diff --git a/tests/test_postgres_reconnect_after_drop.py b/tests/test_postgres_reconnect_after_drop.py index aed458b8..45c1ded2 100644 --- a/tests/test_postgres_reconnect_after_drop.py +++ b/tests/test_postgres_reconnect_after_drop.py @@ -156,3 +156,34 @@ def _pid_without_ending(database) -> int: cursor.execute("SELECT pg_backend_pid()") return cursor.fetchone()[0] + +def test_recovers_after_a_failed_reconnect(observer, tested): + """#242: a failover where the FIRST reconnect cannot reach the new primary. + + ``_reconnect`` closes the dropped link, then ``connect`` fails, leaving a + CLOSED psycopg2 connection. The next statement built its cursor on that + link first, so ``self._conn.cursor()`` raised ``InterfaceError: connection + already closed`` before the reconnect path could run - and the adapter + stayed wedged there until the process restarted. It must instead keep + retrying and recover once a reachable server returns. + + NO MOCKS: a real dropped backend, a real failed reconnect (the adapter's own + connect arguments aimed at a port nothing listens on - what a failover looks + like before the new primary answers), then a real recovery once the good + target is restored. + """ + good = tested.adapter._connect_args + unreachable = (good[0].replace(f":{PG_PORT}/", ":1/"), good[1], good[2], good[3]) + assert unreachable[0] != good[0], "could not aim the reconnect at a dead port" + tested.adapter._connect_args = unreachable + + _drop(observer, _pid(tested)) + with pytest.raises(Exception): + tested.fetch_one("SELECT 1 AS one") # reconnect to the dead port fails + with pytest.raises(Exception): + tested.fetch_one("SELECT 1 AS one") # still down: retried, never wedged + + tested.adapter._connect_args = good # the new primary is reachable now + assert tested.fetch_one("SELECT 1 AS one")["one"] == 1 # recovers + tested.execute(f"INSERT INTO {TABLE} (note) VALUES (?)", ["after-failed-reconnect"]) + assert _notes(observer) == ["after-failed-reconnect"] diff --git a/tina4_python/database/postgres.py b/tina4_python/database/postgres.py index 51db19a2..ce0a0f6d 100644 --- a/tina4_python/database/postgres.py +++ b/tina4_python/database/postgres.py @@ -204,6 +204,24 @@ def _reconnect(self): except Exception: pass + def _ensure_connection(self): + """Rebuild a server-dropped connection BEFORE a cursor is opened on it. + + #236 recovers a link that dies DURING a statement, but execute()/fetch() + build the cursor first, and ``self._conn.cursor()`` raises + ``InterfaceError: connection already closed`` on an already-closed link, + before the reconnect path can run. That is the state a FAILED reconnect + leaves behind - a failover where ``_reconnect`` closed the old link and + the replacement connect could not reach the new primary - after which + every later statement wedged there until the process restarted (#242). + Healing here, when the link is dead and no explicit transaction owns it, + lets a later statement recover once the server returns. A reconnect that + fails again propagates, but leaves the link dead, so the NEXT statement + retries rather than giving up for good. + """ + if not self._in_transaction and self._connection_is_dead(): + self._reconnect() + def _log_reconnect_retry(self, exc: Exception, sql: str): try: from tina4_python.debug import Log @@ -393,6 +411,7 @@ def execute(self, sql: str, params: list = None) -> DatabaseResult: re.search(r"\bRETURNING\b", sql, re.IGNORECASE) ) + self._ensure_connection() cursor = self._conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor = self._exec_with_handling(cursor, sql, params) @@ -489,6 +508,7 @@ def fetch(self, sql: str, params: list = None, # (2.28x), byte-identical output incl. native types (int/bool/Decimal/None) # -- cursor_factory only chooses the row CONTAINER, never value coercion, # so psycopg2's type adaptation is unchanged. + self._ensure_connection() cursor = self._conn.cursor() # #133: a write that returns rows (INSERT ... RETURNING) can be @@ -566,6 +586,7 @@ def fetch_one(self, sql: str, params: list = None) -> dict | None: # v3.13.12: see fetch() — trailing semicolons break the wrappers. sql = self._strip_trailing_semicolons(sql) sql = self._translate_sql(sql, bool(params)) + self._ensure_connection() cursor = self._conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) cursor = self._exec_with_handling(cursor, sql, params) row = cursor.fetchone()