Ryan Malloy
2eb5ac8f0f
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.
356 lines
13 KiB
Python
356 lines
13 KiB
Python
"""One statement per session, and what happens when we forget that.
|
|
|
|
SQLI gives a session a single statement slot. ``SQ_CLOSE``,
|
|
``SQ_RELEASE`` and ``SQ_SFETCH`` all act on whatever statement is
|
|
current — none of them names one. Ordinary use never notices, because a
|
|
non-scrollable cursor materializes its rows and releases the statement
|
|
before returning, so the slot is free again by the time anyone looks.
|
|
|
|
A scrollable cursor is the exception: it holds the slot open on purpose.
|
|
Two things then went wrong, and neither said so.
|
|
|
|
**Another statement on the same connection.** The server returns ``-285``
|
|
for the new statement and *also* destroys the scrollable cursor — its
|
|
next fetch comes back ``-267`` "the transaction has been rolled back,
|
|
all locks released". Two unattributable errors from code that reads as
|
|
completely ordinary: iterate a large result set, run a lookup partway
|
|
through. It is now a ``ProgrammingError`` that says what happened.
|
|
|
|
**The deferred-cleanup queue drained at the wrong moment.** A cursor
|
|
finalizer that can't get the wire lock hands its CLOSE/RELEASE to a
|
|
queue for the next operation to flush. That queue was flushed before
|
|
*every* PDU. But a finalizer enqueues precisely because another thread
|
|
holds the lock, i.e. is mid-statement — so the flush landed inside that
|
|
thread's own statement and released it. The victim saw ``-208`` when it
|
|
happened before the first fetch, ``-267`` between fetch batches. The
|
|
flush now happens only at a statement boundary, where the queued CLOSE
|
|
addresses the orphan it was meant for.
|
|
|
|
A stale queue entry was fatal too: the finalizer enqueues, the cursor
|
|
then gets closed properly, and the leftover CLOSE draws ``-267`` — an
|
|
``OperationalError``, which is in ``WIRE_ERRORS``, which force-closed a
|
|
perfectly healthy connection. Server-reported errors are now told apart
|
|
from wire failures by the presence of a ``sqlcode``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import contextlib
|
|
|
|
import pytest
|
|
|
|
import informix_db
|
|
from informix_db.cursors import _CLOSE_PDU, _RELEASE_PDU
|
|
from tests.conftest import ConnParams
|
|
|
|
|
|
def _connect(conn_params: ConnParams, **kw) -> informix_db.Connection:
|
|
return informix_db.connect(
|
|
host=conn_params.host,
|
|
port=conn_params.port,
|
|
user=conn_params.user,
|
|
password=conn_params.password,
|
|
database=conn_params.database,
|
|
server=conn_params.server,
|
|
connect_timeout=10.0,
|
|
read_timeout=25.0,
|
|
autocommit=True,
|
|
**kw,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# The conflict check — bookkeeping, no server
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _FakeScroll:
|
|
def __init__(self, open_: bool = True) -> None:
|
|
self._server_cursor_open = open_
|
|
|
|
|
|
def _conflict_check(conn: informix_db.Connection, requester: object) -> None:
|
|
conn._check_scroll_cursor_conflict(requester)
|
|
|
|
|
|
def test_conflict_check_is_silent_with_no_scroll_cursor() -> None:
|
|
conn = informix_db.Connection.__new__(informix_db.Connection)
|
|
conn._open_scroll_cursor = None
|
|
_conflict_check(conn, object())
|
|
|
|
|
|
def test_conflict_check_forgets_a_collected_cursor() -> None:
|
|
"""The ref is weak so an abandoned scrollable cursor can still be
|
|
finalized. A dead ref means the slot is free."""
|
|
import weakref
|
|
|
|
conn = informix_db.Connection.__new__(informix_db.Connection)
|
|
victim = _FakeScroll()
|
|
conn._open_scroll_cursor = weakref.ref(victim)
|
|
del victim
|
|
_conflict_check(conn, object())
|
|
assert conn._open_scroll_cursor is None
|
|
|
|
|
|
def test_conflict_check_lets_the_owner_through() -> None:
|
|
"""Re-executing the *same* scrollable cursor is allowed — it closes
|
|
its own server-side cursor first."""
|
|
import weakref
|
|
|
|
conn = informix_db.Connection.__new__(informix_db.Connection)
|
|
owner = _FakeScroll()
|
|
conn._open_scroll_cursor = weakref.ref(owner)
|
|
_conflict_check(conn, owner)
|
|
|
|
|
|
def test_conflict_check_forgets_a_closed_cursor() -> None:
|
|
import weakref
|
|
|
|
conn = informix_db.Connection.__new__(informix_db.Connection)
|
|
done = _FakeScroll(open_=False)
|
|
conn._open_scroll_cursor = weakref.ref(done)
|
|
_conflict_check(conn, object())
|
|
assert conn._open_scroll_cursor is None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Against a real server
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_second_statement_is_refused_while_scroll_cursor_is_open(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
"""Used to be -285 for the new statement plus -267 for the scrollable
|
|
cursor. Now it's one error that names the cause, and the scrollable
|
|
cursor is untouched."""
|
|
with _connect(conn_params) as conn:
|
|
scroll = conn.cursor(scrollable=True)
|
|
scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
|
|
first = scroll.fetch_first()
|
|
assert first is not None
|
|
|
|
other = conn.cursor()
|
|
with pytest.raises(informix_db.ProgrammingError, match="scrollable"):
|
|
other.execute("SELECT COUNT(*) FROM systables")
|
|
|
|
assert scroll.fetch_absolute(1) is not None, (
|
|
"the refused statement must not have disturbed the cursor"
|
|
)
|
|
scroll.close()
|
|
other.execute("SELECT COUNT(*) FROM systables")
|
|
assert other.fetchone() is not None, "slot must free on close"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_executemany_is_refused_too(conn_params: ConnParams) -> None:
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_slot (k INT)")
|
|
scroll = conn.cursor(scrollable=True)
|
|
scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
|
|
scroll.fetch_first()
|
|
with pytest.raises(informix_db.ProgrammingError, match="scrollable"):
|
|
cur.executemany("INSERT INTO t_slot VALUES (?)", [(1,), (2,)])
|
|
scroll.close()
|
|
cur.executemany("INSERT INTO t_slot VALUES (?)", [(1,), (2,)])
|
|
cur.execute("SELECT COUNT(*) FROM t_slot")
|
|
assert cur.fetchone() == (2,)
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_scroll_cursor_can_be_re_executed(conn_params: ConnParams) -> None:
|
|
"""The owner is the one caller allowed past the conflict check, and
|
|
that is only safe because it closes its own server-side cursor
|
|
first. Without that it collides with itself."""
|
|
with _connect(conn_params) as conn:
|
|
scroll = conn.cursor(scrollable=True)
|
|
for _ in range(4):
|
|
scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
|
|
assert scroll.fetch_first() is not None
|
|
scroll.close()
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_abandoning_a_scroll_cursor_frees_the_slot(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
"""Dropping the last reference must let the connection be used again
|
|
— the finalizer closes the cursor and the weak ref goes dead."""
|
|
import gc
|
|
|
|
with _connect(conn_params) as conn:
|
|
scroll = conn.cursor(scrollable=True)
|
|
scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
|
|
scroll.fetch_first()
|
|
del scroll
|
|
gc.collect()
|
|
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT COUNT(*) FROM systables")
|
|
assert cur.fetchone() is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Deferred cleanup
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_queued_cleanup_does_not_land_inside_a_running_statement(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
"""Exactly what a cross-thread finalizer does: it lost the wire lock,
|
|
so it queued its CLOSE/RELEASE while another thread was mid-statement.
|
|
The flush used to happen before that thread's very next PDU — its own
|
|
CURNAME/NFETCH — releasing the statement out from under it."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("CREATE TEMP TABLE t_defer (k INT)")
|
|
cur.executemany(
|
|
"INSERT INTO t_defer VALUES (?)", [(i,) for i in range(50)]
|
|
)
|
|
|
|
original = cur._read_describe_response
|
|
|
|
def enqueue_between_prepare_and_fetch() -> None:
|
|
result = original()
|
|
conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU])
|
|
return result
|
|
|
|
cur._read_describe_response = enqueue_between_prepare_and_fetch
|
|
try:
|
|
cur.execute("SELECT k FROM t_defer ORDER BY k")
|
|
assert len(cur.fetchall()) == 50, (
|
|
"the queued cleanup released our own statement"
|
|
)
|
|
finally:
|
|
cur._read_describe_response = original
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_stale_queued_cleanup_does_not_kill_the_connection(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
"""A queue entry goes stale whenever the cursor gets closed properly
|
|
between enqueue and flush. The server answers the leftover CLOSE with
|
|
-267, which is an OperationalError, which is in WIRE_ERRORS — so a
|
|
stale entry used to force-close a healthy connection."""
|
|
with _connect(conn_params) as conn:
|
|
cur = conn.cursor()
|
|
cur.execute("SELECT FIRST 1 tabid FROM systables")
|
|
cur.fetchall()
|
|
|
|
conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU])
|
|
cur.execute("SELECT FIRST 1 tabid FROM systables")
|
|
assert cur.fetchone() is not None, "stale cleanup killed the connection"
|
|
assert not conn.closed
|
|
assert conn._pending_cleanup == [], "queue should have drained"
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_connection_closes_cleanly_with_a_scroll_cursor_open(
|
|
conn_params: ConnParams,
|
|
) -> None:
|
|
conn = _connect(conn_params)
|
|
scroll = conn.cursor(scrollable=True)
|
|
scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
|
|
scroll.fetch_first()
|
|
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
|