Exergie/dynalab-core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Prepared for playback engine

Browse Source
This commit is contained in:
h3cx
2026-09-11 23:45:27 +02:00
parent 19b356a94d
commit bbc5aac891
14 changed files with 955 additions and 189 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/dynalab_core/__init__.py
+149 -34
View File
@@ -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,
},
)
src/dynalab_core/buffer.py
+4
View File
@@ -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)
src/dynalab_core/derive.py
+163 -21
View File
@@ -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:
src/dynalab_core/dlpak.py
+31 -3
View File
@@ -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),
},
)
src/dynalab_core/errors.py
+8
View File
@@ -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."""
src/dynalab_core/playback.py
+8
View File
@@ -0,0 +1,8 @@
# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# SPDX-License-Identifier: GPL-3.0-or-later
class PlaybackEngine:
def __init__(self) -> None:
pass
src/dynalab_core/protocols/endpoint.py
+279 -75
View File
@@ -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
src/dynalab_core/protocols/json/server.py
+90 -43
View File
@@ -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(
src/dynalab_core/types.py
+15
View File
@@ -0,0 +1,15 @@
# Copyright (C) 2026 Hector van der Aa <hector@h3cx.dev>
# Copyright (C) 2026 Association Exergie <association.exergie@gmail.com>
# 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"
test/manual/core.py
+7 -9
View File
@@ -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()
test/test_core.py
+61
View File
@@ -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()
test/test_derive.py
+43 -3
View File
@@ -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
test/test_dlpak.py
+29
View File
@@ -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
test/test_json_server.py
+68 -1
View File
@@ -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