diff --git a/packages/zero-cache/src/config/zero-config.test.ts b/packages/zero-cache/src/config/zero-config.test.ts index 24006b596d..5c6f0c5835 100644 --- a/packages/zero-cache/src/config/zero-config.test.ts +++ b/packages/zero-cache/src/config/zero-config.test.ts @@ -406,11 +406,11 @@ test('zero-cache --help', () => { rather, it protects the system when the upstream throughput exceeds the downstream throughput. - --change-streamer-flow-control-consensus-padding-seconds number default: 1 + --change-streamer-flow-control-consensus-padding-seconds number default: 0.1 ZERO_CHANGE_STREAMER_FLOW_CONTROL_CONSENSUS_PADDING_SECONDS env - During periodic flow control checks (every 64kb), the amount of time to wait after the - majority of subscribers have acked, after which replication will continue even if - some subscribers have yet to ack. (Note that this is not a timeout for the entire send, + During flow control flushes, the amount of time to wait after the majority of + subscribers have acked, after which replication will continue even if some subscribers + have yet to ack. (Note that this is not a timeout for the entire send, but a timeout that starts after the majority of receivers have acked.) This allows a bounded amount of time for backlogged subscribers to catch up on each flush diff --git a/packages/zero-cache/src/config/zero-config.ts b/packages/zero-cache/src/config/zero-config.ts index db494f2fe3..e20aed427e 100644 --- a/packages/zero-cache/src/config/zero-config.ts +++ b/packages/zero-cache/src/config/zero-config.ts @@ -651,11 +651,11 @@ export const zeroOptions = { }, flowControlConsensusPaddingSeconds: { - type: v.number().default(1), + type: v.number().default(0.1), desc: [ - `During periodic flow control checks (every 64kb), the amount of time to wait after the`, - `majority of subscribers have acked, after which replication will continue even if`, - `some subscribers have yet to ack. (Note that this is not a timeout for the {italic entire} send,`, + `During flow control flushes, the amount of time to wait after the majority of`, + `subscribers have acked, after which replication will continue even if some subscribers`, + `have yet to ack. (Note that this is not a timeout for the {italic entire} send,`, `but a timeout that starts {italic after} the majority of receivers have acked.)`, ``, `This allows a bounded amount of time for backlogged subscribers to catch up on each flush`, diff --git a/packages/zero-cache/src/services/change-streamer/broadcast.test.ts b/packages/zero-cache/src/services/change-streamer/broadcast.test.ts index ea9a855406..f3e962581b 100644 --- a/packages/zero-cache/src/services/change-streamer/broadcast.test.ts +++ b/packages/zero-cache/src/services/change-streamer/broadcast.test.ts @@ -1,9 +1,11 @@ -import {describe, expect, test} from 'vitest'; +import {afterEach, describe, expect, test, vi} from 'vitest'; import {BigIntJSON} from '../../../../shared/src/bigint-json.ts'; import {createSilentLogContext} from '../../../../shared/src/logging-test-utils.ts'; import {sleep} from '../../../../shared/src/sleep.ts'; +import type {Subscription} from '../../types/subscription.ts'; import {ReplicationMessages} from '../replicator/test-utils.ts'; import {Broadcast} from './broadcast.ts'; +import type {Subscriber} from './subscriber.ts'; import {createSubscriber} from './test-utils.ts'; const json = BigIntJSON.stringify; @@ -11,6 +13,58 @@ const json = BigIntJSON.stringify; describe('change-streamer/broadcast', () => { const messages = new ReplicationMessages({issues: 'id'}); const lc = createSilentLogContext(); + const change: [string, 'begin', string] = [ + '11', + 'begin', + json(['begin', messages.begin(), {commitWatermark: '13'}]), + ]; + + afterEach(() => { + vi.useRealTimers(); + vi.restoreAllMocks(); + }); + + async function consumeOne(downstream: Subscription) { + const pipeline = downstream.pipeline; + if (pipeline === undefined) { + throw new Error('Expected pipelined subscription'); + } + const next = await pipeline[Symbol.asyncIterator]().next(); + if (next.done) { + throw new Error('Expected subscription message'); + } + next.value.consumed(); + } + + async function flushCompletions() { + await Promise.resolve(); + await Promise.resolve(); + } + + async function makeSubscribers(count: number) { + const subscribers: Subscriber[] = []; + const downstreams: Subscription[] = []; + for (let i = 0; i < count; i++) { + const [subscriber, , downstream] = createSubscriber('00', true); + subscribers.push(subscriber); + downstreams.push(downstream); + } + await Promise.all(downstreams.map(consumeOne)); + await flushCompletions(); + return {subscribers, downstreams}; + } + + async function consumeChange(downstream: Subscription) { + await consumeOne(downstream); + await flushCompletions(); + } + + async function closeAll(subscribers: readonly Subscriber[]) { + for (const sub of subscribers) { + sub.close(); + } + await flushCompletions(); + } test('without tracking', () => { const [sub1, stream1] = createSubscriber('00', true); @@ -76,6 +130,102 @@ describe('change-streamer/broadcast', () => { } }); + test('flow control releases after majority plus padding', async () => { + vi.useFakeTimers(); + const {subscribers, downstreams} = await makeSubscribers(4); + const broadcast = new Broadcast(subscribers, change, { + lc, + flowControlConsensusPaddingMs: 50, + }); + + await consumeChange(downstreams[0]); + await consumeChange(downstreams[1]); + await vi.advanceTimersByTimeAsync(100); + expect(broadcast.isDone).toBe(false); + + await consumeChange(downstreams[2]); + await vi.advanceTimersByTimeAsync(49); + expect(broadcast.isDone).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + await broadcast.done; + expect(broadcast.isDone).toBe(true); + + await closeAll(subscribers); + }); + + test('flow control timer resets after post-majority progress', async () => { + vi.useFakeTimers(); + const {subscribers, downstreams} = await makeSubscribers(5); + const broadcast = new Broadcast(subscribers, change, { + lc, + flowControlConsensusPaddingMs: 50, + }); + + await consumeChange(downstreams[0]); + await consumeChange(downstreams[1]); + await consumeChange(downstreams[2]); + await vi.advanceTimersByTimeAsync(30); + await consumeChange(downstreams[3]); + + await vi.advanceTimersByTimeAsync(49); + expect(broadcast.isDone).toBe(false); + + await vi.advanceTimersByTimeAsync(1); + await broadcast.done; + expect(broadcast.isDone).toBe(true); + + await closeAll(subscribers); + }); + + test('flow control timer is cleared when all subscribers complete', async () => { + vi.useFakeTimers(); + const clearTimeoutSpy = vi.spyOn(globalThis, 'clearTimeout'); + const {subscribers, downstreams} = await makeSubscribers(4); + const broadcast = new Broadcast(subscribers, change, { + lc, + flowControlConsensusPaddingMs: 50, + }); + + await consumeChange(downstreams[0]); + await consumeChange(downstreams[1]); + await consumeChange(downstreams[2]); + expect(broadcast.isDone).toBe(false); + + await consumeChange(downstreams[3]); + await broadcast.done; + expect(broadcast.isDone).toBe(true); + expect(clearTimeoutSpy).toHaveBeenCalled(); + + await vi.advanceTimersByTimeAsync(50); + expect(broadcast.isDone).toBe(true); + await closeAll(subscribers); + }); + + test('negative flow control padding disables early release', async () => { + vi.useFakeTimers(); + const {subscribers, downstreams} = await makeSubscribers(4); + const broadcast = new Broadcast(subscribers, change, { + lc, + flowControlConsensusPaddingMs: -1, + }); + + await consumeChange(downstreams[0]); + await consumeChange(downstreams[1]); + await consumeChange(downstreams[2]); + expect(broadcast.checkProgress(lc, -1, performance.now() + 1000)).toBe( + false, + ); + await vi.advanceTimersByTimeAsync(1000); + expect(broadcast.isDone).toBe(false); + + await consumeChange(downstreams[3]); + await broadcast.done; + expect(broadcast.isDone).toBe(true); + + await closeAll(subscribers); + }); + test('checkProgress', async () => { const [sub1] = createSubscriber('00', true); const [sub2] = createSubscriber('00', true); diff --git a/packages/zero-cache/src/services/change-streamer/broadcast.ts b/packages/zero-cache/src/services/change-streamer/broadcast.ts index 00aba1a160..9ceaebc72c 100644 --- a/packages/zero-cache/src/services/change-streamer/broadcast.ts +++ b/packages/zero-cache/src/services/change-streamer/broadcast.ts @@ -3,6 +3,8 @@ import {resolver} from '@rocicorp/resolver'; import type {WatermarkedChange} from './change-streamer-service.ts'; import type {Subscriber} from './subscriber.ts'; +type BroadcastChange = WatermarkedChange | readonly WatermarkedChange[]; + /** * Initiates and tracks the progress of a change broadcasted to * a set of subscribers. @@ -10,9 +12,9 @@ import type {Subscriber} from './subscriber.ts'; * Creating a `Broadcast` automatically initiates the send. * * By default, {@link Broadcast.done} resolves when all subscribers - * have acked the change. However, {@link Broadcast.checkProgress()} - * can be called to resolve the broadcast earlier based on the flow - * control policy. + * have acked the change. When flow control options are supplied, + * {@link Broadcast.done} resolves after majority consensus plus the + * configured padding. */ export class Broadcast { /** @@ -21,17 +23,24 @@ export class Broadcast { */ static withoutTracking( subscribers: Iterable, - change: WatermarkedChange, + change: BroadcastChange, ) { + const changes = normalizeChanges(change); for (const sub of subscribers) { - void sub.send(change); + void sub.sendBatch(changes); } } + // Subscribers still gating this broadcast. A subscriber leaves this set when + // its downstream stream has consumed the batch. readonly #pending: Set; + // Completion samples are retained for flow-control logs; they explain which + // subscribers were fast enough to form the majority. readonly #completed: Completed[]; readonly #done = resolver(); + readonly #flowControl: FlowControlOptions | undefined; #isDone = false; + #flowControlTimer: ReturnType | undefined; readonly #watermark: string; readonly #majority: number; @@ -43,22 +52,35 @@ export class Broadcast { * Broadcasts the `change` to the `subscribers` and tracks their * completion. */ - constructor(subscribers: Iterable, change: WatermarkedChange) { + constructor( + subscribers: Iterable, + change: BroadcastChange, + flowControl?: FlowControlOptions, + ) { + const changes = normalizeChanges(change); this.#pending = new Set(subscribers); this.#completed = []; - this.#watermark = change[0]; + this.#flowControl = flowControl; + // The last change in the broadcast is the highest commit/order watermark + // represented by this batch, so it is the useful label for diagnostics. + this.#watermark = changes.at(-1)?.[0] ?? 'none'; + // "More than half" keeps one slow or broken minority from stalling the RM, + // while a one-subscriber topology still waits for that subscriber. this.#majority = Math.floor(this.#pending.size / 2) + 1; for (const sub of this.#pending) { - const changes = sub.numPending + 1; // add one for this `change` + // Log the work visible to the subscriber at send time: its existing + // downstream backlog plus this broadcast's new logical messages. + const numChanges = sub.numPending + changes.length; void sub - .send(change) + .sendBatch(changes) .catch(() => {}) - .finally(() => this.#markCompleted(sub, changes)); + .finally(() => this.#markCompleted(sub, numChanges)); } - // set done if there are no subscribers (mainly for tests) if (this.#pending.size === 0) { + // With no subscribers, there is nobody to apply flow control to; let the + // upstream continue immediately. this.#setDone(); } } @@ -69,12 +91,71 @@ export class Broadcast { this.#pending.delete(sub); if (this.#pending.size === 0) { this.#setDone(); + } else { + this.#resetConsensusTimer(); } } - #setDone() { + #setDone(): boolean { + if (this.#isDone) { + return false; + } this.#isDone = true; + this.#clearConsensusTimer(); this.#done.resolve(); + return true; + } + + #resetConsensusTimer() { + if ( + this.#isDone || + this.#flowControl === undefined || + !(this.#flowControl.flowControlConsensusPaddingMs >= 0) || + this.#completed.length < this.#majority + ) { + // Timer starts only after majority completion and only when early release + // is enabled. Before that, releasing would let the RM outrun the VS fleet. + return; + } + this.#clearConsensusTimer(); + this.#flowControlTimer = setTimeout( + this.#releaseAfterConsensusPadding, + this.#flowControl.flowControlConsensusPaddingMs, + ); + } + + readonly #releaseAfterConsensusPadding = () => { + this.#flowControlTimer = undefined; + const flowControl = this.#flowControl; + if ( + this.#isDone || + flowControl === undefined || + this.#pending.size === 0 || + this.#completed.length < this.#majority + ) { + // The callback can race with normal completion or option changes; in + // those cases there is either nothing left to release or no consensus yet. + return; + } + const now = performance.now(); + if ( + now - this.#latestCompleted < + flowControl.flowControlConsensusPaddingMs + ) { + this.#resetConsensusTimer(); + return; + } + this.#logWithState( + flowControl.lc, + `continuing with ${this.#pending.size} subscriber(s) still pending`, + now - this.#start, + ); + this.#setDone(); + }; + + #clearConsensusTimer() { + clearTimeout(this.#flowControlTimer); + this.#flowControlTimer = undefined; } get isDone(): boolean { @@ -159,9 +240,15 @@ export class Broadcast { flowControlConsensusPaddingMs: number, now: number, ) { + if (this.#isDone) { + return true; + } if (this.#pending.size === 0) { return true; } + if (!(flowControlConsensusPaddingMs >= 0)) { + return false; + } const elapsed = now - this.#start; if (this.#completed.length < this.#majority) { if (elapsed >= 1000) { @@ -206,6 +293,18 @@ export class Broadcast { } } +function normalizeChanges( + change: BroadcastChange, +): readonly WatermarkedChange[] { + return isWatermarkedChange(change) ? [change] : change; +} + +function isWatermarkedChange( + change: BroadcastChange, +): change is WatermarkedChange { + return typeof change[0] === 'string'; +} + /** Tracks the completed result of a single subscriber. */ type Completed = { sub: Subscriber; @@ -214,3 +313,8 @@ type Completed = { /** The elapsed milliseconds. */ elapsed: number; }; + +export type FlowControlOptions = { + readonly lc: LogContext; + readonly flowControlConsensusPaddingMs: number; +}; diff --git a/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts b/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts index 68a26214f4..6918013f87 100644 --- a/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts +++ b/packages/zero-cache/src/services/change-streamer/change-streamer-service.ts @@ -1,4 +1,3 @@ -import {getDefaultHighWaterMark} from 'node:stream'; import type {LogContext} from '@rocicorp/logger'; import {resolver} from '@rocicorp/resolver'; import {unreachable} from '../../../../shared/src/asserts.ts'; @@ -58,6 +57,13 @@ export type TuningOptions = StorerOptions & { flowControlConsensusPaddingSeconds: number; }; +// This is the RM-side byte window for forwarded, already-stringified changes +// waiting on serving-replica flow control. Node's default 16 KiB stream window +// made the RM pause before a row-heavy transaction could fill the VS write +// pipeline. 2 MiB is still a bounded queue, but large enough for the applier to +// batch useful work instead of oscillating between RM sends and VS ACKs. +export const FORWARDER_FLOW_CONTROL_BYTES_THRESHOLD = 2 * 1024 * 1024; + /** * Performs initialization and schema migrations to initialize a ChangeStreamerImpl. */ @@ -366,10 +372,6 @@ class ChangeStreamerImpl implements ChangeStreamerService { await this.#storer.assumeOwnership(this.#purgeLock); this.#purgeLock = null; - // The threshold in (estimated number of) bytes to send() on subscriber - // websockets before `await`-ing the I/O buffers to be ready for more. - const flushBytesThreshold = getDefaultHighWaterMark(false); - while (this.#state.shouldRun()) { let err: unknown; let watermark: string | null = null; @@ -443,8 +445,7 @@ class ChangeStreamerImpl implements ChangeStreamerService { const json = this.#storer.store(watermark, change); const entry: WatermarkedChange = [watermark, change[1].tag, json]; unflushedBytes += json.length; - if (unflushedBytes < flushBytesThreshold) { - // pipeline changes until flushBytesThreshold + if (unflushedBytes < FORWARDER_FLOW_CONTROL_BYTES_THRESHOLD) { this.#forwarder.forward(entry); } else { // Wait for messages to clear socket buffers to ensure that they diff --git a/packages/zero-cache/src/services/change-streamer/forwarder.ts b/packages/zero-cache/src/services/change-streamer/forwarder.ts index ce5a5dd8ca..87a48a9454 100644 --- a/packages/zero-cache/src/services/change-streamer/forwarder.ts +++ b/packages/zero-cache/src/services/change-streamer/forwarder.ts @@ -9,19 +9,29 @@ export type ProgressMonitorOptions = { flowControlConsensusPaddingSeconds: number; }; +const FORWARD_BATCH_SIZE = 64; + export class Forwarder { readonly #lc: LogContext; readonly #progressMonitorOptions: ProgressMonitorOptions; + // Active subscribers receive new changes immediately. Queued subscribers + // connected in the middle of a transaction and must wait for catchup handoff + // so they do not see the tail of a transaction without its beginning. readonly #active = new Set(); readonly #queued = new Set(); #inTransaction = false; + // Batch the data-heavy middle of a transaction before fanning it out to + // every view-syncer. Transaction boundaries still flush immediately so + // subscriber handoff and catchup keep the same observable ordering. + #pending: WatermarkedChange[] = []; #currentBroadcast: Broadcast | undefined; #progressMonitor: NodeJS.Timeout | undefined; + #pendingFlush: NodeJS.Timeout | undefined; constructor( lc: LogContext, - opts: ProgressMonitorOptions = {flowControlConsensusPaddingSeconds: 1}, + opts: ProgressMonitorOptions = {flowControlConsensusPaddingSeconds: 0.1}, ) { this.#lc = lc.withContext('component', 'progress-monitor'); this.#progressMonitorOptions = opts; @@ -39,8 +49,9 @@ export class Forwarder { } const {flowControlConsensusPaddingSeconds} = this.#progressMonitorOptions; - // A negative number disables early flow control release. if (flowControlConsensusPaddingSeconds >= 0) { + // Non-negative padding enables majority-based early release. Negative + // padding is the escape hatch for "wait for every subscriber" debugging. this.#currentBroadcast?.checkProgress( this.#lc, flowControlConsensusPaddingSeconds * 1000, @@ -51,6 +62,7 @@ export class Forwarder { stopProgressMonitor() { clearInterval(this.#progressMonitor); + this.#clearPendingFlush(); } /** @@ -82,11 +94,25 @@ export class Forwarder { * occasionally to avoid memory blowup. */ forward(entry: WatermarkedChange) { - Broadcast.withoutTracking(this.#active.values(), entry); + this.#pending.push(entry); + const batchFull = this.#pending.length >= FORWARD_BATCH_SIZE; + const transactionBoundary = + entry[1] === 'begin' || entry[1] === 'commit' || entry[1] === 'rollback'; + if (batchFull || transactionBoundary) { + // Full batches bound queued JSON and event-loop delay. Transaction + // boundaries flush immediately so subscriber handoff, catchup, and + // rollback/commit visibility stay aligned with the Storer. + this.#flushPendingWithoutTracking(); + } else { + // A quiet transaction may never hit the size threshold, so schedule a + // zero-delay flush to avoid waiting for unrelated future traffic. + this.#schedulePendingFlush(); + } this.#updateActiveSubscribers(entry[1]); } sendStatus(status: Status) { + this.#flushPendingWithoutTracking(); for (const sub of this.#active.values()) { sub.sendStatus(status); } @@ -97,7 +123,18 @@ export class Forwarder { * Promise that resolves when replication should continue. */ async forwardWithFlowControl(entry: WatermarkedChange) { - const broadcast = new Broadcast(this.#active.values(), entry); + const {flowControlConsensusPaddingSeconds} = this.#progressMonitorOptions; + const flowControlConsensusPaddingMs = + flowControlConsensusPaddingSeconds * 1000; + this.#pending.push(entry); + this.#clearPendingFlush(); + const broadcast = new Broadcast( + this.#active.values(), + this.#drainPending(), + flowControlConsensusPaddingMs >= 0 + ? {lc: this.#lc, flowControlConsensusPaddingMs} + : undefined, + ); this.#updateActiveSubscribers(entry[1]); // set for progress tracking @@ -112,6 +149,38 @@ export class Forwarder { } } + #flushPendingWithoutTracking() { + this.#clearPendingFlush(); + const pending = this.#drainPending(); + if (pending.length > 0) { + Broadcast.withoutTracking(this.#active.values(), pending); + } + } + + #schedulePendingFlush() { + if (this.#pendingFlush === undefined) { + this.#pendingFlush = setTimeout( + () => this.#flushPendingWithoutTracking(), + 0, + ); + } + } + + #clearPendingFlush() { + if (this.#pendingFlush !== undefined) { + clearTimeout(this.#pendingFlush); + this.#pendingFlush = undefined; + } + } + + #drainPending(): WatermarkedChange[] { + // Transfer ownership of the backing array instead of copying it; this is on + // the fanout hot path and the next forward() can append to a fresh array. + const pending = this.#pending; + this.#pending = []; + return pending; + } + #updateActiveSubscribers(tag: ChangeTag) { switch (tag) { case 'begin': diff --git a/packages/zero-cache/src/services/change-streamer/subscriber.ts b/packages/zero-cache/src/services/change-streamer/subscriber.ts index bd188dcd73..715f3fa7fe 100644 --- a/packages/zero-cache/src/services/change-streamer/subscriber.ts +++ b/packages/zero-cache/src/services/change-streamer/subscriber.ts @@ -61,6 +61,18 @@ export class Subscriber { } } + async sendBatch(changes: readonly WatermarkedChange[]) { + if (this.#backlog) { + for (const change of changes) { + if (change[0] > this.#watermark) { + this.#backlog.push(change); + } + } + return; + } + await Promise.all(changes.map(change => this.#sendChange(change))); + } + #initialized = false; /**