//! Sync → async execution bridge for the economy handlers. //! //! The UTAS host is a **synchronous, thread-per-connection** server //! ([`crate::Server::serve_listener`] spawns one `std::thread` per socket and //! [`crate::Server::handle_with_ip`] runs to completion on that plain OS //! thread). Most handlers — `/club`, squad, credits, match reward, the Store //! writers — are fully synchronous (Core is reached over a **blocking** reqwest //! client). Only the durable transfer-market + pile handlers are `async`, //! because their store is `sqlx` with the `runtime-tokio` feature. //! //! Rather than sprinkle per-request runtimes, the host owns ONE long-lived //! multi-threaded Tokio runtime for the whole process lifetime and drives the //! few async handlers on it with [`AsyncBridge::block_on`]. //! //! ## Execution contract //! //! * Exactly one runtime exists for the host lifetime; it is created once //! (`Server::from_config`) and shared by every connection thread via `Arc`. //! * `block_on` is called from the synchronous dispatch thread, which is NOT a //! Tokio worker, so the common path is a direct [`tokio::runtime::Runtime::block_on`]. //! * **Nested-runtime safety:** should dispatch ever be invoked from inside a //! Tokio context (it is not in production, but a future async caller or a test //! might), calling `block_on`/`Handle::block_on` on the current thread would //! panic (*"Cannot start a runtime from within a runtime"*). We detect an //! ambient runtime with [`tokio::runtime::Handle::try_current`] and, in that //! case, offload the future onto our own runtime and block the caller on a //! std channel — no runtime is entered on the current thread, so it cannot //! panic. //! * The blocking Core client used *inside* the async handlers is safe here: it //! runs its own private runtime on a dedicated background thread, so calling //! it from a runtime worker parks the worker on a channel rather than entering //! a runtime. The host client is process-lifetime (`Arc`) and never dropped in //! an async context. use std::future::Future; use tokio::runtime::{Builder, Handle, Runtime}; /// A process-lifetime Tokio runtime plus a nested-safe blocking bridge. pub struct AsyncBridge { rt: Runtime, } impl AsyncBridge { /// Build the shared multi-threaded runtime. Multi-threaded (not /// current-thread) so that a handler which parks a worker on the blocking /// Core client still leaves other workers to drive `sqlx` IO to completion. pub fn new() -> std::io::Result { let rt = Builder::new_multi_thread() .worker_threads(4) .enable_all() .thread_name("openfut-econ") .build()?; Ok(Self { rt }) } /// A cloneable handle to the shared runtime (used by callers that want to /// `spawn` onto the same executor rather than block). pub fn handle(&self) -> Handle { self.rt.handle().clone() } /// Drive `fut` to completion from a synchronous caller, returning its output. /// /// The `Send + 'static` bound lets the nested-safety fallback move the future /// onto our runtime; dispatch satisfies it by building `async move` blocks /// that own `Arc` clones of the shared services (never per-request state). pub fn block_on(&self, fut: F) -> F::Output where F: Future + Send + 'static, F::Output: Send + 'static, { match Handle::try_current() { // Common path: a plain OS dispatch thread, no ambient runtime. Err(_) => self.rt.block_on(fut), // Ambient runtime present: driving on the current thread would panic. // Offload to our runtime and block on a std channel — the current // thread never enters a runtime, so it is panic-free. Ok(_) => { let (tx, rx) = std::sync::mpsc::sync_channel(1); self.rt.spawn(async move { let _ = tx.send(fut.await); }); rx.recv().expect("AsyncBridge offloaded task panicked") } } } } #[cfg(test)] mod tests { use super::*; #[test] fn block_on_from_plain_thread_runs_future() { let bridge = AsyncBridge::new().unwrap(); let out = bridge.block_on(async { 2 + 3 }); assert_eq!(out, 5); } #[test] fn block_on_from_plain_thread_drives_async_io() { // A future with an actual await point (timer) completes on our runtime. let bridge = AsyncBridge::new().unwrap(); let out = bridge.block_on(async { tokio::task::yield_now().await; 41 + 1 }); assert_eq!(out, 42); } #[test] fn block_on_is_nested_runtime_safe() { // Simulate the hazardous case: dispatch invoked from WITHIN a Tokio // context. A naive `Runtime::block_on` here panics; the bridge must not. let bridge = AsyncBridge::new().unwrap(); let ambient = Builder::new_multi_thread() .worker_threads(2) .enable_all() .build() .unwrap(); let out = ambient.block_on(async { // We are now inside `ambient`. Calling the bridge must offload, not // panic with "Cannot start a runtime from within a runtime". bridge.block_on(async { tokio::task::yield_now().await; 7 * 6 }) }); assert_eq!(out, 42); } #[test] fn block_on_concurrently_from_many_threads() { // The shared runtime is driven by many dispatch threads at once, exactly // as thread-per-connection dispatch does. let bridge = std::sync::Arc::new(AsyncBridge::new().unwrap()); let mut handles = Vec::new(); for i in 0..16i64 { let b = bridge.clone(); handles.push(std::thread::spawn(move || { b.block_on(async move { tokio::task::yield_now().await; i * 2 }) })); } let sum: i64 = handles.into_iter().map(|h| h.join().unwrap()).sum(); assert_eq!(sum, (0..16i64).map(|i| i * 2).sum::()); } }