Skip to content

fix(op-batcher): solve race condition of BatchSubmitter #5

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Jun 15, 2023
Merged

fix(op-batcher): solve race condition of BatchSubmitter #5

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions op-batcher/batcher/driver.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,14 +10,15 @@ import (
"sync"
"time"

"github.com/ethereum/go-ethereum/core"
"github.com/ethereum/go-ethereum/core/types"
"github.com/ethereum/go-ethereum/log"

"github.com/ethereum-optimism/optimism/op-batcher/metrics"
"github.com/ethereum-optimism/optimism/op-node/eth"
"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
opclient "github.com/ethereum-optimism/optimism/op-service/client"
"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"
)

// BatchSubmitter encapsulates a service responsible for submitting L2 tx
Expand All @@ -27,6 +28,8 @@ type BatchSubmitter struct {

txMgr txmgr.TxManager
wg sync.WaitGroup
// publishReceiveMu solve race condition: publishTxToL1 and handleReceipt access state concurrently
publishReceiveMu sync.Mutex

shutdownCtx context.Context
cancelShutdownCtx context.CancelFunc
Expand Down Expand Up @@ -363,7 +366,9 @@ func (l *BatchSubmitter) publishTxToL1(ctx context.Context, queue *txmgr.Queue[t
l.recordL1Tip(l1tip)

// Collect next transaction data
l.publishReceiveMu.Lock()
txdata, err := l.state.TxData(l1tip.ID())
l.publishReceiveMu.Unlock()
if err == io.EOF {
l.log.Trace("no transaction data available")
return err
Expand Down Expand Up @@ -397,6 +402,9 @@ func (l *BatchSubmitter) sendTransaction(txdata txData, queue *txmgr.Queue[txDat
}

func (l *BatchSubmitter) handleReceipt(r txmgr.TxReceipt[txData]) {
l.publishReceiveMu.Lock()
defer l.publishReceiveMu.Unlock()

// Record TX Status
if r.Err != nil {
l.log.Warn("unable to publish tx", "err", r.Err, "data_size", r.ID.Len())
Expand Down