op-batcher: propagate clock.Clock to gossip builder check. refactor MaxChannelDuration

Author: nonsense

op-acceptance-tests/tests/batcher/batcher_test.go

@@ -0,0 +1,132 @@
+package batcher
+
+import (
+	"testing"
+	"time"
+
+	"github.com/davecgh/go-spew/spew"
+	"github.com/ethereum-optimism/optimism/op-devstack/devtest"
+	"github.com/ethereum-optimism/optimism/op-devstack/dsl"
+	"github.com/ethereum-optimism/optimism/op-devstack/presets"
+	"github.com/ethereum-optimism/optimism/op-devstack/stack"
+	"github.com/ethereum-optimism/optimism/op-devstack/stack/match"
+	"github.com/ethereum-optimism/optimism/op-service/apis"
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum-optimism/optimism/op-service/txplan"
+	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/types"
+	"github.com/ethereum-optimism/optimism/op-test-sequencer/sequencer/seqtypes"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBatcherFullChannelsAfterDowntime(gt *testing.T) {
+	t := devtest.SerialT(gt)
+	sys := presets.NewSingleChainMultiNodeWithTestSeq(t)
+	l := t.Logger()
+	ts_L2 := sys.TestSequencer.Escape().ControlAPI(sys.L2EL.ChainID())
+	ts_L1 := sys.TestSequencer.Escape().ControlAPI(sys.L1Network.ChainID())
+
+	alice := sys.FunderL2.NewFundedEOA(eth.OneWei)
+	cathrine := sys.FunderL2.NewFundedEOA(eth.OneTenthEther)
+
+	cl := sys.L1Network.Escape().L1CLNode(match.FirstL1CL)
+
+	sys.ControlPlane.FakePoSState(cl.ID(), stack.Stop)
+
+	latestUnsafe_A := sys.L2CL.StopSequencer()
+	l.Info("Latest unsafe block after stopping the L2 sequencer", "latestUnsafe", latestUnsafe_A)
+
+	parent := latestUnsafe_A
+	nonce := uint64(0)
+	for j := 0; j < 200; j++ {
+		l1Origin := sys.L1EL.BlockRefByLabel(eth.Unsafe).Hash
+
+		for i := 0; i < 5; i++ {
+			l.Debug("Sequencing L2 block", "iteration", i, "parent", parent)
+			sequenceBlockWithL1Origin(t, ts_L2, parent, l1Origin, alice, cathrine, nonce)
+			nonce++
+
+			parent = sys.L2CL.HeadBlockRef(types.LocalUnsafe).Hash
+
+			sys.AdvanceTime(time.Second * 2)
+		}
+
+		l.Debug("Sequencing L1 block", "iteration_j", j)
+		sequenceBlock(t, ts_L1, common.Hash{})
+	}
+
+	sys.L2CL.StartSequencer()
+
+	l.Info("Current L1 unsafe block", "currentL1Unsafe", sys.L1EL.BlockRefByLabel(eth.Unsafe))
+	sys.ControlPlane.FakePoSState(cl.ID(), stack.Start)
+
+	sys.L2Batcher.Start()
+
+	channels, channelFrames, l2Txs := sys.L2Chain.DeriveData(4) // over the next 4 blocks, collect batches/channels/frames submitted by the batcher on the L1 network, and parse them
+	{
+		for _, c := range channels {
+			l.Info("Channel details", "channelID", c.String(), "frameCount", len(channelFrames[c]), "dataLength_frame0", len(channelFrames[c][0].Data))
+		}
+
+		require.Equal(t, 10, len(channels)) // we expect a total of 10 channels, due to existing MaxChannelDuration bug filed at: https://github.com/ethereum-optimism/optimism/issues/18092
+
+		// values are dependent on:
+		// - MaxPendingTransactions
+		// - number of blocks and transactions sent in the test - 1000 L2 blocks with 1 transaction from cathrine to alice
+		// - MaxL1TxSize (this is set to 40_000 bytes for test purposes)
+		sizeRanges := []struct {
+			min  int
+			max  int
+			note string
+		}{
+			{min: 7_000, max: 10_000, note: "channel 0 - the first 100 blocks"},
+			{min: 100, max: 1000, note: "channel 1 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 2 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 3 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 4 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 5 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 6 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 100, max: 1000, note: "channel 7 - only a few blocks due to racy behavior, nowhere near the channel capacity"},
+			{min: 20_000, max: 40_000, note: "channel 8 - filled to the max capacity, due to blocking behavior because of MaxPendingTransactions limit reached"},
+			{min: 20_000, max: 40_000, note: "channel 9 - filled to the max capacity, due to blocking behavior because of MaxPendingTransactions limit reached"},
+		}
+
+		for i, entry := range sizeRanges {
+			require.LessOrEqual(t, len(channelFrames[channels[i]][0].Data), entry.max, entry.note)
+			require.GreaterOrEqual(t, len(channelFrames[channels[i]][0].Data), entry.min, entry.note)
+		}
+
+		require.Equal(t, len(l2Txs[cathrine.Address()]), 1000) // we expect 1000 transactions total sent from cathrine to alice
+	}
+
+	//sys.L2Chain.PrintChain()
+	//sys.L1Network.PrintChain()
+
+	status := sys.L2CL.SyncStatus()
+	spew.Dump(status)
+}
+
+func sequenceBlock(t devtest.T, ts apis.TestSequencerControlAPI, parent common.Hash) {
+	require.NoError(t, ts.New(t.Ctx(), seqtypes.BuildOpts{Parent: parent}))
+	require.NoError(t, ts.Next(t.Ctx()))
+}
+
+func sequenceBlockWithL1Origin(t devtest.T, ts apis.TestSequencerControlAPI, parent common.Hash, l1Origin common.Hash, alice *dsl.EOA, cathrine *dsl.EOA, nonce uint64) {
+	require.NoError(t, ts.New(t.Ctx(), seqtypes.BuildOpts{Parent: parent, L1Origin: &l1Origin}))
+
+	// include simple transfer tx in opened block
+	{
+		to := cathrine.PlanTransfer(alice.Address(), eth.OneWei)
+		opt := txplan.Combine(to, txplan.WithStaticNonce(nonce))
+		ptx := txplan.NewPlannedTx(opt)
+		signed_tx, err := ptx.Signed.Eval(t.Ctx())
+		require.NoError(t, err, "Expected to be able to evaluate a planned transaction on op-test-sequencer, but got error")
+		txdata, err := signed_tx.MarshalBinary()
+		require.NoError(t, err, "Expected to be able to marshal a signed transaction on op-test-sequencer, but got error")
+
+		err = ts.IncludeTx(t.Ctx(), txdata)
+		require.NoError(t, err, "Expected to be able to include a signed transaction on op-test-sequencer, but got error")
+	}
+
+	require.NoError(t, ts.Next(t.Ctx()))
+}

op-acceptance-tests/tests/batcher/init_test.go

@@ -0,0 +1,33 @@
+package batcher
+
+import (
+	"testing"
+	"time"
+
+	bss "github.com/ethereum-optimism/optimism/op-batcher/batcher"
+	"github.com/ethereum-optimism/optimism/op-devstack/compat"
+	"github.com/ethereum-optimism/optimism/op-devstack/presets"
+	"github.com/ethereum-optimism/optimism/op-devstack/stack"
+	"github.com/ethereum-optimism/optimism/op-devstack/sysgo"
+)
+
+func TestMain(m *testing.M) {
+	presets.DoMain(m, presets.WithSingleChainMultiNode(),
+		presets.WithExecutionLayerSyncOnVerifiers(),
+		presets.WithCompatibleTypes(compat.SysGo),
+		presets.WithNoDiscovery(),
+		presets.WithTimeTravel(),
+		stack.MakeCommon(sysgo.WithBatcherOption(func(id stack.L2BatcherID, cfg *bss.CLIConfig) {
+			cfg.Stopped = true
+
+			// set the blob max size to 40_000 bytes for test purposes
+			cfg.MaxL1TxSize = 40_000
+			cfg.TestUseMaxTxSizeForBlobs = true
+
+			cfg.PollInterval = 1000 * time.Millisecond
+
+			cfg.MaxChannelDuration = 50 // not sure if this is reasonable, or should be larger?
+			cfg.MaxPendingTransactions = 7
+		})),
+	)
+}

op-batcher/batcher/channel.go

@@ -27,7 +27,7 @@ type channel struct {
 }
 
 func newChannel(log log.Logger, metr metrics.Metricer, cfg ChannelConfig, rollupCfg *rollup.Config, latestL1OriginBlockNum uint64, channelOut derive.ChannelOut) *channel {
-	cb := NewChannelBuilderWithChannelOut(cfg, rollupCfg, latestL1OriginBlockNum, channelOut)
+	cb := NewChannelBuilderWithChannelOut(log, cfg, rollupCfg, latestL1OriginBlockNum, channelOut)
 	return &channel{
 		ChannelBuilder:        cb,
 		log:                   log,

op-batcher/batcher/channel_builder.go

@@ -11,6 +11,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/queue"
+	"github.com/ethereum/go-ethereum/log"
 )
 
 var (
@@ -47,6 +48,7 @@ type frameData struct {
 // ChannelBuilder uses a ChannelOut to create a channel with output frame
 // size approximation.
 type ChannelBuilder struct {
+	log       log.Logger
 	cfg       ChannelConfig
 	rollupCfg *rollup.Config
 
@@ -87,8 +89,9 @@ type ChannelBuilder struct {
 	outputBytes int
 }
 
-func NewChannelBuilderWithChannelOut(cfg ChannelConfig, rollupCfg *rollup.Config, latestL1OriginBlockNum uint64, channelOut derive.ChannelOut) *ChannelBuilder {
+func NewChannelBuilderWithChannelOut(log log.Logger, cfg ChannelConfig, rollupCfg *rollup.Config, latestL1OriginBlockNum uint64, channelOut derive.ChannelOut) *ChannelBuilder {
 	cb := &ChannelBuilder{
+		log:       log.With("channel_id", channelOut.ID()),
 		cfg:       cfg,
 		rollupCfg: rollupCfg,
 		co:        channelOut,
@@ -224,8 +227,7 @@ func (c *ChannelBuilder) updateDurationTimeout(l1BlockNum uint64) {
 	if c.cfg.MaxChannelDuration == 0 {
 		return
 	}
-	timeout := l1BlockNum + c.cfg.MaxChannelDuration
-	c.updateTimeout(timeout, ErrMaxDurationReached)
+	c.updateTimeout(l1BlockNum+c.cfg.MaxChannelDuration, ErrMaxDurationReached)
 }
 
 // updateSwTimeout updates the block timeout with the sequencer window timeout
@@ -244,6 +246,7 @@ func (c *ChannelBuilder) updateSwTimeout(l1InfoNumber uint64) {
 // full error reason in case the timeout is hit in the future.
 func (c *ChannelBuilder) updateTimeout(timeoutBlockNum uint64, reason error) {
 	if c.timeout == 0 || c.timeout > timeoutBlockNum {
+		c.log.Debug("setting timeout", "number", timeoutBlockNum, "timeout", c.timeout)
 		c.timeout = timeoutBlockNum
 		c.timeoutReason = reason
 	}
@@ -252,17 +255,12 @@ func (c *ChannelBuilder) updateTimeout(timeoutBlockNum uint64, reason error) {
 // CheckTimeout checks if the channel is timed out at the given block number and
 // in this case marks the channel as full, if it wasn't full already.
 func (c *ChannelBuilder) CheckTimeout(l1BlockNum uint64) {
-	if !c.IsFull() && c.TimedOut(l1BlockNum) {
+	if c.timeout != 0 && !c.IsFull() && l1BlockNum >= c.timeout {
+		c.log.Debug("checking timeout", "l1blockNum", l1BlockNum, "timeout", c.timeout)
 		c.setFullErr(c.timeoutReason)
 	}
 }
 
-// TimedOut returns whether the passed block number is after the timeout block
-// number. If no block timeout is set yet, it returns false.
-func (c *ChannelBuilder) TimedOut(blockNum uint64) bool {
-	return c.timeout != 0 && blockNum >= c.timeout
-}
-
 // IsFull returns whether the channel is full.
 // FullErr returns the reason for the channel being full.
 func (c *ChannelBuilder) IsFull() bool {