feat: reserve configurable WAL maintenance headroom

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

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

@@ -29,7 +29,7 @@ struct Accounting {
 pub struct Budget {
     directory: PathBuf,
     written: Arc<AtomicU64>,
-    data_limit: u64,
+    data_limit: AtomicU64,
     max: u64,
     accounting: Mutex<Accounting>,
     filesystem: Arc<RwLock<()>>,
@@ -58,14 +58,28 @@ 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,
                 pending: 0,
             }),
         }))
     }
+    /// 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<Self>, bytes: usize, data: bool) -> Result<Reservation> {
         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<u64> {
 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));

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)?;