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