diff --git a/Cargo.lock b/Cargo.lock index 016339cb9..752e780ea 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1939,9 +1939,9 @@ checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" [[package]] name = "memmap2" -version = "0.9.10" +version = "0.9.11" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "714098028fe011992e1c3962653c96b2d578c4b4bce9036e15ff220319b1e0e3" +checksum = "d1219ed1b7f229ee7104d281dd01d6802fe28bb6e95d292942c4daacdeb798c0" dependencies = [ "libc", ] diff --git a/crates/uffs-core/src/compact.rs b/crates/uffs-core/src/compact.rs index 5b98502c9..0a1ddb1a9 100644 --- a/crates/uffs-core/src/compact.rs +++ b/crates/uffs-core/src/compact.rs @@ -602,6 +602,56 @@ impl DriveCompactIndex { ids.dedup(); ids } + + /// Extract `filename`'s extension and intern it into [`Self::ext_names`], + /// returning its `extension_id`. + /// + /// Mirrors the build-time logic (`MftIndex::intern_extension` + + /// `ExtensionTable::intern`) so a record created by the USN journal patch + /// path lands under the SAME `extension_id` a full rebuild would assign — + /// otherwise `--ext ` (which resolves the name via [`Self::ext_names`] + /// and looks it up in the [`ExtensionIndex`]) silently misses the new file. + /// + /// Returns `0` (the reserved "no extension" id) for a dotless name, a + /// leading-dot dotfile (`.gitignore`), a trailing-dot name (`file.`), or if + /// the table is already at the `u16::MAX` interning ceiling. + pub(crate) fn intern_extension(&mut self, filename: &str) -> u16 { + // Extension = substring after the LAST dot, where the dot is neither + // the first byte (dotfile) nor the last (trailing dot). + let Some(dot_pos) = filename.rfind('.') else { + return 0; + }; + if dot_pos == 0 || dot_pos + 1 >= filename.len() { + return 0; + } + let Some(raw_ext) = filename.get(dot_pos + 1..) else { + return 0; + }; + let normalized = raw_ext.trim_start_matches('.').to_lowercase(); + if normalized.is_empty() { + return 0; + } + + // Find-or-append. `ext_names[0]` is the reserved "" (no-extension) + // slot, so a real extension never collides with id 0. + if let Some(existing) = self + .ext_names + .iter() + .position(|name| name.as_ref() == normalized) + { + return u16::try_from(existing).unwrap_or(0); + } + let Ok(new_id) = u16::try_from(self.ext_names.len()) else { + // Interning ceiling reached (>= 65 535 distinct extensions); + // fall back to "no extension" rather than wrap. + return 0; + }; + if new_id == u16::MAX { + return 0; + } + self.ext_names.push(normalized.into_boxed_str()); + new_id + } } /// Expand alternate data streams (ADS) for a single record, producing the diff --git a/crates/uffs-core/src/compact_loader.rs b/crates/uffs-core/src/compact_loader.rs index cdd78989c..8676a18c8 100644 --- a/crates/uffs-core/src/compact_loader.rs +++ b/crates/uffs-core/src/compact_loader.rs @@ -454,6 +454,210 @@ pub fn load_mft_file( load_drive(&MftSource::File(mft_path.to_path_buf(), drive), no_cache) } +/// A USN-created file's identity, staged into the index's names blob + +/// extension table via a mutable `drive` borrow BEFORE any record borrow. +/// +/// All fields are `Copy`, so the caller can take a `&mut CompactRecord` +/// after this returns without a borrow conflict. +struct StagedCreate { + /// Byte offset of the staged name in `drive.names`. + name_offset: u32, + /// UTF-8 byte length of the staged name. + name_len: u16, + /// Cached first byte of the name (hot-path metafile gate). + name_first_byte: u8, + /// Interned extension id for the new name (`0` = no extension). + extension_id: u16, + /// Compact index of the parent directory (`u32::MAX` if unmapped). + parent_idx: u32, + /// Real size/timestamps/flags from a targeted MFT read, or all-zero when + /// the USN-only change carried no metadata (a later re-warm fills it). + /// Representation matches `CompactRecord`, so it copies straight in. + meta: uffs_mft::usn::RecordMeta, +} + +/// Append `change`'s filename to the names blob and intern its extension, +/// resolving the parent's compact index. Mutably borrows `drive`, so it +/// must run before any `&mut CompactRecord` borrow. +fn stage_create(drive: &mut DriveCompactIndex, change: &uffs_mft::usn::FileChange) -> StagedCreate { + let extension_id = drive.intern_extension(&change.filename); + let name_start = drive.names.len(); + drive + .names + .as_mut_vec() + .extend_from_slice(change.filename.as_bytes()); + let parent_frs_usize = uffs_mft::frs_to_usize(change.parent_frs.raw()); + let parent_idx = drive + .frs_to_compact + .get(parent_frs_usize) + .copied() + .unwrap_or(u32::MAX); + StagedCreate { + name_offset: uffs_mft::len_to_u32(name_start), + name_len: uffs_mft::len_to_u16(change.filename.len()), + name_first_byte: change.filename.as_bytes().first().copied().unwrap_or(0), + extension_id, + parent_idx, + meta: change.meta.unwrap_or_default(), + } +} + +/// Overwrite an existing compact slot with a reused/re-animated file's +/// identity. Per-file metrics come from the staged metadata — real values +/// when a targeted MFT read backfilled them, else zero (a later re-warm +/// fills them; the USN `FileChange` carries only name + parent). +const fn overwrite_slot(rec: &mut CompactRecord, staged: &StagedCreate) { + rec.name_offset = staged.name_offset; + rec.name_len = staged.name_len; + rec.name_first_byte = staged.name_first_byte; + rec.extension_id = staged.extension_id; + rec.parent_idx = staged.parent_idx; + rec.size = staged.meta.size; + rec.allocated = staged.meta.allocated; + rec.created = staged.meta.created; + rec.modified = staged.meta.modified; + rec.accessed = staged.meta.accessed; + rec.flags = staged.meta.flags; + // Tree metrics are recomputed post-loop (CSR rebuild + compute_path_ + // lengths); never carried by a USN change. + rec.treesize = 0; + rec.tree_allocated = 0; + rec.descendants = 0; + rec.path_len = 0; +} + +/// Apply a delete change: tombstone the slot (`name_len = 0`, parent +/// unmapped so the CSR rebuild drops it) and unmap its FRS so a later batch +/// can't re-animate the tombstone. +fn apply_delete( + drive: &mut DriveCompactIndex, + frs_usize: usize, + compact_idx: u32, + stats: &mut PatchStats, +) { + if compact_idx == u32::MAX { + stats.skipped += 1; + return; + } + if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) { + rec.name_len = 0; + rec.parent_idx = u32::MAX; + if let Some(slot) = drive.frs_to_compact.get_mut(frs_usize) { + *slot = u32::MAX; + } + stats.deleted += 1; + } +} + +/// Apply a create change: overwrite the mapped slot when the MFT record +/// number was reused (tombstone OR stale live record), or append a fresh +/// record + register its FRS mapping when the slot is new. +fn apply_create( + drive: &mut DriveCompactIndex, + change: &uffs_mft::usn::FileChange, + frs_usize: usize, + compact_idx: u32, + stats: &mut PatchStats, +) { + if change.filename.is_empty() { + stats.skipped += 1; + return; + } + // Stage name + interned extension up front (mutable index borrow) so the + // per-record write can take a `&mut CompactRecord` without conflict. + let staged = stage_create(drive, change); + if compact_idx == u32::MAX { + // Brand-new record: append, then register the FRS mapping. NTFS + // reuses freed record numbers and a long-running daemon can outgrow + // the build-time table, so extend + sentinel-fill any gap. + let new_rec = CompactRecord { + size: staged.meta.size, + allocated: staged.meta.allocated, + treesize: 0, + tree_allocated: 0, + created: staged.meta.created, + modified: staged.meta.modified, + accessed: staged.meta.accessed, + name_offset: staged.name_offset, + flags: staged.meta.flags, + parent_idx: staged.parent_idx, + descendants: 0, + name_len: staged.name_len, + extension_id: staged.extension_id, + // path_len filled by `compute_path_lengths` post-loop. + path_len: 0, + name_first_byte: staged.name_first_byte, + _pad: [0; 1], + }; + let new_compact_idx = uffs_mft::len_to_u32(drive.records.len()); + drive.records.as_mut_vec().push(new_rec); + if frs_usize >= drive.frs_to_compact.len() { + drive + .frs_to_compact + .resize(frs_usize.saturating_add(1), u32::MAX); + } + if let Some(slot) = drive.frs_to_compact.get_mut(frs_usize) { + *slot = new_compact_idx; + } + stats.created += 1; + } else if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) { + // The record number is already mapped. A `created` event means NTFS + // reused that slot for a NEW file — the old occupant (a tombstone, OR + // a stale live record whose delete was coalesced/missed) no longer + // exists. Overwrite it wholesale. Skipping a live slot here is what + // dropped FRS-reused recreates (the "delta.pdf vanished" report). + overwrite_slot(rec, &staged); + stats.created += 1; + } +} + +/// Apply a rename change: re-point the name, **re-intern the extension** (a +/// rename can change it: `foo.log` → `foo.pdf`), refresh the first-byte +/// cache, and update `parent_idx`. The FRS keeps its slot, so the mapping is +/// unchanged. +fn apply_rename( + drive: &mut DriveCompactIndex, + change: &uffs_mft::usn::FileChange, + compact_idx: u32, + stats: &mut PatchStats, +) { + if compact_idx == u32::MAX || change.filename.is_empty() { + stats.skipped += 1; + return; + } + let extension_id = drive.intern_extension(&change.filename); + let name_start = drive.names.len(); + drive + .names + .as_mut_vec() + .extend_from_slice(change.filename.as_bytes()); + let new_parent_frs = uffs_mft::frs_to_usize(change.parent_frs.raw()); + let new_parent_compact = drive + .frs_to_compact + .get(new_parent_frs) + .copied() + .unwrap_or(u32::MAX); + if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) { + rec.name_offset = uffs_mft::len_to_u32(name_start); + rec.name_len = uffs_mft::len_to_u16(change.filename.len()); + rec.extension_id = extension_id; + rec.name_first_byte = change.filename.as_bytes().first().copied().unwrap_or(0); + rec.parent_idx = new_parent_compact; + // Apply backfilled size/timestamps/flags when a targeted MFT read + // attached them (corrects a record previously created USN-only with + // zeroed metrics); otherwise leave the existing values untouched. + if let Some(meta) = change.meta { + rec.size = meta.size; + rec.allocated = meta.allocated; + rec.created = meta.created; + rec.modified = meta.modified; + rec.accessed = meta.accessed; + rec.flags = meta.flags; + } + stats.renamed += 1; + } +} + /// Apply USN changes in-place to the compact index. /// /// Mutates records (`parent_idx`, names, flags) and the @@ -489,17 +693,6 @@ pub fn load_mft_file( /// looks up to `u32::MAX` and the function increments `skipped` for /// the whole batch \u2014 the surgical patch silently degrades to a /// no-op so the caller's full-reload fallback path runs. -#[expect( - clippy::too_many_lines, - reason = "Phase 8 surgical-patch loop: the create / delete / rename \ - branches each mutate `drive.frs_to_compact` in a \ - variant-specific way (delete tombstones the slot, create \ - extends + registers, rename leaves it intact); splitting \ - into per-variant helpers would scatter the FRS-mapping \ - invariant across functions and obscure the symmetric \ - treatment that's central to the surgical-patch correctness \ - contract." -)] pub fn apply_usn_patch( drive: &mut DriveCompactIndex, changes: &[uffs_mft::usn::FileChange], @@ -508,10 +701,9 @@ pub fn apply_usn_patch( for change in changes { // Typed `Frs` → raw `u64` lift at the frs_to_compact CSR lookup - // boundary. The mapping table is `Vec` indexed by - // `usize`, so demoting once per change keeps the inner index - // arithmetic on raw values without leaking raw FRS into the - // outer `FileChange` API. + // boundary. The mapping table is `Vec` indexed by `usize`, + // so demoting once per change keeps the inner index arithmetic on + // raw values without leaking raw FRS into the `FileChange` API. let frs_usize = uffs_mft::frs_to_usize(change.frs.raw()); let compact_idx = drive .frs_to_compact @@ -519,127 +711,29 @@ pub fn apply_usn_patch( .copied() .unwrap_or(u32::MAX); + // Per-change disposition trace — enable with `--log-level trace` to + // see exactly which branch each USN event takes and the slot it + // resolved to (the field-debug hook for USN-delta investigations). + tracing::trace!( + drive = %drive.letter, + frs = change.frs.raw(), + name = %change.filename, + created = change.created, + deleted = change.deleted, + renamed = change.renamed, + compact_idx, + mapped = (compact_idx != u32::MAX), + "usn apply: change" + ); + + // The flags are mutually-exclusive net states (resolved in + // `aggregate_changes`), so a simple priority dispatch is correct. if change.deleted { - if compact_idx == u32::MAX { - stats.skipped += 1; - } else if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) { - rec.name_len = 0; - // Clear parent so CSR rebuild excludes this record. - rec.parent_idx = u32::MAX; - // Phase 8: mark the FRS slot unmapped so a future - // batch can't re-animate the tombstone via the - // `compact_idx != u32::MAX` create branch below. - if let Some(slot) = drive.frs_to_compact.get_mut(frs_usize) { - *slot = u32::MAX; - } - stats.deleted += 1; - } + apply_delete(drive, frs_usize, compact_idx, &mut stats); } else if change.created { - if compact_idx != u32::MAX { - // Re-animate a previously deleted slot. - if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) - && rec.name_len == 0 - && !change.filename.is_empty() - { - let name_start = drive.names.len(); - drive - .names - .as_mut_vec() - .extend_from_slice(change.filename.as_bytes()); - rec.name_offset = uffs_mft::len_to_u32(name_start); - rec.name_len = uffs_mft::len_to_u16(change.filename.len()); - } - stats.skipped += 1; - } else if !change.filename.is_empty() { - let name_start = drive.names.len(); - drive - .names - .as_mut_vec() - .extend_from_slice(change.filename.as_bytes()); - - // Typed `ParentFrs` → raw `u64` lift at the - // frs_to_compact CSR lookup boundary (same rationale as - // the `change.frs.raw()` lift above). - let parent_frs_usize = uffs_mft::frs_to_usize(change.parent_frs.raw()); - let parent_compact = drive - .frs_to_compact - .get(parent_frs_usize) - .copied() - .unwrap_or(u32::MAX); - - let new_rec = CompactRecord { - size: 0, - allocated: 0, - treesize: 0, - tree_allocated: 0, - created: 0, - modified: 0, - accessed: 0, - name_offset: uffs_mft::len_to_u32(name_start), - flags: 0, - parent_idx: parent_compact, - descendants: 0, - name_len: uffs_mft::len_to_u16(change.filename.len()), - extension_id: 0, - // path_len is set to 0 here; the full-array - // `compute_path_lengths` call after the USN loop - // will populate the correct value for all records. - path_len: 0, - name_first_byte: change.filename.as_bytes().first().copied().unwrap_or(0), - _pad: [0; 1], - }; - - let new_compact_idx = uffs_mft::len_to_u32(drive.records.len()); - drive.records.as_mut_vec().push(new_rec); - - // Phase 8: register the new FRS → compact_idx mapping - // so future batches that reference this FRS find the - // correct slot. Extend the table if needed (the FRS - // may exceed the build-time max — e.g. NTFS reuses - // freed FRS slots after deletes, and a long-running - // daemon can outgrow the original `frs_to_idx` - // capacity). Sentinel-fill any intermediate gap so - // skipped FRS values still report `u32::MAX`. - if frs_usize >= drive.frs_to_compact.len() { - drive - .frs_to_compact - .resize(frs_usize.saturating_add(1), u32::MAX); - } - if let Some(slot) = drive.frs_to_compact.get_mut(frs_usize) { - *slot = new_compact_idx; - } - stats.created += 1; - } else { - stats.skipped += 1; - } + apply_create(drive, change, frs_usize, compact_idx, &mut stats); } else if change.renamed { - if compact_idx == u32::MAX { - stats.skipped += 1; - } else if let Some(rec) = drive.records.as_mut_slice().get_mut(compact_idx as usize) { - if !change.filename.is_empty() { - let name_start = drive.names.len(); - drive - .names - .as_mut_vec() - .extend_from_slice(change.filename.as_bytes()); - rec.name_offset = uffs_mft::len_to_u32(name_start); - rec.name_len = uffs_mft::len_to_u16(change.filename.len()); - } - - // Typed `ParentFrs` → raw lift on the rename path. - let new_parent_frs = uffs_mft::frs_to_usize(change.parent_frs.raw()); - let new_parent_compact = drive - .frs_to_compact - .get(new_parent_frs) - .copied() - .unwrap_or(u32::MAX); - - // Update parent_idx — CSR rebuild picks this up. - rec.parent_idx = new_parent_compact; - // Rename keeps the FRS in the same compact slot; - // mapping is unchanged. - stats.renamed += 1; - } + apply_rename(drive, change, compact_idx, &mut stats); } else { stats.skipped += 1; } @@ -657,9 +751,29 @@ pub fn apply_usn_patch( // Rebuild extension inverted index so --ext queries reflect USN changes. drive.ext_index = crate::compact::ExtensionIndex::build(&drive.records); + if !changes.is_empty() { + log_batch_summary(drive, changes.len(), &stats); + } + stats } +/// Emit the per-batch USN-apply summary (how the poll mutated the index) +/// at DEBUG. +fn log_batch_summary(drive: &DriveCompactIndex, changes: usize, stats: &PatchStats) { + tracing::debug!( + drive = %drive.letter, + changes, + created = stats.created, + deleted = stats.deleted, + renamed = stats.renamed, + skipped = stats.skipped, + records = drive.records.len(), + ext_index_entries = drive.ext_index.total_entries(), + "usn apply: batch applied" + ); +} + #[cfg(test)] #[path = "compact_loader_tests.rs"] mod tests; diff --git a/crates/uffs-core/src/compact_loader_tests.rs b/crates/uffs-core/src/compact_loader_tests.rs index 41cb5bd86..9b86f4d5e 100644 --- a/crates/uffs-core/src/compact_loader_tests.rs +++ b/crates/uffs-core/src/compact_loader_tests.rs @@ -236,6 +236,161 @@ fn apply_usn_patch_renamed_record_has_new_name_in_blob() { ); } +/// Regression (v0.6.13 field report): a rename that changes the +/// extension (`charlie.log` → `charlie.pdf`) must re-intern the new +/// extension so the record is findable by `--ext pdf` and drops out of +/// `--ext log`. The rename branch used to update only `name_offset` / +/// `name_len`, leaving the stale `extension_id` behind. +#[test] +fn apply_usn_patch_rename_reinterns_extension() { + let mut drive = make_synthetic_drive(); + // FRS 11 → compact_idx 2 ("bar.rs"). Rename it to "bar.pdf". + let changes = vec![FileChange { + frs: 11_u64.into(), + parent_frs: 5_u64.into(), + filename: "bar.pdf".to_owned(), + renamed: true, + ..FileChange::default() + }]; + apply_usn_patch(&mut drive, &changes); + + let pdf_ids = drive.resolve_ext_ids(&["pdf".to_owned()]); + assert_eq!(pdf_ids.len(), 1, "'pdf' must be interned after the rename"); + let pdf_id = *pdf_ids.first().expect("one id"); + let record = drive.records.as_slice().get(2).expect("record 2"); + assert_eq!( + record.extension_id, pdf_id, + "renamed record must carry the new 'pdf' extension_id" + ); + assert_eq!( + record.name_first_byte, b'b', + "first-byte cache must reflect the renamed name" + ); + assert!( + drive.ext_index.get(pdf_id).contains(&2), + "ExtensionIndex.get(pdf) must include the renamed record" + ); +} + +/// Regression (v0.6.13 field report): FRS reuse. NTFS reuses an MFT +/// record number after a delete, so a `created` event can land on a +/// slot whose mapping still points at a *live* (stale) record — e.g. +/// when the prior delete was coalesced away. The create must REPLACE +/// that slot with the new file's identity, not silently skip it (which +/// dropped the new file and was the root of the "delta.pdf vanished" +/// and "recreate after delete loses files" reports). +#[test] +fn apply_usn_patch_create_replaces_live_reused_slot() { + let mut drive = make_synthetic_drive(); + // FRS 11 → compact_idx 2 ("bar.rs", a LIVE record, name_len 6). + // A create for FRS 11 means the record number was reused. + let new_idx = 2_usize; + let changes = vec![FileChange { + frs: 11_u64.into(), + parent_frs: 5_u64.into(), + filename: "reused.pdf".to_owned(), + created: true, + ..FileChange::default() + }]; + apply_usn_patch(&mut drive, &changes); + + let record = drive.records.as_slice().get(new_idx).expect("record 2"); + let name_start = record.name_offset as usize; + let name_end = name_start + record.name_len as usize; + let name_bytes = drive + .names + .as_slice() + .get(name_start..name_end) + .expect("name slice in blob"); + assert_eq!( + name_bytes, b"reused.pdf", + "reused slot must hold the NEW file's name" + ); + let pdf_ids = drive.resolve_ext_ids(&["pdf".to_owned()]); + let pdf_id = *pdf_ids.first().expect("'pdf' interned"); + assert_eq!(record.extension_id, pdf_id, "reused slot tagged 'pdf'"); + assert!( + drive.ext_index.get(pdf_id).contains(&2), + "ExtensionIndex.get(pdf) must include the reused record" + ); +} + +/// Metadata backfill: when the journal source attaches a `RecordMeta` +/// (from a targeted MFT read), the created record carries the real +/// size/timestamps/flags instead of the USN-only zeros. Covers both the +/// append (new FRS) and the overwrite (reused slot) paths, plus rename. +#[test] +fn apply_usn_patch_applies_backfilled_metadata() { + use uffs_mft::usn::RecordMeta; + + let meta = RecordMeta { + size: 1_637_013, + allocated: 1_638_400, + created: 1_700_000_000_000_000, + modified: 1_700_000_500_000_000, + accessed: 1_700_000_900_000_000, + flags: 0x20, // FILE_ATTRIBUTE_ARCHIVE + }; + + // Append path: brand-new FRS 13 with metadata. + let mut appended_drive = make_synthetic_drive(); + let appended_idx = appended_drive.records.len(); + apply_usn_patch(&mut appended_drive, &[FileChange { + frs: 13_u64.into(), + parent_frs: 5_u64.into(), + filename: "report.pdf".to_owned(), + created: true, + meta: Some(meta), + ..FileChange::default() + }]); + let appended = appended_drive + .records + .as_slice() + .get(appended_idx) + .expect("appended"); + assert_eq!( + appended.size, meta.size, + "appended record carries real size" + ); + assert_eq!(appended.modified, meta.modified, "and real modified time"); + assert_eq!(appended.flags, meta.flags, "and real attribute flags"); + + // Overwrite path: a reused live slot (FRS 11 → idx 2) with metadata. + let mut overwrite_drive = make_synthetic_drive(); + apply_usn_patch(&mut overwrite_drive, &[FileChange { + frs: 11_u64.into(), + parent_frs: 5_u64.into(), + filename: "reused.pdf".to_owned(), + created: true, + meta: Some(meta), + ..FileChange::default() + }]); + let overwritten = overwrite_drive + .records + .as_slice() + .get(2) + .expect("reused slot"); + assert_eq!( + overwritten.size, meta.size, + "overwritten slot carries real size" + ); + assert_eq!(overwritten.created, meta.created, "and real created time"); + + // No metadata (USN-only) still yields zeros — unchanged behaviour. + let mut bare_drive = make_synthetic_drive(); + let bare_idx = bare_drive.records.len(); + apply_usn_patch(&mut bare_drive, &[FileChange { + frs: 13_u64.into(), + parent_frs: 5_u64.into(), + filename: "bare.pdf".to_owned(), + created: true, + ..FileChange::default() + }]); + let bare = bare_drive.records.as_slice().get(bare_idx).expect("bare"); + assert_eq!(bare.size, 0, "USN-only create stays zero-size"); + assert_eq!(bare.modified, 0, "USN-only create stays zero-time"); +} + /// Create contract: a newly-created FRS that doesn't map to an /// existing compact slot (`frs_to_compact[frs] == u32::MAX`) appends /// a fresh record at the end with the correct `parent_idx`, @@ -277,6 +432,101 @@ fn apply_usn_patch_created_record_appended_with_correct_parent() { ); } +/// Regression (v0.6.13 field report): a file created via the USN journal +/// patch must become findable by `--ext`. The create branch used to +/// hardcode `extension_id: 0`, so the rebuilt `ExtensionIndex` filed the +/// new file under "no extension" — `uffs report.pdf` found it by name but +/// `uffs report --ext pdf` returned nothing. This pins the whole chain: +/// the extension is interned into `ext_names`, the record carries the real +/// `extension_id`, and the inverted index returns it for that id. +#[test] +fn apply_usn_patch_created_record_is_findable_by_extension() { + let mut drive = make_synthetic_drive(); + let new_idx = drive.records.len(); + + // Pre-condition: "pdf" is unknown on this synthetic drive, so an + // `--ext pdf` query resolves to no ids (the bug's starting point). + assert!( + drive.resolve_ext_ids(&["pdf".to_owned()]).is_empty(), + "fixture must not already know the 'pdf' extension" + ); + + let changes = vec![FileChange { + frs: 13_u64.into(), + parent_frs: 5_u64.into(), + filename: "report.pdf".to_owned(), + created: true, + ..FileChange::default() + }]; + let stats = apply_usn_patch(&mut drive, &changes); + assert_eq!(stats.created, 1, "the create should have applied"); + + // 1. The extension is now interned and resolvable. + let ids = drive.resolve_ext_ids(&["pdf".to_owned()]); + assert_eq!(ids.len(), 1, "'pdf' must resolve to exactly one ext id"); + let pdf_id = *ids.first().expect("one id present"); + assert_ne!( + pdf_id, 0, + "a real extension must not collapse to the no-ext id" + ); + + // 2. The new record carries that extension_id (not the hardcoded 0). + let record = drive + .records + .as_slice() + .get(new_idx) + .expect("created record reachable at the tail"); + assert_eq!( + record.extension_id, pdf_id, + "created record must be tagged with the resolved 'pdf' id" + ); + + // 3. The rebuilt inverted index returns the new record for that id — this is + // exactly what `--ext pdf` walks. + let matches = drive.ext_index.get(pdf_id); + assert!( + matches.contains(&u32::try_from(new_idx).expect("idx fits u32")), + "ExtensionIndex.get(pdf) must include the USN-created record" + ); +} + +/// Companion edge cases for [`DriveCompactIndex::intern_extension`] via the +/// create path: a dotless name and a leading-dot dotfile both resolve to +/// the reserved no-extension id (0) and never pollute `ext_names`. +#[test] +fn apply_usn_patch_dotless_and_dotfile_creates_have_no_extension() { + let mut drive = make_synthetic_drive(); + let ext_names_before = drive.ext_names.len(); + + let changes = vec![ + FileChange { + frs: 13_u64.into(), + parent_frs: 5_u64.into(), + filename: "Makefile".to_owned(), // dotless + created: true, + ..FileChange::default() + }, + FileChange { + frs: 14_u64.into(), + parent_frs: 5_u64.into(), + filename: ".gitignore".to_owned(), // leading-dot dotfile + created: true, + ..FileChange::default() + }, + ]; + apply_usn_patch(&mut drive, &changes); + + let makefile = drive.records.as_slice().get(4).expect("Makefile record"); + let gitignore = drive.records.as_slice().get(5).expect(".gitignore record"); + assert_eq!(makefile.extension_id, 0, "dotless name → no extension"); + assert_eq!(gitignore.extension_id, 0, "dotfile → no extension"); + assert_eq!( + drive.ext_names.len(), + ext_names_before, + "no-extension creates must not append to ext_names" + ); +} + /// Empty-batch fast path: passing zero changes produces all-zero /// stats and leaves the drive byte-for-byte unchanged in shape (same /// record count, same names blob length). Pins that the rebuilt diff --git a/crates/uffs-core/src/search/backend.rs b/crates/uffs-core/src/search/backend.rs index 2651a26fa..93e79e485 100644 --- a/crates/uffs-core/src/search/backend.rs +++ b/crates/uffs-core/src/search/backend.rs @@ -372,11 +372,15 @@ impl MultiDriveBackend { .build() { Ok(compiled_re) => { + // Immutable reborrow so the parallel per-drive closure is + // `Sync` (a captured `&mut` is not). The record-level + // filters are applied inside each scan, before its limit. + let sf: &super::filters::SearchFilters = search_filters; let drive_results: Vec> = self .drives .par_iter() .map(|drive| { - super::query::search_compact_drive_regex(drive, &compiled_re, limit) + super::query::search_compact_drive_regex(drive, &compiled_re, limit, sf) }) .collect(); for drive_rows in drive_results { @@ -410,6 +414,7 @@ impl MultiDriveBackend { // Trigram-accelerated prefix scan (`win*`). `is_prefix` already // proved `is_prefix_pattern` holds, so the strip is infallible. if let Some(prefix) = crate::search::tree::is_prefix_pattern(&needle) { + let sf: &super::filters::SearchFilters = search_filters; let drive_results: Vec> = self .drives .par_iter() @@ -419,6 +424,7 @@ impl MultiDriveBackend { prefix, limit, case_sensitive, + sf, ) }) .collect(); @@ -436,12 +442,13 @@ impl MultiDriveBackend { rows.truncate(limit); } } else { + let sf: &super::filters::SearchFilters = search_filters; let drive_results: Vec> = self .drives .par_iter() .map(|drive| { if is_path { - super::query::search_compact_drive_tree(drive, &needle, limit) + super::query::search_compact_drive_tree(drive, &needle, limit, sf) } else { super::query::search_compact_drive( drive, @@ -450,6 +457,7 @@ impl MultiDriveBackend { case_sensitive, whole_word, match_path, + sf, ) } }) diff --git a/crates/uffs-core/src/search/backend_tests.rs b/crates/uffs-core/src/search/backend_tests.rs index 596140fc2..4f3406dea 100644 --- a/crates/uffs-core/src/search/backend_tests.rs +++ b/crates/uffs-core/src/search/backend_tests.rs @@ -898,6 +898,113 @@ fn search_index_explicit_extensions_not_clobbered() { ); } +/// Build a single C: drive where many files share the substring `pcl5` +/// but only the LAST record (highest index) has extension `pdf`: +/// `pcl5_00.dll` … `pcl5_07.dll` then `pcl5_report.pdf`. +/// +/// Mirrors the field report: `PCL5-Assessment-*.pdf` sits among many +/// `PCL5*.DLL` system files, and the pdf is not among the first matches. +fn build_substring_then_extension_fixture() -> MultiDriveBackend { + use uffs_mft::index::{IndexNameRef, MftIndex, ROOT_FRS, SizeInfo}; + + use crate::compact::build_compact_index; + + let letter = uffs_mft::platform::DriveLetter::C; + let mut mft = MftIndex::new(letter); + let root_off = mft.add_name("."); + let root = mft.get_or_create(ROOT_FRS.into()); + root.stdinfo.set_directory(true); + root.first_name.name = IndexNameRef::new(root_off, 1, true, IndexNameRef::NO_EXTENSION); + root.first_name.parent_frs = Into::into(ROOT_FRS); + + // Eight `pcl5_NN.dll` files first (FRS 200..208), then one pdf last + // (FRS 999 → highest record index). All contain the substring "pcl5". + let add = |index: &mut MftIndex, frs: u64, name: &str, size: u64| { + let off = index.add_name(name); + let ext = index.intern_extension(name); + let rec = index.get_or_create(frs.into()); + rec.first_name.name = + IndexNameRef::new(off, u16::try_from(name.len()).expect("len"), true, ext); + rec.first_name.parent_frs = Into::into(ROOT_FRS); + rec.first_stream.size = SizeInfo { + length: size, + allocated: size.next_multiple_of(512), + }; + rec.stdinfo.flags = 0x20; + }; + for seq in 0..8_u64 { + add(&mut mft, 200 + seq, &format!("pcl5_{seq:02}.dll"), 1000); + } + add(&mut mft, 999, "pcl5_report.pdf", 5000); + + let mut backend = MultiDriveBackend::new(); + let (drive, _, _) = build_compact_index(letter, &mft); + backend.drives.push(drive); + backend +} + +/// Regression (v0.6.13 field report): `uffs pcl5 --ext pdf --limit 5` +/// returned nothing even though `pcl5_report.pdf` exists, matches the +/// substring, and has extension `pdf`. +/// +/// Root cause this pins: the per-drive substring scan truncates to +/// `--limit` BEFORE the extension filter runs (`collect_match_indices` +/// stops at `limit`, then `apply_search_filters` filters the survivors). +/// With 8 `.dll` matches ahead of the single `.pdf`, a limit of 5 fills +/// the result with `.dll`s, then `--ext pdf` removes all of them. +/// +/// The unlimited query in the same test proves the filter logic itself +/// is correct — the bug is purely the limit-before-filter ordering. +#[test] +fn search_substring_with_extension_filter_survives_limit() { + // Diagnostic 1: UNLIMITED — the ext filter alone works, pdf is found. + let mut backend_unlimited = build_substring_then_extension_fixture(); + let mut filters_unlimited = super::super::filters::SearchFilters { + extensions: vec!["pdf".to_owned()], + ..super::super::filters::SearchFilters::default() + }; + let unlimited = backend_unlimited.search(SearchRequest::new("pcl5", &mut filters_unlimited)); + assert!( + unlimited + .rows + .iter() + .any(|row| row.name() == "pcl5_report.pdf"), + "sanity: unlimited 'pcl5 --ext pdf' must find the pdf (got {:?})", + unlimited + .rows + .iter() + .map(DisplayRow::name) + .collect::>() + ); + + // Diagnostic 2: LIMIT 5 — the pdf must STILL be found. On the buggy + // code the limit truncates the 8 .dll matches before the ext filter, + // so this returns zero rows. + let mut backend = build_substring_then_extension_fixture(); + let mut filters_limited = super::super::filters::SearchFilters { + extensions: vec!["pdf".to_owned()], + ..super::super::filters::SearchFilters::default() + }; + let limited = backend.search(SearchRequest { + result_limit: Some(5), + ..SearchRequest::new("pcl5", &mut filters_limited) + }); + assert!( + limited + .rows + .iter() + .any(|row| row.name() == "pcl5_report.pdf"), + "'pcl5 --ext pdf --limit 5' must find the pdf — the extension \ + filter must be applied BEFORE the per-drive limit truncation \ + (got {:?})", + limited + .rows + .iter() + .map(DisplayRow::name) + .collect::>() + ); +} + // ── *.ext + `--hide-system` / `--hide-ads` fast-path regression pins ─ // // 2026-04-19 regression fix: the `is_ext_only()` gate that admits the diff --git a/crates/uffs-core/src/search/dispatch.rs b/crates/uffs-core/src/search/dispatch.rs index 0ba68bea5..06c0a5789 100644 --- a/crates/uffs-core/src/search/dispatch.rs +++ b/crates/uffs-core/src/search/dispatch.rs @@ -339,7 +339,9 @@ pub(super) fn dispatch_regex( .ok()?; let drive_results: Vec> = active_drives .par_iter() - .map(|drive| super::query::search_compact_drive_regex(drive, &compiled_re, limit)) + .map(|drive| { + super::query::search_compact_drive_regex(drive, &compiled_re, limit, search_filters) + }) .collect(); let mut rows: Vec = drive_results.into_iter().flatten().collect(); super::filters::apply_filter(&mut rows, filter_mode); @@ -377,7 +379,7 @@ pub(super) fn dispatch_trigram_or_tree( .par_iter() .map(|drive| { if is_path { - super::query::search_compact_drive_tree(drive, needle, limit) + super::query::search_compact_drive_tree(drive, needle, limit, search_filters) } else if is_prefix { // `is_prefix` was validated upstream via `is_prefix_pattern`; // re-extract the prefix and fall back to the generic scan if @@ -391,6 +393,7 @@ pub(super) fn dispatch_trigram_or_tree( case_sensitive, whole_word, match_path, + search_filters, ) }, |prefix| { @@ -399,6 +402,7 @@ pub(super) fn dispatch_trigram_or_tree( prefix, limit, case_sensitive, + search_filters, ) }, ) @@ -410,6 +414,7 @@ pub(super) fn dispatch_trigram_or_tree( case_sensitive, whole_word, match_path, + search_filters, ) } }) diff --git a/crates/uffs-core/src/search/query/mod.rs b/crates/uffs-core/src/search/query/mod.rs index 7310ef431..9d748e36a 100644 --- a/crates/uffs-core/src/search/query/mod.rs +++ b/crates/uffs-core/src/search/query/mod.rs @@ -180,11 +180,18 @@ pub(crate) fn search_compact_drive_regex( drive: &DriveCompactIndex, compiled_re: ®ex::Regex, limit: usize, + filters: &SearchFilters, ) -> Vec { let mut vp_buf = [0_u8; 4]; let volume_prefix = stack_volume_prefix(&mut vp_buf, drive.letter); let profile = *CACHE_PROFILE; + // Resolve the extension filter for THIS drive so record-level filters are + // applied BEFORE the `.take(limit)` cutoff (see `search_compact_drive`). + let mut local_filters = filters.clone(); + local_filters.resolve_ext_ids_for_drive(drive); + let mut filter_buf: Vec = Vec::with_capacity(256); + let t_match = std::time::Instant::now(); let match_indices: Vec = drive .records @@ -192,7 +199,9 @@ pub(crate) fn search_compact_drive_regex( .enumerate() .filter(|(_, rec)| { let name = rec.name(&drive.names); - !name.is_empty() && compiled_re.is_match(name) + !name.is_empty() + && compiled_re.is_match(name) + && local_filters.matches_record(rec, &drive.names, &mut filter_buf, drive.fold) }) .take(limit) .map(|(idx, _)| uffs_mft::len_to_u32(idx)) @@ -287,7 +296,19 @@ fn collect_match_indices( limit: usize, lower_buf: &mut Vec, matches: &dyn Fn(&str, &mut Vec) -> bool, + filters: &SearchFilters, ) -> Vec { + // Record-level filters (extension, size, dates, hide_system/ads, …) must + // be applied BEFORE the `limit` cutoff. Otherwise a small `--limit` fills + // the result with name-matches that the filter later removes, dropping + // valid matches that sit past the cutoff — the `pcl5 --ext pdf --limit 5` + // regression, where eight `.dll` matches preceded the one `.pdf`. The + // `matches_record` predicate is a no-op for an empty filter set, so an + // unfiltered search keeps its original behaviour. `filters.resolved_ext_ids` + // must already be resolved for THIS drive (see `search_compact_drive`). + let keep = |rec: &CompactRecord, name: &str, buf: &mut Vec| -> bool { + matches(name, buf) && filters.matches_record(rec, &drive.names, buf, drive.fold) + }; match candidates { None => { let mut out = Vec::new(); @@ -296,7 +317,7 @@ fn collect_match_indices( break; } let name = rec.name(&drive.names); - if matches(name, lower_buf) { + if keep(rec, name, lower_buf) { out.push(uffs_mft::len_to_u32(idx)); } } @@ -312,7 +333,7 @@ fn collect_match_indices( continue; }; let name = rec.name(&drive.names); - if matches(name, lower_buf) { + if keep(rec, name, lower_buf) { out.push(idx); } } @@ -330,11 +351,20 @@ pub(crate) fn search_compact_drive( case_sensitive: bool, whole_word: bool, match_path: bool, + filters: &SearchFilters, ) -> Vec { if needle.is_empty() { return Vec::new(); } + // Resolve the extension filter against THIS drive's interning table once, + // up front, so the per-record `matches_record` check inside + // `collect_match_indices` runs before the `limit` cutoff. Cloning keeps the + // parallel per-drive scan free of `&mut` contention (matches the + // numeric_top_n fast path). + let mut local_filters = filters.clone(); + local_filters.resolve_ext_ids_for_drive(drive); + let mut vp_buf = [0_u8; 4]; let volume_prefix = stack_volume_prefix(&mut vp_buf, drive.letter); let is_glob = needle.contains('*') || needle.contains('?'); @@ -402,8 +432,14 @@ pub(crate) fn search_compact_drive( let tri_count = candidates.as_ref().map_or(0, Vec::len); let t_match = std::time::Instant::now(); - let mut match_indices = - collect_match_indices(drive, candidates, limit, &mut fold_buf, &matches); + let mut match_indices = collect_match_indices( + drive, + candidates, + limit, + &mut fold_buf, + &matches, + &local_filters, + ); let match_ms = t_match.elapsed().as_millis(); let match_count = match_indices.len(); @@ -472,13 +508,30 @@ pub(crate) fn search_compact_drive_tree( drive: &DriveCompactIndex, pattern_lower: &str, limit: usize, + filters: &SearchFilters, ) -> Vec { let mut vp_buf = [0_u8; 4]; let volume_prefix = stack_volume_prefix(&mut vp_buf, drive.letter); let profile = *CACHE_PROFILE; + // Resolve the extension filter for THIS drive. When an `--ext` filter is + // active we must NOT let `tree_search` cap the path-walk at `limit` — the + // per-record filter below could otherwise drop matches that sit past the + // cutoff (the `--ext` limit-before-filter class). Over-fetch the full + // path-match set and re-apply `limit` after filtering. The gate is the + // single stable `extensions` field, so it never drifts out of sync with + // `matches_record`'s field set. + let mut local_filters = filters.clone(); + local_filters.resolve_ext_ids_for_drive(drive); + let scan_limit = if local_filters.extensions.is_empty() { + limit + } else { + usize::MAX + }; + let mut filter_buf: Vec = Vec::with_capacity(256); + let t_tree = std::time::Instant::now(); - let match_indices = tree::tree_search(drive, pattern_lower, limit); + let match_indices = tree::tree_search(drive, pattern_lower, scan_limit); let tree_ms = t_tree.elapsed().as_millis(); let match_count = match_indices.len(); @@ -493,6 +546,9 @@ pub(crate) fn search_compact_drive_tree( if name.is_empty() { return None; } + if !local_filters.matches_record(rec, &drive.names, &mut filter_buf, drive.fold) { + return None; + } let (path, path_malformed) = tree::resolve_path_cached_with_malformed( drive, record_idx as usize, @@ -510,6 +566,9 @@ pub(crate) fn search_compact_drive_tree( forensics, )) }) + // Re-apply the limit after filtering (the walk may have over-fetched + // when an extension filter was active — see `scan_limit` above). + .take(limit) .collect(); let resolve_ms = t_resolve.elapsed().as_millis(); diff --git a/crates/uffs-core/src/search/query/prefix_search.rs b/crates/uffs-core/src/search/query/prefix_search.rs index 5f5a19989..bc2b892bc 100644 --- a/crates/uffs-core/src/search/query/prefix_search.rs +++ b/crates/uffs-core/src/search/query/prefix_search.rs @@ -27,11 +27,18 @@ pub(crate) fn search_compact_drive_prefix( prefix: &str, limit: usize, case_sensitive: bool, + filters: &crate::search::filters::SearchFilters, ) -> Vec { let mut vp_buf = [0_u8; 4]; let volume_prefix = stack_volume_prefix(&mut vp_buf, drive.letter); let profile = *CACHE_PROFILE; + // Resolve the extension filter for THIS drive up front so the per-record + // filter runs BEFORE the `limit` cutoff (see `search_compact_drive`). + let mut local_filters = filters.clone(); + local_filters.resolve_ext_ids_for_drive(drive); + let mut filter_buf: Vec = Vec::with_capacity(256); + let t_tri = std::time::Instant::now(); // Get trigram candidates using first 3 chars of prefix. @@ -73,7 +80,9 @@ pub(crate) fn search_compact_drive_prefix( name_folded.starts_with(&prefix_folded) }; - if matches { + if matches + && local_filters.matches_record(rec, &drive.names, &mut filter_buf, drive.fold) + { match_indices.push(rec_idx); if match_indices.len() >= limit { break; diff --git a/crates/uffs-core/src/search/query_tests.rs b/crates/uffs-core/src/search/query_tests.rs index 900a916d0..89f17e860 100644 --- a/crates/uffs-core/src/search/query_tests.rs +++ b/crates/uffs-core/src/search/query_tests.rs @@ -132,7 +132,15 @@ fn build_large_drive(count: usize) -> DriveCompactIndex { #[test] fn search_compact_finds_file_by_name() { let drive = build_test_drive(); - let rows = search_compact_drive(&drive, "readme", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "readme", + 100, + false, + false, + false, + &SearchFilters::default(), + ); assert!( rows.iter().any(|row| row.name() == "readme.txt"), "search for 'readme' must find readme.txt" @@ -144,7 +152,15 @@ fn search_compact_finds_file_by_name() { #[test] fn display_row_fields_match_source_data() { let drive = build_test_drive(); - let rows = search_compact_drive(&drive, "readme", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "readme", + 100, + false, + false, + false, + &SearchFilters::default(), + ); let row = rows .iter() .find(|row| row.name() == "readme.txt") @@ -163,7 +179,15 @@ fn display_row_fields_match_source_data() { #[test] fn display_row_directory_has_tree_metrics() { let drive = build_test_drive(); - let rows = search_compact_drive(&drive, "projects", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "projects", + 100, + false, + false, + false, + &SearchFilters::default(), + ); let row = rows .iter() .find(|row| row.name() == "Projects") @@ -258,7 +282,15 @@ fn multi_drive_search_sort_by_size_desc() { #[test] fn display_row_path_includes_volume_prefix() { let drive = build_test_drive(); - let rows = search_compact_drive(&drive, "readme", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "readme", + 100, + false, + false, + false, + &SearchFilters::default(), + ); let row = rows .iter() .find(|row| row.name() == "readme.txt") @@ -327,8 +359,17 @@ fn prefix_search_matches_generic_glob_path() { // The trigram-accelerated prefix path must return exactly the same set of // rows as the ground-truth generic glob scan. `f000` matches f00000..f00099. let drive = build_large_drive(1_500); - let prefix_rows = search_compact_drive_prefix(&drive, "f000", 10_000, false); - let glob_rows = search_compact_drive(&drive, "f000*", 10_000, false, false, false); + let prefix_rows = + search_compact_drive_prefix(&drive, "f000", 10_000, false, &SearchFilters::default()); + let glob_rows = search_compact_drive( + &drive, + "f000*", + 10_000, + false, + false, + false, + &SearchFilters::default(), + ); let mut prefix_names: Vec<&str> = prefix_rows.iter().map(DisplayRow::name).collect(); let mut glob_names: Vec<&str> = glob_rows.iter().map(DisplayRow::name).collect(); @@ -345,7 +386,7 @@ fn prefix_search_matches_generic_glob_path() { #[test] fn prefix_search_respects_limit() { let drive = build_large_drive(1_500); - let rows = search_compact_drive_prefix(&drive, "f00", 25, false); + let rows = search_compact_drive_prefix(&drive, "f00", 25, false, &SearchFilters::default()); assert!( rows.len() <= 25, "prefix search must respect limit, got {}", @@ -360,7 +401,15 @@ fn large_glob_uses_parallel_resolve_with_correct_rows() { // parallel branch. Verify that path returns every match with intact paths // (no dropped, duplicated, or misordered rows from the chunk reduce). let drive = build_large_drive(9_000); - let rows = search_compact_drive(&drive, "f0*", 20_000, false, false, false); + let rows = search_compact_drive( + &drive, + "f0*", + 20_000, + false, + false, + false, + &SearchFilters::default(), + ); assert_eq!( rows.len(), 9_000, @@ -381,7 +430,7 @@ fn large_glob_uses_parallel_resolve_with_correct_rows() { fn regex_search_finds_matching_files() { let drive = build_test_drive(); let re = regex::Regex::new("(?i)readme").expect("valid regex"); - let rows = search_compact_drive_regex(&drive, &re, 100); + let rows = search_compact_drive_regex(&drive, &re, 100, &SearchFilters::default()); assert!( rows.iter().any(|row| row.name() == "readme.txt"), "regex 'readme' must find readme.txt" @@ -392,7 +441,7 @@ fn regex_search_finds_matching_files() { fn regex_search_no_match_returns_empty() { let drive = build_test_drive(); let re = regex::Regex::new("zzz_no_match[0-9]+").expect("valid regex"); - let rows = search_compact_drive_regex(&drive, &re, 100); + let rows = search_compact_drive_regex(&drive, &re, 100, &SearchFilters::default()); assert!(rows.is_empty(), "regex with no match must return empty"); } @@ -400,7 +449,7 @@ fn regex_search_no_match_returns_empty() { fn regex_search_respects_limit() { let drive = build_large_drive(500); let re = regex::Regex::new("f[0-9]+").expect("valid regex"); - let rows = search_compact_drive_regex(&drive, &re, 10); + let rows = search_compact_drive_regex(&drive, &re, 10, &SearchFilters::default()); assert!( rows.len() <= 10, "regex search must respect limit, got {}", @@ -477,7 +526,15 @@ fn ads_on_directory_display_row_is_not_directory() { let drive = build_ads_on_dir_drive(); // needle must be lowered — search_compact_drive expects pre-lowered for // case-insensitive - let rows = search_compact_drive(&drive, "myfolder:metadata", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "myfolder:metadata", + 100, + false, + false, + false, + &SearchFilters::default(), + ); let ads_row = rows .iter() .find(|row| row.name().contains(':')) @@ -494,7 +551,15 @@ fn normal_directory_display_row_is_directory() { let drive = build_ads_on_dir_drive(); // needle must be lowered — search_compact_drive expects pre-lowered for // case-insensitive - let rows = search_compact_drive(&drive, "myfolder", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "myfolder", + 100, + false, + false, + false, + &SearchFilters::default(), + ); let dir_row = rows .iter() .find(|row| row.name() == "MyFolder") @@ -512,7 +577,15 @@ fn normal_directory_display_row_is_directory() { #[test] fn case_sensitive_search_misses_wrong_case() { let drive = build_test_drive(); - let rows = search_compact_drive(&drive, "README", 100, true, false, false); + let rows = search_compact_drive( + &drive, + "README", + 100, + true, + false, + false, + &SearchFilters::default(), + ); assert!( !rows.iter().any(|row| row.name() == "readme.txt"), "case-sensitive 'README' must not match 'readme.txt'" @@ -524,7 +597,15 @@ fn case_insensitive_search_finds_any_case() { let drive = build_test_drive(); // needle must be pre-lowered for case-insensitive search (caller's // responsibility) - let rows = search_compact_drive(&drive, "readme", 100, false, false, false); + let rows = search_compact_drive( + &drive, + "readme", + 100, + false, + false, + false, + &SearchFilters::default(), + ); assert!( rows.iter().any(|row| row.name() == "readme.txt"), "case-insensitive 'readme' must match 'readme.txt'" @@ -535,7 +616,15 @@ fn case_insensitive_search_finds_any_case() { fn whole_word_search_exact_match() { let drive = build_test_drive(); // Whole-word with exact name (no extension) - let rows = search_compact_drive(&drive, "readme.txt", 100, false, true, false); + let rows = search_compact_drive( + &drive, + "readme.txt", + 100, + false, + true, + false, + &SearchFilters::default(), + ); assert!( rows.iter().any(|row| row.name() == "readme.txt"), "whole-word exact match must find readme.txt" diff --git a/crates/uffs-daemon/src/cache/journal_loop.rs b/crates/uffs-daemon/src/cache/journal_loop.rs index 7dca37b5c..f8ee86541 100644 --- a/crates/uffs-daemon/src/cache/journal_loop.rs +++ b/crates/uffs-daemon/src/cache/journal_loop.rs @@ -43,11 +43,17 @@ use alloc::sync::Arc; use core::time::Duration; -use std::time::Instant; use tokio::sync::watch; use uffs_mft::usn::FileChange; +mod triggers; + +pub(crate) use triggers::{ + ApplyTrigger, DEFAULT_APPLY_INTERVAL_MS, DEFAULT_SAVE_THRESHOLD_AGE, + DEFAULT_SAVE_THRESHOLD_EVENTS, SaveReason, SaveTrigger, +}; + /// Default poll interval for the per-shard journal loop (500 ms). /// /// Overridable at runtime via the `UFFS_USN_POLL_INTERVAL_MS` @@ -56,28 +62,6 @@ use uffs_mft::usn::FileChange; /// without recompiling. pub(crate) const DEFAULT_POLL_INTERVAL_MS: u64 = 500; -/// Default events-since-save threshold for triggering a background -/// compact-cache save (Phase 7 task 7.4). -/// -/// Sized to approximate the plan's "5% churn" criterion at the -/// typical 1.3 GB × ~7 M-record drive shape (`50_000` events ≈ 0.7% -/// churn, comfortably below 5%). Saving more frequently would -/// thrash the disk; less frequently would let the on-disk snapshot -/// drift far enough that a cold-boot replay window grows beyond -/// the cost of an incremental save. -pub(crate) const DEFAULT_SAVE_THRESHOLD_EVENTS: u64 = 50_000; - -/// Default time-since-save threshold for triggering a background -/// compact-cache save (Phase 7 task 7.4) — 5 minutes. -/// -/// Provides a wall-clock ceiling for how stale the on-disk snapshot -/// can get under low-churn workloads (where the events-threshold -/// would never fire on its own). Five minutes matches the cadence -/// of the existing Phase-5 `refresh_usn_for_warm_shards` global -/// tick so the persistence guarantee carries over to the per-shard -/// path without changing the operator-visible recovery window. -pub(crate) const DEFAULT_SAVE_THRESHOLD_AGE: Duration = Duration::from_mins(5); - /// Result of one [`JournalSource::poll`] call. /// /// Carries the change batch, the new cursor for the next call, and @@ -189,6 +173,29 @@ pub(crate) trait PatchSink: Send + Sync + 'static { cursor: u64, ); + /// Patch the in-memory body for `letter` on the apply cadence — the + /// near-live sibling of [`PatchSink::trigger_save`]. + /// + /// The loop calls this (via the per-shard [`ApplyTrigger`]) when + /// buffered changes exist and at least + /// [`JournalLoopConfig::apply_interval`] has elapsed since the last + /// apply / save. Production drains the same pending buffer + /// `trigger_save` uses and runs the surgical patch + body swap so + /// search sees the change — promptly after a quiet period (the + /// interval is already satisfied), or within one apply interval under + /// sustained churn — but **skips** the compact-cache disk write and + /// the cursor persist, which stay the rarer `trigger_save` tick's job. + /// + /// Unlike `trigger_save` this takes **no cursor**: the on-disk + /// cursor must only advance in lockstep with a real on-disk body + /// save, so the apply tick deliberately leaves it pinned. A cold + /// start re-replays the apply-only deltas from the last saved + /// cursor; the body patcher is idempotent on duplicate records, so + /// the re-replay is a no-op against the freshly loaded body. + /// + /// Fire-and-forget, same as `trigger_save`. + fn trigger_apply(&self, letter: uffs_mft::platform::DriveLetter); + /// Notify the sink that the USN journal for `letter` was /// detected to have wrapped (Phase 7 task 7.7). /// @@ -237,96 +244,6 @@ pub(crate) trait CursorStore: Send + Sync + 'static { fn store(&self, letter: uffs_mft::platform::DriveLetter, cursor: u64); } -/// Why a [`PatchSink::trigger_save`] call fired. -/// -/// Encoded so observability surfaces (logs, metrics) can -/// distinguish heavy-churn-driven saves from time-pressure-driven -/// saves; the production sink also passes this through to the -/// compact-cache writer for telemetry. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub(crate) enum SaveReason { - /// `events_since_save >= save_threshold_events` — lots of - /// churn has accumulated and the on-disk snapshot is - /// progressively stale. - EventsExceeded, - /// `Instant::now() - last_save_at >= save_threshold_age` — - /// time-pressure path for low-churn drives where the - /// events threshold would otherwise never fire. - AgeElapsed, -} - -/// Per-shard save-threshold state machine (Phase 7 task 7.4). -/// -/// Tracks the wall-clock time of the last save trigger and the -/// number of events accumulated since. Crossing either the -/// events- or age-threshold (with at least one event pending) -/// produces a [`SaveReason`] and resets both counters. Held -/// inside the [`JournalLoop`] so each per-shard task carries -/// its own independent counters. -#[derive(Debug)] -struct SaveTrigger { - /// Wall-clock time of the last save trigger (or, before any - /// triggers, the loop's spawn time). Compared against - /// `Instant::now()` to compute elapsed-since-last-save. - last_save_at: Instant, - /// Total events accumulated across [`Self::record`] calls - /// since the last save trigger. Compared against - /// `save_threshold_events` to fire the events-based save. - events_since_save: u64, -} - -impl SaveTrigger { - /// Construct a fresh trigger with `last_save_at` set to - /// `Instant::now()` (so the first age-based save can't fire - /// until at least `save_threshold_age` has elapsed since - /// loop spawn). - fn new() -> Self { - Self { - last_save_at: Instant::now(), - events_since_save: 0, - } - } - - /// Record `change_count` events accumulating toward the - /// events-based threshold. Saturating add so a runaway - /// drive can't wrap and silently miss the threshold. - const fn record(&mut self, change_count: u64) { - self.events_since_save = self.events_since_save.saturating_add(change_count); - } - - /// Evaluate the thresholds. - /// - /// **Returns** `Some(reason)` if a save should fire — and - /// resets both counters as a side effect (so the next - /// `evaluate` after a save starts from a clean slate). - /// Returns `None` when no threshold is crossed *or* when no - /// events are pending (zero-churn drives never produce - /// no-op saves). - fn evaluate( - &mut self, - save_threshold_events: u64, - save_threshold_age: Duration, - ) -> Option { - if self.events_since_save == 0 { - return None; - } - let now = Instant::now(); - let elapsed = now.saturating_duration_since(self.last_save_at); - let reason = if self.events_since_save >= save_threshold_events { - Some(SaveReason::EventsExceeded) - } else if elapsed >= save_threshold_age { - Some(SaveReason::AgeElapsed) - } else { - None - }; - if reason.is_some() { - self.last_save_at = now; - self.events_since_save = 0; - } - reason - } -} - /// Configuration for a single [`JournalLoop`] task. /// /// Carries the tuning knobs the production loop reads from env @@ -351,6 +268,12 @@ pub(crate) struct JournalLoopConfig { /// when at least one event is pending. Default /// [`DEFAULT_SAVE_THRESHOLD_AGE`] (5 min). pub(crate) save_threshold_age: Duration, + /// Search-freshness apply cadence — decoupled from the disk-save + /// thresholds above. When buffered changes exist and this long + /// has elapsed since the last apply / save, the loop patches the + /// in-memory body via [`PatchSink::trigger_apply`] so the change + /// becomes searchable. Default [`DEFAULT_APPLY_INTERVAL_MS`] (30 s). + pub(crate) apply_interval: Duration, } impl Default for JournalLoopConfig { @@ -360,6 +283,62 @@ impl Default for JournalLoopConfig { initial_cursor: 0, save_threshold_events: DEFAULT_SAVE_THRESHOLD_EVENTS, save_threshold_age: DEFAULT_SAVE_THRESHOLD_AGE, + apply_interval: Duration::from_millis(DEFAULT_APPLY_INTERVAL_MS), + } + } +} + +/// Env var overriding [`JournalLoopConfig::poll_interval`] (milliseconds). +pub(crate) const POLL_INTERVAL_ENV: &str = "UFFS_USN_POLL_INTERVAL_MS"; + +/// Env var overriding [`JournalLoopConfig::apply_interval`] (milliseconds). +pub(crate) const APPLY_INTERVAL_ENV: &str = "UFFS_USN_APPLY_INTERVAL_MS"; + +impl JournalLoopConfig { + /// Build the production config: [`Self::default`] with the two + /// millisecond-valued env overrides applied when present + parseable. + /// + /// `UFFS_USN_POLL_INTERVAL_MS` dials the poll cadence (long + /// documented; this is the wire-up) and `UFFS_USN_APPLY_INTERVAL_MS` + /// dials the search-freshness apply cadence. A missing, empty, or + /// unparseable value leaves the corresponding default untouched and + /// warn-logs the bad input so a typo is visible rather than silently + /// ignored. A `0` is accepted verbatim (apply / poll on every tick) + /// — useful for tests and aggressive soak runs. + #[must_use] + pub(crate) fn from_env() -> Self { + let mut config = Self::default(); + if let Some(ms) = env_millis(POLL_INTERVAL_ENV) { + config.poll_interval = Duration::from_millis(ms); + } + if let Some(ms) = env_millis(APPLY_INTERVAL_ENV) { + config.apply_interval = Duration::from_millis(ms); + } + config + } +} + +/// Read `name` from the environment as a `u64` millisecond count. +/// +/// Returns `None` (caller keeps the default) when the var is unset, +/// empty, or non-numeric; a non-numeric value also warn-logs so the +/// operator sees the typo instead of silently getting the default. +fn env_millis(name: &str) -> Option { + let raw = std::env::var(name).ok()?; + let trimmed = raw.trim(); + if trimmed.is_empty() { + return None; + } + match trimmed.parse::() { + Ok(ms) => Some(ms), + Err(parse_err) => { + tracing::warn!( + env = name, + value = trimmed, + error = %parse_err, + "Ignoring non-numeric journal-loop interval override; using default" + ); + None } } } @@ -396,6 +375,12 @@ pub(crate) struct JournalLoop { /// Mutated on every non-empty tick by [`SaveTrigger::record`] /// + [`SaveTrigger::evaluate`]. save_trigger: SaveTrigger, + /// Per-shard apply-cadence state — the search-freshness sibling + /// of `save_trigger`. Mutated on every non-empty tick by + /// [`ApplyTrigger::record`] + [`ApplyTrigger::evaluate`], and + /// reset via [`ApplyTrigger::reset_after_save`] when a save tick + /// subsumes the apply. + apply_trigger: ApplyTrigger, /// Last `journal_id` observed from a non-zero-id poll /// (Phase 7 task 7.7 wrap detection). `None` until the first /// non-zero `journal_id` is observed; transitions to @@ -426,6 +411,7 @@ impl JournalLoop { cancel_rx, config, save_trigger: SaveTrigger::new(), + apply_trigger: ApplyTrigger::new(), last_journal_id: None, } } @@ -508,8 +494,8 @@ impl JournalLoop { cursor, &result.changes, &mut self.save_trigger, - self.config.save_threshold_events, - self.config.save_threshold_age, + &mut self.apply_trigger, + &self.config, ); } } @@ -675,44 +661,86 @@ fn log_poll_failure( } } -/// Apply the post-poll change batch to `sink`, or trace-log the -/// no-op tick when `changes` is empty. +/// Buffer the post-poll change batch into `sink`, record it into both +/// cadence triggers, and fire whichever is due via [`fire_due_cadence`] +/// — or trace-log the no-op tick when `changes` is empty. /// -/// On a non-empty tick, also: (a) records the event count into -/// `save_trigger`, (b) evaluates the save thresholds, and (c) -/// fires [`PatchSink::trigger_save`] (passing `cursor` so the sink can -/// persist it in lockstep with the body save) when a threshold crosses. +/// On an idle drive neither cadence fires (both event counters stay at +/// zero), so a quiescent volume costs nothing beyond the poll itself. fn process_tick( sink: &dyn PatchSink, letter: uffs_mft::platform::DriveLetter, cursor: u64, changes: &[FileChange], save_trigger: &mut SaveTrigger, - save_threshold_events: u64, - save_threshold_age: Duration, + apply_trigger: &mut ApplyTrigger, + config: &JournalLoopConfig, ) { if changes.is_empty() { tracing::trace!(drive = %letter, "Journal poll: no changes"); return; } let accepted = sink.accept(letter, changes); - save_trigger.record(changes.len() as u64); - if let Some(reason) = save_trigger.evaluate(save_threshold_events, save_threshold_age) { + let change_count = changes.len() as u64; + save_trigger.record(change_count); + apply_trigger.record(change_count); + + fire_due_cadence(sink, letter, cursor, save_trigger, apply_trigger, config); + + tracing::debug!( + drive = %letter, + accepted, + change_count = changes.len(), + cursor, + "Journal poll: applied tick" + ); +} + +/// Fire whichever cadence is due this tick — **at most one**, since a +/// save drains + applies the same buffer an apply would (a save +/// subsumes an apply). +/// +/// * **Save tick** (rare — 50k events / 5 min): fire +/// [`PatchSink::trigger_save`] (passing `cursor` so the sink persists it in +/// lockstep with the on-disk body), then [`ApplyTrigger::reset_after_save`] +/// so the loop doesn't redundantly re-apply the just-drained buffer. +/// * **Apply tick** (default 30 s): when no save fired, fire +/// [`PatchSink::trigger_apply`] so the in-memory body (and search) goes +/// near-live without the disk write. +/// +/// Extracted from [`process_tick`] so each function stays under +/// clippy's strict-gate cognitive-complexity ceiling. +fn fire_due_cadence( + sink: &dyn PatchSink, + letter: uffs_mft::platform::DriveLetter, + cursor: u64, + save_trigger: &mut SaveTrigger, + apply_trigger: &mut ApplyTrigger, + config: &JournalLoopConfig, +) { + if let Some(reason) = + save_trigger.evaluate(config.save_threshold_events, config.save_threshold_age) + { + // A save drains + applies the buffer AND writes the body to + // disk + persists the cursor — so it subsumes the apply tick. sink.trigger_save(letter, reason, cursor); + apply_trigger.reset_after_save(); tracing::info!( drive = %letter, ?reason, cursor, "Journal poll: triggered background compact-cache save" ); + } else if apply_trigger.evaluate(config.apply_interval) { + // No save this tick: patch the in-memory body so search sees + // the change within the apply interval, leaving disk + // persistence to a later save tick. + sink.trigger_apply(letter); + tracing::debug!( + drive = %letter, + "Journal poll: triggered near-live body apply" + ); } - tracing::debug!( - drive = %letter, - accepted, - change_count = changes.len(), - cursor, - "Journal poll: applied tick" - ); } /// Handle returned by [`spawn_journal_loop`] for cancellation + diff --git a/crates/uffs-daemon/src/cache/journal_loop/sources.rs b/crates/uffs-daemon/src/cache/journal_loop/sources.rs index 733cb4242..c420c27a6 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/sources.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/sources.rs @@ -147,9 +147,18 @@ impl JournalSource for WindowsJournalSource { let (records, next_usn) = uffs_mft::usn::read_usn_journal(self.drive, info.journal_id, start_usn)?; let aggregated = uffs_mft::usn::aggregate_changes(&records); - let changes: Vec = aggregated.into_values().collect(); + let mut changes: Vec = aggregated.into_values().collect(); let next_cursor = u64::try_from(next_usn.raw()).unwrap_or(u64::MAX); + // Backfill real size/timestamps/flags via a targeted MFT read. USN + // records carry only name+parent, so a create/rename would otherwise + // land with size 0 and zero timestamps. This runs HERE (in `poll`, + // on the spawn_blocking thread) — before the registry write-lock is + // taken in `accept` — so it never lengthens the lock hold or touches + // the query path. Best-effort: any failure leaves `meta = None` and + // the records keep their (current) zeroed metrics. + Self::backfill_metadata(self.drive, &mut changes); + Ok(JournalPollResult { changes, next_cursor, @@ -158,6 +167,84 @@ impl JournalSource for WindowsJournalSource { } } +#[cfg(windows)] +impl WindowsJournalSource { + /// Upper bound on targeted MFT reads per poll. A bulk operation (e.g. + /// unzipping thousands of files) can produce a large change set in one + /// 500 ms window; cap the read so a single poll can't stall the loop. + /// Records past the cap keep `meta = None` for this poll and are + /// backfilled on a subsequent one (or by the next full re-warm). + const MAX_TARGETED_READS_PER_POLL: usize = 4096; + + /// Issue one batched targeted MFT read for the created/renamed FRSes in + /// `changes` and attach the recovered [`uffs_mft::usn::RecordMeta`] to + /// each. Deletes need no metadata and are skipped. + fn backfill_metadata( + drive: uffs_mft::platform::DriveLetter, + changes: &mut [uffs_mft::usn::FileChange], + ) { + // Collect the FRSes that need real metadata (creates + renames). + let frs_list: Vec = changes + .iter() + .filter(|change| change.created || change.renamed) + .take(Self::MAX_TARGETED_READS_PER_POLL) + .map(|change| change.frs.raw()) + .collect(); + if frs_list.is_empty() { + return; + } + let Some(scratch) = Self::read_targeted_records(drive, &frs_list) else { + return; + }; + + // Attach the recovered metadata. Representation matches CompactRecord + // exactly (i64 µs timestamps, raw NTFS flags), so it copies straight. + for change in changes.iter_mut() { + if !(change.created || change.renamed) { + continue; + } + if let Some(record) = scratch.find(change.frs) { + change.meta = Some(uffs_mft::usn::RecordMeta { + size: record.first_stream.size.length, + allocated: record.first_stream.size.allocated, + created: record.stdinfo.created, + modified: record.stdinfo.modified, + accessed: record.stdinfo.accessed, + flags: record.stdinfo.flags, + }); + } + } + } + + /// Open the volume (auto-adopting the broker handle when non-elevated — + /// the same path the USN read already uses) and read `frs_list` into a + /// scratch [`MftIndex`](uffs_mft::index::MftIndex). Best-effort: any + /// failure returns `None` (debug-logged), leaving callers' `meta` empty. + fn read_targeted_records( + drive: uffs_mft::platform::DriveLetter, + frs_list: &[u64], + ) -> Option { + let handle = match uffs_mft::platform::VolumeHandle::open(drive) { + Ok(handle) => handle, + Err(err) => { + tracing::debug!(drive = %drive, error = %err, "usn backfill: volume open failed"); + return None; + } + }; + let mut scratch = uffs_mft::index::MftIndex::new(drive); + match uffs_mft::usn::read_targeted_frs_records(&handle, &mut scratch, frs_list) { + Ok(read) => { + tracing::debug!(drive = %drive, requested = frs_list.len(), read, "usn backfill: targeted MFT reads complete"); + Some(scratch) + } + Err(err) => { + tracing::debug!(drive = %drive, error = %err, "usn backfill: targeted reads failed"); + None + } + } + } +} + // ─── No-op cursor store ───────────────────────────────────────────────────── /// Always-empty cursor store: `load` returns 0, `store` is a diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests.rs b/crates/uffs-daemon/src/cache/journal_loop/tests.rs index 76cc28cf0..02b1f38cc 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/tests.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/tests.rs @@ -45,6 +45,7 @@ use super::{ CursorStore, JournalLoopConfig, JournalPollResult, JournalSource, PatchSink, SaveReason, }; +mod apply_cadence; mod backoff; mod basics; mod integration; @@ -175,6 +176,11 @@ struct RecordingSink { /// a successful body save. Lets the loop-level tests assert the /// cursor reaches the sink without coupling to persistence. save_cursors: Mutex>, + /// One entry per `trigger_apply()` call: `letter`. The + /// apply-cadence surface — lets tests assert the near-live body + /// apply fires on the short interval (and that a save tick + /// suppresses a redundant apply on the same buffer). + apply_calls: Mutex>, /// One entry per `journal_wrapped()` call: `letter`. /// Phase 7-D surface — lets tests assert the wrap-detection /// state machine fires when `journal_id` changes between @@ -192,6 +198,7 @@ impl RecordingSink { calls: Mutex::new(Vec::new()), save_calls: Mutex::new(Vec::new()), save_cursors: Mutex::new(Vec::new()), + apply_calls: Mutex::new(Vec::new()), wrap_calls: Mutex::new(Vec::new()), accept_outcome: Mutex::new(true), } @@ -209,6 +216,10 @@ impl RecordingSink { lock_or_recover(&self.save_cursors).clone() } + fn apply_calls(&self) -> Vec { + lock_or_recover(&self.apply_calls).clone() + } + fn wrap_calls(&self) -> Vec { lock_or_recover(&self.wrap_calls).clone() } @@ -230,6 +241,10 @@ impl PatchSink for RecordingSink { lock_or_recover(&self.save_cursors).push(cursor); } + fn trigger_apply(&self, letter: uffs_mft::platform::DriveLetter) { + lock_or_recover(&self.apply_calls).push(letter); + } + fn journal_wrapped(&self, letter: uffs_mft::platform::DriveLetter) { lock_or_recover(&self.wrap_calls).push(letter); } @@ -313,6 +328,10 @@ fn fast_config() -> JournalLoopConfig { initial_cursor: 0, save_threshold_events: u64::MAX, save_threshold_age: Duration::from_hours(24), + // Disabled by default so the generic tick / cancel / cursor + // tests don't accidentally fire an apply; the apply-cadence + // tests override this with a short interval. + apply_interval: Duration::from_hours(24), } } diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests/apply_cadence.rs b/crates/uffs-daemon/src/cache/journal_loop/tests/apply_cadence.rs new file mode 100644 index 000000000..1e8b49581 --- /dev/null +++ b/crates/uffs-daemon/src/cache/journal_loop/tests/apply_cadence.rs @@ -0,0 +1,241 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! Apply-cadence tests — the search-freshness path that decouples the +//! in-memory body patch from the rare compact-cache disk save. +//! +//! Before this split, a newly created / renamed / deleted file stayed +//! invisible to search until a [`super::super::SaveTrigger`] threshold +//! fired (50k events / 5 min): the live USN path buffered changes in +//! `accept` but only patched the searchable body inside `trigger_save`. +//! The [`super::super::ApplyTrigger`] closes that gap — when buffered +//! churn exists and [`super::super::JournalLoopConfig::apply_interval`] +//! has elapsed, the loop fires [`PatchSink::trigger_apply`] to patch +//! the body without the disk write. +//! +//! These tests pin three contracts: +//! +//! 1. **Apply fires without a save** — on the short interval the body is +//! patched (`trigger_apply`) while no `trigger_save` fires. +//! 2. **A save subsumes the apply** — when a save threshold crosses on the same +//! tick, `trigger_save` fires and the redundant `trigger_apply` is +//! suppressed (the save already drained + applied the buffer). +//! 3. **The [`ApplyTrigger`] state machine** — churn-gated, interval- +//! rate-limited, and reset by a save. +//! +//! [`ApplyTrigger`]: super::super::ApplyTrigger +//! [`PatchSink::trigger_apply`]: super::super::PatchSink::trigger_apply + +use alloc::sync::Arc; +use core::time::Duration; + +use super::super::{ + ApplyTrigger, JournalLoopConfig, JournalSource, PatchSink, SaveTrigger, process_tick, + spawn_journal_loop, +}; +use super::{ + CONVERGENCE_DEADLINE, FakeJournalSource, RecordingSink, null_cursor_store, one_change, wait_for, +}; + +/// A [`JournalLoopConfig`] whose apply cadence fires on every tick with +/// churn (`apply_interval == 0`) and whose save thresholds are pinned +/// out of reach, so a `process_tick` exercises the apply path in +/// isolation from any save. +fn apply_only_config() -> JournalLoopConfig { + JournalLoopConfig { + save_threshold_events: u64::MAX, + save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::ZERO, + ..JournalLoopConfig::default() + } +} + +#[test] +fn apply_tick_patches_body_without_save() { + let sink = RecordingSink::new(); + let mut save_trigger = SaveTrigger::new(); + let mut apply_trigger = ApplyTrigger::new(); + let config = apply_only_config(); + + let changes = [one_change(10), one_change(11)]; + process_tick( + &sink as &dyn PatchSink, + uffs_mft::platform::DriveLetter::C, + 100, + &changes, + &mut save_trigger, + &mut apply_trigger, + &config, + ); + + // The body was patched via the apply path... + assert_eq!( + sink.apply_calls().as_slice(), + &[uffs_mft::platform::DriveLetter::C], + "an apply tick with buffered churn must fire trigger_apply exactly once", + ); + // ...and crucially NOT via a disk save (the whole point of the split). + assert!( + sink.save_calls().is_empty(), + "the apply tick must not fire a compact-cache save; got {:?}", + sink.save_calls(), + ); +} + +#[test] +fn save_tick_suppresses_redundant_apply() { + let sink = RecordingSink::new(); + let mut save_trigger = SaveTrigger::new(); + let mut apply_trigger = ApplyTrigger::new(); + // Save crosses on the first event; apply would otherwise fire too + // (interval 0), so this proves the mutual exclusion: the save + // subsumes the apply (it drained + applied the same buffer). + let config = JournalLoopConfig { + save_threshold_events: 1, + save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::ZERO, + ..JournalLoopConfig::default() + }; + + let changes = [one_change(10), one_change(11)]; + process_tick( + &sink as &dyn PatchSink, + uffs_mft::platform::DriveLetter::C, + 100, + &changes, + &mut save_trigger, + &mut apply_trigger, + &config, + ); + + assert_eq!( + sink.save_calls().len(), + 1, + "the save threshold crossed, so exactly one save must fire", + ); + assert!( + sink.apply_calls().is_empty(), + "a save subsumes the apply — no redundant trigger_apply on the same tick; got {:?}", + sink.apply_calls(), + ); +} + +#[test] +fn idle_tick_fires_neither_apply_nor_save() { + let sink = RecordingSink::new(); + let mut save_trigger = SaveTrigger::new(); + let mut apply_trigger = ApplyTrigger::new(); + let config = apply_only_config(); + + // Empty change batch — a no-op poll on a quiescent drive. + process_tick( + &sink as &dyn PatchSink, + uffs_mft::platform::DriveLetter::C, + 100, + &[], + &mut save_trigger, + &mut apply_trigger, + &config, + ); + + assert!(sink.apply_calls().is_empty(), "idle tick must not apply"); + assert!(sink.save_calls().is_empty(), "idle tick must not save"); +} + +#[tokio::test] +async fn loop_applies_body_near_live_without_saving() { + let source = Arc::new(FakeJournalSource::new()); + let sink = Arc::new(RecordingSink::new()); + + // A single create-shaped batch. With a sub-tick apply interval and + // out-of-reach save thresholds, the loop must patch the body (apply) + // within a couple of ticks while never firing a disk save. + source.enqueue_changes(vec![one_change(42)], 100); + + let config = JournalLoopConfig { + poll_interval: Duration::from_millis(5), + save_threshold_events: u64::MAX, + save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::ZERO, + ..JournalLoopConfig::default() + }; + let handle = spawn_journal_loop( + uffs_mft::platform::DriveLetter::C, + Arc::clone(&source) as Arc, + Arc::clone(&sink) as Arc, + null_cursor_store(), + config, + ); + + let sink_for_pred = Arc::clone(&sink); + let applied = wait_for(move || !sink_for_pred.apply_calls().is_empty()).await; + let join = handle.cancel(); + drop(tokio::time::timeout(CONVERGENCE_DEADLINE, join).await); + + assert!( + applied, + "the loop must fire a near-live apply within the convergence deadline", + ); + assert!( + sink.save_calls().is_empty(), + "near-live apply must not trigger a compact-cache save; got {:?}", + sink.save_calls(), + ); +} + +#[test] +fn apply_trigger_requires_churn() { + let mut trigger = ApplyTrigger::new(); + // No events recorded — even a zero interval must not fire. + assert!( + !trigger.evaluate(Duration::ZERO), + "an apply must not fire without buffered churn", + ); +} + +#[test] +fn apply_trigger_fires_once_per_interval_then_resets() { + let mut trigger = ApplyTrigger::new(); + trigger.record(3); + assert!( + trigger.evaluate(Duration::ZERO), + "churn present + interval elapsed must fire", + ); + // The fire reset the churn counter, so a second evaluate with no + // new events must not fire again. + assert!( + !trigger.evaluate(Duration::ZERO), + "evaluate must reset the churn counter after firing", + ); +} + +#[test] +fn apply_trigger_respects_interval() { + let mut trigger = ApplyTrigger::new(); + trigger.record(3); + // Interval far in the future — churn exists but the rate limit + // holds the apply back. + assert!( + !trigger.evaluate(Duration::from_hours(1)), + "an apply must wait for the interval even with churn pending", + ); + // The churn is retained (no reset on a held-back tick), so once the + // interval is satisfied the apply fires. + assert!( + trigger.evaluate(Duration::ZERO), + "retained churn must fire once the interval is satisfied", + ); +} + +#[test] +fn reset_after_save_clears_pending_churn() { + let mut trigger = ApplyTrigger::new(); + trigger.record(5); + // A save tick drained + applied the buffer; the apply trigger must + // forget the churn so it doesn't redundantly re-apply. + trigger.reset_after_save(); + assert!( + !trigger.evaluate(Duration::ZERO), + "reset_after_save must clear the churn guard", + ); +} diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests/integration.rs b/crates/uffs-daemon/src/cache/journal_loop/tests/integration.rs index b9e16c8ff..6ef794ef1 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/tests/integration.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/tests/integration.rs @@ -96,6 +96,7 @@ async fn ten_thousand_events_end_to_end() { initial_cursor: 0, save_threshold_events: SAVE_THRESHOLD_EVENTS, save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests/save_log_message.rs b/crates/uffs-daemon/src/cache/journal_loop/tests/save_log_message.rs index df7073833..b168d2896 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/tests/save_log_message.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/tests/save_log_message.rs @@ -43,7 +43,9 @@ use core::time::Duration; -use super::super::{PatchSink, SaveReason, SaveTrigger, process_tick}; +use super::super::{ + ApplyTrigger, JournalLoopConfig, PatchSink, SaveReason, SaveTrigger, process_tick, +}; use super::{RecordingSink, one_change}; use crate::index::tests::tracing_capture::{CapturedEvent, EventLog}; @@ -71,21 +73,30 @@ fn compact_cache_save_log_message_pins_string_target_and_level() { tracing::callsite::rebuild_interest_cache(); let sink = RecordingSink::new(); - let mut trigger = SaveTrigger::new(); + let mut save_trigger = SaveTrigger::new(); + let mut apply_trigger = ApplyTrigger::new(); // Force the events-threshold path: three changes against a // threshold of one crosses on the first evaluate() call. // Age threshold set generously so it can't be the path that - // fires (we want a deterministic `EventsExceeded` reason). + // fires (we want a deterministic `EventsExceeded` reason), and + // the apply interval is disabled so the save path is the only one + // that can fire on this tick. + let config = JournalLoopConfig { + save_threshold_events: 1, // tight — crosses on the first evaluate + save_threshold_age: Duration::from_hours(1), // generous + apply_interval: Duration::from_hours(1), // disabled for this test + ..JournalLoopConfig::default() + }; let changes = [one_change(10), one_change(11), one_change(12)]; process_tick( &sink as &dyn PatchSink, uffs_mft::platform::DriveLetter::C, 100, // cursor &changes, - &mut trigger, - 1, // save_threshold_events — tight - Duration::from_hours(1), // save_threshold_age — generous + &mut save_trigger, + &mut apply_trigger, + &config, ); // The sink saw the trigger_save callback once with the expected diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests/thresholds.rs b/crates/uffs-daemon/src/cache/journal_loop/tests/thresholds.rs index a19b93702..bc501e340 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/tests/thresholds.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/tests/thresholds.rs @@ -42,6 +42,7 @@ async fn events_threshold_triggers_save() { initial_cursor: 0, save_threshold_events: 5, save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, @@ -91,6 +92,7 @@ async fn age_threshold_triggers_save_with_pending_events() { initial_cursor: 0, save_threshold_events: u64::MAX, save_threshold_age: Duration::from_millis(30), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, @@ -150,6 +152,7 @@ async fn zero_events_drive_does_not_trigger_save() { initial_cursor: 0, save_threshold_events: 1, save_threshold_age: Duration::from_millis(20), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, @@ -202,6 +205,7 @@ async fn counter_resets_after_save() { initial_cursor: 0, save_threshold_events: 5, save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, diff --git a/crates/uffs-daemon/src/cache/journal_loop/tests/wrap_and_persistence.rs b/crates/uffs-daemon/src/cache/journal_loop/tests/wrap_and_persistence.rs index 4f5bb746f..b610b4c6b 100644 --- a/crates/uffs-daemon/src/cache/journal_loop/tests/wrap_and_persistence.rs +++ b/crates/uffs-daemon/src/cache/journal_loop/tests/wrap_and_persistence.rs @@ -90,6 +90,7 @@ async fn cursor_handed_to_sink_on_save_trigger() { initial_cursor: 0, save_threshold_events: 5, save_threshold_age: Duration::from_hours(24), + apply_interval: Duration::from_hours(24), }; let handle = spawn_journal_loop( uffs_mft::platform::DriveLetter::C, diff --git a/crates/uffs-daemon/src/cache/journal_loop/triggers.rs b/crates/uffs-daemon/src/cache/journal_loop/triggers.rs new file mode 100644 index 000000000..0328d5ac4 --- /dev/null +++ b/crates/uffs-daemon/src/cache/journal_loop/triggers.rs @@ -0,0 +1,243 @@ +// SPDX-License-Identifier: MPL-2.0 +// Copyright (c) 2025-2026 SKY, LLC. + +//! Per-shard cadence state machines for the journal loop. +//! +//! Two independent counters govern when the loop drains its pending +//! change buffer, deliberately decoupled so search freshness and disk +//! persistence run on their own clocks: +//! +//! * [`SaveTrigger`] — the rare, expensive **disk save** (default 50k events / +//! 5 min). Crossing either threshold patches the body AND persists the +//! compact cache + cursor. +//! * [`ApplyTrigger`] — the more frequent, disk-free **in-memory apply** +//! (default 30 s). Buffered churn plus an elapsed interval patches the body +//! so a freshly created / renamed / deleted file becomes searchable, without +//! touching disk. +//! +//! A save subsumes an apply (it drains + applies the same buffer), so +//! the loop fires at most one of the two per tick; see +//! [`super::process_tick`]. Extracted from `journal_loop.rs` to keep +//! that file under the workspace file-size policy while keeping the two +//! cadence machines (and the threshold constants they default from) +//! together as one auditable unit. + +use core::time::Duration; +use std::time::Instant; + +/// Default events-since-save threshold for triggering a background +/// compact-cache save (Phase 7 task 7.4). +/// +/// Sized to approximate the plan's "5% churn" criterion at the +/// typical 1.3 GB × ~7 M-record drive shape (`50_000` events ≈ 0.7% +/// churn, comfortably below 5%). Saving more frequently would +/// thrash the disk; less frequently would let the on-disk snapshot +/// drift far enough that a cold-boot replay window grows beyond +/// the cost of an incremental save. +pub(crate) const DEFAULT_SAVE_THRESHOLD_EVENTS: u64 = 50_000; + +/// Default time-since-save threshold for triggering a background +/// compact-cache save (Phase 7 task 7.4) — 5 minutes. +/// +/// Provides a wall-clock ceiling for how stale the on-disk snapshot +/// can get under low-churn workloads (where the events-threshold +/// would never fire on its own). Five minutes matches the cadence +/// of the existing Phase-5 `refresh_usn_for_warm_shards` global +/// tick so the persistence guarantee carries over to the per-shard +/// path without changing the operator-visible recovery window. +pub(crate) const DEFAULT_SAVE_THRESHOLD_AGE: Duration = Duration::from_mins(5); + +/// Default apply interval for the per-shard journal loop — 30 seconds. +/// +/// This is the **search-freshness** knob, decoupled from the much +/// rarer disk-save cadence above. When buffered changes exist and at +/// least this long has elapsed since the last apply / save, the loop +/// patches the in-memory body (via [`super::PatchSink::trigger_apply`]) +/// so a freshly created / renamed / deleted file becomes searchable — +/// instead of waiting up to [`DEFAULT_SAVE_THRESHOLD_AGE`] (5 min) for a +/// disk-save tick to also apply it. +/// +/// Thirty seconds is tuned for the per-apply rebuild cost: each apply +/// clones the body and rebuilds the children / trigram / extension +/// indexes (~600 ms on a 7M-record drive, **independent of batch +/// size**). On a filesystem with constant churn that throttles the +/// rebuild to background noise (~600 ms / 30 s ≈ 2 % of one core per +/// active drive) instead of a continuous drag. Crucially it does *not* +/// blunt the common case: because the trigger fires as soon as the +/// interval has elapsed *since the last apply*, the first change after +/// any quiet period is applied within a poll or two — only sustained, +/// back-to-back churn is batched onto the 30 s cadence. On an idle +/// drive no apply fires at all (the event counter stays at zero). +/// +/// Overridable at runtime via the `UFFS_USN_APPLY_INTERVAL_MS` +/// environment variable, mirroring `UFFS_USN_POLL_INTERVAL_MS`, so soak +/// tests and latency-sensitive setups can dial freshness up or down +/// without recompiling. +pub(crate) const DEFAULT_APPLY_INTERVAL_MS: u64 = 30_000; + +/// Why a [`super::PatchSink::trigger_save`] call fired. +/// +/// Encoded so observability surfaces (logs, metrics) can +/// distinguish heavy-churn-driven saves from time-pressure-driven +/// saves; the production sink also passes this through to the +/// compact-cache writer for telemetry. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum SaveReason { + /// `events_since_save >= save_threshold_events` — lots of + /// churn has accumulated and the on-disk snapshot is + /// progressively stale. + EventsExceeded, + /// `Instant::now() - last_save_at >= save_threshold_age` — + /// time-pressure path for low-churn drives where the + /// events threshold would otherwise never fire. + AgeElapsed, +} + +/// Per-shard save-threshold state machine (Phase 7 task 7.4). +/// +/// Tracks the wall-clock time of the last save trigger and the +/// number of events accumulated since. Crossing either the +/// events- or age-threshold (with at least one event pending) +/// produces a [`SaveReason`] and resets both counters. Held +/// inside the [`super::JournalLoop`] so each per-shard task carries +/// its own independent counters. +#[derive(Debug)] +pub(crate) struct SaveTrigger { + /// Wall-clock time of the last save trigger (or, before any + /// triggers, the loop's spawn time). Compared against + /// `Instant::now()` to compute elapsed-since-last-save. + last_save_at: Instant, + /// Total events accumulated across [`Self::record`] calls + /// since the last save trigger. Compared against + /// `save_threshold_events` to fire the events-based save. + events_since_save: u64, +} + +impl SaveTrigger { + /// Construct a fresh trigger with `last_save_at` set to + /// `Instant::now()` (so the first age-based save can't fire + /// until at least `save_threshold_age` has elapsed since + /// loop spawn). + pub(super) fn new() -> Self { + Self { + last_save_at: Instant::now(), + events_since_save: 0, + } + } + + /// Record `change_count` events accumulating toward the + /// events-based threshold. Saturating add so a runaway + /// drive can't wrap and silently miss the threshold. + pub(super) const fn record(&mut self, change_count: u64) { + self.events_since_save = self.events_since_save.saturating_add(change_count); + } + + /// Evaluate the thresholds. + /// + /// **Returns** `Some(reason)` if a save should fire — and + /// resets both counters as a side effect (so the next + /// `evaluate` after a save starts from a clean slate). + /// Returns `None` when no threshold is crossed *or* when no + /// events are pending (zero-churn drives never produce + /// no-op saves). + pub(super) fn evaluate( + &mut self, + save_threshold_events: u64, + save_threshold_age: Duration, + ) -> Option { + if self.events_since_save == 0 { + return None; + } + let now = Instant::now(); + let elapsed = now.saturating_duration_since(self.last_save_at); + let reason = if self.events_since_save >= save_threshold_events { + Some(SaveReason::EventsExceeded) + } else if elapsed >= save_threshold_age { + Some(SaveReason::AgeElapsed) + } else { + None + }; + if reason.is_some() { + self.last_save_at = now; + self.events_since_save = 0; + } + reason + } +} + +/// Per-shard apply-cadence state machine — the search-freshness +/// counterpart to [`SaveTrigger`]. +/// +/// Where `SaveTrigger` governs the rare, expensive disk save (50k +/// events / 5 min), this governs the more frequent, in-memory body +/// patch (default 30 s, [`DEFAULT_APPLY_INTERVAL_MS`]). Decoupling the +/// two is the whole point: a created / renamed / deleted file must +/// become searchable quickly, but the compact-cache disk write should +/// stay rare. +/// +/// The trigger is purely time-gated with a "has churn" guard — it +/// fires when at least one event has been recorded since the last +/// apply / save **and** [`Self::evaluate`]'s `apply_interval` has +/// elapsed. There is intentionally no event-count fast-path: a huge +/// burst is already caught by `SaveTrigger`'s 50k threshold (a save +/// applies too), so the apply tick only needs to bound *latency*, not +/// volume. +#[derive(Debug)] +pub(crate) struct ApplyTrigger { + /// Wall-clock time of the last apply (or save, which also applies) + /// — or the loop spawn time before any fire. Compared against + /// `Instant::now()` to rate-limit applies to one per + /// `apply_interval`. + last_apply_at: Instant, + /// Events accumulated since the last apply / save. Non-zero is + /// the "there is something to apply" guard; the exact count does + /// not matter (no volume threshold here). + events_since_apply: u64, +} + +impl ApplyTrigger { + /// Construct a fresh trigger with `last_apply_at` set to + /// `Instant::now()` so the first apply can't fire until + /// `apply_interval` has elapsed since loop spawn. + pub(super) fn new() -> Self { + Self { + last_apply_at: Instant::now(), + events_since_apply: 0, + } + } + + /// Record `change_count` events toward the "has churn" guard. + /// Saturating so a runaway drive can't wrap the counter back to + /// zero and suppress an apply. + pub(super) const fn record(&mut self, change_count: u64) { + self.events_since_apply = self.events_since_apply.saturating_add(change_count); + } + + /// Evaluate the apply cadence. + /// + /// **Returns** `true` (and resets the counters) when there is + /// buffered churn and at least `apply_interval` has elapsed since + /// the last apply / save. Returns `false` — without resetting — + /// otherwise, so a not-yet-due tick keeps accumulating. + pub(super) fn evaluate(&mut self, apply_interval: Duration) -> bool { + if self.events_since_apply == 0 { + return false; + } + let now = Instant::now(); + if now.saturating_duration_since(self.last_apply_at) < apply_interval { + return false; + } + self.last_apply_at = now; + self.events_since_apply = 0; + true + } + + /// Reset the trigger because a **save** tick just drained + applied + /// the buffer (a save subsumes an apply). Clears the churn guard + /// and restarts the interval clock so the loop doesn't fire a + /// redundant apply on the same buffer right after a save. + pub(super) fn reset_after_save(&mut self) { + self.last_apply_at = Instant::now(); + self.events_since_apply = 0; + } +} diff --git a/crates/uffs-daemon/src/cache/journal_sink.rs b/crates/uffs-daemon/src/cache/journal_sink.rs index 4f2337d79..86a078781 100644 --- a/crates/uffs-daemon/src/cache/journal_sink.rs +++ b/crates/uffs-daemon/src/cache/journal_sink.rs @@ -32,6 +32,21 @@ //! head reset, so any pending events are stale relative to the new //! cursor) and falls back to the Phase-7 full-reload path. //! +//! ## Apply / save cadence split (search-freshness) +//! +//! Draining the buffer only on the save tick (50k events / 5 min) left +//! freshly created / renamed / deleted files invisible to search for +//! up to 5 minutes. `trigger_apply` decouples the two cadences: the +//! loop fires it on the apply interval (default ~30 s) to drain the buffer +//! into [`ApplyMsg::Apply`], which patches + swaps the in-memory body +//! (search goes near-live) but **skips** the compact-cache disk write +//! and the cursor persist. `trigger_save` keeps doing the full +//! patch-plus-persist on its rare cadence, so a save tick subsumes an +//! apply. The loop never fires both on the same poll; whichever fires +//! drains the buffer. Because only a real body save advances the +//! on-disk cursor, a cold start re-replays the apply-only deltas from +//! the last saved cursor — idempotent against the freshly loaded body. +//! //! Properties of the buffered design: //! //! 1. Preserves FIFO ordering (per-letter and across letters). @@ -97,6 +112,23 @@ enum ApplyMsg { /// shard's save is a no-op, so its cursor must not advance). cursor: u64, }, + /// `trigger_apply` callback — the short apply-cadence sibling of + /// `Save`. The applier runs the same surgical + /// [`crate::cache::ShardEntry::apply_usn_patch_to_body`] + + /// `replace_warm_body` over the drained per-letter buffer so the + /// in-memory body (and therefore search) goes near-live, but + /// **skips** the compact-cache disk write and the cursor persist. + /// Disk persistence stays on the rarer `Save` tick; the cursor only + /// advances on a real body save, so a cold start re-replays the + /// in-between deltas idempotently. + Apply { + /// Drive letter to patch. + letter: uffs_mft::platform::DriveLetter, + /// Drained per-letter event buffer. Empty when no churn + /// accumulated since the last apply / save (the surgical-patch + /// path short-circuits to a no-op). + changes: Vec, + }, /// `journal_wrapped` callback — the journal head reset so any /// pending events are stale; the applier discards them in the /// sink and runs a full @@ -286,6 +318,30 @@ impl PatchSink for RegistryPatchSink { }); } + fn trigger_apply(&self, letter: uffs_mft::platform::DriveLetter) { + // Drain the per-letter buffer just like `trigger_save`, but + // route it to the apply-only path: the body is patched + + // swapped (search goes live) without the compact-cache disk + // write or the cursor persist. Whichever tick (apply or save) + // fires drains the buffer; a save tick subsumes the apply, so + // the loop never fires both on the same poll. + let drained = { + let mut guard = self.lock_pending(); + guard.remove(&letter).unwrap_or_default() + }; + if drained.is_empty() { + // Nothing accumulated since the last drain — no work. (The + // loop only calls this when its event-count says there + // *should* be churn, so an empty drain here just means a save + // tick beat us to it.) + return; + } + let _ignore = self.apply_tx.send(ApplyMsg::Apply { + letter, + changes: drained, + }); + } + fn journal_wrapped(&self, letter: uffs_mft::platform::DriveLetter) { // Wrap means the journal head reset; any buffered events // are stale relative to the new cursor. Discard the @@ -363,6 +419,17 @@ async fn dispatch_msg(idx: &Arc, cursor_store: &dyn CursorStore, m cursor_store.store(letter, cursor); } } + ApplyMsg::Apply { letter, changes } => { + // Apply tick: patch the body + swap it into the registry so + // search goes live, but do NOT persist the compact cache or + // advance the on-disk cursor. Disk persistence + cursor + // advance stay on the rarer `Save` tick; a cold start + // re-replays the in-between deltas idempotently from the + // last saved cursor. + let _applied = idx + .handle_journal_apply(letter, "apply-tick", changes) + .await; + } ApplyMsg::Wrap { letter } => { // Wrap stays on the Phase-7 full-reload path. The // patched-body snapshot is invalidated by the journal diff --git a/crates/uffs-daemon/src/cache/journal_sink/tests.rs b/crates/uffs-daemon/src/cache/journal_sink/tests.rs index e025861bc..cfdd3409e 100644 --- a/crates/uffs-daemon/src/cache/journal_sink/tests.rs +++ b/crates/uffs-daemon/src/cache/journal_sink/tests.rs @@ -223,6 +223,93 @@ async fn trigger_save_with_no_pending_sends_empty_changes() { ); } +/// Pin: `trigger_apply` drains the pending buffer for `letter` +/// and ships it inside `ApplyMsg::Apply { changes }` — the +/// near-live body-patch path, carrying NO cursor (the apply tick +/// never advances the on-disk cursor). The buffer for `letter` is +/// cleared after the drain. +#[tokio::test] +async fn trigger_apply_drains_pending_into_apply_message() { + let (sink, mut rx) = RegistryPatchSink::new_for_test(); + + sink.accept(uffs_mft::platform::DriveLetter::C, &[ + make_change(20), + make_change(21), + ]); + sink.trigger_apply(uffs_mft::platform::DriveLetter::C); + + let ApplyMsg::Apply { letter, changes } = + rx.try_recv().expect("trigger_apply must enqueue Apply") + else { + panic!("expected ApplyMsg::Apply"); + }; + assert_eq!(letter, uffs_mft::platform::DriveLetter::C); + assert_eq!( + changes + .iter() + .map(|change| change.frs.raw()) + .collect::>(), + [20, 21], + "Apply must carry the buffered changes in send order", + ); + + // Pending buffer for 'C' is gone after the drain. + assert!( + pending_frs_for_letter(&sink, uffs_mft::platform::DriveLetter::C).is_none(), + "trigger_apply must remove the per-letter pending entry", + ); +} + +/// Pin: `trigger_apply` on a letter with no buffered events is a +/// pure no-op — it enqueues NO message (unlike `trigger_save`, +/// which always ships a Save so the cursor can advance on the +/// rare save cadence). An empty apply has nothing to patch and +/// no cursor to move, so spending a channel slot + an applier +/// wake-up on it would be wasted work on every quiescent tick. +#[tokio::test] +async fn trigger_apply_with_no_pending_enqueues_nothing() { + let (sink, mut rx) = RegistryPatchSink::new_for_test(); + + sink.trigger_apply(uffs_mft::platform::DriveLetter::Z); + + assert!( + rx.try_recv().is_err(), + "an empty trigger_apply must not enqueue an ApplyMsg", + ); +} + +/// Lockstep safety pin: the apply tick must NEVER persist the +/// cursor — only a real on-disk body save advances it. Even on a +/// successful in-memory patch the on-disk cursor stays pinned to +/// the last save, so a cold start re-replays the apply-only deltas +/// idempotently. Here the letter is unregistered (the patch +/// no-ops), but the contract holds on the success path too: the +/// apply dispatch never calls `cursor_store.store`. +#[tokio::test] +async fn apply_tick_never_persists_cursor() { + let idx = fresh_index_manager(); // no drives registered + let cursor_store = RecordingCursorStore::new(); + let (sink, applier) = RegistryPatchSink::spawn_with_applier( + &idx, + Arc::clone(&cursor_store) as Arc, + ); + + sink.accept(uffs_mft::platform::DriveLetter::C, &[make_change(1)]); + sink.trigger_apply(uffs_mft::platform::DriveLetter::C); + + drop(sink); + let join_result = tokio::time::timeout(core::time::Duration::from_secs(5), applier).await; + join_result + .expect("applier must exit within 5 s") + .expect("applier must not panic"); + + assert!( + cursor_store.store_log().is_empty(), + "the apply tick must never persist the cursor; got {:?}", + cursor_store.store_log(), + ); +} + /// Pin: `journal_wrapped` clears the pending buffer for the /// letter and emits `ApplyMsg::Wrap`. A subsequent /// `trigger_save` then sees an empty buffer (no replay of the diff --git a/crates/uffs-daemon/src/index/journal.rs b/crates/uffs-daemon/src/index/journal.rs index 828ccfadf..d536b0ca9 100644 --- a/crates/uffs-daemon/src/index/journal.rs +++ b/crates/uffs-daemon/src/index/journal.rs @@ -161,15 +161,70 @@ impl IndexManager { reason: &str, changes: Vec, ) -> bool { + match self.apply_to_body(letter, reason, changes).await { + BodyApplyOutcome::Applied(new_body) => { + spawn_compact_cache_save_task(letter, new_body); + true + } + BodyApplyOutcome::NoOp => true, + BodyApplyOutcome::Failed => false, + } + } + + /// Apply-tick sibling of [`IndexManager::handle_journal_save`]: + /// patch the in-memory body and Arc-swap it into the registry, + /// **without** the compact-cache disk write. + /// + /// The per-shard journal loop fires this on the short apply cadence + /// (default ~2 s, [`crate::cache::journal_loop`]) so newly created / + /// renamed / deleted files become searchable within a couple of + /// seconds, while the heavy disk persistence stays on the rare + /// `handle_journal_save` cadence (50k events / 5 min). The apply + /// path deliberately does **not** persist the journal cursor: only a + /// real body save advances the on-disk cursor, so a cold start + /// re-replays from the last saved cursor and the idempotent patcher + /// re-applies the in-between deltas — identical to the pre-split + /// cold-boot window. + /// + /// Returns `true` when the body was patched + swapped (or the batch + /// was empty — a no-op), `false` only on a hard failure (shard not + /// registered / not warm, patch task aborted, or the swap lost a + /// demote race). See [`IndexManager::handle_journal_save`] for the + /// per-failure breakdown. + pub(crate) async fn handle_journal_apply( + &self, + letter: uffs_mft::platform::DriveLetter, + reason: &str, + changes: Vec, + ) -> bool { + !matches!( + self.apply_to_body(letter, reason, changes).await, + BodyApplyOutcome::Failed + ) + } + + /// Shared core of [`IndexManager::handle_journal_save`] and + /// [`IndexManager::handle_journal_apply`]: clone the warm body, + /// apply the buffered batch, and Arc-swap the result into the + /// registry. Returns the patched body on success so the save-tick + /// caller can hand it to the background disk-save task; the + /// apply-tick caller discards it. Stops short of any disk I/O — + /// disk persistence is the save tick's responsibility alone. + async fn apply_to_body( + &self, + letter: uffs_mft::platform::DriveLetter, + reason: &str, + changes: Vec, + ) -> BodyApplyOutcome { if changes.is_empty() { log_save_empty_batch(letter, reason); - return true; + return BodyApplyOutcome::NoOp; } let change_count = changes.len(); let Some(shard) = self.snapshot_shard_for_letter(letter).await else { log_save_no_shard(letter, reason, change_count); - return false; + return BodyApplyOutcome::Failed; }; let (new_body, stats) = match self @@ -179,9 +234,9 @@ impl IndexManager { PatchTaskOutcome::Applied(body, stats) => (body, stats), PatchTaskOutcome::ShardNotWarm => { log_save_shard_demoted(letter, reason, change_count); - return false; + return BodyApplyOutcome::Failed; } - PatchTaskOutcome::TaskAborted => return false, + PatchTaskOutcome::TaskAborted => return BodyApplyOutcome::Failed, }; log_save_patch_applied(letter, reason, change_count, &stats); @@ -190,11 +245,10 @@ impl IndexManager { .apply_journal_body(letter, reason, Arc::clone(&new_body)) .await { - return false; + return BodyApplyOutcome::Failed; } - spawn_compact_cache_save_task(letter, new_body); - true + BodyApplyOutcome::Applied(new_body) } /// Snapshot the shard for `letter` from the registry read-lock, @@ -413,8 +467,27 @@ fn spawn_compact_cache_save_task( }); } +/// Outcome of [`IndexManager::apply_to_body`] — the shared body-patch +/// core behind the save tick and the apply tick. Carries the patched +/// body on success so the save-tick caller can persist it while the +/// apply-tick caller drops it; keeps the empty-batch no-op distinct +/// from a hard failure so each caller maps it to the right `bool`. +enum BodyApplyOutcome { + /// Body cloned, patched, and Arc-swapped into the registry. The + /// save tick hands the inner Arc to the background disk-save task; + /// the apply tick discards it (in-memory swap is all it owes). + Applied(Arc), + /// The batch was empty — nothing to apply. Both ticks treat this + /// as success (a save with no churn is a no-op, not a failure). + NoOp, + /// A hard failure: shard not registered / not warm, the patch task + /// aborted, or the swap lost a demote race. Already logged at the + /// failure site; both ticks surface it as `false`. + Failed, +} + /// Three-way classification of the surgical-patch blocking task's -/// `JoinHandle` result — lets [`IndexManager::handle_journal_save`] +/// `JoinHandle` result — lets [`IndexManager::apply_to_body`] /// match each outcome with its own diagnostic + control-flow /// without resorting to the `Option>` shape `clippy` /// (rightly) flags as a smell. diff --git a/crates/uffs-daemon/src/index/loading.rs b/crates/uffs-daemon/src/index/loading.rs index ac8a78da8..9eda05512 100644 --- a/crates/uffs-daemon/src/index/loading.rs +++ b/crates/uffs-daemon/src/index/loading.rs @@ -215,14 +215,6 @@ impl IndexManager { /// is skipped and an error is logged. This prevents a single stuck /// volume from making the daemon unkillable. #[cfg(windows)] - #[expect( - clippy::print_stderr, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" - )] - #[expect( - clippy::use_debug, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" - )] pub(crate) async fn load_live_drives( &self, drives: &[uffs_mft::platform::DriveLetter], @@ -230,7 +222,7 @@ impl IndexManager { lifecycle: &crate::lifecycle::LifecycleHandle, ) { let total = drives.len(); - eprintln!("[diag] load_live_drives: starting — drives={drives:?} no_cache={no_cache}"); + tracing::debug!(?drives, no_cache, "load_live_drives: starting"); self.set_loading_progress(0, total).await; let join_set = Self::spawn_drive_loaders(drives, no_cache); @@ -324,10 +316,6 @@ impl IndexManager { /// task overruns [`Self::DRIVE_LOAD_TIMEOUT`]. Each completion /// updates the daemon status so clients see incremental progress. #[cfg(windows)] - #[expect( - clippy::print_stderr, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" - )] async fn collect_drive_load_results( &self, mut join_set: tokio::task::JoinSet<( @@ -362,10 +350,6 @@ impl IndexManager { Ok(None) => break, Err(_elapsed) => { let remaining = total.saturating_sub(loaded); - eprintln!( - "[diag] load_live_drives: TIMEOUT — {remaining} drive(s) stuck after {}s", - Self::DRIVE_LOAD_TIMEOUT.as_secs() - ); tracing::error!( remaining, timeout_secs = Self::DRIVE_LOAD_TIMEOUT.as_secs(), @@ -392,10 +376,6 @@ impl IndexManager { /// successful drive into the index, log a partial failure, or log a /// task panic — and bump `loaded` exactly once per outcome. #[cfg(windows)] - #[expect( - clippy::print_stderr, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" - )] async fn handle_drive_load_result( &self, join_result: Result< @@ -419,17 +399,14 @@ impl IndexManager { } Ok((letter, Err(err))) => { *loaded += 1; - eprintln!("[diag] load_live_drives: FAILED drive={letter} error={err:#}"); - // Log the FULL anyhow cause chain to the file sink (the - // detached daemon's stderr above is discarded). `%err` - // alone shows only the outer context, hiding the real - // failure deep in the broker-handle read path. + // Log the FULL anyhow cause chain — `%err` alone shows only + // the outer context, hiding the real failure deep in the + // broker-handle read path. let err_chain = format!("{err:#}"); tracing::error!(drive = %letter, error = %err_chain, "Failed to load live drive"); } Err(err) => { *loaded += 1; - eprintln!("[diag] load_live_drives: PANIC in task error={err}"); tracing::error!(error = %err, "Task panicked loading drive"); } } diff --git a/crates/uffs-daemon/src/lib.rs b/crates/uffs-daemon/src/lib.rs index 81ed35155..8f3646f38 100644 --- a/crates/uffs-daemon/src/lib.rs +++ b/crates/uffs-daemon/src/lib.rs @@ -574,13 +574,14 @@ async fn spawn_journal_loops_for_warm_shards( RegistryPatchSink::spawn_with_applier(idx, Arc::clone(&cursor_store)); let sink_dyn: Arc = sink; - let config = JournalLoopConfig::default(); + let config = JournalLoopConfig::from_env(); let letters = idx.loaded_drive_letters().await; tracing::info!( target: "shard.journal", count = letters.len(), letters = ?letters, poll_interval_ms = config.poll_interval.as_millis(), + apply_interval_ms = config.apply_interval.as_millis(), save_threshold_events = config.save_threshold_events, save_threshold_age_secs = config.save_threshold_age.as_secs(), "Spawning per-shard journal loops", diff --git a/crates/uffs-daemon/src/main.rs b/crates/uffs-daemon/src/main.rs index 02594d5b3..898fa7255 100644 --- a/crates/uffs-daemon/src/main.rs +++ b/crates/uffs-daemon/src/main.rs @@ -117,14 +117,6 @@ struct Cli { } #[tokio::main] -#[expect( - clippy::print_stderr, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" -)] -#[expect( - clippy::use_debug, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" -)] async fn main() -> anyhow::Result<()> { let cli = Cli::parse(); @@ -135,23 +127,22 @@ async fn main() -> anyhow::Result<()> { .or_else(|_| std::env::var("RUST_LOG")) .unwrap_or_else(|_| cli.log_level.clone()); - // [diag] Print to stderr immediately — before the tracing subscriber is - // up — so this always appears in the terminal when uffsd is run directly. - eprintln!("[diag] uffsd main: drives={:?}", cli.drives); - eprintln!("[diag] uffsd main: mft_files={:?}", cli.mft_files); - eprintln!( - "[diag] uffsd main: log_spec={log_spec:?} (cli.log_level={:?})", - cli.log_level - ); - eprintln!("[diag] uffsd main: log_file={:?}", cli.log_file); - eprintln!( - "[diag] uffsd main: env UFFS_LOG={:?} RUST_LOG={:?}", - std::env::var("UFFS_LOG").ok(), - std::env::var("RUST_LOG").ok() - ); - let _guard = uffs_daemon::init_tracing(&log_spec, cli.log_file.as_deref()); + // Startup parameter dump — emitted at DEBUG so it is available for + // diagnostics (`--log-level debug`) but never pollutes the default + // `info` console of a detached/foreground daemon. + tracing::debug!( + drives = ?cli.drives, + mft_files = ?cli.mft_files, + log_spec = %log_spec, + cli_log_level = %cli.log_level, + log_file = ?cli.log_file, + env_uffs_log = ?std::env::var("UFFS_LOG").ok(), + env_rust_log = ?std::env::var("RUST_LOG").ok(), + "uffsd startup parameters" + ); + // Keep copies for potential IPC forwarding (moved into config below). let fwd_drives = cli.drives.clone(); let fwd_mft_files: Vec = cli @@ -201,51 +192,32 @@ fn log_load_response(resp: &LoadDriveResponse) { } /// Forward `--drive` / `--mft-file` to the running daemon via IPC. -#[expect( - clippy::print_stderr, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" -)] -#[expect( - clippy::use_debug, - reason = "[diag] diagnostic tracing — remove after D: drive issue is resolved" -)] fn forward_to_running_daemon( drives: &[uffs_mft::platform::DriveLetter], mft_files: &[String], no_cache: bool, ) -> anyhow::Result<()> { - eprintln!( - "[diag] forward_to_running_daemon: drives={drives:?} mft_files={mft_files:?} no_cache={no_cache}" + tracing::debug!( + ?drives, + ?mft_files, + no_cache, + "forward_to_running_daemon: begin" ); if drives.is_empty() && mft_files.is_empty() { tracing::info!("Daemon is already running. Nothing to load."); - eprintln!("[diag] forward_to_running_daemon: nothing to load — returning"); return Ok(()); } tracing::info!("Daemon is already running — forwarding load request via IPC..."); - eprintln!("[diag] forward_to_running_daemon: connecting to running daemon via IPC..."); let mut client = UffsClientSync::connect()?; - eprintln!("[diag] forward_to_running_daemon: IPC connected OK"); if !drives.is_empty() { - eprintln!("[diag] forward_to_running_daemon: calling load_drive_letters({drives:?})"); - let resp = client.load_drive_letters(drives, no_cache)?; - eprintln!( - "[diag] forward_to_running_daemon: response — loaded={:?} already={:?} errors={:?}", - resp.loaded, resp.already_loaded, resp.errors - ); - log_load_response(&resp); + // Per-item outcomes are logged by `log_load_response`. + log_load_response(&client.load_drive_letters(drives, no_cache)?); } if !mft_files.is_empty() { - eprintln!("[diag] forward_to_running_daemon: calling load_drive({mft_files:?})"); - let resp = client.load_drive(mft_files, no_cache)?; - eprintln!( - "[diag] forward_to_running_daemon: response — loaded={:?} already={:?} errors={:?}", - resp.loaded, resp.already_loaded, resp.errors - ); - log_load_response(&resp); + log_load_response(&client.load_drive(mft_files, no_cache)?); } Ok(()) diff --git a/crates/uffs-mft/src/usn/mod.rs b/crates/uffs-mft/src/usn/mod.rs index 54d791a54..b6b1b1c34 100644 --- a/crates/uffs-mft/src/usn/mod.rs +++ b/crates/uffs-mft/src/usn/mod.rs @@ -235,10 +235,15 @@ impl UsnRecord { /// Categorizes this USN record into a `ChangeType`. #[must_use] pub const fn change_type(&self) -> ChangeType { - if self.reason & reason::FILE_CREATE != 0 { - ChangeType::Created - } else if self.reason & reason::FILE_DELETE != 0 { + // DELETE before CREATE: a single close record can carry both bits + // when a file is created and removed within one open→close cycle + // (e.g. a transient temp file). The net is "gone", so it must + // classify as Deleted. Distinct create/delete events (the FRS-reuse + // case) arrive as separate records and are unaffected by this order. + if self.reason & reason::FILE_DELETE != 0 { ChangeType::Deleted + } else if self.reason & reason::FILE_CREATE != 0 { + ChangeType::Created } else if self.reason & reason::RENAME_NEW_NAME != 0 { ChangeType::Renamed } else if self.reason & (reason::DATA_EXTEND | reason::DATA_TRUNCATION) != 0 { @@ -257,6 +262,33 @@ impl UsnRecord { } } +/// Real per-file metadata that a USN record does NOT carry (size, +/// allocation, timestamps, attribute flags). +/// +/// A `UsnRecord` only conveys the FRS, parent FRS, name, and reason flags, +/// so a file created/renamed via the live journal lands in the index with +/// size 0 and zero timestamps. When the journal source issues a targeted +/// MFT read (`read_targeted_frs_records`) to recover the real values, it +/// attaches them here so `apply_usn_patch` can populate the record fully — +/// matching what a cold rebuild would store. Field representation mirrors +/// `CompactRecord` exactly (i64 µs timestamps, raw NTFS attribute flags), +/// so application is a straight copy with no conversion. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct RecordMeta { + /// Logical file size in bytes. + pub size: u64, + /// Allocated size on disk in bytes. + pub allocated: u64, + /// Creation time (Unix microseconds). + pub created: i64, + /// Last write time (Unix microseconds). + pub modified: i64, + /// Last access time (Unix microseconds). + pub accessed: i64, + /// Raw NTFS `FILE_ATTRIBUTE_*` flags. + pub flags: u32, +} + /// Aggregated changes for a single file (consolidates multiple USN records). // These bools represent independent change flags from USN journal records. // Using a bitflags pattern would add complexity without benefit for this DTO. @@ -283,6 +315,10 @@ pub struct FileChange { pub size_changed: bool, /// Did metadata change? pub metadata_changed: bool, + /// Real size/timestamp/flags metadata, when a targeted MFT read + /// backfilled it (USN records carry none). `None` → the applier leaves + /// the record's metrics zeroed for a later re-warm to fill. + pub meta: Option, } /// Aggregates multiple USN records into per-file changes. @@ -302,10 +338,28 @@ pub fn aggregate_changes(records: &[UsnRecord]) -> HashMap { if !record.filename.is_empty() { entry.filename.clone_from(&record.filename); } + // Records arrive in USN (time) order. The create/delete/rename flags + // are mutually exclusive *net states* for the slot: the LAST such + // event wins, so reusing an MFT record number (delete old → create + // new, same masked FRS) nets to a create, and a transient temp file + // (create → delete) nets to a delete. Size/metadata are independent + // and accumulate. Resolving order here keeps `apply_usn_patch`'s + // simple deleted/created/renamed branch dispatch correct. match record.change_type() { - ChangeType::Created => entry.created = true, - ChangeType::Deleted => entry.deleted = true, - ChangeType::Renamed => entry.renamed = true, + ChangeType::Created => { + entry.created = true; + entry.deleted = false; + entry.renamed = false; + } + ChangeType::Deleted => { + entry.deleted = true; + entry.created = false; + entry.renamed = false; + } + ChangeType::Renamed => { + entry.renamed = true; + entry.deleted = false; + } ChangeType::SizeChanged => entry.size_changed = true, ChangeType::MetadataChanged => entry.metadata_changed = true, ChangeType::Other => {} @@ -432,4 +486,90 @@ mod tests { assert_eq!(seen.get(&Usn::new(2)), Some(&"second")); assert_eq!(seen.get(&Usn::new(3)), None); } + + // ── aggregate_changes net-state resolution (was untested) ─────────── + // + // The original aggregator OR-ed independent created/deleted/renamed + // bools, losing the time order of the USN stream. NTFS reuses an MFT + // record number after a delete, so a "delete old, create new" pair + // lands on the same masked FRS in one poll window — and the net result + // must reflect the LAST event, not both. These pin that. + + use super::{ChangeType, UsnRecord, aggregate_changes, reason}; + use crate::frs::{Frs, ParentFrs}; + + /// Build a minimal `UsnRecord` for a given FRS, reason mask, and name. + fn rec(frs: u64, reason_mask: u32, name: &str) -> UsnRecord { + UsnRecord { + frs: Frs::new(frs), + parent_frs: ParentFrs::new(5), + usn: Usn::new(0), + reason: reason_mask, + file_attributes: 0, + filename: name.to_owned(), + } + } + + #[test] + fn aggregate_delete_then_create_reuse_nets_to_created() { + // FRS 42 deleted (old.txt), then the record number is reused by a + // newly-created new.pdf — the journal emits both for masked FRS 42. + // Net: the slot now holds new.pdf. Must NOT report a delete (that + // would tombstone the brand-new file in apply_usn_patch). + let records = vec![ + rec(42, reason::FILE_DELETE | reason::CLOSE, "old.txt"), + rec(42, reason::FILE_CREATE | reason::CLOSE, "new.pdf"), + ]; + let changes = aggregate_changes(&records); + let change = changes.get(&Frs::new(42)).expect("FRS 42 present"); + assert!(change.created, "net of delete→create reuse is a create"); + assert!(!change.deleted, "must not also report a delete"); + assert_eq!(change.filename, "new.pdf", "carries the new name"); + } + + #[test] + fn aggregate_create_then_delete_nets_to_deleted() { + // A transient file: created then deleted in one window. Net: gone. + let records = vec![ + rec(7, reason::FILE_CREATE, "temp.tmp"), + rec(7, reason::FILE_DELETE | reason::CLOSE, "temp.tmp"), + ]; + let changes = aggregate_changes(&records); + let change = changes.get(&Frs::new(7)).expect("FRS 7 present"); + assert!(change.deleted, "net of create→delete is a delete"); + assert!(!change.created, "must not also report a create"); + } + + #[test] + fn change_type_prefers_delete_when_create_and_delete_coincide() { + // A single close record can carry create+delete in one reason mask + // (file created and removed within one open→close cycle). The net + // is "gone", so it must classify as Deleted, not Created. + let record = rec( + 1, + reason::FILE_CREATE | reason::FILE_DELETE | reason::CLOSE, + "x", + ); + assert!(matches!(record.change_type(), ChangeType::Deleted)); + } + + #[test] + fn aggregate_keeps_unrelated_frs_separate() { + // Sanity: two distinct FRS values never cross-contaminate. + let records = vec![ + rec(10, reason::FILE_CREATE | reason::CLOSE, "a.pdf"), + rec(20, reason::FILE_DELETE | reason::CLOSE, "b.dll"), + ]; + let changes = aggregate_changes(&records); + assert!( + changes + .get(&Frs::new(10)) + .is_some_and(|chg| chg.created && !chg.deleted) + ); + assert!( + changes + .get(&Frs::new(20)) + .is_some_and(|chg| chg.deleted && !chg.created) + ); + } } diff --git a/crates/uffs-update/src/restore.rs b/crates/uffs-update/src/restore.rs index d92162e8e..9c5c13add 100644 --- a/crates/uffs-update/src/restore.rs +++ b/crates/uffs-update/src/restore.rs @@ -91,10 +91,39 @@ fn start_from_command_line(running: &SnapRunning, ready: impl Fn() -> bool) -> b return false; }; // Spawn detached: don't wait, so the relaunched service outlives us. - let spawned = Command::new(&program).args(&args).spawn().is_ok(); + let mut command = Command::new(&program); + command.args(&args); + detach_stdio(&mut command); + let spawned = command.spawn().is_ok(); spawned && wait_until(START_TIMEOUT, &ready) } +/// Sever the relaunched service from this process's console. +/// +/// The captured command line replays the daemon's argv but NOT the stdio +/// redirection the normal `--daemon start` path applies via the client's +/// detached spawn. Without this, a service relaunched here inherits the +/// updater's console: the daemon's `info` tracing (and any startup prints) +/// pour into the user's terminal long after `uffs --update` returns. Null +/// all three handles, and on Windows also set `DETACHED_PROCESS | +/// CREATE_NO_WINDOW` so no console is attached at all. +fn detach_stdio(command: &mut Command) { + command + .stdin(std::process::Stdio::null()) + .stdout(std::process::Stdio::null()) + .stderr(std::process::Stdio::null()); + + #[cfg(windows)] + { + use std::os::windows::process::CommandExt as _; + /// `DETACHED_PROCESS` — the child gets no console. + const DETACHED_PROCESS: u32 = 0x0000_0008; + /// `CREATE_NO_WINDOW` — belt-and-suspenders against a console window. + const CREATE_NO_WINDOW: u32 = 0x0800_0000; + command.creation_flags(DETACHED_PROCESS | CREATE_NO_WINDOW); + } +} + /// Split a captured command line into `(program, args)`. /// /// Naive whitespace split — sufficient for UFFS's switch-style argv diff --git a/scripts/windows/usn-verify.rs b/scripts/windows/usn-verify.rs new file mode 100644 index 000000000..c9c2ee746 --- /dev/null +++ b/scripts/windows/usn-verify.rs @@ -0,0 +1,244 @@ +#!/usr/bin/env rust-script +//! ```cargo +//! [dependencies] +//! anyhow = "1" +//! ``` +//! +//! usn-verify.rs — controlled, repeatable live USN-journal verification. +//! +//! Reproduces the manual "create / search / rename / delete" sequence that +//! PowerShell's multi-line paste kept mangling, but driven from Rust so each +//! step runs in order, every search's output is captured to a file, and the +//! daemon's debug/trace log is collected — all in ONE place so the run is +//! trivial to share. +//! +//! What it exercises (the v0.6.13 USN-delta fixes): +//! * create → file findable by name AND by `--ext`, with REAL size/time +//! (the metadata backfill), not size 0 / 1601-epoch. +//! * rename → `charlie.log` → `charlie.pdf` moves into `--ext pdf`, out of +//! `--ext log`. +//! * delete → `bravo.dll` drops out of `--ext dll`. +//! * FRS reuse → recreating into a just-deleted dir doesn't drop files. +//! +//! ## Binary +//! +//! Uses `~/bin/uffs.exe` (the canonical install path). **Copy your freshly +//! built `target\release\{uffs,uffsd,uffs-broker}.exe` into `~/bin` first** — +//! the daemon that gets spawned is the `uffsd.exe` sitting next to the +//! `uffs.exe` this script invokes, so the install dir is what's under test. +//! +//! ## Usage +//! +//! rust-script scripts\windows\usn-verify.rs +//! +//! ## Output +//! +//! Everything lands in `~/usntest`: +//! * `usntest_*.{pdf,dll,log}` — the files under test +//! * `_run/NN-*.csv` — each search's exact stdout +//! * `_run/uffsd.log` — the daemon's debug+trace log for the run +//! * `_run/usn-apply.log` — just the `usn apply:` / `usn backfill:` +//! lines, extracted for quick sharing +//! +//! Share `_run/` and we can see, per 500 ms poll, exactly what the journal +//! loop did (`created=N deleted=N renamed=N skipped=N`) and whether the +//! targeted metadata read fired. + +use std::fs; +use std::path::{Path, PathBuf}; +use std::process::Command; +use std::thread::sleep; +use std::time::Duration; + +use anyhow::{Context, Result, bail}; + +/// Time to let the per-shard USN loop ingest a batch and patch the live +/// body. With [`APPLY_INTERVAL_MS`] pinned to 500 ms below, the apply +/// tick fires on essentially the first poll that sees the new events, so +/// 3 s is a comfortable margin (the body is searchable well under 1 s +/// after the file op in practice). No 5-minute disk-save wait is needed +/// — the apply tick is decoupled from the rare compact-cache save. +const POLL_SETTLE: Duration = Duration::from_secs(3); +/// Apply-cadence override (ms) for the test daemon — pins +/// `UFFS_USN_APPLY_INTERVAL_MS` low so the near-live body patch fires +/// promptly and deterministically within [`POLL_SETTLE`]. The +/// production default is 30 s (tuned so constant FS churn stays +/// background noise); the harness pins it to 500 ms so the short +/// create / rename / delete rounds don't have to wait that out. +const APPLY_INTERVAL_MS: &str = "500"; +/// Settle time after `--daemon stop` so the socket / PID file clear. +const KILL_SETTLE: Duration = Duration::from_secs(2); +/// Tracing directive: per-change USN trace + daemon-side debug (backfill, +/// journal loop) on top of an `info` baseline. `init_tracing` feeds this +/// straight into `EnvFilter::try_new`, so the full directive form works. +const LOG_SPEC: &str = "info,uffs_core::compact_loader=trace,uffs_daemon=debug"; +/// Bytes written into the headline `.pdf` so the size-backfill assertion is +/// visible (`Size` column should read this, not 0). +const ALPHA_BYTES: usize = 5000; + +/// `~/bin/uffs.exe` — the canonical user-installed **Rust** binary. +/// +/// Pinned to the explicit `uffs.exe` filename on purpose: a bare `uffs` +/// on Windows resolves through `PATHEXT`, where `.com` precedes `.exe`, +/// so if the C++ build (`uffs.com`) is also on `PATH` it would shadow +/// the Rust `uffs.exe` we are trying to exercise. Returning the full +/// path and handing it to `Command::new` bypasses `PATHEXT` resolution +/// entirely, so this script always runs the Rust build under test. +/// Never "simplify" this to a bare `uffs`. +fn uffs_bin() -> PathBuf { + let home = std::env::var_os("USERPROFILE") + .or_else(|| std::env::var_os("HOME")) + .map(PathBuf::from) + .expect("USERPROFILE or HOME must be set"); + let name = if cfg!(windows) { "uffs.exe" } else { "uffs" }; + home.join("bin").join(name) +} + +/// Display name for the cosmetic `$ ...` echo lines. Uses the same +/// `uffs.exe` the script actually spawns so a shared transcript is +/// copy-paste-safe: pasting `uffs ` into a shell could hit the +/// C++ `uffs.com` (see [`uffs_bin`]), but `uffs.exe ` cannot. +fn uffs_display() -> &'static str { + if cfg!(windows) { "uffs.exe" } else { "uffs" } +} + +/// Home directory (`~`) — the scratch tree lives at `~/usntest`. +fn home_dir() -> PathBuf { + std::env::var_os("USERPROFILE") + .or_else(|| std::env::var_os("HOME")) + .map(PathBuf::from) + .expect("USERPROFILE or HOME must be set") +} + +/// Run a `uffs` subcommand inheriting stdout/stderr (for `--status`, daemon +/// control) — the user sees exactly what they would running it by hand. +fn run(uffs: &Path, args: &[&str]) -> Result<()> { + println!("\n$ {} {}", uffs_display(), args.join(" ")); + Command::new(uffs) + .args(args) + .status() + .with_context(|| format!("failed to spawn uffs {}", args.join(" ")))?; + Ok(()) +} + +/// Run a `uffs` search, capture its stdout to `out`, and print a one-line +/// summary (row count + the names found) next to the expectation. +fn capture(uffs: &Path, args: &[&str], out: &Path, expect: &str) -> Result<()> { + let output = Command::new(uffs) + .args(args) + .output() + .with_context(|| format!("failed to spawn uffs {}", args.join(" ")))?; + fs::write(out, &output.stdout).with_context(|| format!("write {}", out.display()))?; + let text = String::from_utf8_lossy(&output.stdout); + + // CSV: one header line + one blank, then data rows; all quoted lines. + let quoted = text.lines().filter(|l| l.starts_with('"')).count(); + let rows = quoted.saturating_sub(1); // minus the header + let names: Vec<&str> = text + .lines() + .filter(|l| l.starts_with("\"C:") || l.starts_with("\"\\\\")) + .filter_map(|l| l.split('"').nth(3)) // 2nd CSV field = Name + .take(8) + .collect(); + + println!("\n$ {} {}", uffs_display(), args.join(" ")); + println!(" expect: {expect}"); + println!(" got: {rows} row(s) {names:?}"); + println!(" saved: {}", out.display()); + Ok(()) +} + +fn main() -> Result<()> { + let uffs = uffs_bin(); + if !uffs.exists() { + bail!( + "uffs binary not found at {}\n\ + Copy your freshly built target\\release\\{{uffs,uffsd,uffs-broker}}.exe \ + into ~/bin first, then re-run.", + uffs.display() + ); + } + + let base = home_dir().join("usntest"); + let run_dir = base.join("_run"); + println!("== UFFS USN verification =="); + println!("binary: {}", uffs.display()); + println!("scratch: {}", base.display()); + println!("artifacts: {}", run_dir.display()); + + // Fresh tree. + let _ = fs::remove_dir_all(&base); + fs::create_dir_all(&run_dir).with_context(|| format!("create {}", run_dir.display()))?; + + run(&uffs, &["--version"])?; + + // ── Restart the daemon with debug+trace logging into the artifacts dir ── + let _ = Command::new(&uffs).args(["--daemon", "stop"]).status(); + sleep(KILL_SETTLE); + println!( + "\n$ {} --daemon start (UFFS_LOG={LOG_SPEC}, UFFS_USN_APPLY_INTERVAL_MS={APPLY_INTERVAL_MS})", + uffs_display() + ); + let status = Command::new(&uffs) + .args(["--daemon", "start"]) + .env("UFFS_LOG", LOG_SPEC) + .env("UFFS_LOG_DIR", &run_dir) + .env("UFFS_USN_APPLY_INTERVAL_MS", APPLY_INTERVAL_MS) + .status() + .context("failed to spawn `uffs --daemon start`")?; + if !status.success() { + bail!("`uffs --daemon start` exited with {status}"); + } + run(&uffs, &["--status"])?; + + // ── Round 1: create four files with distinct extensions ───────────────── + println!("\n== Round 1: create =="); + fs::write(base.join("usntest_alpha.pdf"), "a".repeat(ALPHA_BYTES))?; + fs::write(base.join("usntest_delta.pdf"), b"x")?; + fs::write(base.join("usntest_bravo.dll"), b"x")?; + fs::write(base.join("usntest_charlie.log"), b"x")?; + sleep(POLL_SETTLE); + + capture(&uffs, &["usntest", "--format", "csv"], &run_dir.join("01-name.csv"), "4 files (alpha/delta/bravo/charlie)")?; + capture(&uffs, &["usntest", "--ext", "pdf", "--format", "csv"], &run_dir.join("02-ext-pdf.csv"), "alpha.pdf + delta.pdf")?; + capture(&uffs, &["usntest", "--ext", "dll", "--format", "csv"], &run_dir.join("03-ext-dll.csv"), "bravo.dll")?; + capture(&uffs, &["usntest", "--ext", "log", "--format", "csv"], &run_dir.join("04-ext-log.csv"), "charlie.log")?; + // Metadata backfill: alpha.pdf should show ~ALPHA_BYTES + real timestamps. + capture(&uffs, &["usntest_alpha.pdf", "--format", "csv"], &run_dir.join("05-alpha-meta.csv"), &format!("size ≈ {ALPHA_BYTES}, real (non-1601) timestamps"))?; + + // ── Round 2: rename + delete ──────────────────────────────────────────── + println!("\n== Round 2: rename charlie.log -> charlie.pdf, delete bravo.dll =="); + fs::rename(base.join("usntest_charlie.log"), base.join("usntest_charlie.pdf"))?; + fs::remove_file(base.join("usntest_bravo.dll"))?; + sleep(POLL_SETTLE); + + capture(&uffs, &["usntest", "--ext", "pdf", "--format", "csv"], &run_dir.join("06-ext-pdf-after.csv"), "alpha + delta + charlie (3 pdfs)")?; + capture(&uffs, &["usntest", "--ext", "dll", "--format", "csv"], &run_dir.join("07-ext-dll-after.csv"), "EMPTY (bravo deleted)")?; + capture(&uffs, &["usntest", "--ext", "log", "--format", "csv"], &run_dir.join("08-ext-log-after.csv"), "EMPTY (charlie renamed)")?; + + // ── Stop the daemon to flush the log, then extract the USN lines ──────── + println!("\n== Stopping daemon to flush the log =="); + let _ = Command::new(&uffs).args(["--daemon", "stop"]).status(); + sleep(KILL_SETTLE); + + let log_path = run_dir.join("uffsd.log"); + let apply_path = run_dir.join("usn-apply.log"); + match fs::read_to_string(&log_path) { + Ok(log) => { + let lines: Vec<&str> = log + .lines() + .filter(|l| l.contains("usn apply:") || l.contains("usn backfill:")) + .collect(); + fs::write(&apply_path, lines.join("\n"))?; + println!("extracted {} usn-apply/backfill line(s) -> {}", lines.len(), apply_path.display()); + } + Err(err) => { + println!("(could not read {} — {err}; the daemon log may be elsewhere if UFFS_LOG_DIR was overridden)", log_path.display()); + } + } + + println!("\n== Done =="); + println!("Share the artifacts dir: {}", run_dir.display()); + println!("Key files: 01-name.csv (creates), 06/07/08 (rename+delete), 05-alpha-meta.csv (backfill), usn-apply.log (per-poll dispositions)."); + Ok(()) +} diff --git a/supply-chain/audits.toml b/supply-chain/audits.toml index 9a7289374..c0ba907e2 100644 --- a/supply-chain/audits.toml +++ b/supply-chain/audits.toml @@ -91,6 +91,12 @@ criteria = "safe-to-deploy" delta = "0.1.16 -> 0.1.17" notes = """Delta audit (cargo vet diff 0.1.16 -> 0.1.17). Cargo.toml(.orig): version bump + redox_syscall dep 0.7 -> 0.8. src/lib.rs: (a) cosmetic reorder of a libc MSG_* re-export list (no semantic change); (b) adds one Redox kernel FFI binding redox_relpathat_v0(dirfd, fd, dst_base, dst_len) -> RawResult plus the safe wrappers Fd::relpathat / call::relpathat, mirroring the existing redox_fpath_v1 binding exactly (Error::demux over an unsafe extern call using the caller-provided &mut [u8] buffer's ptr+len). The single new unsafe block is a like-for-like copy of the adjacent fpath binding. libredox is the Redox stable-ABI shim, reachable only via redox_users on target_os="redox"; the extern symbols are never linked on UFFS shipping targets (Windows/macOS). No network/FS-path/process/env additions. Publisher 4lDO2 (Redox maintainer).""" +[[audits.memmap2]] +who = "Robert M1 <50460704+githubrobbi@users.noreply.github.com>" +criteria = "safe-to-deploy" +delta = "0.9.10 -> 0.9.11" +notes = "Delta audit (cargo vet diff 0.9.10 -> 0.9.11), full diff reviewed across all 7 files. This is the RUSTSEC-2026-0186 remediation. Core fix in src/unix.rs + src/windows.rs MmapInner: the range methods (flush_range / flush_async_range / advise / advise_range) now bounds-check `offset > self.len || len > self.len - offset` and return InvalidInput BEFORE the pointer arithmetic — closing the unchecked-pointer-offset unsoundness. Pointer math switched from `self.ptr.offset(-(x as isize))` to `self.ptr.sub(x)` / `.add(x)` (equivalent, with SAFETY comments), the mremap success path swaps via std::mem::replace + mem::forget instead of raw ptr::write, and every FFI/pointer op is wrapped in an explicit `unsafe {}` under a new crate-level #![deny(unsafe_op_in_unsafe_fn)]; `advise`/`unchecked_advise` gain explicit `# Safety` docs. src/lib.rs only forwards those unsafe contracts; Cargo.toml/.orig bump MSRV 1.63->1.65; CHANGELOG + one doc typo. No new dependencies, no new capability (extern/network/FS) surface; the only behavioral change is stricter input validation, which can reject previously-UB calls but never newly accepts. Publisher de-vri-es (memmap2 co-maintainer)." + [[audits.mimalloc]] who = "Robert Nio " criteria = "safe-to-deploy" diff --git a/supply-chain/config.toml b/supply-chain/config.toml index fa79dcd77..b35c986cd 100644 --- a/supply-chain/config.toml +++ b/supply-chain/config.toml @@ -434,10 +434,6 @@ criteria = "safe-to-deploy" version = "0.5.4" criteria = "safe-to-deploy" -[[exemptions.fallible-streaming-iterator]] -version = "0.1.9" -criteria = "safe-to-deploy" - [[exemptions.fast-float2]] version = "0.2.3" criteria = "safe-to-deploy" @@ -978,14 +974,6 @@ criteria = "safe-to-deploy" version = "1.6.0" criteria = "safe-to-deploy" -[[exemptions.rmp]] -version = "0.8.15" -criteria = "safe-to-deploy" - -[[exemptions.rmp-serde]] -version = "1.3.1" -criteria = "safe-to-deploy" - [[exemptions.rustc-hash]] version = "2.1.2" criteria = "safe-to-deploy" diff --git a/supply-chain/imports.lock b/supply-chain/imports.lock index 18cd48112..b6c288831 100644 --- a/supply-chain/imports.lock +++ b/supply-chain/imports.lock @@ -1958,6 +1958,12 @@ criteria = "safe-to-deploy" delta = "0.3.1 -> 0.3.3" aggregated-from = "https://hg.mozilla.org/mozilla-central/raw-file/tip/supply-chain/audits.toml" +[[audits.mozilla.audits.fallible-streaming-iterator]] +who = "Jan-Erik Rediger " +criteria = "safe-to-deploy" +version = "0.1.9" +aggregated-from = "https://raw.githubusercontent.com/mozilla/glean/main/supply-chain/audits.toml" + [[audits.mozilla.audits.fnv]] who = "Bobby Holley " criteria = "safe-to-deploy" @@ -2189,6 +2195,35 @@ As far as I can tell, it does not have any file IO or network access. """ aggregated-from = "https://hg.mozilla.org/mozilla-central/raw-file/tip/supply-chain/audits.toml" +[[audits.mozilla.audits.rmp]] +who = "Ben Dean-Kawamura " +criteria = "safe-to-deploy" +version = "0.8.14" +notes = """ +Very popular crate. 1 instance of unsafe code, which is used to adjust a slice to work around +lifetime issues. No network or file access. +""" +aggregated-from = "https://hg.mozilla.org/mozilla-central/raw-file/tip/supply-chain/audits.toml" + +[[audits.mozilla.audits.rmp]] +who = "Jan-Erik Rediger " +criteria = "safe-to-deploy" +delta = "0.8.14 -> 0.8.15" +aggregated-from = "https://raw.githubusercontent.com/mozilla/glean/main/supply-chain/audits.toml" + +[[audits.mozilla.audits.rmp-serde]] +who = "Ben Dean-Kawamura " +criteria = "safe-to-deploy" +version = "1.3.0" +notes = "Very popular crate. No unsafe code, network or file access." +aggregated-from = "https://hg.mozilla.org/mozilla-central/raw-file/tip/supply-chain/audits.toml" + +[[audits.mozilla.audits.rmp-serde]] +who = "Jan-Erik Rediger " +criteria = "safe-to-deploy" +delta = "1.3.0 -> 1.3.1" +aggregated-from = "https://raw.githubusercontent.com/mozilla/glean/main/supply-chain/audits.toml" + [[audits.mozilla.audits.serde]] who = "Erich Gubler " criteria = "safe-to-deploy"