[OPIK-8240] [BE] fix: split provider-error retryability by status, resume XAUTOCLAIM from its cursor

Two independent defects that each leave a permanently failing online-scoring
message cycling instead of retiring it.

1. A permanent provider 4xx was retried as if transient.

ChatCompletionService.scoreTrace answered every unmappable provider failure
with a blanket InternalServerErrorException, which sits outside
BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS -- so a request that can never
succeed was replayed maxRetries times, once per pendingMessageDuration. The
real status was available the whole time on the HttpException in the cause
chain; it went unread because the existing recovery path classifies by family,
and mapping all of 4xx to ClientErrorException would drop transient 408/429
after one attempt. Hence the blanket 500.

Classify by status code instead: 400/401/403 and the rest of 4xx are
permanent, 408/425/429 and all 5xx stay retryable. scoreTrace gets its own
mapping so create() and the streaming handler keep returning the provider's
status verbatim to HTTP callers. This also closes a pre-existing over-eager
drop on the parseable branch, where a JSON-parseable 429 already became
ClientErrorException and was dropped after one attempt.

2. XAUTOCLAIM never scanned past the first ~100 pending entries.

claimPendingMessages passed StreamMessageId.MIN as the scan start every call
and discarded the cursor Redis returns. Redis caps each XAUTOCLAIM at
COUNT * 10 PEL entries *examined*, so at consumerBatchSize=10 a call inspects
only the first 100 -- and restarting at MIN means nothing past that window is
ever examined. Any backlog above ~100 grows a permanently unreachable tail,
whatever the retry or decode behaviour downstream.

Carry getNextId() forward, resetting to MIN on Redis's 0-0 end-of-pass reply
so the oldest-first bias is preserved. A failed scan deliberately does not
advance the cursor, so its window is retried rather than skipped.

The end-of-pass sentinel is compared numerically, not against
StreamMessageId.MIN/.ALL: those are wire sentinels serializing to "-" and "0",
and neither is equals() to the StreamMessageId(0, 0) Redisson parses 0-0 into
(verified against redisson 4.7.0), so matching on the constants would never
fire and the scan would park at the end of the PEL.

Tests: 132 green across the affected suites. Both fixes mutation-checked --
collapsing the status split to the whole 4xx family fails 5 tests, removing it
entirely fails 7; reverting the cursor to always-MIN fails 3.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: thiagohora

apps/opik-backend/src/main/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriber.java

@@ -124,6 +124,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;
@@ -387,11 +408,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());
@@ -401,14 +422,48 @@ 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. Except on NOGROUP: the group is being recreated, so
+                    // any cursor into the old group's PEL is meaningless.
                     if (isNoGroupError(throwable)) {
+                        claimCursor = StreamMessageId.MIN;
                         return recoverFromNoGroup();
                     }
                     return Mono.just(Map.of());
                 })
                 .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.
+     */
+    // Package-private for unit tests.
+    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()

apps/opik-backend/src/main/java/com/comet/opik/domain/llm/ChatCompletionService.java

@@ -36,6 +36,7 @@
 import java.nio.channels.ClosedChannelException;
 import java.util.List;
 import java.util.Optional;
+import java.util.Set;
 import java.util.concurrent.Callable;
 import java.util.function.Consumer;
 
@@ -147,16 +148,32 @@ public ChatResponse scoreTrace(@NonNull ChatRequest chatRequest,
 
             Optional<ErrorMessage> providerError = provider.getLlmProviderError(runtimeException);
 
-            providerError
-                    .ifPresent(llmProviderError -> failHandlingLLMProviderError(runtimeException, llmProviderError));
+            // Deliberately NOT failHandlingLLMProviderError here, unlike create() and the streaming handler.
+            // Those two answer an HTTP caller, so they want the provider's status verbatim - a 429 upstream
+            // should be a 429 downstream. This method answers the online-scoring subscribers instead, where
+            // the thrown type decides retryability: BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS matches on
+            // ClientErrorException, so the whole 4xx family would be dropped on the first failure - taking
+            // genuinely transient 408s and 429s with it. Route through the retryability split instead, which
+            // separates "this request can never succeed" from "try again later" inside that family.
+            int status = providerError.map(ErrorMessage::getCode)
+                    .filter(ChatCompletionService::isErrorStatus)
+                    .or(() -> findProviderHttpStatus(runtimeException))
+                    .orElse(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode());
 
-            // No failIfProviderReportedHttpStatus here, unlike create() and the streaming handler. This method is
-            // called only by the online-scoring subscribers, never from a resource, so a recovered status reaches no
-            // HTTP client — while BaseRedisSubscriber.NON_RETRYABLE_EXCEPTIONS lists ClientErrorException, so turning
-            // a rate limit into a 429 or a provider timeout into a 408 would make the subscriber ack and drop the
-            // evaluation instead of honouring onlineScoring.maxRetries. Both are RetriableException upstream, so the
-            // blanket 500 is what keeps them retryable.
             log.warn(UNEXPECTED_ERROR_CALLING_LLM_PROVIDER, runtimeException);
+
+            if (isPermanentFailure(status)) {
+                // Non-retryable: the subscriber acks and removes on the first delivery instead of burning
+                // maxRetries x pendingMessageDuration on a request whose outcome cannot change. This is the
+                // OPIK-8193 case - a provider rejecting an oversized/invalid request rejects every retry of
+                // it identically.
+                throw new ClientErrorException(buildDetailedErrorMessage(runtimeException), status,
+                        runtimeException);
+            }
+            // Everything else stays a blanket 500 so it lands outside NON_RETRYABLE_EXCEPTIONS and honours
+            // onlineScoring.maxRetries. Note this keeps transient 4xx (408/429) retryable even though their
+            // real status is a client error, which is exactly the distinction ClientErrorException cannot
+            // express on its own.
             throw new InternalServerErrorException(buildDetailedErrorMessage(runtimeException), runtimeException);
         } finally {
             // Close the Vertex client (reused across retries) to release its GAX threads; other providers self-reclaim.
@@ -292,6 +309,35 @@ private static boolean isErrorStatus(int status) {
         return family == Response.Status.Family.CLIENT_ERROR || family == Response.Status.Family.SERVER_ERROR;
     }
 
+    /**
+     * Statuses in the 4xx family that a retry could still clear, so they must NOT be treated as permanent.
+     *
+     * <p>{@code 408 Request Timeout}, {@code 425 Too Early} and {@code 429 Too Many Requests} all say
+     * "not now", not "not ever" - langchain4j models the latter two as {@code RetriableException} upstream.
+     * They sit in the client-error family purely by HTTP numbering, which is why family alone is too blunt
+     * a discriminator for retryability.
+     */
+    private static final Set<Integer> TRANSIENT_CLIENT_ERRORS = Set.of(
+            Response.Status.REQUEST_TIMEOUT.getStatusCode(),
+            425, // Too Early - no jakarta.ws.rs.core.Response.Status constant
+            429); // Too Many Requests - no jakarta.ws.rs.core.Response.Status constant
+
+    /**
+     * Whether a provider status means "this exact request can never succeed", so the caller should give up
+     * rather than retry.
+     *
+     * <p>True only for the client-error family minus {@link #TRANSIENT_CLIENT_ERRORS}: a malformed,
+     * oversized, unauthorised or forbidden request is rejected identically however many times it is
+     * replayed. Server errors are excluded - a 5xx is the provider having a bad moment, and that is the
+     * textbook retry case. An unrecognised status is treated as retryable, matching
+     * {@code BaseRedisSubscriber}'s own "unknown defaults to retryable for safety" stance.
+     */
+    // Package-private for unit tests.
+    static boolean isPermanentFailure(int status) {
+        return familyOf(status) == Response.Status.Family.CLIENT_ERROR
+                && !TRANSIENT_CLIENT_ERRORS.contains(status);
+    }
+
     /**
      * The status langchain4j's own exception types stand for, used for providers whose clients raise them without an
      * {@link HttpException} in the chain. {@code ContentFilteredException} is covered by its

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberCursorTest.java

@@ -0,0 +1,44 @@
+package com.comet.opik.api.resources.v1.events;
+
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.redisson.api.stream.StreamMessageId;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * Pure-function coverage for {@link BaseRedisSubscriber#nextCursor(StreamMessageId)}, the mapping that
+ * decides where the next {@code XAUTOCLAIM} scan resumes (OPIK-8193).
+ *
+ * <p>Separate from {@link BaseRedisSubscriberUnitTest} because these need no Redis mocks at all, and
+ * living under that class's Mockito setup would trip strict stubbing.
+ */
+@DisplayName("BaseRedisSubscriber claim-cursor mapping")
+class BaseRedisSubscriberCursorTest {
+
+    @Test
+    @DisplayName("Redis's 0-0 end-of-pass reply wraps back to the start of the pending list")
+    void endOfPassWrapsToStart() {
+        // Compared numerically on purpose. StreamMessageId.MIN and .ALL are wire sentinels that serialize
+        // to "-" and "0", and NEITHER is equals() to the StreamMessageId(0, 0) that Redisson parses Redis's
+        // literal 0-0 into -- verified against redisson 4.7.0. Matching on the constants would therefore
+        // never fire, the cursor would stick at the end of the PEL, and the scan would stop finding
+        // anything at all: strictly worse than the always-MIN behaviour this replaces.
+        assertThat(BaseRedisSubscriber.nextCursor(new StreamMessageId(0, 0))).isEqualTo(StreamMessageId.MIN);
+    }
+
+    @Test
+    @DisplayName("A missing next-id is treated as end-of-pass rather than propagating a null")
+    void nullNextIdWrapsToStart() {
+        assertThat(BaseRedisSubscriber.nextCursor(null)).isEqualTo(StreamMessageId.MIN);
+    }
+
+    @Test
+    @DisplayName("A mid-scan position is carried forward unchanged, which is what walks past the first window")
+    void midScanPositionIsCarriedForward() {
+        // Redis caps each XAUTOCLAIM at COUNT * 10 entries EXAMINED, so without carrying this forward the
+        // scan only ever sees the first ~100 pending entries and anything behind them is never reclaimed.
+        var midScan = new StreamMessageId(1787891543627L, 0);
+        assertThat(BaseRedisSubscriber.nextCursor(midScan)).isSameAs(midScan);
+    }
+}

apps/opik-backend/src/test/java/com/comet/opik/api/resources/v1/events/BaseRedisSubscriberUnitTest.java

@@ -604,6 +604,129 @@ void shouldHandleRemoveConsumerTimeoutOnStop() {
         }
     }
 
+    /**
+     * OPIK-8193. Redis caps each {@code XAUTOCLAIM} at {@code COUNT * 10} PEL entries <em>examined</em>,
+     * so a scan that always restarts at {@link StreamMessageId#MIN} only ever inspects the first ~100
+     * pending entries. A backlog above that grows a tail nothing ever reclaims: verified against a real
+     * Redis, entries beyond the window sat at delivery-count 1 for weeks while the head cycled. These
+     * tests pin the cursor being carried forward, which is what walks the rest of the PEL.
+     */
+    @Nested
+    class ClaimCursorTests {
+
+        @BeforeEach
+        void setUp() {
+            whenCreateGroupReturnEmpty();
+            whenRemoveConsumerReturn();
+        }
+
+        @Test
+        void shouldResumeTheNextScanFromTheReturnedCursor() {
+            var fastConfig = CONFIG.toBuilder().claimIntervalRatio(2).build();
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(fastConfig, redissonClient));
+            var starts = new CopyOnWriteArrayList<StreamMessageId>();
+            // Never returns 0-0, so a correct implementation keeps walking forward and a regressed one
+            // (restarting at MIN) is immediately visible in the captured start arguments.
+            var cursor = 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));
+                        return Mono.just(new AutoClaimResult<>(
+                                new StreamMessageId(100L + cursor.incrementAndGet(), 0),
+                                Map.of(),
+                                List.of()));
+                    });
+            whenReadGroupReturnMessages();
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).until(() -> starts.size() >= 3);
+
+            // First scan starts at the beginning; every later scan resumes where the previous one stopped.
+            assertThat(starts.get(0)).isEqualTo(StreamMessageId.MIN);
+            assertThat(starts.get(1)).isEqualTo(new StreamMessageId(101L, 0));
+            assertThat(starts.get(2)).isEqualTo(new StreamMessageId(102L, 0));
+        }
+
+        @Test
+        void shouldWrapBackToTheStartAfterAFullPass() {
+            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));
+                        // Advance once, then report the pass complete.
+                        var nextId = calls.incrementAndGet() == 1
+                                ? new StreamMessageId(500L, 0)
+                                : new StreamMessageId(0, 0);
+                        return Mono.just(new AutoClaimResult<>(nextId, Map.of(), List.of()));
+                    });
+            whenReadGroupReturnMessages();
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS).until(() -> starts.size() >= 3);
+
+            assertThat(starts.get(0)).isEqualTo(StreamMessageId.MIN);
+            assertThat(starts.get(1)).isEqualTo(new StreamMessageId(500L, 0));
+            // 0-0 came back, so the oldest-first bias is restored rather than the scan parking at the end.
+            assertThat(starts.get(2)).isEqualTo(StreamMessageId.MIN);
+        }
+
+        @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;