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

Author: octoper

ARCHITECTURE.md

@@ -352,9 +352,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
@@ -363,7 +365,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

@@ -68,7 +68,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
@@ -83,7 +83,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
     },
   },
@@ -93,8 +93,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/client/client.test.ts

@@ -402,6 +402,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 });
@@ -432,6 +451,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 });

packages/openworkflow/client/client.ts

@@ -286,25 +286,6 @@ function resolveWorkflowConcurrency<Input>(
     };
   }
 
-  let keyValue: unknown;
-  try {
-    keyValue =
-      typeof concurrency.key === "function"
-        ? concurrency.key({ input })
-        : concurrency.key;
-  } catch (error) {
-    throw new Error(
-      `Failed to resolve concurrency key for workflow "${workflowName}"`,
-      { cause: error },
-    );
-  }
-
-  if (typeof keyValue !== "string" || keyValue.trim().length === 0) {
-    throw new Error(
-      `Invalid concurrency key for workflow "${workflowName}": expected a non-empty string`,
-    );
-  }
-
   let limitValue: unknown;
   try {
     limitValue =
@@ -328,6 +309,29 @@ function resolveWorkflowConcurrency<Input>(
     );
   }
 
+  let keyValue: string | null = null;
+  if (concurrency.key !== undefined) {
+    let resolvedKey: unknown;
+    try {
+      resolvedKey =
+        typeof concurrency.key === "function"
+          ? concurrency.key({ input })
+          : concurrency.key;
+    } catch (error) {
+      throw new Error(
+        `Failed to resolve concurrency key for workflow "${workflowName}"`,
+        { cause: error },
+      );
+    }
+
+    if (typeof resolvedKey !== "string" || resolvedKey.trim().length === 0) {
+      throw new Error(
+        `Invalid concurrency key for workflow "${workflowName}": expected a non-empty string`,
+      );
+    }
+    keyValue = resolvedKey;
+  }
+
   return {
     concurrencyKey: keyValue,
     concurrencyLimit: limitValue,

packages/openworkflow/core/workflow-definition.ts

@@ -15,9 +15,10 @@ export interface WorkflowConcurrencyResolverParams<Input> {
  */
 export interface WorkflowConcurrency<Input> {
   /**
-   * Bucket key used to scope concurrency for this run.
+   * Optional bucket key used to scope concurrency for this run.
+   * When omitted, runs use the default workflow+version bucket.
    */
-  readonly key:
+  readonly key?:
     | string
     | ((params: Readonly<WorkflowConcurrencyResolverParams<Input>>) => string);
   /**

packages/openworkflow/postgres/backend.ts

@@ -235,7 +235,11 @@ export class BackendPostgres implements Backend {
 
   private async acquireConcurrencyCreateLock(
     pg: Postgres,
-    params: { workflowName: string; version: string | null; key: string },
+    params: {
+      workflowName: string;
+      version: string | null;
+      key: string | null;
+    },
   ): Promise<void> {
     // Intentionally uses a different lock payload shape than claim-time locks.
     // Create-time lock serializes concurrent creates in the bucket, while
@@ -277,7 +281,7 @@ export class BackendPostgres implements Backend {
     params: {
       workflowName: string;
       version: string | null;
-      key: string;
+      key: string | null;
       limit: number;
     },
   ): Promise<void> {
@@ -288,7 +292,7 @@ export class BackendPostgres implements Backend {
       WHERE "namespace_id" = ${this.namespaceId}
         AND "workflow_name" = ${params.workflowName}
         AND "version" IS NOT DISTINCT FROM ${params.version}
-        AND "concurrency_key" = ${params.key}
+        AND "concurrency_key" IS NOT DISTINCT FROM ${params.key}
         -- Sleeping runs are excluded so long sleeps do not pin historical
         -- limits and block new run creation after config changes.
         AND "status" IN ('pending', 'running')
@@ -418,8 +422,7 @@ export class BackendPostgres implements Backend {
           AND wr."available_at" <= NOW()
           AND (wr."deadline_at" IS NULL OR wr."deadline_at" > NOW())
           AND (
-            wr."concurrency_key" IS NULL
-            OR wr."concurrency_limit" IS NULL
+            wr."concurrency_limit" IS NULL
             OR CASE
               -- cspell:ignore xact hashtextextended
               -- Serialize constrained claims per bucket. pg_try_advisory lock
@@ -447,7 +450,7 @@ export class BackendPostgres implements Backend {
                 WHERE active."namespace_id" = wr."namespace_id"
                   AND active."workflow_name" = wr."workflow_name"
                   AND active."version" IS NOT DISTINCT FROM wr."version"
-                  AND active."concurrency_key" = wr."concurrency_key"
+                  AND active."concurrency_key" IS NOT DISTINCT FROM wr."concurrency_key"
                   AND active."status" = 'running'
                   -- Candidates require available_at <= NOW(); active leased runs
                   -- require available_at > NOW(). Keep explicit self-exclusion

packages/openworkflow/postgres/postgres.test.ts

@@ -105,16 +105,22 @@ describe("postgres", () => {
         const indexes = await pg<
           {
             indexName: string;
+            indexDef: string;
           }[]
         >`
-          SELECT indexname AS "indexName"
+          SELECT
+            indexname AS "indexName",
+            indexdef AS "indexDef"
           FROM pg_indexes
           WHERE schemaname = ${schema}
             AND tablename = 'workflow_runs'
             AND indexname = 'workflow_runs_concurrency_active_idx'
         `;
         /* cspell:enable */
         expect(indexes).toHaveLength(1);
+        expect(indexes[0]?.indexDef).toMatch(
+          /WHERE\s+\((?:"concurrency_limit"|concurrency_limit)\s+IS\s+NOT\s+NULL\)/i,
+        );
       } finally {
         await dropSchema(pg, schema);
       }