From 61b69b38b9f41c3c4fd2947071c2faf8bb8cf0a9 Mon Sep 17 00:00:00 2001 From: Samuel Laferriere Date: Thu, 22 Aug 2024 23:33:18 -0700 Subject: [PATCH 1/2] feat(wip): implement a altda -> txsend nonblocking pipeline to submit altda blobs concurrently --- op-batcher/batcher/driver.go | 103 ++++++++++++++++++++--------------- 1 file changed, 60 insertions(+), 43 deletions(-) diff --git a/op-batcher/batcher/driver.go b/op-batcher/batcher/driver.go index 39ebf2f25b24..7e62a1ac825b 100644 --- a/op-batcher/batcher/driver.go +++ b/op-batcher/batcher/driver.go @@ -297,6 +297,7 @@ const ( func (l *BatchSubmitter) loop() { defer l.wg.Done() + txDataCh := make(chan txData) receiptsCh := make(chan txmgr.TxReceipt[txRef]) queue := txmgr.NewQueue[txRef](l.killCtx, l.Txmgr, l.Config.MaxPendingTransactions) @@ -329,12 +330,41 @@ func (l *BatchSubmitter) loop() { } } }() + go func() { + for { + select { + case txdata := <-txDataCh: + var candidate *txmgr.TxCandidate + var err error + if txdata.asBlob { + if candidate, err = l.blobTxCandidate(txdata); err != nil { + // We could potentially fall through and try a calldata tx instead, but this would + // likely result in the chain spending more in gas fees than it is tuned for, so best + // to just fail. We do not expect this error to trigger unless there is a serious bug + // or configuration issue. + l.state.TxFailed(txdata.ID()) + l.Log.Error("could not create blob tx candidate: %w", err) + } + } else { + // sanity check + if nf := len(txdata.frames); nf != 1 { + l.Log.Crit("Unexpected number of frames in calldata tx", "num_frames", nf) + } + candidate = l.calldataTxCandidate(txdata.CallData()) + } + l.queueTx(txdata, false, candidate, queue, receiptsCh) + case <-l.shutdownCtx.Done(): + l.Log.Info("Shutting down tx sending loop") + return + } + } + }() ticker := time.NewTicker(l.Config.PollInterval) defer ticker.Stop() publishAndWait := func() { - l.publishStateToL1(queue, receiptsCh) + l.prepareStateToSubmitToL1(txDataCh) if !l.Txmgr.IsClosed() { queue.Wait() } else { @@ -369,7 +399,7 @@ func (l *BatchSubmitter) loop() { l.clearState(l.shutdownCtx) continue } - l.publishStateToL1(queue, receiptsCh) + l.prepareStateToSubmitToL1(txDataCh) case <-l.shutdownCtx.Done(): if l.Txmgr.IsClosed() { l.Log.Info("Txmgr is closed, remaining channel data won't be sent") @@ -424,22 +454,24 @@ func (l *BatchSubmitter) waitNodeSync() error { return dial.WaitRollupSync(l.shutdownCtx, l.Log, rollupClient, l1TargetBlock, time.Second*12) } -// publishStateToL1 queues up all pending TxData to be published to the L1, returning when there is +// prepareStateToSubmitToL1 queues up all pending TxData to be published to the L1, returning when there is // no more data to queue for publishing or if there was an error queing the data. -func (l *BatchSubmitter) publishStateToL1(queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) { +func (l *BatchSubmitter) prepareStateToSubmitToL1(txDataCh chan<- txData) { for { // if the txmgr is closed, we stop the transaction sending if l.Txmgr.IsClosed() { l.Log.Info("Txmgr is closed, aborting state publishing") return } - err := l.publishTxToL1(l.killCtx, queue, receiptsCh) + txdata, err := l.prepareTxData(l.killCtx) if err != nil { if err != io.EOF { l.Log.Error("Error publishing tx to l1", "err", err) } return } + // send txdata to be published by the tx sending loop + txDataCh <- txdata } } @@ -482,13 +514,13 @@ func (l *BatchSubmitter) clearState(ctx context.Context) { } } -// publishTxToL1 submits a single state tx to the L1 -func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) error { +// prepareTxData prepares a single state tx to be sent to the L1 +func (l *BatchSubmitter) prepareTxData(ctx context.Context) (txData, error) { // send all available transactions l1tip, err := l.l1Tip(ctx) if err != nil { l.Log.Error("Failed to query L1 tip", "err", err) - return err + return txData{}, err } l.recordL1Tip(l1tip) @@ -497,16 +529,17 @@ func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[t if err == io.EOF { l.Log.Trace("No transaction data available") - return err + return txData{}, err } else if err != nil { l.Log.Error("Unable to get tx data", "err", err) - return err + return txData{}, err } - if err = l.sendTransaction(ctx, txdata, queue, receiptsCh); err != nil { - return fmt.Errorf("BatchSubmitter.sendTransaction failed: %w", err) + if err = l.publishToAltDAUpdateTxData(ctx, txdata); err != nil { + return txData{}, fmt.Errorf("BatchSubmitter.prepareAndQueueTxData failed: %w", err) } - return nil + + return txdata, nil } func (l *BatchSubmitter) safeL1Origin(ctx context.Context) (eth.BlockID, error) { @@ -548,44 +581,28 @@ func (l *BatchSubmitter) cancelBlockingTx(queue *txmgr.Queue[txRef], receiptsCh l.queueTx(txData{}, true, candidate, queue, receiptsCh) } -// sendTransaction creates & queues for sending a transaction to the batch inbox address with the given `txData`. +// publishToAltDAUpdateTxData creates & queues for sending a transaction to the batch inbox address with the given `txData`. // The method will block if the queue's MaxPendingTransactions is exceeded. -func (l *BatchSubmitter) sendTransaction(ctx context.Context, txdata txData, queue *txmgr.Queue[txRef], receiptsCh chan txmgr.TxReceipt[txRef]) error { - var err error - // Do the gas estimation offline. A value of 0 will cause the [txmgr] to estimate the gas limit. - - var candidate *txmgr.TxCandidate - if txdata.asBlob { - if candidate, err = l.blobTxCandidate(txdata); err != nil { - // We could potentially fall through and try a calldata tx instead, but this would - // likely result in the chain spending more in gas fees than it is tuned for, so best - // to just fail. We do not expect this error to trigger unless there is a serious bug - // or configuration issue. - return fmt.Errorf("could not create blob tx candidate: %w", err) - } - } else { +func (l *BatchSubmitter) publishToAltDAUpdateTxData(ctx context.Context, txdata txData) error { + // if AltDA is enabled we post the txdata to the DA Provider and replace it with the commitment. + if l.Config.UseAltDA { // sanity check if nf := len(txdata.frames); nf != 1 { l.Log.Crit("Unexpected number of frames in calldata tx", "num_frames", nf) } - data := txdata.CallData() - // if AltDA is enabled we post the txdata to the DA Provider and replace it with the commitment. - if l.Config.UseAltDA { - comm, err := l.AltDA.SetInput(ctx, data) - if err != nil { - l.Log.Error("Failed to post input to Alt DA", "error", err) - // requeue frame if we fail to post to the DA Provider so it can be retried - l.recordFailedTx(txdata.ID(), err) - return nil - } - l.Log.Info("Set AltDA input", "commitment", comm, "tx", txdata.ID()) - // signal AltDA commitment tx with TxDataVersion1 - data = comm.TxData() + comm, err := l.AltDA.SetInput(ctx, txdata.CallData()) + if err != nil { + l.Log.Error("Failed to post input to Alt DA", "error", err) + // requeue frame if we fail to post to the DA Provider so it can be retried + l.recordFailedTx(txdata.ID(), err) + return err } - candidate = l.calldataTxCandidate(data) + l.Log.Info("Set AltDA input", "commitment", comm, "tx", txdata.ID()) + // TODO: this is kind of weird... we're overwriting the frame's data with the commitment + // but keeping the same ID just so we can delete/requeue the frames on success/failure + txdata.frames[0].data = comm.TxData() } - l.queueTx(txdata, false, candidate, queue, receiptsCh) return nil } From dc86dd146a121d9216413bbeccb7ea5c86e25950 Mon Sep 17 00:00:00 2001 From: Samuel Laferriere Date: Thu, 22 Aug 2024 23:37:49 -0700 Subject: [PATCH 2/2] refactor: publishToAltDAUpdateTxData fct to make it more readable also added TODO comment to run in goroutine --- op-batcher/batcher/driver.go | 39 ++++++++++++++++++------------------ 1 file changed, 20 insertions(+), 19 deletions(-) diff --git a/op-batcher/batcher/driver.go b/op-batcher/batcher/driver.go index 7e62a1ac825b..0ef3820406ec 100644 --- a/op-batcher/batcher/driver.go +++ b/op-batcher/batcher/driver.go @@ -535,8 +535,12 @@ func (l *BatchSubmitter) prepareTxData(ctx context.Context) (txData, error) { return txData{}, err } - if err = l.publishToAltDAUpdateTxData(ctx, txdata); err != nil { - return txData{}, fmt.Errorf("BatchSubmitter.prepareAndQueueTxData failed: %w", err) + // if AltDA is enabled we post the txdata to the DA Provider and replace it with the commitment. + if l.Config.UseAltDA { + // TODO: run this in a goroutine to avoid blocking the main loop + if err = l.publishToAltDAUpdateTxData(ctx, txdata); err != nil { + return txData{}, fmt.Errorf("BatchSubmitter.prepareAndQueueTxData failed: %w", err) + } } return txdata, nil @@ -584,24 +588,21 @@ func (l *BatchSubmitter) cancelBlockingTx(queue *txmgr.Queue[txRef], receiptsCh // publishToAltDAUpdateTxData creates & queues for sending a transaction to the batch inbox address with the given `txData`. // The method will block if the queue's MaxPendingTransactions is exceeded. func (l *BatchSubmitter) publishToAltDAUpdateTxData(ctx context.Context, txdata txData) error { - // if AltDA is enabled we post the txdata to the DA Provider and replace it with the commitment. - if l.Config.UseAltDA { - // sanity check - if nf := len(txdata.frames); nf != 1 { - l.Log.Crit("Unexpected number of frames in calldata tx", "num_frames", nf) - } - comm, err := l.AltDA.SetInput(ctx, txdata.CallData()) - if err != nil { - l.Log.Error("Failed to post input to Alt DA", "error", err) - // requeue frame if we fail to post to the DA Provider so it can be retried - l.recordFailedTx(txdata.ID(), err) - return err - } - l.Log.Info("Set AltDA input", "commitment", comm, "tx", txdata.ID()) - // TODO: this is kind of weird... we're overwriting the frame's data with the commitment - // but keeping the same ID just so we can delete/requeue the frames on success/failure - txdata.frames[0].data = comm.TxData() + // sanity check + if nf := len(txdata.frames); nf != 1 { + l.Log.Crit("Unexpected number of frames in calldata tx", "num_frames", nf) + } + comm, err := l.AltDA.SetInput(ctx, txdata.CallData()) + if err != nil { + l.Log.Error("Failed to post input to Alt DA", "error", err) + // requeue frame if we fail to post to the DA Provider so it can be retried + l.recordFailedTx(txdata.ID(), err) + return err } + l.Log.Info("Set AltDA input", "commitment", comm, "tx", txdata.ID()) + // TODO: this is kind of weird... we're overwriting the frame's data with the commitment + // but keeping the same ID just so we can delete/requeue the frames on success/failure + txdata.frames[0].data = comm.TxData() return nil }