feat(state): add PostgresBackend, a state backend stored in the pgmq …

docs/source/changelog.rst

@@ -5,6 +5,46 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+Unreleased
+----------
+Upgrading
+^^^^^^^^^
+* **A PGMQ queue missing its ``msg_id`` index gets it on its next declaration.** ``declare_queue`` now ensures
+  the index on an already existing queue and not only on one it creates, so a queue that predates it stops seq
+  scanning every partition on each ack without any manual backfill. On a large existing queue the initial index
+  build locks the table for its duration, so plan the first start after the upgrade like any other index
+  creation.
+
+Feat
+^^^^
+* Add ``PostgresBackend``, a write-only state backend for the PostgreSQL/PGMQ broker that stores a message's
+  terminal status in the message itself instead of a dedicated table. The status is merged into the archive the
+  broker already writes on ack/nack, so tracking a message's outcome costs no additional statement, and
+  ``Pending``/``Started`` write nothing at all since PGMQ's own columns already record them. Reading state back
+  is not implemented: ``get_state``, ``get_states``, ``get_states_count`` and ``clean`` raise
+  ``NotImplementedError``, so the dashboard and the state routes of ``remoulade.api`` cannot be served by this
+  backend — query the PGMQ tables directly. ``Message.set_progress`` is silently dropped: storing a progress
+  would mean an ``UPDATE`` on the broker's queue table per call. And ``set_state`` needs the message it is given
+  a status for: the middleware always passes it, but a direct call without it raises ``NotImplementedError``,
+  since a message id can name several rows of a queue at once — a retry is the same message re-enqueued.
+* ``declare_queue`` ensures the PGMQ broker's indexes on every declaration, whether the queue was just created
+  or already existed, so a queue that predates one of them is repaired rather than left to degrade quietly —
+  see *Upgrading*.
+
+Changed
+^^^^^^^
+* ``StateBackend.set_state`` accepts an optional ``message`` keyword argument, letting a backend persist state
+  through the broker's own writes rather than a statement of its own. Backends that do not need it ignore it,
+  but a state backend defined outside remoulade must accept it: the middleware now always passes it, so an
+  override still declared as ``set_state(self, state, ttl=3600)`` raises ``TypeError``.
+* Move the PGMQ broker's hand-written SQL into ``RemouladePostgresClient`` (``remoulade.helpers.postgres_client``).
+* ``PostgresBroker.declare_queue`` now rejects a queue name it could not use as a SQL identifier, since the
+  statements remoulade writes itself interpolate the name into one. The character set is the one remoulade
+  already enforces on every actor declaration; on top of it, the PGMQ broker caps a name at 47 characters, so
+  that the longest index name it derives stays under PostgreSQL's 63-byte limit instead of being truncated into
+  a collision. A name over that length used to be accepted here and now raises ``ValueError`` when actors are
+  declared.
+
 `7.0.0`_ -- 2026-06-15
 ------------
 Breaking changes

docs/source/guide.rst

@@ -449,6 +449,71 @@ table, which is partitioned the same way. Two broker parameters control this:
    a self-managed or hosted PostgreSQL you must enable it yourself.
 
 
+Postgres State Backend
+^^^^^^^^^^^^^^^^^^^^^^
+
+With the Postgres broker, the status of a message can be tracked without a table
+of its own: it is written into the very PGMQ row that carries the message::
+
+  import remoulade
+
+  from remoulade.brokers.postgres import PostgresBroker
+  from remoulade.state import MessageState
+  from remoulade.state.backends import PostgresBackend
+
+  broker = PostgresBroker(url="postgresql://remoulade@localhost:5432/remoulade")
+  broker.add_middleware(MessageState(PostgresBackend(broker)))
+  remoulade.set_broker(broker)
+
+Only the terminal status (``Success``, ``Failure``, ``Skipped``, ``Canceled``) is
+stored, in the message's ``headers`` column. ``Pending`` and ``Started`` write
+nothing at all: PGMQ already records whether a message has been read
+(``read_ct``), when (``last_read_at``), when it was enqueued (``enqueued_at``),
+when it finished (``archived_at``) and — by moving the row from ``pgmq.q_<queue>``
+to ``pgmq.a_<queue>`` — whether it finished at all. The status is handed to the
+broker, which folds it into the archive it performs on ack or nack anyway, so
+tracking a message's outcome costs **no additional statement**.
+
+This backend is **write-only**. ``get_state``, ``get_states``,
+``get_states_count`` and ``clean`` are not implemented and raise
+``NotImplementedError``, so the dashboard and the state routes of
+``remoulade.api`` cannot be served by it. Read the PGMQ tables directly instead::
+
+  SELECT message->>'message_id', message->>'actor_name', headers->>'status', read_ct, archived_at
+  FROM pgmq.a_default
+  WHERE archived_at > now() - interval '1 day';
+
+Because a retried message keeps its ``message_id`` and is re-enqueued as a new
+row, the archive holds one row per attempt, each with its own status — a
+ready-made audit trail.
+
+``Message.set_progress`` is **silently dropped**. It is called while the actor
+runs and has to be visible before the message finishes, so it cannot ride along
+with the ack: every call would be an ``UPDATE`` on the broker's queue table,
+which is its throughput-critical one. Dropping it rather than raising keeps an
+actor that reports its progress working when it is pointed at this backend —
+raising would fail the message mid-work and retry it forever. Report the
+progress of long-running work through your metrics instead.
+
+Two more differences from the Redis state backend are worth planning for:
+
+* **Retention is the archive's.** ``state_ttl`` no longer bounds how long a
+  status is kept; ``archive_retention_interval_in_days`` does, through
+  ``pg_partman``. A status disappears when its message's archive partition is
+  dropped, and purging or dropping a queue destroys the statuses with it.
+* **A status cannot outlive its message.** The backend stores nothing of its own,
+  so it has nowhere to keep a status for a message that is not there.
+
+.. note::
+
+   ``set_state`` needs the message it is recording a status for, and raises
+   ``NotImplementedError`` without it. The state middleware always passes it, so
+   this only concerns a call of your own. A message id would not be enough on its
+   own: a retry is the same message re-enqueued, so one id can name several rows
+   of a queue at once, and nothing in such a call would say which of them the
+   status belongs to.
+
+
 Local Broker
 ^^^^^^^^^^^^^^^
 

remoulade/actor.py

@@ -29,7 +29,7 @@
 
 
 #: The regular expression that represents valid queue names.
-_queue_name_re = re.compile(r"[a-zA-Z_][a-zA-Z0-9._-]*")
+QUEUE_NAME_PATTERN = re.compile(r"[a-zA-Z_][a-zA-Z0-9._-]*")
 
 P = ParamSpec("P")
 R = TypeVar("R")
@@ -126,7 +126,7 @@ def decorator(fn: Callable[P, R]) -> Actor[P, R]:
         queues_names = [queue_name]
         if alternative_queues is not None:
             queues_names += alternative_queues
-        if any(not _queue_name_re.fullmatch(name) for name in queues_names):
+        if any(not QUEUE_NAME_PATTERN.fullmatch(name) for name in queues_names):
             raise ValueError(
                 "Queue names must start with a letter or an underscore followed "
                 "by any number of letters, digits, dashes or underscores."

remoulade/brokers/postgres.py

@@ -26,13 +26,13 @@
 from urllib.parse import urlparse
 
 import psycopg
-from pgmq import SQLAlchemyPGMQueue
 from pgmq.messages import Message as PostgresQueueMessage
 from psycopg import sql as psycopg_sql
 from sqlalchemy import Connection, text
 
 from ..broker import Broker, Consumer, MessageProxy
 from ..errors import QueueJoinTimeout, QueueNotFound, UnsupportedMessageEncoding
+from ..helpers.postgres_client import RemouladePostgresClient, assert_valid_queue_name
 from ..message import Message
 
 if TYPE_CHECKING:
@@ -129,7 +129,7 @@ def __init__(
         self.enable_listen_notify = enable_listen_notify
         self.enqueue_batch_size = enqueue_batch_size
 
-        self.client = SQLAlchemyPGMQueue(
+        self.client = RemouladePostgresClient(
             conn_string=url,
             init_extension=False,
             vt=self.visibility_timeout_seconds,
@@ -217,11 +217,26 @@ def consume(self, queue_name: str, prefetch: int = 1, timeout: int = 30000) -> "
     def declare_queue(self, queue_name: str) -> None:
         """Create a partitioned PGMQ queue if it does not already exist.
 
-        Also ensures the queue table has a btree index on ``msg_id`` — even
-        for pre-existing queues, so queues created before remoulade added the
-        index pick it up on the next declaration. On a large existing queue
-        the initial index build locks the table for its duration.
+        Whether the queue was just created or already existed, this also ensures the
+        indexes remoulade needs on it. Doing it on every declaration is what repairs
+        a queue created by a version of remoulade that did not yet declare one of
+        them: nothing warns when one is missing and nothing fails, the queue merely
+        seq scans every partition on each ``archive`` and ``set_vt``.
+        ``CREATE INDEX IF NOT EXISTS`` is a catalog lookup once the index is there,
+        but on a large existing queue the initial build locks the table for its
+        duration.
+
+        This is the one gate every queue name goes through before it reaches the
+        broker, so it is where the name is checked against
+        :func:`~remoulade.helpers.postgres_client.assert_valid_queue_name`: the
+        statements remoulade writes itself interpolate it as a SQL identifier, which
+        no bind parameter can carry. Failing here means failing loudly at startup,
+        when actors are declared, rather than on a malformed identifier later.
+
+        Raises:
+          ValueError: If ``queue_name`` cannot be used as a SQL identifier.
         """
+        assert_valid_queue_name(queue_name)
         if queue_name in self.queues:
             return
         with self.tx():
@@ -244,29 +259,13 @@ def declare_queue(self, queue_name: str) -> None:
                 if self.enable_listen_notify:
                     self._try_enable_notify(queue_name)
 
-            self._create_msg_id_index(queue_name, self._current_connection)
+            self.client.create_indexes(queue_name, self._current_connection)
 
         self.queues[queue_name] = None
 
         if not queue_exists:
             self.emit_after("declare_queue", queue_name)
 
-    def _create_msg_id_index(self, queue_name: str, connection: "Connection") -> None:
-        """Ensure the queue table has a btree index on ``msg_id``.
-
-        PGMQ's time-partitioned queue tables ship without one, so every
-        ``archive`` (ack/nack) and ``set_vt`` (heartbeat, requeue) lookup seq
-        scans all partitions. Created on the partitioned parent, the index
-        propagates to existing and future partitions. ``msg_id`` never
-        changes, so the index does not defeat HOT updates of ``vt``/``read_ct``.
-
-        The queue name must already be validated (``validate_queue_name``)
-        since it is interpolated as an identifier.
-        """
-        connection.execute(
-            text(f'CREATE INDEX IF NOT EXISTS "q_{queue_name}_msg_id_idx" ON pgmq."q_{queue_name}" (msg_id)')
-        )
-
     def _encode_message(self, message: "Message") -> PostgresPayload:
         """Encode a Remoulade message into a JSON object payload for PGMQ.
 
@@ -768,17 +767,24 @@ def _requeue_message_ids(self, message_ids: list[int]) -> None:
     def _archive_message(self, message: "MessageProxy") -> None:
         """Stop tracking a message and archive it, tolerating transient failures.
 
+        Any header metadata a middleware staged on the message (the state backend
+        recording its outcome, typically) is handed to the client, which merges it
+        as part of the archive rather than by a statement of its own — so
+        recording that a message succeeded or failed costs nothing on top of the
+        ack.
+
         A failed archive (connection blip, pool exhaustion, ...) is logged and
         swallowed rather than propagated: letting it bubble up would kill the
         worker thread, which has no restart logic. The message simply becomes
         visible again once its visibility timeout expires and is redelivered,
-        which is the broker's at-least-once guarantee.
+        which is the broker's at-least-once guarantee. A staged patch is lost
+        with it, and is rebuilt when the message is processed again.
         """
-        if not isinstance(message, _PostgresMessage):
+        if not isinstance(message, PostgresMessage):
             raise ValueError("It must be a PostgresMessage")
         self._unregister_heartbeat_message_id(message._postgres_message.msg_id)
         try:
-            self.client.archive(self.queue_name, message._postgres_message.msg_id)
+            self.client.archive(self.queue_name, message._postgres_message.msg_id, headers=message._header_patch)
         except Exception:
             self.broker.logger.error(
                 "Failed to archive message %s on queue %s; it will be redelivered after its visibility timeout.",
@@ -800,17 +806,15 @@ def nack(self, message: "MessageProxy") -> None:
     @override
     def requeue(self, messages: Iterable["MessageProxy"]) -> None:
         """Make messages visible again immediately by resetting their visibility timeout."""
-        message_ids = [
-            message._postgres_message.msg_id for message in messages if isinstance(message, _PostgresMessage)
-        ]
+        message_ids = [message._postgres_message.msg_id for message in messages if isinstance(message, PostgresMessage)]
         self._requeue_message_ids(message_ids)
 
-    def _build_message(self, postgres_message: PostgresQueueMessage) -> "_PostgresMessage":
+    def _build_message(self, postgres_message: PostgresQueueMessage) -> "PostgresMessage":
         """Wrap a raw PGMQ row as a Remoulade message proxy."""
-        return _PostgresMessage(postgres_message)
+        return PostgresMessage(postgres_message)
 
     @override
-    def __next__(self) -> "_PostgresMessage | None":
+    def __next__(self) -> "PostgresMessage | None":
         """Return the next available message, or ``None`` if the queue stays empty
         or the consumer is at its in-flight capacity."""
         if self.messages:
@@ -850,7 +854,7 @@ def close(self) -> None:
         self.messages.clear()
 
 
-class _PostgresMessage(MessageProxy):
+class PostgresMessage(MessageProxy):
     def __init__(self, postgres_message: PostgresQueueMessage) -> None:
         """Wrap a PGMQ message row as a Remoulade message proxy."""
         payload = postgres_message.message
@@ -866,3 +870,22 @@ def __init__(self, postgres_message: PostgresQueueMessage) -> None:
             raise UnsupportedMessageEncoding("eta option isn't supported with postgres broker")
         super().__init__(message)
         self._postgres_message = postgres_message
+        self._header_patch: dict[str, Any] = {}
+
+    def stage_headers(self, patch: dict[str, Any]) -> bool:
+        """Merge a jsonb patch into the headers this message will be archived with.
+
+        Lets a state backend record a message's outcome without a write of its
+        own: the patch rides along with the archive that ack/nack performs
+        anyway.
+
+        Purely in memory: the patch is flushed by ``ack``/``nack``, folded into
+        the archive statement. Successive calls merge, key by key, so a backend
+        staging several patches over a message's life gets all of them archived.
+
+        Returns:
+          bool: Always True. The return value exists so a caller holding a proxy
+          of an unknown type can tell whether it still owes a write of its own.
+        """
+        self._header_patch.update(patch)
+        return True