From 0ae576020e32814199d6bbad6f6eee25202b7c5a Mon Sep 17 00:00:00 2001 From: nathanfraske Date: Sat, 25 Jul 2026 17:33:38 -0500 Subject: [PATCH 1/2] Harden KVM input and local video recovery --- gui/src-tauri/Cargo.lock | 20 +-- gui/src-tauri/src/backend_recovery.rs | 194 +++++++++++++++++++++++ gui/src-tauri/src/main.rs | 220 +++++++++++++++++++------- gui/src/store.svelte.ts | 12 +- gui/src/tauri.ts | 17 +- gui/src/ui/Console.svelte | 44 ++++-- gui/src/ui/VideoPopout.svelte | 16 +- node/Cargo.lock | 16 +- node/src/mesh.rs | 18 ++- node/src/node_control.rs | 9 +- node/src/video_decode.rs | 195 ++++++++++++++++++++--- 11 files changed, 635 insertions(+), 126 deletions(-) create mode 100644 gui/src-tauri/src/backend_recovery.rs diff --git a/gui/src-tauri/Cargo.lock b/gui/src-tauri/Cargo.lock index 90291e7..9fea0ec 100644 --- a/gui/src-tauri/Cargo.lock +++ b/gui/src-tauri/Cargo.lock @@ -19,7 +19,7 @@ dependencies = [ [[package]] name = "allmystuff-bridge" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "allmystuff-inventory", @@ -28,7 +28,7 @@ dependencies = [ [[package]] name = "allmystuff-cec-consent" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-cec-protocol", "serde", @@ -38,7 +38,7 @@ dependencies = [ [[package]] name = "allmystuff-cec-protocol" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -47,7 +47,7 @@ dependencies = [ [[package]] name = "allmystuff-graph" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -92,7 +92,7 @@ dependencies = [ [[package]] name = "allmystuff-inventory" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -101,7 +101,7 @@ dependencies = [ [[package]] name = "allmystuff-node" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-bridge", "allmystuff-cec-consent", @@ -151,7 +151,7 @@ version = "0.1.0" [[package]] name = "allmystuff-protocol" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "serde", @@ -160,7 +160,7 @@ dependencies = [ [[package]] name = "allmystuff-service" -version = "0.2.46" +version = "0.2.48" dependencies = [ "anyhow", "dirs 5.0.1", @@ -169,7 +169,7 @@ dependencies = [ [[package]] name = "allmystuff-session" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "allmystuff-protocol", @@ -180,7 +180,7 @@ dependencies = [ [[package]] name = "allmystuff-updater" -version = "0.2.46" +version = "0.2.48" dependencies = [ "dirs 5.0.1", "flate2", diff --git a/gui/src-tauri/src/backend_recovery.rs b/gui/src-tauri/src/backend_recovery.rs new file mode 100644 index 0000000..eb4aaf2 --- /dev/null +++ b/gui/src-tauri/src/backend_recovery.rs @@ -0,0 +1,194 @@ +//! Pure decision state for the GUI's local node recovery loop. +//! +//! This module does not start, stop, or contact a process. It turns observed +//! local socket and child-process state into one decision, which keeps the +//! durability policy deterministic and unit-testable. + +pub(crate) const WEDGED_RESTART_ROUNDS: u32 = 3; + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum SocketState { + Ready, + /// The first probe failed, but the node bound its socket during the + /// existing startup grace window. + ReadyAfterGrace, + Unresponsive, +} + +impl SocketState { + pub(crate) const fn label(self) -> &'static str { + match self { + Self::Ready => "ready", + Self::ReadyAfterGrace => "ready_after_grace", + Self::Unresponsive => "unresponsive", + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum Ownership { + GuiOwnedAlive, + /// The GUI either reused a service-owned node or its own child is dead. + NoLiveGuiChild, +} + +impl Ownership { + pub(crate) const fn label(self) -> &'static str { + match self { + Self::GuiOwnedAlive => "gui_owned_alive", + Self::NoLiveGuiChild => "no_live_gui_child", + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct Observation { + pub(crate) socket: SocketState, + pub(crate) ownership: Ownership, + pub(crate) generation: u64, +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum Decision { + Healthy, + StartupCompleted, + WaitForOwnedNode, + RestartOwnedNode, + EnsureNode, +} + +impl Decision { + pub(crate) const fn label(self) -> &'static str { + match self { + Self::Healthy => "healthy", + Self::StartupCompleted => "startup_completed", + Self::WaitForOwnedNode => "wait_for_owned_node", + Self::RestartOwnedNode => "restart_owned_node", + Self::EnsureNode => "ensure_node", + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) struct Evidence { + pub(crate) observation: Observation, + pub(crate) decision: Decision, + pub(crate) prior_wedged_rounds: u32, + pub(crate) wedged_rounds: u32, +} + +#[derive(Debug, Default)] +pub(crate) struct BackendRecovery { + wedged_rounds: u32, +} + +impl BackendRecovery { + pub(crate) fn observe(&mut self, observation: Observation) -> Evidence { + let prior_wedged_rounds = self.wedged_rounds; + let decision = match (observation.socket, observation.ownership) { + (SocketState::Ready, _) => { + self.wedged_rounds = 0; + Decision::Healthy + } + (SocketState::ReadyAfterGrace, _) => { + self.wedged_rounds = 0; + Decision::StartupCompleted + } + (SocketState::Unresponsive, Ownership::NoLiveGuiChild) => { + self.wedged_rounds = 0; + Decision::EnsureNode + } + (SocketState::Unresponsive, Ownership::GuiOwnedAlive) => { + self.wedged_rounds = self.wedged_rounds.saturating_add(1); + if self.wedged_rounds >= WEDGED_RESTART_ROUNDS { + self.wedged_rounds = 0; + Decision::RestartOwnedNode + } else { + Decision::WaitForOwnedNode + } + } + }; + Evidence { + observation, + decision, + prior_wedged_rounds, + wedged_rounds: self.wedged_rounds, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn observation(socket: SocketState, ownership: Ownership, generation: u64) -> Observation { + Observation { + socket, + ownership, + generation, + } + } + + #[test] + fn live_socket_clears_wedge_history() { + let mut recovery = BackendRecovery::default(); + let _ = recovery.observe(observation( + SocketState::Unresponsive, + Ownership::GuiOwnedAlive, + 4, + )); + + let evidence = + recovery.observe(observation(SocketState::Ready, Ownership::GuiOwnedAlive, 4)); + assert_eq!(evidence.decision, Decision::Healthy); + assert_eq!(evidence.prior_wedged_rounds, 1); + assert_eq!(evidence.wedged_rounds, 0); + assert_eq!(evidence.observation.generation, 4); + } + + #[test] + fn starting_node_that_binds_during_grace_is_not_restarted() { + let mut recovery = BackendRecovery::default(); + let evidence = recovery.observe(observation( + SocketState::ReadyAfterGrace, + Ownership::GuiOwnedAlive, + 7, + )); + assert_eq!(evidence.decision, Decision::StartupCompleted); + assert_eq!(evidence.wedged_rounds, 0); + } + + #[test] + fn dead_or_external_node_is_ensured_without_owner_kill() { + let mut recovery = BackendRecovery::default(); + let evidence = recovery.observe(observation( + SocketState::Unresponsive, + Ownership::NoLiveGuiChild, + 0, + )); + assert_eq!(evidence.decision, Decision::EnsureNode); + assert_eq!(evidence.wedged_rounds, 0); + } + + #[test] + fn live_owned_but_wedged_node_restarts_on_existing_third_round() { + let mut recovery = BackendRecovery::default(); + for expected in 1..WEDGED_RESTART_ROUNDS { + let evidence = recovery.observe(observation( + SocketState::Unresponsive, + Ownership::GuiOwnedAlive, + 9, + )); + assert_eq!(evidence.decision, Decision::WaitForOwnedNode); + assert_eq!(evidence.wedged_rounds, expected); + } + let evidence = recovery.observe(observation( + SocketState::Unresponsive, + Ownership::GuiOwnedAlive, + 9, + )); + assert_eq!(evidence.decision, Decision::RestartOwnedNode); + assert_eq!(evidence.prior_wedged_rounds, 2); + assert_eq!(evidence.wedged_rounds, 0); + } +} diff --git a/gui/src-tauri/src/main.rs b/gui/src-tauri/src/main.rs index 77211a1..5c86601 100644 --- a/gui/src-tauri/src/main.rs +++ b/gui/src-tauri/src/main.rs @@ -41,14 +41,53 @@ use serde_json::{json, Value}; use tauri::{Emitter, Manager, RunEvent, State}; use tauri_plugin_autostart::ManagerExt; +mod backend_recovery; mod window_behavior; +use backend_recovery::{ + BackendRecovery, Decision as RecoveryDecision, Observation as RecoveryObservation, + Ownership as RecoveryOwnership, SocketState as RecoverySocketState, WEDGED_RESTART_ROUNDS, +}; + +#[derive(Default)] +struct OwnedNode { + child: Option, + /// Monotonic identity for GUI-owned node processes. Reusing an external + /// service node leaves this unchanged because the GUI does not own it. + generation: u64, +} + +impl OwnedNode { + fn install(&mut self, child: NodeChild) -> u64 { + self.generation = self.generation.saturating_add(1); + self.child = Some(child); + self.generation + } + + fn is_alive(&mut self) -> bool { + self.child.as_mut().is_some_and(NodeChild::is_alive) + } + + fn observation(&mut self) -> (RecoveryOwnership, u64) { + let ownership = if self.is_alive() { + RecoveryOwnership::GuiOwnedAlive + } else { + RecoveryOwnership::NoLiveGuiChild + }; + (ownership, self.generation) + } + + fn take(&mut self) { + self.child.take(); + } +} + struct AppState { node: Arc, /// The node we spawned, if Always-On wasn't already running one. Held so /// it's killed when the app exits (Always-On off => node lives only with /// the app); a reused service node has no child here and keeps running. - node_child: Mutex>, + node_child: Mutex, } // ---- this machine ----------------------------------------------------- @@ -413,13 +452,18 @@ async fn clipboard_pull(state: State<'_, AppState>, route_id: String) -> Result< /// frames — for webviews without WebCodecs, and the bottom rung of the /// console's decode ladder. #[tauri::command] -async fn video_watch(app: tauri::AppHandle, route_id: String, decode: Option) -> u64 { +async fn video_watch( + app: tauri::AppHandle, + route_id: String, + decode: Option, + decoder: Option, +) -> u64 { let state = app.state::(); match state .node .request( "video_watch", - json!({ "route_id": route_id, "decode": decode }), + json!({ "route_id": route_id, "decode": decode, "decoder": decoder }), ) .await { @@ -976,13 +1020,23 @@ async fn open_video_window( app: tauri::AppHandle, key: String, title: String, + decoder: Option, ) -> Result<(), String> { + let decoder = match decoder.as_deref() { + None | Some("automatic") => "automatic", + Some("software") => "software", + Some(other) => { + return Err(format!( + "unsupported local decoder preference {other:?} (automatic | software)" + )) + } + }; open_secondary_window( &app, &format!("video-{}", window_slug(&key)), // The key carries capability/route ids (colons, the route arrow) — // percent-encode so the query survives; URLSearchParams decodes. - format!("index.html?video={}", query_encode(&key)), + format!("index.html?video={}&decoder={decoder}", query_encode(&key)), &title, (880.0, 560.0), (380.0, 260.0), @@ -2217,19 +2271,13 @@ fn heal_node(app: &tauri::AppHandle) { tauri::async_runtime::spawn(async move { // Never heal over our own live serve — replacing its handle would // kill it (see the pump's wedge handling). - if handle - .state::() - .node_child - .lock() - .as_mut() - .map(|c| c.is_alive()) - .unwrap_or(false) - { + if handle.state::().node_child.lock().is_alive() { return; } match ensure_node_running().await { Ok(Some(child)) => { - handle.state::().node_child.lock().replace(child); + let generation = handle.state::().node_child.lock().install(child); + tracing::info!(generation, "installed GUI-owned node during heal"); } Ok(None) => {} Err(e) => tracing::error!("couldn't bring the allmystuff node back up: {e:#}"), @@ -2271,10 +2319,7 @@ fn apply_startup_behavior(app: &tauri::AppHandle) { /// in-process. Reconnects if the node restarts. async fn run_event_pump(app: tauri::AppHandle, node: Arc) { use tokio::sync::mpsc; - // Consecutive grace windows the socket stayed dead while OUR child kept - // running — the wedged-not-gone state. Only a repeat offender earns a - // deliberate, owner-controlled restart. - let mut wedged_rounds: u32 = 0; + let mut recovery = BackendRecovery::default(); loop { // The node may be *gone*, not just restarting — e.g. another client app // (CEC Support) spawned it and exited, taking the kill-on-close serve @@ -2284,64 +2329,111 @@ async fn run_event_pump(app: tauri::AppHandle, node: Arc) { // bound yet) must not read as "gone": respawning over it would // kill-on-drop the very child being waited on, and the stack would // flap spawn/kill forever. - let mut gone = !NodeClient::probe().await; - if gone { + let first_probe_ready = NodeClient::probe().await; + let mut socket = if first_probe_ready { + RecoverySocketState::Ready + } else { + RecoverySocketState::Unresponsive + }; + if !first_probe_ready { for _ in 0..50 { tokio::time::sleep(std::time::Duration::from_millis(200)).await; if NodeClient::probe().await { - gone = false; + socket = RecoverySocketState::ReadyAfterGrace; break; } } } - if gone { - // Socket dead through the grace window — but if OUR child is still - // running, the serve is alive behind a busy/wedged socket, not - // gone. Respawning then spawns a bind-loser and kills the live - // serve when the old handle is replaced: the spawn/kill metronome - // that made every peer connect/reconnect in a loop. Only respawn - // over a child we've confirmed dead; a serve that stays wedged for - // three straight windows gets a deliberate owner restart instead. - let own_alive = app - .state::() - .node_child - .lock() - .as_mut() - .map(|c| c.is_alive()) - .unwrap_or(false); - if own_alive { - wedged_rounds += 1; - if wedged_rounds >= 3 { - tracing::warn!( - "node socket dead across {wedged_rounds} grace windows with our serve alive — restarting it deliberately" - ); - app.state::().node_child.lock().take(); - wedged_rounds = 0; - match ensure_node_running().await { - Ok(Some(child)) => { - app.state::().node_child.lock().replace(child); - } - Ok(None) => {} - Err(e) => tracing::warn!("couldn't bring the node back up: {e:#}"), + + let (ownership, generation) = app.state::().node_child.lock().observation(); + let evidence = recovery.observe(RecoveryObservation { + socket, + ownership, + generation, + }); + match evidence.decision { + RecoveryDecision::Healthy => { + tracing::debug!( + target: "allmystuff::backend_recovery", + socket = evidence.observation.socket.label(), + ownership = evidence.observation.ownership.label(), + generation = evidence.observation.generation, + prior_wedged_rounds = evidence.prior_wedged_rounds, + wedged_rounds = evidence.wedged_rounds, + decision = evidence.decision.label(), + "local node recovery decision" + ); + } + RecoveryDecision::StartupCompleted => { + tracing::info!( + target: "allmystuff::backend_recovery", + socket = evidence.observation.socket.label(), + ownership = evidence.observation.ownership.label(), + generation = evidence.observation.generation, + decision = evidence.decision.label(), + "local node became ready during the existing startup grace window" + ); + } + RecoveryDecision::WaitForOwnedNode => { + tracing::warn!( + target: "allmystuff::backend_recovery", + socket = evidence.observation.socket.label(), + ownership = evidence.observation.ownership.label(), + generation = evidence.observation.generation, + wedged_rounds = evidence.wedged_rounds, + restart_after = WEDGED_RESTART_ROUNDS, + decision = evidence.decision.label(), + "local node socket is unresponsive while the GUI-owned process remains alive" + ); + } + RecoveryDecision::RestartOwnedNode => { + tracing::warn!( + target: "allmystuff::backend_recovery", + socket = evidence.observation.socket.label(), + ownership = evidence.observation.ownership.label(), + generation = evidence.observation.generation, + restart_after = WEDGED_RESTART_ROUNDS, + decision = evidence.decision.label(), + "restarting the unresponsive GUI-owned node" + ); + app.state::().node_child.lock().take(); + match ensure_node_running().await { + Ok(Some(child)) => { + let next_generation = + app.state::().node_child.lock().install(child); + tracing::info!( + target: "allmystuff::backend_recovery", + generation = next_generation, + "installed replacement GUI-owned node" + ); } - } else { - tracing::warn!( - "node socket unresponsive but our serve is still running — not respawning over it ({wedged_rounds}/3)" - ); + Ok(None) => {} + Err(e) => tracing::warn!("couldn't bring the node back up: {e:#}"), } - } else { - wedged_rounds = 0; - tracing::info!("node is gone — bringing it back up"); + } + RecoveryDecision::EnsureNode => { + tracing::info!( + target: "allmystuff::backend_recovery", + socket = evidence.observation.socket.label(), + ownership = evidence.observation.ownership.label(), + generation = evidence.observation.generation, + decision = evidence.decision.label(), + "local node is unavailable; ensuring one is running" + ); match ensure_node_running().await { Ok(Some(child)) => { - app.state::().node_child.lock().replace(child); + let next_generation = + app.state::().node_child.lock().install(child); + tracing::info!( + target: "allmystuff::backend_recovery", + generation = next_generation, + "installed GUI-owned node" + ); } Ok(None) => {} Err(e) => tracing::warn!("couldn't bring the node back up: {e:#}"), } } - } else { - wedged_rounds = 0; } let (tx, mut rx) = mpsc::channel::(256); if let Err(e) = node.subscribe_events(tx).await { @@ -2613,7 +2705,7 @@ fn main() { }; app.manage(AppState { node: node.clone(), - node_child: Mutex::new(None), + node_child: Mutex::new(OwnedNode::default()), }); tauri::async_runtime::spawn(async move { // One node per machine: reuse the Always-On service's node if @@ -2623,7 +2715,13 @@ fn main() { match ensure_node_running().await { Ok(child) => { if let Some(c) = child { - handle.state::().node_child.lock().replace(c); + let generation = + handle.state::().node_child.lock().install(c); + tracing::info!( + target: "allmystuff::backend_recovery", + generation, + "installed initial GUI-owned node" + ); } } Err(e) => tracing::error!("couldn't bring up the allmystuff node: {e:#}"), diff --git a/gui/src/store.svelte.ts b/gui/src/store.svelte.ts index 5770c43..eb64d8b 100644 --- a/gui/src/store.svelte.ts +++ b/gui/src/store.svelte.ts @@ -52,6 +52,7 @@ import { connectRoute, labsSet, tuneRoute, + type NativeDecoderPreference, type StreamTune, type VideoLocalEvent, consoleWindowTarget, @@ -3573,7 +3574,10 @@ class AppStore { * *first* (awaited, so the teardown precedes the popout's fresh offer * on the wire — the same ordering the console's tab switches keep, so * the popout takes the H.264 lane over instead of racing it). */ - async popOutConsoleInput(capId: string) { + async popOutConsoleInput( + capId: string, + decoder: NativeDecoderPreference = "automatic", + ) { if (!isTauri() || isMobile()) return; const cap = this.capability(capId); if (!cap) return; @@ -3586,7 +3590,11 @@ class AppStore { if (owned) await this.disconnect(owned); } const machine = this.machineByAnyId(cap.node); - void openVideoWindow(key, `${cap.label} · ${machine?.label ?? "AllMyStuff"}`); + void openVideoWindow( + key, + `${cap.label} · ${machine?.label ?? "AllMyStuff"}`, + decoder, + ); } /** Lift a room share's tile out into its own OS window. The popout only diff --git a/gui/src/tauri.ts b/gui/src/tauri.ts index 0c7065d..94780ab 100644 --- a/gui/src/tauri.ts +++ b/gui/src/tauri.ts @@ -578,10 +578,12 @@ function parseVideoPacket( * packets — for webviews without WebCodecs, and the bottom rung of the * console's decode ladder. Returns an unwatch fn (a no-op in web mode, * where no frames can arrive anyway). */ +export type NativeDecoderPreference = "automatic" | "software"; + export async function watchVideo( routeId: string, cb: (f: VideoFrameMsg) => void, - opts?: { decode?: boolean }, + opts?: { decode?: boolean; decoder?: NativeDecoderPreference }, ): Promise<() => void> { if (!isTauri()) return () => {}; const { invoke } = await import("@tauri-apps/api/core"); @@ -589,6 +591,7 @@ export async function watchVideo( const token = (await invoke("video_watch", { routeId, decode: opts?.decode ?? false, + decoder: opts?.decoder ?? "automatic", })) as number; let stopped = false; let inFlight = false; @@ -1697,9 +1700,10 @@ export async function isWindowFullscreen(): Promise { export async function openVideoWindow( key: string, title: string, + decoder: NativeDecoderPreference = "automatic", ): Promise { if (!isTauri()) return; - await tryInvoke("open_video_window", { key, title }); + await tryInvoke("open_video_window", { key, title, decoder }); } /** Which stream this window is a popout for, when it was opened by @@ -1709,6 +1713,15 @@ export function videoWindowTarget(): string | null { return new URLSearchParams(window.location.search).get("video"); } +/** Native decoder choice carried into a dedicated video popout. This query + * belongs to the local window only and never enters route negotiation. */ +export function videoWindowDecoderPreference(): NativeDecoderPreference { + if (typeof window === "undefined") return "automatic"; + return new URLSearchParams(window.location.search).get("decoder") === "software" + ? "software" + : "automatic"; +} + /** Pop the CEC Support console out into its own window. One fixed window — * re-invoking focuses it rather than stacking a second console. */ export async function openCecWindow(): Promise { diff --git a/gui/src/ui/Console.svelte b/gui/src/ui/Console.svelte index 95d5a9a..fed6510 100644 --- a/gui/src/ui/Console.svelte +++ b/gui/src/ui/Console.svelte @@ -40,6 +40,7 @@ toggleWindowFullscreen, watchVideo, watchVideoStatus, + type NativeDecoderPreference, type VideoHostStatus, } from "../tauri"; import { @@ -58,6 +59,7 @@ let { windowed = false }: { windowed?: boolean } = $props(); const node = $derived(app.consoleNode); + const kvmSource = $derived(app.isKvm(node ?? undefined)); // What this machine actually shared with us — the console activates with // whatever subset is available and hides the toggles for the rest (a // screen-only share shows the screen, no inert Audio/Control buttons). @@ -206,6 +208,10 @@ // repeatedly. Sticky for the session — a webview whose decoder wedged // once isn't owed a third chance. let nativeDecode = $state(typeof VideoDecoder === "undefined"); + // The decode location and the selected native rung are separate. Automatic + // native fallback may still use a GPU decoder; the visible software choice + // must force OpenH264 on this viewer only. + let nativeDecoderPreference = $state("automatic"); // ---- the quality choices -------------------------------------------- // @@ -332,6 +338,7 @@ openSub = null; // Where to decode is this window's choice; which transport to offer // is the store's (it re-offers the route when that part changes). + nativeDecoderPreference = v === "native" ? "software" : "automatic"; nativeDecode = v === "native" || (v === "auto" && typeof VideoDecoder === "undefined"); app.setConsoleCodec(v === "mjpeg" ? "mjpeg" : v === "auto" ? "auto" : "h264"); } @@ -603,10 +610,17 @@ try { void VideoDecoder.isConfigSupported({ codec: "avc1.42E01F" }) .then((s) => { - if (!s.supported) nativeDecode = true; + if (!s.supported) { + nativeDecoderPreference = "automatic"; + nativeDecode = true; + } }) - .catch(() => (nativeDecode = true)); + .catch(() => { + nativeDecoderPreference = "automatic"; + nativeDecode = true; + }); } catch { + nativeDecoderPreference = "automatic"; nativeDecode = true; } } @@ -701,6 +715,7 @@ // Reading this here makes the ladder's last rung re-run the effect: // flipping it tears the watch down and re-watches in native mode. const native = nativeDecode; + const nativeDecoder = nativeDecoderPreference; // A route CHANGE (screen tab, camera tab, or the teardown between // them) with a picture on the glass = a switch: hold the old frame // dimmed instead of flashing the placeholder. A decode-mode re-wire @@ -788,10 +803,14 @@ let pendingFrame: VideoFrame | null = null; let paintScheduled = false; queuePeek = () => decoder?.decodeQueueSize ?? 0; - decodeModeNote = native ? "native" : ""; + decodeModeNote = native ? `native ${nativeDecoder === "software" ? "sw" : "auto"}` : ""; // "webview (hw)" only asserts we *asked* for hardware (prefer-hardware); // whether WKWebView honours it shows in whether decFps reaches 60. - decodePath = native ? "native (sw)" : "webview (hw)"; + decodePath = native + ? nativeDecoder === "software" + ? "native (sw)" + : "native (auto)" + : "webview (hw)"; const rebuildDecoder = () => { if (decodeMode !== "prefer-software") { @@ -803,6 +822,7 @@ // openh264 decoder. Setting the flag re-runs this effect, which // re-watches the route in native mode (and tears this rung down). console.warn(`video decoder (${codecString}) stalled twice — switching to native decode`); + nativeDecoderPreference = "software"; nativeDecode = true; decodePath = "native (sw)"; askRefresh(); @@ -1033,7 +1053,7 @@ // rebuild as mid-stream stalls — without waiting for the 1s sweep. if (decodeOutputs === 0 && decodeCalls >= 20) rebuildDecoder(); }, - { decode: native }, + { decode: native, decoder: nativeDecoder }, ).then((u) => { // The route may have changed while the subscribe was in flight. if (cancelled) u(); @@ -1509,7 +1529,7 @@ pointerLocked = document.pointerLockElement === stageEl; } function maybePointerLock() { - if (theater && stagePointerActive && !pointerLocked) { + if (theater && stagePointerActive && !kvmSource && !pointerLocked) { void stageEl?.requestPointerLock(); } } @@ -1519,7 +1539,7 @@ }); $effect(() => { // Falling out of theater or control releases the capture. - if (pointerLocked && (!theater || !stagePointerActive)) { + if (pointerLocked && (!theater || !stagePointerActive || kvmSource)) { document.exitPointerLock(); } }); @@ -1535,7 +1555,7 @@ touchMouse.move(e); return; } - if (pointerLocked && stagePointerActive) { + if (pointerLocked && stagePointerActive && !kvmSource) { // Raw deltas, no throttle — the aiming path. if (e.movementX !== 0 || e.movementY !== 0) { app.sendConsoleInput({ kind: "mouse_move_rel", dx: e.movementX, dy: e.movementY }); @@ -1628,7 +1648,7 @@ // mouse; while captured, buttons forward raw (no position re-seat — // the relative stream owns the cursor). if (down) maybePointerLock(); - if (pointerLocked) { + if (pointerLocked && !kvmSource) { e.preventDefault(); app.sendConsoleInput({ kind: "mouse_button", button: e.button, down }); if (down) heldButtons.add(e.button); @@ -2063,7 +2083,9 @@ class="kbtn" title="Pop {selected.label} out into its own window" aria-label="Pop {selected.label} out into its own window" - onclick={() => selectedId && void app.popOutConsoleInput(selectedId)}>⧉ + selectedId && + void app.popOutConsoleInput(selectedId, nativeDecoderPreference)}>⧉ {/if} {/if} @@ -2260,7 +2282,7 @@ aria-label="Pop {inp.label} out into its own window" onclick={(e) => { e.stopPropagation(); - void app.popOutConsoleInput(inp.id); + void app.popOutConsoleInput(inp.id, nativeDecoderPreference); }}>⧉ {/if} diff --git a/gui/src/ui/VideoPopout.svelte b/gui/src/ui/VideoPopout.svelte index 0d53667..8d21884 100644 --- a/gui/src/ui/VideoPopout.svelte +++ b/gui/src/ui/VideoPopout.svelte @@ -31,6 +31,7 @@ sendInput, toggleWindowFullscreen, tuneRoute, + videoWindowDecoderPreference, watchVideo, watchVideoStatus, type StreamTune, @@ -47,6 +48,11 @@ const route = app.catalog.routes.find((r) => r.id === app.videoPopoutLive); return (route && app.capabilityForDisplay(route.from)) ?? null; }); + const sourceMachine = $derived( + sourceCap ? app.machineByAnyId(sourceCap.node) : undefined, + ); + const kvmSource = $derived(app.isKvm(sourceMachine)); + const decoderPreference = videoWindowDecoderPreference(); /** Whether this window owns its route (a console input it wired) — * what gates the quality pills. A watched room share is the sender's * stream to shape. */ @@ -373,7 +379,7 @@ // h264 never arrives here — decode: true means the backend hands // this window ready-to-paint frames. }, - { decode: true }, + { decode: true, decoder: decoderPreference }, ).then((u) => { if (cancelled) u(); else unwatch = u; @@ -438,7 +444,7 @@ pointerLocked = document.pointerLockElement === stageEl; } function maybePointerLock() { - if (fullscreen && controlActive && !pointerLocked) { + if (fullscreen && controlActive && !kvmSource && !pointerLocked) { void stageEl?.requestPointerLock(); } } @@ -448,7 +454,7 @@ }); $effect(() => { // Falling out of fullscreen or control releases the capture. - if (pointerLocked && (!fullscreen || !controlActive)) { + if (pointerLocked && (!fullscreen || !controlActive || kvmSource)) { document.exitPointerLock(); } }); @@ -456,7 +462,7 @@ let lastMoveAt = 0; function onPointerMove(e: PointerEvent) { if (!controlActive) return; - if (pointerLocked) { + if (pointerLocked && !kvmSource) { // Raw deltas, uncoalesced beyond the browser's own batching — this // is the aim path; the 16 ms absolute-move throttle doesn't apply. if (e.movementX !== 0 || e.movementY !== 0) { @@ -478,7 +484,7 @@ // forward even if the cursor was last over a hover-bar button. if (down) stageEl?.focus({ preventScroll: true }); if (down) maybePointerLock(); - if (pointerLocked) { + if (pointerLocked && !kvmSource) { e.preventDefault(); send({ kind: "mouse_button", button: e.button, down }); return; diff --git a/node/Cargo.lock b/node/Cargo.lock index ff5211e..0cfef57 100644 --- a/node/Cargo.lock +++ b/node/Cargo.lock @@ -19,7 +19,7 @@ dependencies = [ [[package]] name = "allmystuff-bridge" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "allmystuff-inventory", @@ -28,7 +28,7 @@ dependencies = [ [[package]] name = "allmystuff-cec-consent" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-cec-protocol", "serde", @@ -38,7 +38,7 @@ dependencies = [ [[package]] name = "allmystuff-cec-protocol" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -47,7 +47,7 @@ dependencies = [ [[package]] name = "allmystuff-graph" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -56,7 +56,7 @@ dependencies = [ [[package]] name = "allmystuff-inventory" -version = "0.2.46" +version = "0.2.48" dependencies = [ "serde", "serde_json", @@ -116,7 +116,7 @@ version = "0.1.0" [[package]] name = "allmystuff-protocol" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "serde", @@ -125,7 +125,7 @@ dependencies = [ [[package]] name = "allmystuff-session" -version = "0.2.46" +version = "0.2.48" dependencies = [ "allmystuff-graph", "allmystuff-protocol", @@ -136,7 +136,7 @@ dependencies = [ [[package]] name = "allmystuff-updater" -version = "0.2.46" +version = "0.2.48" dependencies = [ "dirs", "flate2", diff --git a/node/src/mesh.rs b/node/src/mesh.rs index ee59c9b..739f9fc 100644 --- a/node/src/mesh.rs +++ b/node/src/mesh.rs @@ -50,7 +50,7 @@ use crate::shares::Shares; use crate::sites::{ClientMapping, SitesProxy}; use crate::terminal::{OutMsg, TerminalHost}; use crate::video::{VideoBridge, VideoMode, VideoPacket, VideoSource}; -use crate::video_decode::{Au, DecodeBridge}; +use crate::video_decode::{Au, DecodeBridge, DecoderPreference}; use std::time::{Duration, Instant}; pub struct Mesh { @@ -376,6 +376,9 @@ struct VideoWatcher { /// Whether this window asked the backend to decode H.264 for it /// (raw RGBA frames out) instead of passing access units through. decode: bool, + /// Which native H.264 rung this local window selected. This never leaves + /// the GUI-to-node process boundary. + decoder: DecoderPreference, queue: std::collections::VecDeque>, /// Updated by the window's 16 ms safety poll even when no frame arrived. /// A post-disconnect-request poll is stronger liveness evidence than mere @@ -3836,11 +3839,12 @@ impl Mesh { // Time the pacer's chunk trains as they land — the bandwidth // estimate + delay trend the feedback loop reports back (M3/T1.1). self.note_video_arrival(&route_id, rtp_timestamp, data.len()); - let wants_decode = self + let (wants_decode, decoder_preference) = self .video_watchers .lock() .get(&route_id) - .is_some_and(|w| w.decode); + .map(|w| (w.decode, w.decoder)) + .unwrap_or((false, DecoderPreference::Automatic)); if first { tracing::info!( "first H.264 sample for {route_id} from {} ({} bytes, key={key}, native decode={wants_decode})", @@ -3857,6 +3861,7 @@ impl Mesh { let glitch_rid = route_id.clone(); self.video_decode.feed( &route_id, + decoder_preference, Au { ts_us, key, data }, move |packet| { if let Some(mesh) = mesh.upgrade() { @@ -4139,7 +4144,7 @@ impl Mesh { /// units — for webviews without WebCodecs, and the last rung of the /// console's decode ladder. Returns the claim token to pass back to /// [`Self::video_unwatch`]. - pub fn video_watch(&self, route_id: String, decode: bool) -> u64 { + pub fn video_watch(&self, route_id: String, decode: bool, decoder: DecoderPreference) -> u64 { static NEXT: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(1); let token = NEXT.fetch_add(1, Ordering::Relaxed); if !decode { @@ -4150,7 +4155,9 @@ impl Mesh { // One line per watch claim, so a viewer-side log shows which // window holds each stream and on which decode path — the missing // half of "frames flowing but no window watching". - tracing::info!("window watching {route_id} (native decode: {decode})"); + tracing::info!( + "window watching {route_id} (native decode: {decode}, decoder: {decoder:?})" + ); // A fresh watch (a re-open, an input switch) must start its peer's // video dead-lane grace over. When the previous session closed, its // orphaned frames drained with no route here and left a stale @@ -4184,6 +4191,7 @@ impl Mesh { VideoWatcher { token, decode, + decoder, queue: std::collections::VecDeque::new(), last_poll: Instant::now(), }, diff --git a/node/src/node_control.rs b/node/src/node_control.rs index 7f1cc92..39ea106 100644 --- a/node/src/node_control.rs +++ b/node/src/node_control.rs @@ -51,6 +51,7 @@ use allmystuff_session::{FileEvent, InputAction, TermEvent}; use crate::control_client::{ControlClient, Request}; use crate::mesh::Mesh; use crate::networks_store::DisabledNetworks; +use crate::video_decode::DecoderPreference; use crate::UiSink; // --------------------------------------------------------------------------- @@ -956,7 +957,13 @@ pub async fn dispatch( "video_watch" => { let route_id: String = try_arg!(arg(a, "route_id")); let decode: Option = try_arg!(opt(a, "decode")); - DispatchOut::Json(json!(mesh.video_watch(route_id, decode.unwrap_or(false)))) + let decoder: Option = try_arg!(opt(a, "decoder")); + let decoder = try_arg!(DecoderPreference::parse(decoder.as_deref())); + DispatchOut::Json(json!(mesh.video_watch( + route_id, + decode.unwrap_or(false), + decoder, + ))) } "video_poll" => { let route_id: String = try_arg!(arg(a, "route_id")); diff --git a/node/src/video_decode.rs b/node/src/video_decode.rs index a91d17d..02391bf 100644 --- a/node/src/video_decode.rs +++ b/node/src/video_decode.rs @@ -192,8 +192,35 @@ const STATS_EVERY: Duration = Duration::from_secs(5); /// normal one-frame parser delay from a wedged/unsupported decoder. const ZERO_OUTPUT_AU_LIMIT: u32 = 30; +/// Viewer-local choice for the native H.264 decoder. This is carried only over +/// the GUI-to-node control socket; it is not part of route negotiation or any +/// peer-visible control message. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum DecoderPreference { + #[default] + Automatic, + Software, +} + +impl DecoderPreference { + pub fn parse(value: Option<&str>) -> Result { + match value.unwrap_or("automatic") { + "automatic" => Ok(Self::Automatic), + "software" => Ok(Self::Software), + other => Err(format!( + "unsupported local decoder preference {other:?} (automatic | software)" + )), + } + } + + fn requires_software(self) -> bool { + matches!(self, Self::Software) + } +} + struct RouteDecode { tx: mpsc::SyncSender, + preference: DecoderPreference, /// Set on queue overflow; the thread dumps to the next key unit. need_key: Arc, stop: Arc, @@ -225,8 +252,14 @@ impl DecodeBridge { /// its place (corrupt unit, dumped backlog) so the caller can ask the /// sender to re-key. Both are only captured when this call starts the /// thread. A full queue dumps wholesale and re-keys — see module docs. - pub fn feed(&self, route_id: &str, au: Au, on_frame: F, on_glitch: G) - where + pub fn feed( + &self, + route_id: &str, + preference: DecoderPreference, + au: Au, + on_frame: F, + on_glitch: G, + ) where F: Fn(Vec) + Send + 'static, G: Fn(Option) + Send + 'static, { @@ -234,27 +267,37 @@ impl DecodeBridge { let mut au = au; let mut restarting = false; if let Some(entry) = routes.get_mut(route_id) { - match entry.tx.try_send(au) { - Ok(()) => return, - Err(mpsc::TrySendError::Full(_)) => { - // Deltas past a full queue are useless without their - // predecessors. The decoder thread drops its whole stale - // backlog and requests a fresh key unit. - entry.need_key.store(true, Ordering::SeqCst); - return; - } - Err(mpsc::TrySendError::Disconnected(returned)) => { - // A panicked/returned decoder used to leave a permanent - // tombstone in `routes`: every later feed failed against - // the dead receiver and the display could never restart. - au = returned; - restarting = true; + if entry.preference != preference { + // Decoder selection belongs to this local viewer. Replacing a + // watch with a different preference must replace its worker + // too, otherwise a visible "software" selection can keep + // feeding the already-running hardware decoder indefinitely. + restarting = true; + } else { + match entry.tx.try_send(au) { + Ok(()) => return, + Err(mpsc::TrySendError::Full(_)) => { + // Deltas past a full queue are useless without their + // predecessors. The decoder thread drops its whole stale + // backlog and requests a fresh key unit. + entry.need_key.store(true, Ordering::SeqCst); + return; + } + Err(mpsc::TrySendError::Disconnected(returned)) => { + // A panicked/returned decoder used to leave a permanent + // tombstone in `routes`: every later feed failed against + // the dead receiver and the display could never restart. + au = returned; + restarting = true; + } } } } if restarting { drop(routes.remove(route_id)); - tracing::warn!("native video decoder for {route_id} exited; restarting"); + tracing::warn!( + "native video decoder for {route_id} exited or changed preference; restarting" + ); } // Every fresh decoder starts in `waiting_key`, including the ordinary @@ -277,13 +320,14 @@ impl DecodeBridge { if request_key { on_glitch(None); } - run_decode(&st, &nk, &id, rx, on_frame, on_glitch); + run_decode(&st, &nk, &id, preference, rx, on_frame, on_glitch); }); - tracing::info!("native video decoder started for {route_id}"); + tracing::info!("native video decoder started for {route_id} ({preference:?})"); routes.insert( route_id.to_string(), RouteDecode { tx, + preference, need_key, stop, thread: Some(thread), @@ -350,6 +394,11 @@ impl H264RuntimePolicy { self.delta_failures = Self::DEMOTE_AFTER; } + #[cfg(all(windows, feature = "host"))] + fn note_zero_output_failure(&mut self) { + self.demote(); + } + fn requires_software(&self) -> bool { self.delta_failures >= Self::DEMOTE_AFTER } @@ -602,6 +651,7 @@ fn run_decode( stop: &AtomicBool, need_key: &AtomicBool, route_id: &str, + preference: DecoderPreference, rx: mpsc::Receiver, on_frame: F, on_glitch: G, @@ -724,7 +774,12 @@ fn run_decode( if decoder.is_none() { let built = match stream_codec { AuCodec::H264 => { - let opened = if h264_runtime.requires_software() { + let opened = if preference.requires_software() { + tracing::info!( + "H.264 decoder for {route_id}: viewer selected OpenH264 software" + ); + H264Rung::software() + } else if h264_runtime.requires_software() { tracing::warn!( "H.264 decoder for {route_id}: repeated NVDEC delta failures; keeping this route on OpenH264 software" ); @@ -1056,6 +1111,19 @@ fn run_decode( } } if let Some(e) = broke { + #[cfg(all(windows, feature = "host"))] + if zero_output_aus >= ZERO_OUTPUT_AU_LIMIT + && matches!(decoder.as_ref(), Some(Active::H264(H264Rung::Nvdec(_)))) + { + // An NVDEC session can accept every access unit while never + // surfacing a picture. Reopening the same rung only recreates + // the black-screen loop, so automatic mode takes the portable + // software rung on the next clean entry. + h264_runtime.note_zero_output_failure(); + tracing::warn!( + "H.264 NVDEC for {route_id} produced no pictures across {ZERO_OUTPUT_AU_LIMIT} accepted access units; demoting this route to OpenH264" + ); + } // Corrupt bitstream (a lost unit upstream): drop the decoder, // re-enter at the next IDR. Rate-limited — at frame rate this // would otherwise be a log flood. @@ -1159,6 +1227,7 @@ mod tests { "dead-route".into(), RouteDecode { tx: dead_tx, + preference: DecoderPreference::Automatic, need_key: Arc::new(AtomicBool::new(false)), stop: Arc::new(AtomicBool::new(false)), thread: None, @@ -1168,6 +1237,7 @@ mod tests { let (glitch_tx, glitch_rx) = mpsc::channel(); bridge.feed( "dead-route", + DecoderPreference::Automatic, Au { ts_us: 1, key: false, @@ -1200,6 +1270,7 @@ mod tests { let (glitch_tx, glitch_rx) = mpsc::channel(); bridge.feed( "fresh-delta-route", + DecoderPreference::Automatic, Au { ts_us: 1, key: false, @@ -1220,6 +1291,84 @@ mod tests { bridge.stop("fresh-delta-route"); } + #[test] + fn local_decoder_preference_is_strict_and_defaults_to_automatic() { + assert_eq!( + DecoderPreference::parse(None).unwrap(), + DecoderPreference::Automatic + ); + assert_eq!( + DecoderPreference::parse(Some("automatic")).unwrap(), + DecoderPreference::Automatic + ); + assert_eq!( + DecoderPreference::parse(Some("software")).unwrap(), + DecoderPreference::Software + ); + assert!(DecoderPreference::Software.requires_software()); + assert!(DecoderPreference::parse(Some("nvdec")).is_err()); + } + + #[test] + fn changing_local_decoder_preference_replaces_the_route_worker() { + let bridge = DecodeBridge::new(); + let delta = || Au { + ts_us: 1, + key: false, + data: vec![0, 0, 0, 1, 0x41, 0x9a], + }; + + bridge.feed( + "preference-route", + DecoderPreference::Automatic, + delta(), + |_| {}, + |_| {}, + ); + assert_eq!( + bridge + .routes + .lock() + .get("preference-route") + .map(|route| route.preference), + Some(DecoderPreference::Automatic) + ); + + bridge.feed( + "preference-route", + DecoderPreference::Software, + delta(), + |_| {}, + |_| {}, + ); + assert_eq!( + bridge + .routes + .lock() + .get("preference-route") + .map(|route| route.preference), + Some(DecoderPreference::Software) + ); + bridge.stop("preference-route"); + } + + #[cfg(all(windows, feature = "host"))] + #[test] + fn zero_output_failure_demotes_h264_route_until_restart() { + let mut policy = H264RuntimePolicy::default(); + assert!(!policy.requires_software()); + policy.note_zero_output_failure(); + assert!( + policy.requires_software(), + "a decoder that accepts the zero-output limit must not reopen NVDEC" + ); + policy.reset(); + assert!( + !policy.requires_software(), + "a new route generation gets the automatic hardware ladder again" + ); + } + #[cfg(all(windows, feature = "host"))] #[test] fn repeated_nvdec_delta_failure_demotes_until_route_restart() { @@ -1289,6 +1438,7 @@ mod tests { let tx = tx.clone(); bridge.feed( "r1", + DecoderPreference::Automatic, Au { ts_us: 0, key: shade == 40, // first unit out of a fresh encoder is the IDR @@ -1342,6 +1492,7 @@ mod tests { let sink = tx.clone(); bridge.feed( "resize-route", + DecoderPreference::Automatic, Au { ts_us: seq * 20_000, key: true, @@ -1436,6 +1587,7 @@ mod tests { let sink = frame_tx.clone(); bridge.feed( "paced-h264-route", + DecoderPreference::Automatic, Au { ts_us, key: is_decode_entry(&data), @@ -1540,6 +1692,7 @@ mod tests { let sink = got.clone(); bridge.feed( "route-hevc", + DecoderPreference::Automatic, Au { ts_us: i * 16_667, key: false, From c70d1aeb4eb9e687bc1b09b953c690d033773d56 Mon Sep 17 00:00:00 2001 From: nathanfraske Date: Sat, 25 Jul 2026 19:28:31 -0500 Subject: [PATCH 2/2] Block CEC media hosts in Windows Session 0 --- node/Cargo.toml | 3 + node/src/node_control.rs | 130 ++++++++++++++++++++++++++++++++++++++- 2 files changed, 130 insertions(+), 3 deletions(-) diff --git a/node/Cargo.toml b/node/Cargo.toml index 241bfd1..a3bb016 100644 --- a/node/Cargo.toml +++ b/node/Cargo.toml @@ -245,6 +245,9 @@ windows-sys = { version = "0.60", features = [ # QueryPerformanceCounter/Frequency — mapping DXGI's QPC LastPresentTime # onto the monotonic clock for the M1 capture-age span (win_capture.rs). "Win32_System_Performance", + # ProcessIdToSessionId validates that CEC Support only reuses an + # interactive-session media/input node and never starts one in Session 0. + "Win32_System_RemoteDesktop", # JOBOBJECT_EXTENDED_LIMIT_INFORMATION (daemon_spawn's kill-on-close job) # embeds IO_COUNTERS, which windows-sys gates behind Threading. The GUI # workspace got this feature via unification from another dep; the node diff --git a/node/src/node_control.rs b/node/src/node_control.rs index 39ea106..acd7395 100644 --- a/node/src/node_control.rs +++ b/node/src/node_control.rs @@ -1969,7 +1969,7 @@ async fn ensure_node_current(bin: &Path, pin: &str) { /// [`crate::daemon_spawn::ensure_daemon_running`]'s shape; the GUI will call /// this in Phase B. pub async fn ensure_node_running() -> Result> { - ensure_node_running_pinned(None).await + ensure_node_running_impl(None, false).await } /// Like [`ensure_node_running`], but the caller supplies the AllMyStuff version @@ -1978,10 +1978,105 @@ pub async fn ensure_node_running() -> Result> { /// itself first — the same "keep a sidecar you don't own current" move /// AllMyStuff makes for a reused `myownmesh` (see [`crate::daemon_spawn`]). The /// CEC Support app uses this so a separately-installed AllMyStuff node it reuses -/// is brought up to the version CEC needs to work properly. `pin = None` is -/// exactly [`ensure_node_running`]'s behaviour — no version check at all. +/// is brought up to the version CEC needs to work properly. On Windows this +/// CEC-specific entry point also rejects a Session 0 media/input node. A +/// `pin = None` skips only the version check. pub async fn ensure_node_running_pinned(pin: Option<&str>) -> Result> { + ensure_node_running_impl(pin, true).await +} + +#[cfg(any(windows, test))] +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum PinnedNodeAction { + Reuse, + Spawn, +} + +/// Decide whether CEC Support may reuse or create the shared node. +/// +/// CEC's Windows service runs as LocalSystem in Session 0. It may use an +/// already-running interactive node, but it must never create or reuse a media +/// and input host in Session 0. +#[cfg(any(windows, test))] +fn pinned_node_action(caller_session: u32, node_session: Option) -> Result { + match node_session { + Some(0) => bail!( + "the shared AllMyStuff node is running in Windows Session 0; \ + refusing to attach desktop capture or input" + ), + Some(_) => Ok(PinnedNodeAction::Reuse), + None if caller_session == 0 => bail!( + "no interactive AllMyStuff node is running; refusing to launch \ + desktop capture or input from Windows Session 0" + ), + None => Ok(PinnedNodeAction::Spawn), + } +} + +#[cfg(windows)] +fn windows_process_session(pid: u32) -> Result { + let mut session_id = 0u32; + let ok = unsafe { + windows_sys::Win32::System::RemoteDesktop::ProcessIdToSessionId(pid, &mut session_id) + }; + if ok == 0 { + bail!( + "resolve Windows SessionId for PID {pid}: {}", + std::io::Error::last_os_error() + ); + } + Ok(session_id) +} + +#[cfg(windows)] +async fn windows_node_process_session() -> Result<(u32, u32)> { + let client = NodeClient::new()?; + let stream = client + .connect() + .await + .context("connect to inspect the shared node owner")?; + let pid = stream + .peer_creds() + .context("read shared node pipe credentials")? + .pid() + .ok_or_else(|| anyhow!("the shared node pipe did not report an owner PID"))?; + let session_id = windows_process_session(pid)?; + Ok((pid, session_id)) +} + +async fn ensure_node_running_impl( + pin: Option<&str>, + require_interactive_windows_node: bool, +) -> Result> { + #[cfg(windows)] + let caller = if require_interactive_windows_node { + let pid = std::process::id(); + let session_id = windows_process_session(pid)?; + tracing::info!( + caller_pid = pid, + caller_session_id = session_id, + "validating CEC Support node session ownership" + ); + Some((pid, session_id)) + } else { + None + }; + #[cfg(not(windows))] + let _ = require_interactive_windows_node; + if NodeClient::probe().await { + #[cfg(windows)] + if let Some((caller_pid, caller_session_id)) = caller { + let (node_pid, node_session_id) = windows_node_process_session().await?; + pinned_node_action(caller_session_id, Some(node_session_id))?; + tracing::info!( + caller_pid, + caller_session_id, + node_pid, + node_session_id, + "CEC Support is reusing an interactive AllMyStuff node" + ); + } tracing::info!("existing allmystuff node found on the control socket"); // A node we didn't spawn is already serving. If the caller pins a // version and the on-disk Installed binary is behind it, refresh it so @@ -2011,6 +2106,16 @@ pub async fn ensure_node_running_pinned(pin: Option<&str>) -> Result