Scope batchAppend promise bank per stream

Author: w1am

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,22 @@ 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) return;
+
+          const [resolve, reject] = entry;
           promiseBank.delete(resultingId);
 
           if (resp.hasError()) {
@@ -120,10 +130,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,40 @@ 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]
+        );
+
+        const bAppend = clientB.appendToStream(
+          "sibling_b_during_a_error",
+          jsonTestEvents()
+        );
+
+        await new Promise((r) => setImmediate(r));
+
+        aStream.emit("error", new Error("simulated transport error"));
+
+        const bResult = await bAppend;
+        expect(bResult.success).toBe(true);
+      } finally {
+        await clientA.dispose();
+        await clientB.dispose();
+      }
+    });
   });
 });