ethtxmanager improvements

aggregator/aggregator.go

@@ -275,10 +275,10 @@ func (a *Aggregator) sendFinalProof() {
 				continue
 			}
 			monitoredTxID := buildMonitoredTxID(proof.BatchNumber, proof.BatchNumberFinal)
-			err = a.EthTxManager.Add(ctx, ethTxManagerOwner, monitoredTxID, sender, to, nil, data, nil)
+			err = a.EthTxManager.Add(ctx, ethTxManagerOwner, monitoredTxID, sender, to, nil, data, a.cfg.GasOffset, nil)
 			if err != nil {
-				log := log.WithFields("tx", monitoredTxID)
-				log.Errorf("Error to add batch verification tx to eth tx manager: %v", err)
+				mTxLogger := ethtxmanager.CreateLogger(ethTxManagerOwner, monitoredTxID, sender, to)
+				mTxLogger.Errorf("Error to add batch verification tx to eth tx manager: %v", err)
 				a.handleFailureToAddVerifyBatchToBeMonitored(ctx, proof)
 				continue
 			}
@@ -1027,23 +1027,23 @@ func (hc *healthChecker) Watch(req *grpchealth.HealthCheckRequest, server grpche
 }
 
 func (a *Aggregator) handleMonitoredTxResult(result ethtxmanager.MonitoredTxResult) {
-	resLog := log.WithFields("owner", ethTxManagerOwner, "txId", result.ID)
+	mTxResultLogger := ethtxmanager.CreateMonitoredTxResultLogger(ethTxManagerOwner, result)
 	if result.Status == ethtxmanager.MonitoredTxStatusFailed {
-		resLog.Fatal("failed to send batch verification, TODO: review this fatal and define what to do in this case")
+		mTxResultLogger.Fatal("failed to send batch verification, TODO: review this fatal and define what to do in this case")
 	}
 
 	// monitoredIDFormat: "proof-from-%v-to-%v"
 	idSlice := strings.Split(result.ID, "-")
 	proofBatchNumberStr := idSlice[2]
 	proofBatchNumber, err := strconv.ParseUint(proofBatchNumberStr, encoding.Base10, 0)
 	if err != nil {
-		resLog.Errorf("failed to read final proof batch number from monitored tx: %v", err)
+		mTxResultLogger.Errorf("failed to read final proof batch number from monitored tx: %v", err)
 	}
 
 	proofBatchNumberFinalStr := idSlice[4]
 	proofBatchNumberFinal, err := strconv.ParseUint(proofBatchNumberFinalStr, encoding.Base10, 0)
 	if err != nil {
-		resLog.Errorf("failed to read final proof batch number final from monitored tx: %v", err)
+		mTxResultLogger.Errorf("failed to read final proof batch number final from monitored tx: %v", err)
 	}
 
 	log := log.WithFields("txId", result.ID, "batches", fmt.Sprintf("%d-%d", proofBatchNumber, proofBatchNumberFinal))

aggregator/aggregator_test.go

@@ -54,7 +54,7 @@ func TestSendFinalProof(t *testing.T) {
 		BatchNumberFinal: batchNumFinal,
 	}
 	finalProof := &prover.FinalProof{}
-	cfg := Config{SenderAddress: from.Hex()}
+	cfg := Config{SenderAddress: from.Hex(), GasOffset: uint64(10)}
 
 	testCases := []struct {
 		name    string
@@ -135,7 +135,7 @@ func TestSendFinalProof(t *testing.T) {
 					assert.True(a.verifyingProof)
 				}).Return(&to, data, nil).Once()
 				monitoredTxID := buildMonitoredTxID(batchNum, batchNumFinal)
-				m.ethTxManager.On("Add", mock.Anything, ethTxManagerOwner, monitoredTxID, from, &to, value, data, nil).Return(errBanana).Once()
+				m.ethTxManager.On("Add", mock.Anything, ethTxManagerOwner, monitoredTxID, from, &to, value, data, cfg.GasOffset, nil).Return(errBanana).Once()
 				m.stateMock.On("UpdateGeneratedProof", mock.Anything, recursiveProof, nil).Run(func(args mock.Arguments) {
 					// test is done, stop the sendFinalProof method
 					a.exit()
@@ -160,7 +160,7 @@ func TestSendFinalProof(t *testing.T) {
 					assert.True(a.verifyingProof)
 				}).Return(&to, data, nil).Once()
 				monitoredTxID := buildMonitoredTxID(batchNum, batchNumFinal)
-				m.ethTxManager.On("Add", mock.Anything, ethTxManagerOwner, monitoredTxID, from, &to, value, data, nil).Return(nil).Once()
+				m.ethTxManager.On("Add", mock.Anything, ethTxManagerOwner, monitoredTxID, from, &to, value, data, cfg.GasOffset, nil).Return(nil).Once()
 				ethTxManResult := ethtxmanager.MonitoredTxResult{
 					ID:     monitoredTxID,
 					Status: ethtxmanager.MonitoredTxStatusConfirmed,

aggregator/config.go

@@ -73,4 +73,16 @@ type Config struct {
 	// which a proof in generating state is considered to be stuck and
 	// allowed to be cleared.
 	GeneratingProofCleanupThreshold string `mapstructure:"GeneratingProofCleanupThreshold"`
+
+	// GasOffset is the amount of gas to be added to the gas estimation in order
+	// to provide an amount that is higher than the estimated one. This is used
+	// to avoid the TX getting reverted in case something has changed in the network
+	// state after the estimation which can cause the TX to require more gas to be
+	// executed.
+	//
+	// ex:
+	// gas estimation: 1000
+	// gas offset: 100
+	// final gas: 1100
+	GasOffset uint64 `mapstructure:"GasOffset"`
 }

aggregator/interfaces.go

@@ -29,7 +29,7 @@ type proverInterface interface {
 // ethTxManager contains the methods required to send txs to
 // ethereum.
 type ethTxManager interface {
-	Add(ctx context.Context, owner, id string, from common.Address, to *common.Address, value *big.Int, data []byte, dbTx pgx.Tx) error
+	Add(ctx context.Context, owner, id string, from common.Address, to *common.Address, value *big.Int, data []byte, gasOffset uint64, dbTx pgx.Tx) error
 	Result(ctx context.Context, owner, id string, dbTx pgx.Tx) (ethtxmanager.MonitoredTxResult, error)
 	ResultsByStatus(ctx context.Context, owner string, statuses []ethtxmanager.MonitoredTxStatus, dbTx pgx.Tx) ([]ethtxmanager.MonitoredTxResult, error)
 	ProcessPendingMonitoredTxs(ctx context.Context, owner string, failedResultHandler ethtxmanager.ResultHandler, dbTx pgx.Tx)

cmd/run.go

@@ -173,7 +173,7 @@ func start(cliCtx *cli.Context) error {
 			if poolInstance == nil {
 				poolInstance = createPool(c.Pool, c.State.Batch.Constraints, l2ChainID, st, eventLog)
 			}
-			seq := createSequencer(*c, poolInstance, ethTxManagerStorage, st, eventLog)
+			seq := createSequencer(*c, poolInstance, st, eventLog)
 			go seq.Start(cliCtx.Context)
 		case SEQUENCE_SENDER:
 			ev.Component = event.Component_Sequence_Sender
@@ -217,7 +217,7 @@ func start(cliCtx *cli.Context) error {
 			if poolInstance == nil {
 				poolInstance = createPool(c.Pool, c.State.Batch.Constraints, l2ChainID, st, eventLog)
 			}
-			go runSynchronizer(*c, etherman, etm, st, poolInstance, eventLog)
+			go runSynchronizer(*c, etherman, ethTxManagerStorage, st, poolInstance, eventLog)
 		case ETHTXMANAGER:
 			ev.Component = event.Component_EthTxManager
 			ev.Description = "Running eth tx manager service"
@@ -285,7 +285,7 @@ func newEtherman(c config.Config) (*etherman.Client, error) {
 	return etherman, nil
 }
 
-func runSynchronizer(cfg config.Config, etherman *etherman.Client, ethTxManager *ethtxmanager.Client, st *state.State, pool *pool.Pool, eventLog *event.EventLog) {
+func runSynchronizer(cfg config.Config, etherman *etherman.Client, ethTxManagerStorage *ethtxmanager.PostgresStorage, st *state.State, pool *pool.Pool, eventLog *event.EventLog) {
 	var trustedSequencerURL string
 	var err error
 	if !cfg.IsTrustedSequencer {
@@ -313,8 +313,9 @@ func runSynchronizer(cfg config.Config, etherman *etherman.Client, ethTxManager
 			etherManForL1 = append(etherManForL1, eth)
 		}
 	}
+	etm := ethtxmanager.New(cfg.EthTxManager, etherman, ethTxManagerStorage, st)
 	sy, err := synchronizer.NewSynchronizer(
-		cfg.IsTrustedSequencer, etherman, etherManForL1, st, pool, ethTxManager,
+		cfg.IsTrustedSequencer, etherman, etherManForL1, st, pool, etm,
 		zkEVMClient, eventLog, cfg.NetworkConfig.Genesis, cfg.Synchronizer, cfg.Log.Environment == "development",
 	)
 	if err != nil {
@@ -389,15 +390,13 @@ func runJSONRPCServer(c config.Config, etherman *etherman.Client, chainID uint64
 	}
 }
 
-func createSequencer(cfg config.Config, pool *pool.Pool, etmStorage *ethtxmanager.PostgresStorage, st *state.State, eventLog *event.EventLog) *sequencer.Sequencer {
+func createSequencer(cfg config.Config, pool *pool.Pool, st *state.State, eventLog *event.EventLog) *sequencer.Sequencer {
 	etherman, err := newEtherman(cfg)
 	if err != nil {
 		log.Fatal(err)
 	}
 
-	ethTxManager := ethtxmanager.New(cfg.EthTxManager, etherman, etmStorage, st)
-
-	seq, err := sequencer.New(cfg.Sequencer, cfg.State.Batch, pool, st, etherman, ethTxManager, eventLog)
+	seq, err := sequencer.New(cfg.Sequencer, cfg.State.Batch, pool, st, etherman, eventLog)
 	if err != nil {
 		log.Fatal(err)
 	}

config/config_test.go

@@ -156,6 +156,10 @@ func Test_Defaults(t *testing.T) {
 			path:          "SequenceSender.MaxTxSizeForL1",
 			expectedValue: uint64(131072),
 		},
+		{
+			path:          "SequenceSender.GasOffset",
+			expectedValue: uint64(80000),
+		},
 		{
 			path:          "Etherman.URL",
 			expectedValue: "http://localhost:8545",
@@ -433,7 +437,10 @@ func Test_Defaults(t *testing.T) {
 			path:          "Aggregator.GeneratingProofCleanupThreshold",
 			expectedValue: "10m",
 		},
-
+		{
+			path:          "Aggregator.GasOffset",
+			expectedValue: uint64(0),
+		},
 		{
 			path:          "State.Batch.Constraints.MaxTxsPerBatch",
 			expectedValue: uint64(300),

config/default.go

@@ -144,6 +144,7 @@ LastBatchVirtualizationTimeMaxWaitPeriod = "5s"
 MaxTxSizeForL1 = 131072
 L2Coinbase = "0xf39fd6e51aad88f6f4ce6ab8827279cfffb92266"
 PrivateKey = {Path = "/pk/sequencer.keystore", Password = "testonly"}
+GasOffset = 80000
 
 [Aggregator]
 Host = "0.0.0.0"
@@ -155,6 +156,7 @@ TxProfitabilityMinReward = "1.1"
 ProofStatePollingInterval = "5s"
 CleanupLockedProofsInterval = "2m"
 GeneratingProofCleanupThreshold = "10m"
+GasOffset = 0
 
 [L2GasPriceSuggester]
 Type = "follower"

db/migrations/state/0012.sql

@@ -0,0 +1,8 @@
+-- +migrate Up
+ALTER TABLE state.monitored_txs
+    ADD COLUMN gas_offset DECIMAL(78, 0) NOT NULL DEFAULT 0;
+ALTER TABLE state.monitored_txs ALTER COLUMN gas_offset DROP DEFAULT;
+
+-- +migrate Down
+ALTER TABLE state.monitored_txs
+    DROP COLUMN gas_offset;

db/migrations/state/0012_test.go

@@ -0,0 +1,62 @@
+package migrations_test
+
+import (
+	"database/sql"
+	"testing"
+	"time"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/assert"
+)
+
+// this migration changes length of the token name
+type migrationTest0012 struct{}
+
+func (m migrationTest0012) InsertData(db *sql.DB) error {
+	addMonitoredTx := `
+        INSERT INTO state.monitored_txs (owner, id, from_addr, to_addr, nonce, value, data, gas, gas_price, status, history, block_num, created_at, updated_at) 
+                                VALUES (   $1, $2,        $3,      $4,    $5,    $6,   $7,  $8,        $9,    $10,     $11,       $12,        $13,        $14);`
+
+	args := []interface{}{
+		"owner", "id1", common.HexToAddress("0x111").String(), common.HexToAddress("0x222").String(), 333, 444,
+		[]byte{5, 5, 5}, 666, 777, "status", []string{common.HexToHash("0x888").String()}, 999, time.Now(), time.Now(),
+	}
+	if _, err := db.Exec(addMonitoredTx, args...); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+func (m migrationTest0012) RunAssertsAfterMigrationUp(t *testing.T, db *sql.DB) {
+	addMonitoredTx := `
+	INSERT INTO state.monitored_txs (owner, id, from_addr, to_addr, nonce, value, data, gas, gas_price, status, history, block_num, created_at, updated_at, gas_offset) 
+                            VALUES (   $1, $2,        $3,      $4,    $5,    $6,   $7,  $8,        $9,    $10,     $11,       $12,        $13,        $14,        $15);`
+
+	args := []interface{}{
+		"owner", "id2", common.HexToAddress("0x111").String(), common.HexToAddress("0x222").String(), 333, 444,
+		[]byte{5, 5, 5}, 666, 777, "status", []string{common.HexToHash("0x888").String()}, 999, time.Now(), time.Now(),
+		101010,
+	}
+	_, err := db.Exec(addMonitoredTx, args...)
+	assert.NoError(t, err)
+
+	gasOffset := 999
+
+	getGasOffsetQuery := `SELECT gas_offset FROM state.monitored_txs WHERE id = $1`
+	err = db.QueryRow(getGasOffsetQuery, "id1").Scan(&gasOffset)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, gasOffset)
+
+	err = db.QueryRow(getGasOffsetQuery, "id2").Scan(&gasOffset)
+	assert.NoError(t, err)
+	assert.Equal(t, 101010, gasOffset)
+}
+
+func (m migrationTest0012) RunAssertsAfterMigrationDown(t *testing.T, db *sql.DB) {
+
+}
+
+func TestMigration0012(t *testing.T) {
+	runMigrationTest(t, 12, migrationTest0012{})
+}