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

harmos_runtime/
runtime.rs
