Merge pull request #2041 from cfromknecht/bss-num-confirmations

feat: add confirmation depth awareness to txmgr

.changeset/fuzzy-rivers-double.md

+---
+'@eth-optimism/batch-submitter-service': patch
+---
+
+Adds confirmation depth awareness to txmgr

go/batch-submitter/batch_submitter.go

@@ -164,6 +164,7 @@ func NewBatchSubmitter(cfg Config, gitVersion string) (*BatchSubmitter, error) {
 		GasRetryIncrement:    utils.GasPriceFromGwei(cfg.GasRetryIncrement),
 		ResubmissionTimeout:  cfg.ResubmissionTimeout,
 		ReceiptQueryInterval: time.Second,
+		NumConfirmations:     cfg.NumConfirmations,
 	}
 
 	var batchTxService *Service

go/batch-submitter/drivers/clear_pending_tx_test.go

@@ -32,12 +32,13 @@ func init() {
 }
 
 var (
-	testPrivKey    *ecdsa.PrivateKey
-	testWalletAddr common.Address
-	testChainID    *big.Int // 1
-	testNonce      = uint64(2)
-	testGasPrice   *big.Int // 3
-	testGasLimit   = uint64(4)
+	testPrivKey     *ecdsa.PrivateKey
+	testWalletAddr  common.Address
+	testChainID     = big.NewInt(1)
+	testNonce       = uint64(2)
+	testGasPrice    = big.NewInt(3)
+	testGasLimit    = uint64(4)
+	testBlockNumber = uint64(5)
 )
 
 // TestCraftClearingTx asserts that CraftClearingTx produces the expected
@@ -102,11 +103,20 @@ func TestSignClearingTxEstimateGasFail(t *testing.T) {
 }
 
 type clearPendingTxHarness struct {
-	l1Client drivers.L1Client
+	l1Client *mock.L1Client
 	txMgr    txmgr.TxManager
 }
 
-func newClearPendingTxHarness(l1ClientConfig mock.L1ClientConfig) *clearPendingTxHarness {
+func newClearPendingTxHarnessWithNumConfs(
+	l1ClientConfig mock.L1ClientConfig,
+	numConfirmations uint64,
+) *clearPendingTxHarness {
+
+	if l1ClientConfig.BlockNumber == nil {
+		l1ClientConfig.BlockNumber = func(_ context.Context) (uint64, error) {
+			return testBlockNumber, nil
+		}
+	}
 	if l1ClientConfig.NonceAt == nil {
 		l1ClientConfig.NonceAt = func(_ context.Context, _ common.Address, _ *big.Int) (uint64, error) {
 			return testNonce, nil
@@ -125,6 +135,7 @@ func newClearPendingTxHarness(l1ClientConfig mock.L1ClientConfig) *clearPendingT
 		GasRetryIncrement:    utils.GasPriceFromGwei(5),
 		ResubmissionTimeout:  time.Second,
 		ReceiptQueryInterval: 50 * time.Millisecond,
+		NumConfirmations:     numConfirmations,
 	}, l1Client)
 
 	return &clearPendingTxHarness{
@@ -133,6 +144,10 @@ func newClearPendingTxHarness(l1ClientConfig mock.L1ClientConfig) *clearPendingT
 	}
 }
 
+func newClearPendingTxHarness(l1ClientConfig mock.L1ClientConfig) *clearPendingTxHarness {
+	return newClearPendingTxHarnessWithNumConfs(l1ClientConfig, 1)
+}
+
 // TestClearPendingTxClearingTxÇonfirms asserts the happy path where our
 // clearing transactions confirms unobstructed.
 func TestClearPendingTxClearingTxConfirms(t *testing.T) {
@@ -142,7 +157,8 @@ func TestClearPendingTxClearingTxConfirms(t *testing.T) {
 		},
 		TransactionReceipt: func(_ context.Context, txHash common.Hash) (*types.Receipt, error) {
 			return &types.Receipt{
-				TxHash: txHash,
+				TxHash:      txHash,
+				BlockNumber: big.NewInt(int64(testBlockNumber)),
 			}, nil
 		},
 	})
@@ -190,3 +206,42 @@ func TestClearPendingTxTimeout(t *testing.T) {
 	)
 	require.Equal(t, txmgr.ErrPublishTimeout, err)
 }
+
+// TestClearPendingTxMultipleConfs tests we wait the appropriate number of
+// confirmations for the clearing transaction to confirm.
+func TestClearPendingTxMultipleConfs(t *testing.T) {
+	const numConfs = 2
+
+	// Instantly confirm transaction.
+	h := newClearPendingTxHarnessWithNumConfs(mock.L1ClientConfig{
+		SendTransaction: func(_ context.Context, _ *types.Transaction) error {
+			return nil
+		},
+		TransactionReceipt: func(_ context.Context, txHash common.Hash) (*types.Receipt, error) {
+			return &types.Receipt{
+				TxHash:      txHash,
+				BlockNumber: big.NewInt(int64(testBlockNumber)),
+			}, nil
+		},
+	}, numConfs)
+
+	// The txmgr should timeout waiting for the txn to confirm.
+	err := drivers.ClearPendingTx(
+		"test", context.Background(), h.txMgr, h.l1Client, testWalletAddr,
+		testPrivKey, testChainID,
+	)
+	require.Equal(t, txmgr.ErrPublishTimeout, err)
+
+	// Now set the chain height to the earliest the transaction will be
+	// considered sufficiently confirmed.
+	h.l1Client.SetBlockNumberFunc(func(_ context.Context) (uint64, error) {
+		return testBlockNumber + numConfs - 1, nil
+	})
+
+	// Publishing should succeed.
+	err = drivers.ClearPendingTx(
+		"test", context.Background(), h.txMgr, h.l1Client, testWalletAddr,
+		testPrivKey, testChainID,
+	)
+	require.Nil(t, err)
+}

go/batch-submitter/mock/l1client.go

@@ -13,6 +13,9 @@ import (
 // L1ClientConfig houses the internal methods that are executed by the mock
 // L1Client. Any members left as nil will panic on execution.
 type L1ClientConfig struct {
+	// BlockNumber returns the most recent block number.
+	BlockNumber func(context.Context) (uint64, error)
+
 	// EstimateGas tries to estimate the gas needed to execute a specific
 	// transaction based on the current pending state of the backend blockchain.
 	// There is no guarantee that this is the true gas limit requirement as
@@ -50,6 +53,14 @@ func NewL1Client(cfg L1ClientConfig) *L1Client {
 	}
 }
 
+// BlockNumber returns the most recent block number.
+func (c *L1Client) BlockNumber(ctx context.Context) (uint64, error) {
+	c.mu.RLock()
+	defer c.mu.RUnlock()
+
+	return c.cfg.BlockNumber(ctx)
+}
+
 // EstimateGas executes the mock EstimateGas method.
 func (c *L1Client) EstimateGas(ctx context.Context, call ethereum.CallMsg) (uint64, error) {
 	c.mu.RLock()
@@ -82,6 +93,16 @@ func (c *L1Client) TransactionReceipt(ctx context.Context, txHash common.Hash) (
 	return c.cfg.TransactionReceipt(ctx, txHash)
 }
 
+// SetBlockNumberFunc overwrites the mock BlockNumber method.
+func (c *L1Client) SetBlockNumberFunc(
+	f func(context.Context) (uint64, error)) {
+
+	c.mu.Lock()
+	defer c.mu.Unlock()
+
+	c.cfg.BlockNumber = f
+}
+
 // SetEstimateGasFunc overrwrites the mock EstimateGas method.
 func (c *L1Client) SetEstimateGasFunc(
 	f func(context.Context, ethereum.CallMsg) (uint64, error)) {

go/batch-submitter/txmgr/txmgr.go

@@ -52,6 +52,10 @@ type Config struct {
 	// query the backend to check for confirmations after a tx at a
 	// specific gas price has been published.
 	ReceiptQueryInterval time.Duration
+
+	// NumConfirmations specifies how many blocks are need to consider a
+	// transaction confirmed.
+	NumConfirmations uint64
 }
 
 // TxManager is an interface that allows callers to reliably publish txs,
@@ -71,6 +75,9 @@ type TxManager interface {
 //
 // NOTE: This is a subset of bind.DeployBackend.
 type ReceiptSource interface {
+	// BlockNumber returns the most recent block number.
+	BlockNumber(ctx context.Context) (uint64, error)
+
 	// TransactionReceipt queries the backend for a receipt associated with
 	// txHash. If lookup does not fail, but the transaction is not found,
 	// nil should be returned for both values.
@@ -90,6 +97,10 @@ type SimpleTxManager struct {
 func NewSimpleTxManager(
 	name string, cfg Config, backend ReceiptSource) *SimpleTxManager {
 
+	if cfg.NumConfirmations == 0 {
+		panic("txmgr: NumConfirmations cannot be zero")
+	}
+
 	return &SimpleTxManager{
 		name:    name,
 		cfg:     cfg,
@@ -148,6 +159,7 @@ func (m *SimpleTxManager) Send(
 		// back to the main event loop if found.
 		receipt, err := WaitMined(
 			ctxc, m.backend, tx, m.cfg.ReceiptQueryInterval,
+			m.cfg.NumConfirmations,
 		)
 		if err != nil {
 			log.Debug(name+" send tx failed", "hash", txHash,
@@ -220,6 +232,7 @@ func WaitMined(
 	backend ReceiptSource,
 	tx *types.Transaction,
 	queryInterval time.Duration,
+	numConfirmations uint64,
 ) (*types.Receipt, error) {
 
 	queryTicker := time.NewTicker(queryInterval)
@@ -229,14 +242,42 @@ func WaitMined(
 
 	for {
 		receipt, err := backend.TransactionReceipt(ctx, txHash)
-		if receipt != nil {
-			return receipt, nil
-		}
+		switch {
+		case receipt != nil:
+			txHeight := receipt.BlockNumber.Uint64()
+			tipHeight, err := backend.BlockNumber(ctx)
+			if err != nil {
+				log.Error("Unable to fetch block number", "err", err)
+				break
+			}
 
-		if err != nil {
+			log.Trace("Transaction mined, checking confirmations",
+				"txHash", txHash, "txHeight", txHeight,
+				"tipHeight", tipHeight,
+				"numConfirmations", numConfirmations)
+
+			// The transaction is considered confirmed when
+			// txHeight+numConfirmations-1 <= tipHeight. Note that the -1 is
+			// needed to account for the fact that confirmations have an
+			// inherent off-by-one, i.e. when using 1 confirmation the
+			// transaction should be confirmed when txHeight is equal to
+			// tipHeight. The equation is rewritten in this form to avoid
+			// underflows.
+			if txHeight+numConfirmations <= tipHeight+1 {
+				log.Info("Transaction confirmed", "txHash", txHash)
+				return receipt, nil
+			}
+
+			// Safe to subtract since we know the LHS above is greater.
+			confsRemaining := (txHeight + numConfirmations) - (tipHeight + 1)
+			log.Info("Transaction not yet confirmed", "txHash", txHash,
+				"confsRemaining", confsRemaining)
+
+		case err != nil:
 			log.Trace("Receipt retrievel failed", "hash", txHash,
 				"err", err)
-		} else {
+
+		default:
 			log.Trace("Transaction not yet mined", "hash", txHash)
 		}
 

go/batch-submitter/txmgr/txmgr_test.go

@@ -95,13 +95,23 @@ func newTestHarnessWithConfig(cfg txmgr.Config) *testHarness {
 // newTestHarness initializes a testHarness with a defualt configuration that is
 // suitable for most tests.
 func newTestHarness() *testHarness {
-	return newTestHarnessWithConfig(txmgr.Config{
+	return newTestHarnessWithConfig(configWithNumConfs(1))
+}
+
+func configWithNumConfs(numConfirmations uint64) txmgr.Config {
+	return txmgr.Config{
 		MinGasPrice:          new(big.Int).SetUint64(5),
 		MaxGasPrice:          new(big.Int).SetUint64(50),
 		GasRetryIncrement:    new(big.Int).SetUint64(5),
 		ResubmissionTimeout:  time.Second,
 		ReceiptQueryInterval: 50 * time.Millisecond,
-	})
+		NumConfirmations:     numConfirmations,
+	}
+}
+
+type minedTxInfo struct {
+	gasPrice    *big.Int
+	blockNumber uint64
 }
 
 // mockBackend implements txmgr.ReceiptSource that tracks mined transactions
@@ -109,25 +119,42 @@ func newTestHarness() *testHarness {
 type mockBackend struct {
 	mu sync.RWMutex
 
-	// txHashMinedWithGasPrice tracks the has of a mined transaction to its
-	// gas price.
-	txHashMinedWithGasPrice map[common.Hash]*big.Int
+	// blockHeight tracks the current height of the chain.
+	blockHeight uint64
+
+	// minedTxs maps the hash of a mined transaction to its details.
+	minedTxs map[common.Hash]minedTxInfo
 }
 
 // newMockBackend initializes a new mockBackend.
 func newMockBackend() *mockBackend {
 	return &mockBackend{
-		txHashMinedWithGasPrice: make(map[common.Hash]*big.Int),
+		minedTxs: make(map[common.Hash]minedTxInfo),
 	}
 }
 
 // mine records a (txHash, gasPrice) as confirmed. Subsequent calls to
 // TransactionReceipt with a matching txHash will result in a non-nil receipt.
-func (b *mockBackend) mine(txHash common.Hash, gasPrice *big.Int) {
+// If a nil txHash is supplied this has the effect of mining an empty block.
+func (b *mockBackend) mine(txHash *common.Hash, gasPrice *big.Int) {
 	b.mu.Lock()
 	defer b.mu.Unlock()
 
-	b.txHashMinedWithGasPrice[txHash] = gasPrice
+	b.blockHeight++
+	if txHash != nil {
+		b.minedTxs[*txHash] = minedTxInfo{
+			gasPrice:    gasPrice,
+			blockNumber: b.blockHeight,
+		}
+	}
+}
+
+// BlockNumber returns the most recent block number.
+func (b *mockBackend) BlockNumber(ctx context.Context) (uint64, error) {
+	b.mu.RLock()
+	defer b.mu.RUnlock()
+
+	return b.blockHeight, nil
 }
 
 // TransactionReceipt queries the mockBackend for a mined txHash. If none is
@@ -142,7 +169,7 @@ func (b *mockBackend) TransactionReceipt(
 	b.mu.RLock()
 	defer b.mu.RUnlock()
 
-	gasPrice, ok := b.txHashMinedWithGasPrice[txHash]
+	txInfo, ok := b.minedTxs[txHash]
 	if !ok {
 		return nil, nil
 	}
@@ -150,8 +177,9 @@ func (b *mockBackend) TransactionReceipt(
 	// Return the gas price for the transaction in the GasUsed field so that
 	// we can assert the proper tx confirmed in our tests.
 	return &types.Receipt{
-		TxHash:  txHash,
-		GasUsed: gasPrice.Uint64(),
+		TxHash:      txHash,
+		GasUsed:     txInfo.gasPrice.Uint64(),
+		BlockNumber: big.NewInt(int64(txInfo.blockNumber)),
 	}, nil
 }
 
@@ -168,7 +196,8 @@ func TestTxMgrConfirmAtMinGasPrice(t *testing.T) {
 		tx := types.NewTx(&types.LegacyTx{
 			GasPrice: gasPrice,
 		})
-		h.backend.mine(tx.Hash(), gasPrice)
+		txHash := tx.Hash()
+		h.backend.mine(&txHash, gasPrice)
 		return tx, nil
 	}
 
@@ -220,7 +249,8 @@ func TestTxMgrConfirmsAtMaxGasPrice(t *testing.T) {
 			GasPrice: gasPrice,
 		})
 		if gasPrice.Cmp(h.cfg.MaxGasPrice) == 0 {
-			h.backend.mine(tx.Hash(), gasPrice)
+			txHash := tx.Hash()
+			h.backend.mine(&txHash, gasPrice)
 		}
 		return tx, nil
 	}
@@ -252,7 +282,8 @@ func TestTxMgrConfirmsAtMaxGasPriceDelayed(t *testing.T) {
 		// should still return an error beforehand.
 		if gasPrice.Cmp(h.cfg.MaxGasPrice) == 0 {
 			time.AfterFunc(2*time.Second, func() {
-				h.backend.mine(tx.Hash(), gasPrice)
+				txHash := tx.Hash()
+				h.backend.mine(&txHash, gasPrice)
 			})
 		}
 		return tx, nil
@@ -308,7 +339,8 @@ func TestTxMgrOnlyOnePublicationSucceeds(t *testing.T) {
 		tx := types.NewTx(&types.LegacyTx{
 			GasPrice: gasPrice,
 		})
-		h.backend.mine(tx.Hash(), gasPrice)
+		txHash := tx.Hash()
+		h.backend.mine(&txHash, gasPrice)
 		return tx, nil
 	}
 
@@ -338,7 +370,8 @@ func TestTxMgrConfirmsMinGasPriceAfterBumping(t *testing.T) {
 		// Delay mining the tx with the min gas price.
 		if gasPrice.Cmp(h.cfg.MinGasPrice) == 0 {
 			time.AfterFunc(5*time.Second, func() {
-				h.backend.mine(tx.Hash(), gasPrice)
+				txHash := tx.Hash()
+				h.backend.mine(&txHash, gasPrice)
 			})
 		}
 		return tx, nil
@@ -361,10 +394,10 @@ func TestWaitMinedReturnsReceiptOnFirstSuccess(t *testing.T) {
 	// Create a tx and mine it immediately using the default backend.
 	tx := types.NewTx(&types.LegacyTx{})
 	txHash := tx.Hash()
-	h.backend.mine(txHash, new(big.Int))
+	h.backend.mine(&txHash, new(big.Int))
 
 	ctx := context.Background()
-	receipt, err := txmgr.WaitMined(ctx, h.backend, tx, 50*time.Millisecond)
+	receipt, err := txmgr.WaitMined(ctx, h.backend, tx, 50*time.Millisecond, 1)
 	require.Nil(t, err)
 	require.NotNil(t, receipt)
 	require.Equal(t, receipt.TxHash, txHash)
@@ -383,16 +416,73 @@ func TestWaitMinedCanBeCanceled(t *testing.T) {
 	// Create an unimined tx.
 	tx := types.NewTx(&types.LegacyTx{})
 
-	receipt, err := txmgr.WaitMined(ctx, h.backend, tx, 50*time.Millisecond)
+	receipt, err := txmgr.WaitMined(ctx, h.backend, tx, 50*time.Millisecond, 1)
 	require.Equal(t, err, context.DeadlineExceeded)
 	require.Nil(t, receipt)
 }
 
+// TestWaitMinedMultipleConfs asserts that WaitMiend will properly wait for more
+// than one confirmation.
+func TestWaitMinedMultipleConfs(t *testing.T) {
+	t.Parallel()
+
+	const numConfs = 2
+
+	h := newTestHarnessWithConfig(configWithNumConfs(numConfs))
+	ctxt, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	// Create an unimined tx.
+	tx := types.NewTx(&types.LegacyTx{})
+	txHash := tx.Hash()
+	h.backend.mine(&txHash, new(big.Int))
+
+	receipt, err := txmgr.WaitMined(ctxt, h.backend, tx, 50*time.Millisecond, numConfs)
+	require.Equal(t, err, context.DeadlineExceeded)
+	require.Nil(t, receipt)
+
+	ctxt, cancel = context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+
+	// Mine an empty block, tx should now be confirmed.
+	h.backend.mine(nil, nil)
+	receipt, err = txmgr.WaitMined(ctxt, h.backend, tx, 50*time.Millisecond, numConfs)
+	require.Nil(t, err)
+	require.NotNil(t, receipt)
+	require.Equal(t, txHash, receipt.TxHash)
+}
+
+// TestManagerPanicOnZeroConfs ensures that the NewSimpleTxManager will panic
+// when attempting to configure with NumConfirmations set to zero.
+func TestManagerPanicOnZeroConfs(t *testing.T) {
+	t.Parallel()
+
+	defer func() {
+		if r := recover(); r == nil {
+			t.Fatal("NewSimpleTxManager should panic when using zero conf")
+		}
+	}()
+
+	_ = newTestHarnessWithConfig(configWithNumConfs(0))
+}
+
 // failingBackend implements txmgr.ReceiptSource, returning a failure on the
 // first call but a success on the second call. This allows us to test that the
 // inner loop of WaitMined properly handles this case.
 type failingBackend struct {
-	returnSuccess bool
+	returnSuccessBlockNumber bool
+	returnSuccessReceipt     bool
+}
+
+// BlockNumber for the failingBackend returns errRpcFailure on the first
+// invocation and a fixed block height on subsequent calls.
+func (b *failingBackend) BlockNumber(ctx context.Context) (uint64, error) {
+	if !b.returnSuccessBlockNumber {
+		b.returnSuccessBlockNumber = true
+		return 0, errRpcFailure
+	}
+
+	return 1, nil
 }
 
 // TransactionReceipt for the failingBackend returns errRpcFailure on the first
@@ -400,13 +490,14 @@ type failingBackend struct {
 func (b *failingBackend) TransactionReceipt(
 	ctx context.Context, txHash common.Hash) (*types.Receipt, error) {
 
-	if !b.returnSuccess {
-		b.returnSuccess = true
+	if !b.returnSuccessReceipt {
+		b.returnSuccessReceipt = true
 		return nil, errRpcFailure
 	}
 
 	return &types.Receipt{
-		TxHash: txHash,
+		TxHash:      txHash,
+		BlockNumber: big.NewInt(1),
 	}, nil
 }
 
@@ -424,7 +515,7 @@ func TestWaitMinedReturnsReceiptAfterFailure(t *testing.T) {
 	txHash := tx.Hash()
 
 	ctx := context.Background()
-	receipt, err := txmgr.WaitMined(ctx, &borkedBackend, tx, 50*time.Millisecond)
+	receipt, err := txmgr.WaitMined(ctx, &borkedBackend, tx, 50*time.Millisecond, 1)
 	require.Nil(t, err)
 	require.NotNil(t, receipt)
 	require.Equal(t, receipt.TxHash, txHash)