diff --git a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx index f0f0cbeb8d8..20de27440d5 100644 --- a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx +++ b/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(() => {})) + 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) + }) }) }) diff --git a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts index 588a888233d..47e00885723 100644 --- a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts +++ b/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 {