Rows never checked that they consumed their own payload

Fourteen framing bugs shipped before this. Every one was a column
reading the wrong number of bytes, and not one of them 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 entirely.

All fourteen were detectable for free. The payload is a fully extracted
bytes object of known length, so after decoding N columns the offset has
to land exactly on the end. It didn't, and nobody looked.

_row_not_consumed now raises when it doesn't, naming the column shape
and the byte delta, and it is wired into all three decode paths --
readers fast path, legacy chain, and the generated decoder.

Turning it on found two more bugs on the first run:

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 a BLOB or CLOB anywhere but the final
column position shifted every column after it. The 144 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 -- unnoticed because
read_blob_column resolves through the server's lotofile and never reads
the locator.

NULL composite UDTs skipped their length field, byte for byte the
LVARCHAR NULL bug in a branch twelve lines away.

Both are the same [indicator][int32 length][data] envelope, hand-written
a third and fourth time. _read_udt_envelope is now the only copy, and it
rejects a negative length instead of rewinding the offset into bytes it
already decoded.

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
 # ---------------------------------------------------------------------------

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
+        # Smart LOBs use the UDT envelope, NOT a flat encoded_length field.
-        raw = payload[offset:offset + width]
+        # Measured on Informix 15: a populated BLOB column is 149 bytes
-        offset += width
+        # ([ind=0][len=144][144 hex chars]) and a NULL one is 5, while
-        cls = BlobLocator if col.extended_id == 10 else ClobLocator
+        # encoded_length reports 72. Consuming 72 left 77 bytes behind and
-        return offset, cls(raw=bytes(raw))
+        # 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, body = _read_udt_envelope(payload, offset)
-        offset += 1
+        if body is None:
-        length = int.from_bytes(payload[offset:offset + 4], "big", signed=True)
-        offset += 4
-        raw = payload[offset:offset + length]
-        offset += length
-        if indicator == 1:
             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:
+        # The 4-byte length is part of the envelope and is present even when
-            return offset, None
+        # 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
+            # UDT envelope, not a flat encoded_length field — see the
-            raw = payload[offset:offset + width]
+            # matching branch in _legacy_dispatch_one_column.
-            offset += width
+            offset, body = _read_udt_envelope(payload, offset)
-            cls = BlobLocator if col.extended_id == 10 else ClobLocator
+            values.append(
-            values.append(cls(raw=bytes(raw)))
+                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, body = _read_udt_envelope(payload, offset)
-            offset += 1
+            values.append(
-            length = int.from_bytes(
+                None if body is None
-                payload[offset:offset + 4], "big", signed=True
+                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
+            # Length is present even when NULL — see the matching branch in
-                values.append(None)
+            # _legacy_dispatch_one_column for the wire evidence.
-                continue
             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/connections.py

@@ -38,7 +38,13 @@ 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
@@ -402,7 +408,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,7 +418,7 @@ class Connection:
             try:
                 self._sock.write_all(pdu)
                 self._drain_to_eot()
-            except (ProtocolError, OSError, OperationalError):
+            except WIRE_ERRORS:
                 # Wire is unrecoverable; force-close. Subsequent
                 # ``_send_pdu`` will raise InterfaceError. Server
                 # cleanup of the remaining queued entries happens
@@ -835,7 +840,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 +855,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 +869,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,
@@ -120,7 +125,6 @@ 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
@@ -147,7 +151,7 @@ def _finalize_cursor(
             conn._drain_to_eot()
             conn._send_pdu(_RELEASE_PDU)
             conn._drain_to_eot()
-        except (ProtocolError, OSError) as exc:
+        except WIRE_ERRORS as exc:
             # Wire desync during cleanup — same doctrine as
             # ``_raise_sq_err``: the wire is unrecoverable, force-close
             # the connection. Asymmetric handling of the same failure
@@ -1764,7 +1768,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 +1785,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 +1797,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_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