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

harmos_runtime/storage/
attachments.rs
