fix(relay): Limit maximum number of otel logs deserialized from protos

[klochek]

(This points at a personal branch with which it shares a little boilerplate; will point at main when that one is merged.)

[linear-code]

INGEST-1100

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit da82bf9. Configure here.}

[cursor]

Limit error remapped incorrectly

Medium Severity

otel_proto_deserializer::deserialize already returns logs::Error, including TooManyExpandedLogs, but parse_logs_data maps every failure to Invalid(InvalidProtobuf). Oversized protobuf payloads are still rejected, yet they surface as corrupt protobuf instead of the dedicated limit outcome.

 

^{Reviewed by Cursor Bugbot for commit da82bf9. Configure here.}

[sentry]

Bug: A protobuf parsing error can be misclassified as TooManyExpandedLogs if it occurs after exactly the maximum number of logs have been processed, because the error check uses meter.is_empty().
_{Severity: LOW}

Suggested Fix

Modify the error handling to pattern-match on the specific error type returned from merge_logs_data, such as a potential Error::MeterExhausted, instead of relying on the state of the meter with meter.is_empty(). This will correctly distinguish between exceeding the log limit and a general protobuf decoding failure.

Prompt for AI Agent

Review the code at the location below. A potential bug has been identified by an AI
agent. Verify if this is a real issue. If it is, propose a fix; if not, explain why it's
not valid.

Location: relay-server/src/processing/logs/integrations/otel_proto_deserializer.rs#L179

Potential issue: When parsing a protobuf payload, if the payload contains exactly the
maximum number of allowed log elements followed by corrupt or truncated data, a parsing
error will be incorrectly classified. The code successfully processes all valid
elements, exhausting the `meter`. When it then fails to parse the trailing corrupt data,
the error handling logic checks `meter.is_empty()`, which is true, and incorrectly
returns `Err(logs::Error::TooManyExpandedLogs)`. The correct error should be
`Invalid(DiscardReason::InvalidProtobuf)`. This misclassification leads to incorrect
outcome tracking and metrics for invalid payloads.

_{Did we get this right? 👍 / 👎 to inform future reviews.}