LeaseProvider

LeaseProvider is the distributed leasing contract, separate from event delivery (PubSub). Mastra's signals layer uses it to elect a single owner across multiple processes (for example, serverless invocations) for a resource, most commonly a thread key. The owner is the process that wakes and runs the agent stream, so other processes route follow-up work to it instead of starting a competing run.

Durable agents also claim each run through a lease when the workflows store doesn't support run fencing, so crash recovery can tell whether another process still drives the run.

Leasing is separate from pub/sub. A LeaseProvider implementation needs actual lock coordination, whether through an atomic Redis operation such as SET or Lua or through an in-memory map for a single process. For backends that omit leasing, the signals runtime preserves single-process behavior with a no-op provider.

The built-in RedisStreamsPubSub implements LeaseProvider, which is what enables signals to coordinate across instances in distributed and serverless deployments.

Usage example

You don't construct a LeaseProvider directly. Configure a pub/sub backend that implements it (such as RedisStreamsPubSub) on the Mastra constructor, and the signals runtime uses it automatically for cross-process coordination.

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

To implement leasing in a custom backend, implement the methods below. The signals runtime detects the capability structurally (so it works across package boundaries) and only uses it when all methods are present.

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
}

Methods

Leasing

acquireLease(key, owner, ttlMs)

Atomically tries to acquire a lease on a key. Returns { acquired: true, owner } if the caller claimed the lease, or { acquired: false, owner } where owner is the current holder, so the caller can route follow-up work to them. The same owner can call acquireLease idempotently to renew or re-claim.

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.
}

Returns: Promise<{ acquired: boolean; owner?: string }>

key:

string

The lease key, such as a thread key.

owner:

string

Identifier for the owner, such as a runId. The same owner can call acquireLease idempotently to renew or release.

ttlMs:

number

Time-to-live in milliseconds for the lease.

getLeaseOwner(key)

Reads the current owner of a lease, or undefined if no lease is held.

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

Returns: Promise<string | undefined>

releaseLease(key, owner)

Releases a lease. This is a no-op if the caller isn't the current owner: implementations check ownership atomically before releasing, so a concurrent renewal by another owner is never clobbered.

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

Returns: Promise<void>

renewLease(key, owner, ttlMs)

Renews an existing lease owned by owner, extending its TTL. Returns true if the renewal succeeded and the caller still owns the lease, or false if the lease was lost (TTL expired or another owner took it).

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

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

Returns: Promise<boolean>

transferLease(key, fromOwner, toOwner, ttlMs)

Atomically transfers a held lease from fromOwner to toOwner and refreshes its TTL without releasing the key in between. This gap-free primitive lets a follow-up owner take over the same key immediately after the current owner finishes. For example, a queued follow-up run can take over when a thread run completes. A naive release-then-acquire briefly leaves the key empty. A racing process could then win the freed lease and start a competing run.

Returns true if fromOwner still held the lease and ownership moved to toOwner, or false if the lease was already lost, in which case the caller should fall back to a fresh 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)
}

Returns: Promise<boolean>

注意

Backends that can't transfer a lease atomically must use a best-effort releaseLease(fromOwner) followed by acquireLease(toOwner). They must document that the swap is non-atomic because another process can claim the key between those calls. Keeping the method required gives callers one code path while making atomicity an explicit per-backend decision.

Capability detection

The signals runtime detects LeaseProvider structurally rather than with instanceof, so detection works even when a separately published backend resolves a different copy of @mastra/core. A value is treated as a LeaseProvider when it exposes all five methods (acquireLease, getLeaseOwner, releaseLease, renewLease, transferLease).

When the configured pub/sub backend doesn't implement LeaseProvider, the runtime falls back to an always-win no-op provider. Every caller wins its own lease race, and release, renew, and transfer are inert, which preserves the expected single-process behavior.

Related

• PubSub: The event delivery contract, separate from leasing
• RedisStreamsPubSub: The built-in backend that implements LeaseProvider
• Signals: The runtime that uses leasing to coordinate thread execution across processes
• Channels: Uses leasing to coordinate agent runs in serverless and multi-instance deployments