From 6007171631c7b3a1599c0805f6e7920df332931a Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 14:03:39 +0200 Subject: [PATCH 1/5] EwoksEventReader: log error when last attempt failed when polling for events --- src/ewoksjob/events/readers/base.py | 28 +++++++++++++++++++++++----- 1 file changed, 23 insertions(+), 5 deletions(-) diff --git a/src/ewoksjob/events/readers/base.py b/src/ewoksjob/events/readers/base.py index d8be945..92999a9 100644 --- a/src/ewoksjob/events/readers/base.py +++ b/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 From 88846202b1e71b1063e3cb7ce9db27f6da786ada Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 14:05:39 +0200 Subject: [PATCH 2/5] pass 'timeout' argument to all Ewoks event readers/writers and do not allow unkown arguments --- src/ewoksjob/events/handlers/redis.py | 27 +++++++++++++++++++++----- src/ewoksjob/events/readers/redis.py | 16 ++++++++++++--- src/ewoksjob/events/readers/sqlite3.py | 25 ++++++++++++++++++++++-- 3 files changed, 58 insertions(+), 10 deletions(-) diff --git a/src/ewoksjob/events/handlers/redis.py b/src/ewoksjob/events/handlers/redis.py index 8db0c82..6fadde3 100644 --- a/src/ewoksjob/events/handlers/redis.py +++ b/src/ewoksjob/events/handlers/redis.py @@ -16,17 +16,34 @@ 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 - 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: diff --git a/src/ewoksjob/events/readers/redis.py b/src/ewoksjob/events/readers/redis.py index 94e5892..0bf8d1f 100644 --- a/src/ewoksjob/events/readers/redis.py +++ b/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) diff --git a/src/ewoksjob/events/readers/sqlite3.py b/src/ewoksjob/events/readers/sqlite3.py index 4fecf55..b4ebca3 100644 --- a/src/ewoksjob/events/readers/sqlite3.py +++ b/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 From 7a075b0798b13c2c87d147265cf93ae596a6df02 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 15:53:07 +0200 Subject: [PATCH 3/5] RedisEwoksEventHandler: implement new abstract method --- src/ewoksjob/events/handlers/redis.py | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/ewoksjob/events/handlers/redis.py b/src/ewoksjob/events/handlers/redis.py index 6fadde3..edfb558 100644 --- a/src/ewoksjob/events/handlers/redis.py +++ b/src/ewoksjob/events/handlers/redis.py @@ -35,6 +35,7 @@ def __init__( self._redis_url = url self._ttl = ttl self._timeout = timeout + self._connection = None def _connect(self) -> None: """This is called when no connection exists.""" @@ -49,6 +50,10 @@ def _connect(self) -> None: 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.""" From be66837880c6e9bbf9b2c678c2ee234e98a26c72 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:24:45 +0200 Subject: [PATCH 4/5] tests: concurrent reading/writing of ewoks events with sqlite3 --- src/ewoksjob/tests/test_events_sqlite3.py | 182 ++++++++++++++++++++++ 1 file changed, 182 insertions(+) create mode 100644 src/ewoksjob/tests/test_events_sqlite3.py diff --git a/src/ewoksjob/tests/test_events_sqlite3.py b/src/ewoksjob/tests/test_events_sqlite3.py new file mode 100644 index 0000000..23f3ad0 --- /dev/null +++ b/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, + } From 56c9b875151ab6181aba231e89747d160c976fcb Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:24:55 +0200 Subject: [PATCH 5/5] update changelog --- CHANGELOG.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index e042b4d..2655fa0 100644 --- a/CHANGELOG.md +++ b/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