From f2dc5d434a2fa8811cef8e407ab511f11f331b3b Mon Sep 17 00:00:00 2001 From: dev Date: Tue, 22 Sep 2026 17:05:25 +0000 Subject: [PATCH 1/2] flybus: assert the latest-mode guarantee, not the machine's timing session_over_one_router asserted that a latest subscriber must drop snapshots. bus-v1 section 7 says a latest subscription replaces an undelivered value; it lets a consumer miss values, it does not oblige it to. Under contention the publisher was slow enough that the renderer kept up, saw all twenty snapshots and the test failed on conforming behaviour: 16 of 40 runs beside four busy loops, and 9 of 20 whole-crate runs. The renderer is now held until the publisher's completion is observed rather than until a timer expires, so the coalescing is forced instead of raced for: the subscription keeps the one delivery in flight and one replaceable queued value, and the renderer receives snapshots 1 and 20 of 20. The assertions are the guarantees that hold -- what arrives is in publication order, the last value is the latest published, the stalled spectator never refuses a publication or drops out of the fan-out, and every snapshot the renderer missed is counted as a replacement to the publisher at admission and to the renderer on delivery, so nothing is lost silently. Three more tests in the crate asserted the same kind of race: - latest_replay_is_ordered_ahead_of_a_racing_publish demanded the non-coalesced outcome of a race section 7 allows either way ("bounded mode preserves that order, while latest mode may coalesce it"). It now puts one racing publication to a bounded and a latest subscription at once: bounded must deliver the replay and then the publication, and the latest branch is chosen by that publication's own replaced count. - collection_waits_for_every_retained_owner, and the two disconnect cleanup tests beside it, read the store directory for the unlink that follows the registry update outside the router lock. They use settle_files, like every other unlink check in the suite. - The demo example printed a root count taken before the producer's own release had reached the router, so the line the guide quotes was a race. It waits for the release, the same way it already waits for collection; the printed output is unchanged. Nothing under flybus/src is touched: no routing defect was found. The conformance rows for the credit/queue split, the replaced count, the replay ordering and the latest spectator that cannot refuse a publication now cite what each rewritten test actually varies. --- .../session-framework/bus-conformance.md | 10 +-- .../flysim/crates/flybus/examples/demo.rs | 11 +++- .../flybus/tests/conformance_artifacts.rs | 8 ++- .../flybus/tests/conformance_routing.rs | 64 ++++++++++++++++--- .../flysim/crates/flybus/tests/integration.rs | 61 ++++++++++++++---- 5 files changed, 125 insertions(+), 29 deletions(-) diff --git a/docs/design/session-framework/bus-conformance.md b/docs/design/session-framework/bus-conformance.md index cd90df6..a987554 100644 --- a/docs/design/session-framework/bus-conformance.md +++ b/docs/design/session-framework/bus-conformance.md @@ -153,15 +153,15 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means | `latest`: one queued value, replacing only an undelivered one; replacement releases that entry's roots; delivered or in-use messages are never reclaimed early; maxQueued is exactly 1 | conforms | `router/state.rs::{op_publish (latest branch), op_subscribe}` | `tests/pubsub.rs::latest_coalesces_only_undelivered_values` (both), `tests/conformance_artifacts.rs::latest_mode_holds_at_most_two_roots_delivered_plus_queued` (both) | | `bounded`: FIFO, no coalescing or silent loss; when capacity is unavailable, reject with `BACKPRESSURE` before admitting any delivery | conforms | `router/state.rs::op_publish` (pre-checks every subscriber) | `tests/pubsub.rs::bounded_fifo_and_atomic_backpressure` (both), `tests/conformance_routing.rs::bounded_overflow_rolls_back_all_artifact_roots` (both) | | maxInFlight credits return only on `delivery.consumed`, not on socket write completion | conforms | `router/state.rs::release_owner` (credit returned when the owner is released) | `tests/pubsub.rs::credits_return_only_on_consume` (both), `tests/conformance_routing.rs::bounded_credit_waits_for_every_extracted_artifact` (both) | -| A latest subscriber with all credits in use still has one replaceable queued value | conforms | `router/state.rs::{Sub::queue, dispatch_topic}` (queue and credits are separate) | `tests/bus_acceptance.rs::both_transports_produce_equivalent_behaviour_traces` (events 21 and 24: `publish replaced=1`, then `latest seq=3 replaced=1`) | +| A latest subscriber with all credits in use still has one replaceable queued value | conforms | `router/state.rs::{Sub::queue, dispatch_topic}` (queue and credits are separate) | `tests/bus_acceptance.rs::both_transports_produce_equivalent_behaviour_traces` (events 21 and 24: `publish replaced=1`, then `latest seq=3 replaced=1`), `tests/integration.rs::session_over_one_router` (both: a renderer held for the whole run keeps one delivery in flight and one replaceable value, and receives snapshots 1 and 20 of 20) | | Atomic subscriber/retention snapshot at admission; validate and reserve every queue entry and owner budget before accepting | conforms: one mutex, validate-then-mutate | `router/state.rs::op_publish` | `tests/artifacts.rs::failed_admission_is_atomic` (both) | | A bounded overflow rejects the whole publish: no partial fan-out, no retained-latest update | conforms | `router/state.rs::op_publish` | `tests/conformance_routing.rs::bounded_overflow_rolls_back_all_artifact_roots` (both) | | On acceptance, one `topicSequence` and roots for every delivery and the optional retained value | conforms; a refused publication spends no sequence number | `router/state.rs::op_publish` (`t.sequence += 1` after the checks) | `tests/pubsub.rs::bounded_fifo_and_atomic_backpressure` (both) | | Different topics have no total ordering; multiple publishers follow router acceptance order | conforms: per-topic sequence only | `router/state.rs::Topic::sequence` | `tests/pubsub.rs::retained_replay_clear_delete_and_incarnations` (both) | | The publication reply counts accepted subscriptions and replaced queue entries, not consumers that processed data | conforms | `router/state.rs::op_publish` reply | `tests/pubsub.rs::latest_coalesces_only_undelivered_values` (both) | -| `replaced` on a delivery reports how many undelivered messages were coalesced since that subscription's preceding delivery | conforms | `router/state.rs::{Sub::replaced, dispatch_topic}` (taken at dispatch) | `tests/pubsub.rs::latest_coalesces_only_undelivered_values` (both) | +| `replaced` on a delivery reports how many undelivered messages were coalesced since that subscription's preceding delivery | conforms | `router/state.rs::{Sub::replaced, dispatch_topic}` (taken at dispatch) | `tests/pubsub.rs::latest_coalesces_only_undelivered_values` (both), `tests/integration.rs::session_over_one_router` (both: the 18 replacements the publisher was told about at admission are the same 18 the renderer is told about on delivery, and are exactly the snapshots it did not receive) | | Optional `retained:latest` holds one last message and its artifacts independent of subscribers | conforms | `router/state.rs::op_publish` (retain branch) | `tests/conformance_artifacts.rs::retained_topic_value_holds_a_root_independent_of_subscribers` (both) | -| `replayLatest` enqueues the retained value before subsequent accepted publications; bounded preserves the order, latest may coalesce it | conforms | `router/state.rs::op_subscribe` (replay is enqueued under the subscribe lock) | `tests/conformance_routing.rs::latest_replay_is_ordered_ahead_of_a_racing_publish` (both), `tests/pubsub.rs::retained_replay_clear_delete_and_incarnations` (both) | +| `replayLatest` enqueues the retained value before subsequent accepted publications; bounded preserves the order, latest may coalesce it | conforms | `router/state.rs::op_subscribe` (replay is enqueued under the subscribe lock) | `tests/conformance_routing.rs::latest_replay_is_ordered_ahead_of_a_racing_publish` (both: one racing publication to a bounded and a latest subscription at once; bounded must deliver replay then publication, and the latest branch is chosen by that publication's own `replaced` count, never by which side of the race the dispatcher won), `tests/pubsub.rs::retained_replay_clear_delete_and_incarnations` (both) | | Replay uses the original topicSequence, a fresh deliveryId and explicit roots | conforms | `router/state.rs::op_subscribe` (`add_roots`, the same `Arc`) | `tests/pubsub.rs::retained_replay_clear_delete_and_incarnations` (both) | | Without retention, a zero-subscriber publication retains no ownership after admission | conforms | `router/state.rs::op_publish` | `tests/pubsub.rs::zero_subscriber_publish_retains_nothing` (both) | | Clearing a topic releases only its retained root, not active consumers | conforms | `router/state.rs::op_clear` | `tests/conformance_routing.rs::cleared_topic_gives_no_replay_until_a_fresh_publish` (both) | @@ -260,7 +260,7 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means | The thirteen transport error codes exist with those names | conforms | `error.rs::ErrorCode` | `tests/wire.rs::body_errors_keep_the_connection` (both) | | Three more codes: `CONFLICT`, `NO_TOPIC`, `ARTIFACT_MISMATCH` | deviates-allowed: "Transport errors **include** ..." is not an exhaustive list, and each names a refusal the draft requires but leaves unnamed. Recorded as an amendment in bus-v1 section 12 | `error.rs::ErrorCode` | `tests/pubsub.rs::subscription_and_topic_validation` (both), `tests/artifacts.rs::seal_checks_length_and_digest` (both) | | Before admission report `not-dispatched`; once dispatch might have occurred report `dispatched` or `unknown` conservatively | conforms | `error.rs::BusError::new` (not-dispatched by default), `router/state.rs` dispatched notices, `client/reactor.rs::fail_all` (unknown) | `tests/rpc.rs::{cancellation_states, service_disconnect_fails_calls}` (both), `tests/sol_review_races.rs::writer_failure_terminates_reader_and_pending_work` | -| Bounded subscriptions can reject a publication; latest spectators cannot hold a session transaction indefinitely | conforms: a latest subscriber never causes `BACKPRESSURE` | `router/state.rs::op_publish` (the latest branch skips every capacity check) | `tests/bus_acceptance.rs::a_latest_subscriber_never_refuses_a_publication` (both: 100 publications of 60 KB into one unconsumed slot, six times the bounded pool, none refused, 98 coalesced), with `tests/pubsub.rs::{bounded_fifo_and_atomic_backpressure, saturated_subscriber_does_not_block_control}` (both) for the bounded half | +| Bounded subscriptions can reject a publication; latest spectators cannot hold a session transaction indefinitely | conforms: a latest subscriber never causes `BACKPRESSURE` | `router/state.rs::op_publish` (the latest branch skips every capacity check) | `tests/bus_acceptance.rs::a_latest_subscriber_never_refuses_a_publication` (both: 100 publications of 60 KB into one unconsumed slot, six times the bounded pool, none refused, 98 coalesced), with `tests/pubsub.rs::{bounded_fifo_and_atomic_backpressure, saturated_subscriber_does_not_block_control}` (both) for the bounded half, and `tests/integration.rs::session_over_one_router` (both: 20 snapshot publications accepted by both subscriptions while the presentation consumer reads nothing) | | Sustained pinned-artifact quota exhaustion is surfaced as pressure, not solved by freeing live data | conforms: `QUOTA_EXCEEDED`, never eviction | `router/state.rs::{op_allocate, op_seal}` | `tests/artifacts.rs::quotas_are_enforced` (both) | | Session and application policies choose disconnect, pause or fail; the router does not know which | conforms by absence | `router/state.rs` | `tests/pubsub.rs::saturated_subscriber_does_not_block_control` (both) | @@ -284,7 +284,7 @@ and once over a Unix socket, so `tests/rpc.rs::request_reply_roundtrip` means | 3. Artifacts: allocate/seal/read; publication before seal fails; fan-out owns one object; the last consumer releases; a retained extracted frame survives a message drop | conforms | `tests/artifacts.rs` (28), `tests/conformance_artifacts.rs` (36) | | 4. Faults: sender drops after admission, consumer dies mid-read, reply lost, queued frame replaced, subscription closes with in-use deliveries, router restarts, old release arrives; no double-free, use-after-reuse, unbounded tombstones or hidden replay | conforms | `tests/conformance_routing.rs::{caller_disconnect_detaches_dispatched_call_but_service_keeps_serving, subscriber_disconnect_releases_queued_and_delivered_artifacts_but_not_retention}`, `tests/artifacts.rs::{release_ids_are_watermarked, router_restart_invalidates_old_handles}`, `tests/bus_acceptance.rs::{a_lost_result_leaks_no_roots_and_the_endpoint_cache_still_replays, disconnect_releases_logical_ownership_without_mutating_open_bytes}`, `tests/sol_rereview_regressions.rs` (11) | | 5. RPC cache: an endpoint retains an artifact-bearing result, the original caller consumes it, a domain retry still returns valid bytes, eviction drops the last hold | conforms | `tests/rpc.rs::endpoint_cache_replays_artifact_results` (both), `tests/bus_acceptance.rs::a_retransmission_repeats_the_domain_request_under_a_fresh_call_id` (both) | -| 6. Integration: two parallel fake agents, a complete-batch environment RPC, committed snapshot publication and a deliberately slow presentation consumer on one router | conforms | `tests/integration.rs::session_over_one_router` (both) | +| 6. Integration: two parallel fake agents, a complete-batch environment RPC, committed snapshot publication and a deliberately slow presentation consumer on one router | conforms; the consumer is slow by construction, held until the publisher's completion is observed, so its coalescing is forced rather than raced for | `tests/integration.rs::session_over_one_router` (both) | | 7. Performance: 640x480x60 with three consumers, one delayed; p50/p95/p99 RPC latency, router CPU, copy and readback cost separately, RSS, store live and peak bytes, outstanding roots, collection lag, queue lengths, for one, two and four agents | conforms | `tests/perf.rs::frames_at_60hz_with_three_consumers` (`--ignored`); numbers below | | The first executable example: a counter RPC, a pub/sub observer and a frame artifact held past message consumption, in one small Rust program, no game or browser | conforms | `examples/demo.rs` (`cargo run -p flybus --example demo`), asserted by `tests/example_demo.rs::the_example_shows_a_counter_rpc_an_observer_and_a_held_frame` | diff --git a/services/flysim/crates/flybus/examples/demo.rs b/services/flysim/crates/flybus/examples/demo.rs index 12830de..73f3fb5 100644 --- a/services/flysim/crates/flybus/examples/demo.rs +++ b/services/flysim/crates/flybus/examples/demo.rs @@ -99,8 +99,17 @@ pub async fn run() -> Result, Box> { "published sequence {} to {} subscriber(s)", receipt.topic_sequence, receipt.subscribers )); - // The producer lets go of its own hold; the delivery keeps the bytes alive. + // The producer lets go of its own hold; the delivery keeps the bytes alive. The release + // travels the control lane like any other operation, so the count below waits for it + // instead of reading a number that may still include it. drop(frame); + let released = Instant::now() + Duration::from_secs(10); + while router.stats().artifact_roots > 1 { + if Instant::now() > released { + return Err("the producer's own hold was never released".into()); + } + tokio::time::sleep(Duration::from_millis(1)).await; + } let message = frames.next().await.ok_or("the subscription closed")?; let image = message.artifact("frame")?; diff --git a/services/flysim/crates/flybus/tests/conformance_artifacts.rs b/services/flysim/crates/flybus/tests/conformance_artifacts.rs index cf8a8d2..2eb874b 100644 --- a/services/flysim/crates/flybus/tests/conformance_artifacts.rs +++ b/services/flysim/crates/flybus/tests/conformance_artifacts.rs @@ -179,7 +179,9 @@ async fn collection_waits_for_every_retained_owner(via: Via) { s.owners == 0 && s.sealed_artifacts == 0 && s.store_bytes == 0 }) .await; - assert_eq!(e.files("sealed"), 0); + // The unlink follows the registry update, outside the router lock: wait for the file to + // go rather than assume the two happen together. + e.settle_files("sealed", 0).await; } // ------------------------------------------------------------------------------------------- @@ -521,7 +523,7 @@ async fn disconnect_abandons_an_unsealed_writer(via: Via) { s.artifacts == 0 && s.store_bytes == 0 && s.owners == 0 }) .await; - assert_eq!(e.files("staging"), 0); + e.settle_files("staging", 0).await; } /// An abrupt disconnect must release an explicit hold too, when it was the object's only root. @@ -538,7 +540,7 @@ async fn disconnect_releases_an_explicit_hold(via: Via) { s.sealed_artifacts == 0 && s.owners == 0 && s.store_bytes == 0 }) .await; - assert_eq!(e.files("sealed"), 0); + e.settle_files("sealed", 0).await; } /// A vanished subscriber must give up both a delivery it already holds and one still queued diff --git a/services/flysim/crates/flybus/tests/conformance_routing.rs b/services/flysim/crates/flybus/tests/conformance_routing.rs index 3c0ef30..d8aaa7f 100644 --- a/services/flysim/crates/flybus/tests/conformance_routing.rs +++ b/services/flysim/crates/flybus/tests/conformance_routing.rs @@ -420,14 +420,20 @@ async fn bounded_overflow_rolls_back_all_artifact_roots(via: Via) { } /// bus-v1 section 7: "New subscriptions with replayLatest enqueue it before subsequent accepted -/// publications." A fresh `latest` subscription's replay claims its first in-flight credit -/// immediately (there is nothing else competing for it yet), so a publish accepted right after -/// subscribing must still be observed strictly after the replay, never ahead of or merged with -/// it: each keeps its own delivery. +/// publications ... bounded mode preserves that order, while latest mode may coalesce it before +/// delivery under the ordinary latest rule." The replay is enqueued under the subscribe lock, so +/// a publication admitted after `subscribe` returned is always behind it; what the two modes do +/// with that order is what differs, and one racing publication is put to both at once. +/// +/// The bounded subscription must deliver both values, replay first. The latest subscription +/// either does the same or replaces the still-queued replay, and the router says which in the +/// racing publication's own `replaced` count rather than the test guessing from how fast the +/// dispatcher ran: what it may never do is reorder the two or lose the newer value. async fn latest_replay_is_ordered_ahead_of_a_racing_publish(via: Via) { let e = env(via).await; let admin = e.client("admin").await; let reader = e.client("reader").await; + let viewer = e.client("viewer").await; admin .declare_topic("t.replay-race", Retained::Latest) .await @@ -436,26 +442,66 @@ async fn latest_replay_is_ordered_ahead_of_a_racing_publish(via: Via) { .publish("t.replay-race", obj(json!({"v": "old"})), &[]) .await .unwrap(); - let mut sub = reader + let mut fifo = reader + .subscribe( + "t.replay-race", + SubscriptionConfig::bounded().in_flight(1).replay(true), + ) + .await + .unwrap(); + let mut coalescing = viewer .subscribe( "t.replay-race", SubscriptionConfig::latest().in_flight(1).replay(true), ) .await .unwrap(); - admin + let racing = admin .publish("t.replay-race", obj(json!({"v": "new"})), &[]) .await .unwrap(); - let first = within("the replay arrives first", sub.next()) + assert_eq!( + racing.subscribers, 2, + "one publication, admitted behind both replays" + ); + + let first = within("the replay arrives first", fifo.next()) .await .unwrap(); assert_eq!(first.payload()["v"], "old"); drop(first); // the sole in-flight credit must return before the queued second value moves - let second = within("the racing publish follows, not coalesced away", sub.next()) + let second = within("the racing publish follows, not coalesced away", fifo.next()) .await .unwrap(); - assert_eq!(second.payload()["v"], "new"); + assert_eq!( + (second.payload()["v"].as_str(), second.replaced()), + (Some("new"), 0), + "bounded preserves the order and coalesces nothing" + ); + + let m = within("the latest subscription's first delivery", coalescing.next()) + .await + .unwrap(); + if racing.replaced == 1 { + assert_eq!( + (m.payload()["v"].as_str(), m.replaced()), + (Some("new"), 1), + "a replay still queued is replaced by the newer value, and the delivery says so" + ); + drop(m); + quiet("nothing behind a coalesced replay", coalescing.next()).await; + } else { + assert_eq!( + (racing.replaced, m.payload()["v"].as_str(), m.replaced()), + (0, Some("old"), 0), + "a replay already in flight keeps its own delivery" + ); + drop(m); + let after = within("the racing publish follows it", coalescing.next()) + .await + .unwrap(); + assert_eq!(after.payload()["v"], "new"); + } } /// bus-v1 section 7: clearing releases only the retained root; a later `replayLatest` diff --git a/services/flysim/crates/flybus/tests/integration.rs b/services/flysim/crates/flybus/tests/integration.rs index c4c95ba..f18136b 100644 --- a/services/flysim/crates/flybus/tests/integration.rs +++ b/services/flysim/crates/flybus/tests/integration.rs @@ -1,6 +1,11 @@ //! bus-v1 section 11 item 6: two parallel fake agents, complete-batch environment RPC, //! committed snapshot publication and a deliberately slow presentation consumer, all over one //! router. Generic services only; nothing here knows what a brain or a game is. +//! +//! The presentation consumer is held until the publisher's own completion is observed, so the +//! latest subscription has to coalesce instead of happening to: section 7 lets a latest +//! subscriber miss values, it does not oblige it to, and a test that demands a miss it cannot +//! force is asserting how fast the machine is. mod common; @@ -148,20 +153,32 @@ async fn session_over_one_router(via: Via) { ) .await .unwrap(); + // A renderer that does not read a single snapshot until the publisher has finished every + // one of them. The hold ends on the publisher's observed completion, never on a timer. + let (release, held) = tokio::sync::oneshot::channel::<()>(); let presenting = tokio::spawn(async move { + held.await.unwrap(); let mut seen = Vec::new(); + let mut coalesced = 0; while let Some(m) = slow.next().await { + let (sequence, replaced) = (m.topic_sequence(), m.replaced()); let frame = m.artifact("frame").unwrap(); drop(m); tokio::time::sleep(Duration::from_millis(25)).await; // a slow renderer let bytes = frame.read_all().await.unwrap(); let step = bytes[0] as u64; + assert_eq!( + (step, sequence), + (step, step), + "a delivery carries the frame of the snapshot it announces" + ); + coalesced += replaced; seen.push((step, frame.reference().artifact_id.clone())); if step == STEPS { break; } } - seen + (seen, coalesced) }); let recorder = e.client("recorder").await; let mut all = recorder @@ -179,6 +196,7 @@ async fn session_over_one_router(via: Via) { seq }); + let mut replaced_at_admission = 0; for step in 1..=STEPS { let advanced = coordinator .call_and_wait( @@ -226,8 +244,16 @@ async fn session_over_one_router(via: Via) { ) .await .unwrap(); - assert_eq!(receipt.topic_sequence, step); + assert_eq!( + (receipt.topic_sequence, receipt.subscribers), + (step, 2), + "both subscriptions accept every publication: the stalled latest spectator neither \ + refuses one nor drops out of the fan-out" + ); + replaced_at_admission += receipt.replaced; } + // The publisher is finished, observably, so the renderer may start. + release.send(()).unwrap(); let recorded = within("recorder", recording).await.unwrap(); assert_eq!( @@ -235,17 +261,30 @@ async fn session_over_one_router(via: Via) { (1..=STEPS).map(|s| (s, s)).collect::>(), "the bounded recorder misses nothing" ); - let presented = within("presenter", presenting).await.unwrap(); - assert_eq!( - presented.last().unwrap().0, - STEPS, - "the slow consumer ends on the latest snapshot" - ); + let (presented, coalesced) = within("presenter", presenting).await.unwrap(); + let steps: Vec = presented.iter().map(|(s, _)| *s).collect(); assert!( - presented.len() < STEPS as usize, - "the slow consumer skipped snapshots: {presented:?}" + steps.windows(2).all(|w| w[0] < w[1]), + "what a latest subscription does deliver arrives in publication order: {presented:?}" + ); + assert_eq!( + steps.last().copied(), + Some(STEPS), + "the slow consumer ends on the latest snapshot: {presented:?}" + ); + assert_eq!( + steps, + vec![1, STEPS], + "held for the whole run, the subscription keeps the one delivery already in flight and \ + one replaceable queued value, so the renderer sees the first snapshot and the last, \ + and the eighteen between them were coalesced: {presented:?}" + ); + assert_eq!( + (coalesced, replaced_at_admission), + (STEPS - 2, STEPS - 2), + "every snapshot the renderer missed is counted as a replacement, to the publisher at \ + admission and to the renderer on its next delivery: none is lost silently" ); - assert!(presented.windows(2).all(|w| w[0].0 < w[1].0)); for t in agents { t.abort(); From 9bed780cb96c462a81de0a647247fac260889f04 Mon Sep 17 00:00:00 2001 From: dev Date: Tue, 22 Sep 2026 17:12:46 +0000 Subject: [PATCH 2/2] flybus: wait for the pending-connection events instead of timing them pending_connections_are_bounded_and_hello_expires gave the router 20 ms to register a connection and then slept 80 ms past a 40 ms hello timeout before reading the count once. Both are claims about how fast this box is, not about the router: the first failed once in 20 whole-crate runs beside four busy loops. Both now wait for the event they are about, under the suite's ordinary ten-second bound, so the test still fails if a pending connection never registers or a pending Hello never expires. --- .../crates/flybus/tests/sol_review_races.rs | 18 +++++++++++++----- 1 file changed, 13 insertions(+), 5 deletions(-) diff --git a/services/flysim/crates/flybus/tests/sol_review_races.rs b/services/flysim/crates/flybus/tests/sol_review_races.rs index 9d26b1b..793df64 100644 --- a/services/flysim/crates/flybus/tests/sol_review_races.rs +++ b/services/flysim/crates/flybus/tests/sol_review_races.rs @@ -266,13 +266,14 @@ async fn pending_connections_are_bounded_and_hello_expires() { let router = Router::new(config).unwrap(); let pending = router.connect_in_memory_as("first"); - tokio::time::timeout(Duration::from_millis(20), async { + // Registration happens on the router's own task. How long that takes is this box's + // business; that it happens is the router's. + within("the pending connection occupies the only slot", async { while router.stats().connections != 1 { - tokio::task::yield_now().await; + tokio::time::sleep(Duration::from_millis(1)).await; } }) - .await - .unwrap(); + .await; assert_eq!(router.stats().connections, 1); let refused = Client::connect( router.connect_in_memory_as("second"), @@ -280,7 +281,14 @@ async fn pending_connections_are_bounded_and_hello_expires() { ) .await; assert_eq!(refused.unwrap_err().code, ErrorCode::RouterLost); - tokio::time::sleep(Duration::from_millis(80)).await; + // The 40 ms hello timeout expires on the router's clock: wait for the expiry to be + // observed rather than sleep past it and read the count once. + within("the pending Hello expires", async { + while router.stats().connections != 0 { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await; assert_eq!(router.stats().connections, 0, "pending Hello timed out"); drop(pending);