feat(derive): Holocene Buffer Flushing

bin/client/src/l1/driver.rs

@@ -16,8 +16,8 @@ use kona_derive::{
     pipeline::{DerivationPipeline, Pipeline, PipelineBuilder, StepResult},
     sources::EthereumDataSource,
     stages::{
-        AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
-        L1Traversal,
+        AttributesQueue, BatchQueue, BatchStream, ChannelBank, ChannelReader, FrameQueue,
+        L1Retrieval, L1Traversal,
     },
     traits::{BlobProvider, ChainProvider, L2ChainProvider, OriginProvider},
 };
@@ -46,8 +46,10 @@ pub type OracleAttributesBuilder<O> =
 /// An oracle-backed attributes queue for the derivation pipeline.
 pub type OracleAttributesQueue<DAP, O> = AttributesQueue<
     BatchQueue<
-        ChannelReader<
-            ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<OracleL1ChainProvider<O>>>>>,
+        BatchStream<
+            ChannelReader<
+                ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<OracleL1ChainProvider<O>>>>>,
+            >,
         >,
         OracleL2ChainProvider<O>,
     >,

crates/derive/src/online/pipeline.rs

@@ -11,7 +11,8 @@ use op_alloy_protocol::BlockInfo;
 
 // Pipeline internal stages aren't re-exported at the module-level.
 use crate::stages::{
-    AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval, L1Traversal,
+    AttributesQueue, BatchQueue, BatchStream, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
+    L1Traversal,
 };
 
 /// An online derivation pipeline.
@@ -32,7 +33,11 @@ pub type OnlineAttributesBuilder =
 /// An `online` attributes queue for the derivation pipeline.
 pub type OnlineAttributesQueue<DAP> = AttributesQueue<
     BatchQueue<
-        ChannelReader<ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<AlloyChainProvider>>>>>,
+        BatchStream<
+            ChannelReader<
+                ChannelBank<FrameQueue<L1Retrieval<DAP, L1Traversal<AlloyChainProvider>>>>,
+            >,
+        >,
         AlloyL2ChainProvider,
     >,
     OnlineAttributesBuilder,

crates/derive/src/pipeline/builder.rs

@@ -4,7 +4,8 @@ use super::{
     AttributesBuilder, ChainProvider, DataAvailabilityProvider, DerivationPipeline, L2ChainProvider,
 };
 use crate::stages::{
-    AttributesQueue, BatchQueue, ChannelBank, ChannelReader, FrameQueue, L1Retrieval, L1Traversal,
+    AttributesQueue, BatchQueue, BatchStream, ChannelBank, ChannelReader, FrameQueue, L1Retrieval,
+    L1Traversal,
 };
 use alloc::sync::Arc;
 use core::fmt::Debug;
@@ -16,7 +17,8 @@ type L1RetrievalStage<DAP, P> = L1Retrieval<DAP, L1TraversalStage<P>>;
 type FrameQueueStage<DAP, P> = FrameQueue<L1RetrievalStage<DAP, P>>;
 type ChannelBankStage<DAP, P> = ChannelBank<FrameQueueStage<DAP, P>>;
 type ChannelReaderStage<DAP, P> = ChannelReader<ChannelBankStage<DAP, P>>;
-type BatchQueueStage<DAP, P, T> = BatchQueue<ChannelReaderStage<DAP, P>, T>;
+type BatchStreamStage<DAP, P> = BatchStream<ChannelReaderStage<DAP, P>>;
+type BatchQueueStage<DAP, P, T> = BatchQueue<BatchStreamStage<DAP, P>, T>;
 type AttributesQueueStage<DAP, P, T, B> = AttributesQueue<BatchQueueStage<DAP, P, T>, B>;
 
 /// The `PipelineBuilder` constructs a [DerivationPipeline] using a builder pattern.
@@ -132,8 +134,9 @@ where
         let frame_queue = FrameQueue::new(l1_retrieval);
         let channel_bank = ChannelBank::new(Arc::clone(&rollup_config), frame_queue);
         let channel_reader = ChannelReader::new(channel_bank, Arc::clone(&rollup_config));
+        let batch_stream = BatchStream::new(channel_reader, rollup_config.clone());
         let batch_queue =
-            BatchQueue::new(rollup_config.clone(), channel_reader, l2_chain_provider.clone());
+            BatchQueue::new(rollup_config.clone(), batch_stream, l2_chain_provider.clone());
         let attributes =
             AttributesQueue::new(rollup_config.clone(), batch_queue, attributes_builder);
 

crates/derive/src/stages/batch_queue.rs

@@ -23,6 +23,10 @@ pub trait BatchQueueProvider {
     ///
     /// [ChannelReader]: crate::stages::ChannelReader
     async fn next_batch(&mut self) -> PipelineResult<Batch>;
+
+    /// Allows the [BatchQueue] to flush the buffer in the [crate::stages::BatchStream]
+    /// if an invalid single batch is found. Pre-holocene hardfork, this will be a no-op.
+    fn flush(&mut self);
 }
 
 /// [BatchQueue] is responsible for o rdering unordered batches
@@ -146,6 +150,9 @@ where
                     remaining.push(batch.clone());
                 }
                 BatchValidity::Drop => {
+                    // If we drop a batch, flush previous batches buffered in the BatchStream
+                    // stage.
+                    self.prev.flush();
                     warn!(target: "batch-queue", "Dropping batch with parent: {}", parent.block_info);
                     continue;
                 }
@@ -233,6 +240,7 @@ where
         let data = BatchWithInclusionBlock { inclusion_block: origin, batch };
         // If we drop the batch, validation logs the drop reason with WARN level.
         if data.check_batch(&self.cfg, &self.l1_blocks, parent, &mut self.fetcher).await.is_drop() {
+            self.prev.flush();
             return Ok(());
         }
         self.batches.push(data);

crates/derive/src/stages/batch_stream.rs

@@ -67,6 +67,12 @@ impl<P> BatchQueueProvider for BatchStream<P>
 where
     P: BatchQueueProvider + OriginAdvancer + OriginProvider + ResettableStage + Send + Debug,
 {
+    fn flush(&mut self) {
+        if self.is_active().unwrap_or(false) {
+            self.buffer.clear();
+        }
+    }
+
     async fn next_batch(&mut self) -> PipelineResult<Batch> {
         // If the stage is not active, "pass" the next batch
         // through this stage to the BatchQueue stage.

crates/derive/src/stages/channel_reader.rs

@@ -99,6 +99,9 @@ impl<P> BatchQueueProvider for ChannelReader<P>
 where
     P: ChannelReaderProvider + OriginAdvancer + OriginProvider + ResettableStage + Send + Debug,
 {
+    fn flush(&mut self) { /* noop */
+    }
+
     async fn next_batch(&mut self) -> PipelineResult<Batch> {
         crate::timer!(START, STAGE_ADVANCE_RESPONSE_TIME, &["channel_reader"], timer);
         if let Err(e) = self.set_batch_reader().await {

crates/derive/src/stages/test_utils/batch_queue.rs

@@ -35,6 +35,9 @@ impl OriginProvider for MockBatchQueueProvider {
 
 #[async_trait]
 impl BatchQueueProvider for MockBatchQueueProvider {
+    fn flush(&mut self) { /* noop */
+    }
+
     async fn next_batch(&mut self) -> PipelineResult<Batch> {
         self.batches.pop().ok_or(PipelineError::Eof.temp())?
     }