fix(database): PostgreSQL recovers after a failed reconnect (#242)

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"]

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