op-supervisor: bugfix timestamp Check (#13676)

* op-supervisor: bugfix timestamp Check

* supervisor: include Invalid in SafetyLevel Validation

* supervisor: check if BlockSeal is found when iterator stops

* Don't rewind iterator if Traversal yields ErrStop

* Add Timestamp Invariant Test to E2E

* update unit test

* pr comments re:error message

---------
Co-authored-by: protolambda <proto@protolambda.com>

op-e2e/interop/interop_test.go

@@ -174,9 +174,8 @@ func TestInterop_EmitLogs(t *testing.T) {
 
 		supervisor := s2.SupervisorClient()
 
-		// requireMessage checks the safety level of a log against the supervisor
-		// it also checks that the error is as expected
-		requireMessage := func(chainID string, log gethTypes.Log, expectedSafety types.SafetyLevel, expectedError error) {
+		// helper function to turn a log into an identifier and the expected hash of the payload
+		logToIdentifier := func(chainID string, log gethTypes.Log) (types.Identifier, common.Hash) {
 			client := s2.L2GethClient(chainID)
 			// construct the expected hash of the log's payload
 			// (topics concatenated with data)
@@ -201,23 +200,35 @@ func TestInterop_EmitLogs(t *testing.T) {
 				Timestamp:   block.Time(),
 				ChainID:     types.ChainIDFromBig(s2.ChainID(chainID)),
 			}
-
-			safety, err := supervisor.CheckMessage(context.Background(),
-				identifier,
-				expectedHash,
-			)
-			require.ErrorIs(t, err, expectedError)
-			// the supervisor could progress the safety level more quickly than we expect,
-			// which is why we check for a minimum safety level
-			require.True(t, safety.AtLeastAsSafe(expectedSafety), "log: %v should be at least %s, but is %s", log, expectedSafety.String(), safety.String())
+			return identifier, expectedHash
 		}
+
 		// all logs should be cross-safe
 		for _, log := range logsA {
-			requireMessage(chainA, log, types.CrossSafe, nil)
+			identifier, expectedHash := logToIdentifier(chainA, log)
+			safety, err := supervisor.CheckMessage(context.Background(), identifier, expectedHash)
+			require.NoError(t, err)
+			// the supervisor could progress the safety level more quickly than we expect,
+			// which is why we check for a minimum safety level
+			require.True(t, safety.AtLeastAsSafe(types.CrossSafe), "log: %v should be at least Cross-Safe, but is %s", log, safety.String())
 		}
 		for _, log := range logsB {
-			requireMessage(chainB, log, types.CrossSafe, nil)
+			identifier, expectedHash := logToIdentifier(chainB, log)
+			safety, err := supervisor.CheckMessage(context.Background(), identifier, expectedHash)
+			require.NoError(t, err)
+			// the supervisor could progress the safety level more quickly than we expect,
+			// which is why we check for a minimum safety level
+			require.True(t, safety.AtLeastAsSafe(types.CrossSafe), "log: %v should be at least Cross-Safe, but is %s", log, safety.String())
 		}
+
+		// a log should be invalid if the timestamp is incorrect
+		identifier, expectedHash := logToIdentifier(chainA, logsA[0])
+		// make the timestamp incorrect
+		identifier.Timestamp = 333
+		safety, err := supervisor.CheckMessage(context.Background(), identifier, expectedHash)
+		require.NoError(t, err)
+		require.Equal(t, types.Invalid, safety)
+
 	}
 	config := SuperSystemConfig{
 		mempoolFiltering: false,

op-supervisor/supervisor/backend/backend.go

@@ -385,7 +385,7 @@ func (su *SupervisorBackend) CheckMessage(identifier types.Identifier, payloadHa
 	chainID := identifier.ChainID
 	blockNum := identifier.BlockNumber
 	logIdx := identifier.LogIndex
-	_, err := su.chainDBs.Check(chainID, blockNum, logIdx, logHash)
+	_, err := su.chainDBs.Check(chainID, blockNum, identifier.Timestamp, logIdx, logHash)
 	if errors.Is(err, types.ErrFuture) {
 		su.logger.Debug("Future message", "identifier", identifier, "payloadHash", payloadHash, "err", err)
 		return types.LocalUnsafe, nil

op-supervisor/supervisor/backend/cross/safe_start.go

@@ -12,7 +12,7 @@ import (
 )
 
 type SafeStartDeps interface {
-	Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error)
+	Check(chain types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error)
 
 	CrossDerivedFrom(chainID types.ChainID, derived eth.BlockID) (derivedFrom types.BlockSeal, err error)
 
@@ -61,7 +61,7 @@ func CrossSafeHazards(d SafeStartDeps, chainID types.ChainID, inL1DerivedFrom et
 		if msg.Timestamp < candidate.Timestamp {
 			// If timestamp is older: invariant ensures non-cyclic ordering relative to other messages.
 			// Check that the block that they are included in is cross-safe already.
-			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.LogIdx, msg.Hash)
+			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.Timestamp, msg.LogIdx, msg.Hash)
 			if err != nil {
 				return nil, fmt.Errorf("executing msg %s failed check: %w", msg, err)
 			}
@@ -80,7 +80,7 @@ func CrossSafeHazards(d SafeStartDeps, chainID types.ChainID, inL1DerivedFrom et
 			// Thus check that it was included in a local-safe block,
 			// and then proceed with transitive block checks,
 			// to ensure the local block we depend on is becoming cross-safe also.
-			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.LogIdx, msg.Hash)
+			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.Timestamp, msg.LogIdx, msg.Hash)
 			if err != nil {
 				return nil, fmt.Errorf("executing msg %s failed check: %w", msg, err)
 			}

op-supervisor/supervisor/backend/cross/safe_start_test.go

@@ -322,7 +322,7 @@ type mockSafeStartDeps struct {
 	derivedFromFn func() (derivedFrom types.BlockSeal, err error)
 }
 
-func (m *mockSafeStartDeps) Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
+func (m *mockSafeStartDeps) Check(chain types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
 	if m.checkFn != nil {
 		return m.checkFn()
 	}

op-supervisor/supervisor/backend/cross/safe_update_test.go

@@ -406,7 +406,7 @@ func (m *mockCrossSafeDeps) CrossDerivedFrom(chainID types.ChainID, derived eth.
 	return types.BlockSeal{}, nil
 }
 
-func (m *mockCrossSafeDeps) Check(chainID types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
+func (m *mockCrossSafeDeps) Check(chainID types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
 	if m.checkFn != nil {
 		return m.checkFn(chainID, blockNum, logIdx, logHash)
 	}

op-supervisor/supervisor/backend/cross/unsafe_start.go

@@ -12,7 +12,7 @@ import (
 )
 
 type UnsafeStartDeps interface {
-	Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error)
+	Check(chain types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error)
 
 	IsCrossUnsafe(chainID types.ChainID, block eth.BlockID) error
 
@@ -61,13 +61,16 @@ func CrossUnsafeHazards(d UnsafeStartDeps, chainID types.ChainID,
 		if msg.Timestamp < candidate.Timestamp {
 			// If timestamp is older: invariant ensures non-cyclic ordering relative to other messages.
 			// Check that the block that they are included in is cross-safe already.
-			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.LogIdx, msg.Hash)
+			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.Timestamp, msg.LogIdx, msg.Hash)
 			if err != nil {
 				return nil, fmt.Errorf("executing msg %s failed check: %w", msg, err)
 			}
 			if err := d.IsCrossUnsafe(initChainID, includedIn.ID()); err != nil {
 				return nil, fmt.Errorf("msg %s included in non-cross-unsafe block %s: %w", msg, includedIn, err)
 			}
+			if includedIn.Timestamp != msg.Timestamp {
+				return nil, fmt.Errorf("executing msg %s exists, but has different timestamp than block %s: %w", msg, includedIn, types.ErrConflict)
+			}
 		} else if msg.Timestamp == candidate.Timestamp {
 			// If timestamp is equal: we have to inspect ordering of individual
 			// log events to ensure non-cyclic cross-chain message ordering.
@@ -75,7 +78,7 @@ func CrossUnsafeHazards(d UnsafeStartDeps, chainID types.ChainID,
 			// Thus check that it was included in a local-unsafe block,
 			// and then proceed with transitive block checks,
 			// to ensure the local block we depend on is becoming cross-unsafe also.
-			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.LogIdx, msg.Hash)
+			includedIn, err := d.Check(initChainID, msg.BlockNum, msg.Timestamp, msg.LogIdx, msg.Hash)
 			if err != nil {
 				return nil, fmt.Errorf("executing msg %s failed check: %w", msg, err)
 			}

op-supervisor/supervisor/backend/cross/unsafe_start_test.go

@@ -244,7 +244,7 @@ func TestCrossUnsafeHazards(t *testing.T) {
 		usd.deps = mockDependencySet{}
 		chainID := types.ChainIDFromUInt64(0)
 		candidate := types.BlockSeal{Timestamp: 2}
-		em1 := &types.ExecutingMessage{Chain: types.ChainIndex(0), Timestamp: 1}
+		em1 := &types.ExecutingMessage{Chain: types.ChainIndex(0), Timestamp: 0}
 		execMsgs := []*types.ExecutingMessage{em1}
 		// when there is one execMsg, and the timestamp is less than the candidate,
 		// and IsCrossUnsafe returns no error,
@@ -261,7 +261,7 @@ type mockUnsafeStartDeps struct {
 	isCrossUnsafeFn func() error
 }
 
-func (m *mockUnsafeStartDeps) Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
+func (m *mockUnsafeStartDeps) Check(chain types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
 	if m.checkFn != nil {
 		return m.checkFn()
 	}

op-supervisor/supervisor/backend/cross/unsafe_update_test.go

@@ -146,7 +146,7 @@ func TestCrossUnsafeUpdate(t *testing.T) {
 		usd.openBlockFn = func(chainID types.ChainID, blockNum uint64) (ref eth.BlockRef, logCount uint32, execMsgs map[uint32]*types.ExecutingMessage, err error) {
 			return bl, 3, map[uint32]*types.ExecutingMessage{1: em1, 2: em2}, nil
 		}
-		usd.checkFn = func(chainID types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
+		usd.checkFn = func(chainID types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
 			return types.BlockSeal{Number: 1, Timestamp: 1}, nil
 		}
 		usd.deps = mockDependencySet{}
@@ -193,7 +193,7 @@ type mockCrossUnsafeDeps struct {
 	crossUnsafeFn       func(chainID types.ChainID) (types.BlockSeal, error)
 	openBlockFn         func(chainID types.ChainID, blockNum uint64) (ref eth.BlockRef, logCount uint32, execMsgs map[uint32]*types.ExecutingMessage, err error)
 	updateCrossUnsafeFn func(chain types.ChainID, crossUnsafe types.BlockSeal) error
-	checkFn             func(chainID types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error)
+	checkFn             func(chainID types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error)
 }
 
 func (m *mockCrossUnsafeDeps) CrossUnsafe(chainID types.ChainID) (derived types.BlockSeal, err error) {
@@ -207,9 +207,9 @@ func (m *mockCrossUnsafeDeps) DependencySet() depset.DependencySet {
 	return m.deps
 }
 
-func (m *mockCrossUnsafeDeps) Check(chainID types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
+func (m *mockCrossUnsafeDeps) Check(chainID types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (types.BlockSeal, error) {
 	if m.checkFn != nil {
-		return m.checkFn(chainID, blockNum, logIdx, logHash)
+		return m.checkFn(chainID, blockNum, timestamp, logIdx, logHash)
 	}
 	return types.BlockSeal{}, nil
 }

op-supervisor/supervisor/backend/db/logs/db.go

@@ -315,7 +315,10 @@ func (db *DB) Contains(blockNum uint64, logIdx uint32, logHash common.Hash) (typ
 		panic("expected iterator to stop with error")
 	}
 	if errors.Is(err, types.ErrStop) {
-		h, n, _ := iter.SealedBlock()
+		h, n, ok := iter.SealedBlock()
+		if !ok {
+			return types.BlockSeal{}, fmt.Errorf("iterator stopped but no sealed block found")
+		}
 		timestamp, _ := iter.SealedTimestamp()
 		return types.BlockSeal{
 			Hash:      h,

op-supervisor/supervisor/backend/db/logs/iterator.go

@@ -121,6 +121,10 @@ func (i *iterator) TraverseConditional(fn traverseConditionalFn) error {
 			continue
 		}
 		if err := fn(&i.current); err != nil {
+			// don't rewind to the snapshot if the error is ErrStop
+			if errors.Is(err, types.ErrStop) {
+				return err
+			}
 			i.current = snapshot
 			return err
 		}

op-supervisor/supervisor/backend/db/query.go

@@ -210,12 +210,24 @@ func (db *ChainsDB) CrossDerivedFromBlockRef(chainID types.ChainID, derived eth.
 
 // Check calls the underlying logDB to determine if the given log entry exists at the given location.
 // If the block-seal of the block that includes the log is known, it is returned. It is fully zeroed otherwise, if the block is in-progress.
-func (db *ChainsDB) Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
+func (db *ChainsDB) Check(chain types.ChainID, blockNum uint64, timestamp uint64, logIdx uint32, logHash common.Hash) (includedIn types.BlockSeal, err error) {
 	logDB, ok := db.logDBs.Get(chain)
 	if !ok {
 		return types.BlockSeal{}, fmt.Errorf("%w: %v", types.ErrUnknownChain, chain)
 	}
-	return logDB.Contains(blockNum, logIdx, logHash)
+	includedIn, err = logDB.Contains(blockNum, logIdx, logHash)
+	if err != nil {
+		return types.BlockSeal{}, err
+	}
+	if includedIn.Timestamp != timestamp {
+		return types.BlockSeal{},
+			fmt.Errorf("log exists in block %s, but block timestamp %d does not match %d: %w",
+				includedIn,
+				includedIn.Timestamp,
+				timestamp,
+				types.ErrConflict)
+	}
+	return includedIn, nil
 }
 
 // OpenBlock returns the Executing Messages for the block at the given number on the given chain.

op-supervisor/supervisor/types/types.go

@@ -105,9 +105,10 @@ func (lvl SafetyLevel) String() string {
 	return string(lvl)
 }
 
-func (lvl SafetyLevel) Valid() bool {
+// Validate returns true if the SafetyLevel is one of the recognized levels
+func (lvl SafetyLevel) Validate() bool {
 	switch lvl {
-	case Finalized, CrossSafe, LocalSafe, CrossUnsafe, LocalUnsafe:
+	case Invalid, Finalized, CrossSafe, LocalSafe, CrossUnsafe, LocalUnsafe:
 		return true
 	default:
 		return false
@@ -123,7 +124,7 @@ func (lvl *SafetyLevel) UnmarshalText(text []byte) error {
 		return errors.New("cannot unmarshal into nil SafetyLevel")
 	}
 	x := SafetyLevel(text)
-	if !x.Valid() {
+	if !x.Validate() {
 		return fmt.Errorf("unrecognized safety level: %q", text)
 	}
 	*lvl = x