harmos_runtime/storage/
attachments.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::future::Future;
18use std::pin::Pin;
19use std::sync::Arc;
20// Third Party Imports
21use tokio::sync::watch;
22// Local Imports
23use crate::state::dispatch::{Decode, Dispatch, Restored};
24use crate::state::{Feed, ReplayPayload};
25use crate::storage::consumer::{Attached, JournalConsumer, Recording, Series, Serving, StreamConsumer};
26use crate::{
27    Application, DefinitionId, Error, runtime::Incompatible, state::DecodeError, state::Entry, state::Envelope, state::Kind,
28    state::Position, state::Recorded, state::Transaction,
29};
30
31/* === Constants === */
32
33/* === Macros === */
34
35/* === Type Aliases === */
36
37/* === Structs === */
38
39pub struct Payload<A: Application> {
40    pub prepared: Option<crate::state::Prepared>,
41    pub id: DefinitionId,
42    pub version: u32,
43    pub value: Recorded<A>,
44
45    // Spelled out rather than aliased: the census is the list of names an
46    // author learns, and this one belongs to a field rather than beside them.
47    #[allow(clippy::type_complexity)]
48    pub apply: Option<Box<dyn FnOnce(&mut A::State) + Send>>,
49}
50
51pub struct Recovery<A: Application> {
52    pub commits: Vec<(Position, Position)>,
53    pub frontier: Position,
54    pub entries: Vec<Envelope<Payload<A>>>,
55    pub encoded: Vec<(Position, Vec<u8>)>,
56}
57
58pub(crate) struct Attachments<A: Application> {
59    pub(crate) capture: Option<crate::state::capture::Declaration<A>>,
60    attached: Vec<Box<dyn Attached<A>>>,
61    streams: Vec<Box<dyn Recording<A>>>,
62    series: Option<Box<dyn Serving<A>>>,
63    recovery: Option<Box<dyn Recovering<A>>>,
64    snapshot: Option<Box<dyn Attached<A>>>,
65    retention: Option<Horizon>,
66    tail: usize,
67}
68
69pub(crate) struct Consumers<A: Application> {
70    attached: Vec<Box<dyn Attached<A>>>,
71    streams: Vec<Box<dyn Recording<A>>>,
72    series: Option<Box<dyn Serving<A>>>,
73    recovery: Option<Box<dyn Recovering<A>>>,
74    snapshot: Option<Box<dyn Attached<A>>>,
75    retention: Option<Horizon>,
76    durable: Option<watch::Sender<Position>>,
77    recovered: Vec<Arc<Entry<Recorded<A>>>>,
78    stored: Vec<(Position, Vec<u8>)>,
79    frontier: Position,
80    source_frontier: Position,
81    snapshot_frontier: Position,
82    pruned: Position,
83}
84
85/* === Enums === */
86
87#[derive(Clone, Copy, Debug, PartialEq, Eq)]
88pub enum Horizon {
89    BeneathSnapshot,
90}
91
92/* === Traits === */
93
94pub trait Source<A: Application>: JournalConsumer<A> {
95    fn load(&mut self) -> impl Future<Output = Result<Recovery<A>, Incompatible>> + Send;
96
97    fn read(&mut self, _after: Position, _through: Position) -> impl Future<Output = Result<Recovery<A>, Error>> + Send {
98        async { Err(Error::BeyondTail) }
99    }
100
101    fn prune(&mut self, _beneath: Position) -> impl Future<Output = ()> + Send {
102        async {}
103    }
104}
105
106pub(crate) trait Recovering<A: Application>: Attached<A> {
107    fn load(&mut self) -> Pin<Box<dyn Future<Output = Result<Recovery<A>, Incompatible>> + Send + '_>>;
108
109    fn read(&mut self, after: Position, through: Position) -> Pin<Box<dyn Future<Output = Result<Recovery<A>, Error>> + Send + '_>>;
110
111    fn prune(&mut self, beneath: Position) -> Pin<Box<dyn Future<Output = ()> + Send + '_>>;
112}
113
114/* === Implementations === */
115
116impl<A: Application> Default for Recovery<A> {
117    fn default() -> Self {
118        Self { commits: Vec::new(), frontier: Position::ORIGIN, entries: Vec::new(), encoded: Vec::new() }
119    }
120}
121
122impl<A: Application, S: Source<A>> Recovering<A> for S {
123    fn load(&mut self) -> Pin<Box<dyn Future<Output = Result<Recovery<A>, Incompatible>> + Send + '_>> {
124        Box::pin(Source::load(self))
125    }
126
127    fn read(&mut self, after: Position, through: Position) -> Pin<Box<dyn Future<Output = Result<Recovery<A>, Error>> + Send + '_>> {
128        Box::pin(Source::read(self, after, through))
129    }
130
131    fn prune(&mut self, beneath: Position) -> Pin<Box<dyn Future<Output = ()> + Send + '_>> {
132        Box::pin(Source::prune(self, beneath))
133    }
134}
135
136impl<A: Application> ReplayPayload<A> for Payload<A> {
137    fn prepared(&self) -> Option<crate::state::Prepared> {
138        self.prepared.clone()
139    }
140
141    fn forward(&self) -> Option<Arc<A::Transaction>> {
142        match &self.value {
143            Recorded::Transaction(transaction) => Some(Arc::clone(transaction)),
144            Recorded::Record(_) | Recorded::System(_) => None,
145        }
146    }
147
148    fn restore(
149        envelope: Envelope<Self>, catalog: Option<&Dispatch<A, Self>>, state: &mut A::State,
150    ) -> Result<Option<Recorded<A>>, DecodeError> {
151        let system = match &envelope.payload.value {
152            Recorded::System(fact) => Some(fact.clone()),
153            Recorded::Transaction(_) | Recorded::Record(_) => None,
154        };
155        if let Some(fact) = system {
156            return Ok(Some(Recorded::System(fact)));
157        }
158        let forward = match &envelope.payload.value {
159            Recorded::Transaction(transaction) => Some(Arc::clone(transaction)),
160            Recorded::Record(_) | Recorded::System(_) => None,
161        };
162
163        let Some(catalog) = catalog else {
164            let Payload { value, apply, .. } = envelope.payload;
165            return Ok(match value {
166                Recorded::Transaction(transaction) => {
167                    match apply {
168                        Some(historical) => historical(state),
169                        None => transaction.apply(state),
170                    }
171                    Some(Recorded::Transaction(transaction))
172                }
173                Recorded::Record(record) => Some(Recorded::Record(record)),
174                Recorded::System(fact) => Some(Recorded::System(fact)),
175            });
176        };
177
178        Ok(match catalog.restore(envelope, state)? {
179            Restored::Applied => forward.map(Recorded::Transaction),
180            Restored::Recorded(record) => Some(Recorded::Record(record)),
181            Restored::Skipped => None,
182        })
183    }
184}
185
186impl<A: Application> Attachments<A> {
187    pub(crate) fn has_recovery_owner(&self) -> bool {
188        self.capture.is_some() || self.recovery.is_some() || self.snapshot.is_some() || self.retention.is_some()
189    }
190
191    pub(crate) fn new() -> Self {
192        Self {
193            capture: None,
194            attached: Vec::new(),
195            streams: Vec::new(),
196            series: None,
197            recovery: None,
198            snapshot: None,
199            retention: None,
200            tail: 1024,
201        }
202    }
203
204    pub(crate) fn require_capture(
205        &mut self, limits: crate::state::CaptureLimits, consumer: impl crate::state::JournalCapture<A>,
206    ) -> Result<(), crate::state::CaptureError> {
207        if limits.pending_bytes == 0
208            || limits.max_group_bytes == 0
209            || limits.max_group_bytes > limits.pending_bytes
210            || limits.pending_bytes as usize > tokio::sync::Semaphore::MAX_PERMITS
211        {
212            return Err(crate::state::CaptureError::InvalidLimits);
213        }
214        self.capture = Some(crate::state::capture::Declaration::Callback(crate::state::capture::CallbackDeclaration {
215            limits,
216            consumer: Box::new(consumer),
217        }));
218
219        Ok(())
220    }
221
222    pub(crate) fn attach(&mut self, consumer: impl JournalConsumer<A>) {
223        self.attached.push(Box::new(consumer));
224    }
225
226    pub(crate) fn attach_streams(&mut self, consumer: impl StreamConsumer<A>) {
227        self.streams.push(Box::new(consumer));
228    }
229
230    pub(crate) fn series(&mut self, series: impl Series<A>) {
231        if let Some(demoted) = self.series.replace(Box::new(series)) {
232            self.streams.push(demoted);
233        }
234    }
235
236    pub(crate) fn recover(&mut self, source: impl Source<A>) {
237        if let Some(demoted) = self.recovery.replace(Box::new(source)) {
238            self.attached.push(demoted);
239        }
240    }
241
242    pub(crate) fn snapshot(&mut self, snapshot: impl JournalConsumer<A>) {
243        if let Some(demoted) = self.snapshot.replace(Box::new(snapshot)) {
244            self.attached.push(demoted);
245        }
246    }
247
248    pub(crate) fn retain(&mut self, horizon: Horizon) {
249        self.retention = Some(horizon);
250    }
251
252    pub(crate) fn tail(&mut self, entries: usize) {
253        self.tail = entries.max(1);
254    }
255
256    pub(crate) fn tail_window(&self) -> Result<usize, Incompatible> {
257        if self.retention == Some(Horizon::BeneathSnapshot) && self.snapshot.is_none() {
258            return Err(Incompatible::unreadable("retention beneath a snapshot requires a snapshot declaration"));
259        }
260
261        Ok(self.tail)
262    }
263
264    pub(crate) fn supervised(&mut self) -> Consumers<A> {
265        let snapshot_frontier = self.snapshot.as_ref().map_or(Position::ORIGIN, |snapshot| snapshot.resume());
266
267        Consumers {
268            attached: std::mem::take(&mut self.attached),
269            streams: std::mem::take(&mut self.streams),
270            series: self.series.take(),
271            recovery: self.recovery.take(),
272            snapshot: self.snapshot.take(),
273            retention: self.retention,
274            durable: None,
275            recovered: Vec::new(),
276            stored: Vec::new(),
277            frontier: Position::ORIGIN,
278            source_frontier: Position::ORIGIN,
279            snapshot_frontier,
280            pruned: Position::ORIGIN,
281        }
282    }
283}
284
285impl<A: Application> Default for Attachments<A> {
286    fn default() -> Self {
287        Self::new()
288    }
289}
290
291impl<A: Application> Consumers<A> {
292    pub(crate) fn durability(&mut self, at: Position) -> Option<watch::Receiver<Position>> {
293        let (durable, stored) = watch::channel(at);
294        self.durable = self.recovery.as_ref().map(|_| durable);
295
296        self.durable.as_ref().map(|_| stored)
297    }
298
299    pub(crate) async fn load(&mut self) -> Result<Recovery<A>, Incompatible> {
300        let recovery = match &mut self.recovery {
301            Some(source) => source.load().await,
302            None => Ok(Recovery::default()),
303        }?;
304        self.source_frontier = recovery.frontier;
305
306        Ok(recovery)
307    }
308
309    pub(crate) async fn read(&mut self, after: Position, through: Position) -> Result<Recovery<A>, Error> {
310        match &mut self.recovery {
311            Some(source) => source.read(after, through).await,
312            None => Err(Error::BeyondTail),
313        }
314    }
315
316    pub(crate) fn redelivering(&mut self, recovered: Vec<Arc<Entry<Recorded<A>>>>, stored: Vec<(Position, Vec<u8>)>, frontier: Position) {
317        self.recovered = recovered;
318        self.stored = stored;
319        self.frontier = frontier;
320    }
321
322    pub(crate) async fn drain(&mut self, feed: &Feed<A>, applied: Position) -> Position {
323        let mut settled = applied;
324
325        if let Some(source) = &mut self.recovery {
326            let durable = fold(source.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await;
327
328            if let Some(stored) = &self.durable {
329                stored.send_replace(durable);
330            }
331            self.source_frontier = durable;
332            settled = settled.min(durable);
333        }
334        for consumer in &mut self.attached {
335            settled = settled.min(fold(consumer.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await);
336        }
337        if let Some(snapshot) = &mut self.snapshot {
338            let durable = fold(snapshot.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await;
339            self.snapshot_frontier = durable;
340            settled = settled.min(durable);
341        }
342        self.prune().await;
343        self.forget_redelivered();
344
345        settled
346    }
347
348    pub(crate) async fn settle(&mut self, applied: Position) -> Position {
349        let mut settled = applied;
350
351        if let Some(source) = &mut self.recovery {
352            let durable = source.drain(Vec::new(), applied).await;
353
354            if let Some(stored) = &self.durable {
355                stored.send_replace(durable);
356            }
357            self.source_frontier = durable;
358            settled = settled.min(durable);
359        }
360        for consumer in &mut self.attached {
361            settled = settled.min(consumer.drain(Vec::new(), applied).await);
362        }
363        if let Some(snapshot) = &mut self.snapshot {
364            let durable = snapshot.drain(Vec::new(), applied).await;
365            self.snapshot_frontier = durable;
366            settled = settled.min(durable);
367        }
368        self.prune().await;
369
370        settled
371    }
372
373    pub(crate) fn recordings(&mut self) -> impl Iterator<Item = &mut dyn Recording<A>> {
374        self.streams
375            .iter_mut()
376            .map(|recording| recording.as_mut() as &mut dyn Recording<A>)
377            .chain(self.series.iter_mut().map(|series| series.as_mut() as &mut dyn Recording<A>))
378    }
379
380    pub(crate) async fn row_reader(
381        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<Position>, end: std::ops::Bound<Position>,
382        limits: crate::streams::ReadLimits,
383    ) -> Result<Box<dyn crate::streams::RowSource>, crate::streams::ReadError> {
384        match &mut self.series {
385            Some(series) => series.reader(stream, track, start, end, limits).await,
386            None => Err(crate::streams::ReadError::Unserved),
387        }
388    }
389
390    pub(crate) async fn served(
391        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<Position>, end: std::ops::Bound<Position>,
392    ) -> Result<Vec<crate::state::At<Arc<dyn std::any::Any + Send + Sync>>>, crate::streams::BeneathWindow> {
393        match &mut self.series {
394            Some(series) => series.rows(stream, track, start, end).await,
395            None => Err(crate::streams::BeneathWindow::Unserved),
396        }
397    }
398
399    pub(crate) async fn surveyed(
400        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<u64>, end: std::ops::Bound<u64>,
401        budget: usize,
402    ) -> Result<Vec<crate::streams::Bucket>, crate::streams::BeneathWindow> {
403        match &mut self.series {
404            Some(series) => series.buckets(stream, track, start, end, budget).await,
405            None => Err(crate::streams::BeneathWindow::Unserved),
406        }
407    }
408
409    pub(crate) async fn aggregated(
410        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<u64>, end: std::ops::Bound<u64>,
411        kind: crate::Aggregate,
412    ) -> Result<Vec<crate::streams::Aggregated>, crate::streams::BeneathWindow> {
413        match &mut self.series {
414            Some(series) => series.aggregate(stream, track, start, end, kind).await,
415            None => Err(crate::streams::BeneathWindow::Unserved),
416        }
417    }
418
419    pub(crate) async fn sealed(&mut self, stream: crate::streams::Definition, track: &[String]) -> Position {
420        match &mut self.series {
421            Some(series) => series.seal(stream, track).await,
422            None => Position::ORIGIN,
423        }
424    }
425
426    pub(crate) async fn flush(&mut self) {
427        if let Some(source) = &mut self.recovery {
428            source.flush().await;
429        }
430        for consumer in &mut self.attached {
431            consumer.flush().await;
432        }
433        if let Some(snapshot) = &mut self.snapshot {
434            snapshot.flush().await;
435        }
436        for consumer in self.recordings() {
437            consumer.flush().await;
438        }
439    }
440
441    fn forget_redelivered(&mut self) {
442        let passed = self
443            .recovery
444            .iter()
445            .map(|source| source.resume())
446            .chain(self.attached.iter().map(|consumer| consumer.resume()))
447            .chain(self.snapshot.iter().map(|snapshot| snapshot.resume()))
448            .min()
449            .is_none_or(|behind| behind >= self.frontier);
450
451        if passed {
452            self.recovered = Vec::new();
453            self.stored = Vec::new();
454        }
455    }
456
457    async fn prune(&mut self) {
458        let Some(Horizon::BeneathSnapshot) = self.retention else {
459            return;
460        };
461        let beneath = self.snapshot_frontier;
462        if beneath <= self.pruned || self.source_frontier < beneath {
463            return;
464        }
465        let Some(source) = &mut self.recovery else {
466            return;
467        };
468
469        source.prune(beneath).await;
470        self.pruned = beneath;
471    }
472}
473
474/* === Public Functions === */
475
476pub(crate) fn declare<A: Application>(
477    catalog: &mut Dispatch<A, Payload<A>>, kind: Kind, id: DefinitionId, version: u32, decode: Decode<A>,
478) {
479    match kind {
480        Kind::Transaction => catalog.set_transaction(id, version, apply, decode),
481        Kind::Record => catalog.set_record(id, version, retain),
482    }
483}
484
485/* === Private Functions === */
486
487fn apply<A: Application>(payload: Payload<A>, state: &mut A::State) -> Result<(), DecodeError> {
488    let Payload { prepared, id, version, value, apply } = payload;
489
490    match (value, apply) {
491        (Recorded::Transaction(_), Some(historical)) => {
492            historical(state);
493            Ok(())
494        }
495        (Recorded::Transaction(transaction), None) => {
496            if let Some(prepared) = prepared {
497                crate::state::TransactionCatalog::replay(&*transaction, &prepared, state)
498            } else {
499                transaction.apply(state);
500                Ok(())
501            }
502        }
503        (Recorded::Record(_), _) => Err(mismatched(id, version)),
504        (Recorded::System(_), _) => Err(mismatched(id, version)),
505    }
506}
507
508fn retain<A: Application>(payload: Payload<A>) -> Result<A::Record, DecodeError> {
509    match payload.value {
510        Recorded::Record(record) => Ok(record),
511        Recorded::Transaction(_) => Err(mismatched(payload.id, payload.version)),
512        Recorded::System(_) => Err(mismatched(payload.id, payload.version)),
513    }
514}
515
516fn mismatched(id: DefinitionId, version: u32) -> DecodeError {
517    DecodeError::Malformed { id, version }
518}
519
520async fn fold<A: Application>(
521    consumer: &mut dyn Attached<A>, recovered: &[Arc<Entry<Recorded<A>>>], stored: &[(Position, Vec<u8>)], frontier: Position,
522    feed: &Feed<A>, applied: Position,
523) -> Position {
524    let resume = consumer.resume();
525    let Ok(live) = feed.since(resume.max(frontier)).await else {
526        return consumer.drain(Vec::new(), resume).await;
527    };
528    // What is about to be redelivered is forwarded first in its stored form,
529    // so a durable copy writes those bytes instead of re-encoding history.
530    let owed = stored.partition_point(|(position, _)| *position <= resume);
531    if owed < stored.len() {
532        consumer.redelivered(&stored[owed..]);
533    }
534    let mut entries: Vec<Arc<Entry<Recorded<A>>>> = recovered.iter().filter(|entry| entry.position > resume).map(Arc::clone).collect();
535    entries.extend(live);
536
537    consumer.drain(entries, applied).await
538}
539
540/* === Tests === */
harmos_runtime/storage/attachments.rs

harmos_runtime/storage/
attachments.rs
