feat: Update workflow concurrency to support optional keys and default bucket usage

Author: octoper

ARCHITECTURE.md

@@ -329,9 +329,11 @@ defineWorkflow(
 ```
 
 `key` and `limit` can each be either static values (`string`/`number`) or
-functions of the validated workflow input. They are resolved once when the run
-is created and persisted on `workflow_runs`.
+functions of the validated workflow input, and `key` is optional. They are
+resolved once when the run is created and persisted on `workflow_runs`.
 Resolved keys are stored verbatim; only empty/all-whitespace keys are rejected.
+When `key` is omitted, the run uses the default bucket for
+`namespace_id + workflow_name + version`.
 
 During claim/dequeue, a run is claimable only when the number of active leased
 `running` runs in the same bucket is below the run's `limit`. The bucket scope
@@ -340,7 +342,7 @@ is:
 - `namespace_id`
 - `workflow_name`
 - `version` (version-aware buckets)
-- `concurrency_key`
+- `concurrency_key` (nullable for the default bucket)
 
 `pending`, `sleeping`, and expired-lease `running` runs do not consume
 concurrency slots.

packages/docs/docs/workers.mdx

@@ -98,7 +98,7 @@ defineWorkflow(
   {
     name: "process-order",
     concurrency: {
-      key: ({ input }) => `tenant:${input.tenantId}`, // or: "tenant:acme"
+      key: ({ input }) => `tenant:${input.tenantId}`, // optional
       limit: ({ input }) => input.maxConcurrentOrders, // or: 5
     },
   },
@@ -113,7 +113,7 @@ Workers will only claim a run when the bucket has capacity. Bucket scope is:
 - namespace
 - workflow name
 - workflow version
-- resolved concurrency key
+- resolved concurrency key (or default bucket when key is omitted)
 
 Only active leased `running` runs consume workflow-concurrency slots.
 Resolved keys are stored verbatim; only empty/all-whitespace keys are rejected.

packages/docs/docs/workflows.mdx

@@ -198,7 +198,7 @@ defineWorkflow(
   {
     name: "process-order",
     concurrency: {
-      key: ({ input }) => `tenant:${input.tenantId}`, // or: "tenant:acme"
+      key: ({ input }) => `tenant:${input.tenantId}`, // optional
       limit: ({ input }) => input.maxConcurrentOrders, // or: 5
     },
   },
@@ -208,9 +208,9 @@ defineWorkflow(
 );
 ```
 
-- `key` can be a string or a function `({ input }) => string`
+- `key` is optional; when set it can be a string or `({ input }) => string`
 - `limit` can be a number or a function `({ input }) => number`
-- key must resolve to a non-empty string
+- if provided, key must resolve to a non-empty string
 - limit must resolve to a positive integer
 - resolved keys are stored verbatim; only empty/all-whitespace keys are rejected
 - within active runs (`pending`/`running`) for the same
@@ -221,7 +221,7 @@ When concurrency is configured, runs in the same bucket are constrained by:
 - namespace
 - workflow name
 - workflow version
-- resolved `key`
+- resolved `key` (or the default bucket when key is omitted)
 
 Only actively leased `running` runs consume slots. `pending`, `sleeping`, and
 expired-lease runs do not.

packages/openworkflow/README.md

@@ -67,7 +67,7 @@ For more details, check out our [docs](https://openworkflow.dev/docs).
 - ✅ **Long pauses** - Sleep for seconds or months
 - ✅ **Scheduled runs** - Start workflows at a specific time
 - ✅ **Parallel execution** - Run steps concurrently
-- ✅ **Workflow concurrency** - Limit active runs by key (static or input-based)
+- ✅ **Workflow concurrency** - Limit active runs by bucket (optional key)
 - ✅ **Idempotency keys** - Deduplicate repeated run requests (24h window)
 - ✅ **No extra servers** - Uses your existing database
 - ✅ **Dashboard included** - Monitor and debug workflows
@@ -82,7 +82,7 @@ const workflow = defineWorkflow(
   {
     name: "process-order",
     concurrency: {
-      key: ({ input }) => `tenant:${input.tenantId}`, // or: "tenant:acme"
+      key: ({ input }) => `tenant:${input.tenantId}`, // optional
       limit: ({ input }) => input.maxConcurrentOrders, // or: 5
     },
   },
@@ -92,8 +92,10 @@ const workflow = defineWorkflow(
 );
 ```
 
-`key` must resolve to a non-empty string and `limit` must resolve to a positive
-integer. Invalid values fail run creation.
+`limit` must resolve to a positive integer. If `key` is provided, it must
+resolve to a non-empty string. Invalid values fail run creation.
+When `key` is omitted, runs use the default bucket for
+`namespace + workflow + version`.
 Keys are stored verbatim (for example, `" foo "` and `"foo"` are different
 concurrency keys); only empty or all-whitespace keys are rejected.
 Sleeping runs do not consume workflow-concurrency slots until they are claimed

packages/openworkflow/backend-concurrency.ts

@@ -9,7 +9,7 @@ const INVALID_CONCURRENCY_LIMIT_TYPE_ERROR =
 export const INVALID_CONCURRENCY_LIMIT_VALUE_ERROR =
   'Invalid workflow concurrency metadata: "concurrencyLimit" must be a positive integer or null.';
 const INVALID_CONCURRENCY_PAIR_ERROR =
-  'Invalid workflow concurrency metadata: "concurrencyKey" and "concurrencyLimit" must both be null or both be set.';
+  'Invalid workflow concurrency metadata: "concurrencyLimit" must be set when "concurrencyKey" is provided.';
 export const CONCURRENCY_LIMIT_MISMATCH_ERROR =
   'Invalid workflow concurrency metadata: active runs in the same bucket must use the same "concurrencyLimit".';
 
@@ -19,7 +19,7 @@ export const CONCURRENCY_LIMIT_MISMATCH_ERROR =
 export interface ConcurrencyBucket {
   workflowName: string;
   version: string | null;
-  key: string;
+  key: string | null;
   limit: number;
 }
 
@@ -66,7 +66,7 @@ export function normalizeCreateWorkflowRunParams(
   const concurrencyLimit =
     rawConcurrencyLimit === undefined ? null : rawConcurrencyLimit;
 
-  if ((concurrencyKey === null) !== (concurrencyLimit === null)) {
+  if (concurrencyKey !== null && concurrencyLimit === null) {
     throw new Error(INVALID_CONCURRENCY_PAIR_ERROR);
   }
 
@@ -101,7 +101,7 @@ export function normalizeCreateWorkflowRunParams(
 export function toConcurrencyBucket(
   params: CreateWorkflowRunParams,
 ): ConcurrencyBucket | null {
-  if (params.concurrencyKey === null || params.concurrencyLimit === null) {
+  if (params.concurrencyLimit === null) {
     return null;
   }
 

packages/openworkflow/backend.testsuite.ts

@@ -173,7 +173,7 @@ export function testBackend(options: TestBackendOptions): void {
         expect(created.concurrencyLimit).toBeNull();
       });
 
-      test("rejects mismatched workflow concurrency metadata pairs", async () => {
+      test("rejects key-only workflow concurrency metadata", async () => {
         const base = {
           workflowName: randomUUID(),
           version: null,
@@ -193,19 +193,30 @@ export function testBackend(options: TestBackendOptions): void {
         await expect(
           Promise.resolve().then(() => backend.createWorkflowRun(keyOnly)),
         ).rejects.toThrow(
-          'Invalid workflow concurrency metadata: "concurrencyKey" and "concurrencyLimit" must both be null or both be set.',
+          'Invalid workflow concurrency metadata: "concurrencyLimit" must be set when "concurrencyKey" is provided.',
         );
+      });
+
+      test("accepts limit-only workflow concurrency metadata as default bucket", async () => {
+        const base = {
+          workflowName: randomUUID(),
+          version: null,
+          idempotencyKey: null,
+          input: null,
+          config: {},
+          context: null,
+          availableAt: null,
+          deadlineAt: null,
+        };
 
         const limitOnly = {
           ...base,
           concurrencyKey: null,
           concurrencyLimit: 1,
         };
-        await expect(
-          Promise.resolve().then(() => backend.createWorkflowRun(limitOnly)),
-        ).rejects.toThrow(
-          'Invalid workflow concurrency metadata: "concurrencyKey" and "concurrencyLimit" must both be null or both be set.',
-        );
+        const created = await backend.createWorkflowRun(limitOnly);
+        expect(created.concurrencyKey).toBeNull();
+        expect(created.concurrencyLimit).toBe(1);
       });
 
       test("rejects invalid workflow concurrency limit values", async () => {
@@ -289,6 +300,35 @@ export function testBackend(options: TestBackendOptions): void {
         await teardown(backend);
       });
 
+      test("rejects mixed concurrency limits for the same active default bucket", async () => {
+        const backend = await setup();
+        const workflowName = randomUUID();
+        const version = "v1";
+
+        await createPendingWorkflowRun(backend, {
+          workflowName,
+          version,
+          concurrencyLimit: 1,
+        });
+
+        await expect(
+          backend.createWorkflowRun({
+            workflowName,
+            version,
+            idempotencyKey: null,
+            concurrencyKey: null,
+            concurrencyLimit: 2,
+            input: null,
+            config: {},
+            context: null,
+            availableAt: null,
+            deadlineAt: null,
+          }),
+        ).rejects.toThrow(CONCURRENCY_LIMIT_MISMATCH_ERROR);
+
+        await teardown(backend);
+      });
+
       test("allows changing concurrency limit after terminal runs leave active states", async () => {
         const backend = await setup();
         const workflowName = randomUUID();
@@ -1034,6 +1074,40 @@ export function testBackend(options: TestBackendOptions): void {
         await teardown(backend);
       });
 
+      test("enforces concurrency limit for default workflow-version bucket when key is omitted", async () => {
+        const backend = await setup();
+        const workflowName = randomUUID();
+        const version = "v1";
+        const concurrencyLimit = 1;
+
+        await createPendingWorkflowRun(backend, {
+          workflowName,
+          version,
+          concurrencyLimit,
+        });
+        await createPendingWorkflowRun(backend, {
+          workflowName,
+          version,
+          concurrencyLimit,
+        });
+
+        const firstClaimed = await backend.claimWorkflowRun({
+          workerId: randomUUID(),
+          leaseDurationMs: 100,
+        });
+        expect(firstClaimed).not.toBeNull();
+        expect(firstClaimed?.concurrencyKey).toBeNull();
+        expect(firstClaimed?.concurrencyLimit).toBe(concurrencyLimit);
+
+        const blocked = await backend.claimWorkflowRun({
+          workerId: randomUUID(),
+          leaseDurationMs: 100,
+        });
+        expect(blocked).toBeNull();
+
+        await teardown(backend);
+      });
+
       test("supports limits greater than one", async () => {
         const backend = await setup();
         const workflowName = randomUUID();
@@ -1210,6 +1284,37 @@ export function testBackend(options: TestBackendOptions): void {
         await teardown(backend);
       });
 
+      test("allows claims for different versions in default bucket when key is omitted", async () => {
+        const backend = await setup();
+        const workflowName = randomUUID();
+
+        await createPendingWorkflowRun(backend, {
+          workflowName,
+          version: "v1",
+          concurrencyLimit: 1,
+        });
+        await createPendingWorkflowRun(backend, {
+          workflowName,
+          version: "v2",
+          concurrencyLimit: 1,
+        });
+
+        const firstClaimed = await backend.claimWorkflowRun({
+          workerId: randomUUID(),
+          leaseDurationMs: 100,
+        });
+        const secondClaimed = await backend.claimWorkflowRun({
+          workerId: randomUUID(),
+          leaseDurationMs: 100,
+        });
+
+        expect(firstClaimed).not.toBeNull();
+        expect(secondClaimed).not.toBeNull();
+        expect(secondClaimed?.id).not.toBe(firstClaimed?.id);
+
+        await teardown(backend);
+      });
+
       test("allows claims after the active lease expires", async () => {
         const backend = await setup();
         const workflowName = randomUUID();

packages/openworkflow/client.test.ts

@@ -354,6 +354,25 @@ describe("OpenWorkflow", () => {
     expect(handle.workflowRun.concurrencyLimit).toBe(3);
   });
 
+  test("resolves literal workflow concurrency limit without key", async () => {
+    const backend = await createBackend();
+    const client = new OpenWorkflow({ backend });
+
+    const workflow = client.defineWorkflow(
+      {
+        name: "concurrency-literal-limit-only-test",
+        concurrency: {
+          limit: 3,
+        },
+      },
+      noopFn,
+    );
+    const handle = await workflow.run({ value: 1 });
+
+    expect(handle.workflowRun.concurrencyKey).toBeNull();
+    expect(handle.workflowRun.concurrencyLimit).toBe(3);
+  });
+
   test("resolves function workflow concurrency values from parsed input", async () => {
     const backend = await createBackend();
     const client = new OpenWorkflow({ backend });
@@ -384,6 +403,33 @@ describe("OpenWorkflow", () => {
     expect(handle.workflowRun.concurrencyLimit).toBe(4);
   });
 
+  test("resolves function workflow concurrency limit from parsed input without key", async () => {
+    const backend = await createBackend();
+    const client = new OpenWorkflow({ backend });
+
+    const schema = z.object({
+      limit: z.coerce.number().int().positive(),
+    });
+
+    const workflow = client.defineWorkflow(
+      {
+        name: "concurrency-function-limit-only-test",
+        schema,
+        concurrency: {
+          limit: ({ input }) => Number(input.limit),
+        },
+      },
+      noopFn,
+    );
+
+    const handle = await workflow.run({
+      limit: "4",
+    });
+
+    expect(handle.workflowRun.concurrencyKey).toBeNull();
+    expect(handle.workflowRun.concurrencyLimit).toBe(4);
+  });
+
   test("throws when resolved workflow concurrency key is invalid", async () => {
     const backend = await createBackend();
     const client = new OpenWorkflow({ backend });