Fix partitioned queue worker concurrency tracking (#1272)

## Summary

Fixes partitioned queue dispatch so workflows started from persisted
queue rows preserve their stored `queuePartitionKey` when they are
registered in the worker's in-memory running workflow map.

Without this, `workerConcurrency` can be ineffective for partitioned
queues: DBOS dequeues workflows for a specific partition, but then
starts them without carrying that partition key into
`registerRunningWorkflow`. The running workflow is tracked under
`queuePartitionKey = undefined`, while the next dequeue for the real
partition counts only running workflows with that partition key. As a
result, the worker can repeatedly think there are zero running workflows
for the partition and mark more workflows `PENDING` on every polling
tick.

## Context

We discovered this while running a single DBOS worker with a partitioned
queue configured with only local worker concurrency:

```ts
await DBOS.registerQueue('my-queue', {
    partitionQueue: true,
  workerConcurrency: 1000,
});
```

We enqueued a large batch of workflows using the same
`queuePartitionKey` and expected at most ~1000 workflows in `PENDING`
for that worker/partition. Instead, `PENDING` grew far above the worker
concurrency limit. The queue row was correct, the app version was
current, and there was only one worker process.

The issue appears to be that `executeWorkflowId` loads
`wfStatus.queuePartitionKey` from the database, but does not pass it
back into the internal workflow params. Later, `internalWorkflow`
registers the running workflow using
`params.enqueueOptions?.queuePartitionKey`, so the running workflow is
counted under `undefined` instead of the actual partition.

## Workaround

A workaround is to set global `concurrency` as well as
`workerConcurrency`:

```ts
await DBOS.registerQueue('my-queue', {
    partitionQueue: true,
  workerConcurrency: 1000,
  concurrency: 1000,
});
```

That works because global `concurrency` is enforced from Postgres state
and the SQL filter includes `queue_partition_key`. However, it has
additional overhead: DBOS must count currently pending workflows in the
database, use stricter transaction behavior, and coordinate through the
database even when there is only one worker. We would prefer to avoid
that overhead when `workerConcurrency` alone is sufficient.

## Fix

This PR preserves `wfStatus.queuePartitionKey` when queue-dispatched or
recovered workflows are re-executed from stored workflow status. That
ensures `registerRunningWorkflow` records the correct partition key and
`countRunningWorkflowsForQueue(queue, partitionKey)` enforces
`workerConcurrency` per partition as intended.

Also adds a regression test for a partitioned queue with
`workerConcurrency: 1` and no global `concurrency`. The first workflow
blocks across multiple polling ticks, and the test verifies that no
second workflow in the same partition starts until capacity is
available.

## Tests

```bash
./node_modules/.bin/jest tests/wfqueue.test.ts --runInBand --testNamePattern="partitioned-queue-worker-concurrency"
npm run build
```

Author: jlalmes

src/dbos-executor.ts

@@ -1148,6 +1148,8 @@ export class DBOSExecutor {
 
     // If starting a new workflow, assign a new UUID. Otherwise, use the workflow's original UUID.
     const workflowStartID = !!options?.startNewWorkflow ? undefined : workflowID;
+    const enqueueOptions =
+      wfStatus.queuePartitionKey !== undefined ? { queuePartitionKey: wfStatus.queuePartitionKey } : undefined;
 
     if (methReg?.workflowConfig) {
       return await runWithTopContext(recoverCtx, async () => {
@@ -1157,6 +1159,7 @@ export class DBOSExecutor {
             workflowUUID: workflowStartID,
             configuredInstance: configuredInst,
             queueName: wfStatus.queueName,
+            enqueueOptions,
             executeWorkflow: true,
             deadlineEpochMS: wfStatus.deadlineEpochMS,
             isRecoveryDispatch: !!options?.isRecoveryDispatch,
@@ -1189,6 +1192,7 @@ export class DBOSExecutor {
             workflowUUID: workflowStartID,
             configuredInstance: configuredInst,
             queueName: wfStatus.queueName, // Probably null
+            enqueueOptions,
             executeWorkflow: true,
             isRecoveryDispatch: !!options?.isRecoveryDispatch,
             isQueueDispatch: !!options?.isQueueDispatch,
@@ -1228,6 +1232,7 @@ export class DBOSExecutor {
             tempWfType: TempWorkflowType.send,
             workflowUUID: workflowStartID,
             queueName: wfStatus.queueName,
+            enqueueOptions,
             executeWorkflow: true,
             isRecoveryDispatch: !!options?.isRecoveryDispatch,
             isQueueDispatch: !!options?.isQueueDispatch,

tests/wfqueue.test.ts

@@ -1565,7 +1565,7 @@ describe('queue-time-outs', () => {
 
   beforeEach(async () => {
     await DBOS.launch();
-    for (const ref of [timeoutQueue, dedupRecoveryQueue, partitionQueue]) {
+    for (const ref of [timeoutQueue, dedupRecoveryQueue, partitionQueue, partitionWorkerConcurrencyQueue]) {
       await DBOS.registerQueue(ref.name, { onConflict: 'always_update', ...ref.config });
     }
   });
@@ -1792,6 +1792,10 @@ describe('queue-time-outs', () => {
     name: 'partition-queue',
     config: { partitionQueue: true, concurrency: 1, ...testPolling },
   };
+  const partitionWorkerConcurrencyQueue: QueueRef = {
+    name: 'partition-worker-concurrency-queue',
+    config: { partitionQueue: true, workerConcurrency: 1, ...testPolling },
+  };
 
   const partitionBlockedWorkflow = DBOS.registerWorkflow(
     async () => {
@@ -1809,6 +1813,19 @@ describe('queue-time-outs', () => {
     { name: 'partitionNormalWorkflow' },
   );
 
+  const partitionWorkerConcurrencyBlockingEvent = new Event();
+  const partitionWorkerConcurrencyStartedEvent = new Event();
+  let partitionWorkerConcurrencyStartedCount = 0;
+  const partitionWorkerConcurrencyWorkflow = DBOS.registerWorkflow(
+    async () => {
+      partitionWorkerConcurrencyStartedCount++;
+      partitionWorkerConcurrencyStartedEvent.set();
+      await partitionWorkerConcurrencyBlockingEvent.wait();
+      return DBOS.workflowID;
+    },
+    { name: 'partitionWorkerConcurrencyWorkflow' },
+  );
+
   test('partitioned-queues', async () => {
     const blockedPartitionKey = 'blocked';
     const normalPartitionKey = 'normal';
@@ -1871,6 +1888,38 @@ describe('queue-time-outs', () => {
     }, Error);
   });
 
+  test('partitioned-queue-worker-concurrency', async () => {
+    partitionWorkerConcurrencyBlockingEvent.clear();
+    partitionWorkerConcurrencyStartedEvent.clear();
+    partitionWorkerConcurrencyStartedCount = 0;
+
+    const partitionKey = 'same-partition';
+    const handles = await Promise.all(
+      Array.from({ length: 3 }, () =>
+        DBOS.startWorkflow(partitionWorkerConcurrencyWorkflow, {
+          queueName: partitionWorkerConcurrencyQueue.name,
+          enqueueOptions: { queuePartitionKey: partitionKey },
+        })(),
+      ),
+    );
+
+    try {
+      await partitionWorkerConcurrencyStartedEvent.wait();
+      await sleepms(500);
+      expect(partitionWorkerConcurrencyStartedCount).toBe(1);
+      const statuses = await Promise.all(handles.map((h) => h.getStatus()));
+      expect(statuses.filter((s) => s?.status === StatusString.PENDING)).toHaveLength(1);
+      expect(statuses.filter((s) => s?.status === StatusString.ENQUEUED)).toHaveLength(2);
+      for (const status of statuses) {
+        expect(status?.queuePartitionKey).toBe(partitionKey);
+      }
+    } finally {
+      partitionWorkerConcurrencyBlockingEvent.set();
+    }
+
+    await Promise.all(handles.map((h) => h.getResult()));
+  });
+
   test('explicit-queue-listen-test', async () => {
     await DBOS.shutdown();
     const config = generateDBOSTestConfig();