perf(storage): bound concurrent sync and reuse retained ZIP archives

.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.

.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:

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,20 +17,26 @@ 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):
   Promise<ParsedDocumentWriteObjectResult>;
 ```
 
 The parameters match `writeObject` except `body` is a Web
-`ReadableStream<Uint8Array>`, with optional `signal: AbortSignal`. Consume the
+`ReadableStream<Uint8Array>`, 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.

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)),