From 64b00a98701124846d78ce1d16763c881e18b458 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Wed, 10 Jun 2026 12:59:44 -0500 Subject: [PATCH 1/2] feat: infer and harden bus consumer topology 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]]. --- README.md | 63 ++++-- docs/async-transports.md | 38 +++- src/bus/bus.rs | 7 +- src/bus/handlers.rs | 13 ++ src/bus/kafka_bus.rs | 203 ++++++++++++++++--- src/bus/mod.rs | 8 +- src/bus/nats_bus.rs | 188 ++++++++++++++---- src/bus/postgres_bus.rs | 37 +++- src/bus/rabbit_bus.rs | 171 ++++++++++++---- src/bus/router.rs | 6 + src/bus/topology.rs | 286 +++++++++++++++++++++++++++ src/lock/async_in_memory.rs | 3 +- src/microsvc/message_router.rs | 4 + src/microsvc/service.rs | 58 +++++- tests/distributed_read_model/main.rs | 16 +- tests/kafka_transport/main.rs | 85 +++++++- tests/nats_transport/main.rs | 83 +++++++- tests/postgres_transport/main.rs | 69 ++++++- tests/rabbitmq_transport/main.rs | 98 ++++++++- 19 files changed, 1256 insertions(+), 180 deletions(-) create mode 100644 src/bus/topology.rs diff --git a/README.md b/README.md index 9be9195e..3f75f5ca 100644 --- a/README.md +++ b/README.md @@ -166,20 +166,40 @@ default you replace with a durable adapter. // Persistence: HashMapRepository → durable SQL (features "postgres" / "sqlite") let repo = distributed::PostgresRepository::connect_and_migrate(database_url).await?; let service = distributed::register_handlers!( - Service::new().with_repo(repo.queued().aggregate::()), + Service::new() + .named("todo-api") + .with_repo(repo.queued().aggregate::()), command handlers::todo_create, command handlers::todo_complete, ); // Transport: InMemoryBus → a real broker. The handlers and the // `with_bus(..).run(..)` wiring are unchanged; only this constructor line differs. -// let bus = NatsBus::connect("nats://localhost:4222", "todos", "app").await?; -// let bus = PostgresBus::new(pool, "todos"); -// let bus = RabbitBus::connect("amqp://localhost:5672/%2f", "todos", "app").await?; -// let bus = KafkaBus::connect("localhost:9092", "todos", "app").await?; +let namespace = "todos-prod"; // broker namespace/prefix for this app/environment +// let bus = NatsBus::connect("nats://localhost:4222").namespace(namespace).await?; +// let bus = PostgresBus::new(pool); +// let bus = RabbitBus::connect("amqp://localhost:5672/%2f").namespace(namespace).await?; +// let bus = KafkaBus::connect("localhost:9092").namespace(namespace).await?; service.with_bus(bus).run(RunOptions::idempotent()).await?; ``` +`group` and `namespace` are broker topology names, not the command/event names +your service handles. `register_handlers!` gives the service its command/event +names; `with_bus(bus).run(..)` reads those names through `subscription_plan()` and +passes them to the transport. + +- `Service::named("todo-api")` supplies the default durable consumer `group`. + Use the same service name for every replica of one deployment. For direct + `bus.listen(..)` / `bus.subscribe(..)` consumers that are not a `Service`, set + the group with `bus.group("todo-projections")`. +- `namespace` scopes streams, subjects, topics, queues, or exchanges on a shared + broker. `PostgresBus` does not take `namespace` because the database/schema + behind `pool` already scopes its bus tables. +- Topology names are validated before broker use. Keep groups/service names to + portable deployment IDs (`A-Z`, `a-z`, `0-9`, `_`, `-`); namespaces may also + use `.`. Blank names, whitespace, control characters, path separators, broker + wildcards, and names longer than 128 bytes are rejected. + | Concern | In-memory default | Swap in for production | |---|---|---| | Storage | `HashMapRepository` | `PostgresRepository`, `SqliteRepository` | @@ -787,8 +807,9 @@ transports; only the constructor line changes.** use std::sync::Arc; use distributed::bus::{Bus, BusConsumer, InMemoryBus, RunOptions}; -// Built once — handlers are transport-agnostic. -let service = Arc::new(build_service()); +// Built once — handlers are transport-agnostic. The service name becomes the +// default durable consumer group for broker-backed buses. +let service = Arc::new(build_service().named("order-api")); // Dev/test: in-memory. let bus = InMemoryBus::new(); @@ -798,11 +819,12 @@ bus.listen(service.clone(), RunOptions::idempotent()).await?; // competing bus.subscribe(service.clone(), RunOptions::idempotent()).await?; // fan-out // Production: swap the one constructor line — send/listen/publish/subscribe -// and the handlers are unchanged. -// let bus = NatsBus::connect("nats://localhost:4222", "orders", "app").await?; -// let bus = PostgresBus::new(pool, "orders"); -// let bus = RabbitBus::connect("amqp://localhost:5672/%2f", "orders", "app").await?; -// let bus = KafkaBus::connect("localhost:9092", "orders", "app").await?; +// and the handlers are unchanged. A named Service supplies the consumer group. +let namespace = "orders-prod"; +// let bus = NatsBus::connect("nats://localhost:4222").namespace(namespace).await?; +// let bus = PostgresBus::new(pool); +// let bus = RabbitBus::connect("amqp://localhost:5672/%2f").namespace(namespace).await?; +// let bus = KafkaBus::connect("localhost:9092").namespace(namespace).await?; ``` This is the low-level facade. For a `microsvc::Service`, the one-call convenience @@ -811,9 +833,20 @@ and the event names to `subscribe` from the registered handlers, and makes `repo.outbox(msg).commit(agg)` publish on commit. Drop to `listen` / `subscribe` / `send` / `publish` directly when you need finer control. -Point-to-point vs fan-out is consistently a **consumer-group/identity** choice in -each transport's native topology — the same `group` competes, different `group`s -fan out: +Consumer identity controls the durable broker state in each transport. Command +handlers should normally be owned by one service deployment, with every replica +using the same `group` so the deployment competes as one logical consumer. Event +handlers use distinct `group`s when each service needs its own copy. + +The `group` is not a list of handler names. Handler names come from +`subscription_plan()`; `group` tells the broker which durable consumer, offset, or +queue belongs to this running service. `Service::named(..)` supplies that group +for `service.with_bus(bus).run(..)`; direct `Handlers` or manual +`listen`/`subscribe` calls can set it with `bus.group(..)` or `Handlers::named(..)`. +Groups/service names should use portable deployment IDs (`A-Z`, `a-z`, `0-9`, +`_`, `-`); namespaces may also include `.`. Blank names, whitespace, control +characters, path separators, broker wildcards, and names longer than 128 bytes +are rejected before broker topology is created. | `*Bus` | Feature | `send` / `listen` (competing) | `publish` / `subscribe` (fan-out) | |---|---|---|---| diff --git a/docs/async-transports.md b/docs/async-transports.md index 85ea00cf..10b8bea7 100644 --- a/docs/async-transports.md +++ b/docs/async-transports.md @@ -110,8 +110,9 @@ The app surface is identical across transports; only the constructor changes: use std::sync::Arc; use distributed::bus::{Bus, BusConsumer, InMemoryBus, RunOptions}; -// Built once — handlers are transport-agnostic. -let service = Arc::new(build_service()); +// Built once — handlers are transport-agnostic. The service name becomes the +// default durable consumer group for broker-backed buses. +let service = Arc::new(build_service().named("order-api")); // Dev/test: in-memory. let bus = InMemoryBus::new(); @@ -121,16 +122,33 @@ bus.listen(service.clone(), RunOptions::idempotent()).await?; // competing bus.subscribe(service.clone(), RunOptions::idempotent()).await?; // fan-out // Production: swap the one constructor line — send/listen/publish/subscribe -// and the handlers are unchanged. -// let bus = NatsBus::connect("nats://localhost:4222", "orders", "app").await?; -// let bus = PostgresBus::new(pool, "orders"); -// let bus = RabbitBus::connect("amqp://localhost:5672/%2f", "orders", "app").await?; -// let bus = KafkaBus::connect("localhost:9092", "orders", "app").await?; +// and the handlers are unchanged. A named Service supplies the consumer group. +let namespace = "orders-prod"; +// let bus = NatsBus::connect("nats://localhost:4222").namespace(namespace).await?; +// let bus = PostgresBus::new(pool); +// let bus = RabbitBus::connect("amqp://localhost:5672/%2f").namespace(namespace).await?; +// let bus = KafkaBus::connect("localhost:9092").namespace(namespace).await?; ``` -Point-to-point vs fan-out is consistently a **consumer-group/identity** choice in -each transport's native topology — same `group` competes, different `group`s -fan out: +`group` and `namespace` are broker topology names, not the command/event names +your service handles. Handler names come from the service's `subscription_plan()`. +`Service::named(..)` supplies the default durable consumer `group`: all replicas +of one service deployment use the same value; independent event consumers use +different values so each gets its own event copy. Direct `Handlers` or manual +`listen`/`subscribe` calls can set the group with `bus.group(..)` or +`Handlers::named(..)`. `namespace` scopes streams, subjects, topics, queues, or +exchanges on a shared broker. `PostgresBus` does not take `namespace` because the +database/schema behind `pool` already scopes its bus tables. + +Topology names are validated before broker use. Keep groups/service names to +portable deployment IDs (`A-Z`, `a-z`, `0-9`, `_`, `-`); namespaces may also use +`.`. Blank names, whitespace, control characters, path separators, broker +wildcards, and names longer than 128 bytes are rejected. + +Consumer identity controls the durable broker state in each transport. Command +handlers should normally be owned by one service deployment, with every replica +using the same `group` so the deployment competes as one logical consumer. Event +handlers use distinct `group`s when each service needs its own copy: | `*Bus` | Feature | `send` / `listen` (competing) | `publish` / `subscribe` (fan-out) | | --- | --- | --- | --- | diff --git a/src/bus/bus.rs b/src/bus/bus.rs index cc1ddfd3..49d82f6f 100644 --- a/src/bus/bus.rs +++ b/src/bus/bus.rs @@ -54,8 +54,11 @@ pub trait Bus: Send + Sync { /// Consume side of the bus — pull transports that run a [`run_source`] loop. /// /// `listen`/`subscribe` derive the message names from the router's registered -/// handlers ([`MessageRouter::subscription_plan`]), build the transport's source -/// with the matching topology, and run it. Both run until the source drains/stops. +/// handlers ([`MessageRouter::subscription_plan`]) and use +/// [`MessageRouter::consumer_group`] as the default durable consumer identity +/// when the bus was not configured with an explicit group. They then build the +/// transport's source with the matching topology and run it. Both run until the +/// source drains/stops. /// /// [`run_source`]: super::run_source pub trait BusConsumer: Send + Sync { diff --git a/src/bus/handlers.rs b/src/bus/handlers.rs index 62e516e1..fe1bd89a 100644 --- a/src/bus/handlers.rs +++ b/src/bus/handlers.rs @@ -67,6 +67,7 @@ where /// [`on_event`](Handlers::on_event), then run it with `bus.listen`/`bus.subscribe`. #[derive(Clone, Default)] pub struct Handlers { + group: Option, handlers: HashMap<(MessageKind, String), Arc>, } @@ -76,6 +77,14 @@ impl Handlers { Self::default() } + /// Assign a stable consumer identity for broker adapters that need a durable + /// group when this standalone registry is used directly with `listen` or + /// `subscribe`. + pub fn named(mut self, group: impl Into) -> Self { + self.group = Some(group.into()); + self + } + /// Register a command handler (point-to-point / competing-consumer via `listen`). pub fn on_command(self, name: impl Into, handler: F) -> Self where @@ -103,6 +112,10 @@ impl Handlers { } impl MessageRouter for Handlers { + fn consumer_group(&self) -> Option<&str> { + self.group.as_deref() + } + fn handles(&self, kind: MessageKind, name: &str) -> bool { self.handlers.contains_key(&(kind, name.to_string())) } diff --git a/src/bus/kafka_bus.rs b/src/bus/kafka_bus.rs index e450fa70..c98b54c0 100644 --- a/src/bus/kafka_bus.rs +++ b/src/bus/kafka_bus.rs @@ -18,12 +18,15 @@ //! //! Requires the `kafka` feature. Integration-tested in `tests/kafka_transport`. +use std::future::{Future, IntoFuture}; +use std::pin::Pin; use std::sync::Arc; use std::time::Duration; use super::kafka::{KafkaPublisher, KafkaSource}; use super::{ - run_source, AsyncMessagePublisher, Bus, BusConsumer, MessageRouter, RunOptions, TransportError, + run_source, AsyncMessagePublisher, Bus, BusConsumer, BusTopologyConfig, MessageRouter, + RunOptions, TransportError, }; use super::{Message, MessageKind}; @@ -34,28 +37,99 @@ const DEFAULT_FETCH_TIMEOUT: Duration = Duration::from_secs(8); pub struct KafkaBus { brokers: String, publisher: Arc, - group: String, - namespace: String, + topology: BusTopologyConfig, fetch_timeout: Duration, } +/// Awaitable builder returned by [`KafkaBus::connect`]. +pub struct KafkaBusConnect { + brokers: String, + topology: BusTopologyConfig, + fetch_timeout: Duration, +} + +impl KafkaBusConnect { + /// Set an explicit Kafka consumer group. Service consumers can usually omit + /// this and use [`Service::named`](crate::microsvc::Service::named) instead. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + /// Set the topic/group-id namespace used on the shared Kafka cluster. + pub fn namespace(mut self, namespace: impl Into) -> Self { + self.topology = self.topology.namespace(namespace); + self + } + + /// Override how long a `listen`/`subscribe` poll waits before idling. Kafka + /// group bootstrap/rebalance takes time, so this is generous by default. + pub fn with_fetch_timeout(mut self, timeout: Duration) -> Self { + self.fetch_timeout = timeout; + self + } + + async fn connect(self) -> Result { + let topology = self.topology.validate_for("kafka")?; + let publisher = KafkaPublisher::connect(&self.brokers).await?; + Ok(KafkaBus { + brokers: self.brokers, + publisher: Arc::new(publisher), + topology, + fetch_timeout: self.fetch_timeout, + }) + } +} + +impl IntoFuture for KafkaBusConnect { + type Output = Result; + type IntoFuture = Pin + Send>>; + + fn into_future(self) -> Self::IntoFuture { + Box::pin(self.connect()) + } +} + impl KafkaBus { - /// Connect a producer to `brokers` and build a bus. `group` is the consumer - /// identity (same group ⇒ competing; different groups ⇒ fan-out); `namespace` - /// scopes topics and group ids. - pub async fn connect( + /// Start building a bus connected to Kafka brokers. + /// + /// The returned builder is awaitable: + /// + /// ```ignore + /// let bus = KafkaBus::connect("localhost:9092") + /// .namespace("todos-prod") + /// .await?; + /// ``` + pub fn connect(brokers: &str) -> KafkaBusConnect { + KafkaBusConnect { + brokers: brokers.to_string(), + topology: BusTopologyConfig::default(), + fetch_timeout: DEFAULT_FETCH_TIMEOUT, + } + } + + /// Connect with an explicit group and namespace for direct/low-level use. + pub async fn connect_with( brokers: &str, group: impl Into, namespace: impl Into, ) -> Result { - let publisher = KafkaPublisher::connect(brokers).await?; - Ok(Self { - brokers: brokers.to_string(), - publisher: Arc::new(publisher), - group: group.into(), - namespace: namespace.into(), - fetch_timeout: DEFAULT_FETCH_TIMEOUT, - }) + Self::connect(brokers) + .group(group) + .namespace(namespace) + .await + } + + /// Set an explicit Kafka consumer group on an already-built bus. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + /// Set the topic/group-id namespace used on the shared Kafka cluster. + pub fn namespace(mut self, namespace: impl Into) -> Self { + self.topology = self.topology.namespace(namespace); + self } /// Override how long a `listen`/`subscribe` poll waits before idling. Kafka @@ -65,12 +139,16 @@ impl KafkaBus { self } - fn command_prefix(&self) -> String { - format!("{}.cmd.", self.namespace) + fn validated_namespace(&self) -> Result { + self.topology.namespace_for("kafka") } - fn event_prefix(&self) -> String { - format!("{}.evt.", self.namespace) + fn command_prefix(&self) -> Result { + Ok(format!("{}.cmd.", self.validated_namespace()?)) + } + + fn event_prefix(&self) -> Result { + Ok(format!("{}.evt.", self.validated_namespace()?)) } async fn run( @@ -106,12 +184,12 @@ impl Bus for KafkaBus { async fn send_message(&self, mut message: Message) -> Result<(), TransportError> { // The publisher uses the message name as the topic; namespace it. - message.name = format!("{}{}", self.command_prefix(), message.name); + message.name = format!("{}{}", self.command_prefix()?, message.name); self.publisher.publish(message).await } async fn publish_message(&self, mut message: Message) -> Result<(), TransportError> { - message.name = format!("{}{}", self.event_prefix(), message.name); + message.name = format!("{}{}", self.event_prefix()?, message.name); self.publisher.publish(message).await } } @@ -122,14 +200,21 @@ impl BusConsumer for KafkaBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - let prefix = self.command_prefix(); - let topics: Vec = router - .subscription_plan() + let plan = router.subscription_plan(); + if plan.commands.is_empty() { + return Ok(()); + } + let prefix = self.command_prefix()?; + let topics: Vec = plan .commands .iter() .map(|name| format!("{prefix}{name}")) .collect(); - let group_id = format!("{}.{}.cmd", self.namespace, self.group); + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "kafka")?; + let namespace = self.validated_namespace()?; + let group_id = format!("{namespace}.{group}.cmd"); self.run(router, topics, group_id, prefix, options).await } @@ -138,14 +223,74 @@ impl BusConsumer for KafkaBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - let prefix = self.event_prefix(); - let topics: Vec = router - .subscription_plan() + let plan = router.subscription_plan(); + if plan.events.is_empty() { + return Ok(()); + } + let prefix = self.event_prefix()?; + let topics: Vec = plan .events .iter() .map(|name| format!("{prefix}{name}")) .collect(); - let group_id = format!("{}.{}.evt", self.namespace, self.group); + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "kafka")?; + let namespace = self.validated_namespace()?; + let group_id = format!("{namespace}.{group}.evt"); self.run(router, topics, group_id, prefix, options).await } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::bus::SubscriptionPlan; + use rdkafka::config::ClientConfig; + use rdkafka::producer::FutureProducer; + + struct EmptyRouter; + + impl MessageRouter for EmptyRouter { + fn handles(&self, _kind: MessageKind, _name: &str) -> bool { + false + } + + fn subscription_plan(&self) -> SubscriptionPlan { + SubscriptionPlan::default() + } + + async fn dispatch(&self, _message: &Message) -> Result<(), TransportError> { + Ok(()) + } + } + + fn test_bus() -> KafkaBus { + let producer: FutureProducer = ClientConfig::new() + .set("bootstrap.servers", "localhost:1") + .create() + .unwrap(); + KafkaBus { + brokers: "localhost:1".to_string(), + publisher: Arc::new(KafkaPublisher::new(producer)), + topology: BusTopologyConfig::default(), + fetch_timeout: Duration::from_millis(1), + } + } + + #[tokio::test] + async fn listen_returns_ok_for_empty_plan_without_group() { + let bus = test_bus(); + let router = Arc::new(EmptyRouter); + bus.listen(router, RunOptions::idempotent()).await.unwrap(); + } + + #[tokio::test] + async fn subscribe_returns_ok_for_empty_plan_without_group() { + let bus = test_bus(); + let router = Arc::new(EmptyRouter); + bus.subscribe(router, RunOptions::idempotent()) + .await + .unwrap(); + } +} diff --git a/src/bus/mod.rs b/src/bus/mod.rs index 48c66a34..70e4fc05 100644 --- a/src/bus/mod.rs +++ b/src/bus/mod.rs @@ -112,11 +112,12 @@ mod run_options; mod runner; mod source; mod stable_id; +mod topology; #[cfg(feature = "kafka")] pub use kafka::{KafkaPublisher, KafkaReceived, KafkaSource}; #[cfg(feature = "kafka")] -pub use kafka_bus::KafkaBus; +pub use kafka_bus::{KafkaBus, KafkaBusConnect}; #[cfg(feature = "http")] pub use knative::knative_triggers; #[cfg(feature = "http")] @@ -124,9 +125,9 @@ pub use knative_bus::KnativeBus; #[cfg(feature = "nats")] pub use nats::{NatsJetStreamSource, NatsPublisher, NatsReceived}; #[cfg(feature = "nats")] -pub use nats_bus::NatsBus; +pub use nats_bus::{NatsBus, NatsBusConnect}; #[cfg(feature = "rabbitmq")] -pub use rabbit_bus::RabbitBus; +pub use rabbit_bus::{RabbitBus, RabbitBusConnect}; #[cfg(feature = "rabbitmq")] pub use rabbitmq::{RabbitPublisher, RabbitReceived, RabbitSource}; @@ -145,3 +146,4 @@ pub use run_options::{ConsumerDeliveryMode, InboxHook, NoInbox, RunOptions}; pub use runner::run_source; pub use source::{AsyncMessageSource, ReceivedMessage}; pub use stable_id::{validate_stable_message_id, StableMessageIdError, MAX_STABLE_MESSAGE_ID_LEN}; +pub(crate) use topology::BusTopologyConfig; diff --git a/src/bus/nats_bus.rs b/src/bus/nats_bus.rs index 244e4de1..8c63b7e2 100644 --- a/src/bus/nats_bus.rs +++ b/src/bus/nats_bus.rs @@ -18,6 +18,8 @@ //! //! Requires the `nats` feature. Integration-tested in `tests/nats_transport`. +use std::future::{Future, IntoFuture}; +use std::pin::Pin; use std::sync::Arc; use std::time::Duration; @@ -27,7 +29,8 @@ use async_nats::jetstream::stream::{Config as StreamConfig, Stream}; use super::nats::{NatsJetStreamSource, NatsPublisher}; use super::{ - run_source, AsyncMessagePublisher, Bus, BusConsumer, MessageRouter, RunOptions, TransportError, + run_source, AsyncMessagePublisher, Bus, BusConsumer, BusTopologyConfig, MessageRouter, + RunOptions, TransportError, }; use super::{Message, MessageKind}; @@ -43,24 +46,61 @@ pub struct NatsBus { jetstream: jetstream::Context, cmd_publisher: Arc, evt_publisher: Arc, - group: String, - namespace: String, - stream_name: String, + topology: BusTopologyConfig, fetch_timeout: Duration, } +/// Awaitable builder returned by [`NatsBus::connect`]. +pub struct NatsBusConnect { + url: String, + topology: BusTopologyConfig, + fetch_timeout: Duration, +} + +impl NatsBusConnect { + /// Set an explicit durable consumer group. Service consumers can usually omit + /// this and use [`Service::named`](crate::microsvc::Service::named) instead. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + /// Set the subject/stream namespace used on the shared NATS server. + pub fn namespace(mut self, namespace: impl Into) -> Self { + self.topology = self.topology.namespace(namespace); + self + } + + /// Override how long a `listen`/`subscribe` poll waits before idling. + pub fn with_fetch_timeout(mut self, timeout: Duration) -> Self { + self.fetch_timeout = timeout; + self + } + + async fn connect(self) -> Result { + let topology = self.topology.validate_for("nats")?; + let client = async_nats::connect(&self.url) + .await + .map_err(|err| retryable("nats connect", err))?; + Ok(NatsBus::new(jetstream::new(client)) + .with_topology(topology) + .with_fetch_timeout(self.fetch_timeout)) + } +} + +impl IntoFuture for NatsBusConnect { + type Output = Result; + type IntoFuture = Pin + Send>>; + + fn into_future(self) -> Self::IntoFuture { + Box::pin(self.connect()) + } +} + impl NatsBus { /// Build a bus over an existing JetStream context. - /// - /// `group` is the logical consumer identity (same group ⇒ competing - /// consumers; different groups ⇒ fan-out). `namespace` scopes the stream and - /// subjects so multiple buses can share a server without collision. - pub fn new( - jetstream: jetstream::Context, - group: impl Into, - namespace: impl Into, - ) -> Self { - let namespace = namespace.into(); + pub fn new(jetstream: jetstream::Context) -> Self { + let namespace = BusTopologyConfig::default_namespace(); let cmd_publisher = NatsPublisher::new(jetstream.clone()).with_subject_prefix(format!("{namespace}.cmd")); let evt_publisher = @@ -69,23 +109,55 @@ impl NatsBus { jetstream, cmd_publisher: Arc::new(cmd_publisher), evt_publisher: Arc::new(evt_publisher), - group: group.into(), - stream_name: namespace.to_uppercase().replace(['.', '-'], "_"), - namespace, + topology: BusTopologyConfig::default(), + fetch_timeout: DEFAULT_FETCH_TIMEOUT, + } + } + + /// Start building a bus connected to a NATS server URL. + /// + /// The returned builder is awaitable: + /// + /// ```ignore + /// let bus = NatsBus::connect("nats://localhost:4222") + /// .namespace("todos-prod") + /// .await?; + /// ``` + pub fn connect(url: &str) -> NatsBusConnect { + NatsBusConnect { + url: url.to_string(), + topology: BusTopologyConfig::default(), fetch_timeout: DEFAULT_FETCH_TIMEOUT, } } - /// Connect to a NATS server URL and build a bus. - pub async fn connect( + /// Connect with an explicit group and namespace for direct/low-level use. + pub async fn connect_with( url: &str, group: impl Into, namespace: impl Into, ) -> Result { - let client = async_nats::connect(url) - .await - .map_err(|err| retryable("nats connect", err))?; - Ok(Self::new(jetstream::new(client), group, namespace)) + Self::connect(url).group(group).namespace(namespace).await + } + + /// Set an explicit durable consumer group on an already-built bus. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + fn with_topology(mut self, topology: BusTopologyConfig) -> Self { + self.update_publishers(topology.namespace_unchecked()); + self.topology = topology; + self + } + + /// Set the subject/stream namespace used on the shared NATS server. + pub fn namespace(mut self, namespace: impl Into) -> Self { + let namespace = namespace.into(); + self.update_publishers(&namespace); + self.topology = self.topology.namespace(namespace); + self } /// Override how long a `listen`/`subscribe` poll waits before idling. @@ -96,8 +168,8 @@ impl NatsBus { /// Sanitize the group into a valid NATS consumer-name token. Consumer names /// cannot contain `.`, `*`, `>`, or whitespace, so map them to `_`. - fn durable_base(&self) -> String { - self.group + fn durable_base(group: &str) -> String { + group .chars() .map(|c| match c { '.' | '*' | '>' | ' ' | '\t' | '\n' | '/' | '\\' => '_', @@ -106,15 +178,35 @@ impl NatsBus { .collect() } + fn validated_namespace(&self) -> Result { + self.topology.namespace_for("nats") + } + + fn update_publishers(&mut self, namespace: &str) { + self.cmd_publisher = Arc::new( + NatsPublisher::new(self.jetstream.clone()) + .with_subject_prefix(format!("{namespace}.cmd")), + ); + self.evt_publisher = Arc::new( + NatsPublisher::new(self.jetstream.clone()) + .with_subject_prefix(format!("{namespace}.evt")), + ); + } + + fn stream_name(namespace: &str) -> String { + namespace.to_uppercase().replace(['.', '-'], "_") + } + /// Create-or-open the backing stream (`{namespace}.>`). Called by /// `listen`/`subscribe`; producers should ensure it exists (here, via IaC, or /// by a consumer) before publishing, since JetStream rejects a publish to an /// unbound subject. pub async fn ensure_stream(&self) -> Result { + let namespace = self.validated_namespace()?; self.jetstream .get_or_create_stream(StreamConfig { - name: self.stream_name.clone(), - subjects: vec![format!("{}.>", self.namespace)], + name: Self::stream_name(&namespace), + subjects: vec![format!("{namespace}.>")], ..Default::default() }) .await @@ -159,10 +251,12 @@ impl Bus for NatsBus { } async fn send_message(&self, message: Message) -> Result<(), TransportError> { + self.validated_namespace()?; self.cmd_publisher.publish(message).await } async fn publish_message(&self, message: Message) -> Result<(), TransportError> { + self.validated_namespace()?; self.evt_publisher.publish(message).await } } @@ -173,20 +267,24 @@ impl BusConsumer for NatsBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - let subjects: Vec = router - .subscription_plan() + let plan = router.subscription_plan(); + if plan.commands.is_empty() { + return Ok(()); + } + let namespace = self.validated_namespace()?; + let subjects: Vec = plan .commands .iter() - .map(|name| format!("{}.cmd.{name}", self.namespace)) + .map(|name| format!("{namespace}.cmd.{name}")) .collect(); - if subjects.is_empty() { - return Ok(()); - } + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "nats")?; let source = self .source( - &format!("{}_cmd", self.durable_base()), + &format!("{}_cmd", Self::durable_base(&group)), subjects, - format!("{}.cmd.", self.namespace), + format!("{namespace}.cmd."), ) .await?; run_source(router, source, options).await @@ -197,20 +295,24 @@ impl BusConsumer for NatsBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - let subjects: Vec = router - .subscription_plan() + let plan = router.subscription_plan(); + if plan.events.is_empty() { + return Ok(()); + } + let namespace = self.validated_namespace()?; + let subjects: Vec = plan .events .iter() - .map(|name| format!("{}.evt.{name}", self.namespace)) + .map(|name| format!("{namespace}.evt.{name}")) .collect(); - if subjects.is_empty() { - return Ok(()); - } + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "nats")?; let source = self .source( - &format!("{}_evt", self.durable_base()), + &format!("{}_evt", Self::durable_base(&group)), subjects, - format!("{}.evt.", self.namespace), + format!("{namespace}.evt."), ) .await?; run_source(router, source, options).await diff --git a/src/bus/postgres_bus.rs b/src/bus/postgres_bus.rs index 44ae658e..7cfff558 100644 --- a/src/bus/postgres_bus.rs +++ b/src/bus/postgres_bus.rs @@ -37,7 +37,9 @@ use std::time::Duration; use sqlx::{PgPool, Row}; use super::source::{AsyncMessageSource, ReceivedMessage}; -use super::{run_source, Bus, BusConsumer, MessageRouter, RunOptions, TransportError}; +use super::{ + run_source, Bus, BusConsumer, BusTopologyConfig, MessageRouter, RunOptions, TransportError, +}; use super::{Message, MessageKind}; const DEFAULT_LEASE: Duration = Duration::from_secs(30); @@ -113,22 +115,38 @@ fn message_from_row(row: &sqlx::postgres::PgRow) -> Message { #[derive(Clone)] pub struct PostgresBus { pool: PgPool, - group: String, + topology: BusTopologyConfig, lease: Duration, } impl PostgresBus { - /// Build a bus over an existing pool. `group` is the consumer identity: - /// replicas sharing a `group` compete on the queue (point-to-point) and share - /// one log offset; distinct `group`s each get their own log offset (fan-out). - pub fn new(pool: PgPool, group: impl Into) -> Self { + /// Build a bus over an existing pool. + /// + /// For event subscriptions, `subscribe` uses the router's consumer identity as + /// the durable Postgres log offset. Service consumers usually get that identity + /// from [`Service::named`](crate::microsvc::Service::named). Direct consumers + /// can set it with [`group`](Self::group). Commands are claimed from + /// `bus_queue` by message name, so command replicas compete by listening to the + /// same registered command names. + pub fn new(pool: PgPool) -> Self { Self { pool, - group: group.into(), + topology: BusTopologyConfig::default(), lease: DEFAULT_LEASE, } } + /// Build a bus with an explicit group for direct/low-level use. + pub fn new_with_group(pool: PgPool, group: impl Into) -> Self { + Self::new(pool).group(group) + } + + /// Set an explicit durable event subscription group. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + /// Override the claim lease for `listen` (how long a claimed command stays /// invisible to other workers before it is eligible for redelivery). pub fn with_lease(mut self, lease: Duration) -> Self { @@ -239,10 +257,13 @@ impl BusConsumer for PostgresBus { if names.is_empty() { return Ok(()); } + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "postgres")?; let source = LogSource { pool: self.pool.clone(), names, - consumer: self.group.clone(), + consumer: group, }; run_source(router, source, options).await } diff --git a/src/bus/rabbit_bus.rs b/src/bus/rabbit_bus.rs index 83630024..aa98ffe1 100644 --- a/src/bus/rabbit_bus.rs +++ b/src/bus/rabbit_bus.rs @@ -15,11 +15,15 @@ //! queues, so every group receives every event — fan-out (replicas within a //! group still compete on that group's queue). //! -//! The `group` is the logical consumer identity. `{ns}` (namespace) scopes queue -//! and exchange names so multiple apps can share a broker. +//! The `group` is the logical consumer identity for event subscriptions. Command +//! queues are keyed by command name; replicas compete by consuming the same command +//! queues. `{ns}` (namespace) scopes queue and exchange names so multiple apps can +//! share a broker. //! //! Requires the `rabbitmq` feature. Integration-tested in `tests/rabbitmq_transport`. +use std::future::{Future, IntoFuture}; +use std::pin::Pin; use std::sync::Arc; use lapin::options::{ @@ -31,7 +35,9 @@ use lapin::{Channel, ExchangeKind}; use super::rabbitmq::{connect_channel, message_properties, RabbitReceived}; use super::source::AsyncMessageSource; -use super::{run_source, Bus, BusConsumer, MessageRouter, RunOptions, TransportError}; +use super::{ + run_source, Bus, BusConsumer, BusTopologyConfig, MessageRouter, RunOptions, TransportError, +}; use super::{Message, MessageKind}; fn retryable(context: &str, err: impl std::fmt::Display) -> TransportError { @@ -42,45 +48,117 @@ fn retryable(context: &str, err: impl std::fmt::Display) -> TransportError { pub struct RabbitBus { uri: String, channel: Channel, - group: String, - namespace: String, - events_exchange: String, + topology: BusTopologyConfig, +} + +/// Awaitable builder returned by [`RabbitBus::connect`]. +pub struct RabbitBusConnect { + uri: String, + topology: BusTopologyConfig, +} + +impl RabbitBusConnect { + /// Set an explicit event subscription group. Service consumers can usually + /// omit this and use [`Service::named`](crate::microsvc::Service::named) + /// instead. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self + } + + /// Set the queue/exchange namespace used on the shared RabbitMQ broker. + pub fn namespace(mut self, namespace: impl Into) -> Self { + self.topology = self.topology.namespace(namespace); + self + } + + async fn connect(self) -> Result { + RabbitBus::connect_configured(self.uri, self.topology).await + } +} + +impl IntoFuture for RabbitBusConnect { + type Output = Result; + type IntoFuture = Pin + Send>>; + + fn into_future(self) -> Self::IntoFuture { + Box::pin(self.connect()) + } } impl RabbitBus { - /// Connect to an AMQP URI and build a bus. `group` is the consumer identity - /// (same group ⇒ competing; different groups ⇒ fan-out); `namespace` scopes - /// queue/exchange names. - pub async fn connect( + /// Start building a bus connected to an AMQP URI. + /// + /// The returned builder is awaitable: + /// + /// ```ignore + /// let bus = RabbitBus::connect("amqp://localhost:5672/%2f") + /// .namespace("todos-prod") + /// .await?; + /// ``` + pub fn connect(uri: &str) -> RabbitBusConnect { + RabbitBusConnect { + uri: uri.to_string(), + topology: BusTopologyConfig::default(), + } + } + + /// Connect with an explicit group and namespace for direct/low-level use. + pub async fn connect_with( uri: &str, group: impl Into, namespace: impl Into, ) -> Result { - let channel = connect_channel(uri).await?; + Self::connect(uri).group(group).namespace(namespace).await + } + + async fn connect_configured( + uri: String, + topology: BusTopologyConfig, + ) -> Result { + let topology = topology.validate_for("rabbitmq")?; + let channel = connect_channel(&uri).await?; channel .confirm_select(ConfirmSelectOptions::default()) .await .map_err(|err| retryable("amqp confirm_select", err))?; - let namespace = namespace.into(); Ok(Self { - uri: uri.to_string(), + uri, channel, - group: group.into(), - events_exchange: format!("{namespace}.events"), - namespace, + topology, }) } - fn command_queue(&self, name: &str) -> String { - format!("{}.cmd.{name}", self.namespace) + /// Set an explicit event subscription group on an already-built bus. + pub fn group(mut self, group: impl Into) -> Self { + self.topology = self.topology.group(group); + self } - fn command_prefix(&self) -> String { - format!("{}.cmd.", self.namespace) + /// Set the queue/exchange namespace used on the shared RabbitMQ broker. + pub fn namespace(mut self, namespace: impl Into) -> Self { + self.topology = self.topology.namespace(namespace); + self } - fn group_queue(&self) -> String { - format!("{}.evt.{}", self.namespace, self.group) + fn validated_namespace(&self) -> Result { + self.topology.namespace_for("rabbitmq") + } + + fn command_queue(&self, name: &str) -> Result { + Ok(format!("{}.cmd.{name}", self.validated_namespace()?)) + } + + fn command_prefix(&self) -> Result { + Ok(format!("{}.cmd.", self.validated_namespace()?)) + } + + fn events_exchange(&self) -> Result { + Ok(format!("{}.events", self.validated_namespace()?)) + } + + fn group_queue(&self, group: &str) -> Result { + Ok(format!("{}.evt.{group}", self.validated_namespace()?)) } async fn declare_queue(&self, channel: &Channel, queue: &str) -> Result<(), TransportError> { @@ -98,10 +176,14 @@ impl RabbitBus { Ok(()) } - async fn declare_events_exchange(&self, channel: &Channel) -> Result<(), TransportError> { + async fn declare_events_exchange( + &self, + channel: &Channel, + exchange: &str, + ) -> Result<(), TransportError> { channel .exchange_declare( - ShortString::from(self.events_exchange.as_str()), + ShortString::from(exchange), ExchangeKind::Topic, ExchangeDeclareOptions { durable: true, @@ -149,15 +231,21 @@ impl RabbitBus { &self, router: &R, ) -> Result<(), TransportError> { - self.declare_events_exchange(&self.channel).await?; - let queue = self.group_queue(); - self.declare_queue(&self.channel, &queue).await?; let plan = router.subscription_plan(); + if plan.events.is_empty() { + return Ok(()); + } + let group = self.topology.resolve_consumer_group(router, "rabbitmq")?; + let exchange = self.events_exchange()?; + self.declare_events_exchange(&self.channel, &exchange) + .await?; + let queue = self.group_queue(&group)?; + self.declare_queue(&self.channel, &queue).await?; for name in &plan.events { self.channel .queue_bind( ShortString::from(queue.as_str()), - ShortString::from(self.events_exchange.as_str()), + ShortString::from(exchange.as_str()), ShortString::from(name.as_str()), QueueBindOptions::default(), FieldTable::default(), @@ -183,16 +271,18 @@ impl Bus for RabbitBus { async fn send_message(&self, mut message: Message) -> Result<(), TransportError> { // Default exchange routes by routing key == queue name; declare the queue // so the command is retained until a listener consumes it. - let queue = self.command_queue(message.name()); + let queue = self.command_queue(message.name())?; self.declare_queue(&self.channel, &queue).await?; message.name = queue.clone(); self.publish_confirmed("", &queue, &message).await } async fn publish_message(&self, message: Message) -> Result<(), TransportError> { - self.declare_events_exchange(&self.channel).await?; + let exchange = self.events_exchange()?; + self.declare_events_exchange(&self.channel, &exchange) + .await?; let routing_key = message.name().to_string(); - self.publish_confirmed(&self.events_exchange, &routing_key, &message) + self.publish_confirmed(&exchange, &routing_key, &message) .await } } @@ -203,21 +293,21 @@ impl BusConsumer for RabbitBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - let channel = connect_channel(&self.uri).await?; let plan = router.subscription_plan(); + if plan.commands.is_empty() { + return Ok(()); + } + let channel = connect_channel(&self.uri).await?; let mut queues = Vec::new(); for name in &plan.commands { - let queue = self.command_queue(name); + let queue = self.command_queue(name)?; self.declare_queue(&channel, &queue).await?; queues.push(queue); } - if queues.is_empty() { - return Ok(()); - } let source = RabbitBusSource { channel, queues, - strip_prefix: Some(self.command_prefix()), + strip_prefix: Some(self.command_prefix()?), }; run_source(router, source, options).await } @@ -227,14 +317,17 @@ impl BusConsumer for RabbitBus { router: Arc, options: RunOptions, ) -> Result<(), TransportError> { - self.ensure_subscription(router.as_ref()).await?; if router.subscription_plan().events.is_empty() { return Ok(()); } + self.ensure_subscription(router.as_ref()).await?; + let group = self + .topology + .resolve_consumer_group(router.as_ref(), "rabbitmq")?; let channel = connect_channel(&self.uri).await?; let source = RabbitBusSource { channel, - queues: vec![self.group_queue()], + queues: vec![self.group_queue(&group)?], // Events are published with routing key == the bare event name. strip_prefix: None, }; diff --git a/src/bus/router.rs b/src/bus/router.rs index 4bfec3b1..c0a831b2 100644 --- a/src/bus/router.rs +++ b/src/bus/router.rs @@ -28,6 +28,12 @@ use super::{Message, MessageKind, SubscriptionPlan}; /// [`BusConsumer`](super::BusConsumer): consume it as `Arc` or a generic /// ``, never `Arc`. pub trait MessageRouter: Send + Sync { + /// Stable identity for this consumer, used by broker adapters as the default + /// durable consumer group when the bus itself was not configured with one. + fn consumer_group(&self) -> Option<&str> { + None + } + /// Whether this router has a handler for `(kind, name)`. The runner acks and /// ignores a delivered message it does not handle rather than dead-lettering it. fn handles(&self, kind: MessageKind, name: &str) -> bool; diff --git a/src/bus/topology.rs b/src/bus/topology.rs new file mode 100644 index 00000000..f08989a2 --- /dev/null +++ b/src/bus/topology.rs @@ -0,0 +1,286 @@ +use super::router::MessageRouter; +use super::TransportError; + +pub(crate) const DEFAULT_BUS_NAMESPACE: &str = "default"; +pub(crate) const MAX_TOPOLOGY_NAME_LEN: usize = 128; + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct BusTopologyConfig { + group: Option, + namespace: String, +} + +impl Default for BusTopologyConfig { + fn default() -> Self { + Self { + group: None, + namespace: DEFAULT_BUS_NAMESPACE.to_string(), + } + } +} + +impl BusTopologyConfig { + pub(crate) fn default_namespace() -> &'static str { + DEFAULT_BUS_NAMESPACE + } + + pub(crate) fn group(mut self, group: impl Into) -> Self { + self.group = Some(group.into()); + self + } + + pub(crate) fn namespace(mut self, namespace: impl Into) -> Self { + self.namespace = namespace.into(); + self + } + + pub(crate) fn namespace_unchecked(&self) -> &str { + &self.namespace + } + + pub(crate) fn resolve_consumer_group( + &self, + router: &R, + transport: &str, + ) -> Result { + resolve_consumer_group(self.group.as_deref(), router, transport) + } + + pub(crate) fn namespace_for(&self, transport: &str) -> Result { + validate_namespace(&self.namespace, transport) + } + + pub(crate) fn validate_for(self, transport: &str) -> Result { + let group = self + .group + .map(|group| validate_consumer_group(&group, transport)) + .transpose()?; + let namespace = validate_namespace(&self.namespace, transport)?; + Ok(Self { group, namespace }) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TopologyNameKind { + ConsumerGroup, + Namespace, +} + +impl TopologyNameKind { + fn label(self) -> &'static str { + match self { + Self::ConsumerGroup => "consumer group", + Self::Namespace => "namespace", + } + } + + fn allows_dot(self) -> bool { + matches!(self, Self::Namespace) + } +} + +pub(crate) fn validate_consumer_group( + value: &str, + transport: &str, +) -> Result { + validate_topology_name(value, TopologyNameKind::ConsumerGroup, transport) +} + +pub(crate) fn validate_namespace(value: &str, transport: &str) -> Result { + validate_topology_name(value, TopologyNameKind::Namespace, transport) +} + +pub(crate) fn resolve_consumer_group( + explicit: Option<&str>, + router: &R, + transport: &str, +) -> Result { + let Some(group) = explicit.or_else(|| router.consumer_group()) else { + return Err(TransportError::permanent(format!( + "{transport} bus requires a consumer group; call `Service::named(..)` \ + for service consumers or `bus.group(..)` for direct consumers" + ))); + }; + + validate_consumer_group(group, transport) +} + +fn validate_topology_name( + value: &str, + kind: TopologyNameKind, + transport: &str, +) -> Result { + let label = kind.label(); + let trimmed = value.trim(); + if trimmed.is_empty() { + return Err(topology_error(transport, label, "cannot be empty")); + } + if trimmed.len() != value.len() { + return Err(topology_error( + transport, + label, + "cannot contain leading or trailing whitespace", + )); + } + if value.len() > MAX_TOPOLOGY_NAME_LEN { + return Err(topology_error( + transport, + label, + format!("cannot exceed {MAX_TOPOLOGY_NAME_LEN} bytes"), + )); + } + + for ch in value.chars() { + if ch.is_control() { + return Err(topology_error( + transport, + label, + format!("cannot contain control character {}", display_char(ch)), + )); + } + if ch.is_whitespace() { + return Err(topology_error( + transport, + label, + format!("cannot contain whitespace character {}", display_char(ch)), + )); + } + if matches!(ch, '*' | '>') { + return Err(topology_error( + transport, + label, + format!("cannot contain NATS wildcard {}", display_char(ch)), + )); + } + if matches!(ch, '/' | '\\') { + return Err(topology_error( + transport, + label, + format!("cannot contain path separator {}", display_char(ch)), + )); + } + if ch == '.' && !kind.allows_dot() { + return Err(topology_error( + transport, + label, + "cannot contain `.`; use `-` or `_` for portable group names", + )); + } + if !(ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') || ch == '.') { + return Err(topology_error( + transport, + label, + format!("cannot contain character {}", display_char(ch)), + )); + } + } + + Ok(value.to_string()) +} + +fn topology_error(transport: &str, label: &str, reason: impl std::fmt::Display) -> TransportError { + TransportError::permanent(format!("{transport} bus {label} {reason}")) +} + +fn display_char(ch: char) -> String { + if ch.is_control() { + format!("U+{:04X}", u32::from(ch)) + } else { + format!("`{ch}`") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::bus::{Message, MessageKind, SubscriptionPlan}; + + struct TestRouter { + group: Option<&'static str>, + } + + impl MessageRouter for TestRouter { + fn consumer_group(&self) -> Option<&str> { + self.group + } + + fn handles(&self, _kind: MessageKind, _name: &str) -> bool { + false + } + + fn subscription_plan(&self) -> SubscriptionPlan { + SubscriptionPlan::default() + } + + async fn dispatch(&self, _message: &Message) -> Result<(), TransportError> { + Ok(()) + } + } + + #[test] + fn resolve_consumer_group_prefers_explicit_bus_group() { + let router = TestRouter { + group: Some("service"), + }; + let group = resolve_consumer_group(Some("override"), &router, "test").unwrap(); + assert_eq!(group, "override"); + } + + #[test] + fn resolve_consumer_group_uses_router_identity() { + let router = TestRouter { + group: Some("service"), + }; + let group = resolve_consumer_group(None, &router, "test").unwrap(); + assert_eq!(group, "service"); + } + + #[test] + fn resolve_consumer_group_rejects_missing_identity() { + let router = TestRouter { group: None }; + let err = resolve_consumer_group(None, &router, "test").unwrap_err(); + assert!(err.is_permanent()); + assert!(err.message().contains("Service::named")); + } + + #[test] + fn resolve_consumer_group_rejects_whitespace_identity() { + let router = TestRouter { + group: Some(" service"), + }; + let err = resolve_consumer_group(None, &router, "test").unwrap_err(); + assert!(err.is_permanent()); + assert!(err.message().contains("whitespace")); + } + + #[test] + fn validate_consumer_group_rejects_wildcards() { + let err = validate_consumer_group("orders>*", "nats").unwrap_err(); + assert!(err.message().contains("NATS wildcard")); + } + + #[test] + fn validate_consumer_group_rejects_path_separators() { + let err = validate_consumer_group("tenant/orders", "rabbitmq").unwrap_err(); + assert!(err.message().contains("path separator")); + } + + #[test] + fn validate_consumer_group_rejects_overlong_names() { + let value = "a".repeat(MAX_TOPOLOGY_NAME_LEN + 1); + let err = validate_consumer_group(&value, "kafka").unwrap_err(); + assert!(err.message().contains("cannot exceed")); + } + + #[test] + fn validate_namespace_accepts_dotted_names() { + let namespace = validate_namespace("todos-prod.v1", "kafka").unwrap(); + assert_eq!(namespace, "todos-prod.v1"); + } + + #[test] + fn validate_namespace_rejects_control_characters() { + let err = validate_namespace("todos\nprod", "nats").unwrap_err(); + assert!(err.message().contains("control character")); + } +} diff --git a/src/lock/async_in_memory.rs b/src/lock/async_in_memory.rs index b70343e7..3963498c 100644 --- a/src/lock/async_in_memory.rs +++ b/src/lock/async_in_memory.rs @@ -144,8 +144,7 @@ impl AsyncLock for InMemoryAsyncLock { /// In-memory [`AsyncLockManager`] backed by a `HashMap>`. /// /// Lazily creates one [`InMemoryAsyncLock`] per unique key and returns the same -/// `Arc` for repeated lookups — the async counterpart to -/// [`InMemoryLockManager`](super::InMemoryLockManager). +/// `Arc` for repeated lookups. pub struct InMemoryAsyncLockManager { locks: Mutex>>, } diff --git a/src/microsvc/message_router.rs b/src/microsvc/message_router.rs index 97378df7..068827c0 100644 --- a/src/microsvc/message_router.rs +++ b/src/microsvc/message_router.rs @@ -9,6 +9,10 @@ use crate::bus::{MessageRouter, TransportError}; use crate::microsvc::{Message, MessageKind, Service, SubscriptionPlan}; impl MessageRouter for Service { + fn consumer_group(&self) -> Option<&str> { + self.name() + } + fn handles(&self, kind: MessageKind, name: &str) -> bool { self.handles_message(kind, name) } diff --git a/src/microsvc/service.rs b/src/microsvc/service.rs index 6645da39..8155a5d7 100644 --- a/src/microsvc/service.rs +++ b/src/microsvc/service.rs @@ -188,6 +188,7 @@ impl HandlerBuilder { /// [`Service::new`], adding dependencies and a bus with the `with_*` steps: /// `Service::new().with_repo(repo).with_read_model_store(store).with_bus(bus)`. pub struct Service { + name: Option, dependencies: D, handlers: HashMap<(MessageKind, String), RegisteredHandler>, handler_specs: Vec, @@ -198,6 +199,7 @@ impl Service { /// Build a service around an already-assembled dependency value. pub(crate) fn from_dependencies(dependencies: D) -> Self { Self { + name: None, dependencies, handlers: HashMap::new(), handler_specs: Vec::new(), @@ -205,6 +207,39 @@ impl Service { } } + fn map_dependencies(self, map: impl FnOnce(D) -> N) -> Service { + let Service { + name, dependencies, .. + } = self; + Service { + name, + dependencies: map(dependencies), + handlers: HashMap::new(), + handler_specs: Vec::new(), + runner: None, + } + } + + fn replace_dependencies(self, dependencies: N) -> Service { + self.map_dependencies(|_| dependencies) + } + + /// Assign a stable service/deployment identity. + /// + /// Broker-backed buses use this as the default durable consumer group when the + /// bus itself was not configured with an explicit group. Use the same name for + /// every replica of one service deployment; use different names for independent + /// event consumers that each need their own event copy. + pub fn named(mut self, name: impl Into) -> Self { + self.name = Some(name.into()); + self + } + + /// The stable service/deployment identity, if one was configured. + pub fn name(&self) -> Option<&str> { + self.name.as_deref() + } + /// Fail fast if handlers, specs, or a runner are already registered. The /// dependency builders (`with_repo`, `with_read_model_store`) reconstruct the /// service around a new dependency type, which would otherwise silently drop @@ -474,7 +509,7 @@ impl Service<()> { R: HasRepo + Send + Sync + 'static, { self.assert_no_registrations("with_repo"); - Service::from_dependencies(repo) + self.replace_dependencies(repo) } /// Use a read-model store as the service's dependency. @@ -483,7 +518,7 @@ impl Service<()> { S: HasReadModelStore + Send + Sync + 'static, { self.assert_no_registrations("with_read_model_store"); - Service::from_dependencies(read_model_store) + self.replace_dependencies(read_model_store) } } @@ -499,10 +534,7 @@ impl Service { S: HasReadModelStore + Send + Sync + 'static, { self.assert_no_registrations("with_read_model_store"); - Service::from_dependencies(RepoReadModelDependencies::new( - self.dependencies, - read_model_store, - )) + self.map_dependencies(|repo| RepoReadModelDependencies::new(repo, read_model_store)) } } @@ -555,6 +587,20 @@ mod tests { Service::new() } + #[test] + fn named_service_preserves_identity_across_dependency_builders() { + let service = Service::new() + .named("todo-api") + .with_repo(crate::HashMapRepository::new()) + .with_read_model_store(crate::HashMapRepository::new()); + + assert_eq!(service.name(), Some("todo-api")); + assert_eq!( + crate::bus::MessageRouter::consumer_group(&service), + Some("todo-api") + ); + } + #[tokio::test] async fn dispatch_returns_handler_result() { let service = test_service() diff --git a/tests/distributed_read_model/main.rs b/tests/distributed_read_model/main.rs index 80bb5591..a6b67b8d 100644 --- a/tests/distributed_read_model/main.rs +++ b/tests/distributed_read_model/main.rs @@ -1089,7 +1089,9 @@ fn nats_url() -> Option { #[cfg(feature = "nats")] async fn nats_matrix_bus(ns: &str) -> distributed::bus::NatsBus { let url = nats_url().expect("NATS_URL set"); - let bus = distributed::bus::NatsBus::connect(&url, "matrix", ns) + let bus = distributed::bus::NatsBus::connect(&url) + .group("matrix") + .namespace(ns) .await .expect("nats connect") .with_fetch_timeout(Duration::from_millis(800)); @@ -1144,7 +1146,9 @@ async fn rabbit_matrix_bus( collector: &StdArc>, ) -> distributed::bus::RabbitBus { let url = amqp_url().expect("AMQP_URL set"); - let bus = distributed::bus::RabbitBus::connect(&url, "matrix", ns) + let bus = distributed::bus::RabbitBus::connect(&url) + .group("matrix") + .namespace(ns) .await .expect("rabbit connect"); // Topic exchange drops events with no bound queue, so bind before publishing. @@ -1200,7 +1204,9 @@ fn kafka_brokers() -> Option { #[cfg(feature = "kafka")] async fn kafka_matrix_bus(ns: &str) -> distributed::bus::KafkaBus { let brokers = kafka_brokers().expect("KAFKA_BROKERS set"); - distributed::bus::KafkaBus::connect(&brokers, "matrix", ns) + distributed::bus::KafkaBus::connect(&brokers) + .group("matrix") + .namespace(ns) .await .expect("kafka connect") .with_fetch_timeout(Duration::from_secs(10)) @@ -1255,7 +1261,7 @@ async fn matrix_in_memory_persistence_over_postgres_bus() { return; }; let bus_pool = schema.repository().await.pool().clone(); - let bus = PostgresBus::new(bus_pool, "matrix"); + let bus = PostgresBus::new(bus_pool).group("matrix"); bus.ensure_tables().await.expect("postgres bus tables"); let (collector, collected) = build_collector(); run_checkout_over_bus( @@ -1324,7 +1330,7 @@ async fn postgres_matrix_bus() -> Option { "skipping Postgres-bus matrix cell", ) .await?; - let bus = PostgresBus::new(schema.repository().await.pool().clone(), "matrix"); + let bus = PostgresBus::new(schema.repository().await.pool().clone()).group("matrix"); bus.ensure_tables().await.expect("postgres bus tables"); // The schema has no Drop, so the bus's tables outlive this fixture. Some(bus) diff --git a/tests/kafka_transport/main.rs b/tests/kafka_transport/main.rs index 7e3a2732..6a767e7d 100644 --- a/tests/kafka_transport/main.rs +++ b/tests/kafka_transport/main.rs @@ -35,6 +35,26 @@ fn recording_for(name: &str, kind: MessageKind, rec: Arc>>) -> })) } +fn named_recording_for( + service_name: &str, + name: &str, + kind: MessageKind, + rec: Arc>>, +) -> Arc> { + let leaked: &'static str = Box::leak(name.to_string().into_boxed_str()); + let builder = Service::new().named(service_name.to_string()); + let registered = match kind { + MessageKind::Command => builder.command(leaked), + MessageKind::Event => builder.event(leaked), + }; + Arc::new(registered.handle(move |ctx: &Context<()>| { + rec.lock() + .unwrap() + .push(ctx.message().id().unwrap_or_default().to_string()); + async move { Ok(json!({})) } + })) +} + fn brokers() -> Option { match std::env::var("KAFKA_BROKERS") { Ok(b) => Some(b), @@ -157,7 +177,9 @@ async fn bus_listen_shared_group_consumes_each_command_once() { let Some(brokers) = brokers() else { return }; let ns = unique("ns"); - let producer = KafkaBus::connect(&brokers, "orders", &ns) + let producer = KafkaBus::connect(&brokers) + .group("orders") + .namespace(&ns) .await .expect("connect producer"); let total = 5; @@ -173,7 +195,9 @@ async fn bus_listen_shared_group_consumes_each_command_once() { // First member of group "orders" drains every command. let first = Arc::new(Mutex::new(Vec::new())); - KafkaBus::connect(&brokers, "orders", &ns) + KafkaBus::connect(&brokers) + .group("orders") + .namespace(&ns) .await .unwrap() .with_fetch_timeout(Duration::from_secs(10)) @@ -191,7 +215,9 @@ async fn bus_listen_shared_group_consumes_each_command_once() { // A second member of the SAME group sees nothing — the group already consumed // and committed past these records (point-to-point, not fan-out). let second = Arc::new(Mutex::new(Vec::new())); - KafkaBus::connect(&brokers, "orders", &ns) + KafkaBus::connect(&brokers) + .group("orders") + .namespace(&ns) .await .unwrap() .with_fetch_timeout(Duration::from_secs(6)) @@ -215,7 +241,9 @@ async fn bus_subscribe_fans_out_across_groups() { let Some(brokers) = brokers() else { return }; let ns = unique("ns"); - let producer = KafkaBus::connect(&brokers, "producer", &ns) + let producer = KafkaBus::connect(&brokers) + .group("producer") + .namespace(&ns) .await .expect("connect producer"); let total = 4; @@ -232,7 +260,9 @@ async fn bus_subscribe_fans_out_across_groups() { let expected: Vec = (0..total).map(|i| format!("e{i}")).collect(); for group in ["projections", "audit"] { let rec = Arc::new(Mutex::new(Vec::new())); - KafkaBus::connect(&brokers, group, &ns) + KafkaBus::connect(&brokers) + .group(group) + .namespace(&ns) .await .unwrap() .with_fetch_timeout(Duration::from_secs(10)) @@ -247,3 +277,48 @@ async fn bus_subscribe_fans_out_across_groups() { assert_eq!(ids, expected, "group {group} sees every event"); } } + +#[tokio::test] +async fn bus_subscribe_uses_named_service_as_consumer_group() { + let Some(brokers) = brokers() else { return }; + let ns = unique("ns"); + + let producer = KafkaBus::connect(&brokers) + .namespace(&ns) + .await + .expect("connect producer"); + for i in 0..3 { + producer + .publish_message( + Message::new("order.initialized", MessageKind::Event, b"{}".to_vec()) + .with_id(format!("e{i}")), + ) + .await + .expect("publish event"); + } + + let rec = Arc::new(Mutex::new(Vec::new())); + KafkaBus::connect(&brokers) + .namespace(&ns) + .await + .unwrap() + .with_fetch_timeout(Duration::from_secs(10)) + .subscribe( + named_recording_for( + "order-projection", + "order.initialized", + MessageKind::Event, + rec.clone(), + ), + RunOptions::idempotent(), + ) + .await + .expect("subscriber drains"); + + let mut ids = rec.lock().unwrap().clone(); + ids.sort(); + assert_eq!( + ids, + vec!["e0".to_string(), "e1".to_string(), "e2".to_string()] + ); +} diff --git a/tests/nats_transport/main.rs b/tests/nats_transport/main.rs index 186aa207..6d9b93db 100644 --- a/tests/nats_transport/main.rs +++ b/tests/nats_transport/main.rs @@ -168,6 +168,26 @@ fn recording_service( })) } +fn named_recording_service( + service_name: &str, + name: &str, + kind: MessageKind, + rec: Arc>>, +) -> Arc> { + let leaked: &'static str = Box::leak(name.to_string().into_boxed_str()); + let builder = Service::new().named(service_name.to_string()); + let registered = match kind { + MessageKind::Command => builder.command(leaked), + MessageKind::Event => builder.event(leaked), + }; + Arc::new(registered.handle(move |ctx: &Context<()>| { + rec.lock() + .unwrap() + .push(ctx.message().id().unwrap_or_default().to_string()); + async move { Ok(json!({})) } + })) +} + /// `send` + `listen`: replicas sharing a `group` compete for the command — each /// message is handled exactly once across the pool (point-to-point). #[tokio::test] @@ -176,7 +196,9 @@ async fn bus_send_listen_is_point_to_point_across_a_group() { let namespace = unique("ns").to_lowercase(); let group = "orders"; - let producer = NatsBus::connect(&url, group, &namespace) + let producer = NatsBus::connect(&url) + .group(group) + .namespace(&namespace) .await .expect("connect producer") .with_fetch_timeout(Duration::from_millis(600)); @@ -195,7 +217,9 @@ async fn bus_send_listen_is_point_to_point_across_a_group() { // Two replicas of the same service (same group) drain concurrently. let rec = Arc::new(Mutex::new(Vec::new())); - let bus_a = NatsBus::connect(&url, group, &namespace) + let bus_a = NatsBus::connect(&url) + .group(group) + .namespace(&namespace) .await .unwrap() .with_fetch_timeout(Duration::from_millis(600)); @@ -226,7 +250,9 @@ async fn bus_publish_subscribe_fans_out_across_groups() { let Some(url) = nats_url() else { return }; let namespace = unique("ns").to_lowercase(); - let producer = NatsBus::connect(&url, "publisher", &namespace) + let producer = NatsBus::connect(&url) + .group("publisher") + .namespace(&namespace) .await .expect("connect producer"); producer.ensure_stream().await.expect("ensure stream"); @@ -244,7 +270,9 @@ async fn bus_publish_subscribe_fans_out_across_groups() { let expected: Vec = (0..total).map(|i| format!("e{i}")).collect(); for group in ["projections", "audit"] { - let bus = NatsBus::connect(&url, group, &namespace) + let bus = NatsBus::connect(&url) + .group(group) + .namespace(&namespace) .await .unwrap() .with_fetch_timeout(Duration::from_millis(600)); @@ -260,3 +288,50 @@ async fn bus_publish_subscribe_fans_out_across_groups() { assert_eq!(ids, expected, "group {group} sees every event"); } } + +#[tokio::test] +async fn bus_subscribe_uses_named_service_as_consumer_group() { + let Some(url) = nats_url() else { return }; + let namespace = unique("ns").to_lowercase(); + + let producer = NatsBus::connect(&url) + .namespace(&namespace) + .await + .expect("connect producer"); + producer.ensure_stream().await.expect("ensure stream"); + + for i in 0..3 { + producer + .publish_message( + Message::new("order.initialized", MessageKind::Event, b"{}".to_vec()) + .with_id(format!("e{i}")), + ) + .await + .expect("publish event"); + } + + let rec = Arc::new(Mutex::new(Vec::new())); + NatsBus::connect(&url) + .namespace(&namespace) + .await + .unwrap() + .with_fetch_timeout(Duration::from_millis(600)) + .subscribe( + named_recording_service( + "order-projection", + "order.initialized", + MessageKind::Event, + rec.clone(), + ), + RunOptions::idempotent(), + ) + .await + .expect("subscriber drains"); + + let mut ids = rec.lock().unwrap().clone(); + ids.sort(); + assert_eq!( + ids, + vec!["e0".to_string(), "e1".to_string(), "e2".to_string()] + ); +} diff --git a/tests/postgres_transport/main.rs b/tests/postgres_transport/main.rs index b7cd343b..da5c5b28 100644 --- a/tests/postgres_transport/main.rs +++ b/tests/postgres_transport/main.rs @@ -208,6 +208,26 @@ fn recording_for(name: &str, kind: MessageKind, rec: Arc>>) -> })) } +fn named_recording_for( + service_name: &str, + name: &str, + kind: MessageKind, + rec: Arc>>, +) -> Arc> { + let leaked: &'static str = Box::leak(name.to_string().into_boxed_str()); + let builder = Service::new().named(service_name.to_string()); + let registered = match kind { + MessageKind::Command => builder.command(leaked), + MessageKind::Event => builder.event(leaked), + }; + Arc::new(registered.handle(move |ctx: &Context<()>| { + rec.lock() + .unwrap() + .push(ctx.message().id().unwrap_or_default().to_string()); + async move { Ok(json!({})) } + })) +} + /// `send` + `listen`: the work queue is claimed `FOR UPDATE SKIP LOCKED`, so two /// replicas sharing a `group` compete — each command handled exactly once. #[tokio::test] @@ -216,7 +236,7 @@ async fn bus_send_listen_is_point_to_point_across_a_group() { return; }; let repo = schema.repository().await; - let bus = PostgresBus::new(repo.pool().clone(), "orders"); + let bus = PostgresBus::new(repo.pool().clone()).group("orders"); bus.ensure_tables().await.expect("ensure tables"); let total = 6; @@ -263,7 +283,7 @@ async fn bus_publish_subscribe_fans_out_across_groups() { }; let repo = schema.repository().await; let pool = repo.pool().clone(); - let producer = PostgresBus::new(pool.clone(), "producer"); + let producer = PostgresBus::new(pool.clone()).group("producer"); producer.ensure_tables().await.expect("ensure tables"); let total = 4; @@ -279,7 +299,7 @@ async fn bus_publish_subscribe_fans_out_across_groups() { let expected: Vec = (0..total).map(|i| format!("e{i}")).collect(); for group in ["projections", "audit"] { - let bus = PostgresBus::new(pool.clone(), group); + let bus = PostgresBus::new(pool.clone()).group(group); let rec = Arc::new(Mutex::new(Vec::new())); bus.subscribe( recording_for("order.initialized", MessageKind::Event, rec.clone()), @@ -292,3 +312,46 @@ async fn bus_publish_subscribe_fans_out_across_groups() { assert_eq!(ids, expected, "group {group} sees every event"); } } + +#[tokio::test] +async fn bus_subscribe_uses_named_service_as_consumer_group() { + let Some(schema) = postgres::PostgresTestSchema::create_from_env("bus_named_group", SKIP).await + else { + return; + }; + let repo = schema.repository().await; + let pool = repo.pool().clone(); + let producer = PostgresBus::new(pool.clone()); + producer.ensure_tables().await.expect("ensure tables"); + + for i in 0..3 { + producer + .publish_message( + Message::new("order.initialized", MessageKind::Event, b"{}".to_vec()) + .with_id(format!("e{i}")), + ) + .await + .expect("publish event"); + } + + let rec = Arc::new(Mutex::new(Vec::new())); + PostgresBus::new(pool) + .subscribe( + named_recording_for( + "order-projection", + "order.initialized", + MessageKind::Event, + rec.clone(), + ), + RunOptions::idempotent(), + ) + .await + .expect("subscriber drains"); + + let mut ids = rec.lock().unwrap().clone(); + ids.sort(); + assert_eq!( + ids, + vec!["e0".to_string(), "e1".to_string(), "e2".to_string()] + ); +} diff --git a/tests/rabbitmq_transport/main.rs b/tests/rabbitmq_transport/main.rs index 8c0390b9..26967321 100644 --- a/tests/rabbitmq_transport/main.rs +++ b/tests/rabbitmq_transport/main.rs @@ -62,6 +62,26 @@ fn recording_for(name: &str, kind: MessageKind, rec: Arc>>) -> })) } +fn named_recording_for( + service_name: &str, + name: &str, + kind: MessageKind, + rec: Arc>>, +) -> Arc> { + let leaked: &'static str = Box::leak(name.to_string().into_boxed_str()); + let builder = Service::new().named(service_name.to_string()); + let registered = match kind { + MessageKind::Command => builder.command(leaked), + MessageKind::Event => builder.event(leaked), + }; + Arc::new(registered.handle(move |ctx: &Context<()>| { + rec.lock() + .unwrap() + .push(ctx.message().id().unwrap_or_default().to_string()); + async move { Ok(json!({})) } + })) +} + #[tokio::test] async fn publish_then_consume_round_trips_through_rabbitmq() { let Some(url) = amqp_url() else { return }; @@ -164,7 +184,9 @@ async fn bus_send_listen_is_point_to_point_across_a_group() { let Some(url) = amqp_url() else { return }; let ns = unique("ns").to_lowercase(); - let producer = RabbitBus::connect(&url, "orders", &ns) + let producer = RabbitBus::connect(&url) + .group("orders") + .namespace(&ns) .await .expect("connect producer"); let total = 6; @@ -180,8 +202,16 @@ async fn bus_send_listen_is_point_to_point_across_a_group() { // Two replicas of the same group (separate connections) drain concurrently. let rec = Arc::new(Mutex::new(Vec::new())); - let bus_a = RabbitBus::connect(&url, "orders", &ns).await.unwrap(); - let bus_b = RabbitBus::connect(&url, "orders", &ns).await.unwrap(); + let bus_a = RabbitBus::connect(&url) + .group("orders") + .namespace(&ns) + .await + .unwrap(); + let bus_b = RabbitBus::connect(&url) + .group("orders") + .namespace(&ns) + .await + .unwrap(); let (ra, rb) = tokio::join!( bus_a.listen( recording_for("order.initialize", MessageKind::Command, rec.clone()), @@ -211,7 +241,9 @@ async fn bus_publish_subscribe_fans_out_across_groups() { let Some(url) = amqp_url() else { return }; let ns = unique("ns").to_lowercase(); - let producer = RabbitBus::connect(&url, "producer", &ns) + let producer = RabbitBus::connect(&url) + .group("producer") + .namespace(&ns) .await .expect("connect producer"); @@ -221,8 +253,16 @@ async fn bus_publish_subscribe_fans_out_across_groups() { let audit_rec = Arc::new(Mutex::new(Vec::new())); let svc_proj = recording_for("order.initialized", MessageKind::Event, proj_rec.clone()); let svc_audit = recording_for("order.initialized", MessageKind::Event, audit_rec.clone()); - let bus_proj = RabbitBus::connect(&url, "projections", &ns).await.unwrap(); - let bus_audit = RabbitBus::connect(&url, "audit", &ns).await.unwrap(); + let bus_proj = RabbitBus::connect(&url) + .group("projections") + .namespace(&ns) + .await + .unwrap(); + let bus_audit = RabbitBus::connect(&url) + .group("audit") + .namespace(&ns) + .await + .unwrap(); bus_proj .ensure_subscription(svc_proj.as_ref()) .await @@ -260,3 +300,49 @@ async fn bus_publish_subscribe_fans_out_across_groups() { assert_eq!(proj_ids, expected, "projections sees every event"); assert_eq!(audit_ids, expected, "audit sees every event"); } + +#[tokio::test] +async fn bus_subscribe_uses_named_service_as_consumer_group() { + let Some(url) = amqp_url() else { return }; + let ns = unique("ns").to_lowercase(); + + let producer = RabbitBus::connect(&url) + .namespace(&ns) + .await + .expect("connect producer"); + let bus = RabbitBus::connect(&url) + .namespace(&ns) + .await + .expect("connect subscriber"); + let rec = Arc::new(Mutex::new(Vec::new())); + let service = named_recording_for( + "order-projection", + "order.initialized", + MessageKind::Event, + rec.clone(), + ); + bus.ensure_subscription(service.as_ref()) + .await + .expect("bind subscriber"); + + for i in 0..3 { + producer + .publish_message( + Message::new("order.initialized", MessageKind::Event, b"{}".to_vec()) + .with_id(format!("e{i}")), + ) + .await + .expect("publish event"); + } + + bus.subscribe(service, RunOptions::idempotent()) + .await + .expect("subscriber drains"); + + let mut ids = rec.lock().unwrap().clone(); + ids.sort(); + assert_eq!( + ids, + vec!["e0".to_string(), "e1".to_string(), "e2".to_string()] + ); +} From be3d94ab81da34c63a6da1383958ab319fc2b2a9 Mon Sep 17 00:00:00 2001 From: Patrick Lee Scott Date: Wed, 10 Jun 2026 15:28:28 -0500 Subject: [PATCH 2/2] test: cover named handler consumer group Addresses CodeRabbit review on [[tasks/address-coderabbit-inferred-bus-topology]]. --- src/bus/handlers.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/bus/handlers.rs b/src/bus/handlers.rs index fe1bd89a..ec67b26a 100644 --- a/src/bus/handlers.rs +++ b/src/bus/handlers.rs @@ -176,6 +176,15 @@ mod tests { } } + #[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() { let handlers = Handlers::new()