feat: unify tinfoil retry stack, add timeouts, surface real errors

backend/src/config/settings.ts

@@ -105,7 +105,7 @@ const settingsSchema = z
     corsExposeHeaders: z
       .string()
       .default(
-        'set-auth-token,X-Proxy-Final-Url,X-Proxy-Passthrough-Content-Type,X-Proxy-Passthrough-Mcp-Session-Id,X-Proxy-Passthrough-Mcp-Protocol-Version,X-Proxy-Passthrough-Location,X-Proxy-Passthrough-Anthropic-Version,WWW-Authenticate',
+        'set-auth-token,X-Proxy-Final-Url,X-Proxy-Passthrough-Content-Type,X-Proxy-Passthrough-Mcp-Session-Id,X-Proxy-Passthrough-Mcp-Protocol-Version,X-Proxy-Passthrough-Location,X-Proxy-Passthrough-Anthropic-Version,WWW-Authenticate,Ehbp-Response-Nonce',
       ),
 
     // E2E encryption — when true, devices must complete the trust flow before syncing
@@ -220,7 +220,7 @@ const parseSettings = (): Settings => {
     corsAllowHeaders: process.env.CORS_ALLOW_HEADERS || '',
     corsExposeHeaders:
       process.env.CORS_EXPOSE_HEADERS ||
-      'set-auth-token,X-Proxy-Final-Url,X-Proxy-Passthrough-Content-Type,X-Proxy-Passthrough-Mcp-Session-Id,X-Proxy-Passthrough-Mcp-Protocol-Version,X-Proxy-Passthrough-Location,X-Proxy-Passthrough-Anthropic-Version,WWW-Authenticate',
+      'set-auth-token,X-Proxy-Final-Url,X-Proxy-Passthrough-Content-Type,X-Proxy-Passthrough-Mcp-Session-Id,X-Proxy-Passthrough-Mcp-Protocol-Version,X-Proxy-Passthrough-Location,X-Proxy-Passthrough-Anthropic-Version,WWW-Authenticate,Ehbp-Response-Nonce',
     e2eeEnabled: process.env.E2EE_ENABLED === 'true',
     minAppVersion: process.env.MIN_APP_VERSION || '',
     swaggerEnabled: process.env.SWAGGER_ENABLED === 'true',

backend/src/proxy/streaming.test.ts

@@ -50,6 +50,21 @@ describe('capStream', () => {
     expect(capped.bytesRead()).toBe(data.byteLength)
   })
 
+  it('forwards all bytes when the byte cap is omitted', async () => {
+    const chunks = [new Uint8Array(6), new Uint8Array(5)]
+    const aborts: string[] = []
+    const capped = capStream(makeStream(chunks), {
+      idleTimeoutMs: 5000,
+      onAbort: (reason) => aborts.push(reason),
+    })
+
+    const result = await collectStream(capped.stream)
+
+    expect(result.byteLength).toBe(11)
+    expect(aborts).toEqual([])
+    expect(capped.bytesRead()).toBe(11)
+  })
+
   it('calls onAbort("cap") and terminates when bytes exceed cap', async () => {
     const chunk1 = new Uint8Array(6)
     const chunk2 = new Uint8Array(5)
@@ -79,6 +94,74 @@ describe('capStream', () => {
     expect(aborts).toEqual(['idle'])
   })
 
+  it('errors with the configured reason when error mode expires', async () => {
+    const idleError = new DOMException('opaque stream idle timeout', 'TimeoutError')
+    const aborts: string[] = []
+    const completions: number[] = []
+    const stalled = new ReadableStream<Uint8Array>({ start() {} })
+    const capped = capStream(stalled, {
+      idleTimeoutMs: 0,
+      onIdle: 'error',
+      idleError,
+      onAbort: (reason) => aborts.push(reason),
+      onComplete: (bytesRead) => completions.push(bytesRead),
+    })
+
+    await expect(capped.stream.getReader().read()).rejects.toBe(idleError)
+    expect(aborts).toEqual(['idle'])
+    expect(completions).toEqual([0])
+  })
+
+  it('keeps an externally handled idle error pending after the source aborts', async () => {
+    const idleError = new DOMException('opaque stream idle timeout', 'TimeoutError')
+    const upstreamController = new AbortController()
+    const handledError = Promise.withResolvers<Error>()
+    const stalled = new ReadableStream<Uint8Array>({
+      start(controller) {
+        upstreamController.signal.addEventListener('abort', () => controller.error(idleError), { once: true })
+      },
+    })
+    const capped = capStream(stalled, {
+      idleTimeoutMs: 0,
+      onIdle: 'error',
+      idleError,
+      onAbort: () => upstreamController.abort(idleError),
+      onIdleError: (error) => handledError.resolve(error),
+    })
+    const reader = capped.stream.getReader()
+    const pendingRead = reader.read()
+
+    expect(await handledError.promise).toBe(idleError)
+    const readState = await Promise.race([
+      pendingRead.then(
+        () => 'settled',
+        () => 'settled',
+      ),
+      new Promise<'pending'>((resolve) => setTimeout(() => resolve('pending'), 10)),
+    ])
+    expect(readState).toBe('pending')
+
+    await reader.cancel()
+    await pendingRead
+  })
+
+  it('still relays source errors before an external idle handler fires', async () => {
+    const sourceError = new Error('upstream failed')
+    const failed = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.error(sourceError)
+      },
+    })
+    const capped = capStream(failed, {
+      idleTimeoutMs: 1000,
+      onIdle: 'error',
+      onAbort: () => {},
+      onIdleError: () => {},
+    })
+
+    await expect(capped.stream.getReader().read()).rejects.toBe(sourceError)
+  })
+
   it('resets idle timer on each chunk so a slow-but-steady stream completes', async () => {
     const aborts: string[] = []
     const chunkDelay = 10
@@ -158,6 +241,27 @@ describe('capStream', () => {
     expect(aborts).toEqual([])
   })
 
+  it('clears idle timer and completes once when the source errors', async () => {
+    const sourceError = new Error('upstream failed')
+    const aborts: string[] = []
+    const completions: number[] = []
+    const failed = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.error(sourceError)
+      },
+    })
+    const capped = capStream(failed, {
+      idleTimeoutMs: 0,
+      onAbort: (reason) => aborts.push(reason),
+      onComplete: (bytesRead) => completions.push(bytesRead),
+    })
+
+    await expect(capped.stream.getReader().read()).rejects.toBe(sourceError)
+    await new Promise((resolve) => setImmediate(resolve))
+    expect(aborts).toEqual([])
+    expect(completions).toEqual([0])
+  })
+
   // ---------------------------------------------------------------------------
   // Byte counter + onComplete contract — the observability layer reads these
   // ---------------------------------------------------------------------------

backend/src/proxy/streaming.ts

@@ -2,39 +2,51 @@
  * License, v. 2.0. If a copy of the MPL was not distributed with this
  * file, You can obtain one at http://mozilla.org/MPL/2.0/. */
 
+const defaultIdleTimeoutError = new DOMException('stream idle timeout', 'TimeoutError')
+
+export type CappedStream = {
+  stream: ReadableStream<Uint8Array>
+  /** Total bytes that flowed through the cap. Read after stream completion. */
+  bytesRead: () => number
+}
+
+/** Controls byte limiting and idle-expiry behavior for a relayed stream. */
+export type CapStreamOptions = {
+  /** Maximum bytes accepted before termination. Omit to disable byte limiting. */
+  maxBytes?: number
+  idleTimeoutMs: number
+  /** Whether idle expiry terminates or errors the relayed stream. Defaults to termination. */
+  onIdle?: 'terminate' | 'error'
+  /** Error exposed to readers when `onIdle` is `error`. */
+  idleError?: Error
+  /** Handles error-mode idle expiry at the transport layer. Caller must tear
+   *  down the transport because the relayed stream stays pending. */
+  onIdleError?: (error: Error) => void
+  onAbort: (reason: 'cap' | 'idle') => void
+  /** Fired exactly once after the stream finishes (graceful close, cap-hit,
+   *  idle, source error, or downstream cancel). Receives the total bytes
+   *  that flowed through. Use for post-stream observability emission. */
+  onComplete?: (bytesRead: number) => void
+}
+
 /**
- * Wraps a ReadableStream with a byte-cap and idle-timeout TransformStream.
- * When either limit is exceeded the stream is terminated (not errored) so the
- * client receives a truncated but valid chunked response — the proxy cannot
- * retroactively change the HTTP status because headers have already been sent.
- * `onAbort` is called first so the caller can abort the upstream connection.
+ * Wraps a ReadableStream with an optional byte cap and idle watchdog.
+ * Limit expiry terminates by default because response headers have already
+ * been sent. Error mode rejects unsafe truncation by default or delegates it
+ * to a transport handler. `onAbort` fires first so callers can abort upstream.
  *
  * Returns `bytesRead()` so observability can record the actual transferred byte
  * count after the stream has been consumed. With `content-encoding` passthrough
  * the bytes counted are post-compression (what the wire saw), which is exactly
  * what we want to log.
  */
-export type CappedStream = {
-  stream: ReadableStream<Uint8Array>
-  /** Total bytes that flowed through the cap. Read after stream completion. */
-  bytesRead: () => number
-}
-
-export const capStream = (
-  source: ReadableStream<Uint8Array>,
-  opts: {
-    maxBytes: number
-    idleTimeoutMs: number
-    onAbort: (reason: 'cap' | 'idle') => void
-    /** Fired exactly once after the stream finishes (graceful close, cap-hit,
-     *  idle, source error, or downstream cancel). Receives the total bytes
-     *  that flowed through. Use for post-stream observability emission. */
-    onComplete?: (bytesRead: number) => void
-  },
-): CappedStream => {
+export const capStream = (source: ReadableStream<Uint8Array>, opts: CapStreamOptions): CappedStream => {
   let bytesReceived = 0
   let idleTimer: ReturnType<typeof setTimeout> | undefined
   let completed = false
+  let idleHandled = false
+  const idleError = opts.idleError ?? defaultIdleTimeoutError
+  const externalIdleHandler = opts.onIdle === 'error' ? opts.onIdleError : undefined
 
   const fireComplete = () => {
     if (completed) {
@@ -44,44 +56,79 @@ export const capStream = (
     opts.onComplete?.(bytesReceived)
   }
 
-  const resetIdleTimer = (controller: TransformStreamDefaultController<Uint8Array>) => {
-    clearTimeout(idleTimer)
+  const armIdleTimer = (controller: TransformStreamDefaultController<Uint8Array>) => {
+    if (idleTimer) {
+      idleTimer.refresh()
+      return
+    }
+
     idleTimer = setTimeout(() => {
+      idleTimer = undefined
+      idleHandled = externalIdleHandler !== undefined
       opts.onAbort('idle')
-      controller.terminate()
+      if (opts.onIdle !== 'error') {
+        controller.terminate()
+      } else if (externalIdleHandler) {
+        externalIdleHandler(idleError)
+      } else {
+        controller.error(idleError)
+      }
       fireComplete()
     }, opts.idleTimeoutMs)
   }
 
   const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>({
     start(controller) {
-      resetIdleTimer(controller)
+      armIdleTimer(controller)
     },
     transform(chunk, controller) {
       bytesReceived += chunk.byteLength
-      if (bytesReceived > opts.maxBytes) {
+      if (opts.maxBytes !== undefined && bytesReceived > opts.maxBytes) {
         clearTimeout(idleTimer)
+        idleTimer = undefined
         opts.onAbort('cap')
         controller.terminate()
         fireComplete()
         return
       }
       controller.enqueue(chunk)
-      resetIdleTimer(controller)
+      armIdleTimer(controller)
     },
     flush() {
       clearTimeout(idleTimer)
+      idleTimer = undefined
       fireComplete()
     },
   })
 
-  source.pipeTo(writable).catch(() => {
-    // pipeTo rejects when source errors OR when writable is aborted (e.g., downstream
-    // was cancelled). Clear the idle timer here so it doesn't fire after the stream
-    // has been torn down — running terminate() on an errored controller throws.
+  // pipeTo rejects when the source errors OR the writable is aborted (downstream
+  // cancel). Clear the idle timer so it can't fire after teardown — terminate()
+  // on an errored controller throws.
+  const finishRelay = () => {
     clearTimeout(idleTimer)
+    idleTimer = undefined
     fireComplete()
-  })
+  }
+
+  if (externalIdleHandler) {
+    /** Keeps the relayed body pending after idle expiry so Bun can reset the transport. */
+    const relaySource = async () => {
+      try {
+        await source.pipeTo(writable, { preventAbort: true, preventClose: true })
+        if (!idleHandled) {
+          await writable.close()
+        }
+      } catch (error) {
+        if (!idleHandled) {
+          await writable.abort(error).catch(() => {})
+        }
+      }
+    }
+
+    void relaySource().finally(finishRelay)
+  } else {
+    source.pipeTo(writable).catch(finishRelay)
+  }
 
   return {
     stream: readable,