feat(op-service): Move Tx Creation to the txmgr

op-batcher/batcher/config.go

@@ -31,6 +31,15 @@ type Config struct {
 	From         common.Address
 
 	TxManagerConfig txmgr.Config
+	NetworkTimeout  time.Duration
+
+	// TxManagerTimeout is the maximum amount of time
+	// the driver should wait for the [txmgr] to send a transaction.
+	TxManagerTimeout time.Duration
+
+	// OfflineGasEstimation specifies whether the batcher should calculate
+	// gas estimations offline using the [core.IntrinsicGas] function.
+	OfflineGasEstimation bool
 
 	// RollupConfig is queried at startup
 	Rollup *rollup.Config
@@ -110,6 +119,14 @@ type CLIConfig struct {
 
 	/* Optional Params */
 
+	// TxManagerTimeout is the max amount of time to wait for the [txmgr].
+	// This will default to: 10 * time.Minute.
+	TxManagerTimeout time.Duration
+
+	// OfflineGasEstimation specifies whether the batcher should calculate
+	// gas estimations offline using the [core.IntrinsicGas] function.
+	OfflineGasEstimation bool
+
 	// MaxL1TxSize is the maximum size of a batch tx submitted to L1.
 	MaxL1TxSize uint64
 
@@ -168,19 +185,21 @@ func NewConfig(ctx *cli.Context) CLIConfig {
 		ResubmissionTimeout:       ctx.GlobalDuration(flags.ResubmissionTimeoutFlag.Name),
 
 		/* Optional Flags */
-		MaxChannelDuration: ctx.GlobalUint64(flags.MaxChannelDurationFlag.Name),
-		MaxL1TxSize:        ctx.GlobalUint64(flags.MaxL1TxSizeBytesFlag.Name),
-		TargetL1TxSize:     ctx.GlobalUint64(flags.TargetL1TxSizeBytesFlag.Name),
-		TargetNumFrames:    ctx.GlobalInt(flags.TargetNumFramesFlag.Name),
-		ApproxComprRatio:   ctx.GlobalFloat64(flags.ApproxComprRatioFlag.Name),
-		Stopped:            ctx.GlobalBool(flags.StoppedFlag.Name),
-		Mnemonic:           ctx.GlobalString(flags.MnemonicFlag.Name),
-		SequencerHDPath:    ctx.GlobalString(flags.SequencerHDPathFlag.Name),
-		PrivateKey:         ctx.GlobalString(flags.PrivateKeyFlag.Name),
-		RPCConfig:          rpc.ReadCLIConfig(ctx),
-		LogConfig:          oplog.ReadCLIConfig(ctx),
-		MetricsConfig:      opmetrics.ReadCLIConfig(ctx),
-		PprofConfig:        oppprof.ReadCLIConfig(ctx),
-		SignerConfig:       opsigner.ReadCLIConfig(ctx),
+		OfflineGasEstimation: ctx.GlobalBool(flags.OfflineGasEstimationFlag.Name),
+		TxManagerTimeout:     ctx.GlobalDuration(flags.TxManagerTimeoutFlag.Name),
+		MaxChannelDuration:   ctx.GlobalUint64(flags.MaxChannelDurationFlag.Name),
+		MaxL1TxSize:          ctx.GlobalUint64(flags.MaxL1TxSizeBytesFlag.Name),
+		TargetL1TxSize:       ctx.GlobalUint64(flags.TargetL1TxSizeBytesFlag.Name),
+		TargetNumFrames:      ctx.GlobalInt(flags.TargetNumFramesFlag.Name),
+		ApproxComprRatio:     ctx.GlobalFloat64(flags.ApproxComprRatioFlag.Name),
+		Stopped:              ctx.GlobalBool(flags.StoppedFlag.Name),
+		Mnemonic:             ctx.GlobalString(flags.MnemonicFlag.Name),
+		SequencerHDPath:      ctx.GlobalString(flags.SequencerHDPathFlag.Name),
+		PrivateKey:           ctx.GlobalString(flags.PrivateKeyFlag.Name),
+		RPCConfig:            rpc.ReadCLIConfig(ctx),
+		LogConfig:            oplog.ReadCLIConfig(ctx),
+		MetricsConfig:        opmetrics.ReadCLIConfig(ctx),
+		PprofConfig:          oppprof.ReadCLIConfig(ctx),
+		SignerConfig:         opsigner.ReadCLIConfig(ctx),
 	}
 }

op-batcher/batcher/driver.go

@@ -15,6 +15,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	opcrypto "github.com/ethereum-optimism/optimism/op-service/crypto"
 	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+	"github.com/ethereum/go-ethereum/core"
 	"github.com/ethereum/go-ethereum/core/types"
 	"github.com/ethereum/go-ethereum/log"
 )
@@ -24,7 +25,7 @@ import (
 type BatchSubmitter struct {
 	Config // directly embed the config + sources
 
-	txMgr *TransactionManager
+	txMgr txmgr.TxManager
 	wg    sync.WaitGroup
 	done  chan struct{}
 
@@ -83,13 +84,15 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metri
 	}
 
 	batcherCfg := Config{
-		L1Client:        l1Client,
-		L2Client:        l2Client,
-		RollupNode:      rollupClient,
-		PollInterval:    cfg.PollInterval,
-		TxManagerConfig: txManagerConfig,
-		From:            fromAddress,
-		Rollup:          rcfg,
+		L1Client:             l1Client,
+		L2Client:             l2Client,
+		RollupNode:           rollupClient,
+		PollInterval:         cfg.PollInterval,
+		TxManagerConfig:      txManagerConfig,
+		TxManagerTimeout:     cfg.TxManagerTimeout,
+		OfflineGasEstimation: cfg.OfflineGasEstimation,
+		From:                 fromAddress,
+		Rollup:               rcfg,
 		Channel: ChannelConfig{
 			SeqWindowSize:      rcfg.SeqWindowSize,
 			ChannelTimeout:     rcfg.ChannelTimeout,
@@ -125,10 +128,8 @@ func NewBatchSubmitter(ctx context.Context, cfg Config, l log.Logger, m metrics.
 
 	return &BatchSubmitter{
 		Config: cfg,
-		txMgr: NewTransactionManager(l,
-			cfg.TxManagerConfig, cfg.Rollup.BatchInboxAddress, cfg.Rollup.L1ChainID,
-			cfg.From, cfg.L1Client),
-		state: NewChannelManager(l, m, cfg.Channel),
+		txMgr:  txmgr.NewSimpleTxManager("batcher", l, cfg.TxManagerConfig, cfg.L1Client),
+		state:  NewChannelManager(l, m, cfg.Channel),
 	}, nil
 
 }
@@ -226,7 +227,7 @@ func (l *BatchSubmitter) loadBlocksIntoState(ctx context.Context) {
 
 // loadBlockIntoState fetches & stores a single block into `state`. It returns the block it loaded.
 func (l *BatchSubmitter) loadBlockIntoState(ctx context.Context, blockNumber uint64) (*types.Block, error) {
-	ctx, cancel := context.WithTimeout(ctx, networkTimeout)
+	ctx, cancel := context.WithTimeout(ctx, l.NetworkTimeout)
 	defer cancel()
 	block, err := l.L2Client.BlockByNumber(ctx, new(big.Int).SetUint64(blockNumber))
 	if err != nil {
@@ -244,7 +245,7 @@ func (l *BatchSubmitter) loadBlockIntoState(ctx context.Context, blockNumber uin
 // calculateL2BlockRangeToStore determines the range (start,end] that should be loaded into the local state.
 // It also takes care of initializing some local state (i.e. will modify l.lastStoredBlock in certain conditions)
 func (l *BatchSubmitter) calculateL2BlockRangeToStore(ctx context.Context) (eth.BlockID, eth.BlockID, error) {
-	childCtx, cancel := context.WithTimeout(ctx, networkTimeout)
+	childCtx, cancel := context.WithTimeout(ctx, l.NetworkTimeout)
 	defer cancel()
 	syncStatus, err := l.RollupNode.SyncStatus(childCtx)
 	// Ensure that we have the sync status
@@ -312,8 +313,9 @@ func (l *BatchSubmitter) loop() {
 					l.log.Error("unable to get tx data", "err", err)
 					break
 				}
+
 				// Record TX Status
-				if receipt, err := l.txMgr.SendTransaction(l.ctx, txdata.Bytes()); err != nil {
+				if receipt, err := l.SendTransaction(l.ctx, txdata.Bytes()); err != nil {
 					l.recordFailedTx(txdata.ID(), err)
 				} else {
 					l.recordConfirmedTx(txdata.ID(), receipt)
@@ -335,6 +337,44 @@ func (l *BatchSubmitter) loop() {
 	}
 }
 
+// SendTransaction creates & submits a transaction to the batch inbox address with the given `data`.
+// It currently uses the underlying `txmgr` to handle transaction sending & price management.
+// This is a blocking method. It should not be called concurrently.
+func (l *BatchSubmitter) SendTransaction(ctx context.Context, data []byte) (*types.Receipt, error) {
+	// Do the gas estimation offline if specified
+	// A value of 0 will cause the [txmgr] to estimate the gas limit.
+	var gas uint64
+	if l.OfflineGasEstimation {
+		intrinsicGas, err := core.IntrinsicGas(data, nil, false, true, true, false)
+		if err != nil {
+			return nil, fmt.Errorf("failed to calculate intrinsic gas: %w", err)
+		}
+		gas = intrinsicGas
+	}
+
+	// Create the transaction
+	tx, err := l.txMgr.CraftTx(ctx, txmgr.TxCandidate{
+		To:       l.Rollup.BatchInboxAddress,
+		TxData:   data,
+		From:     l.From,
+		GasLimit: gas,
+	})
+	if err != nil {
+		return nil, fmt.Errorf("failed to create tx: %w", err)
+	}
+
+	// Send the transaction through the txmgr
+	ctx, cancel := context.WithTimeout(ctx, l.TxManagerTimeout)
+	defer cancel()
+	if receipt, err := l.txMgr.Send(ctx, tx); err != nil {
+		l.log.Warn("unable to publish tx", "err", err, "data_size", len(data))
+		return nil, err
+	} else {
+		l.log.Info("tx successfully published", "tx_hash", receipt.TxHash, "data_size", len(data))
+		return receipt, nil
+	}
+}
+
 func (l *BatchSubmitter) recordL1Tip(l1tip eth.L1BlockRef) {
 	if l.lastL1Tip == l1tip {
 		return

op-batcher/batcher/driver_test.go

@@ -0,0 +1,79 @@
+package batcher
+
+import (
+	"context"
+	"math/big"
+	"testing"
+
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/ethereum/go-ethereum/core/types"
+	"github.com/ethereum/go-ethereum/log"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum-optimism/optimism/op-node/rollup"
+	"github.com/ethereum-optimism/optimism/op-node/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/txmgr"
+	"github.com/ethereum-optimism/optimism/op-service/txmgr/mocks"
+)
+
+// TestSendTransaction tests that the driver can send a transaction
+// through the txmgr.
+func TestSendTransaction(t *testing.T) {
+	log := testlog.Logger(t, log.LvlCrit)
+	txMgr := mocks.TxManager{}
+	batcherInboxAddress := common.HexToAddress("0x42000000000000000000000000000000000000ff")
+	chainID := big.NewInt(1)
+	sender := common.HexToAddress("0xdeadbeef")
+	bs := BatchSubmitter{
+		Config: Config{
+			log:                  log,
+			From:                 sender,
+			OfflineGasEstimation: false,
+			Rollup: &rollup.Config{
+				L1ChainID:         chainID,
+				BatchInboxAddress: batcherInboxAddress,
+			},
+		},
+		txMgr: &txMgr,
+	}
+	txData := []byte{0x00, 0x01, 0x02}
+
+	gasTipCap := big.NewInt(136)
+	gasFeeCap := big.NewInt(137)
+	gas := uint64(1337)
+
+	candidate := txmgr.TxCandidate{
+		To:       batcherInboxAddress,
+		TxData:   txData,
+		From:     sender,
+		GasLimit: uint64(0),
+	}
+
+	tx := types.NewTx(&types.DynamicFeeTx{
+		ChainID:   chainID,
+		Nonce:     0,
+		GasTipCap: gasTipCap,
+		GasFeeCap: gasFeeCap,
+		Gas:       gas,
+		To:        &batcherInboxAddress,
+		Data:      txData,
+	})
+	txHash := tx.Hash()
+
+	expectedReceipt := types.Receipt{
+		Type:              1,
+		PostState:         []byte{},
+		Status:            uint64(1),
+		CumulativeGasUsed: gas,
+		TxHash:            txHash,
+		GasUsed:           gas,
+	}
+
+	txMgr.On("CraftTx", mock.Anything, candidate).Return(tx, nil)
+	txMgr.On("Send", mock.Anything, tx).Return(&expectedReceipt, nil)
+
+	receipt, err := bs.SendTransaction(context.Background(), tx.Data())
+	require.NoError(t, err)
+	require.Equal(t, receipt, &expectedReceipt)
+}

op-batcher/flags/flags.go

@@ -1,6 +1,8 @@
 package flags
 
 import (
+	"time"
+
 	"github.com/urfave/cli"
 
 	"github.com/ethereum-optimism/optimism/op-batcher/rpc"
@@ -74,7 +76,17 @@ var (
 	}
 
 	/* Optional flags */
-
+	OfflineGasEstimationFlag = cli.BoolFlag{
+		Name:   "offline-gas-estimation",
+		Usage:  "Whether to use offline gas estimation",
+		EnvVar: opservice.PrefixEnvVar(envVarPrefix, "OFFLINE_GAS_ESTIMATION"),
+	}
+	TxManagerTimeoutFlag = cli.DurationFlag{
+		Name:   "tx-manager-timeout",
+		Usage:  "Maximum duration to wait for L1 transactions, including resubmissions",
+		Value:  10 * time.Minute,
+		EnvVar: opservice.PrefixEnvVar(envVarPrefix, "TX_MANAGER_TIMEOUT"),
+	}
 	MaxChannelDurationFlag = cli.Uint64Flag{
 		Name:   "max-channel-duration",
 		Usage:  "The maximum duration of L1-blocks to keep a channel open. 0 to disable.",
@@ -141,6 +153,8 @@ var requiredFlags = []cli.Flag{
 }
 
 var optionalFlags = []cli.Flag{
+	OfflineGasEstimationFlag,
+	TxManagerTimeoutFlag,
 	MaxChannelDurationFlag,
 	MaxL1TxSizeBytesFlag,
 	TargetL1TxSizeBytesFlag,

op-e2e/migration_test.go

@@ -327,6 +327,8 @@ func TestMigration(t *testing.T) {
 		L1EthRpc:                  forkedL1URL,
 		L2EthRpc:                  gethNode.WSEndpoint(),
 		RollupRpc:                 rollupNode.HTTPEndpoint(),
+		TxManagerTimeout:          10 * time.Minute,
+		OfflineGasEstimation:      true,
 		MaxChannelDuration:        1,
 		MaxL1TxSize:               120_000,
 		TargetL1TxSize:            100_000,

op-e2e/setup.go

@@ -582,10 +582,13 @@ func (cfg SystemConfig) Start(_opts ...SystemConfigOption) (*System, error) {
 	}
 
 	// Batch Submitter
+	txManagerTimeout := 10 * time.Minute
 	sys.BatchSubmitter, err = bss.NewBatchSubmitterFromCLIConfig(bss.CLIConfig{
 		L1EthRpc:                  sys.Nodes["l1"].WSEndpoint(),
 		L2EthRpc:                  sys.Nodes["sequencer"].WSEndpoint(),
 		RollupRpc:                 sys.RollupNodes["sequencer"].HTTPEndpoint(),
+		TxManagerTimeout:          txManagerTimeout,
+		OfflineGasEstimation:      true,
 		MaxChannelDuration:        1,
 		MaxL1TxSize:               120_000,
 		TargetL1TxSize:            100_000,