supervisor: APIs (#11586)

* implement CheckMessage

* Add CheckBlock API

* Cleanup and Unit Tests

* Add Invalid Safety Type

* Use *ChainsDB instead of ChainsDB for SafetyCheckers

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

op-supervisor/supervisor/backend/backend.go

@@ -16,6 +16,7 @@ import (
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/backend/db/heads"
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/backend/db/logs"
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/backend/source"
+	backendTypes "github.com/ethereum-optimism/optimism/op-supervisor/supervisor/backend/types"
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/frontend"
 	"github.com/ethereum-optimism/optimism/op-supervisor/supervisor/types"
 	"github.com/ethereum/go-ethereum/common"
@@ -134,11 +135,51 @@ func (su *SupervisorBackend) Close() error {
 }
 
 func (su *SupervisorBackend) CheckMessage(identifier types.Identifier, payloadHash common.Hash) (types.SafetyLevel, error) {
-	// TODO(protocol-quest#288): hook up to logdb lookup
-	return types.CrossUnsafe, nil
+	chainID := identifier.ChainID
+	blockNum := identifier.BlockNumber
+	logIdx := identifier.LogIndex
+	ok, i, err := su.db.Check(chainID, blockNum, uint32(logIdx), backendTypes.TruncateHash(payloadHash))
+	if err != nil {
+		return types.Invalid, fmt.Errorf("failed to check log: %w", err)
+	}
+	if !ok {
+		return types.Invalid, nil
+	}
+	safest := types.CrossUnsafe
+	// at this point we have the log entry, and we can check if it is safe by various criteria
+	for _, checker := range []db.SafetyChecker{
+		db.NewSafetyChecker(types.Unsafe, su.db),
+		db.NewSafetyChecker(types.Safe, su.db),
+		db.NewSafetyChecker(types.Finalized, su.db),
+	} {
+		if i <= checker.CrossHeadForChain(chainID) {
+			safest = checker.SafetyLevel()
+		}
+	}
+	return safest, nil
 }
 
+// CheckBlock checks if the block is safe according to the safety level
+// The block is considered safe if all logs in the block are safe
+// this is decided by finding the last log in the block and
 func (su *SupervisorBackend) CheckBlock(chainID *hexutil.U256, blockHash common.Hash, blockNumber hexutil.Uint64) (types.SafetyLevel, error) {
-	// TODO(protocol-quest#288): hook up to logdb lookup
-	return types.CrossUnsafe, nil
+	// TODO(#11612): this function ignores blockHash and assumes that the block in the db is the one we are looking for
+	// In order to check block hash, the database must *always* insert a block hash checkpoint, which is not currently done
+	safest := types.CrossUnsafe
+	// find the last log index in the block
+	i, err := su.db.LastLogInBlock(types.ChainID(*chainID), uint64(blockNumber))
+	if err != nil {
+		return types.Invalid, fmt.Errorf("failed to scan block: %w", err)
+	}
+	// at this point we have the extent of the block, and we can check if it is safe by various criteria
+	for _, checker := range []db.SafetyChecker{
+		db.NewSafetyChecker(types.Unsafe, su.db),
+		db.NewSafetyChecker(types.Safe, su.db),
+		db.NewSafetyChecker(types.Finalized, su.db),
+	} {
+		if i <= checker.CrossHeadForChain(types.ChainID(*chainID)) {
+			safest = checker.SafetyLevel()
+		}
+	}
+	return safest, nil
 }

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

@@ -26,6 +26,7 @@ type LogStorage interface {
 	Rewind(newHeadBlockNum uint64) error
 	LatestBlockNum() uint64
 	ClosestBlockInfo(blockNum uint64) (uint64, backendTypes.TruncatedHash, error)
+	ClosestBlockIterator(blockNum uint64) (logs.Iterator, error)
 	Contains(blockNum uint64, logIdx uint32, loghash backendTypes.TruncatedHash) (bool, entrydb.EntryIdx, error)
 	LastCheckpointBehind(entrydb.EntryIdx) (logs.Iterator, error)
 	NextExecutingMessage(logs.Iterator) (backendTypes.ExecutingMessage, error)
@@ -67,9 +68,9 @@ func (db *ChainsDB) Resume() error {
 func (db *ChainsDB) StartCrossHeadMaintenance(ctx context.Context) {
 	go func() {
 		// create three safety checkers, one for each safety level
-		unsafeChecker := NewSafetyChecker(Unsafe, *db)
-		safeChecker := NewSafetyChecker(Safe, *db)
-		finalizedChecker := NewSafetyChecker(Finalized, *db)
+		unsafeChecker := NewSafetyChecker(Unsafe, db)
+		safeChecker := NewSafetyChecker(Safe, db)
+		finalizedChecker := NewSafetyChecker(Finalized, db)
 		// run the maintenance loop every 10 seconds for now
 		ticker := time.NewTicker(time.Second * 10)
 		for {
@@ -91,10 +92,19 @@ func (db *ChainsDB) StartCrossHeadMaintenance(ctx context.Context) {
 	}()
 }
 
+// Check calls the underlying logDB to determine if the given log entry is safe with respect to the checker's criteria.
+func (db *ChainsDB) Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash backendTypes.TruncatedHash) (bool, entrydb.EntryIdx, error) {
+	logDB, ok := db.logDBs[chain]
+	if !ok {
+		return false, 0, fmt.Errorf("%w: %v", ErrUnknownChain, chain)
+	}
+	return logDB.Contains(blockNum, logIdx, logHash)
+}
+
 // UpdateCrossSafeHeads updates the cross-heads of all chains
 // this is an example of how to use the SafetyChecker to update the cross-heads
 func (db *ChainsDB) UpdateCrossSafeHeads() error {
-	checker := NewSafetyChecker(Safe, *db)
+	checker := NewSafetyChecker(Safe, db)
 	return db.UpdateCrossHeads(checker)
 }
 
@@ -162,6 +172,46 @@ func (db *ChainsDB) UpdateCrossHeads(checker SafetyChecker) error {
 	return nil
 }
 
+// LastLogInBlock scans through the logs of the given chain starting from the given block number,
+// and returns the index of the last log entry in that block.
+func (db *ChainsDB) LastLogInBlock(chain types.ChainID, blockNum uint64) (entrydb.EntryIdx, error) {
+	logDB, ok := db.logDBs[chain]
+	if !ok {
+		return 0, fmt.Errorf("%w: %v", ErrUnknownChain, chain)
+	}
+	iter, err := logDB.ClosestBlockIterator(blockNum)
+	if err != nil {
+		return 0, fmt.Errorf("failed to get block iterator for chain %v: %w", chain, err)
+	}
+	ret := entrydb.EntryIdx(0)
+	// scan through using the iterator until the block number exceeds the target
+	for {
+		bn, index, _, err := iter.NextLog()
+		// if we have reached the end of the database, stop
+		if err == io.EOF {
+			break
+		}
+		// all other errors are fatal
+		if err != nil {
+			return 0, fmt.Errorf("failed to read next log entry for chain %v: %w", chain, err)
+		}
+		// if we are now beyond the target block, stop withour updating the return value
+		if bn > blockNum {
+			break
+		}
+		// only update the return value if the block number is the same
+		// it is possible the iterator started before the target block, or that the target block is not in the db
+		if bn == blockNum {
+			ret = entrydb.EntryIdx(index)
+		}
+	}
+	// if we never found the block, return an error
+	if ret == 0 {
+		return 0, fmt.Errorf("block %v not found in chain %v", blockNum, chain)
+	}
+	return ret, nil
+}
+
 // LatestBlockNum returns the latest block number that has been recorded to the logs db
 // for the given chain. It does not contain safety guarantees.
 func (db *ChainsDB) LatestBlockNum(chain types.ChainID) uint64 {

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

@@ -52,6 +52,118 @@ func TestChainsDB_Rewind(t *testing.T) {
 	})
 }
 
+func TestChainsDB_LastLogInBlock(t *testing.T) {
+	// using a chainID of 1 for simplicity
+	chainID := types.ChainIDFromUInt64(1)
+	// get default stubbed components
+	logDB, _, h := setupStubbedForUpdateHeads(chainID)
+	logDB.nextLogs = []nextLogResponse{
+		{10, 1, backendTypes.TruncatedHash{}, nil},
+		{10, 2, backendTypes.TruncatedHash{}, nil},
+		{10, 3, backendTypes.TruncatedHash{}, nil},
+		{10, 4, backendTypes.TruncatedHash{}, nil},
+		{11, 5, backendTypes.TruncatedHash{}, nil},
+	}
+
+	// The ChainsDB is real, but uses only stubbed components
+	db := NewChainsDB(
+		map[types.ChainID]LogStorage{
+			chainID: logDB},
+		&stubHeadStorage{h})
+
+	// LastLogInBlock is expected to:
+	// 1. get a block iterator for block 10 (stubbed)
+	// 2. scan through the iterator until the block number exceeds the target (10)
+	// 3. return the index of the last log in the block (4)
+	index, err := db.LastLogInBlock(chainID, 10)
+	require.NoError(t, err)
+	require.Equal(t, entrydb.EntryIdx(4), index)
+}
+
+func TestChainsDB_LastLogInBlockEOF(t *testing.T) {
+	// using a chainID of 1 for simplicity
+	chainID := types.ChainIDFromUInt64(1)
+	// get default stubbed components
+	logDB, _, h := setupStubbedForUpdateHeads(chainID)
+	logDB.nextLogs = []nextLogResponse{
+		{10, 5, backendTypes.TruncatedHash{}, nil},
+		{10, 6, backendTypes.TruncatedHash{}, nil},
+		{10, 7, backendTypes.TruncatedHash{}, nil},
+		{10, 8, backendTypes.TruncatedHash{}, nil},
+		{10, 9, backendTypes.TruncatedHash{}, nil},
+		{10, 10, backendTypes.TruncatedHash{}, nil},
+	}
+
+	// The ChainsDB is real, but uses only stubbed components
+	db := NewChainsDB(
+		map[types.ChainID]LogStorage{
+			chainID: logDB},
+		&stubHeadStorage{h})
+
+	// LastLogInBlock is expected to:
+	// 1. get a block iterator for block 10 (stubbed)
+	// 2. scan through the iterator and never find the target block
+	// return an error
+	index, err := db.LastLogInBlock(chainID, 10)
+	require.NoError(t, err)
+	require.Equal(t, entrydb.EntryIdx(10), index)
+}
+
+func TestChainsDB_LastLogInBlockNotFound(t *testing.T) {
+	// using a chainID of 1 for simplicity
+	chainID := types.ChainIDFromUInt64(1)
+	// get default stubbed components
+	logDB, _, h := setupStubbedForUpdateHeads(chainID)
+	logDB.nextLogs = []nextLogResponse{
+		{100, 5, backendTypes.TruncatedHash{}, nil},
+		{100, 6, backendTypes.TruncatedHash{}, nil},
+		{100, 7, backendTypes.TruncatedHash{}, nil},
+		{101, 8, backendTypes.TruncatedHash{}, nil},
+		{101, 9, backendTypes.TruncatedHash{}, nil},
+		{101, 10, backendTypes.TruncatedHash{}, nil},
+	}
+
+	// The ChainsDB is real, but uses only stubbed components
+	db := NewChainsDB(
+		map[types.ChainID]LogStorage{
+			chainID: logDB},
+		&stubHeadStorage{h})
+
+	// LastLogInBlock is expected to:
+	// 1. get a block iterator for block 10 (stubbed)
+	// 2. scan through the iterator and never find the target block
+	// return an error
+	_, err := db.LastLogInBlock(chainID, 10)
+	require.ErrorContains(t, err, "block 10 not found")
+}
+
+func TestChainsDB_LastLogInBlockError(t *testing.T) {
+	// using a chainID of 1 for simplicity
+	chainID := types.ChainIDFromUInt64(1)
+	// get default stubbed components
+	logDB, _, h := setupStubbedForUpdateHeads(chainID)
+	logDB.nextLogs = []nextLogResponse{
+		{10, 1, backendTypes.TruncatedHash{}, nil},
+		{10, 2, backendTypes.TruncatedHash{}, nil},
+		{10, 3, backendTypes.TruncatedHash{}, nil},
+		{0, 0, backendTypes.TruncatedHash{}, fmt.Errorf("some error")},
+		{11, 5, backendTypes.TruncatedHash{}, nil},
+	}
+
+	// The ChainsDB is real, but uses only stubbed components
+	db := NewChainsDB(
+		map[types.ChainID]LogStorage{
+			chainID: logDB},
+		&stubHeadStorage{h})
+
+	// LastLogInBlock is expected to:
+	// 1. get a block iterator for block 10 (stubbed)
+	// 2. scan through the iterator and encounter an error
+	// return an error
+	_, err := db.LastLogInBlock(chainID, 10)
+	require.ErrorContains(t, err, "some error")
+}
+
 func TestChainsDB_UpdateCrossHeads(t *testing.T) {
 	// using a chainID of 1 for simplicity
 	chainID := types.ChainIDFromUInt64(1)
@@ -172,7 +284,7 @@ func setupStubbedForUpdateHeads(chainID types.ChainID) (*stubLogDB, *stubChecker
 	// set up stubbed logDB
 	logDB := &stubLogDB{}
 	// the log DB will start the iterator at the checkpoint index
-	logDB.lastCheckpointBehind = &stubIterator{checkpoint}
+	logDB.lastCheckpointBehind = &stubIterator{checkpoint, 0, nil}
 	// rig the log DB to return an error after a certain number of calls to NextExecutingMessage
 	logDB.errAfter = errAfter
 	// set up stubbed executing messages that the ChainsDB can pass to the checker
@@ -237,6 +349,10 @@ func (s *stubChecker) Update(chain types.ChainID, index entrydb.EntryIdx) heads.
 	}
 }
 
+func (s *stubChecker) SafetyLevel() types.SafetyLevel {
+	return types.CrossSafe
+}
+
 type stubHeadStorage struct {
 	heads *heads.Heads
 }
@@ -252,13 +368,27 @@ func (s *stubHeadStorage) Current() *heads.Heads {
 	return s.heads.Copy()
 }
 
+type nextLogResponse struct {
+	blockNum uint64
+	logIdx   uint32
+	evtHash  backendTypes.TruncatedHash
+	err      error
+}
 type stubIterator struct {
 	index        entrydb.EntryIdx
+	nextLogIndex int
+	nextLogs     []nextLogResponse
 }
 
 func (s *stubIterator) NextLog() (uint64, uint32, backendTypes.TruncatedHash, error) {
-	panic("not implemented")
+	if s.nextLogIndex >= len(s.nextLogs) {
+		return 0, 0, backendTypes.TruncatedHash{}, io.EOF
+	}
+	r := s.nextLogs[s.nextLogIndex]
+	s.nextLogIndex++
+	return r.blockNum, r.logIdx, r.evtHash, r.err
 }
+
 func (s *stubIterator) Index() entrydb.EntryIdx {
 	return s.index
 }
@@ -271,6 +401,7 @@ type stubLogDB struct {
 	headBlockNum         uint64
 	emIndex              int
 	executingMessages    []*backendTypes.ExecutingMessage
+	nextLogs             []nextLogResponse
 	lastCheckpointBehind *stubIterator
 	errOverload          error
 	errAfter             int
@@ -282,6 +413,13 @@ func (s *stubLogDB) LastCheckpointBehind(entrydb.EntryIdx) (logs.Iterator, error
 	return s.lastCheckpointBehind, nil
 }
 
+func (s *stubLogDB) ClosestBlockIterator(blockNum uint64) (logs.Iterator, error) {
+	return &stubIterator{
+		index:    entrydb.EntryIdx(99),
+		nextLogs: s.nextLogs,
+	}, nil
+}
+
 func (s *stubLogDB) NextExecutingMessage(i logs.Iterator) (backendTypes.ExecutingMessage, error) {
 	// if error overload is set, return it to simulate a failure condition
 	if s.errOverload != nil && s.emIndex >= s.errAfter {

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

@@ -54,6 +54,10 @@ func (s *stubLogStore) Contains(blockNum uint64, logIdx uint32, loghash types.Tr
 	panic("not supported")
 }
 
+func (s *stubLogStore) ClosestBlockIterator(blockNum uint64) (logs.Iterator, error) {
+	panic("not supported")
+}
+
 func (s *stubLogStore) LastCheckpointBehind(entrydb.EntryIdx) (logs.Iterator, error) {
 	panic("not supported")
 }

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

@@ -202,6 +202,18 @@ func (db *DB) ClosestBlockInfo(blockNum uint64) (uint64, types.TruncatedHash, er
 	return checkpoint.blockNum, entry.hash, nil
 }
 
+// ClosestBlockIterator returns an iterator for the block closest to the specified blockNum.
+// The iterator will start at the search checkpoint for the block, or the first checkpoint before it.
+func (db *DB) ClosestBlockIterator(blockNum uint64) (Iterator, error) {
+	db.rwLock.RLock()
+	defer db.rwLock.RUnlock()
+	checkpointIdx, err := db.searchCheckpoint(blockNum, math.MaxUint32)
+	if err != nil {
+		return nil, fmt.Errorf("no checkpoint at or before block %v found: %w", blockNum, err)
+	}
+	return db.newIterator(checkpointIdx)
+}
+
 // Get returns the truncated hash of the log at the specified blockNum and logIdx,
 // or an error if the log is not found.
 func (db *DB) Get(blockNum uint64, logiIdx uint32) (types.TruncatedHash, error) {

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

@@ -25,6 +25,8 @@ type iterator struct {
 	entriesRead int64
 }
 
+// NextLog returns the next log in the iterator.
+// It scans forward until it finds an initiating event, returning the block number, log index, and event hash.
 func (i *iterator) NextLog() (blockNum uint64, logIdx uint32, evtHash types.TruncatedHash, outErr error) {
 	for i.nextEntryIdx <= i.db.lastEntryIdx() {
 		entryIdx := i.nextEntryIdx

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

@@ -21,25 +21,26 @@ type SafetyChecker interface {
 	Check(chain types.ChainID, blockNum uint64, logIdx uint32, logHash backendTypes.TruncatedHash) bool
 	Update(chain types.ChainID, index entrydb.EntryIdx) heads.OperationFn
 	Name() string
+	SafetyLevel() types.SafetyLevel
 }
 
 // unsafeChecker is a SafetyChecker that uses the unsafe head as the view into the database
 type unsafeChecker struct {
-	chainsDB ChainsDB
+	chainsDB *ChainsDB
 }
 
 // safeChecker is a SafetyChecker that uses the safe head as the view into the database
 type safeChecker struct {
-	chainsDB ChainsDB
+	chainsDB *ChainsDB
 }
 
 // finalizedChecker is a SafetyChecker that uses the finalized head as the view into the database
 type finalizedChecker struct {
-	chainsDB ChainsDB
+	chainsDB *ChainsDB
 }
 
 // NewSafetyChecker creates a new SafetyChecker of the given type
-func NewSafetyChecker(t string, chainsDB ChainsDB) SafetyChecker {
+func NewSafetyChecker(t types.SafetyLevel, chainsDB *ChainsDB) SafetyChecker {
 	switch t {
 	case Unsafe:
 		return &unsafeChecker{
@@ -105,10 +106,22 @@ func (c *finalizedChecker) CrossHeadForChain(chainID types.ChainID) entrydb.Entr
 	return heads.CrossFinalized
 }
 
+func (c *unsafeChecker) SafetyLevel() types.SafetyLevel {
+	return types.CrossUnsafe
+}
+
+func (c *safeChecker) SafetyLevel() types.SafetyLevel {
+	return types.CrossSafe
+}
+
+func (c *finalizedChecker) SafetyLevel() types.SafetyLevel {
+	return types.CrossFinalized
+}
+
 // check checks if the log entry is safe, provided a local head for the chain
 // it is used by the individual SafetyCheckers to determine if a log entry is safe
 func check(
-	chainsDB ChainsDB,
+	chainsDB *ChainsDB,
 	localHead entrydb.EntryIdx,
 	chain types.ChainID,
 	blockNum uint64,

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

@@ -29,7 +29,7 @@ func TestHeadsForChain(t *testing.T) {
 	tcases := []struct {
 		name          string
 		chainID       types.ChainID
-		checkerType   string
+		checkerType   types.SafetyLevel
 		expectedLocal entrydb.EntryIdx
 		expectedCross entrydb.EntryIdx
 	}{
@@ -65,7 +65,7 @@ func TestHeadsForChain(t *testing.T) {
 
 	for _, c := range tcases {
 		t.Run(c.name, func(t *testing.T) {
-			checker := NewSafetyChecker(c.checkerType, *chainsDB)
+			checker := NewSafetyChecker(c.checkerType, chainsDB)
 			localHead := checker.LocalHeadForChain(c.chainID)
 			crossHead := checker.CrossHeadForChain(c.chainID)
 			require.Equal(t, c.expectedLocal, localHead)
@@ -96,7 +96,7 @@ func TestCheck(t *testing.T) {
 
 	tcases := []struct {
 		name             string
-		checkerType      string
+		checkerType      types.SafetyLevel
 		chainID          types.ChainID
 		blockNum         uint64
 		logIdx           uint32
@@ -199,7 +199,7 @@ func TestCheck(t *testing.T) {
 		t.Run(c.name, func(t *testing.T) {
 			// rig the logStore to return the expected response
 			logDB.containsResponse = c.containsResponse
-			checker := NewSafetyChecker(c.checkerType, *chainsDB)
+			checker := NewSafetyChecker(c.checkerType, chainsDB)
 			r := checker.Check(c.chainID, c.blockNum, c.logIdx, c.loghash)
 			// confirm that the expected outcome is correct
 			require.Equal(t, c.expected, r)

op-supervisor/supervisor/types/types.go

@@ -84,10 +84,13 @@ func (lvl *SafetyLevel) UnmarshalText(text []byte) error {
 }
 
 const (
+	CrossFinalized SafetyLevel = "cross-finalized"
 	Finalized      SafetyLevel = "finalized"
+	CrossSafe      SafetyLevel = "cross-safe"
 	Safe           SafetyLevel = "safe"
 	CrossUnsafe    SafetyLevel = "cross-unsafe"
 	Unsafe         SafetyLevel = "unsafe"
+	Invalid        SafetyLevel = "invalid"
 )
 
 type ChainID uint256.Int