fix: Catch and surface body consumption errors in regular flow (#2500)

Closes #2497

- Wrap fetch such that we ensure that we can not only complete the
request but also consume the fetch body in full - and any errors arising
from that are still surfaced as `FetchError`
- Always throw errors from our fetch even if the status code is not in
the `>= 400 && < 500` range - if we don't, it fails in the next step as
`response` is undefined.

In this way, errors during the consumption of the body stream are
handled the same as other network errors, interrupting the stream,
notifying subscribers, etc.

Author: msfstef

.changeset/forty-chefs-cheer.md

@@ -0,0 +1,5 @@
+---
+"@electric-sql/client": patch
+---
+
+Surface errors from consuming response body as `FetchError`s in regular handling flow

packages/sync-service/.env.dev

@@ -10,3 +10,4 @@ OTHER_DATABASE_URL=postgresql://postgres:password@localhost:54322/electric?sslmo
 ELECTRIC_PROFILE_WHERE_CLAUSES=false
 ELECTRIC_OTEL_SAMPLING_RATIO=1
 ELECTRIC_OTEL_DEBUG=false
+ElECTIRC_INSECURE=true

packages/typescript-client/src/client.ts

@@ -21,6 +21,7 @@ import {
   BackoffOptions,
   createFetchWithBackoff,
   createFetchWithChunkBuffer,
+  createFetchWithConsumedMessages,
   createFetchWithResponseHeadersCheck,
 } from './fetch'
 import {
@@ -368,8 +369,10 @@ export class ShapeStream<T extends Row<unknown> = Row>
       },
     })
 
-    this.#fetchClient = createFetchWithResponseHeadersCheck(
-      createFetchWithChunkBuffer(fetchWithBackoffClient)
+    this.#fetchClient = createFetchWithConsumedMessages(
+      createFetchWithResponseHeadersCheck(
+        createFetchWithChunkBuffer(fetchWithBackoffClient)
+      )
     )
   }
 
@@ -510,12 +513,13 @@ export class ShapeStream<T extends Row<unknown> = Row>
             this.#reset(newShapeHandle)
             await this.#publish(e.json as Message<T>[])
             continue
-          } else if (e.status >= 400 && e.status < 500) {
+          } else {
             // Notify subscribers
             this.#sendErrorToSubscribers(e)
 
-            // 400 errors are not actionable without additional user input,
-            // so we exit the loop
+            // errors that have reached this point are not actionable without
+            // additional user input, such as 400s or failures to read the
+            // body of a response, so we exit the loop
             throw e
           }
         } finally {

packages/typescript-client/src/error.ts

@@ -33,10 +33,12 @@ export class FetchError extends Error {
     let json: object | undefined = undefined
 
     const contentType = response.headers.get(`content-type`)
-    if (contentType && contentType.includes(`application/json`)) {
-      json = (await response.json()) as object
-    } else {
-      text = await response.text()
+    if (!response.bodyUsed) {
+      if (contentType && contentType.includes(`application/json`)) {
+        json = (await response.json()) as object
+      } else {
+        text = await response.text()
+      }
     }
 
     return new FetchError(status, text, json, headers, url)

packages/typescript-client/src/fetch.ts

@@ -95,6 +95,32 @@ export function createFetchWithBackoff(
   }
 }
 
+// Ensure body can actually be read in its entirety
+export function createFetchWithConsumedMessages(fetchClient: typeof fetch) {
+  return async (...args: Parameters<typeof fetch>): Promise<Response> => {
+    const url = args[0]
+    const res = await fetchClient(...args)
+    try {
+      if (res.body === null) return res
+      const text = await res.text()
+      return new Response(text, res)
+    } catch (err) {
+      throw new FetchError(
+        res.status,
+        undefined,
+        undefined,
+        Object.fromEntries([...res.headers.entries()]),
+        url.toString(),
+        err instanceof Error
+          ? err.message
+          : typeof err === `string`
+            ? err
+            : `failed to read body`
+      )
+    }
+  }
+}
+
 interface ChunkPrefetchOptions {
   maxChunksToPrefetch: number
 }

packages/typescript-client/test/integration.test.ts

@@ -866,6 +866,42 @@ describe(`HTTP Sync`, () => {
     })
   })
 
+  it(`should handle invalid requests by terminating stream`, async ({
+    expect,
+    issuesTableUrl,
+    aborter,
+  }) => {
+    let error: Error
+    const invalidIssueStream = new ShapeStream<IssueRow>({
+      url: `${BASE_URL}/v1/shape`,
+      params: {
+        table: issuesTableUrl,
+        where: `1=1`,
+      },
+      signal: aborter.signal,
+      // handle: streamState.handle,
+      onError: (err) => {
+        error = err
+      },
+      fetchClient: async (...args) => {
+        const res = await fetch(...args)
+        await res.text()
+        return res
+      },
+    })
+
+    const errorSubscriberPromise = new Promise((_, reject) =>
+      invalidIssueStream.subscribe(() => {}, reject)
+    )
+
+    await expect(errorSubscriberPromise).rejects.toThrow(FetchError)
+    expect(invalidIssueStream.error).instanceOf(FetchError)
+    expect(invalidIssueStream.isConnected()).false
+    expect(error!.message).contains(
+      `Body is unusable: Body has already been read`
+    )
+  })
+
   it(`should detect shape deprecation and restart syncing`, async ({
     expect,
     insertIssues,