fix(files): cap how long a room delivery waits for its refetches (#9427)

Each room delivery waits at most DELIVERY_SETTLE_CAP_MS (one minute, above the slowest list read) for its refetches via withinDeadline, so a refetch that never settles no longer stops every later delivery; past the cap, the next delivery's invalidation cancels and restarts a hung refetch of a query with data. Settling refetches are still waited for and never overlapped.

Author: waleedlatif1

apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx

@@ -14,6 +14,7 @@ vi.mock('@/app/workspace/providers/socket-provider', () => ({
 }))
 
 import {
+  DELIVERY_SETTLE_CAP_MS,
   INVALIDATION_COALESCE_MS,
   useWorkspaceInvalidationRoom,
 } from '@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room'
@@ -152,5 +153,19 @@ describe('useWorkspaceInvalidationRoom', () => {
       await vi.advanceTimersByTimeAsync(INVALIDATION_COALESCE_MS)
       expect(onChanged).toHaveBeenCalledTimes(2)
     })
+
+    it('delivers again once a refetch that never settles outlasts the cap', async () => {
+      const onChanged = vi.fn(() => new Promise<void>(() => {}))
+      mount('ws-1', 'workspace-files', onChanged)
+
+      broadcast('workspace-files-changed', 'ws-1')
+      await vi.advanceTimersByTimeAsync(0)
+      broadcast('workspace-files-changed', 'ws-1')
+      await vi.advanceTimersByTimeAsync(DELIVERY_SETTLE_CAP_MS - 1)
+      expect(onChanged).toHaveBeenCalledTimes(1)
+
+      await vi.advanceTimersByTimeAsync(1 + INVALIDATION_COALESCE_MS)
+      expect(onChanged).toHaveBeenCalledTimes(2)
+    })
   })
 })

apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts

@@ -5,6 +5,7 @@ import { createLogger } from '@sim/logger'
 import type { RoomType } from '@sim/realtime-protocol/rooms'
 import { sleep } from '@sim/utils/helpers'
 import type { Socket } from 'socket.io-client'
+import { withinDeadline } from '@/lib/core/utils/deadline'
 import { useSocket } from '@/app/workspace/providers/socket-provider'
 
 const logger = createLogger('WorkspaceInvalidationRoom')
@@ -17,6 +18,15 @@ const JOIN_RETRY_BASE_MS = 1000
  * uploading files one by one) otherwise refetches every viewer's list once per broadcast.
  */
 export const INVALIDATION_COALESCE_MS = 1000
+/**
+ * The longest a delivery waits for its refetches. A fetch has no timeout of its own, so one that
+ * hangs would otherwise hold the room's deliveries forever. Past this, later deliveries run again:
+ * their invalidation cancels and restarts a hung refetch of a query that has data, while a hung
+ * first load is joined, as it always was, until it settles or its last observer unmounts. A refetch
+ * still progressing past the cap is restarted the same way, so the cap sits well above the
+ * slowest list read seen.
+ */
+export const DELIVERY_SETTLE_CAP_MS = 60_000
 
 interface JoinErrorPayload {
   workspaceId: string
@@ -26,8 +36,8 @@ interface JoinErrorPayload {
 }
 
 /**
- * Runs when the room's resources changed. A returned promise is awaited before the next delivery,
- * so a refetch still in flight is neither cancelled nor overlapped by the next one.
+ * Runs when the room's resources changed. A returned promise is awaited, up to
+ * {@link DELIVERY_SETTLE_CAP_MS}, before the next delivery starts.
  */
 type RoomChangedCallback = () => unknown
 
@@ -85,9 +95,11 @@ function createSharedRoomSubscription(
     try {
       do {
         changedSinceDelivery = false
-        await Promise.allSettled(
+        const settled = Promise.allSettled(
           [...callbacks.values()].map(async (group) => group.values().next().value?.())
         )
+        // `allSettled` never rejects, so the only rejection here is the cap expiring.
+        await withinDeadline(() => settled, Date.now() + DELIVERY_SETTLE_CAP_MS).catch(() => {})
         await sleep(INVALIDATION_COALESCE_MS)
       } while (changedSinceDelivery && !disposed)
     } finally {