mod boot;
mod lifecycle;
pub use boot::{BootstrapCleanup, BootstrapStatus, Startup};
use std::fmt;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::AtomicBool;
use std::sync::{Arc, Mutex, OnceLock, Weak};
use std::task::Poll;
use std::time::Duration;
use tokio::sync::{mpsc, oneshot, watch};
use tokio::task::JoinHandle;
use crate::evidence::{Declaration, Evidence, Vault};
use crate::services::{Once as DeclaredService, ServiceDefinition};
use crate::state::dispatch::Dispatch;
use crate::state::{Feed, Inverting, Origin, ReadRequest, Tail, Undos, replay};
use crate::storage::attachments::{Attachments, Consumers, Horizon, Payload, Source, declare};
use crate::storage::consumer::{JournalConsumer, Series, StreamConsumer};
use crate::streams::{Definition, Holding, Request, StreamManager, Table, Tier};
use crate::{
Application, DefinitionId, Error, operations::LaneSet, operations::OperationPolicy, services::Resident, state::At,
state::CaptureLimits, state::DecodeError, state::Entry, state::JournalCapture, state::Kind, state::Position, state::StateManager,
storage::Cursor,
};
const REQUESTS: usize = 64;
const STOP_GRACE: Duration = Duration::from_millis(100);
pub struct Inspection<A: Application> {
state: Arc<At<A::State>>,
reads: crate::state::Reads<A>,
}
pub struct Inspector<A: Application> {
pub state: Inspection<A>,
source: JoinHandle<()>,
}
pub struct Runtime<A: Application> {
pub state: StateManager<A>,
pub records: crate::records::RecordManager<A>,
pub streams: StreamManager<A>,
pub operations: crate::operations::OperationManager,
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
pub plugin: crate::plugin::PluginHost<A>,
#[cfg(feature = "grpc")]
grpc: crate::grpc::GrpcManager,
pub evidence: Evidence<A>,
stopping: Mutex<Stopping<A>>,
closing: AtomicBool,
storage: Arc<OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
journal_handles: Weak<()>,
participants: Participants,
notices: watch::Sender<Option<String>>,
}
pub struct Handle<A: Application> {
pub state: StateManager<A>,
pub records: crate::records::RecordManager<A>,
pub streams: StreamManager<A>,
pub operations: crate::operations::OperationManager,
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
pub plugin: crate::plugin::PluginHost<A>,
#[cfg(feature = "grpc")]
grpc: crate::grpc::GrpcManager,
pub evidence: Evidence<A>,
storage: Arc<OnceLock<crate::storage::StorageManager<crate::storage::ErasedCatalog>>>,
}
pub struct Builder<A: Application> {
origin: Origin<A::State>,
resources: Arc<A::Resources>,
inverses: Inverting<A>,
catalog: Dispatch<A, Payload<A>>,
definitions: Vec<Definition>,
attachments: Attachments<A>,
managed_state: Option<crate::storage::ReadLimits>,
record_projection: Option<crate::storage::capture::Projection>,
reconciliation: Option<boot::Reconciliation<A>>,
initializers: Vec<boot::Initializer<A>>,
snapshots: Option<crate::state::snapshot::Snapshots<A>>,
lanes: LaneSet,
operations: Vec<crate::operations::CatalogBinding>,
operation_settings: crate::operations::Settings,
operation_middleware: Vec<Arc<dyn crate::operations::middleware::Middleware>>,
services: Vec<DeclaredService<A>>,
evidence: Option<Declaration>,
#[cfg(feature = "simulation")]
simulation: crate::simulation::Configuration<A>,
#[cfg(feature = "simulation")]
live_only: Vec<&'static str>,
}
#[derive(Debug, thiserror::Error)]
#[error("{cause}")]
pub struct Incompatible {
#[source]
cause: Cause,
cleanup: Option<boot::Cleanup>,
}
struct Supervision<A: Application> {
consumers: Consumers<A>,
feed: Feed<A>,
progress: watch::Receiver<Position>,
streams: StreamManager<A>,
delivery: watch::Receiver<()>,
demand: Vec<Holding>,
writer: JoinHandle<()>,
catalog: Arc<Dispatch<A, Payload<A>>>,
inverses: Inverting<A>,
requests: mpsc::Receiver<ReadRequest<A>>,
served: mpsc::Receiver<Request>,
ready: oneshot::Sender<()>,
}
#[derive(Clone)]
struct Participants {
stopping: watch::Sender<bool>,
active: watch::Sender<usize>,
}
pub struct StopParticipant {
stopping: watch::Receiver<bool>,
active: watch::Sender<usize>,
released: bool,
}
enum Stopping<A: Application> {
Running(Option<JoinHandle<Position>>),
Closing {
cleanup: Arc<crate::supervision::cleanup::Cleanup<lifecycle::Shutdown<A>>>,
outcome: Arc<OnceLock<lifecycle::Outcome>>,
progress: watch::Sender<&'static str>,
},
}
#[allow(clippy::type_complexity)]
pub enum Catalog<A: Application> {
Transactions {
definitions: Vec<(Kind, DefinitionId, u32)>,
decode: fn(&DefinitionId, u32, &[u8]) -> Result<Option<A::Transaction>, DecodeError>,
inverses: Option<fn(&A::Transaction, &A::State) -> Option<A::Transaction>>,
},
Streams(Vec<Definition>),
Lanes(LaneSet),
}
#[derive(Debug, thiserror::Error)]
enum Cause {
#[error("the stored history cannot be replayed by this build: {0}")]
Undeclared(DecodeError),
#[error("the stored history cannot be replayed by this build: {0}")]
Unreadable(String),
#[error("the stream catalog cannot be honored: {0}")]
Dishonorable(crate::streams::Refused),
#[error("the work declarations cannot be honored: {0}")]
Work(String),
#[error("the operation declarations cannot be honored: {0}")]
Operations(#[source] crate::Error),
#[error("runtime initialization failed: {0}")]
Initialization(#[source] Box<dyn std::error::Error + Send + Sync>),
#[error("{source}; bootstrap cleanup also failed: {failure}")]
Cleanup {
#[source]
source: Box<Incompatible>,
failure: crate::Error,
},
#[error("application storage reconciliation failed: {0}")]
Reconciliation(String),
#[error("the evidence declarations cannot be honored: {0}")]
Kept(String),
#[cfg(feature = "simulation")]
#[error("this build cannot be simulated after attaching {0}")]
Simulation(String),
}
enum Turn<A: Application> {
Advanced { journal: bool, streams: bool },
Read(ReadRequest<A>),
Serve(Box<Request>),
Stopped,
}
pub trait Registration<A: Application>: Clone + Send {
fn register(self) -> Catalog<A>;
}
impl<A, S, T> Registration<A> for harmos_domain::state::TransactionDefinitions<S, T>
where
A: Application<State = S, Transaction = T>,
{
fn register(self) -> Catalog<A> {
Catalog::Transactions { definitions: self.definitions, decode: self.decode, inverses: Some(self.inverses) }
}
}
impl<A, R> Registration<A> for harmos_domain::state::RecordDefinitions<R>
where
A: Application<Record = R>,
{
fn register(self) -> Catalog<A> {
Catalog::Transactions { definitions: self.definitions, decode: |_, _, _| Ok(None), inverses: None }
}
}
impl<A: Application> Inspection<A> {
pub async fn read<R>(&self, read: impl FnOnce(&A::State) -> R) -> Result<At<R>, Error> {
Ok(At { value: read(&self.state.value), position: self.state.position })
}
pub async fn state_at(&self, origin: Origin<A::State>, at: Position) -> Result<At<A::State>, Error> {
self.reads.state(origin, at).await
}
pub async fn query<X>(&self, after: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
where
X: Clone + Send + 'static,
{
self.query_through(after, self.state.position).await
}
pub async fn query_through<X>(&self, after: Position, through: Position) -> impl Iterator<Item = Result<Entry<X>, Error>>
where
X: Clone + Send + 'static,
{
if through > self.state.position {
return vec![Err(Error::FrontierAhead)].into_iter();
}
if after >= through {
return Vec::new().into_iter();
}
match self.reads.entries(after, through).await {
Ok(entries) => entries
.into_iter()
.filter_map(|entry| {
entry
.value
.select::<X>()
.map(|value| Ok(Entry { prepared: None, position: entry.position, metadata: entry.metadata, value }))
})
.collect::<Vec<_>>()
.into_iter(),
Err(refusal) => vec![Err(refusal)].into_iter(),
}
}
pub fn applied(&self) -> Position {
self.state.position
}
pub fn stored(&self) -> Position {
self.state.position
}
}
impl<A: Application> Drop for Inspector<A> {
fn drop(&mut self) {
self.source.abort();
}
}
impl<A: Application> Clone for Handle<A> {
fn clone(&self) -> Self {
Self {
state: self.state.clone(),
records: self.records.clone(),
streams: self.streams.clone(),
operations: self.operations.clone(),
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
plugin: self.plugin.clone(),
#[cfg(feature = "grpc")]
grpc: self.grpc.clone(),
evidence: self.evidence.clone(),
storage: self.storage.clone(),
}
}
}
impl<A: Application> Handle<A> {
#[cfg(feature = "grpc")]
pub fn grpc(&self) -> &crate::grpc::GrpcManager {
&self.grpc
}
}
impl<A: Application> Runtime<A> {
#[cfg(feature = "grpc")]
pub fn grpc(&self) -> &crate::grpc::GrpcManager {
&self.grpc
}
pub fn builder(origin: impl Into<Origin<A::State>>, resources: impl Into<Arc<A::Resources>>) -> Builder<A> {
Builder {
origin: origin.into(),
resources: resources.into(),
inverses: none::<A>,
catalog: Dispatch::default(),
definitions: Vec::new(),
attachments: Attachments::new(),
managed_state: None,
record_projection: None,
reconciliation: None,
initializers: Vec::new(),
snapshots: None,
lanes: LaneSet::default(),
operations: Vec::new(),
operation_settings: crate::operations::Settings::default(),
operation_middleware: Vec::new(),
services: Vec::new(),
evidence: None,
#[cfg(feature = "simulation")]
simulation: crate::simulation::Configuration::new(),
#[cfg(feature = "simulation")]
live_only: Vec::new(),
}
}
pub async fn inspect(origin: impl Into<Origin<A::State>>, source: impl Source<A>) -> Result<Inspector<A>, Incompatible> {
let mut attachments = Attachments::new();
attachments.recover(source);
let mut consumers = attachments.supervised();
let recovery = consumers.load().await?;
let origin = origin.into();
let anchor = origin.position;
let mut state = At { value: origin.value, position: anchor };
let entries = recovery.entries.into_iter().filter(|entry| entry.position > anchor).collect();
replay(None, entries, &mut state, &mut Undos::default(), none::<A>, 0).map_err(Incompatible::of)?;
state.position = state.position.max(recovery.frontier).max(anchor);
let (reads, requests) = crate::state::Reads::channel(REQUESTS);
let source = tokio::spawn(inspect_reads(consumers, requests));
Ok(Inspector { state: Inspection { state: Arc::new(state), reads }, source })
}
pub fn stop_participant(&self) -> StopParticipant {
self.participants.register()
}
pub fn notices(&self) -> watch::Receiver<Option<String>> {
self.notices.subscribe()
}
pub fn handle(&self) -> Handle<A> {
Handle {
state: self.state.clone(),
records: self.records.clone(),
streams: self.streams.clone(),
operations: self.operations.clone(),
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
plugin: self.plugin.clone(),
#[cfg(feature = "grpc")]
grpc: self.grpc.clone(),
evidence: self.evidence.clone(),
storage: self.storage.clone(),
}
}
}
impl<A: Application> Registration<A> for Vec<Definition> {
fn register(self) -> Catalog<A> {
Catalog::Streams(self)
}
}
impl<A: Application, const N: usize> Registration<A> for [Definition; N] {
fn register(self) -> Catalog<A> {
Catalog::Streams(self.into())
}
}
impl<A: Application> Registration<A> for LaneSet {
fn register(self) -> Catalog<A> {
Catalog::Lanes(self)
}
}
impl<A: Application> Builder<A> {
pub fn register<C: Registration<A> + 'static>(mut self, catalog: C) -> Self {
#[cfg(feature = "simulation")]
self.simulation.register(catalog.clone());
match catalog.register() {
Catalog::Transactions { definitions, decode, inverses } => {
if let Some(inverses) = inverses {
self.inverses = inverses;
}
for (kind, id, version) in definitions {
declare(&mut self.catalog, kind, id, version, decode);
}
}
Catalog::Streams(definitions) => self.definitions.extend(definitions),
Catalog::Lanes(declared) => self.lanes.extend(declared),
}
self
}
pub fn operation<T>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<T::Resources>) -> Self
where
T: for<'ctx> crate::operations::Operation<Context<'ctx> = crate::operations::Context<'ctx>>,
{
let binding = crate::operations::CatalogBinding::operation::<T>(owner.into(), lane, resources);
self.bind_operations(binding)
}
pub fn operations<C, R>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<R>) -> Self
where
C: crate::catalog::Catalog<crate::operations::AuthoredDeclarations>
+ crate::catalog::Catalog<crate::operations::Implementations<R>>,
R: Send + Sync + 'static,
{
let binding = crate::operations::CatalogBinding::local::<C, R>(owner.into(), lane, resources);
self.bind_operations(binding)
}
pub fn wire_operations<C, R>(self, owner: impl Into<String>, lane: impl crate::operations::Lane, resources: Arc<R>) -> Self
where
C: crate::catalog::Catalog<crate::operations::AuthoredDeclarations>
+ crate::catalog::Catalog<crate::operations::WireImplementations<R>>,
R: Send + Sync + 'static,
{
let binding = crate::operations::CatalogBinding::wire::<C, R>(owner.into(), lane, resources);
self.bind_operations(binding)
}
pub fn operation_middleware(mut self, middleware: Arc<dyn crate::operations::middleware::Middleware>) -> Self {
#[cfg(feature = "simulation")]
self.simulation.operation_middleware(middleware.clone());
self.operation_middleware.push(middleware);
self
}
pub fn operation_settings(mut self, settings: crate::operations::Settings) -> Self {
#[cfg(feature = "simulation")]
self.simulation.operation_settings(settings);
self.operation_settings = settings;
self
}
pub(crate) fn bind_operations(mut self, binding: crate::operations::CatalogBinding) -> Self {
#[cfg(feature = "simulation")]
self.simulation.operations(binding.clone());
self.operations.push(binding);
self
}
pub fn service<R>(self, name: impl Into<String>, resident: R) -> Self
where
R: Resident<A>,
{
self.service_with(name, crate::services::POLICY, resident)
}
pub fn service_with<R>(mut self, name: impl Into<String>, policy: OperationPolicy, resident: R) -> Self
where
R: Resident<A>,
{
#[cfg(feature = "simulation")]
self.live_only.push("resident services");
let definition = ServiceDefinition { id: name.into(), policy };
self.services.push(DeclaredService::new(definition, resident));
self
}
pub fn evidence(mut self, vault: impl Vault, kinds: impl IntoIterator<Item = (DefinitionId, u32)>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("an evidence vault");
self.evidence = Some(Declaration::new(vault, kinds));
self
}
pub fn attach(mut self, consumer: impl JournalConsumer<A>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("an attached fold");
self.attachments.attach(consumer);
self
}
pub fn attach_streams(mut self, consumer: impl StreamConsumer<A>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("an attached stream fold");
self.attachments.attach_streams(consumer);
self
}
pub fn series(mut self, series: impl Series<A>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("a series copy");
self.attachments.series(series);
self
}
pub fn recover(mut self, source: impl Source<A>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("a recovery source");
self.attachments.recover(source);
self
}
pub fn snapshot(mut self, snapshot: impl JournalConsumer<A>) -> Self {
#[cfg(feature = "simulation")]
self.live_only.push("a snapshot fold");
self.attachments.snapshot(snapshot);
self
}
pub fn require_capture(mut self, limits: CaptureLimits, consumer: impl JournalCapture<A>) -> Result<Self, crate::state::CaptureError> {
#[cfg(feature = "simulation")]
self.live_only.push("required capture");
self.attachments.require_capture(limits, consumer)?;
Ok(self)
}
pub fn retain(mut self, horizon: Horizon) -> Self {
self.attachments.retain(horizon);
self
}
pub fn tail(mut self, entries: usize) -> Self {
self.attachments.tail(entries);
self
}
#[cfg(feature = "simulation")]
pub fn simulated_service<R>(mut self, name: impl Into<String>, resident: R) -> Self
where
R: Resident<A> + Clone,
{
self.simulation.service(name, resident);
self
}
#[cfg(feature = "simulation")]
pub fn simulated_service_with<R>(mut self, name: impl Into<String>, policy: OperationPolicy, resident: R) -> Self
where
R: Resident<A> + Clone,
{
self.simulation.service_with(name, policy, resident);
self
}
#[cfg(feature = "simulation")]
pub fn simulated_evidence(mut self, vault: impl Vault + Clone, kinds: impl IntoIterator<Item = (DefinitionId, u32)>) -> Self {
self.simulation.evidence(vault, kinds);
self
}
#[cfg(feature = "simulation")]
pub fn entropy_seed(mut self, seed: u64) -> Self {
self.simulation.entropy_seed(seed);
self
}
#[cfg(feature = "simulation")]
pub fn simulate(self) -> Result<crate::simulation::SimulatedRuntime<A>, Incompatible>
where
A::State: Clone + fmt::Debug + PartialEq,
A::Transaction: Clone + fmt::Debug,
A::Record: Clone + fmt::Debug,
{
if !self.live_only.is_empty() {
return Err(Incompatible::simulation(self.live_only.join(", ")));
}
self.simulation.simulate(self.origin, self.resources)
}
#[tracing::instrument(level = "debug", skip_all)]
pub fn runtime(self) -> Startup<A> {
Startup::new(self, None)
}
pub fn managed_state(mut self, limits: crate::storage::ReadLimits) -> Self {
self.managed_state = Some(limits);
#[cfg(feature = "simulation")]
self.live_only.push("managed storage recovery");
self
}
pub fn store_records(mut self, limits: crate::storage::ReadLimits, projection: impl crate::storage::RecordProjection) -> Self {
self.record_projection = Some(crate::storage::capture::Projection { limits, handler: Box::new(projection) });
#[cfg(feature = "simulation")]
self.live_only.push("managed primary record capture");
self
}
pub fn managed_snapshots(mut self, policy: crate::state::SnapshotPolicy) -> Self
where
A::State: serde::Serialize + serde::de::DeserializeOwned,
{
self.snapshots = Some(crate::state::snapshot::Snapshots::new(policy));
#[cfg(feature = "simulation")]
self.live_only.push("managed state snapshots");
self
}
pub fn reconcile<F, Fut>(mut self, reconcile: F) -> Self
where
F: FnOnce(StateManager<A>, crate::storage::StorageManager<crate::storage::ErasedCatalog>) -> Fut + Send + 'static,
Fut: Future<Output = crate::Result<()>> + Send + 'static,
{
self.reconciliation = Some(Box::new(move |journal, storage| Box::pin(reconcile(journal, storage))));
#[cfg(feature = "simulation")]
self.live_only.push("saved application data reconciliation");
self
}
pub fn initialize<F, Fut, E>(mut self, timeout: Duration, initialize: F) -> Self
where
F: Fn(Handle<A>) -> Fut + Clone + Send + Sync + 'static,
Fut: Future<Output = std::result::Result<(), E>> + Send + 'static,
E: std::error::Error + Send + Sync + 'static,
{
#[cfg(feature = "simulation")]
self.simulation.initialize(timeout, initialize.clone());
self.initializers.push(boot::Initializer::new(timeout, initialize));
self
}
pub fn runtime_with_storage<C>(self, storage: &crate::storage::StorageManager<C>) -> Startup<A> {
Startup::new(self, Some(storage.erased()))
}
async fn boot(
self, storage: Option<&crate::storage::StorageManager<crate::storage::ErasedCatalog>>, cleanup: BootstrapCleanup,
) -> Result<Runtime<A>, Incompatible> {
if self
.initializers
.iter()
.any(|initializer| initializer.timeout.is_zero() || tokio::time::Instant::now().checked_add(initializer.timeout).is_none())
{
let source: crate::Error = Error::InvalidInput {
field: "runtime initialization timeout",
reason: "requires a positive representable finite duration",
};
return Err(Incompatible::initialization(Box::new(source)));
}
if self.snapshots.is_some() && self.managed_state.is_none() {
return Err(Incompatible::unreadable("managed snapshots require managed state"));
}
if self.reconciliation.is_some() && self.managed_state.is_none() {
return Err(Incompatible::reconciliation("reconciliation requires managed state"));
}
if self.record_projection.is_some() && self.managed_state.is_none() {
return Err(Incompatible::unreadable("primary record capture requires managed state"));
}
if self.managed_state.is_some() && (storage.is_none() || self.attachments.has_recovery_owner()) {
return Err(Incompatible::unreadable("managed state requires storage and exclusive recovery ownership"));
}
let mut boot = boot::Boot::new(storage, cleanup.clone()).map_err(Incompatible::work)?;
if let Err(error) = self.boot_owned(&mut boot).await {
return match boot.close().await {
Ok(()) => Err(error),
Err(failure) => {
let failure = Arc::new(failure);
let retained = boot.retain_cleanup();
cleanup.failed(retained.clone(), Arc::clone(&failure));
Err(Incompatible {
cause: Cause::Cleanup {
source: Box::new(error),
failure: Error::TaskFailed { domain: "bootstrap cleanup", source: failure },
},
cleanup: retained,
})
}
};
}
Ok(boot.finish())
}
async fn boot_owned(self, boot: &mut boot::Boot<A>) -> Result<(), Incompatible> {
boot.acquire().await.map_err(Incompatible::work)?;
let Self {
origin,
resources,
inverses,
catalog,
definitions,
mut attachments,
managed_state,
record_projection,
reconciliation,
initializers,
snapshots,
lanes,
operations: bindings,
operation_settings,
operation_middleware,
services,
evidence,
..
} = self;
let tail = attachments.tail_window()?;
let mut consumers = attachments.supervised();
let recovery = consumers.load().await?;
let (resume, selections) = consumers.recordings().fold((Cursor::ORIGIN, Vec::new()), |(resume, mut selections), recording| {
selections.push(recording.selections());
(resume.max(recording.resume()), selections)
});
let anchor = origin.position;
let mut state = At { value: origin.value, position: anchor };
let mut undos = Undos::default();
let entries = recovery.entries.into_iter().filter(|entry| entry.position > anchor).collect();
let mut recovered = replay(Some(&catalog), entries, &mut state, &mut undos, inverses, tail).map_err(Incompatible::of)?;
let stored = recovery.encoded.into_iter().filter(|(position, _)| *position > anchor).collect();
state.position = state.position.max(recovery.frontier).max(anchor);
let mut managed_durable = None;
if let Some(limits) = managed_state {
let storage = boot.storage.as_ref().expect("managed state requires storage");
let loaded = StateManager::recover(storage, state, &catalog, inverses, tail, limits, snapshots.as_ref()).await?;
state = loaded.state;
undos = loaded.undos;
recovered = loaded.tail;
managed_durable = Some(loaded.durable);
attachments.capture = Some(crate::state::capture::Declaration::Managed {
storage: storage.clone(),
receipts: loaded.receipts,
projection: record_projection,
snapshots,
});
}
let windows = Table::resuming(definitions, state.position, resume).map_err(Incompatible::dishonored)?;
let demand = windows.hold(selections).map_err(Incompatible::dishonored)?;
let replayed = recovered.clone();
consumers.redelivering(recovered.clone(), stored, state.position);
let durability = managed_durable.or_else(|| consumers.durability(state.position));
let tail = Tail::recovered(state.position, tail, recovered);
let catalog = Arc::new(catalog);
let (journal, writer, requests) = StateManager::start(
state,
inverses,
undos,
REQUESTS,
tail,
durability,
Arc::clone(&catalog),
attachments.capture.take(),
boot.storage.clone(),
crate::state::Timestamp::now,
);
let journal_handles = journal.handle_probe();
let records = crate::records::RecordManager::new(journal.clone());
boot.writer = Some(writer);
boot.state = Some(journal.clone());
let evidence = Evidence::wired(journal.clone(), evidence, &replayed).map_err(Incompatible::kept)?;
let (tier, served) = Tier::channel(REQUESTS);
let streams = StreamManager::wired(windows, journal.progress(), tier);
let delivery = streams.delivery();
operation_settings.validate().map_err(|source| Incompatible { cause: Cause::Operations(source), cleanup: None })?;
let lanes = crate::operations::lane::Lanes::freeze(lanes).map_err(Incompatible::work)?;
let storage = Arc::new(boot.storage.clone().map(OnceLock::from).unwrap_or_default());
let handles = crate::services::Handles {
storage: storage.clone(),
state: journal.clone(),
records: records.clone(),
streams: streams.clone(),
evidence: evidence.clone(),
};
let instances = crate::operations::instances::Instances::start(lanes, Box::new(handles.clone()), resources, &operation_settings);
let mut reserved = std::collections::BTreeSet::new();
for service in &services {
crate::services::validate(&service.definition()).map_err(Incompatible::work)?;
if !reserved.insert(service.id().to_owned()) {
let duplicate: crate::Error = crate::Error::AlreadyRegistered { domain: "service", name: service.id().to_owned() };
return Err(Incompatible::work(duplicate));
}
}
let operations = crate::operations::OperationManager::new(instances, &operation_settings, reserved, boot.identity.clone());
for middleware in operation_middleware {
operations.register_middleware(middleware).map_err(Incompatible::work)?;
}
for binding in bindings {
binding.bind(&operations).map_err(|source| Incompatible { cause: Cause::Operations(source), cleanup: None })?;
}
if let Some(storage) = &boot.storage {
storage.start_on::<A>(&operations).await.map_err(Incompatible::work)?;
}
let (ready, booted) = oneshot::channel();
let supervisor = tokio::spawn(supervise(Supervision {
consumers,
feed: journal.feed(),
progress: journal.progress(),
streams: streams.clone(),
delivery,
demand,
writer: boot.writer.take().expect("boot owns writer"),
catalog,
inverses,
requests,
served,
ready,
}));
#[cfg(feature = "grpc")]
let grpc = crate::grpc::GrpcManager::new::<A>(&operations);
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
let plugin = crate::plugin::PluginHost::new(
operations.clone(),
#[cfg(feature = "plugin-sidecar")]
handles,
)
.map_err(Incompatible::work)?;
boot.runtime = Some(Runtime {
state: journal,
records,
streams,
operations,
#[cfg(all(feature = "plugin", not(target_arch = "wasm32")))]
plugin,
#[cfg(feature = "grpc")]
grpc,
evidence,
stopping: Mutex::new(Stopping::Running(Some(supervisor))),
closing: AtomicBool::new(false),
storage,
journal_handles,
participants: Participants::new(),
notices: watch::channel(None).0,
});
boot.state.take();
booted.await.map_err(Incompatible::work)?;
if let Some(reconcile) = reconciliation {
let runtime = boot.runtime.as_ref().expect("boot owns runtime");
let storage = runtime.storage().expect("reconciliation requires storage");
reconcile(runtime.state.clone(), storage.clone()).await.map_err(Incompatible::reconciliation)?;
storage.flush().await.map_err(Incompatible::reconciliation)?;
storage.flush_recovery().await.map_err(Incompatible::reconciliation)?;
}
let operations = &boot.runtime.as_ref().expect("runtime is owned during boot").operations;
for service in services {
let lease = operations.start_service(service, false, true).map_err(Incompatible::work)?;
operations.hold_resident(lease);
}
for initializer in initializers {
let handle = boot.runtime.as_ref().expect("runtime is owned during initialization").handle();
let initialize = (initializer.initialize)(handle);
match tokio::time::timeout(initializer.timeout, initialize).await {
Ok(result) => result.map_err(Incompatible::initialization)?,
Err(source) => return Err(Incompatible::initialization(Box::new(source))),
}
}
Ok(())
}
}
impl Incompatible {
fn initialization(source: Box<dyn std::error::Error + Send + Sync>) -> Self {
Self { cause: Cause::Initialization(source), cleanup: None }
}
pub fn cleanup_error(&self) -> Option<&crate::Error> {
match (&self.cleanup, &self.cause) {
(Some(_), Cause::Cleanup { failure, .. }) => Some(failure),
_ => None,
}
}
pub async fn retry_cleanup(&mut self) -> crate::Result<()> {
if let Some(cleanup) = &self.cleanup {
cleanup.close().await?;
self.cleanup.take();
}
Ok(())
}
fn reconciliation(cause: impl fmt::Display) -> Self {
Self { cause: Cause::Reconciliation(cause.to_string()), cleanup: None }
}
pub fn unreadable(cause: impl fmt::Display) -> Self {
Self { cause: Cause::Unreadable(cause.to_string()), cleanup: None }
}
fn of(cause: DecodeError) -> Self {
Self { cause: Cause::Undeclared(cause), cleanup: None }
}
pub(crate) fn dishonored(cause: crate::streams::Refused) -> Self {
Self { cause: Cause::Dishonorable(cause), cleanup: None }
}
pub(crate) fn work(cause: impl fmt::Display) -> Self {
Self { cause: Cause::Work(cause.to_string()), cleanup: None }
}
#[cfg(feature = "simulation")]
fn simulation(live_only: String) -> Self {
Self { cause: Cause::Simulation(live_only), cleanup: None }
}
fn kept(cause: String) -> Self {
Self { cause: Cause::Kept(cause), cleanup: None }
}
}
#[cfg(feature = "plugin-sidecar")]
impl From<harmos_sidecar::ExecutableError> for Incompatible {
fn from(error: harmos_sidecar::ExecutableError) -> Self {
Self::work(error)
}
}
impl Participants {
fn new() -> Self {
Self { stopping: watch::channel(false).0, active: watch::channel(0).0 }
}
fn register(&self) -> StopParticipant {
self.active.send_modify(|active| *active += 1);
StopParticipant { stopping: self.stopping.subscribe(), active: self.active.clone(), released: false }
}
async fn stop(&self) {
self.stopping.send_replace(true);
let mut active = self.active.subscribe();
while *active.borrow_and_update() != 0 {
if active.changed().await.is_err() {
break;
}
}
}
}
impl StopParticipant {
pub fn is_stopping(&self) -> bool {
*self.stopping.borrow()
}
pub async fn stopping(&mut self) {
while !*self.stopping.borrow_and_update() {
if self.stopping.changed().await.is_err() {
break;
}
}
}
pub fn finish(mut self) {
self.release();
}
fn release(&mut self) {
if !self.released {
self.active.send_modify(|active| *active = active.saturating_sub(1));
self.released = true;
}
}
}
impl Drop for StopParticipant {
fn drop(&mut self) {
self.release();
}
}
fn none<A: Application>(_change: &A::Transaction, _before: &A::State) -> Option<A::Transaction> {
None
}
async fn supervise<A: Application>(supervision: Supervision<A>) -> Position {
let Supervision {
mut consumers,
feed,
mut progress,
streams,
mut delivery,
demand: _demand,
mut writer,
catalog,
inverses,
mut requests,
mut served,
ready,
} = supervision;
let booted = *progress.borrow_and_update();
consumers.drain(&feed, booted).await;
deliver(&mut consumers, &streams).await;
stand_windows(&streams, &mut consumers, booted).await;
let _ = ready.send(());
loop {
match turn(&mut progress, &mut delivery, &mut requests, &mut served, &mut writer).await {
Turn::Advanced { journal, streams: rows } => {
if journal {
let applied = *progress.borrow_and_update();
consumers.drain(&feed, applied).await;
}
if rows {
delivery.borrow_and_update();
deliver(&mut consumers, &streams).await;
}
}
Turn::Read(request) => answer(request, &mut consumers, Some(&catalog), inverses).await,
Turn::Serve(request) => serve(*request, &mut consumers).await,
Turn::Stopped => break,
}
}
let applied = *progress.borrow();
consumers.drain(&feed, applied).await;
deliver(&mut consumers, &streams).await;
consumers.flush().await;
consumers.settle(applied).await
}
async fn deliver<A: Application>(consumers: &mut Consumers<A>, streams: &StreamManager<A>) {
for recording in consumers.recordings() {
let resume = recording.resume();
let batch = streams.batch(resume);
if batch.reaches() > resume {
recording.drain(batch).await;
}
}
}
async fn stand_windows<A: Application>(streams: &StreamManager<A>, consumers: &mut Consumers<A>, booted: Position) {
for wired in streams.declared() {
let durable = consumers.sealed(wired.definition.clone(), &[]).await;
if durable > Position::ORIGIN && durable <= booted {
wired.window.stands_on(durable);
}
}
}
async fn turn<A: Application>(
progress: &mut watch::Receiver<Position>, delivery: &mut watch::Receiver<()>, requests: &mut mpsc::Receiver<ReadRequest<A>>,
served: &mut mpsc::Receiver<Request>, writer: &mut JoinHandle<()>,
) -> Turn<A> {
let mut advanced = Box::pin(progress.changed());
let mut published = Box::pin(delivery.changed());
std::future::poll_fn(|context| {
if Pin::new(&mut *writer).poll(context).is_ready() {
return Poll::Ready(Turn::Stopped);
}
let journal = match advanced.as_mut().poll(context) {
Poll::Ready(Ok(())) => true,
Poll::Ready(Err(_)) => return Poll::Ready(Turn::Stopped),
Poll::Pending => false,
};
let streams = match published.as_mut().poll(context) {
Poll::Ready(Ok(())) => true,
Poll::Ready(Err(_)) => return Poll::Ready(Turn::Stopped),
Poll::Pending => false,
};
if journal || streams {
Poll::Ready(Turn::Advanced { journal, streams })
} else {
match requests.poll_recv(context) {
Poll::Ready(Some(request)) => Poll::Ready(Turn::Read(request)),
Poll::Ready(None) | Poll::Pending => match served.poll_recv(context) {
Poll::Ready(Some(request)) => Poll::Ready(Turn::Serve(Box::new(request))),
Poll::Ready(None) | Poll::Pending => Poll::Pending,
},
}
}
})
.await
}
async fn answer<A: Application>(
request: ReadRequest<A>, consumers: &mut Consumers<A>, catalog: Option<&Dispatch<A, Payload<A>>>, inverses: Inverting<A>,
) {
match request {
ReadRequest::Entries { after, through, reply } => {
let answer = consumers.read(after, through).await.map(|recovery| {
recovery
.entries
.into_iter()
.map(|envelope| Entry {
prepared: envelope.payload.prepared.clone(),
position: envelope.position,
metadata: envelope.metadata,
value: envelope.payload.value,
})
.collect()
});
let _ = reply.send(answer);
}
ReadRequest::State { origin, at, reply } => {
let answer = async {
let recovery = consumers.read(origin.position, at).await?;
if recovery
.commits
.iter()
.copied()
.chain(recovery.entries.iter().filter_map(|entry| entry.metadata.commit))
.any(|(first, end)| (first <= at && at < end) || (first <= origin.position && origin.position < end))
{
return Err(crate::Error::IncompleteCommit);
}
if recovery.frontier < at {
return Err(crate::Error::BeyondTail);
}
let mut state = At { value: origin.value, position: origin.position };
replay(catalog, recovery.entries, &mut state, &mut Undos::default(), inverses, 0).map_err(|_| crate::Error::Unavailable)?;
state.position = at;
Ok(state)
}
.await;
let _ = reply.send(answer);
}
}
}
async fn serve<A: Application>(request: Request, consumers: &mut Consumers<A>) {
match request {
Request::Reader { stream, track, start, end, limits, permit, reply } => {
if !reply.is_closed() {
let answer = consumers.row_reader(stream, &track, start, end, limits).await;
let _ = reply.send(crate::streams::Opened { source: answer, permit });
}
}
Request::Rows { stream, track, start, end, reply } => {
let answer = consumers.served(stream, &track, start, end).await;
let _ = reply.send(answer);
}
Request::Buckets { stream, track, start, end, budget, reply } => {
let answer = consumers.surveyed(stream, &track, start, end, budget).await;
let _ = reply.send(answer);
}
Request::Aggregate { stream, track, start, end, kind, reply } => {
let answer = consumers.aggregated(stream, &track, start, end, kind).await;
let _ = reply.send(answer);
}
Request::Seal { stream, track, reply } => {
let _ = reply.send(consumers.sealed(stream, &track).await);
}
}
}
async fn inspect_reads<A: Application>(mut consumers: Consumers<A>, mut requests: mpsc::Receiver<ReadRequest<A>>) {
while let Some(request) = requests.recv().await {
answer(request, &mut consumers, None, none::<A>).await;
}
}