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