Code review

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/<entry>/v.m3u8` and `…/o/<entry>/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/<entry>/v.m3u8` and
+`…/o/<entry>/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.

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"));
     }
 

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

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 {

proxy/src/state.rs

@@ -146,6 +146,14 @@ pub struct SourcePolicy {
     pub relabel_segment: RwLock<Option<String>>,
     /// 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<HashSet<String>>,
     /// 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<String>,
     #[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<Arc<SourcePolicy>> {
-        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<Arc<SourcePolicy>> {
+        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<Arc<SourcePolicy>> {
@@ -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);