Release 2026.09.02 — a systematic review, and eleven bugs it found

No new features. The result of going back over the driver looking for
the shape of past bugs rather than for new symptoms.

Two returned wrong data silently: smart LOBs were read 77 bytes short,
shifting every column after them, and NULL composite UDTs skipped their
length field. One did nothing at all: rollback() after a SQL BEGIN WORK
returned successfully having sent nothing, leaving the rows it was asked
to discard in place and the connection back in the pool holding an open
transaction.

The rest: five ordinary query forms failing with -260 because
classification guessed from the first word; a scrollable cursor taking
the whole connection down with it; two unguarded exits leaking failed
statements; deferred cleanup landing inside somebody else's statement,
and a GC-time finalizer that couldn't tell it was already on the thread
holding the wire lock; the async layer borrowing the process-wide thread
pool; and two readers sharing one socket stream without agreeing.

457 integration tests on 15 and 14.10, 456 on 12.10 (one skip, no CTEs
before 14.10) -- up from 414. Every fix has at least one test that fails
against 2026.09.01.

CHANGELOG.md

@@ -2,6 +2,92 @@
 
 All notable changes to `informix-db`. Versioning is [CalVer](https://calver.org/) — `YYYY.MM.DD` for date-based releases, `YYYY.MM.DD.N` for same-day post-releases per PEP 440.
 
+## 2026.09.02 — A systematic review, and eleven bugs it found
+
+No new features. This is the result of going back over the driver looking for the *shape* of past bugs rather than for new symptoms, and it turned up more than expected — including two that silently returned wrong data and one where `rollback()` did nothing at all.
+
+### Rows never checked that they consumed their own payload
+
+Fourteen framing bugs had reached users before this release. Every one was the same defect — a column reading the wrong number of bytes — and not one of them raised where the mistake happened. The wrong width produced a plausible value and corrupted whatever came *next*, so the damage always surfaced in a different column, a different row, or a different statement.
+
+All fourteen were detectable for free. The payload is a fully extracted `bytes` of known length, so after decoding N columns the offset has to land exactly on the end. Now it's checked, on every row, in all three decode paths, and a mismatch names the column shape and the byte delta instead of handing back plausible garbage.
+
+Turning it on found two more bugs on the first run:
+
+**Smart LOBs were read as a flat 72-byte field.** They actually use the UDT envelope — 149 bytes when populated, 5 when NULL. The driver consumed 72 and left 77 behind, so a `BLOB` or `CLOB` anywhere but the final column position shifted every column after it. The 144 envelope bytes are the 72-byte locator *hex-encoded*, which means `BlobLocator.raw` had never held a locator, only the first half of the hex text. Nobody noticed because `read_blob_column` resolves LOBs server-side and never reads it.
+
+**NULL composite UDTs skipped their length field**, byte for byte the LVARCHAR NULL bug in a branch twelve lines away. Both are now the one `_read_udt_envelope`, which also rejects a negative length rather than rewinding into bytes it already decoded.
+
+### `rollback()` could silently do nothing
+
+`cursor.execute("BEGIN WORK")` is a reasonable thing to write, and with autocommit on it opened a real transaction that the connection never learned about. `commit()` and `rollback()` are both guarded by that flag, so **`rollback()` returned successfully having sent nothing, and the rows it was asked to discard survived.** The connection then went back to the pool holding an open transaction and its locks, because the pool's cleanup is guarded by the same flag.
+
+With autocommit off it failed instead, and for a sillier reason: the driver's implicit `SQ_BEGIN` fired first, so the caller's `BEGIN WORK` got `-535`, "already in transaction". The driver and the user competing to open the same transaction, and the user losing.
+
+The server labels these statements and always has. Both spellings of all three now update the connection's state, and the implicit begin is skipped when the caller is doing the job themselves.
+
+### Five ordinary query forms failed with a nonsense error
+
+The driver decided whether to open a cursor by checking if the first word of the SQL was `SELECT`. That gets wrong:
+
+```sql
+-- a leading comment                 →  -260
+/* of any of the three flavours */   →  -260
+{ that Informix accepts }            →  -260
+(SELECT parenthesized)               →  -260
+WITH cte AS (...) SELECT ...         →  -260
+(SELECT a) UNION (SELECT b)          →  -260
+```
+
+`-260` is "Cursor name already in use", which describes neither the cause nor anything the caller did. It says "cursor" because the driver sent `SQ_EXECUTE` where the server was waiting to open one.
+
+The server reports the statement type in the first field of every DESCRIBE response, and the driver had been parsing it into a metadata dict that nothing read. The decision now uses it. `EXECUTE FUNCTION` gets fixed as a side effect — it used to run down the DML path and discard its return value, and now yields it.
+
+### A scrollable cursor took the connection down with it
+
+Informix gives a session one statement slot. A scrollable cursor holds it open on purpose, so any other statement on that connection got `-285` — **and destroyed the scrollable cursor too**, whose next fetch came back `-267`, "the transaction has been rolled back, all locks released". Two unattributable failures from code that reads as entirely ordinary: iterate a large result set, run a lookup partway through.
+
+It is now a `ProgrammingError` that explains the constraint and leaves the cursor alone. Re-executing the *same* scrollable cursor is still allowed and now closes its own server-side cursor first, because it collided with itself as well.
+
+### Statements that failed were sometimes never released
+
+A failed statement stays allocated server-side and collides with the next `PREPARE`, after which every call on that connection returns a nonsense error pointing at the *previous* SQL. That guard existed at six exits and was missing from two, both easy to reach:
+
+- The parameterized-`SELECT` bind drain. Passing a string where the column is an `INT` encodes cleanly, so the rejection comes from the server (`-1213`, `-415`) *after* the guarded step.
+- Opening a scrollable cursor — no guard at all, and the worst place to lack one, since the GC-time fallback is armed on the line after.
+
+All six now share one implementation, which also fixed a defect in the copies that had a cursor to close: `CLOSE` and `RELEASE` shared a single suppression, so a failing `CLOSE` skipped the `RELEASE` — losing the half that actually matters.
+
+### Cleanup could land inside somebody else's statement
+
+`SQ_CLOSE` and `SQ_RELEASE` act on whatever statement is current; neither names one. Two consequences.
+
+A cursor finalizer that can't take the wire lock queues its cleanup for the next operation to flush — but that queue was flushed before *every* PDU, and a finalizer queues precisely because another thread is mid-statement. The flush landed inside that thread's own statement and released it (`-208` before the first fetch, `-267` between fetch batches). Flushing only at a statement boundary is both correct and sufficient.
+
+And the finalizer's own lock probe couldn't see its own thread. `RLock.acquire(blocking=False)` grants a reentrant acquire to the owner, so when GC fired on a thread that was mid-statement, the finalizer concluded it had exclusive access and sent `CLOSE`/`RELEASE` into the running query. Refcounting hides this — a dropped cursor is freed at the drop — but a cursor caught in a reference cycle waits for a collection, and cycles are ordinary. Any traceback holding a cursor makes one.
+
+A stale queue entry was separately fatal: the leftover `CLOSE` draws `-267`, an `OperationalError`, which was treated as a dead wire and force-closed a perfectly healthy connection.
+
+### The async layer borrowed the whole process's thread pool
+
+Every blocking call went through `asyncio.to_thread`, which runs on the event loop's default executor — sized `min(32, cpu_count + 4)`, so six threads on a two-CPU container.
+
+`asyncio.to_thread` cannot interrupt a worker, so a cancelled await leaves the thread running the wire call until the read timeout. Cancellation is ordinary in a web app; a client disconnect cancels the request task. Measured: six cancelled calls against a six-worker default executor starve an unrelated `to_thread` indefinitely, and with the executor held, **four concurrent driver queries never ran at all**. Pool concurrency was capped by the same unrelated number — `max_size=20` on a two-CPU box ran six queries at a time.
+
+Each connection now owns one thread. That's the right size rather than a compromise, since the connection serializes every wire operation on its own lock anyway, and it avoids a deadlock a shared pool-sized executor invites.
+
+### Two readers shared one stream without agreeing
+
+`IfxSocket` owns a read-ahead buffer that the buffered reader fills, while the login path and the connection-level drain read straight from the socket without looking at it. Buffered bytes would be skipped, and skipped bytes in a length-framed protocol don't announce themselves.
+
+Nothing triggers it today, because the server sends one response per request. That's a property of the traffic, not of the code — pipelined `executemany` already puts several responses in flight, and the buffer is connection-scoped precisely so read-ahead *can* cross response boundaries. The two paths now agree by construction.
+
+Also here: `fill_recv_buf` believed whatever byte count it was handed, and that count is almost always a length field straight off the wire — a garbage `0x7FFFFFFF` read as a 2 GB request and sat in `recv` until the read timeout. It now refuses above `IFX_MAX_READ_BYTES` (256 MiB default) with an error naming the number.
+
+### Verified
+
+**457** integration tests on 15.0.1.0.3DE and 14.10.FC7W1DE, **456** on 12.10.FC12W1DE (one skip — 12.10 has no CTEs) — up from 414. Every fix has at least one test that fails against the previous release.
+
 ## 2026.09.01 — TLS traffic fuzzed; no bugs found
 
 Tests and docs only — no behaviour change. TLS was the last untested surface, and it came back clean.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "informix-driver"
-version = "2026.09.01"
+version = "2026.09.02"
 description = "Pure-Python driver for IBM Informix IDS — speaks the SQLI wire protocol over raw sockets. No CSDK, no JVM, no native libraries."
 readme = "README.md"
 license = { text = "MIT" }

src/informix_db/_protocol.py

@@ -29,6 +29,29 @@ class ProtocolError(Exception):
     """Raised when wire bytes can't be parsed (truncated stream, bad framing)."""
 
 
+# Every way a wire read/write can fail, as one tuple.
+#
+# This exists because the tuple was written out by hand in six places and
+# drifted: four of them caught ``(ProtocolError, OSError)`` — which
+# ``IfxSocket`` never raises. It converts *every* socket failure, including
+# clean EOF, into ``OperationalError``, and that is a ``DatabaseError``, not
+# an ``OSError``. So those four handlers could not catch the thing they
+# existed to catch. The worst of them sat in a ``weakref.finalize`` callback,
+# where the escaping exception is printed to stderr and swallowed, leaving a
+# desynchronised connection to be returned to the pool marked healthy.
+#
+# Import this rather than re-typing the members. A tuple in one place cannot
+# drift out of sync with itself.
+def _wire_error_types() -> tuple[type[BaseException], ...]:
+    # Imported lazily: exceptions.py must stay free of protocol imports.
+    from .exceptions import InterfaceError, OperationalError
+
+    return (ProtocolError, OSError, OperationalError, InterfaceError)
+
+
+WIRE_ERRORS: tuple[type[BaseException], ...] = _wire_error_types()
+
+
 # ---------------------------------------------------------------------------
 # Writer
 # ---------------------------------------------------------------------------
@@ -269,6 +292,13 @@ class BufferedSocketReader(IfxStreamReader):
         return b
 
     def skip(self, n: int) -> None:
+        # The base reader's skip delegates to read_exact, which returns
+        # b"" for n <= 0. This one advances the cursor arithmetically, so
+        # without the guard a negative n *rewinds* it — silently
+        # re-decoding bytes already consumed as if they were the next
+        # field. Same divergence as read_exact, which does guard.
+        if n <= 0:
+            return
         sock = self._sock
         sock.fill_recv_buf(n)
         sock._recv_pos += n

src/informix_db/_resultset.py

@@ -25,7 +25,7 @@ from dataclasses import dataclass
 from datetime import timedelta as _timedelta
 from types import MappingProxyType
 
-from ._protocol import IfxStreamReader
+from ._protocol import IfxStreamReader, ProtocolError
 from ._types import IfxType, base_type, is_nullable
 from .converters import (
     _DOUBLE_NULL,
@@ -248,6 +248,114 @@ _COMPOSITE_UDT_TYPES = frozenset({
 _NUMERIC_TYPES = frozenset({_TC_DECIMAL, _TC_MONEY})
 
 
+def _read_udt_envelope(payload: bytes, offset: int) -> tuple[int, bytes | None]:
+    """Read the UDT wire envelope: ``[1-byte indicator][int32 length][data]``.
+
+    Returns ``(new_offset, body)`` with ``body`` ``None`` when the
+    indicator says NULL. The length field is present either way — it
+    belongs to the envelope, not the value — so the offset advances by
+    5 for a NULL and 5 + length otherwise.
+
+    This exists because the same envelope was hand-written four times and
+    three of the copies were wrong, each in a different way:
+
+    * BOOLEAN read ``encoded_length`` (1 byte) instead of the envelope,
+      leaving 5 bytes behind.
+    * The composite-UDT branch returned on the indicator without reading
+      the length, leaving 4 bytes behind.
+    * BLOB/CLOB read a flat 72 bytes, leaving 77 behind.
+
+    Only the UDTVAR(lvarchar) copy was right, and only after two separate
+    fixes. Four copies of one three-line format is four chances to get it
+    wrong; this is one.
+    """
+    indicator = payload[offset]
+    offset += 1
+    length = int.from_bytes(payload[offset : offset + 4], "big", signed=True)
+    offset += 4
+    if indicator == 1:
+        return offset, None
+    if length < 0:
+        raise ProtocolError(
+            f"UDT envelope declared a negative length ({length}) at payload "
+            f"offset {offset - 4}; the wire is desynchronised"
+        )
+    body = payload[offset : offset + length]
+    offset += length
+    return offset, body
+
+
+def _decode_lob_locator(body: bytes, extended_id: int):
+    """Turn a smart-LOB envelope body into a Blob/ClobLocator.
+
+    The body is the 72-byte locator **hex-encoded as 144 ASCII characters**,
+    not the raw locator. We previously consumed a flat 72 bytes, which took
+    the first half of that hex text and handed it to ``BlobLocator`` as if
+    it were binary — so ``BlobLocator.raw`` never actually held a locator.
+    It went unnoticed because ``read_blob_column`` resolves LOBs through
+    the server-side ``lotofile`` function and never uses the locator.
+    """
+    cls = BlobLocator if extended_id == 10 else ClobLocator
+    try:
+        raw = bytes.fromhex(body.decode("ascii"))
+    except (ValueError, UnicodeDecodeError):
+        # Not hex — surface what arrived rather than guessing, but only if
+        # it is already the right size for a locator.
+        raw = bytes(body)
+    if len(raw) != 72:
+        raise ProtocolError(
+            f"smart-LOB locator decoded to {len(raw)} bytes, expected 72 "
+            f"(envelope body was {len(body)} bytes)"
+        )
+    return cls(raw=raw)
+
+
+def _row_not_consumed(
+    offset: int, payload_len: int, columns: list[ColumnInfo]
+) -> None:
+    """Raise when a decoded row didn't consume exactly its payload.
+
+    THIS IS THE CHECK THAT WOULD HAVE CAUGHT EVERY FRAMING BUG WE SHIPPED.
+
+    Thirteen of them reached users, and all thirteen were the same defect:
+    a column consumed the wrong number of bytes. Not one raised at the
+    point of the mistake — the wrong width produced a plausible value and
+    corrupted whatever came *next*, so the damage surfaced in a different
+    column, a different row, or a different statement.
+
+    Every one was detectable here for free. ``payload`` is a fully
+    extracted ``bytes`` object of known length (the wire's even-alignment
+    pad is consumed separately by the caller), so after decoding N columns
+    the offset must land exactly on the end. Anything else means the codec
+    and the server disagree about the row's shape, and continuing can only
+    produce wrong answers.
+
+    A previous attempt at this check was reverted, and the comment
+    explaining why survived long after its reason did: it cited the
+    LVARCHAR odd-length pad, which was itself a bug and has since been
+    removed. With that gone the invariant is exact.
+
+    The message names the column shape because "row decode failed" is not
+    actionable — the whole point is to say *which* column shape and by how
+    many bytes, so the next report arrives with the answer in it.
+    """
+    delta = offset - payload_len
+    direction = "over-read" if delta > 0 else "under-read"
+    shape = ", ".join(
+        f"{c.name}:tc={c.type_code}"
+        f"{'/ext=' + c.extended_name if c.extended_name else ''}"
+        f"/enclen={c.encoded_length}"
+        for c in columns
+    )
+    raise ProtocolError(
+        f"row decoder {direction} by {abs(delta)} byte(s): consumed "
+        f"{offset} of {payload_len} payload bytes. This means the driver "
+        f"and the server disagree about a column's wire width, so every "
+        f"column after the offending one is unreliable. Column shape: "
+        f"[{shape}]"
+    )
+
+
 def _packed_width(encoded_length: int) -> int:
     """On-wire byte width for the four types whose ``encoded_length``
     packs two fields into ``(high << 8) | low``: DECIMAL, MONEY,
@@ -540,6 +648,12 @@ def compile_row_decoder(
             # Unknown kind — abort codegen, caller falls back.
             return None
 
+    # Same end-of-row reconciliation the interpreted paths do. The
+    # generated function is the hot path, so this must be emitted here
+    # too — a check that only guards the slow path guards nothing in
+    # production.
+    lines.append("    if offset != len(payload):")
+    lines.append("        _row_short(offset, len(payload))")
     if val_names:
         lines.append(f"    return ({', '.join(val_names)},)")
     else:
@@ -564,6 +678,13 @@ def compile_row_decoder(
         "_decode_datetime": _decode_datetime,
         "_decode_interval": _decode_interval,
         "_legacy_dispatch_one_column": _legacy_dispatch_one_column,
+        # Bound to this shape's columns so the raised message can name them
+        # without the generated source having to carry the list itself.
+        "_row_short": (
+            lambda offset, payload_len, _cols=columns: _row_not_consumed(
+                offset, payload_len, _cols
+            )
+        ),
         "_UNPACK_SHORT": _UNPACK_SHORT,
         "_UNPACK_INT": _UNPACK_INT,
         "_UNPACK_LONG": _UNPACK_LONG,
@@ -610,11 +731,16 @@ def _legacy_dispatch_one_column(
     """
     # BLOB / CLOB locator (UDTFIXED + extended_id 10/11)
     if tc == _TC_UDTFIXED and col.extended_id in (10, 11):
-        width = col.encoded_length
-        raw = payload[offset:offset + width]
-        offset += width
-        cls = BlobLocator if col.extended_id == 10 else ClobLocator
-        return offset, cls(raw=bytes(raw))
+        # Smart LOBs use the UDT envelope, NOT a flat encoded_length field.
+        # Measured on Informix 15: a populated BLOB column is 149 bytes
+        # ([ind=0][len=144][144 hex chars]) and a NULL one is 5, while
+        # encoded_length reports 72. Consuming 72 left 77 bytes behind and
+        # corrupted every column after a LOB. Found by the end-of-row
+        # reconciliation check, which fired on its first run.
+        offset, body = _read_udt_envelope(payload, offset)
+        if body is None:
+            return offset, None
+        return offset, _decode_lob_locator(body, col.extended_id)
 
     # BOOLEAN. The server describes it as UDTFIXED (41) with
     # extended_name='boolean' and encoded_length=1, but ``encoded_length``
@@ -628,24 +754,28 @@ def _legacy_dispatch_one_column(
     if tc == _TC_UDTFIXED and (
         col.extended_name == "boolean" or col.extended_id == 5
     ):
-        indicator = payload[offset]
-        offset += 1
-        length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
-        offset += 4
-        raw = payload[offset:offset + length]
-        offset += length
-        if indicator == 1:
+        offset, body = _read_udt_envelope(payload, offset)
+        if body is None:
             return offset, None
-        return offset, bool(raw and raw[0] in (ord("t"), ord("T"), 1))
+        return offset, bool(body and body[0] in (ord("t"), ord("T"), 1))
 
     # ROW / COLLECTION composite UDT
     if tc in _COMPOSITE_UDT_TYPES:
         indicator = payload[offset]
         offset += 1
-        if indicator == 1:
-            return offset, None
+        # The 4-byte length is part of the envelope and is present even when
+        # the indicator says NULL — identical to the UDTVAR(lvarchar) branch
+        # below, which decodes the same `[ind][int32 len][data]` shape.
+        # Returning early on the indicator left those 4 bytes unread and
+        # corrupted the following column. Wire evidence, Informix 15,
+        # (INT, SET(INT), VARCHAR) with a NULL set:
+        #     00 00 00 08 | 01 | 00 00 00 00 | 03 78 79 7a
+        #        a = 8    | ind |  length = 0 |  [3]"xyz"
+        # Before the fix the VARCHAR decoded as '' instead of 'xyz'.
         length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
         offset += 4
+        if indicator == 1:
+            return offset, None
         raw = bytes(payload[offset:offset + length])
         offset += length
         if tc == _TC_ROW:
@@ -830,6 +960,8 @@ def parse_tuple_payload(
                 payload, offset, tc, col, encoding
             )
             values.append(value)
+        if offset != len(payload):
+            _row_not_consumed(offset, len(payload), columns)
         return tuple(values)
 
     # Legacy slow path (no pre-compiled readers).
@@ -936,11 +1068,13 @@ def parse_tuple_payload(
         # we read here are an opaque server-side reference, NOT the
         # actual data. Phase 10 lets users fetch via lotofile + SQ_FILE.
         if tc == _TC_UDTFIXED and col.extended_id in (10, 11):
-            width = col.encoded_length
-            raw = payload[offset:offset + width]
-            offset += width
-            cls = BlobLocator if col.extended_id == 10 else ClobLocator
-            values.append(cls(raw=bytes(raw)))
+            # UDT envelope, not a flat encoded_length field — see the
+            # matching branch in _legacy_dispatch_one_column.
+            offset, body = _read_udt_envelope(payload, offset)
+            values.append(
+                None if body is None
+                else _decode_lob_locator(body, col.extended_id)
+            )
             continue
 
         # BOOLEAN — UDT envelope, not a bare byte. See the matching branch
@@ -948,18 +1082,11 @@ def parse_tuple_payload(
         if tc == _TC_UDTFIXED and (
             col.extended_name == "boolean" or col.extended_id == 5
         ):
-            indicator = payload[offset]
-            offset += 1
-            length = int.from_bytes(
-                payload[offset:offset + 4], "big", signed=True
+            offset, body = _read_udt_envelope(payload, offset)
+            values.append(
+                None if body is None
+                else bool(body and body[0] in (ord("t"), ord("T"), 1))
             )
-            offset += 4
-            raw = payload[offset:offset + length]
-            offset += length
-            if indicator == 1:
-                values.append(None)
-            else:
-                values.append(bool(raw and raw[0] in (ord("t"), ord("T"), 1)))
             continue
 
         # ROW / COLLECTION (Phase 12): composite UDTs. Wire format is
@@ -976,13 +1103,15 @@ def parse_tuple_payload(
         if tc in _COMPOSITE_UDT_TYPES:
             indicator = payload[offset]
             offset += 1
-            if indicator == 1:  # null
-                values.append(None)
-                continue
+            # Length is present even when NULL — see the matching branch in
+            # _legacy_dispatch_one_column for the wire evidence.
             length = int.from_bytes(
                 payload[offset:offset + 4], "big", signed=True
             )
             offset += 4
+            if indicator == 1:  # null
+                values.append(None)
+                continue
             raw = bytes(payload[offset:offset + length])
             offset += length
             if tc == _TC_ROW:
@@ -1048,4 +1177,6 @@ def parse_tuple_payload(
     # by Python's slicing semantics for strings — short = harmless).
     # If a future protocol message produces actual garbage here, add a
     # branch-local check at the offending dispatch path.
+    if offset != len(payload):
+        _row_not_consumed(offset, len(payload), columns)
     return tuple(values)

src/informix_db/_socket.py

@@ -17,11 +17,35 @@ the rest of the protocol layer.
 from __future__ import annotations
 
 import contextlib
+import os
 import socket
 import ssl
 
+from ._protocol import ProtocolError
 from .exceptions import InterfaceError, OperationalError
 
+
+def _max_read_bytes() -> int:
+    """Ceiling for a single length-prefixed read, from IFX_MAX_READ_BYTES.
+
+    256 MiB by default: comfortably above any row, string table, or blob
+    chunk a real query produces, and far below the values a desynced
+    stream invents. A genuinely larger single value is possible (a very
+    large TEXT column read in one go), which is why the knob exists.
+    """
+    raw = os.environ.get("IFX_MAX_READ_BYTES")
+    if raw:
+        try:
+            value = int(raw)
+        except ValueError:
+            value = 0
+        if value > 0:
+            return value
+    return 256 * 1024 * 1024
+
+
+MAX_READ_BYTES = _max_read_bytes()
+
 # A ``tls`` parameter to ``IfxSocket`` accepts:
 #   False (default) — plain TCP
 #   True            — TLS with verification disabled (dev / self-signed)
@@ -124,10 +148,42 @@ class IfxSocket:
             raise OperationalError(f"write failed: {e}") from e
 
     def read_exact(self, n: int) -> bytes:
-        """Read exactly ``n`` bytes or raise on EOF / timeout."""
+        """Read exactly ``n`` bytes or raise on EOF / timeout.
+
+        Consumes the read-ahead buffer before touching the socket. This
+        matters because two readers share one stream: ``BufferedSocketReader``
+        fills ``_recv_buf`` (over-reading by design, up to ``_recv_size``),
+        while ``Connection._drain_to_eot``, ``_raise_sq_err`` and the
+        login path call this method directly. Recv'ing here while bytes
+        sat unconsumed in the buffer would skip them, and skipped bytes
+        in a length-framed protocol don't announce themselves — the next
+        read lands mid-field and every subsequent one is wrong.
+
+        No workload triggers it today: the server sends one response per
+        request, so recv returns exactly that response and the buffered
+        reader consumes all of it before control returns here. That is a
+        property of the traffic, not of the code. Pipelined executemany
+        already puts multiple responses in flight, and the buffer is
+        connection-scoped precisely so read-ahead can cross response
+        boundaries. Making the two paths agree by construction costs one
+        branch on a cold path.
+        """
         if self._sock is None:
             raise InterfaceError("socket is closed")
+        if n <= 0:
+            return b""
+        wanted = n
         chunks: list[bytes] = []
+        buffered = len(self._recv_buf) - self._recv_pos
+        if buffered > 0:
+            take = min(buffered, n)
+            chunks.append(
+                bytes(self._recv_buf[self._recv_pos : self._recv_pos + take])
+            )
+            self._recv_pos += take
+            n -= take
+            if n == 0:
+                return chunks[0]
         remaining = n
         while remaining > 0:
             try:
@@ -138,7 +194,8 @@ class IfxSocket:
             if not chunk:
                 self._force_close()
                 raise OperationalError(
-                    f"server closed connection mid-read (wanted {n} bytes, got {n - remaining})"
+                    f"server closed connection mid-read "
+                    f"(wanted {wanted} bytes, got {wanted - remaining})"
                 )
             chunks.append(chunk)
             remaining -= len(chunk)
@@ -155,6 +212,21 @@ class IfxSocket:
         """
         if self._sock is None:
             raise InterfaceError("socket is closed")
+        if need > MAX_READ_BYTES:
+            # ``need`` is almost always a length field straight off the
+            # wire. Trusting it means a corrupt or desynced stream turns
+            # into an allocation of whatever that field happened to say —
+            # a garbage 0x7FFFFFFF reads as a 2 GB request, and the loop
+            # below sits in recv until the read timeout while the buffer
+            # grows. Refusing names the number instead, which is the
+            # diagnostic: a length that absurd means framing is already
+            # lost upstream, not that the row is genuinely that big.
+            raise ProtocolError(
+                f"refusing to read {need} bytes in one field (limit "
+                f"{MAX_READ_BYTES}). Either the wire has desynced, or a "
+                f"single value really is this large — raise the limit "
+                f"with the IFX_MAX_READ_BYTES environment variable."
+            )
         avail = len(self._recv_buf) - self._recv_pos
         if avail >= need:
             return

src/informix_db/aio.py

@@ -63,7 +63,9 @@ import asyncio
 import contextlib
 import functools
 import threading
+import weakref
 from collections.abc import AsyncIterator, Awaitable, Callable
+from concurrent.futures import ThreadPoolExecutor
 from typing import Any, TypeVar
 
 from . import connect as _sync_connect
@@ -80,6 +82,61 @@ def _to_thread(fn: Callable[..., T], *args: Any, **kwargs: Any) -> Awaitable[T]:
     return asyncio.to_thread(fn, *args, **kwargs)
 
 
+def _run_on(
+    executor: ThreadPoolExecutor,
+    fn: Callable[..., T],
+    *args: Any,
+    **kwargs: Any,
+) -> Awaitable[T]:
+    """Await ``fn`` on a specific executor rather than the loop's default."""
+    loop = asyncio.get_running_loop()
+    return loop.run_in_executor(executor, functools.partial(fn, *args, **kwargs))
+
+
+def _connection_executor(conn: _SyncConnection) -> ThreadPoolExecutor:
+    """The dedicated worker thread for one connection, created on demand.
+
+    ``asyncio.to_thread`` runs on the event loop's default executor, which
+    the whole process shares and which is sized from the CPU count —
+    ``min(32, cpu_count + 4)``, so six threads on a two-CPU container.
+    Two consequences, both silent.
+
+    A cancelled await does not stop its worker. ``asyncio.to_thread``
+    cannot interrupt the thread, so it keeps running the wire call until
+    the read timeout expires. Cancellation is ordinary in a web app — a
+    client disconnect cancels the request task — so a handful of them
+    pins every thread in the shared pool and unrelated ``to_thread`` work
+    anywhere else in the process stops dead. Measured: six cancelled
+    calls against a six-worker default executor starve an unrelated
+    ``to_thread`` indefinitely.
+
+    And pool concurrency was capped by the same number without saying so.
+    A pool with ``max_size=20`` on a two-CPU box ran six queries at once.
+
+    One thread per connection is the right size rather than a compromise:
+    the sync connection serializes every wire operation on its own lock,
+    so a second thread could do nothing but wait for the first. It also
+    rules out a deadlock that a shared pool-sized executor invites — with
+    N threads and N connections, N tasks blocked in ``acquire`` occupy
+    every thread while the connection they are waiting for is held by a
+    task that now needs a thread of its own to finish and release it.
+
+    The executor lives on the sync connection so it survives being
+    returned to the pool and handed out again. The finalizer is the
+    backstop for a connection dropped without ``close()``: a
+    ``ThreadPoolExecutor`` that is never shut down leaves its worker
+    parked on the work queue for the life of the process.
+    """
+    executor = getattr(conn, "_async_executor", None)
+    if executor is None:
+        executor = ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="informix-conn"
+        )
+        conn._async_executor = executor
+        weakref.finalize(conn, executor.shutdown, wait=False)
+    return executor
+
+
 class AsyncCursor:
     """Async wrapper over a sync :class:`Cursor`. Each I/O call awaits
     a thread-offloaded version of the sync operation.
@@ -89,10 +146,14 @@ class AsyncCursor:
     paying the thread-hop cost.
     """
 
-    __slots__ = ("_cur",)
+    __slots__ = ("_cur", "_run")
 
-    def __init__(self, cur: _SyncCursor):
+    def __init__(self, cur: _SyncCursor, run: Callable[..., Awaitable[Any]]):
         self._cur = cur
+        # The owning connection's runner, so cursor I/O lands on that
+        # connection's dedicated thread rather than the shared default
+        # executor. See _connection_executor.
+        self._run = run
 
     # -- Pass-through synchronous attributes (no I/O) ---------------------
 
@@ -121,32 +182,32 @@ class AsyncCursor:
     async def execute(
         self, operation: str, parameters: Any = None
     ) -> None:
-        await _to_thread(self._cur.execute, operation, parameters)
+        await self._run(self._cur.execute, operation, parameters)
 
     async def executemany(
         self, operation: str, seq_of_parameters: Any
     ) -> None:
-        await _to_thread(
+        await self._run(
             self._cur.executemany, operation, list(seq_of_parameters)
         )
 
     async def fetchone(self) -> tuple | None:
-        return await _to_thread(self._cur.fetchone)
+        return await self._run(self._cur.fetchone)
 
     async def fetchmany(self, size: int | None = None) -> list[tuple]:
-        return await _to_thread(self._cur.fetchmany, size)
+        return await self._run(self._cur.fetchmany, size)
 
     async def fetchall(self) -> list[tuple]:
-        return await _to_thread(self._cur.fetchall)
+        return await self._run(self._cur.fetchall)
 
     async def close(self) -> None:
-        await _to_thread(self._cur.close)
+        await self._run(self._cur.close)
 
     # Phase 10/11 BLOB helpers (preserve the sync API surface)
     async def read_blob_column(
         self, sql: str, params: tuple = ()
     ) -> bytes | None:
-        return await _to_thread(self._cur.read_blob_column, sql, params)
+        return await self._run(self._cur.read_blob_column, sql, params)
 
     async def write_blob_column(
         self,
@@ -156,7 +217,7 @@ class AsyncCursor:
         *,
         clob: bool = False,
     ) -> None:
-        await _to_thread(
+        await self._run(
             functools.partial(
                 self._cur.write_blob_column,
                 sql, blob_data, params, clob=clob,
@@ -178,14 +239,23 @@ class AsyncCursor:
 class AsyncConnection:
     """Async wrapper over a sync :class:`Connection`."""
 
-    __slots__ = ("_conn",)
+    __slots__ = ("_conn", "_executor")
 
     def __init__(self, conn: _SyncConnection):
         self._conn = conn
+        self._executor = _connection_executor(conn)
+
+    def _run(self, fn: Callable[..., T], *args: Any, **kwargs: Any) -> Awaitable[T]:
+        """Run a blocking connection call on this connection's own thread."""
+        return _run_on(self._executor, fn, *args, **kwargs)
 
     @classmethod
     async def connect(cls, *args: Any, **kwargs: Any) -> AsyncConnection:
-        """Open a connection. Same parameters as :func:`informix_db.connect`."""
+        """Open a connection. Same parameters as :func:`informix_db.connect`.
+
+        The connect itself still goes to the default executor -- there is
+        no connection yet to own a thread, and it is one bounded call.
+        """
         sync_conn = await _to_thread(
             functools.partial(_sync_connect, *args, **kwargs)
         )
@@ -196,22 +266,46 @@ class AsyncConnection:
         return self._conn.closed
 
     async def cursor(self) -> AsyncCursor:
-        sync_cur = await _to_thread(self._conn.cursor)
-        return AsyncCursor(sync_cur)
+        sync_cur = await self._run(self._conn.cursor)
+        return AsyncCursor(sync_cur, self._run)
 
     async def commit(self) -> None:
-        await _to_thread(self._conn.commit)
+        await self._run(self._conn.commit)
 
     async def rollback(self) -> None:
-        await _to_thread(self._conn.rollback)
+        await self._run(self._conn.rollback)
 
     async def close(self) -> None:
-        await _to_thread(self._conn.close)
+        """Close the connection and stop the thread that served it.
+
+        One thread per connection is only affordable if the thread goes
+        away with the connection. ``ThreadPoolExecutor`` workers park on
+        the work queue rather than exiting when idle, so an executor that
+        is never shut down leaks its thread for the life of the process —
+        the ``weakref.finalize`` in ``_connection_executor`` is a backstop
+        for connections dropped without ``close()``, not a substitute for
+        closing here.
+
+        ``wait=False`` because we are on the event loop: the worker has
+        just finished the close and needs no waiting, and blocking the
+        loop to confirm that would be the one thing this module exists to
+        avoid.
+        """
+        try:
+            await self._run(self._conn.close)
+        finally:
+            self._executor.shutdown(wait=False)
+            # Drop the reference too. A shut-down executor rejects new
+            # work, so leaving it attached would turn any later wrap of
+            # this sync connection into a RuntimeError rather than a
+            # fresh thread.
+            with contextlib.suppress(AttributeError):
+                del self._conn._async_executor
 
     async def fast_path_call(
         self, signature: str, *params: object
     ) -> list[object]:
-        return await _to_thread(self._conn.fast_path_call, signature, *params)
+        return await self._run(self._conn.fast_path_call, signature, *params)
 
     # Async context-manager support
     async def __aenter__(self) -> AsyncConnection:
@@ -304,8 +398,15 @@ class AsyncConnectionPool:
     async def release(
         self, conn: AsyncConnection, *, broken: bool = False
     ) -> None:
-        await _to_thread(
-            functools.partial(self._pool.release, conn._conn, broken=broken)
+        # On the connection's own thread, not the default executor. That
+        # thread is idle by definition — the caller is done with the
+        # connection — and it keeps the release off a shared pool that
+        # tasks blocked in ``acquire`` may have filled. Release has to
+        # win that race: it is what frees the connection they are
+        # waiting for.
+        await _run_on(
+            conn._executor,
+            functools.partial(self._pool.release, conn._conn, broken=broken),
         )
 
     @contextlib.asynccontextmanager

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
 
@@ -38,10 +39,16 @@ from ._messages import (
     SLHeader,
     StmtOptions,
 )
-from ._protocol import IfxStreamReader, IfxStreamWriter, ProtocolError, make_pdu_writer
+from ._protocol import (
+    WIRE_ERRORS,
+    IfxStreamReader,
+    IfxStreamWriter,
+    ProtocolError,
+    make_pdu_writer,
+)
 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:
@@ -199,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.
 
@@ -243,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
@@ -263,6 +326,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
@@ -357,22 +425,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.
 
@@ -402,7 +528,6 @@ class Connection:
         discarded; the server-side resources they would have released
         are freed when the session ends anyway.
         """
-        from ._protocol import ProtocolError
 
         with self._cleanup_lock:
             if not self._pending_cleanup:
@@ -413,11 +538,42 @@ class Connection:
             try:
                 self._sock.write_all(pdu)
                 self._drain_to_eot()
-            except (ProtocolError, OSError, OperationalError):
-                # 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()
@@ -596,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:`"
@@ -835,7 +991,6 @@ class Connection:
           [short near_token_len][bytes name][optional pad][short SQ_EOT]
         """
         from . import _errcodes
-        from ._protocol import ProtocolError
 
         sqlcode = struct.unpack("!h", self._sock.read_exact(2))[0]
         isamcode = struct.unpack("!h", self._sock.read_exact(2))[0]
@@ -851,7 +1006,7 @@ class Connection:
                 if name_len & 1:
                     self._sock.read_exact(1)
                 near_token = raw.rstrip(b"\x00").decode("iso-8859-1", errors="replace")
-        except (ProtocolError, OSError):
+        except WIRE_ERRORS:
             pass
         # Phase 28: drain failure means wire desync — force-close so
         # subsequent operations don't inherit the broken state.
@@ -865,7 +1020,7 @@ class Connection:
                 next_tag = struct.unpack("!h", self._sock.read_exact(2))[0]
                 if next_tag == MessageType.SQ_EOT:
                     break
-        except (ProtocolError, OSError, OperationalError):
+        except WIRE_ERRORS:
             self._closed = True
             with contextlib.suppress(Exception):
                 self._sock.close()

src/informix_db/cursors.py

@@ -30,7 +30,12 @@ from typing import TYPE_CHECKING, Any
 
 from . import _errcodes
 from ._messages import MessageType
-from ._protocol import BufferedSocketReader, IfxStreamReader, make_pdu_writer
+from ._protocol import (
+    WIRE_ERRORS,
+    BufferedSocketReader,
+    IfxStreamReader,
+    make_pdu_writer,
+)
 from ._resultset import (
     ColumnInfo,
     compile_column_readers,
@@ -97,6 +102,48 @@ def _make_socket_reader(sock):
     return _SocketReader(sock)
 
 
+# Statement types from the DESCRIBE response's first field. The server
+# tells us exactly what it prepared; these are the two values that mean
+# "this will produce rows".
+_ST_SELECT = 2
+_ST_ROUTINE = 56  # EXECUTE PROCEDURE / EXECUTE FUNCTION
+
+# Transaction control run as SQL. JDBC reads the same three values off
+# the describe and calls setTxBeginState / setTxEndState (IfxSqli, the
+# TxStmt field). Both ``BEGIN`` and ``BEGIN WORK`` report 34; likewise
+# the WORK-less spellings of the other two.
+_ST_TX_BEGIN = 34
+_ST_TX_COMMIT = 35
+_ST_TX_ROLLBACK = 36
+_TX_CONTROL_TYPES = frozenset({_ST_TX_BEGIN, _ST_TX_COMMIT, _ST_TX_ROLLBACK})
+
+
+def _produces_result_set(statement_type: int, ncolumns: int) -> bool:
+    """Whether a prepared statement needs a cursor opened for it.
+
+    This is JDBC's ``IfxSqli.isResultSet`` predicate, and it replaces a
+    first-word check for ``SELECT`` that got five ordinary forms wrong.
+    A leading comment of any of the three Informix flavours, a
+    parenthesized select, a parenthesized UNION, and a CTE were all
+    classified as DML, so the driver sent SQ_EXECUTE where the server
+    expected a cursor. The report back was ``-260 Cursor name already in
+    use``, which describes neither the cause nor anything the caller did.
+
+    The original comment justified the heuristic on the grounds that
+    ``nfields`` can't distinguish these, because ``INSERT INTO t VALUES
+    (?)`` also describes a column. True, and irrelevant: ``statement_type``
+    distinguishes them exactly. That INSERT reports 6 with one field; every
+    SELECT form above reports 2. The value was being parsed into the
+    describe metadata and thrown away.
+
+    ``EXECUTE PROCEDURE``/``FUNCTION`` (56) is the one type that depends on
+    the column count, since a routine may or may not return rows.
+    """
+    if statement_type == _ST_SELECT:
+        return True
+    return statement_type == _ST_ROUTINE and ncolumns > 0
+
+
 def _finalize_cursor(
     conn_ref: weakref.ReferenceType,
     state: list,
@@ -120,13 +167,30 @@ def _finalize_cursor(
     a list (not the cursor object itself) keeps the finalizer's closure
     weak — the cursor remains GC'd-able.
     """
-    from ._protocol import ProtocolError
 
     if not state[0]:
         return  # nothing to release
     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
@@ -147,19 +211,33 @@ def _finalize_cursor(
             conn._drain_to_eot()
             conn._send_pdu(_RELEASE_PDU)
             conn._drain_to_eot()
-        except (ProtocolError, OSError) 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()
+        except WIRE_ERRORS as exc:
+            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:
@@ -247,6 +325,9 @@ class Cursor:
         # from. Empirically the server accepts 0 here even when a real
         # ID was assigned, so this is best-effort tracking.
         self._statement_id: int = 0
+        # DESCRIBE's statement-type field. Decides whether a cursor is
+        # opened -- see _produces_result_set.
+        self._statement_type: int = 0
         # Phase 10: smart-LOB read via ``lotofile(col, path, 'client')``.
         # The server orchestrates a SQ_FILE (98) protocol where it tells
         # us to "open file X, write these bytes, close". We emulate the
@@ -331,6 +412,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 = []
@@ -339,27 +429,36 @@ class Cursor:
         self._rowcount = -1
         self._rows = []
         self._row_index = -1  # before-first-row
+        self._statement_type = 0
         self._statement_already_done = False
 
+        # Step 1: PREPARE — send SQL with numQmarks = 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()
+
         # On a logged DB in non-autocommit mode, the server requires an
         # explicit SQ_BEGIN before the first DML in each transaction.
         # _ensure_transaction is a no-op for autocommit / unlogged DBs,
         # and idempotent within an open transaction.
-        self._conn._ensure_transaction()
+        #
+        # This runs *after* the describe, matching JDBC's
+        # initiateTransaction placement, because until the describe lands
+        # we don't know whether the caller's statement is itself
+        # transaction control. Opening a transaction on their behalf and
+        # then executing their BEGIN WORK gets -535, "already in
+        # transaction" — the driver competing with the user for the same
+        # job and the user losing.
+        if self._statement_type not in _TX_CONTROL_TYPES:
+            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)))
-        self._read_describe_response()
-
-        # Branch on the SQL keyword. We can't use ``self._columns`` /
-        # ``nfields`` here because a parameterized INSERT also returns
-        # a non-empty DESCRIBE (server describes the would-be inserted
-        # row's columns). The SQL-keyword heuristic is what JDBC effectively
-        # does too via its IfxStatement / IfxPreparedStatement subclassing.
-        first_word = sql.lstrip().split(None, 1)[0].upper() if sql.strip() else ""
-        is_select = first_word == "SELECT"
-
-        if is_select:
+        # Ask the server what it just prepared, rather than guessing from
+        # the first word of the SQL.
+        if _produces_result_set(self._statement_type, len(self._columns)):
             if params:
                 self._execute_select_with_params(params)
             else:
@@ -369,6 +468,11 @@ class Cursor:
         else:
             self._execute_dml()
 
+        # The statement succeeded. If it was transaction control, the
+        # server's transaction state just changed and the connection has
+        # to know, or commit() and rollback() silently do nothing.
+        self._note_transaction_control()
+
         # SELECT path: position cursor before the first row so the next
         # ``fetchone()`` returns ``rows[0]``. DML paths leave _row_index
         # at -1 too (no rows to iterate).
@@ -377,6 +481,79 @@ class Cursor:
         if self._description is not None:
             self._row_index = -1
 
+    def _note_transaction_control(self) -> None:
+        """Sync the connection's transaction flag after a successful execute.
+
+        ``BEGIN WORK`` run through ``execute()`` opens a real transaction
+        on the server, and the connection had no idea. With autocommit on
+        nothing stopped it reaching the server, so ``_in_transaction``
+        stayed False while a transaction was open — and both ``commit()``
+        and ``rollback()`` are guarded by that flag. ``rollback()``
+        returned successfully having sent nothing, and the rows it was
+        asked to discard were still there.
+
+        The pool reads the same flag to decide whether a returned
+        connection needs cleaning up, so the connection went back into
+        circulation holding an open transaction and its locks.
+
+        Called only on the success path: a statement that failed did not
+        change the server's transaction state.
+        """
+        statement_type = self._statement_type
+        if statement_type == _ST_TX_BEGIN:
+            self._conn._in_transaction = True
+        elif statement_type in (_ST_TX_COMMIT, _ST_TX_ROLLBACK):
+            self._conn._in_transaction = False
+
+    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.
+
+        A statement that failed is still allocated. Skipping the release
+        bricks the connection: the next PREPARE collides with the leaked
+        one and every subsequent call returns a nonsense error whose
+        offset points back at the *failed* SQL. A duplicate-key violation
+        is about the most ordinary error an application can hit, so "one
+        constraint violation kills the connection" was easy to reach and
+        hard to attribute.
+
+        CLOSE and RELEASE get **separate** suppressions. Putting both in
+        one ``with contextlib.suppress(Exception)`` block reads as "clean
+        up both", but a CLOSE that raises skips the RELEASE entirely —
+        and RELEASE is the one that matters. Failing to close a cursor
+        wastes a handle; failing to release the statement is what breaks
+        the next call.
+
+        Everything here is swallowed rather than propagated. If the wire
+        is genuinely desynced then the cleanup fails too, and the caller
+        is far better served by the real SQL error than by a secondary
+        failure from the cleanup path.
+        """
+        if close_cursor:
+            with contextlib.suppress(Exception):
+                self._conn._send_pdu(self._build_close_pdu())
+                self._drain_to_eot()
+        with contextlib.suppress(Exception):
+            self._conn._send_pdu(self._build_release_pdu())
+            self._drain_to_eot()
+
     def _execute_select_with_params(self, params: tuple) -> None:
         """Parameterized SELECT: SQ_BIND → CURNAME+NFETCH → drain → CLOSE+RELEASE.
 
@@ -395,12 +572,20 @@ class Cursor:
         try:
             pdu = self._build_bind_only_pdu(params)
         except Exception:
-            with contextlib.suppress(Exception):
-                self._conn._send_pdu(self._build_release_pdu())
-                self._drain_to_eot()
+            self._release_after_failure()
             raise
         self._conn._send_pdu(pdu)
-        self._drain_to_eot()
+        try:
+            self._drain_to_eot()
+        except Exception:
+            # The server can reject the BIND itself — a value whose type
+            # doesn't match the described parameter, a bind against a
+            # statement the server has since invalidated. This was the
+            # last unguarded door of the six: the build was covered and
+            # the drain was not, so a server-side bind rejection left the
+            # statement allocated and the *next* execute() failed instead.
+            self._release_after_failure()
+            raise
         # Now open the cursor and fetch — the bound values are in scope
         # for the prepared statement.
         self._execute_select()
@@ -423,9 +608,23 @@ class Cursor:
             self._conn._send_pdu(
                 self._build_curname_scroll_open_pdu(cursor_name)
             )
-            self._drain_to_eot()
+            try:
+                self._drain_to_eot()
+            except Exception:
+                # Opening a scrollable cursor fails like any other
+                # statement — a bad ORDER BY, a permission error, a table
+                # dropped between PREPARE and OPEN. This branch had no
+                # guard at all, and it is the worst place to lack one:
+                # the GC-time finalizer is armed on the line *after* the
+                # drain, so a failure here left the statement allocated
+                # with no fallback whatsoever to reclaim it.
+                self._release_after_failure(close_cursor=True)
+                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
@@ -457,12 +656,9 @@ class Cursor:
             # that returns early), so this path needs its own cleanup —
             # otherwise a mid-fetch failure leaks and the next statement
             # collides with it. Same failure mode as the DML path; see
-            # _execute_dml for what that looks like from the caller's side.
-            with contextlib.suppress(Exception):
-                self._conn._send_pdu(self._build_close_pdu())
-                self._drain_to_eot()
-                self._conn._send_pdu(self._build_release_pdu())
-                self._drain_to_eot()
+            # _release_after_failure for what that looks like from the
+            # caller's side.
+            self._release_after_failure(close_cursor=True)
             raise
 
         self._conn._send_pdu(self._build_close_pdu())
@@ -849,9 +1045,7 @@ class Cursor:
         try:
             pdu = self._build_bind_execute_pdu(params)
         except Exception:
-            with contextlib.suppress(Exception):
-                self._conn._send_pdu(self._build_release_pdu())
-                self._drain_to_eot()
+            self._release_after_failure()
             raise
         self._conn._send_pdu(pdu)
         try:
@@ -860,9 +1054,7 @@ class Cursor:
             # The statement is still allocated server-side even though it
             # failed. See _execute_dml for why skipping this bricks the
             # connection.
-            with contextlib.suppress(Exception):
-                self._conn._send_pdu(self._build_release_pdu())
-                self._drain_to_eot()
+            self._release_after_failure()
             raise
         self._conn._send_pdu(self._build_release_pdu())
         self._drain_to_eot()
@@ -902,9 +1094,7 @@ class Cursor:
             # desynced the release will fail too, and the caller is far
             # better served by the real SQL error than by a secondary
             # failure from the cleanup path.
-            with contextlib.suppress(Exception):
-                self._conn._send_pdu(self._build_release_pdu())
-                self._drain_to_eot()
+            self._release_after_failure()
             raise
         self._conn._send_pdu(self._build_release_pdu())
         self._drain_to_eot()
@@ -962,7 +1152,10 @@ class Cursor:
                     f"expected {first_len} (matching set [0])"
                 )
 
-        # Detect SELECT — not supported in executemany.
+        # Cheap pre-flight reject for the obvious case, so the common
+        # mistake costs no round-trip. The authoritative check is after
+        # PREPARE, below — this one shares the first-word heuristic's
+        # blind spots (leading comments, CTEs, parenthesized selects).
         first_word = operation.lstrip().split(None, 1)[0].upper() if operation.strip() else ""
         if first_word == "SELECT":
             raise NotSupportedError("executemany on SELECT is not supported")
@@ -973,6 +1166,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 = []
@@ -980,18 +1176,33 @@ class Cursor:
             self._rowcount = -1
             self._rows = []
             self._row_index = -1
+            self._statement_type = 0
             self._statement_already_done = False
 
-            # Logged-DB transaction guard — same as execute(). Idempotent
-            # within an open transaction.
-            self._conn._ensure_transaction()
-
             # 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()
 
+            # Now the server has told us what it prepared. A result-set
+            # statement that slipped past the first-word check above --
+            # a CTE, a leading comment, a parenthesized select -- would
+            # otherwise be executed N times down the DML path, which
+            # opens no cursor and answers -260.
+            if _produces_result_set(self._statement_type, len(self._columns)):
+                self._release_after_failure()
+                raise NotSupportedError(
+                    "executemany on a statement that returns rows is not "
+                    "supported"
+                )
+
+            # Logged-DB transaction guard — same as execute(), and for the
+            # same reason placed after the describe rather than before it.
+            if self._statement_type not in _TX_CONTROL_TYPES:
+                self._conn._ensure_transaction()
+
             # Phase 33: pipeline — build all BIND+EXECUTE PDUs first
             # (Python work, no I/O), then send them back-to-back, then
             # drain all responses. Eliminates the per-row round-trip
@@ -1018,9 +1229,7 @@ class Cursor:
                 # already unusable in that case, but attempting the
                 # release costs nothing and the original error is what
                 # propagates either way.
-                with contextlib.suppress(Exception):
-                    self._conn._send_pdu(self._build_release_pdu())
-                    self._drain_to_eot()
+                self._release_after_failure()
                 raise
 
             # Drain N responses. The first error is captured but we
@@ -1316,20 +1525,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
@@ -1633,6 +1830,7 @@ class Cursor:
             elif tag == MessageType.SQ_DESCRIBE:
                 self._columns, meta = parse_describe(reader)
                 self._statement_id = meta.get("statement_id", 0)
+                self._statement_type = meta.get("statement_type", 0)
                 self._description = (
                     [c.to_description_tuple() for c in self._columns] if self._columns else None
                 )
@@ -1764,7 +1962,6 @@ class Cursor:
         (e.g. table or column name for "not found" errors). Empty for
         most syntax errors.
         """
-        from ._protocol import ProtocolError
 
         sqlcode = reader.read_short()
         isamcode = reader.read_short()
@@ -1782,7 +1979,7 @@ class Cursor:
                 if name_len & 1:
                     reader.read_exact(1)  # pad to even
                 near_token = raw.rstrip(b"\x00").decode("iso-8859-1", errors="replace")
-        except (ProtocolError, OSError):
+        except WIRE_ERRORS:
             pass
         # Drain remaining bytes until SQ_EOT. Phase 28: a (ProtocolError,
         # OSError) during drain means the wire is in an unknown state —
@@ -1794,7 +1991,7 @@ class Cursor:
                 t = reader.read_short()
                 if t == MessageType.SQ_EOT:
                     break
-        except (ProtocolError, OSError):
+        except WIRE_ERRORS:
             with contextlib.suppress(Exception):
                 self._conn.close()
 

tests/test_async_threads.py

@@ -0,0 +1,216 @@
+"""The async layer used a thread pool it shared with the whole process.
+
+``asyncio.to_thread`` runs on the event loop's default executor. That
+executor belongs to the process, not to us, and it is sized from the CPU
+count — ``min(32, cpu_count + 4)``, which is six threads on a two-CPU
+container. Every blocking call in ``informix_db.aio`` went through it.
+
+Two consequences, both silent.
+
+**Cancelled calls hold their threads.** ``asyncio.to_thread`` cannot
+interrupt a worker, so a cancelled await leaves the thread running the
+wire call until the read timeout expires. Cancellation is ordinary in a
+web app — a client disconnect cancels the request task — so a handful of
+them pins every thread in the shared pool. Unrelated ``to_thread`` work
+anywhere else in the process then stops dead, and so does the driver.
+
+**Pool concurrency was capped by an unrelated number.** A pool with
+``max_size=20`` on a two-CPU box ran six queries at a time, and nothing
+said so.
+
+Each connection now owns one thread. That is the right size rather than
+a compromise: the sync connection serializes every wire operation on its
+own lock, so a second thread could do nothing but wait for the first. It
+also avoids a deadlock that a shared pool-sized executor invites — with
+N threads and N connections, N tasks blocked in ``acquire`` occupy every
+thread while the connection they wait for is held by a task that now
+needs a thread to finish and release it.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import threading
+import time
+from concurrent.futures import ThreadPoolExecutor
+
+import pytest
+
+from informix_db import aio
+from tests.conftest import ConnParams
+
+pytestmark = pytest.mark.integration
+
+
+def _kw(conn_params: ConnParams) -> dict:
+    return {
+        "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": 20.0,
+        "autocommit": True,
+    }
+
+
+@pytest.mark.asyncio
+async def test_queries_run_with_the_default_executor_fully_held(
+    conn_params: ConnParams,
+) -> None:
+    """The starvation case, with real queries. Four unrelated jobs occupy
+    every thread of a four-worker default executor; the driver must not
+    be waiting behind them."""
+    loop = asyncio.get_running_loop()
+    previous = loop._default_executor
+    loop.set_default_executor(ThreadPoolExecutor(max_workers=4))
+    conns = [await aio.connect(**_kw(conn_params)) for _ in range(4)]
+    cursors = [await c.cursor() for c in conns]
+
+    gate = threading.Event()
+    hogs = [
+        asyncio.create_task(asyncio.to_thread(gate.wait, 30)) for _ in range(4)
+    ]
+    await asyncio.sleep(0.3)
+
+    async def query(cursor, i: int):
+        await cursor.execute(f"SELECT FIRST 1 tabid + {i} FROM systables")
+        return await cursor.fetchone()
+
+    try:
+        rows = await asyncio.wait_for(
+            asyncio.gather(*(query(c, i) for i, c in enumerate(cursors))),
+            timeout=10.0,
+        )
+        assert len(rows) == 4
+        assert all(r is not None for r in rows)
+    finally:
+        gate.set()
+        await asyncio.gather(*hogs, return_exceptions=True)
+        for c in conns:
+            await c.close()
+        # set_default_executor rejects None, which is what the loop
+        # starts with before anything has used to_thread.
+        loop.set_default_executor(previous or ThreadPoolExecutor())
+
+
+@pytest.mark.asyncio
+async def test_each_connection_gets_its_own_thread(
+    conn_params: ConnParams,
+) -> None:
+    a = await aio.connect(**_kw(conn_params))
+    b = await aio.connect(**_kw(conn_params))
+    try:
+        assert a._executor is not b._executor
+        assert a._executor._max_workers == 1, (
+            "more than one thread per connection cannot help — the wire "
+            "lock serializes them anyway"
+        )
+    finally:
+        await a.close()
+        await b.close()
+
+
+@pytest.mark.asyncio
+async def test_cursor_runs_on_its_connections_thread(
+    conn_params: ConnParams,
+) -> None:
+    """A cursor must not fall back to the default executor, or half the
+    work goes back to being shared."""
+    conn = await aio.connect(**_kw(conn_params))
+    try:
+        cur = await conn.cursor()
+        await cur.execute("SELECT FIRST 1 tabid FROM systables")
+        names: list[str] = []
+        await conn._run(lambda: names.append(threading.current_thread().name))
+        assert names[0].startswith("informix-conn")
+        assert cur._run.__self__ is conn
+    finally:
+        await conn.close()
+
+
+@pytest.mark.asyncio
+async def test_pool_reuses_one_thread_per_connection(
+    conn_params: ConnParams,
+) -> None:
+    """The executor lives on the sync connection, so a connection handed
+    out, returned, and handed out again keeps the same thread instead of
+    spawning one per acquire."""
+    pool = await aio.create_pool(**_kw(conn_params), min_size=1, max_size=2)
+    try:
+        seen = []
+        for _ in range(4):
+            async with pool.connection() as conn:
+                cur = await conn.cursor()
+                await cur.execute("SELECT FIRST 1 tabid FROM systables")
+                assert await cur.fetchone() is not None
+                seen.append(id(conn._executor))
+        assert len(set(seen)) <= 2, (
+            f"expected at most one executor per pooled connection, saw "
+            f"{len(set(seen))}"
+        )
+    finally:
+        await pool.close()
+
+
+@pytest.mark.asyncio
+async def test_pool_concurrency_is_not_capped_by_the_default_executor(
+    conn_params: ConnParams,
+) -> None:
+    """Six concurrent pooled queries against a two-worker default
+    executor. Under the old arrangement at most two could run."""
+    loop = asyncio.get_running_loop()
+    previous = loop._default_executor
+    pool = await aio.create_pool(**_kw(conn_params), min_size=6, max_size=6)
+    loop.set_default_executor(ThreadPoolExecutor(max_workers=2))
+    try:
+
+        async def one(i: int):
+            async with pool.connection(timeout=15.0) as conn:
+                cur = await conn.cursor()
+                # Long enough that serialized execution would be obvious.
+                await cur.execute(
+                    "SELECT FIRST 200 a.tabid FROM systables a, systables b"
+                )
+                return len(await cur.fetchall())
+
+        started = time.monotonic()
+        results = await asyncio.wait_for(
+            asyncio.gather(*(one(i) for i in range(6))), timeout=25.0
+        )
+        assert all(r > 0 for r in results)
+        assert len(results) == 6
+        elapsed = time.monotonic() - started
+        assert elapsed < 20.0, f"queries appear serialized ({elapsed:.1f}s)"
+    finally:
+        await pool.close()
+        # set_default_executor rejects None, which is what the loop
+        # starts with before anything has used to_thread.
+        loop.set_default_executor(previous or ThreadPoolExecutor())
+
+
+@pytest.mark.asyncio
+async def test_closing_a_connection_stops_its_thread(
+    conn_params: ConnParams,
+) -> None:
+    """One thread per connection is only affordable if the thread goes
+    away with the connection."""
+    before = threading.active_count()
+    conns = [await aio.connect(**_kw(conn_params)) for _ in range(5)]
+    for c in conns:
+        cur = await c.cursor()
+        await cur.execute("SELECT FIRST 1 tabid FROM systables")
+    for c in conns:
+        await c.close()
+    del conns
+    # Threads exit asynchronously after shutdown; give them a moment.
+    for _ in range(50):
+        if threading.active_count() <= before + 1:
+            break
+        await asyncio.sleep(0.1)
+    assert threading.active_count() <= before + 1, (
+        f"connection threads outlived their connections "
+        f"({before} -> {threading.active_count()})"
+    )

tests/test_row_reconciliation.py

@@ -0,0 +1,230 @@
+"""End-of-row reconciliation, and the two bugs it found immediately.
+
+Fourteen framing bugs reached users before this check existed. Every one
+was the same defect — a column consuming the wrong number of bytes — and
+not one raised at the point of the mistake. The wrong width produced a
+plausible value and corrupted whatever came *next*, so the damage always
+surfaced in a different column, row, or statement.
+
+All fourteen were detectable for free. ``payload`` is a fully extracted
+``bytes`` object of known length, so after decoding N columns the offset
+must land exactly on the end. ``_row_not_consumed`` raises when it
+doesn't, naming the column shape and the byte delta.
+
+Turning the check on found two more bugs on its first run against the
+existing suite:
+
+* **Smart LOBs were read as a flat 72-byte field.** They use the UDT
+  envelope: 149 bytes populated (``[ind=0][len=144][144 hex chars]``),
+  5 bytes NULL. We consumed 72 and left 77 behind, so any LOB not in the
+  final column position corrupted everything after it. The 144 bytes are
+  the 72-byte locator *hex-encoded*, which means ``BlobLocator.raw`` had
+  never actually held a locator — only the first half of the hex text.
+  Nobody noticed because ``read_blob_column`` resolves LOBs through the
+  server-side ``lotofile`` function and never uses the locator.
+
+* **NULL composite UDTs skipped their length field**, byte-for-byte the
+  LVARCHAR NULL bug in the branch twelve lines away.
+
+Both were the same ``[indicator][int32 length][data]`` envelope, written
+out by hand a third and fourth time. ``_read_udt_envelope`` is now the
+single implementation.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import struct
+
+import pytest
+
+import informix_db
+from informix_db._protocol import IfxStreamReader, ProtocolError
+from informix_db._resultset import (
+    ColumnInfo,
+    _read_udt_envelope,
+    parse_tuple_payload,
+)
+from tests.conftest import ConnParams
+
+# ---------------------------------------------------------------------------
+# The check itself — unit level, no server
+# ---------------------------------------------------------------------------
+
+
+def _tuple_pdu(body: bytes) -> IfxStreamReader:
+    """Wrap a payload in the SQ_TUPLE framing parse_tuple_payload expects."""
+    return IfxStreamReader(
+        io.BytesIO(struct.pack("!h", 0) + struct.pack("!i", len(body)) + body)
+    )
+
+
+def test_under_read_is_detected() -> None:
+    cols = [ColumnInfo(name="k", type_code=2, raw_type_code=2, encoded_length=4)]
+    with pytest.raises(ProtocolError, match="under-read by 6"):
+        parse_tuple_payload(_tuple_pdu(struct.pack("!i", 5) + b"LEFTOV"), cols)
+
+
+def test_exact_consumption_passes() -> None:
+    cols = [ColumnInfo(name="k", type_code=2, raw_type_code=2, encoded_length=4)]
+    assert parse_tuple_payload(_tuple_pdu(struct.pack("!i", 5)), cols) == (5,)
+
+
+def test_error_names_the_column_shape() -> None:
+    """"row decode failed" is not actionable. The message has to say which
+    shape and by how much, so the next report arrives with the answer."""
+    cols = [
+        ColumnInfo(name="ident", type_code=2, raw_type_code=2, encoded_length=4),
+        ColumnInfo(name="qty", type_code=2, raw_type_code=2, encoded_length=4),
+    ]
+    body = struct.pack("!i", 1) + struct.pack("!i", 2) + b"UNCONSUMED"
+    with pytest.raises(ProtocolError) as exc:
+        parse_tuple_payload(_tuple_pdu(body), cols)
+    msg = str(exc.value)
+    assert "ident" in msg and "qty" in msg, "message must name the columns"
+    assert "tc=2" in msg
+    assert "consumed 8 of 18" in msg, "message must give the byte counts"
+    assert "under-read by 10" in msg
+
+
+# ---------------------------------------------------------------------------
+# The shared UDT envelope
+# ---------------------------------------------------------------------------
+
+
+def test_envelope_reads_length_even_when_null() -> None:
+    """The length belongs to the envelope, not the value. Returning early
+    on the indicator is the bug that hit LVARCHAR and then composites."""
+    payload = bytes([1]) + struct.pack("!i", 0) + b"NEXT"
+    offset, body = _read_udt_envelope(payload, 0)
+    assert body is None
+    assert offset == 5, "NULL envelope must still consume its length field"
+    assert payload[offset:] == b"NEXT"
+
+
+def test_envelope_reads_body_when_present() -> None:
+    payload = bytes([0]) + struct.pack("!i", 3) + b"abc" + b"NEXT"
+    offset, body = _read_udt_envelope(payload, 0)
+    assert body == b"abc"
+    assert offset == 8
+    assert payload[offset:] == b"NEXT"
+
+
+def test_envelope_rejects_negative_length() -> None:
+    """A negative length off the wire would rewind the offset and silently
+    re-decode earlier bytes as a later column."""
+    payload = bytes([0]) + struct.pack("!i", -4) + b"abc"
+    with pytest.raises(ProtocolError, match="negative length"):
+        _read_udt_envelope(payload, 0)
+
+
+# ---------------------------------------------------------------------------
+# Against a real server
+# ---------------------------------------------------------------------------
+
+pytestmark_integration = pytest.mark.integration
+
+
+def _connect(conn_params: ConnParams) -> 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=15.0, read_timeout=30.0,
+        autocommit=True,
+    )
+
+
+@pytest.mark.integration
+def test_null_composite_does_not_shift_next_column(
+    conn_params: ConnParams,
+) -> None:
+    """A NULL SET made the following VARCHAR return '' instead of its value."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_nullset")
+        cur.execute(
+            "CREATE TABLE t_nullset (a INT, s SET(INT NOT NULL), b VARCHAR(10))"
+        )
+        try:
+            cur.execute("INSERT INTO t_nullset VALUES (7, SET{1,2}, 'abc')")
+            cur.execute("INSERT INTO t_nullset VALUES (8, NULL, 'xyz')")
+            cur.execute("SELECT a, s, b FROM t_nullset ORDER BY a")
+            rows = cur.fetchall()
+            assert rows[0][0] == 7 and rows[0][2] == "abc"
+            assert rows[1] == (8, None, "xyz"), (
+                "NULL composite shifted the following column"
+            )
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_nullset")
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize("populated", [False, True])
+def test_lob_in_non_final_position(
+    conn_params: ConnParams, populated: bool
+) -> None:
+    """A LOB was read as 72 bytes when it occupies 149 (or 5 when NULL), so
+    anything selected after it was garbage."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_lobmid")
+        try:
+            cur.execute(
+                "CREATE TABLE t_lobmid (a INT, b BLOB, c INT, d VARCHAR(10))"
+            )
+        except informix_db.Error:
+            pytest.skip("no sbspace configured; see make ifx-spaces")
+        try:
+            if populated:
+                cur.write_blob_column(
+                    "INSERT INTO t_lobmid VALUES (?, BLOB_PLACEHOLDER, ?, ?)",
+                    b"hello", (1, 222222, "tail"),
+                )
+            else:
+                cur.execute(
+                    "INSERT INTO t_lobmid VALUES (1, NULL, 222222, 'tail')"
+                )
+            cur.execute("SELECT a, b, c, d FROM t_lobmid")
+            a, b, c, d = cur.fetchone()
+            assert (a, c, d) == (1, 222222, "tail"), (
+                "columns after the LOB were shifted"
+            )
+            if populated:
+                assert isinstance(b, informix_db.BlobLocator)
+                assert len(b.raw) == 72, (
+                    "locator must be the decoded 72 bytes, not hex text"
+                )
+            else:
+                assert b is None
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_lobmid")
+
+
+@pytest.mark.integration
+def test_wide_mixed_row_reconciles(conn_params: ConnParams) -> None:
+    """Belt and braces: a row using most of the tricky types must consume
+    its payload exactly. The check runs on every fetch, so merely getting
+    a row back proves reconciliation held."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        cur.execute(
+            "CREATE TEMP TABLE t_recon ("
+            " a INT8 NOT NULL, k LVARCHAR(512), n NCHAR(8), v VARCHAR(32),"
+            " d DECIMAL(16), b BOOLEAN, t DATETIME YEAR TO FRACTION(5),"
+            " tail INT)"
+        )
+        cur.execute(
+            "INSERT INTO t_recon VALUES (?, ?, ?, ?, ?, ?, CURRENT, ?)",
+            (1, "PackageRoot", "nch", "v", None, True, 424242),
+        )
+        cur.execute(
+            "SELECT a, k, n, v, d, b, t, tail FROM t_recon"
+        )
+        row = cur.fetchone()
+        assert row[0] == 1
+        assert row[-1] == 424242

tests/test_socket_reads.py

@@ -0,0 +1,172 @@
+"""Two readers, one stream, and a length field taken on trust.
+
+``IfxSocket`` owns a read-ahead buffer that ``BufferedSocketReader``
+fills and drains. But ``Connection._drain_to_eot``, ``_raise_sq_err``
+and the login path bypass that reader and call ``IfxSocket.read_exact``
+directly, which recv'd from the socket without ever looking at the
+buffer. Bytes sitting in the buffer would simply be skipped, and skipped
+bytes in a length-framed protocol don't announce themselves — the next
+read lands mid-field and every read after it is wrong.
+
+Nothing triggers it today. The server sends one response per request, so
+recv returns exactly that response and the buffered reader consumes all
+of it before control returns to a direct read. That is a property of the
+traffic, not of the code, and the buffer is connection-scoped precisely
+so read-ahead *can* cross response boundaries — pipelined executemany
+already puts several responses in flight. A latent desync waiting on a
+timing change is not a good thing to leave in a wire protocol.
+
+Separately, ``fill_recv_buf`` took its byte count on trust, and that
+count is almost always a length field straight off the wire. A corrupt
+or desynced stream turned into an allocation of whatever the field
+happened to say: a garbage ``0x7FFFFFFF`` reads as a 2 GB request, and
+the fill loop sits in recv until the read timeout while the buffer
+grows. The limit turns that into an error that names the number, which
+is the actual diagnostic — a length that absurd means framing was
+already lost upstream.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from informix_db._protocol import BufferedSocketReader, ProtocolError
+from informix_db._socket import MAX_READ_BYTES, IfxSocket
+
+
+class _FakeSocket:
+    """Stands in for the raw socket. Records what recv actually asked for."""
+
+    def __init__(self, data: bytes = b"") -> None:
+        self.data = data
+        self.pos = 0
+        self.recv_calls: list[int] = []
+
+    def recv(self, n: int) -> bytes:
+        self.recv_calls.append(n)
+        chunk = self.data[self.pos : self.pos + n]
+        self.pos += len(chunk)
+        return chunk
+
+    def close(self) -> None:
+        # The EOF path force-closes; a stand-in has to survive that.
+        pass
+
+
+def _socket_with(buffered: bytes, on_wire: bytes = b"") -> IfxSocket:
+    """An IfxSocket with ``buffered`` already read ahead into _recv_buf."""
+    sock = IfxSocket.__new__(IfxSocket)
+    sock._sock = _FakeSocket(on_wire)
+    sock._recv_buf = bytearray(buffered)
+    sock._recv_pos = 0
+    sock._recv_size = 65536
+    sock._read_timeout = None
+    return sock
+
+
+# ---------------------------------------------------------------------------
+# read_exact must not step over the buffer
+# ---------------------------------------------------------------------------
+
+
+def test_read_exact_consumes_the_buffer_first() -> None:
+    sock = _socket_with(b"BUFFERED", on_wire=b"SOCKET")
+    assert sock.read_exact(8) == b"BUFFERED"
+    assert sock._sock.recv_calls == [], "must not touch the socket at all"
+    assert sock._recv_pos == 8
+
+
+def test_read_exact_spans_buffer_then_socket() -> None:
+    """The interesting case: a read that starts in the buffer and
+    finishes on the wire. Getting this wrong reorders the stream."""
+    sock = _socket_with(b"HEAD", on_wire=b"TAIL")
+    assert sock.read_exact(8) == b"HEADTAIL"
+    assert sock._sock.recv_calls == [4], "only the shortfall comes from recv"
+
+
+def test_read_exact_respects_a_partly_consumed_buffer() -> None:
+    sock = _socket_with(b"XXABCD")
+    sock._recv_pos = 2  # first two bytes already decoded
+    assert sock.read_exact(4) == b"ABCD"
+    assert sock._sock.recv_calls == []
+
+
+def test_read_exact_of_zero_is_empty() -> None:
+    sock = _socket_with(b"DATA")
+    assert sock.read_exact(0) == b""
+    assert sock.read_exact(-5) == b"", "a negative count must not rewind"
+    assert sock._recv_pos == 0
+
+
+def test_short_read_error_reports_the_original_request() -> None:
+    """The message counts bytes; taking some from the buffer must not make
+    it lie about how many were asked for."""
+    from informix_db.exceptions import OperationalError
+
+    sock = _socket_with(b"AB", on_wire=b"")  # 2 buffered, nothing on the wire
+    with pytest.raises(OperationalError, match="wanted 10 bytes"):
+        sock.read_exact(10)
+
+
+def test_buffered_reader_and_direct_read_agree_on_one_stream() -> None:
+    """End to end: a BufferedSocketReader over-reads, then a direct
+    read_exact picks up exactly where it left off."""
+    sock = _socket_with(b"", on_wire=b"\x00\x2aREST-OF-THE-STREAM")
+    reader = BufferedSocketReader(sock)
+    assert reader.read_short() == 42
+    assert len(sock._recv_buf) - sock._recv_pos > 0, (
+        "precondition: the reader must have over-read for this to mean "
+        "anything"
+    )
+    assert sock.read_exact(18) == b"REST-OF-THE-STREAM"
+
+
+# ---------------------------------------------------------------------------
+# fill_recv_buf must not believe an arbitrary length
+# ---------------------------------------------------------------------------
+
+
+def test_absurd_length_is_refused_not_allocated() -> None:
+    sock = _socket_with(b"", on_wire=b"")
+    with pytest.raises(ProtocolError, match="refusing to read"):
+        sock.fill_recv_buf(MAX_READ_BYTES + 1)
+    assert sock._sock.recv_calls == [], "must refuse before any recv"
+
+
+def test_refusal_names_the_knob() -> None:
+    """The error has to be actionable in both directions: framing is lost,
+    or the value genuinely is that big and the limit needs raising."""
+    sock = _socket_with(b"", on_wire=b"")
+    with pytest.raises(ProtocolError) as exc:
+        sock.fill_recv_buf(2**31 - 1)
+    message = str(exc.value)
+    assert "2147483647" in message
+    assert "IFX_MAX_READ_BYTES" in message
+
+
+def test_a_normal_length_is_unaffected() -> None:
+    sock = _socket_with(b"", on_wire=b"x" * 100)
+    sock.fill_recv_buf(100)
+    assert len(sock._recv_buf) - sock._recv_pos >= 100
+
+
+# ---------------------------------------------------------------------------
+# skip
+# ---------------------------------------------------------------------------
+
+
+def test_buffered_skip_does_not_rewind_on_a_negative_count() -> None:
+    """The base reader's skip delegates to read_exact, which guards. This
+    one advances the cursor arithmetically, so an unguarded negative count
+    re-decodes bytes already consumed as if they were the next field."""
+    sock = _socket_with(b"ABCDEFGH")
+    sock._recv_pos = 4
+    BufferedSocketReader(sock).skip(-4)
+    assert sock._recv_pos == 4, "skip must never move the cursor backwards"
+
+
+def test_buffered_skip_advances_normally() -> None:
+    sock = _socket_with(b"ABCDEFGH")
+    reader = BufferedSocketReader(sock)
+    reader.skip(4)
+    assert reader.read_exact(4) == b"EFGH"

tests/test_statement_classification.py

@@ -0,0 +1,216 @@
+"""Deciding whether a statement needs a cursor, by asking rather than guessing.
+
+The driver chose between "open a cursor and fetch" and "execute and
+release" by checking whether the first word of the SQL was ``SELECT``.
+That gets five ordinary forms wrong — a leading comment in any of the
+three Informix flavours, a parenthesized select, a parenthesized UNION,
+and a CTE. All five are perfectly good queries, and all five failed with
+``-260 Cursor name already in use``, an error that describes neither the
+cause nor anything the caller did. It says "cursor" because the driver
+sent SQ_EXECUTE where the server was waiting to open one.
+
+The server had been telling us the answer the whole time.
+``statement_type`` is the first field of the DESCRIBE response, and
+``parse_describe`` has always parsed it into the metadata dict, where
+nothing read it. Every SELECT form above reports 2.
+
+The comment that justified the heuristic said ``nfields`` couldn't
+distinguish these cases, because ``INSERT INTO t VALUES (?)`` also
+describes a column. That much was true — and it argued for the wrong
+conclusion, because ``statement_type`` distinguishes them exactly. That
+INSERT reports 6.
+
+The predicate is now JDBC's ``IfxSqli.isResultSet``: type 2, or type 56
+(``EXECUTE PROCEDURE``/``FUNCTION``) with at least one column, since a
+routine may or may not return rows. That last clause fixes
+``EXECUTE FUNCTION`` as a side effect — it used to run down the DML path
+and discard its return value.
+"""
+
+from __future__ import annotations
+
+import contextlib
+
+import pytest
+
+import informix_db
+from informix_db.cursors import _produces_result_set
+from tests.conftest import ConnParams
+
+# ---------------------------------------------------------------------------
+# The predicate
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize(
+    ("statement_type", "ncolumns", "expected", "why"),
+    [
+        (2, 1, True, "SELECT"),
+        (2, 0, True, "SELECT describing no columns is still a SELECT"),
+        (6, 1, False, "INSERT ... VALUES (?) describes a column but is DML"),
+        (6, 0, False, "INSERT with literals"),
+        (32, 0, False, "DELETE"),
+        (33, 0, False, "UPDATE"),
+        (45, 0, False, "CREATE"),
+        (56, 0, False, "EXECUTE PROCEDURE returning nothing"),
+        (56, 2, True, "EXECUTE FUNCTION returning rows"),
+        (0, 0, False, "unknown type defaults to the non-cursor path"),
+    ],
+)
+def test_result_set_predicate(
+    statement_type: int, ncolumns: int, expected: bool, why: str
+) -> None:
+    assert _produces_result_set(statement_type, ncolumns) is expected, why
+
+
+# ---------------------------------------------------------------------------
+# Against a real server
+# ---------------------------------------------------------------------------
+
+
+def _connect(conn_params: ConnParams) -> 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,
+    )
+
+
+_ONE = "SELECT FIRST 1 tabid FROM systables"
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    ("label", "sql"),
+    [
+        pytest.param("plain", _ONE, id="plain"),
+        pytest.param("lowercase", _ONE.lower(), id="lowercase"),
+        pytest.param("leading-whitespace", f"  \n\t {_ONE}", id="whitespace"),
+        pytest.param("line-comment", f"-- pick one\n{_ONE}", id="line-comment"),
+        pytest.param("block-comment", f"/* pick one */ {_ONE}", id="block-comment"),
+        pytest.param("brace-comment", f"{{ pick one }} {_ONE}", id="brace-comment"),
+        pytest.param(
+            "cte",
+            "WITH c AS (SELECT tabid FROM systables) "
+            "SELECT FIRST 1 tabid FROM c",
+            id="cte",
+        ),
+        pytest.param("parenthesized", f"({_ONE})", id="parenthesized"),
+        pytest.param(
+            "union", f"{_ONE} UNION SELECT 99 FROM systables", id="union"
+        ),
+        pytest.param(
+            "parenthesized-union",
+            f"({_ONE}) UNION (SELECT 99 FROM systables)",
+            id="paren-union",
+        ),
+    ],
+)
+def test_every_select_form_opens_a_cursor(
+    conn_params: ConnParams, label: str, sql: str
+) -> None:
+    """The five non-``plain`` forms below the whitespace case all failed
+    with -260 under the first-word heuristic."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        try:
+            cur.execute(sql)
+        except informix_db.ProgrammingError as exc:
+            # Informix 12.10 has no CTEs and rejects WITH at offset 1.
+            # A syntax error is the server declining the grammar, which
+            # is a different thing from the driver routing it wrongly —
+            # that produced -260, not -201.
+            if getattr(exc, "sqlcode", None) == -201:
+                pytest.skip(f"server does not support this syntax: {label}")
+            raise
+        rows = cur.fetchall()
+        assert rows, f"{label}: expected rows, got none"
+        assert cur.description is not None
+
+
+@pytest.mark.integration
+def test_parameterized_insert_is_not_mistaken_for_a_query(
+    conn_params: ConnParams,
+) -> None:
+    """The case the old comment worried about, and the reason it kept the
+    heuristic: this DESCRIBEs a column. ``statement_type`` says 6."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        cur.execute("CREATE TEMP TABLE t_cls (k INT)")
+        cur.execute("INSERT INTO t_cls VALUES (?)", (7,))
+        assert cur.rowcount == 1
+        cur.execute("SELECT k FROM t_cls")
+        assert cur.fetchall() == [(7,)]
+
+
+@pytest.mark.integration
+def test_execute_function_returns_its_value(conn_params: ConnParams) -> None:
+    """Type 56 with columns. Under the first-word heuristic this ran down
+    the DML path and the return value was discarded."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP FUNCTION ifxdrv_dbl")
+        cur.execute(
+            "CREATE FUNCTION ifxdrv_dbl(n INT) RETURNING INT; "
+            "RETURN n * 2; END FUNCTION"
+        )
+        try:
+            cur.execute("EXECUTE FUNCTION ifxdrv_dbl(21)")
+            assert cur.fetchall() == [(42,)]
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP FUNCTION ifxdrv_dbl")
+
+
+@pytest.mark.integration
+def test_dml_still_takes_the_non_cursor_path(conn_params: ConnParams) -> None:
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        cur.execute("CREATE TEMP TABLE t_cls2 (k INT)")
+        cur.execute("INSERT INTO t_cls2 VALUES (1)")
+        cur.execute("UPDATE t_cls2 SET k = 2")
+        assert cur.rowcount == 1
+        cur.execute("DELETE FROM t_cls2")
+        assert cur.rowcount == 1
+        assert cur.description is None
+
+
+@pytest.mark.integration
+def test_executemany_refuses_a_query_the_first_word_missed(
+    conn_params: ConnParams,
+) -> None:
+    """The pre-flight check shares the heuristic's blind spots, so a
+    comment-prefixed SELECT reaches PREPARE. The post-DESCRIBE check
+    catches it, and the connection stays usable — the refusal releases
+    the statement. (A leading comment rather than a CTE, so this also
+    runs on 12.10, which has no CTEs.)"""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        with pytest.raises(informix_db.NotSupportedError):
+            cur.executemany(
+                "/* batched? no */ SELECT FIRST 1 tabid FROM systables "
+                "WHERE tabid <> ?",
+                [(1,), (2,)],
+            )
+        cur.execute("SELECT FIRST 1 tabid FROM systables")
+        assert cur.fetchone() is not None, "refusal leaked the statement"
+
+
+@pytest.mark.integration
+def test_executemany_still_refuses_a_plain_select(
+    conn_params: ConnParams,
+) -> None:
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        with pytest.raises(informix_db.NotSupportedError):
+            cur.executemany(
+                "SELECT FIRST 1 tabid FROM systables WHERE tabid <> ?",
+                [(1,), (2,)],
+            )

tests/test_statement_release.py

@@ -0,0 +1,189 @@
+"""Every door out of a statement has to release it on the way through.
+
+A statement that fails is still allocated server-side. Skipping the
+RELEASE bricks the connection: the next PREPARE collides with the leaked
+one, and every subsequent call returns a nonsense error whose offset
+points back at the *failed* SQL rather than the new statement.
+
+There are six exits from the execute paths, and the guard was added to
+them one at a time, each after a user hit it:
+
+1. ``_execute_dml`` drain
+2. ``_execute_dml_with_params`` build
+3. ``_execute_dml_with_params`` drain
+4. ``executemany`` pipeline build/send
+5. ``_execute_select_with_params`` build
+6. ``_execute_select`` fetch loop
+
+Two were still open, and both are ordinary to reach:
+
+* **The parameterized-SELECT bind drain.** The build was guarded and the
+  drain was not. Passing a string where the column is an INT gets a
+  clean encode, so the rejection comes from the *server* (-1213, -415)
+  during the bind drain — past the guard. A wrong-typed parameter is
+  about as common as application mistakes get.
+
+* **The scrollable-cursor open.** No guard at all, and the worst place
+  to lack one: the GC-time finalizer is armed on the line *after* the
+  drain, so a failure there left the statement allocated with no
+  fallback of any kind.
+
+The guard is now one helper rather than six hand-rolled copies, which
+also fixed a defect in the copies that had a cursor to close: CLOSE and
+RELEASE shared a single ``contextlib.suppress`` block, so a CLOSE that
+raised skipped the RELEASE — losing the half that actually matters.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import informix_db
+from informix_db.cursors import _RELEASE_PDU, Cursor
+from tests.conftest import ConnParams
+
+# ---------------------------------------------------------------------------
+# The helper — no server needed
+# ---------------------------------------------------------------------------
+
+
+class _FakeConn:
+    def __init__(self) -> None:
+        self.sent: list[bytes] = []
+
+    def _send_pdu(self, pdu: bytes) -> None:
+        self.sent.append(pdu)
+
+
+class _FakeCursor:
+    """Duck-types the four attributes ``_release_after_failure`` touches."""
+
+    def __init__(self, *, close_raises: bool = False) -> None:
+        self._conn = _FakeConn()
+        self._close_raises = close_raises
+
+    def _build_close_pdu(self) -> bytes:
+        return b"CLOSE"
+
+    def _build_release_pdu(self) -> bytes:
+        return _RELEASE_PDU
+
+    def _drain_to_eot(self) -> None:
+        if self._close_raises and self._conn.sent[-1] == b"CLOSE":
+            raise OSError("wire went away mid-close")
+
+
+def test_release_is_sent_even_when_close_fails() -> None:
+    """The regression in the hand-rolled copies. CLOSE and RELEASE shared
+    one suppress block, so a failing CLOSE swallowed the RELEASE — and a
+    lost cursor handle is a nuisance while a lost statement breaks the
+    next call."""
+    cur = _FakeCursor(close_raises=True)
+    Cursor._release_after_failure(cur, close_cursor=True)
+    assert _RELEASE_PDU in cur._conn.sent, (
+        "a CLOSE that raises must not prevent the RELEASE"
+    )
+
+
+def test_close_is_skipped_when_no_cursor_was_opened() -> None:
+    cur = _FakeCursor()
+    Cursor._release_after_failure(cur)
+    assert cur._conn.sent == [_RELEASE_PDU]
+
+
+def test_cleanup_never_propagates() -> None:
+    """Cleanup runs inside an ``except``. If it raises, it replaces the
+    real SQL error with a secondary failure from the cleanup path, which
+    is strictly worse for the caller."""
+
+    class _Hostile(_FakeCursor):
+        def _drain_to_eot(self) -> None:
+            raise OSError("connection reset")
+
+    Cursor._release_after_failure(_Hostile(), close_cursor=True)
+
+
+# ---------------------------------------------------------------------------
+# Against a real server
+# ---------------------------------------------------------------------------
+
+
+def _connect(conn_params: ConnParams) -> 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,
+        # Bounded: a leaked statement used to manifest as a hang.
+        read_timeout=25.0,
+        autocommit=True,
+    )
+
+
+@pytest.mark.integration
+@pytest.mark.parametrize(
+    ("sql", "params"),
+    [
+        pytest.param(
+            "SELECT tabid FROM systables WHERE tabid = ?",
+            ("not-an-int",),
+            id="string-for-int",
+        ),
+        pytest.param(
+            "SELECT ?::INT FROM systables WHERE tabid = 1",
+            (2**40,),
+            id="out-of-range-int",
+        ),
+    ],
+)
+def test_server_rejected_bind_releases_statement(
+    conn_params: ConnParams, sql: str, params: tuple
+) -> None:
+    """A parameterized SELECT whose bind the *server* rejects. The value
+    encodes cleanly, so this lands past the build guard and in the drain
+    that had none. Repeated because a leak only shows on the call after
+    it — the first failure looks fine on its own."""
+    with _connect(conn_params) as conn:
+        cur = conn.cursor()
+        for i in range(4):
+            with pytest.raises(informix_db.Error):
+                cur.execute(sql, params)
+            cur.execute("SELECT FIRST 1 tabid FROM systables")
+            assert cur.fetchone() is not None, f"broken after {i + 1} binds"
+
+
+@pytest.mark.integration
+def test_failed_scroll_open_releases_statement(conn_params: ConnParams) -> None:
+    """FOR UPDATE prepares cleanly and fails at OPEN with -526, which is
+    exactly the branch that had no guard."""
+    with _connect(conn_params) as conn:
+        scroll = conn.cursor(scrollable=True)
+        for i in range(4):
+            with pytest.raises(informix_db.Error):
+                scroll.execute("SELECT tabid FROM systables FOR UPDATE")
+            assert not scroll._server_cursor_open, (
+                "a failed open must not leave the cursor marked live"
+            )
+            other = conn.cursor()
+            other.execute("SELECT FIRST 1 tabid FROM systables")
+            assert other.fetchone() is not None, f"broken after {i + 1} opens"
+            other.close()
+        scroll.close()
+
+
+@pytest.mark.integration
+def test_scrollable_cursor_still_works_after_a_failed_open(
+    conn_params: ConnParams,
+) -> None:
+    """The same cursor object must be reusable — a failed open is an
+    ordinary error, not a terminal state for the cursor."""
+    with _connect(conn_params) as conn:
+        scroll = conn.cursor(scrollable=True)
+        with pytest.raises(informix_db.Error):
+            scroll.execute("SELECT tabid FROM systables FOR UPDATE")
+        scroll.execute("SELECT tabid FROM systables ORDER BY tabid")
+        assert scroll.fetch_first() is not None
+        scroll.close()

tests/test_statement_slot.py

@@ -0,0 +1,355 @@
+"""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

tests/test_transaction_state.py

@@ -0,0 +1,191 @@
+"""Transaction control run as SQL, and the flag that didn't notice.
+
+``Connection._in_transaction`` decides whether ``commit()`` and
+``rollback()`` send anything at all, and the pool reads it to decide
+whether a returned connection needs cleaning up. It was maintained
+solely by the driver's own implicit ``SQ_BEGIN``, so a caller who wrote
+``cursor.execute("BEGIN WORK")`` — an entirely reasonable thing to
+write — walked straight past it.
+
+With autocommit on, nothing stopped that statement reaching the server.
+A transaction opened, the flag stayed False, and ``rollback()`` returned
+successfully having sent nothing. The rows it was asked to discard were
+still there. The connection then went back to the pool holding an open
+transaction and its locks, because the pool's cleanup is guarded by the
+same flag.
+
+With autocommit off it failed instead, and for a sillier reason: the
+driver's implicit ``SQ_BEGIN`` fired first, so the caller's ``BEGIN
+WORK`` got ``-535``, "already in transaction". The driver and the user
+competing to open the same transaction, and the user losing.
+
+The server labels these statements: type 34 for BEGIN, 35 for COMMIT, 36
+for ROLLBACK, with and without the ``WORK`` keyword. JDBC reads the same
+three values off the describe and calls ``setTxBeginState`` /
+``setTxEndState``. It also calls ``initiateTransaction`` *after* the
+describe rather than before, which is what makes the skip possible —
+until the describe lands you can't know the statement is transaction
+control.
+"""
+
+from __future__ import annotations
+
+import contextlib
+
+import pytest
+
+import informix_db
+from informix_db.cursors import _TX_CONTROL_TYPES
+from tests.conftest import ConnParams
+
+pytestmark = pytest.mark.integration
+
+
+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,
+        **kw,
+    )
+
+
+def test_transaction_control_types_are_what_the_server_says(
+    logged_db_params: ConnParams,
+) -> None:
+    """Pin the three constants against a live server rather than trusting
+    the decompiled source. Both spellings must map to the same type."""
+    with _connect(logged_db_params, autocommit=True) as conn:
+        cur = conn.cursor()
+        seen = {}
+        for sql in ("BEGIN WORK", "COMMIT WORK", "ROLLBACK WORK",
+                    "BEGIN", "COMMIT", "ROLLBACK"):
+            with conn._wire_lock:
+                conn._send_pdu(
+                    cur._build_prepare_pdu(sql, num_qmarks=0),
+                    statement_boundary=True,
+                )
+                cur._read_describe_response()
+                cur._release_after_failure()
+            seen[sql] = cur._statement_type
+        assert seen["BEGIN WORK"] == seen["BEGIN"] == 34
+        assert seen["COMMIT WORK"] == seen["COMMIT"] == 35
+        assert seen["ROLLBACK WORK"] == seen["ROLLBACK"] == 36
+        assert set(seen.values()) == _TX_CONTROL_TYPES
+
+
+def test_rollback_after_sql_begin_actually_rolls_back(
+    logged_db_params: ConnParams,
+) -> None:
+    """The data-loss case. rollback() reported success and sent nothing,
+    so the row it was asked to discard survived."""
+    with _connect(logged_db_params, autocommit=True) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_txstate")
+        cur.execute("CREATE TABLE t_txstate (k INT)")
+        try:
+            cur.execute("BEGIN WORK")
+            assert conn._in_transaction, "SQL BEGIN must set the flag"
+            cur.execute("INSERT INTO t_txstate VALUES (1)")
+            conn.rollback()
+            cur.execute("SELECT COUNT(*) FROM t_txstate")
+            assert cur.fetchone() == (0,), "rollback() was a silent no-op"
+            assert not conn._in_transaction
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_txstate")
+
+
+def test_sql_commit_clears_the_flag(logged_db_params: ConnParams) -> None:
+    """The mirror image: with the flag stuck True after a SQL COMMIT, the
+    next rollback() would send SQ_RBWORK with no transaction open and
+    draw -255."""
+    with _connect(logged_db_params, autocommit=True) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_txstate2")
+        cur.execute("CREATE TABLE t_txstate2 (k INT)")
+        try:
+            cur.execute("BEGIN WORK")
+            cur.execute("INSERT INTO t_txstate2 VALUES (1)")
+            cur.execute("COMMIT WORK")
+            assert not conn._in_transaction, "SQL COMMIT must clear the flag"
+            conn.rollback()  # must be a no-op, not a -255
+            cur.execute("SELECT COUNT(*) FROM t_txstate2")
+            assert cur.fetchone() == (1,), "the committed row must survive"
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_txstate2")
+
+
+def test_sql_begin_does_not_collide_with_the_implicit_one(
+    logged_db_params: ConnParams,
+) -> None:
+    """Non-autocommit. The driver's implicit SQ_BEGIN used to fire first
+    and the caller's BEGIN WORK then got -535."""
+    with _connect(logged_db_params, autocommit=False) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_txstate3")
+            conn.commit()
+        cur.execute("CREATE TABLE t_txstate3 (k INT)")
+        conn.commit()
+        try:
+            cur.execute("BEGIN WORK")
+            cur.execute("INSERT INTO t_txstate3 VALUES (1)")
+            conn.rollback()
+            cur.execute("SELECT COUNT(*) FROM t_txstate3")
+            assert cur.fetchone() == (0,)
+            conn.commit()
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_txstate3")
+                conn.commit()
+
+
+def test_ordinary_dml_still_opens_a_transaction(
+    logged_db_params: ConnParams,
+) -> None:
+    """_ensure_transaction moved from before the PREPARE to after the
+    describe. It still has to fire for everything that isn't transaction
+    control, or non-autocommit DML runs outside a transaction."""
+    with _connect(logged_db_params, autocommit=False) as conn:
+        cur = conn.cursor()
+        with contextlib.suppress(Exception):
+            cur.execute("DROP TABLE t_txstate4")
+            conn.commit()
+        cur.execute("CREATE TABLE t_txstate4 (k INT)")
+        conn.commit()
+        try:
+            assert not conn._in_transaction
+            cur.execute("INSERT INTO t_txstate4 VALUES (1)")
+            assert conn._in_transaction, "DML must open a transaction"
+            conn.rollback()
+            cur.execute("SELECT COUNT(*) FROM t_txstate4")
+            assert cur.fetchone() == (0,)
+            conn.commit()
+        finally:
+            with contextlib.suppress(Exception):
+                cur.execute("DROP TABLE t_txstate4")
+                conn.commit()
+
+
+def test_a_failed_transaction_statement_does_not_move_the_flag(
+    logged_db_params: ConnParams,
+) -> None:
+    """The sync runs on the success path only. A COMMIT that the server
+    rejects has not ended anything."""
+    with _connect(logged_db_params, autocommit=True) as conn:
+        cur = conn.cursor()
+        assert not conn._in_transaction
+        with pytest.raises(informix_db.Error):
+            cur.execute("COMMIT WORK")  # -255, nothing to commit
+        assert not conn._in_transaction
+        cur.execute("SELECT FIRST 1 tabid FROM systables")
+        assert cur.fetchone() is not None

uv.lock

@@ -34,7 +34,7 @@ wheels = [
 
 [[package]]
 name = "informix-driver"
-version = "2026.9.1"
+version = "2026.9.2"
 source = { editable = "." }
 
 [package.optional-dependencies]