optimize reconciler scan

Author: icecrasher321

apps/sim/lib/core/workspace-dispatch/adapter.ts

@@ -34,7 +34,7 @@ export interface WorkspaceDispatchStorageAdapter {
     lanes: readonly WorkspaceDispatchLane[]
   ): Promise<number>
   getGlobalQueueDepth(): Promise<number>
-  reconcileGlobalQueueDepth(): Promise<void>
+  reconcileGlobalQueueDepth(knownCount: number): Promise<void>
   popNextWorkspaceId(): Promise<string | null>
   getQueuedWorkspaceCount(): Promise<number>
   hasActiveWorkspace(workspaceId: string): Promise<boolean>

apps/sim/lib/core/workspace-dispatch/memory-store.ts

@@ -290,7 +290,7 @@ export class MemoryWorkspaceDispatchStorage implements WorkspaceDispatchStorageA
     return count
   }
 
-  async reconcileGlobalQueueDepth(): Promise<void> {
+  async reconcileGlobalQueueDepth(_knownCount: number): Promise<void> {
     // no-op: memory store computes depth on the fly
   }
 

apps/sim/lib/core/workspace-dispatch/reconciler.ts

@@ -153,9 +153,39 @@ async function reconcileWaitingWorkspaceTracking(
 }
 
 export async function reconcileWorkspaceDispatchState(): Promise<void> {
-  const activeJobs = await listDispatchJobsByStatuses(['admitting', 'admitted', 'running'])
-  const waitingJobs = await listDispatchJobsByStatuses(['waiting'])
-  const terminalJobs = await listDispatchJobsByStatuses(['completed', 'failed'])
+  const allJobs = await listDispatchJobsByStatuses([
+    'waiting',
+    'admitting',
+    'admitted',
+    'running',
+    'completed',
+    'failed',
+  ])
+
+  const activeJobs: WorkspaceDispatchJobRecord[] = []
+  const waitingJobs: WorkspaceDispatchJobRecord[] = []
+  const terminalJobs: WorkspaceDispatchJobRecord[] = []
+  let nonTerminalCount = 0
+
+  for (const job of allJobs) {
+    switch (job.status) {
+      case 'admitting':
+      case 'admitted':
+      case 'running':
+        activeJobs.push(job)
+        nonTerminalCount++
+        break
+      case 'waiting':
+        waitingJobs.push(job)
+        nonTerminalCount++
+        break
+      case 'completed':
+      case 'failed':
+        terminalJobs.push(job)
+        break
+    }
+  }
+
   let changed = false
 
   for (const record of activeJobs) {
@@ -181,7 +211,7 @@ export async function reconcileWorkspaceDispatchState(): Promise<void> {
     }
   }
 
-  await reconcileGlobalQueueDepth().catch((error) => {
+  await reconcileGlobalQueueDepth(nonTerminalCount).catch((error) => {
     logger.error('Failed to reconcile global queue depth', { error })
   })
 

apps/sim/lib/core/workspace-dispatch/redis-store.ts

@@ -363,14 +363,8 @@ export class RedisWorkspaceDispatchStorage implements WorkspaceDispatchStorageAd
     return count ? Math.max(0, Number.parseInt(count, 10)) : 0
   }
 
-  async reconcileGlobalQueueDepth(): Promise<void> {
-    const allJobs = await this.listDispatchJobsByStatuses([
-      'waiting',
-      'admitting',
-      'admitted',
-      'running',
-    ])
-    await this.redis.set(GLOBAL_DEPTH_KEY, allJobs.length)
+  async reconcileGlobalQueueDepth(knownCount: number): Promise<void> {
+    await this.redis.set(GLOBAL_DEPTH_KEY, knownCount)
   }
 
   async popNextWorkspaceId(): Promise<string | null> {

apps/sim/lib/core/workspace-dispatch/store.ts

@@ -77,8 +77,8 @@ export async function getGlobalQueueDepth(): Promise<number> {
   return getAdapter().getGlobalQueueDepth()
 }
 
-export async function reconcileGlobalQueueDepth(): Promise<void> {
-  return getAdapter().reconcileGlobalQueueDepth()
+export async function reconcileGlobalQueueDepth(knownCount: number): Promise<void> {
+  return getAdapter().reconcileGlobalQueueDepth(knownCount)
 }
 
 export async function popNextWorkspaceId(): Promise<string | null> {