From d80f8011af28d9c8a89053a85da06c383c82a901 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 21:31:54 +0300 Subject: [PATCH 1/2] refactor: build the server-side seconds interval in one helper --- faststream_outbox/client.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/faststream_outbox/client.py b/faststream_outbox/client.py index 6b87420..83fde9d 100644 --- a/faststream_outbox/client.py +++ b/faststream_outbox/client.py @@ -187,6 +187,11 @@ async def ping(self) -> bool: ... ) +def _seconds(name: str) -> "ColumnElement[typing.Any]": + """``make_interval(secs => :name)``, computed by Postgres so worker/DB clock skew can't shift it.""" + return func.make_interval(0, 0, 0, 0, 0, 0, bindparam(name, type_=Float)) + + def _build_fetch_stmt(outbox: "Table") -> "Update": """Build the claim statement once; queues, limit, lease TTL and the new token are bind parameters. @@ -196,8 +201,7 @@ def _build_fetch_stmt(outbox: "Table") -> "Update": """ t = outbox limit = bindparam("limit", type_=Integer) - # ``make_interval(secs => :lease_ttl)`` keeps the cutoff server-side, immune to worker/DB clock skew. - lease_cutoff = func.now() - func.make_interval(0, 0, 0, 0, 0, 0, bindparam("lease_ttl", type_=Float)) + lease_cutoff = func.now() - _seconds("lease_ttl") queue_array = ARRAY(String) queue_name = func.unnest(cast(bindparam("queues", type_=queue_array), queue_array)).label("name") queues = select(queue_name).distinct().cte("queues") @@ -252,9 +256,7 @@ def _build_insert_stmts(table: "Table", channel: str) -> "dict[tuple[bool, bool, The notify variants call ``pg_notify`` in ``RETURNING``, so the NOTIFY costs no extra round trip and fires only when a row actually landed (an ``ON CONFLICT DO NOTHING`` skip returns no row). """ - # Server-side compute keeps timing immune to worker/DB clock skew (mirrors - # client.mark_pending_with_lease). - activate_in = func.now() + func.make_interval(0, 0, 0, 0, 0, 0, bindparam("activate_in_seconds", type_=Float)) + activate_in = func.now() + _seconds("activate_in_seconds") plain = insert(table) timer = pg_insert(table).on_conflict_do_nothing( index_elements=[table.c.queue, table.c.timer_id], @@ -604,7 +606,7 @@ async def mark_pending_with_lease( msg = "OutboxClient.mark_pending_with_lease requires a live AsyncConnection (got None)" raise TypeError(msg) t = self._table - next_attempt_at_expr = func.now() + func.make_interval(0, 0, 0, 0, 0, 0, bindparam("delay", type_=Float)) + next_attempt_at_expr = func.now() + _seconds("delay") stmt = ( update(t) .where(t.c.id == message_id, t.c.acquired_token == acquired_token) From bf2da54d14f20872b3a2ded8b522295d095f9a89 Mon Sep 17 00:00:00 2001 From: Artur Shiriev Date: Sat, 17 Oct 2026 22:18:50 +0300 Subject: [PATCH 2/2] test: track greenlet switches in coverage SQLAlchemy's async layer runs each statement in a greenlet, and the tracer drops lines that run after an untracked switch. On Python 3.11, which can't use sysmon, test_batched_flush_dlq_row_stays_inline lost a line in about a third of runs and failed the 100% gate. --- pyproject.toml | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 9a22af1..ec39723 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -149,9 +149,8 @@ Changelog = "https://github.com/modern-python/faststream-outbox/releases" [tool.coverage.run] omit = ["benchmarks/*"] -# The C tracer intermittently drops executed lines in async tests; 3.11 falls back to it. -core = "sysmon" -disable_warnings = ["no-sysmon"] +# SQLAlchemy's async layer runs statements in greenlets; untracked switches drop executed lines. +concurrency = ["greenlet", "thread"] [tool.coverage.report] fail_under = 100