fix(rate-limiter): hosted-key queue follow-ups from #4756 review

Author: TheodoreSpeaks

apps/sim/background/workflow-column-execution.ts

@@ -70,6 +70,10 @@ export async function executeWorkflowGroupCellJob(
       ...currentPayload,
       groupId: next.id,
       workflowId: next.workflowId,
+      // Re-derive from the target group rather than inheriting the prior group's
+      // value via the spread: a workflow group following an enrichment group would
+      // otherwise carry a stale enrichmentId.
+      enrichmentId: next.enrichmentId,
       executionId: generateId(),
     }
   }

apps/sim/lib/core/rate-limiter/hosted-key/hosted-key-rate-limiter.ts

@@ -65,13 +65,18 @@ function interruptibleSleep(ms: number, signal?: AbortSignal): Promise<void> {
   return new Promise<void>((resolve) => {
     const onAbort = () => {
       clearTimeout(timer)
+      signal.removeEventListener('abort', onAbort)
       resolve()
     }
     const timer = setTimeout(() => {
       signal.removeEventListener('abort', onAbort)
       resolve()
     }, ms)
     signal.addEventListener('abort', onAbort, { once: true })
+    // Re-check after registering: if the signal fired between the guard above and
+    // addEventListener, the 'abort' event already dispatched and our listener would
+    // never run, leaving the sleep to run its full duration. onAbort is idempotent.
+    if (signal.aborted) onAbort()
   })
 }
 

apps/sim/lib/core/rate-limiter/hosted-key/queue.test.ts

@@ -170,24 +170,32 @@ describe('HostedKeyQueue', () => {
   })
 
   describe('refreshHeartbeat', () => {
-    it('writes the heartbeat key with TTL', async () => {
-      mockRedis.set.mockResolvedValueOnce('OK')
+    it('writes the heartbeat key with TTL and re-extends the queue list TTL', async () => {
+      mockRedis.pipeline.exec.mockResolvedValueOnce([
+        [null, 'OK'],
+        [null, 1],
+      ])
 
       await queue.refreshHeartbeat(provider, workspaceId, ticketId)
 
-      expect(mockRedis.set).toHaveBeenCalledWith(
+      expect(mockRedis.pipeline.set).toHaveBeenCalledWith(
         'hosted-queue-tkt:exa:workspace-1:ticket-1',
         '1',
         'EX',
         expect.any(Number)
       )
+      expect(mockRedis.pipeline.expire).toHaveBeenCalledWith(
+        'hosted-queue:exa:workspace-1',
+        expect.any(Number)
+      )
+      expect(mockRedis.pipeline.exec).toHaveBeenCalledTimes(1)
     })
 
     it('is a no-op when Redis is unavailable', async () => {
       redisConfigMockFns.mockGetRedisClient.mockReturnValueOnce(null)
 
       await expect(queue.refreshHeartbeat(provider, workspaceId, ticketId)).resolves.toBeUndefined()
-      expect(mockRedis.set).not.toHaveBeenCalled()
+      expect(mockRedis.multi).not.toHaveBeenCalled()
     })
   })
 

apps/sim/lib/core/rate-limiter/hosted-key/queue.ts

@@ -15,8 +15,11 @@ const TICKET_HEARTBEAT_TTL_SECONDS = 30
 export const HEARTBEAT_REFRESH_INTERVAL_MS = 10_000
 
 /**
- * TTL on the queue list itself. Set on every enqueue. Prevents abandoned queues
- * (whole workspace went silent) from sticking around forever in Redis.
+ * TTL on the queue list itself. Set on every enqueue and re-extended by the head's
+ * heartbeat while it actively waits, so a long-waiting head (whose budget can exceed
+ * this TTL for enterprise async runs) never lets the list expire out from under the
+ * waiters behind it. Prevents abandoned queues (whole workspace went silent) from
+ * sticking around forever in Redis.
  */
 const QUEUE_LIST_TTL_SECONDS = 600
 
@@ -148,7 +151,11 @@ export class HostedKeyQueue {
 
   /**
    * Refresh the ticket's heartbeat. Called periodically by the head while it's
-   * waiting on the bucket so it doesn't get reaped as dead.
+   * waiting on the bucket so it doesn't get reaped as dead. Also re-extends the
+   * queue list TTL: a head whose wait outlives {@link QUEUE_LIST_TTL_SECONDS}
+   * (possible for long enterprise async budgets) would otherwise let the list
+   * expire with no new enqueue to refresh it, dropping every waiter to "missing"
+   * and collapsing FIFO ordering into concurrent bucket racing.
    */
   async refreshHeartbeat(
     provider: string,
@@ -158,9 +165,13 @@ export class HostedKeyQueue {
     const redis = getRedisClient()
     if (!redis) return
 
+    const listKey = queueListKey(provider, billingActorId)
     const hbKey = heartbeatKey(provider, billingActorId, ticketId)
     try {
-      await redis.set(hbKey, '1', 'EX', TICKET_HEARTBEAT_TTL_SECONDS)
+      const pipeline = redis.multi()
+      pipeline.set(hbKey, '1', 'EX', TICKET_HEARTBEAT_TTL_SECONDS)
+      pipeline.expire(listKey, QUEUE_LIST_TTL_SECONDS)
+      await pipeline.exec()
     } catch (error) {
       logger.warn(`Queue heartbeat refresh failed for ${hbKey}`, {
         error: toError(error).message,