Adding gas and byte size limits to collections

access/errors.go

@@ -48,3 +48,13 @@ type InvalidGasLimitError struct {
 func (e InvalidGasLimitError) Error() string {
 	return fmt.Sprintf("transaction gas limit (%d) exceeds the maximum gas limit (%d)", e.Actual, e.Maximum)
 }
+
+// InvalidTxByteSizeError indicates that a transaction byte size exceeds the maximum.
+type InvalidTxByteSizeError struct {
+	Maximum uint64
+	Actual  uint64
+}
+
+func (e InvalidTxByteSizeError) Error() string {
+	return fmt.Sprintf("transaction byte size (%d) exceeds the maximum byte size allowed for a transaction (%d)", e.Actual, e.Maximum)
+}

access/validator.go

@@ -48,6 +48,7 @@ type TransactionValidationOptions struct {
 	AllowUnknownReferenceBlockID bool
 	MaxGasLimit                  uint64
 	CheckScriptsParse            bool
+	MaxTxSizeLimit               uint64
 }
 
 type TransactionValidator struct {
@@ -66,6 +67,11 @@ func NewTransactionValidator(
 }
 
 func (v *TransactionValidator) Validate(tx *flow.TransactionBody) (err error) {
+	err = v.checkTxSizeLimit(tx)
+	if err != nil {
+		return err
+	}
+
 	err = v.checkMissingFields(tx)
 	if err != nil {
 		return err
@@ -91,6 +97,17 @@ func (v *TransactionValidator) Validate(tx *flow.TransactionBody) (err error) {
 	return nil
 }
 
+func (v *TransactionValidator) checkTxSizeLimit(tx *flow.TransactionBody) error {
+	txSize := uint64(tx.ByteSize())
+	if txSize > v.options.MaxTxSizeLimit {
+		return InvalidTxByteSizeError{
+			Actual:  txSize,
+			Maximum: v.options.MaxTxSizeLimit,
+		}
+	}
+	return nil
+}
+
 func (v *TransactionValidator) checkMissingFields(tx *flow.TransactionBody) error {
 	missingFields := tx.MissingFields()
 

cmd/collection/main.go

@@ -46,6 +46,8 @@ func main() {
 	var (
 		txLimit                                uint
 		maxCollectionSize                      uint
+		maxCollectionByteSize                  uint
+		maxCollectionTotalGas                  uint64
 		builderExpiryBuffer                    uint
 		builderPayerRateLimit                  float64
 		builderUnlimitedPayers                 []string
@@ -94,6 +96,10 @@ func main() {
 				"set of payer addresses which are omitted from rate limiting")
 			flags.UintVar(&maxCollectionSize, "builder-max-collection-size", 200,
 				"maximum number of transactions in proposed collections")
+			flags.UintVar(&maxCollectionByteSize, "builder-max-collection-byte-size", 1000000,
+				"maximum byte size of the proposed collection")
+			flags.Uint64Var(&maxCollectionTotalGas, "builder-max-collection-total-gas", uint64(10000000),
+				"maximum total amount of maxgas of transactions in proposed collections")
 			flags.DurationVar(&hotstuffTimeout, "hotstuff-timeout", 60*time.Second,
 				"the initial timeout for the hotstuff pacemaker")
 			flags.DurationVar(&hotstuffMinTimeout, "hotstuff-min-timeout", 2500*time.Millisecond,
@@ -287,6 +293,8 @@ func main() {
 				colMetrics,
 				push,
 				builder.WithMaxCollectionSize(maxCollectionSize),
+				builder.WithMaxCollectionByteSize(maxCollectionByteSize),
+				builder.WithMaxCollectionTotalGas(maxCollectionTotalGas),
 				builder.WithExpiryBuffer(builderExpiryBuffer),
 				builder.WithMaxPayerTransactionRate(builderPayerRateLimit),
 				builder.WithUnlimitedPayers(unlimitedPayers...),

engine/access/rpc/backend/backend.go

@@ -122,6 +122,7 @@ func configureTransactionValidator(state protocol.State) *access.TransactionVali
 			AllowUnknownReferenceBlockID: false,
 			MaxGasLimit:                  flow.DefaultMaxGasLimit,
 			CheckScriptsParse:            true,
+			MaxTxSizeLimit:               flow.DefaultMaxTxSizeLimit,
 		},
 	)
 }

engine/collection/ingest/engine.go

@@ -142,9 +142,11 @@ func (e *Engine) process(originID flow.Identifier, event interface{}) error {
 // from outside the system or routed from another collection node.
 func (e *Engine) onTransaction(originID flow.Identifier, tx *flow.TransactionBody) error {
 
+	txID := tx.ID()
+
 	log := e.log.With().
 		Hex("origin_id", originID[:]).
-		Hex("tx_id", logging.Entity(tx)).
+		Hex("tx_id", txID[:]).
 		Hex("ref_block_id", tx.ReferenceBlockID[:]).
 		Logger()
 
@@ -179,7 +181,6 @@ func (e *Engine) onTransaction(originID flow.Identifier, tx *flow.TransactionBod
 	pool := e.pools.ForEpoch(counter)
 
 	// short-circuit if we have already stored the transaction
-	txID := tx.ID()
 	if pool.Has(txID) {
 		e.log.Debug().Msg("received dupe transaction")
 		return nil
@@ -202,13 +203,16 @@ func (e *Engine) onTransaction(originID flow.Identifier, tx *flow.TransactionBod
 		return fmt.Errorf("node is not assigned to any cluster in this epoch: %d", counter)
 	}
 
+	localClusterFingerPrint := localCluster.Fingerprint()
+	txClusterFingerPrint := txCluster.Fingerprint()
+
 	log = log.With().
-		Hex("local_cluster", logging.ID(localCluster.Fingerprint())).
-		Hex("tx_cluster", logging.ID(txCluster.Fingerprint())).
+		Hex("local_cluster", logging.ID(localClusterFingerPrint)).
+		Hex("tx_cluster", logging.ID(txClusterFingerPrint)).
 		Logger()
 
 	// if our cluster is responsible for the transaction, add it to the mempool
-	if localCluster.Fingerprint() == txCluster.Fingerprint() {
+	if localClusterFingerPrint == txClusterFingerPrint {
 		_ = pool.Add(tx)
 		e.colMetrics.TransactionIngested(txID)
 		log.Debug().Msg("added transaction to pool")

model/flow/constants.go

@@ -19,6 +19,9 @@ const DefaultTransactionExpiryBuffer = 30
 // DefaultMaxGasLimit is the default maximum value for the transaction gas limit.
 const DefaultMaxGasLimit = 9999
 
+// DefaultMaxTxSizeLimit is the default maximum transaction byte size. (64KB)
+const DefaultMaxTxSizeLimit = 64000
+
 // DefaultAuctionWindow defines the length of the auction window at the beginning of
 // an epoch, during which nodes can bid for seats in the committee. Valid epoch events
 // such as setup and commit can only be submitted after this window has passed.

model/flow/transaction.go

@@ -64,6 +64,26 @@ func (tb TransactionBody) Fingerprint() []byte {
 	})
 }
 
+func (tb TransactionBody) ByteSize() int {
+	size := 0
+	size += len(tb.ReferenceBlockID)
+	size += len(tb.Script)
+	for _, arg := range tb.Arguments {
+		size += len(arg)
+	}
+	size += 8 // gas size
+	size += tb.ProposalKey.ByteSize()
+	size += AddressLength                       // payer address
+	size += len(tb.Authorizers) * AddressLength // Authorizers
+	for _, s := range tb.PayloadSignatures {
+		size += s.ByteSize()
+	}
+	for _, s := range tb.EnvelopeSignatures {
+		size += s.ByteSize()
+	}
+	return size
+}
+
 func (tb TransactionBody) ID() Identifier {
 	return MakeID(tb)
 }
@@ -386,6 +406,11 @@ type ProposalKey struct {
 	SequenceNumber uint64
 }
 
+// ByteSize returns the byte size of the proposal key
+func (p ProposalKey) ByteSize() int {
+	return len(p.Address) + 8 + 8
+}
+
 // A TransactionSignature is a signature associated with a specific account key.
 type TransactionSignature struct {
 	Address     Address
@@ -394,6 +419,11 @@ type TransactionSignature struct {
 	Signature   []byte
 }
 
+// ByteSize returns the byte size of the transaction signature
+func (s TransactionSignature) ByteSize() int {
+	return len(s.Address) + 8 + 8 + len(s.Signature)
+}
+
 func (s TransactionSignature) Fingerprint() []byte {
 	return fingerprint.Fingerprint(s.canonicalForm())
 }

module/builder/collection/builder.go

@@ -195,13 +195,25 @@ func (b *Builder) BuildOn(parentID flow.Identifier, setter func(*flow.Header) er
 		minRefID := refChainFinalizedID
 
 		var transactions []*flow.TransactionBody
+		var totalBytesSize uint
+		var totalGas uint64
 		for _, tx := range b.transactions.All() {
 
 			// if we have reached maximum number of transactions, stop
 			if uint(len(transactions)) >= b.config.MaxCollectionSize {
 				break
 			}
 
+			// skip the transaction (keep for the next collection)
+			if totalBytesSize >= b.config.MaxCollectionByteSize {
+				continue
+			}
+
+			// skip the transaction (keep for the next collection)
+			if totalGas >= b.config.MaxCollectionTotalGas {
+				continue
+			}
+
 			// retrieve the main chain header that was used as reference
 			refHeader, err := b.mainHeaders.ByBlockID(tx.ReferenceBlockID)
 			if errors.Is(err, storage.ErrNotFound) {
@@ -251,6 +263,8 @@ func (b *Builder) BuildOn(parentID flow.Identifier, setter func(*flow.Header) er
 			limiter.transactionIncluded(tx)
 
 			transactions = append(transactions, tx)
+			totalBytesSize += uint(tx.ByteSize())
+			totalGas += tx.GasLimit
 		}
 
 		// STEP FOUR: we have a set of transactions that are valid to include

module/builder/collection/builder_test.go

@@ -459,6 +459,42 @@ func (suite *BuilderSuite) TestBuildOn_MaxCollectionSize() {
 	suite.Assert().Equal(builtCollection.Len(), 1)
 }
 
+func (suite *BuilderSuite) TestBuildOn_MaxCollectionByteSize() {
+	// set the max collection byte size to 50
+	suite.builder = builder.NewBuilder(suite.db, trace.NewNoopTracer(), suite.headers, suite.headers, suite.payloads, suite.pool, builder.WithMaxCollectionByteSize(50))
+
+	// build a block
+	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
+	suite.Require().Nil(err)
+
+	// retrieve the built block from storage
+	var built model.Block
+	err = suite.db.View(procedure.RetrieveClusterBlock(header.ID(), &built))
+	suite.Require().Nil(err)
+	builtCollection := built.Payload.Collection
+
+	// should be only 1 transaction in the collection
+	suite.Assert().Equal(builtCollection.Len(), 1)
+}
+
+func (suite *BuilderSuite) TestBuildOn_MaxCollectionTotalGas() {
+	// set the max gas to 10000
+	suite.builder = builder.NewBuilder(suite.db, trace.NewNoopTracer(), suite.headers, suite.headers, suite.payloads, suite.pool, builder.WithMaxCollectionTotalGas(10000))
+
+	// build a block
+	header, err := suite.builder.BuildOn(suite.genesis.ID(), noopSetter)
+	suite.Require().Nil(err)
+
+	// retrieve the built block from storage
+	var built model.Block
+	err = suite.db.View(procedure.RetrieveClusterBlock(header.ID(), &built))
+	suite.Require().Nil(err)
+	builtCollection := built.Payload.Collection
+
+	// should be only 1 transaction in the collection
+	suite.Assert().Equal(builtCollection.Len(), 1)
+}
+
 func (suite *BuilderSuite) TestBuildOn_ExpiredTransaction() {
 
 	// create enough main-chain blocks that an expired transaction is possible

module/builder/collection/config.go

@@ -28,6 +28,12 @@ type Config struct {
 	// UnlimitedPayer is a set of addresses which are not affected by per-payer
 	// rate limiting.
 	UnlimitedPayers map[flow.Address]struct{}
+
+	// MaxCollectionByteSize is the maximum byte size of a collection.
+	MaxCollectionByteSize uint
+
+	// MaxCollectionTotalGas is the maximum of total of gas per collection (sum of maxGasLimit over transactions)
+	MaxCollectionTotalGas uint64
 }
 
 func DefaultConfig() Config {
@@ -36,6 +42,8 @@ func DefaultConfig() Config {
 		ExpiryBuffer:            15,                              // 15 blocks for collections to be included
 		MaxPayerTransactionRate: 0,                               // no rate limiting
 		UnlimitedPayers:         make(map[flow.Address]struct{}), // no unlimited payers
+		MaxCollectionByteSize:   1000000,                         // 1MB
+		MaxCollectionTotalGas:   uint64(10000000),                // 10M
 	}
 }
 
@@ -71,3 +79,15 @@ func WithUnlimitedPayers(payers ...flow.Address) Opt {
 		c.UnlimitedPayers = lookup
 	}
 }
+
+func WithMaxCollectionByteSize(limit uint) Opt {
+	return func(c *Config) {
+		c.MaxCollectionByteSize = limit
+	}
+}
+
+func WithMaxCollectionTotalGas(limit uint64) Opt {
+	return func(c *Config) {
+		c.MaxCollectionTotalGas = limit
+	}
+}