feat(consumer): add a DLQ-by-age load-shedding lever

[onewland]

Stacked on #8166 (BLQ removal). Base is ref/remove-blq; review that first. Retarget to master once #8166 merges.

Summary

Adds DlqByAge, a stateless arroyo strategy for the Rust consumer that sheds load during a backlog. When enabled, it routes a configurable proportion of messages older than a threshold (by Kafka broker timestamp) to the DLQ via the standard InvalidMessage mechanism. The rest are forwarded untouched — so a backlogged consumer catches up to fresh data while the parked backlog is backfilled from the DLQ later.

This replaces the removed BLQ mechanism with something far simpler: no state machine, no consumer panic, no separate producer. It leans entirely on arroyo's existing DLQ buffer + policy.

How it works

• Sits at the front of the pipeline (where BLQRouter was), before decode/process, so dropped messages cost almost nothing.
• For each message: if the per-storage sample rate is 0 → forward (fast path). Else if now - broker_ts <= threshold → forward. Else roll rand < rate: hit → Err(SubmitError::InvalidMessage) (→ DLQ); miss → forward.
• Only inserted when a DLQ topic is configured (dlq_configured flag from consumer.rs). Without a DLQ policy, arroyo logs an InvalidMessage and silently drops it — which would lose data — so we never wire the strategy in that case.

Kept out of Sentry

Age-based routing is an expected load-shedding outcome, not an error. Dead-lettered messages use InvalidMessageReason::Ignored, so arroyo logs them at DEBUG (not ERROR) and the tracing→Sentry layer (logging.rs) does not turn them into Sentry issues. Both reasons produce to the DLQ identically — only the log level differs. This is the same intent as the SilencedDLQMessage path from #8012.

Configuration (sentry-options, read per-message, runtime-tunable)

Option	Type	Default	Meaning
consumer.dlq_by_age_sample_rate_by_storage	dict {storage: number}	{}	fraction (0–1) of too-old messages to dead-letter; no entry ⇒ 0 (disabled)
consumer.dlq_by_age_threshold_seconds	int	3600	age cutoff by broker timestamp

Ships disabled — no storage is enabled and the rate defaults to 0, so this is an inert passthrough until an operator opts a storage in.

Metrics: dlq_by_age.stale_seen and dlq_by_age.routed_to_dlq (storage is already a global metric tag).

Test plan

• cargo test --lib dlq_by_age — 6 unit tests pass: disabled-by-default, explicit rate 0, fresh-forwarded, rate 1 routes all stale, per-storage scoping, and a partial-rate split. Each DLQ'd message is asserted to carry Ignored.
• cargo clippy --all-targets clean; cargo fmt --check clean.

Follow-up (PR 3, ops)

Enable per storage via sentry-options and watch dlq_by_age.routed_to_dlq + DLQ topic lag.

🤖 Generated with Claude Code

[sentry]

Bug: The threshold() method can panic due to an unvalidated, large configuration value from consumer.dlq_by_age_threshold_seconds, causing a consumer crash loop.
_{Severity: MEDIUM}

Suggested Fix

Use the fallible TimeDelta::try_seconds() function instead of TimeDelta::seconds() to handle potentially large values without panicking. Additionally, consider adding a maximum constraint to the consumer.dlq_by_age_threshold_seconds option in the JSON schema to prevent invalid configurations.

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_snuba/src/strategies/dlq_by_age.rs#L91

Potential issue: The `threshold()` method reads the
`consumer.dlq_by_age_threshold_seconds` configuration value and passes it directly to
`TimeDelta::seconds()`. This function will panic if the input value is excessively large
(greater than `i64::MAX / 1_000`). The configuration schema does not enforce a maximum
value, and the code only validates that the number is positive. Because this method is
called for every message processed, a misconfiguration with a very large number would
cause the consumer to enter a persistent crash loop, preventing it from processing any
messages.

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

[onewland]

This doesn't concern me