Add concurrent sends to batcher

op-batcher/batcher/config.go

@@ -26,8 +26,9 @@ type Config struct {
 	RollupNode *sources.RollupClient
 	TxManager  txmgr.TxManager
 
-	NetworkTimeout time.Duration
-	PollInterval   time.Duration
+	NetworkTimeout         time.Duration
+	PollInterval           time.Duration
+	MaxPendingTransactions uint64
 
 	// RollupConfig is queried at startup
 	Rollup *rollup.Config
@@ -76,6 +77,10 @@ type CLIConfig struct {
 	// and creating a new batch.
 	PollInterval time.Duration
 
+	// MaxPendingTransactions is the maximum number of concurrent pending
+	// transactions sent to the transaction manager.
+	MaxPendingTransactions uint64
+
 	// MaxL1TxSize is the maximum size of a batch tx submitted to L1.
 	MaxL1TxSize uint64
 
@@ -128,16 +133,17 @@ func NewConfig(ctx *cli.Context) CLIConfig {
 		PollInterval:    ctx.GlobalDuration(flags.PollIntervalFlag.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),
-		TxMgrConfig:        txmgr.ReadCLIConfig(ctx),
-		RPCConfig:          rpc.ReadCLIConfig(ctx),
-		LogConfig:          oplog.ReadCLIConfig(ctx),
-		MetricsConfig:      opmetrics.ReadCLIConfig(ctx),
-		PprofConfig:        oppprof.ReadCLIConfig(ctx),
+		MaxPendingTransactions: ctx.GlobalUint64(flags.MaxPendingTransactionsFlag.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),
+		TxMgrConfig:            txmgr.ReadCLIConfig(ctx),
+		RPCConfig:              rpc.ReadCLIConfig(ctx),
+		LogConfig:              oplog.ReadCLIConfig(ctx),
+		MetricsConfig:          opmetrics.ReadCLIConfig(ctx),
+		PprofConfig:            oppprof.ReadCLIConfig(ctx),
 	}
 }

op-batcher/batcher/driver.go

@@ -75,13 +75,14 @@ func NewBatchSubmitterFromCLIConfig(cfg CLIConfig, l log.Logger, m metrics.Metri
 	}
 
 	batcherCfg := Config{
-		L1Client:       l1Client,
-		L2Client:       l2Client,
-		RollupNode:     rollupClient,
-		PollInterval:   cfg.PollInterval,
-		NetworkTimeout: cfg.TxMgrConfig.NetworkTimeout,
-		TxManager:      txManager,
-		Rollup:         rcfg,
+		L1Client:               l1Client,
+		L2Client:               l2Client,
+		RollupNode:             rollupClient,
+		PollInterval:           cfg.PollInterval,
+		MaxPendingTransactions: cfg.MaxPendingTransactions,
+		NetworkTimeout:         cfg.TxMgrConfig.NetworkTimeout,
+		TxManager:              txManager,
+		Rollup:                 rcfg,
 		Channel: ChannelConfig{
 			SeqWindowSize:      rcfg.SeqWindowSize,
 			ChannelTimeout:     rcfg.ChannelTimeout,
@@ -301,6 +302,11 @@ func (l *BatchSubmitter) loop() {
 // publishStateToL1 loops through the block data loaded into `state` and
 // submits the associated data to the L1 in the form of channel frames.
 func (l *BatchSubmitter) publishStateToL1(ctx context.Context) {
+	maxPending := l.MaxPendingTransactions
+	if maxPending == 0 {
+		maxPending = 1<<64 - 1
+	}
+
 	for {
 		// Attempt to gracefully terminate the current channel, ensuring that no new frames will be
 		// produced. Any remaining frames must still be published to the L1 to prevent stalling.
@@ -326,20 +332,29 @@ func (l *BatchSubmitter) publishStateToL1(ctx context.Context) {
 		l.recordL1Tip(l1tip)
 
 		// Collect next transaction data
-		txdata, err := l.state.TxData(l1tip.ID())
-		if err == io.EOF {
-			l.log.Trace("no transaction data available")
-			break
-		} else if err != nil {
-			l.log.Error("unable to get tx data", "err", err)
-			break
-		}
-		// Record TX Status
-		if receipt, err := l.sendTransaction(ctx, txdata.Bytes()); err != nil {
-			l.recordFailedTx(txdata.ID(), err)
-		} else {
-			l.recordConfirmedTx(txdata.ID(), receipt)
+		var wg sync.WaitGroup
+		for i := uint64(0); i < maxPending; i++ {
+			var txdata txData
+			txdata, err = l.state.TxData(l1tip.ID())
+			if err == io.EOF {
+				l.log.Trace("no transaction data available")
+				break
+			} else if err != nil {
+				l.log.Error("unable to get tx data", "err", err)
+				break
+			}
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				// Record TX Status
+				if receipt, err := l.sendTransaction(ctx, txdata.Bytes()); err != nil {
+					l.recordFailedTx(txdata.ID(), err)
+				} else {
+					l.recordConfirmedTx(txdata.ID(), receipt)
+				}
+			}()
 		}
+		wg.Wait()
 	}
 }
 

op-batcher/flags/flags.go

@@ -48,6 +48,12 @@ var (
 	}
 
 	// Optional flags
+	MaxPendingTransactionsFlag = cli.Uint64Flag{
+		Name:   "max-pending-tx",
+		Usage:  "The maximum number of pending transactions. 0 for no limit.",
+		Value:  1,
+		EnvVar: opservice.PrefixEnvVar(envVarPrefix, "MAX_PENDING_TX"),
+	}
 	MaxChannelDurationFlag = cli.Uint64Flag{
 		Name:   "max-channel-duration",
 		Usage:  "The maximum duration of L1-blocks to keep a channel open. 0 to disable.",
@@ -96,6 +102,7 @@ var requiredFlags = []cli.Flag{
 }
 
 var optionalFlags = []cli.Flag{
+	MaxPendingTransactionsFlag,
 	MaxChannelDurationFlag,
 	MaxL1TxSizeBytesFlag,
 	TargetL1TxSizeBytesFlag,