feat(wal): reclaim obsolete physical log segments

crates/lance-context-wal/src/raft_log.rs

@@ -11,8 +11,8 @@
 //! opaque descriptor and logical compaction cross one physical sync barrier.
 //! A received Ready snapshot can atomically replace the log and persist its
 //! suffix and HardState. Application snapshot files/transfer, driver integration,
-//! physical reclamation, membership changes and replica replacement remain
-//! higher-level responsibilities.
+//! maintenance scheduling, membership changes and replica replacement remain
+//! higher-level responsibilities. Physical log maintenance is explicit.
 //! Only the owning process may modify the files.
 
 use std::collections::{BTreeMap, BTreeSet};
@@ -29,6 +29,8 @@ use raft_proto::eraftpb::{ConfState, Entry, EntryType, HardState, Snapshot};
 use crate::placement::Layout;
 
 mod checkpoint;
+mod maintenance;
+pub use maintenance::LogMaintenance;
 #[cfg(test)]
 mod fault_tests;
 
@@ -59,6 +61,7 @@ impl MessageExt for EntryIndex {
 struct NodeEngine {
     engine: LogEngine,
     failed: AtomicBool,
+    maintenance_gate: RwLock<()>,
     append_gate: RwLock<()>,
     sync_gate: Mutex<()>,
     synced: AtomicU64,
@@ -76,6 +79,10 @@ impl NodeEngine {
     }
 
     fn write(&self, batch: &mut LogBatch) -> Result<()> {
+        let _maintenance = self
+            .maintenance_gate
+            .read()
+            .map_err(|_| anyhow::anyhow!("maintenance gate poisoned"))?;
         let result = self.append_and_sync(batch);
         if result.is_err() {
             self.failed.store(true, Ordering::Release);
@@ -110,19 +117,28 @@ impl NodeEngine {
                 .write()
                 .map_err(|_| anyhow::anyhow!("append gate poisoned"))?;
             self.check()?;
-            let next = observed
-                .checked_add(1)
-                .context("sync generation exhausted")?;
-            let mut marker = LogBatch::default();
-            marker.put(0, b"sync".to_vec(), next.to_le_bytes().to_vec())?;
-            // Engine::sync() is a no-op in the pinned version: its empty batch
-            // returns before sync. A nonempty marker is required here.
-            self.engine_write(&mut marker, true)?;
-            self.synced.store(next, Ordering::Release);
+            self.sync_locked()?;
         }
         self.check()
     }
 
+    // The caller holds sync_gate and the exclusive append_gate. A maintenance
+    // barrier also covers appends waiting for their original sync generation.
+    fn sync_locked(&self) -> Result<()> {
+        let next = self
+            .synced
+            .load(Ordering::Acquire)
+            .checked_add(1)
+            .context("sync generation exhausted")?;
+        let mut marker = LogBatch::default();
+        marker.put(0, b"sync".to_vec(), next.to_le_bytes().to_vec())?;
+        // Engine::sync() is a no-op in the pinned version: its empty batch
+        // returns before sync. A nonempty marker is required here.
+        self.engine_write(&mut marker, true)?;
+        self.synced.store(next, Ordering::Release);
+        Ok(())
+    }
+
     fn engine_write(&self, batch: &mut LogBatch, sync: bool) -> Result<()> {
         // The caller retains append_gate until failure is published. No other
         // group can complete a successful sync barrier after an uncertain write.
@@ -306,17 +322,59 @@ pub struct PhysicalLog {
     groups: BTreeMap<u64, GroupLog>,
 }
 
+/// Local physical-file tuning, independent of the replicated layout and policy.
+/// Thresholds trigger explicit maintenance; they are not disk-admission quotas.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub struct LogOptions {
+    pub target_file_bytes: u64,
+    /// Native file-count size estimate; oversized batches can exceed it.
+    pub purge_threshold_bytes: u64,
+}
+
+impl Default for LogOptions {
+    fn default() -> Self {
+        Self {
+            target_file_bytes: 64 * 1024 * 1024,
+            purge_threshold_bytes: 256 * 1024 * 1024,
+        }
+    }
+}
+
 impl PhysicalLog {
     pub fn open(directory: &Path, layout: &Layout, node: u64) -> Result<Self> {
-        Self::open_with_file_size(directory, layout, node, 64 * 1024 * 1024)
+        Self::open_with_options(directory, layout, node, LogOptions::default())
     }
 
+    #[cfg(test)]
     fn open_with_file_size(
         directory: &Path,
         layout: &Layout,
         node: u64,
         file_size: u64,
     ) -> Result<Self> {
+        Self::open_with_options(
+            directory,
+            layout,
+            node,
+            LogOptions {
+                target_file_bytes: file_size,
+                purge_threshold_bytes: file_size,
+            },
+        )
+    }
+
+    pub fn open_with_options(
+        directory: &Path,
+        layout: &Layout,
+        node: u64,
+        options: LogOptions,
+    ) -> Result<Self> {
+        ensure!(
+            options.target_file_bytes > 0
+                && options.purge_threshold_bytes >= options.target_file_bytes
+                && usize::try_from(options.purge_threshold_bytes).is_ok(),
+            "invalid physical log thresholds"
+        );
         layout.validate()?;
         ensure!(layout.nodes.contains_key(&node), "unknown local node");
         if !directory.exists() {
@@ -330,7 +388,9 @@ impl PhysicalLog {
         let config = raft_engine::Config {
             dir: directory.to_owned(),
             batch_compression_threshold: raft_engine::ReadableSize(0),
-            target_file_size: raft_engine::ReadableSize(file_size),
+            target_file_size: raft_engine::ReadableSize(options.target_file_bytes),
+            purge_threshold: raft_engine::ReadableSize(options.purge_threshold_bytes),
+            purge_rewrite_threshold: Some(raft_engine::ReadableSize(options.target_file_bytes)),
             enable_log_recycle: false,
             ..Default::default()
         };
@@ -346,6 +406,7 @@ impl PhysicalLog {
         let node = Arc::new(NodeEngine {
             engine,
             failed: AtomicBool::new(false),
+            maintenance_gate: RwLock::new(()),
             append_gate: RwLock::new(()),
             sync_gate: Mutex::new(()),
             synced: AtomicU64::new(0),
@@ -565,6 +626,14 @@ impl Storage for GroupLog {
         _: GetEntriesContext,
     ) -> raft::Result<Vec<Entry>> {
         let state = self.read().map_err(storage_error)?;
+        // Native maintenance can relocate entries and delete their old files.
+        // Keep its exclusive gate out until indexes and payloads are both read.
+        let _read = self
+            .node
+            .maintenance_gate
+            .read()
+            .map_err(|_| storage_error("maintenance gate poisoned"))?;
+        self.node.check().map_err(storage_error)?;
         if low <= state.boundary() {
             return Err(StorageError::Compacted.into());
         }

crates/lance-context-wal/src/raft_log/fault_tests.rs

@@ -16,21 +16,38 @@ type FileHandle = <DefaultFileSystem as FileSystem>::Handle;
 pub(super) struct FaultControl {
     fail_sync: AtomicBool,
     fail_once: AtomicBool,
+    fail_rewrite_sync: AtomicBool,
     sync_calls: AtomicU64,
     entered: Mutex<Option<mpsc::SyncSender<()>>>,
     release: Mutex<Option<mpsc::Receiver<()>>>,
 }
 
 impl FaultControl {
+    pub(super) fn fail_rewrite_sync_for_test(self: &Arc<Self>) -> RewriteSyncFault {
+        self.fail_rewrite_sync.store(true, Ordering::Release);
+        RewriteSyncFault(self.clone())
+    }
+
     #[cfg(feature = "replication")]
     pub(super) fn fail_sync_for_test(&self) {
         self.fail_once.store(true, Ordering::Release);
     }
 }
 
+pub(super) struct RewriteSyncFault(Arc<FaultControl>);
+
+impl Drop for RewriteSyncFault {
+    fn drop(&mut self) {
+        // Disarm before engine destructors, including if an assertion fails
+        // before a native rewrite reaches the intended injection point.
+        self.0.fail_rewrite_sync.store(false, Ordering::Release);
+    }
+}
+
 pub(super) struct FaultHandle {
     inner: Arc<FileHandle>,
     control: Arc<FaultControl>,
+    rewrite: bool,
 }
 
 impl Handle for FaultHandle {
@@ -46,6 +63,7 @@ impl Handle for FaultHandle {
         self.control.sync_calls.fetch_add(1, Ordering::Relaxed);
         if self.control.fail_once.swap(false, Ordering::AcqRel)
             || self.control.fail_sync.load(Ordering::Acquire)
+            || (self.rewrite && self.control.fail_rewrite_sync.swap(false, Ordering::AcqRel))
         {
             if let Some(entered) = self.control.entered.lock().unwrap().take() {
                 entered.send(()).unwrap();
@@ -140,10 +158,13 @@ pub(super) struct FaultFileSystem {
 }
 
 impl FaultFileSystem {
-    fn wrap(&self, inner: FileHandle) -> FaultHandle {
+    fn wrap(&self, inner: FileHandle, path: &Path) -> FaultHandle {
         FaultHandle {
             inner: Arc::new(inner),
             control: self.control.clone(),
+            rewrite: path
+                .extension()
+                .is_some_and(|extension| extension == "rewrite"),
         }
     }
 }
@@ -154,11 +175,14 @@ impl FileSystem for FaultFileSystem {
     type Writer = <DefaultFileSystem as FileSystem>::Writer;
 
     fn create<P: AsRef<Path>>(&self, path: P) -> io::Result<Self::Handle> {
-        Ok(self.wrap(DefaultFileSystem.create(path)?))
+        Ok(self.wrap(DefaultFileSystem.create(path.as_ref())?, path.as_ref()))
     }
 
     fn open<P: AsRef<Path>>(&self, path: P, permission: Permission) -> io::Result<Self::Handle> {
-        Ok(self.wrap(DefaultFileSystem.open(path, permission)?))
+        Ok(self.wrap(
+            DefaultFileSystem.open(path.as_ref(), permission)?,
+            path.as_ref(),
+        ))
     }
 
     fn delete<P: AsRef<Path>>(&self, path: P) -> io::Result<()> {
@@ -183,6 +207,63 @@ fn sync_panic_cannot_ack_another_group() -> Result<()> {
     sync_failure_cannot_ack_another_group(64 * 1024 * 1024)
 }
 
+#[test]
+fn retained_payload_read_does_not_wait_for_another_groups_sync() -> Result<()> {
+    let layout = Layout {
+        cluster: uuid::Uuid::from_u128(1201),
+        nodes: (1..=3)
+            .map(|id| (id, format!("wal-{id}.example.invalid:7800")))
+            .collect(),
+        groups: BTreeMap::from([(10, vec![1, 2, 3]), (20, vec![1, 2, 3])]),
+        buckets: vec![10, 20],
+    };
+    let directory = tempfile::tempdir()?;
+    let log = PhysicalLog::open(&directory.path().join("log"), &layout, 1)?;
+    let retained = Entry {
+        index: 1,
+        term: 1,
+        data: vec![42; 4096].into(),
+        ..Default::default()
+    };
+    let hard = HardState {
+        term: 1,
+        vote: 1,
+        commit: 1,
+        ..Default::default()
+    };
+    let sibling = log.group(20)?;
+    sibling.persist(std::slice::from_ref(&retained), Some(&hard))?;
+    let writer = log.group(10)?;
+    let control = writer.node.fault.clone();
+    let (entered, wait_entered) = mpsc::sync_channel(1);
+    let (release, wait_release) = mpsc::channel();
+    *control.entered.lock().unwrap() = Some(entered);
+    *control.release.lock().unwrap() = Some(wait_release);
+    control.fail_once.store(true, Ordering::Release);
+    let input = retained.clone();
+    let write = std::thread::spawn(move || writer.persist(&[input], Some(&hard)));
+    wait_entered.recv_timeout(Duration::from_secs(10))?;
+
+    // The real filesystem sync is held open. A read of previously durable
+    // sibling data must finish before that sync is released or fails.
+    let (completed, wait_completed) = mpsc::sync_channel(1);
+    let read = std::thread::spawn(move || {
+        completed
+            .send(sibling.entries(1, 2, None, GetEntriesContext::empty(false)))
+            .unwrap();
+    });
+    let observed = wait_completed.recv_timeout(Duration::from_secs(2));
+    release.send(())?;
+    assert!(write.join().unwrap().is_err());
+    read.join().unwrap();
+    assert_eq!(
+        observed.context("payload read waited on an unrelated sync")??,
+        vec![retained]
+    );
+    assert!(log.group(20)?.initial_state().is_err());
+    Ok(())
+}
+
 #[test]
 fn rotation_sync_panic_cannot_ack_another_group() -> Result<()> {
     // Rotation also calls Handle::sync, even for an engine write(sync=false).

crates/lance-context-wal/src/raft_log/maintenance.rs

@@ -0,0 +1,84 @@
+//! Physical segment maintenance after durable logical checkpointing.
+use super::*;
+
+/// A handle to one physical engine's maintenance boundary, shared by every
+/// hosted group. Run it on a bounded blocking worker, outside the Raft loop.
+/// It retains the native directory lock until dropped, including after all
+/// group owners have been dropped.
+pub struct LogMaintenance {
+    node: Arc<NodeEngine>,
+}
+
+impl LogMaintenance {
+    /// Rewrite retained entries and metadata, then delete obsolete native log
+    /// segments. Returned group IDs are checkpoint suggestions, not permission
+    /// to truncate those groups. This never advances a logical checkpoint or
+    /// deletes application snapshots or Lance objects.
+    ///
+    /// Pending appends and checkpoint metadata first cross a physical sync
+    /// barrier. Payload reads and writes are excluded for this maintenance pass;
+    /// an error or panic fences every group on the node. The native pass has no
+    /// per-call byte/time bound, so callers must measure its latency and reserve
+    /// rewrite headroom. This API alone does not impose a disk quota.
+    pub fn run(&self) -> Result<Vec<u64>> {
+        let result = self.run_locked();
+        if result.is_err() {
+            self.node.failed.store(true, Ordering::Release);
+        }
+        result
+    }
+
+    fn run_locked(&self) -> Result<Vec<u64>> {
+        let _maintenance = self
+            .node
+            .maintenance_gate
+            .write()
+            .map_err(|_| anyhow::anyhow!("maintenance gate poisoned"))?;
+        let _sync = self
+            .node
+            .sync_gate
+            .lock()
+            .map_err(|_| anyhow::anyhow!("sync gate poisoned"))?;
+        let _exclusive = self
+            .node
+            .append_gate
+            .write()
+            .map_err(|_| anyhow::anyhow!("append gate poisoned"))?;
+        self.node.check()?;
+        // The maintenance gate drained pending appends and checkpoint writes.
+        // Establish an explicit durable marker before the native rewrite too.
+        self.node.sync_locked()?;
+        let outcome = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            self.node.engine.purge_expired_files()
+        }));
+        let mut groups = match outcome {
+            Ok(Ok(groups)) => groups,
+            Ok(Err(error)) => {
+                self.node.failed.store(true, Ordering::Release);
+                return Err(error).context("uncertain log maintenance; reopen the node log");
+            }
+            Err(_) => {
+                self.node.failed.store(true, Ordering::Release);
+                anyhow::bail!("log maintenance panicked; reopen the node log");
+            }
+        };
+        groups.retain(|group| *group != 0);
+        groups.sort_unstable();
+        groups.dedup();
+        self.node.check()?;
+        Ok(groups)
+    }
+}
+
+impl GroupLog {
+    /// Return a node-wide physical-maintenance handle. One pass serves all
+    /// groups on this engine; scheduling a pass per group repeats the same work.
+    pub fn maintenance(&self) -> LogMaintenance {
+        LogMaintenance {
+            node: self.node.clone(),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests;