跳到主要内容

LeaseProvider

LeaseProvider 是独立于事件投递（PubSub）的分布式租约契约。Mastra 的 signals 层 使用它为资源（最常见为线程键）在多个进程（例如 serverless 调用）中选出单一所有者。所有者是唤醒并运行 agent 流的进程，因此其他进程会将后续工作路由给它，而不是启动竞争性运行。

租约与发布/订阅是不同的关注点。后端仅在确实能够协调锁时才实现 LeaseProvider，例如通过原子 SET/Lua 的 Redis，或用于单进程的内存映射。无法提供租约的后端会省略它；signals 运行时会进行特性检测并回退到无操作 Provider，从而保留单进程行为。

内置的 RedisStreamsPubSub 实现 LeaseProvider，这使 signals 能够在分布式和 serverless 部署中跨实例协调。

使用示例
使用示例的直接链接

无需直接构造 LeaseProvider。在 Mastra 构造函数上配置实现它的发布/订阅后端（例如 RedisStreamsPubSub），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)
acquireleasekey-owner-ttlms的直接链接

以原子方式尝试获取键的租约。若调用者取得租约，返回 { 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
租约键，例如线程键。

owner:

string
所有者标识符，例如 runId。同一所有者可幂等地调用 acquireLease 来续约或释放。

ttlMs:

number
租约的存活时间（毫秒）。

getLeaseOwner(key)
getleaseownerkey的直接链接

读取租约的当前所有者；未持有租约时返回 undefined。

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

返回：Promise<string | undefined>

releaseLease(key, owner)
releaseleasekey-owner的直接链接

释放租约。若调用者不是当前所有者，此操作无效：实现会在释放前以原子方式检查所有权，因此不会覆盖其他所有者的并发续约。

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

返回：Promise<void>

renewLease(key, owner, ttlMs)
renewleasekey-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)
transferleasekey-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，检测也能工作。当一个值暴露所有五个方法（acquireLease、getLeaseOwner、releaseLease、renewLease、transferLease）时，它会被视为 LeaseProvider。

配置的发布/订阅后端未实现 LeaseProvider 时，运行时会回退到始终获胜的无操作 Provider。每个调用者都会赢得自己的租约竞争，而释放、续约和转移均不起作用，从而保留预期的单进程行为。

  • PubSub：独立于租约的事件投递契约
  • RedisStreamsPubSub：实现 LeaseProvider 的内置后端
  • Signals：使用租约协调跨进程线程执行的运行时
  • Channels：使用租约协调 serverless 和多实例部署中的 agent 运行
On this page