feat: KSUDP auto-reconnect, handshake metadata capture

- Auto-reconnect: if no packets for 5s, re-send HANDSHAKE_REQUEST instead
  of SUBSCRIBE_UPDATE. Handles AC restarts and track/car changes without
  restarting rusty-telemetry. Reduces handshake interval from 10s to 5s.
- Handle 408-byte handshake response packets in the streaming recv loop
  (not just initial handshake). On re-handshake, parse metadata and send
  SUBSCRIBE_UPDATE to resume streaming.
- GameInfo struct: captures car_name, track_name, driver_name from the
  KSUDP handshake response. Shared via Arc<Mutex<Option<GameInfo>>>.
- parse_handshake_response(): extracts UTF-16LE strings, cleans null/trim.
- GET /api/game-info endpoint: exposes current game metadata.
- start_recording auto-fills game/track/car from GameInfo when not
  specified by the user.
- create_session auto-fills car_name/track_name from GameInfo.
- Add track_name field to UseCaseSession.
- MissedTickBehavior::Delay on handshake interval to prevent bursts.
This commit is contained in:
Jan-Peter von Hunnius 2026-07-04 22:19:29 +02:00
parent bcacd4b8a7
commit f759101db8
3 changed files with 161 additions and 42 deletions

View file

@ -41,6 +41,7 @@ pub struct AppState {
pub feed_displays: Vec<(FeedDisplayInfo, Arc<Mutex<FeedState>>)>, pub feed_displays: Vec<(FeedDisplayInfo, Arc<Mutex<FeedState>>)>,
pub session_manager: Arc<std::sync::Mutex<SessionManager>>, pub session_manager: Arc<std::sync::Mutex<SessionManager>>,
pub track_model_store: Arc<crate::track_model::TrackModelStore>, pub track_model_store: Arc<crate::track_model::TrackModelStore>,
pub game_info: Arc<std::sync::Mutex<Option<crate::GameInfo>>>,
} }
#[derive(Serialize)] #[derive(Serialize)]
@ -121,7 +122,19 @@ async fn create_session(
let recording_id = state.recording_manager.active_recording_id().await; let recording_id = state.recording_manager.active_recording_id().await;
let mut mgr = lock_sessions(&state); let mut mgr = lock_sessions(&state);
match mgr.create_session(req.use_case, req.label, recording_id) { match mgr.create_session(req.use_case, req.label, recording_id) {
Ok(session) => (StatusCode::CREATED, Json(session)).into_response(), Ok(session) => {
let (car, track) = state.game_info.lock()
.map(|gi| {
let (c, t) = gi.as_ref().map_or((None, None), |i| {
(i.car_name.clone(), i.track_name.clone())
});
(c, t)
})
.unwrap_or((None, None));
mgr.fill_session_metadata(&session.id, car, track);
let updated = mgr.get_session(&session.id);
(StatusCode::CREATED, Json(updated.unwrap_or(session))).into_response()
}
Err(msg) => err(StatusCode::BAD_REQUEST, &msg), Err(msg) => err(StatusCode::BAD_REQUEST, &msg),
} }
} }
@ -334,8 +347,20 @@ async fn get_recording(
async fn start_recording( async fn start_recording(
State(state): State<Arc<AppState>>, State(state): State<Arc<AppState>>,
Json(req): Json<StartRecordingRequest>, Json(mut req): Json<StartRecordingRequest>,
) -> Response { ) -> Response {
// Auto-fill game/track/car from handshake metadata if not specified.
if let Ok(Some(info)) = state.game_info.lock().as_deref() {
if req.game.is_none() {
req.game = Some(Game::AssettoCorsa);
}
if req.track.is_none() {
req.track = info.track_name.clone();
}
if req.car.is_none() {
req.car = info.car_name.clone();
}
}
match state.recording_manager.start_recording(req).await { match state.recording_manager.start_recording(req).await {
Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(), Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(),
Err(e) => err( Err(e) => err(
@ -356,6 +381,11 @@ async fn stop_recording(State(state): State<Arc<AppState>>) -> Response {
} }
} }
async fn get_game_info(State(state): State<Arc<AppState>>) -> Response {
let info = state.game_info.lock().map(|gi| gi.clone()).unwrap_or(None);
ok_json(info.unwrap_or_default())
}
async fn stop_specific_recording( async fn stop_specific_recording(
State(state): State<Arc<AppState>>, State(state): State<Arc<AppState>>,
Path(id): Path<String>, Path(id): Path<String>,
@ -746,6 +776,7 @@ pub fn build_router(state: Arc<AppState>) -> Router {
get(get_feed_data), get(get_feed_data),
) )
.route("/api/live", get(get_live)) .route("/api/live", get(get_live))
.route("/api/game-info", get(get_game_info))
.route("/api/tracks", get(list_tracks)) .route("/api/tracks", get(list_tracks))
.route("/api/tracks/{game}/{track_name}", get(get_track).delete(delete_track)) .route("/api/tracks/{game}/{track_name}", get(get_track).delete(delete_track))
.route("/api/tracks/{game}/{track_name}/corners", get(get_track_corners)) .route("/api/tracks/{game}/{track_name}/corners", get(get_track_corners))

View file

@ -28,6 +28,7 @@ mod analysis;
mod characteristics; mod characteristics;
mod session; mod session;
use serde::Serialize;
use feed_state::*; use feed_state::*;
use ui::*; use ui::*;
@ -53,13 +54,50 @@ use crate::warnings::Warnings;
const VERSION: &str = env!("CARGO_PKG_VERSION"); const VERSION: &str = env!("CARGO_PKG_VERSION");
const BUFFER_SIZE: usize = 65535; const BUFFER_SIZE: usize = 65535;
const RENDER_INTERVAL_MS: u64 = 250; const RENDER_INTERVAL_MS: u64 = 250;
const HANDSHAKE_INTERVAL_SECS: u64 = 10; const HANDSHAKE_INTERVAL_SECS: u64 = 5;
const HANDSHAKE_RESPONSE_SIZE: usize = 408; const HANDSHAKE_RESPONSE_SIZE: usize = 408;
const STALE_TIMEOUT_SECS: u64 = 5;
const HANDSHAKE_REQUEST: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]; const HANDSHAKE_REQUEST: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0];
const SUBSCRIBE_UPDATE: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0]; const SUBSCRIBE_UPDATE: [u8; 12] = [1, 0, 0, 0, 1, 0, 0, 0, 1, 0, 0, 0];
const API_PORT: u16 = 8080; const API_PORT: u16 = 8080;
/// Metadata extracted from the KSUDP handshake response.
/// Updated whenever AC reconnects (track/car change) and shared
/// with the API for auto-filling recording and session metadata.
#[derive(Debug, Clone, Serialize, Default)]
pub struct GameInfo {
#[serde(skip_serializing_if = "Option::is_none")]
pub car_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub track_name: Option<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub driver_name: Option<String>,
}
/// Parse a 408-byte KSUDP HandshakerResponse into GameInfo.
/// Layout: car_name (0-100), driver_name (100-200), identifier (200-204),
/// version (204-208), track_name (208-308), track_config (308-408).
fn parse_handshake_response(data: &[u8]) -> Option<GameInfo> {
if data.len() < HANDSHAKE_RESPONSE_SIZE {
return None;
}
let car_name = decode_utf16le(&data[0..100]);
let driver_name = decode_utf16le(&data[100..200]);
let track_name = decode_utf16le(&data[208..308]);
let clean = |s: String| -> Option<String> {
let trimmed = s.trim_end_matches('\0').trim().to_string();
if trimmed.is_empty() { None } else { Some(trimmed) }
};
Some(GameInfo {
car_name: clean(car_name),
driver_name: clean(driver_name),
track_name: clean(track_name),
})
}
fn api_port() -> u16 { fn api_port() -> u16 {
std::env::var("RUSTY_TELEMETRY_API_PORT") std::env::var("RUSTY_TELEMETRY_API_PORT")
.ok() .ok()
@ -183,6 +221,7 @@ async fn run_feed(
state: Arc<std::sync::Mutex<FeedState>>, state: Arc<std::sync::Mutex<FeedState>>,
warnings: Warnings, warnings: Warnings,
mut shutdown_rx: tokio::sync::watch::Receiver<bool>, mut shutdown_rx: tokio::sync::watch::Receiver<bool>,
game_info: Arc<std::sync::Mutex<Option<GameInfo>>>,
) { ) {
let handshake_target: Option<std::net::SocketAddr> = if feed.needs_handshake { let handshake_target: Option<std::net::SocketAddr> = if feed.needs_handshake {
let target: std::net::SocketAddr = match format!("{}:{}", game_host(), feed.port).parse() { let target: std::net::SocketAddr = match format!("{}:{}", game_host(), feed.port).parse() {
@ -221,34 +260,25 @@ async fn run_feed(
.await .await
{ {
Ok(Ok((len, _src))) if len == HANDSHAKE_RESPONSE_SIZE => { Ok(Ok((len, _src))) if len == HANDSHAKE_RESPONSE_SIZE => {
let car_name = decode_utf16le(&hs_buf[0..100]); let info = parse_handshake_response(&hs_buf[..len]);
let driver_name = decode_utf16le(&hs_buf[100..200]); if let Some(ref info) = info {
let identifier = i32::from_le_bytes([ warnings.info(&format!(
hs_buf[200], hs_buf[201], hs_buf[202], hs_buf[203], "{} :{} Phase 2: HandshakerResponse received",
]); feed.label, feed.port
let version = i32::from_le_bytes([ ));
hs_buf[204], hs_buf[205], hs_buf[206], hs_buf[207], warnings.info(&format!(
]); " car=\"{}\" driver=\"{}\" track=\"{}\"",
let track_name = decode_utf16le(&hs_buf[208..308]); info.car_name.as_deref().unwrap_or("?"),
let track_config = decode_utf16le(&hs_buf[308..408]); info.driver_name.as_deref().unwrap_or("?"),
info.track_name.as_deref().unwrap_or("?"),
warnings.info(&format!( ));
"{} :{} Phase 2: HandshakerResponse received", if let Ok(mut gi) = game_info.lock() {
feed.label, feed.port *gi = Some(info.clone());
)); }
warnings.info(&format!( } else {
" car_name=\"{}\" driver_name=\"{}\" id={} ver={}",
car_name, driver_name, identifier, version
));
warnings.info(&format!(
" track_name=\"{}\" track_config=\"{}\"",
track_name, track_config
));
if identifier != 4242 {
warnings.warn(&format!( warnings.warn(&format!(
"{} :{} Unexpected handshake identifier: {} (expected 4242)", "{} :{} Phase 2: Failed to parse handshake response",
feed.label, feed.port, identifier feed.label, feed.port
)); ));
} }
} }
@ -300,9 +330,13 @@ async fn run_feed(
let mut buf = vec![0u8; BUFFER_SIZE]; let mut buf = vec![0u8; BUFFER_SIZE];
let mut string_cache = CachedStrings::new(); let mut string_cache = CachedStrings::new();
let mut last_recv_time: Option<std::time::Instant> = None;
let stale_timeout = Duration::from_secs(STALE_TIMEOUT_SECS);
let mut hs_interval: Option<tokio::time::Interval> = handshake_target.map(|_| { let mut hs_interval: Option<tokio::time::Interval> = handshake_target.map(|_| {
tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS)) let mut iv = tokio::time::interval(Duration::from_secs(HANDSHAKE_INTERVAL_SECS));
iv.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
iv
}); });
loop { loop {
@ -310,12 +344,32 @@ async fn run_feed(
result = socket.recv_from(&mut buf) => { result = socket.recv_from(&mut buf) => {
match result { match result {
Ok((len, _src)) => { Ok((len, _src)) => {
let parsed: Option<TelemetryFrame> = if feed.needs_handshake { // Handle handshake response packets (AC reconnected with new track/car).
if len != HANDSHAKE_RESPONSE_SIZE { if feed.needs_handshake && len == HANDSHAKE_RESPONSE_SIZE {
parser_ksudp::parse_rt_car_info(&buf[..len]).ok() if let Some(info) = parse_handshake_response(&buf[..len]) {
} else { warnings.info(&format!(
None "{} :{} Re-handshake: car=\"{}\" track=\"{}\"",
feed.label, feed.port,
info.car_name.as_deref().unwrap_or("?"),
info.track_name.as_deref().unwrap_or("?"),
));
if let Ok(mut gi) = game_info.lock() {
*gi = Some(info);
}
if let Some(target) = handshake_target {
let _ = socket.send_to(&SUBSCRIBE_UPDATE, target).await;
}
if let Ok(mut s) = state.lock() {
s.set_handshake_phase(HandshakePhase::Subscribed);
}
} }
continue;
}
last_recv_time = Some(std::time::Instant::now());
let parsed: Option<TelemetryFrame> = if feed.needs_handshake {
parser_ksudp::parse_rt_car_info(&buf[..len]).ok()
} else if feed.name == "pcars" { } else if feed.name == "pcars" {
match parser_pcars::parse_pcars_packet(&buf[..len]) { match parser_pcars::parse_pcars_packet(&buf[..len]) {
Ok(parser_pcars::PcarsPacket::V1Telemetry(f)) => Some(f), Ok(parser_pcars::PcarsPacket::V1Telemetry(f)) => Some(f),
@ -369,11 +423,24 @@ async fn run_feed(
} }
} => { } => {
if let Some(target) = handshake_target { if let Some(target) = handshake_target {
if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await { let is_stale = last_recv_time.is_none_or(|t| t.elapsed() > stale_timeout);
warnings.warn(&format!( if is_stale {
"Periodic SUBSCRIBE_UPDATE send failed for :{}: {}", if let Ok(mut s) = state.lock() {
feed.port, e s.set_handshake_phase(HandshakePhase::WaitingForResponse);
)); }
if let Err(e) = socket.send_to(&HANDSHAKE_REQUEST, target).await {
warnings.warn(&format!(
"HANDSHAKE_REQUEST send failed for :{}: {}",
feed.port, e
));
}
} else {
if let Err(e) = socket.send_to(&SUBSCRIBE_UPDATE, target).await {
warnings.warn(&format!(
"SUBSCRIBE_UPDATE send failed for :{}: {}",
feed.port, e
));
}
} }
} }
} }
@ -452,6 +519,9 @@ async fn main() {
let mut feed_tasks: Vec<tokio::task::JoinHandle<()>> = Vec::new(); let mut feed_tasks: Vec<tokio::task::JoinHandle<()>> = Vec::new();
let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false); let (shutdown_tx, shutdown_rx) = tokio::sync::watch::channel(false);
let game_info: Arc<std::sync::Mutex<Option<GameInfo>>> =
Arc::new(std::sync::Mutex::new(None));
for feed in FEEDS { for feed in FEEDS {
let state = Arc::new(std::sync::Mutex::new(FeedState::new())); let state = Arc::new(std::sync::Mutex::new(FeedState::new()));
@ -507,8 +577,9 @@ async fn main() {
let rm = recording_manager.clone(); let rm = recording_manager.clone();
let w = warnings.clone(); let w = warnings.clone();
let sh = shutdown_rx.clone(); let sh = shutdown_rx.clone();
let gi = game_info.clone();
feed_tasks.push(tokio::spawn(async move { feed_tasks.push(tokio::spawn(async move {
run_feed(feed_clone, socket, rm, state, w, sh).await; run_feed(feed_clone, socket, rm, state, w, sh, gi).await;
})); }));
} }
} }
@ -518,6 +589,7 @@ async fn main() {
feed_displays: feed_displays.clone(), feed_displays: feed_displays.clone(),
session_manager: session_manager.clone(), session_manager: session_manager.clone(),
track_model_store, track_model_store,
game_info: game_info.clone(),
}); });
let api_addr = api_bind_addr(); let api_addr = api_bind_addr();
warnings.info(&format!("REST API server will listen on {}", api_addr)); warnings.info(&format!("REST API server will listen on {}", api_addr));

View file

@ -101,6 +101,8 @@ pub struct UseCaseSession {
pub recording_id: Option<String>, pub recording_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")] #[serde(default, skip_serializing_if = "Option::is_none")]
pub car_name: Option<String>, pub car_name: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub track_name: Option<String>,
} }
impl UseCaseSession { impl UseCaseSession {
@ -117,6 +119,7 @@ impl UseCaseSession {
guidance: self.guidance.clone(), guidance: self.guidance.clone(),
recording_id: self.recording_id.clone(), recording_id: self.recording_id.clone(),
car_name: self.car_name.clone(), car_name: self.car_name.clone(),
track_name: self.track_name.clone(),
} }
} }
} }
@ -335,12 +338,25 @@ impl SessionManager {
guidance, guidance,
recording_id, recording_id,
car_name: None, car_name: None,
track_name: None,
}; };
self.sessions.insert(id.clone(), session.clone()); self.sessions.insert(id.clone(), session.clone());
self.persist_session(&id); self.persist_session(&id);
Ok(session) Ok(session)
} }
/// Update car_name and track_name on a session if they are not yet set.
pub fn fill_session_metadata(&mut self, id: &str, car: Option<String>, track: Option<String>) {
if let Some(session) = self.sessions.get_mut(id) {
if session.car_name.is_none() {
session.car_name = car.filter(|c| !c.is_empty());
}
if session.track_name.is_none() {
session.track_name = track.filter(|t| !t.is_empty());
}
}
}
pub fn list_sessions(&mut self) -> Vec<UseCaseSession> { pub fn list_sessions(&mut self) -> Vec<UseCaseSession> {
let mut sessions: Vec<UseCaseSession> = self let mut sessions: Vec<UseCaseSession> = self
.sessions .sessions