[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,8 @@
 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.UndecodablePayloadException;
+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 +113,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 +164,14 @@ 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 whose payload could not be decoded, counted at detection. "
+                        + "Deliberately not a drop count: the failure is retryable, so one entry is counted once "
+                        + "per delivery until maxRetries retires it, and a pod that decodes it on a later claim "
+                        + "never contributes a drop at all. Alert on a sustained increase, which means entries are "
+                        + "cycling; the removal itself is logged by handleMaxRetriesReached.")
+                .build();
         this.backpressureDropCounter = meter
                 .counterBuilder("%s_%s_backpressure_drops_total".formatted(metricNamespace, metricsBaseName))
                 .setDescription("Total number of events dropped due to backpressure")
@@ -472,9 +483,74 @@ 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). Healthy entries claimed in the same
+        // batch are no longer stranded by it. Retire it through maxRetries rather than deleting it now:
+        // see UndecodablePayloadException for why first-delivery deletion would discard recoverable data.
+        if (message instanceof UndecodableStreamMessage undecodable) {
+            recordUndecodable(messageId, undecodable.cause(), null);
+            // Warn, not error: on its own this is one delivery of one entry, and the retryable path may
+            // still decode it on another pod. handleMaxRetriesReached logs at error when it is finally
+            // removed, which is the event worth waking someone for.
+            log.warn("Undecodable message: messageId '{}', stream '{}', payloadBytes '{}'",
+                    messageId, config.getStreamName(), undecodable.payloadBytes(), undecodable.cause());
+            return Mono.just(ProcessingResult.builder()
+                    .messageId(messageId)
+                    .status(MessageStatus.FAILURE)
+                    // Retryable on purpose: "this pod cannot decode it" is not "nobody can". maxRetries
+                    // bounds the cycling, so it still terminates. See UndecodablePayloadException.
+                    .error(new UndecodablePayloadException(
+                            "Undecodable stream payload of %d bytes".formatted(undecodable.payloadBytes()),
+                            undecodable.cause()))
+                    .context(MessageContext.UNKNOWN)
+                    .build());
+        }
+        // Nothing under the expected field. Two different causes hide behind the same symptom, and they
+        // deserve opposite treatment, so tell them apart by looking for a sentinel among the KEYS:
+        //   - the field name itself failed to decode (LZ4/Kryo on this codec), so the lookup missed. That
+        //     can be version skew between pods, so it is retryable -- and the key sentinel carries the
+        //     cause, which would otherwise be lost entirely since nothing else inspects key objects.
+        //   - the entry genuinely carries no such field. No pod can invent it, so retrying only delays
+        //     the inevitable; keep the pre-existing non-retryable, remove-on-first-delivery behaviour
+        //     (it used to arrive as a NullPointerException out of processEvent).
+        if (message == null) {
+            var keyFailure = undecodableKeyCause(entry.getValue());
+            if (keyFailure != null) {
+                recordUndecodable(messageId, keyFailure, "undecodable_field_name");
+                log.warn("Message field name could not be decoded, payload unreachable: messageId '{}', "
+                        + "stream '{}'", messageId, config.getStreamName(), keyFailure);
+                return Mono.just(ProcessingResult.builder()
+                        .messageId(messageId)
+                        .status(MessageStatus.FAILURE)
+                        .error(new UndecodablePayloadException(
+                                "Field name could not be decoded, no payload under '%s'"
+                                        .formatted(payloadField),
+                                keyFailure))
+                        .context(MessageContext.UNKNOWN)
+                        .build());
+            }
+            // Deliberately NOT on *_undecodable_messages_total: nothing failed to decode here, and
+            // mixing a deterministic malformed entry into that counter would make a decoder alert fire
+            // on it. messageProcessingErrors already counts it via postProcessFailureMessages.
+            recordQueueDelay(messageId);
+            log.warn("Message has no payload under field '{}': messageId '{}', stream '{}'",
+                    payloadField, messageId, config.getStreamName());
+            return Mono.just(ProcessingResult.builder()
+                    .messageId(messageId)
+                    .status(MessageStatus.FAILURE)
+                    // IllegalStateException is non-retryable: removed on first delivery, as before.
+                    .error(new IllegalStateException(
+                            "No payload under field '%s'".formatted(payloadField)))
+                    .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.
+        // message that reaches processEvent (success or failure), and the failure path reuses it for
+        // error attribution. The undecodable and no-payload branches above return before this -- they
+        // have no decoded message to attribute -- and record queue delay themselves.
         var context = messageContext(message);
         var workspaceAttributes = Attributes.of(
                 ErrorMetricsResolver.WORKSPACE_ID_KEY, context.workspaceId(),
@@ -495,12 +571,57 @@ private Mono<ProcessingResult> processMessage(Map.Entry<StreamMessageId, Map<Str
                         .build()))
                 .doFinally(signalType -> {
                     messageProcessingTime.record(System.currentTimeMillis() - startMillis, workspaceAttributes);
-                    extractTimeFromMessageId(messageId)
-                            .ifPresent(messageMillis -> messageQueueDelay
-                                    .record(System.currentTimeMillis() - messageMillis));
+                    recordQueueDelay(messageId);
                 });
     }
 
+    /**
+     * Counts an entry that could not be decoded and keeps it visible in the queue-delay histogram.
+     * <p>
+     * The histogram matters more here than on the success path: a retryable undecodable entry is
+     * delivered up to {@code maxRetries} times, and its growing age is the signal that shows an entry
+     * cycling in a growing PEL. Returning early without recording it would hide exactly the entries
+     * worth noticing.
+     *
+     * @param cause     the decode failure, or {@code null} when there was simply no payload field
+     * @param errorType overrides the {@code error_type} label; {@code null} derives it from {@code cause}
+     */
+    private void recordUndecodable(StreamMessageId messageId, Throwable cause, String errorType) {
+        undecodableMessages.add(1, Attributes.of(
+                ErrorMetricsResolver.ERROR_TYPE_KEY,
+                errorType != null ? errorType : ErrorMetricsResolver.errorType(cause),
+                ErrorMetricsResolver.STREAM_KEY, config.getStreamName()));
+        recordQueueDelay(messageId);
+    }
+
+    /**
+     * Keeps an entry that returns early visible in the queue-delay histogram. Its growing age is the
+     * signal that shows an entry cycling in a growing PEL, so the paths most worth noticing must not be
+     * the ones that skip it.
+     */
+    private void recordQueueDelay(StreamMessageId messageId) {
+        extractTimeFromMessageId(messageId)
+                .ifPresent(messageMillis -> messageQueueDelay.record(System.currentTimeMillis() - messageMillis));
+    }
+
+    /**
+     * The decode failure behind an entry's field name, if the map-key decoder produced a sentinel.
+     * <p>
+     * Reached by inspecting the key objects, which nothing else does: a sentinel key cannot match
+     * {@code payloadField}, so without this the cause would be discarded and a field-name failure would
+     * be indistinguishable from an entry that never carried the field.
+     */
+    private static Throwable undecodableKeyCause(Map<?, ?> valueMap) {
+        if (valueMap == null) {
+            return null;
+        }
+        return valueMap.keySet().stream()
+                .filter(UndecodableStreamMessage.class::isInstance)
+                .map(key -> ((UndecodableStreamMessage) key).cause())
+                .findFirst()
+                .orElse(null);
+    }
+
     private Mono<List<ProcessingResult>> postProcessSuccessMessages(List<ProcessingResult> processingResults) {
         var successIds = processingResults.stream()
                 .filter(processingResult -> processingResult.status() == MessageStatus.SUCCESS)

apps/opik-backend/src/main/java/com/comet/opik/infrastructure/metrics/ErrorMetricsResolver.java

@@ -31,6 +31,8 @@ public class ErrorMetricsResolver {
     public static final AttributeKey<String> WORKSPACE_ID_KEY = AttributeKey.stringKey("workspace_id");
     public static final AttributeKey<String> WORKSPACE_NAME_KEY = AttributeKey.stringKey("workspace_name");
     public static final AttributeKey<String> USER_NAME_KEY = AttributeKey.stringKey("user_name");
+    /** Redis stream name. Shared so subscriber and reaper metrics label the dimension identically. */
+    public static final AttributeKey<String> STREAM_KEY = AttributeKey.stringKey("stream");
     public static final String UNKNOWN = "unknown";
 
     public static String errorType(Throwable throwable) {

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;
@@ -22,8 +25,8 @@
 @AllArgsConstructor
 @Getter
 public enum RedisStreamCodec {
-    JAVA(Constants.JAVA, Suppliers.memoize(() -> new CompositeCodec(new LZ4CodecV2(),
-            new JsonJacksonCodec(buildStreamMapper())))),
+    JAVA(Constants.JAVA, Suppliers.memoize(() -> faultTolerant(new CompositeCodec(new LZ4CodecV2(),
+            new JsonJacksonCodec(buildStreamMapper()))))),
     JSON(Constants.JSON, () -> StringCodec.INSTANCE);
 
     /**
@@ -47,6 +50,108 @@ 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 encoding untouched; every decoder is wrapped so no decode path can throw.
+     * <p>
+     * All three are wrapped deliberately, because each is a distinct wire path and a throw on any of
+     * them lands in {@code CommandDecoder} with no {@code StreamMessageId}:
+     * <ul>
+     * <li>{@code getMapValueDecoder} — the payload. This is the OPIK-8164 path.</li>
+     * <li>{@code getMapKeyDecoder} — the entry's field name. On {@link #JAVA} this resolves to
+     * {@code LZ4CodecV2} over Kryo5, and both LZ4 frame validation and Kryo deserialization throw on
+     * corruption or a format skew across a version bump. This is why the wrapper goes <em>around</em>
+     * the {@link CompositeCodec} rather than around its value codec: the composite routes the map-key
+     * path to its {@code mapKeyCodec}, so wrapping the inner JSON codec would never have covered it.
+     * Field names are a fixed constant written only by Opik publishers, so a failure here is remote —
+     * but its blast radius is identical to the incident. A sentinel key misses the payload-field
+     * lookup, which the null-payload guard in {@code BaseRedisSubscriber.processMessage} then retires
+     * like any other undecodable entry.</li>
+     * <li>{@code getValueDecoder} — plain-value reads. Not reachable through {@link #JAVA}: the
+     * two-arg {@link CompositeCodec} constructor leaves {@code valueCodec} null, so
+     * {@code CompositeCodec.getValueDecoder()} throws {@link NullPointerException} while <em>obtaining</em>
+     * the decoder, which is before any wrapper can intervene. Wrapped anyway because this class is a
+     * general wrapper, not a JAVA-specific one, and a three-arg rewiring would otherwise silently
+     * reopen the hole.</li>
+     * </ul>
+     * The catch is {@link Exception}, not {@link Throwable}: an {@link Error} — an
+     * {@link OutOfMemoryError} while Jackson materializes a multi-megabyte String is the plausible one
+     * here — still propagates, because a JVM in that state should not have its failure recorded as a
+     * routine per-message drop.
+     */
+    @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 tolerant(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;