patch segment timestamp on recovery

Author: MarinPostma

libsql-wal/src/segment/current.rs

@@ -455,6 +455,7 @@ impl<F> CurrentSegment<F> {
             self.file.clone(),
             self.path.clone(),
             self.read_locks.clone(),
+            now,
         )?;
 
         // we only flip the sealed mark when no more error can occur, or we risk to deadlock a read

libsql-wal/src/segment/sealed.rs

@@ -4,10 +4,8 @@ use std::io::{BufWriter, ErrorKind, Write};
 use std::mem::size_of;
 use std::ops::Deref;
 use std::path::{Path, PathBuf};
-use std::sync::{
-    atomic::{AtomicU64, Ordering},
-    Arc,
-};
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
 
 use chrono::prelude::{DateTime, Utc};
 use fst::{Map, MapBuilder, Streamer};
@@ -212,7 +210,7 @@ where
 }
 
 impl<F: FileExt> SealedSegment<F> {
-    pub fn open(file: Arc<F>, path: PathBuf, read_locks: Arc<AtomicU64>) -> Result<Option<Self>> {
+    pub fn open(file: Arc<F>, path: PathBuf, read_locks: Arc<AtomicU64>, now: DateTime<Utc>) -> Result<Option<Self>> {
         let mut header: SegmentHeader = SegmentHeader::new_zeroed();
         file.read_exact_at(header.as_bytes_mut(), 0)?;
 
@@ -230,7 +228,7 @@ impl<F: FileExt> SealedSegment<F> {
         // recover the index, and seal the segment.
         if !header.flags().contains(SegmentFlags::SEALED) {
             assert_eq!(header.index_offset.get(), 0, "{header:?}");
-            return Self::recover(file, path, header).map(Some);
+            return Self::recover(file, path, header, now).map(Some);
         }
 
         let mut slice = vec![0; index_len as usize];
@@ -248,7 +246,7 @@ impl<F: FileExt> SealedSegment<F> {
         }))
     }
 
-    fn recover(file: Arc<F>, path: PathBuf, mut header: SegmentHeader) -> Result<Self> {
+    fn recover(file: Arc<F>, path: PathBuf, mut header: SegmentHeader, now: DateTime<Utc>) -> Result<Self> {
         assert!(!header.is_empty());
         assert_eq!(header.index_size.get(), 0);
         assert_eq!(header.index_offset.get(), 0);
@@ -335,6 +333,7 @@ impl<F: FileExt> SealedSegment<F> {
         header.index_size = index_size.into();
         header.last_commited_frame_no = last_committed.into();
         header.size_after = size_after.into();
+        header.sealed_at_timestamp = (now.timestamp_millis() as u64).into();
         let flags = header.flags();
         header.set_flags(flags | SegmentFlags::SEALED);
         header.recompute_checksum();