From fe3b28ebaba3dcb3f78df6af28ef8da9aca32396 Mon Sep 17 00:00:00 2001
From: Nicolae Vartolomei <nv@redpanda.com>
Date: Mon, 28 Jul 2025 15:53:59 +0100
Subject: [PATCH] Fixed hang in blob operations when a client disconnects
 before the OperationQueue processes the request

In Azurite, all operations are managed through concurrent operation
queues. A bug was identified where operations could hang indefinitely if
the client disconnected before the operation was processed.

This occurred because Azurite would attempt to attach event handlers to
the request's readable stream (body) after it had already been closed by
the client's disconnection. Since a closed stream emits no further
events (like 'data', 'close', or 'error'), the operation would never
complete, causing a permanent hang.

This fix addresses the issue by checking if the request stream is still
readable before attaching any event handlers. This ensures that we only
process requests that are still active, preventing the hang and allowing
the queues to continue processing other operations.
---
 ChangeLog.md                            |  1 +
 src/common/persistence/FSExtentStore.ts | 12 ++++++++++++
 tests/blob/fsStore.test.ts              | 21 +++++++++++++++++++++
 3 files changed, 34 insertions(+)

diff --git a/ChangeLog.md b/ChangeLog.md
index 112b2cc6b..d46175e13 100644
--- a/ChangeLog.md
+++ b/ChangeLog.md
@@ -12,6 +12,7 @@ Blob:
 - Added support for sealing append blobs. (issue #810)
 - Added support for delegation sas with version of 2015-07-05.
 - Fix issue on SQL: Delete a container with blob, then create container/blob with same name, and delete container will fail. (issue #2563)
+- Fixed hang in blob operations when a client disconnects before the OperationQueue processes the request. (issue #2575)
 
 Table:
 
diff --git a/src/common/persistence/FSExtentStore.ts b/src/common/persistence/FSExtentStore.ts
index 31eedb31a..f2521b13f 100644
--- a/src/common/persistence/FSExtentStore.ts
+++ b/src/common/persistence/FSExtentStore.ts
@@ -518,6 +518,18 @@ export default class FSExtentStore implements IExtentStore {
       let count: number = 0;
       let wsEnd = false;
 
+      if (!rs.readable) {
+        this.logger.debug(
+          `FSExtentStore:streamPipe() Readable stream is not readable, rejecting streamPipe.`,
+          contextId
+        );
+        reject(
+          new Error(
+            `FSExtentStore:streamPipe() Readable stream is not readable.`
+          ));
+        return;
+      }
+
       rs.on("data", data => {
         count += data.length;
         if (!ws.write(data)) {
diff --git a/tests/blob/fsStore.test.ts b/tests/blob/fsStore.test.ts
index b0aee39bb..a912f2d3c 100644
--- a/tests/blob/fsStore.test.ts
+++ b/tests/blob/fsStore.test.ts
@@ -47,4 +47,25 @@ describe("FSExtentStore", () => {
     let readable3 = await store.readExtent(extent3);
     assert.strictEqual(await readIntoString(readable3), "Test");
   });
+
+  it("should handle garbage collected input stream during appendExtent @loki", async () => {
+    const store = new FSExtentStore(metadataStore, DEFAULT_BLOB_PERSISTENCE_ARRAY, logger);
+    await store.init();
+
+    const stream1 = Readable.from("Test", { objectMode: false });
+
+    // From manual testing express.js it seems that if the request is aborted
+    // before it is handled/listeners are set up, the stream is destroyed.
+    // This simulates that behavior.
+    stream1.destroy();
+
+    // Then we check that appendExtent handles the destroyed stream
+    // gracefully/does not hang.
+    try {
+      await store.appendExtent(stream1);
+      assert.fail("Expected an error to be thrown due to destroyed stream");
+    } catch (err) {
+      assert.deepStrictEqual(err.message, "FSExtentStore:streamPipe() Readable stream is not readable.");
+    }
+  });
 });