fix(ruby): make Consumer resilient

[NikolayS]

What changed

• immediately re-poll after a successfully acked non-empty batch, so pre-existing backlog drains without one poll_interval delay per batch;
• reconnect and re-LISTEN after transient PG::Error / Pgque::Error failures;
• close failed sessions and use a stop-aware bounded retry wait;
• preserve backoff for empty or intentionally unfinished batches;
• document the behavior and replace the obsolete exit-on-connect-failure test;
• add database-backed resilience coverage.

Why

The new Ruby client could not catch up efficiently after downtime because notifications for existing batches had already fired before it connected. It also exited permanently on routine connection, receive, or ack failures, contrary to the blocking Consumer contract and the behavior of the other first-party clients.

Fixes #313.

Regression coverage

The new five-test suite was run against the old implementation first and failed 5/5:

• backlog stalled at 1/3;
• no recovery from initial connect failure;
• no retry after transient receive failure;
• no reconnect after backend termination;
• Consumer exited instead of waiting for reconnect.

After the fix:

• resilience suite: 5 runs, 22 assertions, 0 failures (repeated under three seeds);
• full Ruby/PostgreSQL suite: 81 runs, 216 assertions, 0 failures/errors;
• gem build + isolated install + load smoke: 0.3.0.rc.1 Pgque::Consumer;
• ruby -c and git diff --check: clean.

Tested with Ruby 3.4.8 and PostgreSQL 18.3. This PR is intentionally a draft and has not been merged.

[NikolayS]

Real-user verification evidence

The original RED was reproduced retrospectively because the implementation and regression tests were committed together. With the final PR test file and the pre-fix library at 73af5c6:

PGQUE_TEST_DSN=postgresql:///pgque_pr_rules_audit \
  ruby -I/tmp/pgque-030-audit/clients/ruby/lib -Iclients/ruby/test \
  clients/ruby/test/test_consumer_resilience.rb

Result: 5 runs, 8 assertions, 5 failures. The failures covered backlog draining, initial connection recovery, receive retry, backend termination, and stop-aware reconnect waiting.

Head verification:

PGQUE_TEST_DSN=postgresql:///pgque_pr_rules_audit ruby -S rake test
gem build pgque.gemspec --output /tmp/pgque-318-0.3.0.rc.1.gem
gem install /tmp/pgque-318-0.3.0.rc.1.gem --install-dir "$(mktemp -d)" --no-document

Result after the follow-up poll-result coverage: 83 runs, 218 assertions, 0 failures, 0 errors, 0 skips. The two explicit poll_once result tests failed on the old implementation with nil and pass on head with false for an empty queue and true for a processed batch. The isolated gem loaded Pgque::Consumer successfully. Head CI: 17/17 checks passed.

[NikolayS]

REV rubric review

Automated ultrareview quota was unavailable, so I executed the five official
REV rubric prompts manually. SOC2 findings were omitted as instructed. This is
a comment-only review, not an approval.

Blocking findings

None. I found no unresolved functional, security, documentation, or test
blocker on head 38d9af41503cd9cd9cf77dd1baf13889448d451c.

Rubric results

• Bug hunter: backlog draining, reconnect/re-LISTEN, failed-session cleanup,
  stop-aware retry, and unfinished-batch backoff are internally consistent.
• Security reviewer: no credential exposure, injection path, privilege change,
  or unsafe error recovery was introduced.
• Docs reviewer: the README accurately explains notification loss, immediate
  backlog polling, retry visibility, and permanent-error behavior.
• Guidelines checker: the diff is surgical and current CI is green. One
  immutable historical constraint remains: test(ruby): cover poll result contract uses a commit type not allowed by current CLAUDE.md. Per policy,
  do not amend or force-push it.
• Test analyzer: the database-backed resilience suite covers initial connect,
  transient receive, backend termination, prompt stop, backlog draining, and
  the poll return contract. The recorded base failures and head successes are
  adequate RED/GREEN evidence.

Nonblocking findings

Only the immutable commit-type history noted above. No current code change is
requested by this review.

[NikolayS]

samorev Code Review Report

• PR: fix(ruby): make Consumer resilient #318 - fix(ruby): make Consumer resilient
• Author: @NikolayS
• AI-Assisted: Yes

Pipeline	Coverage
unknown	Not reported

---

REVIEW FINDINGS (14)

HIGH MR/PR state - Review target is draft

The review target is still marked as draft.
Fix: Mark it ready for review before merge.

CRITICAL CI/Pipeline - Pipeline status is unknown

Provider CI reported status unknown.
Fix: Fix failing checks and rerun review.

HIGH [bugs] Permanent/configuration errors now retry forever, and with the documented default logger ($stderr at FATAL) every retry is completely silent. A consumer pointed at a nonexistent queue, a wrong DSN, or a database it cannot authenticate to will spin indefinitely with zero output and a live process, instead of exiting. Supervisors (systemd, k8s CrashLoopBackOff, exit codes) lose their only misconfiguration signal, and the removed test_running_clears_after_start_failure was the test that pinned the old fail-fast contract.

rescue PG::Error, Pgque::Error => e ... @logger.error("consumer #{@name}: database error; retrying in #{@poll_interval}s: ...") combined with README context the default targets $stderr at FATAL, so the consumer is effectively silent and the new README line Permanent configuration errors such as a missing queue or consumer also retry until stop, SIGTERM, or SIGINT.
Fix: Log retry events at FATAL (or raise the default level to error) so they are visible with the shipped default. Additionally, distinguish non-retryable errors — undefined queue/consumer, authentication failure, PG::UndefinedTable, invalid DSN — and either re-raise them or expose a max_retries / on_error hook so operators can choose fail-fast. Retrying PG::ConnectionBad forever is reasonable; retrying ERROR: queue "typo" does not exist forever is not.

HIGH [bugs] poll_interval was the only rate limiter on redelivery, and next if processed removes it. A batch whose messages are all nacked (e.g. unknown event type — the README documents nack-by-default for unhandled types) still finishes the batch, so poll_once returns true and the loop immediately re-polls. If nacked messages become visible again without a delay, the consumer enters an unbounded hot loop: 100% of a core, continuous transactions against Postgres, and a log line per message, with no drain and no backoff. Previously each redelivery cycle cost one poll_interval.

next if processed in run_session, with processed = true set unconditionally at the end of the conn.transaction block in poll_once, and README: By default the consumer **nacks** any message whose type has no [handler].
Fix: Base the immediate re-poll on progress, not on batch finished — return true only when at least one message was successfully acked/handled, or track consecutive immediate polls and fall back to wait_for_notify_or_stop after N (e.g. 100) back-to-back full batches. Verify the nack retry delay: if it can be zero, the immediate re-poll must not apply to all-nacked batches.

MEDIUM [bugs] Reconnect uses a flat @poll_interval with no exponential backoff, no jitter, and no cap. poll_interval is user-tunable and legitimately small (the new tests themselves use 0.1), so a consumer configured for low latency will attempt ~10 full TCP/auth handshakes per second for the entire duration of a database outage. With a fleet of consumers all retrying in lockstep, this is a synchronized reconnect storm against a database that is trying to recover.

def wait_before_reconnect; deadline = monotonic + @poll_interval; ... — the reconnect delay is exactly the poll interval.
Fix: Use a dedicated backoff independent of poll_interval: exponential from ~100ms to a cap (5–30s) with jitter, reset on a successful session. Keep the existing stop-aware slicing.

MEDIUM [bugs] stop is only prompt during the wait, not during the connect. PG.connect is a blocking call with no connect_timeout enforced; against a blackholed IP or an unresponsive host it can block for the OS TCP timeout (often 2+ minutes). During that window stop, SIGTERM, and SIGINT are all ignored, so a shutdown that the README describes as prompt can hang well past any container termination grace period. The new test does not catch this because it stubs PG.connect to raise instantly.

def run_session; conn = PG.connect(@dsn) (no timeout applied) plus the test stub PG.define_singleton_method(:connect) do |*| connect_calls += 1; raise PG::ConnectionBad, "simulated persistent connection failure" end.
Fix: Inject a default connect_timeout into the DSN when the caller has not set one (e.g. [WAIT_SLICE_SECONDS * 2, poll_interval].max), so a blocked connect returns control to the running? check within a bounded time. Consider a test that stubs PG.connect to sleep rather than raise, to actually exercise stop-during-connect.

MEDIUM [tests] The invariant that makes the immediate re-poll safe — "an unfinished batch (for example after a failed nack) returns false and retains the normal backoff" — is asserted only in a code comment, never in a test. The two new poll_once tests cover empty (false) and fully-processed (true) but not the unfinished path. If a future refactor lets processed = true escape on the unfinished path, the result is the unbounded hot loop described above, and the suite stays green.

# An unfinished batch (for example after a failed nack) returns false and retains the normal backoff. — with only test_poll_once_reports_empty_queue and test_poll_once_reports_processed_batch added.
Fix: Add a third test that forces the nack-failed / batch-left-unfinished path (stub Client#nack to fail, or register a handler that triggers it) and asserts poll_once returns false. This is the highest-value missing test in the PR.

MEDIUM [bugs] While the queue stays non-empty, wait_for_notify_or_stop is never reached, so conn.notifies is never called. libpq buffers incoming NOTIFY messages in the connection's notification list until the client pops them; under sustained load with a NOTIFY per tick, that list grows for as long as the consumer keeps up a non-empty backlog. On a busy queue this is unbounded client-side memory growth, and on the next idle transition the consumer will churn through a large stale backlog of notifications.

next if processed skips wait_for_notify_or_stop(conn), which is the only caller that drains notifications.
Fix: Drain pending notifications non-blockingly on every loop iteration (while conn.notifies; end or conn.wait_for_notify(0)) before the next if processed.

LOW [tests] The timing assertion in the backlog test is vacuous — it can never fail independently. wait_until(timeout: 5) already bounds elapsed time to ~5s, so if drained is true then elapsed < 5 is automatically true, and if drained is false the preceding assert drained fires first. The test's stated purpose (proving no poll_interval wait between batches) rests entirely on the 5s wait_until timeout versus poll_interval: 30.

drained = wait_until(timeout: 5) { seen.size == 3 } / elapsed = monotonic - started / assert_operator elapsed, :<, 5, "backlog took #{elapsed.round(2)}s to drain"
Fix: Either drop the redundant assertion or make it meaningful — assert elapsed < poll_interval (30s) with a wait_until timeout comfortably above one poll interval, so the assertion itself is what distinguishes old from new behavior.

LOW [tests] The tests monkey-patch process-global state (PG.connect, Pgque::Client#receive) and restore it only after thread.join(3). If the join times out, the consumer thread is still running when PG.connect is restored, leaving a live thread issuing real connections into subsequent tests. The patches are also global, so these tests cannot be run under parallelize_me! and will corrupt any concurrently running test.

ensure cons.stop; thread.join(3); PG.define_singleton_method(:connect, original_connect) end — join return value is discarded.
Fix: Assert the join succeeded (refute_nil thread.join(3) or assert thread.join(3), "consumer thread did not exit") before restoring, and add a comment or guard that this file must not run in parallel with others.

LOW [tests] The Client#receive stub captures only positional arguments, while the PG.connect stub in the same file correctly handles **kwargs. If receive takes or gains keyword arguments (a limit:/batch_size: is the natural signature here), Ruby 3 turns them into a trailing positional Hash in |*args| and bind_call then passes that Hash positionally, breaking the call in a way that looks like a production bug rather than a stub bug.

Pgque::Client.define_method(:receive) do |*args| ... original_receive.bind_call(self, *args) end versus PG.define_singleton_method(:connect) do |*args, **kwargs| a few tests earlier.
Fix: Use do |*args, **kwargs, &blk| and original_receive.bind_call(self, *args, **kwargs, &blk) for symmetry and forward compatibility.

LOW [tests] SecureRandom is used but not required, unlike logger, stringio, and uri which are explicitly required at the top of the same file. It currently resolves only via a transitive require from test_helper or a dependency; the test breaks if that incidental require ever goes away.

require_relative "test_helper" / require "logger" / require "stringio" / require "uri" ... app_name = "pgque_ruby_#{SecureRandom.hex(6)}"
Fix: Add require "securerandom".

LOW [tests] The reconnect test reads getvalue(0, 0) from an unordered multi-row-capable query. pg_terminate_backend returns as soon as the signal is sent, so the terminated backend's row can briefly coexist with the new one in pg_stat_activity; with no ORDER BY and no exclusion of old_pid in the query, row 0 may be the dying backend. The test then polls until the stale row disappears, which works but makes the 8s budget depend on backend teardown timing rather than on reconnect timing. It will also fail rather than skip if the test role lacks pg_signal_backend.

result.ntuples.positive? && result.getvalue(0, 0).to_i != old_pid against select pid from pg_stat_activity where application_name = $1 and pid <> pg_backend_pid()
Fix: Push the exclusion into SQL — and pid <> $2 with old_pid — so the check is "a new backend exists" rather than "row 0 happens to be new".

LOW [guidelines] The new test file redefines force_tick, silent_logger, and monotonic locally even though it already include PgqueTest::Helpers, and test_consumer.rb uses force_tick/silent_logger too. These helpers now exist in two places and can drift — a fix to force_tick (e.g. handling a ticker race) would silently apply to one suite only.

include PgqueTest::Helpers followed by def force_tick(conn, queue), def silent_logger, def monotonic in test_consumer_resilience.rb.
Fix: Move the shared helpers into PgqueTest::Helpers and delete the local copies.

---

Summary

Area	Findings	Potential	Filtered
CI/Pipeline	1	0	0
Security	0	0	0
Bugs	0	5	0
Tests	0	6	0
Guidelines	0	1	0
Docs	0	0	0
Metadata	1	0	0

Note:

• Findings: High-confidence issues (8-10/10) - blocking or non-blocking per severity
• Potential: Medium-confidence issues (4-7/10) - review manually
• Filtered: Low-confidence issues (0-3/10) - excluded as likely false positives

Review metadata

provider=github
kind=pr
project=NikolayS/PgQue
number=318
target=github:NikolayS/PgQue#318
state=OPEN
draft=true
diff_lines=457
diff_added=356
diff_removed=21
diff_bytes=14882
comments_count=1
commits_count=2
ci_status=unknown
ci_summary=total=18 success=17 failure=0 pending=0 other=1
prompt=.claude/commands/review-mr.md
blocking=false
posted_by=gh
no_comment=false
live_posting=posted

---

samorev-assisted review (AI analysis by Tanya301/samorev)