op-plasma: track challenges when pipeline is stalled

op-e2e/actions/plasma_test.go

@@ -10,7 +10,6 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/node/safedb"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	plasma "github.com/ethereum-optimism/optimism/op-plasma"
-	"github.com/ethereum-optimism/optimism/op-service/eth"
 	"github.com/ethereum-optimism/optimism/op-service/sources"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 	"github.com/ethereum/go-ethereum/accounts/abi/bind"
@@ -45,10 +44,10 @@ type PlasmaParam func(p *e2eutils.TestParams)
 
 func NewL2PlasmaDA(t Testing, params ...PlasmaParam) *L2PlasmaDA {
 	p := &e2eutils.TestParams{
-		MaxSequencerDrift:   2,
-		SequencerWindowSize: 4,
-		ChannelTimeout:      4,
-		L1BlockTime:         3,
+		MaxSequencerDrift:   40,
+		SequencerWindowSize: 120,
+		ChannelTimeout:      120,
+		L1BlockTime:         12,
 		UsePlasma:           true,
 	}
 	for _, apply := range params {
@@ -163,7 +162,7 @@ func (a *L2PlasmaDA) ActNewL2Tx(t Testing) {
 		a.lastComm = tx.Data[1:]
 	})
 
-	a.miner.ActL1StartBlock(3)(t)
+	a.miner.ActL1StartBlock(12)(t)
 	a.miner.ActL1IncludeTx(a.dp.Addresses.Batcher)(t)
 	a.miner.ActL1EndBlock(t)
 
@@ -191,7 +190,7 @@ func (a *L2PlasmaDA) ActChallengeInput(t Testing, comm []byte, bn uint64) {
 	_, err = a.contract.Deposit(txOpts)
 	require.NoError(t, err)
 
-	a.miner.ActL1StartBlock(3)(t)
+	a.miner.ActL1StartBlock(12)(t)
 	a.miner.ActL1IncludeTx(a.alice.Address())(t)
 	a.miner.ActL1EndBlock(t)
 
@@ -201,38 +200,42 @@ func (a *L2PlasmaDA) ActChallengeInput(t Testing, comm []byte, bn uint64) {
 	_, err = a.contract.Challenge(txOpts, big.NewInt(int64(bn)), comm)
 	require.NoError(t, err)
 
-	a.miner.ActL1StartBlock(3)(t)
+	a.miner.ActL1StartBlock(12)(t)
 	a.miner.ActL1IncludeTx(a.alice.Address())(t)
 	a.miner.ActL1EndBlock(t)
 }
 
 func (a *L2PlasmaDA) ActExpireLastInput(t Testing) {
 	reorgWindow := a.plasmaCfg.ResolveWindow + a.plasmaCfg.ChallengeWindow
 	for a.miner.l1Chain.CurrentBlock().Number.Uint64() <= a.lastCommBn+reorgWindow {
-		a.miner.ActL1StartBlock(3)(t)
+		a.miner.ActL1StartBlock(12)(t)
 		a.miner.ActL1EndBlock(t)
 	}
 }
 
-func (a *L2PlasmaDA) ActResolveLastChallenge(t Testing) {
-	// remove commitment byte prefix
-	input, err := a.storage.GetInput(t.Ctx(), a.lastComm[1:])
-	require.NoError(t, err)
-
+func (a *L2PlasmaDA) ActResolveInput(t Testing, comm []byte, input []byte, bn uint64) {
 	txOpts, err := bind.NewKeyedTransactorWithChainID(a.dp.Secrets.Alice, a.sd.L1Cfg.Config.ChainID)
 	require.NoError(t, err)
 
-	_, err = a.contract.Resolve(txOpts, big.NewInt(int64(a.lastCommBn)), a.lastComm, input)
+	_, err = a.contract.Resolve(txOpts, big.NewInt(int64(bn)), comm, input)
 	require.NoError(t, err)
 
-	a.miner.ActL1StartBlock(3)(t)
+	a.miner.ActL1StartBlock(12)(t)
 	a.miner.ActL1IncludeTx(a.alice.Address())(t)
 	a.miner.ActL1EndBlock(t)
 }
 
+func (a *L2PlasmaDA) ActResolveLastChallenge(t Testing) {
+	// remove commitment byte prefix
+	input, err := a.storage.GetInput(t.Ctx(), a.lastComm[1:])
+	require.NoError(t, err)
+
+	a.ActResolveInput(t, a.lastComm, input, a.lastCommBn)
+}
+
 func (a *L2PlasmaDA) ActL1Blocks(t Testing, n uint64) {
 	for i := uint64(0); i < n; i++ {
-		a.miner.ActL1StartBlock(3)(t)
+		a.miner.ActL1StartBlock(12)(t)
 		a.miner.ActL1EndBlock(t)
 	}
 }
@@ -288,11 +291,7 @@ func TestPlasma_ChallengeExpired(gt *testing.T) {
 	harness.ActL1Blocks(t, 1)
 	harness.sequencer.ActL2PipelineFull(t)
 
-	// make sure that the finalized head was correctly updated on the engine.
-	l2Finalized, err := harness.engCl.L2BlockRefByLabel(t.Ctx(), eth.Finalized)
-	require.NoError(t, err)
-	require.Equal(t, uint64(8), l2Finalized.Number)
-
+	// get new block with same number to compare
 	newBlk, err := harness.engine.EthClient().BlockByNumber(t.Ctx(), blk.Number())
 	require.NoError(t, err)
 
@@ -439,3 +438,187 @@ func TestPlasma_ChallengeReorg(gt *testing.T) {
 	// confirm the reorg did happen
 	require.NotEqual(t, blk.Hash(), newBlk.Hash())
 }
+
+// Sequencer stalls as data is not available, batcher keeps posting, untracked commitments are
+// challenged and resolved, then sequencer resumes and catches up.
+func TestPlasma_SequencerStalledMultiChallenges(gt *testing.T) {
+	if !e2eutils.UsePlasma() {
+		gt.Skip("Plasma is not enabled")
+	}
+
+	t := NewDefaultTesting(gt)
+	a := NewL2PlasmaDA(t)
+
+	// generate some initial L1 blocks.
+	a.ActL1Blocks(t, 5)
+	a.sequencer.ActL1HeadSignal(t)
+
+	// create a new tx on l2 and commit it to l1
+	a.ActNewL2Tx(t)
+
+	// keep track of the related commitment
+	comm1 := a.lastComm
+	input1, err := a.storage.GetInput(t.Ctx(), comm1[1:])
+	bn1 := a.lastCommBn
+	require.NoError(t, err)
+
+	// delete it from the DA provider so the pipeline cannot verify it
+	a.ActDeleteLastInput(t)
+
+	// build more empty l2 unsafe blocks as the l1 origin progresses
+	a.ActL1Blocks(t, 10)
+	a.sequencer.ActBuildToL1HeadUnsafe(t)
+
+	// build another L2 block without advancing derivation
+	a.alice.L2.ActResetTxOpts(t)
+	a.alice.L2.ActSetTxToAddr(&a.dp.Addresses.Bob)(t)
+	a.alice.L2.ActMakeTx(t)
+
+	a.sequencer.ActL2StartBlock(t)
+	a.engine.ActL2IncludeTx(a.alice.Address())(t)
+	a.sequencer.ActL2EndBlock(t)
+
+	a.batcher.ActL2BatchBuffer(t)
+	a.batcher.ActL2ChannelClose(t)
+	a.batcher.ActL2BatchSubmit(t, func(tx *types.DynamicFeeTx) {
+		a.lastComm = tx.Data
+	})
+
+	// include it in L1
+	a.miner.ActL1StartBlock(120)(t)
+	a.miner.ActL1IncludeTx(a.dp.Addresses.Batcher)(t)
+	a.miner.ActL1EndBlock(t)
+
+	a.sequencer.ActL1HeadSignal(t)
+
+	unsafe := a.sequencer.L2Unsafe()
+	unsafeBlk, err := a.engine.EthClient().BlockByHash(t.Ctx(), unsafe.Hash)
+	require.NoError(t, err)
+
+	// advance the pipeline until it errors out as it is still stuck
+	// on deriving the first commitment
+	for i := 0; i < 3; i++ {
+		a.sequencer.ActL2PipelineStep(t)
+	}
+
+	// keep track of the second commitment
+	comm2 := a.lastComm
+	_, err = a.storage.GetInput(t.Ctx(), comm2[1:])
+	require.NoError(t, err)
+	a.lastCommBn = a.miner.l1Chain.CurrentBlock().Number.Uint64()
+
+	// ensure the second commitment is distinct from the first
+	require.NotEqual(t, comm1, comm2)
+
+	// challenge the last commitment while the pipeline is stuck on the first
+	a.ActChallengeLastInput(t)
+
+	// resolve the latest commitment before the first one is event challenged.
+	a.ActResolveLastChallenge(t)
+
+	// now we delete it to force the pipeline to resolve the second commitment
+	// from the challenge data.
+	a.ActDeleteLastInput(t)
+
+	// finally challenge the first commitment
+	a.ActChallengeInput(t, comm1, bn1)
+
+	// resolve it immediately so we can resume derivation
+	a.ActResolveInput(t, comm1, input1, bn1)
+
+	// pipeline can go on
+	a.sequencer.ActL2PipelineFull(t)
+
+	// verify that the chain did not reorg out
+	safeBlk, err := a.engine.EthClient().BlockByNumber(t.Ctx(), unsafeBlk.Number())
+	require.NoError(t, err)
+	require.Equal(t, unsafeBlk.Hash(), safeBlk.Hash())
+}
+
+// Verify that finalization happens based on plasma DA windows.
+// based on l2_batcher_test.go L2Finalization
+func TestPlasma_Finalization(gt *testing.T) {
+	if !e2eutils.UsePlasma() {
+		gt.Skip("Plasma is not enabled")
+	}
+	t := NewDefaultTesting(gt)
+	a := NewL2PlasmaDA(t)
+
+	// build L1 block #1
+	a.ActL1Blocks(t, 1)
+	a.miner.ActL1SafeNext(t)
+
+	// Fill with l2 blocks up to the L1 head
+	a.sequencer.ActL1HeadSignal(t)
+	a.sequencer.ActBuildToL1Head(t)
+
+	a.sequencer.ActL2PipelineFull(t)
+	a.sequencer.ActL1SafeSignal(t)
+	require.Equal(t, uint64(1), a.sequencer.SyncStatus().SafeL1.Number)
+
+	// add L1 block #2
+	a.ActL1Blocks(t, 1)
+	a.miner.ActL1SafeNext(t)
+	a.miner.ActL1FinalizeNext(t)
+	a.sequencer.ActL1HeadSignal(t)
+	a.sequencer.ActBuildToL1Head(t)
+
+	// Catch up derivation
+	a.sequencer.ActL2PipelineFull(t)
+	a.sequencer.ActL1FinalizedSignal(t)
+	a.sequencer.ActL1SafeSignal(t)
+
+	// commit all the l2 blocks to L1
+	a.batcher.ActSubmitAll(t)
+	a.miner.ActL1StartBlock(12)(t)
+	a.miner.ActL1IncludeTx(a.dp.Addresses.Batcher)(t)
+	a.miner.ActL1EndBlock(t)
+
+	// verify
+	a.sequencer.ActL2PipelineFull(t)
+
+	// fill with more unsafe L2 blocks
+	a.sequencer.ActL1HeadSignal(t)
+	a.sequencer.ActBuildToL1Head(t)
+
+	// submit those blocks too, block #4
+	a.batcher.ActSubmitAll(t)
+	a.miner.ActL1StartBlock(12)(t)
+	a.miner.ActL1IncludeTx(a.dp.Addresses.Batcher)(t)
+	a.miner.ActL1EndBlock(t)
+
+	// add some more L1 blocks #5, #6
+	a.miner.ActEmptyBlock(t)
+	a.miner.ActEmptyBlock(t)
+
+	// and more unsafe L2 blocks
+	a.sequencer.ActL1HeadSignal(t)
+	a.sequencer.ActBuildToL1Head(t)
+
+	// move safe/finalize markers: finalize the L1 chain block with the first batch, but not the second
+	a.miner.ActL1SafeNext(t)     // #2 -> #3
+	a.miner.ActL1SafeNext(t)     // #3 -> #4
+	a.miner.ActL1FinalizeNext(t) // #1 -> #2
+	a.miner.ActL1FinalizeNext(t) // #2 -> #3
+
+	// L1 safe and finalized as expected
+	a.sequencer.ActL2PipelineFull(t)
+	a.sequencer.ActL1FinalizedSignal(t)
+	a.sequencer.ActL1SafeSignal(t)
+	a.sequencer.ActL1HeadSignal(t)
+	require.Equal(t, uint64(6), a.sequencer.SyncStatus().HeadL1.Number)
+	require.Equal(t, uint64(4), a.sequencer.SyncStatus().SafeL1.Number)
+	require.Equal(t, uint64(3), a.sequencer.SyncStatus().FinalizedL1.Number)
+	// l2 cannot finalize yet as the challenge window is not passed
+	require.Equal(t, uint64(0), a.sequencer.SyncStatus().FinalizedL2.Number)
+
+	// expire the challenge window so these blocks can no longer be challenged
+	a.ActL1Blocks(t, a.plasmaCfg.ChallengeWindow)
+
+	// advance derivation and finalize plasma via the L1 signal
+	a.sequencer.ActL2PipelineFull(t)
+	a.ActL1Finalized(t)
+
+	// given 12s l1 time and 1s l2 time, l2 should be 12 * 3 = 36 blocks finalized
+	require.Equal(t, uint64(36), a.sequencer.SyncStatus().FinalizedL2.Number)
+}

op-node/rollup/derive/plasma_data_source.go

@@ -50,6 +50,7 @@ func (s *PlasmaDataSource) Next(ctx context.Context) (eth.Data, error) {
 		if err != nil {
 			return nil, err
 		}
+
 		if len(data) == 0 {
 			return nil, NotEnoughData
 		}

op-plasma/damgr.go

@@ -361,15 +361,16 @@ func (d *DA) decodeChallengeStatus(log *types.Log) (ChallengeStatus, Keccak256Co
 	if err != nil {
 		return 0, nil, err
 	}
-	d.log.Debug("decoded challenge status event", "log", log, "event", event)
 	comm, err := DecodeKeccak256(event.ChallengedCommitment)
 	if err != nil {
 		return 0, nil, err
 	}
+	d.log.Debug("decoded challenge status event", "log", log, "event", event, "comm", fmt.Sprintf("%x", comm.Encode()))
 
 	bn := event.ChallengedBlockNumber.Uint64()
-	// if we are not tracking the commitment from processing the l1 origin in derivation,
-	// i.e. someone challenged garbage data, this challenge is invalid.
+	// IsTracking just validates whether the commitment was challenged for the correct block number
+	// if it has been loaded from the batcher inbox before. Spam commitments will be tracked but
+	// ignored and evicted unless derivation encounters the commitment.
 	if !d.state.IsTracking(comm.Encode(), bn) {
 		return 0, nil, fmt.Errorf("%w: %x at block %d", ErrInvalidChallenge, comm.Encode(), bn)
 	}