[E5] MEV-Share private flow — Rust MempoolSource into engine trigger path

[0xfandom]

Follow-up to #123. The Phase 0 Go-side MEV-Share SSE monitor consumer was removed in fa0df83 after the public-mempool path was proven on feat/mempool-tracking-scaffold (PR #118). This issue tracks the proper reintroduction: a Rust-side MempoolSource impl that feeds the same engine trigger path public mempool already uses, not a parallel monitor goroutine that only bumps metrics.

Why monitor-only was wrong

The original cmd/monitor/mev_share.go ran in a separate Go binary, parsed hints, and bumped aether_mev_share_* counters. The engine never saw the events. A real searcher edge requires the engine to act on private flow within the same hot-path detection pipeline as public mempool — counters in a sidecar are observability without action.

To do

Source impl

• new crates/ingestion/src/mev_share.rs defining MevShareSource that implements the existing MempoolSource trait (parallel to AlchemyMempool)
• subscribe to Flashbots MEV-Share SSE (https://mev-share.flashbots.net, override via MEV_SHARE_URL)
• decode hint events using the public schema (https://docs.flashbots.net/flashbots-mev-share/searchers/event-stream)
• map decoded hint → PendingTxEvent and dispatch through EventChannels.pending_tx_tx
• reconnect with exponential backoff on disconnect
• force reconnect after 60 s of read silence (Flashbots hangs sometimes; pattern proved useful in the removed Go monitor)

Source label

• add source: PendingSource enum field on PendingTxEvent with variants Public (default) and MevShare
• AlchemyMempool sets source = Public, MevShareSource sets source = MevShare
• propagate label through pipeline → detection → arb publish → DB row

DB column

• migration 0003_add_source_to_arbs.sql adding source TEXT NOT NULL DEFAULT 'public'
• NewArb.source field in crates/common/src/db.rs and internal/db/ledger.go
• write the source on every insert

Filter for minimal hints

• hints arrive with optional to, callData, functionSelector — the spec lets senders share only the hash. Such hints carry no signal we can decode.
• drop them with aether_mempool_filtered_total{reason="mev_share_minimal_hint"} so dashboards show how often Flashbots gives us useless hints (tier signal — if the rate is high, consider paying for fuller subscription)

Metrics

• aether_mempool_candidates_total{source} counter — split incoming candidates by source
• reuse aether_mempool_filtered_total for drops
• aether_mev_share_reconnects_total / aether_mev_share_errors_total{reason} for stream health
• expose source label on aether_pending_dex_tx_total{source, router, protocol, decoded} (extend existing counter with a new label — coordinate with dashboards)

Wiring

• spawn MevShareSource from crates/grpc-server/src/main.rs alongside AlchemyMempool, gated on MEV_SHARE_ENABLED=1 (or another truthy flag) so default behaviour stays unchanged when the env var is absent
• reuse the existing pipeline shutdown path

Tests

• fixture SSE event → parses into PendingTxEvent with source=MevShare
• minimal hint (no to, no callData) → bumps mempool_filtered_total{mev_share_minimal_hint} and is dropped
• malformed JSON line → bumps mev_share_errors_total{reason="parse"} and stream continues
• disconnect mid-stream → exponential backoff path engages

Proof

• 30-min live shadow run with both sources active
• DB rows present in arbs with both source=public and source=mev_share
• Grafana panel: candidates rate split by source

Out of scope

• Flashbots Protect direct ingest (different endpoint, different auth model)
• MEV-Share publishing (we are a consumer, not a publisher, in this work)
• Multi-provider hint aggregation (BlockNative, etc.)
• Higher-tier paid Flashbots Protect subscription

Branch

New branch off develop once #123 / PR #118 lands. Suggested: feat/mev-share-source.

Reference

Removed Go consumer (for context on schema fields + reconnect pattern): see commit fa0df83 in the diff of PR #118. The Go schema MevShareHint struct is a 1:1 match with the public Flashbots event stream and can be ported as-is to a Rust serde::Deserialize shape.