49. Bounded Brokered Messaging

Important

Dependency examples use 12.0.0-rc.1, the planned first v12 RC. Do not request it from crates.io before it is published; use path dependencies from this source checkout during development.

This tutorial starts with rullst-messaging’s deterministic in-memory broker and then switches the same trait contract to the durable local SQLite adapter. Remote broker protocols remain separate adapters.

1. Enable the umbrella feature

[dependencies]
rullst = { version = "12.0.0-rc.1", features = ["messaging"] }

Use messaging-sqlite instead of messaging for durable local state. A direct dependency uses rullst-messaging = { version = "12.0.0-rc.1", features = ["sqlite"] }.

2. Create a bounded broker

#![allow(unused)]
fn main() {
use rullst::messaging::{BrokerConfig, InMemoryBroker};

fn create_broker() -> Result<(), rullst::messaging::MessagingError> {
let config = BrokerConfig::try_new("billing")?
    .with_limits(
        10_000,        // retained messages
        128,           // consumer-group subscriptions
        5,             // delivery attempts
        1024 * 1024,   // payload bytes
    )?;
let broker = InMemoryBroker::new(config);
let _ = broker;
Ok(())
}
}

The hard ceilings prevent an accidental configuration from turning the local broker into unbounded memory state. Capacity exhaustion fails closed until terminal messages are explicitly purged.

Persist the same contract locally

#![allow(unused)]
fn main() {
use rullst::messaging::{BrokerConfig, SqliteBroker};

async fn open() -> Result<(), rullst::messaging::MessagingError> {
let config = BrokerConfig::try_new("billing")?
    .with_limits(10_000, 128, 5, 1024 * 1024)?;
let broker = SqliteBroker::connect("sqlite://storage/messages.sqlite", config).await?;
let _ = broker;
Ok(())
}
}

All mutations use serialized SQLite write transactions. A committed publish or ACK survives restart, while an uncommitted operation rolls back. In-flight leases also survive; after expiry the next broker operation requeues or dead-letters them. Multiple processes may share one file and namespace, but reopening it with different limits is rejected. This is local durability, not network replication, automatic failover or exactly-once side effects.

Protect header values and payloads

SqliteBroker::connect deliberately retains the compatible plaintext profile. For a new namespace/database, load a 32-byte high-entropy key from a secret manager and select the encrypted profile explicitly:

#![allow(unused)]
fn main() {
use rullst::messaging::{
    BrokerConfig, MessagingKeyring, MessagingStorageKey, SqliteBroker,
};

async fn open_encrypted(
    secret_manager_key: [u8; 32],
) -> Result<(), rullst::messaging::MessagingError> {
let primary = MessagingStorageKey::try_new("messages-2026-09", secret_manager_key)?;
let keyring = MessagingKeyring::new(primary);
let broker = SqliteBroker::connect_encrypted(
    "sqlite://storage/messages.sqlite",
    BrokerConfig::try_new("billing")?,
    keyring,
).await?;
let _ = broker;
Ok(())
}
}

To rotate, put the new primary first and append the old decryption key:

#![allow(unused)]
fn main() {
use rullst::messaging::{MessagingKeyring, MessagingStorageKey};

fn rotate(new_key: [u8; 32], old_key: [u8; 32])
    -> Result<MessagingKeyring, rullst::messaging::MessagingError> {
let keys = MessagingKeyring::new(MessagingStorageKey::try_new(
    "messages-2026-10",
    new_key,
)?)
.with_decryption_key(MessagingStorageKey::try_new(
    "messages-2026-09",
    old_key,
)?)?;
Ok(keys)
}
}

New publications use the primary. Old records are not silently rewritten and their keys remain mandatory until those messages become terminal and are purged. Plaintext/encrypted profiles cannot be mixed or changed in place; move through a new namespace/database with an application-reviewed republish procedure. Passwords are not AES keys, and source literals are unsuitable for production key custody.

The encrypted profile protects header values and payloads. Topic, event and content type, message/key IDs, timestamps, idempotency key, fingerprint and delivery state remain visible metadata. Protect the database file and backups, control rollback, rehearse key recovery and authorize topic access separately.

3. Register a consumer group and publish

#![allow(unused)]
fn main() {
use rullst::messaging::{
    MessageBroker, PublishRequest, StartPosition, SubscriptionRequest, TraceContext,
};

async fn example(
    broker: &rullst::messaging::InMemoryBroker,
) -> Result<(), rullst::messaging::MessagingError> {
broker
    .subscribe(SubscriptionRequest::try_new(
        "invoices",
        "receipt-mailers",
        StartPosition::Earliest,
    )?)
    .await?;

let trace = TraceContext::try_with_state(
    "00-4bf92f3577b34da6a3ce929d0e0e4736-00f067aa0ba902b7-01",
    "vendor=value",
)?;
let request = PublishRequest::try_new(
    "invoices",
    "invoice.paid",
    "invoice/2026-0042/paid/v1",
    br#"{"invoice_id":"2026-0042"}"#.to_vec(),
)?
.with_content_type("application/json")?
.with_trace_context(&trace)?;

let first = broker.publish(request.clone()).await?;
let replay = broker.publish(request).await?;
assert_eq!(first.id(), replay.id());
assert!(replay.is_duplicate());
Ok(())
}
}

Idempotency is scoped to the topic. The same topic/key with different content returns MessagingError::IdempotencyConflict; it never silently overwrites the original publication.

Only validated W3C version-00 traceparent and a conservative tracestate subset are copied. baggage is deliberately excluded because arbitrary application metadata may contain credentials or personal data. Sampling, export, retention and tenant-aware correlation remain host responsibilities.

4. Receive, acknowledge, retry, or dead-letter

#![allow(unused)]
fn main() {
use rullst::messaging::{
    FailureCode, MessageBroker, ReceiveRequest, RetryDisposition,
};
use std::time::Duration;

async fn consume(
    broker: &rullst::messaging::InMemoryBroker,
) -> Result<(), rullst::messaging::MessagingError> {
let request = ReceiveRequest::try_new(
    "invoices",
    "receipt-mailers",
    "worker-1",
    10,
    Duration::from_secs(30),
)?;

for delivery in broker.receive(request).await? {
    let effect_succeeded = true;
    if effect_succeeded {
        broker.ack(delivery.ack_token()).await?;
    } else {
        let disposition = broker
            .retry(
                delivery.ack_token(),
                Duration::from_secs(15),
                FailureCode::try_new("mail.transient")?,
            )
            .await?;
        if disposition == RetryDisposition::DeadLettered {
            // Alert through an application-owned operational channel.
        }
    }
}
Ok(())
}
}

If an adapter needs a stable broker-neutral representation, encode the received envelope explicitly:

#![allow(unused)]
fn main() {
use rullst::messaging::{BrokerConfig, MessageEnvelope, WireEnvelopeCodec};

fn frame(
    envelope: &MessageEnvelope,
) -> Result<(), rullst::messaging::MessagingError> {
let config = BrokerConfig::try_new("billing")?;
let bytes = WireEnvelopeCodec::encode(envelope, &config)?;
let decoded = WireEnvelopeCodec::decode(&bytes, &config)?;
assert_eq!(&decoded, envelope);
Ok(())
}
}

The v1 frame is canonical, bounded by BrokerConfig, namespace-bound and rejects unknown versions, truncation, trailing bytes and non-canonical fields. It carries an accepted envelope only. It does not retain the caller’s publish idempotency key, connect to a broker, or map remote ACK/retention semantics, so it cannot be presented as a Kafka/NATS/RabbitMQ adapter by itself.

An ACK token is an opaque, single-use capability bound to one group and one lease. Once it expires, the message is requeued or dead-lettered at the attempt ceiling. Do not serialize tokens into logs or application records.

5. Design for at-least-once delivery

The consumer may complete the mail/payment/webhook effect and stop before the ACK reaches the broker. Therefore:

1. use delivery.envelope().id() as a stable deduplication key;
2. claim that key transactionally in the side-effect system where possible;
3. ACK only after the effect is durably accepted;
4. treat retries and dead letters as normal operational states;
5. authorize topics and tenant scope in the host application.

The in-memory broker loses state on process exit. The SQLite adapter retains state; its default profile is plaintext, while the explicit encrypted profile protects header values and payloads but not routing/idempotency metadata. The host owns key custody, file permissions, protected backup/restore, rollback detection, retention, disk monitoring and topic/tenant authorization. A future remote adapter is supported only after it passes the shared contract plus its own protocol, restart, and fault matrix; an adapter name alone is not durability evidence.

6. Relay a relational outbox after commit

Enable messaging-orm-outbox on the umbrella crate. Commit the domain mutation and rullst_orm::Outbox::enqueue together as described in the transactional outbox tutorial. Then bind one exact outbox stream to one broker topic:

#![allow(unused)]
fn main() {
use rullst::messaging::{
    InMemoryBroker, MessagingError, OrmOutboxRelay,
};

fn configure_relay(
    broker: InMemoryBroker,
) -> Result<OrmOutboxRelay<InMemoryBroker>, MessagingError> {
    OrmOutboxRelay::try_new("tenant-a", "domain-events", broker)
}
}

A supervised worker claims and relays one event:

#![allow(unused)]
fn main() {
use rullst::messaging::{InMemoryBroker, OrmOutboxRelay};
use rullst_orm::Outbox;

async fn relay_one(relay: &OrmOutboxRelay<InMemoryBroker>) {
    let claim = match Outbox::claim_next("tenant-a", "worker-a", 30, 8).await {
        Ok(Some(claim)) => claim,
        Ok(None) | Err(_) => return,
    };
    match relay.relay_and_ack(claim).await {
        Ok(receipt) if !receipt.outbox_acknowledged() => {
            // The lease expired; a later claim will replay the same event key.
        }
        Ok(_) => {}
        Err(error) => {
            // Record a secret-free operational failure and let the lease expire.
            let _accepted_before_ack_failure = error.accepted_publication();
        }
    }
}
}

The ordering is intentionally publish then ACK. A stop in between causes a new ORM claim to publish the same event key and content; the broker returns its original ID as a duplicate. This is at-least-once relay with bounded idempotency, not an atomic transaction across two systems. Keep the worker supervised and make the final consumer idempotent too.