diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 2c1fb84..3412f9d 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/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) diff --git a/crates/lance-context-wal/src/replication/checkpoint.rs b/crates/lance-context-wal/src/replication/checkpoint.rs index a9715b9..bf7a32c 100644 --- a/crates/lance-context-wal/src/replication/checkpoint.rs +++ b/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>, #[cfg(not(unix))] paths: [PathBuf; 2], locations: BTreeMap, @@ -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) -> Result<()> { + ensure!( + Arc::ptr_eq(self.owner.get_or_init(|| owner.clone()), owner), + "foreign application image owner" + ); + Ok(()) + } + pub fn snapshot(&self) -> Result { 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, diff --git a/crates/lance-context-wal/src/replication/images.rs b/crates/lance-context-wal/src/replication/images.rs index 5caf350..f77a6fc 100644 --- a/crates/lance-context-wal/src/replication/images.rs +++ b/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, + registry: Mutex, + // 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>, + // 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, } impl Owner { - pub fn new(directory: &Path) -> Result> { + pub fn new(directory: &Path, groups: BTreeSet, store: GroupLog) -> Result> { Ok(Arc::new(Self { directory: directory.canonicalize()?, busy: AtomicBool::new(false), + groups, + registry: Mutex::new(Registry::default()), + store, })) } + + pub fn track(self: &Arc, image: &Arc) -> 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, image: &Arc) -> 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) -> Result { 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 { + 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) { diff --git a/crates/lance-context-wal/src/replication/images/collection.rs b/crates/lance-context-wal/src/replication/images/collection.rs new file mode 100644 index 0000000..d01b656 --- /dev/null +++ b/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 { + 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 { + 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, + }) + } +} diff --git a/crates/lance-context-wal/src/replication/source/tests/closed.rs b/crates/lance-context-wal/src/replication/source/tests/closed.rs index 798ee1a..0601abc 100644 --- a/crates/lance-context-wal/src/replication/source/tests/closed.rs +++ b/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, diff --git a/crates/lance-context-wal/src/replication/tests.rs b/crates/lance-context-wal/src/replication/tests.rs index 06f20c9..8d5ffe7 100644 --- a/crates/lance-context-wal/src/replication/tests.rs +++ b/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 { 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) { diff --git a/crates/lance-context-wal/src/replication/tests/image_collection.rs b/crates/lance-context-wal/src/replication/tests/image_collection.rs new file mode 100644 index 0000000..b234d68 --- /dev/null +++ b/crates/lance-context-wal/src/replication/tests/image_collection.rs @@ -0,0 +1,218 @@ +use super::*; +use std::fs; + +fn open(directory: &Path) -> Result> { + (1..=3) + .map(|id| { + Node::open( + &directory.join(id.to_string()), + layout(), + id, + Limits::default(), + ) + }) + .collect() +} + +fn append(nodes: &mut [Node], group: u64, sequence: u64, token: u64) -> Result<()> { + let mut input = batch(sequence); + input.meta.bucket = u32::from(group == 20); + nodes[0].propose(group, token, input)?; + assert!(pump(nodes)? + .iter() + .any(|c| c.request == token + && matches!(c.outcome, Outcome::Committed(Applied::Accepted { .. })))); + Ok(()) +} + +fn checkpoint(node: &mut Node, group: u64, id: u128) -> Result> { + let prepared = node.prepare_checkpoint(group, Uuid::from_u128(id))?.run()?; + let image = prepared.image(); + node.install_checkpoint(prepared)?; + Ok(image) +} + +fn path(directory: &Path, group: u64, id: u128, part: &str) -> std::path::PathBuf { + directory.join(format!( + "context-application-{group}-{}.{part}", + Uuid::from_u128(id) + )) +} + +fn drain(mut task: ImageCollection) -> Result { + assert!(task.run_batch(0).is_err()); + assert!(task.run_batch(4097).is_err()); + let mut total = ImageCollectionProgress::default(); + for _ in 0..512 { + let progress = task.run_batch(1)?; + assert!(progress.scanned_entries <= 1); + total.scanned_entries += progress.scanned_entries; + total.retained_files += progress.retained_files; + total.removed_files += progress.removed_files; + total.removed_bytes += progress.removed_bytes; + if progress.finished { + total.finished = true; + assert_eq!( + task.run_batch(1)?, + ImageCollectionProgress { + finished: true, + ..Default::default() + } + ); + return Ok(total); + } + } + anyhow::bail!("collection cursor made no bounded progress") +} + +#[test] +fn collection_preserves_live_images_sibling_groups_and_concurrent_appends() -> Result<()> { + let directory = tempfile::tempdir()?; + let root = directory.path().join("1"); + let mut nodes = open(directory.path())?; + nodes[0].campaign(10)?; + nodes[0].campaign(20)?; + pump(&mut nodes)?; + append(&mut nodes, 10, 1, 1)?; + append(&mut nodes, 20, 1, 2)?; + drop(checkpoint(&mut nodes[0], 20, 1000)?); + let captured = checkpoint(&mut nodes[0], 10, 1001)?; + append(&mut nodes, 10, 2, 3)?; + drop(checkpoint(&mut nodes[0], 10, 1002)?); + append(&mut nodes, 10, 3, 4)?; + let candidate = nodes[0] + .prepare_checkpoint(10, Uuid::from_u128(1003))? + .run()?; + let abandoned = candidate.image(); + assert!(nodes[0].prepare_image_collection().is_err()); + drop(candidate); + + for part in ["state", "pack"] { + fs::write(path(&root, 10, 1004, part), b"orphan")?; + } + let foreign = path(&root, 99, 1005, "pack"); + let unknown = root.join("context-application-unrecognized.pack"); + fs::write(&foreign, b"foreign")?; + fs::write(&unknown, b"unknown")?; + #[cfg(unix)] + { + std::os::unix::fs::symlink(&foreign, path(&root, 10, 1006, "pack"))?; + fs::hard_link(&foreign, path(&root, 10, 1007, "pack"))?; + } + fs::create_dir(path(&root, 10, 1008, "state"))?; + + let mut task = nodes[0].prepare_image_collection()?; + let (ready, wait_ready) = std::sync::mpsc::sync_channel(1); + let (resume, wait_resume) = std::sync::mpsc::sync_channel(1); + let worker = std::thread::spawn(move || -> Result { + // Inspect one entry, then keep the actual collector alive across writes. + let first = task.run_batch(1)?; + ready.send(())?; + wait_resume.recv_timeout(std::time::Duration::from_secs(30))?; + let mut rest = drain(task)?; + rest.scanned_entries += first.scanned_entries; + rest.retained_files += first.retained_files; + rest.removed_files += first.removed_files; + rest.removed_bytes += first.removed_bytes; + Ok(rest) + }); + wait_ready.recv_timeout(std::time::Duration::from_secs(30))?; + assert!(nodes[0].prepare_image_collection().is_err()); + assert!(nodes[0] + .prepare_checkpoint(20, Uuid::from_u128(1009)) + .is_err()); + append(&mut nodes, 10, 4, 5)?; + resume.send(())?; + let first = worker.join().unwrap()?; + assert_eq!(first.removed_files, 2); + assert_eq!(first.removed_bytes, 12); + assert_eq!(first.retained_files, 8); + assert_eq!( + captured.chunk(checkpoint::Part::Payload, 0, 4096)?, + batch(1).body + ); + for id in 1000..=1003 { + assert!(path(&root, if id == 1000 { 20 } else { 10 }, id, "state").exists()); + } + assert_eq!(fs::read(&foreign)?, b"foreign"); + assert_eq!(fs::read(&unknown)?, b"unknown"); + drop(captured); + drop(abandoned); + + // The collector of one group must protect the other split owner's image. + let mut workers = nodes.remove(0).into_groups(); + drop(workers.pop().unwrap()); + let second = drain(workers[0].prepare_image_collection()?)?; + assert_eq!(second.removed_files, 4); + assert_eq!(second.retained_files, 4); + assert!(path(&root, 20, 1000, "state").exists()); + assert!(path(&root, 10, 1002, "pack").exists()); + assert!(!path(&root, 10, 1001, "pack").exists()); + assert!(!path(&root, 10, 1003, "state").exists()); + let cursor = workers[0].prepare_image_collection()?; + drop(workers); + drop(nodes); + assert!(Node::open(&root, layout(), 1, Limits::default()).is_err()); + drop(cursor); + let recovered = Node::open(&root, layout(), 1, Limits::default())?; + for ((index, _), sequence) in recovered.accepted(10, 0, 128)?.entries.iter().zip(1..=4) { + assert_eq!(recovered.read_accepted(10, *index)?, Some(batch(sequence))); + } + assert_eq!(recovered.accepted(10, 0, 128)?.entries.len(), 4); + assert_eq!(recovered.accepted(20, 0, 128)?.entries.len(), 1); + Ok(()) +} + +#[test] +fn abandoned_receiver_files_are_collected_and_uncertain_sync_fences_collection() -> Result<()> { + let directory = tempfile::tempdir()?; + let mut nodes = open(directory.path())?; + nodes[0].campaign(10)?; + pump(&mut nodes)?; + append(&mut nodes, 10, 1, 1)?; + let image = checkpoint(&mut nodes[0], 10, 2000)?; + let mut message = Message { + from: 1, + to: 2, + term: nodes[0].status(10)?.term, + ..Default::default() + }; + message.set_msg_type(MessageType::MsgSnapshot); + message.set_snapshot(image.snapshot()?); + let packet = Packet { + cluster: nodes[0].layout.cluster, + layout: nodes[0].layout_hash, + policy: nodes[0].policy, + group: 10, + message, + }; + let mut receiver = nodes[1].receive_snapshot(packet, Uuid::from_u128(2001))?; + receiver.write( + checkpoint::Part::Metadata, + 0, + &image.chunk(checkpoint::Part::Metadata, 0, 8)?, + )?; + assert!(nodes[1].prepare_image_collection().is_err()); + drop(receiver); + assert_eq!( + drain(nodes[1].prepare_image_collection()?)?.removed_files, + 2 + ); + let accepted = nodes[1].accepted(10, 0, 128)?; + assert_eq!( + nodes[1].read_accepted(10, accepted.entries[0].0)?, + Some(batch(1)) + ); + + let root = directory.path().join("1"); + let orphan = path(&root, 10, 2002, "pack"); + fs::write(&orphan, b"retain after uncertain sync")?; + let mut collection = nodes[0].prepare_image_collection()?; + nodes[0].groups[&10].raw.store().fail_next_sync_for_test(); + nodes[0].propose(10, 2, batch(2))?; + assert!(nodes[0].poll().is_err()); + assert!(nodes[0].prepare_image_collection().is_err()); + assert!(collection.run_batch(4096).is_err()); + assert!(orphan.exists()); + Ok(()) +} diff --git a/crates/lance-context-wal/tests/raft_log_snapshot_install.rs b/crates/lance-context-wal/tests/raft_log_snapshot_install.rs index df0471f..d3713f1 100644 --- a/crates/lance-context-wal/tests/raft_log_snapshot_install.rs +++ b/crates/lance-context-wal/tests/raft_log_snapshot_install.rs @@ -11,6 +11,11 @@ use raft::{GetEntriesContext, Storage, StorageError}; use raft_proto::eraftpb::{Entry, HardState, Message, MessageType, Snapshot}; use uuid::Uuid; +// A concurrently spawning child briefly inherits other tests' flock descriptors +// before exec closes them. Keep these independent reopen cases out of that +// window; this does not serialize any concurrency exercised inside a case. +static TEST_CASE: std::sync::Mutex<()> = std::sync::Mutex::new(()); + fn layout() -> Layout { Layout { cluster: Uuid::from_u128(23), @@ -90,6 +95,7 @@ fn verify_installed(log: &PhysicalLog, snapshot: &Snapshot) -> Result<()> { #[test] fn received_ready_replaces_divergent_tail_and_recovers_committed_suffix() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); let directory = tempfile::tempdir()?; let path = directory.path().join("log"); let saved; @@ -177,6 +183,7 @@ fn received_ready_replaces_divergent_tail_and_recovers_committed_suffix() -> Res #[test] fn invalid_install_preserves_prior_log_and_vote() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); let directory = tempfile::tempdir()?; let log = seed(&directory.path().join("log"))?; let group = log.group(10)?; @@ -267,6 +274,7 @@ mod process { } #[test] fn install_survives_sigkill() -> Result<()> { + let _case = TEST_CASE.lock().unwrap(); use std::os::unix::process::ExitStatusExt; for phase in ["before", "after"] { let directory = tempfile::tempdir()?; @@ -308,4 +316,81 @@ mod process { } Ok(()) } + + #[test] + #[ignore = "subprocess fixture for the inherited directory lock regression"] + fn inherited_lock_child() {} + + #[test] + fn pre_exec_child_retains_an_unrelated_directory_lock_until_exec() -> Result<()> { + use std::{ + io::{Read, Write}, + os::{fd::AsRawFd, unix::net::UnixStream, unix::process::CommandExt}, + }; + + let _case = TEST_CASE.lock().unwrap(); + let directory = tempfile::tempdir()?; + let path = directory.path().join("log"); + let log = seed(&path)?; + let (mut parent, child) = UnixStream::pair()?; + parent.set_read_timeout(Some(Duration::from_secs(5)))?; + parent.set_write_timeout(Some(Duration::from_secs(5)))?; + let executable = std::env::current_exe()?; + let spawn = std::thread::spawn(move || -> Result<_> { + let fd = child.as_raw_fd(); + let mut command = Command::new(executable); + command + .args(["--ignored", "--exact", "process::inherited_lock_child"]) + .stdout(Stdio::null()) + .stderr(Stdio::null()); + // Only async-signal-safe syscalls and allocation-free error values + // run between fork and exec. The deadline also bounds a failed parent. + unsafe { + command.pre_exec(move || { + let mut byte = 1u8; + if libc::write(fd, (&byte as *const u8).cast(), 1) != 1 { + return Err(std::io::Error::last_os_error()); + } + let mut poll = libc::pollfd { + fd, + events: libc::POLLIN, + revents: 0, + }; + if libc::poll(&mut poll, 1, 10_000) != 1 { + return Err(std::io::Error::from_raw_os_error(libc::ETIMEDOUT)); + } + if libc::read(fd, (&mut byte as *mut u8).cast(), 1) != 1 { + return Err(std::io::Error::from_raw_os_error(libc::EPIPE)); + } + Ok(()) + }); + } + let status = command.spawn()?.wait()?; + drop(child); + Ok(status) + }); + let ready = parent.read_exact(&mut [0]); + drop(log); + let blocked = ready + .as_ref() + .ok() + .map(|_| PhysicalLog::open(&path, &layout(), 2)); + // Always release and join the bounded child before propagating any error. + let released = parent.write_all(&[1]); + let child = spawn + .join() + .map_err(|_| anyhow::anyhow!("spawn thread panicked"))?; + ready?; + released?; + assert!(child?.success()); + let error = blocked + .expect("child ready") + .err() + .context("child must retain the inherited flock until exec")?; + assert!(format!("{error:#}").contains("Failed to lock file")); + let reopened = PhysicalLog::open(&path, &layout(), 2)?; + assert_eq!(read(&reopened.group(10)?, 1, 8)?, original()); + assert_eq!(read(&reopened.group(20)?, 1, 2)?, vec![entry(1, 7)]); + Ok(()) + } } diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index e4d9dd2..b8aeeb3 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -100,6 +100,7 @@ enum Request { ReceiveChunk(bool, u64, Bytes), FinishSnapshot, ReportSnapshot(u64, u64, bool), + CollectImages, } #[derive(Serialize, Deserialize)] enum Response { @@ -111,6 +112,7 @@ enum Response { Error(String), Image(Vec, u64, u64), Chunk(Bytes), + CollectedImages(u64, u64), Ok, } fn send(stream: &mut TcpStream, value: &T) -> Result<()> { @@ -186,6 +188,20 @@ fn run(worker: &mut Worker, request: Request) -> Result { Request::InstallCheckpoint => { node.install_checkpoint(worker.prepared.take().context("no prepared image")?)? } + Request::CollectImages => { + let mut task = node.prepare_image_collection()?; + let mut files = 0; + let mut bytes = 0; + for _ in 0..4096 { + let batch = task.run_batch(1)?; + files += batch.removed_files; + bytes += batch.removed_bytes; + if batch.finished { + return Ok(Response::CollectedImages(files, bytes)); + } + } + anyhow::bail!("test collection did not finish") + } Request::ImageInfo(snapshot) => { let image = worker .images @@ -680,6 +696,9 @@ fn application_snapshot_transfer_interruption_and_full_process_restart() -> Resu // Kill after complete file sync but BEFORE the Raft descriptor transition. drop(c.peers.remove(&1)); c.start(1)?; + assert!( + matches!(c.peers.get_mut(&1).unwrap().call(Request::CollectImages)?, Response::CollectedImages(2, bytes) if bytes > 0) + ); verify_records( c.peers.get_mut(&1).unwrap(), 10, @@ -715,7 +734,27 @@ fn application_snapshot_transfer_interruption_and_full_process_restart() -> Resu "partial snapshot kill was not reached" ); assert_eq!(c.transfers, 1); + // The first chunk can still be buffered when SIGKILL arrives. Check the + // actual orphan lengths, including zero-byte files, rather than assuming + // received bytes were flushed to storage. + let abandoned: Vec<_> = ["state", "pack"] + .into_iter() + .map(|part| { + c.root.path().join("data-3").join(format!( + "context-application-10-{}.{part}", + Uuid::from_u128(10_001) + )) + }) + .collect(); + let mut abandoned_bytes = 0; + for path in &abandoned { + abandoned_bytes += std::fs::metadata(path)?.len(); + } c.start(3)?; + assert!( + matches!(c.peers.get_mut(&3).unwrap().call(Request::CollectImages)?, Response::CollectedImages(2, bytes) if bytes == abandoned_bytes) + ); + assert!(abandoned.iter().all(|path| !path.exists())); verify_records(c.peers.get_mut(&3).unwrap(), 10, &[])?; verify_records( c.peers.get_mut(&3).unwrap(), @@ -745,6 +784,10 @@ fn application_snapshot_transfer_interruption_and_full_process_restart() -> Resu c.start(id)?; } for peer in c.peers.values_mut() { + assert!(matches!( + peer.call(Request::CollectImages)?, + Response::CollectedImages(0, 0) + )); verify_records(peer, 10, &[first.clone(), second.clone()])?; verify_records(peer, 20, std::slice::from_ref(&sibling))?; } diff --git a/crates/lance-context-wal/tests/replication_process/source_closure.rs b/crates/lance-context-wal/tests/replication_process/source_closure.rs index ae4b3d4..01e29e2 100644 --- a/crates/lance-context-wal/tests/replication_process/source_closure.rs +++ b/crates/lance-context-wal/tests/replication_process/source_closure.rs @@ -41,6 +41,18 @@ pub(super) fn close_and_restart( let mut c = cluster.lock().unwrap(); let catalog = c.elect(1, 10, &[1, 2, 3])?; let source = c.elect(4, 20, &[4, 5, 6])?; + let old_image_source = source; + let Response::Image(_, _, old_payload_bytes) = c + .peers + .get_mut(&source) + .unwrap() + .call(Request::PrepareCheckpoint(20, Uuid::new_v4()))? + else { + anyhow::bail!("source input image missing") + }; + assert!(old_payload_bytes > 0); + c.command(source, Request::InstallCheckpoint)?; + c.pump(&[4, 5, 6])?; for id in 4..=6 { assert!(!c .root @@ -118,6 +130,17 @@ pub(super) fn close_and_restart( for id in 1..=6 { c.start(id)?; } + // Old source packs no longer have a live reader after all processes die. + // The installed retirement image still protects its metadata and empty pack. + let Response::CollectedImages(files, bytes) = c + .peers + .get_mut(&old_image_source) + .unwrap() + .call(Request::CollectImages)? + else { + anyhow::bail!("source image collection missing") + }; + assert!(files >= 2 && bytes >= old_payload_bytes); let source = c.elect(4, 20, &[4, 5, 6])?; let recovered_proof = proof(&mut c, source, 600_011, table)?.context("proof lost after all peers restarted")?; diff --git a/docs/design/wal-image-collection.md b/docs/design/wal-image-collection.md new file mode 100644 index 0000000..42bb3e7 --- /dev/null +++ b/docs/design/wal-image-collection.md @@ -0,0 +1,39 @@ +# Collection of obsolete application snapshots + +Tracking: [#378](https://github.com/lance-format/lance-context/issues/378). + +After checkpoint installation, old application `.state` and `.pack` files can +outlive their recovery purpose. `Node::prepare_image_collection` returns a +cursor whose `run_batch` scans at most the requested number of directory entries +(1–4096). It removes only canonical image files for groups hosted by that node. +Unrecognized files, other groups, directories, symlinks and hard links remain. +The caller runs batches on a bounded blocking worker, retains the cursor between +batches, and drains or drops it before preparing another image. + +The node shares one image registry and worker permit across split group owners. +The registry protects every durably installed recovery image, including one +whose group owner has been dropped while sibling groups continue. Weak entries +also protect any older image retained by a reader, outgoing transfer or abandoned +prepared-image handle. Builders, receivers and checkpoint installation exclude +collection through the same permit. Ordinary Raft append and poll work continues. + +Captured images and collection cursors retain the native engine lock. Drop all +readers, transfers and cursors before reopening a failed node directory; a new +incarnation cannot delete files still used by the old one. After a process crash, +reopening validates and registers each installed snapshot before collection is +available. Complete uninstalled candidates and interrupted receive files are +then eligible for collection. An uncertain engine write fences collection too. + +Every successful deleting batch syncs the directory. Its byte count is the sum +of removed file lengths, not device free space or reclaimed physical blocks. +No payload is scanned or rewritten. This operation does not collect Raft-engine +segments, Lance manifests or data files; those require their own dependency and +retention rules. Automatic scheduling and a complete disk quota remain separate +integration work. + +Tests exercise bounded scanning with actual files, ordinary writes while a +collector is retained on another thread, captured-reader retention, dropped +sibling owners, engine-lock lifetime and uncertain-sync fencing. TCP child-process +tests cover uninstalled and partially received snapshots after SIGKILL, and old +source-pack removal after real Lance publication and all-replica restart. The +managed Lance reader then verifies the manifest, transaction and full payloads.