Ewoks events sqlite3 bugs

CHANGELOG.md

@@ -9,8 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
-- Add new fields `workflow_input_schema` and `workflow_output_schema` to the spec
+- Add new fields `workflow_input_schema` and `workflow_output_schema` to the spec.
+- `RedisEwoksEventHandler`: add `disconnect_on_error` and `timeout` argument.
 
+### Fixed
+
+- `Sqlite3EwoksEventHandler`: implement abstract method `_connected()`.
 
 ## [5.0.0] - 2026-07-01
 

doc/howtoguides/events.rst

@@ -33,7 +33,7 @@ Define an Ewoks event handler
 -----------------------------
 
 You can use any handler derived from `logging.Handler` as an Ewoks event handler. For example if you
-have a `Connection` class with a constructor that accepts two arguments, you can define an ewoks
+have a `MyConnection` class with a constructor that accepts two arguments, you can define an ewoks
 event handler as follows:
 
 .. code-block:: python
@@ -46,22 +46,38 @@ event handler as follows:
 
     class MyHandler(EwoksEventHandlerMixIn, ConnectionHandler):
 
-        def _connect(self, uri:str, param1:int, timeout=1) -> None:
-            self._connection = Connection(uri, param1)
-            self._fields = send_events.FIELDS
+        def __init__(
+            self,
+            uri:str,
+            param1: int,
+            disconnect_on_error: bool = False,
+        ):
+            super().__init__(disconnect_on_error=disconnect_on_error)
+            self._uri = uri
+            self._param1 = param1
+            self._connection = None
+
+        def _connect(self) -> None:
+            self._connection = MyConnection(self._uri, self._param1)
 
         def _disconnect(self) -> None:
             del self._connection
             self._connection = None
 
-        def _send_serialized_record(self, srecord):
-            self._connection.send(srecord)
+        def _connected(self) -> bool:
+            return self._connection is not None
 
         def _serialize_record(self, record):
+            """Convert a record to something that can be given to the connection."""
             return {
-                field: self.get_value(record, field, None) for field in self._fields
+                field: self.get_value(record, field, None) for field in self.FIELD_TYPES
             }
 
+        def _send_serialized_record(self, serialized_record):
+            """Send the output from `_serialize_record` to the connection."""
+            self._connection.send(serialized_record)
+
+
 `ConnectionHandler` is an abstract python logging handler which can be used for convenience.
 
 To use this example handler

pyproject.toml

@@ -23,7 +23,7 @@ dependencies = [
     "pyyaml >=5.1",
     "h5py >=2.8",
     "packaging",
-    "ewoksutils >=1.10.0",
+    "ewoksutils >=2.0.0rc1",
     "importlib_metadata;python_version < '3.9'",
     "pydantic >= 2",
     "typing_extensions;python_version < '3.11'",

src/ewokscore/events/handlers/sqlite3.py

@@ -4,5 +4,21 @@
 
 
 class Sqlite3EwoksEventHandler(EwoksEventHandlerMixIn, Sqlite3Handler):
-    def __init__(self, uri: str):
-        super().__init__(uri=uri, table="ewoks_events", field_types=self.FIELD_TYPES)
+    def __init__(
+        self, uri: str, timeout: float = 10, disconnect_on_error: bool = False
+    ):
+        """
+        :param uri: for example "file:/path/to/ewoks_events.db" or
+                    "file:///path/to/ewoks_events.db".
+        :param timeout: native sqlite3 busy timeout: the maximum time to wait
+                        for database locks to be released by other connections.
+                        An event is dropped when the timeout is reached.
+        :param disconnect_on_error: disconnect when emitting an event failed.
+        """
+        super().__init__(
+            uri=uri,
+            table="ewoks_events",
+            field_types=self.FIELD_TYPES,
+            timeout=timeout,
+            disconnect_on_error=disconnect_on_error,
+        )

src/ewokscore/tests/test_events.py

@@ -1,12 +1,16 @@
 import logging
+import threading
 from contextlib import contextmanager
 from logging.handlers import QueueHandler
 from queue import Empty
 from queue import Queue
 
 import pytest
+from ewoksutils import sqlite3_utils
+from ewoksutils.event_utils import FIELD_TYPES
 
 from .. import events
+from ..events import global_state
 
 
 @contextmanager
@@ -130,3 +134,58 @@ def test_workflow_event_error_log_level(caplog):
         get_event()
         assert len(caplog.records) == 1
         assert caplog.records[0].levelno == logging.ERROR
+
+
+def test_concurrent_send(tmp_path):
+    """Sending events concurrently is thread-safe, including the first
+    event which instantiates and registers the event handlers."""
+    uri = f"file:{tmp_path / 'ewoks_events.db'}"
+    handlers = [
+        {
+            "class": "ewokscore.events.handlers.Sqlite3EwoksEventHandler",
+            "arguments": [{"name": "uri", "value": uri}],
+        }
+    ]
+
+    nthreads = 8
+    nevents_per_thread = 10
+    barrier = threading.Barrier(nthreads, timeout=10)
+    exceptions = list()
+
+    def send_events(job_id):
+        execinfo = {
+            "job_id": job_id,
+            "host_name": None,
+            "user_name": None,
+            "process_id": None,
+            "workflow_id": None,
+            "handlers": handlers,
+        }
+        try:
+            # Send the first event from all threads at the same time
+            barrier.wait()
+            for _ in range(nevents_per_thread // 2):
+                events.send_workflow_event(execinfo=execinfo, event="start")
+                events.send_workflow_event(execinfo=execinfo, event="end")
+        except BaseException as ex:
+            exceptions.append(ex)
+
+    threads = [
+        threading.Thread(target=send_events, args=(str(job_id),))
+        for job_id in range(nthreads)
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=30)
+
+    try:
+        assert not exceptions
+        logger = logging.getLogger(global_state.EWOKS_EVENT_LOGGER_NAME)
+        assert len(logger.handlers) == 1
+    finally:
+        events.cleanup()
+
+    with sqlite3_utils.connect(uri, uri=True) as conn:
+        rows = list(sqlite3_utils.select(conn, "ewoks_events", field_types=FIELD_TYPES))
+    assert len(rows) == nthreads * nevents_per_thread