alex/flybrain
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge fix/flybus-coalescing-flake: the flybus suite asserts guarantees, not the machine's timing

Browse source
This commit is contained in:
acamilo 2026-09-22 18:23:06 +00:00
commit 882980823c
6 changed files with 138 additions and 34 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
docs/design/session-framework/bus-conformance.md
View file

@ -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<TopicMsg>`) | `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` |

11
services/flysim/crates/flybus/examples/demo.rs
View file

@ -99,8 +99,17 @@ pub async fn run() -> Result<Vec<String>, Box<dyn std::error::Error>> {
"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")?;

8
services/flysim/crates/flybus/tests/conformance_artifacts.rs
View file

@ -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

64
services/flysim/crates/flybus/tests/conformance_routing.rs
View file

@ -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`

61
services/flysim/crates/flybus/tests/integration.rs
View file

@ -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::<Vec<_>>(),
"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<u64> = 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();

18
services/flysim/crates/flybus/tests/sol_review_races.rs
View file

@ -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);