op-node: reorgs should be typed to force pipeline resets (#3185)

op-node/rollup/derive/engine_queue.go

@@ -279,7 +279,7 @@ func (eq *EngineQueue) forceNextSafeAttributes(ctx context.Context) error {
 			eq.safeAttributes[0].Transactions = deposits
 			return nil
 		}
-		return fmt.Errorf("critical: failed to process block with only deposit transactions: %v", payloadErr)
+		return NewCriticalError(payloadErr, "failed to process block with only deposit transactions")
 	}
 	ref, err := PayloadToBlockRef(payload, &eq.cfg.Genesis)
 	if err != nil {

op-node/rollup/derive/l1_traversal.go

@@ -56,7 +56,7 @@ func (l1t *L1Traversal) Step(ctx context.Context, outer Progress) error {
 		return nil // nil, don't make the pipeline restart if the RPC fails
 	}
 	if l1t.progress.Origin.Hash != nextL1Origin.ParentHash {
-		return fmt.Errorf("detected L1 reorg from %s to %s: %w", l1t.progress.Origin, nextL1Origin, ReorgErr)
+		return NewResetError(ReorgErr, fmt.Sprintf("detected L1 reorg from %s to %s", l1t.progress.Origin, nextL1Origin))
 	}
 	l1t.progress.Origin = nextL1Origin
 	l1t.progress.Closed = false

op-node/rollup/derive/progress.go

@@ -24,12 +24,12 @@ func (pr *Progress) Update(outer Progress) (changed bool, err error) {
 	if pr.Closed {
 		if outer.Closed {
 			if pr.Origin.ID() != outer.Origin.ID() {
-				return true, fmt.Errorf("outer stage changed origin from %s to %s without opening it", pr.Origin, outer.Origin)
+				return true, NewResetError(ReorgErr, fmt.Sprintf("outer stage changed origin from %s to %s without opening it", pr.Origin, outer.Origin))
 			}
 			return false, nil
 		} else {
 			if pr.Origin.Hash != outer.Origin.ParentHash {
-				return true, fmt.Errorf("detected internal pipeline reorg of L1 origin data from %s to %s: %w", pr.Origin, outer.Origin, ReorgErr)
+				return true, NewResetError(ReorgErr, fmt.Sprintf("detected internal pipeline reorg of L1 origin data from %s to %s", pr.Origin, outer.Origin))
 			}
 			pr.Origin = outer.Origin
 			pr.Closed = false
@@ -37,7 +37,7 @@ func (pr *Progress) Update(outer Progress) (changed bool, err error) {
 		}
 	} else {
 		if pr.Origin.ID() != outer.Origin.ID() {
-			return true, fmt.Errorf("outer stage changed origin from %s to %s before closing it", pr.Origin, outer.Origin)
+			return true, NewResetError(ReorgErr, fmt.Sprintf("outer stage changed origin from %s to %s before closing it", pr.Origin, outer.Origin))
 		}
 		if outer.Closed {
 			pr.Closed = true