op-node: op-node interop block verification (devnet-1 scope) (ethereu…

…m-optimism#11611)

* op-node: experimental interop block verification

* op-node: supervisor RPC init

* op-e2e/actions: test interop safety checks

* op-e2e/op-node: test fixes

* op-node: update comments

* op-node: unit-test interop deriver, trigger cross-safe updates on driver step, to poll for supervisor safety changes

* op-node: add more comments, fix flag description

* op-e2e: fix rebase, add missing argument

op-e2e/actions/altda_test.go

@@ -81,7 +81,7 @@ func NewL2AltDA(t Testing, params ...AltDAParam) *L2AltDA {
 
 	daMgr := altda.NewAltDAWithStorage(log, altDACfg, storage, &altda.NoopMetrics{})
 
-	sequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), daMgr, engCl, sd.RollupCfg, 0)
+	sequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), daMgr, engCl, sd.RollupCfg, 0, nil)
 	miner.ActL1SetFeeRecipient(common.Address{'A'})
 	sequencer.ActL2PipelineFull(t)
 
@@ -139,7 +139,7 @@ func (a *L2AltDA) NewVerifier(t Testing) *L2Verifier {
 
 	daMgr := altda.NewAltDAWithStorage(a.log, a.altDACfg, a.storage, &altda.NoopMetrics{})
 
-	verifier := NewL2Verifier(t, a.log, l1F, a.miner.BlobStore(), daMgr, engCl, a.sd.RollupCfg, &sync.Config{}, safedb.Disabled)
+	verifier := NewL2Verifier(t, a.log, l1F, a.miner.BlobStore(), daMgr, engCl, a.sd.RollupCfg, &sync.Config{}, safedb.Disabled, nil)
 
 	return verifier
 }

op-e2e/actions/batch_queue_test.go

@@ -88,7 +88,8 @@ func TestDeriveChainFromNearL1Genesis(gt *testing.T) {
 	// This is the same situation as if op-node restarted at this point.
 	l2Cl, err := sources.NewEngineClient(seqEngine.RPCClient(), logger, nil, sources.EngineClientDefaultConfig(sd.RollupCfg))
 	require.NoError(gt, err)
-	verifier := NewL2Verifier(t, logger, sequencer.l1, miner.BlobStore(), altda.Disabled, l2Cl, sequencer.rollupCfg, sequencer.syncCfg, safedb.Disabled)
+	verifier := NewL2Verifier(t, logger, sequencer.l1, miner.BlobStore(), altda.Disabled,
+		l2Cl, sequencer.rollupCfg, sequencer.syncCfg, safedb.Disabled, nil)
 	verifier.ActL2PipelineFull(t) // Should not get stuck in a reset loop forever
 	require.EqualValues(gt, l2BlockNum, seqEngine.l2Chain.CurrentSafeBlock().Number.Uint64())
 	require.EqualValues(gt, l2BlockNum, seqEngine.l2Chain.CurrentFinalBlock().Number.Uint64())

op-e2e/actions/interop_test.go

@@ -0,0 +1,123 @@
+package actions
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	"github.com/ethereum/go-ethereum/log"
+
+	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/interop"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
+	"github.com/ethereum-optimism/optimism/op-service/testlog"
+	"github.com/ethereum-optimism/optimism/op-service/testutils"
+	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/types"
+)
+
+var _ interop.InteropBackend = (*testutils.MockInteropBackend)(nil)
+
+func TestInteropVerifier(gt *testing.T) {
+	t := NewDefaultTesting(gt)
+	dp := e2eutils.MakeDeployParams(t, defaultRollupTestParams)
+	sd := e2eutils.Setup(t, dp, defaultAlloc)
+	// Temporary work-around: interop needs to be active, for cross-safety to not be instant.
+	// The state genesis in this test is pre-interop however.
+	sd.RollupCfg.InteropTime = new(uint64)
+	logger := testlog.Logger(t, log.LevelDebug)
+	seqMockBackend := &testutils.MockInteropBackend{}
+	l1Miner, seqEng, seq := setupSequencerTest(t, sd, logger,
+		WithVerifierOpts(WithInteropBackend(seqMockBackend)))
+
+	batcher := NewL2Batcher(logger, sd.RollupCfg, DefaultBatcherCfg(dp),
+		seq.RollupClient(), l1Miner.EthClient(), seqEng.EthClient(), seqEng.EngineClient(t, sd.RollupCfg))
+
+	verMockBackend := &testutils.MockInteropBackend{}
+	_, ver := setupVerifier(t, sd, logger,
+		l1Miner.L1Client(t, sd.RollupCfg), l1Miner.BlobStore(), &sync.Config{},
+		WithInteropBackend(verMockBackend))
+
+	seq.ActL2PipelineFull(t)
+	ver.ActL2PipelineFull(t)
+
+	l2ChainID := types.ChainIDFromBig(sd.RollupCfg.L2ChainID)
+	seqMockBackend.ExpectCheckBlock(l2ChainID, 1, types.Unsafe, nil)
+	// create an unsafe L2 block
+	seq.ActL2StartBlock(t)
+	seq.ActL2EndBlock(t)
+	seq.ActL2PipelineFull(t)
+	seqMockBackend.AssertExpectations(t)
+	status := seq.SyncStatus()
+	require.Equal(t, uint64(1), status.UnsafeL2.Number)
+	require.Equal(t, uint64(0), status.CrossUnsafeL2.Number)
+	require.Equal(t, uint64(0), status.LocalSafeL2.Number)
+	require.Equal(t, uint64(0), status.SafeL2.Number)
+
+	// promote it to cross-unsafe in the backend
+	// and see if the node picks up on it
+	seqMockBackend.ExpectCheckBlock(l2ChainID, 1, types.CrossUnsafe, nil)
+	seq.ActInteropBackendCheck(t)
+	seq.ActL2PipelineFull(t)
+	seqMockBackend.AssertExpectations(t)
+	status = seq.SyncStatus()
+	require.Equal(t, uint64(1), status.UnsafeL2.Number)
+	require.Equal(t, uint64(1), status.CrossUnsafeL2.Number, "cross unsafe now")
+	require.Equal(t, uint64(0), status.LocalSafeL2.Number)
+	require.Equal(t, uint64(0), status.SafeL2.Number)
+
+	// submit all new L2 blocks
+	batcher.ActSubmitAll(t)
+	// new L1 block with L2 batch
+	l1Miner.ActL1StartBlock(12)(t)
+	l1Miner.ActL1IncludeTx(sd.RollupCfg.Genesis.SystemConfig.BatcherAddr)(t)
+	l1Miner.ActL1EndBlock(t)
+
+	// Sync the L1 block, to verify the L2 block as local-safe.
+	seqMockBackend.ExpectCheckBlock(l2ChainID, 1, types.CrossUnsafe, nil) // not cross-safe yet
+	seq.ActL1HeadSignal(t)
+	seq.ActL2PipelineFull(t)
+	seqMockBackend.AssertExpectations(t)
+
+	status = seq.SyncStatus()
+	require.Equal(t, uint64(1), status.UnsafeL2.Number)
+	require.Equal(t, uint64(1), status.CrossUnsafeL2.Number)
+	require.Equal(t, uint64(1), status.LocalSafeL2.Number, "local safe changed")
+	require.Equal(t, uint64(0), status.SafeL2.Number)
+
+	// Now mark it as cross-safe
+	seqMockBackend.ExpectCheckBlock(l2ChainID, 1, types.CrossSafe, nil)
+	seq.ActInteropBackendCheck(t)
+	seq.ActL2PipelineFull(t)
+	seqMockBackend.AssertExpectations(t)
+
+	status = seq.SyncStatus()
+	require.Equal(t, uint64(1), status.UnsafeL2.Number)
+	require.Equal(t, uint64(1), status.CrossUnsafeL2.Number)
+	require.Equal(t, uint64(1), status.LocalSafeL2.Number)
+	require.Equal(t, uint64(1), status.SafeL2.Number, "cross-safe reached")
+	require.Equal(t, uint64(0), status.FinalizedL2.Number)
+
+	// The verifier might not see the L2 block that was just derived from L1 as cross-verified yet.
+	verMockBackend.ExpectCheckBlock(l2ChainID, 1, types.Unsafe, nil) // for the local unsafe check
+	verMockBackend.ExpectCheckBlock(l2ChainID, 1, types.Unsafe, nil) // for the local safe check
+	ver.ActL1HeadSignal(t)
+	ver.ActL2PipelineFull(t)
+	verMockBackend.AssertExpectations(t)
+	status = ver.SyncStatus()
+	require.Equal(t, uint64(1), status.UnsafeL2.Number, "synced the block")
+	require.Equal(t, uint64(0), status.CrossUnsafeL2.Number, "not cross-verified yet")
+	require.Equal(t, uint64(1), status.LocalSafeL2.Number, "derived from L1, thus local-safe")
+	require.Equal(t, uint64(0), status.SafeL2.Number, "not yet cross-safe")
+	require.Equal(t, uint64(0), status.FinalizedL2.Number)
+
+	// signal that L1 finalized; the cross-safe block we have should get finalized too
+	l1Miner.ActL1SafeNext(t)
+	l1Miner.ActL1FinalizeNext(t)
+	seq.ActL1SafeSignal(t)
+	seq.ActL1FinalizedSignal(t)
+	seq.ActL2PipelineFull(t)
+	seqMockBackend.AssertExpectations(t)
+
+	status = seq.SyncStatus()
+	require.Equal(t, uint64(1), status.FinalizedL2.Number, "finalized the block")
+}

op-e2e/actions/l2_sequencer.go

@@ -19,6 +19,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/driver"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/interop"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sequencing"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/eth"
@@ -50,8 +51,9 @@ type L2Sequencer struct {
 }
 
 func NewL2Sequencer(t Testing, log log.Logger, l1 derive.L1Fetcher, blobSrc derive.L1BlobsFetcher,
-	altDASrc driver.AltDAIface, eng L2API, cfg *rollup.Config, seqConfDepth uint64) *L2Sequencer {
-	ver := NewL2Verifier(t, log, l1, blobSrc, altDASrc, eng, cfg, &sync.Config{}, safedb.Disabled)
+	altDASrc driver.AltDAIface, eng L2API, cfg *rollup.Config, seqConfDepth uint64,
+	interopBackend interop.InteropBackend) *L2Sequencer {
+	ver := NewL2Verifier(t, log, l1, blobSrc, altDASrc, eng, cfg, &sync.Config{}, safedb.Disabled, interopBackend)
 	attrBuilder := derive.NewFetchingAttributesBuilder(cfg, l1, eng)
 	seqConfDepthL1 := confdepth.NewConfDepth(seqConfDepth, ver.syncStatus.L1Head, l1)
 	l1OriginSelector := &MockL1OriginSelector{

op-e2e/actions/l2_sequencer_test.go

@@ -37,8 +37,30 @@ func EngineWithP2P() EngineOption {
 	}
 }
 
-func setupSequencerTest(t Testing, sd *e2eutils.SetupData, log log.Logger) (*L1Miner, *L2Engine, *L2Sequencer) {
+type sequencerCfg struct {
+	verifierCfg
+}
+
+func defaultSequencerConfig() *sequencerCfg {
+	return &sequencerCfg{verifierCfg: *defaultVerifierCfg()}
+}
+
+type SequencerOpt func(opts *sequencerCfg)
+
+func WithVerifierOpts(opts ...VerifierOpt) SequencerOpt {
+	return func(cfg *sequencerCfg) {
+		for _, opt := range opts {
+			opt(&cfg.verifierCfg)
+		}
+	}
+}
+
+func setupSequencerTest(t Testing, sd *e2eutils.SetupData, log log.Logger, opts ...SequencerOpt) (*L1Miner, *L2Engine, *L2Sequencer) {
 	jwtPath := e2eutils.WriteDefaultJWT(t)
+	cfg := defaultSequencerConfig()
+	for _, opt := range opts {
+		opt(cfg)
+	}
 
 	miner := NewL1Miner(t, log.New("role", "l1-miner"), sd.L1Cfg)
 
@@ -48,7 +70,7 @@ func setupSequencerTest(t Testing, sd *e2eutils.SetupData, log log.Logger) (*L1M
 	l2Cl, err := sources.NewEngineClient(engine.RPCClient(), log, nil, sources.EngineClientDefaultConfig(sd.RollupCfg))
 	require.NoError(t, err)
 
-	sequencer := NewL2Sequencer(t, log.New("role", "sequencer"), l1F, miner.BlobStore(), altda.Disabled, l2Cl, sd.RollupCfg, 0)
+	sequencer := NewL2Sequencer(t, log.New("role", "sequencer"), l1F, miner.BlobStore(), altda.Disabled, l2Cl, sd.RollupCfg, 0, cfg.interopBackend)
 	return miner, engine, sequencer
 }
 

op-e2e/actions/l2_verifier.go

@@ -23,6 +23,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-node/rollup/engine"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/event"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/finality"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/interop"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/status"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/client"
@@ -84,7 +85,10 @@ type safeDB interface {
 	node.SafeDBReader
 }
 
-func NewL2Verifier(t Testing, log log.Logger, l1 derive.L1Fetcher, blobsSrc derive.L1BlobsFetcher, altDASrc driver.AltDAIface, eng L2API, cfg *rollup.Config, syncCfg *sync.Config, safeHeadListener safeDB) *L2Verifier {
+func NewL2Verifier(t Testing, log log.Logger, l1 derive.L1Fetcher,
+	blobsSrc derive.L1BlobsFetcher, altDASrc driver.AltDAIface,
+	eng L2API, cfg *rollup.Config, syncCfg *sync.Config, safeHeadListener safeDB,
+	interopBackend interop.InteropBackend) *L2Verifier {
 	ctx, cancel := context.WithCancel(context.Background())
 	t.Cleanup(cancel)
 
@@ -104,6 +108,10 @@ func NewL2Verifier(t Testing, log log.Logger, l1 derive.L1Fetcher, blobsSrc deri
 		},
 	}
 
+	if interopBackend != nil {
+		sys.Register("interop", interop.NewInteropDeriver(log, cfg, ctx, interopBackend, eng), opts)
+	}
+
 	metrics := &testutils.TestDerivationMetrics{}
 	ec := engine.NewEngineController(eng, log, metrics, cfg, syncCfg,
 		sys.Register("engine-controller", nil, opts))
@@ -316,6 +324,13 @@ func (s *L2Verifier) ActL1FinalizedSignal(t Testing) {
 	require.Equal(t, finalized, s.syncStatus.SyncStatus().FinalizedL1)
 }
 
+func (s *L2Verifier) ActInteropBackendCheck(t Testing) {
+	s.synchronousEvents.Emit(engine.CrossUpdateRequestEvent{
+		CrossUnsafe: true,
+		CrossSafe:   true,
+	})
+}
+
 func (s *L2Verifier) OnEvent(ev event.Event) bool {
 	switch x := ev.(type) {
 	case rollup.L1TemporaryErrorEvent:

op-e2e/actions/l2_verifier_test.go

@@ -3,20 +3,22 @@ package actions
 import (
 	"testing"
 
-	altda "github.com/ethereum-optimism/optimism/op-alt-da"
-	"github.com/ethereum-optimism/optimism/op-node/node/safedb"
 	"github.com/ethereum/go-ethereum/common"
 	"github.com/ethereum/go-ethereum/log"
 	"github.com/stretchr/testify/require"
 
+	altda "github.com/ethereum-optimism/optimism/op-alt-da"
 	"github.com/ethereum-optimism/optimism/op-e2e/e2eutils"
+	"github.com/ethereum-optimism/optimism/op-node/node/safedb"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/derive"
+	"github.com/ethereum-optimism/optimism/op-node/rollup/interop"
 	"github.com/ethereum-optimism/optimism/op-node/rollup/sync"
 	"github.com/ethereum-optimism/optimism/op-service/testlog"
 )
 
 type verifierCfg struct {
 	safeHeadListener safeDB
+	interopBackend   interop.InteropBackend
 }
 
 type VerifierOpt func(opts *verifierCfg)
@@ -27,21 +29,28 @@ func WithSafeHeadListener(l safeDB) VerifierOpt {
 	}
 }
 
+func WithInteropBackend(b interop.InteropBackend) VerifierOpt {
+	return func(opts *verifierCfg) {
+		opts.interopBackend = b
+	}
+}
+
 func defaultVerifierCfg() *verifierCfg {
 	return &verifierCfg{
 		safeHeadListener: safedb.Disabled,
 	}
 }
 
-func setupVerifier(t Testing, sd *e2eutils.SetupData, log log.Logger, l1F derive.L1Fetcher, blobSrc derive.L1BlobsFetcher, syncCfg *sync.Config, opts ...VerifierOpt) (*L2Engine, *L2Verifier) {
+func setupVerifier(t Testing, sd *e2eutils.SetupData, log log.Logger,
+	l1F derive.L1Fetcher, blobSrc derive.L1BlobsFetcher, syncCfg *sync.Config, opts ...VerifierOpt) (*L2Engine, *L2Verifier) {
 	cfg := defaultVerifierCfg()
 	for _, opt := range opts {
 		opt(cfg)
 	}
 	jwtPath := e2eutils.WriteDefaultJWT(t)
 	engine := NewL2Engine(t, log.New("role", "verifier-engine"), sd.L2Cfg, sd.RollupCfg.Genesis.L1, jwtPath, EngineWithP2P())
 	engCl := engine.EngineClient(t, sd.RollupCfg)
-	verifier := NewL2Verifier(t, log.New("role", "verifier"), l1F, blobSrc, altda.Disabled, engCl, sd.RollupCfg, syncCfg, cfg.safeHeadListener)
+	verifier := NewL2Verifier(t, log.New("role", "verifier"), l1F, blobSrc, altda.Disabled, engCl, sd.RollupCfg, syncCfg, cfg.safeHeadListener, cfg.interopBackend)
 	return engine, verifier
 }
 

op-e2e/actions/reorg_test.go

@@ -613,7 +613,7 @@ func RestartOpGeth(gt *testing.T, deltaTimeOffset *hexutil.Uint64) {
 	engRpc := &rpcWrapper{seqEng.RPCClient()}
 	l2Cl, err := sources.NewEngineClient(engRpc, log, nil, sources.EngineClientDefaultConfig(sd.RollupCfg))
 	require.NoError(t, err)
-	sequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), altda.Disabled, l2Cl, sd.RollupCfg, 0)
+	sequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), altda.Disabled, l2Cl, sd.RollupCfg, 0, nil)
 
 	batcher := NewL2Batcher(log, sd.RollupCfg, DefaultBatcherCfg(dp),
 		sequencer.RollupClient(), miner.EthClient(), seqEng.EthClient(), seqEng.EngineClient(t, sd.RollupCfg))
@@ -701,7 +701,7 @@ func ConflictingL2Blocks(gt *testing.T, deltaTimeOffset *hexutil.Uint64) {
 	require.NoError(t, err)
 	l1F, err := sources.NewL1Client(miner.RPCClient(), log, nil, sources.L1ClientDefaultConfig(sd.RollupCfg, false, sources.RPCKindStandard))
 	require.NoError(t, err)
-	altSequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), altda.Disabled, altSeqEngCl, sd.RollupCfg, 0)
+	altSequencer := NewL2Sequencer(t, log, l1F, miner.BlobStore(), altda.Disabled, altSeqEngCl, sd.RollupCfg, 0, nil)
 	altBatcher := NewL2Batcher(log, sd.RollupCfg, DefaultBatcherCfg(dp),
 		altSequencer.RollupClient(), miner.EthClient(), altSeqEng.EthClient(), altSeqEng.EngineClient(t, sd.RollupCfg))
 

op-e2e/actions/sync_test.go

@@ -819,7 +819,7 @@ func TestELSyncTransitionsToCLSyncAfterNodeRestart(gt *testing.T) {
 	PrepareELSyncedNode(t, miner, sequencer, seqEng, verifier, verEng, seqEngCl, batcher, dp)
 
 	// Create a new verifier which is essentially a new op-node with the sync mode of ELSync and default geth engine kind.
-	verifier = NewL2Verifier(t, captureLog, miner.L1Client(t, sd.RollupCfg), miner.BlobStore(), altda.Disabled, verifier.eng, sd.RollupCfg, &sync.Config{SyncMode: sync.ELSync}, defaultVerifierCfg().safeHeadListener)
+	verifier = NewL2Verifier(t, captureLog, miner.L1Client(t, sd.RollupCfg), miner.BlobStore(), altda.Disabled, verifier.eng, sd.RollupCfg, &sync.Config{SyncMode: sync.ELSync}, defaultVerifierCfg().safeHeadListener, nil)
 
 	// Build another 10 L1 blocks on the sequencer
 	for i := 0; i < 10; i++ {
@@ -861,7 +861,7 @@ func TestForcedELSyncCLAfterNodeRestart(gt *testing.T) {
 	PrepareELSyncedNode(t, miner, sequencer, seqEng, verifier, verEng, seqEngCl, batcher, dp)
 
 	// Create a new verifier which is essentially a new op-node with the sync mode of ELSync and erigon engine kind.
-	verifier2 := NewL2Verifier(t, captureLog, miner.L1Client(t, sd.RollupCfg), miner.BlobStore(), altda.Disabled, verifier.eng, sd.RollupCfg, &sync.Config{SyncMode: sync.ELSync, SupportsPostFinalizationELSync: true}, defaultVerifierCfg().safeHeadListener)
+	verifier2 := NewL2Verifier(t, captureLog, miner.L1Client(t, sd.RollupCfg), miner.BlobStore(), altda.Disabled, verifier.eng, sd.RollupCfg, &sync.Config{SyncMode: sync.ELSync, SupportsPostFinalizationELSync: true}, defaultVerifierCfg().safeHeadListener, nil)
 
 	// Build another 10 L1 blocks on the sequencer
 	for i := 0; i < 10; i++ {
@@ -1039,12 +1039,19 @@ func TestSpanBatchAtomicity_Consolidation(gt *testing.T) {
 	verifier.l2PipelineIdle = false
 	for !verifier.l2PipelineIdle {
 		// wait for next pending block
-		verifier.ActL2EventsUntil(t, event.Any(
-			event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent]), 1000, false)
+		verifier.ActL2EventsUntil(t, func(ev event.Event) bool {
+			if event.Is[engine2.SafeDerivedEvent](ev) { // safe updates should only happen once the pending-safe reaches the target.
+				t.Fatal("unexpected next safe update")
+			}
+			return event.Any(event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent])(ev)
+		}, 1000, false)
 		if verifier.L2PendingSafe().Number < targetHeadNumber {
 			// If the span batch is not fully processed, the safe head must not advance.
 			require.Equal(t, verifier.L2Safe().Number, uint64(0))
 		} else {
+			// Make sure we do the post-processing of what safety updates might happen
+			// after the pending-safe event, before the next pending-safe event.
+			verifier.ActL2EventsUntil(t, event.Is[engine2.PendingSafeUpdateEvent], 100, true)
 			// Once the span batch is fully processed, the safe head must advance to the end of span batch.
 			require.Equal(t, verifier.L2Safe().Number, targetHeadNumber)
 			require.Equal(t, verifier.L2Safe(), verifier.L2PendingSafe())
@@ -1088,12 +1095,19 @@ func TestSpanBatchAtomicity_ForceAdvance(gt *testing.T) {
 	verifier.l2PipelineIdle = false
 	for !verifier.l2PipelineIdle {
 		// wait for next pending block
-		verifier.ActL2EventsUntil(t, event.Any(
-			event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent]), 1000, false)
+		verifier.ActL2EventsUntil(t, func(ev event.Event) bool {
+			if event.Is[engine2.SafeDerivedEvent](ev) { // safe updates should only happen once the pending-safe reaches the target.
+				t.Fatal("unexpected next safe update")
+			}
+			return event.Any(event.Is[engine2.PendingSafeUpdateEvent], event.Is[derive.DeriverIdleEvent])(ev)
+		}, 1000, false)
 		if verifier.L2PendingSafe().Number < targetHeadNumber {
 			// If the span batch is not fully processed, the safe head must not advance.
 			require.Equal(t, verifier.L2Safe().Number, uint64(0))
 		} else {
+			// Make sure we do the post-processing of what safety updates might happen
+			// after the pending-safe event, before the next pending-safe event.
+			verifier.ActL2EventsUntil(t, event.Is[engine2.PendingSafeUpdateEvent], 100, true)
 			// Once the span batch is fully processed, the safe head must advance to the end of span batch.
 			require.Equal(t, verifier.L2Safe().Number, targetHeadNumber)
 			require.Equal(t, verifier.L2Safe(), verifier.L2PendingSafe())