feat(Broker): add PgmqBroker#399
Merged
Merged
Conversation
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
2 times, most recently
from
May 21, 2026 14:43
2263626 to
0ad1c3a
Compare
fregogui
reviewed
May 22, 2026
fregogui
reviewed
May 22, 2026
fregogui
reviewed
May 22, 2026
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
May 28, 2026 13:59
4c5155e to
39443d2
Compare
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
May 29, 2026 13:58
eb5c39e to
1940b4b
Compare
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
May 29, 2026 14:53
1940b4b to
f48ad0b
Compare
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
May 29, 2026 16:01
f48ad0b to
23fb4e1
Compare
Contributor
Author
|
If you want to see on paxone https://gitlab.wiremind.io/wiremind/paxone/paxone-backend/-/merge_requests/4967 |
fregogui
reviewed
Jun 1, 2026
fregogui
requested changes
Jun 1, 2026
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
2 times, most recently
from
June 2, 2026 15:49
f81cf6f to
6983122
Compare
fregogui
reviewed
Jun 4, 2026
fregogui
reviewed
Jun 4, 2026
fregogui
reviewed
Jun 4, 2026
fregogui
reviewed
Jun 15, 2026
fregogui
reviewed
Jun 15, 2026
|
PGMQ maintainer here, is there anything I can help you with? Also, it is good if you see our documentation. |
Reconnect the shared listener with capped backoff instead of degrading to polling permanently after the first connection failure. On any connection error the listener drops the connection, wakes consumers so they fall back to polling immediately, then reopens and re-LISTENs on every registered channel; `available` flips back to True so consumers resume LISTEN/NOTIFY automatically. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Remove all SuperBowl dashboard mentions from the docs (getting_started, README). Also remove API artifacts left unused after the legacy state backend removal and unused by the SuperBowl frontend: - DeleteSchema, orphaned by the earlier clean_states route removal - update_job (PUT /scheduled/jobs/<hash>), never called by SuperBowl - GroupMessagesT, a defined-but-unused TypedDict Document the removed DELETE /messages/states and PUT /scheduled/jobs/<hash> routes in the 7.0.0 changelog breaking changes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Complete the 7.0.0 breaking changes with the Encoder and Message method renames (encode/decode -> encode_in_bytes/decode_bytes, plus the new JSON encoder hooks), alongside the already-listed state backend and removed HTTP routes. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The postgres extra now pulls the PGMQ broker dependencies instead of the removed legacy state backend dependencies. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…gine SQLAlchemy defaults `postgresql://` to the psycopg2 dialect, which the project does not depend on. CI installs only psycopg v3, so the raw engine built in the postgres_broker fixture failed at setup with ModuleNotFoundError: No module named 'psycopg2'. Swap the scheme to postgresql+psycopg://, mirroring what PGMQ does internally for the broker. The plain url is kept for PostgresBroker, whose listener opens a raw psycopg.connect() connection that does not understand +psycopg. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Revert the unjustified ^^^ -> --- underline change on the max_retries retry option so it matches the other retry options (min_backoff, max_backoff, retry_when, backoff_strategy). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Explain that queues are partitioned PGMQ queues backed by pg_partman, what archive_partition_interval_in_days / archive_retention_interval_in_days control, and that pg_partman maintenance must run periodically. Also fix the broken sentence about the required PostgreSQL user and remove the unused FUTURE_PARTITION_HORIZON constant. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
archive() runs in process_message's finally, outside the worker loop's except Empty, so a transient DB error during ack/nack used to propagate and kill the worker thread (which has no restart). Swallow and log it like the RabbitMQ consumer does; the message is redelivered after its visibility timeout, honoring at-least-once. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…cripts encode_in_bytes routed through encode_in_json, whose throwaway json.dumps validation pass ran on top of the real serialization: 2 passes for the default JSONEncoder (RabbitMQ/Stub) and 3 for PydanticEncoder. Serialize directly from the data / _encode_in_json output instead, bringing the default path back to a single pass and Pydantic down to two. Output bytes are unchanged. Also drop the misleading `# pragma: no cover` on the default codec, which the Stub broker actually exercises. Remove local_postgres_broker.py / local_postgres_consumer.py: local dev scripts with hard-coded DSNs and prints that should never have been committed. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The existing suite tests the broker's mechanics largely in isolation; the middleware-driven behaviours were only exercised against the stub and RabbitMQ brokers. Add real-Worker tests running through PGMQ: - a flaky actor that fails then succeeds on retry, - an always-failing actor whose exhausted retries end archived (proving the nack does not leave the message to be redelivered forever — no DLQ), - result storage/retrieval via a Results middleware, - a group aggregating results. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…table
join() counted messages by hand-building a Table("q_<name>", schema="pgmq")
and running count(*), coupling the broker to PGMQ's internal table layout.
Use the public metrics() API instead: queue_length covers visible, invisible
in-flight and native delayed messages alike, matching join()'s contract.
Prune the now-unused SQLAlchemy imports.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…e timeout The polling fallback rounded timeout up to a one-second minimum, so a consumer created with timeout=0 blocked ~1s per read instead of being non-blocking as documented. Do a single immediate read when timeout is 0, and validate in the consumer's __init__ that timeout is not negative (matching the broker's other input validation) instead of silently pretending to coerce it. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
The fallback_encoder swallowed every decode error and returned the raw, un-rehydrated payload through the fallback encoder, silently handing actors the wrong data shape and hiding the original failure. Remove the argument so decoding raises explicitly (ActorNotFound, ValidationError, ...). Drop the two tests that asserted the silent-fallback behaviour; the raising paths are already covered. Document both this and the simplejson->Pydantic serialization change (Decimal now encoded as a string) as breaking. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
- Fix copy-pasted Message.decode_json/encode_in_json docstrings that said "bytestring" while they handle JSON objects. - Make convert_days_in_partman_syntax a staticmethod; it never used self. - Reject prefetch < 1 in the consumer with a ValueError (matching the timeout validation) instead of claiming a coercion that never happened. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
…ener The shared LISTEN/NOTIFY listener keyed a single wake event per queue, so a second consumer on the same queue overwrote the first's event and a closing consumer's unregister() silently stopped notifications for a still-active sibling. Track a set of events per queue, wake them all, and only drop a queue's channel routing once its last consumer leaves. Also back `available` with a threading.Event instead of a bare bool so the dispatch thread's writes and the consumer threads' reads are synchronized rather than relying on CPython GIL atomicity; expose it as a read-only bool property so callers and tests are unaffected. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Reintroduce the opt-in fallback_encoder argument so decoding can fall back to another encoder instead of raising. Defaults to None, keeping strict decoding as the default behaviour. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
July 9, 2026 07:12
de7b588 to
6708fbf
Compare
mducros-wm
force-pushed
the
codex-pgmq-broker-j1
branch
from
July 9, 2026 07:29
6708fbf to
99fa686
Compare
* feat(postgres): batch group enqueue with a single PGMQ send_batch Enqueuing a group of N messages on the PostgreSQL/PGMQ broker previously performed N separate INSERTs (and, without group_transaction, N commits). Add a batch-enqueue path so a group fans out in a single send_batch per queue. - Broker.enqueue_many(): runs the enqueue middleware per message (before_enqueue grouped before the write, after_enqueue after it) and delegates the write to _enqueue_many. On failure, after_enqueue(exception) is emitted only for the messages whose before_enqueue already ran, so no span/state leaks. - Broker._enqueue_many(): default loops _enqueue, so rabbitmq/local/stub are unchanged. - PostgresBroker._enqueue_many(): one client.send_batch per queue, reusing the current transaction connection (1 INSERT, 1 connection, 1 commit with group_transaction). - group.run() now uses broker.enqueue_many(self.build()). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> * feat(postgres): make group enqueue send_batch size configurable Add an optional `enqueue_batch_size` to PostgresBroker. When set, _enqueue_many splits each queue's messages into chunks of that size and sends one send_batch per chunk (all within the same transaction), so a very large fan-out does not build a single oversized INSERT. None (default) keeps the current behavior of a single send_batch per queue. Values <= 0 are rejected. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
PGMQ has no server-side equivalent of RabbitMQ's basic_qos, so the consumer kept reading regardless of how many messages were unacked: with any backlog it drained the queue into worker memory, and the heartbeat renewed an ever-growing set of ids every 10s. The resulting giant set_vt updates bloated the queue table, serialized acks behind row locks and collapsed worker throughput until the process restarted. Enforce prefetch as the in-flight cap in the consumer: only read when unacked count is below prefetch, only request the free slots, and wait for an ack to free a slot instead of busy-looping when at capacity. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> fix(postgres-broker): add msg_id index on queue tables Time-partitioned PGMQ queue tables carry no index on msg_id, so every archive (ack/nack) and set_vt (heartbeat, requeue) seq scans all partitions and throughput degrades with backlog size. declare_queue now ensures a btree index on the partitioned parent, idempotently and also for pre-existing queues, so it propagates to existing and future partitions. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
4 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
No description provided.