use std::collections::BTreeSet;
use std::fs;
use std::io::Read as _;
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};
use serde::{Deserialize, Serialize};
use crate::compiler::diagnostics_store::CompilerFactCacheKey;
use crate::compiler::model::{CompilerDiagKey, PlatformTarget};
use crate::compiler::scheduler::{CompilerAcquisitionPlan, CompilerAcquisitionView, ViewIx};
use crate::error::{RailError, RailResult};
use crate::progress;
use crate::source::ContentDigest;
const ACQUISITION_CONTRACT_VERSION: u32 = 3;
const ACQUISITION_IDENTITY_PREFIX: &str = "surface-acquisition-v3-sha256-";
const EVIDENCE_IDENTITY_PREFIX: &str = "surface-acquisition-evidence-v1-sha256-";
const JOURNAL_DIRECTORY: &str = "surface-acquisitions-v3";
const MAX_JOURNAL_BYTES: u64 = 64 * 1024 * 1024;
const MAX_JOURNAL_GROUPS: usize = 256;
const COMPLETION_FLUSH_INTERVAL: Duration = Duration::from_millis(250);
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub(crate) struct CompilerAcquisitionProduct {
pub(crate) package: String,
pub(crate) cargo_target: String,
pub(crate) kind: String,
}
#[derive(Debug, Clone)]
pub(crate) struct CompilerAcquisitionRequest {
pub(crate) workspace_identity: String,
pub(crate) checkout_identity: String,
pub(crate) snapshot_identity: String,
pub(crate) configuration_fingerprint: String,
pub(crate) products: Vec<CompilerAcquisitionProduct>,
pub(crate) resume_manifest: Option<PathBuf>,
}
#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) struct CompilerAcquisitionCargoTarget {
pub(crate) package: String,
pub(crate) target: String,
pub(crate) kinds: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(transparent)]
pub(crate) struct EvidenceIdentity(String);
impl EvidenceIdentity {
fn new(bytes: &[u8]) -> Self {
Self(format!("{EVIDENCE_IDENTITY_PREFIX}{}", ContentDigest::sha256(bytes)))
}
fn validate(&self) -> RailResult<()> {
validate_sha256_identity(&self.0, EVIDENCE_IDENTITY_PREFIX, "acquisition evidence")
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "kebab-case")]
pub(crate) enum FailureClass {
Worker,
Integration,
Coordinator,
Journal,
Sandbox,
Broker,
Cancelled,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(tag = "state", rename_all = "kebab-case", deny_unknown_fields)]
enum DurableViewState {
Pending,
Running { attempt: u32 },
Complete { evidence: EvidenceIdentity },
Failed { class: FailureClass },
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct DurableViewRecord {
view_index: usize,
view_identity: String,
selected_products: Vec<CompilerAcquisitionProduct>,
packages: [String; 1],
target_triple: String,
feature_profile: String,
command_class: String,
cargo_targets: Vec<CompilerAcquisitionCargoTarget>,
attempts: u32,
durable: DurableViewState,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct DurableGroupState {
sequence: u64,
views: Vec<usize>,
terminal: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct JournalSchemas {
compiler_fact: u32,
observation: u32,
native_cache: u32,
collector: u32,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct CompilerAcquisitionHeader {
record: String,
surface_acquisition_contract_version: u32,
acquisition_identity: String,
workspace_identity: String,
checkout_identity: String,
snapshot_identity: String,
configuration_fingerprint: String,
plan_identity: String,
candidate_set_identity: String,
compiler_set_identity: String,
schemas: JournalSchemas,
manifest: String,
resume_command: Vec<String>,
concurrency: usize,
journal_batch: usize,
artifact_soft_limit_bytes: u64,
artifact_hard_limit_bytes: u64,
products: Vec<CompilerAcquisitionProduct>,
view_count: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct DurableSummary {
state: String,
failure: Option<FailureClass>,
pending: usize,
running: usize,
completed: usize,
failed: usize,
}
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
struct AcquisitionJournal {
header: CompilerAcquisitionHeader,
views: Vec<DurableViewRecord>,
groups: Vec<DurableGroupState>,
sequence: u64,
summary: Option<DurableSummary>,
}
pub(crate) struct CompilerAcquisitionJournal {
path: PathBuf,
document: AcquisitionJournal,
dirty_views: BTreeSet<usize>,
pending_completions: usize,
pending_since: Option<Instant>,
revalidation_required: BTreeSet<usize>,
}
impl CompilerAcquisitionJournal {
#[expect(
clippy::too_many_arguments,
reason = "the journal header binds every independent acquisition authority at one boundary"
)]
pub(crate) fn begin(
workspace_root: &Path,
request: &CompilerAcquisitionRequest,
plan: &CompilerAcquisitionPlan,
compiler_set_identity: String,
artifact_soft_limit_bytes: u64,
artifact_hard_limit_bytes: u64,
concurrency: usize,
journal_batch: usize,
) -> RailResult<Self> {
if journal_batch == 0 {
return Err(RailError::message(
"compiler acquisition journal batch must be non-zero",
));
}
let state_root = crate::workspace::cargo_rail_state_root(workspace_root);
let canonical_state_root = crate::utils::canonicalize_existing(&state_root)?;
let directory = state_root.join(JOURNAL_DIRECTORY);
match fs::symlink_metadata(&directory) {
Ok(metadata) if metadata.is_dir() && !crate::utils::is_symlink_or_reparse(&metadata) => {}
Ok(_) => {
return Err(RailError::message(format!(
"Surface acquisition journal directory '{}' is not a real directory",
directory.display()
)));
}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
fs::create_dir(&directory).map_err(|error| {
RailError::message(format!(
"failed to create Surface acquisition journal directory '{}': {error}",
directory.display()
))
})?;
}
Err(error) => return Err(error.into()),
}
let directory = crate::utils::canonicalize_existing(&directory)?;
if directory.parent() != Some(canonical_state_root.as_path()) {
return Err(RailError::message(
"Surface acquisition journal directory escaped the workspace state root",
));
}
let schemas = JournalSchemas {
compiler_fact: crate::compiler::facts::COMPILER_FACT_PROTOCOL_VERSION,
observation: crate::compiler::invocation::OBSERVATION_PROTOCOL_VERSION,
native_cache: crate::compiler::native_cache::native_cache_capability_schema_version(),
collector: crate::compiler::model::COLLECTOR_VERSION,
};
let binding_bytes = serde_json::to_vec(&(
ACQUISITION_CONTRACT_VERSION,
&request.workspace_identity,
&request.checkout_identity,
&request.snapshot_identity,
&request.configuration_fingerprint,
plan.identity().as_str(),
plan.candidate_set_identity(),
&compiler_set_identity,
&schemas,
))?;
let acquisition_identity = format!("{ACQUISITION_IDENTITY_PREFIX}{}", ContentDigest::sha256(&binding_bytes));
let path = directory.join(format!("{acquisition_identity}.json"));
let display_path = display_path(workspace_root, &path);
let resume_command = vec![
"cargo".to_string(),
"rail".to_string(),
"surface".to_string(),
"--resume".to_string(),
display_path.clone(),
"--format".to_string(),
"json".to_string(),
];
let header = CompilerAcquisitionHeader {
record: "manifest".to_string(),
surface_acquisition_contract_version: ACQUISITION_CONTRACT_VERSION,
acquisition_identity,
workspace_identity: request.workspace_identity.clone(),
checkout_identity: request.checkout_identity.clone(),
snapshot_identity: request.snapshot_identity.clone(),
configuration_fingerprint: request.configuration_fingerprint.clone(),
plan_identity: plan.identity().as_str().to_string(),
candidate_set_identity: plan.candidate_set_identity().to_string(),
compiler_set_identity,
schemas,
manifest: display_path.clone(),
resume_command,
concurrency,
journal_batch,
artifact_soft_limit_bytes,
artifact_hard_limit_bytes,
products: request.products.clone(),
view_count: plan.view_count(),
};
let mut views = plan
.execution_order()
.map(|view| durable_view(&header, view))
.collect::<RailResult<Vec<_>>>()?;
let mut sequence = 0;
let mut groups = Vec::new();
let mut revalidation_required = BTreeSet::new();
let mut resumed = false;
if let Some(resume_path) = request.resume_manifest.as_deref() {
let previous = read_resume_journal(workspace_root, resume_path, Some(&header), Some(&views))?;
resumed = true;
sequence = previous.sequence;
groups = previous.groups;
for (position, (view, old)) in views.iter_mut().zip(previous.views).enumerate() {
view.attempts = old.attempts;
view.cargo_targets = old.cargo_targets;
match old.durable {
DurableViewState::Complete { evidence } => {
view.durable = DurableViewState::Complete { evidence };
revalidation_required.insert(position);
}
DurableViewState::Running { .. } | DurableViewState::Pending | DurableViewState::Failed { .. } => {}
}
}
}
let document = AcquisitionJournal {
header,
views,
groups,
sequence,
summary: None,
};
if !resumed {
document.validate()?;
}
let mut journal = Self {
path,
document,
dirty_views: BTreeSet::new(),
pending_completions: 0,
pending_since: None,
revalidation_required,
};
if resumed {
journal.dirty_views.extend(0..journal.document.views.len());
journal.flush()?;
} else {
journal.persist_initial()?;
}
progress!(" Surface acquisition manifest: {display_path}");
Ok(journal)
}
pub(crate) fn evidence_identity(
&self,
index: ViewIx,
diagnostic_keys: &[CompilerDiagKey],
fact_key: Option<&CompilerFactCacheKey>,
) -> RailResult<EvidenceIdentity> {
let view = self.view(index)?;
let bytes = serde_json::to_vec(&(&view.view_identity, diagnostic_keys, fact_key))?;
Ok(EvidenceIdentity::new(&bytes))
}
pub(crate) fn revalidate(&mut self, index: ViewIx, evidence: Option<EvidenceIdentity>) -> RailResult<()> {
let position = self.position(index)?;
if let Some(evidence) = evidence {
evidence.validate()?;
self.document.views[position].durable = DurableViewState::Complete { evidence };
} else {
self.document.views[position].durable = DurableViewState::Pending;
self.document.views[position].cargo_targets.clear();
}
self.revalidation_required.remove(&position);
self.dirty_views.insert(position);
Ok(())
}
pub(crate) fn seal_revalidation(&mut self) -> RailResult<()> {
if !self.revalidation_required.is_empty() {
return Err(RailError::message(
"compiler acquisition journal recovery was not revalidated against every completed view",
));
}
self.flush()
}
pub(crate) fn running_batch(&mut self, indices: &[ViewIx]) -> RailResult<()> {
if indices.is_empty() {
return Err(RailError::message(
"compiler acquisition journal running batch is empty",
));
}
let positions = indices
.iter()
.map(|index| self.position(*index))
.collect::<RailResult<BTreeSet<_>>>()?;
if positions.len() != indices.len()
|| positions.iter().any(|position| {
!matches!(
self.document.views[*position].durable,
DurableViewState::Pending | DurableViewState::Failed { .. }
)
})
{
return Err(RailError::message(
"compiler acquisition journal can start only unique unresolved work",
));
}
for position in positions {
let view = &mut self.document.views[position];
view.attempts = view
.attempts
.checked_add(1)
.ok_or_else(|| RailError::message("compiler acquisition journal attempt count overflowed"))?;
view.durable = DurableViewState::Running { attempt: view.attempts };
self.dirty_views.insert(position);
}
self.flush()
}
pub(crate) fn complete(&mut self, index: ViewIx, evidence: EvidenceIdentity, durable: bool) -> RailResult<()> {
evidence.validate()?;
let position = self.position(index)?;
let view = &mut self.document.views[position];
if !matches!(view.durable, DurableViewState::Running { .. }) {
return Err(RailError::message(
"compiler acquisition journal completed a view that was not running",
));
}
view.durable = if durable {
DurableViewState::Complete { evidence }
} else {
DurableViewState::Pending
};
self.dirty_views.insert(position);
self.pending_completions = self.pending_completions.saturating_add(1);
self.pending_since.get_or_insert_with(Instant::now);
if self.pending_completions >= self.document.header.journal_batch {
self.flush()?;
}
Ok(())
}
pub(crate) fn flush_if_due(&mut self) -> RailResult<()> {
if self
.pending_since
.is_some_and(|started| started.elapsed() >= COMPLETION_FLUSH_INTERVAL)
{
self.flush()?;
}
Ok(())
}
pub(crate) fn completion_flush_timeout(&self) -> Option<Duration> {
self.pending_since.map(|started| {
COMPLETION_FLUSH_INTERVAL
.checked_sub(started.elapsed())
.unwrap_or(Duration::ZERO)
})
}
pub(crate) fn fail(
&mut self,
primary: Option<(ViewIx, Vec<CompilerAcquisitionCargoTarget>, FailureClass)>,
terminal_class: FailureClass,
) -> RailResult<()> {
if let Some((index, cargo_targets, class)) = primary {
let position = self.position(index)?;
let view = &mut self.document.views[position];
if !matches!(view.durable, DurableViewState::Complete { .. }) {
view.durable = DurableViewState::Failed { class };
view.cargo_targets = cargo_targets;
self.dirty_views.insert(position);
}
}
for (position, view) in self.document.views.iter_mut().enumerate() {
if matches!(view.durable, DurableViewState::Running { .. }) {
view.durable = DurableViewState::Failed {
class: FailureClass::Cancelled,
};
self.dirty_views.insert(position);
}
}
self.document.summary = Some(self.summary("partial", Some(terminal_class)));
self.flush()
}
pub(crate) fn finish(&mut self) -> RailResult<()> {
if self
.document
.views
.iter()
.any(|view| matches!(view.durable, DurableViewState::Running { .. }))
{
return Err(RailError::message(
"compiler acquisition journal cannot finish with a running view",
));
}
self.document.summary = Some(self.summary("complete", None));
self.flush()
}
pub(crate) fn resume_help(&self) -> String {
format!(
"resume with '{}' after correcting the source failure; Cargo-Rail will recapture the workspace and revalidate every exact complete evidence object",
self.document.header.resume_command.join(" ")
)
}
pub(crate) fn path(&self) -> &Path {
&self.path
}
fn summary(&self, state: &str, failure: Option<FailureClass>) -> DurableSummary {
let count = |predicate: fn(&DurableViewState) -> bool| {
self.document
.views
.iter()
.filter(|view| predicate(&view.durable))
.count()
};
DurableSummary {
state: state.to_string(),
failure,
pending: count(|state| matches!(state, DurableViewState::Pending)),
running: count(|state| matches!(state, DurableViewState::Running { .. })),
completed: count(|state| matches!(state, DurableViewState::Complete { .. })),
failed: count(|state| matches!(state, DurableViewState::Failed { .. })),
}
}
fn position(&self, index: ViewIx) -> RailResult<usize> {
self.document
.views
.iter()
.position(|view| view.view_index == index.offset())
.ok_or_else(|| RailError::message("Surface acquisition journal view identity disappeared"))
}
fn view(&self, index: ViewIx) -> RailResult<&DurableViewRecord> {
let position = self.position(index)?;
self.document
.views
.get(position)
.ok_or_else(|| RailError::message("Surface acquisition journal view is out of bounds"))
}
fn persist_initial(&mut self) -> RailResult<()> {
let encoded = serde_json::to_vec_pretty(&self.document)?;
self.persist_bytes(&encoded)
}
fn flush(&mut self) -> RailResult<()> {
if self.dirty_views.is_empty() && self.document.summary.is_none() {
return Ok(());
}
let mut next = self.document.clone();
next.sequence = next
.sequence
.checked_add(1)
.ok_or_else(|| RailError::message("compiler acquisition journal sequence overflowed"))?;
next.groups.push(DurableGroupState {
sequence: next.sequence,
views: self.dirty_views.iter().copied().collect(),
terminal: next.summary.is_some(),
});
if next.groups.len() > MAX_JOURNAL_GROUPS {
let excess = next.groups.len() - MAX_JOURNAL_GROUPS;
next.groups.drain(..excess);
}
next.validate()?;
let encoded = serde_json::to_vec_pretty(&next)?;
self.persist_bytes(&encoded)?;
self.document = next;
self.dirty_views.clear();
self.pending_completions = 0;
self.pending_since = None;
Ok(())
}
fn persist_bytes(&mut self, encoded: &[u8]) -> RailResult<()> {
if encoded.len() as u64 > MAX_JOURNAL_BYTES {
return Err(RailError::message(
"Surface acquisition manifest exceeds its byte bound",
));
}
let started = crate::instrumentation::compiler_acquisition_timer();
crate::utils::write_file_atomic(&self.path, encoded)?;
crate::instrumentation::record_compiler_acquisition_journal_write(started, encoded.len(), true, true);
Ok(())
}
}
impl AcquisitionJournal {
fn validate(&self) -> RailResult<()> {
let header = &self.header;
if header.record != "manifest"
|| header.surface_acquisition_contract_version != ACQUISITION_CONTRACT_VERSION
|| header.view_count != self.views.len()
|| header.journal_batch == 0
{
return Err(RailError::message(
"Surface acquisition manifest has an invalid current contract",
));
}
validate_sha256_identity(&header.acquisition_identity, ACQUISITION_IDENTITY_PREFIX, "acquisition")?;
let mut identities = BTreeSet::new();
let mut indices = BTreeSet::new();
for view in &self.views {
let running_attempt = match view.durable {
DurableViewState::Running { attempt } => Some(attempt),
_ => None,
};
if !identities.insert(view.view_identity.as_str())
|| !indices.insert(view.view_index)
|| view.view_index >= self.views.len()
|| view.packages[0].is_empty()
|| running_attempt.is_some_and(|attempt| attempt == 0 || view.attempts != attempt)
{
return Err(RailError::message(
"Surface acquisition manifest has invalid, duplicate, or inconsistent view state",
));
}
if let DurableViewState::Complete { evidence } = &view.durable {
evidence.validate()?;
}
}
let mut previous = None;
for group in &self.groups {
if group.sequence == 0
|| group.sequence > self.sequence
|| previous.is_some_and(|prior| prior >= group.sequence)
|| group.views.windows(2).any(|pair| pair[0] >= pair[1])
|| group.views.iter().any(|ordinal| *ordinal >= self.views.len())
{
return Err(RailError::message(
"Surface acquisition manifest has invalid durability-group ordering",
));
}
previous = Some(group.sequence);
}
if self.groups.last().map_or(self.sequence != 0, |group| {
group.sequence != self.sequence || group.terminal != self.summary.is_some()
}) {
return Err(RailError::message(
"Surface acquisition manifest has an inconsistent durability sequence",
));
}
if let Some(summary) = &self.summary {
let counts = self.views.iter().fold([0_usize; 4], |mut counts, view| {
let offset = match view.durable {
DurableViewState::Pending => 0,
DurableViewState::Running { .. } => 1,
DurableViewState::Complete { .. } => 2,
DurableViewState::Failed { .. } => 3,
};
counts[offset] += 1;
counts
});
let counts_match = counts == [summary.pending, summary.running, summary.completed, summary.failed];
let terminal_state_matches = match (summary.state.as_str(), summary.failure) {
("complete", None) => counts[1] == 0 && counts[3] == 0,
("partial", Some(_)) => counts[1] == 0,
_ => false,
};
if !counts_match || !terminal_state_matches {
return Err(RailError::message(
"Surface acquisition manifest has an inconsistent terminal summary",
));
}
}
Ok(())
}
}
fn durable_view(
header: &CompilerAcquisitionHeader,
view: CompilerAcquisitionView<'_>,
) -> RailResult<DurableViewRecord> {
let view_bytes = serde_json::to_vec(&(
view.command_class(),
PlatformTarget::from(view.platform()),
view.features(),
BTreeSet::from([view.package()]),
view.fact_families(),
))?;
Ok(DurableViewRecord {
view_index: view.index().offset(),
view_identity: format!("compiler-view-v1-sha256-{}", ContentDigest::sha256(&view_bytes)),
selected_products: header
.products
.iter()
.filter(|product| product.package == view.package())
.cloned()
.collect(),
packages: [view.package().to_string()],
target_triple: view.platform().to_string(),
feature_profile: view.features().label(),
command_class: view.command_class().to_string(),
cargo_targets: Vec::new(),
attempts: 0,
durable: DurableViewState::Pending,
})
}
fn display_path(workspace_root: &Path, path: &Path) -> String {
path.strip_prefix(workspace_root)
.unwrap_or(path)
.to_string_lossy()
.replace('\\', "/")
}
fn validate_sha256_identity(value: &str, prefix: &str, label: &str) -> RailResult<()> {
let digest = value
.strip_prefix(prefix)
.ok_or_else(|| RailError::message(format!("Surface {label} identity has the wrong domain or version")))?;
if digest.len() != 64
|| !digest
.bytes()
.all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase())
{
return Err(RailError::message(format!(
"Surface {label} identity is not canonical SHA-256"
)));
}
Ok(())
}
fn resolve_resume_path(workspace_root: &Path, path: &Path) -> RailResult<PathBuf> {
let candidate = if path.is_absolute() {
path.to_path_buf()
} else {
workspace_root.join(path)
};
let metadata = fs::symlink_metadata(&candidate).map_err(|error| {
RailError::message(format!(
"failed to inspect Surface acquisition manifest '{}': {error}",
candidate.display()
))
})?;
if !metadata.is_file() || crate::utils::is_symlink_or_reparse(&metadata) || metadata.len() > MAX_JOURNAL_BYTES {
return Err(RailError::message(format!(
"Surface acquisition manifest '{}' is not a bounded real regular file",
candidate.display()
)));
}
let candidate = crate::utils::canonicalize_existing(&candidate)?;
let state_root = crate::workspace::cargo_rail_state_root(workspace_root);
let authorized = crate::utils::canonicalize_existing(&state_root.join(JOURNAL_DIRECTORY))
.is_ok_and(|directory| candidate.starts_with(directory));
if !authorized {
return Err(RailError::with_help(
format!(
"Surface acquisition manifest '{}' is outside the current workspace's journal authority",
candidate.display()
),
"preserve this manifest; run 'cargo rail surface' without --resume to create current acquisition evidence",
));
}
Ok(candidate)
}
fn read_bounded(path: &Path) -> RailResult<Vec<u8>> {
let mut file = fs::File::open(path).map_err(RailError::from)?;
let expected_len = file.metadata().map_err(RailError::from)?.len();
if expected_len > MAX_JOURNAL_BYTES {
return Err(RailError::message(
"Surface acquisition manifest exceeds its byte bound",
));
}
let mut bytes = Vec::new();
(&mut file)
.take(MAX_JOURNAL_BYTES + 1)
.read_to_end(&mut bytes)
.map_err(RailError::from)?;
if bytes.len() as u64 > MAX_JOURNAL_BYTES {
return Err(RailError::message(
"Surface acquisition manifest exceeds its byte bound",
));
}
if !crate::utils::private_file_matches_path(&file, path, expected_len).map_err(RailError::from)? {
return Err(RailError::message(
"Surface acquisition manifest changed identity while it was read",
));
}
Ok(bytes)
}
fn read_resume_journal(
workspace_root: &Path,
path: &Path,
expected_header: Option<&CompilerAcquisitionHeader>,
expected_views: Option<&[DurableViewRecord]>,
) -> RailResult<AcquisitionJournal> {
let path = resolve_resume_path(workspace_root, path)?;
let bytes = read_bounded(&path)?;
let document: AcquisitionJournal = serde_json::from_slice(&bytes).map_err(|error| {
RailError::with_help(
format!(
"unsupported or malformed Surface acquisition manifest '{}': {error}",
path.display()
),
"preserve this manifest; run 'cargo rail surface' without --resume to create current acquisition evidence",
)
})?;
if document.header.surface_acquisition_contract_version != ACQUISITION_CONTRACT_VERSION {
return Err(RailError::with_help(
format!(
"unsupported Surface acquisition manifest version {} in '{}'",
document.header.surface_acquisition_contract_version,
path.display()
),
"preserve this manifest; run 'cargo rail surface' without --resume to create current acquisition evidence",
));
}
document.validate()?;
if let Some(expected) = expected_header {
validate_binding(&document.header, expected)?;
}
if let Some(expected) = expected_views
&& (document.views.len() != expected.len()
|| document.views.iter().zip(expected).any(|(actual, expected)| {
actual.view_index != expected.view_index || actual.view_identity != expected.view_identity
}))
{
return Err(RailError::message(
"Surface acquisition resume view catalog differs from the current plan",
));
}
Ok(document)
}
fn validate_binding(actual: &CompilerAcquisitionHeader, expected: &CompilerAcquisitionHeader) -> RailResult<()> {
if actual.workspace_identity != expected.workspace_identity
|| actual.checkout_identity != expected.checkout_identity
|| actual.snapshot_identity != expected.snapshot_identity
|| actual.configuration_fingerprint != expected.configuration_fingerprint
|| actual.plan_identity != expected.plan_identity
|| actual.candidate_set_identity != expected.candidate_set_identity
|| actual.compiler_set_identity != expected.compiler_set_identity
|| actual.schemas != expected.schemas
|| actual.acquisition_identity != expected.acquisition_identity
{
return Err(RailError::with_help(
"Surface acquisition resume authority differs from the current workspace, plan, compiler set, or schema",
"rerun 'cargo rail surface' without --resume to plan the changed acquisition",
));
}
Ok(())
}
pub(crate) fn validate_compiler_acquisition_resume(
workspace_root: &Path,
path: &Path,
configuration_fingerprint: &str,
) -> RailResult<()> {
let document = read_resume_journal(workspace_root, path, None, None)?;
if document.header.configuration_fingerprint != configuration_fingerprint {
return Err(RailError::with_help(
"Surface acquisition resume configuration differs from the planned acquisition",
"rerun 'cargo rail surface' without --resume to plan the changed configuration",
));
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn journal_plan(packages: &[&str]) -> RailResult<CompilerAcquisitionPlan> {
let manifests = packages
.iter()
.map(|name| crate::cargo::manifest_analyzer::ParsedManifest {
package_id: cargo_metadata::PackageId {
repr: format!("path+file:///fixture/{name}#{name}@0.1.0"),
},
path: PathBuf::from(format!("/fixture/{name}/Cargo.toml")),
package_name: (*name).to_string(),
declared_features: Default::default(),
required_feature_selections: Default::default(),
condition_feature_selections: Default::default(),
source_cfg_features: Default::default(),
source_cfg_expressions: Default::default(),
retention_identifiers: Default::default(),
development_only_identifiers: Default::default(),
inherits_workspace_msrv: false,
package_fields: Default::default(),
dependencies: Default::default(),
})
.collect::<Vec<_>>();
let typed = packages.iter().map(|name| (*name).to_string()).collect();
let schedule = crate::compiler::scheduler::AnalysisSchedule::for_combined_with_features(
&manifests,
&["default"],
&[],
&typed,
&Default::default(),
None,
)?;
CompilerAcquisitionPlan::from_schedule(&schedule, &[], &["default"])
}
fn request(resume_manifest: Option<PathBuf>) -> CompilerAcquisitionRequest {
CompilerAcquisitionRequest {
workspace_identity: "workspace-v1".to_string(),
checkout_identity: "checkout-v1".to_string(),
snapshot_identity: "snapshot-v1".to_string(),
configuration_fingerprint: "configuration-v1".to_string(),
products: Vec::new(),
resume_manifest,
}
}
fn begin(
root: &Path,
plan: &CompilerAcquisitionPlan,
resume_manifest: Option<PathBuf>,
batch: usize,
) -> CompilerAcquisitionJournal {
fs::create_dir_all(crate::workspace::cargo_rail_state_root(root)).expect("state root");
CompilerAcquisitionJournal::begin(
root,
&request(resume_manifest),
plan,
"compiler-set-v1".to_string(),
1024,
2048,
plan.view_count(),
batch,
)
.expect("journal")
}
fn read_document(path: &Path) -> AcquisitionJournal {
serde_json::from_slice(&fs::read(path).expect("journal bytes")).expect("journal document")
}
fn state(document: &AcquisitionJournal, index: ViewIx) -> &DurableViewState {
&document
.views
.iter()
.find(|view| view.view_index == index.offset())
.expect("view")
.durable
}
fn evidence(journal: &CompilerAcquisitionJournal, index: ViewIx) -> EvidenceIdentity {
journal.evidence_identity(index, &[], None).expect("evidence identity")
}
#[test]
fn unsupported_resume_preserves_progress_and_fresh_acquisition_uses_current_namespace() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha"]).expect("plan");
let journal = begin(root.path(), &plan, None, 1);
let old_directory = crate::workspace::cargo_rail_state_root(root.path()).join("surface-acquisitions-v2");
fs::create_dir_all(&old_directory).expect("old directory");
let old_path = old_directory.join("progress.json");
let old_bytes = b"{\"surface_acquisition_contract_version\":2}";
fs::write(&old_path, old_bytes).expect("old journal");
let error = validate_compiler_acquisition_resume(root.path(), &old_path, "configuration-v1").unwrap_err();
assert!(error.to_string().contains("journal authority"), "{error}");
assert_eq!(fs::read(&old_path).unwrap(), old_bytes);
let fresh = begin(root.path(), &plan, None, 1);
assert_eq!(fresh.document.header.surface_acquisition_contract_version, 3);
assert_eq!(fs::read(&old_path).unwrap(), old_bytes);
let mut invalid = serde_json::to_value(&journal.document).unwrap();
invalid["header"]["surface_acquisition_contract_version"] = serde_json::json!(2);
let invalid = serde_json::to_vec(&invalid).unwrap();
fs::write(&journal.path, &invalid).unwrap();
assert!(validate_compiler_acquisition_resume(root.path(), &journal.path, "configuration-v1").is_err());
assert_eq!(fs::read(&journal.path).unwrap(), invalid);
}
#[test]
fn out_of_order_completions_commit_as_one_canonical_group() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha", "beta", "gamma"]).expect("plan");
let order = plan
.execution_order()
.map(CompilerAcquisitionView::index)
.collect::<Vec<_>>();
let first = order[0];
let last = order[2];
let mut journal = begin(root.path(), &plan, None, 2);
journal.running_batch(&[last, first]).expect("start batch");
journal
.complete(last, evidence(&journal, last), true)
.expect("buffer last completion");
let before_group = read_document(journal.path());
assert!(matches!(state(&before_group, last), DurableViewState::Running { .. }));
journal
.complete(first, evidence(&journal, first), true)
.expect("commit completion group");
let after_group = read_document(journal.path());
assert!(matches!(state(&after_group, first), DurableViewState::Complete { .. }));
assert!(matches!(state(&after_group, last), DurableViewState::Complete { .. }));
assert_eq!(after_group.sequence, 2, "one running batch and one completion batch");
assert_eq!(after_group.groups.len(), 2);
let group = after_group.groups.last().expect("completion group");
assert_eq!(group.views.len(), 2);
assert!(group.views.windows(2).all(|pair| pair[0] < pair[1]));
}
#[test]
fn forced_termination_recovers_complete_views_and_resets_running_views() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha", "beta", "gamma"]).expect("plan");
let order = plan
.execution_order()
.map(CompilerAcquisitionView::index)
.collect::<Vec<_>>();
let complete = order[0];
let interrupted = order[1];
let path = {
let mut journal = begin(root.path(), &plan, None, 1);
journal.running_batch(&[complete]).expect("start complete view");
journal
.complete(complete, evidence(&journal, complete), true)
.expect("complete view");
journal.running_batch(&[interrupted]).expect("start interrupted view");
journal.path().to_path_buf()
};
let mut resumed = begin(root.path(), &plan, Some(path), 1);
assert!(matches!(
resumed.view(complete).expect("complete state").durable,
DurableViewState::Complete { .. }
));
assert!(matches!(
resumed.view(interrupted).expect("interrupted state").durable,
DurableViewState::Pending
));
assert!(resumed.seal_revalidation().is_err());
resumed
.revalidate(complete, Some(evidence(&resumed, complete)))
.expect("revalidate complete evidence");
resumed.seal_revalidation().expect("seal recovery");
}
#[test]
fn failed_replacement_preserves_running_state_and_completed_bytes_resume() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha"]).expect("plan");
let view = plan.execution_order().next().expect("view").index();
let mut journal = begin(root.path(), &plan, None, 1);
journal.running_batch(&[view]).expect("running");
let path = journal.path().to_path_buf();
let saved = path.with_extension("saved");
fs::rename(&path, &saved).unwrap();
fs::create_dir(&path).unwrap();
let error = journal.complete(view, evidence(&journal, view), true).unwrap_err();
assert!(
matches!(state(&read_document(&saved), view), DurableViewState::Running { .. }),
"{error}"
);
fs::remove_dir(&path).unwrap();
fs::rename(&saved, &path).unwrap();
journal.flush().expect("retry real replacement");
assert!(matches!(
state(&read_document(&path), view),
DurableViewState::Complete { .. }
));
drop(journal);
let resumed = begin(root.path(), &plan, Some(path), 1);
assert!(matches!(
state(&read_document(resumed.path()), view),
DurableViewState::Complete { .. }
));
}
#[test]
fn duration_bound_flushes_a_partial_completion_group() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha"]).expect("plan");
let view = plan.execution_order().next().expect("view").index();
let mut journal = begin(root.path(), &plan, None, 8);
journal.running_batch(&[view]).expect("running");
journal
.complete(view, evidence(&journal, view), true)
.expect("buffer completion");
journal.pending_since = Some(
Instant::now()
.checked_sub(COMPLETION_FLUSH_INTERVAL)
.expect("test duration is representable"),
);
journal.flush_if_due().expect("duration flush");
assert!(matches!(
state(&read_document(journal.path()), view),
DurableViewState::Complete { .. }
));
}
#[test]
fn fatal_transition_flushes_completed_progress_and_terminal_failure_class() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha", "beta"]).expect("plan");
let views = plan
.execution_order()
.map(CompilerAcquisitionView::index)
.collect::<Vec<_>>();
let mut journal = begin(root.path(), &plan, None, 8);
journal.running_batch(&views).expect("running batch");
journal
.complete(views[0], evidence(&journal, views[0]), true)
.expect("buffer completed progress");
journal
.fail(Some((views[1], Vec::new(), FailureClass::Worker)), FailureClass::Worker)
.expect("terminal journal");
let terminal = read_document(journal.path());
assert!(matches!(state(&terminal, views[0]), DurableViewState::Complete { .. }));
assert!(matches!(
state(&terminal, views[1]),
DurableViewState::Failed {
class: FailureClass::Worker
}
));
assert_eq!(
terminal.summary.as_ref().and_then(|summary| summary.failure),
Some(FailureClass::Worker)
);
assert_eq!(terminal.summary.as_ref().map(|summary| summary.running), Some(0));
assert!(terminal.groups.last().is_some_and(|group| group.terminal));
}
#[test]
fn terminal_partial_journal_can_enter_a_new_resume_sequence() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha", "beta"]).expect("plan");
let views = plan
.execution_order()
.map(CompilerAcquisitionView::index)
.collect::<Vec<_>>();
let path = {
let mut journal = begin(root.path(), &plan, None, 8);
journal.running_batch(&views).expect("running batch");
journal
.complete(views[0], evidence(&journal, views[0]), true)
.expect("buffer completed progress");
journal
.fail(
Some((views[1], Vec::new(), FailureClass::Coordinator)),
FailureClass::Coordinator,
)
.expect("terminal partial journal");
journal.path().to_path_buf()
};
let resumed = begin(root.path(), &plan, Some(path), 8);
let active = read_document(resumed.path());
assert!(active.summary.is_none());
assert!(active.groups.last().is_some_and(|group| !group.terminal));
assert!(active.groups.iter().any(|group| group.terminal));
assert!(matches!(state(&active, views[0]), DurableViewState::Complete { .. }));
assert!(matches!(state(&active, views[1]), DurableViewState::Pending));
}
#[test]
fn successful_non_durable_view_finishes_but_remains_pending_for_resume() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha"]).expect("plan");
let view = plan.execution_order().next().expect("view").index();
let path = {
let mut journal = begin(root.path(), &plan, None, 1);
journal.running_batch(&[view]).expect("running");
journal
.complete(view, evidence(&journal, view), false)
.expect("non-durable completion");
journal.finish().expect("successful terminal journal");
let terminal = read_document(journal.path());
assert!(matches!(state(&terminal, view), DurableViewState::Pending));
assert_eq!(
terminal.summary.as_ref().map(|summary| summary.state.as_str()),
Some("complete")
);
assert_eq!(terminal.summary.as_ref().map(|summary| summary.pending), Some(1));
journal.path().to_path_buf()
};
let resumed = begin(root.path(), &plan, Some(path), 1);
assert!(matches!(
resumed.view(view).expect("view").durable,
DurableViewState::Pending
));
}
#[test]
fn binding_and_group_corruption_fail_closed() {
let root = tempfile::tempdir().expect("root");
let plan = journal_plan(&["alpha"]).expect("plan");
let journal = begin(root.path(), &plan, None, 1);
let mut mismatched = journal.document.header.clone();
mismatched.compiler_set_identity.push_str("-changed");
assert!(validate_binding(&mismatched, &journal.document.header).is_err());
let mut corrupted = journal.document;
corrupted.sequence = 1;
corrupted.groups.push(DurableGroupState {
sequence: 1,
views: vec![1],
terminal: false,
});
assert!(corrupted.validate().is_err());
corrupted.groups.clear();
corrupted.groups.push(DurableGroupState {
sequence: 1,
views: Vec::new(),
terminal: true,
});
corrupted.summary = Some(DurableSummary {
state: "complete".to_string(),
failure: None,
pending: 0,
running: 0,
completed: 1,
failed: 0,
});
assert!(corrupted.validate().is_err());
}
}