diff --git a/CLAUDE.md b/CLAUDE.md index 9e43749c8..25602bbd5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -9,6 +9,7 @@ A monorepo task runner (like Nx/Turbo) with intelligent caching and dependency r - `crates/vt_graph` — Task dependency graph construction and config loading - `crates/vt_plan` — Execution planning (resolves env vars, working dirs, commands) - `crates/vt_workspace` — Workspace detection and package dependency graph +- `crates/vt_fs_fingerprint` — Filesystem fingerprinting for task caching (pre-run snapshot, traced-access judgment, cache-entry validation) - `crates/fspy*` — File system access tracing (9 crates: supervisor, preload libs, platform backends) - `crates/pty_terminal*` — Cross-platform headless terminal emulator (3 crates) - `crates/vt_path` — Type-safe absolute/relative path system diff --git a/Cargo.lock b/Cargo.lock index afd51f863..238e7bf62 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4143,6 +4143,7 @@ dependencies = [ "ctrlc", "derive_more", "fspy", + "fspy_shared", "futures-util", "materialized_artifact", "materialized_artifact_build", @@ -4151,7 +4152,6 @@ dependencies = [ "owo-colors", "petgraph", "pty_terminal_test_client", - "rayon", "rusqlite", "rustc-hash", "serde", @@ -4163,9 +4163,9 @@ dependencies = [ "tokio", "tokio-util", "tracing", - "twox-hash", "uuid", "vt_client_napi", + "vt_fs_fingerprint", "vt_glob", "vt_graph", "vt_ipc_shared", @@ -4175,7 +4175,6 @@ dependencies = [ "vt_server", "vt_str", "vt_workspace", - "wax", "winapi", "wincode", "zstd", @@ -4250,6 +4249,28 @@ dependencies = [ "vt_str", ] +[[package]] +name = "vt_fs_fingerprint" +version = "0.0.0" +dependencies = [ + "anyhow", + "fspy_shared", + "nix 0.31.2", + "rayon", + "rustc-hash", + "serde", + "tempfile", + "thiserror 2.0.18", + "tracing", + "twox-hash", + "vt_glob", + "vt_graph", + "vt_path", + "vt_str", + "wax", + "wincode", +] + [[package]] name = "vt_glob" version = "0.0.0" diff --git a/Cargo.toml b/Cargo.toml index 1bfc8a785..030f6232f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -148,6 +148,7 @@ tui-term = "0.3.1" twox-hash = "2.1.1" uuid = "1.18.1" vec1 = "1.12.1" +vt_fs_fingerprint = { path = "crates/vt_fs_fingerprint" } vt_glob = { path = "crates/vt_glob" } vt_graph_ser = { path = "crates/vt_graph_ser" } vt_path = { path = "crates/vt_path" } diff --git a/crates/vt/Cargo.toml b/crates/vt/Cargo.toml index 1a59fc291..5040c8ac2 100644 --- a/crates/vt/Cargo.toml +++ b/crates/vt/Cargo.toml @@ -19,12 +19,12 @@ wincode = { workspace = true, features = ["derive"] } clap = { workspace = true, features = ["derive"] } ctrlc = { workspace = true } derive_more = { workspace = true, features = ["debug", "from"] } +fspy_shared = { workspace = true } futures-util = { workspace = true } once_cell = { workspace = true } owo-colors = { workspace = true } petgraph = { workspace = true } pty_terminal_test_client = { workspace = true } -rayon = { workspace = true } rusqlite = { workspace = true, features = ["bundled"] } rustc-hash = { workspace = true } serde = { workspace = true, features = ["derive", "rc"] } @@ -42,9 +42,9 @@ tokio = { workspace = true, features = [ ] } tokio-util = { workspace = true } tracing = { workspace = true } -twox-hash = { workspace = true } materialized_artifact = { workspace = true } uuid = { workspace = true, features = ["v4"] } +vt_fs_fingerprint = { workspace = true } vt_glob = { workspace = true } vt_path = { workspace = true } vt_select = { workspace = true } @@ -54,7 +54,6 @@ vt_ipc_shared = { workspace = true } vt_plan = { workspace = true } vt_server = { workspace = true } vt_workspace = { workspace = true } -wax = { workspace = true } zstd = { workspace = true } # Artifact build-deps must be unconditional: cargo's resolver panics when diff --git a/crates/vt/docs/task-cache.md b/crates/vt/docs/task-cache.md index c39fa45fd..ed863df9b 100644 --- a/crates/vt/docs/task-cache.md +++ b/crates/vt/docs/task-cache.md @@ -167,10 +167,11 @@ The cached execution result: ```rust pub struct CacheEntryValue { - pub post_run_fingerprint: PostRunFingerprint, + pub input_fingerprints: InputFingerprints, // vt_fs_fingerprint (opaque) + pub tracked_env_fingerprints: TrackedEnvFingerprints, pub std_outputs: Arc<[StdOutput]>, pub duration: Duration, - pub globbed_inputs: BTreeMap, + pub output_archive: Option, } ``` @@ -348,19 +349,20 @@ Cache entries are serialized using `bincode` for efficient storage. │ • Stop monitoring │ │ │ │ │ ▼ │ -│ 2. Generate Post-Run Fingerprint │ +│ 2. Conclude the run (vt_fs_fingerprint) │ │ ───────────────────────────────── │ -│ • Hash all accessed files │ -│ • Record file system access patterns │ +│ • Judge traced accesses (read-write overlap check) │ +│ • Fingerprint discovered inputs, collect outputs │ │ │ │ │ ▼ │ │ 3. Create CacheEntryValue │ │ ──────────────────────────── │ │ CacheEntryValue { │ -│ post_run_fingerprint, │ +│ input_fingerprints, │ +│ tracked_env_fingerprints, │ │ std_outputs, │ │ duration, │ -│ globbed_inputs, │ +│ output_archive, │ │ } │ │ │ │ │ ▼ │ @@ -565,13 +567,21 @@ Each `&&` separated command is cached independently. If only terser config chang ### Core Cache Components ``` +crates/vt_fs_fingerprint/src/ # Filesystem fingerprinting (own crate) +├── task_run.rs # TaskFs (pre_run/post_run), Conclusion +├── fingerprint.rs # InputFingerprints, PathFingerprint, InputChange +├── tracked_accesses.rs # fspy access normalization +├── glob.rs # Glob walking + input hashing +└── hash.rs # Content hashing + crates/vt/src/session/ ├── cache/ │ ├── mod.rs # ExecutionCache, CacheEntryKey/Value, FingerprintMismatch │ └── display.rs # Cache status display formatting ├── execute/ -│ ├── mod.rs # execute_spawn, SpawnOutcome -│ ├── fingerprint.rs # PostRunFingerprint, PathFingerprint, DirEntryKind +│ ├── mod.rs # execute_spawn, SpawnOutcome, cache lookup +│ ├── cache_update.rs # Post-run cache update decision +│ ├── post_run.rs # TrackedEnvFingerprints (tracked env validation) │ └── spawn.rs # spawn_with_tracking, fspy integration └── reporter/ └── mod.rs # Reporter traits for cache hit/miss display diff --git a/crates/vt/src/lib.rs b/crates/vt/src/lib.rs index 7a59dfd3b..b5a9e8258 100644 --- a/crates/vt/src/lib.rs +++ b/crates/vt/src/lib.rs @@ -1,5 +1,4 @@ mod cli; -mod collections; mod napi_client; pub mod session; diff --git a/crates/vt/src/session/cache/display.rs b/crates/vt/src/session/cache/display.rs index 87d86958a..5ae8e4bf9 100644 --- a/crates/vt/src/session/cache/display.rs +++ b/crates/vt/src/session/cache/display.rs @@ -250,7 +250,7 @@ mod tests { fn inline_tracked_env_mismatch_preserves_kind() { let added = CacheStatus::Miss(CacheMiss::FingerprintMismatch( FingerprintMismatch::TrackedEnvQueryChanged { - query: crate::session::execute::fingerprint::TrackedEnvQuery::Glob(Str::from( + query: crate::session::execute::post_run::TrackedEnvQuery::Glob(Str::from( "PROBE_*", )), mismatch: EnvMismatch::Added { name: Str::from("PROBE_C") }, diff --git a/crates/vt/src/session/cache/mod.rs b/crates/vt/src/session/cache/mod.rs index cc1972bf2..8db5d9a14 100644 --- a/crates/vt/src/session/cache/mod.rs +++ b/crates/vt/src/session/cache/mod.rs @@ -3,7 +3,7 @@ pub mod archive; pub mod display; -use std::{collections::BTreeMap, fmt::Display, fs::File, io::Write, sync::Arc, time::Duration}; +use std::{fmt::Display, fs::File, io::Write, sync::Arc, time::Duration}; // Re-export display functions for convenience pub use display::format_cache_status_inline; @@ -14,6 +14,8 @@ pub use display::{ use rusqlite::{Connection, OptionalExtension as _}; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; +pub use vt_fs_fingerprint::InputChangeKind; +use vt_fs_fingerprint::{InputChange, InputFingerprints}; use vt_graph::config::ResolvedGlobConfig; use vt_path::{AbsolutePath, RelativePathBuf}; use vt_plan::cache_metadata::{CacheMetadata, ExecutionCacheKey, SpawnFingerprint}; @@ -26,8 +28,8 @@ use wincode::{ }; use super::execute::{ - fingerprint::{PostRunFingerprint, TrackedEnvQuery}, pipe::StdOutput, + post_run::{PostRunMismatch, TrackedEnvFingerprints, TrackedEnvQuery}, }; const TASK_CACHE_PREALLOCATION_SIZE_LIMIT: usize = 256 * 1024 * 1024; @@ -118,19 +120,19 @@ unsafe impl<'de, C: ConfigCore> SchemaRead<'de, C> for DurationSchema { /// Cached execution result for a task. /// -/// Contains the post-run fingerprint (from fspy), captured outputs, -/// execution duration, and explicit input file hashes. +/// Contains the run's input fingerprints and tracked env state, captured +/// outputs, and execution duration. #[derive(Debug, SchemaWrite, SchemaRead, Serialize)] pub struct CacheEntryValue { - pub post_run_fingerprint: PostRunFingerprint, + /// Fingerprints of everything the cached run read. Checked against the + /// filesystem at lookup to decide whether the entry is still valid. + pub input_fingerprints: InputFingerprints, + /// Env vars and bulk env queries observed by runner-aware tools during + /// the run. Checked against the current env context at lookup. + pub tracked_env_fingerprints: TrackedEnvFingerprints, pub std_outputs: Arc<[StdOutput]>, #[wincode(with = "DurationSchema")] pub duration: Duration, - /// Hashes of explicit input files computed from positive globs. - /// Files matching negative globs are already filtered out. - /// Path is relative to workspace root, value is `xxHash3_64` of file content. - /// Stored in the value (not the key) so changes can be detected and reported. - pub globbed_inputs: BTreeMap, /// Filename of the output archive (e.g. `{uuid}.tar.zst`) stored alongside /// `cache.db` in the cache directory. `None` if no output files were produced. pub output_archive: Option, @@ -151,16 +153,6 @@ pub enum CacheMiss { FingerprintMismatch(FingerprintMismatch), } -#[derive(Debug, Clone, Copy, Serialize, Deserialize)] -pub enum InputChangeKind { - /// File content changed but path is the same - ContentModified, - /// New file or folder added - Added, - /// Existing file or folder removed - Removed, -} - /// A single env var difference between a stored fingerprint and the current /// environment. /// @@ -240,11 +232,15 @@ pub enum FingerprintMismatch { }, } -impl From for FingerprintMismatch { - fn from(mismatch: crate::session::execute::fingerprint::PostRunMismatch) -> Self { - use crate::session::execute::fingerprint::PostRunMismatch; +impl From for FingerprintMismatch { + fn from(change: InputChange) -> Self { + Self::InputChanged { kind: change.kind, path: change.path } + } +} + +impl From for FingerprintMismatch { + fn from(mismatch: PostRunMismatch) -> Self { match mismatch { - PostRunMismatch::Input { kind, path } => Self::InputChanged { kind, path }, PostRunMismatch::TrackedEnv(mismatch) => Self::TrackedEnvChanged(mismatch), PostRunMismatch::TrackedEnvQuery { query, mismatch } => { Self::TrackedEnvQueryChanged { query, mismatch } @@ -274,7 +270,7 @@ pub fn split_path(path: &str) -> (Option<&str>, &str) { /// its own cache warm across branch switches, and a cache from a different /// version is simply ignored (it lives in a directory this build never looks /// at) rather than aborting the run. Bumping the version starts a fresh cache. -const CACHE_SCHEMA_VERSION: u32 = 18; +const CACHE_SCHEMA_VERSION: u32 = 19; /// Name of the per-version subdirectory (e.g. `v14`) under the task-cache /// directory that holds the database and output archives for the current @@ -318,71 +314,81 @@ impl ExecutionCache { Ok(()) } - /// Try to hit cache by looking up the cache entry key and validating inputs. - /// Returns `Ok(Ok(cache_value))` on cache hit, `Ok(Err(cache_miss))` on miss. + /// Fetch the stored entry for this task's exact cache key, or the reason + /// the key missed: the task ran before under a different key (command/env, + /// input config, or output config changed — checked in that priority + /// order), or it never ran at all. The old entry's value is never reused, + /// only its key is compared. + /// + /// Whether a fetched entry is still valid is the caller's question. + #[expect( + clippy::significant_drop_tightening, + reason = "lock guard cannot be dropped earlier because the transaction borrows the connection" + )] #[tracing::instrument(level = "debug", skip_all)] - pub async fn try_hit( + pub(crate) async fn fetch_entry( &self, cache_metadata: &CacheMetadata, - globbed_inputs: &BTreeMap, - workspace_root: &AbsolutePath, ) -> anyhow::Result> { - let spawn_fingerprint = &cache_metadata.spawn_fingerprint; - let execution_cache_key = &cache_metadata.execution_cache_key; - let cache_key = CacheEntryKey::from_metadata(cache_metadata); - // Try to find the cache entry by key (spawn fingerprint + input config) - if let Some(cache_value) = self.get_by_cache_key(&cache_key).await? { - // Validate explicit globbed inputs against the stored values - if let Some(mismatch) = - detect_globbed_input_change(&cache_value.globbed_inputs, globbed_inputs) - { - return Ok(Err(CacheMiss::FingerprintMismatch(mismatch))); - } - - // Validate post-run fingerprint (inferred inputs + tracked envs) - if let Some(mismatch) = cache_value - .post_run_fingerprint - .validate(workspace_root, &cache_metadata.unfiltered_envs)? - { - return Ok(Err(CacheMiss::FingerprintMismatch(mismatch.into()))); - } - // Associate the execution key to the cache entry key if not already, - // so that next time we can find it and report what changed - self.upsert_task_fingerprint(execution_cache_key, &cache_key).await?; - return Ok(Ok(cache_value)); - } - - // No cache found with the current cache entry key, - // check if execution key maps to a different cache entry key - if let Some(old_cache_key) = - self.get_cache_key_by_execution_key(execution_cache_key).await? - { - // Destructure to ensure we handle all fields when new ones are added. - // `get_by_cache_key` above returned None for the *current* cache key, - // so at least one field on `old_cache_key` must differ from the - // current metadata — checked in priority order (spawn → input → output). - let CacheEntryKey { - spawn_fingerprint: old_spawn_fingerprint, - input_config: old_input_config, - output_config: old_output_config, - } = old_cache_key; - let mismatch = if old_spawn_fingerprint != *spawn_fingerprint { - FingerprintMismatch::SpawnFingerprint { - old: old_spawn_fingerprint, - new: spawn_fingerprint.clone(), - } - } else if old_input_config != cache_metadata.input_config { - FingerprintMismatch::InputConfig + let (entry, old_cache_key) = { + let mut conn = self.conn.lock().await; + // Both reads run in one deferred read transaction, so they see a + // single database snapshot: the miss classification below may rely + // on the entry fetch having missed even while a concurrent run of + // the same task writes the cache. + let tx = conn.transaction()?; + let entry: Option = get_value(&tx, "cache_entries", &cache_key)?; + let old_cache_key: Option = if entry.is_some() { + None } else { - debug_assert_ne!(old_output_config, cache_metadata.output_config); - FingerprintMismatch::OutputConfig + get_value(&tx, "task_fingerprints", &cache_metadata.execution_cache_key)? }; - return Ok(Err(CacheMiss::FingerprintMismatch(mismatch))); + // Read-only: dropping `tx` (a rollback) is equivalent to a commit. + (entry, old_cache_key) + }; + + if let Some(entry) = entry { + return Ok(Ok(entry)); } - Ok(Err(CacheMiss::NotFound)) + let Some(old_cache_key) = old_cache_key else { + return Ok(Err(CacheMiss::NotFound)); + }; + + // Destructure to ensure we handle all fields when new ones are added. + // The current cache key found no entry in the same snapshot, so at + // least one field on `old_cache_key` must differ from the current + // metadata. + let CacheEntryKey { + spawn_fingerprint: old_spawn_fingerprint, + input_config: old_input_config, + output_config: old_output_config, + } = old_cache_key; + let spawn_fingerprint = &cache_metadata.spawn_fingerprint; + let mismatch = if old_spawn_fingerprint != *spawn_fingerprint { + FingerprintMismatch::SpawnFingerprint { + old: old_spawn_fingerprint, + new: spawn_fingerprint.clone(), + } + } else if old_input_config != cache_metadata.input_config { + FingerprintMismatch::InputConfig + } else { + debug_assert_ne!(old_output_config, cache_metadata.output_config); + FingerprintMismatch::OutputConfig + }; + Ok(Err(CacheMiss::FingerprintMismatch(mismatch))) + } + + /// Associate the task's execution key with the entry key that served a + /// hit, so a later key-level miss can report what changed. + pub(crate) async fn record_hit(&self, cache_metadata: &CacheMetadata) -> anyhow::Result<()> { + self.upsert_task_fingerprint( + &cache_metadata.execution_cache_key, + &CacheEntryKey::from_metadata(cache_metadata), + ) + .await } /// Update cache after successful execution. @@ -419,106 +425,34 @@ impl ExecutionCache { } } -/// Compare stored and current globbed inputs, returning the first changed path. -/// Both maps are `BTreeMap` so we iterate them in sorted lockstep. -fn detect_globbed_input_change( - stored: &BTreeMap, - current: &BTreeMap, -) -> Option { - let mut stored_iter = stored.iter(); - let mut current_iter = current.iter(); - let mut s = stored_iter.next(); - let mut c = current_iter.next(); - - loop { - match (s, c) { - (None, None) => return None, - (Some((sp, _)), None) => { - return Some(FingerprintMismatch::InputChanged { - kind: InputChangeKind::Removed, - path: sp.clone(), - }); - } - (None, Some((cp, _))) => { - return Some(FingerprintMismatch::InputChanged { - kind: InputChangeKind::Added, - path: cp.clone(), - }); - } - (Some((sp, sh)), Some((cp, ch))) => match sp.cmp(cp) { - std::cmp::Ordering::Equal => { - if sh != ch { - return Some(FingerprintMismatch::InputChanged { - kind: InputChangeKind::ContentModified, - path: sp.clone(), - }); - } - s = stored_iter.next(); - c = current_iter.next(); - } - std::cmp::Ordering::Less => { - return Some(FingerprintMismatch::InputChanged { - kind: InputChangeKind::Removed, - path: sp.clone(), - }); - } - std::cmp::Ordering::Greater => { - return Some(FingerprintMismatch::InputChanged { - kind: InputChangeKind::Added, - path: cp.clone(), - }); - } - }, - } - } +/// Fetch and deserialize one value by key from `table` on an already-held +/// connection (or transaction), so callers control how many reads share a +/// snapshot. +fn get_value(conn: &Connection, table: &str, key: &K) -> anyhow::Result> +where + K: SchemaWrite, + V: SchemaReadOwned, +{ + let key_blob = serialize_cache(key)?; + #[expect(clippy::disallowed_macros, reason = "SQL query string for rusqlite requires String")] + let mut select_stmt = conn.prepare_cached(&format!("SELECT value FROM {table} WHERE key=?"))?; + let value_blob: Option> = + select_stmt.query_row::, _, _>([key_blob], |row| row.get(0)).optional()?; + let Some(value_blob) = value_blob else { + return Ok(None); + }; + let value: V = deserialize_cache(&value_blob)?; + Ok(Some(value)) } // Basic database operations impl ExecutionCache { - #[expect( - clippy::significant_drop_tightening, - reason = "lock guard cannot be dropped earlier because prepared statement borrows connection" - )] - async fn get_key_by_value< - K: SchemaWrite, - V: SchemaReadOwned, - >( - &self, - table: &str, - key: &K, - ) -> anyhow::Result> { - let key_blob = serialize_cache(key)?; - let value_blob = { - let conn = self.conn.lock().await; - #[expect( - clippy::disallowed_macros, - reason = "SQL query string for rusqlite requires String" - )] - let mut select_stmt = - conn.prepare_cached(&format!("SELECT value FROM {table} WHERE key=?"))?; - let value_blob: Option> = - select_stmt.query_row::, _, _>([key_blob], |row| row.get(0)).optional()?; - value_blob - }; - let Some(value_blob) = value_blob else { - return Ok(None); - }; - let value: V = deserialize_cache(&value_blob)?; - Ok(Some(value)) - } - async fn get_by_cache_key( &self, cache_key: &CacheEntryKey, ) -> anyhow::Result> { - self.get_key_by_value("cache_entries", cache_key).await - } - - async fn get_cache_key_by_execution_key( - &self, - execution_cache_key: &ExecutionCacheKey, - ) -> anyhow::Result> { - self.get_key_by_value("task_fingerprints", execution_cache_key).await + let conn = self.conn.lock().await; + get_value(&conn, "cache_entries", cache_key) } #[expect( diff --git a/crates/vt/src/session/execute/cache_update.rs b/crates/vt/src/session/execute/cache_update.rs index f13d3f1df..8fc8983dd 100644 --- a/crates/vt/src/session/execute/cache_update.rs +++ b/crates/vt/src/session/execute/cache_update.rs @@ -1,9 +1,10 @@ //! Post-run cache update: decide whether a finished spawn may be cached and, -//! if so, store its fingerprint, captured output, and output archive. +//! if so, store its fingerprints, captured output, and output archive. use std::{collections::BTreeMap, sync::Arc, time::Duration}; use rustc_hash::FxHashSet; +use vt_fs_fingerprint::{Conclusion, PostRunError}; use vt_path::{AbsolutePath, RelativePathBuf}; use vt_plan::cache_metadata::{CacheMetadata, EnvValueHash}; use vt_server::Reports; @@ -11,31 +12,14 @@ use vt_str::Str; use super::{ CacheState, - fingerprint::{PathRead, PostRunFingerprint, TrackedEnvQuery}, - glob, + post_run::{TrackedEnvFingerprints, TrackedEnvQuery}, spawn::ChildOutcome, }; -use crate::{ - collections::HashMap, - session::{ - cache::{CacheEntryValue, ExecutionCache, archive}, - event::{CacheErrorKind, CacheNotUpdatedReason, CacheUpdateStatus, ExecutionError}, - }, +use crate::session::{ + cache::{CacheEntryValue, ExecutionCache, archive}, + event::{CacheErrorKind, CacheNotUpdatedReason, CacheUpdateStatus, ExecutionError}, }; -/// Post-execution summary of what fspy observed for a single task. Fields are -/// cfg-agnostic so the decision logic below doesn't need `cfg(fspy)` — the -/// value is only ever `Some` when tracking happened (see [`observe_fspy`]). -struct TrackingOutcome { - path_reads: HashMap, - /// Auto-output writes after output exclusions are applied. Empty when - /// `output_config.includes_auto` is false. - path_writes: FxHashSet, - /// First path that was both read and written during execution, if any. - /// A non-empty value means caching this task is unsound. - read_write_overlap: Option, -} - type TrackedEnvValues = BTreeMap>; type TrackedEnvQueryValues = BTreeMap>; @@ -59,8 +43,7 @@ pub(super) async fn update_cache( duration: Duration, cancelled: bool, ) -> (CacheUpdateStatus, Option) { - let CacheState { metadata, globbed_inputs, std_outputs, tracking } = state; - let fspy = tracking.fspy.as_ref(); + let CacheState { metadata, task_fs, std_outputs, tracking } = state; if let Some(reports) = reports && reports.cache_disabled @@ -70,15 +53,6 @@ pub(super) async fn update_cache( return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::ToolRequested), None); } - // Tool-reported paths to exclude from auto input tracking. Absolute paths - // are normalized to workspace-relative; anything outside is dropped. - let ignored_input_rels: FxHashSet = reports - .map(|r| normalize_ignored_paths(&r.ignored_inputs, workspace_root)) - .unwrap_or_default(); - let ignored_output_rels: FxHashSet = reports - .map(|r| normalize_ignored_paths(&r.ignored_outputs, workspace_root)) - .unwrap_or_default(); - if cancelled { // Cancelled (Ctrl-C or sibling failure) — result is untrustworthy. return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::Cancelled), None); @@ -89,62 +63,57 @@ pub(super) async fn update_cache( return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::NonZeroExitStatus), None); } - let fspy_outcome = observe_fspy( - outcome, - metadata, - fspy, - &ignored_input_rels, - &ignored_output_rels, - workspace_root, - ); - - if let Some(TrackingOutcome { read_write_overlap: Some(path), .. }) = &fspy_outcome { - // fspy-inferred read-write overlap: the task wrote to a file it also - // read, so the prerun input hashes are stale and caching is unsound. - // (We only check fspy-inferred reads, not globbed_inputs. A task that - // writes to a glob-matched file without reading it produces perpetual - // cache misses but not a correctness bug.) - return ( - CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::InputModified { - path: path.clone(), - }), - None, - ); - } - - if fspy_outcome.is_none() && fspy.is_some() { - // Task requested fspy auto-inference but this binary was built without - // `cfg(fspy)`. Task ran, but we can't compute a valid cache entry - // without tracked path accesses. + // Task requested fspy auto-inference but no trace exists (this binary was + // built without `cfg(fspy)`). Task ran, but we can't compute a valid cache + // entry without tracked path accesses. + #[cfg(fspy)] + let has_trace = outcome.path_accesses.is_some(); + #[cfg(not(fspy))] + let has_trace = false; + if tracking.fspy && !has_trace { return (CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::FspyUnsupported), None); } - // Collect tool-reported tracked envs for the post-run fingerprint. Env - // names that the user already declared are skipped because their values - // are already part of the spawn fingerprint. - let (tracked_envs, tracked_env_queries) = match collect_tracked_reports(reports, metadata) { - Ok(tracked_reports) => tracked_reports, - Err(err) => { + let conclusion = { + #[cfg(fspy)] + let accesses = outcome.path_accesses.as_ref().map(fspy::PathAccessIterable::iter); + #[cfg(not(fspy))] + let accesses: Option>> = None; + + let empty_ignored = FxHashSet::default(); + let reported_ignored_inputs = reports.map_or(&empty_ignored, |r| &r.ignored_inputs); + let reported_ignored_outputs = reports.map_or(&empty_ignored, |r| &r.ignored_outputs); + task_fs.post_run(accesses, reported_ignored_inputs, reported_ignored_outputs) + }; + let (input_fingerprints, outputs) = match conclusion { + Ok(Conclusion::InputModified { path }) => { + // The task wrote a file it also read, so the pre-run input hashes + // are stale and caching is unsound. + return ( + CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::InputModified { path }), + None, + ); + } + Ok(Conclusion::Cacheable { input_fingerprints, outputs }) => (input_fingerprints, outputs), + Err(PostRunError::InputFingerprints(err)) => { return ( CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::CacheDisabled), Some(ExecutionError::PostRunFingerprint(err)), ); } + Err(PostRunError::Outputs(err)) => { + return ( + CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::CacheDisabled), + Some(ExecutionError::Cache { kind: CacheErrorKind::Update, source: err }), + ); + } }; - // Paths already in globbed_inputs are skipped: the overlap check above - // guarantees no input modification, so the prerun hash is the correct - // post-exec hash. - let empty_path_reads = HashMap::default(); - let path_reads = fspy_outcome.as_ref().map_or(&empty_path_reads, |o| &o.path_reads); - let post_run_fingerprint = match PostRunFingerprint::create( - path_reads, - workspace_root, - &globbed_inputs, - tracked_envs, - tracked_env_queries, - ) { - Ok(fingerprint) => fingerprint, + // Collect tool-reported tracked envs for the cache entry. Env names that + // the user already declared are skipped because their values are already + // part of the spawn fingerprint. + let (tracked_envs, tracked_env_queries) = match collect_tracked_reports(reports, metadata) { + Ok(tracked_reports) => tracked_reports, Err(err) => { return ( CacheUpdateStatus::NotUpdated(CacheNotUpdatedReason::CacheDisabled), @@ -153,12 +122,7 @@ pub(super) async fn update_cache( } }; - let output_archive = match collect_and_archive_outputs( - metadata, - fspy_outcome.as_ref(), - workspace_root, - cache_dir, - ) { + let output_archive = match archive_outputs(&outputs, workspace_root, cache_dir) { Ok(archive) => archive, Err(err) => { return ( @@ -169,10 +133,10 @@ pub(super) async fn update_cache( }; let new_cache_value = CacheEntryValue { - post_run_fingerprint, + input_fingerprints, + tracked_env_fingerprints: TrackedEnvFingerprints { tracked_envs, tracked_env_queries }, std_outputs: std_outputs.into(), duration, - globbed_inputs, output_archive, }; match cache.update(metadata, new_cache_value, cache_dir).await { @@ -184,81 +148,6 @@ pub(super) async fn update_cache( } } -/// Summarize the run's fspy observations. `Some` iff tracking was both -/// requested (`tracking.fspy.is_some()`) and compiled in (`cfg(fspy)`). On a -/// `cfg(not(fspy))` build this is always `None`, and [`update_cache`] -/// short-circuits to `FspyUnsupported` when tracking was needed. -/// -/// `path_reads` is gated on `input_config.includes_auto`, filtered by -/// user-configured input negatives, and by tool-reported `ignoreInput` paths. -/// `path_writes` is filtered by user-configured output negatives and -/// tool-reported `ignoreOutput` paths before read-write overlap detection. -fn observe_fspy( - outcome: &ChildOutcome, - metadata: &CacheMetadata, - fspy: Option<&super::FspyTracking<'_>>, - ignored_input_rels: &FxHashSet, - ignored_output_rels: &FxHashSet, - workspace_root: &AbsolutePath, -) -> Option { - #[cfg(fspy)] - { - use super::tracked_accesses::TrackedPathAccesses; - - outcome.path_accesses.as_ref().map(|raw| { - let tracked = TrackedPathAccesses::from_raw(raw, workspace_root); - let filtered_path_reads: HashMap = - // fspy can be attached for auto-output-only tasks. In that - // mode reads must not become inferred inputs. - if metadata.input_config.includes_auto - && let Some(fspy) = fspy - { - tracked - .path_reads - .iter() - .filter(|(path, _)| { - !fspy.input_negative_globs.is_match(path.as_str()) - && !is_ignored(path, ignored_input_rels) - }) - .map(|(path, read)| (path.clone(), *read)) - .collect() - } else { - HashMap::default() - }; - let filtered_path_writes: FxHashSet = - // fspy can also be attached for auto-input-only tasks. In that - // mode writes must not become auto outputs or overlap candidates. - if metadata.output_config.includes_auto - && let Some(fspy) = fspy - { - tracked - .path_writes - .iter() - .filter(|path| { - !fspy.output_negative_globs.is_match(path.as_str()) - && !is_ignored(path, ignored_output_rels) - }) - .cloned() - .collect() - } else { - FxHashSet::default() - }; - let read_write_overlap = - filtered_path_reads.keys().find(|p| filtered_path_writes.contains(*p)).cloned(); - TrackingOutcome { - path_reads: filtered_path_reads, - path_writes: filtered_path_writes, - read_write_overlap, - } - }) - } - #[cfg(not(fspy))] - { - let _ = (outcome, metadata, fspy, ignored_input_rels, ignored_output_rels, workspace_root); - None - } -} - fn collect_tracked_reports( reports: Option<&Reports>, metadata: &CacheMetadata, @@ -273,29 +162,7 @@ fn collect_tracked_reports( .map(Option::unwrap_or_default) } -/// Normalize tool-reported absolute paths to cleaned workspace-relative paths. -/// Paths outside the workspace are dropped — they can't contribute to inputs -/// or outputs. -fn normalize_ignored_paths( - paths: &FxHashSet>, - workspace_root: &AbsolutePath, -) -> FxHashSet { - paths - .iter() - .filter_map(|p| p.strip_prefix(workspace_root).ok().flatten()?.clean().ok()) - .collect() -} - -/// Whether `path` is covered by any `ignored` entry. An ignored entry matches -/// itself (exact file) and everything under it (directory subtree). -fn is_ignored(path: &RelativePathBuf, ignored: &FxHashSet) -> bool { - if ignored.is_empty() { - return false; - } - ignored.contains(path) || ignored.iter().any(|ig| path.strip_prefix(ig).is_some()) -} - -/// Select tool-reported env records to embed in the post-run fingerprint. +/// Select tool-reported env records to embed in the cache entry. /// Names that the user already declared as fingerprinted are skipped because /// their values are already in the spawn fingerprint. fn collect_tracked_envs( @@ -326,8 +193,8 @@ fn collect_tracked_envs( Ok(tracked_envs) } -/// Select tool-reported bulk env query records to embed in the post-run -/// fingerprint. The full match-set is stored as value hashes. +/// Select tool-reported bulk env query records to embed in the cache entry. +/// The full match-set is stored as value hashes. fn collect_tracked_env_queries(reports: &Reports) -> anyhow::Result { let mut tracked_env_queries = BTreeMap::new(); @@ -356,74 +223,23 @@ fn collect_tracked_env_queries(reports: &Reports) -> anyhow::Result, +/// Returns `Some(archive_filename)` if files were archived, `None` if the run +/// produced no output files. +fn archive_outputs( + outputs: &[RelativePathBuf], workspace_root: &AbsolutePath, cache_dir: &AbsolutePath, ) -> anyhow::Result> { - let output_config = &cache_metadata.output_config; - - let mut output_files: FxHashSet = FxHashSet::default(); - - if let Some(t) = tracking { - output_files.extend(t.path_writes.iter().cloned()); - } - - if !output_config.positive_globs.is_empty() { - let glob_paths = glob::collect_glob_paths( - workspace_root, - &output_config.positive_globs, - &output_config.negative_globs, - )?; - output_files.extend(glob_paths); - } - - if output_files.is_empty() { + if outputs.is_empty() { return Ok(None); } - let mut sorted_files: Vec = output_files.into_iter().collect(); - sorted_files.sort(); - let archive_name: Str = vt_str::format!("{}.tar.zst", uuid::Uuid::new_v4()); let archive_path = cache_dir.join(archive_name.as_str()); - archive::create_output_archive(workspace_root, &sorted_files, &archive_path)?; + archive::create_output_archive(workspace_root, outputs, &archive_path)?; Ok(Some(archive_name)) } - -#[cfg(test)] -mod tests { - use std::sync::Arc; - - use rustc_hash::FxHashSet; - use vt_path::{AbsolutePath, RelativePathBuf}; - - use super::normalize_ignored_paths; - - #[test] - fn normalize_ignored_paths_cleans_relative_components() { - let workspace_root = - AbsolutePath::new(if cfg!(windows) { r"C:\repo" } else { "/repo" }).unwrap(); - let ignored = - workspace_root.join(if cfg!(windows) { r"pkg\..\cache" } else { "pkg/../cache" }); - let mut ignored_paths = FxHashSet::default(); - ignored_paths.insert(Arc::::from(ignored)); - - let normalized = normalize_ignored_paths(&ignored_paths, workspace_root); - - let expected = RelativePathBuf::new("cache").unwrap(); - assert!(normalized.contains(&expected)); - } -} diff --git a/crates/vt/src/session/execute/fingerprint.rs b/crates/vt/src/session/execute/fingerprint.rs deleted file mode 100644 index 39320c2fc..000000000 --- a/crates/vt/src/session/execute/fingerprint.rs +++ /dev/null @@ -1,631 +0,0 @@ -//! Post-run fingerprinting for execution caching. -//! -//! This module provides types and functions for creating and validating -//! fingerprints of file system state after task execution. - -use std::{ - collections::BTreeMap, - ffi::OsStr, - fs::File, - io::{self, BufRead}, - sync::Arc, -}; - -use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; -use rustc_hash::FxHashMap; -use serde::{Deserialize, Serialize}; -use vt_path::{AbsolutePath, RelativePathBuf}; -use vt_plan::cache_metadata::EnvValueHash; -use vt_str::Str; -use wincode::{SchemaRead, SchemaWrite}; - -use crate::{ - collections::HashMap, - session::cache::{EnvMismatch, InputChangeKind}, -}; - -#[derive( - SchemaWrite, SchemaRead, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, -)] -pub enum TrackedEnvQuery { - Glob(Str), - Prefix(Str), -} - -/// Path read access info -#[derive(Debug, Clone, Copy)] -pub struct PathRead { - pub read_dir_entries: bool, -} - -/// Post-run fingerprint capturing file state after execution. -/// Used to validate whether cached outputs are still valid. -#[derive(SchemaWrite, SchemaRead, Debug, Default, Serialize)] -pub struct PostRunFingerprint { - /// Paths inferred from fspy during execution with their content fingerprints. - /// Only populated when `input_config.includes_auto` is true. - pub inferred_inputs: HashMap, - - /// Env vars observed via runner-aware IPC `getEnv` with `tracked: true`. - /// Key is the env name; value is the env value hash at execution time, or - /// `None` if unset. Validated at cache lookup against the same plan env - /// context that served the original request. - pub tracked_envs: BTreeMap>, - - /// Bulk env queries (`getEnvs`) made with `tracked: true`. - /// Outer key is the query, inner map is the match-set at execution time - /// (name -> value hash). Validated at cache lookup by re-matching against - /// the current env context and comparing the resulting set. - /// - /// Non-UTF-8 env names are never matched, saved, or treated as errors: - /// they are not returned to the client, so their existence cannot affect - /// task behavior. Values are stricter. A matched env must have a UTF-8 - /// value; the JS client errors when querying a matched non-UTF-8 value, - /// and cache-hit validation treats a currently matched non-UTF-8 value as - /// a changed mismatch so stale cached output is not replayed. - pub tracked_env_queries: BTreeMap>, -} - -/// A mismatch between the stored post-run fingerprint and the current state. -#[derive(Debug, Clone)] -pub enum PostRunMismatch { - /// An inferred input file or directory changed. - Input { kind: InputChangeKind, path: RelativePathBuf }, - /// A tool-tracked env var changed value, appeared, or disappeared. - TrackedEnv(EnvMismatch), - /// A tool-tracked bulk env query's match-set changed between runs. Carries - /// the first differing entry in env-name order. - TrackedEnvQuery { query: TrackedEnvQuery, mismatch: EnvMismatch }, -} - -/// Fingerprint for a single path (file or directory) -#[derive(SchemaWrite, SchemaRead, PartialEq, Eq, Debug, Clone, Serialize, Deserialize)] -pub enum PathFingerprint { - /// Path was not found when fingerprinting - NotFound, - /// File content hash using `xxHash3_64` - FileContentHash(u64), - /// Directory with optional entry listing. - /// `Folder(None)` means the directory was opened but entries were not read - /// (e.g., for `openat` calls). - /// `Folder(Some(_))` contains the directory entries sorted by name. - Folder(Option>), -} - -/// Kind of directory entry -#[derive(SchemaWrite, SchemaRead, PartialEq, Eq, Debug, Clone, Serialize, Deserialize)] -pub enum DirEntryKind { - File, - Dir, - Symlink, -} - -impl PostRunFingerprint { - /// Creates a new fingerprint from path accesses after task execution. - /// - /// Negative glob filtering is done upstream (see - /// [`super::tracked_accesses::TrackedPathAccesses::from_raw`]). - /// Paths already present in `globbed_inputs` are skipped — they are - /// already tracked by the prerun glob fingerprint, and the read-write - /// overlap check in `execute_spawn` guarantees the task did not modify - /// them, so the prerun hash is still correct. - /// - /// # Arguments - /// * `inferred_path_reads` - Map of paths that were read during execution (from fspy) - /// * `base_dir` - Workspace root for resolving relative paths - /// * `globbed_inputs` - Prerun glob fingerprint; paths here are skipped - /// * `tracked_envs` - Tool-requested env vars (name -> value hash), validated on lookup - /// * `tracked_env_queries` - Tool-requested bulk env queries (query -> match-set hashes) - #[tracing::instrument(level = "debug", skip_all, name = "create_post_run_fingerprint")] - pub fn create( - inferred_path_reads: &HashMap, - base_dir: &AbsolutePath, - globbed_inputs: &BTreeMap, - tracked_envs: BTreeMap>, - tracked_env_queries: BTreeMap>, - ) -> anyhow::Result { - let inferred_inputs = inferred_path_reads - .par_iter() - .filter(|(path, _)| !globbed_inputs.contains_key(*path)) - .map(|(relative_path, path_read)| { - let full_path = Arc::::from(base_dir.join(relative_path)); - let fingerprint = fingerprint_path(&full_path, *path_read)?; - Ok((relative_path.clone(), fingerprint)) - }) - .collect::>>()?; - - Ok(Self { inferred_inputs, tracked_envs, tracked_env_queries }) - } - - /// Validates the fingerprint against current filesystem state and the - /// unfiltered env context used by runner-aware IPC. `unfiltered_envs` must - /// be the same plan env context that served the original `getEnv` request, - /// not the filtered env passed to the spawned process. - /// - /// Returns `Some(mismatch)` if anything changed, `None` if all valid. - /// Returns an error if a tracked env is currently present but cannot be - /// represented as UTF-8; treating that value as unset would make cache - /// validation unsound. - #[tracing::instrument(level = "debug", skip_all, name = "validate_post_run_fingerprint")] - pub fn validate( - &self, - base_dir: &AbsolutePath, - unfiltered_envs: &FxHashMap, Arc>, - ) -> anyhow::Result> { - let input_mismatch = self.inferred_inputs.par_iter().find_map_any( - |(input_relative_path, path_fingerprint)| { - let input_full_path = Arc::::from(base_dir.join(input_relative_path)); - let path_read = PathRead { - read_dir_entries: matches!(path_fingerprint, PathFingerprint::Folder(Some(_))), - }; - let current_path_fingerprint = match fingerprint_path(&input_full_path, path_read) { - Ok(ok) => ok, - Err(err) => return Some(Err(err)), - }; - if path_fingerprint == ¤t_path_fingerprint { - None - } else { - let (kind, entry_name) = - determine_change_kind(path_fingerprint, ¤t_path_fingerprint); - let path = if let Some(name) = entry_name { - // For folder changes, build `dir/entry` path - let entry = match RelativePathBuf::new(name.as_str()) { - Ok(p) => p, - Err(e) => return Some(Err(e.into())), - }; - input_relative_path.as_relative_path().join(entry) - } else { - input_relative_path.clone() - }; - Some(Ok(PostRunMismatch::Input { kind, path })) - } - }, - ); - if let Some(result) = input_mismatch { - return result.map(Some); - } - - for (name, stored_value) in &self.tracked_envs { - let current_value = unfiltered_envs - .get(OsStr::new(name.as_str())) - .map(|value| { - let value_str = value.to_str().ok_or_else(|| { - anyhow::anyhow!("tracked env value for {name} is not valid UTF-8") - })?; - Ok::<_, anyhow::Error>(EnvValueHash::new(value_str)) - }) - .transpose()?; - if let Some(mismatch) = - EnvMismatch::compare(name, stored_value.as_ref(), current_value.as_ref()) - { - return Ok(Some(PostRunMismatch::TrackedEnv(mismatch))); - } - } - - for (query, stored_matches) in &self.tracked_env_queries { - let current_matches = match match_env_query(query, unfiltered_envs)? { - EnvQueryValidation::Matches(matches) => matches, - EnvQueryValidation::NonUtf8Value(mismatch) => { - return Ok(Some(PostRunMismatch::TrackedEnvQuery { - query: query.clone(), - mismatch, - })); - } - }; - if let Some(mismatch) = first_env_glob_mismatch(stored_matches, ¤t_matches) { - return Ok(Some(PostRunMismatch::TrackedEnvQuery { - query: query.clone(), - mismatch, - })); - } - } - - Ok(None) - } -} - -/// Build the current match-set for `query` by enumerating the given env -/// snapshot and keeping matching UTF-8 names. If a matching env has a non-UTF-8 -/// value, return a changed mismatch so the stale cache entry is not replayed. -fn match_env_query( - query: &TrackedEnvQuery, - envs: &FxHashMap, Arc>, -) -> anyhow::Result { - Ok(match query { - TrackedEnvQuery::Glob(pattern) => { - let glob = vt_glob::env::EnvGlob::new(pattern.as_str())?; - collect_matching_envs(envs, |name| glob.is_match(name)) - } - TrackedEnvQuery::Prefix(prefix) => { - collect_matching_envs(envs, |name| env_name_starts_with(name, prefix.as_str())) - } - }) -} - -fn collect_matching_envs( - envs: &FxHashMap, Arc>, - is_match: impl Fn(&str) -> bool, -) -> EnvQueryValidation { - let mut matches = BTreeMap::new(); - for (name, value) in envs { - let Some(name_str) = name.to_str() else { - continue; - }; - if !is_match(name_str) { - continue; - } - let Some(value_str) = value.to_str() else { - return EnvQueryValidation::NonUtf8Value(EnvMismatch::Changed { - name: Str::from(name_str), - }); - }; - matches.insert(Str::from(name_str), EnvValueHash::new(value_str)); - } - EnvQueryValidation::Matches(matches) -} - -enum EnvQueryValidation { - Matches(BTreeMap), - NonUtf8Value(EnvMismatch), -} - -#[cfg(not(windows))] -fn env_name_starts_with(name: &str, prefix: &str) -> bool { - name.starts_with(prefix) -} - -#[cfg(windows)] -fn env_name_starts_with(name: &str, prefix: &str) -> bool { - let mut name_chars = name.chars(); - for prefix_char in prefix.chars() { - let Some(name_char) = name_chars.next() else { - return false; - }; - if !name_char.eq_ignore_ascii_case(&prefix_char) { - return false; - } - } - true -} - -/// Find the first deterministic difference between stored and current env -/// glob match-sets. -fn first_env_glob_mismatch( - stored: &BTreeMap, - current: &BTreeMap, -) -> Option { - let mut stored_iter = stored.iter(); - let mut current_iter = current.iter(); - let mut s = stored_iter.next(); - let mut c = current_iter.next(); - - loop { - match (s, c) { - (None, None) => return None, - (Some((name, _)), None) => return Some(EnvMismatch::Removed { name: name.clone() }), - (None, Some((name, _))) => return Some(EnvMismatch::Added { name: name.clone() }), - (Some((sn, sv)), Some((cn, cv))) => match sn.cmp(cn) { - std::cmp::Ordering::Equal => { - if sv != cv { - return Some(EnvMismatch::Changed { name: sn.clone() }); - } - s = stored_iter.next(); - c = current_iter.next(); - } - std::cmp::Ordering::Less => return Some(EnvMismatch::Removed { name: sn.clone() }), - std::cmp::Ordering::Greater => { - return Some(EnvMismatch::Added { name: cn.clone() }); - } - }, - } - } -} - -/// Determine the kind of change between two differing path fingerprints. -/// Caller guarantees `stored != current`. -/// -/// Returns `(kind, entry_name)` where `entry_name` is `Some` for folder changes -/// when a specific added/removed entry can be identified. -fn determine_change_kind<'a>( - stored: &'a PathFingerprint, - current: &'a PathFingerprint, -) -> (InputChangeKind, Option<&'a Str>) { - match (stored, current) { - (PathFingerprint::NotFound, _) => (InputChangeKind::Added, None), - (_, PathFingerprint::NotFound) => (InputChangeKind::Removed, None), - (PathFingerprint::FileContentHash(_), PathFingerprint::FileContentHash(_)) => { - (InputChangeKind::ContentModified, None) - } - (PathFingerprint::Folder(old), PathFingerprint::Folder(new)) => { - determine_folder_change_kind(old.as_ref(), new.as_ref()) - } - // Type changed (file ↔ folder) - _ => (InputChangeKind::Added, None), - } -} - -/// Determine whether a folder change is an addition or removal by comparing entries. -/// Both maps are `BTreeMap` so we iterate them in sorted lockstep. -/// Returns the specific entry name that was added or removed, if identifiable. -fn determine_folder_change_kind<'a>( - old: Option<&'a BTreeMap>, - new: Option<&'a BTreeMap>, -) -> (InputChangeKind, Option<&'a Str>) { - let (Some(old_entries), Some(new_entries)) = (old, new) else { - return (InputChangeKind::Added, None); - }; - - let mut old_iter = old_entries.iter(); - let mut new_iter = new_entries.iter(); - let mut o = old_iter.next(); - let mut n = new_iter.next(); - - loop { - match (o, n) { - (None, None) => return (InputChangeKind::Added, None), - (Some((name, _)), None) => return (InputChangeKind::Removed, Some(name)), - (None, Some((name, _))) => return (InputChangeKind::Added, Some(name)), - (Some((ok, ov)), Some((nk, nv))) => match ok.cmp(nk) { - std::cmp::Ordering::Equal => { - if ov != nv { - return (InputChangeKind::Added, Some(ok)); - } - o = old_iter.next(); - n = new_iter.next(); - } - std::cmp::Ordering::Less => return (InputChangeKind::Removed, Some(ok)), - std::cmp::Ordering::Greater => return (InputChangeKind::Added, Some(nk)), - }, - } - } -} - -/// Check if a directory entry should be ignored in fingerprinting -const fn should_ignore_entry(name: &[u8]) -> bool { - matches!(name, b"." | b".." | b".DS_Store") || name.eq_ignore_ascii_case(b"dist") -} - -/// Fingerprint a single path -pub fn fingerprint_path( - path: &Arc, - path_read: PathRead, -) -> anyhow::Result { - let std_path = path.as_path(); - - let file = match File::open(std_path) { - Ok(file) => file, - Err(err) => { - // On Windows, File::open fails specifically for directories with PermissionDenied - #[cfg(windows)] - { - if err.kind() == io::ErrorKind::PermissionDenied { - // This might be a directory - try reading it as such - return process_directory(std_path, path_read); - } - // On Windows, paths with trailing backslash (from joining empty path) - // fail with NotFound (error code 3). Try as directory in this case. - if err.raw_os_error() == Some(3) && std_path.to_string_lossy().ends_with('\\') { - return process_directory(std_path, path_read); - } - } - if err.kind() != io::ErrorKind::NotFound { - tracing::trace!( - "Uncommon error when opening {:?} for fingerprinting: {}", - std_path, - err - ); - } - // Treat all open errors as NotFound for fingerprinting purposes - return Ok(PathFingerprint::NotFound); - } - }; - - let mut reader = io::BufReader::new(file); - if let Err(io_err) = reader.fill_buf() { - if io_err.kind() != io::ErrorKind::IsADirectory { - return Err(io_err.into()); - } - // Is a directory on Unix - use the optimized nix implementation - #[cfg(unix)] - { - return process_directory_unix(reader.get_ref(), path_read); - } - #[cfg(windows)] - { - return process_directory(std_path, path_read); - } - } - Ok(PathFingerprint::FileContentHash(super::hash::hash_content(reader)?)) -} - -/// Process a directory on Windows using `std::fs::read_dir` -#[cfg(windows)] -#[expect(clippy::disallowed_types, reason = "Windows fallback uses std::path::Path directly")] -fn process_directory( - path: &std::path::Path, - path_read: PathRead, -) -> anyhow::Result { - if !path_read.read_dir_entries { - return Ok(PathFingerprint::Folder(None)); - } - - let mut entries = BTreeMap::new(); - for entry in std::fs::read_dir(path)? { - let entry = entry?; - let name = entry.file_name(); - let name_bytes = name.as_encoded_bytes(); - - if should_ignore_entry(name_bytes) { - continue; - } - - let file_type = entry.file_type()?; - let kind = if file_type.is_file() { - DirEntryKind::File - } else if file_type.is_dir() { - DirEntryKind::Dir - } else { - DirEntryKind::Symlink - }; - - let name_str = name.to_string_lossy(); - entries.insert(Str::from(name_str.as_ref()), kind); - } - - Ok(PathFingerprint::Folder(Some(entries))) -} - -/// Process a directory on Unix using nix for efficiency -#[cfg(unix)] -fn process_directory_unix(file: &File, path_read: PathRead) -> anyhow::Result { - use std::os::fd::AsFd; - - if !path_read.read_dir_entries { - return Ok(PathFingerprint::Folder(None)); - } - - let fd = file.as_fd(); - let mut dir = nix::dir::Dir::from_fd(fd.try_clone_to_owned()?)?; - - let mut entries = BTreeMap::new(); - for entry in dir.iter() { - let entry = entry?; - let name = entry.file_name().to_bytes(); - - if should_ignore_entry(name) { - continue; - } - - let kind = match entry.file_type() { - Some(nix::dir::Type::Directory) => DirEntryKind::Dir, - Some(nix::dir::Type::Symlink) => DirEntryKind::Symlink, - // Treat files and other types as files for fingerprinting - _ => DirEntryKind::File, - }; - - #[expect( - clippy::disallowed_types, - reason = "from_utf8_lossy returns Cow referencing String" - )] - let name_str = String::from_utf8_lossy(name); - entries.insert(Str::from(name_str.as_ref()), kind); - } - - Ok(PathFingerprint::Folder(Some(entries))) -} - -#[cfg(test)] -mod tests { - use std::ffi::{OsStr, OsString}; - - use super::*; - - #[cfg(unix)] - fn non_utf8_os_string() -> OsString { - use std::os::unix::ffi::OsStringExt; - - OsString::from_vec(vec![0xFF]) - } - - #[cfg(windows)] - fn non_utf8_os_string() -> OsString { - use std::os::windows::ffi::OsStringExt; - - OsString::from_wide(&[0xD800]) - } - - #[test] - fn validate_errors_on_current_non_utf8_tracked_env_value() { - let mut tracked_envs = BTreeMap::new(); - tracked_envs.insert(Str::from("PROBE_ENV"), None); - let fingerprint = PostRunFingerprint { tracked_envs, ..PostRunFingerprint::default() }; - - let mut unfiltered_envs = FxHashMap::default(); - unfiltered_envs.insert( - Arc::::from(OsStr::new("PROBE_ENV")), - Arc::::from(non_utf8_os_string()), - ); - - let workspace_root = vt_path::current_dir().expect("cwd"); - let err = fingerprint - .validate(&workspace_root, &unfiltered_envs) - .expect_err("non-UTF-8 tracked env values must error"); - - assert!(err.to_string().contains("tracked env value for PROBE_ENV is not valid UTF-8")); - } - - #[test] - fn validate_reports_current_non_utf8_tracked_env_glob_value_as_changed() { - let mut tracked_env_queries = BTreeMap::new(); - tracked_env_queries.insert(TrackedEnvQuery::Glob(Str::from("PROBE_*")), BTreeMap::new()); - let fingerprint = - PostRunFingerprint { tracked_env_queries, ..PostRunFingerprint::default() }; - - let mut unfiltered_envs = FxHashMap::default(); - unfiltered_envs.insert( - Arc::::from(OsStr::new("PROBE_BAD")), - Arc::::from(non_utf8_os_string()), - ); - - let workspace_root = vt_path::current_dir().expect("cwd"); - let mismatch = - fingerprint.validate(&workspace_root, &unfiltered_envs).expect("validation succeeds"); - - match mismatch { - Some(PostRunMismatch::TrackedEnvQuery { - query, - mismatch: EnvMismatch::Changed { name }, - }) => { - assert_eq!(query, TrackedEnvQuery::Glob(Str::from("PROBE_*"))); - assert_eq!(name.as_str(), "PROBE_BAD"); - } - other => panic!("expected changed tracked env query mismatch, got {other:?}"), - } - } - - #[test] - fn validate_tracked_env_prefix_treats_star_literally() { - let mut tracked_env_queries = BTreeMap::new(); - let mut stored_matches = BTreeMap::new(); - stored_matches.insert(Str::from("PROBE_*A"), EnvValueHash::new("literal")); - tracked_env_queries.insert(TrackedEnvQuery::Prefix(Str::from("PROBE_*")), stored_matches); - let fingerprint = - PostRunFingerprint { tracked_env_queries, ..PostRunFingerprint::default() }; - - let mut unfiltered_envs = FxHashMap::default(); - unfiltered_envs.insert( - Arc::::from(OsStr::new("PROBE_*A")), - Arc::::from(OsStr::new("literal")), - ); - unfiltered_envs.insert( - Arc::::from(OsStr::new("PROBE_XA")), - Arc::::from(OsStr::new("wildcard if interpreted as glob")), - ); - - let workspace_root = vt_path::current_dir().expect("cwd"); - let mismatch = - fingerprint.validate(&workspace_root, &unfiltered_envs).expect("validation succeeds"); - - assert!(mismatch.is_none()); - } - - #[test] - fn validate_ignores_non_utf8_tracked_env_glob_names() { - let mut tracked_env_queries = BTreeMap::new(); - tracked_env_queries.insert(TrackedEnvQuery::Glob(Str::from("PROBE_*")), BTreeMap::new()); - let fingerprint = - PostRunFingerprint { tracked_env_queries, ..PostRunFingerprint::default() }; - - let mut unfiltered_envs = FxHashMap::default(); - unfiltered_envs.insert( - Arc::::from(non_utf8_os_string()), - Arc::::from(OsStr::new("value")), - ); - - let workspace_root = vt_path::current_dir().expect("cwd"); - let mismatch = - fingerprint.validate(&workspace_root, &unfiltered_envs).expect("validation succeeds"); - - assert!(mismatch.is_none()); - } -} diff --git a/crates/vt/src/session/execute/mod.rs b/crates/vt/src/session/execute/mod.rs index ae31c485e..7a136f3af 100644 --- a/crates/vt/src/session/execute/mod.rs +++ b/crates/vt/src/session/execute/mod.rs @@ -1,17 +1,12 @@ mod cache_update; -pub mod fingerprint; -pub mod glob; -mod hash; pub mod pipe; +pub mod post_run; mod scheduler; pub mod spawn; -#[cfg(fspy)] -pub mod tracked_accesses; #[cfg(windows)] mod win_job; use std::{ - collections::BTreeMap, ffi::{OsStr, OsString}, sync::Arc, time::Instant, @@ -19,14 +14,13 @@ use std::{ use futures_util::future::LocalBoxFuture; use tokio_util::sync::CancellationToken; -use vt_glob::path::PathGlobSet; +use vt_fs_fingerprint::TaskFs; use vt_ipc_shared::NODE_CLIENT_PATH_ENV_NAME; -use vt_path::{AbsolutePath, RelativePathBuf}; +use vt_path::AbsolutePath; use vt_plan::{SpawnExecution, cache_metadata::CacheMetadata}; use vt_server::{Recorder, Reports, ServerHandle, StopAccepting, serve}; use self::{ - glob::compute_globbed_inputs, pipe::{PipeSinks, StdOutput, pipe_stdio}, spawn::{ChildHandle, ChildOutcome, SpawnStdio, spawn}, }; @@ -63,6 +57,10 @@ pub enum SpawnOutcome { /// `includes_auto`, which only lives on cache metadata). /// - Cached execution always owns [`PipeWriters`] (piped stdio is forced so /// that output can be captured for replay). +#[expect( + clippy::large_enum_variant, + reason = "one short-lived value per task execution; boxing the cached state would only add indirection" +)] enum ExecutionMode<'a> { Cached { /// Borrowed by [`PipeSinks`] during drain; dropped at end of function. @@ -85,7 +83,9 @@ enum ExecutionMode<'a> { /// a borrow inside [`PipeSinks::capture`]. struct CacheState<'a> { metadata: &'a CacheMetadata, - globbed_inputs: BTreeMap, + /// The run's filesystem story, begun at cache lookup; concluded by the + /// cache-update phase after the child exits. + task_fs: TaskFs<'a>, /// Captured stdout/stderr for cache replay. Written in place during drain; /// always present (possibly empty) once we reach the cache-update phase. std_outputs: Vec, @@ -93,24 +93,19 @@ struct CacheState<'a> { /// available, and fspy path tracing is attached only when auto input or /// output inference needs it. Parts are borrowed in place during the /// wait/join; the struct is never moved out. - tracking: Tracking<'a>, + tracking: Tracking, } /// The IPC server's driver future: resolves with the recorded reports after /// [`StopAccepting::signal`] fires and all in-flight clients drain. type IpcDriver = LocalBoxFuture<'static, Result>; -/// fspy path-tracking state, present only when a cached task needs automatic -/// input or output inference. -struct FspyTracking<'a> { - input_negative_globs: PathGlobSet<'a>, - output_negative_globs: PathGlobSet<'a>, -} - -/// Per-task runner-aware tracking: IPC server handle plus optional fspy state. -/// Lifetime-tied to a single `execute_spawn` call. -struct Tracking<'a> { - fspy: Option>, +/// Per-task runner-aware tracking: IPC server handle plus whether fspy path +/// tracing is attached. Tied to a single `execute_spawn` call. +struct Tracking { + /// fspy path tracing is attached iff a cached task needs automatic input + /// or output inference (`includes_auto` on either side). + fspy: bool, ipc_envs: Vec<(&'static OsStr, OsString)>, ipc_server_fut: IpcDriver, stop_accepting: StopAccepting, @@ -155,28 +150,19 @@ impl<'a> ExecutionMode<'a> { /// `cache_metadata.is_some_and(_)`) at every downstream use site. /// ───────────────────────────────────────────────────────────────────── fn build( - cache_metadata: Option<&'a CacheMetadata>, + cached: Option<(&'a CacheMetadata, TaskFs<'a>)>, stdio_config: StdioConfig, - globbed_inputs: BTreeMap, ) -> Result { - let Some(metadata) = cache_metadata else { + let Some((metadata, task_fs)) = cached else { return Ok(Self::Uncached { pipe_writers: (stdio_config.suggestion == StdioSuggestion::Piped) .then_some(stdio_config.writers), }); }; - let fspy = if metadata.input_config.includes_auto || metadata.output_config.includes_auto { - // Resolve negative globs for fspy path filtering (already - // workspace-root-relative). - let input_negative_globs = PathGlobSet::new(&metadata.input_config.negative_globs) - .map_err(|err| ExecutionError::PostRunFingerprint(err.into()))?; - let output_negative_globs = PathGlobSet::new(&metadata.output_config.negative_globs) - .map_err(|err| ExecutionError::PostRunFingerprint(err.into()))?; - Some(FspyTracking { input_negative_globs, output_negative_globs }) - } else { - None - }; + // fspy path tracing is attached iff auto input or output inference + // needs it. + let fspy = metadata.input_config.includes_auto || metadata.output_config.includes_auto; // Bind runner IPC for every cached task. The merged cache-control API // (`disableCache`) must work even when a task uses explicit inputs and @@ -189,7 +175,7 @@ impl<'a> ExecutionMode<'a> { Ok(Self::Cached { pipe_writers: stdio_config.writers, - state: CacheState { metadata, globbed_inputs, std_outputs: Vec::new(), tracking }, + state: CacheState { metadata, task_fs, std_outputs: Vec::new(), tracking }, }) } @@ -215,7 +201,7 @@ impl<'a> ExecutionMode<'a> { /// whether fspy tracking is on. const fn spawn_config(&self) -> (SpawnStdio, bool) { match self { - Self::Cached { state, .. } => (SpawnStdio::Piped, state.tracking.fspy.is_some()), + Self::Cached { state, .. } => (SpawnStdio::Piped, state.tracking.fspy), Self::Uncached { pipe_writers: Some(_) } => (SpawnStdio::Piped, false), Self::Uncached { pipe_writers: None } => (SpawnStdio::Inherited, false), } @@ -382,8 +368,9 @@ async fn run( // 2. Report execution start with the looked-up cache status (`start()` // runs exactly once on every arm) and either replay the hit — no need - // to execute the command — or carry the globbed inputs into the run. - let (stdio_config, globbed_inputs) = match lookup { + // to execute the command — or carry the run's filesystem story into + // the cache-update phase. + let (stdio_config, cached) = match lookup { CacheLookup::Hit(cached) => { let mut stdio_config = reporter.start(CacheStatus::Hit { replayed_duration: cached.duration }); @@ -395,18 +382,16 @@ async fn run( program_name, )); } - CacheLookup::Miss { miss, globbed_inputs } => { - (reporter.start(CacheStatus::Miss(miss)), globbed_inputs) + CacheLookup::Miss { miss, metadata, task_fs } => { + (reporter.start(CacheStatus::Miss(miss)), Some((metadata, task_fs))) + } + CacheLookup::Disabled => { + (reporter.start(CacheStatus::Disabled(CacheDisabledReason::NoCacheMetadata)), None) } - CacheLookup::Disabled => ( - reporter.start(CacheStatus::Disabled(CacheDisabledReason::NoCacheMetadata)), - BTreeMap::new(), - ), }; // 4. Fold the cache/fspy/stdio decisions into the typed mode. - let mut mode = ExecutionMode::build(cache_metadata, stdio_config, globbed_inputs) - .map_err(Report::failed)?; + let mut mode = ExecutionMode::build(cached, stdio_config).map_err(Report::failed)?; // Measure end-to-end duration here — spawn() doesn't track time. let start = Instant::now(); @@ -498,45 +483,77 @@ async fn run( /// Outcome of the cache-lookup phase. Each variant carries exactly what that /// outcome provides: a hit owns the cached entry to replay, a miss keeps the -/// reason plus the globbed inputs (reused by the cache-update phase after the -/// run), and disabled has neither. -enum CacheLookup { +/// reason plus the begun run (concluded by the cache-update phase after the +/// child exits), and disabled has neither. +enum CacheLookup<'a> { /// Cache hit — the cached entry to replay. Hit(CacheEntryValue), /// Cache miss — the detailed reason (`NotFound` or `FingerprintMismatch`). - Miss { miss: CacheMiss, globbed_inputs: BTreeMap }, + Miss { miss: CacheMiss, metadata: &'a CacheMetadata, task_fs: TaskFs<'a> }, /// Caching is disabled for this task (no cache metadata). Disabled, } -/// Phase 1: compute the globbed inputs and try to hit the cache. -async fn lookup_cache( - cache_metadata: Option<&CacheMetadata>, +/// A cache-lookup failure, reported as an infrastructure error. +const fn lookup_error(source: anyhow::Error) -> Report { + Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source }) +} + +/// Phase 1: fetch the stored entry, begin the run's filesystem story, and +/// decide hit or miss. +/// +/// The fetch comes first — the entry's key does not depend on filesystem +/// state, and `pre_run` wants the stored fingerprints to compare against. +/// A key-level miss arrives already classified (atomically with the fetch). +/// The checks then run in order: input changes (filesystem), tracked envs. +async fn lookup_cache<'a>( + cache_metadata: Option<&'a CacheMetadata>, cache: &ExecutionCache, - workspace_root: &Arc, -) -> Result { - let Some(cache_metadata) = cache_metadata else { + workspace_root: &'a Arc, +) -> Result, Report> { + let Some(metadata) = cache_metadata else { return Ok(CacheLookup::Disabled); }; - // Compute globbed inputs from positive globs at execution time. - // Globs are already workspace-root-relative (resolved at task graph stage). - let globbed_inputs = compute_globbed_inputs( + let entry = cache.fetch_entry(metadata).await.map_err(lookup_error)?; + + let (task_fs, change) = TaskFs::pre_run( workspace_root, - &cache_metadata.input_config.positive_globs, - &cache_metadata.input_config.negative_globs, + &metadata.input_config, + &metadata.output_config, + entry.as_ref().ok().map(|entry| &entry.input_fingerprints), ) - .map_err(|err| { - Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err }) - })?; + .map_err(lookup_error)?; - match cache.try_hit(cache_metadata, &globbed_inputs, workspace_root).await { - Ok(Ok(cached)) => Ok(CacheLookup::Hit(cached)), - Ok(Err(miss)) => Ok(CacheLookup::Miss { miss, globbed_inputs }), - Err(err) => { - Err(Report::failed(ExecutionError::Cache { kind: CacheErrorKind::Lookup, source: err })) - } + let cached = match entry { + Ok(cached) => cached, + Err(miss) => return Ok(CacheLookup::Miss { miss, metadata, task_fs }), + }; + + if let Some(change) = change { + return Ok(CacheLookup::Miss { + miss: CacheMiss::FingerprintMismatch(change.into()), + metadata, + task_fs, + }); } + + if let Some(mismatch) = cached + .tracked_env_fingerprints + .validate_envs(&metadata.unfiltered_envs) + .map_err(lookup_error)? + { + return Ok(CacheLookup::Miss { + miss: CacheMiss::FingerprintMismatch(mismatch.into()), + metadata, + task_fs, + }); + } + + // Remember which entry key served this task so a later key-level miss can + // report what changed. + cache.record_hit(metadata).await.map_err(lookup_error)?; + Ok(CacheLookup::Hit(cached)) } /// Phase 3 (cache hit): replay the captured stdout/stderr and restore the diff --git a/crates/vt/src/session/execute/post_run.rs b/crates/vt/src/session/execute/post_run.rs new file mode 100644 index 000000000..998c04fe5 --- /dev/null +++ b/crates/vt/src/session/execute/post_run.rs @@ -0,0 +1,318 @@ +//! Post-run environment fingerprinting: env values and bulk env queries +//! observed by runner-aware tools during execution, validated again at cache +//! lookup. The filesystem half of post-run fingerprinting lives in +//! [`vt_fs_fingerprint`]. + +use std::{collections::BTreeMap, ffi::OsStr, sync::Arc}; + +use rustc_hash::FxHashMap; +use serde::{Deserialize, Serialize}; +use vt_plan::cache_metadata::EnvValueHash; +use vt_str::Str; +use wincode::{SchemaRead, SchemaWrite}; + +use crate::session::cache::EnvMismatch; + +#[derive( + SchemaWrite, SchemaRead, Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize, +)] +pub enum TrackedEnvQuery { + Glob(Str), + Prefix(Str), +} + +/// Env state observed by runner-aware tools during execution. +/// Used to validate whether cached outputs are still valid. +#[derive(SchemaWrite, SchemaRead, Debug, Default, Serialize)] +pub struct TrackedEnvFingerprints { + /// Env vars observed via runner-aware IPC `getEnv` with `tracked: true`. + /// Key is the env name; value is the env value hash at execution time, or + /// `None` if unset. Validated at cache lookup against the same plan env + /// context that served the original request. + pub tracked_envs: BTreeMap>, + + /// Bulk env queries (`getEnvs`) made with `tracked: true`. + /// Outer key is the query, inner map is the match-set at execution time + /// (name -> value hash). Validated at cache lookup by re-matching against + /// the current env context and comparing the resulting set. + /// + /// Non-UTF-8 env names are never matched, saved, or treated as errors: + /// they are not returned to the client, so their existence cannot affect + /// task behavior. Values are stricter. A matched env must have a UTF-8 + /// value; the JS client errors when querying a matched non-UTF-8 value, + /// and cache-hit validation treats a currently matched non-UTF-8 value as + /// a changed mismatch so stale cached output is not replayed. + pub tracked_env_queries: BTreeMap>, +} + +/// A mismatch between the stored tracked-env fingerprints and the current +/// environment. +#[derive(Debug, Clone)] +pub enum PostRunMismatch { + /// A tool-tracked env var changed value, appeared, or disappeared. + TrackedEnv(EnvMismatch), + /// A tool-tracked bulk env query's match-set changed between runs. Carries + /// the first differing entry in env-name order. + TrackedEnvQuery { query: TrackedEnvQuery, mismatch: EnvMismatch }, +} + +impl TrackedEnvFingerprints { + /// Validates the tracked env state against the unfiltered env context used + /// by runner-aware IPC. `unfiltered_envs` must be the same plan env + /// context that served the original `getEnv` request, not the filtered env + /// passed to the spawned process. + /// + /// Returns `Some(mismatch)` if anything changed, `None` if all valid. + /// Returns an error if a tracked env is currently present but cannot be + /// represented as UTF-8; treating that value as unset would make cache + /// validation unsound. + #[tracing::instrument(level = "debug", skip_all, name = "validate_tracked_envs")] + pub fn validate_envs( + &self, + unfiltered_envs: &FxHashMap, Arc>, + ) -> anyhow::Result> { + for (name, stored_value) in &self.tracked_envs { + let current_value = unfiltered_envs + .get(OsStr::new(name.as_str())) + .map(|value| { + let value_str = value.to_str().ok_or_else(|| { + anyhow::anyhow!("tracked env value for {name} is not valid UTF-8") + })?; + Ok::<_, anyhow::Error>(EnvValueHash::new(value_str)) + }) + .transpose()?; + if let Some(mismatch) = + EnvMismatch::compare(name, stored_value.as_ref(), current_value.as_ref()) + { + return Ok(Some(PostRunMismatch::TrackedEnv(mismatch))); + } + } + + for (query, stored_matches) in &self.tracked_env_queries { + let current_matches = match match_env_query(query, unfiltered_envs)? { + EnvQueryValidation::Matches(matches) => matches, + EnvQueryValidation::NonUtf8Value(mismatch) => { + return Ok(Some(PostRunMismatch::TrackedEnvQuery { + query: query.clone(), + mismatch, + })); + } + }; + if let Some(mismatch) = first_env_glob_mismatch(stored_matches, ¤t_matches) { + return Ok(Some(PostRunMismatch::TrackedEnvQuery { + query: query.clone(), + mismatch, + })); + } + } + + Ok(None) + } +} + +/// Build the current match-set for `query` by enumerating the given env +/// snapshot and keeping matching UTF-8 names. If a matching env has a non-UTF-8 +/// value, return a changed mismatch so the stale cache entry is not replayed. +fn match_env_query( + query: &TrackedEnvQuery, + envs: &FxHashMap, Arc>, +) -> anyhow::Result { + Ok(match query { + TrackedEnvQuery::Glob(pattern) => { + let glob = vt_glob::env::EnvGlob::new(pattern.as_str())?; + collect_matching_envs(envs, |name| glob.is_match(name)) + } + TrackedEnvQuery::Prefix(prefix) => { + collect_matching_envs(envs, |name| env_name_starts_with(name, prefix.as_str())) + } + }) +} + +fn collect_matching_envs( + envs: &FxHashMap, Arc>, + is_match: impl Fn(&str) -> bool, +) -> EnvQueryValidation { + let mut matches = BTreeMap::new(); + for (name, value) in envs { + let Some(name_str) = name.to_str() else { + continue; + }; + if !is_match(name_str) { + continue; + } + let Some(value_str) = value.to_str() else { + return EnvQueryValidation::NonUtf8Value(EnvMismatch::Changed { + name: Str::from(name_str), + }); + }; + matches.insert(Str::from(name_str), EnvValueHash::new(value_str)); + } + EnvQueryValidation::Matches(matches) +} + +enum EnvQueryValidation { + Matches(BTreeMap), + NonUtf8Value(EnvMismatch), +} + +#[cfg(not(windows))] +fn env_name_starts_with(name: &str, prefix: &str) -> bool { + name.starts_with(prefix) +} + +#[cfg(windows)] +fn env_name_starts_with(name: &str, prefix: &str) -> bool { + let mut name_chars = name.chars(); + for prefix_char in prefix.chars() { + let Some(name_char) = name_chars.next() else { + return false; + }; + if !name_char.eq_ignore_ascii_case(&prefix_char) { + return false; + } + } + true +} + +/// Find the first deterministic difference between stored and current env +/// glob match-sets. +fn first_env_glob_mismatch( + stored: &BTreeMap, + current: &BTreeMap, +) -> Option { + let mut stored_iter = stored.iter(); + let mut current_iter = current.iter(); + let mut s = stored_iter.next(); + let mut c = current_iter.next(); + + loop { + match (s, c) { + (None, None) => return None, + (Some((name, _)), None) => return Some(EnvMismatch::Removed { name: name.clone() }), + (None, Some((name, _))) => return Some(EnvMismatch::Added { name: name.clone() }), + (Some((sn, sv)), Some((cn, cv))) => match sn.cmp(cn) { + std::cmp::Ordering::Equal => { + if sv != cv { + return Some(EnvMismatch::Changed { name: sn.clone() }); + } + s = stored_iter.next(); + c = current_iter.next(); + } + std::cmp::Ordering::Less => return Some(EnvMismatch::Removed { name: sn.clone() }), + std::cmp::Ordering::Greater => { + return Some(EnvMismatch::Added { name: cn.clone() }); + } + }, + } + } +} + +#[cfg(test)] +mod tests { + use std::ffi::{OsStr, OsString}; + + use super::*; + + #[cfg(unix)] + fn non_utf8_os_string() -> OsString { + use std::os::unix::ffi::OsStringExt; + + OsString::from_vec(vec![0xFF]) + } + + #[cfg(windows)] + fn non_utf8_os_string() -> OsString { + use std::os::windows::ffi::OsStringExt; + + OsString::from_wide(&[0xD800]) + } + + #[test] + fn validate_errors_on_current_non_utf8_tracked_env_value() { + let mut tracked_envs = BTreeMap::new(); + tracked_envs.insert(Str::from("PROBE_ENV"), None); + let fingerprints = + TrackedEnvFingerprints { tracked_envs, ..TrackedEnvFingerprints::default() }; + + let mut unfiltered_envs = FxHashMap::default(); + unfiltered_envs.insert( + Arc::::from(OsStr::new("PROBE_ENV")), + Arc::::from(non_utf8_os_string()), + ); + + let err = fingerprints + .validate_envs(&unfiltered_envs) + .expect_err("non-UTF-8 tracked env values must error"); + + assert!(err.to_string().contains("tracked env value for PROBE_ENV is not valid UTF-8")); + } + + #[test] + fn validate_reports_current_non_utf8_tracked_env_glob_value_as_changed() { + let mut tracked_env_queries = BTreeMap::new(); + tracked_env_queries.insert(TrackedEnvQuery::Glob(Str::from("PROBE_*")), BTreeMap::new()); + let fingerprints = + TrackedEnvFingerprints { tracked_env_queries, ..TrackedEnvFingerprints::default() }; + + let mut unfiltered_envs = FxHashMap::default(); + unfiltered_envs.insert( + Arc::::from(OsStr::new("PROBE_BAD")), + Arc::::from(non_utf8_os_string()), + ); + + let mismatch = fingerprints.validate_envs(&unfiltered_envs).expect("validation succeeds"); + + match mismatch { + Some(PostRunMismatch::TrackedEnvQuery { + query, + mismatch: EnvMismatch::Changed { name }, + }) => { + assert_eq!(query, TrackedEnvQuery::Glob(Str::from("PROBE_*"))); + assert_eq!(name.as_str(), "PROBE_BAD"); + } + other => panic!("expected changed tracked env query mismatch, got {other:?}"), + } + } + + #[test] + fn validate_tracked_env_prefix_treats_star_literally() { + let mut tracked_env_queries = BTreeMap::new(); + let mut stored_matches = BTreeMap::new(); + stored_matches.insert(Str::from("PROBE_*A"), EnvValueHash::new("literal")); + tracked_env_queries.insert(TrackedEnvQuery::Prefix(Str::from("PROBE_*")), stored_matches); + let fingerprints = + TrackedEnvFingerprints { tracked_env_queries, ..TrackedEnvFingerprints::default() }; + + let mut unfiltered_envs = FxHashMap::default(); + unfiltered_envs.insert( + Arc::::from(OsStr::new("PROBE_*A")), + Arc::::from(OsStr::new("literal")), + ); + unfiltered_envs.insert( + Arc::::from(OsStr::new("PROBE_XA")), + Arc::::from(OsStr::new("wildcard if interpreted as glob")), + ); + + let mismatch = fingerprints.validate_envs(&unfiltered_envs).expect("validation succeeds"); + + assert!(mismatch.is_none()); + } + + #[test] + fn validate_ignores_non_utf8_tracked_env_glob_names() { + let mut tracked_env_queries = BTreeMap::new(); + tracked_env_queries.insert(TrackedEnvQuery::Glob(Str::from("PROBE_*")), BTreeMap::new()); + let fingerprints = + TrackedEnvFingerprints { tracked_env_queries, ..TrackedEnvFingerprints::default() }; + + let mut unfiltered_envs = FxHashMap::default(); + unfiltered_envs.insert( + Arc::::from(non_utf8_os_string()), + Arc::::from(OsStr::new("value")), + ); + + let mismatch = fingerprints.validate_envs(&unfiltered_envs).expect("validation succeeds"); + + assert!(mismatch.is_none()); + } +} diff --git a/crates/vt/src/session/execute/spawn.rs b/crates/vt/src/session/execute/spawn.rs index adff8aac9..c91c6c1df 100644 --- a/crates/vt/src/session/execute/spawn.rs +++ b/crates/vt/src/session/execute/spawn.rs @@ -2,8 +2,8 @@ //! //! [`spawn`] does one thing: hand back the child's stdio pipes plus a //! cancellation-aware `wait` future. Draining the pipes is [`super::pipe`]'s -//! job; normalizing fspy path accesses is [`super::tracked_accesses`]'s (only -//! compiled when `cfg(fspy)` is on). +//! job; the raw path accesses are judged by [`vt_fs_fingerprint`] during the +//! cache update. use std::{ffi::OsStr, io, process::Stdio}; diff --git a/crates/vt/src/session/reporter/summary.rs b/crates/vt/src/session/reporter/summary.rs index c9da7dff2..2d2713498 100644 --- a/crates/vt/src/session/reporter/summary.rs +++ b/crates/vt/src/session/reporter/summary.rs @@ -24,7 +24,7 @@ use crate::session::{ CacheDisabledReason, CacheErrorKind, CacheNotUpdatedReason, CacheStatus, CacheUpdateStatus, ExecutionError, }, - execute::fingerprint::TrackedEnvQuery, + execute::post_run::TrackedEnvQuery, }; // ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ diff --git a/crates/vt_fs_fingerprint/Cargo.toml b/crates/vt_fs_fingerprint/Cargo.toml new file mode 100644 index 000000000..9494a76ce --- /dev/null +++ b/crates/vt_fs_fingerprint/Cargo.toml @@ -0,0 +1,36 @@ +[package] +name = "vt_fs_fingerprint" +version = "0.0.0" +authors.workspace = true +edition.workspace = true +license.workspace = true +publish = false +rust-version.workspace = true + +[dependencies] +anyhow = { workspace = true } +fspy_shared = { workspace = true } +rayon = { workspace = true } +rustc-hash = { workspace = true } +serde = { workspace = true, features = ["derive"] } +thiserror = { workspace = true } +tracing = { workspace = true } +twox-hash = { workspace = true } +vt_glob = { workspace = true } +vt_graph = { workspace = true } +vt_path = { workspace = true } +vt_str = { workspace = true } +wax = { workspace = true } +wincode = { workspace = true, features = ["derive"] } + +[target.'cfg(unix)'.dependencies] +nix = { workspace = true } + +[dev-dependencies] +tempfile = { workspace = true } + +[lints] +workspace = true + +[lib] +doctest = false diff --git a/crates/vt_fs_fingerprint/README.md b/crates/vt_fs_fingerprint/README.md new file mode 100644 index 000000000..434b0ed55 --- /dev/null +++ b/crates/vt_fs_fingerprint/README.md @@ -0,0 +1,41 @@ +# vt_fs_fingerprint + +The filesystem story of one task run, extracted from the execution engine so +it can be reasoned about and tested on its own: what a run read, what it +produced, and whether a previous run's cached result still matches the +filesystem. Everything else about caching — storage, archiving, env tracking, +process spawning — stays with the engine; this crate is the part that decides +what filesystem facts _mean_. + +The API is one typestate with a call per stage: + +- **`TaskFs::pre_run`** — before the task executes: validate its io + configuration, capture the state of its listed inputs, and (when a previous + run's fingerprints were fetched) report the first input that changed since + that run. +- **`TaskFs::post_run`** — after it finished: judge the traced file accesses + and either declare caching unsound (`Conclusion::InputModified` — the task + wrote a path it also read) or return everything the cache should remember + (`Conclusion::Cacheable`: the run's `InputFingerprints` and its outputs). + +`InputFingerprints` is the opaque record linking runs together: `post_run` +produces it, the caller stores it in the cache entry, and a later run's +`pre_run` checks the filesystem against it. + +Why a separate crate: the policy for a path that a task both reads and writes +is unsettled and expected to be rewritten several times. Here it can be driven +directly with synthetic traces and temp directories, instead of only being +observable by running a whole task through the engine. + +Design notes worth knowing: + +- The pre-run snapshot is taken before the task runs as a **soundness** + requirement, not a convenience: a task that modifies one of its own listed + inputs without tracing to catch it must perpetually miss. A snapshot taken + after the run would capture the post-run state and turn that safe perpetual + miss into a false cache hit. +- `pre_run` is handed the previous fingerprints (fetch-first) rather than + exposing a separate comparison method, so comparing after the run — or + against the wrong run's record — is unrepresentable. +- "No filesystem change" from `pre_run` is not yet a cache hit: the engine + still validates tracked environment state, which lives outside this crate. diff --git a/crates/vt/src/collections.rs b/crates/vt_fs_fingerprint/src/collections.rs similarity index 100% rename from crates/vt/src/collections.rs rename to crates/vt_fs_fingerprint/src/collections.rs diff --git a/crates/vt_fs_fingerprint/src/fingerprint.rs b/crates/vt_fs_fingerprint/src/fingerprint.rs new file mode 100644 index 000000000..5b411b17b --- /dev/null +++ b/crates/vt_fs_fingerprint/src/fingerprint.rs @@ -0,0 +1,436 @@ +//! Per-path fingerprints and the run's input-fingerprint record. + +use std::{ + collections::BTreeMap, + fs::File, + io::{self, BufRead}, + sync::Arc, +}; + +use rayon::iter::{IntoParallelRefIterator, ParallelIterator}; +use serde::{Deserialize, Serialize}; +use vt_path::{AbsolutePath, RelativePathBuf}; +use vt_str::Str; +use wincode::{SchemaRead, SchemaWrite}; + +use crate::collections::HashMap; + +/// Path read access info +#[derive(Debug, Clone, Copy)] +pub struct PathRead { + pub read_dir_entries: bool, +} + +/// Fingerprint for a single path (file or directory) +#[derive(SchemaWrite, SchemaRead, PartialEq, Eq, Debug, Clone, Serialize, Deserialize)] +pub enum PathFingerprint { + /// Path was not found when fingerprinting + NotFound, + /// File content hash using `xxHash3_64` + FileContentHash(u64), + /// Directory with optional entry listing. + /// `Folder(None)` means the directory was opened but entries were not read + /// (e.g., for `openat` calls). + /// `Folder(Some(_))` contains the directory entries sorted by name. + Folder(Option>), +} + +/// Kind of directory entry +#[derive(SchemaWrite, SchemaRead, PartialEq, Eq, Debug, Clone, Serialize, Deserialize)] +pub enum DirEntryKind { + File, + Dir, + Symlink, +} + +/// How an input changed since a previous run. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +pub enum InputChangeKind { + /// File content changed but path is the same + ContentModified, + /// New file or folder added + Added, + /// Existing file or folder removed + Removed, +} + +/// The first input found to differ from a previous run. +#[derive(Debug, Clone)] +pub struct InputChange { + pub kind: InputChangeKind, + pub path: RelativePathBuf, +} + +/// Fingerprints of everything a task run read. +/// +/// Opaque: produced by [`TaskFs::post_run`](crate::TaskFs::post_run), stored +/// by the caller, and handed back to a later +/// [`TaskFs::pre_run`](crate::TaskFs::pre_run) to detect input changes. +#[derive(SchemaWrite, SchemaRead, PartialEq, Eq, Debug, Default, Serialize)] +pub struct InputFingerprints { + /// Content hashes of the explicitly-listed inputs, captured before the + /// run (the pre-run snapshot). The read-write-overlap check guarantees + /// the task did not modify them, so they double as the post-run state. + snapshot: BTreeMap, + /// Fingerprints of the inputs the run discovered while it ran (traced + /// reads not already covered by the snapshot). + discovered: HashMap, +} + +impl InputFingerprints { + pub(crate) const fn new( + snapshot: BTreeMap, + discovered: HashMap, + ) -> Self { + Self { snapshot, discovered } + } + + /// Find the first input that differs between this (stored) record and the + /// present: the stored snapshot is diffed against `current_snapshot` + /// (pure), then each discovered input is re-fingerprinted against the + /// filesystem. Check order determines which change gets reported when + /// several exist. + pub(crate) fn find_change( + &self, + current_snapshot: &BTreeMap, + base_dir: &AbsolutePath, + ) -> anyhow::Result> { + if let Some(change) = detect_snapshot_change(&self.snapshot, current_snapshot) { + return Ok(Some(change)); + } + validate_discovered(&self.discovered, base_dir) + } +} + +/// Fingerprint the inputs a run discovered: every traced read not already in +/// the snapshot. Paths in the snapshot are skipped — they are already tracked +/// by the pre-run hashes, and the read-write overlap check guarantees the task +/// did not modify them, so the pre-run hash is still correct. +pub fn fingerprint_discovered( + path_reads: &HashMap, + base_dir: &AbsolutePath, + snapshot: &BTreeMap, +) -> anyhow::Result> { + path_reads + .par_iter() + .filter(|(path, _)| !snapshot.contains_key(*path)) + .map(|(relative_path, path_read)| { + let full_path = Arc::::from(base_dir.join(relative_path)); + let fingerprint = fingerprint_path(&full_path, *path_read)?; + Ok((relative_path.clone(), fingerprint)) + }) + .collect::>>() +} + +/// Re-fingerprint each stored discovered input against the current filesystem +/// state, returning the first mismatch found (parallel, so the pick among +/// several concurrent mismatches is nondeterministic — intentional). +fn validate_discovered( + discovered: &HashMap, + base_dir: &AbsolutePath, +) -> anyhow::Result> { + let change = discovered.par_iter().find_map_any(|(input_relative_path, path_fingerprint)| { + let input_full_path = Arc::::from(base_dir.join(input_relative_path)); + let path_read = PathRead { + read_dir_entries: matches!(path_fingerprint, PathFingerprint::Folder(Some(_))), + }; + let current_path_fingerprint = match fingerprint_path(&input_full_path, path_read) { + Ok(ok) => ok, + Err(err) => return Some(Err(err)), + }; + if path_fingerprint == ¤t_path_fingerprint { + None + } else { + let (kind, entry_name) = + determine_change_kind(path_fingerprint, ¤t_path_fingerprint); + let path = if let Some(name) = entry_name { + // For folder changes, build `dir/entry` path + let entry = match RelativePathBuf::new(name.as_str()) { + Ok(p) => p, + Err(e) => return Some(Err(e.into())), + }; + input_relative_path.as_relative_path().join(entry) + } else { + input_relative_path.clone() + }; + Some(Ok(InputChange { kind, path })) + } + }); + change.transpose() +} + +/// Compare stored and current snapshot hashes, returning the first changed path. +/// Both maps are `BTreeMap` so we iterate them in sorted lockstep. +fn detect_snapshot_change( + stored: &BTreeMap, + current: &BTreeMap, +) -> Option { + let mut stored_iter = stored.iter(); + let mut current_iter = current.iter(); + let mut s = stored_iter.next(); + let mut c = current_iter.next(); + + loop { + match (s, c) { + (None, None) => return None, + (Some((sp, _)), None) => { + return Some(InputChange { kind: InputChangeKind::Removed, path: sp.clone() }); + } + (None, Some((cp, _))) => { + return Some(InputChange { kind: InputChangeKind::Added, path: cp.clone() }); + } + (Some((sp, sh)), Some((cp, ch))) => match sp.cmp(cp) { + std::cmp::Ordering::Equal => { + if sh != ch { + return Some(InputChange { + kind: InputChangeKind::ContentModified, + path: sp.clone(), + }); + } + s = stored_iter.next(); + c = current_iter.next(); + } + std::cmp::Ordering::Less => { + return Some(InputChange { kind: InputChangeKind::Removed, path: sp.clone() }); + } + std::cmp::Ordering::Greater => { + return Some(InputChange { kind: InputChangeKind::Added, path: cp.clone() }); + } + }, + } + } +} + +/// Determine the kind of change between two differing path fingerprints. +/// Caller guarantees `stored != current`. +/// +/// Returns `(kind, entry_name)` where `entry_name` is `Some` for folder changes +/// when a specific added/removed entry can be identified. +fn determine_change_kind<'a>( + stored: &'a PathFingerprint, + current: &'a PathFingerprint, +) -> (InputChangeKind, Option<&'a Str>) { + match (stored, current) { + (PathFingerprint::NotFound, _) => (InputChangeKind::Added, None), + (_, PathFingerprint::NotFound) => (InputChangeKind::Removed, None), + (PathFingerprint::FileContentHash(_), PathFingerprint::FileContentHash(_)) => { + (InputChangeKind::ContentModified, None) + } + (PathFingerprint::Folder(old), PathFingerprint::Folder(new)) => { + determine_folder_change_kind(old.as_ref(), new.as_ref()) + } + // Type changed (file ↔ folder) + _ => (InputChangeKind::Added, None), + } +} + +/// Determine whether a folder change is an addition or removal by comparing entries. +/// Both maps are `BTreeMap` so we iterate them in sorted lockstep. +/// Returns the specific entry name that was added or removed, if identifiable. +fn determine_folder_change_kind<'a>( + old: Option<&'a BTreeMap>, + new: Option<&'a BTreeMap>, +) -> (InputChangeKind, Option<&'a Str>) { + let (Some(old_entries), Some(new_entries)) = (old, new) else { + return (InputChangeKind::Added, None); + }; + + let mut old_iter = old_entries.iter(); + let mut new_iter = new_entries.iter(); + let mut o = old_iter.next(); + let mut n = new_iter.next(); + + loop { + match (o, n) { + (None, None) => return (InputChangeKind::Added, None), + (Some((name, _)), None) => return (InputChangeKind::Removed, Some(name)), + (None, Some((name, _))) => return (InputChangeKind::Added, Some(name)), + (Some((ok, ov)), Some((nk, nv))) => match ok.cmp(nk) { + std::cmp::Ordering::Equal => { + if ov != nv { + return (InputChangeKind::Added, Some(ok)); + } + o = old_iter.next(); + n = new_iter.next(); + } + std::cmp::Ordering::Less => return (InputChangeKind::Removed, Some(ok)), + std::cmp::Ordering::Greater => return (InputChangeKind::Added, Some(nk)), + }, + } + } +} + +/// Check if a directory entry should be ignored in fingerprinting +const fn should_ignore_entry(name: &[u8]) -> bool { + matches!(name, b"." | b".." | b".DS_Store") || name.eq_ignore_ascii_case(b"dist") +} + +/// Fingerprint a single path +pub fn fingerprint_path( + path: &Arc, + path_read: PathRead, +) -> anyhow::Result { + let std_path = path.as_path(); + + let file = match File::open(std_path) { + Ok(file) => file, + Err(err) => { + // On Windows, File::open fails specifically for directories with PermissionDenied + #[cfg(windows)] + { + if err.kind() == io::ErrorKind::PermissionDenied { + // This might be a directory - try reading it as such + return process_directory(std_path, path_read); + } + // On Windows, paths with trailing backslash (from joining empty path) + // fail with NotFound (error code 3). Try as directory in this case. + if err.raw_os_error() == Some(3) && std_path.to_string_lossy().ends_with('\\') { + return process_directory(std_path, path_read); + } + } + if err.kind() != io::ErrorKind::NotFound { + tracing::trace!( + "Uncommon error when opening {:?} for fingerprinting: {}", + std_path, + err + ); + } + // Treat all open errors as NotFound for fingerprinting purposes + return Ok(PathFingerprint::NotFound); + } + }; + + let mut reader = io::BufReader::new(file); + if let Err(io_err) = reader.fill_buf() { + if io_err.kind() != io::ErrorKind::IsADirectory { + return Err(io_err.into()); + } + // Is a directory on Unix - use the optimized nix implementation + #[cfg(unix)] + { + return process_directory_unix(reader.get_ref(), path_read); + } + #[cfg(windows)] + { + return process_directory(std_path, path_read); + } + } + Ok(PathFingerprint::FileContentHash(crate::hash::hash_content(reader)?)) +} + +/// Process a directory on Windows using `std::fs::read_dir` +#[cfg(windows)] +#[expect(clippy::disallowed_types, reason = "Windows fallback uses std::path::Path directly")] +fn process_directory( + path: &std::path::Path, + path_read: PathRead, +) -> anyhow::Result { + if !path_read.read_dir_entries { + return Ok(PathFingerprint::Folder(None)); + } + + let mut entries = BTreeMap::new(); + for entry in std::fs::read_dir(path)? { + let entry = entry?; + let name = entry.file_name(); + let name_bytes = name.as_encoded_bytes(); + + if should_ignore_entry(name_bytes) { + continue; + } + + let file_type = entry.file_type()?; + let kind = if file_type.is_file() { + DirEntryKind::File + } else if file_type.is_dir() { + DirEntryKind::Dir + } else { + DirEntryKind::Symlink + }; + + let name_str = name.to_string_lossy(); + entries.insert(Str::from(name_str.as_ref()), kind); + } + + Ok(PathFingerprint::Folder(Some(entries))) +} + +/// Process a directory on Unix using nix for efficiency +#[cfg(unix)] +fn process_directory_unix(file: &File, path_read: PathRead) -> anyhow::Result { + use std::os::fd::AsFd; + + if !path_read.read_dir_entries { + return Ok(PathFingerprint::Folder(None)); + } + + let fd = file.as_fd(); + let mut dir = nix::dir::Dir::from_fd(fd.try_clone_to_owned()?)?; + + let mut entries = BTreeMap::new(); + for entry in dir.iter() { + let entry = entry?; + let name = entry.file_name().to_bytes(); + + if should_ignore_entry(name) { + continue; + } + + let kind = match entry.file_type() { + Some(nix::dir::Type::Directory) => DirEntryKind::Dir, + Some(nix::dir::Type::Symlink) => DirEntryKind::Symlink, + // Treat files and other types as files for fingerprinting + _ => DirEntryKind::File, + }; + + #[expect( + clippy::disallowed_types, + reason = "from_utf8_lossy returns Cow referencing String" + )] + let name_str = String::from_utf8_lossy(name); + entries.insert(Str::from(name_str.as_ref()), kind); + } + + Ok(PathFingerprint::Folder(Some(entries))) +} + +#[cfg(test)] +mod tests { + use super::*; + + /// The opaque record must survive the cache's serialization round trip + /// with every fingerprint shape it can carry. + #[test] + fn input_fingerprints_wincode_round_trip() { + let mut snapshot = BTreeMap::new(); + snapshot.insert(RelativePathBuf::new("src/a.txt").unwrap(), 42u64); + + let mut entries = BTreeMap::new(); + entries.insert(Str::from("child.txt"), DirEntryKind::File); + entries.insert(Str::from("nested"), DirEntryKind::Dir); + entries.insert(Str::from("link"), DirEntryKind::Symlink); + + let mut discovered = HashMap::default(); + discovered + .insert(RelativePathBuf::new("read.txt").unwrap(), PathFingerprint::FileContentHash(7)); + discovered.insert(RelativePathBuf::new("probed").unwrap(), PathFingerprint::NotFound); + discovered + .insert(RelativePathBuf::new("opened-dir").unwrap(), PathFingerprint::Folder(None)); + discovered.insert( + RelativePathBuf::new("listed-dir").unwrap(), + PathFingerprint::Folder(Some(entries)), + ); + + let fingerprints = InputFingerprints::new(snapshot, discovered); + + let config = wincode::config::Configuration::default(); + let bytes = wincode::config::serialize(&fingerprints, config).unwrap(); + let back: InputFingerprints = wincode::config::deserialize_exact(&bytes, config).unwrap(); + assert_eq!(fingerprints, back); + + let empty = InputFingerprints::default(); + let bytes = wincode::config::serialize(&empty, config).unwrap(); + let back: InputFingerprints = wincode::config::deserialize_exact(&bytes, config).unwrap(); + assert_eq!(empty, back); + } +} diff --git a/crates/vt/src/session/execute/glob.rs b/crates/vt_fs_fingerprint/src/glob.rs similarity index 98% rename from crates/vt/src/session/execute/glob.rs rename to crates/vt_fs_fingerprint/src/glob.rs index cf5287193..6490192fc 100644 --- a/crates/vt/src/session/execute/glob.rs +++ b/crates/vt_fs_fingerprint/src/glob.rs @@ -1,5 +1,5 @@ -//! Glob-based file discovery used by cache input fingerprinting and output -//! archiving. +//! Glob-based file discovery used by input fingerprinting and output +//! collection. //! //! Both rely on the same walker — positive globs collect candidate files, //! negative globs filter them out. The input path adds per-file content @@ -101,7 +101,7 @@ pub fn compute_globbed_inputs( /// Collect file paths matching positive globs, filtered by negative globs. /// /// Like [`compute_globbed_inputs`] but only collects paths (no hashing). -/// Used for determining which output files to archive. +/// Used for determining which output files a run produced. pub fn collect_glob_paths( root: &AbsolutePath, positive_globs: &std::collections::BTreeSet, @@ -130,7 +130,7 @@ fn strip_root( } fn hash_file_content(path: &AbsolutePath) -> io::Result { - super::hash::hash_content(io::BufReader::new(File::open(path.as_path())?)) + crate::hash::hash_content(io::BufReader::new(File::open(path.as_path())?)) } #[cfg(test)] diff --git a/crates/vt/src/session/execute/hash.rs b/crates/vt_fs_fingerprint/src/hash.rs similarity index 81% rename from crates/vt/src/session/execute/hash.rs rename to crates/vt_fs_fingerprint/src/hash.rs index 0bbfb0fac..2ba55b56d 100644 --- a/crates/vt/src/session/execute/hash.rs +++ b/crates/vt_fs_fingerprint/src/hash.rs @@ -1,7 +1,7 @@ use std::{hash::Hasher as _, io}; /// Hash content using 8 KiB buffered `xxHash3_64`. -pub(super) fn hash_content(mut stream: impl io::Read) -> io::Result { +pub fn hash_content(mut stream: impl io::Read) -> io::Result { let mut hasher = twox_hash::XxHash3_64::default(); let mut buf = [0u8; 8192]; loop { diff --git a/crates/vt_fs_fingerprint/src/lib.rs b/crates/vt_fs_fingerprint/src/lib.rs new file mode 100644 index 000000000..5da6dfb20 --- /dev/null +++ b/crates/vt_fs_fingerprint/src/lib.rs @@ -0,0 +1,22 @@ +//! Filesystem fingerprinting for task caching. +//! +//! One run of a task, from the filesystem's point of view: +//! +//! - [`TaskFs::pre_run`] — before the task executes: capture the state of its +//! configured inputs and detect what changed since a previous run. +//! - [`TaskFs::post_run`] — after it finished: judge the traced file accesses +//! and produce everything the cache should remember ([`Conclusion`]). +//! +//! [`InputFingerprints`] is the opaque record that links the two: a run's +//! `post_run` produces it, the caller stores it, and a later run's `pre_run` +//! checks the filesystem against it. + +mod collections; +mod fingerprint; +mod glob; +mod hash; +mod task_run; +mod tracked_accesses; + +pub use fingerprint::{InputChange, InputChangeKind, InputFingerprints}; +pub use task_run::{Conclusion, PostRunError, TaskFs}; diff --git a/crates/vt_fs_fingerprint/src/task_run.rs b/crates/vt_fs_fingerprint/src/task_run.rs new file mode 100644 index 000000000..2bcff3b36 --- /dev/null +++ b/crates/vt_fs_fingerprint/src/task_run.rs @@ -0,0 +1,861 @@ +//! One run of a task, from the filesystem's point of view. + +use std::{ + collections::{BTreeMap, BTreeSet}, + sync::Arc, +}; + +use fspy_shared::ipc::PathAccess; +use rustc_hash::FxHashSet; +use vt_glob::path::PathGlobSet; +use vt_graph::config::ResolvedGlobConfig; +use vt_path::{AbsolutePath, RelativePathBuf}; +use vt_str::Str; + +use crate::{ + collections::HashMap, + fingerprint::{self, InputChange, InputFingerprints, PathRead}, + glob, + tracked_accesses::TrackedPathAccesses, +}; + +/// One run of a task, from the filesystem's point of view. +/// +/// Two stages: [`pre_run`](Self::pre_run) before the task executes, +/// [`post_run`](Self::post_run) after it finished. The final reuse decision +/// belongs to the caller — a run with no input change may still be +/// invalidated by checks outside this crate. +pub struct TaskFs<'a> { + workspace_root: &'a AbsolutePath, + input: &'a ResolvedGlobConfig, + output: &'a ResolvedGlobConfig, + /// Present iff either side has auto tracking — the same condition under + /// which a trace is attached, so filters exist whenever a side counts. + auto_filters: Option>, + /// Content hashes of the listed inputs, captured before the run. + snapshot: BTreeMap, +} + +/// Compiled negative globs for filtering traced accesses, one per side. +struct AutoFilters<'a> { + input_negative_globs: PathGlobSet<'a>, + output_negative_globs: PathGlobSet<'a>, +} + +/// How the run ended, from the filesystem's point of view. +#[derive(Debug)] +pub enum Conclusion { + /// The task wrote a path it also read: the inputs it started from no + /// longer exist as such, so caching this run would be unsound. + InputModified { path: RelativePathBuf }, + /// Caching is sound; this is what the cache should remember. + Cacheable { + /// Fingerprints of everything the run read. + input_fingerprints: InputFingerprints, + /// The files the run produced, sorted. + outputs: Vec, + }, +} + +/// Which half of [`TaskFs::post_run`] failed. +#[derive(Debug, thiserror::Error)] +pub enum PostRunError { + /// The run's input fingerprints could not be computed. + #[error(transparent)] + InputFingerprints(anyhow::Error), + /// The run's outputs could not be collected. + #[error(transparent)] + Outputs(anyhow::Error), +} + +impl<'a> TaskFs<'a> { + /// Before the task runs: validate the task's io configuration, capture + /// the current state of its listed inputs, and — when a previous run's + /// fingerprints are given — report the first input that changed since + /// that run. + /// + /// A returned change of `None` means no input changed on the filesystem + /// (trivially so when `previous` is `None`); whether the previous result + /// can be reused may involve further checks by the caller. + /// + /// # Errors + /// + /// Fails when the io configuration is invalid or the inputs' state cannot + /// be read — in both cases before the task runs. + #[tracing::instrument(level = "debug", skip_all)] + pub fn pre_run( + workspace_root: &'a AbsolutePath, + input: &'a ResolvedGlobConfig, + output: &'a ResolvedGlobConfig, + previous: Option<&InputFingerprints>, + ) -> anyhow::Result<(Self, Option)> { + // Negative globs are only ever matched against traced accesses, so + // they are compiled exactly when a trace will exist (either side has + // auto). Tasks without auto never pay for — or fail on — this. + let auto_filters = if input.includes_auto || output.includes_auto { + Some(AutoFilters { + input_negative_globs: PathGlobSet::new(&input.negative_globs)?, + output_negative_globs: PathGlobSet::new(&output.negative_globs)?, + }) + } else { + None + }; + + let snapshot = glob::compute_globbed_inputs( + workspace_root, + &input.positive_globs, + &input.negative_globs, + )?; + + let task_fs = Self { workspace_root, input, output, auto_filters, snapshot }; + let change = previous + .map(|previous| previous.find_change(&task_fs.snapshot, workspace_root)) + .transpose()? + .flatten(); + Ok((task_fs, change)) + } + + /// After the task ran: judge the traced file accesses and produce what + /// the cache should remember. + /// + /// Traced reads and writes each count only for a side whose configuration + /// has auto tracking, and are filtered by that side's negative globs and + /// the tool-reported ignore paths. `accesses: None` means the run was not + /// traced; the configured outputs are still collected. + /// + /// # Errors + /// + /// Fails when the run's input fingerprints or outputs cannot be collected + /// from the filesystem; the [`PostRunError`] variant says which half. + #[tracing::instrument(level = "debug", skip_all)] + pub fn post_run<'r>( + self, + accesses: Option>>, + reported_ignored_inputs: &FxHashSet>, + reported_ignored_outputs: &FxHashSet>, + ) -> Result { + let tracked = accesses + .map(|raw| TrackedPathAccesses::from_raw(raw, self.workspace_root)) + .unwrap_or_default(); + + // A trace can be attached for auto-output-only tasks; in that mode + // reads must not become discovered inputs. Symmetrically, writes must + // not become outputs (or overlap candidates) for auto-input-only + // tasks. Each side is therefore gated on its own config. + let path_reads: HashMap = if self.input.includes_auto + && let Some(filters) = &self.auto_filters + { + let ignored = normalize_ignored_paths(reported_ignored_inputs, self.workspace_root); + tracked + .path_reads + .iter() + .filter(|(path, _)| { + !filters.input_negative_globs.is_match(path.as_str()) + && !is_ignored(path, &ignored) + }) + .map(|(path, read)| (path.clone(), *read)) + .collect() + } else { + HashMap::default() + }; + let path_writes: FxHashSet = if self.output.includes_auto + && let Some(filters) = &self.auto_filters + { + let ignored = normalize_ignored_paths(reported_ignored_outputs, self.workspace_root); + tracked + .path_writes + .iter() + .filter(|path| { + !filters.output_negative_globs.is_match(path.as_str()) + && !is_ignored(path, &ignored) + }) + .cloned() + .collect() + } else { + FxHashSet::default() + }; + + // The verdict, checked before any fingerprinting so a doomed run does + // no filesystem work: a path both read and written means the pre-run + // snapshot is stale and caching would be unsound. Exact-path equality + // only. (Only traced reads are checked, not the snapshot: a task that + // writes a listed input it never reads causes perpetual cache misses, + // which is wasteful but not a correctness bug.) + if let Some(path) = path_reads.keys().find(|path| path_writes.contains(*path)).cloned() { + return Ok(Conclusion::InputModified { path }); + } + + let discovered = + fingerprint::fingerprint_discovered(&path_reads, self.workspace_root, &self.snapshot) + .map_err(PostRunError::InputFingerprints)?; + + let outputs = collect_outputs( + path_writes, + self.workspace_root, + &self.output.positive_globs, + &self.output.negative_globs, + ) + .map_err(PostRunError::Outputs)?; + + Ok(Conclusion::Cacheable { + input_fingerprints: InputFingerprints::new(self.snapshot, discovered), + outputs, + }) + } +} + +/// Files the run produced: filtered traced writes ∪ configured output-glob +/// matches, sorted. +fn collect_outputs( + writes: FxHashSet, + root: &AbsolutePath, + positive_globs: &BTreeSet, + negative_globs: &BTreeSet, +) -> anyhow::Result> { + let mut files = writes; + if !positive_globs.is_empty() { + files.extend(glob::collect_glob_paths(root, positive_globs, negative_globs)?); + } + let mut sorted: Vec = files.into_iter().collect(); + sorted.sort(); + Ok(sorted) +} + +/// Normalize tool-reported absolute paths to cleaned workspace-relative paths. +/// Paths outside the workspace are dropped — they can't contribute to inputs +/// or outputs. +fn normalize_ignored_paths( + paths: &FxHashSet>, + workspace_root: &AbsolutePath, +) -> FxHashSet { + paths + .iter() + .filter_map(|p| p.strip_prefix(workspace_root).ok().flatten()?.clean().ok()) + .collect() +} + +/// Whether `path` is covered by any `ignored` entry. An ignored entry matches +/// itself (exact file) and everything under it (directory subtree). +fn is_ignored(path: &RelativePathBuf, ignored: &FxHashSet) -> bool { + if ignored.is_empty() { + return false; + } + ignored.contains(path) || ignored.iter().any(|ig| path.strip_prefix(ig).is_some()) +} + +#[cfg(test)] +mod tests { + use std::{ffi::OsStr, fs}; + + use fspy_shared::ipc::AccessMode; + use tempfile::TempDir; + use vt_path::AbsolutePathBuf; + + use super::*; + use crate::InputChangeKind; + + fn workspace() -> (TempDir, AbsolutePathBuf) { + let tmp = TempDir::new().unwrap(); + let root = AbsolutePathBuf::new(tmp.path().to_path_buf()).unwrap(); + (tmp, root) + } + + fn config(auto: bool, positive: &[&str], negative: &[&str]) -> ResolvedGlobConfig { + ResolvedGlobConfig { + includes_auto: auto, + positive_globs: positive.iter().map(|s| (*s).into()).collect(), + negative_globs: negative.iter().map(|s| (*s).into()).collect(), + } + } + + fn rel(s: &str) -> RelativePathBuf { + RelativePathBuf::new(s).unwrap() + } + + fn no_ignored() -> FxHashSet> { + FxHashSet::default() + } + + /// Owns the backing storage for synthetic traced accesses, so tests can + /// drive `post_run` without spawning a traced process. + #[derive(Default)] + struct Trace { + #[cfg(unix)] + entries: Vec<(AccessMode, std::ffi::OsString)>, + #[cfg(windows)] + entries: Vec<(AccessMode, Vec)>, + } + + impl Trace { + fn push(&mut self, mode: AccessMode, path: impl AsRef) { + #[cfg(unix)] + self.entries.push((mode, path.as_ref().to_os_string())); + #[cfg(windows)] + { + use std::os::windows::ffi::OsStrExt as _; + self.entries.push((mode, path.as_ref().encode_wide().collect())); + } + } + + fn accesses(&self) -> impl Iterator> { + self.entries.iter().map(|(mode, path)| { + #[cfg(unix)] + let path: &fspy_shared::ipc::NativePath = path.into(); + #[cfg(windows)] + let path = fspy_shared::ipc::NativePath::from_wide(path); + PathAccess { mode: *mode, path } + }) + } + } + + /// `pre_run` with no previous run; asserts the trivially-empty change. + fn first_run<'a>( + root: &'a AbsolutePath, + input: &'a ResolvedGlobConfig, + output: &'a ResolvedGlobConfig, + ) -> TaskFs<'a> { + let (task_fs, change) = TaskFs::pre_run(root, input, output, None).unwrap(); + assert!(change.is_none(), "a first run has nothing to differ from"); + task_fs + } + + fn cacheable(conclusion: Conclusion) -> (InputFingerprints, Vec) { + match conclusion { + Conclusion::Cacheable { input_fingerprints, outputs } => (input_fingerprints, outputs), + Conclusion::InputModified { path } => panic!("unexpected InputModified at {path}"), + } + } + + /// Run the whole pipeline once: trace → conclude, returning the stored + /// fingerprints for a later run to validate against. + fn conclude( + root: &AbsolutePath, + input: &ResolvedGlobConfig, + output: &ResolvedGlobConfig, + trace: &Trace, + ) -> (InputFingerprints, Vec) { + let task_fs = first_run(root, input, output); + cacheable(task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap()) + } + + /// What changed since the run that stored `previous`? + fn change_since( + root: &AbsolutePath, + input: &ResolvedGlobConfig, + output: &ResolvedGlobConfig, + previous: &InputFingerprints, + ) -> Option { + let (_task_fs, change) = TaskFs::pre_run(root, input, output, Some(previous)).unwrap(); + change + } + + // ── the verdict ───────────────────────────────────────────────────────── + + #[test] + fn read_write_same_path_is_input_modified() { + let (_tmp, root) = workspace(); + fs::write(root.join("file.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("file.txt").as_path()); + trace.push(AccessMode::WRITE, root.join("file.txt").as_path()); + + let conclusion = + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(); + assert!( + matches!(conclusion, Conclusion::InputModified { path } if path == rel("file.txt")) + ); + } + + #[test] + fn single_read_write_access_is_input_modified() { + let (_tmp, root) = workspace(); + fs::write(root.join("file.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + // One O_RDWR-style access carrying both bits. + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("file.txt").as_path()); + + let conclusion = + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(); + assert!( + matches!(conclusion, Conclusion::InputModified { path } if path == rel("file.txt")) + ); + } + + #[test] + fn reading_dir_and_writing_child_is_not_overlap() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("sub")).unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("sub").as_path()); + trace.push(AccessMode::WRITE, root.join("sub/out.txt").as_path()); + + let (_fingerprints, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert_eq!(outputs, vec![rel("sub/out.txt")]); + } + + #[test] + fn input_negative_glob_dissolves_overlap() { + let (_tmp, root) = workspace(); + fs::write(root.join("build.log"), "x").unwrap(); + let input = config(true, &[], &["**/*.log"]); + let output = config(true, &[], &[]); + + let task_fs = first_run(&root, &input, &output); + let mut trace = Trace::default(); + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("build.log").as_path()); + + // The read is excluded by the input negative, so no overlap remains; + // the write still counts as an output. + let (_fingerprints, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert_eq!(outputs, vec![rel("build.log")]); + } + + #[test] + fn output_negative_glob_dissolves_overlap() { + let (_tmp, root) = workspace(); + fs::write(root.join("build.log"), "x").unwrap(); + let input = config(true, &[], &[]); + let output = config(true, &[], &["**/*.log"]); + + let task_fs = first_run(&root, &input, &output); + let mut trace = Trace::default(); + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("build.log").as_path()); + + let (_fingerprints, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert!(outputs.is_empty(), "the write is excluded by the output negative"); + } + + #[test] + fn reported_ignored_input_dissolves_overlap() { + let (_tmp, root) = workspace(); + fs::write(root.join("state.json"), "x").unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("state.json").as_path()); + + let mut ignored = FxHashSet::default(); + ignored.insert(Arc::::from(root.join("state.json"))); + + let (_fingerprints, outputs) = + cacheable(task_fs.post_run(Some(trace.accesses()), &ignored, &no_ignored()).unwrap()); + assert_eq!(outputs, vec![rel("state.json")]); + } + + #[test] + fn reported_ignored_input_covers_subtree() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("gen")).unwrap(); + fs::write(root.join("gen/state.json"), "x").unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("gen/state.json").as_path()); + + let mut ignored = FxHashSet::default(); + ignored.insert(Arc::::from(root.join("gen"))); + + let conclusion = task_fs.post_run(Some(trace.accesses()), &ignored, &no_ignored()).unwrap(); + assert!(matches!(conclusion, Conclusion::Cacheable { .. })); + } + + // ── per-side gating ───────────────────────────────────────────────────── + + #[test] + fn input_auto_only_writes_dont_count() { + let (_tmp, root) = workspace(); + fs::write(root.join("file.txt"), "x").unwrap(); + let input = config(true, &[], &[]); + let output = config(false, &[], &[]); + + let task_fs = first_run(&root, &input, &output); + let mut trace = Trace::default(); + trace.push(AccessMode::READ | AccessMode::WRITE, root.join("file.txt").as_path()); + + // Writes don't count for a non-auto output side: no overlap verdict, + // and no traced outputs. + let (_fingerprints, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert!(outputs.is_empty()); + } + + #[test] + fn output_auto_only_reads_dont_count() { + let (_tmp, root) = workspace(); + fs::write(root.join("read.txt"), "x").unwrap(); + let input = config(false, &[], &[]); + let output = config(true, &[], &[]); + + let task_fs = first_run(&root, &input, &output); + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("read.txt").as_path()); + trace.push(AccessMode::WRITE, root.join("written.txt").as_path()); + + let (previous, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert_eq!(outputs, vec![rel("written.txt")]); + + // The read was never fingerprinted, so changing it goes unnoticed. + fs::write(root.join("read.txt"), "changed").unwrap(); + assert!(change_since(&root, &input, &output, &previous).is_none()); + } + + #[test] + fn untraced_run_still_collects_configured_outputs() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("out")).unwrap(); + fs::write(root.join("out/a.js"), "x").unwrap(); + let input = config(false, &[], &[]); + let output = config(false, &["out/**"], &[]); + + let task_fs = first_run(&root, &input, &output); + let conclusion = task_fs + .post_run(None::>>, &no_ignored(), &no_ignored()) + .unwrap(); + let (_fingerprints, outputs) = cacheable(conclusion); + assert_eq!(outputs, vec![rel("out/a.js")]); + } + + #[test] + fn outputs_union_deduplicates_and_sorts() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("out")).unwrap(); + fs::write(root.join("out/a.js"), "a").unwrap(); + fs::write(root.join("out/b.js"), "b").unwrap(); + let input = config(false, &[], &[]); + let output = config(true, &["out/**"], &[]); + + let task_fs = first_run(&root, &input, &output); + let mut trace = Trace::default(); + // b.js is both traced and glob-matched; zzz.txt only traced. + trace.push(AccessMode::WRITE, root.join("zzz.txt").as_path()); + trace.push(AccessMode::WRITE, root.join("out/b.js").as_path()); + + let (_fingerprints, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert_eq!(outputs, vec![rel("out/a.js"), rel("out/b.js"), rel("zzz.txt")]); + } + + // ── trace normalization ───────────────────────────────────────────────── + + #[test] + fn accesses_outside_the_workspace_are_dropped() { + let (_tmp, root) = workspace(); + let (_other_tmp, other) = workspace(); + fs::write(other.join("outside.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let task_fs = first_run(&root, &io, &io); + let mut trace = Trace::default(); + trace.push(AccessMode::READ, other.join("outside.txt").as_path()); + trace.push(AccessMode::WRITE, other.join("outside.txt").as_path()); + + // Neither the read-write overlap nor the write registers. + let (previous, outputs) = cacheable( + task_fs.post_run(Some(trace.accesses()), &no_ignored(), &no_ignored()).unwrap(), + ); + assert!(outputs.is_empty()); + + // The outside read was never fingerprinted either. + fs::remove_file(other.join("outside.txt")).unwrap(); + assert!(change_since(&root, &io, &io, &previous).is_none()); + } + + #[test] + fn git_accesses_are_skipped() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join(".git")).unwrap(); + fs::write(root.join(".git/index"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join(".git/index").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::remove_file(root.join(".git/index")).unwrap(); + assert!(change_since(&root, &io, &io, &previous).is_none()); + } + + #[test] + fn parent_components_are_cleaned() { + let (tmp, root) = workspace(); + fs::create_dir(root.join("pkg")).unwrap(); + fs::write(root.join("data.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, tmp.path().join("pkg/../data.txt")); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::write(root.join("data.txt"), "changed").unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::ContentModified, path }) if path == rel("data.txt") + )); + } + + // ── discovered inputs, fingerprinted and re-checked ───────────────────── + + #[test] + fn discovered_content_change_is_reported() { + let (_tmp, root) = workspace(); + fs::write(root.join("file.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("file.txt").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + assert!(change_since(&root, &io, &io, &previous).is_none(), "unchanged file"); + + fs::write(root.join("file.txt"), "changed").unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::ContentModified, path }) if path == rel("file.txt") + )); + } + + #[test] + fn discovered_removal_is_reported() { + let (_tmp, root) = workspace(); + fs::write(root.join("file.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("file.txt").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::remove_file(root.join("file.txt")).unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Removed, path }) if path == rel("file.txt") + )); + } + + #[test] + fn discovered_missing_path_that_appears_is_reported_as_added() { + let (_tmp, root) = workspace(); + let io = config(true, &[], &[]); + + // The task probed a path that didn't exist. + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("config.local").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::write(root.join("config.local"), "now exists").unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Added, path }) if path == rel("config.local") + )); + } + + #[test] + fn discovered_file_replaced_by_dir_is_reported() { + let (_tmp, root) = workspace(); + fs::write(root.join("thing"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("thing").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::remove_file(root.join("thing")).unwrap(); + fs::create_dir(root.join("thing")).unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!( + matches!(change, Some(InputChange { kind: InputChangeKind::Added, path }) if path == rel("thing")) + ); + } + + #[test] + fn dir_opened_without_listing_ignores_new_entries() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("dir")).unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ, root.join("dir").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::write(root.join("dir/new.txt"), "x").unwrap(); + assert!(change_since(&root, &io, &io, &previous).is_none()); + } + + #[test] + fn listed_dir_reports_added_and_removed_entries() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("dir")).unwrap(); + fs::write(root.join("dir/old.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ_DIR, root.join("dir").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + fs::write(root.join("dir/new.txt"), "x").unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Added, path }) if path == rel("dir/new.txt") + )); + + fs::remove_file(root.join("dir/new.txt")).unwrap(); + fs::remove_file(root.join("dir/old.txt")).unwrap(); + let change = change_since(&root, &io, &io, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Removed, path }) if path == rel("dir/old.txt") + )); + } + + #[test] + fn listed_dir_does_not_see_ds_store_or_dist() { + let (_tmp, root) = workspace(); + fs::create_dir(root.join("dir")).unwrap(); + fs::write(root.join("dir/keep.txt"), "x").unwrap(); + let io = config(true, &[], &[]); + + let mut trace = Trace::default(); + trace.push(AccessMode::READ_DIR, root.join("dir").as_path()); + let (previous, _outputs) = conclude(&root, &io, &io, &trace); + + // Neither `.DS_Store` nor a `dist` entry (any casing) is visible to + // directory fingerprints. + fs::write(root.join("dir/.DS_Store"), "junk").unwrap(); + fs::create_dir(root.join("dir/DIST")).unwrap(); + assert!(change_since(&root, &io, &io, &previous).is_none()); + } + + // ── listed inputs, snapshotted and diffed ─────────────────────────────── + + /// A listed-inputs-only task: no auto tracking on either side. + fn listed_only() -> (ResolvedGlobConfig, ResolvedGlobConfig) { + (config(false, &["src/**"], &[]), config(false, &[], &[])) + } + + fn listed_workspace() -> (TempDir, AbsolutePathBuf) { + let (tmp, root) = workspace(); + fs::create_dir(root.join("src")).unwrap(); + fs::write(root.join("src/a.txt"), "a").unwrap(); + fs::write(root.join("src/b.txt"), "b").unwrap(); + (tmp, root) + } + + fn conclude_untraced( + root: &AbsolutePath, + input: &ResolvedGlobConfig, + output: &ResolvedGlobConfig, + ) -> InputFingerprints { + let task_fs = first_run(root, input, output); + let conclusion = task_fs + .post_run(None::>>, &no_ignored(), &no_ignored()) + .unwrap(); + cacheable(conclusion).0 + } + + #[test] + fn unchanged_listed_inputs_report_nothing() { + let (_tmp, root) = listed_workspace(); + let (input, output) = listed_only(); + let previous = conclude_untraced(&root, &input, &output); + + assert!(change_since(&root, &input, &output, &previous).is_none()); + } + + #[test] + fn modified_listed_input_is_reported() { + let (_tmp, root) = listed_workspace(); + let (input, output) = listed_only(); + let previous = conclude_untraced(&root, &input, &output); + + fs::write(root.join("src/b.txt"), "changed").unwrap(); + let change = change_since(&root, &input, &output, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::ContentModified, path }) if path == rel("src/b.txt") + )); + } + + #[test] + fn removed_listed_input_is_reported() { + let (_tmp, root) = listed_workspace(); + let (input, output) = listed_only(); + let previous = conclude_untraced(&root, &input, &output); + + fs::remove_file(root.join("src/a.txt")).unwrap(); + let change = change_since(&root, &input, &output, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Removed, path }) if path == rel("src/a.txt") + )); + } + + #[test] + fn added_listed_input_is_reported() { + let (_tmp, root) = listed_workspace(); + let (input, output) = listed_only(); + let previous = conclude_untraced(&root, &input, &output); + + fs::write(root.join("src/c.txt"), "c").unwrap(); + let change = change_since(&root, &input, &output, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Added, path }) if path == rel("src/c.txt") + )); + } + + #[test] + fn first_listed_change_in_path_order_wins() { + let (_tmp, root) = listed_workspace(); + let (input, output) = listed_only(); + let previous = conclude_untraced(&root, &input, &output); + + // Both changed; `src/a.txt` sorts first, so its removal is the answer. + fs::remove_file(root.join("src/a.txt")).unwrap(); + fs::write(root.join("src/b.txt"), "changed").unwrap(); + let change = change_since(&root, &input, &output, &previous); + assert!(matches!( + change, + Some(InputChange { kind: InputChangeKind::Removed, path }) if path == rel("src/a.txt") + )); + } + + #[test] + fn normalize_ignored_paths_cleans_relative_components() { + let workspace_root = + AbsolutePath::new(if cfg!(windows) { r"C:\repo" } else { "/repo" }).unwrap(); + let ignored = + workspace_root.join(if cfg!(windows) { r"pkg\..\cache" } else { "pkg/../cache" }); + let mut ignored_paths = FxHashSet::default(); + ignored_paths.insert(Arc::::from(ignored)); + + let normalized = normalize_ignored_paths(&ignored_paths, workspace_root); + + let expected = RelativePathBuf::new("cache").unwrap(); + assert!(normalized.contains(&expected)); + } +} diff --git a/crates/vt/src/session/execute/tracked_accesses.rs b/crates/vt_fs_fingerprint/src/tracked_accesses.rs similarity index 91% rename from crates/vt/src/session/execute/tracked_accesses.rs rename to crates/vt_fs_fingerprint/src/tracked_accesses.rs index 08de0262b..a3d2f0b8a 100644 --- a/crates/vt/src/session/execute/tracked_accesses.rs +++ b/crates/vt_fs_fingerprint/src/tracked_accesses.rs @@ -3,16 +3,14 @@ //! User-configured negative globs are NOT applied here. They are applied later, //! separately for reads (input config) and writes (output config), since those //! two configs are independent. -#![cfg(fspy)] use std::collections::hash_map::Entry; -use fspy::{AccessMode, PathAccessIterable}; +use fspy_shared::ipc::{AccessMode, PathAccess}; use rustc_hash::FxHashSet; use vt_path::{AbsolutePath, RelativePathBuf}; -use super::fingerprint::PathRead; -use crate::collections::HashMap; +use crate::{collections::HashMap, fingerprint::PathRead}; /// Tracked file accesses from fspy, normalized to workspace-relative paths. #[derive(Default, Debug)] @@ -25,12 +23,15 @@ pub struct TrackedPathAccesses { } impl TrackedPathAccesses { - /// Build from fspy's raw iterable by stripping the workspace prefix and + /// Build from raw accesses by stripping the workspace prefix and /// normalizing `..` components. `.git/*` paths are skipped. User-configured /// negatives are applied by the caller (see module docs). - pub fn from_raw(raw: &PathAccessIterable, workspace_root: &AbsolutePath) -> Self { + pub fn from_raw<'a>( + raw: impl IntoIterator>, + workspace_root: &AbsolutePath, + ) -> Self { let mut accesses = Self::default(); - for access in raw.iter() { + for access in raw { // Strip workspace root and clean `..` components in one pass. // fspy may report paths like `packages/sub-pkg/../shared/dist/output.js`. let relative_path = access.path.strip_path_prefix(workspace_root, |strip_result| {