mod content;
use std::{
collections::{BTreeMap, BTreeSet},
path::Path,
sync::Arc,
};
use futures_util::{StreamExt, stream::FuturesUnordered};
use serde::{Deserialize, Serialize};
use serde_json::json;
use sha2::{Digest, Sha256};
use tokio::sync::Semaphore;
use tokio_util::sync::CancellationToken;
pub use crate::store::history::{Entry, list, remove, show};
pub(crate) use content::{HASH_BUFFER_BYTES, ignore_patterns};
use crate::{
config::{ArtifactName, Eval, EvalDeclaration, Fingerprint, Profile, RepoConfig},
process, runtime, scope,
store::{Execution, Request},
workspace,
};
const MANIFEST_BYTES: usize = 64 * 1024;
const MAX_SUMMARY_PATHS: usize = 10;
const FINGERPRINT_INPUT_VERSION: u32 = 1;
const MANIFEST_DIGEST_PREFIX_BYTES: usize = 8;
const KIND_KEY_FORMAT_PREFIX: &[u8] = b"artifactize-kind-key-v1\n";
const REUSE_KEY_FORMAT_PREFIX: &str = "artifactize-key-v2\n";
pub fn eval_definition_hash(eval: &EvalDeclaration) -> crate::types::DefinitionHash {
let kind = match &eval.profile() {
Profile::Agent { .. } => "agent",
Profile::Human {} => "human",
Profile::Runtime { .. } => "runtime",
};
let mut criteria = json!({
"kind": kind,
"payload": eval.payload(),
"passSchema": eval.pass_schema(),
"failSchema": eval.fail_schema(),
});
if let Profile::Runtime { command, args, .. } = &eval.profile() {
criteria["command"] = json!(command);
criteria["args"] = json!(args);
}
criteria.sort_all_objects();
content::hex(&Sha256::digest(
serde_json::to_vec(&criteria).expect("evaluation criteria is JSON"),
))
.parse()
.expect("a SHA-256 digest in hex is a definition hash")
}
#[derive(Debug, Clone)]
pub struct PreparedFingerprint {
pub value: crate::types::Fingerprint,
pub manifest: Option<Manifest>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct Manifest {
pub inputs: crate::types::Sha256Digest,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub files: Option<BTreeMap<String, crate::types::DigestPrefix>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Key {
pub value: crate::types::ReuseKey,
pub eval_def_hash: crate::types::DefinitionHash,
pub fingerprints: BTreeMap<crate::types::ArtifactName, crate::types::Fingerprint>,
pub artifact_kinds: BTreeMap<crate::types::ArtifactName, crate::config::ArtifactKind>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Unkeyed {
Target,
Dependency(ArtifactName),
}
#[derive(Debug, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct Changes {
pub since_run_id: crate::types::RunId,
#[serde(skip_serializing_if = "Option::is_none")]
pub files: Option<Vec<String>>,
#[serde(skip_serializing_if = "Option::is_none")]
pub dependencies: Option<Vec<String>>,
pub summary: String,
}
pub fn dependencies<'a>(config: &'a RepoConfig, eval: &'a Eval) -> BTreeSet<&'a ArtifactName> {
let target = &config.artifacts[&eval.target];
std::iter::once(&eval.target)
.chain(target.children.values())
.chain(&eval.deps)
.collect()
}
pub(crate) fn fingerprint_targets<'a>(
config: &'a RepoConfig,
required: &BTreeSet<&'a ArtifactName>,
) -> BTreeSet<&'a ArtifactName> {
config
.evals
.iter()
.filter(|eval| required.contains(&eval.target))
.flat_map(|eval| dependencies(config, eval))
.collect()
}
pub fn key(
eval_def_hash: &crate::types::DefinitionHash,
fingerprints: &BTreeMap<crate::types::ArtifactName, crate::types::Fingerprint>,
) -> crate::types::ReuseKey {
let mut digest = Sha256::new();
digest.update(format!("{REUSE_KEY_FORMAT_PREFIX}eval {eval_def_hash}\n"));
for (name, fingerprint) in fingerprints {
digest.update(format!("artifact {name} {fingerprint}\n"));
}
content::hex(&digest.finalize())
.parse()
.expect("SHA-256 is a reuse key")
}
pub fn key_with_kinds(
eval_def_hash: &crate::types::DefinitionHash,
fingerprints: &BTreeMap<crate::types::ArtifactName, crate::types::Fingerprint>,
kinds: &BTreeMap<crate::types::ArtifactName, crate::config::ArtifactKind>,
) -> crate::types::ReuseKey {
if kinds.is_empty() {
return key(eval_def_hash, fingerprints);
}
let mut digest = Sha256::new();
digest.update(KIND_KEY_FORMAT_PREFIX);
digest.update(key(eval_def_hash, fingerprints).as_bytes());
digest.update(serde_json::to_vec(kinds).expect("Artifact kinds are JSON"));
content::hex(&digest.finalize())
.parse()
.expect("SHA-256 is a reuse key")
}
pub fn eval_key(
config: &RepoConfig,
eval: &Eval,
fingerprints: &BTreeMap<&ArtifactName, PreparedFingerprint>,
) -> Result<Key, Unkeyed> {
if !fingerprints.contains_key(&eval.target) {
return Err(Unkeyed::Target);
}
let mut values: BTreeMap<ArtifactName, crate::types::Fingerprint> = BTreeMap::new();
for id in dependencies(config, eval) {
let fingerprint = fingerprints
.get(id)
.ok_or_else(|| Unkeyed::Dependency(id.clone()))?;
values.insert(id.clone(), fingerprint.value.clone());
}
let eval_def_hash = eval_definition_hash(&eval.declaration);
let artifact_kinds = values
.keys()
.map(|id| (id.clone(), config.artifacts[id].kind))
.collect();
Ok(Key {
value: key_with_kinds(&eval_def_hash, &values, &artifact_kinds),
eval_def_hash,
fingerprints: values,
artifact_kinds,
})
}
pub fn eval_keys<'a>(
config: &'a RepoConfig,
fingerprints: &BTreeMap<&ArtifactName, PreparedFingerprint>,
) -> BTreeMap<&'a crate::types::EvalId, Key> {
config
.evals
.iter()
.filter_map(|eval| Some((&eval.id, eval_key(config, eval, fingerprints).ok()?)))
.collect()
}
#[derive(Debug, Clone)]
pub struct Parallelism {
slots: Arc<Semaphore>,
limit: usize,
}
impl Parallelism {
pub fn new(limit: usize) -> Self {
let limit = limit.max(1);
Self {
slots: Arc::new(Semaphore::new(limit)),
limit,
}
}
pub fn available() -> usize {
std::thread::available_parallelism().map_or(1, usize::from)
}
pub fn limit(&self) -> usize {
self.limit
}
}
pub(crate) fn validate_file_inputs(config: &RepoConfig, eval: &Eval) -> Result<(), String> {
let scope = scope::eval_scope(config, eval).map_err(|error| error.to_string())?;
for artifact in scope.artifacts.values() {
scope::validate_file_target(&config.root, artifact).map_err(|error| error.to_string())?;
}
Ok(())
}
pub async fn prepare<'a>(
config: &RepoConfig,
artifacts: impl IntoIterator<Item = &'a ArtifactName>,
output_root: &Path,
parallelism: &Parallelism,
cancellation: CancellationToken,
) -> Result<BTreeMap<&'a ArtifactName, PreparedFingerprint>, String> {
let artifacts: Vec<_> = artifacts.into_iter().collect();
for id in &artifacts {
scope::validate_file_target(&config.root, &config.artifacts[*id])
.map_err(|error| error.to_string())?;
}
let mut seen = BTreeSet::new();
let ids: Vec<&'a ArtifactName> = artifacts
.into_iter()
.filter(|id| config.artifacts[*id].fingerprint.is_some() && seen.insert(*id))
.collect();
let tokens: Vec<_> = ids.iter().map(|_| cancellation.child_token()).collect();
let mut running: FuturesUnordered<_> = ids
.iter()
.zip(&tokens)
.enumerate()
.map(|(index, (id, token))| async move {
let result = match parallelism.slots.acquire().await {
Ok(_slot) if !token.is_cancelled() => {
compute(config, id, output_root, token.clone()).await
}
_ => Err(process::Error::Cancelled.to_string()),
};
(index, result)
})
.collect();
let mut prepared = BTreeMap::new();
let mut failure: Option<(usize, String)> = None;
while let Some((index, result)) = running.next().await {
match result {
Ok(fingerprint) => {
prepared.insert(ids[index], fingerprint);
}
Err(error) if failure.as_ref().is_none_or(|(first, _)| index < *first) => {
for token in &tokens[index + 1..] {
token.cancel();
}
failure = Some((index, error));
}
Err(_) => {}
}
}
match failure {
Some((_, error)) => Err(error),
None => Ok(prepared),
}
}
pub fn reuse(request: &mut Request, execution: &Execution, completed_at: crate::types::Timestamp) {
request.joined = request.execution_id.as_ref() == Some(&execution.id);
request.state = match execution.state.clone() {
crate::store::ExecutionState::Completed { result, .. } => {
crate::store::RequestState::completed(result, completed_at)
}
crate::store::ExecutionState::Failed { error, code, .. } => {
crate::store::RequestState::failed(error, code, completed_at)
}
state => state.into(),
};
request.execution_id = Some(execution.id.clone());
request.profile = execution.profile.clone();
request.options = execution.options.clone();
request.provenance = Some(execution.provenance.clone());
request.usage = None;
request.reused_usage = execution.usage.clone();
request.producer = execution.producer.clone();
request.session = execution
.producer
.as_ref()
.and_then(|producer| producer.session.clone());
request.reviewer = execution.reviewer.clone();
request.origin = execution.origin.clone();
request.queue = None;
request.blocked_reason = None;
}
pub async fn recheck(
config: &RepoConfig,
eval: &Eval,
output_root: &Path,
parallelism: &Parallelism,
cancellation: CancellationToken,
) -> Result<Option<crate::types::ReuseKey>, String> {
validate_file_inputs(config, eval)?;
let fingerprints = prepare(
config,
dependencies(config, eval),
output_root,
parallelism,
cancellation,
)
.await?;
validate_file_inputs(config, eval)?;
Ok(eval_key(config, eval, &fingerprints)
.ok()
.map(|key| key.value))
}
async fn compute(
config: &RepoConfig,
id: &ArtifactName,
output_root: &Path,
cancellation: CancellationToken,
) -> Result<PreparedFingerprint, String> {
scope::validate_file_target(&config.root, &config.artifacts[id])
.map_err(|error| error.to_string())?;
let result = match &config.artifacts[id].fingerprint {
Some(Fingerprint::Graph) => Ok(PreparedFingerprint {
value: crate::graph_structure::read(&config.root)?.fingerprint(),
manifest: None,
}),
Some(Fingerprint::Artifactsum { files, ignore }) => {
content(config, id, files, ignore, &cancellation)
.await
.map_err(|error| format!("Artifactsum for Artifact {id} failed: {error}"))
}
Some(Fingerprint::Script { .. }) => Ok(PreparedFingerprint {
value: script(config, id, output_root, cancellation)
.await
.map_err(|error| format!("Fingerprint script for Artifact {id} failed: {error}"))?,
manifest: None,
}),
None => Err(format!("Artifact {id} declares fingerprint: false.")),
}?;
scope::validate_file_target(&config.root, &config.artifacts[id])
.map_err(|error| error.to_string())?;
Ok(result)
}
async fn content(
config: &RepoConfig,
id: &ArtifactName,
inputs: &[crate::config::LogicalPath],
ignore: &[String],
cancellation: &CancellationToken,
) -> Result<PreparedFingerprint, String> {
let files = content::files(config, id, inputs, ignore, cancellation).await?;
let mut manifest = Manifest {
inputs: files.digest.clone(),
files: Some(
files
.files
.iter()
.map(|(path, file)| {
(
path.to_string(),
content::hex(&file.as_bytes()[..MANIFEST_DIGEST_PREFIX_BYTES])
.parse()
.expect("eight-byte SHA-256 prefix"),
)
})
.collect(),
),
};
if serde_json::to_vec(&manifest)
.expect("manifest is JSON")
.len()
> MANIFEST_BYTES
{
manifest.files = None;
}
Ok(PreparedFingerprint {
value: format!("artifactsum:{}", files.digest)
.parse()
.expect("artifactsum digest is a fingerprint"),
manifest: Some(manifest),
})
}
pub fn changes(
previous: &Execution,
target: &ArtifactName,
current: &Key,
manifest: Option<&Manifest>,
) -> Changes {
let mut parts = Vec::new();
let mut files = None;
if previous.fingerprints.get(target) != current.fingerprints.get(target) {
match previous.manifest.as_ref().zip(manifest) {
Some((old, new)) if old.inputs == new.inputs => files = Some(Vec::new()),
Some((old, new)) => {
files = old
.files
.as_ref()
.zip(new.files.as_ref())
.map(|(old, new)| diff(old, new));
match &files {
Some(files) if !files.is_empty() => {
let shown = files
.iter()
.take(MAX_SUMMARY_PATHS)
.cloned()
.collect::<Vec<_>>()
.join(", ");
parts.push(match files.len() {
0..=MAX_SUMMARY_PATHS => format!("changed: {shown}"),
total => {
format!("changed: {shown} and {} more", total - MAX_SUMMARY_PATHS)
}
});
}
_ => parts.push("inputs changed".to_owned()),
}
}
None => parts.push("fingerprint changed".to_owned()),
}
}
let dependencies = (!previous.fingerprints.is_empty())
.then(|| {
let others = |map: &BTreeMap<crate::types::ArtifactName, crate::types::Fingerprint>| {
map.iter()
.filter(|(name, _)| *name != target)
.map(|(name, value)| (name.clone(), value.clone()))
.collect::<BTreeMap<_, _>>()
};
diff(
&others(&previous.fingerprints),
&others(¤t.fingerprints),
)
})
.filter(|dependencies| !dependencies.is_empty());
for dependency in dependencies.iter().flatten() {
parts.push(if let Some(id) = dependency.strip_prefix('+') {
format!("dependency {id} added")
} else if let Some(id) = dependency.strip_prefix('-') {
format!("dependency {id} removed")
} else {
format!("dependency {dependency} changed")
});
}
if parts.is_empty() {
parts.push("fingerprint changed".to_owned());
}
Changes {
since_run_id: previous.provenance.run_id.clone(),
files,
dependencies,
summary: parts.join("; "),
}
}
fn diff<K: Ord + std::fmt::Display, T: PartialEq>(
old: &BTreeMap<K, T>,
new: &BTreeMap<K, T>,
) -> Vec<String> {
old.keys()
.chain(new.keys())
.collect::<BTreeSet<_>>()
.into_iter()
.filter_map(|key| match (old.get(key), new.get(key)) {
(Some(old), Some(new)) if old != new => Some(key.to_string()),
(None, Some(_)) => Some(format!("+{key}")),
(Some(_), None) => Some(format!("-{key}")),
_ => None,
})
.collect()
}
async fn script(
config: &RepoConfig,
artifact_id: &ArtifactName,
output_root: &Path,
cancellation: CancellationToken,
) -> Result<crate::types::Fingerprint, String> {
if cancellation.is_cancelled() {
return Err(process::Error::Cancelled.to_string());
}
let artifact = &config.artifacts[artifact_id];
let Some(Fingerprint::Script {
command,
args,
files,
timeout_ms,
}) = &artifact.fingerprint
else {
unreachable!("fingerprint script")
};
let cwd = scope::scoped_path(&config.root, artifact.folder()).map_err(|e| e.to_string())?;
for input in files {
scope::scoped_path(&cwd, Path::new(input)).map_err(|e| e.to_string())?;
}
let scope = scope::argv_scope(config, artifact_id, args).map_err(|e| e.to_string())?;
let program = scope::executable(&config.root, &scope, artifact_id, command)
.map_err(|error| format!("Fingerprint executable could not be spawned: {error}"))?;
let args = scope::resolve_argv(config, &scope, artifact_id, args).map_err(|e| e.to_string())?;
let input = json!({"version":FINGERPRINT_INPUT_VERSION,"artifactId":artifact_id});
let output_root =
workspace::prepare_directory(output_root, &config.root).map_err(|e| e.to_string())?;
let disposable = tempfile::Builder::new()
.prefix("fingerprint-")
.tempdir_in(output_root)
.map_err(|e| e.to_string())?;
let result = async {
let mut command = runtime::Command::prepare(
program,
args.iter().map(Into::into).collect(),
&config.root,
disposable.path(),
*timeout_ms,
)
.map_err(|e| e.to_string())?;
command.cwd = cwd;
let output = command
.output(input.to_string().into_bytes(), cancellation.clone())
.await
.map_err(|e| e.to_string())?;
if !output.status.success() {
return Err(format!(
"Fingerprint command exited with {}.",
crate::platform::exit_description(&output.status)
));
}
if output.truncated {
return Err("Fingerprint command output exceeded its limit.".into());
}
validate_output(&output.stdout)
}
.await;
let cleaned = disposable.close();
if cancellation.is_cancelled() {
return Err(process::Error::Cancelled.to_string());
}
let value = result?;
cleaned.map_err(|_| "Fingerprint validation failed.".to_owned())?;
Ok(value)
}
fn validate_output(stdout: &[u8]) -> Result<crate::types::Fingerprint, String> {
let (value, endings) = crate::platform::fingerprint_line_ending(stdout);
if !(1..=crate::types::MAX_FINGERPRINT_BYTES).contains(&value.len())
|| !value
.iter()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b':' | b'-'))
{
return Err(format!(
"stdout must contain 1–128 characters from [A-Za-z0-9._:-], with at most one trailing {endings}."
));
}
String::from_utf8(value.to_vec())
.map_err(|error| error.to_string())?
.parse()
}
#[cfg(test)]
mod tests;