feat(broker): add active_only to flush and replay archived PGMQ messages - #407
Open
mducros-wm wants to merge 5 commits into
Open
feat(broker): add active_only to flush and replay archived PGMQ messages#407mducros-wm wants to merge 5 commits into
mducros-wm wants to merge 5 commits into
Conversation
flush() and flush_all() take an active_only keyword argument, defaulting to False: the messages a broker kept aside once they left the queue are now dropped along with the queue. PostgresBroker empties pgmq.a_<queue> (its rows only, the partitions are left in place for pg_partman to keep managing), RabbitmqBroker gates its XQ purge on the flag and so behaves as before by default, StubBroker and LocalBroker have nothing to spare. Add PostgresBroker.replay_archived_messages(), which sends archived messages back onto their queue and drops them from the archive, filtering on the remoulade message_id, the actor name and an archived_at range, with a dry_run mode. PGMQ can write to the archive but exposes no API to read, count or delete from it, so this goes through SQLAlchemy Core against a declared Table rather than interpolated SQL.
…hive Replace the active_only boolean added in the previous commit with a FlushTarget literal -- "active-only", "dead-only" or "all", defaulting to "all" -- so the messages a broker kept aside once they left the queue can be dropped on their own. A boolean could only express two of the three cases, and dropping just the archive is the case that comes up in practice: a flooded queue is often already drained by its workers, leaving only an oversized archive behind. Each broker maps the target onto its own storage: PostgresBroker onto pgmq.q_<queue> and pgmq.a_<queue>, RabbitmqBroker onto the queue plus its delayed queue and onto the dead letter queue -- the delayed queue holds messages still waiting to be processed, so it counts as active. StubBroker has no dead storage and LocalBroker is a no-op. An unknown target raises rather than being treated as "all", which would drop more than asked. Add PostgresBroker.count_archived_messages, since PGMQ exposes no way to count the archive, and make count_enqueued_messages public alongside it so callers do not have to reach into the untyped pgmq client to count either side.
Review fixes on top of the flush/replay work of this branch. flush went from purge(conn=self._current_connection) to an unconditional engine.begin(), so inside a tx() the purge ran on a second connection. Worse than losing atomicity: pgmq.purge_queue TRUNCATEs, which waits for an ACCESS EXCLUSIVE lock the calling transaction already holds through its uncommitted send -- a deadlock that hangs the process. A _connection() helper now joins the open transaction when there is one and opens its own otherwise, and publishes it on the thread-local state so nested broker calls join it too. replay_archived_messages and count_archived_messages had the same defect. replay_archived_messages documented a "at least one filter is required" guard that was never implemented, and the test named ..._rejects_an_unfiltered_call asserted the opposite of its name. Unfiltered replay is kept and documented; the test is renamed accordingly. Empty collections no longer degrade into replaying the whole archive: `if message_ids:` could not tell "no filter given" from "a filter that matches nothing", so a caller whose log query returned no ids would replay the entire retention window. The SELECT-then-DELETE is now a single DELETE ... RETURNING. The delete locks the rows it selects, so two overlapping replays cannot both pick up the same message and enqueue it twice, and the archive is scanned once under the archived_at predicates that let Postgres prune partitions -- the previous DELETE filtered on the unindexed msg_id alone. Replayed messages go back through enqueue_many instead of send_batch, so the enqueue middleware runs: MessageState resets them to Pending rather than leaving the dashboard reporting the previous outcome, and tracing and metrics see them. Payloads are consequently no longer re-enqueued verbatim, which the docstring now says.
mducros-wm
force-pushed
the
feat/flush-active-only-and-requeue-archived
branch
from
August 28, 2026 15:51
fc08f53 to
32d1eee
Compare
mducros-wm
force-pushed
the
feat/flush-active-only-and-requeue-archived
branch
from
August 31, 2026 13:15
972a125 to
98dbba4
Compare
flush, flush_all and replay_archived_messages take a status filter, read from the status header the pgmq state backend writes on the message itself: flush the successes to reclaim the archive, replay the failures to run them again. Filtering raises NoStateBackend unless that backend is attached. Nothing else writes the header, so a filtered call would otherwise match no row and report having done nothing, which reads exactly like there being nothing to do. The archive table declaration moves to helpers.postgres_client as get_pgmq_table, which now serves the queue table too: a status filter costs the queue its purge, since the rows have to be picked one by one.
compose/compose.yaml ran a plain postgres:16 on 5432 with no pgmq extension, so every postgres test skipped. It now mirrors the services: block of the test workflow: the pgmq image, the ports the test defaults expect, and the extension script tests/docker-compose.yml already mounts.
mducros-wm
force-pushed
the
feat/flush-active-only-and-requeue-archived
branch
from
August 31, 2026 13:51
98dbba4 to
83795f8
Compare
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
1 participant
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.
flush() and flush_all() take an active_only keyword argument, defaulting to False: the messages a broker kept aside once they left the queue are now dropped along with the queue. PostgresBroker empties pgmq.a_ (its rows only, the partitions are left in place for pg_partman to keep managing), RabbitmqBroker gates its XQ purge on the flag and so behaves as before by default, StubBroker and LocalBroker have nothing to spare.
Add PostgresBroker.replay_archived_messages(), which sends archived messages back onto their queue and drops them from the archive, filtering on the remoulade message_id, the actor name and an archived_at range, with a dry_run mode. PGMQ can write to the archive but exposes no API to read, count or delete from it, so this goes through SQLAlchemy Core against a declared Table rather than interpolated SQL.