use std::collections::HashSet;
use std::num::NonZeroUsize;
use std::sync::Arc;
use std::thread::JoinHandle;
use frame_core::component::ComponentId;
use frame_core::event::LifecycleSubscription;
use crate::announcer::Announcer;
use crate::config::FrameConfig;
use crate::doc_binding::DocBinding;
use crate::embedded::EmbeddedLiminal;
use crate::error::HostError;
use crate::page::PageServer;
use crate::runtime::{EventLoggerHandle, HostRuntime};
use crate::serve::{ServeStop, bind};
use crate::spec::{AppSpec, ComponentInstall, ReadinessProbe};
use crate::truth::AppTruth;
const ANNOUNCER_EVENT_BUFFER: usize = 256;
pub struct Application {
runtime: HostRuntime,
logger: EventLoggerHandle,
liminal: EmbeddedLiminal,
doc_binding: Option<DocBinding>,
announcer: Option<Announcer>,
bus_endpoint: String,
components: HashSet<ComponentId>,
truth: Arc<AppTruth>,
truth_recorder: JoinHandle<()>,
}
impl Application {
pub fn boot(config: &FrameConfig, spec: AppSpec) -> Result<Self, HostError> {
if spec.components.is_empty() {
return Err(HostError::ConfigContract {
detail: "the application spec installs no components: a host with nothing to \
host is a composition error, not an empty stack"
.to_owned(),
});
}
let components: HashSet<ComponentId> = spec
.components
.iter()
.map(|install| install.meta.id)
.collect();
let mut runtime = HostRuntime::new(spec.policy)?;
let announcer_subscription: Option<LifecycleSubscription> = match &config.frame.channel {
Some(_) => {
let capacity = NonZeroUsize::new(ANNOUNCER_EVENT_BUFFER)
.ok_or(HostError::SynchronizationPoisoned)?;
Some(runtime.registry().subscribe(capacity)?)
}
None => None,
};
let logger = runtime.spawn_event_logger(components.clone())?;
let (truth, truth_recorder) = AppTruth::spawn(runtime.registry(), components.clone())?;
install_and_verify_ready(&mut runtime, spec.components, spec.readiness)?;
let liminal = match EmbeddedLiminal::boot(&config.bus) {
Ok(liminal) => liminal,
Err(error) => {
if let Err(teardown) = runtime.shutdown(logger) {
tracing::error!(
%teardown,
"frame-core teardown after bus boot failure also failed"
);
}
join_truth_recorder(truth_recorder, "bus boot failure");
return Err(error);
}
};
let bus_endpoint = liminal.websocket_endpoint();
tracing::info!(
endpoint = %bus_endpoint,
"embedded bus component booted; /frame/config.json advertises its real bound WebSocket address as busEndpoint"
);
let doc_binding = match DocBinding::boot(&liminal, config) {
Ok(binding) => binding,
Err(error) => {
teardown_after_boot_failure(
None,
liminal,
runtime,
logger,
truth_recorder,
"document-service",
);
return Err(error);
}
};
let announcer = match connect_announcer(config, &liminal, announcer_subscription) {
Ok(announcer) => announcer,
Err(error) => {
teardown_after_boot_failure(
doc_binding,
liminal,
runtime,
logger,
truth_recorder,
"announcer connect",
);
return Err(error);
}
};
if let Some(announcer) = &announcer {
announcer.attach_truth(Arc::clone(&truth));
}
if let Err(error) = announce_boot_facts(spec.announce, &runtime, announcer.as_ref()) {
teardown_after_boot_failure(
doc_binding,
liminal,
runtime,
logger,
truth_recorder,
"fact announcement",
);
return Err(error);
}
Ok(Self {
runtime,
logger,
liminal,
doc_binding,
announcer,
bus_endpoint,
components,
truth,
truth_recorder,
})
}
pub fn serve(
&self,
config: &FrameConfig,
page: PageServer,
external_stop: impl Future<Output = ()> + Send + 'static,
) -> Result<ServeStop, HostError> {
let tokio_runtime = tokio::runtime::Builder::new_multi_thread()
.enable_all()
.build()
.map_err(|source| HostError::AsyncRuntime { source })?;
tokio_runtime.block_on(async {
let bound = bind(
config,
&self.liminal,
self.bus_endpoint.clone(),
&self.truth,
page,
)
.await?;
if let Some(announcer) = &self.announcer {
announcer.start(self.components.clone())?;
}
bound.serve(external_stop).await
})
}
pub fn shutdown(self) -> Result<(), HostError> {
if let Some(announcer) = &self.announcer {
announcer.close_intake();
}
let doc_shutdown = match self.doc_binding {
None => Ok(()),
Some(binding) => binding.shutdown(),
};
if let Err(ref error) = doc_shutdown {
tracing::error!(%error, "document-service shutdown failed");
}
let host_shutdown = self.runtime.shutdown(self.logger);
if let Err(ref error) = host_shutdown {
tracing::error!(%error, "frame-core ordered shutdown failed");
}
let announcer_shutdown = match self.announcer {
None => Ok(()),
Some(announcer) => {
if host_shutdown.is_ok() {
match announcer.stop() {
Ok(None) => Ok(()),
Ok(Some(death)) => {
tracing::warn!(
death,
"the application-event announcer had died at runtime; its \
recorded death detail is reported here at teardown"
);
Ok(())
}
Err(error) => Err(error),
}
} else {
tracing::error!(
"component shutdown failed; abandoning the announcer pump un-joined \
rather than hanging teardown on transitions that will never arrive"
);
drop(announcer);
Ok(())
}
}
};
if let Err(ref error) = announcer_shutdown {
tracing::error!(%error, "announcer shutdown failed");
}
let truth_shutdown = if host_shutdown.is_ok() {
self.truth_recorder
.join()
.map_err(|_| HostError::TruthRecorderPanicked)
} else {
tracing::error!(
"component shutdown failed; abandoning the app-truth recorder un-joined rather \
than hanging teardown on transitions that will never arrive"
);
drop(self.truth_recorder);
Ok(())
};
if let Err(ref error) = truth_shutdown {
tracing::error!(%error, "app-truth recorder shutdown failed");
}
let liminal_shutdown = self.liminal.shutdown();
if let Err(ref error) = liminal_shutdown {
tracing::error!(%error, "embedded bus graceful shutdown failed");
}
doc_shutdown?;
host_shutdown?;
announcer_shutdown?;
truth_shutdown?;
liminal_shutdown?;
Ok(())
}
}
fn install_and_verify_ready(
runtime: &mut HostRuntime,
components: Vec<ComponentInstall>,
readiness: ReadinessProbe,
) -> Result<(), HostError> {
if let Err(error) = runtime.install(components) {
runtime.abandon_after_boot_failure();
return Err(error);
}
if let Err(source) = readiness(runtime.registry()) {
runtime.abandon_after_boot_failure();
return Err(HostError::Application {
stage: "readiness",
source,
});
}
Ok(())
}
fn connect_announcer(
config: &FrameConfig,
liminal: &EmbeddedLiminal,
subscription: Option<LifecycleSubscription>,
) -> Result<Option<Announcer>, HostError> {
let (Some(channel), Some(subscription)) = (&config.frame.channel, subscription) else {
tracing::warn!(
"no [frame].channel is declared: the application-event announcer is disabled and \
the served page will observe no host events on any channel; declare \
[frame].channel to announce lifecycle transitions, capability denials, and \
application facts"
);
return Ok(None);
};
let auth_token = config
.bus
.auth
.as_ref()
.map(|auth| auth.token.clone().into_bytes());
Announcer::connect(
&liminal.tcp_addr().to_string(),
auth_token.as_deref(),
channel.clone(),
subscription,
)
.map(Some)
}
fn announce_boot_facts(
announce: crate::spec::FactAnnouncement,
runtime: &HostRuntime,
announcer: Option<&Announcer>,
) -> Result<(), HostError> {
let Some(announcer) = announcer else {
tracing::warn!("application facts are not announced: no [frame].channel is declared");
return Ok(());
};
announce(runtime.registry(), announcer).map_err(|source| HostError::Application {
stage: "announce",
source,
})
}
fn teardown_after_boot_failure(
doc_binding: Option<DocBinding>,
liminal: EmbeddedLiminal,
runtime: HostRuntime,
logger: EventLoggerHandle,
truth_recorder: JoinHandle<()>,
failed_stage: &str,
) {
if let Some(binding) = doc_binding
&& let Err(teardown) = binding.shutdown()
{
tracing::error!(
%teardown,
failed_stage,
"document-service teardown after boot failure also failed"
);
}
if let Err(teardown) = liminal.shutdown() {
tracing::error!(
%teardown,
failed_stage,
"embedded bus teardown after boot failure also failed"
);
}
if let Err(teardown) = runtime.shutdown(logger) {
tracing::error!(
%teardown,
failed_stage,
"frame-core teardown after boot failure also failed"
);
}
join_truth_recorder(truth_recorder, failed_stage);
}
fn join_truth_recorder(truth_recorder: JoinHandle<()>, failed_stage: &str) {
if truth_recorder.join().is_err() {
tracing::error!(
failed_stage,
"app-truth recorder thread panicked during teardown after a boot failure"
);
}
}
const DEV_PUMP_WAIT: std::time::Duration = std::time::Duration::from_secs(1);
pub fn run_dev_application(
mut config: FrameConfig,
spec: AppSpec,
page: PageServer,
dev: crate::dev::DevWiring,
) -> Result<(), HostError> {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
config.finalize_page_origins(page.local_addr());
let app = Application::boot(&config, spec)?;
let endpoint = app.liminal.tcp_addr().to_string();
let attachment = frame_conv::BusAttachment {
endpoint: endpoint.clone(),
session_capacity: 1,
operation_window_millis: 10,
inbound_buffer: 16,
answer_window: std::time::Duration::from_secs(30),
};
let (mut conversation, conversation_id) = crate::dev::DevManagementConversation::open(
&attachment,
crate::dev::DevResumeStore::default(),
)
.map_err(|error| HostError::DevManagement {
detail: error.to_string(),
})?;
println!(
"{}",
crate::dev::format_management_line(&endpoint, conversation_id)
);
let mut engine = crate::dev::BarrierEngine::new(crate::dev::NodeMode::Dev, dev.boot);
let support = app
.runtime
.first_support_ref()
.ok_or_else(|| HostError::DevManagement {
detail: "dev wiring requires the application's FFI support module".to_owned(),
})?;
let mut control = crate::dev::DevNodeControl::new(
app.runtime.registry_handle(),
app.runtime.support_swapper(),
dev.meta,
support,
dev.mailbox_proof,
dev.content_proof,
);
let close = Arc::new(AtomicBool::new(false));
let close_for_adapter = Arc::clone(&close);
let (exit_tx, exit_rx) = tokio::sync::oneshot::channel::<crate::dev::ServeExit>();
let adapter = std::thread::Builder::new()
.name("frame-dev-management".to_owned())
.spawn(move || {
let exit = crate::dev::serve(
&mut conversation,
&mut engine,
&mut control,
&close_for_adapter,
DEV_PUMP_WAIT,
);
let _ = exit_tx.send(exit);
})
.map_err(|source| HostError::DevManagement {
detail: format!("adapter thread spawn: {source}"),
})?;
let serve_outcome = app.serve(&config, page, async {
match exit_rx.await {
Ok(exit) => {
tracing::info!(?exit, "dev management adapter exited; ordered stop");
}
Err(_closed) => {
tracing::info!("dev management adapter closed without an exit report");
}
}
});
close.store(true, Ordering::Release);
if adapter.join().is_err() {
tracing::error!("dev management adapter thread panicked during teardown");
}
let shutdown_outcome = app.shutdown();
let stop = serve_outcome?;
if let ServeStop::LiminalExited { detail } = stop {
return Err(HostError::LiminalExited { detail });
}
shutdown_outcome?;
tracing::info!("frame dev server exited cleanly");
Ok(())
}
pub fn run_application(
mut config: FrameConfig,
spec: AppSpec,
page: PageServer,
) -> Result<(), HostError> {
config.finalize_page_origins(page.local_addr());
let app = Application::boot(&config, spec)?;
let serve_outcome = app.serve(&config, page, std::future::pending());
let shutdown_outcome = app.shutdown();
let stop = serve_outcome?;
if let ServeStop::LiminalExited { detail } = stop {
return Err(HostError::LiminalExited { detail });
}
shutdown_outcome?;
tracing::info!("frame server exited cleanly");
Ok(())
}