feat: auto recover from pbss geth unclean shutdown

op-node/rollup/derive/engine_controller.go

@@ -271,6 +271,9 @@ func (e *EngineController) resetBuildingState() {
 // It returns true if the status is acceptable.
 func (e *EngineController) checkNewPayloadStatus(status eth.ExecutePayloadStatus) bool {
 	if e.syncMode == sync.ELSync {
+		if status == eth.ExecutionUnconsistent {
+			return true
+		}
 		if status == eth.ExecutionValid && e.syncStatus == syncStatusStartedEL {
 			e.syncStatus = syncStatusFinishedELButNotFinalized
 		}
@@ -348,6 +351,41 @@ func (e *EngineController) InsertUnsafePayload(ctx context.Context, envelope *et
 	if err != nil {
 		return NewTemporaryError(fmt.Errorf("failed to update insert payload: %w", err))
 	}
+
+	//增加unconsistent逻辑
+	//1. fcu，更新unsafe状态，currentHeads方法
+	//2. reset,推进safe块高(或者直接set好)
+	//3. tryupdateengine (更新geth safe块高)
+	//4.
+	if status.Status == eth.ExecutionUnconsistent {
+		currentL2Info, err := e.getCurrentL2Info(ctx)
+		if err == nil {
+			log.Info("zxl inconsistent state", "unsafe", currentL2Info.Unsafe.Number, "safe", currentL2Info.Safe.Number, "final", currentL2Info.Finalized.Number)
+			e.SetUnsafeHead(currentL2Info.Unsafe)
+			if currentL2Info.Safe.Number > currentL2Info.Unsafe.Number {
+				log.Info("currentL2Info.Safe.Number > currentL2Info.Unsafe.Number", "set safe after", currentL2Info.Unsafe.Number, "set safe before", e.safeHead.Number)
+				e.SetSafeHead(currentL2Info.Unsafe)
+			}
+			if currentL2Info.Finalized.Number > currentL2Info.Unsafe.Number {
+				log.Info("currentL2Info.Finalized.Number > currentL2Info.Unsafe.Number", "set Finalized after", currentL2Info.Unsafe.Number, "set Finalized before", e.safeHead.Number)
+				e.SetFinalizedHead(currentL2Info.Unsafe)
+			}
+		}
+
+		fcs := eth.ForkchoiceState{
+			HeadBlockHash:      e.unsafeHead.Hash,
+			SafeBlockHash:      e.safeHead.Hash,
+			FinalizedBlockHash: e.finalizedHead.Hash,
+		}
+
+		fcRes, err := e.engine.ForkchoiceUpdate(ctx, &fcs, nil)
+		if fcRes.PayloadStatus.Status == eth.ExecutionValid {
+			log.Info("zxl delete response success")
+			e.needFCUCall = false
+			return nil
+		}
+	}
+
 	if !e.checkNewPayloadStatus(status.Status) {
 		payload := envelope.ExecutionPayload
 		return NewTemporaryError(fmt.Errorf("cannot process unsafe payload: new - %v; parent: %v; err: %w",
@@ -360,6 +398,22 @@ func (e *EngineController) InsertUnsafePayload(ctx context.Context, envelope *et
 		SafeBlockHash:      e.safeHead.Hash,
 		FinalizedBlockHash: e.finalizedHead.Hash,
 	}
+
+	//此处更新
+	if status.Status == eth.ExecutionUnconsistent {
+		log.Info("ZXL in unconsistent here")
+		unsafenow, _ := e.engine.L2BlockRefByLabel(ctx, eth.Unsafe)
+		//重置unsafe
+		e.SetUnsafeHead(unsafenow)
+		//强制reset
+		e.SetSafeHead(unsafenow)
+		e.SetFinalizedHead(unsafenow)
+
+		fc.HeadBlockHash = unsafenow.Hash
+		fc.SafeBlockHash = unsafenow.Hash
+		fc.FinalizedBlockHash = unsafenow.Hash
+	}
+
 	if e.syncStatus == syncStatusFinishedELButNotFinalized {
 		fc.SafeBlockHash = envelope.ExecutionPayload.BlockHash
 		fc.FinalizedBlockHash = envelope.ExecutionPayload.BlockHash
@@ -468,3 +522,35 @@ func (e *EngineController) TryBackupUnsafeReorg(ctx context.Context) (bool, erro
 func (e *EngineController) ResetBuildingState() {
 	e.resetBuildingState()
 }
+
+// getCurrentL2Info returns the current finalized, safe and unsafe heads of the execution engine.
+func (e *EngineController) getCurrentL2Info(ctx context.Context) (*sync.FindHeadsResult, error) {
+	finalized, err := e.engine.L2BlockRefByLabel(ctx, eth.Finalized)
+	if err != nil {
+		log.Error("2 err get finalized", "err", err)
+	}
+	if errors.Is(err, ethereum.NotFound) {
+		// default to genesis if we have not finalized anything before.
+		log.Error("not found")
+	}
+	if err != nil {
+		return nil, fmt.Errorf("failed to find the finalized L2 block: %w", err)
+	}
+
+	safe, err := e.engine.L2BlockRefByLabel(ctx, eth.Safe)
+	if errors.Is(err, ethereum.NotFound) {
+		safe = finalized
+	} else if err != nil {
+		return nil, fmt.Errorf("2 failed to find the safe L2 block: %w", err)
+	}
+
+	unsafe, err := e.engine.L2BlockRefByLabel(ctx, eth.Unsafe)
+	if err != nil {
+		return nil, fmt.Errorf("2 failed to find the L2 head block: %w", err)
+	}
+	return &sync.FindHeadsResult{
+		Unsafe:    unsafe,
+		Safe:      safe,
+		Finalized: finalized,
+	}, nil
+}

op-node/rollup/derive/engine_queue.go

@@ -695,6 +695,11 @@ func (eq *EngineQueue) Reset(ctx context.Context, _ eth.L1BlockRef, _ eth.System
 		return NewTemporaryError(fmt.Errorf("failed to find the L2 Heads to start from: %w", err))
 	}
 	finalized, safe, unsafe := result.Finalized, result.Safe, result.Unsafe
+
+	if finalized.Number > safe.Number {
+		finalized = safe
+	}
+
 	l1Origin, err := eq.l1Fetcher.L1BlockRefByHash(ctx, safe.L1Origin.Hash)
 	if err != nil {
 		return NewTemporaryError(fmt.Errorf("failed to fetch the new L1 progress: origin: %v; err: %w", safe.L1Origin, err))

op-service/eth/types.go

@@ -338,6 +338,8 @@ const (
 	ExecutionInvalidBlockHash ExecutePayloadStatus = "INVALID_BLOCK_HASH"
 	// proof-of-stake transition only, not used in rollup
 	ExecutionInvalidTerminalBlock ExecutePayloadStatus = "INVALID_TERMINAL_BLOCK"
+	// given payload is invalid
+	ExecutionUnconsistent ExecutePayloadStatus = "UNCONSISTENT"
 )
 
 type PayloadStatusV1 struct {