return new_frame_notifier and log_id from shared_wal

Author: MarinPostma

libsql-wal/src/segment/current.rs

@@ -14,6 +14,7 @@ use fst::MapBuilder;
 use parking_lot::{Mutex, RwLock};
 use roaring::RoaringBitmap;
 use tokio_stream::Stream;
+use uuid::Uuid;
 use zerocopy::little_endian::U32;
 use zerocopy::{AsBytes, FromZeroes};
 
@@ -54,6 +55,7 @@ impl<F> CurrentSegment<F> {
         db_size: u32,
         tail: Arc<SegmentList<SealedSegment<F>>>,
         salt: u32,
+        log_id: Uuid,
     ) -> Result<Self>
     where
         F: FileExt,
@@ -70,6 +72,7 @@ impl<F> CurrentSegment<F> {
             version: LIBSQL_WAL_VERSION.into(),
             salt: salt.into(),
             page_size: LIBSQL_PAGE_SIZE.into(),
+            log_id: log_id.as_u128().into(),
         };
 
         header.recompute_checksum();
@@ -88,6 +91,10 @@ impl<F> CurrentSegment<F> {
         })
     }
 
+    pub fn log_id(&self) -> Uuid {
+        Uuid::from_u128(self.header.lock().log_id.get())
+    }
+
     pub fn is_empty(&self) -> bool {
         self.count_committed() == 0
     }

libsql-wal/src/segment/mod.rs

@@ -15,7 +15,7 @@ use std::mem::size_of;
 use std::num::NonZeroU64;
 use std::sync::Arc;
 
-use zerocopy::byteorder::little_endian::{U16, U32, U64};
+use zerocopy::byteorder::little_endian::{U128, U16, U32, U64};
 use zerocopy::AsBytes;
 
 use crate::error::{Error, Result};
@@ -63,6 +63,7 @@ pub struct SegmentHeader {
     /// right now we only support 4096, but if se decided to support other sizes,
     /// we could do it without changing the header
     pub page_size: U16,
+    pub log_id: U128,
 
     /// 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

@@ -1,13 +1,14 @@
 use std::collections::BTreeMap;
-use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::sync::Arc;
 use std::time::Instant;
 
 use arc_swap::ArcSwap;
 use crossbeam::deque::Injector;
 use crossbeam::sync::Unparker;
 use parking_lot::{Mutex, MutexGuard};
-use tokio::sync::mpsc;
+use tokio::sync::{mpsc, watch};
+use uuid::Uuid;
 
 use crate::checkpointer::CheckpointMessage;
 use crate::error::{Error, Result};
@@ -83,10 +84,18 @@ impl<IO: Io> SharedWal<IO> {
         Ok(())
     }
 
+    pub fn new_frame_notifier(&self) -> watch::Receiver<u64> {
+        self.new_frame_notifier.subscribe()
+    }
+
     pub fn db_size(&self) -> u32 {
         self.current.load().db_size()
     }
 
+    pub fn log_id(&self) -> Uuid {
+        self.current.load().log_id()
+    }
+
     #[tracing::instrument(skip_all)]
     pub fn begin_read(&self, conn_id: u64) -> ReadTransaction<IO::File> {
         // FIXME: this is not enough to just increment the counter, we must make sure that the segment