# Claims, shared tasks and takeover

> Give one agent ownership of a key, let many callers share one task and its single result, and let another worker take over when one crashes.

Source: https://docs.getrunstate.com/guides/work-once/

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**

```ts
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**

```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**

```ts
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**

```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**

```ts
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**

```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:

```ts
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.

```python
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**

```ts
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**

```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**

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

**Python**

```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](https://docs.getrunstate.com/limits/)). 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**

```ts
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**

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

> **Note: 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**

```ts
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**

```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**

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

**Python**

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

## Related

- [Quotas and concurrency pools](https://docs.getrunstate.com/guides/share-capacity/): make a task start only when the pool slots and quota it needs are available.
- [Task groups and cancellation](https://docs.getrunstate.com/guides/completion-and-cancellation/): stop a run, and finish when enough tasks have succeeded.
- API reference: [Claims](https://docs.getrunstate.com/api/claims/), [Tasks](https://docs.getrunstate.com/api/tasks/), [Work queues](https://docs.getrunstate.com/api/work-queues/).