From d52ab6116a65aedfdf4602b22e586457049395c5 Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Mon, 10 Aug 2026 12:49:51 -0700 Subject: [PATCH 1/6] prefetch files when download archive --- .../__tests__/src/backupDownload.test.ts | 130 ++++++++ .../packages/platform/src/backupDownload.ts | 304 +++++++++++++++--- 2 files changed, 385 insertions(+), 49 deletions(-) diff --git a/client/packages/platform/__tests__/src/backupDownload.test.ts b/client/packages/platform/__tests__/src/backupDownload.test.ts index e9fb6fa99e..2100439fdc 100644 --- a/client/packages/platform/__tests__/src/backupDownload.test.ts +++ b/client/packages/platform/__tests__/src/backupDownload.test.ts @@ -168,6 +168,135 @@ 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((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) => { + const w = sink.getWriter(); + return { + add: async (_name: string, input: ReadableStream) => { + 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('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 +316,7 @@ describe('downloadBackupArchive', () => { }, sink: nullSink(), createWriter: makeWriter([]), + retry: { attempts: 1 }, }), ).rejects.toThrow('Couldn\'t download storage file "loc-9" (HTTP 500).'); }); diff --git a/client/packages/platform/src/backupDownload.ts b/client/packages/platform/src/backupDownload.ts index 68ac929c31..9278f3bbc1 100644 --- a/client/packages/platform/src/backupDownload.ts +++ b/client/packages/platform/src/backupDownload.ts @@ -76,14 +76,200 @@ export type DownloadBackupArchiveOpts = { ) => Promise; signal?: AbortSignal; onProgress?: (progress: BackupDownloadProgress) => void; + /** + * How many entries to fetch ahead of the one currently being written into + * the archive. Each backup entry needs a presigned URL and a fresh HTTP + * connection before its bytes flow; writing entries strictly one at a time + * pays that whole round-trip latency between every file. Fetching a few + * ahead overlaps the next files' latency with the current file's transfer, + * which is the dominant cost for backups with many small storage blobs. + * + * 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 — retrying the open recovers from that and other transient + * failures. `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}). + * + * This only covers failures *before* the writer starts consuming the body: + * once bytes have been written into the archive entry there's no way to + * restart it without HTTP range/resume support, so a mid-stream failure + * still fails the download. + */ + 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; + const isAbortError = (e: unknown): boolean => (e as { name?: string })?.name === 'AbortError'; const errorMessage = (e: unknown): string => e instanceof Error ? e.message : String(e); +// A cancellable sleep: resolves after `ms`, or rejects if the signal aborts +// first so backoff between retries doesn't outlive a cancelled download. +const delay = (ms: number, signal: AbortSignal): Promise => + new Promise((resolve, reject) => { + const onAbort = () => { + clearTimeout(timer); + reject(signal.reason ?? new DOMException('Aborted', 'AbortError')); + }; + const timer = setTimeout(() => { + signal.removeEventListener('abort', onAbort); + resolve(); + }, ms); + signal.addEventListener('abort', onAbort, { once: true }); + }); + +/** + * Opens a body via `open`, retrying on transient failure with abortable + * exponential backoff. `open` is re-invoked from scratch each attempt, so an + * entity file re-mints its presigned URL. An abort propagates immediately; any + * other final failure is wrapped by `describe` into a user-facing message + * naming the entry. + */ +async function openWithRetry( + open: () => Promise>, + opts: { + signal: AbortSignal; + attempts: number; + delayMs: number; + describe: (e: unknown) => string; + }, +): Promise> { + let lastError: unknown; + for (let attempt = 1; attempt <= opts.attempts; attempt++) { + opts.signal.throwIfAborted(); + try { + return await open(); + } catch (e) { + if (isAbortError(e)) throw e; + lastError = e; + if (attempt < opts.attempts) { + const backoff = Math.min( + opts.delayMs * 2 ** (attempt - 1), + MAX_RETRY_DELAY_MS, + ); + await delay(backoff, opts.signal); + } + } + } + throw new Error(opts.describe(lastError)); +} + +/** + * 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. + */ +type PreparedEntry = { + name: string; + input: ReadableStream; + onWriting: () => void; + onAdded: () => void; +}; + +/** + * A thunk that starts fetching one entry (presigned URL + body) and resolves + * once the body stream is available — not once it's fully downloaded. + */ +type EntrySpec = () => Promise; + +/** + * Wraps an ordered stream of entry specs, keeping up to `lookahead` fetches + * in flight while yielding the prepared entries in their original order. A + * background producer pulls specs 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( + specs: AsyncIterable, + lookahead: number, +): AsyncGenerator { + const pipeline: Promise[] = []; + 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 spec (or start its fetch) + // lands in state.producerError for the consumer to throw in order. + const producer = (async () => { + try { + for await (const spec of specs) { + while (pipeline.length >= lookahead) { + await new Promise((resolve) => { + state.onSpace = resolve; + }); + } + const started = spec(); + // 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((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 @@ -132,6 +318,9 @@ export async function downloadBackupArchive( ? backup.uncompressedSize + (backup.filesSize ?? 0) : null; + const retryAttempts = Math.max(1, opts.retry?.attempts ?? DEFAULT_FETCH_ATTEMPTS); + const retryDelayMs = opts.retry?.delayMs ?? DEFAULT_RETRY_DELAY_MS; + const tick = () => onProgress?.({ entitiesCompleted, @@ -221,16 +410,11 @@ export async function downloadBackupArchive( // Entry write order is significant for restore: config.json first, then the // entities/*.jsonl shards, then files/. In particular ALL entity // files must be written before ANY storage file. listFiles returns the - // entity files in write order; this generator yields them to completion, - // then drains the storage queue. - type ArchiveEntry = { - name: string; - input: ReadableStream; - // Fired after the writer finishes consuming the entry, so completion - // counters reflect fully-written files rather than started fetches. - onAdded: () => void; - }; - const entries = (async function* (): AsyncGenerator { + // entity files in write order; this generator yields specs for them in + // order, then drains the storage queue. The specs 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. + const specs = (async function* (): AsyncGenerator { const files = await manager.listFiles(backup.id, { signal }); if (files.length === 0) { throw new Error('No files found for this backup.'); @@ -248,27 +432,34 @@ export async function downloadBackupArchive( tick(); for (const f of files) { - currentEntity = f.name; - tick(); - const url = await manager.getFileUrl(backup.id, f.name, { signal }); - let body: ReadableStream; - try { - body = await fetchBody(url, signal); - } catch (e) { - if (isAbortError(e)) throw e; - throw new Error(`Failed to fetch ${f.name}: ${errorMessage(e)}.`); - } - yield { - name: f.name, - input: countBytes(body), - onAdded: () => { - if (f.name !== 'config.json') entitiesCompleted++; - tick(); - }, + yield async () => { + const body = await openWithRetry( + async () => { + const url = await manager.getFileUrl(backup.id, f.name, { signal }); + return fetchBody(url, signal); + }, + { + signal, + attempts: retryAttempts, + delayMs: retryDelayMs, + describe: (e) => `Failed to fetch ${f.name}: ${errorMessage(e)}.`, + }, + ); + return { + name: f.name, + input: countBytes(body), + onWriting: () => { + currentEntity = f.name; + currentFile = ''; + tick(); + }, + onAdded: () => { + if (f.name !== 'config.json') entitiesCompleted++; + tick(); + }, + }; }; } - currentEntity = ''; - tick(); while (true) { if (storageError) throw storageError; @@ -284,25 +475,32 @@ export async function downloadBackupArchive( queueHead++; } if (file) { - const label = file.path || file.locationId; - currentFile = label; - tick(); - let body: ReadableStream; - try { - body = await fetchBody(file.url, signal); - } catch (e) { - if (isAbortError(e)) throw e; - throw new Error( - `Couldn't download storage file "${label}" (${errorMessage(e)}).`, + const storageFile = file; + const label = storageFile.path || storageFile.locationId; + yield async () => { + const body = await openWithRetry( + () => fetchBody(storageFile.url, signal), + { + signal, + attempts: retryAttempts, + delayMs: retryDelayMs, + describe: (e) => + `Couldn't download storage file "${label}" (${errorMessage(e)}).`, + }, ); - } - yield { - name: `files/${file.locationId}`, - input: countBytes(body), - onAdded: () => { - filesCompleted++; - tick(); - }, + return { + name: `files/${storageFile.locationId}`, + input: countBytes(body), + onWriting: () => { + currentEntity = ''; + currentFile = label; + tick(); + }, + onAdded: () => { + filesCompleted++; + tick(); + }, + }; }; } else if (storageDone) { break; @@ -312,12 +510,16 @@ export async function downloadBackupArchive( }); } } - currentFile = ''; - tick(); if (storageError) throw storageError; })(); + const prefetch = + opts.prefetch != null && opts.prefetch > 0 + ? opts.prefetch + : DEFAULT_PREFETCH; + const entries = prefetchEntries(specs, prefetch); + const sinkWriter = opts.sink.getWriter(); try { // Sink the archive encoder writes into: it tallies the encoded size for @@ -341,11 +543,15 @@ export async function downloadBackupArchive( const writer = await createWriter(countingSink, signal); for await (const entry of entries) { + entry.onWriting(); await writer.add(entry.name, entry.input, { lastModDate: backup.backupAt, }); entry.onAdded(); } + currentEntity = ''; + currentFile = ''; + tick(); // A caller abort that lands after the last entry lets the generator // finish cleanly; don't close and return a complete-looking archive. signal.throwIfAborted(); From e629fc590fc7d9c3d2f0507f21dc8da7c05d8132 Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Mon, 10 Aug 2026 15:23:44 -0700 Subject: [PATCH 2/6] format --- client/packages/platform/src/backupDownload.ts | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/client/packages/platform/src/backupDownload.ts b/client/packages/platform/src/backupDownload.ts index 9278f3bbc1..1419689de1 100644 --- a/client/packages/platform/src/backupDownload.ts +++ b/client/packages/platform/src/backupDownload.ts @@ -318,7 +318,10 @@ export async function downloadBackupArchive( ? backup.uncompressedSize + (backup.filesSize ?? 0) : null; - const retryAttempts = Math.max(1, opts.retry?.attempts ?? DEFAULT_FETCH_ATTEMPTS); + const retryAttempts = Math.max( + 1, + opts.retry?.attempts ?? DEFAULT_FETCH_ATTEMPTS, + ); const retryDelayMs = opts.retry?.delayMs ?? DEFAULT_RETRY_DELAY_MS; const tick = () => From 93d73485049551be1ca75d59201d9e9141a8677e Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Mon, 10 Aug 2026 16:28:21 -0700 Subject: [PATCH 3/6] add a user agent, show cli command in dropdown --- client/packages/cli/src/lib/backupDownload.ts | 41 +++++++++++++++-- client/www/components/dash/Backups.tsx | 45 +++++++++++++++++-- 2 files changed, 79 insertions(+), 7 deletions(-) diff --git a/client/packages/cli/src/lib/backupDownload.ts b/client/packages/cli/src/lib/backupDownload.ts index ffaefa6aca..9f370cd038 100644 --- a/client/packages/cli/src/lib/backupDownload.ts +++ b/client/packages/cli/src/lib/backupDownload.ts @@ -13,6 +13,13 @@ import type { BackupDownloadResult, BackupsManager, } from '@instantdb/platform'; +import version from '../version.js'; + +// node:http(s) send no User-Agent by default, and the presigned URLs sit behind +// CloudFront whose WAF blocks requests with no UA (the NoUserAgent_HEADER rule), +// returning a 403 "Request blocked" page. The browser and the CLI's fetch-based +// API calls carry a UA and pass; this raw request must set one too. +const userAgent = `instant-cli/${version}`; export type { BackupDownloadProgress, @@ -32,7 +39,11 @@ function fetchStream( ): Promise { 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 +55,30 @@ 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): Promise { + 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 { + return ''; + } finally { + 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 +89,8 @@ async function fetchBody( ): Promise> { const res = await fetchStream(url, signal); if (res.statusCode !== 200) { - res.resume(); - throw new Error(`HTTP ${res.statusCode}`); + const body = await readErrorBody(res); + throw new Error(`HTTP ${res.statusCode}${body ? ` — ${body}` : ''}`); } const encoding = res.headers['content-encoding']; let stream: Readable = res; diff --git a/client/www/components/dash/Backups.tsx b/client/www/components/dash/Backups.tsx index d62d705566..3e6792ae82 100644 --- a/client/www/components/dash/Backups.tsx +++ b/client/www/components/dash/Backups.tsx @@ -9,6 +9,7 @@ import { } from 'react'; import { ArrowDownTrayIcon, + CommandLineIcon, EllipsisVerticalIcon, PlusIcon, TrashIcon, @@ -39,6 +40,7 @@ import { DropdownMenu, DropdownMenuContent, DropdownMenuItem, + DropdownMenuSeparator, DropdownMenuTrigger, } from '@/components/DropdownMenu'; import { @@ -245,6 +247,27 @@ function BackupRow({ deleteDialog.onOpen(); } + // The equivalent `instant-cli backup download` invocation, for scripting or + // downloading outside the browser. When the dashboard isn't pointed at the + // CLI's default (production) backend — i.e. dev, staging, or a self-hosted + // instance — prefix the exact apiURI via INSTANT_CLI_API_URI so the CLI hits + // the same server this dashboard does. + const cliDownloadCommand = (() => { + const cmd = `npx instant-cli@latest backup download ${backup.id} --app ${app.id}`; + return config.apiURI === 'https://api.instantdb.com' + ? cmd + : `INSTANT_CLI_API_URI=${config.apiURI} ${cmd}`; + })(); + + async function copyCliDownloadCommand() { + try { + await window.navigator.clipboard.writeText(cliDownloadCommand); + successToast('Copied CLI download command.'); + } catch { + errorToast('Failed to copy to clipboard.'); + } + } + async function deleteBackup() { if (!token) return; setDeleting(true); @@ -304,12 +327,26 @@ function BackupRow({ - Delete backup + Copy CLI download command + {canDelete ? ( + <> + + + {' '} + Delete backup + + + ) : null} ); @@ -345,7 +382,7 @@ function BackupRow({ downloadButton ) } - corner={canDelete ? actionsMenu : null} + corner={actionsMenu} />
From 6f5ca752f2eb69ee178abe70356f4217dae0b142 Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Mon, 10 Aug 2026 16:50:19 -0700 Subject: [PATCH 4/6] cleanup --- client/packages/cli/src/lib/backupDownload.ts | 5 +-- .../packages/platform/src/backupDownload.ts | 41 ++++++++----------- 2 files changed, 19 insertions(+), 27 deletions(-) diff --git a/client/packages/cli/src/lib/backupDownload.ts b/client/packages/cli/src/lib/backupDownload.ts index 9f370cd038..62f3bed159 100644 --- a/client/packages/cli/src/lib/backupDownload.ts +++ b/client/packages/cli/src/lib/backupDownload.ts @@ -15,10 +15,7 @@ import type { } from '@instantdb/platform'; import version from '../version.js'; -// node:http(s) send no User-Agent by default, and the presigned URLs sit behind -// CloudFront whose WAF blocks requests with no UA (the NoUserAgent_HEADER rule), -// returning a 403 "Request blocked" page. The browser and the CLI's fetch-based -// API calls carry a UA and pass; this raw request must set one too. +// Set a user-agent or else cloudfront might block the request const userAgent = `instant-cli/${version}`; export type { diff --git a/client/packages/platform/src/backupDownload.ts b/client/packages/platform/src/backupDownload.ts index 1419689de1..dbf1d2177f 100644 --- a/client/packages/platform/src/backupDownload.ts +++ b/client/packages/platform/src/backupDownload.ts @@ -78,13 +78,9 @@ export type DownloadBackupArchiveOpts = { onProgress?: (progress: BackupDownloadProgress) => void; /** * How many entries to fetch ahead of the one currently being written into - * the archive. Each backup entry needs a presigned URL and a fresh HTTP - * connection before its bytes flow; writing entries strictly one at a time - * pays that whole round-trip latency between every file. Fetching a few - * ahead overlaps the next files' latency with the current file's transfer, - * which is the dominant cost for backups with many small storage blobs. + * the archive. * - * Only fetch *initiation* is parallelised — the bodies are still written in + * 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}. @@ -94,13 +90,12 @@ export type DownloadBackupArchiveOpts = { * 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 — retrying the open recovers from that and other transient - * failures. `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}). + * read it. + * `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}). * - * This only covers failures *before* the writer starts consuming the body: - * once bytes have been written into the archive entry there's no way to + * This only covers failures before the writer starts consuming the body. + * Once bytes have been written into the archive entry there's no way to * restart it without HTTP range/resume support, so a mid-stream failure * still fails the download. */ @@ -186,18 +181,18 @@ 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. */ -type EntrySpec = () => Promise; +type EntryThunk = () => Promise; /** - * Wraps an ordered stream of entry specs, keeping up to `lookahead` fetches + * 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 specs and starts their fetches as space frees up; + * 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( - specs: AsyncIterable, + thunks: AsyncIterable, lookahead: number, ): AsyncGenerator { const pipeline: Promise[] = []; @@ -221,17 +216,17 @@ async function* prefetchEntries( if (w) w(); }; - // Never rejects: a failure to produce the next spec (or start its fetch) + // 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 spec of specs) { + for await (const thunk of thunks) { while (pipeline.length >= lookahead) { await new Promise((resolve) => { state.onSpace = resolve; }); } - const started = spec(); + 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(() => {}); @@ -413,11 +408,11 @@ export async function downloadBackupArchive( // Entry write order is significant for restore: config.json first, then the // entities/*.jsonl shards, then files/. In particular ALL entity // files must be written before ANY storage file. listFiles returns the - // entity files in write order; this generator yields specs for them in - // order, then drains the storage queue. The specs are consumed through + // 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. - const specs = (async function* (): AsyncGenerator { + const thunks = (async function* (): AsyncGenerator { const files = await manager.listFiles(backup.id, { signal }); if (files.length === 0) { throw new Error('No files found for this backup.'); @@ -521,7 +516,7 @@ export async function downloadBackupArchive( opts.prefetch != null && opts.prefetch > 0 ? opts.prefetch : DEFAULT_PREFETCH; - const entries = prefetchEntries(specs, prefetch); + const entries = prefetchEntries(thunks, prefetch); const sinkWriter = opts.sink.getWriter(); try { From 1486a4111fc522d2c0e12b83a52a0cb2f5c2bbb8 Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Mon, 10 Aug 2026 17:11:50 -0700 Subject: [PATCH 5/6] review feedback --- client/packages/cli/src/lib/backupDownload.ts | 17 +++- .../__tests__/src/backupDownload.test.ts | 88 +++++++++++++++++++ .../packages/platform/src/backupDownload.ts | 65 ++++++++++++-- client/www/components/dash/Backups.tsx | 2 +- 4 files changed, 159 insertions(+), 13 deletions(-) diff --git a/client/packages/cli/src/lib/backupDownload.ts b/client/packages/cli/src/lib/backupDownload.ts index 62f3bed159..6041243463 100644 --- a/client/packages/cli/src/lib/backupDownload.ts +++ b/client/packages/cli/src/lib/backupDownload.ts @@ -55,7 +55,14 @@ function pipe(src: Readable, dst: Duplex): Duplex { // 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): Promise { +async function readErrorBody( + res: IncomingMessage, + signal: AbortSignal, +): Promise { + // 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; @@ -69,9 +76,13 @@ async function readErrorBody(res: IncomingMessage): Promise { .replace(/\s+/g, ' ') .trim() .slice(0, 500); - } catch { + } 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(); } } @@ -86,7 +97,7 @@ async function fetchBody( ): Promise> { const res = await fetchStream(url, signal); if (res.statusCode !== 200) { - const body = await readErrorBody(res); + const body = await readErrorBody(res, signal); throw new Error(`HTTP ${res.statusCode}${body ? ` — ${body}` : ''}`); } const encoding = res.headers['content-encoding']; diff --git a/client/packages/platform/__tests__/src/backupDownload.test.ts b/client/packages/platform/__tests__/src/backupDownload.test.ts index 2100439fdc..4378f8a218 100644 --- a/client/packages/platform/__tests__/src/backupDownload.test.ts +++ b/client/packages/platform/__tests__/src/backupDownload.test.ts @@ -268,6 +268,94 @@ describe('downloadBackupArchive', () => { 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({ + 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({ + 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, diff --git a/client/packages/platform/src/backupDownload.ts b/client/packages/platform/src/backupDownload.ts index dbf1d2177f..0c29d29b6b 100644 --- a/client/packages/platform/src/backupDownload.ts +++ b/client/packages/platform/src/backupDownload.ts @@ -107,6 +107,13 @@ const DEFAULT_FETCH_ATTEMPTS = 3; const DEFAULT_RETRY_DELAY_MS = 500; const MAX_RETRY_DELAY_MS = 5000; +// Normalize caller-supplied numeric options so NaN/Infinity/non-integers can't +// alter retry counts or pipeline bounds — fall back to the default instead. +const finitePositiveInt = (v: number | undefined, fallback: number): number => + v != null && Number.isInteger(v) && v > 0 ? v : fallback; +const finiteNonNegative = (v: number | undefined, fallback: number): number => + v != null && Number.isFinite(v) && v >= 0 ? v : fallback; + const isAbortError = (e: unknown): boolean => (e as { name?: string })?.name === 'AbortError'; @@ -128,12 +135,46 @@ const delay = (ms: number, signal: AbortSignal): Promise => signal.addEventListener('abort', onAbort, { once: true }); }); +// Re-emits an already-read first chunk, then streams the rest from `reader`. +// Past that first chunk read errors propagate to the consumer unchanged — by +// then bytes are in the archive entry and it can't be restarted. +function replayFrom( + first: ReadableStreamReadResult, + reader: ReadableStreamDefaultReader, +): ReadableStream { + let replayed = false; + return new ReadableStream({ + async pull(controller) { + if (!replayed) { + replayed = true; + if (first.done) { + controller.close(); + return; + } + controller.enqueue(first.value); + return; + } + const { done, value } = await reader.read(); + if (done) controller.close(); + else controller.enqueue(value); + }, + async cancel(reason) { + await reader.cancel(reason); + }, + }); +} + /** * Opens a body via `open`, retrying on transient failure with abortable * exponential backoff. `open` is re-invoked from scratch each attempt, so an * entity file re-mints its presigned URL. An abort propagates immediately; any * other final failure is wrapped by `describe` into a user-facing message * 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. */ async function openWithRetry( open: () => Promise>, @@ -147,9 +188,15 @@ async function openWithRetry( let lastError: unknown; for (let attempt = 1; attempt <= opts.attempts; attempt++) { opts.signal.throwIfAborted(); + let reader: ReadableStreamDefaultReader | undefined; try { - return await open(); + const body = await open(); + reader = body.getReader(); + const first = await reader.read(); + return replayFrom(first, reader); } catch (e) { + // Release the failed connection before retrying (or giving up). + if (reader) reader.cancel().catch(() => {}); if (isAbortError(e)) throw e; lastError = e; if (attempt < opts.attempts) { @@ -313,11 +360,14 @@ export async function downloadBackupArchive( ? backup.uncompressedSize + (backup.filesSize ?? 0) : null; - const retryAttempts = Math.max( - 1, - opts.retry?.attempts ?? DEFAULT_FETCH_ATTEMPTS, + const retryAttempts = finitePositiveInt( + opts.retry?.attempts, + DEFAULT_FETCH_ATTEMPTS, + ); + const retryDelayMs = finiteNonNegative( + opts.retry?.delayMs, + DEFAULT_RETRY_DELAY_MS, ); - const retryDelayMs = opts.retry?.delayMs ?? DEFAULT_RETRY_DELAY_MS; const tick = () => onProgress?.({ @@ -512,10 +562,7 @@ export async function downloadBackupArchive( if (storageError) throw storageError; })(); - const prefetch = - opts.prefetch != null && opts.prefetch > 0 - ? opts.prefetch - : DEFAULT_PREFETCH; + const prefetch = finitePositiveInt(opts.prefetch, DEFAULT_PREFETCH); const entries = prefetchEntries(thunks, prefetch); const sinkWriter = opts.sink.getWriter(); diff --git a/client/www/components/dash/Backups.tsx b/client/www/components/dash/Backups.tsx index 3e6792ae82..610fbcdafe 100644 --- a/client/www/components/dash/Backups.tsx +++ b/client/www/components/dash/Backups.tsx @@ -256,7 +256,7 @@ function BackupRow({ const cmd = `npx instant-cli@latest backup download ${backup.id} --app ${app.id}`; return config.apiURI === 'https://api.instantdb.com' ? cmd - : `INSTANT_CLI_API_URI=${config.apiURI} ${cmd}`; + : `INSTANT_CLI_API_URI='${config.apiURI}' ${cmd}`; })(); async function copyCliDownloadCommand() { From b384c99a9f8a8a0c429c545690fbf7a1f464606f Mon Sep 17 00:00:00 2001 From: Daniel Woelfel Date: Tue, 11 Aug 2026 10:57:36 -0700 Subject: [PATCH 6/6] Bump version to v1.0.65 --- client/packages/version/src/version.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/client/packages/version/src/version.ts b/client/packages/version/src/version.ts index edaa696c85..f8e438ee50 100644 --- a/client/packages/version/src/version.ts +++ b/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.64'; +const version = 'v1.0.65'; export { version };