use std::collections::{BTreeMap, BTreeSet};
use std::ffi::OsString;
use std::fs::File;
use std::io::{Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, Mutex};
use std::time::{Duration, SystemTime};
use anyhow::{Context, Result, bail};
use foldhash::{HashMap, HashMapExt, HashSet};
use qk_config::Workspace;
use qk_executor::{Capture, Display, Outcome, PreparedTask, execute_captured, read_capture};
use qk_graph::ProjectGraph;
use qk_lockfile::Lockfile;
use qk_taskgraph::{Task, TaskGraph};
use serde_json::{Value, json};
use crate::glob::is_literal;
use crate::inputs::{Resolved, workspace_without_resolution};
use crate::paths::{self, Outputs, Pattern};
pub fn digest_file(path: &Path) -> Result<String> {
let mut hasher = blake3::Hasher::new();
let mut reader = File::open(path)?;
let mut buffer = [0; 64 * 1024];
loop {
let count = reader.read(&mut buffer)?;
if count == 0 {
break;
}
hasher.update(&buffer[..count]);
}
Ok(hasher.finalize().to_hex().to_string())
}
#[derive(Clone, PartialEq, serde::Serialize, serde::Deserialize)]
struct Stamp(Vec<i64>);
impl Stamp {
fn new(metadata: &std::fs::Metadata) -> Self {
let modified = metadata
.modified()
.ok()
.and_then(|time| time.duration_since(SystemTime::UNIX_EPOCH).ok())
.map_or(-1, |duration| duration.as_nanos() as i64);
#[allow(unused_mut)]
let mut values = vec![metadata.len() as i64, modified];
#[cfg(unix)]
{
use std::os::unix::fs::MetadataExt;
values.extend([
metadata.ctime(),
metadata.ctime_nsec(),
metadata.ino() as i64,
i64::from(metadata.mode()),
]);
}
Self(values)
}
fn settled(&self) -> bool {
let modified = SystemTime::UNIX_EPOCH + Duration::from_nanos(self.0[1].max(0) as u64);
self.0[1] >= 0
&& SystemTime::now()
.duration_since(modified)
.is_ok_and(|age| age > Duration::from_secs(2))
}
}
fn digests_path(root: &Path) -> PathBuf {
paths::worktree_state(root).join("digests.json")
}
#[derive(PartialEq, Eq, Hash)]
struct RuntimeKey {
command: String,
environment: BTreeMap<OsString, OsString>,
}
pub struct Snapshot {
pub(crate) files: BTreeSet<String>,
extra_candidates: BTreeSet<String>,
projects: ProjectGraph,
generated_tasks: Vec<(Task, Outputs)>,
task_dependencies: BTreeMap<String, BTreeSet<String>>,
source_ignore: SourceIgnore,
canonical_root: PathBuf,
workspace_prefix: Option<PathBuf>,
patterns: Mutex<HashMap<(String, bool), Arc<Pattern>>>,
digests: Mutex<HashMap<String, (Stamp, String)>>,
digests_changed: std::sync::atomic::AtomicBool,
directories: Mutex<HashSet<PathBuf>>,
runtime: Mutex<HashMap<RuntimeKey, Arc<Mutex<Option<Value>>>>>,
lockfile: Mutex<Option<Arc<Installed>>>,
dependency_outputs: Mutex<HashMap<String, Arc<DependencyOutputs>>>,
outputs_written: std::sync::atomic::AtomicU64,
}
struct DependencyOutputs {
anchors: Vec<String>,
files: Vec<String>,
values: Mutex<HashMap<String, Value>>,
}
struct Installed {
content: String,
lockfile: Option<Lockfile>,
digests: Mutex<HashMap<String, String>>,
}
impl Installed {
fn digest(&self, key: &str, fingerprints: impl FnOnce() -> BTreeSet<String>) -> String {
if let Some(digest) = self.digests.lock().unwrap().get(key) {
return digest.clone();
}
let mut hasher = blake3::Hasher::new();
for fingerprint in fingerprints() {
hasher.update(fingerprint.as_bytes());
hasher.update(b"\n");
}
let digest = hasher.finalize().to_hex().to_string();
self.digests
.lock()
.unwrap()
.insert(key.to_owned(), digest.clone());
digest
}
}
impl Snapshot {
pub fn new(workspace: &Workspace, graph: &TaskGraph, cache_path: &Path) -> Result<Self> {
let canonical_root = workspace.root.canonicalize()?;
let (files, lockfile, (digests, workspace_prefix)) = std::thread::scope(|scope| {
let lockfile = scope.spawn(|| parse_lockfile(&workspace.root.join("pnpm-lock.yaml")));
let digests = scope.spawn(|| {
let digests = load_digests(&workspace.root);
let workspace_prefix = paths::git_path(&workspace.root, "--show-toplevel")
.and_then(|root| root.canonicalize().ok())
.and_then(|root| canonical_root.strip_prefix(root).ok().map(PathBuf::from));
(digests, workspace_prefix)
});
let files = source_files(&workspace.root);
(
files,
lockfile.join().expect("lockfile thread panicked"),
digests.join().expect("digest thread panicked"),
)
});
let mut files = files?;
if let Ok(cache_relative) = cache_path.strip_prefix(&workspace.root) {
files.retain(|path| !Path::new(path).starts_with(cache_relative));
}
let mut generated = BTreeSet::new();
for outputs in graph
.tasks
.values()
.filter_map(|task| Outputs::new(workspace, task).ok())
.filter(Outputs::is_explicit)
{
for anchor in outputs.anchors() {
generated.extend(
under(&files, anchor)
.filter(|path| outputs.matches(path))
.cloned(),
);
}
}
for task in graph.tasks.values() {
let Ok(Some(warm)) = crate::warm::config(workspace, task) else {
continue;
};
if let Ok(paths) = Outputs::from_paths(&warm.kept_paths()) {
for anchor in paths.anchors() {
generated.extend(
under(&files, anchor)
.filter(|path| paths.matches(path))
.cloned(),
);
}
}
}
files.retain(|path| !generated.contains(path));
Ok(Self {
files,
extra_candidates: BTreeSet::new(),
projects: ProjectGraph::build(workspace)?,
source_ignore: SourceIgnore::new(&workspace.root)?,
generated_tasks: graph
.tasks
.values()
.filter_map(|task| {
let outputs = Outputs::new(workspace, task).ok()?;
outputs.is_explicit().then(|| (task.clone(), outputs))
})
.collect(),
task_dependencies: graph
.tasks
.iter()
.map(|(id, task)| (id.clone(), task.dependencies.clone()))
.collect(),
canonical_root,
workspace_prefix,
patterns: Mutex::default(),
digests: Mutex::new(digests),
digests_changed: Default::default(),
directories: Mutex::default(),
runtime: Mutex::default(),
lockfile: Mutex::new(lockfile),
dependency_outputs: Mutex::default(),
outputs_written: Default::default(),
})
}
pub fn save_digests(&self, root: &Path) {
if !self
.digests_changed
.load(std::sync::atomic::Ordering::Relaxed)
{
return;
}
let digests = self.digests.lock().unwrap();
let entries: BTreeMap<&String, (&Stamp, &String)> = digests
.iter()
.filter(|(path, _)| self.files.contains(*path))
.map(|(path, (stamp, digest))| (path, (stamp, digest)))
.collect();
let path = digests_path(root);
let saved = (|| -> Result<()> {
let directory = path.parent().context("digests have a directory")?;
std::fs::create_dir_all(directory)?;
let mut file = tempfile::NamedTempFile::new_in(directory)?;
{
let mut writer = std::io::BufWriter::new(file.as_file_mut());
serde_json::to_writer(&mut writer, &json!({"version": 1, "entries": entries}))?;
writer.flush()?;
}
file.persist(&path)?;
Ok(())
})();
if let Err(error) = saved {
qk_executor::status!("qk: could not save file digests: {error:#}");
}
}
fn safe_parents(&self, root: &Path, path: &str) -> Result<()> {
paths::validate_path(path)?;
let mut current = root.to_owned();
for part in Path::new(path)
.parent()
.into_iter()
.flat_map(Path::components)
{
current.push(part);
if self.directories.lock().unwrap().contains(¤t) {
continue;
}
match std::fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
bail!("input parent is a symlink: {}", current.display())
}
Ok(metadata) if !metadata.is_dir() => {
bail!("input parent is not a directory: {}", current.display())
}
Ok(_) => {
self.directories.lock().unwrap().insert(current.clone());
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(error.into()),
}
}
Ok(())
}
fn installed(&self, root: &Path) -> Result<Option<Arc<Installed>>> {
let absolute = root.join("pnpm-lock.yaml");
let Ok(metadata) = std::fs::symlink_metadata(&absolute) else {
return Ok(None);
};
if !metadata.is_file() {
return Ok(None);
}
let content = self.digest("pnpm-lock.yaml", &absolute, &metadata)?;
let mut current = self.lockfile.lock().unwrap();
if let Some(installed) = &*current
&& installed.content == content
{
return Ok(Some(installed.clone()));
}
let Some(installed) = parse_lockfile(&absolute) else {
return Ok(None);
};
*current = Some(installed.clone());
Ok(Some(installed))
}
pub fn with_candidates(mut self, paths: &[String]) -> Self {
self.files.extend(
paths
.iter()
.filter(|path| !self.source_ignore.matches(path))
.cloned(),
);
self.extra_candidates.extend(paths.iter().cloned());
self
}
fn pattern(&self, pattern: &str, negated: bool) -> Result<Arc<Pattern>> {
let key = (pattern.to_owned(), negated);
if let Some(pattern) = self.patterns.lock().unwrap().get(&key) {
return Ok(pattern.clone());
}
let compiled = Arc::new(Pattern::new(pattern, negated)?);
self.patterns.lock().unwrap().insert(key, compiled.clone());
Ok(compiled)
}
fn digest(&self, path: &str, absolute: &Path, metadata: &std::fs::Metadata) -> Result<String> {
let stamp = Stamp::new(metadata);
if let Some((known, digest)) = self.digests.lock().unwrap().get(path)
&& *known == stamp
{
return Ok(digest.clone());
}
let digest = digest_file(absolute)?;
let after = Stamp::new(&std::fs::symlink_metadata(absolute)?);
if after == stamp && stamp.settled() {
self.digests
.lock()
.unwrap()
.insert(path.to_owned(), (stamp, digest.clone()));
self.digests_changed
.store(true, std::sync::atomic::Ordering::Relaxed);
}
Ok(digest)
}
fn dependency_outputs(
&self,
workspace: &Workspace,
task: &Task,
) -> Result<Arc<DependencyOutputs>> {
if let Some(outputs) = self.dependency_outputs.lock().unwrap().get(&task.id) {
return Ok(outputs.clone());
}
let written = self
.outputs_written
.load(std::sync::atomic::Ordering::SeqCst);
let outputs = Arc::new(self.fresh_dependency_outputs(workspace, task)?);
let mut listed = self.dependency_outputs.lock().unwrap();
if self
.outputs_written
.load(std::sync::atomic::Ordering::SeqCst)
!= written
{
return Ok(outputs);
}
Ok(listed.entry(task.id.clone()).or_insert(outputs).clone())
}
fn fresh_dependency_outputs(
&self,
workspace: &Workspace,
task: &Task,
) -> Result<DependencyOutputs> {
let declared = Outputs::new(workspace, task)?;
let mut files = Vec::new();
for entry in declared.entries(&workspace.root)? {
let (path, metadata) = entry?;
if !metadata.is_dir() {
files.push(path);
}
}
Ok(DependencyOutputs {
anchors: declared.anchors().map(str::to_owned).collect(),
files,
values: Mutex::default(),
})
}
pub(crate) fn outputs_written(&self, workspace: &Workspace, task: &Task) {
let anchors: Option<Vec<String>> = Outputs::new(workspace, task)
.ok()
.filter(Outputs::is_explicit)
.map(|outputs| outputs.anchors().map(str::to_owned).collect());
let overlap = |a: &str, b: &str| {
a == b
|| b.strip_prefix(a).is_some_and(|rest| rest.starts_with('/'))
|| a.strip_prefix(b).is_some_and(|rest| rest.starts_with('/'))
};
let mut listed = self.dependency_outputs.lock().unwrap();
self.outputs_written
.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
listed.retain(|_, outputs| {
anchors.as_ref().is_some_and(|written| {
!written
.iter()
.any(|a| outputs.anchors.iter().any(|b| overlap(a, b)))
})
});
}
fn file_value(&self, root: &Path, path: &str) -> Result<Value> {
self.file_value_with_links(root, path, &mut BTreeSet::new())
}
fn file_value_with_links(
&self,
root: &Path,
path: &str,
visiting: &mut BTreeSet<PathBuf>,
) -> Result<Value> {
self.safe_parents(root, path)?;
let absolute = root.join(path);
let metadata = std::fs::symlink_metadata(&absolute)?;
if metadata.file_type().is_symlink() {
let target = std::fs::read_link(&absolute)?;
let Ok(resolved) = absolute.canonicalize() else {
return Ok(json!({"link":target, "dangling":true}));
};
let Ok(relative) = resolved.strip_prefix(&self.canonical_root) else {
bail!("input symlink resolves outside the workspace: {path}");
};
if resolved.is_file() {
let metadata = std::fs::metadata(&resolved)?;
let key = paths::relative(Path::new(""), relative)?;
return Ok(
json!({"link":target, "content":self.digest(&key, &resolved, &metadata)?,
"mode":crate::store::mode(&metadata) & 0o111 != 0}),
);
}
let relative = paths::relative(Path::new(""), relative)?;
if relative.is_empty() {
bail!("input symlink resolves to the workspace root: {path}");
}
if !visiting.insert(resolved.clone()) {
bail!("directory input symlink cycle: {path}");
}
let mut contents = BTreeMap::new();
for entry in walkdir::WalkDir::new(&resolved).follow_links(false) {
let entry = entry?;
if entry.file_type().is_dir() {
continue;
}
let file = paths::relative(&self.canonical_root, entry.path())?;
let value = self.file_value_with_links(&self.canonical_root, &file, visiting)?;
contents.insert(
file.strip_prefix(&relative).unwrap_or(&file).to_owned(),
value,
);
}
visiting.remove(&resolved);
return Ok(json!({"link":target, "directory":contents}));
}
if !metadata.is_file() {
bail!("input is not a regular file: {path}");
}
let mode = crate::store::mode(&metadata) & 0o111 != 0;
Ok(json!({"content":self.digest(path, &absolute, &metadata)?, "mode":mode}))
}
}
fn parse_lockfile(path: &Path) -> Option<Arc<Installed>> {
let text = std::fs::read_to_string(path).ok()?;
let lockfile = match Lockfile::parse(&text) {
Ok(lockfile) => Some(lockfile),
Err(error) => {
qk_executor::status!("qk: pnpm-lock.yaml: keying tasks by the whole file ({error:#})");
None
}
};
Some(Arc::new(Installed {
content: blake3::hash(text.as_bytes()).to_hex().to_string(),
lockfile,
digests: Mutex::default(),
}))
}
fn load_digests(root: &Path) -> HashMap<String, (Stamp, String)> {
#[derive(serde::Deserialize)]
struct Saved {
version: u32,
entries: HashMap<String, (Stamp, String)>,
}
std::fs::read(digests_path(root))
.ok()
.and_then(|bytes| serde_json::from_slice::<Saved>(&bytes).ok())
.filter(|saved| saved.version == 1)
.map(|saved| saved.entries)
.unwrap_or_default()
}
fn under<'a>(files: &'a BTreeSet<String>, prefix: &'a str) -> impl Iterator<Item = &'a String> {
files
.range::<str, _>((
std::ops::Bound::Included(prefix),
std::ops::Bound::Unbounded,
))
.take_while(move |path| path.starts_with(prefix))
.filter(move |path| path.len() == prefix.len() || path.as_bytes()[prefix.len()] == b'/')
}
pub struct SourceIgnore(ignore::gitignore::Gitignore);
impl SourceIgnore {
pub fn new(root: &Path) -> Result<Self> {
let mut builder = ignore::gitignore::GitignoreBuilder::new(root);
for name in [".gitignore", ".nxignore"] {
let path = root.join(name);
if path.is_file()
&& let Some(error) = builder.add(path)
{
return Err(error.into());
}
}
Ok(Self(builder.build()?))
}
fn matches_directory(&self, path: &str) -> bool {
self.matches(path) || self.0.matched(path, true).is_ignore()
}
pub fn matches(&self, path: &str) -> bool {
Path::new(path)
.ancestors()
.skip(1)
.take_while(|parent| !parent.as_os_str().is_empty())
.any(|parent| self.0.matched(parent, true).is_ignore())
|| self.0.matched(path, false).is_ignore()
}
}
pub fn source_files(root: &Path) -> Result<BTreeSet<String>> {
let lines =
|output: std::io::Result<std::process::Output>| -> Option<Result<BTreeSet<String>>> {
let output = output.ok().filter(|output| output.status.success())?;
Some(
output
.stdout
.split(|byte| *byte == 0)
.filter(|path| !path.is_empty())
.map(|path| {
String::from_utf8(path.to_vec()).context("input paths must be UTF-8")
})
.collect(),
)
};
let (listed, deleted) = if root.join(".nxignore").is_file() {
(None, None)
} else {
std::thread::scope(|scope| {
let deleted =
scope.spawn(|| paths::git(root, &["ls-files", "--deleted", "-z", "--", "."]));
let listed = paths::git(
root,
&[
"ls-files",
"-v",
"--cached",
"--others",
"--exclude-standard",
"-z",
"--",
".",
],
);
(
lines(listed),
lines(deleted.join().expect("git listing thread panicked")),
)
})
};
let candidates = match (listed, deleted) {
(Some(listed), Some(deleted)) => {
let deleted = deleted?;
let mut files = BTreeSet::new();
for entry in listed? {
let Some((tag, path)) = entry.split_once(' ') else {
bail!("unexpected git ls-files entry {entry:?}");
};
let absent =
matches!(tag, "S" | "s") && std::fs::symlink_metadata(root.join(path)).is_err();
if !absent && !deleted.contains(path) {
files.insert(path.to_owned());
}
}
files
}
_ => {
let mut files = BTreeSet::new();
for entry in ignore::WalkBuilder::new(root)
.hidden(false)
.parents(false)
.git_global(false)
.require_git(false)
.ignore(false)
.add_custom_ignore_filename(".nxignore")
.follow_links(false)
.filter_entry(|entry| {
!matches!(
entry.file_name().to_str(),
Some(".git" | ".qk" | "node_modules")
)
})
.build()
{
let entry = entry?;
if entry.file_type().is_some_and(|kind| !kind.is_dir()) {
files.insert(paths::relative(root, entry.path())?);
}
}
files
}
};
let ignore = SourceIgnore::new(root)?;
Ok(candidates
.into_iter()
.filter(|path| {
!ignore.matches(path)
&& !path
.split('/')
.any(|part| matches!(part, ".git" | ".qk" | "node_modules"))
})
.collect())
}
fn json_path<'v>(value: &'v Value, path: &str) -> Option<&'v Value> {
path.split('.')
.try_fold(value, |value, part| value.get(part))
}
fn set_json_path(target: &mut Value, path: &str, value: Value) {
let mut parts = path.split('.').peekable();
let mut current = target;
while let Some(part) = parts.next() {
let Value::Object(object) = current else {
return;
};
if parts.peek().is_none() {
object.insert(part.to_owned(), value);
return;
}
current = object.entry(part).or_insert_with(|| json!({}));
}
}
fn remove_json_path(target: &mut Value, path: &str) {
let (parent, last) = match path.rsplit_once('.') {
Some((parent, last)) => (Some(parent), last),
None => (None, path),
};
let parent = match parent {
Some(parent) => parent
.split('.')
.try_fold(&mut *target, |value, part| value.get_mut(part)),
None => Some(target),
};
if let Some(Value::Object(object)) = parent {
object.remove(last);
}
}
struct FilePattern {
matcher: Arc<Pattern>,
prefix: String,
excluded: bool,
}
#[derive(Default)]
struct FileSelection {
patterns: Vec<FilePattern>,
included: BTreeSet<String>,
excluded: BTreeSet<String>,
}
#[derive(PartialEq, Eq, PartialOrd, Ord)]
enum FileScope {
Workspace,
Project(String),
}
struct Resolver<'a> {
workspace: &'a Workspace,
snapshot: &'a Snapshot,
selected: BTreeSet<String>,
scopes: BTreeMap<FileScope, FileSelection>,
values: BTreeMap<String, Value>,
named_stack: Vec<(String, String)>,
expanded: BTreeSet<(String, String)>,
external: BTreeSet<String>,
prepared: Option<&'a PreparedTask>,
cancelled: &'a AtomicBool,
}
impl Resolver<'_> {
fn input(&mut self, project: &str, input: &Value) -> Result<()> {
match input {
Value::String(name) if name.starts_with('^') => {
self.dependencies_named(project, &name[1..])?;
}
Value::Object(object)
if object.contains_key("input") && !object.contains_key("fileset") =>
{
let name = object["input"]
.as_str()
.context("input must name a named input")?;
let projects = object.get("projects");
let dependencies = object.get("dependencies") == Some(&json!(true))
|| projects == Some(&json!("dependencies"));
let listed = projects.filter(|projects| {
**projects != json!("self") && **projects != json!("dependencies")
});
if (dependencies || listed.is_some()) && !self.named_stack.is_empty() {
bail!("named inputs can only refer to named inputs of their own project");
}
if dependencies {
self.dependencies_named(project, name)?;
} else if let Some(listed) = listed {
let patterns: Vec<String> = match listed {
Value::String(pattern) => vec![pattern.clone()],
other => serde_json::from_value(other.clone())
.context("input projects must be a string or an array")?,
};
for selected in
qk_graph::select_projects(&self.workspace.projects, &patterns, &[])?
{
if self.expanded.insert((selected.clone(), name.to_owned())) {
self.named(&selected, name)?;
}
}
} else {
self.named(project, name)?;
}
}
Value::Object(object)
if object.contains_key("fileset")
&& object.get("dependencies") == Some(&json!(true)) =>
{
if !self.named_stack.is_empty() {
bail!("named inputs can only refer to named inputs of their own project");
}
let pattern = object["fileset"]
.as_str()
.context("fileset input must be a glob")?
.to_owned();
for dependency in self.dependency_closure(project) {
self.fileset(&dependency, &pattern)?;
}
}
Value::Object(object)
if object.len() == 1 && object.contains_key("workingDirectory") =>
{
let mode = object["workingDirectory"].as_str().unwrap_or_default();
let directory = std::env::current_dir()?;
let value = match mode {
"absolute" => directory.to_string_lossy().into_owned(),
"relative" => {
let root = self.workspace.root.canonicalize()?;
let directory = directory.canonicalize()?;
paths::relative(&root, &directory)
.unwrap_or_else(|_| directory.to_string_lossy().into_owned())
}
_ => bail!("workingDirectory must be \"relative\" or \"absolute\""),
};
self.values
.insert(format!("workingDirectory:{mode}"), json!(value));
}
Value::Object(object) if object.contains_key("json") => {
self.json(project, object)?;
}
Value::String(name)
if !name.contains("{projectRoot}") && !name.contains("{workspaceRoot}") =>
{
self.named(project, name)?
}
Value::Object(object) if object.len() == 1 && object.contains_key("fileset") => {
let pattern = object["fileset"]
.as_str()
.context("fileset input must be a glob")?;
self.fileset(project, pattern)?;
}
Value::String(pattern) => self.fileset(project, pattern)?,
Value::Object(object) if object.len() == 1 && object.contains_key("env") => {
let name = object["env"]
.as_str()
.context("env input must name a variable")?;
let Some(prepared) = self.prepared else {
self.values.insert(format!("env:{name}"), Value::Null);
return Ok(());
};
let value = prepared
.env
.get(std::ffi::OsStr::new(name))
.map(|value| {
value
.to_str()
.map(str::to_owned)
.context("declared env input must be UTF-8")
})
.transpose()?;
self.values.insert(format!("env:{name}"), json!(value));
}
Value::Object(object) if object.len() == 1 && object.contains_key("runtime") => {
let command = object["runtime"]
.as_str()
.context("runtime input must be a command")?;
let Some(prepared) = self.prepared else {
self.values
.insert(format!("runtime:{command}"), Value::Null);
return Ok(());
};
let memo = RuntimeKey {
command: command.into(),
environment: prepared.runtime_env.clone(),
};
let slot = self
.snapshot
.runtime
.lock()
.unwrap()
.entry(memo)
.or_default()
.clone();
let mut saved = slot.lock().unwrap();
if let Some(value) = saved.as_ref() {
self.values
.insert(format!("runtime:{command}"), value.clone());
return Ok(());
}
let _profile = crate::profile::span(command, "runtime");
let mut prepared = prepared.clone();
prepared.env = prepared.runtime_env.clone();
prepared.execution.clear();
prepared.node_path = None;
prepared.commands = vec![command.into()];
prepared.cwd = self.workspace.root.clone();
prepared.parallel = false;
let log = tempfile::NamedTempFile::new()?;
let capture = Capture::new(Some(log.as_file().try_clone()?), Display::Hidden);
if execute_captured(&prepared, self.cancelled, Some(&capture))? != Outcome::Success
{
bail!("runtime input did not succeed");
}
let mut stdout = blake3::Hasher::new();
let mut stderr = blake3::Hasher::new();
read_capture(File::open(log.path())?, |error, bytes| {
if error {
stderr.update(bytes);
} else {
stdout.update(bytes);
}
Ok(())
})?;
let value = json!([
stdout.finalize().to_hex().to_string(),
stderr.finalize().to_hex().to_string()
]);
*saved = Some(value.clone());
self.values.insert(format!("runtime:{command}"), value);
}
Value::Object(object) if object.contains_key("dependentTasksOutputFiles") => {
if object
.keys()
.any(|key| !matches!(key.as_str(), "dependentTasksOutputFiles" | "transitive"))
{
bail!("unsupported output input field");
}
let pattern = object["dependentTasksOutputFiles"]
.as_str()
.context("dependentTasksOutputFiles must be a glob")?;
self.snapshot.pattern(pattern, false)?;
let transitive = object
.get("transitive")
.map(|value| value.as_bool().context("transitive must be boolean"))
.transpose()?
.unwrap_or(false);
let value = self
.values
.entry(format!("dependentTasksOutputFiles:{pattern}"))
.or_insert(json!(false));
if transitive {
*value = json!(true);
}
}
Value::Object(object)
if object.len() == 1 && object.contains_key("externalDependencies") =>
{
let names: Vec<String> =
serde_json::from_value(object["externalDependencies"].clone())?;
self.external.extend(names);
}
_ => bail!("unsupported task input declaration"),
}
Ok(())
}
fn fileset(&mut self, project: &str, pattern: &str) -> Result<()> {
let (exclude, pattern) = pattern
.strip_prefix('!')
.map(|pattern| (true, pattern))
.unwrap_or((false, pattern));
let scope = if pattern.contains("{workspaceRoot}") {
FileScope::Workspace
} else {
FileScope::Project(project.to_owned())
};
let pattern = paths::expand(self.workspace, project, pattern)?;
let matcher = self.snapshot.pattern(&pattern, exclude)?;
let prefix = pattern
.split('/')
.take_while(|part| is_literal(part))
.collect::<Vec<_>>()
.join("/");
let candidates: Box<dyn Iterator<Item = &String>> = if prefix.is_empty() {
Box::new(self.snapshot.files.iter())
} else {
Box::new(under(&self.snapshot.files, &prefix))
};
let selection = self.scopes.entry(scope).or_default();
let selected = if exclude {
&mut selection.excluded
} else {
&mut selection.included
};
selected.extend(candidates.filter(|file| matcher.is_match(file)).cloned());
selection.patterns.push(FilePattern {
matcher,
prefix,
excluded: exclude,
});
Ok(())
}
fn dependency_closure(&self, project: &str) -> BTreeSet<String> {
let mut closure = BTreeSet::new();
let mut pending = vec![project.to_owned()];
while let Some(current) = pending.pop() {
for edge in &self.snapshot.projects.dependencies[¤t] {
if edge.target != project && closure.insert(edge.target.clone()) {
pending.push(edge.target.clone());
}
}
}
closure
}
fn dependencies_named(&mut self, project: &str, name: &str) -> Result<()> {
for dependency in self.dependency_closure(project) {
if self.expanded.insert((dependency.clone(), name.to_owned())) {
self.named(&dependency, name)?;
}
}
Ok(())
}
fn json(&mut self, project: &str, object: &serde_json::Map<String, Value>) -> Result<()> {
for key in object.keys() {
if !matches!(key.as_str(), "json" | "fields" | "excludeFields") {
bail!("unsupported json input field {key:?}");
}
}
let path = object["json"]
.as_str()
.context("json input must name a file")?;
let path = paths::expand(self.workspace, project, path)?;
paths::validate_path(&path)?;
let list = |name: &str| -> Result<Option<Vec<String>>> {
object
.get(name)
.map(|value| {
serde_json::from_value(value.clone())
.with_context(|| format!("json input {name} must be an array of strings"))
})
.transpose()
};
let fields = list("fields")?;
let exclude = list("excludeFields")?.unwrap_or_default();
let value = match std::fs::read_to_string(self.workspace.root.join(&path)) {
Ok(text) => {
let mut value: Value =
jsonc_parser::parse_to_serde_value(&text, &Default::default())
.with_context(|| format!("cannot parse json input {path}"))?;
if let Some(fields) = &fields {
let mut kept = json!({});
for field in fields {
if let Some(found) = json_path(&value, field) {
set_json_path(&mut kept, field, found.clone());
}
}
value = kept;
}
for field in &exclude {
remove_json_path(&mut value, field);
}
value
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => Value::Null,
Err(error) => return Err(error.into()),
};
self.values.insert(format!("json:{path}"), value);
Ok(())
}
fn named(&mut self, project: &str, name: &str) -> Result<()> {
let identity = (project.to_owned(), name.to_owned());
if self.named_stack.contains(&identity) {
bail!("named input cycle at {project}:{name}");
}
self.named_stack.push(identity);
let inputs = self.workspace.projects[project]
.named_inputs
.get(name)
.cloned()
.or_else(|| (name == "default").then(|| vec![json!("{projectRoot}/**/*")]))
.with_context(|| format!("unknown named input {name:?}"))?;
for input in &inputs {
self.input(project, input)?;
}
self.named_stack.pop();
Ok(())
}
}
pub fn resolve(
snapshot: &Snapshot,
workspace: &Workspace,
task: &Task,
prepared: Option<&PreparedTask>,
cancelled: &AtomicBool,
) -> Result<Resolved> {
let mut resolver = Resolver {
workspace,
snapshot,
selected: BTreeSet::new(),
scopes: BTreeMap::new(),
values: BTreeMap::new(),
named_stack: Vec::new(),
expanded: BTreeSet::new(),
external: BTreeSet::new(),
prepared,
cancelled,
};
let default = vec![json!("default"), json!("^default")];
for input in task.definition.inputs.as_ref().unwrap_or(&default) {
resolver.input(&task.project, input)?;
}
let mut upstream = BTreeSet::new();
let mut pending: Vec<_> = task.dependencies.iter().cloned().collect();
while let Some(id) = pending.pop() {
if upstream.insert(id.clone())
&& let Some(dependencies) = snapshot.task_dependencies.get(&id)
{
pending.extend(dependencies.iter().cloned());
}
}
let own_outputs = Outputs::new(workspace, task)
.ok()
.filter(Outputs::is_explicit);
for (producer, declared) in &snapshot.generated_tasks {
if producer.id == task.id || !upstream.contains(&producer.id) {
continue;
}
let overlaps = |a: &str, b: &str| {
a.is_empty()
|| a == b
|| a.strip_prefix(b).is_some_and(|rest| rest.starts_with('/'))
|| b.strip_prefix(a).is_some_and(|rest| rest.starts_with('/'))
};
let relevant = resolver
.scopes
.values()
.flat_map(|scope| &scope.patterns)
.filter(|pattern| !pattern.excluded)
.any(|pattern| {
declared.anchors().any(|anchor| {
overlaps(&pattern.prefix, anchor)
&& (!snapshot.source_ignore.matches_directory(anchor)
|| !workspace.root.join(anchor).is_dir())
})
});
if !relevant {
continue;
}
let outputs = snapshot.dependency_outputs(workspace, producer)?;
for path in &outputs.files {
if snapshot.source_ignore.matches(path)
|| own_outputs.as_ref().is_some_and(|own| own.matches(path))
{
continue;
}
for selection in resolver.scopes.values_mut() {
for pattern in &selection.patterns {
if pattern.matcher.is_match(path) {
if pattern.excluded {
selection.excluded.insert(path.clone());
} else {
selection.included.insert(path.clone());
}
}
}
}
}
}
for selection in resolver.scopes.values() {
resolver.selected.extend(
selection
.included
.difference(&selection.excluded)
.filter(|path| own_outputs.as_ref().is_none_or(|own| !own.matches(path)))
.cloned(),
);
}
let readable = snapshot
.installed(&workspace.root)?
.is_some_and(|installed| installed.lockfile.is_some());
let tsconfig = ["tsconfig.base.json", "tsconfig.json"]
.into_iter()
.find(|name| {
workspace.root.join(name).is_file() || snapshot.extra_candidates.contains(*name)
});
let mut mandatory = BTreeSet::new();
for path in tsconfig.into_iter().chain([
"nx.json",
".gitignore",
".nxignore",
qk_config::LOCAL_WORKSPACE,
"package.json",
"pnpm-workspace.yaml",
"pnpm-lock.yaml",
"package-lock.json",
"yarn.lock",
"bun.lock",
]) {
if (workspace.root.join(path).is_file() || snapshot.extra_candidates.contains(path))
&& !(path == "pnpm-lock.yaml" && readable)
{
resolver.selected.insert(path.into());
mandatory.insert(path.into());
}
}
if let Some(extended) = &workspace.extended {
resolver.selected.insert(extended.clone());
mandatory.insert(extended.clone());
}
let mut packages = BTreeSet::from([task.project.clone()]);
let mut pending = packages.clone();
while let Some(project) = pending.pop_first() {
for edge in &snapshot.projects.dependencies[&project] {
if packages.insert(edge.target.clone()) {
pending.insert(edge.target.clone());
}
}
}
for project in &packages {
for name in ["project.json", qk_config::LOCAL_OVERRIDES, "package.json"] {
let path = Path::new(&workspace.projects[project].root).join(name);
let path = path.strip_prefix(".").unwrap_or(&path).to_owned();
if workspace.root.join(&path).is_file() {
let path = path
.to_str()
.context("project path must be UTF-8")?
.replace('\\', "/");
resolver.selected.insert(path.clone());
mandatory.insert(path);
}
}
}
let workspace_file = readable && resolver.selected.remove("pnpm-workspace.yaml");
if workspace_file {
mandatory.remove("pnpm-workspace.yaml");
}
let lockfile = readable.then(|| {
let importers = std::iter::once(".".to_owned())
.chain(
packages
.iter()
.map(|project| workspace.projects[project].root.clone()),
)
.collect();
(importers, resolver.external.clone())
});
Ok(Resolved {
files: resolver.selected,
mandatory,
values: resolver.values,
lockfile,
workspace_file,
external: resolver.external,
})
}
pub fn inputs(
snapshot: &Snapshot,
workspace: &Workspace,
task: &Task,
prepared: &PreparedTask,
dependencies: &BTreeMap<String, String>,
cancelled: &AtomicBool,
) -> Result<Value> {
let Resolved {
files,
mut values,
lockfile,
workspace_file,
..
} = resolve(snapshot, workspace, task, Some(prepared), cancelled)?;
let installed = snapshot.installed(&workspace.root)?;
if let (Some((importers, external)), Some(installed)) = (lockfile, &installed)
&& let Some(lockfile) = &installed.lockfile
{
let importers: BTreeMap<_, _> = importers
.iter()
.filter(|importer| lockfile.has_importer(importer))
.map(|importer| {
let digest = installed.digest(&format!("importer:{importer}"), || {
lockfile.installed(importer).unwrap_or_default()
});
(importer.clone(), digest)
})
.collect();
let external: BTreeMap<_, _> = external
.iter()
.map(|name| {
let digest =
installed.digest(&format!("package:{name}"), || lockfile.package(name));
(name.clone(), digest)
})
.collect();
values.insert(
"lockfile".into(),
json!({
"global": installed.digest("global", || BTreeSet::from([lockfile.global().to_string()])),
"importers": importers,
"external": external,
}),
);
}
let mut files = files
.iter()
.map(|path| {
let value = snapshot
.file_value(&workspace.root, path)
.with_context(|| format!("cannot read input {path}"))?;
Ok((path.clone(), value))
})
.collect::<Result<BTreeMap<_, _>>>()?;
if workspace_file {
match workspace_without_resolution(&workspace.root) {
Some(value) => {
values.insert("pnpm-workspace".into(), value);
}
None => {
files.insert(
"pnpm-workspace.yaml".into(),
snapshot.file_value(&workspace.root, "pnpm-workspace.yaml")?,
);
}
}
}
let mut definition = task.definition.clone();
definition.extra.remove("qk:warm");
definition.extra.remove("qk:threads");
let hash = json!({
"schema":"qk-local-v2", "qk":env!("CARGO_PKG_VERSION"),
"platform":[std::env::consts::OS, std::env::consts::ARCH], "workspace":snapshot.workspace_prefix,
"id":task.id, "args":task.args, "definition":definition, "packageManager":workspace.package_manager,
"files":files, "values":values, "dependencies":dependencies,
});
Ok(hash)
}
pub fn key(inputs: &Value) -> Result<String> {
Ok(blake3::hash(&serde_json::to_vec(inputs)?)
.to_hex()
.to_string())
}
pub fn fingerprint(
snapshot: &Snapshot,
workspace: &Workspace,
task: &Task,
prepared: &PreparedTask,
dependencies: &BTreeMap<String, String>,
cancelled: &AtomicBool,
) -> Result<String> {
key(&inputs(
snapshot,
workspace,
task,
prepared,
dependencies,
cancelled,
)?)
}
pub(crate) fn dependency_keys(
snapshot: &Snapshot,
workspace: &Workspace,
graph: &TaskGraph,
task: &Task,
dependencies: &BTreeMap<String, String>,
cancelled: &AtomicBool,
) -> Result<BTreeMap<String, String>> {
dependency_keys_impl(
snapshot,
workspace,
graph,
task,
dependencies,
cancelled,
false,
)
}
pub(crate) fn recheck_dependency_keys(
snapshot: &Snapshot,
workspace: &Workspace,
graph: &TaskGraph,
task: &Task,
dependencies: &BTreeMap<String, String>,
cancelled: &AtomicBool,
) -> Result<BTreeMap<String, String>> {
dependency_keys_impl(
snapshot,
workspace,
graph,
task,
dependencies,
cancelled,
true,
)
}
fn dependency_keys_impl(
snapshot: &Snapshot,
workspace: &Workspace,
graph: &TaskGraph,
task: &Task,
dependencies: &BTreeMap<String, String>,
cancelled: &AtomicBool,
fresh: bool,
) -> Result<BTreeMap<String, String>> {
let resolved = resolve(snapshot, workspace, task, None, cancelled)?;
let selections: Vec<_> = resolved.dependency_output_inputs().collect();
if selections.is_empty() {
if fresh {
return dependencies
.iter()
.map(|(id, key)| {
let outputs = Outputs::new(workspace, &graph.tasks[id])?;
let key = if outputs.declared() {
crate::output_fingerprint(&workspace.root, &outputs, "")?
} else {
key.clone()
};
Ok((id.clone(), key))
})
.collect();
}
return Ok(dependencies.clone());
}
let mut keys = BTreeMap::new();
let mut fresh_outputs = HashMap::new();
for (pattern, transitive) in selections {
let matcher = snapshot.pattern(pattern, false)?;
let mut selected = task.dependencies.clone();
if transitive {
let mut pending = selected.clone();
while let Some(id) = pending.pop_first() {
for dependency in &graph.tasks[&id].dependencies {
if selected.insert(dependency.clone()) {
pending.insert(dependency.clone());
}
}
}
}
for id in selected {
let outputs = if fresh {
if !fresh_outputs.contains_key(&id) {
fresh_outputs.insert(
id.clone(),
Arc::new(snapshot.fresh_dependency_outputs(workspace, &graph.tasks[&id])?),
);
}
fresh_outputs[&id].clone()
} else {
snapshot.dependency_outputs(workspace, &graph.tasks[&id])?
};
let mut files = BTreeMap::new();
for path in outputs.files.iter().filter(|path| matcher.is_match(path)) {
let known = outputs.values.lock().unwrap().get(path).cloned();
let value = match known {
Some(value) => value,
None => {
let value = snapshot.file_value(&workspace.root, path)?;
if value.get("link").is_none() {
outputs
.values
.lock()
.unwrap()
.insert(path.clone(), value.clone());
}
value
}
};
files.insert(path.clone(), value);
}
keys.insert(format!("{id}:{pattern}"), key(&json!(files))?);
}
}
Ok(keys)
}
#[cfg(test)]
mod tests {
use super::*;
use qk_taskgraph::Request;
fn fixture(outputs: Option<Value>) -> (tempfile::TempDir, Workspace, TaskGraph) {
let root = tempfile::tempdir().unwrap();
std::fs::write(root.path().join("nx.json"), "{}").unwrap();
let mut post = json!({"command": "true"});
if let Some(outputs) = outputs {
post["outputs"] = outputs;
}
std::fs::write(root.path().join("project.json"), json!({"name":"app", "targets": {
"gen": {"command":"true", "outputs":["generated"]},
"post": post,
"build": {"command":"true", "dependsOn":["gen", "post"], "inputs":[{"dependentTasksOutputFiles":"**/*"}]}
}}).to_string()).unwrap();
std::fs::create_dir(root.path().join("generated")).unwrap();
std::fs::write(root.path().join("generated/value"), "one").unwrap();
let workspace = Workspace::load(root.path()).unwrap();
let graph = TaskGraph::build(&workspace, &[Request::parse("app:build").unwrap()]).unwrap();
(root, workspace, graph)
}
#[test]
fn undeclared_writes_invalidate_dependency_values() {
for outputs in [None, Some(json!(["generated"]))] {
let (root, workspace, graph) = fixture(outputs);
let snapshot = Snapshot::new(&workspace, &graph, &root.path().join("cache")).unwrap();
let keys = || {
dependency_keys(
&snapshot,
&workspace,
&graph,
&graph.tasks["app:build"],
&BTreeMap::new(),
&AtomicBool::new(false),
)
.unwrap()
};
let before = keys();
std::fs::write(root.path().join("generated/value"), "different").unwrap();
snapshot.outputs_written(&workspace, &graph.tasks["app:post"]);
assert_ne!(before, keys());
}
}
#[cfg(unix)]
#[test]
fn dependency_symlinks_recheck_targets_outside_output_roots() {
let (root, mut workspace, _) = fixture(Some(json!(["generated"])));
workspace
.projects
.get_mut("app")
.unwrap()
.targets
.get_mut("gen")
.unwrap()
.outputs = Some(vec!["links".into()]);
std::fs::create_dir(root.path().join("links")).unwrap();
std::os::unix::fs::symlink("../generated/value", root.path().join("links/value")).unwrap();
let graph = TaskGraph::build(&workspace, &[Request::parse("app:build").unwrap()]).unwrap();
let snapshot = Snapshot::new(&workspace, &graph, &root.path().join("cache")).unwrap();
let keys = || {
dependency_keys(
&snapshot,
&workspace,
&graph,
&graph.tasks["app:build"],
&BTreeMap::new(),
&AtomicBool::new(false),
)
.unwrap()
};
let before = keys();
std::fs::write(root.path().join("generated/value"), "different").unwrap();
snapshot.outputs_written(&workspace, &graph.tasks["app:post"]);
assert_ne!(before["app:gen:**/*"], keys()["app:gen:**/*"]);
}
}