ValkeyStreamsPubSub

ValkeyStreamsPubSub is a PubSub and LeaseProvider implementation backed by Valkey Streams through Valkey GLIDE. It provides persistent cross-process delivery, consumer groups, redelivery, topic cleanup, and distributed leasing.

Use it for Valkey deployments. Use RedisStreamsPubSub for Redis deployments that use the official redis client. Both integrations implement the same Mastra behavior, with independent tests running against Valkey and Redis.

Installation

npm

npm install @mastra/valkey-streams

pnpm

pnpm add @mastra/valkey-streams

Yarn

yarn add @mastra/valkey-streams

Bun

bun add @mastra/valkey-streams

Usage example

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { ValkeyStreamsPubSub } from '@mastra/valkey-streams'

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

Constructor parameters

url?:

string

= valkey://localhost:6379

Valkey connection URL. Falls back to valkeyOptions.url. rediss:// and valkeys:// URLs enable TLS unless valkeyOptions.useTLS is set.

valkeyOptions?:

ValkeyClientOptions

GLIDE connection options, including native client configuration.

keyPrefix?:

string

= mastra:topic

Prefix for stream keys.

blockMs?:

number

= 1000

How long each read blocks while waiting for events.

maxStreamLength?:

number

= 10000

Approximate maximum entries retained per stream. Set to 0 to disable trimming.

streamIdleTtlMs?:

number

= 604800000 (7 days)

Sliding idle expiry in milliseconds, refreshed on every write to a stream. A stream left idle for the full duration is deleted with its entries and consumer groups. Keep it above the longest time a grouped worker may be offline while still expecting to replay its backlog. Set to 0 to disable expiry; streams that never reach clearTopic then stay in Valkey forever.

reclaimIntervalMs?:

number

= 30000

Interval for reclaiming unacknowledged events. 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 before an event can be reclaimed. Keep it at 30000 or more for evented workflows: a step event reclaimed sooner can be dropped as a duplicate.

maxDeliveryAttempts?:

number

= 5

Maximum nack redeliveries. 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.

logger?:

{ debug?: Function; warn?: Function }

Optional diagnostic logger.

Delivery behavior

ValkeyStreamsPubSub supports pull delivery and doesn't support numeric replay offsets. Grouped subscribers compete for events through a shared consumer group. Subscribers without a group receive fan-out delivery through private consumer groups.

Use startFrom: "latest" to skip retained entries when a group is first created. The default, "earliest", reads retained entries first.

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.

Server configuration

Set maxmemory-policy to noeviction. Consumer groups, unacknowledged entries, and lease keys are live state, and any evicting policy deletes them when the server reaches maxmemory, dropping events and leases. When maxmemory is set and the policy isn't noeviction, ValkeyStreamsPubSub logs a warning once after it first connects. To reclaim streams created without a TTL, follow the Redis Streams runbook with valkey-cli.

Cleanup and shutdown

clearTopic(topic) deletes a topic stream and its consumer groups. flush() waits for in-flight publishes, and close() stops subscriptions and closes GLIDE connections.

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