fix: reject a timer_id over 255 characters before the INSERT

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)

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)

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"):