test: transport × persistence matrix for the distributed read-model flow - #48
Merged
patrickleet merged 16 commits intoMay 30, 2026
Merged
Conversation
First slice of the distributed read-model matrix (async bus facade only, no sync path). Ungate the generic async flow helpers so they are the primary path, and add run_checkout_over_bus<B: Bus + BusConsumer, R>: drive the seat-checkout domain flow + read-model projection + query on persistence R, route the events over transport B, and assert the projected checkout screen. Validated cell: HashMapRepository × InMemoryBus. Refs [[tasks/transport-persistence-matrix]] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The distributed read-model seat-checkout scenario now runs across every async transport × persistence backend, all green against live brokers: transports : InMemoryBus, NatsBus, RabbitBus, KafkaBus, PostgresBus, Knative persistence: HashMapRepository, SqliteRepository, PostgresRepository 12 matrix cells (broker/DB cells skip when their env var is unset): in-memory & sqlite over each of InMemory/NATS/Rabbit/Kafka/Knative, in-memory & postgres-persistence over a Postgres bus / in-memory bus. Knative is a first-class transport cell: KnativeBus POSTs CloudEvents to a local cloud_events_router serving the projection sink (the HTTP/gRPC command ingress is this same Knative surface) — no broker needed. RabbitMQ binds the subscription before publishing (topic exchange drops unrouted events); NATS ensures the stream; Postgres bus ensures its tables. Shared helpers: build_collector (the transport sink), run_checkout_over_bus (pull buses), run_checkout_over_knative (HTTP), project_and_assert_checkout. All on the async bus facade — no sync path. Refs [[tasks/transport-persistence-matrix]] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ndard test onto the async bus
Refactor (not delete) the gold-standard seat_checkout_saga test onto the
async InMemoryBus: same services, choreography, projection, query, and
assertions — the legacy InMemoryQueue/OutboxWorkerThread/Subscribable
wiring is replaced by publish_pending_outbox (claim→publish→complete bridge)
+ bus.subscribe hops. The projection_service/query_service modules are kept.
Complete the matrix to the full 6×3 grid (18 cells), all green against live
brokers: { HashMap, SQLite, Postgres } persistence × { InMemoryBus, NatsBus,
RabbitBus, KafkaBus, PostgresBus, Knative } transport. Postgres-persistence
fixtures + Postgres-bus pairings added; broker/DB cells skip without env.
Full distributed_read_model suite: 23 passed (refactored sync test + 18
matrix cells + 2 async flow tests + HTTP/gRPC command tests).
Refs [[tasks/transport-persistence-matrix]]
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Plus Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
📝 WalkthroughWalkthroughThis test file transitions from a synchronous queue-based outbox worker model to a fully async bus-driven verification flow. Legacy cfg-gates are removed, core event/command functions are ungated for reuse, and the test harness is expanded to include a large transport × persistence matrix covering InMemoryBus, Knative/HTTP, NATS, RabbitMQ, Kafka, and PostgresBus paired with HashMap, SQLite, and Postgres backends. ChangesAsync bus-driven read-model test migration
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
Collaborator
Author
|
@coderabbitai review |
✅ Actions performedReview triggered.
|
First step of Phase 1 (legacy sync bus removal): the pub/sub transport test now publishes events to InMemoryBus and drains them via bus.subscribe, instead of Bus::from_queue(InMemoryQueue) + microsvc::subscribe. Proves the migration pattern; the legacy bus src stays until all consumers are migrated. Refs [[tasks/async-only-consolidation]] Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…arity)
The load -> mutate -> sync -> commit workspace ergonomic existed only over
the sync store traits; the async path used the bare write-plan builder. This
restores parity: the mutation/sync/diff surface is store-independent, so the
same `ReadModelWorkspace` now gains `load_async`/`commit_async` over the
`Async{ReadModelWritePlanStore,RelationalReadModelQueryStore}` traits, plus
`AsyncReadModelLoadBuilder` and `AsyncReadModelWorkspaceExt::workspace_async()`.
No struct extraction or duplicated diff logic: `load`/`commit` move to small
sync- and async-bound impl blocks; everything else stays shared and unbounded.
Proven with async mirrors of the include-hydration and sync-roundtrip tests
on `InMemoryReadModelStore` (impls both async store traits). Sync workspace
API and its tests unchanged.
Part of [[tasks/async-only-consolidation]] (Phase 2).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…on over the async surface Async paths previously bypassed QueuedRepository entirely (AsyncCommitBuilder commits straight through commit_batch_async), so two concurrent async commits to the same aggregate could interleave. This restores the queueing ability for async: `repo.queued_async().async_aggregate::<T>()` serializes per-aggregate get/commit exactly like the sync `.queued().aggregate::<T>()`. Lock primitive (runtime-agnostic — no tokio dep, matching the crate's RPITIT async surface): - AsyncLock / AsyncLockManager traits + InMemoryAsyncLock / InMemoryAsyncLockManager, a hand-rolled waker-based async mutex (try_lock/unlock stay sync; only acquire awaits). QueuedRepository<R, AsyncLockManager> (struct/Clone bound moved to the impls so an async lock manager is accepted): - AsyncGetStream / AsyncTransactionalCommit with the sync locking contract: reads acquire+hold the per-stream lock, commit releases on success and holds on error, multi-locks acquired in sorted/deduped order. Keyed by StreamIdentity::storage_key consistently across get/commit/unlock. - Non-locking forwards (drop-in completeness): AsyncSnapshotStore, AsyncReadModelWritePlanStore, AsyncRelationalReadModelQueryStore, AsyncInboxStore. - AsyncGetWithOpts / AsyncGetAllWithOpts (no_lock opt-out) + AsyncUnlockableRepository. - Queueable::queued_async() / queued_async_with(); AsyncAggregateRepository gains get_with/peek/get_all_with/peek_all/abort/unlock mirroring the sync layer. Adversarial review (3 lenses) found two latent defects in unlock(), both fixed: waking wakers while holding the std Mutex guard could (1) poison/brick the lock if a waker panics and (2) deadlock if a waker synchronously re-polls. unlock() now drains under the guard and wakes outside it; regression tests cover both. Tests: async lock unit tests (incl. re-entrant + panicking waker regressions) and tests/queued_repo_async (mutual exclusion, per-aggregate granularity, no_lock peek, abort release). Sync QueuedRepository API and its tests unchanged. Part of [[tasks/async-only-consolidation]] (Phase 2). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Replaces the legacy `Bus::from_queue`/`microsvc::listen` queue tests with the
async `InMemoryBus` + `BusConsumer::listen` (competing-consumer queues keyed by
command name). The legacy `stats.handled`/`stats.failed` handle has no async
analogue, so:
- success is asserted via domain outcomes (committed aggregate state), not counts;
- failure tolerance is asserted by showing the consumer drains past a failing
message and still processes the rest;
- metadata->Session is verified through `whoami` over the bus (works via
run_source -> dispatch_message -> message_to_session), with a negative control
under FailurePolicy::Stop;
- arbitrary queue names ("counters"/"creates") become command-name routing, so
two services on one bus consume disjoint command queues without competing.
Confirms Phase 1 needs no new runtime capability — metadata->Session already
works and the stats gap is a test-rewrite. microsvc crate: 15 passed.
Part of [[tasks/async-only-consolidation]] (Phase 1).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…InMemoryBus Replaces the threaded legacy-bus choreography (InMemoryQueue + OutboxWorkerThread::spawn_routed + microsvc::listen per service + sleep-poll) with a deterministic, thread-free drive over the async InMemoryBus: - `publish_pending_outbox` claims each service's outbox and forwards messages by destination — worker-addressed messages are point-to-point commands (send_message → consumed via `listen`), saga-addressed messages are events (publish_message → consumed via `subscribe`). - Each round uses a FRESH bus (the in-memory topic log is retained across reads, so a shared bus would re-deliver every prior event to the saga), forwards the pending outbox backlog, then drains the consumers. The loop ends when no service has pending work — i.e. the saga reached Completed. The `stats.handled` assertions (no async analogue) are dropped in favor of the existing domain assertions (saga/order Completed, inventory 95 available / 5 reserved, payment successful). Test 1 (saga_orchestrated) was already bus-free and is unchanged. Both tests pass. Part of [[tasks/async-only-consolidation]] (Phase 1). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ed.rs) tests/sagas/distributed.rs drove the order-fulfillment saga over the raw `bus::Bus`/`Subscribable` API with hand-spawned threads and manual aggregate handling (bus.subscribe(&[names]) -> events.recv() loops). The async InMemoryBus has no raw-receiver equivalent — listen/subscribe are Service-driven — so the file cannot be faithfully migrated; a rewrite would duplicate the async microsvc_saga::saga_distributed test (same saga) plus the matrix metadata coverage. Removed as superseded (owner-confirmed): no coverage is lost. Also drops the now-unused event payloads in tests/sagas/order/events.rs (only distributed.rs constructed them). sagas crate: 7 passed, no warnings. Part of [[tasks/async-only-consolidation]] (Phase 1). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…board onto InMemoryBus The projection handlers of BOTH the gold-standard matrix and the board still decoded via bus::Event (`Event::try_from(ctx.message())` + event.decode/ json_decode + event.id) — so the legacy bus could not be removed without touching the gold-standard test. Decoupled both (refactor, not delete): - decode straight from ctx.message().payload(): serde_json::from_slice for the matrix (JSON), BitcodePayloadCodec::decode for the board (bitcode — identical bytes to the old event.decode()); match on ctx.message().name(); event id from ctx.message().id(). Dropped the bus::Event `event()` helper from both projection handlers/mod.rs. - board main.rs: replaced InMemoryQueue + OutboxWorkerThread + the threaded start_board_projection_service + wait_for_* polling with publish_pending_outbox (fan-out events) + a single bus.subscribe; the projection's monotonic source_version guard makes the per-event-type drain order-independent. Added projections_service::load_board (direct read) replacing the poll loop. matrix: 2 passed (in-memory cell + refactored saga, both exercise the decode); board: 3 passed; sagas: 7 passed. clippy/fmt clean. Part of [[tasks/async-only-consolidation]] (Phase 1 — last test migration before the src removal). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The legacy sync bus is fully superseded by the async bus facade (InMemoryBus + the BusConsumer listen/subscribe + OutboxSource) and had no remaining consumers after the test migrations. Removed: - `src/bus/` entirely (Bus/Subscribable/InMemoryQueue/Listener/Sender/EventBus/ Event/Publisher, ~1.4k lines). - `OutboxWorkerThread` + WorkerStats + OutboxWorkerJoinError (the threaded outbox->bus bridge) and `src/outbox_worker/thread.rs`. - The bus-gated `microsvc::service` surface: `dispatch_event`, `dispatch_listened_event`, `subscribe`/`listen`, `TransportHandle` + `TransportStats`/`TransportJoinError`, and the `From<&Event> for Message` / `TryFrom<&Message> for Event` / `from_bus_event` bridges (+ their unit tests). - The `bus` Cargo feature (out of `default`); `http`/`grpc` no longer depend on it — they use the unconditional `microsvc::Message`, confirmed by building `--features http,grpc`. - The bus-gated crate-root re-exports (`InMemoryQueue`, `bus::Message`, the threaded-worker types). All consumers were migrated first (transport_subscribe/listen, microsvc_saga, the board) or removed as superseded (sagas/distributed.rs), and both projection handlers were decoupled from `bus::Event`. Default test sweep: 238 lib + all integration crates green; `--features http,grpc` builds; clippy/fmt clean. Closes Phase 1 of [[tasks/async-only-consolidation]] — one async bus facade, no sync bus path. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Converts the microsvc handler model from sync to async, the foundation for dropping the sync repository API so all backends are async-only (the sync/async mix was the source of subtle bugs). Core (lib green; integration test crates migrated in follow-up commits): - HandlerFn is now `dyn for<'a> Fn(&'a Context<'a, D>) -> Pin<Box<dyn Future< Output=Result<Value, HandlerError>> + Send + 'a>>`, with an `AsyncHandler<'a,D>` HRTB helper trait so `async fn handle(ctx: &Context<D>)` registers directly. Guards stay synchronous. - Service::dispatch / dispatch_message / dispatch_request / invoke are async. - dependencies.rs: HasRepo/HasReadModelStore now resolve via the ASYNC repo + read-model traits (+ HasRepo for AsyncAggregateRepository / AsyncSnapshotAggregateRepository). - run_source + the http/grpc/knative transports await dispatch. - src unit tests converted (async-closure handlers + awaited dispatch). Handler authors write `async fn handle`; closures need an explicit ctx type annotation and must extract owned values before the `async move` (the future cannot borrow ctx across the await — an HRTB-closure limitation). cargo build (default + --features http,grpc) green; 238 lib tests pass. NOTE: tests/ integration crates still use the sync handler API and are migrated in the following commits (all-or-nothing handler switch). Part of [[tasks/async-only-consolidation]] (Phase 3). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Completes the integration half of the async handler switch: all 21 test crates
(microsvc, sagas, the gold-standard distributed_read_model matrix, the board,
the transport conformance crates, and the ~15 direct-repo crates) now use the
async handler + async repo API exclusively:
- handlers are `async fn handle(ctx: &Context<'_, D>)` with awaited
ctx.repo().get/commit/peek and ctx.repo().outbox(msg).commit(&mut a).await;
read-model handlers use workspace_async()/load_async()/commit_async().await.
- services build with .queued_async().async_aggregate(); inline handler closures
use the `|ctx: &Context<D>| { extract ctx reads; async move { ... } }` form.
- test bodies await dispatch and the now-async repo reads.
Guards stay synchronous. Assertions and domain logic unchanged. The sync repo
trait surface is still present (deleted next); 502 default tests pass, the gold
-standard matrix's gated cells compile, http/grpc/sqlite cells pass.
Part of [[tasks/async-only-consolidation]] (Phase 3).
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Deletes the entire synchronous repository/read-model/snapshot trait surface, now unused after the async handler switch. This eliminates the sync/async mix that was the source of subtle combination bugs: there is exactly one (async) path for every backend. Removed (traits + all backend impls + re-exports): - repository: Get/Commit/Repository (repository.rs), GetOne/GetMany/Gettable (gettable.rs), the TransactionalCommit trait (batch.rs; CommitBatch kept). - snapshot: sync SnapshotStore + sync SnapshotAggregateRepository/SnapshotOutboxCommit. - read_model: sync ReadModelWritePlanStore/RelationalReadModelQueryStore, the sync ReadModelWorkspace load/commit impl, ReadModelLoadBuilder, ReadModelWorkspaceExt, and ReadModelWritePlanBuilder::commit (async equivalents kept). - aggregate: GetAggregate/GetAllAggregates/CommitAggregate + the sync AggregateRepository/AggregateBuilder (AsyncAggregateRepository/Builder kept). - commit_builder: SyncCommitBuilder/SyncStagedCommitBuilder/exts. - outbox: SyncOutboxCommit/SyncOutboxCommitExt (outbox_sync/commit_sync). - hashmap/postgres/sqlite/in-memory backends: their sync impls. - queued_repo: the sync QueuedRepository impls + sync Queueable::queued; the sync lock module (Lock/LockManager/InMemoryLock/InMemoryLockManager) is now fully unused and deleted (Async lock variants kept; LockError kept). - src/ unit tests that exercised the removed sync surface, converted to async. Also converted 5 remaining fully-sync integration crates the earlier sweep missed (bomberman [19 files], read_model_relationship_includes, read_model_commit_bridge, sourced_upcasting, transport_conformance's store_outbox). cargo test: 490 passed / 0 failed; --features http,grpc / postgres / sqlite all build; clippy clean; no sync trait remains in src/. Completes Phase 3 of [[tasks/async-only-consolidation]] — HashMap/SQLite/Postgres all async-only and consistent. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
It is only used by the postgres/sqlite-gated matrix cells, so it (and its `TableSchemaRegistry` import) tripped a dead-code warning on the default build. Gate both with cfg(any(feature = "postgres", feature = "sqlite")) to match the call sites. Default clippy is now fully clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ion) - src/ unit tests (outbox/commit, snapshot/in_memory, snapshot/repository, read_model/in_memory, commit_builder, outbox_worker/store, hashmap_repo): replace the custom busy-poll `block_on` (no-op waker, ignores Poll::Pending — would spin on any yielding future) with `#[tokio::test]`. Transport modules keep their intentionally runtime-free block_on. - board projection handler: `event_version` returns Result<_, HandlerError> instead of panicking on a malformed message id; the handler propagates with `?`. Its unit test now asserts the error path. - tests/todos: the bulk-commit roundtrip now asserts the commit succeeds and that exactly 3 todos are present (was: ignored result + an `if !empty` that masked failures). The concurrency-race commits (deliberately may lose the lock) keep their `let _ =`. 490 tests pass; clippy --all-targets clean. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This was referenced Jun 10, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Stacked on #47 (the bus facade + consumer inbox). Targets
feat/bus-transportsso the diff is just the matrix work; retarget tomainonce #47 merges.What
Runs the gold-standard
tests/distributed_read_modelseat-checkout scenario across the full transport × persistence grid (6 × 3 = 18 cells), on the async bus facade only — proving the abstraction holds end-to-end.Each cell drives the same domain flow → read-model projection → query, with the events routed over its transport and persisted on its backend. Broker/DB cells skip when their env var is unset; in-memory × InMemoryBus runs in a plain
cargo test.Gold-standard test: refactored, not deleted
seat_checkout_saga_reserves_seat_and_projects_user_screennow choreographs its full saga / seat / projection flow over the asyncInMemoryBusinstead of the legacyInMemoryQueue/OutboxWorkerThread/bus::Subscribablewiring:publish_pending_outbox(claim → publish → complete) bridges each service's outbox onto the bus — the new-transport equivalent ofOutboxWorkerThread.bus.subscribedrives each service's event reactions across the choreography hops.projection_service/query_servicemodules, and every assertion are unchanged.The existing async SQLite/Postgres flow tests and the HTTP/gRPC command tests are kept. Knative is a first-class cell (CloudEvents POSTed to a local
cloud_events_router), matching the HTTP/gRPC command-ingress surface.Harness
run_checkout_over_bus<B: Bus + BusConsumer, R>— pull buses.run_checkout_over_knative<R>— the HTTP/CloudEvents path.build_collector(transport sink),project_and_assert_checkout(shared projection + assertions).Verification
cargo test --all-features --test distributed_read_modelagainst live NATS / Postgres / RabbitMQ / Kafka: 23 passed / 0 failed (refactored sync test + 18 matrix cells + 2 async flow tests + HTTP/gRPC). Fullcargo test --all-features: 644 / 0. fmt + clippy clean.No sync bus path is used in the scenario; the crate-wide legacy-bus removal remains the separate
tasks/transport-docs-examples-cutover.🤖 Generated with Claude Code
Summary by CodeRabbit