icydb_core/db/session/
mutation_job.rs

1//! Module: session::mutation_job
2//! Responsibility: charged mutation-job start, state load, and terminal acknowledgement.
3//! Does not own: SQL lowering, Forward/Verify execution, or authorization.
4//! Boundary: trusted session API -> excluded mutation progress record.
5
6use crate::{
7    db::{
8        DbSession, MutationJobError, MutationJobId, MutationJobState,
9        executor::budget::{ExecutionBudgetExceeded, HardExecutionContext},
10        integrity::with_mutation_progress_store,
11    },
12    traits::CanisterKind,
13};
14use icydb_diagnostic_code::{
15    DiagnosticExecutionBudgetResource, DiagnosticExecutionBudgetScope, DiagnosticExecutionLane,
16};
17
18#[cfg(feature = "sql")]
19use crate::db::{
20    integrity::InsertMutationJobResult,
21    mutation_job::{CanonicalMutationIntent, MutationJobRecord},
22};
23
24#[cfg(feature = "sql")]
25const MUTATION_JOB_START_SHAPE: u64 = 0x6d75_7461_7465_0100;
26const MUTATION_JOB_LOAD_SHAPE: u64 = 0x6d75_7461_7465_0101;
27const MUTATION_JOB_ACKNOWLEDGE_SHAPE: u64 = 0x6d75_7461_7465_0102;
28
29impl<C: CanisterKind> DbSession<C> {
30    /// Start one durable trusted fixed SQL mutation job.
31    ///
32    /// SQL is parsed and admitted exactly once for a new identity. The
33    /// catalog-native intent and initial engine checkpoint are durably retained
34    /// before this method returns, and no target row is read or mutated.
35    /// Repeating the same canonical request returns the retained state without
36    /// replacing its operation timestamp or resetting progress.
37    #[cfg(feature = "sql")]
38    pub fn start_trusted_sql_mutation_job(
39        &self,
40        job_id: MutationJobId,
41        sql: &str,
42    ) -> Result<MutationJobState, MutationJobError> {
43        job_id.validate()?;
44        self.charge_mutation_job_operation(
45            DiagnosticExecutionLane::Mutation,
46            MUTATION_JOB_START_SHAPE,
47        )?;
48        let prepared = self.prepare_mutation_job_start(job_id, sql)?;
49        let submitted_intent = CanonicalMutationIntent::decode(&prepared.canonical_intent)?;
50        let submitted = MutationJobRecord::new(
51            job_id,
52            prepared.canonical_intent,
53            prepared.engine_continuation,
54        )?;
55        with_mutation_progress_store::<C, _>(|store| match store.insert_mutation(&submitted)? {
56            InsertMutationJobResult::Inserted => Ok(submitted.state().clone()),
57            InsertMutationJobResult::Occupied(retained) => {
58                resolve_occupied_mutation_job_start(&retained, &submitted_intent)
59            }
60        })
61    }
62
63    /// Load the bounded public state for one retained mutation job.
64    pub fn mutation_job_state(
65        &self,
66        job_id: MutationJobId,
67    ) -> Result<MutationJobState, MutationJobError> {
68        self.charge_mutation_job_operation(
69            DiagnosticExecutionLane::TrustedRead,
70            MUTATION_JOB_LOAD_SHAPE,
71        )?;
72        with_mutation_progress_store::<C, _>(|store| store.load_mutation(job_id))
73            .map(|record| record.state().clone())
74    }
75
76    /// Remove one terminal mutation job after its result has been consumed.
77    ///
78    /// Repeating acknowledgement after a lost response succeeds when the job
79    /// is already absent. Active jobs and stale terminal sequences fail closed.
80    pub fn acknowledge_mutation_job(
81        &self,
82        job_id: MutationJobId,
83        expected_terminal_sequence: u64,
84    ) -> Result<(), MutationJobError> {
85        self.charge_mutation_job_operation(
86            DiagnosticExecutionLane::Mutation,
87            MUTATION_JOB_ACKNOWLEDGE_SHAPE,
88        )?;
89        with_mutation_progress_store::<C, _>(|store| {
90            store.acknowledge_mutation(job_id, expected_terminal_sequence)
91        })
92    }
93
94    fn charge_mutation_job_operation(
95        &self,
96        lane: DiagnosticExecutionLane,
97        shape: u64,
98    ) -> Result<(), MutationJobError> {
99        self.db
100            .request_execution_scope()
101            .charge(
102                HardExecutionContext::new(DiagnosticExecutionBudgetScope::Execution, lane, shape),
103                DiagnosticExecutionBudgetResource::QueryExecutions,
104                1,
105            )
106            .map_err(mutation_job_execution_budget_error)
107    }
108}
109
110#[cfg(feature = "sql")]
111fn resolve_occupied_mutation_job_start(
112    retained: &MutationJobRecord,
113    submitted_intent: &CanonicalMutationIntent,
114) -> Result<MutationJobState, MutationJobError> {
115    let retained_intent = CanonicalMutationIntent::decode(retained.canonical_intent())?;
116    if retained_intent.same_start_request(submitted_intent) {
117        return Ok(retained.state().clone());
118    }
119    if !retained_intent.same_authority(submitted_intent) {
120        return Err(MutationJobError::AuthorityMismatch);
121    }
122    Err(MutationJobError::IdentityConflict)
123}
124
125const fn mutation_job_execution_budget_error(error: ExecutionBudgetExceeded) -> MutationJobError {
126    MutationJobError::ExecutionBudgetExceeded {
127        resource: error.resource().raw(),
128        limit: error.limit(),
129        observed: error.observed(),
130        scope: error.scope().raw(),
131        lane: error.lane().raw(),
132        normalized_shape_fingerprint_prefix: error.normalized_shape_fingerprint_prefix(),
133    }
134}
135
136#[cfg(test)]
137mod tests {
138    use super::*;
139    use crate::{
140        db::{
141            MutationJobAdvanceRequest, MutationJobIdempotencyKey, MutationJobPhase,
142            MutationJobRestartReason, MutationJobStatus, RequestExecutionRoot, StoreRegistry,
143            mutation_job::{MutationJobRecord, MutationJobTransition},
144        },
145        traits::Path,
146    };
147    #[cfg(feature = "sql")]
148    use crate::{
149        db::{
150            data::{AcceptedFixedUpdatePatch, FieldSlot},
151            query::plan::expr::{BinaryOp, Expr, FieldId},
152        },
153        types::Timestamp,
154        value::Value,
155    };
156
157    struct TestCanister;
158
159    impl Path for TestCanister {
160        const PATH: &'static str = "db::session::mutation_job::tests::Canister";
161    }
162
163    impl CanisterKind for TestCanister {
164        const COMMIT_MEMORY_ID: u8 = 244;
165        const COMMIT_STABLE_KEY: &'static str = "icydb.test.mutation-job.commit.v1";
166        const INTEGRITY_PROGRESS_MEMORY_ID: u8 = 245;
167        const INTEGRITY_PROGRESS_STABLE_KEY: &'static str = "icydb.test.mutation-job.progress.v1";
168    }
169
170    thread_local! {
171        static STORE_REGISTRY: StoreRegistry = StoreRegistry::new();
172    }
173
174    fn job_id() -> MutationJobId {
175        MutationJobId::try_from_bytes([19; 32]).expect("nonzero mutation job id should admit")
176    }
177
178    fn session() -> DbSession<TestCanister> {
179        let root = RequestExecutionRoot::__new_runtime_root();
180        DbSession::new(&STORE_REGISTRY, &root)
181    }
182
183    #[test]
184    fn session_load_and_terminal_acknowledgement_preserve_the_store_contract() {
185        let initial = MutationJobRecord::new(job_id(), vec![1, 2], Vec::new())
186            .expect("bounded initial record should admit");
187        with_mutation_progress_store::<TestCanister, _>(|store| {
188            store.insert_mutation(&initial).map(|_| ())
189        })
190        .expect("initial record should insert");
191
192        let session = session();
193        assert_eq!(
194            session.mutation_job_state(job_id()),
195            Ok(initial.state().clone())
196        );
197        assert_eq!(
198            session.acknowledge_mutation_job(job_id(), 0),
199            Err(MutationJobError::Active),
200        );
201
202        let request = MutationJobAdvanceRequest::new(
203            job_id(),
204            0,
205            MutationJobIdempotencyKey::new("authority-drift")
206                .expect("bounded idempotency key should admit"),
207        );
208        let (terminal, _) = initial
209            .apply_transition(
210                &request,
211                MutationJobTransition::new(
212                    MutationJobStatus::RestartRequired(
213                        MutationJobRestartReason::AcceptedSchemaChanged,
214                    ),
215                    MutationJobPhase::Forward,
216                    Vec::new(),
217                    0,
218                    0,
219                    0,
220                ),
221            )
222            .expect("terminal restart receipt should admit");
223        with_mutation_progress_store::<TestCanister, _>(|store| store.replace_mutation(&terminal))
224            .expect("terminal record should replace active state");
225
226        assert_eq!(
227            session.acknowledge_mutation_job(job_id(), 0),
228            Err(MutationJobError::StaleSequence {
229                expected: 0,
230                actual: 1,
231            }),
232        );
233        assert_eq!(session.acknowledge_mutation_job(job_id(), 1), Ok(()));
234        assert_eq!(session.acknowledge_mutation_job(job_id(), 1), Ok(()));
235        assert_eq!(
236            session.mutation_job_state(job_id()),
237            Err(MutationJobError::NotFound),
238        );
239    }
240
241    #[cfg(feature = "sql")]
242    fn canonical_intent(
243        authority: u8,
244        scope_value: u64,
245        timestamp: i64,
246    ) -> CanonicalMutationIntent {
247        let scope = Expr::Binary {
248            op: BinaryOp::Eq,
249            left: Box::new(Expr::Field(FieldId::new("collection_id"))),
250            right: Box::new(Expr::Literal(Value::Nat64(scope_value))),
251        };
252        let patch = AcceptedFixedUpdatePatch::from_canonical_fields(vec![(
253            FieldSlot::from_validated_index(1),
254            vec![3, 4, 5],
255        )])
256        .expect("fixed patch should admit");
257        CanonicalMutationIntent::new(
258            [authority; 16],
259            [authority; 32],
260            "journaled".to_string(),
261            "schema::Token".to_string(),
262            7,
263            11,
264            1,
265            [authority; 16],
266            &scope,
267            &patch,
268            Timestamp::from_millis(timestamp),
269            17,
270        )
271        .expect("canonical intent should admit")
272    }
273
274    #[cfg(feature = "sql")]
275    #[test]
276    fn occupied_start_distinguishes_replay_authority_drift_and_identity_conflict() {
277        let retained_intent = canonical_intent(1, 7, 100);
278        let retained = MutationJobRecord::new(
279            job_id(),
280            retained_intent.encode().expect("intent should encode"),
281            vec![9],
282        )
283        .expect("record should admit");
284
285        assert_eq!(
286            resolve_occupied_mutation_job_start(&retained, &canonical_intent(1, 7, 200)),
287            Ok(retained.state().clone()),
288        );
289        assert_eq!(
290            resolve_occupied_mutation_job_start(&retained, &canonical_intent(2, 7, 200)),
291            Err(MutationJobError::AuthorityMismatch),
292        );
293        assert_eq!(
294            resolve_occupied_mutation_job_start(&retained, &canonical_intent(1, 8, 200)),
295            Err(MutationJobError::IdentityConflict),
296        );
297    }
298}
icydb_core/db/session/mutation_job.rs

icydb_core/db/session/
mutation_job.rs
