From 2eb5ac8f0f06c110dbec06d605afccbc2d5d16ee Mon Sep 17 00:00:00 2001 From: Ryan Malloy Date: Wed, 2 Sep 2026 00:29:04 -0600 Subject: [PATCH] The finalizer's lock probe couldn't see its own thread The cursor finalizer runs at GC time on whatever thread happened to allocate, so it must not touch the wire while a statement owns it. It checked with _wire_lock.acquire(blocking=False), meaning to ask "is anyone using the wire?" -- but an RLock grants a reentrant acquire to its own owner. When GC fired on the thread that was mid-statement, the probe returned True, the finalizer concluded it had exclusive access, and it sent CLOSE/RELEASE into the middle of the statement it had just interrupted. The victim got -208 on valid SQL. Refcounting hid it. A dropped cursor is freed at the drop, before the next statement runs. A cursor caught in a reference cycle waits for a collection instead, and cycles are ordinary -- any traceback that holds a cursor makes one. Reproduced by putting an abandoned scrollable cursor in a cycle and collecting mid-statement. _wire_lock is now a small wrapper that tracks owner and depth, so held_by_current_thread answers the question the finalizer was actually asking. Same-thread GC defers to the cleanup queue exactly as another thread's would. This also retires the _is_owned() call in _ensure_transaction, which was reaching into CPython private API for the same information. The finalizer's error handling had the stale-cleanup defect too: a leftover CLOSE draws -267, an OperationalError, which is in WIRE_ERRORS, so it force-closed a healthy connection over a no-op. Server-reported errors are told apart from wire failures by their sqlcode, matching _drain_pending_cleanup. --- src/informix_db/connections.py | 60 +++++++++++++++++++++- src/informix_db/cursors.py | 56 +++++++++++++++----- tests/test_statement_slot.py | 93 ++++++++++++++++++++++++++++++++++ 3 files changed, 195 insertions(+), 14 deletions(-) diff --git a/src/informix_db/connections.py b/src/informix_db/connections.py index 9ab081f..bc542eb 100644 --- a/src/informix_db/connections.py +++ b/src/informix_db/connections.py @@ -206,6 +206,62 @@ _DEFAULT_ENV: dict[str, str] = { } +class _WireLock: + """A reentrant lock that can say whether *this* thread already holds it. + + ``threading.RLock`` cannot answer that question in public API, and + the difference is not academic. The cursor finalizer runs at GC time + on whatever thread happened to allocate — including a thread that is + at that moment mid-statement holding this lock. It probes with + ``acquire(blocking=False)`` intending to mean "is anyone using the + wire?", but an RLock grants a reentrant acquire to its own owner, so + the probe returns True and the finalizer sends CLOSE/RELEASE into the + middle of the statement it interrupted. The victim gets ``-208``. + + Reachability is not exotic. Refcounting frees a dropped cursor + immediately, before the next statement, which is why this went + unnoticed — but a cursor caught in a reference cycle waits for a + collection instead, and cycles are ordinary in Python. Any traceback + that holds a cursor makes one. + + ``held_by_current_thread`` reads ``_depth``/``_owner`` without the + lock, which is safe: both are mutated only under it, and the only + values another thread can leave behind are a depth of zero or an + owner that isn't us. Either way the answer is False, correctly. + """ + + __slots__ = ("_depth", "_lock", "_owner") + + def __init__(self) -> None: + self._lock = threading.RLock() + self._owner: int | None = None + self._depth = 0 + + def acquire(self, blocking: bool = True, timeout: float = -1) -> bool: + acquired = self._lock.acquire(blocking, timeout) + if acquired: + self._owner = threading.get_ident() + self._depth += 1 + return acquired + + def release(self) -> None: + self._depth -= 1 + if self._depth == 0: + self._owner = None + self._lock.release() + + def __enter__(self) -> _WireLock: + self.acquire() + return self + + def __exit__(self, *exc_info: object) -> None: + self.release() + + @property + def held_by_current_thread(self) -> bool: + return self._depth > 0 and self._owner == threading.get_ident() + + class Connection: """A SQLI session. Owns one TCP socket and the post-login state. @@ -250,7 +306,7 @@ class Connection: # this lock with a timeout, then calls ``conn.rollback()`` — # which itself acquires the lock. Same thread, two acquires. # Reentrance must be cheap and correct. - self._wire_lock = threading.RLock() + self._wire_lock = _WireLock() # Phase 29: deferred-cleanup queue for cursor finalizers that # couldn't acquire the wire lock at GC time. Each entry is a # PDU's worth of bytes (typically a CLOSE or RELEASE) that @@ -696,7 +752,7 @@ class Connection: # method but stable across versions; cheap (~50ns) and only # checks the current thread. If it ever changes shape, drop # this assert — the doc still names the precondition. - assert self._wire_lock._is_owned(), ( + assert self._wire_lock.held_by_current_thread, ( "_ensure_transaction called without _wire_lock held; " "the cursor method that called it must wrap its body in " "`with self._conn._wire_lock:`" diff --git a/src/informix_db/cursors.py b/src/informix_db/cursors.py index cbedd56..70cdc5f 100644 --- a/src/informix_db/cursors.py +++ b/src/informix_db/cursors.py @@ -131,6 +131,24 @@ def _finalize_cursor( conn = conn_ref() if conn is None or conn.closed: return + if conn._wire_lock.held_by_current_thread: + # GC fired on the very thread that is mid-statement. The + # non-blocking acquire below would *succeed* here — an RLock + # grants a reentrant acquire to its own owner — and we would + # send CLOSE/RELEASE into the middle of that statement, killing + # it with -208. Defer instead, exactly as for another thread. + # + # Refcounting hides this: a dropped cursor is freed at the drop, + # before the next statement. A cursor caught in a reference cycle + # waits for a collection instead, and cycles are ordinary — any + # traceback holding a cursor makes one. + conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU]) + _log.debug( + "cursor finalizer: GC ran on the thread holding the wire lock; " + "enqueued CLOSE+RELEASE for deferred cleanup on conn %s", + id(conn), + ) + return if not conn._wire_lock.acquire(blocking=False): # Another thread is mid-operation on this connection. Don't # deadlock; instead, hand the cleanup bytes to the connection's @@ -152,18 +170,32 @@ def _finalize_cursor( conn._send_pdu(_RELEASE_PDU) conn._drain_to_eot() except WIRE_ERRORS as exc: - # Wire desync during cleanup — same doctrine as - # ``_raise_sq_err``: the wire is unrecoverable, force-close - # the connection. Asymmetric handling of the same failure - # mode would be a Hamilton smell. - _log.warning( - "cursor finalizer: wire desync during cleanup; " - "force-closing connection: %r", - exc, - ) - conn._closed = True - with contextlib.suppress(Exception): - conn._sock.close() + if getattr(exc, "sqlcode", None) is not None: + # The *server* rejected the cleanup — typically a stale + # CLOSE for a cursor it no longer has, which answers + # -267 "the transaction has been rolled back". That is + # an OperationalError, which is in WIRE_ERRORS, so this + # branch used to force-close a perfectly healthy + # connection over a no-op. ``_raise_sq_err`` self-drains + # the trailing SQ_EOT, so the wire is still aligned. + # Same distinction as _drain_pending_cleanup. + _log.debug( + "cursor finalizer: server rejected cleanup (stale): %r", + exc, + ) + else: + # No sqlcode: the socket died or framing desynced and we + # can no longer say where a response ends. Force-close — + # same doctrine as ``_raise_sq_err``. Asymmetric handling + # of the same failure mode would be a Hamilton smell. + _log.warning( + "cursor finalizer: wire desync during cleanup; " + "force-closing connection: %r", + exc, + ) + conn._closed = True + with contextlib.suppress(Exception): + conn._sock.close() except InterfaceError: # Connection was closed by another thread between our # ``conn.closed`` check above and the actual write. No-op: diff --git a/tests/test_statement_slot.py b/tests/test_statement_slot.py index 21ff125..31babf2 100644 --- a/tests/test_statement_slot.py +++ b/tests/test_statement_slot.py @@ -260,3 +260,96 @@ def test_connection_closes_cleanly_with_a_scroll_cursor_open( with contextlib.suppress(Exception): conn.close() assert conn.closed + + +# --------------------------------------------------------------------------- +# The wire lock's blind spot +# --------------------------------------------------------------------------- + + +def test_wire_lock_reports_its_own_thread() -> None: + """The whole point. ``RLock.acquire(blocking=False)`` returns True for + the owning thread, so the finalizer's probe could not tell "nobody is + using the wire" from "I am, right now, mid-statement".""" + from informix_db.connections import _WireLock + + lock = _WireLock() + assert not lock.held_by_current_thread + with lock: + assert lock.held_by_current_thread + assert lock.acquire(blocking=False), "must still be reentrant" + lock.release() + assert lock.held_by_current_thread, "still held at depth 1" + assert not lock.held_by_current_thread + + +def test_wire_lock_is_not_held_by_other_threads() -> None: + import threading + + from informix_db.connections import _WireLock + + lock = _WireLock() + seen: list[bool] = [] + entered = threading.Event() + done = threading.Event() + + def holder() -> None: + with lock: + entered.set() + done.wait(5) + + t = threading.Thread(target=holder) + t.start() + entered.wait(5) + seen.append(lock.held_by_current_thread) + done.set() + t.join(5) + assert seen == [False], "another thread's hold must not read as ours" + + +@pytest.mark.integration +def test_finalizer_defers_when_gc_runs_on_the_locking_thread( + conn_params: ConnParams, +) -> None: + """GC fires on whatever thread allocated. When that is the thread + holding the wire lock, the finalizer must write nothing — it used to + acquire the RLock reentrantly and send CLOSE/RELEASE into the running + statement, killing it with -208.""" + import gc + + with _connect(conn_params) as conn: + gc.disable() + try: + victim = conn.cursor(scrollable=True) + victim.execute("SELECT tabid FROM systables ORDER BY tabid") + victim.fetch_first() + # A reference cycle, so collection waits for gc rather than + # happening at the drop. Cycles are ordinary in Python. + cycle = [victim] + cycle.append(cycle) + del victim, cycle + + writes: list[bytes] = [] + original_write = conn._sock.write_all + conn._sock.write_all = lambda b: ( + writes.append(b), + original_write(b), + )[1] + try: + with conn._wire_lock: # stand in for "mid-statement" + gc.collect() + assert writes == [], ( + "finalizer wrote to the wire while a statement " + "owned it" + ) + assert conn._pending_cleanup, ( + "cleanup should have been deferred, not dropped" + ) + finally: + conn._sock.write_all = original_write + finally: + gc.enable() + + cur = conn.cursor() + cur.execute("SELECT COUNT(*) FROM systables") + assert cur.fetchone() is not None