LeaseProvider

LeaseProvider は、イベント配信（PubSub）とは独立した分散リース契約です。Mastra の signals レイヤーは、リソース（多くの場合はスレッドキー）について、複数のプロセス（たとえばサーバーレス関数の呼び出し）から単一の所有者を選出するためにこれを使用します。所有者となったプロセスが Agent ストリームを起動して実行するため、ほかのプロセスは競合する実行を開始せず、後続の処理をそのプロセスにルーティングします。

リースは pub/sub とは別の関心事です。バックエンドが LeaseProvider を実装するのは、Redis のアトミックな SET/Lua や、単一プロセス向けのインメモリマップなど、実際にロックを調整できる場合に限られます。リースに対応できないバックエンドでは実装を省略します。signals ランタイムはこの機能を検出し、no-op Provider にフォールバックすることで、単一プロセスでの動作を維持します。

組み込みの RedisStreamsPubSub は LeaseProvider を実装しているため、signals は分散環境やサーバーレス環境のインスタンス間で連携できます。

使用例

LeaseProvider を直接構築することはありません。これを実装する pub/sub バックエンド（RedisStreamsPubSub など）を Mastra コンストラクターに設定すると、signals ランタイムがプロセス間の連携に自動的に使用します。

src/mastra/index.ts

import { Mastra } from '@mastra/core'
import { RedisStreamsPubSub } from '@mastra/redis-streams'

export const mastra = new Mastra({
  // RedisStreamsPubSub implements both PubSub and LeaseProvider
  pubsub: new RedisStreamsPubSub({
    url: process.env.REDIS_URL,
  }),
})

カスタムバックエンドにリースを実装するには、以下のメソッドを実装します。signals ランタイムは構造に基づいて機能を検出するため（パッケージ境界をまたいでも動作します）、すべてのメソッドが存在する場合にのみ使用します。

src/mastra/pubsub.ts

import { PubSub } from '@mastra/core/events'
import type { LeaseProvider } from '@mastra/core/events'

export class CustomPubSub extends PubSub implements LeaseProvider {
  async acquireLease(key: string, owner: string, ttlMs: number) {
    // Atomically claim the lease, or report the current holder.
    return { acquired: true, owner }
  }

  // ...getLeaseOwner, releaseLease, renewLease, transferLease
}

メソッド

リース

acquireLease(key, owner, ttlMs)

キーのリース取得をアトミックに試みます。呼び出し元がリースを取得した場合は { acquired: true, owner } を返します。取得できなかった場合は、現在の所有者を owner とする { acquired: false, owner } を返すため、呼び出し元は後続の処理をその所有者にルーティングできます。同じ所有者は acquireLease を冪等に呼び出して、更新または再取得できます。

const result = await pubsub.acquireLease('thread:abc', runId, 15000)

if (result.acquired) {
  // This process owns the thread, so wake and run the agent.
} else {
  // result.owner holds the lease, so route the signal to them.
}

戻り値：Promise<{ acquired: boolean; owner?: string }>

key:

string

スレッドキーなどのリースキー。

owner:

string

所有者の識別子（runId など）。同じ所有者は acquireLease を冪等に呼び出して、更新または解放できます。

ttlMs:

number

リースの有効期間（ミリ秒）。

getLeaseOwner(key)

リースの現在の所有者を読み取ります。リースが保持されていない場合は undefined を返します。

const owner = await pubsub.getLeaseOwner('thread:abc')

戻り値：Promise<string | undefined>

releaseLease(key, owner)

リースを解放します。呼び出し元が現在の所有者でない場合は何も行いません。実装では解放前に所有権をアトミックに確認するため、別の所有者による同時更新が上書きされることはありません。

await pubsub.releaseLease('thread:abc', runId)

戻り値：Promise<void>

renewLease(key, owner, ttlMs)

owner が所有する既存のリースを更新し、TTL を延長します。更新に成功し、呼び出し元が引き続きリースを所有している場合は true を返します。リースを失った場合（TTL の期限切れ、または別の所有者による取得）は false を返します。

const stillOwned = await pubsub.renewLease('thread:abc', runId, 15000)

if (!stillOwned) {
  // Lost the lease, so stop renewing and let the new owner take over.
}

戻り値：Promise<boolean>

transferLease(key, fromOwner, toOwner, ttlMs)

保持中のリースを fromOwner から toOwner へアトミックに移譲し、その間にキーを解放することなく TTL を更新します。この切れ目のないプリミティブにより、現在の所有者が完了した直後に、後続の所有者が同じキーを引き継げます。たとえば、スレッドの実行完了時に、キューに入った後続の実行へ引き継ぐことができます。単純な「解放してから取得する」方式では、キーが一時的に空になります。その隙に競合するプロセスが解放されたリースを取得し、競合する実行を開始する可能性があります。

fromOwner が引き続きリースを保持しており、所有権が toOwner に移った場合は true を返します。すでにリースを失っていた場合は false を返します。その場合、呼び出し元は新たに acquireLease を実行するフォールバック処理を行う必要があります。

const transferred = await pubsub.transferLease('thread:abc', currentRunId, nextRunId, 15000)

if (!transferred) {
  // Lease was lost, so acquire fresh instead.
  await pubsub.acquireLease('thread:abc', nextRunId, 15000)
}

戻り値：Promise<boolean>

警告

移譲をアトミックに実行できないバックエンドでも、ベストエフォートの releaseLease(fromOwner) に続けて acquireLease(toOwner) を実行する形で実装し、切り替えが非アトミックであることを明記する必要があります。処理の隙間で競合するプロセスがキーを取得する可能性があるためです。このメソッドを必須にすることで、呼び出し元のコードパスを一本化しつつ、アトミック性をバックエンドごとの明示的な判断にできます。

機能の検出

signals ランタイムは instanceof ではなく構造に基づいて LeaseProvider を検出します。そのため、別途公開されたバックエンドが異なる @mastra/core のコピーを解決した場合でも検出できます。5 つのメソッド（acquireLease、getLeaseOwner、releaseLease、renewLease、transferLease）をすべて公開する値は、LeaseProvider として扱われます。

設定された pub/sub バックエンドが LeaseProvider を実装していない場合、ランタイムは常に取得に成功する no-op Provider にフォールバックします。すべての呼び出し元が自身のリース競争に勝ち、解放、更新、移譲は何も行いません。これにより、想定される単一プロセスでの動作が維持されます。

関連情報

• PubSub：リースとは独立したイベント配信契約
• RedisStreamsPubSub：LeaseProvider を実装する組み込みバックエンド
• Signals：リースを使用してプロセス間のスレッド実行を調整するランタイム
• Channels：サーバーレス環境やマルチインスタンス環境で、リースを使用して Agent の実行を調整