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, Arc<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, Arc<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    #[tracing::instrument(target = "harmos::storage", name = "storage.load", level = "debug", skip_all)]
300    pub(crate) async fn load(&mut self) -> Result<Recovery<A>, Incompatible> {
301        let recovery = match &mut self.recovery {
302            Some(source) => source.load().await,
303            None => Ok(Recovery::default()),
304        }?;
305        self.source_frontier = recovery.frontier;
306
307        Ok(recovery)
308    }
309
310    #[tracing::instrument(target = "harmos::storage", name = "storage.read", level = "debug", skip_all)]
311    pub(crate) async fn read(&mut self, after: Position, through: Position) -> Result<Recovery<A>, Error> {
312        match &mut self.recovery {
313            Some(source) => source.read(after, through).await,
314            None => Err(Error::BeyondTail),
315        }
316    }
317
318    pub(crate) fn redelivering(
319        &mut self, recovered: Vec<Arc<Entry<Recorded<A>>>>, stored: Vec<(Position, Arc<Vec<u8>>)>, frontier: Position,
320    ) {
321        self.recovered = recovered;
322        self.stored = stored;
323        self.frontier = frontier;
324    }
325
326    #[tracing::instrument(target = "harmos::storage", name = "storage.drain", level = "debug", skip_all)]
327    pub(crate) async fn drain(&mut self, feed: &Feed<A>, applied: Position) -> Position {
328        let mut settled = applied;
329
330        if let Some(source) = &mut self.recovery {
331            let durable = fold(source.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await;
332
333            if let Some(stored) = &self.durable {
334                stored.send_replace(durable);
335            }
336            self.source_frontier = durable;
337            settled = settled.min(durable);
338        }
339        for consumer in &mut self.attached {
340            settled = settled.min(fold(consumer.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await);
341        }
342        if let Some(snapshot) = &mut self.snapshot {
343            let durable = fold(snapshot.as_mut(), &self.recovered, &self.stored, self.frontier, feed, applied).await;
344            self.snapshot_frontier = durable;
345            settled = settled.min(durable);
346        }
347        self.prune().await;
348        self.forget_redelivered();
349
350        settled
351    }
352
353    #[tracing::instrument(target = "harmos::storage", name = "storage.settle", level = "debug", skip_all)]
354    pub(crate) async fn settle(&mut self, applied: Position) -> Position {
355        let mut settled = applied;
356
357        if let Some(source) = &mut self.recovery {
358            let durable = source.drain(Vec::new(), applied).await;
359
360            if let Some(stored) = &self.durable {
361                stored.send_replace(durable);
362            }
363            self.source_frontier = durable;
364            settled = settled.min(durable);
365        }
366        for consumer in &mut self.attached {
367            settled = settled.min(consumer.drain(Vec::new(), applied).await);
368        }
369        if let Some(snapshot) = &mut self.snapshot {
370            let durable = snapshot.drain(Vec::new(), applied).await;
371            self.snapshot_frontier = durable;
372            settled = settled.min(durable);
373        }
374        self.prune().await;
375
376        settled
377    }
378
379    pub(crate) fn recordings(&mut self) -> impl Iterator<Item = &mut dyn Recording<A>> {
380        self.streams
381            .iter_mut()
382            .map(|recording| recording.as_mut() as &mut dyn Recording<A>)
383            .chain(self.series.iter_mut().map(|series| series.as_mut() as &mut dyn Recording<A>))
384    }
385
386    #[tracing::instrument(target = "harmos::storage", name = "storage.row_reader", level = "debug", skip_all)]
387    pub(crate) async fn row_reader(
388        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<Position>, end: std::ops::Bound<Position>,
389        limits: crate::streams::ReadLimits,
390    ) -> Result<Box<dyn crate::streams::RowSource>, crate::streams::ReadError> {
391        match &mut self.series {
392            Some(series) => series.reader(stream, track, start, end, limits).await,
393            None => Err(crate::streams::ReadError::Unserved),
394        }
395    }
396
397    #[tracing::instrument(target = "harmos::storage", name = "storage.served", level = "debug", skip_all)]
398    pub(crate) async fn served(
399        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<Position>, end: std::ops::Bound<Position>,
400    ) -> Result<Vec<crate::state::At<Arc<dyn std::any::Any + Send + Sync>>>, crate::streams::BeneathWindow> {
401        match &mut self.series {
402            Some(series) => series.rows(stream, track, start, end).await,
403            None => Err(crate::streams::BeneathWindow::Unserved),
404        }
405    }
406
407    #[tracing::instrument(target = "harmos::storage", name = "storage.surveyed", level = "debug", skip_all)]
408    pub(crate) async fn surveyed(
409        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<u64>, end: std::ops::Bound<u64>,
410        budget: usize,
411    ) -> Result<Vec<crate::streams::Bucket>, crate::streams::BeneathWindow> {
412        match &mut self.series {
413            Some(series) => series.buckets(stream, track, start, end, budget).await,
414            None => Err(crate::streams::BeneathWindow::Unserved),
415        }
416    }
417
418    #[tracing::instrument(target = "harmos::storage", name = "storage.aggregated", level = "debug", skip_all)]
419    pub(crate) async fn aggregated(
420        &mut self, stream: crate::streams::Definition, track: &[String], start: std::ops::Bound<u64>, end: std::ops::Bound<u64>,
421        kind: crate::Aggregate,
422    ) -> Result<Vec<crate::streams::Aggregated>, crate::streams::BeneathWindow> {
423        match &mut self.series {
424            Some(series) => series.aggregate(stream, track, start, end, kind).await,
425            None => Err(crate::streams::BeneathWindow::Unserved),
426        }
427    }
428
429    #[tracing::instrument(target = "harmos::storage", name = "storage.sealed", level = "debug", skip_all)]
430    pub(crate) async fn sealed(&mut self, stream: crate::streams::Definition, track: &[String]) -> Position {
431        match &mut self.series {
432            Some(series) => series.seal(stream, track).await,
433            None => Position::ORIGIN,
434        }
435    }
436
437    #[tracing::instrument(target = "harmos::storage", name = "storage.flush", level = "debug", skip_all)]
438    pub(crate) async fn flush(&mut self) {
439        if let Some(source) = &mut self.recovery {
440            source.flush().await;
441        }
442        for consumer in &mut self.attached {
443            consumer.flush().await;
444        }
445        if let Some(snapshot) = &mut self.snapshot {
446            snapshot.flush().await;
447        }
448        for consumer in self.recordings() {
449            consumer.flush().await;
450        }
451    }
452
453    fn forget_redelivered(&mut self) {
454        let passed = self
455            .recovery
456            .iter()
457            .map(|source| source.resume())
458            .chain(self.attached.iter().map(|consumer| consumer.resume()))
459            .chain(self.snapshot.iter().map(|snapshot| snapshot.resume()))
460            .min()
461            .is_none_or(|behind| behind >= self.frontier);
462
463        if passed {
464            self.recovered = Vec::new();
465            self.stored = Vec::new();
466        }
467    }
468
469    async fn prune(&mut self) {
470        let Some(Horizon::BeneathSnapshot) = self.retention else {
471            return;
472        };
473        let beneath = self.snapshot_frontier;
474        if beneath <= self.pruned || self.source_frontier < beneath {
475            return;
476        }
477        let Some(source) = &mut self.recovery else {
478            return;
479        };
480
481        source.prune(beneath).await;
482        self.pruned = beneath;
483    }
484}
485
486/* === Public Functions === */
487
488pub(crate) fn declare<A: Application>(
489    catalog: &mut Dispatch<A, Payload<A>>, kind: Kind, id: DefinitionId, version: u32, decode: Decode<A>,
490) {
491    match kind {
492        Kind::Transaction => catalog.set_transaction(id, version, apply, decode),
493        Kind::Record => catalog.set_record(id, version, retain),
494    }
495}
496
497/* === Private Functions === */
498
499fn apply<A: Application>(payload: Payload<A>, state: &mut A::State) -> Result<(), DecodeError> {
500    let Payload { prepared, id, version, value, apply } = payload;
501
502    match (value, apply) {
503        (Recorded::Transaction(_), Some(historical)) => {
504            historical(state);
505            Ok(())
506        }
507        (Recorded::Transaction(transaction), None) => {
508            if let Some(prepared) = prepared {
509                crate::state::TransactionCatalog::replay(&*transaction, &prepared, state)
510            } else {
511                transaction.apply(state);
512                Ok(())
513            }
514        }
515        (Recorded::Record(_), _) => Err(mismatched(id, version)),
516        (Recorded::System(_), _) => Err(mismatched(id, version)),
517    }
518}
519
520fn retain<A: Application>(payload: Payload<A>) -> Result<A::Record, DecodeError> {
521    match payload.value {
522        Recorded::Record(record) => Ok(record),
523        Recorded::Transaction(_) => Err(mismatched(payload.id, payload.version)),
524        Recorded::System(_) => Err(mismatched(payload.id, payload.version)),
525    }
526}
527
528fn mismatched(id: DefinitionId, version: u32) -> DecodeError {
529    DecodeError::Malformed { id, version }
530}
531
532async fn fold<A: Application>(
533    consumer: &mut dyn Attached<A>, recovered: &[Arc<Entry<Recorded<A>>>], stored: &[(Position, Arc<Vec<u8>>)], frontier: Position,
534    feed: &Feed<A>, applied: Position,
535) -> Position {
536    let resume = consumer.resume();
537    let Ok(live) = feed.since(resume.max(frontier)).await else {
538        return consumer.drain(Vec::new(), resume).await;
539    };
540    // What is about to be redelivered is forwarded first in its stored form,
541    // so a durable copy writes those bytes instead of re-encoding history.
542    let owed = stored.partition_point(|(position, _)| *position <= resume);
543    if owed < stored.len() {
544        consumer.redelivered(&stored[owed..]);
545    }
546    let mut entries: Vec<Arc<Entry<Recorded<A>>>> = recovered.iter().filter(|entry| entry.position > resume).map(Arc::clone).collect();
547    entries.extend(live);
548
549    consumer.drain(entries, applied).await
550}
551
552/* === Tests === */
harmos_runtime/storage/attachments.rs

harmos_runtime/storage/
attachments.rs
