OpenFUT/openfut-utas-host/src/async_bridge.rs

//! 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<Self> {
        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<F>(&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::<i64>());
    }
}