diff --git a/crates/punktfunk-host/src/inject.rs b/crates/punktfunk-host/src/inject.rs index 07eca85d..1e668f24 100644 --- a/crates/punktfunk-host/src/inject.rs +++ b/crates/punktfunk-host/src/inject.rs @@ -166,132 +166,9 @@ pub fn default_backend() -> Backend { Backend::Unsupported } -/// Host-lifetime pointer/keyboard injector running on its OWN thread, fed over a clonable `Send` -/// channel. The injector backend owns non-`Send` compositor state (a Wayland connection / xkb / EIS -/// socket), so it must live on a single thread; both the GameStream control plane and the native -/// punktfunk/1 plane forward their decoded keyboard/mouse events here instead of injecting inline, so -/// a slow inject (a portal stall, a desktop switch) never head-blocks the network thread's -/// keepalive/retransmit servicing. -pub(crate) struct InjectorService { - tx: std::sync::mpsc::Sender, -} - -impl InjectorService { - pub(crate) fn start() -> InjectorService { - let (tx, rx) = std::sync::mpsc::channel::(); - if let Err(e) = std::thread::Builder::new() - .name("punktfunk-injector".into()) - .spawn(move || injector_service_thread(rx)) - { - tracing::error!(error = %e, "injector service thread spawn failed — pointer/keyboard input disabled"); - } - InjectorService { tx } - } - - /// A sender a session/plane forwards its pointer/keyboard events to. Cloned per caller; dropping a - /// clone does NOT stop the service (it runs while any sender — incl. the service's own — lives). - pub(crate) fn sender(&self) -> std::sync::mpsc::Sender { - self.tx.clone() - } -} - -/// Backoff between reopen attempts after the injector backend fails to open or its worker dies, so a -/// persistently-unavailable portal isn't hammered once per event. -const INJECTOR_REOPEN_BACKOFF: std::time::Duration = std::time::Duration::from_secs(2); - -/// The host-lifetime injector worker: lazily open the pointer/keyboard backend, then inject every -/// forwarded event. Reopen (after [`INJECTOR_REOPEN_BACKOFF`]) on open failure, on a backend change -/// (input follows the active session), or if the backend's worker dies mid-stream. Exits only when -/// every sender has dropped (host shutdown), which drops the injector and closes its portal session. -/// -/// Each wake drains the whole backlog and [`coalesce`]s redundant motion before injecting, so a slow -/// backend never builds up a queue of stale relative-mouse/scroll events (latency) — while button, -/// key, and absolute-move ordering is preserved exactly. -fn injector_service_thread(rx: std::sync::mpsc::Receiver) { - let mut injector: Option> = None; - let mut open_backend: Option = None; - let mut last_failed: Option = None; - while let Ok(first) = rx.recv() { - // Drain everything already queued behind `first` so we coalesce a whole burst at once. - let mut batch = vec![first]; - while let Ok(ev) = rx.try_recv() { - batch.push(ev); - } - - // The resolved input backend (PUNKTFUNK_INPUT_BACKEND, set per connect / mid-stream session - // switch) may have changed since we opened. Reopen against it so input FOLLOWS the active - // session instead of injecting into a stale, still-warm backend (e.g. the managed gamescope's - // EIS socket after the user switched to the KDE desktop). - let want = default_backend(); - if injector.is_some() && open_backend != Some(want) { - tracing::info!( - ?open_backend, - ?want, - "input: backend changed — reopening injector for the active session" - ); - injector = None; - last_failed = None; // re-resolve immediately - } - if injector.is_none() { - // Open on the first event; after a failure wait out the backoff before retrying (a few - // events drop during setup — acceptable, input is lossy). - let ready = last_failed.is_none_or(|t| t.elapsed() >= INJECTOR_REOPEN_BACKOFF); - if ready { - match open(want) { - Ok(i) => { - tracing::info!(backend = ?want, "input injector ready (host-lifetime)"); - injector = Some(i); - open_backend = Some(want); - last_failed = None; - } - Err(e) => { - tracing::warn!(error = %format!("{e:#}"), "pointer/keyboard injection unavailable — will retry"); - last_failed = Some(std::time::Instant::now()); - } - } - } - } - if let Some(inj) = injector.as_mut() { - for ev in coalesce(batch) { - if let Err(e) = inj.inject(&ev) { - // The backend's worker (portal session / EIS socket) died — drop it and reopen on - // a later event (covers a gamescope EIS socket that respawns with its session). - tracing::warn!(error = %format!("{e:#}"), "inject failed — reopening injector"); - injector = None; - open_backend = None; - last_failed = Some(std::time::Instant::now()); - break; // abandon the rest of this batch; the next one reopens - } - } - } - } - tracing::debug!("injector service stopped (host shutting down)"); -} - -/// Coalesce a drained burst: sum consecutive relative-mouse deltas and consecutive same-axis scroll -/// deltas (identical net effect, far fewer injects), passing buttons, keys, absolute moves, and any -/// type change through untouched and in order. Only *adjacent* same-type events merge, so a button -/// or key between two moves flushes the accumulated motion first — ordering is never reshuffled. -fn coalesce(events: Vec) -> Vec { - let mut out: Vec = Vec::with_capacity(events.len()); - for ev in events { - match out.last_mut() { - Some(last) if last.kind == InputKind::MouseMove && ev.kind == InputKind::MouseMove => { - last.x = last.x.saturating_add(ev.x); - last.y = last.y.saturating_add(ev.y); - } - Some(last) - if last.kind == InputKind::MouseScroll - && ev.kind == InputKind::MouseScroll - && last.code == ev.code => - { - last.x = last.x.saturating_add(ev.x); - } - _ => out.push(ev), - } - } - out -} +#[path = "inject/service.rs"] +mod service; +pub(crate) use service::InjectorService; /// How the libei backend reaches its EIS server. KWin goes through the `RemoteDesktop` *portal* /// (with a pre-seeded grant), but GNOME's portal `Start()` needs an interactive approval a @@ -460,57 +337,3 @@ mod sendinput; #[cfg(target_os = "linux")] #[path = "inject/linux/wlr.rs"] mod wlr; - -#[cfg(test)] -mod tests { - use super::*; - - fn mk(kind: InputKind, code: u32, x: i32, y: i32) -> InputEvent { - InputEvent { - kind, - _pad: [0; 3], - code, - x, - y, - flags: 0, - } - } - - #[test] - fn coalesce_sums_adjacent_motion_and_preserves_order() { - let events = vec![ - mk(InputKind::MouseMove, 0, 1, 2), - mk(InputKind::MouseMove, 0, 3, -1), // → summed with the previous move - mk(InputKind::KeyDown, 30, 0, 0), // flushes the move, passes through verbatim - mk(InputKind::MouseMove, 0, 5, 5), // a NEW run after the key (not merged across it) - mk(InputKind::MouseScroll, 0, 1, 0), - mk(InputKind::MouseScroll, 0, 2, 0), // same axis (code 0) → summed - mk(InputKind::MouseScroll, 1, 1, 0), // different axis (code 1) → separate - ]; - let out = coalesce(events); - assert_eq!(out.len(), 5); - assert_eq!( - (out[0].kind, out[0].x, out[0].y), - (InputKind::MouseMove, 4, 1) - ); - assert_eq!(out[1].kind, InputKind::KeyDown); - assert_eq!( - (out[2].kind, out[2].x, out[2].y), - (InputKind::MouseMove, 5, 5) - ); - assert_eq!( - (out[3].kind, out[3].code, out[3].x), - (InputKind::MouseScroll, 0, 3) - ); - assert_eq!( - (out[4].kind, out[4].code, out[4].x), - (InputKind::MouseScroll, 1, 1) - ); - } - - #[test] - fn coalesce_handles_empty_and_singleton() { - assert!(coalesce(vec![]).is_empty()); - assert_eq!(coalesce(vec![mk(InputKind::MouseMove, 0, 7, 8)]).len(), 1); - } -} diff --git a/crates/punktfunk-host/src/inject/service.rs b/crates/punktfunk-host/src/inject/service.rs new file mode 100644 index 00000000..b313c2ca --- /dev/null +++ b/crates/punktfunk-host/src/inject/service.rs @@ -0,0 +1,189 @@ +//! The off-thread injector service (plan §W4, carved out of the inject facade): a host-lifetime +//! pointer/keyboard injector pinned to its OWN thread and fed over a clonable `Send` channel, plus +//! the pre-injection [`coalesce`] pass. The backend owns non-`Send` compositor state (a Wayland +//! connection / xkb / EIS socket), so it must live on one thread; both the GameStream control plane +//! and the native punktfunk/1 plane forward decoded input here instead of injecting inline. + +use super::*; + +/// Host-lifetime pointer/keyboard injector running on its OWN thread, fed over a clonable `Send` +/// channel. The injector backend owns non-`Send` compositor state (a Wayland connection / xkb / EIS +/// socket), so it must live on a single thread; both the GameStream control plane and the native +/// punktfunk/1 plane forward their decoded keyboard/mouse events here instead of injecting inline, so +/// a slow inject (a portal stall, a desktop switch) never head-blocks the network thread's +/// keepalive/retransmit servicing. +pub(crate) struct InjectorService { + tx: std::sync::mpsc::Sender, +} + +impl InjectorService { + pub(crate) fn start() -> InjectorService { + let (tx, rx) = std::sync::mpsc::channel::(); + if let Err(e) = std::thread::Builder::new() + .name("punktfunk-injector".into()) + .spawn(move || injector_service_thread(rx)) + { + tracing::error!(error = %e, "injector service thread spawn failed — pointer/keyboard input disabled"); + } + InjectorService { tx } + } + + /// A sender a session/plane forwards its pointer/keyboard events to. Cloned per caller; dropping a + /// clone does NOT stop the service (it runs while any sender — incl. the service's own — lives). + pub(crate) fn sender(&self) -> std::sync::mpsc::Sender { + self.tx.clone() + } +} + +/// Backoff between reopen attempts after the injector backend fails to open or its worker dies, so a +/// persistently-unavailable portal isn't hammered once per event. +const INJECTOR_REOPEN_BACKOFF: std::time::Duration = std::time::Duration::from_secs(2); + +/// The host-lifetime injector worker: lazily open the pointer/keyboard backend, then inject every +/// forwarded event. Reopen (after [`INJECTOR_REOPEN_BACKOFF`]) on open failure, on a backend change +/// (input follows the active session), or if the backend's worker dies mid-stream. Exits only when +/// every sender has dropped (host shutdown), which drops the injector and closes its portal session. +/// +/// Each wake drains the whole backlog and [`coalesce`]s redundant motion before injecting, so a slow +/// backend never builds up a queue of stale relative-mouse/scroll events (latency) — while button, +/// key, and absolute-move ordering is preserved exactly. +fn injector_service_thread(rx: std::sync::mpsc::Receiver) { + let mut injector: Option> = None; + let mut open_backend: Option = None; + let mut last_failed: Option = None; + while let Ok(first) = rx.recv() { + // Drain everything already queued behind `first` so we coalesce a whole burst at once. + let mut batch = vec![first]; + while let Ok(ev) = rx.try_recv() { + batch.push(ev); + } + + // The resolved input backend (PUNKTFUNK_INPUT_BACKEND, set per connect / mid-stream session + // switch) may have changed since we opened. Reopen against it so input FOLLOWS the active + // session instead of injecting into a stale, still-warm backend (e.g. the managed gamescope's + // EIS socket after the user switched to the KDE desktop). + let want = default_backend(); + if injector.is_some() && open_backend != Some(want) { + tracing::info!( + ?open_backend, + ?want, + "input: backend changed — reopening injector for the active session" + ); + injector = None; + last_failed = None; // re-resolve immediately + } + if injector.is_none() { + // Open on the first event; after a failure wait out the backoff before retrying (a few + // events drop during setup — acceptable, input is lossy). + let ready = last_failed.is_none_or(|t| t.elapsed() >= INJECTOR_REOPEN_BACKOFF); + if ready { + match open(want) { + Ok(i) => { + tracing::info!(backend = ?want, "input injector ready (host-lifetime)"); + injector = Some(i); + open_backend = Some(want); + last_failed = None; + } + Err(e) => { + tracing::warn!(error = %format!("{e:#}"), "pointer/keyboard injection unavailable — will retry"); + last_failed = Some(std::time::Instant::now()); + } + } + } + } + if let Some(inj) = injector.as_mut() { + for ev in coalesce(batch) { + if let Err(e) = inj.inject(&ev) { + // The backend's worker (portal session / EIS socket) died — drop it and reopen on + // a later event (covers a gamescope EIS socket that respawns with its session). + tracing::warn!(error = %format!("{e:#}"), "inject failed — reopening injector"); + injector = None; + open_backend = None; + last_failed = Some(std::time::Instant::now()); + break; // abandon the rest of this batch; the next one reopens + } + } + } + } + tracing::debug!("injector service stopped (host shutting down)"); +} + +/// Coalesce a drained burst: sum consecutive relative-mouse deltas and consecutive same-axis scroll +/// deltas (identical net effect, far fewer injects), passing buttons, keys, absolute moves, and any +/// type change through untouched and in order. Only *adjacent* same-type events merge, so a button +/// or key between two moves flushes the accumulated motion first — ordering is never reshuffled. +fn coalesce(events: Vec) -> Vec { + let mut out: Vec = Vec::with_capacity(events.len()); + for ev in events { + match out.last_mut() { + Some(last) if last.kind == InputKind::MouseMove && ev.kind == InputKind::MouseMove => { + last.x = last.x.saturating_add(ev.x); + last.y = last.y.saturating_add(ev.y); + } + Some(last) + if last.kind == InputKind::MouseScroll + && ev.kind == InputKind::MouseScroll + && last.code == ev.code => + { + last.x = last.x.saturating_add(ev.x); + } + _ => out.push(ev), + } + } + out +} + +#[cfg(test)] +mod tests { + use super::*; + use punktfunk_core::input::{InputEvent, InputKind}; + + fn mk(kind: InputKind, code: u32, x: i32, y: i32) -> InputEvent { + InputEvent { + kind, + _pad: [0; 3], + code, + x, + y, + flags: 0, + } + } + + #[test] + fn coalesce_sums_adjacent_motion_and_preserves_order() { + let events = vec![ + mk(InputKind::MouseMove, 0, 1, 2), + mk(InputKind::MouseMove, 0, 3, -1), // → summed with the previous move + mk(InputKind::KeyDown, 30, 0, 0), // flushes the move, passes through verbatim + mk(InputKind::MouseMove, 0, 5, 5), // a NEW run after the key (not merged across it) + mk(InputKind::MouseScroll, 0, 1, 0), + mk(InputKind::MouseScroll, 0, 2, 0), // same axis (code 0) → summed + mk(InputKind::MouseScroll, 1, 1, 0), // different axis (code 1) → separate + ]; + let out = coalesce(events); + assert_eq!(out.len(), 5); + assert_eq!( + (out[0].kind, out[0].x, out[0].y), + (InputKind::MouseMove, 4, 1) + ); + assert_eq!(out[1].kind, InputKind::KeyDown); + assert_eq!( + (out[2].kind, out[2].x, out[2].y), + (InputKind::MouseMove, 5, 5) + ); + assert_eq!( + (out[3].kind, out[3].code, out[3].x), + (InputKind::MouseScroll, 0, 3) + ); + assert_eq!( + (out[4].kind, out[4].code, out[4].x), + (InputKind::MouseScroll, 1, 1) + ); + } + + #[test] + fn coalesce_handles_empty_and_singleton() { + assert!(coalesce(vec![]).is_empty()); + assert_eq!(coalesce(vec![mk(InputKind::MouseMove, 0, 7, 8)]).len(), 1); + } +}