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        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)),
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, 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), transaction.into()).await
171    }
172
173    pub(crate) fn record_resources(&self) -> (Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>, Arc<tokio::sync::Semaphore>) {
174        (Arc::clone(&self.tracked.catalog), Arc::clone(&self.tracked.record_preparation))
175    }
176
177    #[tracing::instrument(target = "harmos::state", name = "state.append_prepared", level = "debug", skip_all)]
178    pub(crate) async fn append_prepared(
179        &self, value: Recorded<A>, prepared: Prepared, permit: tokio::sync::OwnedSemaphorePermit,
180    ) -> Result<Receipt, Error> {
181        let (reply, answer) = oneshot::channel();
182        let request = Request::Record { value, prepared, preparation: permit, context: self.context.clone(), reply };
183
184        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
185        let permit = match waiting.await {
186            Ok(permit) => permit,
187            Err(_) => return Err(self.unavailable()),
188        };
189        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
190        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
191        answer.await.map_err(|_| self.unavailable())?.map_err(preparation)
192    }
193
194    pub fn decode_transaction(&self, id: &DefinitionId, version: u32, bytes: &[u8]) -> Result<A::Transaction, DecodeError> {
195        self.tracked.catalog.decode_transaction(id, version, bytes)
196    }
197
198    pub async fn query<X>(&self, after: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
199    where
200        X: Clone + Send + 'static,
201    {
202        self.query_through(after, self.applied()).await
203    }
204
205    #[tracing::instrument(target = "harmos::state", name = "state.query_through", level = "debug", skip_all)]
206    pub async fn query_through<X>(&self, after: Position, through: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
207    where
208        X: Clone + Send + 'static,
209    {
210        if self.tracked.poisoned() {
211            return vec![Err(self.unavailable())].into_iter();
212        }
213
214        if through > self.applied() {
215            return vec![Err(Error::FrontierAhead)].into_iter();
216        }
217        let scanned = {
218            let tail = self.tracked.tail.read().await;
219            tail.scan_through::<X>(after, through)
220        };
221
222        match scanned {
223            Ok((found, _)) => found.into_iter().map(Ok).collect::<Vec<_>>().into_iter(),
224            Err(Error::BeyondTail) => match self.tracked.reads.entries(after, through).await {
225                Ok(entries) => entries
226                    .into_iter()
227                    .filter_map(|entry| {
228                        entry
229                            .value
230                            .select::<X>()
231                            .map(|value| Ok(Entry { prepared: None, position: entry.position, metadata: entry.metadata, value }))
232                    })
233                    .collect::<Vec<_>>()
234                    .into_iter(),
235                Err(refusal) => vec![Err(refusal)].into_iter(),
236            },
237            Err(refusal) => vec![Err(refusal)].into_iter(),
238        }
239    }
240
241    #[tracing::instrument(target = "harmos::state", name = "state.watch", level = "debug", skip_all)]
242    pub fn watch<X>(&self, after: Position) -> impl Stream<Item = Result<Entry<X>, Error>>
243    where
244        X: Clone + Send + 'static,
245    {
246        history::watching(Arc::clone(&self.tracked), after)
247    }
248
249    pub fn applied(&self) -> Position {
250        *self.tracked.applied.borrow()
251    }
252
253    pub(crate) fn progress(&self) -> watch::Receiver<Position> {
254        self.tracked.applied.subscribe()
255    }
256
257    pub(crate) fn seal(&self) {
258        self.tracked.sealing.send_replace(true);
259    }
260
261    pub(crate) fn feed(&self) -> Feed<A> {
262        Feed { tracked: Arc::clone(&self.tracked) }
263    }
264
265    pub fn stored(&self) -> Position {
266        *self.tracked.stored.borrow()
267    }
268
269    #[tracing::instrument(target = "harmos::state", name = "state.wait_until_stored", level = "debug", skip_all)]
270    pub async fn wait_until_stored(&self, at: Position) {
271        let mut stored = self.tracked.stored.clone();
272
273        while *stored.borrow_and_update() < at {
274            if stored.changed().await.is_err() {
275                return;
276            }
277        }
278    }
279
280    #[tracing::instrument(target = "harmos::state", name = "state.undo", level = "debug", skip_all)]
281    pub async fn undo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
282        self.revert(principal, focus, Direction::Undo).await
283    }
284
285    #[tracing::instrument(target = "harmos::state", name = "state.redo", level = "debug", skip_all)]
286    pub async fn redo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
287        self.revert(principal, focus, Direction::Redo).await
288    }
289
290    #[tracing::instrument(target = "harmos::state", name = "state.reversals", level = "debug", skip_all)]
291    pub async fn reversals(&self, principal: PrincipalId, focus: Focus<A>) -> Result<At<Reversals>, Error> {
292        if self.tracked.poisoned() {
293            return Err(self.unavailable());
294        }
295
296        let (reply, answer) = oneshot::channel();
297        let request = Request::Reversals { principal, focus, reply };
298
299        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
300        let permit = match waiting.await {
301            Ok(permit) => permit,
302            Err(_) => return Err(self.unavailable()),
303        };
304        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
305        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
306
307        answer.await.map_err(|_| self.unavailable())
308    }
309
310    pub fn as_principal(&self, principal: PrincipalId) -> Self {
311        let mut scoped = self.clone();
312        scoped.context.principal = Some(principal);
313        scoped
314    }
315
316    pub fn caused_by(&self, cause: Position) -> Self {
317        let mut scoped = self.clone();
318        scoped.context.correlation = Some(CorrelationId(cause));
319        scoped.context.causation = Some(CausationId(cause));
320        scoped
321    }
322
323    #[tracing::instrument(target = "harmos::state", name = "state.revert", level = "debug", skip_all)]
324    async fn revert(&self, principal: PrincipalId, focus: Focus<A>, direction: Direction) -> Result<Receipt, Error<TransactionError<A>>> {
325        let (reply, answer) = oneshot::channel();
326        let request = Request::Revert { principal, focus, direction, context: self.context.clone(), reply };
327
328        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
329        let permit = match waiting.await {
330            Ok(permit) => permit,
331            Err(_) => return Err(self.unavailable()),
332        };
333        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
334        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
335        match answer.await {
336            Ok(Ok(receipt)) => Ok(receipt),
337            Ok(Err(Refusal::Preparation(refusal))) => Err(preparation(refusal)),
338            Ok(Err(Refusal::Conflict(refusal))) => Err(match direction {
339                Direction::Undo => Error::UndoConflict(refusal),
340                Direction::Redo => Error::RedoConflict(refusal),
341            }),
342            Ok(Err(Refusal::Nothing)) => Err(match direction {
343                Direction::Undo => Error::NothingToUndo,
344                Direction::Redo => Error::NothingToRedo,
345            }),
346            Err(_) => Err(self.unavailable()),
347        }
348    }
349
350    #[tracing::instrument(target = "harmos::state", name = "state.wait_until_captured", level = "debug", skip_all)]
351    pub async fn wait_until_captured(&self, through: Position) -> Result<(), crate::state::CaptureError> {
352        let Some(mut progress) = self.tracked.capture_progress.clone() else {
353            return Ok(());
354        };
355        loop {
356            if (*progress.borrow_and_update())? >= through {
357                return Ok(());
358            }
359            if progress.changed().await.is_err() {
360                return Err(crate::state::CaptureError::Failed);
361            }
362        }
363    }
364
365    pub fn capture_status(&self) -> Result<(), crate::state::CaptureError> {
366        if self.tracked.capture_failed.load(Ordering::Acquire) { Err(crate::state::CaptureError::Failed) } else { Ok(()) }
367    }
368
369    fn unavailable<E>(&self) -> Error<E> {
370        if self.capture_status().is_err() { Error::CaptureFailed } else { Error::Unavailable }
371    }
372
373    async fn request<E>(&self, key: Option<RequestKey>, transaction: A::Transaction) -> Result<Receipt, Error<E>>
374    where
375        E: From<TransactionError<A>>,
376    {
377        let (reply, answer) = oneshot::channel();
378        let request = Request::Commit { transaction, key, context: self.context.clone(), reply };
379
380        let waiting = self.writer.reserve().instrument(tracing::debug_span!(target: "harmos::state", "state.admission_wait"));
381        let permit = match waiting.await {
382            Ok(permit) => permit,
383            Err(_) => return Err(self.unavailable()),
384        };
385        let queued = tracing::debug_span!(target: "harmos::state", "state.queue");
386        permit.send((request, crate::supervision::executor::SpanContext::new(tracing::Span::current(), queued)));
387        match answer.await {
388            Ok(Ok(receipt)) => Ok(receipt),
389            Ok(Err(Admission::Rejected(refusal))) => Err(Error::Rejected(E::from(refusal))),
390            Ok(Err(Admission::Preparation(refusal))) => Err(preparation(refusal)),
391            Err(_) => Err(self.unavailable()),
392        }
393    }
394}
395
396impl<A: Application> Feed<A> {
397    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
398    #[tracing::instrument(target = "harmos::state", name = "state.shared_page", level = "debug", skip_all)]
399    pub(crate) async fn shared_page(
400        &self, after: Position, through: Position, max_rows: std::num::NonZeroUsize,
401    ) -> Result<history::SharedPage<A>, Error> {
402        self.tracked.tail.read().await.shared_page(after, through, max_rows)
403    }
404
405    #[tracing::instrument(target = "harmos::state", name = "state.since", level = "debug", skip_all)]
406    pub(crate) async fn since(&self, after: Position) -> Result<Vec<Arc<Entry<Recorded<A>>>>, Error> {
407        let tail = self.tracked.tail.read().await;
408
409        tail.since(after)
410    }
411}
412
413impl<A: Application> Clone for StateManager<A> {
414    fn clone(&self) -> Self {
415        Self {
416            handles: Arc::clone(&self.handles),
417            record_schemas: Arc::clone(&self.record_schemas),
418            writer: self.writer.clone(),
419            context: self.context.clone(),
420            tracked: Arc::clone(&self.tracked),
421        }
422    }
423}
424
425/* === Public Functions === */
426
427/* === Private Functions === */
428
429fn preparation<E>(error: PreparationError) -> Error<E> {
430    match error {
431        PreparationError::CaptureFailed => Error::CaptureFailed,
432        error => Error::Preparation(error),
433    }
434}
435
436/* === Tests === */
harmos_runtime/state/manager.rs

harmos_runtime/state/
manager.rs
