feat(clients/python): experimental cooperative consumers

[NikolayS]

Summary

• Add subscribe_subconsumer, unsubscribe_subconsumer, receive_coop, and touch_subconsumer methods on PgqueClient. Each maps 1:1 to its pgque.* SQL counterpart and reuses the existing _wrap_sql_error path.
• Extend the high-level Consumer with subconsumer and dead_interval parameters. When subconsumer is set, the poll loop calls client.receive_coop(...) instead of client.receive(...). dead_interval only applies in coop mode (raises ValueError otherwise). No automatic heartbeat — client.touch_subconsumer is the manual primitive.
• Document the API as Experimental in PgQue 0.2 in the Python client README (caveat block + 2-worker example) and add a parity-matrix row in clients/README.md flagging it experimental.

The cooperative-consumers SQL surface itself is already on main; this PR is purely the Python-client layer over it.

Test plan

• PGQUE_TEST_DSN=postgres://nik@localhost/pgque_coop_py pytest clients/python/tests — 67 passed (9 new in test_coop.py):

  • subscribe_subconsumer returns 1 then 0
  • receive_coop returns messages from a published batch and ack finishes them
  • Two subconsumers under one consumer split batches, no duplicate delivery
  • unsubscribe_subconsumer(..., batch_handling=0) raises on active batch
  • unsubscribe_subconsumer(..., batch_handling=1) routes through retry/DLQ
  • touch_subconsumer returns 1 on a registered row
  • Consumer(subconsumer="worker-1") dispatches a handler and acks
  • Consumer without subconsumer is unchanged (regression)
  • Consumer(dead_interval=..., subconsumer=None) raises ValueError

• Manual two-worker e2e (/tmp/coop_e2e.py): N=40 events across 4 ticks, both Consumer instances configured with subconsumer="worker-1" / worker-2. Output:

  sent N         = 40
  worker-1 saw   = 30 msgs
  worker-2 saw   = 10 msgs
  sum of unique  = 40
  overlap        = 0 (must be 0)
  OK -- disjoint subsets, no duplicate delivery.
  

  Disjoint subsets, sums to N, no duplicate delivery — confirms cooperative allocation works end-to-end through the high-level Consumer.

🤖 Generated with Claude Code

[NikolayS]

REV Code Review Report

• PR: NikolayS/PgQue#216 - feat(clients/python): experimental cooperative consumers
• Author: @NikolayS
• AI-Assisted: Yes (Claude Opus 4.7 co-authored bench/demo commits)
• CI: ✅ success (view run) — Python client tests, TypeScript, Go, pg_cron, pg_tle, verify all passed

---

POTENTIAL ISSUES (4)

Issues with moderate confidence (4-7/10). Review manually — may be false positives.

MEDIUM clients/python/tests/test_coop.py:183 — Test name promises cooperative batch splitting but assertion is vacuously true (confidence: 7/10)

test_two_subconsumers_split_batches_no_duplicates publishes 6 messages in one tick (one PgQ batch). Cooperative allocation assigns a single batch to exactly one subconsumer, so m2 will be empty. ids1.isdisjoint(ids2) is vacuously true for an empty set, and (len(m1) + len(m2)) >= 1 is satisfied even if only one worker ever sees messages. The test validates no-duplication but not the splitting property its name promises.
Suggestion: Use multiple ticks (e.g. via the _publish helper pattern from coop_demo.py) so each subconsumer gets at least one batch, then assert len(m1) > 0 and len(m2) > 0 in addition to the disjointness check.

MEDIUM clients/python/pgque/client.py:320 — receive_coop auto-registration path untested (confidence: 6/10)

The docstring prominently states: "The function auto-registers the coop_main and coop_member rows on first call, so callers do not need to subscribe_subconsumer ahead of time." Every test in test_coop.py calls subscribe_subconsumer explicitly first. If auto-registration ever regresses, no test catches it.
Suggestion: Add a test calling receive_coop directly without a prior subscribe_subconsumer to verify the auto-registration path works end-to-end.

MEDIUM clients/python/tests/test_coop.py — dead_interval (stale-sibling takeover) path has no test (confidence: 6/10)

receive_coop accepts dead_interval and the docstring explains it allows takeover of a stale sibling's batch under a fresh batch_id. This distinct code path is never triggered in the test suite, though the README example shows dead_interval="5 minutes".
Suggestion: Add a test that registers two subconsumers, has one receive and hold a batch past dead_interval, then verifies the other can take over.

LOW clients/python/pgque/client.py:338 — max_messages constraint not surfaced in README (confidence: 5/10)

The receive_coop docstring warns: "set this >= the queue's worst-case batch size or consume the full batch before acking" — because ack(batch_id) advances the cursor past the entire PgQ batch including rows beyond max_messages. The README coop section does not mention this constraint.
Suggestion: Add a short note in the README experimental section pointing readers to the max_messages caveat.

---

Summary

Area	Findings	Potential	Filtered
CI/Actions	0	0	0
Security	0	0	0
Bugs	0	0	0
Tests	0	3	1
Guidelines	0	0	1
Docs	0	1	0
Metadata	0	0	0

• Filtered (tests): touch_subconsumer returning 0 for non-existent row untested — confidence 3/10, too minor.
• Filtered (guidelines): batch_handling: int vs Literal[0, 1] — style preference, not a rule violation.

---

REV-assisted review (AI analysis by postgres-ai/rev)