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

harmos_runtime/
runtime.rs
