Fix/catchup (#778)

* Fix: setting auth web_ui token_ttl_mins = 0 fixed.
* Fix: catchup
* New: added playlist update log view
* Fix: Landing Page
New: "Last Page"  added for Landing Page selection

backend/src/api/model/active_user_manager.rs

@@ -128,6 +128,7 @@ pub struct UserSession {
     pub virtual_id: u32,
     pub provider: Arc<str>,
     pub stream_url: Arc<str>,
+    pub provider_session_headers: HashMap<String, String>,
     pub addr: SocketAddr,
     pub socket_bound: bool,
     pub active_addrs: Vec<SocketAddr>,
@@ -662,6 +663,14 @@ impl ActiveUserManager {
         key
     }
 
+    fn admission_gate_key(username: &str) -> String {
+        let mut key = String::with_capacity(username.len() + 11);
+        key.push_str("admission");
+        key.push('\0');
+        key.push_str(username);
+        key
+    }
+
     fn cleanup_idle_transition_gates(transition_gates: &mut HashMap<String, Arc<Mutex<()>>>) {
         transition_gates.retain(|_, gate| Arc::strong_count(gate) > 1);
     }
@@ -680,6 +689,20 @@ impl ActiveUserManager {
         gate.lock_owned().await
     }
 
+    pub(crate) async fn acquire_user_admission(&self, username: &str) -> tokio::sync::OwnedMutexGuard<()> {
+        let key = Self::admission_gate_key(username);
+        let gate = {
+            let mut transition_gates = self.transition_gates.lock().await;
+            Self::cleanup_idle_transition_gates(&mut transition_gates);
+            Arc::clone(
+                transition_gates
+                    .entry(key)
+                    .or_insert_with(|| Arc::new(Mutex::new(()))),
+            )
+        };
+        gate.lock_owned().await
+    }
+
     fn should_reuse_stream_for_session(existing_stream: &StreamInfo, incoming_channel: &StreamChannel) -> bool {
         existing_stream.channel.item_type.requires_provider_affinity() || incoming_channel.item_type.requires_provider_affinity()
     }
@@ -1586,6 +1609,7 @@ impl ActiveUserManager {
             virtual_id: params.virtual_id,
             provider: params.provider.intern(),
             stream_url: params.stream_url.intern(),
+            provider_session_headers: HashMap::new(),
             addr: *params.addr,
             socket_bound: params.socket_bound,
             active_addrs: vec![*params.addr],
@@ -1961,11 +1985,17 @@ impl ActiveUserManager {
                 session.socket_bound = socket_bound;
                 remember_session_addr(session, *addr);
                 Self::bump_session_transition_version(session);
+                let mut reset_provider_session_headers = false;
                 if &*session.stream_url != stream_url {
                     session.stream_url = stream_url.intern();
+                    reset_provider_session_headers = true;
                 }
                 if &*session.provider != provider {
                     session.provider = provider.intern();
+                    reset_provider_session_headers = true;
+                }
+                if reset_provider_session_headers {
+                    session.provider_session_headers.clear();
                 }
                 // Normalize stale lifecycle states on session refresh.
                 // Expired, PendingProvider, and Preserved sessions cannot stay in those states
@@ -2673,6 +2703,24 @@ impl ActiveUserManager {
         self.update_user_session(username, token).await
     }
 
+    pub async fn update_session_provider_headers(
+        &self,
+        username: &str,
+        token: &str,
+        provider_session_headers: &HashMap<String, String>,
+    ) -> bool {
+        let mut user_connections = self.connections.write().await;
+        let Some(connection_data) = user_connections.by_key.get_mut(username) else {
+            return false;
+        };
+        let Some(session) = connection_data.sessions.iter_mut().find(|session| session.token == token) else {
+            return false;
+        };
+        session.provider_session_headers.clone_from(provider_session_headers);
+        session.ts = current_time_secs();
+        true
+    }
+
     pub async fn pending_provider_version(&self, username: &str, token: &str) -> Option<u64> {
         let user_connections = self.connections.read().await;
         let connection_data = user_connections.by_key.get(username)?;
@@ -3258,7 +3306,7 @@ mod tests {
         model::{PlaylistItemType, StreamChannel, StreamInfo, XtreamCluster},
         utils::Internable,
     };
-    use std::{borrow::Cow, sync::Arc};
+    use std::{borrow::Cow, collections::HashMap, sync::Arc};
     use shared::model::ProxyType;
 
     fn test_channel(virtual_id: u32) -> StreamChannel {
@@ -3326,6 +3374,7 @@ mod tests {
                 virtual_id: 7001,
                 provider: "provider-a".intern(),
                 stream_url: "http://localhost/live.m3u8".intern(),
+                provider_session_headers: HashMap::new(),
                 addr,
                 socket_bound: false,
                 active_addrs: vec![addr],
@@ -3387,6 +3436,7 @@ mod tests {
                 virtual_id: 7002,
                 provider: "provider-a".intern(),
                 stream_url: "http://localhost/live.m3u8".intern(),
+                provider_session_headers: HashMap::new(),
                 addr,
                 socket_bound: false,
                 active_addrs: vec![addr],
@@ -3430,6 +3480,134 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn update_session_provider_headers_updates_existing_session_and_timestamp() {
+        let config = Config::default();
+        let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
+        let event_manager = Arc::new(EventManager::new());
+        let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
+
+        let addr: SocketAddr = "127.0.0.1:55402".parse().unwrap_or_else(|_| unreachable!());
+        let mut user = ProxyUserCredentials::default();
+        user.username = "user-provider-headers".to_string();
+
+        manager
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-provider-headers",
+                virtual_id: 7003,
+                provider: "provider-a",
+                stream_url: "http://localhost/live.m3u8",
+                addr: &addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+
+        let before = manager
+            .get_and_update_user_session(&user.username, "tok-provider-headers")
+            .await
+            .expect("session should exist");
+        let previous_ts = before.ts;
+        let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]);
+
+        assert!(manager
+            .update_session_provider_headers(&user.username, "tok-provider-headers", &headers)
+            .await);
+
+        let after = manager
+            .get_and_update_user_session(&user.username, "tok-provider-headers")
+            .await
+            .expect("session should exist");
+        assert_eq!(after.provider_session_headers, headers);
+        assert!(after.ts >= previous_ts);
+    }
+
+    #[tokio::test]
+    async fn update_session_provider_headers_returns_false_for_missing_user_or_token() {
+        let config = Config::default();
+        let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
+        let event_manager = Arc::new(EventManager::new());
+        let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
+        let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]);
+
+        assert!(!manager
+            .update_session_provider_headers("missing-user", "missing-token", &headers)
+            .await);
+
+        let addr: SocketAddr = "127.0.0.1:55403".parse().unwrap_or_else(|_| unreachable!());
+        let mut user = ProxyUserCredentials::default();
+        user.username = "user-missing-token".to_string();
+        manager
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-existing",
+                virtual_id: 7004,
+                provider: "provider-a",
+                stream_url: "http://localhost/live.m3u8",
+                addr: &addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+
+        assert!(!manager
+            .update_session_provider_headers(&user.username, "tok-missing", &headers)
+            .await);
+    }
+
+    #[tokio::test]
+    async fn create_user_session_clears_provider_headers_when_provider_or_stream_url_changes() {
+        let config = Config::default();
+        let geoip = Arc::new(ArcSwapOption::<GeoIp>::default());
+        let event_manager = Arc::new(EventManager::new());
+        let manager = ActiveUserManager::new(&config, &geoip, &event_manager);
+
+        let addr: SocketAddr = "127.0.0.1:55404".parse().unwrap_or_else(|_| unreachable!());
+        let mut user = ProxyUserCredentials::default();
+        user.username = "user-provider-header-reset".to_string();
+        let headers = HashMap::from([(String::from("cookie"), String::from("sid=abc"))]);
+
+        manager
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-reset",
+                virtual_id: 7005,
+                provider: "provider-a",
+                stream_url: "http://localhost/live-a.m3u8",
+                addr: &addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+        assert!(manager
+            .update_session_provider_headers(&user.username, "tok-reset", &headers)
+            .await);
+
+        manager
+            .create_user_session(CreateUserSessionParams {
+                user: &user,
+                session_token: "tok-reset",
+                virtual_id: 7005,
+                provider: "provider-b",
+                stream_url: "http://localhost/live-b.m3u8",
+                addr: &addr,
+                connection_permission: UserConnectionPermission::Allowed,
+                connection_kind: Some(ConnectionKind::Normal),
+                socket_bound: false,
+            })
+            .await;
+
+        let session = manager
+            .get_and_update_user_session(&user.username, "tok-reset")
+            .await
+            .expect("session should exist");
+        assert!(session.provider_session_headers.is_empty());
+    }
+
     /// `terminate_session` expires a session and removes it.
     #[tokio::test]
     async fn terminate_session_expires_and_removes_session() {
@@ -7787,6 +7965,7 @@ mod tests {
                 virtual_id: 9002,
                 provider: "provider-a".intern(),
                 stream_url: "http://localhost/stream.ts".intern(),
+                provider_session_headers: HashMap::new(),
                 addr,
                 socket_bound: false,
                 active_addrs: vec![addr],
@@ -7828,6 +8007,7 @@ mod tests {
                 virtual_id: 9003,
                 provider: "provider-a".intern(),
                 stream_url: "http://localhost/stream.ts".intern(),
+                provider_session_headers: HashMap::new(),
                 addr,
                 socket_bound: false,
                 active_addrs: vec![addr],
@@ -7902,6 +8082,7 @@ mod tests {
                 virtual_id: 9004,
                 provider: "provider-a".intern(),
                 stream_url: "http://localhost/stream.ts".intern(),
+                provider_session_headers: HashMap::new(),
                 addr,
                 socket_bound: false,
                 active_addrs: vec![addr],

backend/src/api/model/app_state.rs

@@ -390,6 +390,11 @@ fn video_download_changed(a: &crate::model::VideoDownloadConfig, b: &crate::mode
     VideoDownloadConfigDto::from(a) != VideoDownloadConfigDto::from(b)
 }
 
+#[derive(Clone)]
+pub struct ManualPlaylistUpdateRequest {
+    pub targets: Arc<ProcessTargets>,
+}
+
 #[derive(Clone)]
 pub struct AppState {
     pub forced_targets: Arc<ArcSwap<ProcessTargets>>, // as program arguments
@@ -412,7 +417,7 @@ pub struct AppState {
     /// `try_send` deduplicates rapid clicks: if an update is already pending
     /// or the channel is full, the request is silently dropped so at most one
     /// update is queued at any time regardless of how many times the button is clicked.
-    pub manual_update_sender: mpsc::Sender<Arc<ProcessTargets>>,
+    pub manual_update_sender: mpsc::Sender<ManualPlaylistUpdateRequest>,
 }
 
 impl AppState {

backend/src/api/model/event_manager.rs

@@ -1,15 +1,12 @@
 use crate::api::model::streams::{MeterReading, StreamMeterHandle};
 use log::trace;
 use shared::model::{
-    ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanSummary, PlaylistUpdateState,
-    StreamMeterEntry, SystemInfo,
+    ActiveUserConnectionChange, ConfigType, DownloadsDelta, DownloadsResponse, LibraryScanProgressEvent,
+    PlaylistUpdateProgressEvent, PlaylistUpdateState, StreamMeterEntry, SystemInfo,
 };
 use std::{
     collections::HashMap,
-    sync::{
-        atomic::{AtomicUsize, Ordering},
-        Arc,
-    },
+    sync::{atomic::AtomicUsize, atomic::Ordering, Arc},
     time::Duration,
 };
 use tokio::sync::RwLock;
@@ -26,9 +23,9 @@ pub enum EventMessage {
     ActiveProvider(Arc<str>, usize),
     ConfigChange(ConfigType),
     PlaylistUpdate(PlaylistUpdateState),
-    PlaylistUpdateProgress(String, String),
+    PlaylistUpdateProgress(PlaylistUpdateProgressEvent),
     SystemInfoUpdate(SystemInfo),
-    LibraryScanProgress(LibraryScanSummary),
+    LibraryScanProgress(LibraryScanProgressEvent),
     DownloadsUpdate(DownloadsResponse),
     DownloadsDeltaUpdate(DownloadsDelta),
     InputMetadataUpdatesCompleted(Arc<str>),

backend/src/api/model/stream.rs

@@ -31,6 +31,7 @@ pub struct StreamDetails {
     pub(crate) stream_info: ProviderStreamInfo,
     pub provider_name: Option<Arc<str>>,
     pub request_url: Option<Arc<str>>,
+    pub session_headers: Option<HashMap<String, String>>,
     pub grace_period: GracePeriodOptions,
     pub provider_grace_active: bool,
     pub disable_provider_grace: bool,
@@ -51,6 +52,7 @@ impl Clone for StreamDetails {
             stream_info: self.stream_info.clone(),
             provider_name: self.provider_name.clone(),
             request_url: self.request_url.clone(),
+            session_headers: self.session_headers.clone(),
             grace_period: self.grace_period,
             provider_grace_active: self.provider_grace_active,
             disable_provider_grace: self.disable_provider_grace,
@@ -68,6 +70,7 @@ impl StreamDetails {
             stream_info: None,
             provider_name: None,
             request_url: None,
+            session_headers: None,
             grace_period: grace_period_options,
             provider_grace_active: false,
             disable_provider_grace: false,

backend/src/api/model/streams/active_client_stream.rs

@@ -462,6 +462,7 @@ fn create_deferred_provider_open_future(
             stream_url: &stream_url,
             req_headers,
             input_headers: Some(&input.headers),
+            session_headers: stream_details.session_headers.as_ref(),
             disabled_headers: disabled_headers.as_ref(),
             default_user_agent: default_user_agent.as_deref(),
             username: None,
@@ -1370,7 +1371,7 @@ mod tests {
 
         let tokens = CancelTokens::default();
         let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone()));
-        let (manual_update_sender, _) = mpsc::channel::<Arc<ProcessTargets>>(1);
+        let (manual_update_sender, _) = mpsc::channel::<crate::api::model::ManualPlaylistUpdateRequest>(1);
 
         Arc::new(AppState {
             forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets {
@@ -1437,7 +1438,7 @@ mod tests {
 
         let tokens = CancelTokens::default();
         let metadata_manager = Arc::new(MetadataUpdateManager::new(tokens.metadata.clone()));
-        let (manual_update_sender, _) = mpsc::channel::<Arc<ProcessTargets>>(1);
+        let (manual_update_sender, _) = mpsc::channel::<crate::api::model::ManualPlaylistUpdateRequest>(1);
 
         Arc::new(AppState {
             forced_targets: Arc::new(ArcSwap::from_pointee(ProcessTargets {
@@ -1511,6 +1512,7 @@ mod tests {
             stream_info: None,
             provider_name: Some(Arc::clone(provider_name)),
             request_url: Some("http://provider-1.example/live/1".intern()),
+            session_headers: None,
             grace_period: GracePeriodOptions {
                 period_millis: 100,
                 timeout_secs: 0,
@@ -2361,6 +2363,7 @@ mod tests {
             stream_info: None,
             provider_name: Some(provider_name),
             request_url: Some("http://provider-1.example/live/2.ts".intern()),
+            session_headers: None,
             grace_period: GracePeriodOptions {
                 period_millis: 100,
                 timeout_secs: 0,

backend/src/api/model/streams/provider_stream_factory.rs

@@ -56,7 +56,7 @@ create_bitset!(
 #[derive(Debug, Clone)]
 pub struct ProviderStreamFactoryOptions {
     addr: SocketAddr,
-    // item_type: PlaylistItemType,
+    item_type: PlaylistItemType,
     flags: ProviderStreamFactoryFlagsSet,
     buffer_size: usize,
     url: Url,
@@ -80,6 +80,7 @@ pub(crate) struct ProviderStreamFactoryParams<'a> {
     pub stream_url: &'a Url,
     pub req_headers: &'a HeaderMap,
     pub input_headers: Option<&'a HashMap<String, String>>,
+    pub session_headers: Option<&'a HashMap<String, String>>,
     pub disabled_headers: Option<&'a ReverseProxyDisabledHeaderConfig>,
     pub default_user_agent: Option<&'a str>,
     pub username: Option<&'a str>,
@@ -98,6 +99,7 @@ impl ProviderStreamFactoryOptions {
             stream_url,
             req_headers,
             input_headers,
+            session_headers,
             disabled_headers,
             default_user_agent,
             username,
@@ -115,8 +117,15 @@ impl ProviderStreamFactoryOptions {
         let requested_range = get_request_range_start_bytes(&req_headers);
         req_headers.remove("range");
 
+        let merged_input_headers = merge_provider_request_headers(*input_headers, *session_headers);
+
         // We merge configured input headers with the headers from the request.
-        let headers = get_request_headers(*input_headers, Some(&req_headers), *disabled_headers, *default_user_agent);
+        let headers = get_request_headers(
+            merged_input_headers.as_ref(),
+            Some(&req_headers),
+            *disabled_headers,
+            *default_user_agent,
+        );
 
         let default_user_agent = default_user_agent
             .and_then(|ua| {
@@ -151,7 +160,7 @@ impl ProviderStreamFactoryOptions {
         }
 
         Self {
-            // item_type,
+            item_type: *item_type,
             addr: *addr,
             flags,
             buffer_size,
@@ -194,6 +203,9 @@ impl ProviderStreamFactoryOptions {
     #[inline]
     pub fn get_url_as_str(&self) -> &str { self.url.as_str() }
 
+    #[inline]
+    fn get_item_type(&self) -> PlaylistItemType { self.item_type }
+
     #[inline]
     pub fn should_retry_provider_request(&self) -> bool { self.flags.contains(ProviderStreamFactoryFlags::RetryEnabled) }
 
@@ -244,6 +256,23 @@ impl ProviderStreamFactoryOptions {
 
 }
 
+fn merge_provider_request_headers(
+    input_headers: Option<&HashMap<String, String>>,
+    session_headers: Option<&HashMap<String, String>>,
+) -> Option<HashMap<String, String>> {
+    match (input_headers, session_headers) {
+        (None, None) => None,
+        (Some(headers), None) | (None, Some(headers)) => Some(headers.clone()),
+        (Some(input), Some(session)) => {
+            let mut merged = input.clone();
+            for (key, value) in session {
+                merged.insert(key.clone(), value.clone());
+            }
+            Some(merged)
+        }
+    }
+}
+
 fn record_provider_open_failure(
     app_state: &Arc<AppState>,
     stream_options: &ProviderStreamFactoryOptions,
@@ -286,6 +315,17 @@ fn classify_provider_status_error(status: StatusCode) -> &'static str {
     }
 }
 
+fn provider_content_type_looks_like_html(headers: &HeaderMap) -> bool {
+    headers
+        .get(reqwest::header::CONTENT_TYPE)
+        .and_then(|value| value.to_str().ok())
+        .is_some_and(|value| value.split(';').next().unwrap_or_default().trim().eq_ignore_ascii_case("text/html"))
+}
+
+fn should_reject_success_response_content_type(item_type: PlaylistItemType, headers: &HeaderMap) -> bool {
+    !item_type.is_live_adaptive() && provider_content_type_looks_like_html(headers)
+}
+
 #[derive(Clone, Copy, Debug)]
 enum ProviderStreamRequestFailure {
     Status {
@@ -588,6 +628,17 @@ async fn provider_stream_request(
                 debug!("{}", sanitize_sensitive_info(&message));
             }
             if status.is_success() {
+                if should_reject_success_response_content_type(stream_options.get_item_type(), response.headers()) {
+                    debug!(
+                        "Provider returned HTML content for non-adaptive stream {}",
+                        sanitize_sensitive_info(stream_options.get_log_url().as_ref())
+                    );
+                    return Err(ProviderStreamRequestFailure::Status {
+                        status: StatusCode::BAD_GATEWAY,
+                        provider_error_class: "unexpected_content_type",
+                        serve_channel_unavailable: true,
+                    });
+                }
                 let response_info = {
                     // Unfortunately, the HEAD request does not work, so we need this workaround.
                     // We need some header information from the provider, we extract the necessary headers and forward them to the client
@@ -846,6 +897,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers,
             default_user_agent: None,
             username: None,
@@ -866,6 +918,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers,
             default_user_agent: None,
             username: None,
@@ -886,6 +939,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers,
             default_user_agent: None,
             username: None,
@@ -907,6 +961,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers,
             default_user_agent: None,
             username: None,
@@ -934,6 +989,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers: None,
             default_user_agent: None,
             username: None,
@@ -950,6 +1006,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers: None,
             default_user_agent: None,
             username: None,
@@ -980,6 +1037,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers: None,
             default_user_agent: None,
             username: None,
@@ -1010,6 +1068,7 @@ mod tests {
             stream_url: &stream_url,
             req_headers: &req_headers,
             input_headers: None,
+            session_headers: None,
             disabled_headers: None,
             default_user_agent: None,
             username: Some("alice"),
@@ -1042,4 +1101,56 @@ mod tests {
         assert_eq!(info.channel.input_name.as_ref(), "input-a");
         assert_eq!(info.channel.virtual_id, 77);
     }
+
+    #[test]
+    fn html_content_type_is_rejected_for_catchup_streams() {
+        let mut headers = HeaderMap::new();
+        headers.insert(reqwest::header::CONTENT_TYPE, "text/html; charset=UTF-8".parse().unwrap());
+
+        assert!(should_reject_success_response_content_type(PlaylistItemType::Catchup, &headers));
+        assert!(should_reject_success_response_content_type(PlaylistItemType::Video, &headers));
+        assert!(should_reject_success_response_content_type(PlaylistItemType::Live, &headers));
+    }
+
+    #[test]
+    fn html_content_type_is_allowed_for_live_adaptive_streams() {
+        let mut headers = HeaderMap::new();
+        headers.insert(reqwest::header::CONTENT_TYPE, "text/html; charset=UTF-8".parse().unwrap());
+
+        assert!(!should_reject_success_response_content_type(PlaylistItemType::LiveHls, &headers));
+        assert!(!should_reject_success_response_content_type(PlaylistItemType::LiveDash, &headers));
+    }
+
+    #[test]
+    fn session_headers_are_forwarded_to_provider_requests() {
+        let addr = "127.0.0.1:8080".parse().unwrap();
+        let stream_url = Url::parse("http://example.com/live/segment.ts").unwrap();
+        let req_headers = HeaderMap::new();
+        let mut session_headers = HashMap::new();
+        session_headers.insert(String::from("cookie"), String::from("sid=abc; pref=1"));
+        let stream_options =
+            StreamOptions { stream_retry: true, buffer_enabled: true, buffer_size: 1024, pipe_provider_stream: false };
+
+        let options = ProviderStreamFactoryOptions::new(&ProviderStreamFactoryParams {
+            addr,
+            item_type: PlaylistItemType::LiveHls,
+            share_stream: false,
+            stream_options: &stream_options,
+            stream_url: &stream_url,
+            req_headers: &req_headers,
+            input_headers: None,
+            session_headers: Some(&session_headers),
+            disabled_headers: None,
+            default_user_agent: None,
+            username: None,
+            client_ip: None,
+            stream_channel: None,
+            connect_failure_stage: None,
+        });
+
+        assert_eq!(
+            options.get_headers().get(axum::http::header::COOKIE).and_then(|value| value.to_str().ok()),
+            Some("sid=abc; pref=1")
+        );
+    }
 }

backend/src/api/model/streams/shared_stream_manager.rs

@@ -696,12 +696,9 @@ impl SharedStreamManager {
         self.get_shared_state(stream_url).await.map(|s| s.headers.clone())
     }
 
-    pub async fn register_meter_uid(&self, stream_url: &str, meter_uid: u32) {
-        self.meter_uids.write().await.insert(stream_url.to_string(), meter_uid);
-    }
-
-    pub async fn get_meter_uid(&self, stream_url: &str) -> Option<u32> {
-        self.meter_uids.read().await.get(stream_url).copied()
+    pub async fn get_or_register_meter_uid(&self, stream_url: &str, uid_factory: impl FnOnce() -> u32) -> u32 {
+        let mut uids = self.meter_uids.write().await;
+        *uids.entry(stream_url.to_string()).or_insert_with(uid_factory)
     }
 
     async fn forget_subscriber_addr(&self, addr: &SocketAddr) -> Option<Arc<str>> {