use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::fs::{self, File, OpenOptions};
use std::io::{BufReader, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Mutex, OnceLock};
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
use fs2::FileExt;
use semver::Version;
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use thiserror::Error;
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
pub struct ArtifactId(String);
impl ArtifactId {
pub fn new(value: impl Into<String>) -> Result<Self, ArtifactPathError> {
let value = value.into();
let path = Path::new(&value);
let is_one_normal_component = {
let mut components = path.components();
matches!(components.next(), Some(std::path::Component::Normal(_)))
&& components.next().is_none()
};
if value.is_empty()
|| value.contains('/')
|| value.contains('\\')
|| path.is_absolute()
|| !is_one_normal_component
{
return Err(ArtifactPathError::InvalidArtifactId { value });
}
Ok(Self(value))
}
pub fn as_str(&self) -> &str {
&self.0
}
}
impl fmt::Display for ArtifactId {
fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
formatter.write_str(self.as_str())
}
}
const STAGING_DIRECTORY: &str = ".mobench-staging";
const STAGING_QUARANTINE_DIRECTORY: &str = ".mobench-staging-quarantine";
const STAGING_LOCK_FILE: &str = ".mobench-staging.lock";
const WORKSPACE_LOCK_FILE: &str = ".mobench-workspace.lock";
const RUN_READER_LOCK_FILE: &str = ".mobench-run-reader.lock";
const RUN_RETENTION_QUARANTINE_DIRECTORY: &str = ".mobench-run-quarantine";
const LATEST_DIRECTORY: &str = ".mobench-latest";
const LATEST_STAGING_DIRECTORY: &str = "staging";
const LATEST_GENERATIONS_DIRECTORY: &str = "generations";
const LATEST_QUARANTINE_DIRECTORY: &str = "quarantine";
const LATEST_CURRENT_FILE: &str = "current";
const LATEST_LOCK_FILE: &str = ".mobench-latest.lock";
const LATEST_READER_LOCK_FILE: &str = ".mobench-reader.lock";
const LATEST_RETENTION_BOUNDARY_FILE: &str = ".mobench-retention-boundary.json";
const RETAIN_LATEST_GENERATIONS: usize = 8;
const RETAIN_PUBLISHED_RUNS: usize = 32;
const RETAIN_QUARANTINE_ENTRIES: usize = 16;
pub const RUN_MANIFEST_FILE: &str = "mobench-run-manifest.json";
pub const LATEST_MANIFEST_FILE: &str = "manifest.json";
const RUN_MANIFEST_VERSION: u32 = 1;
const LATEST_MANIFEST_VERSION: u32 = 1;
const LATEST_LOCK_TIMEOUT: Duration = Duration::from_secs(5);
const LATEST_LOCK_POLL_INTERVAL: Duration = Duration::from_millis(5);
const MAX_ALLOCATION_ATTEMPTS: usize = 1_024;
static WORKSPACE_SEQUENCE: AtomicU64 = AtomicU64::new(0);
static ACTIVE_WORKSPACES: OnceLock<Mutex<BTreeSet<PathBuf>>> = OnceLock::new();
#[derive(Debug)]
pub struct RunWorkspace {
root: ApprovedRoot,
logical_id: ArtifactId,
expected_latest_generation: Option<String>,
workspace_lock: Option<File>,
staging_path: Option<PathBuf>,
published_path: PathBuf,
}
#[derive(Debug, Clone)]
pub struct PublishedRun {
root: ApprovedRoot,
path: PathBuf,
_reader_lease: Arc<File>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct ManifestArtifact {
pub relative_path: String,
pub size: u64,
pub sha256: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct RunManifest {
pub format_version: u32,
pub producer_version: String,
pub logical_id: String,
pub publication_id: String,
pub expected_latest_generation: Option<String>,
pub artifacts: Vec<ManifestArtifact>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LatestManifestArtifact {
pub source_relative_path: String,
pub destination_relative_path: String,
pub size: u64,
pub sha256: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct LatestManifest {
pub format_version: u32,
pub producer_version: String,
pub generation: String,
pub predecessor_generation: Option<String>,
pub source_logical_id: String,
pub source_publication_id: String,
pub artifacts: Vec<LatestManifestArtifact>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct RetentionBoundary {
format_version: u32,
predecessor_generation: String,
}
#[derive(Debug, Clone)]
pub struct LatestSnapshot {
root: ApprovedRoot,
generation_path: PathBuf,
manifest: LatestManifest,
_reader_lease: Arc<File>,
_source_run_lease: Arc<File>,
#[cfg(unix)]
generation_directory: Arc<File>,
}
#[derive(Debug)]
struct LatestUpdateLock {
file: File,
}
#[derive(Debug)]
struct StableAliasTransaction {
path: PathBuf,
root_path: PathBuf,
installed: Vec<PathBuf>,
backups: Vec<(PathBuf, PathBuf)>,
}
#[derive(Debug)]
enum PointerCommitFailure {
BeforeCommit(ArtifactPathError),
AfterCommit(ArtifactPathError),
}
impl StableAliasTransaction {
fn rollback(self, original: ArtifactPathError) -> ArtifactPathError {
let error =
latest_update_failure(original, &self.installed, &self.backups, self.path.clone());
let rollback_complete = !matches!(&error, ArtifactPathError::LatestRollback { .. });
if rollback_complete {
let _ = fs::remove_dir_all(&self.path);
}
let _ = sync_directory(&self.root_path);
error
}
fn finish(self) {
let _ = fs::remove_dir_all(self.path);
}
}
impl Drop for LatestUpdateLock {
fn drop(&mut self) {
let _ = FileExt::unlock(&self.file);
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LatestArtifact {
source: ArtifactId,
destination: ArtifactId,
}
impl LatestArtifact {
pub fn same(id: ArtifactId) -> Self {
Self {
source: id.clone(),
destination: id,
}
}
pub fn new(source: ArtifactId, destination: ArtifactId) -> Self {
Self {
source,
destination,
}
}
}
impl RunWorkspace {
pub fn allocate(
root: impl AsRef<Path>,
logical_id: &ArtifactId,
) -> Result<Self, ArtifactPathError> {
let root_path = root.as_ref();
fs::create_dir_all(root_path).map_err(|source| ArtifactPathError::Io {
operation: "create artifact root",
path: root_path.to_path_buf(),
source,
})?;
let root = ApprovedRoot::existing(root_path)?;
let expected_latest_generation =
recover_latest_for_writer(&root)?.map(|snapshot| snapshot.manifest.generation);
let _staging_manager_lock = acquire_staging_manager_lock(&root)?;
let staging_root = root.prepare_dir(STAGING_DIRECTORY)?;
quarantine_abandoned_run_staging(&root, &staging_root)?;
for _ in 0..MAX_ALLOCATION_ATTEMPTS {
let nonce = workspace_nonce();
let staging_path = staging_root.join(&nonce);
let published_path = root.path().join(format!("{logical_id}--{nonce}"));
if fs::symlink_metadata(&published_path).is_ok() {
continue;
}
match fs::create_dir(&staging_path) {
Ok(()) => {
if fs::symlink_metadata(&published_path).is_ok() {
let _ = fs::remove_dir(&staging_path);
continue;
}
let workspace_lock_path = staging_path.join(WORKSPACE_LOCK_FILE);
let workspace_lock = match OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&workspace_lock_path)
{
Ok(file) => file,
Err(source) => {
let _ = fs::remove_dir_all(&staging_path);
return Err(ArtifactPathError::Io {
operation: "create run-workspace lock",
path: workspace_lock_path,
source,
});
}
};
if let Err(source) = FileExt::lock_exclusive(&workspace_lock) {
let _ = fs::remove_dir_all(&staging_path);
return Err(ArtifactPathError::Io {
operation: "lock run workspace",
path: workspace_lock_path,
source,
});
}
register_active_workspace(&staging_path);
return Ok(Self {
root,
logical_id: logical_id.clone(),
expected_latest_generation,
workspace_lock: Some(workspace_lock),
staging_path: Some(staging_path),
published_path,
});
}
Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "create run staging directory",
path: staging_path,
source,
});
}
}
}
Err(ArtifactPathError::WorkspaceAllocationExhausted {
logical_id: logical_id.clone(),
})
}
pub fn staging_path(&self) -> &Path {
self.staging_path
.as_deref()
.expect("unpublished workspace retains its staging path")
}
pub fn published_path(&self) -> &Path {
&self.published_path
}
pub fn root(&self) -> &ApprovedRoot {
&self.root
}
pub fn publish(
mut self,
required_files: &[ArtifactId],
) -> Result<PublishedRun, ArtifactPathError> {
let staging_path = self
.staging_path
.as_ref()
.expect("unpublished workspace retains its staging path")
.clone();
let published_path = self.published_path.clone();
match fs::symlink_metadata(&staging_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent { path: staging_path });
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: staging_path,
});
}
Ok(_) => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect run staging directory",
path: staging_path,
source,
});
}
}
let staging_relative = staging_path.strip_prefix(self.root.path()).map_err(|_| {
ArtifactPathError::InvalidRelativePath {
path: staging_path.clone(),
}
})?;
self.root.prepare_dir(staging_relative)?;
let staging_root = ApprovedRoot::existing(&staging_path)?;
for required_file in required_files {
let required_path = staging_root.prepare_file(required_file.as_str())?;
match fs::symlink_metadata(&required_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: required_path,
});
}
Ok(metadata) if !metadata.is_file() => {
return Err(ArtifactPathError::FileDestinationNotFile {
path: required_path,
});
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::RequiredArtifactMissing {
path: required_path,
});
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect required staged artifact",
path: required_path,
source,
});
}
}
}
let manifest_path = staging_path.join(RUN_MANIFEST_FILE);
match fs::symlink_metadata(&manifest_path) {
Ok(_) => {
return Err(ArtifactPathError::ReservedManifestPath {
path: manifest_path,
});
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect reserved run manifest path",
path: manifest_path,
source,
});
}
}
create_run_reader_lease_file(&staging_path)?;
let artifacts =
inspect_artifact_tree(&staging_path, &[WORKSPACE_LOCK_FILE, RUN_READER_LOCK_FILE])?;
let publication_id = published_path
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| ArtifactPathError::NonUtf8ArtifactPath {
path: published_path.clone(),
})?
.to_owned();
let manifest = RunManifest {
format_version: RUN_MANIFEST_VERSION,
producer_version: env!("CARGO_PKG_VERSION").to_owned(),
logical_id: self.logical_id.as_str().to_owned(),
publication_id,
expected_latest_generation: self.expected_latest_generation.clone(),
artifacts,
};
write_json_file(&manifest_path, &manifest, "write run manifest")?;
sync_artifact_tree(&staging_path)?;
match fs::symlink_metadata(&published_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: published_path,
});
}
Ok(_) => {
return Err(ArtifactPathError::PublicationDestinationExists {
path: published_path,
});
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect run publication destination",
path: published_path,
source,
});
}
}
let _staging_manager_lock = acquire_staging_manager_lock(&self.root)?;
let workspace_lock_path = staging_path.join(WORKSPACE_LOCK_FILE);
let workspace_lock = self.workspace_lock.take();
if let Some(workspace_lock) = workspace_lock.as_ref() {
let _ = FileExt::unlock(workspace_lock);
}
drop(workspace_lock);
fs::remove_file(&workspace_lock_path).map_err(|source| ArtifactPathError::Io {
operation: "remove run-workspace lock before publication",
path: workspace_lock_path,
source,
})?;
sync_directory(&staging_path)?;
rename_directory_noreplace(&staging_path, &published_path).map_err(|source| {
ArtifactPathError::Io {
operation: "publish completed run",
path: published_path.clone(),
source,
}
})?;
record_durability_event("publish_run");
let published_sync = sync_directory(
published_path
.parent()
.expect("published run always has an artifact-root parent"),
);
let staging_sync = sync_directory(
staging_path
.parent()
.expect("staging run always has a staging-root parent"),
);
unregister_active_workspace(&staging_path);
self.staging_path = None;
if let Err(source) = published_sync.and(staging_sync) {
return Err(ArtifactPathError::PublicationDurabilityUncertain {
path: published_path,
source: Box::new(source),
});
}
let reader_lease = open_shared_lease(
&published_path.join(RUN_READER_LOCK_FILE),
"open published-run reader lease",
"lock published-run reader lease",
)
.map_err(
|source| ArtifactPathError::PublicationPostCommitMaintenance {
path: published_path.clone(),
source: Box::new(source),
},
)?;
let published = PublishedRun {
root: self.root.clone(),
path: published_path,
_reader_lease: reader_lease,
};
prune_published_runs(&self.root).map_err(|source| {
ArtifactPathError::PublicationPostCommitMaintenance {
path: published.path.clone(),
source: Box::new(source),
}
})?;
Ok(published)
}
}
impl Drop for RunWorkspace {
fn drop(&mut self) {
if let Some(staging_path) = self.staging_path.as_ref() {
if let Ok(_manager_lock) = acquire_staging_manager_lock(&self.root) {
let _ = fs::remove_dir_all(staging_path);
}
unregister_active_workspace(staging_path);
}
}
}
impl PublishedRun {
pub fn path(&self) -> &Path {
&self.path
}
pub fn root(&self) -> &ApprovedRoot {
&self.root
}
pub fn manifest(&self) -> Result<RunManifest, ArtifactPathError> {
validate_run_manifest(&self.path)
}
pub fn refresh_latest(&self, artifacts: &[LatestArtifact]) -> Result<(), ArtifactPathError> {
let published_root = ApprovedRoot::existing(&self.path)?;
let mut prepared: Vec<(PathBuf, PathBuf, ArtifactId, ArtifactId)> =
Vec::with_capacity(artifacts.len());
for artifact in artifacts {
if prepared
.iter()
.any(|(_, _, _, destination)| destination == &artifact.destination)
{
return Err(ArtifactPathError::DuplicateLatestDestination {
id: artifact.destination.clone(),
});
}
let source = published_root.prepare_file(artifact.source.as_str())?;
match fs::symlink_metadata(&source) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent { path: source });
}
Ok(metadata) if !metadata.is_file() => {
return Err(ArtifactPathError::FileDestinationNotFile { path: source });
}
Ok(_) => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::RequiredArtifactMissing { path: source });
}
Err(source_error) => {
return Err(ArtifactPathError::Io {
operation: "inspect published latest source",
path: source,
source: source_error,
});
}
}
let destination = self.root.prepare_file(artifact.destination.as_str())?;
prepared.push((
source,
destination,
artifact.source.clone(),
artifact.destination.clone(),
));
}
let run_manifest = self.manifest()?;
maybe_mutate_latest_source_after_validation();
let latest_root_path = self.root.prepare_dir(LATEST_DIRECTORY)?;
let latest_root = ApprovedRoot::existing(&latest_root_path)?;
latest_root.prepare_dir(LATEST_STAGING_DIRECTORY)?;
latest_root.prepare_dir(LATEST_GENERATIONS_DIRECTORY)?;
latest_root.prepare_dir(LATEST_QUARANTINE_DIRECTORY)?;
let (generation, staging_path, staging_lease) = {
let _latest_update_lock = acquire_latest_update_lock(&self.root)?;
let (generation, staging_path) = allocate_latest_generation(&latest_root)?;
let lease_result = (|| {
create_reader_lease_file(&staging_path)?;
let staging_lease_path = staging_path.join(LATEST_READER_LOCK_FILE);
let staging_lease = OpenOptions::new()
.read(true)
.write(true)
.open(&staging_lease_path)
.map_err(|source| ArtifactPathError::Io {
operation: "open active latest-generation staging lease",
path: staging_lease_path.clone(),
source,
})?;
FileExt::lock_exclusive(&staging_lease).map_err(|source| {
ArtifactPathError::Io {
operation: "lock active latest-generation staging lease",
path: staging_lease_path,
source,
}
})?;
Ok(staging_lease)
})();
match lease_result {
Ok(staging_lease) => (generation, staging_path, staging_lease),
Err(error) => {
let _ = quarantine_latest_path(&latest_root, &staging_path);
return Err(error);
}
}
};
let mut staging_lease = Some(staging_lease);
let preparation_result = (|| {
let mut manifest_artifacts = Vec::with_capacity(prepared.len());
for (source, _, source_id, destination_id) in &prepared {
let staged = staging_path.join(destination_id.as_str());
copy_new_synced(source, &staged, "stage latest generation artifact")?;
let (size, sha256) = digest_file(&staged)?;
let expected = run_manifest
.artifacts
.iter()
.find(|artifact| artifact.relative_path == source_id.as_str())
.ok_or_else(|| ArtifactPathError::ManifestIntegrity {
path: self.path.join(RUN_MANIFEST_FILE),
detail: format!(
"latest source `{source_id}` is not recorded in the run manifest"
),
})?;
if expected.size != size || expected.sha256 != sha256 {
return Err(ArtifactPathError::ManifestIntegrity {
path: source.clone(),
detail: format!(
"latest source `{source_id}` changed after run-manifest validation"
),
});
}
manifest_artifacts.push(LatestManifestArtifact {
source_relative_path: source_id.as_str().to_owned(),
destination_relative_path: destination_id.as_str().to_owned(),
size,
sha256,
});
}
manifest_artifacts.sort_by(|left, right| {
left.destination_relative_path
.cmp(&right.destination_relative_path)
});
sync_artifact_tree(&staging_path)?;
Ok(manifest_artifacts)
})();
let manifest_artifacts = match preparation_result {
Ok(manifest_artifacts) => manifest_artifacts,
Err(error) => {
if let Some(lease) = staging_lease.take() {
let _ = FileExt::unlock(&lease);
drop(lease);
}
if let Err(cleanup) = quarantine_latest_path(&latest_root, &staging_path) {
return Err(ArtifactPathError::LatestStagingCleanup {
original: Box::new(error),
cleanup: Box::new(cleanup),
staging_path,
});
}
return Err(error);
}
};
let generation_path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(&generation);
let mut generation_committed = false;
let generation_result = (|| {
let _latest_update_lock = acquire_latest_update_lock(&self.root)?;
let predecessor = recover_latest_state_locked(&self.root, &latest_root)?;
let observed_generation = predecessor
.as_ref()
.map(|snapshot| snapshot.manifest.generation.clone());
if run_manifest.expected_latest_generation != observed_generation {
return Err(ArtifactPathError::StaleLatestGeneration {
expected: run_manifest.expected_latest_generation.clone(),
observed: observed_generation,
});
}
let manifest = LatestManifest {
format_version: LATEST_MANIFEST_VERSION,
producer_version: env!("CARGO_PKG_VERSION").to_owned(),
generation: generation.clone(),
predecessor_generation: predecessor
.as_ref()
.map(|snapshot| snapshot.manifest.generation.clone()),
source_logical_id: run_manifest.logical_id.clone(),
source_publication_id: run_manifest.publication_id.clone(),
artifacts: manifest_artifacts.clone(),
};
write_json_file(
&staging_path.join(LATEST_MANIFEST_FILE),
&manifest,
"write latest-generation manifest",
)?;
sync_directory(&staging_path)?;
fs::rename(&staging_path, &generation_path).map_err(|source| {
ArtifactPathError::Io {
operation: "publish latest generation",
path: generation_path.clone(),
source,
}
})?;
record_durability_event("publish_generation");
sync_directory(&latest_root.path().join(LATEST_STAGING_DIRECTORY))?;
sync_directory(&latest_root.path().join(LATEST_GENERATIONS_DIRECTORY))?;
if let Some(lease) = staging_lease.take() {
let _ = FileExt::unlock(&lease);
drop(lease);
}
let generation_manifest = validate_latest_manifest(&generation_path, &generation)?;
let candidate = make_latest_snapshot(
self.root.clone(),
generation_path.clone(),
generation_manifest,
)?;
let alias_transaction =
install_stable_aliases_transactional(&self.root, &latest_root, &candidate)?;
match commit_current_pointer(&latest_root, &generation) {
Ok(()) => {
generation_committed = true;
alias_transaction.finish();
}
Err(PointerCommitFailure::BeforeCommit(pointer_error)) => {
return Err(alias_transaction.rollback(pointer_error));
}
Err(PointerCommitFailure::AfterCommit(durability_error)) => {
generation_committed = true;
alias_transaction.finish();
return Err(ArtifactPathError::LatestCommitDurabilityUncertain {
generation: generation.clone(),
source: Box::new(durability_error),
});
}
}
prune_latest_generations(&latest_root, candidate.manifest()).map_err(|source| {
ArtifactPathError::LatestRetentionAfterCommit {
generation: generation.clone(),
source: Box::new(source),
}
})?;
Ok(())
})();
if generation_result.is_err() {
if let Some(lease) = staging_lease.take() {
let _ = FileExt::unlock(&lease);
drop(lease);
}
if staging_path.exists() {
quarantine_latest_path(&latest_root, &staging_path)?;
} else if generation_path.exists() && !generation_committed {
quarantine_latest_path(&latest_root, &generation_path)?;
}
}
generation_result
}
}
impl LatestSnapshot {
pub fn open(root: impl AsRef<Path>) -> Result<Self, ArtifactPathError> {
let root = ApprovedRoot::existing(root)?;
load_current_snapshot(&root, false)?.ok_or_else(|| {
ArtifactPathError::LatestSnapshotUnavailable {
path: root.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE),
}
})
}
pub fn recover_stable_aliases(root: impl AsRef<Path>) -> Result<Self, ArtifactPathError> {
let root = ApprovedRoot::existing(root)?;
let snapshot = recover_latest_for_writer(&root)?.ok_or_else(|| {
ArtifactPathError::LatestSnapshotUnavailable {
path: root.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE),
}
})?;
Ok(snapshot)
}
pub fn generation(&self) -> &str {
&self.manifest.generation
}
pub fn path(&self) -> &Path {
&self.generation_path
}
pub fn root(&self) -> &ApprovedRoot {
&self.root
}
pub fn manifest(&self) -> &LatestManifest {
&self.manifest
}
pub fn open_artifact(&self, id: &ArtifactId) -> Result<File, ArtifactPathError> {
if !self
.manifest
.artifacts
.iter()
.any(|artifact| artifact.destination_relative_path == id.as_str())
{
return Err(ArtifactPathError::LatestArtifactNotInSnapshot { id: id.clone() });
}
let path = self.generation_path.join(id.as_str());
open_snapshot_artifact(self, id).map_err(|source| ArtifactPathError::Io {
operation: "open latest snapshot artifact",
path,
source,
})
}
pub fn read_artifact(&self, id: &ArtifactId) -> Result<Vec<u8>, ArtifactPathError> {
let expected = self
.manifest
.artifacts
.iter()
.find(|artifact| artifact.destination_relative_path == id.as_str())
.ok_or_else(|| ArtifactPathError::LatestArtifactNotInSnapshot { id: id.clone() })?;
let mut file = self.open_artifact(id)?;
let mut contents = Vec::new();
file.read_to_end(&mut contents)
.map_err(|source| ArtifactPathError::Io {
operation: "read latest snapshot artifact",
path: self.generation_path.join(id.as_str()),
source,
})?;
let actual_digest = sha256_bytes(&contents);
if contents.len() as u64 != expected.size || actual_digest != expected.sha256 {
return Err(ArtifactPathError::CorruptGeneration {
path: self.generation_path.join(id.as_str()),
detail: "artifact bytes changed after snapshot validation".to_owned(),
});
}
Ok(contents)
}
}
fn acquire_latest_update_lock(root: &ApprovedRoot) -> Result<LatestUpdateLock, ArtifactPathError> {
acquire_named_lock(
root,
LATEST_LOCK_FILE,
"open latest-artifact lock",
"lock latest artifacts",
)
}
fn acquire_staging_manager_lock(
root: &ApprovedRoot,
) -> Result<LatestUpdateLock, ArtifactPathError> {
acquire_named_lock(
root,
STAGING_LOCK_FILE,
"open run-staging lock",
"lock run staging",
)
}
fn acquire_named_lock(
root: &ApprovedRoot,
name: &str,
open_operation: &'static str,
lock_operation: &'static str,
) -> Result<LatestUpdateLock, ArtifactPathError> {
let path = root.prepare_file(name)?;
let file = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&path)
.map_err(|source| ArtifactPathError::Io {
operation: open_operation,
path: path.clone(),
source,
})?;
let started = Instant::now();
loop {
match FileExt::try_lock_exclusive(&file) {
Ok(()) => return Ok(LatestUpdateLock { file }),
Err(source)
if source.kind() == std::io::ErrorKind::WouldBlock
&& started.elapsed() < LATEST_LOCK_TIMEOUT =>
{
std::thread::sleep(LATEST_LOCK_POLL_INTERVAL);
}
Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {
return Err(ArtifactPathError::LatestLockTimeout { path });
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: lock_operation,
path,
source,
});
}
}
}
}
fn quarantine_abandoned_run_staging(
root: &ApprovedRoot,
staging_root: &Path,
) -> Result<(), ArtifactPathError> {
let quarantine_root = root.prepare_dir(STAGING_QUARANTINE_DIRECTORY)?;
let entries = fs::read_dir(staging_root).map_err(|source| ArtifactPathError::Io {
operation: "list run staging for recovery",
path: staging_root.to_path_buf(),
source,
})?;
for entry in entries {
let entry = entry.map_err(|source| ArtifactPathError::Io {
operation: "inspect run staging entry",
path: staging_root.to_path_buf(),
source,
})?;
let staging_path = entry.path();
if is_active_workspace(&staging_path) {
continue;
}
let metadata =
fs::symlink_metadata(&staging_path).map_err(|source| ArtifactPathError::Io {
operation: "inspect run staging workspace",
path: staging_path.clone(),
source,
})?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
quarantine_run_staging_path(staging_root, &quarantine_root, &staging_path)?;
continue;
}
let workspace_lock_path = staging_path.join(WORKSPACE_LOCK_FILE);
match fs::symlink_metadata(&workspace_lock_path) {
Ok(metadata) if metadata.file_type().is_symlink() || !metadata.is_file() => {
quarantine_run_staging_path(staging_root, &quarantine_root, &staging_path)?;
continue;
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect run-workspace lock",
path: workspace_lock_path,
source,
});
}
}
let workspace_lock = OpenOptions::new()
.read(true)
.write(true)
.create(true)
.truncate(false)
.open(&workspace_lock_path)
.map_err(|source| ArtifactPathError::Io {
operation: "open staged run-workspace lock",
path: workspace_lock_path.clone(),
source,
})?;
match FileExt::try_lock_exclusive(&workspace_lock) {
Ok(()) => {
quarantine_run_staging_path(staging_root, &quarantine_root, &staging_path)?;
}
Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "lock staged run workspace for recovery",
path: workspace_lock_path,
source,
});
}
}
}
prune_quarantine_entries(&quarantine_root, RETAIN_QUARANTINE_ENTRIES)
}
fn quarantine_run_staging_path(
staging_root: &Path,
quarantine_root: &Path,
staging_path: &Path,
) -> Result<(), ArtifactPathError> {
let name = staging_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("unreadable");
let quarantine_path = quarantine_root.join(format!("{name}--{}", workspace_nonce()));
fs::rename(staging_path, &quarantine_path).map_err(|source| ArtifactPathError::Io {
operation: "quarantine abandoned run workspace",
path: staging_path.to_path_buf(),
source,
})?;
sync_directory(staging_root)?;
sync_directory(quarantine_root)
}
fn allocate_latest_generation(
latest_root: &ApprovedRoot,
) -> Result<(String, PathBuf), ArtifactPathError> {
let staging_root = latest_root.prepare_dir(LATEST_STAGING_DIRECTORY)?;
for _ in 0..MAX_ALLOCATION_ATTEMPTS {
let generation = format!("generation-{}", workspace_nonce());
let path = staging_root.join(&generation);
match fs::create_dir(&path) {
Ok(()) => return Ok((generation, path)),
Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => continue,
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "create latest staging generation",
path,
source,
});
}
}
}
Err(ArtifactPathError::LatestTransactionAllocationExhausted)
}
fn quarantine_abandoned_latest_staging(
latest_root: &ApprovedRoot,
) -> Result<(), ArtifactPathError> {
let staging = latest_root.prepare_dir(LATEST_STAGING_DIRECTORY)?;
let entries = fs::read_dir(&staging).map_err(|source| ArtifactPathError::Io {
operation: "list abandoned latest staging",
path: staging.clone(),
source,
})?;
for entry in entries {
let entry = entry.map_err(|source| ArtifactPathError::Io {
operation: "inspect abandoned latest staging entry",
path: staging.clone(),
source,
})?;
let staging_path = entry.path();
let metadata =
fs::symlink_metadata(&staging_path).map_err(|source| ArtifactPathError::Io {
operation: "inspect abandoned latest staging entry",
path: staging_path.clone(),
source,
})?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
quarantine_latest_path(latest_root, &staging_path)?;
continue;
}
let lease_path = staging_path.join(LATEST_READER_LOCK_FILE);
let lease_metadata = match fs::symlink_metadata(&lease_path) {
Ok(metadata) => metadata,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
quarantine_latest_path(latest_root, &staging_path)?;
continue;
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect latest-generation staging lease",
path: lease_path,
source,
});
}
};
if lease_metadata.file_type().is_symlink() || !lease_metadata.is_file() {
quarantine_latest_path(latest_root, &staging_path)?;
continue;
}
let lease = OpenOptions::new()
.read(true)
.write(true)
.open(&lease_path)
.map_err(|source| ArtifactPathError::Io {
operation: "open latest-generation staging lease for recovery",
path: lease_path.clone(),
source,
})?;
match FileExt::try_lock_exclusive(&lease) {
Ok(()) => quarantine_latest_path(latest_root, &staging_path)?,
Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "lock latest-generation staging lease for recovery",
path: lease_path,
source,
});
}
}
}
Ok(())
}
fn quarantine_latest_path(
latest_root: &ApprovedRoot,
source_path: &Path,
) -> Result<(), ArtifactPathError> {
let quarantine = latest_root.prepare_dir(LATEST_QUARANTINE_DIRECTORY)?;
let source_name = source_path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("unreadable");
let destination = quarantine.join(format!("{source_name}--{}", workspace_nonce()));
fs::rename(source_path, &destination).map_err(|source| ArtifactPathError::Io {
operation: "quarantine abandoned latest staging",
path: source_path.to_path_buf(),
source,
})?;
sync_directory(
source_path
.parent()
.expect("latest staging entry always has a parent"),
)?;
sync_directory(&quarantine)?;
prune_quarantine_entries(&quarantine, RETAIN_QUARANTINE_ENTRIES)
}
fn commit_current_pointer(
latest_root: &ApprovedRoot,
generation: &str,
) -> Result<(), PointerCommitFailure> {
ArtifactId::new(generation.to_owned()).map_err(PointerCommitFailure::BeforeCommit)?;
let temporary = latest_root
.path()
.join(format!(".current-{}", workspace_nonce()));
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(&temporary)
.map_err(|source| {
PointerCommitFailure::BeforeCommit(ArtifactPathError::Io {
operation: "create latest pointer candidate",
path: temporary.clone(),
source,
})
})?;
file.write_all(generation.as_bytes())
.and_then(|()| file.write_all(b"\n"))
.map_err(|source| {
PointerCommitFailure::BeforeCommit(ArtifactPathError::Io {
operation: "write latest pointer candidate",
path: temporary.clone(),
source,
})
})?;
file.sync_all().map_err(|source| {
PointerCommitFailure::BeforeCommit(ArtifactPathError::Io {
operation: "sync latest pointer candidate",
path: temporary.clone(),
source,
})
})?;
record_durability_event("sync_pointer_file");
let current = latest_root.path().join(LATEST_CURRENT_FILE);
match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.is_dir() => {
let _ = fs::remove_file(&temporary);
return Err(PointerCommitFailure::BeforeCommit(
ArtifactPathError::FileDestinationNotFile { path: current },
));
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
let _ = fs::remove_file(&temporary);
return Err(PointerCommitFailure::BeforeCommit(ArtifactPathError::Io {
operation: "inspect current latest pointer",
path: current,
source,
}));
}
}
replace_file_atomically(&temporary, ¤t).map_err(|source| {
PointerCommitFailure::BeforeCommit(ArtifactPathError::Io {
operation: "commit latest pointer",
path: current,
source,
})
})?;
record_durability_event("commit_pointer");
sync_directory(latest_root.path()).map_err(PointerCommitFailure::AfterCommit)
}
fn recover_latest_for_writer(
root: &ApprovedRoot,
) -> Result<Option<LatestSnapshot>, ArtifactPathError> {
let _lock = acquire_latest_update_lock(root)?;
let latest_path = root.prepare_dir(LATEST_DIRECTORY)?;
let latest_root = ApprovedRoot::existing(latest_path)?;
latest_root.prepare_dir(LATEST_STAGING_DIRECTORY)?;
latest_root.prepare_dir(LATEST_GENERATIONS_DIRECTORY)?;
latest_root.prepare_dir(LATEST_QUARANTINE_DIRECTORY)?;
recover_latest_state_locked(root, &latest_root)
}
fn recover_latest_state_locked(
root: &ApprovedRoot,
latest_root: &ApprovedRoot,
) -> Result<Option<LatestSnapshot>, ArtifactPathError> {
let (snapshot, needs_alias_refresh) = match load_current_snapshot(root, true) {
Ok(Some(snapshot)) => {
quarantine_generations_outside_committed_chain(latest_root, &snapshot.manifest)?;
(Some(snapshot), true)
}
Ok(None) => (
recover_current_from_generation_chain(root, latest_root)?,
false,
),
Err(error) if is_recoverable_pointer_error(&error) => (
recover_current_from_generation_chain(root, latest_root)?,
false,
),
Err(error) => return Err(error),
};
if let Some(snapshot) = snapshot.as_ref() {
reject_producer_downgrade(&snapshot.manifest.producer_version)?;
if needs_alias_refresh {
install_stable_aliases_transactional(root, latest_root, snapshot)?.finish();
}
}
quarantine_abandoned_latest_staging(latest_root)?;
Ok(snapshot)
}
fn quarantine_generations_outside_committed_chain(
latest_root: &ApprovedRoot,
committed: &LatestManifest,
) -> Result<(), ArtifactPathError> {
let mut committed_chain = BTreeSet::new();
let mut cursor = committed.clone();
loop {
committed_chain.insert(cursor.generation.clone());
let Some(predecessor) = cursor.predecessor_generation.as_ref() else {
break;
};
let cursor_path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(&cursor.generation);
if load_retention_boundary(&cursor_path, &cursor)?.is_some() {
break;
}
let path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(predecessor);
cursor = validate_latest_manifest(&path, predecessor)?;
}
let generations = latest_root.path().join(LATEST_GENERATIONS_DIRECTORY);
let entries = fs::read_dir(&generations).map_err(|source| ArtifactPathError::Io {
operation: "list uncommitted latest generations",
path: generations,
source,
})?;
for entry in entries {
let entry = entry.map_err(|source| ArtifactPathError::Io {
operation: "inspect uncommitted latest generation",
path: latest_root.path().join(LATEST_GENERATIONS_DIRECTORY),
source,
})?;
let path = entry.path();
let generation = entry
.file_name()
.to_str()
.ok_or_else(|| ArtifactPathError::NonUtf8ArtifactPath { path: path.clone() })?
.to_owned();
if committed_chain.contains(&generation) {
continue;
}
quarantine_latest_path(latest_root, &path)?;
}
Ok(())
}
fn is_recoverable_pointer_error(error: &ArtifactPathError) -> bool {
fn is_current(path: &Path) -> bool {
path.file_name().and_then(|name| name.to_str()) == Some(LATEST_CURRENT_FILE)
}
match error {
ArtifactPathError::InvalidLatestPointer { .. } => true,
ArtifactPathError::LatestSnapshotUnavailable { path } => is_current(path),
ArtifactPathError::SymlinkComponent { path }
| ArtifactPathError::FileDestinationNotFile { path } => is_current(path),
ArtifactPathError::Io {
operation, path, ..
} => operation.contains("latest snapshot pointer") && is_current(path),
_ => false,
}
}
fn recover_current_from_generation_chain(
root: &ApprovedRoot,
latest_root: &ApprovedRoot,
) -> Result<Option<LatestSnapshot>, ArtifactPathError> {
let candidate = select_unique_generation_tip(root, latest_root)?;
quarantine_directory_pointer_for_recovery(latest_root)?;
let Some(candidate) = candidate else {
return Ok(None);
};
reject_producer_downgrade(&candidate.manifest.producer_version)?;
let alias_transaction = install_stable_aliases_transactional(root, latest_root, &candidate)?;
match commit_current_pointer(latest_root, candidate.generation()) {
Ok(()) => alias_transaction.finish(),
Err(PointerCommitFailure::BeforeCommit(error)) => {
return Err(alias_transaction.rollback(error));
}
Err(PointerCommitFailure::AfterCommit(error)) => {
alias_transaction.finish();
return Err(ArtifactPathError::LatestCommitDurabilityUncertain {
generation: candidate.generation().to_owned(),
source: Box::new(error),
});
}
}
Ok(Some(candidate))
}
fn quarantine_directory_pointer_for_recovery(
latest_root: &ApprovedRoot,
) -> Result<(), ArtifactPathError> {
let current = latest_root.path().join(LATEST_CURRENT_FILE);
let metadata = match fs::symlink_metadata(¤t) {
Ok(metadata) => metadata,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(()),
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect corrupt latest pointer for recovery",
path: current,
source,
});
}
};
if !metadata.is_dir() || metadata.file_type().is_symlink() {
return Ok(());
}
let quarantine = latest_root.prepare_dir(LATEST_QUARANTINE_DIRECTORY)?;
let destination = quarantine.join(format!("current-corrupt--{}", workspace_nonce()));
fs::rename(¤t, &destination).map_err(|source| ArtifactPathError::Io {
operation: "quarantine corrupt latest pointer directory",
path: current,
source,
})?;
sync_directory(latest_root.path())?;
sync_directory(&quarantine)
}
fn select_unique_generation_tip(
root: &ApprovedRoot,
latest_root: &ApprovedRoot,
) -> Result<Option<LatestSnapshot>, ArtifactPathError> {
let generations_path = latest_root.path().join(LATEST_GENERATIONS_DIRECTORY);
let mut entries = fs::read_dir(&generations_path)
.map_err(|source| ArtifactPathError::Io {
operation: "list latest generations for recovery",
path: generations_path.clone(),
source,
})?
.collect::<Result<Vec<_>, _>>()
.map_err(|source| ArtifactPathError::Io {
operation: "inspect latest generation for recovery",
path: generations_path.clone(),
source,
})?;
entries.sort_by_key(|entry| entry.file_name());
let mut manifests = BTreeMap::new();
let mut retention_boundaries = BTreeSet::new();
for entry in entries {
let path = entry.path();
let metadata = fs::symlink_metadata(&path).map_err(|source| ArtifactPathError::Io {
operation: "inspect latest generation for recovery",
path: path.clone(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent { path });
}
if !metadata.is_dir() {
return Err(ArtifactPathError::DirectoryComponentNotDirectory { path });
}
let generation = entry
.file_name()
.to_str()
.ok_or_else(|| ArtifactPathError::NonUtf8ArtifactPath { path: path.clone() })?
.to_owned();
ArtifactId::new(generation.clone())?;
let manifest = validate_latest_manifest(&path, &generation)?;
if load_retention_boundary(&path, &manifest)?.is_some() {
retention_boundaries.insert(generation.clone());
}
manifests.insert(generation, (path, manifest));
}
if manifests.is_empty() {
return Ok(None);
}
let mut children: BTreeMap<String, Vec<String>> = BTreeMap::new();
let mut predecessors = BTreeSet::new();
for (generation, (_, manifest)) in &manifests {
if retention_boundaries.contains(generation) {
continue;
}
if let Some(predecessor) = manifest.predecessor_generation.as_ref() {
let Some((_, predecessor_manifest)) = manifests.get(predecessor) else {
return Err(ArtifactPathError::BrokenGenerationChain {
generation: generation.clone(),
predecessor: predecessor.clone(),
});
};
reject_chain_downgrade(predecessor_manifest, manifest)?;
children
.entry(predecessor.clone())
.or_default()
.push(generation.clone());
predecessors.insert(predecessor.clone());
}
}
for (predecessor, children) in &children {
if children.len() > 1 {
return Err(ArtifactPathError::AmbiguousGenerationFork {
predecessor: predecessor.clone(),
children: children.clone(),
});
}
}
validate_generation_map_acyclic(&manifests, &retention_boundaries)?;
let tips: Vec<_> = manifests
.keys()
.filter(|generation| !predecessors.contains(*generation))
.cloned()
.collect();
if tips.len() != 1 {
return Err(ArtifactPathError::AmbiguousGenerationTips { tips });
}
let tip = &tips[0];
let (generation_path, manifest) = manifests
.remove(tip)
.expect("selected generation tip came from the manifest map");
Ok(Some(make_latest_snapshot(
root.clone(),
generation_path,
manifest,
)?))
}
fn validate_generation_map_acyclic(
manifests: &BTreeMap<String, (PathBuf, LatestManifest)>,
retention_boundaries: &BTreeSet<String>,
) -> Result<(), ArtifactPathError> {
let mut complete = BTreeSet::new();
for generation in manifests.keys() {
let mut chain = BTreeSet::new();
let mut cursor = generation.as_str();
while !complete.contains(cursor) {
if !chain.insert(cursor.to_owned()) {
return Err(ArtifactPathError::GenerationChainCycle {
generation: cursor.to_owned(),
});
}
if retention_boundaries.contains(cursor) {
break;
}
let Some(predecessor) = manifests
.get(cursor)
.and_then(|(_, manifest)| manifest.predecessor_generation.as_deref())
else {
break;
};
cursor = predecessor;
}
complete.extend(chain);
}
Ok(())
}
fn load_current_snapshot(
root: &ApprovedRoot,
allow_missing: bool,
) -> Result<Option<LatestSnapshot>, ArtifactPathError> {
let latest_path = root.path().join(LATEST_DIRECTORY);
match fs::symlink_metadata(&latest_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent { path: latest_path });
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory { path: latest_path });
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound && allow_missing => {
return Ok(None);
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::LatestSnapshotUnavailable { path: latest_path });
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect latest snapshot root",
path: latest_path,
source,
});
}
}
let latest_root = ApprovedRoot::existing(&latest_path)?;
let current = latest_root.path().join(LATEST_CURRENT_FILE);
let metadata = match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent { path: current });
}
Ok(metadata) if !metadata.is_file() => {
return Err(ArtifactPathError::FileDestinationNotFile { path: current });
}
Ok(metadata) => metadata,
Err(source) if source.kind() == std::io::ErrorKind::NotFound && allow_missing => {
return Ok(None);
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::LatestSnapshotUnavailable { path: current });
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect latest snapshot pointer",
path: current,
source,
});
}
};
if metadata.len() > 512 {
return Err(ArtifactPathError::InvalidLatestPointer { path: current });
}
let generation = fs::read_to_string(¤t)
.map_err(|source| ArtifactPathError::Io {
operation: "read latest snapshot pointer",
path: current.clone(),
source,
})?
.trim()
.to_owned();
ArtifactId::new(generation.clone()).map_err(|_| ArtifactPathError::InvalidLatestPointer {
path: current.clone(),
})?;
let generation_path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(&generation);
match fs::symlink_metadata(&generation_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: generation_path,
});
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: generation_path,
});
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::InvalidLatestPointer { path: current });
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect latest generation",
path: generation_path,
source,
});
}
}
let manifest = validate_latest_manifest(&generation_path, &generation)?;
validate_predecessor_chain(&latest_root, &manifest)?;
Ok(Some(make_latest_snapshot(
root.clone(),
generation_path,
manifest,
)?))
}
fn make_latest_snapshot(
root: ApprovedRoot,
generation_path: PathBuf,
manifest: LatestManifest,
) -> Result<LatestSnapshot, ArtifactPathError> {
let lease_path = generation_path.join(LATEST_READER_LOCK_FILE);
let reader_lease = Arc::new(OpenOptions::new().read(true).open(&lease_path).map_err(
|source| ArtifactPathError::Io {
operation: "open latest-generation reader lease",
path: lease_path.clone(),
source,
},
)?);
FileExt::lock_shared(reader_lease.as_ref()).map_err(|source| ArtifactPathError::Io {
operation: "lock latest-generation reader lease",
path: lease_path,
source,
})?;
let source_run_lease = open_shared_lease(
&root
.path()
.join(&manifest.source_publication_id)
.join(RUN_READER_LOCK_FILE),
"open latest source-run reader lease",
"lock latest source-run reader lease",
)?;
#[cfg(unix)]
let generation_directory = Arc::new(open_directory_no_follow(&generation_path).map_err(
|source| ArtifactPathError::Io {
operation: "pin latest generation directory",
path: generation_path.clone(),
source,
},
)?);
Ok(LatestSnapshot {
root,
generation_path,
manifest,
_reader_lease: reader_lease,
_source_run_lease: source_run_lease,
#[cfg(unix)]
generation_directory,
})
}
fn open_shared_lease(
path: &Path,
open_operation: &'static str,
lock_operation: &'static str,
) -> Result<Arc<File>, ArtifactPathError> {
let lease = Arc::new(OpenOptions::new().read(true).open(path).map_err(|source| {
ArtifactPathError::Io {
operation: open_operation,
path: path.to_path_buf(),
source,
}
})?);
FileExt::lock_shared(lease.as_ref()).map_err(|source| ArtifactPathError::Io {
operation: lock_operation,
path: path.to_path_buf(),
source,
})?;
Ok(lease)
}
fn create_run_reader_lease_file(directory: &Path) -> Result<(), ArtifactPathError> {
create_lease_file(
directory,
RUN_READER_LOCK_FILE,
"create published-run reader lease",
)
}
fn create_reader_lease_file(directory: &Path) -> Result<(), ArtifactPathError> {
create_lease_file(
directory,
LATEST_READER_LOCK_FILE,
"create latest-generation reader lease",
)
}
fn create_lease_file(
directory: &Path,
name: &str,
operation: &'static str,
) -> Result<(), ArtifactPathError> {
let path = directory.join(name);
let file = OpenOptions::new()
.read(true)
.write(true)
.create_new(true)
.open(&path)
.map_err(|source| ArtifactPathError::Io {
operation,
path: path.clone(),
source,
})?;
file.sync_all().map_err(|source| ArtifactPathError::Io {
operation: "sync latest-generation reader lease",
path,
source,
})
}
fn prune_latest_generations(
latest_root: &ApprovedRoot,
current: &LatestManifest,
) -> Result<(), ArtifactPathError> {
let generations_root = latest_root.path().join(LATEST_GENERATIONS_DIRECTORY);
let mut chain = Vec::new();
let mut cursor = current.clone();
loop {
let path = generations_root.join(&cursor.generation);
let boundary = load_retention_boundary(&path, &cursor)?;
chain.push((path, cursor.clone()));
if boundary.is_some() {
break;
}
let Some(predecessor) = cursor.predecessor_generation.as_ref() else {
break;
};
let predecessor_path = generations_root.join(predecessor);
cursor = validate_latest_manifest(&predecessor_path, predecessor)?;
}
if chain.len() <= RETAIN_LATEST_GENERATIONS {
return Ok(());
}
let mut leases = Vec::new();
for (path, _) in &chain[RETAIN_LATEST_GENERATIONS..] {
let lease_path = path.join(LATEST_READER_LOCK_FILE);
let lease = OpenOptions::new()
.read(true)
.write(true)
.open(&lease_path)
.map_err(|source| ArtifactPathError::Io {
operation: "open latest-generation retention lease",
path: lease_path.clone(),
source,
})?;
match FileExt::try_lock_exclusive(&lease) {
Ok(()) => leases.push(lease),
Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => return Ok(()),
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "lock latest generation for retention",
path: lease_path,
source,
});
}
}
}
let (oldest_retained_path, oldest_retained) = &chain[RETAIN_LATEST_GENERATIONS - 1];
let predecessor = oldest_retained
.predecessor_generation
.clone()
.expect("a prunable chain has a predecessor after the retention boundary");
let boundary = RetentionBoundary {
format_version: 1,
predecessor_generation: predecessor,
};
let boundary_path = oldest_retained_path.join(LATEST_RETENTION_BOUNDARY_FILE);
let boundary_temp =
oldest_retained_path.join(format!(".retention-boundary-{}", workspace_nonce()));
write_json_file(
&boundary_temp,
&boundary,
"write latest-generation retention boundary",
)?;
replace_file_atomically(&boundary_temp, &boundary_path).map_err(|source| {
ArtifactPathError::Io {
operation: "commit latest-generation retention boundary",
path: boundary_path.clone(),
source,
}
})?;
sync_directory(oldest_retained_path)?;
let quarantine = latest_root.prepare_dir(LATEST_QUARANTINE_DIRECTORY)?;
let mut retired = Vec::new();
for (path, _) in &chain[RETAIN_LATEST_GENERATIONS..] {
let name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("generation");
let destination = quarantine.join(format!("retired-{name}--{}", workspace_nonce()));
fs::rename(path, &destination).map_err(|source| ArtifactPathError::Io {
operation: "retire latest generation",
path: path.clone(),
source,
})?;
retired.push(destination);
}
sync_directory(&generations_root)?;
sync_directory(&quarantine)?;
drop(leases);
for path in retired {
fs::remove_dir_all(&path).map_err(|source| ArtifactPathError::Io {
operation: "delete retired latest generation",
path,
source,
})?;
}
sync_directory(&quarantine)
}
fn prune_published_runs(root: &ApprovedRoot) -> Result<(), ArtifactPathError> {
let mut runs = Vec::new();
for entry in fs::read_dir(root.path()).map_err(|source| ArtifactPathError::Io {
operation: "list published runs for retention",
path: root.path().to_path_buf(),
source,
})? {
let entry = entry.map_err(|source| ArtifactPathError::Io {
operation: "inspect published run for retention",
path: root.path().to_path_buf(),
source,
})?;
let path = entry.path();
let name = entry.file_name().to_string_lossy().into_owned();
if name.starts_with('.') {
continue;
}
let metadata = fs::symlink_metadata(&path).map_err(|source| ArtifactPathError::Io {
operation: "inspect published run type for retention",
path: path.clone(),
source,
})?;
if metadata.file_type().is_symlink() || !metadata.is_dir() {
continue;
}
if !path.join(RUN_MANIFEST_FILE).is_file() || !path.join(RUN_READER_LOCK_FILE).is_file() {
continue;
}
if validate_run_manifest(&path).is_err() {
continue;
}
let order_key = name
.rsplit_once("--")
.map(|(_, nonce)| nonce.to_owned())
.unwrap_or_else(|| name.clone());
runs.push((order_key, path));
}
runs.sort_by(|left, right| right.0.cmp(&left.0));
if runs.len() <= RETAIN_PUBLISHED_RUNS {
return Ok(());
}
let quarantine = root.prepare_dir(RUN_RETENTION_QUARANTINE_DIRECTORY)?;
for (_, path) in &runs[RETAIN_PUBLISHED_RUNS..] {
let lease_path = path.join(RUN_READER_LOCK_FILE);
let lease = OpenOptions::new()
.read(true)
.write(true)
.open(&lease_path)
.map_err(|source| ArtifactPathError::Io {
operation: "open published-run retention lease",
path: lease_path.clone(),
source,
})?;
match FileExt::try_lock_exclusive(&lease) {
Ok(()) => {}
Err(source) if source.kind() == std::io::ErrorKind::WouldBlock => continue,
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "lock published run for retention",
path: lease_path,
source,
});
}
}
let name = path
.file_name()
.and_then(|name| name.to_str())
.unwrap_or("run");
let retired = quarantine.join(format!("retired-{name}--{}", workspace_nonce()));
fs::rename(path, &retired).map_err(|source| ArtifactPathError::Io {
operation: "retire published run",
path: path.clone(),
source,
})?;
drop(lease);
fs::remove_dir_all(&retired).map_err(|source| ArtifactPathError::Io {
operation: "delete retired published run",
path: retired,
source,
})?;
}
sync_directory(root.path())?;
sync_directory(&quarantine)?;
prune_quarantine_entries(&quarantine, RETAIN_QUARANTINE_ENTRIES)
}
fn prune_quarantine_entries(directory: &Path, retain: usize) -> Result<(), ArtifactPathError> {
let mut entries = fs::read_dir(directory)
.map_err(|source| ArtifactPathError::Io {
operation: "list artifact quarantine for retention",
path: directory.to_path_buf(),
source,
})?
.collect::<Result<Vec<_>, _>>()
.map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact quarantine for retention",
path: directory.to_path_buf(),
source,
})?;
entries.sort_by_key(|entry| std::cmp::Reverse(entry.file_name()));
for entry in entries.into_iter().skip(retain) {
let path = entry.path();
let metadata = fs::symlink_metadata(&path).map_err(|source| ArtifactPathError::Io {
operation: "inspect quarantined artifact for deletion",
path: path.clone(),
source,
})?;
let result = if metadata.file_type().is_symlink() || metadata.is_file() {
fs::remove_file(&path)
} else if metadata.is_dir() {
fs::remove_dir_all(&path)
} else {
fs::remove_file(&path)
};
result.map_err(|source| ArtifactPathError::Io {
operation: "delete expired quarantined artifact",
path,
source,
})?;
}
sync_directory(directory)
}
fn validate_predecessor_chain(
latest_root: &ApprovedRoot,
tip: &LatestManifest,
) -> Result<(), ArtifactPathError> {
let mut visited = BTreeSet::new();
let mut child = tip.clone();
loop {
if !visited.insert(child.generation.clone()) {
return Err(ArtifactPathError::GenerationChainCycle {
generation: child.generation,
});
}
let Some(predecessor) = child.predecessor_generation.as_ref() else {
return Ok(());
};
let child_path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(&child.generation);
if load_retention_boundary(&child_path, &child)?.is_some() {
return Ok(());
}
let predecessor_path = latest_root
.path()
.join(LATEST_GENERATIONS_DIRECTORY)
.join(predecessor);
let predecessor_manifest = match fs::symlink_metadata(&predecessor_path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: predecessor_path,
});
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: predecessor_path,
});
}
Ok(_) => validate_latest_manifest(&predecessor_path, predecessor)?,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Err(ArtifactPathError::BrokenGenerationChain {
generation: child.generation,
predecessor: predecessor.clone(),
});
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect predecessor generation",
path: predecessor_path,
source,
});
}
};
reject_chain_downgrade(&predecessor_manifest, &child)?;
child = predecessor_manifest;
}
}
fn load_retention_boundary(
generation_path: &Path,
manifest: &LatestManifest,
) -> Result<Option<RetentionBoundary>, ArtifactPathError> {
let path = generation_path.join(LATEST_RETENTION_BOUNDARY_FILE);
let metadata = match fs::symlink_metadata(&path) {
Ok(metadata) => metadata,
Err(source) if source.kind() == std::io::ErrorKind::NotFound => return Ok(None),
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect latest-generation retention boundary",
path,
source,
});
}
};
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent { path });
}
if !metadata.is_file() {
return Err(ArtifactPathError::FileDestinationNotFile { path });
}
let body = fs::read(&path).map_err(|source| ArtifactPathError::Io {
operation: "read latest-generation retention boundary",
path: path.clone(),
source,
})?;
let boundary: RetentionBoundary =
serde_json::from_slice(&body).map_err(|source| ArtifactPathError::ManifestJson {
path: path.clone(),
source,
})?;
if boundary.format_version != 1
|| manifest.predecessor_generation.as_deref()
!= Some(boundary.predecessor_generation.as_str())
{
return Err(ArtifactPathError::CorruptGeneration {
path,
detail: "retention boundary does not match the manifest predecessor".to_owned(),
});
}
Ok(Some(boundary))
}
fn reject_chain_downgrade(
predecessor: &LatestManifest,
child: &LatestManifest,
) -> Result<(), ArtifactPathError> {
if parse_producer_version(&predecessor.producer_version)?
> parse_producer_version(&child.producer_version)?
{
return Err(ArtifactPathError::ProducerDowngrade {
existing: predecessor.producer_version.clone(),
attempted: child.producer_version.clone(),
});
}
Ok(())
}
fn reject_producer_downgrade(existing: &str) -> Result<(), ArtifactPathError> {
let existing_version = parse_producer_version(existing)?;
let producer_version = parse_producer_version(env!("CARGO_PKG_VERSION"))?;
if existing_version > producer_version {
return Err(ArtifactPathError::ProducerDowngrade {
existing: existing.to_owned(),
attempted: env!("CARGO_PKG_VERSION").to_owned(),
});
}
Ok(())
}
fn install_stable_aliases_transactional(
root: &ApprovedRoot,
latest_root: &ApprovedRoot,
snapshot: &LatestSnapshot,
) -> Result<StableAliasTransaction, ArtifactPathError> {
let current_aliases: BTreeSet<ArtifactId> = match load_current_snapshot(root, true) {
Ok(Some(current)) => current
.manifest
.artifacts
.iter()
.map(|artifact| ArtifactId::new(artifact.destination_relative_path.clone()))
.collect::<Result<_, _>>()?,
Ok(None) => BTreeSet::new(),
Err(error) if is_recoverable_pointer_error(&error) => BTreeSet::new(),
Err(error) => return Err(error),
};
let next_aliases: BTreeSet<ArtifactId> = snapshot
.manifest
.artifacts
.iter()
.map(|artifact| ArtifactId::new(artifact.destination_relative_path.clone()))
.collect::<Result<_, _>>()?;
let staging_root = latest_root.prepare_dir(LATEST_STAGING_DIRECTORY)?;
let transaction_path = staging_root.join(format!("aliases-{}", workspace_nonce()));
fs::create_dir(&transaction_path).map_err(|source| ArtifactPathError::Io {
operation: "create stable-alias transaction",
path: transaction_path.clone(),
source,
})?;
let transaction_root = ApprovedRoot::existing(&transaction_path)?;
let new_root = transaction_root.prepare_dir("new")?;
let backup_root = transaction_root.prepare_dir("backup")?;
let mut prepared = Vec::with_capacity(snapshot.manifest.artifacts.len());
for artifact in &snapshot.manifest.artifacts {
let id = ArtifactId::new(artifact.destination_relative_path.clone())?;
let source = snapshot.generation_path.join(id.as_str());
let destination = root.prepare_file(id.as_str())?;
let staged = new_root.join(id.as_str());
if let Err(error) = copy_new_synced(&source, &staged, "stage stable latest alias") {
let _ = fs::remove_dir_all(&transaction_path);
return Err(error);
}
prepared.push((staged, destination, id));
}
sync_artifact_tree(&transaction_path)?;
let mut transaction = StableAliasTransaction {
path: transaction_path,
root_path: root.path().to_path_buf(),
installed: Vec::new(),
backups: Vec::new(),
};
for id in current_aliases.difference(&next_aliases) {
let destination = root.prepare_file(id.as_str())?;
match fs::symlink_metadata(&destination) {
Ok(metadata) if metadata.file_type().is_symlink() => {
let original = ArtifactPathError::SymlinkComponent { path: destination };
return Err(transaction.rollback(original));
}
Ok(metadata) if !metadata.is_file() => {
let original = ArtifactPathError::FileDestinationNotFile { path: destination };
return Err(transaction.rollback(original));
}
Ok(_) => {
let backup = backup_root.join(id.as_str());
if let Err(source) = fs::rename(&destination, &backup) {
let original = ArtifactPathError::Io {
operation: "retire obsolete stable latest alias",
path: destination,
source,
};
return Err(transaction.rollback(original));
}
transaction.backups.push((destination, backup));
if let Err(error) =
sync_directory(root.path()).and_then(|()| sync_directory(&backup_root))
{
return Err(transaction.rollback(error));
}
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
let original = ArtifactPathError::Io {
operation: "inspect obsolete stable latest alias",
path: destination,
source,
};
return Err(transaction.rollback(original));
}
}
}
for (staged, destination, id) in prepared {
match fs::symlink_metadata(&destination) {
Ok(metadata) if metadata.file_type().is_symlink() => {
let original = ArtifactPathError::SymlinkComponent { path: destination };
return Err(transaction.rollback(original));
}
Ok(metadata) if !metadata.is_file() => {
let original = ArtifactPathError::FileDestinationNotFile { path: destination };
return Err(transaction.rollback(original));
}
Ok(_) => {
let backup = backup_root.join(id.as_str());
if let Err(source) = fs::rename(&destination, &backup) {
let original = ArtifactPathError::Io {
operation: "back up stable latest alias",
path: destination,
source,
};
return Err(transaction.rollback(original));
}
transaction.backups.push((destination.clone(), backup));
if let Err(error) =
sync_directory(root.path()).and_then(|()| sync_directory(&backup_root))
{
return Err(transaction.rollback(error));
}
}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
let original = ArtifactPathError::Io {
operation: "inspect stable latest alias",
path: destination,
source,
};
return Err(transaction.rollback(original));
}
}
if let Err(original) = maybe_fail_alias_install(transaction.installed.len(), &destination) {
return Err(transaction.rollback(original));
}
if let Err(source) = fs::rename(&staged, &destination) {
let original = ArtifactPathError::Io {
operation: "install stable latest alias",
path: destination,
source,
};
return Err(transaction.rollback(original));
}
transaction.installed.push(destination);
if let Err(error) = sync_directory(root.path()) {
return Err(transaction.rollback(error));
}
}
record_durability_event("install_aliases");
Ok(transaction)
}
fn remove_installed_latest(installed: &[PathBuf]) -> Result<(), ArtifactPathError> {
let mut first_error = None;
for destination in installed.iter().rev() {
match fs::remove_file(destination) {
Ok(()) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {}
Err(source) => {
first_error.get_or_insert_with(|| ArtifactPathError::Io {
operation: "roll back installed latest artifact",
path: destination.clone(),
source,
});
}
}
}
first_error.map_or(Ok(()), Err)
}
fn restore_latest_backups(backups: &[(PathBuf, PathBuf)]) -> Result<(), ArtifactPathError> {
let mut first_error = None;
for (destination, backup) in backups.iter().rev() {
if let Err(source) = fs::rename(backup, destination) {
first_error.get_or_insert_with(|| ArtifactPathError::Io {
operation: "restore previous latest artifact",
path: destination.clone(),
source,
});
}
}
first_error.map_or(Ok(()), Err)
}
fn latest_update_failure(
original: ArtifactPathError,
installed: &[PathBuf],
backups: &[(PathBuf, PathBuf)],
recovery_path: PathBuf,
) -> ArtifactPathError {
let remove_error = remove_installed_latest(installed).err();
let restore_error = restore_latest_backups(backups).err();
let rollback = remove_error.or(restore_error);
match rollback {
Some(rollback) => ArtifactPathError::LatestRollback {
original: Box::new(original),
rollback: Box::new(rollback),
recovery_path,
},
None => original,
}
}
fn validate_run_manifest(run_path: &Path) -> Result<RunManifest, ArtifactPathError> {
let manifest_path = run_path.join(RUN_MANIFEST_FILE);
let manifest: RunManifest = read_json_file(&manifest_path, "read run manifest")?;
if manifest.format_version != RUN_MANIFEST_VERSION {
return Err(ArtifactPathError::UnknownManifestVersion {
path: manifest_path,
found: manifest.format_version,
supported: RUN_MANIFEST_VERSION,
});
}
parse_producer_version(&manifest.producer_version)?;
ArtifactId::new(manifest.logical_id.clone())?;
ArtifactId::new(manifest.publication_id.clone())?;
if let Some(expected) = manifest.expected_latest_generation.as_ref() {
ArtifactId::new(expected.clone())?;
}
let expected_publication = run_path
.file_name()
.and_then(|name| name.to_str())
.ok_or_else(|| ArtifactPathError::NonUtf8ArtifactPath {
path: run_path.to_path_buf(),
})?;
if manifest.publication_id != expected_publication {
return Err(ArtifactPathError::ManifestIntegrity {
path: manifest_path,
detail: "publication identity does not match its directory".to_owned(),
});
}
let actual = inspect_artifact_tree(run_path, &[RUN_MANIFEST_FILE, RUN_READER_LOCK_FILE])?;
if manifest.artifacts != actual {
return Err(ArtifactPathError::ManifestIntegrity {
path: manifest_path,
detail: "run artifact paths, sizes, or digests do not match".to_owned(),
});
}
Ok(manifest)
}
fn validate_latest_manifest(
generation_path: &Path,
expected_generation: &str,
) -> Result<LatestManifest, ArtifactPathError> {
let manifest_path = generation_path.join(LATEST_MANIFEST_FILE);
let manifest: LatestManifest = read_json_file(&manifest_path, "read latest manifest")?;
if manifest.format_version != LATEST_MANIFEST_VERSION {
return Err(ArtifactPathError::UnknownManifestVersion {
path: manifest_path,
found: manifest.format_version,
supported: LATEST_MANIFEST_VERSION,
});
}
parse_producer_version(&manifest.producer_version)?;
ArtifactId::new(manifest.generation.clone())?;
ArtifactId::new(manifest.source_logical_id.clone())?;
ArtifactId::new(manifest.source_publication_id.clone())?;
if let Some(predecessor) = manifest.predecessor_generation.as_ref() {
ArtifactId::new(predecessor.clone())?;
}
if manifest.generation != expected_generation {
return Err(ArtifactPathError::CorruptGeneration {
path: generation_path.to_path_buf(),
detail: "manifest generation does not match current pointer".to_owned(),
});
}
let actual = inspect_artifact_tree(
generation_path,
&[
LATEST_MANIFEST_FILE,
LATEST_READER_LOCK_FILE,
LATEST_RETENTION_BOUNDARY_FILE,
],
)?;
let mut expected = Vec::with_capacity(manifest.artifacts.len());
for artifact in &manifest.artifacts {
ArtifactId::new(artifact.source_relative_path.clone())?;
ArtifactId::new(artifact.destination_relative_path.clone())?;
expected.push(ManifestArtifact {
relative_path: artifact.destination_relative_path.clone(),
size: artifact.size,
sha256: artifact.sha256.clone(),
});
}
expected.sort_by(|left, right| left.relative_path.cmp(&right.relative_path));
if expected != actual {
return Err(ArtifactPathError::CorruptGeneration {
path: generation_path.to_path_buf(),
detail: "generation artifact paths, sizes, or digests do not match".to_owned(),
});
}
Ok(manifest)
}
fn parse_producer_version(version: &str) -> Result<Version, ArtifactPathError> {
Version::parse(version).map_err(|source| ArtifactPathError::InvalidProducerVersion {
version: version.to_owned(),
source,
})
}
fn read_json_file<T: for<'de> Deserialize<'de>>(
path: &Path,
operation: &'static str,
) -> Result<T, ArtifactPathError> {
let metadata = fs::symlink_metadata(path).map_err(|source| ArtifactPathError::Io {
operation,
path: path.to_path_buf(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent {
path: path.to_path_buf(),
});
}
if !metadata.is_file() {
return Err(ArtifactPathError::FileDestinationNotFile {
path: path.to_path_buf(),
});
}
let file = File::open(path).map_err(|source| ArtifactPathError::Io {
operation,
path: path.to_path_buf(),
source,
})?;
serde_json::from_reader(BufReader::new(file)).map_err(|source| {
ArtifactPathError::ManifestJson {
path: path.to_path_buf(),
source,
}
})
}
fn write_json_file<T: Serialize>(
path: &Path,
value: &T,
operation: &'static str,
) -> Result<(), ArtifactPathError> {
let mut encoded =
serde_json::to_vec_pretty(value).map_err(|source| ArtifactPathError::ManifestJson {
path: path.to_path_buf(),
source,
})?;
encoded.push(b'\n');
let mut file = OpenOptions::new()
.write(true)
.create_new(true)
.open(path)
.map_err(|source| ArtifactPathError::Io {
operation,
path: path.to_path_buf(),
source,
})?;
file.write_all(&encoded)
.map_err(|source| ArtifactPathError::Io {
operation,
path: path.to_path_buf(),
source,
})?;
file.sync_all().map_err(|source| ArtifactPathError::Io {
operation: "sync manifest",
path: path.to_path_buf(),
source,
})?;
record_durability_event("sync_manifest");
Ok(())
}
fn inspect_artifact_tree(
root: &Path,
excluded_relative_paths: &[&str],
) -> Result<Vec<ManifestArtifact>, ArtifactPathError> {
let mut artifacts = Vec::new();
inspect_artifact_tree_recursive(root, root, excluded_relative_paths, &mut artifacts)?;
artifacts.sort_by(|left, right| left.relative_path.cmp(&right.relative_path));
Ok(artifacts)
}
fn inspect_artifact_tree_recursive(
root: &Path,
directory: &Path,
excluded_relative_paths: &[&str],
artifacts: &mut Vec<ManifestArtifact>,
) -> Result<(), ArtifactPathError> {
let mut entries = fs::read_dir(directory)
.map_err(|source| ArtifactPathError::Io {
operation: "list artifact directory",
path: directory.to_path_buf(),
source,
})?
.collect::<Result<Vec<_>, _>>()
.map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact directory entry",
path: directory.to_path_buf(),
source,
})?;
entries.sort_by_key(|entry| entry.file_name());
for entry in entries {
let path = entry.path();
let metadata = fs::symlink_metadata(&path).map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact tree entry",
path: path.clone(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent { path });
}
if metadata.is_dir() {
inspect_artifact_tree_recursive(root, &path, excluded_relative_paths, artifacts)?;
continue;
}
if !metadata.is_file() {
return Err(ArtifactPathError::FileDestinationNotFile { path });
}
reject_hardlinked_regular_file(&metadata, &path)?;
let relative_path = manifest_relative_path(root, &path)?;
if excluded_relative_paths.contains(&relative_path.as_str()) {
continue;
}
let (size, sha256) = digest_file(&path)?;
artifacts.push(ManifestArtifact {
relative_path,
size,
sha256,
});
}
Ok(())
}
fn manifest_relative_path(root: &Path, path: &Path) -> Result<String, ArtifactPathError> {
let relative = path
.strip_prefix(root)
.map_err(|_| ArtifactPathError::InvalidRelativePath {
path: path.to_path_buf(),
})?;
let mut components = Vec::new();
for component in relative.components() {
let std::path::Component::Normal(component) = component else {
return Err(ArtifactPathError::InvalidRelativePath {
path: relative.to_path_buf(),
});
};
components.push(component.to_str().ok_or_else(|| {
ArtifactPathError::NonUtf8ArtifactPath {
path: path.to_path_buf(),
}
})?);
}
Ok(components.join("/"))
}
fn digest_file(path: &Path) -> Result<(u64, String), ArtifactPathError> {
let metadata = fs::symlink_metadata(path).map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact for digest",
path: path.to_path_buf(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent {
path: path.to_path_buf(),
});
}
if !metadata.is_file() {
return Err(ArtifactPathError::FileDestinationNotFile {
path: path.to_path_buf(),
});
}
reject_hardlinked_regular_file(&metadata, path)?;
let mut file = File::open(path).map_err(|source| ArtifactPathError::Io {
operation: "open artifact for digest",
path: path.to_path_buf(),
source,
})?;
let mut digest = Sha256::new();
let mut buffer = [0_u8; 64 * 1024];
loop {
let read = file
.read(&mut buffer)
.map_err(|source| ArtifactPathError::Io {
operation: "read artifact for digest",
path: path.to_path_buf(),
source,
})?;
if read == 0 {
break;
}
digest.update(&buffer[..read]);
}
Ok((metadata.len(), format!("{:x}", digest.finalize())))
}
fn sha256_bytes(bytes: &[u8]) -> String {
let mut digest = Sha256::new();
digest.update(bytes);
format!("{:x}", digest.finalize())
}
#[cfg(unix)]
fn open_directory_no_follow(path: &Path) -> std::io::Result<File> {
use std::os::unix::fs::OpenOptionsExt;
OpenOptions::new()
.read(true)
.custom_flags(libc::O_DIRECTORY | libc::O_NOFOLLOW | libc::O_CLOEXEC)
.open(path)
}
#[cfg(unix)]
fn open_snapshot_artifact(snapshot: &LatestSnapshot, id: &ArtifactId) -> std::io::Result<File> {
use std::ffi::CString;
use std::os::fd::{AsRawFd, FromRawFd};
let name = CString::new(id.as_str())
.map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "name contains NUL"))?;
let fd = unsafe {
libc::openat(
snapshot.generation_directory.as_raw_fd(),
name.as_ptr(),
libc::O_RDONLY | libc::O_CLOEXEC | libc::O_NOFOLLOW,
)
};
if fd < 0 {
return Err(std::io::Error::last_os_error());
}
let file = unsafe { File::from_raw_fd(fd) };
if !file.metadata()?.is_file() {
return Err(std::io::Error::new(
std::io::ErrorKind::InvalidData,
"snapshot artifact is not a regular file",
));
}
Ok(file)
}
#[cfg(not(unix))]
fn open_snapshot_artifact(snapshot: &LatestSnapshot, id: &ArtifactId) -> std::io::Result<File> {
File::open(snapshot.generation_path.join(id.as_str()))
}
#[cfg(any(target_os = "macos", target_os = "ios"))]
fn rename_directory_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
use std::ffi::CString;
use std::os::unix::ffi::OsStrExt;
let source = CString::new(source.as_os_str().as_bytes()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidInput, "source contains NUL")
})?;
let destination = CString::new(destination.as_os_str().as_bytes()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidInput, "destination contains NUL")
})?;
let result =
unsafe { libc::renamex_np(source.as_ptr(), destination.as_ptr(), libc::RENAME_EXCL) };
if result == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(any(target_os = "linux", target_os = "android"))]
fn rename_directory_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
use std::ffi::CString;
use std::os::unix::ffi::OsStrExt;
let source = CString::new(source.as_os_str().as_bytes()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidInput, "source contains NUL")
})?;
let destination = CString::new(destination.as_os_str().as_bytes()).map_err(|_| {
std::io::Error::new(std::io::ErrorKind::InvalidInput, "destination contains NUL")
})?;
let result = unsafe {
libc::syscall(
libc::SYS_renameat2,
libc::AT_FDCWD,
source.as_ptr(),
libc::AT_FDCWD,
destination.as_ptr(),
libc::RENAME_NOREPLACE,
)
};
if result == 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(not(any(
target_os = "macos",
target_os = "ios",
target_os = "linux",
target_os = "android",
windows
)))]
fn rename_directory_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
fs::rename(source, destination)
}
#[cfg(windows)]
fn rename_directory_noreplace(source: &Path, destination: &Path) -> std::io::Result<()> {
use std::os::windows::ffi::OsStrExt;
use windows_sys::Win32::Storage::FileSystem::{MOVEFILE_WRITE_THROUGH, MoveFileExW};
let source: Vec<u16> = source
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect();
let destination: Vec<u16> = destination
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect();
let result = unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_WRITE_THROUGH,
)
};
if result != 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(windows)]
fn replace_file_atomically(source: &Path, destination: &Path) -> std::io::Result<()> {
use std::os::windows::ffi::OsStrExt;
use windows_sys::Win32::Storage::FileSystem::{
MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH, MoveFileExW,
};
let source: Vec<u16> = source
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect();
let destination: Vec<u16> = destination
.as_os_str()
.encode_wide()
.chain(std::iter::once(0))
.collect();
let result = unsafe {
MoveFileExW(
source.as_ptr(),
destination.as_ptr(),
MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
)
};
if result != 0 {
Ok(())
} else {
Err(std::io::Error::last_os_error())
}
}
#[cfg(not(windows))]
fn replace_file_atomically(source: &Path, destination: &Path) -> std::io::Result<()> {
fs::rename(source, destination)
}
fn copy_new_synced(
source: &Path,
destination: &Path,
operation: &'static str,
) -> Result<(), ArtifactPathError> {
let source_metadata =
fs::symlink_metadata(source).map_err(|source_error| ArtifactPathError::Io {
operation,
path: source.to_path_buf(),
source: source_error,
})?;
if source_metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent {
path: source.to_path_buf(),
});
}
if !source_metadata.is_file() {
return Err(ArtifactPathError::FileDestinationNotFile {
path: source.to_path_buf(),
});
}
reject_hardlinked_regular_file(&source_metadata, source)?;
let mut source_file = File::open(source).map_err(|source_error| ArtifactPathError::Io {
operation,
path: source.to_path_buf(),
source: source_error,
})?;
let mut destination_file = OpenOptions::new()
.write(true)
.create_new(true)
.open(destination)
.map_err(|source_error| ArtifactPathError::Io {
operation,
path: destination.to_path_buf(),
source: source_error,
})?;
std::io::copy(&mut source_file, &mut destination_file).map_err(|source_error| {
ArtifactPathError::Io {
operation,
path: destination.to_path_buf(),
source: source_error,
}
})?;
destination_file
.sync_all()
.map_err(|source_error| ArtifactPathError::Io {
operation: "sync copied artifact",
path: destination.to_path_buf(),
source: source_error,
})?;
record_durability_event("sync_file");
Ok(())
}
fn sync_artifact_tree(root: &Path) -> Result<(), ArtifactPathError> {
sync_artifact_tree_recursive(root)
}
#[cfg(unix)]
fn reject_hardlinked_regular_file(
metadata: &fs::Metadata,
path: &Path,
) -> Result<(), ArtifactPathError> {
use std::os::unix::fs::MetadataExt;
if metadata.nlink() > 1 {
return Err(ArtifactPathError::HardLinkedArtifact {
path: path.to_path_buf(),
links: metadata.nlink(),
});
}
Ok(())
}
#[cfg(not(unix))]
fn reject_hardlinked_regular_file(
_metadata: &fs::Metadata,
_path: &Path,
) -> Result<(), ArtifactPathError> {
Ok(())
}
fn sync_artifact_tree_recursive(directory: &Path) -> Result<(), ArtifactPathError> {
let entries = fs::read_dir(directory)
.map_err(|source| ArtifactPathError::Io {
operation: "list artifact tree for sync",
path: directory.to_path_buf(),
source,
})?
.collect::<Result<Vec<_>, _>>()
.map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact tree entry for sync",
path: directory.to_path_buf(),
source,
})?;
for entry in entries {
let path = entry.path();
let metadata = fs::symlink_metadata(&path).map_err(|source| ArtifactPathError::Io {
operation: "inspect artifact tree entry for sync",
path: path.clone(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent { path });
}
if metadata.is_dir() {
sync_artifact_tree_recursive(&path)?;
} else if metadata.is_file() {
File::open(&path)
.and_then(|file| file.sync_all())
.map_err(|source| ArtifactPathError::Io {
operation: "sync artifact file",
path: path.clone(),
source,
})?;
record_durability_event("sync_file");
} else {
return Err(ArtifactPathError::FileDestinationNotFile { path });
}
}
sync_directory(directory)
}
#[cfg(unix)]
fn sync_directory(path: &Path) -> Result<(), ArtifactPathError> {
maybe_fail_directory_sync(path)?;
File::open(path)
.and_then(|directory| directory.sync_all())
.map_err(|source| ArtifactPathError::Io {
operation: "sync artifact directory",
path: path.to_path_buf(),
source,
})?;
record_durability_event("sync_directory");
Ok(())
}
#[cfg(not(unix))]
fn sync_directory(_path: &Path) -> Result<(), ArtifactPathError> {
maybe_fail_directory_sync(_path)?;
record_durability_event("directory_sync_unavailable");
Ok(())
}
#[cfg(test)]
thread_local! {
static DURABILITY_EVENTS: std::cell::RefCell<Vec<&'static str>> = const {
std::cell::RefCell::new(Vec::new())
};
static FAIL_ALIAS_INSTALL_AFTER: std::cell::Cell<Option<usize>> = const {
std::cell::Cell::new(None)
};
static MUTATE_LATEST_SOURCE_AFTER_VALIDATION: std::cell::RefCell<Option<(PathBuf, Vec<u8>)>> =
const { std::cell::RefCell::new(None) };
static FAIL_DIRECTORY_SYNC_AFTER_EVENT: std::cell::Cell<Option<&'static str>> =
const { std::cell::Cell::new(None) };
static FAIL_NEXT_DIRECTORY_SYNC: std::cell::Cell<bool> = const { std::cell::Cell::new(false) };
}
#[cfg(test)]
fn record_durability_event(event: &'static str) {
DURABILITY_EVENTS.with(|events| events.borrow_mut().push(event));
FAIL_DIRECTORY_SYNC_AFTER_EVENT.with(|target| {
if target.get() == Some(event) {
target.set(None);
FAIL_NEXT_DIRECTORY_SYNC.with(|fail| fail.set(true));
}
});
}
#[cfg(not(test))]
fn record_durability_event(_event: &'static str) {}
#[cfg(test)]
fn maybe_fail_directory_sync(path: &Path) -> Result<(), ArtifactPathError> {
let fail = FAIL_NEXT_DIRECTORY_SYNC.with(|fail| fail.replace(false));
if fail {
return Err(ArtifactPathError::Io {
operation: "sync artifact directory",
path: path.to_path_buf(),
source: std::io::Error::other("injected post-commit directory sync failure"),
});
}
Ok(())
}
#[cfg(not(test))]
fn maybe_fail_directory_sync(_path: &Path) -> Result<(), ArtifactPathError> {
Ok(())
}
#[cfg(test)]
fn maybe_fail_alias_install(installed: usize, path: &Path) -> Result<(), ArtifactPathError> {
let should_fail = FAIL_ALIAS_INSTALL_AFTER.with(|fail_after| {
let should_fail = fail_after.get() == Some(installed);
if should_fail {
fail_after.set(None);
}
should_fail
});
if should_fail {
return Err(ArtifactPathError::Io {
operation: "install stable latest alias",
path: path.to_path_buf(),
source: std::io::Error::other("injected stable-alias failure"),
});
}
Ok(())
}
#[cfg(test)]
fn maybe_mutate_latest_source_after_validation() {
MUTATE_LATEST_SOURCE_AFTER_VALIDATION.with(|mutation| {
if let Some((path, contents)) = mutation.borrow_mut().take() {
fs::write(path, contents).expect("apply injected post-validation source mutation");
}
});
}
#[cfg(not(test))]
fn maybe_mutate_latest_source_after_validation() {}
#[cfg(not(test))]
fn maybe_fail_alias_install(_installed: usize, _path: &Path) -> Result<(), ArtifactPathError> {
Ok(())
}
fn active_workspaces() -> &'static Mutex<BTreeSet<PathBuf>> {
ACTIVE_WORKSPACES.get_or_init(|| Mutex::new(BTreeSet::new()))
}
fn register_active_workspace(path: &Path) {
active_workspaces()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.insert(path.to_path_buf());
}
fn unregister_active_workspace(path: &Path) {
active_workspaces()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.remove(path);
}
fn is_active_workspace(path: &Path) -> bool {
active_workspaces()
.lock()
.unwrap_or_else(std::sync::PoisonError::into_inner)
.contains(path)
}
fn workspace_nonce() -> String {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_nanos();
let sequence = WORKSPACE_SEQUENCE.fetch_add(1, Ordering::Relaxed);
format!("{timestamp:x}-{:x}-{sequence:x}", std::process::id())
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovedRoot {
path: PathBuf,
}
impl ApprovedRoot {
pub fn existing(path: impl AsRef<Path>) -> Result<Self, ArtifactPathError> {
let path = path.as_ref();
let metadata = fs::symlink_metadata(path).map_err(|source| ArtifactPathError::Io {
operation: "inspect approved root",
path: path.to_path_buf(),
source,
})?;
if metadata.file_type().is_symlink() {
return Err(ArtifactPathError::SymlinkComponent {
path: path.to_path_buf(),
});
}
if !metadata.is_dir() {
return Err(ArtifactPathError::RootNotDirectory(path.to_path_buf()));
}
let path = fs::canonicalize(path).map_err(|source| ArtifactPathError::Io {
operation: "canonicalize approved root",
path: path.to_path_buf(),
source,
})?;
Ok(Self { path })
}
pub fn path(&self) -> &Path {
&self.path
}
pub fn project_dir(&self, relative: impl AsRef<Path>) -> Result<PathBuf, ArtifactPathError> {
let relative = relative.as_ref();
validate_relative_path(relative)?;
let mut current = self.path.clone();
for component in relative.components() {
let std::path::Component::Normal(name) = component else {
unreachable!("relative path was validated before directory projection")
};
current.push(name);
match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: current.clone(),
});
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: current.clone(),
});
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
return Ok(self.path.join(relative));
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect projected artifact directory",
path: current,
source,
});
}
}
}
Ok(current)
}
pub fn prepare_dir(&self, relative: impl AsRef<Path>) -> Result<PathBuf, ArtifactPathError> {
let relative = relative.as_ref();
validate_relative_path(relative)?;
prepare_directory_components(&self.path, relative)
}
pub fn prepare_file(&self, relative: impl AsRef<Path>) -> Result<PathBuf, ArtifactPathError> {
let relative = relative.as_ref();
validate_relative_path(relative)?;
if let Some(parent) = relative
.parent()
.filter(|parent| !parent.as_os_str().is_empty())
{
prepare_directory_components(&self.path, parent)?;
}
let path = self.path.join(relative);
match fs::symlink_metadata(&path) {
Ok(metadata) if metadata.file_type().is_symlink() => {
Err(ArtifactPathError::SymlinkComponent { path })
}
Ok(metadata) if !metadata.is_file() => {
Err(ArtifactPathError::FileDestinationNotFile { path })
}
Ok(_) => Ok(path),
Err(source) if source.kind() == std::io::ErrorKind::NotFound => Ok(path),
Err(source) => Err(ArtifactPathError::Io {
operation: "inspect artifact file",
path,
source,
}),
}
}
}
#[derive(Debug, Error)]
pub enum ArtifactPathError {
#[error("artifact identifier must be one non-empty relative path component: {value:?}")]
InvalidArtifactId { value: String },
#[error("could not allocate a unique workspace for logical run ID {logical_id}")]
WorkspaceAllocationExhausted { logical_id: ArtifactId },
#[error("artifact publication destination already exists: {path}")]
PublicationDestinationExists { path: PathBuf },
#[error("staged run already contains the reserved manifest path: {path}")]
ReservedManifestPath { path: PathBuf },
#[error("required staged artifact is missing: {path}")]
RequiredArtifactMissing { path: PathBuf },
#[error("latest artifact destination was requested more than once: {id}")]
DuplicateLatestDestination { id: ArtifactId },
#[error("could not allocate a unique latest-artifact transaction")]
LatestTransactionAllocationExhausted,
#[error("timed out waiting for the latest-artifact update lock at {path}")]
LatestLockTimeout { path: PathBuf },
#[error("no committed latest snapshot is available at {path}")]
LatestSnapshotUnavailable { path: PathBuf },
#[error("latest pointer is not one valid generation identity: {path}")]
InvalidLatestPointer { path: PathBuf },
#[error("latest snapshot does not contain alias {id}")]
LatestArtifactNotInSnapshot { id: ArtifactId },
#[error("latest generation is corrupt at {path}: {detail}")]
CorruptGeneration { path: PathBuf, detail: String },
#[error("latest-generation chain is broken at {generation}: missing {predecessor}")]
BrokenGenerationChain {
generation: String,
predecessor: String,
},
#[error("latest-generation chain contains a cycle at {generation}")]
GenerationChainCycle { generation: String },
#[error("latest-generation chain forks after {predecessor} into {children:?}")]
AmbiguousGenerationFork {
predecessor: String,
children: Vec<String>,
},
#[error("latest-generation recovery has ambiguous tips: {tips:?}")]
AmbiguousGenerationTips { tips: Vec<String> },
#[error(
"published run has stale latest-generation token: expected {expected:?}, observed {observed:?}"
)]
StaleLatestGeneration {
expected: Option<String>,
observed: Option<String>,
},
#[error("refusing latest-generation producer downgrade from {existing} to {attempted}")]
ProducerDowngrade { existing: String, attempted: String },
#[error("invalid artifact producer version {version:?}: {source}")]
InvalidProducerVersion {
version: String,
#[source]
source: semver::Error,
},
#[error("unsupported manifest version {found} at {path}; supported version is {supported}")]
UnknownManifestVersion {
path: PathBuf,
found: u32,
supported: u32,
},
#[error("artifact manifest integrity check failed at {path}: {detail}")]
ManifestIntegrity { path: PathBuf, detail: String },
#[error("could not encode or decode artifact manifest at {path}: {source}")]
ManifestJson {
path: PathBuf,
#[source]
source: serde_json::Error,
},
#[error(
"latest-artifact update failed ({original}) and rollback was incomplete ({rollback}); recovery material retained at {recovery_path}"
)]
LatestRollback {
original: Box<ArtifactPathError>,
rollback: Box<ArtifactPathError>,
recovery_path: PathBuf,
},
#[error(
"latest-generation preparation failed ({original}) and staging cleanup failed ({cleanup}) at {staging_path}"
)]
LatestStagingCleanup {
original: Box<ArtifactPathError>,
cleanup: Box<ArtifactPathError>,
staging_path: PathBuf,
},
#[error("run is committed at {path}, but directory durability is uncertain: {source}")]
PublicationDurabilityUncertain {
path: PathBuf,
#[source]
source: Box<ArtifactPathError>,
},
#[error("run is committed at {path}, but post-commit retention maintenance failed: {source}")]
PublicationPostCommitMaintenance {
path: PathBuf,
#[source]
source: Box<ArtifactPathError>,
},
#[error(
"latest generation {generation} is committed, but pointer durability is uncertain: {source}"
)]
LatestCommitDurabilityUncertain {
generation: String,
#[source]
source: Box<ArtifactPathError>,
},
#[error("latest generation {generation} is committed, but retention failed: {source}")]
LatestRetentionAfterCommit {
generation: String,
#[source]
source: Box<ArtifactPathError>,
},
#[error("approved artifact root is not a directory: {0}")]
RootNotDirectory(PathBuf),
#[error(
"artifact child path must be a non-empty relative path without parent traversal: {path}"
)]
InvalidRelativePath { path: PathBuf },
#[error("artifact path component is a symbolic link: {path}")]
SymlinkComponent { path: PathBuf },
#[error("artifact directory component is not a directory: {path}")]
DirectoryComponentNotDirectory { path: PathBuf },
#[error("artifact file destination is not a regular file: {path}")]
FileDestinationNotFile { path: PathBuf },
#[error("artifact file has {links} hard links and is not publication-private: {path}")]
HardLinkedArtifact { path: PathBuf, links: u64 },
#[error("artifact path is not valid UTF-8 and cannot be recorded in a manifest: {path:?}")]
NonUtf8ArtifactPath { path: PathBuf },
#[error("failed to {operation} at {path}: {source}")]
Io {
operation: &'static str,
path: PathBuf,
#[source]
source: std::io::Error,
},
}
fn validate_relative_path(path: &Path) -> Result<(), ArtifactPathError> {
if path.as_os_str().is_empty()
|| path
.components()
.any(|component| !matches!(component, std::path::Component::Normal(_)))
{
return Err(ArtifactPathError::InvalidRelativePath {
path: path.to_path_buf(),
});
}
Ok(())
}
fn prepare_directory_components(
root: &Path,
relative: &Path,
) -> Result<PathBuf, ArtifactPathError> {
let mut current = root.to_path_buf();
for component in relative.components() {
let std::path::Component::Normal(name) = component else {
unreachable!("relative path was validated before directory preparation")
};
current.push(name);
match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: current.clone(),
});
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: current.clone(),
});
}
Ok(_) => {}
Err(source) if source.kind() == std::io::ErrorKind::NotFound => {
match fs::create_dir(¤t) {
Ok(()) => {}
Err(source) if source.kind() == std::io::ErrorKind::AlreadyExists => {
match fs::symlink_metadata(¤t) {
Ok(metadata) if metadata.file_type().is_symlink() => {
return Err(ArtifactPathError::SymlinkComponent {
path: current.clone(),
});
}
Ok(metadata) if !metadata.is_dir() => {
return Err(ArtifactPathError::DirectoryComponentNotDirectory {
path: current.clone(),
});
}
Ok(_) => {}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect raced artifact directory",
path: current.clone(),
source,
});
}
}
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "create artifact directory",
path: current.clone(),
source,
});
}
}
}
Err(source) => {
return Err(ArtifactPathError::Io {
operation: "inspect artifact directory",
path: current.clone(),
source,
});
}
}
}
Ok(current)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn artifact_id_rejects_an_empty_component() {
assert!(matches!(
ArtifactId::new(""),
Err(ArtifactPathError::InvalidArtifactId { .. })
));
}
#[test]
fn artifact_id_rejects_values_that_are_not_one_normal_path_component() {
for invalid in [
".",
"..",
"/absolute",
"nested/child",
r"nested\child",
r"C:\absolute",
] {
assert!(
matches!(
ArtifactId::new(invalid),
Err(ArtifactPathError::InvalidArtifactId { .. })
),
"accepted invalid artifact ID {invalid:?}"
);
}
}
#[test]
fn repeated_run_workspace_allocations_are_isolated() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let first = RunWorkspace::allocate(temp.path(), &logical_id).expect("first workspace");
let second = RunWorkspace::allocate(temp.path(), &logical_id).expect("second workspace");
assert_ne!(first.staging_path(), second.staging_path());
assert_ne!(first.published_path(), second.published_path());
assert!(first.staging_path().is_dir());
assert!(second.staging_path().is_dir());
assert!(!first.published_path().exists());
assert!(!second.published_path().exists());
}
#[test]
fn concurrent_run_workspace_allocations_are_isolated() {
use std::collections::BTreeSet;
use std::sync::{Arc, Barrier};
let temp = tempfile::tempdir().expect("tempdir");
let root = Arc::new(temp.path().to_path_buf());
let barrier = Arc::new(Barrier::new(8));
let handles: Vec<_> = (0..8)
.map(|_| {
let root = Arc::clone(&root);
let barrier = Arc::clone(&barrier);
std::thread::spawn(move || {
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
barrier.wait();
let workspace =
RunWorkspace::allocate(root.as_ref(), &logical_id).expect("workspace");
(
workspace.staging_path().to_path_buf(),
workspace.published_path().to_path_buf(),
)
})
})
.collect();
let paths: Vec<_> = handles
.into_iter()
.map(|handle| handle.join().expect("allocation thread"))
.collect();
let staging_paths: BTreeSet<_> = paths.iter().map(|(path, _)| path).collect();
let published_paths: BTreeSet<_> = paths.iter().map(|(_, path)| path).collect();
assert_eq!(staging_paths.len(), paths.len());
assert_eq!(published_paths.len(), paths.len());
}
#[test]
fn repeated_published_runs_do_not_inherit_stale_artifacts() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let required = [
ArtifactId::new("profile.json").expect("valid profile ID"),
ArtifactId::new("summary.md").expect("valid summary ID"),
];
let first = RunWorkspace::allocate(temp.path(), &logical_id).expect("first workspace");
fs::write(first.staging_path().join("profile.json"), "first profile")
.expect("write first profile");
fs::write(first.staging_path().join("summary.md"), "first summary")
.expect("write first summary");
fs::write(first.staging_path().join("stale.data"), "stale").expect("write stale artifact");
let first = first.publish(&required).expect("publish first run");
let second = RunWorkspace::allocate(temp.path(), &logical_id).expect("second workspace");
fs::write(second.staging_path().join("profile.json"), "second profile")
.expect("write second profile");
fs::write(second.staging_path().join("summary.md"), "second summary")
.expect("write second summary");
let second = second.publish(&required).expect("publish second run");
assert_ne!(first.path(), second.path());
assert!(first.path().join("stale.data").is_file());
assert!(!second.path().join("stale.data").exists());
}
#[cfg(unix)]
#[test]
fn run_workspace_rejects_a_symlink_publication_collision() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::tempdir().expect("outside tempdir");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let workspace =
RunWorkspace::allocate(temp.path(), &logical_id).expect("allocate workspace");
fs::write(workspace.staging_path().join("profile.json"), "complete")
.expect("write staged manifest");
symlink(outside.path(), workspace.published_path()).expect("create collision symlink");
let required = [ArtifactId::new("profile.json").expect("valid required file")];
let error = workspace
.publish(&required)
.expect_err("reject symlink collision");
assert!(matches!(error, ArtifactPathError::SymlinkComponent { .. }));
assert!(outside.path().is_dir());
assert!(!outside.path().join("profile.json").exists());
}
#[test]
fn incomplete_run_workspace_does_not_replace_latest_artifacts() {
let temp = tempfile::tempdir().expect("tempdir");
fs::write(temp.path().join("profile.json"), "old profile").expect("seed latest profile");
fs::write(temp.path().join("summary.md"), "old summary").expect("seed latest summary");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let workspace =
RunWorkspace::allocate(temp.path(), &logical_id).expect("allocate workspace");
let published_path = workspace.published_path().to_path_buf();
fs::write(workspace.staging_path().join("profile.json"), "new profile")
.expect("write staged profile");
let required = [
ArtifactId::new("profile.json").expect("valid profile ID"),
ArtifactId::new("summary.md").expect("valid summary ID"),
];
let error = workspace
.publish(&required)
.expect_err("reject incomplete run");
assert!(matches!(
error,
ArtifactPathError::RequiredArtifactMissing { .. }
));
assert!(!published_path.exists());
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("read latest profile"),
"old profile"
);
assert_eq!(
fs::read_to_string(temp.path().join("summary.md")).expect("read latest summary"),
"old summary"
);
}
#[test]
fn published_run_refreshes_stable_latest_artifacts() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let workspace =
RunWorkspace::allocate(temp.path(), &logical_id).expect("allocate workspace");
fs::write(workspace.staging_path().join("profile.json"), "new profile")
.expect("write staged profile");
fs::write(workspace.staging_path().join("summary.md"), "new summary")
.expect("write staged summary");
let required = [
ArtifactId::new("profile.json").expect("valid profile ID"),
ArtifactId::new("summary.md").expect("valid summary ID"),
];
let published = workspace.publish(&required).expect("publish complete run");
published
.refresh_latest(&[
LatestArtifact::same(required[0].clone()),
LatestArtifact::same(required[1].clone()),
])
.expect("refresh latest artifacts");
assert!(published.path().join("profile.json").is_file());
assert!(published.path().join("summary.md").is_file());
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("read latest profile"),
"new profile"
);
assert_eq!(
fs::read_to_string(temp.path().join("summary.md")).expect("read latest summary"),
"new summary"
);
}
#[test]
fn latest_refresh_removes_aliases_omitted_by_the_next_generation() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-run", "first");
refresh_pair(&first).expect("refresh initial aliases");
assert!(temp.path().join("summary.md").is_file());
let logical_id = ArtifactId::new("profile-only-run").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
fs::write(workspace.staging_path().join("profile.json"), "second").expect("write profile");
let profile = ArtifactId::new("profile.json").expect("profile ID");
let published = workspace
.publish(std::slice::from_ref(&profile))
.expect("publish profile-only run");
published
.refresh_latest(&[LatestArtifact::same(profile)])
.expect("refresh profile-only aliases");
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("profile alias"),
"second"
);
assert!(
!temp.path().join("summary.md").exists(),
"obsolete alias survived the generation change"
);
}
#[test]
fn latest_generation_retention_keeps_a_bounded_valid_chain() {
let temp = tempfile::tempdir().expect("tempdir");
for index in 0..(RETAIN_LATEST_GENERATIONS + 4) {
let published = publish_pair(
temp.path(),
&format!("retained-run-{index}"),
&format!("receipt-{index}"),
);
refresh_pair(&published).expect("refresh retained generation");
}
let generations = temp
.path()
.join(LATEST_DIRECTORY)
.join(LATEST_GENERATIONS_DIRECTORY);
assert_eq!(
fs::read_dir(&generations)
.expect("list retained generations")
.count(),
RETAIN_LATEST_GENERATIONS
);
let snapshot = LatestSnapshot::open(temp.path()).expect("open retained latest snapshot");
assert_eq!(
snapshot
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read retained latest artifact"),
format!("receipt-{}", RETAIN_LATEST_GENERATIONS + 3).as_bytes()
);
validate_predecessor_chain(
&ApprovedRoot::existing(temp.path().join(LATEST_DIRECTORY)).expect("open latest root"),
snapshot.manifest(),
)
.expect("retained predecessor chain remains valid");
}
#[test]
fn latest_generation_retention_defers_while_an_old_reader_is_leased() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "leased-run-0", "receipt-0");
refresh_pair(&first).expect("refresh first generation");
let old_reader = LatestSnapshot::open(temp.path()).expect("lease first generation");
for index in 1..=RETAIN_LATEST_GENERATIONS {
let published = publish_pair(
temp.path(),
&format!("leased-run-{index}"),
&format!("receipt-{index}"),
);
refresh_pair(&published).expect("refresh while old reader is leased");
}
let generations = temp
.path()
.join(LATEST_DIRECTORY)
.join(LATEST_GENERATIONS_DIRECTORY);
assert_eq!(
fs::read_dir(&generations)
.expect("list deferred generations")
.count(),
RETAIN_LATEST_GENERATIONS + 1
);
assert_eq!(
old_reader
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read leased generation"),
b"receipt-0"
);
drop(old_reader);
let final_run = publish_pair(temp.path(), "leased-run-final", "receipt-final");
refresh_pair(&final_run).expect("refresh after releasing old reader");
assert_eq!(
fs::read_dir(&generations)
.expect("list pruned generations")
.count(),
RETAIN_LATEST_GENERATIONS
);
}
#[test]
fn published_run_retention_is_bounded_and_defers_leased_runs() {
let temp = tempfile::tempdir().expect("tempdir");
let leased = publish_pair(temp.path(), "published-run-0", "receipt-0");
for index in 1..=RETAIN_PUBLISHED_RUNS {
drop(publish_pair(
temp.path(),
&format!("published-run-{index}"),
&format!("receipt-{index}"),
));
}
let count_runs = || {
fs::read_dir(temp.path())
.expect("list artifact root")
.filter_map(Result::ok)
.filter(|entry| entry.path().join(RUN_MANIFEST_FILE).is_file())
.count()
};
assert_eq!(count_runs(), RETAIN_PUBLISHED_RUNS + 1);
assert_eq!(
fs::read_to_string(leased.path().join("profile.json")).expect("read leased run"),
"receipt-0"
);
drop(leased);
drop(publish_pair(
temp.path(),
"published-run-final",
"receipt-final",
));
assert_eq!(count_runs(), RETAIN_PUBLISHED_RUNS);
}
#[test]
fn abandoned_workspace_quarantine_is_bounded() {
let temp = tempfile::tempdir().expect("tempdir");
let staging = temp.path().join(STAGING_DIRECTORY);
fs::create_dir_all(&staging).expect("create staging root");
for index in 0..(RETAIN_QUARANTINE_ENTRIES + 5) {
let crashed = staging.join(format!("crashed-{index}"));
fs::create_dir(&crashed).expect("create crashed workspace");
fs::write(crashed.join(WORKSPACE_LOCK_FILE), "").expect("write released lock");
}
let workspace = RunWorkspace::allocate(
temp.path(),
&ArtifactId::new("quarantine-writer").expect("logical ID"),
)
.expect("recover abandoned workspaces");
assert_eq!(
fs::read_dir(temp.path().join(STAGING_QUARANTINE_DIRECTORY))
.expect("list bounded quarantine")
.count(),
RETAIN_QUARANTINE_ENTRIES
);
drop(workspace);
}
#[test]
fn concurrent_latest_refreshes_leave_one_complete_alias_pair() {
use std::sync::{Arc, Barrier};
const WRITERS: usize = 8;
let temp = tempfile::tempdir().expect("tempdir");
let required = [
ArtifactId::new("profile.json").expect("valid profile ID"),
ArtifactId::new("summary.md").expect("valid summary ID"),
];
let mut published = Vec::new();
for index in 0..WRITERS {
let logical_id = ArtifactId::new(format!("run-{index}")).expect("valid logical ID");
let workspace =
RunWorkspace::allocate(temp.path(), &logical_id).expect("allocate workspace");
let receipt = format!("run-{index}");
fs::write(workspace.staging_path().join("profile.json"), &receipt)
.expect("write staged profile");
fs::write(workspace.staging_path().join("summary.md"), &receipt)
.expect("write staged summary");
published.push(workspace.publish(&required).expect("publish complete run"));
}
let barrier = Arc::new(Barrier::new(WRITERS));
let handles: Vec<_> = published
.into_iter()
.map(|published| {
let barrier = Arc::clone(&barrier);
let required = required.clone();
std::thread::spawn(move || {
barrier.wait();
published.refresh_latest(&[
LatestArtifact::same(required[0].clone()),
LatestArtifact::same(required[1].clone()),
])
})
})
.collect();
let mut committed = 0;
let mut rejected_as_stale = 0;
for handle in handles {
match handle.join().expect("latest refresh thread") {
Ok(()) => committed += 1,
Err(ArtifactPathError::StaleLatestGeneration { .. }) => rejected_as_stale += 1,
Err(error) => panic!("unexpected latest refresh error: {error}"),
}
}
assert_eq!(committed, 1, "exactly one shared CAS token may commit");
assert_eq!(rejected_as_stale, WRITERS - 1);
let profile =
fs::read_to_string(temp.path().join("profile.json")).expect("read latest profile");
let summary =
fs::read_to_string(temp.path().join("summary.md")).expect("read latest summary");
assert_eq!(profile, summary, "latest aliases came from different runs");
assert!(profile.starts_with("run-"));
}
#[test]
fn rollback_restores_other_backups_after_installed_file_removal_fails() {
let temp = tempfile::tempdir().expect("tempdir");
let blocked_destination = temp.path().join("blocked");
let restorable_destination = temp.path().join("restorable");
let blocked_backup = temp.path().join("blocked.backup");
let restorable_backup = temp.path().join("restorable.backup");
fs::create_dir(&blocked_destination).expect("create removal failure directory");
fs::write(&restorable_destination, "new").expect("write installed file");
fs::write(&blocked_backup, "old blocked").expect("write blocked backup");
fs::write(&restorable_backup, "old restored").expect("write restorable backup");
let original = ArtifactPathError::Io {
operation: "install latest artifact",
path: restorable_destination.clone(),
source: std::io::Error::other("simulated install failure"),
};
let error = latest_update_failure(
original,
&[restorable_destination.clone(), blocked_destination.clone()],
&[
(blocked_destination.clone(), blocked_backup.clone()),
(restorable_destination.clone(), restorable_backup),
],
temp.path().join("retained-rollback"),
);
assert!(matches!(error, ArtifactPathError::LatestRollback { .. }));
assert_eq!(
fs::read_to_string(&restorable_destination).expect("read restored backup"),
"old restored"
);
assert!(blocked_backup.is_file());
}
#[test]
fn incomplete_transaction_rollback_retains_backup_directory() {
let temp = tempfile::tempdir().expect("tempdir");
let transaction_path = temp.path().join("alias-transaction");
let backup_root = transaction_path.join("backup");
fs::create_dir_all(&backup_root).expect("create transaction backup root");
let destination = temp.path().join("profile.json");
fs::create_dir(&destination).expect("create blocking destination directory");
let backup = backup_root.join("profile.json");
fs::write(&backup, "previous").expect("write backup");
let transaction = StableAliasTransaction {
path: transaction_path.clone(),
root_path: temp.path().to_path_buf(),
installed: vec![destination.clone()],
backups: vec![(destination, backup.clone())],
};
let original = ArtifactPathError::Io {
operation: "install stable latest alias",
path: temp.path().join("profile.json"),
source: std::io::Error::other("injected install failure"),
};
let error = transaction.rollback(original);
assert!(matches!(
error,
ArtifactPathError::LatestRollback {
ref recovery_path,
..
} if recovery_path == &transaction_path
));
assert!(transaction_path.is_dir());
assert_eq!(
fs::read_to_string(backup).expect("read retained backup"),
"previous"
);
}
#[test]
fn latest_refresh_prepares_every_source_before_replacing_destinations() {
let temp = tempfile::tempdir().expect("tempdir");
fs::write(temp.path().join("profile.json"), "old profile").expect("seed latest profile");
fs::write(temp.path().join("summary.md"), "old summary").expect("seed latest summary");
let logical_id = ArtifactId::new("android-sample").expect("valid logical ID");
let workspace =
RunWorkspace::allocate(temp.path(), &logical_id).expect("allocate workspace");
fs::write(workspace.staging_path().join("profile.json"), "new profile")
.expect("write staged profile");
fs::write(workspace.staging_path().join("summary.md"), "new summary")
.expect("write staged summary");
let required = [
ArtifactId::new("profile.json").expect("valid profile ID"),
ArtifactId::new("summary.md").expect("valid summary ID"),
];
let published = workspace.publish(&required).expect("publish complete run");
fs::remove_file(published.path().join("summary.md"))
.expect("remove second published source");
let error = published
.refresh_latest(&[
LatestArtifact::same(required[0].clone()),
LatestArtifact::same(required[1].clone()),
])
.expect_err("reject incomplete latest source set");
assert!(matches!(
error,
ArtifactPathError::RequiredArtifactMissing { .. }
));
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("read latest profile"),
"old profile"
);
assert_eq!(
fs::read_to_string(temp.path().join("summary.md")).expect("read latest summary"),
"old summary"
);
}
#[test]
fn approved_root_prepares_nested_directory_beneath_root() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
let prepared = root
.prepare_dir("plots/nested")
.expect("prepare nested directory");
assert_eq!(prepared, root.path().join("plots/nested"));
assert!(prepared.is_dir());
}
#[test]
fn approved_root_projects_a_missing_directory_without_creating_it() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
let projected = root
.project_dir("target/mobench")
.expect("project output directory");
assert_eq!(projected, root.path().join("target/mobench"));
assert!(!projected.exists());
assert!(!root.path().join("target").exists());
}
#[test]
fn approved_root_rejects_paths_that_are_not_relative_descendants() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
for invalid in [Path::new("../escape"), Path::new("/absolute/escape")] {
assert!(matches!(
root.prepare_dir(invalid),
Err(ArtifactPathError::InvalidRelativePath { .. })
));
}
assert!(!root.path().join("../escape").exists());
}
#[test]
fn approved_root_projection_rejects_paths_that_are_not_relative_descendants() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
for invalid in [Path::new("../escape"), Path::new("/absolute/escape")] {
assert!(matches!(
root.project_dir(invalid),
Err(ArtifactPathError::InvalidRelativePath { .. })
));
}
}
#[cfg(unix)]
#[test]
fn approved_root_rejects_symlink_directory_components_without_following_them() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::tempdir().expect("outside tempdir");
symlink(outside.path(), temp.path().join("linked")).expect("create symlink");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
assert!(matches!(
root.prepare_dir("linked/nested"),
Err(ArtifactPathError::SymlinkComponent { .. })
));
assert!(!outside.path().join("nested").exists());
}
#[cfg(unix)]
#[test]
fn approved_root_projection_rejects_symlink_directory_components_without_writing() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::tempdir().expect("outside tempdir");
symlink(outside.path(), temp.path().join("linked")).expect("create symlink");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
assert!(matches!(
root.project_dir("linked/nested"),
Err(ArtifactPathError::SymlinkComponent { .. })
));
assert!(!outside.path().join("nested").exists());
}
#[test]
fn approved_root_prepares_file_parent_without_creating_the_file() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
let prepared = root
.prepare_file("plots/nested/chart.svg")
.expect("prepare artifact file");
assert_eq!(prepared, root.path().join("plots/nested/chart.svg"));
assert!(prepared.parent().expect("parent").is_dir());
assert!(!prepared.exists());
}
#[cfg(unix)]
#[test]
fn approved_root_rejects_preexisting_symlink_file_without_following_it() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approve root");
let plots = root.prepare_dir("plots").expect("prepare plots");
let outside = tempfile::NamedTempFile::new().expect("outside file");
fs::write(outside.path(), "unchanged").expect("seed outside file");
symlink(outside.path(), plots.join("chart.svg")).expect("create symlink");
assert!(matches!(
root.prepare_file("plots/chart.svg"),
Err(ArtifactPathError::SymlinkComponent { .. })
));
assert_eq!(
fs::read_to_string(outside.path()).expect("read outside file"),
"unchanged"
);
}
#[cfg(unix)]
#[test]
fn approved_root_rejects_a_symlink_as_the_root() {
use std::os::unix::fs::symlink;
let parent = tempfile::tempdir().expect("parent tempdir");
let actual_root = tempfile::tempdir().expect("actual root");
let linked_root = parent.path().join("linked-root");
symlink(actual_root.path(), &linked_root).expect("create root symlink");
assert!(matches!(
ApprovedRoot::existing(&linked_root),
Err(ArtifactPathError::SymlinkComponent { path }) if path == linked_root
));
}
fn publish_pair(root: &Path, logical: &str, receipt: &str) -> PublishedRun {
let logical_id = ArtifactId::new(logical).expect("valid logical ID");
let workspace = RunWorkspace::allocate(root, &logical_id).expect("allocate workspace");
fs::write(workspace.staging_path().join("profile.json"), receipt).expect("write profile");
fs::write(workspace.staging_path().join("summary.md"), receipt).expect("write summary");
workspace
.publish(&[
ArtifactId::new("profile.json").expect("profile ID"),
ArtifactId::new("summary.md").expect("summary ID"),
])
.expect("publish pair")
}
fn refresh_pair(published: &PublishedRun) -> Result<(), ArtifactPathError> {
published.refresh_latest(&[
LatestArtifact::same(ArtifactId::new("profile.json").expect("profile ID")),
LatestArtifact::same(ArtifactId::new("summary.md").expect("summary ID")),
])
}
fn durability_events() -> Vec<&'static str> {
DURABILITY_EVENTS.with(|events| std::mem::take(&mut *events.borrow_mut()))
}
fn rewrite_latest_manifest(generation_path: &Path, update: impl FnOnce(&mut LatestManifest)) {
let manifest_path = generation_path.join(LATEST_MANIFEST_FILE);
let mut manifest: LatestManifest =
read_json_file(&manifest_path, "read test manifest").expect("read manifest");
update(&mut manifest);
fs::remove_file(&manifest_path).expect("remove old manifest");
write_json_file(&manifest_path, &manifest, "rewrite test manifest")
.expect("rewrite manifest");
}
fn clone_generation(
source: &LatestSnapshot,
generation: &str,
predecessor: Option<&str>,
) -> PathBuf {
let generations = source
.path()
.parent()
.expect("generation parent")
.to_path_buf();
let destination = generations.join(generation);
fs::create_dir(&destination).expect("create cloned generation");
for artifact in &source.manifest().artifacts {
fs::copy(
source.path().join(&artifact.destination_relative_path),
destination.join(&artifact.destination_relative_path),
)
.expect("copy generation artifact");
}
create_reader_lease_file(&destination).expect("create cloned generation lease");
let mut manifest = source.manifest().clone();
manifest.generation = generation.to_owned();
manifest.predecessor_generation = predecessor.map(str::to_owned);
write_json_file(
&destination.join(LATEST_MANIFEST_FILE),
&manifest,
"write cloned generation manifest",
)
.expect("write cloned manifest");
destination
}
#[test]
fn writer_startup_quarantines_an_unlocked_crash_workspace() {
let temp = tempfile::tempdir().expect("tempdir");
let staging = temp.path().join(STAGING_DIRECTORY);
let crashed = staging.join("crashed-workspace");
fs::create_dir_all(&crashed).expect("create crashed workspace");
fs::write(crashed.join(WORKSPACE_LOCK_FILE), "").expect("create released lock");
fs::write(crashed.join("partial.data"), "partial").expect("write partial artifact");
let workspace = RunWorkspace::allocate(
temp.path(),
&ArtifactId::new("recovery-writer").expect("logical ID"),
)
.expect("allocate after crash");
assert!(!crashed.exists());
let quarantine = temp.path().join(STAGING_QUARANTINE_DIRECTORY);
assert!(
fs::read_dir(quarantine)
.expect("read run quarantine")
.any(|entry| entry
.expect("quarantine entry")
.file_name()
.to_string_lossy()
.starts_with("crashed-workspace--"))
);
drop(workspace);
}
#[test]
fn writer_startup_never_quarantines_an_active_workspace() {
let temp = tempfile::tempdir().expect("tempdir");
let first = RunWorkspace::allocate(
temp.path(),
&ArtifactId::new("active-first").expect("logical ID"),
)
.expect("first workspace");
let first_path = first.staging_path().to_path_buf();
let second = RunWorkspace::allocate(
temp.path(),
&ArtifactId::new("active-second").expect("logical ID"),
)
.expect("second workspace");
assert!(first_path.is_dir(), "active workspace was moved");
assert!(first_path.join(WORKSPACE_LOCK_FILE).is_file());
drop(second);
drop(first);
}
#[test]
fn run_manifest_records_deterministic_recursive_digests_and_identity() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("android-sample").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
fs::write(workspace.staging_path().join("profile.json"), "hello").expect("write profile");
fs::create_dir(workspace.staging_path().join("nested")).expect("create nested");
fs::write(workspace.staging_path().join("nested/notes.txt"), "notes").expect("write notes");
let published = workspace
.publish(&[ArtifactId::new("profile.json").expect("profile ID")])
.expect("publish");
let manifest = published.manifest().expect("verified manifest");
assert_eq!(manifest.format_version, RUN_MANIFEST_VERSION);
assert_eq!(manifest.producer_version, env!("CARGO_PKG_VERSION"));
assert_eq!(manifest.logical_id, "android-sample");
assert_eq!(
manifest.publication_id,
published
.path()
.file_name()
.expect("publication name")
.to_string_lossy()
);
assert_eq!(
manifest
.artifacts
.iter()
.map(|artifact| artifact.relative_path.as_str())
.collect::<Vec<_>>(),
["nested/notes.txt", "profile.json"]
);
let profile = manifest
.artifacts
.iter()
.find(|artifact| artifact.relative_path == "profile.json")
.expect("profile manifest entry");
assert_eq!(profile.size, 5);
assert_eq!(
profile.sha256,
"2cf24dba5fb0a30e26e83b2ac5b9e29e1b161e5c1fa7425e73043362938b9824"
);
}
#[test]
fn publication_syncs_files_and_directories_before_the_atomic_rename() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("ordered-run").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
fs::write(workspace.staging_path().join("profile.json"), "complete")
.expect("write profile");
durability_events();
workspace
.publish(&[ArtifactId::new("profile.json").expect("profile ID")])
.expect("publish");
let events = durability_events();
let publication = events
.iter()
.position(|event| *event == "publish_run")
.expect("publication event");
assert!(events[..publication].contains(&"sync_manifest"));
assert!(events[..publication].contains(&"sync_file"));
assert!(events[..publication].contains(&"sync_directory"));
assert!(events[publication + 1..].contains(&"sync_directory"));
}
#[test]
fn post_rename_sync_failure_reports_committed_run_explicitly() {
let temp = tempfile::tempdir().expect("tempdir");
let logical_id = ArtifactId::new("uncertain-run").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
fs::write(workspace.staging_path().join("profile.json"), "complete")
.expect("write profile");
let expected_path = workspace.published_path().to_path_buf();
FAIL_DIRECTORY_SYNC_AFTER_EVENT.with(|target| target.set(Some("publish_run")));
let error = workspace
.publish(&[ArtifactId::new("profile.json").expect("profile ID")])
.expect_err("surface post-commit durability uncertainty");
assert!(matches!(
error,
ArtifactPathError::PublicationDurabilityUncertain { ref path, .. }
if path == &expected_path
));
assert!(expected_path.join(RUN_MANIFEST_FILE).is_file());
validate_run_manifest(&expected_path).expect("committed run remains valid");
}
#[test]
fn latest_commit_orders_generation_sync_before_the_single_pointer_swap() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "ordered-latest", "receipt");
durability_events();
refresh_pair(&published).expect("refresh latest");
let events = durability_events();
let generation = events
.iter()
.position(|event| *event == "publish_generation")
.expect("generation publication event");
let pointer_sync = events
.iter()
.position(|event| *event == "sync_pointer_file")
.expect("pointer sync event");
let pointer_commit = events
.iter()
.position(|event| *event == "commit_pointer")
.expect("pointer commit event");
assert!(events[..generation].contains(&"sync_manifest"));
assert!(events[generation + 1..pointer_sync].contains(&"sync_directory"));
let aliases = events
.iter()
.position(|event| *event == "install_aliases")
.expect("alias install event");
assert!(generation < pointer_sync);
assert!(
aliases < pointer_sync,
"aliases must precede pointer commit"
);
assert!(pointer_sync < pointer_commit);
assert!(events[pointer_commit + 1..].contains(&"sync_directory"));
}
#[test]
fn post_pointer_sync_failure_reports_committed_generation_explicitly() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "uncertain-latest", "receipt");
FAIL_DIRECTORY_SYNC_AFTER_EVENT.with(|target| target.set(Some("commit_pointer")));
let error = refresh_pair(&published).expect_err("surface pointer durability uncertainty");
let generation = match error {
ArtifactPathError::LatestCommitDurabilityUncertain { generation, .. } => generation,
other => panic!("unexpected error: {other}"),
};
let snapshot = LatestSnapshot::open(temp.path()).expect("committed snapshot remains open");
assert_eq!(snapshot.generation(), generation);
assert_eq!(
snapshot
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read committed artifact"),
b"receipt"
);
}
#[test]
fn latest_snapshot_pins_one_generation_across_concurrent_refreshes() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-run", "first");
refresh_pair(&first).expect("first refresh");
let pinned = LatestSnapshot::open(temp.path()).expect("first snapshot");
let second = publish_pair(temp.path(), "second-run", "second");
refresh_pair(&second).expect("second refresh");
let current = LatestSnapshot::open(temp.path()).expect("second snapshot");
assert_ne!(pinned.generation(), current.generation());
for id in ["profile.json", "summary.md"] {
let id = ArtifactId::new(id).expect("artifact ID");
assert_eq!(pinned.read_artifact(&id).expect("pinned read"), b"first");
assert_eq!(current.read_artifact(&id).expect("current read"), b"second");
}
}
#[cfg(unix)]
#[test]
fn latest_snapshot_open_succeeds_with_a_read_only_artifact_root() {
use std::os::unix::fs::PermissionsExt;
fn set_tree_mode(path: &Path, directory_mode: u32, file_mode: u32) {
let metadata = fs::symlink_metadata(path).expect("tree metadata");
if metadata.is_dir() {
for entry in fs::read_dir(path).expect("read tree") {
set_tree_mode(
&entry.expect("tree entry").path(),
directory_mode,
file_mode,
);
}
fs::set_permissions(path, fs::Permissions::from_mode(directory_mode))
.expect("set directory mode");
} else {
fs::set_permissions(path, fs::Permissions::from_mode(file_mode))
.expect("set file mode");
}
}
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "read-only", "receipt");
refresh_pair(&published).expect("refresh latest");
set_tree_mode(temp.path(), 0o555, 0o444);
let opened = LatestSnapshot::open(temp.path()).expect("read-only snapshot open");
assert_eq!(
opened
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read snapshot artifact"),
b"receipt"
);
set_tree_mode(temp.path(), 0o755, 0o644);
}
#[test]
fn writer_allocation_recovers_mixed_legacy_aliases() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "recover-run", "committed");
refresh_pair(&published).expect("refresh latest");
fs::write(temp.path().join("profile.json"), "interrupted-new")
.expect("simulate interrupted alias install");
fs::write(temp.path().join("summary.md"), "committed").expect("retain old alias");
LatestSnapshot::open(temp.path()).expect("read-only open");
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("unrecovered profile"),
"interrupted-new",
"reader open must not mutate compatibility aliases"
);
let recovery_probe = RunWorkspace::allocate(
temp.path(),
&ArtifactId::new("recovery-probe").expect("probe ID"),
)
.expect("writer startup recovers aliases");
drop(recovery_probe);
let recovered = LatestSnapshot::open(temp.path()).expect("open recovered snapshot");
assert_eq!(
fs::read(temp.path().join("profile.json")).expect("profile alias"),
b"committed"
);
assert_eq!(
fs::read(temp.path().join("summary.md")).expect("summary alias"),
b"committed"
);
assert_eq!(
recovered
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read recovered snapshot"),
b"committed"
);
}
#[test]
fn missing_current_recovers_the_unique_highest_chain_tip() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-chain-run", "first");
refresh_pair(&first).expect("first refresh");
let second = publish_pair(temp.path(), "second-chain-run", "second");
refresh_pair(&second).expect("second refresh");
let expected = LatestSnapshot::open(temp.path()).expect("tip snapshot");
let expected_generation = expected.generation().to_owned();
let current = temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE);
fs::remove_file(¤t).expect("remove current pointer");
fs::write(temp.path().join("profile.json"), "interrupted")
.expect("tamper compatibility alias");
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::LatestSnapshotUnavailable { .. })
));
let recovered =
LatestSnapshot::recover_stable_aliases(temp.path()).expect("recover missing pointer");
assert_eq!(recovered.generation(), expected_generation);
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("recovered profile"),
"second"
);
}
#[test]
fn corrupt_current_recovers_the_unique_valid_chain_tip() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "corrupt-pointer-run", "receipt");
refresh_pair(&published).expect("refresh latest");
let expected = LatestSnapshot::open(temp.path())
.expect("snapshot")
.generation()
.to_owned();
let current = temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE);
fs::write(¤t, "../../invalid\n").expect("corrupt pointer");
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::InvalidLatestPointer { .. })
));
let recovered =
LatestSnapshot::recover_stable_aliases(temp.path()).expect("recover corrupt pointer");
assert_eq!(recovered.generation(), expected);
assert_eq!(
fs::read_to_string(current)
.expect("read repaired pointer")
.trim(),
expected
);
}
#[test]
fn missing_current_fails_closed_on_an_ambiguous_generation_fork() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "fork-root-run", "root");
refresh_pair(&published).expect("refresh root");
let root_snapshot = LatestSnapshot::open(temp.path()).expect("root snapshot");
let root_generation = root_snapshot.generation().to_owned();
clone_generation(&root_snapshot, "fork-child-a", Some(&root_generation));
clone_generation(&root_snapshot, "fork-child-b", Some(&root_generation));
fs::remove_file(temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE))
.expect("remove current pointer");
assert!(matches!(
LatestSnapshot::recover_stable_aliases(temp.path()),
Err(ArtifactPathError::AmbiguousGenerationFork { .. })
));
}
#[test]
fn predecessor_cycle_is_rejected_by_open_and_recovery() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "cycle-first", "first");
refresh_pair(&first).expect("first refresh");
let first_snapshot = LatestSnapshot::open(temp.path()).expect("first snapshot");
let first_generation = first_snapshot.generation().to_owned();
let second = publish_pair(temp.path(), "cycle-second", "second");
refresh_pair(&second).expect("second refresh");
let second_snapshot = LatestSnapshot::open(temp.path()).expect("second snapshot");
let second_generation = second_snapshot.generation().to_owned();
rewrite_latest_manifest(&first_snapshot.generation_path, |manifest| {
manifest.predecessor_generation = Some(second_generation.clone());
});
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::GenerationChainCycle { .. })
));
fs::remove_file(temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE))
.expect("remove current pointer");
assert!(matches!(
LatestSnapshot::recover_stable_aliases(temp.path()),
Err(ArtifactPathError::GenerationChainCycle { .. })
));
assert_ne!(first_generation, second_generation);
}
#[test]
fn broken_predecessor_chain_fails_closed_during_recovery() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "broken-chain", "receipt");
refresh_pair(&published).expect("refresh latest");
let snapshot = LatestSnapshot::open(temp.path()).expect("snapshot");
rewrite_latest_manifest(snapshot.path(), |manifest| {
manifest.predecessor_generation = Some("missing-predecessor".to_owned());
});
fs::remove_file(temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE))
.expect("remove current pointer");
assert!(matches!(
LatestSnapshot::recover_stable_aliases(temp.path()),
Err(ArtifactPathError::BrokenGenerationChain { .. })
));
}
#[test]
fn alias_install_failure_rolls_back_before_pointer_commit() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-run", "first");
refresh_pair(&first).expect("first refresh");
let before = LatestSnapshot::open(temp.path()).expect("snapshot before failure");
let before_generation = before.generation().to_owned();
let second = publish_pair(temp.path(), "second-run", "second");
FAIL_ALIAS_INSTALL_AFTER.with(|fail_after| fail_after.set(Some(1)));
let error = refresh_pair(&second).expect_err("injected alias failure");
assert!(matches!(error, ArtifactPathError::Io { .. }));
let after = LatestSnapshot::open(temp.path()).expect("snapshot after rollback");
assert_eq!(after.generation(), before_generation);
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("profile alias"),
"first"
);
assert_eq!(
fs::read_to_string(temp.path().join("summary.md")).expect("summary alias"),
"first"
);
}
#[test]
fn stale_allocated_run_cannot_replace_a_newer_generation() {
let temp = tempfile::tempdir().expect("tempdir");
let stale = publish_pair(temp.path(), "stale-run", "stale");
let winner = publish_pair(temp.path(), "winner-run", "winner");
refresh_pair(&winner).expect("winner refresh");
let winner_generation = LatestSnapshot::open(temp.path())
.expect("winner snapshot")
.generation()
.to_owned();
let error = refresh_pair(&stale).expect_err("reject stale CAS token");
assert!(matches!(
error,
ArtifactPathError::StaleLatestGeneration {
expected: None,
observed: Some(_)
}
));
let current = LatestSnapshot::open(temp.path()).expect("current snapshot");
assert_eq!(current.generation(), winner_generation);
assert_eq!(
current
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read winner"),
b"winner"
);
}
#[test]
fn abandoned_latest_staging_is_quarantined_before_the_next_commit() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-run", "first");
refresh_pair(&first).expect("first refresh");
let staging = temp
.path()
.join(LATEST_DIRECTORY)
.join(LATEST_STAGING_DIRECTORY);
let abandoned = staging.join("abandoned");
fs::create_dir(&abandoned).expect("create abandoned staging");
fs::write(abandoned.join("partial"), "partial").expect("write partial staging");
let second = publish_pair(temp.path(), "second-run", "second");
refresh_pair(&second).expect("second refresh");
assert!(!abandoned.exists());
let quarantine = temp
.path()
.join(LATEST_DIRECTORY)
.join(LATEST_QUARANTINE_DIRECTORY);
assert!(
fs::read_dir(quarantine)
.expect("read quarantine")
.any(|entry| entry
.expect("quarantine entry")
.file_name()
.to_string_lossy()
.starts_with("abandoned--"))
);
}
#[test]
fn active_latest_staging_lease_defers_recovery_quarantine() {
let temp = tempfile::tempdir().expect("tempdir");
let root = ApprovedRoot::existing(temp.path()).expect("approved root");
let latest_path = root
.prepare_dir(LATEST_DIRECTORY)
.expect("latest directory");
let latest_root = ApprovedRoot::existing(latest_path).expect("approved latest root");
latest_root
.prepare_dir(LATEST_QUARANTINE_DIRECTORY)
.expect("quarantine directory");
let (_, staging_path) = allocate_latest_generation(&latest_root).expect("staging path");
create_reader_lease_file(&staging_path).expect("staging lease file");
let lease_path = staging_path.join(LATEST_READER_LOCK_FILE);
let lease = OpenOptions::new()
.read(true)
.write(true)
.open(&lease_path)
.expect("open staging lease");
FileExt::lock_exclusive(&lease).expect("lock active staging lease");
quarantine_abandoned_latest_staging(&latest_root).expect("active staging recovery pass");
assert!(staging_path.exists(), "active staging was quarantined");
FileExt::unlock(&lease).expect("unlock staging lease");
drop(lease);
quarantine_abandoned_latest_staging(&latest_root).expect("abandoned staging recovery pass");
assert!(!staging_path.exists(), "abandoned staging was retained");
}
#[test]
fn newer_latest_producer_version_rejects_a_downgrade() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "first-run", "first");
refresh_pair(&first).expect("first refresh");
let second = publish_pair(temp.path(), "second-run", "second");
let snapshot = LatestSnapshot::open(temp.path()).expect("snapshot");
let manifest_path = snapshot.path().join(LATEST_MANIFEST_FILE);
let mut manifest: LatestManifest =
read_json_file(&manifest_path, "read test manifest").expect("manifest");
manifest.producer_version = "999.0.0".to_owned();
fs::remove_file(&manifest_path).expect("remove old manifest");
write_json_file(&manifest_path, &manifest, "rewrite test manifest")
.expect("write newer producer manifest");
let error = refresh_pair(&second).expect_err("reject producer downgrade");
assert!(matches!(error, ArtifactPathError::ProducerDowngrade { .. }));
assert_eq!(
fs::read_to_string(temp.path().join("profile.json")).expect("stable profile"),
"first"
);
}
#[test]
fn unknown_latest_manifest_version_fails_closed() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "unknown-version", "receipt");
refresh_pair(&published).expect("refresh latest");
let snapshot = LatestSnapshot::open(temp.path()).expect("snapshot");
let manifest_path = snapshot.path().join(LATEST_MANIFEST_FILE);
let mut manifest: LatestManifest =
read_json_file(&manifest_path, "read test manifest").expect("manifest");
manifest.format_version = LATEST_MANIFEST_VERSION + 1;
fs::remove_file(&manifest_path).expect("remove old manifest");
write_json_file(&manifest_path, &manifest, "rewrite test manifest")
.expect("write unknown manifest");
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::UnknownManifestVersion { .. })
));
}
#[test]
fn corrupt_latest_generation_fails_digest_validation() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "corrupt-generation", "receipt");
refresh_pair(&published).expect("refresh latest");
let snapshot = LatestSnapshot::open(temp.path()).expect("snapshot");
fs::write(snapshot.path().join("profile.json"), "tampered").expect("tamper artifact");
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::CorruptGeneration { .. })
));
}
#[test]
fn pinned_snapshot_rejects_bytes_changed_after_open() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "mutated-after-open", "original");
refresh_pair(&published).expect("refresh latest");
let snapshot = LatestSnapshot::open(temp.path()).expect("open pinned snapshot");
fs::write(snapshot.path().join("profile.json"), "changed").expect("mutate artifact");
assert!(matches!(
snapshot.read_artifact(&ArtifactId::new("profile.json").expect("profile ID")),
Err(ArtifactPathError::CorruptGeneration { .. })
));
}
#[test]
fn latest_refresh_rejects_source_changed_after_run_validation() {
let temp = tempfile::tempdir().expect("tempdir");
let published = publish_pair(temp.path(), "mutated-source", "original");
MUTATE_LATEST_SOURCE_AFTER_VALIDATION.with(|mutation| {
mutation.replace(Some((
published.path().join("profile.json"),
b"changed after validation".to_vec(),
)));
});
let error = refresh_pair(&published).expect_err("reject changed source");
assert!(matches!(error, ArtifactPathError::ManifestIntegrity { .. }));
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::LatestSnapshotUnavailable { .. })
));
}
#[test]
fn invalid_current_directory_does_not_block_first_commit() {
let temp = tempfile::tempdir().expect("tempdir");
let latest = temp.path().join(LATEST_DIRECTORY);
fs::create_dir_all(latest.join(LATEST_CURRENT_FILE)).expect("seed invalid pointer dir");
let published = publish_pair(temp.path(), "first-after-invalid-pointer", "receipt");
refresh_pair(&published).expect("commit after quarantining invalid pointer");
let snapshot = LatestSnapshot::open(temp.path()).expect("open committed snapshot");
assert_eq!(
snapshot
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read committed artifact"),
b"receipt"
);
assert!(
fs::read_dir(latest.join(LATEST_QUARANTINE_DIRECTORY))
.expect("read quarantine")
.any(|entry| entry
.expect("quarantine entry")
.file_name()
.to_string_lossy()
.starts_with("current-corrupt--"))
);
}
#[test]
fn corrupt_off_chain_generation_is_quarantined_without_blocking_writer() {
let temp = tempfile::tempdir().expect("tempdir");
let first = publish_pair(temp.path(), "anchored", "first");
refresh_pair(&first).expect("anchor current generation");
let orphan = temp
.path()
.join(LATEST_DIRECTORY)
.join(LATEST_GENERATIONS_DIRECTORY)
.join("generation-corrupt-orphan");
fs::create_dir(&orphan).expect("create corrupt orphan");
fs::write(orphan.join(LATEST_MANIFEST_FILE), b"not-json").expect("write corrupt orphan");
let second = publish_pair(temp.path(), "after-orphan", "second");
refresh_pair(&second).expect("refresh despite corrupt orphan");
assert!(!orphan.exists());
let snapshot = LatestSnapshot::open(temp.path()).expect("open latest snapshot");
assert_eq!(
snapshot
.read_artifact(&ArtifactId::new("profile.json").expect("profile ID"))
.expect("read latest artifact"),
b"second"
);
}
#[cfg(unix)]
#[test]
fn recursive_publication_validation_rejects_nested_symlinks() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::NamedTempFile::new().expect("outside file");
let logical_id = ArtifactId::new("nested-symlink").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
fs::write(workspace.staging_path().join("profile.json"), "complete")
.expect("write profile");
fs::create_dir(workspace.staging_path().join("nested")).expect("nested directory");
symlink(
outside.path(),
workspace.staging_path().join("nested/linked"),
)
.expect("nested symlink");
assert!(matches!(
workspace.publish(&[ArtifactId::new("profile.json").expect("profile ID")]),
Err(ArtifactPathError::SymlinkComponent { .. })
));
}
#[cfg(unix)]
#[test]
fn publication_rejects_artifacts_hardlinked_outside_the_workspace() {
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::tempdir().expect("outside tempdir");
let logical_id = ArtifactId::new("hardlinked-artifact").expect("logical ID");
let workspace = RunWorkspace::allocate(temp.path(), &logical_id).expect("workspace");
let profile = workspace.staging_path().join("profile.json");
fs::write(&profile, "complete").expect("write profile");
fs::hard_link(&profile, outside.path().join("alias.json")).expect("create hard link");
assert!(matches!(
workspace.publish(&[ArtifactId::new("profile.json").expect("profile ID")]),
Err(ArtifactPathError::HardLinkedArtifact { .. })
));
}
#[cfg(any(
target_os = "macos",
target_os = "ios",
target_os = "linux",
target_os = "android"
))]
#[test]
fn platform_publication_rename_never_replaces_a_raced_directory() {
let temp = tempfile::tempdir().expect("tempdir");
let source = temp.path().join("source");
let destination = temp.path().join("destination");
fs::create_dir(&source).expect("create source");
fs::create_dir(&destination).expect("create raced destination");
fs::write(source.join("source.txt"), "source").expect("write source");
fs::write(destination.join("destination.txt"), "destination").expect("write destination");
let error = rename_directory_noreplace(&source, &destination)
.expect_err("no-replace rename must reject collision");
assert!(matches!(
error.kind(),
std::io::ErrorKind::AlreadyExists | std::io::ErrorKind::Other
));
assert!(source.join("source.txt").is_file());
assert_eq!(
fs::read_to_string(destination.join("destination.txt"))
.expect("read original destination"),
"destination"
);
}
#[cfg(unix)]
#[test]
fn latest_reader_rejects_a_symlink_pointer() {
use std::os::unix::fs::symlink;
let temp = tempfile::tempdir().expect("tempdir");
let outside = tempfile::NamedTempFile::new().expect("outside pointer");
fs::write(outside.path(), "generation-attacker").expect("write outside pointer");
let published = publish_pair(temp.path(), "symlink-pointer", "receipt");
refresh_pair(&published).expect("refresh latest");
let current = temp.path().join(LATEST_DIRECTORY).join(LATEST_CURRENT_FILE);
fs::remove_file(¤t).expect("remove current pointer");
symlink(outside.path(), ¤t).expect("install pointer symlink");
assert!(matches!(
LatestSnapshot::open(temp.path()),
Err(ArtifactPathError::SymlinkComponent { .. })
));
}
}