Prefetch files when downloading archive

client/packages/cli/src/lib/backupDownload.ts

@@ -13,6 +13,10 @@ import type {
   BackupDownloadResult,
   BackupsManager,
 } from '@instantdb/platform';
+import version from '../version.js';
+
+// Set a user-agent or else cloudfront might block the request
+const userAgent = `instant-cli/${version}`;
 
 export type {
   BackupDownloadProgress,
@@ -32,7 +36,11 @@ function fetchStream(
 ): Promise<IncomingMessage> {
   return new Promise((resolve, reject) => {
     const get = url.startsWith('https:') ? httpsGet : httpGet;
-    const req = get(url, { signal }, resolve);
+    const req = get(
+      url,
+      { signal, headers: { 'user-agent': userAgent } },
+      resolve,
+    );
     req.on('error', reject);
   });
 }
@@ -44,6 +52,41 @@ function pipe(src: Readable, dst: Duplex): Duplex {
   return src.pipe(dst);
 }
 
+// S3/CloudFront answer a rejected presigned URL with a short XML body naming
+// the actual cause (SignatureDoesNotMatch, AccessDenied, expired, …). Read a
+// bounded prefix so the error surfaces the reason instead of a bare status.
+async function readErrorBody(
+  res: IncomingMessage,
+  signal: AbortSignal,
+): Promise<string> {
+  // A stalled or dribbling error body must not hang the download; destroying
+  // res ends the read below. The request's own signal already tears res down
+  // on a caller abort, so this only guards the no-abort stall.
+  const timer = setTimeout(() => res.destroy(), 5000);
+  try {
+    const chunks: Buffer[] = [];
+    let total = 0;
+    for await (const chunk of res) {
+      chunks.push(chunk as Buffer);
+      total += (chunk as Buffer).length;
+      if (total >= 2048) break;
+    }
+    return Buffer.concat(chunks)
+      .toString('utf8')
+      .replace(/\s+/g, ' ')
+      .trim()
+      .slice(0, 500);
+  } catch (e) {
+    // A caller-initiated abort must propagate, not be masked as an empty body;
+    // only other read failures fall back to no reason.
+    if (signal.aborted) throw e;
+    return '';
+  } finally {
+    clearTimeout(timer);
+    res.destroy();
+  }
+}
+
 // Fetches a presigned URL with node:http(s), decompressing explicitly: the
 // entity shards are served with `Content-Encoding: zstd` and Node doesn't
 // auto-decompress that. downloadBackupToFile refuses to run without zstd
@@ -54,8 +97,8 @@ async function fetchBody(
 ): Promise<ReadableStream<Uint8Array>> {
   const res = await fetchStream(url, signal);
   if (res.statusCode !== 200) {
-    res.resume();
-    throw new Error(`HTTP ${res.statusCode}`);
+    const body = await readErrorBody(res, signal);
+    throw new Error(`HTTP ${res.statusCode}${body ? ` — ${body}` : ''}`);
   }
   const encoding = res.headers['content-encoding'];
   let stream: Readable = res;

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

@@ -168,6 +168,223 @@ describe('downloadBackupArchive', () => {
     await expect(promise).rejects.toMatchObject({ name: 'AbortError' });
   });
 
+  test('fetches later entries while an earlier one is still being written', 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;
+    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;
+    const createWriter = async (sink: WritableStream<Uint8Array>) => {
+      const w = sink.getWriter();
+      return {
+        add: async (_name: string, input: ReadableStream<Uint8Array>) => {
+          if (firstAdd) {
+            firstAdd = false;
+            await allStarted;
+          }
+          for await (const _chunk of input) {
+            // drain
+          }
+          await w.write(new Uint8Array([0]));
+        },
+        close: () => w.close(),
+      };
+    };
+
+    await downloadBackupArchive({
+      manager,
+      backup,
+      fetchBody: trackingFetch,
+      sink: nullSink(),
+      createWriter,
+    });
+
+    expect(started).toEqual([
+      'config.json',
+      'entities/a.jsonl',
+      'entities/b.jsonl',
+    ]);
+  });
+
+  test('retries a transient failure opening an entry body', async () => {
+    const names: string[] = [];
+    const manager = {
+      listFiles,
+      getFileUrl,
+      streamStorageFiles: async function* () {
+        yield { locationId: 'loc-1', path: 'a.png', url: 'loc-1-url' };
+      },
+    } as any;
+
+    // The storage blob's first open fails once, then succeeds.
+    let storageAttempts = 0;
+    const flakyFetch = async (url: string) => {
+      if (url === 'loc-1-url') {
+        storageAttempts++;
+        if (storageAttempts === 1) throw new Error('ECONNRESET');
+      }
+      return bodyOf(`body:${url}`);
+    };
+
+    const result = await downloadBackupArchive({
+      manager,
+      backup,
+      fetchBody: flakyFetch,
+      sink: nullSink(),
+      createWriter: makeWriter(names),
+      retry: { attempts: 3, delayMs: 0 },
+    });
+
+    expect(storageAttempts).toBe(2);
+    expect(names).toEqual([
+      'config.json',
+      'entities/todos.jsonl',
+      'files/loc-1',
+    ]);
+    expect(result.files).toBe(1);
+  });
+
+  test('retries a body that connects but errors on its first read', async () => {
+    const names: string[] = [];
+    const manager = {
+      listFiles,
+      getFileUrl,
+      streamStorageFiles: async function* () {
+        yield { locationId: 'loc-1', path: 'a.png', url: 'loc-1-url' };
+      },
+    } as any;
+
+    // fetchBody resolves (connection established) but the body errors on its
+    // first read the first time — the shape of a reset idle connection.
+    let storageAttempts = 0;
+    const flakyFetch = async (url: string) => {
+      if (url === 'loc-1-url') {
+        storageAttempts++;
+        if (storageAttempts === 1) {
+          return new ReadableStream<Uint8Array>({
+            pull(controller) {
+              controller.error(new Error('ECONNRESET'));
+            },
+          });
+        }
+      }
+      return bodyOf(`body:${url}`);
+    };
+
+    const result = await downloadBackupArchive({
+      manager,
+      backup,
+      fetchBody: flakyFetch,
+      sink: nullSink(),
+      createWriter: makeWriter(names),
+      retry: { attempts: 3, delayMs: 0 },
+    });
+
+    expect(storageAttempts).toBe(2);
+    expect(names).toEqual([
+      'config.json',
+      'entities/todos.jsonl',
+      'files/loc-1',
+    ]);
+    expect(result.files).toBe(1);
+  });
+
+  test('does not retry once the body has yielded a chunk', async () => {
+    const manager = {
+      listFiles,
+      getFileUrl,
+      streamStorageFiles: async function* () {
+        yield { locationId: 'loc-1', path: 'a.png', url: 'loc-1-url' };
+      },
+    } as any;
+
+    // The body delivers one chunk on the first read, then errors on the next —
+    // a mid-stream failure. Once bytes are flowing the entry can't be
+    // restarted, so this must not retry.
+    let storageAttempts = 0;
+    const fetchBody = async (url: string) => {
+      if (url === 'loc-1-url') {
+        storageAttempts++;
+        let phase = 0;
+        return new ReadableStream<Uint8Array>({
+          pull(controller) {
+            if (phase++ === 0) {
+              controller.enqueue(new TextEncoder().encode('partial'));
+            } else {
+              controller.error(new Error('mid-stream reset'));
+            }
+          },
+        });
+      }
+      return bodyOf(`body:${url}`);
+    };
+
+    await expect(
+      downloadBackupArchive({
+        manager,
+        backup,
+        fetchBody,
+        sink: nullSink(),
+        createWriter: makeWriter([]),
+        retry: { attempts: 3, delayMs: 0 },
+      }),
+    ).rejects.toThrow();
+    expect(storageAttempts).toBe(1);
+  });
+
+  test('gives up after exhausting retries and names the failing entry', async () => {
+    const manager = {
+      listFiles,
+      getFileUrl,
+      streamStorageFiles: async function* () {
+        yield { locationId: 'loc-9', path: null, url: 'bad-url' };
+      },
+    } as any;
+
+    let attempts = 0;
+    await expect(
+      downloadBackupArchive({
+        manager,
+        backup,
+        fetchBody: async (url: string) => {
+          if (url === 'bad-url') {
+            attempts++;
+            throw new Error('HTTP 500');
+          }
+          return bodyOf('x');
+        },
+        sink: nullSink(),
+        createWriter: makeWriter([]),
+        retry: { attempts: 3, delayMs: 0 },
+      }),
+    ).rejects.toThrow('Couldn\'t download storage file "loc-9" (HTTP 500).');
+    expect(attempts).toBe(3);
+  });
+
   test('names a pathless storage file by locationId when its download fails', async () => {
     const manager = {
       listFiles,
@@ -187,6 +404,7 @@ describe('downloadBackupArchive', () => {
         },
         sink: nullSink(),
         createWriter: makeWriter([]),
+        retry: { attempts: 1 },
       }),
     ).rejects.toThrow('Couldn\'t download storage file "loc-9" (HTTP 500).');
   });