Sequencer L2 block parallel processing improvements (0xPolygonHermez#…

…3604) * wip * first implementation of parallel sequencer optmizations and L2 block reorg management * Close sipBatch (if needed) when processing reorg. Halt when 2 consecuties reorgs (same L2 block) * Return error when reserved counters overflow on l2 block process. Log used/reserved counters when closing wip batch * added logs to analyze blocking issue when storing L2 block * Fix unlock mutex in addTxTracker. Set wipTx to nil in RestoreTxsPendingToStore * add high reserved resorces in wipBatch * store high reserved counter on statedb.batch table * Return contextId in ProcessBatchV2 * fix synchornizer test * Set SequentialProcessL2Block to false by default. Update node config documentation * fix non-e2e tests * fix finalizer tests * remove unused code * test * Fix sequencer loadFromPool gofunc. Fix docker compose variables
superksx · Apr 30, 2024 · 8d5cf96 · 8d5cf96
1 parent e261069
commit 8d5cf96
Show file tree

Hide file tree

Showing 41 changed files with 1,208 additions and 497 deletions.
diff --git a/config/config_test.go b/config/config_test.go
@@ -101,6 +101,10 @@ func Test_Defaults(t *testing.T) {
 			path:          "Sequencer.Finalizer.ResourceExhaustedMarginPct",
 			expectedValue: uint32(10),
 		},
+		{
+			path:          "Sequencer.Finalizer.StateRootSyncInterval",
+			expectedValue: types.NewDuration(3600 * time.Second),
+		},
 		{
 			path:          "Sequencer.Finalizer.ForcedBatchesL1BlockConfirmations",
 			expectedValue: uint64(64),
@@ -127,7 +131,7 @@ func Test_Defaults(t *testing.T) {
 		},
 		{
 			path:          "Sequencer.Finalizer.BatchMaxDeltaTimestamp",
-			expectedValue: types.NewDuration(10 * time.Second),
+			expectedValue: types.NewDuration(1800 * time.Second),
 		},
 		{
 			path:          "Sequencer.Finalizer.Metrics.Interval",

diff --git a/config/default.go b/config/default.go
@@ -146,12 +146,13 @@ StateConsistencyCheckInterval = "5s"
 		ForcedBatchesCheckInterval = "10s"
 		L1InfoTreeL1BlockConfirmations = 64
 		L1InfoTreeCheckInterval = "10s"
-		BatchMaxDeltaTimestamp = "10s"
+		BatchMaxDeltaTimestamp = "1800s"
 		L2BlockMaxDeltaTimestamp = "3s"
 		ResourceExhaustedMarginPct = 10
+		StateRootSyncInterval = "3600s"
 		HaltOnBatchNumber = 0
 		SequentialBatchSanityCheck = false
-		SequentialProcessL2Block = true
+		SequentialProcessL2Block = false
 	[Sequencer.Finalizer.Metrics]
 		Interval = "60m"
 		EnableLog = true

diff --git a/config/environments/local/local.node.config.toml b/config/environments/local/local.node.config.toml
@@ -101,9 +101,10 @@ StateConsistencyCheckInterval = "5s"
 		BatchMaxDeltaTimestamp = "120s"
 		L2BlockMaxDeltaTimestamp = "3s"
 		ResourceExhaustedMarginPct = 10
+		StateRootSyncInterval = "360s"
 		HaltOnBatchNumber = 0
 		SequentialBatchSanityCheck = false
-		SequentialProcessL2Block = true
+		SequentialProcessL2Block = false
 	[Sequencer.Finalizer.Metrics]
 		Interval = "60m"
 		EnableLog = true		

diff --git a/db/migrations/state/0021.sql b/db/migrations/state/0021.sql
@@ -0,0 +1,7 @@
+-- +migrate Up
+ALTER TABLE state.batch
+    ADD COLUMN high_reserved_counters JSONB;
+
+-- +migrate Down
+ALTER TABLE state.batch
+    DROP COLUMN high_reserved_counters;
diff --git a/db/migrations/state/0021_test.go b/db/migrations/state/0021_test.go
@@ -0,0 +1,64 @@
+package migrations_test
+
+import (
+	"database/sql"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+type migrationTest0021 struct{}
+
+func (m migrationTest0021) InsertData(db *sql.DB) error {
+	const insertBatch0 = `
+		INSERT INTO state.batch (batch_num, global_exit_root, local_exit_root, acc_input_hash, state_root, timestamp, coinbase, raw_txs_data, forced_batch_num, wip) 
+		VALUES (0,'0x0000', '0x0000', '0x0000', '0x0000', now(), '0x0000', null, null, true)`
+
+	// insert batch
+	_, err := db.Exec(insertBatch0)
+	if err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m migrationTest0021) RunAssertsAfterMigrationUp(t *testing.T, db *sql.DB) {
+	var result int
+
+	// Check column high_reserved_counters exists in state.batch table
+	const getColumn = `SELECT count(*) FROM information_schema.columns WHERE table_name='batch' and column_name='high_reserved_counters'`
+	row := db.QueryRow(getColumn)
+	assert.NoError(t, row.Scan(&result))
+	assert.Equal(t, 1, result)
+
+	const insertBatch0 = `
+		INSERT INTO state.batch (batch_num, global_exit_root, local_exit_root, acc_input_hash, state_root, timestamp, coinbase, raw_txs_data, forced_batch_num, wip, high_reserved_counters) 
+		VALUES (1,'0x0001', '0x0001', '0x0001', '0x0001', now(), '0x0001', null, null, true, '{"Steps": 1890125}')`
+
+	// insert batch 1
+	_, err := db.Exec(insertBatch0)
+	assert.NoError(t, err)
+
+	const insertBatch1 = `
+		INSERT INTO state.batch (batch_num, global_exit_root, local_exit_root, acc_input_hash, state_root, timestamp, coinbase, raw_txs_data, forced_batch_num, wip, high_reserved_counters) 
+		VALUES (2,'0x0002', '0x0002', '0x0002', '0x0002', now(), '0x0002', null, null, false, '{"Steps": 1890125}')`
+
+	// insert batch 2
+	_, err = db.Exec(insertBatch1)
+	assert.NoError(t, err)
+}
+
+func (m migrationTest0021) RunAssertsAfterMigrationDown(t *testing.T, db *sql.DB) {
+	var result int
+
+	// Check column high_reserved_counters doesn't exists in state.batch table
+	const getCheckedColumn = `SELECT count(*) FROM information_schema.columns WHERE table_name='batch' and column_name='high_reserved_counters'`
+	row := db.QueryRow(getCheckedColumn)
+	assert.NoError(t, row.Scan(&result))
+	assert.Equal(t, 0, result)
+}
+
+func TestMigration0021(t *testing.T) {
+	runMigrationTest(t, 21, migrationTest0021{})
+}
diff --git a/docs/config-file/node-config-doc.html b/docs/config-file/node-config-doc.html
diff --git a/docs/config-file/node-config-doc.md b/docs/config-file/node-config-doc.md
@@ -2083,6 +2083,7 @@ StateConsistencyCheckInterval="5s"
 | - [L1InfoTreeCheckInterval](#Sequencer_Finalizer_L1InfoTreeCheckInterval )                     | No      | string  | No         | -          | Duration                                                                                                                                                                                                      |
 | - [BatchMaxDeltaTimestamp](#Sequencer_Finalizer_BatchMaxDeltaTimestamp )                       | No      | string  | No         | -          | Duration                                                                                                                                                                                                      |
 | - [L2BlockMaxDeltaTimestamp](#Sequencer_Finalizer_L2BlockMaxDeltaTimestamp )                   | No      | string  | No         | -          | Duration                                                                                                                                                                                                      |
+| - [StateRootSyncInterval](#Sequencer_Finalizer_StateRootSyncInterval )                         | No      | string  | No         | -          | Duration                                                                                                                                                                                                      |
 | - [HaltOnBatchNumber](#Sequencer_Finalizer_HaltOnBatchNumber )                                 | No      | integer | No         | -          | HaltOnBatchNumber specifies the batch number where the Sequencer will stop to process more transactions and generate new batches.<br />The Sequencer will halt after it closes the batch equal to this number |
 | - [SequentialBatchSanityCheck](#Sequencer_Finalizer_SequentialBatchSanityCheck )               | No      | boolean | No         | -          | SequentialBatchSanityCheck indicates if the reprocess of a closed batch (sanity check) must be done in a<br />sequential way (instead than in parallel)                                                       |
 | - [SequentialProcessL2Block](#Sequencer_Finalizer_SequentialProcessL2Block )                   | No      | boolean | No         | -          | SequentialProcessL2Block indicates if the processing of a L2 Block must be done in the same finalizer go func instead<br />in the processPendingL2Blocks go func                                              |
@@ -2216,7 +2217,7 @@ ForcedBatchesCheckInterval="10s"
 
 **Default:** `"10s"`
 
-**Description:** L1InfoTreeCheckInterval is the wait time to check if the L1InfoRoot has been updated
+**Description:** L1InfoTreeCheckInterval is the time interval to check if the L1InfoRoot has been updated
 
 **Examples:** 
 
@@ -2240,7 +2241,7 @@ L1InfoTreeCheckInterval="10s"
 
 **Type:** : `string`
 
-**Default:** `"10s"`
+**Default:** `"30m0s"`
 
 **Description:** BatchMaxDeltaTimestamp is the resolution of the timestamp used to close a batch
 
@@ -2254,10 +2255,10 @@ L1InfoTreeCheckInterval="10s"
 "300ms"
 ```
 
-**Example setting the default value** ("10s"):
+**Example setting the default value** ("30m0s"):
 ```
 [Sequencer.Finalizer]
-BatchMaxDeltaTimestamp="10s"
+BatchMaxDeltaTimestamp="30m0s"
 ```
 
 #### <a name="Sequencer_Finalizer_L2BlockMaxDeltaTimestamp"></a>10.7.9. `Sequencer.Finalizer.L2BlockMaxDeltaTimestamp`
@@ -2286,7 +2287,34 @@ BatchMaxDeltaTimestamp="10s"
 L2BlockMaxDeltaTimestamp="3s"
 ```
 
-#### <a name="Sequencer_Finalizer_HaltOnBatchNumber"></a>10.7.10. `Sequencer.Finalizer.HaltOnBatchNumber`
+#### <a name="Sequencer_Finalizer_StateRootSyncInterval"></a>10.7.10. `Sequencer.Finalizer.StateRootSyncInterval`
+
+**Title:** Duration
+
+**Type:** : `string`
+
+**Default:** `"1h0m0s"`
+
+**Description:** StateRootSyncInterval indicates how often the stateroot generated by the L2 block process will be synchronized with
+the stateroot used in the tx-by-tx execution
+
+**Examples:** 
+
+```json
+"1m"
+```
+
+```json
+"300ms"
+```
+
+**Example setting the default value** ("1h0m0s"):
+```
+[Sequencer.Finalizer]
+StateRootSyncInterval="1h0m0s"
+```
+
+#### <a name="Sequencer_Finalizer_HaltOnBatchNumber"></a>10.7.11. `Sequencer.Finalizer.HaltOnBatchNumber`
 
 **Type:** : `integer`
 
@@ -2301,7 +2329,7 @@ The Sequencer will halt after it closes the batch equal to this number
 HaltOnBatchNumber=0
 ```
 
-#### <a name="Sequencer_Finalizer_SequentialBatchSanityCheck"></a>10.7.11. `Sequencer.Finalizer.SequentialBatchSanityCheck`
+#### <a name="Sequencer_Finalizer_SequentialBatchSanityCheck"></a>10.7.12. `Sequencer.Finalizer.SequentialBatchSanityCheck`
 
 **Type:** : `boolean`
 
@@ -2316,22 +2344,22 @@ sequential way (instead than in parallel)
 SequentialBatchSanityCheck=false
 ```
 
-#### <a name="Sequencer_Finalizer_SequentialProcessL2Block"></a>10.7.12. `Sequencer.Finalizer.SequentialProcessL2Block`
+#### <a name="Sequencer_Finalizer_SequentialProcessL2Block"></a>10.7.13. `Sequencer.Finalizer.SequentialProcessL2Block`
 
 **Type:** : `boolean`
 
-**Default:** `true`
+**Default:** `false`
 
 **Description:** SequentialProcessL2Block indicates if the processing of a L2 Block must be done in the same finalizer go func instead
 in the processPendingL2Blocks go func
 
-**Example setting the default value** (true):
+**Example setting the default value** (false):
 ```
 [Sequencer.Finalizer]
-SequentialProcessL2Block=true
+SequentialProcessL2Block=false
 ```
 
-#### <a name="Sequencer_Finalizer_Metrics"></a>10.7.13. `[Sequencer.Finalizer.Metrics]`
+#### <a name="Sequencer_Finalizer_Metrics"></a>10.7.14. `[Sequencer.Finalizer.Metrics]`
 
 **Type:** : `object`
 **Description:** Metrics is the config for the sequencer metrics
@@ -2341,7 +2369,7 @@ SequentialProcessL2Block=true
 | - [Interval](#Sequencer_Finalizer_Metrics_Interval )   | No      | string  | No         | -          | Duration                                           |
 | - [EnableLog](#Sequencer_Finalizer_Metrics_EnableLog ) | No      | boolean | No         | -          | EnableLog is a flag to enable/disable metrics logs |
 
-##### <a name="Sequencer_Finalizer_Metrics_Interval"></a>10.7.13.1. `Sequencer.Finalizer.Metrics.Interval`
+##### <a name="Sequencer_Finalizer_Metrics_Interval"></a>10.7.14.1. `Sequencer.Finalizer.Metrics.Interval`
 
 **Title:** Duration
 
@@ -2367,7 +2395,7 @@ SequentialProcessL2Block=true
 Interval="1h0m0s"
 ```
 
-##### <a name="Sequencer_Finalizer_Metrics_EnableLog"></a>10.7.13.2. `Sequencer.Finalizer.Metrics.EnableLog`
+##### <a name="Sequencer_Finalizer_Metrics_EnableLog"></a>10.7.14.2. `Sequencer.Finalizer.Metrics.EnableLog`
 
 **Type:** : `boolean`
 

diff --git a/docs/config-file/node-config-schema.json b/docs/config-file/node-config-schema.json
@@ -825,7 +825,7 @@
 						"L1InfoTreeCheckInterval": {
 							"type": "string",
 							"title": "Duration",
-							"description": "L1InfoTreeCheckInterval is the wait time to check if the L1InfoRoot has been updated",
+							"description": "L1InfoTreeCheckInterval is the time interval to check if the L1InfoRoot has been updated",
 							"default": "10s",
 							"examples": [
 								"1m",
@@ -836,7 +836,7 @@
 							"type": "string",
 							"title": "Duration",
 							"description": "BatchMaxDeltaTimestamp is the resolution of the timestamp used to close a batch",
-							"default": "10s",
+							"default": "30m0s",
 							"examples": [
 								"1m",
 								"300ms"
@@ -852,6 +852,16 @@
 								"300ms"
 							]
 						},
+						"StateRootSyncInterval": {
+							"type": "string",
+							"title": "Duration",
+							"description": "StateRootSyncInterval indicates how often the stateroot generated by the L2 block process will be synchronized with\nthe stateroot used in the tx-by-tx execution",
+							"default": "1h0m0s",
+							"examples": [
+								"1m",
+								"300ms"
+							]
+						},
 						"HaltOnBatchNumber": {
 							"type": "integer",
 							"description": "HaltOnBatchNumber specifies the batch number where the Sequencer will stop to process more transactions and generate new batches.\nThe Sequencer will halt after it closes the batch equal to this number",
@@ -865,7 +875,7 @@
 						"SequentialProcessL2Block": {
 							"type": "boolean",
 							"description": "SequentialProcessL2Block indicates if the processing of a L2 Block must be done in the same finalizer go func instead\nin the processPendingL2Blocks go func",
-							"default": true
+							"default": false
 						},
 						"Metrics": {
 							"properties": {

diff --git a/event/event.go b/event/event.go
@@ -50,6 +50,8 @@ const (
 	EventID_ReservedZKCountersOverflow EventID = "RESERVED ZKCOUNTERS OVERFLOW"
 	// EventID_InvalidInfoRoot is triggered when an invalid l1InfoRoot was synced
 	EventID_InvalidInfoRoot EventID = "INVALID INFOROOT"
+	// EventID_L2BlockReorg is triggered when a L2 block reorg has happened in the sequencer
+	EventID_L2BlockReorg EventID = "L2 BLOCK REORG"
 	// Source_Node is the source of the event
 	Source_Node Source = "node"
 

diff --git a/go.mod b/go.mod
@@ -3,7 +3,7 @@ module github.com/0xPolygonHermez/zkevm-node
 go 1.21
 
 require (
-	github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240422135400-0df0d27226b3
+	github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240426122934-6f47d2485fc1
 	github.com/didip/tollbooth/v6 v6.1.2
 	github.com/dop251/goja v0.0.0-20230806174421-c933cf95e127
 	github.com/ethereum/go-ethereum v1.13.11

diff --git a/go.sum b/go.sum
@@ -39,8 +39,8 @@ cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9
 dario.cat/mergo v1.0.0 h1:AGCNq9Evsj31mOgNPcLyXc+4PNABt905YmuqPYYpBWk=
 dario.cat/mergo v1.0.0/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
-github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240422135400-0df0d27226b3 h1:g5IMJalQxVRNfnXrzQG7bx2COktaFBf1mNuF4SLuQss=
-github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240422135400-0df0d27226b3/go.mod h1:0QkAXcFa92mFJrCbN3UPUJGJYes851yEgYHLONnaosE=
+github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240426122934-6f47d2485fc1 h1:4wbCJOGcZ8BTuOfNFrcZ1cAVfTWaX1W9EYHaDx3imLc=
+github.com/0xPolygonHermez/zkevm-data-streamer v0.2.3-0.20240426122934-6f47d2485fc1/go.mod h1:0QkAXcFa92mFJrCbN3UPUJGJYes851yEgYHLONnaosE=
 github.com/AndreasBriese/bbloom v0.0.0-20190306092124-e2d15f34fcf9/go.mod h1:bOvUY6CB00SOBii9/FifXqc0awNKxLFCL/+pkDPuyl8=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78 h1:w+iIsaOQNcT7OZ575w+acHgRric5iCyQh+xv+KJ4HB8=
 github.com/Azure/go-ansiterm v0.0.0-20170929234023-d6e3b3328b78/go.mod h1:LmzpDX56iTiv29bbRTIsUNlaFfuhWRQBWjQdVyAevI8=

diff --git a/sequencer/addrqueue.go b/sequencer/addrqueue.go
@@ -121,22 +121,25 @@ func (a *addrQueue) IsEmpty() bool {
 }
 
 // deleteTx deletes the tx from the addrQueue
-func (a *addrQueue) deleteTx(txHash common.Hash) (deletedReadyTx *TxTracker) {
+func (a *addrQueue) deleteTx(txHash common.Hash) (deletedTx *TxTracker, isReady bool) {
 	txHashStr := txHash.String()
 
 	if (a.readyTx != nil) && (a.readyTx.HashStr == txHashStr) {
 		log.Infof("deleting readyTx %s from addrQueue %s", txHashStr, a.fromStr)
 		prevReadyTx := a.readyTx
 		a.readyTx = nil
-		return prevReadyTx
+		return prevReadyTx, true
 	} else {
+		var deletedTx *TxTracker
 		for _, txTracker := range a.notReadyTxs {
 			if txTracker.HashStr == txHashStr {
+				deletedTx = txTracker
 				log.Infof("deleting notReadyTx %s from addrQueue %s", txHashStr, a.fromStr)
 				delete(a.notReadyTxs, txTracker.Nonce)
+				break
 			}
 		}
-		return nil
+		return deletedTx, false
 	}
 }
 
@@ -158,6 +161,22 @@ func (a *addrQueue) deletePendingTxToStore(txHash common.Hash) {
 	}
 }
 
+func (a *addrQueue) getTransactions() []*TxTracker {
+	// TODO: Add test for this function
+
+	txsList := []*TxTracker{}
+
+	if a.readyTx != nil {
+		txsList = append(txsList, a.readyTx)
+	}
+
+	for _, tx := range a.notReadyTxs {
+		txsList = append(txsList, tx)
+	}
+
+	return txsList
+}
+
 // updateCurrentNonceBalance updates the nonce and balance of the addrQueue and updates the ready and notReady txs
 func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (newReadyTx, prevReadyTx *TxTracker, toDelete []*TxTracker) {
 	var oldReadyTx *TxTracker = nil
@@ -179,7 +198,7 @@ func (a *addrQueue) updateCurrentNonceBalance(nonce *uint64, balance *big.Int) (
 				}
 			}
 			for _, txTracker := range txsToDelete {
-				log.Infof("deleting notReadyTx with nonce %d from addrQueue %s", txTracker.Nonce, a.fromStr)
+				log.Infof("deleting notReadyTx with nonce %d from addrQueue %s, reason: %s", txTracker.Nonce, a.fromStr, *txTracker.FailedReason)
 				delete(a.notReadyTxs, txTracker.Nonce)
 			}
 		}

diff --git a/sequencer/addrqueue_test.go b/sequencer/addrqueue_test.go
@@ -164,11 +164,11 @@ func TestAddrQueue(t *testing.T) {
 	t.Run("Delete readyTx 0x01", func(t *testing.T) {
 		tc := addTxTestCases[2]
 		tx := newTestTxTracker(tc.hash, tc.nonce, tc.gasPrice, tc.cost)
-		deltx := addr.deleteTx(tx.Hash)
+		deltx, isReady := addr.deleteTx(tx.Hash)
 		if !(addr.readyTx == nil) {
 			t.Fatalf("Error readyTx not nil. Expected=%s, Actual=%s", "", addr.readyTx.HashStr)
 		}
-		if !(deltx.HashStr == tx.HashStr) {
+		if !isReady || !(deltx.HashStr == tx.HashStr) {
 			t.Fatalf("Error returning deletedReadyTx. Expected=%s, Actual=%s", tx.HashStr, deltx.HashStr)
 		}
 	})