diff --git a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.test.ts b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.test.ts index b0b038176bf..a8c60f3d8d1 100644 --- a/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.test.ts +++ b/apps/sim/app/workspace/[workspaceId]/home/hooks/stream/handle-resource-event.test.ts @@ -336,7 +336,7 @@ describe('recovered resource events', () => { { name: 'a loaded transcript', cached: true }, { name: 'a transcript still on its first load', cached: false }, ])( - 'cancels a stuck read of $name at most once, so a slow replacement finishes and is read again', + 'doubles the hold after replacing a stuck read of $name, so a slow replacement finishes and is read again', async ({ cached }) => { const queryClient = new QueryClient() const { server, reads, unsubscribe } = mountedChat(queryClient, { @@ -377,6 +377,49 @@ describe('recovered resource events', () => { } } ) + + it.each([ + { name: 'a loaded transcript', cached: true }, + { name: 'a transcript still on its first load', cached: false }, + ])( + 'still replaces a second stuck read of $name once its doubled hold passes', + async ({ cached }) => { + const queryClient = new QueryClient() + const { server, reads, unsubscribe } = mountedChat(queryClient, { + cached, + hangFirstRead: true, + }) + reads.mockImplementationOnce(() => new Promise(() => {})) + const ctx = createStreamLoopContext( + makeStreamLoopDeps({ queryClient, chatIdRef: ref('chat-1') }) + ) + vi.useFakeTimers({ toFake: ['Date'] }) + try { + handleResourceEvent(ctx, replayed(tables[0], 1)) + await vi.waitFor(() => expect(reads).toHaveBeenCalledOnce()) + + vi.setSystemTime(Date.now() + 30_000) + handleResourceEvent(ctx, replayed(tables[1], 2)) + await vi.waitFor(() => expect(reads).toHaveBeenCalledTimes(2)) + + vi.setSystemTime(Date.now() + 30_000) + handleResourceEvent(ctx, replayed(tables[1], 3)) + await flushMacrotask() + expect(reads).toHaveBeenCalledTimes(2) + + server.stored = tables + vi.setSystemTime(Date.now() + 30_000) + handleResourceEvent(ctx, replayed(tables[2], 4)) + + await vi.waitFor(() => expect(reads).toHaveBeenCalledTimes(3)) + expect(await settledResources(queryClient)).toEqual(tables) + expect(reads).toHaveBeenCalledTimes(3) + } finally { + vi.useRealTimers() + unsubscribe() + } + } + ) }) describe('handleResourceEvent saved-view pins', () => { diff --git a/apps/sim/hooks/queries/mothership-chats.ts b/apps/sim/hooks/queries/mothership-chats.ts index 075aba15a7c..1fff2e9885e 100644 --- a/apps/sim/hooks/queries/mothership-chats.ts +++ b/apps/sim/hooks/queries/mothership-chats.ts @@ -110,16 +110,16 @@ export const mothershipChatKeys = { } /** - * How long a chat's reload in flight holds back later requests. Past it a hung fetch no longer - * blocks the chat: the next request starts its own reload, which replaces the stuck one. + * How long a chat's first reload in flight holds back later requests. Past its hold a reload is + * presumed stuck and the next request replaces it, so a hung fetch never blocks the chat for good. */ const CHAT_DETAIL_REFRESH_HOLD_MS = 30 * 1000 interface ChatDetailRefresh { startedAt: number + /** Doubles with each stuck reload a chain replaces, so a slow load gets the time it needs. */ + holdMs: number requestedAgain: boolean - /** Whether this reload replaced one past the hold; a chain cancels a slow load at most once. */ - replaced: boolean } /** Per client, the chats with a reload in flight. */ @@ -131,33 +131,31 @@ const chatDetailRefreshes = new WeakMap() chatDetailRefreshes.set(queryClient, refreshes) const inFlight = refreshes.get(chatId) - if (inFlight && Date.now() - inFlight.startedAt < CHAT_DETAIL_REFRESH_HOLD_MS) { + if (inFlight && Date.now() - inFlight.startedAt < inFlight.holdMs) { inFlight.requestedAgain = true return } - // A query with no data yet joins its fetch in flight instead of replacing it, so a stuck - // one is cancelled first, but only once per chain: a load that is slow rather than hung - // would otherwise restart every hold and never finish. Later reloads in the chain join the - // read in flight, cached or not, and since it may predate this request, read again after. const refresh: ChatDetailRefresh = { startedAt: Date.now(), - requestedAgain: inFlight?.replaced === true, - replaced: inFlight !== undefined, + holdMs: inFlight ? inFlight.holdMs * 2 : CHAT_DETAIL_REFRESH_HOLD_MS, + requestedAgain: false, } refreshes.set(chatId, refresh) const queryKey = mothershipChatKeys.detail(chatId) - const reload = - inFlight && !inFlight.replaced - ? queryClient - .cancelQueries({ queryKey, exact: true }) - .then(() => queryClient.invalidateQueries({ queryKey })) - : queryClient.invalidateQueries({ queryKey }, { cancelRefetch: !inFlight }) + // A query with no data yet joins its fetch in flight instead of replacing it, so a stuck + // one is cancelled first. + const reload = inFlight + ? queryClient + .cancelQueries({ queryKey, exact: true }) + .then(() => queryClient.invalidateQueries({ queryKey })) + : queryClient.invalidateQueries({ queryKey }) void reload.finally(() => { if (refreshes.get(chatId) !== refresh) return refreshes.delete(chatId) diff --git a/apps/sim/lib/cleanup/chat-cleanup.integration.ts b/apps/sim/lib/cleanup/chat-cleanup.integration.ts index 8cf97ded8d2..14d8424cfef 100644 --- a/apps/sim/lib/cleanup/chat-cleanup.integration.ts +++ b/apps/sim/lib/cleanup/chat-cleanup.integration.ts @@ -1,5 +1,5 @@ /** Exercises chat cleanup's storage-key collection against real PostgreSQL message rows. */ -import { db } from '@sim/db' +import { db, dbFor } from '@sim/db' import { copilotChats, copilotMessages, user } from '@sim/db/schema' import { generateId } from '@sim/utils/id' import { compareStrings } from '@sim/utils/string' @@ -30,7 +30,7 @@ describe('chat cleanup file collection in PostgreSQL', () => { afterAll(async () => { await db.delete(copilotChats).where(eq(copilotChats.userId, userId)) await db.delete(user).where(eq(user.id, userId)) - await db.$client.end() + await Promise.all([db.$client.end(), dbFor('cleanup').$client.end()]) }) async function seedChat(contents: unknown[], deletedSeqs: number[] = []): Promise {