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.

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.

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.

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