Gather, stop and wait

Sometimes you need every answer, sometimes five good ones, sometimes the first acceptable result is enough. Once you have it, the rest of the swarm should stop picking up work. This guide covers:

• gather: submit many inputs to a queue and return when enough results are accepted;
• runs: stop or finish everything in a run from one call;
• sleep and together: durable waits for “later” and “when everyone is here”.

gather

gather submits a list of inputs to a queue as one group. Workers handle the jobs as usual. As each result arrives, your accept function decides whether it counts. Once need results are accepted, gather returns them and cancels the rest.

TypeScript

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

// Done when five drafts score above 0.8.
const accepted = await review.gather(drafts, {
  need: 5,
  accept: (result, draft) => result.score > 0.8,
  timeout: '10m',
});
console.log(`${accepted.length} good drafts`);

Python

from review import review

# Done when five drafts score above 0.8.
accepted = review.gather(
    drafts,
    need=5,
    accept=lambda result, draft: result["score"] > 0.8,
    timeout="10m",
)
print(f"{len(accepted)} good drafts")
# in async code: await review.agather(drafts, need=5, ...)

• need is a number, or 'all' (the default). With 'all', gather returns when every job has finished.
• accept runs in the producer and receives the result and the input it came from. Without it, every result is accepted.
• The return value is the accepted results in the order they were accepted.
• If need can no longer be reached, because too many jobs failed or were rejected, gather raises TaskFailedError.
• If timeout runs out first, it raises TimeoutError.

What happens when enough is done

In the same step that records the decision, runstate cancels the group’s remaining jobs:

• workers asking for more of the group’s work get nothing;
• jobs that haven’t started are cancelled;
• handlers that are still running see context().signal abort, and results that arrive afterwards don’t change the outcome.

A group holds at most 100 inputs.

Runs

Every call belongs to a run. By default every process joins the run named by RUNSTATE_RUN (default). To name one in code, use runstate.run(name). It returns a client bound to that run:

TypeScript

import runstate from 'runstate-sdk';

const nightly = runstate.run('nightly-2026-09-17', { deadline: '2h' });

const fetchPrices = nightly.once(fetchPricesFor);
await fetchPrices('eur');

Python

import runstate

nightly = runstate.run("nightly-2026-09-17", deadline="2h")


@nightly.once
def fetch_prices(currency: str) -> dict: ...


fetch_prices("eur")

• The run is created by the first call that needs it. Every other process that uses the same name joins it.
• deadline is a duration from creation. After it passes, nothing new starts and calls raise StoppedError.
• once results and queue jobs belong to the run. When the run is stopped, finished or past its deadline, the next call with the same name starts a fresh run.

Stop a run

Stopping a run is how you hit stop on a swarm. Call it from any process in the run:

TypeScript

await runstate.stop();

Python

runstate.stop()  # await runstate.astop()

After stop(), for the run and every child run:

• Nothing new starts. New wrapped calls, take, sleep and together raise StoppedError.
• Queued work is cancelled. Jobs that haven’t started won’t start.
• Running work is asked to stop, not killed. context().signal aborts in every wrapped call in flight. Check it in long work.
• Workers drain. serve() stops taking work, waits up to 30 seconds for running handlers, and returns.
• Recorded results stay readable in the console.

runstate.finish() ends a run the same way but records that it finished normally. Both are idempotent.

Child runs

A child run groups part of the work so you can stop it on its own. Stopping a run stops its children too.

TypeScript

const retries = runstate.child('retry-batch', { deadline: '10m' });
const retryFetch = retries.once(fetchPricesFor);
await retryFetch('usd');
await retries.stop(); // stops only the child

Python

retries = runstate.child("retry-batch", deadline="10m")


@retries.once
def retry_fetch(currency: str) -> dict: ...


retry_fetch("usd")
retries.stop()  # stops only the child

React to a stop in your code

TypeScript

const crawl = runstate.slots('browser', async (urls: string[]) => {
  const { signal } = runstate.context()!;
  const pages = [];
  for (const url of urls) {
    if (signal.aborted) break; // the run stopped: hand back what we have
    pages.push(await fetchPage(url, { signal }));
  }
  return pages;
});

Python

@runstate.slots("browser")
def crawl(urls: list[str]) -> list:
    signal = runstate.context().signal
    pages = []
    for url in urls:
        if signal.is_set():
            break  # the run stopped: hand back what we have
        pages.append(fetch_page(url))
    return pages

sleep

sleep is a durable timer in the run. A process that restarts and calls sleep with the same name waits only for the time that is left.

TypeScript

await runstate.sleep('cooldown:acme', '10m');
await runstate.sleep('report-ready', { until: new Date('2026-09-18T06:00:00Z') });

Python

from datetime import datetime, timezone

runstate.sleep("cooldown:acme", "10m")  # await runstate.asleep(...)
runstate.sleep("report-ready", until=datetime(2026, 9, 18, 6, 0, tzinfo=timezone.utc))

• The duration is a number of seconds or a string such as '10m'. To wait until a point in time, pass { until: Date } in TypeScript or until= with a datetime in Python.
• If the run stops first, sleep raises StoppedError.
• Names are per run. A name that already fired returns at once when used again in the same run, so give each distinct wait its own name, for example cooldown:acme rather than cooldown.

together

together waits until a given number of different workers have arrived under the same name. Use it for “start the merge step when all three shards are done”.

TypeScript

// In each of the three shard workers, after its own work:
await runstate.together('merge', 3, { timeout: '5m' });
await mergeResults();

Python

# In each of the three shard workers, after its own work:
runstate.together("merge", 3, timeout="5m")  # await runstate.atogether(...)
merge_results()

• Workers are counted by their worker label: the worker option, which defaults to <hostname>-<pid>. A restarted process with the same label counts once. If you set worker yourself, give each process its own label.
• A worker that arrives after the group already went through continues straight away.
• Without timeout, together waits without a limit. If timeout runs out, it raises TimeoutError. If the run stops, it raises StoppedError.

Related

• Events and monitoring: watch jobs finish and fail as they happen.
• TypeScript SDK and Python SDK reference.
• Other languages: HTTP API reference.