"""Regression tests for error recovery, cursor lifecycle, and concurrency. The type-matrix suite covers what a value looks like on the wire. This covers everything around it: what happens after a statement fails, what happens to abandoned cursors, and what happens when more than one thread or task is involved. Two bugs here were found by fuzzing, and both are the kind that never show up in a single-threaded happy-path test: * **A failed statement was never released.** Successful DML sent PREPARE → EXECUTE → RELEASE; failing DML sent PREPARE → EXECUTE and stopped. The leaked statement collided with the next PREPARE, and every subsequent call on that connection returned a nonsense error whose offset pointed back at the *failed* SQL. A duplicate-key violation — about the most ordinary error an application can hit — killed the connection outright. * **Cancelling a pool acquire leaked the connection.** ``asyncio.to_thread`` cannot interrupt its worker, so a cancelled waiter left the worker to finish and hand back a connection nobody owned. Under HTTP load, where client disconnects cancel request tasks precisely while they wait for a connection, the pool dies one slot at a time. The tests that follow assert on *data* wherever they can, not merely the absence of an exception: a worker reads back a value only it wrote, so a crossed wire fails even when nothing raises. """ from __future__ import annotations import asyncio import contextlib import threading import pytest import informix_db from informix_db import aio 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, # Bounded: a leaked statement used to manifest as a hang. read_timeout=25.0, **kw, ) def _pool_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": 15.0, "read_timeout": 30.0, } # --------------------------------------------------------------------------- # Error recovery — a failed statement must not poison the connection # --------------------------------------------------------------------------- FAILING_SQL = [ pytest.param("SELECT FROM WHERE", id="syntax"), pytest.param("SELECT * FROM no_such_table_xyz", id="no_such_table"), pytest.param("SELECT no_such_col FROM systables", id="no_such_column"), pytest.param("SELECT no_such_function_xyz(1) FROM systables", id="bad_func"), pytest.param("SELECT 1/0 FROM systables", id="div_zero"), pytest.param("SELECT 'notanumber'::INT FROM systables", id="bad_cast"), ] @pytest.mark.parametrize("sql", FAILING_SQL) def test_connection_survives_failed_statement( conn_params: ConnParams, sql: str ) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() for _ in range(4): # repeat: a leak accumulates with pytest.raises(informix_db.Error): cur.execute(sql) cur.fetchall() cur.execute("SELECT FIRST 1 tabid FROM systables ORDER BY tabid") assert cur.fetchone() is not None @pytest.mark.parametrize("autocommit", [True, False]) def test_duplicate_key_does_not_kill_the_connection( conn_params: ConnParams, autocommit: bool ) -> None: """The bug that started this file. A unique-constraint violation left the prepared statement allocated server-side; the next statement then failed with a nonsense error (-255 "Not in transaction" in autocommit, -285 otherwise) and the connection never recovered.""" with _connect(conn_params, autocommit=autocommit) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_dup (k INT PRIMARY KEY)") cur.execute("INSERT INTO t_dup VALUES (1)") if not autocommit: conn.commit() for i in range(5): with pytest.raises(informix_db.IntegrityError): cur.execute("INSERT INTO t_dup VALUES (1)") if not autocommit: conn.rollback() cur.execute("SELECT k FROM t_dup") assert cur.fetchall() == [(1,)], f"broken after {i + 1} violations" def test_failed_dml_with_params_releases_statement( conn_params: ConnParams, ) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_nn (s VARCHAR(8) NOT NULL, k INT)") for _ in range(4): with pytest.raises(informix_db.Error): cur.execute("INSERT INTO t_nn VALUES (?, ?)", (None, 1)) cur.execute("SELECT COUNT(*) FROM t_nn") assert cur.fetchone() == (0,) def test_new_cursor_works_after_error(conn_params: ConnParams) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() with pytest.raises(informix_db.Error): cur.execute("SELECT * FROM no_such_table_xyz") other = conn.cursor() other.execute("SELECT FIRST 1 tabid FROM systables") assert other.fetchone() is not None other.close() # --------------------------------------------------------------------------- # Fetch batching — NFETCH is a byte budget, so batch edges move with width # --------------------------------------------------------------------------- @pytest.mark.parametrize("n", [0, 1, 2, 3, 63, 64, 65, 199, 200, 201, 512, 1025]) def test_row_count_exact_across_fetch_styles( conn_params: ConnParams, n: int ) -> None: """Every fetch style must agree, at counts that straddle plausible server batch boundaries. A batching off-by-one shows up as a wrong count, a duplicate, or a dropped row.""" with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() cur.execute( "CREATE TEMP TABLE t_batch " "(k INT, a VARCHAR(32), b INT, c VARCHAR(32), d INT)" ) if n: cur.executemany( "INSERT INTO t_batch VALUES (?, ?, ?, ?, ?)", [(i, f"a{i}", i * 2, f"c{i}", i * 3) for i in range(n)], ) want = [(i,) for i in range(n)] sql = "SELECT k FROM t_batch ORDER BY k" cur.execute(sql) assert cur.fetchall() == want cur.execute(sql) got = [] while (row := cur.fetchone()) is not None: got.append(row) assert got == want for size in (1, 7, 100): cur.execute(sql) got = [] while chunk := cur.fetchmany(size): got.extend(chunk) assert got == want, f"fetchmany({size}) disagreed at n={n}" cur.execute(sql) assert list(cur) == want # --------------------------------------------------------------------------- # Cursor lifecycle # --------------------------------------------------------------------------- def test_abandoned_partial_fetches_do_not_leak( conn_params: ConnParams, ) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_aband (k INT)") cur.executemany( "INSERT INTO t_aband VALUES (?)", [(i,) for i in range(200)] ) for i in range(25): c = conn.cursor() c.execute("SELECT k FROM t_aband ORDER BY k") c.fetchone() if i % 2: c.close() else: del c # rely on the finalizer cur.execute("SELECT COUNT(*) FROM t_aband") assert cur.fetchone() == (200,) def test_re_execute_mid_fetch(conn_params: ConnParams) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_reex (k INT)") cur.executemany( "INSERT INTO t_reex VALUES (?)", [(i,) for i in range(100)] ) for _ in range(10): cur.execute("SELECT k FROM t_reex ORDER BY k") cur.fetchone() # abandon mid-fetch cur.execute("SELECT k FROM t_reex ORDER BY k") assert len(cur.fetchall()) == 100 def test_interleaved_cursors_on_one_connection( conn_params: ConnParams, ) -> None: with _connect(conn_params, autocommit=True) as conn: cur = conn.cursor() cur.execute("CREATE TEMP TABLE t_inter (k INT)") cur.executemany( "INSERT INTO t_inter VALUES (?)", [(i,) for i in range(50)] ) a, b = conn.cursor(), conn.cursor() a.execute("SELECT k FROM t_inter ORDER BY k") b.execute("SELECT k FROM t_inter ORDER BY k DESC") assert [a.fetchone() for _ in range(5)] == [(i,) for i in range(5)] assert [b.fetchone() for _ in range(5)] == [ (i,) for i in range(49, 44, -1) ] a.close() b.close() # --------------------------------------------------------------------------- # Concurrency # --------------------------------------------------------------------------- def test_threads_sharing_one_connection_do_not_interleave( conn_params: ConnParams, ) -> None: """One connection is one socket. Each thread reads back a value only it supplied, so crossed wires fail even if nothing raises.""" failures: list[str] = [] lock = threading.Lock() with _connect(conn_params, autocommit=True) as conn: def worker(tid: int) -> None: try: for r in range(10): cur = conn.cursor() token = tid * 1000 + r cur.execute( "SELECT FIRST 1 ?::INT, ?::VARCHAR(16) FROM systables", (token, f"t{tid}"), ) row = cur.fetchone() if row != (token, f"t{tid}"): with lock: failures.append(f"thread {tid}: got {row!r}") return cur.close() except Exception as exc: with lock: failures.append(f"thread {tid}: {type(exc).__name__}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(6)] for t in threads: t.start() for t in threads: t.join() assert not failures, failures def test_pool_hands_out_clean_connections_under_load( conn_params: ConnParams, ) -> None: """Borrowers that fail must still return a usable connection.""" failures: list[str] = [] lock = threading.Lock() pool = informix_db.create_pool( **_pool_kw(conn_params), min_size=2, max_size=4, acquire_timeout=30.0, autocommit=True, ) try: def worker(tid: int) -> None: try: for r in range(8): with pool.connection() as conn: cur = conn.cursor() if r % 3 == 0: with contextlib.suppress(informix_db.Error): cur.execute("SELECT * FROM no_such_tbl_xyz") token = tid * 1000 + r cur.execute( "SELECT FIRST 1 ?::INT FROM systables", (token,) ) got = cur.fetchone() if got != (token,): with lock: failures.append( f"thread {tid}: got {got!r}, want {token}" ) return cur.close() except Exception as exc: with lock: failures.append(f"thread {tid}: {type(exc).__name__}: {exc}") threads = [threading.Thread(target=worker, args=(i,)) for i in range(6)] for t in threads: t.start() for t in threads: t.join() finally: with contextlib.suppress(Exception): pool.close() assert not failures, failures def test_pool_does_not_leak_transactions_between_borrowers( conn_params: ConnParams, ) -> None: """max_size=1 so both borrowers get the same underlying connection.""" pool = informix_db.create_pool( **_pool_kw(conn_params), min_size=1, max_size=1, acquire_timeout=30.0, autocommit=False, ) try: with pool.connection() as conn: cur = conn.cursor() with contextlib.suppress(Exception): cur.execute("DROP TABLE t_pool_txn") cur.execute("CREATE TABLE t_pool_txn (k INT)") conn.commit() try: for i in range(5): with pool.connection() as conn: # dirties, never commits conn.cursor().execute( "INSERT INTO t_pool_txn VALUES (?)", (i,) ) with pool.connection() as conn: # must not see it cur = conn.cursor() cur.execute("SELECT COUNT(*) FROM t_pool_txn") assert cur.fetchone() == (0,), "transaction leaked" conn.rollback() finally: with pool.connection() as conn: cur = conn.cursor() with contextlib.suppress(Exception): cur.execute("DROP TABLE t_pool_txn") conn.commit() finally: with contextlib.suppress(Exception): pool.close() # --------------------------------------------------------------------------- # Async # --------------------------------------------------------------------------- async def test_async_pool_survives_cancelled_acquires( conn_params: ConnParams, ) -> None: """Cancelling a task blocked on acquire used to destroy that pool slot permanently: ``asyncio.to_thread`` cannot interrupt its worker, so the worker finished and handed back a connection nobody owned. Two holders keep both connections checked out; three waiters block on acquire and are cancelled there. Afterwards the pool must still serve. """ pool = await aio.create_pool( **_pool_kw(conn_params), min_size=1, max_size=2, autocommit=True ) try: release = asyncio.Event() async def holder(i: int) -> None: async with pool.connection() as conn: cur = await conn.cursor() await cur.execute( "SELECT FIRST 1 ?::INT FROM systables", (i,) ) await cur.fetchone() await release.wait() async def waiter(i: int) -> tuple | None: async with pool.connection() as conn: cur = await conn.cursor() await cur.execute( "SELECT FIRST 1 ?::INT FROM systables", (i,) ) return await cur.fetchone() for cycle in range(2): release.clear() holders = [asyncio.create_task(holder(i)) for i in range(2)] await asyncio.sleep(0.5) # both connections held waiters = [asyncio.create_task(waiter(100 + i)) for i in range(3)] await asyncio.sleep(0.5) # all blocked on acquire for t in waiters: t.cancel() await asyncio.gather(*waiters, return_exceptions=True) release.set() await asyncio.gather(*holders, return_exceptions=True) await asyncio.sleep(0.3) # orphan returns land got = await asyncio.wait_for(waiter(200 + cycle), timeout=15) assert got == (200 + cycle,), ( f"pool starved after cycle {cycle} — cancelled acquires leaked" ) finally: with contextlib.suppress(Exception): await pool.close() async def test_async_pool_concurrent_tasks_get_their_own_data( conn_params: ConnParams, ) -> None: pool = await aio.create_pool( **_pool_kw(conn_params), min_size=2, max_size=4, autocommit=True ) try: async def worker(tid: int) -> None: for r in range(6): async with pool.connection() as conn: cur = await conn.cursor() token = tid * 1000 + r await cur.execute( "SELECT FIRST 1 ?::INT FROM systables", (token,) ) assert await cur.fetchone() == (token,) await asyncio.gather(*(worker(i) for i in range(6))) finally: with contextlib.suppress(Exception): await pool.close()