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