chore(sequencer-relayer): change blob submitter to use boxed blocks

crates/astria-sequencer-relayer/src/relayer/mod.rs

@@ -1,4 +1,4 @@
 use std::{
     path::PathBuf,
     sync::Arc,
     time::Duration,
@@ -83,6 +83,9 @@
     IncludeRollup,
 };
 
+type ForwardFut<'a> =
+    Fuse<BoxFuture<'a, Result<(), tokio::sync::mpsc::error::SendError<Box<SequencerBlock>>>>>;
+
 pub(crate) struct Relayer {
     /// A token to notify relayer that it should shut down.
     #[expect(
@@ -172,7 +175,7 @@
         // future to forward a sequencer block to the celestia-submission-task.
         // gets set in the select-loop if the task is at capacity.
         let mut forward_once_free: Fuse<
-            BoxFuture<Result<(), tokio::sync::mpsc::error::SendError<SequencerBlock>>>,
+            BoxFuture<Result<(), tokio::sync::mpsc::error::SendError<Box<SequencerBlock>>>>,
         > = Fuse::terminated();
 
         self.state.set_ready();
@@ -282,17 +285,15 @@
         block: SequencerBlock,
         block_stream: &mut read::BlockStream,
         submitter: write::BlobSubmitterHandle,
-        forward: &mut Fuse<
-            BoxFuture<Result<(), tokio::sync::mpsc::error::SendError<SequencerBlock>>>,
-        >,
+        forward: &mut ForwardFut,
     ) -> eyre::Result<()> {
         assert!(
             forward.is_terminated(),
             "block stream must be paused and not yield blocks when the blob submitter is \
              congested and this future is in-flight",
         );
 
-        if let Err(error) = submitter.try_send(block) {
+        if let Err(error) = submitter.try_send(Box::new(block)) {
             debug!(
                 // Just print the error directly: TrySendError has no cause chain.
                 %error,

crates/astria-sequencer-relayer/src/relayer/write/mod.rs

@@ -1,4 +1,4 @@
 //! A task writing sequencer blocks to Celestia.
 //!
 //! [`BlobSubmitter`] receives [`SequencerBlock`]s over a channel,
 //! converts them to Celestia [`Blob`]s, and writes them to Celestia
@@ -86,7 +86,7 @@
 
 #[derive(Clone)]
 pub(super) struct BlobSubmitterHandle {
-    tx: mpsc::Sender<SequencerBlock>,
+    tx: mpsc::Sender<Box<SequencerBlock>>,
 }
 
 impl BlobSubmitterHandle {
@@ -95,8 +95,8 @@
     /// This is a thin wrapper around [`mpsc::Sender::try_send`].
     pub(super) fn try_send(
         &self,
-        block: SequencerBlock,
-    ) -> Result<(), TrySendError<SequencerBlock>> {
+        block: Box<SequencerBlock>,
+    ) -> Result<(), TrySendError<Box<SequencerBlock>>> {
         self.tx.try_send(block)
     }
 
@@ -105,8 +105,8 @@
     /// This is a thin wrapper around [`mpsc::Sender::send`].
     pub(super) async fn send(
         &self,
-        block: SequencerBlock,
-    ) -> Result<(), SendError<SequencerBlock>> {
+        block: Box<SequencerBlock>,
+    ) -> Result<(), SendError<Box<SequencerBlock>>> {
         self.tx.send(block).await
     }
 }
@@ -116,7 +116,7 @@
     client_builder: CelestiaClientBuilder,
 
     /// The channel over which sequencer blocks are received.
-    blocks: mpsc::Receiver<SequencerBlock>,
+    blocks: mpsc::Receiver<Box<SequencerBlock>>,
 
     /// The accumulator of all data that will be submitted to Celestia on the next submission.
     next_submission: NextSubmission,
@@ -253,7 +253,7 @@
                             sequencer_height = %block.height(),
                             "skipping sequencer block as already included in previous submission"
                         ));
-                    } else if let Err(error) = self.add_sequencer_block_to_next_submission(block) {
+                    } else if let Err(error) = self.add_sequencer_block_to_next_submission(*block) {
                         break Err(error).wrap_err(
                             "critically failed adding Sequencer block to next submission"
                         );