Skip to content

Commit 04147f7

Browse files
committed
fix(chat): stop losing sends aborted during mount-settling
1 parent 507def6 commit 04147f7

2 files changed

Lines changed: 220 additions & 4 deletions

File tree

Lines changed: 183 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,183 @@
1+
/**
2+
* @vitest-environment jsdom
3+
*
4+
* Regression tests for the mount-settling send loss: a send started on a
5+
* fresh chat surface used to be silently dropped when React ran the unmount
6+
* cleanup mid-flight (a Suspense hide/reveal cycles every effect shortly
7+
* after Home mounts), aborting the fetch before it dispatched. The fix routes
8+
* idle sends through the durable message queue and restores the queued entry
9+
* when the cleanup abort strikes before the server received the request.
10+
*/
11+
import { act, type ReactNode } from 'react'
12+
import { sleep } from '@sim/utils/helpers'
13+
import { QueryClient, QueryClientProvider } from '@tanstack/react-query'
14+
import { createRoot, type Root } from 'react-dom/client'
15+
import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'
16+
17+
const { mockRequestJson, navigationMocks } = vi.hoisted(() => ({
18+
mockRequestJson: vi.fn(),
19+
navigationMocks: {
20+
usePathname: vi.fn(() => '/workspace/ws-1/home'),
21+
useRouter: vi.fn(() => ({ push: vi.fn(), replace: vi.fn(), prefetch: vi.fn() })),
22+
useSearchParams: vi.fn(() => new URLSearchParams()),
23+
},
24+
}))
25+
26+
vi.mock('next/navigation', () => navigationMocks)
27+
28+
vi.mock('@/lib/api/client/request', async (importOriginal) => ({
29+
...(await importOriginal<typeof import('@/lib/api/client/request')>()),
30+
requestJson: mockRequestJson,
31+
}))
32+
33+
import { useChat } from '@/app/workspace/[workspaceId]/home/hooks/use-chat'
34+
import { useMothershipQueueStore } from '@/stores/mothership-queue/store'
35+
36+
interface NetworkState {
37+
/** How the chat POST behaves for the next call. */
38+
postBehavior: 'hang' | 'accept'
39+
postCalls: number
40+
}
41+
42+
const state: NetworkState = { postBehavior: 'hang', postCalls: 0 }
43+
44+
/** An SSE response whose stream ends immediately without a terminal event. */
45+
function emptySseResponse(): Response {
46+
const stream = new ReadableStream<Uint8Array>({
47+
start(controller) {
48+
controller.close()
49+
},
50+
})
51+
return new Response(stream, { status: 200, headers: { 'Content-Type': 'text/event-stream' } })
52+
}
53+
54+
function abortError(): Error {
55+
const error = new Error('Aborted')
56+
error.name = 'AbortError'
57+
return error
58+
}
59+
60+
async function fetchStub(input: RequestInfo | URL, init?: RequestInit): Promise<Response> {
61+
const url = String(input instanceof Request ? input.url : input)
62+
63+
if (url.includes('/api/mothership/chat') && init?.method === 'POST') {
64+
state.postCalls++
65+
if (state.postBehavior === 'accept') return emptySseResponse()
66+
return new Promise<Response>((_, reject) => {
67+
const signal = init?.signal
68+
if (!signal) return
69+
if (signal.aborted) {
70+
reject(abortError())
71+
return
72+
}
73+
signal.addEventListener('abort', () => reject(abortError()), { once: true })
74+
})
75+
}
76+
77+
return new Response(JSON.stringify({ error: 'not found' }), { status: 404 })
78+
}
79+
80+
const mountedRoots: Root[] = []
81+
let queryClient: QueryClient
82+
83+
function renderUseChat(): {
84+
getResult: () => ReturnType<typeof useChat>
85+
unmount: () => void
86+
} {
87+
;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true
88+
queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } })
89+
const container = document.createElement('div')
90+
const root = createRoot(container)
91+
mountedRoots.push(root)
92+
let result: ReturnType<typeof useChat> | undefined
93+
94+
function Probe() {
95+
result = useChat('ws-1', undefined)
96+
return null
97+
}
98+
99+
act(() => {
100+
root.render(
101+
<QueryClientProvider client={queryClient}>{(<Probe />) as ReactNode}</QueryClientProvider>
102+
)
103+
})
104+
105+
return {
106+
getResult: () => {
107+
if (result === undefined) throw new Error('Hook result is not ready')
108+
return result
109+
},
110+
unmount: () => act(() => root.unmount()),
111+
}
112+
}
113+
114+
/** Every queued message across all chat keys, flattened. */
115+
function allQueuedMessages() {
116+
return Object.values(useMothershipQueueStore.getState().queues).flat()
117+
}
118+
119+
async function waitFor(predicate: () => boolean, budgetMs = 2000): Promise<void> {
120+
const deadline = Date.now() + budgetMs
121+
while (!predicate()) {
122+
if (Date.now() > deadline) throw new Error('waitFor timed out')
123+
await act(async () => {
124+
await sleep(10)
125+
})
126+
}
127+
}
128+
129+
describe('useChat mount-settling send recovery', () => {
130+
beforeEach(() => {
131+
vi.stubGlobal('fetch', fetchStub)
132+
state.postBehavior = 'hang'
133+
state.postCalls = 0
134+
mockRequestJson.mockResolvedValue({ chats: [] })
135+
useMothershipQueueStore.setState({ queues: {}, editing: {} })
136+
window.sessionStorage.clear()
137+
})
138+
139+
afterEach(() => {
140+
for (const root of mountedRoots.splice(0)) {
141+
act(() => root.unmount())
142+
}
143+
queryClient?.clear()
144+
vi.unstubAllGlobals()
145+
vi.clearAllMocks()
146+
})
147+
148+
it('restores a send the unmount cleanup aborted before the server received it', async () => {
149+
const { getResult, unmount } = renderUseChat()
150+
151+
await act(async () => {
152+
void getResult().sendMessage('hello from the palette')
153+
})
154+
await waitFor(() => state.postCalls === 1)
155+
156+
// The dispatch claimed the queue head when the optimistic send applied.
157+
expect(allQueuedMessages()).toHaveLength(0)
158+
159+
// The cleanup abort (same code path a Suspense hide/reveal runs during
160+
// mount-settling) fires while the POST is still awaiting the server.
161+
unmount()
162+
await waitFor(() => allQueuedMessages().length === 1)
163+
164+
expect(allQueuedMessages()[0].content).toBe('hello from the palette')
165+
})
166+
167+
it('does not re-queue a send the server already received', async () => {
168+
state.postBehavior = 'accept'
169+
const { getResult, unmount } = renderUseChat()
170+
171+
await act(async () => {
172+
void getResult().sendMessage('already accepted')
173+
})
174+
await waitFor(() => state.postCalls === 1)
175+
176+
unmount()
177+
await act(async () => {
178+
await sleep(50)
179+
})
180+
181+
expect(allQueuedMessages()).toHaveLength(0)
182+
})
183+
})

apps/sim/app/workspace/[workspaceId]/home/hooks/use-chat.ts

Lines changed: 37 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -1343,6 +1343,12 @@ export function useChat(
13431343
const queueDispatchActionsRef = useRef<QueueDispatchAction[]>([])
13441344
const queueDispatchTaskRef = useRef<Promise<void> | null>(null)
13451345
const queueDispatchEpochRef = useRef(0)
1346+
/**
1347+
* Set when the in-flight dispatch was killed by the unmount cleanup before
1348+
* reaching the server. Lets the restore path re-queue the message across the
1349+
* epoch bump that same cleanup performs.
1350+
*/
1351+
const restorableCleanupAbortRef = useRef(false)
13461352
const queueDispatchLoopRef = useRef<() => Promise<void>>(async () => {})
13471353
const enqueueQueueDispatchRef = useRef<(action: QueueDispatchActionInput) => Promise<void>>(
13481354
async () => {}
@@ -3465,6 +3471,8 @@ export function useChat(
34653471
: undefined
34663472

34673473
let consumedByTranscript = false
3474+
let sendReachedServer = false
3475+
let sendAbortSignal: AbortSignal | null = null
34683476

34693477
setError(null)
34703478
setTransportStreaming()
@@ -3697,6 +3705,7 @@ export function useChat(
36973705
}
36983706
const abortController = new AbortController()
36993707
abortControllerRef.current = abortController
3708+
sendAbortSignal = abortController.signal
37003709

37013710
const resourceAttachments = buildResourceAttachments(
37023711
resourcesRef.current,
@@ -3725,6 +3734,7 @@ export function useChat(
37253734
}),
37263735
signal: abortController.signal,
37273736
})
3737+
sendReachedServer = true
37283738

37293739
// Capture for propagation on side-channel calls + non-React
37303740
// tool-completion callbacks (via trace-context singleton).
@@ -3823,7 +3833,19 @@ export function useChat(
38233833
}
38243834
}
38253835
} catch (err) {
3826-
if (err instanceof Error && err.name === 'AbortError') return consumedByTranscript
3836+
if (err instanceof Error && err.name === 'AbortError') {
3837+
if (sendAbortSignal?.reason === 'unmount:client_cleanup' && !sendReachedServer) {
3838+
/* The mount-settling effect cycle (Suspense hide/reveal) ran the
3839+
unmount cleanup while this send was still pre-dispatch. Nothing
3840+
reached the server, so the send is fully recoverable: withdraw
3841+
the optimistic pair and report not-consumed so the queued entry
3842+
is restored and re-dispatched when effects re-run. */
3843+
rollbackOptimisticSend()
3844+
restorableCleanupAbortRef.current = true
3845+
return false
3846+
}
3847+
return consumedByTranscript
3848+
}
38273849
if (isStreamSchemaValidationError(err)) {
38283850
setError(err.message)
38293851
if (gen !== undefined && streamGenRef.current === gen) {
@@ -3912,9 +3934,15 @@ export function useChat(
39123934
return
39133935
}
39143936

3915-
await startSendMessage(message, fileAttachments, contexts)
3937+
/* Even an idle-path send goes through the durable queue: a direct
3938+
startSendMessage has no backing entry, so the cleanup abort that runs
3939+
when a Suspense hide/reveal cycles effects during mount-settling would
3940+
silently drop it. The dispatch loop claims the head immediately, so
3941+
the message never renders as queued. */
3942+
queueStore.enqueue(activeChatKey, createQueuedMessage(message, fileAttachments, contexts))
3943+
void enqueueQueueDispatchRef.current({ type: 'send_head' })
39163944
},
3917-
[workspaceId, startSendMessage, createQueuedMessage]
3945+
[workspaceId, createQueuedMessage]
39183946
)
39193947
useEffect(() => {
39203948
if (typeof window === 'undefined') return
@@ -4461,7 +4489,10 @@ export function useChat(
44614489
clearQueuedSendHandoffState(msg.id)
44624490
}
44634491
clearQueuedSendHandoffClaim(msg.id)
4464-
if (!removedFromQueue || options.epoch !== queueDispatchEpochRef.current) {
4492+
if (!removedFromQueue) {
4493+
return
4494+
}
4495+
if (options.epoch !== queueDispatchEpochRef.current && !restorableCleanupAbortRef.current) {
44654496
return
44664497
}
44674498
// If the user explicitly removed this message during dispatch, honor
@@ -4487,6 +4518,7 @@ export function useChat(
44874518
// between dispatch scheduling and this send.
44884519
const liveMsg = queueAtSend[currentIndex]
44894520
activeQueuedSendHandoff = options.queuedSendHandoff ?? liveMsg.queuedSendHandoff
4521+
restorableCleanupAbortRef.current = false
44904522
const consumed = await startSendMessage(
44914523
liveMsg.content,
44924524
liveMsg.fileAttachments,
@@ -4502,6 +4534,7 @@ export function useChat(
45024534
} catch {
45034535
restoreQueuedMessage(activeQueuedSendHandoff)
45044536
} finally {
4537+
restorableCleanupAbortRef.current = false
45054538
setDispatchingHeadId((current) => (current === msg.id ? null : current))
45064539
queuedMessageDispatchIdsRef.current.delete(msg.id)
45074540
userRemovedDuringDispatchRef.current.delete(msg.id)

0 commit comments

Comments
 (0)