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/.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: diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md index 09e58a2..104795f 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,37 @@ 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. 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 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 +`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 +174,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..2674ee2 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 })); } } @@ -48,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({ @@ -75,7 +86,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 +97,7 @@ async function runWorker(archivePath, imageCount, concurrency) { archivePath, String(imageCount), String(concurrency), + cache ? 'cache' : 'range', ], { stdio: ['ignore', 'pipe', 'inherit'] }, ); @@ -97,12 +109,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 +127,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 +143,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 +166,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 +192,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 +243,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 +266,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..b9757b4 --- /dev/null +++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts @@ -0,0 +1,843 @@ +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'; +import type { + ParsedDocumentObject, + ParsedDocumentObjectHead, + ParsedDocumentStorage, + ParsedDocumentStorageLimits, + ParsedDocumentSyncProgress, + ParsedDocumentWriteObjectStreamParams, + ParsedDocumentWriteObjectResult, +} 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 originSequential?: boolean; + readonly cache?: boolean; + readonly failPath?: string; + readonly wrapErrors?: boolean; + 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 isSequentialOrigin: boolean = + options.originSequential === true && + typeof input === 'string' && + input.startsWith('https://archive.example/'); + const match: RegExpExecArray | 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) + : 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"' }; + } catch (error) { + if (options.wrapErrors) throw new Error('Storage adapter write failed', { cause: error }); + throw error; + } 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: { '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; + }); + 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); + expect(fixture.finished).toHaveLength(2); + expect(pendingWrites).toBe(0); + expect(settledWrites).toBeGreaterThan(0); + 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; + const cleanupStarted: Promise = new Promise((resolve) => { + startCleanup = resolve; + }); + let releaseCleanup: (() => void) | undefined; + const cleanupGate: Promise = new Promise((resolve) => { + releaseCleanup = 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' }, + ]); + const sync: Promise = fixture.sync({ maxEntriesPerSync: 1 }); + try { + 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 }); + }); + + 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('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({ + 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; + 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', 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; + }, + }); + 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('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; + 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/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'; + } +} 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..627b77a 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,13 +20,22 @@ import { 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 { readonly canSkipEntries: boolean; readonly identity: string; readonly downloadIdentity: string; + readonly archiveCacheBypassed?: boolean; + readonly isCached?: boolean; readonly entries: AsyncIterable | Iterable; readonly close: () => Promise; } @@ -37,97 +49,158 @@ 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 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); 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, + archiveDownloadIdentity, + 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 +215,24 @@ 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, + archiveDownloadIdentity, + 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 +243,147 @@ export async function syncRemoteResultArchive( completed: true, commit, }; + } catch (error) { + controller.abort(error); + await Promise.all(running); + const hasCacheTransportFailure: boolean = + isArchiveCacheTransportFailure(error) || isArchiveCacheTransportFailure(failure); + 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, + archiveDownloadIdentity, + 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 { +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)[] = [ + '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, + }); + await entry.waitForRead?.(); + 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, + archiveDownloadIdentity: previous?.archiveDownloadIdentity, + 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,30 +391,50 @@ 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, + // Cached reads already pin a strong ETag and exact lengths; CDN timestamps can drift. + checkResourceChanges: !prepared?.isCached, 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, }); + const producers: Set> = new Set(); try { const entries = await reader.getEntries(); const files: FileEntry[] = entries.filter( @@ -218,17 +449,28 @@ 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) + for (const entry of files) { + let entryRead: RangeEntryRead | undefined; yield { path: normalizeResultPath(entry.filename), - read: (): ReadableStream => readRangeEntry(entry, params.signal), + contentLength: entry.uncompressedSize, + 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. + await Promise.all(producers); await reader.close(); }, }; @@ -241,19 +483,28 @@ async function openArchive(params: StreamedSyncParams): Promise { function readRangeEntry( entry: FileEntry, signal: AbortSignal | undefined, -): ReadableStream { - const stream = new TransformStream(); + producers: Set>, +): 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. - void entry + const completed: 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); + 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); }); - return stream.readable; + producers.add(producer); + return { body: stream.readable, completed }; } async function writeStream( diff --git a/src/types/storage.ts b/src/types/storage.ts index 80051ca..1a95fe7 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,10 @@ 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; /** Content hashes for verifying a sequential retry without a strong HTTP validator. */ readonly archiveEntryHashes?: readonly string[]; @@ -104,6 +112,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 +129,10 @@ export interface ParsedDocumentStorage { readSyncProgress( params: ParsedDocumentRevisionParams, ): Promise; - writeSyncProgress(params: ParsedDocumentSyncProgress): Promise; + writeSyncProgress( + params: ParsedDocumentSyncProgress, + options?: ParsedDocumentSyncProgressOptions, + ): Promise; } export interface ParsedDocumentSyncScheduler { @@ -126,6 +142,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;