comet-ml
diff --git a/‎apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java‎
Lines changed: 65 additions & 2 deletions b/‎apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java‎
Lines changed: 65 additions & 2 deletions
diff --git a/‎apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberTest.java‎
Lines changed: 103 additions & 0 deletions b/‎apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberTest.java‎
Lines changed: 103 additions & 0 deletions
diff --git a/‎apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberUnitTest.java‎
Lines changed: 109 additions & 0 deletions b/‎apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberUnitTest.java‎
Lines changed: 109 additions & 0 deletions
@@ -127,6 +127,27 @@ public abstract class BaseRedisSubscriber<M> implements Managed {
     private final LongHistogram listPendingTime;
     private final LongCounter unexpectedErrors;
 
+    /**
+     * Resume point for the next {@code XAUTOCLAIM} scan.
+     *
+     * <p>Redis caps each {@code XAUTOCLAIM} at {@code COUNT * 10} PEL entries <em>examined</em> (not
+     * claimed), so with the default {@code consumerBatchSize} of 10 a single call only ever inspects the
+     * first 100 entries of the pending list. Restarting every scan at {@link StreamMessageId#MIN} —
+     * which this class used to do, discarding the cursor Redis hands back — means anything past that
+     * first window is never examined at all: a backlog above ~100 entries develops a permanently
+     * unreachable tail, whatever the retry or decode behaviour further down. Carrying the returned
+     * cursor forward walks the whole PEL instead, a window at a time.
+     *
+     * <p>Redis returns {@code 0-0} once a pass has covered the entire pending list; that resets this
+     * back to {@link StreamMessageId#MIN} so the next pass starts over from the oldest entry, keeping
+     * the oldest-first bias the original code intended.
+     *
+     * <p>Volatile rather than synchronized: {@code concatMap} in {@link #setupStreamListener()} already
+     * serializes claim calls, so this is only ever written by one claim at a time; volatile just
+     * publishes that write to whichever scheduler thread runs the next one.
+     */
+    private volatile StreamMessageId claimCursor = StreamMessageId.MIN;
+
     private volatile RStreamReactive<String, M> stream;
     private volatile Disposable streamSubscription;
     private volatile Scheduler timerScheduler;
@@ -398,11 +419,11 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 consumerId,
                 config.getPendingMessageDuration().toJavaDuration().toMillis(),
                 TimeUnit.MILLISECONDS,
-                StreamMessageId.MIN, // Start from the beginning of pending list
+                claimCursor, // Resume where the previous scan stopped; see claimCursor
                 config.getConsumerBatchSize())
                 .subscribeOn(consumerScheduler)
                 .filter(Objects::nonNull)
-                .map(AutoClaimResult::getMessages)
+                .map(this::advanceCursorAndExtractMessages)
                 .filter(Objects::nonNull)
                 .doOnSuccess(claimedMessages -> {
                     claimedMessages = Objects.requireNonNullElse(claimedMessages, Map.of());
@@ -412,6 +433,9 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 .onErrorResume(throwable -> {
                     claimErrors.add(1);
                     log.error("Error claiming pending messages", throwable);
+                    // A failed scan leaves the cursor where it was, so the next attempt retries the same
+                    // window rather than skipping it. The NOGROUP case is different, but recoverFromNoGroup
+                    // resets the cursor itself so both it and the read path get the same treatment.
                     if (isNoGroupError(throwable)) {
                         return recoverFromNoGroup();
                     }
@@ -420,6 +444,35 @@ private Mono<Map<StreamMessageId, Map<String, M>>> claimPendingMessages() {
                 .doFinally(signalType -> claimTime.record(System.currentTimeMillis() - startMillis));
     }
 
+    /**
+     * Advances {@link #claimCursor} to where Redis says the next scan should resume, then returns the
+     * batch this scan actually claimed.
+     *
+     * <p>A {@code null} or {@code 0-0} next-id means the pass reached the end of the pending list, so the
+     * cursor goes back to {@link StreamMessageId#MIN} and the following pass starts from the oldest entry
+     * again.
+     */
+    private Map<StreamMessageId, Map<String, M>> advanceCursorAndExtractMessages(AutoClaimResult<String, M> result) {
+        claimCursor = nextCursor(result.getNextId());
+        return result.getMessages();
+    }
+
+    /**
+     * The cursor to use for the next scan, given the next-id Redis returned.
+     *
+     * <p>Compared numerically rather than against {@link StreamMessageId#MIN} / {@link StreamMessageId#ALL}:
+     * those are wire sentinels that serialize to {@code -} and {@code 0}, and neither is
+     * {@code equals()} to the {@code StreamMessageId(0, 0)} that Redisson parses Redis's literal
+     * {@code 0-0} end-of-pass reply into. Matching on the constants would therefore never fire, and the
+     * scan would run off the end of the PEL and stay there instead of wrapping.
+     */
+    private static StreamMessageId nextCursor(StreamMessageId nextId) {
+        if (nextId == null || (nextId.getId0() == 0 && nextId.getId1() == 0)) {
+            return StreamMessageId.MIN;
+        }
+        return nextId;
+    }
+
     private Mono<Map<StreamMessageId, Map<String, M>>> readMessages() {
         var startMillis = System.currentTimeMillis();
         var streamReadGroupArgs = StreamReadGroupArgs.neverDelivered()
@@ -453,6 +506,16 @@ private boolean isNoGroupError(Throwable throwable) {
     private Mono<Map<StreamMessageId, Map<String, M>>> recoverFromNoGroup() {
         log.warn("Recreating not found consumer group '{}' for stream '{}'",
                 config.getConsumerGroupName(), config.getStreamName());
+        // The cursor indexes the OLD group's pending list, so it means nothing once the group is
+        // recreated. Reset here rather than at the call sites: NOGROUP surfaces from both readMessages
+        // and claimPendingMessages, and the read path is the likelier of the two to notice it first
+        // (reads run on every tick that is not a claim tick).
+        //
+        // Leaving it stale is not self-correcting on a busy stream. A scan starting above the recreated
+        // PEL's entries only wraps once it exhausts the list, and entries arriving after the stale
+        // position keep giving it work at the high end -- so the wrap can be deferred indefinitely while
+        // the oldest entries, the ones this scan exists to reach, are never examined.
+        claimCursor = StreamMessageId.MIN;
         return createConsumerGroup()
                 .onErrorResume(throwable -> {
                     log.error("Failed to recreate consumer group '{}' for stream '{}'",
 
@@ -44,6 +44,7 @@
 import java.util.concurrent.CopyOnWriteArraySet;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.stream.IntStream;
 import java.util.stream.Stream;
 
 import static com.comet.opik.api.resources.utils.TestUtils.waitForMillis;
@@ -429,6 +430,108 @@ void shouldRecoverFromNoGroupOnReadAndContinueProcessing() {
         }
     }
 
+    /**
+     * OPIK-8240, against a real Redis PEL deeper than one {@code XAUTOCLAIM} scan window.
+     * <p>
+     * Redis caps each {@code XAUTOCLAIM} at {@code COUNT * 10} pending entries <em>examined</em> — not
+     * claimed — and answers with the position the next scan should resume from. Discarding that position
+     * and restarting every scan at {@link StreamMessageId#MIN} is invisible while the head of the PEL
+     * drains, because the entries that leave make room in the window. It stops being invisible as soon
+     * as the head stops draining: the budget is then spent re-examining the same first {@code COUNT * 10}
+     * entries and everything behind them is never looked at again, whatever the retry behaviour further
+     * down.
+     * <p>
+     * Both halves of the cursor contract are observable here, and only on a real Redis: the reply that
+     * this class consumes is produced by Redis's own PEL scan, not by anything the subscriber controls.
+     */
+    @Nested
+    class ClaimCursorTests {
+
+        // Gives XAUTOCLAIM a 100-entry examine budget per scan (COUNT * 10).
+        private static final int CLAIM_BATCH_SIZE = 10;
+        private static final int SCAN_WINDOW = CLAIM_BATCH_SIZE * 10;
+        // Deliberately deeper than one window: the last 50 are the entries a cursor-less scan strands.
+        private static final int BACKLOG_SIZE = SCAN_WINDOW + 50;
+        private static final int CLAIM_TIMEOUT_SECONDS = 30;
+
+        private TestStreamConfiguration deepBacklogConfig;
+        private RStreamReactive<String, String> deepBacklogStream;
+
+        @BeforeEach
+        void setUp() {
+            deepBacklogConfig = TestStreamConfiguration.create().toBuilder()
+                    .consumerBatchSize(CLAIM_BATCH_SIZE)
+                    // Every poll claims. Nothing is published after start(), so a read would only park the
+                    // concatMap on its long poll and stretch the run out for no coverage.
+                    .claimIntervalRatio(1)
+                    .pendingMessageDuration(io.dropwizard.util.Duration.milliseconds(100))
+                    // High enough that nothing retires mid-run. Retirement removes entries from the PEL,
+                    // which would let even a cursor-less scan crawl to the tail eventually and pass this
+                    // test for the wrong reason.
+                    .maxRetries(Integer.MAX_VALUE)
+                    .build();
+            deepBacklogStream = redissonClient.getStream(
+                    deepBacklogConfig.getStreamName(), deepBacklogConfig.getCodec());
+            deepBacklogStream.delete().block();
+        }
+
+        @Test
+        void shouldReachPendingEntriesBeyondTheFirstScanWindowAndThenWrap() {
+            var messages = IntStream.range(0, BACKLOG_SIZE)
+                    .mapToObj(index -> "backlog-%03d-%s".formatted(index, UUID.randomUUID()))
+                    .toList();
+            deepBacklogStream.createGroup(
+                    StreamCreateGroupArgs.name(deepBacklogConfig.getConsumerGroupName()).makeStream()).block();
+
+            // concatMap, not flatMap: stream ids must follow publication order for "the head of the PEL"
+            // to mean the first element of this list.
+            Flux.fromIterable(messages)
+                    .concatMap(message -> deepBacklogStream.add(
+                            StreamAddArgs.entry(TestStreamConfiguration.PAYLOAD_FIELD, message)))
+                    .collectList()
+                    .block();
+
+            // Delivered to a consumer that never acks, which is the backlog a crashed or restarted pod
+            // leaves behind: XLEN and the PEL both hold the full depth.
+            var crashedConsumerId = "crashed-consumer-%s".formatted(UUID.randomUUID());
+            var delivered = deepBacklogStream.readGroup(
+                    deepBacklogConfig.getConsumerGroupName(), crashedConsumerId,
+                    StreamReadGroupArgs.neverDelivered()
+                            .count(BACKLOG_SIZE)
+                            .timeout(deepBacklogConfig.getLongPollingDuration().toJavaDuration()))
+                    .block();
+            assertThat(delivered).hasSize(BACKLOG_SIZE);
+
+            // Idle for longer than min-idle-time, so every entry is claimable rather than skipped.
+            waitForMillis(deepBacklogConfig.getPendingMessageDuration().toMilliseconds() + 100);
+
+            // Every delivery fails retryably, so nothing is acked and the PEL keeps its full depth for the
+            // whole run — the condition that makes the scan window a ceiling rather than a batch size.
+            var deliveries = new ConcurrentHashMap<String, AtomicInteger>();
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(
+                    deepBacklogConfig, redissonClient, message -> {
+                        deliveries.computeIfAbsent(message, key -> new AtomicInteger()).incrementAndGet();
+                        return Mono.error(new RuntimeException("Retryable, so the entry stays pending"));
+                    }));
+            subscriber.start();
+
+            // The tail is reached. Without the cursor this stalls at the first SCAN_WINDOW entries: each
+            // scan restarts at MIN, spends its whole examine budget on the head, and returns having never
+            // looked further.
+            await().atMost(CLAIM_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(deliveries.keySet())
+                            .containsExactlyInAnyOrderElementsOf(messages));
+
+            // And the scan wraps instead of parking at the end of the PEL. Redis answers a completed pass
+            // with 0-0, which resets the cursor to the oldest entry; without that reset the oldest entry
+            // would never be delivered a second time and the subscriber would go quiet on a PEL that is
+            // still full.
+            await().atMost(CLAIM_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(deliveries.get(messages.getFirst()).get())
+                            .isGreaterThan(1));
+        }
+    }
+
     @Nested
     class RetryTests {
 
 
@@ -622,6 +622,58 @@ void shouldHandleInvalidMessageIdTimestamp() {
     @Nested
     class NoGroupErrorTests {
 
+        /**
+         * Regression for review feedback on OPIK-8240: {@code recoverFromNoGroup} is reached from BOTH
+         * {@code readMessages} and {@code claimPendingMessages}, and the cursor reset was originally only
+         * at the claim call site. The read path is the likelier of the two to notice NOGROUP first, since
+         * reads run on every tick that is not a claim tick.
+         *
+         * <p>A cursor left pointing into the old group's pending list is not self-correcting on a busy
+         * stream: a scan starting above the recreated PEL's entries only wraps once it exhausts the list,
+         * and entries arriving after the stale position keep feeding it at the high end, so the oldest
+         * entries can go unexamined indefinitely -- the exact starvation this fix exists to remove.
+         */
+        @Test
+        void shouldResetTheClaimCursorWhenTheGroupIsRecreatedViaTheReadPath() {
+            whenCreateGroupReturnEmpty();
+            whenRemoveConsumerReturn();
+            var fastConfig = CONFIG.toBuilder().claimIntervalRatio(2).build();
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(fastConfig, redissonClient));
+            var starts = new CopyOnWriteArrayList<StreamMessageId>();
+            var claims = new AtomicInteger();
+
+            // First claim advances the cursor well past the start of the PEL.
+            when(stream.autoClaim(
+                    eq(fastConfig.getConsumerGroupName()),
+                    anyString(),
+                    eq(fastConfig.getPendingMessageDuration().toJavaDuration().toMillis()),
+                    eq(TimeUnit.MILLISECONDS),
+                    any(StreamMessageId.class),
+                    eq(fastConfig.getConsumerBatchSize())))
+                    .thenAnswer(invocation -> {
+                        starts.add(invocation.getArgument(4));
+                        claims.incrementAndGet();
+                        return Mono.just(new AutoClaimResult<>(
+                                new StreamMessageId(9_000L, 0), Map.of(), List.of()));
+                    });
+            // Reads then hit NOGROUP, which recreates the group underneath us.
+            when(stream.readGroup(eq(fastConfig.getConsumerGroupName()), anyString(),
+                    any(StreamReadGroupArgs.class)))
+                    .thenAnswer(invocation -> Mono
+                            .error(new RuntimeException("NOGROUP No such key stream or consumer group")));
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).until(() -> starts.size() >= 2);
+
+            // The recreated group has a fresh PEL, so the next scan must start from the beginning rather
+            // than from the position it reached in the group that no longer exists.
+            assertThat(starts.getFirst()).isEqualTo(StreamMessageId.MIN);
+            assertThat(starts.get(1)).isEqualTo(StreamMessageId.MIN);
+        }
+
         @Test
         void shouldRecoverOnClaimAndNotDie() {
             whenCreateGroupReturnEmpty();
@@ -767,6 +819,63 @@ private static PendingEntry pendingEntry(StreamMessageId messageId, long deliver
         return entry;
     }
 
+    /**
+     * OPIK-8240, for the one part of the claim cursor a real Redis cannot be made to produce on demand:
+     * a failing {@code XAUTOCLAIM}.
+     * <p>
+     * The cursor's normal behaviour -- walking past the first {@code COUNT * 10} examine window and
+     * wrapping when Redis reports the pass complete -- is covered end to end in
+     * {@link BaseRedisSubscriberTest.ClaimCursorTests} against a PEL deeper than that window, so it is
+     * deliberately not duplicated here. What is left is the error path: the cursor must NOT advance past
+     * a window whose scan failed, or that window's entries are skipped until the next full wrap. Making
+     * a real Redis fail one XAUTOCLAIM mid-run and succeed on the next is not something the container
+     * exposes, so it is mocked.
+     */
+    @Nested
+    class ClaimCursorTests {
+
+        @BeforeEach
+        void setUp() {
+            whenCreateGroupReturnEmpty();
+            whenRemoveConsumerReturn();
+        }
+
+        @Test
+        void shouldRetryTheSameWindowAfterAFailedScan() {
+            var fastConfig = CONFIG.toBuilder().claimIntervalRatio(2).build();
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(fastConfig, redissonClient));
+            var starts = new CopyOnWriteArrayList<StreamMessageId>();
+            var calls = new AtomicInteger();
+            when(stream.autoClaim(
+                    eq(fastConfig.getConsumerGroupName()),
+                    anyString(),
+                    eq(fastConfig.getPendingMessageDuration().toJavaDuration().toMillis()),
+                    eq(TimeUnit.MILLISECONDS),
+                    any(StreamMessageId.class),
+                    eq(fastConfig.getConsumerBatchSize())))
+                    .thenAnswer(invocation -> {
+                        starts.add(invocation.getArgument(4));
+                        if (calls.incrementAndGet() == 1) {
+                            return Mono.just(new AutoClaimResult<>(
+                                    new StreamMessageId(700L, 0), Map.of(), List.of()));
+                        }
+                        // A failed scan must not advance the cursor, or the window it covered would be
+                        // skipped entirely and its entries left unreachable until the next full wrap.
+                        return Mono.error(new RuntimeException("Redis autoClaim error"));
+                    });
+            whenReadGroupReturnMessages();
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).until(() -> starts.size() >= 3);
+
+            assertThat(starts.get(1)).isEqualTo(new StreamMessageId(700L, 0));
+            assertThat(starts.get(2)).isEqualTo(new StreamMessageId(700L, 0));
+        }
+    }
+
     private TestRedisSubscriber trackSubscriber(TestRedisSubscriber subscriber) {
         subscribers.add(subscriber);
         return subscriber;