RedisStreamsPubSub
RedisStreamsPubSub は PubSub の実装で、Redis Streams を基盤とします。永続化、コンシューマーグループ、障害時の再配信に対応し、プロセスやホストをまたいでイベントを配信します。また、LeaseProvider も実装しているため、signals レイヤーはインスタンス間でリソースごとに単一の所有者を選出できます。これにより、分散環境やサーバーレス環境で signals が実行を調整できます。
複数のサービスがイベントストリームを共有する分散デプロイで使用します。単一プロセスでの配信には EventEmitterPubSub を使用してください。Google Cloud では GoogleCloudPubSub を使用してください。
各トピックは Redis のストリームキーに対応します。グループを指定したサブスクリプションでは Redis のコンシューマーグループを使用するため、メンバー間でラウンドロビン方式により処理を分担します。グループを指定しないサブスクリプションではプライベートなコンシューマーグループが作成されるため、すべてのサブスクライバーがすべてのイベントを受信します。
RedisStreamsPubSub は pull トランスポートです。コンシューマーは XREADGROUP でイベントを読み取るため、Mastra が代わりに読み取るオーケストレーション Worker を実行します。
インストールインストールへの直接リンク
- npm
- pnpm
- Yarn
- Bun
npm install @mastra/redis-streams
pnpm add @mastra/redis-streams
yarn add @mastra/redis-streams
bun add @mastra/redis-streams
使用例使用例への直接リンク
Redis の接続 URL を指定します。
import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'
export const mastra = new Mastra({
pubsub: new RedisStreamsPubSub({
url: 'redis://localhost:6379',
}),
})
コンストラクターパラメーターコンストラクターパラメーターへの直接リンク
url?:
redisOptions.url を使用します。keyPrefix?:
<keyPrefix>:<topic> に対応します。blockMs?:
redisOptions?:
redis クライアントへ渡すオプション。maxStreamLength?:
streamIdleTtlMs?:
clearTopic が処理します。この設定は、clearTopic の呼び出しに到達しないストリーム（クラッシュした実行など）のメモリ使用量だけを制限します。0 以上の整数である必要があります。デフォルトは 0（無効）です。reclaimIntervalMs?:
reclaimIdleMs?:
maxDeliveryAttempts?:
nack によって再配信される最大回数。上限を無効にするには Infinity を渡します。logger?:
プロパティプロパティへの直接リンク
supportedModes:
["pull"] を返します。メソッドメソッドへの直接リンク
RedisStreamsPubSub は PubSub 契約を実装します。以下のメソッドには、この実装固有の動作があります。
subscribe(topic, cb, options?)subscribetopic-cb-optionsへの直接リンク
トピックを購読します。options.group を指定すると、グループのメンバーは Redis のコンシューマーグループを通じてイベントを分担します。グループを指定しない場合、サブスクライバーはプライベートなコンシューマーグループを通じてすべてのイベントを受信します。
await pubsub.subscribe('workflow.events', (event, ack, nack) => {
console.log(event)
})
flush()flushへの直接リンク
処理中の publish が完了するまで待機します。
await pubsub.flush()
clearTopic(topic)cleartopictopicへの直接リンク
トピックのストリームと、そのストリーム上のすべてのコンシューマーグループを削除し、完了したトピックが保持し続けるメモリを解放します。Mastra の実行ライフサイクル（durable Agent とイベント駆動型 Workflow エンジン）は、実行が終了状態に達するとこれを自動的に呼び出します。トピックが今後一切読み取られないことを確認してから、自分で呼び出してください。これはベストエフォートで実行され、例外をスローしません。失敗は warn レベルでログに記録されます。ストリームの削除時に接続中のサブスクライバーは自動的に復旧しますが、削除されたエントリは受信できません。
自動クリーンアップには、clearTopic をサポートするバージョンの @mastra/core と @mastra/redis-streams が両方とも必要です。ランタイムはキャッシュレイヤーを介して呼び出しをルーティングするため、実行終了時のストリーム削除を有効にするには、2 つのパッケージを同時にアップグレードしてください。
await pubsub.clearTopic('workflow.events.run-123')
close()closeへの直接リンク
Redis 接続を閉じ、すべてのサブスクリプションを停止します。正常なシャットダウン時に呼び出してください。
await pubsub.close()
再配信と再取得再配信と再取得への直接リンク
サブスクライバーが nack を呼び出すと、deliveryAttempt をインクリメントしてイベントが再度 publish され、元のイベントには確認応答が行われます。イベントが maxDeliveryAttempts に達すると、再配信されずに破棄されます。これとは別に、各サブスクリプションは、グループ内の以前のコンシューマーが読み取ったものの確認応答しなかったイベントを定期的に再取得します。この動作は reclaimIntervalMs と reclaimIdleMs で制御されます。
分散リース分散リースへの直接リンク
RedisStreamsPubSub は、同じ Redis 接続上に LeaseProvider 契約を実装します。signals ランタイムはこれを使用して単一の所有者（通常はスレッドキーごと）を選出します。これにより、インスタンスをまたいで 1 つのプロセスだけが Agent を起動して実行し、ほかのプロセスは後続の処理を所有者へルーティングします。この仕組みによって、signals はサーバーレス環境やマルチインスタンス環境で動作します。共有リースがなければ、各インスタンスがそれぞれ競合する実行を開始します。
リースキーは、トピックと同じ keyPrefix の名前空間に <keyPrefix>:lease:<key> として配置されます。すべての操作はアトミックです。acquireLease は SET NX PX を使用し、自身の TTL を冪等に更新します。releaseLease、renewLease、transferLease は、変更前に所有権を確認する Lua スクリプトを使用するため、別の所有者による同時更新が上書きされることはありません。
これらのメソッドを直接呼び出す必要はありません。RedisStreamsPubSub を pubsub バックエンドとして設定するだけで、ランタイムがこの機能を検出して使用します。メソッド契約の詳細については、LeaseProvider を参照してください。