RedisStreamsPubSub

RedisStreamsPubSub is a PubSub implementation backed by Redis Streams. It delivers events across processes and hosts, with persistence, consumer groups, and redelivery on failure. It also implements LeaseProvider, so the signals layer can elect a single owner per resource across instances, which is what lets signals coordinate runs in distributed and serverless deployments.

Use it for distributed deployments where several services share an event stream. For single-process delivery, use EventEmitterPubSub. For Google Cloud, use GoogleCloudPubSub.

Each topic maps to a Redis stream key. Subscriptions with a group use a Redis consumer group, so members share work round-robin. Subscriptions without a group create a private consumer group, so every subscriber receives every event.

RedisStreamsPubSub is a pull transport: consumers read events with XREADGROUP, so Mastra runs an orchestration worker to read on its behalf.

Installation

npm

npm install @mastra/redis-streams

pnpm

pnpm add @mastra/redis-streams

Yarn

yarn add @mastra/redis-streams

Bun

bun add @mastra/redis-streams

Usage example

Provide a Redis connection URL.

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

export const mastra = new Mastra({
  pubsub: new RedisStreamsPubSub({
    url: 'redis://localhost:6379',
  }),
})

Constructor parameters

url?:

string

= redis://localhost:6379

Redis connection URL. Falls back to redisOptions.url.

keyPrefix?:

string

= mastra:topic

Prefix for stream keys. Each topic maps to <keyPrefix>:<topic>.

blockMs?:

number

= 1000

How long, in milliseconds, each read blocks while waiting for new events.

redisOptions?:

RedisClientOptions

Options passed to the underlying redis client for advanced configuration.

maxStreamLength?:

number

= 10000

Approximate maximum number of entries kept per stream. Set to 0 to disable trimming.

streamIdleTtlMs?:

number

= 0

Idle expiry in milliseconds: a sliding TTL refreshed on every write (publish, nack retry, group re-creation). Each write resets it, so an actively-written stream never expires mid-flight; a stream left idle for the full duration is deleted by Redis automatically. Note that only writes refresh the TTL — a consumer slowly draining a backlog does not — so set it well above the longest expected gap between writes on a live topic. This is a backstop, not the primary cleanup — clearTopic handles normal end-of-lifecycle deletion; this only bounds memory for streams that never reach a clearTopic call (e.g. a crashed run). Must be a non-negative integer. Defaults to 0 (disabled).

reclaimIntervalMs?:

number

= 30000

How often, in milliseconds, a subscription reclaims events that an earlier consumer read but never acknowledged. Set to 0 to disable.

reclaimIdleMs?:

number

= 60000

Minimum idle time, in milliseconds, before a pending event is eligible for reclaim. Keep this well above typical processing time to avoid double delivery.

maxDeliveryAttempts?:

number

= 5

Maximum times an event is redelivered through nack before it is dropped. Pass Infinity to disable the cap.

logger?:

{ debug?: Function; warn?: Function }

Optional logger for diagnostics. When omitted, suppressed errors are silent.

Properties

supportedModes:

ReadonlyArray<"pull" | "push">

Returns ["pull"].

Methods

RedisStreamsPubSub implements the PubSub contract. The methods below have behavior specific to this implementation.

subscribe(topic, cb, options?)

Subscribes to a topic. With options.group, members of the group share events through a Redis consumer group. Without a group, the subscriber receives every event through a private consumer group.

await pubsub.subscribe('workflow.events', (event, ack, nack) => {
  console.log(event)
})

flush()

Waits for in-flight publishes to complete.

await pubsub.flush()

clearTopic(topic)

Deletes a topic's stream and every consumer group on it, freeing the memory a finished topic would otherwise hold. Mastra's run lifecycles — durable agents and the evented workflow engine — call this automatically when a run reaches a terminal state; call it yourself only once nothing will read the topic again. It's best-effort and never throws — failures are logged at warn level. A subscriber still attached when the stream is deleted recovers on its own but misses the deleted entries.

Automatic cleanup requires @mastra/core and @mastra/redis-streams versions that both support clearTopic: the runtime routes the call through its caching layer, so upgrade the two packages together to get end-of-run stream deletion.

await pubsub.clearTopic('workflow.events.run-123')

close()

Closes the Redis connections and stops all subscriptions. Call this during graceful shutdown.

await pubsub.close()

Redelivery and reclaim

When a subscriber calls nack, the event is republished with an incremented deliveryAttempt and the original is acknowledged. Once an event reaches maxDeliveryAttempts, it's dropped instead of redelivered. Separately, each subscription periodically reclaims events that an earlier consumer in the group read but never acknowledged, controlled by reclaimIntervalMs and reclaimIdleMs.

Distributed leasing

RedisStreamsPubSub implements the LeaseProvider contract on top of the same Redis connection. The signals runtime uses it to elect a single owner (usually per thread key) so that across instances only one process wakes and runs the agent, and others route follow-up work to the holder. This is what makes signals work on serverless and multi-instance deployments; without a shared lease, each instance would start its own competing run.

Lease keys are namespaced under the same keyPrefix as topics, as <keyPrefix>:lease:<key>. All operations are atomic: acquireLease uses SET NX PX and refreshes its own TTL idempotently, while releaseLease, renewLease, and transferLease use Lua scripts that check ownership before mutating, so a concurrent renewal from another owner is never clobbered.

You don't call these methods directly. Configuring RedisStreamsPubSub as the pubsub backend is enough for the runtime to detect and use the capability. See LeaseProvider for the full method contract.