icydb_core/db/session/
resumable_job.rs

1//! Module: session::resumable_job
2//! Responsibility: idempotent proof-checked application progress advancement.
3//! Does not own: application authorization, accumulator meaning, or page selection.
4//! Boundary: one synchronous operation closure -> excluded durable progress record.
5
6use crate::{
7    db::{
8        CompareProofAndAdvanceError, DbSession, ExhaustiveReadError, ReadSetRevisionProof,
9        ResumableJobAdvance, ResumableJobAdvanceReceipt, ResumableJobAdvanceRequest,
10        ResumableJobError, ResumableJobId, ResumableJobRecord, ResumableJobState,
11        ResumableJobStatus,
12        integrity::with_resumable_progress_store,
13        resource::budget::{ExecutionBudgetExceeded, HardExecutionContext},
14    },
15    traits::CanisterKind,
16};
17use icydb_diagnostic_code::{
18    DiagnosticExecutionBudgetResource, DiagnosticExecutionBudgetScope, DiagnosticExecutionLane,
19};
20
21const RESUMABLE_JOB_START_SHAPE: u64 = 0x7265_7375_6d65_0101;
22const RESUMABLE_JOB_LOAD_SHAPE: u64 = 0x7265_7375_6d65_0102;
23const RESUMABLE_JOB_ADVANCE_SHAPE: u64 = 0x7265_7375_6d65_0103;
24const RESUMABLE_JOB_ACKNOWLEDGE_SHAPE: u64 = 0x7265_7375_6d65_0104;
25
26impl<C: CanisterKind> DbSession<C> {
27    /// Create one durable application-owned job in IcyDB's excluded progress
28    /// domain. Every protected source store must be journaled.
29    pub fn start_resumable_job(
30        &self,
31        job_id: ResumableJobId,
32        proof: ReadSetRevisionProof,
33        initial_application_state: Vec<u8>,
34    ) -> Result<ResumableJobState, ResumableJobError> {
35        self.charge_resumable_operation(
36            DiagnosticExecutionLane::Mutation,
37            RESUMABLE_JOB_START_SHAPE,
38        )?;
39        self.verify_durable_read_set_revision_proof(&proof)
40            .map_err(map_exhaustive_error)?;
41        let record = ResumableJobRecord::new(job_id, proof, initial_application_state)?;
42        with_resumable_progress_store::<C, _>(|store| store.insert_resumable(&record))?;
43        Ok(record.state().clone())
44    }
45
46    /// Load the bounded current application state for one retained job.
47    pub fn resumable_job_state(
48        &self,
49        job_id: ResumableJobId,
50    ) -> Result<ResumableJobState, ResumableJobError> {
51        self.charge_resumable_operation(
52            DiagnosticExecutionLane::TrustedRead,
53            RESUMABLE_JOB_LOAD_SHAPE,
54        )?;
55        with_resumable_progress_store::<C, _>(|store| store.load_resumable(job_id))
56            .map(|record| record.state().clone())
57    }
58
59    /// Remove one completed or invalidated job after the application has
60    /// durably consumed its terminal result.
61    ///
62    /// The expected sequence prevents acknowledgement of a replaced state.
63    /// Repeating an acknowledgement after a lost reply succeeds when the job
64    /// is already absent. Active jobs with remaining continuation fail closed.
65    /// Recovery admission must succeed before retained progress can be removed.
66    pub fn acknowledge_resumable_job(
67        &self,
68        job_id: ResumableJobId,
69        expected_sequence: u64,
70    ) -> Result<(), ResumableJobError> {
71        self.charge_resumable_operation(
72            DiagnosticExecutionLane::Mutation,
73            RESUMABLE_JOB_ACKNOWLEDGE_SHAPE,
74        )?;
75        self.db
76            .ensure_recovered_control_state()
77            .map_err(|_| ResumableJobError::Internal)?;
78        with_resumable_progress_store::<C, _>(|store| {
79            let record = match store.load_resumable(job_id) {
80                Ok(record) => record,
81                Err(ResumableJobError::NotFound) => return Ok(()),
82                Err(error) => return Err(error),
83            };
84            if record.state().sequence != expected_sequence {
85                return Err(ResumableJobError::StaleSequence {
86                    expected: expected_sequence,
87                    actual: record.state().sequence,
88                });
89            }
90            let terminal = record.state().sequence > 0
91                && matches!(
92                    record.state().status,
93                    ResumableJobStatus::Completed | ResumableJobStatus::Invalidated
94                );
95            if !terminal {
96                return Err(ResumableJobError::NotTerminal);
97            }
98            store.remove_resumable(job_id)
99        })
100    }
101
102    /// Execute at most one application page, then atomically replace its progress
103    /// record after rechecking the complete source proof.
104    ///
105    /// Replaying the same sequence and idempotency key returns the persisted
106    /// receipt without invoking `operation` again. The closure is synchronous;
107    /// no `.await`, timer, or external call can split proof comparison from the
108    /// final progress write.
109    ///
110    /// The closure's side effects are not part of that atomic replacement. An
111    /// operation error, invalid advance, failed progress write, or post-page
112    /// invalidation does not undo writes already performed by the closure.
113    /// Invalidation can be returned in an `Ok` receipt. Exact replay suppresses
114    /// the closure only after a receipt has been retained successfully.
115    /// Prefer read-only page work or independently idempotent effects. If an IC
116    /// update relies on message rollback, its caller must trap on every error
117    /// and invalidated receipt before returning; this cannot undo effects from
118    /// earlier messages or external calls and provides no native rollback.
119    pub fn compare_proof_and_advance<E>(
120        &self,
121        request: &ResumableJobAdvanceRequest,
122        operation: impl FnOnce(&ResumableJobState) -> Result<ResumableJobAdvance, E>,
123    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
124        self.charge_resumable_operation(
125            DiagnosticExecutionLane::Mutation,
126            RESUMABLE_JOB_ADVANCE_SHAPE,
127        )?;
128        request.idempotency_key.validate()?;
129        if request.job_id.to_bytes() == [0; 32] {
130            return Err(ResumableJobError::InvalidJobId.into());
131        }
132
133        let record =
134            with_resumable_progress_store::<C, _>(|store| store.load_resumable(request.job_id))?;
135        if let Some(receipt) = exact_replay(&record, request) {
136            return Ok(receipt.clone());
137        }
138        ensure_request_can_advance(&record, request)?;
139
140        if let Err(error) = self.verify_durable_read_set_revision_proof(&record.state().proof) {
141            return Self::persist_or_report_pre_page_invalidation(record, request, error);
142        }
143
144        let advance = operation(record.state()).map_err(CompareProofAndAdvanceError::Operation)?;
145        advance.validate()?;
146
147        if let Err(error) = self.verify_durable_read_set_revision_proof(&record.state().proof) {
148            return Self::persist_or_report_post_page_invalidation(record, request, error);
149        }
150
151        let (candidate, receipt) = record.apply_advance(request, advance)?;
152        Self::replace_resumable_job_if_current(request, &candidate)?;
153        Ok(receipt)
154    }
155
156    fn persist_or_report_pre_page_invalidation<E>(
157        record: ResumableJobRecord,
158        request: &ResumableJobAdvanceRequest,
159        error: ExhaustiveReadError,
160    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
161        Self::persist_or_report_invalidation(record, request, error)
162    }
163
164    fn persist_or_report_post_page_invalidation<E>(
165        record: ResumableJobRecord,
166        request: &ResumableJobAdvanceRequest,
167        error: ExhaustiveReadError,
168    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
169        Self::persist_or_report_invalidation(record, request, error)
170    }
171
172    fn persist_or_report_invalidation<E>(
173        record: ResumableJobRecord,
174        request: &ResumableJobAdvanceRequest,
175        error: ExhaustiveReadError,
176    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
177        match error {
178            ExhaustiveReadError::Revision(error) if error.is_source_change() => {
179                let (invalidated, receipt) = record.invalidate(request)?;
180                Self::replace_resumable_job_if_current(request, &invalidated)?;
181                Ok(receipt)
182            }
183            other => Err(map_exhaustive_error(other).into()),
184        }
185    }
186
187    fn replace_resumable_job_if_current<E>(
188        request: &ResumableJobAdvanceRequest,
189        candidate: &ResumableJobRecord,
190    ) -> Result<(), CompareProofAndAdvanceError<E>> {
191        with_resumable_progress_store::<C, _>(|store| {
192            let current = store.load_resumable(request.job_id)?;
193            if exact_replay(&current, request).is_some() {
194                return Err(ResumableJobError::StaleSequence {
195                    expected: request.expected_sequence,
196                    actual: current.state().sequence,
197                });
198            }
199            ensure_request_can_advance(&current, request)?;
200            store.replace_resumable(candidate)
201        })?;
202        Ok(())
203    }
204
205    fn charge_resumable_operation(
206        &self,
207        lane: DiagnosticExecutionLane,
208        shape: u64,
209    ) -> Result<(), ResumableJobError> {
210        self.db
211            .request_execution_scope()
212            .charge(
213                HardExecutionContext::new(DiagnosticExecutionBudgetScope::Execution, lane, shape),
214                DiagnosticExecutionBudgetResource::QueryExecutions,
215                1,
216            )
217            .map_err(resumable_execution_budget_error)
218    }
219}
220
221fn exact_replay<'a>(
222    record: &'a ResumableJobRecord,
223    request: &ResumableJobAdvanceRequest,
224) -> Option<&'a ResumableJobAdvanceReceipt> {
225    record.last_receipt().filter(|receipt| {
226        receipt.request_sequence == request.expected_sequence
227            && receipt.idempotency_key() == &request.idempotency_key
228    })
229}
230
231fn ensure_request_can_advance(
232    record: &ResumableJobRecord,
233    request: &ResumableJobAdvanceRequest,
234) -> Result<(), ResumableJobError> {
235    if record.state().status == ResumableJobStatus::Invalidated {
236        return Err(ResumableJobError::Invalidated);
237    }
238    if record.state().status == ResumableJobStatus::Completed {
239        return Err(ResumableJobError::Completed);
240    }
241    if record.state().sequence != request.expected_sequence {
242        return Err(ResumableJobError::StaleSequence {
243            expected: request.expected_sequence,
244            actual: record.state().sequence,
245        });
246    }
247    Ok(())
248}
249
250fn map_exhaustive_error(error: ExhaustiveReadError) -> ResumableJobError {
251    match error {
252        ExhaustiveReadError::Revision(error) => ResumableJobError::SourceProof(error),
253        ExhaustiveReadError::Query(_) => ResumableJobError::Internal,
254    }
255}
256
257const fn resumable_execution_budget_error(error: ExecutionBudgetExceeded) -> ResumableJobError {
258    ResumableJobError::ExecutionBudgetExceeded {
259        resource: error.resource().raw(),
260        limit: error.limit(),
261        observed: error.observed(),
262        scope: error.scope().raw(),
263        lane: error.lane().raw(),
264        normalized_shape_fingerprint_prefix: error.normalized_shape_fingerprint_prefix(),
265    }
266}
icydb_core/db/session/resumable_job.rs

icydb_core/db/session/
resumable_job.rs
