test: drive a real StreamConstraintsException through a real stream end to end

You were right to push on this, and my earlier wording was wrong. I said
the Exception arm "has always handled the literal incident" -- I meant
since the first commit OF THIS PR, which introduced the wrapper. Pre-PR
there was no try/catch in the codec at all, so the StreamConstraintsException
propagated up through CommandDecoder exactly as the original OPIK-8164
analysis said. Nothing of ours caught it.

And the drop mechanism was not actually tested for a real decode failure.
Every existing test either injected a pre-built UndecodableStreamMessage
into a mocked readGroup, or exercised the decoder in isolation. None
reproduced the thing that happened: written successfully, then failing on
the way back out below the subscriber.

New BaseRedisSubscriberTest.UndecodablePayloadTests does that on real
Redis -- a JAVA-shaped codec with a 512-byte string limit standing in for
production's 100 MB, so the write/read asymmetry reproduces at a few
hundred bytes. It asserts the entry writes fine, the healthy entry behind
it still processes, the undecodable one is NOT removed on first delivery,
and that it does leave once the delivery count reaches maxRetries.

Mutation-checked, and the mutant is the incident: making tolerant() a
no-op produces

  io.netty.handler.codec.DecoderException:
    com.fasterxml.jackson.core.exc.StreamConstraintsException:
    String value length (513) exceeds the maximum allowed (512, ...)

which is the production stack trace's exact shape, and the test fails at
the FIRST await -- the healthy entry behind it never processes either.
That is the batch-stranding half of the wedge, now covered.

Two things the run exposed, both documented in the test:

- Retirement is driven by the delivery count, which only advances when
  XAUTOCLAIM redelivers after pendingMessageDuration. At the shipped
  online-scoring values (10m, maxRetries 3) an undecodable entry sits in
  the stream for ~30 minutes, re-read and re-decoded each cycle, before
  removal. Bounded, not eliminated -- and at production payload sizes each
  cycle re-attempts a large materialization. The test compresses the
  timers to 500ms/2 to stay fast.
- My first attempt at this test published before subscriber.start(), so
  the consumer group was created at '$' and never saw the entries. That
  was a test bug, not a code one, but it is why the ordering is now
  explicit with a comment.

RedisStreamCodec.faultTolerant widens from package-private to public
@VisibleForTesting so the subscriber test package can build the codec.

59 green, 0 skipped.

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

Author: thiagohora

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/redis/RedisStreamCodec.java

@@ -63,7 +63,7 @@ static ObjectMapper buildStreamMapper() {
      * directions: a pod on either build writes what the other can read.
      */
     @VisibleForTesting
-    static Codec faultTolerant(Codec delegate) {
+    public static Codec faultTolerant(Codec delegate) {
         return new FaultTolerantCodec(delegate);
     }
 

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

@@ -1,8 +1,10 @@
 package com.comet.opik.api.resources.v1.events;
 
 import com.comet.opik.api.resources.utils.RedisContainerUtils;
+import com.comet.opik.infrastructure.redis.RedisStreamCodec;
 import com.comet.opik.podam.PodamFactoryUtils;
 import com.comet.opik.utils.JsonUtils;
+import com.fasterxml.jackson.core.StreamReadConstraints;
 import com.redis.testcontainers.RedisContainer;
 import jakarta.ws.rs.ClientErrorException;
 import jakarta.ws.rs.NotFoundException;
@@ -22,7 +24,10 @@
 import org.redisson.api.stream.StreamCreateGroupArgs;
 import org.redisson.api.stream.StreamMessageId;
 import org.redisson.api.stream.StreamReadGroupArgs;
+import org.redisson.client.codec.Codec;
+import org.redisson.codec.CompositeCodec;
 import org.redisson.codec.JsonJacksonCodec;
+import org.redisson.codec.LZ4CodecV2;
 import org.redisson.config.Config;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
@@ -89,6 +94,99 @@ void tearDown() {
         subscribers.forEach(BaseRedisSubscriber::stop);
     }
 
+    /**
+     * The OPIK-8164 mechanism end to end, on real Redis, with a real decode failure.
+     * <p>
+     * Every other drop test in this PR either injects a pre-built {@code UndecodableStreamMessage} into
+     * a mocked {@code readGroup}, or exercises the decoder in isolation. Neither reproduces the thing
+     * that actually happened: the payload is written successfully, because Jackson has no
+     * serialization-side limit, and then breaches {@code maxStringLength} on the way back out — inside
+     * Redisson's {@code CommandDecoder}, below {@code BaseRedisSubscriber}, which pre-PR is exactly why
+     * the entry could never be acked and the stream wedged permanently.
+     * <p>
+     * A small {@code maxStringLength} stands in for production's 100 MB so the asymmetry is reproduced
+     * at a few hundred bytes instead of tens of megabytes. The write path is unbounded either way, so
+     * the mechanism is identical.
+     * <p>
+     * Worth knowing what this run exposed about timing, because the retry timers are compressed here
+     * and are not in production. Retirement is driven by the delivery count, which only advances when
+     * {@code XAUTOCLAIM} redelivers the entry after {@code pendingMessageDuration}. At the shipped
+     * online-scoring values ({@code pendingMessageDuration: 10m}, {@code maxRetries: 3}) an undecodable
+     * entry therefore stays in the stream for roughly 30 minutes, being re-read and re-decoded on each
+     * cycle, before it is removed. The wedge is bounded rather than eliminated — and for a payload of
+     * production size, each cycle re-attempts a large materialization. That is the cost of not deleting
+     * on first delivery; it is the right trade while a newer pod might still decode the entry, but it
+     * is not free.
+     */
+    @Nested
+    class UndecodablePayloadTests {
+
+        private static final int SMALL_STRING_LIMIT = 512;
+
+        private TestStreamConfiguration smallLimitConfig;
+        private RStreamReactive<String, String> smallLimitStream;
+
+        /** The JAVA codec's shape, but with a string limit small enough to breach cheaply. */
+        private Codec smallLimitCodec() {
+            var mapper = JsonUtils.getMapper().copy();
+            mapper.getFactory().setStreamReadConstraints(
+                    StreamReadConstraints.builder().maxStringLength(SMALL_STRING_LIMIT).build());
+            return RedisStreamCodec.faultTolerant(
+                    new CompositeCodec(new LZ4CodecV2(), new JsonJacksonCodec(mapper)));
+        }
+
+        @BeforeEach
+        void setUp() {
+            smallLimitConfig = TestStreamConfiguration.create().toBuilder()
+                    .codec(smallLimitCodec())
+                    // Retirement is driven by the delivery count, which only rises when XAUTOCLAIM
+                    // redelivers the entry after pendingMessageDuration. The class default is 2 minutes
+                    // and maxRetries 3, so an undecodable entry would sit in the stream for ~6 minutes.
+                    // Compressed here to keep the test quick; see the class javadoc for what this means
+                    // at the shipped values.
+                    .pendingMessageDuration(io.dropwizard.util.Duration.milliseconds(500))
+                    .claimIntervalRatio(2)
+                    .maxRetries(2)
+                    .build();
+            smallLimitStream = redissonClient.getStream(
+                    smallLimitConfig.getStreamName(), smallLimitConfig.getCodec());
+            smallLimitStream.delete().block();
+        }
+
+        @Test
+        void shouldDrainAnEntryThatOnlyFailsOnRead() {
+            var subscriber = trackSubscriber(
+                    TestRedisSubscriber.createSubscriber(smallLimitConfig, redissonClient));
+            var oversized = "a".repeat(SMALL_STRING_LIMIT + 1);
+            var healthy = "well-within-limits";
+
+            // Start first: the consumer group is created at '$', so it only sees entries added after
+            // start(). Publishing beforehand leaves them permanently undelivered and proves nothing.
+            subscriber.start();
+
+            // Writes fine: Jackson constrains reads, not writes. This is the asymmetry behind OPIK-8164.
+            smallLimitStream.add(StreamAddArgs.entry(TestStreamConfiguration.PAYLOAD_FIELD, oversized)).block();
+            smallLimitStream.add(StreamAddArgs.entry(TestStreamConfiguration.PAYLOAD_FIELD, healthy)).block();
+            assertThat(smallLimitStream.size().block()).isEqualTo(2);
+
+            // The healthy entry behind it still gets through -- pre-PR it was stranded in the same batch.
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(subscriber.getSuccessMessageCount().get()).isEqualTo(1));
+
+            // The undecodable entry is NOT gone yet: it is retryable, so it stays pending for another
+            // consumer rather than being deleted on sight.
+            assertThat(smallLimitStream.size().block()).isEqualTo(1);
+
+            // And it does eventually leave, once the delivery count reaches maxRetries -- bounded, not
+            // permanent. This is the half that distinguishes this fix from the pre-PR wedge.
+            await().atMost(AWAIT_TIMEOUT_SECONDS * 10, TimeUnit.SECONDS)
+                    .untilAsserted(() -> assertThat(smallLimitStream.size().block()).isZero());
+
+            // It never reached processEvent: the sentinel is intercepted in processMessage.
+            assertThat(subscriber.getFailedMessageCount().get()).isZero();
+        }
+    }
+
     @Nested
     class SuccessTests {