Merge pull request #2142 from DYNOSuprovo/fix/workspace-changes-abort-listener

fix(workspaces): clean up abort listener on retry timer and return early when aborted

docs/architecture/overview.md

@@ -125,4 +125,4 @@ The rules that keep the processes out of each other's way:
 ## Known gaps
 
 - The sources and Studio lists reload themselves on a workspace event rather than through TanStack Query: they are component state, so nothing else that shows a document or an artifact is invalidated by the same event.
-- The retry `wait()` in [`workspace-changes.ts`](../../surfsense_local/frontend/src/features/workspaces/workspace-changes.ts) never removes its `abort` listener when its timer fires, so a stream that keeps dropping, as it does while the API is down, adds one listener to the workspace's signal per attempt, about 360 an hour, until the last list leaves. It also waits out its timer when the signal is already aborted, so a stream torn down between attempts takes up to 10 seconds to stop.
+

surfsense_local/frontend/src/features/workspaces/workspace-changes.test.ts

@@ -46,6 +46,7 @@ function listen(workspaceId: number, kind: "documents" | "artifacts") {
 afterEach(() => {
   for (const unsubscribe of unsubscribes.splice(0)) unsubscribe()
   vi.unstubAllGlobals()
+  vi.useRealTimers()
 })
 
 describe("workspace changes", () => {
@@ -98,4 +99,62 @@ describe("workspace changes", () => {
       { timeout: 4000 }
     )
   })
+
+  it("removes every abort listener wait() adds across dropped streams", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] })
+    const first = eventStream()
+    const second = eventStream()
+    const third = eventStream()
+    const fetchMock = apiStreaming([
+      first.response,
+      second.response,
+      third.response,
+    ])
+
+    listen(1, "documents")
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1))
+
+    const signal = fetchMock.mock.calls[0][1]?.signal as AbortSignal
+    const addListenerSpy = vi.spyOn(signal, "addEventListener")
+    const removeListenerSpy = vi.spyOn(signal, "removeEventListener")
+
+    first.drop()
+    await vi.advanceTimersByTimeAsync(1_000)
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(2))
+
+    second.drop()
+    await vi.advanceTimersByTimeAsync(2_000)
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(3))
+
+    const abortAdds = addListenerSpy.mock.calls.filter(([event]) => event === "abort")
+    const abortRemoves = removeListenerSpy.mock.calls.filter(
+      ([event]) => event === "abort"
+    )
+
+    expect(abortAdds.length).toBeGreaterThanOrEqual(2)
+    expect(abortRemoves.length).toBe(abortAdds.length)
+  })
+
+  it("tears the subscription down during a retry wait without waiting out the timer", async () => {
+    vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout"] })
+    const stream = eventStream()
+    const fetchMock = apiStreaming([stream.response])
+
+    const unsubscribe = subscribeToWorkspaceChanges(1, "documents", vi.fn())
+    unsubscribes.push(unsubscribe)
+    await vi.waitFor(() => expect(fetchMock).toHaveBeenCalledTimes(1))
+
+    const signal = fetchMock.mock.calls[0][1]?.signal as AbortSignal
+
+    stream.drop()
+    await vi.waitFor(() => expect(vi.getTimerCount()).toBe(1))
+
+    unsubscribe()
+
+    expect(signal.aborted).toBe(true)
+    expect(vi.getTimerCount()).toBe(0)
+
+    await vi.advanceTimersByTimeAsync(10_000)
+    expect(fetchMock).toHaveBeenCalledTimes(1)
+  })
 })

surfsense_local/frontend/src/features/workspaces/workspace-changes.ts

@@ -16,16 +16,17 @@ const LAST_RETRY_MS = 10_000
 const subscriptions = new Map<number, Subscription>()
 
 function wait(milliseconds: number, signal: AbortSignal) {
+  if (signal.aborted) return Promise.resolve()
   return new Promise<void>((resolve) => {
-    const timeout = window.setTimeout(resolve, milliseconds)
-    signal.addEventListener(
-      "abort",
-      () => {
-        window.clearTimeout(timeout)
-        resolve()
-      },
-      { once: true }
-    )
+    const onAbort = () => {
+      window.clearTimeout(timeout)
+      resolve()
+    }
+    const timeout = window.setTimeout(() => {
+      signal.removeEventListener("abort", onAbort)
+      resolve()
+    }, milliseconds)
+    signal.addEventListener("abort", onAbort, { once: true })
   })
 }