1use std::collections::BTreeMap;
17use std::fs;
18use std::io::Write;
19use std::path::PathBuf;
20use std::time::{Duration, SystemTime, UNIX_EPOCH};
21
22use anyhow::{Context, Result, bail};
23use qk_config::Workspace;
24use qk_taskgraph::{Task, TaskGraph};
25use serde::{Deserialize, Serialize};
26use serde_json::Value;
27
28use crate::paths::{self, Outputs};
29use crate::record::{self, Artifact, WarmGroup, WarmRecord, validate_link};
30use crate::store::{mode, set_mode, symlink};
31use crate::{Cache, hash::digest_file};
32
33pub const RESTORED_AT: std::time::SystemTime = std::time::UNIX_EPOCH;
38
39const KEPT: usize = 8;
41
42#[derive(Clone, Debug, Default)]
44pub struct Warm {
45 pub identity: String,
47 pub outputs: bool,
48 pub paths: Vec<String>,
50 pub env: BTreeMap<String, String>,
52 pub directory: bool,
54 pub max_size: Option<u64>,
55 pub remote: bool,
56 pub preserve_mtimes: bool,
59 pub portable: bool,
61 pub key: Vec<KeyPart>,
63 pub restore_keys: Option<usize>,
66 pub survive: Vec<String>,
69 pub background: bool,
72}
73
74#[derive(Clone, Debug)]
76pub enum KeyPart {
77 File(String),
79 Env(String),
81}
82
83pub fn config(workspace: &Workspace, task: &Task) -> Result<Option<Warm>> {
85 let Some(value) = task.definition.extra.get("qk:warm") else {
86 return Ok(None);
87 };
88 let object = value.as_object().context("qk:warm must be an object")?;
89 for key in object.keys() {
90 if !matches!(
91 key.as_str(),
92 "outputs"
93 | "paths"
94 | "env"
95 | "maxSize"
96 | "remote"
97 | "mtimes"
98 | "portable"
99 | "key"
100 | "restoreKeys"
101 | "group"
102 | "survive"
103 | "save"
104 ) {
105 bail!("unknown qk:warm field {key:?}");
106 }
107 }
108 let flag = |name: &str, default: bool| -> Result<bool> {
109 object.get(name).map_or(Ok(default), |value| {
110 value
111 .as_bool()
112 .with_context(|| format!("qk:warm.{name} must be true or false"))
113 })
114 };
115 let group = object
116 .get("group")
117 .map(|value| {
118 value
119 .as_str()
120 .filter(|name| !name.is_empty())
121 .context("qk:warm.group must be a name")
122 })
123 .transpose()?;
124 if group.is_some() && (object.contains_key("outputs") || object.contains_key("paths")) {
125 bail!("a qk:warm.group shares {{warm}} alone; outputs and paths belong to one task");
126 }
127 let identity = group.map_or_else(|| task.id.clone(), |group| format!("group {group}"));
128 const SURVIVE_SHAPE: &str = "qk:warm.survive must be an array of target names";
129 let survive = object
130 .get("survive")
131 .map(|value| {
132 value
133 .as_array()
134 .context(SURVIVE_SHAPE)?
135 .iter()
136 .map(|target| {
137 target
138 .as_str()
139 .filter(|target| !target.is_empty())
140 .map(str::to_owned)
141 .context(SURVIVE_SHAPE)
142 })
143 .collect::<Result<Vec<_>>>()
144 })
145 .transpose()?
146 .unwrap_or_default();
147 let directory = directory(workspace, &identity);
148 let directory_text = directory.to_str().context("warm directory must be UTF-8")?;
149 let workspace_text = workspace
150 .root
151 .to_str()
152 .context("workspace root must be UTF-8")?;
153 let project_root = &workspace.projects[&task.project].root;
154 let project_directory = if project_root == "." {
155 workspace.root.clone()
156 } else {
157 workspace.root.join(project_root)
158 };
159 let project_text = project_directory
160 .to_str()
161 .context("project root must be UTF-8")?;
162 let expand = |text: &str| -> Result<String> {
163 if text.contains("{options.") || text.contains("{args.") {
164 bail!("dynamic warm environment values are not supported yet");
165 }
166 Ok(text
167 .replace("{workspaceRoot}", workspace_text)
168 .replace("{projectRoot}", project_text)
169 .replace("{warm}", directory_text))
170 };
171 let mut paths = Vec::new();
172 for path in object
173 .get("paths")
174 .map(|value| {
175 value
176 .as_array()
177 .context("qk:warm.paths must be an array of paths")
178 })
179 .transpose()?
180 .into_iter()
181 .flatten()
182 {
183 let path = path
184 .as_str()
185 .context("qk:warm.paths must be an array of paths")?;
186 if path.contains("{warm}") {
187 bail!("qk:warm.paths are workspace paths; {{warm}} is kept already");
188 }
189 let (excluded, path) = match path.strip_prefix('!') {
190 Some(path) => (true, path),
191 None => (false, path),
192 };
193 let path = paths::expand(workspace, &task.project, path)?
194 .trim_end_matches('/')
195 .to_owned();
196 paths::validate_path(&path)?;
197 paths.push(if excluded { format!("!{path}") } else { path });
198 }
199 let mut env = BTreeMap::new();
200 let mut uses_directory = false;
201 if let Some(values) = object.get("env") {
202 for (name, value) in values
203 .as_object()
204 .context("qk:warm.env must be an object")?
205 {
206 let value = value
207 .as_str()
208 .context("qk:warm.env values must be strings")?;
209 uses_directory |= value.contains("{warm}");
210 env.insert(name.clone(), expand(value)?);
211 }
212 }
213 let max_size = match object.get("maxSize") {
214 None => None,
215 Some(Value::String(text)) => Some(crate::parse_size(text)?),
216 Some(Value::Number(number)) => number.as_u64(),
217 Some(_) => bail!("qk:warm.maxSize must be a size such as \"2GB\""),
218 };
219 let preserve_mtimes = match object.get("mtimes").map(|value| value.as_str()) {
220 None | Some(Some("epoch")) => false,
221 Some(Some("preserve")) => true,
222 Some(_) => bail!("qk:warm.mtimes must be \"epoch\" or \"preserve\""),
223 };
224 const KEY_SHAPE: &str = "qk:warm.key must be an array of paths and {\"env\": name} objects";
225 let mut key = Vec::new();
226 for part in object
227 .get("key")
228 .map(|value| value.as_array().context(KEY_SHAPE))
229 .transpose()?
230 .into_iter()
231 .flatten()
232 {
233 key.push(match part {
234 Value::String(path) => {
235 let path = paths::expand(workspace, &task.project, path)?;
236 paths::validate_path(&path)?;
237 KeyPart::File(path)
238 }
239 Value::Object(object) if object.len() == 1 => KeyPart::Env(
240 object
241 .get("env")
242 .and_then(Value::as_str)
243 .context(KEY_SHAPE)?
244 .to_owned(),
245 ),
246 _ => bail!(KEY_SHAPE),
247 });
248 }
249 let restore_keys = object
250 .get("restoreKeys")
251 .map(|value| {
252 value
253 .as_u64()
254 .and_then(|count| usize::try_from(count).ok())
255 .filter(|count| *count <= key.len())
256 .context("qk:warm.restoreKeys must be a count no larger than qk:warm.key")
257 })
258 .transpose()?;
259 let background = match object.get("save").map(Value::as_str) {
260 None | Some(Some("wait")) => false,
261 Some(Some("background")) => true,
262 Some(_) => bail!("qk:warm.save must be \"wait\" or \"background\""),
263 };
264 Ok(Some(Warm {
265 identity,
266 outputs: flag("outputs", false)?,
267 paths,
268 env,
269 directory: uses_directory,
270 max_size,
271 remote: flag("remote", true)?,
272 preserve_mtimes,
273 portable: flag("portable", false)?,
274 key,
275 restore_keys,
276 survive,
277 background,
278 }))
279}
280
281fn directory(workspace: &Workspace, identity: &str) -> PathBuf {
283 paths::worktree_state(&workspace.root)
284 .join("warm")
285 .join(&blake3::hash(identity.as_bytes()).to_hex()[..32])
286}
287
288impl Warm {
289 pub fn kept_paths(&self) -> Vec<String> {
292 self.paths
293 .iter()
294 .filter(|path| !path.starts_with('!'))
295 .cloned()
296 .collect()
297 }
298}
299
300#[derive(Default)]
303pub struct Kept {
304 roots: Vec<PathBuf>,
306}
307
308impl Drop for Kept {
309 fn drop(&mut self) {
310 for root in self.roots.iter().rev() {
311 release(root);
312 }
313 }
314}
315
316#[derive(Default, Serialize, Deserialize)]
318struct Stash {
319 #[serde(skip)]
320 root: PathBuf,
321 #[serde(skip)]
322 workspace: PathBuf,
323 moved: Vec<String>,
325}
326
327struct Active {
330 users: usize,
331 stash: Stash,
332 _lock: fs::File,
333}
334
335static ACTIVE: std::sync::Mutex<BTreeMap<PathBuf, Active>> = std::sync::Mutex::new(BTreeMap::new());
338
339impl Stash {
340 fn manifest(root: &std::path::Path) -> PathBuf {
342 root.join("moved.json")
343 }
344
345 fn save_manifest(&self) -> Result<()> {
348 let mut file = tempfile::NamedTempFile::new_in(&self.root)?;
349 file.write_all(&serde_json::to_vec(self)?)?;
350 file.as_file().sync_all()?;
351 file.persist(Self::manifest(&self.root))?;
352 Ok(())
353 }
354
355 fn put_back(&self) -> bool {
361 let mut complete = true;
362 for (index, path) in self.moved.iter().enumerate().rev() {
363 let kept = self.root.join(index.to_string());
364 match fs::symlink_metadata(&kept) {
365 Ok(_) => {}
366 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
367 Err(error) => {
368 complete = false;
369 qk_executor::status!(
370 "qk: could not inspect saved {path}, kept for the next run ({error})"
371 );
372 continue;
373 }
374 }
375 let original = self.workspace.join(path);
376 let result = paths::safe_parents(&self.workspace, path).and_then(|()| {
377 remove(&original)?;
378 if let Some(parent) = original.parent() {
379 fs::create_dir_all(parent)?;
380 }
381 fs::rename(&kept, &original)?;
382 Ok(())
383 });
384 if let Err(error) = result {
385 complete = false;
386 qk_executor::status!(
387 "qk: could not put {path} back, kept for the next run ({error:#})"
388 );
389 }
390 }
391 if complete {
392 let _ = fs::remove_dir_all(&self.root);
393 }
394 complete
395 }
396}
397
398fn remove(path: &std::path::Path) -> std::io::Result<()> {
400 match fs::symlink_metadata(path) {
401 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
402 Err(error) => Err(error),
403 Ok(metadata) if metadata.is_dir() => fs::remove_dir_all(path),
404 Ok(_) => fs::remove_file(path),
405 }
406}
407
408fn release(root: &std::path::Path) {
411 let mut active = ACTIVE
412 .lock()
413 .unwrap_or_else(std::sync::PoisonError::into_inner);
414 let Some(entry) = active.get_mut(root) else {
415 return;
416 };
417 entry.users -= 1;
418 if entry.users > 0 {
419 return;
420 }
421 let entry = active.remove(root).expect("present above");
422 drop(active);
425 entry.stash.put_back();
426}
427
428fn names(owner: &Task, entry: &str, task: &Task) -> bool {
430 match entry.split_once(':') {
431 Some((project, target)) => task.project == project && task.target == target,
432 None => task.project == owner.project && task.target == entry,
433 }
434}
435
436fn depends_on(graph: &TaskGraph, from: &str, on: &str) -> bool {
438 let mut seen = std::collections::BTreeSet::new();
439 let mut next = vec![from];
440 while let Some(id) = next.pop() {
441 for dependency in graph
442 .tasks
443 .get(id)
444 .into_iter()
445 .flat_map(|task| &task.dependencies)
446 {
447 if dependency == on {
448 return true;
449 }
450 if seen.insert(dependency.as_str()) {
451 next.push(dependency);
452 }
453 }
454 }
455 false
456}
457
458pub fn keep_across(workspace: &Workspace, graph: &TaskGraph, task: &Task) -> Kept {
461 let mut kept = Kept::default();
462 for owner in graph.tasks.values() {
463 let Ok(Some(warm)) = config(workspace, owner) else {
464 continue;
465 };
466 if !warm.survive.iter().any(|entry| names(owner, entry, task))
467 || !depends_on(graph, &owner.id, &task.id)
468 {
469 continue;
470 }
471 match acquire(workspace, owner, &warm) {
472 Ok(root) => kept.roots.push(root),
473 Err(error) => qk_executor::status!(
474 "qk: {}: warm paths not kept across {} ({error:#})",
475 owner.id,
476 task.id
477 ),
478 }
479 }
480 kept
481}
482
483fn acquire(workspace: &Workspace, owner: &Task, warm: &Warm) -> Result<PathBuf> {
486 let state = paths::worktree_state(&workspace.root).join("warm-kept");
487 let root = state.join(hash32(&owner.id));
488 fs::create_dir_all(&state)?;
489 let lock = fs::OpenOptions::new()
490 .create(true)
491 .truncate(false)
492 .read(true)
493 .write(true)
494 .open(state.join(format!("{}.lock", hash32(&owner.id))))?;
495 let mut active = loop {
498 let mut active = ACTIVE
499 .lock()
500 .unwrap_or_else(std::sync::PoisonError::into_inner);
501 if let Some(entry) = active.get_mut(&root) {
502 entry.users += 1;
503 return Ok(root);
504 }
505 match fs2::FileExt::try_lock_exclusive(&lock) {
506 Ok(()) => break active,
507 Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => {
508 drop(active);
509 std::thread::sleep(Duration::from_millis(20));
510 }
511 Err(error) => return Err(error.into()),
512 }
513 };
514 let left = match fs::read(Stash::manifest(&root)) {
521 Ok(bytes) => {
522 let mut left: Stash = serde_json::from_slice(&bytes)
523 .context("invalid warm stash manifest; saved paths retained for recovery")?;
524 left.root = root.clone();
525 left.workspace = workspace.root.clone();
526 Some(left)
527 }
528 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
529 if root.is_dir() && fs::read_dir(&root)?.next().is_some() {
530 bail!("missing warm stash manifest; saved paths retained for recovery");
531 }
532 None
533 }
534 Err(error) => return Err(error).context("cannot read warm stash; saved paths retained"),
535 };
536 let retained = left.filter(|left| !left.put_back());
537 let stash = stash(workspace, &root, warm, retained)?;
538 active.insert(
539 root.clone(),
540 Active {
541 users: 1,
542 stash,
543 _lock: lock,
544 },
545 );
546 Ok(root)
547}
548
549fn stash(
553 workspace: &Workspace,
554 root: &std::path::Path,
555 warm: &Warm,
556 retained: Option<Stash>,
557) -> Result<Stash> {
558 let mut stash = match retained {
559 Some(retained) => retained,
560 None => {
561 let _ = fs::remove_dir_all(root);
562 Stash {
563 root: root.to_owned(),
564 workspace: workspace.root.clone(),
565 moved: Vec::new(),
566 }
567 }
568 };
569 let held: Vec<String> = stash
570 .moved
571 .iter()
572 .enumerate()
573 .filter_map(
574 |(index, path)| match fs::symlink_metadata(root.join(index.to_string())) {
575 Ok(_) => Some(Ok(path.clone())),
576 Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
577 Err(error) => Some(Err(error).context("cannot inspect retained warm path")),
578 },
579 )
580 .collect::<Result<_>>()?;
581 let overlaps = |path: &str| {
582 held.iter().any(|held| {
583 held == path
584 || path.starts_with(&format!("{held}/"))
585 || held.starts_with(&format!("{path}/"))
586 })
587 };
588 let mut kept = std::collections::BTreeSet::new();
591 for path in warm.kept_paths() {
592 if let Ok(found) = Outputs::from_paths(std::slice::from_ref(&path))
593 .and_then(|outputs| outputs.paths(&workspace.root))
594 {
595 kept.extend(found);
596 }
597 }
598 let topmost: Vec<String> = kept
600 .iter()
601 .filter(|path| {
602 !std::path::Path::new(path.as_str())
603 .ancestors()
604 .skip(1)
605 .any(|ancestor| {
606 ancestor
607 .to_str()
608 .is_some_and(|ancestor| kept.contains(ancestor))
609 })
610 })
611 .filter(|path| !overlaps(path))
612 .cloned()
613 .collect();
614 if topmost.is_empty() {
615 return Ok(stash);
616 }
617 fs::create_dir_all(root)?;
618 for path in topmost {
619 let moved = paths::safe_parents(&workspace.root, &path).and_then(|()| {
620 stash.moved.push(path.clone());
621 stash.save_manifest()?;
622 fs::rename(
625 workspace.root.join(&path),
626 root.join((stash.moved.len() - 1).to_string()),
627 )
628 .inspect_err(|_| {
629 stash.moved.pop();
630 })?;
631 Ok(())
632 });
633 if let Err(error) = moved {
634 stash.save_manifest()?;
637 stash.put_back();
638 return Err(error);
639 }
640 }
641 Ok(stash)
642}
643
644pub fn check_overlaps(workspace: &Workspace, graph: &TaskGraph) -> Result<()> {
647 let mut claimed: Vec<(String, &str)> = Vec::new();
648 for (id, task) in &graph.tasks {
649 let Some(warm) = config(workspace, task)? else {
650 continue;
651 };
652 for path in warm.kept_paths() {
653 if let Some((_, other)) = claimed.iter().find(|(known, _)| {
654 known == &path
655 || known.starts_with(&format!("{path}/"))
656 || path.starts_with(&format!("{known}/"))
657 }) {
658 bail!("{id} and {other} both keep {path} as warm state");
659 }
660 claimed.push((path, id));
661 }
662 }
663 Ok(())
664}
665
666impl WarmGroup {
667 fn size(&self, cache: &Cache) -> u64 {
669 self.artifacts
670 .values()
671 .filter_map(|artifact| match artifact {
672 Artifact::File { blob, .. } => {
673 fs::metadata(cache.root.join("blobs").join(blob)).ok()
674 }
675 _ => None,
676 })
677 .map(|metadata| metadata.len())
678 .sum()
679 }
680}
681
682#[derive(Default, Serialize, Deserialize)]
685struct RestoredFiles {
686 groups: BTreeMap<String, BTreeMap<String, (Vec<i64>, Artifact)>>,
687}
688
689#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
691pub struct Restored {
692 pub source: String,
695 pub groups: Vec<String>,
696 pub files: usize,
697 pub bytes: u64,
698}
699
700#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
702#[serde(rename_all = "camelCase")]
703pub struct WarmReport {
704 pub restored: Option<Restored>,
706 #[serde(default, skip_serializing_if = "Vec::is_empty")]
708 pub present: Vec<String>,
709 pub save_ms: Option<u64>,
712 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
714 pub background: bool,
715}
716
717#[derive(Clone)]
720struct Location {
721 name: &'static str,
722 base: PathBuf,
723 outputs: Option<Outputs>,
725}
726
727impl Location {
728 fn paths(&self) -> Result<std::collections::BTreeSet<String>> {
730 if !self.base.exists() {
731 return Ok(Default::default());
732 }
733 match &self.outputs {
734 Some(outputs) => outputs.paths(&self.base),
735 None => {
736 let mut paths = std::collections::BTreeSet::new();
737 for entry in walkdir::WalkDir::new(&self.base)
738 .min_depth(1)
739 .follow_links(false)
740 {
741 let entry = entry?;
742 paths.insert(paths::relative(&self.base, entry.path())?);
743 }
744 Ok(paths)
745 }
746 }
747 }
748
749 fn matches(&self, path: &str) -> bool {
751 self.outputs
752 .as_ref()
753 .is_none_or(|outputs| outputs.matches(path))
754 }
755}
756
757fn locations(workspace: &Workspace, task: &Task, warm: &Warm) -> Result<Vec<Location>> {
759 let mut locations = Vec::new();
760 if warm.outputs {
761 locations.push(Location {
762 name: "outputs",
763 base: workspace.root.clone(),
764 outputs: Some(Outputs::new(workspace, task)?),
765 });
766 }
767 if !warm.paths.is_empty() {
768 locations.push(Location {
769 name: "paths",
770 base: workspace.root.clone(),
771 outputs: Some(Outputs::from_paths(&warm.paths)?),
772 });
773 }
774 if warm.directory {
775 locations.push(Location {
776 name: "directory",
777 base: directory(workspace, &warm.identity),
778 outputs: None,
779 });
780 }
781 Ok(locations)
782}
783
784fn current_branch(workspace: &Workspace) -> Option<String> {
787 for name in ["GITHUB_HEAD_REF", "GITHUB_REF_NAME"] {
788 if let Ok(value) = std::env::var(name)
789 && !value.is_empty()
790 {
791 return Some(value);
792 }
793 }
794 let output = paths::git(&workspace.root, &["rev-parse", "--abbrev-ref", "HEAD"]).ok()?;
795 let branch = String::from_utf8(output.stdout).ok()?.trim().to_owned();
796 (output.status.success() && !branch.is_empty() && branch != "HEAD").then_some(branch)
797}
798
799fn branches(workspace: &Workspace) -> Vec<String> {
802 let default = workspace
803 .config
804 .default_base
805 .clone()
806 .unwrap_or_else(|| "main".into());
807 let default = default
808 .strip_prefix("origin/")
809 .unwrap_or(&default)
810 .to_owned();
811 let mut branches: Vec<String> = current_branch(workspace).into_iter().collect();
812 if !branches.contains(&default) {
813 branches.push(default);
814 }
815 branches
816}
817
818fn stamp(metadata: &fs::Metadata) -> Vec<i64> {
820 let modified = metadata
821 .modified()
822 .ok()
823 .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
824 .map_or(-1, |duration| duration.as_nanos() as i64);
825 #[allow(unused_mut)]
826 let mut stamp = vec![metadata.len() as i64, modified, i64::from(mode(metadata))];
827 #[cfg(unix)]
828 {
829 use std::os::unix::fs::MetadataExt;
830 stamp.extend([
831 metadata.ctime(),
832 metadata.ctime_nsec(),
833 metadata.ino() as i64,
834 ]);
835 }
836 stamp
837}
838
839fn worktree(workspace: &Workspace) -> String {
841 fs::canonicalize(&workspace.root)
842 .unwrap_or_else(|_| workspace.root.clone())
843 .to_string_lossy()
844 .into_owned()
845}
846
847fn current_commit(workspace: &Workspace) -> Option<String> {
849 let output = paths::git(&workspace.root, &["rev-parse", "HEAD"]).ok()?;
850 let commit = String::from_utf8(output.stdout).ok()?.trim().to_owned();
851 (output.status.success() && !commit.is_empty()).then_some(commit)
852}
853
854fn key_digests(
857 workspace: &Workspace,
858 warm: &Warm,
859 prepared: &qk_executor::PreparedTask,
860) -> Result<Vec<String>> {
861 warm.key
862 .iter()
863 .map(|part| {
864 Ok(match part {
865 KeyPart::File(path) => {
866 paths::safe_parents(&workspace.root, path)?;
867 let absolute = workspace.root.join(path);
868 if absolute.is_file() {
869 format!("file:{}", digest_file(&absolute)?)
870 } else {
871 "file:absent".to_owned()
872 }
873 }
874 KeyPart::Env(name) => match prepared
877 .execution
878 .get(std::ffi::OsStr::new(name))
879 .or_else(|| prepared.env.get(std::ffi::OsStr::new(name)))
880 {
881 Some(value) => format!(
882 "env:{}",
883 blake3::hash(value.as_encoded_bytes()).to_hex()[..32].to_owned()
884 ),
885 None => "env:unset".to_owned(),
886 },
887 })
888 })
889 .collect()
890}
891
892fn key_match(warm: &Warm, current: &[String], saved: &[String]) -> Option<u8> {
896 if saved == current {
897 return Some(0);
898 }
899 let count = warm.restore_keys?;
900 (saved.len() == current.len() && saved[..count] == current[..count]).then_some(1)
901}
902
903fn behind(workspace: &Workspace, commit: &str) -> Option<u64> {
906 let ancestor = paths::git(
907 &workspace.root,
908 &["merge-base", "--is-ancestor", commit, "HEAD"],
909 )
910 .ok()?;
911 if !ancestor.status.success() {
912 return None;
913 }
914 let output = paths::git(
915 &workspace.root,
916 &["rev-list", "--count", &format!("{commit}..HEAD")],
917 )
918 .ok()?;
919 String::from_utf8(output.stdout).ok()?.trim().parse().ok()
920}
921
922fn hash32(text: &str) -> String {
924 blake3::hash(text.as_bytes()).to_hex()[..32].to_owned()
925}
926
927fn nanos(time: SystemTime) -> Option<i64> {
929 time.duration_since(UNIX_EPOCH)
930 .ok()
931 .and_then(|duration| i64::try_from(duration.as_nanos()).ok())
932}
933
934fn now_ms() -> u64 {
936 SystemTime::now()
937 .duration_since(UNIX_EPOCH)
938 .map_or(0, |duration| duration.as_millis() as u64)
939}
940
941fn restored_files_path(workspace: &Workspace, warm: &Warm) -> PathBuf {
943 paths::worktree_state(&workspace.root)
944 .join("warm-restored")
945 .join(format!("{}.json", hash32(&warm.identity)))
946}
947
948fn read_restored_files(workspace: &Workspace, warm: &Warm) -> RestoredFiles {
950 fs::read(restored_files_path(workspace, warm))
951 .ok()
952 .and_then(|bytes| serde_json::from_slice(&bytes).ok())
953 .unwrap_or_default()
954}
955
956impl Cache {
957 fn warm_prefix(&self, identity: &str) -> String {
959 hash32(&format!(
960 "{identity}\0{}\0{}",
961 std::env::consts::OS,
962 std::env::consts::ARCH
963 ))
964 }
965
966 fn lock_warm(&self, warm: &Warm) -> Result<fs::File> {
969 let directory = self.root.join("locks");
970 fs::create_dir_all(&directory)?;
971 let file = fs::OpenOptions::new()
972 .create(true)
973 .truncate(false)
974 .read(true)
975 .write(true)
976 .open(directory.join(format!("warm-{}", self.warm_prefix(&warm.identity))))?;
977 fs2::FileExt::lock_exclusive(&file)?;
978 Ok(file)
979 }
980
981 fn warm_record(&self, record: &WarmRecord) -> PathBuf {
982 self.root.join("warm").join(format!(
983 "{}-{}.json",
984 self.warm_prefix(&record.task),
985 hash32(&format!("{}\0{}", record.worktree, record.key.join("\0")))
986 ))
987 }
988
989 fn warm_records(&self, warm: &Warm) -> Vec<(PathBuf, WarmRecord)> {
991 let prefix = format!("{}-", self.warm_prefix(&warm.identity));
992 let mut records: Vec<(PathBuf, WarmRecord)> = fs::read_dir(self.root.join("warm"))
993 .into_iter()
994 .flatten()
995 .flatten()
996 .filter(|entry| {
997 entry
998 .file_name()
999 .to_str()
1000 .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".json"))
1001 })
1002 .filter_map(|entry| {
1003 let record =
1004 WarmRecord::read(&fs::read(entry.path()).ok()?, &warm.identity).ok()?;
1005 Some((entry.path(), record))
1006 })
1007 .collect();
1008 records.sort_by_key(|(_, record)| std::cmp::Reverse(record.saved));
1009 records
1010 }
1011
1012 fn choose_warm(
1018 &self,
1019 workspace: &Workspace,
1020 warm: &Warm,
1021 current: &[String],
1022 ) -> Option<(WarmRecord, String)> {
1023 let own = worktree(workspace);
1024 let mut candidates: Vec<(u8, bool, WarmRecord)> = self
1025 .warm_records(warm)
1026 .into_iter()
1027 .filter_map(|(_, record)| {
1028 let quality = key_match(warm, current, &record.key)?;
1029 let other = record.worktree != own;
1030 (!other || warm.portable).then_some((quality, other, record))
1031 })
1032 .collect();
1033 candidates.sort_by_key(|(quality, other, _)| (*quality, *other));
1035 let rank = candidates
1036 .first()
1037 .map(|(quality, other, _)| (*quality, *other))?;
1038 let tied = candidates
1039 .iter()
1040 .take_while(|(quality, other, _)| (*quality, *other) == rank)
1041 .count();
1042 let index = if tied > 1 {
1043 (0..tied)
1044 .min_by_key(|&index| {
1045 let behind = candidates[index]
1046 .2
1047 .commit
1048 .as_deref()
1049 .and_then(|commit| behind(workspace, commit));
1050 (behind.is_none(), behind, index)
1051 })
1052 .unwrap_or(0)
1053 } else {
1054 0
1055 };
1056 let (_, other, record) = candidates.swap_remove(index);
1057 let source = if other {
1058 format!("worktree {}", record.worktree)
1059 } else {
1060 "local".to_owned()
1061 };
1062 Some((record, source))
1063 }
1064
1065 fn store_warm(&self, record: &WarmRecord) -> Result<()> {
1066 let path = self.warm_record(record);
1067 let directory = path.parent().context("warm records have a directory")?;
1068 fs::create_dir_all(directory)?;
1069 record::publish(&self.root, &path, record)
1070 }
1071
1072 pub(crate) fn restore_warm(
1078 &self,
1079 workspace: &Workspace,
1080 task: &Task,
1081 warm: &Warm,
1082 prepared: &qk_executor::PreparedTask,
1083 ) -> Result<(Restored, Vec<String>)> {
1084 let mut restored = Restored::default();
1085 let _lock = self.lock_warm(warm)?;
1086 let mut present = Vec::new();
1088 let mut missing = Vec::new();
1089 for location in locations(workspace, task, warm)? {
1090 if location.paths()?.is_empty() {
1091 missing.push(location);
1092 } else {
1093 present.push(location.name.to_owned());
1094 }
1095 }
1096 if missing.is_empty() {
1097 return Ok((restored, present));
1098 }
1099 let current = key_digests(workspace, warm, prepared)?;
1100 let (record, source) = match self.choose_warm(workspace, warm, ¤t) {
1101 Some(chosen) => chosen,
1102 None => match self.remote_warm(workspace, task, warm, ¤t) {
1103 Some((record, branch)) => (record, format!("remote {branch}")),
1104 None => return Ok((restored, present)),
1105 },
1106 };
1107 let keep_mtimes = warm.preserve_mtimes && source == "local";
1108 restored.source = source;
1109 let mut noted = read_restored_files(workspace, warm);
1110 for location in missing {
1111 let Some(group) = record.groups.get(location.name) else {
1112 continue;
1113 };
1114 fs::create_dir_all(&location.base)?;
1115 let mut files = BTreeMap::new();
1116 for (path, artifact) in &group.artifacts {
1117 paths::safe_parents(&location.base, path)?;
1118 if !location.matches(path) {
1119 bail!("warm state for {} holds {path}, outside its paths", task.id);
1120 }
1121 let destination = location.base.join(path);
1122 if let Some(parent) = destination.parent() {
1123 fs::create_dir_all(parent)?;
1124 }
1125 match artifact {
1126 Artifact::File { blob, mode } => {
1127 let source = self.root.join("blobs").join(blob);
1128 if digest_file(&source)? != *blob {
1129 bail!("warm state for {} has a corrupt blob", task.id);
1130 }
1131 fs::copy(&source, &destination)?;
1132 let modified = keep_mtimes
1133 .then(|| group.mtimes.get(path))
1134 .flatten()
1135 .and_then(|nanos| u64::try_from(*nanos).ok())
1136 .map_or(RESTORED_AT, |nanos| {
1137 UNIX_EPOCH + Duration::from_nanos(nanos)
1138 });
1139 fs::File::options()
1140 .write(true)
1141 .open(&destination)?
1142 .set_modified(modified)?;
1143 set_mode(&destination, *mode)?;
1144 restored.files += 1;
1145 restored.bytes += fs::metadata(&destination)?.len();
1146 files.insert(
1147 path.clone(),
1148 (
1149 stamp(&fs::symlink_metadata(&destination)?),
1150 artifact.clone(),
1151 ),
1152 );
1153 }
1154 Artifact::Directory { mode } => {
1155 fs::create_dir_all(&destination)?;
1156 set_mode(&destination, *mode)?;
1157 }
1158 Artifact::Symlink { target, directory } => {
1159 validate_link(path, target)?;
1160 symlink(target, &destination, *directory)?;
1161 }
1162 }
1163 }
1164 noted.groups.insert(location.name.to_owned(), files);
1165 restored.groups.push(location.name.to_owned());
1166 }
1167 if !restored.groups.is_empty() {
1168 let path = restored_files_path(workspace, warm);
1169 fs::create_dir_all(path.parent().context("restore notes have a directory")?)?;
1170 fs::write(path, serde_json::to_vec(¬ed)?)?;
1171 }
1172 Ok((restored, present))
1173 }
1174
1175 fn remote_warm(
1178 &self,
1179 workspace: &Workspace,
1180 task: &Task,
1181 warm: &Warm,
1182 current: &[String],
1183 ) -> Option<(WarmRecord, String)> {
1184 let remote = self
1185 .remote
1186 .as_ref()
1187 .filter(|_| warm.remote && warm.portable)?;
1188 for branch in branches(workspace) {
1189 match remote.fetch_warm(&self.root, &warm.identity, &branch) {
1190 Ok(Some(bytes)) => {
1191 let Some(record) = WarmRecord::read(&bytes, &warm.identity).ok() else {
1192 continue;
1193 };
1194 if key_match(warm, current, &record.key).is_none() {
1197 continue;
1198 }
1199 if self.store_warm(&record).is_err() {
1200 return None;
1201 }
1202 return Some((record, branch));
1203 }
1204 Ok(None) => {}
1205 Err(error) => {
1206 qk_executor::status!(
1207 "qk: {}: remote warm state unavailable ({error:#})",
1208 task.id
1209 );
1210 continue;
1213 }
1214 }
1215 }
1216 None
1217 }
1218
1219 pub(crate) fn save_warm(
1222 &self,
1223 workspace: &Workspace,
1224 task: &Task,
1225 warm: &Warm,
1226 prepared: &qk_executor::PreparedTask,
1227 ) -> Result<()> {
1228 self.save_prepared(&Save::prepare(workspace, task, warm, prepared)?)
1229 }
1230
1231 pub(crate) fn save_warm_in_background(
1234 &self,
1235 workspace: &Workspace,
1236 task: &Task,
1237 warm: &Warm,
1238 prepared: &qk_executor::PreparedTask,
1239 ) -> Result<()> {
1240 let save = Save::prepare(workspace, task, warm, prepared)?;
1241 let cache = Cache {
1242 remote: self.remote.clone(),
1243 ..Cache::new(self.root.clone())
1244 };
1245 let handle = std::thread::spawn(move || {
1246 if let Err(error) = cache.save_prepared(&save) {
1247 qk_executor::status!("qk: {}: warm state not saved ({error:#})", save.task);
1248 }
1249 });
1250 self.saves.lock().unwrap().push(handle);
1251 Ok(())
1252 }
1253
1254 fn save_prepared(&self, save: &Save) -> Result<()> {
1259 let warm = &save.warm;
1260 let _lock = self.lock_warm(warm)?;
1261 let records = self.warm_records(warm);
1262 let previous = records
1264 .iter()
1265 .filter(|(_, record)| record.worktree == save.worktree)
1266 .min_by_key(|(_, record)| record.key != save.key)
1267 .map(|(_, record)| record);
1268 let noted: RestoredFiles = fs::read(&save.notes)
1269 .ok()
1270 .and_then(|bytes| serde_json::from_slice(&bytes).ok())
1271 .unwrap_or_default();
1272 let mut record = WarmRecord::new(
1273 warm.identity.clone(),
1274 save.worktree.clone(),
1275 save.commit.clone(),
1276 now_ms(),
1277 save.key.clone(),
1278 );
1279 let gone = |error: &std::io::Error| error.kind() == std::io::ErrorKind::NotFound;
1280 for location in &save.locations {
1281 if !location.base.exists() {
1282 continue;
1283 }
1284 let known = previous.and_then(|previous| previous.groups.get(location.name));
1285 let restored_files = noted.groups.get(location.name);
1286 let mut group = WarmGroup::default();
1287 let mut pending = Vec::new();
1288 for path in location.paths()? {
1289 paths::safe_parents(&location.base, &path)?;
1290 let absolute = location.base.join(&path);
1291 let metadata = match fs::symlink_metadata(&absolute) {
1292 Err(error) if gone(&error) => continue,
1293 result => result?,
1294 };
1295 let artifact = if metadata.file_type().is_symlink() {
1296 let target = match fs::read_link(&absolute) {
1297 Err(error) if gone(&error) => continue,
1298 result => result?,
1299 };
1300 let target = target
1301 .to_str()
1302 .context("symlink target must be UTF-8")?
1303 .to_owned();
1304 validate_link(&path, &target)?;
1305 Artifact::Symlink {
1306 target,
1307 directory: absolute.is_dir(),
1308 }
1309 } else if metadata.is_file() {
1310 let now = stamp(&metadata);
1311 let saved = known.and_then(|known| {
1312 (known.stamps.get(&path) == Some(&now))
1313 .then(|| known.artifacts.get(&path))
1314 .flatten()
1315 });
1316 let restored = restored_files
1317 .and_then(|files| files.get(&path))
1318 .filter(|(stamp, _)| *stamp == now)
1319 .map(|(_, artifact)| artifact);
1320 let reused = saved.or(restored).filter(|artifact| match artifact {
1321 Artifact::File { blob, .. } => self.root.join("blobs").join(blob).is_file(),
1322 _ => false,
1323 });
1324 let Some(artifact) = reused.cloned() else {
1325 pending.push((path, absolute, metadata));
1326 continue;
1327 };
1328 group.stamps.insert(path.clone(), now);
1329 if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1330 group.mtimes.insert(path.clone(), modified);
1331 }
1332 artifact
1333 } else if metadata.is_dir() {
1334 Artifact::Directory {
1335 mode: mode(&metadata),
1336 }
1337 } else {
1338 continue;
1339 };
1340 group.artifacts.insert(path, artifact);
1341 }
1342 let sources: Vec<_> = pending
1343 .iter()
1344 .map(|(_, absolute, _)| absolute.clone())
1345 .collect();
1346 for ((path, _, metadata), result) in pending.into_iter().zip(self.put_blobs(&sources)) {
1347 let blob = match result {
1348 Ok(blob) => blob,
1349 Err(error) if error.downcast_ref::<std::io::Error>().is_some_and(gone) => {
1350 continue;
1351 }
1352 Err(error) => return Err(error),
1353 };
1354 group.stamps.insert(path.clone(), stamp(&metadata));
1355 if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1356 group.mtimes.insert(path.clone(), modified);
1357 }
1358 group.artifacts.insert(
1359 path,
1360 Artifact::File {
1361 blob,
1362 mode: mode(&metadata),
1363 },
1364 );
1365 }
1366 if let Some(limit) = warm.max_size
1367 && group.size(self) > limit
1368 {
1369 qk_executor::status!(
1370 "qk: {}: warm {} is over its maxSize and was not saved",
1371 save.task,
1372 location.name
1373 );
1374 continue;
1375 }
1376 record.groups.insert(location.name.to_owned(), group);
1377 }
1378 self.store_warm(&record)?;
1379 let _ = fs::remove_file(&save.notes);
1380 let saved = self.warm_record(&record);
1383 for (path, _) in records
1384 .into_iter()
1385 .filter(|(path, _)| *path != saved)
1386 .skip(KEPT - 1)
1387 {
1388 let _ = fs::remove_file(path);
1389 }
1390 if warm.remote
1392 && warm.portable
1393 && let Some(remote) = &self.remote
1394 && let Some(branch) = &save.branch
1395 {
1396 for group in record.groups.values_mut() {
1397 group.stamps.clear();
1398 }
1399 remote.upload_warm(
1400 &self.root,
1401 &warm.identity,
1402 branch,
1403 serde_json::to_vec(&record)?,
1404 );
1405 }
1406 Ok(())
1407 }
1408}
1409
1410struct Save {
1413 task: String,
1414 warm: Warm,
1415 locations: Vec<Location>,
1416 worktree: String,
1417 commit: Option<String>,
1418 branch: Option<String>,
1419 key: Vec<String>,
1420 notes: PathBuf,
1421}
1422
1423impl Save {
1424 fn prepare(
1426 workspace: &Workspace,
1427 task: &Task,
1428 warm: &Warm,
1429 prepared: &qk_executor::PreparedTask,
1430 ) -> Result<Self> {
1431 Ok(Self {
1432 task: task.id.clone(),
1433 warm: warm.clone(),
1434 locations: locations(workspace, task, warm)?,
1435 worktree: worktree(workspace),
1436 commit: current_commit(workspace),
1437 branch: current_branch(workspace),
1438 key: key_digests(workspace, warm, prepared)?,
1439 notes: restored_files_path(workspace, warm),
1440 })
1441 }
1442}
1443
1444#[cfg(test)]
1445mod tests {
1446 use super::*;
1447 use qk_taskgraph::Request;
1448 use serde_json::json;
1449
1450 fn fixture() -> (tempfile::TempDir, Workspace, TaskGraph) {
1452 let temp = tempfile::tempdir().unwrap();
1453 fs::write(temp.path().join("nx.json"), "{}").unwrap();
1454 fs::write(
1455 temp.path().join("project.json"),
1456 json!({
1457 "name": "app", "targets": {
1458 "one": {"command": "true", "qk:warm": {"paths": ["scratch-one"]}},
1459 "two": {"command": "true", "qk:warm": {"paths": ["scratch-two"]}}
1460 }
1461 })
1462 .to_string(),
1463 )
1464 .unwrap();
1465 for path in ["scratch-one", "scratch-two"] {
1466 fs::create_dir(temp.path().join(path)).unwrap();
1467 fs::write(temp.path().join(path).join("state"), "saved").unwrap();
1468 }
1469 let workspace = Workspace::load(temp.path()).unwrap();
1470 let graph = TaskGraph::build(
1471 &workspace,
1472 &[
1473 Request::parse("app:one").unwrap(),
1474 Request::parse("app:two").unwrap(),
1475 ],
1476 )
1477 .unwrap();
1478 (temp, workspace, graph)
1479 }
1480
1481 #[test]
1483 fn waiting_for_an_owner_does_not_block_other_releases() {
1484 let (_temp, workspace, graph) = fixture();
1485 let one = &graph.tasks["app:one"];
1486 let two = &graph.tasks["app:two"];
1487 let first = acquire(&workspace, one, &config(&workspace, one).unwrap().unwrap()).unwrap();
1488 let state = paths::worktree_state(&workspace.root).join("warm-kept");
1489 let lock = fs::OpenOptions::new()
1490 .create(true)
1491 .truncate(false)
1492 .read(true)
1493 .write(true)
1494 .open(state.join(format!("{}.lock", hash32(&two.id))))
1495 .unwrap();
1496 fs2::FileExt::lock_exclusive(&lock).unwrap();
1497 let (released, received) = std::sync::mpsc::channel();
1498 let completed = std::thread::scope(|scope| {
1499 scope.spawn(|| {
1500 let second =
1501 acquire(&workspace, two, &config(&workspace, two).unwrap().unwrap()).unwrap();
1502 release(&second);
1503 });
1504 std::thread::sleep(Duration::from_millis(100));
1505 scope.spawn(|| {
1506 release(&first);
1507 released.send(()).unwrap();
1508 });
1509 let completed = received.recv_timeout(Duration::from_secs(2)).is_ok();
1510 fs2::FileExt::unlock(&lock).unwrap();
1511 completed
1512 });
1513 assert!(
1514 completed,
1515 "a blocked owner lock prevented an unrelated release"
1516 );
1517 assert_eq!(
1518 fs::read_to_string(workspace.root.join("scratch-one/state")).unwrap(),
1519 "saved"
1520 );
1521 }
1522
1523 #[test]
1525 fn unreadable_ownership_preserves_saved_paths() {
1526 let (_temp, workspace, graph) = fixture();
1527 let owner = &graph.tasks["app:one"];
1528 let warm = config(&workspace, owner).unwrap().unwrap();
1529 let root = acquire(&workspace, owner, &warm).unwrap();
1530 let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1531 drop(entry); fs::write(Stash::manifest(&root), "{").unwrap();
1533 assert!(acquire(&workspace, owner, &warm).is_err());
1534 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1535 fs::remove_file(Stash::manifest(&root)).unwrap();
1536 assert!(acquire(&workspace, owner, &warm).is_err());
1537 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1538 fs::create_dir(Stash::manifest(&root)).unwrap();
1539 assert!(acquire(&workspace, owner, &warm).is_err());
1540 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1541 }
1542
1543 #[cfg(unix)]
1545 #[test]
1546 fn inspection_errors_keep_recovery_incomplete() {
1547 let (_temp, workspace, graph) = fixture();
1548 let owner = &graph.tasks["app:one"];
1549 let warm = config(&workspace, owner).unwrap().unwrap();
1550 let root = acquire(&workspace, owner, &warm).unwrap();
1551 let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1552 let saved = root.with_extension("saved");
1553 fs::rename(&root, &saved).unwrap();
1554 fs::write(&root, "blocked").unwrap(); assert!(!entry.stash.put_back());
1556 let retained = Stash {
1557 root: root.clone(),
1558 workspace: workspace.root.clone(),
1559 moved: entry.stash.moved.clone(),
1560 };
1561 assert!(stash(&workspace, &root, &warm, Some(retained)).is_err());
1562 assert_eq!(fs::read_to_string(saved.join("0/state")).unwrap(), "saved");
1563 assert_eq!(fs::read_to_string(&root).unwrap(), "blocked");
1564 }
1565}