From 2b0443529075079c26d209220794f640fb78b2af Mon Sep 17 00:00:00 2001 From: Beinan Date: Wed, 21 Oct 2026 05:49:25 +0000 Subject: [PATCH] feat(wal): reclaim obsolete physical log segments --- crates/lance-context-wal/src/raft_log.rs | 95 ++++++++-- .../src/raft_log/fault_tests.rs | 87 ++++++++- .../src/raft_log/maintenance.rs | 84 +++++++++ .../src/raft_log/maintenance/tests.rs | 167 ++++++++++++++++++ crates/lance-context-wal/src/replication.rs | 36 +++- .../tests/replication_process.rs | 41 ++++- docs/design/wal-log-maintenance.md | 44 +++++ 7 files changed, 533 insertions(+), 21 deletions(-) create mode 100644 crates/lance-context-wal/src/raft_log/maintenance.rs create mode 100644 crates/lance-context-wal/src/raft_log/maintenance/tests.rs create mode 100644 docs/design/wal-log-maintenance.md diff --git a/crates/lance-context-wal/src/raft_log.rs b/crates/lance-context-wal/src/raft_log.rs index faf8e22..71822be 100644 --- a/crates/lance-context-wal/src/raft_log.rs +++ b/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, } +/// 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::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::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 { + 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> { 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()); } diff --git a/crates/lance-context-wal/src/raft_log/fault_tests.rs b/crates/lance-context-wal/src/raft_log/fault_tests.rs index 23d0c41..b700d30 100644 --- a/crates/lance-context-wal/src/raft_log/fault_tests.rs +++ b/crates/lance-context-wal/src/raft_log/fault_tests.rs @@ -16,21 +16,38 @@ type FileHandle = ::Handle; pub(super) struct FaultControl { fail_sync: AtomicBool, fail_once: AtomicBool, + fail_rewrite_sync: AtomicBool, sync_calls: AtomicU64, entered: Mutex>>, release: Mutex>>, } impl FaultControl { + pub(super) fn fail_rewrite_sync_for_test(self: &Arc) -> 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); + +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, control: Arc, + 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 = ::Writer; fn create>(&self, path: P) -> io::Result { - Ok(self.wrap(DefaultFileSystem.create(path)?)) + Ok(self.wrap(DefaultFileSystem.create(path.as_ref())?, path.as_ref())) } fn open>(&self, path: P, permission: Permission) -> io::Result { - Ok(self.wrap(DefaultFileSystem.open(path, permission)?)) + Ok(self.wrap( + DefaultFileSystem.open(path.as_ref(), permission)?, + path.as_ref(), + )) } fn delete>(&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). diff --git a/crates/lance-context-wal/src/raft_log/maintenance.rs b/crates/lance-context-wal/src/raft_log/maintenance.rs new file mode 100644 index 0000000..8dd1c6c --- /dev/null +++ b/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, +} + +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> { + let result = self.run_locked(); + if result.is_err() { + self.node.failed.store(true, Ordering::Release); + } + result + } + + fn run_locked(&self) -> Result> { + 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; diff --git a/crates/lance-context-wal/src/raft_log/maintenance/tests.rs b/crates/lance-context-wal/src/raft_log/maintenance/tests.rs new file mode 100644 index 0000000..f936318 --- /dev/null +++ b/crates/lance-context-wal/src/raft_log/maintenance/tests.rs @@ -0,0 +1,167 @@ +use super::*; +use std::{fs, path::PathBuf}; + +fn layout() -> Layout { + Layout { + cluster: uuid::Uuid::from_u128(1200), + 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], + } +} + +fn entry(group: u64, index: u64) -> Entry { + Entry { + index, + term: 1, + data: vec![(group ^ index) as u8; 16 * 1024].into(), + ..Default::default() + } +} + +fn append(group: &GroupLog, first: u64, last: u64) -> Result<()> { + group.persist( + &(first..=last) + .map(|index| entry(group.group, index)) + .collect::>(), + Some(&HardState { + term: 1, + vote: 1, + commit: last, + ..Default::default() + }), + ) +} + +fn checkpoint(group: &GroupLog, index: u64) -> Result { + let mut snapshot = Snapshot::default(); + snapshot.mut_metadata().index = index; + snapshot.mut_metadata().term = 1; + snapshot + .mut_metadata() + .set_conf_state(group.initial_state()?.conf_state); + // Storage-level tests use an opaque descriptor. End-to-end tests must first + // materialize and sync the actual application files referenced by a cut. + snapshot.data = format!("maintenance-test/{index}").into_bytes().into(); + group.checkpoint(&snapshot)?; + Ok(snapshot) +} + +fn files(directory: &Path) -> Result> { + let mut found = BTreeMap::new(); + for entry in fs::read_dir(directory)? { + let entry = entry?; + if entry + .path() + .extension() + .is_some_and(|extension| extension == "raftlog" || extension == "rewrite") + { + found.insert(entry.path(), entry.metadata()?.len()); + } + } + Ok(found) +} + +fn verify(group: &GroupLog, first: u64, last: u64) -> Result<()> { + assert_eq!(group.first_index()?, first); + assert_eq!(group.last_index()?, last); + assert_eq!(group.initial_state()?.hard_state.commit, last); + assert_eq!( + group.entries(first, last + 1, None, GetEntriesContext::empty(false))?, + (first..=last) + .map(|index| entry(group.group, index)) + .collect::>() + ); + Ok(()) +} + +#[test] +fn repeated_physical_collection_preserves_suffix_sibling_and_recovery() -> Result<()> { + let root = tempfile::tempdir()?; + let directory = root.path().join("log"); + let layout = layout(); + let log = PhysicalLog::open_with_file_size(&directory, &layout, 1, 4096)?; + let group = log.group(10)?; + let sibling = log.group(20)?; + group.bind_application(b"maintenance-test/v1")?; + append(&sibling, 1, 4)?; + let maintenance = group.maintenance(); + let unrelated = directory.join("context-application-unrelated.pack"); + fs::write(&unrelated, b"not a native engine segment")?; + let mut removed = BTreeSet::new(); + let mut snapshot = Snapshot::default(); + let mut maximum_retained_bytes = 0; + for round in 0..16 { + let last = (round + 1) * 32; + append(&group, round * 32 + 1, last)?; + snapshot = checkpoint(&group, last - 4)?; + let before = files(&directory)?; + for _ in 0..4 { + assert!(maintenance.run()?.iter().all(|id| [10, 20].contains(id))); + } + let after = files(&directory)?; + removed.extend( + before + .keys() + .filter(|file| !after.contains_key(*file)) + .cloned(), + ); + maximum_retained_bytes = maximum_retained_bytes.max(after.values().sum::()); + verify(&group, last - 3, last)?; + verify(&sibling, 1, 4)?; + assert_eq!(fs::read(&unrelated)?, b"not a native engine segment"); + } + assert!(!removed.is_empty(), "no physical file was reclaimed"); + // 8 MiB was appended across 16 rounds. Check actual retained file lengths, + // rather than equating a logical truncation count with physical collection. + assert!( + maximum_retained_bytes < 2 * 1024 * 1024, + "physical retention grew beyond the fixture bound: {maximum_retained_bytes}" + ); + drop((group, sibling, log)); + assert!(PhysicalLog::open(&directory, &layout, 1).is_err()); + drop(maintenance); + let reopened = PhysicalLog::open(&directory, &layout, 1)?; + let group = reopened.group(10)?; + group.bind_application(b"maintenance-test/v1")?; + assert_eq!(group.snapshot(508, 2)?, snapshot); + verify(&group, 509, 512)?; + verify(&reopened.group(20)?, 1, 4)?; + Ok(()) +} + +#[test] +fn rewrite_sync_failure_fences_all_groups_and_recovers_retained_records() -> Result<()> { + let root = tempfile::tempdir()?; + let directory = root.path().join("log"); + let layout = layout(); + let log = PhysicalLog::open_with_file_size(&directory, &layout, 1, 4096)?; + let group = log.group(10)?; + let sibling = log.group(20)?; + group.bind_application(b"maintenance-test/v1")?; + append(&sibling, 1, 4)?; + append(&group, 1, 32)?; + let snapshot = checkpoint(&group, 28)?; + let maintenance = group.maintenance(); + // Fail the rewrite queue's sync, after the append-queue marker barrier. + let fault = group.node.fault.fail_rewrite_sync_for_test(); + assert!(maintenance.run().is_err()); + assert!(group.initial_state().is_err()); + assert!(sibling.initial_state().is_err()); + assert!(sibling.persist(&[], None).is_err()); + assert!(sibling + .entries(1, 5, None, GetEntriesContext::empty(false)) + .is_err()); + assert!(maintenance.run().is_err()); + drop((fault, maintenance, group, sibling, log)); + let reopened = PhysicalLog::open(&directory, &layout, 1)?; + let group = reopened.group(10)?; + group.bind_application(b"maintenance-test/v1")?; + assert_eq!(group.snapshot(28, 2)?, snapshot); + verify(&group, 29, 32)?; + verify(&reopened.group(20)?, 1, 4)?; + append(&reopened.group(20)?, 5, 5)?; + Ok(()) +} diff --git a/crates/lance-context-wal/src/replication.rs b/crates/lance-context-wal/src/replication.rs index 3412f9d..9f62ed8 100644 --- a/crates/lance-context-wal/src/replication.rs +++ b/crates/lance-context-wal/src/replication.rs @@ -35,7 +35,7 @@ use uuid::Uuid; use crate::{ placement::Layout, - raft_log::{GroupLog, PhysicalLog}, + raft_log::{GroupLog, LogMaintenance, LogOptions, PhysicalLog}, record::{Batch, BatchMeta}, }; pub use images::{CheckpointTask, PreparedCheckpoint, SnapshotReceiver}; @@ -218,8 +218,9 @@ struct Group { } /// Owns all Raft groups in one shared physical engine. Do not call concurrently -/// or retain storage handles outside it. An internal/disk error fences this node; -/// drop it, reconcile outstanding requests and reopen the original directories. +/// or retain raw group storage handles outside it. An internal/disk error fences +/// this node; drop it and any maintenance handles, reconcile outstanding requests +/// and reopen the original directories. pub struct Node { signer: Option>, source_signers: BTreeMap>, @@ -263,6 +264,18 @@ impl Node { } pub fn open(directory: &Path, layout: Arc, id: u64, limits: Limits) -> Result { + Self::open_with_storage_options(directory, layout, id, limits, LogOptions::default()) + } + + /// Open with local log rotation and maintenance thresholds. These settings + /// do not change the replicated layout, application policy or disk quota. + pub fn open_with_storage_options( + directory: &Path, + layout: Arc, + id: u64, + limits: Limits, + storage: LogOptions, + ) -> Result { let layout_hash = layout.fingerprint()?; let identity = limits.identity()?; ensure!( @@ -270,7 +283,7 @@ impl Node { "source signing policy must cover exactly the configured groups" ); let policy = Sha256::digest(&identity).into(); - let log = PhysicalLog::open(directory, &layout, id)?; + let log = PhysicalLog::open_with_options(directory, &layout, id, storage)?; let hosted: std::collections::BTreeSet<_> = layout .groups .iter() @@ -359,6 +372,21 @@ impl Node { Ok(()) } + /// Prepare one physical-engine maintenance handle, shared by all groups on + /// this node. Run it on a bounded blocking worker. A native pass excludes + /// payload reads and persistence and can delay polling until it completes. + pub fn prepare_log_maintenance(&self) -> Result { + self.healthy()?; + Ok(self + .groups + .first_key_value() + .context("node has no hosted groups")? + .1 + .raw + .store() + .maintenance()) + } + /// Install the catalog's private dependency-attestation key. The persisted /// policy pins its public key. Source-only replicas need only the public key. pub fn set_catalog_signer(&mut self, signer: Arc) -> Result<()> { diff --git a/crates/lance-context-wal/tests/replication_process.rs b/crates/lance-context-wal/tests/replication_process.rs index b8aeeb3..9a918c7 100644 --- a/crates/lance-context-wal/tests/replication_process.rs +++ b/crates/lance-context-wal/tests/replication_process.rs @@ -94,6 +94,7 @@ enum Request { Page(u64), PrepareCheckpoint(u64, Uuid), InstallCheckpoint, + MaintainLog, ImageInfo(Vec), ImageChunk(Vec, bool, u64, usize), BeginSnapshot(WirePacket, Uuid), @@ -113,6 +114,7 @@ enum Response { Image(Vec, u64, u64), Chunk(Bytes), CollectedImages(u64, u64), + LogMaintained(Vec), Ok, } fn send(stream: &mut TcpStream, value: &T) -> Result<()> { @@ -202,6 +204,11 @@ fn run(worker: &mut Worker, request: Request) -> Result { } anyhow::bail!("test collection did not finish") } + Request::MaintainLog => { + return Ok(Response::LogMaintained( + node.prepare_log_maintenance()?.run()?, + )); + } Request::ImageInfo(snapshot) => { let image = worker .images @@ -283,13 +290,19 @@ fn replication_peer_fixture() -> Result<()> { } else { Limits::default() }; - let mut node = Node::open( + let mut node = Node::open_with_storage_options( &root.join(format!("data-{id}")), Arc::new(serde_json::from_slice(&std::fs::read( root.join("layout.json"), )?)?), id, limits.clone(), + // Exercise actual segment rotation and physical removal with the small + // process fixtures; these thresholds are not benchmark configuration. + lance_context_wal::raft_log::LogOptions { + target_file_bytes: 4096, + purge_threshold_bytes: 4096, + }, )?; let key_path = root.join(format!("catalog-key-{id}.pkcs8")); if key_path.exists() { @@ -721,6 +734,32 @@ fn application_snapshot_transfer_interruption_and_full_process_restart() -> Resu second_ack, Outcome::Committed(Applied::Accepted { .. }) )); + let directory = c.root.path().join(format!("data-{leader}")); + let mut old_segments = Vec::new(); + for entry in std::fs::read_dir(&directory)? { + let path = entry?.path(); + if path + .extension() + .is_some_and(|extension| extension == "raftlog") + { + old_segments.push(path); + } + } + for _ in 0..4 { + assert!(matches!( + c.peers + .get_mut(&leader) + .unwrap() + .call(Request::MaintainLog)?, + Response::LogMaintained(_) + )); + } + assert!( + old_segments.iter().any(|path| !path.exists()), + "native maintenance did not remove a physical segment" + ); + // The application image and suffix must still serve the lagging replica + // after physical removal, interrupted transfer and every process restarting. c.interrupt_next_snapshot = true; for _ in 0..30 { c.command(leader, Request::Tick)?; diff --git a/docs/design/wal-log-maintenance.md b/docs/design/wal-log-maintenance.md new file mode 100644 index 0000000..012938c --- /dev/null +++ b/docs/design/wal-log-maintenance.md @@ -0,0 +1,44 @@ +# Physical Raft log maintenance + +Tracking: [#378](https://github.com/lance-format/lance-context/issues/378). + +Logical checkpointing removes entries from the retained index, but old native +segments can still occupy disk. `GroupLog::maintenance` and +`Node::prepare_log_maintenance` return one physical-engine handle. Run its `run` +method on a bounded blocking worker, once per node rather than once per group. +It asks raft-engine to rewrite retained entries and metadata before deleting +obsolete native segments. Suggested checkpoint group IDs do not authorize +logical truncation; application snapshot validation and durability still decide +that boundary. Application images and Lance objects are outside this operation. + +A checkpoint may already have changed the native in-memory index while waiting +for its physical sync. Maintenance therefore drains writes through a separate +node-wide gate, acquires the shared sync gate, and persists a nonempty marker +before rewriting or deleting segments. The maintenance gate excludes payload +reads as well. Ordinary payload reads and ordinary sync barriers can still run +concurrently; readers do not take the append gate used to coalesce syncs. +The engine's empty `sync()` call does not provide that barrier. Native errors or +panics fence every group sharing the physical engine. Drop every storage and +maintenance handle before reopening after an uncertain outcome. + +`LogOptions` controls local rotation and rewrite thresholds without changing the +replicated policy. Defaults rotate near 64 MiB and make explicit maintenance +eligible above 256 MiB in the append queue, with a 64 MiB rewrite threshold. +The native trigger estimates size from file count times target size. Oversized +batches can make actual file lengths larger than that estimate. These are +triggers, not hard disk limits; capacity decisions need actual disk observations. +No maintenance runs automatically. + +The native pass has no per-call byte or time budget. It can delay reads and +persistence while holding the gate. A production scheduler must reserve rewrite +headroom, measure this delay and coordinate admission and maintenance. This +initial boundary does not establish bounded steady-state capacity or improve +measured throughput by itself. + +The storage tests use real files and repeated append/checkpoint/rewrite cycles, +check actual retained file lengths and removed paths, preserve an uncheckpointed +sibling group, and reopen to verify every retained entry. An injected rewrite +sync failure exercises the shared failure fence and recovery. The TCP process +fixture removes native segments before transferring an application snapshot to a +lagging replica, interrupts that transfer, then restarts all replicas and checks +the snapshot, suffix, sibling data and original retry receipt.