harmos_runtime/state/
manager.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12/* === Re-Exports === */
13
14/* === Imports === */
15
16// Standard Library Imports
17use std::sync::atomic::{AtomicBool, Ordering};
18use std::sync::{Arc, Weak};
19// Third Party Imports
20use futures_core::Stream;
21use tokio::sync::{RwLock, mpsc, oneshot, watch};
22use tokio::task::JoinHandle;
23use tracing::Instrument;
24// Local Imports
25use super::ManagedState;
26use super::transaction::TransactionError;
27use super::writer::{Admission, Direction, Refusal, Request};
28use super::{
29    At, CausationId, CorrelationId, DecodeError, DefinitionId, Entry, Error, Focus, Inverting, Position, PreparationError, Prepared,
30    PrincipalId, ReadRequest, Reads, Receipt, Recorded, RequestKey, Reversals, Tail, Transaction, Undos, capture, dispatch, history,
31    writer,
32};
33use crate::Application;
34
35/* === Constants === */
36
37/* === Macros === */
38
39/* === Type Aliases === */
40
41/* === Structs === */
42
43pub(crate) struct TrackedState<A: Application> {
44    pub(crate) state: RwLock<At<A::State>>,
45    pub(crate) tail: RwLock<Tail<A>>,
46    pub(super) applied: watch::Sender<Position>,
47    pub(super) stored: watch::Receiver<Position>,
48    pub(super) poisoned: AtomicBool,
49    pub(crate) sealing: watch::Sender<bool>,
50    pub(super) capture_failed: Arc<AtomicBool>,
51    pub(super) capture_progress: Option<watch::Receiver<Result<Position, crate::state::CaptureError>>>,
52    pub(super) catalog: Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>,
53    pub(super) reads: Reads<A>,
54    pub(super) record_preparation: Arc<tokio::sync::Semaphore>,
55    pub(super) clock: fn() -> super::Timestamp,
56}
57
58#[derive(Clone, Debug, Default)]
59pub(crate) struct Context {
60    pub(super) principal: Option<PrincipalId>,
61    pub(super) correlation: Option<CorrelationId>,
62    pub(super) causation: Option<CausationId>,
63}
64
65pub struct StateManager<A: Application> {
66    handles: Arc<()>,
67    record_schemas: Arc<crate::records::Registry>,
68    writer: mpsc::Sender<writer::QueuedRequest<A>>,
69    context: Context,
70    pub(super) tracked: Arc<TrackedState<A>>,
71}
72
73pub(crate) struct Feed<A: Application> {
74    pub(super) tracked: Arc<TrackedState<A>>,
75}
76
77/* === Enums === */
78
79/* === Traits === */
80
81/* === Implementations === */
82
83impl<A: Application> TrackedState<A> {
84    pub(super) fn unavailable<E>(&self) -> Error<E> {
85        if self.capture_failed.load(Ordering::Acquire) { Error::CaptureFailed } else { Error::Unavailable }
86    }
87
88    pub(super) fn poison(&self) {
89        self.poisoned.store(true, Ordering::Release);
90    }
91
92    pub(crate) fn poisoned(&self) -> bool {
93        self.poisoned.load(Ordering::Acquire)
94    }
95}
96
97impl<A: Application> StateManager<A> {
98    pub fn state(&self) -> ManagedState<A> {
99        ManagedState::new(Arc::clone(&self.tracked))
100    }
101
102    pub(crate) fn handle_probe(&self) -> Weak<()> {
103        Arc::downgrade(&self.handles)
104    }
105
106    pub(crate) fn record_schemas(&self) -> Arc<crate::records::Registry> {
107        Arc::clone(&self.record_schemas)
108    }
109
110    // The private boot boundary receives already-owned journal components.
111    #[allow(clippy::too_many_arguments)]
112    pub(crate) fn start(
113        origin: At<A::State>, inverses: Inverting<A>, undos: Undos<A>, requests: usize, tail: Tail<A>,
114        durability: Option<watch::Receiver<Position>>, catalog: Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>,
115        capture: Option<capture::Declaration<A>>, storage: Option<crate::storage::StorageManager<crate::storage::ErasedCatalog>>,
116        schema_settings: Arc<crate::config::SchemaSettings>, clock: fn() -> super::Timestamp,
117    ) -> (Self, JoinHandle<()>, mpsc::Receiver<ReadRequest<A>>) {
118        let position = origin.position;
119        let (sender, receiver) = mpsc::channel(requests.max(1));
120        let (reads, history) = Reads::channel(requests);
121        let (applied, _) = watch::channel(position);
122        let (durable, bridged) = watch::channel(position);
123        let stored = durability.unwrap_or(bridged);
124        let capture_failed = Arc::new(AtomicBool::new(false));
125        let capture = capture.map(|capture| capture.start(capture_failed.clone(), position));
126        let tracked = Arc::new(TrackedState {
127            state: RwLock::new(origin),
128            tail: RwLock::new(tail),
129            applied,
130            stored,
131            poisoned: AtomicBool::new(false),
132            sealing: watch::channel(false).0,
133            capture_failed,
134            capture_progress: capture.as_ref().map(capture::Capture::progress),
135            record_preparation: Arc::new(tokio::sync::Semaphore::new(1)),
136            catalog,
137            reads,
138            clock,
139        });
140        let running = tokio::spawn(writer::run(receiver, Arc::clone(&tracked), durable, inverses, undos, capture));
141
142        (
143            Self {
144                handles: Arc::new(()),
145                record_schemas: Arc::new(crate::records::Registry::new(storage, schema_settings)),
146                writer: sender,
147                context: Context::default(),
148                tracked,
149            },
150            running,
151            history,
152        )
153    }
154
155    #[tracing::instrument(target = "harmos::state", name = "state.commit", level = "debug", skip_all)]
156    pub async fn commit<T>(&self, transaction: T) -> Result<Receipt, Error<T::Error>>
157    where
158        T: Transaction<A::State> + Into<A::Transaction>,
159        T::Error: From<TransactionError<A>>,
160    {
161        self.request(None, crate::operations::Response::new(transaction.into())).await
162    }
163
164    #[tracing::instrument(target = "harmos::state", name = "state.commit_keyed", level = "debug", skip_all)]
165    pub async fn commit_keyed<T>(&self, key: RequestKey, transaction: T) -> Result<Receipt, Error<T::Error>>
166    where
167        T: Transaction<A::State> + Into<A::Transaction>,
168        T::Error: From<TransactionError<A>>,
169    {
170        self.request(Some(key), crate::operations::Response::new(transaction.into())).await
171    }
172
173    /// Commits a transaction projected from an external owner without copying its payload.
174    ///
175    /// After queue acceptance, cancelling this caller does not release the owner:
176    /// writer, capture and history references retain it until their last transaction
177    /// handle drops. Allocations made while applying the transaction remain separate.
178    pub async fn commit_owned(
179        &self, transaction: crate::operations::Response<A::Transaction>,
180    ) -> Result<Receipt, Error<<A::Transaction as Transaction<A::State>>::Error>> {
181        self.request(None, transaction).await
182    }
183
184    /// Commits an externally owned transaction with the existing request-key deduplication.
185    ///
186    /// Accepted work retains its owner through writer, capture and history references
187    /// despite caller cancellation. A duplicate key releases the unused supplied owner;
188    /// this does not account for allocations made while applying a transaction.
189    pub async fn commit_keyed_owned(
190        &self, key: RequestKey, transaction: crate::operations::Response<A::Transaction>,
191    ) -> Result<Receipt, Error<<A::Transaction as Transaction<A::State>>::Error>> {
192        self.request(Some(key), transaction).await
193    }
194
195    pub(crate) fn record_resources(&self) -> (Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>, Arc<tokio::sync::Semaphore>) {
196        (Arc::clone(&self.tracked.catalog), Arc::clone(&self.tracked.record_preparation))
197    }
198
199    #[tracing::instrument(target = "harmos::state", name = "state.append_prepared", level = "debug", skip_all)]
200    pub(crate) async fn append_prepared(
201        &self, value: Recorded<A>, prepared: Prepared, permit: tokio::sync::OwnedSemaphorePermit,
202    ) -> Result<Receipt, Error> {
203        let (reply, answer) = oneshot::channel();
204        let request = Request::Record {
205            value,
206            prepared,
207            preparation: permit,
208            context: self.context.clone(),
209            reply,
210        };
211
212        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
213        let permit = match waiting.await {
214            Ok(permit) => permit,
215            Err(_) => return Err(self.unavailable()),
216        };
217        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
218        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
219        answer.await.map_err(|_| self.unavailable())?.map_err(preparation)
220    }
221
222    pub fn decode_transaction(&self, id: &DefinitionId, version: u32, bytes: &[u8]) -> Result<A::Transaction, DecodeError> {
223        self.tracked.catalog.decode_transaction(id, version, bytes)
224    }
225
226    pub async fn query<X>(&self, after: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
227    where
228        X: Clone + Send + 'static,
229    {
230        self.query_through(after, self.applied()).await
231    }
232
233    #[tracing::instrument(target = "harmos::state", name = "state.query_through", level = "debug", skip_all)]
234    pub async fn query_through<X>(&self, after: Position, through: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
235    where
236        X: Clone + Send + 'static,
237    {
238        if self.tracked.poisoned() {
239            return vec![Err(self.unavailable())].into_iter();
240        }
241
242        if through > self.applied() {
243            return vec![Err(Error::FrontierAhead)].into_iter();
244        }
245        let scanned = {
246            let tail = self.tracked.tail.read().await;
247            tail.scan_through::<X>(after, through)
248        };
249
250        match scanned {
251            Ok((found, _)) => found.into_iter().map(Ok).collect::<Vec<_>>().into_iter(),
252            Err(Error::BeyondTail) => match self.tracked.reads.entries(after, through).await {
253                Ok(entries) => entries
254                    .into_iter()
255                    .filter_map(|entry| {
256                        entry.value.select::<X>().map(|value| {
257                            Ok(Entry {
258                                prepared: None,
259                                position: entry.position,
260                                metadata: entry.metadata,
261                                value,
262                            })
263                        })
264                    })
265                    .collect::<Vec<_>>()
266                    .into_iter(),
267                Err(refusal) => vec![Err(refusal)].into_iter(),
268            },
269            Err(refusal) => vec![Err(refusal)].into_iter(),
270        }
271    }
272
273    #[tracing::instrument(target = "harmos::state", name = "state.watch", level = "debug", skip_all)]
274    pub fn watch<X>(&self, after: Position) -> impl Stream<Item = Result<Entry<X>, Error>>
275    where
276        X: Clone + Send + 'static,
277    {
278        history::watching(Arc::clone(&self.tracked), after)
279    }
280
281    pub fn applied(&self) -> Position {
282        *self.tracked.applied.borrow()
283    }
284
285    pub(crate) fn progress(&self) -> watch::Receiver<Position> {
286        self.tracked.applied.subscribe()
287    }
288
289    pub(crate) fn seal(&self) {
290        self.tracked.sealing.send_replace(true);
291    }
292
293    pub(crate) fn feed(&self) -> Feed<A> {
294        Feed { tracked: Arc::clone(&self.tracked) }
295    }
296
297    pub fn stored(&self) -> Position {
298        *self.tracked.stored.borrow()
299    }
300
301    #[tracing::instrument(target = "harmos::state", name = "state.wait_until_stored", level = "debug", skip_all)]
302    pub async fn wait_until_stored(&self, at: Position) {
303        let mut stored = self.tracked.stored.clone();
304
305        while *stored.borrow_and_update() < at {
306            if stored.changed().await.is_err() {
307                return;
308            }
309        }
310    }
311
312    #[tracing::instrument(target = "harmos::state", name = "state.undo", level = "debug", skip_all)]
313    pub async fn undo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
314        self.revert(principal, focus, Direction::Undo).await
315    }
316
317    #[tracing::instrument(target = "harmos::state", name = "state.redo", level = "debug", skip_all)]
318    pub async fn redo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
319        self.revert(principal, focus, Direction::Redo).await
320    }
321
322    #[tracing::instrument(target = "harmos::state", name = "state.reversals", level = "debug", skip_all)]
323    pub async fn reversals(&self, principal: PrincipalId, focus: Focus<A>) -> Result<At<Reversals>, Error> {
324        if self.tracked.poisoned() {
325            return Err(self.unavailable());
326        }
327
328        let (reply, answer) = oneshot::channel();
329        let request = Request::Reversals { principal, focus, reply };
330
331        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
332        let permit = match waiting.await {
333            Ok(permit) => permit,
334            Err(_) => return Err(self.unavailable()),
335        };
336        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
337        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
338
339        answer.await.map_err(|_| self.unavailable())
340    }
341
342    pub fn as_principal(&self, principal: PrincipalId) -> Self {
343        let mut scoped = self.clone();
344        scoped.context.principal = Some(principal);
345        scoped
346    }
347
348    pub fn caused_by(&self, cause: Position) -> Self {
349        let mut scoped = self.clone();
350        scoped.context.correlation = Some(CorrelationId(cause));
351        scoped.context.causation = Some(CausationId(cause));
352        scoped
353    }
354
355    #[tracing::instrument(target = "harmos::state", name = "state.revert", level = "debug", skip_all)]
356    async fn revert(&self, principal: PrincipalId, focus: Focus<A>, direction: Direction) -> Result<Receipt, Error<TransactionError<A>>> {
357        let (reply, answer) = oneshot::channel();
358        let request = Request::Revert {
359            principal,
360            focus,
361            direction,
362            context: self.context.clone(),
363            reply,
364        };
365
366        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
367        let permit = match waiting.await {
368            Ok(permit) => permit,
369            Err(_) => return Err(self.unavailable()),
370        };
371        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
372        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
373        match answer.await {
374            Ok(Ok(receipt)) => Ok(receipt),
375            Ok(Err(Refusal::Preparation(refusal))) => Err(preparation(refusal)),
376            Ok(Err(Refusal::Conflict(refusal))) => Err(match direction {
377                Direction::Undo => Error::UndoConflict(refusal),
378                Direction::Redo => Error::RedoConflict(refusal),
379            }),
380            Ok(Err(Refusal::Nothing)) => Err(match direction {
381                Direction::Undo => Error::NothingToUndo,
382                Direction::Redo => Error::NothingToRedo,
383            }),
384            Err(_) => Err(self.unavailable()),
385        }
386    }
387
388    #[tracing::instrument(target = "harmos::state", name = "state.wait_until_captured", level = "debug", skip_all)]
389    pub async fn wait_until_captured(&self, through: Position) -> Result<(), crate::state::CaptureError> {
390        let Some(mut progress) = self.tracked.capture_progress.clone() else {
391            return Ok(());
392        };
393        loop {
394            if (*progress.borrow_and_update())? >= through {
395                return Ok(());
396            }
397            if progress.changed().await.is_err() {
398                return Err(crate::state::CaptureError::Failed);
399            }
400        }
401    }
402
403    pub fn capture_status(&self) -> Result<(), crate::state::CaptureError> {
404        if self.tracked.capture_failed.load(Ordering::Acquire) { Err(crate::state::CaptureError::Failed) } else { Ok(()) }
405    }
406
407    fn unavailable<E>(&self) -> Error<E> {
408        if self.capture_status().is_err() { Error::CaptureFailed } else { Error::Unavailable }
409    }
410
411    async fn request<E>(
412        &self, key: Option<RequestKey>, transaction: crate::operations::Response<A::Transaction>,
413    ) -> Result<Receipt, Error<E>>
414    where
415        E: From<TransactionError<A>>,
416    {
417        let (reply, answer) = oneshot::channel();
418        let request = Request::Commit {
419            transaction,
420            key,
421            context: self.context.clone(),
422            reply,
423        };
424
425        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
426        let permit = match waiting.await {
427            Ok(permit) => permit,
428            Err(_) => return Err(self.unavailable()),
429        };
430        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
431        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
432        match answer.await {
433            Ok(Ok(receipt)) => Ok(receipt),
434            Ok(Err(Admission::Rejected(refusal))) => Err(Error::Rejected(E::from(refusal))),
435            Ok(Err(Admission::Preparation(refusal))) => Err(preparation(refusal)),
436            Err(_) => Err(self.unavailable()),
437        }
438    }
439}
440
441impl<A: Application> Feed<A> {
442    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
443    #[tracing::instrument(target = "harmos::state", name = "state.shared_page", level = "debug", skip_all)]
444    pub(crate) async fn shared_page(
445        &self, after: Position, through: Position, max_rows: std::num::NonZeroUsize, bytes: impl Fn(&A::Record) -> usize,
446    ) -> Result<history::SharedPage<A>, Error> {
447        self.tracked.tail.read().await.shared_page_sized(after, through, max_rows, bytes)
448    }
449
450    #[tracing::instrument(target = "harmos::state", name = "state.since", level = "debug", skip_all)]
451    pub(crate) async fn since(&self, after: Position) -> Result<Vec<Arc<Entry<Recorded<A>>>>, Error> {
452        let tail = self.tracked.tail.read().await;
453
454        tail.since(after)
455    }
456}
457
458impl<A: Application> Clone for StateManager<A> {
459    fn clone(&self) -> Self {
460        Self {
461            handles: Arc::clone(&self.handles),
462            record_schemas: Arc::clone(&self.record_schemas),
463            writer: self.writer.clone(),
464            context: self.context.clone(),
465            tracked: Arc::clone(&self.tracked),
466        }
467    }
468}
469
470/* === Public Functions === */
471
472/* === Private Functions === */
473
474fn preparation<E>(error: PreparationError) -> Error<E> {
475    match error {
476        PreparationError::CaptureFailed => Error::CaptureFailed,
477        error => Error::Preparation(error),
478    }
479}
480
481/* === Tests === */
harmos_runtime/state/manager.rs

harmos_runtime/state/
manager.rs
