From da54e1f06a8908071be74472fda9aada58f65398 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 20:17:57 +0300 Subject: [PATCH] fix: reject a timer_id over 255 characters before the INSERT SQLAlchemy 2.0.x binds String(255) values as $1::VARCHAR(255), and Postgres truncates an explicit cast instead of rejecting it, so an over-long timer_id was stored truncated and could dedup against a different timer sharing its first 255 characters. Newer SQLAlchemy raised an opaque DBAPIError instead. Validate it with the queue. --- faststream_outbox/response.py | 8 +++++++- tests/test_integration.py | 14 ++++++++++++++ tests/test_unit.py | 16 ++++++++++++++++ 3 files changed, 37 insertions(+), 1 deletion(-) diff --git a/faststream_outbox/response.py b/faststream_outbox/response.py index a9e612f..1af3844 100644 --- a/faststream_outbox/response.py +++ b/faststream_outbox/response.py @@ -22,7 +22,7 @@ from faststream_outbox.message import CORRELATION_ID_HEADER -# Matches the ``queue`` column width in ``make_outbox_table`` (``String(255)``). +# Matches the ``queue`` and ``timer_id`` column width in ``make_outbox_table`` (``String(255)``). _MAX_QUEUE_LENGTH = 255 # Single source for the ``request()`` rejection message, shared by the broker, producer, @@ -37,6 +37,7 @@ def _validate_publish_args( session: object, activate_in: dt.timedelta | None, activate_at: dt.datetime | None, + timer_id: str | None = None, ) -> None: """Fail-fast validation shared by every real outbox publish entry point. @@ -68,6 +69,9 @@ def _validate_publish_args( if len(queue) > _MAX_QUEUE_LENGTH: msg = f"queue must be at most {_MAX_QUEUE_LENGTH} characters (got {len(queue)})" raise ValueError(msg) + if timer_id is not None and len(timer_id) > _MAX_QUEUE_LENGTH: + msg = f"timer_id must be at most {_MAX_QUEUE_LENGTH} characters (got {len(timer_id)})" + raise ValueError(msg) # Marker exception raised by programming guards on the dispatch path (e.g. the @@ -101,6 +105,7 @@ def __init__( session=session, activate_in=activate_in, activate_at=activate_at, + timer_id=timer_id, ) # P4: timer_id / correlation_id are per-row single-publish concepts the batch # path silently drops (each batched row gets its own auto correlation_id and @@ -203,6 +208,7 @@ def __init__( session=session, activate_in=activate_in, activate_at=activate_at, + timer_id=timer_id, ) if correlation_id is None and headers: correlation_id = headers.get(CORRELATION_ID_HEADER) diff --git a/tests/test_integration.py b/tests/test_integration.py index 24d96aa..afbdadf 100644 --- a/tests/test_integration.py +++ b/tests/test_integration.py @@ -2482,6 +2482,20 @@ async def test_broker_without_engine_publishes_on_the_callers_session( assert len(result.scalars().all()) == 3 +async def test_publish_rejects_an_oversize_timer_id_without_storing_a_row( + pg_engine: AsyncEngine, outbox_table: Table +) -> None: + """SQLAlchemy 2.0.x binds String(255) as ``$1::VARCHAR(255)``, which Postgres truncates instead of rejecting.""" + broker = OutboxBroker(pg_engine, outbox_table=outbox_table) + session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) + async with session_factory() as session, session.begin(): + with pytest.raises(ValueError, match="timer_id must be at most 255 characters"): + await broker.publish({"n": 1}, queue="orders", session=session, timer_id="t" * 256) + async with pg_engine.connect() as conn: + result = await conn.execute(select(outbox_table.c.id)) + assert result.scalars().all() == [] + + async def test_notify_deduped_within_transaction_same_queue(pg_engine, outbox_table) -> None: broker = OutboxBroker(pg_engine, outbox_table=outbox_table) session_factory = async_sessionmaker(pg_engine, expire_on_commit=False) diff --git a/tests/test_unit.py b/tests/test_unit.py index 59e6f48..69ec336 100644 --- a/tests/test_unit.py +++ b/tests/test_unit.py @@ -890,6 +890,22 @@ def test_publish_command_rejects_oversize_queue() -> None: OutboxPublishCommand(b"x", queue="q" * 256, session=_make_session_mock()) +def test_publish_command_rejects_oversize_timer_id() -> None: + """timer_id over the String(255) column limit is rejected before SQLAlchemy can truncate it.""" + with pytest.raises(ValueError, match="timer_id must be at most 255 characters"): + OutboxPublishCommand(b"x", queue="q", session=_make_session_mock(), timer_id="t" * 256) + + +def test_publish_command_accepts_timer_id_at_the_column_limit() -> None: + cmd = OutboxPublishCommand(b"x", queue="q", session=_make_session_mock(), timer_id="t" * 255) + assert cmd.timer_id == "t" * 255 + + +def test_outbox_response_rejects_oversize_timer_id() -> None: + with pytest.raises(ValueError, match="timer_id must be at most 255 characters"): + OutboxResponse(b"x", queue="q", session=_make_session_mock(), timer_id="t" * 256) + + def test_publish_command_rejects_timer_id_for_batch() -> None: """P4: timer_id is meaningless for a batch (multiple bodies) — reject, don't silently drop.""" with pytest.raises(ValueError, match="batch"):