Skip to content

Pub/Sub Delivery and Scaling

How a published event reaches a worker, where consumer groups exist, and what that means when you add replicas.

The delivery tree

flowchart TB
    S[(Redis stream<br/>one per topic)]
    S --> A1["api pod 1<br/>consumer group = pod name"]
    S --> A2["api pod 2<br/>consumer group = pod name"]
    S --> A3["api pod 3<br/>consumer group = pod name"]
    A1 -->|in-process fan-out| R1["Runtime pod A<br/>one Subscribe stream per topic"]
    A2 -->|in-process fan-out| R2["Runtime pod B"]
    A1 -->|in-process fan-out| R3["Runtime pod C"]
    R1 --> I1["worker 1<br/>inbox per topic"]
    R1 --> I2["worker 2<br/>inbox per topic"]
    R3 --> I3["worker 3<br/>inbox per topic"]
    I1 --> O1["worker 1 outbox"]
    I2 --> O2["worker 2 outbox"]
    I3 --> O3["worker 3 outbox"]
    O1 --> P["PublishEvents"]
    O2 --> P
    O3 --> P

Four hops, each with its own unit of consumption:

Hop Unit Delivery semantics
Redis → daprd one consumer group per API pod per topic (consumerID: {podName}) Competing inside the group, broadcast across groups: every API pod receives every message
API process → gRPC subscribers one Dapr subscription per topic per API process, shared by every Pubsub.Subscribe caller In-process fan-out: every subscriber receives every message
Runtime pod → workers one gRPC subscription per topic per Runtime pod Every local worker registered for the topic receives every message
Worker one Dapr actor, one inbox per topic One actor turn at a time; order preserved within a topic

The WorkManager and its Runtime do not have consumer groups of their own. Only the API pods hold Dapr/Redis consumer groups; everything below them is fan-out.

Consumer groups

  • Groups are per API pod, not per Runtime pod or per worker. The count is roughly API replicas × topics with at least one subscriber.
  • A group is created lazily with the first subscriber for a topic and is not deleted when the subscription closes. Dapr's Redis component never removes it, so every rollout leaves groups named after pods that no longer exist; the deployment role removes them (see consumer_groups.yml in virtufin/k3s-common).
  • Because every API pod has its own group, the group's "competing consumer" behavior stays inside one pod. Across pods the effect is broadcast — which is what makes every Runtime pod see everything.

Acknowledgment: the broker is not a retry mechanism

TopicDaprSubscriptionRegistry.Deliver writes the event into every subscriber's channel and returns success. Dapr acknowledges the group entry right after that, before any worker processes it. Consequences:

  • a live pending entry exists for microseconds;
  • a worker failure is not redelivered by the broker; the Runtime publishes it once to dlq.<topic> (see Worker Lifecycle);
  • a handler-level fault returns Retry: Dapr retries up to maxRedeliveryCount (3, redeliveryIntervalInSeconds: 1), then dead-letters to dlq.pubsub;
  • a subscriber whose channel is full logs Failed to enqueue and is reaped by the subscription health sweep (one minute); its Runtime reconnects.

No replay from the group: subscriptions start at the tail

Dapr would resume an existing group at its last-delivered-id, and create a missing group at the stream's start ("0"). Either way a new subscription would be fed old entries. Before opening a subscription, the gateway therefore

  1. creates its own group at $ (or moves it there with XGROUP SETID), and
  2. acknowledges every entry still pending for that group,

and refuses to subscribe if that fails rather than replaying silently. This is also why a worker that subscribes late does not receive history: catch-up is explicit, through the API's replay_since (bounded to 10 000 entries / 64 MB by default), which reads the stream without moving any consumer group. The Runtime uses it for per-worker cursors (Worker Lifecycle).

Operational consequences, all handled by the deployment:

  • Orphan groups after a rollout are removed by virtufin_backend's consumer-group cleanup.
  • Parked readers: a closed subscription's XREADGROUP ... BLOCK 0 reader can take one message into the pending list and never ack it. The gateway wakes parked sidecar readers (CLIENT UNBLOCK) when subscriptions close and acknowledges entries pending longer than five seconds, matching the no-catch-up policy.
  • Memory: streamTTL: 1m trims only when something publishes; a CronJob trims idle streams above a Redis memory threshold, never past what an active group still needs.

Scaling

Add What it adds What it costs
API replica fan-out headroom and another broker consumer group every message is delivered once more, per API pod
Runtime pod (one per node in the reference deployment) capacity to host workers, engines, inboxes no new consumer groups; one more gRPC subscriber per topic it hosts
WorkManager pod catalog/API throughput nothing on the pub/sub path

What does not scale: a single worker. One worker is one actor, one turn at a time — copying it does not share its load. And ordering is guaranteed only within a topic; two topics interleave freely.

To scale a hot stream, split it across topics (per symbol, per account, per shard) and run one worker per topic; Dapr placement spreads the workers across Runtime pods.

Identical workers: fan-out or exactly one?

The same worker code can be deployed in two shapes. The difference is only the topic plan.

Fan-out: copies on one topic

Three copies subscribe to one topic. One event is published; each copy processes it and publishes its own output. This is broadcast, by design: two deployments of the same package duplicate the work and its outputs.

Exactly one: one topic per copy

Three copies, each subscribed to its own topic, and one event published per topic. Each event is processed by exactly one worker. This is not load sharing of a single stream; it is partitioning, and the upstream publisher (or a router worker) has to decide the topic.

flowchart LR
    E1[event A] --> T1["x.part.0"] --> W1[worker 0]
    E2[event B] --> T2["x.part.1"] --> W2[worker 1]
    E3[event C] --> T3["x.part.2"] --> W3[worker 2]

There is no competing-consumer mode: you cannot have copies share one topic and have each message handled once. That behavior required worker groups and a distributed lock, both removed in 0.3.0.

Runnable example

examples/scaling_workers.py deploys both shapes against a live API and counts the markers each worker publishes:

pip install virtufin-api
VIRTUFIN_API_HOST=api.virtufin.com VIRTUFIN_API_PORT=443 VIRTUFIN_API_TLS=1 \
VIRTUFIN_API_KEY=... python3 examples/scaling_workers.py

Real output from the demo deployment (6 workers created and deleted by the script):

fan-out: 1 event -> 3 marker(s) from 3 worker(s) (fanout-0, fanout-1, fanout-2)
exactly-one: 3 events -> 3 marker(s); per-event counts [1, 1, 1]
  part-0 handled 01db1286, part-1 handled 1a7cc981, part-2 handled 338f80b1

fan-out: PASS
exactly-one: PASS

The script prints PASS/FAIL from the marker counts, so it doubles as a regression check for the delivery model.

A single worker still sees each message once

Within one worker, the actor instance is unique and the inbox is FIFO, so a message is processed once. The only repeats are deliberate: input or command cursors are batched (every few hundred milliseconds), so an event finished in the last batch before a crash is replayed to the next activation. Workers are expected to tolerate a repeated input.

Competing consumers (group)

A worker can join a group: several copies of it then split a topic instead of every copy processing every event. The default (no group) is unchanged: a dedicated worker receives every event, which is what the fan-out and partition examples above show.

Configuration

CreateWorkerRequest field Meaning
group Group name. Empty means dedicated (broadcast). Letters, digits, -, _, .; up to 64 characters.
group_size Number of members, 1..1024. Every live member must use the same value.
group_index This member's slot, 0 <= group_index < group_size. Must be unique among live members that share a topic.

All three are immutable after creation and are returned by ListWorkers.

The slot rule

An event belongs to slot hash(routing key) % group_size, and only the member whose group_index equals that slot processes it. The routing key is the CloudEvent subject when present, else its id:

  • with a subject (the platform's per-entity key), all events of one entity land on the same member, so per-entity order is preserved;
  • without one, the event id is used: each event is still handled exactly once, but one entity's events may be spread over different members.

The hash (FNV-1a over UTF-8 plus the Murmur3 avalanche step) is pinned by tests; changing it would remap every event.

Two consequences follow from "exactly one by construction":

  • there is no coordination: no claim, no lock, no per-message state write;
  • every member still receives every event (delivery stays a broadcast) and simply drops what it does not own before enqueueing it. The network cost is unchanged; the actor work is divided.

Command topics (wrk.<type>.command.<id>) bypass the split: a command addresses one worker, so the slot filter never applies to it.

Coverage and lifecycle

  • StartWorker refuses a duplicate (group, group_index) among running members that share a topic (FAILED_PRECONDITION).
  • Incomplete coverage is allowed (members start one by one) and logs a warning: group X on [topics] has N of M members running.
  • While a slot has no live member, the events for that slot are dropped by the other members — the same tail policy as any worker that is not subscribed. There is no failover and no work stealing: a slow member lags its slot.
  • Grouped workers do not use input cursors: a skipped event would pin a cursor, and a replay would re-enter the slot race. On a move they start at the tail, exactly like replay=off. Commands keep their own cursor.
  • group_size and group_index cannot be changed in place. To reshard, stop the group and recreate its members with the new size (a short gap), or give the new generation a different group name (e.g. pairs-v2) so old and new never process the same slot at once.

Example

The runnable example deploys all three shapes against one API:

pip install virtufin-api
VIRTUFIN_API_HOST=api.virtufin.com VIRTUFIN_API_PORT=443 VIRTUFIN_API_TLS=1 \
VIRTUFIN_API_KEY=... python3 examples/scaling_workers.py

Phase 3 creates three workers with group="scaling-demo", group_size=3, group_index=0..2 on one topic and publishes three events:

fan-out: 1 event -> 3 marker(s) from 3 worker(s) (fanout-0, fanout-1, fanout-2)
exactly-one: 3 events -> 3 marker(s); per-event counts [1, 1, 1]
group: 3 events -> 3 marker(s); per-event counts [1, 1, 1]
  part-0 handled ..., part-1 handled ..., part-2 handled ...
  scaling-demo/0 handled ..., scaling-demo/1 handled ..., scaling-demo/2 handled ...

The group phase prints one marker per event (never one per member), while the fan-out phase prints three markers for a single event.

Inspecting it on a cluster

Against the Redis primary (namespace redis in the reference deployment):

# one group per API pod per topic, with its last-delivered id and pending count
redis-cli XINFO GROUPS <topic>

# anything a closed subscription's parked reader left behind
redis-cli XPENDING <topic> <api-pod-name>

The worker lifecycle event stream (sys.workmanager.worker.lifecycle) carries worker_id and topic, which is the quickest way to see which workers a topic reaches.

History: worker groups

Before 0.3.0 a worker could carry a group, and members of a group coordinated through a distributed Redis lock (GroupLockOptions, AcquireGroupLockAsync, the VIRTUFIN_WORKER_GROUP environment variable) so that exactly one member processed each message. That lock — and its failure modes — was removed in 0.3.0, and the old group proto fields (4 and 5) were retired.

Competing consumers came back in 0.39.0 as the deterministic group / group_size / group_index slotting described above: the same externally visible outcome (each message handled by one member) without a coordinator.