Improve session tracking

CHANGELOG.md

@@ -9,6 +9,7 @@
 - Added detailed shared-stream/buffer/provider logging to trace lag, cache persistence, and session/provider lifecycle events.
 - Connection registration failures now trigger an explicit disconnect so zombie sockets don’t linger.
 - Shared stream shutdown now drops registry locks before releasing provider handles to prevent cross-lock stalls.
+ - Session tracking now matches repeated HLS segment connections by session token so a single user keeps one active connection count even when new TCP sockets are opened.
 
 
 # 3.2.0 (2025-11-14)

backend/src/api/api_utils.rs

@@ -742,7 +742,7 @@ pub async fn force_provider_stream_response(
             .await;
         stream_channel.shared = share_stream;
         let stream =
-            ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, req_headers)
+            ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, Some(&user_session.token), req_headers)
                 .await;
 
         let (status_code, header_map) =
@@ -845,7 +845,7 @@ pub async fn stream_response(
         };
         stream_channel.shared = share_stream;
         let stream =
-            ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, req_headers)
+            ActiveClientStream::new(stream_details, app_state, user, connection_permission, fingerprint, stream_channel, None, req_headers)
                 .await;
         let stream_resp = if share_stream {
             debug_if_enabled!(
@@ -982,7 +982,7 @@ async fn try_shared_stream_response_if_any(
             stream_details.provider_name = provider;
             stream_channel.shared = true;
             let stream =
-                ActiveClientStream::new(stream_details, app_state, user, connect_permission, fingerprint, stream_channel, req_headers)
+                ActiveClientStream::new(stream_details, app_state, user, connect_permission, fingerprint, stream_channel, None, req_headers)
                     .await
                     .boxed();
             let mut response = axum::response::Response::builder().status(status_code);

backend/src/api/model/active_user_manager.rs

@@ -255,7 +255,7 @@ impl ActiveUserManager {
 
     #[allow(clippy::too_many_arguments)]
     pub async fn update_connection(&self, username: &str, max_connections: u32, fingerprint: &Fingerprint,
-                                   provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) -> Option<StreamInfo> {
+                                   provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>, session_token: Option<&str>) -> Option<StreamInfo> {
         let stream_info = {
             let mut user_connections = self.connections.write().await;
 
@@ -271,17 +271,30 @@ impl ActiveUserManager {
                 .or_insert_with(|| UserConnectionData::new(0, max_connections));
             connection_data.max_connections = max_connections;
 
-            let existing_stream_info = connection_data
-                .streams
-                .iter_mut()
-                .find(|s| s.addr == fingerprint.addr)
-                .map(|stream_info| {
-                    stream_info.channel = stream_channel.clone();
-                    stream_info.provider = provider.to_string();
-                    stream_info.clone()
-                });
+            let user_agent_string = user_agent.to_string();
 
-            if let Some(stream_info) = existing_stream_info { stream_info } else {
+            let mut existing_stream_info = session_token.and_then(|token| {
+                connection_data.streams.iter_mut().find(|s| s.session_token.as_deref() == Some(token))
+            });
+
+            if existing_stream_info.is_none() {
+                existing_stream_info = connection_data
+                    .streams
+                    .iter_mut()
+                    .find(|s| s.addr == fingerprint.addr);
+            }
+
+            if let Some(stream_info) = existing_stream_info {
+                stream_info.channel = stream_channel.clone();
+                stream_info.provider = provider.to_string();
+                stream_info.addr = *fingerprint.addr;
+                stream_info.user_agent = user_agent_string.clone();
+                stream_info.ts = current_time_secs();
+                if session_token.is_some() {
+                    stream_info.session_token = session_token.map(|token| token.to_string());
+                }
+                stream_info.clone()
+            } else {
                 let country = {
                     let geoip = self.geo_ip.load();
                     if let Some(geoip_db) = (*geoip).as_ref() {
@@ -297,8 +310,9 @@ impl ActiveUserManager {
                     &fingerprint.client_ip,
                     provider,
                     stream_channel,
-                    user_agent.to_string(),
+                    user_agent_string,
                     country,
+                    session_token,
                 );
                 connection_data.connections += 1;
                 let connections = connection_data.connections;

backend/src/api/model/connection_manager.rs

@@ -69,8 +69,8 @@ impl ConnectionManager {
 
     #[allow(clippy::too_many_arguments)]
     pub async fn update_connection(&self, username: &str, max_connections: u32, fingerprint: &Fingerprint,
-                                provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>) {
-        if let Some(stream_info) = self.user_manager.update_connection(username, max_connections, fingerprint, provider, stream_channel, user_agent).await {
+                                provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>, session_token: Option<&str>) {
+        if let Some(stream_info) = self.user_manager.update_connection(username, max_connections, fingerprint, provider, stream_channel, user_agent, session_token).await {
             self.event_manager.send_event(EventMessage::ActiveUser(ActiveUserConnectionChange::Updated(stream_info)));
         } else {
             warn!("Failed to register connection for user {username} at {}; disconnecting client", fingerprint.addr);

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

@@ -39,6 +39,7 @@ impl ActiveClientStream {
                       connection_permission: UserConnectionPermission,
                       fingerprint: &Fingerprint,
                       stream_channel: StreamChannel,
+                      session_token: Option<&str>,
                       req_headers: &HeaderMap) -> Self {
         if connection_permission == UserConnectionPermission::Exhausted {
             error!("Something is wrong this should not happen");
@@ -49,7 +50,7 @@ impl ActiveClientStream {
 
         let user_agent = req_headers.get(USER_AGENT).map(|h| String::from_utf8_lossy(h.as_bytes())).unwrap_or_default();
 
-        app_state.connection_manager.update_connection(username, user.max_connections, fingerprint, &provider_name, stream_channel, user_agent).await;
+        app_state.connection_manager.update_connection(username, user.max_connections, fingerprint, &provider_name, stream_channel, user_agent, session_token).await;
         if let Some((_,_,_m_, Some(cvt))) = stream_details.stream_info.as_ref() {
             app_state.connection_manager.update_stream_detail(&fingerprint.addr, *cvt).await;
         }

shared/src/model/stream_info.rs

@@ -64,10 +64,12 @@ pub struct StreamInfo {
     pub ts: u64,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub country: Option<String>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub session_token: Option<String>,
 }
 
 impl StreamInfo {
-    pub fn new(username: &str, addr: &SocketAddr, client_ip: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option<String>) -> Self {
+    pub fn new(username: &str, addr: &SocketAddr, client_ip: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option<String>, session_token: Option<&str>) -> Self {
         Self {
             username: username.to_string(),
             channel: stream_channel,
@@ -77,6 +79,7 @@ impl StreamInfo {
             user_agent,
             ts: current_time_secs(),
             country,
+            session_token: session_token.map(|token| token.to_string()),
         }
     }
-}
+}