fix: cancel failed archive Range reads before decoder cleanup

.changeset/fix-range-cancellation.md

@@ -0,0 +1,11 @@
+---
+'@ontos-ai/knowhere-sdk': patch
+---
+
+Cancel all Range reads and owned decoder output streams when an archive read fails or synchronization is aborted. This prevents cache validation failures from waiting indefinitely on decoder cleanup, preserves contiguous checkpoints and cache bypass for retry, and reports non-206 cache responses with their HTTP status.
+
+Abort invalid prefetched Range responses before a decoder reads them, and propagate archive transport cancellation to storage adapters waiting for capacity. Preserve cached-setup fallback and retry checkpoints.
+
+Tag beta releases with the checked-out source commit when the workflow dispatch ref differs from the selected build ref.
+
+Recheck cancellation after waiting for reader-pointer publication and metadata verification, and between final/current pointer writes. Cancelled ordinary and progressive synchronization calls cannot start later pointer writes or return a completed response.

docs/streamed-parsed-document-sync.md

@@ -92,6 +92,13 @@ Re-enqueued running or failed transfers continue from their successful checkpoin
 Existing remote-chunk limits (`remotePageSize`, `maxPagesPerSync`, and
 `syncDeadlineMs`) continue to bound reconstruction when no archive is available.
 
+Range validation failures also cancel the signal supplied to active storage writes,
+including adapters waiting for capacity before reading their body. Response validation
+aborts the archive transport immediately, even when the decoder has not yet consumed
+that response. Failed cached setup keeps its cancellation isolated so origin fallback
+can proceed. Adapters must reject on signal cancellation and settle their resources
+before returning; callers must not rely only on a failed body read to release capacity.
+
 ## Optional original ZIP
 
 `retainResultArchive: true` streams the exact source archive to `result.zip` after

scripts/publish-beta-release.mjs

@@ -24,6 +24,8 @@ const packageJson = releaseUtils.readJson('package.json');
 const packageName = packageJson.name;
 const version = packageJson.version;
 const tagName = `v${version}`;
+// workflow_dispatch can check out inputs.ref while GITHUB_SHA still identifies the workflow ref.
+const sourceCommit = releaseUtils.readCommand('git', ['rev-parse', 'HEAD']).trim();
 
 if (!version.includes(`-${distTag}.`)) {
   throw new Error(`Beta publish expected ${version} to include prerelease tag ${distTag}`);
@@ -42,8 +44,7 @@ const releaseNotes = [
   `Beta snapshot publish for \`${packageName}@${version}\`.`,
   '',
   `- npm dist-tag: \`${distTag}\``,
-  `- git ref: \`${process.env.GITHUB_REF_NAME ?? 'local'}\``,
-  `- commit: \`${process.env.GITHUB_SHA ?? 'local'}\``,
+  `- source commit: \`${sourceCommit}\``,
   '',
   'This prerelease was generated from pending changesets in CI without committing prerelease state back to the source branch.',
 ].join('\n');
@@ -59,6 +60,6 @@ releaseUtils.runCommand('gh', [
   '--notes-file',
   notesPath,
   '--target',
-  process.env.GITHUB_SHA ?? 'HEAD',
+  sourceCommit,
   '--prerelease',
 ]);

src/__tests__/beta-release-contract.test.ts

@@ -0,0 +1,81 @@
+import { spawnSync } from 'node:child_process';
+import { mkdtemp, mkdir, readFile, writeFile, copyFile, rm } from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { delimiter, join } from 'node:path';
+import { expect, it } from 'vitest';
+
+it('tags the checked-out beta source when workflow metadata identifies another ref', async (): Promise<void> => {
+  const directory: string = await mkdtemp(join(tmpdir(), 'knowhere-beta-contract-'));
+  const bin: string = join(directory, 'bin');
+  const scripts: string = join(directory, 'scripts');
+  const actualCommit: string = 'bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb';
+  try {
+    await Promise.all([mkdir(bin), mkdir(scripts)]);
+    await Promise.all(
+      ['publish-beta-release.mjs', 'release-utils.mjs'].map((name: string) =>
+        copyFile(join(process.cwd(), 'scripts', name), join(scripts, name)),
+      ),
+    );
+    await writeFile(
+      join(directory, 'package.json'),
+      JSON.stringify({ name: '@ontos-ai/knowhere-sdk', version: '2.6.0' }),
+    );
+    const commandBody: string = `
+import { readFileSync, writeFileSync } from 'node:fs';
+const args = process.argv.slice(2);
+if (COMMAND === 'git' && args.join(' ') === 'rev-parse HEAD') {
+  console.log('${actualCommit}');
+} else if (COMMAND === 'pnpm' && args[0] === 'exec' && args[1] === 'changeset') {
+  if (args[2] === 'version') {
+    const manifest = JSON.parse(readFileSync('package.json', 'utf8'));
+    manifest.version = '2.6.1-beta.20990101000000';
+    writeFileSync('package.json', JSON.stringify(manifest));
+  }
+} else if (COMMAND === 'pnpm' && args[0] === 'list') {
+  console.log(JSON.stringify([{ path: process.cwd() }]));
+} else if (COMMAND === 'npm' && args[0] === 'view') {
+  process.exit(0);
+} else if (COMMAND === 'gh' && args[0] === 'release' && args[1] === 'view') {
+  process.exit(1);
+} else if (COMMAND === 'gh' && args[0] === 'release' && args[1] === 'create') {
+  writeFileSync(process.env.RELEASE_CAPTURE_PATH, JSON.stringify({
+    target: args[args.indexOf('--target') + 1],
+    notes: readFileSync(args[args.indexOf('--notes-file') + 1], 'utf8')
+  }));
+} else {
+  throw new Error('Unexpected release boundary command: ' + COMMAND + ' ' + args.join(' '));
+}
+`;
+    await Promise.all(
+      ['git', 'pnpm', 'npm', 'gh'].map((name: string) =>
+        writeFile(
+          join(bin, name),
+          `#!/usr/bin/env node\nconst COMMAND = '${name}';\n${commandBody}`,
+          { mode: 0o755 },
+        ),
+      ),
+    );
+    const capturePath: string = join(directory, 'release.json');
+    const result = spawnSync(process.execPath, [join(scripts, 'publish-beta-release.mjs')], {
+      cwd: directory,
+      encoding: 'utf8',
+      env: {
+        ...process.env,
+        PATH: `${bin}${delimiter}${process.env.PATH ?? ''}`,
+        GITHUB_SHA: 'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa',
+        GITHUB_REF_NAME: 'main',
+        RELEASE_CAPTURE_PATH: capturePath,
+      },
+    });
+    expect(result.status, result.stderr).toBe(0);
+    const captured: unknown = JSON.parse(await readFile(capturePath, 'utf8'));
+    expect(captured).toMatchObject({ target: actualCommit });
+    const notes: unknown =
+      typeof captured === 'object' && captured !== null && 'notes' in captured
+        ? captured.notes
+        : undefined;
+    expect(notes).toContain(actualCommit);
+  } finally {
+    await rm(directory, { recursive: true, force: true });
+  }
+});

src/knowledge/__tests__/progressive-snapshots.test.ts

@@ -3,6 +3,8 @@ import JSZip from 'jszip';
 import { afterEach, describe, expect, it, vi } from 'vitest';
 import { Knowhere } from '../../client.js';
 import type { JobResult, JobSnapshot } from '../../types/job.js';
+import type { DocumentChunkListResponse } from '../../types/document.js';
+import { serializeSnapshotPublication } from '../serialize-snapshot-publication.js';
 import type {
   ParsedDocumentObject,
   ParsedDocumentStorage,
@@ -132,6 +134,109 @@ afterEach(() => {
 });
 
 describe('progressive immutable snapshots', () => {
+  it.each(['legacy', 'baseline', 'final'] as const)(
+    'does not publish a %s reader pointer after cancellation during metadata verification',
+    async (stage): Promise<void> => {
+      const fixture: ProgressiveFixture = await createFixture();
+      if (stage === 'legacy') {
+        fixture.job.snapshots = undefined;
+        fixture.job.currentSnapshotId = undefined;
+      } else if (stage === 'final') {
+        fixture.job.snapshots = [fixture.final];
+        fixture.job.currentSnapshotId = fixture.final.snapshotId;
+        fixture.job.resultUrl = fixture.final.resultUrl;
+      }
+      const controller: AbortController = new AbortController();
+      const cancellation: Error = new Error('Cancelled during publication metadata verification');
+      const listChunks = vi.spyOn(fixture.client.documents, 'listChunks').getMockImplementation();
+      if (!listChunks) throw new Error('Missing document metadata fixture');
+      vi.spyOn(fixture.client.documents, 'listChunks').mockImplementation(
+        async (documentId, params): Promise<DocumentChunkListResponse> => {
+          const response: DocumentChunkListResponse = await listChunks(documentId, params);
+          if (!params?.includeAssetUrls) controller.abort(cancellation);
+          return response;
+        },
+      );
+      await expect(
+        fixture.knowledge.syncParsedDocument({
+          documentId: 'doc_test',
+          signal: controller.signal,
+        }),
+      ).rejects.toBe(cancellation);
+      expect(fixture.objects.has('current/.knowhere-sdk/current-snapshot.json')).toBe(false);
+      expect(fixture.objects.has('current/.knowhere-sdk/final-snapshot.json')).toBe(false);
+    },
+  );
+
+  it.each(['legacy', 'baseline', 'final'] as const)(
+    'does not publish a queued %s reader pointer after cancellation',
+    async (stage): Promise<void> => {
+      const fixture: ProgressiveFixture = await createFixture();
+      if (stage === 'legacy') {
+        fixture.job.snapshots = undefined;
+        fixture.job.currentSnapshotId = undefined;
+      } else if (stage === 'final') {
+        fixture.job.snapshots = [fixture.final];
+        fixture.job.currentSnapshotId = fixture.final.snapshotId;
+        fixture.job.resultUrl = fixture.final.resultUrl;
+      }
+      let releasePublication: () => void = (): void => {
+        throw new Error('Publication gate not initialized');
+      };
+      const gate: Promise<void> = new Promise((resolve): void => {
+        releasePublication = resolve;
+      });
+      const held: Promise<void> = serializeSnapshotPublication(
+        fixture.storage,
+        'doc_test',
+        (): Promise<void> => gate,
+      );
+      const controller: AbortController = new AbortController();
+      const cancellation: Error = new Error('Cancelled while reader publication was queued');
+      const operation = fixture.knowledge.syncParsedDocument({
+        documentId: 'doc_test',
+        signal: controller.signal,
+      });
+      const rejection = expect(operation).rejects.toBe(cancellation);
+      try {
+        await vi.waitFor((): void => {
+          expect(
+            [...fixture.objects.keys()].some((key: string): boolean =>
+              key.endsWith('/.knowhere-sdk/commit.json'),
+            ),
+          ).toBe(true);
+        });
+        controller.abort(cancellation);
+      } finally {
+        releasePublication();
+      }
+      await held;
+      await rejection;
+      expect(fixture.objects.has('current/.knowhere-sdk/current-snapshot.json')).toBe(false);
+      expect(fixture.objects.has('current/.knowhere-sdk/final-snapshot.json')).toBe(false);
+    },
+  );
+
+  it('stops current reader publication when cancelled during the final pointer write', async (): Promise<void> => {
+    const fixture: ProgressiveFixture = await createFixture();
+    fixture.job.snapshots = [fixture.final];
+    fixture.job.currentSnapshotId = fixture.final.snapshotId;
+    fixture.job.resultUrl = fixture.final.resultUrl;
+    const controller: AbortController = new AbortController();
+    const cancellation: Error = new Error('Cancelled during final reader pointer write');
+    const writeObject = fixture.storage.writeObject.bind(fixture.storage);
+    fixture.storage.writeObject = async (params): Promise<ParsedDocumentWriteObjectResult> => {
+      const result: ParsedDocumentWriteObjectResult = await writeObject(params);
+      if (params.path === '.knowhere-sdk/final-snapshot.json') controller.abort(cancellation);
+      return result;
+    };
+    await expect(
+      fixture.knowledge.syncParsedDocument({ documentId: 'doc_test', signal: controller.signal }),
+    ).rejects.toBe(cancellation);
+    expect(fixture.objects.has('current/.knowhere-sdk/final-snapshot.json')).toBe(true);
+    expect(fixture.objects.has('current/.knowhere-sdk/current-snapshot.json')).toBe(false);
+  });
+
   it('copies final under a new local key despite sharing the upstream Job Result and keeps pinned baseline readable', async () => {
     const fixture = await createFixture();
     const first = await fixture.knowledge.syncParsedDocument({ jobId: 'job_test' });

src/knowledge/knowledge.ts

@@ -209,14 +209,17 @@ export class Knowledge {
   ): Promise<KnowledgeSyncParsedDocumentResponse> {
     const response: KnowledgeSyncParsedDocumentResponse =
       await this.performParsedDocumentSync(params);
+    params.signal?.throwIfAborted();
     if (response.completed && !response.snapshot) {
       const storage = this.requireParsedStorageConfig().storage;
       await serializeSnapshotPublication(storage, response.documentId, async (): Promise<void> => {
+        params.signal?.throwIfAborted();
         if (params.documentId || params.jobId) {
           const current: DocumentChunkListResponse = await this.client.documents.listChunks(
             response.documentId,
             { page: 1, pageSize: 1 },
           );
+          params.signal?.throwIfAborted();
           if (params.jobId && current.jobId !== params.jobId) return;
           if (!params.jobId && current.jobResultId !== response.revisionKey) return;
         }
@@ -231,6 +234,7 @@ export class Knowledge {
         });
       });
     }
+    params.signal?.throwIfAborted();
     return response;
   }
 
@@ -341,15 +345,18 @@ export class Knowledge {
       signal: params.signal,
       retainResultArchive: params.retainResultArchive,
     });
+    params.signal?.throwIfAborted();
     if (response.completed && snapshot.snapshotId === params.job.currentSnapshotId) {
       await serializeSnapshotPublication(
         params.parsedStorageConfig.storage,
         params.documentId,
         async (): Promise<void> => {
+          params.signal?.throwIfAborted();
           const current: DocumentChunkListResponse = await this.client.documents.listChunks(
             params.documentId,
             { page: 1, pageSize: 1 },
           );
+          params.signal?.throwIfAborted();
           if (current.jobResultId !== snapshot.jobResultId) return;
           const pointer: Uint8Array = Buffer.from(
             JSON.stringify({
@@ -365,6 +372,7 @@ export class Knowledge {
               body: pointer,
               contentType: 'application/json; charset=utf-8',
             });
+            params.signal?.throwIfAborted();
           }
           await params.parsedStorageConfig.storage.writeObject({
             documentId: params.documentId,
@@ -376,6 +384,7 @@ export class Knowledge {
         },
       );
     }
+    params.signal?.throwIfAborted();
     return response;
   }
 

src/storage/__tests__/parsed-document-archive-cache.test.ts

@@ -225,6 +225,17 @@ describe('prepareCachedArchive', (): void => {
     );
   });
 
+  it('reports a non-206 response without mistaking missing version headers for a revision change', async (): Promise<void> => {
+    const fixture: CacheFixture = createFixture();
+    const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage });
+    const cancel = vi.fn();
+    const response: Response = new Response(new ReadableStream<Uint8Array>({ cancel }), {
+      status: 503,
+    });
+    expect((): Response => prepared.validateResponse(response)).toThrow(/HTTP 503/);
+    expect(cancel).toHaveBeenCalledOnce();
+  });
+
   it('rejects a mismatched cached range total before consuming bytes', async (): Promise<void> => {
     const fixture: CacheFixture = createFixture();
     const prepared = await prepareCachedArchive({ ...document, storage: fixture.storage });