src/document-workbench/doc0-upload-client.ts

@@ -21,6 +21,16 @@ export interface Doc0UploadReceipt {
   pipelineStage: string;
 }
 
+/** The POST may have committed even if its response could not be verified. */
+export class Doc0UploadOutcomeUnknownError extends Error {
+  constructor() {
+    super('DOC0 upload outcome or receipt could not be verified; reconcile using the original idempotency key');
+    this.name = 'Doc0UploadOutcomeUnknownError';
+  }
+}
+
+const MAX_RECEIPT_BYTES = 128 * 1024;
+
 /**
  * Internal client only. The caller must enforce verified tenant/project/actor
  * permissions and data handling policy. The coordinated DOC0/N0D/ST0RE
@@ -87,21 +97,45 @@ export class Doc0UploadClient {
       }),
     );
     body.set('file', new Blob([new Uint8Array(input.bytes)], { type: input.mimeType }), input.filename);
-    const response = await this.fetchImpl(`${this.baseUrl}/v1/doc0/documents`, {
-      method: 'POST',
-      redirect: 'error',
-      headers: {
-        authorization: `Bearer ${this.options.serviceToken}`,
-        'x-tenant-id': input.tenantId,
-        'x-actor-id': input.actorId,
-        'x-request-id': input.requestId,
-        'x-idempotency-key': input.idempotencyKey,
-      },
-      body,
-      signal: AbortSignal.timeout(this.options.timeoutMs ?? 120_000),
-    });
-    if (!response.ok) throw new Error(`DOC0 upload failed: HTTP ${response.status}`);
-    const data = (await response.json()) as {
+    let response: Response;
+    try {
+      response = await this.fetchImpl(`${this.baseUrl}/v1/doc0/documents`, {
+        method: 'POST',
+        redirect: 'error',
+        headers: {
+          authorization: `Bearer ${this.options.serviceToken}`,
+          'x-tenant-id': input.tenantId,
+          'x-actor-id': input.actorId,
+          'x-request-id': input.requestId,
+          'x-idempotency-key': input.idempotencyKey,
+        },
+        body,
+        signal: AbortSignal.timeout(this.options.timeoutMs ?? 120_000),
+      });
+    } catch { throw new Doc0UploadOutcomeUnknownError(); }
+    // Even a non-2xx response does not prove the write did not commit.
+    if (!response.ok || !response.body) throw new Doc0UploadOutcomeUnknownError();
+    let decoded: unknown;
+    let reader: ReadableStreamDefaultReader<Uint8Array>;
+    try { reader = response.body.getReader(); }
+    catch { throw new Doc0UploadOutcomeUnknownError(); }
+    try {
+      const parts: Uint8Array[] = [];
+      let length = 0;
+      for (;;) {
+        const chunk = await reader.read();
+        if (chunk.done) break;
+        length += chunk.value.byteLength;
+        if (length > MAX_RECEIPT_BYTES) {
+          void reader.cancel().catch(() => undefined);
+          throw new Error('receipt too large');
+        }
+        parts.push(chunk.value);
+      }
+      decoded = JSON.parse(new TextDecoder('utf-8', { fatal: true }).decode(Buffer.concat(parts))) as unknown;
+    } catch { throw new Doc0UploadOutcomeUnknownError(); }
+    finally { reader.releaseLock(); }
+    const data = decoded as {
       document?: {
         documentId?: unknown;
         tenantId?: unknown;
@@ -117,8 +151,8 @@ export class Doc0UploadClient {
         pipelineStage?: unknown;
       };
     };
-    const doc = data.document;
-    const version = data.currentVersion;
+    const doc = data?.document;
+    const version = data?.currentVersion;
     if (
       !doc ||
       !version ||
@@ -136,7 +170,7 @@ export class Doc0UploadClient {
       typeof version.pipelineStage !== 'string' ||
       !version.pipelineStage
     ) {
-      throw new Error('DOC0 upload receipt does not match the submitted document');
+      throw new Doc0UploadOutcomeUnknownError();
     }
     return {
       documentId: doc.documentId,

tests/document-workbench/doc0-upload-client.test.ts

@@ -1,6 +1,6 @@
 import { createHash } from 'node:crypto';
 import { describe, expect, it, vi } from 'vitest';
-import { Doc0UploadClient } from '../../src/document-workbench/doc0-upload-client';
+import { Doc0UploadClient, Doc0UploadOutcomeUnknownError } from '../../src/document-workbench/doc0-upload-client';
 
 const bytes = new TextEncoder().encode('source bytes');
 const input = {
@@ -85,4 +85,22 @@ describe('DOC0 upload client', () => {
     expect(() => new Doc0UploadClient({ ...base, baseUrl: credentialUrl.toString() })).toThrow(/HTTPS/);
     expect(() => new Doc0UploadClient({ ...base, baseUrl: 'https://doc0.test', timeoutMs: 0 })).toThrow(/limits/);
   });
+  it('treats transport failure and non-success responses as unknown outcomes without retrying', async () => {
+    for (const response of [new Error('connection reset'), new Response('unavailable', { status: 503 })]) {
+      const fetchImpl = vi.fn(async () => { if (response instanceof Error) throw response; return response; });
+      const client = new Doc0UploadClient({ baseUrl: 'https://doc0.test', serviceToken: 'test', maxBytes: 100, fetchImpl });
+      await expect(client.upload(input)).rejects.toBeInstanceOf(Doc0UploadOutcomeUnknownError);
+      expect(fetchImpl).toHaveBeenCalledOnce();
+    }
+  });
+  it('rejects broken, malformed and oversized receipts as unknown outcomes', async () => {
+    const broken = new ReadableStream<Uint8Array>({ start(controller) { controller.error(new Error('broken')); } });
+    const locked = new Response('{}');
+    const lock = locked.body!.getReader();
+    for (const response of [new Response(broken), new Response('{'), new Response('x'.repeat(128 * 1024 + 1)), locked]) {
+      const client = new Doc0UploadClient({ baseUrl: 'https://doc0.test', serviceToken: 'test', maxBytes: 100, fetchImpl: async () => response });
+      await expect(client.upload(input)).rejects.toBeInstanceOf(Doc0UploadOutcomeUnknownError);
+    }
+    lock.releaseLock();
+  });
 });