add sealed_at_timestamp field to SegmentHeader

Author: MarinPostma

libsql-wal/src/registry.rs

@@ -542,7 +542,7 @@ where
         )?;
         // sealing must the last fallible operation, because we don't want to end up in a situation
         // where the current log is sealed and it wasn't swapped.
-        if let Some(sealed) = current.seal()? {
+        if let Some(sealed) = current.seal(self.io.now())? {
             new.tail().push(sealed.clone());
             maybe_store_segment(
                 self.storage.as_ref(),

libsql-wal/src/segment/current.rs

@@ -9,6 +9,7 @@ use std::sync::{
     Arc,
 };
 
+use chrono::{DateTime, Utc};
 use crossbeam_skiplist::SkipMap;
 use fst::MapBuilder;
 use parking_lot::{Mutex, RwLock};
@@ -74,6 +75,7 @@ impl<F> CurrentSegment<F> {
             page_size: LIBSQL_PAGE_SIZE.into(),
             log_id: log_id.as_u128().into(),
             frame_count: 0.into(),
+            sealed_at_timestamp: 0.into(),
         };
 
         header.recompute_checksum();
@@ -408,7 +410,7 @@ impl<F> CurrentSegment<F> {
 
     /// It is expected that sealing is performed under a write lock
     #[tracing::instrument(skip_all)]
-    pub fn seal(&self) -> Result<Option<SealedSegment<F>>>
+    pub fn seal(&self, now: DateTime<Utc>) -> Result<Option<SealedSegment<F>>>
     where
         F: FileExt,
     {
@@ -442,6 +444,7 @@ impl<F> CurrentSegment<F> {
         header.index_size = index_size.into();
         let flags = header.flags();
         header.set_flags(flags | SegmentFlags::SEALED);
+        header.sealed_at_timestamp = (now.timestamp_millis() as u64).into();
         header.recompute_checksum();
         self.file.write_all_at(header.as_bytes(), 0)?;
 

libsql-wal/src/segment/mod.rs

@@ -66,6 +66,8 @@ pub struct SegmentHeader {
     /// we could do it without changing the header
     pub page_size: U16,
     pub log_id: U128,
+    /// ms, from unix epoch
+    pub sealed_at_timestamp: U64,
 
     /// checksum of the header fields, excluding the checksum itself. This field must be the last
     pub header_cheksum: U32,

libsql-wal/src/shared_wal.rs

@@ -79,7 +79,7 @@ impl<IO: Io> SharedWal<IO> {
         }
         // The current segment will not be used anymore. It's empty, but we still seal it so that
         // the next startup doesn't find an unsealed segment.
-        self.current.load().seal()?;
+        self.current.load().seal(self.io.now())?;
         tracing::info!("namespace shutdown");
         Ok(())
     }