sync: Add CGT with develop

Makefile

@@ -34,6 +34,7 @@ golang-docker: ## Builds Docker images for Go components using buildx
 	GIT_COMMIT=$$(git rev-parse HEAD) \
 	GIT_DATE=$$(git show -s --format='%ct') \
 	IMAGE_TAGS=$$(git rev-parse HEAD),latest \
+	KONA_VERSION=$$(jq -r .version kona/version.json) \
 	docker buildx bake \
 			--progress plain \
 			--load \

docker-bake.hcl

@@ -10,10 +10,6 @@ variable "KONA_VERSION" {
   default = "none"
 }
 
-variable "ASTERISC_VERSION" {
-  default = "v1.3.0"
-}
-
 variable "GIT_COMMIT" {
   default = "dev"
 }
@@ -149,7 +145,6 @@ target "op-challenger" {
     GIT_DATE = "${GIT_DATE}"
     OP_CHALLENGER_VERSION = "${OP_CHALLENGER_VERSION}"
     KONA_VERSION="${KONA_VERSION}"
-    ASTERISC_VERSION="${ASTERISC_VERSION}"
   }
   target = "op-challenger-target"
   platforms = split(",", PLATFORMS)

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

@@ -0,0 +1,116 @@
+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())
+
+	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)
+			time.Sleep(20 * time.Millisecond) // failed to force-include tx: type: 2 sender; err: nonce too high
+		}
+
+		l.Debug("Sequencing L1 block", "iteration_j", j)
+		sys.TestSequencer.SequenceBlock(t, sys.L1Network.ChainID(), 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, 2, len(channels)) // we expect a total of 2 channels
+
+		// 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: 30_000, max: 40_000, note: "channel 0 - filled to the max capacity"},
+			{min: 30_000, max: 40_000, note: "channel 1 - remaining data, filling channel close to max capacity"},
+		}
+
+		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
+	}
+
+	status := sys.L2CL.SyncStatus()
+	spew.Dump(status)
+}
+
+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
+			cfg.MaxPendingTransactions = 7
+		})),
+	)
+}

op-acceptance-tests/tests/flashblocks/flashblocks_stream_test.go

@@ -82,7 +82,7 @@ func TestFlashblocksStream(gt *testing.T) {
 	defer close(builderOutput)
 	builderDone := make(chan struct{})
 	go func() {
-		err := oprbuilderNode.FlashblocksClient().ListenFor(ctx, logger.With("stream_source", "op-rbuilder"), testDuration, builderOutput, builderDone)
+		err := oprbuilderNode.FlashblocksClient().ReadAll(ctx, logger.With("stream_source", "op-rbuilder"), testDuration, builderOutput, builderDone)
 		require.NoError(t, err)
 	}()
 	builderMessages := make([]string, 0)
@@ -92,7 +92,7 @@ func TestFlashblocksStream(gt *testing.T) {
 	doneListening := make(chan struct{})
 	streamedMessages := make([]string, 0)
 	go func() {
-		err := rollupBoostNode.FlashblocksClient().ListenFor(ctx, logger.With("stream_source", "rollup-boost"), testDuration, output, doneListening)
+		err := rollupBoostNode.FlashblocksClient().ReadAll(ctx, logger.With("stream_source", "rollup-boost"), testDuration, output, doneListening)
 		require.NoError(t, err)
 	}()
 

op-acceptance-tests/tests/flashblocks/flashblocks_transfer_test.go

@@ -45,21 +45,22 @@ func TestFlashblocksTransfer(gt *testing.T) {
 	_, span = tracer.Start(ctx, fmt.Sprintf("test chain %s", sys.L2Chain.String()))
 	defer span.End()
 
-	doneListening := make(chan struct{})
-	output := make(chan []byte, 100)
-
 	// Drive a couple blocks on the test sequencer so the faucet L2 funding tx has a chance to land before we rely on it.
 	driveViaTestSequencer(t, sys, 2)
 
 	alice := sys.FunderL2.NewFundedEOA(eth.ThreeHundredthsEther)
 	bob := sys.Wallet.NewEOA(sys.L2EL)
 	bobAddress := bob.Address().Hex()
 
-	// flashblocks listener
+	// flashblocks listener - start goroutine and wait for it to be running
 	flashblocksClient := sys.L2RollupBoost.FlashblocksClient()
+	output := make(chan []byte, 100)
+	doneListening := make(chan struct{})
 	go func() {
-		err := flashblocksClient.ListenFor(ctx, logger, 20*time.Second, output, doneListening)
-		t.Require().NoError(err, "failed to listen for flashblocks")
+		err := flashblocksClient.ReadAll(ctx, logger.With("stream_source", "rollup-boost"), 20*time.Second, output, doneListening)
+		if err != nil {
+			t.Require().NoError(err, "failed to listen for flashblocks")
+		}
 	}()
 
 	var executedTransaction *txplan.PlannedTx

op-acceptance-tests/tests/interop/reorgs/l2_reorgs_after_l1_reorg_test.go

@@ -8,10 +8,8 @@ import (
 	"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-supervisor/supervisor/types"
-	"github.com/ethereum-optimism/optimism/op-test-sequencer/sequencer/seqtypes"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/stretchr/testify/require"
 )
@@ -66,7 +64,6 @@ func testL2ReorgAfterL1Reorg(gt *testing.T, n int, preChecks, postChecks checksF
 	ctx := t.Ctx()
 
 	sys := presets.NewSimpleInterop(t)
-	ts := sys.TestSequencer.Escape().ControlAPI(sys.L1Network.ChainID())
 
 	cl := sys.L1Network.Escape().L1CLNode(match.FirstL1CL)
 
@@ -76,7 +73,7 @@ func testL2ReorgAfterL1Reorg(gt *testing.T, n int, preChecks, postChecks checksF
 
 	// sequence a few L1 and L2 blocks
 	for range n + 1 {
-		sequenceL1Block(t, ts, common.Hash{})
+		sys.TestSequencer.SequenceBlock(t, sys.L1Network.ChainID(), common.Hash{})
 
 		sys.L2ChainA.WaitForBlock()
 		sys.L2ChainA.WaitForBlock()
@@ -101,7 +98,7 @@ func testL2ReorgAfterL1Reorg(gt *testing.T, n int, preChecks, postChecks checksF
 	tipL2_preReorg := sys.L2ELA.BlockRefByLabel(eth.Unsafe)
 
 	// reorg the L1 chain -- sequence an alternative L1 block from divergence block parent
-	sequenceL1Block(t, ts, divergence.ParentHash)
+	sys.TestSequencer.SequenceBlock(t, sys.L1Network.ChainID(), divergence.ParentHash)
 
 	// continue building on the alternative L1 chain
 	sys.ControlPlane.FakePoSState(cl.ID(), stack.Start)
@@ -160,8 +157,3 @@ func testL2ReorgAfterL1Reorg(gt *testing.T, n int, preChecks, postChecks checksF
 	// post reorg test validations and checks
 	postChecks(t, sys)
 }
-
-func sequenceL1Block(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()))
-}

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 {