diff --git a/README.md b/README.md index 6c48948..550f027 100644 --- a/README.md +++ b/README.md @@ -254,6 +254,39 @@ rdc completion zsh > ~/.zfunc/_rdc rdc completion fish > ~/.config/fish/completions/rdc.fish ``` +## Diagnosing RPC connection resets + +For an intermittent RPC error, set `RDC_DAEMON_LOG` before starting a new daemon. +The file is appended to and retained after startup or failure. It records request +methods, IDs, handler/send timings, connection errors, the daemon PID, and the +loaded RenderDoc version/module path. It also enables Python's `faulthandler` to +write native-fault tracebacks to that file where supported. For locally launched +daemons, native stderr is retained separately at `.stderr` to +avoid competing file writers on Windows. + +For example, in Windows PowerShell: + +```powershell +$env:RDC_DAEMON_LOG = Join-Path $PWD "rdc-rpc.log" +rdc --session rpc-debug open .\capture.rdc --timeout 120 +rdc --session rpc-debug script .\pick_once.py *> .\rdc-client.txt +rdc --session rpc-debug status *>> .\rdc-client.txt +rdc --session rpc-debug close +Remove-Item Env:RDC_DAEMON_LOG +``` + +Use a separate log file for each daemon, with an existing parent directory. For +Split sessions, configure the variable on the server before launching its daemon; +setting it on a client connected to an existing daemon has no effect. Request +parameters and session tokens are not included in the request-stage logs, but +native stderr and tracebacks may contain application details. + +Run the reproducing script once, then check `status` separately: scripts can have +side effects and are not automatically retried. Include the client output, daemon +log and its `.stderr` companion, exact commands, and rdc-cli version/commit in the +bug report. Diagnostics do not guarantee a traceback for every native termination +and do not establish the cause of a `PickPixel` failure by themselves. + ## Development ```bash diff --git a/src/rdc/daemon_server.py b/src/rdc/daemon_server.py index 3dfdcbd..f3d9b3b 100644 --- a/src/rdc/daemon_server.py +++ b/src/rdc/daemon_server.py @@ -2,8 +2,10 @@ import argparse import atexit +import faulthandler import json import logging +import os import secrets import shutil import socket @@ -170,6 +172,20 @@ def _cleanup_temp(state: DaemonState) -> None: _log = logging.getLogger("rdc.daemon") +def _configure_diagnostics() -> logging.FileHandler | None: + """Enable persistent request and native-fault logs when explicitly requested.""" + path = os.environ.get("RDC_DAEMON_LOG") + if not path: + return None + handler = logging.FileHandler(Path(path).expanduser(), encoding="utf-8") + handler.setFormatter(logging.Formatter("%(asctime)s %(levelname)s pid=%(process)d %(message)s")) + _log.addHandler(handler) + _log.setLevel(logging.INFO) + faulthandler.enable(file=handler.stream, all_threads=True) + _log.info("daemon diagnostics enabled: python=%s platform=%s", sys.version, sys.platform) + return handler + + class AdapterMatch(NamedTuple): """Adapter identity parsed from a D3D12 structured-data chunk.""" @@ -755,7 +771,8 @@ def run_server( # pragma: no cover conn.settimeout(10.0) try: line = _recv_line(conn) - except (ValueError, TimeoutError): + except (ValueError, OSError) as exc: + _log.warning("RPC receive failed from %s: %s", _addr, exc) continue if not line: continue @@ -769,10 +786,20 @@ def run_server( # pragma: no cover } try: conn.sendall((json.dumps(error_resp) + "\n").encode("utf-8")) - except OSError: - pass + except OSError as exc: + _log.warning("RPC send failed: stage=parse error: %s", exc) continue + method = request.get("method", "") + request_id = request.get("id") + started = time.perf_counter() + _log.info("RPC request started: method=%s id=%s", method, request_id) response, running = _process_request(request, state) + _log.info( + "RPC handler completed: method=%s id=%s elapsed_ms=%.1f", + method, + request_id, + (time.perf_counter() - started) * 1000, + ) binary_path = response.get("result", {}).pop("_binary_path", None) try: payload = json.dumps(response) + "\n" @@ -785,17 +812,29 @@ def run_server( # pragma: no cover } payload = json.dumps(err_resp) + "\n" binary_path = None + stage = "response" try: conn.sendall(payload.encode("utf-8")) if binary_path: - try: - with Path(binary_path).open("rb") as bf: - while chunk := bf.read(65536): - conn.sendall(chunk) - except OSError: - _log.warning("failed to send binary payload: %s", binary_path) - except OSError: - pass + stage = "binary payload" + with Path(binary_path).open("rb") as bf: + while chunk := bf.read(65536): + conn.sendall(chunk) + except OSError as exc: + _log.warning( + "RPC send failed: method=%s id=%s stage=%s: %s", + method, + request_id, + stage, + exc, + ) + else: + _log.info( + "RPC response sent: method=%s id=%s elapsed_ms=%.1f", + method, + request_id, + (time.perf_counter() - started) * 1000, + ) last_activity = time.time() # The shutdown RPC runs cleanup via _handle_shutdown before exiting the @@ -818,6 +857,7 @@ def main() -> None: # pragma: no cover parser.add_argument("--remote-url", default=None) parser.add_argument("--gpu", default=None) args = parser.parse_args() + _configure_diagnostics() state = DaemonState( capture=args.capture, current_eid=0, token=args.token, gpu_pref=args.gpu or "" @@ -832,6 +872,14 @@ def main() -> None: # pragma: no cover sys.stderr.write(f"error: {err}\n") sys.exit(1) + if _log.isEnabledFor(logging.INFO): + _log.info( + "replay ready: renderdoc=%s module=%s api=%s remote=%s", + _detect_version(state.rd) if state.rd is not None else None, + getattr(state.rd, "__file__", None), + state.api_name, + state.is_remote, + ) run_server( host=args.host, port=args.port, diff --git a/src/rdc/services/session_service.py b/src/rdc/services/session_service.py index e8fc855..cb0de8f 100644 --- a/src/rdc/services/session_service.py +++ b/src/rdc/services/session_service.py @@ -74,8 +74,16 @@ def start_daemon( # replay the daemon's verbose upload logging would fill the OS pipe buffer # and block before serving its first ping if no one drains it, which deadlocks # wait_for_ping. A file has no buffer limit and we read its tail on failure. - stderr_file = tempfile.NamedTemporaryFile( # noqa: SIM115 - mode="w+", prefix="rdc-daemon-", suffix=".stderr", delete=False + diagnostic_path = os.environ.get("RDC_DAEMON_LOG") + # Keep native stderr separate from the diagnostic FileHandler. On Windows, + # subprocesses inherit the handle's position but not the CRT append flag; + # a second writer to the same file could overwrite newer log records. + stderr_file = ( + Path(f"{diagnostic_path}.stderr").expanduser().open("a", encoding="utf-8") + if diagnostic_path + else tempfile.NamedTemporaryFile( # noqa: SIM115 + mode="w+", prefix="rdc-daemon-", suffix=".stderr", delete=False + ) ) proc = subprocess.Popen( cmd, @@ -85,12 +93,13 @@ def start_daemon( **_platform.popen_flags(), ) proc._rdc_stderr_path = stderr_file.name # type: ignore[attr-defined] + proc._rdc_preserve_stderr = bool(diagnostic_path) # type: ignore[attr-defined] stderr_file.close() return proc def _read_daemon_stderr(proc: subprocess.Popen[str]) -> str: - """Return the tail of the daemon's redirected stderr, then remove the file.""" + """Read the stderr tail, removing only automatically-created temporary logs.""" path = getattr(proc, "_rdc_stderr_path", None) if not path: return "" @@ -99,22 +108,22 @@ def _read_daemon_stderr(proc: subprocess.Popen[str]) -> str: except OSError: return "" finally: - try: - Path(path).unlink() - except OSError: - pass + if getattr(proc, "_rdc_preserve_stderr", False) is not True: + try: + Path(path).unlink() + except OSError: + pass return "\n".join(text.splitlines()[-20:]).strip() def _discard_daemon_stderr(proc: subprocess.Popen[str]) -> None: """Best-effort removal of the daemon's stderr temp file after a successful start. - On POSIX the daemon keeps writing to the now-unlinked inode; on Windows the - file is still open so unlink may fail and is silently ignored (the OS reclaims - it on process exit). + Explicit diagnostic logs are retained. Otherwise, on POSIX the daemon keeps + writing to the now-unlinked inode; on Windows an open file may prevent unlink. """ path = getattr(proc, "_rdc_stderr_path", None) - if path: + if path and getattr(proc, "_rdc_preserve_stderr", False) is not True: try: Path(path).unlink() except OSError: diff --git a/tests/unit/test_rpc_reset_diagnostics.py b/tests/unit/test_rpc_reset_diagnostics.py new file mode 100644 index 0000000..a2a2840 --- /dev/null +++ b/tests/unit/test_rpc_reset_diagnostics.py @@ -0,0 +1,306 @@ +"""Connection failures must stay local to one RPC and leave useful diagnostics.""" + +from __future__ import annotations + +import json +import logging +import os +import socket +import sys +from pathlib import Path +from types import SimpleNamespace +from typing import Any, TextIO +from unittest.mock import MagicMock + +import pytest + +from rdc import daemon_server +from rdc.protocol import shutdown_request, status_request +from rdc.services import session_service + +_TOKEN = "private-session-token-do-not-log" + + +def _connection( + request: dict[str, object] | None = None, + *, + recv_error: Exception | None = None, + send_error: Exception | None = None, +) -> MagicMock: + conn = MagicMock(spec=socket.socket) + conn.__enter__.return_value = conn + if recv_error is not None: + conn.recv.side_effect = recv_error + else: + conn.recv.return_value = (json.dumps(request) + "\n").encode() + if send_error is not None: + conn.sendall.side_effect = send_error + return conn + + +def _serve( + monkeypatch: pytest.MonkeyPatch, *connections: MagicMock, temp_dir: Path | None = None +) -> MagicMock: + """Inject only socket I/O; exercise the real dispatcher and status/shutdown.""" + listener = MagicMock(spec=socket.socket) + listener.__enter__.return_value = listener + listener.accept.side_effect = [(conn, ("127.0.0.1", 12345)) for conn in connections] + monkeypatch.setattr(daemon_server.socket, "socket", lambda *_a, **_kw: listener) + state = daemon_server.DaemonState(capture="test.rdc", current_eid=42, token=_TOKEN) + state.temp_dir = temp_dir + daemon_server.run_server("127.0.0.1", 39999, state, idle_timeout_s=0) + return listener + + +def _assert_status_and_shutdown(status: MagicMock, shutdown: MagicMock) -> None: + response = json.loads(status.sendall.call_args.args[0]) + assert response["id"] == 2 + assert response["result"]["current_eid"] == 42 + assert response["result"]["capture"] == "test.rdc" + assert json.loads(shutdown.sendall.call_args.args[0])["result"]["ok"] is True + + +@pytest.mark.parametrize( + "failure", + [ + ConnectionResetError(10054, "peer reset connection"), + TimeoutError("peer did not finish sending"), + ValueError("recv_line: message exceeds max_bytes limit"), + ], + ids=["reset", "timeout", "oversized-request"], +) +def test_receive_failure_does_not_block_later_status( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + failure: Exception, +) -> None: + broken = _connection(recv_error=failure) + status = _connection(status_request(_TOKEN, 2)) + shutdown = _connection(shutdown_request(_TOKEN, 3)) + with caplog.at_level(logging.INFO, logger="rdc.daemon"): + listener = _serve(monkeypatch, broken, status, shutdown) + + assert listener.accept.call_count == 3 + broken.sendall.assert_not_called() + _assert_status_and_shutdown(status, shutdown) + assert any( + record.levelno == logging.WARNING and "RPC receive failed" in record.getMessage() + for record in caplog.records + ) + assert _TOKEN not in caplog.text + + +@pytest.mark.parametrize( + "failure", + [ConnectionResetError(10054, "peer reset connection"), BrokenPipeError("peer closed")], + ids=["reset", "broken-pipe"], +) +def test_send_failure_does_not_block_later_status( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, + failure: Exception, +) -> None: + broken = _connection(status_request(_TOKEN, 1), send_error=failure) + status = _connection(status_request(_TOKEN, 2)) + shutdown = _connection(shutdown_request(_TOKEN, 3)) + with caplog.at_level(logging.INFO, logger="rdc.daemon"): + listener = _serve(monkeypatch, broken, status, shutdown) + + assert listener.accept.call_count == 3 + broken.sendall.assert_called_once() + _assert_status_and_shutdown(status, shutdown) + assert any( + record.levelno == logging.WARNING + and "RPC send failed" in record.getMessage() + and "method=status" in record.getMessage() + and "id=1" in record.getMessage() + and "stage=response" in record.getMessage() + for record in caplog.records + ) + assert not any( + "RPC response sent" in record.getMessage() and "id=1" in record.getMessage() + for record in caplog.records + ) + assert _TOKEN not in caplog.text + + +def test_rpc_diagnostics_record_phases_without_request_params( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + request = status_request(_TOKEN, 2) + request["params"]["private_note"] = "private-request-param-do-not-log" + status = _connection(request) + shutdown = _connection(shutdown_request(_TOKEN, 3)) + with caplog.at_level(logging.INFO, logger="rdc.daemon"): + _serve(monkeypatch, status, shutdown) + + _assert_status_and_shutdown(status, shutdown) + messages = [record.getMessage() for record in caplog.records if record.levelno == logging.INFO] + for phase in ("RPC request started", "RPC handler completed", "RPC response sent"): + assert any( + phase in message and "method=status" in message and "id=2" in message + for message in messages + ) + for phase in ("RPC handler completed", "RPC response sent"): + assert any( + phase in message and "id=2" in message and "elapsed_ms=" in message + for message in messages + ) + assert _TOKEN not in caplog.text + assert "private-request-param-do-not-log" not in caplog.text + + +def test_binary_send_reset_reports_stage_and_allows_later_status( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, tmp_path: Path +) -> None: + data_dir = tmp_path / "exports" + data_dir.mkdir() + data_path = data_dir / "output.bin" + data_path.write_bytes(b"binary payload") + request = {"id": 1, "method": "file_read", "params": {"_token": _TOKEN, "path": str(data_path)}} + broken = _connection(request) + broken.sendall.side_effect = [None, ConnectionResetError(10054, "peer reset connection")] + status = _connection(status_request(_TOKEN, 2)) + shutdown = _connection(shutdown_request(_TOKEN, 3)) + with caplog.at_level(logging.INFO, logger="rdc.daemon"): + _serve(monkeypatch, broken, status, shutdown, temp_dir=data_dir) + + assert broken.sendall.call_count == 2 + _assert_status_and_shutdown(status, shutdown) + assert "RPC send failed: method=file_read id=1 stage=binary payload" in caplog.text + assert not any( + "RPC response sent" in record.getMessage() and "id=1" in record.getMessage() + for record in caplog.records + ) + assert _TOKEN not in caplog.text + + +@pytest.mark.parametrize("env_value", [None, ""], ids=["unset", "empty"]) +def test_diagnostics_are_opt_in(monkeypatch: pytest.MonkeyPatch, env_value: str | None) -> None: + if env_value is None: + monkeypatch.delenv("RDC_DAEMON_LOG", raising=False) + else: + monkeypatch.setenv("RDC_DAEMON_LOG", env_value) + enable = MagicMock() + monkeypatch.setattr(daemon_server.faulthandler, "enable", enable) + logger = logging.Logger("test-diagnostics-disabled") + monkeypatch.setattr(daemon_server, "_log", logger) + + assert daemon_server._configure_diagnostics() is None + assert logger.handlers == [] + enable.assert_not_called() + + +def test_diagnostics_file_records_rpc_without_token( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + log_path = tmp_path / "daemon.log" + monkeypatch.setenv("RDC_DAEMON_LOG", str(log_path)) + enable = MagicMock() + monkeypatch.setattr(daemon_server.faulthandler, "enable", enable) + logger = logging.Logger("test-diagnostics-enabled") + monkeypatch.setattr(daemon_server, "_log", logger) + + handler = daemon_server._configure_diagnostics() + assert isinstance(handler, logging.FileHandler) + try: + enable.assert_called_once_with(file=handler.stream, all_threads=True) + assert logger.getEffectiveLevel() == logging.INFO + status = _connection(status_request(_TOKEN, 2)) + shutdown = _connection(shutdown_request(_TOKEN, 3)) + _serve(monkeypatch, status, shutdown) + handler.flush() + contents = log_path.read_text(encoding="utf-8") + assert f"pid={os.getpid()}" in contents + assert "python=" in contents and sys.platform in contents + assert "RPC request started: method=status id=2" in contents + assert "RPC handler completed: method=status id=2 elapsed_ms=" in contents + assert "RPC response sent: method=status id=2 elapsed_ms=" in contents + assert _TOKEN not in contents + finally: + logger.removeHandler(handler) + handler.close() + + +def test_main_logs_loaded_replay_version_and_module( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + monkeypatch.setattr( + sys, + "argv", + ["daemon_server", "--port", "39999", "--capture", "test.rdc", "--token", _TOKEN], + ) + monkeypatch.setattr(daemon_server._platform, "install_shutdown_signal", lambda: None) + monkeypatch.setattr(daemon_server, "_configure_diagnostics", lambda: None) + serve = MagicMock() + monkeypatch.setattr(daemon_server, "run_server", serve) + + def load_replay(state: daemon_server.DaemonState) -> None: + state.rd = SimpleNamespace( + GetVersionString=lambda: "1.46", __file__="actual-renderdoc-module.pyd" + ) + state.api_name = "Vulkan" + + monkeypatch.setattr(daemon_server, "_load_replay", load_replay) + with caplog.at_level(logging.INFO, logger="rdc.daemon"): + daemon_server.main() + + serve.assert_called_once() + assert ( + "replay ready: renderdoc=(1, 46) module=actual-renderdoc-module.pyd api=Vulkan" + in caplog.text + ) + assert _TOKEN not in caplog.text + + +def test_start_daemon_appends_native_stderr_to_separate_diagnostics_file( + monkeypatch: pytest.MonkeyPatch, tmp_path: Path +) -> None: + log_path = tmp_path / "daemon.log" + log_path.write_text("previous RPC diagnostics\n", encoding="utf-8") + stderr_path = Path(str(log_path) + ".stderr") + stderr_path.write_text("previous native stderr\n", encoding="utf-8") + monkeypatch.setenv("RDC_DAEMON_LOG", str(log_path)) + monkeypatch.setattr(session_service, "_renderdoc_available", lambda: False) + proc = MagicMock() + streams: list[TextIO] = [] + + def popen(_cmd: list[str], **kwargs: Any) -> MagicMock: + stream = kwargs["stderr"] + streams.append(stream) + assert Path(stream.name) == stderr_path + stream.write("native replay stderr\n") + stream.flush() + return proc + + monkeypatch.setattr(session_service.subprocess, "Popen", popen) + assert session_service.start_daemon("test.rdc", 39999, _TOKEN) is proc + assert proc._rdc_stderr_path == str(stderr_path) + assert proc._rdc_preserve_stderr is True + assert streams[0].closed + assert ( + stderr_path.read_text(encoding="utf-8") == "previous native stderr\nnative replay stderr\n" + ) + assert log_path.read_text(encoding="utf-8") == "previous RPC diagnostics\n" + + +@pytest.mark.parametrize( + "cleanup", ["read", "discard"], ids=["failed-startup", "successful-startup"] +) +def test_requested_native_stderr_file_survives_cleanup(tmp_path: Path, cleanup: str) -> None: + log_path = tmp_path / "daemon.log" + log_path.write_text("RPC diagnostics\n", encoding="utf-8") + stderr_path = Path(str(log_path) + ".stderr") + contents = "\n".join(f"diagnostic line {i}" for i in range(25)) + "\n" + stderr_path.write_text(contents, encoding="utf-8") + proc = MagicMock() + proc._rdc_stderr_path = str(stderr_path) + proc._rdc_preserve_stderr = True + + if cleanup == "read": + assert session_service._read_daemon_stderr(proc) == "\n".join(contents.splitlines()[-20:]) + else: + session_service._discard_daemon_stderr(proc) + + assert stderr_path.read_text(encoding="utf-8") == contents + assert log_path.read_text(encoding="utf-8") == "RPC diagnostics\n"