ADR: NATS Messaging Abstraction

Author: Michael Schleicher

This ADR proposes a type-safe, mockable messaging abstraction for inter-service communication in Arminius, and asks reviewers to agree on its interface. This document describes what that design is, why it is shaped that way, and which alternatives were weighed.

What This Is Really About

Arminius is event-driven: services coordinate by emitting events and answering commands. Which events a service emits and which commands it answers is the contract it offers the rest of the system - and that is domain logic, owned by the domain.

The goal of this abstraction is to let a domain declare its messages as plain Rust types (e.g. “emits RoomCreated”, “answers GetDevice with a device id”) and write its logic against those types alone. How a message travels - NATS subjects, JetStream, serialization, acknowledgement - is an architectural concern the domain should not have to care about; changing the transport must not ripple into domain logic.

So the domain side stays clean: a message is just a type, domain code reads as emit this event / handle this command, and the transport (with any performance tuning) lives behind the seam. The rest of this ADR is the technical means to that end.

Context

Today, services share a common NatsConnection type that handles connecting and stream setup. On top of it, each service hand-writes its own messaging layer: a wrapper client, a subject type, and the payload serialisation, all built directly against NatsConnection and async-nats/JetStream. Nothing is shared, so the same plumbing is reinvented per service and every wrapper stays bound to the concrete transport.

This has two recurring problems:

• Transport-coupled. Every wrapper depends on async_nats/JetStream directly. Subject layout, stream names, retention, and consumer configuration are spread across these per-service wrappers rather than centralized.
• Untestable without a broker. Because the transport is hard-wired, exercising a handler in a unit test requires a running NATS instance.

We want a bus where a message is published and received purely through traits - no per-service wrapper and no manual serde at the callsite - where subject and JetStream configuration are centralized inside the abstraction, and where handlers can be unit-tested without a running NATS instance.

Decision

Define messaging as a set of transport-agnostic traits: Event/Command describe a message, Publisher/Subscriber and Requester/Responder move it. Domain code depends only on these traits; the NATS/JetStream implementation lives behind them (see Crate Organisation). The rest of this document specifies that interface.

Message Types

There are two different message types we currently support in Arminius, Events and Commands. First we define the interface that the domain logic uses to publish, subscribe, request, and respond to those messages. A concrete message is then a plain Rust type that implements the matching carrier trait, which the interface traits are generic over.

Messages should support compatibility to older messages and versioning to provide safety for future changes. All messages will use JSON as a transport format.

The following signatures and trait bounds shown here are not set in stone. Migrating real callsites may surface a need to adjust them, and this ADR will be updated to match.

Events

An event is a state change in the system: Something happened and the service wants to notify interested parties. Delivery matters - the event must arrive - but the emitting service does not care who, or how many, listeners there are.

Interface

A type becomes an event by implementing Event. It binds the type to the subject it lives on, plus the subjects of older versions it can still consume:

pub trait Event {
    /// The subject this event type lives under (versioned; may contain `*`).
    const SUBJECT: &'static str;
    /// Subjects of older compatible versions, consumed and upcast via `From`.
    const COMPATIBLE_SUBJECTS: &'static [&'static str] = &[];
}

Publisher and Subscriber provide the interface to work with the Event:

/// Publisher that can publish message onto their specific subjects
pub trait Publisher {
    fn publish<E>(&self, message: E) -> impl Future<Output = Result<(), Error>>
    where E: Event + Clone + 'static;
}

/// Subscriber will receive on subjects tied to the corresponding type.
pub trait Subscriber {
    fn subscribe<E: Event + 'static>(
        &self,
    ) -> impl Future<Output = Result<
        impl Stream<Item = Result<impl ReceivedEvent<E> + Send + Sync + 'static, Error>>,
        Error,
    >>;
}

/// Helper trait to convey acknowledgment and tracing context.
pub trait ReceivedEvent<E: Event>: Deref<Target = E> {
    fn ack(&self) -> impl Future<Output = Result<(), Error>> + Send;
    fn context(&self) -> &opentelemetry::Context;
}

Example

The #[derive(Event)] macro is purely a convenience: It generates the Event implementation from #[subject(...)] attributes. A type can implement Event by hand just as well.

#[derive(Clone, Serialize, Deserialize, Event)]
#[subject(version = "1")]
struct RoomCreated { room_id: String }

async fn announce(publisher: &impl Publisher, room_id: String) {
    publisher.publish(RoomCreated { room_id }).await.unwrap();
}

async fn consume(subscriber: &impl Subscriber) {
    let stream = subscriber.subscribe::<RoomCreated>().await.unwrap();
    futures::pin_mut!(stream);
    while let Some(Ok(event)) = stream.next().await {
        println!("room created: {}", event.room_id);
        event.ack().await.unwrap();
    }
}

Commands

Commands are requests that require a response from another service. In a microservice architecture services should stay self-reliant and avoid querying others: a service subscribes to the Events it needs and keeps its own copy of that information, so that an outage of another service does not take the whole system down. Sometimes that pattern cannot hold - information must be synced live, or requested because it may not be stored elsewhere - and that is what commands are for.

Interface

A type becomes a command by implementing Command, which binds it to its subject and - crucially - its reply type:

pub trait Command {
    /// The subject this command is sent on (versioned; may contain `*`).
    const SUBJECT: &'static str;
    /// The reply type. `Requester::request` returns exactly this.
    type Reply;

    /// How often the command may reach its handler. Defaults to `AtLeastOnce`.
    const DELIVERY: Delivery = Delivery::AtLeastOnce;
}

Because type Reply lives on the command type, request and response are bound at compile time - a request can never be matched to the wrong reply type. The command declares only the reply type, not where the reply goes. The reply subject is handled by the NATS implementation (a per-request inbox in a header), never by the callsite.

DELIVERY sits on the command type for the same reason. Request-reply gives a command one responder, so every instance of the answering service derives the same guarantee and they cannot disagree about it. Events carry no such constant, because an event has many subscribers and whether handling it twice is safe is a property of each subscriber, not of the event.

The constant decides one thing in the implementation: whether the consumer may hand the command out again. AtLeastOnce leaves redelivery on, so a handler that dies before answering has its command picked up by another attempt. AtMostOnce says the handler has an effect that must not be repeated, so it acknowledges the command before starting work and the transport caps its consumer at a single delivery - an expiring acknowledgement window then hands the command to nobody.

Requester and Responder provide the interface to work with the Command:

/// Requester sends a command and awaits its typed reply.
pub trait Requester {
    fn request<C: Command + Send>(&self, cmd: C)
        -> impl Future<Output = Result<C::Reply, Error>> + Send;
}

/// Responder subscribes to commands of one type and answers each with a typed reply.
pub trait Responder {
    fn respond<C: Command + Send + 'static>(self)
        -> impl Future<Output = Result<
            impl Stream<Item = Result<impl ReceivedCommand<C>, Error>> + Send + 'static,
            Error,
        >> + Send + 'static;
}

/// Helper trait to convey the acknowledgement, response and tracing context
pub trait ReceivedCommand<C: Command>: Deref<Target = C> {
    /// Take responsibility for the command without answering it yet.
    fn ack(&self) -> impl Future<Output = Result<(), Error>> + Send;
    /// Send the response and acknowledge the command.
    fn reply(&self, response: C::Reply) -> impl Future<Output = Result<(), Error>> + Send;
    fn context(&self) -> &opentelemetry::Context;
}

Example

The #[derive(Command)] macro is purely a convenience: it generates the trait Command implementation from #[subject(...)] / #[command(...)] attributes.

/// Request to create a new room and if successful get the generated room id.
#[derive(Serialize, Deserialize, Command)]
#[subject("rooms.create")]
#[command(reply = Result<String, CreateRoomError>)]
pub struct CreateRoomCommand { pub name: String }

async fn create(requester: impl Requester) {
    let cmd = CreateRoomCommand { name: "lobby".into() };
    // `reply` is the newly generated room id.
    let reply: Result<String, CreateRoomError> = requester.request(cmd).await.unwrap();
}

async fn serve(responder: impl Responder) {
    let stream = responder.respond::<CreateRoomCommand>().await.unwrap();
    futures::pin_mut!(stream);
    while let Some(Ok(cmd)) = stream.next().await {
        let result = handle(&*cmd);
        cmd.reply(result).await.unwrap();
    }
}

Errors

Every operation returns the same transport-agnostic Error, so domain code never depends on async_nats error types:

pub enum Error {
    SendFailed { error: String },
    SubscriptionFailed { subject: String, error: String },
    ReceptionFailed { subject: String, error: String },
    AckFailed { error: String },
}

Subject and JetStream Configuration Are Centralized

Because the subject is a property carried by the trait - Event::SUBJECT and Command::SUBJECT - callsites never spell a subject out. They name a type; the type knows its subject.

• Versioned subjects. A subject defaults to <StructName>.v<version>. Per-instance subjects use {field} placeholders: each becomes a * wildcard in SUBJECT (the subscribe/stream pattern) and is interpolated into the concrete subject() (the publish target). For example "rooms.{room_id}.events.v1" gives a subscribe pattern of "rooms.*.events.v1" and a publish target of "rooms.7.events.v1".
• Stream and consumer configuration lives entirely in arminius-lib-nats. Stream naming, retention policy, durable consumer names (the consumer name is the service name, so instances of one service load-balance automatically), and ack windows are set in the implementation. Streams are created idempotently on first use. Callsites configure none of this. Configuration is per-connection, with one exception carried on the message type itself: a command declaring Delivery::AtMostOnce gets a consumer that never redelivers it.
• Schema evolution is also behind the trait. COMPATIBLE_SUBJECTS lets a subscriber consume older versions and upcast them via From, with transitive subject matching across a version chain. The derive’s compatible = [..] attribute is just the convenience that populates this constant and wires up the upcast.

Testing without a Broker

arminius-lib-message-bus ships an in-memory implementation of the traits behind a mock feature, enabled as a dev-dependency so it never ships in release builds. Because service code is written against &impl Publisher / &impl Subscriber / impl ReceivedCommand, the mock substitutes directly:

• MockBus implements both Publisher and Subscriber. Published events are delivered to active subscriptions on a matching subject and recorded for assertions via published_as::<T>(). Clones share backing state.
• MockPublisher<T> captures published events of one type for assertions.
• MockSubscriber wraps a fixed Vec of events into a stream plus parallel “checkers” that assert each event was acknowledged.
• MockReceivedCommand + ReplyChecker let a test run a command handler and read back the captured reply (or observe that the handler dropped the command without replying).

This satisfies the requirement that handlers be unit-testable with no running NATS instance. Tests are run with cargo nextest run for process isolation.

These mocks are deliberately behavioural fakes (a publish really delivers to live subscribers) rather than expectation mocks à la mockall::automock.

Crate Organisation

The abstraction is split across three crates so the interface is decoupled from any concrete transport:

The key property is the layering: callsites depend only on the traits. Service code accepts &impl Publisher, &impl Subscriber, &impl Requester or impl Responder and never names a transport type. (Responder is the exception: respond takes self by value so the returned stream is 'static.) The NATS implementation can therefore be swapped for the in-memory mock with no change to the code under test, and nothing about JetStream leaks above the trait surface.

Alternatives Considered

• Raw NATS subjects + manual serde at each callsite (status quo). Rejected: this is exactly the untyped, transport-coupled, broker-dependent situation the abstraction exists to remove.
• A single untyped envelope/wrapper type (subject field + serde_json::Value body). Loses per-message types and compile-time checking of subjects and - critically - request/reply pairing. Rejected: it does not make the bus type-safe.
• Hand-written trait impls, no derive. Viable, and still supported - but writing SUBJECT/version constants and the payload functions for every type by hand is repetitive and easy to get wrong. The derive is sugar over this, not a replacement for it.
• A runtime subject registry/config file. Rejected: moves subject definitions out of the type and back to runtime, losing the compile-time guarantees and the “no raw subject at the callsite” property the design is built on.

Consequences

Gained

• compile-time–checked subjects and reply types
• a transport-agnostic callsite surface
• unit-testable handlers via mocks
• centralized JetStream configuration
• built-in versioning and schema evolution
• OpenTelemetry context propagation through the trait surface

Cost

• a proc-macro layer to maintain, plus a learning curve for the attribute syntax
• the traits bake in JetStream’s delivery model: ack(), redelivery and durable consumers are assumed by the interface, and Delivery names the two guarantees a transport has to offer, so one without those semantics would not drop in cleanly