From 7d3c2fd8bef5a69f03125fb1070adcfc8d34a3f3 Mon Sep 17 00:00:00 2001 From: DarkBreakpoint <243206744+DarkBreakpoint@users.noreply.github.com> Date: Wed, 19 Nov 2025 23:00:25 -0600 Subject: [PATCH] Improve session tracking --- CHANGELOG.md | 1 + backend/src/api/api_utils.rs | 6 +-- backend/src/api/model/active_user_manager.rs | 38 +++++++++++++------ backend/src/api/model/connection_manager.rs | 4 +- .../api/model/streams/active_client_stream.rs | 3 +- shared/src/model/stream_info.rs | 7 +++- 6 files changed, 39 insertions(+), 20 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 34c06c500..a9a038d44 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/backend/src/api/api_utils.rs b/backend/src/api/api_utils.rs index 6f075ef00..81b866256 100644 --- a/backend/src/api/api_utils.rs +++ b/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); diff --git a/backend/src/api/model/active_user_manager.rs b/backend/src/api/model/active_user_manager.rs index 7242c8f5b..227eff1dd 100644 --- a/backend/src/api/model/active_user_manager.rs +++ b/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 { + provider: &str, stream_channel: StreamChannel, user_agent: Cow<'_, str>, session_token: Option<&str>) -> Option { 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; diff --git a/backend/src/api/model/connection_manager.rs b/backend/src/api/model/connection_manager.rs index d62acd2c0..f1e692546 100644 --- a/backend/src/api/model/connection_manager.rs +++ b/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); diff --git a/backend/src/api/model/streams/active_client_stream.rs b/backend/src/api/model/streams/active_client_stream.rs index 12ef62b99..d87559c8d 100644 --- a/backend/src/api/model/streams/active_client_stream.rs +++ b/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; } diff --git a/shared/src/model/stream_info.rs b/shared/src/model/stream_info.rs index ccd7fb963..ef8c8680a 100644 --- a/shared/src/model/stream_info.rs +++ b/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, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub session_token: Option, } impl StreamInfo { - pub fn new(username: &str, addr: &SocketAddr, client_ip: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option) -> Self { + pub fn new(username: &str, addr: &SocketAddr, client_ip: &str, provider: &str, stream_channel: StreamChannel, user_agent: String, country: Option, 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()), } } -} \ No newline at end of file +}