344 ewoks events sqlite3 bugs

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- `RedisEwoksEventHandler`: add `disconnect_on_error` and `timeout` argument.
+- `RedisEwoksEventReader`: add `timeout` argument and do not allow unknown arguments.
+- `Sqlite3EwoksEventReader`: add `timeout` argument and do not allow unknown arguments.
+
+### Fixed
+
+`EwoksEventReader`: log error when last attempt failed when polling for events.
+
 ## [1.6.0rc1] - 2026-07-13
 
 ### Added

src/ewoksjob/events/handlers/redis.py

@@ -16,22 +16,44 @@
 class RedisEwoksEventHandler(EwoksEventHandlerMixIn, ConnectionHandler):
     # TODO: https://redisql.redbeardlab.com/blog/python/using-redisql-with-python/
 
-    def __init__(self, url: str, ttl=None):
-        """An example url is "redis://localhost:10003?db=2"."""
+    def __init__(
+        self,
+        url: str,
+        ttl: Optional[int] = None,
+        timeout: float = 10,
+        disconnect_on_error: bool = False,
+    ):
+        """
+        :param url: for example "redis://localhost:10003?db=2".
+        :param ttl: time-to-live in seconds of the record keys.
+        :param timeout: native redis socket timeout: the maximum time to wait
+                        for connecting and for command replies.
+                        A record is dropped when the timeout is reached.
+        :param disconnect_on_error: disconnect when emitting a record failed.
+        """
+        super().__init__(disconnect_on_error=disconnect_on_error)
         self._redis_url = url
         self._ttl = ttl
-        super().__init__()
+        self._timeout = timeout
+        self._connection = None
 
-    def _connect(self, timeout=1) -> None:
+    def _connect(self) -> None:
         """This is called when no connection exists."""
         client_name = f"ewoks:writer:{socket.gethostname()}:{os.getpid()}"
         self._connection = redis.Redis.from_url(
-            self._redis_url, client_name=client_name
+            self._redis_url,
+            client_name=client_name,
+            socket_connect_timeout=self._timeout,
+            socket_timeout=self._timeout,
         )
 
     def _disconnect(self) -> None:
         """This is called when a connection exists and is connected."""
         self._connection.close()
+        self._connection = None
+
+    def _connected(self) -> bool:
+        return self._connection is not None
 
     def _serialize_record(self, record: logging.LogRecord) -> Optional[RedisRecordType]:
         """Convert a record to something that can be given to the connection."""

src/ewoksjob/events/readers/base.py

@@ -1,4 +1,5 @@
 import json
+import logging
 import time
 from datetime import datetime
 from threading import Event
@@ -15,6 +16,7 @@
 except ImportError:
     Variable = VariableContainer = None
 
+logger = logging.getLogger(__name__)
 
 EventType = Dict[str, str]
 
@@ -54,20 +56,32 @@ def poll_events(
             interval = max(interval, 0.001)
         start = time.time()
         n = 0
+        exception = None
         while True:
             try:
                 events = list(self.get_events(**filters))
-            except Exception as e:
-                if "no such table" not in str(e):
+            except Exception as ex:
+                if not self._retry_get_events_exception(ex):
                     raise
+                if exception is None:
+                    logger.warning("Reading events failed (keep retrying): %s", ex)
+                exception = ex
             else:
+                exception = None
                 events = events[n:]
                 n += len(events)
                 yield from events
-            if timeout is not None and (time.time() - start) > timeout:
-                return
-            if stop_event is not None and stop_event.is_set():
+
+            timed_out = timeout is not None and (time.time() - start) > timeout
+            stopped = stop_event is not None and stop_event.is_set()
+            if timed_out or stopped:
+                if exception is not None:
+                    logger.error(
+                        "Stop polling events (last attempt failed)",
+                        exc_info=exception,
+                    )
                 return
+
             time.sleep(interval)
 
     def get_events(self, **filters) -> Iterator[EventType]:
@@ -168,3 +182,7 @@ def split_filter(
             raise TypeError("starttime needs to be a datetime object")
         post_filter = {"starttime": starttime, "endtime": endtime}
         return is_equal_filter, post_filter
+
+    @staticmethod
+    def _retry_get_events_exception(ex: Exception) -> bool:
+        return True

src/ewoksjob/events/readers/redis.py

@@ -10,10 +10,20 @@
 
 
 class RedisEwoksEventReader(EwoksEventReader):
-    def __init__(self, url: str, **_):
-        client_name = f"ewoks:reader:{socket.gethostname()}:{os.getpid()}"
-        self._proxy = redis.Redis.from_url(url, client_name=client_name)
+    def __init__(self, url: str, timeout: float = 10):
+        """
+        :param url: for example "redis://localhost:10003?db=2".
+        :param timeout: native redis socket timeout: the maximum time to wait
+                        for connecting and for command replies.
+        """
         super().__init__()
+        client_name = f"ewoks:reader:{socket.gethostname()}:{os.getpid()}"
+        self._proxy = redis.Redis.from_url(
+            url,
+            client_name=client_name,
+            socket_connect_timeout=timeout,
+            socket_timeout=timeout,
+        )
 
     def wait_events(self, **kwargs) -> Iterator[EventType]:
         yield from self.poll_events(**kwargs)

src/ewoksjob/events/readers/sqlite3.py

@@ -9,9 +9,15 @@
 
 
 class Sqlite3EwoksEventReader(EwoksEventReader):
-    def __init__(self, uri: str, **_) -> None:
+    def __init__(self, uri: str, timeout: float = 10) -> None:
+        """
+        :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.
+        """
         super().__init__()
         self._uri = uri
+        self._timeout = timeout
         self.__connection = None
         self.__sql_types = sqlite3_utils.python_to_sql_types(FIELD_TYPES)
 
@@ -24,7 +30,9 @@ def close(self):
     @property
     def _connection(self):
         if self.__connection is None:
-            self.__connection = sqlite3.connect(self._uri, uri=True)
+            self.__connection = sqlite3.connect(
+                self._uri, uri=True, timeout=self._timeout
+            )
         return self.__connection
 
     def wait_events(self, **kwargs) -> Iterator[EventType]:
@@ -38,3 +46,16 @@ def get_events(self, **filters) -> Iterator[EventType]:
             sql_types=self.__sql_types,
             **filters,
         )
+
+    @staticmethod
+    def _retry_get_events_exception(ex: Exception) -> bool:
+        if not isinstance(ex, sqlite3.OperationalError):
+            return False
+        error_message = str(ex)
+        if "no such table" in error_message:
+            # no event was published yet.
+            return True
+        if "database is locked" in error_message:
+            # transient lock by another connection.
+            return True
+        return False

src/ewoksjob/tests/test_events_sqlite3.py

@@ -0,0 +1,182 @@
+import sqlite3
+import threading
+import time
+
+from ewokscore import events
+
+from ..events.readers.sqlite3 import Sqlite3EwoksEventReader
+
+
+def test_concurrent_read_write(sqlite3_ewoks_events):
+    """Events published while a reader is polling must all be received."""
+    handlers, reader = sqlite3_ewoks_events
+
+    njobs = 4
+    nevents_per_job = 20
+
+    def publish_events(job_id):
+        execinfo = _execinfo(handlers, job_id=job_id)
+        for _ in range(nevents_per_job // 2):
+            events.send_workflow_event(execinfo=execinfo, event="start")
+            events.send_workflow_event(execinfo=execinfo, event="end")
+
+    # Publish concurrently (first event instantiates and registers the event handlers)
+    threads = [
+        threading.Thread(target=publish_events, args=(str(job_id),))
+        for job_id in range(njobs)
+    ]
+    for thread in threads:
+        thread.start()
+
+    # Poll for events concurrently
+    try:
+        evts = list(reader.wait_events(timeout=10))
+    finally:
+        for thread in threads:
+            thread.join(timeout=10)
+
+    # Check that all events were published
+    assert len(evts) == njobs * nevents_per_job
+
+
+def test_read_blocked_by_writer(tmp_path, sqlite3_ewoks_events):
+    """A reader must retry reading while a writer locks the database."""
+    lock_seconds = 0.5
+    retry_period = lock_seconds / 5  # do not retry long enough
+    retry_timeout = 5 * lock_seconds  # retry long enough
+
+    # Add events to the database
+    handlers, _ = sqlite3_ewoks_events
+    execinfo = _execinfo(handlers)
+    events.send_workflow_event(execinfo=execinfo, event="start")
+
+    # Writer locks the database
+    write_lock_acquired = threading.Event()
+
+    def write_lock_database():
+        conn = sqlite3.connect(str(tmp_path / "ewoks_events.db"))
+        try:
+            # Start write transaction
+            conn.execute("BEGIN EXCLUSIVE")
+            write_lock_acquired.set()
+
+            # Open write transaction locks the db
+            time.sleep(lock_seconds)
+
+            # Finish transaction
+            conn.commit()
+        finally:
+            conn.close()
+
+    uri = f"file:{tmp_path / 'ewoks_events.db'}"
+
+    # Read while locked: do not retry long enough
+    write_lock_acquired.clear()
+    thread = threading.Thread(target=write_lock_database)
+    thread.start()
+    try:
+        assert write_lock_acquired.wait(timeout=10)
+        with Sqlite3EwoksEventReader(uri=uri, timeout=retry_period) as reader:
+            evts = list(reader.wait_events(timeout=retry_period))
+    finally:
+        thread.join(timeout=10)
+
+    # Check that no events were read
+    assert len(evts) == 0
+
+    # Read while locked: retry long enough
+    write_lock_acquired.clear()
+    thread = threading.Thread(target=write_lock_database)
+    thread.start()
+    try:
+        assert write_lock_acquired.wait(timeout=10)
+        with Sqlite3EwoksEventReader(uri=uri, timeout=retry_period) as reader:
+            evts = list(reader.wait_events(timeout=retry_timeout))
+    finally:
+        thread.join(timeout=10)
+
+    # Check that all events were read
+    assert len(evts) == 1
+
+
+def test_failed_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys):
+    """A writer must wait for read locks to be released."""
+    _test_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys, succeed=False)
+
+
+def test_succeeded_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys):
+    """A writer must wait for read locks to be released."""
+    _test_write_blocked_by_reader(tmp_path, sqlite3_ewoks_events, capsys, succeed=True)
+
+
+def _test_write_blocked_by_reader(
+    tmp_path, sqlite3_ewoks_events, capsys, succeed: bool
+):
+    """A writer must wait for read locks to be released."""
+    lock_seconds = 0.5
+    if succeed:
+        retry_timeout = 5 * lock_seconds  # retry long enough
+    else:
+        retry_timeout = lock_seconds / 5  # do not retry long enough
+
+    # Add start event to the database
+    handlers, reader = sqlite3_ewoks_events
+    timeout_arg = {"name": "timeout", "value": retry_timeout}
+    handlers[0]["arguments"].append(timeout_arg)
+    execinfo = _execinfo(handlers)
+    events.send_workflow_event(execinfo=execinfo, event="start")
+
+    # Reader
+    read_lock_acquired = threading.Event()
+
+    def read_lock_database():
+        conn = sqlite3.connect(str(tmp_path / "ewoks_events.db"))
+        try:
+            # Start read transaction
+            conn.execute("BEGIN")
+            conn.execute("SELECT COUNT(*) FROM ewoks_events").fetchone()
+            read_lock_acquired.set()
+
+            # Open read transaction locks the db
+            time.sleep(lock_seconds)
+
+            # Finish transaction
+            conn.rollback()
+        finally:
+            conn.close()
+
+    # Write while locked
+    read_lock_acquired.clear()
+    thread = threading.Thread(target=read_lock_database)
+    thread.start()
+
+    _ = capsys.readouterr()
+    try:
+        assert read_lock_acquired.wait(timeout=10)
+        events.send_workflow_event(execinfo=execinfo, event="end")
+    finally:
+        thread.join(timeout=10)
+
+    # Check the error log of the dropped event
+    captured = capsys.readouterr()
+    if succeed:
+        assert "--- Logging error ---" not in captured.err
+    else:
+        assert "--- Logging error ---" in captured.err
+        assert "sqlite3.OperationalError: database is locked" in captured.err
+
+    # Check that the expected events were published
+    evts = list(reader.wait_events(timeout=1))
+    excepted = 2 if succeed else 1
+    assert len(evts) == excepted
+
+
+def _execinfo(handlers, job_id="123") -> dict:
+    return {
+        "job_id": job_id,
+        "workflow_id": "456",
+        "host_name": None,
+        "user_name": None,
+        "process_id": None,
+        "handlers": handlers,
+    }