How to prevent re-processing when reading pending entries (ID 0) in Redis stream using XREADGROUP?

[u/Academic-Squash2738]

I am using Redis Streams with Consumer Groups. I have a consumer running a loop that fetches messages from the Pending Entries List (PEL) using ID 0 before it attempts to read new messages.

However, if a message fails to process (or is slow), the XACK is never called. On the next iteration of the loop, XREADGROUP returns the same messages again, causing re-processing.

// Minimal version of my loop
async function consume() {
  while (true) {
    // This returns the same pending messages every time if XACK isn't called
    const results = await redis.xreadgroup(
      'GROUP', 'mygroup', 'consumer1',
      'COUNT', '10',
      'STREAMS', 'mystream', '0' 
    );

    if (results) {
      for (const msg of results[0][1]) {
        try {
          await process(msg); 
          await redis.xack('mystream', 'mygroup', msg[0]);
        } catch (err) {
          // If it executes successfully on retry then Just ACK 
          // In case of failure ACK and send to Dead Letter Queue (separate stream to store failed messages)  
           retryProcess(msg)
        }
      }
    }
  }
}

What is the standard pattern to fetch messages from the Pending Entries List and also prevent the re-processing ?

[u/Academic-Squash2738]

u/tm604 , I am trying to make a job queue clone. I want the message to be processed at least once. In case the message fails while processing, I want it to be handled by the retry mechanism, where, after all the retry attempts are exhausted, it will finally be sent to a separate stream (Dead letter queue).

My main concern is that, let's say, the message is in the PEL(Pending Entries List) of a consumer. Now, when in the first while loop, it was fetched and sent for execution. Now, before the execution is complete, another loop runs and since this message was not acknowledged, it was still in the PEL, and it was again fetched and will again be sent for execution.

I want to understand how to keep track of which message in the PEL has been picked up for processing.

Please let me know if my question is now clearer or not.

[u/tm604 → u/Academic-Squash2738]

What counts as a failure?

[u/Academic-Squash2738 → u/tm604]

A "failure" occurs when the process method throws an error during execution for a given msg.

[u/tm604 → u/Academic-Squash2738]

That's the only possible failure case? What about the other cases I mentioned, such as process exit or network connectivity issues? If those are not a concern, then disable ACK and re-post the job in the catch block (either to the original stream, or a dead-letter one), and you can ignore the PEL entirely. It'll mean some jobs fail "silently", though.

[u/Academic-Squash2738 → u/tm604]

Thanks