[DEV-1818] Fix batchAppend cross-stream error contamination

packages/db-client/src/streams/appendToStream/batchAppend.ts

@@ -30,14 +30,19 @@ import {
 
 import type { AppendToStreamOptions } from ".";
 
+type PromiseBank = Map<
+  string,
+  [resolve: (r: AppendResult) => void, reject: (error: Error) => void]
+>;
+
 const streamCache = new WeakMap<
   StreamsClient,
   Promise<ReturnType<StreamsClient["batchAppend"]>>
 >();
 
-const promiseBank = new Map<
-  string,
-  [resolve: (r: AppendResult) => void, reject: (error: Error) => void]
+const promiseBanks = new WeakMap<
+  ReturnType<StreamsClient["batchAppend"]>,
+  PromiseBank
 >();
 
 export const batchAppend = async function (
@@ -55,17 +60,28 @@ export const batchAppend = async function (
   const stream = await this.GRPCStreamCreator(
     StreamsClient,
     "appendToStream",
-    (client) =>
-      client
+    (client) => {
+      const promiseBank: PromiseBank = new Map();
+
+      const batchStream = client
         .batchAppend(
           ...this.callArguments(baseOptions, {
             deadline: Infinity,
           })
         )
         .on("data", (resp: BatchAppendResp) => {
           const resultingId = parseUUID(resp.getCorrelationId()!);
-          const [resolve, reject] = promiseBank.get(resultingId)!;
+          const entry = promiseBank.get(resultingId);
+
+          if (!entry) {
+            debug.command_grpc(
+              "batchAppend: dropping response for unknown correlationId %s",
+              resultingId
+            );
+            return;
+          }
 
+          const [resolve, reject] = entry;
           promiseBank.delete(resultingId);
 
           if (resp.hasError()) {
@@ -120,10 +136,22 @@ export const batchAppend = async function (
             reject(convertToCommandError(error));
           }
           promiseBank.clear();
-        }),
+        });
+
+      promiseBanks.set(batchStream, promiseBank);
+
+      return batchStream;
+    },
     streamCache
   )();
 
+  const promiseBank = promiseBanks.get(stream);
+  if (!promiseBank) {
+    throw new Error(
+      "batchAppend could not find the promise bank for the stream."
+    );
+  }
+
   return new Promise(async (...batchPromise) => {
     promiseBank.set(correlationId, batchPromise);
 

packages/test/src/streams/appendToStream-batch-append.test.ts

@@ -111,6 +111,53 @@ describe("appendToStream - batch append", () => {
         (128 * 5_000) / 1024
       );
     });
+
+    test("A stream error does not reject in-flight appends on sibling streams", async () => {
+      const clientA = KurrentDBClient.connectionString(node.connectionString());
+      const clientB = KurrentDBClient.connectionString(node.connectionString());
+      const aSpy = jest.spyOn(
+        clientA,
+        "GRPCStreamCreator" as never
+      ) as unknown as jest.SpiedFunction<KurrentDBClient["GRPCStreamCreator"]>;
+
+      try {
+        await clientA.appendToStream("sibling_a_warmup", jsonTestEvents());
+        await clientB.appendToStream("sibling_b_warmup", jsonTestEvents());
+
+        const aStream = await extractBatchStream.call(
+          clientA,
+          ...aSpy.mock.calls[0]
+        );
+
+        let aSettled = false;
+        let bSettled = false;
+        const aAppend = clientA
+          .appendToStream("sibling_a_during_a_error", jsonTestEvents())
+          .finally(() => {
+            aSettled = true;
+          });
+        const bAppend = clientB
+          .appendToStream("sibling_b_during_a_error", jsonTestEvents())
+          .finally(() => {
+            bSettled = true;
+          });
+
+        await new Promise((r) => setImmediate(r));
+
+        expect(aSettled).toBe(false);
+        expect(bSettled).toBe(false);
+
+        aStream.emit("error", new Error("simulated transport error"));
+
+        await expect(aAppend).rejects.toThrow();
+
+        const bResult = await bAppend;
+        expect(bResult.success).toBe(true);
+      } finally {
+        await clientA.dispose();
+        await clientB.dispose();
+      }
+    });
   });
 });