diff --git a/pyproject.toml b/pyproject.toml index 17c8a07..67ea91f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -18,4 +18,5 @@ build-backend = "uv_build" [dependency-groups] dev = [ "pytest>=9.1.1", + "rich>=15.0.0", ] diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 3f02431..c6efbda 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -2,6 +2,7 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +import logging import threading from typing import Literal @@ -14,6 +15,10 @@ from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer +log = logging.getLogger(__name__) +log.addHandler(logging.NullHandler()) + + class Core: def __init__(self, config: CoreConfig) -> None: self._state: Literal["uninitd", "initd", "started", "stopping", "stopped"] = ( @@ -22,24 +27,60 @@ class Core: self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() self._core_config: CoreConfig = config - self._json_server = JsonServer(self._core_config) self._connector_registry = ConnectorRegistry() + self._json_server = JsonServer(self._core_config, self._connector_registry) self._state = "initd" def start(self) -> None: if self._state != "initd": + log.warning( + "Core start rejected in state %s", + self._state, + extra={ + "event": "core.start_rejected", + "core_state": self._state, + "expected_state": "initd", + }, + ) raise CoreStateMismatchError( f'Unable to start DynaLab Core, expected state to be "initd", found {self._state}' ) + + log.info( + "Core %s starting on %s", + self._core_version.get_version(), + self._core_config.bind_str(), + extra={ + "event": "core.starting", + "core_version": self._core_version.get_version(), + "host": self._core_config.host, + "port": self._core_config.port, + }, + ) self._json_server.start() self._state = "started" + log.info( + "Core started", + extra={"event": "core.started", "core_state": self._state}, + ) def wait(self, timeout: float | None = None) -> None: - self._stop_event.wait(timeout) + try: + self._stop_event.wait(timeout) + except KeyboardInterrupt: + raise def stop(self) -> None: + log.info( + "Core stopping", + extra={"event": "core.stopping", "core_state": self._state}, + ) self._json_server.stop() self._connector_registry.stop() self._stop_event.set() self._state = "stopped" + log.info( + "Core stopped", + extra={"event": "core.stopped", "core_state": self._state}, + ) diff --git a/src/dynalab_core/constants.py b/src/dynalab_core/constants.py index a43904f..9893b29 100644 --- a/src/dynalab_core/constants.py +++ b/src/dynalab_core/constants.py @@ -2,7 +2,18 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +from uuid import uuid4 from dynalab_core.protocols.common import VersionDescriptor +from dynalab_core.protocols.constants import PROTOCOL_VERSION +from dynalab_core.protocols.packets.handshake import DynaLabHello CORE_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) + +HELLO_PACKET = DynaLabHello( + instance_id=uuid4(), + core_version=CORE_VERSION, + protocol_version=PROTOCOL_VERSION, + heartbeat_interval_ms=1000, + heartbeat_timeout_ms=5000, +) diff --git a/src/dynalab_core/protocols/json/constants.py b/src/dynalab_core/protocols/constants.py similarity index 100% rename from src/dynalab_core/protocols/json/constants.py rename to src/dynalab_core/protocols/constants.py diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 5fec425..39dc6f1 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -2,36 +2,106 @@ # Copyright (C) 2026 Association Exergie # SPDX-License-Identifier: GPL-3.0-or-later +import logging from queue import Empty, Full, Queue -from threading import RLock, Thread import threading +from threading import RLock, Thread +from time import monotonic, sleep from uuid import UUID + +from dynalab_core.constants import HELLO_PACKET from dynalab_core.protocols.errors import ( ConnectorEndpointQueueFullError, ConnectorRegistryAlreadyRegisteredError, ) from dynalab_core.protocols.packets import ProtocolMessage -from dynalab_core.protocols.packets.handshake import ConnectorHello +from dynalab_core.protocols.packets.handshake import ( + ConnectorHello, + HandshakeAccepted, + HandshakeRejected, +) +from dynalab_core.protocols.packets.heartbeat import Heartbeat + +log = logging.getLogger(__name__) class ConnectorEndpoint: - def __init__(self, hello: ConnectorHello) -> None: + def __init__( + self, + hello: ConnectorHello, + timeout_event: threading.Event | None = None, + ) -> None: + # external IO queues self._packet_ingress_queue: Queue[ProtocolMessage] = Queue(524288) self._packet_egress_queue: Queue[ProtocolMessage] = Queue(524288) + # internal IO queues + self._heartbeat_ingress_queue: Queue[ProtocolMessage] = Queue(524288) + self._heartbeat_egress_queue: Queue[ProtocolMessage] = Queue(524288) + self._connector_hello = hello - self._worker_thread = Thread( - target=self._worker, - name=f"endpoint_worker_{self._connector_hello.connector_uuid}", + self._timed_out_event = timeout_event or threading.Event() + self._stop_event = threading.Event() + self._input_worker_stopped_event = threading.Event() + self._input_worker_thread = Thread( + target=self._input_worker, + name=f"endpoint_input_worker_{self._connector_hello.connector_uuid}", daemon=True, ) - self._stop_event = threading.Event() - self._stopped_event = threading.Event() + self._output_worker_stopped_event = threading.Event() + self._output_worker_thread = Thread( + target=self._output_worker, + name=f"endpoint_output_worker_{self._connector_hello.connector_uuid}", + daemon=True, + ) + self._heartbeat_worker_stopped_event = threading.Event() + self._heartbeat_worker_thread = Thread( + target=self._heartbeat_worker, + name=f"endpoint_heartbeat_worker_{self._connector_hello.connector_uuid}", + daemon=True, + ) + self._input_worker_thread.start() + self._output_worker_thread.start() def stop(self) -> None: + connector_uuid = str(self.uuid()) + log.debug( + "Stopping connector endpoint %s", + connector_uuid, + extra={"event": "endpoint.stopping", "connector_uuid": connector_uuid}, + ) self._stop_event.set() - if not self._worker_thread.is_alive(): - self._stopped_event.set() - self._stopped_event.wait(10) + + if not self._input_worker_thread.is_alive(): + self._input_worker_stopped_event.set() + if not self._output_worker_thread.is_alive(): + self._output_worker_stopped_event.set() + if not self._heartbeat_worker_thread.is_alive(): + self._heartbeat_worker_stopped_event.set() + + self._input_worker_stopped_event.wait(1) + self._output_worker_stopped_event.wait(1) + self._heartbeat_worker_stopped_event.wait(1) + + if ( + not self._input_worker_stopped_event.is_set() + or not self._output_worker_stopped_event.is_set() + or not self._heartbeat_worker_stopped_event.is_set() + ): + log.error( + "Connector endpoint %s did not stop within 10 seconds", + connector_uuid, + extra={ + "event": "endpoint.stop_timeout", + "connector_uuid": connector_uuid, + "timeout_s": 10, + }, + ) + else: + log.debug( + "Connector endpoint %s stopped", + connector_uuid, + extra={"event": "endpoint.stopped", "connector_uuid": connector_uuid}, + ) def uuid(self) -> UUID: return self._connector_hello.connector_uuid @@ -40,49 +110,163 @@ class ConnectorEndpoint: try: self._packet_ingress_queue.put_nowait(packet) except Full: + self._log_queue_full("ingress", self._packet_ingress_queue) raise ConnectorEndpointQueueFullError - def get_egress_packet(self, timeout: float | None) -> ProtocolMessage | None: - try: - packet = self._packet_egress_queue.get(block=True, timeout=timeout) - except Empty: - return None - else: - return packet + def get_egress_packet(self, timeout: float | None) -> ProtocolMessage: + return self._packet_egress_queue.get(block=True, timeout=timeout) - def get_egress_packet_no_wait(self) -> ProtocolMessage | None: - try: - packet = self._packet_egress_queue.get_nowait() - except Empty: - return None - else: - return packet + def get_egress_packet_no_wait(self) -> ProtocolMessage: + return self._packet_egress_queue.get_nowait() - def _get_ingress_packet(self, timeout: float | None) -> ProtocolMessage | None: - try: - packet = self._packet_ingress_queue.get(block=True, timeout=timeout) - except Empty: - return None - else: - return packet + def _get_ingress_packet(self, timeout: float | None) -> ProtocolMessage: + return self._packet_ingress_queue.get(block=True, timeout=timeout) - def _get_ingress_packet_no_wait(self) -> ProtocolMessage | None: - try: - packet = self._packet_ingress_queue.get_nowait() - except Empty: - return None - else: - return packet + def _get_ingress_packet_no_wait(self) -> ProtocolMessage: + return self._packet_ingress_queue.get_nowait() def _put_egress_packet(self, packet: ProtocolMessage) -> None: try: self._packet_egress_queue.put_nowait(packet) except Full: + self._log_queue_full("egress", self._packet_egress_queue) raise ConnectorEndpointQueueFullError - def _worker(self) -> None: - # TODO: build endpoint worker thread - pass + def _log_queue_full(self, direction: str, queue: Queue[ProtocolMessage]) -> None: + connector_uuid = str(self.uuid()) + log.warning( + "Connector %s %s queue is full (%d/%d)", + connector_uuid, + direction, + queue.qsize(), + queue.maxsize, + extra={ + "event": "endpoint.queue_full", + "connector_uuid": connector_uuid, + "queue_direction": direction, + "queue_size": queue.qsize(), + "queue_capacity": queue.maxsize, + }, + ) + + def _heartbeat_worker(self) -> None: + last_send: int = 0 + last_recieved: int = 0 + first_received: bool = False + ctr: int = 0 + try: + while not self._stop_event.is_set(): + now = round(monotonic() * 1000) + if now > last_send + HELLO_PACKET.heartbeat_interval_ms: + try: + self._heartbeat_egress_queue.put_nowait( + Heartbeat(sequence=ctr, send_timestamp=now) + ) + last_send = now + ctr += 1 + log.debug("Sent heartbeat") + except Full: + sleep(0.01) + + try: + received = self._heartbeat_ingress_queue.get_nowait() + except Empty: + sleep(0.01) + else: + if isinstance(received, Heartbeat): + first_received = True + if received.return_timestamp is not None: + last_recieved = received.return_timestamp + + if not first_received: + if ctr > round( + HELLO_PACKET.heartbeat_timeout_ms + / HELLO_PACKET.heartbeat_interval_ms + ): + self._timed_out_event.set() + return + elif last_send > last_recieved + HELLO_PACKET.heartbeat_timeout_ms: + self._timed_out_event.set() + return + finally: + self._heartbeat_worker_stopped_event.set() + + def _input_worker(self) -> None: + connector_uuid = str(self.uuid()) + log.debug( + "Connector endpoint input worker %s started", + self._input_worker_thread.name, + extra={ + "event": "endpoint.input_worker_started", + "connector_uuid": connector_uuid, + "thread_name": self._input_worker_thread.name, + }, + ) + while not self._stop_event.is_set(): + try: + message = self._get_ingress_packet_no_wait() + except Empty: + self._stop_event.wait(0.01) + continue + else: + if isinstance(message, Heartbeat): + self._heartbeat_ingress_queue.put_nowait(message) + self._input_worker_stopped_event.set() + log.debug( + "Connector endpoint input worker %s stopped", + self._input_worker_thread.name, + extra={ + "event": "endpoint.input_worker_stopped", + "connector_uuid": connector_uuid, + "thread_name": self._input_worker_thread.name, + }, + ) + + def _output_worker(self) -> None: + connector_uuid = str(self.uuid()) + log.debug( + "Connector endpoint output worker %s started", + self._input_worker_thread.name, + extra={ + "event": "endpoint.output_worker_started", + "connector_uuid": connector_uuid, + "thread_name": self._input_worker_thread.name, + }, + ) + + if self._connector_hello.protocol_version == HELLO_PACKET.protocol_version: + self._put_egress_packet(HandshakeAccepted(accepted_signals=[])) + log.debug(f"Accepted handshake for endpoint {self.uuid()}") + else: + self._put_egress_packet( + HandshakeRejected(reason="Protocol versions mismatch") + ) + log.debug(f"Rejected handshake for endpoint {self.uuid()}") + + self._heartbeat_worker_thread.start() + + while not self._stop_event.is_set(): + messages: list[ProtocolMessage] = [] + try: + messages.append(self._heartbeat_egress_queue.get_nowait()) + except Empty: + self._stop_event.wait(0.01) + pass + + for message in messages: + self._put_egress_packet(message) + + self._output_worker_stopped_event.set() + + log.debug( + "Connector endpoint output worker %s stopped", + self._input_worker_thread.name, + extra={ + "event": "endpoint.output_worker_stopped", + "connector_uuid": connector_uuid, + "thread_name": self._input_worker_thread.name, + }, + ) class ConnectorRegistry: @@ -90,30 +274,99 @@ class ConnectorRegistry: self._endpoints: dict[UUID, ConnectorEndpoint] = {} self._lock = RLock() - def register(self, hello: ConnectorHello) -> None: - endpoint = ConnectorEndpoint(hello) + def register( + self, hello: ConnectorHello, timeout_event: threading.Event + ) -> ConnectorEndpoint: + endpoint = ConnectorEndpoint(hello, timeout_event) with self._lock: current = self._endpoints.get(hello.connector_uuid) if current is not None: + log.warning( + "Connector %s is already registered", + hello.connector_uuid, + extra={ + "event": "connector.registration_rejected", + "connector_uuid": str(hello.connector_uuid), + "reason": "duplicate_uuid", + }, + ) raise ConnectorRegistryAlreadyRegisteredError self._endpoints[hello.connector_uuid] = endpoint + endpoint_count = len(self._endpoints) + + log.info( + "Registered connector %s (%s)", + hello.connector_name, + hello.connector_uuid, + extra={ + "event": "connector.registered", + "connector_uuid": str(hello.connector_uuid), + "connector_name": hello.connector_name, + "connector_version": hello.connector_version, + "protocol_version": hello.protocol_version.get_version(), + "endpoint_count": endpoint_count, + }, + ) return endpoint def unregister(self, endpoint: ConnectorEndpoint) -> None: + connector_uuid = endpoint.uuid() with self._lock: - current = self._endpoints.get(endpoint.uuid()) + current = self._endpoints.get(connector_uuid) if current is endpoint: - del self._endpoints[endpoint.uuid()] + del self._endpoints[connector_uuid] + endpoint_count = len(self._endpoints) + else: + log.debug( + "Connector %s was not registered", + connector_uuid, + extra={ + "event": "connector.unregister_noop", + "connector_uuid": str(connector_uuid), + }, + ) + return + + endpoint.stop() + + log.info( + "Unregistered connector %s", + connector_uuid, + extra={ + "event": "connector.unregistered", + "connector_uuid": str(connector_uuid), + "endpoint_count": endpoint_count, + }, + ) def get(self, connector_uuid: UUID) -> ConnectorEndpoint | None: with self._lock: return self._endpoints.get(connector_uuid) def stop(self) -> None: - for endpoint in self._endpoints.values(): + with self._lock: + endpoints = tuple(self._endpoints.values()) + self._endpoints.clear() + + log.debug( + "Stopping connector registry with %d endpoint(s)", + len(endpoints), + extra={ + "event": "connector_registry.stopping", + "endpoint_count": len(endpoints), + }, + ) + for endpoint in endpoints: endpoint.stop() + log.debug( + "Connector registry stopped", + extra={ + "event": "connector_registry.stopped", + "endpoint_count": 0, + }, + ) diff --git a/src/dynalab_core/protocols/json/errors.py b/src/dynalab_core/protocols/json/errors.py index 65a284d..cee5d1b 100644 --- a/src/dynalab_core/protocols/json/errors.py +++ b/src/dynalab_core/protocols/json/errors.py @@ -17,3 +17,10 @@ class JsonServerStartupError(JsonServerError): def __init__(self, error: Exception) -> None: self.error = error super().__init__(f"JSON server failed to start: {error}") + + +class JsonServerValueError(JsonServerError): + """JsonServer incorrect value detected""" + + def __init__(self, message: str) -> None: + super().__init__(str) diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index 0a1c4f4..5aa8ae2 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -4,21 +4,31 @@ import asyncio from asyncio import Server +import logging +from queue import Empty import threading from threading import Thread +from time import monotonic +from uuid import uuid4 from dynalab_core.config import CoreConfig +from dynalab_core.constants import HELLO_PACKET +from dynalab_core.protocols.endpoint import ConnectorEndpoint, ConnectorRegistry +from dynalab_core.protocols.errors import ConnectorRegistryAlreadyRegisteredError from dynalab_core.protocols.json.errors import ( JsonServerStartupError, JsonServerTimeoutError, ) +from dynalab_core.protocols.json.wire import read_message, write_message +from dynalab_core.protocols.packets.handshake import ConnectorHello + +log = logging.getLogger(__name__) class JsonServer: def __init__( - self, - config: CoreConfig, + self, config: CoreConfig, connector_registry: ConnectorRegistry ) -> None: self._thread: Thread | None = None self._server: Server | None = None @@ -29,38 +39,102 @@ class JsonServer: self._timeout: float = 10 self._stop_event = threading.Event() self._stopped_event = threading.Event() + self._connector_registry = connector_registry + self._handler_tasks: set[asyncio.Task[None]] = set() def start(self) -> None: + log.info( + "Starting JSON server on %s", + self._config.bind_str(), + extra={ + "event": "json_server.starting", + "host": self._config.host, + "port": self._config.port, + "timeout_s": self._timeout, + }, + ) self._thread = Thread( target=self._json_server_thread_main, name="json_server_thread", daemon=True ) self._thread.start() res = self._started_event.wait(self._timeout) - print(f"result: {res}") if not res: + log.error( + "JSON server did not start within %.1f seconds", + self._timeout, + extra={ + "event": "json_server.start_timeout", + "host": self._config.host, + "port": self._config.port, + "timeout_s": self._timeout, + }, + ) raise JsonServerTimeoutError if self._startup_error is not None: raise JsonServerStartupError(self._startup_error) + bound_addresses = [ + str(sock.getsockname()) for sock in (self._server.sockets or []) + ] + log.info( + "JSON server listening on %s", + ", ".join(bound_addresses), + extra={ + "event": "json_server.started", + "bound_addresses": bound_addresses, + }, + ) + def stop(self) -> None: + log.info("Stopping JSON server", extra={"event": "json_server.stopping"}) self._stop_event.set() - self._stopped_event.wait(10) + if not self._stopped_event.wait(10): + log.error( + "JSON server did not stop within 10 seconds", + extra={"event": "json_server.stop_timeout", "timeout_s": 10}, + ) + else: + log.info("JSON server stopped", extra={"event": "json_server.stopped"}) def _json_server_thread_main(self) -> None: - asyncio.run(self._run_json_server()) + log.debug( + "JSON server thread started", + extra={ + "event": "json_server.thread_started", + "thread_name": threading.current_thread().name, + }, + ) + try: + asyncio.run(self._run_json_server()) + finally: + log.debug( + "JSON server thread stopped", + extra={ + "event": "json_server.thread_stopped", + "thread_name": threading.current_thread().name, + }, + ) + self._stopped_event.set() async def _run_json_server(self) -> None: try: self._server = await asyncio.start_server( - self._handle_json_connection, + self._start_connection_handler, host=self._config.host, port=self._config.port, ) except OSError as error: - print( - f"Failed to start ingress server on {self._config.bind_str()}: {error}" + log.exception( + "Failed to start JSON server on %s", + self._config.bind_str(), + extra={ + "event": "json_server.start_failed", + "host": self._config.host, + "port": self._config.port, + "exception_type": type(error).__name__, + }, ) self._startup_error = error if not self._debug_timeout_test: @@ -69,21 +143,262 @@ class JsonServer: if not self._debug_timeout_test: self._started_event.set() if self._startup_error is not None: - self._stopped_event.set() return async with self._server: - await asyncio.to_thread(self._stop_event.wait) - self._stopped_event.set() + await self._wait_for_thread_event(self._stop_event) + + tasks = tuple(self._handler_tasks) + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + + log.debug("All handlers stopped") + + async def _wait_for_thread_event(self, event: threading.Event) -> None: + while not event.is_set(): + await asyncio.sleep(0.01) + + async def _input_task( + self, + reader: asyncio.StreamReader, + endpoint: ConnectorEndpoint, + ) -> None: + log.debug(f"Started input task for {endpoint.uuid()}") + while not self._stop_event.is_set(): + message = await read_message(reader) + endpoint.put_ingress_packet(message) + + async def _output_task( + self, + writer: asyncio.StreamWriter, + endpoint: ConnectorEndpoint, + ) -> None: + log.debug(f"Started output task for {endpoint.uuid()}") + while not self._stop_event.is_set(): + try: + message = endpoint.get_egress_packet_no_wait() + except Empty: + await asyncio.sleep(0.01) + continue + await write_message(writer, message) + + def _start_connection_handler( + self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter + ) -> None: + task = asyncio.create_task(self._handle_json_connection(reader, writer)) + self._handler_tasks.add(task) + task.add_done_callback(self._handler_finished) + + def _handler_finished(self, task: asyncio.Task[None]) -> None: + self._handler_tasks.discard(task) + + if task.cancelled(): + return + + error = task.exception() + if error is not None: + log.error( + "Connection handler terminated with an exception", + exc_info=(type(error), error, error.__traceback__), + ) async def _handle_json_connection( self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter ) -> None: - print(f"Connected to: {writer.get_extra_info('peername')}") + connection_id = uuid4().hex[:12] + peer_address = str(writer.get_extra_info("peername")) + started_at = monotonic() + reason = "handshake_incomplete" + connector_endpoint: ConnectorEndpoint | None = None + connection_tasks: list[asyncio.Task[None]] = [] + + context = { + "connection_id": connection_id, + "peer_address": peer_address, + } + log.debug( + "Accepted connection %s from %s", + connection_id, + peer_address, + extra={"event": "json_connection.accepted", **context}, + ) - # TODO: Build out handler to handshake, register peer and pipe packets into connector queues try: - await reader.readline() + try: + await asyncio.wait_for(write_message(writer, HELLO_PACKET), 5.0) + except TimeoutError: + reason = "hello_write_timeout" + log.debug( + "Connection %s timed out while sending server hello", + connection_id, + extra={ + "event": "connector.handshake_timed_out", + "phase": "write_server_hello", + "timeout_s": 5.0, + **context, + }, + ) + return + + try: + message = await asyncio.wait_for(read_message(reader), timeout=5.0) + except TimeoutError: + reason = "connector_hello_timeout" + log.debug( + "Connection %s timed out waiting for connector hello", + connection_id, + extra={ + "event": "connector.handshake_timed_out", + "phase": "read_connector_hello", + "timeout_s": 5.0, + **context, + }, + ) + return + + if not isinstance(message, ConnectorHello): + reason = "unexpected_message" + log.warning( + "Connection %s sent %s instead of connector_hello", + connection_id, + message.type, + extra={ + "event": "connector.handshake_rejected", + "reason": reason, + "message_type": message.type, + **context, + }, + ) + return + + connector_hello = message + timeout_event = threading.Event() + connector_endpoint = self._connector_registry.register( + connector_hello, timeout_event + ) + + reason = "handler_completed" + log.info( + "Connector %s connected from %s", + connector_hello.connector_uuid, + peer_address, + extra={ + "event": "connector.connected", + "connector_uuid": str(connector_hello.connector_uuid), + "connector_name": connector_hello.connector_name, + "connector_version": connector_hello.connector_version, + **context, + }, + ) + + input_task = asyncio.create_task( + self._input_task(reader, connector_endpoint) + ) + output_task = asyncio.create_task( + self._output_task(writer, connector_endpoint) + ) + server_stop_task = asyncio.create_task( + self._wait_for_thread_event(self._stop_event) + ) + timeout_task = asyncio.create_task( + self._wait_for_thread_event(timeout_event) + ) + connection_tasks = [ + input_task, + output_task, + server_stop_task, + timeout_task, + ] + + done, _ = await asyncio.wait( + connection_tasks, return_when=asyncio.FIRST_COMPLETED + ) + if timeout_task in done: + reason = "endpoint_timeout" + elif server_stop_task in done: + reason = "server_shutdown" + + for task in done: + if task is input_task or task is output_task: + task.result() + + except asyncio.CancelledError: + reason = "server_shutdown" + log.debug( + "Connection %s cancelled during server shutdown", + connection_id, + extra={"event": "json_connection.cancelled", **context}, + ) + raise + except ConnectionError as error: + reason = "peer_disconnected" + log.debug( + "Connection %s disconnected: %s", + connection_id, + error, + extra={ + "event": "connector.handshake_disconnected", + "exception_type": type(error).__name__, + **context, + }, + ) + except ValueError as error: + reason = "invalid_frame" + log.warning( + "Connection %s sent an invalid handshake frame: %s", + connection_id, + error, + extra={ + "event": "connector.handshake_rejected", + "reason": reason, + **context, + }, + ) + except ConnectorRegistryAlreadyRegisteredError: + reason = "duplicate_connector" + log.debug( + "Connection %s rejected because its connector is already registered", + connection_id, + extra={ + "event": "json_connection.rejected", + "reason": reason, + **context, + }, + ) + except Exception: + reason = "handler_failed" + log.exception( + "Connection %s handler failed", + connection_id, + extra={"event": "json_connection.failed", **context}, + ) + raise finally: + for task in connection_tasks: + task.cancel() + if connection_tasks: + await asyncio.gather(*connection_tasks, return_exceptions=True) + writer.close() - await writer.wait_closed() - print("Disconnected") + try: + await writer.wait_closed() + except ConnectionError: + pass + + if connector_endpoint is not None: + await asyncio.to_thread( + self._connector_registry.unregister, connector_endpoint + ) + + duration_ms = round((monotonic() - started_at) * 1000) + log.debug( + "Closed connection %s after %d ms (%s)", + connection_id, + duration_ms, + reason, + extra={ + "event": "json_connection.closed", + "reason": reason, + "duration_ms": duration_ms, + **context, + }, + ) diff --git a/src/dynalab_core/protocols/json/wire.py b/src/dynalab_core/protocols/json/wire.py new file mode 100644 index 0000000..9c1f6d4 --- /dev/null +++ b/src/dynalab_core/protocols/json/wire.py @@ -0,0 +1,32 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: MIT + +from asyncio import StreamReader, StreamWriter +from pydantic import TypeAdapter + +from dynalab_core.protocols.packets import ProtocolMessage + +_MESSAGE_ADAPTER = TypeAdapter(ProtocolMessage) + + +async def write_message(writer: StreamWriter, message: ProtocolMessage) -> None: + writer.write(message.model_dump_json().encode("utf-8") + b"\n") + await writer.drain() + + +async def read_message( + reader: StreamReader, *, max_frame_bytes: int = 65_536 +) -> ProtocolMessage: + line = await reader.readline() + + if not line: + raise ConnectionError("Peer disconnected") + + if len(line) > max_frame_bytes: + raise ValueError("Incoming protocol frame exceeds maximum size") + + if not line.endswith(b"\n"): + raise ValueError("Protocol frame is missing newline delimiter") + + return _MESSAGE_ADAPTER.validate_json(line) diff --git a/src/dynalab_core/protocols/packets/__init__.py b/src/dynalab_core/protocols/packets/__init__.py index 8da7159..055158c 100644 --- a/src/dynalab_core/protocols/packets/__init__.py +++ b/src/dynalab_core/protocols/packets/__init__.py @@ -8,9 +8,10 @@ from dynalab_core.protocols.packets.handshake import ( HandshakeAccepted, HandshakeRejected, ) +from dynalab_core.protocols.packets.heartbeat import Heartbeat ProtocolMessage = Annotated[ - DynaLabHello | ConnectorHello | HandshakeAccepted | HandshakeRejected, + DynaLabHello | ConnectorHello | HandshakeAccepted | HandshakeRejected | Heartbeat, Field(discriminator="type"), ] diff --git a/src/dynalab_core/protocols/packets/handshake.py b/src/dynalab_core/protocols/packets/handshake.py index b4f0ccc..21b8a8c 100644 --- a/src/dynalab_core/protocols/packets/handshake.py +++ b/src/dynalab_core/protocols/packets/handshake.py @@ -7,7 +7,7 @@ from uuid import UUID from pydantic import BaseModel, Field from dynalab_core.protocols.common import VersionDescriptor -from dynalab_core.protocols.json.constants import PROTOCOL_VERSION +from dynalab_core.protocols.constants import PROTOCOL_VERSION class DynaLabHello(BaseModel): diff --git a/src/dynalab_core/protocols/packets/heartbeat.py b/src/dynalab_core/protocols/packets/heartbeat.py new file mode 100644 index 0000000..35f0d1d --- /dev/null +++ b/src/dynalab_core/protocols/packets/heartbeat.py @@ -0,0 +1,14 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + + +from typing import Literal +from pydantic import BaseModel + + +class Heartbeat(BaseModel): + type: Literal["heartbeat"] = "heartbeat" + sequence: int + send_timestamp: int + return_timestamp: int | None = None diff --git a/test/manual/core.py b/test/manual/core.py index 604e84b..a4656d5 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -1,17 +1,28 @@ -from time import sleep +import logging + +from rich.logging import RichHandler + from dynalab_core import Core from dynalab_core.config import CoreConfig +logging.basicConfig( + level=logging.DEBUG, + format="%(name)s %(message)s", + datefmt="%H:%M:%S", + handlers=[RichHandler(rich_tracebacks=True, show_path=False)], + force=True, +) +log = logging.getLogger(__name__) + config = CoreConfig(port=8765) -print(config) +log.info("Configured manual core on %s", config.bind_str()) dl_core = Core(config) dl_core.start() -print("started") try: while True: dl_core.wait(1) - print("waiting") except KeyboardInterrupt: + log.info("Received keyboard interrupt") dl_core.stop() diff --git a/test/manual/peer.py b/test/manual/peer.py index 3dc02f4..11b83b5 100644 --- a/test/manual/peer.py +++ b/test/manual/peer.py @@ -1,4 +1,25 @@ import asyncio +import logging +from uuid import uuid4 + +from dynalab_core.protocols.packets import ProtocolMessage +from rich.logging import RichHandler + +from dynalab_core.protocols.common import VersionDescriptor +from dynalab_core.protocols.constants import PROTOCOL_VERSION +from dynalab_core.protocols.json.wire import read_message, write_message +from dynalab_core.protocols.packets.handshake import ConnectorHello, DynaLabHello + +CONNECTOR_VERSION = VersionDescriptor(type="alpha", major=0, minor=0, patch=1) + +logging.basicConfig( + level=logging.DEBUG, + format="%(name)s %(message)s", + datefmt="%H:%M:%S", + handlers=[RichHandler(rich_tracebacks=True, show_path=False)], + force=True, +) +log = logging.getLogger(__name__) async def main() -> None: @@ -7,14 +28,42 @@ async def main() -> None: port=8765, ) - print("Connected") + log.info("Connected to DynaLab core") try: - await asyncio.sleep(10) + try: + message = await asyncio.wait_for(read_message(reader), timeout=30.0) + except TimeoutError: + log.warning("Timed out waiting for server hello") + return + if not isinstance(message, DynaLabHello): + log.warning("Expected server hello, received %s", message.type) + return + dynalab_hello = message + log.debug("Received hello from core instance %s", dynalab_hello.instance_id) + connector_hello = ConnectorHello( + connector_uuid=uuid4(), + protocol_version=PROTOCOL_VERSION, + connector_name="Test connector", + connector_version=CONNECTOR_VERSION.get_version(), + ) + try: + await asyncio.wait_for(write_message(writer, connector_hello), timeout=5.0) + except TimeoutError: + log.warning("Timed out sending connector hello") + return + log.info("Connector hello sent for %s", connector_hello.connector_uuid) + + try: + while True: + message = await read_message(reader) + log.info(message) + except KeyboardInterrupt: + return finally: writer.close() await writer.wait_closed() - print("Disconnected") + log.info("Disconnected from DynaLab core") asyncio.run(main()) diff --git a/test/test_core.py b/test/test_core.py index 731bf9d..9649d8d 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -1,3 +1,5 @@ +import logging + import pytest from dynalab_core import Core from dynalab_core.config import CoreConfig @@ -5,14 +7,28 @@ from dynalab_core.errors import CoreStateMismatchError from test.common import find_available_port -def test_core_cannot_start_twice() -> None: +def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None: port = find_available_port(8765) core = Core(CoreConfig(port=port)) - core.start() + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + core.start() - try: - with pytest.raises(CoreStateMismatchError): - core.start() - finally: - core.stop() + try: + with pytest.raises(CoreStateMismatchError): + core.start() + finally: + core.stop() + + events = [getattr(record, "event", None) for record in caplog.records] + assert events.count("core.starting") == 1 + assert "core.start_rejected" in events + assert "core.stopped" in events + + rejection = next( + record + for record in caplog.records + if getattr(record, "event", None) == "core.start_rejected" + ) + assert rejection.levelno == logging.WARNING + assert rejection.core_state == "started" diff --git a/test/test_endpoint_logging.py b/test/test_endpoint_logging.py new file mode 100644 index 0000000..1e1da9d --- /dev/null +++ b/test/test_endpoint_logging.py @@ -0,0 +1,41 @@ +import logging +from queue import Queue +from uuid import uuid4 + +import pytest + +from dynalab_core.protocols.constants import PROTOCOL_VERSION +from dynalab_core.protocols.endpoint import ConnectorEndpoint +from dynalab_core.protocols.errors import ConnectorEndpointQueueFullError +from dynalab_core.protocols.packets import ProtocolMessage +from dynalab_core.protocols.packets.handshake import ConnectorHello + + +def test_full_endpoint_queue_is_logged(caplog: pytest.LogCaptureFixture) -> None: + hello = ConnectorHello( + connector_uuid=uuid4(), + protocol_version=PROTOCOL_VERSION, + connector_name="Test connector", + connector_version="0.1.0-test", + ) + endpoint = ConnectorEndpoint(hello) + endpoint._packet_ingress_queue = Queue[ProtocolMessage](maxsize=1) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + try: + endpoint.put_ingress_packet(hello) + with pytest.raises(ConnectorEndpointQueueFullError): + endpoint.put_ingress_packet(hello) + finally: + endpoint.stop() + + queue_record = next( + record + for record in caplog.records + if getattr(record, "event", None) == "endpoint.queue_full" + ) + assert queue_record.levelno == logging.WARNING + assert queue_record.connector_uuid == str(hello.connector_uuid) + assert queue_record.queue_direction == "ingress" + assert queue_record.queue_size == 1 + assert queue_record.queue_capacity == 1 diff --git a/test/test_json_server.py b/test/test_json_server.py index 10a0a6b..6fd43c2 100644 --- a/test/test_json_server.py +++ b/test/test_json_server.py @@ -1,37 +1,141 @@ +import logging +import socket +import time +from uuid import uuid4 + import pytest from dynalab_core import Core from dynalab_core.config import CoreConfig +from dynalab_core.protocols.constants import PROTOCOL_VERSION from dynalab_core.protocols.json.errors import ( JsonServerStartupError, JsonServerTimeoutError, ) +from dynalab_core.protocols.packets.handshake import ConnectorHello from test.common import find_available_port -def test_json_server_raises_timeout_error() -> None: +def _wait_for_event( + caplog: pytest.LogCaptureFixture, event: str, timeout: float = 1.0 +) -> logging.LogRecord: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + for record in caplog.records: + if getattr(record, "event", None) == event: + return record + time.sleep(0.01) + raise AssertionError(f"Log event {event!r} was not emitted") + + +def test_json_server_raises_timeout_error( + caplog: pytest.LogCaptureFixture, +) -> None: port = find_available_port(8765) core = Core(CoreConfig(port=port)) core._json_server._debug_timeout_test = True core._json_server._timeout = 0.01 - try: - with pytest.raises(JsonServerTimeoutError): - core.start() - finally: - core.stop() + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + try: + with pytest.raises(JsonServerTimeoutError): + core.start() + finally: + core.stop() + + timeout_record = next( + record + for record in caplog.records + if getattr(record, "event", None) == "json_server.start_timeout" + ) + assert timeout_record.levelno == logging.ERROR + assert timeout_record.port == port -def test_json_server_raises_startup_error() -> None: +def test_json_server_raises_startup_error( + caplog: pytest.LogCaptureFixture, +) -> None: port = find_available_port(8765) core1 = Core(CoreConfig(port=port)) core2 = Core(CoreConfig(port=port)) - core1.start() + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + core1.start() - try: - with pytest.raises(JsonServerStartupError): - core2.start() - finally: - core1.stop() - core2.stop() + try: + with pytest.raises(JsonServerStartupError): + core2.start() + finally: + core1.stop() + core2.stop() + + failure_record = next( + record + for record in caplog.records + if getattr(record, "event", None) == "json_server.start_failed" + ) + assert failure_record.levelno == logging.ERROR + assert failure_record.port == port + assert failure_record.exc_info is not None + + +def test_json_server_logs_invalid_handshake( + caplog: pytest.LogCaptureFixture, +) -> None: + port = find_available_port(8765) + core = Core(CoreConfig(port=port)) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + core.start() + try: + with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: + server_hello = peer.makefile("rb").readline() + assert server_hello + peer.sendall(b"not-json\n") + + rejection = _wait_for_event( + caplog, "connector.handshake_rejected" + ) + finally: + core.stop() + + assert rejection.levelno == logging.WARNING + assert rejection.reason == "invalid_frame" + assert rejection.connection_id + assert rejection.peer_address + + +def test_json_server_waits_for_connection_handlers_on_stop( + caplog: pytest.LogCaptureFixture, +) -> None: + port = find_available_port(8765) + core = Core(CoreConfig(port=port)) + connector_uuid = uuid4() + connector_hello = ConnectorHello( + connector_uuid=connector_uuid, + protocol_version=PROTOCOL_VERSION, + connector_name="Test connector", + connector_version="0.1.0-test", + ) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + core.start() + try: + with socket.create_connection(("127.0.0.1", port), timeout=1) as peer: + peer_file = peer.makefile("rb") + assert peer_file.readline() + peer.sendall( + connector_hello.model_dump_json().encode("utf-8") + b"\n" + ) + assert b'"type":"handshake_accepted"' in peer_file.readline() + + core.stop() + finally: + if core._state != "stopped": + core.stop() + + events = [getattr(record, "event", None) for record in caplog.records] + assert "json_server.stop_timeout" not in events + assert core._json_server._stopped_event.is_set() + assert not core._json_server._handler_tasks + assert core._connector_registry.get(connector_uuid) is None diff --git a/uv.lock b/uv.lock index 8675709..3e67e4a 100644 --- a/uv.lock +++ b/uv.lock @@ -31,13 +31,17 @@ dependencies = [ [package.dev-dependencies] dev = [ { name = "pytest" }, + { name = "rich" }, ] [package.metadata] requires-dist = [{ name = "pydantic", specifier = ">=2.13.4" }] [package.metadata.requires-dev] -dev = [{ name = "pytest", specifier = ">=9.1.1" }] +dev = [ + { name = "pytest", specifier = ">=9.1.1" }, + { name = "rich", specifier = ">=15.0.0" }, +] [[package]] name = "iniconfig" @@ -48,6 +52,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "markdown-it-py" +version = "4.2.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "mdurl" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/06/ff/7841249c247aa650a76b9ee4bbaeae59370dc8bfd2f6c01f3630c35eb134/markdown_it_py-4.2.0.tar.gz", hash = "sha256:04a21681d6fbb623de53f6f364d352309d4094dd4194040a10fd51833e418d49", size = 82454, upload-time = "2026-05-07T12:08:28.36Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/81/4da04ced5a082363ecfa159c010d200ecbd959ae410c10c0264a38cac0f5/markdown_it_py-4.2.0-py3-none-any.whl", hash = "sha256:9f7ebbcd14fe59494226453aed97c1070d83f8d24b6fc3a3bcf9a38092641c4a", size = 91687, upload-time = "2026-05-07T12:08:27.182Z" }, +] + +[[package]] +name = "mdurl" +version = "0.1.2" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d6/54/cfe61301667036ec958cb99bd3efefba235e65cdeb9c84d24a8293ba1d90/mdurl-0.1.2.tar.gz", hash = "sha256:bb413d29f5eea38f31dd4754dd7377d4465116fb207585f97bf925588687c1ba", size = 8729, upload-time = "2022-08-14T12:40:10.846Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/b3/38/89ba8ad64ae25be8de66a6d463314cf1eb366222074cfda9ee839c56a4b4/mdurl-0.1.2-py3-none-any.whl", hash = "sha256:84008a41e51615a49fc9966191ff91509e3c40b939176e643fd50a5c2196b8f8", size = 9979, upload-time = "2022-08-14T12:40:09.779Z" }, +] + [[package]] name = "packaging" version = "26.2" @@ -162,6 +187,19 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/24/25/1de2678b631f5a49215c6c96fff41ba892b0a34df68d6d80292b1b48aa7f/pytest-9.1.1-py3-none-any.whl", hash = "sha256:37a86b45efb9a47a61a36449063e8e18d0cab3161329fc099eb21783169c4f0c", size = 386536, upload-time = "2026-06-19T10:58:31.347Z" }, ] +[[package]] +name = "rich" +version = "15.0.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "markdown-it-py" }, + { name = "pygments" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/c0/8f/0722ca900cc807c13a6a0c696dacf35430f72e0ec571c4275d2371fca3e9/rich-15.0.0.tar.gz", hash = "sha256:edd07a4824c6b40189fb7ac9bc4c52536e9780fbbfbddf6f1e2502c31b068c36", size = 230680, upload-time = "2026-04-12T08:24:00.75Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/82/3b/64d4899d73f91ba49a8c18a8ff3f0ea8f1c1d75481760df8c68ef5235bf5/rich-15.0.0-py3-none-any.whl", hash = "sha256:33bd4ef74232fb73fe9279a257718407f169c09b78a87ad3d296f548e27de0bb", size = 310654, upload-time = "2026-04-12T08:24:02.83Z" }, +] + [[package]] name = "typing-extensions" version = "4.16.0"