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 and atomically retain its next
103    /// state 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    pub fn compare_proof_and_advance<E>(
110        &self,
111        request: &ResumableJobAdvanceRequest,
112        operation: impl FnOnce(&ResumableJobState) -> Result<ResumableJobAdvance, E>,
113    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
114        self.charge_resumable_operation(
115            DiagnosticExecutionLane::Mutation,
116            RESUMABLE_JOB_ADVANCE_SHAPE,
117        )?;
118        request.idempotency_key.validate()?;
119        if request.job_id.to_bytes() == [0; 32] {
120            return Err(ResumableJobError::InvalidJobId.into());
121        }
122
123        let record =
124            with_resumable_progress_store::<C, _>(|store| store.load_resumable(request.job_id))?;
125        if let Some(receipt) = exact_replay(&record, request) {
126            return Ok(receipt.clone());
127        }
128        ensure_request_can_advance(&record, request)?;
129
130        if let Err(error) = self.verify_durable_read_set_revision_proof(&record.state().proof) {
131            return Self::persist_or_report_pre_page_invalidation(record, request, error);
132        }
133
134        let advance = operation(record.state()).map_err(CompareProofAndAdvanceError::Operation)?;
135        advance.validate()?;
136
137        if let Err(error) = self.verify_durable_read_set_revision_proof(&record.state().proof) {
138            return Self::persist_or_report_post_page_invalidation(record, request, error);
139        }
140
141        let (candidate, receipt) = record.apply_advance(request, advance)?;
142        Self::replace_resumable_job_if_current(request, &candidate)?;
143        Ok(receipt)
144    }
145
146    fn persist_or_report_pre_page_invalidation<E>(
147        record: ResumableJobRecord,
148        request: &ResumableJobAdvanceRequest,
149        error: ExhaustiveReadError,
150    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
151        Self::persist_or_report_invalidation(record, request, error)
152    }
153
154    fn persist_or_report_post_page_invalidation<E>(
155        record: ResumableJobRecord,
156        request: &ResumableJobAdvanceRequest,
157        error: ExhaustiveReadError,
158    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
159        Self::persist_or_report_invalidation(record, request, error)
160    }
161
162    fn persist_or_report_invalidation<E>(
163        record: ResumableJobRecord,
164        request: &ResumableJobAdvanceRequest,
165        error: ExhaustiveReadError,
166    ) -> Result<ResumableJobAdvanceReceipt, CompareProofAndAdvanceError<E>> {
167        match error {
168            ExhaustiveReadError::Revision(error) if error.is_source_change() => {
169                let (invalidated, receipt) = record.invalidate(request)?;
170                Self::replace_resumable_job_if_current(request, &invalidated)?;
171                Ok(receipt)
172            }
173            other => Err(map_exhaustive_error(other).into()),
174        }
175    }
176
177    fn replace_resumable_job_if_current<E>(
178        request: &ResumableJobAdvanceRequest,
179        candidate: &ResumableJobRecord,
180    ) -> Result<(), CompareProofAndAdvanceError<E>> {
181        with_resumable_progress_store::<C, _>(|store| {
182            let current = store.load_resumable(request.job_id)?;
183            if exact_replay(&current, request).is_some() {
184                return Err(ResumableJobError::StaleSequence {
185                    expected: request.expected_sequence,
186                    actual: current.state().sequence,
187                });
188            }
189            ensure_request_can_advance(&current, request)?;
190            store.replace_resumable(candidate)
191        })?;
192        Ok(())
193    }
194
195    fn charge_resumable_operation(
196        &self,
197        lane: DiagnosticExecutionLane,
198        shape: u64,
199    ) -> Result<(), ResumableJobError> {
200        self.db
201            .request_execution_scope()
202            .charge(
203                HardExecutionContext::new(DiagnosticExecutionBudgetScope::Execution, lane, shape),
204                DiagnosticExecutionBudgetResource::QueryExecutions,
205                1,
206            )
207            .map_err(resumable_execution_budget_error)
208    }
209}
210
211fn exact_replay<'a>(
212    record: &'a ResumableJobRecord,
213    request: &ResumableJobAdvanceRequest,
214) -> Option<&'a ResumableJobAdvanceReceipt> {
215    record.last_receipt().filter(|receipt| {
216        receipt.request_sequence == request.expected_sequence
217            && receipt.idempotency_key() == &request.idempotency_key
218    })
219}
220
221fn ensure_request_can_advance(
222    record: &ResumableJobRecord,
223    request: &ResumableJobAdvanceRequest,
224) -> Result<(), ResumableJobError> {
225    if record.state().status == ResumableJobStatus::Invalidated {
226        return Err(ResumableJobError::Invalidated);
227    }
228    if record.state().status == ResumableJobStatus::Completed {
229        return Err(ResumableJobError::Completed);
230    }
231    if record.state().sequence != request.expected_sequence {
232        return Err(ResumableJobError::StaleSequence {
233            expected: request.expected_sequence,
234            actual: record.state().sequence,
235        });
236    }
237    Ok(())
238}
239
240fn map_exhaustive_error(error: ExhaustiveReadError) -> ResumableJobError {
241    match error {
242        ExhaustiveReadError::Revision(error) => ResumableJobError::SourceProof(error),
243        ExhaustiveReadError::Query(_) => ResumableJobError::Internal,
244    }
245}
246
247const fn resumable_execution_budget_error(error: ExecutionBudgetExceeded) -> ResumableJobError {
248    ResumableJobError::ExecutionBudgetExceeded {
249        resource: error.resource().raw(),
250        limit: error.limit(),
251        observed: error.observed(),
252        scope: error.scope().raw(),
253        lane: error.lane().raw(),
254        normalized_shape_fingerprint_prefix: error.normalized_shape_fingerprint_prefix(),
255    }
256}
icydb_core/db/session/resumable_job.rs

icydb_core/db/session/
resumable_job.rs
