use crate::{
model::{
artifacts::ArtifactChecksumRecord,
attempt_journal::{ObservationOutcomeRecord, ObservationReceiptRequest},
fence_acquisition::{FenceAcquisitionError, FenceAcquisitionPayload, original_authority},
fence_obligation::FenceObligationRecord,
fence_reconciliation::{
FenceReconciliationIntent, FenceReconciliationObservation, FenceReconciliationRequest,
FenceReconciliationRequestError,
},
operation_plan::OperationPlanError,
},
ops::persistence::{
AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
ExecutionStageGuard, ExecutionWorkflowPersistenceError, FenceObligationPersistenceError,
FenceObligationRequirement, read_fence_obligation,
},
policy::fence_reconciliation::{FenceReconciliationError, validate},
ports::fence_reconciliation::{FenceReconciliationProvider, FenceReconciliationProviderError},
};
use thiserror::Error;
pub async fn observe_fence_acquisition<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
requirement: FenceObligationRequirement<'_>,
obligation: &FenceObligationRecord,
payload: &FenceAcquisitionPayload,
challenge: ArtifactChecksumRecord,
provider: &mut impl FenceReconciliationProvider,
admit: impl AsyncFnOnce(&FenceReconciliationRequest<'_, '_>) -> Result<(), E>,
) -> Result<FenceReconciliationObservation, FenceReconciliationExecutionError<E>> {
let plan = stage.plan();
let layout = stage.layout()?;
read_fence_obligation(layout, plan, requirement, &obligation.digest())?;
let authority = plan.attempt_authority(obligation.acquisition_operation())?;
let mut journal = AttemptJournalGuard::open(layout, &authority)?;
original_authority(plan, obligation, journal.record()?, payload)?;
let intent = FenceReconciliationIntent::new(plan, obligation, journal.record()?, challenge)?;
journal.reserve_planned_observation(
&plan.digest(),
intent.mutation_attempt(),
intent.digest().hash(),
)?;
let request = FenceReconciliationRequest::new(&intent, journal.record()?)?;
admit(&request)
.await
.map_err(FenceReconciliationExecutionError::Admission)?;
read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
let observation = provider
.observe_fence_acquisition(&request, journal.record()?)
.await?;
let association = match journal.record() {
Ok(record) => record,
Err(source) => {
return Err(FenceReconciliationExecutionError::AfterReplyJournal {
source,
observation: Box::new(observation),
});
}
};
if let Err(source) = validate(&request, association, &observation) {
return Err(FenceReconciliationExecutionError::Association {
source,
observation: Box::new(observation),
});
}
let layout = match stage.layout() {
Ok(layout) => layout,
Err(source) => {
return Err(FenceReconciliationExecutionError::AfterReplyStage {
source,
observation: Box::new(observation),
});
}
};
if let Err(source) = read_fence_obligation(layout, plan, requirement, &obligation.digest()) {
return Err(FenceReconciliationExecutionError::AfterReplyObligation {
source,
observation: Box::new(observation),
});
}
Ok(observation)
}
pub async fn settle_fence_acquisition_observation<E: std::error::Error + 'static>(
stage: &ExecutionStageGuard<'_>,
requirement: FenceObligationRequirement<'_>,
obligation: &FenceObligationRecord,
payload: &FenceAcquisitionPayload,
challenge: ArtifactChecksumRecord,
observation: &FenceReconciliationObservation,
qualify: impl AsyncFnOnce(
&FenceReconciliationRequest<'_, '_>,
&FenceReconciliationObservation,
) -> Result<ObservationReceiptRequest, E>,
) -> Result<ObservationOutcomeRecord, FenceReconciliationSettlementError<E>> {
let plan = stage.plan();
let layout = stage.layout()?;
read_fence_obligation(layout, plan, requirement, &obligation.digest())?;
let authority = plan.attempt_authority(obligation.acquisition_operation())?;
let mut journal = AttemptJournalGuard::open(layout, &authority)?;
original_authority(plan, obligation, journal.record()?, payload)?;
journal.admit_planned_progress(&plan.digest())?;
let intent = FenceReconciliationIntent::new(plan, obligation, journal.record()?, challenge)?;
let request = FenceReconciliationRequest::new(&intent, journal.record()?)?;
let admitted = validate(&request, journal.record()?, observation)?;
let receipt = qualify(&request, observation)
.await
.map_err(FenceReconciliationSettlementError::Qualification)?;
if receipt.attempt != admitted.observation_attempt()
|| receipt.request != admitted.request().hash()
|| receipt.outcome != admitted.outcome()
|| receipt.evidence != admitted.evidence().hash()
{
return Err(FenceReconciliationSettlementError::ReceiptMismatch);
}
validate(&request, journal.record()?, observation)?;
read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
let outcome = receipt.outcome;
journal.record_planned_observation(&plan.digest(), receipt)?;
read_fence_obligation(stage.layout()?, plan, requirement, &obligation.digest())?;
Ok(outcome)
}
#[derive(Debug, Error)]
pub enum FenceReconciliationSettlementError<E: std::error::Error + 'static> {
#[error(transparent)]
Plan(#[from] OperationPlanError),
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Obligation(#[from] FenceObligationPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Payload(#[from] FenceAcquisitionError),
#[error(transparent)]
Progress(#[from] ExecutionProgressPersistenceError),
#[error(transparent)]
Request(#[from] FenceReconciliationRequestError),
#[error(transparent)]
Association(#[from] FenceReconciliationError),
#[error("fence observation settlement qualification failed: {0}")]
Qualification(#[source] E),
#[error("fence observation settlement receipt mismatch")]
ReceiptMismatch,
}
#[derive(Debug, Error)]
pub enum FenceReconciliationExecutionError<E: std::error::Error + 'static> {
#[error(transparent)]
Plan(#[from] OperationPlanError),
#[error(transparent)]
Stage(#[from] ExecutionWorkflowPersistenceError),
#[error(transparent)]
Obligation(#[from] FenceObligationPersistenceError),
#[error(transparent)]
Journal(#[from] AttemptJournalError),
#[error(transparent)]
Payload(#[from] FenceAcquisitionError),
#[error(transparent)]
Progress(#[from] ExecutionProgressPersistenceError),
#[error(transparent)]
Request(#[from] FenceReconciliationRequestError),
#[error("fresh fence reconciliation admission failed: {0}")]
Admission(#[source] E),
#[error(transparent)]
Provider(#[from] FenceReconciliationProviderError),
#[error("fence reconciliation association failed: {source}")]
Association {
source: FenceReconciliationError,
observation: Box<FenceReconciliationObservation>,
},
#[error("fence reconciliation journal changed after reply: {source}")]
AfterReplyJournal {
source: AttemptJournalError,
observation: Box<FenceReconciliationObservation>,
},
#[error("fence reconciliation stage changed after reply: {source}")]
AfterReplyStage {
source: ExecutionWorkflowPersistenceError,
observation: Box<FenceReconciliationObservation>,
},
#[error("fence reconciliation obligation changed after reply: {source}")]
AfterReplyObligation {
source: FenceObligationPersistenceError,
observation: Box<FenceReconciliationObservation>,
},
}