From 3f8b408809a6cb276fa0292e2d1221cfbb990b46 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 15:42:11 +0200 Subject: [PATCH 1/8] ConnectionHandler: always call super().handleError --- src/ewoksutils/logging_utils/connection.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/src/ewoksutils/logging_utils/connection.py b/src/ewoksutils/logging_utils/connection.py index 3938ba9..20fee75 100644 --- a/src/ewoksutils/logging_utils/connection.py +++ b/src/ewoksutils/logging_utils/connection.py @@ -66,10 +66,9 @@ def _ensure_connection(self) -> bool: return False def handleError(self, record): - if self.closeOnError and self._connected(): + if self._disconnect_on_error and self._connected(): self._disconnect() - else: - super().handleError(record) + super().handleError(record) def emit(self, record): try: From 3cba897eb1211399dcccd65af97b5d90e2d6e6c2 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 15:42:53 +0200 Subject: [PATCH 2/8] ConnectionHandler: remove unused retry logic --- src/ewoksutils/logging_utils/connection.py | 36 +++------------------- 1 file changed, 4 insertions(+), 32 deletions(-) diff --git a/src/ewoksutils/logging_utils/connection.py b/src/ewoksutils/logging_utils/connection.py index 20fee75..afb41fc 100644 --- a/src/ewoksutils/logging_utils/connection.py +++ b/src/ewoksutils/logging_utils/connection.py @@ -1,5 +1,4 @@ import logging -import time from abc import abstractmethod from typing import Any @@ -13,13 +12,6 @@ def __init__(self): super().__init__() self._connection = None self.closeOnError = False - self._retry_time = None - # - # Exponential backoff parameters. - # - self._retry_start = 1.0 - self._retry_max = 30.0 - self._retry_factor = 2.0 @abstractmethod def _connect(self, timeout=1) -> None: @@ -44,27 +36,6 @@ def _send_serialized_record(self, srecord: Any): def _connected(self) -> bool: return self._connection is not None - def _ensure_connection(self) -> bool: - if self._connected(): - return True - now = time.time() - if self._retry_time is not None and now < self._retry_time: - return False - self._connect() - if self._connected(): - # Connection succeeded: no delay for next connection attempt - self._retry_time = None - return True - # Connection failed: no next connection attempt before _retry_time - if self._retry_time is None: - self._retry_period = self._retry_start - else: - self._retry_period = self._retry_period * self._retry_factor - if self._retry_period > self._retry_max: - self._retry_period = self._retry_max - self._retry_time = now + self._retry_period - return False - def handleError(self, record): if self._disconnect_on_error and self._connected(): self._disconnect() @@ -72,9 +43,10 @@ def handleError(self, record): def emit(self, record): try: - if self._ensure_connection(): - s = self._serialize_record(record) - self._send_serialized_record(s) + if not self._connected(): + self._connect() + s = self._serialize_record(record) + self._send_serialized_record(s) except Exception: self.handleError(record) From 597ab1bd0b7ec6202e7c5b8aff34315b7dbab528 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:07:44 +0200 Subject: [PATCH 3/8] ConnectionHandler: disconnect_on_error argument and abstract method --- src/ewoksutils/logging_utils/connection.py | 22 +++++++++++++--------- 1 file changed, 13 insertions(+), 9 deletions(-) diff --git a/src/ewoksutils/logging_utils/connection.py b/src/ewoksutils/logging_utils/connection.py index afb41fc..8478591 100644 --- a/src/ewoksutils/logging_utils/connection.py +++ b/src/ewoksutils/logging_utils/connection.py @@ -4,17 +4,20 @@ class ConnectionHandler(logging.Handler): - """A python handler with a generic underlying connection. The - only requirement is that the connection closes itself on garbage collection. + """A python handler with a generic underlying connection. The only + requirements are that the connection closes itself on garbage + collection and that the connection handles timeouts natively. """ - def __init__(self): + def __init__(self, disconnect_on_error: bool = False): + """ + :param disconnect_on_error: disconnect when emitting a record failed + """ super().__init__() - self._connection = None - self.closeOnError = False + self._disconnect_on_error = disconnect_on_error @abstractmethod - def _connect(self, timeout=1) -> None: + def _connect(self) -> None: """This is called when no connection exists.""" pass @@ -23,6 +26,10 @@ def _disconnect(self) -> None: """This is called when a connection exists and is connected.""" pass + @abstractmethod + def _connected(self) -> bool: + pass + @abstractmethod def _serialize_record(self, record: logging.LogRecord) -> Any: """Convert a record to something that can be given to the connection.""" @@ -33,9 +40,6 @@ def _send_serialized_record(self, srecord: Any): """Send the output from `_serialize_record` to the connection.""" pass - def _connected(self) -> bool: - return self._connection is not None - def handleError(self, record): if self._disconnect_on_error and self._connected(): self._disconnect() From 30d07a2996a53c1402612429668406e3115f6d98 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:09:04 +0200 Subject: [PATCH 4/8] Sqlite3Handler: timeout and implement abstract method --- src/ewoksutils/logging_utils/sqlite3.py | 83 ++++++++++++++----------- 1 file changed, 47 insertions(+), 36 deletions(-) diff --git a/src/ewoksutils/logging_utils/sqlite3.py b/src/ewoksutils/logging_utils/sqlite3.py index ff17b3b..ca37490 100644 --- a/src/ewoksutils/logging_utils/sqlite3.py +++ b/src/ewoksutils/logging_utils/sqlite3.py @@ -1,5 +1,4 @@ import sqlite3 -import time from typing import Any from typing import Dict from typing import List @@ -13,39 +12,62 @@ class Sqlite3Handler(ConnectionHandler): - def __init__(self, uri: str, table: str, field_types: Dict): + def __init__( + self, + uri: str, + table: str, + field_types: Dict, + timeout: float = 10, + disconnect_on_error: bool = False, + ): + """ + :param uri: for example "file:/path/to/test.db" or "file:///path/to/test.db". + :param table: name of the database table in which records are inserted + (the table is created when missing). + :param field_types: mapping from record attribute names (table columns) + to python types. + :param timeout: native sqlite3 busy timeout: the maximum time to wait + for database locks to be released by other connections. + 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._uri = uri + self._timeout = timeout self._field_sql_types = sqlite3_utils.python_to_sql_types(field_types) + self._ensure_table_query = sqlite3_utils.ensure_table_query( table, self._field_sql_types ) self._insert_row_query = sqlite3_utils.insert_query( table, len(self._field_sql_types) ) + + self._connection = None self._connection_context = None - super().__init__() - def _connect(self, timeout=1) -> None: - if self._connection is not None: - raise RuntimeError("Already connected") - ctx = None + def _connect(self) -> None: + ctx = sqlite3_utils.connect( + self._uri, timeout=self._timeout, uri=True, check_same_thread=False + ) try: - ctx = sqlite3_utils.connect( - self._uri, timeout=timeout, uri=True, check_same_thread=False - ) conn = ctx.__enter__() self._sql_query(self._ensure_table_query, conn=conn) - except (OSError, TimeoutError): - if ctx is not None: - ctx.__exit__(None, None, None) + except BaseException: + ctx.__exit__(None, None, None) self._connection = None self._connection_context = None - else: - self._connection = conn - self._connection_context = ctx + raise + self._connection = conn + self._connection_context = ctx def _disconnect(self) -> None: self._connection_context.__exit__(None, None, None) + self._connection = None + self._connection_context = None + + def _connected(self) -> bool: + return self._connection is not None def _send_serialized_record(self, values: Optional[Sqlite3RecordType]): if values: @@ -63,27 +85,16 @@ def _sql_query( sql: str, parameters: Sequence = tuple(), conn: Optional[sqlite3.Connection] = None, - timeout=1, ) -> None: if conn is None: conn = self._connection - exception = None - t0 = time.time() - while True: - try: - conn.execute(sql, parameters) - break - except sqlite3.OperationalError as e: - exception = e - t1 = time.time() - if timeout is not None and (t1 - t0) > timeout: - raise TimeoutError("cannot execute SQL query") from exception - while True: + try: + conn.execute(sql, parameters) + conn.commit() + except BaseException: + # Do not leave the failed query pending in an open transaction. try: - conn.commit() - break - except sqlite3.OperationalError as e: - exception = e - t1 = time.time() - if timeout is not None and (t1 - t0) > timeout: - raise TimeoutError("cannot commit SQL query") from exception + conn.rollback() + except sqlite3.OperationalError: + pass + raise From 73251e2dad7411c9f07cdf255eae49a69500af42 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:09:33 +0200 Subject: [PATCH 5/8] test_connection_handler: implement abstract method --- src/ewoksutils/tests/test_logging_utils.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/src/ewoksutils/tests/test_logging_utils.py b/src/ewoksutils/tests/test_logging_utils.py index e8d930f..fbde624 100644 --- a/src/ewoksutils/tests/test_logging_utils.py +++ b/src/ewoksutils/tests/test_logging_utils.py @@ -40,13 +40,20 @@ def __del__(self): connected = False class MyHandler(ConnectionHandler): - def _connect(self, timeout=1) -> None: + def __init__(self, disconnect_on_error: bool = False): + super().__init__(disconnect_on_error=disconnect_on_error) + self._connection = None + + def _connect(self) -> None: self._connection = Connection() def _disconnect(self) -> None: del self._connection self._connection = None + def _connected(self) -> bool: + return self._connection is not None + def _send_serialized_record(self, srecord): destination.append(srecord) From 977378fb5e9b8b0af441b36320db6e1a7d72734f Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:09:59 +0200 Subject: [PATCH 6/8] test Sqlite3Handler in terms of concurrency --- src/ewoksutils/tests/test_logging_sqlite3.py | 135 +++++++++++++++++++ 1 file changed, 135 insertions(+) create mode 100644 src/ewoksutils/tests/test_logging_sqlite3.py diff --git a/src/ewoksutils/tests/test_logging_sqlite3.py b/src/ewoksutils/tests/test_logging_sqlite3.py new file mode 100644 index 0000000..8120c1e --- /dev/null +++ b/src/ewoksutils/tests/test_logging_sqlite3.py @@ -0,0 +1,135 @@ +import logging +import sqlite3 +import threading +import time + +from .. import sqlite3_utils +from ..logging_utils.sqlite3 import Sqlite3Handler + +FIELD_TYPES = {"field1": 0, "field2": ""} + + +def test_concurrent_read_write(tmp_path): + """Records emitted from several handlers while a reader is polling + must all be inserted.""" + uri = str(tmp_path / "test.db") + nwriters = 4 + nrecords = 20 + + # Barrier which ensures that reading and writing overlaps + started = threading.Barrier(nwriters + 1, timeout=10) + + # Publish concurrently + def write_records(value): + handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES) + try: + started.wait() + for i in range(nrecords): + handler.handle(_make_record(field1=value, field2=str(i))) + finally: + handler.close() + + writer_threads = [ + threading.Thread(target=write_records, args=(value,)) + for value in range(nwriters) + ] + + # Read concurrently + stop_reading = threading.Event() + + def read_records(): + with sqlite3_utils.connect(uri, uri=True) as conn: + started.wait() + while not stop_reading.is_set(): + try: + _ = list( + sqlite3_utils.select(conn, "mytable", field_types=FIELD_TYPES) + ) + except sqlite3.OperationalError: + pass + time.sleep(0.01) + + reader_thread = threading.Thread(target=read_records) + reader_thread.start() + try: + for thread in writer_threads: + thread.start() + for thread in writer_threads: + thread.join(timeout=60) + finally: + stop_reading.set() + reader_thread.join(timeout=10) + + # Check that all records were published + assert len(_select_all_records(uri)) == nwriters * nrecords + + +def test_write_blocked_by_reader(tmp_path): + """A writer must wait for read locks to be released.""" + lock_seconds = 0.5 + retry_period = lock_seconds / 5 # do not retry long enough + retry_timeout = 5 * lock_seconds # retry long enough + + # Add record to the database + uri = str(tmp_path / "test.db") + handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES) + handler.handle(_make_record(field1=1)) + + # Reader + read_lock_acquired = threading.Event() + + def read_lock_database(): + conn = sqlite3.connect(uri) + try: + # Start read transaction + conn.execute("BEGIN") + conn.execute("SELECT COUNT(*) FROM mytable").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: do not retry long enough + read_lock_acquired.clear() + thread = threading.Thread(target=read_lock_database) + thread.start() + + handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES, timeout=retry_period) + try: + assert read_lock_acquired.wait(timeout=10) + handler.handle(_make_record(field1=2)) + finally: + thread.join(timeout=10) + handler.close() + + # Write while locked: retry long enough + read_lock_acquired.clear() + thread = threading.Thread(target=read_lock_database) + thread.start() + + handler = Sqlite3Handler(uri, "mytable", FIELD_TYPES, timeout=retry_timeout) + try: + assert read_lock_acquired.wait(timeout=10) + # Write while the reader locks + handler.handle(_make_record(field1=3)) + finally: + thread.join(timeout=10) + handler.close() + + # Check that the expected records were published + records = _select_all_records(uri) + assert [record["field1"] for record in records] == [1, 3] + + +def _make_record(**fields) -> logging.LogRecord: + return logging.makeLogRecord(fields) + + +def _select_all_records(uri: str) -> list: + with sqlite3_utils.connect(uri, uri=True) as conn: + return list(sqlite3_utils.select(conn, "mytable", field_types=FIELD_TYPES)) From dfa04ac399decfce5e9bb6f3ea31973cd27a34d0 Mon Sep 17 00:00:00 2001 From: woutdenolf Date: Tue, 14 Jul 2026 16:10:08 +0200 Subject: [PATCH 7/8] modify changelog --- CHANGELOG.md | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index d816db0..e4ec44f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `ConnectionHandler`: add `disconnect_on_error` arguments. +- `Sqlite3Handler`: add `disconnect_on_error` and `timeout` arguments. + +### Changed + +- `ConnectionHandler` has abstract method `_connected()`. + ## [1.10.0] - 2026-07-01 ### Added From 1679573a2ea96547a4e1aacab445d3634e135607 Mon Sep 17 00:00:00 2001 From: Wout De Nolf Date: Wed, 15 Jul 2026 17:27:12 +0200 Subject: [PATCH 8/8] Update CHANGELOG.md MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Loïc Huder <42204205+loichuder@users.noreply.github.com> --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e4ec44f..ef2c34d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,7 +14,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- `ConnectionHandler` has abstract method `_connected()`. +- `ConnectionHandler` `_connected()` method is now abstract ## [1.10.0] - 2026-07-01