Claims, shared tasks and takeover

Ten agents can reach for the same piece of work. Only one should do it, the others should get its result, and if the one doing it disappears, someone else should finish. runstate covers this with three things that share one mechanism, the lease:

• Claims: exclusive ownership of a key, such as company:acme.
• Shared tasks: callers asking for the same work join one task and all read its single result.
• Crash takeover: work held by a dead worker returns to the queue when its lease expires, and a late result from that worker is rejected.

The examples assume a client and a run. Workers attach to a run they were given the id of:

TypeScript

import { Runstate } from 'runstate-sdk';

const rs = new Runstate({ holder: 'researcher-1' });
const run = rs.scope(process.env.RUNSTATE_SCOPE_ID!); // or: await rs.scopes.create()

Python

import os

from runstate import AsyncRunstate

rs = AsyncRunstate(holder="researcher-1")
run = rs.scope(os.environ["RUNSTATE_SCOPE_ID"])  # or: await rs.scopes.create()

Own a key with a claim

A claim gives one holder exclusive ownership of a key inside a run. run() acquires the claim, keeps its lease renewed while your function runs, and releases it when the function returns or throws.

TypeScript

const report = await run.claim('company:acme').run(
  async (lease) => {
    console.log(`researching acme as generation ${lease.generation}`);
    return researchCompany('acme', { signal: lease.signal });
  },
  { wait: true, timeoutMs: 120_000 },
);

Python

async def research(lease):
    print(f"researching acme as generation {lease.generation}")
    return await research_company("acme")


report = await run.claim("company:acme").run(research, wait=True, timeout_ms=120_000)

If another agent owns the key:

• with wait: false (the default for claims) the call throws ConflictError with code CLAIM_HELD straight away;
• with wait: true the SDK retries every ~300 ms (with jitter) until the claim is free or timeoutMs passes (default 30 s), then throws WaitTimeoutError.

To check without waiting and without an exception, use tryAcquire() and release the lease yourself:

TypeScript

const attempt = await run.claim('company:acme').tryAcquire();
if (attempt.kind === 'busy') {
  console.log('another agent is already on acme');
} else {
  try {
    await researchCompany('acme');
  } finally {
    await attempt.lease.release();
  }
}

Python

attempt = await run.claim("company:acme").try_acquire()
if attempt["kind"] == "busy":
    print("another agent is already on acme")
else:
    lease = attempt["lease"]
    try:
        await research_company("acme")
    finally:
        await lease.release()

Lease length

Claims last leaseSeconds (default 30, allowed 5 to 600) and the SDK renews them in the background at roughly a third of that interval. The lease length is how long a crashed owner blocks everyone else: shorter leases mean faster takeover, longer leases tolerate longer network stalls. Renewals are free; they don’t count as billable operations.

When the lease is lost

Renewal can fail, for example if the process is paused for longer than the lease or the network drops. The SDK then marks the lease lost and stops renewing. It never quietly re-acquires it: by then another agent may own the key.

This is where the two SDKs differ.

TypeScript

In TypeScript, run() does not interrupt your function when the lease is lost. It aborts lease.signal and sets lease.isLost; your function keeps running until it returns. Check either one in long-running work:

await run.claim('file:report.md').run(async (lease) => {
  for (const section of sections) {
    if (lease.isLost) throw new Error('lost ownership of report.md, stopping');
    await writeSection(section, { fencingToken: lease.token(), signal: lease.signal });
  }
});

Python

In Python, run() cancels your coroutine when the lease is lost and raises LeaseLostError from run(). A plain (non-async) function runs in a worker thread and can’t be interrupted: run() stops waiting for it and raises, but the thread runs to completion.

from runstate import LeaseLostError


async def write_report(lease):
    for section in sections:
        lease.raise_if_lost()
        await write_section(section, fencing_token=lease.token())


try:
    await run.claim("file:report.md").run(write_report)
except LeaseLostError:
    print("lost ownership of report.md; another agent may own it now")

lease.is_lost, lease.lost_event (an asyncio.Event) and lease.raise_if_lost() are available when you hold a lease yourself.

Concurrency pool leases (run.pool(name).run()) behave the same way in each language.

Protecting external systems with the fencing token

lease.token() returns the lease’s fencing token. runstate rejects requests that carry a stale token. If the work writes to a system you control (a database row, a file store), store the token with the write and refuse writes carrying an older one. That makes a late write from a previous owner harmless outside runstate too. Tokens are opaque: compare them for equality with the token you last accepted rather than parsing them; lease.generation is the value that increases with each new owner.

Shared tasks

A claim prevents two agents from doing the same thing at once. A task goes further: the work runs once, its result is recorded durably, and any number of callers can read it.

A task is identified by its key within a run. Submitting the same key with the same input returns the existing task instead of creating a new one. Submitting the same key with different input is rejected with IDEMPOTENCY_CONFLICT, because it would be two different pieces of work under one name.

TypeScript

await rs.mailboxes.ensure('research', { mode: 'WORK' });

// Any number of planners can run this. They all get the same task.
const ticket = await run.mailbox('research').submit(
  { company: 'acme' },
  { key: 'research:acme', subscriberId: 'planner-7' },
);

const outcome = await ticket.result({ timeoutMs: 300_000 });
if (outcome.state === 'SUCCEEDED') {
  console.log('report', outcome.outcome);
} else {
  console.log(`task ended ${outcome.state}: ${outcome.reason ?? 'no reason'}`);
}

Python

await rs.mailboxes.ensure("research", mode="WORK")

# Any number of planners can run this. They all get the same task.
ticket = await run.mailbox("research").submit(
    {"company": "acme"},
    key="research:acme",
    subscriber_id="planner-7",
)

outcome = await ticket.result(timeout_ms=300_000)
if outcome["state"] == "SUCCEEDED":
    print("report", outcome["outcome"])
else:
    print(f"task ended {outcome['state']}: {outcome['reason']}")

• result() polls the task (every 250 ms by default) until it reaches a terminal state: SUCCEEDED, FAILED, CANCELLED or EXPIRED. A timeout throws WaitTimeoutError in your process only; the task keeps going.
• outcome is the payload the worker completed the task with (null / None if it completed without one). reason explains non-successful endings.
• subscriberId records who is interested in the task. You can also subscribe later with ticket.join(subscriberId, scopeId) and unsubscribe with ticket.detach(subscriberId). A task with zero subscribers is not cancelled; to stop a task, call ticket.cancel(reason).
• deadline (an ISO 8601 timestamp) makes a task EXPIRED if it hasn’t finished by then.

Get a result back after a restart

A ticket is just a task id. If the process that submitted the task crashes, re-attach by id and read the recorded result. Nothing is re-run.

TypeScript

const ticket = rs.scope(runId).task(taskId);
const view = await ticket.status();
console.log(view.state, view.attempt, view.outcome);

Python

ticket = rs.scope(run_id).task(task_id)
view = await ticket.status()
print(view["state"], view["attempt"], view.get("outcome"))

Results are kept for your plan’s retention window (7 days on the free plan, see Limits & plans). After that the task is tombstoned: status() still returns its state and reason, with tombstoned: true and outcome: null, so you can tell an expired result from an unknown task.

Submitting a key that already has a task in the run, even a finished or tombstoned one, returns that task rather than starting the work again. Use a new key (or a new run) when you really want the work redone.

Workers and crash takeover

Workers take tasks from a work queue. The simplest loop is consume():

TypeScript

const controller = new AbortController();
process.on('SIGTERM', () => controller.abort());

await run.mailbox('research').consume<{ company: string }>(
  async (data, delivery) => {
    const report = await researchCompany(data.company);
    await delivery.complete({ payload: report });
  },
  {
    concurrency: 4, // handlers running at once in this process
    leaseSeconds: 60, // renewed while each handler runs
    signal: controller.signal, // stop taking work, drain, requeue the rest
    onError: (err, delivery) => console.error(`attempt ${delivery.attempt} failed`, err),
  },
);

Python

import asyncio
import signal

stop = asyncio.Event()
asyncio.get_running_loop().add_signal_handler(signal.SIGTERM, stop.set)


async def handle(data, delivery):
    report = await research_company(data["company"])
    await delivery.complete(result={"payload": report})


def log_error(err, delivery):
    print(f"attempt {delivery.attempt} failed: {err}")


await run.mailbox("research").consume(
    handle,
    concurrency=4,  # handlers running at once in this process
    lease_seconds=60,  # renewed while each handler runs
    stop_event=stop,  # stop taking work, drain, requeue the rest
    on_error=log_error,
)

What consume() does for each delivery:

• renews the delivery’s lease while your handler runs;
• if the handler returns, completes the delivery (with no payload, unless you already called delivery.complete(...) yourself);
• if the handler throws, calls onError / on_error and puts the task back on the queue with backoff;
• when you stop it (abort the signal / set the stop event), takes no new work, waits up to shutdownTimeoutMs / shutdown_timeout (30 s) for running handlers, and requeues whatever is still unfinished;
• stops cleanly if the run is cancelled (ScopeCancelledError), rethrows authentication, configuration and not-found errors, and keeps polling through other errors.

Inside a handler you decide how a delivery ends:

Call	Effect on the task
delivery.complete({ payload }) / complete(result={"payload": ...})	SUCCEEDED; payload becomes the recorded result.
delivery.retry()	Back on the queue with backoff; the next delivery has attempt + 1. When the queue’s attempt limit (5) is used up, the task becomes FAILED.
delivery.reject()	Dead-lettered without retry; the task becomes FAILED.

Each delivery settles once: after the first complete, retry or reject, further calls on the same delivery do nothing.

What happens when a worker dies

1. The worker stops renewing its lease (the process crashed, was killed, or lost the network).
2. When the lease expires, runstate puts the task back on the queue. The next worker to receive it sees attempt incremented.
3. If the old worker comes back and tries to complete, the completion is rejected with STALE_CLAIM (LeaseLostError), and the task keeps the result from whichever worker completed it with a valid lease.

The lease length (leaseSeconds, minimum 5) sets how quickly a dead worker’s task is picked up again.

One recorded result, not one execution

runstate records exactly one result per task. If a worker crashes after doing the work but before completing, the next worker does the work again. Make external side effects idempotent, or check the fencing token in the system you write to.

Receiving manually

consume() covers most workers. For full control, receive() returns one delivery (or null / None when the queue is empty). A delivery from receive() is leased for 30 seconds and is not renewed for you: call delivery.renew() for longer work, then settle it yourself.

TypeScript

const delivery = await run.mailbox('research').receive<{ company: string }>();
if (delivery !== null) {
  try {
    await delivery.renew(120); // extend before long work
    const report = await researchCompany(delivery.data.company);
    await delivery.complete({ payload: report });
  } catch (err) {
    await delivery.retry();
    throw err;
  }
}

Python

delivery = await run.mailbox("research").receive()
if delivery is not None:
    try:
        await delivery.renew(120)  # extend before long work
        report = await research_company(delivery.data["company"])
        await delivery.complete(result={"payload": report})
    except Exception:
        await delivery.retry()
        raise

Plain messages

If you don’t need a task and its result, send() puts a message on a queue. An optional workKey deduplicates sends to the same queue and run for 7 days: a duplicate returns the original message with replay: true, and a different payload under the same workKey is rejected with IDEMPOTENCY_CONFLICT.

TypeScript

const sent = await run.mailbox('research').send({ company: 'acme' }, { workKey: 'crawl:acme' });
console.log(sent.id, sent.replay);

Python

sent = await run.mailbox("research").send({"company": "acme"}, work_key="crawl:acme")
print(sent["id"], sent["replay"])

Related

• Quotas and concurrency pools: make a task start only when the pool slots and quota it needs are available.
• Task groups and cancellation: stop a run, and finish when enough tasks have succeeded.
• API reference: Claims, Tasks, Work queues.