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.

Author: nvartolomei

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:
 

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)) {

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.");
+    }
+  });
 });