diff --git a/crates/uffs-client/src/connect.rs b/crates/uffs-client/src/connect.rs index 639cfae6c..1fa75eb64 100644 --- a/crates/uffs-client/src/connect.rs +++ b/crates/uffs-client/src/connect.rs @@ -269,6 +269,34 @@ impl UffsClient { Ok(response) } + /// Ask the daemon which drives the given search would promote + /// (the dispatch promote-plan, bloom pre-check included) — + /// `warm_plan` RPC, daemon v0.6.38+. + /// + /// Send the SAME params the search itself will carry; the daemon + /// resolves them through its own filter pipeline, so the answer is + /// bit-exact with what dispatching that search would promote. An + /// empty `needs_promote` means the query serves without any + /// re-warm. Older daemons answer with a method-not-found + /// [`crate::error::ClientError::DaemonError`] + /// (code [`crate::protocol::ERR_METHOD_NOT_FOUND`]); callers + /// gate-keeping on this should fall back to a tier-based check. + /// + /// # Errors + /// + /// Returns a `ClientError` on connection, protocol, or timeout + /// failure, and the method-not-found daemon error described above. + pub async fn warm_plan( + &mut self, + params: &SearchParams, + ) -> Result { + let value = serde_json::to_value(params) + .map_err(|err| crate::error::ClientError::Protocol(err.to_string()))?; + let result = self.send_request("warm_plan", Some(value)).await?; + serde_json::from_value(result) + .map_err(|err| crate::error::ClientError::Protocol(err.to_string())) + } + /// List loaded drives. /// /// # Errors diff --git a/crates/uffs-client/src/protocol/response.rs b/crates/uffs-client/src/protocol/response.rs index af69d5726..be0c6f652 100644 --- a/crates/uffs-client/src/protocol/response.rs +++ b/crates/uffs-client/src/protocol/response.rs @@ -14,7 +14,7 @@ use serde::{Deserialize, Serialize}; pub use super::response_journal::{ChangedSinceParams, ChangedSinceResponse, JournalChange}; pub use super::response_status::{ DaemonPaths, DaemonStatus, DriveInfo, DriveMemoryInfo, DrivesResponse, LiveUpdateInfo, - ShardTier, StatsResponse, StatusResponse, + ShardTier, StatsResponse, StatusResponse, WarmPlanResponse, }; pub use super::response_tiering::{ DEFAULT_PRELOAD_PIN_MINUTES, DriveTierStatus, ForgetParams, ForgetResponse, HibernateParams, diff --git a/crates/uffs-client/src/protocol/response_status.rs b/crates/uffs-client/src/protocol/response_status.rs index 72211c388..08d19ef8f 100644 --- a/crates/uffs-client/src/protocol/response_status.rs +++ b/crates/uffs-client/src/protocol/response_status.rs @@ -25,6 +25,28 @@ pub struct DrivesResponse { pub drives: Vec, } +/// Response for the `warm_plan` method (v0.6.38+). +/// +/// Carries the drives the daemon's dispatch would actually promote for +/// the given search params — Parked/Cold shards in scope whose bloom +/// cannot prove them irrelevant to the query's resolved extension +/// filter. +/// +/// This is the daemon-authoritative answer external cold-index gates +/// must consume instead of inferring readiness from `DriveInfo::tier` +/// alone: a Parked drive whose bloom proves "no matching extensions" +/// serves the query instantly with zero rows and MUST NOT be gated or +/// force-warmed (field 2026-08-24: the MCP gate's tier-only view +/// reported such a drive "warming — retry" and paged its body back +/// into RAM for a query the bloom had already answered). +#[derive(Debug, Serialize, Deserialize)] +pub struct WarmPlanResponse { + /// Drives the search would promote, in registry order. Empty + /// means the query dispatches without any re-warm — the caller's + /// gate must pass. + pub needs_promote: Vec, +} + /// Memory-tiering state of a single shard, as surfaced over the wire. /// /// Mirrors the daemon-internal `ShardState` enum (`crates/uffs-daemon/ diff --git a/crates/uffs-core/src/compact.rs b/crates/uffs-core/src/compact.rs index a31fa6652..9d08efcc6 100644 --- a/crates/uffs-core/src/compact.rs +++ b/crates/uffs-core/src/compact.rs @@ -267,9 +267,33 @@ impl DriveCompactIndex { /// mutation — then resets `delta = None` so subsequent searches take the /// zero-overhead base fast path. /// + /// Fold any pending delta overlay into fresh bases so this index + /// is safe to serialize — the boundary every save path must cross. + /// + /// A delta-carrying index is UNSERIALIZABLE as-is: the patch path + /// appends created records (the header's record count grows) while + /// `children` / `trigram` / `ext_index` stay the Arc-shared bases, + /// so the writer emits a children CSR sized for the OLD record + /// count and the reader — which sizes the section from the header + /// — mis-walks every later section (field signature: `bloom + /// k_hashes out of range`, quarantining an otherwise-healthy cache + /// on the next start; winbox M/C/D/S 2026-08-24, and the 2026-08 + /// "drive C for ten days" incident). Even an aligned save would + /// persist a trigram base missing the new records — a silent + /// substring-search hole after reload. No-op when `delta` is + /// already `None`, so calling it unconditionally before a save is + /// free in the steady state. + pub fn fold_delta_for_save(&mut self) { + if self.delta.is_some() { + self.compact_base(); + } + } + /// O(total records); the per-apply path drives toward this running only /// occasionally (every [`TRIGRAM_COMPACT_THRESHOLD`] touched records) or - /// before serialization (the on-disk cache is always delta-free). + /// before serialization (the on-disk cache is always delta-free — + /// enforced by [`Self::fold_delta_for_save`] + + /// `save_compact_cache_background`'s refusal guard). pub(crate) fn compact_base(&mut self) { self.trigram = Arc::new(TrigramIndex::build(&self.records, &self.names, self.fold)); self.ext_index = Arc::new(ExtensionIndex::build(&self.records)); diff --git a/crates/uffs-core/src/compact_cache.rs b/crates/uffs-core/src/compact_cache.rs index 9bac52e40..ca0e6ef30 100644 --- a/crates/uffs-core/src/compact_cache.rs +++ b/crates/uffs-core/src/compact_cache.rs @@ -707,20 +707,39 @@ fn parse_compact_body( return Err("truncated trigram header"); } let tkc = read_u32(data, pe) as usize; - let (trigram_loaded, after_tri) = if version >= 6 && tkc > 0 { + let (trigram_loaded, after_tri) = if version >= 6 { + // The v6+ writer ALWAYS emits the full trigram CSR — keys, + // offsets, and the values count — even for an empty index: + // `TrigramIndex` maintains the CSR invariant `offsets.len() + // == keys.len() + 1` in memory, so an empty index still + // serialises its `[0]` offsets entry plus `vc = 0`. The old + // `tkc > 0` fast-path consumed only the 4-byte key count in + // the empty case, leaving those 8 bytes unread and shifting + // every later section (ext names, bloom, trie, frs) — the + // `bloom k_hashes out of range` corruption class that + // quarantined otherwise-valid caches (2026-08 "drive C for + // ten days"; 2026-08-24 winbox C/D/S quarantine storm). let (ks, ke, oe) = (pe + 4, pe + 4 + tkc * 8, pe + 4 + tkc * 8 + (tkc + 1) * 4); if data.len() < oe + 4 { return Err("truncated trigram CSR"); } - let tk: Vec = aligned_vec_from_bytes(data.get(ks..ke).ok_or("trigram keys")?); - let to: Vec = aligned_vec_from_bytes(data.get(ke..oe).ok_or("trigram offsets")?); let vc = read_u32(data, oe) as usize; let ve = oe + 4 + vc * 4; if data.len() < ve { return Err("truncated trigram values"); } - let tv: Vec = aligned_vec_from_bytes(data.get(oe + 4..ve).ok_or("trigram values")?); - (Some(TrigramIndex::from_csr(tk, to, tv)), ve) + if tkc > 0 { + let tk: Vec = aligned_vec_from_bytes(data.get(ks..ke).ok_or("trigram keys")?); + let to: Vec = aligned_vec_from_bytes(data.get(ke..oe).ok_or("trigram offsets")?); + let tv: Vec = + aligned_vec_from_bytes(data.get(oe + 4..ve).ok_or("trigram values")?); + (Some(TrigramIndex::from_csr(tk, to, tv)), ve) + } else { + // Empty on disk: report `None` so the assembler rebuilds + // from records — adopting the empty index verbatim would + // silently break substring search on a non-empty drive. + (None, ve) + } } else { (None, pe + 4) }; @@ -1018,6 +1037,21 @@ pub fn save_compact_cache(index: &DriveCompactIndex) -> io::Result<()> { /// # Errors /// Returns an error only if compression or directory creation fails. pub fn save_compact_cache_background(index: &DriveCompactIndex) -> io::Result<()> { + // Boundary enforcement of the "on-disk cache is always delta-free" + // invariant: a delta-carrying index serialises a children CSR sized + // for the pre-patch record count while the header carries the + // post-patch count, mis-aligning every later section for the reader + // (see `DriveCompactIndex::fold_delta_for_save` for the full + // mechanism + field incidents). Callers fold the delta first; this + // guard turns any future path that forgets into a loud error + // instead of a poisoned cache that quarantines on the next start. + if index.delta.is_some() { + return Err(io::Error::other( + "refusing to persist a delta-carrying compact index: fold the delta first \ + (DriveCompactIndex::fold_delta_for_save) — its base sections are stale \ + against the patched records and would serialize misaligned", + )); + } let profile = std::env::var_os("UFFS_CACHE_PROFILE").is_some(); let t_compress = Instant::now(); diff --git a/crates/uffs-core/src/compact_cache/parked.rs b/crates/uffs-core/src/compact_cache/parked.rs index ba925793c..cafa4f8dc 100644 --- a/crates/uffs-core/src/compact_cache/parked.rs +++ b/crates/uffs-core/src/compact_cache/parked.rs @@ -200,9 +200,17 @@ fn find_v9_filters_offset(data: &[u8]) -> Result<(usize, u64), &'static str> { return Err("truncated trigram header"); } - // Trigram CSR (always present for v >= 6). + // Trigram CSR (always present for v >= 6, INCLUDING when empty: + // the writer serialises the in-memory CSR invariant `offsets.len() + // == keys.len() + 1`, so an empty index still emits its `[0]` + // offsets entry plus a zero values count). The old `key_count > + // 0` fast-path treated the count word as a bare "no trigram" + // sentinel — a shape no writer ever produced — and left 8 bytes + // unconsumed, shifting the ext-names/bloom/trie reads that follow + // (the `bloom k_hashes out of range` corruption class; see + // `parse_compact_body` for the full incident note). let trigram_key_count = read_u32(data, postings_end) as usize; - let after_trigram = if trigram_key_count > 0 { + let after_trigram = { let keys_end = postings_end .checked_add(4) .and_then(|x| x.checked_add(trigram_key_count.checked_mul(8)?)) @@ -222,15 +230,6 @@ fn find_v9_filters_offset(data: &[u8]) -> Result<(usize, u64), &'static str> { return Err("truncated trigram values"); } values_end - } else { - // v >= 6 with `trigram_key_count == 0` is the legacy "no - // trigram on disk" sentinel; v9 always emits a real - // trigram, but a malformed v9 buffer could land here. - // The 4-byte sentinel header has been consumed; advance - // past it so the next section follows correctly. - postings_end - .checked_add(4) - .ok_or("after-trigram-sentinel overflow")? }; // Skip the v7+ ext_names table without allocating. v9 always diff --git a/crates/uffs-core/src/compact_cache/tests.rs b/crates/uffs-core/src/compact_cache/tests.rs index dfade0131..27bcb33cd 100644 --- a/crates/uffs-core/src/compact_cache/tests.rs +++ b/crates/uffs-core/src/compact_cache/tests.rs @@ -701,3 +701,11 @@ fn load_compact_cache_at_quarantines_nothing_itself() { } assert!(path.exists(), "inner loader must not touch the file"); } + +// The cache-poisoning regression cluster (empty-trigram shape, the +// delta-carrying misalignment, and the save-boundary refusal) lives in +// a nested sibling file so this module stays under the workspace +// 800-LOC ceiling. `#[path]` keeps it a child of `tests`, so it can +// use this module's `make_test_index` fixture directly. +#[path = "tests_cache_poison.rs"] +mod cache_poison; diff --git a/crates/uffs-core/src/compact_cache/tests_cache_poison.rs b/crates/uffs-core/src/compact_cache/tests_cache_poison.rs new file mode 100644 index 000000000..e21a93d30 --- /dev/null +++ b/crates/uffs-core/src/compact_cache/tests_cache_poison.rs @@ -0,0 +1,173 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! Cache-poisoning regression cluster: shapes a serialized compact +//! cache must survive (empty trigram CSR) or must never assume +//! (delta-carrying base sections), plus the save-boundary refusal. +//! Split from [`super`] (the `tests` module) for the 800-LOC policy; +//! shares its [`super::make_test_index`] fixture. + +use super::super::*; +use super::make_test_index; + +/// Regression: a cache saved with an EMPTY trigram index must still +/// round-trip every later section intact. +/// +/// The writer always emits the full trigram CSR — offsets carries +/// `keys.len() + 1` entries, so an empty index still writes its `[0]` +/// offsets entry plus the values count. The reader's old `tkc > 0` +/// fast-path skipped only the 4-byte key count in the empty case, +/// leaving 8 bytes unconsumed and shifting the ext-names / bloom / +/// trie / frs sections — surfacing as `bloom k_hashes out of range` +/// and a quarantined cache (2026-08 "drive C for ten days" incident; +/// 2026-08-24 winbox C/D/S quarantine storm). +#[test] +fn empty_trigram_round_trips_without_shifting_later_sections() { + let mut index = make_test_index(); + index.trigram = Arc::new(TrigramIndex::empty()); + // Give the later sections real content so a shift is detectable. + index.ext_names = vec![Box::from("rs"), Box::from("toml")]; + index.bloom = Some(index.build_bloom()); + index.path_trie = Some(index.build_path_trie()); + + let serialized = serialize_compact(&index); + let (loaded, _tri_ms) = deserialize_compact(&serialized, uffs_mft::platform::DriveLetter::T) + .unwrap_or_else(|err| panic!("empty-trigram cache must deserialize, got: {err}")); + + assert_eq!( + loaded.ext_names, + vec![Box::from("rs"), Box::from("toml")], + "ext_names must survive an empty trigram section" + ); + let bloom = loaded.bloom.as_ref().expect("bloom section must load"); + assert!( + bloom.contains(b"rs"), + "the loaded bloom must answer for the drive's real extensions" + ); + assert_eq!( + loaded.frs_to_compact, index.frs_to_compact, + "frs_to_compact must survive an empty trigram section" + ); + // The empty on-disk trigram is rebuilt from records (an empty + // index over a non-empty drive would silently break substring + // search); records are non-empty here, so the rebuild is real. + assert!( + loaded.trigram.posting_count() > 0, + "trigram must be rebuilt from records, not adopted empty" + ); +} + +/// The parked-tier loader walks the same section chain with its own +/// arithmetic (`find_v9_filters_offset`) — pin that it, too, lands on +/// the real bloom when the trigram section is empty (its old +/// `key_count > 0` branch believed in a bare-sentinel shape no writer +/// ever produced, shifting the filter reads by 8 bytes). +#[test] +fn parked_loader_finds_filters_past_an_empty_trigram() { + let mut index = make_test_index(); + index.trigram = Arc::new(TrigramIndex::empty()); + index.ext_names = vec![Box::from("rs"), Box::from("toml")]; + index.bloom = Some(index.build_bloom()); + index.path_trie = Some(index.build_path_trie()); + + let serialized = serialize_compact(&index); + let parked = deserialize_parked_body(&serialized, uffs_mft::platform::DriveLetter::T) + .unwrap_or_else(|err| panic!("parked load must survive an empty trigram, got: {err}")); + assert!( + parked.bloom.contains(b"rs"), + "the parked bloom must answer for the drive's real extensions" + ); +} + +/// Regression: a DELTA-CARRYING index must never serialize — its +/// Arc-shared base sections are stale against the patched records. +/// +/// A create appends a record (the header's `rc` grows) while +/// `children` stays the base CSR with `rc_base + 1` offsets; the +/// reader sizes the children section from the header, so every later +/// section is read shifted and the cache quarantines on the next +/// start (field: `bloom k_hashes out of range`, winbox M/C/D/S +/// 2026-08-24 — caches written BY the then-current daemon between two +/// restarts). [`DriveCompactIndex::fold_delta_for_save`] is the +/// repair the save path applies; pin both halves. +#[test] +fn delta_carrying_index_misaligns_and_fold_repairs_it() { + let mut index = make_test_index(); + index.bloom = Some(index.build_bloom()); + index.path_trie = Some(index.build_path_trie()); + + // Simulate the surgical-patch state: one created record appended + // (name "qux" appended to the blob), bases left Arc-shared/stale, + // delta overlay armed. + let mut records: Vec = index.records.as_slice().to_vec(); + let mut names: Vec = index.names.as_slice().to_vec(); + let name_offset = u32::try_from(names.len()).unwrap_or(u32::MAX); + names.extend_from_slice(b"qux"); + records.push(CompactRecord { + name_offset, + parent_idx: 0, + name_len: 3, + name_first_byte: b'q', + ..CompactRecord::default() + }); + index.records = ColumnStorage::from_vec(records); + index.names = ColumnStorage::from_vec(names); + index.delta = Some(crate::compact::IndexDelta::default()); + + // The poisoned bytes can NEVER round-trip faithfully: the children + // CSR on disk was written for the 3 base records, so the created + // record's parent→child edge does not exist in the file (the delta + // that held it is not serialised), and the header-driven reader + // walks the shifted sections — on big real-world drives that + // surfaces as a parse error and a quarantined cache ("bloom + // k_hashes out of range"); on this small fixture the shifted bytes + // happen to parse, which is the SILENT face of the same bug. + // Either face proves the shape must never reach disk. + let poisoned = serialize_compact(&index); + match deserialize_compact(&poisoned, uffs_mft::platform::DriveLetter::T) { + Err(_parse_error) => {} // the quarantine face + Ok((corrupt, _tri_ms)) => { + let mut root_children: Vec = Vec::new(); + corrupt.for_each_child(0, |child| root_children.push(child)); + assert!( + !root_children.contains(&3), + "the created record's children edge was never written — a load that \ + reports it would falsify the poison analysis" + ); + } + } + + // The repair the save path applies: fold the delta, then serialize. + index.fold_delta_for_save(); + assert!(index.delta.is_none(), "fold must clear the overlay"); + let healthy = serialize_compact(&index); + let (loaded, _tri_ms) = deserialize_compact(&healthy, uffs_mft::platform::DriveLetter::T) + .unwrap_or_else(|err| panic!("folded index must round-trip, got: {err}")); + assert_eq!(loaded.records.len(), 4, "the created record must persist"); + let mut root_children: Vec = Vec::new(); + loaded.for_each_child(0, |child| root_children.push(child)); + assert!( + root_children.contains(&3), + "the folded children CSR must carry the created record's edge" + ); + assert!( + loaded.bloom.is_some(), + "the bloom section must land intact after the fold" + ); +} + +/// The save boundary refuses a delta-carrying index outright — the +/// second line of defence behind the save path's fold, so any future +/// writer path that forgets to fold fails loudly instead of writing a +/// poisoned cache. +#[test] +fn background_save_refuses_a_delta_carrying_index() { + let mut index = make_test_index(); + index.delta = Some(crate::compact::IndexDelta::default()); + let err = save_compact_cache_background(&index) + .expect_err("a delta-carrying index must be refused at the save boundary"); + assert!( + err.to_string().contains("delta"), + "the refusal must name the delta invariant: {err}" + ); +} diff --git a/crates/uffs-daemon/src/cache/shard.rs b/crates/uffs-daemon/src/cache/shard.rs index d6fda3b25..df815ddb1 100644 --- a/crates/uffs-daemon/src/cache/shard.rs +++ b/crates/uffs-daemon/src/cache/shard.rs @@ -518,10 +518,19 @@ impl ShardEntry { /// revisits the layout. /// /// [`DriveCompactIndex::frs_to_compact`]: uffs_core::compact::DriveCompactIndex::frs_to_compact + /// `fold_delta` — pass `true` on a SAVE-bound patch: the returned + /// body will be serialised, and a delta-carrying index must never + /// reach the writer (its base CSR sections are stale against the + /// patched records — see + /// [`DriveCompactIndex::fold_delta_for_save`]). The apply-only + /// tick passes `false` and keeps the cheap overlay. + /// + /// [`DriveCompactIndex::fold_delta_for_save`]: uffs_core::compact::DriveCompactIndex::fold_delta_for_save #[must_use] pub(crate) fn apply_usn_patch_to_body( &self, changes: &[uffs_mft::usn::FileChange], + fold_delta: bool, ) -> Option<( Arc, uffs_core::compact_loader::PatchStats, @@ -539,6 +548,9 @@ impl ShardEntry { // the small delta, not the hundreds-of-MB inverted indexes. let mut owned: DriveCompactIndex = (**body_arc).clone(); let stats = uffs_core::compact_loader::apply_usn_patch(&mut owned, changes); + if fold_delta { + owned.fold_delta_for_save(); + } Some((Arc::new(owned), stats)) } } diff --git a/crates/uffs-daemon/src/cache/shard/tests.rs b/crates/uffs-daemon/src/cache/shard/tests.rs index 17f992d7d..34d04009d 100644 --- a/crates/uffs-daemon/src/cache/shard/tests.rs +++ b/crates/uffs-daemon/src/cache/shard/tests.rs @@ -523,7 +523,7 @@ fn apply_usn_patch_to_body_returns_new_arc_on_warm() { // Empty change batch — the method must still produce a fresh // Arc so the caller's swap path is exercised even on no-op ticks. - let result = shard.apply_usn_patch_to_body(&[]); + let result = shard.apply_usn_patch_to_body(&[], false); let (new_body, stats) = result.expect("Warm shard must yield Some"); assert!( @@ -554,7 +554,7 @@ fn apply_usn_patch_to_body_returns_none_on_parked() { ..FileChange::default() }]; - let result = shard.apply_usn_patch_to_body(&changes); + let result = shard.apply_usn_patch_to_body(&changes, false); assert!( result.is_none(), "Parked shard has no in-memory body — must return None" @@ -570,7 +570,7 @@ fn apply_usn_patch_to_body_returns_none_on_cold() { let stats = Arc::new(DriveStats::new()); let shard = ShardEntry::new_cold(uffs_mft::platform::DriveLetter::C, stats); - let result = shard.apply_usn_patch_to_body(&[]); + let result = shard.apply_usn_patch_to_body(&[], false); assert!( result.is_none(), "Cold shard has no in-memory body — must return None" @@ -599,7 +599,7 @@ fn apply_usn_patch_to_body_lands_delete_on_new_arc() { }]; let (new_body, stats) = shard - .apply_usn_patch_to_body(&changes) + .apply_usn_patch_to_body(&changes, false) .expect("Warm shard yields Some"); assert_eq!(stats.deleted, 1, "exactly one delete should land"); @@ -630,3 +630,52 @@ fn apply_usn_patch_to_body_lands_delete_on_new_arc() { "original body Arc must be unaffected by patch on the clone" ); } + +/// Save-bound patches fold the delta overlay; apply-only patches keep +/// it. The body handed to the background disk save must be +/// delta-free — a delta-carrying index serialises its base children +/// CSR against a grown record count and poisons the cache for the +/// next start (`bloom k_hashes out of range`; see +/// `DriveCompactIndex::fold_delta_for_save`). A create is the +/// trigger shape: it appends a record, which is exactly what desyncs +/// the Arc-shared bases. +#[test] +fn save_bound_patch_folds_delta_apply_only_keeps_it() { + let create = || { + vec![FileChange { + frs: 77_u64.into(), + parent_frs: 5_u64.into(), + filename: "fresh.txt".to_owned(), + created: true, + ..FileChange::default() + }] + }; + + let body = Arc::new(make_test_body(uffs_mft::platform::DriveLetter::C)); + let shard = ShardEntry::new_warm(uffs_mft::platform::DriveLetter::C, Arc::clone(&body)); + let (applied, stats) = shard + .apply_usn_patch_to_body(&create(), false) + .expect("Warm shard yields Some"); + assert_eq!(stats.created, 1, "the create must land"); + assert!( + applied.delta.is_some(), + "apply-only patch must keep the cheap delta overlay" + ); + + let shard2 = ShardEntry::new_warm(uffs_mft::platform::DriveLetter::C, body); + let (saved, stats2) = shard2 + .apply_usn_patch_to_body(&create(), true) + .expect("Warm shard yields Some"); + assert_eq!( + stats2.created, 1, + "the create must land on the save path too" + ); + assert!( + saved.delta.is_none(), + "save-bound patch must fold the delta before the body reaches the writer" + ); + assert!( + saved.records.len() > 2, + "the created record must survive the fold" + ); +} diff --git a/crates/uffs-daemon/src/handler.rs b/crates/uffs-daemon/src/handler.rs index c21792bc5..fa1640018 100644 --- a/crates/uffs-daemon/src/handler.rs +++ b/crates/uffs-daemon/src/handler.rs @@ -5,9 +5,8 @@ //! [`crate::index::IndexManager`]. use uffs_client::protocol::response::{ - DEFAULT_PRELOAD_PIN_MINUTES, FacetValuesParams, FacetValuesResponse, ForgetParams, - ForgetResponse, HibernateParams, HibernateResponse, LoadDriveParams, LoadDriveResponse, - PreloadParams, PreloadResponse, RefreshParams, SearchPayload, StatusDrivesResponse, + FacetValuesParams, FacetValuesResponse, ForgetParams, ForgetResponse, LoadDriveParams, + LoadDriveResponse, RefreshParams, SearchPayload, StatusDrivesResponse, }; use uffs_client::protocol::{ AggregateSpecWire, ERR_DRIVE_BUSY, ERR_INVALID_PARAMS, ERR_METHOD_NOT_FOUND, RpcErrorResponse, @@ -52,6 +51,12 @@ mod diff_handler; #[path = "handler_journal.rs"] mod journal_handler; +// The memory-tiering handlers (`hibernate`, `preload`, `warm_plan`) live in +// a sibling file for the same 800-LOC policy reason; `#[path]` keeps them +// `impl RequestHandler` methods the dispatcher calls directly. +#[path = "handler_tiering.rs"] +mod tiering_handler; + /// Request handler holding shared daemon state. pub(crate) struct RequestHandler { /// Shared index manager. @@ -92,6 +97,7 @@ impl RequestHandler { // respectively. "hibernate" => self.handle_hibernate(id, req).await, "preload" => self.handle_preload(id, req).await, + "warm_plan" => self.handle_warm_plan(id, req).await, "forget" => self.handle_forget(id, req).await, "status_drives" => self.handle_status_drives(id).await, "changed_since" => self.handle_changed_since(id, req).await, @@ -578,112 +584,6 @@ impl RequestHandler { serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() } - /// Handle `hibernate` method (Phase 8-B). - /// - /// Parses [`HibernateParams`] from the JSON-RPC envelope, walks - /// the registry via [`IndexManager::hibernate_shards`], and - /// returns the structured [`HibernateResponse`] reporting - /// drives demoted from each pre-call tier plus drives that were - /// already at the bottom. - /// - /// Empty `drives` in the params means "every loaded drive"; - /// non-matching letters in a non-empty `drives` filter are - /// silently dropped (the operator audit lives on the - /// `already_cold` field of the response, which lists only - /// drives the daemon actually knows about). - /// - /// Malformed params (anything that fails to deserialise as - /// [`HibernateParams`]) fall back to the empty-default - /// (hibernate every drive); the wire contract is "best-effort - /// match" rather than "strict reject" because the all-loaded - /// path is always safe and an over-strict reject would surprise - /// scripts that send slightly-non-canonical JSON. - async fn handle_hibernate(&self, id: u64, req: &RpcRequest) -> String { - let params: HibernateParams = req - .params - .as_ref() - .and_then(|val| serde_json::from_value(val.clone()).ok()) - .unwrap_or_default(); - let outcome = self.index.hibernate_shards(¶ms.drives).await; - let response = HibernateResponse { - hot_demoted: outcome.hot_demoted, - warm_demoted: outcome.warm_demoted, - parked_demoted: outcome.parked_demoted, - already_cold: outcome.already_cold, - }; - let result = serde_json::to_value(&response).unwrap_or_default(); - serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() - } - - /// Handle `preload` method (Phase 8-C). - /// - /// Parses [`PreloadParams`] from the JSON-RPC envelope, loops - /// over the requested drives calling - /// [`IndexManager::preload_drive`] for each, and aggregates the - /// per-drive [`crate::index::tiering_ops::PreloadOutcome`]s into - /// a single [`PreloadResponse`]. - /// - /// Validates that the params include at least one drive — an - /// empty `drives` vector returns [`ERR_INVALID_PARAMS`] so a - /// caller's mistyped script doesn't silently succeed. The pin - /// duration defaults to [`DEFAULT_PRELOAD_PIN_MINUTES`] when - /// the params omit `pin_minutes`. - async fn handle_preload(&self, id: u64, req: &RpcRequest) -> String { - let params: PreloadParams = req - .params - .as_ref() - .and_then(|val| serde_json::from_value(val.clone()).ok()) - .unwrap_or_default(); - if params.drives.is_empty() { - return serde_json::to_string(&RpcErrorResponse::error( - Some(id), - ERR_INVALID_PARAMS, - "preload: `drives` must contain at least one drive letter", - )) - .unwrap_or_default(); - } - let pin_minutes = params.pin_minutes.unwrap_or(DEFAULT_PRELOAD_PIN_MINUTES); - - let mut promoted: Vec = Vec::new(); - let mut already_hot: Vec = Vec::new(); - let mut errors: Vec = Vec::new(); - let mut latest_pin_until_ms: i64 = 0; - - for &letter in ¶ms.drives { - use crate::index::tiering_ops::PreloadOutcome; - match self.index.preload_drive(letter, pin_minutes).await { - PreloadOutcome::Promoted { pin_until_ms, .. } => { - promoted.push(letter); - latest_pin_until_ms = i64::try_from(pin_until_ms).unwrap_or(i64::MAX); - } - PreloadOutcome::AlreadyHot { pin_until_ms } => { - already_hot.push(letter); - latest_pin_until_ms = i64::try_from(pin_until_ms).unwrap_or(i64::MAX); - } - PreloadOutcome::UnknownDrive => { - errors.push(format!("{letter}: drive not loaded")); - } - PreloadOutcome::LoadFailed => { - errors.push(format!("{letter}: body load failed")); - } - PreloadOutcome::Busy { from_state } => { - errors.push(format!( - "{letter}: drive busy in transient state ({from_state})" - )); - } - } - } - - let response = PreloadResponse { - promoted, - already_hot, - errors, - pin_until_unix_ms: latest_pin_until_ms, - }; - let result = serde_json::to_value(&response).unwrap_or_default(); - serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() - } - /// Handle `forget` method (Phase 8-D). /// /// Parses [`ForgetParams`] from the JSON-RPC envelope and diff --git a/crates/uffs-daemon/src/handler_tiering.rs b/crates/uffs-daemon/src/handler_tiering.rs new file mode 100644 index 000000000..5f6693d84 --- /dev/null +++ b/crates/uffs-daemon/src/handler_tiering.rs @@ -0,0 +1,157 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! Memory-tiering RPC handlers: `hibernate`, `preload`, and +//! `warm_plan`. +//! +//! Split out of `handler.rs` to keep the dispatcher under the +//! workspace 800-LOC policy ceiling; `#[path]` in `handler.rs` +//! re-attaches this file to the same `impl RequestHandler`, so the +//! dispatch table calls these as ordinary methods +//! (`self.handle_hibernate(...)` etc.) with no call-site changes. +//! +//! `warm_plan` (v0.6.38) is the daemon-authoritative cold-index +//! answer for external gates: given the SAME `SearchParams` a search +//! would carry, it returns the promote set dispatch would actually +//! execute — Parked/Cold shards in scope whose bloom cannot prove +//! them irrelevant to the resolved extension filter. Gates that +//! instead infer readiness from the tier marker alone over-gate: +//! a bloom-skippable Parked drive answers its query instantly with +//! zero rows and must not be reported "warming" or force-promoted +//! (field 2026-08-24, winbox drive E:). + +use uffs_client::protocol::response::{ + DEFAULT_PRELOAD_PIN_MINUTES, HibernateParams, HibernateResponse, PreloadParams, + PreloadResponse, WarmPlanResponse, +}; +use uffs_client::protocol::{ERR_INVALID_PARAMS, RpcErrorResponse, RpcRequest, RpcResponse}; + +use super::RequestHandler; + +impl RequestHandler { + /// Handle `warm_plan`: report which drives the given search would + /// promote, without promoting anything. + /// + /// Params are a full [`uffs_client::protocol::SearchParams`] — + /// the caller sends the SAME params it is about to search with, + /// and the daemon resolves them through the same + /// canonicalisation + filter pipeline the search itself uses + /// ([`crate::index::IndexManager::warm_plan`]), so the answer is + /// bit-exact with what dispatch will do. Validation matches + /// `handle_search` (S4.4.3 pattern-length guard included). + pub(super) async fn handle_warm_plan(&self, id: u64, req: &RpcRequest) -> String { + let params = match Self::parse_and_validate_search_params(req) { + Ok(params) => params, + Err(parse_err) => return parse_err.to_rpc_error_json(id), + }; + let needs_promote = self.index.warm_plan(¶ms).await; + let result = serde_json::to_value(&WarmPlanResponse { needs_promote }).unwrap_or_default(); + serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() + } + + /// Handle `hibernate` method (Phase 8-B). + /// + /// Parses [`HibernateParams`] from the JSON-RPC envelope, walks + /// the registry via [`crate::index::IndexManager::hibernate_shards`], and + /// returns the structured [`HibernateResponse`] reporting + /// drives demoted from each pre-call tier plus drives that were + /// already at the bottom. + /// + /// Empty `drives` in the params means "every loaded drive"; + /// non-matching letters in a non-empty `drives` filter are + /// silently dropped (the operator audit lives on the + /// `already_cold` field of the response, which lists only + /// drives the daemon actually knows about). + /// + /// Malformed params (anything that fails to deserialise as + /// [`HibernateParams`]) fall back to the empty-default + /// (hibernate every drive); the wire contract is "best-effort + /// match" rather than "strict reject" because the all-loaded + /// path is always safe and an over-strict reject would surprise + /// scripts that send slightly-non-canonical JSON. + pub(super) async fn handle_hibernate(&self, id: u64, req: &RpcRequest) -> String { + let params: HibernateParams = req + .params + .as_ref() + .and_then(|val| serde_json::from_value(val.clone()).ok()) + .unwrap_or_default(); + let outcome = self.index.hibernate_shards(¶ms.drives).await; + let response = HibernateResponse { + hot_demoted: outcome.hot_demoted, + warm_demoted: outcome.warm_demoted, + parked_demoted: outcome.parked_demoted, + already_cold: outcome.already_cold, + }; + let result = serde_json::to_value(&response).unwrap_or_default(); + serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() + } + + /// Handle `preload` method (Phase 8-C). + /// + /// Parses [`PreloadParams`] from the JSON-RPC envelope, loops + /// over the requested drives calling + /// [`crate::index::IndexManager::preload_drive`] for each, and aggregates + /// the per-drive [`crate::index::tiering_ops::PreloadOutcome`]s into + /// a single [`PreloadResponse`]. + /// + /// Validates that the params include at least one drive — an + /// empty `drives` vector returns [`ERR_INVALID_PARAMS`] so a + /// caller's mistyped script doesn't silently succeed. The pin + /// duration defaults to [`DEFAULT_PRELOAD_PIN_MINUTES`] when + /// the params omit `pin_minutes`. + pub(super) async fn handle_preload(&self, id: u64, req: &RpcRequest) -> String { + let params: PreloadParams = req + .params + .as_ref() + .and_then(|val| serde_json::from_value(val.clone()).ok()) + .unwrap_or_default(); + if params.drives.is_empty() { + return serde_json::to_string(&RpcErrorResponse::error( + Some(id), + ERR_INVALID_PARAMS, + "preload: `drives` must contain at least one drive letter", + )) + .unwrap_or_default(); + } + let pin_minutes = params.pin_minutes.unwrap_or(DEFAULT_PRELOAD_PIN_MINUTES); + + let mut promoted: Vec = Vec::new(); + let mut already_hot: Vec = Vec::new(); + let mut errors: Vec = Vec::new(); + let mut latest_pin_until_ms: i64 = 0; + + for &letter in ¶ms.drives { + use crate::index::tiering_ops::PreloadOutcome; + match self.index.preload_drive(letter, pin_minutes).await { + PreloadOutcome::Promoted { pin_until_ms, .. } => { + promoted.push(letter); + latest_pin_until_ms = i64::try_from(pin_until_ms).unwrap_or(i64::MAX); + } + PreloadOutcome::AlreadyHot { pin_until_ms } => { + already_hot.push(letter); + latest_pin_until_ms = i64::try_from(pin_until_ms).unwrap_or(i64::MAX); + } + PreloadOutcome::UnknownDrive => { + errors.push(format!("{letter}: drive not loaded")); + } + PreloadOutcome::LoadFailed => { + errors.push(format!("{letter}: body load failed")); + } + PreloadOutcome::Busy { from_state } => { + errors.push(format!( + "{letter}: drive busy in transient state ({from_state})" + )); + } + } + } + + let response = PreloadResponse { + promoted, + already_hot, + errors, + pin_until_unix_ms: latest_pin_until_ms, + }; + let result = serde_json::to_value(&response).unwrap_or_default(); + serde_json::to_string(&RpcResponse::success(id, result)).unwrap_or_default() + } +} diff --git a/crates/uffs-daemon/src/index/dispatch.rs b/crates/uffs-daemon/src/index/dispatch.rs index 0bcda38b8..a1416f630 100644 --- a/crates/uffs-daemon/src/index/dispatch.rs +++ b/crates/uffs-daemon/src/index/dispatch.rs @@ -100,32 +100,11 @@ impl IndexManager { ext_terms: &[String], ) { // ── Phase 1: read-lock detection (fast path) ─────────── - // Identify Parked/Cold shards in the touched set. Single - // read-lock acquisition; the registry's `iter()` is a Vec - // walk, no allocation beyond the `needs_promote` Vec. - // - // Phase 4 Commit F — for Parked shards we additionally - // probe the bloom against `ext_terms`: a miss means the - // shard provably has no records matching the ext filter, - // so we skip the promote entirely (zero-RAM-touch - // contract). Cold shards drop their bloom on demote, so - // they always promote. Empty `ext_terms` short-circuits - // to the Phase-3 always-promote behaviour. - let needs_promote: Vec = { - let guard = self.index.read().await; - guard - .iter() - .filter(|shard| params_drives.is_empty() || params_drives.contains(&shard.drive)) - .filter(|shard| { - matches!( - shard.state(), - crate::cache::ShardState::Parked | crate::cache::ShardState::Cold - ) - }) - .filter(|shard| Self::bloom_pre_check_should_promote(shard, ext_terms)) - .map(|shard| shard.drive) - .collect() - }; + // See [`Self::promote_plan`] — the same computation also + // answers the `warm_plan` RPC, so external gates (the MCP + // cold-index gate) and this dispatch can never disagree + // about which drives a query would actually promote. + let needs_promote = self.promote_plan(params_drives, ext_terms).await; if needs_promote.is_empty() { return; } @@ -448,6 +427,64 @@ impl IndexManager { body } + /// The drives a dispatch with this scope + resolved ext-term set + /// would need to promote: every Parked/Cold shard in the touched + /// set whose bloom cannot prove it irrelevant. + /// + /// Single read-lock acquisition; the registry's `iter()` is a Vec + /// walk, no allocation beyond the returned Vec. + /// + /// Phase 4 Commit F — for Parked shards we additionally probe the + /// bloom against `ext_terms`: a miss means the shard provably has + /// no records matching the ext filter, so the promote is skipped + /// entirely (zero-RAM-touch contract). Cold shards drop their + /// bloom on demote, so they always promote. Empty `ext_terms` + /// short-circuits to the Phase-3 always-promote behaviour. + /// + /// Shared verbatim by [`Self::ensure_warm_for_dispatch`] (which + /// then executes the promotes) and the `warm_plan` RPC (which + /// reports the set to external gates) — one computation, so the + /// MCP cold-index gate can never contradict what dispatch will + /// actually do (field 2026-08-24: the gate's tier-only view + /// declared a bloom-skippable Parked drive "not ready" and + /// force-promoted its body for a query the bloom had already + /// answered). + pub(crate) async fn promote_plan( + &self, + params_drives: &[uffs_mft::platform::DriveLetter], + ext_terms: &[String], + ) -> Vec { + let guard = self.index.read().await; + guard + .iter() + .filter(|shard| params_drives.is_empty() || params_drives.contains(&shard.drive)) + .filter(|shard| { + matches!( + shard.state(), + crate::cache::ShardState::Parked | crate::cache::ShardState::Cold + ) + }) + .filter(|shard| Self::bloom_pre_check_should_promote(shard, ext_terms)) + .map(|shard| shard.drive) + .collect() + } + + /// The `warm_plan` RPC body: the promote set the SAME search would + /// trigger, derived through the same canonicalisation + filter + /// resolution `run_search_over` performs before its ensure-warm + /// call — bit-exact with dispatch by construction, because both + /// funnel into [`Self::promote_plan`]. + pub(crate) async fn warm_plan( + &self, + params: &uffs_client::protocol::SearchParams, + ) -> Vec { + let mut effective = params.clone(); + effective.populate_canonical_fields(); + let filters = super::search_filters_build::build_search_filters(&effective); + self.promote_plan(&effective.drives, &filters.extensions) + .await + } + /// Phase 4 Commit F — bloom pre-check for Parked shards. /// /// Returns `true` when the shard must be promoted (the search diff --git a/crates/uffs-daemon/src/index/journal.rs b/crates/uffs-daemon/src/index/journal.rs index 5bcc70663..5f703efc4 100644 --- a/crates/uffs-daemon/src/index/journal.rs +++ b/crates/uffs-daemon/src/index/journal.rs @@ -161,7 +161,10 @@ impl IndexManager { reason: &str, changes: Vec, ) -> bool { - match self.apply_to_body(letter, reason, changes).await { + // Save-bound: fold any delta overlay inside the patch task so + // the body handed to the disk writer is delta-free (the + // serialization invariant `fold_delta_for_save` documents). + match self.apply_to_body(letter, reason, changes, true).await { BodyApplyOutcome::Applied(new_body) => { spawn_compact_cache_save_task(letter, new_body); true @@ -197,8 +200,11 @@ impl IndexManager { reason: &str, changes: Vec, ) -> bool { + // Apply-only: keep the cheap delta overlay — nothing is + // serialised on this tick, and folding every ~2 s apply would + // be an O(total-records) rebuild per tick. !matches!( - self.apply_to_body(letter, reason, changes).await, + self.apply_to_body(letter, reason, changes, false).await, BodyApplyOutcome::Failed ) } @@ -215,6 +221,7 @@ impl IndexManager { letter: uffs_mft::platform::DriveLetter, reason: &str, changes: Vec, + fold_delta: bool, ) -> BodyApplyOutcome { if changes.is_empty() { log_save_empty_batch(letter, reason); @@ -228,7 +235,7 @@ impl IndexManager { }; let (new_body, stats) = match self - .run_surgical_patch_task(&shard, letter, reason, changes) + .run_surgical_patch_task(&shard, letter, reason, changes, fold_delta) .await { PatchTaskOutcome::Applied(body, stats) => (body, stats), @@ -281,13 +288,14 @@ impl IndexManager { letter: uffs_mft::platform::DriveLetter, reason: &str, changes: Vec, + fold_delta: bool, ) -> PatchTaskOutcome { let background_io = Arc::clone(&self.background_io); let shard_for_patch = Arc::clone(shard); let change_count = changes.len(); let patch_result = tokio::task::spawn_blocking(move || { let _bg_scope = crate::cache::background_io::BackgroundIoScope::enter(background_io); - shard_for_patch.apply_usn_patch_to_body(&changes) + shard_for_patch.apply_usn_patch_to_body(&changes, fold_delta) }) .await; classify_patch_result(patch_result, letter, reason, change_count) diff --git a/crates/uffs-daemon/src/index/mod.rs b/crates/uffs-daemon/src/index/mod.rs index 6dccde03f..a966a76bd 100644 --- a/crates/uffs-daemon/src/index/mod.rs +++ b/crates/uffs-daemon/src/index/mod.rs @@ -25,6 +25,7 @@ mod predicates; mod projection; mod refresh; pub(crate) mod search; +mod search_filters_build; mod stats; mod status_drives; mod test_helpers; diff --git a/crates/uffs-daemon/src/index/search.rs b/crates/uffs-daemon/src/index/search.rs index 38c8366c9..6174b1eb5 100644 --- a/crates/uffs-daemon/src/index/search.rs +++ b/crates/uffs-daemon/src/index/search.rs @@ -23,7 +23,6 @@ use uffs_core::search::backend::{ DriveIndex, FilterMode, PhaseTimings, SearchRequest, SortSpec, search_index, }; use uffs_core::search::field::FieldId; -use uffs_core::search::filters::{SearchFilterParams, SearchFilters}; use super::IndexManager; @@ -117,43 +116,11 @@ impl IndexManager { SearchFilterMode::All => FilterMode::All, }; - let ep = &effective_params; - let mut filters = SearchFilters::from_params(&SearchFilterParams { - hide_system: ep.hide_system, - hide_ads: ep.hide_ads, - min_size: ep.min_size, - max_size: ep.max_size, - min_descendants: ep.min_descendants, - max_descendants: ep.max_descendants, - newer: ep.newer.as_deref(), - older: ep.older.as_deref(), - newer_created: ep.newer_created.as_deref(), - older_created: ep.older_created.as_deref(), - newer_accessed: ep.newer_accessed.as_deref(), - older_accessed: ep.older_accessed.as_deref(), - attr_filter: ep.attr.as_deref(), - ext_filter: ep.ext.as_deref(), - exclude: ep.exclude.as_deref(), - path_contains: ep.path_contains.as_deref(), - path_excludes: ep.path_excludes.as_deref(), - type_filter: ep.type_filter.as_deref(), - min_bulkiness: ep.min_bulkiness, - max_bulkiness: ep.max_bulkiness, - min_name_len: ep.min_name_len, - max_name_len: ep.max_name_len, - min_path_len: ep.min_path_len, - max_path_len: ep.max_path_len, - min_allocated: ep.min_allocated, - max_allocated: ep.max_allocated, - min_treesize: ep.min_treesize, - max_treesize: ep.max_treesize, - min_tree_allocated: ep.min_tree_allocated, - max_tree_allocated: ep.max_tree_allocated, - allowed_months: &ep.allowed_months, - }); - // Display-only: select the malformed-name render mode for resolved - // paths + the name column (`--normalize-malformed`). - filters.normalize_malformed = ep.normalize_malformed; + // Resolution shared with the `warm_plan` RPC (see + // `search_filters_build`): both must derive the SAME + // `filters.extensions`, or an external warm gate could + // contradict the bloom pre-check below. + let mut filters = super::search_filters_build::build_search_filters(&effective_params); // Overlay canonical predicates that can be compiled into the hot // path (size / descendant bounds). diff --git a/crates/uffs-daemon/src/index/search_filters_build.rs b/crates/uffs-daemon/src/index/search_filters_build.rs new file mode 100644 index 000000000..0f9fee06d --- /dev/null +++ b/crates/uffs-daemon/src/index/search_filters_build.rs @@ -0,0 +1,60 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! The [`SearchParams`] → [`SearchFilters`] resolution, shared by the +//! search pipeline and the `warm_plan` RPC. +//! +//! Factored out of `run_search_over` so the promote-plan the daemon +//! reports to external gates (`warm_plan`, consumed by the MCP +//! cold-index gate) is derived through the *same* filter resolution +//! the search itself will use — most importantly the resolved +//! extension-term set that drives the Phase 4 bloom pre-check. Two +//! copies of this mapping would eventually disagree on exactly the +//! field the bloom contract depends on. + +use uffs_client::protocol::SearchParams; +use uffs_core::search::filters::{SearchFilterParams, SearchFilters}; + +/// Resolve a search's record/display filter set from its (already +/// canonicalised) wire params. Pure parameter mapping: predicates +/// compilation, diff-mode overlays, and other pipeline-stage mutations +/// stay with the caller. +pub(super) fn build_search_filters(ep: &SearchParams) -> SearchFilters { + let mut filters = SearchFilters::from_params(&SearchFilterParams { + hide_system: ep.hide_system, + hide_ads: ep.hide_ads, + min_size: ep.min_size, + max_size: ep.max_size, + min_descendants: ep.min_descendants, + max_descendants: ep.max_descendants, + newer: ep.newer.as_deref(), + older: ep.older.as_deref(), + newer_created: ep.newer_created.as_deref(), + older_created: ep.older_created.as_deref(), + newer_accessed: ep.newer_accessed.as_deref(), + older_accessed: ep.older_accessed.as_deref(), + attr_filter: ep.attr.as_deref(), + ext_filter: ep.ext.as_deref(), + exclude: ep.exclude.as_deref(), + path_contains: ep.path_contains.as_deref(), + path_excludes: ep.path_excludes.as_deref(), + type_filter: ep.type_filter.as_deref(), + min_bulkiness: ep.min_bulkiness, + max_bulkiness: ep.max_bulkiness, + min_name_len: ep.min_name_len, + max_name_len: ep.max_name_len, + min_path_len: ep.min_path_len, + max_path_len: ep.max_path_len, + min_allocated: ep.min_allocated, + max_allocated: ep.max_allocated, + min_treesize: ep.min_treesize, + max_treesize: ep.max_treesize, + min_tree_allocated: ep.min_tree_allocated, + max_tree_allocated: ep.max_tree_allocated, + allowed_months: &ep.allowed_months, + }); + // Display-only: select the malformed-name render mode for resolved + // paths + the name column (`--normalize-malformed`). + filters.normalize_malformed = ep.normalize_malformed; + filters +} diff --git a/crates/uffs-daemon/src/index/tests/ensure_warm.rs b/crates/uffs-daemon/src/index/tests/ensure_warm.rs index 828c53a45..a6171e732 100644 --- a/crates/uffs-daemon/src/index/tests/ensure_warm.rs +++ b/crates/uffs-daemon/src/index/tests/ensure_warm.rs @@ -351,7 +351,7 @@ async fn ensure_warm_for_dispatch_promotes_in_parallel() { /// (folded basenames + extensions). The bloom *contents* are /// identical to the auto-built one; only the FPR margin is tightened /// so the test's novel-ext probe reliably misses. -fn build_test_drive_with_tight_bloom() -> uffs_core::compact::DriveCompactIndex { +pub(super) fn build_test_drive_with_tight_bloom() -> uffs_core::compact::DriveCompactIndex { use uffs_core::bloom::Bloom; /// Tighter than the production `SHARD_BLOOM_TARGET_FPR` (1 %) so diff --git a/crates/uffs-daemon/src/index/tests/mod.rs b/crates/uffs-daemon/src/index/tests/mod.rs index 96355576f..e25c72015 100644 --- a/crates/uffs-daemon/src/index/tests/mod.rs +++ b/crates/uffs-daemon/src/index/tests/mod.rs @@ -57,6 +57,7 @@ mod manager; mod registry; mod shard_ttl_events; mod tiering_ops; +mod warm_plan; // Exposed at `pub(crate)` so the shared `EventLog` / `CapturedEvent` // scaffold (already `pub(crate)`) can be imported from sibling test // modules such as `crate::cache::journal_loop::tests::save_log_message` diff --git a/crates/uffs-daemon/src/index/tests/warm_plan.rs b/crates/uffs-daemon/src/index/tests/warm_plan.rs new file mode 100644 index 000000000..ba5b39961 --- /dev/null +++ b/crates/uffs-daemon/src/index/tests/warm_plan.rs @@ -0,0 +1,134 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! `IndexManager::promote_plan` / `warm_plan` parity tests (v0.6.38). +//! +//! Split from [`super::ensure_warm`] to hold that file under the +//! workspace 800-LOC ceiling; shares its bloom fixture +//! ([`super::ensure_warm::build_test_drive_with_tight_bloom`]) so the +//! plan tests and the dispatch tests exercise identical shards. + +#![expect( + clippy::std_instead_of_alloc, + reason = "test code — `std::sync::Arc` matches the rest of the daemon's test fixtures" +)] + +use std::sync::Arc; + +use super::IndexManager; +use super::body_loader_fakes::PanickingBodyLoader; +use super::ensure_warm::build_test_drive_with_tight_bloom; +// ── `warm_plan` RPC parity with dispatch (v0.6.38) ───────────────── +// +// The MCP cold-index gate consumes `promote_plan`/`warm_plan` instead +// of inferring readiness from the tier marker; these pin that the plan +// is bit-exact with what `ensure_warm_for_dispatch` would promote — +// bloom pre-check included (field 2026-08-24: the tier-only gate +// declared a bloom-skippable Parked drive E: "not ready" and +// force-promoted its body for an ext-filtered query the bloom had +// already answered). + +/// Bloom miss ⇒ the plan is empty: the gate must pass the query +/// through, because dispatch would promote nothing. +#[tokio::test] +async fn promote_plan_is_empty_on_bloom_miss() { + use crate::cache::ShardState; + + let (tx, _rx) = crate::events::event_channel(); + let mgr = IndexManager::with_body_loader_for_test(None, tx, Arc::new(PanickingBodyLoader)); + mgr.add_drive(build_test_drive_with_tight_bloom()).await; + assert!( + mgr.demote_letter_for_test(uffs_mft::platform::DriveLetter::C, ShardState::Parked) + .await + ); + + // `csv` is novel to the fixture (its extensions are md/rs/toml/bin). + let plan = mgr + .promote_plan(&[uffs_mft::platform::DriveLetter::C], &["csv".to_owned()]) + .await; + assert!( + plan.is_empty(), + "bloom miss must yield an empty plan — got {plan:?}" + ); +} + +/// Bloom hit and no-ext-filter both plan the promote; a Warm shard +/// never appears in the plan. +#[tokio::test] +async fn promote_plan_names_drives_dispatch_would_promote() { + use crate::cache::ShardState; + + let (tx, _rx) = crate::events::event_channel(); + let mgr = IndexManager::with_body_loader_for_test(None, tx, Arc::new(PanickingBodyLoader)); + mgr.add_drive(build_test_drive_with_tight_bloom()).await; + + // Warm shard: plan must be empty regardless of filter. + assert!( + mgr.promote_plan(&[], &[]).await.is_empty(), + "a Warm shard must never be planned for promotion" + ); + + assert!( + mgr.demote_letter_for_test(uffs_mft::platform::DriveLetter::C, ShardState::Parked) + .await + ); + + // Ext present in the fixture → bloom hit → planned. + assert_eq!( + mgr.promote_plan(&[], &["rs".to_owned()]).await, + vec![uffs_mft::platform::DriveLetter::C], + "bloom hit must plan the promote" + ); + // No ext filter → Phase-3 always-promote behaviour. + assert_eq!( + mgr.promote_plan(&[], &[]).await, + vec![uffs_mft::platform::DriveLetter::C], + "no ext filter must plan every Parked shard in scope" + ); + // Out-of-scope drive filter → not planned. + assert!( + mgr.promote_plan(&[uffs_mft::platform::DriveLetter::D], &[]) + .await + .is_empty(), + "a drive outside the query scope must not be planned" + ); +} + +/// `warm_plan` resolves the ext-term set from wire `SearchParams` +/// through the same filter pipeline the search uses — the RPC-facing +/// half of the parity contract. +#[tokio::test] +async fn warm_plan_resolves_ext_filter_like_the_search_does() { + use crate::cache::ShardState; + + let (tx, _rx) = crate::events::event_channel(); + let mgr = IndexManager::with_body_loader_for_test(None, tx, Arc::new(PanickingBodyLoader)); + mgr.add_drive(build_test_drive_with_tight_bloom()).await; + assert!( + mgr.demote_letter_for_test(uffs_mft::platform::DriveLetter::C, ShardState::Parked) + .await + ); + + let mut params_novel = uffs_client::protocol::SearchParams { + pattern: "*".to_owned(), + ext: Some("csv".to_owned()), + ..Default::default() + }; + params_novel.populate_canonical_fields(); + assert!( + mgr.warm_plan(¶ms_novel).await.is_empty(), + "--ext csv (novel to the fixture) must plan nothing" + ); + + let mut params_hit = uffs_client::protocol::SearchParams { + pattern: "*".to_owned(), + ext: Some("rs".to_owned()), + ..Default::default() + }; + params_hit.populate_canonical_fields(); + assert_eq!( + mgr.warm_plan(¶ms_hit).await, + vec![uffs_mft::platform::DriveLetter::C], + "--ext rs (present in the fixture) must plan the promote" + ); +} diff --git a/crates/uffs-mcp/src/tools/aggregate.rs b/crates/uffs-mcp/src/tools/aggregate.rs index 4b94e87ac..b3d4c4044 100644 --- a/crates/uffs-mcp/src/tools/aggregate.rs +++ b/crates/uffs-mcp/src/tools/aggregate.rs @@ -136,7 +136,9 @@ pub(crate) async fn run( // Cold-index contract: return a retryable "warming" error instead of // blocking silently for the length of a re-warm (see `tools::warm`). - super::warm::warm_gate(client, ¶ms.drives).await?; + // Full params, not just the drive scope: the daemon's plan consults + // the query's resolved ext/type filter against each Parked bloom. + super::warm::warm_gate(client, ¶ms).await?; // Log the exact RPC payload for debugging parity with API validation. tracing::info!( diff --git a/crates/uffs-mcp/src/tools/facet_values.rs b/crates/uffs-mcp/src/tools/facet_values.rs index 36b2913b1..b3898bdac 100644 --- a/crates/uffs-mcp/src/tools/facet_values.rs +++ b/crates/uffs-mcp/src/tools/facet_values.rs @@ -105,7 +105,9 @@ pub(crate) async fn run( // Cold-index contract: return a retryable "warming" error instead of // blocking silently for the length of a re-warm (see `tools::warm`). - super::warm::warm_gate(client, ¶ms.drives).await?; + // Full params, not just the drive scope: the daemon's plan consults + // the query's resolved ext/type filter against each Parked bloom. + super::warm::warm_gate(client, ¶ms).await?; let mut response = client .search(¶ms) diff --git a/crates/uffs-mcp/src/tools/info.rs b/crates/uffs-mcp/src/tools/info.rs index 105ab3aa1..9f4087f82 100644 --- a/crates/uffs-mcp/src/tools/info.rs +++ b/crates/uffs-mcp/src/tools/info.rs @@ -32,6 +32,9 @@ pub(crate) async fn run( // Cold-index contract, scoped to the path's own drive: an info // lookup on a parked drive must not block for the re-warm either. + // Minimal match-all params: an info lookup has no ext filter, so + // the daemon's plan is "promote every Parked/Cold drive in scope" + // — the body genuinely is needed to resolve the path. let scope: Vec = args .path .chars() @@ -39,7 +42,12 @@ pub(crate) async fn run( .and_then(|ch| uffs_mft::platform::DriveLetter::parse(ch).ok()) .into_iter() .collect(); - super::warm::warm_gate(client, &scope).await?; + let scope_params = uffs_client::protocol::SearchParams { + pattern: "*".to_owned(), + drives: scope, + ..Default::default() + }; + super::warm::warm_gate(client, &scope_params).await?; let response = client .info(&args.path) diff --git a/crates/uffs-mcp/src/tools/search.rs b/crates/uffs-mcp/src/tools/search.rs index f39a6811c..01ba29240 100644 --- a/crates/uffs-mcp/src/tools/search.rs +++ b/crates/uffs-mcp/src/tools/search.rs @@ -366,7 +366,9 @@ pub(crate) async fn run( // Cold-index contract: return a retryable "warming" error instead of // blocking silently for the length of a re-warm (see `tools::warm`). - super::warm::warm_gate(client, &search_params.drives).await?; + // Full params, not just the drive scope: the daemon's plan consults + // the query's resolved ext filter against each Parked bloom. + super::warm::warm_gate(client, &search_params).await?; tracing::debug!( params_json = %serde_json::to_string(&search_params).unwrap_or_default(), diff --git a/crates/uffs-mcp/src/tools/warm.rs b/crates/uffs-mcp/src/tools/warm.rs index b9fb60124..9633fd377 100644 --- a/crates/uffs-mcp/src/tools/warm.rs +++ b/crates/uffs-mcp/src/tools/warm.rs @@ -37,10 +37,26 @@ //! (MCP progress notifications) was considered and rejected: hosts do //! not extend their tool-call patience on progress events, so the //! call would still read as hung. +//! +//! # Who decides "not ready": the daemon, not this gate +//! +//! Since v0.6.38 the gate asks the daemon's `warm_plan` RPC for the +//! exact promote set its dispatch would execute for the query — which +//! includes the Phase-4 bloom pre-check. A Parked drive whose bloom +//! proves it holds nothing matching the query's extension filter is +//! ready *by definition* (dispatch skips it and it contributes zero +//! rows), so it must neither gate the call nor be force-warmed. The +//! previous tier-only inference did both (field 2026-08-24, winbox +//! drive E:), and its warm trigger — pattern-only, no ext filter — +//! bypassed the daemon's own bloom skip and paged the body back into +//! RAM. The tier-based check survives only as `not_ready`, the +//! fallback for pre-`warm_plan` daemons, where it is conservative +//! (over-gates, never under-gates). use uffs_client::connect::UffsClient; -use uffs_client::protocol::SearchParams; +use uffs_client::error::ClientError; use uffs_client::protocol::response::{DriveInfo, ShardTier}; +use uffs_client::protocol::{ERR_METHOD_NOT_FOUND, SearchParams}; use crate::error::BridgeError; @@ -75,24 +91,51 @@ fn not_ready( .collect() } -/// Gate a query tool on index warmth: `Ok(())` when every scoped drive -/// is `Warm`/`Hot`; otherwise kick a detached re-warm and return the -/// retry-shaped error described in the module docs. +/// Whether a client error is the daemon saying it does not know the +/// `warm_plan` method (pre-v0.6.38): the signal to fall back to the +/// legacy tier-based gate rather than fail the query tool. +const fn is_method_not_found(err: &ClientError) -> bool { + matches!(err, ClientError::DaemonError { code, .. } if *code == ERR_METHOD_NOT_FOUND) +} + +/// Gate a query tool on index warmth: `Ok(())` when the daemon's own +/// dispatch would promote nothing for `params`; otherwise kick a +/// detached re-warm of exactly the drives the daemon named and return +/// the retry-shaped error described in the module docs. +/// +/// The decision is daemon-authoritative (`warm_plan` RPC): the daemon +/// computes the promote set through the same filter resolution + bloom +/// pre-check its dispatch uses, so a Parked drive whose bloom proves it +/// irrelevant to the query's extension filter passes the gate and is +/// never force-promoted. Deciding from the tier marker alone — the +/// pre-v0.6.38 behaviour, kept as the fallback for older daemons — +/// over-gates exactly that case (field 2026-08-24, winbox drive `E:` — an +/// ext-filtered query the bloom had already answered was reported +/// "warming — retry" while the warm trigger paged the drive's body +/// back into RAM, defeating the Phase-4 zero-RAM-touch contract). /// /// # Errors /// /// Returns [`BridgeError::Daemon`] when scoped drives are re-warming -/// (the retry contract) or when the tier probe itself fails. +/// (the retry contract) or when the plan/tier probe itself fails. pub(crate) async fn warm_gate( client: &mut UffsClient, - scope: &[uffs_mft::platform::DriveLetter], + params: &SearchParams, ) -> Result<(), BridgeError> { - let drives = client - .drives() - .await - .map_err(|err| BridgeError::Daemon(format!("warm check failed: {err}")))?; - - let cold = not_ready(&drives.drives, scope); + let cold = match client.warm_plan(params).await { + Ok(plan) => plan.needs_promote, + Err(err) if is_method_not_found(&err) => { + // Pre-`warm_plan` daemon: legacy tier-based gate. Strictly + // more conservative than the plan (it cannot consult the + // bloom), never less — a query is at worst gated when it + // could have passed, exactly the pre-v0.6.38 behaviour. + let drives = client.drives().await.map_err(|drives_err| { + BridgeError::Daemon(format!("warm check failed: {drives_err}")) + })?; + not_ready(&drives.drives, ¶ms.drives) + } + Err(err) => return Err(BridgeError::Daemon(format!("warm check failed: {err}"))), + }; if cold.is_empty() { return Ok(()); } @@ -109,14 +152,14 @@ pub(crate) async fn warm_gate( tracing::warn!("warm trigger: daemon connect failed — retry will re-trigger"); return; }; - let mut params = SearchParams { + let mut trigger_params = SearchParams { pattern: WARM_TRIGGER_PATTERN.to_owned(), drives: trigger_drives.clone(), limit: Some(1), ..Default::default() }; - params.populate_canonical_fields(); - match warm_client.search(¶ms).await { + trigger_params.populate_canonical_fields(); + match warm_client.search(&trigger_params).await { Ok(_) => tracing::info!(drives = ?trigger_drives, "warm trigger completed"), Err(err) => tracing::warn!(%err, "warm trigger search failed"), } @@ -204,4 +247,24 @@ mod tests { let drives = vec![info('C', None)]; assert_eq!(not_ready(&drives, &[]), Vec::::new()); } + + /// Only the daemon's method-not-found answer selects the legacy + /// tier-based fallback; every other failure propagates. + #[test] + fn only_method_not_found_selects_the_legacy_fallback() { + use uffs_client::error::ClientError; + use uffs_client::protocol::ERR_METHOD_NOT_FOUND; + + use super::is_method_not_found; + + assert!(is_method_not_found(&ClientError::DaemonError { + code: ERR_METHOD_NOT_FOUND, + message: "Method not found: warm_plan".to_owned(), + })); + assert!(!is_method_not_found(&ClientError::DaemonError { + code: -32_602_i32, + message: "Missing or invalid search params".to_owned(), + })); + assert!(!is_method_not_found(&ClientError::ConnectionClosed)); + } }