diff --git a/packages/db-client/src/streams/appendToStream/batchAppend.ts b/packages/db-client/src/streams/appendToStream/batchAppend.ts index fd5a5734..5bb68a1f 100644 --- a/packages/db-client/src/streams/appendToStream/batchAppend.ts +++ b/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> >(); -const promiseBank = new Map< - string, - [resolve: (r: AppendResult) => void, reject: (error: Error) => void] +const promiseBanks = new WeakMap< + ReturnType, + PromiseBank >(); export const batchAppend = async function ( @@ -55,8 +60,10 @@ 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, @@ -64,8 +71,17 @@ export const batchAppend = async function ( ) .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); diff --git a/packages/test/src/streams/appendToStream-batch-append.test.ts b/packages/test/src/streams/appendToStream-batch-append.test.ts index 53834c74..a56fca31 100644 --- a/packages/test/src/streams/appendToStream-batch-append.test.ts +++ b/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; + + 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(); + } + }); }); });