From d9814723b3428cd05869b8fff9d954e37a9dea18 Mon Sep 17 00:00:00 2001 From: kptdobe Date: Wed, 29 Apr 2026 11:44:46 +0200 Subject: [PATCH 1/3] fix: consumed body on retry --- src/storage/object/copy.js | 6 ++- src/storage/version/put.js | 7 +++- test/storage/object/copy.test.js | 53 +++++++++++++++++++++++++ test/storage/version/put.test.js | 67 +++++++++++++++++++++++++++++--- 4 files changed, 126 insertions(+), 7 deletions(-) diff --git a/src/storage/object/copy.js b/src/storage/object/copy.js index cca9977b..98e73e6f 100644 --- a/src/storage/object/copy.js +++ b/src/storage/object/copy.js @@ -94,11 +94,15 @@ export const copyFile = async (config, env, daCtx, sourceKey, details, isRename) env, { bucket: daCtx.bucket, org: daCtx.org, key: sourceKey }, ); + // Buffer the ReadableStream so the body survives retries inside putObjectWithVersion. + const originalBody = original.body instanceof ReadableStream + ? await new Response(original.body).arrayBuffer() + : original.body; return /* await */ putObjectWithVersion(env, daCtx, { bucket: daCtx.bucket, org: daCtx.org, key: Key, - body: original.body, + body: originalBody, contentLength: original.contentLength, type: original.contentType, }); diff --git a/src/storage/version/put.js b/src/storage/version/put.js index 765146be..ab556dcb 100644 --- a/src/storage/version/put.js +++ b/src/storage/version/put.js @@ -35,6 +35,8 @@ export function getContentLength(body) { return new Blob([body]).size; } else if (body instanceof File) { return body.size; + } else if (body instanceof ArrayBuffer) { + return body.byteLength; } return undefined; } @@ -323,10 +325,13 @@ export async function putObjectWithVersion( export async function postObjectVersionWithLabel(label, env, daCtx) { const { body, contentLength, contentType } = await getObject(env, daCtx); + // Buffer the ReadableStream so the body survives retries inside putObjectWithVersion. + // A ReadableStream can only be consumed once; ArrayBuffer can be reused freely. + const bodyBuffer = body instanceof ReadableStream ? await new Response(body).arrayBuffer() : body; const { bucket, org, key } = daCtx; const resp = await putObjectWithVersion(env, daCtx, { - bucket, org, key, body, contentLength, type: contentType, label, + bucket, org, key, body: bodyBuffer, contentLength, type: contentType, label, }, true); if (resp.status !== 200) return { status: resp.status }; diff --git a/test/storage/object/copy.test.js b/test/storage/object/copy.test.js index 335fb345..a3982ff0 100644 --- a/test/storage/object/copy.test.js +++ b/test/storage/object/copy.test.js @@ -598,6 +598,59 @@ describe('Object copy', () => { assert.strictEqual(puwv[0].u.type, 'text/html'); }); + it('buffers ReadableStream body to ArrayBuffer before calling putObjectWithVersion (stream must survive retry)', async () => { + // Regression test for: ReadableStream disturbed on putObjectWithVersion retry. + // copyFile fetches original.body (a ReadableStream) and passes it to + // putObjectWithVersion. If the main PUT fails with 412 and retries, the stream + // is already consumed in the Cloudflare runtime ("disturbed") and the retry + // returns 500. The fix buffers the stream to ArrayBuffer before the call so + // the body can survive retries. + const error = { $metadata: { httpStatusCode: 412 } }; + + const mockS3Client = class { + // eslint-disable-next-line class-methods-use-this + send() { throw error; } + + middlewareStack = { add: () => {} }; + }; + + const mockGetObject = async (e, u, h) => { + if (u.key === 'xsrc/abc/def.html' && !h) { + return { + body: ReadableStream.from([new TextEncoder().encode('original body')]), + contentLength: 13, + contentType: 'text/html', + }; + } + }; + + const puwv = []; + const mockPutObjectWithVersion = async (e, c, u) => { + puwv.push({ e, c, u }); + return { status: 200 }; + }; + + // eslint-disable-next-line no-shadow + const { copyFile } = await esmock('../../../src/storage/object/copy.js', { + '../../../src/storage/object/get.js': { default: mockGetObject }, + '../../../src/storage/version/put.js': { putObjectWithVersion: mockPutObjectWithVersion }, + '@aws-sdk/client-s3': { S3Client: mockS3Client }, + }); + + const env = { dacollab: { fetch: () => ({ body: { cancel: () => {} } }) } }; + const daCtx = { bucket: 'mybucket', org: 'xorg' }; + daCtx.aclCtx = await getAclCtx(env, daCtx.org, daCtx.users, '/'); + const details = { source: 'xsrc', destination: 'xdst' }; + + await copyFile({}, env, daCtx, 'xsrc/abc/def.html', details, false); + + assert.strictEqual(puwv.length, 1); + // The body must be an ArrayBuffer so it survives retries inside putObjectWithVersion. + // A ReadableStream here means the stream was not buffered and would be disturbed on retry. + assert(puwv[0].u.body instanceof ArrayBuffer, 'body must be buffered to ArrayBuffer before putObjectWithVersion'); + assert.strictEqual(puwv[0].u.contentLength, 13); + }); + it('Copy content when origin does not exists', async () => { const error = { $metadata: { httpStatusCode: 404, hi: 'ha' }, diff --git a/test/storage/version/put.test.js b/test/storage/version/put.test.js index 09e86961..68a3dcfb 100644 --- a/test/storage/version/put.test.js +++ b/test/storage/version/put.test.js @@ -12,7 +12,7 @@ /* eslint-disable no-unused-vars,camelcase */ import assert from 'node:assert'; import esmock from 'esmock'; -import { PutObjectCommand } from '@aws-sdk/client-s3'; +import { PutObjectCommand, CopyObjectCommand } from '@aws-sdk/client-s3'; describe('Version Put', () => { it('Test putObjectWithVersion retry on new document', async () => { @@ -469,7 +469,7 @@ describe('Version Put', () => { // eslint-disable-next-line consistent-return const mockGetObject = async (e, u, h) => { if (e === env && !h) { - const body = ReadableStream.from('doccontent'); + const body = ReadableStream.from([new TextEncoder().encode('doccontent')]); return { body, contentType: 'text/html', @@ -526,7 +526,8 @@ describe('Version Put', () => { assert.equal(10, s3INMSent[0].input.ContentLength); assert.equal(1, s3Sent.length); - assert(s3Sent[0].input.Body instanceof ReadableStream); + // body is buffered to ArrayBuffer so it survives retries inside putObjectWithVersion + assert(s3Sent[0].input.Body instanceof ArrayBuffer); assert.equal('mybucket', s3Sent[0].input.Bucket); assert.equal('org123/q/r/t', s3Sent[0].input.Key); assert.equal('q/r/t', s3Sent[0].input.Metadata.Path); @@ -2263,7 +2264,7 @@ describe('Version Put', () => { }; const mockGetObject = async () => ({ - body: ReadableStream.from('doccontent'), + body: ReadableStream.from([new TextEncoder().encode('doccontent')]), contentType: 'text/html', contentLength: 10, metadata: { id: 'doc-id', version: 'ver-1' }, @@ -2301,7 +2302,7 @@ describe('Version Put', () => { }; const mockGetObject = async () => ({ - body: ReadableStream.from('doccontent'), + body: ReadableStream.from([new TextEncoder().encode('doccontent')]), contentType: 'text/html', contentLength: 10, metadata: { id: 'doc-id', version: 'ver-1' }, @@ -2962,6 +2963,62 @@ describe('Version Put', () => { }); describe('postObjectVersionWithLabel', () => { + it('returns 201 when main PUT 412s once then succeeds (ReadableStream body must survive retry)', async () => { + // Regression test for: ReadableStream disturbed on putObjectWithVersion retry. + // The real S3/R2 SDK consumes the request body before returning 412. When + // putObjectWithVersion retries with the same update.body ReadableStream, the + // stream is already disturbed, causing a TypeError and a 500 response. + // + // The fix buffers the stream to ArrayBuffer before the first PUT so the body + // survives retries. The mock enforces this by throwing when it sees a + // ReadableStream on the retry (simulating Cloudflare's "disturbed" error). + const req = { json: async () => ({ label: 'my-label' }) }; + const env = {}; + const ctx = { + bucket: 'mybucket', org: 'org123', key: 'doc.html', ext: 'html', users: [], + }; + + const mockGetObject = async () => ({ + body: ReadableStream.from([new TextEncoder().encode('doccontent')]), + contentType: 'text/html', + contentLength: 10, + status: 200, + metadata: { id: 'doc-id', version: 'v1' }, + }); + + let mainCallCount = 0; + const mainClient = { + async send(cmd) { + mainCallCount += 1; + if (mainCallCount === 1) { + const err = new Error('412'); + err.$metadata = { httpStatusCode: 412 }; + throw err; + } + // On retry: a ReadableStream body means it was not buffered — the real + // Cloudflare runtime would throw "disturbed" here. Enforce that invariant. + if (cmd.input.Body instanceof ReadableStream) { + throw new TypeError('This ReadableStream is disturbed (has already been read from), and cannot be used as a body.'); + } + return { $metadata: { httpStatusCode: 200 } }; + }, + }; + const versionClient = { + async send() { return { $metadata: { httpStatusCode: 200 } }; }, + }; + + const { postObjectVersion } = await esmock('../../../src/storage/version/put.js', { + '../../../src/storage/object/get.js': { default: mockGetObject }, + '../../../src/storage/utils/version.js': { + ifNoneMatch: () => versionClient, + ifMatch: () => mainClient, + }, + }); + + const resp = await postObjectVersion(req, env, ctx); + assert.equal(201, resp.status); + }); + it('returns 500 when versionCreated is false (version already exists / 412)', async () => { const mockGetObject = async () => ({ body: 'doc content', From 4981e2dfbe31ef4706ca45e880ba31999e2db431 Mon Sep 17 00:00:00 2001 From: kptdobe Date: Wed, 29 Apr 2026 12:19:02 +0200 Subject: [PATCH 2/3] test: add coverage for ArrayBuffer branch in getContentLength and non-ReadableStream body in copyFile Co-Authored-By: Claude Sonnet 4.6 --- test/storage/object/copy.test.js | 45 ++++++++++++++++++++++++++++++++ test/storage/version/put.test.js | 8 ++++++ 2 files changed, 53 insertions(+) diff --git a/test/storage/object/copy.test.js b/test/storage/object/copy.test.js index a3982ff0..f85fa0c9 100644 --- a/test/storage/object/copy.test.js +++ b/test/storage/object/copy.test.js @@ -651,6 +651,51 @@ describe('Object copy', () => { assert.strictEqual(puwv[0].u.contentLength, 13); }); + it('passes non-ReadableStream body through unchanged to putObjectWithVersion', async () => { + const error = { $metadata: { httpStatusCode: 412 } }; + + const mockS3Client = class { + // eslint-disable-next-line class-methods-use-this + send() { throw error; } + + middlewareStack = { add: () => {} }; + }; + + const preBuffered = new TextEncoder().encode('pre-buffered').buffer; + const mockGetObject = async (e, u, h) => { + if (u.key === 'xsrc/abc/def.html' && !h) { + return { + body: preBuffered, + contentLength: preBuffered.byteLength, + contentType: 'text/html', + }; + } + }; + + const puwv = []; + const mockPutObjectWithVersion = async (e, c, u) => { + puwv.push({ e, c, u }); + return { status: 200 }; + }; + + // eslint-disable-next-line no-shadow + const { copyFile } = await esmock('../../../src/storage/object/copy.js', { + '../../../src/storage/object/get.js': { default: mockGetObject }, + '../../../src/storage/version/put.js': { putObjectWithVersion: mockPutObjectWithVersion }, + '@aws-sdk/client-s3': { S3Client: mockS3Client }, + }); + + const env = { dacollab: { fetch: () => ({ body: { cancel: () => {} } }) } }; + const daCtx = { bucket: 'mybucket', org: 'xorg' }; + daCtx.aclCtx = await getAclCtx(env, daCtx.org, daCtx.users, '/'); + const details = { source: 'xsrc', destination: 'xdst' }; + + await copyFile({}, env, daCtx, 'xsrc/abc/def.html', details, false); + + assert.strictEqual(puwv.length, 1); + assert.strictEqual(puwv[0].u.body, preBuffered, 'non-ReadableStream body must be passed through unchanged'); + }); + it('Copy content when origin does not exists', async () => { const error = { $metadata: { httpStatusCode: 404, hi: 'ha' }, diff --git a/test/storage/version/put.test.js b/test/storage/version/put.test.js index 68a3dcfb..ffac24bf 100644 --- a/test/storage/version/put.test.js +++ b/test/storage/version/put.test.js @@ -13,8 +13,16 @@ import assert from 'node:assert'; import esmock from 'esmock'; import { PutObjectCommand, CopyObjectCommand } from '@aws-sdk/client-s3'; +import { getContentLength } from '../../../src/storage/version/put.js'; describe('Version Put', () => { + describe('getContentLength', () => { + it('returns byteLength for ArrayBuffer body', () => { + const buf = new ArrayBuffer(17); + assert.strictEqual(getContentLength(buf), 17); + }); + }); + it('Test putObjectWithVersion retry on new document', async () => { const getObjectCalls = []; const mockGetObject = async (e, u, nb) => { From 57a94d593e992a7f040a1e03bcf5e64160281b34 Mon Sep 17 00:00:00 2001 From: kptdobe Date: Wed, 29 Apr 2026 12:47:46 +0200 Subject: [PATCH 3/3] chore: remove assertion --- test/storage/version/put.test.js | 2 -- 1 file changed, 2 deletions(-) diff --git a/test/storage/version/put.test.js b/test/storage/version/put.test.js index ffac24bf..2e54efc9 100644 --- a/test/storage/version/put.test.js +++ b/test/storage/version/put.test.js @@ -534,8 +534,6 @@ describe('Version Put', () => { assert.equal(10, s3INMSent[0].input.ContentLength); assert.equal(1, s3Sent.length); - // body is buffered to ArrayBuffer so it survives retries inside putObjectWithVersion - assert(s3Sent[0].input.Body instanceof ArrayBuffer); assert.equal('mybucket', s3Sent[0].input.Bucket); assert.equal('org123/q/r/t', s3Sent[0].input.Key); assert.equal('q/r/t', s3Sent[0].input.Metadata.Path);