fix(global): give the bus its own runtime so it outlives the caller's - #2
fix(global): give the bus its own runtime so it outlives the caller's#2senamakel wants to merge 6 commits into
Conversation
Checkpoint of work in progress, touching crates/tinybus/src/connection.rs. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
The bus's broker and connection loops now run on a single-threaded runtime owned by the singleton, rather than the caller's runtime. This ensures the bus outlives any one runtime, such as under `cargo test` where each test tears down its own, preventing silent failures from a dead reactor. The runtime is built lazily on first use and never dropped, with `init_with` remaining the exception as the caller owns that connection's tasks. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
Add two tests that verify the global OnceBus continues to work after the runtime that initialised it has been dropped, and that the shared runtime is created only once regardless of how many times init is called. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
Dropping a `Runtime` blocks until its tasks finish, which is forbidden in async contexts and would panic for anyone holding an `OnceBus` in a local rather than a `static`. The new `Drop` implementation calls `shutdown_background` to return immediately, making local use safe. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
The test helper was refactored to separate subscription from event collection, ensuring a subscription is established before publishing to avoid missing events. This makes the delivery test more reliable by guaranteeing the subscriber is ready before any publish occurs. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
The bus runtime now lives on its own OS thread using a current-thread tokio runtime instead of a multi-threaded one, because the slim no-default-features build lacks the rt-multi-thread feature. A dedicated thread also avoids panics when dropping the runtime from an async context, and one thread is sufficient for the broker and connection loops. Auto-committed-on: dragonfly Co-authored-by: Medulla <medulla@tinyhumans.ai>
📝 WalkthroughWalkthroughThe change adds a public connection handshake method and moves ChangesOwned runtime and connection initialization
Sequence Diagram(s)sequenceDiagram
participant OnceBus
participant BusRuntime
participant Connection
participant Broker
OnceBus->>BusRuntime: enter owned runtime
BusRuntime->>Connection: attach transport
Connection->>Broker: send Hello
Broker-->>Connection: return unique name
Connection-->>OnceBus: complete initialization
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@crates/tinybus/src/connection.rs`:
- Around line 137-138: Update the repeated-handshake documentation near the
connection handshake method to accurately state that each repeated call sends
another Hello and replaces unique_name with the broker’s returned from_name; do
not describe it as requesting a second unique name.
In `@crates/tinybus/src/global.rs`:
- Around line 399-407: Update the test the_bus_runtime_is_built_once_and_shared
to invoke BUS.runtime() twice after initialization and assert that both calls
return the same runtime reference, rather than only checking BUS.runtime.get().
Keep the existing repeated init calls to establish initialization before
exercising runtime reuse.
- Around line 143-149: Serialize first-time bus initialization across
init_in_process, init_over, and init_with using one shared async initialization
guard. After acquiring the guard, recheck self.bus and return the existing bus
when initialization already completed; otherwise perform runtime, connection,
and broker creation while holding the guard. Add a concurrent initialization
test asserting that exactly one broker and connection are created.
- Around line 351-366: Update wait_for and an_in_process_bus_delivers_end_to_end
to use a channel for event delivery instead of polling the shared Vec<Tick>
state. Receive events asynchronously until n items are collected, wrapping the
receive operation in tokio::time::timeout and preserving the existing
deadline/failure behavior.
🪄 Autofix
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5879d524-4092-401b-a35a-a3ec50402f0b
📒 Files selected for processing (2)
crates/tinybus/src/connection.rscrates/tinybus/src/global.rs
| /// Idempotent in the only sense that matters: calling it twice would | ||
| /// request a second unique name, so don't. |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
Correct the repeated-handshake documentation.
The broker returns the attached peer's from_name for every Hello. A second call does not request a second unique name. State that repeated calls send Hello again and replace unique_name with the broker response, or reject repeated calls explicitly.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/tinybus/src/connection.rs` around lines 137 - 138, Update the
repeated-handshake documentation near the connection handshake method to
accurately state that each repeated call sends another Hello and replaces
unique_name with the broker’s returned from_name; do not describe it as
requesting a second unique name.
| /// The runtime this bus's tasks live on, started on first use. | ||
| fn runtime(&self) -> Result<&BusRuntime> { | ||
| if let Some(existing) = self.runtime.get() { | ||
| return Ok(existing); | ||
| } | ||
| let started = BusRuntime::start()?; | ||
| Ok(self.runtime.get_or_init(|| started)) |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Serialize first-time initialization before spawning tasks.
Two concurrent callers can both observe an unset bus. Each caller can then start a runtime, attach a connection, and, for init_in_process, spawn a broker before init_with selects one bus. The losing in-process broker remains on the owned runtime after its connection is dropped.
Use one async initialization guard for init_in_process, init_over, and init_with. Recheck self.bus after acquiring that guard. Add a concurrent initialization test that verifies only one broker and connection are created.
Also applies to: 160-176, 185-195
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/tinybus/src/global.rs` around lines 143 - 149, Serialize first-time
bus initialization across init_in_process, init_over, and init_with using one
shared async initialization guard. After acquiring the guard, recheck self.bus
and return the existing bus when initialization already completed; otherwise
perform runtime, connection, and broker creation while holding the guard. Add a
concurrent initialization test asserting that exactly one broker and connection
are created.
| /// Wait for `n` events to land, or fail on a deadline. | ||
| async fn wait_for(seen: &Arc<Mutex<Vec<Tick>>>, n: usize) -> Vec<Tick> { | ||
| let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(5); | ||
| loop { | ||
| { | ||
| let guard = seen.lock().await; | ||
| if guard.len() >= n { | ||
| return guard.clone(); | ||
| } | ||
| } | ||
| assert!( | ||
| tokio::time::Instant::now() < deadline, | ||
| "the bus delivered nothing" | ||
| ); | ||
| tokio::task::yield_now().await; | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -eu
printf '%s\n' '--- candidate file map ---'
ast-grep outline crates/tinybus/src/global.rs --lang rust 2>/dev/null || true
printf '%s\n' '--- target implementation ---'
sed -n '300,390p' crates/tinybus/src/global.rs
printf '%s\n' '--- event capture and wait_for references ---'
rg -n -C 4 'wait_for|seen|Tick|mpsc|channel|send\(' crates/tinybus/src/global.rsRepository: tinyhumansai/tinybus
Length of output: 11661
🏁 Script executed:
#!/bin/bash
set -eu
python3 - <<'PY'
from pathlib import Path
import re
path = Path("crates/tinybus/src/global.rs")
text = path.read_text()
match = re.search(
r"async fn wait_for\(seen:.*?\n \}\n",
text,
re.S,
)
if not match:
raise SystemExit("wait_for not found")
body = match.group(0)
print("--- wait_for behavioral markers ---")
for marker in ("seen.lock().await", "tokio::task::yield_now().await", "tokio::time::timeout", "recv().await"):
print(f"{marker}: {marker in body}")
print("--- wait_for call sites ---")
for number, line in enumerate(text.splitlines(), 1):
if "wait_for(" in line:
print(f"{number}: {line.strip()}")
print("--- direct polling loop markers ---")
for number, line in enumerate(text.splitlines(), 1):
if "while seen.lock().await.is_empty()" in line or "tokio::task::yield_now().await" in line:
print(f"{number}: {line.strip()}")
PY
printf '%s\n' '--- surrounding helper callers ---'
sed -n '335,410p' crates/tinybus/src/global.rsRepository: tinyhumansai/tinybus
Length of output: 3729
Await delivery instead of polling shared state.
Replace the shared Vec polling in wait_for with a channel and receive events under tokio::time::timeout. Apply the same change to an_in_process_bus_delivers_end_to_end.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/tinybus/src/global.rs` around lines 351 - 366, Update wait_for and
an_in_process_bus_delivers_end_to_end to use a channel for event delivery
instead of polling the shared Vec<Tick> state. Receive events asynchronously
until n items are collected, wrapping the receive operation in
tokio::time::timeout and preserving the existing deadline/failure behavior.
Source: Coding guidelines
| fn the_bus_runtime_is_built_once_and_shared() { | ||
| static BUS: OnceBus<Tick> = OnceBus::new(); | ||
| let runtime = tokio::runtime::Runtime::new().unwrap(); | ||
| runtime.block_on(async { | ||
| BUS.init_in_process(config()).await.unwrap(); | ||
| BUS.init_in_process(config()).await.unwrap(); | ||
| }); | ||
| // One runtime, however many times `init` is called. | ||
| assert!(BUS.runtime.get().is_some()); |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Make the runtime-reuse test exercise runtime reuse.
The second init_in_process returns at line 160 because bus already exists. It does not call runtime(). The assertion only proves that a runtime exists.
Call BUS.runtime() twice and compare the returned references, or add test-only startup instrumentation.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/tinybus/src/global.rs` around lines 399 - 407, Update the test
the_bus_runtime_is_built_once_and_shared to invoke BUS.runtime() twice after
initialization and assert that both calls return the same runtime reference,
rather than only checking BUS.runtime.get(). Keep the existing repeated init
calls to establish initialization before exercising runtime reuse.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: a126943ee5
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let connection = { | ||
| let _guard = self.runtime()?.handle.enter(); | ||
| Connection::attach(transport.into()) |
There was a problem hiding this comment.
When transport is a UnixTransport created by UnixTransport::connect on the caller's runtime, entering the owned runtime only for Connection::attach moves the tasks but does not re-register the underlying Tokio UnixStream with the new reactor. Once the caller's runtime is dropped, its reactor is shut down and this supposedly persistent connection stops doing I/O. Construct or register the socket on the owned runtime rather than accepting an already runtime-bound transport here.
Useful? React with 👍 / 👎.
| let (_handle, seen) = watch(&BUS); | ||
| BUS.publish(Tick(1)); | ||
| assert_eq!(wait_for(&seen, 1).await, vec![Tick(1)]); | ||
| }); |
There was a problem hiding this comment.
When a subscriber created on the initial runtime is retained or forget() is used for a process-lifetime startup subscription, OnceBus::subscribe still reaches subscriber::spawn, whose plain tokio::spawn binds the handler task to that caller runtime. Dropping the first runtime therefore silently kills the subscription even though the connection survives; this test masks the case by dropping _handle before dropping first and then creating a new subscription. Keep one handle across the boundary and spawn subscription tasks through the owned runtime.
AGENTS.md reference: AGENTS.md:L58-L60
Useful? React with 👍 / 👎.
1 participant
The bug
A process-wide bus outlives any one runtime — but its tasks did not.
OnceBus::init_in_processspawned the broker and this peer's reader/writer loops with plaintokio::spawn, so they belonged to whichever runtime calledinitfirst.That is fine in production, where the bus is initialised on the runtime that then runs for the life of the process. It is a trap under
cargo test, where every#[tokio::test]builds and tears down its own: the first test to win theOnceLockleaves every later test holding a bus attached to a dead reactor. Publishes go nowhere, subscribers never wake, and nothing anywhere reports an error — the failure is silent, which is what makes it worth a fix rather than a note.It is not hypothetical: it showed up in the OpenHuman migration as
init()returning an error for every test after the first, because announcing the peer manifest was hitting a dead connection. That was mitigated downstream by making the announce advisory, which is correct on its own merits but treats the symptom — this is the cause.Scope, stated honestly: I initially thought this also explained
memory::sync_pipeline_e2e_tests::multi_batch_volume_builds_full_treefailing downstream. It does not. That test asserted on delivery after a singleyield_now, which was enough when the bus was a channel with an inline handler and is not enough now that an event crosses two task hops; it is fixed in the OpenHuman PR by waiting for delivery rather than assuming it. This change is worth having regardless, but it is not what fixed that test.The fix
OnceBusowns a runtime on a dedicated thread and enters it around the spawning, so the bus's lifetime matches thestatic's — which is what everything reaching for a global bus already assumes.Current-thread, not multi-thread.
Builder::new_multi_threadneeds tokio'srt-multi-threadfeature, which the slim--no-default-featuresbuild does not have. A bus that could not be a singleton in the slim build would defeat the point of the slim build, so the runtime is a current-thread one driven by its own OS thread. One thread is enough regardless: the broker routes messages and the connection loops shuffle frames, none of it CPU-bound.Shutdown is drop-shaped, not explicit.
BusRuntimeholds aoneshot::Sender; dropping it ends the thread'sblock_on, which drops the runtime on that thread, where blocking is allowed. That is deliberate — an in-placeRuntimedrop panics inside an async context, so anyone holding anOnceBusin a local rather than astaticwould have hit it. (I hit it: it is why this shape exists rather than aDropimpl callingshutdown_background.)Connection::handshakeis split out ofconnect. AnEnterGuardis!Send, so holding one across an await would make the future!Sendfor every caller.attach— where the spawning happens — is synchronous, so the guard is held across it and dropped before the awaitedHello.init_withis deliberately unchanged: the caller supplied that connection and owns its tasks, so it is left where it was built.Testing
a_bus_outlives_the_runtime_that_initialised_itreproduces the exact shape — initialise on one runtime, drop it, use the bus from another — and fails without this change.All gates green under the CI toolchain (1.97.1,
RUSTFLAGS=-D warnings): fmt, clippy,--all-features(129 tests),--no-default-features --features macros(126), the slimcargo check, and docs.