diff --git a/.changeset/fix-range-cancellation.md b/.changeset/fix-range-cancellation.md
new file mode 100644
index 0000000..8c5d7ed
--- /dev/null
+++ b/.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.
diff --git a/docs/streamed-parsed-document-sync.md b/docs/streamed-parsed-document-sync.md
index 86e168c..c620b72 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/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 });
+  }
+});
diff --git a/src/knowledge/__tests__/progressive-snapshots.test.ts b/src/knowledge/__tests__/progressive-snapshots.test.ts
index 173ced2..c0f31b6 100644
--- a/src/knowledge/__tests__/progressive-snapshots.test.ts
+++ b/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' });
diff --git a/src/knowledge/knowledge.ts b/src/knowledge/knowledge.ts
index 43475ca..9cbd4f2 100644
--- a/src/knowledge/knowledge.ts
+++ b/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;
   }
 
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 2c09691..0680500 100644
--- a/src/storage/__tests__/streamed-sync-concurrency.test.ts
+++ b/src/storage/__tests__/streamed-sync-concurrency.test.ts
@@ -1,6 +1,8 @@
+import type { ReadableStreamReadResult } from 'node:stream/web';
 import { createHash } from 'crypto';
 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';
@@ -16,6 +18,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[];
@@ -29,6 +32,9 @@ interface Fixture {
 
 async function createFixture(
   options: {
+    readonly entropyBytes?: number;
+    readonly fileCount?: number;
+    readonly largeEntries?: boolean;
     readonly sequential?: boolean;
     readonly originSequential?: boolean;
     readonly cache?: boolean;
@@ -41,7 +47,15 @@ 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.entropyBytes
+        ? randomBytes(options.entropyBytes)
+        : options.largeEntries
+          ? `body-${index}`.repeat(262144)
+          : `body-${index}`,
+    );
   const archive: Uint8Array = await zip.generateAsync({
     type: 'uint8array',
     compression: 'DEFLATE',
@@ -136,6 +150,7 @@ async function createFixture(
       Promise.resolve('https://cache.example/result.zip');
   }
   return {
+    archive,
     storage,
     objects,
     progress,
@@ -294,6 +309,357 @@ 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.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();
+    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 11de26a..c68bfd9 100644
--- a/src/storage/streamed-parsed-document-sync.ts
+++ b/src/storage/streamed-parsed-document-sync.ts
@@ -1,9 +1,16 @@
+import type { ReadableStreamReadResult } from 'node:stream/web';
 import { createHash } from 'crypto';
 import type { JobSnapshot } from '../types/job.js';
 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 {
@@ -33,6 +40,7 @@ interface RangeEntryRead {
 }
 
 export interface ArchiveSource {
+  readonly signal?: AbortSignal;
   readonly readArchive?: () => ReadableStream<Uint8Array>;
   readonly canSkipEntries: boolean;
   readonly identity: string;
@@ -73,9 +81,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 =
@@ -84,6 +92,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;
@@ -431,6 +445,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,
@@ -462,32 +528,48 @@ 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,
     // 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: params.signal });
-      } catch (error) {
-        params.signal?.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 HttpRangeReader(resultUrl, options), {
+  const reader = new ZipReader(new AbortableHttpRangeReader(resultUrl, options, rangeController), {
     useWebWorkers: false,
     checkCrc32: true,
   });
@@ -506,6 +588,7 @@ async function openArchiveTransport(
       )
       .digest('hex');
     return {
+      signal: rangeSignal,
       identity: previous?.archiveIdentity === identity ? identity : archiveIdentity,
       downloadIdentity: identity,
       archiveCacheBypassed,
@@ -528,7 +611,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(),
@@ -542,6 +625,7 @@ async function openArchiveTransport(
       },
     };
   } catch (error) {
+    rangeController.abort(error);
     await reader.close();
     throw error;
   }
@@ -558,6 +642,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 })
@@ -568,6 +658,7 @@ function readRangeEntry(
       streamController?.error(error);
     })
     .finally((): void => {
+      signal?.removeEventListener('abort', cancelOutput);
       producers.delete(producer);
     });
   producers.add(producer);