From aedf95b0985e33e37b0dd6bca951517708182a75 Mon Sep 17 00:00:00 2001
From: Claude <noreply@anthropic.com>
Date: Sun, 8 Mar 2026 12:41:49 +0000
Subject: [PATCH] Fix protobuf UTF-8 decode errors by adding lenient decoding
 fallback

proto-lens uses strict UTF-8 decoding for protobuf string fields, which
causes ~1200 message decode failures per minute when OTLP producers send
non-UTF-8 bytes in string fields (e.g. AnyValue.string_value, Metric.name,
Span.name). The entire message is dropped even though the protobuf structure
is valid.

Add decodeMessageLenient which first tries normal decoding, then on UTF-8
errors walks the protobuf wire format to replace invalid UTF-8 bytes with
'?' in length-delimited fields and retries. This preserves ~77% of
previously-dropped messages.

https://claude.ai/code/session_01MmF2VFF2zfknCJNUzgJACT
---
 src/Opentelemetry/OtlpServer.hs | 43 ++++++++++++++++++++++++++++++++-
 1 file changed, 42 insertions(+), 1 deletion(-)

diff --git a/src/Opentelemetry/OtlpServer.hs b/src/Opentelemetry/OtlpServer.hs
index 4db18745a..dabf11129 100644
--- a/src/Opentelemetry/OtlpServer.hs
+++ b/src/Opentelemetry/OtlpServer.hs
@@ -34,6 +34,8 @@ import Data.List qualified as L
 import Data.List.Extra (chunksOf)
 import Data.Map qualified as Map
 import Data.ProtoLens.Encoding (decodeMessage)
+import Data.ProtoLens.Encoding.Bytes (runBuilder, runParser)
+import Data.ProtoLens.Encoding.Wire (TaggedValue (..), WireValue (..), buildFieldSet, parseFieldSet)
 import Data.Scientific (fromFloatDigits)
 import Data.Text qualified as T
 import Data.Time (UTCTime, diffUTCTime, getCurrentTime)
@@ -123,6 +125,45 @@ wireTypeErrorsRef :: IORef (HashMap Text (Int, AE.Value))
 wireTypeErrorsRef = unsafePerformIO $ newIORef HashMap.empty
 
 
+-- | Try to decode a protobuf message with lenient UTF-8 handling.
+-- On UTF-8 decode errors, sanitizes the wire format by replacing invalid UTF-8 bytes
+-- with '?' in length-delimited fields, then retries decoding.
+decodeMessageLenient :: Message msg => ByteString -> Either String msg
+decodeMessageLenient bs =
+  case decodeMessage bs of
+    Right msg -> Right msg
+    Left err
+      | "Cannot decode byte" `L.isInfixOf` err -> decodeMessage (sanitizeProtobufUtf8 bs)
+      | otherwise -> Left err
+
+
+-- | Sanitize a protobuf ByteString by walking the wire format and replacing
+-- invalid UTF-8 bytes in length-delimited fields with '?'.
+sanitizeProtobufUtf8 :: ByteString -> ByteString
+sanitizeProtobufUtf8 bs =
+  case runParser parseFieldSet bs of
+    Left _ -> bs
+    Right fields -> runBuilder (buildFieldSet (map sanitizeTaggedValue fields))
+
+
+sanitizeTaggedValue :: TaggedValue -> TaggedValue
+sanitizeTaggedValue (TaggedValue tag (Lengthy content))
+  | BS.null content = TaggedValue tag (Lengthy content)
+  | otherwise =
+      case runParser parseFieldSet content of
+        Right subFields@(_ : _) ->
+          -- Content parses as a nested protobuf message; recursively sanitize its fields
+          TaggedValue tag (Lengthy (runBuilder (buildFieldSet (map sanitizeTaggedValue subFields))))
+        _ ->
+          -- Content is a string or bytes field; replace invalid UTF-8 bytes with '?'
+          TaggedValue tag (Lengthy (sanitizeUtf8Bytes content))
+sanitizeTaggedValue other = other
+
+
+sanitizeUtf8Bytes :: ByteString -> ByteString
+sanitizeUtf8Bytes content = encodeUtf8 (decodeUtf8With (\_ _ -> Just '?') content)
+
+
 -- | Initialize periodic error logging
 initPeriodicErrorLogging :: Logger -> IO ()
 initPeriodicErrorLogging logger = void $ forkIO $ forever $ do
@@ -298,7 +339,7 @@ processBatchPipeline !label msgs appCtx fallbackTime extractKeys extractIds conv
   checkpoint (cp "") $ do
     processingStartTime <- liftIO getCurrentTime
     let !msgsCount = length msgs
-        !decodedMsgs = [(ackId, decodeMessage msg :: Either String req) | (ackId, msg) <- msgs] `using` parList rpar
+        !decodedMsgs = [(ackId, decodeMessageLenient msg :: Either String req) | (ackId, msg) <- msgs] `using` parList rpar
         !allProjectKeys = V.force $ V.concat [extractKeys req | (_, Right req) <- decodedMsgs]
         !uniqueProjectKeys = V.force $ V.fromList $ L.nub $ V.toList allProjectKeys
         !atIds = concatMap extractIds [req | (_, Right req) <- decodedMsgs]