fix: consumed body on retry

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,
         });

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 };

test/storage/object/copy.test.js

@@ -598,6 +598,104 @@ 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('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' },

test/storage/version/put.test.js

@@ -12,9 +12,17 @@
 /* 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';
+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) => {
@@ -469,7 +477,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 +534,6 @@ describe('Version Put', () => {
     assert.equal(10, s3INMSent[0].input.ContentLength);
 
     assert.equal(1, s3Sent.length);
-    assert(s3Sent[0].input.Body instanceof ReadableStream);
     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 +2270,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 +2308,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 +2969,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',