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

harmos_runtime/
runtime.rs
