[OPIK-8192] [BE] fix: drop undecodable scoring stream messages instead of wedging the stream

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

@@ -4,6 +4,7 @@
 import com.comet.opik.infrastructure.StreamConfiguration;
 import com.comet.opik.infrastructure.auth.RequestContext;
 import com.comet.opik.infrastructure.metrics.ErrorMetricsResolver;
+import com.comet.opik.infrastructure.redis.UndecodableStreamMessage;
 import io.dropwizard.lifecycle.Managed;
 import io.opentelemetry.api.GlobalOpenTelemetry;
 import io.opentelemetry.api.common.Attributes;
@@ -111,6 +112,7 @@ public abstract class BaseRedisSubscriber<M> implements Managed {
     private final LongHistogram messageProcessingTime;
     private final LongHistogram messageQueueDelay;
     private final LongCounter messageProcessingErrors;
+    private final LongCounter undecodableMessages;
     private final LongCounter backpressureDropCounter;
     private final LongCounter claimErrors;
     private final LongHistogram claimTime;
@@ -161,6 +163,12 @@ protected BaseRedisSubscriber(
                 .counterBuilder("%s_%s_processing_errors".formatted(metricNamespace, metricsBaseName))
                 .setDescription("Errors when processing messages")
                 .build();
+        this.undecodableMessages = meter
+                .counterBuilder("%s_%s_undecodable_messages_total".formatted(metricNamespace, metricsBaseName))
+                .setDescription("Stream entries dropped because their payload could not be decoded. Non-zero means "
+                        + "data was discarded: a payload the codec cannot read is unrecoverable, and keeping it "
+                        + "would wedge the stream for every consumer. Alert on any increase.")
+                .build();
         this.backpressureDropCounter = meter
                 .counterBuilder("%s_%s_backpressure_drops_total".formatted(metricNamespace, metricsBaseName))
                 .setDescription("Total number of events dropped due to backpressure")
@@ -472,6 +480,26 @@ private Mono<ProcessingResult> processMessage(Map.Entry<StreamMessageId, Map<Str
                     .context(MessageContext.UNKNOWN)
                     .build());
         }
+        // A payload the codec could not decode. It arrives as a sentinel rather than an exception
+        // precisely so this point is reachable at all -- the throw used to happen inside Redisson, below
+        // here and before any messageId existed, which is why such an entry could never be acked or
+        // removed and wedged the stream permanently (OPIK-8164). Drop it: it will never become decodable,
+        // and every redelivery strands the healthy entries claimed alongside it.
+        if (message instanceof UndecodableStreamMessage undecodable) {
+            undecodableMessages.add(1);
+            log.error("Dropping undecodable message: messageId '{}', stream '{}', payloadBytes '{}'",
+                    messageId, config.getStreamName(), undecodable.payloadBytes(), undecodable.cause());
+            return Mono.just(ProcessingResult.builder()
+                    .messageId(messageId)
+                    .status(MessageStatus.FAILURE)
+                    // IllegalArgumentException is in NON_RETRYABLE_EXCEPTIONS, so postProcessFailureMessages
+                    // acks and removes without consulting the delivery count.
+                    .error(new IllegalArgumentException(
+                            "Undecodable stream payload of %d bytes".formatted(undecodable.payloadBytes()),
+                            undecodable.cause()))
+                    .context(MessageContext.UNKNOWN)
+                    .build());
+        }
         var startMillis = System.currentTimeMillis();
         // Resolve the workspace/user once: the processing-time histogram is tagged with it for every
         // message (success or failure), and the failure path reuses it for error attribution.

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

@@ -8,9 +8,12 @@
 import com.google.common.base.Suppliers;
 import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.RequiredArgsConstructor;
 import lombok.experimental.UtilityClass;
 import org.redisson.client.codec.Codec;
 import org.redisson.client.codec.StringCodec;
+import org.redisson.client.protocol.Decoder;
+import org.redisson.client.protocol.Encoder;
 import org.redisson.codec.CompositeCodec;
 import org.redisson.codec.JsonJacksonCodec;
 import org.redisson.codec.LZ4CodecV2;
@@ -23,7 +26,7 @@
 @Getter
 public enum RedisStreamCodec {
     JAVA(Constants.JAVA, Suppliers.memoize(() -> new CompositeCodec(new LZ4CodecV2(),
-            new JsonJacksonCodec(buildStreamMapper())))),
+            faultTolerant(new JsonJacksonCodec(buildStreamMapper()))))),
     JSON(Constants.JSON, () -> StringCodec.INSTANCE);
 
     /**
@@ -47,6 +50,87 @@ static ObjectMapper buildStreamMapper() {
         return mapper;
     }
 
+    /**
+     * Wraps a codec so a payload it cannot decode yields an {@link UndecodableStreamMessage} instead of
+     * throwing.
+     * <p>
+     * A throw here happens inside Redisson's {@code CommandDecoder}, below {@code BaseRedisSubscriber} and
+     * before any {@code StreamMessageId} exists, so the entry can never be acked or removed and the stream
+     * wedges permanently (OPIK-8164). Returning a value keeps the failure in the normal message flow, where
+     * the id is known and the entry can be dropped, counted and logged.
+     * <p>
+     * Encoders are untouched, so the wire format is byte-identical and a rolling upgrade is safe in both
+     * directions: a pod on either build writes what the other can read.
+     */
+    @VisibleForTesting
+    static Codec faultTolerant(Codec delegate) {
+        return new FaultTolerantCodec(delegate);
+    }
+
+    /**
+     * Delegates everything except decoding, which cannot throw.
+     * <p>
+     * Both the map-value and plain-value decoders are wrapped: streams decode the payload through
+     * {@code getMapValueDecoder}, but {@link CompositeCodec} may reach for either depending on the
+     * operation, and a decoder that throws on one path defeats the purpose.
+     */
+    @RequiredArgsConstructor
+    private static final class FaultTolerantCodec implements Codec {
+
+        private final Codec delegate;
+
+        private static Decoder<Object> tolerant(Decoder<Object> decoder) {
+            return (buf, state) -> {
+                int payloadBytes = buf.readableBytes();
+                try {
+                    return decoder.decode(buf, state);
+                } catch (Exception decodeFailure) {
+                    // Consume whatever the failed decode left behind, so the buffer looks the same to
+                    // Redisson as it would after a successful decode.
+                    if (buf.isReadable()) {
+                        buf.skipBytes(buf.readableBytes());
+                    }
+                    return new UndecodableStreamMessage(payloadBytes, decodeFailure);
+                }
+            };
+        }
+
+        @Override
+        public Decoder<Object> getMapValueDecoder() {
+            return tolerant(delegate.getMapValueDecoder());
+        }
+
+        @Override
+        public Encoder getMapValueEncoder() {
+            return delegate.getMapValueEncoder();
+        }
+
+        @Override
+        public Decoder<Object> getMapKeyDecoder() {
+            return delegate.getMapKeyDecoder();
+        }
+
+        @Override
+        public Encoder getMapKeyEncoder() {
+            return delegate.getMapKeyEncoder();
+        }
+
+        @Override
+        public Decoder<Object> getValueDecoder() {
+            return tolerant(delegate.getValueDecoder());
+        }
+
+        @Override
+        public Encoder getValueEncoder() {
+            return delegate.getValueEncoder();
+        }
+
+        @Override
+        public ClassLoader getClassLoader() {
+            return delegate.getClassLoader();
+        }
+    }
+
     private final String name;
     private final Supplier<Codec> codecSupplier;
 

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

@@ -0,0 +1,21 @@
+package com.comet.opik.infrastructure.redis;
+
+/**
+ * Stand-in for a stream entry whose payload the codec could not decode.
+ * <p>
+ * Without this, a decode failure is thrown inside Redisson's {@code CommandDecoder} — below
+ * {@code BaseRedisSubscriber}, before any {@code StreamMessageId} is in hand. With no id there is
+ * nothing to ack, retry or remove, so the entry is redelivered forever and, at
+ * {@code consumerBatchSize > 1}, strands every healthy entry claimed alongside it. That is the
+ * permanent wedge behind OPIK-8164: one oversized trace took two production scoring streams to
+ * {@code pending == XLEN} and 19.66 GiB.
+ * <p>
+ * Returning this instead of throwing keeps the failure inside the normal message flow, where the id
+ * is known, so {@code BaseRedisSubscriber} can drop the entry, count it and log it. A payload we
+ * cannot decode will never become decodable, so dropping is the only outcome that terminates.
+ *
+ * @param payloadBytes readable bytes the decoder was handed, for sizing the offending entry
+ * @param cause        the decode failure, kept for the log rather than rethrown
+ */
+public record UndecodableStreamMessage(int payloadBytes, Throwable cause) {
+}

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

@@ -1,5 +1,6 @@
 package com.comet.opik.api.resources.v1.events;
 
+import com.comet.opik.infrastructure.redis.UndecodableStreamMessage;
 import com.comet.opik.podam.PodamFactoryUtils;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeAll;
@@ -269,6 +270,58 @@ void shouldHandleEmptyListEntryValue() {
                     });
         }
 
+        /**
+         * OPIK-8192. Once an undecodable payload arrives as a <em>value</em> rather than as a throw from
+         * inside Redisson, it carries a messageId and is dropped. Before that it never reached here at all,
+         * was redelivered forever, and at {@code consumerBatchSize > 1} stranded every healthy entry
+         * claimed with it -- the permanent wedge in OPIK-8164.
+         * <p>
+         * Scope: this pins the drop and the consumer surviving it. It does <em>not</em> pin the explicit
+         * {@link UndecodableStreamMessage} branch in {@code processMessage} -- with that branch removed the
+         * sentinel still reaches {@code processEvent}, fails the generic cast, and is dropped as a
+         * non-retryable {@code ClassCastException}, so this test passes either way. What the branch adds is
+         * the dedicated counter and a log naming the stream and payload size, which is the difference
+         * between "we discarded a 20 MB trace" and an opaque cast error. Asserting the counter would need
+         * OTel metric test infrastructure this module does not have. The load-bearing guard against the
+         * wedge returning is {@code FaultTolerantStreamCodecTest}.
+         */
+        @Test
+        void shouldDropUndecodableMessageAndKeepConsuming() {
+            var readCount = new AtomicInteger();
+            var undecodableId = new StreamMessageId(System.currentTimeMillis(), 0);
+            var subscriber = trackSubscriber(TestRedisSubscriber.createSubscriber(CONFIG, redissonClient));
+            whenAutoClaimReturnEmpty(subscriber.getConsumerId());
+            when(stream.readGroup(eq(CONFIG.getConsumerGroupName()), anyString(), any(StreamReadGroupArgs.class)))
+                    .thenAnswer(invocation -> {
+                        int count = readCount.incrementAndGet();
+                        if (count == 1) {
+                            return Mono.just(Map.of(undecodableId,
+                                    Map.of(TestStreamConfiguration.PAYLOAD_FIELD,
+                                            new UndecodableStreamMessage(20_054_016,
+                                                    new IllegalStateException(
+                                                            "String value length exceeds maximum")))));
+                        }
+                        return Mono.just(Map.of(new StreamMessageId(System.currentTimeMillis(), count),
+                                Map.of(TestStreamConfiguration.PAYLOAD_FIELD,
+                                        podamFactory.manufacturePojo(String.class))));
+                    });
+            whenAckReturn();
+            whenRemoveReturn();
+
+            subscriber.start();
+
+            await().atMost(AWAIT_TIMEOUT_SECONDS, TimeUnit.SECONDS)
+                    .untilAsserted(() -> {
+                        // The undecodable entry is acked and removed -- dropped, not redelivered.
+                        verify(stream).ack(eq(CONFIG.getConsumerGroupName()),
+                                eq(new StreamMessageId[]{undecodableId}));
+                        verify(stream).remove(eq(new StreamMessageId[]{undecodableId}));
+                        // It never reaches processEvent, and healthy messages behind it still process.
+                        assertThat(subscriber.getFailedMessageCount().get()).isEqualTo(0);
+                        assertThat(subscriber.getSuccessMessageCount().get()).isGreaterThan(1);
+                    });
+        }
+
         @Test
         void shouldNotDieOnProcessingError() {
             // Subscriber that throws on first message, succeeds on subsequent

apps/opik-backend/src/test/java/com/comet/opik/infrastructure/redis/FaultTolerantStreamCodecTest.java

@@ -0,0 +1,100 @@
+package com.comet.opik.infrastructure.redis;
+
+import com.fasterxml.jackson.core.StreamReadConstraints;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+import org.redisson.client.codec.Codec;
+import org.redisson.codec.JsonJacksonCodec;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The decode side of {@link RedisStreamCodec#JAVA} must never throw: a throw lands inside Redisson's
+ * {@code CommandDecoder}, below {@code BaseRedisSubscriber} and before any {@code StreamMessageId} exists,
+ * so the entry can never be acked or removed and the stream wedges permanently (OPIK-8164).
+ * <p>
+ * Container-free on purpose -- this is decoder behaviour, and the wedge it prevents needs no Redis to
+ * demonstrate.
+ */
+@DisplayName("Fault-tolerant stream codec")
+class FaultTolerantStreamCodecTest {
+
+    private static final int SMALL_STRING_LIMIT = 64;
+
+    /** A mapper whose string limit is small enough to breach without allocating megabytes. */
+    private static Codec codecWithSmallStringLimit() {
+        var mapper = new ObjectMapper();
+        mapper.getFactory().setStreamReadConstraints(
+                StreamReadConstraints.builder().maxStringLength(SMALL_STRING_LIMIT).build());
+        return RedisStreamCodec.faultTolerant(new JsonJacksonCodec(mapper));
+    }
+
+    private static ByteBuf json(String payload) {
+        return Unpooled.wrappedBuffer(payload.getBytes(StandardCharsets.UTF_8));
+    }
+
+    @Test
+    @DisplayName("a payload over the string limit decodes to a sentinel instead of throwing")
+    void oversizedPayloadYieldsSentinel() throws IOException {
+        var oversized = "\"%s\"".formatted("a".repeat(SMALL_STRING_LIMIT + 1));
+        var buf = json(oversized);
+        var payloadBytes = buf.readableBytes();
+
+        var decoded = codecWithSmallStringLimit().getMapValueDecoder().decode(buf, null);
+
+        assertThat(decoded).isInstanceOf(UndecodableStreamMessage.class);
+        var undecodable = (UndecodableStreamMessage) decoded;
+        assertThat(undecodable.payloadBytes()).isEqualTo(payloadBytes);
+        assertThat(undecodable.cause()).isNotNull();
+        // The size is reported from the buffer, so it survives the failed decode consuming it.
+        assertThat(buf.isReadable()).isFalse();
+    }
+
+    @Test
+    @DisplayName("a well-formed payload still decodes normally")
+    void wellFormedPayloadStillDecodes() throws IOException {
+        var decoded = codecWithSmallStringLimit().getMapValueDecoder().decode(json("\"within limits\""), null);
+
+        assertThat(decoded).isEqualTo("within limits");
+    }
+
+    @Test
+    @DisplayName("malformed JSON decodes to a sentinel rather than throwing")
+    void malformedJsonYieldsSentinel() throws IOException {
+        var decoded = codecWithSmallStringLimit().getMapValueDecoder().decode(json("{not json"), null);
+
+        assertThat(decoded).isInstanceOf(UndecodableStreamMessage.class);
+    }
+
+    /**
+     * The wire format must not move. Encoders are handed through untouched, so a pod on either build
+     * writes bytes the other can read and a rolling upgrade is safe in both directions.
+     */
+    @Test
+    @DisplayName("encoders are the delegate's own, so the wire format is unchanged")
+    void encodersAreUntouched() {
+        var delegate = new JsonJacksonCodec(new ObjectMapper());
+        var tolerant = RedisStreamCodec.faultTolerant(delegate);
+
+        assertThat(tolerant.getMapValueEncoder()).isSameAs(delegate.getMapValueEncoder());
+        assertThat(tolerant.getMapKeyEncoder()).isSameAs(delegate.getMapKeyEncoder());
+        assertThat(tolerant.getValueEncoder()).isSameAs(delegate.getValueEncoder());
+        assertThat(tolerant.getMapKeyDecoder()).isSameAs(delegate.getMapKeyDecoder());
+        assertThat(tolerant.getClassLoader()).isSameAs(delegate.getClassLoader());
+    }
+
+    @Test
+    @DisplayName("the shipped JAVA codec is fault tolerant")
+    void shippedJavaCodecIsFaultTolerant() throws IOException {
+        // Guards the wiring, not the wrapper: the enum must hand the stream a decoder that cannot throw.
+        var decoded = RedisStreamCodec.JAVA.getCodec().getMapValueDecoder().decode(json("{not json"), null);
+
+        assertThat(decoded).isInstanceOf(UndecodableStreamMessage.class);
+    }
+}