From f7f29a815b0a51af35b537c4b04dfe58dcdebb28 Mon Sep 17 00:00:00 2001 From: Michael Date: Thu, 15 Oct 2026 13:08:24 +0200 Subject: [PATCH] Reconnect PostgreSQL after the server drops the connection A failover, an idle-session reap or pg_terminate_backend closes the connection under a running app. The adapter kept the dead psycopg2 connection, so every later query raised "connection already closed" until the process was restarted. The adapter now keeps its connect arguments and, when psycopg2 has marked the connection closed, opens a new one. The statement that hit the dropped link is retried once, but only when no explicit transaction was open and no uncommitted work was pending. Otherwise the caller sees the error and the next statement runs on the new connection. This mirrors the Firebird adapter's dead-connection handling. Session state set with SET after connecting does not carry over to the new connection; connect options do. Signed-off-by: Michael --- tests/test_postgres_reconnect_after_drop.py | 158 ++++++++++++++++++++ tina4_python/database/postgres.py | 71 ++++++++- 2 files changed, 226 insertions(+), 3 deletions(-) create mode 100644 tests/test_postgres_reconnect_after_drop.py diff --git a/tests/test_postgres_reconnect_after_drop.py b/tests/test_postgres_reconnect_after_drop.py new file mode 100644 index 00000000..aed458b8 --- /dev/null +++ b/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] + diff --git a/tina4_python/database/postgres.py b/tina4_python/database/postgres.py index 04cb039d..51db19a2 100644 --- a/tina4_python/database/postgres.py +++ b/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 "" + 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