feat(wal): collect obsolete application snapshot files

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

@@ -7,13 +7,15 @@
 //!
 //! The owner schedules ticks, bounded network I/O and calls on one thread. This
 //! component supports explicit application checkpoints and prepared snapshot
-//! transfer. It retains bounded metadata and all image files: it is not the
-//! production service, a physical retention policy, or a bounded disk allocator.
+//! transfer and explicit collection of unreferenced application images. Raft
+//! segment reclamation, production scheduling and bounded disk admission remain
+//! separate responsibilities.
 //! Message scope checks are not authentication. Do not expose it to untrusted
 //! transport or enable ingestion without the remaining admission/lifecycle work.
 
 pub mod checkpoint;
 mod images;
+pub use images::{ImageCollection, ImageCollectionProgress};
 pub mod publication;
 pub mod query;
 pub mod source;
@@ -269,7 +271,14 @@ impl Node {
         );
         let policy = Sha256::digest(&identity).into();
         let log = PhysicalLog::open(directory, &layout, id)?;
-        let images = images::Owner::new(directory)?;
+        let hosted: std::collections::BTreeSet<_> = layout
+            .groups
+            .iter()
+            .filter(|(_, voters)| voters.contains(&id))
+            .map(|(group, _)| *group)
+            .collect();
+        let image_store = log.group(*hosted.iter().next().context("node has no hosted group")?)?;
+        let images = images::Owner::new(directory, hosted, image_store)?;
         let mut groups = BTreeMap::new();
         let logger = slog::Logger::root(slog::Discard, slog::o!());
         for (group, voters) in &layout.groups {
@@ -282,6 +291,7 @@ impl Node {
             let (mut state, image) = if store.first_index()? > 1 {
                 let snapshot = store.snapshot(store.first_index()? - 1, id)?;
                 let loaded = checkpoint::load(directory, &snapshot, &layout, &limits, *group)?;
+                images.installed(&loaded.image)?;
                 (loaded.state, Some(loaded.image))
             } else {
                 (State::default(), None)

crates/lance-context-wal/src/replication/checkpoint.rs

@@ -178,10 +178,11 @@ struct Location {
 }
 
 /// Open immutable files keep an outgoing transfer or reader attached to its
-/// captured image. The current increment does not delete any image files.
+/// captured image. Collection retains its files while any image owner is alive.
 pub struct Image {
     descriptor: Descriptor,
     files: [File; 2],
+    owner: std::sync::OnceLock<Arc<super::images::Owner>>,
     #[cfg(not(unix))]
     paths: [PathBuf; 2],
     locations: BTreeMap<u64, Location>,
@@ -195,6 +196,18 @@ impl std::fmt::Debug for Image {
 }
 
 impl Image {
+    pub(super) fn identity(&self) -> (u64, Uuid) {
+        (self.descriptor.binding.group, self.descriptor.id)
+    }
+
+    pub(super) fn keep_owner(&self, owner: &Arc<super::images::Owner>) -> Result<()> {
+        ensure!(
+            Arc::ptr_eq(self.owner.get_or_init(|| owner.clone()), owner),
+            "foreign application image owner"
+        );
+        Ok(())
+    }
+
     pub fn snapshot(&self) -> Result<Snapshot> {
         self.descriptor.snapshot()
     }
@@ -506,6 +519,7 @@ pub(super) fn build(
         image: Arc::new(Image {
             descriptor,
             files,
+            owner: std::sync::OnceLock::new(),
             locations,
             #[cfg(not(unix))]
             paths,
@@ -641,6 +655,7 @@ pub(super) fn load(
         image: Arc::new(Image {
             descriptor,
             files: [metadata_file, pack_file],
+            owner: std::sync::OnceLock::new(),
             locations,
             #[cfg(not(unix))]
             paths,

crates/lance-context-wal/src/replication/images.rs

@@ -2,21 +2,76 @@ use super::*;
 use checkpoint::{Cut, Image, Loaded, Part};
 use raft_proto::eraftpb::Snapshot;
 use std::{
+    collections::{BTreeMap, BTreeSet},
     path::PathBuf,
-    sync::atomic::{AtomicBool, Ordering},
+    sync::{
+        atomic::{AtomicBool, Ordering},
+        Mutex, Weak,
+    },
 };
 
+mod collection;
+pub use collection::{ImageCollection, ImageCollectionProgress};
+
 pub(super) struct Owner {
     directory: PathBuf,
     busy: AtomicBool,
+    groups: BTreeSet<u64>,
+    registry: Mutex<Registry>,
+    // Images keep this owner alive. Its engine lock excludes a new incarnation
+    // until readers and transfers from the old incarnation have released them.
+    store: GroupLog,
+}
+
+#[derive(Default)]
+struct Registry {
+    live: BTreeMap<(u64, Uuid), Weak<Image>>,
+    // A split group owner may be dropped while its siblings keep running.
+    // Its durable recovery image remains required even without a live Arc.
+    installed: BTreeMap<u64, (u64, Uuid)>,
 }
 impl Owner {
-    pub fn new(directory: &Path) -> Result<Arc<Self>> {
+    pub fn new(directory: &Path, groups: BTreeSet<u64>, store: GroupLog) -> Result<Arc<Self>> {
         Ok(Arc::new(Self {
             directory: directory.canonicalize()?,
             busy: AtomicBool::new(false),
+            groups,
+            registry: Mutex::new(Registry::default()),
+            store,
         }))
     }
+
+    pub fn track(self: &Arc<Self>, image: &Arc<Image>) -> Result<()> {
+        let key = image.identity();
+        ensure!(self.groups.contains(&key.0), "unhosted application image");
+        image.keep_owner(self)?;
+        let mut registry = self
+            .registry
+            .lock()
+            .map_err(|_| anyhow::anyhow!("image registry poisoned"))?;
+        let live = &mut registry.live;
+        live.retain(|_, image| image.strong_count() > 0);
+        if let Some(previous) = live.get(&key).and_then(Weak::upgrade) {
+            ensure!(
+                Arc::ptr_eq(&previous, image),
+                "duplicate live image identity"
+            );
+        } else {
+            live.insert(key, Arc::downgrade(image));
+        }
+        Ok(())
+    }
+
+    pub fn installed(self: &Arc<Self>, image: &Arc<Image>) -> Result<()> {
+        self.track(image)?;
+        let mut registry = self
+            .registry
+            .lock()
+            .map_err(|_| anyhow::anyhow!("image registry poisoned"))?;
+        let key = image.identity();
+        registry.installed.insert(key.0, key);
+        Ok(())
+    }
     fn acquire(self: &Arc<Self>) -> Result<Permit> {
         self.busy
             .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
@@ -64,6 +119,7 @@ impl CheckpointTask {
             state,
             |index, meta| read_body(&store, previous.as_deref(), index, meta),
         )?;
+        permit.0.track(&loaded.image)?;
         Ok(PreparedCheckpoint {
             permit,
             group,
@@ -100,10 +156,12 @@ impl SnapshotReceiver {
         self.receiver.write(part, offset, bytes)
     }
     pub fn finish(self) -> Result<PreparedCheckpoint> {
+        let loaded = self.receiver.finish()?;
+        self.permit.0.track(&loaded.image)?;
         Ok(PreparedCheckpoint {
             permit: self.permit,
             group: self.packet.group,
-            loaded: self.receiver.finish()?,
+            loaded,
             received: Some(self.packet),
         })
     }
@@ -184,7 +242,12 @@ impl Node {
             prepared.loaded.state.applied <= g.state.applied,
             "checkpoint exceeds applied state"
         );
-        if let Err(error) = g.raw.store().checkpoint(&snapshot) {
+        if let Err(error) = g
+            .raw
+            .store()
+            .checkpoint(&snapshot)
+            .and_then(|_| self.images.installed(&prepared.loaded.image))
+        {
             self.failed = true;
             return Err(error);
         }
@@ -315,6 +378,7 @@ pub(super) fn install_received(
         ready.entries(),
         ready.hs().context("received snapshot has no HardState")?,
     )?;
+    prepared.permit.0.installed(&prepared.loaded.image)?;
     g.state = prepared.loaded.state;
     g.image = Some(prepared.loaded.image);
     for (_, (request, _)) in std::mem::take(&mut g.pending) {

crates/lance-context-wal/src/replication/images/collection.rs

@@ -0,0 +1,137 @@
+//! Collection of application-image files in the exclusively owned node directory.
+use super::*;
+use std::{
+    ffi::OsStr,
+    fs::{self, File, ReadDir},
+};
+
+const MAX_SCAN_ENTRIES: usize = 4096;
+
+/// One bounded directory-scan batch. Removed bytes are file lengths, not a
+/// measurement of storage-device free space or physical bytes reclaimed.
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub struct ImageCollectionProgress {
+    pub scanned_entries: u64,
+    pub retained_files: u64,
+    pub removed_files: u64,
+    pub removed_bytes: u64,
+    pub finished: bool,
+}
+
+/// A resumable scan on a bounded blocking worker. Its node-wide image permit
+/// excludes builders, receivers and checkpoint installation, while ordinary
+/// append/poll work may continue. Drain or drop it before creating another image.
+/// A storage handle keeps the exclusive engine lock alive even if all Node
+/// owners are dropped; another incarnation cannot race this collector.
+pub struct ImageCollection {
+    permit: Permit,
+    entries: ReadDir,
+    protected: BTreeSet<(u64, Uuid)>,
+    finished: bool,
+}
+
+impl ImageCollection {
+    /// Inspect at most `max_entries` directory entries, including unrelated or
+    /// protected files. Only canonical, unreferenced image files of hosted groups
+    /// are removed. A successful deleting batch syncs the containing directory.
+    /// Retain this cursor for the next batch; restarting each scan can starve
+    /// later entries. No file payload is decoded, read or rewritten.
+    pub fn run_batch(&mut self, max_entries: usize) -> Result<ImageCollectionProgress> {
+        ensure!(
+            (1..=MAX_SCAN_ENTRIES).contains(&max_entries),
+            "invalid image collection batch size"
+        );
+        self.permit.0.store.initial_state()?;
+        let mut progress = ImageCollectionProgress::default();
+        for _ in 0..max_entries {
+            if self.finished {
+                break;
+            }
+            let Some(entry) = self.entries.next() else {
+                self.finished = true;
+                break;
+            };
+            let entry = entry?;
+            progress.scanned_entries += 1;
+            let Some(key) = image_key(&entry.file_name()) else {
+                continue;
+            };
+            if !self.permit.0.groups.contains(&key.0) {
+                continue;
+            }
+            if self.protected.contains(&key) {
+                progress.retained_files += 1;
+                continue;
+            }
+            // The node owns this directory. Keep unexpected symlinks, hard
+            // links, subdirectories and names outside its image namespace.
+            let metadata = fs::symlink_metadata(entry.path())?;
+            if !metadata.file_type().is_file() {
+                continue;
+            }
+            #[cfg(unix)]
+            {
+                use std::os::unix::fs::MetadataExt;
+                if metadata.nlink() != 1 {
+                    continue;
+                }
+            }
+            let bytes = progress
+                .removed_bytes
+                .checked_add(metadata.len())
+                .context("image collection byte counter overflow")?;
+            fs::remove_file(entry.path())?;
+            progress.removed_files += 1;
+            progress.removed_bytes = bytes;
+        }
+        if progress.removed_files > 0 {
+            File::open(&self.permit.0.directory)?.sync_all()?;
+        }
+        progress.finished = self.finished;
+        Ok(progress)
+    }
+}
+
+fn image_key(name: &OsStr) -> Option<(u64, Uuid)> {
+    let name = name.to_str()?.strip_prefix("context-application-")?;
+    let stem = name
+        .strip_suffix(".state")
+        .or_else(|| name.strip_suffix(".pack"))?;
+    let (group, id) = stem.split_once('-')?;
+    let number: u64 = group.parse().ok()?;
+    let uuid = Uuid::parse_str(id).ok()?;
+    (number > 0 && number.to_string() == group && !uuid.is_nil() && uuid.to_string() == id)
+        .then_some((number, uuid))
+}
+
+impl Node {
+    /// Collect obsolete and abandoned application-image files. Current installed
+    /// images and captured readers/transfers remain protected, including images
+    /// of sibling group owners returned by `into_groups`. This does not collect
+    /// Raft-engine segments or Lance objects and does not enforce a disk quota.
+    pub fn prepare_image_collection(&self) -> Result<ImageCollection> {
+        self.healthy()?;
+        let permit = self.images.acquire()?;
+        let protected = {
+            let mut registry = self
+                .images
+                .registry
+                .lock()
+                .map_err(|_| anyhow::anyhow!("image registry poisoned"))?;
+            registry.live.retain(|_, image| image.strong_count() > 0);
+            registry
+                .live
+                .keys()
+                .chain(registry.installed.values())
+                .copied()
+                .collect()
+        };
+        let entries = fs::read_dir(&self.images.directory)?;
+        Ok(ImageCollection {
+            permit,
+            entries,
+            protected,
+            finished: false,
+        })
+    }
+}

crates/lance-context-wal/src/replication/source/tests/closed.rs

@@ -232,11 +232,15 @@ fn empty_payload_images_preserve_latest_retry_and_captured_old_images() -> Resul
         assert_eq!(prepared.image().bytes(Part::Payload), 0);
         node.install_checkpoint(prepared)?;
     }
-    c.reopen()?;
+    // A captured image remains readable after its node is dropped, and keeps
+    // the directory locked until its last reader releases it.
+    c.nodes.clear();
     assert_eq!(
         old_image.chunk(Part::Payload, 4096, 4096)?,
         Bytes::from(vec![2; 4096])
     );
+    drop(old_image);
+    c.reopen()?;
     let retry = Batch::new(
         c.table,
         1,

crates/lance-context-wal/src/replication/tests.rs

@@ -1,6 +1,9 @@
 use super::*;
 use bytes::Bytes;
 
+#[path = "tests/image_collection.rs"]
+mod image_collection;
+
 pub(super) fn layout() -> Arc<Layout> {
     Arc::new(Layout {
         cluster: Uuid::from_u128(1),
@@ -176,6 +179,11 @@ fn application_images_keep_payloads_tails_suffix_and_lagging_peer_recovery() ->
     assert!(nodes[2].groups[&10].raw.store().first_index()? > 1);
     assert_eq!(nodes[2].groups[&10].state, nodes[0].groups[&10].state);
     drop(nodes);
+    assert!(
+        open().is_err(),
+        "captured transfer must retain the engine lock"
+    );
+    drop(old_image);
     let mut nodes = open()?;
     for node in &nodes {
         for ((index, _), sequence) in node.accepted(10, 0, 128)?.entries.into_iter().zip(1..=2) {