Keep the lifecycle callbacks, and the query string out of the exception log

docs/architecture.md

@@ -59,6 +59,17 @@ to an ordered disconnect executor. Runtime-bus disconnect notifications can
 perform synchronous I/O, so they must not run on Tornado's event-loop thread or
 delay unrelated handshakes.
 
+A disconnect that arrives while admission is still running is deferred until
+the connect lifecycle finishes, so a client can never be published after its
+own cleanup has run. At shutdown, the connect and disconnect executors are
+drained rather than cancelled: dropping either would leave a client marked
+connected on the runtime bus after the server that admitted it is gone.
+
+When no executor is installed -- an embedded or test harness that mounts the
+handler without `HiveMindWebsocketProtocol.run()` -- the callback runs
+synchronously on the caller's thread instead. Transport integrators relying on
+that compatibility path keep the historical behaviour.
+
 ## Authorization
 
 Clients connect with a URL query parameter:

hivemind_websocket_protocol/__init__.py

@@ -1,35 +1,37 @@
 import asyncio
 import binascii
 import copy
 import dataclasses
 import hashlib
 import logging
 import math
 import os
 import os.path
 import random
 import time
 from collections import OrderedDict
 from concurrent.futures import Future, ThreadPoolExecutor
 from os import makedirs
 from os.path import exists, join
 from socket import gethostname
 from threading import Lock, get_ident
 from typing import Any, Optional
 
 import pybase64
 from OpenSSL import crypto
 from hivemind_plugin_manager.protocols import NetworkProtocol
 from ovos_bus_client.session import Session
 from ovos_utils.log import LOG
 from ovos_utils.xdg_utils import xdg_data_home
 from poorman_handshake import HandShake, PasswordHandShake, check_password_strength
+import traceback
+
 from tornado import ioloop
 from tornado import web
 from tornado.iostream import StreamClosedError
 from tornado.websocket import WebSocketClosedError, WebSocketHandler
 
 from hivemind_bus_client.message import HiveMessageType
 try:
     from hivemind_core.config import runtime_password_min_bits
 except ImportError:  # released hivemind-core without the helper
@@ -40,34 +42,34 @@
         ).strip().lower()
         return 0.0 if disabled in ("1", "true", "yes", "on") else 40.0
 
 from hivemind_core.protocol import (
     HiveMindListenerProtocol,
     HiveMindClientConnection,
     HiveMindNodeType
 )
 from hivemind_plugin_manager.protocols import ClientCallbacks
 from hivemind_plugin_manager.database import AbstractRemoteDB, Client
 
 from hivemind_websocket_protocol._client_ip import (
     parse_networks,
     resolve_client_ip,
 )
 from hivemind_websocket_protocol.health import (
     LOCAL_HEALTH_PATH,
     HiveMindWebApplication,
     LocalHealthHandler,
 )
 from hivemind_websocket_protocol._metrics import (
     ADMISSION_QUEUE,
     INBOUND_PROCESSING,
     INBOUND_QUEUE,
     REDIS_COMMAND,
     REDIS_DESERIALIZE,
 )
 from hivemind_websocket_protocol._prometheus import (
     HiveMindMetricsHandler,
     load_metric_collectors,
 )
 
 
 DEFAULT_TRUSTED_HEADERS = "x-hivemind-client-ip,x-forwarded-for,x-real-ip"
@@ -103,8 +105,8 @@
 
 
 def _private_key_fingerprint(
     path: Optional[str],
 ) -> Optional[tuple[str, int, int, int, int]]:
     """Return a cheap rotation-aware fingerprint for a listener private key."""
     if not path:
         return None
@@ -127,7 +129,7 @@
     return handshake
 
 
 def _new_client_handshake(path: Optional[str]) -> HandShake:
     """Create isolated handshake state without reparsing an unchanged RSA key."""
     fingerprint = _private_key_fingerprint(path)
     if fingerprint is None:
@@ -156,7 +158,7 @@
 
 def _new_password_handshake(
     password: str,
     min_bits: Optional[float] = None,
 ) -> PasswordHandShake:
     """Validate the credential away from Tornado's event loop."""
     if min_bits is None:
@@ -193,7 +195,7 @@
 
 
 def _finish_websocket_write(future: Any,
                             completion: Optional[Future] = None) -> None:
     """Consume asynchronous write failures so closed peers stay routine."""
     if completion is None:
         future.exception()
@@ -251,7 +253,7 @@
 def _write_websocket_message(handler: WebSocketHandler,
                              payload: str,
                              is_binary: bool,
                              completion: Optional[Future] = None) -> Future:
     """Write a frame and observe both synchronous and future failures."""
     completion = completion or Future()
     if completion.done():
@@ -313,7 +315,10 @@
         return default
     try:
         parsed = int(value)
-    except (TypeError, ValueError):
+    except (TypeError, ValueError, OverflowError):
+        # OverflowError with the others: int(float("inf")) raises it, and a
+        # non-finite worker count reaching this from configuration aborted
+        # startup instead of falling back to the documented default.
         LOG.warning(f"Ignoring invalid {name}: {value!r}")
         return default
     if parsed < 1:
@@ -345,7 +350,7 @@
         hm_protocol (Optional[HiveMindListenerProtocol]): The protocol instance for handling HiveMind messages.
     """
     config: dict[str, Any] = dataclasses.field(default_factory=dict)
     hm_protocol: Optional[HiveMindListenerProtocol] = None
     callbacks: ClientCallbacks = dataclasses.field(default_factory=ClientCallbacks)
 
     def _websocket_ping_settings(self) -> dict[str, float]:
@@ -393,7 +398,7 @@
     def _metrics_listener_settings(
         self,
         websocket_port: int,
     ) -> Optional[tuple[str, int]]:
         enabled = _boolean(
             self.config.get(
                 "metrics_enabled",
@@ -677,14 +682,17 @@
             HiveMindTornadoWebSocket.slow_admission_log_ms = (
                 DEFAULT_SLOW_ADMISSION_LOG_MS
             )
+            # Admission and inbound work is discardable at teardown: whatever
+            # it would have produced, nobody is left to receive.
             auth_executor.shutdown(wait=True, cancel_futures=True)
             handshake_executor.shutdown(wait=True, cancel_futures=True)
             inbound_executor.shutdown(wait=True, cancel_futures=True)
-            connect_lifecycle_executor.shutdown(
-                wait=True,
-                cancel_futures=True,
-            )
-            disconnect_executor.shutdown(wait=True, cancel_futures=True)
+            # The lifecycle pair is not. A connect callback already queued owns
+            # the presence its disconnect has to clear, and dropping either
+            # leaves a client marked connected on the runtime bus after the
+            # server that admitted it is gone. Let them finish.
+            connect_lifecycle_executor.shutdown(wait=True)
+            disconnect_executor.shutdown(wait=True)
         if startup_error is not None:
             raise startup_error
 
@@ -982,6 +990,23 @@
     def _peer_label(self, peer: str) -> str:
         return f"{peer} ({self.source_ip})" if self.source_ip else peer
 
+    def log_exception(self, typ, value, tb) -> None:
+        """Keep a query string out of the uncaught-exception log too.
+
+        ``_request_summary`` only covers the ordinary request line. Tornado's
+        exception logger prints ``self.request`` directly, and
+        ``HTTPServerRequest.__repr__`` includes the URI -- so a credential
+        passed as a query parameter survived the redaction that the normal path
+        already applied.
+        """
+        if isinstance(value, web.HTTPError):
+            return super().log_exception(typ, value, tb)
+        LOG.error(
+            "Uncaught exception %s\n%s",
+            self._request_summary(),
+            "".join(traceback.format_exception(typ, value, tb)),
+        )
+
     def _request_summary(self) -> str:
         """Keep query-string credentials out of Tornado request logs."""
         return (

tests/test_protocol_unit.py

@@ -1765,3 +1765,62 @@ def _run():
     assert not t.is_alive()
     assert (cert_dir / "gen-me.crt").exists()
     assert (cert_dir / "gen-me.key").exists()
+
+
+def test_a_non_finite_worker_count_falls_back_instead_of_aborting_startup():
+    """`int(float("inf"))` raises OverflowError, not ValueError.
+
+    A non-finite worker count reaching this from configuration used to abort
+    startup rather than fall back to the documented default -- the one outcome
+    the fallback exists to prevent.
+    """
+    from hivemind_websocket_protocol import _positive_int
+
+    for value in (float("inf"), float("-inf"), float("nan")):
+        assert _positive_int(value, 7, "workers") == 7, value
+    # The ordinary invalid inputs keep behaving as they did.
+    assert _positive_int("nonsense", 7, "workers") == 7
+    assert _positive_int(0, 7, "workers") == 7
+    assert _positive_int(None, 7, "workers") == 7
+    assert _positive_int(3, 7, "workers") == 3
+
+
+def test_an_uncaught_exception_log_carries_no_query_string():
+    """Tornado prints `self.request`, whose repr includes the URI.
+
+    `_request_summary` only covers the ordinary request line, so a credential
+    passed as a query parameter survived into the exception log that the normal
+    path already redacted.
+    """
+    import logging
+    from unittest.mock import MagicMock
+
+    from hivemind_websocket_protocol import HiveMindTornadoWebSocket
+
+    handler = HiveMindTornadoWebSocket.__new__(HiveMindTornadoWebSocket)
+    handler.request = MagicMock()
+    handler.request.remote_ip = "203.0.113.7"
+    handler.request.method = "GET"
+    handler.request.uri = "/?authorization=c2VjcmV0OnRva2Vu"
+    handler.request.__repr__ = lambda _self: (
+        "HTTPServerRequest(uri='/?authorization=c2VjcmV0OnRva2Vu')"
+    )
+
+    records = []
+    import hivemind_websocket_protocol as module
+
+    class _Recorder:
+        def error(self, message, *args):
+            records.append(message % args if args else message)
+
+    original = module.LOG
+    module.LOG = _Recorder()
+    try:
+        handler.log_exception(ValueError, ValueError("boom"), None)
+    finally:
+        module.LOG = original
+
+    assert records, "the exception was not logged at all"
+    joined = "\n".join(records)
+    assert "c2VjcmV0OnRva2Vu" not in joined, joined
+    assert "authorization=" not in joined, joined