feat(eap-items): Route partition-boundary stragglers to a late-arrivals topic

[onewland]

Splits late-arriving eap-items messages off the regular DLQ into a dedicated snuba-eap-items-late-arrivals topic, so the DLQ can go back to meaning "genuinely invalid" and the late-arrivals stream becomes safe for automated replay.

Today the partition-boundary killswitch in the EAP items processor anyhow::bail!s on prior-week messages, and the same DLQ topic also collects protobuf-decode failures and other true-invalid cases. Any drain consumer pointed at the DLQ will faithfully re-attempt those, looping forever, so we can never automate replay of the partition-boundary stragglers without first manually scrubbing the DLQ.

The mechanism mirrors the existing ProduceReplacements pattern in errors. The processor now returns a typed InsertOrLateArrival<T> enum; a new ProduceLateArrivals strategy in the pipeline fans out the variants — Insert(b) to ClickHouse, LateArrival(payload) to the side topic carrying the original Kafka payload. A new late_arrivals_topic field on the storage stream-loader plumbs the topic name through Python ConsumerConfig to Rust.

Considered (and rejected) two other shapes:

• A pre-processor router doing a partial protobuf decode (mirrors BLQRouter). Would duplicate the protobuf decode work the processor already does.
• An error-routing wrapper that intercepts a sentinel anyhow::Error. Relies on fragile string/type matching on errors.

The typed-enum shape is the cleanest fit and lines up with what's already in the codebase for replacements.

Notes for reviewers

• Runtime config key renamed: eap_items_dlq_grace_period_min:<storage> → eap_items_late_arrival_grace_period_min:<storage>. The metric is also renamed: eap_items.messages.dlqed_prior_partition → eap_items.messages.routed_late_arrival. Dashboards and alerts need to be updated in lockstep with the deploy. The killswitch code path no longer mentions "DLQ" anywhere — it never routes there.

• eap_items_dlq_replay.yaml is intentionally NOT changed. The existing DLQ topic still carries a backlog of partition-boundary stragglers from the previous regime; the replay consumer can still drain that backlog. After the DLQ is drained, repointing/renaming the replay storage to consume the new topic is a follow-up PR.

• New Topic enum entry EAP_ITEMS_LATE_ARRIVALS (snuba-eap-items-late-arrivals) requires the Kafka topic to be provisioned in ops before this rolls out. If the topic is missing in any environment, the consumer will fail to start.

• Storages without a late_arrivals_topic configured: the factory uses ProduceLateArrivals::disabled, which silently drops the LateArrival branch. Matches the existing "killswitch disabled when runtime config is unset" semantic. Today only eap_items configures the topic; eap_items_dlq_replay (and any future EAP storage) opt out by leaving it unset.

• Order-safety for the rollout: the Rust ConsumerConfig.late_arrivals_topic has #[serde(default)], so the Rust side deserializes whether or not the Python side emits the field. The two halves can land independently.

• Drive-by: added # type: ignore[attr-defined] to the pre-existing confluent_kafka.admin import in table_storage.py. The symbols resolve at runtime but the stubs don't explicitly re-export them; pre-commit's mypy started flagging the file once I modified it.

Tests

• New processors::eap_items::tests::test_process_message_routes_late_arrival_when_armed and the row-binary variant cover the integration via patch_str_config_for_test.
• New strategies::late_arrivals::tests::forwards_insert_diverts_late_arrival covers the strategy fan-out directly.
• Existing partition-boundary unit tests were renamed from *_dlq_* to *_route_* and now assert should_route_late_arrival semantics.

All 146 Rust tests pass; relevant Python storage/consumer tests pass.