mod evict;
mod glob;
mod hash;
mod ignored;
mod inputs;
mod paths;
pub mod profile;
mod record;
mod remote;
mod store;
pub mod warm;
use std::collections::BTreeMap;
use std::path::PathBuf;
use std::sync::OnceLock;
use std::sync::atomic::{AtomicBool, Ordering};
use anyhow::Result;
use qk_config::Workspace;
use qk_executor::{Capture, Outcome, PreparedTask, execute, execute_captured};
use qk_taskgraph::{Task, TaskGraph};
use serde_json::Value;
pub use evict::{Pruned, max_size, parse_size, prune, prune_automatic};
pub use glob::Pattern;
pub use hash::{DependencyGlob, Selects, SourceIgnore, dependency_globs, source_files};
pub use inputs::{
InputChange, InputChanges, JsonFields, ROOT_TSCONFIG_FILES, Resolution, Resolved, TsConfig,
manifest_for_dependents, resolve_each, resolve_tasks, selectively_hash_tsconfig,
without_resolution,
};
pub use paths::expand as expand_project_path;
pub use paths::{
Outputs, cache_directory, cache_location, clear_worktree_state, resolved_outputs,
worktree_state,
};
pub struct Cache {
pub root: PathBuf,
snapshot: OnceLock<std::result::Result<hash::Snapshot, String>>,
remote: Option<std::sync::Arc<remote::Remote>>,
reuse: bool,
kept: std::sync::Mutex<std::collections::BTreeSet<String>>,
saves: std::sync::Mutex<Vec<std::thread::JoinHandle<()>>>,
cleanup: std::sync::Mutex<Option<store::Cleanup>>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "kebab-case")]
pub enum CacheStatus {
LocalHit,
RemoteHit,
Miss,
Uncached,
}
pub struct Execution {
pub log: qk_executor::RecordedLog,
pub inputs_unchanged: bool,
}
pub struct TaskResult {
pub outcome: Outcome,
pub cache: CacheStatus,
pub key: Option<(String, Value)>,
pub warm: Option<warm::WarmReport>,
pub execution: Option<Execution>,
}
impl TaskResult {
pub fn uncached(outcome: Outcome) -> Self {
Self {
outcome,
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(),
}
}
pub fn for_observation(workspace: &Workspace) -> Self {
Self {
reuse: false,
..Self::new(paths::cache_location(workspace))
}
}
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)
}
}
pub fn finish(&self, workspace: &Workspace) {
if !self.reuse {
return;
}
if let Some(Ok(snapshot)) = self.snapshot.get() {
snapshot.save_digests(&workspace.root);
}
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,
cancelled: &AtomicBool,
) -> Result<TaskResult> {
let result = self.run_task(workspace, graph, task, prepared, 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,
cancelled: &AtomicBool,
) -> Result<TaskResult> {
let cacheable = task.definition.cache == Some(true);
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()
};
let entry_label = |warm: &warm::Warm| match warm.group() {
Some(group) => format!(" (group {group})"),
None => String::new(),
};
let mut running = prepared.clone();
for warm in &entries {
for (name, value) in &warm.env {
running.execution.insert(name.into(), value.into());
}
}
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 bypass = |reason: String| {
if cacheable {
qk_executor::status!("qk: {}: cache bypassed ({reason})", task.id);
}
let _profile = profile::span(&task.id, "execute");
execute(prepared, cancelled).map(TaskResult::uncached)
};
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, 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));
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:#}")),
};
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);
return Ok(TaskResult {
outcome,
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, &key, cancelled);
return Ok(TaskResult {
outcome,
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));
}
Err(error) => return bypass(format!("cache lock unavailable: {error}")),
};
if waited
&& hash::recheck_dependency_keys(snapshot, workspace, graph, task, 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());
}
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(true) => {
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,
cache: if claimed {
CacheStatus::RemoteHit
} else {
CacheStatus::LocalHit
},
key: keyed,
warm: None,
execution: None,
});
}
Ok(false) => {}
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(false);
}
self.restore(
&workspace.root,
&task.id,
&task.dependencies,
&key,
&outputs,
&prepared.display,
qk_executor::Shown::RemoteCache,
)
});
match fetched {
Ok(true) => {
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,
cache: CacheStatus::RemoteHit,
key: keyed,
warm: None,
execution: None,
});
}
Ok(false) => {}
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, &key, cancelled);
let execution = capture.take_log().map(|log| Execution {
log,
inputs_unchanged,
});
if outcome == Outcome::Success && inputs_unchanged && capture.healthy() {
let publishing = profile::span(&task.id, "publish");
let published = self.publish(&workspace.root, &key, &outputs, log.path());
drop(publishing);
match published {
Ok(()) => {
store::record_outputs(&workspace.root, &task.id, &key, &outputs);
if let Some(remote) = &self.remote {
remote.upload(&self.root, &key);
}
}
Err(error) => {
qk_executor::status!("qk: {}: could not save cache entry ({error})", task.id);
}
}
}
Ok(TaskResult {
outcome,
cache: CacheStatus::Miss,
key: keyed,
warm: warm_report,
execution,
})
}
#[allow(clippy::too_many_arguments)]
fn inputs_match(
&self,
snapshot: &hash::Snapshot,
workspace: &Workspace,
graph: &TaskGraph,
task: &Task,
prepared: &PreparedTask,
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, 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
}
}
pub type Keyed = std::result::Result<(String, Value), String>;
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))?;
Ok(graph
.tasks
.values()
.map(|task| {
let keyed = key_now(&snapshot, workspace, graph, task, prepared, cancelled);
(task.id.clone(), keyed)
})
.collect())
}
fn key_now(
snapshot: &hash::Snapshot,
workspace: &Workspace,
graph: &TaskGraph,
task: &Task,
prepared: &BTreeMap<String, PreparedTask>,
cancelled: &AtomicBool,
) -> Keyed {
let failed = |error: anyhow::Error| format!("{}: {error:#}", task.id);
let dependencies =
hash::dependency_keys(snapshot, workspace, graph, task, 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))
}
#[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);
}
}
#[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);
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");
{
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"
);
}
#[test]
fn complete_directory_restores_preserve_outputs() {
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);
cache.publish(root, &key, &outputs, &log).unwrap();
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
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!(restored);
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(),
std::path::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!(kept);
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();
}
}
}