Disable prefetching on backup downloads

client/packages/platform/__tests__/src/backupDownload.test.ts

@@ -168,44 +168,49 @@ describe('downloadBackupArchive', () => {
     await expect(promise).rejects.toMatchObject({ name: 'AbortError' });
   });
 
-  test('fetches later entries while an earlier one is still being written', async () => {
+  test('fetches each entry body only when the writer reaches it, never ahead', async () => {
     const files = [
       { name: 'config.json', size: 1 },
       { name: 'entities/a.jsonl', size: 1 },
       { name: 'entities/b.jsonl', size: 1 },
     ];
     const started: string[] = [];
-    let resolveAllStarted!: () => void;
-    const allStarted = new Promise<void>((r) => {
-      resolveAllStarted = r;
-    });
 
     const manager = {
       listFiles: async () => files,
       getFileUrl,
-      // A fetch records that it started and, once every entry's fetch has
-      // begun, releases the writer below.
       streamStorageFiles: async function* () {},
     } as any;
+    // Records the moment a body is fetched. Downloads are strictly sequential,
+    // so this fires only when the writer reaches the entry — never ahead.
     const trackingFetch = async (url: string) => {
       started.push(url);
-      if (started.length === files.length) resolveAllStarted();
       return bodyOf(`body:${url}`);
     };
 
-    // The first entry's write can't finish until every fetch has started. A
-    // strictly sequential downloader would deadlock — the second file's fetch
-    // would wait on the first file's write, which waits on all fetches — so
-    // this test only completes because later fetches run ahead of the writer.
-    let firstAdd = true;
+    // A pair of gates per entry: `began` resolves when the encoder starts
+    // writing that entry; the encoder then blocks on `release` until the test
+    // lets it proceed. This lets us pause on each entry in turn and check that
+    // no later body has been pulled ahead — the browser-fetch bug that buffered
+    // whole entities before their turn would show a later fetch in `started`
+    // while the current write is blocked.
+    const deferred = () => {
+      let resolve!: () => void;
+      const promise = new Promise<void>((r) => {
+        resolve = r;
+      });
+      return { promise, resolve };
+    };
+    const began = files.map(deferred);
+    const release = files.map(deferred);
+    let addCount = 0;
     const createWriter = async (sink: WritableStream<Uint8Array>) => {
       const w = sink.getWriter();
       return {
         add: async (_name: string, input: ReadableStream<Uint8Array>) => {
-          if (firstAdd) {
-            firstAdd = false;
-            await allStarted;
-          }
+          const i = addCount++;
+          began[i].resolve();
+          await release[i].promise;
           for await (const _chunk of input) {
             // drain
           }
@@ -215,14 +220,25 @@ describe('downloadBackupArchive', () => {
       };
     };
 
-    await downloadBackupArchive({
+    const done = downloadBackupArchive({
       manager,
       backup,
       fetchBody: trackingFetch,
       sink: nullSink(),
       createWriter,
     });
 
+    // Walk the entries one at a time. When each write begins, exactly the
+    // bodies up to and including it have been fetched — nothing ahead. Release
+    // it and move to the next; the next body must not have been fetched until
+    // this write completed.
+    for (let i = 0; i < files.length; i++) {
+      await began[i].promise;
+      expect(started).toEqual(files.slice(0, i + 1).map((f) => f.name));
+      release[i].resolve();
+    }
+    await done;
+
     expect(started).toEqual([
       'config.json',
       'entities/a.jsonl',

client/packages/platform/src/backupDownload.ts

@@ -76,21 +76,9 @@ export type DownloadBackupArchiveOpts = {
   ) => Promise<BackupArchiveWriter>;
   signal?: AbortSignal;
   onProgress?: (progress: BackupDownloadProgress) => void;
-  /**
-   * How many entries to fetch ahead of the one currently being written into
-   * the archive.
-   *
-   * Only fetch *initiation* is parallelised, the bodies are still written in
-   * order and consumed one at a time, so an in-flight prefetched body buffers
-   * only to its stream's high-water mark and memory stays bounded. Defaults to
-   * {@link DEFAULT_PREFETCH}.
-   */
-  prefetch?: number;
   /**
    * Retry policy for *opening* an entry's body (fetching its presigned URL and
-   * getting a live response). Prefetching keeps body connections open, paused,
-   * while earlier entries write, so a queued connection can be reset before we
-   * read it.
+   * getting a live response).
    * `attempts` is the total number of tries (default {@link DEFAULT_FETCH_ATTEMPTS});
    * `delayMs` is the base for exponential backoff (default {@link DEFAULT_RETRY_DELAY_MS}).
    *
@@ -102,7 +90,6 @@ export type DownloadBackupArchiveOpts = {
   retry?: { attempts?: number; delayMs?: number };
 };
 
-const DEFAULT_PREFETCH = 4;
 const DEFAULT_FETCH_ATTEMPTS = 3;
 const DEFAULT_RETRY_DELAY_MS = 500;
 const MAX_RETRY_DELAY_MS = 5000;
@@ -172,9 +159,9 @@ function replayFrom(
  * naming the entry.
  *
  * The first chunk is read inside the retry scope, so a body that connects but
- * fails on its first read — the shape of a prefetched connection reset while it
- * sat idle — is re-fetched too, since nothing has been written to the archive
- * yet. Only failures once bytes are flowing are treated as unrecoverable.
+ * fails on its first read is re-fetched too, since nothing has been written
+ * to the archive yet. Only failures once bytes are flowing are treated as
+ * unrecoverable.
  */
 async function openWithRetry(
   open: () => Promise<ReadableStream<Uint8Array>>,
@@ -212,10 +199,9 @@ async function openWithRetry(
 }
 
 /**
- * One archive entry whose body fetch has already been started. `onWriting`
- * runs when the encoder begins consuming it (so progress reflects the entry
- * actually streaming, not one prefetched ahead); `onAdded` runs once it's
- * fully written.
+ * One archive entry whose body fetch has been started. `onWriting` runs when
+ * the encoder begins consuming it (so progress reflects the entry actually
+ * streaming); `onAdded` runs once it's fully written.
  */
 type PreparedEntry = {
   name: string;
@@ -225,93 +211,12 @@ type PreparedEntry = {
 };
 
 /**
- * A thunk that starts fetching one entry (presigned URL + body) and resolves
- * once the body stream is available — not once it's fully downloaded.
+ * A thunk that fetches one entry (presigned URL + body) and resolves once the
+ * body stream is available — not once it's fully downloaded. Called only when
+ * the entry is about to be written, so its body never downloads ahead.
  */
 type EntryThunk = () => Promise<PreparedEntry>;
 
-/**
- * Wraps an ordered stream of entry thunks, keeping up to `lookahead` fetches
- * in flight while yielding the prepared entries in their original order. A
- * background producer pulls thunks and starts their fetches as space frees up;
- * the consumer awaits each in turn. Preserves order and backpressure: at most
- * `lookahead` bodies are ever in flight, and the producer parks when the
- * pipeline is full or the source is waiting for more work.
- */
-async function* prefetchEntries(
-  thunks: AsyncIterable<EntryThunk>,
-  lookahead: number,
-): AsyncGenerator<PreparedEntry> {
-  const pipeline: Promise<PreparedEntry>[] = [];
-  const state: {
-    done: boolean;
-    producerError: unknown;
-    // Woken when the producer pushes an entry (or finishes).
-    onItem: (() => void) | null;
-    // Woken when the consumer frees a pipeline slot.
-    onSpace: (() => void) | null;
-  } = { done: false, producerError: null, onItem: null, onSpace: null };
-
-  const wakeItem = () => {
-    const w = state.onItem;
-    state.onItem = null;
-    if (w) w();
-  };
-  const wakeSpace = () => {
-    const w = state.onSpace;
-    state.onSpace = null;
-    if (w) w();
-  };
-
-  // Never rejects: a failure to produce the next thunk (or start its fetch)
-  // lands in state.producerError for the consumer to throw in order.
-  const producer = (async () => {
-    try {
-      for await (const thunk of thunks) {
-        while (pipeline.length >= lookahead) {
-          await new Promise<void>((resolve) => {
-            state.onSpace = resolve;
-          });
-        }
-        const started = thunk();
-        // The consumer awaits `started` in order; attach a no-op catch so a
-        // fetch that rejects before then isn't reported as unhandled.
-        started.catch(() => {});
-        pipeline.push(started);
-        wakeItem();
-      }
-    } catch (e) {
-      state.producerError = e;
-    } finally {
-      state.done = true;
-      wakeItem();
-    }
-  })();
-
-  try {
-    while (true) {
-      if (pipeline.length === 0) {
-        if (state.done) {
-          if (state.producerError) throw state.producerError;
-          break;
-        }
-        await new Promise<void>((resolve) => {
-          state.onItem = resolve;
-        });
-        continue;
-      }
-      const entry = await pipeline.shift()!;
-      wakeSpace();
-      yield entry;
-    }
-  } finally {
-    // On early exit (abort/error), let the producer unwind — its own signal
-    // teardown resolves the source's waits — without blocking here.
-    wakeSpace();
-    producer.catch(() => {});
-  }
-}
-
 /**
  * Downloads a backup into a single archive written to `opts.sink`: entries
  * in the canonical restore order (`config.json`, then the
@@ -459,9 +364,9 @@ export async function downloadBackupArchive(
   // entities/*.jsonl shards, then files/<locationId>. In particular ALL entity
   // files must be written before ANY storage file. listFiles returns the
   // entity files in write order; this generator yields thunks for them in
-  // order, then drains the storage queue. The thunks are consumed through
-  // prefetchEntries, which starts a bounded number of the fetches ahead of the
-  // encoder while keeping this order and one-at-a-time writing.
+  // order, then drains the storage queue. The consumer calls each thunk in turn
+  // and writes it to completion before the next, so bodies download one at a
+  // time and never ahead of the encoder.
   const thunks = (async function* (): AsyncGenerator<EntryThunk> {
     const files = await manager.listFiles(backup.id, { signal });
     if (files.length === 0) {
@@ -562,9 +467,6 @@ export async function downloadBackupArchive(
     if (storageError) throw storageError;
   })();
 
-  const prefetch = finitePositiveInt(opts.prefetch, DEFAULT_PREFETCH);
-  const entries = prefetchEntries(thunks, prefetch);
-
   const sinkWriter = opts.sink.getWriter();
   try {
     // Sink the archive encoder writes into: it tallies the encoded size for
@@ -587,7 +489,9 @@ export async function downloadBackupArchive(
     });
 
     const writer = await createWriter(countingSink, signal);
-    for await (const entry of entries) {
+
+    for await (const thunk of thunks) {
+      const entry = await thunk();
       entry.onWriting();
       await writer.add(entry.name, entry.input, {
         lastModDate: backup.backupAt,

client/packages/version/src/version.ts

@@ -2,6 +2,6 @@
 // Update the version here and merge your code to main to
 // publish a new version of all of the packages to npm.
 
-const version = 'v1.0.65';
+const version = 'v1.0.66';
 
 export { version };