Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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)
})
})
})
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand All @@ -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
Expand All @@ -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

Expand Down Expand Up @@ -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(() => {})
Comment thread
waleedlatif1 marked this conversation as resolved.
Resolved
Comment thread
waleedlatif1 marked this conversation as resolved.
Resolved
await sleep(INVALIDATION_COALESCE_MS)
} while (changedSinceDelivery && !disposed)
} finally {
Expand Down
Loading