diff --git a/src/informix_db/_protocol.py b/src/informix_db/_protocol.py index e9df9c1..b5fb043 100644 --- a/src/informix_db/_protocol.py +++ b/src/informix_db/_protocol.py @@ -292,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 diff --git a/src/informix_db/_socket.py b/src/informix_db/_socket.py index e77faf9..5b7b226 100644 --- a/src/informix_db/_socket.py +++ b/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 diff --git a/src/informix_db/cursors.py b/src/informix_db/cursors.py index c8240d3..fde6080 100644 --- a/src/informix_db/cursors.py +++ b/src/informix_db/cursors.py @@ -108,6 +108,15 @@ def _make_socket_reader(sock): _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. @@ -423,12 +432,6 @@ class Cursor: self._statement_type = 0 self._statement_already_done = False - # 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() - # 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. @@ -438,6 +441,21 @@ class Cursor: ) 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. + # + # 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() + # 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)): @@ -450,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). @@ -458,6 +481,30 @@ 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``. @@ -1132,10 +1179,6 @@ class Cursor: 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), @@ -1155,6 +1198,11 @@ class Cursor: "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 diff --git a/tests/test_socket_reads.py b/tests/test_socket_reads.py new file mode 100644 index 0000000..c34acde --- /dev/null +++ b/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" diff --git a/tests/test_transaction_state.py b/tests/test_transaction_state.py new file mode 100644 index 0000000..d99e551 --- /dev/null +++ b/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