feat(ingest): subscription-driven feed reconciler (#62)

Activate only the feeds this host is subscribed to, and add/remove them at
runtime as subscriptions change — all feed detection in one place.

- ingest::subscriptions: the single detector, reading host subscriptions from
  `doublezero status --json` (the `multicast_groups` S:<code> entries — the
  authoritative per-host view), with shred-group IPs resolved via `multicast
  group list`. feeds.rs gains a group `code` (tiredsolid/scottsdale).
- ingest::reconcile: a periodic reconciler (--subscription-refresh-secs, default
  30) diffs desired-vs-running and spawns/aborts market-data receivers, the WS
  sink, and the shred forwarder. The WS sink activates only when a market-data
  feed is subscribed (so a shreds-only host serves no WS and can't collide with
  an existing :8081 service); shred sources track the subscribed edge-solana-*
  groups. Teardown is JoinHandle::abort(); a died feed self-heals next tick.
- Default-on with fail-open: no `doublezero` CLI -> static always-on set; a
  transient CLI error -> keep current activations. --subscription-gating-disable
  forces the static model (the e2e harness uses it — it feeds synthetic multicast).
- Supersedes the shred forwarder's one-shot discovery: remove discover_groups /
  resolve_sources / decide_activation in favour of the reconciler.

Docs + CHANGELOG updated; cargo build/clippy/test all green.

Author: juan-malbeclabs

CHANGELOG.md

@@ -8,6 +8,24 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ## [Unreleased]
 
 ### Added
+- **Subscription-driven feed activation** — the bridge now activates only the feeds this host is
+  actually subscribed to, and adds/removes them at runtime as subscriptions change:
+  - A single detector (`src/ingest/subscriptions.rs`) reads the host's subscriptions from
+    `doublezero status --json` (`multicast_groups`, the `S:<code>` entries — the authoritative
+    per-host view, unlike the network-wide `multicast group list`), resolving shred-group IPs via
+    `multicast group list` (`src/shred/discovery.rs::parse_group_code_ips`). Each market-data feed
+    now carries its group `code` (`src/ingest/feeds.rs`: `tiredsolid` = Hyperliquid, `scottsdale` =
+    Phoenix).
+  - A periodic reconciler (`src/ingest/reconcile.rs`) polls every `--subscription-refresh-secs`
+    (`DZ_SUBSCRIPTION_REFRESH_SECS`, default 30) and diffs desired-vs-running, spawning/aborting
+    market-data receivers, the WebSocket sink, and the shred forwarder. The **WebSocket sink comes
+    up only when ≥1 market-data feed is subscribed** (so a shreds-only host serves no WS and can't
+    collide with an existing `:8081` service); shred sources come from the subscribed
+    `edge-solana-*` groups.
+  - **Default-on with fail-open**: with no `doublezero` CLI (running from source) gating falls open
+    to the static always-on set; a transient CLI failure keeps the current activations rather than
+    flapping. `--subscription-gating-disable` (`DZ_SUBSCRIPTION_GATING_DISABLE`) forces the static
+    model. A single feed dying no longer exits the process — the reconciler respawns it.
 - Cross-source de-duplication **win metrics**, surfacing how the edge feed beats the
   original/public sources in both quantity and latency at each de-dup contest:
   - Quotes/trades (`src/ingest/arbiter.rs`): the staleness floor and windowed dedup now report

CLAUDE.md

@@ -35,15 +35,26 @@ defaults to `warn,doublezero_edge_connect=info` (our crate at `info`, deps quiet
 
 ## Architecture
 
-One WS-server task plus **one receiver task per selected feed** share a single
+A WS-server task plus **one receiver task per active feed** share a single
 `tokio::sync::broadcast` channel of `FeedMessage` (the fan-out backbone) and a `Mutex<HashMap>`
-instrument snapshot. `main.rs` selects feeds (`--feed`, or all of `ingest::feeds::FEEDS` by
-default), builds the shared `Arbiter` around the broadcast `Sender`, spawns the receivers into a
-`JoinSet` (and the optional public WS input feeders), and exits if the WS server, any receiver, or a
-feeder task returns. **Exception:** the WS listener is bound *eagerly* before its task is spawned,
-and a bind failure (e.g. the port is already taken) is logged as a warning and the process runs
-without the sink — never fatal, so a taken WS port can't crash-loop the container and tear down the
-DoubleZero tunnel.
+instrument snapshot. `main.rs` selects the *candidate* feeds (`--feed`, or all of
+`ingest::feeds::FEEDS` by default), builds the shared `Arbiter` around the broadcast `Sender`, and
+hands everything to the **subscription reconciler** (`ingest::reconcile`), which is the single
+activation authority: it decides *which* of those feeds (plus the WS sink and the shred forwarder)
+actually run, based on what the host is subscribed to. `main.rs`'s top-level `select!` then awaits
+the reconciler plus the independently-spawned public WS input feeders and metrics endpoint; the
+process exits only if one of those tasks panics.
+
+**Activation is subscription-driven and dynamic** (`ingest::reconcile` + `ingest::subscriptions`):
+the reconciler polls the host's multicast subscriptions from `doublezero status --json` every
+`--subscription-refresh-secs` and reconciles the running task set — spawning receivers for
+newly-subscribed feeds, aborting ones that go away, bringing the **WS sink up only when ≥1
+market-data feed is subscribed** (bound non-fatally: a taken port disables the sink but never
+crash-loops the tunnel), and restarting the shred forwarder when its subscribed source set changes.
+It is **default-on with fail-open**: no `doublezero` CLI (running from source) → the static
+always-on set; a transient CLI error → keep current activations. `--subscription-gating-disable`
+forces the static model. A single feed dying no longer exits the process — the reconciler respawns
+it on the next tick.
 
 Ingest has **two source transports** that converge on one shared `arbiter` before the broadcast:
 the always-on DZ Edge **multicast** receivers, and optional **public WebSocket** feeders (off by
@@ -53,8 +64,9 @@ race in the arbiter's one per-`(venue, symbol)` floor (see `ingest/arbiter.rs` b
 cross-source duplicates collapse and the public copy fills in only when the edge gaps.
 
 Modules are grouped by role under `src/`:
-- **`ingest/`** — the source→`FeedMessage` pipeline (always on): `feeds`, `receiver`,
-  `processor`, `book`, `subscriber`, `arbiter`, the optional public feeders (`public_feeder`
+- **`ingest/`** — the source→`FeedMessage` pipeline: `feeds`, `receiver`, `processor`, `book`,
+  `subscriber`, `arbiter`, the **`subscriptions`** detector + **`reconcile`** activation loop (which
+  decide what runs — see Architecture above), the optional public feeders (`public_feeder`
   scaffolding + `ws_feeder`/`phoenix_feeder` venues), and the codecs (`codec`, `codec_common`,
   `codec_midpoint`, `codec_mbo`). Intra-pipeline references use `crate::ingest::*`; this half knows
   nothing about how the data is re-served.
@@ -91,10 +103,24 @@ Modules are grouped by role under `src/`:
 - **root** — `model` (shared wire types/clocks/snapshots) and `main`.
 
 - **`ingest/feeds.rs`** — the hardcoded feed registry: each `Feed` is one multicast group mapped to one
-  venue, with a `FeedKind` (which protocol) and `FeedPorts` (`TwoPort` for TOB/Midpoint, or
-  `ThreePort` adding a snapshot port for MBO). `FEEDS` is the built-in list; add a row to ingest
-  another venue (sibling-protocol rows are added once their live endpoints are known). `--feed
-  <venue>` selects a subset; consumers then filter by venue over the WS.
+  venue, with a group `code` (`tiredsolid`/`scottsdale` — the identifier `doublezero status` reports,
+  matched by the reconciler), a `FeedKind` (which protocol) and `FeedPorts` (`TwoPort` for
+  TOB/Midpoint, or `ThreePort` adding a snapshot port for MBO). `FEEDS` is the built-in list; add a
+  row to ingest another venue (sibling-protocol rows are added once their live endpoints are known).
+  `--feed <venue>` selects a subset; consumers then filter by venue over the WS.
+- **`ingest/subscriptions.rs`** — the single **detection** place. `detect()` shells out to
+  `doublezero status --json` and returns the host's subscribed group **codes** (the `S:<code>`
+  entries of `multicast_groups` — the authoritative per-host view), plus a code→IP map from
+  `multicast group list` (`shred::discovery::parse_group_code_ips`) for the shred groups (market-data
+  IPs come from `FEEDS`). Classifies into `market_data_feeds()` (subscribed enabled feeds) and
+  `shred_sources()` (subscribed `edge-solana-*` → `ip:port`). Sync `Command` soft-fail; the
+  `Detected` enum distinguishes `CliMissing` (fail open) from `Unavailable` (transient, keep current).
+- **`ingest/reconcile.rs`** — the **activation authority**. `Reconciler::run()` polls `detect()`
+  every `--subscription-refresh-secs`, computes the desired set (market-data receivers, WS on iff a
+  market-data feed is subscribed, shred sources), and applies the diff via a pure `plan()`
+  (spawn/abort). Owns all `JoinHandle`s; teardown is `abort()` (clean — sockets close on drop). Reaps
+  finished handles so a died feed respawns. Fail-open / `--subscription-gating-disable` route through
+  one `static_desired()`.
 - **`ingest/receiver.rs`** — the ingest hot path. All socket plumbing is **protocol-agnostic and shared**:
   `bind_multicast`, `recv_with_ts` (kernel timestamps), `wait_for_interface_ip`, the `IDLE_REJOIN`
   watchdog, `emit_status`, and `SeqTracker`. `drive()` is a generic receive loop over **N ports**
@@ -167,8 +193,9 @@ Modules are grouped by role under `src/`:
   PROTOCOL.md v1 surface: optional per-client subscribe/unsubscribe filtering (empty filter list =
   firehose), app ping/pong + server WS-ping heartbeat with idle-timeout reaping, and the limits
   (max clients/subs/inbound-rate, broadcast backpressure where a slow client drops oldest). The
-  listener is bound via `ws::bind()` (separate from `ws::serve()`) so `main.rs` can treat a bind
-  failure as non-fatal — a taken port disables the sink but leaves the tunnel running.
+  listener is bound via `ws::bind()` (separate from `ws::serve()`) so the reconciler can treat a bind
+  failure as non-fatal — a taken port disables the sink but leaves the tunnel running — and activate
+  the sink only once a market-data feed is subscribed.
 - **`model.rs`** — wire types (`NormalizedQuote`/`NormalizedTrade`/`NormalizedMidpoint`/
   `NormalizedDepth`/`NormalizedInstrument`, the `FeedMessage` tagged enum) and the `now_ns()` /
   `now_mono_ns()` clocks. The `InstrumentSnapshot` and `DepthSnapshot` are both keyed by
@@ -209,9 +236,12 @@ Modules are grouped by role under `src/`:
 - `--iface` accepts an interface name (resolved to its IPv4 via `ip -4 -o addr show`) or an IPv4
   literal directly.
 - **Source/sink activation is uniform**: a source or sink runs when its key config value is
-  non-empty/present. ws (output) is on by default (non-empty default `--ws-bind`; `--ws-bind ""`
-  disables it); the public WS input feeder is **off** by default (on when `--ws-input-coins` is
-  non-empty). README has the full activation tables.
+  non-empty/present, **and** (for the subscription-gated ones — market-data receivers, the WS sink,
+  the shred forwarder) when the reconciler sees the host subscribed to the relevant group. ws
+  (output) is *configured* by a non-empty `--ws-bind` (`--ws-bind ""` disables it outright) but only
+  *activated* when a market-data feed is subscribed; the public WS input feeder is **off** by
+  default (on when `--ws-input-coins` is non-empty) and is **not** subscription-gated. README has the
+  full activation tables.
 - No TLS on the **service surface** — the WebSocket output and multicast input target a trusted/local
   network; terminate TLS at a reverse proxy if exposed. The **one** exception is the outbound
   `wss://` client in `ingest/ws_feeder.rs` (public HL feed), which uses rustls + bundled webpki roots.

README.md

@@ -94,10 +94,19 @@ DZ_SECRET=DZ_… DZ_NAME=Custom-Container-Name curl -fsSL https://get.doublezero
 **Bridge variables.** The installer relays **any** non-empty bridge env var straight through to
 the container, so the bridge is tuned entirely from the one-liner. Common ones: `DZ_IFACE`,
 `DZ_RECV_BUF`, `WS_BIND` and the `WS_*` limits, `METRICS_BIND` (turn on the Prometheus `/metrics`
-endpoint — off by default), `RUST_LOG`, and the shred forwarder's `DZ_SHRED_*` (notably
-`DZ_SHRED_DEDUP_MODE` and `DZ_SHRED_RPC_URL`). The full list with defaults is the `Args`
+endpoint — off by default), `RUST_LOG`, the shred forwarder's `DZ_SHRED_*` (notably
+`DZ_SHRED_DEDUP_MODE` and `DZ_SHRED_RPC_URL`), and the reconciler's `DZ_SUBSCRIPTION_REFRESH_SECS`
+/ `DZ_SUBSCRIPTION_GATING_DISABLE`. The full list with defaults is the `Args`
 struct in [`src/main.rs`](src/main.rs); per-feature config lives in the [docs](docs/) (see below).
 
+> **Subscription-driven activation.** The bridge only runs the feeds this host is actually
+> subscribed to: a reconciler polls `doublezero status` every `DZ_SUBSCRIPTION_REFRESH_SECS`
+> (default 30) and activates/deactivates market-data receivers, the shred forwarder, and the
+> WebSocket sink as subscriptions change. The **WebSocket sink comes up only when a market-data feed
+> is subscribed** — so a shreds-only host serves no WS (and won't collide with an existing `:8081`
+> service) with no config. Running from source without the `doublezero` CLI, gating falls open to
+> the static always-on behaviour; `DZ_SUBSCRIPTION_GATING_DISABLE=1` forces that model explicitly.
+
 > **Logging defaults.** Unset, `RUST_LOG` defaults to `warn,doublezero_edge_connect=info`: the
 > bridge's own startup/operational lines stay at `info` while noisy dependency chatter is held to
 > `warn`. Set `RUST_LOG=debug` for verbose output. The installer also caps the container log on

docs/output-sinks.md

@@ -7,12 +7,18 @@ sink can't stall ingest. Every flag also reads from the env var shown.
 
 | Sink | Default | Enable / disable | Config flags (env) |
 |------|---------|------------------|--------------------|
-| **WebSocket** (`sinks::ws`) | **on** | on unless `--ws-bind` is empty; `--ws-bind ""` disables it | `--ws-bind` (`WS_BIND`, default `0.0.0.0:8081`) + the `--ws-*` limits |
+| **WebSocket** (`sinks::ws`) | **on when subscribed** | configured unless `--ws-bind` is empty (`--ws-bind ""` disables it); *activated* only when ≥1 market-data feed is subscribed | `--ws-bind` (`WS_BIND`, default `0.0.0.0:8081`) + the `--ws-*` limits |
 | **Metrics** (`sinks::metrics`) | **off** | on when `--metrics-bind` is non-empty | `--metrics-bind` (`METRICS_BIND`, default empty) |
 
-A sink is active when its key config value is non-empty/present; the WebSocket sink simply ships
-a non-empty default bind, so it is on unless you explicitly clear it, while the metrics endpoint
-ships an empty default, so it is off unless you give it a bind.
+The metrics endpoint is active when its key config value is non-empty. The WebSocket sink ships a
+non-empty default bind (so it's *configured* unless you clear it), but the **subscription
+reconciler** only *activates* it once this host is actually subscribed to a market-data feed —
+so a shreds-only host serves no WebSocket and can't collide with an existing `:8081` service, with
+no manual config. Its listener is bound non-fatally: a taken port disables the sink for that cycle
+(retried on the next reconcile) but never crash-loops the process or the DoubleZero tunnel. Running
+from source without the `doublezero` CLI, gating falls open and the sink is active whenever
+configured. See the main README for the reconciler flags (`--subscription-refresh-secs`,
+`--subscription-gating-disable`).
 
 ## Metrics (Prometheus)
 

docs/shred-forwarding.md

@@ -13,11 +13,15 @@ and never drops a datagram bound for a healthy one. `--shred-forward` targets sh
 local/fast sinks: sends are sequential per destination, so a slow or remote sink throttles the
 whole forwarder (and sheds load); the send sockets pin no egress interface.
 
-It **activates on discovery**: by default it shells out to `doublezero multicast group list
---json-compact` and selects the activated groups whose `code` starts with `--shred-code-prefix`
-(default `edge-solana-`), binding each on `--shred-port` (default `7733`). If the CLI is missing,
-errors, or finds no matching group, the forwarder stays off. Pass `--shred-source GROUP:PORT`
-(repeatable) to override discovery entirely.
+It **activates on subscription**: source resolution is driven by the shared subscription reconciler
+(`ingest::reconcile`, see the main README/CLAUDE.md), which reads the groups **this host is
+subscribed to** from `doublezero status` every `--subscription-refresh-secs` and picks the
+subscribed groups whose `code` starts with `--shred-code-prefix` (default `edge-solana-`), resolving
+each to its multicast IP via `doublezero multicast group list --json-compact` and binding it on
+`--shred-port` (default `7733`). The forwarder starts when ≥1 such group is subscribed, and is
+**restarted whenever that set changes** (a newly-subscribed group is picked up, a dropped one torn
+down) — no process restart needed. If the `doublezero` CLI is missing, the forwarder runs only from
+an explicit `--shred-source GROUP:PORT` (repeatable), which bypasses subscription discovery entirely.
 
 ## Deduplication
 
@@ -100,7 +104,8 @@ forwarder backpressure the **newest** datagram is shed (with a periodic drop-cou
 than blocking ingest. Discovery binds every matched group; a group this host isn't actually
 receiving on simply stays idle and periodically rejoins (harmless).
 
-Source resolution is **one-shot at startup**: if the `doublezero` CLI isn't ready when the bridge
-boots, or a group activates later, those groups aren't picked up until the process restarts
-(periodic re-discovery is a follow-up). Once a group is resolved, its receiver survives interface
-flap via the rejoin watchdog.
+Source resolution is **refreshed periodically** (every `--subscription-refresh-secs`): the bridge
+starts before `doublezero connect multicast` runs, so subscriptions aren't present at boot — the
+reconciler picks them up on a later tick, and likewise reacts to a group subscribed/unsubscribed at
+runtime by restarting the forwarder with the new source set. Once a group is running, its receiver
+survives interface flap via the rejoin watchdog.

src/ingest/feeds.rs

@@ -69,6 +69,12 @@ pub struct Feed {
     /// Venue name stamped on every instrument and message from this feed. Matches the
     /// edge-feed-spec source registry name (e.g. "Hyperliquid" for SourceID 1).
     pub venue: &'static str,
+    /// The DoubleZero multicast group **code** for this feed's group (e.g. "tiredsolid",
+    /// "scottsdale") — the identifier `doublezero status`/`multicast group list` report. The
+    /// subscription reconciler matches this against the host's subscribed `S:<code>` entries to
+    /// decide whether to activate the feed. Multiple feeds (e.g. a venue's TOB + MBO) can share one
+    /// code, since they ride the same multicast group.
+    pub code: &'static str,
     /// Which edge-feed-spec protocol this feed speaks (selects decoder + processor).
     pub kind: FeedKind,
     /// Multicast group for the feed.
@@ -99,6 +105,7 @@ pub const FEEDS: &[Feed] = &[
     // tiredsolid). Each feed gets its own receiver + reference-data state, keyed by group address.
     Feed {
         venue: "Hyperliquid",
+        code: "tiredsolid",
         kind: FeedKind::TopOfBook,
         group: Ipv4Addr::new(233, 84, 178, 15),
         ports: FeedPorts::TwoPort {
@@ -112,6 +119,7 @@ pub const FEEDS: &[Feed] = &[
     // Depth-only: TOB owns this venue's trades.
     Feed {
         venue: "Hyperliquid",
+        code: "tiredsolid",
         kind: FeedKind::MarketByOrder,
         group: Ipv4Addr::new(233, 84, 178, 15),
         ports: FeedPorts::ThreePort {
@@ -123,6 +131,7 @@ pub const FEEDS: &[Feed] = &[
     },
     Feed {
         venue: "Phoenix",
+        code: "scottsdale",
         kind: FeedKind::TopOfBook,
         group: Ipv4Addr::new(233, 84, 178, 18),
         ports: FeedPorts::TwoPort {
@@ -150,6 +159,23 @@ mod tests {
         }
     }
 
+    #[test]
+    fn every_feed_has_a_group_code() {
+        // The reconciler matches `code` against `doublezero status` subscriptions, so every row
+        // must carry one. Both Hyperliquid rows share the group `tiredsolid`; Phoenix is `scottsdale`.
+        for f in FEEDS {
+            assert!(!f.code.is_empty(), "{} {:?} has no code", f.venue, f.kind);
+        }
+        for f in FEEDS {
+            let expected = match f.venue {
+                "Hyperliquid" => "tiredsolid",
+                "Phoenix" => "scottsdale",
+                other => panic!("unexpected venue {other}"),
+            };
+            assert_eq!(f.code, expected, "{} has wrong code", f.venue);
+        }
+    }
+
     #[test]
     fn hyperliquid_has_tob_and_mbo() {
         let kinds: std::collections::HashSet<FeedKind> = FEEDS

src/ingest/mod.rs

@@ -15,5 +15,7 @@ pub mod phoenix_feeder;
 pub mod processor;
 pub mod public_feeder;
 pub mod receiver;
+pub mod reconcile;
 pub mod subscriber;
+pub mod subscriptions;
 pub mod ws_feeder;