From 0673119d450663db77e819d7b544d038235607f9 Mon Sep 17 00:00:00 2001
From: suguanYang <wangbinqi77@gmail.com>
Date: Mon, 26 Oct 2026 21:45:17 -0700
Subject: [PATCH 1/3] fix: cancel archive reads before decoder cleanup

---
 .changeset/fix-range-cancellation.md          |   5 +
 .../parsed-document-archive-cache.test.ts     |  11 +
 .../streamed-sync-concurrency.test.ts         | 226 +++++++++++++++++-
 src/storage/parsed-document-archive-cache.ts  |  10 +-
 src/storage/streamed-parsed-document-sync.ts  |  82 ++++++-
 5 files changed, 324 insertions(+), 10 deletions(-)
 create mode 100644 .changeset/fix-range-cancellation.md

diff --git a/.changeset/fix-range-cancellation.md b/.changeset/fix-range-cancellation.md
new file mode 100644
index 0000000..b08fcfa
--- /dev/null
+++ b/.changeset/fix-range-cancellation.md
@@ -0,0 +1,5 @@
+---
+"@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.
diff --git a/src/storage/__tests__/parsed-document-archive-cache.test.ts b/src/storage/__tests__/parsed-document-archive-cache.test.ts
index b503680..f3917ff 100644
--- a/src/storage/__tests__/parsed-document-archive-cache.test.ts
+++ b/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 });
diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts
index b9757b4..94cfdba 100644
--- a/src/storage/__tests__/streamed-sync-concurrency.test.ts
+++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts
@@ -1,3 +1,4 @@
+import type { ReadableStreamReadResult } from 'node:stream/web';
 import JSZip from 'jszip';
 import { ZipReader, type FileEntry } from '@zip.js/zip.js';
 import { afterEach, describe, expect, it, vi } from 'vitest';
@@ -15,6 +16,7 @@ import type {
 import { syncRemoteResultArchive } from '../streamed-parsed-document-sync.js';
 
 interface Fixture {
+  readonly archive: Uint8Array;
   readonly storage: ParsedDocumentStorage;
   readonly objects: Map<string, Uint8Array>;
   readonly progress: ParsedDocumentSyncProgress[];
@@ -28,6 +30,8 @@ interface Fixture {
 
 async function createFixture(
   options: {
+    readonly fileCount?: number;
+    readonly largeEntries?: boolean;
     readonly sequential?: boolean;
     readonly originSequential?: boolean;
     readonly cache?: boolean;
@@ -40,7 +44,11 @@ async function createFixture(
   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}`);
+  for (let index: number = 0; index < (options.fileCount ?? 8); index += 1)
+    zip.file(
+      `file-${index}.txt`,
+      options.largeEntries ? `body-${index}`.repeat(262144) : `body-${index}`,
+    );
   const archive: Uint8Array = await zip.generateAsync({
     type: 'uint8array',
     compression: 'DEFLATE',
@@ -135,6 +143,7 @@ async function createFixture(
       Promise.resolve('https://cache.example/result.zip');
   }
   return {
+    archive,
     storage,
     objects,
     progress,
@@ -278,6 +287,221 @@ describe('bounded parsed archive concurrency', () => {
     expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false);
   });
 
+  it.each([
+    { failure: 'etag', concurrency: 1 },
+    { failure: 'etag', concurrency: 2 },
+    { failure: 'etag', concurrency: 4 },
+    { failure: 'http', concurrency: 4 },
+    { failure: 'truncated', concurrency: 4 },
+  ] as const)(
+    'cancels cached Range reads before decoder cleanup after $failure failure at concurrency $concurrency',
+    async ({ failure, concurrency }) => {
+      const fixture: Fixture = await createFixture({
+        cache: true,
+        largeEntries: true,
+        fileCount: 4,
+        wrapErrors: true,
+      });
+      const archive: Buffer = Buffer.from(fixture.archive);
+      const headerOffset: number = archive.indexOf(Buffer.from('file-0.txt')) - 30;
+      expect(archive.readUInt32LE(headerOffset)).toBe(0x04034b50);
+      const dataOffset: number = headerOffset + 30 + Buffer.byteLength('file-0.txt');
+      const dataEnd: number = dataOffset + archive.readUInt32LE(headerOffset + 18);
+      let isInvalid: boolean = false;
+      let markFailure: (() => void) | undefined;
+      const failedResponse: Promise<void> = new Promise((resolve) => {
+        markFailure = resolve;
+      });
+      const originalFetch: typeof fetch = fetch;
+      vi.stubGlobal(
+        'fetch',
+        async (input: string | URL | Request, init?: RequestInit): Promise<Response> => {
+          const response: Response = await originalFetch(input, init);
+          const url: string =
+            typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
+          const range: string | null = new Headers(init?.headers).get('range');
+          const offset: number = Number(/^bytes=(\d+)-/.exec(range ?? '')?.[1] ?? -1);
+          if (
+            !isInvalid ||
+            !url.startsWith('https://cache.example/') ||
+            offset < dataOffset ||
+            offset >= dataEnd
+          )
+            return response;
+          markFailure?.();
+          const headers: Headers = new Headers(response.headers);
+          if (failure === 'http') {
+            await response.body?.cancel();
+            return new Response('Service unavailable', { status: 503 });
+          }
+          if (failure === 'truncated') {
+            const bytes: Uint8Array = new Uint8Array(await response.arrayBuffer());
+            return new Response(bytes.subarray(0, bytes.length - 1), {
+              status: response.status,
+              headers,
+            });
+          }
+          headers.set('etag', '"changed-cache"');
+          return new Response(response.body, { status: response.status, headers });
+        },
+      );
+      const sync = (signal?: AbortSignal, maxEntriesPerSync?: number): Promise<unknown> =>
+        syncRemoteResultArchive({
+          documentId: 'document',
+          revisionKey: 'revision',
+          jobId: 'job',
+          resultUrl: 'https://archive.example/result.zip',
+          storage: fixture.storage,
+          cacheResultArchive: true,
+          signal,
+          limits: {
+            maxConcurrentFiles: concurrency,
+            checkpointFileInterval: 10,
+            maxEntriesPerSync,
+          },
+        });
+      expect(await sync(undefined, 2)).toMatchObject({ completed: false });
+      isInvalid = true;
+      // eslint-disable-next-line @typescript-eslint/unbound-method -- called with the current reader below.
+      const getEntries = ZipReader.prototype.getEntries;
+      vi.spyOn(ZipReader.prototype, 'getEntries').mockImplementation(async function (
+        this: ZipReader<unknown>,
+        options,
+      ) {
+        const entries = await getEntries.call(this, options);
+        for (const entry of entries) {
+          if (entry.directory || entry.filename !== 'file-0.txt') continue;
+          const getData = entry.getData.bind(entry);
+          vi.spyOn(entry, 'getData').mockImplementation(async (writable, entryOptions) => {
+            if (!isInvalid) return getData(writable, entryOptions);
+            if (!(writable instanceof WritableStream))
+              throw new Error('Expected a stream decoder sink.');
+            const writer: WritableStreamDefaultWriter<unknown> = writable.getWriter();
+            const destination: WritableStream<Uint8Array> = new WritableStream<Uint8Array>({
+              write: (chunk): Promise<void> => writer.write(chunk),
+              close: (): Promise<void> => writer.close(),
+              async abort(error: unknown): Promise<void> {
+                // Decoder cleanup needs its input cancelled before releasing the output lock.
+                await new Promise<void>((resolve) => {
+                  if (entryOptions?.signal?.aborted) resolve();
+                  else
+                    entryOptions?.signal?.addEventListener('abort', (): void => resolve(), {
+                      once: true,
+                    });
+                });
+                await writer.abort(error);
+              },
+            });
+            try {
+              return await getData(destination, entryOptions);
+            } finally {
+              writer.releaseLock();
+            }
+          });
+        }
+        return entries;
+      });
+      const controller: AbortController = new AbortController();
+      const outcome: Promise<string> = sync(controller.signal).then(
+        (): string => 'completed',
+        (): string => 'rejected',
+      );
+      try {
+        await failedResponse;
+        expect(
+          await Promise.race([
+            outcome,
+            new Promise<string>((resolve) => setTimeout((): void => resolve('pending'), 250)),
+          ]),
+        ).toBe('rejected');
+      } finally {
+        controller.abort(new Error('Release the regression fixture after assertion.'));
+        await outcome;
+      }
+      expect(fixture.progress.at(-1)).toMatchObject({
+        nextArchiveEntry: 2,
+        archiveCacheBypassed: true,
+      });
+      expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false);
+      const writesAtReturn: number = fixture.finished.length;
+      await new Promise<void>((resolve) => setTimeout(resolve, 20));
+      expect(fixture.finished).toHaveLength(writesAtReturn);
+      isInvalid = false;
+      expect(await sync()).toMatchObject({ completed: true });
+    },
+  );
+
+  it('interrupts a decoder awaiting output cleanup when the caller cancels', async () => {
+    const fixture: Fixture = await createFixture();
+    const controller: AbortController = new AbortController();
+    const reason: Error = new Error('Caller cancelled the archive.');
+    let heldWriter: WritableStreamDefaultWriter<Uint8Array> | undefined;
+    let markWaiting: (() => void) | undefined;
+    const waiting: Promise<void> = new Promise((resolve) => {
+      markWaiting = resolve;
+    });
+    let hasSettledProducer: boolean = false;
+    let markConsumed: (() => void) | undefined;
+    const consumed: Promise<void> = new Promise((resolve) => {
+      markConsumed = resolve;
+    });
+    fixture.storage.writeObjectStream = async (
+      params,
+    ): Promise<ParsedDocumentWriteObjectResult> => {
+      const reader: ReadableStreamDefaultReader<Uint8Array> = params.body.getReader();
+      try {
+        while (true) {
+          const part: ReadableStreamReadResult<Uint8Array> = await reader.read();
+          if (part.done) break;
+          params.signal?.throwIfAborted();
+          markConsumed?.();
+        }
+        return params;
+      } finally {
+        reader.releaseLock();
+      }
+    };
+    const entry: FileEntry = {
+      filename: 'manifest.json',
+      directory: false,
+      uncompressedSize: 16,
+      signature: 0,
+      async getData(writable: WritableStream<Uint8Array>): Promise<void> {
+        const writer: WritableStreamDefaultWriter<Uint8Array> = writable.getWriter();
+        heldWriter = writer;
+        try {
+          await writer.write(Buffer.from('{"job_id":"job"}'));
+          markWaiting?.();
+          await writer.closed;
+        } finally {
+          writer.releaseLock();
+          hasSettledProducer = true;
+        }
+      },
+    } as FileEntry;
+    vi.spyOn(ZipReader.prototype, 'getEntries').mockResolvedValue([entry]);
+    const outcome: Promise<string> = fixture.sync(undefined, controller.signal).then(
+      (): string => 'completed',
+      (): string => 'rejected',
+    );
+    try {
+      await waiting;
+      await consumed;
+      controller.abort(reason);
+      expect(
+        await Promise.race([
+          outcome,
+          new Promise<string>((resolve) => setTimeout((): void => resolve('pending'), 100)),
+        ]),
+      ).toBe('rejected');
+      expect(hasSettledProducer).toBe(true);
+      expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false);
+    } finally {
+      await heldWriter?.abort(reason).catch((): void => undefined);
+      await outcome;
+    }
+  });
+
   it('rejects a producer failure while its writable is locked without hanging the sink', async () => {
     const fixture: Fixture = await createFixture();
     let rejectProducer: (() => void) | undefined;
diff --git a/src/storage/parsed-document-archive-cache.ts b/src/storage/parsed-document-archive-cache.ts
index 4c09d54..ea076c7 100644
--- a/src/storage/parsed-document-archive-cache.ts
+++ b/src/storage/parsed-document-archive-cache.ts
@@ -359,15 +359,17 @@ function validateCachedResponse(
   marker: ArchiveCacheMarker,
   signal: AbortSignal | undefined,
 ): Response {
-  if (response.headers.get('etag') !== marker.storageEntityTag)
+  if (response.status !== 206 || !response.body)
     rejectResponse(
       response,
-      new ArchiveCacheTransportError('Parsed result archive cache response version changed.'),
+      new ArchiveCacheTransportError(
+        `Parsed result archive cache requires a ranged response (HTTP ${response.status}).`,
+      ),
     );
-  if (response.status !== 206 || !response.body)
+  if (response.headers.get('etag') !== marker.storageEntityTag)
     rejectResponse(
       response,
-      new ArchiveCacheTransportError('Parsed result archive cache requires a ranged response.'),
+      new ArchiveCacheTransportError('Parsed result archive cache response version changed.'),
     );
   const range: RegExpExecArray | null = /^bytes (\d+)-(\d+)\/(\d+)$/.exec(
     response.headers.get('content-range') ?? '',
diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts
index 627b77a..c5254be 100644
--- a/src/storage/streamed-parsed-document-sync.ts
+++ b/src/storage/streamed-parsed-document-sync.ts
@@ -1,8 +1,15 @@
+import type { ReadableStreamReadResult } from 'node:stream/web';
 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 {
+  HttpRangeReader,
+  ZipReader,
+  type CreateReadableOptions,
+  type FileEntry,
+  type HttpOptions,
+} from '@zip.js/zip.js';
 
 import type { KnowledgeSyncParsedDocumentResponse } from '../knowledge/types.js';
 import type {
@@ -374,6 +381,58 @@ async function persistArchiveCacheBypass(
   return progress;
 }
 
+/** Abort decoder inputs before waiting for their output-stream cleanup. */
+class AbortableHttpRangeReader extends HttpRangeReader {
+  constructor(
+    url: string,
+    options: HttpOptions,
+    private readonly controller: AbortController,
+  ) {
+    super(url, options);
+  }
+
+  override createReadable(options?: CreateReadableOptions): ReadableStream<Uint8Array> {
+    const reader: ReadableStreamDefaultReader<Uint8Array> = super
+      .createReadable(options)
+      .getReader();
+    const cancellation: AbortController = this.controller;
+    return new ReadableStream<Uint8Array>(
+      {
+        async pull(output): Promise<void> {
+          try {
+            const chunk: ReadableStreamReadResult<Uint8Array> = await reader.read();
+            if (chunk.done) {
+              reader.releaseLock();
+              output.close();
+            } else output.enqueue(chunk.value);
+          } catch (error) {
+            cancellation.abort(error);
+            reader.releaseLock();
+            output.error(error);
+          }
+        },
+        async cancel(reason: unknown): Promise<void> {
+          try {
+            await reader.cancel(reason);
+          } finally {
+            reader.releaseLock();
+          }
+        },
+      },
+      { highWaterMark: 0 },
+    );
+  }
+
+  override async readUint8Array(index: number, length: number): Promise<Uint8Array> {
+    try {
+      return await super.readUint8Array(index, length);
+    } catch (error) {
+      this.controller.abort(error);
+      throw error;
+    }
+  }
+}
+
 async function openArchiveTransport(
   params: StreamedSyncParams,
   previous: ParsedDocumentSyncProgress | null,
@@ -405,6 +464,11 @@ async function openArchiveTransport(
     );
     return { ...source, archiveCacheBypassed };
   }
+  // Isolate setup failure from the document signal so cached setup can still fall back.
+  const rangeController: AbortController = new AbortController();
+  const rangeSignal: AbortSignal = params.signal
+    ? AbortSignal.any([params.signal, rangeController.signal])
+    : rangeController.signal;
   const options: HttpOptions = {
     forceRangeRequests: true,
     preventHeadRequest: true,
@@ -413,9 +477,9 @@ async function openArchiveTransport(
     fetch: async (input, init): Promise<Response> => {
       let response: Response;
       try {
-        response = await fetch(input, { ...init, signal: params.signal });
+        response = await fetch(input, { ...init, signal: rangeSignal });
       } catch (error) {
-        params.signal?.throwIfAborted();
+        rangeSignal.throwIfAborted();
         if (prepared?.isCached)
           throw new ArchiveCacheTransportError(
             'Parsed result archive cache transport unavailable.',
@@ -430,7 +494,7 @@ async function openArchiveTransport(
       return prepared ? prepared.validateResponse(response) : response;
     },
   };
-  const reader = new ZipReader(new HttpRangeReader(resultUrl, options), {
+  const reader = new ZipReader(new AbortableHttpRangeReader(resultUrl, options, rangeController), {
     useWebWorkers: false,
     checkCrc32: true,
   });
@@ -461,7 +525,7 @@ async function openArchiveTransport(
             path: normalizeResultPath(entry.filename),
             contentLength: entry.uncompressedSize,
             read: (): ReadableStream<Uint8Array> => {
-              entryRead = readRangeEntry(entry, params.signal, producers);
+              entryRead = readRangeEntry(entry, rangeSignal, producers);
               return entryRead.body;
             },
             waitForRead: (): Promise<void> => entryRead?.completed ?? Promise.resolve(),
@@ -475,6 +539,7 @@ async function openArchiveTransport(
       },
     };
   } catch (error) {
+    rangeController.abort(error);
     await reader.close();
     throw error;
   }
@@ -491,6 +556,12 @@ function readRangeEntry(
       streamController = controller;
     },
   });
+  // Interrupt the owned output even when decoder cleanup still holds its writable lock.
+  const cancelOutput = (): void => {
+    streamController?.error(signal?.reason);
+  };
+  signal?.addEventListener('abort', cancelOutput, { once: true });
+  if (signal?.aborted) cancelOutput();
   // The producer and sink run together. The TransformStream applies sink backpressure.
   const completed: Promise<void> = entry
     .getData(stream.writable, { signal, checkCrc32: true, useWebWorkers: false })
@@ -501,6 +572,7 @@ function readRangeEntry(
       streamController?.error(error);
     })
     .finally((): void => {
+      signal?.removeEventListener('abort', cancelOutput);
       producers.delete(producer);
     });
   producers.add(producer);

From 511cf7715e333c9d8ae796e9af4e5437a15c08a4 Mon Sep 17 00:00:00 2001
From: suguanYang <wangbinqi77@gmail.com>
Date: Tue, 27 Oct 2026 06:52:14 -0700
Subject: [PATCH 2/3] fix: cancel prefetched archive failures before storage
 reads

---
 .changeset/fix-range-cancellation.md          |   4 +-
 docs/streamed-parsed-document-sync.md         |   7 +
 .../streamed-sync-concurrency.test.ts         | 146 +++++++++++++++++-
 src/storage/streamed-parsed-document-sync.ts  |  49 ++++--
 4 files changed, 188 insertions(+), 18 deletions(-)

diff --git a/.changeset/fix-range-cancellation.md b/.changeset/fix-range-cancellation.md
index b08fcfa..f41a01c 100644
--- a/.changeset/fix-range-cancellation.md
+++ b/.changeset/fix-range-cancellation.md
@@ -1,5 +1,7 @@
 ---
-"@ontos-ai/knowhere-sdk": patch
+'@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.
diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md
index 104795f..66541fc 100644
--- a/docs/streamed-parsed-document-sync.md
+++ b/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
diff --git a/src/storage/__tests__/streamed-sync-concurrency.test.ts b/src/storage/__tests__/streamed-sync-concurrency.test.ts
index 94cfdba..f8218cf 100644
--- a/src/storage/__tests__/streamed-sync-concurrency.test.ts
+++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts
@@ -1,6 +1,7 @@
 import type { ReadableStreamReadResult } from 'node:stream/web';
 import JSZip from 'jszip';
-import { ZipReader, type FileEntry } from '@zip.js/zip.js';
+import { randomBytes } from 'node:crypto';
+import { ZipReader, configure, 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';
@@ -30,6 +31,7 @@ interface Fixture {
 
 async function createFixture(
   options: {
+    readonly entropyBytes?: number;
     readonly fileCount?: number;
     readonly largeEntries?: boolean;
     readonly sequential?: boolean;
@@ -47,7 +49,11 @@ async function createFixture(
   for (let index: number = 0; index < (options.fileCount ?? 8); index += 1)
     zip.file(
       `file-${index}.txt`,
-      options.largeEntries ? `body-${index}`.repeat(262144) : `body-${index}`,
+      options.entropyBytes
+        ? randomBytes(options.entropyBytes)
+        : options.largeEntries
+          ? `body-${index}`.repeat(262144)
+          : `body-${index}`,
     );
   const archive: Uint8Array = await zip.generateAsync({
     type: 'uint8array',
@@ -431,6 +437,142 @@ describe('bounded parsed archive concurrency', () => {
     },
   );
 
+  it.each(['http', 'etag', 'network', 'origin-http'] as const)(
+    'cancels queued upload consumers after unread cached %s failure',
+    async (failure): Promise<void> => {
+      const originalWorkerCount: number =
+        typeof navigator === 'undefined' ? 2 : navigator.hardwareConcurrency;
+      configure({ maxWorkers: 1 });
+      const fixture: Fixture = await createFixture({
+        cache: true,
+        entropyBytes: 1_200_000,
+        fileCount: 2,
+      });
+      const synchronize = (signal?: AbortSignal, maxEntriesPerSync?: number): Promise<unknown> =>
+        syncRemoteResultArchive({
+          documentId: 'document',
+          revisionKey: 'revision',
+          jobId: 'job',
+          resultUrl: 'https://archive.example/result.zip',
+          storage: fixture.storage,
+          cacheResultArchive: failure !== 'origin-http',
+          limits: { maxConcurrentFiles: 2, maxEntriesPerSync },
+          signal,
+        });
+      await synchronize(undefined, 2);
+      const archive: Buffer = Buffer.from(fixture.archive);
+      const badOffset: number =
+        archive.indexOf(Buffer.from('file-1.txt')) + Buffer.byteLength('file-1.txt');
+      let markFailure: (() => void) | undefined;
+      const failedResponse: Promise<void> = new Promise((resolve) => {
+        markFailure = resolve;
+      });
+      const originalFetch: typeof fetch = fetch;
+      let isInvalid: boolean = true;
+      vi.stubGlobal(
+        'fetch',
+        async (input: string | URL | Request, init?: RequestInit): Promise<Response> => {
+          const url: string =
+            typeof input === 'string' ? input : input instanceof URL ? input.href : input.url;
+          const offset: number = Number(
+            /^bytes=(\d+)-/.exec(new Headers(init?.headers).get('range') ?? '')?.[1] ?? -1,
+          );
+          const response: Response = await originalFetch(input, init);
+          if (
+            isInvalid &&
+            url.startsWith(
+              failure === 'origin-http' ? 'https://archive.example/' : 'https://cache.example/',
+            ) &&
+            offset === badOffset
+          ) {
+            markFailure?.();
+            if (failure === 'etag') {
+              const headers: Headers = new Headers(response.headers);
+              headers.set('etag', '"invalid-cache"');
+              return new Response(response.body, { status: response.status, headers });
+            }
+            await response.body?.cancel();
+            if (failure === 'network') throw new TypeError('Cached transport unavailable');
+            const body: ReadableStream<Uint8Array> | string =
+              failure === 'origin-http'
+                ? new ReadableStream<Uint8Array>({
+                    cancel: () =>
+                      new Promise<void>((resolve) => {
+                        if (init?.signal?.aborted) resolve();
+                        else
+                          init?.signal?.addEventListener('abort', () => resolve(), { once: true });
+                      }),
+                  })
+                : 'Cached response unavailable';
+            return new Response(body, { status: 500 });
+          }
+          return response;
+        },
+      );
+      let release: (() => void) | undefined;
+      const permit: Promise<void> = new Promise((resolve) => {
+        release = resolve;
+      });
+      const write = fixture.storage.writeObjectStream!.bind(fixture.storage);
+      fixture.storage.writeObjectStream = async (
+        params: ParsedDocumentWriteObjectStreamParams,
+      ): Promise<ParsedDocumentWriteObjectResult> => {
+        // Models a backpressured adapter waiting for another document's capacity.
+        if (params.path.startsWith('file-')) {
+          await Promise.race([
+            permit,
+            new Promise<never>((_resolve, reject) => {
+              if (params.signal?.aborted)
+                reject(
+                  new Error('Storage capacity wait cancelled', { cause: params.signal.reason }),
+                );
+              else
+                params.signal?.addEventListener(
+                  'abort',
+                  () =>
+                    reject(
+                      new Error('Storage capacity wait cancelled', {
+                        cause: params.signal?.reason,
+                      }),
+                    ),
+                  { once: true },
+                );
+            }),
+          ]);
+        }
+        return write(params);
+      };
+      const controller: AbortController = new AbortController();
+      const outcome: Promise<string> = synchronize(controller.signal).then(
+        (): string => 'completed',
+        (): string => 'rejected',
+      );
+      try {
+        await failedResponse;
+        expect(
+          await Promise.race([
+            outcome,
+            new Promise<string>((resolve) => setTimeout(() => resolve('pending'), 250)),
+          ]),
+        ).toBe('rejected');
+        expect(fixture.progress.at(-1)?.archiveCacheBypassed === true).toBe(
+          failure !== 'origin-http',
+        );
+        expect(fixture.objects.has('.knowhere-sdk/commit.json')).toBe(false);
+      } finally {
+        release?.();
+        controller.abort(new Error('Release the backpressured fixture'));
+        await outcome;
+        configure({ maxWorkers: originalWorkerCount });
+      }
+      const writesAtReturn: number = fixture.finished.length;
+      await new Promise<void>((resolve) => setTimeout(resolve, 20));
+      expect(fixture.finished).toHaveLength(writesAtReturn);
+      isInvalid = false;
+      expect(await synchronize()).toMatchObject({ completed: true });
+    },
+  );
+
   it('interrupts a decoder awaiting output cleanup when the caller cancels', async () => {
     const fixture: Fixture = await createFixture();
     const controller: AbortController = new AbortController();
diff --git a/src/storage/streamed-parsed-document-sync.ts b/src/storage/streamed-parsed-document-sync.ts
index c5254be..a88201f 100644
--- a/src/storage/streamed-parsed-document-sync.ts
+++ b/src/storage/streamed-parsed-document-sync.ts
@@ -38,6 +38,7 @@ interface RangeEntryRead {
 }
 
 export interface ArchiveSource {
+  readonly signal?: AbortSignal;
   readonly canSkipEntries: boolean;
   readonly identity: string;
   readonly downloadIdentity: string;
@@ -75,9 +76,9 @@ export async function syncRemoteResultArchive(
   const signal: AbortSignal = params.signal
     ? AbortSignal.any([params.signal, controller.signal])
     : controller.signal;
-  const transferParams: StreamedSyncParams = { ...params, signal };
+  const openingParams: StreamedSyncParams = { ...params, signal };
   const previous: ParsedDocumentSyncProgress | null = await params.storage.readSyncProgress(params);
-  const openedSource: ArchiveSource = await openArchive(transferParams, previous);
+  const openedSource: ArchiveSource = await openArchive(openingParams, 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 =
@@ -86,6 +87,12 @@ export async function syncRemoteResultArchive(
     previous.archiveDownloadIdentity === openedSource.downloadIdentity
       ? { ...openedSource, identity: previous.archiveIdentity }
       : openedSource;
+  // A transport error must also interrupt sinks that are waiting for capacity,
+  // before they consume the entry stream and can observe its failure.
+  const transferParams: StreamedSyncParams = {
+    ...openingParams,
+    signal: source.signal ?? signal,
+  };
   const archiveDownloadIdentity: string | undefined = source.isCached
     ? source.downloadIdentity
     : previous?.archiveDownloadIdentity;
@@ -475,23 +482,34 @@ async function openArchiveTransport(
     // Cached reads already pin a strong ETag and exact lengths; CDN timestamps can drift.
     checkResourceChanges: !prepared?.isCached,
     fetch: async (input, init): Promise<Response> => {
-      let response: Response;
       try {
-        response = await fetch(input, { ...init, signal: rangeSignal });
-      } catch (error) {
-        rangeSignal.throwIfAborted();
-        if (prepared?.isCached)
-          throw new ArchiveCacheTransportError(
-            'Parsed result archive cache transport unavailable.',
+        let response: Response;
+        try {
+          response = await fetch(input, { ...init, signal: rangeSignal });
+        } catch (error) {
+          rangeSignal.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) {
+          const error: Error = new Error(
+            `Parsed result Range request failed (HTTP ${response.status}).`,
           );
+          rangeController.abort(error);
+          await response.body?.cancel(error).catch((): void => undefined);
+          throw error;
+        }
+        return prepared ? prepared.validateResponse(response) : response;
+      } catch (error) {
+        // zip.js can open a response before a decoder has started reading it.
+        // Abort here rather than waiting for the reader's pull to fail.
+        rangeController.abort(error);
         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 prepared ? prepared.validateResponse(response) : response;
     },
   };
   const reader = new ZipReader(new AbortableHttpRangeReader(resultUrl, options, rangeController), {
@@ -513,6 +531,7 @@ async function openArchiveTransport(
       )
       .digest('hex');
     return {
+      signal: rangeSignal,
       identity: previous?.archiveIdentity === identity ? identity : archiveIdentity,
       downloadIdentity: identity,
       archiveCacheBypassed,

From e367de113b2d79ee9a344e174e06bfaa18646833 Mon Sep 17 00:00:00 2001
From: suguanYang <wangbinqi77@gmail.com>
Date: Tue, 27 Oct 2026 08:21:54 -0700
Subject: [PATCH 3/3] fix: tag beta releases with the checked-out source commit

---
 .changeset/fix-range-cancellation.md        |  2 +
 scripts/publish-beta-release.mjs            |  7 +-
 src/__tests__/beta-release-contract.test.ts | 81 +++++++++++++++++++++
 3 files changed, 87 insertions(+), 3 deletions(-)
 create mode 100644 src/__tests__/beta-release-contract.test.ts

diff --git a/.changeset/fix-range-cancellation.md b/.changeset/fix-range-cancellation.md
index f41a01c..002cb02 100644
--- a/.changeset/fix-range-cancellation.md
+++ b/.changeset/fix-range-cancellation.md
@@ -5,3 +5,5 @@
 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.
diff --git a/scripts/publish-beta-release.mjs b/scripts/publish-beta-release.mjs
index 106020d..4edc8bb 100644
--- a/scripts/publish-beta-release.mjs
+++ b/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',
 ]);
diff --git a/src/__tests__/beta-release-contract.test.ts b/src/__tests__/beta-release-contract.test.ts
new file mode 100644
index 0000000..c8aefe4
--- /dev/null
+++ b/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 });
+  }
+});