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

harmos_runtime/state/
history.rs
