feat: infer and harden bus consumer topology

[patrickleet]

Adds Service::named-derived consumer groups, awaitable bus constructors, shared topology validation, and inferred-group transport coverage.

Implements [[tasks/infer-bus-topology-from-service-name]] and [[tasks/harden-inferred-bus-topology]].

Summary by CodeRabbit

• New Features

  • Services can be named via .named() to provide a stable consumer-group identity.
  • Transport connectors now use a fluent builder (connect().group(...).namespace(...)) for group/namespace configuration.

• Documentation

  • Updated quickstarts and Service Bus docs to clarify group vs. namespace, show named services, and document topology-name validation rules.

• Tests

  • Added integration tests validating named-service consumer-group behavior across transports.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: ec90d37c-a859-45d1-b610-561f1f6ad418

📥 Commits

Reviewing files that changed from the base of the PR and between 64b00a9 and be3d94a.

📒 Files selected for processing (1)

• src/bus/handlers.rs

---

📝 Walkthrough

Walkthrough

This PR establishes stable consumer identity and topology scoping for distributed message buses by introducing Service::named() for services and extending MessageRouter::consumer_group() for routers, refactoring transport adapters to fluent builders (e.g., KafkaBus::connect(brokers).group(...).namespace(...)), and centralizing topology validation/resolution in BusTopologyConfig.

Changes

Consumer Identity, Topology Validation, and Transport Builder Refactoring

Layer / File(s)	Summary
Message Router Consumer Group Contract src/bus/router.rs, src/bus/handlers.rs, src/microsvc/message_router.rs	Add consumer_group(): Option<&str> method to MessageRouter trait (default None); add private group: Option<String> field and public named(...) fluent method to Handlers; implement consumer_group() on Handlers (returns configured group) and Service<D> (returns service name).
Topology Validation and Consumer Group Resolution src/bus/topology.rs	Introduce BusTopologyConfig with optional group and default namespace; implement character/length validation for consumer groups and namespaces (allowing . in namespaces but not groups); provide resolve_consumer_group(...) that prefers explicit override then falls back to router identity; centralize error formatting for topology violations.
Service Stable Broker Identity src/microsvc/service.rs	Add private name: Option<String> field to Service<D> with named(...) and name() accessor APIs; refactor dependency builders (with_repo, with_read_model_store) using internal map_dependencies and replace_dependencies helpers to preserve name across type changes; add test verifying identity preservation.
Bus Consumer Identity Documentation src/bus/bus.rs	Update BusConsumer trait docs to clarify that listen/subscribe default to MessageRouter::consumer_group when explicit group is not configured.
README and API Documentation for Topology README.md, docs/async-transports.md	Document topology semantics: Service::named(..) provides durable consumer group, namespace scopes broker topology names; explain validation rules and competing-vs-fan-out identity behavior; update quickstart and production examples to show named services and namespaced buses.
Bus Module Topology Exports src/bus/mod.rs	Add topology module; export new *BusConnect builder types alongside existing bus structs for Kafka, NATS, RabbitMQ; expose BusTopologyConfig as crate-private re-export.
Kafka Bus Awaitable Builder and Topology Integration src/bus/kafka_bus.rs, tests/kafka_transport/main.rs	Introduce KafkaBusConnect awaitable builder with .group(), .namespace(), .with_fetch_timeout() methods; refactor KafkaBus::connect(brokers) to return builder; add connect_with(...) convenience method; store topology instead of group/namespace strings; update command_prefix/event_prefix to validate namespace and return Result; update listen/subscribe to resolve consumer groups via topology; update tests with fluent API usage and add named-service consumer group verification test.
NATS Bus Awaitable Builder and Topology Integration src/bus/nats_bus.rs, tests/nats_transport/main.rs	Update NatsBus::new to accept only JetStream context; introduce NatsBusConnect builder; add NatsBus::connect(url) builder and connect_with(...) convenience; refactor to store topology; update durable_base to take resolved group string; refactor stream/subject naming via validated namespace; update listen/subscribe to resolve consumer groups via topology; update tests with fluent API usage and add named-service consumer group verification test.
Postgres Bus Topology Integration src/bus/postgres_bus.rs, tests/postgres_transport/main.rs	Refactor PostgresBus to store topology instead of explicit group; change new(pool) to default topology and add new_with_group(pool, group) for explicit group; add group(...) method; update subscribe to resolve consumer group via topology.resolve_consumer_group(...); update tests with fluent .group() API and add named-service consumer group verification test.
RabbitMQ Bus Awaitable Builder and Topology Integration src/bus/rabbit_bus.rs, tests/rabbitmq_transport/main.rs	Introduce RabbitBusConnect awaitable builder; refactor RabbitBus::connect(uri) to return builder; add connect_with(...) convenience; refactor to store topology instead of group/namespace/events_exchange; introduce topology helper methods for queue/exchange naming; update declare_events_exchange to accept explicit exchange name; refactor ensure_subscription, send_message, publish_message, listen, subscribe to use topology helpers and resolve consumer groups; update tests with fluent API usage and add named-service consumer group verification test.
Distributed Read Model Matrix Test Updates tests/distributed_read_model/main.rs	Update persistence × transport matrix fixtures to use fluent bus configuration: NATS/RabbitMQ/Kafka construct with URL/brokers then apply .group("matrix") and .namespace(ns) via chaining; Postgres uses PostgresBus::new(pool).group("matrix").
Lock Manager Documentation Clarification src/lock/async_in_memory.rs	Update doc comment for InMemoryAsyncLockManager to clarify behavior of repeated lookups returning the same Arc.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Possibly related PRs

• hops-ops/distributed#51: This PR builds directly on the router/handlers refactoring by extending MessageRouter/Handlers with durable consumer identity and wiring it through transport adapter connection logic.
• hops-ops/distributed#48: Both PRs modify tests/distributed_read_model/main.rs to update the read-model test setup with new async bus facade configuration using .group(...) and .namespace(...) chaining.

"🐰
I named the service with a hop and a twitch,
Topology tidy, no more pitch,
Builders chain groups and namespaces bright,
Consumers find their seats each night,
Hops celebrate the message-flight!"

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately summarizes the main change: refactoring bus consumer topology inference and validation across multiple bus transports and service configuration.
Docstring Coverage	✅ Passed	Docstring coverage is 100.00% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ 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 codex/inferred-bus-topology-hardening

---

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 and usage tips.}

[coderabbitai]

🧹 Nitpick comments (1)

src/bus/handlers.rs (1)

80-86: ⚡ Quick win

Add a direct unit test for Handlers::named identity exposure.

This new contract is central to inferred group resolution, but there’s no local test asserting named(..) flows through MessageRouter::consumer_group().

Proposed test addition

 #[cfg(test)]
 mod tests {
     use super::*;
@@
+    #[test]
+    fn named_handlers_expose_consumer_group() {
+        let handlers = Handlers::new().named("order-projection");
+        assert_eq!(
+            crate::bus::MessageRouter::consumer_group(&handlers),
+            Some("order-projection")
+        );
+    }
+
     #[test]
     fn subscription_plan_groups_by_kind() {

Also applies to: 115-117

🤖 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/bus/handlers.rs` around lines 80 - 86, Add a focused unit test that
constructs a Handlers instance, calls Handlers::named("some-group"), wires it
through whatever builder/registration path you have so it produces a
MessageRouter (or obtains the router via the same factory used in production),
then assert that MessageRouter::consumer_group() returns "some-group";
specifically reference Handlers::named and MessageRouter::consumer_group() in
the test to ensure the named identity flows through the
registry/listen/subscribe path unchanged.

🤖 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/bus/handlers.rs`:
- Around line 80-86: Add a focused unit test that constructs a Handlers
instance, calls Handlers::named("some-group"), wires it through whatever
builder/registration path you have so it produces a MessageRouter (or obtains
the router via the same factory used in production), then assert that
MessageRouter::consumer_group() returns "some-group"; specifically reference
Handlers::named and MessageRouter::consumer_group() in the test to ensure the
named identity flows through the registry/listen/subscribe path unchanged.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: b98a9adb-3a4e-47a5-b5b7-205e4531b455

📥 Commits

Reviewing files that changed from the base of the PR and between f4a76bc and 64b00a9.

📒 Files selected for processing (19)

• README.md
• docs/async-transports.md
• src/bus/bus.rs
• src/bus/handlers.rs
• src/bus/kafka_bus.rs
• src/bus/mod.rs
• src/bus/nats_bus.rs
• src/bus/postgres_bus.rs
• src/bus/rabbit_bus.rs
• src/bus/router.rs
• src/bus/topology.rs
• src/lock/async_in_memory.rs
• src/microsvc/message_router.rs
• src/microsvc/service.rs
• tests/distributed_read_model/main.rs
• tests/kafka_transport/main.rs
• tests/nats_transport/main.rs
• tests/postgres_transport/main.rs
• tests/rabbitmq_transport/main.rs

[patrickleet]

Addressed in be3d94a: added named_handlers_expose_consumer_group, which constructs Handlers::new().named("order-projection") and asserts that MessageRouter::consumer_group() exposes the same identity.