1use std::collections::{BTreeMap, BTreeSet};
25use std::fs;
26use std::io::Write;
27use std::path::PathBuf;
28use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
29use std::time::{Duration, SystemTime, UNIX_EPOCH};
30
31use anyhow::{Context, Result, bail};
32use qk_config::Workspace;
33use qk_taskgraph::{Task, TaskGraph};
34use serde::{Deserialize, Serialize};
35use serde_json::Value;
36
37use crate::paths::{self, Outputs};
38use crate::record::{self, Artifact, WarmGroup, WarmRecord, validate_link};
39use crate::store::{copy_blob_contents, mode, set_mode, set_modified, symlink};
40use crate::{Cache, hash::digest_file, profile};
41
42pub const RESTORED_AT: std::time::SystemTime = std::time::UNIX_EPOCH;
47
48const GROUP: &str = "group ";
50
51pub const FIELDS: &[&str] = &[
55 "outputs",
56 "paths",
57 "env",
58 "maxSize",
59 "remote",
60 "mtimes",
61 "portable",
62 "key",
63 "restoreKeys",
64 "group",
65 "survive",
66 "save",
67 "restore",
68];
69
70const KEPT: usize = 8;
72const RESTORERS: usize = 8;
74
75type PendingFile<'a> = (&'a String, &'a str, u32, PathBuf, SystemTime);
77type RestoredFile = (Vec<i64>, u64);
79
80#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
83pub enum Portable {
84 #[default]
86 Own,
87 Path,
92 Anywhere,
94}
95
96#[derive(Clone, Debug, Default)]
98pub struct Warm {
99 pub identity: String,
101 pub outputs: bool,
102 pub paths: Vec<String>,
104 pub env: BTreeMap<String, String>,
106 pub directory: bool,
108 pub max_size: Option<u64>,
109 pub remote: bool,
110 pub preserve_mtimes: bool,
113 pub portable: Portable,
115 pub key: Vec<KeyPart>,
117 pub restore_keys: Option<usize>,
120 pub survive: Vec<String>,
123 pub background: bool,
126 pub each_path: bool,
129}
130
131#[derive(Clone, Debug)]
133pub enum KeyPart {
134 File(String),
136 Env(String),
138}
139
140pub fn config(workspace: &Workspace, task: &Task) -> Result<Vec<Warm>> {
146 const SHAPE: &str = "qk:warm must be an object or an array of objects";
147 let objects = match task.definition.extra.get("qk:warm") {
148 None => return Ok(Vec::new()),
149 Some(Value::Object(object)) => vec![object],
150 Some(Value::Array(entries)) => entries
151 .iter()
152 .map(|entry| entry.as_object().context(SHAPE))
153 .collect::<Result<_>>()?,
154 Some(_) => bail!(SHAPE),
155 };
156 let mut entries: Vec<Warm> = Vec::new();
157 for object in objects {
158 let warm = entry(workspace, task, object)?;
159 if entries.iter().any(|other| other.identity == warm.identity) {
160 bail!(match warm.group() {
161 Some(group) => format!("qk:warm names group {group:?} twice"),
162 None => "only one qk:warm entry may go without a group".to_owned(),
163 });
164 }
165 for name in warm.env.keys() {
166 if entries.iter().any(|other| other.env.contains_key(name)) {
167 bail!("qk:warm sets {name} in more than one entry");
168 }
169 }
170 entries.push(warm);
171 }
172 Ok(entries)
173}
174
175fn entry(
177 workspace: &Workspace,
178 task: &Task,
179 object: &serde_json::Map<String, Value>,
180) -> Result<Warm> {
181 let flag = |name: &str, default: bool| -> Result<bool> {
182 object.get(name).map_or(Ok(default), |value| {
183 value
184 .as_bool()
185 .with_context(|| format!("qk:warm.{name} must be true or false"))
186 })
187 };
188 let group = object
189 .get("group")
190 .map(|value| {
191 value
192 .as_str()
193 .filter(|name| !name.is_empty())
194 .context("qk:warm.group must be a name")
195 })
196 .transpose()?;
197 if group.is_some()
199 && ["outputs", "paths", "survive"]
200 .iter()
201 .any(|key| object.contains_key(*key))
202 {
203 bail!(
204 "a qk:warm.group shares {{warm}} alone; outputs, paths and survive belong to one task"
205 );
206 }
207 let identity = group.map_or_else(|| task.id.clone(), |group| format!("{GROUP}{group}"));
208 const SURVIVE_SHAPE: &str = "qk:warm.survive must be an array of target names";
209 let survive = object
210 .get("survive")
211 .map(|value| {
212 value
213 .as_array()
214 .context(SURVIVE_SHAPE)?
215 .iter()
216 .map(|target| {
217 target
218 .as_str()
219 .filter(|target| !target.is_empty())
220 .map(str::to_owned)
221 .context(SURVIVE_SHAPE)
222 })
223 .collect::<Result<Vec<_>>>()
224 })
225 .transpose()?
226 .unwrap_or_default();
227 let directory = directory(workspace, &identity);
228 let directory_text = directory.to_str().context("warm directory must be UTF-8")?;
229 let workspace_text = workspace
230 .root
231 .to_str()
232 .context("workspace root must be UTF-8")?;
233 let project_root = &workspace.projects[&task.project].root;
234 let project_directory = if project_root == "." {
235 workspace.root.clone()
236 } else {
237 workspace.root.join(project_root)
238 };
239 let project_text = project_directory
240 .to_str()
241 .context("project root must be UTF-8")?;
242 let expand = |text: &str| -> Result<String> {
243 if text.contains("{options.") || text.contains("{args.") {
244 bail!("dynamic warm environment values are not supported yet");
245 }
246 Ok(text
247 .replace("{workspaceRoot}", workspace_text)
248 .replace("{projectRoot}", project_text)
249 .replace("{warm}", directory_text))
250 };
251 let mut paths = Vec::new();
252 for path in object
253 .get("paths")
254 .map(|value| {
255 value
256 .as_array()
257 .context("qk:warm.paths must be an array of paths")
258 })
259 .transpose()?
260 .into_iter()
261 .flatten()
262 {
263 let path = path
264 .as_str()
265 .context("qk:warm.paths must be an array of paths")?;
266 if path.contains("{warm}") {
267 bail!("qk:warm.paths are workspace paths; {{warm}} is kept already");
268 }
269 let (excluded, path) = match path.strip_prefix('!') {
270 Some(path) => (true, path),
271 None => (false, path),
272 };
273 dynamic(path)?;
274 let path = paths::expand(workspace, &task.project, path)?
275 .trim_end_matches('/')
276 .to_owned();
277 paths::validate_path(&path)?;
278 paths.push(if excluded { format!("!{path}") } else { path });
279 }
280 let mut env = BTreeMap::new();
281 let mut uses_directory = false;
282 if let Some(values) = object.get("env") {
283 for (name, value) in values
284 .as_object()
285 .context("qk:warm.env must be an object")?
286 {
287 let value = value
288 .as_str()
289 .context("qk:warm.env values must be strings")?;
290 uses_directory |= value.contains("{warm}");
291 env.insert(name.clone(), expand(value)?);
292 }
293 }
294 let max_size = match object.get("maxSize") {
295 None => None,
296 Some(Value::String(text)) => Some(crate::parse_size(text)?),
297 Some(Value::Number(number)) => number.as_u64(),
298 Some(_) => bail!("qk:warm.maxSize must be a size such as \"2GB\""),
299 };
300 let preserve_mtimes = match object.get("mtimes").map(|value| value.as_str()) {
301 None | Some(Some("epoch")) => false,
302 Some(Some("preserve")) => true,
303 Some(_) => bail!("qk:warm.mtimes must be \"epoch\" or \"preserve\""),
304 };
305 const KEY_SHAPE: &str = "qk:warm.key must be an array of paths and {\"env\": name} objects";
306 let mut key = Vec::new();
307 for part in object
308 .get("key")
309 .map(|value| value.as_array().context(KEY_SHAPE))
310 .transpose()?
311 .into_iter()
312 .flatten()
313 {
314 key.push(match part {
315 Value::String(path) => {
316 dynamic(path)?;
317 let path = paths::expand(workspace, &task.project, path)?;
318 paths::validate_path(&path)?;
319 KeyPart::File(path)
320 }
321 Value::Object(object) if object.len() == 1 => KeyPart::Env(
322 object
323 .get("env")
324 .and_then(Value::as_str)
325 .context(KEY_SHAPE)?
326 .to_owned(),
327 ),
328 _ => bail!(KEY_SHAPE),
329 });
330 }
331 let restore_keys = object
332 .get("restoreKeys")
333 .map(|value| {
334 value
335 .as_u64()
336 .and_then(|count| usize::try_from(count).ok())
337 .filter(|count| *count <= key.len())
338 .context("qk:warm.restoreKeys must be a count no larger than qk:warm.key")
339 })
340 .transpose()?;
341 let each_path = match object.get("restore").map(Value::as_str) {
342 None | Some(Some("whole")) => false,
343 Some(Some("each-path")) if !paths.is_empty() => true,
344 Some(Some("each-path")) => bail!("qk:warm.restore \"each-path\" needs qk:warm.paths"),
345 Some(_) => bail!("qk:warm.restore must be \"whole\" or \"each-path\""),
346 };
347 let background = match object.get("save").map(Value::as_str) {
348 None | Some(Some("wait")) => false,
349 Some(Some("background")) => true,
350 Some(_) => bail!("qk:warm.save must be \"wait\" or \"background\""),
351 };
352 Ok(Warm {
353 identity,
354 outputs: flag("outputs", false)?,
355 paths,
356 env,
357 directory: uses_directory,
358 max_size,
359 remote: flag("remote", true)?,
360 preserve_mtimes,
361 portable: match object.get("portable") {
362 None | Some(Value::Bool(false)) => Portable::Own,
363 Some(Value::Bool(true)) => Portable::Anywhere,
364 Some(Value::String(path)) if path == "path" => Portable::Path,
365 Some(_) => bail!("qk:warm.portable must be true, false or \"path\""),
366 },
367 key,
368 restore_keys,
369 survive,
370 background,
371 each_path,
372 })
373}
374
375fn dynamic(path: &str) -> Result<()> {
379 if path.contains("{options.") || path.contains("{args.") {
380 bail!("dynamic cache paths are not supported yet");
381 }
382 Ok(())
383}
384
385fn directory(workspace: &Workspace, identity: &str) -> PathBuf {
386 paths::worktree_state(&workspace.root)
387 .join("warm")
388 .join(&blake3::hash(identity.as_bytes()).to_hex()[..32])
389}
390
391impl Warm {
392 pub fn group(&self) -> Option<&str> {
394 self.identity.strip_prefix(GROUP)
395 }
396
397 pub fn kept_paths(&self) -> Vec<String> {
400 self.paths
401 .iter()
402 .filter(|path| !path.starts_with('!'))
403 .cloned()
404 .collect()
405 }
406}
407
408pub fn kept_paths(entries: &[Warm]) -> Vec<String> {
410 entries.iter().flat_map(Warm::kept_paths).collect()
411}
412
413#[derive(Default)]
416pub struct Kept {
417 roots: Vec<PathBuf>,
419}
420
421impl Drop for Kept {
422 fn drop(&mut self) {
423 for root in self.roots.iter().rev() {
424 release(root);
425 }
426 }
427}
428
429#[derive(Default, Serialize, Deserialize)]
431struct Stash {
432 #[serde(skip)]
433 root: PathBuf,
434 #[serde(skip)]
435 workspace: PathBuf,
436 moved: Vec<String>,
438}
439
440struct Active {
443 users: usize,
444 stash: Stash,
445 _lock: fs::File,
446}
447
448static ACTIVE: std::sync::Mutex<BTreeMap<PathBuf, Active>> = std::sync::Mutex::new(BTreeMap::new());
451
452impl Stash {
453 fn manifest(root: &std::path::Path) -> PathBuf {
455 root.join("moved.json")
456 }
457
458 fn save_manifest(&self) -> Result<()> {
461 let mut file = tempfile::NamedTempFile::new_in(&self.root)?;
462 file.write_all(&serde_json::to_vec(self)?)?;
463 file.as_file().sync_all()?;
464 file.persist(Self::manifest(&self.root))?;
465 Ok(())
466 }
467
468 fn put_back(&self) -> bool {
474 let mut complete = true;
475 for (index, path) in self.moved.iter().enumerate().rev() {
476 let kept = self.root.join(index.to_string());
477 match fs::symlink_metadata(&kept) {
478 Ok(_) => {}
479 Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
480 Err(error) => {
481 complete = false;
482 qk_executor::status!(
483 "qk: could not inspect saved {path}, kept for the next run ({error})"
484 );
485 continue;
486 }
487 }
488 let original = self.workspace.join(path);
489 let result = paths::safe_parents(&self.workspace, path).and_then(|()| {
490 remove(&original)?;
491 if let Some(parent) = original.parent() {
492 fs::create_dir_all(parent)?;
493 }
494 fs::rename(&kept, &original)?;
495 Ok(())
496 });
497 if let Err(error) = result {
498 complete = false;
499 qk_executor::status!(
500 "qk: could not put {path} back, kept for the next run ({error:#})"
501 );
502 }
503 }
504 if complete {
505 let _ = fs::remove_dir_all(&self.root);
506 }
507 complete
508 }
509}
510
511fn remove(path: &std::path::Path) -> std::io::Result<()> {
513 match fs::symlink_metadata(path) {
514 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
515 Err(error) => Err(error),
516 Ok(metadata) if metadata.is_dir() => fs::remove_dir_all(path),
517 Ok(_) => fs::remove_file(path),
518 }
519}
520
521fn release(root: &std::path::Path) {
524 let mut active = ACTIVE
525 .lock()
526 .unwrap_or_else(std::sync::PoisonError::into_inner);
527 let Some(entry) = active.get_mut(root) else {
528 return;
529 };
530 entry.users -= 1;
531 if entry.users > 0 {
532 return;
533 }
534 let entry = active.remove(root).expect("present above");
535 drop(active);
538 entry.stash.put_back();
539}
540
541fn names(owner: &Task, entry: &str, task: &Task) -> bool {
543 match entry.split_once(':') {
544 Some((project, target)) => task.project == project && task.target == target,
545 None => task.project == owner.project && task.target == entry,
546 }
547}
548
549fn depends_on(graph: &TaskGraph, from: &str, on: &str) -> bool {
551 let mut seen = std::collections::BTreeSet::new();
552 let mut next = vec![from];
553 while let Some(id) = next.pop() {
554 for dependency in graph
555 .tasks
556 .get(id)
557 .into_iter()
558 .flat_map(|task| &task.dependencies)
559 {
560 if dependency == on {
561 return true;
562 }
563 if seen.insert(dependency.as_str()) {
564 next.push(dependency);
565 }
566 }
567 }
568 false
569}
570
571pub fn keep_across(workspace: &Workspace, graph: &TaskGraph, task: &Task) -> Kept {
574 let mut kept = Kept::default();
575 for owner in graph.tasks.values() {
576 let Ok(entries) = config(workspace, owner) else {
577 continue;
578 };
579 let Some(warm) = entries.into_iter().find(|warm| warm.group().is_none()) else {
581 continue;
582 };
583 if !warm.survive.iter().any(|entry| names(owner, entry, task))
584 || !depends_on(graph, &owner.id, &task.id)
585 {
586 continue;
587 }
588 match acquire(workspace, owner, &warm) {
589 Ok(root) => kept.roots.push(root),
590 Err(error) => qk_executor::status!(
591 "qk: {}: warm paths not kept across {} ({error:#})",
592 owner.id,
593 task.id
594 ),
595 }
596 }
597 kept
598}
599
600fn acquire(workspace: &Workspace, owner: &Task, warm: &Warm) -> Result<PathBuf> {
603 let state = paths::worktree_state(&workspace.root).join("warm-kept");
604 let root = state.join(hash32(&owner.id));
605 fs::create_dir_all(&state)?;
606 let lock = fs::OpenOptions::new()
607 .create(true)
608 .truncate(false)
609 .read(true)
610 .write(true)
611 .open(state.join(format!("{}.lock", hash32(&owner.id))))?;
612 let mut active = loop {
615 let mut active = ACTIVE
616 .lock()
617 .unwrap_or_else(std::sync::PoisonError::into_inner);
618 if let Some(entry) = active.get_mut(&root) {
619 entry.users += 1;
620 return Ok(root);
621 }
622 match fs2::FileExt::try_lock_exclusive(&lock) {
623 Ok(()) => break active,
624 Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => {
625 drop(active);
626 std::thread::sleep(Duration::from_millis(20));
627 }
628 Err(error) => return Err(error.into()),
629 }
630 };
631 let left = match fs::read(Stash::manifest(&root)) {
638 Ok(bytes) => {
639 let mut left: Stash = serde_json::from_slice(&bytes)
640 .context("invalid warm stash manifest; saved paths retained for recovery")?;
641 left.root = root.clone();
642 left.workspace = workspace.root.clone();
643 Some(left)
644 }
645 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
646 if root.is_dir() && fs::read_dir(&root)?.next().is_some() {
647 bail!("missing warm stash manifest; saved paths retained for recovery");
648 }
649 None
650 }
651 Err(error) => return Err(error).context("cannot read warm stash; saved paths retained"),
652 };
653 let retained = left.filter(|left| !left.put_back());
654 let stash = stash(workspace, &root, warm, retained)?;
655 active.insert(
656 root.clone(),
657 Active {
658 users: 1,
659 stash,
660 _lock: lock,
661 },
662 );
663 Ok(root)
664}
665
666fn stash(
670 workspace: &Workspace,
671 root: &std::path::Path,
672 warm: &Warm,
673 retained: Option<Stash>,
674) -> Result<Stash> {
675 let mut stash = match retained {
676 Some(retained) => retained,
677 None => {
678 let _ = fs::remove_dir_all(root);
679 Stash {
680 root: root.to_owned(),
681 workspace: workspace.root.clone(),
682 moved: Vec::new(),
683 }
684 }
685 };
686 let held: Vec<String> = stash
687 .moved
688 .iter()
689 .enumerate()
690 .filter_map(
691 |(index, path)| match fs::symlink_metadata(root.join(index.to_string())) {
692 Ok(_) => Some(Ok(path.clone())),
693 Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
694 Err(error) => Some(Err(error).context("cannot inspect retained warm path")),
695 },
696 )
697 .collect::<Result<_>>()?;
698 let overlaps = |path: &str| {
699 held.iter().any(|held| {
700 held == path
701 || path.starts_with(&format!("{held}/"))
702 || held.starts_with(&format!("{path}/"))
703 })
704 };
705 let mut kept = std::collections::BTreeSet::new();
708 for path in warm.kept_paths() {
709 if let Ok(found) = Outputs::from_paths(std::slice::from_ref(&path))
710 .and_then(|outputs| outputs.paths(&workspace.root))
711 {
712 kept.extend(found);
713 }
714 }
715 let topmost: Vec<String> = kept
717 .iter()
718 .filter(|path| {
719 !std::path::Path::new(path.as_str())
720 .ancestors()
721 .skip(1)
722 .any(|ancestor| {
723 ancestor
724 .to_str()
725 .is_some_and(|ancestor| kept.contains(ancestor))
726 })
727 })
728 .filter(|path| !overlaps(path))
729 .cloned()
730 .collect();
731 if topmost.is_empty() {
732 return Ok(stash);
733 }
734 fs::create_dir_all(root)?;
735 for path in topmost {
736 let moved = paths::safe_parents(&workspace.root, &path).and_then(|()| {
737 stash.moved.push(path.clone());
738 stash.save_manifest()?;
739 fs::rename(
742 workspace.root.join(&path),
743 root.join((stash.moved.len() - 1).to_string()),
744 )
745 .inspect_err(|_| {
746 stash.moved.pop();
747 })?;
748 Ok(())
749 });
750 if let Err(error) = moved {
751 stash.save_manifest()?;
754 stash.put_back();
755 return Err(error);
756 }
757 }
758 Ok(stash)
759}
760
761pub fn check_overlaps(workspace: &Workspace, graph: &TaskGraph) -> Result<()> {
764 let mut claimed: Vec<(String, &str)> = Vec::new();
765 for (id, task) in &graph.tasks {
766 for path in kept_paths(&config(workspace, task)?) {
767 if let Some((_, other)) = claimed.iter().find(|(known, _)| {
768 known == &path
769 || known.starts_with(&format!("{path}/"))
770 || path.starts_with(&format!("{known}/"))
771 }) {
772 bail!("{id} and {other} both keep {path} as warm state");
773 }
774 claimed.push((path, id));
775 }
776 }
777 Ok(())
778}
779
780impl WarmGroup {
781 fn size(&self, cache: &Cache) -> u64 {
783 self.artifacts
784 .values()
785 .filter_map(|artifact| match artifact {
786 Artifact::File { blob, .. } => {
787 fs::metadata(cache.root.join("blobs").join(blob)).ok()
788 }
789 _ => None,
790 })
791 .map(|metadata| metadata.len())
792 .sum()
793 }
794}
795
796#[derive(Default, Serialize, Deserialize)]
799struct RestoredFiles {
800 groups: BTreeMap<String, GroupFiles>,
801}
802
803type GroupFiles = BTreeMap<String, (Vec<i64>, Artifact)>;
805
806#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
808pub struct Restored {
809 #[serde(default, skip_serializing_if = "Option::is_none")]
811 pub group: Option<String>,
812 pub source: String,
815 pub groups: Vec<String>,
816 pub files: usize,
817 pub bytes: u64,
818}
819
820#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
822#[serde(rename_all = "camelCase")]
823pub struct WarmReport {
824 #[serde(default, skip_serializing_if = "Vec::is_empty")]
826 pub restored: Vec<Restored>,
827 #[serde(default, skip_serializing_if = "Vec::is_empty")]
831 pub present: Vec<String>,
832 pub save_ms: Option<u64>,
835 #[serde(default, skip_serializing_if = "std::ops::Not::not")]
837 pub background: bool,
838}
839
840#[derive(Clone)]
843struct Location {
844 name: &'static str,
845 base: PathBuf,
846 outputs: Option<Outputs>,
848 each: Vec<Outputs>,
851}
852
853impl Location {
854 fn paths(&self) -> Result<std::collections::BTreeSet<String>> {
856 if !self.base.exists() {
857 return Ok(Default::default());
858 }
859 match &self.outputs {
860 Some(outputs) => outputs.paths(&self.base),
861 None => {
862 let mut paths = std::collections::BTreeSet::new();
863 for entry in walkdir::WalkDir::new(&self.base)
864 .min_depth(1)
865 .follow_links(false)
866 {
867 let entry = entry?;
868 paths.insert(paths::relative(&self.base, entry.path())?);
869 }
870 Ok(paths)
871 }
872 }
873 }
874
875 fn matches(&self, path: &str) -> bool {
877 self.outputs
878 .as_ref()
879 .is_none_or(|outputs| outputs.matches(path))
880 }
881
882 fn present(&self) -> Result<Option<Vec<&Outputs>>> {
885 let mut present = Vec::new();
886 for outputs in &self.each {
887 if !outputs.paths(&self.base)?.is_empty() {
888 present.push(outputs);
889 }
890 }
891 Ok((!present.is_empty() && present.len() < self.each.len()).then_some(present))
892 }
893}
894
895fn remove_restored(location: &Location, kept: &[&Outputs]) {
899 let Ok(paths) = location.paths() else {
900 return;
901 };
902 for path in paths.iter().rev() {
903 if kept.iter().any(|outputs| outputs.matches(path)) {
904 continue;
905 }
906 let absolute = location.base.join(path);
907 let _ = match fs::symlink_metadata(&absolute) {
908 Ok(metadata) if metadata.is_dir() => fs::remove_dir(&absolute),
909 Ok(_) => fs::remove_file(&absolute),
910 Err(_) => continue,
911 };
912 }
913}
914
915fn locations(workspace: &Workspace, task: &Task, warm: &Warm) -> Result<Vec<Location>> {
917 let mut locations = Vec::new();
918 if warm.outputs {
919 locations.push(Location {
920 name: "outputs",
921 base: workspace.root.clone(),
922 outputs: Some(Outputs::new(workspace, task)?),
923 each: Vec::new(),
924 });
925 }
926 if !warm.paths.is_empty() {
927 let mut each = Vec::new();
928 if warm.each_path {
929 let (excluded, included): (Vec<_>, Vec<_>) =
930 warm.paths.iter().partition(|path| path.starts_with('!'));
931 for path in included {
932 let mut patterns = vec![path.clone()];
933 patterns.extend(excluded.iter().map(|path| (*path).clone()));
934 each.push(Outputs::from_paths(&patterns)?);
935 }
936 }
937 locations.push(Location {
938 name: "paths",
939 base: workspace.root.clone(),
940 outputs: Some(Outputs::from_paths(&warm.paths)?),
941 each,
942 });
943 }
944 if warm.directory {
945 locations.push(Location {
946 name: "directory",
947 base: directory(workspace, &warm.identity),
948 outputs: None,
949 each: Vec::new(),
950 });
951 }
952 Ok(locations)
953}
954
955pub(crate) fn current_branch(workspace: &Workspace) -> Option<String> {
958 for name in ["GITHUB_HEAD_REF", "GITHUB_REF_NAME"] {
959 if let Ok(value) = std::env::var(name)
960 && !value.is_empty()
961 {
962 return Some(value);
963 }
964 }
965 let output = paths::git(&workspace.root, &["rev-parse", "--abbrev-ref", "HEAD"]).ok()?;
966 let branch = String::from_utf8(output.stdout).ok()?.trim().to_owned();
967 (output.status.success() && !branch.is_empty() && branch != "HEAD").then_some(branch)
968}
969
970pub(crate) fn branches(workspace: &Workspace) -> Vec<String> {
973 let default = workspace
974 .config
975 .default_base
976 .clone()
977 .unwrap_or_else(|| "main".into());
978 let default = default
979 .strip_prefix("origin/")
980 .unwrap_or(&default)
981 .to_owned();
982 let mut branches: Vec<String> = current_branch(workspace).into_iter().collect();
983 if !branches.contains(&default) {
984 branches.push(default);
985 }
986 branches
987}
988
989fn stamp(metadata: &fs::Metadata) -> Vec<i64> {
991 let modified = metadata
992 .modified()
993 .ok()
994 .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
995 .map_or(-1, |duration| duration.as_nanos() as i64);
996 #[allow(unused_mut)]
997 let mut stamp = vec![metadata.len() as i64, modified, i64::from(mode(metadata))];
998 #[cfg(unix)]
999 {
1000 use std::os::unix::fs::MetadataExt;
1001 stamp.extend([
1002 metadata.ctime(),
1003 metadata.ctime_nsec(),
1004 metadata.ino() as i64,
1005 ]);
1006 }
1007 stamp
1008}
1009
1010fn remote_identity(warm: &Warm, worktree: &str) -> String {
1015 match warm.portable {
1016 Portable::Path => format!("{}\0path\0{worktree}", warm.identity),
1017 _ => warm.identity.clone(),
1018 }
1019}
1020
1021fn worktree(workspace: &Workspace) -> String {
1022 fs::canonicalize(&workspace.root)
1023 .unwrap_or_else(|_| workspace.root.clone())
1024 .to_string_lossy()
1025 .into_owned()
1026}
1027
1028fn current_commit(workspace: &Workspace) -> Option<String> {
1030 let output = paths::git(&workspace.root, &["rev-parse", "HEAD"]).ok()?;
1031 let commit = String::from_utf8(output.stdout).ok()?.trim().to_owned();
1032 (output.status.success() && !commit.is_empty()).then_some(commit)
1033}
1034
1035fn key_digests(
1038 workspace: &Workspace,
1039 warm: &Warm,
1040 prepared: &qk_executor::PreparedTask,
1041) -> Result<Vec<String>> {
1042 warm.key
1043 .iter()
1044 .map(|part| {
1045 Ok(match part {
1046 KeyPart::File(path) => {
1047 paths::safe_parents(&workspace.root, path)?;
1048 let absolute = workspace.root.join(path);
1049 if absolute.is_file() {
1050 format!("file:{}", digest_file(&absolute)?)
1051 } else {
1052 "file:absent".to_owned()
1053 }
1054 }
1055 KeyPart::Env(name) => match prepared
1058 .execution
1059 .get(std::ffi::OsStr::new(name))
1060 .or_else(|| prepared.env.get(std::ffi::OsStr::new(name)))
1061 {
1062 Some(value) => format!(
1063 "env:{}",
1064 blake3::hash(value.as_encoded_bytes()).to_hex()[..32].to_owned()
1065 ),
1066 None => "env:unset".to_owned(),
1067 },
1068 })
1069 })
1070 .collect()
1071}
1072
1073fn key_match(warm: &Warm, current: &[String], saved: &[String]) -> Option<u8> {
1077 if saved == current {
1078 return Some(0);
1079 }
1080 let count = warm.restore_keys?;
1081 (saved.len() == current.len() && saved[..count] == current[..count]).then_some(1)
1082}
1083
1084fn descends_from(workspace: &Workspace, commit: &str) -> bool {
1086 paths::git(
1087 &workspace.root,
1088 &["merge-base", "--is-ancestor", commit, "HEAD"],
1089 )
1090 .is_ok_and(|output| output.status.success())
1091}
1092
1093fn behind(workspace: &Workspace, commit: &str) -> Option<u64> {
1096 let ancestor = paths::git(
1097 &workspace.root,
1098 &["merge-base", "--is-ancestor", commit, "HEAD"],
1099 )
1100 .ok()?;
1101 if !ancestor.status.success() {
1102 return None;
1103 }
1104 let output = paths::git(
1105 &workspace.root,
1106 &["rev-list", "--count", &format!("{commit}..HEAD")],
1107 )
1108 .ok()?;
1109 String::from_utf8(output.stdout).ok()?.trim().parse().ok()
1110}
1111
1112fn hash32(text: &str) -> String {
1114 blake3::hash(text.as_bytes()).to_hex()[..32].to_owned()
1115}
1116
1117fn nanos(time: SystemTime) -> Option<i64> {
1119 time.duration_since(UNIX_EPOCH)
1120 .ok()
1121 .and_then(|duration| i64::try_from(duration.as_nanos()).ok())
1122}
1123
1124fn now_ms() -> u64 {
1126 SystemTime::now()
1127 .duration_since(UNIX_EPOCH)
1128 .map_or(0, |duration| duration.as_millis() as u64)
1129}
1130
1131fn restored_files_path(workspace: &Workspace, warm: &Warm) -> PathBuf {
1133 paths::worktree_state(&workspace.root)
1134 .join("warm-restored")
1135 .join(format!("{}.json", hash32(&warm.identity)))
1136}
1137
1138fn read_restored_files(workspace: &Workspace, warm: &Warm) -> RestoredFiles {
1140 fs::read(restored_files_path(workspace, warm))
1141 .ok()
1142 .and_then(|bytes| serde_json::from_slice(&bytes).ok())
1143 .unwrap_or_default()
1144}
1145
1146impl Cache {
1147 fn warm_prefix(&self, identity: &str) -> String {
1149 hash32(&format!(
1150 "{identity}\0{}\0{}",
1151 std::env::consts::OS,
1152 std::env::consts::ARCH
1153 ))
1154 }
1155
1156 fn lock_warm(&self, warm: &Warm) -> Result<fs::File> {
1159 let directory = self.root.join("locks");
1160 fs::create_dir_all(&directory)?;
1161 let file = fs::OpenOptions::new()
1162 .create(true)
1163 .truncate(false)
1164 .read(true)
1165 .write(true)
1166 .open(directory.join(format!("warm-{}", self.warm_prefix(&warm.identity))))?;
1167 fs2::FileExt::lock_exclusive(&file)?;
1168 Ok(file)
1169 }
1170
1171 fn warm_record(&self, record: &WarmRecord) -> PathBuf {
1172 self.root.join("warm").join(format!(
1173 "{}-{}.json",
1174 self.warm_prefix(&record.task),
1175 hash32(&format!("{}\0{}", record.worktree, record.key.join("\0")))
1176 ))
1177 }
1178
1179 fn warm_records(&self, warm: &Warm) -> Vec<(PathBuf, WarmRecord)> {
1181 let prefix = format!("{}-", self.warm_prefix(&warm.identity));
1182 let mut records: Vec<(PathBuf, WarmRecord)> = fs::read_dir(self.root.join("warm"))
1183 .into_iter()
1184 .flatten()
1185 .flatten()
1186 .filter(|entry| {
1187 entry
1188 .file_name()
1189 .to_str()
1190 .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".json"))
1191 })
1192 .filter_map(|entry| {
1193 let record =
1194 WarmRecord::read(&fs::read(entry.path()).ok()?, &warm.identity).ok()?;
1195 Some((entry.path(), record))
1196 })
1197 .collect();
1198 records.sort_by_key(|(_, record)| std::cmp::Reverse(record.saved));
1199 records
1200 }
1201
1202 fn choose_warm(
1209 &self,
1210 workspace: &Workspace,
1211 warm: &Warm,
1212 current: &[String],
1213 related: bool,
1214 ) -> Option<(WarmRecord, String)> {
1215 let own = worktree(workspace);
1216 let mut candidates: Vec<(u8, bool, WarmRecord)> = self
1217 .warm_records(warm)
1218 .into_iter()
1219 .filter_map(|(_, record)| {
1220 let quality = key_match(warm, current, &record.key)?;
1221 let other = record.worktree != own;
1222 (!other || warm.portable == Portable::Anywhere).then_some((quality, other, record))
1223 })
1224 .filter(|(_, _, record)| {
1225 let descends = record
1226 .commit
1227 .as_deref()
1228 .is_some_and(|commit| descends_from(workspace, commit));
1229 descends == related
1230 })
1231 .collect();
1232 candidates.sort_by_key(|(quality, other, _)| (*quality, *other));
1234 let rank = candidates
1235 .first()
1236 .map(|(quality, other, _)| (*quality, *other))?;
1237 let tied = candidates
1238 .iter()
1239 .take_while(|(quality, other, _)| (*quality, *other) == rank)
1240 .count();
1241 let index = if tied > 1 {
1242 (0..tied)
1243 .min_by_key(|&index| {
1244 let behind = candidates[index]
1245 .2
1246 .commit
1247 .as_deref()
1248 .and_then(|commit| behind(workspace, commit));
1249 (behind.is_none(), behind, index)
1250 })
1251 .unwrap_or(0)
1252 } else {
1253 0
1254 };
1255 let (_, other, record) = candidates.swap_remove(index);
1256 let source = if other {
1257 format!("worktree {}", record.worktree)
1258 } else {
1259 "local".to_owned()
1260 };
1261 Some((record, source))
1262 }
1263
1264 fn store_warm(&self, record: &WarmRecord) -> Result<()> {
1265 let path = self.warm_record(record);
1266 let directory = path.parent().context("warm records have a directory")?;
1267 fs::create_dir_all(directory)?;
1268 record::publish(&self.root, &path, record)
1269 }
1270
1271 pub(crate) fn restore_warm(
1277 &self,
1278 workspace: &Workspace,
1279 task: &Task,
1280 warm: &Warm,
1281 prepared: &qk_executor::PreparedTask,
1282 ) -> Result<(Restored, Vec<String>)> {
1283 let mut restored = Restored {
1284 group: warm.group().map(str::to_owned),
1285 ..Restored::default()
1286 };
1287 let _lock = self.lock_warm(warm)?;
1288 let mut present = Vec::new();
1290 let mut missing = Vec::new();
1291 let locations = locations(workspace, task, warm)?;
1292 for location in &locations {
1293 if location.paths()?.is_empty() {
1294 missing.push((location, Vec::new()));
1295 } else if let Some(kept) = location.present()? {
1296 missing.push((location, kept));
1299 } else {
1300 present.push(warm.group().map_or(location.name, |group| group).to_owned());
1301 }
1302 }
1303 if missing.is_empty() {
1304 return Ok((restored, present));
1305 }
1306 let choosing = profile::span(&task.id, "warm_restore_choose");
1307 let current = key_digests(workspace, warm, prepared)?;
1308 let chosen = self
1310 .choose_warm(workspace, warm, ¤t, true)
1311 .or_else(|| {
1312 self.remote_warm(workspace, task, warm, ¤t)
1313 .map(|(record, branch)| (record, format!("remote {branch}")))
1314 })
1315 .or_else(|| self.choose_warm(workspace, warm, ¤t, false));
1316 let Some((record, source)) = chosen else {
1317 return Ok((restored, present));
1318 };
1319 drop(choosing);
1320 let _files = profile::span(&task.id, "warm_restore_files");
1321 let keep_mtimes = warm.preserve_mtimes && source == "local";
1322 restored.source = source;
1323 let mut noted = read_restored_files(workspace, warm);
1324 for (location, kept) in missing {
1325 let Some(group) = record.groups.get(location.name) else {
1326 continue;
1327 };
1328 let (files, bytes) =
1329 match self.restore_location(&task.id, location, &kept, group, keep_mtimes) {
1330 Ok(restored) => restored,
1331 Err(error) => {
1332 remove_restored(location, &kept);
1336 return Err(error);
1337 }
1338 };
1339 restored.files += files.len();
1340 restored.bytes += bytes;
1341 if kept.is_empty() {
1342 noted.groups.insert(location.name.to_owned(), files);
1343 } else {
1344 noted
1346 .groups
1347 .entry(location.name.to_owned())
1348 .or_default()
1349 .extend(files);
1350 }
1351 restored.groups.push(location.name.to_owned());
1352 }
1353 if !restored.groups.is_empty() {
1354 let path = restored_files_path(workspace, warm);
1355 fs::create_dir_all(path.parent().context("restore notes have a directory")?)?;
1356 fs::write(path, serde_json::to_vec(¬ed)?)?;
1357 }
1358 Ok((restored, present))
1359 }
1360
1361 fn restore_location(
1364 &self,
1365 task: &str,
1366 location: &Location,
1367 kept: &[&Outputs],
1368 group: &WarmGroup,
1369 keep_mtimes: bool,
1370 ) -> Result<(GroupFiles, u64)> {
1371 fs::create_dir_all(&location.base)?;
1372 let mut created = BTreeSet::new();
1375 let mut directories = Vec::new();
1376 let mut pending = Vec::new();
1377 for (path, artifact) in &group.artifacts {
1378 paths::safe_parents(&location.base, path)?;
1379 if !location.matches(path) {
1380 bail!("warm state for {task} holds {path}, outside its paths");
1381 }
1382 if kept.iter().any(|outputs| outputs.matches(path)) {
1383 continue;
1384 }
1385 let destination = location.base.join(path);
1386 if let Some(parent) = destination.parent()
1387 && created.insert(parent.to_owned())
1388 {
1389 fs::create_dir_all(parent)?;
1390 }
1391 match artifact {
1392 Artifact::File { blob, mode } => {
1393 let modified = keep_mtimes
1394 .then(|| group.mtimes.get(path))
1395 .flatten()
1396 .and_then(|nanos| u64::try_from(*nanos).ok())
1397 .map_or(RESTORED_AT, |nanos| {
1398 UNIX_EPOCH + Duration::from_nanos(nanos)
1399 });
1400 pending.push((path, blob.as_str(), *mode, destination, modified));
1401 }
1402 Artifact::Directory { mode } => {
1403 fs::create_dir_all(&destination)?;
1404 created.insert(destination.clone());
1405 directories.push((destination, *mode));
1406 }
1407 Artifact::Symlink { target, directory } => {
1408 validate_link(path, target)?;
1409 symlink(target, &destination, *directory)?;
1410 }
1411 }
1412 }
1413 let copied = self
1414 .restore_warm_files(&pending)
1415 .with_context(|| format!("warm state for {task} not restored"))?;
1416 for (destination, mode) in directories {
1418 set_mode(&destination, mode)?;
1419 }
1420 let mut files = BTreeMap::new();
1421 let mut bytes = 0;
1422 for ((path, blob, mode, _, _), (metadata, size)) in pending.into_iter().zip(copied) {
1423 bytes += size;
1424 let artifact = Artifact::File {
1425 blob: blob.to_owned(),
1426 mode,
1427 };
1428 files.insert(path.clone(), (metadata, artifact));
1429 }
1430 Ok((files, bytes))
1431 }
1432
1433 fn restore_warm_files(&self, files: &[PendingFile]) -> Result<Vec<RestoredFile>> {
1436 let restore =
1437 |(_, blob, mode, destination, modified): &PendingFile| -> Result<RestoredFile> {
1438 let source = self.root.join("blobs").join(blob);
1439 if copy_blob_contents(&source, destination)? != *blob {
1440 let _ = fs::remove_file(destination);
1442 bail!("corrupt blob {blob}");
1443 }
1444 set_modified(destination, *modified)?;
1446 set_mode(destination, *mode)?;
1447 let metadata = fs::symlink_metadata(destination)?;
1448 Ok((stamp(&metadata), metadata.len()))
1449 };
1450 let workers = files.len().div_ceil(64).clamp(1, RESTORERS);
1451 if workers == 1 {
1452 return files.iter().map(restore).collect();
1453 }
1454 let next = AtomicUsize::new(0);
1457 let failed = AtomicBool::new(false);
1458 let mut results: Vec<(usize, Result<RestoredFile>)> = std::thread::scope(|scope| {
1459 let handles: Vec<_> = (0..workers)
1460 .map(|_| {
1461 scope.spawn(|| {
1462 let mut done = Vec::new();
1463 while !failed.load(Ordering::Relaxed) {
1464 let index = next.fetch_add(1, Ordering::Relaxed);
1465 let Some(file) = files.get(index) else { break };
1466 let result = restore(file);
1467 if result.is_err() {
1468 failed.store(true, Ordering::Relaxed);
1469 }
1470 done.push((index, result));
1471 }
1472 done
1473 })
1474 })
1475 .collect();
1476 handles
1477 .into_iter()
1478 .flat_map(|handle| handle.join().expect("warm restore worker panicked"))
1479 .collect()
1480 });
1481 results.sort_by_key(|(index, _)| *index);
1482 results.into_iter().map(|(_, result)| result).collect()
1483 }
1484
1485 fn remote_warm(
1488 &self,
1489 workspace: &Workspace,
1490 task: &Task,
1491 warm: &Warm,
1492 current: &[String],
1493 ) -> Option<(WarmRecord, String)> {
1494 let remote = self
1495 .remote
1496 .as_ref()
1497 .filter(|_| warm.remote && warm.portable != Portable::Own)?;
1498 let own = worktree(workspace);
1499 let placed =
1501 |saved: Option<&str>| warm.portable != Portable::Path || saved == Some(own.as_str());
1502 for branch in branches(workspace) {
1503 let suits = |saved: &[String], worktree: Option<&str>| {
1504 key_match(warm, current, saved).is_some() && placed(worktree)
1505 };
1506 let name = remote_identity(warm, &own);
1507 match remote.fetch_warm(&self.root, &name, &task.id, &branch, suits) {
1508 Ok(Some((bytes, _pin))) => {
1509 let Some(record) = WarmRecord::read(&bytes, &warm.identity).ok() else {
1510 continue;
1511 };
1512 if key_match(warm, current, &record.key).is_none()
1515 || !placed(Some(&record.worktree))
1516 {
1517 continue;
1518 }
1519 if self.store_warm(&record).is_err() {
1520 return None;
1521 }
1522 return Some((record, branch));
1523 }
1524 Ok(None) => {}
1525 Err(error) => {
1526 qk_executor::status!(
1527 "qk: {}: remote warm state unavailable ({error:#})",
1528 task.id
1529 );
1530 continue;
1533 }
1534 }
1535 }
1536 None
1537 }
1538
1539 pub(crate) fn save_warm(
1542 &self,
1543 workspace: &Workspace,
1544 task: &Task,
1545 warm: &Warm,
1546 prepared: &qk_executor::PreparedTask,
1547 ) -> Result<()> {
1548 self.save_prepared(&Save::prepare(workspace, task, warm, prepared)?)
1549 }
1550
1551 pub(crate) fn save_warm_in_background(
1554 &self,
1555 workspace: &Workspace,
1556 task: &Task,
1557 warm: &Warm,
1558 prepared: &qk_executor::PreparedTask,
1559 ) -> Result<()> {
1560 let save = Save::prepare(workspace, task, warm, prepared)?;
1561 let cache = Cache {
1562 remote: self.remote.clone(),
1563 ..Cache::new(self.root.clone())
1564 };
1565 let handle = std::thread::spawn(move || {
1566 if let Err(error) = cache.save_prepared(&save) {
1567 qk_executor::status!("qk: {}: warm state not saved ({error:#})", save.task);
1568 }
1569 });
1570 self.saves.lock().unwrap().push(handle);
1571 Ok(())
1572 }
1573
1574 fn save_prepared(&self, save: &Save) -> Result<()> {
1579 let warm = &save.warm;
1580 let _lock = self.lock_warm(warm)?;
1581 let _save = profile::span(&save.task, "warm_save");
1582 let records = self.warm_records(warm);
1583 let previous = records
1585 .iter()
1586 .filter(|(_, record)| record.worktree == save.worktree)
1587 .min_by_key(|(_, record)| record.key != save.key)
1588 .map(|(_, record)| record);
1589 let noted: RestoredFiles = fs::read(&save.notes)
1590 .ok()
1591 .and_then(|bytes| serde_json::from_slice(&bytes).ok())
1592 .unwrap_or_default();
1593 let mut record = WarmRecord::new(
1594 warm.identity.clone(),
1595 save.worktree.clone(),
1596 save.commit.clone(),
1597 now_ms(),
1598 save.key.clone(),
1599 );
1600 let gone = |error: &std::io::Error| error.kind() == std::io::ErrorKind::NotFound;
1601 for location in &save.locations {
1602 if !location.base.exists() {
1603 continue;
1604 }
1605 let known = previous.and_then(|previous| previous.groups.get(location.name));
1606 let restored_files = noted.groups.get(location.name);
1607 let mut group = WarmGroup::default();
1608 let mut pending = Vec::new();
1609 for path in location.paths()? {
1610 paths::safe_parents(&location.base, &path)?;
1611 let absolute = location.base.join(&path);
1612 let metadata = match fs::symlink_metadata(&absolute) {
1613 Err(error) if gone(&error) => continue,
1614 result => result?,
1615 };
1616 let artifact = if metadata.file_type().is_symlink() {
1617 let target = match fs::read_link(&absolute) {
1618 Err(error) if gone(&error) => continue,
1619 result => result?,
1620 };
1621 let target = target
1622 .to_str()
1623 .context("symlink target must be UTF-8")?
1624 .to_owned();
1625 validate_link(&path, &target)?;
1626 Artifact::Symlink {
1627 target,
1628 directory: absolute.is_dir(),
1629 }
1630 } else if metadata.is_file() {
1631 let now = stamp(&metadata);
1632 let saved = known.and_then(|known| {
1633 (known.stamps.get(&path) == Some(&now))
1634 .then(|| known.artifacts.get(&path))
1635 .flatten()
1636 });
1637 let restored = restored_files
1638 .and_then(|files| files.get(&path))
1639 .filter(|(stamp, _)| *stamp == now)
1640 .map(|(_, artifact)| artifact);
1641 let reused = saved.or(restored).filter(|artifact| match artifact {
1642 Artifact::File { blob, .. } => self.root.join("blobs").join(blob).is_file(),
1643 _ => false,
1644 });
1645 let Some(artifact) = reused.cloned() else {
1646 pending.push((path, absolute, metadata));
1647 continue;
1648 };
1649 group.stamps.insert(path.clone(), now);
1650 if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1651 group.mtimes.insert(path.clone(), modified);
1652 }
1653 artifact
1654 } else if metadata.is_dir() {
1655 Artifact::Directory {
1656 mode: mode(&metadata),
1657 }
1658 } else {
1659 continue;
1660 };
1661 group.artifacts.insert(path, artifact);
1662 }
1663 let storing = profile::span(&save.task, "warm_save_blobs");
1664 let sources: Vec<_> = pending
1665 .iter()
1666 .map(|(_, absolute, _)| absolute.clone())
1667 .collect();
1668 for ((path, _, metadata), result) in pending.into_iter().zip(self.put_blobs(&sources)) {
1669 let blob = match result {
1670 Ok(blob) => blob,
1671 Err(error) if error.downcast_ref::<std::io::Error>().is_some_and(gone) => {
1672 continue;
1673 }
1674 Err(error) => return Err(error),
1675 };
1676 group.stamps.insert(path.clone(), stamp(&metadata));
1677 if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1678 group.mtimes.insert(path.clone(), modified);
1679 }
1680 group.artifacts.insert(
1681 path,
1682 Artifact::File {
1683 blob,
1684 mode: mode(&metadata),
1685 },
1686 );
1687 }
1688 drop(storing);
1689 if let Some(limit) = warm.max_size
1690 && group.size(self) > limit
1691 {
1692 qk_executor::status!(
1693 "qk: {}: warm {} is over its maxSize and was not saved",
1694 save.task,
1695 location.name
1696 );
1697 continue;
1698 }
1699 record.groups.insert(location.name.to_owned(), group);
1700 }
1701 self.store_warm(&record)?;
1702 let _ = fs::remove_file(&save.notes);
1703 let saved = self.warm_record(&record);
1706 for (path, _) in records
1707 .into_iter()
1708 .filter(|(path, _)| *path != saved)
1709 .skip(KEPT - 1)
1710 {
1711 let _ = fs::remove_file(path);
1712 }
1713 if warm.remote
1715 && warm.portable != Portable::Own
1716 && let Some(remote) = &self.remote
1717 && let Some(branch) = &save.branch
1718 {
1719 for group in record.groups.values_mut() {
1720 group.stamps.clear();
1721 }
1722 remote.upload_warm(
1723 &self.root,
1724 &warm.identity,
1725 &remote_identity(warm, &save.worktree),
1726 branch,
1727 serde_json::to_vec(&record)?,
1728 );
1729 }
1730 Ok(())
1731 }
1732}
1733
1734struct Save {
1737 task: String,
1738 warm: Warm,
1739 locations: Vec<Location>,
1740 worktree: String,
1741 commit: Option<String>,
1742 branch: Option<String>,
1743 key: Vec<String>,
1744 notes: PathBuf,
1745}
1746
1747impl Save {
1748 fn prepare(
1750 workspace: &Workspace,
1751 task: &Task,
1752 warm: &Warm,
1753 prepared: &qk_executor::PreparedTask,
1754 ) -> Result<Self> {
1755 Ok(Self {
1756 task: task.id.clone(),
1757 warm: warm.clone(),
1758 locations: locations(workspace, task, warm)?,
1759 worktree: worktree(workspace),
1760 commit: current_commit(workspace),
1761 branch: current_branch(workspace),
1762 key: key_digests(workspace, warm, prepared)?,
1763 notes: restored_files_path(workspace, warm),
1764 })
1765 }
1766}
1767
1768#[cfg(test)]
1769mod tests {
1770 use super::*;
1771 use qk_taskgraph::Request;
1772 use serde_json::json;
1773
1774 fn fixture() -> (tempfile::TempDir, Workspace, TaskGraph) {
1776 let temp = tempfile::tempdir().unwrap();
1777 fs::write(temp.path().join("nx.json"), "{}").unwrap();
1778 fs::write(
1779 temp.path().join("project.json"),
1780 json!({
1781 "name": "app", "targets": {
1782 "one": {"command": "true", "qk:warm": {"paths": ["scratch-one"]}},
1783 "two": {"command": "true", "qk:warm": {"paths": ["scratch-two"]}}
1784 }
1785 })
1786 .to_string(),
1787 )
1788 .unwrap();
1789 for path in ["scratch-one", "scratch-two"] {
1790 fs::create_dir(temp.path().join(path)).unwrap();
1791 fs::write(temp.path().join(path).join("state"), "saved").unwrap();
1792 }
1793 let workspace = Workspace::load(temp.path()).unwrap();
1794 let graph = TaskGraph::build(
1795 &workspace,
1796 &[
1797 Request::parse("app:one").unwrap(),
1798 Request::parse("app:two").unwrap(),
1799 ],
1800 )
1801 .unwrap();
1802 (temp, workspace, graph)
1803 }
1804
1805 fn entries(warm: Value) -> Result<Vec<Warm>> {
1807 let temp = tempfile::tempdir().unwrap();
1808 fs::write(temp.path().join("nx.json"), "{}").unwrap();
1809 fs::write(
1810 temp.path().join("project.json"),
1811 json!({"name": "app", "targets": {"build": {"command": "true", "qk:warm": warm}}})
1812 .to_string(),
1813 )
1814 .unwrap();
1815 let workspace = Workspace::load(temp.path()).unwrap();
1816 let graph = TaskGraph::build(&workspace, &[Request::parse("app:build").unwrap()]).unwrap();
1817 config(&workspace, &graph.tasks["app:build"])
1818 }
1819
1820 #[test]
1821 fn an_array_holds_the_task_entry_beside_groups() {
1822 let entries = entries(json!([
1823 {"group": "metro", "portable": true, "env": {"METRO_CACHE_ROOT": "{warm}/metro"}},
1824 {"paths": ["build"], "mtimes": "preserve"}
1825 ]))
1826 .unwrap();
1827 assert_eq!(entries.len(), 2);
1828 assert_eq!(entries[0].group(), Some("metro"));
1829 assert!(entries[0].portable == Portable::Anywhere && entries[0].directory);
1830 assert_eq!(entries[1].group(), None);
1831 assert!(entries[1].portable == Portable::Own && entries[1].preserve_mtimes);
1832 assert_eq!(kept_paths(&entries), ["build"]);
1833 }
1834
1835 #[test]
1836 fn portable_is_a_boolean_or_the_same_path() {
1837 let portable = |value| {
1838 entries(json!({"paths": ["build"], "portable": value}))
1839 .map(|entries| entries[0].portable)
1840 };
1841 assert_eq!(portable(json!("path")).unwrap(), Portable::Path);
1842 assert_eq!(portable(json!(false)).unwrap(), Portable::Own);
1843 assert_eq!(
1844 format!("{:#}", portable(json!("anywhere")).unwrap_err()),
1845 "qk:warm.portable must be true, false or \"path\""
1846 );
1847 }
1848
1849 #[test]
1850 fn entries_cannot_share_saves_or_variables() {
1851 let error = |warm| format!("{:#}", entries(warm).unwrap_err());
1852 assert_eq!(
1853 error(json!([{"paths": ["one"]}, {"paths": ["two"]}])),
1854 "only one qk:warm entry may go without a group"
1855 );
1856 assert_eq!(
1857 error(json!([{"group": "metro"}, {"group": "metro"}])),
1858 "qk:warm names group \"metro\" twice"
1859 );
1860 assert_eq!(
1861 error(json!([
1862 {"group": "one", "env": {"CACHE": "{warm}/one"}},
1863 {"group": "two", "env": {"CACHE": "{warm}/two"}}
1864 ])),
1865 "qk:warm sets CACHE in more than one entry"
1866 );
1867 assert_eq!(
1868 error(json!(["build"])),
1869 "qk:warm must be an object or an array of objects"
1870 );
1871 }
1872
1873 #[test]
1875 fn waiting_for_an_owner_does_not_block_other_releases() {
1876 let (_temp, workspace, graph) = fixture();
1877 let one = &graph.tasks["app:one"];
1878 let two = &graph.tasks["app:two"];
1879 let first = acquire(&workspace, one, &config(&workspace, one).unwrap()[0]).unwrap();
1880 let state = paths::worktree_state(&workspace.root).join("warm-kept");
1881 let lock = fs::OpenOptions::new()
1882 .create(true)
1883 .truncate(false)
1884 .read(true)
1885 .write(true)
1886 .open(state.join(format!("{}.lock", hash32(&two.id))))
1887 .unwrap();
1888 fs2::FileExt::lock_exclusive(&lock).unwrap();
1889 let (released, received) = std::sync::mpsc::channel();
1890 let completed = std::thread::scope(|scope| {
1891 scope.spawn(|| {
1892 let second =
1893 acquire(&workspace, two, &config(&workspace, two).unwrap()[0]).unwrap();
1894 release(&second);
1895 });
1896 std::thread::sleep(Duration::from_millis(100));
1897 scope.spawn(|| {
1898 release(&first);
1899 released.send(()).unwrap();
1900 });
1901 let completed = received.recv_timeout(Duration::from_secs(2)).is_ok();
1902 fs2::FileExt::unlock(&lock).unwrap();
1903 completed
1904 });
1905 assert!(
1906 completed,
1907 "a blocked owner lock prevented an unrelated release"
1908 );
1909 assert_eq!(
1910 fs::read_to_string(workspace.root.join("scratch-one/state")).unwrap(),
1911 "saved"
1912 );
1913 }
1914
1915 #[test]
1917 fn unreadable_ownership_preserves_saved_paths() {
1918 let (_temp, workspace, graph) = fixture();
1919 let owner = &graph.tasks["app:one"];
1920 let warm = config(&workspace, owner).unwrap().remove(0);
1921 let root = acquire(&workspace, owner, &warm).unwrap();
1922 let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1923 drop(entry); fs::write(Stash::manifest(&root), "{").unwrap();
1925 assert!(acquire(&workspace, owner, &warm).is_err());
1926 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1927 fs::remove_file(Stash::manifest(&root)).unwrap();
1928 assert!(acquire(&workspace, owner, &warm).is_err());
1929 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1930 fs::create_dir(Stash::manifest(&root)).unwrap();
1931 assert!(acquire(&workspace, owner, &warm).is_err());
1932 assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1933 }
1934
1935 #[cfg(unix)]
1937 #[test]
1938 fn inspection_errors_keep_recovery_incomplete() {
1939 let (_temp, workspace, graph) = fixture();
1940 let owner = &graph.tasks["app:one"];
1941 let warm = config(&workspace, owner).unwrap().remove(0);
1942 let root = acquire(&workspace, owner, &warm).unwrap();
1943 let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1944 let saved = root.with_extension("saved");
1945 fs::rename(&root, &saved).unwrap();
1946 fs::write(&root, "blocked").unwrap(); assert!(!entry.stash.put_back());
1948 let retained = Stash {
1949 root: root.clone(),
1950 workspace: workspace.root.clone(),
1951 moved: entry.stash.moved.clone(),
1952 };
1953 assert!(stash(&workspace, &root, &warm, Some(retained)).is_err());
1954 assert_eq!(fs::read_to_string(saved.join("0/state")).unwrap(), "saved");
1955 assert_eq!(fs::read_to_string(&root).unwrap(), "blocked");
1956 }
1957}