sim_lib_operation_gate/
operation_service.rs

1use std::{collections::BTreeMap, sync::Arc};
2
3use sim_kernel::{ContentId, Datum, Symbol};
4use sim_lib_journal::{
5    Journal, JournalBackend, JournalEntry, JournalObject, Lease, VerifiedSnapshot,
6};
7
8use crate::{durable::*, operation_error::OperationError};
9
10/// Complete verified durable record for one operation.
11#[derive(Clone, Debug, PartialEq, Eq)]
12pub struct OperationRecord {
13    intent: OperationIntent,
14    grant: OperationGrant,
15    attempt: Option<OperationAttempt>,
16    dispatch: Option<OperationDispatch>,
17    receipt: Option<PerformerReceipt>,
18}
19
20impl OperationRecord {
21    /// Returns the immutable semantic intent.
22    pub const fn intent(&self) -> &OperationIntent {
23        &self.intent
24    }
25
26    /// Returns the separately recorded grant.
27    pub const fn grant(&self) -> &OperationGrant {
28        &self.grant
29    }
30
31    /// Returns the attempt selected before dispatch, if dispatch occurred.
32    pub const fn attempt(&self) -> Option<&OperationAttempt> {
33        self.attempt.as_ref()
34    }
35
36    /// Returns the durable performer handoff, if it occurred.
37    pub const fn dispatch(&self) -> Option<&OperationDispatch> {
38        self.dispatch.as_ref()
39    }
40
41    /// Returns the raw performer acknowledgement, if it became durable.
42    pub const fn receipt(&self) -> Option<&PerformerReceipt> {
43        self.receipt.as_ref()
44    }
45
46    /// Derives the exact three-state public projection.
47    pub fn state(&self) -> DurableOperationState {
48        Projection {
49            intent: self.intent.clone(),
50            grant: self.grant.clone(),
51            dispatch: self.dispatch.clone(),
52            attempt: self.attempt.clone(),
53            receipt: self.receipt.clone(),
54        }
55        .state()
56    }
57}
58
59/// Journal-backed owner of durable operation intent, dispatch, and raw receipts.
60pub struct OperationService<B: JournalBackend> {
61    journal: Journal<Arc<B>>,
62    lease: Option<Lease>,
63}
64
65impl<B: JournalBackend> OperationService<B> {
66    /// Creates a service over one backend value without acquiring authority.
67    pub fn new(backend: B) -> Self {
68        Self::from_shared(Arc::new(backend))
69    }
70
71    /// Creates a service over a shared backend, supporting crash/reopen tests.
72    pub fn from_shared(backend: Arc<B>) -> Self {
73        Self {
74            journal: Journal::new(backend),
75            lease: None,
76        }
77    }
78
79    /// Acquires a fresh fenced journal lease after start or recovery.
80    pub fn resume(&mut self) -> Result<(), OperationError> {
81        self.lease = Some(self.journal.acquire_lease()?);
82        Ok(())
83    }
84
85    /// Reconstructs one operation solely from a verified journal snapshot.
86    pub fn state(
87        &self,
88        operation: &OperationId,
89    ) -> Result<Option<DurableOperationState>, OperationError> {
90        let projections = project(self.journal.verified_snapshot()?)?;
91        Ok(projections.get(operation).map(Projection::state))
92    }
93
94    /// Reconstructs the complete durable record from one verified snapshot.
95    pub fn record(
96        &self,
97        operation: &OperationId,
98    ) -> Result<Option<OperationRecord>, OperationError> {
99        let projections = project(self.journal.verified_snapshot()?)?;
100        Ok(projections.get(operation).map(Projection::record))
101    }
102
103    /// Persists intent and dispatch, calls the performer once, then persists its receipt.
104    ///
105    /// If recovery finds a durable dispatch, this method returns that state and
106    /// never calls the performer. Missing acknowledgement likewise returns the
107    /// durable dispatch state; it is not converted into failure or retry authority.
108    pub fn execute(
109        &mut self,
110        intent: &OperationIntent,
111        grant: &OperationGrant,
112        attempt: &OperationAttempt,
113        performer: &mut dyn OperationPerformer,
114    ) -> Result<DurableOperationState, OperationError> {
115        self.lease.as_ref().ok_or(OperationError::NotResumed)?;
116        intent.verify()?;
117        grant.verify()?;
118        attempt.verify()?;
119        if grant.operation != intent.id {
120            return Err(OperationError::GrantMismatch);
121        }
122        if attempt.operation != intent.id {
123            return Err(OperationError::AttemptMismatch);
124        }
125
126        let projections = project(self.journal.verified_snapshot()?)?;
127        match projections.get(intent.id()) {
128            Some(existing) if existing.intent != *intent => {
129                return Err(OperationError::ContradictoryIntent);
130            }
131            Some(existing) if existing.dispatch.is_some() => return Ok(existing.state()),
132            Some(existing) if existing.grant != *grant => {
133                return Err(OperationError::GrantMismatch);
134            }
135            Some(_) => {}
136            None => self.append(
137                "intent-persisted",
138                vec![intent.canonical_datum(), grant.canonical_datum()],
139            )?,
140        }
141
142        let dispatch =
143            OperationDispatch::new(intent.id.clone(), grant.id.clone(), attempt.id.clone())?;
144        self.append(
145            "dispatched",
146            vec![dispatch.canonical_datum(), attempt.canonical_datum()],
147        )?;
148        let raw = match performer.perform(&dispatch) {
149            PerformerResponse::Receipt(raw) => raw,
150            PerformerResponse::AcknowledgementMissing => {
151                return Ok(DurableOperationState::Dispatched {
152                    intent: intent.id.clone(),
153                    dispatch: dispatch.id,
154                });
155            }
156        };
157        let receipt = PerformerReceipt::new(dispatch.id.clone(), raw)?;
158        self.append("receipt-persisted", vec![receipt.canonical_datum()])?;
159        Ok(DurableOperationState::ReceiptPersisted {
160            dispatch: dispatch.id,
161            receipt: receipt.id,
162        })
163    }
164
165    fn append(&self, kind: &'static str, datums: Vec<Datum>) -> Result<(), OperationError> {
166        let lease = self.lease.as_ref().ok_or(OperationError::NotResumed)?;
167        let objects = datums
168            .into_iter()
169            .map(JournalObject::from_datum)
170            .collect::<Result<Vec<_>, _>>()?;
171        let payloads = objects.iter().map(|object| object.id.clone()).collect();
172        let expected = self.journal.head()?;
173        let sequence = expected
174            .as_ref()
175            .map_or(Some(0), |head| head.sequence.checked_add(1))
176            .ok_or(OperationError::SequenceExhausted)?;
177        let entry = JournalEntry::new(
178            sequence,
179            expected.as_ref().map(|head| head.entry.clone()),
180            Symbol::qualified("operation", kind),
181            payloads,
182        );
183        self.journal
184            .publish(lease, expected.as_ref(), objects, vec![entry])?;
185        Ok(())
186    }
187}
188
189#[derive(Clone)]
190struct Projection {
191    intent: OperationIntent,
192    grant: OperationGrant,
193    dispatch: Option<OperationDispatch>,
194    attempt: Option<OperationAttempt>,
195    receipt: Option<PerformerReceipt>,
196}
197
198impl Projection {
199    fn state(&self) -> DurableOperationState {
200        match (&self.dispatch, &self.receipt) {
201            (Some(dispatch), Some(receipt)) => DurableOperationState::ReceiptPersisted {
202                dispatch: dispatch.id.clone(),
203                receipt: receipt.id.clone(),
204            },
205            (Some(dispatch), None) => DurableOperationState::Dispatched {
206                intent: self.intent.id.clone(),
207                dispatch: dispatch.id.clone(),
208            },
209            (None, None) => DurableOperationState::IntentPersisted {
210                intent: self.intent.id.clone(),
211            },
212            (None, Some(_)) => unreachable!("projection construction refuses receipt first"),
213        }
214    }
215
216    fn record(&self) -> OperationRecord {
217        OperationRecord {
218            intent: self.intent.clone(),
219            grant: self.grant.clone(),
220            attempt: self.attempt.clone(),
221            dispatch: self.dispatch.clone(),
222            receipt: self.receipt.clone(),
223        }
224    }
225}
226
227fn project(
228    snapshot: VerifiedSnapshot,
229) -> Result<BTreeMap<OperationId, Projection>, OperationError> {
230    let mut operations = BTreeMap::<OperationId, Projection>::new();
231    for entry in snapshot.entries() {
232        if entry.kind == Symbol::qualified("operation", "intent-persisted") {
233            require_payload_count(entry, 2)?;
234            let intent =
235                OperationIntent::from_datum(snapshot_datum(&snapshot, &entry.payloads[0])?)?;
236            let grant = OperationGrant::from_datum(snapshot_datum(&snapshot, &entry.payloads[1])?)?;
237            if grant.operation != intent.id {
238                return Err(OperationError::GrantMismatch);
239            }
240            if operations
241                .insert(
242                    intent.id.clone(),
243                    Projection {
244                        intent,
245                        grant,
246                        dispatch: None,
247                        attempt: None,
248                        receipt: None,
249                    },
250                )
251                .is_some()
252            {
253                return Err(OperationError::DuplicateIntent);
254            }
255        } else if entry.kind == Symbol::qualified("operation", "dispatched") {
256            require_payload_count(entry, 2)?;
257            let dispatch =
258                OperationDispatch::from_datum(snapshot_datum(&snapshot, &entry.payloads[0])?)?;
259            let attempt =
260                OperationAttempt::from_datum(snapshot_datum(&snapshot, &entry.payloads[1])?)?;
261            let projection = operations
262                .get_mut(&dispatch.operation)
263                .ok_or(OperationError::InvalidTransition("dispatch before intent"))?;
264            if projection.dispatch.is_some()
265                || dispatch.grant != projection.grant.id
266                || attempt.operation != projection.intent.id
267                || dispatch.attempt != attempt.id
268            {
269                return Err(OperationError::InvalidTransition("conflicting dispatch"));
270            }
271            projection.dispatch = Some(dispatch);
272            projection.attempt = Some(attempt);
273        } else if entry.kind == Symbol::qualified("operation", "receipt-persisted") {
274            require_payload_count(entry, 1)?;
275            let receipt =
276                PerformerReceipt::from_datum(snapshot_datum(&snapshot, &entry.payloads[0])?)?;
277            let projection = operations
278                .values_mut()
279                .find(|projection| {
280                    projection
281                        .dispatch
282                        .as_ref()
283                        .is_some_and(|dispatch| dispatch.id == receipt.dispatch)
284                })
285                .ok_or(OperationError::InvalidTransition("receipt before dispatch"))?;
286            if projection.receipt.replace(receipt).is_some() {
287                return Err(OperationError::InvalidTransition("duplicate receipt"));
288            }
289        }
290    }
291    Ok(operations)
292}
293fn snapshot_datum<'a>(
294    snapshot: &'a VerifiedSnapshot,
295    id: &ContentId,
296) -> Result<&'a Datum, OperationError> {
297    snapshot
298        .datum(id)
299        .ok_or(OperationError::NonCanonical("missing operation payload"))
300}
301
302fn require_payload_count(entry: &JournalEntry, expected: usize) -> Result<(), OperationError> {
303    if entry.payloads.len() != expected {
304        return Err(OperationError::NonCanonical("operation event payloads"));
305    }
306    Ok(())
307}
sim_lib_operation_gate/operation_service.rs

sim_lib_operation_gate/
operation_service.rs
