diff --git a/start.sh b/start.sh index 8bd54bf..f70d591 100755 --- a/start.sh +++ b/start.sh @@ -113,7 +113,10 @@ socat TCP-LISTEN:12345,reuseaddr,fork \ EXEC:"pacat --playback --device=mic_sink --format=s16le --rate=48000 --channels=1 --latency-msec=20" & echo "socat mic listener started on TCP :12345" -# ── Start streaming agent ───────────────────────────────────────────────────── +# ── Start streaming agent (with auto-restart) ───────────────────────────────── +# gst_agent exits with code 1 on ICE failure or other fatal errors. +# Without a restart loop, the container sits dead until manually restarted. +# 3 second delay before each restart to avoid a tight crash loop. while true; do python3 ~/streaming-agent/gst_agent.py echo "[gst_agent] exited with code $?, restarting in 3s..." diff --git a/streaming-agent/gst_agent.py b/streaming-agent/gst_agent.py index 051539a..9209064 100755 --- a/streaming-agent/gst_agent.py +++ b/streaming-agent/gst_agent.py @@ -20,6 +20,7 @@ import asyncio import json import logging import os +import queue as _queue import subprocess import sys import threading @@ -69,15 +70,63 @@ def cf_put(path, body): raise RuntimeError(f"CF API error: {data['errorDescription']}") return data -# ── Input forwarding via xdotool ────────────────────────────────────────────── -def xdotool(*args): +# ── Input forwarding via a serialized xdotool worker ────────────────────────── +# Previously every input event spawned a fresh `xdotool` process via Popen with +# NO ordering guarantee. Under load (rapid mouse-move + key events) a keyup +# could execute before its keydown, leaving a key logically stuck down in X11 — +# that's the "keyboard stops responding until handoff/restart" bug. +# +# Fix: push all input commands onto a single queue processed by ONE worker +# thread. Commands execute strictly in the order they were received, so a +# keydown always runs before its matching keyup. Each command still runs as a +# short-lived xdotool invocation, but serialized, so ordering is guaranteed. +_xdotool_queue = _queue.Queue() + +def _xdotool_worker(): env = {**os.environ, 'DISPLAY': DISPLAY} - subprocess.Popen( - ['xdotool'] + list(args), - env=env, - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) + while True: + args = _xdotool_queue.get() + + # Coalesce a run of consecutive mouse moves into one, but only by + # peeking non-destructively: we drain moves, and if we hit a non-move + # command we execute the accumulated move, then execute that command. + if args and args[0] == 'mousemove_relative': + sum_dx = int(args[-2]) + sum_dy = int(args[-1]) + pending_cmd = None + while True: + try: + nxt = _xdotool_queue.get_nowait() + except _queue.Empty: + break + if nxt and nxt[0] == 'mousemove_relative': + sum_dx += int(nxt[-2]) + sum_dy += int(nxt[-1]) + else: + pending_cmd = nxt + break + # Execute the accumulated movement as a single command + _run_xdotool(env, ['mousemove_relative', '--', str(sum_dx), str(sum_dy)]) + # If we pulled a non-move command out of the queue, execute it too + if pending_cmd is not None: + _run_xdotool(env, [str(a) for a in pending_cmd]) + else: + _run_xdotool(env, [str(a) for a in args]) + +def _run_xdotool(env, argv): + try: + subprocess.run( + ['xdotool'] + argv, + env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + timeout=2, + ) + except Exception as e: + log.error(f'xdotool error: {e}') + +def xdotool(*args): + _xdotool_queue.put(args) # Track held keys so we can release them all on controller change _held_keys = set() @@ -225,8 +274,15 @@ class CSSStreamAgent: GstWebRTC.WebRTCICEConnectionState.COMPLETED): self._ice_ready.set() elif state == GstWebRTC.WebRTCICEConnectionState.FAILED: - log.error('ICE failed') - self.loop.quit() + log.error('ICE failed — exiting process so start.sh can restart it') + # os._exit forces immediate termination of the entire process, + # including the main thread blocked in asyncio. self.loop.quit() + # alone only stops the GLib thread, leaving the asyncio signaling + # loop running forever — which is why it got stuck instead of + # restarting. + os._exit(1) + elif state == GstWebRTC.WebRTCICEConnectionState.DISCONNECTED: + log.warning('ICE disconnected — monitoring for recovery') def set_remote_description(self, sdp_text, sdp_type='answer'): _, sdp = GstSdp.SDPMessage.new_from_text(sdp_text) @@ -261,6 +317,60 @@ class CSSStreamAgent: finally: self.stop() +# ── CSS process watchdog ────────────────────────────────────────────────────── +def css_watchdog(): + # If cstrike_win64.exe crashes (e.g. Wine c0000409 stack overrun), the + # stream keeps running but shows a black screen forever with no recovery. + # This watchdog detects the crash and relaunches CSS directly via revLoader, + # the same way start.sh does on initial boot. gst_agent keeps running and + # will simply start capturing the relaunched game once it renders. + # + # We only start watching once CSS has been seen alive at least once, to + # avoid racing its initial startup. Require several consecutive misses to + # avoid false positives during a normal controller-triggered relaunch + # (the signaling server kills+relaunches CSS on controller change). + home = os.path.expanduser('~') + revloader = os.path.join(home, 'Counter-Strike Source', 'revLoader.exe') + seen_alive = False + missing_count = 0 + while True: + time.sleep(5) + try: + result = subprocess.run( + ['pgrep', '-f', 'cstrike_win64.exe'], + capture_output=True, text=True + ) + alive = result.returncode == 0 + except Exception: + alive = True # on error, assume alive (don't false-trigger) + + if alive: + seen_alive = True + missing_count = 0 + continue + + if not seen_alive: + continue + + missing_count += 1 + log.warning(f'CSS process not found ({missing_count}/4)') + # 4 consecutive misses (~20s) — long enough to not collide with the + # signaling server's own kill+relaunch cycle on controller change. + if missing_count >= 4: + log.error('CSS appears crashed — relaunching revLoader') + try: + env = dict(os.environ, DISPLAY=DISPLAY) + subprocess.Popen(['wine', revloader], env=env, + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL) + log.info('revLoader relaunched by watchdog') + except Exception as e: + log.error(f'watchdog relaunch failed: {e}') + # Reset so we give the relaunched game time to come up before + # counting misses again. + seen_alive = False + missing_count = 0 + # ── Signaling ───────────────────────────────────────────────────────────────── async def run_signaling(session_id, video_track_name, audio_track_name): log.info(f'Connecting to signaling server {SIG_URL}') @@ -290,6 +400,10 @@ def main(): Gst.init(None) log.info(f'Starting — {CAP_W}x{CAP_H}@{CAP_FPS}fps display={DISPLAY}') + # Start the serialized xdotool input worker (guarantees keydown/keyup order) + xw = threading.Thread(target=_xdotool_worker, daemon=True) + xw.start() + agent = CSSStreamAgent() agent.build_pipeline() agent.start() @@ -342,6 +456,10 @@ def main(): log.info(f'Audio track: {audio_track_name}') log.info(f'Streaming — sessionId={session_id}') + # Start CSS crash watchdog (relaunches the game if it dies unexpectedly) + wd = threading.Thread(target=css_watchdog, daemon=True) + wd.start() + asyncio.run(run_signaling(session_id, video_track_name, audio_track_name)) if __name__ == '__main__':