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