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(|_| {
743 recorded::Others::new(
744 self.workspace_outputs
745 .get_or_init(|| workspace_outputs(workspace)),
746 upstream(graph, task),
747 &own,
748 )
749 });
750 let ignore = settings
751 .as_ref()
752 .and_then(|settings| {
753 self.ignored_reads(workspace, task, prepared, &settings.ignore_reads)
754 .ok()
755 })
756 .unwrap_or_default();
757 let produced = settings
759 .as_ref()
760 .map(|_| produced(snapshot, workspace, graph, task));
761 let unlisted = recorded::Unlisted {
762 own: own.clone(),
763 others,
764 ignore: &ignore,
765 produced: produced.as_ref(),
766 };
767 let inputs = &keyed.as_ref().expect("a cacheable task is keyed").1;
768 let mut unkeyable = None;
770 let mut recordings = settings.and_then(|settings| {
771 let weak = hash::weak_key(
772 snapshot,
773 workspace,
774 graph,
775 task,
776 prepared,
777 inputs,
778 &settings.ignore_reads,
779 &ignore.scope(),
780 produced.as_ref().and_then(recorded::Produced::weak),
781 cancelled,
782 );
783 match weak {
784 Ok(weak) => Some((recorded::load(&self.root, &weak), weak, settings)),
785 Err(error) => {
786 qk_executor::status!(
787 "qk: {}: not keyed by its recordings ({error:#})",
788 task.id
789 );
790 unkeyable = Some((settings.mode, format!("{error:#}")));
791 None
792 }
793 }
794 });
795 let mut shadowing = None;
796 let mut replaced = Vec::new();
799 if let Some((sets, weak, settings)) = &mut recordings {
800 match lookup::lookup(settings.mode, self.cold) {
801 lookup::Lookup::Replace => {
803 replaced = sets
804 .iter()
805 .filter_map(|set| {
806 recording_key(weak, set, snapshot, workspace, &unlisted, task)
807 })
808 .collect();
809 }
810 lookup::Lookup::Shadow => {
811 let _lookup = profile::span(&task.id, "recorded_lookup");
812 shadowing = Some(self.start_shadow(
813 workspace, task, snapshot, &outputs, &unlisted, sets, weak, &key,
814 ));
815 }
816 lookup::Lookup::Restore => {
817 let recorded_profile = profile::span(&task.id, "recorded_restore");
818 let found = self.by_recordings(
819 workspace, task, prepared, snapshot, &outputs, &unlisted, sets, weak, &key,
820 cancelled,
821 );
822 drop(recorded_profile);
823 let remote = matches!(found, lookup::Found::Remote(_));
824 if let lookup::Found::Local(strong) | lookup::Found::Remote(strong) = found {
825 if let Some(store) = self.remote.as_ref().filter(|_| !remote) {
826 store.held(&strong);
827 }
828 recorded::touch(&self.root, weak);
829 qk_executor::report::event(qk_executor::report::Event::Cache {
830 id: task.id.clone(),
831 cached: if remote {
832 qk_executor::report::Cached::RemoteHit
833 } else {
834 qk_executor::report::Cached::Hit
835 },
836 });
837 return Ok(TaskResult {
838 outcome: Outcome::Success,
839 cache: if remote {
840 CacheStatus::RemoteHit
841 } else {
842 CacheStatus::LocalHit
843 },
844 key: keyed,
845 warm: None,
846 execution: None,
847 recorded: lookup::recorded(
848 Some(settings.mode),
849 !sets.is_empty(),
850 lookup::Taken::Recording(strong),
851 ),
852 shadow: None,
853 });
854 }
855 }
856 }
857 }
858 let mut settle = |recordings: &mut Option<(Vec<recorded::PathSet>, String, _)>| {
860 let shadowing = shadowing.take()?;
861 let (sets, weak, _) = recordings.as_mut()?;
862 let _wait = profile::span(&task.id, "recorded_wait");
863 Some(self.settle_shadow(
864 shadowing, workspace, task, snapshot, &outputs, &unlisted, sets, weak,
865 ))
866 };
867 let mode = recordings.as_ref().map(|(_, _, settings)| settings.mode);
870 let had_sets = |recordings: &Option<(Vec<recorded::PathSet>, String, _)>| {
871 recordings
872 .as_ref()
873 .is_some_and(|(sets, _, _)| !sets.is_empty())
874 };
875 let declared = lookup::by_declared(
876 mode,
877 self.cold,
878 recordings.as_ref().map_or(&[], |(sets, _, _)| sets),
879 |sets| {
880 hash::coverage(snapshot, workspace, graph, task, prepared, cancelled)
881 .is_ok_and(|coverage| self.covers_all(&coverage, snapshot, workspace, sets))
882 },
883 );
884 let verdict = |had_sets: bool, taken: lookup::Taken| match &unkeyable {
886 Some((mode, reason)) => lookup::unkeyable(*mode, reason, taken),
887 None => lookup::recorded(mode, had_sets, taken),
888 };
889 let unrecorded = verdict(had_sets(&recordings), lookup::Taken::Declared(key.clone()));
890 let prefetched = declared
893 && self.remote.as_ref().is_some_and(|remote| {
894 remote.resolved(&key);
895 remote.prefetched(&key)
896 });
897 let local_profile = profile::span(&task.id, "local_restore");
898 let restored = if declared && !self.cold {
899 self.restore(
900 &workspace.root,
901 &task.id,
902 &task.dependencies,
903 &key,
904 &outputs,
905 &prepared.display,
906 if prefetched {
907 qk_executor::Shown::RemoteCache
908 } else {
909 qk_executor::Shown::LocalCache
910 },
911 )
912 } else {
913 Ok(false)
914 };
915 match restored {
916 Ok(true) => {
917 let claimed = match &self.remote {
918 Some(remote) if prefetched && remote.claim(&key) => true,
919 Some(remote) => {
920 remote.held(&key);
921 false
922 }
923 None => false,
924 };
925 qk_executor::report::event(qk_executor::report::Event::Cache {
926 id: task.id.clone(),
927 cached: if claimed {
928 qk_executor::report::Cached::RemoteHit
929 } else {
930 qk_executor::report::Cached::Hit
931 },
932 });
933 let shadow = settle(&mut recordings);
934 let shadow = self.with_changes(
935 shadow,
936 recordings.as_ref(),
937 (snapshot, workspace, graph, task, prepared, &unlisted),
938 cancelled,
939 );
940 return Ok(TaskResult {
941 outcome: Outcome::Success,
942 cache: if claimed {
943 CacheStatus::RemoteHit
944 } else {
945 CacheStatus::LocalHit
946 },
947 key: keyed,
948 warm: None,
949 execution: None,
950 recorded: unrecorded,
951 shadow,
952 });
953 }
954 Ok(false) => {}
955 Err(error) => {
956 qk_executor::status!("qk: {}: ignoring unusable cache entry ({error})", task.id)
957 }
958 }
959 drop(local_profile);
960 if let Some(remote) = self.remote.as_ref().filter(|_| declared && !self.cold) {
961 let fetch_profile = profile::span(&task.id, "remote_fetch");
962 let fetched = remote.fetch(&self.root, &key, &task.id, cancelled);
963 drop(fetch_profile);
964 let fetched = fetched.and_then(|found| {
965 if !found {
966 return Ok(false);
967 }
968 self.restore(
969 &workspace.root,
970 &task.id,
971 &task.dependencies,
972 &key,
973 &outputs,
974 &prepared.display,
975 qk_executor::Shown::RemoteCache,
976 )
977 });
978 match fetched {
979 Ok(true) => {
980 qk_executor::report::event(qk_executor::report::Event::Cache {
981 id: task.id.clone(),
982 cached: qk_executor::report::Cached::RemoteHit,
983 });
984 let shadow = settle(&mut recordings);
985 let shadow = self.with_changes(
986 shadow,
987 recordings.as_ref(),
988 (snapshot, workspace, graph, task, prepared, &unlisted),
989 cancelled,
990 );
991 return Ok(TaskResult {
992 outcome: Outcome::Success,
993 cache: CacheStatus::RemoteHit,
994 key: keyed,
995 warm: None,
996 execution: None,
997 recorded: unrecorded,
998 shadow,
999 });
1000 }
1001 Ok(false) => {}
1002 Err(error) => {
1003 qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", task.id)
1004 }
1005 }
1006 }
1007 let shadow = settle(&mut recordings);
1009 qk_executor::report::event(qk_executor::report::Event::Cache {
1010 id: task.id.clone(),
1011 cached: qk_executor::report::Cached::Miss,
1012 });
1013 let log = match tempfile::NamedTempFile::new_in(self.root.join("tmp")) {
1014 Ok(log) => log,
1015 Err(error) => return bypass(format!("cache log unavailable: {error}")),
1016 };
1017 let capture =
1018 Capture::new(Some(log.as_file().try_clone()?), prepared.display.clone()).retain_log();
1019 let restored = before(running);
1020 let keying = prepared
1023 .recording
1024 .as_ref()
1025 .filter(|recording| recording.keys());
1026 let listed = match (&recordings, keying) {
1027 (Some(_), Some(_)) => {
1028 let project_root = workspace
1029 .projects
1030 .get(&task.project)
1031 .map_or("", |project| project.root.as_str());
1032 recorded::Before::take(
1033 &workspace.root,
1034 unlisted.directories(project_root),
1035 &unlisted,
1036 )
1037 }
1038 _ => recorded::Before::default(),
1039 };
1040 let started = recorded::now(&self.root.join("tmp"));
1041 let executing = profile::span(&task.id, "execute");
1042 let outcome = execute_captured(running, cancelled, Some(&capture))?;
1043 drop(executing);
1044 let warm_report = after(outcome, restored, running);
1045 let inputs_unchanged =
1046 self.inputs_match(snapshot, workspace, graph, task, prepared, &key, cancelled);
1047 let execution = capture.take_log().map(|log| Execution {
1048 log,
1049 inputs_unchanged,
1050 });
1051 let path_set = match (&recordings, keying) {
1052 (Some((_, _, settings)), Some(recording)) => Some(
1053 self.path_set(
1054 workspace,
1055 task,
1056 recording,
1057 prepared,
1058 settings,
1059 &own,
1060 produced.as_ref(),
1061 )
1062 .and_then(|set| {
1063 if set.is_empty() {
1065 return Err("its trace named no path in the workspace".to_owned());
1066 }
1067 if let Some(reason) = set.unkeyable(snapshot, &workspace.root) {
1069 return Err(reason);
1070 }
1071 match set
1072 .changed_since(&workspace.root, started, &unlisted, &listed)
1073 .map(|changed| changed.to_string())
1074 {
1075 Some(reason) => Err(reason),
1076 None => Ok(set),
1077 }
1078 }),
1079 ),
1080 _ => None,
1081 };
1082 if let Some(Err(reason)) = &path_set {
1083 qk_executor::status!("qk: {}: not recorded ({reason})", task.id);
1084 }
1085 let mut published = false;
1086 let kept = outcome == Outcome::Success && inputs_unchanged && capture.healthy();
1087 if kept {
1088 let publishing = profile::span(&task.id, "publish");
1089 let publish = self.publish(&workspace.root, &key, &outputs, log.path());
1090 drop(publishing);
1091 match publish {
1092 Ok(()) => {
1093 published = true;
1094 store::record_outputs(&workspace.root, &task.id, &key, &outputs);
1095 if let Some(remote) = &self.remote {
1096 remote.upload(&self.root, &key);
1097 }
1098 for strong in &replaced {
1099 if let Err(error) =
1100 self.publish(&workspace.root, strong, &outputs, log.path())
1101 {
1102 qk_executor::status!(
1103 "qk: {}: could not replace a recorded cache entry ({error})",
1104 task.id
1105 );
1106 }
1107 }
1108 }
1109 Err(error) => {
1110 qk_executor::status!("qk: {}: could not save cache entry ({error})", task.id);
1111 }
1112 }
1113 }
1114 let check = match &shadow {
1118 Some(Shadow::WouldHit { key: recorded, .. }) => match outcome {
1119 Outcome::Failed(code) => Some(Check {
1120 matched: false,
1121 differences: vec![format!("failed (exit {code})")],
1122 unchecked: 0,
1123 }),
1124 Outcome::Success if published => {
1125 let _checking = profile::span(&task.id, "recorded_check");
1126 let settings = recordings.as_ref().map(|(_, _, settings)| settings);
1127 self.check(&task.id, &key, recorded, settings)
1128 }
1129 _ => None,
1130 },
1131 _ => None,
1132 };
1133 let ran = match (path_set, &recordings) {
1134 (Some(Err(reason)), _) => lookup::Ran::Refused(reason),
1135 (Some(Ok(set)), Some((_, weak, _))) if kept => {
1137 let saved = recorded::with_values(weak, set, snapshot, &workspace.root, &unlisted)
1138 .and_then(|(strong, set)| {
1139 self.publish(&workspace.root, &strong, &outputs, log.path())?;
1140 recorded::save(&self.root, weak, set)?;
1141 if let Some(remote) = &self.remote {
1144 remote.upload(&self.root, &strong);
1145 if let Some(record) = recorded::record(&self.root, weak) {
1146 remote.upload_path_sets(&self.root, weak, record);
1147 }
1148 }
1149 Ok(strong)
1150 });
1151 match saved {
1152 Ok(key) => lookup::Ran::Saved(key),
1153 Err(error) => {
1154 qk_executor::status!("qk: {}: recording not saved ({error:#})", task.id);
1155 lookup::Ran::Unsaved(format!("{error:#}"))
1156 }
1157 }
1158 }
1159 (Some(Ok(_)), _) => {
1160 let why = if outcome != Outcome::Success {
1162 None
1163 } else if !inputs_unchanged {
1164 Some("its inputs changed while it ran")
1165 } else {
1166 Some("its log could not be written")
1167 };
1168 if let Some(why) = why {
1169 qk_executor::status!("qk: {}: recording not saved ({why})", task.id);
1170 }
1171 lookup::Ran::Unkept
1172 }
1173 (None, _) => match (&unkeyable, keying) {
1174 (Some((_, reason)), Some(_)) => lookup::Ran::Refused(reason.clone()),
1176 _ => lookup::Ran::Unkeyed,
1177 },
1178 };
1179 let recorded = verdict(had_sets(&recordings), lookup::Taken::Ran(ran));
1180 let shadow = match shadow {
1181 Some(Shadow::WouldHit { key: recorded, .. }) => Some(Shadow::WouldHit {
1182 key: recorded,
1183 check,
1184 }),
1185 shadow => shadow,
1186 };
1187 Ok(TaskResult {
1188 outcome,
1189 cache: CacheStatus::Miss,
1190 key: keyed,
1191 warm: warm_report,
1192 execution,
1193 recorded,
1194 shadow,
1195 })
1196 }
1197
1198 fn check(
1202 &self,
1203 task: &str,
1204 ran: &str,
1205 recorded: &str,
1206 settings: Option<&ultracache::Settings>,
1207 ) -> Option<Check> {
1208 let compared = self
1209 .artifacts(ran, task)
1210 .and_then(|ran| Ok(ran.zip(self.artifacts(recorded, task)?)))
1211 .and_then(|compared| {
1212 let unchecked = settings.map(ultracache::Settings::unchecked).transpose()?;
1213 Ok(compared.map(|compared| (compared, unchecked)))
1214 });
1215 match compared {
1216 Ok(Some(((mut ran, mut recorded), unchecked))) => {
1217 let mut left_out = 0;
1218 if let Some(unchecked) = unchecked {
1219 left_out = ran
1220 .keys()
1221 .chain(recorded.keys())
1222 .filter(|path| unchecked(path))
1223 .collect::<BTreeSet<_>>()
1224 .len();
1225 ran.retain(|path, _| !unchecked(path));
1226 recorded.retain(|path, _| !unchecked(path));
1227 }
1228 let differences = store::differences(&ran, &recorded, UNCOVERED);
1229 Some(Check {
1230 matched: differences.is_empty(),
1231 differences,
1232 unchecked: left_out,
1233 })
1234 }
1235 Ok(None) => None,
1236 Err(error) => {
1237 qk_executor::status!(
1238 "qk: {task}: not compared with its recorded result ({error:#})"
1239 );
1240 None
1241 }
1242 }
1243 }
1244
1245 fn with_changes(
1249 &self,
1250 shadow: Option<Shadow>,
1251 recordings: Option<&(Vec<recorded::PathSet>, String, ultracache::Settings)>,
1252 (snapshot, workspace, graph, task, prepared, unlisted): (
1253 &hash::Snapshot,
1254 &Workspace,
1255 &TaskGraph,
1256 &Task,
1257 &PreparedTask,
1258 &recorded::Unlisted,
1259 ),
1260 cancelled: &AtomicBool,
1261 ) -> Option<Shadow> {
1262 let (Some(Shadow::WouldMiss { .. }), Some((sets, _, _))) = (&shadow, recordings) else {
1263 return shadow;
1264 };
1265 let Some(newest) = sets.first() else {
1266 return shadow;
1267 };
1268 let _explaining = profile::span(&task.id, "recorded_changes");
1269 let Ok(coverage) = hash::coverage(snapshot, workspace, graph, task, prepared, cancelled)
1270 else {
1271 return shadow;
1272 };
1273 let declared = self.covers_all(&coverage, snapshot, workspace, sets);
1274 let outputs = self
1275 .workspace_outputs
1276 .get_or_init(|| workspace_outputs(workspace));
1277 let output = |path: &str| outputs.matches(path);
1278 let uncovered: BTreeMap<&str, &str> = coverage
1279 .uncovered(snapshot, &workspace.root, newest, &output)
1280 .collect();
1281 let named = |paths: Vec<String>| {
1282 let more = paths.len().saturating_sub(UNCOVERED - 1);
1283 let mut paths = paths;
1284 if more > 1 {
1285 paths.truncate(UNCOVERED - 1);
1286 paths.push(format!("{more} more"));
1287 }
1288 paths
1289 };
1290 if newest.recorded.is_some() {
1291 let changed = newest
1292 .changed(snapshot, &workspace.root, unlisted)
1293 .unwrap_or_default()
1294 .into_iter()
1295 .map(|(path, how)| match uncovered.contains_key(path) {
1296 true => format!("{path} ({how})"),
1297 false => format!("{path} ({how}, declared)"),
1298 })
1299 .collect();
1300 return Some(Shadow::WouldMiss {
1301 changed: named(changed),
1302 uncovered: vec![],
1303 declared,
1304 });
1305 }
1306 let uncovered = uncovered
1307 .into_iter()
1308 .map(|(path, how)| format!("{path} ({how})"))
1309 .collect();
1310 Some(Shadow::WouldMiss {
1311 changed: vec![],
1312 uncovered: named(uncovered),
1313 declared,
1314 })
1315 }
1316
1317 fn covers_all(
1321 &self,
1322 coverage: &hash::Coverage,
1323 snapshot: &hash::Snapshot,
1324 workspace: &Workspace,
1325 sets: &[recorded::PathSet],
1326 ) -> bool {
1327 let outputs = self
1328 .workspace_outputs
1329 .get_or_init(|| workspace_outputs(workspace));
1330 let output = |path: &str| outputs.matches(path);
1331 sets.iter()
1332 .all(|set| coverage.covers(snapshot, &workspace.root, set, &output))
1333 }
1334
1335 #[allow(clippy::too_many_arguments)]
1340 fn by_recordings(
1341 &self,
1342 workspace: &Workspace,
1343 task: &Task,
1344 prepared: &PreparedTask,
1345 snapshot: &hash::Snapshot,
1346 outputs: &paths::Outputs,
1347 unlisted: &recorded::Unlisted,
1348 sets: &mut Vec<recorded::PathSet>,
1349 weak: &str,
1350 declared: &str,
1351 cancelled: &AtomicBool,
1352 ) -> lookup::Found {
1353 let strong =
1354 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1355 let results = Restoring {
1356 cache: self,
1357 workspace,
1358 task,
1359 prepared,
1360 outputs,
1361 cancelled,
1362 };
1363 let search = match lookup::Search::here(sets, strong, &results) {
1364 Ok(found) => return found,
1365 Err(search) => search,
1366 };
1367 let Some(remote) = self.remote.as_ref().filter(|_| {
1368 lookup::asks_store(lookup::Lookup::Restore, sets, || self.holds_entry(declared))
1369 }) else {
1370 return lookup::Found::Nothing;
1371 };
1372 let added = self.merged(
1373 remote.fetch_path_sets(&self.root, weak, &task.id),
1374 weak,
1375 task,
1376 );
1377 search.there(added, sets, strong, &results)
1378 }
1379
1380 #[allow(clippy::too_many_arguments)]
1385 fn start_shadow(
1386 &self,
1387 workspace: &Workspace,
1388 task: &Task,
1389 snapshot: &hash::Snapshot,
1390 outputs: &paths::Outputs,
1391 unlisted: &recorded::Unlisted,
1392 sets: &[recorded::PathSet],
1393 weak: &str,
1394 declared: &str,
1395 ) -> Shadowing {
1396 let strong =
1397 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1398 let checking = Checking {
1399 cache: self,
1400 task,
1401 outputs,
1402 };
1403 let search = match lookup::Search::here(sets, strong, &checking) {
1404 Ok(found) => return Shadowing::Settled(lookup::shadow(found, sets)),
1405 Err(search) => search,
1406 };
1407 let Some(remote) = self.remote.as_ref().filter(|_| {
1408 lookup::asks_store(lookup::Lookup::Shadow, sets, || self.holds_entry(declared))
1409 }) else {
1410 return Shadowing::Settled(lookup::unfound(sets));
1411 };
1412 let (remote, root, weak, id) = (
1413 std::sync::Arc::clone(remote),
1414 self.root.clone(),
1415 weak.to_owned(),
1416 task.id.clone(),
1417 );
1418 Shadowing::Fetching {
1419 search,
1420 fetch: std::thread::spawn(move || remote.fetch_path_sets(&root, &weak, &id)),
1421 }
1422 }
1423
1424 #[allow(clippy::too_many_arguments)]
1428 fn settle_shadow(
1429 &self,
1430 shadowing: Shadowing,
1431 workspace: &Workspace,
1432 task: &Task,
1433 snapshot: &hash::Snapshot,
1434 outputs: &paths::Outputs,
1435 unlisted: &recorded::Unlisted,
1436 sets: &mut Vec<recorded::PathSet>,
1437 weak: &str,
1438 ) -> Shadow {
1439 let (search, fetch) = match shadowing {
1440 Shadowing::Settled(shadow) => return shadow,
1441 Shadowing::Fetching { search, fetch } => (search, fetch),
1442 };
1443 let fetched = fetch
1444 .join()
1445 .unwrap_or_else(|_| Err(anyhow::anyhow!("fetching them panicked")));
1446 let added = self.merged(fetched, weak, task);
1447 let strong =
1448 |set: &recorded::PathSet| recording_key(weak, set, snapshot, workspace, unlisted, task);
1449 let checking = Checking {
1450 cache: self,
1451 task,
1452 outputs,
1453 };
1454 let found = search.there(added, sets, strong, &checking);
1455 lookup::shadow(found, sets)
1456 }
1457
1458 fn merged(
1461 &self,
1462 fetched: Result<Option<Vec<u8>>>,
1463 weak: &str,
1464 task: &Task,
1465 ) -> Vec<recorded::PathSet> {
1466 fetched
1467 .and_then(|record| match record {
1468 Some(record) => recorded::merge(&self.root, weak, &record),
1469 None => Ok(Vec::new()),
1470 })
1471 .unwrap_or_else(|error| {
1472 qk_executor::status!("qk: {}: remote path sets unavailable ({error:#})", task.id);
1473 Vec::new()
1474 })
1475 }
1476
1477 fn holds_entry(&self, key: &str) -> bool {
1479 self.root
1480 .join("entries")
1481 .join(format!("{key}.json"))
1482 .is_file()
1483 }
1484
1485 #[allow(clippy::too_many_arguments)]
1487 fn path_set(
1488 &self,
1489 workspace: &Workspace,
1490 task: &Task,
1491 recording: &qk_executor::Recording,
1492 prepared: &PreparedTask,
1493 settings: &ultracache::Settings,
1494 own: &[&paths::Outputs],
1495 produced: Option<&recorded::Produced>,
1496 ) -> std::result::Result<recorded::PathSet, String> {
1497 let nothing = recorded::Produced::default();
1498 let produced = produced.unwrap_or(¬hing);
1499 let failed = |error: anyhow::Error| format!("{error:#}");
1500 let canonical = workspace
1501 .root
1502 .canonicalize()
1503 .map_err(|e| failed(e.into()))?;
1504 let keep = |path: &std::path::Path, operation: qk_input_analysis::Operation| {
1507 [&workspace.root, &canonical].iter().any(|root| {
1508 path.strip_prefix(root).is_ok_and(|relative| {
1509 let mut installed = false;
1510 for component in relative.components() {
1511 match component.as_os_str().to_str() {
1512 Some(".git") => return false,
1513 Some("node_modules") => installed = true,
1514 _ => {}
1515 }
1516 }
1517 !installed
1518 || relative.to_str().is_some_and(|relative| {
1519 let relative = relative.replace('\\', "/");
1520 produced.traced(&relative, &operation)
1521 })
1522 })
1523 })
1524 };
1525 let limits = qk_input_analysis::strace::Limits::kept(&keep, false);
1528 let accesses = recording.whole_accesses(&prepared.cwd, &limits)?;
1529 let ignore = self
1530 .ignored_reads(workspace, task, prepared, &settings.ignore_reads)
1531 .map_err(failed)?;
1532 let state = [self.root.clone(), worktree_state(&workspace.root)]
1534 .iter()
1535 .filter_map(|path| {
1536 [&workspace.root, &canonical]
1537 .iter()
1538 .find_map(|root| path.strip_prefix(root).ok())
1539 .and_then(|relative| relative.to_str())
1540 .map(|relative| relative.replace('\\', "/"))
1541 })
1542 .collect();
1543 let exclusions = recorded::Exclusions {
1544 roots: [&workspace.root, &canonical],
1545 outputs: own,
1546 ignore,
1547 state,
1548 produced,
1549 };
1550 Ok(recorded::PathSet::from_accesses(
1551 &accesses,
1552 &exclusions,
1553 &canonical,
1554 ))
1555 }
1556
1557 #[allow(clippy::too_many_arguments)]
1559 fn inputs_match(
1560 &self,
1561 snapshot: &hash::Snapshot,
1562 workspace: &Workspace,
1563 graph: &TaskGraph,
1564 task: &Task,
1565 prepared: &PreparedTask,
1566 before: &str,
1567 cancelled: &AtomicBool,
1568 ) -> bool {
1569 if cancelled.load(Ordering::SeqCst) {
1570 return false;
1571 }
1572 let _profile = profile::span(&task.id, "inputs_recheck");
1573 let after = hash::recheck_dependency_keys(snapshot, workspace, graph, task, cancelled)
1574 .and_then(|dependencies| {
1575 hash::fingerprint(
1576 snapshot,
1577 workspace,
1578 task,
1579 prepared,
1580 &dependencies,
1581 cancelled,
1582 )
1583 });
1584 if after.as_deref().ok() != Some(before) {
1585 if task.definition.cache == Some(true) {
1586 if self.reuse {
1587 qk_executor::status!(
1588 "qk: {}: not caching because inputs changed during execution",
1589 task.id
1590 );
1591 } else {
1592 qk_executor::status!(
1593 "qk: {}: excluding execution from flaky results because inputs changed during execution",
1594 task.id
1595 );
1596 }
1597 }
1598 return false;
1599 }
1600 true
1601 }
1602}
1603
1604pub type Keyed = std::result::Result<(String, Value), String>;
1606
1607pub fn keys(
1611 workspace: &Workspace,
1612 graph: &TaskGraph,
1613 prepared: &BTreeMap<String, PreparedTask>,
1614 cancelled: &AtomicBool,
1615) -> Result<BTreeMap<String, Keyed>> {
1616 let snapshot = hash::Snapshot::new(workspace, graph, &paths::cache_location(workspace))?;
1617 Ok(graph
1618 .tasks
1619 .values()
1620 .map(|task| {
1621 let keyed = key_now(&snapshot, workspace, graph, task, prepared, cancelled);
1622 (task.id.clone(), keyed)
1623 })
1624 .collect())
1625}
1626
1627fn key_now(
1628 snapshot: &hash::Snapshot,
1629 workspace: &Workspace,
1630 graph: &TaskGraph,
1631 task: &Task,
1632 prepared: &BTreeMap<String, PreparedTask>,
1633 cancelled: &AtomicBool,
1634) -> Keyed {
1635 let failed = |error: anyhow::Error| format!("{}: {error:#}", task.id);
1636 let dependencies =
1637 hash::dependency_keys(snapshot, workspace, graph, task, cancelled).map_err(failed)?;
1638 let inputs = hash::inputs(
1639 snapshot,
1640 workspace,
1641 task,
1642 &prepared[&task.id],
1643 &dependencies,
1644 cancelled,
1645 )
1646 .map_err(failed)?;
1647 Ok((hash::key(&inputs).map_err(failed)?, inputs))
1648}
1649
1650#[doc(hidden)]
1652pub fn run_upload_worker() -> anyhow::Result<()> {
1653 remote::background::run()
1654}
1655
1656fn workspace_outputs(workspace: &Workspace) -> recorded::WorkspaceOutputs {
1662 let mut outputs = Vec::new();
1663 for (name, project) in &workspace.projects {
1664 for (target, definition) in &project.targets {
1665 if let Some(scratch) = scratch_outputs(workspace, name, definition) {
1666 let configurations = std::iter::once(None)
1667 .chain(definition.configurations.keys().cloned().map(Some));
1668 for configuration in configurations {
1669 outputs.push((
1670 (name.clone(), target.clone(), configuration),
1671 scratch.clone(),
1672 ));
1673 }
1674 }
1675 let configured = definition
1676 .configurations
1677 .iter()
1678 .map(|(configuration, options)| {
1679 let mut definition = definition.clone();
1680 definition.options.extend(options.clone());
1681 (Some(configuration.clone()), definition)
1682 });
1683 for (configuration, definition) in
1684 std::iter::once((None, definition.clone())).chain(configured)
1685 {
1686 let task = Task {
1687 id: format!("{name}:{target}"),
1688 project: name.clone(),
1689 target: target.clone(),
1690 configuration: configuration.clone(),
1691 args: Vec::new(),
1692 definition,
1693 dependencies: Default::default(),
1694 edges: Default::default(),
1695 };
1696 if let Some(declared) = paths::Outputs::new(workspace, &task)
1697 .ok()
1698 .filter(paths::Outputs::declared)
1699 {
1700 outputs.push(((name.clone(), target.clone(), configuration), declared));
1701 }
1702 }
1703 }
1704 }
1705 let roots = workspace
1706 .projects
1707 .iter()
1708 .map(|(name, project)| (name.clone(), project.root.clone()))
1709 .collect();
1710 recorded::WorkspaceOutputs::new(outputs, &roots)
1711}
1712
1713fn scratch_outputs(
1716 workspace: &Workspace,
1717 project: &str,
1718 definition: &qk_config::Target,
1719) -> Option<paths::Outputs> {
1720 let directories =
1721 scratch::directories_of(workspace, project, definition.extra.get("qk:sandbox")).ok()?;
1722 let patterns: Vec<String> = directories
1723 .iter()
1724 .filter_map(|directory| {
1725 let relative = directory.strip_prefix(&workspace.root).ok()?.to_str()?;
1726 Some(glob::escape(&relative.replace('\\', "/")))
1727 })
1728 .collect();
1729 if patterns.is_empty() {
1730 return None;
1731 }
1732 paths::Outputs::from_paths(&patterns).ok()
1733}
1734
1735fn produced(
1740 snapshot: &hash::Snapshot,
1741 workspace: &Workspace,
1742 graph: &TaskGraph,
1743 task: &Task,
1744) -> recorded::Produced {
1745 let (upstream, declared): (Vec<_>, Vec<_>) = graph
1746 .output_dependencies(task, true)
1747 .iter()
1748 .filter_map(|id| graph.keyed_task(id))
1749 .filter_map(|dependency| {
1750 let outputs = paths::Outputs::new(workspace, dependency)
1751 .ok()
1752 .filter(paths::Outputs::declared)?;
1753 let (declared, _) = paths::resolved_outputs(workspace, dependency).ok()?;
1754 Some((outputs, declared))
1755 })
1756 .unzip();
1757 let include_ignored = snapshot
1758 .include_ignored_roots(workspace, task)
1759 .unwrap_or_else(|error| {
1760 qk_executor::status!(
1761 "qk: {}: includeIgnored inputs not kept in recordings ({error:#})",
1762 task.id
1763 );
1764 Vec::new()
1765 });
1766 recorded::Produced::new(upstream, include_ignored).declared(declared)
1767}
1768
1769fn upstream(graph: &TaskGraph, task: &Task) -> BTreeSet<recorded::TargetOf> {
1773 let of = |task: &Task| {
1774 (
1775 task.project.clone(),
1776 task.target.clone(),
1777 task.configuration.clone(),
1778 )
1779 };
1780 graph
1783 .output_dependencies(task, true)
1784 .iter()
1785 .filter_map(|id| graph.keyed_task(id))
1786 .map(of)
1787 .collect()
1788}
1789
1790#[cfg(test)]
1791mod tests {
1792 use super::*;
1793
1794 #[test]
1795 fn concurrent_blob_writes_keep_results_ordered_and_repair_corruption() {
1796 let workspace = tempfile::tempdir().unwrap();
1797 let root = workspace.path();
1798 let cache = Cache::new(root.join("cache"));
1799 cache.initialize().unwrap();
1800 let mut sources = Vec::new();
1801 for index in 0..130 {
1802 let source = root.join(format!("source-{index}"));
1803 std::fs::write(&source, format!("content-{}", index % 65)).unwrap();
1804 sources.push(source);
1805 }
1806 let digest = cache.put_blob(&sources[0]).unwrap();
1807 std::fs::write(cache.root.join("blobs").join(&digest), "corrupt").unwrap();
1808 std::fs::remove_file(&sources[31]).unwrap();
1809 let results = cache.put_blobs(&sources);
1810 assert_eq!(results.len(), sources.len());
1811 for (index, result) in results.into_iter().enumerate() {
1812 if index == 31 {
1813 assert_eq!(
1814 result
1815 .unwrap_err()
1816 .downcast_ref::<std::io::Error>()
1817 .unwrap()
1818 .kind(),
1819 std::io::ErrorKind::NotFound
1820 );
1821 continue;
1822 }
1823 let digest = result.unwrap();
1824 let held = cache.root.join("blobs").join(&digest);
1825 assert_eq!(
1826 std::fs::read(&held).unwrap(),
1827 format!("content-{}", index % 65).as_bytes()
1828 );
1829 assert_eq!(hash::digest_file(&held).unwrap(), digest);
1830 }
1831 }
1832
1833 #[cfg(unix)]
1835 #[test]
1836 fn a_blob_the_store_holds_is_not_written_again() {
1837 use std::os::unix::fs::MetadataExt;
1838 let workspace = tempfile::tempdir().unwrap();
1839 let root = workspace.path();
1840 let cache = Cache::new(root.join(".qk/cache"));
1841 cache.initialize().unwrap();
1842 std::fs::write(root.join("first"), "same").unwrap();
1843 std::fs::write(root.join("second"), "same").unwrap();
1844 std::fs::write(root.join("other"), "different").unwrap();
1845 let first = cache.put_blob(&root.join("first")).unwrap();
1846 let blob = cache.root.join("blobs").join(&first);
1847 let stored = std::fs::metadata(&blob).unwrap().ino();
1848 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1849 assert_eq!(std::fs::metadata(&blob).unwrap().ino(), stored);
1850 std::fs::write(&blob, "corrupt").unwrap();
1852 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1853 assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1854 {
1856 use std::os::unix::fs::PermissionsExt;
1857 std::fs::set_permissions(&blob, std::fs::Permissions::from_mode(0o000)).unwrap();
1858 }
1859 assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
1860 assert_eq!(std::fs::read(&blob).unwrap(), b"same");
1861 let other = cache.put_blob(&root.join("other")).unwrap();
1862 assert_ne!(other, first);
1863 assert_eq!(
1864 std::fs::read(cache.root.join("blobs").join(other)).unwrap(),
1865 b"different"
1866 );
1867 }
1868
1869 #[test]
1872 fn complete_directory_restores_preserve_outputs() {
1873 let workspace = tempfile::tempdir().unwrap();
1874 let root = workspace.path();
1875 std::fs::create_dir_all(root.join("dist/nested/empty")).unwrap();
1876 std::fs::write(root.join("dist/nested/a.txt"), "a").unwrap();
1877 std::fs::write(root.join("dist/b.bin"), [0, 1, 2]).unwrap();
1878 std::fs::create_dir_all(root.join("bundle/empty")).unwrap();
1879 std::fs::write(root.join("bundle/readonly"), "readonly").unwrap();
1880 #[cfg(unix)]
1881 {
1882 use std::os::unix::fs::PermissionsExt;
1883 let script = root.join("dist/run.sh");
1884 std::fs::write(&script, "#!/bin/sh\n").unwrap();
1885 std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
1886 std::os::unix::fs::symlink("nested/a.txt", root.join("dist/link")).unwrap();
1887 std::fs::set_permissions(
1888 root.join("dist/nested"),
1889 std::fs::Permissions::from_mode(0o500),
1890 )
1891 .unwrap();
1892 std::fs::set_permissions(
1893 root.join("bundle/readonly"),
1894 std::fs::Permissions::from_mode(0o444),
1895 )
1896 .unwrap();
1897 }
1898 let log = root.join("log");
1899 std::fs::write(&log, "").unwrap();
1900
1901 let cache = Cache::new(root.join(".qk/cache"));
1902 cache.initialize().unwrap();
1903 let outputs = paths::Outputs::from_patterns(&["dist", "bundle"]);
1904 let key = "0".repeat(64);
1905 cache.publish(root, &key, &outputs, &log).unwrap();
1906
1907 #[cfg(unix)]
1908 {
1909 use std::os::unix::fs::PermissionsExt;
1910 std::fs::set_permissions(
1913 root.join("dist/nested"),
1914 std::fs::Permissions::from_mode(0o755),
1915 )
1916 .unwrap();
1917 }
1918 std::fs::remove_dir_all(root.join("dist")).unwrap();
1919 std::fs::remove_dir_all(root.join("bundle")).unwrap();
1920 let before = std::time::SystemTime::now();
1921 let restored = cache
1922 .restore(
1923 root,
1924 "app:build",
1925 &Default::default(),
1926 &key,
1927 &outputs,
1928 &qk_executor::Display::Stream,
1929 qk_executor::Shown::LocalCache,
1930 )
1931 .unwrap();
1932 assert!(restored);
1933 assert!(root.join("dist/nested/empty").is_dir());
1934 assert!(root.join("bundle/empty").is_dir());
1935 assert_eq!(
1936 std::fs::read_to_string(root.join("bundle/readonly")).unwrap(),
1937 "readonly"
1938 );
1939 #[cfg(unix)]
1940 {
1941 assert_eq!(
1942 std::fs::read_link(root.join("dist/link")).unwrap(),
1943 std::path::Path::new("nested/a.txt")
1944 );
1945 assert_eq!(
1946 store::mode(&std::fs::metadata(root.join("dist/run.sh")).unwrap()),
1947 0o755
1948 );
1949 assert_eq!(
1950 store::mode(&std::fs::metadata(root.join("dist/nested")).unwrap()),
1951 0o500
1952 );
1953 assert_eq!(
1954 store::mode(&std::fs::metadata(root.join("bundle/readonly")).unwrap()),
1955 0o444
1956 );
1957 }
1958 let path = root.join("dist/nested/a.txt");
1959 let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
1960 assert!(modified >= before);
1961 let kept = cache
1962 .restore(
1963 root,
1964 "app:build",
1965 &Default::default(),
1966 &key,
1967 &outputs,
1968 &qk_executor::Display::Hidden,
1969 qk_executor::Shown::LocalCache,
1970 )
1971 .unwrap();
1972 assert!(kept);
1973 assert_eq!(
1974 std::fs::metadata(&path).unwrap().modified().unwrap(),
1975 modified
1976 );
1977 #[cfg(unix)]
1978 {
1979 use std::os::unix::fs::PermissionsExt;
1980 std::fs::set_permissions(
1981 root.join("dist/nested"),
1982 std::fs::Permissions::from_mode(0o755),
1983 )
1984 .unwrap();
1985 }
1986 }
1987
1988 #[test]
1989 fn listings_leave_out_other_targets_outputs_but_the_upstreams() {
1990 let temp = tempfile::tempdir().unwrap();
1991 let root = temp.path();
1992 std::fs::write(root.join("nx.json"), "{}").unwrap();
1993 std::fs::create_dir_all(root.join("app")).unwrap();
1994 std::fs::create_dir_all(root.join("lib")).unwrap();
1995 std::fs::write(
1998 root.join("project.json"),
1999 r#"{"name":"workspace","targets":{"cache":{"command":"true",
2000 "outputs":["{projectRoot}/.cache/build","{projectRoot}/app/extra/out"]}}}"#,
2001 )
2002 .unwrap();
2003 std::fs::write(
2004 root.join("app/project.json"),
2005 r#"{"name":"app","root":"./app","targets":{
2006 "build":{"command":"true","outputs":["{options.outputPath}"],
2007 "options":{"outputPath":"app/dist"},
2008 "configurations":{"production":{"outputPath":"app/dist-production"}}},
2009 "lint":{"command":"true"},
2010 "browser":{"command":"true","qk:sandbox":{"scratch":["{projectRoot}/.vitest"]}},
2011 "cover":{"command":"true","outputs":["{projectRoot}/.coverage/unit"]},
2012 "cover-browser":{"command":"true","outputs":["{projectRoot}/.coverage/browser"]},
2013 "wasm":{"command":"true","outputs":["{projectRoot}/target/wasm/release/*.wasm"]},
2014 "test":{"command":"true","dependsOn":[{"projects":["lib"],"target":"build"}],
2015 "configurations":{"production":{}}}}}"#,
2016 )
2017 .unwrap();
2018 std::fs::write(
2019 root.join("lib/project.json"),
2020 r#"{"name":"lib","targets":{"build":{"command":"true","outputs":["{options.outputPath}"],
2021 "options":{"outputPath":"lib/dist"},
2022 "qk:sandbox":{"scratch":["{projectRoot}/.scratch"]},
2023 "configurations":{"production":{"outputPath":"lib/dist-production"}}}}}"#,
2024 )
2025 .unwrap();
2026 let workspace = Workspace::load(root).unwrap();
2027 let graph = qk_taskgraph::TaskGraph::build(
2028 &workspace,
2029 &[
2030 qk_taskgraph::Request::parse("app:lint").unwrap(),
2031 qk_taskgraph::Request::parse("app:cover").unwrap(),
2032 qk_taskgraph::Request::parse("app:test").unwrap(),
2033 qk_taskgraph::Request::parse("app:test:production").unwrap(),
2034 ],
2035 )
2036 .unwrap();
2037 let outputs = workspace_outputs(&workspace);
2038 let others =
2039 |task: &str| recorded::Others::new(&outputs, upstream(&graph, &graph.tasks[task]), &[]);
2040 let lint = others("app:lint");
2041 assert!(lint.matches("app/dist/main.js"));
2042 assert!(lint.matches("app/dist-production/main.js"));
2043 assert!(lint.matches("lib/dist"));
2044 assert!(!lint.matches("app/src/main.ts"));
2045 assert!(lint.matches("app/.vitest"));
2046 assert!(lint.matches("app/.vitest/attachments/diff.png"));
2047 assert!(lint.matches("app/target"));
2049 assert!(lint.matches("app/target/wasm"));
2050 assert!(!lint.matches("app"));
2051 assert!(lint.matches(".cache"));
2053 assert!(lint.matches("app/extra"));
2054 assert!(!lint.matches("lib"));
2055 let unlisted = recorded::Unlisted {
2058 own: Vec::new(),
2059 others: Some(lint),
2060 ignore: &ultracache::IgnoredReads::none(),
2061 produced: None,
2062 };
2063 let directories = unlisted.directories("lib");
2064 for directory in [".", ".cache", "app", "app/extra"] {
2065 assert!(directories.contains(directory), "{directory}");
2066 }
2067 assert!(!others("app:test").matches("lib/dist"));
2070 assert!(others("app:test").matches("lib/dist-production"));
2071 assert!(others("app:cover").matches("app/.coverage"));
2074 let production = others("app:test:production");
2075 assert!(!production.matches("lib/dist-production"));
2076 assert!(production.matches("lib/dist"));
2077 assert!(others("app:lint").matches("lib/.scratch"));
2080 assert!(!others("app:test").matches("lib/.scratch"));
2081 assert!(!production.matches("lib/.scratch"));
2082 }
2083}