diff --git a/crates/lance-context-wal/README.md b/crates/lance-context-wal/README.md index 83ffd27..f99e6c1 100644 --- a/crates/lance-context-wal/README.md +++ b/crates/lance-context-wal/README.md @@ -15,8 +15,8 @@ here. A receipt identifies a position and is not authentication or proof of comm cargo test -p lance-context-wal --features placement --lib ``` -Compatibility tests preserve a layout fingerprint recorded by a real three-node -experiment and a fixed session-hash vector. Other tests reject ambiguous ownership +Compatibility tests preserve a synthetic layout fingerprint and a fixed +session-hash vector. Other tests reject ambiguous ownership and ensure routes survive JSON serialization and restart. The independent `admission` feature provides one proposal budget shared by all @@ -26,6 +26,13 @@ counters account for peer, control and rewrite traffic between directory scans. Refresh measures file lengths and allocated blocks, rather than estimating segment sizes. Successful admission does not walk the directory. +`Budget::limit_data` can lower the shared producer ceiling to leave more space +for checkpoint copies and log rewrites. It never raises the total budget or +cancels outstanding reservations. Maintenance callers use the total-budget +reservation path; using producer admission for checkpoints can prevent recovery +of space precisely when producers are throttled. The default 25% control window +alone does not guarantee enough temporary space for full-copy maintenance. + This is proposal admission, not filesystem space reservation or a limit on already accepted replication traffic. Callers must supply a monotonic write counter covering every group and protect file lifecycle changes with the shared diff --git a/crates/lance-context-wal/src/admission.rs b/crates/lance-context-wal/src/admission.rs index 62e7597..a1e45ec 100644 --- a/crates/lance-context-wal/src/admission.rs +++ b/crates/lance-context-wal/src/admission.rs @@ -29,7 +29,7 @@ struct Accounting { pub struct Budget { directory: PathBuf, written: Arc, - data_limit: u64, + data_limit: AtomicU64, max: u64, accounting: Mutex, filesystem: Arc>, @@ -58,7 +58,7 @@ impl Budget { written, filesystem, max: max as u64, - data_limit: (max - reserve) as u64, + data_limit: AtomicU64::new((max - reserve) as u64), accounting: Mutex::new(Accounting { measured, counter_at_measurement: counter, @@ -66,6 +66,20 @@ impl Budget { }), })) } + /// Tighten new-proposal admission to leave additional space for checkpoint + /// copies and engine rewrites. All groups observe the same ceiling. Existing + /// reservations remain charged; lowering the limit does not revoke them. + /// The total budget and already accepted replication traffic are unchanged. + pub fn limit_data(&self, bytes: usize) -> Result<()> { + let bytes = u64::try_from(bytes)?; + let _state = self.accounting.lock().unwrap(); + ensure!( + bytes > 0 && bytes <= self.data_limit.load(Ordering::Acquire), + "data limit must be positive and cannot raise the producer ceiling" + ); + self.data_limit.store(bytes, Ordering::Release); + Ok(()) + } fn accounted(&self, state: &Accounting) -> u64 { state.measured.saturating_add( self.written @@ -79,7 +93,11 @@ impl Budget { pub fn reserve(self: &Arc, bytes: usize, data: bool) -> Result { let bytes = u64::try_from(bytes)?; let mut state = self.accounting.lock().unwrap(); - let limit = if data { self.data_limit } else { self.max }; + let limit = if data { + self.data_limit.load(Ordering::Acquire) + } else { + self.max + }; ensure!( self.accounted(&state) .saturating_add(state.pending) @@ -124,7 +142,7 @@ impl Budget { measured_disk_bytes: state.measured, accounted_disk_bytes: self.accounted(&state), reserved_proposal_bytes: state.pending, - data_limit: self.data_limit, + data_limit: self.data_limit.load(Ordering::Acquire), max_log_bytes: self.max, } } @@ -169,6 +187,31 @@ fn retained_disk_bytes(directory: &Path) -> Result { mod tests { use super::*; #[test] + fn lower_ceiling_keeps_pending_proposals_and_maintenance_charged() -> Result<()> { + let directory = tempfile::tempdir()?; + let budget = Budget::new( + directory.path(), + Arc::new(AtomicU64::new(0)), + 8 * 1024 * 1024, + Arc::new(RwLock::new(())), + )?; + let data = budget.reserve(4 * 1024 * 1024, true)?; + budget.limit_data(2 * 1024 * 1024)?; + assert!(budget.limit_data(0).is_err()); + assert!(budget.limit_data(3 * 1024 * 1024).is_err()); + assert_eq!(budget.snapshot().data_limit, 2 * 1024 * 1024); + assert_eq!(budget.snapshot().reserved_proposal_bytes, 4 * 1024 * 1024); + assert!(budget.reserve(1, true).is_err()); + let maintenance = budget.reserve(4 * 1024 * 1024, false)?; + assert!(budget.reserve(1, false).is_err()); + drop(data); + // Maintenance still occupies the shared producer ceiling. + assert!(budget.reserve(1, true).is_err()); + drop(maintenance); + assert!(budget.reserve(2 * 1024 * 1024, true).is_ok()); + Ok(()) + } + #[test] fn racing_groups_share_admission_and_preserve_control_window() -> Result<()> { let directory = tempfile::tempdir()?; let written = Arc::new(AtomicU64::new(0)); diff --git a/crates/lance-context-wal/src/placement.rs b/crates/lance-context-wal/src/placement.rs index 5ea42bd..71815d2 100644 --- a/crates/lance-context-wal/src/placement.rs +++ b/crates/lance-context-wal/src/placement.rs @@ -139,24 +139,25 @@ pub struct Receipt { mod tests { use super::*; - fn measured_layout() -> Layout { + fn synthetic_layout() -> Layout { Layout { - cluster: Uuid::parse_str("b2da8f23-3a69-4538-9f86-2c10a61540cc").unwrap(), - nodes: (1..=3).map(|id| (id, format!( - "context-wal-lab-20261018-{}.context-wal-lab-20261018.souschef.svc.cluster.local:7800", id-1))).collect(), - groups: (1..=3).map(|id| (id, vec![1,2,3])).collect(), + cluster: Uuid::parse_str("00000000-0000-4000-8000-000000000001").unwrap(), + nodes: (1..=3) + .map(|id| (id, format!("wal-{}.example.invalid:7800", id - 1))) + .collect(), + groups: (1..=3).map(|id| (id, vec![1, 2, 3])).collect(), buckets: (0..1024).map(|bucket| 1 + bucket % 3).collect(), } } #[test] - fn layout_fingerprint_matches_recorded_multinode_experiment() -> Result<()> { - // Fixed compatibility vector from the real PHX protocol10 experiment. + fn layout_fingerprint_matches_synthetic_compatibility_vector() -> Result<()> { + // Synthetic compatibility vector, independently checked with OpenSSL. // Changing serialization or the hash domain must not reroute old data. - let layout = measured_layout(); + let layout = synthetic_layout(); let expected = [ - 32, 110, 106, 153, 114, 83, 142, 215, 88, 81, 79, 126, 93, 225, 129, 65, 94, 127, 167, - 140, 244, 218, 38, 47, 110, 21, 222, 171, 71, 7, 133, 7, + 44, 57, 66, 143, 150, 96, 112, 119, 182, 55, 27, 115, 233, 38, 124, 127, 206, 205, 105, + 78, 241, 190, 203, 129, 173, 136, 224, 91, 83, 23, 187, 33, ]; assert_eq!(layout.fingerprint()?, expected); let decoded: Layout = serde_json::from_slice(&serde_json::to_vec(&layout)?)?; @@ -172,19 +173,19 @@ mod tests { #[test] fn placement_rejects_ambiguous_or_incomplete_ownership() { - let mut layout = measured_layout(); + let mut layout = synthetic_layout(); layout.groups.insert(1, vec![1, 1, 3]); assert!(layout.validate().is_err()); - let mut layout = measured_layout(); + let mut layout = synthetic_layout(); layout.nodes.insert(2, layout.nodes[&1].clone()); assert!(layout.validate().is_err()); - let mut layout = measured_layout(); + let mut layout = synthetic_layout(); layout.buckets[0] = 99; assert!(layout.validate().is_err()); - let mut layout = measured_layout(); + let mut layout = synthetic_layout(); layout.nodes.insert(4, "unused:7800".into()); assert!(layout.validate().is_err()); - let mut layout = measured_layout(); + let mut layout = synthetic_layout(); layout.groups = (1..=9).map(|id| (id, vec![1, 2, 3])).collect(); layout.buckets = (0..16).map(|id| 1 + id % 9).collect(); assert!(layout.validate().is_err()); @@ -197,7 +198,7 @@ mod tests { // SHA-256 of the literal v1 prefix and three LE-u64 length=1 fields // a/b/c, independently checked with OpenSSL: c5f0901a5b7f96b9... . assert_eq!(bucket_for("a", "b", "c", 65536)?, 0xf0c5); - let layout = measured_layout(); + let layout = synthetic_layout(); let restored: Layout = serde_json::from_slice(&serde_json::to_vec(&layout)?)?; for session in ["session-1", "session-2", "训练会话", ""] { let bucket = layout.session_bucket("tenant", "namespace/table", session)?;