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

harmos_runtime/state/
history.rs
