interop: add strong consistency guarantees (ethereum-optimism#19505)

* interop: add consistency guarantees with minimal diff

Add 5 targeted consistency fixes to the supernode interop activity,
restructured around an observe/decide/execute pattern with a pure
Decide() function for testability.

Fixes:
- Crash-safe invalidations via WAL pattern in VerifiedDB
- L1 fork consistency check via byNumberConsistencyChecker
- Accepted snapshot re-verification (L2 drift + L1 canonicality)
- Deny-list decision provenance via DecisionTimestamp + pruning
- Queued resets to eliminate race between Reset() and main loop

Additional hardening:
- Stale logsDB detection in loadLogs (same-height hash mismatch)
- Idempotent startup recovery (WAL replay, deny-list prune, logsDB trim)
- Deterministic chain iteration (sorted by ChainID)
- VerifiedDB internal locking (sync.RWMutex)
- RewindEngine atomic guard against concurrent resets
- ErrL1AtSafeHeadNotFound mapped to ethereum.NotFound
- WAL preserved on invalidation failure for retry
- VerifiedBlockAtL1 scan bounded by activationTimestamp

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* interop: unify recovery and transition apply path

Squashed follow-up commits:
- interop: unify recovery and transition apply path
- interop: remove dead wrappers and stale comments
- interop: make rewind transition building explicit
- interop: fail closed on invalid rewind WAL
- interop: report all rewind apply errors
- chain-container: drop unused ClearDenied hook
- chain-container: remove dead denylist clear helper
- chain-container: restrict rewinds to interop callers
- Revert "chain-container: restrict rewinds to interop callers"
- chain-container: document reorg footgun
- supernode: make interop reset callback a silent no-op
- interop: clarify block-time warning

Signed-off-by: Karl Floersch <karl@oplabs.co>

* Update op-supernode/supernode/chain_container/chain_container.go

Co-authored-by: George Knee <georgeknee@googlemail.com>

* interop: address review nits and add follow-up issue ref

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Signed-off-by: Karl Floersch <karl@oplabs.co>
Co-authored-by: opsuperchain <opsuperchain@slop.bot>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
Co-authored-by: George Knee <georgeknee@googlemail.com>

Author: 4 people

op-supernode/supernode/activity/interop/algo.go

@@ -73,39 +73,49 @@ func (i *Interop) verifyInteropMessages(ts uint64, blocksAtTimestamp blockPerCha
 	}
 
 	for chainID, expectedBlock := range blocksAtTimestamp {
-		db, ok := i.logsDBs[chainID]
-		if !ok {
-			// Skip chains that we don't have a logsDB for
-			// This can happen if blocksAtTimestamp includes chains not registered with the interop activity
-			continue
-		}
+		var (
+			blockRef eth.BlockRef
+			execMsgs map[uint32]*types.ExecutingMessage
+			err      error
+		)
+		if frontierBlock, ok := i.frontierView.block(chainID); ok {
+			blockRef = frontierBlock.ref
+			execMsgs = frontierBlock.execMsgs
+		} else {
+			db, ok := i.logsDBs[chainID]
+			if !ok {
+				// Skip chains that we don't have a logsDB for
+				// This can happen if blocksAtTimestamp includes chains not registered with the interop activity
+				continue
+			}
 
-		// Get the block from the logsDB
-		blockRef, _, execMsgs, err := db.OpenBlock(expectedBlock.Number)
-		if err != nil {
-			// OpenBlock fails for the first block in the DB because it tries to find the parent.
-			// Handle this by checking if this is the first sealed block and using FirstSealedBlock instead.
-			if errors.Is(err, types.ErrSkipped) {
-				firstBlock, firstErr := db.FirstSealedBlock()
-				if firstErr != nil {
-					return Result{}, fmt.Errorf("chain %s: failed to open block %d and failed to get first block: %w", chainID, expectedBlock.Number, err)
-				}
-				if firstBlock.Number == expectedBlock.Number {
-					// This is the first block in the logsDB. Use FirstSealedBlock info.
-					// The first block has no executing messages (since we can't verify them without prior data).
-					if firstBlock.Hash != expectedBlock.Hash {
-						i.log.Warn("first block hash mismatch",
-							"chain", chainID,
-							"expected", expectedBlock.Hash,
-							"got", firstBlock.Hash,
-						)
-						result.InvalidHeads[chainID] = expectedBlock
+			// Get the block from the logsDB
+			blockRef, _, execMsgs, err = db.OpenBlock(expectedBlock.Number)
+			if err != nil {
+				// OpenBlock fails for the first block in the DB because it tries to find the parent.
+				// Handle this by checking if this is the first sealed block and using FirstSealedBlock instead.
+				if errors.Is(err, types.ErrSkipped) {
+					firstBlock, firstErr := db.FirstSealedBlock()
+					if firstErr != nil {
+						return Result{}, fmt.Errorf("chain %s: failed to open block %d and failed to get first block: %w", chainID, expectedBlock.Number, err)
+					}
+					if firstBlock.Number == expectedBlock.Number {
+						// This is the first block in the logsDB. Use FirstSealedBlock info.
+						// The first block has no executing messages (since we can't verify them without prior data).
+						if firstBlock.Hash != expectedBlock.Hash {
+							i.log.Warn("first block hash mismatch",
+								"chain", chainID,
+								"expected", expectedBlock.Hash,
+								"got", firstBlock.Hash,
+							)
+							result.InvalidHeads[chainID] = expectedBlock
+						}
+						result.L2Heads[chainID] = expectedBlock
+						continue
 					}
-					result.L2Heads[chainID] = expectedBlock
-					continue
 				}
+				return Result{}, fmt.Errorf("chain %s: failed to open block %d: %w", chainID, expectedBlock.Number, err)
 			}
-			return Result{}, fmt.Errorf("chain %s: failed to open block %d: %w", chainID, expectedBlock.Number, err)
 		}
 
 		// Verify the block hash matches what we expect
@@ -177,6 +187,14 @@ func (i *Interop) verifyExecutingMessage(executingChain eth.ChainID, executingTi
 		Checksum:  execMsg.Checksum,
 	}
 
+	// Same-timestamp dependencies may live in the current frontier view rather
+	// than accepted-history logsDB.
+	if execMsg.Timestamp == executingTimestamp {
+		if _, ok := i.frontierView.contains(execMsg.ChainID, query); ok {
+			return nil
+		}
+	}
+
 	// Check if the initiating message exists in the source chain's logsDB
 	_, err := sourceDB.Contains(query)
 	return err

op-supernode/supernode/activity/interop/algo_test.go

@@ -929,9 +929,12 @@ func (m *algoMockChain) RewindEngine(ctx context.Context, timestamp uint64, inva
 	return nil
 }
 func (m *algoMockChain) BlockTime() uint64 { return 1 }
-func (m *algoMockChain) InvalidateBlock(ctx context.Context, height uint64, payloadHash common.Hash) (bool, error) {
+func (m *algoMockChain) InvalidateBlock(ctx context.Context, height uint64, payloadHash common.Hash, decisionTimestamp uint64) (bool, error) {
 	return false, nil
 }
+func (m *algoMockChain) PruneDeniedAtOrAfterTimestamp(timestamp uint64) (map[uint64][]common.Hash, error) {
+	return nil, nil
+}
 func (m *algoMockChain) IsDenied(height uint64, payloadHash common.Hash) (bool, error) {
 	return false, nil
 }

op-supernode/supernode/activity/interop/checker.go

@@ -0,0 +1,42 @@
+package interop
+
+import (
+	"context"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+)
+
+// l1ByNumberSource provides L1 block lookups by number for consistency checking.
+type l1ByNumberSource interface {
+	L1BlockRefByNumber(ctx context.Context, num uint64) (eth.L1BlockRef, error)
+}
+
+// byNumberConsistencyChecker verifies that a set of L1 block IDs all belong to
+// the same L1 fork by comparing each against the canonical chain.
+type byNumberConsistencyChecker struct {
+	l1 l1ByNumberSource
+}
+
+func newByNumberConsistencyChecker(l1 l1ByNumberSource) *byNumberConsistencyChecker {
+	if l1 == nil {
+		return nil
+	}
+	return &byNumberConsistencyChecker{l1: l1}
+}
+
+// SameL1Chain returns true if all non-zero heads belong to the same canonical L1 chain.
+func (c *byNumberConsistencyChecker) SameL1Chain(ctx context.Context, heads []eth.BlockID) (bool, error) {
+	for _, head := range heads {
+		if head == (eth.BlockID{}) {
+			continue
+		}
+		canonical, err := c.l1.L1BlockRefByNumber(ctx, head.Number)
+		if err != nil {
+			return false, err
+		}
+		if canonical.Hash != head.Hash {
+			return false, nil
+		}
+	}
+	return true, nil
+}

op-supernode/supernode/activity/interop/checker_test.go

@@ -0,0 +1,77 @@
+package interop
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+type mockL1Source struct {
+	blocks map[uint64]eth.L1BlockRef
+}
+
+func (m *mockL1Source) L1BlockRefByNumber(ctx context.Context, num uint64) (eth.L1BlockRef, error) {
+	ref, ok := m.blocks[num]
+	if !ok {
+		return eth.L1BlockRef{}, fmt.Errorf("block %d not found", num)
+	}
+	return ref, nil
+}
+
+func TestByNumberConsistencyChecker_SameL1Chain(t *testing.T) {
+	t.Parallel()
+
+	hashA := common.HexToHash("0xaaaa")
+	hashB := common.HexToHash("0xbbbb")
+
+	source := &mockL1Source{
+		blocks: map[uint64]eth.L1BlockRef{
+			100: {Hash: hashA, Number: 100},
+			200: {Hash: hashB, Number: 200},
+		},
+	}
+	checker := newByNumberConsistencyChecker(source)
+
+	t.Run("all match canonical", func(t *testing.T) {
+		same, err := checker.SameL1Chain(context.Background(), []eth.BlockID{
+			{Hash: hashA, Number: 100},
+			{Hash: hashB, Number: 200},
+		})
+		require.NoError(t, err)
+		require.True(t, same)
+	})
+
+	t.Run("mismatch detected", func(t *testing.T) {
+		same, err := checker.SameL1Chain(context.Background(), []eth.BlockID{
+			{Hash: hashA, Number: 100},
+			{Hash: common.HexToHash("0xdead"), Number: 200}, // wrong hash
+		})
+		require.NoError(t, err)
+		require.False(t, same)
+	})
+
+	t.Run("zero block IDs skipped", func(t *testing.T) {
+		same, err := checker.SameL1Chain(context.Background(), []eth.BlockID{
+			{},
+			{Hash: hashA, Number: 100},
+			{},
+		})
+		require.NoError(t, err)
+		require.True(t, same)
+	})
+
+	t.Run("empty heads list", func(t *testing.T) {
+		same, err := checker.SameL1Chain(context.Background(), []eth.BlockID{})
+		require.NoError(t, err)
+		require.True(t, same)
+	})
+
+	t.Run("nil checker returns nil", func(t *testing.T) {
+		nilChecker := newByNumberConsistencyChecker(nil)
+		require.Nil(t, nilChecker)
+	})
+}

op-supernode/supernode/activity/interop/cycle.go

@@ -177,6 +177,13 @@ func (i *Interop) verifyCycleMessages(ts uint64, blocksAtTimestamp map[eth.Chain
 	// collect all EMs for the given blocks per chain
 	chainEMs := make(map[eth.ChainID]map[uint32]*types.ExecutingMessage)
 	for chainID, blockID := range blocksAtTimestamp {
+		if frontierBlock, ok := i.frontierView.block(chainID); ok {
+			if frontierBlock.ref.Time == ts {
+				chainEMs[chainID] = frontierBlock.execMsgs
+			}
+			continue
+		}
+
 		db, ok := i.logsDBs[chainID]
 		if !ok {
 			// Chain not in logsDBs - skip it for cycle verification

op-supernode/supernode/activity/interop/decide_test.go

@@ -0,0 +1,130 @@
+package interop
+
+import (
+	"testing"
+
+	"github.com/ethereum-optimism/optimism/op-service/eth"
+	"github.com/ethereum/go-ethereum/common"
+	"github.com/stretchr/testify/require"
+)
+
+func TestCheckPreconditions(t *testing.T) {
+	t.Parallel()
+
+	tests := []struct {
+		name string
+		obs  RoundObservation
+		want *Decision
+	}{
+		{
+			name: "pause when paused",
+			obs: RoundObservation{
+				Paused:       true,
+				ChainsReady:  true,
+				L1Consistent: true,
+			},
+			want: ptrDecision(DecisionWait),
+		},
+		{
+			name: "wait when chains not ready",
+			obs: RoundObservation{
+				ChainsReady: false,
+			},
+			want: ptrDecision(DecisionWait),
+		},
+		{
+			name: "rewind when L1 inconsistent",
+			obs: RoundObservation{
+				ChainsReady:  true,
+				L1Consistent: false,
+			},
+			want: ptrDecision(DecisionRewind),
+		},
+		{
+			name: "proceed when preconditions are satisfied",
+			obs: RoundObservation{
+				ChainsReady:  true,
+				L1Consistent: true,
+			},
+			want: nil,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			got := checkPreconditions(tt.obs)
+			if tt.want == nil {
+				require.Nil(t, got)
+				return
+			}
+			require.NotNil(t, got)
+			require.Equal(t, *tt.want, got.Decision)
+		})
+	}
+}
+
+func TestDecideVerifiedResult(t *testing.T) {
+	t.Parallel()
+
+	ts := uint64(1000)
+	validResult := Result{
+		Timestamp:   ts + 1,
+		L1Inclusion: eth.BlockID{Hash: common.HexToHash("0xl1"), Number: 50},
+		L2Heads: map[eth.ChainID]eth.BlockID{
+			eth.ChainIDFromUInt64(1): {Hash: common.HexToHash("0xa"), Number: 100},
+		},
+	}
+	invalidResult := Result{
+		Timestamp:   ts + 1,
+		L1Inclusion: eth.BlockID{Hash: common.HexToHash("0xl1"), Number: 50},
+		L2Heads: map[eth.ChainID]eth.BlockID{
+			eth.ChainIDFromUInt64(1): {Hash: common.HexToHash("0xa"), Number: 100},
+		},
+		InvalidHeads: map[eth.ChainID]eth.BlockID{
+			eth.ChainIDFromUInt64(2): {Hash: common.HexToHash("0xbad"), Number: 200},
+		},
+	}
+
+	tests := []struct {
+		name     string
+		verified Result
+		want     Decision
+	}{
+		{
+			name:     "wait when verification result is empty",
+			verified: Result{},
+			want:     DecisionWait,
+		},
+		{
+			name:     "invalidate on invalid verification",
+			verified: invalidResult,
+			want:     DecisionInvalidate,
+		},
+		{
+			name:     "advance on valid verification",
+			verified: validResult,
+			want:     DecisionAdvance,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			t.Parallel()
+			got := decideVerifiedResult(RoundObservation{}, tt.verified)
+			require.Equal(t, tt.want, got.Decision, "unexpected decision")
+
+			if tt.want == DecisionInvalidate {
+				require.NotEmpty(t, got.Result.InvalidHeads, "invalidate should carry invalid heads")
+			}
+			if tt.want == DecisionAdvance {
+				require.False(t, got.Result.IsEmpty(), "advance should carry result")
+				require.True(t, got.Result.IsValid(), "advance result should be valid")
+			}
+		})
+	}
+}
+
+func ptrDecision(d Decision) *Decision {
+	return &d
+}