diff --git a/README.md b/README.md index 9dd313e..7c04937 100644 --- a/README.md +++ b/README.md @@ -826,17 +826,26 @@ Three consequences worth knowing: **Audio in its own `#EXT-X-MEDIA` rendition.** Some providers — pluto on every device cohort — offer no muxed variant at all: every `#EXT-X-STREAM-INF` defers its audio to a separate rendition playlist. Following the variant alone would ring, and serve, **video only**. The engine therefore rings the -**pair**: one ring entry holds the video segment *and* its audio partner, matched on the upstream media -sequence, and the entry URL answers with a small **master we author** over two media playlists of our own -(`…/o//v.m3u8` and `…/o//a.m3u8`). - -Two properties make this safe, and both are load-bearing: - +**pair**: one ring entry holds the video segment *and* its audio partner, and the entry URL answers with a +small **master we author** over two media playlists of our own (`…/o//v.m3u8` and +`…/o//a.m3u8`). + +Three properties make this safe, and all three are load-bearing: + +- **The pair is matched on the wall clock, not the sequence number.** `#EXT-X-PROGRAM-DATE-TIME` dates the + media itself, so it survives a renumbering; the media sequence only *looks* like a cross-rendition identity. + Pluto renumbers the two renditions independently across a session renewal — its stitcher ends the playlist + every ~25 s — so a fresh video playlist can open at sequence 10 against the audio's 11 **for the same + media**, and index pairing then puts every pair of that session about one segment out. The sequence index + remains the fallback for a source that publishes no PDT, where an aligned pair resolves to the same segment + either way. Each lane's *own* sequence still matters once a partner is picked: an absent `#EXT-X-KEY` IV is + derived from it (RFC 8216 §5.2), and the two lanes' numbers are exactly what diverge. - **One offset, both lanes.** A single affine shift is computed from the *video* lane's DTS and applied to both renditions, so the source's authored A/V skew is translated rather than replaced. Computing an offset per lane would manufacture a lip-sync error that was not in the source. A skew guard declines the pair - outright if the two renditions ever drift more than half a second apart, and a declined pair publishes - **both** lanes verbatim so they stay in sync with each other. + outright if the lanes' offset ever moves more than half a second from the skew locked on the first pair — + it bounds the DRIFT, not the skew's own magnitude, which a source is free to author as large as it likes — + and a declined pair publishes **both** lanes verbatim so they stay in sync with each other. - **Both lanes get the PID remap.** An ad creative is JIT-transmuxed into separate video and audio sources with their own arbitrary PSI, so the pids churn on *both* sides of a pod edge — normalising only the video would leave the audio track dying at every break. diff --git a/proxy/src/edge.rs b/proxy/src/edge.rs index f0c407a..daeba5a 100644 --- a/proxy/src/edge.rs +++ b/proxy/src/edge.rs @@ -299,7 +299,7 @@ mod tests { #[test] fn classifies_stream_mounts() { assert_eq!(stream_source("/api/v1/dlhd/aHR0cA"), Some("dlhd")); - assert_eq!(stream_source("/api/ext/v1/dami/h/aHR0cA"), Some("dami")); + assert_eq!(stream_source("/api/ext/v1/pluto/h/aHR0cA"), Some("pluto")); assert_eq!(stream_source("/api/v1/dulo/x"), Some("dulo")); } diff --git a/proxy/src/origin.rs b/proxy/src/origin.rs index 6d295ae..270dab7 100644 --- a/proxy/src/origin.rs +++ b/proxy/src/origin.rs @@ -58,6 +58,16 @@ const MIN_SEGMENTS: usize = 3; /// is then instant (the ring is still warm) instead of paying a fresh resolve + prebuffer. const IDLE_GRACE: Duration = Duration::from_secs(30); +/// How long a STRUCTURAL decline is remembered after the ingest that discovered it died. +/// +/// The verdict is expensive — a Node resolve round-trip plus an upstream entry fetch — and it is stable by +/// definition ("this shape can never be ringed"), so re-deriving it on every manifest poll is pure waste. +/// It still has to EXPIRE: a provider that starts publishing fMP4 can stop again, and the memo outlives the +/// task that could otherwise re-test it. Aged from when the decline was recorded, not from last access, so a +/// channel under continuous polling is still retried on this cadence rather than staying declined for as +/// long as someone keeps watching it. +const INELIGIBLE_MEMO_TTL: Duration = Duration::from_secs(60); + /// Cadence for the "still idle?" check. Cheap — it only wakes to compare two integers and a timestamp. const IDLE_TICK: Duration = Duration::from_secs(5); @@ -305,12 +315,24 @@ pub struct Origin { /// `resolve_media`; read by `wait_ready`, which is what lets `serve_entry`/`serve_ts` decline and hand the /// request back to the ordinary rewrite path. ineligible: RwLock>, + /// When `ineligible` was recorded — the memo's OWN age, which is what `INELIGIBLE_MEMO_TTL` is measured + /// against. Deliberately not `last_access`: polling the channel must not keep a stale decline alive. + ineligible_at: Mutex>, /// Set once the ingest learns this upstream is DEMUXED — the audio rendition it rings beside the video. /// /// Read by Side-2 to pick the renderer: the HLS one authors an `#EXT-X-MEDIA` from it (so the client still /// sees the track labelled as upstream labelled it), and `serve_ts` routes to the INTERLEAVING producer /// (`ts_ring_pair_producer` → `tsweave`), which folds the pair into one program on the way out. demuxed_audio: RwLock>, + /// Which KIND of playlist the entry was last answered with — `true` master, `false` media playlist. + /// + /// Diagnostic only, and deliberately so. `serve_entry` re-decides the kind from `demuxed_audio` on every + /// poll, so a mid-session flip changes what one URL *is*, and RFC 8216 gives a client no reload semantics + /// for a media playlist that has become a master — players fail the load. There is no in-session repair + /// (after the flip, EITHER shape describes a ring the client's session cannot use), and pinning the kind + /// would only trade a failed reload for silent video. So the transition is recorded and named instead: + /// without it, "the stream just stopped" is indistinguishable from a stall in every log we keep. + last_entry_master: RwLock>, /// S3/UND — the last structural fault that retired an upstream on this channel, and how many upstreams /// have been retired for one. Reported on the `iop` health frame so a channel quietly hopping providers /// is visible in Active Streams, not only in the log. That invisibility is what let a false positive run @@ -350,13 +372,24 @@ pub enum Lane { Audio, } +/// Seeds `Origin::generation` so that no two origins — including two incarnations of the SAME channel — ever +/// share one. +/// +/// `reset_ring` bumps the generation, which is what makes a stale segment URL fail cleanly after a failover. +/// That guard only ever compared a URL against the CURRENT registry entry, so it held within one origin and +/// not across them: a respawned origin used to start at generation 0 with `next_seq` back at 0, re-issuing +/// the exact `-.ts` names its predecessor had already handed out. A client holding a stale manifest +/// then got a 200 and DIFFERENT media, which is precisely what the generation exists to prevent. Process-wide +/// and monotonic, because the value only ever has to differ — nothing reads meaning into it. +static GENERATION_SEED: AtomicU64 = AtomicU64::new(1); + impl Origin { fn new(ring_cap_bytes: u64) -> Self { Self { ring: RwLock::new(VecDeque::new()), ring_bytes: AtomicU64::new(0), next_seq: AtomicU64::new(0), - generation: AtomicU64::new(0), + generation: AtomicU64::new(GENERATION_SEED.fetch_add(1, Ordering::Relaxed)), subscribers: AtomicU32::new(0), last_access: Mutex::new(Instant::now()), target_duration_ms: AtomicU64::new(0), @@ -368,7 +401,9 @@ impl Origin { evicted_segments: AtomicU64::new(0), disc_seq: AtomicU64::new(0), ineligible: RwLock::new(None), + ineligible_at: Mutex::new(None), demuxed_audio: RwLock::new(None), + last_entry_master: RwLock::new(None), last_suspect: RwLock::new(None), suspect_retires: AtomicU32::new(0), upstream_shape: RwLock::new(None), @@ -396,9 +431,18 @@ impl Origin { /// Record a STRUCTURAL mismatch and wake anyone waiting on a window that is never coming. fn mark_ineligible(&self, reason: String) { *self.ineligible.write_ok() = Some(reason); + *self.ineligible_at.lock_ok() = Some(Instant::now()); self.notify.notify_waiters(); } + /// Whether this origin carries a decline still worth trusting — the whole reason a dead ingest's registry + /// entry is kept rather than removed. `subscribe` reads it to answer from the memo instead of paying for + /// a resolve; anything older than `INELIGIBLE_MEMO_TTL` reads as "re-test it". + fn declined_recently(&self) -> bool { + self.ineligible.read_ok().is_some() + && self.ineligible_at.lock_ok().is_some_and(|t| t.elapsed() < INELIGIBLE_MEMO_TTL) + } + /// Append a segment and evict from the front until the ring fits its byte cap. /// /// Returns how many segments were evicted. The `MIN_SEGMENTS` floor is enforced HERE rather than by the @@ -573,13 +617,25 @@ pub fn subscribe(state: &AppState, source: &str, entry: &str, pl: Option<&str>, let (origin, start) = { let mut map = state.origins().lock_ok(); match map.get(&key) { - Some(o) => { + // A LIVE ingest — share it. `stopping` is the whole test: the teardown sets it before it drops + // the registry entry, so between those two moments the key is still present while the task that + // serves it has already left its loop. Reusing that origin hands the caller a lease on a ring + // nothing will refill, and (worse) reports `start = false`, so no replacement is ever spawned. + Some(o) if !o.stopping.load(Ordering::Relaxed) => { // A re-resolve may have changed the cap; apply it to the live ring so raising the dial takes // effect without a restart (it shrinks lazily, as new pushes evict against the new cap). o.ring_cap_bytes.store(cap, Ordering::Relaxed); (o.clone(), false) } - None => { + // A DEAD ingest that recorded a structural decline. Keep answering from the memo: `wait_ready` + // short-circuits on `ineligible`, so the caller falls back to the manifest rewrite immediately — + // no resolve, no spawn, no telemetry — which is the whole point of not re-deriving a verdict that + // cost a Node round-trip and an upstream fetch. Nothing is spawned, so nothing sweeps this entry: + // the TTL inside `declined_recently` is what eventually retires it, via the arm below. + Some(o) if o.declined_recently() => (o.clone(), false), + // Either no entry, a dead one, or a decline that has aged out. Insert REPLACES the stale entry, + // which is why the ingest guard removes only an origin it still owns. + _ => { let o = Arc::new(Origin::new(cap)); map.insert(key.clone(), o.clone()); (o, true) @@ -611,6 +667,7 @@ struct PollPlaylists { audio: Option<(Url, String)>, } +#[derive(Clone)] struct IngestCtx { state: AppState, origin: Arc, @@ -620,6 +677,57 @@ struct IngestCtx { key: String, } +/// Runs the ingest teardown on EVERY exit — including a panic. +/// +/// It used to be a tail of `ingest`'s body, which covered every `break` and nothing else. A panic in the +/// segment or PSI parsers unwound straight past it and left the registry holding an entry with +/// `stopping == false`: `subscribe` then saw a live ingest that did not exist, returned `start = false` +/// forever, and the channel could never be respawned — not even by an entry poll, which is the one path that +/// recovers every other kind of ingest death. The raw-TS producers, meanwhile, sat on `wait_for_segment` +/// waking every 30 s to re-read a flag that was never going to flip. +struct IngestGuard { + ctx: IngestCtx, + rid: String, +} + +impl Drop for IngestGuard { + fn drop(&mut self) { + // Mark the ingest dead BEFORE dropping the registry entry, and wake every reader. Without this a + // raw-TS producer parked on `wait_for_segment` would keep re-waiting forever against a ring nothing + // refills. + self.ctx.origin.stopping.store(true, Ordering::Relaxed); + self.ctx.origin.notify.notify_waiters(); + // A STRUCTURAL decline is KEPT: the entry is the memo, and `subscribe` answers from it until the TTL + // retires it. Everything else is removed — but only if the key still holds THIS origin. `subscribe` + // replaces a stopping entry, so a slow teardown could otherwise delete a healthy successor that has + // already taken the key. + if self.ctx.origin.ineligible().is_some() { + // The memo is the VERDICT, never the media. A decline can land on a re-resolve long after the + // ring filled (an upstream that turns fMP4 mid-session), and a retained ring is served, not just + // held: `wait_ready` tests `ring_depth` BEFORE `ineligible`, so a full window answers `Ready::Yes` + // and the channel plays a frozen loop until the memo ages out — while the ring's RAM is pinned for + // as long as the entry survives, which is unbounded because nothing sweeps it. Dropping the window + // frees the bytes AND lets `wait_ready` fall through to the decline, which is the whole point. + self.ctx.origin.reset_ring(); + } else { + let mut map = self.ctx.state.origins().lock_ok(); + if map.get(&self.ctx.key).is_some_and(|o| Arc::ptr_eq(o, &self.ctx.origin)) { + map.remove(&self.ctx.key); + } + } + report_iop(&self.ctx, "closed"); + log::info("iop", &self.rid, || { + format!( + "ingest stop {}/{} — {} segment(s), {} MiB ingested", + self.ctx.source, + crate::proxy::host_of(&self.ctx.entry), + self.ctx.origin.ingested_segments.load(Ordering::Relaxed), + self.ctx.origin.ingested_bytes.load(Ordering::Relaxed) / (1024 * 1024) + ) + }); + } +} + /// The Side-1 loop: resolve → follow the media playlist → fetch + decrypt each new segment → push to the ring. /// /// Structured as one task per channel so backpressure, failover and shutdown are all local to it. Every line @@ -630,6 +738,9 @@ async fn ingest(ctx: IngestCtx) { log::info("iop", &rid, || { format!("ingest start {}/{}", ctx.source, crate::proxy::host_of(&ctx.entry)) }); + // Armed FIRST, so every exit below — `break`, or a panic out of a parser — is a clean teardown. It holds + // its own handle on the context because the loop keeps using `ctx` by value throughout. + let _guard = IngestGuard { ctx: ctx.clone(), rid: rid.clone() }; let mut prev = PrevSeg::default(); let mut next_upstream_seq: i64 = -1; @@ -669,14 +780,10 @@ async fn ingest(ctx: IngestCtx) { // whether `resolve_media` found an audio rendition — but both are held so a re-resolve can change shape // without rebuilding the task. let mut pair_splicer = crate::tsnorm::PairSplicer::new(); - let mut warned_splice: std::collections::HashSet = std::collections::HashSet::new(); + let mut warned_splice: std::collections::HashSet<&'static str> = std::collections::HashSet::new(); let mut warned_pairing = false; // Latch for the one-shot line naming which key is pairing this source's two renditions. let mut pairing_logged = false; - // S3/UND — the undecodable-upstream detector. Scoped to `playerSelectable` sources (dlhd): retiring an - // upstream is only useful where there are alternates to walk to, and every other source would just - // re-resolve the same dead provider on a 2 s loop. - let undecodable_watch = ctx.source == "dlhd"; let mut probe_segments: u32 = 0; // The fault currently repeating, and how many segments in a row have shown it. let mut suspect_run: Option<(crate::tsseg::Suspect, u32)> = None; @@ -828,6 +935,17 @@ async fn ingest(ctx: IngestCtx) { continue; } }; + // S3/UND — the undecodable-upstream detector. Scoped to sources that HAVE alternates to walk to: + // retiring an upstream is only useful where another one can take over, and on a single-upstream + // source the retirement would just re-resolve the same dead provider on a 2 s loop. It used to be + // `ctx.source == "dlhd"`, the crate's only hardcoded provider id. + // + // Read per poll off the MOUNT source's policy, exactly like this loop's other knobs (headers, + // timeouts, allow_private) — not off the serving candidate's. The two differ only after a failover + // onto another provider, whose grant files its policy under its own `policySource`; for attempt 0 + // they are the same object. Worth knowing when reading this: a child's capability does not flip the + // parent's watch, which is the existing behaviour of every knob here rather than a rule of this one. + let undecodable_watch = policy.player_selectable.load(Ordering::Relaxed); let client = ctx.state.client_for( policy.connect_timeout_ms.load(Ordering::Relaxed), policy.max_redirects.load(Ordering::Relaxed), @@ -859,7 +977,7 @@ async fn ingest(ctx: IngestCtx) { let audio_seg = match &ap { None => None, Some((aurl, apl)) => match pair_audio(seg, upstream_seq, apl) { - PairPick::Found(a) => Some((aurl.clone(), a.clone())), + PairPick::Found(a, aseq) => Some((aurl.clone(), a.clone(), aseq)), // The audio partner is already gone, so this pair can never be completed. Drop the video // segment too: publishing it unpaired would put the two playlists on different windows, // and the next segment's sequence check reports the gap honestly. @@ -941,11 +1059,14 @@ async fn ingest(ctx: IngestCtx) { None => continue, // a gap: the NEXT ingested segment will see the sequence jump and splice }; - // The audio half, against the RENDITION's own base url and key cache. The implicit-IV derivation - // (RFC 8216 §5.2) uses the same media sequence, which is exactly what pairing established. + // The audio half, against the RENDITION's own base url, own key cache and — the part PDT pairing + // changed — its OWN media sequence. RFC 8216 §5.2 derives an absent IV from the segment's media + // sequence, and pairing on the wall clock exists precisely because the two lanes' sequences + // diverge: handing the video's number to a renumbered audio lane decrypts its first CBC block + // (the leading TS packet, usually the PAT) against the wrong IV. let audio_plain = match &audio_seg { None => None, - Some((aurl, aseg)) => { + Some((aurl, aseg, aseq)) => { let aseg_url = match aurl.join(&aseg.uri) { Ok(u) => u, Err(_) => continue, @@ -957,7 +1078,7 @@ async fn ingest(ctx: IngestCtx) { } policy.hosts.write_ok().insert(h.to_lowercase()); } - match fetch_segment(&ctx, &rid, &client, &policy, aurl, aseg, &aseg_url, upstream_seq, read_timeout_ms, &mut audio_key_cache).await { + match fetch_segment(&ctx, &rid, &client, &policy, aurl, aseg, &aseg_url, *aseq, read_timeout_ms, &mut audio_key_cache).await { Some(b) => Some(b), // No partner bytes ⇒ no pair. Dropping BOTH keeps the two published windows aligned; // the next segment's sequence check turns the hole into an honest splice. @@ -1118,13 +1239,16 @@ async fn ingest(ctx: IngestCtx) { // nobody received. Upstream's splice is then signalled the old way — a visible decoder reset // beats a stream we mis-rewrote. On the paired path it is all-or-nothing: both lanes go verbatim // together, so they stay in sync with each other. - let mut declined: Option = None; + // The stable cause travels WITH the message: the latch below keys on the former, the log prints + // the latter. Keying on the message is what defeated the latch — every one of these reasons + // interpolates a live measurement. + let mut declined: Option<(&'static str, String)> = None; let (plain, audio_out, absorbed) = match (normalize, audio_plain) { (false, a) => (plain, a, false), (true, Some(araw)) => match pair_splicer.normalize_pair(&plain, &araw) { Some((v, a)) => (Bytes::from(v), Some(Bytes::from(a)), true), None => { - declined = Some(pair_splicer.last_decline().to_string()); + declined = Some((pair_splicer.last_decline_slug(), pair_splicer.last_decline().to_string())); pair_splicer.reset(); (plain, Some(araw), false) } @@ -1132,7 +1256,11 @@ async fn ingest(ctx: IngestCtx) { (true, None) => match splicer.normalize(&plain) { Some(b) => (Bytes::from(b), None, true), None => { - declined = Some("no PSI, or a program shape the published layout cannot carry".to_string()); + // One fixed sentence, so it is already its own key. + declined = Some(( + "muxed-no-psi", + "no PSI, or a program shape the published layout cannot carry".to_string(), + )); splicer.reset(); (plain, None, false) } @@ -1141,8 +1269,8 @@ async fn ingest(ctx: IngestCtx) { // Latched per DISTINCT reason rather than once per ingest. One latch hid the thing that matters — // whether a pod edge declines for the same cause every time (a shape to handle) or for a // different one each time (a bug in this pass). - if let Some(why) = declined { - if warned_splice.insert(why.clone()) { + if let Some((cause, why)) = declined { + if warned_splice.insert(cause) { log::warn("iop", &rid, || { format!("splice normalisation declined — {why}; publishing verbatim and signalling the splice") }); @@ -1320,26 +1448,19 @@ async fn ingest(ctx: IngestCtx) { } } - // Mark the ingest dead BEFORE dropping the registry entry, and wake every reader. Without this a raw-TS - // producer parked on `wait_for_segment` would keep re-waiting forever against a ring nothing refills. - ctx.origin.stopping.store(true, Ordering::Relaxed); - ctx.origin.notify.notify_waiters(); - ctx.state.origins().lock_ok().remove(&ctx.key); - report_iop(&ctx, "closed"); - log::info("iop", &rid, || { - format!( - "ingest stop {}/{} — {} segment(s), {} MiB ingested", - ctx.source, - crate::proxy::host_of(&ctx.entry), - ctx.origin.ingested_segments.load(Ordering::Relaxed), - ctx.origin.ingested_bytes.load(Ordering::Relaxed) / (1024 * 1024) - ) - }); + // The teardown itself is `IngestGuard::drop`, which runs here and on every other way out of this task. } /// Which audio segment partners a video one — the outcome of the pairing lookup. enum PairPick<'a> { - Found(&'a SegRef), + /// The partner segment, plus **its own** absolute media sequence. + /// + /// The sequence has to travel with the pick because the whole point of PDT pairing is that the two lanes + /// renumber independently — so the video lane's number is not a stand-in for the audio lane's, and RFC + /// 8216 §5.2 derives an absent `#EXT-X-KEY` IV from the segment's OWN media sequence. On the index + /// fallback the two are equal by construction, which is why passing the video's was correct until PDT + /// pairing existed. + Found(&'a SegRef, i64), /// The audio window has already rolled past this video segment; the pair can never complete. RolledPast, /// The audio lane has not published this far yet — HOLD the video segment and retry next poll. @@ -1385,19 +1506,21 @@ fn pair_tolerance_ms(video_duration: f64, target_duration: f64) -> i64 { /// today changes behaviour: an aligned pair resolves to the same segment either way. fn pair_audio<'a>(video: &SegRef, upstream_seq: i64, apl: &'a crate::tsmux::MediaPlaylist) -> PairPick<'a> { if let Some(vt) = video.pdt_ms { - let mut best: Option<(&SegRef, i64)> = None; - for a in &apl.segments { + // The INDEX rides along with the pick: the audio lane's own sequence is `media_sequence + index`, and + // this is the only path that can land on an index the video's sequence does not name. + let mut best: Option<(&SegRef, i64, usize)> = None; + for (i, a) in apl.segments.iter().enumerate() { let Some(at) = a.pdt_ms else { continue }; let d = (at - vt).abs(); - if best.is_none_or(|(_, bd)| d < bd) { - best = Some((a, d)); + if best.is_none_or(|(_, bd, _)| d < bd) { + best = Some((a, d, i)); } } // Only trust this path when the audio lane actually dates itself; a lane with no PDT at all falls // through to the index rather than being declared "rolled past". - if let Some((a, d)) = best { + if let Some((a, d, i)) = best { if d <= pair_tolerance_ms(video.duration, apl.target_duration) { - return PairPick::Found(a); + return PairPick::Found(a, apl.media_sequence + i as i64); } // Out of tolerance: which SIDE decides hold-vs-drop. Before the window ⇒ the partner is already // gone; after it ⇒ it has not been published yet. @@ -1411,7 +1534,9 @@ fn pair_audio<'a>(video: &SegRef, upstream_seq: i64, apl: &'a crate::tsmux::Medi return PairPick::RolledPast; } match apl.segments.get(idx as usize) { - Some(a) => PairPick::Found(a), + // `media_sequence + idx` IS `upstream_seq` here, by the line above — written out rather than reusing + // the video's number so the two arms state the same rule. + Some(a) => PairPick::Found(a, apl.media_sequence + idx), None => PairPick::NotYet, } } @@ -1464,6 +1589,13 @@ async fn resolve_media(ctx: &IngestCtx, rid: &str, escalate: bool, reason: Optio }); // Its own write site: this arm returns before the manifest handling below ever runs. *ctx.origin.upstream_shape.write_ok() = Some("ts".to_string()); + // …and so does the `demuxed_audio` write, which is why it has to be repeated here. A bare TS socket + // is ONE muxed stream (`push_cut` pushes `audio: None`), so leaving a previous demuxed resolve's + // rendition in place outlives the upstream that justified it — and every reader of the flag then + // describes a ring that no longer exists: `serve_entry` keeps authoring a master, `serve_playlist` + // keeps listing an audio lane whose segments 404, and `serve_ts` keeps dispatching to the pair + // producer, which declines to its cap and ends the socket on a flag the reconnect re-reads unchanged. + *ctx.origin.demuxed_audio.write_ok() = None; return Some(MediaSource::RawTs(Box::pin(stream), first)); } // A manifest: drain the (small) remainder into text. @@ -2002,6 +2134,57 @@ async fn wait_ready(origin: &Arc, rid: &str) -> Ready { } } +/// The viewer HEARTBEAT, in one place. +/// +/// A manifest poll is what keeps a viewer "active" — exactly as on the proxy path, the difference being that +/// these bytes came from RAM, so no upstream fetch was involved and none is reported. Every endpoint that +/// answers a manifest owes one, and the shape must be identical everywhere: `noteViewer` keys on +/// (ip, ua, username, channel), so a field that drifts between the entry and the lane endpoints would split +/// one client into two viewers on demuxed channels only — an accounting bug visible as a phantom viewer +/// rather than as an error. It was three copies of this literal before; one is enough. +fn note_viewer(state: &AppState, mount_path: &str, source: &str, entry: &str, id: &crate::proxy::Identity, bytes: usize) { + state.report(serde_json::json!({ + "kind": "viewer", "source": source, "entryUrl": entry, + "ip": id.ip, "ua": id.ua, "username": id.username, + "playerType": if mount_path == "/api/ext/v1" { "externalPlayer" } else { "appPlayer" }, + "bytes": bytes as u64, + })); +} + +/// Snapshot the published window and its discontinuity counter, IN THAT ORDER. +/// +/// The order is the point, and it is why this is a function rather than two lines at each call site. The two +/// are separate atomics, so an eviction landing between them leaves a one-poll skew either way — but reading +/// the COUNTER first makes it an UNDER-count, where the tag is still in the window and the client counts it +/// itself. The other order double-counts it. (Monotonicity holds regardless: the counter only ever rises.) +/// Every renderer needs both, and a rule this easy to reverse should exist once. +fn window_snapshot(origin: &Origin) -> (u64, Vec>) { + let disc_seq = origin.disc_seq(); + (disc_seq, origin.window()) +} + +/// Name the moment the entry's playlist KIND changes, once per change. +/// +/// A client that fetched a media playlist reloads that same URL for the rest of its session; getting a master +/// back is not a shape it has any handling for. The flip itself is unavoidable — a NEW client must be told the +/// truth about the ring it is joining — so what this buys is the ability to read a session that died at +/// exactly this instant as what it was, rather than as an unexplained stall. See `Origin::last_entry_master`. +fn note_entry_shape(origin: &Origin, is_master: bool, rid: &str) { + // ONE lock acquisition: swap the current shape in and judge what was there. `Some(!is_master)` is the + // whole "it changed" test — the field is a bool, so the only other populated value IS the other shape. + if origin.last_entry_master.write_ok().replace(is_master) == Some(!is_master) { + let name = |m: bool| if m { "master" } else { "media playlist" }; + log::warn("oop", rid, || { + format!( + "entry shape changed {} → {} mid-session (the upstream re-resolved to the other lane shape) — \ + clients already polling this URL will fail their next reload and have to start a new session", + name(!is_master), + name(is_master) + ) + }); + } +} + /// SIDE-2 ENTRY: subscribe, wait for a playable window, and serve OUR manifest. /// /// The lease is dropped when this returns — a polling client renews it on every poll, and the ingest's idle @@ -2031,28 +2214,20 @@ pub async fn serve_entry( Ready::Ineligible => return None, Ready::TimedOut => return Some(crate::proxy::text(503, "stream warming up: no playable window yet")), } - // Read the counter BEFORE snapshotting the window. The two are separate atomics, so an eviction landing - // between them leaves a one-poll skew either way — but this order makes it an UNDER-count, where the tag - // is still in the window and the client counts it itself. The other order double-counts it. (Monotonicity - // holds regardless: the counter only ever rises.) // A DEMUXED origin answers the entry with an authored MASTER over the two lanes; a muxed one answers // with the single media playlist, byte-identically to before pairing existed. - if let Some(m) = origin.demuxed_audio() { + let demuxed = origin.demuxed_audio(); + note_entry_shape(&origin, demuxed.is_some(), rid); + if let Some(m) = demuxed { origin.touch(); // the master itself carries no segments, but it is still a live client let body = render_master(mount_path, source, entry, &m, token, pl); log::info("oop", rid, || { format!("origin master served (1 variant + audio rendition \"{}\", {} bytes)", m.audio.name, body.len()) }); - state.report(serde_json::json!({ - "kind": "viewer", "source": source, "entryUrl": entry, - "ip": id.ip, "ua": id.ua, "username": id.username, - "playerType": if mount_path == "/api/ext/v1" { "externalPlayer" } else { "appPlayer" }, - "bytes": body.len() as u64, - })); + note_viewer(state, mount_path, source, entry, id, body.len()); return Some(crate::proxy::raw(200, "application/vnd.apple.mpegurl", body.into_bytes())); } - let disc_seq = origin.disc_seq(); - let window = origin.window(); + let (disc_seq, window) = window_snapshot(&origin); let body = render_media_playlist( &window, origin.target_duration(), @@ -2068,24 +2243,22 @@ pub async fn serve_entry( log::info("oop", rid, || { format!("origin manifest served ({} segment(s), {} bytes)", window.len(), body.len()) }); - // A manifest poll is the viewer heartbeat, exactly as on the proxy path — the difference is that these - // bytes came from RAM, so no upstream fetch was involved and none is reported. - state.report(serde_json::json!({ - "kind": "viewer", "source": source, "entryUrl": entry, - "ip": id.ip, "ua": id.ua, "username": id.username, - "playerType": if mount_path == "/api/ext/v1" { "externalPlayer" } else { "appPlayer" }, - "bytes": body.len() as u64, - })); + note_viewer(state, mount_path, source, entry, id, body.len()); Some(crate::proxy::raw(200, "application/vnd.apple.mpegurl", body.into_bytes())) } /// SIDE-2 PLAYLIST: one lane's authored media playlist, for a demuxed origin's authored master. /// -/// Answered straight from the registry like `serve_segment` — no resolve, no Node round-trip. It does NOT -/// `subscribe`: the ingest already exists (the master could not have been served otherwise), and `window()` -/// refreshes `last_access`, which is the half of the idle check that a lease-less reader can keep alive. That -/// matters because the master is fetched ONCE — if the heartbeat stayed only on the entry, every demuxed -/// channel would look idle after `IDLE_GRACE` and be reaped out from under a watching client. +/// SUBSCRIBES, exactly as `serve_entry` does — and that is the whole difference between a demuxed session +/// that survives an ingest death and one that does not. `window()` refreshing `last_access` keeps a polling +/// client's origin alive, which is why the idle sweep never reaps one out from under a watcher; but it cannot +/// bring an ingest BACK. Every other kind of ingest exit — upstream ENDLIST, a structural decline, empty-poll +/// exhaustion, a panic — removes the registry entry, and a demuxed client fetches its master exactly ONCE, so +/// it has no reason ever to touch a subscribing endpoint again: it would poll a 404 forever while a muxed +/// client, whose player re-reads the entry, is respawned transparently. +/// +/// The policy comes from the cache rather than a resolve, so this stays a RAM-only path: no Node round-trip, +/// no upstream fetch, nothing `buildGrant`'s stored-entry gate has to see. #[allow(clippy::too_many_arguments)] pub async fn serve_playlist( state: &AppState, @@ -2098,18 +2271,40 @@ pub async fn serve_playlist( id: &crate::proxy::Identity, rid: &str, ) -> axum::response::Response { - let key = crate::state::target_key(source, entry); - let origin = match state.origins().lock_ok().get(&key) { - Some(o) => o.clone(), - None => { - log::warn("oop", rid, || format!("playlist {lane:?}: no live ingest for {source}")); + // The gate on restarting an ingest from a LANE poll: this (source, entry) must be one we have actually + // resolved. `hop_policy` would answer here too, but its fallback is the mount SOURCE's policy, which + // exists for any source that ever served anything — so it would let a client spawn an ingest, and the + // Node resolve round-trips behind it, for any entry string it cared to encode into an `o/` URL. These + // routes are otherwise side-effect-free by design. A session that legitimately needs the restart always + // has the record: its entry was resolved to serve the master it is polling the lanes of. + let Some(policy) = state.resolved_target_policy(source, entry) else { + log::warn("oop", rid, || format!("playlist {lane:?}: {source} entry was never resolved here — not starting an ingest")); + return crate::proxy::text(404, "not found: no live ingest"); + }; + let lease = subscribe(state, source, entry, pl, &policy); + let origin = lease.origin().clone(); + match wait_ready(&origin, rid).await { + Ready::Yes => {} + // A lane URL has no rewrite fallback to hand back to — the client is already inside our authored + // master — so the honest answer is the same 404 a lane that carries no media gets. + Ready::Ineligible => { + log::warn("oop", rid, || format!("playlist {lane:?}: this upstream cannot be ringed")); return crate::proxy::text(404, "not found: no live ingest"); } - }; - // Same ordering rule as `serve_entry`: read the counter BEFORE the window so a concurrent eviction - // under-counts rather than double-counts. - let disc_seq = origin.disc_seq(); - let window = origin.window(); + Ready::TimedOut => return crate::proxy::text(503, "stream warming up: no playable window yet"), + } + let (disc_seq, window) = window_snapshot(&origin); + // The audio lane is rendered off the SAME ladder as the video one, so nothing downstream checks that the + // entries actually carry a second lane — a ring holding muxed segments would publish a full ladder of + // `-a` URIs and 404 every one of them at `serve_segment`. Fail the playlist instead: one honest 404 the + // client can act on beats a valid-looking rendition whose every segment misses. (A non-empty window is + // required so a ring caught mid-reset reads as "not yet", which is what the empty ladder already says.) + if matches!(lane, Lane::Audio) && !window.is_empty() && window.iter().all(|s| s.audio.is_none()) { + log::warn("oop", rid, || { + format!("playlist {lane:?}: the ring holds no audio lane ({} segment(s)) — 404", window.len()) + }); + return crate::proxy::text(404, "not found: lane not carried"); + } let body = render_media_playlist( &window, origin.target_duration(), @@ -2125,14 +2320,8 @@ pub async fn serve_playlist( log::info("oop", rid, || { format!("origin manifest served ({lane:?} lane, {} segment(s), {} bytes)", window.len(), body.len()) }); - // The viewer heartbeat. `noteViewer` keys on (ip, ua, username, channel), so the two lanes' polls - // collapse to ONE viewer rather than double-counting the client. - state.report(serde_json::json!({ - "kind": "viewer", "source": source, "entryUrl": entry, - "ip": id.ip, "ua": id.ua, "username": id.username, - "playerType": if mount_path == "/api/ext/v1" { "externalPlayer" } else { "appPlayer" }, - "bytes": body.len() as u64, - })); + // Both lanes' polls collapse to ONE viewer — see `note_viewer` for why the shape must not drift. + note_viewer(state, mount_path, source, entry, id, body.len()); crate::proxy::raw(200, "application/vnd.apple.mpegurl", body.into_bytes()) } @@ -2312,8 +2501,11 @@ async fn ts_ring_producer( // viewer joins the ring at a different point and therefore sits on its own timeline. let mut splicer = crate::tsnorm::Splicer::new(); let mut warned_splice = false; + // Two paths reach the close emit now — the ingest-stopping `break` and the lane-changed `break 'outer` + // — so the reason has to be threaded, exactly as the pair producer threads its own. + let mut close_reason = "ingest_stopped"; - loop { + 'outer: loop { let window = origin.window(); // Fell off the back of the ring — the client reads slower than the ingest writes, or the ring is too // small for this bitrate. Skipping forward drops video, so say so plainly rather than silently gapping. @@ -2339,22 +2531,47 @@ async fn ts_ring_producer( let from = next_seq; // snapshot: the filter closure borrows it, the body reassigns it for seg in window.iter().filter(|s| s.seq >= from) { next_seq = seg.seq + 1; + // THE RING TURNED DEMUXED under this socket. `demuxed` was sampled once, at open (`serve_ts`), so + // a re-resolve onto a master with a separate audio rendition leaves this producer emitting + // `seg.bytes` — which is now the VIDEO lane alone. Silent video, for the rest of the session. + // + // End the socket, which is the same recovery the pair producer takes on the opposite flip: the + // client reconnects, `serve_ts` re-reads the flag, and dispatches to `ts_ring_pair_producer`. + // Unlike a weave decline this needs no cap — an entry carrying an audio lane is not a transient + // shape, it is the ingest having changed what the ring holds. + if seg.audio.is_some() { + log::warn("oop", &ctx.rid, || { + format!("ring turned demuxed at seq={} — ending the muxed socket so the client reconnects into the interleaving producer", seg.seq) + }); + close_reason = "lane_changed"; + break 'outer; + } // Declining is a designed outcome: a segment carrying no PSI, or a program shape the published // layout cannot express, is served verbatim. That reinstates the upstream splice for that one // segment — a visible glitch — which still beats emitting a stream we mis-rewrote. let body = match normalize.then(|| splicer.normalize(&seg.bytes)).flatten() { Some(bytes) => Bytes::from(bytes), None => { - // Only a genuine DECLINE is worth a warning. With the switch off there is nothing to - // decline — serving verbatim is the requested behaviour, and saying otherwise would send - // an operator hunting a stream shape that was never the problem. - if normalize && !warned_splice { - warned_splice = true; - log::warn("oop", &ctx.rid, || { - "splice normalisation declined (no PSI, or a program shape the published layout \ - cannot carry) — serving upstream timestamps as-is" - .to_string() - }); + // Only a genuine DECLINE is worth acting on. With the switch off there is nothing to + // decline — serving verbatim is the requested behaviour, the guard above already dropped + // the timeline once, and warning would send an operator hunting a stream shape that was + // never the problem. + if normalize { + // DROP THE TIMELINE, same as the ingest's identical arm and this producer's own + // ring-skip path. The client is about to receive this segment on UPSTREAM timestamps, + // so anchoring the NEXT one against the published clock would stamp it behind media + // the client already holds — a backwards jump, on a bare TS socket that has no + // `#EXT-X-DISCONTINUITY` to explain it. Re-anchoring instead keeps the output + // contiguous with what actually went out. + splicer.reset(); + if !warned_splice { + warned_splice = true; + log::warn("oop", &ctx.rid, || { + "splice normalisation declined (no PSI, or a program shape the published \ + layout cannot carry) — serving upstream timestamps as-is" + .to_string() + }); + } } seg.bytes.clone() } @@ -2389,9 +2606,8 @@ async fn ts_ring_producer( if pending_bytes > 0 { ctx.state.report(serde_json::json!({ "kind": "sbytes", "streamId": stream_id, "bytes": pending_bytes })); } - // Single predecessor (the ingest-stopping break above), so this can be a literal. - ctx.state.report(serde_json::json!({ "kind": "close", "streamId": stream_id, "reason": "ingest_stopped" })); - log::info("oop", &ctx.rid, || format!("origin raw-TS session close ({stream_id})")); + ctx.state.report(serde_json::json!({ "kind": "close", "streamId": stream_id, "reason": close_reason })); + log::info("oop", &ctx.rid, || format!("origin raw-TS session close ({stream_id}, {close_reason})")); } /// SIDE-2 RAW TS, DEMUXED (S3/RMX): follow the ring, weaving each PAIR into one socket. @@ -2425,7 +2641,7 @@ async fn ts_ring_pair_producer( let mut weaver = crate::tsweave::PairWeaver::new(); // Latched per DISTINCT reason, like the ingest's — one latch would hide whether a pod edge declines for // the same cause every time (a shape to handle) or a different one each time (a bug in the pass). - let mut warned_declines: std::collections::HashSet = std::collections::HashSet::new(); + let mut warned_declines: std::collections::HashSet<&'static str> = std::collections::HashSet::new(); let mut consecutive_declines: u32 = 0; let mut warned_switch = false; // Two paths reach this producer's close emit — the ingest-stopping `break` and the declines `break 'outer` @@ -2472,15 +2688,18 @@ async fn ts_ring_pair_producer( Bytes::from(b) } None => { - let why = match seg.audio.as_ref() { - Some(_) => weaver.last_decline().to_string(), + // Cause first, message second: the latch keys on the cause, the log prints the message. + // The weave's reasons interpolate live measurements, so latching on the text meant every + // declined pair minted a new key — a warn per segment, and a set that only grew. + let (cause, why) = match seg.audio.as_ref() { + Some(_) => (weaver.last_decline_slug(), weaver.last_decline().to_string()), // The origin re-resolved onto a muxed upstream mid-session. Ending on the decline cap // is the recovery: the client reconnects and dispatches to `ts_ring_producer`. - None => "the ring entry carries no audio lane".to_string(), + None => ("no-audio-lane", "the ring entry carries no audio lane".to_string()), }; weaver.reset(); consecutive_declines += 1; - if warned_declines.insert(why.clone()) { + if warned_declines.insert(cause) { log::warn("oop", &ctx.rid, || { format!("interleave declined — {why}; skipping the pair") }); @@ -2624,8 +2843,17 @@ pub(crate) struct RingFootprint { /// would mean cloning the `Arc`s out and DROPPING the registry guard before any `ring.read_ok()`. pub(crate) fn ring_footprint(origins: &Mutex>>) -> RingFootprint { let map = origins.lock_ok(); - let mut f = RingFootprint { origins: map.len(), subscribed: 0, bytes: 0, cap_bytes: 0 }; + let mut f = RingFootprint { origins: 0, subscribed: 0, bytes: 0, cap_bytes: 0 }; for o in map.values() { + // A STOPPING entry is not a live channel and must not be counted as one. The registry keeps a + // declined origin as the memo that stops its verdict being re-derived on every poll — the ring is + // dropped, so it costs no media, but counting it would overstate both the origin count and the + // headroom this number exists to size an eviction budget against. An ordinary teardown removes its + // entry outright, so in practice a declined origin is the only thing this skips. + if o.stopping.load(Ordering::Relaxed) { + continue; + } + f.origins += 1; if o.subscribers.load(Ordering::Relaxed) > 0 { f.subscribed += 1; } @@ -2666,6 +2894,43 @@ mod tests { } } + /// The stale-segment guard has to hold ACROSS incarnations, not just within one. + /// + /// A respawned origin used to start at generation 0 with `next_seq` back at 0, so it re-issued the exact + /// `-.ts` names its predecessor had handed out and `serve_segment` answered a stale URL with a + /// 200 and different media. + #[test] + fn two_incarnations_of_a_channel_never_share_a_generation() { + let first = Origin::new(1000); + let second = Origin::new(1000); + assert_ne!( + first.generation(), + second.generation(), + "a respawned origin must not reuse its predecessor's segment-URL namespace" + ); + // And a reset inside one incarnation still moves it on, which is the guard's original job. + let before = first.generation(); + first.reset_ring(); + assert!(first.generation() > before, "reset_ring must still advance the generation"); + } + + /// The decline memo is what stops an unringable shape re-resolving on every poll — and the TTL is what + /// stops it outliving a provider that fixed itself. Nothing sweeps a retained entry (the task that would + /// have is the one that died), so the age check IS the expiry. + #[test] + fn a_decline_memo_expires_so_the_channel_is_retried() { + let o = Origin::new(1000); + assert!(!o.declined_recently(), "an origin with no verdict must never read as declined"); + + o.mark_ineligible("fMP4 (#EXT-X-MAP) is not concatenable".to_string()); + assert!(o.declined_recently(), "a fresh decline is answered from the memo"); + + // Backdate past the TTL: the next subscribe must re-test the upstream rather than trust this. + *o.ineligible_at.lock_ok() = Some(Instant::now() - INELIGIBLE_MEMO_TTL - Duration::from_secs(1)); + assert!(!o.declined_recently(), "a stale decline must expire so the shape is re-tested"); + assert!(o.ineligible().is_some(), "expiry is about the memo's AGE, not about forgetting the reason"); + } + #[test] fn ring_evicts_oldest_to_stay_under_the_byte_cap() { let o = Origin::new(1000); // 1000-byte cap @@ -2710,9 +2975,18 @@ mod tests { for (i, vs) in v.segments.iter().enumerate() { let useq = v.media_sequence + i as i64; match pair_audio(vs, useq, &a) { - PairPick::Found(p) => { + PairPick::Found(p, aseq) => { let d = (p.pdt_ms.unwrap() - vs.pdt_ms.unwrap()).abs(); assert!(d < 100, "seq {useq} paired to media {d} ms away — that is a different segment"); + // The pick must carry the AUDIO lane's own number, not the video's. This is the number + // RFC 8216 §5.2 turns into an absent-IV, so borrowing the video's here decrypts the + // partner's first CBC block against the wrong IV on exactly the renewal this test models. + assert_eq!( + aseq, + a.media_sequence + i as i64, + "seq {useq} paired correctly but reported the wrong media sequence for the partner" + ); + assert_eq!(aseq, useq + 1, "the renumbering must be visible in the reported sequence"); } _ => panic!("seq {useq} failed to pair despite the audio being present"), } @@ -2732,7 +3006,12 @@ mod tests { for (i, vs) in v.segments.iter().enumerate() { let useq = v.media_sequence + i as i64; let by_pdt = match pair_audio(vs, useq, &a) { - PairPick::Found(p) => p.uri.clone(), + PairPick::Found(p, aseq) => { + // Aligned lanes: the reported sequence must be the video's, which is what makes passing + // `upstream_seq` to the audio fetch correct on every source that pairs by index. + assert_eq!(aseq, useq, "an aligned lane must report the same sequence"); + p.uri.clone() + } _ => panic!("aligned pair must resolve"), }; let by_index = a.segments[(useq - a.media_sequence) as usize].uri.clone(); @@ -2746,7 +3025,10 @@ mod tests { let v = lane(10, 0, 3, 5.0, false); let a = lane(10, 0, 3, 5.0, false); match pair_audio(&v.segments[1], 11, &a) { - PairPick::Found(p) => assert_eq!(p.uri, "seg11.ts", "index pairing still selects positionally"), + PairPick::Found(p, aseq) => { + assert_eq!(p.uri, "seg11.ts", "index pairing still selects positionally"); + assert_eq!(aseq, 11, "the fallback reports the index it selected on"); + } _ => panic!("the fallback must still pair"), } // …including its rolled-past arm. @@ -2813,14 +3095,23 @@ mod tests { // `idle` keeps zero subscribers: it is inside its IDLE_GRACE window, still holding RAM. Counting it in // `bytes` but not in `subscribed` is the whole point — a footprint that only saw watched channels // would under-report exactly when a burst of just-closed channels is what filled memory. + // A DECLINED origin is retained in the registry as a memo, with its ring dropped and `stopping` set. + // It is not a live channel and must not appear in either count, or the number this metric exists to + // size an eviction budget against would include channels that can never ingest again. + let declined = Arc::new(Origin::new(7_000)); + declined.mark_ineligible("fMP4 (#EXT-X-MAP) is not concatenable".to_string()); + declined.stopping.store(true, Ordering::Relaxed); + let map: HashMap> = - [("a".to_string(), watched), ("b".to_string(), idle)].into_iter().collect(); + [("a".to_string(), watched), ("b".to_string(), idle), ("c".to_string(), declined)] + .into_iter() + .collect(); let f = ring_footprint(&Mutex::new(map)); - assert_eq!(f.origins, 2); + assert_eq!(f.origins, 2, "the retained decline is not a live origin"); assert_eq!(f.subscribed, 1, "the idle origin still costs RAM but has no viewer"); assert_eq!(f.bytes, 2_100, "1500 + 600 — every ring, watched or not"); - assert_eq!(f.cap_bytes, 14_000, "Σ per-channel caps: headroom, not a global ceiling"); + assert_eq!(f.cap_bytes, 14_000, "Σ per-channel caps: headroom, not a global ceiling — and not the decline's 7000"); } #[test] diff --git a/proxy/src/probe.rs b/proxy/src/probe.rs index 5228086..a59af65 100644 --- a/proxy/src/probe.rs +++ b/proxy/src/probe.rs @@ -1,6 +1,6 @@ //! The channel-probe endpoint (PRB, P1.3) — the successor to the removed streamProbe sweep; reads manifest-declared decode metadata. //! -//! Node's `sources/probeAll.ts` RESOLVES every Active channel (dulo/dlhd/dami adapter logic, throttled) then +//! Node's `sources/probeAll.ts` RESOLVES every Active channel (dulo/dlhd adapter logic, throttled) then //! POSTs the resolved `{ id, target, upstreamHeaders }` batch here. This binary FETCHES each target //! concurrently (bounded), decides liveness (a 2xx that parses as a manifest = live), and extracts the //! declared decode metadata via the SAME parser the live proxy uses (`manifest::extract_media`). It writes diff --git a/proxy/src/proxy.rs b/proxy/src/proxy.rs index 8428b5a..4d96f15 100644 --- a/proxy/src/proxy.rs +++ b/proxy/src/proxy.rs @@ -118,8 +118,11 @@ pub async fn serve_stream( }; let rid = log::rid(source, &entry); // A DEMUXED origin also publishes its two authored media playlists here, for the same reason its - // segments live here: answered from the ring, no resolve, no Node round-trip, and never seen by - // buildGrant's stored-entry gate. + // segments live here: answered from the ring, and never seen by buildGrant's stored-entry gate. + // The bytes are still RAM-only — the policy comes from the cache — but unlike `o/` segments this + // SUBSCRIBES, so a poll that finds no live ingest restarts one (which does resolve). That is what + // lets a demuxed session survive an ingest death: its client fetched the master once and has no + // other subscribing endpoint to poll. let lane = match file { "v.m3u8" => Some(crate::origin::Lane::Video), "a.m3u8" => Some(crate::origin::Lane::Audio), @@ -363,7 +366,7 @@ pub async fn serve_stream( // entry / cold hop rides the live mirror — and a failover-pinned stream never snaps back to its dead // parent) and fail this request (the player refetches). // · ENTRY — a transport failure always enters the walk: a fresh resolve of the SAME pinned candidate - // first (Node re-runs resolveStream → dlhd/dami reprobeMirror — the pre-failover mirror rotation), + // first (Node re-runs resolveStream → dlhd reprobeMirror — the pre-failover mirror rotation), // then, when failoverEnabled, the NEXT candidates in Node's order. A DEFINITIVE non-2xx enters the // walk only when failoverOnDefiniteError is on (default keeps the forward-verbatim semantics). if resp.is_none() && is_hop { diff --git a/proxy/src/state.rs b/proxy/src/state.rs index 4dba880..c2030f4 100644 --- a/proxy/src/state.rs +++ b/proxy/src/state.rs @@ -146,6 +146,14 @@ pub struct SourcePolicy { pub relabel_segment: RwLock>, /// Permit private/loopback upstream IPs (LAN sources); false for public-CDN sources. pub allow_private: AtomicBool, + /// Whether the SERVING adapter has alternate upstreams to walk to — Node's `adapter.playerSelectable`. + /// + /// The undecodable-upstream detector (S3/UND, `origin.rs`) is scoped to it: retiring a provider is only + /// useful where there is another one to retire it FOR, and on a single-upstream source the retirement + /// would just re-resolve the same dead provider on a 2 s loop. It rides the grant because that capability + /// is the adapter's, and the adapter lives in Node — the data plane used to test `source == "dlhd"`, + /// which was the crate's only hardcoded provider id and silently excluded the next such adapter. + pub player_selectable: AtomicBool, /// The growing SSRF allowlist (lowercased hosts): seed = resolved master host, grown from manifest children. pub hosts: RwLock>, /// PXY-2: the resolved proxy-config CLIENT knobs for this source's streams (from the grant). proxy.rs @@ -200,6 +208,7 @@ impl SourcePolicy { headers: RwLock::new(Vec::new()), relabel_segment: RwLock::new(None), allow_private: AtomicBool::new(false), + player_selectable: AtomicBool::new(false), hosts: RwLock::new(HashSet::new()), connect_timeout_ms: AtomicU64::new(15000), max_redirects: AtomicU32::new(10), @@ -227,6 +236,10 @@ pub struct Grant { pub relabel_segment: Option, #[serde(rename = "allowPrivate")] pub allow_private: bool, + /// S3/UND: does the serving adapter have alternate upstreams? `default` → false → an older Node degrades + /// to "no undecodable detection", which is the safe direction: the detector only ever RETIRES an upstream. + #[serde(rename = "playerSelectable", default)] + pub player_selectable: bool, // PXY-2: the resolved (Custom→Default→env) proxy config. Node already merged headerOverrides into // upstreamHeaders, so this struct declares the knobs Rust applies: connectTimeoutMs + maxRedirects (P2, // client-level), readTimeoutMs + bufferSizeKb (P3.1/RSL, per-stream) and outputFormat (hls|ts, P3.2/DST). @@ -462,7 +475,7 @@ impl AppState { /// FOG: force a FRESH resolve of a SPECIFIC candidate (bypass the target cache) and re-cache the /// result — pinning the stream's cursor to that attempt. attempt 0 = the channel itself (Node re-runs - /// `resolveStream`, which drives dlhd/dami `reprobeMirror()` — the pre-failover "mirror failover"); + /// `resolveStream`, which drives dlhd `reprobeMirror()` — the pre-failover "mirror failover"); /// attempt N >= 1 = the channel's Nth ordered failover child, resolved via the child's own adapter. pub async fn resolve_at( &self, @@ -613,21 +626,33 @@ impl AppState { /// entry record falls back to the mount source's policy (today's behavior). Touches last_access so an /// actively-polling session (hops only — HLS players rarely re-request the ENTRY) keeps its cursor. pub fn hop_policy(&self, source: &str, entry: &str) -> Option> { - if !entry.is_empty() { - let policy_key = { - let mut m = self.targets.lock_ok(); - m.get_mut(&target_key(source, entry)).map(|e| { - e.last_access = Instant::now(); - e.policy_key.clone() - }) - }; - if let Some(pk) = policy_key { - if let Some(p) = self.get(&pk) { - return Some(p); - } - } + self.resolved_target_policy(source, entry).or_else(|| self.get(source)) + } + + /// The policy for a target this process has ACTUALLY resolved — `hop_policy`'s strict half, with no + /// mount-source fallback. + /// + /// The distinction is a gate, not an optimisation. Falling back to the source's policy answers "is this + /// a source we know", which is true of every source that ever served anything; requiring the target + /// record answers "is this an entry we have resolved", which is what a caller needs before it may act on + /// an entry string a client supplied. `origin::serve_playlist` uses it for exactly that: a lane poll may + /// restart a dead ingest for a channel we were serving, and must not start one for an arbitrary URL. + /// + /// A record is written on every resolve and never swept (`expires` only governs REUSE), so this reads as + /// "resolved at some point in this process" — which is what makes it a usable gate rather than a race + /// against `TARGET_TTL`. + pub fn resolved_target_policy(&self, source: &str, entry: &str) -> Option> { + if entry.is_empty() { + return None; } - self.get(source) + let policy_key = { + let mut m = self.targets.lock_ok(); + m.get_mut(&target_key(source, entry)).map(|e| { + e.last_access = Instant::now(); + e.policy_key.clone() + }) + }; + self.get(&policy_key?) } pub fn get(&self, source: &str) -> Option> { @@ -691,6 +716,7 @@ impl AppState { *policy.headers.write_ok() = grant.upstream_headers.into_iter().collect(); *policy.relabel_segment.write_ok() = grant.relabel_segment; policy.allow_private.store(grant.allow_private, Ordering::Relaxed); + policy.player_selectable.store(grant.player_selectable, Ordering::Relaxed); // PXY-2: record the resolved client knobs so proxy.rs selects the matching upstream client per hop. policy.connect_timeout_ms.store(grant.proxy_config.connect_timeout_ms, Ordering::Relaxed); policy.max_redirects.store(grant.proxy_config.max_redirects, Ordering::Relaxed); diff --git a/proxy/src/tsmux.rs b/proxy/src/tsmux.rs index 5af16ba..4047ca3 100644 --- a/proxy/src/tsmux.rs +++ b/proxy/src/tsmux.rs @@ -19,7 +19,7 @@ //! the PCR/PTS reset); a truly seamless splice would need RMX. //! //! Durability reuses the RSL layer: playlist + segment fetches go through `fetch_with_retry` (transient retry), -//! and a persistent media-playlist failure re-resolves the entry (driving dlhd/dami `reprobeMirror` failover). +//! and a persistent media-playlist failure re-resolves the entry (driving dlhd `reprobeMirror` failover). //! Telemetry uses the SOCKET model (noteSocketViewer* — explicit open/close, a 60s no-byte backstop) rather //! than the 30s poll-recency model, since a continuous stream never polls: `open` → Node mints a connId; periodic //! `sbytes` → egress; `close` → session end. @@ -395,7 +395,11 @@ pub(crate) fn unsupported_encryption(body: &str) -> Option { /// is most of pluto and dlhd) reads as unencrypted. /// /// Returns the literal `"NONE"` rather than an Option so a consumer can distinguish MEASURED cleartext from -/// an absent reading — on this panel those mean opposite things. +/// an absent reading — on this panel those mean opposite things. Three states, then: `"NONE"` (we read the +/// playlist and it is in the clear), a METHOD we read (`"AES-128"`, `"SAMPLE-AES"`, …), and `"UNKNOWN"` (a +/// key tag IS present but its METHOD could not be read, so the content is encrypted by something we cannot +/// name). A caller gating on `!= "NONE"` therefore treats an unreadable key as encrypted, which is the safe +/// direction: the tag's presence is itself the evidence. /// /// Last key wins: a KEY applies until the next one replaces it, so what the window ENDS on is the current /// state. Only meaningful on a MEDIA playlist; a master carries no `#EXT-X-KEY` and would always answer NONE. @@ -408,7 +412,11 @@ pub(crate) fn encryption_method(body: &str) -> String { .find(|(k, _)| k.eq_ignore_ascii_case("METHOD")) .map(|(_, v)| v.trim().trim_matches('"').to_ascii_uppercase()) .unwrap_or_default(); - method = if m.is_empty() { "NONE".to_string() } else { m }; + // A key tag we cannot READ is not a reading of cleartext. RFC 8216 makes METHOD mandatory, so an + // empty one means a malformed upstream — but `NONE` is this function's word for MEASURED + // cleartext, and answering it here would tell the operator the exact opposite of what the tag's + // presence proves. `UNKNOWN` is the third state the doc above promises consumers. + method = if m.is_empty() { "UNKNOWN".to_string() } else { m }; } } method @@ -450,14 +458,36 @@ pub(crate) struct AudioRendition { pub describes_video: bool, } -/// Every `#EXT-X-MEDIA:TYPE=AUDIO` rendition in a master that carries its own `URI=`, resolved absolute. +/// ONE pass over a master's `#EXT-X-MEDIA:TYPE=AUDIO` lines, answering both questions the caller has: +/// which renditions are actually demuxed, and which GROUPS are already muxed into their variant. /// /// Per RFC 8216 §4.3.4.1 an `#EXT-X-MEDIA` WITHOUT a `URI` means that rendition is already present in the /// referencing variant's own playlist — so only the URI-bearing ones are actually demuxed, and a bare /// "does this master mention EXT-X-MEDIA" test would false-positive on every muxed stream that merely /// labels its audio track. -fn demuxed_audio_renditions(body: &str, base: &Url) -> Vec { +/// +/// THE GROUP VERDICT IS NOT THE MEMBER VERDICT, which is why both come out of the same pass. The +/// per-rendition rule is right on its own but says nothing about the group, and a group can hold both kinds. +/// Live pluto does exactly that: +/// +/// ```text +/// #EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="audio",NAME="Original",DEFAULT=YES,CHANNELS="2" ← no URI +/// #EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="audio",NAME="English",AUTOSELECT=YES,URI="…", +/// CHARACTERISTICS="public.accessibility.describes-video" +/// ``` +/// +/// The programme audio is the URI-less `DEFAULT` one — muxed into the variant — and the only URI-bearing +/// member is an audio-description track. Judging the group by its URI-bearing members alone made this look +/// demuxed, so the origin ringed the video against the DESCRIPTION and the viewer got a narrator instead of +/// the programme. One URI-less member is proof the variant is self-sufficient. +/// +/// These two verdicts USED to be two functions, each re-deriving the same line filter, `split_attrs` call, +/// quote-stripping `val` closure and TYPE=AUDIO gate over the same body. Keeping one copy is what stops a +/// future fix to attribute handling landing on one scan and not the other — which would skew exactly the +/// demuxed-vs-muxed verdict the pluto case above shows the cost of getting wrong. +fn audio_media(body: &str, base: &Url) -> (Vec, HashSet) { let mut out = Vec::new(); + let mut muxed = HashSet::new(); for l in body.split('\n') { let Some(attrs) = l.trim_start().strip_prefix("#EXT-X-MEDIA:") else { continue }; let attrs = split_attrs(attrs); @@ -470,8 +500,16 @@ fn demuxed_audio_renditions(body: &str, base: &Url) -> Vec { if !val("TYPE").is_some_and(|t| t.eq_ignore_ascii_case("AUDIO")) { continue; } - // No URI ⇒ muxed into the variant ⇒ following the variant alone still yields audio. - let Some(uri) = val("URI").filter(|u| !u.is_empty()) else { continue }; + // No URI ⇒ muxed into the variant ⇒ following the variant alone still yields audio. That is the + // GROUP's verdict, not just this member's: one URI-less rendition makes the whole group playable + // from the variant, so it is collected here rather than by a second scan that would have to + // re-derive the same attribute rules and stay in lockstep with them. + let Some(uri) = val("URI").filter(|u| !u.is_empty()) else { + if let Some(g) = val("GROUP-ID") { + muxed.insert(g); + } + continue; + }; let (Some(group), Ok(url)) = (val("GROUP-ID"), base.join(&uri)) else { continue }; let yes = |k: &str| val(k).is_some_and(|v| v.eq_ignore_ascii_case("YES")); out.push(AudioRendition { @@ -486,49 +524,9 @@ fn demuxed_audio_renditions(body: &str, base: &Url) -> Vec { .is_some_and(|c| c.to_ascii_lowercase().contains("public.accessibility.describes-video")), }); } - out + (out, muxed) } -/// Audio groups that contain at least one rendition WITHOUT a `URI` — i.e. groups whose audio is already -/// inside the referencing variant. -/// -/// The per-rendition rule (`demuxed_audio_renditions` skips URI-less entries) is right on its own but says -/// nothing about the GROUP, and a group can hold both kinds. Live pluto does exactly that: -/// -/// ```text -/// #EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="audio",NAME="Original",DEFAULT=YES,CHANNELS="2" ← no URI -/// #EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="audio",NAME="English",AUTOSELECT=YES,URI="…", -/// CHARACTERISTICS="public.accessibility.describes-video" -/// ``` -/// -/// The programme audio is the URI-less `DEFAULT` one — muxed into the variant — and the only URI-bearing -/// member is an audio-description track. Judging the group by its URI-bearing members alone made this look -/// demuxed, so the origin ringed the video against the DESCRIPTION and the viewer got a narrator instead of -/// the programme. RFC 8216 §4.3.4.1 is unambiguous that a URI-less rendition is present in the variant, so -/// one such member is proof the variant is self-sufficient. -fn muxed_audio_groups(body: &str) -> HashSet { - let mut out = HashSet::new(); - for l in body.split('\n') { - let Some(attrs) = l.trim_start().strip_prefix("#EXT-X-MEDIA:") else { continue }; - let attrs = split_attrs(attrs); - let val = |k: &str| { - attrs - .iter() - .find(|(a, _)| a.eq_ignore_ascii_case(k)) - .map(|(_, v)| v.trim().trim_matches('"').to_string()) - }; - if !val("TYPE").is_some_and(|t| t.eq_ignore_ascii_case("AUDIO")) { - continue; - } - if val("URI").is_some_and(|u| !u.is_empty()) { - continue; // a demuxed member says nothing about the group - } - if let Some(g) = val("GROUP-ID") { - out.insert(g); - } - } - out -} /// Which rendition of a group to follow: `DEFAULT=YES` wins, else `AUTOSELECT=YES`, else the first the master /// listed. The same order a player would apply with no user preference expressed. @@ -596,10 +594,10 @@ struct VariantAttrs { /// the rendition to follow beside it. A passthrough raw-TS caller falls back to the HLS rewrite on that; the /// origin rings the pair, and its raw-TS renderer interleaves it (`tsweave`) rather than declining. pub(crate) fn pick_variant(body: &str, base: &Url) -> Option { - let renditions = demuxed_audio_renditions(body, base); // A group holding even ONE URI-less rendition has its audio inside the variant, so the variant is - // playable on its own and must not be treated as demuxed — see `muxed_audio_groups`. - let muxed = muxed_audio_groups(body); + // playable on its own and must not be treated as demuxed — see `audio_media`, which decides both in + // one pass so the two verdicts can never drift apart. + let (renditions, muxed) = audio_media(body, base); let demuxed: HashSet = renditions.iter().map(|r| r.group.clone()).filter(|g| !muxed.contains(g)).collect(); let mut best: Option<(i64, String, VariantAttrs)> = None; // audio-safe variants only @@ -1234,6 +1232,16 @@ mod tests { // at all are both genuinely cleartext. assert_eq!(encryption_method("#EXTM3U\n#EXT-X-KEY:METHOD=NONE\n#EXTINF:6.0,\nseg0.ts\n"), "NONE"); assert_eq!(encryption_method("#EXTM3U\n#EXTINF:6.0,\nseg0.ts\n"), "NONE"); + + // …and a key tag whose METHOD is missing or empty is NEITHER. RFC 8216 requires the attribute, so + // this is a malformed packager — but the tag's presence proves the content is encrypted by SOMETHING, + // and reporting the measurement for cleartext would state the opposite of the only fact available. + let no_method = "#EXTM3U\n#EXT-X-KEY:URI=\"k.key\",IV=0x0123\n#EXTINF:6.0,\nseg0.ts\n"; + assert_eq!(encryption_method(no_method), "UNKNOWN", "an unreadable key tag is not cleartext"); + let empty_method = "#EXTM3U\n#EXT-X-KEY:METHOD=,URI=\"k.key\"\n#EXTINF:6.0,\nseg0.ts\n"; + assert_eq!(encryption_method(empty_method), "UNKNOWN"); + // The gate every consumer uses must read both as encrypted. + assert_ne!(encryption_method(no_method), "NONE"); } #[test] diff --git a/proxy/src/tsnorm.rs b/proxy/src/tsnorm.rs index 5253e0f..974362b 100644 --- a/proxy/src/tsnorm.rs +++ b/proxy/src/tsnorm.rs @@ -770,6 +770,14 @@ pub(crate) struct PairSplicer { /// bug in this pass — the same lesson that got the two URL-shape splice heuristics removed. Every early /// return below names itself, and the layout ones print what they were actually handed. last_decline: String, + /// The CAUSE of that decline, as a stable key — the format string the message was built from, with the + /// per-occurrence measurements still to be filled in. + /// + /// The two are not interchangeable. Callers latch a "warn once per distinct reason" set on this, and + /// keying that on `last_decline` instead defeats it completely: every message here interpolates live + /// numbers (a drift in ms, a pid dump), so on the churning sources this pass exists for, every single + /// decline mints a new key — a warn per segment forever, and a set that only grows. + last_slug: &'static str, } /// A 33-bit clock delta read as a SIGNED millisecond offset — skews are small either side of zero, and @@ -795,6 +803,7 @@ impl PairSplicer { audio_cc: HashMap::new(), locked_skew: None, last_decline: String::new(), + last_slug: "", } } @@ -803,6 +812,11 @@ impl PairSplicer { &self.last_decline } + /// The stable cause behind `last_decline` — what a per-reason log latch must key on. See `last_slug`. + pub(crate) fn last_decline_slug(&self) -> &'static str { + self.last_slug + } + /// Forget the timeline, both layouts and the latched skew. Same contract as `Splicer::reset`. pub(crate) fn reset(&mut self) { self.clock.reset(); @@ -820,17 +834,23 @@ impl PairSplicer { /// Rewrite one paired segment. `None` ⇒ NEITHER lane may be published rewritten. pub(crate) fn normalize_pair(&mut self, video: &[u8], audio: &[u8]) -> Option<(Vec, Vec)> { // Every arm names itself, so the `iop` log can say WHICH stage refused this pair. + // + // The FORMAT STRING doubles as the cause key: it is a `&'static str` that is constant per site, while + // everything that varies between occurrences lives in the interpolated arguments. That split is what + // lets a caller latch one warning per cause without either losing the measurements from the message or + // minting a new key for every drift value. A new decline site gets a distinct key for free. macro_rules! decline { - ($($why:tt)*) => {{ - self.last_decline = format!($($why)*); + ($fmt:literal $(, $arg:expr)* $(,)?) => {{ + self.last_slug = $fmt; + self.last_decline = format!($fmt $(, $arg)*); return None; }}; } macro_rules! need { - ($e:expr, $($why:tt)*) => { + ($e:expr, $fmt:literal $(, $arg:expr)* $(,)?) => { match $e { Some(v) => v, - None => decline!($($why)*), + None => decline!($fmt $(, $arg)*), } }; } diff --git a/proxy/src/tsweave.rs b/proxy/src/tsweave.rs index 9015ad2..a917d9e 100644 --- a/proxy/src/tsweave.rs +++ b/proxy/src/tsweave.rs @@ -50,18 +50,31 @@ const NULL_PID: u16 = 0x1FFF; /// also forced at the head of every woven pair, so a client never waits a whole interval to find the program. const PSI_INTERVAL_PKTS: usize = 250; -/// How far BEFORE the video lane's first timestamp the merge's sort origin sits, in 90 kHz ticks (2 s). +/// How far before the video lane's first timestamp the merge's sort origin sits AT MINIMUM, in 90 kHz ticks +/// (2 s). /// /// The sort compares `forward_gap(anchor, key)`, which is wrap-safe but one-directional: a key that sits /// *before* the anchor reads as almost a full wrap ahead and would sort last. Audio may legitimately lead -/// video (a negative skew) and carried-over audio may precede the next pair's first frame, but `PairSplicer`'s -/// `SKEW_TOLERANCE` already bounds both at 0.5 s — so 2 s is four times the headroom that guard allows, and -/// still six orders of magnitude inside the 33-bit wrap. +/// video (a negative skew) and carried-over audio may precede the next pair's first frame. +/// +/// A FLOOR, not a bound — and it was written as a bound. The justification used to be that `PairSplicer`'s +/// `SKEW_TOLERANCE` caps the lead at 0.5 s, so 2 s was "four times the headroom". That reading is wrong: +/// `SKEW_TOLERANCE` bounds how far one pair's skew may DRIFT from the LOCKED skew, never the locked skew's +/// own magnitude, and the first pair latches whatever it finds unconditionally. Pairing on +/// `#EXT-X-PROGRAM-DATE-TIME` accepts a partner up to HALF A SEGMENT away (2.5 s on pluto's 5 s ladder), so a +/// locked lead past 2 s is reachable — and then EVERY audio key sat before the anchor, so every audio unit +/// sorted after every video unit and the socket stepped ~2.5 s backwards at each pair seam. +/// +/// `weave` now widens the anchor to whatever actually leads within the block. This constant only keeps the +/// ordinary case byte-identical to what it was. const ANCHOR_BACKSTOP: u64 = 2 * 90_000; -/// Ceiling on the audio held back for the next pair (see `weave` step 5). A trailing run bounded by the 0.5 s -/// skew guard is a few KiB; anything approaching this cap means the lanes are not shaped the way the guard -/// implies, so the carry is abandoned and that audio is emitted in place rather than growing without bound. +/// Ceiling on the audio held back for the next pair (see `weave` step 5). A trailing run is a few KiB on any +/// normally-shaped pair; anything approaching this cap means the lanes are not shaped like a pair at all, so +/// the carry is abandoned and that audio is emitted in place rather than growing without bound. +/// +/// Deliberately NOT justified by `SKEW_TOLERANCE` — see `ANCHOR_BACKSTOP` for why that guard bounds drift +/// rather than lead, and why a bound derived from it would be a bound on nothing. const MAX_CARRY_BYTES: usize = 256 * 1024; /// Which buffer a `Unit`'s bytes live in. Indices, not references, so the emit step can hold `&mut self`. @@ -123,6 +136,10 @@ pub(crate) struct PairWeaver { /// `PairSplicer::last_decline` — a decline that only says "declined" cannot distinguish a shape this pass /// genuinely cannot carry from a bug in this pass. last_decline: String, + /// The stable CAUSE of that decline — the format string, before its measurements are filled in. Callers + /// latch their per-reason warning on this; see `PairSplicer::last_slug` for why the message itself cannot + /// serve as the key. + last_slug: &'static str, } impl PairWeaver { @@ -136,6 +153,7 @@ impl PairWeaver { carry_units: Vec::new(), since_psi: 0, last_decline: String::new(), + last_slug: "", } } @@ -144,6 +162,12 @@ impl PairWeaver { &self.last_decline } + /// The stable cause behind it, for a per-reason log latch. Forwarded unchanged when the decline came from + /// the splicer, so one key names one cause across both layers. + pub(crate) fn last_decline_slug(&self) -> &'static str { + self.last_slug + } + /// Forget the timeline and the seam carry-over. Same contract as `PairSplicer::reset`, and it MUST be /// called wherever the client skips or a pair declines, or the next pair would be spaced against a clock /// whose media nobody received. @@ -160,9 +184,12 @@ impl PairWeaver { /// the caller must `reset()` and skip it. There is deliberately no "serve verbatim" fallback here — the /// muxed path has one because a single transport stream concatenates, and two do not. pub(crate) fn weave(&mut self, video: &[u8], audio: &[u8]) -> Option> { + // The format string is the cause key; the interpolated values are the measurement. See the twin macro + // in `PairSplicer::normalize_pair`. macro_rules! decline { - ($($why:tt)*) => {{ - self.last_decline = format!($($why)*); + ($fmt:literal $(, $arg:expr)* $(,)?) => {{ + self.last_slug = $fmt; + self.last_decline = format!($fmt $(, $arg)*); return None; }}; } @@ -170,7 +197,13 @@ impl PairWeaver { // 1. One clock, canonical pids, skew-guarded — all of it `PairSplicer`'s, none of it ours. let (vout, aout) = match self.pair.normalize_pair(video, audio) { Some(p) => p, - None => decline!("{}", self.pair.last_decline()), + // Forwarded by hand rather than through `decline!`: the reason is already formatted, and its slug + // belongs to the splicer. Passing it through the macro would key the latch on "{}". + None => { + self.last_slug = self.pair.last_decline_slug(); + self.last_decline = self.pair.last_decline().to_string(); + return None; + } }; // The carry belongs to the PREVIOUS pair's buffers, so take it now: from here on `self` must be free @@ -225,7 +258,20 @@ impl PairWeaver { // 5. Merge. `forward_gap` from an anchor placed safely before the video lane's first stamp, so the // comparison is wrap-safe; the sort is STABLE, which is what keeps each pid's packets in order when // a fragment unit inherits its predecessor's key. - let anchor = vfirst.wrapping_sub(ANCHOR_BACKSTOP) & CLOCK_MASK; + // The origin is placed before the EARLIEST key actually present, not before a fixed guess: the + // locked A/V skew has no bounded magnitude (see `ANCHOR_BACKSTOP`), so a lead past the backstop + // used to put every audio key behind the origin, where `forward_gap` reads it as nearly a full + // wrap and sorts it last. Keyless units are skipped deliberately — `key_or_zero` gives them 0, + // which is not a timestamp, and they keep sorting last exactly as before. + let mut back = ANCHOR_BACKSTOP; + for u in carry_units.iter().chain(vunits.iter()).chain(aunits[..split].iter()) { + let Some(k) = u.key else { continue }; + let lead = forward_gap(k, vfirst); // how far this key sits BEFORE the video lane's first + if lead < CLOCK_WRAP / 2 && lead > back { + back = lead; + } + } + let anchor = vfirst.wrapping_sub(back) & CLOCK_MASK; let mut all: Vec = Vec::with_capacity(carry_units.len() + vunits.len() + aunits.len()); all.extend_from_slice(&carry_units); all.extend_from_slice(&vunits); @@ -749,6 +795,43 @@ mod tests { ); } + /// THE REGRESSION the fixed anchor caused: an audio lane leading video by more than `ANCHOR_BACKSTOP`. + /// + /// Reachable because `SKEW_TOLERANCE` bounds DRIFT from the locked skew, not the locked skew itself, and + /// PDT pairing accepts a partner up to half a segment away. With a fixed origin every audio key landed + /// before it, read as almost a full wrap, and sorted after ALL video — the socket stepped backwards at + /// every seam while HLS off the same ring played fine. + #[test] + fn an_audio_lane_leading_past_the_backstop_still_leaves_in_timestamp_order() { + const LEAD: u64 = 225_000; // 2.5 s at 90 kHz — beyond the 2 s backstop + let vbase: u64 = 2_000_000; + // The audio lane is LONGER as well as earlier, so its run actually spans the video's — a lane that + // both starts and ends before the video would be segregated by any correct merge, and would test + // nothing about the anchor. + let v = video_lane(vbase, 6); + let a = audio_lane(vbase.wrapping_sub(LEAD) & CLOCK_MASK, 70); + let mut w = PairWeaver::new(); + let out = w.weave(&v, &a).expect("a leading audio lane is still a publishable pair"); + + let keys = keys_in_order(&out); + assert!(keys.len() >= 10, "both lanes are represented ({} units)", keys.len()); + for w2 in keys.windows(2) { + assert!( + forward_gap(w2[0].1, w2[1].1) < CLOCK_WRAP / 2, + "units never step backwards: {:?} then {:?}", + w2[0], + w2[1] + ); + } + // The failure mode was total segregation — every audio unit after every video unit — so assert the + // lanes actually interleave rather than merely being monotonic. + let pids: Vec = keys.iter().map(|(p, _)| *p).collect(); + assert!( + pids.windows(2).filter(|w| w[0] != w[1]).count() >= 8, + "the lanes interleave rather than concatenate: {pids:?}" + ); + } + #[test] fn the_authored_av_skew_survives_the_weave() { // The one property a muxer must not break. `PairSplicer` applies ONE offset derived from the video diff --git a/server/src/epg/fastSelfEpg.ts b/server/src/epg/fastSelfEpg.ts index 5a1927e..1ff23ee 100644 --- a/server/src/epg/fastSelfEpg.ts +++ b/server/src/epg/fastSelfEpg.ts @@ -1,5 +1,5 @@ // fastSelfEpg — the source-agnostic WRITE / LINK / UPSERT half of a FAST source's self-EPG, extracted from -// epg/tubi.ts + epg/dlhd.ts + epg/dami.ts (which were three copies of the same mechanics). It carries NO +// epg/tubi.ts + epg/dlhd.ts (which had grown into copies of the same mechanics). It carries NO // per-source fetch/parse logic: a source's epg/.ts produces already-mapped EpgChannel/Program docs (Samsung // reuses the shared XMLTV mappers; an inline-program source builds them with a tiny mapper) and calls these. // @@ -16,7 +16,7 @@ import { PlaylistChannel } from '../models/PlaylistChannel.js'; /** * Per-source REPLACE of the guide stores (epgchannels + programs, both scoped by `source`) from already-mapped * docs. Returns the new counts PLUS the distinct bare channelIds present (the playlist hook self-links those via - * linkFastSelfEpg). The same pattern Gracenote / EPG-PW / tubi / dlhd / dami use. + * linkFastSelfEpg). The same pattern Gracenote / EPG-PW / tubi / dlhd use. */ export async function writeFastEpg( sourceId: string, @@ -35,7 +35,7 @@ export async function writeFastEpg( /** * Self-link a source's still-UNTOUCHED PlaylistChannels onto its own guide (FILL-ONLY-IF-UNTOUCHED — generalized - * from dlhd/dami's afterSync). The filter requires epg == null AND epgState == null, so a user link/unlink (or a + * from dlhd's afterSync). The filter requires epg == null AND epgState == null, so a user link/unlink (or a * crosswalk that already claimed the row) is never overwritten. Returns the number of channels linked. */ export async function linkFastSelfEpg(sourceId: string, channelIds: string[]): Promise { diff --git a/server/src/epg/local.ts b/server/src/epg/local.ts index 92510fc..4af88a2 100644 --- a/server/src/epg/local.ts +++ b/server/src/epg/local.ts @@ -1,4 +1,4 @@ -// Local Now playlist-bound self-EPG. Unlike a registry source (dlhd/dami) whose afterSync hook writes its +// Local Now playlist-bound self-EPG. Unlike a registry source (dlhd) whose afterSync hook writes its // guide, a Local Now playlist is a CUSTOM playlist (one per market) so its guide is written INSIDE the // playlist's own sync (../sources/adapters/local/import.ts → syncLocalPlaylist). This module owns the guide // shapes + the per-source replace + the EpgSource upsert, shared by that sync AND the standalone EPG sync @@ -7,7 +7,7 @@ // ⚠️ Per-playlist namespacing: every Local playlist has its OWN EpgSource (id === the playlist id), and its // epgchannels/programs are scoped by `source === `. The composite guide key is // ":" (EpgChannel._id == Program.channelId), joined to a PlaylistChannel by -// `${epg}:${tvg_id}` — exactly the dlhd/dami convention, just keyed by the playlist instead of a source id. +// `${epg}:${tvg_id}` — exactly the dlhd convention, just keyed by the playlist instead of a source id. // Programs come INLINE with the catalog (~5 per channel), so the guide refreshes on every market fetch. import { EpgSource } from '../models/EpgSource.js'; @@ -115,7 +115,7 @@ export async function writeLocalEpg( } } - // Per-source (per-playlist) replace — the same pattern dlhd/dami/tubi use. + // Per-source (per-playlist) replace — the same pattern dlhd/tubi use. await EpgChannel.deleteMany({ source: playlistId }); if (channelDocs.length) await EpgChannel.insertMany(channelDocs, { ordered: false }); await Program.deleteMany({ source: playlistId }); diff --git a/server/src/models/EpgSource.ts b/server/src/models/EpgSource.ts index f7c4cab..5c95dbc 100644 --- a/server/src/models/EpgSource.ts +++ b/server/src/models/EpgSource.ts @@ -19,7 +19,7 @@ export interface EpgSourceDoc { auto: boolean; interval: string; builtin?: boolean; - // true ⇒ this EPG source was created by a playlist's afterSync automation (the tubi/dlhd/dami self-EPG rows), + // true ⇒ this EPG source was created by a playlist's afterSync automation (the tubi/dlhd self-EPG rows), // not added by a user. Bound rows hide manual sync + schedule controls in the UI (the playlist owns the // refresh cadence). Set in the $set block of upsert{Tubi,Dlhd,Dami}EpgSource so it is re-asserted on every sync. playlistBinding: boolean; @@ -37,7 +37,7 @@ export interface EpgSourceDoc { xmlGeneratedCount: number; xmlFailCount: number; // Gracenote provenance (null for non-Gracenote / legacy rows). - source: string | null; // lowercase kind discriminator: 'gracenote' | 'epg-pw' | 'jesmann' | 'tubi' | 'dlhd' | 'dami' | 'local' | 'xml file' | 'remote url' + source: string | null; // lowercase kind discriminator: 'gracenote' | 'epg-pw' | 'jesmann' | 'tubi' | 'dlhd' | 'local' | 'xml file' | 'remote url' location: string | null; lineup_Type: string | null; // provider.type: 'OTA' | 'CABLE' | 'SATELLITE' postalCode: string | null; @@ -65,7 +65,7 @@ const EpgSourceSchema = new Schema( auto: { type: Boolean, required: true }, interval: { type: String, required: true }, builtin: { type: Boolean }, - // Set true by the tubi/dlhd/dami self-EPG upserts (playlist afterSync); gates the UI's sync/schedule controls. + // Set true by the tubi/dlhd self-EPG upserts (playlist afterSync); gates the UI's sync/schedule controls. playlistBinding: { type: Boolean, required: true, default: false }, // List position for the drag-to-reorder UI; GET sorts by { order: 1, name: 1 }. order: { type: Number, required: true, default: 0 }, diff --git a/server/src/models/PlaylistChannel.ts b/server/src/models/PlaylistChannel.ts index b3627e0..e301929 100644 --- a/server/src/models/PlaylistChannel.ts +++ b/server/src/models/PlaylistChannel.ts @@ -46,7 +46,7 @@ export interface PlaylistChannelDoc { // distinct from a stored null (its original tvg_id was unlinked). Never in a $set/$setOnInsert bucket, // so it rides re-sync untouched. See services/failover.ts failoverDisbandUpdate for the restore. origTvgId?: string | null; - // Operator's preferred upstream "player" for sources that expose several (adapter.playerSelectable — dlhd/dami's + // Operator's preferred upstream "player" for sources that expose several (adapter.playerSelectable — dlhd's // DaddyLive Player 1..N). 1-based; null/absent = inherit the source-wide default (Settings.dlhdPlayer). Read at // resolve time by the seam (buildGrant) and honored+failed-over by the adapter's resolveStream. OPTIONAL — older // docs lack it (treat undefined as null). $setOnInsert-only, like the failover fields, so it survives re-sync. diff --git a/server/src/models/Settings.ts b/server/src/models/Settings.ts index eea574c..3e2a6a3 100644 --- a/server/src/models/Settings.ts +++ b/server/src/models/Settings.ts @@ -33,7 +33,7 @@ import { Schema, model } from 'mongoose'; // log). Global operator toggle, edited on the Settings screen; consumed only by the SPA. // Legacy rows hold only 'inapp'/'debug' and stay valid — settings/translate.ts coerces // anything unrecognized back to 'inapp', so no migration is needed. -// - dlhdPlayer — source-wide DEFAULT upstream player for DaddyLive (dlhd/dami) channels that expose several +// - dlhdPlayer — source-wide DEFAULT upstream player for DaddyLive (dlhd) channels that expose several // (0 = Auto/first; 1..N = a specific player). A per-channel override (PlaylistChannel.playerPref) // wins over it. Cached into the dlhd resolver at boot + on every save (settings/applyDlhdPlayer.ts) // so the hot resolve path reads it with no DB hit. @@ -53,7 +53,7 @@ export interface SettingsDoc { offset: string; // DST-aware UTC offset ('±HHMM') derived from `timezone` on save; stamped onto programs + emitted in the guide darkMode: boolean; videoPlayer: 'inapp' | 'ultimate' | 'debug'; // which player the slide-out renders ('inapp' default; 'ultimate' = popup player window; 'debug' = diagnostic HUD) - dlhdPlayer: number; // source-wide default DaddyLive player (0 = Auto/first; 1..N) for dlhd/dami channels without a per-channel override + dlhdPlayer: number; // source-wide default DaddyLive player (0 = Auto/first; 1..N) for dlhd channels without a per-channel override nameservers: string | null; // comma-separated outbound-fetch resolver IP(s); null/blank = OS resolver (DEFAULT_NAMESERVERS 8.8.8.8,8.8.4.4 seeds first boot) logLevel: number; // GLOBAL 1|2|3 log verbosity — app + Rust proxy engine (default 2; formerly dnsLogLevel) maxmindAccountId: string | null; // MaxMind GeoLite2 web-service account id (null = geo disabled) diff --git a/server/src/proxy/resolveSeam.ts b/server/src/proxy/resolveSeam.ts index d7748b3..bbc394a 100644 --- a/server/src/proxy/resolveSeam.ts +++ b/server/src/proxy/resolveSeam.ts @@ -13,15 +13,15 @@ import { logMilestone, logTrace } from '../logs/tier.js'; // churn-prone provider logic in TypeScript; Rust just fetches + rewrites + pipes. // // Faithfulness notes (verified against the adapters): -// · upstreamHeaders is per-stream CONSTANT — snapshot once here (for dlhd/dami this captures the rotating +// · upstreamHeaders is per-stream CONSTANT — snapshot once here (for dlhd this captures the rotating // playerReferer per stream, which is MORE correct than the shared module global the old proxy replayed). // The (Default)/(Custom) proxy-config `headerOverrides` are merged ON TOP here (operator wins), so Rust // replays the final header set unchanged — the one proxy-config knob applied Node-side (see CFG/PXY-2). // · The SSRF allowlist is OBSERVATIONAL: Rust seeds it from the resolved target host and grows it from the -// hosts it rewrites out of each manifest (all of dulo/dlhd/dami enable dynamic-allow), so the grant needs +// hosts it rewrites out of each manifest (all of dulo/dlhd enable dynamic-allow), so the grant needs // NO host list — only `allowPrivate` (false for these public-CDN sources; a future LAN source flips it). // · relabelSegment is derived by PROBING the adapter's relabel rule with a sentinel content-type, so the -// core stays generic (no per-source branch): dulo passes the sentinel through → null; dlhd/dami force +// core stays generic (no per-source branch): dulo passes the sentinel through → null; dlhd forces // 'video/mp2t' on segments → 'video/mp2t'. // · proxyConfig is the resolved (Custom app_ → Default app → env) knob set (proxyconfig/resolve.ts). Rust // applies connectTimeoutMs + maxRedirects (P2 → its upstream client), readTimeoutMs + bufferSizeKb (P3.1/RSL @@ -36,10 +36,20 @@ export interface ResolveGrant { upstreamHeaders: Record; /** Force this content-type on non-manifest (segment) responses; null = pass upstream through. */ relabelSegment: string | null; - /** Permit private/loopback upstream IPs (LAN sources). false for the public-CDN sources (dulo/dlhd/dami). */ + /** Permit private/loopback upstream IPs (LAN sources). false for the public-CDN sources (dulo/dlhd). */ allowPrivate: boolean; /** Whether the request URL needed server-side resolution (vs a direct passthrough entry). */ isEntry: boolean; + /** + * Does the SERVING adapter have alternate upstreams to walk to (`adapter.playerSelectable`)? + * + * S3/UND: the local origin's undecodable-upstream detector is scoped to this. Retiring an upstream only + * helps where another one can take over — on a single-upstream source the retirement just re-resolves the + * same dead provider on a 2 s loop. It rides the grant because the capability belongs to the adapter, and + * the data plane must not know adapter names: it used to test `source === 'dlhd'` in Rust, which silently + * excluded the next playerSelectable adapter from detection until someone edited and redeployed the crate. + */ + playerSelectable: boolean; /** The resolved (Default/Custom) data-plane config for this stream — Rust applies the LIVE knobs, carries the rest. */ proxyConfig: RuntimeProxyConfig; /** @@ -84,7 +94,8 @@ function mergeUpstreamHeaders( return out; } -// Read a channel's per-channel player OVERRIDE (for playerSelectable sources — dlhd/dami). Returns the 1-based +// Read a channel's per-channel player OVERRIDE (for playerSelectable sources — dlhd today, and any +// adapter that sets the flag). Returns the 1-based // preference, or 0 when unset (the adapter's resolveStream then falls back to the cached source-wide default). // Mirrors buildFailoverGrant's reverse lookup: exact by (streamEntryUrl, pl) when the composed M3U stamped ?pl, // else a DETERMINISTIC no-pl fallback (canonical source-playlist doc, then the lexically-first clone copy). One @@ -197,7 +208,7 @@ export async function buildGrant( // m3u/serialize.ts). The in-app appPlayer path carries no ?pl → the Default applies (CFG/PXY-2). const proxyConfig = await resolveProxyConfig(pl); - // Snapshot the per-stream upstream headers against the resolved target (dlhd/dami: the CDN-host branch → + // Snapshot the per-stream upstream headers against the resolved target (dlhd: the CDN-host branch → // { Referer: playerReferer(), UA }; dulo: a constant map — it ignores the url arg), then merge the operator // headerOverrides ON TOP (operator wins, CASE-INSENSITIVELY — HTTP header names are case-insensitive and Rust // normalizes them, so a `referer` override must beat the adapter's `Referer`, not race it). This is the one @@ -265,6 +276,7 @@ export async function buildGrant( relabelSegment, allowPrivate: false, isEntry, + playerSelectable: !!adapter.playerSelectable, proxyConfig, adSignature: adapter.proxy.adSignature ?? null, policySource: source, @@ -414,6 +426,9 @@ async function buildFailoverGrant( relabelSegment, allowPrivate: false, isEntry, + // The CHILD's capability, for the same reason as its signature below: a failover onto a single-upstream + // provider must not keep the parent's alternates-exist promise, and vice versa. + playerSelectable: !!candAdapter.playerSelectable, proxyConfig, // The CHILD's own signature, like its headers/relabel — a cross-provider backup must not inherit the // parent provider's ad shape (same reason policySource names candSource). diff --git a/server/src/proxy/telemetryIngest.ts b/server/src/proxy/telemetryIngest.ts index d928071..f029b0c 100644 --- a/server/src/proxy/telemetryIngest.ts +++ b/server/src/proxy/telemetryIngest.ts @@ -167,6 +167,26 @@ function num(v: unknown): number { return typeof v === 'number' && Number.isFinite(v) && v >= 0 ? v : 0; } +// ── ABSENT ≠ ZERO, for the fields a sidecar may not have ──────────────────────────────────────────────── +// The reasoning above holds for a counter every sidecar has always sent. It does NOT hold for a field this +// release added: an older Rust sidecar (mid-upgrade, or the aio image's split rebuild) simply omits those +// keys, and `num()` would turn the omission into 0 — a MEASUREMENT we never took. The panel is built for the +// difference: its version-skew branches read `undefined` as "not reported by this sidecar" and anything else +// as authoritative, so coercing here silently makes those branches unreachable and replaces "unknown" with a +// confident, wrong number. These three keep the omission intact. +function optNum(v: unknown): number | undefined { + return typeof v === 'number' && Number.isFinite(v) && v >= 0 ? v : undefined; +} +function optBool(v: unknown): boolean | undefined { + return typeof v === 'boolean' ? v : undefined; +} +/** A tri-state string: the reason, `null` for "reported, and there is none", `undefined` for "not reported". + * The middle state is load-bearing — for `ineligible`, `null` is an authoritative "this upstream is fine". */ +function optTri(v: unknown): string | null | undefined { + if (v === undefined) return undefined; + return typeof v === 'string' && v ? v.slice(0, 48) : null; +} + function applyEvent(e: TelemetryEvent): void { if (!e || typeof e !== 'object') return; const ip = str(e.ip); @@ -225,26 +245,26 @@ function applyEvent(e: TelemetryEvent): void { subscribers: num(e.subscribers), ringSegments: num(e.ringSegments), ringBytes: num(e.ringBytes), - channelRingCapBytes: num(e.channelRingCapBytes), - ringSeconds: num(e.ringSeconds), + channelRingCapBytes: optNum(e.channelRingCapBytes), + ringSeconds: optNum(e.ringSeconds), // Booleans go through `=== true`, NEVER num() — num() tests `typeof v === 'number'` and would map // `true` to 0, i.e. permanently false. Same shape as the cue branch's `profileChanged` below. floorBeatsCap: e.floorBeatsCap === true, headSeq: num(e.headSeq), generation: num(e.generation), - discSeq: num(e.discSeq), - discInWindow: num(e.discInWindow), + discSeq: optNum(e.discSeq), + discInWindow: optNum(e.discInWindow), ingestedSegments: num(e.ingestedSegments), ingestedBytes: num(e.ingestedBytes), evictedSegments: num(e.evictedSegments), targetDuration: num(e.targetDuration), - demuxed: e.demuxed === true, + demuxed: optBool(e.demuxed), upstreamShape: optStr(e.upstreamShape) ?? null, encryption: optStr(e.encryption) ?? null, // Tri-state, so it takes the `suspect` shape rather than str(): str() coerces null to '' and the // difference between "eligible" and "declined, reason unknown" would be lost. - ineligible: typeof e.ineligible === 'string' && e.ineligible ? e.ineligible.slice(0, 48) : null, - suspect: typeof e.suspect === 'string' && e.suspect ? e.suspect.slice(0, 48) : null, + ineligible: optTri(e.ineligible), + suspect: optTri(e.suspect), suspectRetires: num(e.suspectRetires), at: Date.now(), }); diff --git a/server/src/proxyconfig/seed.ts b/server/src/proxyconfig/seed.ts index cdcd1de..1c3701d 100644 --- a/server/src/proxyconfig/seed.ts +++ b/server/src/proxyconfig/seed.ts @@ -27,8 +27,24 @@ export async function seedProxyConfig(): Promise { // mode already hides an unknown path on read, so this is not about correctness — it is about not carrying a // dead key through config exports and backup restores forever. Runs on EVERY boot (cheap: a no-op once the // field is gone) rather than gated behind a version marker the collection does not have. - const pruned = await ProxyConfig.updateMany({ adPolicy: { $exists: true } }, { $unset: { adPolicy: '' } }); - if (pruned.modifiedCount > 0) { + // + // `strict: false` is LOAD-BEARING, not defensive. `adPolicy` is — correctly — absent from the schema, and + // Mongoose's strict UPDATE casting deletes unknown paths from the update document: the `$unset` is emptied, + // the emptied `$unset` operator is then dropped, and `updateMany` returns `{ acknowledged: false }` without + // ever reaching the database. Worse, that return shape carries no `modifiedCount`, so `undefined > 0` is + // false and the success log below can never fire either — a migration that silently does nothing and + // silently says nothing. (mongoose 8: castUpdate.js `skip = isStrict && !schematype …` → `delete obj[key]`, + // then `isEmptyObject(val)` → `delete ret[op]`, then query.js returns before the driver call.) + const pruned = await ProxyConfig.updateMany( + { adPolicy: { $exists: true } }, + { $unset: { adPolicy: '' } }, + { strict: false }, + ); + if (!pruned.acknowledged) { + // The failure this migration already had once. Named rather than swallowed, so the next person who + // re-tightens the options learns it from a log line instead of from a stale key in a config export. + logger.warn('seed', 'proxy config: adPolicy migration was cast away before reaching the database — not applied'); + } else if (pruned.modifiedCount > 0) { logger.ok('seed', `proxy config: dropped the removed adPolicy field from ${pruned.modifiedCount} doc(s)`); } } diff --git a/server/src/routes/playlists.ts b/server/src/routes/playlists.ts index bf5734a..52677dd 100644 --- a/server/src/routes/playlists.ts +++ b/server/src/routes/playlists.ts @@ -462,7 +462,7 @@ async function cascadeDeleteBuiltinPlaylist(p: { ); } - // 2. The playlist-bound self-EPG source (tubi/dlhd/dami self-EPG; id === src, playlistBinding:true), if any. + // 2. The playlist-bound self-EPG source (tubi/dlhd self-EPG; id === src, playlistBinding:true), if any. // cascadeDeleteEpgSource unlinks every playlistchannel linked to it — INCLUDING this built-in's own // channels — and drops its programs/epgchannels/cronjob. dulo has none (crosswalk-only) → no-op. const bound = (await EpgSource.findOne( @@ -704,7 +704,7 @@ playlistsRouter.put('/:id/channels/:channelId', requireAdmin, async (req, res, n $set[key] = body[key]; } } - // playerPref: preferred upstream player for playerSelectable sources (dlhd/dami). A 1-based integer, or + // playerPref: preferred upstream player for playerSelectable sources (dlhd). A 1-based integer, or // null to clear it (inherit the source-wide default). Numeric, so it can't ride the string loop above. An // out-of-range pick is accepted but clamps to the lead player at resolve time (resolveStream.ts). if (body.playerPref !== undefined) { diff --git a/server/src/services/failover.ts b/server/src/services/failover.ts index 526a3ce..ce2f804 100644 --- a/server/src/services/failover.ts +++ b/server/src/services/failover.ts @@ -15,7 +15,7 @@ export interface FailoverEpgSnapshot { } // A grouped child's epgState is NEVER null: the fill-only sync writers (epg/fastSelfEpg.ts, -// sources/epgCrosswalk.ts, the dlhd/dami/tubi/local afterSync self-links) all match +// sources/epgCrosswalk.ts, the dlhd/tubi/local afterSync self-links) all match // { epg: null, epgState: null } as "untouched" — a child inheriting an unlinked parent's nulls would be // independently re-linked to its OWN guide id on the next sync, silently diverging from the parent. export function inheritedEpgState(snap: FailoverEpgSnapshot): 'matched' | 'unmatched' { diff --git a/server/src/sources/adapters/_fast/dynamicAllow.ts b/server/src/sources/adapters/_fast/dynamicAllow.ts index 8a3d7aa..303a2ed 100644 --- a/server/src/sources/adapters/_fast/dynamicAllow.ts +++ b/server/src/sources/adapters/_fast/dynamicAllow.ts @@ -2,9 +2,9 @@ // dlhd's allowlist (adapters/dlhd/config.ts UPSTREAM_ALLOW / isAllowedHost / allowHost): a Set seeded with a // source's known CDN domain suffixes that the source GROWS at runtime — `allow()` for a host learned by // resolveStream (the resolved master CDN), `onPlaylistChildHost()` for a host seen inside a resolved playlist. -// Each FAST source gets its OWN instance (no shared module state, unlike dlhd↔dami which intentionally share an -// upstream). Private/loopback/link-local targets are ALWAYS blocked via the shared core/ssrf.ts guard — the -// dynamic set only ever widens to public CDN hosts. +// Each FAST source gets its OWN instance: the allow-set widens only from that source's own resolves, so one +// source's CDN can never authorise another's. Private/loopback/link-local targets are ALWAYS blocked via the +// shared core/ssrf.ts guard — the dynamic set only ever widens to public CDN hosts. import { isPrivateHost } from '../../core/ssrf.js'; diff --git a/server/src/sources/adapters/distro.ts b/server/src/sources/adapters/distro.ts index 12d9621..cd8c8bc 100644 --- a/server/src/sources/adapters/distro.ts +++ b/server/src/sources/adapters/distro.ts @@ -99,7 +99,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the distro self-EPG from the separate epg/query.php schedule (keyed off the same `raw` catalog rows // buildSource consumed), upsert the 'distro' EpgSource, and self-link the still-untouched channels onto it. // Live-only (the caller guards on `live` so a snapshot fallback never overwrites a good guide). FILL-ONLY-IF- -// UNTOUCHED — same posture as dlhd/dami/samsung/vizio/lg/vidaa/whale/xumo/freelivesports. +// UNTOUCHED — same posture as dlhd/samsung/vizio/lg/vidaa/whale/xumo/freelivesports. async function applyDistroSelfEpg(sourceId: string, raw: DistroRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/dlhd/config.ts b/server/src/sources/adapters/dlhd/config.ts index 5054613..64be27e 100644 --- a/server/src/sources/adapters/dlhd/config.ts +++ b/server/src/sources/adapters/dlhd/config.ts @@ -71,7 +71,7 @@ export const UA = // or renames). Index i (0-based) here == "Player i+1" in the UI. export const PLAYER_PREFIXES = ['stream', 'cast', 'watch', 'plus', 'casting', 'player'] as const; -// The source-wide DEFAULT player (0 = Auto/first; 1..N = a specific player) for every dlhd/dami channel that +// The source-wide DEFAULT player (0 = Auto/first; 1..N = a specific player) for every dlhd channel that // carries no per-channel override. Cached module-level (like _base) so the hot resolve path reads it with NO // DB hit; refreshed from the Settings singleton at boot + on every settings save (settings/applyDlhdPlayer.ts). let _playerDefault = 0; diff --git a/server/src/sources/adapters/freelivesports.ts b/server/src/sources/adapters/freelivesports.ts index 003af9e..65db4bd 100644 --- a/server/src/sources/adapters/freelivesports.ts +++ b/server/src/sources/adapters/freelivesports.ts @@ -95,7 +95,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the freelivesports self-EPG from the inline epg.entries (the same `raw` catalog rows buildSource // consumed), upsert the 'freelivesports' EpgSource, and self-link the still-untouched channels onto it. Live-only // (the caller guards on `live` so a snapshot fallback never overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — -// same posture as dlhd/dami/samsung/vizio/lg/vidaa/whale. +// same posture as dlhd/samsung/vizio/lg/vidaa/whale. async function applyFlsSelfEpg(sourceId: string, raw: FlsRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/lg.ts b/server/src/sources/adapters/lg.ts index 6d3b9e0..695e1e5 100644 --- a/server/src/sources/adapters/lg.ts +++ b/server/src/sources/adapters/lg.ts @@ -92,7 +92,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the lg self-EPG from the inline programs (the same `raw` catalog rows buildSource consumed), upsert the // 'lg' EpgSource, and self-link the still-untouched channels onto it. Live-only (the caller guards on `live` so a -// snapshot fallback never overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/dami/samsung/vizio. +// snapshot fallback never overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/samsung/vizio. async function applyLgSelfEpg(sourceId: string, raw: LgRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/samsung.ts b/server/src/sources/adapters/samsung.ts index c8e6c5c..32a41a9 100644 --- a/server/src/sources/adapters/samsung.ts +++ b/server/src/sources/adapters/samsung.ts @@ -135,7 +135,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the samsung self-EPG from the per-region XMLTV, upsert the 'samsung' EpgSource, and self-link the still- // untouched channels onto it. Live-only (the caller guards on `live` so a snapshot fallback never overwrites a -// good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/dami. +// good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd. async function applySamsungSelfEpg(sourceId: string): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/tubi.ts b/server/src/sources/adapters/tubi.ts index 883e180..e35542f 100644 --- a/server/src/sources/adapters/tubi.ts +++ b/server/src/sources/adapters/tubi.ts @@ -9,7 +9,7 @@ // shared rewriter (core/playlist.ts) routes the #EXT-X-KEY URI back through the proxy so the key decrypts. // // tubi is UNIQUE in that it carries its OWN EPG inline: afterSync (the source-agnostic post-sync hook) -// attaches EPG the dlhd/dami TWO-TIER way — (1) a committed gracenote crosswalk (TUBI_EPG_ADDON_FILE, ported +// attaches EPG the dlhd TWO-TIER way — (1) a committed gracenote crosswalk (TUBI_EPG_ADDON_FILE, ported // from FastChannels' exact per-content_id tmsid map) links the curated US linear channels to a real Gracenote // guide so they share a standard grid + cross-source-dedupe, then (2) tubi's own inline guide // (epgchannels/programs from this same listing) is written, the 'tubi' EpgSource upserted, and the REMAINING @@ -159,7 +159,7 @@ const tubiAdapter: SourceAdapter = { // ── post-sync hook: tubi carries its own EPG (gracenote crosswalk THEN self-EPG) ───────────────────── // Runs after syncLive upserts/prunes the channel stores, off the SAME listing (`raw`) this sync fetched. - // The dlhd/dami TWO-TIER pattern: a committed gracenote crosswalk claims the curated US linear channels + // The dlhd TWO-TIER pattern: a committed gracenote crosswalk claims the curated US linear channels // first (so a Tubi "CBS News" shares a STANDARD guide + cross-source-dedupes with the same channel from // other sources), then tubi's own inline-program self-EPG fills the remainder. Both are // FILL-ONLY-IF-UNTOUCHED (epg == null AND epgState == null), so a user link/unlink/remap always survives. diff --git a/server/src/sources/adapters/vidaa.ts b/server/src/sources/adapters/vidaa.ts index 0f2f1cf..3f6026d 100644 --- a/server/src/sources/adapters/vidaa.ts +++ b/server/src/sources/adapters/vidaa.ts @@ -87,7 +87,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the vidaa self-EPG from the separate /epg/grid schedule (keyed off the same `raw` catalog rows buildSource // consumed), upsert the 'vidaa' EpgSource, and self-link the still-untouched channels onto it. Live-only (the // caller guards on `live` so a snapshot fallback never overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same -// posture as dlhd/dami/samsung/vizio/lg. +// posture as dlhd/samsung/vizio/lg. async function applyVidaaSelfEpg(sourceId: string, raw: VidaaRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/vizio.ts b/server/src/sources/adapters/vizio.ts index 29d5b4b..ced289b 100644 --- a/server/src/sources/adapters/vizio.ts +++ b/server/src/sources/adapters/vizio.ts @@ -88,7 +88,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the vizio self-EPG from the airings schedule grid (the same `raw` catalog rows buildSource consumed), // upsert the 'vizio' EpgSource, and self-link the still-untouched channels onto it. Live-only (the caller guards // on `live` so a snapshot fallback never overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same posture as -// dlhd/dami/samsung. +// dlhd/samsung. async function applyVizioSelfEpg(sourceId: string, raw: VizioRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/whale.ts b/server/src/sources/adapters/whale.ts index f95e3e7..613938d 100644 --- a/server/src/sources/adapters/whale.ts +++ b/server/src/sources/adapters/whale.ts @@ -95,7 +95,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the whale self-EPG from the SEPARATE /epg fetch (keyed off the same `raw` catalog rows buildSource // consumed — currentProgram descriptions ride along), upsert the 'whale' EpgSource, and self-link the // still-untouched channels onto it. Live-only (the caller guards on `live` so a snapshot fallback never -// overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/dami/samsung/vizio/lg/vidaa. +// overwrites a good guide). FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/samsung/vizio/lg/vidaa. async function applyWhaleSelfEpg(sourceId: string, raw: WhaleRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/adapters/xumo.ts b/server/src/sources/adapters/xumo.ts index 7807a61..e29852b 100644 --- a/server/src/sources/adapters/xumo.ts +++ b/server/src/sources/adapters/xumo.ts @@ -101,7 +101,7 @@ async function resolveStream(entryUrl: string): Promise<{ masterUrl: string }> { // Build the xumo self-EPG from the SEPARATE paginated market guide (the asset metadata rides along in each page // response, so no per-program asset fetch), upsert the 'xumo' EpgSource, and self-link the still-untouched // channels onto it. Live-only (the caller guards on `live` so a snapshot fallback never overwrites a good guide). -// FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/dami/samsung/vizio/lg/vidaa/whale. +// FILL-ONLY-IF-UNTOUCHED — same posture as dlhd/samsung/vizio/lg/vidaa/whale. async function applyXumoSelfEpg(sourceId: string, raw: XumoRow[]): Promise { const { offset, defaulted } = await resolveProgramOffset(); if (defaulted) logger.warn('seed', `[${sourceId}] settings offset unset — guide times stored as UTC (+0000)`); diff --git a/server/src/sources/core/streamTelemetry.ts b/server/src/sources/core/streamTelemetry.ts index 1171bee..ed8b5f6 100644 --- a/server/src/sources/core/streamTelemetry.ts +++ b/server/src/sources/core/streamTelemetry.ts @@ -270,6 +270,35 @@ export interface MediaInfo { // channelKey → merged decode metadata. Bounded by the CHANNEL AGGREGATE's lifetime: the tick that drops a // cold aggregate drops this with it (see the `channels.delete` site). Deliberately NOT in statsHub's // activeKeys prune block with the display-only maps — the reasons are spelled out at that delete. +/** Every DISPLAY-ONLY per-channel map, registered on creation so one sweep prunes them all. + * + * The pattern this replaces was a hand-written `prune(activeKeys)` per map plus a matching call in + * statsHub — six of each by the end. Nothing enforced the pairing: a new map whose prune or whose call site + * was forgotten grew without bound, keyed by every channel ever played, and nothing would ever say so. + * Registering at declaration makes the omission unrepresentable. + * + * DISPLAY-ONLY is the entry requirement, not a description. `activeKeys` is built by statsHub and pruned + * only while an admin socket is open, so a map that any DETECTION path reads must not be in here — see + * `mediaByChannel`, which is deliberately kept out and torn down with the channel aggregate instead. */ +const displayMaps: Map[] = []; + +function displayMap(): Map { + const m = new Map(); + displayMaps.push(m as Map); + return m; +} + +/** Drop every display-only map's entries for channels that are no longer active. One sweep, all maps. */ +export function pruneChannelDisplayMaps(activeKeys: Set): void { + for (const m of displayMaps) { + for (const key of m.keys()) if (!activeKeys.has(key)) m.delete(key); + } +} + +// NOT a `displayMap`, deliberately: the client-shortfall heuristic reads this one's `bandwidth`, so pruning +// it from `activeKeys` would make buffering DETECTION depend on whether an admin has the screen open. It is +// torn down with the channel aggregate instead — see the long note at its delete site for the two reasons +// and for why a sweep over this map is not equivalent. const mediaByChannel = new Map(); /** @@ -319,7 +348,7 @@ export interface FailoverServing { candidateName: string; // the serving child's tvg_name (display) } -const failoverByChannel = new Map(); // channelKey → serving candidate +const failoverByChannel = displayMap(); // channelKey → serving candidate /** Record (or clear, with null) which failover candidate a channel's grants currently target. */ export function noteFailoverServing(source: string, entryUrl: string, f: FailoverServing | null): void { @@ -334,17 +363,13 @@ export function failoverFor(channelKey: string): FailoverServing | null { } /** Drop failover attribution for channels no longer active (statsHub calls this with the live key set). */ -export function pruneFailoverServing(activeKeys: Set): void { - for (const key of failoverByChannel.keys()) if (!activeKeys.has(key)) failoverByChannel.delete(key); -} - // ── Upstream attribution: which HOST is actually carrying this channel ───────────────────────────────── // The channel row names a SOURCE (`dlhd`), which for a multi-provider source says nothing about which of its // interchangeable providers is on air right now. The resolve seam knows — it just discarded it. Same // in-memory idiom and the same (parent source, parent entry) key as the failover map above, so a child // serving under its parent's identity files its host under the parent, where statsHub can join it. -const hostByChannel = new Map(); // channelKey → entry-hop host +const hostByChannel = displayMap(); // channelKey → entry-hop host /** Record the host a channel's grant currently resolves to. Callers MUST pass the caller's own (source, * entryUrl), never a resolved candidate's — see noteFailoverServing for why. */ @@ -360,10 +385,6 @@ export function upstreamHostFor(channelKey: string): string | null { } /** Drop host attribution for channels no longer active (statsHub calls this with the live key set). */ -export function pruneUpstreamHost(activeKeys: Set): void { - for (const key of hostByChannel.keys()) if (!activeKeys.has(key)) hostByChannel.delete(key); -} - // ── The proxy config a stream was actually GRANTED ────────────────────────────────────────────────────── // Answers the one question the panel could not: "I set Raw-TS and I am still being served HLS — why?" The // served half is `delivery`; this is the requested half, captured where it is resolved rather than re-read @@ -380,7 +401,7 @@ export interface RequestedConfig { spliceNormalize: boolean; } -const requestedByChannel = new Map(); // channelKey → config resolved into the grant +const requestedByChannel = displayMap(); // channelKey → config resolved into the grant /** Record the proxy config resolved into a channel's grant. */ export function noteRequestedConfig(source: string, entryUrl: string, c: RequestedConfig): void { @@ -393,10 +414,6 @@ export function requestedConfigFor(channelKey: string): RequestedConfig | null { } /** Drop requested config for channels no longer active (statsHub calls this with the live key set). */ -export function pruneRequestedConfig(activeKeys: Set): void { - for (const key of requestedByChannel.keys()) if (!activeKeys.has(key)) requestedByChannel.delete(key); -} - // ── How this channel's last viewer session ENDED ─────────────────────────────────────────────────────── // A closed session leaves the `clients` map by definition, so nothing about it survives into the live // snapshot — `closeSession` feeds ClosedSession → ViewSession, which is History, not Active Streams. This map @@ -417,7 +434,7 @@ export interface LastClose { socketBound: boolean; } -const lastCloseByChannel = new Map(); // channelKey → how the last session ended +const lastCloseByChannel = displayMap(); // channelKey → how the last session ended /** The last session end for a channel (null = no session has ended on it yet). */ export function lastCloseFor(channelKey: string): LastClose | null { @@ -425,10 +442,6 @@ export function lastCloseFor(channelKey: string): LastClose | null { } /** Drop close attribution for channels no longer active (statsHub calls this with the live key set). */ -export function pruneLastClose(activeKeys: Set): void { - for (const key of lastCloseByChannel.keys()) if (!activeKeys.has(key)) lastCloseByChannel.delete(key); -} - // ── S3/ORIGIN ingest health (the `iop` side) ─────────────────────────────────────────────────────────── // Everything else in this file measures EGRESS — bytes we sent to a viewer. Origin mode adds a second, // independent quantity: what ONE ingest pulled from upstream on behalf of N viewers. Conflating them would @@ -444,29 +457,33 @@ export interface IngestHealth { /** THIS channel's live applied cap in bytes — the denominator `ringBytes` needs to mean anything. NOT the * configured `originRingMb` (a shrink is applied lazily) and NOT the process-wide Σ that * `noteRingFootprint` carries; all three legitimately differ at the same instant. */ - channelRingCapBytes: number; + /** UNDEFINED means "this sidecar did not report it" — a distinct state from any value, and the reason + * these fields are not plain numbers/booleans. An older sidecar (mid-upgrade, or the aio image's split + * rebuild) omits the keys this release added; coercing them to 0/false/null at the ingest seam would + * state a measurement nobody took, and the panel's version-skew branches all test `undefined`. */ + channelRingCapBytes: number | undefined; /** Σ of the held segments' own durations — the real window length, as opposed to * `ringSegments × targetDuration`, which over-reads by each segment's gap below the window max. */ - ringSeconds: number; + ringSeconds: number | undefined; /** The byte cap could not be honored because the MIN_SEGMENTS floor won: this channel's bitrate does not * fit its ring budget. While true, `ringBytes` legitimately exceeds `channelRingCapBytes`. */ floorBeatsCap: boolean; headSeq: number; // our next sequence — monotonic for the life of the ingest generation: number; // bumped on a failover ring reset /** RFC 8216's EXT-X-DISCONTINUITY-SEQUENCE: discontinuity tags that have already LEFT the window. */ - discSeq: number; + discSeq: number | undefined; /** Discontinuity tags still INSIDE the window. Disjoint from `discSeq` — never sum the two. */ - discInWindow: number; + discInWindow: number | undefined; ingestedSegments: number; ingestedBytes: number; // UPSTREAM bytes — distinct from egress; one of these can serve N viewers evictedSegments: number; targetDuration: number; /** True when the origin paired a separate audio rendition into every segment (a DEMUXED upstream). */ - demuxed: boolean; + demuxed: boolean | undefined; /** Non-null ⇒ the origin DECLINED this upstream (fMP4 / SAMPLE-AES / unpairable audio) and the rewrite * path is serving the client instead. The ingest keeps reporting either way, so without this field a * declined channel is indistinguishable from a healthy ring-backed one. */ - ineligible: string | null; + ineligible: string | null | undefined; /** What the ORIGIN's upstream turned out to be — 'ts' | 'hls-master' | 'hls-media'. The authoritative * reading for a ring-backed channel; the passthrough rewriter reports the same idea onto MediaInfo, and a * channel with an INELIGIBLE origin legitimately has both. */ @@ -478,11 +495,11 @@ export interface IngestHealth { /** S3/UND: slug of the last structural fault that retired an upstream (`undecodable-video`, * `not-transport-stream`), or null if none. Non-null means this channel has been hopping providers — * a state every other field here reports as healthy, because fetching IS working. */ - suspect: string | null; + suspect: string | null | undefined; suspectRetires: number; } -const ingestByChannel = new Map(); // channelKey → last ingest snapshot +const ingestByChannel = displayMap(); // channelKey → last ingest snapshot /** Record the data plane's latest `iop` snapshot for a channel. */ export function noteIngest(source: string, entryUrl: string, h: IngestHealth): void { @@ -495,10 +512,6 @@ export function ingestFor(channelKey: string): IngestHealth | null { } /** Drop ingest health for channels no longer active (statsHub calls this with the live key set). */ -export function pruneIngest(activeKeys: Set): void { - for (const key of ingestByChannel.keys()) if (!activeKeys.has(key)) ingestByChannel.delete(key); -} - // ── S3/CUE: ad-break state (per channel) ──────────────────────────────────────────────────────────────── // An EVENT stream, not a health snapshot: the sidecar sends exactly two frames per break (open/close), so // this map holds "what is happening on this channel right now" plus a small rolling tally the operator can @@ -526,7 +539,7 @@ export interface AdBreakState { at: number; // Date.now() of the last cue event } -const adBreakByChannel = new Map(); +const adBreakByChannel = displayMap(); /** Record an ad-break edge from the data plane. `state` is 'open' or 'close'. */ export function noteAdBreak( @@ -554,10 +567,6 @@ export function adBreakFor(channelKey: string): AdBreakState | null { } /** Drop ad-break state for channels no longer active (statsHub calls this with the live key set). */ -export function pruneAdBreaks(activeKeys: Set): void { - for (const key of adBreakByChannel.keys()) if (!activeKeys.has(key)) adBreakByChannel.delete(key); -} - // ── S3/ORIGIN aggregate ring footprint (process-wide) ────────────────────────────────────────────────── // The map above is PER-CHANNEL and is pruned against the active-stream set, so it cannot answer "how much RAM // are the rings holding" — an origin inside its 30s idle grace still owns its bytes with no active row to hang diff --git a/server/src/sources/probeAll.ts b/server/src/sources/probeAll.ts index 0bbc5c1..4b8707a 100644 --- a/server/src/sources/probeAll.ts +++ b/server/src/sources/probeAll.ts @@ -6,7 +6,7 @@ // // HYBRID by design — the churny per-source logic stays in Node, the durable byte work stays in Rust: // 1. RESOLVE in Node (throttled): buildGrant() runs the same adapter logic the resolve seam does (dulo -// Supabase session, dlhd 3-hop scrape + mirror, dami delegation) → a { target, upstreamHeaders } grant. +// Supabase session, dlhd 3-hop scrape + mirror) → a { target, upstreamHeaders } grant. // 2. FETCH + ANALYZE in Rust: the resolved batch is POSTed to the sidecar's /probe endpoint, which fetches // each target concurrently and reports liveness (a 2xx manifest = live) + decode via the SAME parser the // live proxy uses (manifest::extract_media). diff --git a/server/src/sources/types.ts b/server/src/sources/types.ts index 1ffc553..2b71796 100644 --- a/server/src/sources/types.ts +++ b/server/src/sources/types.ts @@ -152,7 +152,7 @@ export interface SourceAdapter { /** * Opt-in: this source exposes multiple interchangeable upstream "players" per channel that the operator can * PREFER (a source-wide default + per-channel override, honored + failed-over by resolveStream via opts.player). - * dlhd/dami set this (DaddyLive's Player 1..N). Absent/false ⇒ the resolve seam never reads a player pref and + * dlhd sets this (DaddyLive's Player 1..N). Absent/false ⇒ the resolve seam never reads a player pref and * the SPA hides the picker. Purely a capability flag; the resolution logic lives in the adapter. */ playerSelectable?: boolean; diff --git a/server/src/stats/statsHub.ts b/server/src/stats/statsHub.ts index 4ef8b84..eba3ae8 100644 --- a/server/src/stats/statsHub.ts +++ b/server/src/stats/statsHub.ts @@ -16,7 +16,7 @@ import { WebSocket } from 'ws'; import { logger } from '../sources/core/logger.js'; -import { snapshotRaw, mediaFor, failoverFor, pruneFailoverServing, ingestFor, pruneIngest, adBreakFor, pruneAdBreaks, upstreamHostFor, pruneUpstreamHost, requestedConfigFor, pruneRequestedConfig, lastCloseFor, pruneLastClose, onSessionClose, onBufferEvent, type ClosedSession, type MediaInfo, type FailoverServing, type IngestHealth, type AdBreakState, type RequestedConfig, type LastClose } from '../sources/core/streamTelemetry.js'; +import { snapshotRaw, mediaFor, failoverFor, ingestFor, adBreakFor, upstreamHostFor, requestedConfigFor, lastCloseFor, pruneChannelDisplayMaps, onSessionClose, onBufferEvent, type ClosedSession, type MediaInfo, type FailoverServing, type IngestHealth, type AdBreakState, type RequestedConfig, type LastClose } from '../sources/core/streamTelemetry.js'; import { humanVideoCodec, humanAudioCodec, humanContainer, humanResolution, parseFps } from '../sources/core/decodeLabels.js'; import { streamKey, phaseFor, type StreamPhase } from '../sources/core/streamState.js'; import { PlaylistChannel } from '../models/PlaylistChannel.js'; @@ -238,12 +238,9 @@ export async function buildDisplaySnapshot(): Promise { } // Drop debounce state for channels no longer active (keeps the map bounded to live channels). for (const key of bufferDebounce.keys()) if (!activeKeys.has(key)) bufferDebounce.delete(key); - pruneFailoverServing(activeKeys); - pruneIngest(activeKeys); - pruneAdBreaks(activeKeys); - pruneUpstreamHost(activeKeys); - pruneRequestedConfig(activeKeys); - pruneLastClose(activeKeys); + // One sweep over every display-only per-channel map. This used to be one call per map, which meant a new + // map needed a new call remembered HERE as well as a prune written there — see `pruneChannelDisplayMaps`. + pruneChannelDisplayMaps(activeKeys); return out; } @@ -317,7 +314,13 @@ function broadcast(payload: unknown): void { } async function pushSnapshot(only?: WebSocket): Promise { - const text = JSON.stringify({ type: 'active-streams', streams: await buildDisplaySnapshot() }); + // `at` is THIS process's clock at send time, and it is load-bearing rather than informational. Every + // timestamp in the payload below — `lastSeen`, `connectedAt`, `ingest.at`, `lastClose.at` — is stamped with + // the same Date.now(), so a client that ages them against ITS OWN clock is subtracting two different + // clocks: a few seconds of skew is enough to pin a channel permanently live or permanently dead, and the + // browser reading is worst exactly when the stream dies. Sending our clock alongside lets the SPA turn all + // of them back into same-clock subtractions (see `serverNow` in useStreamStats). + const text = JSON.stringify({ type: 'active-streams', at: Date.now(), streams: await buildDisplaySnapshot() }); if (only) send(only, text); else for (const ws of sockets) send(ws, text); } diff --git a/src/composables/useStreamStats.ts b/src/composables/useStreamStats.ts index 277bc42..2fbf19a 100644 --- a/src/composables/useStreamStats.ts +++ b/src/composables/useStreamStats.ts @@ -57,6 +57,26 @@ let ws: WebSocket | null = null; let refCount = 0; let reconnectTimer: ReturnType | null = null; +// ── THE SERVER'S CLOCK, as observed from here ─────────────────────────────────────────────────────────── +// Every timestamp the backend sends is stamped with ITS `Date.now()`, so measuring one against OURS +// subtracts two different clocks — see the note above, which is the same trap `recvAt` exists to dodge. Where +// a same-clock stamp is available (a frame's arrival) that remains the better tool. This is for the +// timestamps where none is: a viewer's `connectedAt`, a session's `lastClose.at`, the age of the FIRST +// ingest frame we ever see. One offset, re-estimated every snapshot, turns all of those into same-clock +// subtractions. The round trip inflates it by the network delay (single-digit ms), which is nothing against +// the 10s liveness gate and the 30-90s staleness window it feeds. +let serverOffset = 0; + +function noteServerClock(at: number): void { + if (Number.isFinite(at)) serverOffset = at - Date.now(); +} + +/** The current time on the SERVER's clock. Use for any timestamp the backend stamped; never for measuring + * something this browser observed (that is what a local `recvAt` is for). */ +export function serverNow(): number { + return Date.now() + serverOffset; +} + function ingest(streams: ActiveStream[]): void { ACTIVE_STREAMS.value = streams; const present = new Set(); @@ -73,7 +93,15 @@ function ingest(streams: ActiveStream[]): void { if (!ing) continue; const prev = ingestMeta[s.channelId]; if (!prev) { - ingestMeta[s.channelId] = { lastAt: ing.at, recvAt: now, lastBytes: ing.ingestedBytes, mbps: null }; + // FIRST SIGHT is the one case with no previous frame to compare against, and stamping it `now` asserts + // a freshness we have not observed: Node re-broadcasts the last `iop` snapshot for as long as viewers + // remain, so an ingest that died ten minutes ago arrives looking exactly like one that reported a + // moment ago — and the panel would paint green for the whole staleness window at precisely the moment + // an operator opened it to find out why the stream is frozen. Back-date our arrival stamp by the + // frame's OWN age, measured server-clock to server-clock, so `ingestAge` tells the truth on the first + // render. Clamped at 0: a small negative is offset noise, never evidence of a frame from the future. + const age = Math.max(0, serverNow() - ing.at); + ingestMeta[s.channelId] = { lastAt: ing.at, recvAt: now - age, lastBytes: ing.ingestedBytes, mbps: null }; } else if (ing.at !== prev.lastAt) { const dtMs = ing.at - prev.lastAt; const dBytes = ing.ingestedBytes - prev.lastBytes; @@ -136,6 +164,9 @@ function connect(): void { at?: number; side?: 'upstream' | 'client'; }; + // Re-estimated on EVERY snapshot, before the payload is read: the frame's own `at` is the only + // observation of the server's clock we get, and everything downstream ages against it. + if (typeof msg.at === 'number' && msg.type === 'active-streams') noteServerClock(msg.at); if (msg.type === 'active-streams' && Array.isArray(msg.streams)) ingest(msg.streams); else if (msg.type === 'view-session' && msg.session) ingestSession(msg.session); else if (msg.type === 'buffer-event' && msg.channelKey) ingestBufferEvent(msg); diff --git a/src/screens/ActiveStreamsScreen.vue b/src/screens/ActiveStreamsScreen.vue index d73b48d..684d3ed 100644 --- a/src/screens/ActiveStreamsScreen.vue +++ b/src/screens/ActiveStreamsScreen.vue @@ -9,7 +9,7 @@ import Segmented from '../components/Segmented.vue'; import ChannelLogo from '../components/ChannelLogo.vue'; import LivelineChart from '../components/LivelineChart.vue'; import { ACTIVE_STREAMS, CHANNELS, EPG_PROGRAMS, fetchProgramsFor, flagEmoji, type ActiveStream, type Program, type StreamClient } from '../data'; -import { useStreamStats } from '../composables/useStreamStats'; +import { useStreamStats, serverNow } from '../composables/useStreamStats'; // Live snapshot over the /api/stream-stats WebSocket (updates ACTIVE_STREAMS in place). Only show streams // whose channelId resolves to a real channel in the global list. @@ -213,7 +213,8 @@ watch(liveStreams, loadNowNext, { immediate: true }); // Per-client display helpers. function rateKB(bps: number) { return (bps / 1024).toFixed(0); } -function sinceLabel(ts: number) { const m = Math.floor((Date.now() - ts) / 60000); return m < 1 ? 'just now' : m < 60 ? `${m}m` : `${Math.floor(m / 60)}h ${m % 60}m`; } +// `ts` is a SERVER stamp (a viewer's `connectedAt`), so it ages against the server's clock, not ours. +function sinceLabel(ts: number) { const m = Math.floor((serverNow() - ts) / 60000); return m < 1 ? 'just now' : m < 60 ? `${m}m` : `${Math.floor(m / 60)}h ${m % 60}m`; } // S3/UND: the data plane sends a stable slug (so it can also be recorded against the burnt provider); this // is the operator-facing wording. An unknown slug is shown verbatim rather than hidden — a newer sidecar // reporting a fault this SPA has not learned yet must not silently disappear from the row. @@ -263,7 +264,9 @@ function staleMs(ing: ActiveStream['ingest']): number { } function ageLabel(ms: number): string { if (!Number.isFinite(ms)) return '—'; - const s = Math.round(ms / 1000); + // Clamped: an age is never negative, and `s < 90` below would happily render one verbatim ("-42s ago"). + // Callers now pass a same-clock difference, so this is a backstop against the next caller that does not. + const s = Math.max(0, Math.round(ms / 1000)); return s < 90 ? `${s}s` : `${Math.round(s / 60)}m`; } function mib(bytes: number): string { return (bytes / 1048576).toFixed(1); } @@ -326,7 +329,13 @@ function breakAge(s: ActiveStream): number { const b = s.adBreak; if (!b) return 0; const cur = breakSeen[s.channelId]; - if (!cur || cur.at !== b.at) { breakSeen[s.channelId] = { at: b.at, recvAt: Date.now() }; return 0; } + if (!cur || cur.at !== b.at) { + // Back-dated by the break's own age for the same reason the ingest seed is: a break that opened before + // this screen did must not read as brand new. A genuinely new break still ages to ~0. + const age = Math.max(0, serverNow() - b.at); + breakSeen[s.channelId] = { at: b.at, recvAt: Date.now() - age }; + return age; + } return Date.now() - cur.recvAt; } @@ -507,7 +516,10 @@ function stateManifest(): StageState { function stateOutput(): StageState { const s = sel.value; if (!s) return { tone: 'na', text: '—', title: 'No stream selected.' }; - const now = Date.now(); + // `lastSeen` / `connectedAt` are stamped by NODE. Aging them against the browser clock made a machine even + // ten seconds fast read EVERY row as stale — so this stage, the default-open one, said "draining" forever + // while viewers were demonstrably streaming, and the `no egress` fault below could never fire. + const now = serverNow(); // Only a list KNOWN to describe this channel may drive a verdict — see `clientsOf`. const rows = selClients.value; if (!rows) return { tone: 'unknown', text: 'no reading', title: 'Session list not loaded for this channel yet.' }; @@ -821,7 +833,9 @@ function onRailKey(e: KeyboardEvent): void { · serving {{ deliveryLabel(sel.delivery) }} · fell back to HLS — - + + @@ -1084,7 +1099,7 @@ function onRailKey(e: KeyboardEvent): void {