ic_backup/workflow/ic_mutation/
mod.rs

1//! Single original mutation reservation and dispatch under continuous journal exclusion.
2
3use crate::ports::IcProviderError;
4use crate::{
5    model::{
6        attempt_journal::AttemptAuthorityRecord,
7        ic_mutation::{IcMutationAcknowledgement, IcMutationRequest, IcMutationRequestError},
8        ic_request::IcManagementRequestRecord,
9    },
10    ops::persistence::{
11        AttemptJournalError, AttemptJournalGuard, ExecutionProgressPersistenceError,
12        ExecutionStageGuard, ExecutionWorkflowPersistenceError,
13    },
14    ports::ic_mutation::IcMutationProvider,
15};
16
17// Domain callbacks retain their admission/re-admission order and post-reply errors.
18// The journal drops on return, before a caller can publish a stage checkpoint.
19pub(super) async fn submit_with<E, T>(
20    stage: &ExecutionStageGuard<'_>,
21    authority: &AttemptAuthorityRecord,
22    payload: &IcManagementRequestRecord,
23    provider: &mut impl IcMutationProvider,
24    admit: impl AsyncFnOnce(&IcMutationRequest<'_>) -> Result<(), E>,
25    complete: impl AsyncFnOnce(
26        &mut AttemptJournalGuard<'_>,
27        &IcMutationRequest<'_>,
28        &IcMutationAcknowledgement,
29    ) -> T,
30) -> Result<(IcMutationAcknowledgement, T), E>
31where
32    E: From<IcMutationRequestError> + From<ExecutionReservationError> + From<IcProviderError>,
33{
34    payload
35        .validate_mutation_binding(authority.binding())
36        .map_err(IcMutationRequestError::from)
37        .map_err(E::from)?;
38    execute_reserved(
39        stage,
40        authority,
41        |_| Ok(()),
42        |journal| {
43            IcMutationRequest::new(
44                stage.plan(),
45                authority.binding().operation_sequence(),
46                journal,
47                payload,
48            )
49            .map_err(E::from)
50        },
51        admit,
52        async |request, journal| {
53            provider
54                .submit_mutation(request, journal)
55                .await
56                .map_err(E::from)
57        },
58        complete,
59    )
60    .await
61}
62
63// The sole open -> original validation -> reserve -> bind -> admit -> submit
64// sequence. Request/reply semantics stay with each typed caller. Keep the same
65// journal through both awaits and completion; checkpointing happens after return.
66pub(super) async fn execute_reserved<Q, R, T, E>(
67    stage: &ExecutionStageGuard<'_>,
68    authority: &AttemptAuthorityRecord,
69    validate: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<(), E>,
70    bind: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<Q, E>,
71    admit: impl AsyncFnOnce(&Q) -> Result<(), E>,
72    submit: impl AsyncFnOnce(&Q, &crate::model::attempt_journal::AttemptJournalRecord) -> Result<R, E>,
73    complete: impl AsyncFnOnce(&mut AttemptJournalGuard<'_>, &Q, &R) -> T,
74) -> Result<(R, T), E>
75where
76    E: From<ExecutionReservationError>,
77{
78    let mut journal = AttemptJournalGuard::open(
79        stage
80            .layout()
81            .map_err(ExecutionReservationError::from)
82            .map_err(E::from)?,
83        authority,
84    )
85    .map_err(ExecutionReservationError::from)
86    .map_err(E::from)?;
87    validate(
88        journal
89            .record()
90            .map_err(ExecutionReservationError::from)
91            .map_err(E::from)?,
92    )?;
93    journal
94        .reserve_planned_mutation(&stage.plan().digest())
95        .map_err(ExecutionReservationError::from)
96        .map_err(E::from)?;
97    let request = bind(
98        journal
99            .record()
100            .map_err(ExecutionReservationError::from)
101            .map_err(E::from)?,
102    )?;
103    admit(&request).await?;
104    let reply = submit(
105        &request,
106        journal
107            .record()
108            .map_err(ExecutionReservationError::from)
109            .map_err(E::from)?,
110    )
111    .await?;
112    let result = complete(&mut journal, &request, &reply).await;
113    Ok((reply, result))
114}
115
116/// Common original-stage and durable-reservation failure before a reply exists.
117/// Every failure retains original spending and supplies no retry authority.
118#[derive(Debug, thiserror::Error)]
119pub enum ExecutionReservationError {
120    /// Retained stage or ancestor admission failed.
121    #[error(transparent)]
122    Stage(#[from] ExecutionWorkflowPersistenceError),
123    /// Original selected journal could not be admitted.
124    #[error(transparent)]
125    Journal(#[from] AttemptJournalError),
126    /// Complete original progress or bounded reservation failed.
127    #[error(transparent)]
128    Progress(#[from] ExecutionProgressPersistenceError),
129}
130
131#[cfg(all(test, unix))]
132mod tests;
ic_backup/workflow/ic_mutation/mod.rs

ic_backup/workflow/ic_mutation/
mod.rs
