use super::{
AttemptJournalError, AttemptJournalGuard, BackupLayoutGuard, ExecutionProgressPersistenceError,
ExecutionSettlementCheckpointError, ExecutionSettlementPersistenceError, JournalLock,
JournalLockError, OperationPlanPersistenceError, PersistenceError, read_execution_progress,
read_execution_settlement, read_json, read_operation_plan, reconcile_execution_settlement,
replay_execution_settlement,
};
use crate::model::{
artifacts::ArtifactChecksumRecord,
execution_workflow::{
ExecutionStageBindingRecord, ExecutionStagePredecessorRecord, ExecutionWorkflowError,
ExecutionWorkflowRecord, MAX_EXECUTION_STAGE_BYTES, MAX_EXECUTION_WORKFLOW_BYTES,
},
operation_plan::{OperationPlanError, OperationPlanRecord},
};
use crate::policy::execution_progress::ExecutionProgressView;
use std::{
collections::{BTreeMap, BTreeSet, btree_map::Entry},
fs,
path::PathBuf,
};
use thiserror::Error;
const WORKFLOW_FILE: &str = "execution-workflow.json";
const BINDING_FILE: &str = "stage-binding.json";
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ExecutionWorkflowSettlementView {
workflow: ArtifactChecksumRecord,
stages: Vec<ExecutionStageSettlementView>,
}
impl ExecutionWorkflowSettlementView {
#[must_use]
pub const fn workflow(&self) -> &ArtifactChecksumRecord {
&self.workflow
}
#[must_use]
pub fn stages(&self) -> &[ExecutionStageSettlementView] {
&self.stages
}
}
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct ExecutionStageSettlementView {
stage_sequence: u64,
binding: ArtifactChecksumRecord,
settlement: ArtifactChecksumRecord,
}
impl ExecutionStageSettlementView {
#[must_use]
pub const fn stage_sequence(&self) -> u64 {
self.stage_sequence
}
#[must_use]
pub const fn binding(&self) -> &ArtifactChecksumRecord {
&self.binding
}
#[must_use]
pub const fn settlement(&self) -> &ArtifactChecksumRecord {
&self.settlement
}
}
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum StagePreparationBarrier {
Plan,
Binding,
Journal(u64),
}
pub fn create_execution_workflow(
layout: &BackupLayoutGuard,
record: &ExecutionWorkflowRecord,
) -> Result<(), ExecutionWorkflowPersistenceError> {
layout.check_root()?;
let bytes = super::json::encode_json(record, MAX_EXECUTION_WORKFLOW_BYTES)?;
super::immutable_record::create_encoded_record(layout, WORKFLOW_FILE, &bytes)
}
pub fn read_execution_workflow(
layout: &BackupLayoutGuard,
expected: &ArtifactChecksumRecord,
) -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError> {
super::immutable_record::read(
layout,
WORKFLOW_FILE,
MAX_EXECUTION_WORKFLOW_BYTES,
|record: &ExecutionWorkflowRecord| &record.digest() == expected,
|| ExecutionWorkflowPersistenceError::DigestMismatch,
)
}
pub(crate) fn read_execution_stage_occupancy(
layout: &BackupLayoutGuard,
expected: &ArtifactChecksumRecord,
) -> Result<BTreeSet<u64>, ExecutionWorkflowPersistenceError> {
let workflow = read_execution_workflow(layout, expected)?;
let mut occupied = BTreeSet::new();
for stage in workflow.stages() {
let sequence = stage.operation_sequence();
match fs::symlink_metadata(stage_path(layout, sequence)) {
Ok(metadata) if metadata.is_dir() => {
occupied.insert(sequence);
}
Ok(_) => return Err(ExecutionWorkflowPersistenceError::UnsafeStage),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(error) => return Err(PersistenceError::from(error).into()),
}
}
read_execution_workflow(layout, expected)?;
Ok(occupied)
}
pub fn read_complete_execution_workflow(
layout: &BackupLayoutGuard,
expected: &ArtifactChecksumRecord,
) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
read_complete_with(layout, expected, || {
read_execution_workflow(layout, expected)
})
}
pub fn read_complete_execution_workflow_locked(
layout: &BackupLayoutGuard,
journal: &JournalLock,
expected: &ArtifactChecksumRecord,
) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
if journal.journal() != layout.root().join(WORKFLOW_FILE) {
return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
}
read_complete_with(layout, expected, || {
layout.check_root()?;
let workflow: ExecutionWorkflowRecord =
read_json(journal.journal(), MAX_EXECUTION_WORKFLOW_BYTES)?;
super::json::check_json_size(&workflow, MAX_EXECUTION_WORKFLOW_BYTES)?;
if &workflow.digest() != expected {
return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
}
Ok(workflow)
})
}
fn read_complete_with(
layout: &BackupLayoutGuard,
expected: &ArtifactChecksumRecord,
read_original: impl Fn() -> Result<ExecutionWorkflowRecord, ExecutionWorkflowPersistenceError>,
) -> Result<ExecutionWorkflowSettlementView, ExecutionStageCheckpointError> {
let workflow = read_original()?;
let mut stages = Vec::with_capacity(workflow.stages().len());
for allocated in workflow.stages() {
let sequence = allocated.operation_sequence();
let stage = acquire_stage(layout, sequence)?;
let binding: ExecutionStageBindingRecord =
read_json(&stage.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)
.map_err(ExecutionWorkflowPersistenceError::from)?;
let (binding, plan) = read_binding(&stage, &workflow, sequence, &binding.digest())?;
let settlement = replay_execution_settlement(&stage, &plan.digest())?;
read_binding(&stage, &workflow, sequence, &binding.digest())?;
let view = ExecutionStageSettlementView {
stage_sequence: sequence,
binding: binding.digest(),
settlement: settlement.digest(),
};
stages.push((binding, view));
}
for (binding, _) in &stages {
for predecessor in binding.predecessors() {
let index = stages
.binary_search_by_key(&predecessor.stage_sequence(), |(_, view)| {
view.stage_sequence()
})
.map_err(|_| ExecutionWorkflowPersistenceError::DigestMismatch)?;
let actual = &stages[index].1;
if predecessor.binding() != actual.binding()
|| predecessor.settlement() != actual.settlement()
{
return Err(ExecutionWorkflowPersistenceError::DigestMismatch.into());
}
}
}
read_original()?;
Ok(ExecutionWorkflowSettlementView {
workflow: expected.clone(),
stages: stages.into_iter().map(|(_, view)| view).collect(),
})
}
#[derive(Debug)]
pub struct ExecutionStageGuard<'a> {
workflow_layout: &'a BackupLayoutGuard,
stage_layout: BackupLayoutGuard,
binding: ExecutionStageBindingRecord,
plan: OperationPlanRecord,
}
impl<'a> ExecutionStageGuard<'a> {
pub fn reconcile_checkpoint(
&self,
learned_evidence: ArtifactChecksumRecord,
) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
self.reconcile_checkpoint_with(learned_evidence, || {})
}
fn reconcile_checkpoint_with(
&self,
learned_evidence: ArtifactChecksumRecord,
after_reconciliation: impl FnOnce(),
) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
let settlement = reconcile_execution_settlement(self.layout()?, &self.plan.digest())?;
after_reconciliation();
self.layout()?;
Ok(ExecutionStagePredecessorRecord::new(
self.binding.stage_sequence(),
self.binding.digest(),
settlement.digest(),
learned_evidence,
))
}
pub fn replay_checkpoint(
&self,
learned_evidence: ArtifactChecksumRecord,
) -> Result<ExecutionStagePredecessorRecord, ExecutionStageCheckpointError> {
let settlement = replay_execution_settlement(self.layout()?, &self.plan.digest())?;
self.layout()?;
Ok(ExecutionStagePredecessorRecord::new(
self.binding.stage_sequence(),
self.binding.digest(),
settlement.digest(),
learned_evidence,
))
}
pub fn resume(
workflow_layout: &'a BackupLayoutGuard,
expected_workflow: &ArtifactChecksumRecord,
sequence: u64,
expected_binding: &ArtifactChecksumRecord,
) -> Result<(Self, ExecutionProgressView), ExecutionStageResumeError> {
let stage = Self::open(
workflow_layout,
expected_workflow,
sequence,
expected_binding,
)?;
let progress = read_execution_progress(&stage.stage_layout, &stage.plan.digest())?;
stage.layout()?;
Ok((stage, progress))
}
pub fn prepare(
workflow_layout: &'a BackupLayoutGuard,
binding: ExecutionStageBindingRecord,
plan: OperationPlanRecord,
) -> Result<Self, ExecutionStagePreparationError> {
Self::prepare_with(workflow_layout, binding, plan, |_| {})
}
fn prepare_with(
workflow_layout: &'a BackupLayoutGuard,
binding: ExecutionStageBindingRecord,
plan: OperationPlanRecord,
mut barrier: impl FnMut(StagePreparationBarrier),
) -> Result<Self, ExecutionStagePreparationError> {
let authorities = plan.attempt_authorities()?;
let stage = Self::create_with(workflow_layout, binding, plan, &mut barrier)?;
let layout = stage.layout()?;
for authority in authorities {
let sequence = authority.binding().operation_sequence();
drop(AttemptJournalGuard::create(layout, authority)?);
barrier(StagePreparationBarrier::Journal(sequence));
}
stage.layout()?;
Ok(stage)
}
fn create_with(
workflow_layout: &'a BackupLayoutGuard,
binding: ExecutionStageBindingRecord,
plan: OperationPlanRecord,
mut barrier: impl FnMut(StagePreparationBarrier),
) -> Result<Self, ExecutionWorkflowPersistenceError> {
let workflow = read_execution_workflow(workflow_layout, binding.workflow())?;
binding.validate(&workflow, &plan)?;
let binding_bytes = super::json::encode_json(&binding, MAX_EXECUTION_STAGE_BYTES)?;
let plan_bytes = super::json::encode_json(
&plan,
crate::model::operation_plan::MAX_OPERATION_PLAN_BYTES,
)?;
validate_predecessors(workflow_layout, &workflow, &binding)?;
let path = stage_path(workflow_layout, binding.stage_sequence());
ic_host_fs::durable::create_private_directory(&path)
.map_err(|error| PersistenceError::Io(crate::ops::directory_creation_io(error)))?;
let stage_layout = acquire_stage(workflow_layout, binding.stage_sequence())?;
super::operation_plan::create_encoded(&stage_layout, &plan_bytes)?;
barrier(StagePreparationBarrier::Plan);
super::json::create_encoded(&stage_layout.root().join(BINDING_FILE), &binding_bytes)?;
barrier(StagePreparationBarrier::Binding);
workflow_layout.check_root()?;
Ok(Self {
workflow_layout,
stage_layout,
binding,
plan,
})
}
pub fn open(
workflow_layout: &'a BackupLayoutGuard,
expected_workflow: &ArtifactChecksumRecord,
sequence: u64,
expected_binding: &ArtifactChecksumRecord,
) -> Result<Self, ExecutionWorkflowPersistenceError> {
let workflow = read_execution_workflow(workflow_layout, expected_workflow)?;
workflow
.stage(sequence)
.map_err(|_| ExecutionWorkflowError::UnknownStage(sequence))?;
let stage_layout = acquire_stage(workflow_layout, sequence)?;
let (binding, plan) = read_binding(&stage_layout, &workflow, sequence, expected_binding)?;
validate_predecessors(workflow_layout, &workflow, &binding)?;
workflow_layout.check_root()?;
Ok(Self {
workflow_layout,
stage_layout,
binding,
plan,
})
}
#[must_use]
pub const fn plan(&self) -> &OperationPlanRecord {
&self.plan
}
#[must_use]
pub const fn binding(&self) -> &ExecutionStageBindingRecord {
&self.binding
}
#[must_use]
pub const fn workflow_layout(&self) -> &BackupLayoutGuard {
self.workflow_layout
}
pub fn layout(&self) -> Result<&BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
let workflow = read_execution_workflow(self.workflow_layout, self.binding.workflow())?;
read_binding(
&self.stage_layout,
&workflow,
self.binding.stage_sequence(),
&self.binding.digest(),
)?;
validate_predecessors(self.workflow_layout, &workflow, &self.binding)?;
Ok(&self.stage_layout)
}
}
#[derive(Debug, Error)]
pub enum ExecutionStageCheckpointError {
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Settlement(#[from] ExecutionSettlementCheckpointError),
}
#[derive(Debug, Error)]
pub enum ExecutionStagePreparationError {
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Authority(#[from] OperationPlanError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
}
#[derive(Debug, Error)]
pub enum ExecutionStageResumeError {
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Progress(#[from] ExecutionProgressPersistenceError),
}
fn read_binding(
layout: &BackupLayoutGuard,
workflow: &ExecutionWorkflowRecord,
sequence: u64,
expected: &ArtifactChecksumRecord,
) -> Result<(ExecutionStageBindingRecord, OperationPlanRecord), ExecutionWorkflowPersistenceError> {
layout.check_root()?;
let binding: ExecutionStageBindingRecord =
read_json(&layout.root().join(BINDING_FILE), MAX_EXECUTION_STAGE_BYTES)?;
super::json::check_json_size(&binding, MAX_EXECUTION_STAGE_BYTES)?;
if binding.stage_sequence() != sequence || &binding.digest() != expected {
return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
}
let plan = read_operation_plan(layout, binding.plan())?;
binding.validate(workflow, &plan)?;
Ok((binding, plan))
}
fn validate_predecessors(
layout: &BackupLayoutGuard,
workflow: &ExecutionWorkflowRecord,
binding: &ExecutionStageBindingRecord,
) -> Result<(), ExecutionWorkflowPersistenceError> {
let mut expected: BTreeMap<_, _> = binding
.predecessors()
.iter()
.map(|row| {
(
row.stage_sequence(),
(row.binding().clone(), row.settlement().clone()),
)
})
.collect();
let mut pending: Vec<_> = expected.keys().copied().collect();
while let Some(sequence) = pending.pop() {
let (expected_binding, expected_settlement) = &expected[&sequence];
let original = {
let predecessor = acquire_stage(layout, sequence)?;
let (original, _) = read_binding(&predecessor, workflow, sequence, expected_binding)?;
read_execution_settlement(&predecessor, original.plan(), expected_settlement)?;
original
};
for ancestor in original.predecessors() {
match expected.entry(ancestor.stage_sequence()) {
Entry::Vacant(entry) => {
pending.push(ancestor.stage_sequence());
entry.insert((ancestor.binding().clone(), ancestor.settlement().clone()));
}
Entry::Occupied(entry) => {
let (expected_binding, expected_settlement) = entry.get();
if expected_binding != ancestor.binding()
|| expected_settlement != ancestor.settlement()
{
return Err(ExecutionWorkflowPersistenceError::DigestMismatch);
}
}
}
}
}
Ok(())
}
fn stage_path(layout: &BackupLayoutGuard, sequence: u64) -> PathBuf {
layout.root().join(format!("execution-stage-{sequence}"))
}
fn acquire_stage(
layout: &BackupLayoutGuard,
sequence: u64,
) -> Result<BackupLayoutGuard, ExecutionWorkflowPersistenceError> {
layout.check_root()?;
let path = stage_path(layout, sequence);
let metadata = fs::symlink_metadata(&path).map_err(PersistenceError::from)?;
if !metadata.is_dir() {
return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
}
let stage = BackupLayoutGuard::acquire(&path)?;
if stage.root() != path {
return Err(ExecutionWorkflowPersistenceError::UnsafeStage);
}
layout.check_root()?;
Ok(stage)
}
#[derive(Debug, Error)]
pub enum ExecutionWorkflowPersistenceError {
#[error("execution workflow/stage digest mismatch")]
DigestMismatch,
#[error("unsafe execution stage directory")]
UnsafeStage,
#[error(transparent)]
Model(#[from] ExecutionWorkflowError),
#[error(transparent)]
Plan(#[from] OperationPlanPersistenceError),
#[error(transparent)]
Settlement(#[from] ExecutionSettlementPersistenceError),
#[error(transparent)]
Lock(#[from] JournalLockError),
#[error(transparent)]
Persistence(#[from] PersistenceError),
}
#[cfg(all(test, unix))]
mod tests;