diff --git a/backend/app/src/api/endpoints/hls_api/manifest.rs b/backend/app/src/api/endpoints/hls_api/manifest.rs index 602d58731..99963cf62 100644 --- a/backend/app/src/api/endpoints/hls_api/manifest.rs +++ b/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.")] diff --git a/backend/app/src/api/endpoints/hls_api/manifest/canonical_owner.rs b/backend/app/src/api/endpoints/hls_api/manifest/canonical_owner.rs index 4f9277013..280a22b0b 100644 --- a/backend/app/src/api/endpoints/hls_api/manifest/canonical_owner.rs +++ b/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, + session: &HlsSessionHandle, + proxy_session_id: &ProxySessionId, + access_lease_id: &HlsAccessLeaseId, + expected: &HlsAccessLease, +) -> Result> { + 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, ¤t_session) + || sessions.session_incarnation(session).is_none() + || sessions.session_incarnation(session) != sessions.session_incarnation(¤t_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, ¤t)) { + return Err(unavailable()); + } + Ok(lease) +} + +pub(in crate::api::endpoints::hls_api) async fn await_hls_terminal_or_refresh( + app_state: &Arc, + session: &HlsSessionHandle, + expected: &HlsAccessLease, + mut lease: HlsAccessLease, + mut commits: tokio::sync::watch::Receiver<()>, + refresh: impl std::future::Future, +) -> Option { + 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, 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. diff --git a/backend/app/src/api/endpoints/hls_api/tests/runtime_policy.rs b/backend/app/src/api/endpoints/hls_api/tests/runtime_policy.rs index b89b53bf9..fc98a8733 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/runtime_policy.rs +++ b/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 diff --git a/backend/app/src/api/endpoints/hls_api/tests/support/origins.rs b/backend/app/src/api/endpoints/hls_api/tests/support/origins.rs index bed96a471..e7deff7d7 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/support/origins.rs +++ b/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, pub(in crate::api::endpoints::hls_api::tests) body: Arc<[u8]>, + pub(in crate::api::endpoints::hls_api::tests) gate: Option>, } 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!( diff --git a/backend/app/src/api/endpoints/hls_api/tests/terminal_tail.rs b/backend/app/src/api/endpoints/hls_api/tests/terminal_tail.rs index 3f85ed718..4865c0a40 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/terminal_tail.rs +++ b/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, pub(in crate::api::endpoints::hls_api::tests) app_state: Arc, @@ -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, +) -> 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, diff --git a/backend/app/src/api/endpoints/hls_api/tests/terminal_tail_terminal_cutover.rs b/backend/app/src/api/endpoints/hls_api/tests/terminal_tail_terminal_cutover.rs index 27a79e02a..893c830a8 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/terminal_tail_terminal_cutover.rs +++ b/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(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, + blocked: tokio::sync::watch::Receiver, +} + +type PendingRefresh = futures::future::Shared>; + +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 { + 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::>(); + 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() { diff --git a/backend/app/src/api/endpoints/hls_terminal_response.rs b/backend/app/src/api/endpoints/hls_terminal_response.rs index a53838a7e..f8017eabb 100644 --- a/backend/app/src/api/endpoints/hls_terminal_response.rs +++ b/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(¤t, 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; diff --git a/backend/app/src/mcp_stdio.rs b/backend/app/src/mcp_stdio.rs index 87e6b0bc8..f228141f7 100644 --- a/backend/app/src/mcp_stdio.rs +++ b/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, diff --git a/backend/hls/src/availability/terminal_cutover.rs b/backend/hls/src/availability/terminal_cutover.rs index 807196a4a..0f8bd5ca4 100644 --- a/backend/hls/src/availability/terminal_cutover.rs +++ b/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, diff --git a/backend/hls/src/manager/publication.rs b/backend/hls/src/manager/publication.rs index 18e26eb14..cb58957f9 100644 --- a/backend/hls/src/manager/publication.rs +++ b/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 { &self.transient_resources } diff --git a/backend/hls/src/manager/terminal.rs b/backend/hls/src/manager/terminal.rs index 6952559a2..f73f84cda 100644 --- a/backend/hls/src/manager/terminal.rs +++ b/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( diff --git a/backend/hls/src/manager/tests/terminal.rs b/backend/hls/src/manager/tests/terminal.rs index d18ffbc12..30fcaacde 100644 --- a/backend/hls/src/manager/tests/terminal.rs +++ b/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); } diff --git a/backend/hls/src/session.rs b/backend/hls/src/session.rs index 54dd587bc..04d157ad1 100644 --- a/backend/hls/src/session.rs +++ b/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, + 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 { 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() } diff --git a/backend/hls/src/terminal_commit.rs b/backend/hls/src/terminal_commit.rs index 7dfef779f..3ab53bdfc 100644 --- a/backend/hls/src/terminal_commit.rs +++ b/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>, } @@ -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())); diff --git a/shared/Cargo.toml b/shared/Cargo.toml index 93df683b6..cb971388d 100644 --- a/shared/Cargo.toml +++ b/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"] }