qk_cache/

lib.rs

//! Local content-addressed cache shared by linked Git worktrees.

mod evict;
mod glob;
mod hash;
mod inputs;
mod paths;
pub mod profile;
mod record;
mod remote;
mod store;
pub mod warm;

use std::collections::BTreeMap;
use std::path::{Path, PathBuf};
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};

use anyhow::{Context, Result};
use qk_config::Workspace;
use qk_executor::{Capture, Outcome, PreparedTask, execute, execute_captured};
use qk_taskgraph::{Task, TaskGraph};
use serde_json::{Value, json};

pub use evict::{Pruned, max_size, parse_size, prune, prune_automatic};
pub use glob::Pattern;
pub use hash::{SourceIgnore, source_files};
pub use inputs::{
    InputChange, InputChanges, Resolution, Resolved, manifest_for_dependents, resolve_each,
    resolve_tasks, without_resolution,
};
/// Expands `{projectRoot}` and `{workspaceRoot}` in a project's path or glob.
pub use paths::expand as expand_project_path;
pub use paths::{
    Outputs, cache_directory, cache_location, clear_worktree_state, resolved_outputs,
    worktree_state,
};

/// The cache for one run; its workspace snapshot is taken on first use.
pub struct Cache {
    pub root: PathBuf,
    snapshot: OnceLock<std::result::Result<hash::Snapshot, String>>,
    remote: Option<std::sync::Arc<remote::Remote>>,
    /// Whether results and warm state may be restored or published.
    reuse: bool,
    /// The tasks whose outputs this run left exactly as they were.
    kept: std::sync::Mutex<std::collections::BTreeSet<String>>,
    /// Warm state being saved in the background.
    saves: std::sync::Mutex<Vec<std::thread::JoinHandle<()>>>,
    cleanup: std::sync::Mutex<Option<store::Cleanup>>,
}

/// A task's output fingerprint for its dependents' keys, or why it has none:
/// `"<task id>: <reason>"` for the task where fingerprinting first failed.
pub type Fingerprint = std::result::Result<String, String>;

/// Where a task's result came from.
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum CacheStatus {
    LocalHit,
    RemoteHit,
    Miss,
    /// Ran without consulting the cache: not cacheable, bypassed or disabled.
    Uncached,
}

/// One actual keyed execution, independent of reusable cache entries.
pub struct Execution {
    pub log: qk_executor::RecordedLog,
    pub inputs_unchanged: bool,
}

pub struct TaskResult {
    pub outcome: Outcome,
    pub fingerprint: Fingerprint,
    pub cache: CacheStatus,
    /// The task's key and what it was computed from, when it has one.
    pub key: Option<(String, Value)>,
    /// What warm state did, for targets that keep it.
    pub warm: Option<warm::WarmReport>,
    pub execution: Option<Execution>,
}

impl TaskResult {
    pub fn uncached(outcome: Outcome, reason: String) -> Self {
        Self {
            outcome,
            fingerprint: Err(reason),
            cache: CacheStatus::Uncached,
            key: None,
            warm: None,
            execution: None,
        }
    }
}

impl Cache {
    pub fn new(root: PathBuf) -> Self {
        Self {
            root,
            snapshot: OnceLock::new(),
            remote: None,
            reuse: true,
            kept: Default::default(),
            saves: Default::default(),
            cleanup: Default::default(),
        }
    }

    /// Records verified executions without configuring or using result stores.
    pub fn for_observation(workspace: &Workspace) -> Self {
        Self {
            reuse: false,
            ..Self::new(paths::cache_location(workspace))
        }
    }

    /// The workspace's cache, with the remote store its nx.json `s3` key
    /// configures, if any. Credentials and modes are read from `environment`,
    /// which includes the workspace's dotenv files. A remote store that cannot
    /// be used is reported and left out; the local cache still works.
    pub fn for_workspace(
        workspace: &Workspace,
        environment: &BTreeMap<std::ffi::OsString, std::ffi::OsString>,
    ) -> Self {
        let root = paths::cache_location(workspace);
        let remote = match remote::configure(
            workspace.config.extra.get("s3"),
            workspace.config.extra.get("storeServer"),
            environment,
        ) {
            Ok(remote) => remote.map(std::sync::Arc::new),
            Err(error) => {
                qk_executor::status!("qk: remote cache disabled: {error:#}");
                None
            }
        };
        if let Some(remote) = &remote {
            remote.read_index(&root);
        }
        Self {
            remote,
            ..Self::new(root)
        }
    }

    /// Saves what later runs reuse and waits for background saves, uploads
    /// and retired-output cleanup, reporting failures.
    pub fn finish(&self, workspace: &Workspace) {
        if !self.reuse {
            return;
        }
        if let Some(Ok(snapshot)) = self.snapshot.get() {
            snapshot.save_digests(&workspace.root);
        }
        // Before the uploads, which a save may add to.
        for save in std::mem::take(&mut *self.saves.lock().unwrap()) {
            let _ = save.join();
        }
        if let Some(remote) = &self.remote {
            let failures = remote.finish();
            for failure in &failures {
                qk_executor::status!("qk: remote cache upload failed for {failure}");
            }
        }
        let _profile = profile::span("cache", "restore_cleanup_wait");
        self.cleanup.lock().unwrap().take();
    }

    fn snapshot(
        &self,
        workspace: &Workspace,
        graph: &TaskGraph,
    ) -> std::result::Result<&hash::Snapshot, String> {
        self.snapshot
            .get_or_init(|| {
                hash::Snapshot::new(workspace, graph, &self.root)
                    .map_err(|error| format!("cannot snapshot the workspace: {error:#}"))
            })
            .as_ref()
            .map_err(Clone::clone)
    }

    pub fn run(
        &self,
        workspace: &Workspace,
        graph: &TaskGraph,
        task: &Task,
        prepared: &PreparedTask,
        dependencies: &BTreeMap<String, Fingerprint>,
        cancelled: &AtomicBool,
    ) -> Result<TaskResult> {
        let result = self.run_task(workspace, graph, task, prepared, dependencies, cancelled);
        if let Some(Ok(snapshot)) = self.snapshot.get() {
            snapshot.outputs_written(workspace, task);
        }
        result
    }

    fn run_task(
        &self,
        workspace: &Workspace,
        graph: &TaskGraph,
        task: &Task,
        prepared: &PreparedTask,
        dependencies: &BTreeMap<String, Fingerprint>,
        cancelled: &AtomicBool,
    ) -> Result<TaskResult> {
        let cacheable = task.definition.cache == Some(true);
        // Held until the task is done, whether it ran or came from the cache.
        let _kept = self
            .reuse
            .then(|| warm::keep_across(workspace, graph, task));
        let entries = if self.reuse {
            warm::config(workspace, task).unwrap_or_else(|error| {
                qk_executor::status!("qk: {}: warm state ignored ({error:#})", task.id);
                Vec::new()
            })
        } else {
            Vec::new()
        };
        // Names an entry in messages: the group it shares, if any.
        let entry_label = |warm: &warm::Warm| match warm.group() {
            Some(group) => format!(" (group {group})"),
            None => String::new(),
        };
        // The warm variables point tools at their state; they are not inputs,
        // so the key is computed from the task without them.
        let mut running = prepared.clone();
        for warm in &entries {
            for (name, value) in &warm.env {
                running.execution.insert(name.into(), value.into());
            }
        }
        // What each entry restored, and which groups were on disk already.
        let before = |running: &PreparedTask| -> (Vec<warm::Restored>, Vec<String>) {
            let (mut restored, mut present) = (Vec::new(), Vec::new());
            for warm in &entries {
                match self
                    .initialize()
                    .and_then(|()| self.restore_warm(workspace, task, warm, running))
                {
                    Ok((entry, groups)) => {
                        if !entry.groups.is_empty() {
                            restored.push(entry);
                        }
                        present.extend(groups);
                    }
                    Err(error) => qk_executor::status!(
                        "qk: {}: warm state{} not restored ({error:#})",
                        task.id,
                        entry_label(warm)
                    ),
                }
            }
            (restored, present)
        };
        let after = |outcome: Outcome,
                     (restored, present): (Vec<warm::Restored>, Vec<String>),
                     running: &PreparedTask|
         -> Option<warm::WarmReport> {
            if entries.is_empty() {
                return None;
            }
            let mut save_ms = None;
            let mut background = false;
            for warm in entries.iter().filter(|_| outcome == Outcome::Success) {
                let started = std::time::Instant::now();
                let saved = if warm.background {
                    self.save_warm_in_background(workspace, task, warm, running)
                } else {
                    self.save_warm(workspace, task, warm, running)
                };
                match saved {
                    Ok(()) if warm.background => background = true,
                    Ok(()) => {
                        *save_ms.get_or_insert(0) += started.elapsed().as_millis() as u64;
                    }
                    Err(error) => qk_executor::status!(
                        "qk: {}: warm state{} not saved ({error:#})",
                        task.id,
                        entry_label(warm)
                    ),
                }
            }
            Some(warm::WarmReport {
                restored,
                present,
                save_ms,
                background,
            })
        };
        let fallback = |reason: String| {
            let _profile = profile::span(&task.id, "execute");
            execute(prepared, cancelled).map(|outcome| TaskResult::uncached(outcome, reason))
        };
        // Reports why this task runs uncached, attributing the reason to this task.
        let bypass = |reason: String| {
            if cacheable {
                qk_executor::status!("qk: {}: cache bypassed ({reason})", task.id);
            }
            fallback(format!("{}: {reason}", task.id))
        };
        let dependencies = match dependencies
            .iter()
            .map(|(id, key)| key.clone().map(|key| (id.clone(), key)))
            .collect::<std::result::Result<BTreeMap<_, _>, _>>()
        {
            Ok(dependencies) => dependencies,
            Err(root) => {
                // Keep the original task and reason so chains stay readable.
                if cacheable {
                    qk_executor::status!("qk: {}: cache bypassed (depends on {root})", task.id);
                }
                return fallback(root);
            }
        };
        let snapshot = match self.snapshot(workspace, graph) {
            Ok(snapshot) => snapshot,
            Err(reason) => return bypass(reason),
        };
        let dependency_profile = profile::span(&task.id, "dependency_inputs");
        let dependencies =
            match hash::dependency_keys(snapshot, workspace, graph, task, &dependencies, cancelled)
            {
                Ok(dependencies) => dependencies,
                Err(error) => return bypass(format!("{error:#}")),
            };
        drop(dependency_profile);
        let input_profile = profile::span(&task.id, "inputs");
        let (key, inputs) = match hash::inputs(
            snapshot,
            workspace,
            task,
            prepared,
            &dependencies,
            cancelled,
        )
        .and_then(|inputs| Ok((hash::key(&inputs)?, inputs)))
        {
            Ok(key) => key,
            Err(error) => return bypass(format!("{error:#}")),
        };
        drop(input_profile);
        let keyed = Some((key.clone(), inputs));
        // As in Nx, the task sees its hash.
        running
            .execution
            .insert("NX_TASK_HASH".into(), key.clone().into());
        let running = &running;
        let outputs = match paths::Outputs::new(workspace, task) {
            Ok(outputs) => outputs,
            Err(error) => return bypass(format!("{error:#}")),
        };
        // Only dependents read a fingerprint. Hashing the outputs of a task none
        // depends on, after it ran, can mean reading gigabytes for nothing.
        let depended_on = graph
            .tasks
            .values()
            .any(|other| other.dependencies.contains(&task.id));
        let fingerprint_outputs = |key: &str| {
            if depended_on {
                outputs_fingerprint(task, &workspace.root, &outputs, key)
            } else {
                Err(format!("{}: no task depends on it", task.id))
            }
        };
        if !cacheable {
            let restored = before(running);
            let executing = profile::span(&task.id, "execute");
            let outcome = execute(running, cancelled)?;
            drop(executing);
            let warm_report = after(outcome, restored, running);
            let fingerprint = if self.unchanged(
                snapshot,
                workspace,
                graph,
                task,
                prepared,
                &dependencies,
                &key,
                outcome,
                cancelled,
            ) {
                fingerprint_outputs(&key)
            } else {
                Err(format!("{}: did not complete unchanged", task.id))
            };
            return Ok(TaskResult {
                outcome,
                fingerprint,
                cache: CacheStatus::Uncached,
                key: keyed,
                warm: warm_report,
                execution: None,
            });
        }
        if !self.reuse {
            let capture = Capture::new(None, prepared.display.clone()).retain_log();
            let executing = profile::span(&task.id, "execute");
            let outcome = execute_captured(running, cancelled, Some(&capture))?;
            drop(executing);
            let inputs_unchanged = self.inputs_match(
                snapshot,
                workspace,
                graph,
                task,
                prepared,
                &dependencies,
                &key,
                cancelled,
            );
            let fingerprint = if outcome == Outcome::Success && inputs_unchanged {
                fingerprint_outputs(&key)
            } else {
                Err(format!("{}: did not complete unchanged", task.id))
            };
            return Ok(TaskResult {
                outcome,
                fingerprint,
                cache: CacheStatus::Uncached,
                key: keyed,
                warm: None,
                execution: capture.take_log().map(|log| Execution {
                    log,
                    inputs_unchanged,
                }),
            });
        }
        if let Err(error) = self.initialize() {
            return bypass(format!("cache unavailable: {error}"));
        }
        let (_lock, waited) = match self.lock(&key, cancelled) {
            Ok(Some(lock)) => lock,
            Ok(None) => {
                return Ok(TaskResult::uncached(
                    Outcome::Cancelled,
                    format!("{}: cancelled", task.id),
                ));
            }
            Err(error) => return bypass(format!("cache lock unavailable: {error}")),
        };
        // Inputs may have changed while another worktree held this key's lock.
        if waited
            && hash::recheck_dependency_keys(
                snapshot,
                workspace,
                graph,
                task,
                &dependencies,
                cancelled,
            )
            .and_then(|dependencies| {
                hash::fingerprint(
                    snapshot,
                    workspace,
                    task,
                    prepared,
                    &dependencies,
                    cancelled,
                )
            })
            .ok()
            .as_ref()
                != Some(&key)
        {
            return bypass("inputs changed while waiting for another run".into());
        }
        // An entry fetched ahead of this task is local, but it came from the
        // remote store, and is reported as such.
        let prefetched = self.remote.as_ref().is_some_and(|remote| {
            remote.resolved(&key);
            remote.prefetched(&key)
        });
        let local_profile = profile::span(&task.id, "local_restore");
        match self.restore(
            &workspace.root,
            &task.id,
            &task.dependencies,
            &key,
            &outputs,
            &prepared.display,
            if prefetched {
                qk_executor::Shown::RemoteCache
            } else {
                qk_executor::Shown::LocalCache
            },
        ) {
            Ok(Some(fingerprint)) => {
                let claimed = match &self.remote {
                    Some(remote) if prefetched && remote.claim(&key) => true,
                    Some(remote) => {
                        remote.held(&key);
                        false
                    }
                    None => false,
                };
                qk_executor::report::event(qk_executor::report::Event::Cache {
                    id: task.id.clone(),
                    cached: if claimed {
                        qk_executor::report::Cached::RemoteHit
                    } else {
                        qk_executor::report::Cached::Hit
                    },
                });
                return Ok(TaskResult {
                    outcome: Outcome::Success,
                    fingerprint: Ok(fingerprint),
                    cache: if claimed {
                        CacheStatus::RemoteHit
                    } else {
                        CacheStatus::LocalHit
                    },
                    key: keyed,
                    warm: None,
                    execution: None,
                });
            }
            Ok(None) => {}
            Err(error) => {
                qk_executor::status!("qk: {}: ignoring unusable cache entry ({error})", task.id)
            }
        }
        drop(local_profile);
        if let Some(remote) = &self.remote {
            let fetch_profile = profile::span(&task.id, "remote_fetch");
            let fetched = remote.fetch(&self.root, &key, &task.id, cancelled);
            drop(fetch_profile);
            let fetched = fetched.and_then(|found| {
                if !found {
                    return Ok(None);
                }
                self.restore(
                    &workspace.root,
                    &task.id,
                    &task.dependencies,
                    &key,
                    &outputs,
                    &prepared.display,
                    qk_executor::Shown::RemoteCache,
                )
            });
            match fetched {
                Ok(Some(fingerprint)) => {
                    qk_executor::report::event(qk_executor::report::Event::Cache {
                        id: task.id.clone(),
                        cached: qk_executor::report::Cached::RemoteHit,
                    });
                    return Ok(TaskResult {
                        outcome: Outcome::Success,
                        fingerprint: Ok(fingerprint),
                        cache: CacheStatus::RemoteHit,
                        key: keyed,
                        warm: None,
                        execution: None,
                    });
                }
                Ok(None) => {}
                Err(error) => {
                    qk_executor::status!("qk: {}: remote cache unavailable ({error:#})", task.id)
                }
            }
        }
        qk_executor::report::event(qk_executor::report::Event::Cache {
            id: task.id.clone(),
            cached: qk_executor::report::Cached::Miss,
        });
        let log = match tempfile::NamedTempFile::new_in(self.root.join("tmp")) {
            Ok(log) => log,
            Err(error) => return bypass(format!("cache log unavailable: {error}")),
        };
        let capture =
            Capture::new(Some(log.as_file().try_clone()?), prepared.display.clone()).retain_log();
        let restored = before(running);
        let executing = profile::span(&task.id, "execute");
        let outcome = execute_captured(running, cancelled, Some(&capture))?;
        drop(executing);
        let warm_report = after(outcome, restored, running);
        let inputs_unchanged = self.inputs_match(
            snapshot,
            workspace,
            graph,
            task,
            prepared,
            &dependencies,
            &key,
            cancelled,
        );
        let execution = capture.take_log().map(|log| Execution {
            log,
            inputs_unchanged,
        });
        let fingerprint = if outcome != Outcome::Success || !inputs_unchanged {
            Err(format!("{}: did not complete unchanged", task.id))
        } else if !capture.healthy() {
            fingerprint_outputs(&key)
        } else {
            // The saved manifest already hashes every output; reuse it.
            let publishing = profile::span(&task.id, "publish");
            let published = self.publish(&workspace.root, &key, &outputs, log.path());
            drop(publishing);
            match published {
                Ok(fingerprint) => {
                    store::record_outputs(&workspace.root, &task.id, &key, &outputs);
                    if let Some(remote) = &self.remote {
                        remote.upload(&self.root, &key);
                    }
                    Ok(fingerprint)
                }
                Err(error) => {
                    qk_executor::status!("qk: {}: could not save cache entry ({error})", task.id);
                    fingerprint_outputs(&key)
                }
            }
        };
        Ok(TaskResult {
            outcome,
            fingerprint,
            cache: CacheStatus::Miss,
            key: keyed,
            warm: warm_report,
            execution,
        })
    }

    /// Whether a successful task's inputs still match the key it ran under.
    #[allow(clippy::too_many_arguments)]
    fn unchanged(
        &self,
        snapshot: &hash::Snapshot,
        workspace: &Workspace,
        graph: &TaskGraph,
        task: &Task,
        prepared: &PreparedTask,
        dependencies: &BTreeMap<String, String>,
        before: &str,
        outcome: Outcome,
        cancelled: &AtomicBool,
    ) -> bool {
        if outcome != Outcome::Success || cancelled.load(Ordering::SeqCst) {
            return false;
        }
        self.inputs_match(
            snapshot,
            workspace,
            graph,
            task,
            prepared,
            dependencies,
            before,
            cancelled,
        )
    }

    /// Rechecks declared inputs for successful and failed executions alike.
    #[allow(clippy::too_many_arguments)]
    fn inputs_match(
        &self,
        snapshot: &hash::Snapshot,
        workspace: &Workspace,
        graph: &TaskGraph,
        task: &Task,
        prepared: &PreparedTask,
        dependencies: &BTreeMap<String, String>,
        before: &str,
        cancelled: &AtomicBool,
    ) -> bool {
        if cancelled.load(Ordering::SeqCst) {
            return false;
        }
        let _profile = profile::span(&task.id, "inputs_recheck");
        let after = hash::recheck_dependency_keys(
            snapshot,
            workspace,
            graph,
            task,
            dependencies,
            cancelled,
        )
        .and_then(|dependencies| {
            hash::fingerprint(
                snapshot,
                workspace,
                task,
                prepared,
                &dependencies,
                cancelled,
            )
        });
        if after.as_deref().ok() != Some(before) {
            if task.definition.cache == Some(true) {
                if self.reuse {
                    qk_executor::status!(
                        "qk: {}: not caching because inputs changed during execution",
                        task.id
                    );
                } else {
                    qk_executor::status!(
                        "qk: {}: excluding execution from flaky results because inputs changed during execution",
                        task.id
                    );
                }
            }
            return false;
        }
        true
    }
}

/// Per-output values for dependents' keys. Disk and manifest must agree exactly,
/// so a dependent's key is the same whether this task was restored or executed.
pub(crate) fn file_output(content: &str, mode: u32) -> Value {
    json!({"content": content, "mode": mode})
}

pub(crate) fn link_output(target: &str) -> Value {
    json!({"link": target})
}

pub(crate) fn directory_output() -> Value {
    json!("directory")
}

/// What a task's dependents are keyed on. A task that declares outputs is
/// known to them only through those outputs, so a run that reproduces them
/// leaves its dependents cached however its own inputs changed. A task
/// without declared outputs may affect its dependents in ways qk cannot see,
/// so they are keyed on its inputs.
pub(crate) fn combine_outputs(
    input: &str,
    files: &BTreeMap<String, Value>,
    declared: bool,
) -> Result<String> {
    let identity = if declared { "declared outputs" } else { input };
    Ok(blake3::hash(&serde_json::to_vec(&(identity, files))?)
        .to_hex()
        .to_string())
}

fn outputs_fingerprint(
    task: &Task,
    root: &Path,
    outputs: &paths::Outputs,
    input: &str,
) -> Fingerprint {
    output_fingerprint(root, outputs, input)
        .map_err(|error| format!("{}: cannot fingerprint outputs: {error:#}", task.id))
}

fn output_fingerprint(root: &Path, outputs: &paths::Outputs, input: &str) -> Result<String> {
    let mut files = BTreeMap::new();
    for entry in outputs.entries(root)? {
        let (path, metadata) = entry?;
        paths::safe_parents(root, &path)?;
        let absolute = root.join(&path);
        let value = if metadata.file_type().is_symlink() {
            link_output(
                std::fs::read_link(absolute)?
                    .to_str()
                    .context("symlink target must be UTF-8")?,
            )
        } else if metadata.is_file() {
            file_output(&hash::digest_file(&absolute)?, store::mode(&metadata))
        } else {
            directory_output()
        };
        files.insert(path, value);
    }
    combine_outputs(input, &files, outputs.declared())
}

/// A task's key, and the inputs it is computed from, or why it has none.
pub type Keyed = std::result::Result<(String, Value), String>;

/// The key each task in `graph` would be cached under if it ran now, and what
/// it is computed from, without running a task. Runtime inputs are executed.
/// A dependency counts as having succeeded, fingerprinted by its outputs as
/// they are on disk.
pub fn keys(
    workspace: &Workspace,
    graph: &TaskGraph,
    prepared: &BTreeMap<String, PreparedTask>,
    cancelled: &AtomicBool,
) -> Result<BTreeMap<String, Keyed>> {
    let snapshot = hash::Snapshot::new(workspace, graph, &paths::cache_location(workspace))?;
    let mut keys = BTreeMap::new();
    let mut fingerprints: BTreeMap<String, Fingerprint> = BTreeMap::new();
    let mut pending: Vec<&Task> = graph.tasks.values().collect();
    // Dependencies first, as the runner keys them.
    while !pending.is_empty() {
        let waiting = pending.len();
        pending.retain(|task| {
            if !task
                .dependencies
                .iter()
                .all(|id| fingerprints.contains_key(id))
            {
                return true;
            }
            let keyed = key_now(
                &snapshot,
                workspace,
                graph,
                task,
                prepared,
                &fingerprints,
                cancelled,
            );
            let fingerprint = if task.definition.continuous == Some(true) {
                Err(format!(
                    "{}: continuous tasks are not fingerprinted",
                    task.id
                ))
            } else if !prepared[&task.id].ready_when.is_empty() {
                Err(format!(
                    "{}: tasks with readyWhen are not fingerprinted",
                    task.id
                ))
            } else {
                keyed.clone().and_then(|(key, _)| {
                    let outputs = paths::Outputs::new(workspace, task)
                        .map_err(|error| format!("{}: {error:#}", task.id))?;
                    outputs_fingerprint(task, &workspace.root, &outputs, &key)
                })
            };
            fingerprints.insert(task.id.clone(), fingerprint);
            keys.insert(task.id.clone(), keyed);
            false
        });
        anyhow::ensure!(pending.len() < waiting, "the task graph has a cycle");
    }
    Ok(keys)
}

fn key_now(
    snapshot: &hash::Snapshot,
    workspace: &Workspace,
    graph: &TaskGraph,
    task: &Task,
    prepared: &BTreeMap<String, PreparedTask>,
    fingerprints: &BTreeMap<String, Fingerprint>,
    cancelled: &AtomicBool,
) -> Keyed {
    let dependencies = task
        .dependencies
        .iter()
        .map(|id| fingerprints[id].clone().map(|key| (id.clone(), key)))
        .collect::<std::result::Result<BTreeMap<_, _>, _>>()?;
    let failed = |error: anyhow::Error| format!("{}: {error:#}", task.id);
    let dependencies =
        hash::dependency_keys(snapshot, workspace, graph, task, &dependencies, cancelled)
            .map_err(failed)?;
    let inputs = hash::inputs(
        snapshot,
        workspace,
        task,
        &prepared[&task.id],
        &dependencies,
        cancelled,
    )
    .map_err(failed)?;
    Ok((hash::key(&inputs).map_err(failed)?, inputs))
}

/// Runs the private detached-upload protocol used by the qk executable.
#[doc(hidden)]
pub fn run_upload_worker() -> anyhow::Result<()> {
    remote::background::run()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn concurrent_blob_writes_keep_results_ordered_and_repair_corruption() {
        let workspace = tempfile::tempdir().unwrap();
        let root = workspace.path();
        let cache = Cache::new(root.join("cache"));
        cache.initialize().unwrap();
        let mut sources = Vec::new();
        for index in 0..130 {
            let source = root.join(format!("source-{index}"));
            std::fs::write(&source, format!("content-{}", index % 65)).unwrap();
            sources.push(source);
        }
        let digest = cache.put_blob(&sources[0]).unwrap();
        std::fs::write(cache.root.join("blobs").join(&digest), "corrupt").unwrap();
        std::fs::remove_file(&sources[31]).unwrap();
        let results = cache.put_blobs(&sources);
        assert_eq!(results.len(), sources.len());
        for (index, result) in results.into_iter().enumerate() {
            if index == 31 {
                assert_eq!(
                    result
                        .unwrap_err()
                        .downcast_ref::<std::io::Error>()
                        .unwrap()
                        .kind(),
                    std::io::ErrorKind::NotFound
                );
                continue;
            }
            let digest = result.unwrap();
            let held = cache.root.join("blobs").join(&digest);
            assert_eq!(
                std::fs::read(&held).unwrap(),
                format!("content-{}", index % 65).as_bytes()
            );
            assert_eq!(hash::digest_file(&held).unwrap(), digest);
        }
    }

    /// Content the store holds already is reused, not copied over again.
    #[cfg(unix)]
    #[test]
    fn a_blob_the_store_holds_is_not_written_again() {
        use std::os::unix::fs::MetadataExt;
        let workspace = tempfile::tempdir().unwrap();
        let root = workspace.path();
        let cache = Cache::new(root.join(".qk/cache"));
        cache.initialize().unwrap();
        std::fs::write(root.join("first"), "same").unwrap();
        std::fs::write(root.join("second"), "same").unwrap();
        std::fs::write(root.join("other"), "different").unwrap();
        let first = cache.put_blob(&root.join("first")).unwrap();
        let blob = cache.root.join("blobs").join(&first);
        let stored = std::fs::metadata(&blob).unwrap().ino();
        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
        assert_eq!(std::fs::metadata(&blob).unwrap().ino(), stored);
        // A stored blob whose content no longer matches is stored anew.
        std::fs::write(&blob, "corrupt").unwrap();
        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
        assert_eq!(std::fs::read(&blob).unwrap(), b"same");
        // So is one that cannot be read.
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(&blob, std::fs::Permissions::from_mode(0o000)).unwrap();
        }
        assert_eq!(cache.put_blob(&root.join("second")).unwrap(), first);
        assert_eq!(std::fs::read(&blob).unwrap(), b"same");
        let other = cache.put_blob(&root.join("other")).unwrap();
        assert_ne!(other, first);
        assert_eq!(
            std::fs::read(cache.root.join("blobs").join(other)).unwrap(),
            b"different"
        );
    }

    /// Complete-directory restore preserves contents, links, modes, timestamps
    /// and fingerprints, and the subsequent hit keeps restored files in place.
    #[test]
    fn manifest_and_disk_output_fingerprints_agree() {
        let workspace = tempfile::tempdir().unwrap();
        let root = workspace.path();
        std::fs::create_dir_all(root.join("dist/nested/empty")).unwrap();
        std::fs::write(root.join("dist/nested/a.txt"), "a").unwrap();
        std::fs::write(root.join("dist/b.bin"), [0, 1, 2]).unwrap();
        std::fs::create_dir_all(root.join("bundle/empty")).unwrap();
        std::fs::write(root.join("bundle/readonly"), "readonly").unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let script = root.join("dist/run.sh");
            std::fs::write(&script, "#!/bin/sh\n").unwrap();
            std::fs::set_permissions(&script, std::fs::Permissions::from_mode(0o755)).unwrap();
            std::os::unix::fs::symlink("nested/a.txt", root.join("dist/link")).unwrap();
            std::fs::set_permissions(
                root.join("dist/nested"),
                std::fs::Permissions::from_mode(0o500),
            )
            .unwrap();
            std::fs::set_permissions(
                root.join("bundle/readonly"),
                std::fs::Permissions::from_mode(0o444),
            )
            .unwrap();
        }
        let log = root.join("log");
        std::fs::write(&log, "").unwrap();

        let cache = Cache::new(root.join(".qk/cache"));
        cache.initialize().unwrap();
        let outputs = paths::Outputs::from_patterns(&["dist", "bundle"]);
        let key = "0".repeat(64);
        let saved = cache.publish(root, &key, &outputs, &log).unwrap();
        assert_eq!(saved, output_fingerprint(root, &outputs, &key).unwrap());

        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            // The manifest retains the restrictive mode; cleanup needs write
            // permission on the old directory to remove its files.
            std::fs::set_permissions(
                root.join("dist/nested"),
                std::fs::Permissions::from_mode(0o755),
            )
            .unwrap();
        }
        std::fs::remove_dir_all(root.join("dist")).unwrap();
        std::fs::remove_dir_all(root.join("bundle")).unwrap();
        let before = std::time::SystemTime::now();
        let restored = cache
            .restore(
                root,
                "app:build",
                &Default::default(),
                &key,
                &outputs,
                &qk_executor::Display::Stream,
                qk_executor::Shown::LocalCache,
            )
            .unwrap();
        assert_eq!(restored, Some(saved.clone()));
        assert_eq!(saved, output_fingerprint(root, &outputs, &key).unwrap());
        assert!(root.join("dist/nested/empty").is_dir());
        assert!(root.join("bundle/empty").is_dir());
        assert_eq!(
            std::fs::read_to_string(root.join("bundle/readonly")).unwrap(),
            "readonly"
        );
        #[cfg(unix)]
        {
            assert_eq!(
                std::fs::read_link(root.join("dist/link")).unwrap(),
                Path::new("nested/a.txt")
            );
            assert_eq!(
                store::mode(&std::fs::metadata(root.join("dist/run.sh")).unwrap()),
                0o755
            );
            assert_eq!(
                store::mode(&std::fs::metadata(root.join("dist/nested")).unwrap()),
                0o500
            );
            assert_eq!(
                store::mode(&std::fs::metadata(root.join("bundle/readonly")).unwrap()),
                0o444
            );
        }
        let path = root.join("dist/nested/a.txt");
        let modified = std::fs::metadata(&path).unwrap().modified().unwrap();
        assert!(modified >= before);
        let kept = cache
            .restore(
                root,
                "app:build",
                &Default::default(),
                &key,
                &outputs,
                &qk_executor::Display::Hidden,
                qk_executor::Shown::LocalCache,
            )
            .unwrap();
        assert_eq!(kept, Some(saved));
        assert_eq!(
            std::fs::metadata(&path).unwrap().modified().unwrap(),
            modified
        );
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            std::fs::set_permissions(
                root.join("dist/nested"),
                std::fs::Permissions::from_mode(0o755),
            )
            .unwrap();
        }
    }
}