//! Registry methods that own the BWE / pacer / peer-reap concern. //! //! Split from [`super`] because it's a distinct concern from routing //! UDP-to-client and fanning out events — and the M5.3 additions //! pushed the parent over the line-count budget. Contains the public //! accessors the forwarder uses, [`Registry::reap_dead`], and the //! per-fanout pacer-layer refresh (fires once per `MediaData`). use str0m::media::Rid; use crate::client::layer; use crate::propagate::ClientId; use oxpulse_sfu_kit::bwe::estimator::BandwidthEstimator; use super::Registry; /// Fallback simulcast ladder used when a publisher hasn't been observed /// emitting any RID yet (bootstrap window before the first MediaData /// arrives, or re-init after a track reset). Keeps the existing /// full ladder available pacer contract until per-publisher /// `active_rids` has real data. const DEFAULT_SIMULCAST_LADDER: [Rid; 3] = [layer::LOW, layer::MEDIUM, layer::HIGH]; impl Registry { /// Mutable access to the [`BandwidthEstimator`]. Exposed so the /// UDP loop / str0m-event plumbing can feed TWCC samples in /// without a circular dependency through [`super::Client`]. pub fn bandwidth_mut(&mut self) -> &mut BandwidthEstimator { &mut self.bandwidth } /// Shared read of the current estimator — used by fanout and the /// metrics gauge publisher. pub fn bandwidth(&self) -> &BandwidthEstimator { &self.bandwidth } /// Per-client pacer iterator (Phase B migration). /// /// Previously returned a registry-level Pacer. After Phase B the pacer /// is per-client (Client::pacer: SubscriberPacer). Preserved for the /// Phase 2 ProbeController seam (docs/ROADMAP.md). Do NOT remove. #[allow(dead_code)] pub fn pacers_mut(&mut self) -> impl Iterator { self.clients.iter_mut().map(|c| &mut c.pacer) } /// Drop dead clients and update metrics + BWE + pacer state. /// Moved from [`super`] so the single code path touches every /// per-peer pool that needs eviction on disconnect. /// /// Also clears the per-peer Prometheus label series on /// `sfu_bandwidth_estimate_bps{peer_id}` and /// `sfu_pacer_layer_total{peer_id, rid}` — otherwise reconnect /// churn would grow `peer_id` cardinality without bound /// (every new connection allocates a fresh [`ClientId`]). pub fn reap_dead(&mut self) { let detector = &mut self.detector; let metrics = &self.metrics; let bandwidth = &mut self.bandwidth; self.clients.retain(|c| { let alive = c.is_alive(); if !alive { detector.remove_peer(&c.id.0); // CAST INVARIANT (cross-crate newtype): kit `ClientId` and // crate `ClientId` both wrap a `u64`. Numeric round-trip is // valid while both sides stay u64-backed; revisit on kit // representation change. (Not `// SAFETY:` — no `unsafe`.) bandwidth.reap_dead(oxpulse_sfu_kit::propagate::ClientId(*c.id)); // Per-client SubscriberPacer drops automatically with c. metrics.client_disconnect_total.inc(); metrics.active_participants.dec(); // Scrub label series so peer_id cardinality doesn't // grow forever across reconnects. `remove_label_values` // returns `Err` when the series was never observed // (e.g. client disconnected before a BWE sample landed // or before the pacer picked a tier) — non-fatal. let peer_label = (*c.id).to_string(); let _ = metrics .bandwidth_estimate_bps .remove_label_values(&[&peer_label]); let _ = metrics .client_delivered_media_count .remove_label_values(&[&peer_label]); // Task #18: drop pacer_tick_throttled_total{peer_id} series so // SFU_PACER_FLOOR reconnect churn doesn't grow cardinality. let _ = metrics .pacer_tick_throttled_total .remove_label_values(&[&peer_label]); for rid_label in PACER_RID_LABELS { let _ = metrics .pacer_layer_total .remove_label_values(&[&peer_label, rid_label]); // M6.1: scrub layer_transitions_total{from,to,peer} series // for this peer to prevent cardinality growth on reconnects. // NOTE (#435): PACER_RID_LABELS is a compile-time const // (&["q","h","f","other"]) — always non-empty, always valid. // The nested loop is O(4²)=16 iterations by design: removes // every (from,to) RID pair. Rust's slice iterator is always // bounds-safe; no runtime validation needed. for to_label in PACER_RID_LABELS { let _ = metrics.layer_transitions_total.remove_label_values(&[ rid_label, to_label, &peer_label, ]); } } // F2b-2: drop chat_relay_{tx,rx}_bytes_total{dc, client_id} // series for both dc values. `client_id` label equals the // ClientId u64 stringified — same format as `peer_label`. // `chat_relay_dropped_total` uses (dc, reason) only — no // client_id — so it is not in scope. for dc in ["data", "ctrl"] { let _ = metrics .chat_relay_tx_bytes_total .remove_label_values(&[dc, &peer_label]); let _ = metrics .chat_relay_rx_bytes_total .remove_label_values(&[dc, &peer_label]); } // Phase 8 T10: drop voice_relay_{tx,rx}_bytes_total{client_id} // series. Label schema: single `client_id` label (no `dc`). // `voice_relay_dropped` uses (reason) only — no client_id — // so it does not need scrubbing here. let _ = metrics .voice_relay_tx_bytes_total .remove_label_values(&[&peer_label]); let _ = metrics .voice_relay_rx_bytes_total .remove_label_values(&[&peer_label]); // Phase 2c review fix (MAJOR 1): drop bwe-hint counter series // on disconnect so reconnect churn doesn't grow cardinality. let _ = metrics .sfu_bwe_hint_received_total .remove_label_values(&[&peer_label]); let _ = metrics .sfu_bwe_hint_throttled_total .remove_label_values(&[&peer_label]); // MAJOR-2: decrement voice_relay_active_channels gauge. // with_voice_dc increments on open; we mirror it here on // reap so reconnect storms don't monotonically inflate the // gauge. Only decrement when the client actually had a voice // DC — relay clients (voice_data_cid == None) never inc'd. if c.voice_data_cid.is_some() { metrics .voice_relay_active_channels .with_label_values(&["voice"]) .dec(); } // chat gauge dec — mirror voice path (followup to T10 MAJOR-2). // with_chat_dcs increments both {data} and {ctrl} on open; // decrement both here on reap. Only decrement when the client // called with_chat_dcs — relay-origin clients that skipped it // have chat_data_cid == None and never inc'd the gauge. if c.chat_data_cid.is_some() { metrics .chat_relay_active_channels .with_label_values(&["data"]) .dec(); } if c.chat_ctrl_cid.is_some() { metrics .chat_relay_active_channels .with_label_values(&["ctrl"]) .dec(); } // reactions gauge dec — only when the gauge was actually // incremented (Event::ChannelOpen fired). Using the // `reactions_dc_opened` flag (not `reactions_dc_cid.is_some()`) // prevents double-dec and guards against v0.12.22 clients that // called `with_reactions_dc` but never completed SCTP DCEP. if c.reactions_dc_opened { metrics .chat_relay_active_channels .with_label_values(&["reactions"]) .dec(); } } alive }); // 2026-05-06 post-mortem: keep `active_rooms` in lockstep with // population. Single-room SFU → 0 if empty, else 1. Idempotent. if self.clients.is_empty() { self.metrics.active_rooms.set(0); } // Update solo_since after reap: // * 0 clients → room empty, clear. // * 1 client → start solo clock if not already running. // * ≥2 clients → multiple peers, clear. match self.clients.len() { 0 => { self.solo_since = None; } 1 => { if self.solo_since.is_none() { self.solo_since = Some(std::time::Instant::now()); } } _ => { self.solo_since = None; } } } /// Consult the pacer for each subscriber's current tier and /// publish the bandwidth-estimate / layer-selection metrics. /// Invoked once per `MediaData` fanout pass; cheap (O(clients)). /// /// `origin` identifies the *publisher* whose RIDs drive the /// `available_rids` pacer input. Subscribers other than the origin /// pick their layer from whatever the publisher has been seen /// emitting (see [`crate::client::Client::active_rids`]); the /// origin themselves still runs so the pacer state and metrics /// stay in sync across self-fanout passes. /// /// `now` is threaded in from [`super::poll::Registry::fanout_pending`] /// (one clock read per drain pass) so the task #18 min-tick floor below /// is deterministic under test, mirroring the kit's own /// `update_pacer_layers` (`oxpulse_sfu_kit::registry::drive`). pub(super) fn update_pacer_layers(&mut self, origin: ClientId, now: std::time::Instant) { // Snapshot the publisher's currently-emitted RIDs before the // mut-loop so the borrow checker lets us index other clients. // Empty ⇒ bootstrap / non-simulcast; substitute the full ladder // so we don't silently wedge subscribers at audio-only before // the first packet. let publisher_rids: Vec = self .clients .iter() .find(|c| c.id == origin) .map(|c| c.active_rids()) .unwrap_or_default(); let available: &[Rid] = if publisher_rids.is_empty() { &DEFAULT_SIMULCAST_LADDER } else { &publisher_rids }; let floor_enabled = crate::pacer_floor::pacer_floor_enabled(); // G3 P3b: hoist the BWE temporal-cap kill-switch read once per pass. // The cap's ≥100ms floor is independent of SFU_PACER_FLOOR — both // flags feed the SAME `pacer_tick_ready` call (single call per // client/pass), so the cap engages even when the pacer floor is off // and vice-versa. The draft's double-call of the SIDE-EFFECTING // `pacer_tick_ready` made the feature DEAD under the intended prod // config (SFU_PACER_FLOOR=1) — do not reintroduce that. let bwe_cap_enabled = crate::bwe_temporal_cap::bwe_temporal_cap_enabled(); for client in self.clients.iter_mut() { // CAST INVARIANT: same u64-backed `ClientId` rule as in // `reap_dead` above. Update both sides if kit changes the // representation. // // V0 (issue #2310): fetch the estimate AND which ceiling term binds // the kit's combined_bps min()-chain in ONE lookup. `budget` (`.0`) // is bit-identical to the previous `estimate_bps` result, so the // pacer decision and the bandwidth_estimate_bps gauge are unchanged // — purely additive observability. let (budget, binding_term) = match self.bandwidth.estimate_with_term( oxpulse_sfu_kit::propagate::ClientId(*client.id), std::time::Instant::now(), ) { Some((bps, term)) => (Some(bps), Some(term)), None => (None, None), }; let prev_layer = client.desired_layer; let peer_label = (*client.id).to_string(); // Task #18 (ADR-13-style min-tick floor): this loop runs on every // ~20-30ms MediaData. Without a floor, two below-threshold ticks // land inside that window and collapse the kit's SUSPEND_STREAK // debounce (intended ~200ms horizon) down to ~40ms, so a brief // loss burst forces a spurious SuspendVideo. Gate the FSM- // advancing call at most once per PACER_MIN_TICK_INTERVAL (100ms) // per subscriber; a throttled tick keeps last tick's chosen layer // and skips straight to the metrics snapshot below, so gauges // stay fresh even on a throttled pass. Behind SFU_PACER_FLOOR, // default off (see `crate::pacer_floor`). // // G3 P3b: the BWE temporal-cap's ≥100ms floor shares the SAME // `pacer_tick_ready` call — `tick_ready` is computed once per // client/pass and gates BOTH the pacer FSM advance and the cap // selection. When neither flag is on, `tick_ready` is trivially // `true` (no `pacer_tick_ready` call → no `last_pacer_tick` side // effect), preserving byte-identical FSM behaviour. let tick_ready = if floor_enabled || bwe_cap_enabled { client.pacer_tick_ready(now, oxpulse_sfu_kit::bwe::PACER_MIN_TICK_INTERVAL) } else { true }; let chosen = if floor_enabled && !tick_ready { self.metrics .pacer_tick_throttled_total .with_label_values(&[&peer_label]) .inc(); Some(client.desired_layer) } else { // GoogCC is now embedded in BandwidthEstimator::PerSubscriber and // applied as a ceiling inside combined_bps() → estimate_with_term(). // A separate merge gate here would double-count GoogCC. Trust // the kit (anatolykoptev/oxpulse-sfu-kit issue #17 resolved in // v0.11.4). The `budget` value from estimate_with_term() already // incorporates the GoogCC ceiling alongside Kalman + native + hint // (its `.0` is bit-identical to the former estimate_bps() result). client.pacer_select_layer(budget, available) }; // G4: on an ACTUAL layer switch (desired layer changed AND // video is still active — `chosen` is Some), proactively // request a keyframe on the TARGET RID so the subscriber's // decoder gets a keyframe for the new layer immediately // instead of freezing until a reactive PLI (visible stutter) // or the publisher's next periodic keyframe. Worst on // slow/variable internet where the pacer switches exactly // when the user can least afford a freeze. Throttled // per-target-RID inside `emit_keyframe_on_layer_switch` // (mirrors `request_keyframe_throttled`'s MIN_GAP pattern) so // a rapid up/down oscillation cannot storm the publisher. // `chosen == Some(prev_layer)` covers NoChange and // clamped-to-same; `chosen == None` covers SuspendVideo / // GoAudioOnly (no keyframe while video is suspended). if chosen != Some(prev_layer) { if let Some(new_rid) = chosen { client.emit_keyframe_on_layer_switch(new_rid, &mut self.to_propagate); } } // Phase 2c: emit PeerSuspended when pacer tier changes. // `pending_tier_emit` is set by pacer_select_layer based on PacerAction; // None means no state transition occurred this tick. if let Some(new_tier) = client.pending_tier_emit.take() { let changed = match &client.last_emitted_tier { Some(prev) => prev != &new_tier, // First emission: emit only if not VideoMax (boot assumption // is that peers start in video-max state). None => !matches!(new_tier, crate::propagate::SuspendTier::VideoMax), }; if changed { let peer_id = client.id; client.last_emitted_tier = Some(new_tier.clone()); self.to_propagate .push_back(crate::propagate::Propagated::PeerSuspended { peer_id, tier: new_tier, }); } } if let Some(bps) = budget { self.metrics .bandwidth_estimate_bps .with_label_values(&[&peer_label]) .set(bps as i64); } // V0 (issue #2310): record which term is binding combined_bps this // pass. term-only label (no peer_id) — federated-safe, aggregated // across all subscribers. Emitted on EVERY pass, including // pacer-throttled ones, because the ceiling is binding regardless of // whether the pacer FSM advanced this tick. if let Some(term) = binding_term { self.metrics .combined_bps_binding_term .with_label_values(&[term.as_str()]) .inc(); } // Per-peer media-throughput snapshot — diagnoses // 'connected but no media' cases. Cheap (atomic load // + gauge set). self.metrics .client_delivered_media_count .with_label_values(&[&peer_label]) .set(client.delivered_media_count() as i64); if let Some(rid) = chosen { self.metrics .pacer_layer_total .with_label_values(&[&peer_label, rid_label_for(rid)]) .inc(); // M6.1: record a transition only when the layer actually changed. if rid != prev_layer { self.metrics .layer_transitions_total .with_label_values(&[ rid_label_for(prev_layer), rid_label_for(rid), &peer_label, ]) .inc(); } } // G3 P3b: SFU-BWE-driven per-subscriber temporal-layer cap. // Runs only when the kill-switch is on AND the tick was not // throttled (the cap's ≥100ms floor shares `tick_ready` with // the pacer floor). NOT routed through PacerAction — the cap // is a pure function of `budget_bps` + the Schmitt state in // `bwe_vfm_temporal_cap`. Cap-only label (no peer_id) → 3 // series, federated-safe. #[cfg(feature = "vfm")] if bwe_cap_enabled && tick_ready { let cap = client.bwe_select_temporal_cap(budget); let label = match cap { 0 => "0", 1 => "1", _ => "uncapped", }; self.metrics .bwe_temporal_cap_total .with_label_values(&[label]) .inc(); } } } } /// Every RID label `update_pacer_layers` could ever emit onto /// `sfu_pacer_layer_total`. Kept in lockstep with [`rid_label_for`] — /// if you add a variant there, mirror it here so `reap_dead` scrubs /// the matching label series on disconnect. /// /// Visible to sibling registry modules so [`super::Registry::insert`]'s /// session-steal eviction path can run the same scrub without /// duplicating the list. pub(super) const PACER_RID_LABELS: &[&str] = &["q", "h", "f", "other"]; /// Map a simulcast `Rid` to its Prometheus label (`q` / `h` / `f`). fn rid_label_for(rid: Rid) -> &'static str { if rid == layer::LOW { "q" } else if rid == layer::MEDIUM { "h" } else if rid == layer::HIGH { "f" } else { "other" } }