[Storage] Remove unneeded bounds on journal::contiguous::Contiguous (#2064)

Author: danlaine

storage/src/journal/contiguous/fixed.rs

@@ -620,9 +620,7 @@ impl<E: Storage + Metrics, A: CodecFixed<Cfg = ()>> Journal<E, A> {
 }
 
 // Implement Contiguous trait for fixed-length journals
-impl<E: Storage + Metrics, A: CodecFixed<Cfg = ()> + Send + Sync> super::Contiguous
-    for Journal<E, A>
-{
+impl<E: Storage + Metrics, A: CodecFixed<Cfg = ()>> super::Contiguous for Journal<E, A> {
     type Item = A;
 
     async fn append(&mut self, item: Self::Item) -> Result<u64, Error> {

storage/src/journal/contiguous/mod.rs

@@ -32,16 +32,14 @@ pub trait Contiguous {
     ///
     /// Returns an error if the underlying storage operation fails or if the item cannot
     /// be encoded.
-    fn append(
-        &mut self,
-        item: Self::Item,
-    ) -> impl std::future::Future<Output = Result<u64, Error>> + Send;
+    fn append(&mut self, item: Self::Item)
+        -> impl std::future::Future<Output = Result<u64, Error>>;
 
     /// Return the total number of items that have been appended to the journal.
     ///
     /// This count is NOT affected by pruning. The next appended item will receive this
     /// position as its value.
-    fn size(&self) -> impl std::future::Future<Output = u64> + Send;
+    fn size(&self) -> impl std::future::Future<Output = u64>;
 
     /// Return the position of the oldest item still retained in the journal.
     ///
@@ -50,9 +48,7 @@ pub trait Contiguous {
     /// After pruning, this returns the position of the first item that remains.
     /// Note that due to section/blob alignment, this may be less than the `min_position`
     /// passed to `prune()`.
-    fn oldest_retained_pos(
-        &self,
-    ) -> impl std::future::Future<Output = Result<Option<u64>, Error>> + Send;
+    fn oldest_retained_pos(&self) -> impl std::future::Future<Output = Result<Option<u64>, Error>>;
 
     /// Prune items at positions strictly less than `min_position`.
     ///
@@ -72,7 +68,7 @@ pub trait Contiguous {
     fn prune(
         &mut self,
         min_position: u64,
-    ) -> impl std::future::Future<Output = Result<bool, Error>> + Send;
+    ) -> impl std::future::Future<Output = Result<bool, Error>>;
 
     /// Rewind the journal to the given size, discarding items from the end.
     ///
@@ -95,7 +91,7 @@ pub trait Contiguous {
     ///
     /// Returns [Error::InvalidRewind] if size is invalid (too large or points to pruned data).
     /// Returns an error if the underlying storage operation fails.
-    fn rewind(&mut self, size: u64) -> impl std::future::Future<Output = Result<(), Error>> + Send;
+    fn rewind(&mut self, size: u64) -> impl std::future::Future<Output = Result<(), Error>>;
 
     /// Return a stream of all items in the journal starting from `start_pos`.
     ///
@@ -112,31 +108,28 @@ pub trait Contiguous {
         buffer: NonZeroUsize,
     ) -> impl std::future::Future<
         Output = Result<impl Stream<Item = Result<(u64, Self::Item), Error>> + '_, Error>,
-    > + Send;
+    >;
 
     /// Read the item at the given position.
     ///
     /// # Errors
     ///
     /// - Returns [Error::ItemPruned] if the item at `position` has been pruned.
     /// - Returns [Error::ItemOutOfRange] if the item at `position` does not exist.
-    fn read(
-        &self,
-        position: u64,
-    ) -> impl std::future::Future<Output = Result<Self::Item, Error>> + Send;
+    fn read(&self, position: u64) -> impl std::future::Future<Output = Result<Self::Item, Error>>;
 
     /// Sync all pending writes to storage.
     ///
     /// This ensures all previously appended items are durably persisted.
-    fn sync(&mut self) -> impl std::future::Future<Output = Result<(), Error>> + Send;
+    fn sync(&mut self) -> impl std::future::Future<Output = Result<(), Error>>;
 
     /// Close the journal, syncing all pending writes and releasing resources.
-    fn close(self) -> impl std::future::Future<Output = Result<(), Error>> + Send;
+    fn close(self) -> impl std::future::Future<Output = Result<(), Error>>;
 
     /// Destroy the journal, removing all associated storage.
     ///
     /// This method consumes the journal and deletes all persisted data including blobs,
     /// metadata, and any other storage artifacts. Use this for cleanup in tests or when
     /// permanently removing a journal.
-    fn destroy(self) -> impl std::future::Future<Output = Result<(), Error>> + Send;
+    fn destroy(self) -> impl std::future::Future<Output = Result<(), Error>>;
 }

storage/src/journal/contiguous/variable.rs

@@ -151,7 +151,7 @@ pub struct Journal<E: Storage + Metrics, V: Codec> {
     oldest_retained_pos: u64,
 }
 
-impl<E: Storage + Metrics, V: Codec + Send> Journal<E, V> {
+impl<E: Storage + Metrics, V: Codec> Journal<E, V> {
     /// Initialize a contiguous variable journal.
     ///
     /// # Crash Recovery
@@ -475,53 +475,6 @@ impl<E: Storage + Metrics, V: Codec + Send> Journal<E, V> {
         Ok(pruned)
     }
 
-    /// Return a stream of all items in the journal starting from `start_pos`.
-    ///
-    /// Each item is yielded as a tuple `(position, item)` where position is the item's
-    /// position in the journal.
-    ///
-    /// # Errors
-    ///
-    /// Returns an error if `start_pos` exceeds the journal size or if any storage/decoding
-    /// errors occur during replay.
-    pub async fn replay(
-        &self,
-        start_pos: u64,
-        buffer_size: NonZeroUsize,
-    ) -> Result<Pin<Box<dyn Stream<Item = Result<(u64, V), Error>> + Send + '_>>, Error> {
-        // Validate start position is within bounds.
-        if start_pos < self.oldest_retained_pos {
-            return Err(Error::ItemPruned(start_pos));
-        }
-        if start_pos > self.size {
-            return Err(Error::ItemOutOfRange(start_pos));
-        }
-
-        // If replaying at exactly size, return empty stream
-        if start_pos == self.size {
-            return Ok(Box::pin(stream::empty()));
-        }
-
-        // Use offsets index to find offset to start from, calculate section from position
-        let start_offset = self.offsets.read(start_pos).await?;
-        let start_section = position_to_section(start_pos, self.items_per_section);
-        let data_stream = self
-            .data
-            .replay(start_section, start_offset, buffer_size)
-            .await?;
-
-        // Transform the stream to include position information
-        let transformed = data_stream.enumerate().map(move |(idx, result)| {
-            result.map(|(_section, _offset, _size, item)| {
-                // Calculate position: start_pos + items read
-                let pos = start_pos + idx as u64;
-                (pos, item)
-            })
-        });
-
-        Ok(Box::pin(transformed))
-    }
-
     /// Read the item at the given position.
     ///
     /// # Errors
@@ -803,8 +756,57 @@ impl<E: Storage + Metrics, V: Codec + Send> Journal<E, V> {
     }
 }
 
+impl<E: Storage + Metrics, V: Codec + Send> Journal<E, V> {
+    /// Return a stream of all items in the journal starting from `start_pos`.
+    ///
+    /// Each item is yielded as a tuple `(position, item)` where position is the item's
+    /// position in the journal.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if `start_pos` exceeds the journal size or if any storage/decoding
+    /// errors occur during replay.
+    pub async fn replay(
+        &self,
+        start_pos: u64,
+        buffer_size: NonZeroUsize,
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<(u64, V), Error>> + '_>>, Error> {
+        // Validate start position is within bounds.
+        if start_pos < self.oldest_retained_pos {
+            return Err(Error::ItemPruned(start_pos));
+        }
+        if start_pos > self.size {
+            return Err(Error::ItemOutOfRange(start_pos));
+        }
+
+        // If replaying at exactly size, return empty stream
+        if start_pos == self.size {
+            return Ok(Box::pin(stream::empty()));
+        }
+
+        // Use offsets index to find offset to start from, calculate section from position
+        let start_offset = self.offsets.read(start_pos).await?;
+        let start_section = position_to_section(start_pos, self.items_per_section);
+        let data_stream = self
+            .data
+            .replay(start_section, start_offset, buffer_size)
+            .await?;
+
+        // Transform the stream to include position information
+        let transformed = data_stream.enumerate().map(move |(idx, result)| {
+            result.map(|(_section, _offset, _size, item)| {
+                // Calculate position: start_pos + items read
+                let pos = start_pos + idx as u64;
+                (pos, item)
+            })
+        });
+
+        Ok(Box::pin(transformed))
+    }
+}
+
 // Implement Contiguous trait for variable-length items
-impl<E: Storage + Metrics, V: Codec + Send + Sync> Contiguous for Journal<E, V> {
+impl<E: Storage + Metrics, V: Codec + Send> Contiguous for Journal<E, V> {
     type Item = V;
 
     async fn append(&mut self, item: Self::Item) -> Result<u64, Error> {