LeaseProvider

LeaseProvider 是分散式租約 contract，與 event 傳遞（PubSub）分開。Mastra 的 signal layer 會使用它，在多個 process（例如 serverless invocation）之間為某項資源選出單一擁有者，通常是 thread key。擁有者是喚醒並執行 Agent stream 的 process，因此其他 process 會將後續工作路由給它，而不會啟動互相競爭的 run。

租約與 pub/sub 是不同的關注事項。後端只有在確實能協調鎖定時才實作 LeaseProvider，例如透過原子 SET/Lua 的 Redis，或單一 process 使用的記憶體內 map。無法提供租約的後端會省略此能力；signal runtime 會偵測此功能，並改用不執行任何操作的 Provider，以保留單一 process 行為。

內建的 RedisStreamsPubSub 會實作 LeaseProvider，讓 signal 能在分散式與 serverless 部署中跨 instance 協調。

使用範例

你不會直接建構 LeaseProvider。請在 Mastra constructor 上設定實作它的 pub/sub 後端（例如 RedisStreamsPubSub），signal runtime 就會自動使用它進行跨 process 協調。

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,
  }),
})

若要在自訂後端中實作租約，請實作下列方法。signal runtime 會從結構上偵測此能力（因此可跨 package boundary 運作），且只有所有方法都存在時才會使用。

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)

以原子方式嘗試取得 key 的租約。如果 caller 取得租約，傳回 { acquired: true, owner }；否則傳回 { acquired: false, owner }，其中 owner 是目前持有者，caller 可將後續工作路由給對方。同一 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

租約 key，例如 thread key。

owner:

string

擁有者的識別碼，例如 runId。同一 owner 可重複呼叫 acquireLease，以冪等方式續約或釋出。

ttlMs:

number

租約的存留時間，以毫秒為單位。

getLeaseOwner(key)

讀取租約目前的擁有者；若租約未被持有，則傳回 undefined。

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

傳回：Promise<string | undefined>

releaseLease(key, owner)

釋出租約。如果 caller 不是目前擁有者，此方法不會執行任何操作：實作會在釋出前以原子方式檢查擁有權，因此絕不會覆蓋其他擁有者的並行續約。

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

傳回：Promise<void>

renewLease(key, owner, ttlMs)

續訂 owner 所擁有的現有租約，延長其 TTL。如果續訂成功且 caller 仍擁有租約，傳回 true；如果租約已遺失（TTL 到期或由其他 owner 取得），則傳回 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，中途不會釋出 key。此無空窗 primitive 讓後續 owner 能在目前 owner 完成後，立即接管相同 key。例如，thread run 完成後，排入 queue 的後續 run 可以接管。若採用先釋出再取得的方式，key 會短暫呈現空白，競爭中的 process 可能搶先取得已釋出的租約並啟動互相競爭的 run。

如果 fromOwner 仍持有租約，且擁有權已移至 toOwner，傳回 true；如果租約已遺失，則傳回 false，此時 caller 應改用新的 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)，並記載交換不是原子操作，因為競爭中的 process 可能在空窗期間取得 key。維持此方法為必填，可讓 caller 使用單一路徑，並將原子性明確列為各後端的決策。

能力偵測

signal runtime 會從結構上偵測 LeaseProvider，而不使用 instanceof；即使分開發布的後端解析到不同的 @mastra/core 副本，偵測仍可運作。當值公開全部五個方法（acquireLease、getLeaseOwner、releaseLease、renewLease、transferLease）時，就會視為 LeaseProvider。

如果設定的 pub/sub 後端未實作 LeaseProvider，runtime 會改用永遠獲勝且不執行任何操作的 Provider。每個 caller 都會在自己的租約競爭中獲勝，而釋出、續訂與移轉不會產生作用，以保留預期的單一 process 行為。

相關內容

• PubSub：event 傳遞 contract，與租約分開
• RedisStreamsPubSub：實作 LeaseProvider 的內建後端
• Signal：使用租約跨 process 協調 thread 執行的 runtime
• Channel：使用租約在 serverless 與多 instance 部署中協調 Agent run