feat: Add reason to InvalidMessage

[lcian]

As of getsentry/snuba#7945, in some cases Snuba can return RunTaskError::InvalidMessage to avoid processing some messages and intentionally have them end up in DLQ.
From the POV of arroyo these are proper failures, which causes a ton of unnecessary logs to be emitted.
This introduces an IgnoredMessage.reason field that can hold Invalid for legit errors and Ignored for intentional drops.

[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 1a5c5c9. Configure here.}

[untitaker]

I think we should add severity field to InvalidMessage instead of having another enum variant. This will make it easier to find all uses that need updating.

[sentry]

Bug: Messages with InvalidMessageReason::Ignored are counted towards the DLQ limit. A high volume of ignored messages can trigger a panic when the limit is reached.
_{Severity: HIGH}

Suggested Fix

In DlqPolicyWrapper::produce, check for InvalidMessageReason::Ignored and bypass the call to dlq_limit_state.record_invalid_message for such messages. This will prevent intentionally ignored messages from contributing to the invalid message limit that can cause a panic.

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: rust-arroyo/src/processing/dlq.rs#L643-L645

Potential issue: The new `InvalidMessageReason::Ignored` variant is intended for
messages that are intentionally skipped. However, in `DlqPolicyWrapper::produce`, these
"ignored" messages are still counted towards the DLQ limit by `record_invalid_message`.
If a high volume of ignored messages is received, this limit can be exceeded, causing
the consumer to `panic!`. This behavior is inconsistent with the semantics of "ignoring"
a message, as it can lead to a crash for a non-error condition.

[lcian]

How about this way @untitaker?
I can also rename it to like Severity::High/Low or Severity::Error/Info (but that would be a bit misleading because the logging is actually error+info when legit error, or debug+debug when ignored), just let me know what you prefer.

[untitaker]

that works for me. the sentry-bot issue seems real but since we didn't hit it with snuba yet it's not important

[untitaker]

feel free to merge and cut a new bugfix release

[lcian]

Thanks @untitaker.

I just realized that if someone is constructing InvalidMessage this will be a breaking change.
With a GH code search, I can only see Snuba and https://github.com/getsentry/streams doing that.
For Snuba it should be fine because it builds the container with --locked.
streams doesn't use --locked but it's a Python lib that ships the rust part as a binary as far as I understand, so that should also be fine.

So is this okay to release as a patch?

[untitaker]

@lcian i don't have a problem with breaking changes, but if you want to do them i think it would be best if you drive the entire migration:

1. merge pr
2. make arroyo release
3. bump in all repos and fix the compilation issues on the go

So is this okay to release as a patch?

ok let's do minor.

but the point is not about patch vs minor. the point is that, yes, breaking changes are fine, but you have to roll them out asap. pushing out a breaking change and then either:

1. letting these changes be unreleased within arroyo main
2. or even just letting applications be outdated

..puts us in a bad position when we want to make a new bugfix release in a hurry during inc response, because then we'll have to deal with the breaking change. and it almost doesn't matter if the breaking change was released as minor or patch release.

this could be solved by backporting bugfixes to old major/minor releases but i don't think we want to go through that effort.

[lcian]

Basically I just wanted to avoid a situation where some image is built on the fly without --locked, which would cause it to pick up the new version and fail to deploy or something like that. I guess that doesn't make any sense as images are always built from main etc.
But I figured best to double check, given that I didn't completely know where and how this library is used.
I'll release a minor and resolve the breakages in all users.

[untitaker]

where some image is built on the fly without --locke

we should not have this situation at all, and if it is, then we need to fix the app. streams is just a library, yeah.