add durable frame_no to proto

Author: MarinPostma

bottomless/src/replicator.rs

@@ -1652,6 +1652,7 @@ impl Replicator {
                     let frame = RpcFrame {
                         data: frame_to_inject.bytes(),
                         timestamp: None,
+                        durable_frame_no: None,
                     };
                     injector.inject_frame(frame).await?;
                     applied_wal_frame = true;

libsql-replication/proto/replication_log.proto

@@ -36,6 +36,7 @@ message Frame {
     // if this frames is a commit frame, then this can be set
     // to the time when the transaction was commited
     optional int64 timestamp = 2;
+    optional int64 durable_frame_no = 3;
 }
 
 message Frames {

libsql-replication/src/generated/wal_log.rs

@@ -83,6 +83,8 @@ pub struct Frame {
     /// to the time when the transaction was commited
     #[prost(int64, optional, tag = "2")]
     pub timestamp: ::core::option::Option<i64>,
+    #[prost(int64, optional, tag = "3")]
+    pub durable_frame_no: ::core::option::Option<i64>,
 }
 #[allow(clippy::derive_partial_eq_without_eq)]
 #[derive(Clone, PartialEq, ::prost::Message)]

libsql-replication/src/replicator.rs

@@ -772,6 +772,7 @@ mod test {
                         .map(|f| RpcFrame {
                             data: f.bytes(),
                             timestamp: None,
+                            durable_frame_no: None,
                         })
                         .take(2)
                         .map(Ok)
@@ -785,6 +786,7 @@ mod test {
                         .map(|f| RpcFrame {
                             data: f.bytes(),
                             timestamp: None,
+                            durable_frame_no: None,
                         })
                         .map(Ok)
                         .collect::<Vec<_>>();

libsql-server/src/rpc/replication/libsql_replicator.rs

@@ -102,6 +102,7 @@ where
                         Poll::Ready(Some(Ok(RpcFrame {
                             data,
                             timestamp: None,
+                            durable_frame_no: None,
                         })))
                     }
                     WalFlavor::Sqlite => {
@@ -116,6 +117,7 @@ where
                         Poll::Ready(Some(Ok(RpcFrame {
                             data: frame.bytes(),
                             timestamp: None,
+                            durable_frame_no: None,
                         })))
                     }
                 }

libsql-server/src/rpc/replication/replication_log.rs

@@ -187,6 +187,7 @@ fn map_frame_stream_output(
         Ok((frame, ts)) => Ok(Frame {
             data: frame.bytes(),
             timestamp: ts.map(|ts| ts.timestamp_millis()),
+            durable_frame_no: None,
         }),
         Err(LogReadError::SnapshotRequired) => Err(Status::new(
             tonic::Code::FailedPrecondition,
@@ -431,6 +432,7 @@ mod snapshot_stream {
                         yield Ok(Frame {
                             data: libsql_replication::frame::Frame::from(frame).bytes(),
                             timestamp: None,
+                            durable_frame_no: None,
                         });
                     }
                     Err(e) => {

libsql/src/replication/local_client.rs

@@ -47,7 +47,7 @@ impl ReplicatorClient for LocalClient {
     async fn next_frames(&mut self) -> Result<Self::FrameStream, Error> {
         match self.frames.take() {
             Some(Frames::Vec(f)) => {
-                let iter = f.into_iter().map(|f| RpcFrame { data: f.bytes(), timestamp: None }).map(Ok);
+                let iter = f.into_iter().map(|f| RpcFrame { data: f.bytes(), timestamp: None, durable_frame_no: None }).map(Ok);
                 Ok(Box::pin(tokio_stream::iter(iter)))
             }
             Some(f @ Frames::Snapshot(_)) => {
@@ -72,7 +72,7 @@ impl ReplicatorClient for LocalClient {
                             next.header_mut().size_after = size_after.into();
                         }
                         let frame = Frame::from(next);
-                        yield RpcFrame { data: frame.bytes(), timestamp: None };
+                        yield RpcFrame { data: frame.bytes(), timestamp: None, durable_frame_no: None };
                     }
                 };