ic_backup/workflow/application_ingress/
mod.rs

1//! Single opaque update or original recovery observation under existing stage authority.
2//!
3//! Applications retain protocol interpretation, fresh authority and durable byte custody.
4//! Host ingress receipts never settle application-owned paid effects or terminal release.
5
6use crate::{
7    model::{
8        application_ingress::{
9            ApplicationIngressError, ApplicationIngressPayload, ApplicationMutationRequest,
10            ApplicationObservationRequest,
11        },
12        artifacts::ArtifactChecksumRecord,
13        attempt_journal::{
14            AttemptJournalRecord, MutationOutcomeRecord, MutationReceiptRequest,
15            ObservationOutcomeRecord, ObservationReceiptRequest,
16        },
17        operation_plan::OperationPlanError,
18    },
19    ops::persistence::{AttemptJournalError, ExecutionStageCheckpointError, ExecutionStageGuard},
20    workflow::{ExecutionReservationError, execution_workflow::RetainedStageEvidence},
21};
22use std::future::Future;
23
24/// Exact already-reserved request; constructing a request never authorizes redispatch.
25#[derive(Debug)]
26pub enum ApplicationRequest<'a> {
27    /// One new original mutation.
28    Mutation(ApplicationMutationRequest<'a>),
29    /// One separately reserved observation of the original pending mutation.
30    Observation(ApplicationObservationRequest<'a>),
31}
32impl ApplicationRequest<'_> {
33    /// Read the exact bounded application bytes.
34    #[must_use]
35    pub const fn payload(&self) -> &ApplicationIngressPayload {
36        match self {
37            Self::Mutation(r) => r.payload(),
38            Self::Observation(r) => r.payload(),
39        }
40    }
41    fn validate(&self, journal: &AttemptJournalRecord) -> Result<(), ApplicationIngressError> {
42        match self {
43            Self::Mutation(r) => r.validate_journal(journal),
44            Self::Observation(r) => r.validate_journal(journal),
45        }
46    }
47    fn record_qualified<E: std::error::Error + 'static>(
48        &self,
49        journal: &mut crate::ops::persistence::AttemptJournalGuard<'_>,
50        qualified: &ApplicationQualification,
51    ) -> Result<(), ApplicationExecutionCause<E>> {
52        match self {
53            ApplicationRequest::Mutation(r) => {
54                if qualified.outcome != ObservationOutcomeRecord::Applied {
55                    return Err(ApplicationExecutionCause::MutationOutcome);
56                }
57                super::qualified_receipt::record_applied(
58                    journal,
59                    MutationReceiptRequest {
60                        attempt: r.mutation_attempt(),
61                        request: self.payload().digest(),
62                        outcome: MutationOutcomeRecord::Applied,
63                        evidence: qualified.evidence.clone(),
64                    },
65                    r.mutation_attempt(),
66                    self.payload().digest().hash(),
67                    ApplicationExecutionCause::MutationOutcome,
68                    || Ok(()),
69                )?;
70            }
71            ApplicationRequest::Observation(r) => {
72                journal.record_observation(ObservationReceiptRequest {
73                    attempt: r.observation_attempt(),
74                    request: self.payload().digest(),
75                    outcome: qualified.outcome,
76                    evidence: qualified.evidence.clone(),
77                })?;
78            }
79        }
80        Ok(())
81    }
82}
83
84/// Owned bounded reply, including any transport-specific evidence retained by its type.
85pub trait ApplicationReply: std::fmt::Debug + 'static {
86    /// Read raw bytes without application decoding.
87    fn bytes(&self) -> &[u8];
88}
89impl ApplicationReply for Vec<u8> {
90    fn bytes(&self) -> &[u8] {
91        self
92    }
93}
94
95/// Independently qualified application claims; matching bytes alone supply none.
96#[derive(Debug)]
97pub struct ApplicationQualification {
98    /// Mutation replies must be Applied; observations may explicitly remain uncertain.
99    pub outcome: ObservationOutcomeRecord,
100    /// Durable authenticated original-request/reply attribution evidence.
101    pub evidence: ArtifactChecksumRecord,
102    /// Application-qualified learned input for the existing stage checkpoint.
103    pub learned: ArtifactChecksumRecord,
104}
105
106impl ApplicationQualification {
107    fn checkpoint(
108        self,
109        stage: &ExecutionStageGuard<'_>,
110    ) -> Result<
111        (ObservationOutcomeRecord, Option<RetainedStageEvidence>),
112        ExecutionStageCheckpointError,
113    > {
114        let retained = if self.outcome == ObservationOutcomeRecord::Applied {
115            stage.reconcile_checkpoint(self.learned.clone())?;
116            Some(RetainedStageEvidence::new(
117                stage.binding().digest(),
118                self.learned,
119            ))
120        } else {
121            None
122        };
123        Ok((self.outcome, retained))
124    }
125}
126
127/// Mandatory application authority, single-dispatch transport and reply qualification.
128///
129/// Remote admission/qualification work requires its own separately accounted authority.
130/// `submit` must durably retain original signed ingress before dispatch and exact replies
131/// before returning, including before any cancellable qualification. No method may retry
132/// or infer a receipt. Use Agent's existing application-ingress provider for transport.
133pub trait ApplicationIngressProvider {
134    /// Typed application/transport refusal; an uncertain failure grants no resend.
135    type Error: std::error::Error + 'static;
136    /// Exact owned reply and transport evidence.
137    type Reply: ApplicationReply;
138    /// Re-admit original application custody synchronously, including after each await.
139    /// # Errors
140    /// Refuses changed authority, byte custody or original application intent.
141    fn check(&mut self, request: &ApplicationRequest<'_>) -> Result<(), Self::Error>;
142    /// Qualify current membership, permissions, safety and never-dispatched custody.
143    fn admit(
144        &mut self,
145        request: &ApplicationRequest<'_>,
146    ) -> impl Future<Output = Result<(), Self::Error>>;
147    /// Dispatch exactly once through the selected already-reserved lane.
148    fn submit(
149        &mut self,
150        request: &ApplicationRequest<'_>,
151        journal: &AttemptJournalRecord,
152    ) -> impl Future<Output = Result<Self::Reply, Self::Error>>;
153    /// Decode the exact protocol reply and qualify its original attribution and custody.
154    fn qualify(
155        &mut self,
156        request: &ApplicationRequest<'_>,
157        reply: &Self::Reply,
158    ) -> impl Future<Output = Result<ApplicationQualification, Self::Error>>;
159}
160
161/// Successful local settlement, without terminal, paid-effect or reference-release authority.
162#[derive(Debug)]
163pub struct ApplicationExecution<R> {
164    /// Exact owned response, including original transport evidence.
165    pub reply: R,
166    /// Explicit qualified outcome; no observation result grants automatic mutation reissue.
167    pub outcome: ObservationOutcomeRecord,
168    /// Existing checkpoint only when the original mutation is Applied and the stage complete.
169    pub retained: Option<RetainedStageEvidence>,
170}
171
172/// Exact local or integration refusal, with original spending retained.
173#[derive(Debug, thiserror::Error)]
174pub enum ApplicationExecutionCause<E: std::error::Error + 'static> {
175    /// Original stage/journal/reservation admission failed.
176    #[error(transparent)]
177    Reservation(#[from] ExecutionReservationError),
178    /// Original plan cannot supply the requested operation.
179    #[error(transparent)]
180    Plan(#[from] OperationPlanError),
181    /// Original opaque payload or reserved request differs.
182    #[error(transparent)]
183    Request(#[from] ApplicationIngressError),
184    /// Receipt persistence or current journal admission failed.
185    #[error(transparent)]
186    Journal(#[from] AttemptJournalError),
187    /// Existing stage checkpoint failed after journal exclusion was released.
188    #[error(transparent)]
189    Checkpoint(#[from] ExecutionStageCheckpointError),
190    /// Required application admission, transport or qualification failed.
191    #[error("application ingress callback failed: {0}")]
192    Application(#[source] E),
193    /// This API checkpoints a complete singleton stage.
194    #[error("application ingress requires a singleton stage")]
195    NonSingletonStage,
196    /// No unresolved original mutation exists for observation reservation.
197    #[error("original pending mutation is absent")]
198    NoPendingMutation,
199    /// A direct mutation reply cannot record an inferred negative or uncertain receipt.
200    #[error("direct application mutation requires qualified Applied evidence")]
201    MutationOutcome,
202    /// The provider violated the original reply ceiling.
203    #[error("application reply exceeds original ceiling")]
204    ReplyTooLarge,
205}
206
207/// A returned response survives every post-reply failure; cancellation still needs durable custody.
208#[derive(Debug, thiserror::Error)]
209pub enum ApplicationExecutionError<E: std::error::Error + 'static, R: std::fmt::Debug> {
210    /// No reply was returned; original consumed authority remains occupied.
211    #[error(transparent)]
212    BeforeReply(#[from] ApplicationExecutionCause<E>),
213    /// Exact response remains owned even if qualification, persistence or checkpointing failed.
214    #[error("application reply could not settle: {source}")]
215    AfterReply {
216        /// Typed refusal, never replacement spending authority.
217        source: ApplicationExecutionCause<E>,
218        /// Original bounded response and transport evidence.
219        reply: Box<R>,
220    },
221}
222
223/// Execute one new original opaque mutation in a singleton stage and publish its existing stage checkpoint.
224/// # Errors
225/// Rejects changed/occupied originals, admission, submission or unqualified replies.
226pub async fn execute<P: ApplicationIngressProvider>(
227    stage: &ExecutionStageGuard<'_>,
228    sequence: u64,
229    payload: &ApplicationIngressPayload,
230    provider: &mut P,
231) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
232    run(stage, sequence, payload, provider, false).await
233}
234
235/// Observe an original pending mutation using separate finite observation authority.
236///
237/// Lost observations remain occupied. NotApplied/Uncertain settle only that observation;
238/// they publish no complete stage checkpoint and supply no automatic mutation resend.
239/// # Errors
240/// Rejects absent/changed pending originals, spent observations and unqualified responses.
241pub async fn reconcile<P: ApplicationIngressProvider>(
242    stage: &ExecutionStageGuard<'_>,
243    sequence: u64,
244    payload: &ApplicationIngressPayload,
245    provider: &mut P,
246) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
247    run(stage, sequence, payload, provider, true).await
248}
249
250async fn run<P: ApplicationIngressProvider>(
251    stage: &ExecutionStageGuard<'_>,
252    sequence: u64,
253    payload: &ApplicationIngressPayload,
254    provider: &mut P,
255    observation: bool,
256) -> Result<ApplicationExecution<P::Reply>, ApplicationExecutionError<P::Error, P::Reply>> {
257    type Cause<E> = ApplicationExecutionCause<E>;
258    if stage.plan().operations().len() != 1 {
259        return Err(Cause::NonSingletonStage.into());
260    }
261    let authority = stage
262        .plan()
263        .attempt_authority(sequence)
264        .map_err(Cause::from)?;
265    if observation {
266        payload.validate_observation_binding(authority.binding())
267    } else {
268        payload.validate_mutation_binding(authority.binding())
269    }
270    .map_err(Cause::from)?;
271    let (reply, result) = super::ic_mutation::dispatch_reserved(
272        stage,
273        &authority,
274        provider,
275        |journal| {
276            if observation {
277                let mutation = journal
278                    .record()?
279                    .view()
280                    .pending_mutation
281                    .ok_or(Cause::NoPendingMutation)?;
282                journal
283                    .reserve_planned_observation(
284                        &stage.plan().digest(),
285                        mutation,
286                        payload.digest().hash(),
287                    )
288                    .map_err(ExecutionReservationError::from)?;
289            } else {
290                journal
291                    .reserve_planned_mutation(&stage.plan().digest())
292                    .map_err(ExecutionReservationError::from)?;
293            }
294            Ok::<_, Cause<P::Error>>(())
295        },
296        |journal| {
297            Ok(if observation {
298                ApplicationRequest::Observation(ApplicationObservationRequest::new(
299                    stage.plan(),
300                    sequence,
301                    journal,
302                    payload,
303                )?)
304            } else {
305                ApplicationRequest::Mutation(ApplicationMutationRequest::new(
306                    stage.plan(),
307                    sequence,
308                    journal,
309                    payload,
310                )?)
311            })
312        },
313        async |provider, request, journal| {
314            provider.check(request).map_err(Cause::Application)?;
315            provider.admit(request).await.map_err(Cause::Application)?;
316            stage.layout().map_err(ExecutionReservationError::from)?;
317            request.validate(journal)?;
318            provider.check(request).map_err(Cause::Application)?;
319            provider
320                .submit(request, journal)
321                .await
322                .map_err(Cause::Application)
323        },
324        async |provider, journal, request, reply| {
325            stage.layout().map_err(ExecutionReservationError::from)?;
326            request.validate(journal.record()?)?;
327            provider.check(request).map_err(Cause::Application)?;
328            if reply.bytes().len() > payload.reply_limit() {
329                return Err(Cause::ReplyTooLarge);
330            }
331            let qualified = provider
332                .qualify(request, reply)
333                .await
334                .map_err(Cause::Application)?;
335            stage.layout().map_err(ExecutionReservationError::from)?;
336            request.validate(journal.record()?)?;
337            provider.check(request).map_err(Cause::Application)?;
338            request.record_qualified(journal, &qualified)?;
339            Ok(qualified)
340        },
341    )
342    .await?;
343    // dispatch_reserved has released the attempt guard before checkpoint publication.
344    let result = result.and_then(|qualified| qualified.checkpoint(stage).map_err(Cause::from));
345    let (reply, (outcome, retained)) =
346        super::qualified_receipt::retain_reply(reply, result, |source, reply| {
347            ApplicationExecutionError::AfterReply { source, reply }
348        })?;
349    Ok(ApplicationExecution {
350        reply,
351        outcome,
352        retained,
353    })
354}
355
356#[cfg(test)]
357mod tests;
ic_backup/workflow/application_ingress/mod.rs

ic_backup/workflow/application_ingress/
mod.rs
