Skip to content

perf(zero-cache): negotiate cumulative stream acks #10

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
Karavil wants to merge 1 commit into from
Closed

perf(zero-cache): negotiate cumulative stream acks #10

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 15 additions & 3 deletions packages/zero-cache/src/services/change-streamer/change-streamer-http.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,8 @@ const SNAPSHOT_PATH_PATTERN = '/replication/:version/snapshot';
const CHANGES_PATH_PATTERN = '/replication/:version/changes';
const PATH_REGEX = /\/replication\/v(?<version>\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`;
Expand Down Expand Up @@ -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,
Expand Down Expand Up @@ -243,23 +248,30 @@ export class ChangeStreamerHttpClient implements ChangeStreamer {

async subscribe(ctx: SubscriberContext): Promise<Source<Downstream>> {
const ws = await this.#openChangesWebSocket(ctx);
return streamIn(this.#lc, ws, downstreamSchema);
return streamIn(this.#lc, ws, downstreamSchema, {
ack: 'cumulative-if-supported',
});
}

async subscribeBatched(
ctx: SubscriberContext,
): Promise<Source<StreamInPayload<Downstream>>> {
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()}`);
}
}
Expand Down
113 changes: 98 additions & 15 deletions packages/zero-cache/src/types/streams.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -224,6 +224,7 @@ describe('streams with internal acks', () => {
let cleanedUp: Promise<Message[]>;
let cleanup: (m: Message[]) => void;
let ackMessages: number[];
let enableCumulativeAckCapability: boolean;
let streamBatchMessages: number;
let port: number;

Expand All @@ -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),
Expand All @@ -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);
});
Expand All @@ -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<Message>,
consumer: (await streamIn(lc, ws, messageSchema, {
ack,
})) as Subscription<Message>,
};
}

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<Message>
>,
consumer: (await streamInBatches(lc, ws, messageSchema, {
ack,
})) as Subscription<StreamInPayload<Message>>,
};
}

Expand Down Expand Up @@ -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<Result>[] = [];
for (let i = 0; i < messageCount; i++) {
Expand All @@ -396,6 +405,29 @@ describe('streams with internal acks', () => {
consumed();
}

expect(await Promise.all(results)).toEqual(
Array<Result>(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<Result>[] = [];
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<Result>(messageCount).fill('consumed'),
);
Expand All @@ -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<Result>[] = [];
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<Result>(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;
Expand Down Expand Up @@ -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<Result>[] = [];
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<Result>(messageCount).fill('consumed'),
);
expect(ackMessages).toEqual([3]);
expect(await cleanedUp).toEqual([]);
});

test('pipelined backwards ack closes connection', async () => {
const received: JSONValue[] = [];
const closed = resolver<void>();
const results: Promise<Result>[] = [];
Expand All @@ -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 () => {
Expand Down
Loading