1mod ahead;
4mod evict;
5pub mod expectations;
6mod glob;
7mod hash;
8mod ignored;
9mod inputs;
10mod lookup;
11mod paths;
12pub mod profile;
13pub mod projections;
14mod record;
15pub mod recorded;
16mod remote;
17pub mod scratch;
18mod store;
19pub mod ultracache;
20mod verified;
21pub mod warm;
22
23use std::collections::{BTreeMap, BTreeSet};
24use std::path::PathBuf;
25use std::sync::OnceLock;
26use std::sync::atomic::{AtomicBool, Ordering};
27
28use anyhow::Result;
29use qk_config::Workspace;
30use qk_executor::{Capture, Outcome, PreparedTask, execute, execute_captured};
31use qk_taskgraph::{Task, TaskGraph};
32use serde_json::Value;
33
34pub use evict::{Pruned, max_size, parse_size, prune, prune_automatic};
35pub use glob::Pattern;
36pub use hash::{
37 DependencyGlob, Selects, SourceIgnore, dependency_globs, keyed_definition, source_files,
38 task_configuration,
39};
40pub use inputs::{
41 InputChange, InputChanges, JsonFields, ROOT_TSCONFIG_FILES, Resolution, Resolved, TsConfig,
42 manifest_for_dependents, resolve_each, resolve_tasks, selectively_hash_tsconfig,
43 without_resolution,
44};
45pub use paths::expand as expand_project_path;
47pub use paths::git as workspace_git;
48pub use paths::{
49 Outputs, cache_directory, cache_location, clear_worktree_state, resolved_outputs,
50 worktree_state,
51};
52
53fn remote_settings(workspace: &Workspace) -> (Option<&Value>, Option<&Value>) {
57 let extra = &workspace.config.extra;
58 (extra.get("s3"), extra.get("qk:storeServer"))
59}
60
61fn execute_uncached(
64 task: &PreparedTask,
65 cancelled: &AtomicBool,
66) -> Result<(Outcome, Option<Execution>)> {
67 if task.sandbox.is_none() {
68 return Ok((execute(task, cancelled)?, None));
69 }
70 let capture = Capture::new(None, task.display.clone()).retain_log();
71 let outcome = execute_captured(task, cancelled, Some(&capture))?;
72 let execution = capture.take_log().map(|log| Execution {
73 log,
74 inputs_unchanged: false,
75 });
76 Ok((outcome, execution))
77}
78
79fn scratch_patterns<'a>(
82 root: &'a std::path::Path,
83 prepared: &'a PreparedTask,
84) -> impl Iterator<Item = Pattern> + 'a {
85 prepared.scratch.iter().filter_map(move |scratch| {
86 let relative = scratch.directory.strip_prefix(root).ok()?.to_str()?;
87 let named = glob::escape(&relative.replace('\\', "/"));
89 Pattern::fileset(&named, false).ok()
90 })
91}
92
93impl Cache {
94 fn ignored_reads(
98 &self,
99 workspace: &Workspace,
100 task: &Task,
101 prepared: &PreparedTask,
102 written: &[String],
103 ) -> Result<ultracache::IgnoredReads> {
104 let projects = written
105 .iter()
106 .any(|glob| glob.starts_with('^'))
107 .then(|| {
108 self.projects
109 .get_or_init(|| qk_graph::ProjectGraph::build(workspace).ok())
110 .as_ref()
111 })
112 .flatten();
113 Ok(
114 ultracache::IgnoredReads::new(written, workspace, projects, &task.project)?
115 .with_patterns(scratch_patterns(&workspace.root, prepared)),
116 )
117 }
118}
119
120pub struct Cache {
122 pub root: PathBuf,
123 snapshot: OnceLock<std::result::Result<hash::Snapshot, String>>,
124 remote: Option<std::sync::Arc<remote::Remote>>,
125 reuse: bool,
127 cold: bool,
131 cleared: std::sync::Mutex<std::collections::BTreeSet<String>>,
134 prefetches: OnceLock<warm::Prefetches>,
136 verified: OnceLock<verified::Verified>,
139 kept: std::sync::Mutex<std::collections::BTreeSet<String>>,
141 saves: std::sync::Mutex<Vec<std::thread::JoinHandle<()>>>,
143 cleanup: std::sync::Mutex<Option<store::Cleanup>>,
144 workspace_outputs: OnceLock<recorded::WorkspaceOutputs>,
146 projects: OnceLock<Option<qk_graph::ProjectGraph>>,
149}
150
151#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
153#[serde(rename_all = "kebab-case")]
154pub enum CacheStatus {
155 LocalHit,
156 RemoteHit,
157 Miss,
158 Uncached,
160}
161
162pub struct Execution {
164 pub log: qk_executor::RecordedLog,
165 pub inputs_unchanged: bool,
166}
167
168pub struct TaskResult {
169 pub outcome: Outcome,
170 pub cache: CacheStatus,
171 pub key: Option<(String, Value)>,
173 pub warm: Option<warm::WarmReport>,
175 pub execution: Option<Execution>,
176 pub recorded: Option<Recorded>,
178 pub shadow: Option<Shadow>,
181}
182
183#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
185#[serde(rename_all = "camelCase", tag = "kind")]
186pub enum Shadow {
187 WouldHit {
189 key: String,
190 #[serde(default, skip_serializing_if = "Option::is_none")]
192 check: Option<Check>,
193 },
194 WouldMiss {
196 #[serde(default, skip_serializing_if = "Vec::is_empty")]
203 changed: Vec<String>,
204 #[serde(default, skip_serializing_if = "Vec::is_empty")]
208 uncovered: Vec<String>,
209 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
213 declared: bool,
214 },
215 Unrecorded,
217}
218
219pub const UNCOVERED: usize = 5;
221
222#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
227#[serde(rename_all = "camelCase")]
228pub struct Check {
229 pub matched: bool,
230 #[serde(default, skip_serializing_if = "Vec::is_empty")]
233 pub differences: Vec<String>,
234 #[serde(default, skip_serializing_if = "is_zero")]
237 pub unchecked: usize,
238}
239
240fn is_zero(count: &usize) -> bool {
241 *count == 0
242}
243
244enum Shadowing {
248 Settled(Shadow),
251 Fetching {
254 search: lookup::Search,
255 fetch: std::thread::JoinHandle<Result<Option<Vec<u8>>>>,
256 },
257}
258
259fn recording_key(
262 weak: &str,
263 set: &recorded::PathSet,
264 snapshot: &hash::Snapshot,
265 workspace: &Workspace,
266 unlisted: &recorded::Unlisted,
267 task: &Task,
268) -> Option<String> {
269 recorded::strong_key(weak, set, snapshot, &workspace.root, unlisted)
270 .inspect_err(|error| {
271 qk_executor::status!("qk: {}: ignoring a path set ({error:#})", task.id);
272 })
273 .ok()
274}
275
276struct Restoring<'a> {
279 cache: &'a Cache,
280 workspace: &'a Workspace,
281 task: &'a Task,
282 prepared: &'a PreparedTask,
283 outputs: &'a paths::Outputs,
284 cancelled: &'a AtomicBool,
285}
286
287impl Restoring<'_> {
288 fn restored(&self, key: &str, shown: qk_executor::Shown) -> bool {
289 self.cache
290 .restore(
291 &self.workspace.root,
292 &self.task.id,
293 &self.task.dependencies,
294 key,
295 self.outputs,
296 &self.prepared.display,
297 shown,
298 )
299 .unwrap_or_else(|error| {
300 qk_executor::status!(
301 "qk: {}: ignoring unusable cache entry ({error})",
302 self.task.id
303 );
304 false
305 })
306 }
307}
308
309impl lookup::Results for Restoring<'_> {
310 fn here(&self, key: &str) -> bool {
311 self.restored(key, qk_executor::Shown::LocalCache)
312 }
313
314 fn there(&self, key: &str) -> bool {
315 let Some(remote) = &self.cache.remote else {
316 return false;
317 };
318 remote
319 .fetch(&self.cache.root, key, &self.task.id, self.cancelled)
320 .unwrap_or_else(|error| {
321 qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", self.task.id);
322 false
323 })
324 && self.restored(key, qk_executor::Shown::RemoteCache)
325 }
326}
327
328struct Checking<'a> {
332 cache: &'a Cache,
333 task: &'a Task,
334 outputs: &'a paths::Outputs,
335}
336
337impl lookup::Results for Checking<'_> {
338 fn here(&self, key: &str) -> bool {
339 self.cache
340 .usable(&self.task.id, key, self.outputs)
341 .unwrap_or(false)
342 }
343
344 fn there(&self, key: &str) -> bool {
345 let Some(remote) = &self.cache.remote else {
346 return false;
347 };
348 remote.holds(key).unwrap_or(false)
349 }
350}
351
352#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
354#[serde(rename_all = "camelCase", tag = "kind")]
355pub enum Recorded {
356 Hit { key: String },
358 Saved { key: String },
360 Missed,
362 Unrecorded,
365 Declared { key: String },
368 Failed { reason: String },
370}
371
372impl TaskResult {
373 pub fn uncached(outcome: Outcome) -> Self {
374 Self {
375 outcome,
376 cache: CacheStatus::Uncached,
377 key: None,
378 warm: None,
379 execution: None,
380 recorded: None,
381 shadow: None,
382 }
383 }
384}
385
386impl Cache {
387 pub fn new(root: PathBuf) -> Self {
388 Self {
389 root,
390 snapshot: OnceLock::new(),
391 remote: None,
392 reuse: true,
393 cold: false,
394 cleared: Default::default(),
395 prefetches: OnceLock::new(),
396 verified: OnceLock::new(),
397 kept: Default::default(),
398 saves: Default::default(),
399 cleanup: Default::default(),
400 workspace_outputs: OnceLock::new(),
401 projects: OnceLock::new(),
402 }
403 }
404
405 pub(crate) fn verified(&self) -> &verified::Verified {
407 self.verified
408 .get_or_init(|| verified::Verified::load(&self.root))
409 }
410
411 pub fn for_observation(workspace: &Workspace) -> Self {
413 Self {
414 reuse: false,
415 ..Self::new(paths::cache_location(workspace))
416 }
417 }
418
419 pub fn for_workspace(
425 workspace: &Workspace,
426 environment: &BTreeMap<std::ffi::OsString, std::ffi::OsString>,
427 ) -> Self {
428 let root = paths::cache_location(workspace);
429 let (s3, server) = remote_settings(workspace);
430 let remote = match remote::configure(s3, server, environment) {
431 Ok(remote) => remote.map(std::sync::Arc::new),
432 Err(error) => {
433 qk_executor::status!("qk: remote cache disabled: {error:#}");
434 None
435 }
436 };
437 if let Some(remote) = &remote {
438 remote.read_index(&root);
439 }
440 Self {
441 remote,
442 ..Self::new(root)
443 }
444 }
445
446 pub fn cold(self) -> Self {
449 Self { cold: true, ..self }
450 }
451
452 pub fn finish(&self, workspace: &Workspace) {
455 if !self.reuse {
456 return;
457 }
458 if let Some(Ok(snapshot)) = self.snapshot.get() {
459 snapshot.save_digests(&workspace.root);
460 }
461 self.finish_prefetches();
462 for save in std::mem::take(&mut *self.saves.lock().unwrap()) {
464 let _ = save.join();
465 }
466 if let Some(remote) = &self.remote {
467 let failures = remote.finish();
468 for failure in &failures {
469 qk_executor::status!("qk: remote cache upload failed for {failure}");
470 }
471 }
472 let _profile = profile::span("cache", "restore_cleanup_wait");
473 self.cleanup.lock().unwrap().take();
474 }
475
476 fn snapshot(
477 &self,
478 workspace: &Workspace,
479 graph: &TaskGraph,
480 ) -> std::result::Result<&hash::Snapshot, String> {
481 self.snapshot
482 .get_or_init(|| {
483 hash::Snapshot::new(workspace, graph, &self.root)
484 .map_err(|error| format!("cannot snapshot the workspace: {error:#}"))
485 })
486 .as_ref()
487 .map_err(Clone::clone)
488 }
489
490 pub fn run(
491 &self,
492 workspace: &Workspace,
493 graph: &TaskGraph,
494 task: &Task,
495 prepared: &PreparedTask,
496 cancelled: &AtomicBool,
497 ) -> Result<TaskResult> {
498 let result = self.run_task(workspace, graph, task, prepared, cancelled);
499 if let Some(Ok(snapshot)) = self.snapshot.get() {
500 snapshot.outputs_written(workspace, task);
501 }
502 result
503 }
504
505 fn run_task(
506 &self,
507 workspace: &Workspace,
508 graph: &TaskGraph,
509 task: &Task,
510 prepared: &PreparedTask,
511 cancelled: &AtomicBool,
512 ) -> Result<TaskResult> {
513 let cacheable = task.definition.cache == Some(true);
514 let _kept = self
516 .reuse
517 .then(|| warm::keep_across(workspace, graph, task));
518 let entries = if self.reuse {
519 warm::config(workspace, task).unwrap_or_else(|error| {
520 qk_executor::status!("qk: {}: warm state ignored ({error:#})", task.id);
521 Vec::new()
522 })
523 } else {
524 Vec::new()
525 };
526 let entry_label = |warm: &warm::Warm| match warm.group() {
528 Some(group) => format!(" (group {group})"),
529 None => String::new(),
530 };
531 let mut running = prepared.clone();
534 for warm in &entries {
535 for (name, value) in &warm.env {
536 running.execution.insert(name.into(), value.into());
537 }
538 }
539 let uncleared = std::sync::Mutex::new(std::collections::BTreeSet::new());
543 let before = |running: &PreparedTask| -> (Vec<warm::Restored>, Vec<String>) {
544 let (mut restored, mut present) = (Vec::new(), Vec::new());
545 for warm in &entries {
546 if self.cold {
547 if let Err(error) = self.clear_warm(workspace, task, warm) {
549 qk_executor::status!(
550 "qk: {}: warm state{} not cleared, so not saved ({error:#})",
551 task.id,
552 entry_label(warm)
553 );
554 uncleared.lock().unwrap().insert(warm.identity.clone());
555 }
556 continue;
557 }
558 match self
559 .initialize()
560 .and_then(|()| self.restore_warm(workspace, task, warm, running))
561 {
562 Ok((entry, groups)) => {
563 if !entry.groups.is_empty() {
564 restored.push(entry);
565 }
566 present.extend(groups);
567 }
568 Err(error) => qk_executor::status!(
569 "qk: {}: warm state{} not restored ({error:#})",
570 task.id,
571 entry_label(warm)
572 ),
573 }
574 }
575 (restored, present)
576 };
577 let after = |outcome: Outcome,
578 (restored, present): (Vec<warm::Restored>, Vec<String>),
579 running: &PreparedTask|
580 -> Option<warm::WarmReport> {
581 if entries.is_empty() {
582 return None;
583 }
584 let mut save_ms = None;
585 let mut background = false;
586 let uncleared = uncleared.lock().unwrap();
587 for warm in entries
588 .iter()
589 .filter(|warm| outcome == Outcome::Success && !uncleared.contains(&warm.identity))
590 {
591 let started = std::time::Instant::now();
592 let saved = if warm.background {
593 self.save_warm_in_background(workspace, task, warm, running)
594 } else {
595 self.save_warm(workspace, task, warm, running)
596 };
597 match saved {
598 Ok(()) if warm.background => background = true,
599 Ok(()) => {
600 *save_ms.get_or_insert(0) += started.elapsed().as_millis() as u64;
601 }
602 Err(error) => qk_executor::status!(
603 "qk: {}: warm state{} not saved ({error:#})",
604 task.id,
605 entry_label(warm)
606 ),
607 }
608 }
609 Some(warm::WarmReport {
610 restored,
611 present,
612 save_ms,
613 background,
614 })
615 };
616 let bypass = |reason: String| {
618 if cacheable {
619 qk_executor::status!("qk: {}: cache bypassed ({reason})", task.id);
620 }
621 let _profile = profile::span(&task.id, "execute");
622 let (outcome, execution) = execute_uncached(prepared, cancelled)?;
623 Ok(TaskResult {
624 execution,
625 ..TaskResult::uncached(outcome)
626 })
627 };
628 let snapshot = match self.snapshot(workspace, graph) {
629 Ok(snapshot) => snapshot,
630 Err(reason) => return bypass(reason),
631 };
632 let dependency_profile = profile::span(&task.id, "dependency_inputs");
633 let dependencies = match hash::dependency_keys(snapshot, workspace, graph, task, cancelled)
634 {
635 Ok(dependencies) => dependencies,
636 Err(error) => return bypass(format!("{error:#}")),
637 };
638 drop(dependency_profile);
639 let input_profile = profile::span(&task.id, "inputs");
640 let (key, inputs) = match hash::inputs(
641 snapshot,
642 workspace,
643 task,
644 prepared,
645 &dependencies,
646 cancelled,
647 )
648 .and_then(|inputs| Ok((hash::key(&inputs)?, inputs)))
649 {
650 Ok(key) => key,
651 Err(error) => return bypass(format!("{error:#}")),
652 };
653 drop(input_profile);
654 let keyed = Some((key.clone(), inputs));
655 running
657 .execution
658 .insert("NX_TASK_HASH".into(), key.clone().into());
659 let running = &running;
660 let outputs = match paths::Outputs::new(workspace, task) {
661 Ok(outputs) => outputs,
662 Err(error) => return bypass(format!("{error:#}")),
663 };
664 if !cacheable {
665 let restored = before(running);
666 let executing = profile::span(&task.id, "execute");
667 let (outcome, execution) = execute_uncached(running, cancelled)?;
668 drop(executing);
669 let warm_report = after(outcome, restored, running);
670 return Ok(TaskResult {
671 outcome,
672 cache: CacheStatus::Uncached,
673 key: keyed,
674 warm: warm_report,
675 execution,
676 recorded: None,
677 shadow: None,
678 });
679 }
680 if !self.reuse {
681 let capture = Capture::new(None, prepared.display.clone()).retain_log();
682 let executing = profile::span(&task.id, "execute");
683 let outcome = execute_captured(running, cancelled, Some(&capture))?;
684 drop(executing);
685 let inputs_unchanged =
686 self.inputs_match(snapshot, workspace, graph, task, prepared, &key, cancelled);
687 return Ok(TaskResult {
688 outcome,
689 cache: CacheStatus::Uncached,
690 key: keyed,
691 warm: None,
692 execution: capture.take_log().map(|log| Execution {
693 log,
694 inputs_unchanged,
695 }),
696 recorded: None,
697 shadow: None,
698 });
699 }
700 if let Err(error) = self.initialize() {
701 return bypass(format!("cache unavailable: {error}"));
702 }
703 let (_lock, waited) = match self.lock(&key, cancelled) {
704 Ok(Some(lock)) => lock,
705 Ok(None) => {
706 return Ok(TaskResult::uncached(Outcome::Cancelled));
707 }
708 Err(error) => return bypass(format!("cache lock unavailable: {error}")),
709 };
710 if waited
712 && hash::recheck_dependency_keys(snapshot, workspace, graph, task, cancelled)
713 .and_then(|dependencies| {
714 hash::fingerprint(
715 snapshot,
716 workspace,
717 task,
718 prepared,
719 &dependencies,
720 cancelled,
721 )
722 })
723 .ok()
724 .as_ref()
725 != Some(&key)
726 {
727 return bypass("inputs changed while waiting for another run".into());
728 }
729 let settings = ultracache::Settings::of(task).unwrap_or_else(|error| {
732 qk_executor::status!("qk: {}: qk:ultracache ignored ({error:#})", task.id);
733 None
734 });
735 let warm_outputs = paths::Outputs::from_paths(&warm::kept_paths(&entries)).ok();
736 let own: Vec<&paths::Outputs> = std::iter::once(&outputs)
737 .chain(warm_outputs.as_ref())
738 .collect();
739 let others = settings.as_ref().map(|_| recorded::Others {
743 outputs: self
744 .workspace_outputs
745 .get_or_init(|| workspace_outputs(workspace)),
746 upstream: upstream(graph, task),
747 });
748 let ignore = settings
749 .as_ref()
750 .and_then(|settings| {
751 self.ignored_reads(workspace, task, prepared, &settings.ignore_reads)
752 .ok()
753 })
754 .unwrap_or_default();
755 let unlisted = recorded::Unlisted {
756 own: own.clone(),
757 others,
758 ignore: &ignore,
759 };
760 let inputs = &keyed.as_ref().expect("a cacheable task is keyed").1;
761 let mut unkeyable = None;
763 let mut recordings = settings.and_then(|settings| {
764 let weak = hash::weak_key(
765 snapshot,
766 workspace,
767 graph,
768 task,
769 prepared,
770 inputs,
771 &settings.ignore_reads,
772 &ignore.scope(),
773 cancelled,
774 );
775 match weak {
776 Ok(weak) => Some((recorded::load(&self.root, &weak), weak, settings)),
777 Err(error) => {
778 qk_executor::status!(
779 "qk: {}: not keyed by its recordings ({error:#})",
780 task.id
781 );
782 unkeyable = Some((settings.mode, format!("{error:#}")));
783 None
784 }
785 }
786 });
787 let mut shadowing = None;
788 let mut replaced = Vec::new();
791 if let Some((sets, weak, settings)) = &mut recordings {
792 match lookup::lookup(settings.mode, self.cold) {
793 lookup::Lookup::Replace => {
795 replaced = sets
796 .iter()
797 .filter_map(|set| {
798 recording_key(weak, set, snapshot, workspace, &unlisted, task)
799 })
800 .collect();
801 }
802 lookup::Lookup::Shadow => {
803 let _lookup = profile::span(&task.id, "recorded_lookup");
804 shadowing = Some(self.start_shadow(
805 workspace, task, snapshot, &outputs, &unlisted, sets, weak, &key,
806 ));
807 }
808 lookup::Lookup::Restore => {
809 let recorded_profile = profile::span(&task.id, "recorded_restore");
810 let found = self.by_recordings(
811 workspace, task, prepared, snapshot, &outputs, &unlisted, sets, weak, &key,
812 cancelled,
813 );
814 drop(recorded_profile);
815 let remote = matches!(found, lookup::Found::Remote(_));
816 if let lookup::Found::Local(strong) | lookup::Found::Remote(strong) = found {
817 if let Some(store) = self.remote.as_ref().filter(|_| !remote) {
818 store.held(&strong);
819 }
820 recorded::touch(&self.root, weak);
821 qk_executor::report::event(qk_executor::report::Event::Cache {
822 id: task.id.clone(),
823 cached: if remote {
824 qk_executor::report::Cached::RemoteHit
825 } else {
826 qk_executor::report::Cached::Hit
827 },
828 });
829 return Ok(TaskResult {
830 outcome: Outcome::Success,
831 cache: if remote {
832 CacheStatus::RemoteHit
833 } else {
834 CacheStatus::LocalHit
835 },
836 key: keyed,
837 warm: None,
838 execution: None,
839 recorded: lookup::recorded(
840 Some(settings.mode),
841 !sets.is_empty(),
842 lookup::Taken::Recording(strong),
843 ),
844 shadow: None,
845 });
846 }
847 }
848 }
849 }
850 let mut settle = |recordings: &mut Option<(Vec<recorded::PathSet>, String, _)>| {
852 let shadowing = shadowing.take()?;
853 let (sets, weak, _) = recordings.as_mut()?;
854 let _wait = profile::span(&task.id, "recorded_wait");
855 Some(self.settle_shadow(
856 shadowing, workspace, task, snapshot, &outputs, &unlisted, sets, weak,
857 ))
858 };
859 let mode = recordings.as_ref().map(|(_, _, settings)| settings.mode);
862 let had_sets = |recordings: &Option<(Vec<recorded::PathSet>, String, _)>| {
863 recordings
864 .as_ref()
865 .is_some_and(|(sets, _, _)| !sets.is_empty())
866 };
867 let declared = lookup::by_declared(
868 mode,
869 self.cold,
870 recordings.as_ref().map_or(&[], |(sets, _, _)| sets),
871 |sets| {
872 hash::coverage(snapshot, workspace, graph, task, prepared, cancelled)
873 .is_ok_and(|coverage| self.covers_all(&coverage, snapshot, workspace, sets))
874 },
875 );
876 let verdict = |had_sets: bool, taken: lookup::Taken| match &unkeyable {
878 Some((mode, reason)) => lookup::unkeyable(*mode, reason, taken),
879 None => lookup::recorded(mode, had_sets, taken),
880 };
881 let unrecorded = verdict(had_sets(&recordings), lookup::Taken::Declared(key.clone()));
882 let prefetched = declared
885 && self.remote.as_ref().is_some_and(|remote| {
886 remote.resolved(&key);
887 remote.prefetched(&key)
888 });
889 let local_profile = profile::span(&task.id, "local_restore");
890 let restored = if declared && !self.cold {
891 self.restore(
892 &workspace.root,
893 &task.id,
894 &task.dependencies,
895 &key,
896 &outputs,
897 &prepared.display,
898 if prefetched {
899 qk_executor::Shown::RemoteCache
900 } else {
901 qk_executor::Shown::LocalCache
902 },
903 )
904 } else {
905 Ok(false)
906 };
907 match restored {
908 Ok(true) => {
909 let claimed = match &self.remote {
910 Some(remote) if prefetched && remote.claim(&key) => true,
911 Some(remote) => {
912 remote.held(&key);
913 false
914 }
915 None => false,
916 };
917 qk_executor::report::event(qk_executor::report::Event::Cache {
918 id: task.id.clone(),
919 cached: if claimed {
920 qk_executor::report::Cached::RemoteHit
921 } else {
922 qk_executor::report::Cached::Hit
923 },
924 });
925 let shadow = settle(&mut recordings);
926 let shadow = self.with_changes(
927 shadow,
928 recordings.as_ref(),
929 (snapshot, workspace, graph, task, prepared, &unlisted),
930 cancelled,
931 );
932 return Ok(TaskResult {
933 outcome: Outcome::Success,
934 cache: if claimed {
935 CacheStatus::RemoteHit
936 } else {
937 CacheStatus::LocalHit
938 },
939 key: keyed,
940 warm: None,
941 execution: None,
942 recorded: unrecorded,
943 shadow,
944 });
945 }
946 Ok(false) => {}
947 Err(error) => {
948 qk_executor::status!("qk: {}: ignoring unusable cache entry ({error})", task.id)
949 }
950 }
951 drop(local_profile);
952 if let Some(remote) = self.remote.as_ref().filter(|_| declared && !self.cold) {
953 let fetch_profile = profile::span(&task.id, "remote_fetch");
954 let fetched = remote.fetch(&self.root, &key, &task.id, cancelled);
955 drop(fetch_profile);
956 let fetched = fetched.and_then(|found| {
957 if !found {
958 return Ok(false);
959 }
960 self.restore(
961 &workspace.root,
962 &task.id,
963 &task.dependencies,
964 &key,
965 &outputs,
966 &prepared.display,
967 qk_executor::Shown::RemoteCache,
968 )
969 });
970 match fetched {
971 Ok(true) => {
972 qk_executor::report::event(qk_executor::report::Event::Cache {
973 id: task.id.clone(),
974 cached: qk_executor::report::Cached::RemoteHit,
975 });
976 let shadow = settle(&mut recordings);
977 let shadow = self.with_changes(
978 shadow,
979 recordings.as_ref(),
980 (snapshot, workspace, graph, task, prepared, &unlisted),
981 cancelled,
982 );
983 return Ok(TaskResult {
984 outcome: Outcome::Success,
985 cache: CacheStatus::RemoteHit,
986 key: keyed,
987 warm: None,
988 execution: None,
989 recorded: unrecorded,
990 shadow,
991 });
992 }
993 Ok(false) => {}
994 Err(error) => {
995 qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", task.id)
996 }
997 }
998 }
999 let shadow = settle(&mut recordings);
1001 qk_executor::report::event(qk_executor::report::Event::Cache {
1002 id: task.id.clone(),
1003 cached: qk_executor::report::Cached::Miss,
1004 });
1005 let log = match tempfile::NamedTempFile::new_in(self.root.join("tmp")) {
1006 Ok(log) => log,
1007 Err(error) => return bypass(format!("cache log unavailable: {error}")),
1008 };
1009 let capture =
1010 Capture::new(Some(log.as_file().try_clone()?), prepared.display.clone()).retain_log();
1011 let restored = before(running);
1012 let keying = prepared
1015 .recording
1016 .as_ref()
1017 .filter(|recording| recording.keys());
1018 let listed = match (&recordings, keying) {
1019 (Some(_), Some(_)) => {
1020 let project_root = workspace
1021 .projects
1022 .get(&task.project)
1023 .map_or("", |project| project.root.as_str());
1024 recorded::Before::take(
1025 &workspace.root,
1026 unlisted.directories(project_root),
1027 &unlisted,
1028 )
1029 }
1030 _ => recorded::Before::default(),
1031 };
1032 let started = recorded::now(&self.root.join("tmp"));
1033 let executing = profile::span(&task.id, "execute");
1034 let outcome = execute_captured(running, cancelled, Some(&capture))?;
1035 drop(executing);
1036 let warm_report = after(outcome, restored, running);
1037 let inputs_unchanged =
1038 self.inputs_match(snapshot, workspace, graph, task, prepared, &key, cancelled);
1039 let execution = capture.take_log().map(|log| Execution {
1040 log,
1041 inputs_unchanged,
1042 });
1043 let path_set = match (&recordings, keying) {
1044 (Some((_, _, settings)), Some(recording)) => Some(
1045 self.path_set(workspace, task, recording, prepared, settings, &own)
1046 .and_then(|set| {
1047 if set.is_empty() {
1049 return Err("its trace named no path in the workspace".to_owned());
1050 }
1051 match set
1052 .changed_since(&workspace.root, started, &unlisted, &listed)
1053 .map(|changed| changed.to_string())
1054 {
1055 Some(reason) => Err(reason),
1056 None => Ok(set),
1057 }
1058 }),
1059 ),
1060 _ => None,
1061 };
1062 if let Some(Err(reason)) = &path_set {
1063 qk_executor::status!("qk: {}: not recorded ({reason})", task.id);
1064 }
1065 let mut published = false;
1066 let kept = outcome == Outcome::Success && inputs_unchanged && capture.healthy();
1067 if kept {
1068 let publishing = profile::span(&task.id, "publish");
1069 let publish = self.publish(&workspace.root, &key, &outputs, log.path());
1070 drop(publishing);
1071 match publish {
1072 Ok(()) => {
1073 published = true;
1074 store::record_outputs(&workspace.root, &task.id, &key, &outputs);
1075 if let Some(remote) = &self.remote {
1076 remote.upload(&self.root, &key);
1077 }
1078 for strong in &replaced {
1079 if let Err(error) =
1080 self.publish(&workspace.root, strong, &outputs, log.path())
1081 {
1082 qk_executor::status!(
1083 "qk: {}: could not replace a recorded cache entry ({error})",
1084 task.id
1085 );
1086 }
1087 }
1088 }
1089 Err(error) => {
1090 qk_executor::status!("qk: {}: could not save cache entry ({error})", task.id);
1091 }
1092 }
1093 }
1094 let check = match &shadow {
1098 Some(Shadow::WouldHit { key: recorded, .. }) => match outcome {
1099 Outcome::Failed(code) => Some(Check {
1100 matched: false,
1101 differences: vec![format!("failed (exit {code})")],
1102 unchecked: 0,
1103 }),
1104 Outcome::Success if published => {
1105 let _checking = profile::span(&task.id, "recorded_check");
1106 let settings = recordings.as_ref().map(|(_, _, settings)| settings);
1107 self.check(&task.id, &key, recorded, settings)
1108 }
1109 _ => None,
1110 },
1111 _ => None,
1112 };
1113 let ran = match (path_set, &recordings) {
1114 (Some(Err(reason)), _) => lookup::Ran::Refused(reason),
1115 (Some(Ok(set)), Some((_, weak, _))) if kept => {
1117 let saved = recorded::with_values(weak, set, snapshot, &workspace.root, &unlisted)
1118 .and_then(|(strong, set)| {
1119 self.publish(&workspace.root, &strong, &outputs, log.path())?;
1120 recorded::save(&self.root, weak, set)?;
1121 if let Some(remote) = &self.remote {
1124 remote.upload(&self.root, &strong);
1125 if let Some(record) = recorded::record(&self.root, weak) {
1126 remote.upload_path_sets(&self.root, weak, record);
1127 }
1128 }
1129 Ok(strong)
1130 });
1131 match saved {
1132 Ok(key) => lookup::Ran::Saved(key),
1133 Err(error) => {
1134 qk_executor::status!("qk: {}: recording not saved ({error:#})", task.id);
1135 lookup::Ran::Unsaved(format!("{error:#}"))
1136 }
1137 }
1138 }
1139 (Some(Ok(_)), _) => {
1140 let why = if outcome != Outcome::Success {
1142 None
1143 } else if !inputs_unchanged {
1144 Some("its inputs changed while it ran")
1145 } else {
1146 Some("its log could not be written")
1147 };
1148 if let Some(why) = why {
1149 qk_executor::status!("qk: {}: recording not saved ({why})", task.id);
1150 }
1151 lookup::Ran::Unkept
1152 }
1153 (None, _) => match (&unkeyable, keying) {
1154 (Some((_, reason)), Some(_)) => lookup::Ran::Refused(reason.clone()),
1156 _ => lookup::Ran::Unkeyed,
1157 },
1158 };
1159 let recorded = verdict(had_sets(&recordings), lookup::Taken::Ran(ran));
1160 let shadow = match shadow {
1161 Some(Shadow::WouldHit { key: recorded, .. }) => Some(Shadow::WouldHit {
1162 key: recorded,
1163 check,
1164 }),
1165 shadow => shadow,
1166 };
1167 Ok(TaskResult {
1168 outcome,
1169 cache: CacheStatus::Miss,
1170 key: keyed,
1171 warm: warm_report,
1172 execution,
1173 recorded,
1174 shadow,
1175 })
1176 }
1177
1178 fn check(
1182 &self,
1183 task: &str,
1184 ran: &str,
1185 recorded: &str,
1186 settings: Option<&ultracache::Settings>,
1187 ) -> Option<Check> {
1188 let compared = self
1189 .artifacts(ran, task)
1190 .and_then(|ran| Ok(ran.zip(self.artifacts(recorded, task)?)))
1191 .and_then(|compared| {
1192 let unchecked = settings.map(ultracache::Settings::unchecked).transpose()?;
1193 Ok(compared.map(|compared| (compared, unchecked)))
1194 });
1195 match compared {
1196 Ok(Some(((mut ran, mut recorded), unchecked))) => {
1197 let mut left_out = 0;
1198 if let Some(unchecked) = unchecked {
1199 left_out = ran
1200 .keys()
1201 .chain(recorded.keys())
1202 .filter(|path| unchecked(path))
1203 .collect::<BTreeSet<_>>()
1204 .len();
1205 ran.retain(|path, _| !unchecked(path));
1206 recorded.retain(|path, _| !unchecked(path));
1207 }
1208 let differences = store::differences(&ran, &recorded, UNCOVERED);
1209 Some(Check {
1210 matched: differences.is_empty(),
1211 differences,
1212 unchecked: left_out,
1213 })
1214 }
1215 Ok(None) => None,
1216 Err(error) => {
1217 qk_executor::status!(
1218 "qk: {task}: not compared with its recorded result ({error:#})"
1219 );
1220 None
1221 }
1222 }
1223 }
1224
1225 fn with_changes(
1229 &self,
1230 shadow: Option<Shadow>,
1231 recordings: Option<&(Vec<recorded::PathSet>, String, ultracache::Settings)>,
1232 (snapshot, workspace, graph, task, prepared, unlisted): (
1233 &hash::Snapshot,
1234 &Workspace,
1235 &TaskGraph,
1236 &Task,
1237 &PreparedTask,
1238 &recorded::Unlisted,
1239 ),
1240 cancelled: &AtomicBool,
1241 ) -> Option<Shadow> {
1242 let (Some(Shadow::WouldMiss { .. }), Some((sets, _, _))) = (&shadow, recordings) else {
1243 return shadow;
1244 };
1245 let Some(newest) = sets.first() else {
1246 return shadow;
1247 };
1248 let _explaining = profile::span(&task.id, "recorded_changes");
1249 let Ok(coverage) = hash::coverage(snapshot, workspace, graph, task, prepared, cancelled)
1250 else {
1251 return shadow;
1252 };
1253 let declared = self.covers_all(&coverage, snapshot, workspace, sets);
1254 let outputs = self
1255 .workspace_outputs
1256 .get_or_init(|| workspace_outputs(workspace));
1257 let output = |path: &str| outputs.matches(path);
1258 let uncovered: BTreeMap<&str, &str> = coverage
1259 .uncovered(snapshot, &workspace.root, newest, &output)
1260 .collect();
1261 let named = |paths: Vec<String>| {
1262 let more = paths.len().saturating_sub(UNCOVERED - 1);
1263 let mut paths = paths;
1264 if more > 1 {
1265 paths.truncate(UNCOVERED - 1);
1266 paths.push(format!("{more} more"));
1267 }
1268 paths
1269 };
1270 if newest.recorded.is_some() {
1271 let changed = newest
1272 .changed(snapshot, &workspace.root, unlisted)
1273 .unwrap_or_default()
1274 .into_iter()
1275 .map(|(path, how)| match uncovered.contains_key(path) {
1276 true => format!("{path} ({how})"),
1277 false => format!("{path} ({how}, declared)"),
1278 })
1279 .collect();
1280 return Some(Shadow::WouldMiss {
1281 changed: named(changed),
1282 uncovered: vec![],
1283 declared,
1284 });
1285 }
1286 let uncovered = uncovered
1287 .into_iter()
1288 .map(|(path, how)| format!("{path} ({how})"))
1289 .collect();
1290 Some(Shadow::WouldMiss {
1291 changed: vec![],
1292 uncovered: named(uncovered),
1293 declared,
1294 })
1295 }
1296
1297 fn covers_all(
1301 &self,
1302 coverage: &hash::Coverage,
1303 snapshot: &hash::Snapshot,
1304 workspace: &Workspace,
1305 sets: &[recorded::PathSet],
1306 ) -> bool {
1307 let outputs = self
1308 .workspace_outputs
1309 .get_or_init(|| workspace_outputs(workspace));
1310 let output = |path: &str| outputs.matches(path);
1311 sets.iter()
1312 .all(|set| coverage.covers(snapshot, &workspace.root, set, &output))
1313 }
1314
1315 #[allow(clippy::too_many_arguments)]
1320 fn by_recordings(
1321 &self,
1322 workspace: &Workspace,
1323 task: &Task,
1324 prepared: &PreparedTask,
1325 snapshot: &hash::Snapshot,
1326 outputs: &paths::Outputs,
1327 unlisted: &recorded::Unlisted,
1328 sets: &mut Vec<recorded::PathSet>,
1329 weak: &str,
1330 declared: &str,
1331 cancelled: &AtomicBool,
1332 ) -> lookup::Found {
1333 let strong =
1334 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1335 let results = Restoring {
1336 cache: self,
1337 workspace,
1338 task,
1339 prepared,
1340 outputs,
1341 cancelled,
1342 };
1343 let search = match lookup::Search::here(sets, strong, &results) {
1344 Ok(found) => return found,
1345 Err(search) => search,
1346 };
1347 let Some(remote) = self.remote.as_ref().filter(|_| {
1348 lookup::asks_store(lookup::Lookup::Restore, sets, || self.holds_entry(declared))
1349 }) else {
1350 return lookup::Found::Nothing;
1351 };
1352 let added = self.merged(
1353 remote.fetch_path_sets(&self.root, weak, &task.id),
1354 weak,
1355 task,
1356 );
1357 search.there(added, sets, strong, &results)
1358 }
1359
1360 #[allow(clippy::too_many_arguments)]
1365 fn start_shadow(
1366 &self,
1367 workspace: &Workspace,
1368 task: &Task,
1369 snapshot: &hash::Snapshot,
1370 outputs: &paths::Outputs,
1371 unlisted: &recorded::Unlisted,
1372 sets: &[recorded::PathSet],
1373 weak: &str,
1374 declared: &str,
1375 ) -> Shadowing {
1376 let strong =
1377 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1378 let checking = Checking {
1379 cache: self,
1380 task,
1381 outputs,
1382 };
1383 let search = match lookup::Search::here(sets, strong, &checking) {
1384 Ok(found) => return Shadowing::Settled(lookup::shadow(found, sets)),
1385 Err(search) => search,
1386 };
1387 let Some(remote) = self.remote.as_ref().filter(|_| {
1388 lookup::asks_store(lookup::Lookup::Shadow, sets, || self.holds_entry(declared))
1389 }) else {
1390 return Shadowing::Settled(lookup::unfound(sets));
1391 };
1392 let (remote, root, weak, id) = (
1393 std::sync::Arc::clone(remote),
1394 self.root.clone(),
1395 weak.to_owned(),
1396 task.id.clone(),
1397 );
1398 Shadowing::Fetching {
1399 search,
1400 fetch: std::thread::spawn(move || remote.fetch_path_sets(&root, &weak, &id)),
1401 }
1402 }
1403
1404 #[allow(clippy::too_many_arguments)]
1408 fn settle_shadow(
1409 &self,
1410 shadowing: Shadowing,
1411 workspace: &Workspace,
1412 task: &Task,
1413 snapshot: &hash::Snapshot,
1414 outputs: &paths::Outputs,
1415 unlisted: &recorded::Unlisted,
1416 sets: &mut Vec<recorded::PathSet>,
1417 weak: &str,
1418 ) -> Shadow {
1419 let (search, fetch) = match shadowing {
1420 Shadowing::Settled(shadow) => return shadow,
1421 Shadowing::Fetching { search, fetch } => (search, fetch),
1422 };
1423 let fetched = fetch
1424 .join()
1425 .unwrap_or_else(|_| Err(anyhow::anyhow!("fetching them panicked")));
1426 let added = self.merged(fetched, weak, task);
1427 let strong =
1428 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1429 let checking = Checking {
1430 cache: self,
1431 task,
1432 outputs,
1433 };
1434 let found = search.there(added, sets, strong, &checking);
1435 lookup::shadow(found, sets)
1436 }
1437
1438 fn merged(
1441 &self,
1442 fetched: Result<Option<Vec<u8>>>,
1443 weak: &str,
1444 task: &Task,
1445 ) -> Vec<recorded::PathSet> {
1446 fetched
1447 .and_then(|record| match record {
1448 Some(record) => recorded::merge(&self.root, weak, &record),
1449 None => Ok(Vec::new()),
1450 })
1451 .unwrap_or_else(|error| {
1452 qk_executor::status!("qk: {}: remote path sets unavailable ({error:#})", task.id);
1453 Vec::new()
1454 })
1455 }
1456
1457 fn holds_entry(&self, key: &str) -> bool {
1459 self.root
1460 .join("entries")
1461 .join(format!("{key}.json"))
1462 .is_file()
1463 }
1464
1465 fn path_set(
1467 &self,
1468 workspace: &Workspace,
1469 task: &Task,
1470 recording: &qk_executor::Recording,
1471 prepared: &PreparedTask,
1472 settings: &ultracache::Settings,
1473 own: &[&paths::Outputs],
1474 ) -> std::result::Result<recorded::PathSet, String> {
1475 let failed = |error: anyhow::Error| format!("{error:#}");
1476 let canonical = workspace
1477 .root
1478 .canonicalize()
1479 .map_err(|e| failed(e.into()))?;
1480 let keep = |path: &std::path::Path| {
1483 [&workspace.root, &canonical].iter().any(|root| {
1484 path.strip_prefix(root).is_ok_and(|relative| {
1485 !relative.components().any(|component| {
1486 matches!(
1487 component.as_os_str().to_str(),
1488 Some("node_modules" | ".git")
1489 )
1490 })
1491 })
1492 })
1493 };
1494 let limits = qk_input_analysis::strace::Limits::kept(&keep, false);
1497 let accesses = recording.whole_accesses(&prepared.cwd, &limits)?;
1498 let ignore = self
1499 .ignored_reads(workspace, task, prepared, &settings.ignore_reads)
1500 .map_err(failed)?;
1501 let state = [self.root.clone(), worktree_state(&workspace.root)]
1503 .iter()
1504 .filter_map(|path| {
1505 [&workspace.root, &canonical]
1506 .iter()
1507 .find_map(|root| path.strip_prefix(root).ok())
1508 .and_then(|relative| relative.to_str())
1509 .map(|relative| relative.replace('\\', "/"))
1510 })
1511 .collect();
1512 let exclusions = recorded::Exclusions {
1513 roots: [&workspace.root, &canonical],
1514 outputs: own,
1515 ignore,
1516 state,
1517 };
1518 Ok(recorded::PathSet::from_accesses(
1519 &accesses,
1520 &exclusions,
1521 &canonical,
1522 ))
1523 }
1524
1525 #[allow(clippy::too_many_arguments)]
1527 fn inputs_match(
1528 &self,
1529 snapshot: &hash::Snapshot,
1530 workspace: &Workspace,
1531 graph: &TaskGraph,
1532 task: &Task,
1533 prepared: &PreparedTask,
1534 before: &str,
1535 cancelled: &AtomicBool,
1536 ) -> bool {
1537 if cancelled.load(Ordering::SeqCst) {
1538 return false;
1539 }
1540 let _profile = profile::span(&task.id, "inputs_recheck");
1541 let after = hash::recheck_dependency_keys(snapshot, workspace, graph, task, cancelled)
1542 .and_then(|dependencies| {
1543 hash::fingerprint(
1544 snapshot,
1545 workspace,
1546 task,
1547 prepared,
1548 &dependencies,
1549 cancelled,
1550 )
1551 });
1552 if after.as_deref().ok() != Some(before) {
1553 if task.definition.cache == Some(true) {
1554 if self.reuse {
1555 qk_executor::status!(
1556 "qk: {}: not caching because inputs changed during execution",
1557 task.id
1558 );
1559 } else {
1560 qk_executor::status!(
1561 "qk: {}: excluding execution from flaky results because inputs changed during execution",
1562 task.id
1563 );
1564 }
1565 }
1566 return false;
1567 }
1568 true
1569 }
1570}
1571
1572pub type Keyed = std::result::Result<(String, Value), String>;
1574
1575pub fn keys(
1579 workspace: &Workspace,
1580 graph: &TaskGraph,
1581 prepared: &BTreeMap<String, PreparedTask>,
1582 cancelled: &AtomicBool,
1583) -> Result<BTreeMap<String, Keyed>> {
1584 let snapshot = hash::Snapshot::new(workspace, graph, &paths::cache_location(workspace))?;
1585 Ok(graph
1586 .tasks
1587 .values()
1588 .map(|task| {
1589 let keyed = key_now(&snapshot, workspace, graph, task, prepared, cancelled);
1590 (task.id.clone(), keyed)
1591 })
1592 .collect())
1593}
1594
1595fn key_now(
1596 snapshot: &hash::Snapshot,
1597 workspace: &Workspace,
1598 graph: &TaskGraph,
1599 task: &Task,
1600 prepared: &BTreeMap<String, PreparedTask>,
1601 cancelled: &AtomicBool,
1602) -> Keyed {
1603 let failed = |error: anyhow::Error| format!("{}: {error:#}", task.id);
1604 let dependencies =
1605 hash::dependency_keys(snapshot, workspace, graph, task, cancelled).map_err(failed)?;
1606 let inputs = hash::inputs(
1607 snapshot,
1608 workspace,
1609 task,
1610 &prepared[&task.id],
1611 &dependencies,
1612 cancelled,
1613 )
1614 .map_err(failed)?;
1615 Ok((hash::key(&inputs).map_err(failed)?, inputs))
1616}
1617
1618#[doc(hidden)]
1620pub fn run_upload_worker() -> anyhow::Result<()> {
1621 remote::background::run()
1622}
1623
1624fn workspace_outputs(workspace: &Workspace) -> recorded::WorkspaceOutputs {
1630 let mut outputs = Vec::new();
1631 for (name, project) in &workspace.projects {
1632 for (target, definition) in &project.targets {
1633 if let Some(scratch) = scratch_outputs(workspace, name, definition) {
1634 let configurations = std::iter::once(None)
1635 .chain(definition.configurations.keys().cloned().map(Some));
1636 for configuration in configurations {
1637 outputs.push((
1638 (name.clone(), target.clone(), configuration),
1639 scratch.clone(),
1640 ));
1641 }
1642 }
1643 let configured = definition
1644 .configurations
1645 .iter()
1646 .map(|(configuration, options)| {
1647 let mut definition = definition.clone();
1648 definition.options.extend(options.clone());
1649 (Some(configuration.clone()), definition)
1650 });
1651 for (configuration, definition) in
1652 std::iter::once((None, definition.clone())).chain(configured)
1653 {
1654 let task = Task {
1655 id: format!("{name}:{target}"),
1656 project: name.clone(),
1657 target: target.clone(),
1658 configuration: configuration.clone(),
1659 args: Vec::new(),
1660 definition,
1661 dependencies: Default::default(),
1662 edges: Default::default(),
1663 };
1664 if let Some(declared) = paths::Outputs::new(workspace, &task)
1665 .ok()
1666 .filter(paths::Outputs::declared)
1667 {
1668 outputs.push(((name.clone(), target.clone(), configuration), declared));
1669 }
1670 }
1671 }
1672 }
1673 let roots = workspace
1674 .projects
1675 .iter()
1676 .map(|(name, project)| (name.clone(), project.root.clone()))
1677 .collect();
1678 recorded::WorkspaceOutputs::new(outputs, &roots)
1679}
1680
1681fn scratch_outputs(
1684 workspace: &Workspace,
1685 project: &str,
1686 definition: &qk_config::Target,
1687) -> Option<paths::Outputs> {
1688 let directories =
1689 scratch::directories_of(workspace, project, definition.extra.get("qk:sandbox")).ok()?;
1690 let patterns: Vec<String> = directories
1691 .iter()
1692 .filter_map(|directory| {
1693 let relative = directory.strip_prefix(&workspace.root).ok()?.to_str()?;
1694 Some(glob::escape(&relative.replace('\\', "/")))
1695 })
1696 .collect();
1697 if patterns.is_empty() {
1698 return None;
1699 }
1700 paths::Outputs::from_paths(&patterns).ok()
1701}
1702
1703fn upstream(graph: &TaskGraph, task: &Task) -> BTreeSet<recorded::TargetOf> {
1707 let of = |task: &Task| {
1708 (
1709 task.project.clone(),
1710 task.target.clone(),
1711 task.configuration.clone(),
1712 )
1713 };
1714 let mut targets = BTreeSet::new();
1715 let mut seen = BTreeSet::new();
1716 let mut pending: Vec<&String> = task.dependencies.iter().collect();
1717 while let Some(id) = pending.pop() {
1718 if !seen.insert(id) {
1719 continue;
1720 }
1721 if let Some(dependency) = graph.tasks.get(id) {
1722 targets.insert(of(dependency));
1723 pending.extend(&dependency.dependencies);
1724 }
1725 }
1726 targets
1727}
1728
1729#[cfg(test)]
1730mod tests {
1731 use super::*;
1732
1733 #[test]
1734 fn concurrent_blob_writes_keep_results_ordered_and_repair_corruption() {
1735 let workspace = tempfile::tempdir().unwrap();
1736 let root = workspace.path();
1737 let cache = Cache::new(root.join("cache"));
1738 cache.initialize().unwrap();
1739 let mut sources = Vec::new();
1740 for index in 0..130 {
1741 let source = root.join(format!("source-{index}"));
1742 std::fs::write(&source, format!("content-{}", index % 65)).unwrap();
1743 sources.push(source);
1744 }
1745 let digest = cache.put_blob(&sources[0]).unwrap();
1746 std::fs::write(cache.root.join("blobs").join(&digest), "corrupt").unwrap();
1747 std::fs::remove_file(&sources[31]).unwrap();
1748 let results = cache.put_blobs(&sources);
1749 assert_eq!(results.len(), sources.len());
1750 for (index, result) in results.into_iter().enumerate() {
1751 if index == 31 {
1752 assert_eq!(
1753 result
1754 .unwrap_err()
1755 .downcast_ref::<std::io::Error>()
1756 .unwrap()
1757 .kind(),
1758 std::io::ErrorKind::NotFound
1759 );
1760 continue;
1761 }
1762 let digest = result.unwrap();
1763 let held = cache.root.join("blobs").join(&digest);
1764 assert_eq!(
1765 std::fs::read(&held).unwrap(),
1766 format!("content-{}", index % 65).as_bytes()
1767 );
1768 assert_eq!(hash::digest_file(&held).unwrap(), digest);
1769 }
1770 }
1771
1772 #[cfg(unix)]
1774 #[test]
1775 fn a_blob_the_store_holds_is_not_written_again() {
1776 use std::os::unix::fs::MetadataExt;
1777 let workspace = tempfile::tempdir().unwrap();
1778 let root = workspace.path();
1779 let cache = Cache::new(root.join(".qk/cache"));
1780 cache.initialize().unwrap();
1781 std::fs::write(root.join("first"), "same").unwrap();
1782 std::fs::write(root.join("second"), "same").unwrap();
1783 std::fs::write(root.join("other"), "different").unwrap();
1784 let first = cache.put_blob(&root.join("first")).unwrap();
1785 let blob = cache.root.join("blobs").join(&first);
1786 let stored = std::fs::metadata(&blob).unwrap().ino();
1787 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1788 assert_eq!(std::fs::metadata(&blob).unwrap().ino(), stored);
1789 std::fs::write(&blob, "corrupt").unwrap();
1791 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1792 assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1793 {
1795 use std::os::unix::fs::PermissionsExt;
1796 std::fs::set_permissions(&blob, std::fs::Permissions::from_mode(0o000)).unwrap();
1797 }
1798 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1799 assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1800 let other = cache.put_blob(&root.join("other")).unwrap();
1801 assert_ne!(other, first);
1802 assert_eq!(
1803 std::fs::read(cache.root.join("blobs").join(other)).unwrap(),
1804 b"different"
1805 );
1806 }
1807
1808 #[test]
1811 fn complete_directory_restores_preserve_outputs() {
1812 let workspace = tempfile::tempdir().unwrap();
1813 let root = workspace.path();
1814 std::fs::create_dir_all(root.join("dist/nested/empty")).unwrap();
1815 std::fs::write(root.join("dist/nested/a.txt"), "a").unwrap();
1816 std::fs::write(root.join("dist/b.bin"), [0, 1, 2]).unwrap();
1817 std::fs::create_dir_all(root.join("bundle/empty")).unwrap();
1818 std::fs::write(root.join("bundle/readonly"), "readonly").unwrap();
1819 #[cfg(unix)]
1820 {
1821 use std::os::unix::fs::PermissionsExt;
1822 let script = root.join("dist/run.sh");
1823 std::fs::write(&script, "#!/bin/sh\n").unwrap();
1824 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1825 std::os::unix::fs::symlink("nested/a.txt", root.join("dist/link")).unwrap();
1826 std::fs::set_permissions(
1827 root.join("dist/nested"),
1828 std::fs::Permissions::from_mode(0o500),
1829 )
1830 .unwrap();
1831 std::fs::set_permissions(
1832 root.join("bundle/readonly"),
1833 std::fs::Permissions::from_mode(0o444),
1834 )
1835 .unwrap();
1836 }
1837 let log = root.join("log");
1838 std::fs::write(&log, "").unwrap();
1839
1840 let cache = Cache::new(root.join(".qk/cache"));
1841 cache.initialize().unwrap();
1842 let outputs = paths::Outputs::from_patterns(&["dist", "bundle"]);
1843 let key = "0".repeat(64);
1844 cache.publish(root, &key, &outputs, &log).unwrap();
1845
1846 #[cfg(unix)]
1847 {
1848 use std::os::unix::fs::PermissionsExt;
1849 std::fs::set_permissions(
1852 root.join("dist/nested"),
1853 std::fs::Permissions::from_mode(0o755),
1854 )
1855 .unwrap();
1856 }
1857 std::fs::remove_dir_all(root.join("dist")).unwrap();
1858 std::fs::remove_dir_all(root.join("bundle")).unwrap();
1859 let before = std::time::SystemTime::now();
1860 let restored = cache
1861 .restore(
1862 root,
1863 "app:build",
1864 &Default::default(),
1865 &key,
1866 &outputs,
1867 &qk_executor::Display::Stream,
1868 qk_executor::Shown::LocalCache,
1869 )
1870 .unwrap();
1871 assert!(restored);
1872 assert!(root.join("dist/nested/empty").is_dir());
1873 assert!(root.join("bundle/empty").is_dir());
1874 assert_eq!(
1875 std::fs::read_to_string(root.join("bundle/readonly")).unwrap(),
1876 "readonly"
1877 );
1878 #[cfg(unix)]
1879 {
1880 assert_eq!(
1881 std::fs::read_link(root.join("dist/link")).unwrap(),
1882 std::path::Path::new("nested/a.txt")
1883 );
1884 assert_eq!(
1885 store::mode(&std::fs::metadata(root.join("dist/run.sh")).unwrap()),
1886 0o755
1887 );
1888 assert_eq!(
1889 store::mode(&std::fs::metadata(root.join("dist/nested")).unwrap()),
1890 0o500
1891 );
1892 assert_eq!(
1893 store::mode(&std::fs::metadata(root.join("bundle/readonly")).unwrap()),
1894 0o444
1895 );
1896 }
1897 let path = root.join("dist/nested/a.txt");
1898 let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
1899 assert!(modified >= before);
1900 let kept = cache
1901 .restore(
1902 root,
1903 "app:build",
1904 &Default::default(),
1905 &key,
1906 &outputs,
1907 &qk_executor::Display::Hidden,
1908 qk_executor::Shown::LocalCache,
1909 )
1910 .unwrap();
1911 assert!(kept);
1912 assert_eq!(
1913 std::fs::metadata(&path).unwrap().modified().unwrap(),
1914 modified
1915 );
1916 #[cfg(unix)]
1917 {
1918 use std::os::unix::fs::PermissionsExt;
1919 std::fs::set_permissions(
1920 root.join("dist/nested"),
1921 std::fs::Permissions::from_mode(0o755),
1922 )
1923 .unwrap();
1924 }
1925 }
1926
1927 #[test]
1928 fn listings_leave_out_other_targets_outputs_but_the_upstreams() {
1929 let temp = tempfile::tempdir().unwrap();
1930 let root = temp.path();
1931 std::fs::write(root.join("nx.json"), "{}").unwrap();
1932 std::fs::create_dir_all(root.join("app")).unwrap();
1933 std::fs::create_dir_all(root.join("lib")).unwrap();
1934 std::fs::write(
1937 root.join("project.json"),
1938 r#"{"name":"workspace","targets":{"cache":{"command":"true",
1939 "outputs":["{projectRoot}/.cache/build","{projectRoot}/app/extra/out"]}}}"#,
1940 )
1941 .unwrap();
1942 std::fs::write(
1943 root.join("app/project.json"),
1944 r#"{"name":"app","root":"./app","targets":{
1945 "build":{"command":"true","outputs":["{options.outputPath}"],
1946 "options":{"outputPath":"app/dist"},
1947 "configurations":{"production":{"outputPath":"app/dist-production"}}},
1948 "lint":{"command":"true"},
1949 "browser":{"command":"true","qk:sandbox":{"scratch":["{projectRoot}/.vitest"]}},
1950 "cover":{"command":"true","outputs":["{projectRoot}/.coverage/unit"]},
1951 "cover-browser":{"command":"true","outputs":["{projectRoot}/.coverage/browser"]},
1952 "wasm":{"command":"true","outputs":["{projectRoot}/target/wasm/release/*.wasm"]},
1953 "test":{"command":"true","dependsOn":[{"projects":["lib"],"target":"build"}],
1954 "configurations":{"production":{}}}}}"#,
1955 )
1956 .unwrap();
1957 std::fs::write(
1958 root.join("lib/project.json"),
1959 r#"{"name":"lib","targets":{"build":{"command":"true","outputs":["{options.outputPath}"],
1960 "options":{"outputPath":"lib/dist"},
1961 "qk:sandbox":{"scratch":["{projectRoot}/.scratch"]},
1962 "configurations":{"production":{"outputPath":"lib/dist-production"}}}}}"#,
1963 )
1964 .unwrap();
1965 let workspace = Workspace::load(root).unwrap();
1966 let graph = qk_taskgraph::TaskGraph::build(
1967 &workspace,
1968 &[
1969 qk_taskgraph::Request::parse("app:lint").unwrap(),
1970 qk_taskgraph::Request::parse("app:cover").unwrap(),
1971 qk_taskgraph::Request::parse("app:test").unwrap(),
1972 qk_taskgraph::Request::parse("app:test:production").unwrap(),
1973 ],
1974 )
1975 .unwrap();
1976 let outputs = workspace_outputs(&workspace);
1977 let others = |task: &str| recorded::Others {
1978 outputs: &outputs,
1979 upstream: upstream(&graph, &graph.tasks[task]),
1980 };
1981 let lint = others("app:lint");
1982 assert!(lint.matches("app/dist/main.js"));
1983 assert!(lint.matches("app/dist-production/main.js"));
1984 assert!(lint.matches("lib/dist"));
1985 assert!(!lint.matches("app/src/main.ts"));
1986 assert!(lint.matches("app/.vitest"));
1987 assert!(lint.matches("app/.vitest/attachments/diff.png"));
1988 assert!(lint.matches("app/target"));
1990 assert!(lint.matches("app/target/wasm"));
1991 assert!(!lint.matches("app"));
1992 assert!(lint.matches(".cache"));
1994 assert!(lint.matches("app/extra"));
1995 assert!(!lint.matches("lib"));
1996 let unlisted = recorded::Unlisted {
1999 own: Vec::new(),
2000 others: Some(lint),
2001 ignore: &ultracache::IgnoredReads::none(),
2002 };
2003 let directories = unlisted.directories("lib");
2004 for directory in [".", ".cache", "app", "app/extra"] {
2005 assert!(directories.contains(directory), "{directory}");
2006 }
2007 assert!(!others("app:test").matches("lib/dist"));
2010 assert!(others("app:test").matches("lib/dist-production"));
2011 assert!(others("app:cover").matches("app/.coverage"));
2014 let production = others("app:test:production");
2015 assert!(!production.matches("lib/dist-production"));
2016 assert!(production.matches("lib/dist"));
2017 assert!(others("app:lint").matches("lib/.scratch"));
2020 assert!(!others("app:test").matches("lib/.scratch"));
2021 assert!(!production.matches("lib/.scratch"));
2022 }
2023}