harmos_runtime/state/
history.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::collections::{BTreeMap, VecDeque};
18use std::fmt;
19use std::future::Future;
20use std::num::NonZeroUsize;
21use std::pin::Pin;
22use std::sync::Arc;
23use std::task::{Context, Poll};
24// Third Party Imports
25use futures_core::Stream;
26use tokio::sync::{mpsc, oneshot, watch};
27use tracing::Instrument;
28// Local Imports
29use super::TrackedState;
30use super::entry::{At, Entry, Link, Position, PrincipalId};
31use super::envelope::{Envelope, Recorded};
32use super::transaction::dispatch::Dispatch;
33use super::transaction::{DecodeError, Inverting};
34use crate::Application;
35use crate::error::Error;
36use crate::supervision::executor::InContext;
37
38/* === Constants === */
39
40/* === Macros === */
41
42/* === Type Aliases === */
43
44type Predicate<A> = Arc<dyn Fn(&<A as Application>::Transaction) -> bool + Send + Sync>;
45
46type Step<A, X> = Pin<Box<dyn Future<Output = (Parts<A>, Result<Vec<Entry<X>>, Error>)> + Send>>;
47
48/* === Structs === */
49
50pub struct Origin<S> {
51    pub value: S,
52    pub position: Position,
53}
54
55pub(crate) struct Reads<A: Application> {
56    sender: mpsc::Sender<ReadRequest<A>>,
57}
58
59pub struct Focus<A: Application> {
60    predicate: Option<Predicate<A>>,
61}
62
63#[derive(Clone, Copy, Debug, PartialEq, Eq)]
64pub struct Reversals {
65    pub undoable: bool,
66    pub redoable: bool,
67}
68
69pub(crate) struct Undoable<A: Application> {
70    pub(crate) principal: PrincipalId,
71    pub(crate) forward: Arc<A::Transaction>,
72    pub(crate) inverse: Option<Arc<A::Transaction>>,
73    pub(crate) undone: Option<Position>,
74}
75
76pub(crate) struct Undos<A: Application> {
77    pub(crate) entries: BTreeMap<Position, Undoable<A>>,
78}
79
80pub(crate) struct Tail<A: Application> {
81    pub(crate) entries: VecDeque<Arc<Entry<Recorded<A>>>>,
82    capacity: usize,
83    floor: Position,
84}
85
86pub(crate) struct SharedPage<A: Application> {
87    pub(crate) entries: Vec<Arc<Entry<Recorded<A>>>>,
88    pub(crate) reaches: Position,
89}
90
91struct Parts<A: Application> {
92    trace: crate::supervision::executor::SpanContext,
93    tracked: Arc<TrackedState<A>>,
94    frontier: Position,
95    doorbell: watch::Receiver<Position>,
96}
97
98struct Watching<A: Application, X> {
99    parts: Option<Parts<A>>,
100    step: Option<Step<A, X>>,
101    pending: VecDeque<Entry<X>>,
102    done: bool,
103}
104
105/* === Enums === */
106
107pub(crate) enum ReadRequest<A: Application> {
108    Entries {
109        trace: crate::supervision::executor::SpanContext,
110        after: Position,
111        through: Position,
112        reply: oneshot::Sender<Result<Vec<Entry<Recorded<A>>>, Error>>,
113    },
114    State {
115        trace: crate::supervision::executor::SpanContext,
116        origin: Origin<A::State>,
117        at: Position,
118        reply: oneshot::Sender<Result<At<A::State>, Error>>,
119    },
120}
121
122/* === Traits === */
123
124pub(crate) trait ReplayPayload<A: Application>: Sized {
125    fn prepared(&self) -> Option<crate::state::Prepared> {
126        None
127    }
128
129    fn forward(&self) -> Option<Arc<A::Transaction>>;
130    fn restore(
131        envelope: Envelope<Self>, catalog: Option<&Dispatch<A, Self>>, state: &mut A::State,
132    ) -> Result<Option<Recorded<A>>, DecodeError>;
133}
134
135/* === Implementations === */
136
137impl<S> Origin<S> {
138    pub fn initial(value: S) -> Self {
139        Self { value, position: Position::ORIGIN }
140    }
141
142    pub fn at(value: S, position: Position) -> Self {
143        Self { value, position }
144    }
145}
146
147impl<S> From<S> for Origin<S> {
148    fn from(value: S) -> Self {
149        Self::initial(value)
150    }
151}
152
153impl<S> From<At<S>> for Origin<S> {
154    fn from(origin: At<S>) -> Self {
155        Self::at(origin.value, origin.position)
156    }
157}
158
159impl<A: Application> Reads<A> {
160    pub(crate) fn channel(capacity: usize) -> (Self, mpsc::Receiver<ReadRequest<A>>) {
161        let (sender, receiver) = mpsc::channel(capacity.max(1));
162
163        (Self { sender }, receiver)
164    }
165
166    #[tracing::instrument(target = "harmos::state", name = "state.entries", level = "debug", skip_all)]
167    pub(crate) async fn entries(&self, after: Position, through: Position) -> Result<Vec<Entry<Recorded<A>>>, Error> {
168        let (reply, answer) = oneshot::channel();
169        self.sender
170            .send(ReadRequest::Entries {
171                after,
172                through,
173                reply,
174                trace: crate::supervision::executor::SpanContext::new(tracing::Span::current(), tracing::Span::none()),
175            })
176            .await
177            .map_err(|_| Error::BeyondTail)?;
178
179        answer.await.map_err(|_| Error::Unavailable)?
180    }
181
182    #[tracing::instrument(target = "harmos::state", name = "state.state", level = "debug", skip_all)]
183    pub(crate) async fn state(&self, origin: Origin<A::State>, at: Position) -> Result<At<A::State>, Error> {
184        if origin.position > at {
185            return Err(Error::OriginAhead);
186        }
187        if origin.position == at {
188            return Ok(At { value: origin.value, position: at });
189        }
190
191        let (reply, answer) = oneshot::channel();
192        self.sender
193            .send(ReadRequest::State {
194                origin,
195                at,
196                reply,
197                trace: crate::supervision::executor::SpanContext::new(tracing::Span::current(), tracing::Span::none()),
198            })
199            .await
200            .map_err(|_| Error::BeyondTail)?;
201
202        answer.await.map_err(|_| Error::Unavailable)?
203    }
204}
205
206impl<A: Application> Clone for Reads<A> {
207    fn clone(&self) -> Self {
208        Self { sender: self.sender.clone() }
209    }
210}
211
212impl<A: Application> Focus<A> {
213    pub fn all() -> Self {
214        Self { predicate: None }
215    }
216
217    pub fn matching(predicate: impl Fn(&A::Transaction) -> bool + Send + Sync + 'static) -> Self {
218        Self { predicate: Some(Arc::new(predicate)) }
219    }
220
221    fn admits(&self, transaction: &A::Transaction) -> bool {
222        self.predicate.as_ref().is_none_or(|predicate| predicate(transaction))
223    }
224}
225
226impl<A: Application> Clone for Focus<A> {
227    fn clone(&self) -> Self {
228        Self { predicate: self.predicate.clone() }
229    }
230}
231
232impl<A: Application> fmt::Debug for Focus<A> {
233    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
234        formatter.write_str(match self.predicate {
235            Some(_) => "Focus::matching(..)",
236            None => "Focus::all()",
237        })
238    }
239}
240
241impl<A: Application> Undos<A> {
242    pub(crate) fn record(
243        &mut self, position: Position, principal: Option<&PrincipalId>, link: Option<Link>, forward: &Arc<A::Transaction>,
244        inverse: Option<A::Transaction>,
245    ) {
246        match link {
247            Some(Link::UndoOf(original)) => {
248                if let Some(entry) = self.entries.get_mut(&original) {
249                    entry.undone = Some(position);
250                }
251            }
252            Some(Link::RedoOf(original)) => {
253                if let Some(entry) = self.entries.get_mut(&original) {
254                    entry.undone = None;
255                    // The redo re-applied the change, so the inverse captured
256                    // against the state it just left is the one that undoes it
257                    // next time.
258                    entry.inverse = inverse.map(Arc::new);
259                }
260            }
261            None => {
262                let Some(principal) = principal else {
263                    return;
264                };
265                // A new forward commit truncates this principal's redo branch:
266                // the present has diverged from the future those entries would
267                // restore, so redoing onto it would be a lie. The truncation is
268                // personal — undo is per principal, and another principal's
269                // redo is still guarded by the inverse's re-check on commit.
270                self.entries.retain(|_, entry| entry.principal != *principal || entry.undone.is_none());
271                self.entries.insert(
272                    position,
273                    Undoable { principal: principal.clone(), forward: Arc::clone(forward), inverse: inverse.map(Arc::new), undone: None },
274                );
275            }
276        }
277    }
278
279    pub(crate) fn undoable(&self, principal: &PrincipalId, focus: &Focus<A>) -> Option<(Position, Arc<A::Transaction>)> {
280        for (position, entry) in self.entries.iter().rev() {
281            if entry.principal != *principal {
282                continue;
283            }
284            if entry.undone.is_some() || !focus.admits(&entry.forward) {
285                continue;
286            }
287            let inverse = entry.inverse.as_ref()?;
288            return Some((*position, Arc::clone(inverse)));
289        }
290        None
291    }
292
293    pub(crate) fn redoable(&self, principal: &PrincipalId, focus: &Focus<A>) -> Option<(Position, Arc<A::Transaction>)> {
294        self.entries
295            .iter()
296            .filter(|(_, entry)| entry.principal == *principal && focus.admits(&entry.forward))
297            .filter_map(|(position, entry)| entry.undone.map(|at| (at, *position, Arc::clone(&entry.forward))))
298            .max_by_key(|(at, _position, _forward)| *at)
299            .map(|(_at, position, forward)| (position, forward))
300    }
301
302    pub(crate) fn reversals(&self, principal: &PrincipalId, focus: &Focus<A>) -> Reversals {
303        Reversals { undoable: self.undoable(principal, focus).is_some(), redoable: self.redoable(principal, focus).is_some() }
304    }
305}
306
307impl<A: Application> Default for Undos<A> {
308    fn default() -> Self {
309        Self { entries: BTreeMap::new() }
310    }
311}
312
313impl<A: Application> Tail<A> {
314    pub(crate) fn new(origin: Position, capacity: usize) -> Self {
315        Self { entries: VecDeque::new(), capacity: capacity.max(1), floor: origin }
316    }
317
318    pub(crate) fn recovered(frontier: Position, capacity: usize, entries: Vec<Arc<Entry<Recorded<A>>>>) -> Self {
319        if entries.is_empty() {
320            return Self::new(frontier, capacity);
321        }
322        let floor = entries.first().map_or(frontier, |entry| Position(entry.position.0.saturating_sub(1)));
323
324        Self { entries: entries.into(), capacity: capacity.max(1), floor }
325    }
326
327    pub(crate) fn push_group(&mut self, entries: Vec<Entry<Recorded<A>>>) -> Vec<Arc<Entry<Recorded<A>>>> {
328        while self.entries.len() + entries.len() > self.capacity && !self.entries.is_empty() {
329            let evicted = self.entries.pop_front().expect("nonempty tail");
330            let frontier = evicted.metadata.commit.map_or(evicted.position, |(_, end)| end);
331            self.floor = frontier;
332            while self.entries.front().is_some_and(|entry| entry.position <= frontier) {
333                self.entries.pop_front();
334            }
335        }
336        let entries: Vec<_> = entries.into_iter().map(Arc::new).collect();
337        self.entries.extend(entries.iter().cloned());
338        entries
339    }
340
341    pub(crate) fn splits_commit(&self, position: Position) -> bool {
342        self.entries.iter().any(|entry| entry.metadata.commit.is_some_and(|(first, end)| first <= position && position < end))
343    }
344
345    pub(crate) fn since(&self, after: Position) -> Result<Vec<Arc<Entry<Recorded<A>>>>, Error> {
346        if after < self.floor {
347            return Err(Error::BeyondTail);
348        }
349
350        Ok(self.entries.iter().filter(|entry| entry.position > after).map(Arc::clone).collect())
351    }
352
353    pub(crate) fn scan<X: Clone + 'static>(&self, after: Position) -> Result<(Vec<Entry<X>>, Position), Error> {
354        self.scan_through(after, Position(u64::MAX))
355    }
356
357    pub(crate) fn scan_through<X: Clone + 'static>(&self, after: Position, through: Position) -> Result<(Vec<Entry<X>>, Position), Error> {
358        let mut frontier = after;
359        let mut found = Vec::new();
360        loop {
361            let page = self.shared_page(frontier, through, NonZeroUsize::new(1024).unwrap())?;
362            if page.entries.is_empty() {
363                break;
364            }
365            frontier = page.reaches;
366            for entry in page.entries {
367                if let Some(value) = entry.value.select::<X>() {
368                    found.push(Entry { prepared: None, position: entry.position, metadata: entry.metadata.clone(), value });
369                }
370            }
371        }
372
373        Ok((found, frontier))
374    }
375
376    #[tracing::instrument(target = "harmos::state", name = "state.shared_page", level = "debug", skip_all)]
377    pub(crate) fn shared_page(&self, after: Position, through: Position, max_rows: NonZeroUsize) -> Result<SharedPage<A>, Error> {
378        if after < self.floor {
379            return Err(Error::BeyondTail);
380        }
381        let start = self.entries.partition_point(|entry| entry.position <= after);
382        let count = max_rows.get().min(self.entries.len() - start);
383        let mut entries = Vec::with_capacity(count);
384        let mut reaches = after;
385        for entry in self.entries.iter().skip(start).take(count) {
386            if entry.position > through {
387                break;
388            }
389            reaches = entry.position;
390            entries.push(Arc::clone(entry));
391        }
392        Ok(SharedPage { entries, reaches })
393    }
394}
395
396impl<A: Application, X> Unpin for Watching<A, X> {}
397
398impl<A, X> Stream for Watching<A, X>
399where
400    A: Application,
401    X: Clone + Send + 'static,
402{
403    type Item = Result<Entry<X>, Error>;
404
405    fn poll_next(self: Pin<&mut Self>, context: &mut Context<'_>) -> Poll<Option<Self::Item>> {
406        let watching = self.get_mut();
407
408        if let Some(entry) = watching.pending.pop_front() {
409            return Poll::Ready(Some(Ok(entry)));
410        }
411        if watching.done {
412            return Poll::Ready(None);
413        }
414
415        let mut step = match watching.step.take() {
416            Some(step) => step,
417            None => match watching.parts.take() {
418                Some(parts) => {
419                    let dispatcher = parts.trace.dispatcher.clone();
420                    let span = tracing::dispatcher::with_default(
421                        &dispatcher,
422                        || tracing::trace_span!(target: "harmos::state", parent: &parts.trace.span, "state.watch_batch", after = parts.frontier.0),
423                    );
424                    Box::pin(next(parts).instrument(span).in_context(dispatcher))
425                }
426                None => {
427                    watching.done = true;
428                    return Poll::Ready(None);
429                }
430            },
431        };
432
433        match step.as_mut().poll(context) {
434            Poll::Pending => {
435                watching.step = Some(step);
436                Poll::Pending
437            }
438            Poll::Ready((parts, Ok(found))) => {
439                watching.parts = Some(parts);
440                watching.pending.extend(found);
441                match watching.pending.pop_front() {
442                    Some(entry) => Poll::Ready(Some(Ok(entry))),
443                    None => {
444                        watching.done = true;
445                        Poll::Ready(None)
446                    }
447                }
448            }
449            Poll::Ready((parts, Err(refusal))) => {
450                watching.parts = Some(parts);
451                watching.done = true;
452                Poll::Ready(Some(Err(refusal)))
453            }
454        }
455    }
456}
457
458/* === Public Functions === */
459
460#[tracing::instrument(target = "harmos::state", name = "state.replay", level = "debug", skip_all)]
461pub(crate) fn replay<A, W>(
462    catalog: Option<&Dispatch<A, W>>, recorded: Vec<Envelope<W>>, state: &mut At<A::State>, undos: &mut Undos<A>, inverses: Inverting<A>,
463    tail: usize,
464) -> Result<Vec<Arc<Entry<Recorded<A>>>>, DecodeError>
465where
466    A: Application,
467    W: ReplayPayload<A>,
468{
469    let mut redelivered = VecDeque::new();
470
471    for envelope in recorded {
472        let prepared = envelope.payload.prepared();
473        let position = envelope.position;
474        let metadata = envelope.metadata.clone();
475        let link = metadata.link;
476        let principal = metadata.principal.clone();
477        let forward = envelope.payload.forward();
478        let inverse = forward.as_ref().and_then(|change| inverses(change, &state.value));
479        let recovered = W::restore(envelope, catalog, &mut state.value)?;
480
481        if let Some(forward) = &forward {
482            undos.record(position, principal.as_ref(), link, forward, inverse);
483        }
484        if let Some(value) = recovered
485            && tail != 0
486        {
487            // Keep a complete newest group, even when it exceeds the target window.
488            let first = metadata.commit.map_or(position, |(first, _)| first);
489            while redelivered.len() >= tail && redelivered.front().is_some_and(|entry: &Arc<Entry<Recorded<A>>>| entry.position < first) {
490                let end = redelivered.front().map(|entry| entry.metadata.commit.map_or(entry.position, |(_, end)| end)).expect("nonempty");
491                while redelivered.front().is_some_and(|entry| entry.position <= end) {
492                    redelivered.pop_front();
493                }
494            }
495            redelivered.push_back(Arc::new(Entry { prepared, position, metadata, value }));
496        }
497        state.position = position;
498    }
499
500    Ok(redelivered.into())
501}
502
503pub(crate) fn watching<A, X>(tracked: Arc<TrackedState<A>>, after: Position) -> impl Stream<Item = Result<Entry<X>, Error>>
504where
505    A: Application,
506    X: Clone + Send + 'static,
507{
508    let doorbell = tracked.applied.subscribe();
509
510    Watching {
511        parts: Some(Parts {
512            tracked,
513            frontier: after,
514            doorbell,
515            trace: crate::supervision::executor::SpanContext::new(tracing::Span::current(), tracing::Span::none()),
516        }),
517        step: None,
518        pending: VecDeque::new(),
519        done: false,
520    }
521}
522
523/* === Private Functions === */
524
525async fn next<A, X>(mut parts: Parts<A>) -> (Parts<A>, Result<Vec<Entry<X>>, Error>)
526where
527    A: Application,
528    X: Clone + Send + 'static,
529{
530    loop {
531        if parts.tracked.poisoned() {
532            return (parts, Err(Error::Unavailable));
533        }
534
535        let scanned = {
536            let tail = parts.tracked.tail.read().await;
537            tail.scan::<X>(parts.frontier)
538        };
539
540        match scanned {
541            Err(refusal) => return (parts, Err(refusal)),
542            Ok((found, frontier)) => {
543                parts.frontier = frontier;
544                if !found.is_empty() {
545                    return (parts, Ok(found));
546                }
547            }
548        }
549
550        if parts.doorbell.changed().await.is_err() {
551            return (parts, Err(Error::Unavailable));
552        }
553    }
554}
555
556/* === Tests === */
harmos_runtime/state/history.rs

harmos_runtime/state/
history.rs
