feat: resume download on network errors (#133)

incognitojam · Copilot · web-flow · commit 1f1c3ca6c581 · 2025-04-11T18:01:00.000+01:00
* fix: retry on network errors

* wrap in error

* format

* .

* test interruption

* short and fast

* test the tests (1)

* test the tests (2)

* test the tests (3)

* test the tests (4)

* test the tests (5)

* revert

* try a few times

* rename

* thanks copilot

* less spam

* hmm ok

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;

* warn

* fewer repeats

* still works

* spaces

* ,

* test: reduce retryDelay

* cleanup

* jsdoc

* byteLength - thanks copilot

---------

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;
diff --git a/src/utils/image.js b/src/utils/image.js
@@ -1,6 +1,8 @@
 import { useEffect, useRef } from 'react'
 import { XzReadableStream } from 'xz-decompress'
 
+import { fetchStream } from './stream'
+
 /**
  * Progress callback
  *
@@ -49,21 +51,7 @@ export class ImageManager {
     }
 
     console.debug(`[ImageManager] Downloading ${image.name} from ${archiveUrl}`)
-    const response = await fetch(archiveUrl, { mode: 'cors' })
-    if (!response.ok) {
-      throw new Error(`Fetch failed: ${response.status} ${response.statusText}`)
-    }
-
-    const contentLength = +response.headers.get('Content-Length')
-    let receivedLength = 0
-    const transform = new TransformStream({
-      transform(chunk, controller) {
-        receivedLength += chunk.byteLength
-        onProgress?.(receivedLength / contentLength)
-        controller.enqueue(chunk)
-      },
-    })
-    let stream = response.body.pipeThrough(transform)
+    let stream = await fetchStream(archiveUrl, { mode: 'cors' }, { onProgress })
     try {
       if (image.compressed) {
         stream = new XzReadableStream(stream)
diff --git a/src/utils/manifest.test.js b/src/utils/manifest.test.js
@@ -4,6 +4,7 @@ import config from '../config'
 import { ImageManager } from './image'
 import { getManifest } from './manifest'
 
+const CI = import.meta.env.CI
 const MANIFEST_BRANCH = import.meta.env.MANIFEST_BRANCH
 
 const imageManager = new ImageManager()
@@ -64,7 +65,10 @@ for (const [branch, manifestUrl] of Object.entries(config.manifests)) {
           }
         })
 
-        test.skipIf(big && !MANIFEST_BRANCH)('download', { timeout: (big ? 11 * 60 : 20) * 1000 }, async () => {
+        test.skipIf(big && !MANIFEST_BRANCH)('download', {
+          timeout: (big ? 11 * 60 : 20) * 1000,
+          repeats: CI && !MANIFEST_BRANCH ? 2 : 1,
+        }, async () => {
           await imageManager.downloadImage(image)
         })
       })
diff --git a/src/utils/stream.js b/src/utils/stream.js
@@ -0,0 +1,89 @@
+/**
+ * @param {string|URL} url
+ * @param {RequestInit} [requestOptions]
+ * @param {object} [options]
+ * @param {number} options.maxRetries
+ * @param {number} options.retryDelay
+ */
+export async function fetchStream(url, requestOptions = {}, options = {}) {
+  const maxRetries = options.maxRetries || 3
+  const retryDelay = options.retryDelay || 1000
+
+  /** @param {Response} response */
+  const getContentLength = (response) => {
+    const total = response.headers.get('Content-Length')
+    const range = response.headers.get('Content-Range')
+    if (range) {
+      const match = range.match(/\/(\d+)$/)
+      if (match) return parseInt(match[1], 10)
+    }
+    if (total) return parseInt(total, 10)
+    throw new Error('Content-Length not found in response headers')
+  }
+
+  /**
+   * @param {number} startByte
+   * @param {AbortSignal} signal
+   */
+  const fetchRange = async (startByte, signal) => {
+    const headers = { ...(requestOptions.headers || {}) }
+    if (startByte > 0) {
+      headers['range'] = `bytes=${startByte}-`
+    }
+
+    const response = await fetch(url, {
+      ...requestOptions,
+      headers,
+      signal
+    })
+    if (!response.ok && response.status !== 206 && response.status !== 200) {
+      throw new Error(`Fetch error: ${response.status}`)
+    }
+    return response
+  }
+
+  return new ReadableStream({
+    start() {
+      this.startByte = 0
+      this.contentLength = null
+      this.abortController = new AbortController()
+    },
+
+    async pull(streamController) {
+      for (let attempt = 0; attempt <= maxRetries; attempt++) {
+        try {
+          const response = await fetchRange(this.startByte, this.abortController.signal)
+          if (this.contentLength === null) {
+            this.contentLength = getContentLength(response)
+          }
+
+          const reader = response.body.getReader()
+          while (true) {
+            const { done, value } = await reader.read()
+            if (done) {
+              streamController.close()
+              return
+            }
+
+            this.startByte += value.byteLength
+            streamController.enqueue(value)
+            options.onProgress?.(this.startByte / this.contentLength)
+          }
+        } catch (err) {
+          console.warn(`Attempt ${attempt + 1} failed:`, err)
+          if (attempt === maxRetries) {
+            this.abortController.abort()
+            streamController.error(new Error('Max retries reached', { cause: err }))
+            return
+          }
+          await new Promise((res) => setTimeout(res, retryDelay))
+        }
+      }
+    },
+
+    cancel(reason) {
+      console.warn('Stream canceled:', reason)
+      this.abortController.abort()
+    },
+  })
+}
diff --git a/src/utils/stream.test.js b/src/utils/stream.test.js
@@ -0,0 +1,131 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+import { fetchStream } from './stream'
+
+const encoder = new TextEncoder()
+const decoder = new TextDecoder()
+
+function mockResponse({
+  body = [],
+  headers = {},
+  status = 200,
+  failAfter = -1,
+}) {
+  const chunks = body.map(chunk => encoder.encode(chunk))
+  return {
+    ok: status >= 200 && status < 300,
+    status,
+    headers: {
+      get: (key) => headers[key.toLowerCase()] || null,
+    },
+    body: {
+      getReader() {
+        let readCount = 0
+        return {
+          read() {
+            if (failAfter >= 0 && readCount > failAfter) {
+              return Promise.reject(new Error('Network error'))
+            }
+            if (readCount < chunks.length) {
+              return Promise.resolve({ done: false, value: chunks[readCount++] })
+            }
+            return Promise.resolve({ done: true })
+          },
+        }
+      },
+    }
+  }
+}
+
+async function readText(stream) {
+  const reader = stream.getReader()
+  const chunks = []
+  while (true) {
+    const { done, value } = await reader.read()
+    if (done) break
+    chunks.push(decoder.decode(value))
+  }
+  return chunks.join('')
+}
+
+describe('fetchStream', () => {
+  const retryDelay = 1
+
+  beforeEach(() => {
+    global.fetch = vi.fn()
+  })
+
+  it('downloads content and tracks progress', async () => {
+    global.fetch.mockResolvedValueOnce(mockResponse({
+      body: ['Hello ', 'World'],
+      headers: { 'content-length': '11' },
+    }))
+
+    const progress = []
+    const stream = await fetchStream('https://example.com', {}, {
+      onProgress: progress.push.bind(progress),
+    })
+
+    expect(await readText(stream)).toBe('Hello World')
+    expect(progress[progress.length - 1]).toBe(1)
+  })
+
+  it('retries and uses Range header after failure', async () => {
+    global.fetch
+      .mockResolvedValueOnce(
+        mockResponse({
+          body: ['partial'],
+          headers: { 'content-length': '12' },
+          status: 500,
+        })
+      )
+      .mockResolvedValueOnce(
+        mockResponse({
+          body: [' data'],
+          headers: {
+            'content-length': '12',
+            'content-range': 'bytes 7-11/12',
+          },
+        })
+      )
+
+    const stream = await fetchStream('https://example.com', {}, { maxRetries: 1, retryDelay })
+
+    expect(await readText(stream)).toBe(' data')
+    expect(global.fetch).toHaveBeenCalledTimes(2)
+  })
+
+  it('resumes download when reader fails mid-stream', async () => {
+    global.fetch
+      .mockResolvedValueOnce(
+        mockResponse({
+          body: ['First part'],
+          headers: { 'content-length': '22' },
+          failAfter: 0,
+        })
+      )
+      .mockResolvedValueOnce(
+        mockResponse({
+          body: [' second part'],
+          headers: {
+            'content-length': '12',
+            'content-range': 'bytes 10-21/22',
+          },
+        })
+      )
+
+    const stream = await fetchStream('https://example.com', {}, { maxRetries: 1, retryDelay })
+
+    expect(await readText(stream)).toBe('First part second part')
+    const { headers } = global.fetch.mock.calls[1][1]
+    expect(headers['range']).toBe('bytes=10-')
+  })
+
+  it('throws after max retries', async () => {
+    global.fetch.mockRejectedValue(new Error('network error'))
+
+    const stream = await fetchStream('https://example.com', {}, { maxRetries: 2, retryDelay })
+
+    await expect(stream.getReader().read()).rejects.toThrow('Max retries reached')
+    expect(global.fetch).toHaveBeenCalledTimes(3)
+  })
+})
