harmos_runtime/
runtime.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10/* === Modules === */
11
12mod boot;
13mod lifecycle;
14
15/* === Re-Exports === */
16
17pub use boot::{BootstrapCleanup, BootstrapStatus, Startup};
18
19/* === Imports === */
20
21// Standard Library Imports
22use std::fmt;
23use std::future::Future;
24use std::pin::Pin;
25use std::sync::atomic::AtomicBool;
26use std::sync::{Arc, Mutex, OnceLock, Weak};
27use std::task::Poll;
28use std::time::Duration;
29// Third Party Imports
30use tokio::sync::{mpsc, oneshot, watch};
31use tokio::task::JoinHandle;
32// Local Imports
33use crate::evidence::{Declaration, Evidence, Vault};
34use crate::services::{Once as DeclaredService, ServiceDefinition};
35use crate::state::dispatch::Dispatch;
36use crate::state::{Feed, Inverting, Origin, ReadRequest, Tail, Undos, replay};
37use crate::storage::attachments::{Attachments, Consumers, Horizon, Payload, Source, declare};
38use crate::storage::consumer::{JournalConsumer, Series, StreamConsumer};
39use crate::streams::{Definition, Holding, Request, StreamManager, Table, Tier};
40use crate::{
41    Application, DefinitionId, Error, operations::LaneSet, operations::OperationPolicy, services::Resident, state::At,
42    state::CaptureLimits, state::DecodeError, state::Entry, state::JournalCapture, state::Kind, state::Position, state::StateManager,
43    storage::Cursor,
44};
45
46/* === Constants === */
47
48const REQUESTS: usize = 64;
49
50const STOP_GRACE: Duration = Duration::from_millis(100);
51
52/* === Macros === */
53
54/* === Type Aliases === */
55
56/* === Structs === */
57
58pub struct Inspection<A: Application> {
59    state: Arc<At<A::State>>,
60    reads: crate::state::Reads<A>,
61}
62
63pub struct Inspector<A: Application> {
64    pub state: Inspection<A>,
65    source: JoinHandle<()>,
66}
67
68pub struct Runtime<A: Application> {
69    pub state: StateManager<A>,
70    pub records: crate::records::RecordManager<A>,
71    pub streams: StreamManager<A>,
72    pub operations: crate::operations::OperationManager,
73
74    #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
75    pub plugin: crate::plugin::PluginHost<A>,
76
77    #[cfg(feature = "grpc")]
78    grpc: crate::grpc::GrpcManager,
79
80    pub evidence: Evidence<A>,
81    stopping: Mutex<Stopping<A>>,
82    closing: AtomicBool,
83    storage: Arc<OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
84    journal_handles: Weak<()>,
85    participants: Participants,
86    notices: watch::Sender<Option<String>>,
87}
88
89pub struct Handle<A: Application> {
90    pub state: StateManager<A>,
91    pub records: crate::records::RecordManager<A>,
92    pub streams: StreamManager<A>,
93    pub operations: crate::operations::OperationManager,
94
95    #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
96    pub plugin: crate::plugin::PluginHost<A>,
97
98    #[cfg(feature = "grpc")]
99    grpc: crate::grpc::GrpcManager,
100
101    pub evidence: Evidence<A>,
102    storage: Arc<OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
103}
104
105pub struct Builder<A: Application> {
106    origin: Origin<A::State>,
107    resources: Arc<A::Resources>,
108    inverses: Inverting<A>,
109    catalog: Dispatch<A, Payload<A>>,
110    definitions: Vec<Definition>,
111    attachments: Attachments<A>,
112    managed_state: Option<crate::storage::ReadLimits>,
113    record_projection: Option<crate::storage::capture::Projection>,
114    reconciliation: Option<boot::Reconciliation<A>>,
115    initializers: Vec<boot::Initializer<A>>,
116    snapshots: Option<crate::state::snapshot::Snapshots<A>>,
117    lanes: LaneSet,
118    operations: Vec<crate::operations::CatalogBinding>,
119    operation_settings: crate::operations::Settings,
120    operation_middleware: Vec<Arc<dyn crate::operations::middleware::Middleware>>,
121    services: Vec<DeclaredService<A>>,
122    evidence: Option<Declaration>,
123
124    #[cfg(feature = "simulation")]
125    simulation: crate::simulation::Configuration<A>,
126
127    #[cfg(feature = "simulation")]
128    live_only: Vec<&'static str>,
129}
130
131#[derive(Debug, thiserror::Error)]
132#[error("{cause}")]
133pub struct Incompatible {
134    #[source]
135    cause: Cause,
136
137    cleanup: Option<boot::Cleanup>,
138}
139
140struct Supervision<A: Application> {
141    consumers: Consumers<A>,
142    feed: Feed<A>,
143    progress: watch::Receiver<Position>,
144    streams: StreamManager<A>,
145    delivery: watch::Receiver<()>,
146    demand: Vec<Holding>,
147    writer: JoinHandle<()>,
148    catalog: Arc<Dispatch<A, Payload<A>>>,
149    inverses: Inverting<A>,
150    requests: mpsc::Receiver<ReadRequest<A>>,
151    served: mpsc::Receiver<Request>,
152    ready: oneshot::Sender<()>,
153}
154
155#[derive(Clone)]
156struct Participants {
157    stopping: watch::Sender<bool>,
158    active: watch::Sender<usize>,
159}
160
161pub struct StopParticipant {
162    stopping: watch::Receiver<bool>,
163    active: watch::Sender<usize>,
164    released: bool,
165}
166
167/* === Enums === */
168
169enum Stopping<A: Application> {
170    Running(Option<JoinHandle<Position>>),
171    Closing {
172        cleanup: Arc<crate::supervision::cleanup::Cleanup<lifecycle::Shutdown<A>>>,
173        outcome: Arc<OnceLock<lifecycle::Outcome>>,
174        progress: watch::Sender<&'static str>,
175    },
176}
177
178#[allow(clippy::type_complexity)]
179pub enum Catalog<A: Application> {
180    Transactions {
181        definitions: Vec<(Kind, DefinitionId, u32)>,
182        decode: fn(&DefinitionId, u32, &[u8]) -> Result<Option<A::Transaction>, DecodeError>,
183        inverses: Option<fn(&A::Transaction, &A::State) -> Option<A::Transaction>>,
184    },
185    Streams(Vec<Definition>),
186    Lanes(LaneSet),
187}
188
189#[derive(Debug, thiserror::Error)]
190enum Cause {
191    #[error("the stored history cannot be replayed by this build: {0}")]
192    Undeclared(DecodeError),
193
194    #[error("the stored history cannot be replayed by this build: {0}")]
195    Unreadable(String),
196
197    #[error("the stream catalog cannot be honored: {0}")]
198    Dishonorable(crate::streams::Refused),
199
200    #[error("the work declarations cannot be honored: {0}")]
201    Work(String),
202
203    #[error("the operation declarations cannot be honored: {0}")]
204    Operations(#[source] crate::Error),
205
206    #[error("runtime initialization failed: {0}")]
207    Initialization(#[source] Box<dyn std::error::Error + Send + Sync>),
208
209    #[error("{source}; bootstrap cleanup also failed: {failure}")]
210    Cleanup {
211        #[source]
212        source: Box<Incompatible>,
213
214        failure: crate::Error,
215    },
216
217    #[error("application storage reconciliation failed: {0}")]
218    Reconciliation(String),
219
220    #[error("the evidence declarations cannot be honored: {0}")]
221    Kept(String),
222
223    #[cfg(feature = "simulation")]
224    #[error("this build cannot be simulated after attaching {0}")]
225    Simulation(String),
226}
227
228enum Turn<A: Application> {
229    Advanced { journal: bool, streams: bool },
230    Read(ReadRequest<A>),
231    Serve(Box<Request>),
232    Stopped,
233}
234
235/* === Traits === */
236
237pub trait Registration<A: Application>: Clone + Send {
238    fn register(self) -> Catalog<A>;
239}
240
241/* === Implementations === */
242
243impl<A, S, T> Registration<A> for harmos_domain::state::TransactionDefinitions<S, T>
244where
245    A: Application<State = S, Transaction = T>,
246{
247    fn register(self) -> Catalog<A> {
248        Catalog::Transactions { definitions: self.definitions, decode: self.decode, inverses: Some(self.inverses) }
249    }
250}
251
252impl<A, R> Registration<A> for harmos_domain::state::RecordDefinitions<R>
253where
254    A: Application<Record = R>,
255{
256    fn register(self) -> Catalog<A> {
257        Catalog::Transactions { definitions: self.definitions, decode: |_, _, _| Ok(None), inverses: None }
258    }
259}
260
261impl<A: Application> Inspection<A> {
262    pub async fn read<R>(&self, read: impl FnOnce(&A::State) -> R) -> Result<At<R>, Error> {
263        Ok(At { value: read(&self.state.value), position: self.state.position })
264    }
265
266    pub async fn state_at(&self, origin: Origin<A::State>, at: Position) -> Result<At<A::State>, Error> {
267        self.reads.state(origin, at).await
268    }
269
270    pub async fn query<X>(&self, after: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
271    where
272        X: Clone + Send + 'static,
273    {
274        self.query_through(after, self.state.position).await
275    }
276
277    pub async fn query_through<X>(&self, after: Position, through: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
278    where
279        X: Clone + Send + 'static,
280    {
281        if through > self.state.position {
282            return vec![Err(Error::FrontierAhead)].into_iter();
283        }
284        if after >= through {
285            return Vec::new().into_iter();
286        }
287
288        match self.reads.entries(after, through).await {
289            Ok(entries) => entries
290                .into_iter()
291                .filter_map(|entry| {
292                    entry
293                        .value
294                        .select::<X>()
295                        .map(|value| Ok(Entry { prepared: None, position: entry.position, metadata: entry.metadata, value }))
296                })
297                .collect::<Vec<_>>()
298                .into_iter(),
299            Err(refusal) => vec![Err(refusal)].into_iter(),
300        }
301    }
302
303    pub fn applied(&self) -> Position {
304        self.state.position
305    }
306
307    pub fn stored(&self) -> Position {
308        self.state.position
309    }
310}
311
312impl<A: Application> Drop for Inspector<A> {
313    fn drop(&mut self) {
314        self.source.abort();
315    }
316}
317
318// Hand-written, because a derived `Clone` would demand `A: Clone` of an
319// application that is only ever a marker.
320impl<A: Application> Clone for Handle<A> {
321    fn clone(&self) -> Self {
322        Self {
323            state: self.state.clone(),
324            records: self.records.clone(),
325            streams: self.streams.clone(),
326            operations: self.operations.clone(),
327            #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
328            plugin: self.plugin.clone(),
329            #[cfg(feature = "grpc")]
330            grpc: self.grpc.clone(),
331            evidence: self.evidence.clone(),
332            storage: self.storage.clone(),
333        }
334    }
335}
336
337impl<A: Application> Handle<A> {
338    #[cfg(feature = "grpc")]
339    pub fn grpc(&self) -> &crate::grpc::GrpcManager {
340        &self.grpc
341    }
342}
343
344impl<A: Application> Runtime<A> {
345    #[cfg(feature = "grpc")]
346    pub fn grpc(&self) -> &crate::grpc::GrpcManager {
347        &self.grpc
348    }
349
350    pub fn builder(origin: impl Into<Origin<A::State>>, resources: impl Into<Arc<A::Resources>>) -> Builder<A> {
351        Builder {
352            origin: origin.into(),
353            resources: resources.into(),
354            inverses: none::<A>,
355            catalog: Dispatch::default(),
356            definitions: Vec::new(),
357            attachments: Attachments::new(),
358            managed_state: None,
359            record_projection: None,
360            reconciliation: None,
361            initializers: Vec::new(),
362            snapshots: None,
363            lanes: LaneSet::default(),
364            operations: Vec::new(),
365            operation_settings: crate::operations::Settings::default(),
366            operation_middleware: Vec::new(),
367            services: Vec::new(),
368            evidence: None,
369            #[cfg(feature = "simulation")]
370            simulation: crate::simulation::Configuration::new(),
371            #[cfg(feature = "simulation")]
372            live_only: Vec::new(),
373        }
374    }
375
376    pub async fn inspect(origin: impl Into<Origin<A::State>>, source: impl Source<A>) -> Result<Inspector<A>, Incompatible> {
377        let mut attachments = Attachments::new();
378        attachments.recover(source);
379        let mut consumers = attachments.supervised();
380        let recovery = consumers.load().await?;
381        let origin = origin.into();
382        let anchor = origin.position;
383        let mut state = At { value: origin.value, position: anchor };
384        let entries = recovery.entries.into_iter().filter(|entry| entry.position > anchor).collect();
385        replay(None, entries, &mut state, &mut Undos::default(), none::<A>, 0).map_err(Incompatible::of)?;
386        state.position = state.position.max(recovery.frontier).max(anchor);
387
388        let (reads, requests) = crate::state::Reads::channel(REQUESTS);
389        let source = tokio::spawn(inspect_reads(consumers, requests));
390
391        Ok(Inspector { state: Inspection { state: Arc::new(state), reads }, source })
392    }
393
394    pub fn stop_participant(&self) -> StopParticipant {
395        self.participants.register()
396    }
397
398    pub fn notices(&self) -> watch::Receiver<Option<String>> {
399        self.notices.subscribe()
400    }
401
402    pub fn handle(&self) -> Handle<A> {
403        Handle {
404            state: self.state.clone(),
405            records: self.records.clone(),
406            streams: self.streams.clone(),
407            operations: self.operations.clone(),
408            #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
409            plugin: self.plugin.clone(),
410            #[cfg(feature = "grpc")]
411            grpc: self.grpc.clone(),
412            evidence: self.evidence.clone(),
413            storage: self.storage.clone(),
414        }
415    }
416}
417
418impl<A: Application> Registration<A> for Vec<Definition> {
419    fn register(self) -> Catalog<A> {
420        Catalog::Streams(self)
421    }
422}
423
424impl<A: Application, const N: usize> Registration<A> for [Definition; N] {
425    fn register(self) -> Catalog<A> {
426        Catalog::Streams(self.into())
427    }
428}
429
430impl<A: Application> Registration<A> for LaneSet {
431    fn register(self) -> Catalog<A> {
432        Catalog::Lanes(self)
433    }
434}
435
436impl<A: Application> Builder<A> {
437    pub fn register<C: Registration<A> + 'static>(mut self, catalog: C) -> Self {
438        #[cfg(feature = "simulation")]
439        self.simulation.register(catalog.clone());
440
441        match catalog.register() {
442            Catalog::Transactions { definitions, decode, inverses } => {
443                if let Some(inverses) = inverses {
444                    self.inverses = inverses;
445                }
446                for (kind, id, version) in definitions {
447                    declare(&mut self.catalog, kind, id, version, decode);
448                }
449            }
450            Catalog::Streams(definitions) => self.definitions.extend(definitions),
451            Catalog::Lanes(declared) => self.lanes.extend(declared),
452        }
453        self
454    }
455
456    pub fn operation<T>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<T::Resources>) -> Self
457    where
458        T: for<'ctx> crate::operations::Operation<Context<'ctx> = crate::operations::Context<'ctx>>,
459    {
460        let binding = crate::operations::CatalogBinding::operation::<T>(owner.into(), lane, resources);
461        self.bind_operations(binding)
462    }
463
464    pub fn operations<C, R>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<R>) -> Self
465    where
466        C: crate::catalog::Catalog<crate::operations::AuthoredDeclarations>
467            + crate::catalog::Catalog<crate::operations::Implementations<R>>,
468        R: Send + Sync + 'static,
469    {
470        let binding = crate::operations::CatalogBinding::local::<C, R>(owner.into(), lane, resources);
471        self.bind_operations(binding)
472    }
473
474    pub fn wire_operations<C, R>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<R>) -> Self
475    where
476        C: crate::catalog::Catalog<crate::operations::AuthoredDeclarations>
477            + crate::catalog::Catalog<crate::operations::WireImplementations<R>>,
478        R: Send + Sync + 'static,
479    {
480        let binding = crate::operations::CatalogBinding::wire::<C, R>(owner.into(), lane, resources);
481        self.bind_operations(binding)
482    }
483
484    pub fn operation_middleware(mut self, middleware: Arc<dyn crate::operations::middleware::Middleware>) -> Self {
485        #[cfg(feature = "simulation")]
486        self.simulation.operation_middleware(middleware.clone());
487        self.operation_middleware.push(middleware);
488        self
489    }
490
491    pub fn operation_settings(mut self, settings: crate::operations::Settings) -> Self {
492        #[cfg(feature = "simulation")]
493        self.simulation.operation_settings(settings);
494        self.operation_settings = settings;
495        self
496    }
497
498    pub(crate) fn bind_operations(mut self, binding: crate::operations::CatalogBinding) -> Self {
499        #[cfg(feature = "simulation")]
500        self.simulation.operations(binding.clone());
501        self.operations.push(binding);
502        self
503    }
504
505    pub fn service<R>(self, name: impl Into<String>, resident: R) -> Self
506    where
507        R: Resident<A>,
508    {
509        self.service_with(name, crate::services::POLICY, resident)
510    }
511
512    pub fn service_with<R>(mut self, name: impl Into<String>, policy: OperationPolicy, resident: R) -> Self
513    where
514        R: Resident<A>,
515    {
516        #[cfg(feature = "simulation")]
517        self.live_only.push("resident services");
518
519        let definition = ServiceDefinition { id: name.into(), policy };
520        self.services.push(DeclaredService::new(definition, resident));
521        self
522    }
523
524    pub fn evidence(mut self, vault: impl Vault, kinds: impl IntoIterator<Item = (DefinitionId, u32)>) -> Self {
525        #[cfg(feature = "simulation")]
526        self.live_only.push("an evidence vault");
527
528        self.evidence = Some(Declaration::new(vault, kinds));
529        self
530    }
531
532    pub fn attach(mut self, consumer: impl JournalConsumer<A>) -> Self {
533        #[cfg(feature = "simulation")]
534        self.live_only.push("an attached fold");
535
536        self.attachments.attach(consumer);
537        self
538    }
539
540    pub fn attach_streams(mut self, consumer: impl StreamConsumer<A>) -> Self {
541        #[cfg(feature = "simulation")]
542        self.live_only.push("an attached stream fold");
543
544        self.attachments.attach_streams(consumer);
545        self
546    }
547
548    pub fn series(mut self, series: impl Series<A>) -> Self {
549        #[cfg(feature = "simulation")]
550        self.live_only.push("a series copy");
551
552        self.attachments.series(series);
553        self
554    }
555
556    pub fn recover(mut self, source: impl Source<A>) -> Self {
557        #[cfg(feature = "simulation")]
558        self.live_only.push("a recovery source");
559
560        self.attachments.recover(source);
561        self
562    }
563
564    pub fn snapshot(mut self, snapshot: impl JournalConsumer<A>) -> Self {
565        #[cfg(feature = "simulation")]
566        self.live_only.push("a snapshot fold");
567
568        self.attachments.snapshot(snapshot);
569        self
570    }
571
572    pub fn require_capture(mut self, limits: CaptureLimits, consumer: impl JournalCapture<A>) -> Result<Self, crate::state::CaptureError> {
573        #[cfg(feature = "simulation")]
574        self.live_only.push("required capture");
575
576        self.attachments.require_capture(limits, consumer)?;
577
578        Ok(self)
579    }
580
581    pub fn retain(mut self, horizon: Horizon) -> Self {
582        self.attachments.retain(horizon);
583        self
584    }
585
586    pub fn tail(mut self, entries: usize) -> Self {
587        self.attachments.tail(entries);
588        self
589    }
590
591    #[cfg(feature = "simulation")]
592    pub fn simulated_service<R>(mut self, name: impl Into<String>, resident: R) -> Self
593    where
594        R: Resident<A> + Clone,
595    {
596        self.simulation.service(name, resident);
597        self
598    }
599
600    #[cfg(feature = "simulation")]
601    pub fn simulated_service_with<R>(mut self, name: impl Into<String>, policy: OperationPolicy, resident: R) -> Self
602    where
603        R: Resident<A> + Clone,
604    {
605        self.simulation.service_with(name, policy, resident);
606        self
607    }
608
609    #[cfg(feature = "simulation")]
610    pub fn simulated_evidence(mut self, vault: impl Vault + Clone, kinds: impl IntoIterator<Item = (DefinitionId, u32)>) -> Self {
611        self.simulation.evidence(vault, kinds);
612        self
613    }
614
615    #[cfg(feature = "simulation")]
616    pub fn entropy_seed(mut self, seed: u64) -> Self {
617        self.simulation.entropy_seed(seed);
618        self
619    }
620
621    #[cfg(feature = "simulation")]
622    pub fn simulate(self) -> Result<crate::simulation::SimulatedRuntime<A>, Incompatible>
623    where
624        A::State: Clone + fmt::Debug + PartialEq,
625        A::Transaction: Clone + fmt::Debug,
626        A::Record: Clone + fmt::Debug,
627    {
628        if !self.live_only.is_empty() {
629            return Err(Incompatible::simulation(self.live_only.join(", ")));
630        }
631
632        self.simulation.simulate(self.origin, self.resources)
633    }
634
635    #[tracing::instrument(level = "debug", skip_all)]
636    pub fn runtime(self) -> Startup<A> {
637        Startup::new(self, None)
638    }
639
640    pub fn managed_state(mut self, limits: crate::storage::ReadLimits) -> Self {
641        self.managed_state = Some(limits);
642        #[cfg(feature = "simulation")]
643        self.live_only.push("managed storage recovery");
644        self
645    }
646
647    pub fn store_records(mut self, limits: crate::storage::ReadLimits, projection: impl crate::storage::RecordProjection) -> Self {
648        self.record_projection = Some(crate::storage::capture::Projection { limits, handler: Box::new(projection) });
649        #[cfg(feature = "simulation")]
650        self.live_only.push("managed primary record capture");
651        self
652    }
653
654    pub fn managed_snapshots(mut self, policy: crate::state::SnapshotPolicy) -> Self
655    where
656        A::State: serde::Serialize + serde::de::DeserializeOwned,
657    {
658        self.snapshots = Some(crate::state::snapshot::Snapshots::new(policy));
659        #[cfg(feature = "simulation")]
660        self.live_only.push("managed state snapshots");
661        self
662    }
663
664    pub fn reconcile<F, Fut>(mut self, reconcile: F) -> Self
665    where
666        F: FnOnce(StateManager<A>, crate::storage::StorageManager<crate::storage::ErasedCatalog>) -> Fut + Send + 'static,
667        Fut: Future<Output = crate::Result<()>> + Send + 'static,
668    {
669        self.reconciliation = Some(Box::new(move |journal, storage| Box::pin(reconcile(journal, storage))));
670        #[cfg(feature = "simulation")]
671        self.live_only.push("saved application data reconciliation");
672        self
673    }
674
675    pub fn initialize<F, Fut, E>(mut self, timeout: Duration, initialize: F) -> Self
676    where
677        F: Fn(Handle<A>) -> Fut + Clone + Send + Sync + 'static,
678        Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
679        E: std::error::Error + Send + Sync + 'static,
680    {
681        #[cfg(feature = "simulation")]
682        self.simulation.initialize(timeout, initialize.clone());
683        self.initializers.push(boot::Initializer::new(timeout, initialize));
684        self
685    }
686
687    pub fn runtime_with_storage<C>(self, storage: &crate::storage::StorageManager<C>) -> Startup<A> {
688        Startup::new(self, Some(storage.erased()))
689    }
690
691    async fn boot(
692        self, storage: Option<&crate::storage::StorageManager<crate::storage::ErasedCatalog>>, cleanup: BootstrapCleanup,
693    ) -> Result<Runtime<A>, Incompatible> {
694        if self
695            .initializers
696            .iter()
697            .any(|initializer| initializer.timeout.is_zero() || tokio::time::Instant::now().checked_add(initializer.timeout).is_none())
698        {
699            let source: crate::Error = Error::InvalidInput {
700                field: "runtime initialization timeout",
701                reason: "requires a positive representable finite duration",
702            };
703            return Err(Incompatible::initialization(Box::new(source)));
704        }
705        if self.snapshots.is_some() && self.managed_state.is_none() {
706            return Err(Incompatible::unreadable("managed snapshots require managed state"));
707        }
708        if self.reconciliation.is_some() && self.managed_state.is_none() {
709            return Err(Incompatible::reconciliation("reconciliation requires managed state"));
710        }
711        if self.record_projection.is_some() && self.managed_state.is_none() {
712            return Err(Incompatible::unreadable("primary record capture requires managed state"));
713        }
714        if self.managed_state.is_some() && (storage.is_none() || self.attachments.has_recovery_owner()) {
715            return Err(Incompatible::unreadable("managed state requires storage and exclusive recovery ownership"));
716        }
717        let mut boot = boot::Boot::new(storage, cleanup.clone()).map_err(Incompatible::work)?;
718        if let Err(error) = self.boot_owned(&mut boot).await {
719            return match boot.close().await {
720                Ok(()) => Err(error),
721                Err(failure) => {
722                    let failure = Arc::new(failure);
723                    let retained = boot.retain_cleanup();
724                    cleanup.failed(retained.clone(), Arc::clone(&failure));
725                    Err(Incompatible {
726                        cause: Cause::Cleanup {
727                            source: Box::new(error),
728                            failure: Error::TaskFailed { domain: "bootstrap cleanup", source: failure },
729                        },
730                        cleanup: retained,
731                    })
732                }
733            };
734        }
735        Ok(boot.finish())
736    }
737
738    async fn boot_owned(self, boot: &mut boot::Boot<A>) -> Result<(), Incompatible> {
739        boot.acquire().await.map_err(Incompatible::work)?;
740        let Self {
741            origin,
742            resources,
743            inverses,
744            catalog,
745            definitions,
746            mut attachments,
747            managed_state,
748            record_projection,
749            reconciliation,
750            initializers,
751            snapshots,
752            lanes,
753            operations: bindings,
754            operation_settings,
755            operation_middleware,
756            services,
757            evidence,
758            ..
759        } = self;
760        let tail = attachments.tail_window()?;
761        let mut consumers = attachments.supervised();
762        let recovery = consumers.load().await?;
763        let (resume, selections) = consumers.recordings().fold((Cursor::ORIGIN, Vec::new()), |(resume, mut selections), recording| {
764            selections.push(recording.selections());
765
766            (resume.max(recording.resume()), selections)
767        });
768        let anchor = origin.position;
769        let mut state = At { value: origin.value, position: anchor };
770        let mut undos = Undos::default();
771        let entries = recovery.entries.into_iter().filter(|entry| entry.position > anchor).collect();
772        let mut recovered = replay(Some(&catalog), entries, &mut state, &mut undos, inverses, tail).map_err(Incompatible::of)?;
773        let stored = recovery.encoded.into_iter().filter(|(position, _)| *position > anchor).collect();
774        state.position = state.position.max(recovery.frontier).max(anchor);
775        let mut managed_durable = None;
776        if let Some(limits) = managed_state {
777            let storage = boot.storage.as_ref().expect("managed state requires storage");
778            let loaded = StateManager::recover(storage, state, &catalog, inverses, tail, limits, snapshots.as_ref()).await?;
779            state = loaded.state;
780            undos = loaded.undos;
781            recovered = loaded.tail;
782            managed_durable = Some(loaded.durable);
783            attachments.capture = Some(crate::state::capture::Declaration::Managed {
784                storage: storage.clone(),
785                receipts: loaded.receipts,
786                projection: record_projection,
787                snapshots,
788            });
789        }
790        let windows = Table::resuming(definitions, state.position, resume).map_err(Incompatible::dishonored)?;
791        let demand = windows.hold(selections).map_err(Incompatible::dishonored)?;
792        let replayed = recovered.clone();
793        consumers.redelivering(recovered.clone(), stored, state.position);
794
795        let durability = managed_durable.or_else(|| consumers.durability(state.position));
796        let tail = Tail::recovered(state.position, tail, recovered);
797        let catalog = Arc::new(catalog);
798        let (journal, writer, requests) = StateManager::start(
799            state,
800            inverses,
801            undos,
802            REQUESTS,
803            tail,
804            durability,
805            Arc::clone(&catalog),
806            attachments.capture.take(),
807            boot.storage.clone(),
808            crate::state::Timestamp::now,
809        );
810        let journal_handles = journal.handle_probe();
811        let records = crate::records::RecordManager::new(journal.clone());
812        boot.writer = Some(writer);
813        boot.state = Some(journal.clone());
814        // Evidence folds from the same replayed order the state above it
815        // folded from, so a boot learns what was finalized from facts rather
816        // than from whatever files happen to be lying in a vault.
817        let evidence = Evidence::wired(journal.clone(), evidence, &replayed).map_err(Incompatible::kept)?;
818        // The windows are wired before the consumers start, so a recorder
819        // attached below never misses a row published in the gap between the
820        // two.
821        let (tier, served) = Tier::channel(REQUESTS);
822        let streams = StreamManager::wired(windows, journal.progress(), tier);
823        let delivery = streams.delivery();
824        operation_settings.validate().map_err(|source| Incompatible { cause: Cause::Operations(source), cleanup: None })?;
825        let lanes = crate::operations::lane::Lanes::freeze(lanes).map_err(Incompatible::work)?;
826        let storage = Arc::new(boot.storage.clone().map(OnceLock::from).unwrap_or_default());
827        let handles = crate::services::Handles {
828            storage: storage.clone(),
829            state: journal.clone(),
830            records: records.clone(),
831            streams: streams.clone(),
832            evidence: evidence.clone(),
833        };
834        let instances = crate::operations::instances::Instances::start(lanes, Box::new(handles.clone()), resources, &operation_settings);
835        let mut reserved = std::collections::BTreeSet::new();
836        for service in &services {
837            crate::services::validate(&service.definition()).map_err(Incompatible::work)?;
838            if !reserved.insert(service.id().to_owned()) {
839                let duplicate: crate::Error = crate::Error::AlreadyRegistered { domain: "service", name: service.id().to_owned() };
840                return Err(Incompatible::work(duplicate));
841            }
842        }
843        let operations = crate::operations::OperationManager::new(instances, &operation_settings, reserved, boot.identity.clone());
844        for middleware in operation_middleware {
845            operations.register_middleware(middleware).map_err(Incompatible::work)?;
846        }
847        for binding in bindings {
848            binding.bind(&operations).map_err(|source| Incompatible { cause: Cause::Operations(source), cleanup: None })?;
849        }
850        if let Some(storage) = &boot.storage {
851            storage.start_on::<A>(&operations).await.map_err(Incompatible::work)?;
852        }
853        let (ready, booted) = oneshot::channel();
854        let supervisor = tokio::spawn(supervise(Supervision {
855            consumers,
856            feed: journal.feed(),
857            progress: journal.progress(),
858            streams: streams.clone(),
859            delivery,
860            demand,
861            writer: boot.writer.take().expect("boot owns writer"),
862            catalog,
863            inverses,
864            requests,
865            served,
866            ready,
867        }));
868        #[cfg(feature = "grpc")]
869        let grpc = crate::grpc::GrpcManager::new::<A>(&operations);
870        #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
871        let plugin = crate::plugin::PluginHost::new(
872            operations.clone(),
873            #[cfg(feature = "plugin-sidecar")]
874            handles,
875        )
876        .map_err(Incompatible::work)?;
877        boot.runtime = Some(Runtime {
878            state: journal,
879            records,
880            streams,
881            operations,
882            #[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
883            plugin,
884            #[cfg(feature = "grpc")]
885            grpc,
886            evidence,
887            stopping: Mutex::new(Stopping::Running(Some(supervisor))),
888            closing: AtomicBool::new(false),
889            storage,
890            journal_handles,
891            participants: Participants::new(),
892            notices: watch::channel(None).0,
893        });
894        boot.state.take();
895        booted.await.map_err(Incompatible::work)?;
896        if let Some(reconcile) = reconciliation {
897            let runtime = boot.runtime.as_ref().expect("boot owns runtime");
898            let storage = runtime.storage().expect("reconciliation requires storage");
899            reconcile(runtime.state.clone(), storage.clone()).await.map_err(Incompatible::reconciliation)?;
900            storage.flush().await.map_err(Incompatible::reconciliation)?;
901            storage.flush_recovery().await.map_err(Incompatible::reconciliation)?;
902        }
903        let operations = &boot.runtime.as_ref().expect("runtime is owned during boot").operations;
904        for service in services {
905            let lease = operations.start_service(service, false, true).map_err(Incompatible::work)?;
906            operations.hold_resident(lease);
907        }
908        for initializer in initializers {
909            let handle = boot.runtime.as_ref().expect("runtime is owned during initialization").handle();
910            let initialize = (initializer.initialize)(handle);
911            match tokio::time::timeout(initializer.timeout, initialize).await {
912                Ok(result) => result.map_err(Incompatible::initialization)?,
913                Err(source) => return Err(Incompatible::initialization(Box::new(source))),
914            }
915        }
916        Ok(())
917    }
918}
919
920impl Incompatible {
921    fn initialization(source: Box<dyn std::error::Error + Send + Sync>) -> Self {
922        Self { cause: Cause::Initialization(source), cleanup: None }
923    }
924
925    pub fn cleanup_error(&self) -> Option<&crate::Error> {
926        match (&self.cleanup, &self.cause) {
927            (Some(_), Cause::Cleanup { failure, .. }) => Some(failure),
928            _ => None,
929        }
930    }
931
932    pub async fn retry_cleanup(&mut self) -> crate::Result<()> {
933        if let Some(cleanup) = &self.cleanup {
934            cleanup.close().await?;
935            self.cleanup.take();
936        }
937        Ok(())
938    }
939
940    fn reconciliation(cause: impl fmt::Display) -> Self {
941        Self { cause: Cause::Reconciliation(cause.to_string()), cleanup: None }
942    }
943
944    pub fn unreadable(cause: impl fmt::Display) -> Self {
945        Self { cause: Cause::Unreadable(cause.to_string()), cleanup: None }
946    }
947
948    fn of(cause: DecodeError) -> Self {
949        Self { cause: Cause::Undeclared(cause), cleanup: None }
950    }
951
952    pub(crate) fn dishonored(cause: crate::streams::Refused) -> Self {
953        Self { cause: Cause::Dishonorable(cause), cleanup: None }
954    }
955
956    pub(crate) fn work(cause: impl fmt::Display) -> Self {
957        Self { cause: Cause::Work(cause.to_string()), cleanup: None }
958    }
959
960    #[cfg(feature = "simulation")]
961    fn simulation(live_only: String) -> Self {
962        Self { cause: Cause::Simulation(live_only), cleanup: None }
963    }
964
965    fn kept(cause: String) -> Self {
966        Self { cause: Cause::Kept(cause), cleanup: None }
967    }
968}
969
970#[cfg(feature = "plugin-sidecar")]
971impl From<harmos_sidecar::ExecutableError> for Incompatible {
972    fn from(error: harmos_sidecar::ExecutableError) -> Self {
973        Self::work(error)
974    }
975}
976
977impl Participants {
978    fn new() -> Self {
979        Self { stopping: watch::channel(false).0, active: watch::channel(0).0 }
980    }
981
982    fn register(&self) -> StopParticipant {
983        self.active.send_modify(|active| *active += 1);
984
985        StopParticipant { stopping: self.stopping.subscribe(), active: self.active.clone(), released: false }
986    }
987
988    async fn stop(&self) {
989        self.stopping.send_replace(true);
990        let mut active = self.active.subscribe();
991        while *active.borrow_and_update() != 0 {
992            if active.changed().await.is_err() {
993                break;
994            }
995        }
996    }
997}
998
999impl StopParticipant {
1000    pub fn is_stopping(&self) -> bool {
1001        *self.stopping.borrow()
1002    }
1003
1004    pub async fn stopping(&mut self) {
1005        while !*self.stopping.borrow_and_update() {
1006            if self.stopping.changed().await.is_err() {
1007                break;
1008            }
1009        }
1010    }
1011
1012    pub fn finish(mut self) {
1013        self.release();
1014    }
1015
1016    fn release(&mut self) {
1017        if !self.released {
1018            self.active.send_modify(|active| *active = active.saturating_sub(1));
1019            self.released = true;
1020        }
1021    }
1022}
1023
1024impl Drop for StopParticipant {
1025    fn drop(&mut self) {
1026        self.release();
1027    }
1028}
1029
1030/* === Public Functions === */
1031
1032/* === Private Functions === */
1033
1034fn none<A: Application>(_change: &A::Transaction, _before: &A::State) -> Option<A::Transaction> {
1035    None
1036}
1037
1038async fn supervise<A: Application>(supervision: Supervision<A>) -> Position {
1039    let Supervision {
1040        mut consumers,
1041        feed,
1042        mut progress,
1043        streams,
1044        mut delivery,
1045        demand: _demand,
1046        mut writer,
1047        catalog,
1048        inverses,
1049        mut requests,
1050        mut served,
1051        ready,
1052    } = supervision;
1053    // Starting the consumers is a step of boot rather than a consequence of the
1054    // next commit: a fold whose checkpoint predates the crash is owed entries
1055    // the moment the runtime is up, however quiet the writer then stays.
1056    let booted = *progress.borrow_and_update();
1057    consumers.drain(&feed, booted).await;
1058    deliver(&mut consumers, &streams).await;
1059    stand_windows(&streams, &mut consumers, booted).await;
1060    let _ = ready.send(());
1061
1062    loop {
1063        match turn(&mut progress, &mut delivery, &mut requests, &mut served, &mut writer).await {
1064            Turn::Advanced { journal, streams: rows } => {
1065                if journal {
1066                    let applied = *progress.borrow_and_update();
1067                    consumers.drain(&feed, applied).await;
1068                }
1069                if rows {
1070                    delivery.borrow_and_update();
1071                    deliver(&mut consumers, &streams).await;
1072                }
1073            }
1074            Turn::Read(request) => answer(request, &mut consumers, Some(&catalog), inverses).await,
1075            Turn::Serve(request) => serve(*request, &mut consumers).await,
1076            Turn::Stopped => break,
1077        }
1078    }
1079
1080    let applied = *progress.borrow();
1081    consumers.drain(&feed, applied).await;
1082    deliver(&mut consumers, &streams).await;
1083    consumers.flush().await;
1084
1085    // What the flush made durable is read back through the contract's own
1086    // question, so the reported frontier includes the batches it just wrote.
1087    consumers.settle(applied).await
1088}
1089
1090async fn deliver<A: Application>(consumers: &mut Consumers<A>, streams: &StreamManager<A>) {
1091    for recording in consumers.recordings() {
1092        let resume = recording.resume();
1093        let batch = streams.batch(resume);
1094
1095        if batch.reaches() > resume {
1096            recording.drain(batch).await;
1097        }
1098    }
1099}
1100
1101async fn stand_windows<A: Application>(streams: &StreamManager<A>, consumers: &mut Consumers<A>, booted: Position) {
1102    for wired in streams.declared() {
1103        let durable = consumers.sealed(wired.definition.clone(), &[]).await;
1104
1105        if durable > Position::ORIGIN && durable <= booted {
1106            wired.window.stands_on(durable);
1107        }
1108    }
1109}
1110
1111async fn turn<A: Application>(
1112    progress: &mut watch::Receiver<Position>, delivery: &mut watch::Receiver<()>, requests: &mut mpsc::Receiver<ReadRequest<A>>,
1113    served: &mut mpsc::Receiver<Request>, writer: &mut JoinHandle<()>,
1114) -> Turn<A> {
1115    let mut advanced = Box::pin(progress.changed());
1116    let mut published = Box::pin(delivery.changed());
1117
1118    std::future::poll_fn(|context| {
1119        if Pin::new(&mut *writer).poll(context).is_ready() {
1120            return Poll::Ready(Turn::Stopped);
1121        }
1122        let journal = match advanced.as_mut().poll(context) {
1123            Poll::Ready(Ok(())) => true,
1124            Poll::Ready(Err(_)) => return Poll::Ready(Turn::Stopped),
1125            Poll::Pending => false,
1126        };
1127        let streams = match published.as_mut().poll(context) {
1128            Poll::Ready(Ok(())) => true,
1129            Poll::Ready(Err(_)) => return Poll::Ready(Turn::Stopped),
1130            Poll::Pending => false,
1131        };
1132
1133        if journal || streams {
1134            Poll::Ready(Turn::Advanced { journal, streams })
1135        } else {
1136            match requests.poll_recv(context) {
1137                Poll::Ready(Some(request)) => Poll::Ready(Turn::Read(request)),
1138                Poll::Ready(None) | Poll::Pending => match served.poll_recv(context) {
1139                    Poll::Ready(Some(request)) => Poll::Ready(Turn::Serve(Box::new(request))),
1140                    Poll::Ready(None) | Poll::Pending => Poll::Pending,
1141                },
1142            }
1143        }
1144    })
1145    .await
1146}
1147
1148async fn answer<A: Application>(
1149    request: ReadRequest<A>, consumers: &mut Consumers<A>, catalog: Option<&Dispatch<A, Payload<A>>>, inverses: Inverting<A>,
1150) {
1151    match request {
1152        ReadRequest::Entries { after, through, reply } => {
1153            let answer = consumers.read(after, through).await.map(|recovery| {
1154                recovery
1155                    .entries
1156                    .into_iter()
1157                    .map(|envelope| Entry {
1158                        prepared: envelope.payload.prepared.clone(),
1159                        position: envelope.position,
1160                        metadata: envelope.metadata,
1161                        value: envelope.payload.value,
1162                    })
1163                    .collect()
1164            });
1165            let _ = reply.send(answer);
1166        }
1167        ReadRequest::State { origin, at, reply } => {
1168            let answer = async {
1169                let recovery = consumers.read(origin.position, at).await?;
1170                if recovery
1171                    .commits
1172                    .iter()
1173                    .copied()
1174                    .chain(recovery.entries.iter().filter_map(|entry| entry.metadata.commit))
1175                    .any(|(first, end)| (first <= at && at < end) || (first <= origin.position && origin.position < end))
1176                {
1177                    return Err(crate::Error::IncompleteCommit);
1178                }
1179                if recovery.frontier < at {
1180                    return Err(crate::Error::BeyondTail);
1181                }
1182                let mut state = At { value: origin.value, position: origin.position };
1183                replay(catalog, recovery.entries, &mut state, &mut Undos::default(), inverses, 0).map_err(|_| crate::Error::Unavailable)?;
1184                state.position = at;
1185
1186                Ok(state)
1187            }
1188            .await;
1189            let _ = reply.send(answer);
1190        }
1191    }
1192}
1193
1194async fn serve<A: Application>(request: Request, consumers: &mut Consumers<A>) {
1195    match request {
1196        Request::Reader { stream, track, start, end, limits, permit, reply } => {
1197            if !reply.is_closed() {
1198                let answer = consumers.row_reader(stream, &track, start, end, limits).await;
1199                let _ = reply.send(crate::streams::Opened { source: answer, permit });
1200            }
1201        }
1202        Request::Rows { stream, track, start, end, reply } => {
1203            let answer = consumers.served(stream, &track, start, end).await;
1204            let _ = reply.send(answer);
1205        }
1206        Request::Buckets { stream, track, start, end, budget, reply } => {
1207            let answer = consumers.surveyed(stream, &track, start, end, budget).await;
1208            let _ = reply.send(answer);
1209        }
1210        Request::Aggregate { stream, track, start, end, kind, reply } => {
1211            let answer = consumers.aggregated(stream, &track, start, end, kind).await;
1212            let _ = reply.send(answer);
1213        }
1214        Request::Seal { stream, track, reply } => {
1215            let _ = reply.send(consumers.sealed(stream, &track).await);
1216        }
1217    }
1218}
1219
1220async fn inspect_reads<A: Application>(mut consumers: Consumers<A>, mut requests: mpsc::Receiver<ReadRequest<A>>) {
1221    while let Some(request) = requests.recv().await {
1222        answer(request, &mut consumers, None, none::<A>).await;
1223    }
1224}
1225
1226/* === Tests === */
harmos_runtime/runtime.rs

harmos_runtime/
runtime.rs
