Skip to main content

qk_cache/
warm.rs

1//! Warm state: scratch state a task runs faster with, kept between runs of
2//! the same task and restored before it runs, but never part of a result.
3//! A target opts in with `qk:warm`:
4//!
5//! - `outputs: true`: the task's previous outputs, so an incremental tool
6//!   such as `tsc --build` starts from its last build;
7//! - `paths`: scratch directories in the workspace;
8//! - `env`: variables for the task, which may name `{warm}`, a directory qk
9//!   keeps for the task outside the working tree.
10//!
11//! A group already present on disk is left alone. Otherwise it is restored
12//! from this worktree's own save. `portable: true` also permits other
13//! worktrees' and remote saves. It is saved after successful runs, one record
14//! per worktree.
15//!
16//! `qk:warm` may also be an array of such objects, each restored and saved on
17//! its own, so state that relocates can be portable beside state that cannot.
18
19use std::collections::BTreeMap;
20use std::fs;
21use std::io::Write;
22use std::path::PathBuf;
23use std::time::{Duration, SystemTime, UNIX_EPOCH};
24
25use anyhow::{Context, Result, bail};
26use qk_config::Workspace;
27use qk_taskgraph::{Task, TaskGraph};
28use serde::{Deserialize, Serialize};
29use serde_json::Value;
30
31use crate::paths::{self, Outputs};
32use crate::record::{self, Artifact, WarmGroup, WarmRecord, validate_link};
33use crate::store::{mode, set_mode, symlink};
34use crate::{Cache, hash::digest_file};
35
36/// When restored files say they were written. Warm state comes from another
37/// checkout, so it must look older than every file in this one: tools that
38/// trust timestamps, as `tsc --build` does, then check the sources against it
39/// instead of taking a restored build for an up-to-date one.
40pub const RESTORED_AT: std::time::SystemTime = std::time::UNIX_EPOCH;
41
42/// What a group's identity starts with, before its name.
43const GROUP: &str = "group ";
44
45/// How many saves are kept for a task, across worktrees and keys.
46const KEPT: usize = 8;
47
48/// A target's `qk:warm`.
49#[derive(Clone, Debug, Default)]
50pub struct Warm {
51    /// What its saves belong to: the task, or the group it shares them with.
52    pub identity: String,
53    pub outputs: bool,
54    /// Workspace-relative scratch paths.
55    pub paths: Vec<String>,
56    /// Variables for the task, with tokens expanded.
57    pub env: BTreeMap<String, String>,
58    /// Whether any variable names `{warm}`, which makes it a group.
59    pub directory: bool,
60    pub max_size: Option<u64>,
61    pub remote: bool,
62    /// Whether a worktree's own save comes back with the modification times it
63    /// was saved with, rather than [`RESTORED_AT`].
64    pub preserve_mtimes: bool,
65    /// Whether another worktree's save, or the remote's, may be restored.
66    pub portable: bool,
67    /// What decides whether a save suits this checkout, in order.
68    pub key: Vec<KeyPart>,
69    /// How many leading parts of the key a save must match when none matches
70    /// it whole, or `None` to restore only exact matches.
71    pub restore_keys: Option<usize>,
72    /// Dependencies that delete the paths, which are moved aside while one
73    /// runs: `target` in the task's project, or `project:target`.
74    pub survive: Vec<String>,
75    /// Whether saving happens after the task has reported, in the
76    /// background, rather than before.
77    pub background: bool,
78}
79
80/// One part of a warm key.
81#[derive(Clone, Debug)]
82pub enum KeyPart {
83    /// A workspace-relative file, by its content.
84    File(String),
85    /// An environment variable, by its value.
86    Env(String),
87}
88
89/// The target's `qk:warm` entries, none without one.
90///
91/// Entries of an array keep their saves apart, so at most one of them may
92/// belong to the task itself, and the rest name groups. No two may set the
93/// same variable.
94pub fn config(workspace: &Workspace, task: &Task) -> Result<Vec<Warm>> {
95    const SHAPE: &str = "qk:warm must be an object or an array of objects";
96    let objects = match task.definition.extra.get("qk:warm") {
97        None => return Ok(Vec::new()),
98        Some(Value::Object(object)) => vec![object],
99        Some(Value::Array(entries)) => entries
100            .iter()
101            .map(|entry| entry.as_object().context(SHAPE))
102            .collect::<Result<_>>()?,
103        Some(_) => bail!(SHAPE),
104    };
105    let mut entries: Vec<Warm> = Vec::new();
106    for object in objects {
107        let warm = entry(workspace, task, object)?;
108        if entries.iter().any(|other| other.identity == warm.identity) {
109            bail!(match warm.group() {
110                Some(group) => format!("qk:warm names group {group:?} twice"),
111                None => "only one qk:warm entry may go without a group".to_owned(),
112            });
113        }
114        for name in warm.env.keys() {
115            if entries.iter().any(|other| other.env.contains_key(name)) {
116                bail!("qk:warm sets {name} in more than one entry");
117            }
118        }
119        entries.push(warm);
120    }
121    Ok(entries)
122}
123
124/// One `qk:warm` object.
125fn entry(
126    workspace: &Workspace,
127    task: &Task,
128    object: &serde_json::Map<String, Value>,
129) -> Result<Warm> {
130    for key in object.keys() {
131        if !matches!(
132            key.as_str(),
133            "outputs"
134                | "paths"
135                | "env"
136                | "maxSize"
137                | "remote"
138                | "mtimes"
139                | "portable"
140                | "key"
141                | "restoreKeys"
142                | "group"
143                | "survive"
144                | "save"
145        ) {
146            bail!("unknown qk:warm field {key:?}");
147        }
148    }
149    let flag = |name: &str, default: bool| -> Result<bool> {
150        object.get(name).map_or(Ok(default), |value| {
151            value
152                .as_bool()
153                .with_context(|| format!("qk:warm.{name} must be true or false"))
154        })
155    };
156    let group = object
157        .get("group")
158        .map(|value| {
159            value
160                .as_str()
161                .filter(|name| !name.is_empty())
162                .context("qk:warm.group must be a name")
163        })
164        .transpose()?;
165    // `survive` keeps `paths` across a dependency, which a group has none of.
166    if group.is_some()
167        && ["outputs", "paths", "survive"]
168            .iter()
169            .any(|key| object.contains_key(*key))
170    {
171        bail!(
172            "a qk:warm.group shares {{warm}} alone; outputs, paths and survive belong to one task"
173        );
174    }
175    let identity = group.map_or_else(|| task.id.clone(), |group| format!("{GROUP}{group}"));
176    const SURVIVE_SHAPE: &str = "qk:warm.survive must be an array of target names";
177    let survive = object
178        .get("survive")
179        .map(|value| {
180            value
181                .as_array()
182                .context(SURVIVE_SHAPE)?
183                .iter()
184                .map(|target| {
185                    target
186                        .as_str()
187                        .filter(|target| !target.is_empty())
188                        .map(str::to_owned)
189                        .context(SURVIVE_SHAPE)
190                })
191                .collect::<Result<Vec<_>>>()
192        })
193        .transpose()?
194        .unwrap_or_default();
195    let directory = directory(workspace, &identity);
196    let directory_text = directory.to_str().context("warm directory must be UTF-8")?;
197    let workspace_text = workspace
198        .root
199        .to_str()
200        .context("workspace root must be UTF-8")?;
201    let project_root = &workspace.projects[&task.project].root;
202    let project_directory = if project_root == "." {
203        workspace.root.clone()
204    } else {
205        workspace.root.join(project_root)
206    };
207    let project_text = project_directory
208        .to_str()
209        .context("project root must be UTF-8")?;
210    let expand = |text: &str| -> Result<String> {
211        if text.contains("{options.") || text.contains("{args.") {
212            bail!("dynamic warm environment values are not supported yet");
213        }
214        Ok(text
215            .replace("{workspaceRoot}", workspace_text)
216            .replace("{projectRoot}", project_text)
217            .replace("{warm}", directory_text))
218    };
219    let mut paths = Vec::new();
220    for path in object
221        .get("paths")
222        .map(|value| {
223            value
224                .as_array()
225                .context("qk:warm.paths must be an array of paths")
226        })
227        .transpose()?
228        .into_iter()
229        .flatten()
230    {
231        let path = path
232            .as_str()
233            .context("qk:warm.paths must be an array of paths")?;
234        if path.contains("{warm}") {
235            bail!("qk:warm.paths are workspace paths; {{warm}} is kept already");
236        }
237        let (excluded, path) = match path.strip_prefix('!') {
238            Some(path) => (true, path),
239            None => (false, path),
240        };
241        let path = paths::expand(workspace, &task.project, path)?
242            .trim_end_matches('/')
243            .to_owned();
244        paths::validate_path(&path)?;
245        paths.push(if excluded { format!("!{path}") } else { path });
246    }
247    let mut env = BTreeMap::new();
248    let mut uses_directory = false;
249    if let Some(values) = object.get("env") {
250        for (name, value) in values
251            .as_object()
252            .context("qk:warm.env must be an object")?
253        {
254            let value = value
255                .as_str()
256                .context("qk:warm.env values must be strings")?;
257            uses_directory |= value.contains("{warm}");
258            env.insert(name.clone(), expand(value)?);
259        }
260    }
261    let max_size = match object.get("maxSize") {
262        None => None,
263        Some(Value::String(text)) => Some(crate::parse_size(text)?),
264        Some(Value::Number(number)) => number.as_u64(),
265        Some(_) => bail!("qk:warm.maxSize must be a size such as \"2GB\""),
266    };
267    let preserve_mtimes = match object.get("mtimes").map(|value| value.as_str()) {
268        None | Some(Some("epoch")) => false,
269        Some(Some("preserve")) => true,
270        Some(_) => bail!("qk:warm.mtimes must be \"epoch\" or \"preserve\""),
271    };
272    const KEY_SHAPE: &str = "qk:warm.key must be an array of paths and {\"env\": name} objects";
273    let mut key = Vec::new();
274    for part in object
275        .get("key")
276        .map(|value| value.as_array().context(KEY_SHAPE))
277        .transpose()?
278        .into_iter()
279        .flatten()
280    {
281        key.push(match part {
282            Value::String(path) => {
283                let path = paths::expand(workspace, &task.project, path)?;
284                paths::validate_path(&path)?;
285                KeyPart::File(path)
286            }
287            Value::Object(object) if object.len() == 1 => KeyPart::Env(
288                object
289                    .get("env")
290                    .and_then(Value::as_str)
291                    .context(KEY_SHAPE)?
292                    .to_owned(),
293            ),
294            _ => bail!(KEY_SHAPE),
295        });
296    }
297    let restore_keys = object
298        .get("restoreKeys")
299        .map(|value| {
300            value
301                .as_u64()
302                .and_then(|count| usize::try_from(count).ok())
303                .filter(|count| *count <= key.len())
304                .context("qk:warm.restoreKeys must be a count no larger than qk:warm.key")
305        })
306        .transpose()?;
307    let background = match object.get("save").map(Value::as_str) {
308        None | Some(Some("wait")) => false,
309        Some(Some("background")) => true,
310        Some(_) => bail!("qk:warm.save must be \"wait\" or \"background\""),
311    };
312    Ok(Warm {
313        identity,
314        outputs: flag("outputs", false)?,
315        paths,
316        env,
317        directory: uses_directory,
318        max_size,
319        remote: flag("remote", true)?,
320        preserve_mtimes,
321        portable: flag("portable", false)?,
322        key,
323        restore_keys,
324        survive,
325        background,
326    })
327}
328
329/// The directory `{warm}` names for a task or group, in the worktree's state.
330fn directory(workspace: &Workspace, identity: &str) -> PathBuf {
331    paths::worktree_state(&workspace.root)
332        .join("warm")
333        .join(&blake3::hash(identity.as_bytes()).to_hex()[..32])
334}
335
336impl Warm {
337    /// The group the entry shares its saves with, if it names one.
338    pub fn group(&self) -> Option<&str> {
339        self.identity.strip_prefix(GROUP)
340    }
341
342    /// The scratch paths without their exclusions: what the task keeps,
343    /// whether or not it is saved, and so never an input.
344    pub fn kept_paths(&self) -> Vec<String> {
345        self.paths
346            .iter()
347            .filter(|path| !path.starts_with('!'))
348            .cloned()
349            .collect()
350    }
351}
352
353/// The scratch paths every entry keeps.
354pub fn kept_paths(entries: &[Warm]) -> Vec<String> {
355    entries.iter().flat_map(Warm::kept_paths).collect()
356}
357
358/// Paths moved aside while a dependency that deletes them runs, and moved
359/// back, over whatever it left there, when this is dropped.
360#[derive(Default)]
361pub struct Kept {
362    /// The stashes this dependency holds a share of.
363    roots: Vec<PathBuf>,
364}
365
366impl Drop for Kept {
367    fn drop(&mut self) {
368        for root in self.roots.iter().rev() {
369            release(root);
370        }
371    }
372}
373
374/// One task's paths, moved into its directory in the worktree's state.
375#[derive(Default, Serialize, Deserialize)]
376struct Stash {
377    #[serde(skip)]
378    root: PathBuf,
379    #[serde(skip)]
380    workspace: PathBuf,
381    /// Each moved path, relative to the workspace, by the order it was moved.
382    moved: Vec<String>,
383}
384
385/// A stash in use, and how many of the running dependencies share it. The
386/// lock keeps another qk process from taking it for abandoned.
387struct Active {
388    users: usize,
389    stash: Stash,
390    _lock: fs::File,
391}
392
393/// The stashes this process holds, by root. Dependencies that run at once
394/// share one: the first moves the paths, the last puts them back.
395static ACTIVE: std::sync::Mutex<BTreeMap<PathBuf, Active>> = std::sync::Mutex::new(BTreeMap::new());
396
397impl Stash {
398    /// The ownership record for the numbered paths in a stash.
399    fn manifest(root: &std::path::Path) -> PathBuf {
400        root.join("moved.json")
401    }
402
403    /// Publishes ownership before moving a path, without truncating the last
404    /// valid record. Sync the file before the atomic replacement.
405    fn save_manifest(&self) -> Result<()> {
406        let mut file = tempfile::NamedTempFile::new_in(&self.root)?;
407        file.write_all(&serde_json::to_vec(self)?)?;
408        file.as_file().sync_all()?;
409        file.persist(Self::manifest(&self.root))?;
410        Ok(())
411    }
412
413    /// Moves the paths back, replacing what is there now, and removes the
414    /// stash once every path is back. A path whose parents are not plain
415    /// directories inside the workspace, as when a dependency put a link or a
416    /// file in their place, is not touched and stays in the stash for the
417    /// next run. Whether every path went back.
418    fn put_back(&self) -> bool {
419        let mut complete = true;
420        for (index, path) in self.moved.iter().enumerate().rev() {
421            let kept = self.root.join(index.to_string());
422            match fs::symlink_metadata(&kept) {
423                Ok(_) => {}
424                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
425                Err(error) => {
426                    complete = false;
427                    qk_executor::status!(
428                        "qk: could not inspect saved {path}, kept for the next run ({error})"
429                    );
430                    continue;
431                }
432            }
433            let original = self.workspace.join(path);
434            let result = paths::safe_parents(&self.workspace, path).and_then(|()| {
435                remove(&original)?;
436                if let Some(parent) = original.parent() {
437                    fs::create_dir_all(parent)?;
438                }
439                fs::rename(&kept, &original)?;
440                Ok(())
441            });
442            if let Err(error) = result {
443                complete = false;
444                qk_executor::status!(
445                    "qk: could not put {path} back, kept for the next run ({error:#})"
446                );
447            }
448        }
449        if complete {
450            let _ = fs::remove_dir_all(&self.root);
451        }
452        complete
453    }
454}
455
456/// Removes a path itself, including a symlink, without following it.
457fn remove(path: &std::path::Path) -> std::io::Result<()> {
458    match fs::symlink_metadata(path) {
459        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
460        Err(error) => Err(error),
461        Ok(metadata) if metadata.is_dir() => fs::remove_dir_all(path),
462        Ok(_) => fs::remove_file(path),
463    }
464}
465
466/// Gives up one dependency's share of a stash, putting the paths back when
467/// it was the last.
468fn release(root: &std::path::Path) {
469    let mut active = ACTIVE
470        .lock()
471        .unwrap_or_else(std::sync::PoisonError::into_inner);
472    let Some(entry) = active.get_mut(root) else {
473        return;
474    };
475    entry.users -= 1;
476    if entry.users > 0 {
477        return;
478    }
479    let entry = active.remove(root).expect("present above");
480    // Another dependency taking the stash again waits on the lock, not on
481    // every other task's stash.
482    drop(active);
483    entry.stash.put_back();
484}
485
486/// Whether `entry`, from `owner`'s `survive`, names `task`.
487fn names(owner: &Task, entry: &str, task: &Task) -> bool {
488    match entry.split_once(':') {
489        Some((project, target)) => task.project == project && task.target == target,
490        None => task.project == owner.project && task.target == entry,
491    }
492}
493
494/// Whether `from` depends on `on`, directly or through other tasks.
495fn depends_on(graph: &TaskGraph, from: &str, on: &str) -> bool {
496    let mut seen = std::collections::BTreeSet::new();
497    let mut next = vec![from];
498    while let Some(id) = next.pop() {
499        for dependency in graph
500            .tasks
501            .get(id)
502            .into_iter()
503            .flat_map(|task| &task.dependencies)
504        {
505            if dependency == on {
506                return true;
507            }
508            if seen.insert(dependency.as_str()) {
509                next.push(dependency);
510            }
511        }
512    }
513    false
514}
515
516/// Moves aside the warm paths of every task in the graph that depends on
517/// `task` and names it in `survive`, so that `task` cannot delete them.
518pub fn keep_across(workspace: &Workspace, graph: &TaskGraph, task: &Task) -> Kept {
519    let mut kept = Kept::default();
520    for owner in graph.tasks.values() {
521        let Ok(entries) = config(workspace, owner) else {
522            continue;
523        };
524        // Only the task's own entry has paths to keep.
525        let Some(warm) = entries.into_iter().find(|warm| warm.group().is_none()) else {
526            continue;
527        };
528        if !warm.survive.iter().any(|entry| names(owner, entry, task))
529            || !depends_on(graph, &owner.id, &task.id)
530        {
531            continue;
532        }
533        match acquire(workspace, owner, &warm) {
534            Ok(root) => kept.roots.push(root),
535            Err(error) => qk_executor::status!(
536                "qk: {}: warm paths not kept across {} ({error:#})",
537                owner.id,
538                task.id
539            ),
540        }
541    }
542    kept
543}
544
545/// Takes a share of `owner`'s stash, moving its paths aside if no running
546/// dependency has yet.
547fn acquire(workspace: &Workspace, owner: &Task, warm: &Warm) -> Result<PathBuf> {
548    let state = paths::worktree_state(&workspace.root).join("warm-kept");
549    let root = state.join(hash32(&owner.id));
550    fs::create_dir_all(&state)?;
551    let lock = fs::OpenOptions::new()
552        .create(true)
553        .truncate(false)
554        .read(true)
555        .write(true)
556        .open(state.join(format!("{}.lock", hash32(&owner.id))))?;
557    // Only the nonblocking attempt holds ACTIVE. Waiting for another process
558    // must leave this process's other stashes free to acquire and release.
559    let mut active = loop {
560        let mut active = ACTIVE
561            .lock()
562            .unwrap_or_else(std::sync::PoisonError::into_inner);
563        if let Some(entry) = active.get_mut(&root) {
564            entry.users += 1;
565            return Ok(root);
566        }
567        match fs2::FileExt::try_lock_exclusive(&lock) {
568            Ok(()) => break active,
569            Err(error) if error.raw_os_error() == fs2::lock_contended_error().raw_os_error() => {
570                drop(active);
571                std::thread::sleep(Duration::from_millis(20));
572            }
573            Err(error) => return Err(error.into()),
574        }
575    };
576    // Under the lock, a stash on disk is one no run holds: one an earlier run
577    // could not put back, or left when it was stopped. It goes back first.
578    // What cannot go back yet, as when its parent is not a directory, stays
579    // in the stash this dependency holds, beside the paths that did go back,
580    // and is put back after the dependency, which may well recreate the
581    // parent.
582    let left = match fs::read(Stash::manifest(&root)) {
583        Ok(bytes) => {
584            let mut left: Stash = serde_json::from_slice(&bytes)
585                .context("invalid warm stash manifest; saved paths retained for recovery")?;
586            left.root = root.clone();
587            left.workspace = workspace.root.clone();
588            Some(left)
589        }
590        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
591            if root.is_dir() && fs::read_dir(&root)?.next().is_some() {
592                bail!("missing warm stash manifest; saved paths retained for recovery");
593            }
594            None
595        }
596        Err(error) => return Err(error).context("cannot read warm stash; saved paths retained"),
597    };
598    let retained = left.filter(|left| !left.put_back());
599    let stash = stash(workspace, &root, warm, retained)?;
600    active.insert(
601        root.clone(),
602        Active {
603            users: 1,
604            stash,
605            _lock: lock,
606        },
607    );
608    Ok(root)
609}
610
611/// Moves the warm paths into a stash at `root`: a new one, or `retained`,
612/// what an earlier run could not put back, which keeps its entries. A path
613/// that is one of those, or inside or around one, stays where it is.
614fn stash(
615    workspace: &Workspace,
616    root: &std::path::Path,
617    warm: &Warm,
618    retained: Option<Stash>,
619) -> Result<Stash> {
620    let mut stash = match retained {
621        Some(retained) => retained,
622        None => {
623            let _ = fs::remove_dir_all(root);
624            Stash {
625                root: root.to_owned(),
626                workspace: workspace.root.clone(),
627                moved: Vec::new(),
628            }
629        }
630    };
631    let held: Vec<String> = stash
632        .moved
633        .iter()
634        .enumerate()
635        .filter_map(
636            |(index, path)| match fs::symlink_metadata(root.join(index.to_string())) {
637                Ok(_) => Some(Ok(path.clone())),
638                Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
639                Err(error) => Some(Err(error).context("cannot inspect retained warm path")),
640            },
641        )
642        .collect::<Result<_>>()?;
643    let overlaps = |path: &str| {
644        held.iter().any(|held| {
645            held == path
646                || path.starts_with(&format!("{held}/"))
647                || held.starts_with(&format!("{path}/"))
648        })
649    };
650    // Each path on its own: one whose parents are not usable now, as one
651    // retained above may have, has nothing live there to move.
652    let mut kept = std::collections::BTreeSet::new();
653    for path in warm.kept_paths() {
654        if let Ok(found) = Outputs::from_paths(std::slice::from_ref(&path))
655            .and_then(|outputs| outputs.paths(&workspace.root))
656        {
657            kept.extend(found);
658        }
659    }
660    // The topmost paths only: moving one moves what is below it.
661    let topmost: Vec<String> = kept
662        .iter()
663        .filter(|path| {
664            !std::path::Path::new(path.as_str())
665                .ancestors()
666                .skip(1)
667                .any(|ancestor| {
668                    ancestor
669                        .to_str()
670                        .is_some_and(|ancestor| kept.contains(ancestor))
671                })
672        })
673        .filter(|path| !overlaps(path))
674        .cloned()
675        .collect();
676    if topmost.is_empty() {
677        return Ok(stash);
678    }
679    fs::create_dir_all(root)?;
680    for path in topmost {
681        let moved = paths::safe_parents(&workspace.root, &path).and_then(|()| {
682            stash.moved.push(path.clone());
683            stash.save_manifest()?;
684            // An entry put back earlier left its slot empty, and a retained
685            // one's slot is taken; a new one goes at the end either way.
686            fs::rename(
687                workspace.root.join(&path),
688                root.join((stash.moved.len() - 1).to_string()),
689            )
690            .inspect_err(|_| {
691                stash.moved.pop();
692            })?;
693            Ok(())
694        });
695        if let Err(error) = moved {
696            // What was moved before goes back, rather than waiting on a run
697            // that holds no share of it.
698            stash.save_manifest()?;
699            stash.put_back();
700            return Err(error);
701        }
702    }
703    Ok(stash)
704}
705
706/// Two tasks in a run cannot keep the same scratch path, since both would
707/// write it.
708pub fn check_overlaps(workspace: &Workspace, graph: &TaskGraph) -> Result<()> {
709    let mut claimed: Vec<(String, &str)> = Vec::new();
710    for (id, task) in &graph.tasks {
711        for path in kept_paths(&config(workspace, task)?) {
712            if let Some((_, other)) = claimed.iter().find(|(known, _)| {
713                known == &path
714                    || known.starts_with(&format!("{path}/"))
715                    || path.starts_with(&format!("{known}/"))
716            }) {
717                bail!("{id} and {other} both keep {path} as warm state");
718            }
719            claimed.push((path, id));
720        }
721    }
722    Ok(())
723}
724
725impl WarmGroup {
726    /// Total saved file bytes, excluding directories and links.
727    fn size(&self, cache: &Cache) -> u64 {
728        self.artifacts
729            .values()
730            .filter_map(|artifact| match artifact {
731                Artifact::File { blob, .. } => {
732                    fs::metadata(cache.root.join("blobs").join(blob)).ok()
733                }
734                _ => None,
735            })
736            .map(|metadata| metadata.len())
737            .sum()
738    }
739}
740
741/// What this worktree last restored, per group: each file's metadata once
742/// restored and the artifact it came from, so a save need not read it again.
743#[derive(Default, Serialize, Deserialize)]
744struct RestoredFiles {
745    groups: BTreeMap<String, BTreeMap<String, (Vec<i64>, Artifact)>>,
746}
747
748/// What a restore brought back.
749#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
750pub struct Restored {
751    /// The group the entry shares its saves with, absent for the task's own.
752    #[serde(default, skip_serializing_if = "Option::is_none")]
753    pub group: Option<String>,
754    /// `local` for this worktree's own save, `worktree <root>` for another
755    /// worktree's, or `remote <branch>`.
756    pub source: String,
757    pub groups: Vec<String>,
758    pub files: usize,
759    pub bytes: u64,
760}
761
762/// What warm state did for one run of a task.
763#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
764#[serde(rename_all = "camelCase")]
765pub struct WarmReport {
766    /// What each entry restored before the run, for those that restored any.
767    #[serde(default, skip_serializing_if = "Vec::is_empty")]
768    pub restored: Vec<Restored>,
769    /// The groups already on disk before the run, which were left as they
770    /// were: `outputs`, `paths` and `directory` for the task's own entry, and
771    /// a named group's name for its directory.
772    #[serde(default, skip_serializing_if = "Vec::is_empty")]
773    pub present: Vec<String>,
774    /// How long saving the entries after the run took, when any was saved
775    /// before the task reported.
776    pub save_ms: Option<u64>,
777    /// Whether any entry was saved in the background, after the task reported.
778    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
779    pub background: bool,
780}
781
782/// A warm group: its name, the directory its paths are relative to, and
783/// which paths below that directory belong to it.
784#[derive(Clone)]
785struct Location {
786    name: &'static str,
787    base: PathBuf,
788    /// `None` for everything below `base`.
789    outputs: Option<Outputs>,
790}
791
792impl Location {
793    /// The group's paths on disk, relative to `base`.
794    fn paths(&self) -> Result<std::collections::BTreeSet<String>> {
795        if !self.base.exists() {
796            return Ok(Default::default());
797        }
798        match &self.outputs {
799            Some(outputs) => outputs.paths(&self.base),
800            None => {
801                let mut paths = std::collections::BTreeSet::new();
802                for entry in walkdir::WalkDir::new(&self.base)
803                    .min_depth(1)
804                    .follow_links(false)
805                {
806                    let entry = entry?;
807                    paths.insert(paths::relative(&self.base, entry.path())?);
808                }
809                Ok(paths)
810            }
811        }
812    }
813
814    /// Whether a relative path belongs to this configured group.
815    fn matches(&self, path: &str) -> bool {
816        self.outputs
817            .as_ref()
818            .is_none_or(|outputs| outputs.matches(path))
819    }
820}
821
822/// Resolves output, scratch and external environment groups for this task.
823fn locations(workspace: &Workspace, task: &Task, warm: &Warm) -> Result<Vec<Location>> {
824    let mut locations = Vec::new();
825    if warm.outputs {
826        locations.push(Location {
827            name: "outputs",
828            base: workspace.root.clone(),
829            outputs: Some(Outputs::new(workspace, task)?),
830        });
831    }
832    if !warm.paths.is_empty() {
833        locations.push(Location {
834            name: "paths",
835            base: workspace.root.clone(),
836            outputs: Some(Outputs::from_paths(&warm.paths)?),
837        });
838    }
839    if warm.directory {
840        locations.push(Location {
841            name: "directory",
842            base: directory(workspace, &warm.identity),
843            outputs: None,
844        });
845    }
846    Ok(locations)
847}
848
849/// The branch being built: from CI when it checks out a detached head, else
850/// from git. `None` when there is no branch to name.
851fn current_branch(workspace: &Workspace) -> Option<String> {
852    for name in ["GITHUB_HEAD_REF", "GITHUB_REF_NAME"] {
853        if let Ok(value) = std::env::var(name)
854            && !value.is_empty()
855        {
856            return Some(value);
857        }
858    }
859    let output = paths::git(&workspace.root, &["rev-parse", "--abbrev-ref", "HEAD"]).ok()?;
860    let branch = String::from_utf8(output.stdout).ok()?.trim().to_owned();
861    (output.status.success() && !branch.is_empty() && branch != "HEAD").then_some(branch)
862}
863
864/// The branches to take warm state from, in order: the current one, then the
865/// workspace's default branch.
866fn branches(workspace: &Workspace) -> Vec<String> {
867    let default = workspace
868        .config
869        .default_base
870        .clone()
871        .unwrap_or_else(|| "main".into());
872    let default = default
873        .strip_prefix("origin/")
874        .unwrap_or(&default)
875        .to_owned();
876    let mut branches: Vec<String> = current_branch(workspace).into_iter().collect();
877    if !branches.contains(&default) {
878        branches.push(default);
879    }
880    branches
881}
882
883/// Metadata used to detect files unchanged since their save or restore.
884fn stamp(metadata: &fs::Metadata) -> Vec<i64> {
885    let modified = metadata
886        .modified()
887        .ok()
888        .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
889        .map_or(-1, |duration| duration.as_nanos() as i64);
890    #[allow(unused_mut)]
891    let mut stamp = vec![metadata.len() as i64, modified, i64::from(mode(metadata))];
892    #[cfg(unix)]
893    {
894        use std::os::unix::fs::MetadataExt;
895        stamp.extend([
896            metadata.ctime(),
897            metadata.ctime_nsec(),
898            metadata.ino() as i64,
899        ]);
900    }
901    stamp
902}
903
904/// This worktree's root, as records name it.
905fn worktree(workspace: &Workspace) -> String {
906    fs::canonicalize(&workspace.root)
907        .unwrap_or_else(|_| workspace.root.clone())
908        .to_string_lossy()
909        .into_owned()
910}
911
912/// The checkout commit, absent outside Git workspaces.
913fn current_commit(workspace: &Workspace) -> Option<String> {
914    let output = paths::git(&workspace.root, &["rev-parse", "HEAD"]).ok()?;
915    let commit = String::from_utf8(output.stdout).ok()?.trim().to_owned();
916    (output.status.success() && !commit.is_empty()).then_some(commit)
917}
918
919/// A digest of each part of the warm key, as this checkout and the task's
920/// environment have them.
921fn key_digests(
922    workspace: &Workspace,
923    warm: &Warm,
924    prepared: &qk_executor::PreparedTask,
925) -> Result<Vec<String>> {
926    warm.key
927        .iter()
928        .map(|part| {
929            Ok(match part {
930                KeyPart::File(path) => {
931                    paths::safe_parents(&workspace.root, path)?;
932                    let absolute = workspace.root.join(path);
933                    if absolute.is_file() {
934                        format!("file:{}", digest_file(&absolute)?)
935                    } else {
936                        "file:absent".to_owned()
937                    }
938                }
939                // As the task's process gets them: its execution variables
940                // over its environment, and nothing else.
941                KeyPart::Env(name) => match prepared
942                    .execution
943                    .get(std::ffi::OsStr::new(name))
944                    .or_else(|| prepared.env.get(std::ffi::OsStr::new(name)))
945                {
946                    Some(value) => format!(
947                        "env:{}",
948                        blake3::hash(value.as_encoded_bytes()).to_hex()[..32].to_owned()
949                    ),
950                    None => "env:unset".to_owned(),
951                },
952            })
953        })
954        .collect()
955}
956
957/// How well a save's key suits this checkout: `Some(0)` when it matches
958/// whole, `Some(1)` when it matches only in the leading parts `restoreKeys`
959/// allows, `None` when it does not suit.
960fn key_match(warm: &Warm, current: &[String], saved: &[String]) -> Option<u8> {
961    if saved == current {
962        return Some(0);
963    }
964    let count = warm.restore_keys?;
965    (saved.len() == current.len() && saved[..count] == current[..count]).then_some(1)
966}
967
968/// How many commits `commit` is behind the checkout's `HEAD`, or `None` when
969/// it is not an ancestor of it.
970fn behind(workspace: &Workspace, commit: &str) -> Option<u64> {
971    let ancestor = paths::git(
972        &workspace.root,
973        &["merge-base", "--is-ancestor", commit, "HEAD"],
974    )
975    .ok()?;
976    if !ancestor.status.success() {
977        return None;
978    }
979    let output = paths::git(
980        &workspace.root,
981        &["rev-list", "--count", &format!("{commit}..HEAD")],
982    )
983    .ok()?;
984    String::from_utf8(output.stdout).ok()?.trim().parse().ok()
985}
986
987/// A compact stable filename component for task and group identities.
988fn hash32(text: &str) -> String {
989    blake3::hash(text.as_bytes()).to_hex()[..32].to_owned()
990}
991
992/// A representable timestamp for restoring saved modification times.
993fn nanos(time: SystemTime) -> Option<i64> {
994    time.duration_since(UNIX_EPOCH)
995        .ok()
996        .and_then(|duration| i64::try_from(duration.as_nanos()).ok())
997}
998
999/// The save time used to order records across worktrees.
1000fn now_ms() -> u64 {
1001    SystemTime::now()
1002        .duration_since(UNIX_EPOCH)
1003        .map_or(0, |duration| duration.as_millis() as u64)
1004}
1005
1006/// Where a worktree notes what it last restored for a task or group.
1007fn restored_files_path(workspace: &Workspace, warm: &Warm) -> PathBuf {
1008    paths::worktree_state(&workspace.root)
1009        .join("warm-restored")
1010        .join(format!("{}.json", hash32(&warm.identity)))
1011}
1012
1013/// Reads optional restore metadata; a missing note forces fresh file reads.
1014fn read_restored_files(workspace: &Workspace, warm: &Warm) -> RestoredFiles {
1015    fs::read(restored_files_path(workspace, warm))
1016        .ok()
1017        .and_then(|bytes| serde_json::from_slice(&bytes).ok())
1018        .unwrap_or_default()
1019}
1020
1021impl Cache {
1022    /// The file-name prefix every save of a task or group shares.
1023    fn warm_prefix(&self, identity: &str) -> String {
1024        hash32(&format!(
1025            "{identity}\0{}\0{}",
1026            std::env::consts::OS,
1027            std::env::consts::ARCH
1028        ))
1029    }
1030
1031    /// Serializes restoring and saving a task's or group's state, which the
1032    /// tasks of a group may do at once.
1033    fn lock_warm(&self, warm: &Warm) -> Result<fs::File> {
1034        let directory = self.root.join("locks");
1035        fs::create_dir_all(&directory)?;
1036        let file = fs::OpenOptions::new()
1037            .create(true)
1038            .truncate(false)
1039            .read(true)
1040            .write(true)
1041            .open(directory.join(format!("warm-{}", self.warm_prefix(&warm.identity))))?;
1042        fs2::FileExt::lock_exclusive(&file)?;
1043        Ok(file)
1044    }
1045
1046    fn warm_record(&self, record: &WarmRecord) -> PathBuf {
1047        self.root.join("warm").join(format!(
1048            "{}-{}.json",
1049            self.warm_prefix(&record.task),
1050            hash32(&format!("{}\0{}", record.worktree, record.key.join("\0")))
1051        ))
1052    }
1053
1054    /// Every save of the task or group, newest first.
1055    fn warm_records(&self, warm: &Warm) -> Vec<(PathBuf, WarmRecord)> {
1056        let prefix = format!("{}-", self.warm_prefix(&warm.identity));
1057        let mut records: Vec<(PathBuf, WarmRecord)> = fs::read_dir(self.root.join("warm"))
1058            .into_iter()
1059            .flatten()
1060            .flatten()
1061            .filter(|entry| {
1062                entry
1063                    .file_name()
1064                    .to_str()
1065                    .is_some_and(|name| name.starts_with(&prefix) && name.ends_with(".json"))
1066            })
1067            .filter_map(|entry| {
1068                let record =
1069                    WarmRecord::read(&fs::read(entry.path()).ok()?, &warm.identity).ok()?;
1070                Some((entry.path(), record))
1071            })
1072            .collect();
1073        records.sort_by_key(|(_, record)| std::cmp::Reverse(record.saved));
1074        records
1075    }
1076
1077    /// The save to restore from, and how to name where it came from. A save
1078    /// whose key matches whole comes before one that matches in part; then
1079    /// this worktree's own before another's, which only a portable target
1080    /// restores; then the one made at the commit nearest behind `HEAD`; then
1081    /// the newest.
1082    fn choose_warm(
1083        &self,
1084        workspace: &Workspace,
1085        warm: &Warm,
1086        current: &[String],
1087    ) -> Option<(WarmRecord, String)> {
1088        let own = worktree(workspace);
1089        let mut candidates: Vec<(u8, bool, WarmRecord)> = self
1090            .warm_records(warm)
1091            .into_iter()
1092            .filter_map(|(_, record)| {
1093                let quality = key_match(warm, current, &record.key)?;
1094                let other = record.worktree != own;
1095                (!other || warm.portable).then_some((quality, other, record))
1096            })
1097            .collect();
1098        // Newest first within each rank, as warm_records returned them.
1099        candidates.sort_by_key(|(quality, other, _)| (*quality, *other));
1100        let rank = candidates
1101            .first()
1102            .map(|(quality, other, _)| (*quality, *other))?;
1103        let tied = candidates
1104            .iter()
1105            .take_while(|(quality, other, _)| (*quality, *other) == rank)
1106            .count();
1107        let index = if tied > 1 {
1108            (0..tied)
1109                .min_by_key(|&index| {
1110                    let behind = candidates[index]
1111                        .2
1112                        .commit
1113                        .as_deref()
1114                        .and_then(|commit| behind(workspace, commit));
1115                    (behind.is_none(), behind, index)
1116                })
1117                .unwrap_or(0)
1118        } else {
1119            0
1120        };
1121        let (_, other, record) = candidates.swap_remove(index);
1122        let source = if other {
1123            format!("worktree {}", record.worktree)
1124        } else {
1125            "local".to_owned()
1126        };
1127        Some((record, source))
1128    }
1129
1130    fn store_warm(&self, record: &WarmRecord) -> Result<()> {
1131        let path = self.warm_record(record);
1132        let directory = path.parent().context("warm records have a directory")?;
1133        fs::create_dir_all(directory)?;
1134        record::publish(&self.root, &path, record)
1135    }
1136
1137    /// Restores each warm group not already on disk from the save
1138    /// [`Cache::choose_warm`] picks, or the remote's, and notes what it
1139    /// restored so the next save can tell which files are unchanged. Files are
1140    /// dated [`RESTORED_AT`], unless the target keeps modification times and
1141    /// the save is this worktree's own.
1142    pub(crate) fn restore_warm(
1143        &self,
1144        workspace: &Workspace,
1145        task: &Task,
1146        warm: &Warm,
1147        prepared: &qk_executor::PreparedTask,
1148    ) -> Result<(Restored, Vec<String>)> {
1149        let mut restored = Restored {
1150            group: warm.group().map(str::to_owned),
1151            ..Restored::default()
1152        };
1153        let _lock = self.lock_warm(warm)?;
1154        // Local state wins: it is the newest for this checkout.
1155        let mut present = Vec::new();
1156        let mut missing = Vec::new();
1157        for location in locations(workspace, task, warm)? {
1158            if location.paths()?.is_empty() {
1159                missing.push(location);
1160            } else {
1161                present.push(warm.group().map_or(location.name, |group| group).to_owned());
1162            }
1163        }
1164        if missing.is_empty() {
1165            return Ok((restored, present));
1166        }
1167        let current = key_digests(workspace, warm, prepared)?;
1168        let (record, source) = match self.choose_warm(workspace, warm, &current) {
1169            Some(chosen) => chosen,
1170            None => match self.remote_warm(workspace, task, warm, &current) {
1171                Some((record, branch)) => (record, format!("remote {branch}")),
1172                None => return Ok((restored, present)),
1173            },
1174        };
1175        let keep_mtimes = warm.preserve_mtimes && source == "local";
1176        restored.source = source;
1177        let mut noted = read_restored_files(workspace, warm);
1178        for location in missing {
1179            let Some(group) = record.groups.get(location.name) else {
1180                continue;
1181            };
1182            fs::create_dir_all(&location.base)?;
1183            let mut files = BTreeMap::new();
1184            for (path, artifact) in &group.artifacts {
1185                paths::safe_parents(&location.base, path)?;
1186                if !location.matches(path) {
1187                    bail!("warm state for {} holds {path}, outside its paths", task.id);
1188                }
1189                let destination = location.base.join(path);
1190                if let Some(parent) = destination.parent() {
1191                    fs::create_dir_all(parent)?;
1192                }
1193                match artifact {
1194                    Artifact::File { blob, mode } => {
1195                        let source = self.root.join("blobs").join(blob);
1196                        if digest_file(&source)? != *blob {
1197                            bail!("warm state for {} has a corrupt blob", task.id);
1198                        }
1199                        fs::copy(&source, &destination)?;
1200                        let modified = keep_mtimes
1201                            .then(|| group.mtimes.get(path))
1202                            .flatten()
1203                            .and_then(|nanos| u64::try_from(*nanos).ok())
1204                            .map_or(RESTORED_AT, |nanos| {
1205                                UNIX_EPOCH + Duration::from_nanos(nanos)
1206                            });
1207                        fs::File::options()
1208                            .write(true)
1209                            .open(&destination)?
1210                            .set_modified(modified)?;
1211                        set_mode(&destination, *mode)?;
1212                        restored.files += 1;
1213                        restored.bytes += fs::metadata(&destination)?.len();
1214                        files.insert(
1215                            path.clone(),
1216                            (
1217                                stamp(&fs::symlink_metadata(&destination)?),
1218                                artifact.clone(),
1219                            ),
1220                        );
1221                    }
1222                    Artifact::Directory { mode } => {
1223                        fs::create_dir_all(&destination)?;
1224                        set_mode(&destination, *mode)?;
1225                    }
1226                    Artifact::Symlink { target, directory } => {
1227                        validate_link(path, target)?;
1228                        symlink(target, &destination, *directory)?;
1229                    }
1230                }
1231            }
1232            noted.groups.insert(location.name.to_owned(), files);
1233            restored.groups.push(location.name.to_owned());
1234        }
1235        if !restored.groups.is_empty() {
1236            let path = restored_files_path(workspace, warm);
1237            fs::create_dir_all(path.parent().context("restore notes have a directory")?)?;
1238            fs::write(path, serde_json::to_vec(&noted)?)?;
1239        }
1240        Ok((restored, present))
1241    }
1242
1243    /// The task's warm state from the remote store: the current branch's, else
1244    /// the default branch's. It is kept locally from then on.
1245    fn remote_warm(
1246        &self,
1247        workspace: &Workspace,
1248        task: &Task,
1249        warm: &Warm,
1250        current: &[String],
1251    ) -> Option<(WarmRecord, String)> {
1252        let remote = self
1253            .remote
1254            .as_ref()
1255            .filter(|_| warm.remote && warm.portable)?;
1256        for branch in branches(workspace) {
1257            match remote.fetch_warm(&self.root, &warm.identity, &branch) {
1258                Ok(Some(bytes)) => {
1259                    let Some(record) = WarmRecord::read(&bytes, &warm.identity).ok() else {
1260                        continue;
1261                    };
1262                    // A branch's save that does not suit this checkout gives
1263                    // way to the next branch's.
1264                    if key_match(warm, current, &record.key).is_none() {
1265                        continue;
1266                    }
1267                    if self.store_warm(&record).is_err() {
1268                        return None;
1269                    }
1270                    return Some((record, branch));
1271                }
1272                Ok(None) => {}
1273                Err(error) => {
1274                    qk_executor::status!(
1275                        "qk: {}: remote warm state unavailable ({error:#})",
1276                        task.id
1277                    );
1278                    // An invalid record or unavailable blob on this branch
1279                    // does not rule out a usable default-branch save.
1280                    continue;
1281                }
1282            }
1283        }
1284        None
1285    }
1286
1287    /// Saves the task's warm groups after a successful run, as this worktree's
1288    /// save.
1289    pub(crate) fn save_warm(
1290        &self,
1291        workspace: &Workspace,
1292        task: &Task,
1293        warm: &Warm,
1294        prepared: &qk_executor::PreparedTask,
1295    ) -> Result<()> {
1296        self.save_prepared(&Save::prepare(workspace, task, warm, prepared)?)
1297    }
1298
1299    /// Saves the task's warm groups on a thread of its own, which
1300    /// [`Cache::finish`] waits for.
1301    pub(crate) fn save_warm_in_background(
1302        &self,
1303        workspace: &Workspace,
1304        task: &Task,
1305        warm: &Warm,
1306        prepared: &qk_executor::PreparedTask,
1307    ) -> Result<()> {
1308        let save = Save::prepare(workspace, task, warm, prepared)?;
1309        let cache = Cache {
1310            remote: self.remote.clone(),
1311            ..Cache::new(self.root.clone())
1312        };
1313        let handle = std::thread::spawn(move || {
1314            if let Err(error) = cache.save_prepared(&save) {
1315                qk_executor::status!("qk: {}: warm state not saved ({error:#})", save.task);
1316            }
1317        });
1318        self.saves.lock().unwrap().push(handle);
1319        Ok(())
1320    }
1321
1322    /// Saves what [`Save::prepare`] gathered. A file whose metadata matches
1323    /// this worktree's last save or restore keeps its blob unread; one that
1324    /// is gone by the time it is read, as another task may remove it, is left
1325    /// out.
1326    fn save_prepared(&self, save: &Save) -> Result<()> {
1327        let warm = &save.warm;
1328        let _lock = self.lock_warm(warm)?;
1329        let records = self.warm_records(warm);
1330        // This worktree's save under the same key, else under any.
1331        let previous = records
1332            .iter()
1333            .filter(|(_, record)| record.worktree == save.worktree)
1334            .min_by_key(|(_, record)| record.key != save.key)
1335            .map(|(_, record)| record);
1336        let noted: RestoredFiles = fs::read(&save.notes)
1337            .ok()
1338            .and_then(|bytes| serde_json::from_slice(&bytes).ok())
1339            .unwrap_or_default();
1340        let mut record = WarmRecord::new(
1341            warm.identity.clone(),
1342            save.worktree.clone(),
1343            save.commit.clone(),
1344            now_ms(),
1345            save.key.clone(),
1346        );
1347        let gone = |error: &std::io::Error| error.kind() == std::io::ErrorKind::NotFound;
1348        for location in &save.locations {
1349            if !location.base.exists() {
1350                continue;
1351            }
1352            let known = previous.and_then(|previous| previous.groups.get(location.name));
1353            let restored_files = noted.groups.get(location.name);
1354            let mut group = WarmGroup::default();
1355            let mut pending = Vec::new();
1356            for path in location.paths()? {
1357                paths::safe_parents(&location.base, &path)?;
1358                let absolute = location.base.join(&path);
1359                let metadata = match fs::symlink_metadata(&absolute) {
1360                    Err(error) if gone(&error) => continue,
1361                    result => result?,
1362                };
1363                let artifact = if metadata.file_type().is_symlink() {
1364                    let target = match fs::read_link(&absolute) {
1365                        Err(error) if gone(&error) => continue,
1366                        result => result?,
1367                    };
1368                    let target = target
1369                        .to_str()
1370                        .context("symlink target must be UTF-8")?
1371                        .to_owned();
1372                    validate_link(&path, &target)?;
1373                    Artifact::Symlink {
1374                        target,
1375                        directory: absolute.is_dir(),
1376                    }
1377                } else if metadata.is_file() {
1378                    let now = stamp(&metadata);
1379                    let saved = known.and_then(|known| {
1380                        (known.stamps.get(&path) == Some(&now))
1381                            .then(|| known.artifacts.get(&path))
1382                            .flatten()
1383                    });
1384                    let restored = restored_files
1385                        .and_then(|files| files.get(&path))
1386                        .filter(|(stamp, _)| *stamp == now)
1387                        .map(|(_, artifact)| artifact);
1388                    let reused = saved.or(restored).filter(|artifact| match artifact {
1389                        Artifact::File { blob, .. } => self.root.join("blobs").join(blob).is_file(),
1390                        _ => false,
1391                    });
1392                    let Some(artifact) = reused.cloned() else {
1393                        pending.push((path, absolute, metadata));
1394                        continue;
1395                    };
1396                    group.stamps.insert(path.clone(), now);
1397                    if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1398                        group.mtimes.insert(path.clone(), modified);
1399                    }
1400                    artifact
1401                } else if metadata.is_dir() {
1402                    Artifact::Directory {
1403                        mode: mode(&metadata),
1404                    }
1405                } else {
1406                    continue;
1407                };
1408                group.artifacts.insert(path, artifact);
1409            }
1410            let sources: Vec<_> = pending
1411                .iter()
1412                .map(|(_, absolute, _)| absolute.clone())
1413                .collect();
1414            for ((path, _, metadata), result) in pending.into_iter().zip(self.put_blobs(&sources)) {
1415                let blob = match result {
1416                    Ok(blob) => blob,
1417                    Err(error) if error.downcast_ref::<std::io::Error>().is_some_and(gone) => {
1418                        continue;
1419                    }
1420                    Err(error) => return Err(error),
1421                };
1422                group.stamps.insert(path.clone(), stamp(&metadata));
1423                if let Some(modified) = metadata.modified().ok().and_then(nanos) {
1424                    group.mtimes.insert(path.clone(), modified);
1425                }
1426                group.artifacts.insert(
1427                    path,
1428                    Artifact::File {
1429                        blob,
1430                        mode: mode(&metadata),
1431                    },
1432                );
1433            }
1434            if let Some(limit) = warm.max_size
1435                && group.size(self) > limit
1436            {
1437                qk_executor::status!(
1438                    "qk: {}: warm {} is over its maxSize and was not saved",
1439                    save.task,
1440                    location.name
1441                );
1442                continue;
1443            }
1444            record.groups.insert(location.name.to_owned(), group);
1445        }
1446        self.store_warm(&record)?;
1447        let _ = fs::remove_file(&save.notes);
1448        // The oldest saves beyond the kept number go; the blobs they alone
1449        // cite are evicted with the cache.
1450        let saved = self.warm_record(&record);
1451        for (path, _) in records
1452            .into_iter()
1453            .filter(|(path, _)| *path != saved)
1454            .skip(KEPT - 1)
1455        {
1456            let _ = fs::remove_file(path);
1457        }
1458        // Stamps describe this machine's files; the remote record goes without.
1459        if warm.remote
1460            && warm.portable
1461            && let Some(remote) = &self.remote
1462            && let Some(branch) = &save.branch
1463        {
1464            for group in record.groups.values_mut() {
1465                group.stamps.clear();
1466            }
1467            remote.upload_warm(
1468                &self.root,
1469                &warm.identity,
1470                branch,
1471                serde_json::to_vec(&record)?,
1472            );
1473        }
1474        Ok(())
1475    }
1476}
1477
1478/// What a save needs from the workspace, gathered where the task ran so that
1479/// the save can run elsewhere.
1480struct Save {
1481    task: String,
1482    warm: Warm,
1483    locations: Vec<Location>,
1484    worktree: String,
1485    commit: Option<String>,
1486    branch: Option<String>,
1487    key: Vec<String>,
1488    notes: PathBuf,
1489}
1490
1491impl Save {
1492    /// Captures the effective key and locations before a background save.
1493    fn prepare(
1494        workspace: &Workspace,
1495        task: &Task,
1496        warm: &Warm,
1497        prepared: &qk_executor::PreparedTask,
1498    ) -> Result<Self> {
1499        Ok(Self {
1500            task: task.id.clone(),
1501            warm: warm.clone(),
1502            locations: locations(workspace, task, warm)?,
1503            worktree: worktree(workspace),
1504            commit: current_commit(workspace),
1505            branch: current_branch(workspace),
1506            key: key_digests(workspace, warm, prepared)?,
1507            notes: restored_files_path(workspace, warm),
1508        })
1509    }
1510}
1511
1512#[cfg(test)]
1513mod tests {
1514    use super::*;
1515    use qk_taskgraph::Request;
1516    use serde_json::json;
1517
1518    /// Two independent owners let a blocked file lock contend with a release.
1519    fn fixture() -> (tempfile::TempDir, Workspace, TaskGraph) {
1520        let temp = tempfile::tempdir().unwrap();
1521        fs::write(temp.path().join("nx.json"), "{}").unwrap();
1522        fs::write(
1523            temp.path().join("project.json"),
1524            json!({
1525                "name": "app", "targets": {
1526                    "one": {"command": "true", "qk:warm": {"paths": ["scratch-one"]}},
1527                    "two": {"command": "true", "qk:warm": {"paths": ["scratch-two"]}}
1528                }
1529            })
1530            .to_string(),
1531        )
1532        .unwrap();
1533        for path in ["scratch-one", "scratch-two"] {
1534            fs::create_dir(temp.path().join(path)).unwrap();
1535            fs::write(temp.path().join(path).join("state"), "saved").unwrap();
1536        }
1537        let workspace = Workspace::load(temp.path()).unwrap();
1538        let graph = TaskGraph::build(
1539            &workspace,
1540            &[
1541                Request::parse("app:one").unwrap(),
1542                Request::parse("app:two").unwrap(),
1543            ],
1544        )
1545        .unwrap();
1546        (temp, workspace, graph)
1547    }
1548
1549    /// The `qk:warm` entries of an `app:build` whose target declares `warm`.
1550    fn entries(warm: Value) -> Result<Vec<Warm>> {
1551        let temp = tempfile::tempdir().unwrap();
1552        fs::write(temp.path().join("nx.json"), "{}").unwrap();
1553        fs::write(
1554            temp.path().join("project.json"),
1555            json!({"name": "app", "targets": {"build": {"command": "true", "qk:warm": warm}}})
1556                .to_string(),
1557        )
1558        .unwrap();
1559        let workspace = Workspace::load(temp.path()).unwrap();
1560        let graph = TaskGraph::build(&workspace, &[Request::parse("app:build").unwrap()]).unwrap();
1561        config(&workspace, &graph.tasks["app:build"])
1562    }
1563
1564    #[test]
1565    fn an_array_holds_the_task_entry_beside_groups() {
1566        let entries = entries(json!([
1567            {"group": "metro", "portable": true, "env": {"METRO_CACHE_ROOT": "{warm}/metro"}},
1568            {"paths": ["build"], "mtimes": "preserve"}
1569        ]))
1570        .unwrap();
1571        assert_eq!(entries.len(), 2);
1572        assert_eq!(entries[0].group(), Some("metro"));
1573        assert!(entries[0].portable && entries[0].directory);
1574        assert_eq!(entries[1].group(), None);
1575        assert!(!entries[1].portable && entries[1].preserve_mtimes);
1576        assert_eq!(kept_paths(&entries), ["build"]);
1577    }
1578
1579    #[test]
1580    fn entries_cannot_share_saves_or_variables() {
1581        let error = |warm| format!("{:#}", entries(warm).unwrap_err());
1582        assert_eq!(
1583            error(json!([{"paths": ["one"]}, {"paths": ["two"]}])),
1584            "only one qk:warm entry may go without a group"
1585        );
1586        assert_eq!(
1587            error(json!([{"group": "metro"}, {"group": "metro"}])),
1588            "qk:warm names group \"metro\" twice"
1589        );
1590        assert_eq!(
1591            error(json!([
1592                {"group": "one", "env": {"CACHE": "{warm}/one"}},
1593                {"group": "two", "env": {"CACHE": "{warm}/two"}}
1594            ])),
1595            "qk:warm sets CACHE in more than one entry"
1596        );
1597        assert_eq!(
1598            error(json!(["build"])),
1599            "qk:warm must be an object or an array of objects"
1600        );
1601    }
1602
1603    /// An external owner lock must not prevent another stash's final release.
1604    #[test]
1605    fn waiting_for_an_owner_does_not_block_other_releases() {
1606        let (_temp, workspace, graph) = fixture();
1607        let one = &graph.tasks["app:one"];
1608        let two = &graph.tasks["app:two"];
1609        let first = acquire(&workspace, one, &config(&workspace, one).unwrap()[0]).unwrap();
1610        let state = paths::worktree_state(&workspace.root).join("warm-kept");
1611        let lock = fs::OpenOptions::new()
1612            .create(true)
1613            .truncate(false)
1614            .read(true)
1615            .write(true)
1616            .open(state.join(format!("{}.lock", hash32(&two.id))))
1617            .unwrap();
1618        fs2::FileExt::lock_exclusive(&lock).unwrap();
1619        let (released, received) = std::sync::mpsc::channel();
1620        let completed = std::thread::scope(|scope| {
1621            scope.spawn(|| {
1622                let second =
1623                    acquire(&workspace, two, &config(&workspace, two).unwrap()[0]).unwrap();
1624                release(&second);
1625            });
1626            std::thread::sleep(Duration::from_millis(100));
1627            scope.spawn(|| {
1628                release(&first);
1629                released.send(()).unwrap();
1630            });
1631            let completed = received.recv_timeout(Duration::from_secs(2)).is_ok();
1632            fs2::FileExt::unlock(&lock).unwrap();
1633            completed
1634        });
1635        assert!(
1636            completed,
1637            "a blocked owner lock prevented an unrelated release"
1638        );
1639        assert_eq!(
1640            fs::read_to_string(workspace.root.join("scratch-one/state")).unwrap(),
1641            "saved"
1642        );
1643    }
1644
1645    /// Damaged or absent ownership metadata never makes saved contents disposable.
1646    #[test]
1647    fn unreadable_ownership_preserves_saved_paths() {
1648        let (_temp, workspace, graph) = fixture();
1649        let owner = &graph.tasks["app:one"];
1650        let warm = config(&workspace, owner).unwrap().remove(0);
1651        let root = acquire(&workspace, owner, &warm).unwrap();
1652        let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1653        drop(entry); // Simulate a stopped process, without returning its paths.
1654        fs::write(Stash::manifest(&root), "{").unwrap();
1655        assert!(acquire(&workspace, owner, &warm).is_err());
1656        assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1657        fs::remove_file(Stash::manifest(&root)).unwrap();
1658        assert!(acquire(&workspace, owner, &warm).is_err());
1659        assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1660        fs::create_dir(Stash::manifest(&root)).unwrap();
1661        assert!(acquire(&workspace, owner, &warm).is_err());
1662        assert_eq!(fs::read_to_string(root.join("0/state")).unwrap(), "saved");
1663    }
1664
1665    /// Inspection errors are incomplete recovery, not evidence of an empty slot.
1666    #[cfg(unix)]
1667    #[test]
1668    fn inspection_errors_keep_recovery_incomplete() {
1669        let (_temp, workspace, graph) = fixture();
1670        let owner = &graph.tasks["app:one"];
1671        let warm = config(&workspace, owner).unwrap().remove(0);
1672        let root = acquire(&workspace, owner, &warm).unwrap();
1673        let entry = ACTIVE.lock().unwrap().remove(&root).unwrap();
1674        let saved = root.with_extension("saved");
1675        fs::rename(&root, &saved).unwrap();
1676        fs::write(&root, "blocked").unwrap(); // Numbered slot inspection returns ENOTDIR.
1677        assert!(!entry.stash.put_back());
1678        let retained = Stash {
1679            root: root.clone(),
1680            workspace: workspace.root.clone(),
1681            moved: entry.stash.moved.clone(),
1682        };
1683        assert!(stash(&workspace, &root, &warm, Some(retained)).is_err());
1684        assert_eq!(fs::read_to_string(saved.join("0/state")).unwrap(), "saved");
1685        assert_eq!(fs::read_to_string(&root).unwrap(), "blocked");
1686    }
1687}