Once and queues

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 two ways:

• once wraps a function so it runs once per run for the same arguments. Every caller, in any process, gets the same result.
• queue hands work to worker processes. Producers call it like a function; workers run serve().

Both are built on leases, so work held by a dead process is picked up by someone else.

Do work once

TypeScript

import runstate from 'runstate-sdk';

const research = runstate.once(researchCompany);

// Any number of agents can call this. The function runs once; they all get its result.
const report = await research('acme');

Python

import runstate


@runstate.once
def research_company(name: str) -> dict: ...


# Any number of agents can call this. The function runs once; they all get its result.
report = research_company("acme")

What happens on each call:

1. The SDK hashes the arguments into a key, <name>:<hash>.
2. If nobody has started that key in the run, this caller runs the function and records the result.
3. If another caller is running it, this caller waits and gets the same result.
4. If the key already has a result in the run, it comes back straight away.

Callers with different arguments don’t wait for each other. research('acme') and research('globex') run in parallel.

Choose the key

By default the key covers every argument. Arguments must be JSON values: strings, numbers, booleans, null, lists and plain objects (in Python also tuples, dataclasses and pydantic models). Anything else is a ConfigError.

When an argument shouldn’t count, such as a client object or a timestamp, pass key:

TypeScript

const research = runstate.once(
  async (company: Company, http: HttpClient) => researchCompany(company, http),
  { name: 'research', key: (company) => company.domain },
);

Python

@runstate.once(name="research", key=lambda company, http: company.domain)
def research(company: Company, http: HttpClient) -> dict: ...

name defaults to the function name. Give it explicitly when you wrap an anonymous function, or when a TypeScript process and a Python process should share results: both SDKs hash arguments the same way, so equal names and arguments give equal keys.

Results

• The result must be JSON-serializable and at most 64 KB. A larger result raises ResultTooLargeError. Store big outputs elsewhere, such as object storage, and return a reference.
• Results belong to the run. After the run is stopped or finished, the next call starts a fresh run and runs the function again. See runs.

Failures and retries

If the function throws, the error goes back to that caller and the next waiting caller tries again. After attempts failures (3 by default), every caller gets OnceFailedError with the last error message, until the run ends.

TypeScript

import runstate, { OnceFailedError } from 'runstate-sdk';

const research = runstate.once(researchCompany, { attempts: 5 });

try {
  await research('acme');
} catch (err) {
  if (err instanceof OnceFailedError) console.error(`gave up after ${err.attempts} attempts: ${err.message}`);
  else throw err;
}

Python

import runstate


@runstate.once(attempts=5)
def research_company(name: str) -> dict: ...


try:
    research_company("acme")
except runstate.OnceFailedError as err:
    print(f"gave up after {err.attempts} attempts: {err}")

When a process dies

The caller running the function holds a lease (30 seconds by default, set with lease). The SDK renews it while the function runs. If the process crashes, the lease lapses and a waiting caller takes the work over.

If the first process was only slow and comes back after the takeover, runstate rejects its result. The SDK aborts context().signal and raises LeaseLostError. Check the signal in long work so it stops early:

TypeScript

const writeReport = runstate.once(async function writeReport(name: string) {
  const { signal } = runstate.context()!;
  const sections = [];
  for (const part of parts) {
    signal.throwIfAborted();
    sections.push(await writeSection(name, part, { signal }));
  }
  return sections;
});

Python

@runstate.once
def write_report(name: str) -> list:
    signal = runstate.context().signal
    sections = []
    for part in PARTS:
        if signal.is_set():
            raise RuntimeError("stopping")
        sections.append(write_section(name, part))
    return sections

One recorded result, not one execution

runstate records exactly one result per key. If a process crashes after doing the work but before the result is recorded, the next caller does the work again. Make external side effects safe to repeat.

Hand work to workers with a queue

once runs the function in whichever process calls first. When the work should run on dedicated workers, for example machines with browsers or GPUs, use a queue.

Define the queue once, in a module both sides import:

TypeScript

research.ts

import runstate from 'runstate-sdk';

export const research = runstate.queue(
  'research',
  async (input: { company: string }) => researchCompany(input.company),
  { attempts: 5, lease: '60s', concurrency: 4 },
);

Python

research.py

import runstate


@runstate.queue("research", attempts=5, lease="60s", concurrency=4)
def research(input: dict) -> dict:
    return research_company(input["company"])

The handler takes one argument, the input, and returns the result. Both must be JSON-serializable and at most 64 KB.

Producers

TypeScript

import { research } from './research.ts';

// Submit and wait for the result.
const report = await research({ company: 'acme' });

// Or submit now and collect later.
const job = await research.send({ company: 'globex' });
console.log(`submitted ${job.id}`);
const later = await job.result();

Python

from research import research

# Submit and wait for the result.
report = research({"company": "acme"})

# Or submit now and collect later.
job = research.send({"company": "globex"})
print(f"submitted {job.id}")
later = job.result()

In async code, use await research.asend(...) and await job.aresult(). If the handler is async def, calling the queue is awaitable too.

Shared tasks

Submitting the same input twice in a run returns the same job, so any number of producers can ask for the same work and all get its one result. The job’s key is a hash of the input. Pass key to the queue to decide what counts as the same work:

TypeScript

export const research = runstate.queue('research', researchHandler, {
  key: (input: { company: string; requestedAt: string }) => input.company,
});

Python

@runstate.queue("research", key=lambda input: input["company"])
def research(input: dict) -> dict: ...

A finished job stays the answer for the rest of the run. Start a fresh run to do the work again.

Workers

TypeScript

worker.ts

import runstate from 'runstate-sdk';
import './research.ts';

await runstate.serve(); // until the run stops or the process gets SIGTERM

Python

worker.py

import runstate

import research  # noqa: F401  (defines the queue)

runstate.serve()  # until the run stops or the process gets SIGTERM

serve() runs every queue defined in the process:

• takes jobs, up to concurrency at once, and renews each job’s lease while the handler runs;
• records the handler’s return value as the job’s result;
• if the handler throws, puts the job back with backoff. After attempts failures (5 by default), callers get TaskFailedError with the last error message;
• when the run stops, or the process gets SIGTERM or SIGINT, takes no new work and waits up to 30 seconds for running handlers.

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 lapses, the job goes back to the queue. The next worker to take it sees context().attempt incremented.
3. If the old worker comes back and tries to record a result, the result is dropped quietly. No LeaseLostError reaches your code, and the job’s caller gets the result of the worker that finished it.

The lease (30 seconds by default) sets how quickly a dead worker’s job is picked up again.

Queues that need resources

A queue can name the slots, limit and budget each job needs. A worker takes the job and all of them in one step, so a job never holds a browser while it waits for a rate limit:

TypeScript

export const render = runstate.queue('render', async (input: { url: string }) => renderPage(input.url), {
  slots: 'browser',
  limit: { name: 'search-api', units: 1 },
  budget: { name: 'model-usd', reserve: 0.4 },
});

Python

@runstate.queue(
    "render",
    slots="browser",
    limit={"name": "search-api", "units": 1},
    budget={"name": "model-usd", "reserve": 0.40},
)
def render(input: dict) -> dict: ...

The named resources must already exist. See Limits and slots and Spend budgets.

Related

• Gather, stop and wait: stop when enough jobs have good results, and stop a whole run.
• TypeScript SDK and Python SDK reference.
• Other languages: HTTP API reference.