feat(hls): add optimistic startup mode (#914)

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.

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;

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;

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<u64, crate::SegmentRevisionGuard>| {
+        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<u64, crate::SegmentRevisionGuard>| {
+        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(())
+}

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);

backend/hls/src/renderer.rs

@@ -225,7 +225,10 @@ pub fn renderer_candidate_window_proxy_seqs(session: &HlsSession) -> Vec<u64> {
 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<Vec<u64>> {
     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<HlsSession> {
+        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::<Vec<_>>() {
+            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<crate::SegmentRevisionGuard> {
+        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 };

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(())
+}

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)
             })

backend/hls/src/segment_fetcher/startup_fill.rs

@@ -244,7 +244,7 @@ async fn try_prepare_startup_fill(
     expected: Option<u64>,
     deadline: Instant,
 ) -> Result<Option<PreparedStartupFill>, 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() {

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,
 };

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(())
+}

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<bool>)> {
+    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<crate::SegmentRevisionStore>,
@@ -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(_))
                 })

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();

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.

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": "جميع الأهداف",

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",

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",

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": "Все цели",

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();

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 {

shared/src/model/config/hls/startup_config_tests.rs

@@ -6,7 +6,12 @@ fn startup_defaults_and_modes_roundtrip() -> Result<(), Box<dyn std::error::Erro
     legacy.prepare()?;
     assert_eq!(legacy.startup.mode, HlsStartupMode::Conservative);
     assert!(serde_json::to_value(&legacy)?.get("startup").is_none());
-    for mode in [HlsStartupMode::Conservative, HlsStartupMode::FirstReady, HlsStartupMode::Progressive] {
+    for mode in [
+        HlsStartupMode::Conservative,
+        HlsStartupMode::FirstReady,
+        HlsStartupMode::Progressive,
+        HlsStartupMode::Optimistic,
+    ] {
         let startup = HlsStartupConfigDto { mode, max_progressive_segments: 7, ..Default::default() };
         startup.prepare()?;
         let serialized = serde_json::to_string(&startup)?;
@@ -30,3 +35,17 @@ fn invalid_startup_limits_are_rejected() {
         assert!(config.prepare().is_err());
     }
 }
+
+#[test]
+fn startup_mode_capabilities() {
+    use std::str::FromStr;
+    assert_eq!(HlsStartupMode::from_str("optimistic").ok(), Some(HlsStartupMode::Optimistic));
+    assert!(HlsStartupMode::Optimistic.streams_raw_revisions());
+    assert!(HlsStartupMode::Progressive.streams_raw_revisions());
+    assert!(!HlsStartupMode::FirstReady.streams_raw_revisions());
+    assert!(!HlsStartupMode::Conservative.streams_raw_revisions());
+    assert!(HlsStartupMode::Optimistic.publishes_pending_successors());
+    for mode in [HlsStartupMode::Conservative, HlsStartupMode::FirstReady, HlsStartupMode::Progressive] {
+        assert!(!mode.publishes_pending_successors());
+    }
+}

test/fixtures/testkit/scenarios/m3u-hls-share-optimistic-missing-head.yml

@@ -0,0 +1,50 @@
+schema_version: 1
+name: m3u-hls-share-optimistic-missing-head
+tuliprox:
+  base_url: http://fixture.invalid
+  execution_mode: isolated_fixture
+  fixture_stream:
+    share_live_hls: true
+    share_live_mpeg_ts: false
+    hls_startup_mode: optimistic
+  bootstrap:
+    command: "${env:TULIPROX_TESTKIT_SUT_BINARY}"
+    arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"]
+    readiness_url: "{base_url}/healthcheck"
+origin:
+  run_id: m3u-hls-share-optimistic-missing-head
+  hls_profile:
+    prefix_delay_millis: 1000
+    window_segments: 6
+    missing_head_segments: 2
+actors:
+  - id: viewer
+    agent: local
+    username: viewer
+    user_agent: tuliprox-testkit/viewer
+  - id: second-viewer
+    agent: local
+    username: other
+    user_agent: tuliprox-testkit/second-viewer
+channels:
+  shared-channel: { origin_marker: 17, protocol: hls }
+steps:
+  - command_id: first-viewer-starts
+    start: { actor: viewer, playback_id: first, session_group: first, channel: shared-channel }
+    await: { valid_frames: 2 }
+    assert_origin: { no_limit_rejections: true }
+  - command_id: second-viewer-joins
+    start: { actor: second-viewer, playback_id: second, session_group: second, channel: shared-channel }
+    await: { valid_frames: 2 }
+    assert_origin: { no_limit_rejections: true }
+  - command_id: first-viewer-stops
+    stop: { playback_id: first }
+  - command_id: second-viewer-stops
+    stop: { playback_id: second }
+    assert_runtime: { active_users: 0, active_user_connections: 0 }
+policy_contract:
+  user_access_control: true
+  users:
+    viewer: { max_connections: 4 }
+    other: { max_connections: 4 }
+  provider_max_connections: 8

test/fixtures/testkit/scenarios/m3u-hls-share-optimistic-startup.yml

@@ -0,0 +1,50 @@
+schema_version: 1
+name: m3u-hls-share-optimistic-startup
+tuliprox:
+  base_url: http://fixture.invalid
+  execution_mode: isolated_fixture
+  fixture_stream:
+    share_live_hls: true
+    share_live_mpeg_ts: false
+    hls_startup_mode: optimistic
+  bootstrap:
+    command: "${env:TULIPROX_TESTKIT_SUT_BINARY}"
+    arguments: ["-s", "-c", "{config_file}", "-i", "{source_file}", "-a", "{api_proxy_file}"]
+    readiness_url: "{base_url}/healthcheck"
+origin:
+  run_id: m3u-hls-share-optimistic-startup
+  hls_profile:
+    prefix_delay_millis: 1000
+    window_segments: 6
+    missing_head_segments: 0
+actors:
+  - id: viewer
+    agent: local
+    username: viewer
+    user_agent: tuliprox-testkit/viewer
+  - id: second-viewer
+    agent: local
+    username: other
+    user_agent: tuliprox-testkit/second-viewer
+channels:
+  shared-channel: { origin_marker: 17, protocol: hls }
+steps:
+  - command_id: first-viewer-starts
+    start: { actor: viewer, playback_id: first, session_group: first, channel: shared-channel }
+    await: { valid_frames: 2 }
+    assert_origin: { no_limit_rejections: true }
+  - command_id: second-viewer-joins
+    start: { actor: second-viewer, playback_id: second, session_group: second, channel: shared-channel }
+    await: { valid_frames: 2 }
+    assert_origin: { no_limit_rejections: true }
+  - command_id: first-viewer-stops
+    stop: { playback_id: first }
+  - command_id: second-viewer-stops
+    stop: { playback_id: second }
+    assert_runtime: { active_users: 0, active_user_connections: 0 }
+policy_contract:
+  user_access_control: true
+  users:
+    viewer: { max_connections: 4 }
+    other: { max_connections: 4 }
+  provider_max_connections: 8

tools/tuliprox-testkit/src/hls_probe.rs

@@ -15,6 +15,7 @@ pub enum StartupMode {
     Conservative,
     FirstReady,
     Progressive,
+    Optimistic,
 }
 
 impl fmt::Display for StartupMode {
@@ -23,6 +24,7 @@ impl fmt::Display for StartupMode {
             Self::Conservative => "conservative",
             Self::FirstReady => "first_ready",
             Self::Progressive => "progressive",
+            Self::Optimistic => "optimistic",
         })
     }
 }

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
+}