wait for n confirmations for a txn

churner/churner_test.go

@@ -24,6 +24,7 @@ func TestProcessChurnRequest(t *testing.T) {
 		LoggerConfig: logging.DefaultCLIConfig(),
 		EthClientConfig: geth.EthClientConfig{
 			PrivateKeyString: churnerPrivateKeyHex,
+			NumConfirmations: 0,
 		},
 	}
 	metrics := churner.NewMetrics("9001", logger)

churner/tests/churner_test.go

@@ -150,6 +150,7 @@ func createTransactorFromScratch(privateKey, operatorStateRetriever, serviceMana
 	ethClientCfg := geth.EthClientConfig{
 		RPCURL:           rpcURL,
 		PrivateKeyString: privateKey,
+		NumConfirmations: 0,
 	}
 
 	gethClient, err := geth.NewClient(ethClientCfg, logger)

common/geth/cli.go

@@ -6,13 +6,15 @@ import (
 )
 
 var (
-	rpcUrlFlagName     = "chain.rpc"
-	privateKeyFlagName = "chain.private-key"
+	rpcUrlFlagName           = "chain.rpc"
+	privateKeyFlagName       = "chain.private-key"
+	numConfirmationsFlagName = "chain.num-confirmations"
 )
 
 type EthClientConfig struct {
 	RPCURL           string
 	PrivateKeyString string
+	NumConfirmations int
 }
 
 func EthClientFlags(envPrefix string) []cli.Flag {
@@ -29,13 +31,21 @@ func EthClientFlags(envPrefix string) []cli.Flag {
 			Required: true,
 			EnvVar:   common.PrefixEnvVar(envPrefix, "PRIVATE_KEY"),
 		},
+		cli.IntFlag{
+			Name:     numConfirmationsFlagName,
+			Usage:    "Number of confirmations to wait for",
+			Required: false,
+			Value:    3,
+			EnvVar:   common.PrefixEnvVar(envPrefix, "NUM_CONFIRMATIONS"),
+		},
 	}
 }
 
 func ReadEthClientConfig(ctx *cli.Context) EthClientConfig {
 	cfg := EthClientConfig{}
 	cfg.RPCURL = ctx.GlobalString(rpcUrlFlagName)
 	cfg.PrivateKeyString = ctx.GlobalString(privateKeyFlagName)
+	cfg.NumConfirmations = ctx.GlobalInt(numConfirmationsFlagName)
 	return cfg
 }
 
@@ -44,5 +54,6 @@ func ReadEthClientConfig(ctx *cli.Context) EthClientConfig {
 func ReadEthClientConfigRPCOnly(ctx *cli.Context) EthClientConfig {
 	cfg := EthClientConfig{}
 	cfg.RPCURL = ctx.GlobalString(rpcUrlFlagName)
+	cfg.NumConfirmations = ctx.GlobalInt(numConfirmationsFlagName)
 	return cfg
 }

common/geth/client.go

@@ -6,6 +6,7 @@ import (
 	"errors"
 	"fmt"
 	"math/big"
+	"time"
 
 	"github.com/Layr-Labs/eigenda/common"
 	"github.com/ethereum/go-ethereum"
@@ -31,6 +32,7 @@ type EthClient struct {
 	NoSendTransactOpts *bind.TransactOpts
 	Contracts          map[gethcommon.Address]*bind.BoundContract
 	Logger             common.Logger
+	numConfirmations   int
 }
 
 var _ common.EthClient = (*EthClient)(nil)
@@ -72,12 +74,13 @@ func NewClient(config EthClientConfig, logger common.Logger) (*EthClient, error)
 	}
 
 	c := &EthClient{
-		RPCURL:         config.RPCURL,
-		privateKey:     privateKey,
-		AccountAddress: accountAddress,
-		Client:         chainClient,
-		Contracts:      make(map[gethcommon.Address]*bind.BoundContract),
-		Logger:         logger,
+		RPCURL:           config.RPCURL,
+		privateKey:       privateKey,
+		AccountAddress:   accountAddress,
+		Client:           chainClient,
+		Contracts:        make(map[gethcommon.Address]*bind.BoundContract),
+		Logger:           logger,
+		numConfirmations: config.NumConfirmations,
 	}
 
 	c.NoSendTransactOpts = opts
@@ -207,7 +210,7 @@ func (c *EthClient) EstimateGasPriceAndLimitAndSendTx(
 }
 
 func (c *EthClient) EnsureTransactionEvaled(ctx context.Context, tx *types.Transaction, tag string) (*types.Receipt, error) {
-	receipt, err := bind.WaitMined(ctx, c.Client, tx)
+	receipt, err := c.waitMined(ctx, tx)
 	if err != nil {
 		return nil, fmt.Errorf("EnsureTransactionEvaled: failed to wait for transaction (%s) to mine: %w", tag, err)
 	}
@@ -219,6 +222,43 @@ func (c *EthClient) EnsureTransactionEvaled(ctx context.Context, tx *types.Trans
 	return receipt, nil
 }
 
+// waitMined waits for tx to be mined on the blockchain.
+// Taken from https://github.com/ethereum/go-ethereum/blob/master/accounts/abi/bind/util.go#L32,
+// but added a check for number of confirmations.
+func (c *EthClient) waitMined(ctx context.Context, tx *types.Transaction) (*types.Receipt, error) {
+	queryTicker := time.NewTicker(6 * time.Second)
+	defer queryTicker.Stop()
+
+	for {
+		receipt, err := c.TransactionReceipt(ctx, tx.Hash())
+		if err == nil {
+			chainTip, err := c.BlockNumber(ctx)
+			if err == nil {
+				if receipt.BlockNumber.Uint64()+uint64(c.numConfirmations) > chainTip {
+					c.Logger.Trace("EnsureTransactionEvaled: transaction has been mined but don't have enough confirmations at current chain tip", "txnBlockNumber", receipt.BlockNumber.Uint64(), "numConfirmations", c.numConfirmations, "chainTip", chainTip)
+				} else {
+					return receipt, nil
+				}
+			} else {
+				c.Logger.Trace("EnsureTransactionEvaled: failed to get chain tip while waiting for transaction to mine", "err", err)
+			}
+		}
+
+		if errors.Is(err, ethereum.NotFound) {
+			c.Logger.Trace("Transaction not yet mined")
+		} else {
+			c.Logger.Trace("Receipt retrieval failed", "err", err)
+		}
+
+		// Wait for the next round.
+		select {
+		case <-ctx.Done():
+			return nil, ctx.Err()
+		case <-queryTicker.C:
+		}
+	}
+}
+
 // getGasFeeCap returns the gas fee cap for a transaction, calculated as:
 // gasFeeCap = 2 * baseFee + gasTipCap
 // Rationale: https://www.blocknative.com/blog/eip-1559-fees

core/eth/tx.go

@@ -384,9 +384,10 @@ func (t *Transactor) GetOperatorStakesForQuorums(ctx context.Context, quorums []
 	return state, nil
 }
 
-// ConfirmBatch confirms a batch header and signature aggregation. The signature aggregation must satisfy the quorum thresholds
+// BuildConfirmBatchTxn builds a transaction to confirm a batch header and signature aggregation. The signature aggregation must satisfy the quorum thresholds
 // specified in the batch header. If the signature aggregation does not satisfy the quorum thresholds, the transaction will fail.
-func (t *Transactor) ConfirmBatch(ctx context.Context, batchHeader core.BatchHeader, quorums map[core.QuorumID]*core.QuorumResult, signatureAggregation core.SignatureAggregation) (*types.Receipt, error) {
+// Note that this function returns a transaction without publishing it to the blockchain. The caller is responsible for publishing the transaction.
+func (t *Transactor) BuildConfirmBatchTxn(ctx context.Context, batchHeader core.BatchHeader, quorums map[core.QuorumID]*core.QuorumResult, signatureAggregation core.SignatureAggregation) (*types.Transaction, error) {
 	quorumNumbers := quorumParamsToQuorumNumbers(quorums)
 	nonSignerOperatorIds := make([][32]byte, len(signatureAggregation.NonSigners))
 	for i := range signatureAggregation.NonSigners {
@@ -441,9 +442,15 @@ func (t *Transactor) ConfirmBatch(ctx context.Context, batchHeader core.BatchHea
 		NonSignerStakeIndices:        checkSignaturesIndices.NonSignerStakeIndices,
 	}
 
-	tx, err := t.Bindings.EigenDAServiceManager.ConfirmBatch(t.EthClient.GetNoSendTransactOpts(), batchH, signatureChecker)
+	return t.Bindings.EigenDAServiceManager.ConfirmBatch(t.EthClient.GetNoSendTransactOpts(), batchH, signatureChecker)
+}
+
+// ConfirmBatch confirms a batch header and signature aggregation. The signature aggregation must satisfy the quorum thresholds
+// specified in the batch header. If the signature aggregation does not satisfy the quorum thresholds, the transaction will fail.
+func (t *Transactor) ConfirmBatch(ctx context.Context, batchHeader core.BatchHeader, quorums map[core.QuorumID]*core.QuorumResult, signatureAggregation core.SignatureAggregation) (*types.Receipt, error) {
+	tx, err := t.BuildConfirmBatchTxn(ctx, batchHeader, quorums, signatureAggregation)
 	if err != nil {
-		t.Logger.Error("Failed to confirm batch", "err", err)
+		t.Logger.Error("Failed to build a ConfirmBatch txn", "err", err)
 		return nil, err
 	}
 

core/indexer/state_test.go

@@ -67,6 +67,7 @@ func mustMakeOperatorTransactor(env *deploy.Config, op deploy.OperatorVars, logg
 	config := geth.EthClientConfig{
 		RPCURL:           deployer.RPC,
 		PrivateKeyString: op.NODE_PRIVATE_KEY,
+		NumConfirmations: 0,
 	}
 
 	c, err := geth.NewClient(config, logger)
@@ -86,6 +87,7 @@ func mustMakeTestClients(env *deploy.Config, privateKey string, logger common.Lo
 	config := geth.EthClientConfig{
 		RPCURL:           deployer.RPC,
 		PrivateKeyString: privateKey,
+		NumConfirmations: 0,
 	}
 
 	client, err := geth.NewClient(config, logger)

core/thegraph/state_integration_test.go

@@ -111,6 +111,7 @@ func mustMakeTestClient(t *testing.T, env *deploy.Config, privateKey string, log
 	config := geth.EthClientConfig{
 		RPCURL:           deployer.RPC,
 		PrivateKeyString: privateKey,
+		NumConfirmations: 0,
 	}
 
 	client, err := geth.NewClient(config, logger)

inabox/deploy/config.go

@@ -171,6 +171,7 @@ func (env *Config) generateDisperserVars(ind int, key, address, logPath, dbPath,
 		DISPERSER_SERVER_METRICS_HTTP_PORT:      "9093",
 		DISPERSER_SERVER_CHAIN_RPC:              "",
 		DISPERSER_SERVER_PRIVATE_KEY:            "123",
+		DISPERSER_SERVER_NUM_CONFIRMATIONS:      "0",
 
 		DISPERSER_SERVER_REGISTERED_QUORUM_ID:       "0",
 		DISPERSER_SERVER_TOTAL_UNAUTH_THROUGHPUT:    "10000000",
@@ -217,6 +218,7 @@ func (env *Config) generateBatcherVars(ind int, key, graphUrl, logPath string) B
 		BATCHER_AWS_ENDPOINT_URL:            "",
 		BATCHER_FINALIZER_INTERVAL:          "6m",
 		BATCHER_ENCODING_REQUEST_QUEUE_SIZE: "500",
+		BATCHER_NUM_CONFIRMATIONS:           "0",
 	}
 
 	env.applyDefaults(&v, "BATCHER", "batcher", ind)
@@ -301,6 +303,7 @@ func (env *Config) generateOperatorVars(ind int, name, key, churnerUrl, logPath,
 		NODE_NUM_BATCH_VALIDATORS:        "128",
 		NODE_PUBLIC_IP_PROVIDER:          "mockip",
 		NODE_PUBLIC_IP_CHECK_INTERVAL:    "10s",
+		NODE_NUM_CONFIRMATIONS:           "0",
 	}
 
 	env.applyDefaults(&v, "NODE", "opr", ind)

inabox/deploy/env_vars.go

@@ -27,6 +27,8 @@ type DisperserVars struct {
 
 	DISPERSER_SERVER_PRIVATE_KEY string
 
+	DISPERSER_SERVER_NUM_CONFIRMATIONS string
+
 	DISPERSER_SERVER_STD_LOG_LEVEL string
 
 	DISPERSER_SERVER_FILE_LOG_LEVEL string
@@ -114,6 +116,8 @@ type BatcherVars struct {
 
 	BATCHER_PRIVATE_KEY string
 
+	BATCHER_NUM_CONFIRMATIONS string
+
 	BATCHER_STD_LOG_LEVEL string
 
 	BATCHER_FILE_LOG_LEVEL string
@@ -262,6 +266,8 @@ type OperatorVars struct {
 
 	NODE_PRIVATE_KEY string
 
+	NODE_NUM_CONFIRMATIONS string
+
 	NODE_STD_LOG_LEVEL string
 
 	NODE_FILE_LOG_LEVEL string
@@ -315,6 +321,8 @@ type RetrieverVars struct {
 
 	RETRIEVER_PRIVATE_KEY string
 
+	RETRIEVER_NUM_CONFIRMATIONS string
+
 	RETRIEVER_STD_LOG_LEVEL string
 
 	RETRIEVER_FILE_LOG_LEVEL string
@@ -344,23 +352,25 @@ type ChurnerVars struct {
 
 	CHURNER_EIGENDA_SERVICE_MANAGER string
 
+	CHURNER_ENABLE_METRICS string
+
 	CHURNER_PER_PUBLIC_KEY_RATE_LIMIT string
 
+	CHURNER_METRICS_HTTP_PORT string
+
 	CHURNER_CHAIN_RPC string
 
 	CHURNER_PRIVATE_KEY string
 
+	CHURNER_NUM_CONFIRMATIONS string
+
 	CHURNER_STD_LOG_LEVEL string
 
 	CHURNER_FILE_LOG_LEVEL string
 
 	CHURNER_LOG_PATH string
 
 	CHURNER_INDEXER_PULL_INTERVAL string
-
-	CHURNER_ENABLE_METRICS string
-
-	CHURNER_METRICS_HTTP_PORT string
 }
 
 func (vars ChurnerVars) getEnvMap() map[string]string {

inabox/tests/integration_suite_test.go

@@ -20,6 +20,7 @@ import (
 	"github.com/Layr-Labs/eigenda/inabox/deploy"
 	"github.com/Layr-Labs/eigenda/pkg/encoding/kzgEncoder"
 	gcommon "github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/rpc"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
 	"github.com/ory/dockertest/v3"
@@ -40,6 +41,7 @@ var (
 	bucketTableName   = "test-BucketStore"
 	logger            common.Logger
 	ethClient         common.EthClient
+	rpcClient         common.RPCEthClient
 	mockRollup        *rollupbindings.ContractMockRollup
 	retrievalClient   clients.RetrievalClient
 )
@@ -112,8 +114,11 @@ var _ = BeforeSuite(func() {
 	ethClient, err = geth.NewClient(geth.EthClientConfig{
 		RPCURL:           testConfig.Deployers[0].RPC,
 		PrivateKeyString: pk,
+		NumConfirmations: 3,
 	}, logger)
 	Expect(err).To(BeNil())
+	rpcClient, err = rpc.Dial(testConfig.Deployers[0].RPC)
+	Expect(err).To(BeNil())
 	mockRollup, err = rollupbindings.NewContractMockRollup(gcommon.HexToAddress(testConfig.MockRollup), ethClient)
 	Expect(err).To(BeNil())
 	err = setupRetrievalClient(testConfig)
@@ -124,6 +129,7 @@ func setupRetrievalClient(testConfig *deploy.Config) error {
 	ethClientConfig := geth.EthClientConfig{
 		RPCURL:           testConfig.Deployers[0].RPC,
 		PrivateKeyString: "351b8eca372e64f64d514f90f223c5c4f86a04ff3dcead5c27293c547daab4ca", // just random private key
+		NumConfirmations: 3,
 	}
 	client, err := geth.NewClient(ethClientConfig, logger)
 	if err != nil {

inabox/tests/integration_test.go

@@ -30,6 +30,10 @@ var _ = Describe("Inabox Integration", func() {
 		Expect(err).To(BeNil())
 		_, err = ethClient.EstimateGasPriceAndLimitAndSendTx(ctx, tx, "RegisterValidator", big.NewInt(1e18))
 		Expect(err).To(BeNil())
+		for i := 0; i < 3; i++ {
+			err = rpcClient.CallContext(ctx, nil, "evm_mine")
+			Expect(err).To(BeNil())
+		}
 
 		disp := traffic.NewDisperserClient(&traffic.Config{
 			Hostname:        "localhost",
@@ -81,6 +85,11 @@ var _ = Describe("Inabox Integration", func() {
 					_, err = ethClient.EstimateGasPriceAndLimitAndSendTx(ctx, tx, "PostCommitment", nil)
 					Expect(err).To(BeNil())
 					break loop
+				} else {
+					for i := 0; i < 3; i++ {
+						err = rpcClient.CallContext(ctx, nil, "evm_mine")
+						Expect(err).To(BeNil())
+					}
 				}
 			}
 		}

node/plugin/cmd/main.go

@@ -47,6 +47,7 @@ func main() {
 		plugin.BlsOperatorStateRetrieverFlag,
 		plugin.EigenDAServiceManagerFlag,
 		plugin.ChurnerUrlFlag,
+		plugin.NumConfirmationsFlag,
 	}
 	app.Name = "eigenda-node-plugin"
 	app.Usage = "EigenDA Node Plugin"
@@ -98,6 +99,7 @@ func pluginOps(ctx *cli.Context) {
 	ethConfig := geth.EthClientConfig{
 		RPCURL:           config.ChainRpcUrl,
 		PrivateKeyString: *privateKey,
+		NumConfirmations: config.NumConfirmations,
 	}
 	client, err := geth.NewClient(ethConfig, logger)
 	if err != nil {