pass segment timestamp to storage

Author: MarinPostma

libsql-wal/src/segment/mod.rs

@@ -15,6 +15,8 @@ use std::mem::size_of;
 use std::num::NonZeroU64;
 use std::sync::Arc;
 
+use chrono::DateTime;
+use chrono::Utc;
 use zerocopy::byteorder::little_endian::{U128, U16, U32, U64};
 use zerocopy::AsBytes;
 
@@ -171,6 +173,7 @@ pub trait Segment: Send + Sync + 'static {
     async fn read_frame_offset_async<B>(&self, offset: u32, buf: B) -> (B, Result<()>)
     where
         B: IoBufMut + Send + 'static;
+    fn timestamp(&self) -> DateTime<Utc>;
 
     fn destroy<IO: Io>(&self, io: &IO) -> impl Future<Output = ()>;
 }
@@ -196,15 +199,12 @@ impl<T: Segment> Segment for Arc<T> {
         self.as_ref().index()
     }
 
-    fn read_page(&self, page_no: u32, max_frame_no: u64, buf: &mut [u8]) -> io::Result<bool> {
-        self.as_ref().read_page(page_no, max_frame_no, buf)
+    fn is_storable(&self) -> bool {
+        self.as_ref().is_storable()
     }
 
-    async fn read_frame_offset_async<B>(&self, offset: u32, buf: B) -> (B, Result<()>)
-    where
-        B: IoBufMut + Send + 'static,
-    {
-        self.as_ref().read_frame_offset_async(offset, buf).await
+    fn read_page(&self, page_no: u32, max_frame_no: u64, buf: &mut [u8]) -> io::Result<bool> {
+        self.as_ref().read_page(page_no, max_frame_no, buf)
     }
 
     fn is_checkpointable(&self) -> bool {
@@ -215,12 +215,19 @@ impl<T: Segment> Segment for Arc<T> {
         self.as_ref().size_after()
     }
 
+    async fn read_frame_offset_async<B>(&self, offset: u32, buf: B) -> (B, Result<()>)
+    where
+        B: IoBufMut + Send + 'static,
+    {
+        self.as_ref().read_frame_offset_async(offset, buf).await
+    }
+
     fn destroy<IO: Io>(&self, io: &IO) -> impl Future<Output = ()> {
         self.as_ref().destroy(io)
     }
 
-    fn is_storable(&self) -> bool {
-        self.as_ref().is_storable()
+    fn timestamp(&self) -> DateTime<Utc> {
+        self.as_ref().timestamp()
     }
 }
 

libsql-wal/src/segment/sealed.rs

@@ -9,6 +9,7 @@ use std::sync::{
     Arc,
 };
 
+use chrono::prelude::{DateTime, Utc};
 use fst::{Map, MapBuilder, Streamer};
 use zerocopy::{AsBytes, FromZeroes};
 
@@ -156,6 +157,17 @@ where
         &self.index
     }
 
+    fn is_storable(&self) -> bool {
+        // we don't store unordered segments, since they only happen in two cases:
+        // - in a replica: no need for storage
+        // - in a primary, on recovery from storage: we don't want to override remote
+        // segment.
+        !self
+            .header()
+            .flags()
+            .contains(SegmentFlags::FRAME_UNORDERED)
+    }
+
     fn read_page(&self, page_no: u32, max_frame_no: u64, buf: &mut [u8]) -> std::io::Result<bool> {
         if self.header().start_frame_no.get() > max_frame_no {
             return Ok(false);
@@ -171,16 +183,6 @@ where
         Ok(false)
     }
 
-    async fn read_frame_offset_async<B>(&self, offset: u32, buf: B) -> (B, Result<()>)
-    where
-        B: IoBufMut + Send + 'static,
-    {
-        assert_eq!(buf.bytes_total(), size_of::<Frame>());
-        let frame_offset = frame_offset(offset);
-        let (buf, ret) = self.file.read_exact_at_async(buf, frame_offset as _).await;
-        (buf, ret.map_err(Into::into))
-    }
-
     fn is_checkpointable(&self) -> bool {
         let read_locks = self.read_locks.load(Ordering::Relaxed);
         tracing::debug!(read_locks);
@@ -191,6 +193,16 @@ where
         self.header().size_after()
     }
 
+    async fn read_frame_offset_async<B>(&self, offset: u32, buf: B) -> (B, Result<()>)
+    where
+        B: IoBufMut + Send + 'static,
+    {
+        assert_eq!(buf.bytes_total(), size_of::<Frame>());
+        let frame_offset = frame_offset(offset);
+        let (buf, ret) = self.file.read_exact_at_async(buf, frame_offset as _).await;
+        (buf, ret.map_err(Into::into))
+    }
+
     fn destroy<IO: crate::io::Io>(&self, io: &IO) -> impl std::future::Future<Output = ()> {
         async move {
             if let Err(e) = io.remove_file_async(&self.path).await {
@@ -199,15 +211,9 @@ where
         }
     }
 
-    fn is_storable(&self) -> bool {
-        // we don't store unordered segments, since they only happen in two cases:
-        // - in a replica: no need for storage
-        // - in a primary, on recovery from storage: we don't want to override remote
-        // segment.
-        !self
-            .header()
-            .flags()
-            .contains(SegmentFlags::FRAME_UNORDERED)
+    fn timestamp(&self) -> DateTime<Utc> {
+        assert_ne!(self.header().sealed_at_timestamp.get(), 0, "segment was not sealed properly");
+        DateTime::from_timestamp_millis(self.header().sealed_at_timestamp.get() as _).expect("this should be a guaranteed roundtrip with DateTime::timestamp_millis")
     }
 }
 

libsql-wal/src/storage/backend/mod.rs

@@ -22,7 +22,7 @@ pub struct SegmentMeta {
     pub segment_id: Uuid,
     pub start_frame_no: u64,
     pub end_frame_no: u64,
-    pub created_at: DateTime<Utc>,
+    pub segment_timestamp: DateTime<Utc>,
 }
 
 pub struct RestoreRequest {}

libsql-wal/src/storage/backend/s3.rs

@@ -754,7 +754,7 @@ mod tests {
                     segment_id: Uuid::new_v4(),
                     start_frame_no: 0u64.into(),
                     end_frame_no: 64u64.into(),
-                    created_at: Utc::now(),
+                    segment_timestamp: Utc::now(),
                 },
                 std::fs::File::open(&f_path).unwrap(),
                 index,
@@ -776,7 +776,7 @@ mod tests {
                     segment_id: Uuid::new_v4(),
                     start_frame_no: 64u64.into(),
                     end_frame_no: 128u64.into(),
-                    created_at: Utc::now(),
+                    segment_timestamp: Utc::now(),
                 },
                 std::fs::File::open(&f_path).unwrap(),
                 index,

libsql-wal/src/storage/compaction/mod.rs

@@ -304,7 +304,7 @@ impl<B> Compactor<B> {
                     segment_id: Uuid::new_v4(),
                     start_frame_no: start,
                     end_frame_no: end,
-                    created_at: Utc::now(),
+                    segment_timestamp: Utc::now(),
                 },
                 out_file,
                 out_index.into_inner().unwrap(),

libsql-wal/src/storage/job.rs

@@ -69,7 +69,7 @@ where
             namespace: self.request.namespace.clone(),
             start_frame_no: segment.start_frame_no(),
             end_frame_no: segment.last_committed(),
-            created_at: io.now(),
+            segment_timestamp: segment.timestamp(),
         };
         let config = self
             .request
@@ -108,6 +108,7 @@ mod test {
     // use std::sync::atomic::AtomicBool;
     use std::sync::Arc;
 
+    use chrono::prelude::DateTime;
     use chrono::Utc;
     // use fst::{Map, Streamer};
     // use libsql_sys::rusqlite::OpenFlags;
@@ -428,6 +429,10 @@ mod test {
             fn is_storable(&self) -> bool {
                 true
             }
+
+            fn timestamp(&self) -> DateTime<Utc> {
+                Utc::now()
+            }
         }
 
         struct TestBackend;