From 2dd5827b40ca3b1ba727a6073c54288805564d86 Mon Sep 17 00:00:00 2001 From: Alp Karavil <46430499+Karavil@users.noreply.github.com> Date: Tue, 26 May 2026 14:34:54 -0400 Subject: [PATCH] =?UTF-8?q?=E2=9A=A1=20perf:=20negotiate=20cumulative=20st?= =?UTF-8?q?ream=20acks?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Why: reduce websocket ACK churn while keeping new clients compatible with old v6 stream servers during rollout. * advertise cumulative ACK capability before enabling cumulative sender semantics * make new clients fall back to per-message ACKs without the capability frame * add benchmark ACK mode controls for cumulative websocket runs --- .../change-streamer/change-streamer-http.ts | 18 +- packages/zero-cache/src/types/streams.test.ts | 113 ++++++- packages/zero-cache/src/types/streams.ts | 284 ++++++++++++++++-- 3 files changed, 369 insertions(+), 46 deletions(-) diff --git a/packages/zero-cache/src/services/change-streamer/change-streamer-http.ts b/packages/zero-cache/src/services/change-streamer/change-streamer-http.ts index f92f15da7c..3921ebd8b6 100644 --- a/packages/zero-cache/src/services/change-streamer/change-streamer-http.ts +++ b/packages/zero-cache/src/services/change-streamer/change-streamer-http.ts @@ -38,6 +38,8 @@ const SNAPSHOT_PATH_PATTERN = '/replication/:version/snapshot'; const CHANGES_PATH_PATTERN = '/replication/:version/changes'; const PATH_REGEX = /\/replication\/v(?\d+)\/(changes|snapshot)$/; const STREAM_BATCH_MESSAGES = 256; +const ACK_MODE_PARAM = 'ack'; +const CUMULATIVE_ACK_MODE = 'cumulative'; const SNAPSHOT_PATH = `/replication/v${PROTOCOL_VERSION}/snapshot`; const CHANGES_PATH = `/replication/v${PROTOCOL_VERSION}/changes`; @@ -148,7 +150,10 @@ export class ChangeStreamerHttpServer extends HttpService { // batching after parse/ACK. Older clients omit the flag and keep the // original one-message stream shape. const streamBatchRequested = url.searchParams.get('streamBatch') === '1'; + const cumulativeAckRequested = + url.searchParams.get(ACK_MODE_PARAM) === CUMULATIVE_ACK_MODE; void streamOutStringified(this._lc, downstream, ws, { + ack: cumulativeAckRequested ? 'cumulative' : undefined, batch: streamBatchRequested ? {maxMessages: STREAM_BATCH_MESSAGES} : undefined, @@ -243,7 +248,9 @@ export class ChangeStreamerHttpClient implements ChangeStreamer { async subscribe(ctx: SubscriberContext): Promise> { const ws = await this.#openChangesWebSocket(ctx); - return streamIn(this.#lc, ws, downstreamSchema); + return streamIn(this.#lc, ws, downstreamSchema, { + ack: 'cumulative-if-supported', + }); } async subscribeBatched( @@ -251,15 +258,20 @@ export class ChangeStreamerHttpClient implements ChangeStreamer { ): Promise>> { const ws = await this.#openChangesWebSocket(ctx); // Keep websocket batches visible to the incremental syncer so it can hand a - // whole received frame to the write worker before ACKing the frame messages. - return streamInBatches(this.#lc, ws, downstreamSchema); + // whole received frame to the write worker before sending the cumulative ACK. + return streamInBatches(this.#lc, ws, downstreamSchema, { + ack: 'cumulative-if-supported', + }); } async #openChangesWebSocket(ctx: SubscriberContext) { const uri = await this.#resolveChangeStreamer(CHANGES_PATH); const params = getParams(ctx); + // Cumulative ACKs require a positive capability frame from the server. Old + // v6 servers ignore these query params and keep per-message ACK semantics. params.set('streamBatch', '1'); + params.set(ACK_MODE_PARAM, CUMULATIVE_ACK_MODE); return new WebSocket(uri + `?${params.toString()}`); } } diff --git a/packages/zero-cache/src/types/streams.test.ts b/packages/zero-cache/src/types/streams.test.ts index 0601a29303..9487dce0ad 100644 --- a/packages/zero-cache/src/types/streams.test.ts +++ b/packages/zero-cache/src/types/streams.test.ts @@ -224,6 +224,7 @@ describe('streams with internal acks', () => { let cleanedUp: Promise; let cleanup: (m: Message[]) => void; let ackMessages: number[]; + let enableCumulativeAckCapability: boolean; let streamBatchMessages: number; let port: number; @@ -238,6 +239,7 @@ describe('streams with internal acks', () => { consumed = new Queue(); ackMessages = []; + enableCumulativeAckCapability = false; streamBatchMessages = 1; producer = Subscription.create({ consumed: m => consumed.enqueue(m), @@ -262,7 +264,12 @@ describe('streams with internal acks', () => { streamBatchMessages > 1 ? {maxMessages: streamBatchMessages} : undefined; - const streamOptions = {batch: batchOptions}; + const streamOptions = { + ack: enableCumulativeAckCapability + ? ('cumulative' as const) + : undefined, + batch: batchOptions, + }; void streamOut(lc, producer, ws, streamOptions); void streamOutStringified(lc, stringifiedProducer, ws, streamOptions); }); @@ -279,25 +286,27 @@ describe('streams with internal acks', () => { await server.close(); }); - async function startReceiver() { + async function startReceiver( + ack: 'cumulative' | 'cumulative-if-supported' = 'cumulative', + ) { ws = new WebSocket(`http://localhost:${port}/`); return { ws, - consumer: (await streamIn( - lc, - ws, - messageSchema, - )) as Subscription, + consumer: (await streamIn(lc, ws, messageSchema, { + ack, + })) as Subscription, }; } - async function startBatchReceiver() { + async function startBatchReceiver( + ack: 'cumulative' | 'cumulative-if-supported' = 'cumulative', + ) { ws = new WebSocket(`http://localhost:${port}/`); return { ws, - consumer: (await streamInBatches(lc, ws, messageSchema)) as Subscription< - StreamInPayload - >, + consumer: (await streamInBatches(lc, ws, messageSchema, { + ack, + })) as Subscription>, }; } @@ -381,7 +390,7 @@ describe('streams with internal acks', () => { ]); }); - test('pipelined stream uses per-message acks', async () => { + test('pipelined cumulative ack batches consumed messages', async () => { const messageCount = 96; const results: Promise[] = []; for (let i = 0; i < messageCount; i++) { @@ -396,6 +405,29 @@ describe('streams with internal acks', () => { consumed(); } + expect(await Promise.all(results)).toEqual( + Array(messageCount).fill('consumed'), + ); + expect(ackMessages).toEqual([96]); + consumer.cancel(); + expect(await cleanedUp).toEqual([]); + }); + + test('negotiated cumulative ack falls back to per-message without capability', async () => { + const messageCount = 96; + const results: Promise[] = []; + for (let i = 0; i < messageCount; i++) { + results.push(producer.push({from: i, to: i + 1, str: `msg-${i}`}).result); + } + + const {consumer} = await startReceiver('cumulative-if-supported'); + await vi.waitFor(() => expect(consumer.queued).toBe(messageCount)); + + const entries = await drainPipeline(messageCount, consumer); + for (const {consumed} of entries) { + consumed(); + } + expect(await Promise.all(results)).toEqual( Array(messageCount).fill('consumed'), ); @@ -406,6 +438,30 @@ describe('streams with internal acks', () => { expect(await cleanedUp).toEqual([]); }); + test('negotiated cumulative ack batches consumed messages after capability', async () => { + enableCumulativeAckCapability = true; + const messageCount = 96; + const results: Promise[] = []; + for (let i = 0; i < messageCount; i++) { + results.push(producer.push({from: i, to: i + 1, str: `msg-${i}`}).result); + } + + const {consumer} = await startReceiver('cumulative-if-supported'); + await vi.waitFor(() => expect(consumer.queued).toBe(messageCount)); + + const entries = await drainPipeline(messageCount, consumer); + for (const {consumed} of entries) { + consumed(); + } + + expect(await Promise.all(results)).toEqual( + Array(messageCount).fill('consumed'), + ); + expect(ackMessages).toEqual([96]); + consumer.cancel(); + expect(await cleanedUp).toEqual([]); + }); + test('pipelined outbound stream batches queued messages', async () => { streamBatchMessages = 4; const messageCount = 10; @@ -498,11 +554,36 @@ describe('streams with internal acks', () => { ], }, ]); - await vi.waitFor(() => expect(ackMessages).toEqual([1, 2, 3, 4, 5, 6])); + await vi.waitFor(() => expect(ackMessages).toEqual([4, 6])); expect(await cleanedUp).toEqual([]); }); - test('pipelined unexpected ack closes connection', async () => { + test('pipelined cumulative ack flushes pending ack on cleanup', async () => { + const messageCount = 3; + const results: Promise[] = []; + for (let i = 0; i < messageCount; i++) { + results.push(producer.push({from: i, to: i + 1, str: `msg-${i}`}).result); + } + + const {consumer} = await startReceiver(); + await vi.waitFor(() => expect(consumer.queued).toBe(messageCount)); + + const entries = await drainPipeline(messageCount, consumer); + for (const {consumed} of entries) { + consumed(); + } + expect(ackMessages).toEqual([]); + + consumer.cancel(); + + expect(await Promise.all(results)).toEqual( + Array(messageCount).fill('consumed'), + ); + expect(ackMessages).toEqual([3]); + expect(await cleanedUp).toEqual([]); + }); + + test('pipelined backwards ack closes connection', async () => { const received: JSONValue[] = []; const closed = resolver(); const results: Promise[] = []; @@ -518,8 +599,10 @@ describe('streams with internal acks', () => { await vi.waitFor(() => expect(received).toHaveLength(2)); ws.send(JSON.stringify({ack: 2})); + expect(await Promise.all(results)).toEqual(['consumed', 'consumed']); + + ws.send(JSON.stringify({ack: 1})); await closed.promise; - expect(await Promise.all(results)).toEqual(['unconsumed', 'unconsumed']); }); test('pipelined (unconsumed)', async () => { diff --git a/packages/zero-cache/src/types/streams.ts b/packages/zero-cache/src/types/streams.ts index 053d104731..38ddc9b2c5 100644 --- a/packages/zero-cache/src/types/streams.ts +++ b/packages/zero-cache/src/types/streams.ts @@ -29,6 +29,17 @@ import { // Consistent with Postgres keepalives, and shorter than the // commonly used default idle timeout of 1 minute. const PING_INTERVAL_MS = 30_000; +// Batch stream ACKs so row-heavy RM -> serving-replica traffic spends CPU on +// apply work instead of ACK writes. The count threshold keeps hot streams from +// sending one ACK per frame; the short timer keeps quiet streams from waiting +// for an arbitrary future frame before releasing sender-side backpressure. +const CUMULATIVE_ACK_EVERY = 128; +const CUMULATIVE_ACK_INTERVAL_MS = 5; +const CUMULATIVE_ACK_STREAM_OPTION = 'cumulative-ack-v1'; +const CUMULATIVE_ACK_STREAM_OPTIONS_FRAME = BigIntJSON.stringify({ + control: 'stream-options', + capabilities: [CUMULATIVE_ACK_STREAM_OPTION], +}); export type Source = AsyncIterable & { /** @@ -296,6 +307,7 @@ export function streamOutStringified( } type StreamOutOptions = { + ack?: 'per-message' | 'cumulative' | undefined; batch?: {maxMessages?: number | undefined} | undefined; }; @@ -310,22 +322,61 @@ async function streamOutInternal( sendPingsForLiveness(lc, sink, PING_INTERVAL_MS); const closer = WebSocketCloser.forSource(lc, sink, source); - const acks = new Queue(); - sink.addEventListener('message', ({data}) => { - try { - acks.enqueue(parseAck(data)); - } catch (e) { - lc.error?.(`error parsing ack`, e); - closer.close(e); - } - }); try { let nextID = 0; const {pipeline} = source; + if (options.ack === 'cumulative' && pipeline === undefined) { + throw new Error('Cumulative ACKs require a pipelined source'); + } + if (options.ack === 'cumulative') { + sink.send(CUMULATIVE_ACK_STREAM_OPTIONS_FRAME); + } if (pipeline) { const iterator = pipeline[Symbol.asyncIterator](); const maxBatchMessages = Math.max(1, options.batch?.maxMessages ?? 1); + let lastAck = 0; + const pending = new Map void>(); + sink.addEventListener('message', ({data}) => { + try { + const {ack} = parseAck(data); + if (ack < lastAck) { + // ACKs are cumulative and therefore monotonic. Moving backwards + // would mean the receiver is describing a different stream. + throw new Error(`Ack moved backwards from ${lastAck} to ${ack}`); + } + if (ack > nextID) { + if (nextID === 0 && pending.size === 0) { + // A reconnect can deliver a stale ACK before this side has sent + // anything on the new stream. It releases nothing, so ignore it. + return; + } + // After we have sent at least one frame, ACKing a future ID would + // release work the receiver could not have processed. + throw new Error( + `Unexpected ack ${ack}; only sent through ${nextID}`, + ); + } + if (ack === lastAck) { + // Duplicate cumulative ACK. Nothing new can be released. + return; + } + lastAck = ack; + // RM -> VS catchup can keep thousands of future stream IDs pending; + // pending is insertion-ordered by stream ID, so ACK cleanup stops at + // the first future ID and only pays for frames that can be released. + for (const [id, consumed] of pending) { + if (id > ack) { + break; + } + consumed(); + pending.delete(id); + } + } catch (e) { + lc.error?.(`error handling ack`, e); + closer.close(e); + } + }); lc.debug?.(`started pipelined outbound stream`); for (;;) { @@ -360,24 +411,14 @@ async function streamOutInternal( remaining--; if (remaining === 0) { // A single source payload can expand to several websocket IDs. - // The source is consumed only after all IDs are ACKed. + // The source is consumed only after all IDs are cumulatively ACKed. consumed(); } }; for (const msg of messages) { const id = ++nextID; + pending.set(id, consumeOne); outbound.push({id, msg}); - void (async () => { - const {ack} = await acks.dequeue(); - // lc.debug?.(`received ack`, ack); - if (ack !== id) { - throw new Error(`Unexpected ack for ${id}: ${ack}`); - } - consumeOne(); - })().catch(e => { - lc.error?.(`error handling ack`, e); - closer.close(e); - }); } } if (outbound.length === 0) { @@ -389,6 +430,16 @@ async function streamOutInternal( sink.send(data); } } else { + const acks = new Queue(); + sink.addEventListener('message', ({data}) => { + try { + acks.enqueue(parseAck(data)); + } catch (e) { + lc.error?.(`error parsing ack`, e); + closer.close(e); + } + }); + lc.debug?.(`started synchronous outbound stream`); for await (const payload of source) { const messages = expandPayload(payload); @@ -444,22 +495,27 @@ export function streamIn( lc: LogContext, source: WebSocket, schema: v.Type, + options: StreamInOptions = {}, ): Promise> { - return streamInInternal(lc, source, schema, false) as Promise>; + return streamInInternal(lc, source, schema, options, false) as Promise< + Source + >; } export function streamInBatches( lc: LogContext, source: WebSocket, schema: v.Type, + options: StreamInOptions = {}, ): Promise>> { - return streamInInternal(lc, source, schema, true); + return streamInInternal(lc, source, schema, options, true); } async function streamInInternal( lc: LogContext, source: WebSocket, schema: v.Type, + options: StreamInOptions, preserveBatches: boolean, ): Promise>> { expectPingsForLiveness(lc, source, PING_INTERVAL_MS); @@ -468,6 +524,7 @@ async function streamInInternal( msg: schema, id: v.number(), }); + const acker = new CumulativeAcker(source, options.ack === 'cumulative'); const sink: Subscription< StreamInPayload, @@ -476,14 +533,15 @@ async function streamInInternal( { consumed: streamOrBatch => { if (isStreamedBatch(streamOrBatch)) { - for (const {id} of streamOrBatch) { - source.send(JSON.stringify({ack: id} satisfies Ack)); - } + acker.consumedBatch(streamOrBatch); } else { - source.send(JSON.stringify({ack: streamOrBatch.id} satisfies Ack)); + acker.consumed(streamOrBatch.id); } }, - cleanup: () => closer.close(), + cleanup: () => { + acker.close(); + closer.close(); + }, }, streamOrBatch => isStreamedBatch(streamOrBatch) @@ -504,7 +562,19 @@ async function streamInInternal( } try { const value = BigIntJSON.parse(data); + if (isStreamOptionsFrame(value)) { + if ( + options.ack === 'cumulative-if-supported' && + value.capabilities.includes(CUMULATIVE_ACK_STREAM_OPTION) + ) { + acker.enableCumulative(); + } + return; + } const messages = parseStreamedMessages(value, streamedSchema); + for (const msg of messages) { + acker.received(msg.id); + } if (preserveBatches && messages.length > 1) { // The RM already grouped these messages into one websocket frame. Keep // that boundary visible so the VS can batch one write-worker apply. @@ -570,6 +640,164 @@ function parseStreamedMessages( return [v.parse(value, streamedSchema, 'passthrough')]; } +function isStreamOptionsFrame( + value: JSONValue, +): value is {readonly capabilities: readonly string[]} { + return ( + typeof value === 'object' && + value !== null && + !Array.isArray(value) && + (value as {readonly control?: unknown}).control === 'stream-options' && + Array.isArray((value as {readonly capabilities?: unknown}).capabilities) && + (value as {readonly capabilities: readonly unknown[]}).capabilities.every( + capability => typeof capability === 'string', + ) + ); +} + +type StreamInOptions = { + ack?: 'per-message' | 'cumulative' | 'cumulative-if-supported' | undefined; +}; + +class CumulativeAcker { + readonly #ws: WebSocket; + #batch: boolean; + // IDs that have been consumed after a gap. They become ACKable only when all + // earlier IDs are consumed. + readonly #outOfOrder = new Set(); + // highestReceived is a parser-side fact; highestAckable is a consumer-side + // fact. Keeping them separate lets us flush early when the receiver has caught + // up to everything it has actually seen. + #highestReceived = 0; + #highestAckable = 0; + #lastSent = 0; + #pendingSinceLastSend = 0; + + #timer: NodeJS.Timeout | undefined; + + constructor(ws: WebSocket, batch: boolean) { + this.#ws = ws; + this.#batch = batch; + } + + enableCumulative() { + this.#batch = true; + } + + received(id: number) { + this.#validateID(id); + this.#highestReceived = Math.max(this.#highestReceived, id); + } + + consumed(id: number) { + this.#validateID(id); + if (id <= this.#highestAckable) { + // A late completion for an already-ackable ID cannot advance the + // cumulative watermark. + return; + } + const previous = this.#highestAckable; + if (id === this.#highestAckable + 1) { + this.#highestAckable = id; + while (this.#outOfOrder.delete(this.#highestAckable + 1)) { + this.#highestAckable++; + } + } else { + this.#outOfOrder.add(id); + } + const advanced = this.#highestAckable - previous; + if (advanced === 0) { + // The receiver finished a future ID before a gap. Do not ACK past the + // missing ID; the sender may still need to retain it. + return; + } + this.#pendingSinceLastSend += advanced; + const belowBatchThreshold = + this.#pendingSinceLastSend < CUMULATIVE_ACK_EVERY; + const receiverCaughtUp = + this.#pendingSinceLastSend > 1 && + this.#highestAckable === this.#highestReceived; + if (this.#batch && belowBatchThreshold && !receiverCaughtUp) { + // Hot streams wait for the count threshold or timer so ACK traffic stays + // small. If the receiver has caught up to everything it has seen, flush + // now so the sender can release flow control promptly. + this.#schedule(); + } else { + this.flush(); + } + } + + consumedBatch(messages: readonly Streamed[]) { + if (messages.length === 0) { + return; + } + let expected = this.#highestAckable + 1; + for (const {id} of messages) { + this.#validateID(id); + if (id !== expected) { + // Contiguous transport batches are the hot path. If a caller hands us + // an out-of-order batch, fall back to the per-ID logic so gaps are + // tracked exactly. + for (const fallback of messages) { + this.consumed(fallback.id); + } + return; + } + expected++; + } + + const previous = this.#highestAckable; + const last = messages.at(-1); + assert(last !== undefined, 'non-empty batch must have a last id'); + this.#highestAckable = last.id; + while (this.#outOfOrder.delete(this.#highestAckable + 1)) { + this.#highestAckable++; + } + const advanced = this.#highestAckable - previous; + if (advanced === 0) { + return; + } + this.#pendingSinceLastSend += advanced; + this.flush(); + } + + flush() { + this.#clearTimer(); + if (this.#highestAckable <= this.#lastSent) { + // Avoid duplicate ACK frames when a timer fires after an eager flush. + return; + } + if (this.#ws.readyState !== this.#ws.OPEN) { + return; + } + this.#ws.send(JSON.stringify({ack: this.#highestAckable} satisfies Ack)); + this.#lastSent = this.#highestAckable; + this.#pendingSinceLastSend = 0; + } + + close() { + this.flush(); + this.#clearTimer(); + } + + #schedule() { + this.#timer ??= setTimeout(() => this.flush(), CUMULATIVE_ACK_INTERVAL_MS); + } + + #clearTimer() { + if (this.#timer !== undefined) { + clearTimeout(this.#timer); + this.#timer = undefined; + } + } + + #validateID(id: number) { + if (!Number.isSafeInteger(id) || id <= 0) { + throw new Error(`Invalid stream message id ${id}`); + } + } +} + class WebSocketCloser { readonly #lc: LogContext; readonly #ws: WebSocket;