[TxPool] Introduce limit on enqueued transactions (#687)

* introduce limit on the number of (currently) enqueued transactions

command/server/config/config.go

@@ -51,8 +51,9 @@ type Network struct {
 
 // TxPool defines the TxPool configuration params
 type TxPool struct {
-	PriceLimit uint64 `json:"price_limit" yaml:"price_limit"`
-	MaxSlots   uint64 `json:"max_slots" yaml:"max_slots"`
+	PriceLimit         uint64 `json:"price_limit" yaml:"price_limit"`
+	MaxSlots           uint64 `json:"max_slots" yaml:"max_slots"`
+	MaxAccountEnqueued uint64 `json:"max_account_enqueued" yaml:"max_account_enqueued"`
 }
 
 // Headers defines the HTTP response headers required to enable CORS.
@@ -97,8 +98,9 @@ func DefaultConfig() *Config {
 		Telemetry:  &Telemetry{},
 		ShouldSeal: true,
 		TxPool: &TxPool{
-			PriceLimit: 0,
-			MaxSlots:   4096,
+			PriceLimit:         0,
+			MaxSlots:           4096,
+			MaxAccountEnqueued: 128,
 		},
 		LogLevel:    "INFO",
 		RestoreFile: "",

command/server/params.go

@@ -29,6 +29,7 @@ const (
 	jsonRPCBatchRequestLimitFlag = "json-rpc-batch-request-limit"
 	jsonRPCBlockRangeLimitFlag   = "json-rpc-block-range-limit"
 	maxSlotsFlag                 = "max-slots"
+	maxEnqueuedFlag              = "max-enqueued"
 	blockGasTargetFlag           = "block-gas-target"
 	secretsConfigFlag            = "secrets-config"
 	restoreFlag                  = "restore"
@@ -162,14 +163,15 @@ func (p *serverParams) generateConfig() *server.Config {
 			MaxOutboundPeers: p.rawConfig.Network.MaxOutboundPeers,
 			Chain:            p.genesisConfig,
 		},
-		DataDir:        p.rawConfig.DataDir,
-		Seal:           p.rawConfig.ShouldSeal,
-		PriceLimit:     p.rawConfig.TxPool.PriceLimit,
-		MaxSlots:       p.rawConfig.TxPool.MaxSlots,
-		SecretsManager: p.secretsConfig,
-		RestoreFile:    p.getRestoreFilePath(),
-		BlockTime:      p.rawConfig.BlockTime,
-		LogLevel:       hclog.LevelFromString(p.rawConfig.LogLevel),
-		LogFilePath:    p.logFileLocation,
+		DataDir:            p.rawConfig.DataDir,
+		Seal:               p.rawConfig.ShouldSeal,
+		PriceLimit:         p.rawConfig.TxPool.PriceLimit,
+		MaxSlots:           p.rawConfig.TxPool.MaxSlots,
+		MaxAccountEnqueued: p.rawConfig.TxPool.MaxAccountEnqueued,
+		SecretsManager:     p.secretsConfig,
+		RestoreFile:        p.getRestoreFilePath(),
+		BlockTime:          p.rawConfig.BlockTime,
+		LogLevel:           hclog.LevelFromString(p.rawConfig.LogLevel),
+		LogFilePath:        p.logFileLocation,
 	}
 }

command/server/server.go

@@ -179,6 +179,13 @@ func setFlags(cmd *cobra.Command) {
 		"maximum slots in the pool",
 	)
 
+	cmd.Flags().Uint64Var(
+		&params.rawConfig.TxPool.MaxAccountEnqueued,
+		maxEnqueuedFlag,
+		defaultConfig.TxPool.MaxAccountEnqueued,
+		"maximum number of enqueued transactions per account",
+	)
+
 	cmd.Flags().Uint64Var(
 		&params.rawConfig.BlockTime,
 		blockTimeFlag,

server/config.go

@@ -21,9 +21,10 @@ type Config struct {
 	GRPCAddr   *net.TCPAddr
 	LibP2PAddr *net.TCPAddr
 
-	PriceLimit uint64
-	MaxSlots   uint64
-	BlockTime  uint64
+	PriceLimit         uint64
+	MaxAccountEnqueued uint64
+	MaxSlots           uint64
+	BlockTime          uint64
 
 	Telemetry *Telemetry
 	Network   *network.Config

server/server.go

@@ -209,9 +209,10 @@ func NewServer(config *Config) (*Server, error) {
 			m.network,
 			m.serverMetrics.txpool,
 			&txpool.Config{
-				Sealing:    m.config.Seal,
-				MaxSlots:   m.config.MaxSlots,
-				PriceLimit: m.config.PriceLimit,
+				Sealing:            m.config.Seal,
+				MaxSlots:           m.config.MaxSlots,
+				PriceLimit:         m.config.PriceLimit,
+				MaxAccountEnqueued: m.config.MaxAccountEnqueued,
 			},
 		)
 		if err != nil {

txpool/account.go

@@ -11,7 +11,10 @@ import (
 // Each account (value) is bound to one address (key).
 type accountsMap struct {
 	sync.Map
+
 	count uint64
+
+	maxEnqueuedLimit uint64
 }
 
 // Intializes an account for the given address.
@@ -24,6 +27,9 @@ func (m *accountsMap) initOnce(addr types.Address, nonce uint64) *account {
 		newAccount.enqueued = newAccountQueue()
 		newAccount.promoted = newAccountQueue()
 
+		//	set the limit for enqueued txs
+		newAccount.maxEnqueued = m.maxEnqueuedLimit
+
 		// set the nonce
 		newAccount.setNonce(nonce)
 
@@ -151,6 +157,9 @@ type account struct {
 	enqueued, promoted *accountQueue
 	nextNonce          uint64
 	demotions          uint
+
+	//	maximum number of enqueued transactions
+	maxEnqueued uint64
 }
 
 // getNonce returns the next expected nonce for this account.
@@ -214,6 +223,10 @@ func (a *account) enqueue(tx *types.Transaction) error {
 	a.enqueued.lock(true)
 	defer a.enqueued.unlock()
 
+	if a.enqueued.length() == a.maxEnqueued {
+		return ErrMaxEnqueuedLimitReached
+	}
+
 	// reject low nonce tx
 	if tx.Nonce < a.getNonce() {
 		return ErrNonceTooLow

txpool/txpool.go

@@ -30,18 +30,19 @@ const (
 
 // errors
 var (
-	ErrIntrinsicGas        = errors.New("intrinsic gas too low")
-	ErrBlockLimitExceeded  = errors.New("exceeds block gas limit")
-	ErrNegativeValue       = errors.New("negative value")
-	ErrExtractSignature    = errors.New("cannot extract signature")
-	ErrInvalidSender       = errors.New("invalid sender")
-	ErrTxPoolOverflow      = errors.New("txpool is full")
-	ErrUnderpriced         = errors.New("transaction underpriced")
-	ErrNonceTooLow         = errors.New("nonce too low")
-	ErrInsufficientFunds   = errors.New("insufficient funds for gas * price + value")
-	ErrInvalidAccountState = errors.New("invalid account state")
-	ErrAlreadyKnown        = errors.New("already known")
-	ErrOversizedData       = errors.New("oversized data")
+	ErrIntrinsicGas            = errors.New("intrinsic gas too low")
+	ErrBlockLimitExceeded      = errors.New("exceeds block gas limit")
+	ErrNegativeValue           = errors.New("negative value")
+	ErrExtractSignature        = errors.New("cannot extract signature")
+	ErrInvalidSender           = errors.New("invalid sender")
+	ErrTxPoolOverflow          = errors.New("txpool is full")
+	ErrUnderpriced             = errors.New("transaction underpriced")
+	ErrNonceTooLow             = errors.New("nonce too low")
+	ErrInsufficientFunds       = errors.New("insufficient funds for gas * price + value")
+	ErrInvalidAccountState     = errors.New("invalid account state")
+	ErrAlreadyKnown            = errors.New("already known")
+	ErrOversizedData           = errors.New("oversized data")
+	ErrMaxEnqueuedLimitReached = errors.New("maximum number of enqueued transactions reached")
 )
 
 // indicates origin of a transaction
@@ -79,9 +80,10 @@ type signer interface {
 }
 
 type Config struct {
-	PriceLimit uint64
-	MaxSlots   uint64
-	Sealing    bool
+	PriceLimit         uint64
+	MaxSlots           uint64
+	MaxAccountEnqueued uint64
+	Sealing            bool
 }
 
 /* All requests are passed to the main loop
@@ -183,8 +185,8 @@ func NewTxPool(
 		forks:       forks,
 		store:       store,
 		metrics:     metrics,
-		accounts:    accountsMap{},
 		executables: newPricedQueue(),
+		accounts:    accountsMap{maxEnqueuedLimit: config.MaxAccountEnqueued},
 		index:       lookupMap{all: make(map[types.Hash]*types.Transaction)},
 		gauge:       slotGauge{height: 0, max: config.MaxSlots},
 		priceLimit:  config.PriceLimit,