refactor: seqno map

Author: marvin-j97

src/lib.rs

@@ -118,7 +118,7 @@ mod keyspace;
 mod locked_file;
 mod meta_keyspace;
 mod path;
-mod poison_dart;
+mod poison;
 mod readable;
 mod recovery;
 mod snapshot;

src/recovery.rs

@@ -141,7 +141,7 @@ pub fn recover_sealed_memtables(
         let raw_reader = JournalReader::new(journal_path)?;
         let reader = JournalBatchReader::new(raw_reader);
 
-        let mut watermarks: HashMap<InternalKeyspaceId, EvictionWatermark> = HashMap::default();
+        let mut watermarks: HashMap<_, EvictionWatermark> = HashMap::default();
 
         for batch in reader {
             let batch = batch?;
@@ -216,8 +216,8 @@ pub fn recover_sealed_memtables(
 
             let keyspace_lsn = tree.get_highest_persisted_seqno();
 
-            // IMPORTANT: Only apply sealed memtables to keyspaces
-            // that have a lower seqno to avoid double flushing
+            // IMPORTANT: Apply the sealed memtable only if it contains data newer than the
+            // keyspace's persisted LSN.
             let should_skip_sealed_memtable =
                 keyspace_lsn.is_some_and(|keyspace_lsn| keyspace_lsn >= wm.lsn);
 

src/supervisor.rs

@@ -50,4 +50,24 @@ impl Supervisor {
     pub fn new(inner: SupervisorInner) -> Self {
         Self(Arc::new(inner))
     }
+
+    pub fn build_seqno_map(
+        &self,
+        keyspaces: &Keyspaces,
+    ) -> Vec<crate::journal::manager::EvictionWatermark> {
+        use crate::AbstractTree;
+
+        let mut seqnos = Vec::with_capacity(keyspaces.len());
+
+        for keyspace in keyspaces.values() {
+            if let Some(lsn) = keyspace.tree.get_highest_memtable_seqno() {
+                seqnos.push(crate::journal::manager::EvictionWatermark {
+                    lsn,
+                    keyspace: keyspace.clone(),
+                });
+            }
+        }
+
+        seqnos
+    }
 }

src/worker_pool.rs

@@ -3,11 +3,10 @@
 // (found in the LICENSE-* files in the repository)
 
 use crate::{
-    compaction::worker::run as run_compaction, flush::worker::run as run_flush,
-    journal::manager::EvictionWatermark, poison_dart::PoisonDart, stats::Stats,
-    supervisor::Supervisor, Keyspace,
+    compaction::worker::run as run_compaction, flush::worker::run as run_flush, poison::PoisonDart,
+    stats::Stats, supervisor::Supervisor, Keyspace,
 };
-use lsm_tree::{AbstractTree, MemtableId};
+use lsm_tree::MemtableId;
 use std::{
     borrow::Cow,
     sync::{atomic::AtomicUsize, Arc, Mutex},
@@ -164,18 +163,7 @@ fn worker_tick(ctx: &WorkerState) -> crate::Result<bool> {
                         #[expect(clippy::expect_used)]
                         let keyspaces = ctx.supervisor.keyspaces.write().expect("lock is poisoned");
 
-                        let mut seqnos = Vec::with_capacity(keyspaces.len());
-
-                        for keyspace in keyspaces.values() {
-                            if let Some(lsn) = keyspace.tree.get_highest_memtable_seqno() {
-                                seqnos.push(EvictionWatermark {
-                                    lsn,
-                                    keyspace: keyspace.clone(),
-                                });
-                            }
-                        }
-
-                        seqnos
+                        ctx.supervisor.build_seqno_map(&keyspaces)
                     };
 
                     journal_manager.rotate_journal(&mut journal_writer, seqno_map)?;