Ewoks events sqlite3 bugs

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` `_connected()` method is now abstract
+
 ## [1.10.0] - 2026-07-01
 
 ### Added

src/ewoksutils/logging_utils/connection.py

@@ -1,28 +1,23 @@
 import logging
-import time
 from abc import abstractmethod
 from typing import Any
 
 
 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._retry_time = None
-        #
-        # Exponential backoff parameters.
-        #
-        self._retry_start = 1.0
-        self._retry_max = 30.0
-        self._retry_factor = 2.0
+        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
 
@@ -31,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."""
@@ -41,41 +40,17 @@ 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 _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.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:
-            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)
 

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

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))

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)