From 9120345b70284069a7c8cd5e83f64eb4968b4267 Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Mon, 13 Jul 2026 12:46:03 +0300 Subject: [PATCH 1/2] feat(client): add MAXCOUNT and MAXSIZE options to XREAD and XREADGROUP Redis 8.10 adds cumulative reply caps across all streams in XREAD and XREADGROUP. Unlike COUNT (per-stream), MAXCOUNT caps the total number of entries and MAXSIZE soft-caps the total reply size in bytes across all stream keys. Both are exposed as optional numeric fields on the existing options objects and emitted as uppercase tokens in canonical order before STREAMS, omitted when unset. Purely additive: response shape, blocking, consumer-group, and stream-id semantics are unchanged and the existing Streams reply parser is reused. No client-side validation is added, so server errors propagate as-is. Options are since 8.10.0. The XREADGROUP option push order was aligned to the canonical order (COUNT MAXCOUNT MAXSIZE BLOCK CLAIM NOACK). Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/lib/commands/XREAD.spec.ts | 59 ++++++++++++++++ packages/client/lib/commands/XREAD.ts | 14 +++- .../client/lib/commands/XREADGROUP.spec.ts | 67 ++++++++++++++++++- packages/client/lib/commands/XREADGROUP.ts | 20 ++++-- 4 files changed, 154 insertions(+), 6 deletions(-) diff --git a/packages/client/lib/commands/XREAD.spec.ts b/packages/client/lib/commands/XREAD.spec.ts index 963989a2359..542e1387384 100644 --- a/packages/client/lib/commands/XREAD.spec.ts +++ b/packages/client/lib/commands/XREAD.spec.ts @@ -89,6 +89,45 @@ describe('XREAD', () => { ['XREAD', 'COUNT', '1', 'BLOCK', '0', 'STREAMS', 'key', '0-0'] ); }); + + it('with MAXCOUNT', () => { + assert.deepEqual( + parseArgs(XREAD, { + key: 'key', + id: '0-0' + }, { + MAXCOUNT: 3 + }), + ['XREAD', 'MAXCOUNT', '3', 'STREAMS', 'key', '0-0'] + ); + }); + + it('with MAXSIZE', () => { + assert.deepEqual( + parseArgs(XREAD, { + key: 'key', + id: '0-0' + }, { + MAXSIZE: 65536 + }), + ['XREAD', 'MAXSIZE', '65536', 'STREAMS', 'key', '0-0'] + ); + }); + + it('with COUNT, MAXCOUNT, MAXSIZE, BLOCK', () => { + assert.deepEqual( + parseArgs(XREAD, { + key: 'key', + id: '0-0' + }, { + COUNT: 2, + MAXCOUNT: 3, + MAXSIZE: 65536, + BLOCK: 0 + }), + ['XREAD', 'COUNT', '2', 'MAXCOUNT', '3', 'MAXSIZE', '65536', 'BLOCK', '0', 'STREAMS', 'key', '0-0'] + ); + }); }); testUtils.testAll('client.xRead', async client => { @@ -146,4 +185,24 @@ describe('XREAD', () => { } }); + testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'client.xRead with MAXCOUNT caps entries cumulatively across streams', async client => { + await Promise.all([ + client.xAdd('{t}s1', '1-0', { field: 'v1' }), + client.xAdd('{t}s1', '2-0', { field: 'v2' }), + client.xAdd('{t}s2', '1-0', { field: 'v3' }), + client.xAdd('{t}s2', '2-0', { field: 'v4' }) + ]); + + const reply = await client.xRead([ + { key: '{t}s1', id: '0' }, + { key: '{t}s2', id: '0' } + ], { + MAXCOUNT: 3 + }); + + assert.ok(reply); + const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0); + assert.equal(total, 3); + }, GLOBAL.SERVERS.OPEN); + }); diff --git a/packages/client/lib/commands/XREAD.ts b/packages/client/lib/commands/XREAD.ts index 19a6945fd5e..e4c1384e1db 100644 --- a/packages/client/lib/commands/XREAD.ts +++ b/packages/client/lib/commands/XREAD.ts @@ -39,12 +39,16 @@ export function pushXReadStreams(parser: CommandParser, streams: XReadStreams) { /** * Options for the XREAD command - * + * * @property COUNT - Limit the number of entries returned per stream + * @property MAXCOUNT - Cumulative cap on the total number of entries returned across all streams (Redis 8.10+) + * @property MAXSIZE - Soft cumulative cap on the total server reply size in bytes across all streams (Redis 8.10+) * @property BLOCK - Milliseconds to block waiting for new entries (0 for indefinite) */ export interface XReadOptions { COUNT?: number; + MAXCOUNT?: number; + MAXSIZE?: number; BLOCK?: number; } @@ -57,6 +61,14 @@ export default { parser.push('COUNT', options.COUNT.toString()); } + if (options?.MAXCOUNT !== undefined) { + parser.push('MAXCOUNT', options.MAXCOUNT.toString()); + } + + if (options?.MAXSIZE !== undefined) { + parser.push('MAXSIZE', options.MAXSIZE.toString()); + } + if (options?.BLOCK !== undefined) { parser.push('BLOCK', options.BLOCK.toString()); } diff --git a/packages/client/lib/commands/XREADGROUP.spec.ts b/packages/client/lib/commands/XREADGROUP.spec.ts index 89ba4b05321..06cc6346a50 100644 --- a/packages/client/lib/commands/XREADGROUP.spec.ts +++ b/packages/client/lib/commands/XREADGROUP.spec.ts @@ -117,7 +117,48 @@ describe('XREADGROUP', () => { NOACK: true, CLAIM: 100 }), - ['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'NOACK', 'CLAIM', '100', 'STREAMS', 'key', '0-0'] + ['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'CLAIM', '100', 'NOACK', 'STREAMS', 'key', '0-0'] + ); + }); + + it('with MAXCOUNT', () => { + assert.deepEqual( + parseArgs(XREADGROUP, 'group', 'consumer', { + key: 'key', + id: '0-0' + }, { + MAXCOUNT: 3 + }), + ['XREADGROUP', 'GROUP', 'group', 'consumer', 'MAXCOUNT', '3', 'STREAMS', 'key', '0-0'] + ); + }); + + it('with MAXSIZE', () => { + assert.deepEqual( + parseArgs(XREADGROUP, 'group', 'consumer', { + key: 'key', + id: '0-0' + }, { + MAXSIZE: 65536 + }), + ['XREADGROUP', 'GROUP', 'group', 'consumer', 'MAXSIZE', '65536', 'STREAMS', 'key', '0-0'] + ); + }); + + it('with COUNT, MAXCOUNT, MAXSIZE, BLOCK, CLAIM, NOACK', () => { + assert.deepEqual( + parseArgs(XREADGROUP, 'group', 'consumer', { + key: 'key', + id: '0-0' + }, { + COUNT: 2, + MAXCOUNT: 3, + MAXSIZE: 65536, + BLOCK: 0, + CLAIM: 100, + NOACK: true + }), + ['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '2', 'MAXCOUNT', '3', 'MAXSIZE', '65536', 'BLOCK', '0', 'CLAIM', '100', 'NOACK', 'STREAMS', 'key', '0-0'] ); }); }); @@ -233,4 +274,28 @@ describe('XREADGROUP', () => { assert.equal(typeof readGroupReply[0].messages[0].millisElapsedFromDelivery, 'number'); assert.equal(typeof readGroupReply[0].messages[0].deliveriesCounter, 'number'); }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'xReadGroup - MAXCOUNT caps entries cumulatively across streams', async client => { + await Promise.all([ + client.xGroupCreate('{t}s1', 'group', '0', { MKSTREAM: true }), + client.xGroupCreate('{t}s2', 'group', '0', { MKSTREAM: true }) + ]); + await Promise.all([ + client.xAdd('{t}s1', '1-0', { field: 'v1' }), + client.xAdd('{t}s1', '2-0', { field: 'v2' }), + client.xAdd('{t}s2', '1-0', { field: 'v3' }), + client.xAdd('{t}s2', '2-0', { field: 'v4' }) + ]); + + const reply = await client.xReadGroup('group', 'consumer', [ + { key: '{t}s1', id: '>' }, + { key: '{t}s2', id: '>' } + ], { + MAXCOUNT: 3 + }); + + assert.ok(reply); + const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0); + assert.equal(total, 3); + }, GLOBAL.SERVERS.OPEN); }); diff --git a/packages/client/lib/commands/XREADGROUP.ts b/packages/client/lib/commands/XREADGROUP.ts index 7ff9b9c6460..9f3b1974ea2 100644 --- a/packages/client/lib/commands/XREADGROUP.ts +++ b/packages/client/lib/commands/XREADGROUP.ts @@ -7,12 +7,16 @@ import { transformStreamsMessagesReplyResp2, transformStreamsMessagesReplyResp3C * Options for the XREADGROUP command * * @property COUNT - Limit the number of entries returned per stream + * @property MAXCOUNT - Cumulative cap on the total number of entries returned across all streams (Redis 8.10+) + * @property MAXSIZE - Soft cumulative cap on the total server reply size in bytes across all streams (Redis 8.10+) * @property BLOCK - Milliseconds to block waiting for new entries (0 for indefinite) * @property NOACK - Skip adding the message to the PEL (Pending Entries List) * @property CLAIM - Prepend PEL entries that are at least this many milliseconds old */ export interface XReadGroupOptions { COUNT?: number; + MAXCOUNT?: number; + MAXSIZE?: number; BLOCK?: number; NOACK?: boolean; CLAIM?: number; @@ -33,18 +37,26 @@ export default { parser.push('COUNT', options.COUNT.toString()); } - if (options?.BLOCK !== undefined) { - parser.push('BLOCK', options.BLOCK.toString()); + if (options?.MAXCOUNT !== undefined) { + parser.push('MAXCOUNT', options.MAXCOUNT.toString()); } - if (options?.NOACK) { - parser.push('NOACK'); + if (options?.MAXSIZE !== undefined) { + parser.push('MAXSIZE', options.MAXSIZE.toString()); + } + + if (options?.BLOCK !== undefined) { + parser.push('BLOCK', options.BLOCK.toString()); } if (options?.CLAIM !== undefined) { parser.push('CLAIM', options.CLAIM.toString()); } + if (options?.NOACK) { + parser.push('NOACK'); + } + pushXReadStreams(parser, streams); }, /** From 530b0b66f5f253deb924ebae4e54f45c17c52744 Mon Sep 17 00:00:00 2001 From: Nikolay Karadzhov Date: Mon, 13 Jul 2026 13:30:45 +0300 Subject: [PATCH 2/2] test(client): add MAXSIZE behavior tests; fix XREAD COUNT zero handling Add cumulative MAXSIZE cap integration tests for XREAD and XREADGROUP (MAXSIZE 1 over two streams returns exactly one entry, matching the server's global soft-cap semantics), gated to Redis 8.10+. Fix XREAD to guard COUNT with `\!== undefined` instead of a truthy check, so `COUNT: 0` is forwarded consistently with MAXCOUNT/MAXSIZE/BLOCK and with XREADGROUP. Co-Authored-By: Claude Opus 4.8 (1M context) --- packages/client/lib/commands/XREAD.spec.ts | 22 ++++++++++++++++ packages/client/lib/commands/XREAD.ts | 2 +- .../client/lib/commands/XREADGROUP.spec.ts | 26 +++++++++++++++++++ 3 files changed, 49 insertions(+), 1 deletion(-) diff --git a/packages/client/lib/commands/XREAD.spec.ts b/packages/client/lib/commands/XREAD.spec.ts index 542e1387384..b1a5aad4dea 100644 --- a/packages/client/lib/commands/XREAD.spec.ts +++ b/packages/client/lib/commands/XREAD.spec.ts @@ -205,4 +205,26 @@ describe('XREAD', () => { assert.equal(total, 3); }, GLOBAL.SERVERS.OPEN); + testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'client.xRead with MAXSIZE caps reply size cumulatively across streams', async client => { + await Promise.all([ + client.xAdd('{t}s1', '1-0', { field: 'v1' }), + client.xAdd('{t}s1', '2-0', { field: 'v2' }), + client.xAdd('{t}s2', '1-0', { field: 'v3' }), + client.xAdd('{t}s2', '2-0', { field: 'v4' }) + ]); + + // MAXSIZE is a soft cumulative byte cap across all streams; the server always + // returns at least one entry, so MAXSIZE of 1 byte yields exactly one entry total. + const reply = await client.xRead([ + { key: '{t}s1', id: '0' }, + { key: '{t}s2', id: '0' } + ], { + MAXSIZE: 1 + }); + + assert.ok(reply); + const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0); + assert.equal(total, 1); + }, GLOBAL.SERVERS.OPEN); + }); diff --git a/packages/client/lib/commands/XREAD.ts b/packages/client/lib/commands/XREAD.ts index e4c1384e1db..a484edf5058 100644 --- a/packages/client/lib/commands/XREAD.ts +++ b/packages/client/lib/commands/XREAD.ts @@ -57,7 +57,7 @@ export default { parseCommand(parser: CommandParser, streams: XReadStreams, options?: XReadOptions) { parser.push('XREAD'); - if (options?.COUNT) { + if (options?.COUNT !== undefined) { parser.push('COUNT', options.COUNT.toString()); } diff --git a/packages/client/lib/commands/XREADGROUP.spec.ts b/packages/client/lib/commands/XREADGROUP.spec.ts index 06cc6346a50..f46d16f6fb3 100644 --- a/packages/client/lib/commands/XREADGROUP.spec.ts +++ b/packages/client/lib/commands/XREADGROUP.spec.ts @@ -298,4 +298,30 @@ describe('XREADGROUP', () => { const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0); assert.equal(total, 3); }, GLOBAL.SERVERS.OPEN); + + testUtils.testWithClientIfVersionWithinRange([[8, 10], 'LATEST'], 'xReadGroup - MAXSIZE caps reply size cumulatively across streams', async client => { + await Promise.all([ + client.xGroupCreate('{t}s1', 'group', '0', { MKSTREAM: true }), + client.xGroupCreate('{t}s2', 'group', '0', { MKSTREAM: true }) + ]); + await Promise.all([ + client.xAdd('{t}s1', '1-0', { field: 'v1' }), + client.xAdd('{t}s1', '2-0', { field: 'v2' }), + client.xAdd('{t}s2', '1-0', { field: 'v3' }), + client.xAdd('{t}s2', '2-0', { field: 'v4' }) + ]); + + // MAXSIZE is a soft cumulative byte cap across all streams; the server always + // returns at least one entry, so MAXSIZE of 1 byte yields exactly one entry total. + const reply = await client.xReadGroup('group', 'consumer', [ + { key: '{t}s1', id: '>' }, + { key: '{t}s2', id: '>' } + ], { + MAXSIZE: 1 + }); + + assert.ok(reply); + const total = reply.reduce((sum, stream) => sum + stream.messages.length, 0); + assert.equal(total, 1); + }, GLOBAL.SERVERS.OPEN); });