fix(table): enforce shared concurrencyKey cap on database batchEnqueueAndWait

Author: TheodoreSpeaks

apps/sim/lib/core/async-jobs/backends/database.test.ts

@@ -2,6 +2,7 @@
  * @vitest-environment node
  */
 import { dbChainMock, dbChainMockFns, resetDbChainMock } from '@sim/testing'
+import { sleep } from '@sim/utils/helpers'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 vi.mock('@sim/db', () => ({
@@ -78,3 +79,36 @@ describe('DatabaseJobQueue enqueue', () => {
     })
   })
 })
+
+describe('DatabaseJobQueue batchEnqueueAndWait', () => {
+  beforeEach(() => {
+    vi.clearAllMocks()
+    resetDbChainMock()
+  })
+
+  it('caps overlapping batches sharing a concurrencyKey at the shared limit', async () => {
+    const queue = new DatabaseJobQueue()
+    let inFlight = 0
+    let maxInFlight = 0
+    const makeItem = () => ({
+      payload: {},
+      options: {
+        concurrencyKey: 'table-1',
+        concurrencyLimit: 2,
+        runner: async () => {
+          inFlight += 1
+          maxInFlight = Math.max(maxInFlight, inFlight)
+          await sleep(1)
+          inFlight -= 1
+        },
+      },
+    })
+
+    await Promise.all([
+      queue.batchEnqueueAndWait('workflow-group-cell', [makeItem(), makeItem()]),
+      queue.batchEnqueueAndWait('workflow-group-cell', [makeItem(), makeItem()]),
+    ])
+
+    expect(maxInFlight).toBe(2)
+  })
+})

apps/sim/lib/core/async-jobs/backends/database.ts

@@ -207,7 +207,23 @@ export class DatabaseJobQueue implements JobQueueBackend {
         inlineCancelKeyControllers.set(cancelKey, controller)
         tracked.push({ key: cancelKey, controller })
       }
-      return runner(item.payload, controller.signal).catch((err) => {
+      // Same shared-key semaphore as `runInline`: without it, overlapping
+      // batches on one concurrencyKey (e.g. two dispatches on one table) would
+      // each run their full window concurrently instead of sharing the cap.
+      const { concurrencyKey, concurrencyLimit } = item.options ?? {}
+      const run = async () => {
+        if (concurrencyKey && concurrencyLimit && concurrencyLimit > 0) {
+          await acquireSlot(concurrencyKey, concurrencyLimit)
+          try {
+            await runner(item.payload, controller.signal)
+          } finally {
+            releaseSlot(concurrencyKey)
+          }
+          return
+        }
+        await runner(item.payload, controller.signal)
+      }
+      return run().catch((err) => {
         logger.error(`[${type}] Inline run failed`, {
           cancelKey,
           error: toError(err).message,