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. 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. Members share work round-robin. Subscriptions without a group create a private consumer group. 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

Requires Redis 7.0 or later. The reclaim loop relies on XCLAIM removing trimmed entries from the pending list, which earlier versions don't do.

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',
  }),
})

Redis Cluster

Pass cluster instead of url/redisOptions to connect to a Redis Cluster (for example, AWS ElastiCache with cluster mode enabled). The options are forwarded to createCluster() from redis.

src/mastra/index.ts

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

export const mastra = new Mastra({
  pubsub: new RedisStreamsPubSub({
    cluster: {
      rootNodes: [{ url: 'redis://node-1:6379' }, { url: 'redis://node-2:6379' }],
    },
  }),
})

Bring your own client

To control client construction yourself (TLS, credential providers, and so on), pass an unconnected redis client as client. The pubsub uses it as its writer and calls client.duplicate() for each subscription's blocking reader. Every connection stays distinct. The pubsub owns the client's lifecycle (it connects on first use and quits it on close()), so don't share it with the rest of your app.

src/mastra/index.ts

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

const pubsub = new RedisStreamsPubSub({
  client: createClient({ url: process.env.REDIS_URL, socket: { tls: true } }),
})

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.

cluster?:

RedisClusterOptions

Connect to a Redis Cluster. Options are passed to createCluster() from redis. Mutually exclusive with url, redisOptions, and client.

client?:

RedisClientType | RedisClusterType

A pre-configured, unconnected redis client (standalone or cluster) used as the writer; readers are created with client.duplicate(). The pubsub owns its lifecycle (connects lazily, quits on close()). Mutually exclusive with url, redisOptions, and cluster.

maxStreamLength?:

number

= 10000

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

streamIdleTtlMs?:

number

= 604800000 (7 days)

Idle expiry in milliseconds: a sliding TTL refreshed on every write to a stream (publish, subscribe creating its consumer group, nack retry). 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, together with its entries and consumer groups. Only writes refresh the TTL — a consumer slowly draining a backlog does not — so keep it well above the longest expected gap between writes on a live topic, and above the longest a grouped worker may be offline while still expecting to replay its backlog on restart. For topics with a clear end of life (workflow and agent runs), clearTopic deletes the stream eagerly and the TTL only covers streams that never reach that call, such as a crashed run. For open-ended topics that are never cleared (for example, per-conversation streams), the TTL is the only mechanism that reclaims their memory. Set to 0 to disable expiry; streams that never reach clearTopic then stay in Redis forever. Must be a non-negative integer.

reclaimIntervalMs?:

number

= 30000

How often, in milliseconds, a subscription reclaims events that an earlier consumer read but never acknowledged. Set to 0 to disable. While it's enabled, workflow recovery leaves evented runs to this reclaim instead of restarting them. With 0, the pubsub no longer redelivers a stopped worker's events, so workflow recovery restarts evented runs when an instance boots, as it does without a broker.

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, and at 30000 or more for evented workflows: a step event reclaimed sooner can be dropped as a duplicate.

maxDeliveryAttempts?:

number

= 5

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

inFlightTimeoutMs?:

number

= 0

How long a handler may hold an event without acking or nacking before the reclaim loop nacks it on its behalf. The event is republished with an incremented deliveryAttempt, so maxDeliveryAttempts still applies. Set this to recover hung handlers in single-consumer groups. 0 disables it.

publishRetryBudgetMs?:

number

= 120000

How long, in milliseconds, publish keeps retrying a write that failed with a transient connection error, such as a closed socket, ECONNRESET, or Redis answering LOADING after a restart. When the budget runs out, publish rejects with PublishRetryBudgetExhaustedError instead of hanging. PublishRetryBudgetExhaustedError is exported from @mastra/redis-streams.

logger?:

{ debug?: Function; warn?: Function }

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

Properties

supportedModes:

ReadonlyArray<"pull" | "push">

Returns ["pull"].

supportsOffsets:

boolean

Returns false. Redis stream anchors support start positions, but not the numeric indexes used by subscribeFromOffset().

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.

Set options.startFrom to "latest" to skip retained entries when Redis creates the consumer group. The default, "earliest", reads retained entries first. This option doesn't reset an existing group's checkpoint.

await pubsub.subscribe(
  'workflow.events',
  (event, ack, nack) => {
    console.log(event)
  },
  { group: 'live-workers', startFrom: 'latest' },
)

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.

Not every topic reaches a clearTopic call. Topics without a defined end of life (per-conversation streams, application feeds) are never cleared, and a stream that misses its cleanup (for example, a crashed run) is never retried. The streamIdleTtlMs sliding TTL (7 days by default) reclaims those streams once they go idle. Deleting a thread with Memory.deleteThread() also clears that thread's stream.

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

trimTopic(topic, { runId, producedBefore? })

Pages through the stream with XRANGE and runs XDEL for every entry published with runId. With producedBefore, only entries produced at or before that time are deleted, skipping approval and suspension prompts. Mastra uses this to trim a long run each time its messages are saved. Mastra calls this once a thread run's messages are persisted, so an idle thread's stream empties instead of growing to maxLen. Because entries are matched by runId, a run resumed after a restart also removes the entries it published before the restart. Entries from other runs, including those from other processes, are never touched. Best-effort. Failures are logged at warn level.

await pubsub.trimTopic('agent.thread.thread-abc', { runId: 'run-123' })

close()

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

await pubsub.close()

Redis configuration

Set maxmemory-policy to noeviction on the Redis instance. Consumer groups, unacknowledged entries, and lease keys are live state, not cache. Under any evicting policy, Redis deletes them when it reaches maxmemory: lease keys and streams with a TTL are eviction candidates under volatile-* policies, and every key is a candidate under allkeys-* policies. Lost leases let two processes run the same thread, and evicted streams silently drop events.

With noeviction, Redis rejects writes instead of deleting data when memory runs out, so size maxmemory for your traffic and rely on streamIdleTtlMs to reclaim idle streams. Managed services often default to an evicting policy (Amazon ElastiCache defaults to volatile-lru), so check the setting explicitly.

When maxmemory is set and the policy isn't noeviction, RedisStreamsPubSub logs a warning once after it first connects. It reads the policy with INFO memory, because many managed services block CONFIG GET.

redis-server --maxmemory-policy noeviction

Reclaim streams created without a TTL

Streams written while streamIdleTtlMs was 0, including by versions before the 7-day default, have no TTL. They stay in Redis until a new write gives them one. To reclaim them:

1. Switch the instance to noeviction first. Under a volatile-* policy, adding a TTL makes a live stream an eviction candidate.
2. Deploy every process that writes to this Redis with a version that defaults the TTL, or with an explicit streamIdleTtlMs greater than 0. A write with expiry turned off doesn't refresh an existing TTL, so a stream that an old writer still uses would expire mid-use.
3. Give every stream without a TTL the expiry it would have had: 7 days after its last write. Streams idle for longer than that expire immediately, which frees memory right away. PEXPIRE runs even on an instance that's rejecting writes because it's out of memory.

# Replace mastra:topic with your keyPrefix, and add your connection flags
# (-h, -p, -a, --tls) to every redis-cli call. On a cluster, run it against each primary.
TTL_MS=604800000
NOW_MS=$(redis-cli time | awk 'NR == 1 { s = $1 } NR == 2 { printf "%.0f", s * 1000 + $1 / 1000 }')
redis-cli --scan --pattern 'mastra:topic:*' | while read -r key; do
  [ "$(redis-cli type "$key")" = "stream" ] || continue
  [ "$(redis-cli pttl "$key")" = "-1" ] || continue
  last_ms=$(redis-cli --raw xinfo stream "$key" | awk 'prev == "last-generated-id" { split($0, id, "-"); print id[1]; exit } { prev = $0 }')
  [ -n "$last_ms" ] || continue
  if [ "$last_ms" = "0" ]; then
    remaining=$TTL_MS # never written to, only subscribed
  else
    remaining=$(( TTL_MS - (NOW_MS - last_ms) ))
    [ "$remaining" -lt 1 ] && remaining=1
    [ "$remaining" -gt "$TTL_MS" ] && remaining=$TTL_MS
  fi
  redis-cli pexpire "$key" "$remaining" > /dev/null
done

The loop only touches streams without a TTL, so it's safe to run repeatedly. After step 2, a stream that's still in use keeps its data, because every write refreshes the TTL.

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.

A subscription never reclaims an event its own handler is still processing, so a slow handler isn't invoked twice for the same event. If a handler hangs, a different consumer in the group reclaims the event once it has been idle for reclaimIdleMs. In a single-consumer group there's no sibling to do that, so set inFlightTimeoutMs to have the subscription nack the event itself after that long.

Evented agent worker takeover

With Postgres storage, distributed evented agents publish worker orchestration events through Redis. If a worker exits before acknowledging an event, another worker can reclaim it and finish the run while the original API process stays alive. Configure the API process with MASTRA_WORKERS=false and worker processes with MASTRA_WORKERS=orchestration. All processes must share the same Postgres database, Redis connection settings, and keyPrefix.

Keep reclaimIdleMs above 30 seconds (the default is 60 seconds). A lower value can reclaim an event while the stopped worker still holds the step, causing the event to be acknowledged without running.

Runs started on older versions can finish after upgrading, but don't have guaranteed crash recovery. Saved steps without occurrence identifiers keep their existing unguarded behavior. Other storage adapters keep distributed internal orchestration on the worker process.

An interrupted step can execute again, so make external side effects idempotent. Completed step occurrences are protected against stale redeliveries, and a completed loop iteration reuses its saved result and loop decision. Guarded execution persists the records needed for recovery even when shouldPersistSnapshot returns false. Resume, time travel, restart races, and individual foreach items aren't covered by these occurrence guards. See make tools safe to repeat for agent tools.

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.