add db index for gas_price (#154)

* add db index for gas_price

* add dynamic config QueryPendingTxsLimit

db/migrations/pool/1001.sql

@@ -0,0 +1,5 @@
+-- +migrate Up
+CREATE INDEX IF NOT EXISTS idx_transaction_gas_price ON pool.transaction (gas_price);
+
+-- +migrate Down
+DROP INDEX IF EXISTS pool.idx_transaction_gas_price;

docs/config-file/node-config-doc.md

@@ -2572,6 +2572,7 @@ CheckLastL2BlockHashOnCloseBatch=true
 | - [StreamServer](#Sequencer_StreamServer )                                           | No      | object          | No         | -          | StreamServerCfg is the config for the stream server                                                 |
 | - [PackBatchSpacialList](#Sequencer_PackBatchSpacialList )                           | No      | array of string | No         | -          | XLayer config<br />PackBatchSpacialList is the list of addresses that will have a special gas price |
 | - [GasPriceMultiple](#Sequencer_GasPriceMultiple )                                   | No      | number          | No         | -          | GasPriceMultiple is the multiple of the gas price                                                   |
+| - [QueryPendingTxsLimit](#Sequencer_QueryPendingTxsLimit )                           | No      | integer         | No         | -          | QueryPendingTxsLimit is used to limit amount txs from the db                                        |
 
 ### <a name="Sequencer_DeletePoolTxsL1BlockConfirmations"></a>10.1. `Sequencer.DeletePoolTxsL1BlockConfirmations`
 
@@ -3230,6 +3231,20 @@ PackBatchSpacialList is the list of addresses that will have a special gas price
 GasPriceMultiple=0
 ```
 
+### <a name="Sequencer_QueryPendingTxsLimit"></a>10.11. `Sequencer.QueryPendingTxsLimit`
+
+**Type:** : `integer`
+
+**Default:** `0`
+
+**Description:** QueryPendingTxsLimit is used to limit amount txs from the db
+
+**Example setting the default value** (0):
+```
+[Sequencer]
+QueryPendingTxsLimit=0
+```
+
 ## <a name="SequenceSender"></a>11. `[SequenceSender]`
 
 **Type:** : `object`

docs/config-file/node-config-schema.json

@@ -1240,6 +1240,11 @@
 					"type": "number",
 					"description": "GasPriceMultiple is the multiple of the gas price",
 					"default": 0
+				},
+				"QueryPendingTxsLimit": {
+					"type": "integer",
+					"description": "QueryPendingTxsLimit is used to limit amount txs from the db",
+					"default": 0
 				}
 			},
 			"additionalProperties": false,

pool/interfaces.go

@@ -21,7 +21,7 @@ type storage interface {
 	GetPendingTxHashesSince(ctx context.Context, since time.Time) ([]common.Hash, error)
 	GetTxsByFromAndNonce(ctx context.Context, from common.Address, nonce uint64) ([]Transaction, error)
 	GetTxsByStatus(ctx context.Context, state TxStatus, limit uint64) ([]Transaction, error)
-	GetNonWIPPendingTxs(ctx context.Context) ([]Transaction, error)
+	GetNonWIPPendingTxs(ctx context.Context, limit uint64) ([]Transaction, error)
 	IsTxPending(ctx context.Context, hash common.Hash) (bool, error)
 	SetGasPrices(ctx context.Context, l2GasPrice uint64, l1GasPrice uint64) error
 	DeleteGasPricesHistoryOlderThan(ctx context.Context, date time.Time) error

pool/pgpoolstorage/pgpoolstorage.go

@@ -172,17 +172,24 @@ func (p *PostgresPoolStorage) GetTxsByStatus(ctx context.Context, status pool.Tx
 }
 
 // GetNonWIPPendingTxs returns an array of transactions
-func (p *PostgresPoolStorage) GetNonWIPPendingTxs(ctx context.Context) ([]pool.Transaction, error) {
+// limit parameter is used to limit amount txs from the db,
+// if limit = 0, then there is no limit
+func (p *PostgresPoolStorage) GetNonWIPPendingTxs(ctx context.Context, limit uint64) ([]pool.Transaction, error) {
 	var (
 		rows pgx.Rows
 		err  error
 		sql  string
 	)
 
-	sql = `SELECT encoded, status, received_at, is_wip, ip, cumulative_gas_used, used_keccak_hashes, used_poseidon_hashes, used_poseidon_paddings, used_mem_aligns,
-		used_arithmetics, used_binaries, used_steps, used_sha256_hashes, failed_reason, reserved_zkcounters FROM pool.transaction WHERE is_wip IS FALSE and status = $1`
-	rows, err = p.db.Query(ctx, sql, pool.TxStatusPending)
-
+	if limit == 0 {
+		sql = `SELECT encoded, status, received_at, is_wip, ip, cumulative_gas_used, used_keccak_hashes, used_poseidon_hashes, used_poseidon_paddings, used_mem_aligns,
+		used_arithmetics, used_binaries, used_steps, used_sha256_hashes, failed_reason, reserved_zkcounters FROM pool.transaction WHERE is_wip IS FALSE and status = $1 ORDER BY gas_price DESC`
+		rows, err = p.db.Query(ctx, sql, pool.TxStatusPending)
+	} else {
+		sql = `SELECT encoded, status, received_at, is_wip, ip, cumulative_gas_used, used_keccak_hashes, used_poseidon_hashes, used_poseidon_paddings, used_mem_aligns,
+		used_arithmetics, used_binaries, used_steps, used_sha256_hashes, failed_reason, reserved_zkcounters FROM pool.transaction WHERE is_wip IS FALSE and status = $1 ORDER BY gas_price DESC LIMIT $2`
+		rows, err = p.db.Query(ctx, sql, pool.TxStatusPending, limit)
+	}
 	if err != nil {
 		return nil, err
 	}

pool/pool.go

@@ -356,8 +356,8 @@ func (p *Pool) GetPendingTxs(ctx context.Context, limit uint64) ([]Transaction,
 }
 
 // GetNonWIPPendingTxs from the pool
-func (p *Pool) GetNonWIPPendingTxs(ctx context.Context) ([]Transaction, error) {
-	return p.storage.GetNonWIPPendingTxs(ctx)
+func (p *Pool) GetNonWIPPendingTxs(ctx context.Context, limit uint64) ([]Transaction, error) {
+	return p.storage.GetNonWIPPendingTxs(ctx, limit)
 }
 
 // GetSelectedTxs gets selected txs from the pool db

sequencer/apollo_xlayer.go

@@ -13,6 +13,7 @@ type ApolloConfig struct {
 	FullBatchSleepDuration types.Duration
 	PackBatchSpacialList   []string
 	GasPriceMultiple       float64
+	QueryPendingTxsLimit   uint64
 
 	sync.RWMutex
 }
@@ -41,6 +42,7 @@ func UpdateConfig(apolloConfig Config) {
 	getApolloConfig().FullBatchSleepDuration = apolloConfig.Finalizer.FullBatchSleepDuration
 	getApolloConfig().PackBatchSpacialList = apolloConfig.PackBatchSpacialList
 	getApolloConfig().GasPriceMultiple = apolloConfig.GasPriceMultiple
+	getApolloConfig().QueryPendingTxsLimit = apolloConfig.QueryPendingTxsLimit
 	getApolloConfig().Unlock()
 }
 
@@ -82,3 +84,14 @@ func getGasPriceMultiple(gpMul float64) float64 {
 
 	return ret
 }
+
+func getQueryPendingTxsLimit(limit uint64) uint64 {
+	ret := limit
+	if getApolloConfig().Enable() {
+		getApolloConfig().RLock()
+		defer getApolloConfig().RUnlock()
+		ret = getApolloConfig().QueryPendingTxsLimit
+	}
+
+	return ret
+}

sequencer/config.go

@@ -36,6 +36,8 @@ type Config struct {
 	PackBatchSpacialList []string `mapstructure:"PackBatchSpacialList"`
 	// GasPriceMultiple is the multiple of the gas price
 	GasPriceMultiple float64 `mapstructure:"GasPriceMultiple"`
+	// QueryPendingTxsLimit is used to limit amount txs from the db
+	QueryPendingTxsLimit uint64 `mapstructure:"QueryPendingTxsLimit"`
 }
 
 // StreamServerCfg contains the data streamer's configuration properties

sequencer/interfaces.go

@@ -19,7 +19,7 @@ type txPool interface {
 	DeleteFailedTransactionsOlderThan(ctx context.Context, date time.Time) error
 	DeleteTransactionByHash(ctx context.Context, hash common.Hash) error
 	MarkWIPTxsAsPending(ctx context.Context) error
-	GetNonWIPPendingTxs(ctx context.Context) ([]pool.Transaction, error)
+	GetNonWIPPendingTxs(ctx context.Context, limit uint64) ([]pool.Transaction, error)
 	UpdateTxStatus(ctx context.Context, hash common.Hash, newStatus pool.TxStatus, isWIP bool, failedReason *string) error
 	GetTxZkCountersByHash(ctx context.Context, hash common.Hash) (*state.ZKCounters, *state.ZKCounters, error)
 	UpdateTxWIPStatus(ctx context.Context, hash common.Hash, isWIP bool) error

sequencer/sequencer.go

@@ -200,7 +200,7 @@ func (s *Sequencer) expireOldWorkerTxs(ctx context.Context) {
 // loadFromPool keeps loading transactions from the pool
 func (s *Sequencer) loadFromPool(ctx context.Context) {
 	for {
-		poolTransactions, err := s.pool.GetNonWIPPendingTxs(ctx)
+		poolTransactions, err := s.pool.GetNonWIPPendingTxs(ctx, getQueryPendingTxsLimit(s.cfg.QueryPendingTxsLimit))
 		if err != nil && err != pool.ErrNotFound {
 			log.Errorf("error loading txs from pool, error: %v", err)
 		}