LeaseProvider

LeaseProvider 是與事件傳送（PubSub）分開的分散式租約合約。Mastra 的 signals 層使用它，在多個進程（例如 serverless invocation）之間為資源（最常見是 thread key）選出單一擁有者。擁有者是喚醒並執行 Agent 串流的進程，因此其他進程會將後續工作路由給它，而不會開始互相競爭的執行。

租約與 pub/sub 是不同範疇。只有確實能協調鎖定的後端才會實作 LeaseProvider，例如透過原子 SET/Lua 操作的 Redis，或供單一進程使用的記憶體 map。無法提供租約的後端會省略它；signals runtime 會偵測此功能，並改用不執行操作的 Provider，以保留單一進程行為。

內置的 RedisStreamsPubSub 會實作 LeaseProvider，讓 signals 能在分散式及 serverless 部署中跨執行個體協調。

使用範例

你不會直接建構 LeaseProvider。請在 Mastra 建構函數上設定實作它的 pub/sub 後端（例如 RedisStreamsPubSub），signals runtime 便會自動使用它作跨進程協調。

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 runtime 會從結構上偵測此功能（因此可跨套件邊界運作），並只會在所有方法均存在時使用它。

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 的租約。若呼叫者取得租約，傳回 { acquired: true, owner }；否則傳回 { acquired: false, owner }，其中 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。同一擁有者可重複呼叫 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，期間不會解除 key。此無空隙的原語可讓後續擁有者在目前擁有者完成後，立即接管同一 key。例如，thread 執行完成時，已排隊的後續執行可立即接管。若單純先解除再取得，key 會短暫變成空置，競爭中的進程可能取得已釋放租約並開始另一個互相競爭的執行。

如果 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)，並註明交換並非原子操作，因為競爭中的進程可能在空隙期間取得 key。將此方法列為必要方法，代表呼叫者只有一條程式碼路徑，而原子性則由各後端明確決定。

功能偵測

signals runtime 會從結構上而非透過 instanceof 偵測 LeaseProvider，因此即使獨立發佈的後端解析至另一份 @mastra/core，偵測仍可運作。公開全部五個方法（acquireLease、getLeaseOwner、releaseLease、renewLease、transferLease）的值會被視為 LeaseProvider。

若設定的 pub/sub 後端沒有實作 LeaseProvider，runtime 會改用一律勝出的無操作 Provider。每個呼叫者都會勝出自己的租約競爭，而解除、續期及轉移操作均不產生作用，藉此保留預期的單一進程行為。

相關內容

• PubSub：與租約分開的事件傳送合約
• RedisStreamsPubSub：實作 LeaseProvider 的內置後端
• Signals：使用租約協調跨進程 thread 執行的 runtime
• Channels：使用租約在 serverless 及多執行個體部署中協調 Agent 執行