diff --git a/backend/app/src/api/endpoints/hls_api/manifest/cached.rs b/backend/app/src/api/endpoints/hls_api/manifest/cached.rs index 79366b04b..64bc0c748 100644 --- a/backend/app/src/api/endpoints/hls_api/manifest/cached.rs +++ b/backend/app/src/api/endpoints/hls_api/manifest/cached.rs @@ -26,6 +26,54 @@ use tuliprox_hls::api::{ ProxySessionId, }; +type HlsPreparedManifest = + (HlsMaterializedSharedManifest, HlsLeaseManifestSnapshot, HlsPublishedTransientResourceIds, u64); + +enum HlsLiveWindowPreparation { + Prepared(HlsPreparedManifest), + NoRenderableWindow { lease: Box, plan: tuliprox_hls::api::HlsLiveWindowPlan }, + Unavailable(&'static str), +} + +/// Retained bytes are never reconstructed from the successor's metadata. The +/// publication transaction revalidates this selection, including all bindings. +fn retained_live_window_response( + lease: &tuliprox_hls::api::HlsAccessLease, + options: HlsCachedManifestOptions, + now_ms: u64, +) -> Result { + if lease.state != HlsAccessLeaseState::Activated + || lease.playback_mode != tuliprox_hls::api::HlsLeasePlaybackMode::Live + { + return Err("lease_not_active_live"); + } + if options.requires_newer_manifest() { + return Err("fresh_commit_required"); + } + let previous = lease.last_manifest_snapshot.as_ref().ok_or("no_previous_snapshot")?; + published_live_window_response(previous, now_ms) +} + +fn published_live_window_response( + previous: &HlsLeaseManifestSnapshot, + now_ms: u64, +) -> Result { + if now_ms > previous.delivered_at_ms.saturating_add(previous.playlist_duration_ms) { + return Err("previous_window_expired"); + } + let published = previous + .startup_revisions + .as_ref() + .and_then(|publication| publication.window_plan.as_ref()) + .ok_or("previous_window_unavailable")?; + Ok(( + HlsMaterializedSharedManifest { body: published.body.to_string(), mode: "normal", initial_strip_outcome: None }, + previous.clone(), + HlsPublishedTransientResourceIds::default(), + now_ms, + )) +} + #[allow(clippy::too_many_arguments)] async fn prepare_live_window_response( app_state: &Arc, @@ -34,48 +82,49 @@ async fn prepare_live_window_response( fingerprint: Option<&crate::model::Fingerprint>, options: HlsCachedManifestOptions, started_at_ms: u64, -) -> Option<(HlsMaterializedSharedManifest, HlsLeaseManifestSnapshot, HlsPublishedTransientResourceIds, u64)> { +) -> HlsLiveWindowPreparation { + use tuliprox_hls::api::{HlsManifestRenderer, RenderError}; + let now_ms = current_time_millis(); - let lease = - app_state.hls.proxy.access_lease_response_snapshot(view.access_lease_id, view.proxy_session_id, now_ms).await?; + let Some(lease) = + app_state.hls.proxy.access_lease_response_snapshot(view.access_lease_id, view.proxy_session_id, now_ms).await + else { + return HlsLiveWindowPreparation::Unavailable("lease_unavailable"); + }; let (prepared, queued) = { let mut state = session.write().await; state.live_window_strip = app_state.hls.proxy.strip(); if !tuliprox_hls::api::hls_live_window_origin_allowed_for_request(&state, options, started_at_ms, now_ms) { - return None; + return HlsLiveWindowPreparation::Unavailable("origin_freshness_or_identity"); } // Keep ordinary forward session commits current when their own window // is ready. A lease's independent entry never depends on this attempt. let _ = state.render_and_store_manifest(now_ms); - let desired = state.plan_live_window(tuliprox_hls::api::HlsLiveWindowView::for_lease(&lease))?; + let Some(desired) = state.plan_live_window(tuliprox_hls::api::HlsLiveWindowView::for_lease(&lease)) else { + return HlsLiveWindowPreparation::Unavailable("no_window_plan"); + }; if let Some(previous) = lease .last_manifest_snapshot .as_ref() .filter(|previous| desired.visible.first().is_some_and(|first| *first < previous.first_proxy_seq)) { - if now_ms > previous.delivered_at_ms.saturating_add(previous.playlist_duration_ms) { - return None; - } - let published = previous.startup_revisions.as_ref()?.window_plan.as_ref()?; - return Some(( - HlsMaterializedSharedManifest { - body: published.body.to_string(), - mode: "normal", - initial_strip_outcome: None, - }, - previous.clone(), - HlsPublishedTransientResourceIds::default(), - now_ms, - )); + return match published_live_window_response(previous, now_ms) { + Ok(prepared) => HlsLiveWindowPreparation::Prepared(prepared), + Err(reason) => HlsLiveWindowPreparation::Unavailable(reason), + }; } - let plan = state.plan_live_window_for_lease(&lease)?; + let Some(plan) = state.plan_live_window_for_lease(&lease) else { + return HlsLiveWindowPreparation::Unavailable("no_lease_window_plan"); + }; let backpressure = app_state.hls.proxy.segment_worker_pool().classify_backpressure_for_session(&state); let report = state.queue_manifest_fetch_candidates(now_ms, backpressure.allows_prefetch()); let queued = report.render_window_queued + report.prefetch_queued > 0; - let prepared = tuliprox_hls::api::HlsManifestRenderer::render_live_window(&state, &plan, now_ms).ok().and_then( - |rendered| { - // An explicit planned view bypasses generic strip. PreserveFullManifest - // continues to mean the origin's protected manifest semantics only. + let prepared = match HlsManifestRenderer::render_live_window(&state, &plan, now_ms) { + Err(RenderError::NoRenderableWindow) => { + HlsLiveWindowPreparation::NoRenderableWindow { lease: Box::new(lease), plan } + } + Err(_) => HlsLiveWindowPreparation::Unavailable("render_invalid_state"), + Ok(rendered) => { let materialized = HlsMaterializedSharedManifest { body: super::materialize_hls_access_manifest( &rendered.body, @@ -95,7 +144,7 @@ async fn prepare_live_window_response( }, ), }; - let snapshot = observe_hls_lease_manifest_snapshot_derivation( + match observe_hls_lease_manifest_snapshot_derivation( app_state, view.proxy_session_id, view.access_lease_id, @@ -108,11 +157,17 @@ async fn prepare_live_window_response( }, now_ms, ), - ) - .ok()??; - Some((materialized, snapshot, HlsPublishedTransientResourceIds::default(), now_ms)) - }, - ); + ) { + Ok(Some(snapshot)) => HlsLiveWindowPreparation::Prepared(( + materialized, + snapshot, + HlsPublishedTransientResourceIds::default(), + now_ms, + )), + _ => HlsLiveWindowPreparation::Unavailable("snapshot_derivation_failed"), + } + } + }; (prepared, queued) }; if queued { @@ -366,6 +421,11 @@ pub(in crate::api::endpoints::hls_api) async fn try_hls_cached_manifest_response .prepare_access_lease_manifest_publication(access_lease_id, &proxy_session_id, started_at_ms) .await else { + debug!( + "HLS live window unavailable: proxy_session={} lease={} status=503 reason=publication_guard_unavailable", + safe_proxy_session_id(&proxy_session_id), + safe_hls_access_lease_id(access_lease_id), + ); return Some(hls_cached_manifest_temporarily_unavailable()); }; let view = HlsCachedManifestViewContext::new( @@ -385,17 +445,35 @@ pub(in crate::api::endpoints::hls_api) async fn try_hls_cached_manifest_response .await; } if fast_start || !cached.wait_for_initial_commit { + let mut retained_plan = None; let prepared = if fast_start { - let prepared = + let preparation = prepare_live_window_response(app_state, session, &view, fingerprint, options, started_at_ms).await; - if prepared.is_none() { - if wait_for_hls_startup_evidence(started_at, options.wait_timeout).await { - continue; + let selection = match preparation { + HlsLiveWindowPreparation::Prepared(prepared) => Ok(prepared), + HlsLiveWindowPreparation::NoRenderableWindow { lease, plan } => { + retained_plan = Some(plan); + retained_live_window_response(&lease, options, current_time_millis()) + } + HlsLiveWindowPreparation::Unavailable(reason) => Err(reason), + }; + match selection { + Ok(prepared) => Some(prepared), + Err(reason) => { + if wait_for_hls_startup_evidence(started_at, options.wait_timeout).await { + continue; + } + if access_lease_state == HlsAccessLeaseState::Pending { + return None; + } + debug!( + "HLS live window unavailable: proxy_session={} lease={} status=503 reason={reason}", + safe_proxy_session_id(&proxy_session_id), + safe_hls_access_lease_id(access_lease_id), + ); + return Some(hls_cached_manifest_temporarily_unavailable()); } - return (access_lease_state != HlsAccessLeaseState::Pending) - .then(hls_cached_manifest_temporarily_unavailable); } - prepared } else if let Some(transient) = cached.transient_body { let materialized = view.materialize(&transient.body, "transient", transient.window_policy); let published_resource_ids = if transient.window_policy.preserves_full_manifest() { @@ -497,24 +575,61 @@ pub(in crate::api::endpoints::hls_api) async fn try_hls_cached_manifest_response } let startup_snapshot = snapshot.clone(); let admission_at_ms = current_time_millis(); - let outcome = app_state - .hls - .proxy - .commit_access_lease_manifest_publication_with_resources( - access_lease_id, - &proxy_session_id, - publication_guard, - snapshot, - published_resource_ids, - admission_at_ms, - ) - .await; + let outcome = if retained_plan.is_some() { + app_state + .hls + .proxy + .republish_access_lease_live_window( + session, + access_lease_id, + &proxy_session_id, + publication_guard, + snapshot, + options, + started_at_ms, + admission_at_ms, + ) + .await + } else { + app_state + .hls + .proxy + .commit_access_lease_manifest_publication_with_resources( + access_lease_id, + &proxy_session_id, + publication_guard, + snapshot, + published_resource_ids, + admission_at_ms, + ) + .await + }; if fast_start && !outcome.is_committed() { + if outcome + == tuliprox_hls::api::HlsLeaseManifestPublicationOutcome::Rejected( + tuliprox_hls::api::HlsLeaseManifestPublicationRejectReason::SnapshotChanged, + ) + { + continue; + } if wait_for_hls_startup_evidence(started_at, options.wait_timeout).await { continue; } + debug!( + "HLS live window unavailable: proxy_session={} lease={} status=503 reason=publication_rejected outcome={outcome:?}", + safe_proxy_session_id(&proxy_session_id), safe_hls_access_lease_id(access_lease_id), + ); return Some(hls_cached_manifest_temporarily_unavailable()); } + if let Some(plan) = retained_plan { + debug!( + "HLS live window retained: proxy_session={} lease={} reason=no_renderable_window status=200 previous_first={} previous_last={} desired_first={:?} desired_last={:?} remaining_validity_ms={}", + safe_proxy_session_id(&proxy_session_id), safe_hls_access_lease_id(access_lease_id), + startup_snapshot.first_proxy_seq, startup_snapshot.last_proxy_seq, + plan.visible.first(), plan.visible.last(), + startup_snapshot.delivered_at_ms.saturating_add(startup_snapshot.playlist_duration_ms).saturating_sub(current_time_millis()), + ); + } if let Some(snapshot_generation) = outcome.snapshot_generation() { let published_at_ms = current_time_millis(); let first_startup_publication = diff --git a/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window.rs b/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window.rs index b583118dd..46dfdc3bf 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window.rs +++ b/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window.rs @@ -1,3 +1,5 @@ +mod retained; + use super::{normal_manifest, response_body, test_app_state_with_hls_proxy, try_test_hls_cached_manifest_response}; use crate::api::model::{HlsAccessLeaseId, HlsAccessLeaseState, HlsProxyManager, HlsSessionKey, SegmentCacheStatus}; use axum::http::StatusCode; diff --git a/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window/retained.rs b/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window/retained.rs new file mode 100644 index 000000000..218d389a3 --- /dev/null +++ b/backend/app/src/api/endpoints/hls_api/tests/manifest_live_window/retained.rs @@ -0,0 +1,580 @@ +use super::{ + origin, prepare_revisions, response_body, test_app_state_with_hls_proxy, try_test_hls_cached_manifest_response, +}; +use crate::api::model::AppState; +use axum::http::StatusCode; +use shared::model::{HlsCacheConfigDto, HlsStartupMode, HlsStripMode}; +use std::{sync::Arc, time::Duration}; +use tuliprox_core::{ + model::{HlsCacheConfig, StripConfig}, + utils::current_time_millis, +}; +use tuliprox_hls::api::{ + HlsAccessLease, HlsAccessLeaseId, HlsAccessLeaseState, HlsCachedManifestOptions, + HlsLeaseManifestPublicationOutcome as Outcome, HlsLeaseManifestPublicationRejectReason as Reject, + HlsLeaseManifestSnapshot, HlsProxyManager, HlsSessionHandle, HlsSessionKey, ProxySessionId, +}; + +const MODES: [HlsStartupMode; 3] = + [HlsStartupMode::FirstReady, HlsStartupMode::Progressive, HlsStartupMode::Optimistic]; + +struct Fixture { + _directory: tempfile::TempDir, + app: Arc, + manager: Arc, + session: HlsSessionHandle, + proxy: ProxySessionId, + lease_id: HlsAccessLeaseId, + previous: HlsLeaseManifestSnapshot, + body: bytes::Bytes, +} + +impl Fixture { + async fn new(mode: HlsStartupMode) -> Self { + let directory = tempfile::tempdir().unwrap(); + let mut dto = HlsCacheConfigDto { + cache_path: Some(directory.path().to_string_lossy().into_owned()), + max_segments_prefetch: 0, + ..Default::default() + }; + dto.startup.mode = mode; + dto.strip.value = 0; + let manager = Arc::new(HlsProxyManager::with_hls_cache_config(&HlsCacheConfig::from(&dto))); + let app = test_app_state_with_hls_proxy(manager.clone()); + let session = + manager.get_or_create_session(HlsSessionKey::new(1, "12345"), b"secret", current_time_millis()).await; + let proxy = session.read().await.proxy_session_id.clone(); + { + let mut state = session.write().await; + state.apply_origin_manifest(&origin(0, 6, &[4])).unwrap(); + prepare_revisions(&mut state, &[3, 4, 5]); + // Real revision files make GC and media requests exercise the same + // immutable objects that this lease publishes. + for seq in [3, 4, 5] { + let old = state.startup.as_mut().unwrap().revisions.remove(&seq).unwrap(); + let revision = + state.startup.as_ref().unwrap().store.create(proxy.clone(), seq, old.revision().key.kind).unwrap(); + let metadata = manager + .segment_cache() + .write_bytes_and_commit(&revision.revision().key, format!("bound-{seq}").as_bytes()) + .await + .unwrap(); + revision.revision().complete(metadata); + state.startup.as_mut().unwrap().revisions.insert(seq, revision); + } + state.render_and_store_manifest(current_time_millis()).unwrap(); + } + let lease_id = HlsAccessLeaseId(super::super::support::grant_hls_proxy_lease(&app, &proxy.0).await); + let response = try_test_hls_cached_manifest_response( + &app, + &session, + &lease_id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let body = response_body(response).await; + let mut lease = manager.access_lease_response_snapshot(&lease_id, &proxy, current_time_millis()).await.unwrap(); + lease.active_until_ms = Some(current_time_millis() + 120_000); + let previous = lease.last_manifest_snapshot.as_ref().unwrap().clone(); + manager.prepare_access_lease(lease).await; + Self { _directory: directory, app, manager, session, proxy, lease_id, previous, body } + } + + async fn advance(&self) { self.session.write().await.apply_origin_manifest(&origin(6, 6, &[4])).unwrap(); } + + async fn request(&self, options: HlsCachedManifestOptions) -> axum::response::Response { + try_test_hls_cached_manifest_response( + &self.app, + &self.session, + &self.lease_id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + options, + ) + .await + .unwrap() + } + + async fn lease(&self) -> HlsAccessLease { + self.manager.access_lease_response_snapshot(&self.lease_id, &self.proxy, current_time_millis()).await.unwrap() + } + + async fn republish_at(&self, now_ms: u64, snapshot: HlsLeaseManifestSnapshot) -> Outcome { + let guard = + self.manager.prepare_access_lease_manifest_publication(&self.lease_id, &self.proxy, now_ms).await.unwrap(); + self.manager + .republish_access_lease_live_window( + &self.session, + &self.lease_id, + &self.proxy, + guard, + snapshot, + HlsCachedManifestOptions::initial(Duration::ZERO), + now_ms, + now_ms, + ) + .await + } +} + +#[tokio::test] +async fn live_window_retains_twenty_zero_wait_polls_until_the_successor_gate_opens() { + for mode in MODES { + let fixture = Fixture::new(mode).await; + fixture.advance().await; + let (release, gate) = tokio::sync::oneshot::channel(); + let session = fixture.session.clone(); + let successor = tokio::spawn(async move { + gate.await.unwrap(); + prepare_revisions(&mut *session.write().await, &[6, 7, 8, 9, 10, 11]); + }); + for _ in 0..20 { + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::OK, "{mode}"); + assert!(!response.headers().contains_key(axum::http::header::RETRY_AFTER)); + assert_eq!(response_body(response).await, fixture.body); + let published = fixture.lease().await.last_manifest_snapshot.unwrap(); + assert_eq!(published.delivered_at_ms, fixture.previous.delivered_at_ms); + assert_eq!(published.source_commit_identity, fixture.previous.source_commit_identity); + assert_eq!(published.last_visible_media_end_ms, fixture.previous.last_visible_media_end_ms); + assert_eq!(published.startup_revisions, fixture.previous.startup_revisions); + assert!(published.snapshot_generation > fixture.previous.snapshot_generation); + } + assert!(!successor.is_finished()); + assert!(!fixture.session.read().await.segment_prefetch_queue.is_empty(), "successor work stays queued"); + release.send(()).unwrap(); + successor.await.unwrap(); + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::OK); + assert_ne!(response_body(response).await, fixture.body); + let next = fixture.lease().await.last_manifest_snapshot.unwrap(); + assert_eq!(next.first_proxy_seq, 6); + assert!(next.first_proxy_seq > fixture.previous.first_proxy_seq); + assert!(next.discontinuity_sequence >= fixture.previous.discontinuity_sequence); + } +} + +#[tokio::test] +async fn live_window_retained_deadline_is_inclusive_and_never_slides() { + for mode in MODES { + let fixture = Fixture::new(mode).await; + fixture.advance().await; + let deadline = fixture.previous.delivered_at_ms + fixture.previous.playlist_duration_ms; + assert!(fixture.republish_at(deadline, fixture.previous.clone()).await.is_committed(), "{mode}"); + let current = fixture.lease().await.last_manifest_snapshot.unwrap(); + assert_eq!(current.delivered_at_ms, fixture.previous.delivered_at_ms); + let result = fixture.republish_at(deadline + 1, current).await; + assert_eq!(result, Outcome::Rejected(Reject::ManifestGenerationUnavailable)); + } +} + +#[tokio::test] +async fn live_window_failed_successor_does_not_poison_the_bound_predecessor() { + for mode in MODES { + let fixture = Fixture::new(mode).await; + fixture.advance().await; + { + let mut state = fixture.session.write().await; + prepare_revisions(&mut state, &[7, 8, 9, 10, 11]); + let revision = state + .startup + .as_ref() + .unwrap() + .store + .create(fixture.proxy.clone(), 6, tuliprox_hls::SegmentRevisionKind::Raw) + .unwrap(); + revision.revision().fail(); + state.startup.as_mut().unwrap().revisions.insert(6, revision); + } + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response_body(response).await, fixture.body); + } +} + +#[tokio::test] +async fn live_window_retained_publication_revalidates_concurrent_state_changes() { + for change in ["newer", "terminal", "lease", "admission", "session", "origin", "missing_revision", "fresh_commit"] { + let fixture = Fixture::new(HlsStartupMode::Optimistic).await; + fixture.advance().await; + let now = current_time_millis(); + let guard = fixture + .manager + .prepare_access_lease_manifest_publication(&fixture.lease_id, &fixture.proxy, now) + .await + .unwrap(); + match change { + "newer" => { + prepare_revisions(&mut *fixture.session.write().await, &[6, 7, 8, 9, 10, 11]); + assert_eq!( + fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await.status(), + StatusCode::OK + ); + } + "terminal" => { + let mut lease = fixture.lease().await; + lease.playback_mode = tuliprox_hls::api::HlsLeasePlaybackMode::TerminalUnavailable { + decision_generation: 1, + reason: tuliprox_hls::HlsTerminalTailCompatibility::MissingAsset, + }; + fixture.manager.prepare_access_lease(lease).await; + } + "lease" | "admission" => { + let mut lease = fixture.lease().await; + if change == "lease" { + lease.issued_at_ms += 1; + } else { + lease.admission_generation += 1; + } + fixture.manager.prepare_access_lease(lease).await; + } + "session" => { + let key = fixture.session.read().await.key.clone(); + fixture.manager.sessions().remove_session(&key, &fixture.proxy).await.unwrap(); + fixture.manager.get_or_create_session(key, b"secret", now).await; + } + "origin" => fixture.session.write().await.activity.origin_work_generation += 1, + "missing_revision" => { + fixture.manager.access_leases().write().await.prune_revision_bindings(&fixture.proxy, 4); + } + "fresh_commit" => { + fixture.session.write().await.fresh_manifest_commit_required = + Some(tuliprox_hls::api::HlsFreshManifestRequiredReason::PreviousHardManifestFailure); + } + _ => unreachable!(), + } + let before = fixture.lease().await; + let result = fixture + .manager + .republish_access_lease_live_window( + &fixture.session, + &fixture.lease_id, + &fixture.proxy, + guard, + fixture.previous.clone(), + HlsCachedManifestOptions::initial(Duration::ZERO), + now, + now, + ) + .await; + assert!(!result.is_committed(), "{change}: {result:?}"); + if change == "newer" { + assert_eq!(result, Outcome::Rejected(Reject::SnapshotChanged)); + } + assert_eq!(fixture.lease().await.last_manifest_snapshot, before.last_manifest_snapshot, "{change}"); + } +} + +#[tokio::test] +async fn live_window_retained_fallback_preserves_each_leases_body_and_freshness_requirements() { + let fixture = Fixture::new(HlsStartupMode::FirstReady).await; + let second_id = HlsAccessLeaseId("another-live-lease".into()); + let response = try_test_hls_cached_manifest_response( + &fixture.app, + &fixture.session, + &second_id, + HlsAccessLeaseState::Pending, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + let second_body = response_body(response).await; + let mut second = fixture + .manager + .access_lease_response_snapshot(&second_id, &fixture.proxy, current_time_millis()) + .await + .unwrap(); + second.state = HlsAccessLeaseState::Activated; + second.active_until_ms = Some(current_time_millis() + 60_000); + fixture.manager.prepare_access_lease(second).await; + fixture.advance().await; + assert_eq!( + response_body(fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await).await, + fixture.body + ); + let response = try_test_hls_cached_manifest_response( + &fixture.app, + &fixture.session, + &second_id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response_body(response).await, second_body); + assert_ne!(second_body, fixture.body); + let response = fixture + .request( + HlsCachedManifestOptions::initial(Duration::ZERO) + .requiring_newer_manifest(fixture.previous.delivered_at_ms), + ) + .await; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); +} + +#[tokio::test] +async fn live_window_gc_keeps_bound_bytes_during_fallback_and_releases_them_after_retention() { + let mut fixture = Fixture::new(HlsStartupMode::FirstReady).await; + fixture.advance().await; + let now = current_time_millis(); + let gc = fixture.manager.garbage_collector(); + let mut policy = gc.policy().as_ref().clone(); + policy.cache_duration_ms = 0; + gc.update_config(policy, gc.rewrite_secret_fingerprint()); + let mut paths = Vec::new(); + for _ in 0..3 { + assert_eq!(fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await.status(), StatusCode::OK); + gc.run_once(now + 1).await.unwrap(); + for segment in fixture.previous.visible_segments.iter() { + let response = super::super::support::get_response(fixture.app.clone(), &segment.uri, None).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response_body(response).await.as_ref(), format!("bound-{}", segment.proxy_seq).as_bytes()); + let revision = fixture + .manager + .access_leases() + .read() + .await + .published_segment_revision(&fixture.lease_id, segment.proxy_seq) + .unwrap(); + if let tuliprox_hls::SegmentRevisionState::Complete(metadata) = revision.revision().state() { + paths.push(metadata.path); + } + } + } + prepare_revisions(&mut *fixture.session.write().await, &[6, 7, 8, 9, 10, 11]); + assert_eq!(fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await.status(), StatusCode::OK); + fixture.previous.startup_revisions = None; // Release the test's own historical pins. + gc.run_once(now + 60_000).await.unwrap(); + gc.run_once(now + 60_001).await.unwrap(); + assert!(paths.iter().all(|path| !path.exists()), "obsolete bound revision files must become reclaimable"); +} + +#[tokio::test] +async fn live_window_retained_fallback_does_not_require_complete_successors_in_progressive_modes() { + for mode in [HlsStartupMode::Progressive, HlsStartupMode::Optimistic] { + let fixture = Fixture::new(mode).await; + fixture.advance().await; + { + let mut state = fixture.session.write().await; + prepare_revisions(&mut state, &[6]); + for seq in 7..=11 { + let revision = state + .startup + .as_ref() + .unwrap() + .store + .create(fixture.proxy.clone(), seq, tuliprox_hls::SegmentRevisionKind::Raw) + .unwrap(); + if mode == HlsStartupMode::Progressive { + revision.revision().prefix_available.store(1, std::sync::atomic::Ordering::Release); + } + state.startup.as_mut().unwrap().revisions.insert(seq, revision); + } + } + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::OK); + assert_ne!(response_body(response).await, fixture.body); + assert_eq!(fixture.lease().await.last_manifest_snapshot.unwrap().first_proxy_seq, 6); + assert!(!fixture.session.read().await.startup.as_ref().unwrap().revisions[&7].revision().is_complete()); + } +} + +#[tokio::test] +async fn live_window_unusable_bound_predecessor_and_missing_snapshot_keep_the_error_path() { + for unusable in [false, true] { + let fixture = Fixture::new(HlsStartupMode::Optimistic).await; + if unusable { + // Publish an optimistic pending successor, then fail that exact + // bound revision before the next origin window arrives. + { + let mut state = fixture.session.write().await; + let revision = state + .startup + .as_ref() + .unwrap() + .store + .create(fixture.proxy.clone(), 5, tuliprox_hls::SegmentRevisionKind::Raw) + .unwrap(); + state.startup.as_mut().unwrap().revisions.insert(5, revision); + } + // A complete binding cannot silently be replaced: use a new lease + // to publish this pending tail under the normal publication rules. + let id = HlsAccessLeaseId("pending-predecessor".into()); + let response = try_test_hls_cached_manifest_response( + &fixture.app, + &fixture.session, + &id, + HlsAccessLeaseState::Pending, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let mut lease = fixture + .manager + .access_lease_response_snapshot(&id, &fixture.proxy, current_time_millis()) + .await + .unwrap(); + lease.state = HlsAccessLeaseState::Activated; + lease.active_until_ms = Some(current_time_millis() + 60_000); + fixture.manager.prepare_access_lease(lease).await; + fixture.session.read().await.startup.as_ref().unwrap().revisions[&5].revision().fail(); + fixture.advance().await; + let response = try_test_hls_cached_manifest_response( + &fixture.app, + &fixture.session, + &id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + } else { + let mut lease = fixture.lease().await; + lease.last_manifest_snapshot = None; + fixture.manager.prepare_access_lease(lease).await; + fixture.advance().await; + assert_eq!( + fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await.status(), + StatusCode::SERVICE_UNAVAILABLE + ); + } + } +} + +#[tokio::test] +async fn live_window_retained_response_dispatches_successor_work_before_returning() { + let fixture = Fixture::new(HlsStartupMode::FirstReady).await; + fixture.advance().await; + let gate = fixture.manager.segment_worker_pool().pause_next_origin_prepare_for_test(); + let response = super::super::super::try_hls_cached_manifest_response( + &fixture.app, + Some(&super::super::test_fingerprint()), + &fixture.session, + &fixture.lease_id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + None, + HlsCachedManifestOptions::initial(Duration::ZERO), + super::super::super::HlsRuntimeBandwidthLearningContext::Disabled, + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response_body(response).await, fixture.body); + tokio::time::timeout(Duration::from_secs(1), gate.entered.notified()).await.expect("successor scheduler was woken"); + assert!(!fixture.session.read().await.segment_prefetch_queue.is_empty()); + fixture.manager.access_leases().write().await.remove_access_lease(&fixture.lease_id); + gate.prepare.notify_one(); + tokio::time::timeout(Duration::from_secs(1), gate.prepared.notified()).await.unwrap(); + gate.proceed.notify_one(); +} + +#[tokio::test] +async fn live_window_retained_response_rejects_expired_missing_bindings_and_inactive_leases() { + for reason in ["expired", "bindings", "idle", "pending", "origin_generation"] { + let fixture = Fixture::new(HlsStartupMode::Progressive).await; + fixture.advance().await; + match reason { + "expired" => { + let mut lease = fixture.lease().await; + let previous = lease.last_manifest_snapshot.as_mut().unwrap(); + previous.delivered_at_ms = current_time_millis() - previous.playlist_duration_ms - 1; + fixture.manager.prepare_access_lease(lease).await; + } + "bindings" => fixture.manager.access_leases().write().await.prune_revision_bindings(&fixture.proxy, 4), + "idle" | "pending" => { + let mut lease = fixture.lease().await; + lease.state = if reason == "idle" { HlsAccessLeaseState::Idle } else { HlsAccessLeaseState::Pending }; + fixture.manager.prepare_access_lease(lease).await; + } + "origin_generation" => fixture.session.write().await.activity.origin_work_generation += 1, + _ => unreachable!(), + } + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE, "{reason}"); + assert_eq!(response.headers()[axum::http::header::RETRY_AFTER], "1"); + assert!(response_body(response).await.is_empty()); + } +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn live_window_concurrent_retained_polls_reselect_without_overwriting_newer_snapshots() { + let fixture = Arc::new(Fixture::new(HlsStartupMode::Optimistic).await); + fixture.advance().await; + let barrier = Arc::new(tokio::sync::Barrier::new(20)); + let mut requests = tokio::task::JoinSet::new(); + for _ in 0..20 { + let fixture = fixture.clone(); + let barrier = barrier.clone(); + requests.spawn(async move { + barrier.wait().await; + let response = fixture.request(HlsCachedManifestOptions::initial(Duration::ZERO)).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response_body(response).await, fixture.body); + }); + } + while let Some(result) = requests.join_next().await { + result.unwrap(); + } + let current = fixture.lease().await.last_manifest_snapshot.unwrap(); + assert_eq!(current.snapshot_generation, fixture.previous.snapshot_generation + 20); + assert_eq!(current.delivered_at_ms, fixture.previous.delivered_at_ms); +} + +#[tokio::test] +async fn live_window_snapshot_derivation_failure_cannot_select_the_old_body() { + let fixture = Fixture::new(HlsStartupMode::FirstReady).await; + fixture.advance().await; + prepare_revisions(&mut *fixture.session.write().await, &[6, 7, 8, 9, 10, 11]); + // All segments render, but materialization exceeds the existing snapshot + // URI bound. This is a representation error, not a render readiness error. + let prefix = format!("/{}", "x".repeat(400_000)); + let response = try_test_hls_cached_manifest_response( + &fixture.app, + &fixture.session, + &fixture.lease_id, + HlsAccessLeaseState::Activated, + &StripConfig { mode: HlsStripMode::Segments, value: 0 }, + Some(&prefix), + HlsCachedManifestOptions::initial(Duration::ZERO), + ) + .await + .unwrap(); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(fixture.lease().await.last_manifest_snapshot.unwrap(), fixture.previous); +} + +#[tokio::test] +async fn live_window_retained_deadline_is_rechecked_after_lock_acquisition() { + let fixture = Fixture::new(HlsStartupMode::FirstReady).await; + fixture.advance().await; + let mut lease = fixture.lease().await; + let previous = lease.last_manifest_snapshot.as_mut().unwrap(); + previous.delivered_at_ms = current_time_millis() - previous.playlist_duration_ms - 1; + let previous = previous.clone(); + let stale_request_time = previous.delivered_at_ms + previous.playlist_duration_ms; + fixture.manager.prepare_access_lease(lease).await; + // The request's captured timestamp was still on the inclusive boundary; + // the publication transaction must use its current clock as well. + assert_eq!( + fixture.republish_at(stale_request_time, previous).await, + Outcome::Rejected(Reject::ManifestGenerationUnavailable), + ); +} diff --git a/backend/app/tests/hls_manifest_window.rs b/backend/app/tests/hls_manifest_window.rs new file mode 100644 index 000000000..bf5827978 --- /dev/null +++ b/backend/app/tests/hls_manifest_window.rs @@ -0,0 +1,199 @@ +//! Real process regression against the existing local Postman hls-window fixture. +//! +//! Start its simulation with a fixed port and an NDJSON log: +//! `/Users/philippmielke/.local/bin/postman simulation run postman/simulations/hls-window.sim.yaml --port 4926 --no-history --output ndjson` +//! +//! Run through `./bin/test.sh -p tuliprox --test hls_manifest_window -- --ignored`, +//! setting `TULIPROX_HLS_WINDOW_SIMULATION_ORIGIN=http://127.0.0.1:4926/origin` +//! and `TULIPROX_HLS_WINDOW_EVIDENCE_DIR` to an absolute output directory. +//! The member serves real HLS metadata; a loopback TS origin gates new bytes. +//! Only the URL-free client protocol is suitable for sharing; keep raw logs local. +use std::{ + io::Write as _, + process::Stdio, + sync::{ + atomic::{AtomicUsize, Ordering}, + Arc, + }, + time::Duration, +}; +use tokio::{ + io::{AsyncBufReadExt, AsyncWriteExt, BufReader}, + net::TcpListener, + sync::watch, +}; + +const BODY: &[u8] = + include_bytes!(concat!(env!("CARGO_MANIFEST_DIR"), "/../../test/fixtures/hls/channel_unavailable.ts")); + +fn media_url(base: &str, playlist: &str) -> String { + let uri = playlist.lines().find(|line| !line.is_empty() && !line.starts_with('#')).unwrap(); + reqwest::Url::parse(base).unwrap().join(uri).unwrap().to_string() +} + +fn sequence(body: &str, tag: &str) -> u64 { + body.lines().find_map(|line| line.strip_prefix(tag)).unwrap_or("0").parse().unwrap() +} + +#[tokio::test] +#[allow( + clippy::too_many_lines, + reason = "Keep the real process, streaming gate and HTTP assertions in one isolated scenario." +)] +#[ignore = "requires local Postman hls-window simulation; see module documentation"] +async fn live_window_http_republishes_same_lease_while_successor_is_gated() { + let simulation = std::env::var("TULIPROX_HLS_WINDOW_SIMULATION_ORIGIN").expect("Postman origin URL"); + let evidence = + std::path::PathBuf::from(std::env::var("TULIPROX_HLS_WINDOW_EVIDENCE_DIR").expect("evidence directory")); + std::fs::create_dir_all(&evidence).unwrap(); + let mut protocol = std::fs::File::create(evidence.join("client.ndjson")).unwrap(); + let client = reqwest::Client::builder().timeout(Duration::from_secs(10)).build().unwrap(); + for mode in ["first_ready", "progressive", "optimistic"] { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let segment_origin = format!("http://{}", listener.local_addr().unwrap()); + let configure = |first| { + client + .post(format!("{simulation}/configure")) + .json(&serde_json::json!({"firstSequence": first, "segmentOrigin": segment_origin})) + }; + assert_eq!(configure(0).send().await.unwrap().status(), 204); + let catalog = format!( + "#EXTM3U\n#EXTINF:-1 tvg-id=\"window\" group-title=\"Test\",Window Channel\n{simulation}/index.m3u8\n" + ); + let (release, released) = watch::channel(false); + let attempts = Arc::new(AtomicUsize::new(0)); + let counted = attempts.clone(); + let origin = tokio::spawn(async move { + let mut children = tokio::task::JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let (socket, _) = accepted.unwrap(); + let catalog = catalog.clone(); + let mut released = released.clone(); + let attempts = counted.clone(); + children.spawn(async move { + let mut reader = BufReader::new(socket); + let mut request = String::new(); reader.read_line(&mut request).await.unwrap(); + let path = request.split_whitespace().nth(1).unwrap().to_string(); + loop { + let mut line = String::new(); reader.read_line(&mut line).await.unwrap(); + if line == "\r\n" || line.is_empty() { break; } + } + let mut socket = reader.into_inner(); + let body = if path == "/catalog.m3u" { catalog.as_bytes() } else { BODY }; + let seq = path.strip_prefix('/').and_then(|path| path.strip_suffix(".ts")) + .and_then(|seq| seq.parse::().ok()); + if seq.is_some_and(|seq| seq >= 6) { + attempts.fetch_add(1, Ordering::AcqRel); + while !*released.borrow_and_update() { + if released.changed().await.is_err() { return; } + } + } + let headers = format!("HTTP/1.1 200 OK\r\nContent-Length: {}\r\nConnection: close\r\n\r\n", body.len()); + if socket.write_all(headers.as_bytes()).await.is_ok() { let _ = socket.write_all(body).await; } + }); + } + _ = children.join_next(), if !children.is_empty() => {} + } + } + }); + let directory = tempfile::tempdir().unwrap(); + let root = directory.path(); + for name in ["web", "storage", "backup"] { + std::fs::create_dir(root.join(name)).unwrap(); + } + let bound = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let port = bound.local_addr().unwrap().port(); + drop(bound); + let config = format!("api:\n host: 127.0.0.1\n port: {port}\n web_root: '{}'\nworking_dir: '{}'\nbackup_dir: '{}'\nupdate_on_boot: true\nweb_ui:\n enabled: false\nlog:\n log_level: debug\nreverse_proxy:\n rewrite_secret: 0123456789abcdef0123456789abcdef\n hls_cache:\n cache_path: '{}'\n strip:\n mode: segments\n value: 0\n origin_segment_timeout_ms: 10000\n max_concurrent_segment_fetches_per_session: 6\n initial_manifest_wait_timeout_secs: 3\n startup:\n mode: {mode}\n segment_repair:\n max_level: off\n", root.join("web").display(), root.join("storage").display(), root.join("backup").display(), root.join("cache").display()); + let sources = format!("inputs:\n - name: window-origin\n type: m3u\n url: {segment_origin}/catalog.m3u\n max_connections: 8\nsources:\n - inputs: [window-origin]\n targets:\n - name: window\n output:\n - type: m3u\n options:\n share_live_streams:\n hls: true\n"); + let users = format!("server:\n - name: local\n protocol: http\n host: 127.0.0.1\n port: '{port}'\n timezone: UTC\n message: Window test\nuser:\n - target: window\n credentials:\n - username: viewer\n password: test-password\n proxy: reverse\n server: local\n status: Active\n max_connections: 1\n"); + for (name, body) in [("config.yml", config), ("source.yml", sources), ("api-proxy.yml", users)] { + std::fs::write(root.join(name), body).unwrap(); + } + let log_file = evidence.join(format!("sut-{mode}.log")); + let log = std::fs::File::create(&log_file).unwrap(); + let mut process = tokio::process::Command::new(env!("CARGO_BIN_EXE_tuliprox")) + .args(["-s", "-c", "config.yml", "-i", "source.yml", "-a", "api-proxy.yml"]) + .current_dir(root) + .env("RUST_LOG", "debug") + .stdout(Stdio::from(log.try_clone().unwrap())) + .stderr(Stdio::from(log)) + .kill_on_drop(true) + .spawn() + .unwrap(); + let base = format!("http://127.0.0.1:{port}"); + let deadline = tokio::time::Instant::now() + Duration::from_secs(30); + let entry = loop { + assert!( + process.try_wait().unwrap().is_none(), + "SUT exited: {}", + std::fs::read_to_string(&log_file).unwrap() + ); + if let Ok(response) = client.get(format!("{base}/m3u?username=viewer&password=test-password")).send().await + { + if let Ok(body) = response.text().await { + if let Some(url) = body.lines().find(|line| line.starts_with("http")) { + break url.to_owned(); + } + } + } + assert!(tokio::time::Instant::now() < deadline, "catalog startup timeout"); + tokio::time::sleep(Duration::from_millis(25)).await; + }; + let master = client.get(&entry).send().await.unwrap().error_for_status().unwrap().text().await.unwrap(); + let variant = media_url(&entry, &master); + let initial = client.get(&variant).send().await.unwrap(); + assert_eq!(initial.status(), 200); + let previous = initial.text().await.unwrap(); + writeln!(protocol, "{}", serde_json::json!({"mode":mode,"phase":"initial","status":200,"mediaSequence":sequence(&previous,"#EXT-X-MEDIA-SEQUENCE:")})).unwrap(); + let segment = client.get(media_url(&variant, &previous)).send().await.unwrap(); + assert_eq!(segment.status(), 200); + assert!(!segment.bytes().await.unwrap().is_empty()); + assert_eq!(configure(6).send().await.unwrap().status(), 204); + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + while attempts.load(Ordering::Acquire) == 0 { + let response = client.get(&variant).send().await.unwrap(); + assert_eq!(response.status(), 200, "{mode}: waiting for origin refresh"); + assert_eq!(response.text().await.unwrap(), previous); + writeln!(protocol, "{}", serde_json::json!({"mode":mode,"phase":"refresh","status":200,"mediaSequence":sequence(&previous,"#EXT-X-MEDIA-SEQUENCE:")})).unwrap(); + assert!(tokio::time::Instant::now() < deadline, "successor origin request was never made"); + tokio::time::sleep(Duration::from_millis(25)).await; + } + for poll in 0..20 { + let response = client.get(&variant).send().await.unwrap(); + let status = response.status().as_u16(); + assert_eq!(status, 200, "{mode} held poll {poll}"); + assert!(response.headers().get("retry-after").is_none()); + assert_eq!(response.text().await.unwrap(), previous); + writeln!(protocol, "{}", serde_json::json!({"mode":mode,"phase":"held","poll":poll,"status":status,"mediaSequence":sequence(&previous,"#EXT-X-MEDIA-SEQUENCE:"),"successorOriginRequests":attempts.load(Ordering::Acquire)})).unwrap(); + } + let retained_responses = std::fs::read_to_string(&log_file) + .unwrap() + .lines() + .filter(|line| line.contains("HLS live window retained:")) + .count(); + assert!(retained_responses >= 20, "{mode}: held HTTP requests must exercise the retained-window branch"); + release.send(true).unwrap(); + let deadline = tokio::time::Instant::now() + Duration::from_secs(8); + loop { + let response = client.get(&variant).send().await.unwrap(); + assert_eq!(response.status(), 200); + let next = response.text().await.unwrap(); + if sequence(&next, "#EXT-X-MEDIA-SEQUENCE:") > sequence(&previous, "#EXT-X-MEDIA-SEQUENCE:") { + assert!( + sequence(&next, "#EXT-X-DISCONTINUITY-SEQUENCE:") + >= sequence(&previous, "#EXT-X-DISCONTINUITY-SEQUENCE:") + ); + writeln!(protocol, "{}", serde_json::json!({"mode":mode,"phase":"advanced","status":200,"mediaSequence":sequence(&next,"#EXT-X-MEDIA-SEQUENCE:"),"successorOriginRequests":attempts.load(Ordering::Acquire),"retainedResponses":retained_responses})).unwrap(); + break; + } + writeln!(protocol, "{}", serde_json::json!({"mode":mode,"phase":"released","status":200,"mediaSequence":sequence(&next,"#EXT-X-MEDIA-SEQUENCE:")})).unwrap(); + assert!(tokio::time::Instant::now() < deadline, "successor did not publish"); + tokio::time::sleep(Duration::from_millis(25)).await; + } + process.kill().await.unwrap(); + origin.abort(); + } +} diff --git a/backend/hls/src/lease.rs b/backend/hls/src/lease.rs index 8d7308637..24790e6d6 100644 --- a/backend/hls/src/lease.rs +++ b/backend/hls/src/lease.rs @@ -188,6 +188,7 @@ pub enum HlsLeaseManifestPublicationRejectReason { LeaseNotLive, SourceRegressive, ManifestGenerationUnavailable, + SnapshotChanged, RevisionConflict, SnapshotGenerationExhausted, } diff --git a/backend/hls/src/lib.rs b/backend/hls/src/lib.rs index ec5502bb7..68e89f11a 100644 --- a/backend/hls/src/lib.rs +++ b/backend/hls/src/lib.rs @@ -137,7 +137,9 @@ pub mod api { lease::{ new_hls_access_lease_id, HlsAccessLease, HlsAccessLeaseActivation, HlsAccessLeaseId, HlsAccessLeasePendingDeadline, HlsAccessLeaseState, HlsAccessLeaseTiming, HlsAccessLeaseTouch, - HlsFreshManifestRequiredReason, HlsLeaseStartupAdmissionState, HlsMediaLeaseIdentity, HlsPlaybackFamilyKey, + HlsFreshManifestRequiredReason, HlsLeaseManifestPublicationOutcome, + HlsLeaseManifestPublicationRejectReason, HlsLeaseStartupAdmissionState, HlsMediaLeaseIdentity, + HlsPlaybackFamilyKey, }, lifecycle::{HlsLifecycleEvent, HlsLifecycleEventKey}, live_window::{HlsLiveWindowPlan, HlsLiveWindowView}, @@ -185,7 +187,7 @@ pub mod api { cold_start_retry_after_seconds, maybe_trigger_origin_refresh_with_outcome, trigger_origin_refresh_sync, HlsManifestCommitRequirement, HlsOriginRefreshTriggerOutcome, HlsPostRefreshRuntime, OriginRefreshRequest, }, - renderer::{HlsManifestRenderer, RenderedManifest}, + renderer::{HlsManifestRenderer, RenderError, RenderedManifest}, resource_fetch::{ retry_after_secs_from_ms, HlsOriginResourceClients, HlsOriginResourceFetchError, HlsResourceFetchAttempt, }, diff --git a/backend/hls/src/manager/publication.rs b/backend/hls/src/manager/publication.rs index c5b5e407f..1c6b0e501 100644 --- a/backend/hls/src/manager/publication.rs +++ b/backend/hls/src/manager/publication.rs @@ -487,6 +487,7 @@ impl HlsProxyManager { snapshot, published_transient_resource_ids, now_ms, + None, ) .await; } diff --git a/backend/hls/src/manager/revision_recovery.rs b/backend/hls/src/manager/revision_recovery.rs index 13627b180..985be2727 100644 --- a/backend/hls/src/manager/revision_recovery.rs +++ b/backend/hls/src/manager/revision_recovery.rs @@ -19,6 +19,55 @@ pub(crate) struct HlsLeaseRuntime { pub(super) terminal_commit_retries: Arc, } +fn validate_retained_live_window( + leases: &mut crate::HlsAccessLeaseStore, + session: &crate::HlsSession, + lease_id: &HlsAccessLeaseId, + snapshot: &crate::media_reserve::HlsLeaseManifestSnapshot, + options: crate::manifest_commit::HlsCachedManifestOptions, + started_at_ms: u64, + now_ms: u64, +) -> Result<(), crate::lease::HlsLeaseManifestPublicationRejectReason> { + use crate::lease::HlsLeaseManifestPublicationRejectReason as Reject; + let Some(lease) = leases.response_snapshot(lease_id, &session.proxy_session_id, now_ms) else { + return Err(Reject::LeaseUnavailable); + }; + if lease.last_manifest_snapshot.as_ref() != Some(snapshot) { + return Err(Reject::SnapshotChanged); + } + if now_ms > snapshot.delivered_at_ms.saturating_add(snapshot.playlist_duration_ms) + || lease.state != crate::HlsAccessLeaseState::Activated + || options.requires_newer_manifest() + || session.fresh_manifest_commit_required.is_some() + || !crate::manifest_commit::hls_live_window_origin_allowed_for_request(session, options, started_at_ms, now_ms) + { + return Err(Reject::ManifestGenerationUnavailable); + } + let Some(publication) = snapshot.startup_revisions.as_ref() else { + return Err(Reject::RevisionConflict); + }; + if publication + .window_plan + .as_ref() + .is_none_or(|window| window.plan.origin_work_generation != session.activity.origin_work_generation) + { + return Err(Reject::ManifestGenerationUnavailable); + } + // GC can prune a snapshot's revision map. Every advertised URI must + // still resolve to the exact lease binding before recovery runs. + if snapshot.visible_segments.is_empty() + || snapshot.visible_proxy_seqs().any(|seq| { + publication + .revisions + .get(&seq) + .is_none_or(|revision| leases.published_segment_revision(lease_id, seq).as_ref() != Some(revision)) + }) + { + return Err(Reject::RevisionConflict); + } + Ok(()) +} + impl HlsProxyManager { pub(super) fn revision_recovery_runtime(&self) -> HlsLeaseRuntime { HlsLeaseRuntime { @@ -38,6 +87,37 @@ impl HlsProxyManager { let _ = self.segment_worker_pool.revision_recovery.set(Arc::new(self.revision_recovery_runtime())); } + /// Reuses the publication transaction, fencing a retained response to the + /// session handle and request freshness policy that selected it. + #[allow(clippy::too_many_arguments)] + pub async fn republish_access_lease_live_window( + &self, + session: &crate::HlsSessionHandle, + lease_id: &HlsAccessLeaseId, + session_id: &ProxySessionId, + expected: crate::lease::HlsLeaseManifestPublicationGuard, + snapshot: crate::media_reserve::HlsLeaseManifestSnapshot, + options: crate::manifest_commit::HlsCachedManifestOptions, + started_at_ms: u64, + now_ms: u64, + ) -> crate::lease::HlsLeaseManifestPublicationOutcome { + self.commit_recovered_startup_manifest( + lease_id, + session_id, + expected, + snapshot, + crate::HlsPublishedTransientResourceIds::default(), + now_ms, + Some((session, options, started_at_ms)), + ) + .await + } + + #[allow( + clippy::too_many_arguments, + clippy::too_many_lines, + reason = "Keep generation checks, revision recovery and publication under the same transaction locks." + )] pub(super) async fn commit_recovered_startup_manifest( &self, lease_id: &HlsAccessLeaseId, @@ -46,6 +126,7 @@ impl HlsProxyManager { mut snapshot: crate::media_reserve::HlsLeaseManifestSnapshot, resources: crate::HlsPublishedTransientResourceIds, now_ms: u64, + retained_request: Option<(&crate::HlsSessionHandle, crate::manifest_commit::HlsCachedManifestOptions, u64)>, ) -> crate::lease::HlsLeaseManifestPublicationOutcome { use crate::lease::{ HlsLeaseManifestPublicationOutcome as Outcome, HlsLeaseManifestPublicationRejectReason as Reject, @@ -53,8 +134,15 @@ impl HlsProxyManager { let Some(current) = self.sessions.hold_current_proxy_session(session_id).await else { return Outcome::Rejected(Reject::SessionMismatch); }; + if retained_request.is_some_and(|(expected_session, _, _)| !Arc::ptr_eq(current.session(), expected_session)) { + return Outcome::Rejected(Reject::SessionMismatch); + } let mut leases = self.access_leases.write().await; let mut session = current.session().write().await; + // A retained request may have waited for these locks. Recheck its + // deadline at the transaction without moving behind the request observation. + let now_ms = + if retained_request.is_some() { now_ms.max(tuliprox_core::utils::current_time_millis()) } else { now_ms }; if !leases.media_identity_is_current(lease_id, session_id, expected.media_identity(), now_ms) || session.is_gc_marked_for_removal() { @@ -69,6 +157,19 @@ impl HlsProxyManager { .and_then(|lease| lease.last_manifest_snapshot) .as_ref() == Some(&snapshot); + if let Some((_, options, started_at_ms)) = retained_request { + if let Err(reason) = validate_retained_live_window( + &mut leases, + &session, + lease_id, + &snapshot, + options, + started_at_ms, + now_ms, + ) { + return Outcome::Rejected(reason); + } + } if let Some(publication) = snapshot.startup_revisions.as_mut() { if let Some(window) = &publication.window_plan { let plan = &window.plan; @@ -110,6 +211,16 @@ impl HlsProxyManager { } } } + if retained_request.is_some() + && snapshot.startup_revisions.as_ref().is_none_or(|publication| { + publication + .revisions + .get(&snapshot.first_proxy_seq) + .is_none_or(|head| !head.revision().is_startup_head_ready(publication.mode)) + }) + { + return Outcome::Rejected(Reject::RevisionConflict); + } let published = snapshot .startup_revisions .as_ref() diff --git a/backend/hls/src/manifest_commit.rs b/backend/hls/src/manifest_commit.rs index 9cb072f4d..a60f7c951 100644 --- a/backend/hls/src/manifest_commit.rs +++ b/backend/hls/src/manifest_commit.rs @@ -64,7 +64,7 @@ impl HlsCachedManifestOptions { self } - const fn requires_newer_manifest(self) -> bool { + pub const fn requires_newer_manifest(self) -> bool { self.newer_than_rendered_at_ms.is_some() || self.after_origin_manifest_commits.is_some() } } diff --git a/backend/hls/src/segment_fetcher/tests/live_window.rs b/backend/hls/src/segment_fetcher/tests/live_window.rs index 6fbbb1410..b7017424e 100644 --- a/backend/hls/src/segment_fetcher/tests/live_window.rs +++ b/backend/hls/src/segment_fetcher/tests/live_window.rs @@ -157,7 +157,10 @@ async fn live_window_short_empty_readiness_and_stale_plans_are_explicit() { state.activity.origin_work_generation += 1; assert!(!plan.is_current(&state)); assert!(state.plan_live_window(HlsLiveWindowView::Entry).is_none()); - assert!(HlsManifestRenderer::render_live_window(&state, &plan, 0).is_err()); + assert_eq!( + HlsManifestRenderer::render_live_window(&state, &plan, 0), + Err(crate::renderer::RenderError::InvalidState), + ); } } } diff --git a/postman/mocks/hls-window/config.yaml b/postman/mocks/hls-window/config.yaml new file mode 100644 index 000000000..b2c649790 --- /dev/null +++ b/postman/mocks/hls-window/config.yaml @@ -0,0 +1,6 @@ +version: 1 +name: hls-window +interceptRequests: true +scenarios: + - name: default + path: default.js diff --git a/postman/mocks/hls-window/default.js b/postman/mocks/hls-window/default.js new file mode 100644 index 000000000..dda1eaf7f --- /dev/null +++ b/postman/mocks/hls-window/default.js @@ -0,0 +1,35 @@ +const http = require('http'); + +// Reuses the existing hls-window fixture. Optional controls advance origin +// metadata while a loopback streaming fixture gates actual TS responses. +http.createServer(async (req, res) => { + if (req.method === 'POST' && req.url === '/configure') { + const config = req.body; + if (!config || !/^http:\/\/127\.0\.0\.1:\d+$/.test(config.segmentOrigin) || + !Number.isSafeInteger(config.firstSequence) || config.firstSequence < 0) { + res.writeHead(400); + return res.end('invalid loopback fixture configuration'); + } + await pm.state.set('hls-window.config', config); + res.writeHead(204); + return res.end(); + } + if (req.method === 'GET' && req.url === '/index.m3u8') { + const config = await pm.state.get('hls-window.config'); + const first = config ? config.firstSequence : 0; + const segmentOrigin = config ? config.segmentOrigin + '/' : ''; + res.writeHead(200, { 'content-type': 'application/vnd.apple.mpegurl' }); + return res.end('#EXTM3U\n#EXT-X-TARGETDURATION:4\n' + + `#EXT-X-MEDIA-SEQUENCE:${first}\n` + + Array.from({ length: 6 }, (_, i) => `#EXTINF:4,\n${segmentOrigin}${first + i}.ts\n`).join('')); + } + if (req.method === 'GET' && /^\/[0-5]\.ts$/.test(req.url)) { + res.writeHead(200, { 'content-type': 'video/mp2t', 'content-length': 12 }); + return setTimeout(() => { + res.write('segment-'); + setTimeout(() => res.end('body'), 350); + }, 30); + } + res.writeHead(501); + res.end('endpoint outside the HLS window fixture contract'); +}).listen(process.env.PORT); diff --git a/postman/simulations/hls-window.sim.yaml b/postman/simulations/hls-window.sim.yaml new file mode 100644 index 000000000..fbce063ef --- /dev/null +++ b/postman/simulations/hls-window.sim.yaml @@ -0,0 +1,5 @@ +simulation: hls-window +routing: path +mocks: + - routeKey: origin + path: ../mocks/hls-window/config.yaml