1use crate::application_shutdown;
3use crate::async_runtime::monotonic_now;
4use crate::properties::PropertyGraphGuard;
5use crate::{application_shutdown::Cleanup, PropertyGraph, ShutdownTicket, TargetInfo};
6use crate::{async_runtime::AsyncDispatcher, AsyncError, AsyncLimits, AsyncScope};
7use std::time::Duration;
8use std::{
9 any::Any,
10 cell::{Cell, RefCell},
11 rc::{Rc, Weak},
12};
13use std::{
14 future::Future,
15 pin::Pin,
16 task::{Context, Poll, Waker},
17};
18
19pub type AppResult<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
21
22pub const HOST_MESSAGE_LIMIT: usize = 64 * 1024;
24type HostHandler = Rc<dyn Fn(Vec<u8>) -> AppResult<Vec<u8>>>;
25type ReplyCallback = Box<dyn FnOnce(Result<Vec<u8>, String>)>;
26
27struct HostReply(RefCell<Option<ReplyCallback>>);
30impl HostReply {
31 fn finish(&self, result: Result<Vec<u8>, String>) {
32 let callback = self.0.borrow_mut().take();
33 if let Some(callback) = callback {
34 callback(result);
35 }
36 }
37}
38impl Drop for HostReply {
39 fn drop(&mut self) {
40 self.finish(Err("application closed before replying".into()));
41 }
42}
43
44#[derive(Clone)]
47pub struct AppOptions {
48 pub async_limits: AsyncLimits,
49 pub startup_data: Vec<u8>,
52 pub shutdown_deadline: Duration,
54}
55impl Default for AppOptions {
56 fn default() -> Self {
57 Self {
58 async_limits: AsyncLimits::default(),
59 startup_data: Vec::new(),
60 shutdown_deadline: Duration::from_secs(2),
61 }
62 }
63}
64
65pub struct InitContext {
69 pub target: TargetInfo,
70 lifetime: AppContext,
71 startup_data: Vec<u8>,
72}
73impl InitContext {
74 pub fn lifetime(&self) -> &AppContext {
76 &self.lifetime
77 }
78 pub fn startup_data(&self) -> &[u8] {
81 &self.startup_data
82 }
83 pub fn on_host_request(
89 &self,
90 handler: impl Fn(Vec<u8>) -> AppResult<Vec<u8>> + 'static,
91 ) -> AppResult {
92 let inner = self
93 .lifetime
94 .inner
95 .upgrade()
96 .ok_or("application is closed")?;
97 if inner.phase.get() != AppPhase::Preparing {
98 return Err("register host requests during initialization".into());
99 }
100 let mut slot = inner.host_handler.borrow_mut();
101 if slot.is_some() {
102 return Err("host request handler already installed".into());
103 }
104 *slot = Some(Rc::new(handler));
105 Ok(())
106 }
107}
108
109#[derive(Clone, Copy, Debug, PartialEq, Eq)]
111pub enum AppPhase {
112 Preparing,
113 Active,
114 Closing,
115 Closed,
116}
117
118#[derive(Clone, Copy, Debug, PartialEq, Eq)]
120pub enum StopReason {
121 HostClosed,
122 Replaced,
123 PreparationDiscarded,
124}
125
126pub(crate) struct AppInner {
127 pub(crate) phase: Cell<AppPhase>,
128 target: TargetInfo,
129 pub(crate) graph: PropertyGraph,
130 dispatcher: AsyncDispatcher,
131 scope: AsyncScope,
132 initialized: Cell<bool>,
133 startup_data: Vec<u8>,
134 host_handler: RefCell<Option<HostHandler>>,
135 retained: RefCell<Vec<Rc<dyn Any>>>,
136 cleanup: RefCell<Vec<Rc<dyn Cleanup>>>,
137 pub(crate) shutdown_ticket: RefCell<Option<ShutdownTicket>>,
138 shutdown_deadline: Duration,
139 shutdown_started: Cell<Option<Duration>>,
140 deadline_reported: Cell<bool>,
141 shutdown_errors: RefCell<Vec<String>>,
142 close_waiters: RefCell<Vec<Weak<RefCell<Option<Waker>>>>>,
143}
144
145#[derive(Clone)]
148pub struct AppContext {
149 inner: Weak<AppInner>,
150}
151impl AppContext {
152 #[doc(hidden)]
157 pub fn request(&self, bytes: Vec<u8>, reply: impl FnOnce(Result<Vec<u8>, String>) + 'static) {
158 let reply = Rc::new(HostReply(RefCell::new(Some(Box::new(reply)))));
159 let enqueue = || -> Result<(), String> {
160 if bytes.len() > HOST_MESSAGE_LIMIT {
161 return Err("host request exceeds 64 KiB".into());
162 }
163 let inner = self.inner.upgrade().ok_or("application is closed")?;
164 let handler = inner
165 .host_handler
166 .borrow()
167 .clone()
168 .ok_or("application has no host request handler")?;
169 let scope = self.async_scope().map_err(|e| e.to_string())?;
170 let delivered_reply = reply.clone();
171 let completion = scope
172 .completion(move |bytes| {
173 let result =
174 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| handler(bytes)))
175 .map_err(|panic| crate::async_runtime::panic_message(panic))
176 .and_then(|result| result.map_err(|e| e.to_string()))
177 .and_then(|bytes| {
178 if bytes.len() <= HOST_MESSAGE_LIMIT {
179 Ok(bytes)
180 } else {
181 Err("host reply exceeds 64 KiB".into())
182 }
183 })
184 .map_err(|error| {
185 if error.len() <= HOST_MESSAGE_LIMIT {
186 error
187 } else {
188 "host error reply exceeds 64 KiB".into()
189 }
190 });
191 delivered_reply.finish(result);
192 })
193 .map_err(|e| e.to_string())?;
194 completion
195 .complete(bytes)
196 .map_err(|_| "application is closed".to_string())
197 };
198 if let Err(error) = enqueue() {
199 reply.finish(Err(error));
200 }
201 }
202 pub fn async_scope(&self) -> Result<AsyncScope, AsyncError> {
204 let inner = self.inner.upgrade().ok_or(AsyncError::Closed)?;
205 if matches!(inner.phase.get(), AppPhase::Closing | AppPhase::Closed) {
206 return Err(AsyncError::Closed);
207 }
208 Ok(inner.scope.clone())
209 }
210 pub fn set_task_error_sink(&self, sink: impl Fn(String) + 'static) -> Result<(), AsyncError> {
214 let inner = self.inner.upgrade().ok_or(AsyncError::Closed)?;
215 if matches!(inner.phase.get(), AppPhase::Closing | AppPhase::Closed) {
216 return Err(AsyncError::Closed);
217 }
218 inner.dispatcher.set_error_sink(sink);
219 Ok(())
220 }
221 pub fn id(&self) -> Option<u64> {
223 self.inner.upgrade().map(|inner| inner.graph.id())
224 }
225 #[doc(hidden)]
228 pub fn retain_cleanup(&self, cleanup: Rc<dyn Cleanup>) -> Result<(), AsyncError> {
229 if let Some(inner) = self.inner.upgrade() {
230 if !matches!(inner.phase.get(), AppPhase::Closing | AppPhase::Closed) {
231 inner.cleanup.borrow_mut().push(cleanup);
232 return Ok(());
233 }
234 }
235 let orphan = ApplicationInstance::prepare(TargetInfo::default(), AppOptions::default())
239 .expect("default lifecycle limits are valid");
240 orphan.inner.cleanup.borrow_mut().push(cleanup);
241 orphan.finish_close(StopReason::PreparationDiscarded);
242 Err(AsyncError::Closed)
243 }
244 pub fn phase(&self) -> AppPhase {
246 self.inner
247 .upgrade()
248 .map_or(AppPhase::Closed, |inner| inner.phase.get())
249 }
250 pub fn closed(&self) -> AppClosed {
253 let wake = Rc::new(RefCell::new(None));
254 if let Some(inner) = self.inner.upgrade() {
255 let mut waiters = inner.close_waiters.borrow_mut();
256 waiters.retain(|waiter| waiter.strong_count() != 0);
257 waiters.push(Rc::downgrade(&wake));
258 }
259 AppClosed {
260 context: self.clone(),
261 wake,
262 }
263 }
264 pub fn target(&self) -> Result<TargetInfo, AsyncError> {
266 self.inner
267 .upgrade()
268 .map(|inner| inner.target)
269 .ok_or(AsyncError::Closed)
270 }
271}
272
273pub struct AppClosed {
276 context: AppContext,
277 wake: Rc<RefCell<Option<Waker>>>,
278}
279impl Future for AppClosed {
280 type Output = ();
281 fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<()> {
282 let Some(inner) = self.context.inner.upgrade() else {
283 return Poll::Ready(());
284 };
285 if inner.phase.get() == AppPhase::Closed {
286 return Poll::Ready(());
287 }
288 *self.wake.borrow_mut() = Some(cx.waker().clone());
289 Poll::Pending
290 }
291}
292
293thread_local! {
294 static PREPARING_APP: RefCell<Option<Weak<AppInner>>> = const { RefCell::new(None) };
295}
296
297#[doc(hidden)]
300#[derive(Clone)]
301pub struct ApplicationInstance {
302 pub(crate) inner: Rc<AppInner>,
303}
304impl Drop for ApplicationInstance {
305 fn drop(&mut self) {
306 if Rc::strong_count(&self.inner) == 1 && self.inner.phase.get() != AppPhase::Closed {
307 if std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
308 let reason = if self.inner.phase.get() == AppPhase::Preparing {
309 StopReason::PreparationDiscarded
310 } else {
311 StopReason::HostClosed
312 };
313 self.begin_close();
314 self.finish_close(reason);
315 }))
316 .is_err()
317 {
318 log::error!("discarded application teardown panicked");
319 }
320 }
321 }
322}
323impl ApplicationInstance {
324 pub fn prepare(target: TargetInfo, options: AppOptions) -> AppResult<Self> {
325 if options.startup_data.len() > HOST_MESSAGE_LIMIT {
326 return Err("startup data exceeds 64 KiB".into());
327 }
328 if options.async_limits.callbacks_per_turn == 0
329 || options.async_limits.callback_time.is_zero()
330 {
331 return Err("async delivery budgets must be positive".into());
332 }
333 let graph = PropertyGraph::new(|| {});
334 let dispatcher = AsyncDispatcher::new(options.async_limits);
335 let scope = dispatcher.scope()?;
336 Ok(Self {
337 inner: Rc::new(AppInner {
338 phase: Cell::new(AppPhase::Preparing),
339 target,
340 graph,
341 dispatcher,
342 scope,
343 initialized: Cell::new(false),
344 startup_data: options.startup_data,
345 host_handler: RefCell::new(None),
346 retained: RefCell::new(Vec::new()),
347 cleanup: RefCell::new(Vec::new()),
348 shutdown_ticket: RefCell::new(None),
349 shutdown_deadline: options.shutdown_deadline,
350 shutdown_started: Cell::new(None),
351 deadline_reported: Cell::new(false),
352 shutdown_errors: RefCell::new(Vec::new()),
353 close_waiters: RefCell::new(Vec::new()),
354 }),
355 })
356 }
357 pub fn initialize<R>(&self, init: impl FnOnce(&InitContext) -> AppResult<R>) -> AppResult<R> {
360 if self.inner.phase.get() != AppPhase::Preparing || self.inner.initialized.replace(true) {
361 return Err("application initializer can only run once before mounting".into());
362 }
363 let _entered = self.enter();
364 let cx = InitContext {
365 target: self.inner.target,
366 lifetime: self.context(),
367 startup_data: self.inner.startup_data.clone(),
368 };
369 match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| init(&cx))) {
370 Ok(Ok(root)) => Ok(root),
371 Ok(Err(error)) => {
372 self.finish_close(StopReason::PreparationDiscarded);
373 Err(error)
374 }
375 Err(panic) => {
376 self.finish_close(StopReason::PreparationDiscarded);
377 std::panic::resume_unwind(panic)
378 }
379 }
380 }
381 pub fn retain_root<T: 'static>(&self, root: Rc<T>) {
384 assert_eq!(self.inner.phase.get(), AppPhase::Preparing);
385 self.inner.retained.borrow_mut().push(root);
386 }
387 pub fn for_runtime(target: TargetInfo) -> Self {
388 PREPARING_APP
389 .with(|app| app.borrow().as_ref().and_then(Weak::upgrade))
390 .map(|inner| Self { inner })
391 .unwrap_or_else(|| {
392 Self::prepare(target, AppOptions::default())
393 .expect("empty configuration cannot fail")
394 })
395 }
396 pub fn context(&self) -> AppContext {
397 AppContext {
398 inner: Rc::downgrade(&self.inner),
399 }
400 }
401 pub fn dispatcher(&self) -> AsyncDispatcher {
402 self.inner.dispatcher.clone()
403 }
404 pub fn has_pending_work(&self) -> bool {
405 self.inner.graph.has_pending() || self.inner.dispatcher.has_pending()
406 }
407 pub fn set_waker(&self, wake: impl Fn() + Send + Sync + 'static) {
408 let wake = std::sync::Arc::new(wake);
409 let delivery_wake = wake.clone();
410 self.inner.graph.set_waker(move || wake());
411 self.inner.dispatcher.set_waker(move || delivery_wake());
412 }
413 pub fn property_graph(&self) -> PropertyGraph {
414 self.inner.graph.clone()
415 }
416 pub fn enter(&self) -> ApplicationGuard {
417 let previous = PREPARING_APP.with(|app| app.replace(Some(Rc::downgrade(&self.inner))));
418 ApplicationGuard {
419 previous,
420 _graph: self.inner.graph.enter(),
421 }
422 }
423 pub fn activate(&self) {
424 if self.inner.phase.get() != AppPhase::Preparing {
425 return;
426 }
427 self.inner.phase.set(AppPhase::Active);
428 let _entered = self.enter();
429 self.inner.dispatcher.activate();
430 }
431 pub fn begin_close(&self) -> bool {
432 match self.inner.phase.get() {
433 AppPhase::Closing | AppPhase::Closed => false,
434 _ => {
435 self.inner.phase.set(AppPhase::Closing);
436 self.inner.dispatcher.close();
437 self.inner.host_handler.borrow_mut().take();
438 let wake = application_shutdown::wake_route();
441 self.inner.dispatcher.set_waker(move || wake());
442 self.inner.graph.stop_publications();
443 true
444 }
445 }
446 }
447 pub fn finish_close(&self, _reason: StopReason) {
448 if self.inner.phase.get() == AppPhase::Closed || self.inner.shutdown_started.get().is_some()
449 {
450 return;
451 }
452 self.begin_close();
453 self.inner.graph.close();
454 self.inner.shutdown_started.set(Some(monotonic_now()));
455 application_shutdown::retire(self.inner.clone());
456 self.advance_shutdown();
457 if self.inner.phase.get() != AppPhase::Closed {
458 application_shutdown::deadline_wake(self.inner.shutdown_deadline);
459 }
460 application_shutdown::pump_shutdowns();
461 }
462 pub(crate) fn advance_shutdown(&self) {
463 loop {
464 if let Err(error) = self.inner.dispatcher.drain(|| {}) {
466 log::error!("task error sink panicked during shutdown: {error}");
467 }
468 let result = self
469 .inner
470 .shutdown_ticket
471 .borrow()
472 .as_ref()
473 .map(ShutdownTicket::result);
474 match result {
475 Some(None) => break,
476 Some(Some(result)) => {
477 self.inner.shutdown_ticket.borrow_mut().take();
478 if let Err(error) = result {
479 log::error!("application cleanup failed: {error}");
480 self.inner.shutdown_errors.borrow_mut().push(error);
481 }
482 self.inner.cleanup.borrow_mut().pop();
483 }
484 None => {}
485 }
486 let service = self.inner.cleanup.borrow().last().cloned();
487 let Some(service) = service else {
488 let retained = std::mem::take(&mut *self.inner.retained.borrow_mut());
489 self.inner.phase.set(AppPhase::Closed);
490 drop(retained);
491 let waiters = std::mem::take(&mut *self.inner.close_waiters.borrow_mut());
492 for waiter in waiters {
493 if let Some(slot) = waiter.upgrade() {
494 let wake = slot.borrow_mut().take();
495 if let Some(wake) = wake {
496 wake.wake();
497 }
498 }
499 }
500 break;
501 };
502 let ticket =
503 std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| service.begin_shutdown()))
504 .unwrap_or_else(|panic| {
505 let error = crate::async_runtime::panic_message(panic);
506 log::error!(
507 "application cleanup panicked; retaining dependencies: {error}"
508 );
509 self.inner.shutdown_errors.borrow_mut().push(error);
510 ShutdownTicket::unacknowledged()
511 });
512 ticket.set_waker(application_shutdown::wake_route());
513 *self.inner.shutdown_ticket.borrow_mut() = Some(ticket);
514 }
515 if self.inner.phase.get() != AppPhase::Closed
516 && !self.inner.deadline_reported.get()
517 && self.inner.shutdown_started.get().is_some_and(|started| {
518 monotonic_now().saturating_sub(started) >= self.inner.shutdown_deadline
519 })
520 {
521 self.inner.deadline_reported.set(true);
522 log::error!("application shutdown deadline expired; resources and dependencies remain retained until acknowledgement");
523 }
524 }
525}
526
527#[doc(hidden)]
529pub struct ApplicationGuard {
530 previous: Option<Weak<AppInner>>,
531 _graph: PropertyGraphGuard,
532}
533impl Drop for ApplicationGuard {
534 fn drop(&mut self) {
535 PREPARING_APP.with(|app| {
536 app.replace(self.previous.take());
537 });
538 }
539}
540
541#[cfg(test)]
542mod tests {
543 use super::*;
544 #[test]
545 fn host_requests_are_staged_bounded_and_cancelled_with_the_application() {
546 let app = ApplicationInstance::prepare(
547 TargetInfo::default(),
548 AppOptions {
549 startup_data: b"configured by caller".to_vec(),
550 ..Default::default()
551 },
552 )
553 .unwrap();
554 let calls = Rc::new(Cell::new(0));
555 app.initialize(|cx| {
556 assert_eq!(cx.startup_data(), b"configured by caller");
557 let calls = calls.clone();
558 cx.on_host_request(move |bytes| {
559 calls.set(calls.get() + 1);
560 Ok(bytes)
561 })?;
562 assert!(cx.on_host_request(|bytes| Ok(bytes)).is_err());
563 Ok(())
564 })
565 .unwrap();
566 let replies = Rc::new(RefCell::new(Vec::new()));
567 let observed = replies.clone();
568 app.context().request(b"one".to_vec(), move |result| {
569 observed.borrow_mut().push(result)
570 });
571 app.dispatcher().drain(|| {}).unwrap();
572 assert_eq!(calls.get(), 0);
573 app.activate();
574 app.dispatcher().drain(|| {}).unwrap();
575 assert_eq!(&*replies.borrow(), &[Ok(b"one".to_vec())]);
576 let observed = replies.clone();
577 app.context()
578 .request(vec![0; HOST_MESSAGE_LIMIT + 1], move |result| {
579 observed.borrow_mut().push(result)
580 });
581 assert!(replies.borrow()[1].as_ref().unwrap_err().contains("64 KiB"));
582 let observed = replies.clone();
583 app.context().request(b"cancel".to_vec(), move |result| {
584 observed.borrow_mut().push(result)
585 });
586 app.finish_close(StopReason::HostClosed);
587 assert_eq!(calls.get(), 1);
588 assert_eq!(replies.borrow().len(), 3);
589 assert!(replies.borrow()[2].is_err());
590 }
591 #[test]
592 fn host_request_failure_is_observable_without_unwinding_into_host() {
593 let app =
594 ApplicationInstance::prepare(TargetInfo::default(), AppOptions::default()).unwrap();
595 app.initialize(|cx| cx.on_host_request(|_| panic!("request failed")))
596 .unwrap();
597 app.activate();
598 let reply = Rc::new(RefCell::new(None));
599 let received = reply.clone();
600 app.context().request(Vec::new(), move |result| {
601 *received.borrow_mut() = Some(result)
602 });
603 app.dispatcher().drain(|| {}).unwrap();
604 assert!(reply
605 .borrow()
606 .as_ref()
607 .unwrap()
608 .as_ref()
609 .unwrap_err()
610 .contains("request failed"));
611 }
612 #[test]
613 fn host_error_replies_obey_the_boundary_size_limit() {
614 let app =
615 ApplicationInstance::prepare(TargetInfo::default(), AppOptions::default()).unwrap();
616 app.initialize(|cx| cx.on_host_request(|_| Err("x".repeat(HOST_MESSAGE_LIMIT + 1).into())))
617 .unwrap();
618 app.activate();
619 let reply = Rc::new(RefCell::new(None));
620 let received = reply.clone();
621 app.context().request(Vec::new(), move |result| {
622 *received.borrow_mut() = Some(result)
623 });
624 app.dispatcher().drain(|| {}).unwrap();
625 let reply = reply.borrow();
626 let error = reply.as_ref().unwrap().as_ref().unwrap_err();
627 assert_eq!(error, "host error reply exceeds 64 KiB");
628 }
629 #[test]
630 fn initializer_runs_once_in_its_graph_and_preserves_property_identity() {
631 let app =
632 ApplicationInstance::prepare(TargetInfo::default(), AppOptions::default()).unwrap();
633 let context = app.context();
634 let calls = Cell::new(0);
635 let root = app
636 .initialize(|cx| {
637 calls.set(calls.get() + 1);
638 assert_eq!(cx.lifetime().id(), context.id());
639 assert_eq!(
640 ApplicationInstance::for_runtime(cx.target).context().id(),
641 context.id()
642 );
643 Ok(crate::Property::new(42usize))
644 })
645 .unwrap();
646 let shared = root.clone();
647 shared.set(43);
648 assert_eq!(root.get(), 43);
649 assert!(app
650 .initialize(|_| {
651 calls.set(99);
652 Ok(())
653 })
654 .is_err());
655 assert_eq!(calls.get(), 1);
656 app.activate();
657 app.activate();
658 assert_eq!(context.phase(), AppPhase::Active);
659 app.finish_close(StopReason::HostClosed);
660 assert_eq!(context.phase(), AppPhase::Closed);
661 }
662 #[test]
663 fn context_does_not_own_application_and_dropped_preparation_closes() {
664 let app =
665 ApplicationInstance::prepare(TargetInfo::default(), AppOptions::default()).unwrap();
666 let context = app.context();
667 drop(app);
668 assert_eq!(context.phase(), AppPhase::Closed);
669 assert!(context.async_scope().is_err());
670 }
671}
672
673impl crate::async_runtime::AsyncScopeSource for AppContext {
674 fn work_scope(&self) -> Result<AsyncScope, AsyncError> {
675 self.async_scope()
676 }
677}
678
679mod handler_return {
680 pub trait Sealed {}
681 impl Sealed for () {}
682}
683#[doc(hidden)]
690#[diagnostic::on_unimplemented(
691 message = "Pax event handlers must return () synchronously",
692 note = "Start asynchronous work through an explicitly owned executor or scoped task helper; async fn handlers are not awaited."
693)]
694pub trait SynchronousHandlerReturn: handler_return::Sealed {}
695impl SynchronousHandlerReturn for () {}
696#[doc(hidden)]
697pub fn synchronous_handler<T: SynchronousHandlerReturn>(_: T) {}