From 0ddc3e21cc537caa42dca7ce74830bde3f3ef454 Mon Sep 17 00:00:00 2001 From: euzu <33094714+euzu@users.noreply.github.com> Date: Mon, 12 Oct 2026 22:43:06 +0200 Subject: [PATCH] feat(hls): add optimistic startup mode (#914) --- CHANGELOG.md | 4 + .../hls_api/tests/manifest_entry_freshness.rs | 2 + backend/hls/src/lease/revision_bindings.rs | 6 +- backend/hls/src/lease/tests/startup.rs | 78 ++++++++++++ backend/hls/src/media_reserve.rs | 43 ++++++- backend/hls/src/renderer.rs | 73 ++++++++++- backend/hls/src/response/tests/lifecycle.rs | 75 +++++++++++ backend/hls/src/segment_fetcher/scheduler.rs | 7 +- .../hls/src/segment_fetcher/startup_fill.rs | 40 ++++-- backend/hls/src/segment_fetcher/tests.rs | 6 +- .../hls/src/segment_fetcher/tests/startup.rs | 116 +++++++++++++++++- .../hls/src/segment_fetcher/tests/support.rs | 50 +++++++- backend/hls/src/segment_revision.rs | 37 +++++- .../configuration/shared-hls-configuration.md | 18 ++- frontend/public/assets/i18n/ar.json | 1 + frontend/public/assets/i18n/en.json | 1 + frontend/public/assets/i18n/es.json | 1 + frontend/public/assets/i18n/ru.json | 1 + .../config/reverse_proxy_config_view.rs | 5 + shared/src/model/config/hls/startup.rs | 10 ++ .../model/config/hls/startup_config_tests.rs | 21 +++- .../m3u-hls-share-optimistic-missing-head.yml | 50 ++++++++ .../m3u-hls-share-optimistic-startup.yml | 50 ++++++++ tools/tuliprox-testkit/src/hls_probe.rs | 2 + .../tests/hls_fast_start_process.rs | 12 +- 25 files changed, 674 insertions(+), 35 deletions(-) create mode 100644 test/fixtures/testkit/scenarios/m3u-hls-share-optimistic-missing-head.yml create mode 100644 test/fixtures/testkit/scenarios/m3u-hls-share-optimistic-startup.yml diff --git a/CHANGELOG.md b/CHANGELOG.md index 343e03416..a0377adc3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -189,6 +189,10 @@ reader lifetime and deferred repairs are bounded by the new `startup` limits; revision files share the existing disk cache limits. Archive, encrypted and fMP4/MAP sessions keep the existing path. The Web UI exposes the mode and its limits. See [Optional startup modes](docs/src/configuration/shared-hls-configuration.md#optional-startup-modes). +- **Optimistic fast start for shared live HLS.** `startup.mode: optimistic` works like `progressive`, but publishes + the first playlist as soon as the head segment has data and its successors were accepted by the origin, instead + of waiting for data from three segments. Successor fetches start together with the head. Opt-in; the default + stays `conservative`. - **Personal Web UI table layouts.** Show or hide optional columns, reorder them with mouse, touch, or keyboard, and save layouts per account. Restore defaults, cancel drafts, and resolve conflicting saves without losing edits. diff --git a/backend/app/src/api/endpoints/hls_api/tests/manifest_entry_freshness.rs b/backend/app/src/api/endpoints/hls_api/tests/manifest_entry_freshness.rs index cf4425a93..ae6bf363c 100644 --- a/backend/app/src/api/endpoints/hls_api/tests/manifest_entry_freshness.rs +++ b/backend/app/src/api/endpoints/hls_api/tests/manifest_entry_freshness.rs @@ -114,9 +114,11 @@ async fn new_playback_on_retained_session_starts_on_current_origin_window() { (HlsStartupMode::Progressive, 16_000, true), (HlsStartupMode::FirstReady, 16_000, true), (HlsStartupMode::Conservative, 16_000, true), + (HlsStartupMode::Optimistic, 16_000, true), (HlsStartupMode::Progressive, 300_000, false), (HlsStartupMode::FirstReady, 300_000, false), (HlsStartupMode::Conservative, 300_000, false), + (HlsStartupMode::Optimistic, 300_000, false), ] { let head = Arc::new(AtomicU64::new(0)); let origin = spawn_moving_origin(Arc::clone(&head)).await; diff --git a/backend/hls/src/lease/revision_bindings.rs b/backend/hls/src/lease/revision_bindings.rs index c582b9ce8..3bc1bcb99 100644 --- a/backend/hls/src/lease/revision_bindings.rs +++ b/backend/hls/src/lease/revision_bindings.rs @@ -63,10 +63,12 @@ pub(super) fn commit_lease_revision_bindings(lease: &mut HlsAccessLease, snapsho if let Some(publication) = &snapshot.startup_revisions { let current = lease.revision_bindings.as_ref(); if publication.revisions.iter().any(|(seq, revision)| { - !revision.revision().is_publishable() + !revision.revision().is_window_publishable(publication.mode) || current.is_some_and(|current| { current.retired_through.is_some_and(|retired| *seq <= retired) - || current.revisions.get(seq).is_some_and(|bound| bound != revision) + || current.revisions.get(seq).is_some_and(|bound| { + bound != revision && !bound.revision().is_replaceable_in(publication.mode) + }) }) }) { return false; diff --git a/backend/hls/src/lease/tests/startup.rs b/backend/hls/src/lease/tests/startup.rs index 3812bd2f1..89ff91062 100644 --- a/backend/hls/src/lease/tests/startup.rs +++ b/backend/hls/src/lease/tests/startup.rs @@ -81,3 +81,81 @@ fn startup_claim_only_accepts_the_published_head_once() { assert!(!leases.try_claim_progressive_startup(&lease_id, 40, 2102)); assert!(!leases.try_claim_progressive_startup(&lease_id, 41, 2102)); } + +#[test] +fn optimistic_binding_accepts_pending_successors_but_not_failed_or_replaced_ones() -> std::io::Result<()> { + let store = crate::SegmentRevisionStore::default(); + let session = ProxySessionId("optimistic".into()); + let snapshot_with = |revisions: std::collections::BTreeMap| { + let mut snapshot = manifest_snapshot(1); + snapshot.startup_revisions = Some(Arc::new(crate::HlsManifestRevisions { + mode: shared::model::HlsStartupMode::Optimistic, + retained_start_seq: 0, + revisions, + })); + snapshot + }; + let mut owners = std::collections::BTreeMap::new(); + for seq in 0..3 { + owners.insert(seq, store.create(session.clone(), seq, crate::SegmentRevisionKind::Raw)?); + } + owners + .get(&0) + .ok_or_else(|| std::io::Error::other("head"))? + .revision() + .prefix_available + .store(1, std::sync::atomic::Ordering::Release); + let mut bound = lease(HlsAccessLeaseId("optimistic".into()), &session.0, 1000); + assert!(super::super::commit_lease_revision_bindings(&mut bound, &snapshot_with(owners.clone()))); + + let mut replaced = owners.clone(); + replaced.insert(1, store.create(session.clone(), 1, crate::SegmentRevisionKind::Raw)?); + assert!(!super::super::commit_lease_revision_bindings(&mut bound, &snapshot_with(replaced))); + + let mut fresh_lease = lease(HlsAccessLeaseId("failed".into()), &session.0, 1000); + owners.get(&2).ok_or_else(|| std::io::Error::other("successor"))?.revision().fail(); + assert!(!super::super::commit_lease_revision_bindings(&mut fresh_lease, &snapshot_with(owners))); + Ok(()) +} + +#[test] +fn failed_unexposed_successor_may_be_rebound_but_exposed_bytes_never_change() -> std::io::Result<()> { + let store = crate::SegmentRevisionStore::default(); + let session = ProxySessionId("rebind".into()); + let snapshot_with = |revisions: std::collections::BTreeMap| { + let mut snapshot = manifest_snapshot(1); + snapshot.startup_revisions = Some(Arc::new(crate::HlsManifestRevisions { + mode: shared::model::HlsStartupMode::Optimistic, + retained_start_seq: 0, + revisions, + })); + snapshot + }; + let budget = crate::ProgressiveBudgetManager::new(tuliprox_core::model::HlsStartupConfig::default()); + let mut owners = std::collections::BTreeMap::new(); + for seq in 0..3 { + owners.insert(seq, store.create(session.clone(), seq, crate::SegmentRevisionKind::Raw)?); + } + let head = owners.get(&0).ok_or_else(|| std::io::Error::other("head"))?.clone(); + let mut replay = + crate::progressive_startup::ProgressiveReplay::new(&budget).ok_or_else(|| std::io::Error::other("slot"))?; + replay.push(b"prefix")?; + *head.revision().replay.lock().map_err(|_| std::io::Error::other("replay"))? = Some(replay); + head.revision().prefix_available.store(6, std::sync::atomic::Ordering::Release); + let mut bound = lease(HlsAccessLeaseId("rebind".into()), &session.0, 1000); + assert!(super::super::commit_lease_revision_bindings(&mut bound, &snapshot_with(owners.clone()))); + + // The origin fails a listed successor before any byte reached a client; its retry may replace it. + owners.get(&1).ok_or_else(|| std::io::Error::other("successor"))?.revision().fail(); + let mut retried = owners.clone(); + retried.insert(1, store.create(session.clone(), 1, crate::SegmentRevisionKind::Raw)?); + assert!(super::super::commit_lease_revision_bindings(&mut bound, &snapshot_with(retried.clone()))); + + // A revision that streamed bytes stays bound even after it failed. + assert!(head.revision().expose()); + head.revision().fail(); + let mut replaced_head = retried; + replaced_head.insert(0, store.create(session, 0, crate::SegmentRevisionKind::Raw)?); + assert!(!super::super::commit_lease_revision_bindings(&mut bound, &snapshot_with(replaced_head))); + Ok(()) +} diff --git a/backend/hls/src/media_reserve.rs b/backend/hls/src/media_reserve.rs index 186b6aedd..4ef0849d9 100644 --- a/backend/hls/src/media_reserve.rs +++ b/backend/hls/src/media_reserve.rs @@ -559,7 +559,7 @@ pub fn evaluate_startup_admission(input: HlsStartupAdmissionInput<'_>) -> HlsSta publication .revisions .get(&segment.proxy_seq) - .is_some_and(|revision| revision.revision().is_publishable()) + .is_some_and(|revision| revision.revision().is_window_publishable(publication.mode)) }) && publication .revisions @@ -1012,6 +1012,47 @@ mod tests { Ok(()) } + #[test] + fn optimistic_admission_needs_head_bytes_and_unfailed_successors() -> std::io::Result<()> { + let store = crate::SegmentRevisionStore::default(); + let mut manifest = manifest(); + let mut owners = std::collections::BTreeMap::new(); + for seq in 0..3 { + owners.insert( + seq, + store.create(crate::ProxySessionId("admission".into()), seq, crate::SegmentRevisionKind::Raw)?, + ); + } + manifest.startup_revisions = Some(Arc::new(crate::HlsManifestRevisions { + mode: shared::model::HlsStartupMode::Optimistic, + retained_start_seq: 0, + revisions: owners, + })); + let timeline = HlsReadyTimelineSnapshot { units: Arc::from([]) }; + assert!(matches!( + admission(&manifest, &timeline, HlsStartupAdmissionOriginState::Healthy, 0).decision, + HlsStartupAdmissionDecision::Reject(_) + )); + let publication = manifest.startup_revisions.as_ref().ok_or_else(|| std::io::Error::other("publication"))?; + publication + .revisions + .get(&0) + .ok_or_else(|| std::io::Error::other("head"))? + .revision() + .prefix_available + .store(1, std::sync::atomic::Ordering::Release); + assert_eq!( + admission(&manifest, &timeline, HlsStartupAdmissionOriginState::Healthy, 0).decision, + HlsStartupAdmissionDecision::Admit + ); + publication.revisions.get(&2).ok_or_else(|| std::io::Error::other("successor"))?.revision().fail(); + assert!(matches!( + admission(&manifest, &timeline, HlsStartupAdmissionOriginState::Healthy, 0).decision, + HlsStartupAdmissionDecision::Reject(_) + )); + Ok(()) + } + #[test] fn healthy_startup_admission_uses_contiguous_visible_ready_window() { let admission = admission(&manifest(), &ready_timeline(), HlsStartupAdmissionOriginState::Healthy, u64::MAX); diff --git a/backend/hls/src/renderer.rs b/backend/hls/src/renderer.rs index 148475bdc..f9f2580eb 100644 --- a/backend/hls/src/renderer.rs +++ b/backend/hls/src/renderer.rs @@ -225,7 +225,10 @@ pub fn renderer_candidate_window_proxy_seqs(session: &HlsSession) -> Vec { fn is_renderable(entry: &SegmentEntry, session: &HlsSession) -> bool { if let Some(startup) = &session.startup { return entry.is_fast_start_eligible() - && startup.revisions.get(&entry.proxy_seq).is_some_and(|revision| revision.revision().is_publishable()); + && startup + .revisions + .get(&entry.proxy_seq) + .is_some_and(|revision| revision.revision().is_window_publishable(startup.config.mode)); } match entry.status { SegmentCacheStatus::Ready { .. } => {} @@ -250,7 +253,11 @@ fn is_renderable(entry: &SegmentEntry, session: &HlsSession) -> bool { fn select_window(session: &HlsSession, render_gap_segments: usize, tail_policy: RenderTailPolicy) -> Option> { let head_seq = session.publishable_origin_head_proxy_seq?; - if session.startup.is_some() { + if let Some(startup) = &session.startup { + // The head always needs body bytes; only optimistic successors may still be waiting for theirs. + if !startup.revisions.get(&head_seq).is_some_and(|revision| revision.revision().is_publishable()) { + return None; + } let planned_tail = session.publishable_origin_tail_proxy_seq?.checked_sub(u64::try_from(render_gap_segments).ok()?)?; let wanted = if session.published_live_origin_baseline.is_none() { @@ -589,6 +596,68 @@ mod tests { ) } + fn optimistic_startup_session(mode: shared::model::HlsStartupMode) -> std::io::Result { + let mut session = session(); + session + .apply_origin_manifest(&normal_manifest( + "#EXTM3U\n#EXT-X-TARGETDURATION:2\n#EXT-X-MEDIA-SEQUENCE:0\n#EXTINF:2,\n0.ts\n#EXTINF:2,\n1.ts\n#EXTINF:2,\n2.ts\n", + )) + .map_err(|error| std::io::Error::other(format!("{error:?}")))?; + let config = tuliprox_core::model::HlsStartupConfig { mode, ..Default::default() }; + let worker = std::sync::Arc::new(crate::HlsSegmentWorkerPool::default()); + let store = std::sync::Arc::new(crate::SegmentRevisionStore::default()); + let mut revisions = std::collections::BTreeMap::new(); + for seq in session.segments.keys().copied().collect::>() { + revisions + .insert(seq, store.create(session.proxy_session_id.clone(), seq, crate::SegmentRevisionKind::Raw)?); + } + session.startup = Some(crate::HlsSessionStartup { + config: config.clone(), + first_data_timeout_ms: 10_000, + worker: std::sync::Arc::downgrade(&worker), + access_leases: std::sync::Arc::new(tokio::sync::RwLock::new(crate::HlsAccessLeaseStore::default())), + revisions, + policy_fixed: true, + store, + budget: crate::ProgressiveBudgetManager::new(config), + }); + Ok(session) + } + + fn startup_revision(session: &HlsSession, offset: usize) -> std::io::Result { + session + .startup + .as_ref() + .and_then(|startup| startup.revisions.values().nth(offset).cloned()) + .ok_or_else(|| std::io::Error::other("revision")) + } + + #[test] + fn optimistic_startup_publishes_pending_successors_once_the_head_has_bytes() -> std::io::Result<()> { + for (mode, renders) in + [(shared::model::HlsStartupMode::Optimistic, true), (shared::model::HlsStartupMode::Progressive, false)] + { + let mut session = optimistic_startup_session(mode)?; + assert!(matches!(session.render_and_store_manifest(10), Err(RenderError::NoRenderableWindow))); + startup_revision(&session, 0)?.revision().prefix_available.store(1, std::sync::atomic::Ordering::Release); + let rendered = session.render_and_store_manifest(20); + assert_eq!(rendered.is_ok(), renders, "mode={mode}"); + if let Ok(rendered) = rendered { + assert_eq!(rendered.segment_proxy_seqs.len(), 3); + } + } + Ok(()) + } + + #[test] + fn optimistic_startup_window_stops_before_a_failed_successor() -> std::io::Result<()> { + let mut session = optimistic_startup_session(shared::model::HlsStartupMode::Optimistic)?; + startup_revision(&session, 0)?.revision().prefix_available.store(1, std::sync::atomic::Ordering::Release); + startup_revision(&session, 1)?.revision().fail(); + assert!(matches!(session.render_and_store_manifest(10), Err(RenderError::NoRenderableWindow))); + Ok(()) + } + fn mark_all_segments_ready(session: &mut HlsSession) { for segment in session.segments.values_mut() { segment.status = SegmentCacheStatus::Ready { content_length: 1024, ready_at_ms: 10 }; diff --git a/backend/hls/src/response/tests/lifecycle.rs b/backend/hls/src/response/tests/lifecycle.rs index da261ac2d..5d5083d1c 100644 --- a/backend/hls/src/response/tests/lifecycle.rs +++ b/backend/hls/src/response/tests/lifecycle.rs @@ -129,3 +129,78 @@ async fn dropped_full_range_segment_body_records_request_without_completion() { assert_eq!(fixture.meter.bytes_total(), u64::try_from(first_chunk.len()).expect("first chunk size")); assert_eq!(fixture.access.active_readers(), 0); } + +#[tokio::test] +async fn optimistic_successor_that_fails_after_publication_is_a_permanent_segment_failure() -> std::io::Result<()> { + let fixture = CachedSegmentFixture::new(b"canonical-body".to_vec()).await; + let seq = fixture.segment_file.proxy_seq; + let head_seq = seq.checked_sub(1).ok_or_else(|| std::io::Error::other("fixture head"))?; + let config = tuliprox_core::model::HlsStartupConfig { + mode: shared::model::HlsStartupMode::Optimistic, + ..Default::default() + }; + let store = Arc::new(crate::SegmentRevisionStore::default()); + let head = store.create(fixture.proxy_session_id.clone(), head_seq, crate::SegmentRevisionKind::Raw)?; + head.revision().prefix_available.store(1, Ordering::Release); + let successor = store.create(fixture.proxy_session_id.clone(), seq, crate::SegmentRevisionKind::Raw)?; + let revisions = std::collections::BTreeMap::from([(head_seq, head), (seq, successor.clone())]); + fixture.session.write().await.startup = Some(crate::HlsSessionStartup { + config: config.clone(), + first_data_timeout_ms: 1000, + worker: Weak::new(), + access_leases: Arc::clone(fixture.manager.access_leases()), + revisions: revisions.clone(), + policy_fixed: true, + store, + budget: crate::ProgressiveBudgetManager::new(config.clone()), + }); + let segment = |proxy_seq: u64, uri: &str| crate::HlsLeaseManifestSegment { + proxy_seq, + duration_ms: 4000, + uri: uri.into(), + discontinuity_before: false, + map_ref_ready: true, + encryption: None, + }; + let snapshot = crate::HlsLeaseManifestSnapshot { + startup_revisions: Some(Arc::new(crate::HlsManifestRevisions { + mode: config.mode, + retained_start_seq: head_seq, + revisions, + })), + delivery_mode: crate::HlsManifestDeliveryMode::NormalCacheTimeline, + source_commit_identity: crate::HlsManifestCommitIdentity::new(1), + uri_materialization: None, + finalized_transient_manifest_generation: None, + snapshot_generation: 0, + delivered_at_ms: current_time_millis(), + first_proxy_seq: head_seq, + last_proxy_seq: seq, + visible_segments: Arc::from([segment(head_seq, "11.ts"), segment(seq, "12.ts")]), + discontinuity_sequence: 0, + target_duration_ms: 4000, + playlist_duration_ms: 8000, + last_visible_media_end_ms: 8000, + active_map: None, + active_encryption: None, + container: crate::HlsMediaContainer::MpegTs, + }; + { + let mut leases = fixture.manager.access_leases().write().await; + let now = current_time_millis(); + let guard = leases + .prepare_manifest_publication(&fixture.lease_id, &fixture.proxy_session_id, now) + .ok_or_else(|| std::io::Error::other("publication guard"))?; + assert!(leases + .commit_manifest_publication(&fixture.lease_id, &fixture.proxy_session_id, guard, snapshot, now) + .is_committed()); + } + // Listing does not expose a revision; only streaming bytes does. The origin + // failing a listed successor therefore fails just that one segment. + successor.revision().fail(); + assert!(matches!( + fixture.serve(None).await, + HlsResourceServeOutcome::Failure(HlsResourceServeFailure::PermanentFailed { .. }) + )); + Ok(()) +} diff --git a/backend/hls/src/segment_fetcher/scheduler.rs b/backend/hls/src/segment_fetcher/scheduler.rs index d282c2e6e..99cc90e4f 100644 --- a/backend/hls/src/segment_fetcher/scheduler.rs +++ b/backend/hls/src/segment_fetcher/scheduler.rs @@ -269,10 +269,15 @@ impl HlsSegmentWorkerPool { return None; } let initial_barrier = session.startup.as_ref().is_some_and(|startup| { + let mode = startup.config.mode; + // Optimistic startup lists successors before their bodies arrive, + // so they must start together with the head. + if mode.publishes_pending_successors() { + return false; + } let Some(head) = session.publishable_origin_head_proxy_seq else { return true; }; - let mode = startup.config.mode; startup.revisions.get(&head).is_none_or(|revision| { mode != shared::model::HlsStartupMode::Conservative && !revision.revision().is_startup_head_ready(mode) }) diff --git a/backend/hls/src/segment_fetcher/startup_fill.rs b/backend/hls/src/segment_fetcher/startup_fill.rs index 8b767405b..3d48ab8f1 100644 --- a/backend/hls/src/segment_fetcher/startup_fill.rs +++ b/backend/hls/src/segment_fetcher/startup_fill.rs @@ -244,7 +244,7 @@ async fn try_prepare_startup_fill( expected: Option, deadline: Instant, ) -> Result, SegmentFetchError> { - let (store, budget, mut mode, limit, session_id) = { + let (store, budget, mut mode, limit, session_id, decides_policy) = { let session = context.session.read().await; if !session.segments.get(&snapshot.proxy_seq).is_some_and(super::super::SegmentEntry::is_fast_start_eligible) { return Err(SegmentFetchError::UnsupportedStartupMedia); @@ -259,13 +259,13 @@ async fn try_prepare_startup_fill( startup.config.mode, startup.config.max_progressive_bytes_per_segment.get(), session.proxy_session_id.clone(), + startup_fill_decides_policy(&session, snapshot.proxy_seq), ) }; - let mut replay = (mode == shared::model::HlsStartupMode::Progressive - && expected.is_none_or(|length| length <= limit)) - .then(|| super::super::progressive_startup::ProgressiveReplay::new(&budget)) - .flatten(); - if mode == shared::model::HlsStartupMode::Progressive && replay.is_none() { + let mut replay = (mode.streams_raw_revisions() && expected.is_none_or(|length| length <= limit)) + .then(|| super::super::progressive_startup::ProgressiveReplay::new(&budget)) + .flatten(); + if mode.streams_raw_revisions() && replay.is_none() && decides_policy { let mut session = context.session.write().await; if session.startup.as_ref().is_some_and(|startup| !startup.policy_fixed) { if let Some(startup) = &mut session.startup { @@ -275,7 +275,7 @@ async fn try_prepare_startup_fill( debug!("HLS startup fallback: reason=replay_admission effective_mode=first_ready"); } } - let raw = mode == shared::model::HlsStartupMode::Progressive; + let raw = mode.streams_raw_revisions(); let kind = if raw { super::super::SegmentRevisionKind::Raw } else { super::super::SegmentRevisionKind::Processed }; let revision = store.create(session_id, snapshot.proxy_seq, kind).map_err(|error| SegmentFetchError::cache_body(&error))?; @@ -285,7 +285,7 @@ async fn try_prepare_startup_fill( .ok_or_else(|| SegmentFetchError::cache_body(&std::io::Error::other("revision peak overflow")))?; let mut reservation = match context.segment_cache.reserve_revision_peak(&revision.revision().key, peak).await { Ok(reservation) => reservation, - Err(_) if raw && fallback_to_first_ready(&context.session).await => { + Err(_) if raw && decides_policy && fallback_to_first_ready(&context.session).await => { debug!("HLS startup fallback: reason=disk_capacity effective_mode=first_ready"); return Ok(None); } @@ -345,13 +345,31 @@ async fn bind_startup_revision( "startup fetch generation retired", ))); } - if let Some(startup) = &mut session.startup { - startup.policy_fixed = true; - startup.revisions.insert(snapshot.proxy_seq, revision.clone()); + let decides_policy = startup_fill_decides_policy(&session, snapshot.proxy_seq); + let Some(startup) = &mut session.startup else { return Ok(()) }; + startup.policy_fixed |= decides_policy; + startup.revisions.insert(snapshot.proxy_seq, revision.clone()); + // An optimistic window may become complete when its last successor is + // accepted, after the head already has bytes. + if startup.config.mode.publishes_pending_successors() && session.last_rendered_manifest.is_none() { + if let Err(error) = session.render_and_store_manifest(current_time_millis()) { + debug!( + "HLS optimistic startup render after successor binding failed: proxy_seq={} error={error:?}", + snapshot.proxy_seq + ); + } } Ok(()) } +/// Optimistic startup fills successors alongside the head, so only the head's +/// fill may fix the session policy or fall back to `first_ready`. A successor +/// without replay stays raw and is delivered once complete. +fn startup_fill_decides_policy(session: &super::super::HlsSession, proxy_seq: u64) -> bool { + !session.startup_mode().publishes_pending_successors() + || session.publishable_origin_head_proxy_seq == Some(proxy_seq) +} + async fn fallback_to_first_ready(session: &HlsSessionHandle) -> bool { let mut session = session.write().await; if session.published_live_origin_baseline.is_some() { diff --git a/backend/hls/src/segment_fetcher/tests.rs b/backend/hls/src/segment_fetcher/tests.rs index a681a1046..e3fc02b48 100644 --- a/backend/hls/src/segment_fetcher/tests.rs +++ b/backend/hls/src/segment_fetcher/tests.rs @@ -17,7 +17,7 @@ mod support; use self::support::{ clear_scheduled_prefetch, cold_startup_round, commit_test_key_ready, committed_segment, encode_test_body, encrypted_fetch_context, fetch_context, fetch_context_with_access_lease, grant_usable_worker_access_lease, - grant_usable_worker_access_lease_at, install_startup, normal_manifest, prefix_drop_rounds, - shared_key_fetch_and_wait, spawn_segment_server, spawn_sequence_response_server, temp_cache_files, - test_segment_repair_manager, TestContentEncoding, TestOriginResponse, TestSegmentServer, + grant_usable_worker_access_lease_at, headers_only_successor_origin, install_startup, normal_manifest, + prefix_drop_rounds, shared_key_fetch_and_wait, spawn_segment_server, spawn_sequence_response_server, + temp_cache_files, test_segment_repair_manager, TestContentEncoding, TestOriginResponse, TestSegmentServer, }; diff --git a/backend/hls/src/segment_fetcher/tests/startup.rs b/backend/hls/src/segment_fetcher/tests/startup.rs index 4b6a7c12e..0a2afd23b 100644 --- a/backend/hls/src/segment_fetcher/tests/startup.rs +++ b/backend/hls/src/segment_fetcher/tests/startup.rs @@ -1,9 +1,11 @@ use super::{ - clear_scheduled_prefetch, cold_startup_round, fetch_context, install_startup, prefix_drop_rounds, + clear_scheduled_prefetch, cold_startup_round, fetch_context, fetch_context_with_access_lease, + grant_usable_worker_access_lease_at, headers_only_successor_origin, install_startup, prefix_drop_rounds, spawn_segment_server, spawn_sequence_response_server, SegmentFetchPolicy, SegmentFetchPriority, }; use crate::{HlsOriginResourceFetchError, SegmentCacheStatus}; use std::{sync::Arc, time::Duration}; +use tuliprox_core::utils::current_time_millis; #[tokio::test] async fn capacity_deferred_segment_requeues_only_after_capacity_revision_changes() { @@ -70,12 +72,19 @@ async fn capacity_deferred_segment_requeues_only_after_capacity_revision_changes } #[tokio::test] -async fn known_oversized_startup_body_falls_back_before_replay_or_publication() -> std::io::Result<()> { +async fn known_oversized_startup_body_falls_back_before_replay_or_publication_for_raw_modes() -> std::io::Result<()> { + for mode in [shared::model::HlsStartupMode::Progressive, shared::model::HlsStartupMode::Optimistic] { + known_oversized_startup_body_falls_back(mode).await?; + } + Ok(()) +} + +async fn known_oversized_startup_body_falls_back(mode: shared::model::HlsStartupMode) -> std::io::Result<()> { let server = spawn_sequence_response_server(Vec::new()).await; let directory = tempfile::tempdir()?; let policy = SegmentFetchPolicy::default(); let (worker, context, _) = fetch_context(&server, &directory, &policy).await; - install_startup(&context, &worker, shared::model::HlsStartupMode::Progressive).await; + install_startup(&context, &worker, mode).await; let snapshot = worker .next_fetch_snapshot(&context, 10, &policy) .await @@ -105,7 +114,11 @@ async fn known_oversized_startup_body_falls_back_before_replay_or_publication() #[tokio::test] async fn cold_startup_publication_and_first_data_precede_successor_eof() -> std::io::Result<()> { - for mode in [shared::model::HlsStartupMode::FirstReady, shared::model::HlsStartupMode::Progressive] { + for mode in [ + shared::model::HlsStartupMode::FirstReady, + shared::model::HlsStartupMode::Progressive, + shared::model::HlsStartupMode::Optimistic, + ] { cold_startup_round(mode, None).await?; } Ok(()) @@ -113,7 +126,11 @@ async fn cold_startup_publication_and_first_data_precede_successor_eof() -> std: #[tokio::test] async fn first_ready_and_progressive_keep_one_origin_fill_and_exact_bytes() -> std::io::Result<()> { - for mode in [shared::model::HlsStartupMode::FirstReady, shared::model::HlsStartupMode::Progressive] { + for mode in [ + shared::model::HlsStartupMode::FirstReady, + shared::model::HlsStartupMode::Progressive, + shared::model::HlsStartupMode::Optimistic, + ] { let server = spawn_segment_server(0).await; let directory = tempfile::tempdir()?; let policy = SegmentFetchPolicy::default(); @@ -134,7 +151,7 @@ async fn first_ready_and_progressive_keep_one_origin_fill_and_exact_bytes() -> s let metadata = revision.revision().wait_complete(tokio::time::Instant::now()).await?; assert_eq!(tokio::fs::read(metadata.path).await?, b"body:/1.ts"); assert_eq!(server.requests.lock().await.iter().filter(|request| request.starts_with("GET /1.ts ")).count(), 1); - if mode == shared::model::HlsStartupMode::Progressive { + if mode.streams_raw_revisions() { assert_eq!(revision.revision().key.kind, crate::SegmentRevisionKind::Raw); assert_eq!(context.session.read().await.startup.as_ref().map(|s| s.budget.usage()), Some((0, 0))); } @@ -146,3 +163,90 @@ async fn first_ready_and_progressive_keep_one_origin_fill_and_exact_bytes() -> s async fn progressive_exposes_prefix_before_origin_eof_and_drop_is_terminal() -> std::io::Result<()> { prefix_drop_rounds().await } + +#[tokio::test] +async fn optimistic_publishes_before_successor_body_bytes() -> std::io::Result<()> { + for (mode, publishes) in + [(shared::model::HlsStartupMode::Optimistic, true), (shared::model::HlsStartupMode::Progressive, false)] + { + let (server, release) = headers_only_successor_origin().await?; + let directory = tempfile::tempdir()?; + let policy = SegmentFetchPolicy { max_session_segment_fetches: 3, ..Default::default() }; + let (worker, context, _) = fetch_context_with_access_lease(&server, &directory, &policy, false).await; + install_startup(&context, &worker, mode).await; + let session_id = context.session.read().await.proxy_session_id.clone(); + grant_usable_worker_access_lease_at(&worker, &session_id, current_time_millis()).await; + worker.wake_scheduler(context.clone(), current_time_millis()).await; + let published = tokio::time::timeout(Duration::from_millis(500), async { + loop { + if context.session.read().await.last_rendered_manifest.is_some() { + return; + } + tokio::time::sleep(Duration::from_millis(5)).await; + } + }) + .await + .is_ok(); + assert_eq!(published, publishes, "mode={mode}"); + release.send_replace(true); + } + Ok(()) +} + +async fn optimistic_head_and_successor_snapshots( + server: &super::TestSegmentServer, + directory: &tempfile::TempDir, +) -> std::io::Result<(super::SegmentFetchContext, super::super::SegmentFetchSnapshot, super::super::SegmentFetchSnapshot)> +{ + let policy = SegmentFetchPolicy::default(); + let (worker, context, _) = fetch_context(server, directory, &policy).await; + install_startup(&context, &worker, shared::model::HlsStartupMode::Optimistic).await; + let first = + worker.next_fetch_snapshot(&context, 10, &policy).await.ok_or_else(|| std::io::Error::other("first"))?; + let second = + worker.next_fetch_snapshot(&context, 10, &policy).await.ok_or_else(|| std::io::Error::other("second"))?; + let head_seq = context.session.read().await.publishable_origin_head_proxy_seq; + let (head, successor) = if Some(first.proxy_seq) == head_seq { (first, second) } else { (second, first) }; + assert_eq!(Some(head.proxy_seq), head_seq); + let mut session = context.session.write().await; + let startup = session.startup.as_mut().ok_or_else(|| std::io::Error::other("startup policy"))?; + startup.config.max_progressive_bytes_per_segment = shared::model::Bytes::new(8); + drop(session); + Ok((context, head, successor)) +} + +#[tokio::test] +async fn optimistic_successor_binding_first_does_not_block_the_head_fallback() -> std::io::Result<()> { + let server = spawn_sequence_response_server(Vec::new()).await; + let directory = tempfile::tempdir()?; + let (context, head, successor) = optimistic_head_and_successor_snapshots(&server, &directory).await?; + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + let bound_successor = super::super::prepare_startup_fill(&context, &successor, Some(4), deadline) + .await + .map_err(|error| std::io::Error::other(format!("{error:?}")))?; + assert!(bound_successor.raw); + let prepared_head = super::super::prepare_startup_fill(&context, &head, Some(12), deadline) + .await + .map_err(|error| std::io::Error::other(format!("{error:?}")))?; + assert!(!prepared_head.raw); + assert_eq!(context.session.read().await.startup_mode(), shared::model::HlsStartupMode::FirstReady); + Ok(()) +} + +#[tokio::test] +async fn optimistic_successor_replay_rejection_keeps_the_session_policy() -> std::io::Result<()> { + let server = spawn_sequence_response_server(Vec::new()).await; + let directory = tempfile::tempdir()?; + let (context, head, successor) = optimistic_head_and_successor_snapshots(&server, &directory).await?; + let deadline = tokio::time::Instant::now() + Duration::from_secs(1); + let bound_successor = super::super::prepare_startup_fill(&context, &successor, Some(12), deadline) + .await + .map_err(|error| std::io::Error::other(format!("{error:?}")))?; + assert!(bound_successor.raw); + assert_eq!(context.session.read().await.startup_mode(), shared::model::HlsStartupMode::Optimistic); + let prepared_head = super::super::prepare_startup_fill(&context, &head, Some(4), deadline) + .await + .map_err(|error| std::io::Error::other(format!("{error:?}")))?; + assert!(prepared_head.raw); + Ok(()) +} diff --git a/backend/hls/src/segment_fetcher/tests/support.rs b/backend/hls/src/segment_fetcher/tests/support.rs index 42a32a45d..0ee1a9984 100644 --- a/backend/hls/src/segment_fetcher/tests/support.rs +++ b/backend/hls/src/segment_fetcher/tests/support.rs @@ -481,6 +481,54 @@ pub(in crate::segment_fetcher::tests) async fn prefix_origin( Ok((TestSegmentServer { base_url, requests, task }, release)) } +/// Origin whose head (`/1.ts`) sends a body prefix at once, while every +/// successor answers with headers only until `release` is set. +pub(in crate::segment_fetcher::tests) async fn headers_only_successor_origin( +) -> std::io::Result<(TestSegmentServer, tokio::sync::watch::Sender)> { + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await?; + let base_url = format!("http://{}", listener.local_addr()?); + let requests = Arc::new(Mutex::new(Vec::new())); + let recorded = Arc::clone(&requests); + let (release, released) = tokio::sync::watch::channel(false); + let task = tokio::spawn(async move { + let mut children = tokio::task::JoinSet::new(); + loop { + tokio::select! { + accepted = listener.accept() => { + let Ok((socket, _)) = accepted else { break; }; + let recorded = Arc::clone(&recorded); + let mut released = released.clone(); + children.spawn(async move { + let mut reader = BufReader::new(socket); + let mut request = String::new(); + tokio::io::AsyncBufReadExt::read_line(&mut reader, &mut request).await?; + let head = request.starts_with("GET /1.ts "); + recorded.lock().await.push(request); + loop { + let mut line = String::new(); + tokio::io::AsyncBufReadExt::read_line(&mut reader, &mut line).await?; + if line == "\r\n" || line.is_empty() { break; } + } + let mut socket = reader.into_inner(); + socket.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 12\r\nConnection: close\r\n\r\n").await?; + if head { + socket.write_all(b"segment-").await?; + } + while !*released.borrow_and_update() { + released.changed().await.map_err(std::io::Error::other)?; + } + socket.write_all(if head { b"body".as_slice() } else { b"segment-body".as_slice() }).await?; + socket.shutdown().await?; + Ok::<_, std::io::Error>(()) + }); + } + _ = children.join_next(), if !children.is_empty() => {} + } + } + }); + Ok((TestSegmentServer { base_url, requests, task }, release)) +} + pub(in crate::segment_fetcher::tests) type StartupSoakResources<'a> = ( &'a tempfile::TempDir, &'a Arc, @@ -581,7 +629,7 @@ pub(in crate::segment_fetcher::tests) async fn wait_for_cold_publication( let mut session = context.session.write().await; if let Ok(rendered) = session.render_and_store_manifest(current_time_millis()) { let head = session.startup.as_ref().and_then(|startup| startup.revisions.get(&1)).cloned(); - if mode == shared::model::HlsStartupMode::Progressive + if mode.streams_raw_revisions() || head.as_ref().is_some_and(|head| { matches!(*head.revision().subscribe().borrow(), crate::SegmentRevisionState::Complete(_)) }) diff --git a/backend/hls/src/segment_revision.rs b/backend/hls/src/segment_revision.rs index 48fd1ed3d..7d37ac25f 100644 --- a/backend/hls/src/segment_revision.rs +++ b/backend/hls/src/segment_revision.rs @@ -112,6 +112,15 @@ impl SegmentRevision { exposed } + /// An optimistic successor is listed before its body arrives. If it fails + /// before receiving any byte, a new attempt may take over its URI. + pub(crate) fn is_replaceable_in(&self, mode: HlsStartupMode) -> bool { + mode.publishes_pending_successors() + && matches!(*self.state.borrow(), SegmentRevisionState::Failed) + && !self.exposed.load(Ordering::Acquire) + && self.prefix_available.load(Ordering::Acquire) == 0 + } + pub(crate) fn replay_exhausted(&self) { self.state.send_if_modified(|state| { if !matches!(state, SegmentRevisionState::Pending) { @@ -144,11 +153,22 @@ impl SegmentRevision { pub fn is_startup_head_ready(&self, mode: HlsStartupMode) -> bool { match mode { HlsStartupMode::FirstReady => self.is_complete(), - HlsStartupMode::Progressive => self.is_publishable(), + HlsStartupMode::Progressive | HlsStartupMode::Optimistic => self.is_publishable(), HlsStartupMode::Conservative => false, } } + /// Whether this revision may be listed in a startup window. Optimistic + /// startup lists a successor as soon as the origin accepted it; delivery + /// of a successor always waits for its complete body. + pub fn is_window_publishable(&self, mode: HlsStartupMode) -> bool { + if mode.publishes_pending_successors() { + !matches!(*self.state.borrow(), SegmentRevisionState::Failed) + } else { + self.is_publishable() + } + } + pub fn is_publishable(&self) -> bool { match &*self.state.borrow() { SegmentRevisionState::Failed => false, @@ -296,6 +316,21 @@ mod tests { Ok(()) } + #[test] + fn optimistic_window_accepts_pending_successors_but_never_failed_ones() -> io::Result<()> { + let store = SegmentRevisionStore::default(); + let owner = store.create(ProxySessionId("window".into()), 1, SegmentRevisionKind::Raw)?; + let revision = owner.revision(); + assert!(revision.is_window_publishable(HlsStartupMode::Optimistic)); + assert!(!revision.is_window_publishable(HlsStartupMode::Progressive)); + assert!(!revision.is_startup_head_ready(HlsStartupMode::Optimistic)); + revision.prefix_available.store(1, Ordering::Release); + assert!(revision.is_startup_head_ready(HlsStartupMode::Optimistic)); + revision.fail(); + assert!(!revision.is_window_publishable(HlsStartupMode::Optimistic)); + Ok(()) + } + #[test] fn exposed_replay_exhaustion_is_terminal() -> io::Result<()> { let store = SegmentRevisionStore::default(); diff --git a/docs/src/configuration/shared-hls-configuration.md b/docs/src/configuration/shared-hls-configuration.md index 74d5f2b12..869539f5c 100644 --- a/docs/src/configuration/shared-hls-configuration.md +++ b/docs/src/configuration/shared-hls-configuration.md @@ -270,6 +270,13 @@ The complete session delivers immutable raw TS revisions, including subsequent r Repair and the corruption watchdog continue on the processed cache copy. Raw delivery does not correct source timestamp or transport-stream defects; prefer `first_ready` when a source needs repair. +`optimistic` behaves like `progressive` and publishes the initial playlist earlier. The head segment needs a +body prefix, while listed successors only need an accepted HTTP 200 origin response. Successor fetches start +together with the head, within the configured per-session concurrency. A client request for a successor waits +for its complete body. If the origin fails a successor after it was listed, only that segment fails for the +client; already published media never changes. Replay or disk fallback switches the session to `first_ready` +exactly like `progressive`. + Fast startup is limited to live, clear MPEG-TS objects without MAP or origin byte ranges. Archive, encrypted, fragmented-MP4 and transient sessions use their existing path. Before publication an incompatible source can fall back to that path. After a Fast Start manifest is published, a refresh requiring transient delivery is rejected to preserve @@ -284,17 +291,18 @@ disk cache limits. Conservative behavior remains the default; existing sessions reload. Restart after changing to `conservative` when an immediate rollback of running sessions is required. Existing provider and fetch concurrency limits remain in effect. The initial playlist needs at least three -publishable segments. With fewer than three concurrent fills, a cold session can still wait for an earlier -origin EOF before that window is available. Raising the existing per-session concurrency limit to three is an -operator choice and must fit the provider's connection allowance; startup mode never raises it automatically. +publishable segments; `optimistic` counts successors the origin has accepted. With fewer than three concurrent +fills, a cold session can still wait for an earlier origin EOF before that window is available. Raising the +existing per-session concurrency limit to three is an operator choice and must fit the provider's connection +allowance; startup mode never raises it automatically. Measure both first data and decoded-picture time before enabling progressive mode in production. Short successful response runs do not establish long-term player, origin-failure or resource stability. ## Reproducible startup checks -The testkit has isolated shared-HLS scenarios for `conservative`, `first_ready` and `progressive`, plus missing-head -scenarios for both fast modes. Existing scenarios retain their original origin behavior. The new +The testkit has isolated shared-HLS scenarios for `conservative`, `first_ready`, `progressive` and `optimistic`, plus +missing-head scenarios for every fast mode. Existing scenarios retain their original origin behavior. The new `origin.hls_profile` publishes six segments, sends a complete synthetic frame immediately, and delays the second frame by `prefix_delay_millis`. `missing_head_segments` returns alternating HTTP 404/410 at the head; `abort_segments` injects a body error after the prefix. `unknown_length` omits origin Content-Length. diff --git a/frontend/public/assets/i18n/ar.json b/frontend/public/assets/i18n/ar.json index e49fd3224..9a2a812aa 100644 --- a/frontend/public/assets/i18n/ar.json +++ b/frontend/public/assets/i18n/ar.json @@ -916,6 +916,7 @@ "HLS_STARTUP_MODE_CONSERVATIVE": "محافظ", "HLS_STARTUP_MODE_FIRST_READY": "أول جاهز", "HLS_STARTUP_MODE_PROGRESSIVE": "تدريجي", + "HLS_STARTUP_MODE_OPTIMISTIC": "متفائل", "BACK": "رجوع", "BOUQUETS": "الباقات", "ALL_TARGETS": "جميع الأهداف", diff --git a/frontend/public/assets/i18n/en.json b/frontend/public/assets/i18n/en.json index 612a8cce3..abe10f21c 100644 --- a/frontend/public/assets/i18n/en.json +++ b/frontend/public/assets/i18n/en.json @@ -916,6 +916,7 @@ "HLS_STARTUP_MODE_CONSERVATIVE": "Conservative", "HLS_STARTUP_MODE_FIRST_READY": "First ready", "HLS_STARTUP_MODE_PROGRESSIVE": "Progressive", + "HLS_STARTUP_MODE_OPTIMISTIC": "Optimistic", "ALL_TARGETS": "All targets", "ACCEPT_INSECURE_SSL_CERTIFICATES": "Accept Insecure SSL Certificates", "ACTIVE": "Active", diff --git a/frontend/public/assets/i18n/es.json b/frontend/public/assets/i18n/es.json index 6bce949af..6354039d6 100644 --- a/frontend/public/assets/i18n/es.json +++ b/frontend/public/assets/i18n/es.json @@ -916,6 +916,7 @@ "HLS_STARTUP_MODE_CONSERVATIVE": "Conservador", "HLS_STARTUP_MODE_FIRST_READY": "Primero listo", "HLS_STARTUP_MODE_PROGRESSIVE": "Progresivo", + "HLS_STARTUP_MODE_OPTIMISTIC": "Optimista", "ALL_TARGETS": "Todos los destinos", "ACCEPT_INSECURE_SSL_CERTIFICATES": "Aceptar certificados SSL no seguros", "ACTIVE": "Activo", diff --git a/frontend/public/assets/i18n/ru.json b/frontend/public/assets/i18n/ru.json index 4ed56f1f6..bf64455a1 100644 --- a/frontend/public/assets/i18n/ru.json +++ b/frontend/public/assets/i18n/ru.json @@ -916,6 +916,7 @@ "HLS_STARTUP_MODE_CONSERVATIVE": "Консервативный", "HLS_STARTUP_MODE_FIRST_READY": "Первый готовый", "HLS_STARTUP_MODE_PROGRESSIVE": "Прогрессивный", + "HLS_STARTUP_MODE_OPTIMISTIC": "Оптимистичный", "BACK": "Назад", "BOUQUETS": "Букеты", "ALL_TARGETS": "Все цели", diff --git a/frontend/src/app/components/config/reverse_proxy_config_view.rs b/frontend/src/app/components/config/reverse_proxy_config_view.rs index d6a5940c5..ff8de152c 100644 --- a/frontend/src/app/components/config/reverse_proxy_config_view.rs +++ b/frontend/src/app/components/config/reverse_proxy_config_view.rs @@ -1000,6 +1000,11 @@ pub fn ReverseProxyConfigView() -> Html { html! { translate.t("LABEL.HLS_STARTUP_MODE_PROGRESSIVE") }, selected_startup == HlsStartupMode::Progressive, ), + DropDownOption::new( + "optimistic", + html! { translate.t("LABEL.HLS_STARTUP_MODE_OPTIMISTIC") }, + selected_startup == HlsStartupMode::Optimistic, + ), ]); let set_startup = { let hls_cache_state = hls_cache_state.clone(); diff --git a/shared/src/model/config/hls/startup.rs b/shared/src/model/config/hls/startup.rs index b6aef504c..240f7f0ae 100644 --- a/shared/src/model/config/hls/startup.rs +++ b/shared/src/model/config/hls/startup.rs @@ -13,14 +13,24 @@ pub enum HlsStartupMode { Conservative, FirstReady, Progressive, + Optimistic, } crate::impl_str_enum!(HlsStartupMode, "HLS startup mode", Conservative => "conservative", FirstReady => "first_ready", Progressive => "progressive", + Optimistic => "optimistic", ); +impl HlsStartupMode { + /// Raw revisions stream to the first viewer before origin EOF. + pub const fn streams_raw_revisions(self) -> bool { matches!(self, Self::Progressive | Self::Optimistic) } + + /// Startup successors are publishable once the origin accepted them, before body bytes arrive. + pub const fn publishes_pending_successors(self) -> bool { matches!(self, Self::Optimistic) } +} + #[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Eq)] #[serde(default, deny_unknown_fields)] pub struct HlsStartupConfigDto { diff --git a/shared/src/model/config/hls/startup_config_tests.rs b/shared/src/model/config/hls/startup_config_tests.rs index 1a638366a..4d54c37ed 100644 --- a/shared/src/model/config/hls/startup_config_tests.rs +++ b/shared/src/model/config/hls/startup_config_tests.rs @@ -6,7 +6,12 @@ fn startup_defaults_and_modes_roundtrip() -> Result<(), Box "conservative", Self::FirstReady => "first_ready", Self::Progressive => "progressive", + Self::Optimistic => "optimistic", }) } } diff --git a/tools/tuliprox-testkit/tests/hls_fast_start_process.rs b/tools/tuliprox-testkit/tests/hls_fast_start_process.rs index 6ccfa159b..0c0673a8c 100644 --- a/tools/tuliprox-testkit/tests/hls_fast_start_process.rs +++ b/tools/tuliprox-testkit/tests/hls_fast_start_process.rs @@ -68,7 +68,7 @@ async fn run(mode: StartupMode, missing: u32, abort: bool, unknown_length: bool) let client = reqwest::Client::new(); let playlist_url = format!("{}/m3u?username=testkit&password={}", fixture.base_url, fixture_password()); let urls = discover_playlist(&client, &playlist_url).await?; - let progressive = matches!(mode, StartupMode::Progressive); + let progressive = matches!(mode, StartupMode::Progressive | StartupMode::Optimistic); let result = probe( urls.playback_url(17)?, &RunId::new(&run_id), @@ -150,3 +150,13 @@ async fn progressive_known_length_drop_is_not_successful_eof() -> Result<(), Tes async fn progressive_chunked_drop_is_not_successful_eof() -> Result<(), TestkitError> { run(StartupMode::Progressive, 0, true, true).await } +#[tokio::test] +#[ignore = "requires built SUT via TULIPROX_TESTKIT_SUT_BINARY"] +async fn optimistic_delivers_prefix_before_origin_eof() -> Result<(), TestkitError> { + run(StartupMode::Optimistic, 0, false, false).await +} +#[tokio::test] +#[ignore = "requires built SUT via TULIPROX_TESTKIT_SUT_BINARY"] +async fn optimistic_skips_404_and_410_head() -> Result<(), TestkitError> { + run(StartupMode::Optimistic, 2, false, false).await +}