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    dispatch_reserved(
79        stage,
80        authority,
81        &mut (),
82        |journal| {
83            validate(
84                journal
85                    .record()
86                    .map_err(ExecutionReservationError::from)
87                    .map_err(E::from)?,
88            )?;
89            journal
90                .reserve_planned_mutation(&stage.plan().digest())
91                .map_err(ExecutionReservationError::from)
92                .map_err(E::from)?;
93            Ok(())
94        },
95        bind,
96        async |(), request, journal| {
97            admit(request).await?;
98            submit(request, journal).await
99        },
100        async |(), journal, request, reply| complete(journal, request, reply).await,
101    )
102    .await
103}
104
105// One exclusion/binding/dispatch/completion owner for management and opaque ingress.
106// Reservation selects the existing mutation or observation journal transition.
107pub(super) async fn dispatch_reserved<Q, R, T, E, C>(
108    stage: &ExecutionStageGuard<'_>,
109    authority: &AttemptAuthorityRecord,
110    context: &mut C,
111    reserve: impl FnOnce(&mut AttemptJournalGuard<'_>) -> Result<(), E>,
112    bind: impl FnOnce(&crate::model::attempt_journal::AttemptJournalRecord) -> Result<Q, E>,
113    dispatch: impl AsyncFnOnce(
114        &mut C,
115        &Q,
116        &crate::model::attempt_journal::AttemptJournalRecord,
117    ) -> Result<R, E>,
118    complete: impl AsyncFnOnce(&mut C, &mut AttemptJournalGuard<'_>, &Q, &R) -> T,
119) -> Result<(R, T), E>
120where
121    E: From<ExecutionReservationError>,
122{
123    let mut journal = AttemptJournalGuard::open(
124        stage
125            .layout()
126            .map_err(ExecutionReservationError::from)
127            .map_err(E::from)?,
128        authority,
129    )
130    .map_err(ExecutionReservationError::from)
131    .map_err(E::from)?;
132    reserve(&mut journal)?;
133    let request = bind(
134        journal
135            .record()
136            .map_err(ExecutionReservationError::from)
137            .map_err(E::from)?,
138    )?;
139    let reply = dispatch(
140        context,
141        &request,
142        journal
143            .record()
144            .map_err(ExecutionReservationError::from)
145            .map_err(E::from)?,
146    )
147    .await?;
148    let result = complete(context, &mut journal, &request, &reply).await;
149    Ok((reply, result))
150}
151
152/// Common original-stage and durable-reservation failure before a reply exists.
153/// Every failure retains original spending and supplies no retry authority.
154#[derive(Debug, thiserror::Error)]
155pub enum ExecutionReservationError {
156    /// Retained stage or ancestor admission failed.
157    #[error(transparent)]
158    Stage(#[from] ExecutionWorkflowPersistenceError),
159    /// Original selected journal could not be admitted.
160    #[error(transparent)]
161    Journal(#[from] AttemptJournalError),
162    /// Complete original progress or bounded reservation failed.
163    #[error(transparent)]
164    Progress(#[from] ExecutionProgressPersistenceError),
165}
166
167#[cfg(all(test, unix))]
168mod tests;
ic_backup/workflow/ic_mutation/mod.rs

ic_backup/workflow/ic_mutation/
mod.rs
