wip (hls): final cutover

backend/app/src/api/endpoints/hls_api/manifest.rs

@@ -15,6 +15,8 @@ mod access;
 mod admission;
 mod cached;
 mod canonical_owner;
+#[cfg(test)]
+pub(super) use canonical_owner::{await_hls_terminal_or_refresh, current_hls_refresh_lease};
 mod provisioning;
 mod publication;
 #[allow(unused_imports, reason = "Retains the existing module interface in production and test builds.")]

backend/app/src/api/endpoints/hls_api/manifest/canonical_owner.rs

@@ -393,6 +393,99 @@ pub(in crate::api::endpoints::hls_api) enum HlsManifestRefreshOrdering {
     AwaitBeforeTerminalEvaluation,
 }
 
+pub(in crate::api::endpoints::hls_api) async fn current_hls_refresh_lease(
+    app_state: &Arc<AppState>,
+    session: &HlsSessionHandle,
+    proxy_session_id: &ProxySessionId,
+    access_lease_id: &HlsAccessLeaseId,
+    expected: &HlsAccessLease,
+) -> Result<HlsAccessLease, Box<axum::response::Response>> {
+    let unavailable =
+        || Box::new(hls_terminal_failed_closed_response(HlsTerminalFailedClosedReason::LeaseStateUnavailable));
+    let sessions = app_state.hls.proxy.sessions();
+    let current_session = sessions.get_by_proxy_session_id(proxy_session_id).await.ok_or_else(unavailable)?;
+    if !Arc::ptr_eq(session, &current_session)
+        || sessions.session_incarnation(session).is_none()
+        || sessions.session_incarnation(session) != sessions.session_incarnation(&current_session)
+    {
+        return Err(unavailable());
+    }
+    let lease = app_state
+        .hls
+        .proxy
+        .access_lease_response_snapshot(access_lease_id, proxy_session_id, app_state.hls.proxy.now_ms())
+        .await
+        .ok_or_else(unavailable)?;
+    let terminal_transition = expected.playback_mode == HlsLeasePlaybackMode::Live
+        && matches!(
+            lease.playback_mode,
+            HlsLeasePlaybackMode::TerminalTail(_)
+                | HlsLeasePlaybackMode::TerminalCustomResponse(_)
+                | HlsLeasePlaybackMode::TerminalUnavailable { .. }
+        );
+    let expected_admission = if terminal_transition {
+        expected.admission_generation.checked_add(1)
+    } else {
+        Some(expected.admission_generation)
+    };
+    if lease.issued_at_ms != expected.issued_at_ms || Some(lease.admission_generation) != expected_admission {
+        return Err(unavailable());
+    }
+    if !sessions.get_by_proxy_session_id(proxy_session_id).await.is_some_and(|current| Arc::ptr_eq(session, &current)) {
+        return Err(unavailable());
+    }
+    Ok(lease)
+}
+
+pub(in crate::api::endpoints::hls_api) async fn await_hls_terminal_or_refresh(
+    app_state: &Arc<AppState>,
+    session: &HlsSessionHandle,
+    expected: &HlsAccessLease,
+    mut lease: HlsAccessLease,
+    mut commits: tokio::sync::watch::Receiver<()>,
+    refresh: impl std::future::Future<Output = bool>,
+) -> Option<axum::response::Response> {
+    let proxy_session_id = &expected.proxy_session_id;
+    let access_lease_id = &expected.lease_id;
+    tokio::pin!(refresh);
+    loop {
+        if let Some(response) = hls_terminal_playback_response(&lease, proxy_session_id, access_lease_id) {
+            return Some(response);
+        }
+        tokio::select! {
+            _ = &mut refresh => break,
+            changed = commits.changed() => {
+                if changed.is_err() {
+                    // The session handle normally keeps the sender alive.
+                    refresh.await;
+                    break;
+                }
+                lease = match current_hls_refresh_lease(
+                    app_state, session, proxy_session_id, access_lease_id, expected,
+                ).await {
+                    Ok(lease) => lease,
+                    Err(response) => return Some(*response),
+                };
+            }
+        }
+    }
+    let lease = match current_hls_refresh_lease(app_state, session, proxy_session_id, access_lease_id, expected).await {
+        Ok(lease) => lease,
+        Err(response) => return Some(*response),
+    };
+    resolve_hls_terminal_manifest_state(
+        app_state,
+        session,
+        proxy_session_id,
+        access_lease_id,
+        lease,
+        app_state.hls.proxy.now_ms(),
+    )
+    .await
+    .err()
+    .map(|response| *response)
+}
+
 pub(in crate::api::endpoints::hls_api) async fn trigger_hls_canonical_manifest_refresh(
     app_state: &Arc<AppState>,
     session: &HlsSessionHandle,
@@ -407,6 +500,35 @@ pub(in crate::api::endpoints::hls_api) async fn trigger_hls_canonical_manifest_r
             hls_direct_refresh_follow_up(app_state, session, proxy_session_id, refresh_request, outcome).await
         }
         HlsManifestRefreshOrdering::AwaitBeforeTerminalEvaluation => {
+            let initial = app_state
+                .hls
+                .proxy
+                .access_lease_response_snapshot(access_lease_id, proxy_session_id, app_state.hls.proxy.now_ms())
+                .await;
+            if let Some(expected) = initial
+                .as_ref()
+                .filter(|lease| lease.state == HlsAccessLeaseState::Activated && lease.last_manifest_snapshot.is_some())
+            {
+                // Subscribe before checking again: a commit in either gap is
+                // observed through current state or the retained watch version.
+                let commits = session.read().await.subscribe_terminal_commits();
+                let lease =
+                    match current_hls_refresh_lease(app_state, session, proxy_session_id, access_lease_id, expected)
+                        .await
+                    {
+                        Ok(lease) => lease,
+                        Err(response) => return Some(*response),
+                    };
+                return await_hls_terminal_or_refresh(
+                    app_state,
+                    session,
+                    expected,
+                    lease,
+                    commits,
+                    trigger_origin_refresh_sync(refresh_request),
+                )
+                .await;
+            }
             // An already-owned refresh cannot be joined through this call, but
             // it must not bypass the lease-specific terminal decision. The
             // in-flight owner will still publish its eventual progress/failure.

backend/app/src/api/endpoints/hls_api/tests/runtime_policy.rs

@@ -21,6 +21,7 @@ pub(in crate::api::endpoints::hls_api::tests) async fn assert_runtime_policy_ter
 
 #[tokio::test]
 async fn resource_access_denial_commits_and_preserves_user_exhausted_tail() {
+    let _ = env_logger::Builder::from_env(env_logger::Env::default()).is_test(true).try_init();
     let fixture = runtime_policy_endpoint_fixture(true).await;
     let session = fixture
         .app_state

backend/app/src/api/endpoints/hls_api/tests/support/origins.rs

@@ -47,15 +47,16 @@ pub(in crate::api::endpoints::hls_api::tests) struct TestBinaryOriginResponse {
     pub(in crate::api::endpoints::hls_api::tests) status: StatusCode,
     pub(in crate::api::endpoints::hls_api::tests) location: Option<String>,
     pub(in crate::api::endpoints::hls_api::tests) body: Arc<[u8]>,
+    pub(in crate::api::endpoints::hls_api::tests) gate: Option<tokio::sync::watch::Receiver<bool>>,
 }
 
 impl TestBinaryOriginResponse {
     pub(in crate::api::endpoints::hls_api::tests) fn new(status: StatusCode, body: Arc<[u8]>) -> Self {
-        Self { status, location: None, body }
+        Self { status, location: None, body, gate: None }
     }
 
     pub(in crate::api::endpoints::hls_api::tests) fn redirect(location: String) -> Self {
-        Self { status: StatusCode::FOUND, location: Some(location), body: Arc::from(&b""[..]) }
+        Self { status: StatusCode::FOUND, location: Some(location), body: Arc::from(&b""[..]), gate: None }
     }
 }
 
@@ -91,7 +92,14 @@ pub(in crate::api::endpoints::hls_api::tests) async fn spawn_test_binary_origin(
                 if path_has_extension(path, "m3u8") {
                     manifest_requests.fetch_add(1, Ordering::SeqCst);
                 }
-                let TestBinaryOriginResponse { status, location, body } = handler(path);
+                let TestBinaryOriginResponse { status, location, body, gate } = handler(path);
+                if let Some(mut gate) = gate {
+                    while !*gate.borrow_and_update() {
+                        if gate.changed().await.is_err() {
+                            return;
+                        }
+                    }
+                }
                 let reason = status.canonical_reason().unwrap_or("Status");
                 let location_header = location.map_or_else(String::new, |location| format!("Location: {location}\r\n"));
                 let response = format!(

backend/app/src/api/endpoints/hls_api/tests/terminal_tail.rs

@@ -4,9 +4,9 @@ use super::{
     get_status, media_uri_count, normal_manifest, path_has_extension, prepare_pending_test_hls_access_lease,
     proxy_session_id_from_variant_uri, regression_origin_manifest, response_body, single_variant_master_playlist,
     spawn_test_binary_origin, spawn_test_encrypted_hls_origin, store_test_sources_with_target, terminal_test_asset,
-    test_app_state_with_hls_proxy_and_inputs, test_app_state_with_inputs, test_fingerprint, test_hls_access_context,
-    test_hls_entry_stream_context, test_m3u_hls_item, test_m3u_hls_share_target, try_test_hls_cached_manifest_response,
-    TestBinaryOriginResponse, TestSegmentOrigin, AES_TEST_KEY_BYTES, AES_TEST_MANIFEST, AES_TEST_PLAINTEXT_SEGMENT,
+    test_app_state_with_hls_proxy_and_inputs, test_fingerprint, test_hls_access_context, test_hls_entry_stream_context,
+    test_m3u_hls_item, test_m3u_hls_share_target, try_test_hls_cached_manifest_response, TestBinaryOriginResponse,
+    TestSegmentOrigin, AES_TEST_KEY_BYTES, AES_TEST_MANIFEST, AES_TEST_PLAINTEXT_SEGMENT,
 };
 use crate::{
     api::model::{
@@ -1119,6 +1119,7 @@ pub(in crate::api::endpoints::hls_api::tests) fn test_beast_hls_proxy(
 }
 
 pub(in crate::api::endpoints::hls_api::tests) struct PublicationLateFixture {
+    pub(in crate::api::endpoints::hls_api::tests) _directory: tempfile::TempDir,
     pub(in crate::api::endpoints::hls_api::tests) origin: TestSegmentOrigin,
     pub(in crate::api::endpoints::hls_api::tests) origin_phase: Arc<AtomicUsize>,
     pub(in crate::api::endpoints::hls_api::tests) app_state: Arc<AppState>,
@@ -1151,6 +1152,15 @@ pub(in crate::api::endpoints::hls_api::tests) async fn publication_late_fixture(
         TestBinaryOriginResponse::new(StatusCode::OK, Arc::clone(&segment))
     }))
     .await;
+    publication_fixture_with_origin(HlsCacheConfigDto::default(), origin, origin_phase).await
+}
+
+pub(in crate::api::endpoints::hls_api::tests) async fn publication_fixture_with_origin(
+    mut config: HlsCacheConfigDto,
+    origin: TestSegmentOrigin,
+    origin_phase: Arc<AtomicUsize>,
+) -> PublicationLateFixture {
+    let directory = tempfile::tempdir().unwrap();
     let input = ConfigInput {
         id: 1,
         name: Arc::from("publication-late-request-input"),
@@ -1162,7 +1172,9 @@ pub(in crate::api::endpoints::hls_api::tests) async fn publication_late_fixture(
     };
     let mut target = test_m3u_hls_share_target();
     target.name = "default".to_string();
-    let app_state = test_app_state_with_inputs(vec![Arc::new(input.clone())]);
+    config.cache_path = Some(directory.path().to_string_lossy().into_owned());
+    let manager = Arc::new(HlsProxyManager::with_hls_cache_config(&HlsCacheConfig::from(&config)));
+    let app_state = test_app_state_with_hls_proxy_and_inputs(manager, vec![Arc::new(input.clone())]);
     enable_hls_cache(&app_state);
     enable_channel_unavailable_custom_response(&app_state);
     configure_default_test_server(&app_state);
@@ -1213,6 +1225,7 @@ pub(in crate::api::endpoints::hls_api::tests) async fn publication_late_fixture(
         .await
         .expect("shared publication-late session");
     PublicationLateFixture {
+        _directory: directory,
         origin,
         origin_phase,
         app_state,

backend/app/src/api/endpoints/hls_api/tests/terminal_tail_terminal_cutover.rs

@@ -14,7 +14,756 @@ use crate::api::model::{
     ProxySessionId,
 };
 use axum::http::{header, StatusCode};
-use std::sync::Arc;
+use std::{
+    fmt::Write as _,
+    sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    },
+};
+
+async fn within_real_second<F: std::future::Future>(future: F) -> F::Output {
+    tokio::pin!(future);
+    let started = std::time::Instant::now();
+    loop {
+        let polled = futures::poll!(future.as_mut());
+        assert!(started.elapsed() < std::time::Duration::from_secs(1), "real-time safety deadline");
+        if let std::task::Poll::Ready(value) = polled {
+            return value;
+        }
+        // Keep the paused runtime runnable: a response after commit must not
+        // require Tokio to auto-advance to an outstanding origin timeout.
+        tokio::task::yield_now().await;
+    }
+}
+
+struct DelayedTerminalRefresh {
+    fixture: super::terminal_tail::PublicationLateFixture,
+    release: tokio::sync::watch::Sender<bool>,
+    blocked: tokio::sync::watch::Receiver<bool>,
+}
+
+type PendingRefresh = futures::future::Shared<futures::future::BoxFuture<'static, bool>>;
+
+impl DelayedTerminalRefresh {
+    async fn new(mode: shared::model::HlsStartupMode) -> Self {
+        let _ = env_logger::Builder::from_env(env_logger::Env::default()).is_test(true).try_init();
+        let playlist = |first: u64| {
+            let mut body = format!("#EXTM3U\n#EXT-X-TARGETDURATION:12\n#EXT-X-MEDIA-SEQUENCE:{first}\n");
+            for seq in first..first + 6 {
+                writeln!(body, "#EXTINF:10.027,\n{seq}.ts").unwrap();
+            }
+            Arc::<[u8]>::from(body.into_bytes())
+        };
+        let (initial, progressed) = (playlist(100), playlist(101));
+        let phase = Arc::new(AtomicUsize::new(0));
+        let handler_phase = Arc::clone(&phase);
+        let attempts = AtomicUsize::new(0);
+        let (release, gate) = tokio::sync::watch::channel(false);
+        let (entered, blocked) = tokio::sync::watch::channel(false);
+        let origin = super::spawn_test_binary_origin(Arc::new(move |path| {
+            if !super::path_has_extension(path, "m3u8") {
+                return super::TestBinaryOriginResponse::new(
+                    StatusCode::OK,
+                    Arc::from(
+                        include_bytes!(concat!(
+                            env!("CARGO_MANIFEST_DIR"),
+                            "/../../test/fixtures/hls/channel_unavailable.ts"
+                        ))
+                        .as_slice(),
+                    ),
+                );
+            }
+            if handler_phase.load(Ordering::SeqCst) == 0 {
+                return super::TestBinaryOriginResponse::new(StatusCode::OK, Arc::clone(&initial));
+            }
+            if attempts.fetch_add(1, Ordering::SeqCst) == 0 {
+                return super::TestBinaryOriginResponse::new(StatusCode::BAD_GATEWAY, Arc::from(&b"retry"[..]));
+            }
+            entered.send_replace(true);
+            let status = if handler_phase.load(Ordering::SeqCst) == 2 { StatusCode::NOT_FOUND } else { StatusCode::OK };
+            let mut response = super::TestBinaryOriginResponse::new(status, Arc::clone(&progressed));
+            response.gate = Some(gate.clone());
+            response
+        }))
+        .await;
+        let mut config = shared::model::HlsCacheConfigDto {
+            origin_manifest_timeout_ms: shared::model::Millis::new(120_000),
+            ..Default::default()
+        };
+        config.startup.mode = mode;
+        let fixture = super::terminal_tail::publication_fixture_with_origin(config, origin, phase).await;
+        super::terminal_tail::wait_for_ready_timeline(&fixture.session, 3).await;
+        let lease = fixture
+            .app_state
+            .hls
+            .proxy
+            .access_lease_response_snapshot(
+                &fixture.access_lease_id,
+                &fixture.proxy_session_id,
+                fixture.app_state.hls.proxy.now_ms(),
+            )
+            .await
+            .unwrap();
+        let snapshot = lease.last_manifest_snapshot.as_ref().unwrap();
+        let preparation_observer = fixture.session.read().await.subscribe_terminal_commits();
+        super::terminal_tail::prepare_terminal_cutover_bundle(&fixture.app_state, snapshot).await;
+        assert!(!preparation_observer.has_changed().unwrap(), "bundle preparation must not publish a commit signal");
+        drop(preparation_observer);
+        let response = get_response(Arc::clone(&fixture.app_state), &snapshot.visible_segments[0].uri, None).await;
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(!response_body(response).await.is_empty());
+        let base = fixture.app_state.hls.proxy.now_ms();
+        tokio::time::pause();
+        fixture.app_state.hls.proxy.follow_tokio_clock_for_test(base);
+        {
+            let mut session = fixture.session.write().await;
+            session.origin_refresh.next_fetch_allowed_at_ms = 0;
+            session.origin_control.last_media_progress_at_ms = Some(base.saturating_sub(24_000));
+        }
+        fixture.origin_phase.store(1, Ordering::SeqCst);
+        Self { fixture, release, blocked }
+    }
+
+    async fn reserve(&self) -> u64 {
+        let manager = &self.fixture.app_state.hls.proxy;
+        let now = manager.now_ms();
+        let lease = manager
+            .access_lease_response_snapshot(&self.fixture.access_lease_id, &self.fixture.proxy_session_id, now)
+            .await
+            .unwrap();
+        let manifest = lease.last_manifest_snapshot.as_ref().unwrap();
+        let session = self.fixture.session.read().await;
+        let reserve = tuliprox_hls::api::evaluate_lease_reserve(tuliprox_hls::api::HlsLeaseReserveInput {
+            manifest,
+            cursor: &lease.playback_cursor,
+            ready_timeline: &session.ready_timeline_snapshot(
+                lease.playback_cursor.ready_timeline_start_proxy_seq(manifest.first_proxy_seq),
+                now,
+            ),
+            now_ms: now,
+            recovery_trigger_budget: tuliprox_hls::api::HlsRecoveryTriggerBudgetMs::from_millis(0),
+            origin_path_degraded: session.origin_control.path_condition.is_degraded(),
+            recovery_committed: false,
+        });
+        assert!(reserve.playback_position_known);
+        reserve.guaranteed_reserve_ms
+    }
+
+    async fn waiting_request(&mut self) -> tokio::task::JoinHandle<axum::response::Response> {
+        let app = Arc::clone(&self.fixture.app_state);
+        let uri = self.fixture.media_playlist_uri.clone();
+        let request = tokio::spawn(async move { get_response(app, &uri, None).await });
+        self.wait_for_blocked_origin().await;
+        assert!(!request.is_finished(), "request must wait with positive reserve");
+        request
+    }
+
+    async fn wait_for_blocked_origin(&mut self) {
+        within_real_second(async {
+            loop {
+                if self.fixture.session.read().await.origin_control.path_condition.is_degraded() {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await;
+        // Drive only the pre-commit retry stagger, allowing the native socket
+        // to reach its explicit gate before advancing to media exhaustion.
+        for _ in 0..20 {
+            if *self.blocked.borrow() {
+                break;
+            }
+            tokio::time::advance(std::time::Duration::from_millis(250)).await;
+            let started = std::time::Instant::now();
+            while started.elapsed() < std::time::Duration::from_millis(5) {
+                tokio::task::yield_now().await;
+            }
+        }
+        within_real_second(self.blocked.wait_for(|entered| *entered)).await.unwrap();
+        assert!(self.fixture.session.read().await.origin_refresh.in_flight);
+        assert!(self.reserve().await > 0);
+    }
+
+    fn refresh_request(&self) -> tuliprox_hls::api::OriginRefreshRequest {
+        let fixture = &self.fixture;
+        let mut request = super::terminal_tail::regression_origin_refresh_request(
+            &fixture.app_state,
+            Arc::clone(&fixture.session),
+            &format!("{}/channel/index.m3u8", fixture.origin.base_url),
+            Some(fixture.access_lease_id.clone()),
+        );
+        request.now_ms = fixture.app_state.hls.proxy.now_ms();
+        request.post_refresh_runtime =
+            Some(tuliprox_hls::api::HlsPostRefreshRuntime { ctx: fixture.app_state.hls_ctx().downgrade() });
+        request.origin_io = Some(super::provider_admission::test_hls_origin_io_context(&fixture.app_state));
+        request
+    }
+
+    async fn pending_refresh(&mut self) -> PendingRefresh {
+        use futures::FutureExt as _;
+        let mut refresh = tuliprox_hls::api::trigger_origin_refresh_sync(self.refresh_request()).boxed().shared();
+        assert!(futures::poll!(&mut refresh).is_pending());
+        self.wait_for_blocked_origin().await;
+        refresh
+    }
+
+    async fn lease(&self) -> tuliprox_hls::api::HlsAccessLease {
+        let fixture = &self.fixture;
+        fixture
+            .app_state
+            .hls
+            .proxy
+            .access_lease_response_snapshot(
+                &fixture.access_lease_id,
+                &fixture.proxy_session_id,
+                fixture.app_state.hls.proxy.now_ms(),
+            )
+            .await
+            .unwrap()
+    }
+
+    async fn consume_reserve(&self) -> u64 {
+        let mut commits = self.fixture.session.read().await.subscribe_terminal_commits();
+        tokio::time::advance(std::time::Duration::from_millis(self.reserve().await)).await;
+        within_real_second(commits.changed()).await.unwrap();
+        assert!(matches!(
+            self.lease().await.playback_mode,
+            HlsLeasePlaybackMode::TerminalTail(_) | HlsLeasePlaybackMode::TerminalUnavailable { .. }
+        ));
+        assert!(self.fixture.session.read().await.origin_refresh.in_flight);
+        assert!(!*self.release.borrow());
+        self.fixture.app_state.hls.proxy.now_ms()
+    }
+
+    async fn finish_refresh(&self, refresh: PendingRefresh) {
+        self.release.send_replace(true);
+        assert!(within_real_second(refresh).await);
+        let state = self.fixture.session.read().await;
+        assert!(!state.origin_refresh.in_flight);
+        assert_eq!(state.activity.active_origin_work_count.load(std::sync::atomic::Ordering::Acquire), 0);
+        assert!(state.origin_account_io_lease.as_ref().is_none_or(|lease| !lease.is_active_or_acquiring()));
+    }
+}
+
+#[tokio::test]
+async fn terminal_notification_releases_conservative_manifest_before_origin_refresh_finishes() {
+    delayed_terminal_response_before_refresh(shared::model::HlsStartupMode::Conservative).await;
+}
+
+#[derive(Clone, Copy)]
+enum CommitSignalRace {
+    Unsubscribed,
+    Subscribed,
+    Checked,
+}
+
+async fn terminal_notification_race(race: CommitSignalRace) {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Optimistic).await;
+    let refresh = controlled.pending_refresh().await;
+    let expected = controlled.lease().await;
+    let fixture = &controlled.fixture;
+    let mut commits = None;
+    let mut checked = None;
+    if !matches!(race, CommitSignalRace::Unsubscribed) {
+        commits = Some(fixture.session.read().await.subscribe_terminal_commits());
+    }
+    if matches!(race, CommitSignalRace::Checked) {
+        checked = Some(
+            super::super::current_hls_refresh_lease(
+                &fixture.app_state,
+                &fixture.session,
+                &fixture.proxy_session_id,
+                &fixture.access_lease_id,
+                &expected,
+            )
+            .await
+            .unwrap(),
+        );
+    }
+    let commit_at = controlled.consume_reserve().await;
+    let commits = match commits {
+        Some(commits) => commits,
+        None => fixture.session.read().await.subscribe_terminal_commits(),
+    };
+    let checked = match checked {
+        Some(lease) => lease,
+        None => super::super::current_hls_refresh_lease(
+            &fixture.app_state,
+            &fixture.session,
+            &fixture.proxy_session_id,
+            &fixture.access_lease_id,
+            &expected,
+        )
+        .await
+        .unwrap(),
+    };
+    let response = within_real_second(super::super::await_hls_terminal_or_refresh(
+        &fixture.app_state,
+        &fixture.session,
+        &expected,
+        checked,
+        commits,
+        refresh.clone(),
+    ))
+    .await
+    .unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(String::from_utf8(response_body(response).await.to_vec()).unwrap().ends_with("#EXT-X-ENDLIST\n"));
+    assert_eq!(fixture.app_state.hls.proxy.now_ms(), commit_at);
+    assert!(fixture.session.read().await.origin_refresh.in_flight);
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    controlled.finish_refresh(refresh).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_commit_before_subscription_is_not_lost() {
+    terminal_notification_race(CommitSignalRace::Unsubscribed).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_commit_between_subscription_and_check_is_not_lost() {
+    terminal_notification_race(CommitSignalRace::Subscribed).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_commit_between_check_and_wait_is_not_lost() {
+    terminal_notification_race(CommitSignalRace::Checked).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_wakes_all_waiters_without_another_refresh_or_generation() {
+    use futures::FutureExt as _;
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Progressive).await;
+    let refresh = controlled.pending_refresh().await;
+    let expected = controlled.lease().await;
+    let fixture = &controlled.fixture;
+    let starts = fixture.app_state.hls.proxy.metrics().snapshot().refresh_started;
+    let mut waiters = Vec::new();
+    for _ in 0..3 {
+        let commits = fixture.session.read().await.subscribe_terminal_commits();
+        let lease = super::super::current_hls_refresh_lease(
+            &fixture.app_state,
+            &fixture.session,
+            &fixture.proxy_session_id,
+            &fixture.access_lease_id,
+            &expected,
+        )
+        .await
+        .unwrap();
+        let mut waiter = super::super::await_hls_terminal_or_refresh(
+            &fixture.app_state,
+            &fixture.session,
+            &expected,
+            lease,
+            commits,
+            refresh.clone(),
+        )
+        .boxed();
+        assert!(futures::poll!(&mut waiter).is_pending());
+        waiters.push(waiter);
+    }
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 3);
+    let commit_at = controlled.consume_reserve().await;
+    let responses = within_real_second(futures::future::join_all(waiters)).await;
+    let mut bodies = Vec::new();
+    for response in responses {
+        let response = response.unwrap();
+        assert_eq!(response.status(), StatusCode::OK);
+        bodies.push(response_body(response).await);
+    }
+    assert_eq!(bodies[0], bodies[1]);
+    assert_eq!(bodies[1], bodies[2]);
+    assert_eq!(fixture.app_state.hls.proxy.now_ms(), commit_at);
+    assert_eq!(fixture.app_state.hls.proxy.metrics().snapshot().refresh_started, starts);
+    assert_eq!(controlled.lease().await.admission_generation, expected.admission_generation + 1);
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    controlled.finish_refresh(refresh).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_client_abort_drops_observer_without_aborting_origin() {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Optimistic).await;
+    let request = controlled.waiting_request().await;
+    let fixture = &controlled.fixture;
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 1);
+    request.abort();
+    assert!(request.await.unwrap_err().is_cancelled());
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    assert!(fixture.session.read().await.origin_refresh.in_flight);
+    controlled.release.send_replace(true);
+    within_real_second(async {
+        while fixture.session.read().await.origin_refresh.in_flight {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await;
+    assert_eq!(
+        fixture.session.read().await.activity.active_origin_work_count.load(std::sync::atomic::Ordering::Acquire),
+        0
+    );
+}
+
+#[tokio::test]
+async fn terminal_notification_successful_refresh_before_exhaustion_keeps_manifest_live() {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::FirstReady).await;
+    let waiting = controlled.waiting_request().await;
+    controlled.release.send_replace(true);
+    let response = within_real_second(waiting).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(!String::from_utf8(response_body(response).await.to_vec()).unwrap().contains("#EXT-X-ENDLIST"));
+    assert_eq!(controlled.lease().await.playback_mode, HlsLeasePlaybackMode::Live);
+    assert!(controlled.reserve().await > 0);
+    assert_eq!(controlled.fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+}
+
+#[derive(Clone, Copy)]
+enum RefreshWaitReplacement {
+    Session,
+    Lease,
+    Admission,
+}
+
+#[tokio::test]
+async fn terminal_notification_singleflight_followers_return_live_without_new_origin_starts() {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Progressive).await;
+    let owner = controlled.waiting_request().await;
+    let fixture = &controlled.fixture;
+    let starts = fixture.app_state.hls.proxy.metrics().snapshot().refresh_started;
+    for _ in 0..4 {
+        let response =
+            within_real_second(get_response(Arc::clone(&fixture.app_state), &fixture.media_playlist_uri, None)).await;
+        assert_eq!(response.status(), StatusCode::OK);
+        assert!(!String::from_utf8(response_body(response).await.to_vec()).unwrap().contains("#EXT-X-ENDLIST"));
+        assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 1);
+    }
+    assert_eq!(fixture.app_state.hls.proxy.metrics().snapshot().refresh_started, starts);
+    assert!(!owner.is_finished());
+    controlled.release.send_replace(true);
+    assert_eq!(within_real_second(owner).await.unwrap().status(), StatusCode::OK);
+}
+
+#[tokio::test]
+async fn terminal_notification_refresh_failure_keeps_positive_reserve_live_and_releases_work() {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::FirstReady).await;
+    controlled.fixture.origin_phase.store(2, std::sync::atomic::Ordering::SeqCst);
+    let waiting = controlled.waiting_request().await;
+    controlled.release.send_replace(true);
+    let response = within_real_second(waiting).await.unwrap();
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(!String::from_utf8(response_body(response).await.to_vec()).unwrap().contains("#EXT-X-ENDLIST"));
+    assert!(controlled.reserve().await > 0);
+    assert_eq!(controlled.lease().await.playback_mode, HlsLeasePlaybackMode::Live);
+    let session = controlled.fixture.session.read().await;
+    assert!(!session.origin_refresh.in_flight);
+    assert!(session.origin_control.path_condition.is_degraded());
+    assert_eq!(session.activity.active_origin_work_count.load(std::sync::atomic::Ordering::Acquire), 0);
+    assert_eq!(session.terminal_commit_observer_count_for_test(), 0);
+}
+
+#[tokio::test]
+async fn terminal_notification_unavailable_releases_waiter_before_origin_finishes() {
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Optimistic).await;
+    let waiting = controlled.waiting_request().await;
+    controlled.fixture.app_state.app_config.custom_stream_response.store(None);
+    let commit_at = controlled.consume_reserve().await;
+    let response = within_real_second(waiting).await.unwrap();
+    assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+    assert!(response_body(response).await.is_empty());
+    assert!(matches!(controlled.lease().await.playback_mode, HlsLeasePlaybackMode::TerminalUnavailable { .. }));
+    assert_eq!(controlled.fixture.app_state.hls.proxy.now_ms(), commit_at);
+    assert!(controlled.fixture.session.read().await.origin_refresh.in_flight);
+    assert_eq!(controlled.fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    controlled.release.send_replace(true);
+    within_real_second(async {
+        while controlled.fixture.session.read().await.origin_refresh.in_flight {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await;
+}
+
+async fn terminal_notification_replacement(replacement: RefreshWaitReplacement) {
+    use futures::FutureExt as _;
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Optimistic).await;
+    let refresh = controlled.pending_refresh().await;
+    let expected = controlled.lease().await;
+    let fixture = &controlled.fixture;
+    let commits = fixture.session.read().await.subscribe_terminal_commits();
+    let checked = super::super::current_hls_refresh_lease(
+        &fixture.app_state,
+        &fixture.session,
+        &fixture.proxy_session_id,
+        &fixture.access_lease_id,
+        &expected,
+    )
+    .await
+    .unwrap();
+    let mut waiting = super::super::await_hls_terminal_or_refresh(
+        &fixture.app_state,
+        &fixture.session,
+        &expected,
+        checked,
+        commits,
+        refresh.clone(),
+    )
+    .boxed();
+    assert!(futures::poll!(&mut waiting).is_pending());
+    controlled.consume_reserve().await;
+    match replacement {
+        RefreshWaitReplacement::Session => {
+            let key = fixture.session.read().await.key.clone();
+            fixture.app_state.hls.proxy.sessions().remove_session(&key, &fixture.proxy_session_id).await.unwrap();
+            let replacement = fixture
+                .app_state
+                .hls
+                .proxy
+                .get_or_create_session(
+                    key,
+                    &fixture.app_state.get_encrypt_secret(),
+                    fixture.app_state.hls.proxy.now_ms(),
+                )
+                .await;
+            assert!(!Arc::ptr_eq(&replacement, &fixture.session));
+        }
+        RefreshWaitReplacement::Lease | RefreshWaitReplacement::Admission => {
+            let mut leases = fixture.app_state.hls.proxy.access_leases().write().await;
+            let mut lease = leases.remove_access_lease(&fixture.access_lease_id).unwrap();
+            match replacement {
+                RefreshWaitReplacement::Lease => lease.issued_at_ms += 1,
+                RefreshWaitReplacement::Admission => lease.admission_generation += 1,
+                RefreshWaitReplacement::Session => unreachable!(),
+            }
+            leases.prepare_access_lease(lease);
+        }
+    }
+    let response = within_real_second(waiting).await.unwrap();
+    assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+    assert!(response_body(response).await.is_empty());
+    assert!(fixture.session.read().await.origin_refresh.in_flight);
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    controlled.finish_refresh(refresh).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_revalidates_session_incarnation_before_responding() {
+    terminal_notification_replacement(RefreshWaitReplacement::Session).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_revalidates_lease_incarnation_before_responding() {
+    terminal_notification_replacement(RefreshWaitReplacement::Lease).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_revalidates_admission_before_responding() {
+    terminal_notification_replacement(RefreshWaitReplacement::Admission).await;
+}
+
+#[tokio::test]
+#[allow(clippy::too_many_lines, reason = "One controlled two-lease notification and recovery scenario.")]
+async fn terminal_notification_for_another_lease_keeps_own_request_waiting_for_recovery() {
+    use futures::FutureExt as _;
+    let mut controlled = DelayedTerminalRefresh::new(shared::model::HlsStartupMode::Optimistic).await;
+    let fixture = &controlled.fixture;
+    let other = HlsAccessLeaseId("foreign-terminal-observer".to_owned());
+    let snapshot = within_real_second(async {
+        super::prepare_pending_test_hls_access_lease(&fixture.app_state, &fixture.proxy_session_id, &other).await;
+        let original = controlled.lease().await;
+        {
+            let mut leases = fixture.app_state.hls.proxy.access_leases().write().await;
+            let mut lease = leases.remove_access_lease(&other).unwrap();
+            lease.username = original.username;
+            lease.user_session_token = original.user_session_token;
+            leases.prepare_access_lease(lease);
+        }
+        let manifest = super::try_test_hls_cached_manifest_response(
+            &fixture.app_state,
+            &fixture.session,
+            &other,
+            tuliprox_hls::api::HlsAccessLeaseState::Pending,
+            &fixture.app_state.hls.proxy.strip(),
+            None,
+            tuliprox_hls::api::HlsCachedManifestOptions::initial(std::time::Duration::ZERO),
+        )
+        .await
+        .unwrap();
+        assert_eq!(manifest.status(), StatusCode::OK);
+        response_body(manifest).await;
+        let manager = &fixture.app_state.hls.proxy;
+        assert!(manager
+            .activate_access_lease(
+                &other,
+                &fixture.proxy_session_id,
+                manager.now_ms(),
+                tuliprox_hls::api::HlsAccessLeaseTiming { active_window_ms: 120_000, valid_window_ms: 180_000 }
+            )
+            .await
+            .is_activated());
+        let snapshot = manager
+            .access_lease_response_snapshot(&other, &fixture.proxy_session_id, manager.now_ms())
+            .await
+            .unwrap()
+            .last_manifest_snapshot
+            .unwrap();
+        let segment =
+            get_response(Arc::clone(&fixture.app_state), snapshot.visible_segments.last().unwrap().uri.as_ref(), None)
+                .await;
+        assert_eq!(segment.status(), StatusCode::OK);
+        assert!(!response_body(segment).await.is_empty());
+        snapshot
+    })
+    .await;
+    let refresh = controlled.pending_refresh().await;
+    let expected = controlled.lease().await;
+    let fixture = &controlled.fixture;
+    let manager = &fixture.app_state.hls.proxy;
+    let notifications = fixture.session.read().await.subscribe_terminal_commits();
+    let checked = super::super::current_hls_refresh_lease(
+        &fixture.app_state,
+        &fixture.session,
+        &fixture.proxy_session_id,
+        &fixture.access_lease_id,
+        &expected,
+    )
+    .await
+    .unwrap();
+    let mut request = super::super::await_hls_terminal_or_refresh(
+        &fixture.app_state,
+        &fixture.session,
+        &expected,
+        checked,
+        notifications,
+        refresh.clone(),
+    )
+    .boxed();
+    assert!(futures::poll!(&mut request).is_pending());
+    let other_lease =
+        manager.access_lease_response_snapshot(&other, &fixture.proxy_session_id, manager.now_ms()).await.unwrap();
+    let exhaustion = other_lease.playback_cursor.first_segment_started_at_ms.unwrap()
+        + snapshot.visible_segments.last().unwrap().duration_ms;
+    let mut commits = fixture.session.read().await.subscribe_terminal_commits();
+    tokio::time::advance(std::time::Duration::from_millis(exhaustion.saturating_sub(manager.now_ms()))).await;
+    within_real_second(commits.changed()).await.unwrap();
+    assert!(matches!(
+        manager
+            .access_lease_response_snapshot(&other, &fixture.proxy_session_id, manager.now_ms())
+            .await
+            .unwrap()
+            .playback_mode,
+        HlsLeasePlaybackMode::TerminalTail(_)
+    ));
+    assert_eq!(controlled.lease().await.playback_mode, HlsLeasePlaybackMode::Live);
+    assert!(controlled.reserve().await > 0);
+    assert!(futures::poll!(&mut request).is_pending(), "a foreign commit is only a wake-up signal");
+    drop(commits);
+    controlled.release.send_replace(true);
+    assert!(within_real_second(request).await.is_none());
+    controlled.finish_refresh(refresh).await;
+    let response = get_response(Arc::clone(&fixture.app_state), &fixture.media_playlist_uri, None).await;
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(!String::from_utf8(response_body(response).await.to_vec()).unwrap().contains("#EXT-X-ENDLIST"));
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+}
+
+#[tokio::test]
+async fn terminal_notification_releases_first_ready_manifest_before_origin_refresh_finishes() {
+    delayed_terminal_response_before_refresh(shared::model::HlsStartupMode::FirstReady).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_releases_progressive_manifest_before_origin_refresh_finishes() {
+    delayed_terminal_response_before_refresh(shared::model::HlsStartupMode::Progressive).await;
+}
+
+#[tokio::test]
+async fn terminal_notification_releases_optimistic_manifest_before_origin_refresh_finishes() {
+    delayed_terminal_response_before_refresh(shared::model::HlsStartupMode::Optimistic).await;
+}
+
+async fn delayed_terminal_response_before_refresh(mode: shared::model::HlsStartupMode) {
+    let mut controlled = DelayedTerminalRefresh::new(mode).await;
+    let waiting = controlled.waiting_request().await;
+    let fixture = &controlled.fixture;
+    let manager = &fixture.app_state.hls.proxy;
+    let previous = manager
+        .access_lease_response_snapshot(&fixture.access_lease_id, &fixture.proxy_session_id, manager.now_ms())
+        .await
+        .unwrap()
+        .last_manifest_snapshot
+        .unwrap();
+    let work = Arc::clone(&fixture.session.read().await.activity.active_origin_work_count);
+    let progress_before = fixture.session.read().await.origin_control.progress_generation;
+    assert!(work.load(std::sync::atomic::Ordering::Acquire) > 0);
+    let io = fixture.session.read().await.origin_account_io_lease.clone().unwrap();
+    assert!(io.is_active_or_acquiring());
+    let starts = manager.metrics().snapshot().refresh_started;
+    let mut commits = fixture.session.read().await.subscribe_terminal_commits();
+    let reserve = controlled.reserve().await;
+    assert!(reserve > 0);
+    assert!(!waiting.is_finished());
+    let safety_started = std::time::Instant::now();
+    tokio::time::advance(std::time::Duration::from_millis(reserve)).await;
+    within_real_second(commits.changed()).await.unwrap();
+    let commit_at = manager.now_ms();
+    assert_eq!(controlled.reserve().await, 0);
+    let response_started = std::time::Instant::now();
+    let response = within_real_second(waiting).await.unwrap();
+    let response_delay_us = response_started.elapsed().as_micros();
+    assert!(
+        safety_started.elapsed() < std::time::Duration::from_secs(1),
+        "owner commit and HTTP response share one real second"
+    );
+    assert_eq!(manager.now_ms(), commit_at, "no virtual time after terminal commit");
+    assert_eq!(response.status(), StatusCode::OK);
+    assert!(!response.headers().contains_key(header::RETRY_AFTER));
+    assert!(fixture.session.read().await.origin_refresh.in_flight);
+    assert!(!*controlled.release.borrow());
+    let body = String::from_utf8(response_body(response).await.to_vec()).unwrap();
+    let last_live = previous.visible_segments.last().unwrap().uri.as_ref();
+    let media = body.lines().filter(|line| !line.starts_with('#') && !line.is_empty()).collect::<Vec<_>>();
+    let first_cvr = media.iter().position(|uri| uri.contains("/terminal/")).unwrap();
+    assert_eq!(media[first_cvr - 1], last_live);
+    assert!(body.contains("#EXT-X-DISCONTINUITY\n"));
+    assert!(body.ends_with("#EXT-X-ENDLIST\n"));
+    let cvr = get_response(Arc::clone(&fixture.app_state), media[first_cvr], None).await;
+    assert_eq!(cvr.status(), StatusCode::OK);
+    assert!(!response_body(cvr).await.is_empty());
+    let plan = super::terminal_tail::terminal_generation_for_lease(
+        &fixture.app_state,
+        &fixture.proxy_session_id,
+        &fixture.access_lease_id,
+    )
+    .await;
+    eprintln!("terminal_refresh_proof mode={mode:?} reserve_before_ms={reserve} commit_at_ms={commit_at} response_at_ms={} response_delay_us={response_delay_us} origin_in_flight={} origin_gate_released={} last_live_preserved={} cvr_status=200 generation={}",
+        manager.now_ms(), fixture.session.read().await.origin_refresh.in_flight, *controlled.release.borrow(), media[first_cvr - 1] == last_live, plan.0);
+    drop(commits);
+    assert_eq!(fixture.session.read().await.terminal_commit_observer_count_for_test(), 0);
+    controlled.release.send_replace(true);
+    within_real_second(async {
+        while fixture.session.read().await.origin_refresh.in_flight {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await;
+    assert_eq!(work.load(std::sync::atomic::Ordering::Acquire), 0);
+    assert!(!io.is_active_or_acquiring());
+    assert_eq!(fixture.session.read().await.origin_refresh.last_success_at_ms, Some(commit_at));
+    assert!(fixture.session.read().await.origin_control.progress_generation > progress_before);
+    assert_eq!(manager.metrics().snapshot().refresh_started, starts);
+    assert_eq!(
+        super::terminal_tail::terminal_generation_for_lease(
+            &fixture.app_state,
+            &fixture.proxy_session_id,
+            &fixture.access_lease_id
+        )
+        .await,
+        plan
+    );
+    let replay = get_response(Arc::clone(&fixture.app_state), &fixture.media_playlist_uri, None).await;
+    assert_eq!(response_body(replay).await, body.as_bytes());
+}
 
 #[tokio::test]
 async fn terminal_lease_manifest_is_inline_immutable_endlist_on_canonical_path() {

backend/app/src/api/endpoints/hls_terminal_response.rs

@@ -259,7 +259,7 @@ pub(super) async fn resolve_hls_terminal_manifest_state(
         .await;
         match hls_terminal_endpoint_action(resolution) {
             HlsTerminalEndpointAction::ServeLive => {
-                let snapshot_now_ms = current_time_millis();
+                let snapshot_now_ms = app_state.hls.proxy.now_ms();
                 let current =
                     current_hls_terminal_lease(app_state, proxy_session_id, access_lease_id, snapshot_now_ms).await?;
                 if let Some(response) = hls_terminal_playback_response(&current, proxy_session_id, access_lease_id) {
@@ -280,9 +280,13 @@ pub(super) async fn resolve_hls_terminal_manifest_state(
                 reevaluations = reevaluations.saturating_add(1);
             }
             HlsTerminalEndpointAction::ReloadTerminal => {
-                let reloaded =
-                    current_hls_terminal_lease(app_state, proxy_session_id, access_lease_id, current_time_millis())
-                        .await?;
+                let reloaded = current_hls_terminal_lease(
+                    app_state,
+                    proxy_session_id,
+                    access_lease_id,
+                    app_state.hls.proxy.now_ms(),
+                )
+                .await?;
                 return match hls_terminal_playback_response(&reloaded, proxy_session_id, access_lease_id) {
                     Some(response) => Err(Box::new(response)),
                     None => Err(Box::new(hls_terminal_failed_closed_response(
@@ -296,7 +300,7 @@ pub(super) async fn resolve_hls_terminal_manifest_state(
                         HlsTerminalFailedClosedReason::RuntimeUnavailable,
                     )));
                 }
-                evaluation_now_ms = current_time_millis();
+                evaluation_now_ms = app_state.hls.proxy.now_ms();
                 let reloaded =
                     current_hls_terminal_lease(app_state, proxy_session_id, access_lease_id, evaluation_now_ms).await?;
                 let reloaded_session = current_hls_terminal_session(app_state, proxy_session_id).await?;
@@ -313,5 +317,3 @@ pub(super) async fn resolve_hls_terminal_manifest_state(
         }
     }
 }
-
-use tuliprox_core::utils::current_time_millis;

backend/app/src/mcp_stdio.rs

@@ -16,7 +16,7 @@ const CODE_UPSTREAM_FAILURE: i64 = -32000;
 const CODE_UNAUTHORIZED: i64 = -32001;
 const CODE_INTERNAL_ERROR: i64 = -32603;
 const CONNECT_TIMEOUT: Duration = Duration::from_secs(10);
-const REQUEST_TIMEOUT: Duration = Duration::from_secs(300);
+const REQUEST_TIMEOUT: Duration = Duration::from_mins(5);
 
 struct BridgeError {
     code: i64,

backend/hls/src/availability/terminal_cutover.rs

@@ -306,7 +306,7 @@ pub(crate) async fn commit_terminal_tail_if_lease_reserve_requires_cutover_detai
         return HlsDetailedTerminalResolution::resolved(HlsTerminalResolution::LiveAllowed);
     }
     let evaluation = evaluate_lease_terminal_cutover(ctx, lease, manifest, &state, now_ms);
-    log::debug!("HLS terminal reserve decision: proxy_session={} lease={} start_at_ms={:?} position_known={} position_ms={} reserve_ms={} session_target_duration_ms={} snapshot_generation={} may_publish={}",
+    log::trace!("HLS terminal reserve decision: proxy_session={} lease={} start_at_ms={:?} position_known={} position_ms={} reserve_ms={} session_target_duration_ms={} snapshot_generation={} may_publish={}",
         crate::safe_proxy_session_id(proxy_session_id), crate::safe_hls_access_lease_id(&lease.lease_id),
         lease.playback_cursor.first_segment_started_at_ms, evaluation.reserve.playback_position_known,
         evaluation.reserve.conservative_playback_position_ms, evaluation.reserve.guaranteed_reserve_ms,
@@ -619,39 +619,22 @@ pub(super) async fn commit_ready_terminal_bundle(
         Ok(plan) => {
             if preparation.required_terminal_media_key == Some(plan.media_preparation_key()) {
                 let commit_now_ms = context.ctx.hls_proxy.terminal_commit_now_ms();
-                let generation = plan.generation.0;
-                let base_proxy_tail = plan.base_manifest.last_proxy_seq;
-                let outcome = context.ctx.hls_proxy.commit_access_lease_terminal_if_generation_matches(
-                    HlsTerminalCommitRequest {
-                        session: context.session,
-                        lease_id: context.lease_id,
-                        proxy_session_id: context.proxy_session_id,
-                        preparation: &preparation,
-                        now_ms: commit_now_ms,
-                        payload: HlsTerminalCommitPayload::Tail {
-                            plan: Arc::new(plan),
-                            media_guard: base_evidence.into_commit_guard(),
-                        },
-                        asset_revision_guard: terminal_asset_revision_guard(
-                            context.ctx,
-                            expected_asset.reason,
-                            Some(expected_asset),
-                        ),
+                context.ctx.hls_proxy.commit_access_lease_terminal_if_generation_matches(HlsTerminalCommitRequest {
+                    session: context.session,
+                    lease_id: context.lease_id,
+                    proxy_session_id: context.proxy_session_id,
+                    preparation: &preparation,
+                    now_ms: commit_now_ms,
+                    payload: HlsTerminalCommitPayload::Tail {
+                        plan: Arc::new(plan),
+                        media_guard: base_evidence.into_commit_guard(),
                     },
-                );
-                if outcome == HlsTerminalCommitOutcome::Committed {
-                    debug!(
-                        "HLS runtime custom tail committed: proxy_session={} lease={} reason={} \
-                         generation={} base_proxy_tail={} asset_revision={:016x}",
-                        crate::safe_proxy_session_id(context.proxy_session_id),
-                        crate::safe_hls_access_lease_id(context.lease_id),
-                        expected_asset.reason.as_label(),
-                        generation,
-                        base_proxy_tail,
-                        expected_asset.media.revision
-                    );
-                }
-                outcome
+                    asset_revision_guard: terminal_asset_revision_guard(
+                        context.ctx,
+                        expected_asset.reason,
+                        Some(expected_asset),
+                    ),
+                })
             } else {
                 commit_terminal_unavailable(
                     context,

backend/hls/src/manager/publication.rs

@@ -300,10 +300,8 @@ impl HlsProxyManager {
     pub fn now_ms(&self) -> u64 { self.terminal_commit_clock.now_ms() }
 
     /// Anchor the scheduling clock to tokio's, for `start_paused` tests.
-    #[cfg(test)]
-    pub(crate) fn follow_tokio_clock_for_test(&self, base_ms: u64) {
-        self.terminal_commit_clock.follow_tokio_clock(base_ms);
-    }
+    #[cfg(any(test, feature = "test-support"))]
+    pub fn follow_tokio_clock_for_test(&self, base_ms: u64) { self.terminal_commit_clock.follow_tokio_clock(base_ms); }
 
     pub fn transient_resources(&self) -> &Arc<TransientResourceStore> { &self.transient_resources }
 

backend/hls/src/manager/terminal.rs

@@ -724,7 +724,12 @@ impl HlsProxyManager {
             return outcome;
         }
         session.live_window_demands.remove(lease_id);
-        Self::finish_terminal_commit(leases, session, proxy_session_id, preparation);
+        let tail = if let (HlsTerminalLeaseDecision::Tail(plan), false) = (decision, protection_capacity_exceeded) {
+            Some(plan.as_ref())
+        } else {
+            None
+        };
+        Self::finish_terminal_commit(leases, session, command, tail);
         HlsTerminalCommitOutcome::Committed
     }
 
@@ -759,16 +764,18 @@ impl HlsProxyManager {
             return outcome;
         }
         session.live_window_demands.remove(lease_id);
-        Self::finish_terminal_commit(leases, session, proxy_session_id, &command.preparation);
+        Self::finish_terminal_commit(leases, session, command, None);
         HlsTerminalCommitOutcome::Committed
     }
 
     fn finish_terminal_commit(
         leases: &mut HlsAccessLeaseStore,
         session: &mut super::super::HlsSession,
-        proxy_session_id: &ProxySessionId,
-        preparation: &crate::lease::HlsTerminalTailPreparation,
+        command: &HlsTerminalCommitCommand,
+        tail: Option<&crate::terminal_tail::HlsTerminalTailPlan>,
     ) {
+        let proxy_session_id = &command.key.proxy_session_id;
+        let preparation = &command.preparation;
         if !preparation.trigger.is_runtime_policy() && preparation.reserve.is_exhausted() {
             session.require_live_entry_revalidation(preparation.origin_progress_generation);
         }
@@ -778,6 +785,23 @@ impl HlsProxyManager {
         } else {
             super::super::origin_progress::HlsOriginProgressPhase::TerminalPartial
         };
+        if let Some(plan) = tail {
+            log::debug!(
+                "HLS runtime custom tail committed: proxy_session={} lease={} reason={} \
+                 generation={} base_proxy_tail={} asset_revision={:016x} \
+                 position_ms={} reserve_ms={} snapshot_generation={}",
+                crate::safe_proxy_session_id(proxy_session_id),
+                crate::safe_hls_access_lease_id(&command.key.lease_id),
+                preparation.trigger.reason().as_label(),
+                plan.generation.0,
+                plan.base_manifest.last_proxy_seq,
+                plan.media_preparation_key().asset.revision,
+                preparation.reserve.conservative_playback_position_ms,
+                preparation.reserve.guaranteed_reserve_ms,
+                preparation.manifest_snapshot_generation
+            );
+        }
+        session.notify_terminal_commit();
     }
 
     fn schedule_terminal_commit_retry(

backend/hls/src/manager/tests/terminal.rs

@@ -1214,6 +1214,7 @@ async fn hls_terminal_commit_protection_capacity_has_typed_unavailable_outcome()
         ))
         .await
         .expect("terminal preparation");
+    let commits = session.read().await.subscribe_terminal_commits();
 
     assert_eq!(
         commit_terminal_plan(
@@ -1239,6 +1240,7 @@ async fn hls_terminal_commit_protection_capacity_has_typed_unavailable_outcome()
             ..
         }
     ));
+    assert!(commits.has_changed().unwrap(), "TerminalUnavailable is a completed state change too");
     let session = session.read().await;
     assert_eq!(session.terminal_tail_protection_count(), HLS_TERMINAL_TAIL_PROTECTION_CAPACITY);
     assert!(session.terminal_tail_protection(&lease_id).is_none());
@@ -1276,6 +1278,8 @@ async fn unavailable_after_expired_splice_deadline_still_rejects_recovered_origi
 #[tokio::test]
 async fn final_terminal_authorization_rejects_every_positive_reserve_and_unknown_zero() {
     let (manager, session, proxy, lease, preparation, plan) = prepared_terminal_commit_fixture("final-zero-gate").await;
+    let mut commits = session.read().await.subscribe_terminal_commits();
+    assert!(!commits.has_changed().unwrap(), "preparation is not a terminal commit");
     let margin = preparation.reserve.transition_margin.as_millis();
     let acquisition = crate::recovery_timing::HlsTerminalCommitAcquisitionBudgetMs::from_retry_policy().as_millis();
     for (reserve_ms, known) in [(margin + acquisition, true), (margin, true), (1, true), (0, false)] {
@@ -1290,9 +1294,15 @@ async fn final_terminal_authorization_rejects_every_positive_reserve_and_unknown
             manager.access_lease_response_snapshot(&lease, &proxy, 2_000).await.unwrap().playback_mode,
             HlsLeasePlaybackMode::Live
         );
+        assert!(!commits.has_changed().unwrap(), "rejected commits must not wake requests");
     }
     assert_eq!(
         commit_terminal_plan(&manager, &session, &lease, &proxy, &preparation, 2_000, plan),
         HlsTerminalCommitOutcome::Committed
     );
+    assert!(commits.has_changed().unwrap());
+    commits.borrow_and_update();
+    assert!(!commits.has_changed().unwrap());
+    drop(commits);
+    assert_eq!(session.read().await.terminal_commit_observer_count_for_test(), 0);
 }

backend/hls/src/session.rs

@@ -26,7 +26,7 @@ use std::{
     fmt,
     sync::Arc,
 };
-use tokio::sync::Notify;
+use tokio::sync::{watch, Notify};
 
 /// Session mode selected for a shared HLS content session.
 #[derive(Debug, Clone, PartialEq, Eq)]
@@ -349,6 +349,7 @@ pub struct HlsSession {
     fresh_manifest_commit_requirement_generation: u64,
     pub segment_failure_tracker: HlsSegmentFailureTracker,
     terminal_tail_protections: HashMap<HlsAccessLeaseId, HlsTerminalTailProtection>,
+    terminal_commit_tx: watch::Sender<()>,
     gc_marked_for_removal: bool,
 }
 
@@ -437,10 +438,19 @@ impl HlsSession {
             fresh_manifest_commit_requirement_generation: 0,
             segment_failure_tracker: HlsSegmentFailureTracker::default(),
             terminal_tail_protections: HashMap::new(),
+            terminal_commit_tx: watch::channel(()).0,
             gc_marked_for_removal: false,
         }
     }
 
+    /// Observes successful terminal commits; callers must revalidate their own lease.
+    pub fn subscribe_terminal_commits(&self) -> watch::Receiver<()> { self.terminal_commit_tx.subscribe() }
+
+    #[cfg(any(test, feature = "test-support"))]
+    pub fn terminal_commit_observer_count_for_test(&self) -> usize { self.terminal_commit_tx.receiver_count() }
+
+    pub(crate) fn notify_terminal_commit(&self) { self.terminal_commit_tx.send_replace(()); }
+
     pub fn next_manifest_commit_identity(&self, rendered_at_ms: u64) -> Option<HlsManifestCommitIdentity> {
         self.manifest_commit_generation
             .checked_add(1)
@@ -1041,6 +1051,7 @@ impl fmt::Debug for HlsSession {
             .field("fresh_manifest_commit_requirement_generation", &self.fresh_manifest_commit_requirement_generation)
             .field("segment_failure_tracker", &self.segment_failure_tracker)
             .field("terminal_tail_protections_len", &self.terminal_tail_protections.len())
+            .field("terminal_commit_tx", &self.terminal_commit_tx)
             .field("gc_marked_for_removal", &self.gc_marked_for_removal)
             .finish()
     }

backend/hls/src/terminal_commit.rs

@@ -38,7 +38,7 @@ pub(super) struct HlsTerminalCommitClock {
     /// timers and leaves the wall clock behind, so a loop that sleeps until a
     /// wall-clock deadline never reaches it and spins until its budget is gone.
     /// Anchoring both to tokio's clock makes such a test deterministic.
-    #[cfg(test)]
+    #[cfg(any(test, feature = "test-support"))]
     virtual_origin: std::sync::Mutex<Option<(u64, tokio::time::Instant)>>,
 }
 
@@ -46,7 +46,7 @@ impl Default for HlsTerminalCommitClock {
     fn default() -> Self {
         Self {
             fixed_now_ms: AtomicU64::new(HLS_TERMINAL_COMMIT_LIVE_CLOCK),
-            #[cfg(test)]
+            #[cfg(any(test, feature = "test-support"))]
             virtual_origin: std::sync::Mutex::new(None),
         }
     }
@@ -54,7 +54,7 @@ impl Default for HlsTerminalCommitClock {
 
 impl HlsTerminalCommitClock {
     pub(super) fn now_ms(&self) -> u64 {
-        #[cfg(test)]
+        #[cfg(any(test, feature = "test-support"))]
         if let Some((base_ms, origin)) = *self.virtual_origin.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
         {
             let elapsed = tokio::time::Instant::now().saturating_duration_since(origin).as_millis();
@@ -69,7 +69,7 @@ impl HlsTerminalCommitClock {
     }
 
     /// Anchor this clock to tokio's, so `tokio::time::advance` moves it too.
-    #[cfg(test)]
+    #[cfg(any(test, feature = "test-support"))]
     pub(super) fn follow_tokio_clock(&self, base_ms: u64) {
         *self.virtual_origin.lock().unwrap_or_else(std::sync::PoisonError::into_inner) =
             Some((base_ms, tokio::time::Instant::now()));

shared/Cargo.toml

@@ -47,3 +47,4 @@ js-sys = "0.3.104"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 getrandom = "0.4.3"
+pest = { workspace = true, features = ["std"] }