UN-3712 [FEAT] PG-queue: Redis-signalled result delivery (remove result poll-storm)#2168
Conversation
|
Important Review skippedAuto reviews are disabled on base/target branches other than the default branch. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
| Filename | Overview |
|---|---|
| workers/queue_backend/pg_queue/result_backend.py | Adds Redis wake-up signaling, deadline-aware polling fallback, and recovery after client construction failures. |
| workers/tests/test_pg_result_signal.py | Adds unit and integration coverage for signaling, fallback, timeout, and recovery behavior. |
| docker/sample.env | Documents the optional Redis result-signal backend and its polling default. |
Reviews (3): Last reviewed commit: "UN-3712 [FIX] address PR review — BLPOP ..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
…lt poll-storm) Replace the per-in-flight-task DB poll in the PG-queue result backend with an optional Redis wake-up signal (RPUSH producer / BLPOP consumer), gated by PG_RESULT_SIGNAL_BACKEND (default poll = unchanged behaviour). The result stays durable in Postgres; only the reply key rides Redis, so DB load scales with throughput instead of concurrency. Falls back to the backoff poll on any Redis error, so correctness never depends on the signal. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
muhammad-ali-e
force-pushed
the
feat/UN-3445-pg-result-redis-signal
branch
from
bc63550 to
86f7565
Compare
muhammad-ali-e
left a comment
muhammad-ali-e
left a comment
There was a problem hiding this comment.
PR Review Toolkit — automated multi-agent review (code-reviewer, silent-failure-hunter, type-design-analyzer, test-analyzer, comment-analyzer, code-simplifier).
The design is solid: DB stays the source of truth, the signal is best-effort, and correctness never depends on Redis — a lost/absent token costs only latency. No correctness strand was found. The findings below are latency-regression, error-taxonomy, typing, test-coverage, and comment-accuracy issues.
Note on duplicates: the two existing greptile P2 comments — "Socket Timeout Aborts BLPOP" (line 191) and "Transient Failure Disables Signaling" (line 193) — are the two highest-impact issues here and I independently confirmed both (the BLPOP one against redis-py 5.2.1 Connection.read_response, which applies socket_timeout to blocking reads). I have not re-posted them to avoid duplication; the comment below on the round-trip test is the new companion finding to the socket-timeout issue.
Comments are grouped in this single review, ordered roughly by priority.
| @pytest.fixture | ||
| def real_redis(self): | ||
| redis = pytest.importorskip("redis") | ||
| client = redis.Redis(host="localhost", port=6379, socket_connect_timeout=2) |
There was a problem hiding this comment.
[High] The only live-BLPOP test masks the socket_timeout=5 bug (companion to greptile's Socket Timeout Aborts BLPOP).
This round-trip test builds its own client redis.Redis(host=..., socket_connect_timeout=2) with no socket_timeout (defaults to None = block forever) and injects it, bypassing the production _get_result_redis_client() which carries socket_timeout=5. It also stores after only time.sleep(0.5), well inside 5s. So the test stays green while the production config aborts every BLPOP that blocks >5s.
Fix: exercise a client with the production socket_timeout=5 and add a case where the token arrives after socket_timeout seconds (store after ~6s, short fallback constant) to prove BLPOP still wakes rather than degrading to poll. That test fails today and pins the greptile fix.
| client.blpop.return_value = None # BLPOP timed out, no token | ||
| monkeypatch.setattr(rb_mod, "_get_result_redis_client", lambda: client) | ||
| rb = _backend_with_get_result(None) # never ready | ||
| assert rb.wait_for_result("k", timeout=0.05) is None |
There was a problem hiding this comment.
[Medium] The central resilience property is untested: a missed signal must still deliver via the backstop SELECT.
test_timeout_returns_none has blpop→None and get_result→None forever, so it only proves timeout. Nothing exercises the design's core claim (module docstring L30-33): BLPOP misses/returns None (lost token / fallback tick) but the authoritative get_result on the next loop iteration finds the committed row and returns it.
Add: client.blpop.return_value=None, get_result side_effect [None, _ROW], generous timeout → assert it returns _ROW and blpop was called (i.e. it went through the loop, not the immediate race check). This catches any refactor that treats a None BLPOP as "give up" and strands the caller.
| row = self.get_result(task_id) | ||
| if row is not None: | ||
| return row | ||
| client = _get_result_redis_client() |
There was a problem hiding this comment.
[Medium] First-call Redis client build runs inline on the hot path and can ignore the caller's timeout (Sentinel mode).
_get_result_redis_client() is called here (and in store_result→_signal_ready after the DB commit). In Sentinel mode create_redis_client pings-and-retries for up to ~5 minutes before raising. So a wait_for_result(timeout=30) can actually block for minutes on its first wait during a sentinel blip, and store_result can stall the executor consumer for minutes after a commit — the build ignores the caller's deadline entirely.
Consider building the client at worker init (off the request path), or bounding the first build by remaining here. Best fixed together with the transient-disable issue greptile flagged on L193.
| block = min(remaining, _REDIS_FALLBACK_POLL_SECONDS) | ||
| try: | ||
| client.blpop([key], timeout=block) | ||
| except Exception: |
There was a problem hiding this comment.
[Medium] except Exception around BLPOP conflates expected timeouts and permanent misconfig, both logged as a transient "blip".
This catches redis.TimeoutError (normal: no token yet), but also AuthenticationError, ResponseError/WRONGTYPE/NOPERM, and unrelated programming errors — all degraded to poll with an identical WARNING. A systematic failure (wrong REDIS_PASSWORD, revoked ACL) then looks like a one-off and floods one misleading WARNING per wait while the fast path silently never works.
Catch redis.exceptions.RedisError (or TimeoutError+ConnectionError) instead of bare Exception, and treat an expected no-token TimeoutError as normal (do the backstop get_result and keep looping) rather than logging "BLPOP failed" and abandoning the fast path.
| _redis_client_unavailable = False | ||
|
|
||
|
|
||
| def _get_result_redis_client() -> Any: |
There was a problem hiding this comment.
[Low] Type the Redis client as redis_lib.Redis | None instead of Any.
Both _redis_client_singleton (L172) and this accessor's return are Any, yet the docstring says this mirrors redis_barrier._get_redis_client, which types the identical singleton concretely (redis_barrier.py:200,203), and create_redis_client already returns redis.Redis. Any is contagious and silences checking on every client.pipeline()/rpush/expire/blpop call (e.g. a blpop(key) vs blpop([key]) typo would be invisible). | None is also more honest — None (build-failed) is a real, branched-on state. Pure annotation change; redis_barrier already imports redis as redis_lib in this package.
| failure on a fresh connection is a genuine error, not a stale reap. If a | ||
| long-lived backend ever gains a one-shot ``get_result`` lookup, revisit | ||
| this. | ||
| relies on: the connection is never idle long enough to be reaped when |
There was a problem hiding this comment.
[Low] Consider a PgResultRow TypedDict for the {status, result, error} row.
The non-obvious invariant this docstring explains in prose — a present row may still have result is None (a failed outcome or a forget-ed tombstone), disambiguated by status — is exactly what a type should carry. A TypedDict(status: str, result: dict|None, error: str) keeps the wire dict (zero runtime cost, the backend reader consumes it as a dict), names the shared return of get_result/wait_for_result/_wait_via_redis, and encodes the result: … | None nullability statically. Keep status: str (not the enum) to match the raw DB string the reader compares against.
| relies on: the connection is never idle long enough to be reaped when | ||
| ``get_result`` runs. In ``poll`` mode the wait polls every ~0.2–2s so the | ||
| connection stays warm; in ``redis`` mode :meth:`_wait_via_redis` ``close``s | ||
| the connection before each (possibly long) ``BLPOP``, so every |
There was a problem hiding this comment.
[Low] Comment accuracy: "every get_result runs on a freshly-opened connection" is overstated.
_wait_via_redis calls close() at the top of the loop, so the post-BLPOP read and the deadline check are fresh — but the immediate race-check get_result (L465) runs before the loop with no preceding close(). It's fresh only because the sole production caller constructs a new backend per wait (executor_rpc.py), not because _wait_via_redis closes before it. Suggest scoping the claim to post-BLPOP reads so the no-reconnect-retry justification stays airtight.
| polling ``pg_task_result`` once per in-flight task scales DB load with | ||
| *concurrency*, not throughput (every waiter SELECTs on a backoff) — which | ||
| saturated the DB at high concurrency. When ``PG_RESULT_SIGNAL_BACKEND=redis``, | ||
| :meth:`store_result` RPUSHes a tiny ready-token (the reply key ONLY — never the |
There was a problem hiding this comment.
[Low] Comment accuracy: "does ONE authoritative SELECT" undercounts (and the opening "polls" line is now stale).
The normal miss-then-wake path does at least two SELECTs per wait — the publish-before-wait race check (L465) and the post-BLPOP read (L499) — plus one fallback SELECT per 30s on a long wait. The O(1)-per-wait-vs-O(N)-polls point is right, but "ONE" is literally inaccurate; suggest "an authoritative SELECT on wake (a small constant per wait, not one-per-poll)".
Relatedly, the module-overview line (top of file, not in this diff) still says "the blocking caller polls wait_for_result until the row appears", which now contradicts the event-driven redis path documented 17 lines below. Worth making mechanism-neutral while here.
…ch cooldown, narrowed redis except, resilience test - Build the signal Redis client with socket_timeout=35s (>30s BLPOP block) so the blocking BLPOP is not cut short and forced to poll. - Replace the permanent unavailable-latch with a 30s cooldown so a transient Redis restart/failover self-heals instead of disabling the signal for life. - Narrow the BLPOP except to redis.exceptions.RedisError; non-redis errors propagate instead of silently degrading to poll. - Type the client singleton as redis.Redis | None. - Tests: missed-signal-still-delivers-via-fallback-SELECT, socket_timeout regression guard, cooldown retry, and non-redis-error propagation. - Docstring accuracy fixes (event-driven wait, SELECT count, get_result conn). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
|
Addressed the review — thanks, pushed P2/High — BLPOP socket-timeout (the important one): the signal client was built with P2/Medium — transient failure permanently disabled signalling: replaced the permanent Medium — Medium — resilience untested: added Low: concrete typing ( Deferred: the |
|
muhammad-ali-e
merged commit 0c2334b
into
feat/UN-3445-pg-queue-integration
1 participant
What
Replace the per-in-flight-task DB poll in the PG-queue result backend with an optional Redis wake-up signal (RPUSH producer / BLPOP consumer), gated by
PG_RESULT_SIGNAL_BACKEND(defaultpoll= unchanged behaviour).Why
Each in-flight file's
wait_for_resultSELECTspg_task_resultevery ≤2s. Load-testing the API-deployment path at 400 concurrent users pinned the txn DB CPU at 100% — DB load scaled with concurrency, not throughput (~190 SELECTs/s of pure polling). Moving the "result is ready" edge off the DB makes DB load scale with throughput.PG
LISTEN/NOTIFYwas evaluated and rejected: the receiver side can't survive transaction-pooled PgBouncer. Redis is already a hard dependency of the stack; only the wake-up signal moves — the queue and the durable result stay in Postgres.How
store_resultRPUSHes a tiny ready-token (reply key only, never the payload/PII) after the row commits;wait_for_resultBLPOPs it, then does ONE authoritative SELECT.PG_RESULT_SIGNAL_BACKEND=redis|poll, defaultpoll) — ships dark, flips per environment; producer/consumer roll independently.docker/sample.envdocuments the flag.Testing
workers/tests/test_pg_result_signal.py— 19 unit tests (flag routing, publish-before-wait race, Redis-down → poll fallback, poll-mode parity, signal-never-raises, no-PII-in-signal) + 1 real round-trip (live Redis RPUSH/BLPOP + real Postgres cross-connection). All green; existingtest_pg_result_backend.py(poll path) still green.Ticket: UN-3712
🤖 Generated with Claude Code