A session has one statement slot and the driver acted as if it had many
SQLI gives a session a single statement slot. SQ_CLOSE, SQ_RELEASE and SQ_SFETCH 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. A scrollable cursor holds the slot open on purpose, and two things went wrong there without saying so. Another statement on the same connection got -285, and the scrollable cursor was collateral damage: its next fetch came back -267, "the transaction has been rolled back, all locks released". Two errors, neither naming the cause, from code that reads as completely ordinary -- iterate a large result set, run a lookup partway through. It is now a ProgrammingError that explains the constraint, and the scrollable cursor is left alone. Re-executing the same scrollable cursor is the one caller allowed past the check, and it now closes its own server-side cursor first, because it collided with itself too. The deferred-cleanup queue was flushed before every PDU. But a finalizer enqueues precisely because it lost the wire lock, meaning another thread is mid-statement -- so the flush landed inside that thread's own statement and released it. The victim saw -208 when it landed before the first fetch, -267 between fetch batches. Flushing only at a statement boundary is both correct and sufficient: cleanup that misses one statement is picked up by the next. A stale queue entry was fatal on top of that. The finalizer enqueues, the cursor is then closed properly, and the leftover CLOSE draws -267 -- an OperationalError, which is in WIRE_ERRORS, which force-closed a healthy connection. Server-reported errors are now told apart from wire failures by whether they carry a sqlcode. Noted while tracing this: _build_close_pdu and _build_release_pdu write SQ_ID followed by write_int(opcode), which frames correctly only because the four bytes happen to land as [statement_id=0][opcode]. JDBC's sendStatementID writes the real id there, and _read_describe_response has been parsing it into _statement_id all along without ever using it. Addressing CLOSE/RELEASE correctly is not enough for multiplexing on its own -- a second cursor's SFETCH addressed to its own id returns -259 -- so that stays open rather than half-done.
This commit is contained in:
parent
0d8bd57ba9
commit
ac00a25cb7
@ -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()
|
||||
|
||||
@ -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
|
||||
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
|
||||
|
||||
262
tests/test_statement_slot.py
Normal file
262
tests/test_statement_slot.py
Normal file
@ -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
|
||||
Loading…
x
Reference in New Issue
Block a user