feat(derive): wire up the batch span stage

op-rs · Sep 26, 2024 · dac37f6 · dac37f6
1 parent a610b16
commit dac37f6
Show file tree

Hide file tree

Showing 4 changed files with 21 additions and 11 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/bin/client/src/l1/driver.rs b/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>,
     >,

diff --git a/crates/derive/src/online/pipeline.rs b/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,

diff --git a/crates/derive/src/pipeline/builder.rs b/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);