op-program: Consolidate cascading block replacements

op-e2e/actions/interop/proofs_test.go

@@ -443,6 +443,8 @@ func TestInteropFaultProofs_Cycle(gt *testing.T) {
 }
 
 func TestInteropFaultProofs_CascadeInvalidBlock(gt *testing.T) {
+	// TODO(#14307): Support cascading block invalidations
+	gt.Skip("TODO(#14307): Support cascading block invalidations")
 	t := helpers.NewDefaultTesting(gt)
 
 	system := dsl.NewInteropDSL(t)
@@ -458,31 +460,39 @@ func TestInteropFaultProofs_CascadeInvalidBlock(gt *testing.T) {
 		emitterContract.Deploy(alice),
 	))
 
-	// Initiating messages on chain A
-	system.AddL2Block(actors.ChainA, dsl.WithL2BlockTransactions(
-		emitterContract.EmitMessage(alice, "chainA message"),
-	))
-	chainAInitTx := emitterContract.LastEmittedMessage()
-	system.AddL2Block(actors.ChainB)
-	system.SubmitBatchData()
+	assertHeads(t, actors.ChainA, 1, 0, 1, 0)
+	assertHeads(t, actors.ChainB, 1, 0, 1, 0)
 
-	// Create a message with a conflicting payload on chain B, that also emits an initiating message
-	system.AddL2Block(actors.ChainB, dsl.WithL2BlockTransactions(
-		system.InboxContract.Execute(alice, chainAInitTx, dsl.WithPayload([]byte("this message was never emitted"))),
-		emitterContract.EmitMessage(alice, "chainB message"),
-	), dsl.WithL1BlockCrossUnsafe())
-	chainBExecTx := system.InboxContract.LastTransaction()
-	chainBExecTx.CheckIncluded()
-	chainBInitTx := emitterContract.LastEmittedMessage()
-
-	// Create a message with a valid message on chain A, pointing to the initiating message on B from the same block
-	// as an invalid message.
-	system.AddL2Block(actors.ChainA,
-		dsl.WithL2BlockTransactions(system.InboxContract.Execute(alice, chainBInitTx)),
-		dsl.WithL1BlockCrossUnsafe(),
+	// Create initiating and executing messages within the same block
+	var (
+		chainAExecTx *dsl.GeneratedTransaction
+		chainBExecTx *dsl.GeneratedTransaction
+		chainBInitTx *dsl.GeneratedTransaction
 	)
-	chainAExecTx := system.InboxContract.LastTransaction()
-	chainAExecTx.CheckIncluded()
+	{
+		actors.ChainA.Sequencer.ActL2StartBlock(t)
+		actors.ChainB.Sequencer.ActL2StartBlock(t)
+
+		chainAInitTx := emitterContract.EmitMessage(alice, "chainA message")(actors.ChainA)
+		chainAInitTx.Include()
+
+		// Create messages with a conflicting payload on chain B, while also emitting an initiating message
+		chainBExecTx := system.InboxContract.Execute(alice, chainAInitTx,
+			dsl.WithPayload([]byte("this message was never emitted")))(actors.ChainB)
+		chainBExecTx.Include()
+		chainBInitTx = emitterContract.EmitMessage(alice, "chainB message")(actors.ChainB)
+		chainBInitTx.Include()
+
+		// Create a message with a valid message on chain A, pointing to the initiating message on B from the same block
+		// as an invalid message.
+		chainAExecTx = system.InboxContract.Execute(alice, chainBInitTx)(actors.ChainA)
+		chainAExecTx.Include()
+
+		actors.ChainA.Sequencer.ActL2EndBlock(t)
+		actors.ChainB.Sequencer.ActL2EndBlock(t)
+	}
+	assertHeads(t, actors.ChainA, 2, 0, 1, 0)
+	assertHeads(t, actors.ChainB, 2, 0, 1, 0)
 
 	system.SubmitBatchData(func(opts *dsl.SubmitBatchDataOpts) {
 		opts.SkipCrossSafeUpdate = true
@@ -513,19 +523,13 @@ func TestInteropFaultProofs_CascadeInvalidBlock(gt *testing.T) {
 			disputedClaim:      optimisticEnd.Marshal(),
 			disputedTraceIndex: consolidateStep,
 			expectValid:        false,
-			// TODO(#14306): Support cascading re-orgs in op-program
-			skipProgram:    true,
-			skipChallenger: true,
 		},
 		{
 			name:               "Consolidate-ReplaceInvalidBlocks",
 			agreedClaim:        preConsolidation,
 			disputedClaim:      crossSafeEnd.Marshal(),
 			disputedTraceIndex: consolidateStep,
 			expectValid:        true,
-			// TODO(#14306): Support cascading re-orgs in op-program
-			skipProgram:    true,
-			skipChallenger: true,
 		},
 	}
 	runFppAndChallengerTests(gt, system, tests)

op-program/client/interop/consolidate.go

@@ -20,6 +20,8 @@ import (
 	"github.com/ethereum/go-ethereum/log"
 )
 
+var ErrInvalidBlockReplacement = errors.New("invalid block replacement error")
+
 // ReceiptsToExecutingMessages returns the executing messages in the receipts indexed by their position in the log.
 func ReceiptsToExecutingMessages(depset depset.ChainIndexFromID, receipts ethtypes.Receipts) (map[uint32]*supervisortypes.ExecutingMessage, uint32, error) {
 	execMsgs := make(map[uint32]*supervisortypes.ExecutingMessage)
@@ -39,17 +41,19 @@ func ReceiptsToExecutingMessages(depset depset.ChainIndexFromID, receipts ethtyp
 	return execMsgs, curr, nil
 }
 
-func fetchAgreedBlockHashes(oracle l2.Oracle, superRoot *eth.SuperV1) ([]common.Hash, error) {
-	agreedBlockHashes := make([]common.Hash, len(superRoot.Chains))
-	for i, chain := range superRoot.Chains {
-		output := oracle.OutputByRoot(common.Hash(chain.Output), chain.ChainID)
-		outputV0, ok := output.(*eth.OutputV0)
-		if !ok {
-			return nil, fmt.Errorf("unsupported L2 output version: %d", output.Version())
-		}
-		agreedBlockHashes[i] = common.Hash(outputV0.BlockHash)
-	}
-	return agreedBlockHashes, nil
+type consolidateState struct {
+	*types.TransitionState
+	replacedChains map[eth.ChainID]bool
+}
+
+func (s *consolidateState) isReplaced(chainID eth.ChainID) bool {
+	return s.replacedChains[chainID]
+}
+
+func (s *consolidateState) setReplaced(transitionStateIndex int, chainID eth.ChainID, outputRoot eth.Bytes32, replacementBlockHash common.Hash) {
+	s.PendingProgress[transitionStateIndex].OutputRoot = outputRoot
+	s.PendingProgress[transitionStateIndex].BlockHash = replacementBlockHash
+	s.replacedChains[chainID] = true
 }
 
 func RunConsolidation(
@@ -61,87 +65,137 @@ func RunConsolidation(
 	superRoot *eth.SuperV1,
 	tasks taskExecutor,
 ) (eth.Bytes32, error) {
+	consolidateState := consolidateState{
+		TransitionState: &types.TransitionState{
+			PendingProgress: make([]types.OptimisticBlock, len(transitionState.PendingProgress)),
+			SuperRoot:       transitionState.SuperRoot,
+			Step:            transitionState.Step,
+		},
+		replacedChains: make(map[eth.ChainID]bool),
+	}
+	// We will be updating the transition state as blocks are replaced, so make a copy
+	copy(consolidateState.PendingProgress, transitionState.PendingProgress)
+	// Use a reference to the transition state so the consolidate oracle has a recent view.
+	// The TransitionStateByRoot method isn't expected to be used during consolidation,
+	// but we pass the state for safety in case this changes in the future.
+	consolidateOracle := NewConsolidateOracle(l2PreimageOracle, consolidateState.TransitionState)
+
+	// Keep consolidating until there are no more invalid blocks to replace
+loop:
+	for {
+		err := singleRoundConsolidation(logger, bootInfo, l1PreimageOracle, consolidateOracle, &consolidateState, superRoot, tasks)
+		switch {
+		case err == nil:
+			break loop
+		case errors.Is(err, ErrInvalidBlockReplacement):
+			continue
+		default:
+			return eth.Bytes32{}, err
+		}
+	}
+
+	var consolidatedChains []eth.ChainIDAndOutput
+	for i, chain := range superRoot.Chains {
+		consolidatedChains = append(consolidatedChains, eth.ChainIDAndOutput{
+			ChainID: chain.ChainID,
+			Output:  consolidateState.PendingProgress[i].OutputRoot,
+		})
+	}
+	consolidatedSuper := &eth.SuperV1{
+		Timestamp: superRoot.Timestamp + 1,
+		Chains:    consolidatedChains,
+	}
+	return eth.SuperRoot(consolidatedSuper), nil
+}
+
+func singleRoundConsolidation(
+	logger log.Logger,
+	bootInfo *boot.BootInfoInterop,
+	l1PreimageOracle l1.Oracle,
+	l2PreimageOracle *ConsolidateOracle,
+	consolidateState *consolidateState,
+	superRoot *eth.SuperV1,
+	tasks taskExecutor,
+) error {
 	// The depset is the same for all chains. So it suffices to use any chain ID
 	depset, err := bootInfo.Configs.DependencySet(superRoot.Chains[0].ChainID)
 	if err != nil {
-		return eth.Bytes32{}, fmt.Errorf("failed to get dependency set: %w", err)
-	}
-	deps, err := newConsolidateCheckDeps(depset, bootInfo, transitionState, superRoot.Chains, l2PreimageOracle)
-	if err != nil {
-		return eth.Bytes32{}, fmt.Errorf("failed to create consolidate check deps: %w", err)
+		return fmt.Errorf("failed to get dependency set: %w", err)
 	}
-	agreedBlockHashes, err := fetchAgreedBlockHashes(l2PreimageOracle, superRoot)
+	deps, err := newConsolidateCheckDeps(depset, bootInfo, consolidateState.TransitionState, superRoot.Chains, l2PreimageOracle)
 	if err != nil {
-		return eth.Bytes32{}, err
+		return fmt.Errorf("failed to create consolidate check deps: %w", err)
 	}
-	// TODO(#14306): Handle cascading reorgs
-	// invalidChains tracks blocks that need to be replaced with a deposits-only block.
-	// The replacement is done after a first pass on all chains to avoid "contaminating" the caonical block
-	// oracle in a way that alters the result of hazard checks after a reorg.
 	invalidChains := make(map[eth.ChainID]*ethtypes.Block)
 
 	for i, chain := range superRoot.Chains {
-		progress := transitionState.PendingProgress[i]
+		// Do not check chains that have been replaced with a deposits-only block.
+		// They are already cross-safe because deposits-only blocks cannot contain executing messages.
+		if consolidateState.isReplaced(chain.ChainID) {
+			continue
+		}
 
+		agreedOutput := l2PreimageOracle.OutputByRoot(common.Hash(chain.Output), chain.ChainID)
+		agreedOutputV0, ok := agreedOutput.(*eth.OutputV0)
+		if !ok {
+			return fmt.Errorf("unsupported L2 output version: %d", agreedOutput.Version())
+		}
+		agreedBlockHash := common.Hash(agreedOutputV0.BlockHash)
+
+		progress := consolidateState.PendingProgress[i]
 		// It's possible that the optimistic block is not canonical.
 		// So we use the blockDataByHash hint to trigger a block rebuild to ensure that the block data, including receipts, are available.
-		_ = l2PreimageOracle.BlockDataByHash(agreedBlockHashes[i], progress.BlockHash, chain.ChainID)
+		_ = l2PreimageOracle.BlockDataByHash(agreedBlockHash, progress.BlockHash, chain.ChainID)
 
-		optimisticBlock, receipts := l2PreimageOracle.ReceiptsByBlockHash(progress.BlockHash, chain.ChainID)
-		execMsgs, _, err := ReceiptsToExecutingMessages(deps.DependencySet(), receipts)
-		switch {
-		case errors.Is(err, supervisortypes.ErrUnknownChain):
-			invalidChains[chain.ChainID] = optimisticBlock
-			continue
-		case err != nil:
-			return eth.Bytes32{}, err
-		}
+		optimisticBlock, _ := l2PreimageOracle.ReceiptsByBlockHash(progress.BlockHash, chain.ChainID)
 
 		candidate := supervisortypes.BlockSeal{
 			Hash:      progress.BlockHash,
 			Number:    optimisticBlock.NumberU64(),
 			Timestamp: optimisticBlock.Time(),
 		}
-		if err := checkHazards(logger, deps, candidate, chain.ChainID, execMsgs); err != nil {
+		if err := checkHazards(logger, deps, candidate, chain.ChainID); err != nil {
 			if !isInvalidMessageError(err) {
-				return eth.Bytes32{}, err
+				return err
 			}
 			invalidChains[chain.ChainID] = optimisticBlock
 		}
 	}
 
-	var consolidatedChains []eth.ChainIDAndOutput
+	if len(invalidChains) == 0 {
+		return nil
+	}
+
 	for i, chain := range superRoot.Chains {
 		if optimisticBlock, ok := invalidChains[chain.ChainID]; ok {
 			chainAgreedPrestate := superRoot.Chains[i]
-			_, outputRoot, err := buildDepositOnlyBlock(
+			replacementBlockHash, outputRoot, err := buildDepositOnlyBlock(
 				logger,
 				bootInfo,
 				l1PreimageOracle,
 				l2PreimageOracle,
 				chainAgreedPrestate,
 				tasks,
 				optimisticBlock,
+				// Update the preimage oracle database with the replaced block data
+				l2PreimageOracle.KeyValueStore(),
 			)
 			if err != nil {
-				return eth.Bytes32{}, err
+				return err
 			}
-			consolidatedChains = append(consolidatedChains, eth.ChainIDAndOutput{
-				ChainID: chain.ChainID,
-				Output:  outputRoot,
-			})
-		} else {
-			consolidatedChains = append(consolidatedChains, eth.ChainIDAndOutput{
-				ChainID: chain.ChainID,
-				Output:  transitionState.PendingProgress[i].OutputRoot,
-			})
+			logger.Info(
+				"Replaced block",
+				"chain", chain.ChainID,
+				"replacedBlock", eth.ToBlockID(optimisticBlock),
+				"replacementBlockHash", replacementBlockHash,
+				"outputRoot", outputRoot,
+				"replacedOutputRoot", superRoot.Chains[i].Output,
+			)
+			superRoot.Chains[i].Output = outputRoot
+			consolidateState.setReplaced(i, chain.ChainID, outputRoot, replacementBlockHash)
 		}
 	}
-	consolidatedSuper := &eth.SuperV1{
-		Timestamp: superRoot.Timestamp + 1,
-		Chains:    consolidatedChains,
-	}
-	return eth.SuperRoot(consolidatedSuper), nil
+	return ErrInvalidBlockReplacement
 }
 
 func isInvalidMessageError(err error) bool {
@@ -158,7 +212,7 @@ type ConsolidateCheckDeps interface {
 	cross.UnsafeStartDeps
 }
 
-func checkHazards(logger log.Logger, deps ConsolidateCheckDeps, candidate supervisortypes.BlockSeal, chainID eth.ChainID, execMsgs map[uint32]*supervisortypes.ExecutingMessage) error {
+func checkHazards(logger log.Logger, deps ConsolidateCheckDeps, candidate supervisortypes.BlockSeal, chainID eth.ChainID) error {
 	hazards, err := cross.CrossUnsafeHazards(deps, logger, chainID, candidate)
 	if err != nil {
 		return err
@@ -309,6 +363,7 @@ func buildDepositOnlyBlock(
 	chainAgreedPrestate eth.ChainIDAndOutput,
 	tasks taskExecutor,
 	optimisticBlock *ethtypes.Block,
+	db l2.KeyValueStore,
 ) (common.Hash, eth.Bytes32, error) {
 	rollupCfg, err := bootInfo.Configs.RollupConfig(chainAgreedPrestate.ChainID)
 	if err != nil {
@@ -327,6 +382,7 @@ func buildDepositOnlyBlock(
 		l1PreimageOracle,
 		l2PreimageOracle,
 		optimisticBlock,
+		db,
 	)
 	if err != nil {
 		return common.Hash{}, eth.Bytes32{}, err

op-program/client/interop/interop.go

@@ -53,6 +53,7 @@ type taskExecutor interface {
 		l1Oracle l1.Oracle,
 		l2Oracle l2.Oracle,
 		optimisticBlock *ethtypes.Block,
+		db l2.KeyValueStore,
 	) (blockHash common.Hash, outputRoot eth.Bytes32, err error)
 }
 
@@ -214,6 +215,7 @@ func (t *interopTaskExecutor) BuildDepositOnlyBlock(
 	l1Oracle l1.Oracle,
 	l2Oracle l2.Oracle,
 	optimisticBlock *ethtypes.Block,
+	db l2.KeyValueStore,
 ) (common.Hash, eth.Bytes32, error) {
 	return tasks.BuildDepositOnlyBlock(
 		logger,
@@ -224,5 +226,6 @@ func (t *interopTaskExecutor) BuildDepositOnlyBlock(
 		agreedL2OutputRoot,
 		l1Oracle,
 		l2Oracle,
+		db,
 	)
 }