diff --git a/client/packages/platform/__tests__/src/backupDownload.test.ts b/client/packages/platform/__tests__/src/backupDownload.test.ts index 4378f8a218..a1b93d3a49 100644 --- a/client/packages/platform/__tests__/src/backupDownload.test.ts +++ b/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((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((r) => { + resolve = r; + }); + return { promise, resolve }; + }; + const began = files.map(deferred); + const release = files.map(deferred); + let addCount = 0; const createWriter = async (sink: WritableStream) => { const w = sink.getWriter(); return { add: async (_name: string, input: ReadableStream) => { - if (firstAdd) { - firstAdd = false; - await allStarted; - } + const i = addCount++; + began[i].resolve(); + await release[i].promise; for await (const _chunk of input) { // drain } @@ -215,7 +220,7 @@ describe('downloadBackupArchive', () => { }; }; - await downloadBackupArchive({ + const done = downloadBackupArchive({ manager, backup, fetchBody: trackingFetch, @@ -223,6 +228,17 @@ describe('downloadBackupArchive', () => { 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', diff --git a/client/packages/platform/src/backupDownload.ts b/client/packages/platform/src/backupDownload.ts index 0c29d29b6b..5e57d304c9 100644 --- a/client/packages/platform/src/backupDownload.ts +++ b/client/packages/platform/src/backupDownload.ts @@ -76,21 +76,9 @@ 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. - * - * 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>, @@ -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; -/** - * 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, - 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 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((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((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/. 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 { 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, diff --git a/client/packages/version/src/version.ts b/client/packages/version/src/version.ts index f8e438ee50..0a016fc4d7 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.65'; +const version = 'v1.0.66'; export { version };