Pub/Sub

Topic subscriptions and fan-out publish.

See Pub/Sub for the full guide (delivery semantics, lifecycle, cross-SDK topics).

Method	Description
`subscriber(topic, name, fn, options?)`	Register `fn` as `name`, subscribed to `topic`. Returns the queue (chainable, like `task`).
`publish(topic, args?, options?)` → `Promise<Job[]>`	Fan a message out to every active subscription — one job each. Empty array when nothing is subscribed.
`declareSubscriptions()` → `Promise<void>`	Write pending durable subscriptions to storage. Call in a producer-only process; `runWorker()` does this automatically.
`unsubscribe(topic, name)` → `Promise<boolean>`	Remove a subscription.
`pauseSubscription(topic, name)` / `resumeSubscription(topic, name)` → `Promise<boolean>`	Stop/resume deliveries without unregistering.
`listSubscriptions(topic?)` → `Promise<Subscription[]>`	All subscriptions, or one topic's active ones.
`listTopics()` → `Promise<string[]>`	Distinct topics with at least one subscription.
`topicStats(topic?)` → `Promise<TopicStat[]>`	Backlog snapshot per subscription, optionally filtered to one topic: `topic`, `subscription`, `taskName`, `queue`, `active`, `durable`, `pending`, `running`, `dead`, `oldestPendingAgeMs`. Every registered subscription appears — paused and ephemeral ones included — even at zero backlog. Computed live off indexed columns, so it is safe to poll.
`reapEphemeralSubscriptions()` → `Promise<number>`	Drop ephemeral subscriptions whose owning worker is gone. Runs automatically on the worker heartbeat cadence; exposed for operational tooling.
`subscribeLog(topic, name)` → `Promise<void>`	Register a durable log subscription — a named cursor with no handler. Writes immediately, so register it before the publishes it should see.
`logConsumer(topic, name, handler, opts?)` → `this`	Register a managed consumer: the durable log subscription plus, once a worker runs, a poll loop that pulls messages, calls `handler(...args)` per message (the handler may return a `Promise` — it is awaited), and advances the cursor. `opts`: `pollIntervalMs` (default 1000), `batchSize` (default 100), `onError` (`"retry"` leaves a failed message un-acked to re-read, default; `"skip"` acks past it).
`declareTopic(name, opts?)` → `Promise<void>`	Declare a log topic so its publishes are retained even with no subscriber (removing the late-join boundary). `opts.retention` (seconds) bounds a sub-less backlog. Idempotent.
`listDeclaredTopics()` → `Promise<DeclaredTopic[]>`	List declared topics: `name`, `mode`, `retentionMs`, `createdAt`.
`readTopic(topic, name, limit?)` → `Promise<TopicMessage[]>`	Pull up to `limit` (default 100) messages after `name`'s cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then `ackTopic()`.
`ackTopic(topic, name, cursor)` → `Promise<boolean>`	Advance a log subscription's cursor to `cursor` (a message `id`) — a monotonic high-water mark. `false` if nothing moved.
`leaseTopic(topic, name, opts?)` → `Promise<TopicMessage[]>`	Per-message alternative to the cursor read: lease up to `opts.limit` (default 100) messages for `opts.visibility` seconds (default 30), tracked individually so a nack or lease timeout redelivers just that message. Don't mix with `readTopic`/`ackTopic` on one subscription.
`ackMessage(topic, name, messageId)` / `nackMessage(topic, name, messageId)` → `Promise<boolean>`	Ack (done, never redelivered) or nack (redeliver now) one leased message. `false` if there was no un-acked delivery.
`topicLogStats()` → `Promise<TopicLogStat[]>`	Lag snapshot for every log subscription: `topic`, `subscription`, `cursor`, `lag`, `oldestUnackedAgeMs`.

See Log topics for the cursor, at-least-once, and retention semantics.

`TopicMessage`

One message pulled from a log topic, returned by readTopic() and leaseTopic().

Field	Type	Description
`id`	`string`	Message id — pass to `ackTopic()` (cursor read) or to `ackMessage()`/`nackMessage()` (per-message lease).
`args`	`unknown[]`	Deserialized positional args from the `publish()` call.
`metadata`	`Record<string, unknown>`	Caller metadata, if any.
`notes`	`Record<string, unknown>`	Structured notes, if any.
`createdAt`	`number`	Unix-millisecond publish time.

Method

Description

subscriber(topic, name, fn, options?)

publish(topic, args?, options?) → Promise<Job[]>

Fan a message out to every active subscription — one job each. Empty array when nothing is subscribed.

declareSubscriptions() → Promise<void>

Write pending durable subscriptions to storage. Call in a producer-only process; runWorker() does this automatically.

unsubscribe(topic, name) → Promise<boolean>

Remove a subscription.

pauseSubscription(topic, name) / resumeSubscription(topic, name) → Promise<boolean>

Stop/resume deliveries without unregistering.

listSubscriptions(topic?) → Promise<Subscription[]>

All subscriptions, or one topic's active ones.

listTopics() → Promise<string[]>

Distinct topics with at least one subscription.

topicStats(topic?) → Promise<TopicStat[]>

Backlog snapshot per subscription, optionally filtered to one topic: topic, subscription, taskName, queue, active, durable, pending, running, dead, oldestPendingAgeMs. Every registered subscription appears — paused and ephemeral ones included — even at zero backlog. Computed live off indexed columns, so it is safe to poll.

reapEphemeralSubscriptions() → Promise<number>

Drop ephemeral subscriptions whose owning worker is gone. Runs automatically on the worker heartbeat cadence; exposed for operational tooling.

subscribeLog(topic, name) → Promise<void>

Register a durable log subscription — a named cursor with no handler. Writes immediately, so register it before the publishes it should see.

logConsumer(topic, name, handler, opts?) → this

Register a managed consumer: the durable log subscription plus, once a worker runs, a poll loop that pulls messages, calls handler(...args) per message (the handler may return a Promise — it is awaited), and advances the cursor. opts: pollIntervalMs (default 1000), batchSize (default 100), onError ("retry" leaves a failed message un-acked to re-read, default; "skip" acks past it).

declareTopic(name, opts?) → Promise<void>

Declare a log topic so its publishes are retained even with no subscriber (removing the late-join boundary). opts.retention (seconds) bounds a sub-less backlog. Idempotent.

listDeclaredTopics() → Promise<DeclaredTopic[]>

List declared topics: name, mode, retentionMs, createdAt.

readTopic(topic, name, limit?) → Promise<TopicMessage[]>

Pull up to limit (default 100) messages after name's cursor, oldest first and exclusive of it. Empty once caught up. At-least-once — process, then ackTopic().

ackTopic(topic, name, cursor) → Promise<boolean>

Advance a log subscription's cursor to cursor (a message id) — a monotonic high-water mark. false if nothing moved.

leaseTopic(topic, name, opts?) → Promise<TopicMessage[]>

Per-message alternative to the cursor read: lease up to opts.limit (default 100) messages for opts.visibility seconds (default 30), tracked individually so a nack or lease timeout redelivers just that message. Don't mix with readTopic/ackTopic on one subscription.

ackMessage(topic, name, messageId) / nackMessage(topic, name, messageId) → Promise<boolean>

Ack (done, never redelivered) or nack (redeliver now) one leased message. false if there was no un-acked delivery.

topicLogStats() → Promise<TopicLogStat[]>

Lag snapshot for every log subscription: topic, subscription, cursor, lag, oldestUnackedAgeMs.

TopicMessage

One message pulled from a log topic, returned by readTopic() and leaseTopic().

Field	Type	Description
`id`	`string`	Message id — pass to `ackTopic()` (cursor read) or to `ackMessage()`/`nackMessage()` (per-message lease).
`args`	`unknown[]`	Deserialized positional args from the `publish()` call.
`metadata`	`Record<string, unknown>`	Caller metadata, if any.
`notes`	`Record<string, unknown>`	Structured notes, if any.
`createdAt`	`number`	Unix-millisecond publish time.