Reconnect PostgreSQL after the server drops the connection

tests/test_postgres_reconnect_after_drop.py

@@ -0,0 +1,158 @@
+# Copyright (c) 2026 Code Infinity
+# SPDX-License-Identifier: MPL-2.0
+# This Source Code Form is subject to the terms of the Mozilla Public
+# License, v. 2.0. If a copy of the MPL was not distributed with this
+# file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+"""PostgreSQL: a connection the server drops is replaced, not reused forever.
+
+A failover, an idle-session reap or ``pg_terminate_backend`` closes the
+connection under a running app. The adapter used to keep the dead psycopg2
+connection, so every later query raised ``InterfaceError: connection already
+closed`` until the process was restarted - while a health check that does not
+touch the database kept the process looking healthy.
+
+Now the adapter reconnects from its original connect arguments. The statement
+that hit the dropped link is retried once, but only when nothing uncommitted
+went down with it. Inside an explicit transaction, or with uncommitted work
+pending, the caller sees the error, and the next statement runs on a new
+connection. This mirrors the Firebird adapter's dead-connection handling.
+
+NO MOCKS. A real PostgreSQL; the drop is a real ``pg_terminate_backend`` issued
+from a second connection, and every assertion reads what the server holds.
+"""
+
+import os
+import socket
+from urllib.parse import urlparse
+
+import pytest
+
+PG_URL = os.environ.get("TINA4_TEST_PG_URL", "postgres://tina4:tina4@localhost:55432/tina4_py")
+_PARSED = urlparse(PG_URL)
+PG_HOST, PG_PORT = _PARSED.hostname or "localhost", _PARSED.port or 5432
+TABLE = "reconnect_py_drop"
+
+
+def _pg_reachable() -> bool:
+    try:
+        with socket.create_connection((PG_HOST, PG_PORT), timeout=1.0):
+            return True
+    except OSError:
+        return False
+
+
+pytestmark = pytest.mark.skipif(
+    not _pg_reachable(),
+    reason=f"PostgreSQL not reachable at {PG_HOST}:{PG_PORT} - skip integration test",
+)
+
+
+def _connect(**kwargs):
+    from tina4_python.database import Database
+    return Database(PG_URL, **kwargs)
+
+
+@pytest.fixture
+def observer():
+    """A second connection: seeds the table, drops the tested one, reads back."""
+    database = _connect()
+    database.execute(f"DROP TABLE IF EXISTS {TABLE}")
+    database.execute(f"CREATE TABLE {TABLE} (id serial PRIMARY KEY, note text)")
+    yield database
+    database.execute(f"DROP TABLE IF EXISTS {TABLE}")
+    database.close()
+
+
+@pytest.fixture
+def tested():
+    database = _connect()
+    yield database
+    database.close()
+
+
+def _pid(database) -> int:
+    return database.fetch_one("SELECT pg_backend_pid() AS pid")["pid"]
+
+
+def _drop(observer, pid: int):
+    row = observer.fetch_one("SELECT pg_terminate_backend(?) AS gone", [pid])
+    assert row["gone"] is True
+
+
+def _notes(observer) -> list:
+    return [r["note"] for r in observer.fetch(f"SELECT note FROM {TABLE} ORDER BY id").records]
+
+
+def test_a_read_after_the_server_drops_the_connection_runs_on_a_new_one(observer, tested):
+    before = _pid(tested)
+    _drop(observer, before)
+
+    after = _pid(tested)
+
+    assert after != before
+    assert _pid(tested) == after
+
+
+def test_every_read_path_recovers(observer, tested):
+    _drop(observer, _pid(tested))
+    assert tested.fetch(f"SELECT note FROM {TABLE}").records == []
+
+    _drop(observer, _pid(tested))
+    assert tested.fetch_one("SELECT 1 AS one")["one"] == 1
+
+
+def test_a_write_after_the_drop_lands_exactly_once(observer, tested):
+    _drop(observer, _pid(tested))
+
+    tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["after-drop"])
+
+    assert _notes(observer) == ["after-drop"]
+
+
+def test_a_pooled_connection_recovers(observer):
+    pooled = _connect(pool=2)
+    try:
+        _drop(observer, _pid(pooled))
+        assert _pid(pooled) > 0
+        pooled.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["pooled"])
+        assert _notes(observer) == ["pooled"]
+    finally:
+        pooled.close()
+
+
+def test_inside_a_transaction_the_drop_surfaces_and_the_next_statement_recovers(observer, tested):
+    tested.start_transaction()
+    tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["in-txn"])
+    _drop(observer, _pid(tested))
+
+    with pytest.raises(Exception):
+        tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["second"])
+    try:
+        tested.rollback()
+    except Exception:
+        pass
+
+    tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["next"])
+    assert _notes(observer) == ["next"]
+
+
+def test_with_uncommitted_work_pending_the_statement_is_not_retried_alone(observer, tested):
+    tested.adapter.autocommit = False
+    tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["uncommitted"])
+    _drop(observer, _pid_without_ending(tested))
+
+    with pytest.raises(Exception):
+        tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["second"])
+
+    tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["next"])
+    tested.commit()
+    assert _notes(observer) == ["next"]
+
+
+def _pid_without_ending(database) -> int:
+    """The backend pid, read on the open transaction without closing it."""
+    cursor = database.adapter._conn.cursor()
+    cursor.execute("SELECT pg_backend_pid()")
+    return cursor.fetchone()[0]
+

tina4_python/database/postgres.py

@@ -33,6 +33,7 @@ class PostgreSQLAdapter(SqlCrudMixin, DatabaseAdapter):
     def __init__(self):
         super().__init__()
         self._conn = None
+        self._connect_args = None
         self._in_transaction: bool = False
         self.last_error: str | None = None
 
@@ -147,11 +148,74 @@ def _exec_with_handling(self, cursor, sql: str, params=None):
         diagnose, and the heal step also defers to explicit txns.
         """
         self._heal_aborted_txn()
+        nothing_pending = self._nothing_pending()
+        try:
+            self._safe_execute(cursor, sql, params)
+            return cursor
+        except Exception as e:
+            if self._in_transaction or not self._connection_is_dead():
+                self._on_query_error(e, sql, params)
+                raise
+            self._reconnect()
+            if not nothing_pending:
+                # Uncommitted work went down with the old connection, so the
+                # statement is not retried on its own; the caller sees the error.
+                self._on_query_error(e, sql, params)
+                raise
+            self._log_reconnect_retry(e, sql)
+        cursor = self._conn.cursor(cursor_factory=type(cursor))
         try:
             self._safe_execute(cursor, sql, params)
         except Exception as e:
             self._on_query_error(e, sql, params)
             raise
+        return cursor
+
+    def _connection_is_dead(self) -> bool:
+        """True when psycopg2 has marked the link closed under us (the server
+        dropped it, or the socket was reset). An adapter the caller closed
+        itself has ``_conn = None`` and is not "dead" - it stays closed.
+        """
+        return self._conn is not None and self._conn.closed != 0
+
+    def _nothing_pending(self) -> bool:
+        """True when the connection holds no uncommitted work, so a statement
+        that fails because the link died can be run again on a new one."""
+        try:
+            import psycopg2.extensions as _ext
+            return self._conn.info.transaction_status == _ext.TRANSACTION_STATUS_IDLE
+        except Exception:
+            return False
+
+    def _reconnect(self):
+        """Replace a dropped connection with a new one, opened from the
+        arguments the original :meth:`connect` received. Session state set with
+        ``SET`` after connecting does not carry over; connect ``options`` do.
+        """
+        try:
+            self._conn.close()
+        except Exception:
+            pass
+        connection_string, username, password, kwargs = self._connect_args
+        self.connect(connection_string, username, password, **kwargs)
+        try:
+            from tina4_python.debug import Log
+            Log.warning("PostgreSQL connection was closed by the server - reconnected.")
+        except Exception:
+            pass
+
+    def _log_reconnect_retry(self, exc: Exception, sql: str):
+        try:
+            from tina4_python.debug import Log
+            preview = sql.strip().splitlines()[0][:200] if sql else "<no sql>"
+            Log.warning(
+                f"PostgreSQL statement hit a dropped connection "
+                f"({exc.__class__.__name__}: {str(exc).strip()}) - retrying once "
+                f"on the new connection.",
+                sql=preview,
+            )
+        except Exception:
+            pass
 
     def _on_query_error(self, exc: Exception, sql: str, params=None):
         """Handle a failed query: log, store, auto-rollback if implicit.
@@ -286,6 +350,7 @@ def connect(self, connection_string: str, username: str = "", password: str = ""
                 "    uv add tina4-python[all-db]    # all five database drivers"
             )
 
+        self._connect_args = (connection_string, username, password, dict(kwargs))
         parsed = urlparse(connection_string)
 
         # Percent-DECODED: urlparse leaves userinfo escaped.
@@ -329,7 +394,7 @@ def execute(self, sql: str, params: list = None) -> DatabaseResult:
         )
 
         cursor = self._conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
-        self._exec_with_handling(cursor, sql, params)
+        cursor = self._exec_with_handling(cursor, sql, params)
 
         # Capture rowcount NOW, before the lastval()/SAVEPOINT probes below run
         # their own cursor.execute() calls and overwrite cursor.rowcount. A
@@ -441,7 +506,7 @@ def fetch(self, sql: str, params: list = None,
         # crashed on the server, which is the swap ADR-0024 exists to protect.
         paginated = not (is_write or limit is None or limit <= 0 or self._has_trailing_limit(sql))
         paginated_sql, paginated_params = self._paginate_sql(sql, params, limit, offset, paginated)
-        self._exec_with_handling(cursor, paginated_sql, paginated_params)
+        cursor = self._exec_with_handling(cursor, paginated_sql, paginated_params)
         columns = [d[0] for d in cursor.description] if cursor.description else []
         indexes = range(len(columns))
         rows = [
@@ -502,7 +567,7 @@ def fetch_one(self, sql: str, params: list = None) -> dict | None:
         sql = self._strip_trailing_semicolons(sql)
         sql = self._translate_sql(sql, bool(params))
         cursor = self._conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
-        self._exec_with_handling(cursor, sql, params)
+        cursor = self._exec_with_handling(cursor, sql, params)
         row = cursor.fetchone()
         result = self._decode_blobs(dict(row)) if row else None
         # v3.13.15 (#51): close the implicit transaction so the connection