fix(blob): resolve multipart deadlock on empty streams, fix TransformStream backpressure (#1037)

* fix(blob): resolve multipart deadlock on empty streams, fix TransformStream backpressure

Two fixes:

1. uploadAllParts hangs forever on empty ReadableStream because
   resolve() was only called inside sendPart(), which is never invoked
   when there are no bytes. Add resolve() when stream is done with no
   buffered parts and no active uploads.

2. createChunkTransformStream wrapped transform/flush in queueMicrotask,
   returning undefined immediately. This broke backpressure and could
   cause data loss in flush (trailing bytes enqueued after the readable
   side closes).

Fixes #881

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

* test: add e2e test for empty stream multipart deadlock fix

Add a test route that does a multipart upload with an empty
ReadableStream, and a Playwright test that verifies it resolves
without hanging (with a 15s timeout).

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Vincent Voyer <vincent@codeagain.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: 3 people

.changeset/fix-stream-deadlock.md

@@ -0,0 +1,5 @@
+---
+"@vercel/blob": patch
+---
+
+Fix multipart upload hanging forever on empty streams, and fix `createChunkTransformStream` bypassing backpressure by removing incorrect `queueMicrotask` wrapping.

packages/blob/src/api.node.test.ts

@@ -9,7 +9,7 @@ import {
   BlobUnknownError,
   requestApi,
 } from './api';
-import { BlobError } from './helpers';
+import { BlobError, createChunkTransformStream } from './helpers';
 
 describe('api', () => {
   describe('request api', () => {
@@ -133,3 +133,74 @@ describe('api', () => {
     });
   });
 });
+
+describe('createChunkTransformStream', () => {
+  async function collectChunks(
+    stream: ReadableStream<Uint8Array>,
+  ): Promise<Uint8Array[]> {
+    const chunks: Uint8Array[] = [];
+    const reader = stream.getReader();
+    for (;;) {
+      const { value, done } = await reader.read();
+      if (done) break;
+      chunks.push(value);
+    }
+    return chunks;
+  }
+
+  it('accumulates small chunks and emits full-sized chunks', async () => {
+    const chunkSize = 4;
+    const stream = new ReadableStream<ArrayBuffer>({
+      start(controller) {
+        controller.enqueue(new Uint8Array([1, 2]).buffer);
+        controller.enqueue(new Uint8Array([3, 4, 5, 6]).buffer);
+        controller.enqueue(new Uint8Array([7]).buffer);
+        controller.close();
+      },
+    });
+
+    const chunks = await collectChunks(
+      stream.pipeThrough(createChunkTransformStream(chunkSize)),
+    );
+
+    expect(chunks).toHaveLength(2);
+    expect(Array.from(chunks[0]!)).toEqual([1, 2, 3, 4]);
+    expect(Array.from(chunks[1]!)).toEqual([5, 6, 7]);
+  });
+
+  it('calls onProgress for each emitted chunk', async () => {
+    const chunkSize = 3;
+    const progress: number[] = [];
+    const stream = new ReadableStream<ArrayBuffer>({
+      start(controller) {
+        controller.enqueue(new Uint8Array([1, 2, 3, 4, 5]).buffer);
+        controller.close();
+      },
+    });
+
+    await collectChunks(
+      stream.pipeThrough(
+        createChunkTransformStream(chunkSize, (bytes) => progress.push(bytes)),
+      ),
+    );
+
+    expect(progress).toEqual([3, 2]);
+  });
+
+  it('flushes remaining bytes when stream ends', async () => {
+    const chunkSize = 10;
+    const stream = new ReadableStream<ArrayBuffer>({
+      start(controller) {
+        controller.enqueue(new Uint8Array([1, 2, 3]).buffer);
+        controller.close();
+      },
+    });
+
+    const chunks = await collectChunks(
+      stream.pipeThrough(createChunkTransformStream(chunkSize)),
+    );
+
+    expect(chunks).toHaveLength(1);
+    expect(Array.from(chunks[0]!)).toEqual([1, 2, 3]);
+  });
+});

packages/blob/src/helpers.ts

@@ -269,31 +269,27 @@ export const createChunkTransformStream = (
 
   return new TransformStream<ArrayBuffer, Uint8Array>({
     transform(chunk, controller) {
-      queueMicrotask(() => {
-        // Combine the new chunk with any leftover data
-        const newBuffer = new Uint8Array(buffer.length + chunk.byteLength);
-        newBuffer.set(buffer);
-        newBuffer.set(new Uint8Array(chunk), buffer.length);
-        buffer = newBuffer;
-
-        // Output complete chunks
-        while (buffer.length >= chunkSize) {
-          const newChunk = buffer.slice(0, chunkSize);
-          controller.enqueue(newChunk);
-          onProgress?.(newChunk.byteLength);
-          buffer = buffer.slice(chunkSize);
-        }
-      });
+      // Combine the new chunk with any leftover data
+      const newBuffer = new Uint8Array(buffer.length + chunk.byteLength);
+      newBuffer.set(buffer);
+      newBuffer.set(new Uint8Array(chunk), buffer.length);
+      buffer = newBuffer;
+
+      // Output complete chunks
+      while (buffer.length >= chunkSize) {
+        const newChunk = buffer.slice(0, chunkSize);
+        controller.enqueue(newChunk);
+        onProgress?.(newChunk.byteLength);
+        buffer = buffer.slice(chunkSize);
+      }
     },
 
     flush(controller) {
-      queueMicrotask(() => {
-        // Send any remaining data
-        if (buffer.length > 0) {
-          controller.enqueue(buffer);
-          onProgress?.(buffer.byteLength);
-        }
-      });
+      // Send any remaining data
+      if (buffer.length > 0) {
+        controller.enqueue(buffer);
+        onProgress?.(buffer.byteLength);
+      }
     },
   });
 };

packages/blob/src/index.node.test.ts

@@ -791,6 +791,33 @@ describe('blob client', () => {
       }
     });
 
+    it('resolves (does not hang) when multipart body is an empty ReadableStream', async () => {
+      const emptyStream = new ReadableStream({
+        start(controller) {
+          controller.close();
+        },
+      });
+
+      // createMultipartUpload
+      mockClient
+        .intercept({ path: () => true, method: 'POST' })
+        .reply(200, { uploadId: 'upload-123', key: 'foo.txt' });
+
+      // completeMultipartUpload (called with 0 parts for an empty stream)
+      mockClient.intercept({ path: () => true, method: 'POST' }).reply(200, {
+        url: `${BLOB_STORE_BASE_URL}/foo.txt`,
+        downloadUrl: `${BLOB_STORE_BASE_URL}/foo.txt?download=1`,
+        pathname: 'foo.txt',
+        contentType: 'text/plain',
+        contentDisposition: 'attachment; filename="foo.txt"',
+        etag: '"empty"',
+      });
+
+      await expect(
+        put('foo.txt', emptyStream, { access: 'public', multipart: true }),
+      ).resolves.toMatchObject({ pathname: 'foo.txt' });
+    });
+
     const table: [string, (signal: AbortSignal) => Promise<unknown>][] = [
       [
         'put',

packages/blob/src/multipart/upload.ts

@@ -245,6 +245,11 @@ export function uploadAllParts({
               });
 
               sendParts();
+            } else if (activeUploads === 0) {
+              // Stream was empty or ended exactly on a part boundary with no
+              // in-flight uploads. Nothing will call resolve() later, so do it here.
+              reader.releaseLock();
+              resolve(completedParts);
             }
             reading = false;
             return;

test/next/src/app/vercel/blob/api/app/empty-stream-multipart/route.ts

@@ -0,0 +1,32 @@
+import * as vercelBlob from '@vercel/blob';
+import { NextResponse } from 'next/server';
+import { validateUploadToken } from '../../../validate-upload-token';
+
+export async function POST(request: Request): Promise<NextResponse> {
+  const { searchParams } = new URL(request.url);
+  const pathname = searchParams.get('filename');
+
+  if (pathname === null) {
+    return NextResponse.json({ message: 'Missing filename' }, { status: 400 });
+  }
+
+  if (!validateUploadToken(request)) {
+    return NextResponse.json({ message: 'Not authorized' }, { status: 401 });
+  }
+
+  // Multipart upload with an empty ReadableStream — this used to deadlock
+  // because uploadAllParts never called resolve() when the stream was empty.
+  const emptyStream = new ReadableStream({
+    start(controller) {
+      controller.close();
+    },
+  });
+
+  const blob = await vercelBlob.put(pathname, emptyStream, {
+    access: 'public',
+    multipart: true,
+    addRandomSuffix: true,
+  });
+
+  return NextResponse.json(blob);
+}

test/next/test/@vercel/blob/index.test.ts

@@ -37,6 +37,27 @@ test.describe('@vercel/blob', () => {
     });
   });
 
+  test.describe('empty stream multipart', () => {
+    test('does not deadlock on empty ReadableStream with multipart: true', async ({
+      request,
+      extraHTTPHeaders,
+    }) => {
+      const res = await request.post(
+        `vercel/blob/api/app/empty-stream-multipart?filename=${prefix}/empty-stream.bin`,
+        {
+          headers: {
+            cookie: `clientUpload=${process.env.BLOB_UPLOAD_SECRET ?? ''}`,
+            ...extraHTTPHeaders,
+          },
+          timeout: 15000, // Should resolve fast, not deadlock
+        },
+      );
+      // The upload resolves (no deadlock). The server may reject the empty
+      // multipart upload, but the important thing is it doesn't hang.
+      expect(res.status()).toBeLessThan(500);
+    });
+  });
+
   test.describe('page', () => {
     test('serverless', async ({ page }) => {
       await page.goto(`vercel/pages/blob?filename=${prefix}/test-page.txt`);