feat!: async microsvc transports + persistence matrix + async-only consolidation

[patrickleet]

The umbrella PR to main. Builds the async microsvc transport layer, proves the distributed read-model flow across a transport×persistence matrix, and consolidates the crate to a single async-only repository + bus surface (removing the sync/async mix that caused subtle combination bugs).

184 files, +13,390 / −8,045 (39 commits). Each slice was implemented, put through adversarial multi-agent review with fixes applied, and committed separately. cargo test (default, no brokers): 490 passed / 0 failed; broker/feature cells verified against real infra; clippy --all-targets clean.

---

1. Async microservice transports

microsvc owns handler registration + dispatch; transport adapters own receive/ack/retry/publish. Shared vocabulary lives in microsvc::transport.

• Core: TransportError, FailurePolicy/FailureAction, RunOptions/InboxHook, TransportCapabilities, stable-id rules; the AsyncMessageSource/ReceivedMessage + run_source receive loop; AsyncMessagePublisher, OutboxMessage→Message, OutboxDispatcher (claim→publish→complete); a reusable conformance harness + in-memory reference.
• Bus facade: Bus (produce) + BusConsumer (listen/subscribe) over InMemoryBus/NatsBus/PostgresBus/RabbitBus/KafkaBus/KnativeBus.
• Consumer inbox: a CommitBatch participant (consumer_inbox table) giving effectively-once (consumer, message_id) semantics across in-memory/SQLite/Postgres.
• Adapters (feature-gated, verified against real brokers via docker): Postgres (OutboxSource, SKIP LOCKED + lease), NATS JetStream, RabbitMQ, Kafka, and Knative/HTTP CloudEvents ingress (200/503/422 mapping; knative_triggers() from subscription_plan()).
• CI: compose.yaml + reusable .github/workflows/integration-*.yaml broker jobs on PRs and push-to-main.

2. Transport × persistence matrix

The gold-standard seat-checkout saga (domain flow → read-model projection → query) runs across a 6×3 grid — { InMemoryBus, NatsBus, RabbitBus, KafkaBus, PostgresBus, Knative } × { HashMap, SQLite, Postgres } — plus the HTTP/gRPC command surface, proving the abstraction holds over every transport and store. The same generic async flow + project_message_async + query helpers drive all cells.

3. Async-only consolidation (breaking)

End state: one async bus facade, one async repository surface; HashMap/SQLite/Postgres all behave the same.

• Legacy sync bus removed — src/bus/ (~1.4k lines), OutboxWorkerThread, the bus-gated service.rs surface + Event↔Message bridges, and the bus Cargo feature (http/grpc no longer depend on it). Projection handlers decoupled from bus::Event.
• Async ergonomic builders — ReadModelWorkspace gains load_async/commit_async/workspace_async(); QueuedRepository gains a runtime-agnostic waker-based async lock + the full async repo surface, so .queued_async().async_aggregate::<T>() serializes per-aggregate get/commit (an adversarial review caught + fixed two latent unlock defects).
• Async handler model — HandlerFn is a boxed Send future with an AsyncHandler HRTB trait, so async fn handle(ctx: &Context<'_, D>) registers directly; dispatch/invoke are async; guards stay sync.
• Sync repository API deleted — Get/Commit/Repository/GetOne/GetMany/TransactionalCommit, sync SnapshotStore/ReadModelWritePlanStore/RelationalReadModelQueryStore, sync AggregateRepository/QueuedRepository/CommitBuilder/OutboxCommit, and the now-unused sync lock module — across all backends.

⚠️ Breaking changes

• Handlers are now async fn; Service::dispatch/dispatch_message/dispatch_request are async.
• The synchronous repository/read-model/snapshot/lock APIs are removed — use the Async* equivalents, .queued_async(), and .async_aggregate().
• The legacy bus::* module and OutboxWorkerThread are gone; use microsvc::transport (InMemoryBus + OutboxSource/run_source).

🤖 Generated with Claude Code

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Removes distributed processed-message idempotency (DB, API, stores, repos, tests) and adds a new async transport subsystem: core traits, runner, publishers, adapters (NATS/RabbitMQ/Knative), outbox dispatcher/source with explicit-ID claiming, conformance tests, and CI/compose updates.

Changes

Processed Message Idempotency Removal

Layer / File(s)	Summary
Database schema cleanup migrations/postgres/0001_initial.sql, migrations/sqlite/0001_initial.sql	Drop read_model_processed_messages table from initial migrations; Postgres adds aggregate_events(recorded_at) index.
Read-model session and API surface src/read_model/session.rs, src/read_model/mod.rs, src/repository/async_repository.rs, src/lib.rs	Remove ProcessedMessageMark and related API: ReadModelWritePlan now only contains mutations; ReadModelCommitOutcome simplified; ReadModelAdapterCapabilities drops processed_messages; is_processed/is_processed_async removed; mark_processed APIs removed and re-exports cleared.
In-memory and HashMap store implementations src/read_model/in_memory.rs, src/hashmap_repo/repository.rs	Remove processed_messages state, staging, and checks; adapt apply/commit signatures; drop is_processed delegations; update unit tests accordingly.
Postgres and SQLite repositories src/postgres_repo/mod.rs, src/sqlite_repo/mod.rs	Remove SQL-level processed-message deduplication and insertion; sql_read_model_capabilities drops the flag; always commit read-model transactions after applying plans; add optional message_ids filtering to outbox claim queries.
Outbox & table plumbing src/outbox/table.rs	Construct TableWritePlan with only mutations vector; tests adjusted accordingly.
Tests and examples various tests/*	Remove or simplify tests that asserted processed-message marking or duplicate-skip behavior; delete the distributed-idempotency test file and adjust many projection/handler tests to no longer call mark_processed.
Exports src/lib.rs	Remove ProcessedMessageMark from crate re-exports.

Async Transport Subsystem

Layer / File(s)	Summary
Transport module foundation and core contracts src/microsvc/mod.rs, src/microsvc/transport/mod.rs, src/microsvc/transport/source.rs	Add transport submodule; define AsyncMessageSource and ReceivedMessage contracts and document receive/publish semantics.
Runner and run options src/microsvc/transport/runner.rs, src/microsvc/transport/run_options.rs, src/microsvc/transport/stable_id.rs	Implement run_source loop with failure-policy routing, inbox vs idempotent delivery modes, and stable message-id validation; include unit tests for ordering, failure routing, and inbox behavior.
Transport error / failure policy src/microsvc/transport/error.rs, src/microsvc/transport/failure_policy.rs	Add TransportError/TransportErrorKind with classifier and mapping to FailureAction via FailurePolicy::resolve.
Publisher and outbox dispatcher/source src/microsvc/transport/publisher.rs, src/microsvc/transport/outbox_dispatch.rs, src/microsvc/transport/outbox_source.rs	Add AsyncMessagePublisher trait and default publish_batch; add OutboxDispatcher to claim → map → publish → settle outbox rows and OutboxSource to expose outbox rows as an AsyncMessageSource.
Transport capabilities and adapters src/microsvc/transport/capabilities.rs, src/microsvc/transport/nats.rs, src/microsvc/transport/rabbitmq.rs, src/microsvc/transport/knative.rs	Record per-transport capabilities; add NATS JetStream, RabbitMQ, and Knative/CloudEvents adapters with publish/receive/settle implementations and adapter tests.
Conformance tests, feature flags, CI, compose tests/transport_conformance/*, tests/transport_in_memory/*, tests/nats_transport/*, tests/rabbitmq_transport/*, tests/knative_cloudevents/*, Cargo.toml, .github/workflows/on-pr-quality.yaml, compose.yaml	Add reusable transport conformance harness, in-memory and adapter integration tests, new Cargo feature flags for adapters, CI job update to run postgres transport tests, and compose services for RabbitMQ/Kafka/NATS.

Outbox Worker Enhancement

Layer / File(s)	Summary
Outbox claim-by-ids src/outbox_worker/store.rs, tests/persistent_repository_conformance/outbox.rs	Add optional message_ids: Option<Vec<String>> to ClaimOutboxMessages, implement ClaimOutboxMessages::for_ids(...), and adapt sync/async stores to honor explicit-id claim filters; add conformance tests verifying only requested IDs are claimed.

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• patrickleet/sourced_rust#44: Prior work on read-model processed-message idempotency and schema that this PR removes or replaces.
• patrickleet/sourced_rust#35: Introduced read-model session/write-plan machinery and processed-message primitives removed by this PR.
• patrickleet/sourced_rust#37: Related changes touching processed-message handling and repository commit logic.

"I nibble through rows and headers bright,
No processed marks left in morning light,
Runners hum, publishers hop in time,
Outbox claims and settles one message, one chime,
A rabbit cheers the tests that pass tonight." 🐰✨

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/bus-transports

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/read_model/session.rs`:
- Around line 59-68: ReadModelCommitOutcome currently always reports applied
(ReadModelCommitOutcome::applied() / was_applied() -> true) which removes the
storage-side replay/dedupe signal; restore a proper commit outcome enum or add
explicit skipped/replay semantics and wire them into the storage adapters:
implement a non-applied outcome (e.g., Skipped), make
ReadModelCommitOutcome::was_applied return true only for the applied variant,
and ensure the storage implementations call mark_processed/was_skipped and
populate/read the read_model_processed_messages behavior used by Postgres/SQLite
adapters so the write-plan commit boundary acts as a replay barrier; update the
methods ReadModelCommitOutcome, applied, was_applied, mark_processed,
was_skipped and ensure read_model_processed_messages handling is present in the
storage adapter code paths or alternatively make projector inbox/stable-id
enforcement explicit in the public API so non-idempotent handlers are rejected.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 2d2a8743-5492-4d7d-99bf-616acca9dbe0

📥 Commits

Reviewing files that changed from the base of the PR and between c4656dc and bb10271.

📒 Files selected for processing (45)

• migrations/postgres/0001_initial.sql
• migrations/sqlite/0001_initial.sql
• src/commit_builder/mod.rs
• src/hashmap_repo/repository.rs
• src/lib.rs
• src/microsvc/mod.rs
• src/microsvc/transport/capabilities.rs
• src/microsvc/transport/error.rs
• src/microsvc/transport/failure_policy.rs
• src/microsvc/transport/mod.rs
• src/microsvc/transport/outbox_dispatch.rs
• src/microsvc/transport/publisher.rs
• src/microsvc/transport/run_options.rs
• src/microsvc/transport/runner.rs
• src/microsvc/transport/source.rs
• src/microsvc/transport/stable_id.rs
• src/outbox/table.rs
• src/outbox_worker/store.rs
• src/postgres_repo/mod.rs
• src/read_model/in_memory.rs
• src/read_model/mod.rs
• src/read_model/session.rs
• src/repository/async_repository.rs
• src/sqlite_repo/mod.rs
• tests/async_repository/main.rs
• tests/distributed_read_model/main.rs
• tests/distributed_read_model/projection_service/handlers/checkout.rs
• tests/distributed_read_model/projection_service/handlers/seat.rs
• tests/distributed_read_model/projection_service/mod.rs
• tests/distributed_read_model/projection_service/service.rs
• tests/distributed_read_model_board/main.rs
• tests/distributed_read_model_board/projections_service/handlers/board.rs
• tests/distributed_read_model_board/projections_service/mod.rs
• tests/hashmap_repository_conformance/main.rs
• tests/persistent_repository_conformance/outbox.rs
• tests/persistent_repository_conformance/read_models.rs
• tests/postgres_repository/main.rs
• tests/postgres_repository_conformance/main.rs
• tests/read_model_distributed_idempotency/main.rs
• tests/read_model_relationship_includes/main.rs
• tests/read_model_session/main.rs
• tests/sqlite_repository/main.rs
• tests/sqlite_repository_conformance/main.rs
• tests/transport_conformance/mod.rs
• tests/transport_in_memory/main.rs

💤 Files with no reviewable changes (9)

• tests/read_model_distributed_idempotency/main.rs
• tests/distributed_read_model/projection_service/service.rs
• src/commit_builder/mod.rs
• migrations/sqlite/0001_initial.sql
• tests/distributed_read_model_board/projections_service/mod.rs
• tests/distributed_read_model_board/projections_service/handlers/board.rs
• tests/read_model_relationship_includes/main.rs
• migrations/postgres/0001_initial.sql
• src/repository/async_repository.rs

[patrickleet]

Progress update: conformance harness + Postgres transport (OutboxSource, outbox-backed) landed and verified against real Postgres (concurrent SKIP-LOCKED claim safety, retry, dead-letter). compose.yaml now defines postgres/rabbitmq/kafka/nats for local integration testing, and the Postgres transport runs in CI. Next: NATS/RabbitMQ adapters + CI jobs, Knative ingress, docs cutover.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

src/microsvc/transport/outbox_source.rs (1)

171-188: ⚡ Quick win

Clarify or replace the hand-rolled block_on to avoid hangs when futures can suspend.

outbox_source.rs’s helper busy-spins on Poll::Pending and never registers wakeups. The current tests don’t hit Pending because they use HashMapOutboxStore, whose AsyncOutboxStore methods are pure async move blocks with no .await (so they complete synchronously). If these tests ever switch to stores like SqliteOutboxStore/PostgresOutboxStore (which do .await), they can hang—so either use a real test executor or explicitly document/enforce the “never suspends” precondition for this helper.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/microsvc/transport/outbox_source.rs` around lines 171 - 188, The custom
block_on in outbox_source.rs busy-loops and never registers wakeups, causing
hangs when futures suspend; replace it with a real executor call (e.g. use
futures::executor::block_on or accept a runtime handle and call
tokio::runtime::Handle::block_on) in the block_on function so suspended futures
are properly woken and polled; alternatively change the API around
AsyncOutboxStore consumers to run on an injected executor or document/enforce
that AsyncOutboxStore methods never .await and add tests/assertions to prevent
using stores (like SqliteOutboxStore/PostgresOutboxStore) that suspend. Ensure
references to the function name block_on in outbox_source.rs are updated to use
the chosen executor approach.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/microsvc/transport/outbox_source.rs`:
- Around line 62-69: The builder currently allows lease == Duration::ZERO and
batch_size == 0 which silently misconfigures behavior (recv() will behave
incorrectly); add upfront validation in the builder methods `with_lease` and
`with_batch_size` to reject invalid values: in `with_lease(lease: Duration)`
assert/validate that `lease > Duration::ZERO` and return/panic with a clear
message if not, and in `with_batch_size(batch_size: usize)` assert/validate that
`batch_size > 0` and return/panic with a clear message if not; keep these checks
inside the existing `with_lease` and `with_batch_size` functions so callers fail
fast rather than causing `recv()` to silently misbehave.

---

Nitpick comments:
In `@src/microsvc/transport/outbox_source.rs`:
- Around line 171-188: The custom block_on in outbox_source.rs busy-loops and
never registers wakeups, causing hangs when futures suspend; replace it with a
real executor call (e.g. use futures::executor::block_on or accept a runtime
handle and call tokio::runtime::Handle::block_on) in the block_on function so
suspended futures are properly woken and polled; alternatively change the API
around AsyncOutboxStore consumers to run on an injected executor or
document/enforce that AsyncOutboxStore methods never .await and add
tests/assertions to prevent using stores (like
SqliteOutboxStore/PostgresOutboxStore) that suspend. Ensure references to the
function name block_on in outbox_source.rs are updated to use the chosen
executor approach.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: e64e78d8-31d7-4cae-b5b8-0ceb073d0b9b

📥 Commits

Reviewing files that changed from the base of the PR and between bb10271 and 70298e1.

📒 Files selected for processing (5)

• .github/workflows/on-pr-quality.yaml
• compose.yaml
• src/microsvc/transport/mod.rs
• src/microsvc/transport/outbox_source.rs
• tests/postgres_transport/main.rs

✅ Files skipped from review due to trivial changes (1)

• .github/workflows/on-pr-quality.yaml

🚧 Files skipped from review as they are similar to previous changes (1)

• src/microsvc/transport/mod.rs

[coderabbitai]

🧹 Nitpick comments (2)

tests/nats_transport/main.rs (1)

62-72: 💤 Low value

Box::leak creates intentional memory leaks.

Using Box::leak(subject.clone().into_boxed_str()) to obtain a &'static str works but leaks memory on each test run. Since these are integration tests that run a bounded number of times, the practical impact is negligible. However, if the Service::event() API could accept &str or String instead of &'static str, this pattern wouldn't be necessary.

Also applies to: 108-120

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/nats_transport/main.rs` around lines 62 - 72, The test leaks memory by
using Box::leak(subject.clone().into_boxed_str()) to produce a &'static str for
Service::event; change the Service::event API to accept an owned String or a
generic Into<Cow<'static, str>>/impl Into<Cow<'_, str>> (or &str) so callers can
pass subject.clone() or subject.as_str() without leaking, then update the test
calls that use Box::leak (including the other occurrence at lines 108-120) to
pass the String/&str directly to Service::event.

src/microsvc/transport/nats.rs (1)

171-186: 💤 Low value

Consider documenting or making batch size configurable.

Fetching one message at a time (max_messages(1)) is straightforward but may limit throughput in high-volume scenarios. This is acceptable for a first implementation, but a future enhancement could allow configuring a prefetch buffer or larger batch size with internal iteration.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/microsvc/transport/nats.rs` around lines 171 - 186, The recv method
currently hardcodes batch().max_messages(1) which limits throughput; make the
batch size configurable by adding a field (e.g., batch_size: usize or u32) to
the NATS transport struct and replace the literal 1 in recv with that field (use
consumer.batch().max_messages(self.batch_size) and keep
expires(self.fetch_timeout) as-is), update the constructor/initialization to
accept a configurable default (and document the new field in the struct
comment), and ensure NatsReceived::from_jetstream is still applied per message
when iterating the returned batch; consider validating a sensible default and
bounds for batch_size.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Nitpick comments:
In `@src/microsvc/transport/nats.rs`:
- Around line 171-186: The recv method currently hardcodes
batch().max_messages(1) which limits throughput; make the batch size
configurable by adding a field (e.g., batch_size: usize or u32) to the NATS
transport struct and replace the literal 1 in recv with that field (use
consumer.batch().max_messages(self.batch_size) and keep
expires(self.fetch_timeout) as-is), update the constructor/initialization to
accept a configurable default (and document the new field in the struct
comment), and ensure NatsReceived::from_jetstream is still applied per message
when iterating the returned batch; consider validating a sensible default and
bounds for batch_size.

In `@tests/nats_transport/main.rs`:
- Around line 62-72: The test leaks memory by using
Box::leak(subject.clone().into_boxed_str()) to produce a &'static str for
Service::event; change the Service::event API to accept an owned String or a
generic Into<Cow<'static, str>>/impl Into<Cow<'_, str>> (or &str) so callers can
pass subject.clone() or subject.as_str() without leaking, then update the test
calls that use Box::leak (including the other occurrence at lines 108-120) to
pass the String/&str directly to Service::event.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 1e64c495-1c45-4533-b7ac-cbb2d7b89eaf

📥 Commits

Reviewing files that changed from the base of the PR and between 70298e1 and b7ca692.

📒 Files selected for processing (5)

• .github/workflows/on-pr-quality.yaml
• Cargo.toml
• src/microsvc/transport/mod.rs
• src/microsvc/transport/nats.rs
• tests/nats_transport/main.rs

🚧 Files skipped from review as they are similar to previous changes (2)

• src/microsvc/transport/mod.rs
• .github/workflows/on-pr-quality.yaml

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/microsvc/transport/rabbitmq.rs`:
- Around line 120-137: The connect function currently declares the queue with
FieldTable::default() which causes messages rejected with requeue: false (in
your dead_letter/park handlers) to be discarded; update the connect function
(the connect method that calls connect_channel and queue_declare, and returns
Self::new) to either (A) build and pass a FieldTable that includes the
dead-letter/parking args (e.g., "x-dead-letter-exchange" and, if needed,
"x-dead-letter-routing-key"/"x-dead-letter-queue" entries) derived from
configuration or env and use that instead of FieldTable::default(), or (B)
validate that DLX/parking config exists and return a clear TransportError
(instead of declaring without args) when it does not; ensure you still wrap AMQP
errors with retryable("amqp queue_declare", err) and return the error type used
by connect (TransportError).
- Around line 63-80: message_properties currently only writes headers and
message_id, so content_type is lost; update message_properties to set AMQP
BasicProperties.content_type when Message.content_type is present (use
properties = properties.with_content_type(ShortString::from(ct)) where ct is
message.content_type.as_str()), and also update RabbitReceived::from_delivery to
read delivery.properties.content_type (if present) and pass that through to
Message::new instead of hardcoding "application/json" so the original
content_type survives the round-trip.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 2b995cbd-5a86-44d7-96f9-4fdb04832b45

📥 Commits

Reviewing files that changed from the base of the PR and between b7ca692 and a5b9a8b.

📒 Files selected for processing (5)

• .github/workflows/on-pr-quality.yaml
• Cargo.toml
• src/microsvc/transport/mod.rs
• src/microsvc/transport/rabbitmq.rs
• tests/rabbitmq_transport/main.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/microsvc/transport/mod.rs

[coderabbitai]

⚠️ Potential issue | 🟠 Major | 🏗️ Heavy lift

Fix RabbitMQ default connect to avoid silently discarding dead_letter/park failures

connect declares the durable queue with FieldTable::default() (no x-dead-letter-exchange) while the dead_letter/park handlers reject with requeue: false (lines ~211-225). With no DLX configured, messages routed to dead_letter/park will be discarded instead of dead-lettered/parked (lines ~120-137).

Update the default connect path to either (a) declare DLX/parking-related queue args (e.g., x-dead-letter-exchange, and any required routing-key args) or (b) fail loudly with a clear TransportError when DLX/parking isn’t configured rather than rejecting without requeue.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/microsvc/transport/rabbitmq.rs` around lines 120 - 137, The connect
function currently declares the queue with FieldTable::default() which causes
messages rejected with requeue: false (in your dead_letter/park handlers) to be
discarded; update the connect function (the connect method that calls
connect_channel and queue_declare, and returns Self::new) to either (A) build
and pass a FieldTable that includes the dead-letter/parking args (e.g.,
"x-dead-letter-exchange" and, if needed,
"x-dead-letter-routing-key"/"x-dead-letter-queue" entries) derived from
configuration or env and use that instead of FieldTable::default(), or (B)
validate that DLX/parking config exists and return a clear TransportError
(instead of declaring without args) when it does not; ensure you still wrap AMQP
errors with retryable("amqp queue_declare", err) and return the error type used
by connect (TransportError).

[patrickleet]

Skipping this one, with reason. dead_letter/park reject with requeue: false, which is the AMQP signal for "stop redelivery." With no DLX configured the broker drops the message, which is the correct terminal outcome for "give up on this message" under at-least-once delivery — not a silent bug. The handler doc already states this ("routes to the queue's dead-letter exchange if one is configured, otherwise drops").

Forcing a DLX declaration (or failing loudly) in the default connect would impose a dead-letter/parking topology this thin adapter shouldn't own and would break the common no-DLX dev/test path. A configurable DLX/parking exchange is a reasonable enhancement, but it's a follow-up rather than part of this PR.

[coderabbitai]

@patrickleet, that's a sound rationale. The AMQP reject(requeue: false) semantics are well-defined — broker drops without a DLX, which is the correct terminal outcome for this thin adapter under at-least-once delivery. Noted that the doc already covers the conditional DLX routing. I'll keep this in mind for future reviews of this adapter.

Would you like me to open a follow-up GitHub issue to track the configurable DLX/parking-exchange enhancement for a later PR?

(╥_╥)🐇

---

✏️ Learnings added

Learnt from: patrickleet
Repo: patrickleet/sourced_rust PR: 47
File: src/microsvc/transport/rabbitmq.rs:122-139
Timestamp: 2026-05-29T06:09:28.874Z
Learning: In `src/microsvc/transport/rabbitmq.rs`, `RabbitSource::connect` intentionally declares the queue with `FieldTable::default()` (no `x-dead-letter-exchange`). The `dead_letter`/`park` settlement methods reject with `requeue: false`, which causes the broker to drop the message when no DLX is configured — this is the correct terminal outcome under at-least-once delivery, not a silent bug. Configurable DLX/parking-exchange support is a planned follow-up, not part of the initial PR. The doc comments already state this conditional behavior.

Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

src/microsvc/transport/knative.rs (1)

86-92: 💤 Low value

Consider validating ce-source as required per CloudEvents spec.

The CloudEvents specification requires source as a mandatory context attribute alongside id and type. Currently, only ce-id and ce-type are validated, so malformed events missing ce-source will be accepted. If strict CloudEvents compliance is desired, add validation for ce-source.

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@src/microsvc/transport/knative.rs` around lines 86 - 92, The parse_binary
function currently validates ce-id and ce-type but not ce-source; add validation
to require the ce-source header per the CloudEvents spec by reading
header(headers, "ce-source") and returning an Err (e.g., ok_or("missing
ce-source header")) when absent, then use the resulting source value like the
existing id/name/content_type variables so malformed events without ce-source
are rejected.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@src/microsvc/transport/knative.rs`:
- Around line 190-193: The trigger_name built in knative_triggers (using
subscriber_service and event) can produce invalid Kubernetes names; add a
sanitize function (e.g., sanitize_k8s_name) and call it when composing
trigger_name to enforce RFC1123: convert to lowercase, replace any
non-alphanumeric characters with '-', collapse repeated '-' runs, trim
leading/trailing '-' so the name starts/ends with alphanumeric, and truncate to
63 characters (preserving uniqueness if needed); update the trigger_name
construction (and any other places using subscriber_service or event for
resource names) to use this sanitizer before formatting.

---

Nitpick comments:
In `@src/microsvc/transport/knative.rs`:
- Around line 86-92: The parse_binary function currently validates ce-id and
ce-type but not ce-source; add validation to require the ce-source header per
the CloudEvents spec by reading header(headers, "ce-source") and returning an
Err (e.g., ok_or("missing ce-source header")) when absent, then use the
resulting source value like the existing id/name/content_type variables so
malformed events without ce-source are rejected.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro Plus

Run ID: 5c0ce1d2-ba21-424e-b169-fc46a2537bc8

📥 Commits

Reviewing files that changed from the base of the PR and between a5b9a8b and 512b688.

📒 Files selected for processing (3)

• src/microsvc/transport/knative.rs
• src/microsvc/transport/mod.rs
• tests/knative_cloudevents/main.rs

🚧 Files skipped from review as they are similar to previous changes (1)

• src/microsvc/transport/mod.rs

[patrickleet]

All broker adapters (NATS, RabbitMQ, Kafka) + Postgres transport + Knative ingress are implemented and verified locally against real services (full cargo test --all-features + the four integration suites green: 34 test binaries, 0 failures). CI integration jobs were extracted into reusable ./.github/workflows/integration-*.yaml and now run on PRs and push-to-main. Remaining: the examples cutover / legacy-bus removal (breaking — needs your sign-off) and a Postgres poll/NOTIFY consumer daemon.

[patrickleet]

Added: the Bus facade (send/listen + publish/subscribe)

Built the servicebus-style ergonomic surface on top of the transport adapters (tasks/build-transport-bus-facade, spec specs/transport-bus-facade):

• Bus (produce: send/publish/send_message/publish_message) + BusConsumer (consume: listen/subscribe, generic over the service, deriving names from the service's command/event handlers and running through the shared run_source). Handler code and dispatch_message are unchanged across transports.
• Six *Bus types, each shining in its native topology — point-to-point vs fan-out is a consumer-group/identity choice, not a flattened trait:
  • InMemoryBus — competing queue / retained-log-per-cursor (dev/test reference).
  • NatsBus — shared durable consumer (listen) / per-group durable (subscribe).
  • PostgresBus — bus_queue FOR UPDATE SKIP LOCKED (listen) / bus_log + per-consumer bus_offset Kafka-style fan-out (subscribe).
  • RabbitBus — default exchange → durable queue (listen) / topic exchange → per-group queue (subscribe).
  • KafkaBus — shared consumer group (listen) / group-per-service (subscribe).
  • KnativeBus — Bus-only (produce → broker-ingress POST); consume = manifests(&plan, &subscriptions) role-based Broker/Trigger YAML + cloud_events_router (per-type /cloudevent/<type> routes).

Each transport's integration binary gained a competing-consumer test (one delivery across a shared group) and a fan-out test (every group sees every event), verified against the real broker. cargo test --all-features = 612 passed / 0 failed against live NATS / Postgres / RabbitMQ / Kafka.

docs/async-transports.md documents the facade + a transport-swap example.

Note on Decision #8: PostgresBus uses the spec's sanctioned claim-lease work queue rather than sqlxmq — sqlxmq's always-on push JobRunner doesn't compose with the facade's uniform drain-to-idle run_source model that every other *Bus shares (a claim-lease source returns Ok(None) when empty and stops cleanly). The reasoning is recorded in the postgres_bus.rs module header; sqlxmq remains a documented future backend.

🤖 Generated with Claude Code