Skip to main content

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
Direct link to Installation

npm install @mastra/valkey-streams

Usage example
Direct link to 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
Direct link to Constructor parameters

url?:

string
= valkey://localhost:6379
Valkey connection URL. Falls back to valkeyOptions.url.

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
= 0
Sliding idle expiry refreshed on stream writes. Set to 0 to disable.

reclaimIntervalMs?:

number
= 30000
Interval for reclaiming unacknowledged events. Set to 0 to disable.

reclaimIdleMs?:

number
= 60000
Minimum idle time before an event can be reclaimed.

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
Direct link to 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.

Cleanup and shutdown
Direct link to 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()
On this page