replicator returns Box<Frame>

Author: MarinPostma

libsql-wal/src/replication/injector.rs

@@ -97,7 +97,7 @@ mod test {
         primary_shared.last_committed_frame_no();
         for _ in 0..2 {
             let frame = stream.next().await.unwrap().unwrap();
-            injector.insert_frame(Box::new(frame)).await.unwrap();
+            injector.insert_frame(frame).await.unwrap();
         }
 
         replica_conn
@@ -118,7 +118,7 @@ mod test {
             .unwrap();
 
         let frame = stream.next().await.unwrap().unwrap();
-        injector.insert_frame(Box::new(frame)).await.unwrap();
+        injector.insert_frame(frame).await.unwrap();
 
         replica_conn
             .query_row("select count(*) from test", (), |r| {

libsql-wal/src/replication/replicator.rs

@@ -4,11 +4,13 @@ use roaring::RoaringBitmap;
 use tokio::sync::watch;
 use tokio_stream::{Stream, StreamExt};
 
-use crate::error::Result;
 use crate::io::Io;
+use crate::replication::Error;
 use crate::segment::Frame;
 use crate::shared_wal::SharedWal;
 
+use super::Result;
+
 pub struct Replicator<IO: Io> {
     shared: Arc<SharedWal<IO>>,
     new_frame_notifier: watch::Receiver<u64>,
@@ -38,7 +40,7 @@ impl<IO: Io> Replicator<IO> {
     ///
     /// In a single replication step, the replicator guarantees that a minimal set of frames is
     /// sent to the replica.
-    pub fn frame_stream(&mut self) -> impl Stream<Item = Result<Frame>> + '_ {
+    pub fn frame_stream(&mut self) -> impl Stream<Item = Result<Box<Frame>>> + '_ {
         async_stream::try_stream! {
             loop {
                 let most_recent_frame_no = *self
@@ -54,6 +56,7 @@ impl<IO: Io> Replicator<IO> {
                     let (stream, replicated_until, size_after) = current.frame_stream_from(self.next_frame_no, &mut seen);
                     let should_replicate_from_tail = replicated_until != self.next_frame_no;
 
+
                     // replicate from current
                     {
                         tokio::pin!(stream);
@@ -62,7 +65,7 @@ impl<IO: Io> Replicator<IO> {
 
                         loop {
                             let Some(frame) = stream.next().await else { break };
-                            let mut frame = frame?;
+                            let mut frame = frame.map_err(|e| Error::CurrentSegment(e.into()))?;
                             commit_frame_no = frame.header().frame_no().max(commit_frame_no);
                             if stream.peek().await.is_none() && !should_replicate_from_tail {
                                 frame.header_mut().set_size_after(size_after);