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