Broadcast — per-instance fan-out

A consumer group splits a topic across the instances of a service: three replicas, and each message goes to exactly one of them. That is the right shape for work.

It is the wrong shape for a signal. When a cache entry goes stale, every replica holds its own copy and every replica has to hear about it. Delivering the invalidation to one of three replicas leaves two serving stale data.

.broadcast() is the other shape. Every instance subscribes as itself, and every instance receives every message.

use shove::{TopologyBuilder, define_topic};
 
define_topic!(pub CacheInvalidations, InvalidateKey,
    TopologyBuilder::new("cache-invalidations")
        .broadcast()
        .build()
);

Consume it through broadcast_subscriber() rather than consumer_group():

let broker = Broker::<InMemory>::new(config).await?;
let mut subscriber = broker.broadcast_subscriber();
 
subscriber.subscribe::<CacheInvalidations, _>(EvictHandler, ConsumerOptions::new())?;
subscriber.run_until_timeout(shutdown_signal(), Duration::from_secs(5)).await;

Best-effort is the contract, not a limitation

Broadcast is deliberately lossy. Each constraint below exists because the alternative is a durable per-instance group — which is just N consumer groups plus a lifecycle problem, and which shove already supports if that is what you need.

Deliver-new only. An instance receives what is published while it is subscribed. Nothing is replayed for an instance that was down, and nothing accumulates for one that has not started. This is correct for invalidation — a process that just booted has an empty cache, so there is nothing stale to invalidate — and it is what removes the need to store a per-instance offset anywhere.

No retry chain. .dlq(), .dlq_named(), .hold_queue() and .sequenced() each panic in build() when combined with .broadcast(). (.for_consumer_group() panics too — broadcast is already a per-process group, so a named one would name nothing.) This is not an oversight to be filled in later. Redelivery to one subscriber of a fan-out is not expressible on most of these brokers, and a shared DLQ would collect one copy of every failure per subscriber — the failure of a single message would look like N failures.

So .broadcast() implies allow_message_loss, and the retry budget is pinned to zero no matter what ConsumerOptions::with_max_retries says:

That is the same terminal path a topology with no DLQ already takes, so an existing data-loss alert on shove_messages_discarded_total covers broadcast without changes. If losing a message on handler failure is unacceptable for your use case, broadcast is the wrong primitive — use a consumer group per reader with for_consumer_group.

Defer deserves one warning. A broadcast subscription is a single delivery loop, so a deferred message is retried in place, holding up the messages behind it — and because Defer never spends the retry budget, a handler that always defers stalls the subscription indefinitely. There is nowhere else to put the message: no second consumer to hand the backlog to, and no DLQ or hold queue to park it in. Use Defer on a broadcast topic only for a condition you expect to clear on its own.

One consumer per instance. There is no autoscaling knob on BroadcastSubscriber, because a second consumer inside the same process would split the fan-out rather than duplicate it — competing consumption sneaking back in through the wrong door. Registering a .broadcast() topology on consumer_group() or consumer_supervisor() returns a ShoveError::Topology for the same reason.

Ephemeral identity. The subscription is created when the delivery loop starts and destroyed when it ends, including when the process dies without draining — on NATS a hard kill leaves the consumer for the server's inactivity threshold and no longer; on RabbitMQ the queue is exclusive and auto_delete, so the broker removes it when the connection goes, drained or not; and on Redis and Kafka there is no per-subscriber state to leave in the first place. Nothing accumulates per pod restart. This is the property that the obvious workarounds — a per-pod UUID consumer group on Kafka or Redis — fail: each of those leaves broker-side state behind on every restart, and nothing reaps it.

A broker outage is a gap, not a stop. A broadcast subscription reconnects with backoff like any other consumer, rather than resolving with the error. That is deliberate: BroadcastSubscriber reports a task's error only through SupervisorOutcome, and only once run_until_timeout returns — so a subscription that gave up on the first blip would look healthy while receiving nothing for the rest of the process's life, which is a far worse failure than missing a window. On NATS the reconnect creates a fresh ephemeral consumer at DeliverPolicy::New, so whatever was published during the outage is not replayed; the previous consumer is deleted rather than left to accumulate. On Redis the read position is carried across the reconnect instead of resetting to $, so entries that arrived during the outage are still delivered — provided they survived Redis persistence and the stream's MAXLEN bound. Kafka and RabbitMQ behave like NATS rather than like Redis: Kafka re-assigns at the then-current tail, and a RabbitMQ reconnect declares a fresh server-named queue and rebinds it, so anything the fanout exchange had nowhere to route during the window is already gone.

Per-backend semantics

The contract above is backend-independent. How each backend delivers it is not: one of them changes how the stream is declared, and another changes the publisher's behaviour.

Every backend except SQS implements broadcast on this version, so there is no longer a planned row: broadcast_subscriber() compiles everywhere but Broker<Sqs>, where the gate is permanent rather than pending. Four of the rows carry a caveat that affects how you design around them — the NATS stream declaration, the Redis stream bound, the RabbitMQ publisher route and the Kafka group.id — and each has its own section below.

SQS is a compile error

Broker<Sqs> has no broadcast_subscriber() method at all:

let broker = Broker::<Sqs>::new(config).await?;
let _ = broker.broadcast_subscriber();
// error: no method named `broadcast_subscriber` found for struct `Broker<Sqs>`
// note: `Sqs` has no ephemeral per-instance subscription primitive

The gate is the HasBroadcast capability trait, which Sqs does not implement — the same mechanism that keeps consumer_group() off Broker<Sqs>. Per-instance fan-out on SQS would mean creating and deleting a real queue plus an SNS subscription for every process, and a queue leaked by a pod that died badly costs money for as long as nobody notices. If you need fan-out on AWS, publish to an SNS topic and have each instance manage its own subscription — that lifecycle is yours to own, and shove does not pretend otherwise.

NATS: a broadcast topology declares an Interest stream

NATS is the one backend where .broadcast() changes how the stream is declared. shove's default retention is WorkQueue, and JetStream refuses both halves of the ephemeral broadcast consumer on such a stream:

AckPolicy::None     -> "consumer in pull mode requires ack policy"        (10084)
DeliverPolicy::New  -> "consumer must be deliver all on workqueue stream" (10101)

So a .broadcast() topology is declared Interest instead. That is not a workaround but the closest match to the contract: with AckPolicy::None, interest is satisfied on delivery, so the stream retains nothing — neither messages already fanned out, nor messages published while no subscriber exists. This is why NATS needs no "bound the stream" caveat of the kind Redis does.

Two consequences:

• Pinning the retention is refused. nats_stream_config(NatsRetention::WorkQueue) on a broadcast topology fails at declare() with a ShoveError::Topology naming the conflict, rather than surfacing later as an opaque consumer-create error. Interest (the default) and Limits (bounded replay, at the cost of retaining messages nobody will read) both work.
• Retention is immutable on an existing stream. Adding .broadcast() to a topic that already has a WorkQueue stream makes declare() fail — JetStream cannot change a stream's retention policy. Recreate the stream, or use a new topic name.

Teardown deletes the ephemeral consumer explicitly when the delivery loop ends. An ephemeral consumer is not removed by the client disconnecting, so the server's inactivity threshold (30s) is only the backstop for a process that dies without draining.

Redis: bound the stream

Broadcast reads with XREAD, which never acknowledges and never trims. Nothing about a broadcast subscriber causes the stream to shrink. Declare the stream with a MAXLEN bound, or publish with XADD ... MAXLEN ~ N; otherwise it grows without limit and the memory cost is eventually the whole retention. Note that MAXLEN ~ trims at radix-node granularity, so the stream can sit somewhat above the bound — it is a cap on growth, not an exact length.

A broadcast topology on Redis declares nothing: no consumer group, and not even the stream (a blocking XREAD against a key that does not exist yet is valid and delivers the first entry written while it waits). That absence is the feature — a group created here would never be read from and nothing would ever reap it.

RabbitMQ: the publisher route changes

RabbitMQ is the one backend where broadcast changes the publish side as well as the consume side. A normal topology publishes to the default exchange with the queue name as routing key. A broadcast topology publishes to the {queue}-fanout exchange instead, and each instance binds its own exclusive auto-delete queue to it.

The consequence: a publisher and its subscribers must agree on the topology. Publishing a .broadcast() topic from a service still running an older build sends the message to the plain queue, where no broadcast subscriber is listening, and it is silently never delivered. Deploy the publisher and the subscribers together, or accept a gap.

declare_topology::<T>() declares the exchange and nothing else — no queue, because a broadcast topology has none. A publisher-only service still needs to call it, exactly as a sequenced topic's publisher needs the consistent-hash exchange declared. A subscriber declares the exchange itself before binding, so a subscriber-only process works either way.

Kafka: a groupless subscription, and one inert group.id

A Kafka broadcast subscriber assign()s every partition of the topic at its current end offset. It never calls subscribe() and never commits, so the broker receives no JoinGroup and no OffsetCommit: no group is created, nothing lands in __consumer_offsets, and kafka-consumer-groups --list shows nothing for the topic no matter how many times instances restart.

One wrinkle worth knowing about if you go looking at client config: librdkafka refuses assign() on a consumer handle with no group.id at all, so shove configures the fixed string {queue}-broadcast. It is inert — nothing joins under it and nothing commits to it — and it is deliberately not a per-process value, because a per-restart identifier showing up in client tooling is the same accumulating residue this feature exists to avoid.

Because the assignment is taken at the tail, a reconnect re-assigns at the then-current tail: messages published while a subscriber was disconnected are not replayed. That is deliver-new applied to the reconnect window rather than a separate caveat.

reset_consumer_group_offsets::<T>() returns a validation error for a broadcast topic. There is no group to re-anchor and no stored position to reset, and rewriting the {queue}-consumer offsets — which nothing on the broadcast path reads — would be an ops-facing call that succeeded at nothing.

Choosing between broadcast and a consumer group

The last row is deliberately not automated. It needs a stable per-instance identity and something to clean up the group when an instance is retired for good, and both of those are deployment decisions shove cannot make for you.