diff --git a/src/document-workbench/doc0-upload-client.ts b/src/document-workbench/doc0-upload-client.ts index f69b438..d90ce7b 100644 --- a/src/document-workbench/doc0-upload-client.ts +++ b/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; + 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, diff --git a/tests/document-workbench/doc0-upload-client.test.ts b/tests/document-workbench/doc0-upload-client.test.ts index a1230ba..8d978d0 100644 --- a/tests/document-workbench/doc0-upload-client.test.ts +++ b/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({ 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(); + }); });