From bbc5aac891bf9cc34d36f5a58671a265197130d3 Mon Sep 17 00:00:00 2001 From: Hector van der Aa Date: Fri, 11 Sep 2026 23:45:27 +0200 Subject: [PATCH] Prepared for playback engine --- src/dynalab_core/__init__.py | 183 ++++++++--- src/dynalab_core/buffer.py | 4 + src/dynalab_core/derive.py | 184 +++++++++-- src/dynalab_core/dlpak.py | 34 ++- src/dynalab_core/errors.py | 8 + src/dynalab_core/playback.py | 8 + src/dynalab_core/protocols/endpoint.py | 354 +++++++++++++++++----- src/dynalab_core/protocols/json/server.py | 133 +++++--- src/dynalab_core/types.py | 15 + test/manual/core.py | 16 +- test/test_core.py | 61 ++++ test/test_derive.py | 46 ++- test/test_dlpak.py | 29 ++ test/test_json_server.py | 69 ++++- 14 files changed, 955 insertions(+), 189 deletions(-) create mode 100644 src/dynalab_core/playback.py create mode 100644 src/dynalab_core/types.py create mode 100644 test/test_dlpak.py diff --git a/src/dynalab_core/__init__.py b/src/dynalab_core/__init__.py index 2a6b86b..06179c0 100644 --- a/src/dynalab_core/__init__.py +++ b/src/dynalab_core/__init__.py @@ -5,6 +5,7 @@ from datetime import datetime, timezone import logging +from pathlib import Path from queue import Empty, Queue import threading from threading import Lock, Thread @@ -18,13 +19,18 @@ from dynalab_core.config import CoreConfig from dynalab_core.constants import CORE_VERSION from dynalab_core.derive import DeriveRegistry from dynalab_core.dlpak import DLPak -from dynalab_core.errors import CoreStateMismatchError +from dynalab_core.errors import ( + CoreModeIncorrectError, + CoreModeSwitchImpossibleError, + CoreStateMismatchError, +) from dynalab_core.protocols.endpoint import ConnectorRegistry from dynalab_core.protocols.common import VersionDescriptor from dynalab_core.protocols.json.server import JsonServer from dynalab_core.protocols.packets import ProtocolMessage from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor +from dynalab_core.types import CoreMode, CoreModeState from dynalab_core.values import Value @@ -34,15 +40,17 @@ log.addHandler(logging.NullHandler()) class Core: def __init__(self, config: CoreConfig) -> None: - self._state: Literal["uninitd", "initd", "started", "stopping", "stopped"] = ( - "unintid" - ) + self._state: Literal[ + "uninitd", "initd", "started", "stopping", "stopped", "stop_incomplete" + ] = "unintid" + self._mode: CoreModeState = CoreModeState() self._recording = threading.Event() self._recording_buffer = ValueBuffer() self._recording_timestamp: datetime = datetime.now(timezone.utc) self._processing_buffer: DLPak | None = None self._core_version: VersionDescriptor = CORE_VERSION self._stop_event: threading.Event = threading.Event() + self._input_worker_stopped_event: threading.Event = threading.Event() self._core_config: CoreConfig = config self._live_values: dict[UUID, Value] = {} self._live_values_lock = Lock() @@ -56,7 +64,7 @@ class Core: self._derive_registry = DeriveRegistry( self._data_input_queue, self._get_live_value_descriptor ) - self._connector_registry = ConnectorRegistry(self._data_input_queue) + self._connector_registry = ConnectorRegistry(self._data_input_queue, self._mode) self._json_server = JsonServer(self._core_config, self._connector_registry) self._state = "initd" @@ -100,30 +108,83 @@ class Core: except KeyboardInterrupt: raise - def stop(self) -> None: + def stop(self) -> bool: log.info( "Core stopping", extra={"event": "core.stopping", "core_state": self._state}, ) - self._json_server.stop() - self._connector_registry.stop() - self._derive_registry.stop() + json_server_stopped = self._json_server.stop() + connector_registry_stopped = self._connector_registry.stop() + derive_registry_stopped = self._derive_registry.stop() self._stop_event.set() - self._state = "stopped" - log.info( - "Core stopped", - extra={"event": "core.stopped", "core_state": self._state}, + input_worker_stopped = self._input_worker_stopped_event.wait(5.0) + if not input_worker_stopped: + log.error( + "Core input worker did not stop within 5 seconds", + extra={ + "event": "core.input_worker_stop_timeout", + "thread_name": self._input_worker_thread.name, + "timeout_s": 5.0, + }, + ) + stopped = all( + ( + json_server_stopped, + connector_registry_stopped, + derive_registry_stopped, + input_worker_stopped, + ) ) + if stopped: + self._state = "stopped" + log.info( + "Core stopped", + extra={"event": "core.stopped", "core_state": self._state}, + ) + else: + self._state = "stop_incomplete" + log.error( + "Core stop incomplete", + extra={ + "event": "core.stop_incomplete", + "core_state": self._state, + "json_server_stopped": json_server_stopped, + "connector_registry_stopped": connector_registry_stopped, + "derive_registry_stopped": derive_registry_stopped, + "input_worker_stopped": input_worker_stopped, + }, + ) + return stopped + + def set_mode(self, mode: CoreMode) -> None: + if self._recording.is_set(): + raise CoreModeSwitchImpossibleError + + self._mode.value = mode + with self._live_values_lock: + self._live_values = {} def start_recording(self) -> None: + if self._mode.value != "realtime": + raise CoreModeIncorrectError self._recording_buffer.clear() self._recording_timestamp = datetime.now(timezone.utc) self._recording.set() + log.info( + "Recording started", + extra={ + "event": "recording.started", + "recording_timestamp": self._recording_timestamp.isoformat(), + }, + ) def stop_recording(self) -> None: + if self._mode.value != "realtime": + raise CoreModeIncorrectError self._recording.clear() time.sleep(1) self._recording_buffer.normalize() + self._recording_buffer.sort() self._processing_buffer = DLPak() self._processing_buffer.set_data(self._recording_buffer) connector_signal_descriptors = ( @@ -135,6 +196,22 @@ class Core: self._recording_timestamp, connector_signal_descriptors + derive_signal_descriptors, ) + log.info( + "Recording stopped with %d sample(s) from %d signal(s)", + len(self._recording_buffer), + len(self._recording_buffer.get_signal_ids()), + extra={ + "event": "recording.stopped", + "sample_count": len(self._recording_buffer), + "signal_count": len(self._recording_buffer.get_signal_ids()), + "duration_ms": round( + ( + datetime.now(timezone.utc) - self._recording_timestamp + ).total_seconds() + * 1000 + ), + }, + ) def get_signal_descriptor(self, signal_id: UUID) -> SignalDescriptor | None: signal = self._connector_registry.get_signal_descriptor(signal_id) @@ -174,6 +251,10 @@ class Core: def unbind_derive_unit(self, unit_id: UUID) -> None: self._derive_registry.unregister(unit_id) + def write(self, dir: str | Path, filename: str) -> None: + if self._processing_buffer is not None: + self._processing_buffer.write(dir, filename) + def _get_live_value_descriptor(self, signal_id: UUID) -> ValueDescriptor | None: with self._live_values_lock: live_value = self._live_values.get(signal_id) @@ -183,29 +264,63 @@ class Core: return live_value.to_descriptor(signal_id) def _input_worker(self) -> None: - while not self._stop_event.is_set(): - try: - message = self._data_input_queue.get_nowait() - except Empty: - self._stop_event.wait(0.01) - continue + log.debug( + "Core input worker started", + extra={ + "event": "core.input_worker_started", + "thread_name": threading.current_thread().name, + }, + ) + try: + while not self._stop_event.is_set(): + try: + message = self._data_input_queue.get_nowait() + except Empty: + self._stop_event.wait(0.01) + continue - if isinstance(message, ValueDescriptor): - with self._live_values_lock: - live_value = self._live_values.get(message.signal_id) + if isinstance(message, ValueDescriptor): + with self._live_values_lock: + live_value = self._live_values.get(message.signal_id) - if live_value is None: - signal_descriptor = self.get_signal_descriptor(message.signal_id) + if live_value is None: + signal_descriptor = self.get_signal_descriptor( + message.signal_id + ) - live_value = Value( - signal_descriptor.timeout_ms if signal_descriptor else 2000 + live_value = Value( + signal_descriptor.timeout_ms + if signal_descriptor + else 2000 + ) + self._live_values[message.signal_id] = live_value + + live_value.update(message.value, message.timestamp) + if self._recording.is_set(): + self._recording_buffer.append( + message.timestamp, message.signal_id, message.value ) - self._live_values[message.signal_id] = live_value - live_value.update(message.value, message.timestamp) - if self._recording.is_set(): - self._recording_buffer.append( - message.timestamp, message.signal_id, message.value - ) - - self._derive_registry.put_data(message) + self._derive_registry.put_data(message) + except Exception as error: + log.exception( + "Core input worker failed", + extra={ + "event": "core.input_worker_failed", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + "message_type": type(message).__name__ + if "message" in locals() + else None, + }, + ) + self._stop_event.set() + finally: + self._input_worker_stopped_event.set() + log.debug( + "Core input worker stopped", + extra={ + "event": "core.input_worker_stopped", + "thread_name": threading.current_thread().name, + }, + ) diff --git a/src/dynalab_core/buffer.py b/src/dynalab_core/buffer.py index 61a4463..c2c55e3 100644 --- a/src/dynalab_core/buffer.py +++ b/src/dynalab_core/buffer.py @@ -42,6 +42,10 @@ class ValueBuffer: for timestamp, signal_id, value in self._samples ] + def sort(self) -> None: + with self._lock: + self._samples.sort(key=lambda sample: sample[0]) + def export_csv(self) -> str: with self._lock: samples = list(self._samples) diff --git a/src/dynalab_core/derive.py b/src/dynalab_core/derive.py index 160a463..1eac251 100644 --- a/src/dynalab_core/derive.py +++ b/src/dynalab_core/derive.py @@ -78,21 +78,66 @@ class DeriveUnit: self._worker_thread.start() def _worker_function(self) -> None: - while not self._stop_event.is_set(): - try: - input_args = self._input_queue.get(timeout=0.1) - except Empty: - continue + derive_uuid = str(self._id) + log.debug( + "Derive worker %s started", + derive_uuid, + extra={ + "event": "derive.worker_started", + "derive_uuid": derive_uuid, + "thread_name": threading.current_thread().name, + }, + ) + try: + while not self._stop_event.is_set(): + try: + input_args = self._input_queue.get(timeout=0.1) + except Empty: + continue - if len(input_args) != self._num_input_args: - continue + if len(input_args) != self._num_input_args: + log.warning( + "Derive worker %s received %d argument(s), expected %d", + derive_uuid, + len(input_args), + self._num_input_args, + extra={ + "event": "derive.arguments_rejected", + "derive_uuid": derive_uuid, + "expected_argument_count": self._num_input_args, + "actual_argument_count": len(input_args), + }, + ) + continue - value = self._process_function(*input_args, self._return_signal) - self._return_queue.put(value) - self._stopped_event.set() + value = self._process_function(*input_args, self._return_signal) + self._return_queue.put(value) + except Exception as error: + log.exception( + "Derive worker %s failed", + derive_uuid, + extra={ + "event": "derive.worker_failed", + "derive_uuid": derive_uuid, + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) + finally: + self._stopped_event.set() + log.debug( + "Derive worker %s stopped", + derive_uuid, + extra={ + "event": "derive.worker_stopped", + "derive_uuid": derive_uuid, + "thread_name": threading.current_thread().name, + }, + ) def put_data(self, input: list[ValueDescriptor]) -> None: - self._input_queue.put(input) + if not self._stopped_event.is_set(): + self._input_queue.put(input) def process_offline(self, input_args: list[ValueDescriptor]) -> ValueDescriptor: if len(input_args) == self._num_input_args: @@ -102,9 +147,21 @@ class DeriveUnit: def get_input_signals(self) -> list[SignalDescriptor]: return self._input_signals - def stop(self) -> None: + def stop(self) -> bool: self._stop_event.set() - self._stopped_event.wait(5.0) + if not self._stopped_event.wait(5.0): + log.error( + "Derive worker %s did not stop within 5 seconds", + self._id, + extra={ + "event": "derive.stop_timeout", + "derive_uuid": str(self._id), + "thread_name": self._worker_thread.name, + "timeout_s": 5.0, + }, + ) + return False + return True def uuid(self) -> UUID: return self._id @@ -112,6 +169,9 @@ class DeriveUnit: def return_signal(self) -> SignalDescriptor: return self._return_signal + def is_running(self) -> bool: + return not self._stopped_event.is_set() + class DeriveRegistry: def __init__( @@ -174,14 +234,34 @@ class DeriveRegistry: with self._lock: self._units[unit_id] = unit + log.info( + "Registered derive unit %s", + unit_id, + extra={ + "event": "derive.registered", + "derive_uuid": str(unit_id), + "input_signal_ids": [str(signal.id) for signal in input_signals], + "output_signal_id": str(output_signal.id), + }, + ) + return unit - def unregister(self, unit_uuid: UUID) -> None: + def unregister(self, unit_uuid: UUID) -> bool: with self._lock: current = self._units.get(unit_uuid) if current is not None: - current.stop() + if not current.stop(): + log.error( + "Unable to unregister derive unit %s because it did not stop", + unit_uuid, + extra={ + "event": "derive.unregister_incomplete", + "derive_uuid": str(unit_uuid), + }, + ) + return False del self._units[unit_uuid] else: log.debug( @@ -192,7 +272,14 @@ class DeriveRegistry: "derive_uuid": str(unit_uuid), }, ) - return + return True + + log.info( + "Unregistered derive unit %s", + unit_uuid, + extra={"event": "derive.unregistered", "derive_uuid": str(unit_uuid)}, + ) + return True def get(self, unit_uuid: UUID) -> DeriveUnit | None: with self._lock: @@ -203,6 +290,8 @@ class DeriveRegistry: units = list(self._units.values()) for unit in units: + if not unit.is_running(): + continue signal = unit.return_signal() if signal.id == signal_id: return signal @@ -215,7 +304,8 @@ class DeriveRegistry: signals: list[SignalDescriptor] = [] for unit in units: - signals.append(unit.return_signal()) + if unit.is_running(): + signals.append(unit.return_signal()) return signals @@ -223,19 +313,55 @@ class DeriveRegistry: if not self._routing_stop_event.is_set(): self._routing_queue.put_nowait(message) - def stop(self) -> None: + def stop(self) -> bool: + log.debug( + "Stopping derive registry", + extra={"event": "derive_registry.stopping"}, + ) self._routing_stop_event.set() if not self._routing_worker_thread.is_alive(): self._routing_stopped_event.set() - self._routing_stopped_event.wait(5.0) + routing_stopped = self._routing_stopped_event.wait(5.0) + if not routing_stopped: + log.error( + "Derive routing worker did not stop within 5 seconds", + extra={ + "event": "derive.routing_stop_timeout", + "thread_name": self._routing_worker_thread.name, + "timeout_s": 5.0, + }, + ) with self._lock: units = list(self._units.values()) - for unit in units: - unit.stop() + unit_results = [unit.stop() for unit in units] + units_stopped = all(unit_results) self._units.clear() + stopped = routing_stopped and units_stopped + if stopped: + log.debug( + "Derive registry stopped", + extra={"event": "derive_registry.stopped", "unit_count": 0}, + ) + else: + log.error( + "Derive registry stop incomplete", + extra={ + "event": "derive_registry.stop_incomplete", + "routing_worker_stopped": routing_stopped, + "unit_workers_stopped": units_stopped, + }, + ) + return stopped def _routing_worker(self) -> None: + log.debug( + "Derive routing worker started", + extra={ + "event": "derive.routing_worker_started", + "thread_name": threading.current_thread().name, + }, + ) try: while not self._routing_stop_event.is_set(): try: @@ -244,8 +370,24 @@ class DeriveRegistry: continue self._route_data(message) + except Exception as error: + log.exception( + "Derive routing worker failed", + extra={ + "event": "derive.routing_worker_failed", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) finally: self._routing_stopped_event.set() + log.debug( + "Derive routing worker stopped", + extra={ + "event": "derive.routing_worker_stopped", + "thread_name": threading.current_thread().name, + }, + ) def _route_data(self, message: ValueDescriptor) -> None: with self._lock: diff --git a/src/dynalab_core/dlpak.py b/src/dynalab_core/dlpak.py index be27118..cb79622 100644 --- a/src/dynalab_core/dlpak.py +++ b/src/dynalab_core/dlpak.py @@ -3,6 +3,7 @@ # SPDX-License-Identifier: GPL-3.0-or-later from datetime import datetime, timezone +import logging from pathlib import Path from typing import Literal from zipfile import ZIP_DEFLATED, ZipFile @@ -13,6 +14,8 @@ from dynalab_core.buffer import ValueBuffer from dynalab_core.errors import DLPakNoDataError from dynalab_core.protocols.packets.handshake import SignalDescriptor +log = logging.getLogger(__name__) + class RecordManifest(BaseModel): timestamp: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) @@ -62,6 +65,31 @@ class DLPak: output_path = output_dir / file_path - with ZipFile(output_path, mode="w", compression=ZIP_DEFLATED) as archive: - archive.writestr("manifest.json", self._manifest.model_dump_json(indent=2)) - archive.writestr("data.csv", self._data.export_csv()) + try: + with ZipFile(output_path, mode="w", compression=ZIP_DEFLATED) as archive: + archive.writestr( + "manifest.json", self._manifest.model_dump_json(indent=2) + ) + archive.writestr("data.csv", self._data.export_csv()) + except Exception as error: + log.exception( + "Failed to write DLPak archive to %s", + output_path.name, + extra={ + "event": "dlpak.write_failed", + "output_filename": output_path.name, + "exception_type": type(error).__name__, + }, + ) + raise + + log.info( + "Wrote DLPak archive to %s", + output_path.name, + extra={ + "event": "dlpak.written", + "output_filename": output_path.name, + "sample_count": len(self._data), + "signal_count": len(self._manifest.signals), + }, + ) diff --git a/src/dynalab_core/errors.py b/src/dynalab_core/errors.py index 89c0955..008e6ad 100644 --- a/src/dynalab_core/errors.py +++ b/src/dynalab_core/errors.py @@ -7,6 +7,14 @@ class CoreError(Exception): """DynaLab Core error.""" +class CoreModeIncorrectError(CoreError): + """DynaLab Core mode error.""" + + +class CoreModeSwitchImpossibleError(CoreError): + """DynaLab Core mode error.""" + + class CoreStateMismatchError(CoreError): """DynaLab Core state error.""" diff --git a/src/dynalab_core/playback.py b/src/dynalab_core/playback.py new file mode 100644 index 0000000..7ace54f --- /dev/null +++ b/src/dynalab_core/playback.py @@ -0,0 +1,8 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + + +class PlaybackEngine: + def __init__(self) -> None: + pass diff --git a/src/dynalab_core/protocols/endpoint.py b/src/dynalab_core/protocols/endpoint.py index 17e4026..64d4079 100644 --- a/src/dynalab_core/protocols/endpoint.py +++ b/src/dynalab_core/protocols/endpoint.py @@ -5,10 +5,11 @@ import logging from queue import Empty, Full, Queue import threading -from threading import RLock, Thread -from time import monotonic, monotonic_ns, sleep +from threading import Lock, RLock, Thread +from time import monotonic, sleep from uuid import UUID +from dynalab_core.types import CoreModeState from dynalab_core.constants import HELLO_PACKET, INTERNAL_CONNECTOR_HELLO from dynalab_core.protocols.errors import ( ConnectorEndpointQueueFullError, @@ -34,6 +35,7 @@ class ConnectorEndpoint: timeout_event: threading.Event, reject_event: threading.Event, core_input_queue: Queue, + core_mode: CoreModeState, ) -> None: # external IO queues self._packet_ingress_queue: Queue[ProtocolMessage] = Queue(524288) @@ -43,10 +45,14 @@ class ConnectorEndpoint: self._heartbeat_ingress_queue: Queue[ProtocolMessage] = Queue(524288) self._heartbeat_egress_queue: Queue[ProtocolMessage] = Queue(524288) + self._core_mode = core_mode + self._connector_hello = hello self._timed_out_event = timeout_event self._reject_event = reject_event self._stop_event = threading.Event() + self._queue_log_lock = Lock() + self._queue_log_state: dict[str, tuple[float, int]] = {} self._input_worker_stopped_event = threading.Event() self._input_worker_thread = Thread( target=self._input_worker, @@ -65,10 +71,12 @@ class ConnectorEndpoint: name=f"endpoint_heartbeat_worker_{self._connector_hello.connector_uuid}", daemon=True, ) + + def start(self) -> None: self._input_worker_thread.start() self._output_worker_thread.start() - def stop(self) -> None: + def stop(self) -> bool: connector_uuid = str(self.uuid()) log.debug( "Stopping connector endpoint %s", @@ -94,20 +102,22 @@ class ConnectorEndpoint: or not self._heartbeat_worker_stopped_event.is_set() ): log.error( - "Connector endpoint %s did not stop within 10 seconds", + "Connector endpoint %s did not stop within 3 seconds", connector_uuid, extra={ "event": "endpoint.stop_timeout", "connector_uuid": connector_uuid, - "timeout_s": 10, + "timeout_s": 3, }, ) + return False else: log.debug( "Connector endpoint %s stopped", connector_uuid, extra={"event": "endpoint.stopped", "connector_uuid": connector_uuid}, ) + return True def uuid(self) -> UUID: return self._connector_hello.connector_uuid @@ -126,7 +136,9 @@ class ConnectorEndpoint: try: self._packet_ingress_queue.put_nowait(packet) except Full: - self._log_queue_full("ingress", self._packet_ingress_queue) + self._log_queue_full( + "ingress", self._packet_ingress_queue, action="rejected" + ) raise ConnectorEndpointQueueFullError def get_egress_packet(self, timeout: float | None) -> ProtocolMessage: @@ -145,10 +157,26 @@ class ConnectorEndpoint: try: self._packet_egress_queue.put_nowait(packet) except Full: - self._log_queue_full("egress", self._packet_egress_queue) + self._log_queue_full("egress", self._packet_egress_queue, action="rejected") raise ConnectorEndpointQueueFullError - def _log_queue_full(self, direction: str, queue: Queue[ProtocolMessage]) -> None: + def _log_queue_full( + self, + direction: str, + queue: Queue[ProtocolMessage], + message_type: str | None = None, + action: str = "dropped", + ) -> None: + now = monotonic() + with self._queue_log_lock: + last_logged, suppressed_count = self._queue_log_state.get( + direction, (0.0, 0) + ) + if now - last_logged < 1.0: + self._queue_log_state[direction] = (last_logged, suppressed_count + 1) + return + self._queue_log_state[direction] = (now, 0) + connector_uuid = str(self.uuid()) log.warning( "Connector %s %s queue is full (%d/%d)", @@ -162,12 +190,16 @@ class ConnectorEndpoint: "queue_direction": direction, "queue_size": queue.qsize(), "queue_capacity": queue.maxsize, + "message_type": message_type, + "action": action, + "suppressed_count": suppressed_count, }, ) def _heartbeat_worker(self) -> None: + connector_uuid = str(self.uuid()) last_send: int = 0 - last_recieved: int = 0 + last_received: int = 0 first_received: bool = False ctr: int = 0 try: @@ -180,8 +212,13 @@ class ConnectorEndpoint: ) last_send = now ctr += 1 - log.debug("Sent heartbeat") except Full: + self._log_queue_full( + "heartbeat_egress", + self._heartbeat_egress_queue, + "Heartbeat", + action="retrying", + ) sleep(0.01) try: @@ -192,19 +229,57 @@ class ConnectorEndpoint: if isinstance(received, Heartbeat): first_received = True if received.return_timestamp is not None: - last_recieved = received.return_timestamp - log.debug("Received heartbeat") + last_received = received.return_timestamp if not first_received: if ctr > round( HELLO_PACKET.heartbeat_timeout_ms / HELLO_PACKET.heartbeat_interval_ms ): + log.warning( + "Connector %s timed out before its first heartbeat response", + connector_uuid, + extra={ + "event": "endpoint.heartbeat_timeout", + "connector_uuid": connector_uuid, + "heartbeat_interval_ms": HELLO_PACKET.heartbeat_interval_ms, + "heartbeat_timeout_ms": HELLO_PACKET.heartbeat_timeout_ms, + "initial_heartbeat_received": False, + "last_sent_ms": last_send, + "last_received_ms": None, + }, + ) self._timed_out_event.set() return - elif last_send > last_recieved + HELLO_PACKET.heartbeat_timeout_ms: + elif last_send > last_received + HELLO_PACKET.heartbeat_timeout_ms: + log.warning( + "Connector %s heartbeat timed out", + connector_uuid, + extra={ + "event": "endpoint.heartbeat_timeout", + "connector_uuid": connector_uuid, + "heartbeat_interval_ms": HELLO_PACKET.heartbeat_interval_ms, + "heartbeat_timeout_ms": HELLO_PACKET.heartbeat_timeout_ms, + "initial_heartbeat_received": True, + "last_sent_ms": last_send, + "last_received_ms": last_received, + }, + ) self._timed_out_event.set() return + except Exception as error: + log.exception( + "Connector %s heartbeat worker failed", + connector_uuid, + extra={ + "event": "endpoint.worker_failed", + "connector_uuid": connector_uuid, + "worker_role": "heartbeat", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) + self._reject_event.set() finally: self._heartbeat_worker_stopped_event.set() @@ -219,55 +294,101 @@ class ConnectorEndpoint: "thread_name": self._input_worker_thread.name, }, ) - last_log = 0 - while not self._stop_event.is_set(): - now = monotonic_ns() - messages: list[ProtocolMessage] = [] - try: - while not self._packet_ingress_queue.empty(): - messages.append(self._get_ingress_packet_no_wait()) - except Empty: - continue - else: - for message in messages: - if isinstance(message, Heartbeat): - self._heartbeat_ingress_queue.put_nowait(message) - elif isinstance(message, ValueDescriptor): - if message.signal_id in ( - signal.id for signal in self._connector_hello.signals + try: + while not self._stop_event.is_set(): + messages: list[ProtocolMessage] = [] + try: + while not self._packet_ingress_queue.empty(): + messages.append(self._get_ingress_packet_no_wait()) + except Empty: + continue + else: + if not messages: + self._stop_event.wait(0.01) + continue + for message in messages: + if isinstance(message, Heartbeat): + try: + self._heartbeat_ingress_queue.put_nowait(message) + except Full: + self._log_queue_full( + "heartbeat_ingress", + self._heartbeat_ingress_queue, + type(message).__name__, + ) + elif ( + isinstance(message, ValueDescriptor) + and self._core_mode.value == "realtime" ): - self._core_input_queue.put_nowait(message) - elif isinstance(message, ValueBatch): - for value in message.values: - if value.signal_id in ( + if message.signal_id in ( signal.id for signal in self._connector_hello.signals ): - self._core_input_queue.put_nowait(value) - - if now > last_log + 2000 * 1_000_000: - log.debug(f"Endpoint queue size: {self._packet_ingress_queue.qsize()}") - last_log = now - 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, - }, - ) + try: + self._core_input_queue.put_nowait(message) + except Full: + self._log_queue_full( + "core_input", + self._core_input_queue, + type(message).__name__, + action="disconnecting", + ) + self._reject_event.set() + return + elif ( + isinstance(message, ValueBatch) + and self._core_mode.value == "realtime" + ): + for value in message.values: + if value.signal_id in ( + signal.id + for signal in self._connector_hello.signals + ): + try: + self._core_input_queue.put_nowait(value) + except Full: + self._log_queue_full( + "core_input", + self._core_input_queue, + type(value).__name__, + action="disconnecting", + ) + self._reject_event.set() + return + except Exception as error: + log.exception( + "Connector %s input worker failed", + connector_uuid, + extra={ + "event": "endpoint.worker_failed", + "connector_uuid": connector_uuid, + "worker_role": "input", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) + self._reject_event.set() + finally: + 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()) try: - connector_uuid = str(self.uuid()) log.debug( "Connector endpoint output worker %s started", - self._input_worker_thread.name, + self._output_worker_thread.name, extra={ "event": "endpoint.output_worker_started", "connector_uuid": connector_uuid, - "thread_name": self._input_worker_thread.name, + "thread_name": self._output_worker_thread.name, }, ) @@ -278,7 +399,17 @@ class ConnectorEndpoint: self._put_egress_packet( HandshakeRejected(reason="Protocol versions mismatch") ) - log.debug(f"Rejected handshake for endpoint {self.uuid()}") + log.warning( + "Rejected connector %s due to protocol version mismatch", + connector_uuid, + extra={ + "event": "endpoint.handshake_rejected", + "connector_uuid": connector_uuid, + "reason": "protocol_version_mismatch", + "protocol_version": self._connector_hello.protocol_version.get_version(), + "signal_count": len(self._connector_hello.signals), + }, + ) self._reject_event.set() return @@ -286,13 +417,30 @@ class ConnectorEndpoint: self._put_egress_packet( HandshakeRejected(reason="No signals available") ) - log.debug(f"Rejected handshake for endpoint {self.uuid()}") + log.warning( + "Rejected connector %s because it exposes no signals", + connector_uuid, + extra={ + "event": "endpoint.handshake_rejected", + "connector_uuid": connector_uuid, + "reason": "no_signals", + "protocol_version": self._connector_hello.protocol_version.get_version(), + "signal_count": 0, + }, + ) self._reject_event.set() return self._put_egress_packet(HandshakeAccepted(accepted_signals=[])) - log.debug(f"Accepted handshake for endpoint {self.uuid()}") - log.debug(f"Accepted signals: {self._connector_hello.signals}") + log.debug( + "Accepted handshake for connector %s", + connector_uuid, + extra={ + "event": "endpoint.handshake_accepted", + "connector_uuid": connector_uuid, + "advertised_signal_count": len(self._connector_hello.signals), + }, + ) self._heartbeat_worker_thread.start() @@ -305,26 +453,40 @@ class ConnectorEndpoint: else: self._put_egress_packet(message) + except Exception as error: + log.exception( + "Connector %s output worker failed", + connector_uuid, + extra={ + "event": "endpoint.worker_failed", + "connector_uuid": connector_uuid, + "worker_role": "output", + "thread_name": threading.current_thread().name, + "exception_type": type(error).__name__, + }, + ) + self._reject_event.set() finally: self._output_worker_stopped_event.set() log.debug( "Connector endpoint output worker %s stopped", - self._input_worker_thread.name, + self._output_worker_thread.name, extra={ "event": "endpoint.output_worker_stopped", "connector_uuid": connector_uuid, - "thread_name": self._input_worker_thread.name, + "thread_name": self._output_worker_thread.name, }, ) class ConnectorRegistry: - def __init__(self, core_input_queue: Queue) -> None: + def __init__(self, core_input_queue: Queue, core_mode: CoreModeState) -> None: self._endpoints: dict[UUID, ConnectorEndpoint] = {} self._internal_connector: ConnectorHello = INTERNAL_CONNECTOR_HELLO self._lock = RLock() self._core_input_queue = core_input_queue + self._core_mode = core_mode def add_internal_signal(self, signal: SignalDescriptor) -> None: with self._lock: @@ -342,10 +504,6 @@ class ConnectorRegistry: timeout_event: threading.Event, reject_event: threading.Event, ) -> ConnectorEndpoint: - endpoint = ConnectorEndpoint( - hello, timeout_event, reject_event, self._core_input_queue - ) - with self._lock: current = self._endpoints.get(hello.connector_uuid) @@ -361,9 +519,35 @@ class ConnectorRegistry: ) raise ConnectorRegistryAlreadyRegisteredError + endpoint = ConnectorEndpoint( + hello, + timeout_event, + reject_event, + self._core_input_queue, + self._core_mode, + ) + self._endpoints[hello.connector_uuid] = endpoint endpoint_count = len(self._endpoints) + try: + endpoint.start() + except Exception as error: + endpoint.stop() + with self._lock: + if self._endpoints.get(hello.connector_uuid) is endpoint: + del self._endpoints[hello.connector_uuid] + log.exception( + "Failed to start connector endpoint %s", + hello.connector_uuid, + extra={ + "event": "connector.registration_failed", + "connector_uuid": str(hello.connector_uuid), + "exception_type": type(error).__name__, + }, + ) + raise + log.info( "Registered connector %s (%s)", hello.connector_name, @@ -380,13 +564,22 @@ class ConnectorRegistry: return endpoint - def unregister(self, endpoint: ConnectorEndpoint) -> None: + def unregister(self, endpoint: ConnectorEndpoint) -> bool: connector_uuid = endpoint.uuid() with self._lock: current = self._endpoints.get(connector_uuid) if current is endpoint: - endpoint.stop() + if not endpoint.stop(): + log.error( + "Unable to unregister connector %s because its endpoint did not stop", + connector_uuid, + extra={ + "event": "connector.unregister_incomplete", + "connector_uuid": str(connector_uuid), + }, + ) + return False del self._endpoints[connector_uuid] endpoint_count = len(self._endpoints) else: @@ -398,7 +591,7 @@ class ConnectorRegistry: "connector_uuid": str(connector_uuid), }, ) - return + return True log.info( "Unregistered connector %s", @@ -409,6 +602,7 @@ class ConnectorRegistry: "endpoint_count": endpoint_count, }, ) + return True def get(self, connector_uuid: UUID) -> ConnectorEndpoint | None: with self._lock: @@ -451,7 +645,7 @@ class ConnectorRegistry: return signals - def stop(self) -> None: + def stop(self) -> bool: with self._lock: endpoints = tuple(self._endpoints.values()) self._endpoints.clear() @@ -464,12 +658,22 @@ class ConnectorRegistry: "endpoint_count": len(endpoints), }, ) - for endpoint in endpoints: - endpoint.stop() - log.debug( - "Connector registry stopped", - extra={ - "event": "connector_registry.stopped", - "endpoint_count": 0, - }, - ) + endpoint_results = [endpoint.stop() for endpoint in endpoints] + stopped = all(endpoint_results) + if stopped: + log.debug( + "Connector registry stopped", + extra={ + "event": "connector_registry.stopped", + "endpoint_count": 0, + }, + ) + else: + log.error( + "Connector registry stop incomplete", + extra={ + "event": "connector_registry.stop_incomplete", + "failed_endpoint_count": endpoint_results.count(False), + }, + ) + return stopped diff --git a/src/dynalab_core/protocols/json/server.py b/src/dynalab_core/protocols/json/server.py index a5c84b1..7ad120b 100644 --- a/src/dynalab_core/protocols/json/server.py +++ b/src/dynalab_core/protocols/json/server.py @@ -6,10 +6,9 @@ import asyncio from asyncio import Server import logging from queue import Empty -from statistics import mean import threading from threading import Thread -from time import monotonic, monotonic_ns +from time import monotonic from uuid import uuid4 @@ -22,7 +21,7 @@ from dynalab_core.protocols.json.errors import ( JsonServerTimeoutError, ) from dynalab_core.protocols.json.wire import read_message, write_message -from dynalab_core.protocols.packets.handshake import ConnectorHello +from dynalab_core.protocols.packets.handshake import ConnectorHello, HandshakeAccepted log = logging.getLogger(__name__) @@ -88,7 +87,7 @@ class JsonServer: }, ) - def stop(self) -> None: + def stop(self) -> bool: log.info("Stopping JSON server", extra={"event": "json_server.stopping"}) self._stop_event.set() if not self._stopped_event.wait(10): @@ -96,8 +95,10 @@ class JsonServer: "JSON server did not stop within 10 seconds", extra={"event": "json_server.stop_timeout", "timeout_s": 10}, ) + return False else: log.info("JSON server stopped", extra={"event": "json_server.stopped"}) + return True def _json_server_thread_main(self) -> None: log.debug( @@ -109,6 +110,22 @@ class JsonServer: ) try: asyncio.run(self._run_json_server()) + except Exception as error: + startup_complete = self._server is not None + log.exception( + "JSON server thread failed", + extra={ + "event": "json_server.thread_failed", + "thread_name": threading.current_thread().name, + "host": self._config.host, + "port": self._config.port, + "startup_complete": startup_complete, + "exception_type": type(error).__name__, + }, + ) + if not startup_complete: + self._startup_error = error + self._started_event.set() finally: log.debug( "JSON server thread stopped", @@ -163,27 +180,34 @@ class JsonServer: reader: asyncio.StreamReader, endpoint: ConnectorEndpoint, ) -> None: - log.debug(f"Started input task for {endpoint.uuid()}") - intervals: list[int] = [] - last_val: int = 0 - last_print: int = 0 + connector_uuid = str(endpoint.uuid()) + log.debug( + "Started input task for connector %s", + connector_uuid, + extra={ + "event": "json_connection.input_task_started", + "connector_uuid": connector_uuid, + }, + ) while not self._stop_event.is_set(): message = await read_message(reader) endpoint.put_ingress_packet(message) - now = monotonic_ns() - intervals.append(now - last_val) - last_val = now - if now > last_print + 2000 * 1_000_000: - log.debug(f"Server input frequency: {1_000_000 / mean(intervals)}kHz") - last_print = now - intervals = [] async def _output_task( self, writer: asyncio.StreamWriter, endpoint: ConnectorEndpoint, + connection_context: dict[str, str], ) -> None: - log.debug(f"Started output task for {endpoint.uuid()}") + connector_uuid = str(endpoint.uuid()) + log.debug( + "Started output task for connector %s", + connector_uuid, + extra={ + "event": "json_connection.output_task_started", + "connector_uuid": connector_uuid, + }, + ) while not self._stop_event.is_set(): try: message = endpoint.get_egress_packet_no_wait() @@ -191,6 +215,17 @@ class JsonServer: await asyncio.sleep(0.01) continue await write_message(writer, message) + if isinstance(message, HandshakeAccepted): + log.info( + "Connector %s connected from %s", + connector_uuid, + connection_context["peer_address"], + extra={ + "event": "connector.connected", + "connector_uuid": connector_uuid, + **connection_context, + }, + ) def _start_connection_handler( self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter @@ -209,6 +244,10 @@ class JsonServer: if error is not None: log.error( "Connection handler terminated with an exception", + extra={ + "event": "json_connection.task_failed", + "exception_type": type(error).__name__, + }, exc_info=(type(error), error, error.__traceback__), ) @@ -219,6 +258,7 @@ class JsonServer: peer_address = str(writer.get_extra_info("peername")) started_at = monotonic() reason = "handshake_incomplete" + phase = "write_server_hello" connector_endpoint: ConnectorEndpoint | None = None connection_tasks: list[asyncio.Task[None]] = [] @@ -250,6 +290,7 @@ class JsonServer: ) return + phase = "read_connector_hello" try: message = await asyncio.wait_for(read_message(reader), timeout=5.0) except TimeoutError: @@ -282,6 +323,7 @@ class JsonServer: return connector_hello = message + context["connector_uuid"] = str(connector_hello.connector_uuid) timeout_event = threading.Event() reject_event = threading.Event() connector_endpoint = self._connector_registry.register( @@ -289,24 +331,12 @@ class JsonServer: ) 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, - }, - ) - + phase = "streaming" input_task = asyncio.create_task( self._input_task(reader, connector_endpoint) ) output_task = asyncio.create_task( - self._output_task(writer, connector_endpoint) + self._output_task(writer, connector_endpoint, context) ) server_stop_task = asyncio.create_task( self._wait_for_thread_event(self._stop_event) @@ -326,9 +356,9 @@ class JsonServer: done, _ = await asyncio.wait( connection_tasks, return_when=asyncio.FIRST_COMPLETED ) - if timeout_task in done: + if timeout_event.is_set(): reason = "endpoint_timeout" - elif reject_event in done: + elif reject_event.is_set(): reason = "connection_rejected" elif server_stop_task in done: reason = "server_shutdown" @@ -348,11 +378,12 @@ class JsonServer: except ConnectionError as error: reason = "peer_disconnected" log.debug( - "Connection %s disconnected: %s", + "Connection %s disconnected during %s", connection_id, - error, + phase, extra={ - "event": "connector.handshake_disconnected", + "event": "json_connection.disconnected", + "phase": phase, "exception_type": type(error).__name__, **context, }, @@ -360,12 +391,16 @@ class JsonServer: except ValueError as error: reason = "invalid_frame" log.warning( - "Connection %s sent an invalid handshake frame: %s", + "Connection %s sent an invalid frame during %s", connection_id, - error, + phase, extra={ - "event": "connector.handshake_rejected", + "event": "connector.handshake_rejected" + if phase != "streaming" + else "connector.frame_rejected", "reason": reason, + "phase": phase, + "exception_type": type(error).__name__, **context, }, ) @@ -380,14 +415,18 @@ class JsonServer: **context, }, ) - except Exception: + except Exception as error: reason = "handler_failed" log.exception( "Connection %s handler failed", connection_id, - extra={"event": "json_connection.failed", **context}, + extra={ + "event": "json_connection.failed", + "phase": phase, + "exception_type": type(error).__name__, + **context, + }, ) - raise finally: for task in connection_tasks: task.cancel() @@ -397,8 +436,16 @@ class JsonServer: writer.close() try: await writer.wait_closed() - except ConnectionError: - pass + except ConnectionError as error: + log.debug( + "Connection %s was already closed during cleanup", + connection_id, + extra={ + "event": "json_connection.cleanup_disconnected", + "exception_type": type(error).__name__, + **context, + }, + ) if connector_endpoint is not None: await asyncio.to_thread( diff --git a/src/dynalab_core/types.py b/src/dynalab_core/types.py new file mode 100644 index 0000000..7812ffa --- /dev/null +++ b/src/dynalab_core/types.py @@ -0,0 +1,15 @@ +# Copyright (C) 2026 Hector van der Aa +# Copyright (C) 2026 Association Exergie +# SPDX-License-Identifier: GPL-3.0-or-later + + +from dataclasses import dataclass +from typing import Literal + + +CoreMode = Literal["realtime", "playback"] + + +@dataclass +class CoreModeState: + value: CoreMode = "realtime" diff --git a/test/manual/core.py b/test/manual/core.py index 7bad49f..9af786e 100644 --- a/test/manual/core.py +++ b/test/manual/core.py @@ -30,13 +30,6 @@ dl_core = Core(config) dl_core.start() -dl_core.start_recording() - -for i in range(2): - dl_core.wait(1) - values = dl_core.get_all_live_values() - log.debug(f"Core values: {values}") - unit_id = dl_core.bind_derive_unit( processing_str, [ @@ -50,6 +43,13 @@ unit_id = dl_core.bind_derive_unit( SignalDescriptor(id=uuid4(), name="Dummy sum", type="number", timeout_ms=5000), ) +for i in range(10): + dl_core.wait(1) + values = dl_core.get_all_live_values() + log.debug(f"Core values: {values}") + +dl_core.set_mode("playback") + try: while True: @@ -58,6 +58,4 @@ try: log.debug(f"Core values: {values}") except KeyboardInterrupt: log.info("Received keyboard interrupt") - dl_core.stop_recording() - dl_core._processing_buffer.write("./", "output.dlpak") dl_core.stop() diff --git a/test/test_core.py b/test/test_core.py index 9649d8d..7a67a73 100644 --- a/test/test_core.py +++ b/test/test_core.py @@ -1,9 +1,14 @@ import logging +import threading +from uuid import uuid4 import pytest from dynalab_core import Core from dynalab_core.config import CoreConfig from dynalab_core.errors import CoreStateMismatchError +from dynalab_core.protocols.endpoint import ConnectorEndpoint +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import ConnectorHello from test.common import find_available_port @@ -32,3 +37,59 @@ def test_core_cannot_start_twice(caplog: pytest.LogCaptureFixture) -> None: ) assert rejection.levelno == logging.WARNING assert rejection.core_state == "started" + + +def test_core_input_worker_logs_failure( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + core = Core(CoreConfig(port=find_available_port(8765))) + + def fail_routing(message: ValueDescriptor) -> None: + raise RuntimeError("routing failed") + + monkeypatch.setattr(core._derive_registry, "put_data", fail_routing) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + core.start() + try: + core._data_input_queue.put_nowait( + ValueDescriptor(signal_id=uuid4(), value=1.0, timestamp=1) + ) + assert core._input_worker_stopped_event.wait(1.0) + finally: + core.stop() + + failures = [ + record + for record in caplog.records + if getattr(record, "event", None) == "core.input_worker_failed" + ] + assert len(failures) == 1 + assert failures[0].exception_type == "RuntimeError" + assert failures[0].message_type == "ValueDescriptor" + assert failures[0].exc_info is not None + + +def test_core_mode_is_shared_with_endpoints() -> None: + core = Core(CoreConfig(port=find_available_port(8765))) + endpoint = ConnectorEndpoint( + ConnectorHello( + connector_uuid=uuid4(), + connector_name="test", + connector_version="1.0", + signals=[], + ), + threading.Event(), + threading.Event(), + core._data_input_queue, + core._connector_registry._core_mode, + ) + + try: + assert endpoint._core_mode is core._mode + + core.set_mode("playback") + + assert endpoint._core_mode.value == "playback" + finally: + core.stop() diff --git a/test/test_derive.py b/test/test_derive.py index 51a8244..05b5259 100644 --- a/test/test_derive.py +++ b/test/test_derive.py @@ -1,7 +1,9 @@ +import logging from queue import Empty, Queue import time from uuid import UUID, uuid4 +import pytest from dynalab_core.derive import DeriveRegistry from dynalab_core.protocols.packets.data import ValueDescriptor from dynalab_core.protocols.packets.handshake import SignalDescriptor @@ -19,6 +21,14 @@ def add(left: ValueDescriptor, right: ValueDescriptor, output: SignalDescriptor) ) """ +FAILING_PROCESSING_FUNCTION = """ +from dynalab_core.protocols.packets.data import ValueDescriptor +from dynalab_core.protocols.packets.handshake import SignalDescriptor + +def fail(value: ValueDescriptor, output: SignalDescriptor) -> ValueDescriptor: + raise RuntimeError("derive failed") +""" + def test_derive_registry_routes_with_live_values() -> None: output_queue: Queue[ValueDescriptor] = Queue() @@ -32,9 +42,7 @@ def test_derive_registry_routes_with_live_values() -> None: unit = registry.register( PROCESSING_FUNCTION, [left_signal, right_signal], output_signal ) - right_value = ValueDescriptor( - signal_id=right_signal.id, value=2.0, timestamp=1 - ) + right_value = ValueDescriptor(signal_id=right_signal.id, value=2.0, timestamp=1) live_values[right_signal.id] = right_value left_value = ValueDescriptor(signal_id=left_signal.id, value=3.0, timestamp=2) @@ -59,6 +67,7 @@ def test_derive_registry_routes_with_live_values() -> None: assert result.value == 5.0 assert result.timestamp == left_value.timestamp + def test_derive_registry_ignores_unknown_unit_on_unregister() -> None: registry = DeriveRegistry(Queue(), lambda signal_id: None) @@ -66,3 +75,34 @@ def test_derive_registry_ignores_unknown_unit_on_unregister() -> None: registry.unregister(uuid4()) finally: registry.stop() + + +def test_derive_worker_logs_failure_and_stops( + caplog: pytest.LogCaptureFixture, +) -> None: + input_signal = SignalDescriptor(id=uuid4(), name="Input", type="number") + output_signal = SignalDescriptor(id=uuid4(), name="Output", type="number") + registry = DeriveRegistry(Queue(), lambda signal_id: None) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + try: + unit = registry.register( + FAILING_PROCESSING_FUNCTION, [input_signal], output_signal + ) + unit.put_data( + [ValueDescriptor(signal_id=input_signal.id, value=1.0, timestamp=1)] + ) + assert unit._stopped_event.wait(1.0) + assert registry.get_all_signal_descriptors() == [] + finally: + registry.stop() + + failures = [ + record + for record in caplog.records + if getattr(record, "event", None) == "derive.worker_failed" + ] + assert len(failures) == 1 + assert failures[0].derive_uuid == str(unit.uuid()) + assert failures[0].exception_type == "RuntimeError" + assert failures[0].exc_info is not None diff --git a/test/test_dlpak.py b/test/test_dlpak.py new file mode 100644 index 0000000..7577acc --- /dev/null +++ b/test/test_dlpak.py @@ -0,0 +1,29 @@ +import logging +from datetime import datetime, timezone +from uuid import uuid4 + +import pytest +from dynalab_core.buffer import ValueBuffer +from dynalab_core.dlpak import DLPak +from dynalab_core.protocols.packets.handshake import SignalDescriptor + + +def test_dlpak_logs_written_archive(caplog: pytest.LogCaptureFixture, tmp_path) -> None: + signal = SignalDescriptor(id=uuid4(), name="Signal", type="number") + buffer = ValueBuffer() + buffer.append(1, signal.id, 2.0) + package = DLPak() + package.set_data(buffer) + package.set_manifest(datetime.now(timezone.utc), [signal]) + + with caplog.at_level(logging.INFO, logger="dynalab_core"): + package.write(tmp_path, "recording") + + record = next( + record + for record in caplog.records + if getattr(record, "event", None) == "dlpak.written" + ) + assert record.output_filename == "recording.dlpak" + assert record.sample_count == 1 + assert record.signal_count == 1 diff --git a/test/test_json_server.py b/test/test_json_server.py index d762583..b020f89 100644 --- a/test/test_json_server.py +++ b/test/test_json_server.py @@ -1,3 +1,4 @@ +import asyncio import logging import socket import time @@ -7,6 +8,7 @@ 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.common import VersionDescriptor from dynalab_core.protocols.json.errors import ( JsonServerStartupError, JsonServerTimeoutError, @@ -91,7 +93,7 @@ def test_json_server_logs_invalid_handshake( 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") + peer.sendall(b'{"secret":"must-not-appear-in-logs"}\n') rejection = _wait_for_event(caplog, "connector.handshake_rejected") finally: @@ -101,6 +103,69 @@ def test_json_server_logs_invalid_handshake( assert rejection.reason == "invalid_frame" assert rejection.connection_id assert rejection.peer_address + assert "must-not-appear-in-logs" not in caplog.text + + +def test_json_server_reports_unexpected_thread_startup_failure( + caplog: pytest.LogCaptureFixture, monkeypatch: pytest.MonkeyPatch +) -> None: + async def fail_start_server(*args: object, **kwargs: object) -> None: + raise RuntimeError("unexpected startup failure") + + monkeypatch.setattr(asyncio, "start_server", fail_start_server) + core = Core(CoreConfig(port=find_available_port(8765))) + + with caplog.at_level(logging.DEBUG, logger="dynalab_core"): + try: + with pytest.raises(JsonServerStartupError): + core.start() + finally: + core.stop() + + failures = [ + record + for record in caplog.records + if getattr(record, "event", None) == "json_server.thread_failed" + ] + assert len(failures) == 1 + assert failures[0].startup_complete is False + assert failures[0].exception_type == "RuntimeError" + assert failures[0].exc_info is not None + + +def test_rejected_handshake_is_not_logged_as_connected( + 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=VersionDescriptor(type="alpha", major=999, minor=0, patch=0), + connector_name="Incompatible connector", + connector_version="0.1.0-test", + signals=[SignalDescriptor(id=uuid4(), name="Dummy signal", type="number")], + ) + + 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_rejected"' in peer_file.readline() + rejection = _wait_for_event(caplog, "endpoint.handshake_rejected") + finally: + core.stop() + + assert rejection.reason == "protocol_version_mismatch" + assert rejection.connector_uuid == str(connector_uuid) + assert not any( + getattr(record, "event", None) == "connector.connected" + and getattr(record, "connector_uuid", None) == str(connector_uuid) + for record in caplog.records + ) def test_json_server_waits_for_connection_handlers_on_stop( @@ -133,6 +198,8 @@ def test_json_server_waits_for_connection_handlers_on_stop( events = [getattr(record, "event", None) for record in caplog.records] assert "json_server.stop_timeout" not in events + assert "connector.connected" in events + assert events.index("connector.registered") < events.index("connector.connected") assert core._json_server._stopped_event.is_set() assert not core._json_server._handler_tasks assert core._connector_registry.get(connector_uuid) is None