supervisor: Record new L1 on ProvideL1 and Notify L2 Finality (#13678)

* Record new L1 on ProvideL1 and Notify L2 Finality

* lint ; expand unit tests

* Add comment for safe use of RecordNewL1

op-supervisor/supervisor/backend/backend.go

@@ -517,6 +517,10 @@ func (su *SupervisorBackend) UpdateLocalSafe(ctx context.Context, chainID types.
 	return nil
 }
 
+func (su *SupervisorBackend) RecordNewL1(ctx context.Context, chain types.ChainID, ref eth.BlockRef) error {
+	return su.chainDBs.RecordNewL1(chain, ref)
+}
+
 // Access to synchronous processing for tests
 // ----------------------------
 

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

@@ -114,60 +114,65 @@ func (db *ChainsDB) UpdateFinalizedL1(finalized eth.BlockRef) error {
 	db.logger.Info("Updated finalized L1", "finalizedL1", finalized)
 	db.finalizedL1.Unlock()
 
-	// whenever the L1 Finalized changes, the L2 Finalized may change, notify subscribers
-	db.NotifyL2Finalized()
+	// whenver the L1 Finalized changes, the L2 Finalized may change, notify subscribers
+	for _, chain := range db.depSet.Chains() {
+		db.NotifyL2Finalized(chain)
+	}
 
 	return nil
 }
 
-// NotifyL2Finalized notifies all L2 finality subscribers of the latest L2 finalized block, per chain.
-func (db *ChainsDB) NotifyL2Finalized() {
-	for _, chain := range db.depSet.Chains() {
-		f, err := db.Finalized(chain)
-		if err != nil {
-			db.logger.Error("Failed to get finalized L2 block", "chain", chain, "err", err)
-			continue
-		}
-		sub, ok := db.l2FinalityFeeds.Get(chain)
-		if ok {
-			sub.Send(f)
-		}
+// NotifyL2Finalized notifies all L2 finality subscribers of the latest L2 finalized block for the given chain.
+func (db *ChainsDB) NotifyL2Finalized(chain types.ChainID) {
+	f, err := db.Finalized(chain)
+	if err != nil {
+		db.logger.Error("Failed to get finalized L1 block", "chain", chain, "err", err)
+		return
+	}
+	sub, ok := db.l2FinalityFeeds.Get(chain)
+	if ok {
+		sub.Send(f)
 	}
 }
 
-// RecordNewL1 records a new L1 block in the database.
-// it uses the latest derived L2 block as the derived block for the new L1 block.
-func (db *ChainsDB) RecordNewL1(ref eth.BlockRef) error {
-	for _, chain := range db.depSet.Chains() {
-		// get local derivation database
-		ldb, ok := db.localDBs.Get(chain)
-		if !ok {
-			return fmt.Errorf("cannot RecordNewL1 to chain %s: %w", chain, types.ErrUnknownChain)
-		}
-		// get the latest derived and derivedFrom blocks
-		derivedFrom, derived, err := ldb.Latest()
-		if err != nil {
-			return fmt.Errorf("failed to get latest derivedFrom for chain %s: %w", chain, err)
-		}
-		// make a ref from the latest derived block
-		derivedParent, err := ldb.PreviousDerived(derived.ID())
-		if errors.Is(err, types.ErrFuture) {
-			db.logger.Warn("Empty DB, Recording first L1 block", "chain", chain, "err", err)
-		} else if err != nil {
-			db.logger.Warn("Failed to get latest derivedfrom to insert new L1 block", "chain", chain, "err", err)
-			return err
-		}
-		derivedRef := derived.MustWithParent(derivedParent.ID())
-		// don't push the new L1 block if it's not newer than the latest derived block
-		if derivedFrom.Number >= ref.Number {
-			db.logger.Warn("L1 block has already been processed for this height", "chain", chain, "block", ref, "latest", derivedFrom)
-			continue
-		}
-		// the database is extended with the new L1 and the existing L2
-		if err = db.UpdateLocalSafe(chain, ref, derivedRef); err != nil {
-			db.logger.Error("Failed to update local safe", "chain", chain, "block", ref, "derived", derived, "err", err)
-			return err
-		}
+// RecordNewL1 records a new L1 block in the database for a given chain.
+// It uses the latest derived L2 block as the derived block for the new L1 block.
+// It also triggers L2 Finality Notifications, as a new L1 may change L2 finality.
+// NOTE: callers to this function are responsible for ensuring that advancing the L1 block is correct
+// (ie that no further L2 blocks need to be recorded) because if the L1 block is recorded with a gap in derived blocks,
+// the database is considered corrupted and the supervisor will not be able to proceed without pruning the database.
+// The database cannot protect against this because it is does not know how many L2 blocks to expect for a given L1 block.
+func (db *ChainsDB) RecordNewL1(chain types.ChainID, ref eth.BlockRef) error {
+	// get local derivation database
+	ldb, ok := db.localDBs.Get(chain)
+	if !ok {
+		return fmt.Errorf("cannot RecordNewL1 to chain %s: %w", chain, types.ErrUnknownChain)
+	}
+	// get the latest derived and derivedFrom blocks
+	derivedFrom, derived, err := ldb.Latest()
+	if err != nil {
+		return fmt.Errorf("failed to get latest derivedFrom for chain %s: %w", chain, err)
+	}
+	// make a ref from the latest derived block
+	derivedParent, err := ldb.PreviousDerived(derived.ID())
+	if errors.Is(err, types.ErrFuture) {
+		db.logger.Warn("Empty DB, Recording first L1 block", "chain", chain, "err", err)
+	} else if err != nil {
+		db.logger.Warn("Failed to get latest derivedfrom to insert new L1 block", "chain", chain, "err", err)
+		return err
+	}
+	derivedRef := derived.MustWithParent(derivedParent.ID())
+	// don't push the new L1 block if it's not newer than the latest derived block
+	if derivedFrom.Number >= ref.Number {
+		db.logger.Warn("L1 block has already been processed for this height", "chain", chain, "block", ref, "latest", derivedFrom)
+		return nil
+	}
+	// the database is extended with the new L1 and the existing L2
+	if err = db.UpdateLocalSafe(chain, ref, derivedRef); err != nil {
+		db.logger.Error("Failed to update local safe", "chain", chain, "block", ref, "derived", derived, "err", err)
+		return err
 	}
+	// now tht the db has the new L1, we can attempt to to notify the L2 finality subscribers
+	db.NotifyL2Finalized(chain)
 	return nil
 }

op-supervisor/supervisor/backend/syncnode/controller_test.go

@@ -137,6 +137,7 @@ var _ SyncControl = (*mockSyncControl)(nil)
 type mockBackend struct {
 	updateLocalUnsafeFn func(ctx context.Context, chainID types.ChainID, head eth.BlockRef) error
 	updateLocalSafeFn   func(ctx context.Context, chainID types.ChainID, derivedFrom eth.BlockRef, lastDerived eth.BlockRef) error
+	recordL1Fn          func(ctx context.Context, chain types.ChainID, ref eth.BlockRef) error
 }
 
 func (m *mockBackend) LocalSafe(ctx context.Context, chainID types.ChainID) (pair types.DerivedIDPair, err error) {
@@ -177,6 +178,13 @@ func (m *mockBackend) L1BlockRefByNumber(ctx context.Context, number uint64) (et
 	return eth.L1BlockRef{}, nil
 }
 
+func (m *mockBackend) RecordNewL1(ctx context.Context, chain types.ChainID, ref eth.BlockRef) error {
+	if m.recordL1Fn != nil {
+		return m.recordL1Fn(ctx, chain, ref)
+	}
+	return nil
+}
+
 var _ backend = (*mockBackend)(nil)
 
 func sampleDepSet(t *testing.T) depset.DependencySet {

op-supervisor/supervisor/backend/syncnode/node.go

@@ -38,6 +38,7 @@ type backend interface {
 	SafeDerivedAt(ctx context.Context, chainID types.ChainID, derivedFrom eth.BlockID) (derived eth.BlockID, err error)
 	Finalized(ctx context.Context, chainID types.ChainID) (eth.BlockID, error)
 	L1BlockRefByNumber(ctx context.Context, number uint64) (eth.L1BlockRef, error)
+	RecordNewL1(ctx context.Context, chainID types.ChainID, l1 eth.BlockRef) error
 }
 
 const (
@@ -357,6 +358,15 @@ func (m *ManagedNode) onExhaustL1Event(completed types.DerivedBlockRefPair) {
 		// but does not fit on the derivation state.
 		return
 	}
+	// now that the node has the next L1 block, we can add it to the database
+	// this ensures that only the L1 *or* the L2 ever increments in the derivation database,
+	// as RecordNewL1 will insert the new L1 block with the latest L2 block
+	ctx, cancel := context.WithTimeout(m.ctx, internalTimeout)
+	defer cancel()
+	err = m.backend.RecordNewL1(ctx, m.chainID, nextL1)
+	if err != nil {
+		m.log.Warn("Failed to record new L1 block", "l1Block", nextL1, "err", err)
+	}
 }
 
 func (m *ManagedNode) AwaitSentCrossUnsafeUpdate(ctx context.Context, minNum uint64) error {

op-supervisor/supervisor/backend/syncnode/node_test.go

@@ -28,6 +28,8 @@ func TestEventResponse(t *testing.T) {
 	nodeUnsafe := 0
 	nodeDerivation := 0
 	nodeExhausted := 0
+	// recordL1 is called along with nodeExhausted
+	recordL1 := 0
 
 	// the node will call UpdateCrossUnsafe when a cross-unsafe event is received from the database
 	syncCtrl.updateCrossUnsafeFn = func(ctx context.Context, id eth.BlockID) error {
@@ -61,6 +63,10 @@ func TestEventResponse(t *testing.T) {
 		nodeExhausted++
 		return nil
 	}
+	backend.recordL1Fn = func(ctx context.Context, chainID types.ChainID, ref eth.L1BlockRef) error {
+		recordL1++
+		return nil
+	}
 	// TODO(#13595): rework node-reset, and include testing for it here
 
 	node.Start()
@@ -85,4 +91,7 @@ func TestEventResponse(t *testing.T) {
 			nodeDerivation >= 1 &&
 			nodeExhausted >= 1
 	}, 4*time.Second, 250*time.Millisecond)
+
+	// recordL1 is called every time nodeExhausted is called
+	require.Equal(t, nodeExhausted, recordL1)
 }