fix: Clear react caches when stream is errored (#1898)

Fixes #1690

.changeset/quiet-bees-prove.md

@@ -0,0 +1,5 @@
+---
+"@electric-sql/react": patch
+---
+
+Clear caches when cached stream is in errored state or is explicitly aborted

packages/react-hooks/src/react-hooks.tsx

@@ -32,35 +32,49 @@ export function getShapeStream<T extends Row<unknown>>(
 ): ShapeStream<T> {
   const shapeHash = sortedOptionsHash(options)
 
-  // If the stream is already cached, return
+  // If the stream is already cached, return it if valid
   if (streamCache.has(shapeHash)) {
-    // Return the ShapeStream
-    return streamCache.get(shapeHash)! as ShapeStream<T>
-  } else {
-    const newShapeStream = new ShapeStream<T>(options)
-
-    streamCache.set(shapeHash, newShapeStream)
+    const stream = streamCache.get(shapeHash)! as ShapeStream<T>
+    if (stream.error === undefined && !stream.options.signal?.aborted) {
+      return stream
+    }
 
-    // Return the created shape
-    return newShapeStream
+    // if stream is cached but errored/aborted, remove it and related shapes
+    streamCache.delete(shapeHash)
+    shapeCache.delete(stream)
   }
+
+  const newShapeStream = new ShapeStream<T>(options)
+
+  streamCache.set(shapeHash, newShapeStream)
+
+  // Return the created shape
+  return newShapeStream
 }
 
 export function getShape<T extends Row<unknown>>(
   shapeStream: ShapeStream<T>
 ): Shape<T> {
-  // If the stream is already cached, return
+  // If the stream is already cached, return it if valid
   if (shapeCache.has(shapeStream)) {
-    // Return the ShapeStream
-    return shapeCache.get(shapeStream)! as Shape<T>
-  } else {
-    const newShape = new Shape<T>(shapeStream)
-
-    shapeCache.set(shapeStream, newShape)
+    if (
+      shapeStream.error === undefined &&
+      !shapeStream.options.signal?.aborted
+    ) {
+      return shapeCache.get(shapeStream)! as Shape<T>
+    }
 
-    // Return the created shape
-    return newShape
+    // if stream is cached but errored/aborted, remove it and related shapes
+    streamCache.delete(sortedOptionsHash(shapeStream.options))
+    shapeCache.delete(shapeStream)
   }
+
+  const newShape = new Shape<T>(shapeStream)
+
+  shapeCache.set(shapeStream, newShape)
+
+  // Return the created shape
+  return newShape
 }
 
 export interface UseShapeResult<T extends Row<unknown> = Row> {

packages/react-hooks/test/react-hooks.test.tsx

@@ -60,6 +60,37 @@ describe(`useShape`, () => {
     )
   })
 
+  it(`should re-sync a shape after an interrupt`, async ({
+    aborter,
+    issuesTableUrl,
+    insertIssues,
+  }) => {
+    const manualAborter = new AbortController()
+    renderHook(() =>
+      useShape({
+        url: `${BASE_URL}/v1/shape/${issuesTableUrl}`,
+        signal: manualAborter.signal,
+        subscribe: false,
+      })
+    )
+
+    manualAborter.abort()
+
+    const [id] = await insertIssues({ title: `test row` })
+
+    const { result } = renderHook(() =>
+      useShape({
+        url: `${BASE_URL}/v1/shape/${issuesTableUrl}`,
+        signal: aborter?.signal,
+        subscribe: false,
+      })
+    )
+
+    await waitFor(() =>
+      expect(result.current.data).toEqual([{ id: id, title: `test row` }])
+    )
+  })
+
   it(`should expose isLoading status`, async ({ issuesTableUrl }) => {
     const { result } = renderHook(() =>
       useShape({