feat(state): add PostgresBackend, a state backend stored in the pgmq … - #408
Open
mducros-wm wants to merge 17 commits into
Open
feat(state): add PostgresBackend, a state backend stored in the pgmq …#408mducros-wm wants to merge 17 commits into
mducros-wm wants to merge 17 commits into
Conversation
…message The backend keeps no store of its own: a message's state *is* its pgmq row. Pending/Started, the three timestamps and the attempt count are derived from PGMQ's enqueued_at, last_read_at, archived_at and read_ct, and the terminal status is merged into the archive the broker already writes on ack/nack, so tracking a message's whole lifecycle costs no additional statement. To make that merge possible, StateBackend.set_state takes an optional `message` keyword argument, which the state middleware now always passes. A state backend defined outside remoulade must accept it. Unlike the Redis and stub backends, get_states/get_states_count honour their filters, sorting and pagination in SQL. set_progress is unsupported and raises, since storing a progress would mean an UPDATE on the queue table per call. The broker's hand-written SQL moves into RemouladePostgresClient (remoulade.helpers.postgres_client), which also gains the indexes the state lookups need: (message->>'message_id') on queue and archive tables, and (headers->>'status') on archive tables. Index creation happens with the queue, so existing queues need a one-off backfill -- see the changelog.
mducros-wm
commented
Aug 20, 2026
mducros-wm
commented
Aug 20, 2026
mducros-wm
commented
Aug 20, 2026
PostgresBroker's hand-written statements interpolate the queue name into a SQL identifier, which no bind parameter can carry. Check it at declare_queue, the single gate every queue name goes through, so a name that cannot be used as an identifier fails loudly when actors are declared. Reuse the character set remoulade already enforces on every actor declaration (actor.QUEUE_NAME_PATTERN, promoted from _queue_name_re) rather than defining a second, divergent rule. Only the length bound is specific to PostgreSQL: past 47 characters the longest index name remoulade derives (q_<queue>_msg_id_idx) exceeds PostgreSQL's 63-byte cap, gets truncated, and two long queue names collide on one index.
… on every declare Two fixes from the review of this branch. PostgresBackend.set_state passed no connection to patch_headers, so the UPDATE took a second connection from the pool instead of the broker's open transaction. Inside broker.tx() -- what group.run() uses with group_transaction=True -- it could not see the rows that transaction had just written, so the status was dropped and patch_headers reported False; and a thread already holding a connection asked the pool for another, blocking until the pool timed out. Route it through the broker's transaction, which is None outside one and lets the client open a transaction for the single statement, as before. declare_queue no longer leaves an existing queue's indexes to a manual create_indexes call. Nothing warns when one is missing and nothing fails -- the queue merely seq scans every partition on each archive and set_vt -- so the declaration is what has to repair it, rather than an operator knowing to. It now ensures them on every declaration, new queue or not, which makes the create_partitioned_queue override a passthrough; drop it.
PostgresBackend.set_state patched headers on whatever queue the state named. A Message built by hand -- Message(queue_name=...) or Message.copy -- skips the validation the actor API does, so that name is plain application data, and patch_headers interpolates it into the UPDATE as a SQL identifier. The reachable path is an enqueue that fails: Broker.enqueue emits after_enqueue(exception=...) for the message, MessageState saves Failure, and the state carries the undeclared queue name. emit_after logs and swallows, so what it produced was a wasted UPDATE on the throughput-critical queue table and a misleading Postgres traceback logged at critical, hiding the QueueNotFound the caller actually got. Statement execution never succeeded -- the tail of the statement spans lines a `--` cannot reach, and psycopg's extended protocol needs both bind params satisfied -- but the name has no business reaching SQL. Check it against broker.queues, the queues the broker declared, the same guard the broker already applies on enqueue and consume. A queue that is not there holds no message to patch, so there is nothing to look in either way.
…y client pgmq sets SQLAlchemyPGMQueue to None rather than failing when its SQLAlchemy backend cannot be imported, so RemouladePostgresClient subclasses None and the module raises TypeError, not ImportError. That escaped the guard, failing `import remoulade.state.backends` outright and taking RedisBackend and StubBackend down with it in an environment that never asked for Postgres.
set_state raised NotImplementedError on any state carrying a progress. Since Message.set_progress goes through set_state, an actor reporting its progress in a loop failed mid-work and retried forever -- losing far more than the progress it could not store. Ignore the progress instead: a state that also carries a terminal status now gets that status recorded rather than losing everything to the exception. Nothing else is needed, a progress-only state already falls through the TERMINAL_STATUSES check just below. Public API untouched. Also two docs that no longer described the code: stage_headers promised that "a progress update followed by a terminal status both land", unreachable by design, and a comment still named _PostgresMessage, renamed PostgresMessage on this branch.
PostgresBackend had two ways to record a status: staged on the in-flight message and written by the archive the broker performs anyway, or, with no message at hand, an UPDATE on the queue table matched on (message->>'message_id'). That second path has no legitimate caller. MessageState always passes the message; the processing hooks pass a PostgresMessage and take the staged route, and the only hook passing a plain Message is after_enqueue, which records a status just when the enqueue raised -- and an enqueue that raised wrote no row, so there is nothing to patch. What is left is a direct set_state call, where a message id is not enough to name a row: a retry is the same message re-enqueued, so an id can match several rows at once and nothing in such a call says which one the status belongs to. So drop patch_headers and raise NotImplementedError when no message is given. This also drops, as dead code, the transaction routing added in 6aa8d69 and the broker.queues check added in e9eb62c -- both of which guarded that path, and both of which are what made it clear the path had no caller worth keeping. And it drops the (message->>'message_id') index, whose only reader was patch_headers: one less index to maintain on every insert into the throughput-critical table.
fregogui
reviewed
Aug 21, 2026
fregogui
requested changes
Aug 21, 2026
PostgresBackend recorded a status without ever running a statement: it staged a
jsonb patch on the in-flight message, and the broker folded it into the archive
it performs on ack/nack anyway. That coupling is not worth what it saves.
The status is now an UPDATE on pgmq.q_<queue>, run at set_state time -- one
statement per processed message, on top of the ack. The hook order allows it:
emit_after("process_message") runs before post_process_message, so the row is
still enqueued when the terminal status is written, and pgmq.archive carries the
headers over. A status now also survives an archive that failed and got
swallowed, where a staged patch was lost with it.
The row is named by State.delivery_id, a new field the state middleware fills
from the in-flight message (MessageProxy.delivery_id, None on the base class and
the PGMQ msg_id on PostgresMessage). Carrying the id in the State rather than as
a set_state kwarg gives StateBackend.set_state its signature back, so this branch
no longer breaks a state backend defined out of tree. A state without a
delivery_id records nothing, which is the failed-enqueue case: MessageState
reports a Failure, but an enqueue that raised wrote no row to record it on -- so
the NotImplementedError 6396c1f introduced for that goes away with it.
Targeting stays on msg_id rather than (message->>'message_id'): a retry keeps the
message_id, so two rows of a queue can share it, and the index such a statement
would need was dropped in 6396c1f.
The broker loses PostgresMessage.stage_headers, its _header_patch and the archive
override that merged them, so it knows nothing of the state backend any more and
_archive_message is a plain pgmq.archive again.
The Unreleased section had grown a bullet per commit, in Upgrading/Feat/Changed, several of which described the same change from different angles. Release it as 7.1.0 with one bullet per change a reader upgrading from 7.0.0 can act on. Folded the msg_id index backfill and its table-lock warning into a single Fix bullet, moved the queue name validation to Breaking changes, where it belongs since a name over 47 characters used to be accepted, and dropped the move of the broker's SQL into RemouladePostgresClient: internal, with nothing for a reader to do about it.
It is the only method added on this branch that implements a base class one without saying so. Everything else either overrides nothing (patch_headers, create_indexes, _connection, _run, PostgresBackend.client, _build_message) or is already decorated (__next__, ack, nack, requeue, close, delivery_id).
mducros-wm
force-pushed
the
feat/pgmq-state-backend
branch
from
August 24, 2026 09:34
5b4a9bf to
d187cba
Compare
fregogui
approved these changes
Aug 26, 2026
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
2 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.
…message
The backend keeps no store of its own: a message's state is its pgmq row. Pending/Started, the three timestamps and the attempt count are derived from PGMQ's enqueued_at, last_read_at, archived_at and read_ct, and the terminal status is merged into the archive the broker already writes on ack/nack, so tracking a message's whole lifecycle costs no additional statement.
To make that merge possible, StateBackend.set_state takes an optional
messagekeyword argument, which the state middleware now always passes. A state backend defined outside remoulade must accept it.Unlike the Redis and stub backends, get_states/get_states_count honour their filters, sorting and pagination in SQL. set_progress is unsupported and raises, since storing a progress would mean an UPDATE on the queue table per call.
The broker's hand-written SQL moves into RemouladePostgresClient (remoulade.helpers.postgres_client), which also gains the indexes the state lookups need: (message->>'message_id') on queue and archive tables, and (headers->>'status') on archive tables. Index creation happens with the queue, so existing queues need a one-off backfill -- see the changelog.