feat(replacer): optional concurrent replacement processing

[sentry-junior]

Spike / discussion PR: show what multi-message replacement concurrency could look like on one errors-replacer consumer.

Why

Today the errors replacer processes one kafka message at a time (RunTask → flush_batch([one])). Intra-message work already fans out across clickhouse shards, but independent replacements (especially different projects) cannot overlap.

During the US replacements backlog discussion we asked whether we could process more than one replacement at once without adding replicas. This PR sketches that path.

What changed

• New CLI flag: --concurrency (default 1, current behavior).
• concurrency > 1 switches the strategy to arroyo RunTaskInThreads.
• Same-project_id replacements stay serial via per-project locks.
• Shared worker state (offset map, project time counter, connection refresh, schema init, optimize) is locked for thread safety.
• Factory/CLI unit coverage for the strategy choice.

What this does not do

• Does not help a single hot project/group on one partition much — that work is still one-at-a-time by design.
• Cross-project messages on the same partition may complete out of kafka order when concurrency > 1. That is an intentional spike tradeoff and needs a real correctness review before any prod enablement.
• No deploy/ops wiring; default stays 1.

How to try

snuba replacer --storage errors --concurrency 4 ...

Test plan

• tests/cli/test_replacer.py expects concurrency passthrough
• factory tests for RunTask vs RunTaskInThreads and invalid concurrency
• full replacer integration suite in CI
• if ever considered for prod: load-test mixed-project backlog and same-project ordering guarantees

Requested by PDPM.

--

View Junior Session in Sentry