diff --git a/crates/fbuild-build/tests/cache_survives_tar_extract.rs b/crates/fbuild-build/tests/cache_survives_tar_extract.rs index 825f2444..ef5c2439 100644 --- a/crates/fbuild-build/tests/cache_survives_tar_extract.rs +++ b/crates/fbuild-build/tests/cache_survives_tar_extract.rs @@ -172,9 +172,10 @@ fn compiler_signature_survives_toolchain_path_change() { let flags = vec!["-O2".to_string(), "-DFOO=1".to_string()]; let pre_flags = vec!["-Wall".to_string()]; let extra_flags = vec!["-I/some/include".to_string()]; + let build_unflags: Vec = Vec::new(); - let sig_a = build_rebuild_signature(&path_a, &flags, &pre_flags, &extra_flags); - let sig_b = build_rebuild_signature(&path_b, &flags, &pre_flags, &extra_flags); + let sig_a = build_rebuild_signature(&path_a, &flags, &pre_flags, &extra_flags, &build_unflags); + let sig_b = build_rebuild_signature(&path_b, &flags, &pre_flags, &extra_flags, &build_unflags); assert_eq!( sig_a, sig_b, @@ -184,7 +185,7 @@ fn compiler_signature_survives_toolchain_path_change() { let alt_filename = if cfg!(windows) { "clang.exe" } else { "clang" }; let path_c = toolchain_a.path().join(alt_filename); - let sig_c = build_rebuild_signature(&path_c, &flags, &pre_flags, &extra_flags); + let sig_c = build_rebuild_signature(&path_c, &flags, &pre_flags, &extra_flags, &build_unflags); assert_ne!( sig_a, sig_c, "build_rebuild_signature collapsed two different compilers to the same signature; \ diff --git a/crates/fbuild-cli/src/cli/args.rs b/crates/fbuild-cli/src/cli/args.rs index 7762cf3e..34ac84ac 100644 --- a/crates/fbuild-cli/src/cli/args.rs +++ b/crates/fbuild-cli/src/cli/args.rs @@ -729,7 +729,23 @@ pub enum DaemonAction { /// Show lock status (project locks, serial sessions) Locks, /// Clear stale locks - ClearLocks, + ClearLocks { + /// Include serial sessions in stale lock cleanup + #[arg(long)] + serial: bool, + /// Close only sessions that the daemon can classify as stale + #[arg(long)] + stale: bool, + /// Target a serial session by port (for example COM11 or /dev/ttyUSB0) + #[arg(long)] + port: Option, + /// Target a serial session by client id + #[arg(long = "client-id")] + client_id: Option, + /// Close a targeted live serial session; requires --port or --client-id + #[arg(long)] + force: bool, + }, /// Show disk cache statistics CacheStats, /// Run disk cache garbage collection diff --git a/crates/fbuild-cli/src/cli/daemon_cmd.rs b/crates/fbuild-cli/src/cli/daemon_cmd.rs index 05bb7db4..6e3e85a0 100644 --- a/crates/fbuild-cli/src/cli/daemon_cmd.rs +++ b/crates/fbuild-cli/src/cli/daemon_cmd.rs @@ -111,8 +111,21 @@ pub async fn run_daemon(action: DaemonAction) -> fbuild_core::Result<()> { DaemonAction::Locks => { run_daemon_locks(&client).await?; } - DaemonAction::ClearLocks => { - run_daemon_clear_locks(&client).await?; + DaemonAction::ClearLocks { + serial, + stale, + port, + client_id, + force, + } => { + let request = daemon_client::ClearLocksRequest { + serial, + stale, + port, + client_id, + force, + }; + run_daemon_clear_locks(&client, request).await?; } DaemonAction::CacheStats => { run_daemon_cache_stats(&client).await?; @@ -347,9 +360,64 @@ pub async fn run_daemon_locks(client: &DaemonClient) -> fbuild_core::Result<()> for lock in &status.port_locks { let state = if lock.is_held { "HELD" } else { "FREE" }; let writer = lock.writer_client_id.as_deref().unwrap_or("none"); + let owner = lock.owner_client_id.as_deref().unwrap_or("none"); output::result(format!( - " {} [{}] open={} writer={} readers={}", - lock.port, state, lock.is_open, writer, lock.reader_count + " {} [{}] open={} baud={} owner={} writer={} readers={} age={} last_activity={} rx={} tx={}", + lock.port, + state, + lock.is_open, + lock.baud_rate, + owner, + writer, + lock.reader_count, + format_age_seconds(lock.session_age_seconds), + format_age_seconds(lock.last_activity_age_seconds), + lock.total_bytes_read, + lock.total_bytes_written + )); + if let Some(age) = lock.last_read_age_seconds { + output::result(format!(" last_read: {}", format_age_seconds(age))); + } + if let Some(age) = lock.last_write_age_seconds { + output::result(format!(" last_write: {}", format_age_seconds(age))); + } + for client in &lock.clients { + let pid = client + .pid + .map(|pid| pid.to_string()) + .unwrap_or_else(|| "unknown".to_string()); + let alive = client + .process_alive + .map(|alive| if alive { "alive" } else { "dead" }) + .unwrap_or("unknown"); + output::result(format!( + " client {} pid={} ({})", + client.client_id, pid, alive + )); + if let Some(exe) = &client.exe { + output::result(format!(" exe: {}", exe)); + } + if let Some(cwd) = &client.cwd { + output::result(format!(" cwd: {}", cwd)); + } + if let Some(argv) = &client.argv { + if !argv.is_empty() { + output::result(format!(" argv: {}", argv.join(" "))); + } + } + } + } + } + + if !status.pending_serial_attaches.is_empty() { + output::result("Pending Serial Attaches:"); + for pending in &status.pending_serial_attaches { + output::result(format!( + " #{} port={} client={} age={}", + pending.id, + pending.port.as_deref().unwrap_or("unknown"), + pending.client_id.as_deref().unwrap_or("unknown"), + format_age_seconds(pending.age_seconds) )); } } @@ -375,17 +443,35 @@ pub async fn run_daemon_locks(client: &DaemonClient) -> fbuild_core::Result<()> Ok(()) } -pub async fn run_daemon_clear_locks(client: &DaemonClient) -> fbuild_core::Result<()> { +pub async fn run_daemon_clear_locks( + client: &DaemonClient, + request: daemon_client::ClearLocksRequest, +) -> fbuild_core::Result<()> { if !client.health().await { output::result("daemon is not running"); return Ok(()); } - let result = client.clear_locks().await?; + let result = client.clear_locks_with(&request).await?; output::result(&result.message); if result.cleared_count > 0 { output::result(format!("Cleared {} lock(s)", result.cleared_count)); } + if result.cleared_project_count > 0 { + output::result(format!( + "Cleared {} project lock(s)", + result.cleared_project_count + )); + } + if result.cleared_serial_count > 0 { + output::result(format!( + "Cleared serial sessions: {}", + result.cleared_serial_sessions.join(", ") + )); + } + for refusal in result.refused { + output::warn(refusal); + } Ok(()) } @@ -688,3 +774,7 @@ pub fn format_uptime(seconds: f64) -> String { format!("{}h {}m", secs / 3600, (secs % 3600) / 60) } } + +fn format_age_seconds(seconds: f64) -> String { + format_uptime(seconds.max(0.0)) +} diff --git a/crates/fbuild-cli/src/daemon_client.rs b/crates/fbuild-cli/src/daemon_client.rs index 953e8024..51560b4c 100644 --- a/crates/fbuild-cli/src/daemon_client.rs +++ b/crates/fbuild-cli/src/daemon_client.rs @@ -546,9 +546,18 @@ impl DaemonClient { /// Clear stale locks on the daemon. pub async fn clear_locks(&self) -> fbuild_core::Result { + self.clear_locks_with(&ClearLocksRequest::default()).await + } + + /// Clear stale locks on the daemon with targeting options. + pub async fn clear_locks_with( + &self, + request: &ClearLocksRequest, + ) -> fbuild_core::Result { let resp = self .client .post(format!("{}/api/locks/clear", self.base_url)) + .json(request) .timeout(fbuild_core::time::MEDIUM_HTTP_TIMEOUT) .send() .await diff --git a/crates/fbuild-cli/src/daemon_client/types.rs b/crates/fbuild-cli/src/daemon_client/types.rs index 9a2dcc86..48b0cd8e 100644 --- a/crates/fbuild-cli/src/daemon_client/types.rs +++ b/crates/fbuild-cli/src/daemon_client/types.rs @@ -227,6 +227,8 @@ pub struct LockStatusResponse { pub success: bool, pub port_locks: Vec, pub project_locks: Vec, + #[serde(default)] + pub pending_serial_attaches: Vec, pub stale_locks: Vec, } @@ -237,8 +239,70 @@ pub struct PortLockInfo { #[allow(dead_code)] pub holder_description: Option, pub is_open: bool, + #[serde(default)] + pub owner_client_id: Option, pub writer_client_id: Option, pub reader_count: usize, + #[serde(default)] + #[allow(dead_code)] + pub reader_client_ids: Vec, + #[serde(default)] + pub baud_rate: u32, + #[serde(default)] + #[allow(dead_code)] + pub started_at: f64, + #[serde(default)] + pub session_age_seconds: f64, + #[serde(default)] + #[allow(dead_code)] + pub last_activity_at: f64, + #[serde(default)] + pub last_activity_age_seconds: f64, + #[serde(default)] + #[allow(dead_code)] + pub last_read_at: Option, + #[serde(default)] + pub last_read_age_seconds: Option, + #[serde(default)] + #[allow(dead_code)] + pub last_write_at: Option, + #[serde(default)] + pub last_write_age_seconds: Option, + #[serde(default)] + pub total_bytes_read: u64, + #[serde(default)] + pub total_bytes_written: u64, + #[serde(default)] + pub clients: Vec, +} + +#[derive(Debug, Deserialize)] +pub struct SerialClientLockInfo { + pub client_id: String, + #[serde(default)] + pub pid: Option, + #[serde(default)] + pub process_alive: Option, + #[serde(default)] + pub exe: Option, + #[serde(default)] + pub cwd: Option, + #[serde(default)] + pub argv: Option>, +} + +#[derive(Debug, Deserialize)] +pub struct PendingSerialAttachInfo { + pub id: u64, + #[serde(default)] + pub client_id: Option, + #[serde(default)] + pub port: Option, + #[serde(default)] + #[allow(dead_code)] + pub started_at: f64, + #[serde(default)] + pub age_seconds: f64, } #[derive(Debug, Deserialize)] @@ -247,11 +311,33 @@ pub struct ProjectLockInfo { pub is_held: bool, } +#[derive(Debug, Default, Serialize)] +pub struct ClearLocksRequest { + #[serde(default)] + pub serial: bool, + #[serde(default)] + pub stale: bool, + #[serde(skip_serializing_if = "Option::is_none")] + pub port: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub client_id: Option, + #[serde(default)] + pub force: bool, +} + #[derive(Debug, Deserialize)] pub struct ClearLocksResponse { #[allow(dead_code)] pub success: bool, pub cleared_count: usize, + #[serde(default)] + pub cleared_project_count: usize, + #[serde(default)] + pub cleared_serial_count: usize, + #[serde(default)] + pub cleared_serial_sessions: Vec, + #[serde(default)] + pub refused: Vec, pub message: String, } diff --git a/crates/fbuild-daemon/src/context.rs b/crates/fbuild-daemon/src/context.rs index fc69418a..a2b04044 100644 --- a/crates/fbuild-daemon/src/context.rs +++ b/crates/fbuild-daemon/src/context.rs @@ -7,7 +7,7 @@ use fbuild_core::install_status::InstallStatus; use fbuild_core::DaemonState; use fbuild_serial::SharedSerialManager; use std::path::PathBuf; -use std::sync::atomic::{AtomicBool, AtomicUsize}; +use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize}; use std::sync::{Arc, Weak}; use std::time::Instant; use tokio::sync::Mutex; @@ -124,6 +124,14 @@ fn compute_binary_mtime() -> f64 { .unwrap_or(0.0) } +#[derive(Debug, Clone)] +pub struct PendingSerialAttachInfo { + pub id: u64, + pub started_at: f64, + pub client_id: Option, + pub port: Option, +} + /// Shared state for the daemon, passed to all axum handlers via `State`. pub struct DaemonContext { /// When the daemon started. @@ -145,6 +153,8 @@ pub struct DaemonContext { /// self-evict mid-attach and the client would see a forcibly-closed /// WebSocket. See ISSUES.md "Self-eviction during pending serial attach". pub pending_serial_attaches: Arc, + pub pending_serial_attach_details: DashMap, + pending_serial_attach_next_id: AtomicU64, /// Current daemon state (idle, building, deploying, etc.). pub daemon_state: Arc>, /// Description of the current operation (e.g. project dir being built). @@ -227,6 +237,8 @@ impl DaemonContext { is_shutting_down: Arc::new(AtomicBool::new(false)), operation_in_progress: Arc::new(AtomicBool::new(false)), pending_serial_attaches: Arc::new(AtomicUsize::new(0)), + pending_serial_attach_details: DashMap::new(), + pending_serial_attach_next_id: AtomicU64::new(1), daemon_state: Arc::new(std::sync::RwLock::new(DaemonState::Idle)), current_operation: Arc::new(std::sync::RwLock::new(None)), dependency_install: Arc::new(std::sync::RwLock::new(None)), @@ -258,6 +270,52 @@ impl DaemonContext { } } + pub fn begin_pending_serial_attach(&self) -> u64 { + let id = self + .pending_serial_attach_next_id + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let started_at = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64(); + self.pending_serial_attaches + .fetch_add(1, std::sync::atomic::Ordering::Relaxed); + self.pending_serial_attach_details.insert( + id, + PendingSerialAttachInfo { + id, + started_at, + client_id: None, + port: None, + }, + ); + id + } + + pub fn update_pending_serial_attach(&self, id: u64, client_id: String, port: String) { + if let Some(mut entry) = self.pending_serial_attach_details.get_mut(&id) { + entry.client_id = Some(client_id); + entry.port = Some(port); + } + } + + pub fn end_pending_serial_attach(&self, id: u64) { + if self.pending_serial_attach_details.remove(&id).is_some() { + self.pending_serial_attaches + .fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + } + } + + pub fn pending_serial_attach_infos(&self) -> Vec { + let mut infos: Vec<_> = self + .pending_serial_attach_details + .iter() + .map(|entry| entry.value().clone()) + .collect(); + infos.sort_by_key(|info| info.id); + infos + } + /// Check whether the daemon is completely idle (no ops, no serial sessions, /// no pending serial attaches). Pending attaches are counted separately /// because a WebSocket client may be in the middle of opening a port; if diff --git a/crates/fbuild-daemon/src/handlers/locks.rs b/crates/fbuild-daemon/src/handlers/locks.rs index 8b47e43f..b813aea6 100644 --- a/crates/fbuild-daemon/src/handlers/locks.rs +++ b/crates/fbuild-daemon/src/handlers/locks.rs @@ -1,14 +1,172 @@ //! Lock status and management handlers. use crate::context::DaemonContext; -use crate::models::{ClearLocksResponse, LockStatusResponse, PortLockInfo, ProjectLockInfo}; +use crate::models::{ + ClearLocksRequest, ClearLocksResponse, LockStatusResponse, PendingSerialAttachLockInfo, + PortLockInfo, ProjectLockInfo, SerialClientLockInfo, +}; use axum::extract::State; use axum::Json; +use fbuild_serial::{PortSessionInfo, SerialClientInfo}; use std::sync::Arc; +fn now_unix_secs() -> f64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64() +} + +fn age_seconds(now: f64, timestamp: f64) -> f64 { + (now - timestamp).max(0.0) +} + +fn optional_age_seconds(now: f64, timestamp: Option) -> Option { + timestamp.map(|ts| age_seconds(now, ts)) +} + +fn serial_client_lock_info(client: &SerialClientInfo) -> SerialClientLockInfo { + SerialClientLockInfo { + client_id: client.client_id.clone(), + pid: client.metadata.pid, + process_alive: client.metadata.pid.map(is_pid_alive), + exe: client.metadata.exe.clone(), + cwd: client.metadata.cwd.clone(), + argv: client.metadata.argv.clone(), + } +} + +fn port_matches(actual: &str, requested: &str) -> bool { + if cfg!(windows) { + actual.eq_ignore_ascii_case(requested) + } else { + actual == requested + } +} + +fn session_matches_request(session: &PortSessionInfo, request: &ClearLocksRequest) -> bool { + let port_matches_request = request + .port + .as_deref() + .is_some_and(|port| port_matches(&session.port, port)); + let client_matches_request = request.client_id.as_deref().is_some_and(|client_id| { + session.owner_client_id.as_deref() == Some(client_id) + || session.writer_client_id.as_deref() == Some(client_id) + || session + .reader_client_ids + .iter() + .any(|reader| reader == client_id) + || session + .clients + .iter() + .any(|client| client.client_id == client_id) + }); + + if request.port.is_none() && request.client_id.is_none() { + request.serial && request.stale + } else { + port_matches_request || client_matches_request + } +} + +fn session_stale_reason(session: &PortSessionInfo) -> Option { + if session.writer_client_id.is_none() && session.reader_count == 0 { + return Some("no active reader or writer clients".to_string()); + } + + if let Some(owner_id) = session.owner_client_id.as_deref() { + if let Some(owner) = session.clients.iter().find(|c| c.client_id == owner_id) { + if let Some(pid) = owner.metadata.pid { + return if is_pid_alive(pid) { + None + } else { + Some(format!("owner pid {pid} is no longer running")) + }; + } + } + } + + let known_pids: Vec = session + .clients + .iter() + .filter_map(|client| client.metadata.pid) + .collect(); + if !known_pids.is_empty() && known_pids.iter().all(|pid| !is_pid_alive(*pid)) { + return Some(format!( + "all known client pid(s) are gone: {}", + known_pids + .iter() + .map(u32::to_string) + .collect::>() + .join(", ") + )); + } + + None +} + +fn live_refusal_reason(session: &PortSessionInfo) -> String { + if let Some(owner_id) = session.owner_client_id.as_deref() { + if let Some(owner) = session.clients.iter().find(|c| c.client_id == owner_id) { + if let Some(pid) = owner.metadata.pid { + if is_pid_alive(pid) { + return format!("owner client {owner_id} pid {pid} is still running"); + } + } + } + } + if session + .clients + .iter() + .any(|client| client.metadata.pid.is_some()) + { + return "at least one attached client pid is still running".to_string(); + } + "no stale evidence available for active session".to_string() +} + +fn is_pid_alive(pid: u32) -> bool { + #[cfg(unix)] + { + if pid > i32::MAX as u32 { + return false; + } + // SAFETY: kill(pid, 0) probes existence without sending a signal. + if unsafe { libc::kill(pid as i32, 0) } == 0 { + true + } else { + std::io::Error::last_os_error().raw_os_error() == Some(libc::EPERM) + } + } + #[cfg(windows)] + { + const PROCESS_QUERY_LIMITED_INFORMATION: u32 = 0x1000; + type Handle = *mut std::ffi::c_void; + #[link(name = "kernel32")] + extern "system" { + fn OpenProcess(desired_access: u32, inherit_handle: i32, process_id: u32) -> Handle; + fn CloseHandle(handle: Handle) -> i32; + } + unsafe { + let h = OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, 0, pid); + if h.is_null() { + false + } else { + CloseHandle(h); + true + } + } + } + #[cfg(not(any(unix, windows)))] + { + let _ = pid; + true + } +} + /// GET /api/locks/status pub async fn lock_status(State(ctx): State>) -> Json { - // Gather project lock status + let now = now_unix_secs(); let project_locks: Vec = ctx .project_locks .iter() @@ -21,17 +179,50 @@ pub async fn lock_status(State(ctx): State>) -> Json = sessions - .into_iter() + .iter() .map(|s| PortLockInfo { port: s.port.clone(), is_held: s.is_open, holder_description: s.owner_client_id.clone(), is_open: s.is_open, - writer_client_id: s.writer_client_id, + owner_client_id: s.owner_client_id.clone(), + writer_client_id: s.writer_client_id.clone(), reader_count: s.reader_count, + reader_client_ids: s.reader_client_ids.clone(), + baud_rate: s.baud_rate, + started_at: s.started_at, + session_age_seconds: age_seconds(now, s.started_at), + last_activity_at: s.last_activity_at, + last_activity_age_seconds: age_seconds(now, s.last_activity_at), + last_read_at: s.last_read_at, + last_read_age_seconds: optional_age_seconds(now, s.last_read_at), + last_write_at: s.last_write_at, + last_write_age_seconds: optional_age_seconds(now, s.last_write_at), + total_bytes_read: s.total_bytes_read, + total_bytes_written: s.total_bytes_written, + clients: s.clients.iter().map(serial_client_lock_info).collect(), + }) + .collect(); + + let mut stale_locks: Vec = sessions + .iter() + .filter_map(|s| { + session_stale_reason(s).map(|reason| format!("serial {}: {}", s.port, reason)) + }) + .collect(); + stale_locks.sort(); + + let pending_serial_attaches = ctx + .pending_serial_attach_infos() + .into_iter() + .map(|info| PendingSerialAttachLockInfo { + id: info.id, + client_id: info.client_id, + port: info.port, + started_at: info.started_at, + age_seconds: age_seconds(now, info.started_at), }) .collect(); @@ -39,36 +230,99 @@ pub async fn lock_status(State(ctx): State>) -> Json>) -> Json { - // Remove project lock entries that are not currently held - let mut cleared = 0usize; +pub async fn clear_locks( + State(ctx): State>, + request: Option>, +) -> Json { + let request = request.map(|Json(req)| req).unwrap_or_default(); + let mut cleared_project_count = 0usize; let keys: Vec<_> = ctx.project_locks.iter().map(|e| e.key().clone()).collect(); for key in keys { if let Some(entry) = ctx.project_locks.get(&key) { if entry.value().try_lock().is_ok() { - // Not held — safe to remove the stale entry drop(entry); ctx.project_locks.remove(&key); - cleared += 1; + cleared_project_count += 1; + } + } + } + + let mut cleared_serial_sessions = Vec::new(); + let mut refused = Vec::new(); + let serial_requested = request.serial + || request.stale + || request.force + || request.port.is_some() + || request.client_id.is_some(); + + if request.force && request.port.is_none() && request.client_id.is_none() { + refused.push("--force requires --port or --client-id".to_string()); + } else if serial_requested + && !request.stale + && request.port.is_none() + && request.client_id.is_none() + { + refused.push("serial cleanup requires --stale, --port, or --client-id".to_string()); + } else if serial_requested { + let sessions = ctx.serial_manager.get_port_sessions(); + let mut matched_any = false; + for session in sessions { + if !session_matches_request(&session, &request) { + continue; } + matched_any = true; + let stale_reason = session_stale_reason(&session); + if stale_reason.is_none() && !request.force { + refused.push(format!( + "refused to close serial {}: {}; pass --force with --port/--client-id to close it", + session.port, + live_refusal_reason(&session) + )); + continue; + } + + match ctx + .serial_manager + .close_port(&session.port, "clear_locks") + .await + { + Ok(()) => cleared_serial_sessions.push(session.port), + Err(err) => { + refused.push(format!("failed to close serial {}: {}", session.port, err)) + } + } + } + if !matched_any { + refused.push("no matching serial sessions found".to_string()); } } - let message = if cleared > 0 { - format!("Cleared {} stale lock(s)", cleared) + let cleared_serial_count = cleared_serial_sessions.len(); + let cleared_count = cleared_project_count + cleared_serial_count; + let message = if !refused.is_empty() && cleared_count == 0 { + format!("No locks cleared; {}", refused.join("; ")) + } else if !refused.is_empty() { + format!("Cleared {} lock(s); {}", cleared_count, refused.join("; ")) + } else if cleared_count > 0 { + format!("Cleared {} stale lock(s)", cleared_count) } else { "No stale locks found".to_string() }; Json(ClearLocksResponse { success: true, - cleared_count: cleared, + cleared_count, + cleared_project_count, + cleared_serial_count, + cleared_serial_sessions, + refused, message, }) } @@ -76,6 +330,7 @@ pub async fn clear_locks(State(ctx): State>) -> Json, port: &str, client: &str, pid: Option) { + let now = now_unix_secs(); + let mut session = SerialSession::new(port.to_string(), 115200); + session.is_open = true; + session.owner_client_id = Some(client.to_string()); + session.writer_client_id = Some(client.to_string()); + session.reader_client_ids.insert(client.to_string()); + session.started_at = now - 10.0; + session.last_activity_at = now - 5.0; + session.client_metadata.insert( + client.to_string(), + SerialClientMetadata { + pid, + exe: Some("python".to_string()), + cwd: Some("/tmp/fbuild-test".to_string()), + argv: Some(vec!["python".to_string(), "-".to_string()]), + }, + ); + ctx.serial_manager.insert_session_for_test(session); + } + #[tokio::test] async fn lock_status_reports_held_project_locks_without_stale_entries() { let ctx = test_context(); @@ -99,10 +375,8 @@ mod tests { let Json(status) = lock_status(State(ctx)).await; assert!(status.success); - assert!( - status.stale_locks.is_empty(), - "stale_locks is reserved for future durable stale-lock detection" - ); + assert!(status.stale_locks.is_empty()); + assert!(status.pending_serial_attaches.is_empty()); assert_eq!(status.project_locks.len(), 1); assert_eq!( status.project_locks[0].project_dir, @@ -113,6 +387,45 @@ mod tests { drop(guard); } + #[tokio::test] + async fn lock_status_reports_serial_owner_metadata_and_age() { + let ctx = test_context(); + insert_serial_session(&ctx, "COM_TEST_META", "client-1", Some(std::process::id())); + + let Json(status) = lock_status(State(ctx)).await; + + assert_eq!(status.port_locks.len(), 1); + let lock = &status.port_locks[0]; + assert_eq!(lock.port, "COM_TEST_META"); + assert_eq!(lock.owner_client_id.as_deref(), Some("client-1")); + assert_eq!(lock.writer_client_id.as_deref(), Some("client-1")); + assert_eq!(lock.reader_client_ids, vec!["client-1".to_string()]); + assert!(lock.session_age_seconds >= 0.0); + assert!(lock.last_activity_age_seconds >= 0.0); + assert_eq!(lock.clients.len(), 1); + assert_eq!(lock.clients[0].pid, Some(std::process::id())); + assert_eq!(lock.clients[0].process_alive, Some(true)); + assert_eq!(lock.clients[0].exe.as_deref(), Some("python")); + } + + #[tokio::test] + async fn lock_status_reports_pending_serial_attach_age() { + let ctx = test_context(); + let id = ctx.begin_pending_serial_attach(); + ctx.update_pending_serial_attach(id, "client-2".to_string(), "COM_PENDING".to_string()); + + let Json(status) = lock_status(State(ctx.clone())).await; + + assert_eq!(status.pending_serial_attaches.len(), 1); + let pending = &status.pending_serial_attaches[0]; + assert_eq!(pending.id, id); + assert_eq!(pending.client_id.as_deref(), Some("client-2")); + assert_eq!(pending.port.as_deref(), Some("COM_PENDING")); + assert!(pending.age_seconds >= 0.0); + + ctx.end_pending_serial_attach(id); + } + #[tokio::test] async fn clear_locks_removes_only_unheld_project_lock_entries() { let ctx = test_context(); @@ -124,13 +437,87 @@ mod tests { .insert(unheld.clone(), Arc::new(Mutex::new(()))); ctx.project_locks.insert(held.clone(), held_lock.clone()); - let Json(response) = clear_locks(State(ctx.clone())).await; + let Json(response) = clear_locks(State(ctx.clone()), None).await; assert!(response.success); assert_eq!(response.cleared_count, 1); + assert_eq!(response.cleared_project_count, 1); + assert_eq!(response.cleared_serial_count, 0); assert!(!ctx.project_locks.contains_key(&unheld)); assert!(ctx.project_locks.contains_key(&held)); drop(guard); } + + #[tokio::test] + async fn clear_locks_refuses_live_serial_session_without_force() { + let ctx = test_context(); + insert_serial_session( + &ctx, + "COM_TEST_LIVE", + "client-live", + Some(std::process::id()), + ); + + let req = ClearLocksRequest { + serial: true, + port: Some("COM_TEST_LIVE".to_string()), + ..Default::default() + }; + let Json(response) = clear_locks(State(ctx.clone()), Some(Json(req))).await; + + assert_eq!(response.cleared_serial_count, 0); + assert_eq!(ctx.serial_manager.get_port_sessions().len(), 1); + assert!( + response + .refused + .iter() + .any(|msg| msg.contains("still running")), + "expected live-session refusal, got {:?}", + response.refused + ); + } + + #[tokio::test] + async fn clear_locks_force_closes_targeted_serial_session() { + let ctx = test_context(); + insert_serial_session( + &ctx, + "COM_TEST_FORCE", + "client-force", + Some(std::process::id()), + ); + + let req = ClearLocksRequest { + serial: true, + port: Some("COM_TEST_FORCE".to_string()), + force: true, + ..Default::default() + }; + let Json(response) = clear_locks(State(ctx.clone()), Some(Json(req))).await; + + assert_eq!(response.cleared_serial_count, 1); + assert_eq!( + response.cleared_serial_sessions, + vec!["COM_TEST_FORCE".to_string()] + ); + assert!(ctx.serial_manager.get_port_sessions().is_empty()); + } + + #[tokio::test] + async fn clear_locks_stale_closes_dead_owner_serial_session() { + let ctx = test_context(); + insert_serial_session(&ctx, "COM_TEST_STALE", "client-stale", Some(u32::MAX)); + + let req = ClearLocksRequest { + serial: true, + stale: true, + ..Default::default() + }; + let Json(response) = clear_locks(State(ctx.clone()), Some(Json(req))).await; + + assert_eq!(response.cleared_serial_count, 1); + assert!(response.refused.is_empty()); + assert!(ctx.serial_manager.get_port_sessions().is_empty()); + } } diff --git a/crates/fbuild-daemon/src/handlers/operations/deploy.rs b/crates/fbuild-daemon/src/handlers/operations/deploy.rs index cce9f625..721fbe54 100644 --- a/crates/fbuild-daemon/src/handlers/operations/deploy.rs +++ b/crates/fbuild-daemon/src/handlers/operations/deploy.rs @@ -959,7 +959,7 @@ pub async fn deploy( let open_result = tokio::time::timeout( POST_DEPLOY_OPEN_TIMEOUT, ctx.serial_manager - .open_port(&monitor_port, baud_rate, &request_id, None), + .open_port(&monitor_port, baud_rate, &request_id, None, None), ) .await; let open_err: Option = match open_result { @@ -989,7 +989,10 @@ pub async fn deploy( } // Subscribe to broadcast channel - let mut rx = match ctx.serial_manager.attach_reader(&monitor_port, &request_id) { + let mut rx = match ctx + .serial_manager + .attach_reader(&monitor_port, &request_id, None) + { Some(rx) => rx, None => { return ( diff --git a/crates/fbuild-daemon/src/handlers/operations/monitor.rs b/crates/fbuild-daemon/src/handlers/operations/monitor.rs index 73226731..8ac62220 100644 --- a/crates/fbuild-daemon/src/handlers/operations/monitor.rs +++ b/crates/fbuild-daemon/src/handlers/operations/monitor.rs @@ -260,7 +260,7 @@ pub async fn monitor( let open_result = tokio::time::timeout( std::time::Duration::from_secs(SERIAL_OPEN_PORT_TIMEOUT_SECS), ctx.serial_manager - .open_port(&port, baud_rate, &request_id, None), + .open_port(&port, baud_rate, &request_id, None, None), ) .await; match open_result { @@ -295,7 +295,7 @@ pub async fn monitor( || req.timeout.is_some(); if has_conditions { - let mut rx = match ctx.serial_manager.attach_reader(&port, &request_id) { + let mut rx = match ctx.serial_manager.attach_reader(&port, &request_id, None) { Some(rx) => rx, None => { return ( diff --git a/crates/fbuild-daemon/src/handlers/websockets.rs b/crates/fbuild-daemon/src/handlers/websockets.rs index f1000817..13771b28 100644 --- a/crates/fbuild-daemon/src/handlers/websockets.rs +++ b/crates/fbuild-daemon/src/handlers/websockets.rs @@ -66,21 +66,24 @@ pub async fn ws_serial_monitor( /// `open_port` to complete its USB re-enumeration retries). struct PendingAttachGuard { ctx: Arc, + id: u64, } impl PendingAttachGuard { fn new(ctx: Arc) -> Self { - ctx.pending_serial_attaches - .fetch_add(1, std::sync::atomic::Ordering::Relaxed); - Self { ctx } + let id = ctx.begin_pending_serial_attach(); + Self { ctx, id } + } + + fn set_target(&self, client_id: String, port: String) { + self.ctx + .update_pending_serial_attach(self.id, client_id, port); } } impl Drop for PendingAttachGuard { fn drop(&mut self) { - self.ctx - .pending_serial_attaches - .fetch_sub(1, std::sync::atomic::Ordering::Relaxed); + self.ctx.end_pending_serial_attach(self.id); } } @@ -115,7 +118,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { // Mark this attach as pending so the self-eviction loop won't shut the // daemon down while we're waiting for `open_port` to finish (USB // re-enumeration on Windows can take 10+ seconds). - let _attach_guard = PendingAttachGuard::new(ctx.clone()); + let attach_guard = PendingAttachGuard::new(ctx.clone()); // Wait for the attach message (FastLED/fbuild#808: bounded so an // idle client cannot tie up a tokio task slot forever). @@ -134,7 +137,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { return; } }; - let (client_id, port, baud_rate, pre_acquire_writer) = match first_frame { + let (client_id, port, baud_rate, pre_acquire_writer, client_metadata) = match first_frame { Some(Ok(Message::Text(text))) => { match serde_json::from_str::(&text) { Ok(SerialClientMessage::Attach { @@ -143,12 +146,14 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { baud_rate, open_if_needed, pre_acquire_writer, + client_metadata, }) => { + attach_guard.set_target(client_id.clone(), port.clone()); // Open port if needed if open_if_needed { if let Err(e) = ctx .serial_manager - .open_port(&port, baud_rate, &client_id, None) + .open_port(&port, baud_rate, &client_id, None, client_metadata.clone()) .await { let err_msg = SerialServerMessage::Error { @@ -160,7 +165,13 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { return; } } - (client_id, port, baud_rate, pre_acquire_writer) + ( + client_id, + port, + baud_rate, + pre_acquire_writer, + client_metadata, + ) } Ok(_) => { let err_msg = SerialServerMessage::Error { @@ -202,7 +213,10 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { }; // Attach reader - let mut rx = match ctx.serial_manager.attach_reader(&port, &client_id) { + let mut rx = match ctx + .serial_manager + .attach_reader(&port, &client_id, client_metadata.clone()) + { Some(rx) => rx, None => { let err_msg = SerialServerMessage::Error { @@ -230,6 +244,7 @@ async fn handle_serial_ws(mut socket: WebSocket, ctx: Arc) { cleanup_ws_serial_session(&ctx, &port, &client_id, writer_acquired).await; return; } + drop(attach_guard); // Concurrent reader / writer / inbound split (issue #749). // diff --git a/crates/fbuild-daemon/src/lib.rs b/crates/fbuild-daemon/src/lib.rs index 8edd9012..1afeba61 100644 --- a/crates/fbuild-daemon/src/lib.rs +++ b/crates/fbuild-daemon/src/lib.rs @@ -33,6 +33,7 @@ pub mod broker; pub mod context; pub mod device_manager; pub mod handlers; +pub mod lock_models; pub mod log_layer; pub mod models; pub mod status_manager; diff --git a/crates/fbuild-daemon/src/lock_models.rs b/crates/fbuild-daemon/src/lock_models.rs new file mode 100644 index 00000000..289fa7c2 --- /dev/null +++ b/crates/fbuild-daemon/src/lock_models.rs @@ -0,0 +1,91 @@ +//! Lock status and cleanup request/response models. + +use serde::{Deserialize, Serialize}; + +/// GET /api/locks/status response. +#[derive(Debug, Serialize)] +pub struct LockStatusResponse { + pub success: bool, + pub port_locks: Vec, + pub project_locks: Vec, + pub pending_serial_attaches: Vec, + pub stale_locks: Vec, +} + +/// Lock information for a serial port. +#[derive(Debug, Serialize)] +pub struct PortLockInfo { + pub port: String, + pub is_held: bool, + pub holder_description: Option, + pub is_open: bool, + pub owner_client_id: Option, + pub writer_client_id: Option, + pub reader_count: usize, + pub reader_client_ids: Vec, + pub baud_rate: u32, + pub started_at: f64, + pub session_age_seconds: f64, + pub last_activity_at: f64, + pub last_activity_age_seconds: f64, + pub last_read_at: Option, + pub last_read_age_seconds: Option, + pub last_write_at: Option, + pub last_write_age_seconds: Option, + pub total_bytes_read: u64, + pub total_bytes_written: u64, + pub clients: Vec, +} + +/// Best-effort owner metadata for a serial session client. +#[derive(Debug, Serialize)] +pub struct SerialClientLockInfo { + pub client_id: String, + pub pid: Option, + pub process_alive: Option, + pub exe: Option, + pub cwd: Option, + pub argv: Option>, +} + +/// WebSocket attach currently in progress. +#[derive(Debug, Serialize)] +pub struct PendingSerialAttachLockInfo { + pub id: u64, + pub client_id: Option, + pub port: Option, + pub started_at: f64, + pub age_seconds: f64, +} + +/// Lock information for a project directory. +#[derive(Debug, Serialize)] +pub struct ProjectLockInfo { + pub project_dir: String, + pub is_held: bool, +} + +/// POST /api/locks/clear request. +#[derive(Debug, Default, Deserialize)] +pub struct ClearLocksRequest { + #[serde(default)] + pub serial: bool, + #[serde(default)] + pub stale: bool, + pub port: Option, + pub client_id: Option, + #[serde(default)] + pub force: bool, +} + +/// POST /api/locks/clear response. +#[derive(Debug, Serialize)] +pub struct ClearLocksResponse { + pub success: bool, + pub cleared_count: usize, + pub cleared_project_count: usize, + pub cleared_serial_count: usize, + pub cleared_serial_sessions: Vec, + pub refused: Vec, + pub message: String, +} diff --git a/crates/fbuild-daemon/src/models.rs b/crates/fbuild-daemon/src/models.rs index 73ca181d..aa3d8e02 100644 --- a/crates/fbuild-daemon/src/models.rs +++ b/crates/fbuild-daemon/src/models.rs @@ -4,6 +4,11 @@ use std::collections::BTreeMap; use serde::{Deserialize, Serialize}; +pub use crate::lock_models::{ + ClearLocksRequest, ClearLocksResponse, LockStatusResponse, PendingSerialAttachLockInfo, + PortLockInfo, ProjectLockInfo, SerialClientLockInfo, +}; + /// POST /api/build #[derive(Debug, Deserialize)] pub struct BuildRequest { @@ -310,41 +315,6 @@ pub struct DeviceListResponse { pub devices: Vec, } -/// GET /api/locks/status response. -#[derive(Debug, Serialize)] -pub struct LockStatusResponse { - pub success: bool, - pub port_locks: Vec, - pub project_locks: Vec, - pub stale_locks: Vec, -} - -/// Lock information for a serial port. -#[derive(Debug, Serialize)] -pub struct PortLockInfo { - pub port: String, - pub is_held: bool, - pub holder_description: Option, - pub is_open: bool, - pub writer_client_id: Option, - pub reader_count: usize, -} - -/// Lock information for a project directory. -#[derive(Debug, Serialize)] -pub struct ProjectLockInfo { - pub project_dir: String, - pub is_held: bool, -} - -/// POST /api/locks/clear response. -#[derive(Debug, Serialize)] -pub struct ClearLocksResponse { - pub success: bool, - pub cleared_count: usize, - pub message: String, -} - /// POST /api/install-deps request. #[derive(Debug, Deserialize)] pub struct InstallDepsRequest { diff --git a/crates/fbuild-python/src/async_serial_monitor.rs b/crates/fbuild-python/src/async_serial_monitor.rs index cb756aff..f8276598 100644 --- a/crates/fbuild-python/src/async_serial_monitor.rs +++ b/crates/fbuild-python/src/async_serial_monitor.rs @@ -137,6 +137,7 @@ impl AsyncSerialMonitor { baud_rate, open_if_needed: true, pre_acquire_writer: true, + client_metadata: Some(crate::messages::ClientMetadata::current()), }; let attach_json = serde_json::to_string(&attach) .expect("fbuild-python: ClientMessage::Attach serialization is infallible"); diff --git a/crates/fbuild-python/src/messages.rs b/crates/fbuild-python/src/messages.rs index e2b1edd9..5eaa06c8 100644 --- a/crates/fbuild-python/src/messages.rs +++ b/crates/fbuild-python/src/messages.rs @@ -9,6 +9,33 @@ pub(crate) type WsStream = pub(crate) type WsSink = futures::stream::SplitSink; pub(crate) type WsSource = futures::stream::SplitStream; +#[derive(Debug, Clone, Serialize)] +pub(crate) struct ClientMetadata { + #[serde(skip_serializing_if = "Option::is_none")] + pub pid: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub exe: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cwd: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub argv: Option>, +} + +impl ClientMetadata { + pub(crate) fn current() -> Self { + Self { + pid: Some(std::process::id()), + exe: std::env::current_exe() + .ok() + .map(|p| p.to_string_lossy().into_owned()), + cwd: std::env::current_dir() + .ok() + .map(|p| p.to_string_lossy().into_owned()), + argv: Some(std::env::args().collect()), + } + } +} + /// Messages we receive from the daemon (subset we care about). #[derive(Debug, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -99,6 +126,8 @@ pub(crate) enum ClientMessage { baud_rate: u32, open_if_needed: bool, pre_acquire_writer: bool, + #[serde(skip_serializing_if = "Option::is_none")] + client_metadata: Option, }, Write { data: String, diff --git a/crates/fbuild-python/src/serial_monitor.rs b/crates/fbuild-python/src/serial_monitor.rs index 46c2d65d..652e5542 100644 --- a/crates/fbuild-python/src/serial_monitor.rs +++ b/crates/fbuild-python/src/serial_monitor.rs @@ -84,6 +84,7 @@ impl SerialMonitor { baud_rate: self.baud_rate, open_if_needed: true, pre_acquire_writer: true, + client_metadata: Some(crate::messages::ClientMetadata::current()), }; let attach_json = serde_json::to_string(&attach) .expect("fbuild-python: ClientMessage::Attach serialization is infallible"); diff --git a/crates/fbuild-serial/src/lib.rs b/crates/fbuild-serial/src/lib.rs index 0ab888f2..bb76b6eb 100644 --- a/crates/fbuild-serial/src/lib.rs +++ b/crates/fbuild-serial/src/lib.rs @@ -42,6 +42,8 @@ pub mod preemption; pub mod rpc_validate; pub mod session; -pub use manager::{PortSessionInfo, SharedSerialManager}; -pub use messages::{SerialClientMessage, SerialServerMessage, SerialStreamEvent}; +pub use manager::{PortSessionInfo, SerialClientInfo, SharedSerialManager}; +pub use messages::{ + SerialClientMessage, SerialClientMetadata, SerialServerMessage, SerialStreamEvent, +}; pub use session::SerialSession; diff --git a/crates/fbuild-serial/src/manager.rs b/crates/fbuild-serial/src/manager.rs index 024e1580..dd9d3e69 100644 --- a/crates/fbuild-serial/src/manager.rs +++ b/crates/fbuild-serial/src/manager.rs @@ -5,7 +5,7 @@ //! readers, exclusive writer) and the Windows USB-CDC write strategy. use crate::crash_decoder::CrashDecoder; -use crate::messages::SerialStreamEvent; +use crate::messages::{SerialClientMetadata, SerialStreamEvent}; use crate::preemption::PreemptionTracker; use crate::session::SerialSession; use dashmap::DashMap; @@ -19,11 +19,31 @@ const OUTPUT_BUFFER_CAP: usize = 10_000; const BROADCAST_CHANNEL_SIZE: usize = 1024; const READ_BUF_SIZE: usize = 4096; +fn now_unix_secs() -> f64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64() +} + +fn now_unix_millis() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_millis() + .min(u128::from(u64::MAX)) as u64 +} + +fn millis_to_unix_secs(ms: u64) -> Option { + (ms > 0).then(|| ms as f64 / 1000.0) +} + /// Per-port output buffer shared with the background reader. Separate from /// `SerialSession` so we don't need `SerialSession: Clone`. struct PortOutputBuffer { buffer: std::sync::Mutex>, total_bytes_read: std::sync::atomic::AtomicU64, + last_read_at_ms: std::sync::atomic::AtomicU64, } /// Central serial port manager. One instance per daemon. @@ -76,11 +96,19 @@ impl SharedSerialManager { baud_rate: u32, client_id: &str, family: Option, + client_metadata: Option, ) -> fbuild_core::Result<()> { let session_key = self.resolve_port_key(port); // If already open, just return Ok - if let Some(session) = self.sessions.get(&session_key) { + if let Some(mut session) = self.sessions.get_mut(&session_key) { if session.is_open { + session.last_activity_at = now_unix_secs(); + if let Some(metadata) = client_metadata { + session + .client_metadata + .insert(client_id.to_string(), metadata); + } + drop(session); self.bump_close_generation(&session_key); tracing::info!(port, client_id, "port already open, reusing"); return Ok(()); @@ -199,6 +227,7 @@ impl SharedSerialManager { let port_buf = Arc::new(PortOutputBuffer { buffer: std::sync::Mutex::new(VecDeque::with_capacity(OUTPUT_BUFFER_CAP)), total_bytes_read: std::sync::atomic::AtomicU64::new(0), + last_read_at_ms: std::sync::atomic::AtomicU64::new(0), }); self.output_buffers .insert(port_name.clone(), Arc::clone(&port_buf)); @@ -228,6 +257,9 @@ impl SharedSerialManager { port_buf .total_bytes_read .fetch_add(n as u64, Ordering::Relaxed); + port_buf + .last_read_at_ms + .store(now_unix_millis(), Ordering::Relaxed); // Split into complete lines while let Some(newline_pos) = partial_line.find('\n') { @@ -284,15 +316,19 @@ impl SharedSerialManager { }; let mut session = SerialSession::new(port_name.clone(), baud_rate); + let now = now_unix_secs(); session.is_open = true; session.owner_client_id = Some(client_id.to_string()); + if let Some(metadata) = client_metadata { + session + .client_metadata + .insert(client_id.to_string(), metadata); + } session.serial_handle = Some(serial_handle); session.reader_handle = Some(reader_handle); session.stop_flag = stop_flag; - session.started_at = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs_f64(); + session.started_at = now; + session.last_activity_at = now; self.sessions.insert(port_name.clone(), session); self.bump_close_generation(&port_name); @@ -399,7 +435,10 @@ impl SharedSerialManager { // Update stats if let Some(mut session) = self.sessions.get_mut(&session_key) { + let now = now_unix_secs(); session.total_bytes_written += bytes_written as u64; + session.last_write_at = Some(now); + session.last_activity_at = now; } Ok(bytes_written) @@ -604,6 +643,7 @@ impl SharedSerialManager { &self, port: &str, client_id: &str, + client_metadata: Option, ) -> Option> { let session_key = self.resolve_port_key(port); let rx = self @@ -611,7 +651,13 @@ impl SharedSerialManager { .get(&session_key) .map(|tx| tx.subscribe())?; if let Some(mut session) = self.sessions.get_mut(&session_key) { + session.last_activity_at = now_unix_secs(); session.reader_client_ids.insert(client_id.to_string()); + if let Some(metadata) = client_metadata { + session + .client_metadata + .insert(client_id.to_string(), metadata); + } drop(session); self.bump_close_generation(&session_key); } @@ -623,6 +669,7 @@ impl SharedSerialManager { let session_key = self.resolve_port_key(port); if let Some(mut session) = self.sessions.get_mut(&session_key) { session.reader_client_ids.remove(client_id); + session.last_activity_at = now_unix_secs(); } } @@ -729,6 +776,7 @@ impl SharedSerialManager { Arc::new(PortOutputBuffer { buffer: std::sync::Mutex::new(VecDeque::with_capacity(OUTPUT_BUFFER_CAP)), total_bytes_read: std::sync::atomic::AtomicU64::new(0), + last_read_at_ms: std::sync::atomic::AtomicU64::new(0), }) }) .clone(); @@ -817,6 +865,7 @@ impl SharedSerialManager { ))); } session.writer_client_id = Some(client_id.to_string()); + session.last_activity_at = now_unix_secs(); drop(session); self.bump_close_generation(&session_key); Ok(()) @@ -834,6 +883,7 @@ impl SharedSerialManager { if let Some(mut session) = self.sessions.get_mut(&session_key) { if session.writer_client_id.as_deref() == Some(client_id) { session.writer_client_id = None; + session.last_activity_at = now_unix_secs(); } } } @@ -907,18 +957,53 @@ impl SharedSerialManager { .iter() .map(|entry| { let s = entry.value(); + let last_read_at = self.output_buffers.get(entry.key()).and_then(|buf| { + millis_to_unix_secs(buf.last_read_at_ms.load(Ordering::Relaxed)) + }); + let total_bytes_read = self + .output_buffers + .get(entry.key()) + .map(|buf| buf.total_bytes_read.load(Ordering::Relaxed)) + .unwrap_or(s.total_bytes_read); + let mut reader_client_ids: Vec = + s.reader_client_ids.iter().cloned().collect(); + reader_client_ids.sort(); + let mut clients: Vec = s + .client_metadata + .iter() + .map(|(client_id, metadata)| SerialClientInfo { + client_id: client_id.clone(), + metadata: metadata.clone(), + }) + .collect(); + clients.sort_by(|a, b| a.client_id.cmp(&b.client_id)); PortSessionInfo { port: s.port.clone(), is_open: s.is_open, writer_client_id: s.writer_client_id.clone(), reader_count: s.reader_client_ids.len(), + reader_client_ids, owner_client_id: s.owner_client_id.clone(), baud_rate: s.baud_rate, + started_at: s.started_at, + last_activity_at: s.last_activity_at, + last_read_at, + last_write_at: s.last_write_at, + total_bytes_read, + total_bytes_written: s.total_bytes_written, + clients, } }) .collect() } + /// Test hook for higher-level daemon handlers that need an in-memory + /// serial session without opening a real OS port. + #[cfg(any(test, debug_assertions))] + pub fn insert_session_for_test(&self, session: SerialSession) { + self.sessions.insert(session.port.clone(), session); + } + async fn open_physical_serial( port: &str, baud_rate: u32, @@ -1006,6 +1091,9 @@ impl SharedSerialManager { port_buf .total_bytes_read .fetch_add(n as u64, Ordering::Relaxed); + port_buf + .last_read_at_ms + .store(now_unix_millis(), Ordering::Relaxed); while let Some(newline_pos) = partial_line.find('\n') { let line = partial_line[..newline_pos].trim_end().to_string(); @@ -1087,8 +1175,23 @@ pub struct PortSessionInfo { pub is_open: bool, pub writer_client_id: Option, pub reader_count: usize, + pub reader_client_ids: Vec, pub owner_client_id: Option, pub baud_rate: u32, + pub started_at: f64, + pub last_activity_at: f64, + pub last_read_at: Option, + pub last_write_at: Option, + pub total_bytes_read: u64, + pub total_bytes_written: u64, + pub clients: Vec, +} + +/// Snapshot of a serial client attached to a port session. +#[derive(Debug, Clone)] +pub struct SerialClientInfo { + pub client_id: String, + pub metadata: SerialClientMetadata, } impl Default for SharedSerialManager { diff --git a/crates/fbuild-serial/src/manager/tests.rs b/crates/fbuild-serial/src/manager/tests.rs index d7658a5a..96fdbbc2 100644 --- a/crates/fbuild-serial/src/manager/tests.rs +++ b/crates/fbuild-serial/src/manager/tests.rs @@ -151,7 +151,7 @@ fn open_port_does_not_starve_runtime_workers() { let open_task = tokio::spawn(async move { let _ = mgr_open - .open_port(&bogus_port, 115200, "test_client", None) + .open_port(&bogus_port, 115200, "test_client", None, None) .await; }); @@ -215,7 +215,10 @@ fn attach_reader_missing_broadcaster_does_not_mutate_session_state() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: None, + client_metadata: Default::default(), elf_path: None, serial_handle: None, reader_handle: None, @@ -223,7 +226,7 @@ fn attach_reader_missing_broadcaster_does_not_mutate_session_state() { }, ); - let result = mgr.attach_reader(port, client); + let result = mgr.attach_reader(port, client, None); assert!( result.is_none(), "attach_reader must return None when broadcaster is absent" @@ -273,7 +276,10 @@ async fn detach_then_close_releases_port_for_lone_monitor() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(client.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: None, reader_handle: None, @@ -325,7 +331,10 @@ async fn grace_close_removes_idle_port_after_delay() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(client.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: None, reader_handle: None, @@ -363,7 +372,10 @@ async fn grace_close_is_canceled_by_new_reader() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(client.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: None, reader_handle: None, @@ -372,7 +384,7 @@ async fn grace_close_is_canceled_by_new_reader() { ); assert!(mgr.close_port_after_grace_if_idle(port, client, Duration::from_millis(25))); - let rx = mgr.attach_reader(port, next_client); + let rx = mgr.attach_reader(port, next_client, None); assert!( rx.is_some(), "new reader should attach during pending close grace window" @@ -409,7 +421,10 @@ async fn stale_deploy_preemption_close_preserves_new_monitor_session() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(original_client.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: Some(Arc::new(Mutex::new(Box::new(fake)))), reader_handle: None, @@ -422,7 +437,7 @@ async fn stale_deploy_preemption_close_preserves_new_monitor_session() { // close must not remove the new monitor session or later writes fail // with "port not open" (FastLED/fbuild#811). let stale_generation = mgr.bump_close_generation(port); - assert!(mgr.attach_reader(port, monitor_client).is_some()); + assert!(mgr.attach_reader(port, monitor_client, None).is_some()); assert_ne!(mgr.close_generation(port), Some(stale_generation)); let closed = mgr @@ -494,6 +509,7 @@ async fn rebind_preserves_session_and_routes_writes_to_new_handle() { Arc::new(PortOutputBuffer { buffer: std::sync::Mutex::new(VecDeque::with_capacity(OUTPUT_BUFFER_CAP)), total_bytes_read: std::sync::atomic::AtomicU64::new(0), + last_read_at_ms: std::sync::atomic::AtomicU64::new(0), }), ); let (old_fake, _old_writes) = FakeSerialPort::new(old_port); @@ -511,7 +527,10 @@ async fn rebind_preserves_session_and_routes_writes_to_new_handle() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(writer.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: Some(Arc::new(Mutex::new(Box::new(old_fake)))), reader_handle: None, @@ -722,7 +741,10 @@ async fn write_to_port_completes_partial_writes_and_reports_full_length() { total_bytes_read: 0, total_bytes_written: 0, started_at: 0.0, + last_activity_at: 0.0, + last_write_at: None, owner_client_id: Some(writer.to_string()), + client_metadata: Default::default(), elf_path: None, serial_handle: Some(Arc::new(Mutex::new(Box::new(fake)))), reader_handle: None, diff --git a/crates/fbuild-serial/src/messages.rs b/crates/fbuild-serial/src/messages.rs index 1ee48438..10f60ed6 100644 --- a/crates/fbuild-serial/src/messages.rs +++ b/crates/fbuild-serial/src/messages.rs @@ -6,6 +6,22 @@ use serde::{Deserialize, Serialize}; +/// Best-effort owner metadata supplied by serial monitor clients. +/// +/// Older clients omit this entirely; the daemon treats every field as +/// optional so the attach protocol stays backward-compatible. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct SerialClientMetadata { + #[serde(default, skip_serializing_if = "Option::is_none")] + pub pid: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub exe: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub cwd: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub argv: Option>, +} + /// Messages sent by the client to the daemon. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "type", rename_all = "snake_case")] @@ -16,6 +32,8 @@ pub enum SerialClientMessage { baud_rate: u32, open_if_needed: bool, pre_acquire_writer: bool, + #[serde(default, skip_serializing_if = "Option::is_none")] + client_metadata: Option, }, Write { /// Base64-encoded data. @@ -134,9 +152,16 @@ mod tests { baud_rate: 115200, open_if_needed: true, pre_acquire_writer: false, + client_metadata: Some(SerialClientMetadata { + pid: Some(1234), + exe: Some("python".into()), + cwd: Some("/work".into()), + argv: Some(vec!["python".into(), "-".into()]), + }), }; let json = serde_json::to_string(&msg).unwrap(); assert!(json.contains("\"type\":\"attach\"")); + assert!(json.contains("\"pid\":1234")); let parsed: SerialClientMessage = serde_json::from_str(&json).unwrap(); match parsed { SerialClientMessage::Attach { @@ -145,17 +170,31 @@ mod tests { baud_rate, open_if_needed, pre_acquire_writer, + client_metadata, } => { assert_eq!(client_id, "c1"); assert_eq!(port, "COM3"); assert_eq!(baud_rate, 115200); assert!(open_if_needed); assert!(!pre_acquire_writer); + assert_eq!(client_metadata.unwrap().pid, Some(1234)); } _ => panic!("expected Attach"), } } + #[test] + fn client_attach_without_metadata_still_deserializes() { + let json = r#"{"type":"attach","client_id":"c1","port":"COM3","baud_rate":115200,"open_if_needed":true,"pre_acquire_writer":false}"#; + let parsed: SerialClientMessage = serde_json::from_str(json).unwrap(); + match parsed { + SerialClientMessage::Attach { + client_metadata, .. + } => assert!(client_metadata.is_none()), + _ => panic!("expected Attach"), + } + } + #[test] fn client_write_roundtrip() { let msg = SerialClientMessage::Write { diff --git a/crates/fbuild-serial/src/session.rs b/crates/fbuild-serial/src/session.rs index a6ecb338..63252d4f 100644 --- a/crates/fbuild-serial/src/session.rs +++ b/crates/fbuild-serial/src/session.rs @@ -1,7 +1,8 @@ //! Per-port serial session state. -use std::collections::HashSet; +use crate::messages::SerialClientMetadata; use std::collections::VecDeque; +use std::collections::{HashMap, HashSet}; use std::path::PathBuf; use std::sync::atomic::AtomicBool; use std::sync::Arc; @@ -22,8 +23,12 @@ pub struct SerialSession { pub total_bytes_read: u64, pub total_bytes_written: u64, pub started_at: f64, + pub last_activity_at: f64, + pub last_write_at: Option, /// Client that opened the port. pub owner_client_id: Option, + /// Best-effort metadata keyed by serial client id. + pub client_metadata: HashMap, /// Path to firmware ELF for crash decoding. pub elf_path: Option, /// The underlying serial port handle. @@ -36,6 +41,10 @@ pub struct SerialSession { impl SerialSession { pub fn new(port: String, baud_rate: u32) -> Self { + let now = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs_f64(); Self { port, baud_rate, @@ -45,8 +54,11 @@ impl SerialSession { output_buffer: VecDeque::with_capacity(10_000), total_bytes_read: 0, total_bytes_written: 0, - started_at: 0.0, + started_at: now, + last_activity_at: now, + last_write_at: None, owner_client_id: None, + client_metadata: HashMap::new(), elf_path: None, serial_handle: None, reader_handle: None,