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;
23// Local Imports
24use super::ManagedState;
25use super::transaction::TransactionError;
26use super::writer::{Admission, Direction, Refusal, Request};
27use super::{
28    At, CausationId, CorrelationId, DecodeError, DefinitionId, Entry, Error, Focus, Inverting, Position, PreparationError, Prepared,
29    PrincipalId, ReadRequest, Reads, Receipt, Recorded, RequestKey, Reversals, Tail, Transaction, Undos, capture, dispatch, history,
30    writer,
31};
32use crate::Application;
33
34/* === Constants === */
35
36/* === Macros === */
37
38/* === Type Aliases === */
39
40/* === Structs === */
41
42pub(crate) struct TrackedState<A: Application> {
43    pub(crate) state: RwLock<At<A::State>>,
44    pub(crate) tail: RwLock<Tail<A>>,
45    pub(super) applied: watch::Sender<Position>,
46    pub(super) stored: watch::Receiver<Position>,
47    pub(super) poisoned: AtomicBool,
48    pub(crate) sealing: watch::Sender<bool>,
49    pub(super) capture_failed: Arc<AtomicBool>,
50    pub(super) capture_progress: Option<watch::Receiver<Result<Position, crate::state::CaptureError>>>,
51    pub(super) catalog: Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>,
52    pub(super) reads: Reads<A>,
53    pub(super) record_preparation: Arc<tokio::sync::Semaphore>,
54    pub(super) clock: fn() -> super::Timestamp,
55}
56
57#[derive(Clone, Debug, Default)]
58pub(crate) struct Context {
59    pub(super) principal: Option<PrincipalId>,
60    pub(super) correlation: Option<CorrelationId>,
61    pub(super) causation: Option<CausationId>,
62}
63
64pub struct StateManager<A: Application> {
65    handles: Arc<()>,
66    record_schemas: Arc<crate::records::Registry>,
67    writer: mpsc::Sender<Request<A>>,
68    context: Context,
69    pub(super) tracked: Arc<TrackedState<A>>,
70}
71
72pub(crate) struct Feed<A: Application> {
73    pub(super) tracked: Arc<TrackedState<A>>,
74}
75
76/* === Enums === */
77
78/* === Traits === */
79
80/* === Implementations === */
81
82impl<A: Application> TrackedState<A> {
83    pub(super) fn unavailable<E>(&self) -> Error<E> {
84        if self.capture_failed.load(Ordering::Acquire) { Error::CaptureFailed } else { Error::Unavailable }
85    }
86
87    pub(super) fn poison(&self) {
88        self.poisoned.store(true, Ordering::Release);
89    }
90
91    pub(crate) fn poisoned(&self) -> bool {
92        self.poisoned.load(Ordering::Acquire)
93    }
94}
95
96impl<A: Application> StateManager<A> {
97    pub fn state(&self) -> ManagedState<A> {
98        ManagedState::new(Arc::clone(&self.tracked))
99    }
100
101    pub(crate) fn handle_probe(&self) -> Weak<()> {
102        Arc::downgrade(&self.handles)
103    }
104
105    pub(crate) fn record_schemas(&self) -> Arc<crate::records::Registry> {
106        Arc::clone(&self.record_schemas)
107    }
108
109    // The private boot boundary receives already-owned journal components.
110    #[allow(clippy::too_many_arguments)]
111    pub(crate) fn start(
112        origin: At<A::State>, inverses: Inverting<A>, undos: Undos<A>, requests: usize, tail: Tail<A>,
113        durability: Option<watch::Receiver<Position>>, catalog: Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>,
114        capture: Option<capture::Declaration<A>>, storage: Option<crate::storage::StorageManager<crate::storage::ErasedCatalog>>,
115        clock: fn() -> super::Timestamp,
116    ) -> (Self, JoinHandle<()>, mpsc::Receiver<ReadRequest<A>>) {
117        let position = origin.position;
118        let (sender, receiver) = mpsc::channel(requests.max(1));
119        let (reads, history) = Reads::channel(requests);
120        let (applied, _) = watch::channel(position);
121        let (durable, bridged) = watch::channel(position);
122        let stored = durability.unwrap_or(bridged);
123        let capture_failed = Arc::new(AtomicBool::new(false));
124        let capture = capture.map(|capture| capture.start(capture_failed.clone(), position));
125        let tracked = Arc::new(TrackedState {
126            state: RwLock::new(origin),
127            tail: RwLock::new(tail),
128            applied,
129            stored,
130            poisoned: AtomicBool::new(false),
131            sealing: watch::channel(false).0,
132            capture_failed,
133            capture_progress: capture.as_ref().map(capture::Capture::progress),
134            record_preparation: Arc::new(tokio::sync::Semaphore::new(1)),
135            catalog,
136            reads,
137            clock,
138        });
139        let running = tokio::spawn(writer::run(receiver, Arc::clone(&tracked), durable, inverses, undos, capture));
140
141        (
142            Self {
143                handles: Arc::new(()),
144                record_schemas: Arc::new(crate::records::Registry::new(storage)),
145                writer: sender,
146                context: Context::default(),
147                tracked,
148            },
149            running,
150            history,
151        )
152    }
153
154    pub async fn commit<T>(&self, transaction: T) -> Result<Receipt, Error<T::Error>>
155    where
156        T: Transaction<A::State> + Into<A::Transaction>,
157        T::Error: From<TransactionError<A>>,
158    {
159        self.request(None, transaction.into()).await
160    }
161
162    pub async fn commit_keyed<T>(&self, key: RequestKey, transaction: T) -> Result<Receipt, Error<T::Error>>
163    where
164        T: Transaction<A::State> + Into<A::Transaction>,
165        T::Error: From<TransactionError<A>>,
166    {
167        self.request(Some(key), transaction.into()).await
168    }
169
170    pub(crate) fn record_resources(&self) -> (Arc<dispatch::Dispatch<A, crate::storage::Payload<A>>>, Arc<tokio::sync::Semaphore>) {
171        (Arc::clone(&self.tracked.catalog), Arc::clone(&self.tracked.record_preparation))
172    }
173
174    pub(crate) async fn append_prepared(
175        &self, value: Recorded<A>, prepared: Prepared, permit: tokio::sync::OwnedSemaphorePermit,
176    ) -> Result<Receipt, Error> {
177        let (reply, answer) = oneshot::channel();
178        let request = Request::Record { value, prepared, preparation: permit, context: self.context.clone(), reply };
179
180        if self.writer.send(request).await.is_err() {
181            return Err(self.unavailable());
182        }
183        answer.await.map_err(|_| self.unavailable())?.map_err(preparation)
184    }
185
186    pub fn decode_transaction(&self, id: &DefinitionId, version: u32, bytes: &[u8]) -> Result<A::Transaction, DecodeError> {
187        self.tracked.catalog.decode_transaction(id, version, bytes)
188    }
189
190    pub async fn query<X>(&self, after: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
191    where
192        X: Clone + Send + 'static,
193    {
194        self.query_through(after, self.applied()).await
195    }
196
197    pub async fn query_through<X>(&self, after: Position, through: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
198    where
199        X: Clone + Send + 'static,
200    {
201        if self.tracked.poisoned() {
202            return vec![Err(self.unavailable())].into_iter();
203        }
204
205        if through > self.applied() {
206            return vec![Err(Error::FrontierAhead)].into_iter();
207        }
208        let scanned = {
209            let tail = self.tracked.tail.read().await;
210            tail.scan_through::<X>(after, through)
211        };
212
213        match scanned {
214            Ok((found, _)) => found.into_iter().map(Ok).collect::<Vec<_>>().into_iter(),
215            Err(Error::BeyondTail) => match self.tracked.reads.entries(after, through).await {
216                Ok(entries) => entries
217                    .into_iter()
218                    .filter_map(|entry| {
219                        entry
220                            .value
221                            .select::<X>()
222                            .map(|value| Ok(Entry { prepared: None, position: entry.position, metadata: entry.metadata, value }))
223                    })
224                    .collect::<Vec<_>>()
225                    .into_iter(),
226                Err(refusal) => vec![Err(refusal)].into_iter(),
227            },
228            Err(refusal) => vec![Err(refusal)].into_iter(),
229        }
230    }
231
232    pub fn watch<X>(&self, after: Position) -> impl Stream<Item = Result<Entry<X>, Error>>
233    where
234        X: Clone + Send + 'static,
235    {
236        history::watching(Arc::clone(&self.tracked), after)
237    }
238
239    pub fn applied(&self) -> Position {
240        *self.tracked.applied.borrow()
241    }
242
243    pub(crate) fn progress(&self) -> watch::Receiver<Position> {
244        self.tracked.applied.subscribe()
245    }
246
247    pub(crate) fn seal(&self) {
248        self.tracked.sealing.send_replace(true);
249    }
250
251    pub(crate) fn feed(&self) -> Feed<A> {
252        Feed { tracked: Arc::clone(&self.tracked) }
253    }
254
255    pub fn stored(&self) -> Position {
256        *self.tracked.stored.borrow()
257    }
258
259    pub async fn wait_until_stored(&self, at: Position) {
260        let mut stored = self.tracked.stored.clone();
261
262        while *stored.borrow_and_update() < at {
263            if stored.changed().await.is_err() {
264                return;
265            }
266        }
267    }
268
269    pub async fn undo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
270        self.revert(principal, focus, Direction::Undo).await
271    }
272
273    pub async fn redo(&self, principal: PrincipalId, focus: Focus<A>) -> Result<Receipt, Error<TransactionError<A>>> {
274        self.revert(principal, focus, Direction::Redo).await
275    }
276
277    pub async fn reversals(&self, principal: PrincipalId, focus: Focus<A>) -> Result<At<Reversals>, Error> {
278        if self.tracked.poisoned() {
279            return Err(self.unavailable());
280        }
281
282        let (reply, answer) = oneshot::channel();
283        let request = Request::Reversals { principal, focus, reply };
284
285        if self.writer.send(request).await.is_err() {
286            return Err(self.unavailable());
287        }
288
289        answer.await.map_err(|_| self.unavailable())
290    }
291
292    pub fn as_principal(&self, principal: PrincipalId) -> Self {
293        let mut scoped = self.clone();
294        scoped.context.principal = Some(principal);
295        scoped
296    }
297
298    pub fn caused_by(&self, cause: Position) -> Self {
299        let mut scoped = self.clone();
300        scoped.context.correlation = Some(CorrelationId(cause));
301        scoped.context.causation = Some(CausationId(cause));
302        scoped
303    }
304
305    async fn revert(&self, principal: PrincipalId, focus: Focus<A>, direction: Direction) -> Result<Receipt, Error<TransactionError<A>>> {
306        let (reply, answer) = oneshot::channel();
307        let request = Request::Revert { principal, focus, direction, context: self.context.clone(), reply };
308
309        if self.writer.send(request).await.is_err() {
310            return Err(self.unavailable());
311        }
312        match answer.await {
313            Ok(Ok(receipt)) => Ok(receipt),
314            Ok(Err(Refusal::Preparation(refusal))) => Err(preparation(refusal)),
315            Ok(Err(Refusal::Conflict(refusal))) => Err(match direction {
316                Direction::Undo => Error::UndoConflict(refusal),
317                Direction::Redo => Error::RedoConflict(refusal),
318            }),
319            Ok(Err(Refusal::Nothing)) => Err(match direction {
320                Direction::Undo => Error::NothingToUndo,
321                Direction::Redo => Error::NothingToRedo,
322            }),
323            Err(_) => Err(self.unavailable()),
324        }
325    }
326
327    pub async fn wait_until_captured(&self, through: Position) -> Result<(), crate::state::CaptureError> {
328        let Some(mut progress) = self.tracked.capture_progress.clone() else {
329            return Ok(());
330        };
331        loop {
332            if (*progress.borrow_and_update())? >= through {
333                return Ok(());
334            }
335            if progress.changed().await.is_err() {
336                return Err(crate::state::CaptureError::Failed);
337            }
338        }
339    }
340
341    pub fn capture_status(&self) -> Result<(), crate::state::CaptureError> {
342        if self.tracked.capture_failed.load(Ordering::Acquire) { Err(crate::state::CaptureError::Failed) } else { Ok(()) }
343    }
344
345    fn unavailable<E>(&self) -> Error<E> {
346        if self.capture_status().is_err() { Error::CaptureFailed } else { Error::Unavailable }
347    }
348
349    async fn request<E>(&self, key: Option<RequestKey>, transaction: A::Transaction) -> Result<Receipt, Error<E>>
350    where
351        E: From<TransactionError<A>>,
352    {
353        let (reply, answer) = oneshot::channel();
354        let request = Request::Commit { transaction, key, context: self.context.clone(), reply };
355
356        if self.writer.send(request).await.is_err() {
357            return Err(self.unavailable());
358        }
359        match answer.await {
360            Ok(Ok(receipt)) => Ok(receipt),
361            Ok(Err(Admission::Rejected(refusal))) => Err(Error::Rejected(E::from(refusal))),
362            Ok(Err(Admission::Preparation(refusal))) => Err(preparation(refusal)),
363            Err(_) => Err(self.unavailable()),
364        }
365    }
366}
367
368impl<A: Application> Feed<A> {
369    #[cfg(any(feature = "plugin", feature = "plugin-guest", feature = "plugin-sidecar"))]
370    pub(crate) async fn shared_page(
371        &self, after: Position, through: Position, max_rows: std::num::NonZeroUsize,
372    ) -> Result<history::SharedPage<A>, Error> {
373        self.tracked.tail.read().await.shared_page(after, through, max_rows)
374    }
375
376    pub(crate) async fn since(&self, after: Position) -> Result<Vec<Arc<Entry<Recorded<A>>>>, Error> {
377        let tail = self.tracked.tail.read().await;
378
379        tail.since(after)
380    }
381}
382
383impl<A: Application> Clone for StateManager<A> {
384    fn clone(&self) -> Self {
385        Self {
386            handles: Arc::clone(&self.handles),
387            record_schemas: Arc::clone(&self.record_schemas),
388            writer: self.writer.clone(),
389            context: self.context.clone(),
390            tracked: Arc::clone(&self.tracked),
391        }
392    }
393}
394
395/* === Public Functions === */
396
397/* === Private Functions === */
398
399fn preparation<E>(error: PreparationError) -> Error<E> {
400    match error {
401        PreparationError::CaptureFailed => Error::CaptureFailed,
402        error => Error::Preparation(error),
403    }
404}
405
406/* === Tests === */
harmos_runtime/state/manager.rs

harmos_runtime/state/
manager.rs
