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

= 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

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

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()