feat(xreadgroup): add claim attribute (#3122)

* feat(xreadgroup): add claim attribute

the CLAIM attribute can be used to instruct redis to return
PEL ( Pending Entries List ) entries with their respective
deliveries and ms since last delivery

* remove m01 from test matrix

* add jsdoc

.github/workflows/tests.yml

@@ -22,7 +22,7 @@ jobs:
       fail-fast: false
       matrix:
         node-version: ["18", "20", "22"]
-        redis-version: ["rs-7.4.0-v1", "8.0.2", "8.2", "8.4-M01-pre", "8.4-RC1-pre"]
+        redis-version: ["rs-7.4.0-v1", "8.0.2", "8.2", "8.4-RC1-pre"]
     steps:
       - uses: actions/checkout@v4
         with:

packages/client/lib/commands/XREADGROUP.spec.ts

@@ -93,6 +93,33 @@ describe('XREADGROUP', () => {
         ['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'NOACK', 'STREAMS', 'key', '0-0']
       );
     });
+
+    it('with CLAIM', () => {
+      assert.deepEqual(
+        parseArgs(XREADGROUP, 'group', 'consumer', {
+          key: 'key',
+          id: '0-0'
+        }, {
+          CLAIM: 100
+        }),
+        ['XREADGROUP', 'GROUP', 'group', 'consumer', 'CLAIM', '100', 'STREAMS', 'key', '0-0']
+      );
+    });
+
+    it('with COUNT, BLOCK, NOACK, CLAIM', () => {
+      assert.deepEqual(
+        parseArgs(XREADGROUP, 'group', 'consumer', {
+          key: 'key',
+          id: '0-0'
+        }, {
+          COUNT: 1,
+          BLOCK: 0,
+          NOACK: true,
+          CLAIM: 100
+        }),
+        ['XREADGROUP', 'GROUP', 'group', 'consumer', 'COUNT', '1', 'BLOCK', '0', 'NOACK', 'CLAIM', '100', 'STREAMS', 'key', '0-0']
+      );
+    });
   });
 
   testUtils.testAll('xReadGroup - null', async client => {
@@ -156,35 +183,54 @@ describe('XREADGROUP', () => {
     cluster: GLOBAL.CLUSTERS.OPEN
   });
 
-  testUtils.testWithClient('client.xReadGroup should throw with resp3 and unstableResp3: false', async client => {
+  testUtils.testAll('xReadGroup - without CLAIM should not include delivery fields', async client => {
-    assert.throws(
+    const [, id] = await Promise.all([
-      () => client.xReadGroup('group', 'consumer', {
+      client.xGroupCreate('key', 'group', '$', {
-        key: 'key',
+        MKSTREAM: true
-        id: '>'
       }),
-      {
+      client.xAdd('key', '*', { field: 'value' })
-        message: 'Some RESP3 results for Redis Query Engine responses may change. Refer to the readme for guidance'
+    ]);
-      }
+
-    );
+    const readGroupReply = await client.xReadGroup('group', 'consumer', {
+      key: 'key',
+      id: '>'
+    });
+
+    assert.ok(readGroupReply);
+    assert.equal(readGroupReply[0].messages[0].millisElapsedFromDelivery, undefined);
+    assert.equal(readGroupReply[0].messages[0].deliveriesCounter, undefined);
   }, {
-    ...GLOBAL.SERVERS.OPEN,
+    client: GLOBAL.SERVERS.OPEN,
-    clientOptions: {
+    cluster: GLOBAL.CLUSTERS.OPEN
-      RESP: 3
-    }
   });
 
-  testUtils.testWithClient('client.xReadGroup should not throw with resp3 and unstableResp3: true', async client => {
+  testUtils.testWithClientIfVersionWithinRange([[8,4], 'LATEST'],'xReadGroup - with CLAIM should include delivery fields', async client => {
-    assert.doesNotThrow(
+    const [, id] = await Promise.all([
-      () => client.xReadGroup('group', 'consumer', {
+      client.xGroupCreate('key', 'group', '$', {
-        key: 'key',
+        MKSTREAM: true
-        id: '>'
+      }),
-      })
+      client.xAdd('key', '*', { field: 'value' })
-    );
+    ]);
-  }, {
+
-    ...GLOBAL.SERVERS.OPEN,
+    // First read to add message to PEL
-    clientOptions: {
+    await client.xReadGroup('group', 'consumer', {
-      RESP: 3,
+      key: 'key',
-      unstableResp3: true
+      id: '>'
-    }
+    });
-  });
+
+    // Read with CLAIM to get delivery fields
+    const readGroupReply = await client.xReadGroup('group', 'consumer2', {
+      key: 'key',
+      id: '>'
+    }, {
+      CLAIM: 0
+    });
+
+    assert.ok(readGroupReply);
+    assert.equal(readGroupReply[0].messages[0].id, id);
+    assert.ok(readGroupReply[0].messages[0].millisElapsedFromDelivery !== undefined);
+    assert.ok(readGroupReply[0].messages[0].deliveriesCounter !== undefined);
+    assert.equal(typeof readGroupReply[0].messages[0].millisElapsedFromDelivery, 'number');
+    assert.equal(typeof readGroupReply[0].messages[0].deliveriesCounter, 'number');
+  }, GLOBAL.SERVERS.OPEN);
 });

packages/client/lib/commands/XREADGROUP.ts

@@ -9,11 +9,13 @@ import { transformStreamsMessagesReplyResp2 } from './generic-transformers';
  * @property COUNT - Limit the number of entries returned per stream
  * @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;
   BLOCK?: number;
   NOACK?: boolean;
+  CLAIM?: number;
 }
 
 export default {
@@ -50,6 +52,10 @@ export default {
       parser.push('NOACK');
     }
 
+    if (options?.CLAIM !== undefined) {
+      parser.push('CLAIM', options.CLAIM.toString());
+    }
+
     pushXReadStreams(parser, streams);
   },
   /**
@@ -59,5 +65,4 @@ export default {
     2: transformStreamsMessagesReplyResp2,
     3: undefined as unknown as () => ReplyUnion
   },
-  unstableResp3: true,
 } as const satisfies Command;

packages/client/lib/commands/generic-transformers.ts

@@ -514,19 +514,25 @@ export function parseArgs(command: Command, ...args: Array<any>): CommandArgumen
 
 export type StreamMessageRawReply = TuplesReply<[
   id: BlobStringReply,
-  message: ArrayReply<BlobStringReply>
+  message: ArrayReply<BlobStringReply>,
+  millisElapsedFromDelivery?: NumberReply,
+  deliveriesCounter?: NumberReply
 ]>;
 
 export type StreamMessageReply = {
   id: BlobStringReply,
   message: MapReply<BlobStringReply | string, BlobStringReply>,
+  millisElapsedFromDelivery?: number
+  deliveriesCounter?: number
 };
 
 export function transformStreamMessageReply(typeMapping: TypeMapping | undefined, reply: StreamMessageRawReply): StreamMessageReply {
-  const [ id, message ] = reply as unknown as UnwrapReply<typeof reply>;
+  const [ id, message, millisElapsedFromDelivery, deliveriesCounter ] = reply as unknown as UnwrapReply<typeof reply>;
   return {
     id: id,
-    message: transformTuplesReply(message, undefined, typeMapping)
+    message: transformTuplesReply(message, undefined, typeMapping),
+    ...(millisElapsedFromDelivery !== undefined ? { millisElapsedFromDelivery: Number(millisElapsedFromDelivery) } : {}),
+    ...(deliveriesCounter !== undefined ? { deliveriesCounter: Number(deliveriesCounter) } : {})
   };
 }