From c0953fd2c8572e3b9b6fec1674913a7e7089c44d Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 10:07:32 -0700 Subject: [PATCH 1/8] perf(storage): bound concurrent archive sync and reuse retained ZIPs --- .changeset/parsed-archive-concurrency.md | 8 + docs/streamed-parsed-document-sync.md | 68 ++- scripts/validation/streamed-sync-memory.mjs | 87 ++- .../validation/streamed-sync-process-kill.mjs | 128 +++++ src/index.ts | 1 + src/knowledge/knowledge.ts | 10 +- src/knowledge/types.ts | 2 + .../parsed-document-archive-cache.test.ts | 519 ++++++++++++++++++ .../streamed-sync-concurrency.test.ts | 466 ++++++++++++++++ src/storage/parsed-document-archive-cache.ts | 453 +++++++++++++++ .../sequential-parsed-document-archive.ts | 20 +- src/storage/streamed-parsed-document-sync.ts | 363 +++++++++--- src/types/storage.ts | 22 +- 13 files changed, 2025 insertions(+), 122 deletions(-) create mode 100644 .changeset/parsed-archive-concurrency.md create mode 100644 scripts/validation/streamed-sync-process-kill.mjs create mode 100644 src/storage/__tests__/parsed-document-archive-cache.test.ts create mode 100644 src/storage/__tests__/streamed-sync-concurrency.test.ts create mode 100644 src/storage/parsed-document-archive-cache.ts diff --git a/.changeset/parsed-archive-concurrency.md b/.changeset/parsed-archive-concurrency.md new file mode 100644 index 0000000..456c400 --- /dev/null +++ b/.changeset/parsed-archive-concurrency.md @@ -0,0 +1,8 @@ +--- +'@ontos-ai/knowhere-sdk': minor +--- + +Add bounded concurrent Range archive extraction, count/time checkpoint batching, +stream content lengths and storage entity tags, and optional source ZIP caching +with persistent fallback and immutable transport validation. Preserve sequential +streaming, existing storage adapters, archive order, and checkpoint identities. diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md index 09e58a2..8902d83 100644 --- a/docs/streamed-parsed-document-sync.md +++ b/docs/streamed-parsed-document-sync.md @@ -2,8 +2,9 @@ `knowledge.syncParsedDocument({ documentId })` resolves the published revision and refreshes its signed job-result URL on every invocation. It reads the ZIP central -directory over HTTP Range GETs and copies each result file to storage before -extracting another file. The default remote path does not call `jobs.load()` or +directory over HTTP Range GETs and streams each result file to storage. Range +archives support explicitly bounded concurrent entry writes; the default remains +one active entry. The default remote path does not call `jobs.load()` or construct a `ParseResult`. Signed GET URLs do not require HEAD authorization. The snapshot preserves the exact uncompressed bytes and result-relative paths @@ -16,8 +17,8 @@ replace the result with a reconstruction. ## Storage adapter -Existing `ParsedDocumentStorage` adapters remain supported. They receive one -entry at a time through `writeObject`. An adapter can add: +Existing `ParsedDocumentStorage` adapters remain supported. They receive entry bytes through `writeObject`; the default still copies one +entry at a time. An adapter can add: ```typescript writeObjectStream?(params: ParsedDocumentWriteObjectStreamParams): @@ -25,11 +26,17 @@ writeObjectStream?(params: ParsedDocumentWriteObjectStreamParams): ``` The parameters match `writeObject` except `body` is a Web -`ReadableStream`, with optional `signal: AbortSignal`. Consume the +`ReadableStream`, with optional `signal: AbortSignal` and +`contentLength` for the expected uncompressed entry size. Consume the entire body with backpressure before resolving. Reject an interrupted or partial write; expose the completed object only when the write succeeds. The SDK advances -the checkpoint after that successful write. The byte-only fallback buffers one -entry at a time, so its peak memory depends on the largest individual entry. +the checkpoint only after successful writes form a contiguous prefix in archive +order. `ParsedDocumentObjectHead` and `ParsedDocumentWriteObjectResult` may expose +an `entityTag` for immutable object version checks. The byte-only fallback buffers one +active entry, so its peak memory depends on the largest entries and configured +concurrency. Adapters may accept a second optional `writeSyncProgress` argument +with `{ signal, isSegmentBoundary }`. It marks invocation boundaries and terminal +checkpoints; older adapters that accept only the progress object remain valid. ## Batches and retries @@ -39,6 +46,9 @@ const knowledge = client.knowledge.withParsedStorage({ limits: { maxEntriesPerSync: 100, archiveSyncDeadlineMs: 120_000, + maxConcurrentFiles: 4, // Default 1; sequential archives always use 1. + checkpointFileInterval: 10, // Default 1. + checkpointIntervalMs: 2_000, }, }); @@ -47,14 +57,21 @@ const result = await knowledge.syncParsedDocument({ revisionKey, signal: abortController.signal, retainResultArchive: true, + cacheResultArchive: true, // Optional; default false. }); if (!result.completed) { // Persist or re-enqueue this same document/revision for another invocation. } ``` -Each invocation copies at most 100 new entries by default and checks its 120-second -budget between entries. `completed: false` leaves the last successful entry +Each invocation starts at most 100 new entries by default and checks its +120-second budget before scheduling another entry. All five archive limits must +be positive safe integers; invalid settings fail before API or download requests. +At a checkpoint count or time threshold, scheduling pauses, active writes settle, +and the contiguous successful prefix is persisted before more entries start. +At most `checkpointFileInterval + maxConcurrentFiles - 1` successful entries can +await a checkpoint. Out-of-order successes never advance the cursor past a failed +or unfinished entry. Segment boundaries also drain and checkpoint active writes. `completed: false` leaves the last successful entry checkpoint available for the next invocation. `nextArchiveEntry` is a zero-based index of the next snapshot entry, excluding directories, nested `result.zip`, and reserved SDK metadata. `pinnedRevisionKey`, `archiveIdentity`, and @@ -65,7 +82,10 @@ mismatches fail before writing; publication is checked again before committing. With HTTP Range support, retries skip completed entries after validating archive identity and central-directory metadata. Without Range support, retries read the source sequentially again and verify completed entry hashes before continuing. -Caller cancellation rejects and preserves the last successful checkpoint. +Caller cancellation rejects, cancels active extraction, and waits for active +storage writes before returning. Cancellation and other handled failures attempt +a final checkpoint of the contiguous successful prefix; a failed checkpoint may +require copying those entries again. An explicit sync after a completed checkpoint starts a fresh bounded transfer, so callers can rebuild a snapshot whose persisted objects are missing or corrupt. Re-enqueued running or failed transfers continue from their successful checkpoint. @@ -80,6 +100,28 @@ invalidate the committed snapshot. Use a stream-capable adapter to retain a large archive without buffering it. A byte-only adapter must buffer that optional ZIP object. The default does not retain `result.zip`. +## Optional source archive cache + +`cacheResultArchive: true` streams the source ZIP to `result.zip` before extraction, +then reads entry ranges from that storage object. It is independent of final +snapshot publication: `result.zip` and its SDK cache marker do not constitute a +commit. The option requires a stream sink, `headObject`, `getObjectUrl`, a strong +upstream ETag, and a storage ETag and content length. A marker records the origin +identity, SHA-256, byte count, and immutable storage version after a complete upload. + +Each invocation refreshes the signed origin URL and checks its canonical identity. +A cached ZIP must match its marker, HEAD metadata, and Range transport ETag/total +length; each response also validates its actual byte count. Checkpoint identity +continues to use the origin identity and central directory, so switching between +origin and cached transport preserves existing checkpoints. Corrupt content and +changed origin identities fail rather than committing a snapshot. + +Optional caching has a thirty-second budget. Missing capabilities, interrupted +uploads, cache metadata failures, or unavailable Range transport persist +`archiveCacheBypassed: true` before falling back to the original streaming path. +Later segments retain that decision. `retainResultArchive` continues to control +optional post-commit retention independently. + ## Servers without Range support The SDK first probes support with a ranged GET and cancels an ignored Range body. @@ -123,9 +165,3 @@ whole-archive GET. The two concurrent jobs share one process and stream a combin allocations, and garbage-collection/native-allocation variance; they do not establish a safe concurrency limit for a deployed application. Storage adapters must preserve stream backpressure for the same memory behavior. - -The complete `pnpm test:ci` run passes 380 tests and all configured coverage -thresholds (75.45% branches, 85.95% statements, 88.88% functions, 86.60% lines). -The streaming and lazy ParseResult storage modules each have 100% line coverage. -`pnpm lint`, `pnpm typecheck`, `pnpm format:check`, and `pnpm build` also pass; -lint reports three pre-existing warnings in unrelated tests. diff --git a/scripts/validation/streamed-sync-memory.mjs b/scripts/validation/streamed-sync-memory.mjs index 42de489..d331aa5 100644 --- a/scripts/validation/streamed-sync-memory.mjs +++ b/scripts/validation/streamed-sync-memory.mjs @@ -2,33 +2,42 @@ import assert from 'node:assert/strict'; import { spawn } from 'node:child_process'; import { randomFillSync } from 'node:crypto'; import { createReadStream, createWriteStream } from 'node:fs'; -import { mkdtemp, rm, stat } from 'node:fs/promises'; +import { mkdtemp, stat } from 'node:fs/promises'; import { createServer } from 'node:http'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { Writable } from 'node:stream'; +import { Readable, Writable } from 'node:stream'; +import { pipeline } from 'node:stream/promises'; import { fileURLToPath } from 'node:url'; import { ZipWriter } from '@zip.js/zip.js'; const MEBIBYTE = 1024 * 1024; const isWorker = process.argv[2] === '--worker'; if (isWorker) { - await validateWorker(process.argv[3], Number(process.argv[4]), Number(process.argv[5] ?? 1)); + await validateWorker( + process.argv[3], + Number(process.argv[4]), + Number(process.argv[5] ?? 1), + process.argv[6] === 'cache', + ); } else { const directory = await mkdtemp(join(tmpdir(), 'knowhere-memory-validation-')); try { - for (const [sizeMebibytes, imageCount, concurrency] of [ - [281, 922, 1], - [562, 1844, 1], - [281, 922, 2], + for (const [sizeMebibytes, imageCount, concurrency, cache] of [ + [281, 922, 1, false], + [562, 1844, 1, false], + [281, 922, 2, false], + [281, 922, 1, true], + [562, 1844, 1, true], + [281, 922, 2, true], ]) { const archivePath = join(directory, `${sizeMebibytes}.zip`); if (!(await stat(archivePath).catch(() => null))) await generateArchive(archivePath, sizeMebibytes * MEBIBYTE, imageCount); - await runWorker(archivePath, imageCount, concurrency); + await runWorker(archivePath, imageCount, concurrency, cache); } } finally { - await rm(directory, { recursive: true, force: true }); + console.error(JSON.stringify({ generatedFixturesPreservedAt: directory })); } } @@ -75,7 +84,7 @@ function textStream(value) { }); } -async function runWorker(archivePath, imageCount, concurrency) { +async function runWorker(archivePath, imageCount, concurrency, cache) { await new Promise((resolve, reject) => { const child = spawn( process.execPath, @@ -86,6 +95,7 @@ async function runWorker(archivePath, imageCount, concurrency) { archivePath, String(imageCount), String(concurrency), + cache ? 'cache' : 'range', ], { stdio: ['ignore', 'pipe', 'inherit'] }, ); @@ -97,12 +107,16 @@ async function runWorker(archivePath, imageCount, concurrency) { }); } -async function validateWorker(archivePath, imageCount, concurrency) { +async function validateWorker(archivePath, imageCount, concurrency, cache) { const { Knowledge } = await import('../../dist/index.mjs'); const { size } = await stat(archivePath); let rangeRequests = 0; let wholeRequests = 0; + const cachedArchives = new Map(); + let cacheRangeRequests = 0; const server = createServer((request, response) => { + const cachedArchive = cachedArchives.get(new URL(request.url, 'http://fixture').pathname); + const selectedArchive = cachedArchive ?? archivePath; if (request.method !== 'GET') { response.writeHead(403).end(); return; @@ -111,7 +125,8 @@ async function validateWorker(archivePath, imageCount, concurrency) { let start = 0; let end = size - 1; if (range) { - rangeRequests += 1; + if (cachedArchive) cacheRangeRequests += 1; + else rangeRequests += 1; const match = /^bytes=(\d*)-(\d*)$/.exec(range); if (!match) { response.writeHead(416).end(); @@ -126,9 +141,9 @@ async function validateWorker(archivePath, imageCount, concurrency) { 'accept-ranges': 'bytes', 'content-length': end - start + 1, ...(range ? { 'content-range': `bytes ${start}-${end}/${size}` } : {}), - etag: '"synthetic-immutable-archive"', + etag: cachedArchive ? '"storage-archive-version"' : '"synthetic-immutable-archive"', }); - const source = createReadStream(archivePath, { start, end, highWaterMark: 64 * 1024 }); + const source = createReadStream(selectedArchive, { start, end, highWaterMark: 64 * 1024 }); response.on('close', () => source.destroy()); source.pipe(response); }); @@ -149,7 +164,22 @@ async function validateWorker(archivePath, imageCount, concurrency) { const metadata = new Map(); let progress = null; let calls = 0; + const cachePath = `${archivePath}.copy-${copyIndex}.cached.zip`; + const cacheUrl = `http://127.0.0.1:${address.port}/cache-${copyIndex}.zip`; const storage = { + async headObject(params) { + if (params.path !== 'result.zip' || !cachedArchives.has(`/cache-${copyIndex}.zip`)) + return null; + return { + ...params, + url: cacheUrl, + contentLength: size, + entityTag: '"storage-archive-version"', + }; + }, + async getObjectUrl() { + return cacheUrl; + }, readObject(params) { const body = metadata.get(params.path); return Promise.resolve(body ? { ...params, body } : null); @@ -160,6 +190,13 @@ async function validateWorker(archivePath, imageCount, concurrency) { return Promise.resolve(params); }, async writeObjectStream(params) { + if (params.path === 'result.zip') { + await pipeline(Readable.fromWeb(params.body), createWriteStream(cachePath), { + signal: params.signal, + }); + cachedArchives.set(`/cache-${copyIndex}.zip`, cachePath); + return { ...params, url: cacheUrl, entityTag: '"storage-archive-version"' }; + } const parts = []; const isMetadata = params.path === 'manifest.json' || params.path === 'chunks.json'; for await (const part of params.body) { @@ -204,12 +241,19 @@ async function validateWorker(archivePath, imageCount, concurrency) { }, }, }; - const knowledge = new Knowledge(client).withParsedStorage({ storage }); + const knowledge = new Knowledge(client).withParsedStorage({ + storage, + limits: { maxConcurrentFiles: 4, checkpointFileInterval: 10, checkpointIntervalMs: 2000 }, + }); let result; do { calls += 1; syncCalls += 1; - result = await knowledge.syncParsedDocument({ documentId: 'doc_memory' }); + result = await knowledge.syncParsedDocument({ + documentId: 'doc_memory', + cacheResultArchive: cache, + retainResultArchive: cache, + }); assert(calls <= Math.ceil((imageCount + 2) / 100) + 1, 'Sync must make bounded progress'); } while (!result.completed); return result; @@ -220,16 +264,25 @@ async function validateWorker(archivePath, imageCount, concurrency) { ); sample(); assert.equal(copiedImages, imageCount * concurrency); - assert.equal(wholeRequests, 0, 'Primary path must only issue Range requests'); + assert.equal( + wholeRequests, + cache ? concurrency : 0, + 'Each cached archive must require exactly one full source download', + ); + if (cache) assert(cacheRangeRequests > 0, 'Expanded entries must read from retained archive'); assert(peakRssBytes < 512 * MEBIBYTE, `Peak RSS exceeds 512 MiB: ${peakRssBytes}`); console.log( JSON.stringify({ archiveMebibytes: Number((size / MEBIBYTE).toFixed(2)), imageCount, concurrency, + fileConcurrency: 4, + cacheEnabled: cache, + scope: 'SDK streams and disk cache; HTTP fixture shares measured process', copiedMebibytes: Number((copiedBytes / MEBIBYTE).toFixed(2)), syncCalls, rangeRequests, + cacheRangeRequests, wholeRequests, baselineRssMebibytes: Number((baselineRssBytes / MEBIBYTE).toFixed(2)), peakRssMebibytes: Number((peakRssBytes / MEBIBYTE).toFixed(2)), diff --git a/scripts/validation/streamed-sync-process-kill.mjs b/scripts/validation/streamed-sync-process-kill.mjs new file mode 100644 index 0000000..75d5a61 --- /dev/null +++ b/scripts/validation/streamed-sync-process-kill.mjs @@ -0,0 +1,128 @@ +import assert from 'node:assert/strict'; +import { fork } from 'node:child_process'; +import { mkdtemp, readFile, writeFile } from 'node:fs/promises'; +import { createServer } from 'node:http'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import JSZip from 'jszip'; + +if (process.argv[2] === '--child') await runChild(); +else await runParent(); + +async function runChild() { + const { Knowledge, DiskParsedDocumentStorage } = await import('../../dist/index.mjs'); + const [directory, url, mode] = process.argv.slice(3); + const storage = new DiskParsedDocumentStorage(directory); + const writes = []; + const originalWrite = storage.writeObject.bind(storage); + storage.writeObject = async (parameters) => { + const written = await originalWrite(parameters); + if (!parameters.path.startsWith('.knowhere-sdk/')) writes.push(parameters.path); + return written; + }; + const writeProgress = storage.writeSyncProgress.bind(storage); + storage.writeSyncProgress = async (progress, options) => { + if (mode === 'kill' && progress.status === 'running') { + process.send({ phase: 'checkpoint-held', entry: progress.nextArchiveEntry, writes }); + await new Promise(() => {}); + } + await writeProgress(progress, options); + }; + const client = { + jobs: { + get: async () => ({ jobId: 'job_kill', isDone: true, resultUrl: url }), + load() { + throw new Error('Full ParseResult load forbidden'); + }, + }, + documents: { + listChunks: async () => ({ + documentId: 'doc_kill', + jobId: 'job_kill', + jobResultId: 'revision_kill', + chunks: [], + pagination: { page: 1, pageSize: 1, total: 0, totalPages: 0 }, + }), + }, + }; + const knowledge = new Knowledge(client).withParsedStorage({ + storage, + limits: { maxConcurrentFiles: 4, checkpointFileInterval: 10, checkpointIntervalMs: 2000 }, + }); + const result = await knowledge.syncParsedDocument({ documentId: 'doc_kill' }); + assert(result.completed); + process.send({ phase: 'resumed', completed: result.completed, writes }); + process.disconnect(); +} + +async function runParent() { + const directory = await mkdtemp(join(tmpdir(), 'knowhere-kill-validation-')); + const zip = new JSZip(); + zip.file('manifest.json', '{"job_id":"job_kill"}'); + zip.file('chunks.json', '[]'); + for (let index = 0; index < 30; index += 1) + zip.file(`images/${index}.png`, Buffer.alloc(64 * 1024, index)); + const archive = await zip.generateAsync({ type: 'nodebuffer', compression: 'DEFLATE' }); + const server = createServer((request, response) => { + const match = /^bytes=(\d*)-(\d*)$/.exec(request.headers.range ?? ''); + const start = match?.[1] ? Number(match[1]) : 0; + const end = match?.[2] ? Math.min(Number(match[2]), archive.length - 1) : archive.length - 1; + response.writeHead(match ? 206 : 200, { + 'content-length': end - start + 1, + etag: '"kill-fixture"', + ...(match ? { 'content-range': `bytes ${start}-${end}/${archive.length}` } : {}), + }); + response.end(archive.subarray(start, end + 1)); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const url = `http://127.0.0.1:${server.address().port}/archive.zip`; + try { + const killed = await launch('kill'); + assert( + killed.entry > 0 && killed.entry <= 13, + `Uncheckpointed successes exceed13: ${killed.entry}`, + ); + assert.equal(killed.writes.length, killed.entry); + const resumed = await launch('resume'); + assert(resumed.completed); + const retransmitted = resumed.writes.filter((path) => killed.writes.includes(path)); + assert.equal(retransmitted.length, killed.writes.length); + assert(retransmitted.length <= 13); + const evidence = { + fileConcurrency: 4, + checkpointFileInterval: 10, + signal: 'SIGKILL', + unpersistedSuccessfulFiles: killed.entry, + retransmittedFiles: retransmitted.length, + completed: true, + fixtureDirectory: directory, + }; + await writeFile(join(directory, 'evidence.json'), JSON.stringify(evidence, null, 2)); + console.log(JSON.stringify(evidence)); + } finally { + await new Promise((resolve) => server.close(resolve)); + } + function launch(mode) { + return new Promise((resolve, reject) => { + const child = fork(fileURLToPath(import.meta.url), ['--child', directory, url, mode], { + stdio: ['ignore', 'ignore', 'inherit', 'ipc'], + }); + let message; + const deadline = setTimeout(() => { + child.kill('SIGKILL'); + reject(new Error('Process validation timed out')); + }, 30000); + child.on('error', reject); + child.on('message', (value) => { + message = value; + if (mode === 'kill' && value.phase === 'checkpoint-held') child.kill('SIGKILL'); + }); + child.on('exit', (code, signal) => { + clearTimeout(deadline); + if (mode === 'kill' ? signal === 'SIGKILL' : code === 0) resolve(message); + else reject(new Error(`Unexpected child exit: ${code}/${signal}`)); + }); + }); + } +} diff --git a/src/index.ts b/src/index.ts index 7658ae6..02f58a6 100644 --- a/src/index.ts +++ b/src/index.ts @@ -78,6 +78,7 @@ export type { ParsedDocumentStorageLimits, ParsedDocumentRevisionParams, ParsedDocumentSyncProgress, + ParsedDocumentSyncProgressOptions, ParsedDocumentSyncScheduler, ParsedDocumentWriteObjectParams, ParsedDocumentWriteObjectStreamParams, diff --git a/src/knowledge/knowledge.ts b/src/knowledge/knowledge.ts index 1525d2b..7972eaf 100644 --- a/src/knowledge/knowledge.ts +++ b/src/knowledge/knowledge.ts @@ -18,7 +18,10 @@ import { syncParseResultToParsedDocumentStorage, } from '../storage/parsed-document-storage.js'; import { keysToCamel } from '../lib/utils.js'; -import { syncRemoteResultArchive } from '../storage/streamed-parsed-document-sync.js'; +import { + syncRemoteResultArchive, + validateArchiveSyncLimits, +} from '../storage/streamed-parsed-document-sync.js'; import { LocalKnowledgeStore } from './local-store.js'; import type { IndexedKnowledgeChunk, @@ -205,6 +208,7 @@ export class Knowledge { params.signal?.throwIfAborted(); const normalized = normalizeDocumentReference(params); const parsedStorageConfig = this.requireParsedStorageConfig(); + validateArchiveSyncLimits(parsedStorageConfig.limits); if (normalized.documentId) { return this.syncRemoteDocumentToParsedStorage({ @@ -213,6 +217,7 @@ export class Knowledge { parsedStorageConfig, signal: params.signal, retainResultArchive: params.retainResultArchive, + cacheResultArchive: params.cacheResultArchive, }); } @@ -229,6 +234,7 @@ export class Knowledge { limits: parsedStorageConfig.limits, signal: params.signal, retainResultArchive: params.retainResultArchive, + cacheResultArchive: params.cacheResultArchive, }); } @@ -1130,6 +1136,7 @@ export class Knowledge { parsedStorageConfig: ParsedDocumentStorageConfig; signal?: AbortSignal; retainResultArchive?: boolean; + cacheResultArchive?: boolean; }): Promise { params.signal?.throwIfAborted(); const firstPage = await this.client.documents.listChunks(params.documentId, { @@ -1158,6 +1165,7 @@ export class Knowledge { limits: params.parsedStorageConfig.limits, signal: params.signal, retainResultArchive: params.retainResultArchive, + cacheResultArchive: params.cacheResultArchive, verifyRevision: async (): Promise => { const current = await this.client.documents.listChunks(params.documentId, { page: 1, diff --git a/src/knowledge/types.ts b/src/knowledge/types.ts index 148cf36..ad3fb62 100644 --- a/src/knowledge/types.ts +++ b/src/knowledge/types.ts @@ -121,6 +121,8 @@ export interface KnowledgeSyncParsedDocumentParams extends KnowledgeDocumentRefe readonly signal?: AbortSignal; /** Optionally retain the exact source ZIP as result.zip after committing the snapshot. */ readonly retainResultArchive?: boolean; + /** Optionally cache the source ZIP in storage before extracting its files. */ + readonly cacheResultArchive?: boolean; } export interface KnowledgeSyncParsedDocumentResponse { diff --git a/src/storage/__tests__/parsed-document-archive-cache.test.ts b/src/storage/__tests__/parsed-document-archive-cache.test.ts new file mode 100644 index 0000000..b503680 --- /dev/null +++ b/src/storage/__tests__/parsed-document-archive-cache.test.ts @@ -0,0 +1,519 @@ +import { createHash } from 'crypto'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import type { + ParsedDocumentObject, + ParsedDocumentStorage, + ParsedDocumentWriteObjectResult, +} from '../../types/storage.js'; +import { prepareCachedArchive } from '../parsed-document-archive-cache.js'; + +interface CacheFixture { + readonly storage: ParsedDocumentStorage; + readonly objects: Map; + readonly requests: Array<{ readonly url: string; readonly range: string | null }>; + readonly archive: Uint8Array; + readonly getStorageVersion: () => string; +} + +function createFixture(): CacheFixture { + const archive: Uint8Array = new TextEncoder().encode('complete archive bytes'); + const objects: Map = new Map(); + const requests: CacheFixture['requests'] = []; + let storageVersion: string = '"storage-0"'; + let urlSequence: number = 0; + const storage: ParsedDocumentStorage = { + readObject(params): Promise { + const body: Uint8Array | undefined = objects.get(params.path); + return Promise.resolve(body ? { ...params, body } : null); + }, + writeObject(params): Promise { + objects.set(params.path, params.body); + return Promise.resolve(params); + }, + async writeObjectStream(params): Promise { + const parts: Uint8Array[] = []; + for await (const chunk of params.body) parts.push(chunk); + const body: Uint8Array = Buffer.concat(parts); + objects.set(params.path, body); + storageVersion = '"storage-1"'; + return { ...params, entityTag: storageVersion }; + }, + headObject(params) { + const body: Uint8Array | undefined = objects.get(params.path); + return Promise.resolve( + body ? { ...params, contentLength: body.length, entityTag: storageVersion } : null, + ); + }, + getObjectUrl(): Promise { + urlSequence += 1; + return Promise.resolve(`https://cache.example/result.zip?signature=${urlSequence}`); + }, + readSyncProgress(): Promise { + return Promise.resolve(null); + }, + writeSyncProgress(): Promise { + return Promise.resolve(); + }, + }; + vi.stubGlobal( + 'fetch', + vi.fn((input: string | URL | Request, init?: RequestInit): Promise => { + const url: string = + typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + const range: string | null = new Headers(init?.headers).get('range'); + requests.push({ url, range }); + const isCache: boolean = url.startsWith('https://cache.example/'); + const body: Uint8Array = isCache ? (objects.get('result.zip') ?? new Uint8Array()) : archive; + const entityTag: string = isCache ? storageVersion : '"origin-version"'; + if (range) + return Promise.resolve( + new Response(body.slice(0, 1), { + status: 206, + headers: { + etag: entityTag, + 'content-length': '1', + 'content-range': `bytes 0-0/${body.length}`, + }, + }), + ); + return Promise.resolve( + new Response(body, { headers: { etag: entityTag, 'content-length': String(body.length) } }), + ); + }), + ); + return { storage, objects, requests, archive, getStorageVersion: (): string => storageVersion }; +} + +const document = { + documentId: 'document-test', + revisionKey: 'revision-test', + jobId: 'job-test', + resultUrl: 'https://origin.example/result.zip?signature=1', +} as const; + +afterEach((): void => { + vi.unstubAllGlobals(); + vi.useRealTimers(); +}); + +describe('prepareCachedArchive', (): void => { + it('streams the origin once and publishes a version-bound completion marker afterward', async (): Promise => { + const fixture: CacheFixture = createFixture(); + + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect(prepared.isCached).toBe(true); + expect(prepared.resultUrl).toMatch(/^https:\/\/cache\.example\//); + expect(new Uint8Array(fixture.objects.get('result.zip')!)).toEqual(fixture.archive); + expect( + fixture.requests.filter( + (request): boolean => request.url.includes('origin') && !request.range, + ), + ).toHaveLength(1); + const markers: string[] = [...fixture.objects.keys()].filter((path): boolean => + path.startsWith('.knowhere-sdk/archive-cache/'), + ); + expect(markers).toHaveLength(1); + const marker: unknown = JSON.parse(Buffer.from(fixture.objects.get(markers[0])!).toString()); + expect(marker).toMatchObject({ + version: 1, + jobId: document.jobId, + revisionKey: document.revisionKey, + originIdentity: prepared.downloadIdentity, + contentLength: fixture.archive.length, + sha256: createHash('sha256').update(fixture.archive).digest('hex'), + storageEntityTag: fixture.getStorageVersion(), + }); + expect(JSON.stringify(marker)).not.toContain('signature='); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + + it('reuses a persisted marker across invocations with freshly signed transport URLs', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const first = await prepareCachedArchive({ ...document, storage: fixture.storage }); + const second = await prepareCachedArchive({ + ...document, + storage: fixture.storage, + resultUrl: 'https://origin.example/result.zip?signature=2', + }); + + expect(second.isCached).toBe(true); + expect(second.resultUrl).not.toBe(first.resultUrl); + expect(second.downloadIdentity).toBe(first.downloadIdentity); + expect( + fixture.requests.filter( + (request): boolean => request.url.includes('origin') && !request.range, + ), + ).toHaveLength(1); + expect(fixture.requests.some((request): boolean => request.url.endsWith('signature=2'))).toBe( + true, + ); + }); + + it('honors a previous bypass without attempting cache storage again', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const headObject = vi.spyOn(fixture.storage, 'headObject'); + const writeObjectStream = vi.spyOn(fixture.storage, 'writeObjectStream'); + const onBypass = vi.fn((): Promise => Promise.resolve()); + + const prepared = await prepareCachedArchive({ + ...document, + storage: fixture.storage, + archiveCacheBypassed: true, + onBypass, + }); + + expect(prepared.archiveCacheBypassed).toBe(true); + expect(prepared.resultUrl).toBe(document.resultUrl); + expect(headObject).not.toHaveBeenCalled(); + expect(writeObjectStream).not.toHaveBeenCalled(); + expect(onBypass).not.toHaveBeenCalled(); + }); + + it('persists bypass before falling back when streaming storage is unavailable', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const storage: ParsedDocumentStorage = { ...fixture.storage, writeObjectStream: undefined }; + let hasBypassed: boolean = false; + + const prepared = await prepareCachedArchive({ + ...document, + storage, + onBypass: (): Promise => { + hasBypassed = true; + return Promise.resolve(); + }, + }); + + expect(hasBypassed).toBe(true); + expect(prepared.isCached).toBe(false); + expect(prepared.archiveCacheBypassed).toBe(true); + expect(fixture.objects.size).toBe(0); + }); + + it('rejects changed CDN versions before exposing a cached range', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect( + (): Response => + prepared.validateResponse( + new Response(new Uint8Array([1]), { + status: 206, + headers: { + etag: '"stale-cdn-version"', + 'content-range': `bytes 0-0/${fixture.archive.length}`, + 'content-length': '1', + }, + }), + ), + ).toThrow(/version/); + }); + + it('cancels a cached response when its version headers are rejected', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + const cancel = vi.fn(); + const response: Response = new Response(new ReadableStream({ cancel }), { + status: 206, + headers: { etag: '"stale-version"' }, + }); + + expect((): Response => prepared.validateResponse(response)).toThrow(/version/); + expect(cancel).toHaveBeenCalledOnce(); + expect((): Response => prepared.validateResponse(response)).toThrow( + expect.objectContaining({ name: 'ArchiveCacheTransportError' }), + ); + }); + + it('rejects a mismatched cached range total before consuming bytes', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect( + (): Response => + prepared.validateResponse( + new Response(new Uint8Array([1]), { + status: 206, + headers: { + etag: fixture.getStorageVersion(), + 'content-range': 'bytes 0-0/99999', + 'content-length': '1', + }, + }), + ), + ).toThrow(/length/); + }); + + it('rejects truncated cached range streams', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + const response: Response = prepared.validateResponse( + new Response(new Uint8Array([1]), { + status: 206, + headers: { + etag: fixture.getStorageVersion(), + 'content-range': `bytes 0-1/${fixture.archive.length}`, + 'content-length': '2', + }, + }), + ); + + await expect(response.arrayBuffer()).rejects.toThrow(/length/); + }); + + it('bypasses stale cache HEAD versions and preserves canonical upstream identity', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const first = await prepareCachedArchive({ ...document, storage: fixture.storage }); + vi.spyOn(fixture.storage, 'headObject').mockImplementation((params) => + Promise.resolve({ + ...params, + entityTag: '"stale-head"', + contentLength: fixture.archive.length, + }), + ); + const onBypass = vi.fn((): Promise => Promise.resolve()); + + const prepared = await prepareCachedArchive({ + ...document, + storage: fixture.storage, + onBypass, + }); + + expect(prepared.isCached).toBe(false); + expect(prepared.downloadIdentity).toBe(first.downloadIdentity); + expect(onBypass).toHaveBeenCalledOnce(); + }); + + it('does not publish a marker after an interrupted archive upload', async (): Promise => { + const fixture: CacheFixture = createFixture(); + vi.spyOn(fixture.storage, 'writeObjectStream').mockImplementation(async (params) => { + const reader: ReadableStreamDefaultReader = params.body.getReader(); + await reader.read(); + await reader.cancel(); + throw new Error('Interrupted storage upload'); + }); + const onBypass = vi.fn((): Promise => Promise.resolve()); + + const prepared = await prepareCachedArchive({ + ...document, + storage: fixture.storage, + onBypass, + }); + + expect(prepared.archiveCacheBypassed).toBe(true); + expect( + [...fixture.objects.keys()].some((path): boolean => path.includes('archive-cache')), + ).toBe(false); + expect(onBypass).toHaveBeenCalledOnce(); + }); + + it('bypasses a cache transport that ignores Range requests', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const url: string = + typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + if (url.includes('cache.example')) + return new Response(fixture.archive, { + headers: { + etag: fixture.getStorageVersion(), + 'content-length': String(fixture.archive.length), + }, + }); + return originalFetch(input, init); + }, + ); + + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect(prepared.isCached).toBe(false); + expect(prepared.archiveCacheBypassed).toBe(true); + }); + + it('rejects a cache probe that sends a larger range than the requested byte', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const url: string = + typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + if (url.includes('cache.example')) + return new Response(fixture.archive, { + status: 206, + headers: { + etag: fixture.getStorageVersion(), + 'content-length': String(fixture.archive.length), + 'content-range': `bytes 0-${fixture.archive.length - 1}/${fixture.archive.length}`, + }, + }); + return originalFetch(input, init); + }, + ); + + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect(prepared.isCached).toBe(false); + expect(prepared.archiveCacheBypassed).toBe(true); + }); + + it('bypasses an origin without a strong validator', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + response.headers.set('etag', 'W/"weak-origin"'); + return response; + }, + ); + + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect(prepared.archiveCacheBypassed).toBe(true); + expect(fixture.objects.size).toBe(0); + }); + + it('does not trust malformed persisted cache metadata', async (): Promise => { + const fixture: CacheFixture = createFixture(); + await prepareCachedArchive({ ...document, storage: fixture.storage }); + const markerPath: string | undefined = [...fixture.objects.keys()].find((path): boolean => + path.startsWith('.knowhere-sdk/archive-cache/'), + ); + fixture.objects.set(markerPath!, Buffer.from('{"version":1,"status":"completed"}')); + + const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage }); + + expect(prepared.archiveCacheBypassed).toBe(true); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + + it('propagates bypass persistence failure before allowing upstream fallback', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const storage: ParsedDocumentStorage = { ...fixture.storage, getObjectUrl: undefined }; + + await expect( + prepareCachedArchive({ + ...document, + storage, + onBypass: (): Promise => Promise.reject(new Error('Checkpoint unavailable')), + }), + ).rejects.toThrow('Checkpoint unavailable'); + }); + + it('bounds unavailable storage work to thirty seconds', async (): Promise => { + vi.useFakeTimers(); + const fixture: CacheFixture = createFixture(); + vi.spyOn(fixture.storage, 'headObject').mockImplementation( + () => new Promise((): void => undefined), + ); + const onBypass = vi.fn((): Promise => Promise.resolve()); + + const preparation = prepareCachedArchive({ ...document, storage: fixture.storage, onBypass }); + await vi.advanceTimersByTimeAsync(30_001); + const prepared = await preparation; + + expect(prepared.archiveCacheBypassed).toBe(true); + expect(onBypass).toHaveBeenCalledOnce(); + }); + + it('includes the origin probe in the thirty-second cache budget', async (): Promise => { + vi.useFakeTimers(); + const fixture: CacheFixture = createFixture(); + vi.stubGlobal('fetch', (): Promise => new Promise((): void => undefined)); + const onBypass = vi.fn((): Promise => Promise.resolve()); + let hasFinished: boolean = false; + const preparation = prepareCachedArchive({ + ...document, + storage: fixture.storage, + onBypass, + }).then((prepared) => { + hasFinished = true; + return prepared; + }); + + await vi.advanceTimersByTimeAsync(30_001); + + expect(hasFinished).toBe(true); + const prepared = await preparation; + expect(prepared.archiveCacheBypassed).toBe(true); + expect(prepared.downloadIdentity).toBeUndefined(); + expect(onBypass).toHaveBeenCalledOnce(); + }); + + it('aborts an in-flight upload and waits for its cleanup before bypassing', async (): Promise => { + vi.useFakeTimers(); + const fixture: CacheFixture = createFixture(); + let releaseCleanup: (() => void) | undefined; + let hasSettled: boolean = false; + let hasSettledBeforeBypass: boolean = false; + vi.spyOn(fixture.storage, 'writeObjectStream').mockImplementation( + (params): Promise => + new Promise((_resolve, reject): void => { + params.signal!.addEventListener('abort', (): void => { + releaseCleanup = (): void => { + hasSettled = true; + reject(new Error('Upload aborted after cleanup')); + }; + }); + }), + ); + const onBypass = vi.fn((): Promise => { + hasSettledBeforeBypass = hasSettled; + return Promise.resolve(); + }); + const preparation = prepareCachedArchive({ ...document, storage: fixture.storage, onBypass }); + + await vi.advanceTimersByTimeAsync(30_001); + + expect(releaseCleanup).toBeDefined(); + const bypassCallsBeforeCleanup: number = onBypass.mock.calls.length; + releaseCleanup!(); + expect((await preparation).archiveCacheBypassed).toBe(true); + expect(bypassCallsBeforeCleanup).toBe(0); + expect(hasSettledBeforeBypass).toBe(true); + expect(onBypass).toHaveBeenCalledOnce(); + }); + + it('fails when origin bytes no longer match the pinned download identity', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + if (!new Headers(init?.headers).has('range')) + return new Response(fixture.archive, { + headers: { etag: '"changed-origin"', 'content-length': String(fixture.archive.length) }, + }); + return originalFetch(input, init); + }, + ); + + await expect(prepareCachedArchive({ ...document, storage: fixture.storage })).rejects.toThrow( + /origin.*identity/i, + ); + expect( + [...fixture.objects.keys()].some((path): boolean => path.includes('archive-cache')), + ).toBe(false); + }); + + it('does not disguise cancellation as an optional cache bypass', async (): Promise => { + const fixture: CacheFixture = createFixture(); + const controller: AbortController = new AbortController(); + controller.abort(new Error('Caller cancelled')); + const onBypass = vi.fn((): Promise => Promise.resolve()); + + await expect( + prepareCachedArchive({ + ...document, + storage: fixture.storage, + signal: controller.signal, + onBypass, + }), + ).rejects.toThrow('Caller cancelled'); + expect(onBypass).not.toHaveBeenCalled(); + }); +}); diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts new file mode 100644 index 0000000..1ff39a7 --- /dev/null +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -0,0 +1,466 @@ +import JSZip from 'jszip'; +import { afterEach, describe, expect, it, vi } from 'vitest'; +import { Knowledge } from '../../knowledge/knowledge.js'; +import type { Knowhere } from '../../client.js'; +import type { + ParsedDocumentObject, + ParsedDocumentObjectHead, + ParsedDocumentStorage, + ParsedDocumentStorageLimits, + ParsedDocumentSyncProgress, + ParsedDocumentWriteObjectStreamParams, +} from '../../types/storage.js'; +import { syncRemoteResultArchive } from '../streamed-parsed-document-sync.js'; + +interface Fixture { + readonly storage: ParsedDocumentStorage; + readonly objects: Map; + readonly progress: ParsedDocumentSyncProgress[]; + readonly started: string[]; + readonly finished: string[]; + readonly lengths: Map; + readonly fetch: ReturnType; + readonly sync: (limits?: ParsedDocumentStorageLimits, signal?: AbortSignal) => Promise; + readonly getPeak: () => number; +} + +async function createFixture( + options: { + readonly sequential?: boolean; + readonly cache?: boolean; + readonly failPath?: string; + readonly onStart?: (path: string) => void; + readonly delays?: Readonly>; + } = {}, +): Promise { + const zip: JSZip = new JSZip(); + zip.file('manifest.json', '{"job_id":"job"}'); + zip.file('chunks.json', '[]'); + for (let index: number = 0; index < 8; index += 1) zip.file(`file-${index}.txt`, `body-${index}`); + const archive: Uint8Array = await zip.generateAsync({ + type: 'uint8array', + compression: 'DEFLATE', + }); + const fetchMock: ReturnType = vi.fn( + (_input: unknown, init?: RequestInit): Promise => { + const range: string | null = new Headers(init?.headers).get('range'); + const match: RegExpExecArray | null = + range && !options.sequential ? /^bytes=(\d*)-(\d*)$/.exec(range) : null; + const start: number = match?.[1] ? Number(match[1]) : 0; + const end: number = match?.[2] + ? Math.min(Number(match[2]), archive.length - 1) + : archive.length - 1; + return Promise.resolve( + new Response(archive.slice(start, end + 1), { + status: match ? 206 : 200, + headers: { + etag: '"archive"', + 'content-length': String(end - start + 1), + ...(match ? { 'content-range': `bytes ${start}-${end}/${archive.length}` } : {}), + }, + }), + ); + }, + ); + vi.stubGlobal('fetch', fetchMock); + const objects: Map = new Map(); + const progress: ParsedDocumentSyncProgress[] = []; + const started: string[] = []; + const finished: string[] = []; + const lengths: Map = new Map(); + let active: number = 0; + let peak: number = 0; + const storage: ParsedDocumentStorage = { + readObject(params): Promise { + const body: Uint8Array | undefined = objects.get(params.path); + return Promise.resolve(body ? { ...params, body } : null); + }, + writeObject(params) { + objects.set(params.path, params.body); + return Promise.resolve(params); + }, + async writeObjectStream(params: ParsedDocumentWriteObjectStreamParams) { + started.push(params.path); + options.onStart?.(params.path); + lengths.set( + params.path, + 'contentLength' in params ? Number(params.contentLength) : undefined, + ); + active += 1; + peak = Math.max(peak, active); + try { + await new Promise((resolve) => + setTimeout(resolve, options.delays?.[params.path] ?? 2), + ); + if (params.path === options.failPath) throw new Error(`write failed: ${params.path}`); + const parts: Uint8Array[] = []; + for await (const part of params.body) { + params.signal?.throwIfAborted(); + parts.push(part); + } + objects.set(params.path, Buffer.concat(parts)); + finished.push(params.path); + return { ...params, entityTag: '"archive"' }; + } finally { + active -= 1; + } + }, + readSyncProgress: () => Promise.resolve(progress.at(-1) ?? null), + writeSyncProgress: (value) => { + progress.push(value); + return Promise.resolve(); + }, + }; + if (options.cache) { + storage.headObject = (params): Promise => { + const body: Uint8Array | undefined = objects.get(params.path); + return Promise.resolve( + body ? { ...params, contentLength: body.length, entityTag: '"archive"' } : null, + ); + }; + storage.getObjectUrl = (): Promise => + Promise.resolve('https://cache.example/result.zip'); + } + return { + storage, + objects, + progress, + started, + finished, + lengths, + fetch: fetchMock, + getPeak: (): number => peak, + sync: (limits, signal): Promise => + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage, + limits, + signal, + }), + }; +} + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllGlobals(); +}); + +describe('bounded parsed archive concurrency', () => { + it('copies Range entries concurrently and provides the expected uncompressed length', async () => { + const fixture: Fixture = await createFixture({ delays: { 'manifest.json': 20 } }); + await fixture.sync({ maxConcurrentFiles: 3, checkpointFileInterval: 4 }); + expect(fixture.getPeak()).toBe(3); + expect(fixture.lengths.get('manifest.json')).toBe(Buffer.byteLength('{"job_id":"job"}')); + expect(fixture.progress.at(-1)).toMatchObject({ status: 'completed', nextArchiveEntry: 10 }); + }); + + it('batches checkpoints and stops scheduling while the checkpoint is blocked', async () => { + const fixture: Fixture = await createFixture(); + let releaseCheckpoint: (() => void) | undefined; + const checkpointGate: Promise = new Promise((resolve) => { + releaseCheckpoint = resolve; + }); + let markCheckpoint: (() => void) | undefined; + const checkpointStarted: Promise = new Promise((resolve) => { + markCheckpoint = resolve; + }); + fixture.storage.writeSyncProgress = async (value): Promise => { + fixture.progress.push(value); + if (fixture.progress.length === 1) { + markCheckpoint?.(); + await checkpointGate; + } + }; + const sync: Promise = fixture.sync({ + maxConcurrentFiles: 3, + checkpointFileInterval: 4, + }); + await checkpointStarted; + const startedAtCheckpoint: number = fixture.started.length; + await new Promise((resolve) => setTimeout(resolve, 15)); + expect(fixture.started.length).toBe(startedAtCheckpoint); + releaseCheckpoint?.(); + await sync; + const running: ParsedDocumentSyncProgress[] = fixture.progress.filter( + (value) => value.status === 'running', + ); + expect(running.length).toBeLessThan(10); + expect(running[0].nextArchiveEntry).toBeGreaterThanOrEqual(4); + expect(running[0].nextArchiveEntry).toBeLessThanOrEqual(6); + }); + + it('flushes only the contiguous successful prefix and settles all writes after a failure', async () => { + const fixture: Fixture = await createFixture({ + failPath: 'file-0.txt', + delays: { 'manifest.json': 1, 'chunks.json': 1, 'file-0.txt': 30 }, + }); + await expect( + fixture.sync({ maxConcurrentFiles: 3, checkpointFileInterval: 20 }), + ).rejects.toThrow(/write failed/); + expect(fixture.progress.at(-1)?.nextArchiveEntry).toBe(2); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + const writesAtReturn: number = fixture.finished.length; + await new Promise((resolve) => setTimeout(resolve, 40)); + expect(fixture.finished.length).toBe(writesAtReturn); + }); + + it('flushes a successful prefix on caller cancellation and settles active writes', async () => { + const controller: AbortController = new AbortController(); + const fixture: Fixture = await createFixture({ + delays: { 'file-0.txt': 30 }, + onStart: (path): void => { + if (path === 'file-0.txt') + setTimeout((): void => controller.abort(new Error('Caller stopped')), 5); + }, + }); + await expect( + fixture.sync({ maxConcurrentFiles: 3, checkpointFileInterval: 20 }, controller.signal), + ).rejects.toThrow(/Caller stopped/); + expect(fixture.progress.at(-1)?.nextArchiveEntry).toBe(2); + const completedAtReturn: number = fixture.finished.length; + await new Promise((resolve) => setTimeout(resolve, 35)); + expect(fixture.finished).toHaveLength(completedAtReturn); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + + it('counts started entries against the segment limit and marks its checkpoint boundary', async () => { + const fixture: Fixture = await createFixture(); + const options: Array<{ readonly isSegmentBoundary?: boolean } | undefined> = []; + fixture.storage.writeSyncProgress = (value, checkpointOptions): Promise => { + fixture.progress.push(value); + options.push(checkpointOptions); + return Promise.resolve(); + }; + expect( + await fixture.sync({ + maxConcurrentFiles: 3, + checkpointFileInterval: 20, + maxEntriesPerSync: 4, + }), + ).toMatchObject({ completed: false }); + expect(fixture.started).toHaveLength(4); + expect(fixture.progress.at(-1)?.nextArchiveEntry).toBe(4); + expect(options.at(-1)?.isSegmentBoundary).toBe(true); + }); + + it('forces sequential archive sources to use one active entry', async () => { + const fixture: Fixture = await createFixture({ sequential: true }); + await fixture.sync({ maxConcurrentFiles: 4, checkpointFileInterval: 4 }); + expect(fixture.getPeak()).toBe(1); + expect(fixture.lengths.get('manifest.json')).toBe(Buffer.byteLength('{"job_id":"job"}')); + }); + + it('flushes on the elapsed checkpoint interval before starting more entries', async () => { + const fixture: Fixture = await createFixture({ + delays: { 'manifest.json': 10, 'chunks.json': 10 }, + }); + await fixture.sync({ + maxConcurrentFiles: 2, + checkpointFileInterval: 50, + checkpointIntervalMs: 3, + }); + expect(fixture.progress[0].nextArchiveEntry).toBe(2); + expect(fixture.progress.filter((value) => value.status === 'running').length).toBeGreaterThan( + 1, + ); + }); + + it.each([ + 'maxConcurrentFiles', + 'checkpointFileInterval', + 'checkpointIntervalMs', + 'maxEntriesPerSync', + 'archiveSyncDeadlineMs', + ] as const)('rejects invalid %s before any network access', async (name) => { + const fixture: Fixture = await createFixture(); + await expect(fixture.sync({ [name]: 0 })).rejects.toThrow(new RegExp(name)); + expect(fixture.fetch).not.toHaveBeenCalled(); + }); + + it('uses one full origin fetch when caching and retaining the archive together', async () => { + const fixture: Fixture = await createFixture({ cache: true }); + await syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + retainResultArchive: true, + }); + expect(fixture.started.filter((path) => path === 'result.zip')).toHaveLength(1); + const fullFetches: number = fixture.fetch.mock.calls.filter((call: readonly unknown[]) => { + const init: RequestInit | undefined = call[1] as RequestInit | undefined; + return !new Headers(init?.headers).has('range'); + }).length; + expect(fullFetches).toBe(1); + }); + + it('verifies a legacy sequential checkpoint when resuming through cached Range transport', async () => { + const options = { sequential: true, cache: true }; + const fixture: Fixture = await createFixture(options); + await fixture.sync({ maxEntriesPerSync: 2 }); + const checkpoint: string | undefined = fixture.progress.at(-1)?.archiveIdentity; + options.sequential = false; + await syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }); + expect(fixture.progress.at(-1)).toMatchObject({ + status: 'completed', + archiveIdentity: checkpoint, + }); + expect(fixture.started.filter((path) => path === 'manifest.json')).toHaveLength(1); + expect(fixture.started.filter((path) => path === 'chunks.json')).toHaveLength(1); + }); + + it('falls back to the origin if cached transport changes during archive setup', async () => { + const fixture: Fixture = await createFixture({ cache: true }); + const originalFetch: typeof fetch = fetch; + let cacheRequests: number = 0; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + if ( + (typeof input === 'string' + ? input + : input instanceof URL + ? input.href + : input.url + ).startsWith('https://cache.example/') + ) { + cacheRequests += 1; + if (cacheRequests > 2) { + const headers: Headers = new Headers(response.headers); + headers.set('etag', '"changed-cache"'); + return new Response(response.body, { status: response.status, headers }); + } + } + return response; + }, + ); + expect( + await syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }), + ).toMatchObject({ completed: true }); + expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(true); + }); + + it.each(['etag', 'http', 'network'] as const)( + 'persists cache bypass after an entry transport %s failure and retries from the origin', + async (mode) => { + let isInvalid: boolean = false; + const fixture: Fixture = await createFixture({ + cache: true, + onStart: (path): void => { + if (path === 'file-0.txt') isInvalid = true; + }, + }); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + if ( + (typeof input === 'string' + ? input + : input instanceof URL + ? input.href + : input.url + ).startsWith('https://cache.example/') && + isInvalid + ) { + if (mode === 'http') return new Response(null, { status: 503 }); + if (mode === 'network') throw new Error('Cache transport unavailable'); + const headers: Headers = new Headers(response.headers); + headers.set('etag', '"changed-cache"'); + return new Response(response.body, { status: response.status, headers }); + } + return response; + }, + ); + const sync = (): Promise => + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }); + await expect(sync()).rejects.toThrow(); + expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(true); + expect(await sync()).toMatchObject({ completed: true }); + }, + ); + + it('rejects a changed sequential origin after cache bypass without an initial identity', async () => { + const fixture: Fixture = await createFixture({ sequential: true }); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + if (!new Headers(init?.headers).has('range')) { + const headers: Headers = new Headers(response.headers); + headers.set('etag', '"changed-origin"'); + return new Response(response.body, { status: response.status, headers }); + } + return response; + }, + ); + await expect( + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }), + ).rejects.toThrow(/identity changed/); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + + it('persists cache bypass before extracting when the adapter has no cache capabilities', async () => { + const fixture: Fixture = await createFixture(); + await syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }); + expect(fixture.progress[0].archiveCacheBypassed).toBe(true); + expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(true); + }); + + it('validates limits before resolving a remote document reference', async () => { + const listChunks = vi.fn(); + const client: Knowhere = { documents: { listChunks } } as unknown as Knowhere; + const fixture: Fixture = await createFixture(); + const knowledge: Knowledge = new Knowledge(client).withParsedStorage({ + storage: fixture.storage, + limits: { maxConcurrentFiles: 0 }, + }); + await expect(knowledge.syncParsedDocument({ documentId: 'document' })).rejects.toThrow( + /maxConcurrentFiles/, + ); + expect(listChunks).not.toHaveBeenCalled(); + }); +}); diff --git a/src/storage/parsed-document-archive-cache.ts b/src/storage/parsed-document-archive-cache.ts new file mode 100644 index 0000000..4c09d54 --- /dev/null +++ b/src/storage/parsed-document-archive-cache.ts @@ -0,0 +1,453 @@ +import { createHash, type Hash } from 'crypto'; +import { ArchiveCacheTransportError } from './archive-cache-transport-error.js'; +import type { + ParsedDocumentObject, + ParsedDocumentObjectHead, + ParsedDocumentObjectParams, + ParsedDocumentStorage, + ParsedDocumentWriteObjectResult, +} from '../types/storage.js'; + +interface ArchiveCacheParams { + readonly storage: ParsedDocumentStorage; + readonly documentId: string; + readonly revisionKey: string; + readonly jobId: string; + readonly resultUrl: string; + readonly signal?: AbortSignal; + readonly archiveCacheBypassed?: boolean; + readonly onBypass?: () => Promise; +} + +interface PreparedArchive { + readonly resultUrl: string; + readonly downloadIdentity: string | undefined; + readonly archiveCacheBypassed: boolean; + readonly isCached: boolean; + readonly validateResponse: (response: Response) => Response; +} + +interface ArchiveCacheMarker { + readonly version: 1; + readonly jobId: string; + readonly revisionKey: string; + readonly originIdentity: string; + readonly contentLength: number; + readonly sha256: string; + readonly storageEntityTag: string; +} + +interface ArchiveOrigin { + readonly identity: string; + readonly contentLength: number | undefined; + readonly entityTag: string | null; +} + +interface ArchiveCacheObjectVersion { + readonly contentLength: number; + readonly entityTag: string; +} + +const ARCHIVE_CACHE_TIMEOUT_MS: number = 30_000; + +class ArchiveOriginIntegrityError extends Error {} + +/** Cache metadata records a complete ZIP transfer, never a completed document sync. */ +export async function prepareCachedArchive(params: ArchiveCacheParams): Promise { + params.signal?.throwIfAborted(); + if (params.archiveCacheBypassed) return createFallback(params); + if ( + !params.storage.writeObjectStream || + !params.storage.headObject || + !params.storage.getObjectUrl + ) { + await params.onBypass?.(); + return createFallback(params); + } + + const controller: AbortController = new AbortController(); + const signal: AbortSignal = params.signal + ? AbortSignal.any([params.signal, controller.signal]) + : controller.signal; + const mutations: Set> = new Set(); + let rejectCancellation: (() => void) | undefined; + const cancellation: Promise = new Promise((_resolve, reject): void => { + rejectCancellation = (): void => { + const reason: unknown = signal.reason; + reject( + reason instanceof Error + ? reason + : new Error('Parsed result archive cache was cancelled.', { cause: reason }), + ); + }; + signal.addEventListener('abort', rejectCancellation, { once: true }); + }); + const timer: ReturnType = setTimeout((): void => { + controller.abort(new Error('Parsed result archive cache exceeded its thirty-second budget.')); + }, ARCHIVE_CACHE_TIMEOUT_MS); + let origin: ArchiveOrigin | undefined; + const preparation: Promise = (async (): Promise => { + origin = await probeArchiveOrigin({ ...params, signal }); + signal.throwIfAborted(); + if (!hasStrongEntityTag(origin.entityTag) || origin.contentLength === undefined) + throw new Error('Parsed result archive cache requires a strong origin version and length.'); + return prepareArchiveCache(params, origin, signal, mutations); + })(); + try { + return await Promise.race([preparation, cancellation]); + } catch (error) { + controller.abort(error); + // Stream adapters must reject on abort after releasing their network resources. Wait for + // their cleanup before fallback can start a second transfer against the same capacity. + await Promise.allSettled([...mutations]); + params.signal?.throwIfAborted(); + if (error instanceof ArchiveOriginIntegrityError) throw error; + // Persist this decision before the caller starts the old upstream stream path. + await params.onBypass?.(); + return createFallback(params, origin); + } finally { + clearTimeout(timer); + if (rejectCancellation) signal.removeEventListener('abort', rejectCancellation); + } +} + +function createFallback(params: ArchiveCacheParams, origin?: ArchiveOrigin): PreparedArchive { + return { + resultUrl: params.resultUrl, + downloadIdentity: origin?.identity, + archiveCacheBypassed: true, + isCached: false, + validateResponse(response: Response): Response { + if (origin && createOriginIdentity(params.jobId, response) !== origin.identity) + rejectResponse( + response, + new ArchiveOriginIntegrityError('Parsed result origin archive identity changed.'), + ); + return response; + }, + }; +} + +async function probeArchiveOrigin(params: ArchiveCacheParams): Promise { + const response: Response = await fetch(params.resultUrl, { + headers: { Range: 'bytes=0-0' }, + signal: params.signal, + }); + try { + if (!response.ok) throw new Error(`Parsed result download failed (HTTP ${response.status}).`); + return { + identity: createOriginIdentity(params.jobId, response), + contentLength: readOriginLength(response), + entityTag: response.headers.get('etag'), + }; + } finally { + await response.body?.cancel(); + } +} + +async function prepareArchiveCache( + params: ArchiveCacheParams, + origin: ArchiveOrigin, + signal: AbortSignal, + mutations: Set>, +): Promise { + const object: ParsedDocumentObjectParams = { ...params, path: 'result.zip' }; + const initialHead: ParsedDocumentObjectHead | null = await params.storage.headObject!(object); + signal.throwIfAborted(); + let marker: ArchiveCacheMarker | null = null; + if (initialHead) { + const version: ArchiveCacheObjectVersion = readObjectVersion(initialHead); + const stored: ParsedDocumentObject | null = await params.storage.readObject({ + ...object, + path: createMarkerPath(origin.identity, version.entityTag), + }); + signal.throwIfAborted(); + if (stored) marker = readMarker(stored, params, origin, version); + } + if (!marker) marker = await storeArchive(params, origin, signal, mutations); + signal.throwIfAborted(); + const resultUrl: string | null = await params.storage.getObjectUrl!(object); + signal.throwIfAborted(); + if (!resultUrl) throw new Error('Parsed result archive cache has no readable object URL.'); + const validateResponse = (response: Response): Response => + validateCachedResponse(response, marker, params.signal); + const response: Response = await fetch(resultUrl, { + headers: { Range: 'bytes=0-0', 'If-Match': marker.storageEntityTag }, + signal, + }); + try { + if ( + response.headers.get('content-range') !== `bytes 0-0/${marker.contentLength}` || + response.headers.get('content-length') !== '1' + ) + rejectResponse( + response, + new ArchiveCacheTransportError('Parsed result archive cache probe exceeded one byte.'), + ); + const validated: Response = validateResponse(response); + // The probe is exactly one byte. Reading it also checks a truncated CDN response. + await validated.arrayBuffer(); + } catch (error) { + await response.body?.cancel().catch((): void => undefined); + throw error; + } + signal.throwIfAborted(); + return { + resultUrl, + downloadIdentity: origin.identity, + archiveCacheBypassed: false, + isCached: true, + validateResponse, + }; +} + +async function storeArchive( + params: ArchiveCacheParams, + origin: ArchiveOrigin, + signal: AbortSignal, + mutations: Set>, +): Promise { + const response: Response = await fetch(params.resultUrl, { + headers: { 'If-Match': origin.entityTag! }, + signal, + }); + if (signal.aborted) { + await response.body?.cancel(); + signal.throwIfAborted(); + } + if (!response.ok || !response.body) { + await response.body?.cancel(); + if (response.status === 412) + throw new ArchiveOriginIntegrityError('Parsed result origin archive identity changed.'); + throw new Error(`Parsed result archive cache download failed (HTTP ${response.status}).`); + } + if (createOriginIdentity(params.jobId, response) !== origin.identity) { + await response.body.cancel(); + throw new ArchiveOriginIntegrityError('Parsed result origin archive identity changed.'); + } + const hash: Hash = createHash('sha256'); + let contentLength: number = 0; + let hasCompleted: boolean = false; + const body: ReadableStream = response.body.pipeThrough( + new TransformStream({ + transform(chunk, controller): void { + signal.throwIfAborted(); + contentLength += chunk.length; + if (contentLength > origin.contentLength!) + throw new ArchiveOriginIntegrityError('Parsed result origin archive length changed.'); + hash.update(chunk); + controller.enqueue(chunk); + }, + flush(): void { + if (contentLength !== origin.contentLength) + throw new ArchiveOriginIntegrityError('Parsed result origin archive length changed.'); + hasCompleted = true; + }, + }), + ); + const object: ParsedDocumentObjectParams = { ...params, path: 'result.zip' }; + let write: ParsedDocumentWriteObjectResult; + try { + signal.throwIfAborted(); + write = await settleMutation( + params.storage.writeObjectStream!({ + ...object, + body, + contentType: 'application/zip', + contentLength: origin.contentLength, + signal, + }), + mutations, + ); + } catch (error) { + await body.cancel(error).catch((): void => undefined); + throw error; + } + signal.throwIfAborted(); + if (!hasCompleted) + throw new Error('Parsed result archive cache sink did not consume the stream.'); + if (!hasStrongEntityTag(write.entityTag)) + throw new Error('Parsed result archive cache requires a strong object version.'); + const head: ParsedDocumentObjectHead | null = await params.storage.headObject!(object); + signal.throwIfAborted(); + if (!head) throw new Error('Parsed result archive cache was not persisted.'); + const actualVersion: ArchiveCacheObjectVersion = readObjectVersion(head); + if (write.entityTag !== actualVersion.entityTag || actualVersion.contentLength !== contentLength) + throw new Error('Parsed result archive cache object version or length changed during upload.'); + const marker: ArchiveCacheMarker = { + version: 1, + jobId: params.jobId, + revisionKey: params.revisionKey, + originIdentity: origin.identity, + contentLength, + sha256: hash.digest('hex'), + storageEntityTag: actualVersion.entityTag, + }; + signal.throwIfAborted(); + await settleMutation( + params.storage.writeObject({ + ...object, + path: createMarkerPath(origin.identity, actualVersion.entityTag), + body: Buffer.from(JSON.stringify(marker)), + contentType: 'application/json; charset=utf-8', + }), + mutations, + ); + signal.throwIfAborted(); + return marker; +} + +async function settleMutation( + mutation: Promise, + mutations: Set>, +): Promise { + mutations.add(mutation); + try { + return await mutation; + } finally { + mutations.delete(mutation); + } +} + +function readMarker( + object: ParsedDocumentObject, + params: ArchiveCacheParams, + origin: ArchiveOrigin, + version: ArchiveCacheObjectVersion, +): ArchiveCacheMarker { + const value: unknown = JSON.parse(Buffer.from(object.body).toString('utf8')); + if ( + typeof value !== 'object' || + value === null || + !('version' in value) || + value.version !== 1 || + !('jobId' in value) || + value.jobId !== params.jobId || + !('revisionKey' in value) || + value.revisionKey !== params.revisionKey || + !('originIdentity' in value) || + value.originIdentity !== origin.identity || + !('contentLength' in value) || + value.contentLength !== origin.contentLength || + value.contentLength !== version.contentLength || + !('storageEntityTag' in value) || + value.storageEntityTag !== version.entityTag || + !('sha256' in value) || + typeof value.sha256 !== 'string' || + !/^[a-f0-9]{64}$/.test(value.sha256) + ) + throw new Error('Parsed result archive cache completion marker is invalid.'); + return { + version: 1, + jobId: params.jobId, + revisionKey: params.revisionKey, + originIdentity: origin.identity, + contentLength: version.contentLength, + storageEntityTag: version.entityTag, + sha256: value.sha256, + }; +} + +function readObjectVersion(object: ParsedDocumentObjectHead): ArchiveCacheObjectVersion { + if (!hasStrongEntityTag(object.entityTag) || !isPositiveInteger(object.contentLength)) + throw new Error('Parsed result archive cache requires a strong object version and length.'); + return { entityTag: object.entityTag, contentLength: object.contentLength }; +} + +function validateCachedResponse( + response: Response, + marker: ArchiveCacheMarker, + signal: AbortSignal | undefined, +): Response { + if (response.headers.get('etag') !== marker.storageEntityTag) + rejectResponse( + response, + new ArchiveCacheTransportError('Parsed result archive cache response version changed.'), + ); + if (response.status !== 206 || !response.body) + rejectResponse( + response, + new ArchiveCacheTransportError('Parsed result archive cache requires a ranged response.'), + ); + const range: RegExpExecArray | null = /^bytes (\d+)-(\d+)\/(\d+)$/.exec( + response.headers.get('content-range') ?? '', + ); + const contentLength: number = Number(response.headers.get('content-length')); + if ( + !range || + Number(range[3]) !== marker.contentLength || + Number(range[1]) > Number(range[2]) || + Number(range[2]) >= marker.contentLength || + contentLength !== Number(range[2]) - Number(range[1]) + 1 + ) + rejectResponse( + response, + new ArchiveCacheTransportError('Parsed result archive cache response length changed.'), + ); + let actualLength: number = 0; + return new Response( + response.body.pipeThrough( + new TransformStream({ + transform(chunk, controller): void { + signal?.throwIfAborted(); + actualLength += chunk.length; + if (actualLength > contentLength) + throw new ArchiveCacheTransportError( + 'Parsed result archive cache response length changed.', + ); + controller.enqueue(chunk); + }, + flush(): void { + if (actualLength !== contentLength) + throw new ArchiveCacheTransportError( + 'Parsed result archive cache response length changed.', + ); + }, + }), + ), + { status: response.status, statusText: response.statusText, headers: response.headers }, + ); +} + +function rejectResponse(response: Response, error: Error): never { + void response.body?.cancel(error).catch((): void => undefined); + throw error; +} + +function createMarkerPath(originIdentity: string, entityTag: string): string { + const identity: string = createHash('sha256') + .update(JSON.stringify([originIdentity, entityTag])) + .digest('hex'); + return `.knowhere-sdk/archive-cache/${identity}.json`; +} + +function createOriginIdentity(jobId: string, response: Response): string { + // Keep legacy checkpoints stable when switching between upstream and cached transports. + return createHash('sha256') + .update( + JSON.stringify([ + jobId, + response.headers.get('etag'), + response.headers.get('last-modified'), + response.headers.get('content-range')?.split('/').at(-1) ?? + response.headers.get('content-length'), + ]), + ) + .digest('hex'); +} + +function readOriginLength(response: Response): number | undefined { + const value: number = Number( + response.headers.get('content-range')?.split('/').at(-1) ?? + response.headers.get('content-length'), + ); + return isPositiveInteger(value) ? value : undefined; +} + +function hasStrongEntityTag(value: string | null | undefined): value is string { + return typeof value === 'string' && value.length > 0 && !value.startsWith('W/'); +} + +function isPositiveInteger(value: number | undefined): value is number { + return typeof value === 'number' && Number.isSafeInteger(value) && value > 0; +} diff --git a/src/storage/sequential-parsed-document-archive.ts b/src/storage/sequential-parsed-document-archive.ts index bde87f3..00263ab 100644 --- a/src/storage/sequential-parsed-document-archive.ts +++ b/src/storage/sequential-parsed-document-archive.ts @@ -20,30 +20,34 @@ import { export async function openSequentialArchive( params: StreamedSyncParams, identity: string, + validateResponse?: (response: Response) => Response, ): Promise { - const response: Response = await fetch(params.resultUrl, { signal: params.signal }); + let response: Response = await fetch(params.resultUrl, { signal: params.signal }); if (!response.ok || !response.body) throw new Error(`Parsed result download failed (HTTP ${response.status}).`); - if (createDownloadIdentity(params.jobId, response) !== identity) { + if (validateResponse) response = validateResponse(response); + else if (createDownloadIdentity(params.jobId, response) !== identity) { await response.body.cancel(); throw new Error('Parsed result archive identity changed before sequential download.'); } + const body: ReadableStream | null = response.body; + if (!body) throw new Error('Parsed result download returned an empty body.'); let directory: string; try { directory = await mkdtemp(join(tmpdir(), 'knowhere-sync-')); } catch (error) { - await response.body.cancel(); + await body.cancel(); throw new ParsedDocumentArchiveError(error); } let file: FileHandle; try { file = await open(join(directory, 'archive.zip'), 'w+'); } catch (error) { - await response.body.cancel(); + await body.cancel(); await rm(directory, { recursive: true, force: true }); throw new ParsedDocumentArchiveError(error); } - const input: Readable = Readable.fromWeb(response.body); + const input: Readable = Readable.fromWeb(body); const parser = Parse({ forceStream: true }); const checksums: Map = new Map(); let archiveSize: number = 0; @@ -79,6 +83,12 @@ export async function openSequentialArchive( const path: string = normalizeResultPath(entry.path); yield { path, + contentLength: + !(entry.vars.flags & 0x08) && + 'uncompressedSize' in entry.vars && + typeof entry.vars.uncompressedSize === 'number' + ? entry.vars.uncompressedSize + : undefined, read: (): ReadableStream => { let checksum: number = 0; let size: number = 0; diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts index fcbb53b..df07d33 100644 --- a/src/storage/streamed-parsed-document-sync.ts +++ b/src/storage/streamed-parsed-document-sync.ts @@ -1,4 +1,7 @@ import { createHash } from 'crypto'; +import { ValidationError } from '../errors/index.js'; +import { ArchiveCacheTransportError } from './archive-cache-transport-error.js'; +import { prepareCachedArchive } from './parsed-document-archive-cache.js'; import { HttpRangeReader, ZipReader, type FileEntry, type HttpOptions } from '@zip.js/zip.js'; import type { KnowledgeSyncParsedDocumentResponse } from '../knowledge/types.js'; @@ -17,6 +20,7 @@ import { export interface ArchiveEntry { readonly path: string; + readonly contentLength?: number; readonly read: () => ReadableStream; } @@ -24,6 +28,8 @@ export interface ArchiveSource { readonly canSkipEntries: boolean; readonly identity: string; readonly downloadIdentity: string; + readonly archiveCacheBypassed?: boolean; + readonly isCached?: boolean; readonly entries: AsyncIterable | Iterable; readonly close: () => Promise; } @@ -37,97 +43,146 @@ export interface StreamedSyncParams { readonly limits?: ParsedDocumentStorageLimits; readonly signal?: AbortSignal; readonly retainResultArchive?: boolean; + readonly cacheResultArchive?: boolean; readonly verifyRevision?: () => Promise; } -/** Copies one entry at a time; no ParseResult or whole archive is materialized. */ +/** Copies bounded streams concurrently and checkpoints only the contiguous successful prefix. */ export async function syncRemoteResultArchive( params: StreamedSyncParams, ): Promise { params.signal?.throwIfAborted(); + validateArchiveSyncLimits(params.limits); const startedAt: number = Date.now(); - const maxEntries: number = Math.max(1, params.limits?.maxEntriesPerSync ?? 100); - const deadlineMs: number = Math.max(1, params.limits?.archiveSyncDeadlineMs ?? 120_000); + const maxEntries: number = params.limits?.maxEntriesPerSync ?? 100; + const deadlineMs: number = params.limits?.archiveSyncDeadlineMs ?? 120_000; + const checkpointFileInterval: number = params.limits?.checkpointFileInterval ?? 1; + const checkpointIntervalMs: number = params.limits?.checkpointIntervalMs ?? 2_000; + const controller: AbortController = new AbortController(); + const signal: AbortSignal = params.signal + ? AbortSignal.any([params.signal, controller.signal]) + : controller.signal; + const transferParams: StreamedSyncParams = { ...params, signal }; const previous: ParsedDocumentSyncProgress | null = await params.storage.readSyncProgress(params); - const source: ArchiveSource = await openArchive(params); + const source: ArchiveSource = await openArchive(transferParams, previous); + const concurrency: number = source.canSkipEntries ? (params.limits?.maxConcurrentFiles ?? 1) : 1; + const nextEntry: number = + previous?.status === 'completed' ? 0 : (previous?.nextArchiveEntry ?? 0); const entryHashes: string[] = []; + const successfulHashes: Map = new Map(); + const running: Set> = new Set(); const paths: Set = new Set(); let entryIndex: number = 0; - let copiedEntries: number = 0; - const nextEntry: number = - previous?.status === 'completed' ? 0 : (previous?.nextArchiveEntry ?? 0); + let startedEntries: number = 0; + let successfulEntries: number = 0; + let checkpointEntry: number = nextEntry; + let checkpointAt: number = Date.now(); + let failure: Error | undefined; + let archiveCacheBypassed: boolean | undefined = source.archiveCacheBypassed; + + async function flushProgress( + isSegmentBoundary: boolean, + isRecovering: boolean = false, + ): Promise { + let contiguousEntry: number = checkpointEntry; + while (successfulHashes.has(contiguousEntry)) { + entryHashes[contiguousEntry] = successfulHashes.get(contiguousEntry)!; + contiguousEntry += 1; + } + if (contiguousEntry === checkpointEntry) return; + await params.storage.writeSyncProgress( + { + documentId: params.documentId, + revisionKey: params.revisionKey, + pinnedRevisionKey: params.revisionKey, + archiveIdentity: source.identity, + archiveCacheBypassed, + archiveEntryHashes: entryHashes.slice(0, contiguousEntry), + nextArchiveEntry: contiguousEntry, + status: 'running', + updatedAt: new Date().toISOString(), + }, + { signal: isRecovering ? undefined : signal, isSegmentBoundary }, + ); + for (let index: number = checkpointEntry; index < contiguousEntry; index += 1) + successfulHashes.delete(index); + checkpointEntry = contiguousEntry; + successfulEntries = successfulHashes.size; + checkpointAt = Date.now(); + } + + function startEntry(entry: ArchiveEntry, index: number): void { + const task: Promise = copyEntry(transferParams, entry) + .then( + (hash: string): void => { + successfulHashes.set(index, hash); + successfulEntries += 1; + }, + (error: unknown): void => { + if (failure === undefined) + failure = error instanceof Error ? error : new Error(String(error)); + controller.abort(error); + }, + ) + .finally((): void => { + running.delete(task); + }); + running.add(task); + } + + async function drainEntries(): Promise { + await Promise.all(running); + if (failure !== undefined) throw failure; + signal.throwIfAborted(); + } + try { - if (previous?.archiveIdentity && previous.archiveIdentity !== source.identity) { + if (previous?.archiveIdentity && previous.archiveIdentity !== source.identity) throw new Error('Parsed result archive identity changed; start a new revision.'); - } - if (previous?.pinnedRevisionKey && previous.pinnedRevisionKey !== params.revisionKey) { + if (previous?.pinnedRevisionKey && previous.pinnedRevisionKey !== params.revisionKey) throw new Error('Parsed result revision changed while resuming archive sync.'); - } for await (const entry of source.entries) { - params.signal?.throwIfAborted(); + if (failure !== undefined) throw failure; + signal.throwIfAborted(); if (paths.has(entry.path)) throw new Error(`Duplicate parsed result path: ${entry.path}`); paths.add(entry.path); - if ( - entryIndex >= nextEntry && - copiedEntries > 0 && - (copiedEntries >= maxEntries || Date.now() - startedAt >= deadlineMs) - ) { - return { documentId: params.documentId, revisionKey: params.revisionKey, completed: false }; - } - if (entryIndex < nextEntry && source.canSkipEntries) { + if (entryIndex < nextEntry) { const previousHash: string | undefined = previous?.archiveEntryHashes?.[entryIndex]; if (!previousHash) throw new Error('Parsed archive checkpoint is missing an entry hash.'); + if ( + !source.canSkipEntries && + (await copyEntry(transferParams, entry, true)) !== previousHash + ) + throw new Error('Parsed result archive entry changed while resuming.'); entryHashes.push(previousHash); entryIndex += 1; continue; } - const hash = createHash('sha256'); - const body: ReadableStream = entry.read().pipeThrough( - new TransformStream({ - transform(chunk, controller): void { - params.signal?.throwIfAborted(); - hash.update(chunk); - controller.enqueue(chunk); - }, - }), - ); - if (entryIndex < nextEntry) { - await body.pipeTo(new WritableStream(), { signal: params.signal }); - } else { - await writeStream(params.storage, { - documentId: params.documentId, - revisionKey: params.revisionKey, - path: entry.path, - body, - contentType: inferContentTypeFromPath(entry.path), - signal: params.signal, - }); + if ( + startedEntries > 0 && + (startedEntries >= maxEntries || Date.now() - startedAt >= deadlineMs) + ) { + await drainEntries(); + await flushProgress(true); + return { documentId: params.documentId, revisionKey: params.revisionKey, completed: false }; } - const entryHash: string = createHash('sha256') - .update(entry.path) - .update(hash.digest()) - .digest('hex'); - if (entryIndex < nextEntry && previous?.archiveEntryHashes?.[entryIndex] !== entryHash) { - throw new Error('Parsed result archive entry changed while resuming.'); + if ( + successfulEntries >= checkpointFileInterval || + Date.now() - checkpointAt >= checkpointIntervalMs + ) { + await drainEntries(); + await flushProgress(false); } - entryHashes.push(entryHash); + startEntry(entry, entryIndex); entryIndex += 1; - if (entryIndex <= nextEntry) continue; - copiedEntries += 1; - await params.storage.writeSyncProgress({ - documentId: params.documentId, - revisionKey: params.revisionKey, - pinnedRevisionKey: params.revisionKey, - archiveIdentity: source.identity, - archiveEntryHashes: [...entryHashes], - nextArchiveEntry: entryIndex, - status: 'running', - updatedAt: new Date().toISOString(), - }); + startedEntries += 1; + if (running.size >= concurrency) await Promise.race(running); } + await drainEntries(); + await flushProgress(true); await validateRequiredObjects(params, paths); await params.verifyRevision?.(); - params.signal?.throwIfAborted(); + signal.throwIfAborted(); const commit: ParsedDocumentCommit = { version: 1, documentId: params.documentId, @@ -142,19 +197,23 @@ export async function syncRemoteResultArchive( body: Buffer.from(JSON.stringify(commit, null, 2)), contentType: 'application/json; charset=utf-8', }); - await params.storage.writeSyncProgress({ - documentId: params.documentId, - revisionKey: params.revisionKey, - pinnedRevisionKey: params.revisionKey, - archiveIdentity: source.identity, - archiveEntryHashes: entryHashes, - nextArchiveEntry: entryIndex, - status: 'completed', - updatedAt: new Date().toISOString(), - }); - if (params.retainResultArchive) { + await params.storage.writeSyncProgress( + { + documentId: params.documentId, + revisionKey: params.revisionKey, + pinnedRevisionKey: params.revisionKey, + archiveIdentity: source.identity, + archiveCacheBypassed, + archiveEntryHashes: entryHashes, + nextArchiveEntry: entryIndex, + status: 'completed', + updatedAt: new Date().toISOString(), + }, + { signal, isSegmentBoundary: true }, + ); + if (params.retainResultArchive && !source.isCached) { try { - await retainArchive(params, source.downloadIdentity); + await retainArchive(transferParams, source.downloadIdentity); } catch { /* Optional archive retention never invalidates a committed snapshot. */ } @@ -165,14 +224,133 @@ export async function syncRemoteResultArchive( completed: true, commit, }; + } catch (error) { + controller.abort(error); + await Promise.all(running); + const hasCacheTransportFailure: boolean = + error instanceof ArchiveCacheTransportError || failure instanceof ArchiveCacheTransportError; + if (hasCacheTransportFailure) archiveCacheBypassed = true; + await flushProgress(true, true).catch((): void => undefined); + if (hasCacheTransportFailure) { + await persistArchiveCacheBypass( + { ...params, signal: undefined }, + { + documentId: params.documentId, + revisionKey: params.revisionKey, + pinnedRevisionKey: params.revisionKey, + archiveIdentity: source.identity, + archiveEntryHashes: entryHashes.slice(0, checkpointEntry), + nextArchiveEntry: checkpointEntry, + status: 'running', + updatedAt: new Date().toISOString(), + }, + ).catch((): void => undefined); + } + throw failure ?? error; } finally { + controller.abort(); + await Promise.all(running); await source.close(); } } -async function openArchive(params: StreamedSyncParams): Promise { +/** Validate configuration before job lookup, download, or storage writes. */ +export function validateArchiveSyncLimits(limits: ParsedDocumentStorageLimits | undefined): void { + const names: readonly (keyof ParsedDocumentStorageLimits)[] = [ + 'maxEntriesPerSync', + 'archiveSyncDeadlineMs', + 'maxConcurrentFiles', + 'checkpointFileInterval', + 'checkpointIntervalMs', + ]; + for (const name of names) { + const value: number | undefined = limits?.[name]; + if (value !== undefined && (!Number.isSafeInteger(value) || value <= 0)) + throw new ValidationError(`${name} must be a positive safe integer.`); + } +} + +async function copyEntry( + params: StreamedSyncParams, + entry: ArchiveEntry, + isVerification: boolean = false, +): Promise { + const hash = createHash('sha256'); + const body: ReadableStream = entry.read().pipeThrough( + new TransformStream({ + transform(chunk, controller): void { + params.signal?.throwIfAborted(); + hash.update(chunk); + controller.enqueue(chunk); + }, + }), + ); + if (isVerification) + await body.pipeTo(new WritableStream(), { signal: params.signal }); + else + await writeStream(params.storage, { + documentId: params.documentId, + revisionKey: params.revisionKey, + path: entry.path, + body, + contentType: inferContentTypeFromPath(entry.path), + contentLength: entry.contentLength, + signal: params.signal, + }); + return createHash('sha256').update(entry.path).update(hash.digest()).digest('hex'); +} + +async function openArchive( + params: StreamedSyncParams, + previous: ParsedDocumentSyncProgress | null, +): Promise { + const prepared: Awaited> | undefined = + params.cacheResultArchive + ? await prepareCachedArchive({ + ...params, + archiveCacheBypassed: previous?.archiveCacheBypassed, + onBypass: async (): Promise => { + await persistArchiveCacheBypass(params, previous); + }, + }) + : undefined; + try { + return await openArchiveTransport(params, previous, prepared); + } catch (error) { + if (!prepared?.isCached) throw error; + params.signal?.throwIfAborted(); + const bypass: ParsedDocumentSyncProgress = await persistArchiveCacheBypass(params, previous); + return openArchive({ ...params, cacheResultArchive: false }, bypass); + } +} + +async function persistArchiveCacheBypass( + params: StreamedSyncParams, + previous: ParsedDocumentSyncProgress | null, +): Promise { + const progress: ParsedDocumentSyncProgress = { + ...(previous?.status === 'completed' ? {} : previous), + documentId: params.documentId, + revisionKey: params.revisionKey, + pinnedRevisionKey: params.revisionKey, + archiveCacheBypassed: true, + status: 'running', + updatedAt: new Date().toISOString(), + }; + await params.storage.writeSyncProgress(progress, { signal: params.signal }); + return progress; +} + +async function openArchiveTransport( + params: StreamedSyncParams, + previous: ParsedDocumentSyncProgress | null, + prepared: Awaited> | undefined, +): Promise { + const resultUrl: string = prepared?.resultUrl ?? params.resultUrl; + const archiveCacheBypassed: boolean | undefined = + prepared?.archiveCacheBypassed ?? previous?.archiveCacheBypassed; // Probe with GET, then cancel ignored Range responses before a reader can buffer them. - const probe: Response = await fetch(params.resultUrl, { + let probe: Response = await fetch(resultUrl, { headers: { Range: 'bytes=0-0' }, signal: params.signal, }); @@ -180,27 +358,45 @@ async function openArchive(params: StreamedSyncParams): Promise { await probe.body?.cancel(); throw new Error(`Parsed result download failed (HTTP ${probe.status}).`); } - const identity: string = createDownloadIdentity(params.jobId, probe); + if (prepared) probe = prepared.validateResponse(probe); + const identity: string = + prepared?.downloadIdentity ?? createDownloadIdentity(params.jobId, probe); const supportsRange: boolean = probe.status === 206 && probe.headers.has('content-range'); await probe.body?.cancel(); if (!supportsRange) { const { openSequentialArchive } = await import('./sequential-parsed-document-archive.js'); - return openSequentialArchive(params, identity); + const source: ArchiveSource = await openSequentialArchive( + { ...params, resultUrl }, + identity, + prepared?.isCached ? prepared.validateResponse : undefined, + ); + return { ...source, archiveCacheBypassed }; } const options: HttpOptions = { forceRangeRequests: true, preventHeadRequest: true, checkResourceChanges: true, fetch: async (input, init): Promise => { - const response: Response = await fetch(input, { ...init, signal: params.signal }); + let response: Response; + try { + response = await fetch(input, { ...init, signal: params.signal }); + } catch (error) { + params.signal?.throwIfAborted(); + if (prepared?.isCached) + throw new ArchiveCacheTransportError( + 'Parsed result archive cache transport unavailable.', + ); + throw error; + } + if (prepared?.isCached) return prepared.validateResponse(response); if (response.status !== 206) { await response.body?.cancel(); throw new Error(`Parsed result Range request failed (HTTP ${response.status}).`); } - return response; + return prepared ? prepared.validateResponse(response) : response; }, }; - const reader = new ZipReader(new HttpRangeReader(params.resultUrl, options), { + const reader = new ZipReader(new HttpRangeReader(resultUrl, options), { useWebWorkers: false, checkCrc32: true, }); @@ -218,13 +414,16 @@ async function openArchive(params: StreamedSyncParams): Promise { ) .digest('hex'); return { - identity: archiveIdentity, + identity: previous?.archiveIdentity === identity ? identity : archiveIdentity, downloadIdentity: identity, - canSkipEntries: true, + archiveCacheBypassed, + isCached: prepared?.isCached, + canSkipEntries: previous?.archiveIdentity !== identity, entries: (function* (): Generator { for (const entry of files) yield { path: normalizeResultPath(entry.filename), + contentLength: entry.uncompressedSize, read: (): ReadableStream => readRangeEntry(entry, params.signal), }; })(), diff --git a/src/types/storage.ts b/src/types/storage.ts index 80051ca..fe06326 100644 --- a/src/types/storage.ts +++ b/src/types/storage.ts @@ -51,6 +51,7 @@ export interface ParsedDocumentObjectParams extends ParsedDocumentRevisionParams } export interface ParsedDocumentObjectHead extends ParsedDocumentObjectParams { + readonly entityTag?: string; readonly contentType?: string; readonly contentLength?: number; readonly metadata?: Readonly>; @@ -73,10 +74,13 @@ export interface ParsedDocumentWriteObjectStreamParams extends Omit< 'body' > { readonly body: ReadableStream; + /** Expected uncompressed entry length, when supplied by the archive directory. */ + readonly contentLength?: number; readonly signal?: AbortSignal; } export interface ParsedDocumentWriteObjectResult extends ParsedDocumentObjectParams { + readonly entityTag?: string; readonly url?: string; } @@ -96,6 +100,8 @@ export interface ParsedDocumentSyncProgress { readonly nextChunkPage?: number; readonly pinnedRevisionKey?: string; readonly archiveIdentity?: string; + /** Once true, later segments use the origin rather than retrying optional archive caching. */ + readonly archiveCacheBypassed?: boolean; readonly nextArchiveEntry?: number; /** Content hashes for verifying a sequential retry without a strong HTTP validator. */ readonly archiveEntryHashes?: readonly string[]; @@ -104,6 +110,11 @@ export interface ParsedDocumentSyncProgress { readonly error?: string; } +export interface ParsedDocumentSyncProgressOptions { + readonly signal?: AbortSignal; + readonly isSegmentBoundary?: boolean; +} + export interface ParsedDocumentStorage { readObject(params: ParsedDocumentObjectParams): Promise; writeObject(params: ParsedDocumentWriteObjectParams): Promise; @@ -116,7 +127,10 @@ export interface ParsedDocumentStorage { readSyncProgress( params: ParsedDocumentRevisionParams, ): Promise; - writeSyncProgress(params: ParsedDocumentSyncProgress): Promise; + writeSyncProgress( + params: ParsedDocumentSyncProgress, + options?: ParsedDocumentSyncProgressOptions, + ): Promise; } export interface ParsedDocumentSyncScheduler { @@ -126,6 +140,12 @@ export interface ParsedDocumentSyncScheduler { export interface ParsedDocumentStorageLimits { /** ZIP entries copied per invocation (default 100). */ readonly maxEntriesPerSync?: number; + /** Maximum concurrent entry writes for Range archives (default 1). */ + readonly maxConcurrentFiles?: number; + /** Successful files between checkpoints (default 1). */ + readonly checkpointFileInterval?: number; + /** Maximum scheduling interval between checkpoints (default 2000ms). */ + readonly checkpointIntervalMs?: number; /** ZIP copy budget checked between entries (default 120000ms). */ readonly archiveSyncDeadlineMs?: number; readonly remotePageSize?: number; From a8cdc758328d88ca7ed84bd5070e3bbe1ad19947 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 10:08:54 -0700 Subject: [PATCH 2/8] fix(storage): include cache transport error module --- src/storage/archive-cache-transport-error.ts | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 src/storage/archive-cache-transport-error.ts diff --git a/src/storage/archive-cache-transport-error.ts b/src/storage/archive-cache-transport-error.ts new file mode 100644 index 0000000..14b5bfb --- /dev/null +++ b/src/storage/archive-cache-transport-error.ts @@ -0,0 +1,6 @@ +export class ArchiveCacheTransportError extends Error { + constructor(message: string) { + super(message); + this.name = 'ArchiveCacheTransportError'; + } +} From 5bc40d84a5cacb5f68621c2935b9a6957bf7ab7d Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 10:13:50 -0700 Subject: [PATCH 3/8] test: synchronize archive cancellation with completed prefix --- .../streamed-sync-concurrency.test.ts | 64 +++++++++++++++---- 1 file changed, 52 insertions(+), 12 deletions(-) diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts index 1ff39a7..d9fb30a 100644 --- a/src/storage/__tests__/streamed-sync-concurrency.test.ts +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -9,6 +9,7 @@ import type { ParsedDocumentStorageLimits, ParsedDocumentSyncProgress, ParsedDocumentWriteObjectStreamParams, + ParsedDocumentWriteObjectResult, } from '../../types/storage.js'; import { syncRemoteResultArchive } from '../streamed-parsed-document-sync.js'; @@ -209,20 +210,59 @@ describe('bounded parsed archive concurrency', () => { it('flushes a successful prefix on caller cancellation and settles active writes', async () => { const controller: AbortController = new AbortController(); - const fixture: Fixture = await createFixture({ - delays: { 'file-0.txt': 30 }, - onStart: (path): void => { - if (path === 'file-0.txt') - setTimeout((): void => controller.abort(new Error('Caller stopped')), 5); - }, + const fixture: Fixture = await createFixture({ delays: { 'chunks.json': 30 } }); + const writeStream: ParsedDocumentStorage['writeObjectStream'] = + fixture.storage.writeObjectStream?.bind(fixture.storage); + if (!writeStream) throw new Error('Cancellation fixture requires stream storage.'); + const successfulPaths: Set = new Set(); + let confirmPrefix: (() => void) | undefined; + const prefixCompleted: Promise = new Promise((resolve) => { + confirmPrefix = resolve; }); - await expect( - fixture.sync({ maxConcurrentFiles: 3, checkpointFileInterval: 20 }, controller.signal), - ).rejects.toThrow(/Caller stopped/); + let confirmPendingWrite: (() => void) | undefined; + const pendingWriteStarted: Promise = new Promise((resolve) => { + confirmPendingWrite = resolve; + }); + const cancelled: Promise = new Promise((resolve) => { + controller.signal.addEventListener('abort', (): void => resolve(), { once: true }); + }); + let pendingWrites: number = 0; + let settledWrites: number = 0; + fixture.storage.writeObjectStream = async ( + params: ParsedDocumentWriteObjectStreamParams, + ): Promise => { + if (params.path === 'manifest.json' || params.path === 'chunks.json') { + const result: ParsedDocumentWriteObjectResult = await writeStream(params); + successfulPaths.add(params.path); + if (successfulPaths.size === 2) confirmPrefix?.(); + return result; + } + pendingWrites += 1; + confirmPendingWrite?.(); + try { + await cancelled; + params.signal?.throwIfAborted(); + throw new Error('Cancellation signal was not propagated to the active write.'); + } finally { + pendingWrites -= 1; + settledWrites += 1; + } + }; + const sync: Promise = fixture.sync( + { maxConcurrentFiles: 3, checkpointFileInterval: 20 }, + controller.signal, + ); + const rejectedSync: Promise = expect(sync).rejects.toThrow(/Caller stopped/); + await Promise.all([prefixCompleted, pendingWriteStarted]); + expect(successfulPaths).toEqual(new Set(['manifest.json', 'chunks.json'])); + expect(fixture.finished).toHaveLength(2); + expect(pendingWrites).toBeGreaterThan(0); + controller.abort(new Error('Caller stopped')); + await rejectedSync; expect(fixture.progress.at(-1)?.nextArchiveEntry).toBe(2); - const completedAtReturn: number = fixture.finished.length; - await new Promise((resolve) => setTimeout(resolve, 35)); - expect(fixture.finished).toHaveLength(completedAtReturn); + expect(fixture.finished).toHaveLength(2); + expect(pendingWrites).toBe(0); + expect(settledWrites).toBeGreaterThan(0); expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); }); From 74dfd81da607fe14e732b91aed4fa4797d338ba0 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 10:18:38 -0700 Subject: [PATCH 4/8] fix: await archive producers before finishing sync segments --- .../streamed-sync-concurrency.test.ts | 60 +++++++++++++++++++ src/storage/streamed-parsed-document-sync.ts | 27 ++++++--- 2 files changed, 79 insertions(+), 8 deletions(-) diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts index d9fb30a..ae4edd0 100644 --- a/src/storage/__tests__/streamed-sync-concurrency.test.ts +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -1,4 +1,5 @@ import JSZip from 'jszip'; +import { ZipReader, type FileEntry } from '@zip.js/zip.js'; import { afterEach, describe, expect, it, vi } from 'vitest'; import { Knowledge } from '../../knowledge/knowledge.js'; import type { Knowhere } from '../../client.js'; @@ -266,6 +267,65 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); }); + it('waits for Range producer cleanup before returning an incomplete segment', async () => { + const fixture: Fixture = await createFixture(); + let startCleanup: (() => void) | undefined; + const cleanupStarted: Promise = new Promise((resolve) => { + startCleanup = resolve; + }); + let releaseCleanup: (() => void) | undefined; + const cleanupGate: Promise = new Promise((resolve) => { + releaseCleanup = resolve; + }); + let finishCheckpoint: (() => void) | undefined; + const checkpointWritten: Promise = new Promise((resolve) => { + finishCheckpoint = resolve; + }); + let hasCleanedProducer: boolean = false; + const entry: FileEntry = { + filename: 'manifest.json', + directory: false, + uncompressedSize: 16, + signature: 0, + async getData(writable: WritableStream): Promise { + const writer: WritableStreamDefaultWriter = writable.getWriter(); + try { + await writer.write(Buffer.from('{"job_id":"job"}')); + await writer.close(); + } finally { + writer.releaseLock(); + startCleanup?.(); + await cleanupGate; + hasCleanedProducer = true; + } + }, + } as FileEntry; + vi.spyOn(ZipReader.prototype, 'getEntries').mockResolvedValue([ + entry, + { ...entry, filename: 'chunks.json' }, + ]); + fixture.storage.writeSyncProgress = (value): Promise => { + fixture.progress.push(value); + finishCheckpoint?.(); + return Promise.resolve(); + }; + const sync: Promise = fixture.sync({ maxEntriesPerSync: 1 }); + try { + await Promise.all([cleanupStarted, checkpointWritten]); + const state: string = await Promise.race([ + sync.then((): string => 'returned'), + new Promise((resolve) => setImmediate((): void => resolve('waiting'))), + ]); + expect(state).toBe('waiting'); + expect(hasCleanedProducer).toBe(false); + } finally { + releaseCleanup?.(); + await sync; + } + expect(hasCleanedProducer).toBe(true); + expect(await sync).toMatchObject({ completed: false }); + }); + it('counts started entries against the segment limit and marks its checkpoint boundary', async () => { const fixture: Fixture = await createFixture(); const options: Array<{ readonly isSegmentBoundary?: boolean } | undefined> = []; diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts index df07d33..5b4478d 100644 --- a/src/storage/streamed-parsed-document-sync.ts +++ b/src/storage/streamed-parsed-document-sync.ts @@ -400,6 +400,7 @@ async function openArchiveTransport( useWebWorkers: false, checkCrc32: true, }); + const producers: Set> = new Set(); try { const entries = await reader.getEntries(); const files: FileEntry[] = entries.filter( @@ -424,10 +425,12 @@ async function openArchiveTransport( yield { path: normalizeResultPath(entry.filename), contentLength: entry.uncompressedSize, - read: (): ReadableStream => readRangeEntry(entry, params.signal), + read: (): ReadableStream => readRangeEntry(entry, params.signal, producers), }; })(), close: async (): Promise => { + // zip.js closes stream output before getData finishes its own cleanup. + await Promise.all(producers); await reader.close(); }, }; @@ -440,18 +443,26 @@ async function openArchiveTransport( function readRangeEntry( entry: FileEntry, signal: AbortSignal | undefined, + producers: Set>, ): ReadableStream { const stream = new TransformStream(); // The producer and sink run together. The TransformStream applies sink backpressure. - void entry + const producer: Promise = entry .getData(stream.writable, { signal, checkCrc32: true, useWebWorkers: false }) - .catch(async (error: unknown): Promise => { - try { - await stream.writable.abort(error); - } catch { - /* Producer already aborted the stream. */ - } + .then( + (): void => undefined, + async (error: unknown): Promise => { + try { + await stream.writable.abort(error); + } catch { + /* Producer already aborted the stream. */ + } + }, + ) + .finally((): void => { + producers.delete(producer); }); + producers.add(producer); return stream.readable; } From 041027015343cdb3fa60d52cf88003de637aec61 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 10:21:47 -0700 Subject: [PATCH 5/8] fix(release): resolve main when publishing a feature beta --- .github/workflows/release.yml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index fa93d23..5b43fbe 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -99,6 +99,12 @@ jobs: fetch-depth: 0 ref: ${{ inputs.ref }} + - name: Resolve Changesets base branch + run: | + if ! git show-ref --verify --quiet refs/heads/main; then + git branch main origin/main + fi + - name: Setup pnpm uses: pnpm/action-setup@v4 with: From 26d0a3543b45efd8454d70dc28e051a9bbebd3e2 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 11:01:38 -0700 Subject: [PATCH 6/8] fix: preserve archive cache identity across CDN timestamp drift Cached Blob Range responses keep stable strong ETags and byte lengths while Last-Modified varies. Use the SDK cache validator for cached reads, propagate errors through the owned stream controller, and await each producer before advancing checkpoints. Preserve cache bypass through storage error causes. --- docs/streamed-parsed-document-sync.md | 5 +- .../streamed-sync-concurrency.test.ts | 174 ++++++++++++++++-- src/storage/streamed-parsed-document-sync.ts | 62 +++++-- 3 files changed, 209 insertions(+), 32 deletions(-) diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md index 8902d83..b03b8f2 100644 --- a/docs/streamed-parsed-document-sync.md +++ b/docs/streamed-parsed-document-sync.md @@ -111,7 +111,10 @@ identity, SHA-256, byte count, and immutable storage version after a complete up Each invocation refreshes the signed origin URL and checks its canonical identity. A cached ZIP must match its marker, HEAD metadata, and Range transport ETag/total -length; each response also validates its actual byte count. Checkpoint identity +length; each response also validates its actual byte count. Cached reads use these +strong version and length checks instead of zip.js Last-Modified comparisons, +because CDN timestamps can vary between responses. Origin reads retain zip.js +resource-change checks. Checkpoint identity continues to use the origin identity and central directory, so switching between origin and cached transport preserves existing checkpoints. Corrupt content and changed origin identities fail rather than committing a snapshot. diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts index ae4edd0..3a2fa5c 100644 --- a/src/storage/__tests__/streamed-sync-concurrency.test.ts +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -31,6 +31,7 @@ async function createFixture( readonly sequential?: boolean; readonly cache?: boolean; readonly failPath?: string; + readonly wrapErrors?: boolean; readonly onStart?: (path: string) => void; readonly delays?: Readonly>; } = {}, @@ -103,6 +104,9 @@ async function createFixture( objects.set(params.path, Buffer.concat(parts)); finished.push(params.path); return { ...params, entityTag: '"archive"' }; + } catch (error) { + if (options.wrapErrors) throw new Error('Storage adapter write failed', { cause: error }); + throw error; } finally { active -= 1; } @@ -267,6 +271,77 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); }); + it('rejects a producer failure while its writable is locked without hanging the sink', async () => { + const fixture: Fixture = await createFixture(); + let rejectProducer: (() => void) | undefined; + const producerRejected: Promise = new Promise((resolve) => { + rejectProducer = resolve; + }); + let heldWriter: WritableStreamDefaultWriter | undefined; + const entry: FileEntry = { + filename: 'manifest.json', + directory: false, + uncompressedSize: 16, + signature: 0, + getData(writable: WritableStream): Promise { + heldWriter = writable.getWriter(); + void heldWriter.closed.catch((): void => undefined); + rejectProducer?.(); + return Promise.reject(new Error('Locked producer failed')); + }, + } as FileEntry; + vi.spyOn(ZipReader.prototype, 'getEntries').mockResolvedValue([entry]); + fixture.storage.writeObjectStream = async ( + params: ParsedDocumentWriteObjectStreamParams, + ): Promise => { + await params.body.pipeTo(new WritableStream()); + return params; + }; + const result: Promise = fixture.sync().then( + (): string => 'completed', + (error: unknown): string => (error instanceof Error ? error.message : 'unknown failure'), + ); + try { + await producerRejected; + const state: string = await Promise.race([ + result, + new Promise((resolve) => setImmediate((): void => resolve('pending'))), + ]); + expect(state).toBe('Locked producer failed'); + } finally { + await heldWriter?.abort(new Error('Release held producer fixture')); + heldWriter?.releaseLock(); + await result; + } + }); + + it('rejects a late producer error after body EOF before checkpointing successful bytes', async () => { + const fixture: Fixture = await createFixture(); + const entry: FileEntry = { + filename: 'manifest.json', + directory: false, + uncompressedSize: 16, + signature: 0, + async getData(writable: WritableStream): Promise { + const writer: WritableStreamDefaultWriter = writable.getWriter(); + try { + await writer.write(Buffer.from('{"job_id":"job"}')); + await writer.close(); + } finally { + writer.releaseLock(); + } + throw new Error('Late producer failed'); + }, + } as FileEntry; + vi.spyOn(ZipReader.prototype, 'getEntries').mockResolvedValue([ + entry, + { ...entry, filename: 'chunks.json' }, + ]); + await expect(fixture.sync({ maxEntriesPerSync: 1 })).rejects.toThrow('Late producer failed'); + expect(fixture.progress).toHaveLength(0); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + it('waits for Range producer cleanup before returning an incomplete segment', async () => { const fixture: Fixture = await createFixture(); let startCleanup: (() => void) | undefined; @@ -277,10 +352,6 @@ describe('bounded parsed archive concurrency', () => { const cleanupGate: Promise = new Promise((resolve) => { releaseCleanup = resolve; }); - let finishCheckpoint: (() => void) | undefined; - const checkpointWritten: Promise = new Promise((resolve) => { - finishCheckpoint = resolve; - }); let hasCleanedProducer: boolean = false; const entry: FileEntry = { filename: 'manifest.json', @@ -304,25 +375,22 @@ describe('bounded parsed archive concurrency', () => { entry, { ...entry, filename: 'chunks.json' }, ]); - fixture.storage.writeSyncProgress = (value): Promise => { - fixture.progress.push(value); - finishCheckpoint?.(); - return Promise.resolve(); - }; const sync: Promise = fixture.sync({ maxEntriesPerSync: 1 }); try { - await Promise.all([cleanupStarted, checkpointWritten]); + await cleanupStarted; const state: string = await Promise.race([ sync.then((): string => 'returned'), new Promise((resolve) => setImmediate((): void => resolve('waiting'))), ]); expect(state).toBe('waiting'); expect(hasCleanedProducer).toBe(false); + expect(fixture.progress).toHaveLength(0); } finally { releaseCleanup?.(); await sync; } expect(hasCleanedProducer).toBe(true); + expect(fixture.progress.at(-1)?.nextArchiveEntry).toBe(1); expect(await sync).toMatchObject({ completed: false }); }); @@ -421,6 +489,68 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.started.filter((path) => path === 'chunks.json')).toHaveLength(1); }); + it('accepts cached Range responses with stable entity tags and changing Last-Modified metadata', async () => { + let hasStartedSnapshot: boolean = false; + const fixture: Fixture = await createFixture({ + cache: true, + onStart: (path): void => { + if (path !== 'result.zip') hasStartedSnapshot = true; + }, + }); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + const url: string = + typeof input === 'string' ? input : input instanceof URL ? input.href : input.url; + if (!url.startsWith('https://cache.example/')) return response; + const headers: Headers = new Headers(response.headers); + headers.set( + 'last-modified', + hasStartedSnapshot ? 'Tue, 01 Oct 2024 00:00:01 GMT' : 'Tue, 01 Oct 2024 00:00:00 GMT', + ); + return new Response(response.body, { status: response.status, headers }); + }, + ); + expect( + await syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + limits: { maxConcurrentFiles: 3, checkpointFileInterval: 4 }, + }), + ).toMatchObject({ completed: true }); + expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(false); + }); + + it('preserves origin Range checks when Last-Modified changes during extraction', async () => { + let hasStartedSnapshot: boolean = false; + const fixture: Fixture = await createFixture({ + onStart: (): void => { + hasStartedSnapshot = true; + }, + }); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + const headers: Headers = new Headers(response.headers); + headers.set( + 'last-modified', + hasStartedSnapshot ? 'Tue, 01 Oct 2024 00:00:01 GMT' : 'Tue, 01 Oct 2024 00:00:00 GMT', + ); + return new Response(response.body, { status: response.status, headers }); + }, + ); + await expect(fixture.sync({ maxConcurrentFiles: 3 })).rejects.toThrow('HTTP resource changed'); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + it('falls back to the origin if cached transport changes during archive setup', async () => { const fixture: Fixture = await createFixture({ cache: true }); const originalFetch: typeof fetch = fetch; @@ -460,12 +590,20 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(true); }); - it.each(['etag', 'http', 'network'] as const)( - 'persists cache bypass after an entry transport %s failure and retries from the origin', - async (mode) => { + it.each([ + ['etag', false], + ['etag', true], + ['http', false], + ['http', true], + ['network', false], + ['network', true], + ] as const)( + 'persists cache bypass after transport %s failure with adapter wrapping %s', + async (mode, wrapErrors) => { let isInvalid: boolean = false; const fixture: Fixture = await createFixture({ cache: true, + wrapErrors, onStart: (path): void => { if (path === 'file-0.txt') isInvalid = true; }, @@ -508,6 +646,16 @@ describe('bounded parsed archive concurrency', () => { }, ); + it('preserves a storage failure with a cyclic cause without hanging recovery', async () => { + const fixture: Fixture = await createFixture(); + const error: Error = new Error('Cyclic storage failure'); + error.cause = error; + fixture.storage.writeObjectStream = (): Promise => + Promise.reject(error); + await expect(fixture.sync()).rejects.toBe(error); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + }); + it('rejects a changed sequential origin after cache bypass without an initial identity', async () => { const fixture: Fixture = await createFixture({ sequential: true }); const originalFetch: typeof fetch = fetch; diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts index 5b4478d..c90c646 100644 --- a/src/storage/streamed-parsed-document-sync.ts +++ b/src/storage/streamed-parsed-document-sync.ts @@ -22,6 +22,12 @@ export interface ArchiveEntry { readonly path: string; readonly contentLength?: number; readonly read: () => ReadableStream; + readonly waitForRead?: () => Promise; +} + +interface RangeEntryRead { + readonly body: ReadableStream; + readonly completed: Promise; } export interface ArchiveSource { @@ -228,7 +234,7 @@ export async function syncRemoteResultArchive( controller.abort(error); await Promise.all(running); const hasCacheTransportFailure: boolean = - error instanceof ArchiveCacheTransportError || failure instanceof ArchiveCacheTransportError; + isArchiveCacheTransportFailure(error) || isArchiveCacheTransportFailure(failure); if (hasCacheTransportFailure) archiveCacheBypassed = true; await flushProgress(true, true).catch((): void => undefined); if (hasCacheTransportFailure) { @@ -254,6 +260,17 @@ export async function syncRemoteResultArchive( } } +function isArchiveCacheTransportFailure(error: unknown): boolean { + const causes: Set = new Set(); + let current: unknown = error; + while (current instanceof Error && !causes.has(current)) { + if (current instanceof ArchiveCacheTransportError) return true; + causes.add(current); + current = current.cause; + } + return false; +} + /** Validate configuration before job lookup, download, or storage writes. */ export function validateArchiveSyncLimits(limits: ParsedDocumentStorageLimits | undefined): void { const names: readonly (keyof ParsedDocumentStorageLimits)[] = [ @@ -297,6 +314,7 @@ async function copyEntry( contentLength: entry.contentLength, signal: params.signal, }); + await entry.waitForRead?.(); return createHash('sha256').update(entry.path).update(hash.digest()).digest('hex'); } @@ -375,7 +393,8 @@ async function openArchiveTransport( const options: HttpOptions = { forceRangeRequests: true, preventHeadRequest: true, - checkResourceChanges: true, + // Cached reads already pin a strong ETag and exact lengths; CDN timestamps can drift. + checkResourceChanges: !prepared?.isCached, fetch: async (input, init): Promise => { let response: Response; try { @@ -421,12 +440,18 @@ async function openArchiveTransport( isCached: prepared?.isCached, canSkipEntries: previous?.archiveIdentity !== identity, entries: (function* (): Generator { - for (const entry of files) + for (const entry of files) { + let entryRead: RangeEntryRead | undefined; yield { path: normalizeResultPath(entry.filename), contentLength: entry.uncompressedSize, - read: (): ReadableStream => readRangeEntry(entry, params.signal, producers), + read: (): ReadableStream => { + entryRead = readRangeEntry(entry, params.signal, producers); + return entryRead.body; + }, + waitForRead: (): Promise => entryRead?.completed ?? Promise.resolve(), }; + } })(), close: async (): Promise => { // zip.js closes stream output before getData finishes its own cleanup. @@ -444,26 +469,27 @@ function readRangeEntry( entry: FileEntry, signal: AbortSignal | undefined, producers: Set>, -): ReadableStream { - const stream = new TransformStream(); +): RangeEntryRead { + let streamController: TransformStreamDefaultController | undefined; + const stream = new TransformStream({ + start(controller): void { + streamController = controller; + }, + }); // The producer and sink run together. The TransformStream applies sink backpressure. - const producer: Promise = entry + const completed: Promise = entry .getData(stream.writable, { signal, checkCrc32: true, useWebWorkers: false }) - .then( - (): void => undefined, - async (error: unknown): Promise => { - try { - await stream.writable.abort(error); - } catch { - /* Producer already aborted the stream. */ - } - }, - ) + .then((): void => undefined); + const producer: Promise = completed + .catch((error: unknown): void => { + // Controller errors reach the sink even while zip.js owns the writable lock. + streamController?.error(error); + }) .finally((): void => { producers.delete(producer); }); producers.add(producer); - return stream.readable; + return { body: stream.readable, completed }; } async function writeStream( From 7e6c5b2faf2fec715c50cf1c08d0c4d91dd884f1 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 11:02:23 -0700 Subject: [PATCH 7/8] test(storage): validate concurrent copies with distinct image payloads --- scripts/validation/streamed-sync-memory.mjs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/scripts/validation/streamed-sync-memory.mjs b/scripts/validation/streamed-sync-memory.mjs index d331aa5..2674ee2 100644 --- a/scripts/validation/streamed-sync-memory.mjs +++ b/scripts/validation/streamed-sync-memory.mjs @@ -57,6 +57,8 @@ async function generateArchive(archivePath, totalBytes, imageCount) { const block = randomFillSync(Buffer.alloc(64 * 1024)); let remaining = totalBytes; for (let index = 0; index < imageCount; index += 1) { + // Distinct payloads make CRC checks detect cross-entry mixups during concurrency. + randomFillSync(block); let imageBytes = Math.min(remaining, Math.ceil(totalBytes / imageCount)); remaining -= imageBytes; const body = new ReadableStream({ From 7b5a6f8af94ce2a1e1a88defd4a513ac34551fb1 Mon Sep 17 00:00:00 2001 From: suguanYang Date: Mon, 26 Oct 2026 11:20:59 -0700 Subject: [PATCH 8/8] fix: resume cached archives through sequential origin fallback --- docs/streamed-parsed-document-sync.md | 10 +- .../streamed-sync-concurrency.test.ts | 133 +++++++++++++++++- src/storage/streamed-parsed-document-sync.ts | 17 ++- src/types/storage.ts | 2 + 4 files changed, 157 insertions(+), 5 deletions(-) diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md index b03b8f2..104795f 100644 --- a/docs/streamed-parsed-document-sync.md +++ b/docs/streamed-parsed-document-sync.md @@ -116,8 +116,14 @@ strong version and length checks instead of zip.js Last-Modified comparisons, because CDN timestamps can vary between responses. Origin reads retain zip.js resource-change checks. Checkpoint identity continues to use the origin identity and central directory, so switching between -origin and cached transport preserves existing checkpoints. Corrupt content and -changed origin identities fail rather than committing a snapshot. +origin and cached Range transport preserves existing checkpoints. New cached +checkpoints also retain `archiveDownloadIdentity`, the validated strong canonical +origin identity. If a later cache bypass uses an origin without Range support, +that proof preserves the directory identity while the sequential reader verifies +every completed entry hash before continuing. A changed origin or saved-prefix +mismatch fails; older Range checkpoints without that proof keep their existing +strict identity checks. Corrupt content and changed origin identities fail rather +than committing a snapshot. Optional caching has a thirty-second budget. Missing capabilities, interrupted uploads, cache metadata failures, or unavailable Range transport persist diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts index 3a2fa5c..b9757b4 100644 --- a/src/storage/__tests__/streamed-sync-concurrency.test.ts +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -29,6 +29,7 @@ interface Fixture { async function createFixture( options: { readonly sequential?: boolean; + readonly originSequential?: boolean; readonly cache?: boolean; readonly failPath?: string; readonly wrapErrors?: boolean; @@ -45,10 +46,16 @@ async function createFixture( compression: 'DEFLATE', }); const fetchMock: ReturnType = vi.fn( - (_input: unknown, init?: RequestInit): Promise => { + (input: unknown, init?: RequestInit): Promise => { const range: string | null = new Headers(init?.headers).get('range'); + const isSequentialOrigin: boolean = + options.originSequential === true && + typeof input === 'string' && + input.startsWith('https://archive.example/'); const match: RegExpExecArray | null = - range && !options.sequential ? /^bytes=(\d*)-(\d*)$/.exec(range) : null; + range && !options.sequential && !isSequentialOrigin + ? /^bytes=(\d*)-(\d*)$/.exec(range) + : null; const start: number = match?.[1] ? Number(match[1]) : 0; const end: number = match?.[2] ? Math.min(Number(match[2]), archive.length - 1) @@ -489,6 +496,128 @@ describe('bounded parsed archive concurrency', () => { expect(fixture.started.filter((path) => path === 'chunks.json')).toHaveLength(1); }); + it('resumes cached Range checkpoints through sequential origin after persistent cache bypass', async () => { + const fixture: Fixture = await createFixture({ cache: true, originSequential: true }); + const sync = (maxEntriesPerSync: number): Promise => + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + limits: { maxEntriesPerSync }, + }); + expect(await sync(1)).toMatchObject({ completed: false }); + const checkpoint: ParsedDocumentSyncProgress | undefined = fixture.progress.at(-1); + fixture.storage.getObjectUrl = (): Promise => Promise.resolve(null); + expect(await sync(1)).toMatchObject({ completed: false }); + expect(await sync(100)).toMatchObject({ completed: true }); + expect(typeof checkpoint?.archiveDownloadIdentity).toBe('string'); + for (const progress of fixture.progress) { + expect(progress.archiveIdentity).toBe(checkpoint?.archiveIdentity); + expect(progress.archiveDownloadIdentity).toBe(checkpoint?.archiveDownloadIdentity); + } + expect(fixture.progress.at(-1)?.archiveCacheBypassed).toBe(true); + expect(fixture.started.filter((path) => path === 'manifest.json')).toHaveLength(1); + expect(fixture.started.filter((path) => path === 'chunks.json')).toHaveLength(1); + expect(await sync(1)).toMatchObject({ completed: false }); + expect(fixture.progress.at(-1)).toMatchObject({ + archiveIdentity: checkpoint?.archiveIdentity, + archiveDownloadIdentity: checkpoint?.archiveDownloadIdentity, + nextArchiveEntry: 1, + }); + }); + + it.each(['changed origin', 'missing origin proof', 'changed saved prefix'] as const)( + 'rejects sequential fallback from cached Range with %s', + async (mode) => { + const fixture: Fixture = await createFixture({ cache: true, originSequential: true }); + const sync = (): Promise => + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + limits: { maxEntriesPerSync: 1 }, + }); + await sync(); + fixture.storage.getObjectUrl = (): Promise => Promise.resolve(null); + const checkpoint: ParsedDocumentSyncProgress = fixture.progress.at(-1)!; + if (mode === 'changed origin') { + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + const headers: Headers = new Headers(response.headers); + headers.set('etag', '"changed-origin"'); + return new Response(response.body, { status: response.status, headers }); + }, + ); + } else { + fixture.progress.push({ + ...checkpoint, + ...(mode === 'missing origin proof' + ? { archiveDownloadIdentity: undefined } + : { archiveEntryHashes: ['different saved entry hash'] }), + }); + } + await expect(sync()).rejects.toThrow( + mode === 'changed saved prefix' ? /entry changed while resuming/ : /identity changed/, + ); + expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false); + expect(fixture.started.filter((path) => path !== 'result.zip')).toHaveLength(1); + }, + ); + + it('retains cached origin proof when extraction failure persists bypass before sequential resume', async () => { + let isInvalid: boolean = false; + const fixture: Fixture = await createFixture({ + cache: true, + originSequential: true, + wrapErrors: true, + onStart: (path): void => { + if (path === 'file-0.txt') isInvalid = true; + }, + }); + const originalFetch: typeof fetch = fetch; + vi.stubGlobal( + 'fetch', + async (input: string | URL | Request, init?: RequestInit): Promise => { + const response: Response = await originalFetch(input, init); + if (typeof input !== 'string' || !input.startsWith('https://cache.example/') || !isInvalid) + return response; + const headers: Headers = new Headers(response.headers); + headers.set('etag', '"changed-cache"'); + return new Response(response.body, { status: response.status, headers }); + }, + ); + const sync = (): Promise => + syncRemoteResultArchive({ + documentId: 'document', + revisionKey: 'revision', + jobId: 'job', + resultUrl: 'https://archive.example/result.zip', + storage: fixture.storage, + cacheResultArchive: true, + }); + await expect(sync()).rejects.toThrow(); + const checkpoint: ParsedDocumentSyncProgress | undefined = fixture.progress.at(-1); + expect(checkpoint).toMatchObject({ + archiveCacheBypassed: true, + nextArchiveEntry: 2, + }); + expect(typeof checkpoint?.archiveDownloadIdentity).toBe('string'); + expect(await sync()).toMatchObject({ completed: true }); + expect(fixture.progress.at(-1)).toMatchObject({ + archiveIdentity: checkpoint?.archiveIdentity, + archiveDownloadIdentity: checkpoint?.archiveDownloadIdentity, + }); + }); + it('accepts cached Range responses with stable entity tags and changing Last-Modified metadata', async () => { let hasStartedSnapshot: boolean = false; const fixture: Fixture = await createFixture({ diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts index c90c646..627b77a 100644 --- a/src/storage/streamed-parsed-document-sync.ts +++ b/src/storage/streamed-parsed-document-sync.ts @@ -70,7 +70,18 @@ export async function syncRemoteResultArchive( : controller.signal; const transferParams: StreamedSyncParams = { ...params, signal }; const previous: ParsedDocumentSyncProgress | null = await params.storage.readSyncProgress(params); - const source: ArchiveSource = await openArchive(transferParams, previous); + const openedSource: ArchiveSource = await openArchive(transferParams, previous); + // A validated cache can support Range even when its unchanged origin cannot. + // Keep the directory identity while the sequential reader verifies every saved hash. + const source: ArchiveSource = + !openedSource.canSkipEntries && + previous?.archiveIdentity && + previous.archiveDownloadIdentity === openedSource.downloadIdentity + ? { ...openedSource, identity: previous.archiveIdentity } + : openedSource; + const archiveDownloadIdentity: string | undefined = source.isCached + ? source.downloadIdentity + : previous?.archiveDownloadIdentity; const concurrency: number = source.canSkipEntries ? (params.limits?.maxConcurrentFiles ?? 1) : 1; const nextEntry: number = previous?.status === 'completed' ? 0 : (previous?.nextArchiveEntry ?? 0); @@ -102,6 +113,7 @@ export async function syncRemoteResultArchive( revisionKey: params.revisionKey, pinnedRevisionKey: params.revisionKey, archiveIdentity: source.identity, + archiveDownloadIdentity, archiveCacheBypassed, archiveEntryHashes: entryHashes.slice(0, contiguousEntry), nextArchiveEntry: contiguousEntry, @@ -209,6 +221,7 @@ export async function syncRemoteResultArchive( revisionKey: params.revisionKey, pinnedRevisionKey: params.revisionKey, archiveIdentity: source.identity, + archiveDownloadIdentity, archiveCacheBypassed, archiveEntryHashes: entryHashes, nextArchiveEntry: entryIndex, @@ -245,6 +258,7 @@ export async function syncRemoteResultArchive( revisionKey: params.revisionKey, pinnedRevisionKey: params.revisionKey, archiveIdentity: source.identity, + archiveDownloadIdentity, archiveEntryHashes: entryHashes.slice(0, checkpointEntry), nextArchiveEntry: checkpointEntry, status: 'running', @@ -351,6 +365,7 @@ async function persistArchiveCacheBypass( documentId: params.documentId, revisionKey: params.revisionKey, pinnedRevisionKey: params.revisionKey, + archiveDownloadIdentity: previous?.archiveDownloadIdentity, archiveCacheBypassed: true, status: 'running', updatedAt: new Date().toISOString(), diff --git a/src/types/storage.ts b/src/types/storage.ts index fe06326..1a95fe7 100644 --- a/src/types/storage.ts +++ b/src/types/storage.ts @@ -100,6 +100,8 @@ export interface ParsedDocumentSyncProgress { readonly nextChunkPage?: number; readonly pinnedRevisionKey?: string; readonly archiveIdentity?: string; + /** Strong canonical origin identity retained when a cached archive resumes sequentially. */ + readonly archiveDownloadIdentity?: string; /** Once true, later segments use the origin rather than retrying optional archive caching. */ readonly archiveCacheBypassed?: boolean; readonly nextArchiveEntry?: number;