Singleton Workflows

A singleton workflow has at most one run in progress at a time. A nightly sync, a per-user onboarding flow or a cache rebuild are typical examples: starting a second copy while the first one is still running would do the same work twice.

You get this behavior by starting every run with the same workflowRunId. A run ID is reserved while its run is in progress, and released once the run finishes, fails or is cancelled.

There are two ways to handle a new trigger while a run is in progress:

In both patterns, a trigger after the previous run has ended starts a new run, even with exactly the same body.

Keep the running one#

Trigger the workflow with a fixed workflowRunId. If a run with that ID is still in progress, client.trigger throws and no new run is started.

import { Client } from "@upstash/workflow";import { QstashError } from "@upstash/qstash";const client = new Client({ token: "<QSTASH_TOKEN>" });try {  const { workflowRunId } = await client.trigger({    url: "https://<YOUR_WORKFLOW_ENDPOINT>/nightly-sync",    workflowRunId: "nightly-sync",    body: { source: "crm" },  });  console.log(`started ${workflowRunId}`); // started wfr_nightly-sync} catch (error) {  if (!(error instanceof QstashError) || error.status !== 400) {    throw error;  }  // a 400 can also mean an invalid request, so check that a run is in progress  const { runs } = await client.logs({    filter: { workflowRunId: "wfr_nightly-sync", state: "RUN_STARTED" },  });  if (runs.length === 0) {    throw error;  }  console.log("nightly-sync is already running, skipping");}

What client.trigger returns:

• No run in progress: a new run starts and the call resolves to { workflowRunId: "wfr_nightly-sync" }.
• A run is in progress: the call throws a QstashError with status: 400. The running run is not affected.

QStash returns 400 for other invalid trigger requests too, such as a malformed URL, so status alone doesn’t tell you that the ID is taken. The example checks with client.logs that a run with the ID is in progress before skipping. If that run ends between the two calls, runs is empty and the error is rethrown, so you can retry the trigger.

Replace the running one#

To always run the latest request, first cancel the existing run by its prefixed run ID (for example, wfr_rebuild-cache), then start a new one with the unprefixed workflowRunId (for example, rebuild-cache). If the previous run hasn’t finished yet, it is stopped and the new run takes its place.

import { Client } from "@upstash/workflow";const client = new Client({ token: "<QSTASH_TOKEN>" });const { cancelled } = await client.cancel("wfr_rebuild-cache");console.log(`cancelled ${cancelled} run(s)`);const { workflowRunId } = await client.trigger({  url: "https://<YOUR_WORKFLOW_ENDPOINT>/rebuild-cache",  workflowRunId: "rebuild-cache",  body: { reason: "settings changed" },});console.log(`started ${workflowRunId}`); // started wfr_rebuild-cache

What each call returns:

• client.cancel resolves to { cancelled: 1 } when a run was in progress and got cancelled, and to { cancelled: 0 } when there was nothing to cancel. It doesn’t throw in either case, so you can call it unconditionally.
• client.trigger resolves to { workflowRunId: "wfr_rebuild-cache" }. The cancel is completed before client.cancel returns, so the ID is free when you trigger the new run.

The cancelled run stops right away. If one of its steps is executing on your endpoint at that moment, that code runs to the end, but its result is discarded. Steps, waits and context.call results of the cancelled run are never added to the new run, even though both runs share the same ID.