← Use cases

Subscription replay — rewind a stream to any timestamp.

Your consumer had a bug last Tuesday. You shipped a fix today. You need to reprocess every message since the bug went live — without disturbing the production consumer that's still running. With Queen this is one line of configuration: spawn a new consumer group with subscriptionFrom set to the moment the bug went live. The new group reads from that exact timestamp; the production group's offset is untouched.

Numbers are from a verified run on a single Mac with one PostgreSQL 18 container. Reproduce with examples/use-case-replay.js.

1. The problem

The "I need to reprocess events from a point in time" problem comes up constantly:

• A consumer had a bug for the last 18 hours; you deployed a fix and need to re-run every message since the bug landed.
• A downstream system was down for 2 hours; you need to replay only that window into a new pipeline.
• You added a new analytics consumer and want it to backfill the last week of traffic before going live.
• You're debugging in staging by replaying production traffic from a specific incident timestamp.

How the major alternatives handle it:

2. The pattern

Queen stores each consumer group's progress as a row in queen.partition_consumers keyed by (partition_id, consumer_group). To replay, you spawn a brand-new consumer group that's never been seen before. The first pop registers it in queen.consumer_groups_metadata with the subscription anchor — either 'all', 'new', or a specific ISO timestamp.

Because the two consumer groups have separate offset rows, they read the same partitions completely independently. The production worker doesn't pause, restart, or notice anything. The replay worker is isolated by construction.

3. Quickstart

Minimal reproducible setup: push 1,500 events with a deploy boundary marked between idx 499 and idx 500. One CG reads everything; a new CG spawned afterwards reads only what came at or after the boundary.

// npm install queen-mq        (Node 22+)
import { Queen } from 'queen-mq'

const url = 'http://localhost:6632'
const q   = new Queen({ url, handleSignals: false })

const QUEUE = 'order-events'

await q.queue(QUEUE).config({
  leaseTime:                 30,
  retryLimit:                3,
  retentionEnabled:          true,
  retentionSeconds:          3600,     // keep history for 1 hour
  completedRetentionSeconds: 600
}).create()

// Production consumer — sees every message ever pushed.
const ac = new AbortController()
;(async () => {
  await q.queue(QUEUE)
    .group('prod')
    .batch(50)
    .each()
    .subscriptionMode('all')   // ★ first pop registers prod with mode='all'
    .partitions(8)
    .consume(async (msg) => {
      // handle in production
    }, { signal: ac.signal })
})()

// Producer (interleaved with prod consumer running):
//   - push 500 events  ───►  sleep 800 ms  ───►  capture deployAt  ───►  sleep 800 ms
//   - push 500 more events
//   - sleep 600 ms
//   - push 500 more events

Now the deploy fix has shipped and we want to reprocess every event created at or after deployAt. We spawn a new consumer group with that timestamp:

// New CG, never seen before. First pop locks in the subscription anchor.
const reprocessAc = new AbortController()
;(async () => {
  await q.queue(QUEUE)
    .group('reprocess')
    .batch(50)
    .each()
    .subscriptionFrom(deployAt)   // ★ ISO timestamp — picks up the 'timestamp' mode
    .partitions(8)
    .consume(async (msg) => {
      // re-run your fixed pipeline for this message
    }, { signal: reprocessAc.signal })
})()

// Verified run (1,500 events total, deploy boundary at idx 500):
//   'prod' CG received:       1500  (expected 1500)
//   'reprocess' CG received:  1000  (expected ≈ 1000)
//   smallest idx in reprocess: 500  (matches deploy boundary)

Note that 'prod' and 'reprocess' consume the same queue, on the same broker, against the same database — there is no contention or impact between them. The same pattern works at any scale: the reprocess CG can be a separate Pod, a separate cluster, a notebook, or a CLI invocation.

4. Cleaning up a consumer group when you're done

Once the reprocess completes, you typically want to delete the one-shot CG so its registration doesn't linger. Use the admin API:

// Delete CG state + subscription metadata for one queue:
await q.admin.deleteConsumerGroupForQueue(
  'reprocess',     // consumer group
  'order-events',  // queue
  true             // deleteMetadata: also drop the subscription_timestamp row
)

queue.delete() intentionally does not cascade to queen.consumer_groups_metadata, because consumer groups subscribe across multiple queues (and across namespace/task selectors). To wipe the metadata too, pass deleteMetadata: true. This is the same pattern as queenctl cg delete --queue ....

5. The three subscription modes

The mode is locked in on the consumer group's first pop. Subsequent pops by the same CG honour the recorded anchor — they ignore any subscriptionMode / subscriptionFrom you pass. To change the anchor, you either rename the CG, or delete it via the admin API (see §4) and let the next pop re-register it.

6. Why the two CGs don't interfere

Per-CG isolation is enforced at the schema level. Every (partition_id, consumer_group) pair gets its own row in queen.partition_consumers holding that CG's last_consumed_id. The pop stored procedure (pop_unified_batch_v4) filters by that row alone — no cross-CG read or write happens anywhere on the hot path.

SELECT consumer_group,
       count(*)                AS partitions_consumed,
       sum(total_messages_consumed) AS lifetime_messages,
       max(last_consumed_at)   AS last_seen
FROM   queen.partition_consumers pc
JOIN   queen.partitions p   ON p.id = pc.partition_id
JOIN   queen.queues     q   ON q.id = p.queue_id
WHERE  q.name = 'order-events'
GROUP  BY consumer_group;

You'll see one row per CG. Each one advances independently as that CG processes messages. There's no shared "topic offset" that one CG can clobber on another.

7. Common pitfalls

8. References

← Back to use cases