diff --git a/src/informix_db/connections.py b/src/informix_db/connections.py index 0e50edc..9ab081f 100644 --- a/src/informix_db/connections.py +++ b/src/informix_db/connections.py @@ -18,6 +18,7 @@ import socket as socket_mod import ssl import struct import threading +import weakref from io import BytesIO from pathlib import Path @@ -47,7 +48,7 @@ from ._protocol import ( ) from ._socket import IfxSocket from .cursors import Cursor -from .exceptions import InterfaceError, OperationalError +from .exceptions import InterfaceError, OperationalError, ProgrammingError # Default capability bits the JDBC reference sends. Validated against # 01-connect-only.socat.log via the PDU diff in tests/test_pdu_match.py: @@ -269,6 +270,11 @@ class Connection: # under ``_wire_lock``. self._pending_cleanup: list[bytes] = [] self._cleanup_lock = threading.Lock() + # Weak ref to the scrollable cursor currently holding the + # session's statement slot, or None. Weak so that abandoning a + # scrollable cursor still lets its finalizer run — a strong ref + # here would keep the very object alive whose GC we depend on. + self._open_scroll_cursor: weakref.ref | None = None # Logged-DB transaction state: True iff there's an open server-side # transaction (SQ_BEGIN sent, not yet committed/rolled-back). The # cursor uses this to decide whether to send an implicit SQ_BEGIN @@ -363,22 +369,80 @@ class Connection: raise InterfaceError("connection is closed") return Cursor(self, scrollable=scrollable) - def _send_pdu(self, pdu: bytes) -> None: + def _send_pdu(self, pdu: bytes, *, statement_boundary: bool = False) -> None: """Send an assembled PDU. Used by Cursor. - Phase 29: opportunistically drains any pending cleanup PDUs - from the deferred-cleanup queue *before* sending the new PDU. - Caller must hold ``_wire_lock`` (every actual call site already - does — execute/executemany/_sfetch_at, commit, rollback, - fast_path_call, etc.). The drain happens under that lock so - the queued cleanup atomically completes before the next op. + Caller must hold ``_wire_lock`` (every actual call site does). + + ``statement_boundary=True`` additionally drains the deferred + cleanup queue first. **Only pass it when this PDU is the first + of a new statement**, meaning no statement is currently open + server-side. + + The queue previously drained before *every* PDU, which was wrong + in a way that took a repro to see. ``SQ_CLOSE`` and + ``SQ_RELEASE`` carry no statement identifier — they act on the + server's *current* statement. A finalizer enqueues precisely + because it lost the race for the wire lock, which means another + thread is mid-statement; that thread's next ``_send_pdu`` was + then its own ``CURNAME``/``NFETCH``, and the drain released the + statement out from under it. The caller saw a nonsense error on + valid SQL: ``-208`` when injected before the first fetch, + ``-267`` "transaction has been rolled back" between fetch + batches. Both point nowhere near the actual cause, and the + window is exactly the window in which enqueueing happens. + + Draining only at a boundary costs nothing: cleanup that misses + one statement is picked up by the next. """ if self._closed: raise InterfaceError("connection is closed") - if self._pending_cleanup: + if statement_boundary and self._pending_cleanup: self._drain_pending_cleanup() self._sock.write_all(pdu) + def _check_scroll_cursor_conflict(self, requester: object) -> None: + """Refuse to start a statement while a scrollable cursor is open. + + A server-side scrollable cursor occupies the session's statement + slot, and SQLI gives us no way to address around it: ``SQ_CLOSE``, + ``SQ_RELEASE`` and ``SQ_SFETCH`` all act on the session's current + statement. + + The server does not refuse politely. Starting another statement + returns ``-285``, and the scrollable cursor is collateral damage + — its next fetch comes back ``-267`` "the transaction has been + rolled back, all locks released". Two unattributable failures + from code that looks entirely ordinary: iterate a large result + set with a scrollable cursor, run a lookup query partway through. + + Multiplexing is presumably expressible — JDBC prefixes every + statement-scoped PDU with the statement id, which the server does + assign distinctly (0 and 1 for two concurrent cursors). But the + id alone is not sufficient: addressing the second cursor's + ``SQ_SFETCH`` to its own id returns ``-259`` "cursor not open". + Until that is understood, refusing is the honest behaviour. It + costs the caller a second connection and it never corrupts. + """ + ref = self._open_scroll_cursor + if ref is None: + return + other = ref() + if ( + other is None + or other is requester + or not getattr(other, "_server_cursor_open", False) + ): + self._open_scroll_cursor = None + return + raise ProgrammingError( + "a scrollable cursor is open on this connection; Informix " + "allows only one statement per session, so running another " + "statement here would fail with -285 and destroy the " + "scrollable cursor as well. Close the scrollable cursor " + "first, or use a separate connection for the other statement." + ) + def _enqueue_cleanup(self, pdus: list[bytes]) -> None: """Append cleanup PDUs to the deferred queue. @@ -418,11 +482,42 @@ class Connection: try: self._sock.write_all(pdu) self._drain_to_eot() - except WIRE_ERRORS: - # Wire is unrecoverable; force-close. Subsequent - # ``_send_pdu`` will raise InterfaceError. Server - # cleanup of the remaining queued entries happens - # implicitly at session end. + except Exception as exc: + if getattr(exc, "sqlcode", None) is not None: + # The *server* rejected the cleanup — a stale entry + # for a cursor it no longer has. Queued cleanup goes + # stale routinely: the finalizer enqueues, then the + # cursor gets closed properly before the drain runs. + # + # This is not a wire problem. ``_raise_sq_err`` + # self-drains the trailing SQ_EOT, so the wire is + # still aligned and the remaining entries are still + # worth sending. + # + # It must not escape, and it must not be treated as + # fatal. Both were wrong before: a stale CLOSE draws + # ``-267``, which is an OperationalError, which is in + # WIRE_ERRORS — so a stale queue entry force-closed a + # perfectly healthy connection. And this runs at the + # start of somebody else's statement, so letting it + # out would fail their good SQL with an error about a + # cursor they never opened. + _log.debug( + "deferred cleanup rejected by server (stale entry), " + "continuing: %r", + exc, + ) + continue + if not isinstance(exc, WIRE_ERRORS): + _log.warning( + "unexpected error draining deferred cleanup: %r", exc + ) + # No sqlcode means the failure came from the wire, not + # the server: the socket died, or framing desynced and we + # can no longer say where a response ends. Force-close. + # Subsequent ``_send_pdu`` raises InterfaceError. The + # server-side resources the remaining entries would have + # freed are released when the session ends anyway. self._closed = True with contextlib.suppress(Exception): self._sock.close() diff --git a/src/informix_db/cursors.py b/src/informix_db/cursors.py index 1221a21..cbedd56 100644 --- a/src/informix_db/cursors.py +++ b/src/informix_db/cursors.py @@ -335,6 +335,15 @@ class Cursor: def _execute_under_wire_lock(self, sql: str, params: tuple) -> None: """Wire-bound body of ``execute``. Caller MUST hold ``_wire_lock``.""" + # Someone else's scrollable cursor owns the session's statement + # slot -- refuse rather than draw a -285 and destroy their cursor. + self._conn._check_scroll_cursor_conflict(self) + # Our own previous scrollable cursor is still open server-side. + # Re-executing over the top of it collides exactly the same way, + # so close it first. This is the one caller allowed past the + # check above, and it is only safe because of this. + if self._scrollable and self._server_cursor_open: + self._close_server_cursor() # Reset previous-execute state. self._description = None self._columns = [] @@ -352,7 +361,12 @@ class Cursor: self._conn._ensure_transaction() # Step 1: PREPARE — send SQL with numQmarks = len(params). - self._conn._send_pdu(self._build_prepare_pdu(sql, num_qmarks=len(params))) + # statement_boundary: nothing is open server-side yet, so this is + # the one safe moment to flush a finalizer's deferred cleanup. + self._conn._send_pdu( + self._build_prepare_pdu(sql, num_qmarks=len(params)), + statement_boundary=True, + ) self._read_describe_response() # Branch on the SQL keyword. We can't use ``self._columns`` / @@ -381,6 +395,24 @@ class Cursor: if self._description is not None: self._row_index = -1 + def _close_server_cursor(self) -> None: + """Free the server-side scrollable cursor. Caller MUST hold ``_wire_lock``. + + Best-effort: a wire failure here is swallowed. Both callers are + already past the point where reporting it would help — one is + closing the cursor, the other is about to run a new statement + that will report its own failure if the wire is really gone. + """ + try: + self._conn._send_pdu(self._build_close_pdu()) + self._drain_to_eot() + self._conn._send_pdu(self._build_release_pdu()) + self._drain_to_eot() + except Exception: + pass + self._server_cursor_open = False + self._conn._open_scroll_cursor = None + def _release_after_failure(self, *, close_cursor: bool = False) -> None: """Best-effort server-side cleanup after a statement fails. @@ -480,6 +512,9 @@ class Cursor: raise self._server_cursor_open = True self._finalizer_state[0] = True # arm the GC-time fallback + # The connection needs to know its statement slot is taken, + # so the next statement can refuse instead of drawing -285. + self._conn._open_scroll_cursor = weakref.ref(self) self._scroll_total_rows = None return # don't close; cursor stays live for SQ_SFETCH # Phase 35: NFETCH loop — keep fetching until a response yields @@ -1018,6 +1053,9 @@ class Cursor: # under the wire lock — N rows commit atomically with respect # to other threads on the connection. with self._conn._wire_lock: + self._conn._check_scroll_cursor_conflict(self) + if self._scrollable and self._server_cursor_open: + self._close_server_cursor() # Reset per-execute state. self._description = None self._columns = [] @@ -1033,7 +1071,8 @@ class Cursor: # PREPARE once. self._conn._send_pdu( - self._build_prepare_pdu(sql, num_qmarks=first_len) + self._build_prepare_pdu(sql, num_qmarks=first_len), + statement_boundary=True, ) self._read_describe_response() @@ -1359,20 +1398,8 @@ class Cursor: if self._closed: return if self._scrollable and self._server_cursor_open: - # Phase 27: hold the wire lock during CLOSE+RELEASE so we - # don't interleave with another thread's pending op on the - # connection. Best-effort: any wire failure here is - # swallowed (the caller is closing; we don't want to mask - # whatever caused them to close). - try: - with self._conn._wire_lock: - self._conn._send_pdu(self._build_close_pdu()) - self._drain_to_eot() - self._conn._send_pdu(self._build_release_pdu()) - self._drain_to_eot() - except Exception: - pass - self._server_cursor_open = False + with self._conn._wire_lock: + self._close_server_cursor() # Phase 28: explicit close has handled the server-side resources # (or tried to). Disarm the finalizer so it doesn't fire later # for nothing — and clear the state flag as a belt-and-suspenders diff --git a/tests/test_statement_slot.py b/tests/test_statement_slot.py new file mode 100644 index 0000000..21ff125 --- /dev/null +++ b/tests/test_statement_slot.py @@ -0,0 +1,262 @@ +"""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