diff --git a/docs/design/session-framework/bus-conformance.md b/docs/design/session-framework/bus-conformance.md index 00e72b3..5c52155 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(); 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);