pipeline replay mode

rollup/pipeline/pipeline.go

@@ -53,13 +53,14 @@ var (
 )
 
 type Pipeline struct {
-	chain     *core.BlockChain
-	vmConfig  vm.Config
-	parent    *types.Block
-	start     time.Time
-	wg        sync.WaitGroup
-	ctx       context.Context
-	cancelCtx context.CancelFunc
+	chain      *core.BlockChain
+	vmConfig   vm.Config
+	parent     *types.Block
+	start      time.Time
+	wg         sync.WaitGroup
+	ctx        context.Context
+	cancelCtx  context.CancelFunc
+	replayMode bool
 
 	// accumulators
 	ccc            *circuitcapacitychecker.CircuitCapacityChecker
@@ -114,6 +115,12 @@ func (p *Pipeline) WithBeforeTxHook(beforeTxHook func()) *Pipeline {
 	return p
 }
 
+// WithReplayMode enables the replay mode that allows L1 messages to be skipped
+func (p *Pipeline) WithReplayMode() *Pipeline {
+	p.replayMode = true
+	return p
+}
+
 func (p *Pipeline) Start(deadline time.Time) error {
 	p.start = time.Now()
 	p.txnQueue = make(chan *types.Transaction)
@@ -266,7 +273,7 @@ func (p *Pipeline) traceAndApplyStage(txsIn <-chan *types.Transaction) (<-chan e
 				return
 			}
 
-			if tx.IsL1MessageTx() && tx.AsL1MessageTx().QueueIndex != p.nextL1MsgIndex {
+			if !p.replayMode && tx.IsL1MessageTx() && tx.AsL1MessageTx().QueueIndex != p.nextL1MsgIndex {
 				// Continue, we might still be able to include some L2 messages
 				sendCancellable(resCh, ErrUnexpectedL1MessageIndex, p.ctx.Done())
 				continue

rollup/state_processor/pipelined_state_processor.go

@@ -43,7 +43,7 @@ func (p *Processor) Process(block *types.Block, statedb *state.StateDB, cfg vm.C
 		}
 	}
 
-	pl := pipeline.NewPipeline(p.chain, cfg, statedb, header, nextL1MsgIndex, p.ccc)
+	pl := pipeline.NewPipeline(p.chain, cfg, statedb, header, nextL1MsgIndex, p.ccc).WithReplayMode()
 	pl.Start(time.Now().Add(time.Minute))
 	defer pl.Release()