Skip to main content

qk_cache/
lib.rs

1//! Local content-addressed cache shared by linked Git worktrees.
2
3mod evict;
4pub mod expectations;
5mod glob;
6mod hash;
7mod ignored;
8mod inputs;
9mod paths;
10pub mod profile;
11mod record;
12pub mod recorded;
13mod remote;
14mod store;
15pub mod ultracache;
16pub mod warm;
17
18use std::collections::{BTreeMap, BTreeSet};
19use std::path::PathBuf;
20use std::sync::OnceLock;
21use std::sync::atomic::{AtomicBool, Ordering};
22
23use anyhow::Result;
24use qk_config::Workspace;
25use qk_executor::{Capture, Outcome, PreparedTask, execute, execute_captured};
26use qk_taskgraph::{Task, TaskGraph};
27use serde_json::Value;
28
29pub use evict::{Pruned, max_size, parse_size, prune, prune_automatic};
30pub use glob::Pattern;
31pub use hash::{
32    DependencyGlob, Selects, SourceIgnore, dependency_globs, keyed_definition, source_files,
33    task_configuration,
34};
35pub use inputs::{
36    InputChange, InputChanges, JsonFields, ROOT_TSCONFIG_FILES, Resolution, Resolved, TsConfig,
37    manifest_for_dependents, resolve_each, resolve_tasks, selectively_hash_tsconfig,
38    without_resolution,
39};
40/// Expands `{projectRoot}` and `{workspaceRoot}` in a project's path or glob.
41pub use paths::expand as expand_project_path;
42pub use paths::git as workspace_git;
43pub use paths::{
44    Outputs, cache_directory, cache_location, clear_worktree_state, resolved_outputs,
45    worktree_state,
46};
47
48/// The nx.json settings of the workspace's remote store: its `s3`, and its
49/// `storeServer` or `qk:storeServer`. Nx never reads `qk:storeServer`, so a
50/// workspace can name a store server for qk while Nx's own cache plugin keeps
51/// `s3`.
52fn remote_settings(workspace: &Workspace) -> (Option<&Value>, Option<&Value>) {
53    let extra = &workspace.config.extra;
54    match extra.get("qk:storeServer") {
55        Some(server) => (None, Some(server)),
56        None => (extra.get("s3"), extra.get("storeServer")),
57    }
58}
59
60/// The cache for one run; its workspace snapshot is taken on first use.
61pub struct Cache {
62    pub root: PathBuf,
63    snapshot: OnceLock<std::result::Result<hash::Snapshot, String>>,
64    remote: Option<std::sync::Arc<remote::Remote>>,
65    /// Whether results and warm state may be restored or published.
66    reuse: bool,
67    /// The tasks whose outputs this run left exactly as they were.
68    kept: std::sync::Mutex<std::collections::BTreeSet<String>>,
69    /// Warm state being saved in the background.
70    saves: std::sync::Mutex<Vec<std::thread::JoinHandle<()>>>,
71    cleanup: std::sync::Mutex<Option<store::Cleanup>>,
72    /// Every target's declared outputs, which recordings' listings leave out.
73    workspace_outputs: OnceLock<recorded::WorkspaceOutputs>,
74}
75
76/// Where a task's result came from.
77#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
78#[serde(rename_all = "kebab-case")]
79pub enum CacheStatus {
80    LocalHit,
81    RemoteHit,
82    Miss,
83    /// Ran without consulting the cache: not cacheable, bypassed or disabled.
84    Uncached,
85}
86
87/// One actual keyed execution, independent of reusable cache entries.
88pub struct Execution {
89    pub log: qk_executor::RecordedLog,
90    pub inputs_unchanged: bool,
91}
92
93pub struct TaskResult {
94    pub outcome: Outcome,
95    pub cache: CacheStatus,
96    /// The task's key and what it was computed from, when it has one.
97    pub key: Option<(String, Value)>,
98    /// What warm state did, for targets that keep it.
99    pub warm: Option<warm::WarmReport>,
100    pub execution: Option<Execution>,
101    /// How a task keyed by its recordings was keyed: see [`recorded`].
102    pub recorded: Option<Recorded>,
103    /// For a task with `qk:ultracache` mode `"record"`, what its recordings
104    /// would have done.
105    pub shadow: Option<Shadow>,
106}
107
108/// What the recordings of a task keyed by its declared inputs would have done.
109#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
110#[serde(rename_all = "camelCase", tag = "kind")]
111pub enum Shadow {
112    /// One of its path sets' keys had a result, here or in the remote store.
113    WouldHit {
114        key: String,
115        /// For a task that ran all the same, whether it gave that result.
116        #[serde(default, skip_serializing_if = "Option::is_none")]
117        check: Option<Check>,
118    },
119    /// None of its path sets' keys had a result.
120    WouldMiss {
121        /// For a task its declared inputs took from the cache, the paths of
122        /// its newest path set that are not as they were when it was
123        /// recorded, as `<path> (<how>)`, `<how>` ending in `, declared`
124        /// where its declared inputs cover the path. A change they do not
125        /// cover either matters, so they missed it, or does not, and its
126        /// recording keys what it need not. At most [`UNCOVERED`].
127        #[serde(default, skip_serializing_if = "Vec::is_empty")]
128        changed: Vec<String>,
129        /// Where the newest path set keeps no record of how its paths were,
130        /// what it holds that the declared inputs leave out, as
131        /// `<path> (<how>)`: one of them changed. At most [`UNCOVERED`].
132        #[serde(default, skip_serializing_if = "Vec::is_empty")]
133        uncovered: Vec<String>,
134    },
135    /// It has no path set yet.
136    Unrecorded,
137}
138
139/// How many uncovered paths a [`Shadow::WouldMiss`] names.
140pub const UNCOVERED: usize = 5;
141
142/// Whether a task that ran where a recording would have taken a result from
143/// the cache gave that result: it succeeded, with the same output files.
144/// Its terminal output is not compared, since timings and the like differ
145/// from run to run.
146#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
147#[serde(rename_all = "camelCase")]
148pub struct Check {
149    pub matched: bool,
150    /// How the run differed, as `failed (exit <code>)` or `<path> (<how>)`,
151    /// at most [`UNCOVERED`].
152    #[serde(default, skip_serializing_if = "Vec::is_empty")]
153    pub differences: Vec<String>,
154}
155
156/// A lookup by a task's recordings in mode `"record"`, started before its
157/// lookup by its declared key so that the two wait on the remote store
158/// together rather than one after the other.
159enum Shadowing {
160    /// Answered without the store: a recording's result is held here, or
161    /// there is no store to ask.
162    Settled(Shadow),
163    /// Waiting for the path sets the store holds, with the keys of the path
164    /// sets held here, none of which has a result here.
165    Fetching {
166        keys: Vec<String>,
167        fetch: std::thread::JoinHandle<Result<Option<Vec<u8>>>>,
168    },
169}
170
171/// The key of one of a task's path sets, or none, said, when the path set
172/// cannot key it.
173fn recording_key(
174    weak: &str,
175    set: &recorded::PathSet,
176    snapshot: &hash::Snapshot,
177    workspace: &Workspace,
178    unlisted: &recorded::Unlisted,
179    task: &Task,
180) -> Option<String> {
181    recorded::strong_key(weak, set, snapshot, &workspace.root, unlisted)
182        .inspect_err(|error| {
183            qk_executor::status!("qk: {}: ignoring a path set ({error:#})", task.id);
184        })
185        .ok()
186}
187
188/// What a lookup by a task's recordings found.
189enum Found {
190    /// A result under this key in the local cache.
191    Local(String),
192    /// A result under this key in the remote store.
193    Remote(String),
194    Nothing,
195}
196
197/// How a task with `qk:ultracache` was keyed.
198#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
199#[serde(rename_all = "camelCase", tag = "kind")]
200pub enum Recorded {
201    /// One of its path sets' keys had a result, restored under that key.
202    Hit { key: String },
203    /// This run recorded it, and stored its result under its new key.
204    Saved { key: String },
205    /// None of its path sets' keys had a result, and this run did not record.
206    Missed,
207    /// It has no path set yet, and this run did not record one: it was keyed
208    /// by its declared inputs.
209    Unrecorded,
210    /// None of its path sets' keys had a result, and its declared key, which
211    /// covers what each of them read, did: restored under that key.
212    Declared { key: String },
213    /// Recording it failed, with why.
214    Failed { reason: String },
215}
216
217impl TaskResult {
218    pub fn uncached(outcome: Outcome) -> Self {
219        Self {
220            outcome,
221            cache: CacheStatus::Uncached,
222            key: None,
223            warm: None,
224            execution: None,
225            recorded: None,
226            shadow: None,
227        }
228    }
229}
230
231impl Cache {
232    pub fn new(root: PathBuf) -> Self {
233        Self {
234            root,
235            snapshot: OnceLock::new(),
236            remote: None,
237            reuse: true,
238            kept: Default::default(),
239            saves: Default::default(),
240            cleanup: Default::default(),
241            workspace_outputs: OnceLock::new(),
242        }
243    }
244
245    /// Records verified executions without configuring or using result stores.
246    pub fn for_observation(workspace: &Workspace) -> Self {
247        Self {
248            reuse: false,
249            ..Self::new(paths::cache_location(workspace))
250        }
251    }
252
253    /// The workspace's cache, with the remote store its nx.json `s3`,
254    /// `storeServer` or `qk:storeServer` key configures, if any. Credentials
255    /// and modes are read from `environment`, which includes the workspace's
256    /// dotenv files. A remote store that cannot be used is reported and left
257    /// out; the local cache still works.
258    pub fn for_workspace(
259        workspace: &Workspace,
260        environment: &BTreeMap<std::ffi::OsString, std::ffi::OsString>,
261    ) -> Self {
262        let root = paths::cache_location(workspace);
263        let (s3, server) = remote_settings(workspace);
264        let remote = match remote::configure(s3, server, environment) {
265            Ok(remote) => remote.map(std::sync::Arc::new),
266            Err(error) => {
267                qk_executor::status!("qk: remote cache disabled: {error:#}");
268                None
269            }
270        };
271        if let Some(remote) = &remote {
272            remote.read_index(&root);
273        }
274        Self {
275            remote,
276            ..Self::new(root)
277        }
278    }
279
280    /// Saves what later runs reuse and waits for background saves, uploads
281    /// and retired-output cleanup, reporting failures.
282    pub fn finish(&self, workspace: &Workspace) {
283        if !self.reuse {
284            return;
285        }
286        if let Some(Ok(snapshot)) = self.snapshot.get() {
287            snapshot.save_digests(&workspace.root);
288        }
289        // Before the uploads, which a save may add to.
290        for save in std::mem::take(&mut *self.saves.lock().unwrap()) {
291            let _ = save.join();
292        }
293        if let Some(remote) = &self.remote {
294            let failures = remote.finish();
295            for failure in &failures {
296                qk_executor::status!("qk: remote cache upload failed for {failure}");
297            }
298        }
299        let _profile = profile::span("cache", "restore_cleanup_wait");
300        self.cleanup.lock().unwrap().take();
301    }
302
303    fn snapshot(
304        &self,
305        workspace: &Workspace,
306        graph: &TaskGraph,
307    ) -> std::result::Result<&hash::Snapshot, String> {
308        self.snapshot
309            .get_or_init(|| {
310                hash::Snapshot::new(workspace, graph, &self.root)
311                    .map_err(|error| format!("cannot snapshot the workspace: {error:#}"))
312            })
313            .as_ref()
314            .map_err(Clone::clone)
315    }
316
317    pub fn run(
318        &self,
319        workspace: &Workspace,
320        graph: &TaskGraph,
321        task: &Task,
322        prepared: &PreparedTask,
323        cancelled: &AtomicBool,
324    ) -> Result<TaskResult> {
325        let result = self.run_task(workspace, graph, task, prepared, cancelled);
326        if let Some(Ok(snapshot)) = self.snapshot.get() {
327            snapshot.outputs_written(workspace, task);
328        }
329        result
330    }
331
332    fn run_task(
333        &self,
334        workspace: &Workspace,
335        graph: &TaskGraph,
336        task: &Task,
337        prepared: &PreparedTask,
338        cancelled: &AtomicBool,
339    ) -> Result<TaskResult> {
340        let cacheable = task.definition.cache == Some(true);
341        // Held until the task is done, whether it ran or came from the cache.
342        let _kept = self
343            .reuse
344            .then(|| warm::keep_across(workspace, graph, task));
345        let entries = if self.reuse {
346            warm::config(workspace, task).unwrap_or_else(|error| {
347                qk_executor::status!("qk: {}: warm state ignored ({error:#})", task.id);
348                Vec::new()
349            })
350        } else {
351            Vec::new()
352        };
353        // Names an entry in messages: the group it shares, if any.
354        let entry_label = |warm: &warm::Warm| match warm.group() {
355            Some(group) => format!(" (group {group})"),
356            None => String::new(),
357        };
358        // The warm variables point tools at their state; they are not inputs,
359        // so the key is computed from the task without them.
360        let mut running = prepared.clone();
361        for warm in &entries {
362            for (name, value) in &warm.env {
363                running.execution.insert(name.into(), value.into());
364            }
365        }
366        // What each entry restored, and which groups were on disk already.
367        let before = |running: &PreparedTask| -> (Vec<warm::Restored>, Vec<String>) {
368            let (mut restored, mut present) = (Vec::new(), Vec::new());
369            for warm in &entries {
370                match self
371                    .initialize()
372                    .and_then(|()| self.restore_warm(workspace, task, warm, running))
373                {
374                    Ok((entry, groups)) => {
375                        if !entry.groups.is_empty() {
376                            restored.push(entry);
377                        }
378                        present.extend(groups);
379                    }
380                    Err(error) => qk_executor::status!(
381                        "qk: {}: warm state{} not restored ({error:#})",
382                        task.id,
383                        entry_label(warm)
384                    ),
385                }
386            }
387            (restored, present)
388        };
389        let after = |outcome: Outcome,
390                     (restored, present): (Vec<warm::Restored>, Vec<String>),
391                     running: &PreparedTask|
392         -> Option<warm::WarmReport> {
393            if entries.is_empty() {
394                return None;
395            }
396            let mut save_ms = None;
397            let mut background = false;
398            for warm in entries.iter().filter(|_| outcome == Outcome::Success) {
399                let started = std::time::Instant::now();
400                let saved = if warm.background {
401                    self.save_warm_in_background(workspace, task, warm, running)
402                } else {
403                    self.save_warm(workspace, task, warm, running)
404                };
405                match saved {
406                    Ok(()) if warm.background => background = true,
407                    Ok(()) => {
408                        *save_ms.get_or_insert(0) += started.elapsed().as_millis() as u64;
409                    }
410                    Err(error) => qk_executor::status!(
411                        "qk: {}: warm state{} not saved ({error:#})",
412                        task.id,
413                        entry_label(warm)
414                    ),
415                }
416            }
417            Some(warm::WarmReport {
418                restored,
419                present,
420                save_ms,
421                background,
422            })
423        };
424        // Reports why this task runs uncached.
425        let bypass = |reason: String| {
426            if cacheable {
427                qk_executor::status!("qk: {}: cache bypassed ({reason})", task.id);
428            }
429            let _profile = profile::span(&task.id, "execute");
430            execute(prepared, cancelled).map(TaskResult::uncached)
431        };
432        let snapshot = match self.snapshot(workspace, graph) {
433            Ok(snapshot) => snapshot,
434            Err(reason) => return bypass(reason),
435        };
436        let dependency_profile = profile::span(&task.id, "dependency_inputs");
437        let dependencies = match hash::dependency_keys(snapshot, workspace, graph, task, cancelled)
438        {
439            Ok(dependencies) => dependencies,
440            Err(error) => return bypass(format!("{error:#}")),
441        };
442        drop(dependency_profile);
443        let input_profile = profile::span(&task.id, "inputs");
444        let (key, inputs) = match hash::inputs(
445            snapshot,
446            workspace,
447            task,
448            prepared,
449            &dependencies,
450            cancelled,
451        )
452        .and_then(|inputs| Ok((hash::key(&inputs)?, inputs)))
453        {
454            Ok(key) => key,
455            Err(error) => return bypass(format!("{error:#}")),
456        };
457        drop(input_profile);
458        let keyed = Some((key.clone(), inputs));
459        // As in Nx, the task sees its hash.
460        running
461            .execution
462            .insert("NX_TASK_HASH".into(), key.clone().into());
463        let running = &running;
464        let outputs = match paths::Outputs::new(workspace, task) {
465            Ok(outputs) => outputs,
466            Err(error) => return bypass(format!("{error:#}")),
467        };
468        if !cacheable {
469            let restored = before(running);
470            let executing = profile::span(&task.id, "execute");
471            let outcome = execute(running, cancelled)?;
472            drop(executing);
473            let warm_report = after(outcome, restored, running);
474            return Ok(TaskResult {
475                outcome,
476                cache: CacheStatus::Uncached,
477                key: keyed,
478                warm: warm_report,
479                execution: None,
480                recorded: None,
481                shadow: None,
482            });
483        }
484        if !self.reuse {
485            let capture = Capture::new(None, prepared.display.clone()).retain_log();
486            let executing = profile::span(&task.id, "execute");
487            let outcome = execute_captured(running, cancelled, Some(&capture))?;
488            drop(executing);
489            let inputs_unchanged =
490                self.inputs_match(snapshot, workspace, graph, task, prepared, &key, cancelled);
491            return Ok(TaskResult {
492                outcome,
493                cache: CacheStatus::Uncached,
494                key: keyed,
495                warm: None,
496                execution: capture.take_log().map(|log| Execution {
497                    log,
498                    inputs_unchanged,
499                }),
500                recorded: None,
501                shadow: None,
502            });
503        }
504        if let Err(error) = self.initialize() {
505            return bypass(format!("cache unavailable: {error}"));
506        }
507        let (_lock, waited) = match self.lock(&key, cancelled) {
508            Ok(Some(lock)) => lock,
509            Ok(None) => {
510                return Ok(TaskResult::uncached(Outcome::Cancelled));
511            }
512            Err(error) => return bypass(format!("cache lock unavailable: {error}")),
513        };
514        // Inputs may have changed while another worktree held this key's lock.
515        if waited
516            && hash::recheck_dependency_keys(snapshot, workspace, graph, task, cancelled)
517                .and_then(|dependencies| {
518                    hash::fingerprint(
519                        snapshot,
520                        workspace,
521                        task,
522                        prepared,
523                        &dependencies,
524                        cancelled,
525                    )
526                })
527                .ok()
528                .as_ref()
529                != Some(&key)
530        {
531            return bypass("inputs changed while waiting for another run".into());
532        }
533        // Keyed by its recordings, a task's path sets stand in for its
534        // declared file inputs: see `recorded`.
535        let settings = ultracache::Settings::of(task).unwrap_or_else(|error| {
536            qk_executor::status!("qk: {}: qk:ultracache ignored ({error:#})", task.id);
537            None
538        });
539        let warm_outputs = paths::Outputs::from_paths(&warm::kept_paths(&entries)).ok();
540        let own: Vec<&paths::Outputs> = std::iter::once(&outputs)
541            .chain(warm_outputs.as_ref())
542            .collect();
543        // What its recordings' listings leave out, by the workspace's
544        // configuration rather than what this run schedules, so that the
545        // same recording keys the same in any run.
546        let others = settings.as_ref().map(|_| recorded::Others {
547            outputs: self
548                .workspace_outputs
549                .get_or_init(|| workspace_outputs(workspace)),
550            upstream: upstream(graph, task),
551        });
552        let ignore: Vec<Pattern> = settings
553            .as_ref()
554            .map(|settings| {
555                settings
556                    .ignore_reads
557                    .iter()
558                    .filter_map(|glob| Pattern::new(glob, false).ok())
559                    .collect()
560            })
561            .unwrap_or_default();
562        let unlisted = recorded::Unlisted {
563            own: own.clone(),
564            others,
565            ignore: &ignore,
566        };
567        let inputs = &keyed.as_ref().expect("a cacheable task is keyed").1;
568        let mut recordings = settings.and_then(|settings| {
569            let weak = hash::weak_key(
570                snapshot,
571                workspace,
572                graph,
573                task,
574                prepared,
575                inputs,
576                &settings.ignore_reads,
577                cancelled,
578            );
579            match weak {
580                Ok(weak) => Some((recorded::load(&self.root, &weak), weak, settings)),
581                Err(error) => {
582                    qk_executor::status!(
583                        "qk: {}: not keyed by its recordings ({error:#})",
584                        task.id
585                    );
586                    None
587                }
588            }
589        });
590        let mut shadowing = None;
591        if let Some((sets, weak, settings)) = &mut recordings {
592            if settings.mode == ultracache::Mode::Record {
593                let _lookup = profile::span(&task.id, "recorded_lookup");
594                shadowing =
595                    Some(self.start_shadow(workspace, task, snapshot, &unlisted, sets, weak));
596            } else {
597                let recorded_profile = profile::span(&task.id, "recorded_restore");
598                let found = self.by_recordings(
599                    workspace, task, prepared, snapshot, &outputs, &unlisted, sets, weak, &key,
600                    cancelled,
601                );
602                drop(recorded_profile);
603                if let found @ (Found::Local(_) | Found::Remote(_)) = found {
604                    let (strong, remote) = match found {
605                        Found::Local(strong) => {
606                            if let Some(remote) = &self.remote {
607                                remote.held(&strong);
608                            }
609                            (strong, None)
610                        }
611                        Found::Remote(strong) => (strong, Some(())),
612                        Found::Nothing => unreachable!(),
613                    };
614                    recorded::touch(&self.root, weak);
615                    qk_executor::report::event(qk_executor::report::Event::Cache {
616                        id: task.id.clone(),
617                        cached: if remote.is_some() {
618                            qk_executor::report::Cached::RemoteHit
619                        } else {
620                            qk_executor::report::Cached::Hit
621                        },
622                    });
623                    return Ok(TaskResult {
624                        outcome: Outcome::Success,
625                        cache: if remote.is_some() {
626                            CacheStatus::RemoteHit
627                        } else {
628                            CacheStatus::LocalHit
629                        },
630                        key: keyed,
631                        warm: None,
632                        execution: None,
633                        recorded: Some(Recorded::Hit { key: strong }),
634                        shadow: None,
635                    });
636                }
637            }
638        }
639        // What its recordings would have done, once the store has said.
640        let mut settle = |recordings: &mut Option<(Vec<recorded::PathSet>, String, _)>| {
641            let shadowing = shadowing.take()?;
642            let (sets, weak, _) = recordings.as_mut()?;
643            let _wait = profile::span(&task.id, "recorded_wait");
644            Some(self.settle_shadow(shadowing, workspace, task, snapshot, &unlisted, sets, weak))
645        };
646        // A task with path sets is keyed by them alone, unless its declared
647        // inputs, as they are now, cover what each of them read: otherwise
648        // its declared inputs are what a recording found wanting. In mode
649        // "record" it is keyed by its declared inputs all the same.
650        let declared = recordings.as_ref().is_none_or(|(sets, _, settings)| {
651            settings.mode == ultracache::Mode::Record
652                || sets.is_empty()
653                || hash::coverage(snapshot, workspace, graph, task, prepared, cancelled).is_ok_and(
654                    |coverage| {
655                        let outputs = self
656                            .workspace_outputs
657                            .get_or_init(|| workspace_outputs(workspace));
658                        let output = |path: &str| outputs.matches(path);
659                        sets.iter()
660                            .all(|set| coverage.covers(snapshot, &workspace.root, set, &output))
661                    },
662                )
663        });
664        // How its recordings keyed it, unless it runs and records.
665        let unrecorded = recordings
666            .as_ref()
667            .filter(|(_, _, settings)| settings.mode == ultracache::Mode::On)
668            .map(|(sets, _, _)| {
669                if sets.is_empty() {
670                    Recorded::Unrecorded
671                } else {
672                    Recorded::Declared { key: key.clone() }
673                }
674            });
675        // An entry fetched ahead of this task is local, but it came from the
676        // remote store, and is reported as such.
677        let prefetched = declared
678            && self.remote.as_ref().is_some_and(|remote| {
679                remote.resolved(&key);
680                remote.prefetched(&key)
681            });
682        let local_profile = profile::span(&task.id, "local_restore");
683        let restored = if declared {
684            self.restore(
685                &workspace.root,
686                &task.id,
687                &task.dependencies,
688                &key,
689                &outputs,
690                &prepared.display,
691                if prefetched {
692                    qk_executor::Shown::RemoteCache
693                } else {
694                    qk_executor::Shown::LocalCache
695                },
696            )
697        } else {
698            Ok(false)
699        };
700        match restored {
701            Ok(true) => {
702                let claimed = match &self.remote {
703                    Some(remote) if prefetched && remote.claim(&key) => true,
704                    Some(remote) => {
705                        remote.held(&key);
706                        false
707                    }
708                    None => false,
709                };
710                qk_executor::report::event(qk_executor::report::Event::Cache {
711                    id: task.id.clone(),
712                    cached: if claimed {
713                        qk_executor::report::Cached::RemoteHit
714                    } else {
715                        qk_executor::report::Cached::Hit
716                    },
717                });
718                let shadow = settle(&mut recordings);
719                let shadow = self.with_changes(
720                    shadow,
721                    recordings.as_ref(),
722                    (snapshot, workspace, graph, task, prepared, &unlisted),
723                    cancelled,
724                );
725                return Ok(TaskResult {
726                    outcome: Outcome::Success,
727                    cache: if claimed {
728                        CacheStatus::RemoteHit
729                    } else {
730                        CacheStatus::LocalHit
731                    },
732                    key: keyed,
733                    warm: None,
734                    execution: None,
735                    recorded: unrecorded,
736                    shadow,
737                });
738            }
739            Ok(false) => {}
740            Err(error) => {
741                qk_executor::status!("qk: {}: ignoring unusable cache entry ({error})", task.id)
742            }
743        }
744        drop(local_profile);
745        if let Some(remote) = self.remote.as_ref().filter(|_| declared) {
746            let fetch_profile = profile::span(&task.id, "remote_fetch");
747            let fetched = remote.fetch(&self.root, &key, &task.id, cancelled);
748            drop(fetch_profile);
749            let fetched = fetched.and_then(|found| {
750                if !found {
751                    return Ok(false);
752                }
753                self.restore(
754                    &workspace.root,
755                    &task.id,
756                    &task.dependencies,
757                    &key,
758                    &outputs,
759                    &prepared.display,
760                    qk_executor::Shown::RemoteCache,
761                )
762            });
763            match fetched {
764                Ok(true) => {
765                    qk_executor::report::event(qk_executor::report::Event::Cache {
766                        id: task.id.clone(),
767                        cached: qk_executor::report::Cached::RemoteHit,
768                    });
769                    let shadow = settle(&mut recordings);
770                    let shadow = self.with_changes(
771                        shadow,
772                        recordings.as_ref(),
773                        (snapshot, workspace, graph, task, prepared, &unlisted),
774                        cancelled,
775                    );
776                    return Ok(TaskResult {
777                        outcome: Outcome::Success,
778                        cache: CacheStatus::RemoteHit,
779                        key: keyed,
780                        warm: None,
781                        execution: None,
782                        recorded: unrecorded,
783                        shadow,
784                    });
785                }
786                Ok(false) => {}
787                Err(error) => {
788                    qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", task.id)
789                }
790            }
791        }
792        // Settled before it runs: recording it merges the path sets fetched.
793        let shadow = settle(&mut recordings);
794        qk_executor::report::event(qk_executor::report::Event::Cache {
795            id: task.id.clone(),
796            cached: qk_executor::report::Cached::Miss,
797        });
798        let log = match tempfile::NamedTempFile::new_in(self.root.join("tmp")) {
799            Ok(log) => log,
800            Err(error) => return bypass(format!("cache log unavailable: {error}")),
801        };
802        let capture =
803            Capture::new(Some(log.as_file().try_clone()?), prepared.display.clone()).retain_log();
804        let restored = before(running);
805        let listed = match (&recordings, &prepared.recording) {
806            (Some(_), Some(_)) => {
807                let project_root = workspace
808                    .projects
809                    .get(&task.project)
810                    .map_or("", |project| project.root.as_str());
811                recorded::Before::take(
812                    &workspace.root,
813                    unlisted.directories(project_root),
814                    &unlisted,
815                )
816            }
817            _ => recorded::Before::default(),
818        };
819        let started = recorded::now(&self.root.join("tmp"));
820        let executing = profile::span(&task.id, "execute");
821        let outcome = execute_captured(running, cancelled, Some(&capture))?;
822        drop(executing);
823        let warm_report = after(outcome, restored, running);
824        let inputs_unchanged =
825            self.inputs_match(snapshot, workspace, graph, task, prepared, &key, cancelled);
826        let execution = capture.take_log().map(|log| Execution {
827            log,
828            inputs_unchanged,
829        });
830        let mut recorded = unrecorded.map(|unrecorded| match &recordings {
831            Some((sets, _, _)) if !sets.is_empty() => Recorded::Missed,
832            _ => unrecorded,
833        });
834        let path_set = match (&recordings, &prepared.recording) {
835            (Some((_, _, settings)), Some(recording)) => Some(
836                self.path_set(workspace, recording, prepared, settings, &own)
837                    .and_then(|set| {
838                        // Keyed by nothing, a recording would match any workspace.
839                        if set.is_empty() {
840                            return Err("its trace named no path in the workspace".to_owned());
841                        }
842                        match set
843                            .changed_since(&workspace.root, started, &unlisted, &listed)
844                            .map(|changed| changed.to_string())
845                        {
846                            Some(reason) => Err(reason),
847                            None => Ok(set),
848                        }
849                    }),
850            ),
851            _ => None,
852        };
853        if let Some(Err(reason)) = &path_set {
854            qk_executor::status!("qk: {}: not recorded ({reason})", task.id);
855            recorded = Some(Recorded::Failed {
856                reason: reason.clone(),
857            });
858        }
859        let mut published = false;
860        let kept = outcome == Outcome::Success && inputs_unchanged && capture.healthy();
861        if kept {
862            let publishing = profile::span(&task.id, "publish");
863            let publish = self.publish(&workspace.root, &key, &outputs, log.path());
864            drop(publishing);
865            match publish {
866                Ok(()) => {
867                    published = true;
868                    store::record_outputs(&workspace.root, &task.id, &key, &outputs);
869                    if let Some(remote) = &self.remote {
870                        remote.upload(&self.root, &key);
871                    }
872                }
873                Err(error) => {
874                    qk_executor::status!("qk: {}: could not save cache entry ({error})", task.id);
875                }
876            }
877        }
878        // Ran where a recording would have taken a result from the cache:
879        // whether it gave that result, read before this run's own recording
880        // can replace it.
881        let check = match &shadow {
882            Some(Shadow::WouldHit { key: recorded, .. }) => match outcome {
883                Outcome::Failed(code) => Some(Check {
884                    matched: false,
885                    differences: vec![format!("failed (exit {code})")],
886                }),
887                Outcome::Success if published => {
888                    let _checking = profile::span(&task.id, "recorded_check");
889                    self.check(&task.id, &key, recorded)
890                }
891                _ => None,
892            },
893            _ => None,
894        };
895        if kept {
896            // Its result is kept under the key of what it read, too.
897            if let (Some((_, weak, _)), Some(Ok(set))) = (&recordings, path_set) {
898                let saved = recorded::with_values(weak, set, snapshot, &workspace.root, &unlisted)
899                    .and_then(|(strong, set)| {
900                        self.publish(&workspace.root, &strong, &outputs, log.path())?;
901                        recorded::save(&self.root, weak, set)?;
902                        // Shared, the sets are merged with those the store
903                        // held when this task looked it up.
904                        if let Some(remote) = &self.remote {
905                            remote.upload(&self.root, &strong);
906                            if let Some(record) = recorded::record(&self.root, weak) {
907                                remote.upload_path_sets(&self.root, weak, record);
908                            }
909                        }
910                        Ok(strong)
911                    });
912                recorded = Some(match saved {
913                    Ok(key) => Recorded::Saved { key },
914                    Err(error) => {
915                        qk_executor::status!("qk: {}: recording not saved ({error:#})", task.id);
916                        Recorded::Failed {
917                            reason: format!("{error:#}"),
918                        }
919                    }
920                });
921            }
922        }
923        let shadow = match shadow {
924            Some(Shadow::WouldHit { key: recorded, .. }) => Some(Shadow::WouldHit {
925                key: recorded,
926                check,
927            }),
928            shadow => shadow,
929        };
930        Ok(TaskResult {
931            outcome,
932            cache: CacheStatus::Miss,
933            key: keyed,
934            warm: warm_report,
935            execution,
936            recorded,
937            shadow,
938        })
939    }
940
941    /// Whether the result just published under `ran` is the one under
942    /// `recorded`, here or in the remote store: `None` when that is not to
943    /// be had.
944    fn check(&self, task: &str, ran: &str, recorded: &str) -> Option<Check> {
945        let compared = self
946            .artifacts(ran, task)
947            .and_then(|ran| Ok(ran.zip(self.artifacts(recorded, task)?)));
948        match compared {
949            Ok(Some((ran, recorded))) => {
950                let differences = store::differences(&ran, &recorded, UNCOVERED);
951                Some(Check {
952                    matched: differences.is_empty(),
953                    differences,
954                })
955            }
956            Ok(None) => None,
957            Err(error) => {
958                qk_executor::status!(
959                    "qk: {task}: not compared with its recorded result ({error:#})"
960                );
961                None
962            }
963        }
964    }
965
966    /// `shadow`, naming what changed in the newest path set since it was
967    /// recorded, or what it holds that the declared inputs leave out, when no
968    /// recording would have hit.
969    fn with_changes(
970        &self,
971        shadow: Option<Shadow>,
972        recordings: Option<&(Vec<recorded::PathSet>, String, ultracache::Settings)>,
973        (snapshot, workspace, graph, task, prepared, unlisted): (
974            &hash::Snapshot,
975            &Workspace,
976            &TaskGraph,
977            &Task,
978            &PreparedTask,
979            &recorded::Unlisted,
980        ),
981        cancelled: &AtomicBool,
982    ) -> Option<Shadow> {
983        let (Some(Shadow::WouldMiss { .. }), Some((sets, _, _))) = (&shadow, recordings) else {
984            return shadow;
985        };
986        let Some(newest) = sets.first() else {
987            return shadow;
988        };
989        let _explaining = profile::span(&task.id, "recorded_changes");
990        let Ok(coverage) = hash::coverage(snapshot, workspace, graph, task, prepared, cancelled)
991        else {
992            return shadow;
993        };
994        let outputs = self
995            .workspace_outputs
996            .get_or_init(|| workspace_outputs(workspace));
997        let output = |path: &str| outputs.matches(path);
998        let uncovered: BTreeMap<&str, &str> = coverage
999            .uncovered(snapshot, &workspace.root, newest, &output)
1000            .collect();
1001        let named = |paths: Vec<String>| {
1002            let more = paths.len().saturating_sub(UNCOVERED - 1);
1003            let mut paths = paths;
1004            if more > 1 {
1005                paths.truncate(UNCOVERED - 1);
1006                paths.push(format!("{more} more"));
1007            }
1008            paths
1009        };
1010        if newest.recorded.is_some() {
1011            let changed = newest
1012                .changed(snapshot, &workspace.root, unlisted)
1013                .unwrap_or_default()
1014                .into_iter()
1015                .map(|(path, how)| match uncovered.contains_key(path) {
1016                    true => format!("{path} ({how})"),
1017                    false => format!("{path} ({how}, declared)"),
1018                })
1019                .collect();
1020            return Some(Shadow::WouldMiss {
1021                changed: named(changed),
1022                uncovered: vec![],
1023            });
1024        }
1025        let uncovered = uncovered
1026            .into_iter()
1027            .map(|(path, how)| format!("{path} ({how})"))
1028            .collect();
1029        Some(Shadow::WouldMiss {
1030            changed: vec![],
1031            uncovered: named(uncovered),
1032        })
1033    }
1034
1035    /// Looks a task up by its recordings and restores what it finds: by each
1036    /// path set's key in the local cache, then, with none there, among the
1037    /// path sets and results the remote store holds, adding its path sets to
1038    /// `sets`.
1039    #[allow(clippy::too_many_arguments)]
1040    fn by_recordings(
1041        &self,
1042        workspace: &Workspace,
1043        task: &Task,
1044        prepared: &PreparedTask,
1045        snapshot: &hash::Snapshot,
1046        outputs: &paths::Outputs,
1047        unlisted: &recorded::Unlisted,
1048        sets: &mut Vec<recorded::PathSet>,
1049        weak: &str,
1050        declared: &str,
1051        cancelled: &AtomicBool,
1052    ) -> Found {
1053        let strong =
1054            |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1055        let restored = |strong: &str, shown| {
1056            self.restore(
1057                &workspace.root,
1058                &task.id,
1059                &task.dependencies,
1060                strong,
1061                outputs,
1062                &prepared.display,
1063                shown,
1064            )
1065            .unwrap_or_else(|error| {
1066                qk_executor::status!("qk: {}: ignoring unusable cache entry ({error})", task.id);
1067                false
1068            })
1069        };
1070        let local = |strong: &str| restored(strong, qk_executor::Shown::LocalCache);
1071        let mut keys = Vec::new();
1072        for set in sets.iter() {
1073            if let Some(strong) = strong(set) {
1074                if local(&strong) {
1075                    return Found::Local(strong);
1076                }
1077                keys.push(strong);
1078            }
1079        }
1080        let Some(remote) = &self.remote else {
1081            return Found::Nothing;
1082        };
1083        // Without path sets here, a result here under the declared key is
1084        // the task's, and asking the store for path sets would only delay
1085        // it, run after run, and have them shadow it.
1086        if sets.is_empty()
1087            && self
1088                .root
1089                .join("entries")
1090                .join(format!("{declared}.json"))
1091                .is_file()
1092        {
1093            return Found::Nothing;
1094        }
1095        // Path sets other machines recorded, new here.
1096        let fetched = remote
1097            .fetch_path_sets(&self.root, weak, &task.id)
1098            .and_then(|record| match record {
1099                Some(record) => recorded::merge(&self.root, weak, &record),
1100                None => Ok(Vec::new()),
1101            });
1102        match fetched {
1103            Ok(added) => {
1104                for set in added {
1105                    if let Some(strong) = strong(&set) {
1106                        if local(&strong) {
1107                            sets.push(set);
1108                            return Found::Local(strong);
1109                        }
1110                        keys.push(strong);
1111                    }
1112                    sets.push(set);
1113                }
1114            }
1115            Err(error) => {
1116                qk_executor::status!("qk: {}: remote path sets unavailable ({error:#})", task.id)
1117            }
1118        }
1119        for strong in keys {
1120            let found = remote
1121                .fetch(&self.root, &strong, &task.id, cancelled)
1122                .unwrap_or_else(|error| {
1123                    qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", task.id);
1124                    false
1125                })
1126                && restored(&strong, qk_executor::Shown::RemoteCache);
1127            if found {
1128                return Found::Remote(strong);
1129            }
1130        }
1131        Found::Nothing
1132    }
1133
1134    /// Starts looking a task in mode `"record"` up by its recordings, without
1135    /// restoring anything: by each path set's key in the local cache, and with
1136    /// none there, by fetching the path sets the remote store holds on a
1137    /// thread of its own, for [`Self::settle_shadow`] to finish.
1138    fn start_shadow(
1139        &self,
1140        workspace: &Workspace,
1141        task: &Task,
1142        snapshot: &hash::Snapshot,
1143        unlisted: &recorded::Unlisted,
1144        sets: &[recorded::PathSet],
1145        weak: &str,
1146    ) -> Shadowing {
1147        let mut keys = Vec::new();
1148        for set in sets {
1149            if let Some(strong) = recording_key(weak, set, snapshot, workspace, unlisted, task) {
1150                if self.holds_entry(&strong) {
1151                    return Shadowing::Settled(Shadow::WouldHit {
1152                        key: strong,
1153                        check: None,
1154                    });
1155                }
1156                keys.push(strong);
1157            }
1158        }
1159        let Some(remote) = &self.remote else {
1160            return Shadowing::Settled(if sets.is_empty() {
1161                Shadow::Unrecorded
1162            } else {
1163                Shadow::WouldMiss {
1164                    changed: vec![],
1165                    uncovered: vec![],
1166                }
1167            });
1168        };
1169        let (remote, root, weak, id) = (
1170            std::sync::Arc::clone(remote),
1171            self.root.clone(),
1172            weak.to_owned(),
1173            task.id.clone(),
1174        );
1175        Shadowing::Fetching {
1176            keys,
1177            fetch: std::thread::spawn(move || remote.fetch_path_sets(&root, &weak, &id)),
1178        }
1179    }
1180
1181    /// What a task's recordings would have done: [`Self::start_shadow`]'s
1182    /// lookup, finished with the path sets the store held, which are added
1183    /// to `sets`, and the results it holds.
1184    #[allow(clippy::too_many_arguments)]
1185    fn settle_shadow(
1186        &self,
1187        shadowing: Shadowing,
1188        workspace: &Workspace,
1189        task: &Task,
1190        snapshot: &hash::Snapshot,
1191        unlisted: &recorded::Unlisted,
1192        sets: &mut Vec<recorded::PathSet>,
1193        weak: &str,
1194    ) -> Shadow {
1195        let (mut keys, fetch) = match shadowing {
1196            Shadowing::Settled(shadow) => return shadow,
1197            Shadowing::Fetching { keys, fetch } => (keys, fetch),
1198        };
1199        let fetched = fetch
1200            .join()
1201            .unwrap_or_else(|_| Err(anyhow::anyhow!("fetching them panicked")))
1202            .and_then(|record| match record {
1203                Some(record) => recorded::merge(&self.root, weak, &record),
1204                None => Ok(Vec::new()),
1205            });
1206        match fetched {
1207            Ok(added) => {
1208                for set in added {
1209                    if let Some(strong) =
1210                        recording_key(weak, &set, snapshot, workspace, unlisted, task)
1211                    {
1212                        if self.holds_entry(&strong) {
1213                            sets.push(set);
1214                            return Shadow::WouldHit {
1215                                key: strong,
1216                                check: None,
1217                            };
1218                        }
1219                        keys.push(strong);
1220                    }
1221                    sets.push(set);
1222                }
1223            }
1224            Err(error) => {
1225                qk_executor::status!("qk: {}: remote path sets unavailable ({error:#})", task.id)
1226            }
1227        }
1228        let remote = self
1229            .remote
1230            .as_ref()
1231            .expect("only a remote store is fetched from");
1232        keys.into_iter()
1233            .find(|strong| remote.holds(strong).unwrap_or(false))
1234            .map_or(
1235                if sets.is_empty() {
1236                    Shadow::Unrecorded
1237                } else {
1238                    Shadow::WouldMiss {
1239                        changed: vec![],
1240                        uncovered: vec![],
1241                    }
1242                },
1243                |key| Shadow::WouldHit { key, check: None },
1244            )
1245    }
1246
1247    /// Whether the local cache holds a result under `key`.
1248    fn holds_entry(&self, key: &str) -> bool {
1249        self.root
1250            .join("entries")
1251            .join(format!("{key}.json"))
1252            .is_file()
1253    }
1254
1255    /// The path set of a task's recording, or why there is none.
1256    fn path_set(
1257        &self,
1258        workspace: &Workspace,
1259        recording: &qk_executor::Recording,
1260        prepared: &PreparedTask,
1261        settings: &ultracache::Settings,
1262        own: &[&paths::Outputs],
1263    ) -> std::result::Result<recorded::PathSet, String> {
1264        let failed = |error: anyhow::Error| format!("{error:#}");
1265        let canonical = workspace
1266            .root
1267            .canonicalize()
1268            .map_err(|e| failed(e.into()))?;
1269        // Only what a path set can hold is kept and counted: a build reads
1270        // far more in node_modules and outside the workspace than in it.
1271        let keep = |path: &std::path::Path| {
1272            [&workspace.root, &canonical].iter().any(|root| {
1273                path.strip_prefix(root).is_ok_and(|relative| {
1274                    !relative.components().any(|component| {
1275                        matches!(
1276                            component.as_os_str().to_str(),
1277                            Some("node_modules" | ".git")
1278                        )
1279                    })
1280                })
1281            })
1282        };
1283        let limits = qk_input_analysis::strace::Limits {
1284            records: 1_000_000,
1285            bytes: 256 * 1024 * 1024,
1286            accesses: 500_000,
1287            keep: Some(&keep),
1288        };
1289        recording.settle().map_err(failed)?;
1290        let (accesses, diagnostics) =
1291            qk_input_analysis::strace::collect_within(&recording.directory, &prepared.cwd, &limits)
1292                .map_err(failed)?;
1293        // A partial trace would key the task by part of what it read.
1294        if let Some(diagnostic) = diagnostics.first() {
1295            return Err(diagnostic.clone());
1296        }
1297        let ignore = settings
1298            .ignore_reads
1299            .iter()
1300            .map(|glob| Pattern::new(glob, false))
1301            .collect::<Result<Vec<_>>>()
1302            .map_err(failed)?;
1303        // qk's own directories, where they lie inside the workspace.
1304        let state = [self.root.clone(), worktree_state(&workspace.root)]
1305            .iter()
1306            .filter_map(|path| {
1307                [&workspace.root, &canonical]
1308                    .iter()
1309                    .find_map(|root| path.strip_prefix(root).ok())
1310                    .and_then(|relative| relative.to_str())
1311                    .map(|relative| relative.replace('\\', "/"))
1312            })
1313            .collect();
1314        let exclusions = recorded::Exclusions {
1315            roots: [&workspace.root, &canonical],
1316            outputs: own,
1317            ignore,
1318            state,
1319        };
1320        Ok(recorded::PathSet::from_accesses(
1321            &accesses,
1322            &exclusions,
1323            &canonical,
1324        ))
1325    }
1326
1327    /// Rechecks declared inputs for successful and failed executions alike.
1328    #[allow(clippy::too_many_arguments)]
1329    fn inputs_match(
1330        &self,
1331        snapshot: &hash::Snapshot,
1332        workspace: &Workspace,
1333        graph: &TaskGraph,
1334        task: &Task,
1335        prepared: &PreparedTask,
1336        before: &str,
1337        cancelled: &AtomicBool,
1338    ) -> bool {
1339        if cancelled.load(Ordering::SeqCst) {
1340            return false;
1341        }
1342        let _profile = profile::span(&task.id, "inputs_recheck");
1343        let after = hash::recheck_dependency_keys(snapshot, workspace, graph, task, cancelled)
1344            .and_then(|dependencies| {
1345                hash::fingerprint(
1346                    snapshot,
1347                    workspace,
1348                    task,
1349                    prepared,
1350                    &dependencies,
1351                    cancelled,
1352                )
1353            });
1354        if after.as_deref().ok() != Some(before) {
1355            if task.definition.cache == Some(true) {
1356                if self.reuse {
1357                    qk_executor::status!(
1358                        "qk: {}: not caching because inputs changed during execution",
1359                        task.id
1360                    );
1361                } else {
1362                    qk_executor::status!(
1363                        "qk: {}: excluding execution from flaky results because inputs changed during execution",
1364                        task.id
1365                    );
1366                }
1367            }
1368            return false;
1369        }
1370        true
1371    }
1372}
1373
1374/// A task's key, and the inputs it is computed from, or why it has none.
1375pub type Keyed = std::result::Result<(String, Value), String>;
1376
1377/// The key each task in `graph` would be cached under if it ran now, and what
1378/// it is computed from, without running a task. Runtime inputs are executed,
1379/// and the dependency outputs a task reads are taken as they are on disk.
1380pub fn keys(
1381    workspace: &Workspace,
1382    graph: &TaskGraph,
1383    prepared: &BTreeMap<String, PreparedTask>,
1384    cancelled: &AtomicBool,
1385) -> Result<BTreeMap<String, Keyed>> {
1386    let snapshot = hash::Snapshot::new(workspace, graph, &paths::cache_location(workspace))?;
1387    Ok(graph
1388        .tasks
1389        .values()
1390        .map(|task| {
1391            let keyed = key_now(&snapshot, workspace, graph, task, prepared, cancelled);
1392            (task.id.clone(), keyed)
1393        })
1394        .collect())
1395}
1396
1397fn key_now(
1398    snapshot: &hash::Snapshot,
1399    workspace: &Workspace,
1400    graph: &TaskGraph,
1401    task: &Task,
1402    prepared: &BTreeMap<String, PreparedTask>,
1403    cancelled: &AtomicBool,
1404) -> Keyed {
1405    let failed = |error: anyhow::Error| format!("{}: {error:#}", task.id);
1406    let dependencies =
1407        hash::dependency_keys(snapshot, workspace, graph, task, cancelled).map_err(failed)?;
1408    let inputs = hash::inputs(
1409        snapshot,
1410        workspace,
1411        task,
1412        &prepared[&task.id],
1413        &dependencies,
1414        cancelled,
1415    )
1416    .map_err(failed)?;
1417    Ok((hash::key(&inputs).map_err(failed)?, inputs))
1418}
1419
1420/// Runs the private detached-upload protocol used by the qk executable.
1421#[doc(hidden)]
1422pub fn run_upload_worker() -> anyhow::Result<()> {
1423    remote::background::run()
1424}
1425
1426/// The declared outputs of every target in the workspace, as their targets
1427/// declare them and as each of their configurations does with its options,
1428/// which can move an output; one whose outputs cannot be read is left out.
1429fn workspace_outputs(workspace: &Workspace) -> recorded::WorkspaceOutputs {
1430    let mut outputs = Vec::new();
1431    for (name, project) in &workspace.projects {
1432        for (target, definition) in &project.targets {
1433            let configured = definition
1434                .configurations
1435                .iter()
1436                .map(|(configuration, options)| {
1437                    let mut definition = definition.clone();
1438                    definition.options.extend(options.clone());
1439                    (Some(configuration.clone()), definition)
1440                });
1441            for (configuration, definition) in
1442                std::iter::once((None, definition.clone())).chain(configured)
1443            {
1444                let task = Task {
1445                    id: format!("{name}:{target}"),
1446                    project: name.clone(),
1447                    target: target.clone(),
1448                    configuration: configuration.clone(),
1449                    args: Vec::new(),
1450                    definition,
1451                    dependencies: Default::default(),
1452                    edges: Default::default(),
1453                };
1454                if let Some(declared) = paths::Outputs::new(workspace, &task)
1455                    .ok()
1456                    .filter(paths::Outputs::declared)
1457                {
1458                    outputs.push(((name.clone(), target.clone(), configuration), declared));
1459                }
1460            }
1461        }
1462    }
1463    let roots = workspace
1464        .projects
1465        .iter()
1466        .map(|(name, project)| (name.clone(), project.root.clone()))
1467        .collect();
1468    recorded::WorkspaceOutputs::new(outputs, &roots)
1469}
1470
1471/// The targets of every task `task` depends on, each with the configuration
1472/// it runs with. Its own target is not among them: its outputs are left out
1473/// of its listings as other targets' are.
1474fn upstream(graph: &TaskGraph, task: &Task) -> BTreeSet<recorded::TargetOf> {
1475    let of = |task: &Task| {
1476        (
1477            task.project.clone(),
1478            task.target.clone(),
1479            task.configuration.clone(),
1480        )
1481    };
1482    let mut targets = BTreeSet::new();
1483    let mut seen = BTreeSet::new();
1484    let mut pending: Vec<&String> = task.dependencies.iter().collect();
1485    while let Some(id) = pending.pop() {
1486        if !seen.insert(id) {
1487            continue;
1488        }
1489        if let Some(dependency) = graph.tasks.get(id) {
1490            targets.insert(of(dependency));
1491            pending.extend(&dependency.dependencies);
1492        }
1493    }
1494    targets
1495}
1496
1497#[cfg(test)]
1498mod tests {
1499    use super::*;
1500
1501    #[test]
1502    fn concurrent_blob_writes_keep_results_ordered_and_repair_corruption() {
1503        let workspace = tempfile::tempdir().unwrap();
1504        let root = workspace.path();
1505        let cache = Cache::new(root.join("cache"));
1506        cache.initialize().unwrap();
1507        let mut sources = Vec::new();
1508        for index in 0..130 {
1509            let source = root.join(format!("source-{index}"));
1510            std::fs::write(&source, format!("content-{}", index % 65)).unwrap();
1511            sources.push(source);
1512        }
1513        let digest = cache.put_blob(&sources[0]).unwrap();
1514        std::fs::write(cache.root.join("blobs").join(&digest), "corrupt").unwrap();
1515        std::fs::remove_file(&sources[31]).unwrap();
1516        let results = cache.put_blobs(&sources);
1517        assert_eq!(results.len(), sources.len());
1518        for (index, result) in results.into_iter().enumerate() {
1519            if index == 31 {
1520                assert_eq!(
1521                    result
1522                        .unwrap_err()
1523                        .downcast_ref::<std::io::Error>()
1524                        .unwrap()
1525                        .kind(),
1526                    std::io::ErrorKind::NotFound
1527                );
1528                continue;
1529            }
1530            let digest = result.unwrap();
1531            let held = cache.root.join("blobs").join(&digest);
1532            assert_eq!(
1533                std::fs::read(&held).unwrap(),
1534                format!("content-{}", index % 65).as_bytes()
1535            );
1536            assert_eq!(hash::digest_file(&held).unwrap(), digest);
1537        }
1538    }
1539
1540    /// Content the store holds already is reused, not copied over again.
1541    #[cfg(unix)]
1542    #[test]
1543    fn a_blob_the_store_holds_is_not_written_again() {
1544        use std::os::unix::fs::MetadataExt;
1545        let workspace = tempfile::tempdir().unwrap();
1546        let root = workspace.path();
1547        let cache = Cache::new(root.join(".qk/cache"));
1548        cache.initialize().unwrap();
1549        std::fs::write(root.join("first"), "same").unwrap();
1550        std::fs::write(root.join("second"), "same").unwrap();
1551        std::fs::write(root.join("other"), "different").unwrap();
1552        let first = cache.put_blob(&root.join("first")).unwrap();
1553        let blob = cache.root.join("blobs").join(&first);
1554        let stored = std::fs::metadata(&blob).unwrap().ino();
1555        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1556        assert_eq!(std::fs::metadata(&blob).unwrap().ino(), stored);
1557        // A stored blob whose content no longer matches is stored anew.
1558        std::fs::write(&blob, "corrupt").unwrap();
1559        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1560        assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1561        // So is one that cannot be read.
1562        {
1563            use std::os::unix::fs::PermissionsExt;
1564            std::fs::set_permissions(&blob, std::fs::Permissions::from_mode(0o000)).unwrap();
1565        }
1566        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1567        assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1568        let other = cache.put_blob(&root.join("other")).unwrap();
1569        assert_ne!(other, first);
1570        assert_eq!(
1571            std::fs::read(cache.root.join("blobs").join(other)).unwrap(),
1572            b"different"
1573        );
1574    }
1575
1576    /// Complete-directory restore preserves contents, links, modes and
1577    /// timestamps, and the subsequent hit keeps restored files in place.
1578    #[test]
1579    fn complete_directory_restores_preserve_outputs() {
1580        let workspace = tempfile::tempdir().unwrap();
1581        let root = workspace.path();
1582        std::fs::create_dir_all(root.join("dist/nested/empty")).unwrap();
1583        std::fs::write(root.join("dist/nested/a.txt"), "a").unwrap();
1584        std::fs::write(root.join("dist/b.bin"), [0, 1, 2]).unwrap();
1585        std::fs::create_dir_all(root.join("bundle/empty")).unwrap();
1586        std::fs::write(root.join("bundle/readonly"), "readonly").unwrap();
1587        #[cfg(unix)]
1588        {
1589            use std::os::unix::fs::PermissionsExt;
1590            let script = root.join("dist/run.sh");
1591            std::fs::write(&script, "#!/bin/sh\n").unwrap();
1592            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1593            std::os::unix::fs::symlink("nested/a.txt", root.join("dist/link")).unwrap();
1594            std::fs::set_permissions(
1595                root.join("dist/nested"),
1596                std::fs::Permissions::from_mode(0o500),
1597            )
1598            .unwrap();
1599            std::fs::set_permissions(
1600                root.join("bundle/readonly"),
1601                std::fs::Permissions::from_mode(0o444),
1602            )
1603            .unwrap();
1604        }
1605        let log = root.join("log");
1606        std::fs::write(&log, "").unwrap();
1607
1608        let cache = Cache::new(root.join(".qk/cache"));
1609        cache.initialize().unwrap();
1610        let outputs = paths::Outputs::from_patterns(&["dist", "bundle"]);
1611        let key = "0".repeat(64);
1612        cache.publish(root, &key, &outputs, &log).unwrap();
1613
1614        #[cfg(unix)]
1615        {
1616            use std::os::unix::fs::PermissionsExt;
1617            // The manifest retains the restrictive mode; cleanup needs write
1618            // permission on the old directory to remove its files.
1619            std::fs::set_permissions(
1620                root.join("dist/nested"),
1621                std::fs::Permissions::from_mode(0o755),
1622            )
1623            .unwrap();
1624        }
1625        std::fs::remove_dir_all(root.join("dist")).unwrap();
1626        std::fs::remove_dir_all(root.join("bundle")).unwrap();
1627        let before = std::time::SystemTime::now();
1628        let restored = cache
1629            .restore(
1630                root,
1631                "app:build",
1632                &Default::default(),
1633                &key,
1634                &outputs,
1635                &qk_executor::Display::Stream,
1636                qk_executor::Shown::LocalCache,
1637            )
1638            .unwrap();
1639        assert!(restored);
1640        assert!(root.join("dist/nested/empty").is_dir());
1641        assert!(root.join("bundle/empty").is_dir());
1642        assert_eq!(
1643            std::fs::read_to_string(root.join("bundle/readonly")).unwrap(),
1644            "readonly"
1645        );
1646        #[cfg(unix)]
1647        {
1648            assert_eq!(
1649                std::fs::read_link(root.join("dist/link")).unwrap(),
1650                std::path::Path::new("nested/a.txt")
1651            );
1652            assert_eq!(
1653                store::mode(&std::fs::metadata(root.join("dist/run.sh")).unwrap()),
1654                0o755
1655            );
1656            assert_eq!(
1657                store::mode(&std::fs::metadata(root.join("dist/nested")).unwrap()),
1658                0o500
1659            );
1660            assert_eq!(
1661                store::mode(&std::fs::metadata(root.join("bundle/readonly")).unwrap()),
1662                0o444
1663            );
1664        }
1665        let path = root.join("dist/nested/a.txt");
1666        let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
1667        assert!(modified >= before);
1668        let kept = cache
1669            .restore(
1670                root,
1671                "app:build",
1672                &Default::default(),
1673                &key,
1674                &outputs,
1675                &qk_executor::Display::Hidden,
1676                qk_executor::Shown::LocalCache,
1677            )
1678            .unwrap();
1679        assert!(kept);
1680        assert_eq!(
1681            std::fs::metadata(&path).unwrap().modified().unwrap(),
1682            modified
1683        );
1684        #[cfg(unix)]
1685        {
1686            use std::os::unix::fs::PermissionsExt;
1687            std::fs::set_permissions(
1688                root.join("dist/nested"),
1689                std::fs::Permissions::from_mode(0o755),
1690            )
1691            .unwrap();
1692        }
1693    }
1694
1695    #[test]
1696    fn listings_leave_out_other_targets_outputs_but_the_upstreams() {
1697        let temp = tempfile::tempdir().unwrap();
1698        let root = temp.path();
1699        std::fs::write(root.join("nx.json"), "{}").unwrap();
1700        std::fs::create_dir_all(root.join("app")).unwrap();
1701        std::fs::create_dir_all(root.join("lib")).unwrap();
1702        // A project at the workspace root, with an output in a folder of its
1703        // own and one inside another project.
1704        std::fs::write(
1705            root.join("project.json"),
1706            r#"{"name":"workspace","targets":{"cache":{"command":"true",
1707                "outputs":["{projectRoot}/.cache/build","{projectRoot}/app/extra/out"]}}}"#,
1708        )
1709        .unwrap();
1710        std::fs::write(
1711            root.join("app/project.json"),
1712            r#"{"name":"app","root":"./app","targets":{
1713                "build":{"command":"true","outputs":["{options.outputPath}"],
1714                  "options":{"outputPath":"app/dist"},
1715                  "configurations":{"production":{"outputPath":"app/dist-production"}}},
1716                "lint":{"command":"true"},
1717                "cover":{"command":"true","outputs":["{projectRoot}/.coverage/unit"]},
1718                "cover-browser":{"command":"true","outputs":["{projectRoot}/.coverage/browser"]},
1719                "wasm":{"command":"true","outputs":["{projectRoot}/target/wasm/release/*.wasm"]},
1720                "test":{"command":"true","dependsOn":[{"projects":["lib"],"target":"build"}],
1721                  "configurations":{"production":{}}}}}"#,
1722        )
1723        .unwrap();
1724        std::fs::write(
1725            root.join("lib/project.json"),
1726            r#"{"name":"lib","targets":{"build":{"command":"true","outputs":["{options.outputPath}"],
1727                "options":{"outputPath":"lib/dist"},
1728                "configurations":{"production":{"outputPath":"lib/dist-production"}}}}}"#,
1729        )
1730        .unwrap();
1731        let workspace = Workspace::load(root).unwrap();
1732        let graph = qk_taskgraph::TaskGraph::build(
1733            &workspace,
1734            &[
1735                qk_taskgraph::Request::parse("app:lint").unwrap(),
1736                qk_taskgraph::Request::parse("app:cover").unwrap(),
1737                qk_taskgraph::Request::parse("app:test").unwrap(),
1738                qk_taskgraph::Request::parse("app:test:production").unwrap(),
1739            ],
1740        )
1741        .unwrap();
1742        let outputs = workspace_outputs(&workspace);
1743        let others = |task: &str| recorded::Others {
1744            outputs: &outputs,
1745            upstream: upstream(&graph, &graph.tasks[task]),
1746        };
1747        let lint = others("app:lint");
1748        assert!(lint.matches("app/dist/main.js"));
1749        assert!(lint.matches("app/dist-production/main.js"));
1750        assert!(lint.matches("lib/dist"));
1751        assert!(!lint.matches("app/src/main.ts"));
1752        // The directories holding an output count as it, not its project's.
1753        assert!(lint.matches("app/target"));
1754        assert!(lint.matches("app/target/wasm"));
1755        assert!(!lint.matches("app"));
1756        // A root project's folders are top-level, but never another project.
1757        assert!(lint.matches(".cache"));
1758        assert!(lint.matches("app/extra"));
1759        assert!(!lint.matches("lib"));
1760        // Before the task runs, the directories those folders come and go
1761        // in are listed, another project's root among them.
1762        let unlisted = recorded::Unlisted {
1763            own: Vec::new(),
1764            others: Some(lint),
1765            ignore: &[],
1766        };
1767        let directories = unlisted.directories("lib");
1768        for directory in [".", ".cache", "app", "app/extra"] {
1769            assert!(directories.contains(directory), "{directory}");
1770        }
1771        // A dependency's output is the test's input, and only that of the
1772        // configuration it depends on.
1773        assert!(!others("app:test").matches("lib/dist"));
1774        assert!(others("app:test").matches("lib/dist-production"));
1775        // A folder holding the task's own output and another target's is
1776        // left out, as both are.
1777        assert!(others("app:cover").matches("app/.coverage"));
1778        let production = others("app:test:production");
1779        assert!(!production.matches("lib/dist-production"));
1780        assert!(production.matches("lib/dist"));
1781    }
1782}