mirror of
https://github.com/euzu/tuliprox.git
synced 2026-10-30 10:42:29 +01:00
fix (hls): retain published manifests during live window transitions
This commit is contained in:
knylbyte
@@ -26,6 +26,54 @@ use tuliprox_hls::api::{
|
||||
ProxySessionId,
|
||||
};
|
||||
|
||||
type HlsPreparedManifest =
|
||||
(HlsMaterializedSharedManifest, HlsLeaseManifestSnapshot, HlsPublishedTransientResourceIds, u64);
|
||||
|
||||
enum HlsLiveWindowPreparation {
|
||||
Prepared(HlsPreparedManifest),
|
||||
NoRenderableWindow { lease: Box<tuliprox_hls::api::HlsAccessLease>, 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<HlsPreparedManifest, &'static str> {
|
||||
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<HlsPreparedManifest, &'static str> {
|
||||
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<AppState>,
|
||||
@@ -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 =
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<AppState>,
|
||||
manager: Arc<HlsProxyManager>,
|
||||
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),
|
||||
);
|
||||
}
|
||||
@@ -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::<u64>().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();
|
||||
}
|
||||
}
|
||||
@@ -188,6 +188,7 @@ pub enum HlsLeaseManifestPublicationRejectReason {
|
||||
LeaseNotLive,
|
||||
SourceRegressive,
|
||||
ManifestGenerationUnavailable,
|
||||
SnapshotChanged,
|
||||
RevisionConflict,
|
||||
SnapshotGenerationExhausted,
|
||||
}
|
||||
|
||||
@@ -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,
|
||||
},
|
||||
|
||||
@@ -487,6 +487,7 @@ impl HlsProxyManager {
|
||||
snapshot,
|
||||
published_transient_resource_ids,
|
||||
now_ms,
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
@@ -19,6 +19,55 @@ pub(crate) struct HlsLeaseRuntime {
|
||||
pub(super) terminal_commit_retries: Arc<crate::terminal_commit::HlsTerminalCommitRetryCoordinator>,
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,6 @@
|
||||
version: 1
|
||||
name: hls-window
|
||||
interceptRequests: true
|
||||
scenarios:
|
||||
- name: default
|
||||
path: 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);
|
||||
@@ -0,0 +1,5 @@
|
||||
simulation: hls-window
|
||||
routing: path
|
||||
mocks:
|
||||
- routeKey: origin
|
||||
path: ../mocks/hls-window/config.yaml
|
||||
Reference in New Issue
Block a user