Ryan Malloy
5ad296419c
IfxSocket owns a read-ahead buffer that BufferedSocketReader fills and
drains. 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 looking at the buffer. Bytes sitting in the
buffer were skipped, and skipped bytes in a length-framed protocol do
not 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. read_exact now drains the
buffer first, which costs one branch on a cold path and makes the two
paths agree by construction rather than by luck.
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 reads as a 2 GB request and the fill loop sits in recv until
the read timeout while the buffer grows. It now refuses above
IFX_MAX_READ_BYTES (256 MiB default) with an error naming the number,
which is the actual diagnostic: a length that absurd means framing was
already lost upstream.
BufferedSocketReader.skip advanced the cursor arithmetically with no
guard, so a negative count rewound it and re-decoded consumed bytes as
the next field. The base reader's skip delegates to read_exact and does
guard; this one diverged.
Transaction control run as SQL desynced Connection._in_transaction,
which is what commit() and rollback() are guarded by and what the pool
reads to decide whether a returned connection needs cleaning up. With
autocommit on, cursor.execute("BEGIN WORK") opened a real transaction
while the flag stayed False, 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. With autocommit off it failed instead: the driver's implicit
SQ_BEGIN fired first and the caller's BEGIN WORK got -535.
The server labels these -- 34 BEGIN, 35 COMMIT, 36 ROLLBACK, with and
without the WORK keyword, measured on all three versions. JDBC reads the
same values off the describe and calls setTxBeginState/setTxEndState.
_ensure_transaction moves to after the describe, matching JDBC's
initiateTransaction placement, which is what makes it possible to skip
for transaction control at all -- until the describe lands you cannot
know that is what the statement is.
173 lines
6.7 KiB
Python
173 lines
6.7 KiB
Python
"""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"
|