Transports

Introduction

Featuremicro The transport moves messages between services. Your handlers and your ServiceClient calls never name it: the transport is a setting, so a service that runs on NATS today can move to Kafka with a configuration change and a restart.

Every transport carries the same thing, an envelope: a pattern (deals.get, deal.won), a set of headers (the message id, the request id, the trace context, credentials) and an encoded payload. What differs is how the envelope travels, what happens when a service is down, and how far each broker goes to deliver it exactly once.

Choosing a transport

If you have no broker yet, start with NATS. It is one small binary, request-reply is built into the protocol, JetStream makes events durable, and navree dock up runs it for development (see Dock).

Configuring the transport

Name the transport and where it is in the micro section. Each transport reads its own subsection for everything else:

[micro]
service = "notifications"
transport = "nats"
url = "nats://127.0.0.1:4222"

[micro.nats]
stream = "CRM"

In production, set the connection from the environment:

APP_MICRO__TRANSPORT=nats
APP_MICRO__URL=tls://nats.internal:4222
APP_MICRO__NATS__TOKEN=change-me

In code, each transport has a constructor on Transport, which MicroModule::for_root accepts directly:

use navree::micro::Transport;

Navree::app().module(MicroModule::for_root(Transport::kafka(["kafka-1:9092", "kafka-2:9092"])))

Delivery guarantees

Requests

A request is delivered to one instance of the service that answers the pattern: the instances share the work. The caller waits until the reply arrives or the deadline passes. If the deadline passes first, send returns MicroError::Timeout, and the handler may or may not have run: the request may have been lost on the way, or the reply on the way back.

Requests are not retried unless you ask with .retries(n), and then with the same message id, so a handler with the inbox on runs once. See retries.

Events

On every broker, events are delivered at least once:

1. emit returns once the broker has accepted the event (or once the event is in the outbox, inside a unit of work).
2. Each subscribing service receives it; one instance of that service handles it.
3. The instance acknowledges the event only after the handler returned Ok and its unit of work committed.
4. If the handler fails, or the instance dies before acknowledging, the broker delivers the event again.

So an event is never lost once emit has returned, but a handler can see it twice: after a crash between the commit and the acknowledgment, or after a broker failover. The inbox makes that harmless by skipping message ids it has already handled, in the same transaction as the handler's writes.

TCP, and NATS without JetStream, deliver events at most once: an event sent while the receiving service is down is gone. Use them for events you can afford to miss, such as cache invalidation.

Ordering

Every transport keeps the order in which one producer emitted events with the same key. Set the key on emit; events without one may be handled in any order:

self.client.emit("deal.updated", &event).key(event.deal_id).await?;

On the consuming side, Navree handles events with the same key one after another, and events with different keys in parallel, up to micro.concurrency. A retried event loses its place: the events after it may be handled first. Ordering has the details per transport.

Serialization

Payloads are encoded with micro.codec:

• json (the default): readable in every broker's tools, and what services written in other languages expect.
• msgpack: smaller and faster to encode, with the same serde types.

A payload is any type that implements Serialize and DeserializeOwned. Decoding follows serde's rules: unknown fields are ignored, and a missing field fails unless it has #[serde(default)]. A request whose payload doesn't decode gets a 400 reply; an event whose payload doesn't decode goes straight to the dead letters, because retrying it can't help.

Changing a payload

Services deploy one at a time, so for a while old and new versions of a payload are both in flight:

• Adding a field is safe when the receiver gives it a default (#[serde(default)], or an Option).
• Removing or renaming a field, or changing its type, is a new message: give it a new pattern (deals.get.v2), keep handling the old one until every sender has moved, then delete it.

The shared contract crate from the introduction makes the compiler check both sides of each change.

The envelope

Headers travel as the broker's native headers: NATS headers, Kafka record headers, AMQP message properties, Redis stream entry fields and MQTT 5 user properties (TCP frames carry them in the envelope). A service in another language talks to a Navree service by publishing the JSON payload where the transport puts the pattern, with these headers:

All of them are constants of navree::micro::headers (headers::MESSAGE_ID, headers::DEADLINE, ...).

A reply's body is the handler's result, or the error in Navree's JSON shape ({ "statusCode", "message", "error", "code" }, plus errors for validation failures).

micro.max_message_bytes (1 MiB by default) caps the encoded payload in both directions. A larger payload fails on the sending side with MicroError::PayloadTooLarge, before it reaches the broker. Send ids, not documents: put files in storage and send their path.

TCP

TCP connects services directly, with no broker. Each service listens on micro.tcp.addr, and the client finds the service of a pattern in micro.tcp.routes:

[micro]
transport = "tcp"

[micro.tcp]
addr = "0.0.0.0:4000"

[micro.tcp.routes]
"deals.*" = "deals.internal:4000"
"companies.*" = "companies.internal:4000"

Frames are length-prefixed envelopes over long-lived connections, with up to pool_size connections to each peer and many requests in flight on each. A pattern no route covers goes to micro.url, when it is set; a peer that doesn't handle a pattern answers "no responders" at once. emit sends the event to every route whose prefix matches, once, and doesn't wait for anyone to handle it: TCP has no storage, so an event for a service that is down is lost, and a failed handler is retried inside the receiving process. TCP has no dead-letter destination of its own: use micro.events.dead_letter = "database" or "log". Use TCP for request-response between a few services that know each other, and a broker for events.

NATS

Featuremicro-nats Requests use NATS request-reply on a subject equal to the pattern, with the service's instances in one queue group named after micro.service, so each request goes to one instance. When no instance listens, NATS says so at once and send fails with MicroError::NoResponders instead of waiting for the deadline.

Events go through JetStream on the subject <stream>.events.<pattern> (NAVREE.events.deal.won). micro:setup (or micro.auto_setup at boot) creates the stream micro.nats.stream, which stores every subject under <stream>., and one durable pull consumer per subscribing service and pattern, named <service>_<pattern> with dots as underscores (notifications_deal_won). A service that is down when an event is published gets it when it comes back, for as long as the stream keeps it. A failed event is negatively acknowledged with its backoff as the delay; an ordered handler's consumer has one event in flight at a time. Broadcast handlers use a plain core subscription, and dead letters go to <stream>.dead.<service>.<pattern>, stored by the same stream.

[micro]
transport = "nats"
url = "nats://nats-1:4222,nats://nats-2:4222"   # several servers, comma-separated

[micro.nats]
stream = "CRM"
replicas = 3
max_age_hours = 168

User names and passwords go in the URL: nats://deals:[email protected]:4222.

Kafka

Featuremicro-kafka Navree talks to Kafka through a small client of its own on the Kafka wire protocol, in pure Rust. It uses Kafka's consumer group protocol from KIP-848, where the broker assigns the partitions, so it needs Kafka 4.0 or later. It reads record batches compressed with gzip or snappy, and writes them uncompressed.

Each event pattern is a topic (deal.won), and each subscribing service is a consumer group named after micro.service. The event's key is the record key, hashed like the Java client does, so events with the same key land on the same partition and keep their order. Navree commits a partition's offset once every earlier record of it was handled, that is once each handler returned Ok and its unit of work committed, and releases a partition the group takes away only after the events it handed out are done. A new group starts at the end of its topics (start_from = "latest") or at their beginning ("earliest"); after that it continues from its committed offsets, so a service that has been listening gets the events published while it was down.

Publishing an event creates its topic when it is missing. Broadcast handlers read every partition of their topic from the end, without a group.

A failed event doesn't block its partition: it is copied to the service's retry topic (<service>.retry) with its attempt count and the time it becomes due, and handled again from there; after micro.events.max_attempts it goes to the dead letters.

Request-response works too: requests go to the pattern's topic and replies to the reply topic of the asking service (<service>.replies), which each of its instances reads. A request to a pattern whose topic doesn't exist fails at once with MicroError::NoResponders. Its latency is Kafka's, so prefer NATS, TCP or gRPC for chatty request-response and use Kafka for events.

[micro]
transport = "kafka"
url = "kafka-1:9092,kafka-2:9092,kafka-3:9092"

[micro.kafka]
partitions = 12
replication_factor = 3
tls = true
sasl = { mechanism = "SCRAM-SHA-512", username = "deals", password = "..." }

Set the password from the environment: APP_MICRO__KAFKA__SASL__PASSWORD.

RabbitMQ

Featuremicro-rabbitmq Navree declares one topic exchange, micro.rabbitmq.exchange, and publishes every message to it with the pattern as the routing key:

• Events. Each subscribing service gets a queue per pattern, <service>.<pattern>, bound to the exchange. Every service gets its own copy; the instances of one service compete for the messages in its queue.
• Requests. One queue per pattern, shared by the instances of the service that answers it. Replies come back through RabbitMQ's direct reply-to.
• Publishing. emit waits for the broker's publisher confirm, so an event is stored when emit returns.

Retries wait in a delay queue (<service>.<pattern>.retry, per-message TTL) and come back to the main queue when due, carrying navree-attempt. A request is published as mandatory, so one that no queue is bound for comes back at once as MicroError::NoResponders. Broadcast handlers each get an exclusive, auto-deleted queue, and an ordered handler's queue has a single active consumer. With micro.events.dead_letter = "transport", dead letters go to the <exchange>.dead topic exchange, routed by pattern.

[micro]
transport = "rabbitmq"
url = "amqps://deals:[email protected]:5671/crm"

[micro.rabbitmq]
exchange = "crm"
prefetch = 32

Redis Streams

Featuremicro-redis Each event pattern is a stream, <prefix><pattern> (navree:deal.won), and each subscribing service is a consumer group; each instance is a consumer in it. Navree acknowledges an entry with XACK once the handler's unit of work committed. Entries that an instance took but never acknowledged, because it crashed, are claimed by another instance after claim_after_ms, and the claim counts as a new attempt.

An entry holds the headers as fields and the payload in the field payload. A failed event's entry gets its idle time set (XCLAIM .. IDLE) so that the claim happens when its backoff is over. Requests use a stream per pattern too, read by the answering service's group; replies go to a short-lived stream per instance, <prefix>replies.<instance>. A pattern no service has subscribed to fails at once with MicroError::NoResponders. Broadcast handlers read with XREAD, without a group, and dead letters go to the stream <prefix>dead. Navree needs Redis 6.2 or later (XAUTOCLAIM).

Streams live in Redis's memory, so Navree trims them as it adds entries: each stream keeps about max_len entries. Size it for the longest outage you want a consumer to survive.

[micro]
transport = "redis"
url = "rediss://default:[email protected]:6380/0"

[micro.redis]
prefix = "crm:"
max_len = 500000

MQTT

Featuremicro-mqtt MQTT needs a broker that speaks MQTT 5. A pattern becomes a topic under micro.mqtt.prefix, with dots turned into slashes: deal.won is navree/deal/won.

• Events are published with QoS 1 (at least once). Each service subscribes with a shared subscription, $share/<service>/navree/deal/won, so one instance of each service gets each event. Sessions persist (session_expiry_secs), so events published while a service restarts wait for it on the broker.
• Requests use MQTT 5's response topic and correlation data; replies arrive on <prefix>/replies/<client id>, a topic private to the instance. A request nobody subscribes to fails at once with MicroError::NoResponders when the broker reports "no matching subscribers" in its PUBACK.
• Retries. MQTT has no negative acknowledgment, so a failed event stays unacknowledged while the instance retries it with the configured backoff before it dead-letters it; if the instance goes away first, the broker hands the event out again.
• Reconnecting. On reconnect, unacknowledged publishes are sent again and lost subscriptions are restored. Dead letters go to <prefix>/dead/<service>/<topic>.

Setting up the broker

Brokers need streams, topics, exchanges and queues before messages flow. micro:setup declares everything the app's patterns need: the JetStream stream and consumers, the Kafka topics (the patterns' topics, and <service>.retry, <service>.dead and <service>.replies), the RabbitMQ exchange, queues and bindings, the Redis consumer groups. MQTT topics need no declaration. It only adds what is missing, so running it again is safe:

cargo run -p notifications -- micro:setup

nats: stream CRM exists
nats: created consumer notifications_deal_won (deal.won)
nats: created consumer notifications_deal_assigned (deal.assigned)

In debug builds micro.auto_setup runs the same step at boot. In production, run micro:setup from your deploy, next to migrate, with credentials allowed to change the broker's topology; the services themselves can then run with credentials that only publish and consume.

Connections and failures

• Connecting. The app connects when the listener starts (serve or micro:listen) or on its first send or emit. Commands that don't use the transport, such as migrate, never connect. A connection attempt gives up after micro.connect_timeout_ms (5000).
• Reconnecting. A dropped connection is retried with exponential backoff, from 100 ms up to 30 seconds. While it is down, send fails at once with MicroError::Transport, emit inside a unit of work goes to the outbox and is published once the connection is back, and the readiness check reports the service as not ready.
• Stopping. On shutdown the listener stops taking new messages, waits for the ones in flight (each within its own time limit), acknowledges them, then closes the connection. Events that were taken but not finished are redelivered to another instance.

Several transports in one app

A service listens on one transport, but it can send to others. Name each extra client in micro.clients:

[micro.clients.billing]
transport = "tcp"
url = "billing.internal:4000"
request_timeout_ms = 2000

With MicroModule::for_root(options), every client in options.clients is registered. With from_config(), the configuration is read after the modules register, so name the clients in code too:

Navree::app().module(MicroModule::from_config().client("billing"))

A name with no [micro.clients.<name>] section fails the boot. Then inject the client by name:

#[injectable]
pub struct InvoiceService {
    #[inject(name = "billing")]
    billing: Arc<ServiceClient>,
}

The unnamed ServiceClient uses the app's own transport. A named client takes the same keys as the micro section (transport, url, codec, request_timeout_ms and the transport's subsection) and sends only; handlers are always served from the app's own transport.

Writing a transport

A transport is a type that implements MicroTransport. Navree gives it envelopes to send and publish, and a channel for the deliveries it receives; it handles the pipeline, the codec, retries and the inbox itself:

use std::time::Duration;

use navree::BoxFuture;
use navree::micro::{
    Delivery, DeliveryActions, Envelope, MicroError, MicroTransport, Subscription,
};
use tokio::sync::mpsc;

pub struct SqsTransport {
    // the client, queue URLs, ...
}

impl MicroTransport for SqsTransport {
    fn name(&self) -> &'static str {
        "sqs"
    }

    /// Sends a request and waits for its reply, until `envelope.deadline()`.
    fn request(&self, envelope: Envelope) -> BoxFuture<'_, Result<Envelope, MicroError>> {
        Box::pin(async move { todo!() })
    }

    /// Publishes an event; returns once the broker has stored it.
    fn publish(&self, envelope: Envelope) -> BoxFuture<'_, Result<(), MicroError>> {
        Box::pin(async move { todo!() })
    }

    /// Receives the subscribed patterns and hands each message over as a
    /// `Delivery`, until Navree closes the channel (`deliveries.closed()`).
    fn listen(
        &self,
        subscriptions: Vec<Subscription>,
        deliveries: mpsc::Sender<Delivery>,
    ) -> BoxFuture<'_, Result<(), MicroError>> {
        Box::pin(async move { todo!() })
    }
}

/// What Navree calls back for one received SQS message.
struct SqsActions {
    // the receipt handle, ...
}

impl DeliveryActions for SqsActions {
    fn ack(self: Box<Self>) -> BoxFuture<'static, Result<(), MicroError>> {
        Box::pin(async move { todo!() }) // delete the message
    }

    fn retry_in(self: Box<Self>, delay: Duration) -> BoxFuture<'static, Result<(), MicroError>> {
        Box::pin(async move { todo!() }) // change its visibility timeout to `delay`
    }

    fn reply(self: Box<Self>, reply: Envelope) -> BoxFuture<'static, Result<(), MicroError>> {
        Box::pin(async move { todo!() }) // send `reply` to the asker, then delete
    }
}

Each Subscription names a pattern, its kind (PatternKind::Message or PatternKind::Event), the group that shares it (Group::Service(name), or Group::Instance(id) for broadcast events), whether it is ordered, and its ack_wait. For each message it receives, the transport sends Delivery::new(envelope, index, attempt, actions) into the channel, where index is the position of the matching subscription and actions implements DeliveryActions: Navree calls ack, retry_in(delay) or reply(envelope) once the handler is done. The trait's other methods have defaults you override as needed: setup (what micro:setup declares), supports_wildcards, is_connected (the readiness check), dead_letter (the transport dead-letter destination) and close.

navree::micro::conformance holds the behavior every transport must have, as checks you run against yours: implement its Harness (which builds instances of a service and a client) and call conformance::run(&harness) from a test. Navree's own transports pass it. Install the transport with Transport::custom:

Navree::app().module(MicroModule::for_root(Transport::custom(SqsTransport::new(queue_url))))

Testing

Transport::memory() is a broker inside the process: requests, events, consumer groups, retries and dead letters behave as they do on a real broker, without the network, and a group that has subscribed once keeps its events while none of its instances listens. MicroFake wraps it with stubs and assertions, and is the usual way to test a service; see request-response and events.

To test against a real broker, run it in CI the way navree dock up does locally, point APP_MICRO__URL at it, and run micro:setup before the tests.