> Discover all available pages from the documentation index: https://mastra.ai/llms.txt

# ValkeyStreamsPubSub

`ValkeyStreamsPubSub` is a [`PubSub`](https://mastra.ai/reference/pubsub/base) and [`LeaseProvider`](https://mastra.ai/reference/pubsub/lease-provider) implementation backed by Valkey Streams through [Valkey GLIDE](https://github.com/valkey-io/valkey-glide). It provides persistent cross-process delivery, consumer groups, redelivery, topic cleanup, and distributed leasing.

Use it for Valkey deployments. Use [`RedisStreamsPubSub`](https://mastra.ai/reference/pubsub/redis-streams) 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**:

```bash
npm install @mastra/valkey-streams
```

**pnpm**:

```bash
pnpm add @mastra/valkey-streams
```

**Yarn**:

```bash
yarn add @mastra/valkey-streams
```

**Bun**:

```bash
bun add @mastra/valkey-streams
```

## Usage example

```typescript
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 connection URL. Falls back to valkeyOptions.url. (Default: `valkey://localhost:6379`)

**valkeyOptions** (`ValkeyClientOptions`): GLIDE connection options, including native client configuration.

**keyPrefix** (`string`): Prefix for stream keys. (Default: `mastra:topic`)

**blockMs** (`number`): How long each read blocks while waiting for events. (Default: `1000`)

**maxStreamLength** (`number`): Approximate maximum entries retained per stream. Set to 0 to disable trimming. (Default: `10000`)

**streamIdleTtlMs** (`number`): Sliding idle expiry refreshed on stream writes. Set to 0 to disable. (Default: `0`)

**reclaimIntervalMs** (`number`): Interval for reclaiming unacknowledged events. Set to 0 to disable. (Default: `30000`)

**reclaimIdleMs** (`number`): Minimum idle time before an event can be reclaimed. (Default: `60000`)

**maxDeliveryAttempts** (`number`): Maximum nack redeliveries. Pass Infinity to disable the cap. (Default: `5`)

**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.

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