Concurrency Controls

[pranaygp]

Concurrency

Updated RFC

In light of a comment from @ale-vigano below, we decided to completely redo the concurrency spec. The idea is extremely simple now. workflow will simply expose a native lock primitive that acts as a distributed semaphore and is meant to be used inside "workflow" functions.

async function foo() {
  "use workflow"
  await using lock("key", { concurrency: 30 })
  // continues when we get to acquire the lock, only 30 at a time
}

Similar to hook and sleep, lock will consume a a concurrency slot on a queue while it is "in scope", and if no concurrency slots are available, it will suspend the workflow, and resume when there is one.

It turns out that we can model the various variations of multi tenancy queuing, prority queueing, etc. as syntactic sugar on top of the lock primitive, and the lock primitive is a lot more powerful and extensible as a primitive.

We'll fill out the details here soon with examples and the exact API spec, and how it works with the world spec to be compatible with any queue, etc.

---

Initial draft for posteritry

Copied verbatim from a draft I wrote on June 28, 2025. Some details may need a refresh and more thinking but I wanted to post this to get commentary and see if/where it breaks in the real world. Please comment on how well/poorly this would work for your use case and lets factor it into the actual implementation.

// The easiest way to limit concurrency is to set a global maxConcurrency
// directly on a step. This is useful to limit, say, the global number
// of DB connections, or requests to a provider/API, etc.

async function maxConcurrentStep() {
  "use step";
  // ...
}
maxConcurrentStep.maxConcurrency = 100;

// When used on a workflow, each workflow has to complete (or fail/cancel)
// to allow more workflows to occupy concurrency slots. Use this carefully
// with long running workflows, or workflows that never end, as it will
// prevent new workflow executions.

async function maxConcurrentWorkflow() {
  "use workflow";
  // ...
}
maxConcurrentWorkflow.maxConcurrency = 100;

// Concurrency can be customized during invocation. The global limits
// set on the step/worklfow will still be respected, but we can set
// tighter dynamic limits. This is often useful when providing different
// constraints for different customers/tiers and also for implementing
// per-tenant concurrency constraints so that no single customer clogs
// up expensive resources.

import { useStep } from "workflow"

async function parentWorkflow() {
  "use workflow";
  
  // `useStep` allows us to configure various aspects of step invocation.
  // Invoking a step directly is syntactically equivalent to using useStep
  // with all the default configurations
  const freeTierStep = useStep(maxConcurrentStep, {
	  // The key creates a new concurrency queue/bucket for the free tier
		concurrencyKey: "free-tier" 
		maxConcurrency: 10
  });
  
  // While free tier shares a global concurrency. Pro members
  // members each get their own dedicated queues for faster processing
  const proTierStep = useStep(maxConcurrentStep, {
	  // We can just include a user ID in the concurrency key
		concurrencyKey: `${user.id}-pro-tier`
		maxConcurrency: 5
  });
  
  // Now, we can just execute the steps like any other function
  await freeTierStep();
  await proTierStep();
}

// An astute reader might notice in the above setup, a free user
// could potentially run 10 parallel steps if there are no other
// free users. Twice as much as the pro tier. So, to be more fair,
// let's have it so each free tier member has maxConcurrency of 1
// AND all free tiers together share a maxConcurrency of 20, so that
// pro tier users will never be blocked (by the global step concurrency)

// To do this, we can rewrite the steps to have 2 separate global queues
// and use those:

async function doExpensiveThing() {
  // A regular function
}

// Free tier will share a max pool of 20
async function doExpensiveThingFreeTier() {
  "use step";
  return doExpensiveThing();
}
doExpensiveThingFreeTier.maxConcurrency = 20

// And we reserve a capacity of 80 for pro tier users
async function doExpensiveThingProTier() {
  "use step";
  return doExpensiveThing();
}
doExpensiveThingProTier.maxConcurrency = 80


// Now the workflow:

async function parentWorkflow() {
  "use workflow";
  
  // Each free tier user only gets 1 max concurrency
  const freeTierStep = useStep(doExpensiveThingFreeTier, {
		concurrencyKey: user.id
		maxConcurrency: 1
  });
  
	// While pro tier users get 5 each
  const proTierStep = useStep(doExpensiveThingProTier, {
	  // We can just include a user ID in the concurrency key
		concurrencyKey: user.id
		maxConcurrency: 5
  });
  
  // ...
}

---

Note: useStep is a placeholder name. It's not a replacement for "use step" which still needs to be used to properly demarcate step functions to the bundler. It's just a function that lets you augment the declared behavior of a step at invoke time. For instance, it could even be used to augment the default maxRetries. Open to naming suggestions :)

[pranaygp]

Another note: .maxConcurrency and .maxRetries might become a setDefaults({ maxConcurrency, maxRetries }) macro function you can call exactly once inside the step itself. It's functionally equivalent, simple syntactic sugar, and then we get type safety and IDE completions on it

[eersnington]

Leaning towards a setDefaults() way within the step too.

async function maxConcurrentStep() {
  "use step";
  // ...
}
maxConcurrentStep.maxConcurrency = 100;

a problem I see with this is that a dev can define these configuration anywhere in the file, and might cause confusion within a team. I prefer everything about how a step behaves to be visible inside it

[eersnington]

Any thoughs on something like this?

import { maxConcurrency } from "workflow"

async function maxConcurrentSteps() {
  "use step";
  maxConcurrency({ key: "freetier", limit: 20 }) // define limits here
  // .. expensive thing
}

With this, you can have also global configuration within the workflow config

[ale-vigano]

I think @eersnington has a point, I thought the same on a first read

Maybe:

{
"use workflow"
await using lock("key", maxConcurrent: 30)
// continues when we get to acquire the lock, only 30 at a time
}

This is possible to do now now with a simple hook (except for the using but it could help to make sure it releases)

[pranaygp]

hmm honestly this sounds like an even cleaner interface...

it reminds me of async-sema and works really well across steps and workflows. supports declarative defaults and runtime changes and gives you a lot more flexibility

the downside is that you still have to actually run the step just to check the lock? the compile time approach offloads the behavior into the queue implementation itself and avoids having to actually spin up processes

maybe if we can make this work like a macro at compile time we can support something really elegant (think also: imort and import(), similar syntax for build time and runtime module usage)

[ale-vigano]

the downside is that you still have to actually run the step just to check

No if this is implemented on the workflow alone.

I had encountered feelings about the step declaring it's own retries and now it seems it will also declare concurrency, to me that's workflow/caller logic/config.

🙋🏻 I'd personally put everything related to orchestration back into the workflow code, making the call flow more explicit and steps simpler (at risk of verbosity)

[pranaygp]

Interesting.... I like the idea and I'm gonna cook more on this 🤔

[pranaygp]

@ale-vigano thank you so much for this thought! It led to an interesting discussion we're really excited about. I've just updated the RFC

[ale-vigano]

I’m glad I was able to help!

[Pom4H]

You could also lean on default params as a discoverable place for “static” config and let the bundler pick that up instead of extra APIs.

async function myWorkflow(...args, options = { maxConcurrency: 10 }) {
  "use workflow";
}

async function myStep(...args, options = { maxConcurrency: 5 }) {
  "use step";
}

The idea:
• If the last parameter is an options object literal with known keys (maxConcurrency, maxRetries, etc.), the bundler treats it as compile-time defaults.
• At runtime, userland can still override by passing options explicitly, without introducing yet another config surface.
• Semantics stay local and readable: you see how the step/workflow behaves right where it’s defined, but you’re not forced into a custom DSL or decorators.

Feels like a light-weight middle ground between .maxConcurrency properties and an in-body setDefaults() call.

[Pom4H]

initially thought of this too and I really like but I concluded that it felt too disruptive because we're literally hacking into how you'd expect JS to work.

say you skipped reading the chapter in the docs about this stuff and to just included the options in a function implementation, you'd be so confused on what's going on

I could be dissuaded though - just my thought

what about these arguments in the start() function?

Like the super function in a constructor in a class?

[pranaygp]

Right but you're saying that the last option on any step/workflow is automatically considered workflow specific options?

like, what if I'm building some other business step in my app:

// Will try the request 3 times in the SAME process, not across processes
async function databaseRequest(query: string, options: { maxRetries: number } = { maxRetries = 3 }) {
  "use step";
  for (const i = 0; i < maxRetries; i++) {
    const { response } =  await db.sql(query);
    if (response !== null) return response
  }
  throw new Error("failed")
}

In this code, maxRetries is actually user thing and nothing to do with actual step retries. Maybe I'm making a request to something external API that natively supports retrying and I actually want to pass maxRetries to that backend.

I think it's extremely paradigm breaking if we just take control of the last argument in your function

[pranaygp]

everything inside workflow and step is still "just javascript" btw. use step and use workflow don't break that - it just moves where the code gets executed - but the javascript code you read in a workflow should work exactly like you would expect javascript to work. workflow doesn't change the semantics of the language itself, just the runtime/environment

[pranaygp]

this would also break the spread operator completely

myStep(...[...]); // last element in the array becomes the options?

async function anotherStep(...args: any[]) {
  "use step";
  
  // what is the last value in args here?
  // also what is `arguments` global refer to
}

This approach doesn't really seem to work

[Pom4H]

I agree—JS semantics shouldn’t be hijacked. The “last-param as options” trick leaks into core language expectations.

A lightweight alternative that keeps everything explicit and opt-in is a tiny, branded helper the bundler recognizes—no special rules for function parameters, no implicit “last argument” magic.

// library side
declare const WORKFLOW_OPTIONS: unique symbol;

type WorkflowConfig = {
  maxConcurrency?: number;
  maxRetries?: number;
  concurrencyKey?: string;
};

type Branded<T extends WorkflowConfig> =
  Readonly<T> & { readonly [WORKFLOW_OPTIONS]: true };

/** Mark object literal as workflow config (compile-time signal) */
export function options<T extends WorkflowConfig>(cfg: T): Branded<T> {
  return cfg as Branded<T>; // runtime: plain object; compile-time: branded
}

Usage

import { options } from "workflow";

async function myWorkflow(
  // signature stays regular JS; default is explicit
  cfg = options({ maxConcurrency: 10 })
) {
  "use workflow";
}

Step with defaults:

async function myStep(
  cfg = options({ maxConcurrency: 5, maxRetries: 3 }),
  ...args: unknown[]
) {
  "use step";
}

Call-site override (override wins):

await myStep(
  options({ maxConcurrency: plan === "pro" ? 10 : 2 }),
  payload
);
// Plain object (type error; not branded)
await myStep({ maxConcurrency: 10 });

Symmetry with start():

await start(myWorkflow, args, options({ maxConcurrency: 20 }));

This keeps semantics pure, avoids last-arg ambiguity, and still gives a small, discoverable surface for static concurrency/retry config—co-located, overridable, and easy to reason about.

[cramforce]

What is the default concurrency key? Does it include the deployment ID or is it purely derived from the function and file name?

[eersnington]

this can be dervied from the stepid exposed in getStepMetadata() function right, for steps?

[eersnington]

this can be dervied from the stepid exposed in getStepMetadata() function right, for steps?

thought about something like this initially but this is a bit verbose

import { getStepMetadata, maxConcurrency } from "workflow";

async function expensiveStep(userId: string, amount: number) {
  "use step";
  const { stepId } = getStepMetadata();
  maxConcurrency({ limit: 50, key: `expensive-step:${stepId}`});
// ..
}

but since metadata is already exposed within a step function, why not make allow maxConcurrency to have access step metadata, and use stepId as the key?

import { maxConcurrency } from "workflow";

async function expensiveStep(userId: string, amount: number) {
  "use step";
  maxConcurrency({ limit: 50 }); // default key = current stepId automatically
// ..
}

[pranaygp]

@cramforce in my proposal, it's the stepId. So it stays consistent across versions but unstable across renames etc

But you make a good prompt here. Maybe we should not have a default key at all for concurrency 🤔

[cramforce]

Maybe the real solution is to allow me to override the step ID

[eersnington]

actually, what happens when there's a new deployment and if the concurrency key is derived from stepId?

[nemrow]

I agree with @eersnington on it being contained within the function itself. However I like @pranaygp's proposed setDefaults macro function. To me the most intuitive syntax would be:

import { setDefaults } from "workflow"

async function step() {
  "use step";
  setDefaults({ maxConcurrency: 5, maxRetries: 1 })
  // .. 
}

Im assuming more default options are not too far down the road

[pranaygp]

hmm the goal is to (eventually) have priority queues. some default behavior exists on how tasks get reprioritized but you'd have finer control on choosing how to retry

you could always customize retry behavior by setting maxRetries to 1, and having a wrapping "higher order step" that handles requeueing failed tasks in the workflow context (and then you ship that higher order retrying as an npm package)

[pranaygp]

for the sake of this RFC, I haven't decided. what would you prefer - retries happen immediately or go to the back of the queue?

[MattPua]

I might be an outlier, but i feel like retries shouldn't necessarily be immediate and instead be back of the queue.

I consider longer running tasks that actually are big resource consumers, and take a long time. If the retry is immediate, it may favour retrying a really long task again, which could further increase the backlog of queued tasks.

But if it was FILO, than you dont worry about making that backlog too big

[pranaygp]

yeah back of the queue makes sense (but ideally still a priority queue with good defaults)

front of the queue would mean a bug could cause you to hang everything else in the queue constantly. especially if you set .maxRetries = Infinity

[eersnington]

+1 this is good cause workflows will have similar high load and lowload behavior and remain stable when shit goes down

[nemrow]

Can these defaults be changed on the fly? Does this scenario play out properly in your head?

import { setDefaults } from "workflow"
import { customerAConcurrency } from "../customer-configs"

async function parentWorkflow() {
  "use workflow";
  setDefaults({ maxConcurrency: customerAConcurrency })
  // ...
}

Assuming yes; if the customerAConcurrency goes from 100 to 50 while there are 100 currently running, what is the expected behavior for the extra 50 workflows already running in that queue?

[pranaygp]

great question I should've answered in the main RFC

I think the answer is you can't change the default on a step once you've shipped a version. It should be a build time failure, and you can fix it by renaming the key (or more aggressively renaming the step function itself)

the thinking here was that we could bootstrap the queues needed with their setting at build time. you can imagine that setting this concurrency config literally causes an sqs queue to be created in the aws world (via terraform/sst/cdk/whatever)

[pranaygp]

but maybe a better dx is that we actually allow you to change it and it will update the constraint on the target queue and leave it to the world's queue implementation on how to handle it. i expect most queues can gracefully handle increasing and decreasing concurrency settings and draining in flight requests.

[nemrow]

Unless there are some other implications I am not predicting, I feel the better dx (selfishly) would be for more flexibility and allow it to change during runtime.

[MattPua]

One other thing I'm curious about, is if there would be a way to move the priority of a specific step.

So if we have global workflow limits that are currently in place, and the queue is growing, but theres one particular job for one particular key that I want to expedite, could there be a mechanism to do that.

[pranaygp]

yup. It's not in this RFC yet but priority queueing does seem pretty important - #301 (reply in thread)

[MattPua]

is there also a general max number of total steps that can be executed as a general infra constraint?

[pranaygp]

Within a run? or globally?
globally, no such limit on Vercel world- but you do pay for every step invocation

within a run - there is a max size on the event log, so if the event log grows too large it can cause the replays to become incredibly slow and effectively non performant. We don't yet have concrete numbers on what this max event log size but it hasn't been prohibitive on most nearly any use case we're seeing on workflow yet. We'll need to do another benchmark soon and tackle a lot of more low hanging fruit on performance

[MattPua]

What use case is there for max retries infinite? Seems like a good way to get a runaway bill unexpectedly

…

On Tue, Nov 11, 2025, 8:10 p.m. Pranay Prakash ***@***.***> wrote: yup. It's not in this RFC yet but priority queueing does seem pretty important - #301 (reply in thread) <#301 (reply in thread)> — Reply to this email directly, view it on GitHub <#301 (reply in thread)>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AAZYLDTLYW6FF542VPTDKSD34KCHTAVCNFSM6AAAAACLXOAOE2VHI2DSMVQWIX3LMV43URDJONRXK43TNFXW4Q3PNVWWK3TUHMYTIOJUGE2DAMY> . You are receiving this because you commented.Message ID: ***@***.***>

[pranaygp]

You can think of using maxRetries = Infinity along with return/throw FatalError as similar to doing while(true) with a break in javascript. There are some patterns where you might want to do that 🤷

While it's mose illustrative, the initial implementation of sleep was actually a step that used RetryableError with delays to keep sleeping long enough till the time passed and then it just returned. We used Infinity to make it work forever

but you never have to

[eersnington]

been thinking about this point -> how would maxConcurrency interact with long running or suspended workflows (sleep("time") api)?

[zhm]

would this become a recommended way to have “unique jobs” (one per organization, one per slack channel, etc)? Instead of a singleton workflow + infinite loop + hook resume, you can use a concurrency of 1 and fire a new workflow? One of the things I’ve been trying to figure out with infinite workflows + hooks is figuring out when/where to start() them so you know they’re always running by the time you resume them.