Skip to content
runstatedocs
HomeConsoleGet started

Task groups and cancellation

Decide exactly once when a set of tasks is done (first accepted, N accepted, or all finished), stop handing out work in the same step, and cancel a whole run from one call.

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. runstate gives you two tools:

  • Task groups decide once, under races, whether a set of tasks is done, and close the group to new work in the same step.
  • Run cancellation refuses new work for a run and everything under it, immediately, on the server.

Barriers and timers, lower-level building blocks for “wait until everyone arrives” and “wake me later”, are at the end.

Task groups

Section titled “Task groups”

A task group owns its own run (the gate) and delivers its member tasks through a work queue. You submit members, workers complete them, and something in your system gives each successful member a verdict: accept or reject. The group finalizes as soon as its condition is decided.

A group is open to new members until you close it (group.close()), until expectedMembers members have joined, or until it finalizes. Closed groups show state SEALED. A result that depends on the full member list, such as “the threshold can no longer be reached” or “every member has finished”, is only decided once the group is closed. While it is open, a group can still succeed, and it can fail only when its deadline passes or its run is cancelled.

condition Finalizes with SUCCESS when Finalizes with FAILURE when
FIRST_ACCEPTED the first member is accepted (first_accepted) its deadline passes (deadline)
N_ACCEPTED (with threshold) threshold members are accepted (threshold_reached), open or closed the group is closed and accepted members plus members that could still be accepted can’t reach threshold (threshold_unreachable), or the deadline passes
ALL_TERMINAL the group is closed, every member has finished, at least one was accepted and none rejected (all_terminal) the group is closed and every member has finished otherwise (all_terminal), or the deadline passes

The value in parentheses is the outcomeReason.

Create a group and submit members

Section titled “Create a group and submit members”
await rs.mailboxes.ensure('reviews', { mode: 'WORK' });
const run = await rs.scopes.create();


// Done when five drafts have been accepted.
const group = await run.groups.create({
  mailbox: 'reviews',
  condition: 'N_ACCEPTED',
  threshold: 5,
  parentScopeId: run.id, // cancelling the run also closes the group
});


const tickets = [];
for (let i = 0; i < 12; i++) {
  tickets.push(await group.submit({ draft: i }, { key: `draft-${i}` }));
}
// No more members: from now on the group can also fail as unreachable.
await group.close();


// Workers must attach to the group's own run to receive its members.
const { scopeId: groupRunId } = await group.status();
console.log(`start workers with RUNSTATE_SCOPE_ID=${groupRunId}`);
  • The queue named in mailbox must exist before you create the group (NotFoundError otherwise).
  • A group holds at most 100 members. group.joinTask(taskId) / join_task(task_id) adds an existing task while the group is open.
  • deadline (ISO 8601) finalizes the group with FAILURE if it hasn’t decided by then.
  • Workers are ordinary queue consumers (see Claims, shared tasks and takeover) attached to the group’s run id from group.status().

Close the group

Section titled “Close the group”

group.close() stops new members from joining and evaluates the group right away, in the same step. It returns { state, outcome, outcomeReason }: state is SEALED while the group is still undecided, or FINALIZED if closing decided it (for example an N_ACCEPTED group with too few members left that could be accepted). Calling close() again is safe; on a finalized group it returns the recorded outcome.

If you know the member count up front, pass expectedMembers / expected_members (1 to 100) to create() instead. The group closes itself when that many members have joined, so the submit that adds the last member also closes it. For N_ACCEPTED, expectedMembers can’t be lower than threshold. After the group is closed, submitting a new member fails with ConflictError; resubmitting a member that already joined (same key and input) returns that member.

Until a group is closed, members that end FAILED, CANCELLED or EXPIRED don’t fail it: you may still add members that succeed. After the group is closed they stop counting toward the threshold.

Record verdicts and wait for the decision

Section titled “Record verdicts and wait for the decision”

Accepting requires the member task to have SUCCEEDED; rejecting requires it to have finished. Both are idempotent.

// A reviewer process: judge each finished member.
for (const ticket of tickets) {
  const result = await ticket.result({ timeoutMs: 600_000 });
  if (result.state !== 'SUCCEEDED') continue;
  const verdict = isGoodDraft(result.outcome) ? await group.accept(ticket.id) : await group.reject(ticket.id);
  if (verdict.groupState === 'FINALIZED') break;
}


const decision = await group.wait({ timeoutMs: 600_000 });
console.log(decision.outcome, decision.outcomeReason); // e.g. SUCCESS threshold_reached

A process that restarts can re-attach with run.groups.byId(groupId) (Python by_id) and call status() or wait() again.

What finalizing does

Section titled “What finalizing does”

In the same transaction that records the decision, runstate cancels the group’s run. From that moment:

  • workers asking for more of the group’s work are refused, and consume() loops on the group’s run stop;
  • members that haven’t finished are cancelled;
  • verdicts that arrive afterwards are recorded with late: true and never change the outcome.

Group status() shows every member with its task state, verdict and late flag.

Cancel a run

Section titled “Cancel a run”

Cancelling a run is how you hit stop on a swarm.

const run = await rs.scopes.create({
  deadline: new Date(Date.now() + 60 * 60 * 1000).toISOString(), // optional: stop accepting work after an hour
});
const shard = await run.child(); // a sub-run, cancelled along with its parent


// ... later, from any process that knows the run id:
await rs.scope(run.id).cancel();


const status = await shard.status();
console.log(status.storedState, status.effective); // ACTIVE INACTIVE

After cancel(), for the run and every run below it:

  • New work is refused with SCOPE_CANCELLED (ScopeCancelledError): claims, sends, submits, receives, admissions, pool acquisitions, quota takes and budget reservations.
  • Queued work is cancelled. Tasks that haven’t finished become CANCELLED; queued messages and armed timers are cancelled.
  • Running agents are asked to stop, not killed. Lease renewals are refused too, so the SDK marks held claims, pool leases and deliveries as lost: lease.signal aborts in TypeScript, and in Python run() cancels its handler. consume() stops taking work.
  • Agents can still hand back what they hold. Releasing claims and pool leases and retrying or rejecting deliveries remain allowed. Completing a delivery is refused.
  • Recorded results stay readable. Cancelling never hides a task result that was already recorded.

status() returns storedState (the run’s own state: ACTIVE, CANCELLED or COMPLETED) and effective, which is INACTIVE if the run or any ancestor is no longer active or past its deadline.

complete() closes a run the same way but records that it finished normally. Either is final: a run can’t be reopened, and cancelling a completed run returns a conflict.

A run with a deadline refuses new work with DEADLINE_EXCEEDED once the deadline passes. childLimit caps how many child runs a run can have (default 1000). Two calls to rs.scopes.create() always create two runs. Pass name to create() or child() to label a run; status() returns it, but runs are never looked up by name.

Barriers

Section titled “Barriers”

A barrier releases everyone waiting on it once target distinct participants have arrived: a wait-for-all rendezvous, for example “start the merge step when all three shards are done”.

// Coordinator:
const barrier = await run.barriers.create({ target: 3 });


// Each shard, given barrier.id:
const handle = rs.scope(run.id).barriers.get(barrier.id);
await handle.arrive({ key: 'shard-2' }); // idempotent per key
const view = await handle.wait({ timeoutMs: 300_000 });
console.log(view.state, `${view.arrivals}/${view.target}`);

wait() polls until the barrier is RELEASED (default timeout 60 s) and throws ConflictError with BARRIER_CANCELLED if the barrier is cancelled, for example by cancelling its run. Arriving after release returns BARRIER_RELEASED. The HTTP API can advance a released barrier to a new epoch for reuse (POST /barriers/{id}/nextEpoch); neither SDK wraps that call yet. Creating a barrier is configuration and needs a key with resource_config; arriving and waiting need coordination_write and coordination_read.

Timers

Section titled “Timers”

A timer is a durable wakeup at a time (at, ISO 8601) or after a delay (afterSeconds). It fires on the server whether or not anyone is waiting.

const timer = await run.timers.create({ afterSeconds: 300 });
console.log(`fires at ${timer.fireAt.toISOString()}`);


const { firedAt } = await timer.wait({ timeoutMs: 360_000 });
console.log(`follow-up check at ${firedAt.toISOString()}`);
  • wait() polls for the timer to fire and acknowledges it in the same call. Have only one process wait on a given timer.
  • timer.cancel() disarms a timer that hasn’t fired. Cancelling the run cancels its armed timers.
  • run.timers.get(timerId) re-attaches to a timer by id after a restart.
Section titled “Related”