agentmux_srv/backend/
blocking.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Running a blocking section from code that may be on an async worker.
5//!
6//! A synchronous call that can take seconds (SQLite with a busy timeout, file
7//! I/O, per-process OS queries) must not run inline on a Tokio worker. It is
8//! not only that one worker that stops: the worker that happens to be driving
9//! the I/O driver stops polling it, and no other worker takes the driver over
10//! until that one parks again. For the length of the call no socket is
11//! serviced — the HTTP health endpoint included — while timers and
12//! already-queued tasks keep running, so the server looks alive in its logs
13//! and answers nothing. That is what the launcher's liveness probe saw before
14//! it recycled srv. See
15//! `docs/analysis/ANALYSIS_SRV_HTTP_STALL_IO_DRIVER_STARVATION_2026_09_26.md`.
16
17/// Run `f`, handing this worker's other tasks (and the I/O driver, if it holds
18/// it) to another thread first when called on a multi-thread runtime worker.
19///
20/// Synchronous and order-preserving, unlike `spawn_blocking`: the caller
21/// continues only after `f` returns, so a write that must land before the
22/// next one still does.
23///
24/// - Multi-thread runtime worker: `block_in_place`.
25/// - Blocking-pool thread, or no runtime: nothing to hand off — runs `f`.
26/// - Current-thread runtime (most `#[tokio::test]`s): `block_in_place` panics
27///   there and there is no other worker to hand off to — runs `f`.
28pub fn off_async_worker<R>(f: impl FnOnce() -> R) -> R {
29    let on_multi_thread = tokio::runtime::Handle::try_current()
30        .is_ok_and(|h| h.runtime_flavor() == tokio::runtime::RuntimeFlavor::MultiThread);
31    if on_multi_thread {
32        tokio::task::block_in_place(f)
33    } else {
34        f()
35    }
36}
37
38#[cfg(test)]
39mod tests {
40    use super::off_async_worker;
41
42    #[test]
43    fn runs_without_a_runtime() {
44        assert_eq!(off_async_worker(|| 7), 7);
45    }
46
47    #[tokio::test(flavor = "current_thread")]
48    async fn does_not_panic_on_a_current_thread_runtime() {
49        // block_in_place would panic here; the helper must fall back.
50        assert_eq!(off_async_worker(|| 7), 7);
51    }
52
53    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
54    async fn runs_on_a_multi_thread_worker() {
55        assert_eq!(off_async_worker(|| 7), 7);
56    }
57
58    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
59    async fn runs_on_a_blocking_pool_thread() {
60        let v = tokio::task::spawn_blocking(|| off_async_worker(|| 7)).await.unwrap();
61        assert_eq!(v, 7);
62    }
63
64    /// The point of the helper: while a worker sits in a long blocking section,
65    /// the runtime still services I/O. One worker makes this deterministic —
66    /// with the section inline, that worker IS the I/O driver and nothing else
67    /// can poll it, so the round-trip below would wait the full 1.5s;
68    /// `block_in_place` hands its core to another thread first.
69    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
70    async fn io_is_still_serviced_while_a_worker_blocks() {
71        use tokio::io::{AsyncReadExt, AsyncWriteExt};
72        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
73        let addr = listener.local_addr().unwrap();
74        tokio::spawn(async move {
75            loop {
76                let (mut s, _) = listener.accept().await.unwrap();
77                tokio::spawn(async move {
78                    let mut b = [0u8; 4];
79                    let _ = s.read(&mut b).await;
80                    let _ = s.write_all(b"pong").await;
81                });
82            }
83        });
84        let (entered_tx, entered_rx) = std::sync::mpsc::channel::<()>();
85        let blocker = tokio::spawn(async move {
86            off_async_worker(move || {
87                entered_tx.send(()).unwrap();
88                std::thread::sleep(std::time::Duration::from_millis(1500));
89            });
90        });
91        // Start the round-trip only once the worker is inside the section.
92        tokio::task::spawn_blocking(move || entered_rx.recv().unwrap()).await.unwrap();
93        let started = std::time::Instant::now();
94        let mut c = tokio::net::TcpStream::connect(addr).await.unwrap();
95        c.write_all(b"ping").await.unwrap();
96        let mut b = [0u8; 4];
97        c.read_exact(&mut b).await.unwrap();
98        assert_eq!(&b, b"pong");
99        assert!(
100            started.elapsed() < std::time::Duration::from_millis(800),
101            "round-trip took {:?} while another worker was blocked",
102            started.elapsed()
103        );
104        blocker.await.unwrap();
105    }
106}
agentmux_srv/backend/blocking.rs

agentmux_srv/backend/
blocking.rs
