Durable fan-out publish/subscribe with consumer groups, a retained message log, backfill, and replay

Pub/Sub

Working example → Order events fanned out to independent billing, fulfillment, and analytics groups, with replay and pause/resume: apps/backend/src/triggers/pubsub.ts

Topic() defines a durable publish/subscribe channel. Subscription() attaches a consumer group to it. Every group receives every message; within a group, deliveries are load-balanced across that group's nodes with the queue pipeline's full at-least-once guarantees — claims, heartbeats, retries with backoff, and dead-letter queues, all inherited for free. A retained message log backs late-joining consumers (backfill) and manual replay.

This is fan-out, not a Kafka-style offset log: there's no partitioning or consumer-managed offsets. Each group's progress is tracked by the same durable job pipeline every other module uses.

Quick start

events/orders.ts

import { Topic, Subscription } from "oqronkit";

interface OrderPlaced {
  orderId: string;
  amount: number;
}

export const orderEvents = Topic<OrderPlaced>({
  name: "order-events",
  retention: { maxAgeMs: 30 * 86_400_000, maxCount: 500_000 }, // 30 days
});

// Each group below gets EVERY published order, independently.
Subscription<OrderPlaced>({
  topic: "order-events",
  group: "billing",
  retries: { max: 3, strategy: "exponential", baseDelay: 2_000 },
  deadLetter: { enabled: true },
  handler: async (ctx) => {
    await chargeCard(ctx.data.orderId, ctx.data.amount);
  },
});

Subscription<OrderPlaced>({
  topic: "order-events",
  group: "fulfillment",
  handler: async (ctx) => {
    await createShipment(ctx.data.orderId);
  },
});

routes/checkout.ts

// One publish → one durable log entry + one delivery job per group
const messageId = await orderEvents.publish(
  { orderId: order.id, amount: order.total },
  { idempotencyKey: `order-${order.id}` }, // republish converges, per group
);

`publish()` options

Option	Type	Purpose
`idempotencyKey`	`string`	Becomes the message id — republishing with the same key converges to ONE delivery per group
`headers`	`Record<string, string>`	Arbitrary metadata, carried through to every group's `ctx.headers`
`correlationId`	`string`	Carried through to `ctx.correlationId` — thread related messages together
`delayMs`	`number`	Delay this publish's deliveries by N ms (applied per group, at the broker)
`expiresAt`	`Date \| number`	Deliveries claimed after this time are silently acked (skipped), never run

Consumer groups

A group is a named set of nodes that share one logical subscription — declare the same (topic, group) pair on multiple nodes and they load-balance that group's deliveries, exactly like Worker() nodes sharing a topic. Different groups never compete with each other:

publish('order-events', {...})
  ├─ billing      (3 nodes, load-balanced) → 1 delivery job, claimed by whichever node polls first
  ├─ fulfillment  (1 node)                 → 1 delivery job
  └─ analytics    (1 node)                 → 1 delivery job

Three independent deliveries. A retrying/dead billing message never blocks fulfillment or analytics.

With an explicit idempotencyKey, every group still gets its own deduplicated delivery (messageId:group) — republishing the same event never double-delivers, and never collapses the fan-out.

Topic and group names can't contain : — it separates the internal delivery queue's segments (topic:<topic>:<group>, a reserved prefix no Queue/Worker/Webhook may use).

Delivery lifecycle

publish(msg)
  ├─ write the durable message log (topic_msg:<topic>) — durable-first, before any delivery exists
  ├─ for every registered group: enqueue ONE deduplicated delivery job
  └─ emit pubsub:message:published

  poll ─► atomic claim ─► expired? ─► ack, skip
             │
             ├─ filter(data) === false? ─► ack, skip (handler never runs)
             │
             ├─ emit pubsub:delivery:claimed
             ├─ run the handler
             │
             ├─ success ─► completed · emit pubsub:delivery:acked
             └─ throw    ─► retry with backoff, or dead-letter when exhausted
                            (emits pubsub:delivery:dead)

  every attempt = a run (module: 'topic', name: 'topic:<topic>:<group>')

Publishing is durable-first: the message log write must succeed before any delivery job exists, so a crash between "published" and "delivered" can never lose the message — only delay it (the boot reconciler recovers orphaned deliveries and emits pubsub:reconciliation:repaired).

Pausing a group pauses consumption only — publishes keep landing in the log and new delivery jobs keep queuing; resumeGroup drains the backlog.

Handler context (`ctx`)

Property	Type	Description
`ctx.id`	`string`	Message id — the publish's `idempotencyKey`, or an auto-generated one
`ctx.topic`	`string`	Topic name
`ctx.group`	`string`	This subscription's consumer group
`ctx.data`	`T`	Typed message payload
`ctx.headers`	`Record<string, string> \| undefined`	Headers passed to `publish()`
`ctx.correlationId`	`string \| undefined`	Correlation id passed to `publish()`
`ctx.publishedAt`	`Date`	When the message was published
`ctx.attempt`	`number`	Current delivery attempt (1-based)
`ctx.maxAttempts`	`number`	Max attempts configured for this group
`ctx.duration`	`number`	Live elapsed ms since this attempt started
`ctx.signal`	`AbortSignal`	Fires on timeout or cancellation
`ctx.aborted`	`boolean`	Shorthand for `signal.aborted`
`ctx.log(level, msg)` / `ctx.log.info/warn/error(msg)`	`Function`	Structured logging — recorded on the run

Backfill for new groups

A group joining after messages were already published can catch up on the retained log — a one-time replay guarded by a distributed lock, so multiple nodes registering the same new group don't double-backfill:

Subscription<OrderPlaced>({
  topic: "order-events",
  group: "analytics", // added months after order-events went live
  startFrom: "earliest", // or a Date / epoch ms — "everything since then"
  handler: async (ctx) => {
    /* ... */
  },
});

`startFrom`	Behavior
`'latest'` (default)	New group sees only messages published after it registers
`'earliest'`	One-time backfill of the entire retained log
`Date` / epoch `number`	One-time backfill of messages published at or after that point

Backfill runs once per group (tracked on its control record); it never re-runs on later restarts.

Replay

Re-deliver a range on demand — useful after fixing a consumer bug, without touching the other groups:

// Re-deliver everything since a given time to ONE group
const count = await orderEvents.replay("billing", {
  from: lastKnownGoodDeploy,
});

Each replay() call is a fresh delivery generation (its own dedupe token) — it's intentional re-delivery, not blocked by the original publish's dedupe ids, so replaying twice delivers twice.

Retention

The message log is pruned on an internal 5-minute sweep, guarded by a distributed lock so only one node prunes a given topic at a time:

Topic({
  name: "order-events",
  retention: {
    maxAgeMs: 7 * 86_400_000, // default: 7 days
    maxCount: 100_000, // default: trims the OLDEST entries beyond this
  },
});

The log shares the append-heavy, bulk-pruned runs storage class — see Storage Model.

Inspecting the log

const recent = await orderEvents.getMessages({ limit: 50 }); // newest first
const since = await orderEvents.getMessages({ from: someDate });
const groups = await orderEvents.listGroups(); // control records
// [{ group: 'billing', status: 'active', startFrom: 'latest', backfilled: true, createdAt }, ...]

Reliability semantics

Situation	Outcome
Node crashes mid-delivery	Claim TTL expires → another node retries (at-least-once)
Node crashes right after `publish()`	Log write was durable-first → boot reconciler re-enqueues any orphaned delivery
One group's handler keeps failing	Retries with backoff, then dead-letters — other groups are unaffected
Same message published twice with `idempotencyKey`	Deduplicated per group — one delivery each
A group is added after the topic has history	`startFrom: 'earliest'`/a date backfills it once
A message expires before a group consumes it	Silently acked (skipped), never delivered

Control & environments

await orderEvents.pauseGroup("fulfillment"); // consumption pause; publishes keep buffering
await orderEvents.resumeGroup("fulfillment"); // drains the backlog

// Gate a topic or a specific group's node independently
Topic({ name: "order-events", environments: ["production"] });
Subscription({
  topic: "order-events",
  group: "fulfillment",
  environments: ["production", "staging"],
});

A production-only Topic is inert elsewhere (publish() throws); a gated Subscription never registers its group or consumes on that node — see Environments & Microservices.

Configuration reference

Topic()

Option	Default	Purpose
`name`	—	Unique topic name (no `:`)
`retention.maxAgeMs`	7 days	Age-based log prune
`retention.maxCount`	`100_000`	Count-based log trim (oldest first)
`validate`	—	`(message) => boolean \| string` — reject at publish time
`environments`	all	Environment allow-list
`enabled`	`true`	Definition-level switch
`version`	`0`	Definition version (downgrade protection)

Subscription()

Option	Default	Purpose
`topic` / `group`	—	Which topic, which consumer group (no `:` in either)
`startFrom`	`'latest'`	New-group backfill: `'latest'` \| `'earliest'` \| `Date`/epoch
`filter`	—	`(data) => boolean` — `false` acks without invoking the handler
`concurrency`	`5`	Parallel deliveries per node
`retries` / `timeout`	—	Same shape as Task Queue
`deadLetter.onDead`	—	Receives the `{ messageId, topicName, group, payload, ... }` delivery, not a full job record
`guaranteedWorker`	`true`	Heartbeat extends the claim while the handler runs
`pollIntervalMs`	`500`	Poll cadence for this group
`environments`	all	Environment allow-list for this group only

`ITopic` instance methods

Everything Topic() returns, in one place:

Method	Returns	Description
`publish(message, opts?)`	`Promise<string>` (message id)	Durable-first: log write, then one deduplicated delivery job per registered group
`replay(group, opts?)`	`Promise<number>` (messages enqueued)	Re-deliver a log range to ONE group — a fresh delivery generation, not blocked by the original publish's dedupe
`pauseGroup(group)`	`Promise<void>`	Pause consumption for a group — publishes keep buffering
`resumeGroup(group)`	`Promise<void>`	Resume consumption — drains the backlog
`listGroups()`	`Promise<TopicGroupRecord[]>`	Every registered group's control record (`status`, `startFrom`, `backfilled`, `createdAt`)
`getMessages(opts?)`	`Promise<TopicMessageRecord[]>`	Read the retained log — newest first by default, or `{ from, limit }`

Next Steps

Distributed Worker — the single-consumer analog Topic's delivery pipeline is built on
Crash Safety — the at-least-once pipeline deliveries ride on
Storage Model — where the message log and group state live

Pub/Sub

On this page