test: dedup test suites and add failure-path coverage

[patrickleet]

Two halves: remove ~1,100 lines of copy-pasted test scaffolding with zero coverage loss (and one coverage gain), then add the failure-path coverage the suites were missing. No src/ changes.

Part A — dedup

1. Repository conformance mains → one macro
The hashmap/sqlite/postgres conformance mains were 17 near-identical #[tokio::test] shims each. They are now a shared repository_conformance_tests!() macro in tests/persistent_repository_conformance/mod.rs over a normalized async fn repository() -> Option<Backend> factory; each main is ~15 lines. This also fixes a coverage drift: only the hashmap main ran the read_models scenarios — sqlite and postgres now run the full 20-case list including read_models (the SQL factories bootstrap the conformance table), and the #![allow(dead_code)] that papered over the drift is gone.

2. Bus behaviour scenarios ×5 → transport conformance
The point-to-point / fan-out / named-service consumer-group scenarios were copy-pasted byte-identical-modulo-construction across the five bus transport mains. They are now generic scenario fns in tests/transport_conformance/ taking a bus-factory closure; each main keeps a small factory + shims. Kept local on purpose: Kafka's offset-commit point-to-point proof and RabbitMQ's bind-before-publish fan-out + named-service variants — their transport-specific semantics are the point. The sqlite suite's message/id micro-helpers were promoted alongside.

3. Conformance-redundant tests deleted
optimistic_conflict_rolls_back_other_stream_and_snapshot and duplicate_stream_identity_is_rejected_before_sql_writes (postgres main) and snapshots_persist_by_full_stream_identity (both SQL mains) re-proved scenarios the shared conformance suite already runs against every backend. The remaining raw-SQL dialect assertions were deliberately not merged.

4. tests/support/ modules

• support/ids.rs — unique_id/run_token/unique (was copied 6×)
• support/outbox.rs — the scan-four-statuses find_outbox_by_id loop (was copied 7×)
• support/sqlite.rs — the temp-file TempDb with WAL/SHM cleanup (was copied 2×)
• support/env.rs — the broker_env skip guard (3 broker suites)

tests/todos and tests/read_models untouched (owned elsewhere).

Part B — new coverage (all env-gated; suites skip cleanly without services)

5. Broker failure paths — the real-broker suites were entirely happy-path; the ack/nack/dead-letter contract was proven only against in-memory fakes and the SQL buses. Per broker (NATS, RabbitMQ, Kafka):

• retryable_failure_is_redelivered_then_succeeds — Nak / requeue / seek-back redelivery, then drain
• permanent-failure routing — RabbitMQ: the rejected message actually lands in a DLX-configured dead-letter queue; NATS: the MSG_TERMINATED advisory (the parking destination) fires and the message is not redelivered to the durable; Kafka: the adapter has no native DLQ (dead_letter = offset-commit skip), so the test proves no-redelivery-to-group + flow-continues instead
• undecodable_payload_dead_letters_without_blocking — raw garbage on the subject/queue/topic is dead-lettered and subsequent messages still flow

6. Composed at-least-once — publish succeeds, completion write fails (simulated crash), row reclaimed after lease expiry and republished (duplicate delivery, same stable id), both deliveries replayed through run_source into a consumer whose handler commits its effect atomically with an inbox receipt. Exactly one effect commits; the duplicate is acked, not nacked, and the second attempt's effect row is fenced.

7. QueuedRepository stale-lease fencing — writer A holds a durable SqliteLockManager lease, the TTL expires, writer B steals it and commits, then A commits its stale load → A observes ConcurrentWrite (the optimistic check is the backstop once the lock can't protect it) and B's write survives.

8. SQLite 1000-event batch — commit_batch_with_1000_events_round_trips exercises the 999-bind-param multi-row INSERT chunking, asserting count, contiguous sequences, and payloads across chunk seams.

9. Postgres mid-commit fault injection — an ACCESS EXCLUSIVE lock parks a commit_batch mid-transaction, pg_blocking_pids pinpoints the blocked backend, pg_terminate_backend kills it. Nothing persists and the commit fails with a Storage error.

src findings (documented, not fixed — src/ is owned by other agents)

1. SQLSTATE 57P01 classified permanent (src/sqlx_repo/mod.rs::is_sqlx_transient): a terminated/killed postgres backend surfaces as 57P01, which is classified retryable: false because only 40001/40P01 are whitelisted among Database errors. A lost connection is an infrastructure hiccup and should be retryable — as-is, an outbox dispatcher would fail/dead-letter a row because postgres restarted mid-commit. The related connection-failure classes (08***, 57P02, 57P03) have the same gap. The new backend_termination_mid_commit_rolls_back_and_nothing_persists test pins the current behavior with a comment saying which assertion to flip once fixed.

2. Outbox lease deadlines truncate to whole seconds (src/outbox/message.rs::lease_deadline_secs uses .as_secs()): a lease's expiry floor-truncates, so a sub-second lease (or any lease landing just before a second boundary) can be born up to ~1s shorter than requested — even already expired, at which point the store fences the claiming worker's own complete with InvalidState. Found when the composed at-least-once test flaked under load with a 100ms lease; the test now models the retry as a fresh worker with a comfortable lease, and the quirk is called out in the commit.

3. Service::dispatch_message swallows payload decode failures (src/microsvc/service.rs, message_to_json_input fallback): a non-JSON payload falls back to Value::Null input instead of surfacing HandlerError::DecodeFailed, so an undecodable message is silently dispatched rather than dead-lettered (the DecodeFailed → permanent mapping in microsvc/error.rs is unreachable from this path). Discovered while writing the undecodable_payload_* tests — they use a payload-decoding handler as the decode point, which is the framework's effective contract today. Worth deciding whether that fallback is intentional.

Test results

$ cargo test --workspace --all-features --all-targets
  (AMQP_URL, KAFKA_BROKERS, NATS_URL, DATABASE_URL all set — broker suites exercised for real)
45 suites, 825 tests: all passing, 0 failed

Highlights:
  hashmap/sqlite/postgres_repository_conformance: 20 passed each (sqlite/postgres +3 read_models cases vs main)
  nats_transport:      8 passed (3 new failure-path tests)
  rabbitmq_transport:  8 passed (3 new failure-path tests, incl. real DLQ routing)
  kafka_transport:     8 passed (3 new failure-path tests)
  transport_in_memory: 13 passed (new composed at-least-once test)
  queued_repo:         6 passed (new stale-lease fencing test)
  sqlite_repository:   11 passed (new 1000-event chunking test)
  postgres_repository: 9 passed (new mid-commit termination test)

cargo fmt clean; cargo clippy --workspace --all-features --all-targets carries only the two warnings that pre-exist on main (todos MutexGuard-across-await, lib manual_async_fn).

🤖 Generated with Claude Code

https://claude.ai/code/session_01DzYSVLas93c7LbgHJWsW7n

[coderabbitai]

Warning

Review limit reached

@patrickleet, you've reached your PR review limit, so we couldn't start this review.

Next review available in: 30 minutes

Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available.
You're only billed for reviews past your plan's rate limits ($0.25/file).

How can I continue?

After more reviews become available, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews.

How do review limits work?

CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability.

For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window.

Please refer docs for additional details.

Review details

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: f19e37e3-6c14-4ce6-927b-f078c92437bf

📥 Commits

Reviewing files that changed from the base of the PR and between ab74c4d and a852fa8.

📒 Files selected for processing (24)

• tests/distributed_read_model/main.rs
• tests/hashmap_repository_conformance/main.rs
• tests/kafka_transport/main.rs
• tests/nats_transport/main.rs
• tests/persistent_repository_conformance/inbox.rs
• tests/persistent_repository_conformance/mod.rs
• tests/persistent_repository_conformance/outbox.rs
• tests/persistent_repository_conformance/read_models.rs
• tests/persistent_repository_conformance/scenario.rs
• tests/postgres_repository/main.rs
• tests/postgres_repository_conformance/main.rs
• tests/postgres_transport/main.rs
• tests/queued_repo/main.rs
• tests/rabbitmq_transport/main.rs
• tests/sql_lock_manager/main.rs
• tests/sqlite_repository/main.rs
• tests/sqlite_repository_conformance/main.rs
• tests/sqlite_transport/main.rs
• tests/support/env.rs
• tests/support/ids.rs
• tests/support/outbox.rs
• tests/support/sqlite.rs
• tests/transport_conformance/mod.rs
• tests/transport_in_memory/main.rs

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch review/tests-dedup-coverage

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}