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

docs/source/changelog.rst

@@ -5,6 +5,30 @@ Changelog
 
 All notable changes to this project will be documented in this file.
 
+`7.1.0`_ -- 2026-08-21
+----------------------
+Feat
+^^^^
+* Add ``PostgresBackend``, a write-only state backend for the PostgreSQL/PGMQ broker that stores a message's
+  terminal status in the pgmq message itself rather than in a table of its own: one ``UPDATE`` per processed
+  message, ``Pending``/``Started`` left to PGMQ's own columns, ``Message.set_progress`` silently dropped, and
+  retention bounded by the archive's rather than by ``state_ttl``. No read path — ``get_state``, ``get_states``,
+  ``get_states_count`` and ``clean`` raise ``NotImplementedError``
+* ``State`` carries a ``delivery_id``, the broker's own id for the delivery a state was observed on, which
+  ``PostgresBackend`` uses to name the row to write on. ``None`` for a broker that has no such id.
+
+Fix
+^^^
+* ``PostgresBroker.declare_queue`` ensures the ``msg_id`` index on a queue that already exists and not only on
+  one it creates, so a queue predating the index stops seq scanning every partition on each ack. The initial
+  build locks the table for its duration.
+
+Breaking changes
+^^^^^^^^^^^^^^^^
+* ``PostgresBroker`` rejects a queue name it could not use as a SQL identifier, and caps it 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. Declaring an actor on a longer name now raises ``ValueError``.
+
 `7.0.0`_ -- 2026-06-15
 ------------
 Breaking changes

docs/source/guide.rst

@@ -449,6 +449,51 @@ 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, message states can be tracked without a table of their
+own: ``PostgresBackend`` writes the status in the ``headers`` column of the PGMQ
+row that already 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 statuses (``Success``, ``Failure``, ``Skipped``, ``Canceled``)
+are stored. ``Pending`` and ``Started`` write nothing, because PGMQ already
+records them: ``enqueued_at``, ``read_ct`` and ``last_read_at`` on the row.
+This backend is **write-only**: ``get_state``, ``get_states``,
+``get_states_count`` and ``clean`` raise ``NotImplementedError``
+
+  SELECT message->>'message_id', message->>'actor_name', headers->>'status', read_ct, archived_at
+  FROM pgmq.a_default
+  WHERE archived_at > now() - interval '1 day';
+
+``Message.set_progress`` is silently dropped..
+
+Two differences with the Redis state backend are worth planning for:
+
+* ``state_ttl`` is ignored. Retention is the archive's, driven by
+  ``archive_retention_interval_in_days`` and ``pg_partman``: a status is gone once
+  the partition holding its message is dropped. Purging or dropping a queue
+  destroys the statuses too.
+* A status cannot outlive its message, since the backend keeps no store of its own.
+
+.. note::
+
+   A status is written on the row named by ``State.delivery_id``, not by
+   ``message_id``: a retry re-enqueues the same message, so one ``message_id`` can
+   name several rows of a queue at once and nothing 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/broker.py

@@ -624,6 +624,14 @@ def fail(self):
         """Mark this message for rejection."""
         self.failed = True
 
+    @property
+    def delivery_id(self):
+        """The broker's own id for this delivery, or ``None`` if it has none.
+
+        One per delivery, unlike ``message_id``, which a retry keeps.
+        """
+        return None
+
     def __getattr__(self, name):
         return getattr(self._message, name)
 

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,
@@ -189,7 +189,8 @@ def _try_enable_notify(self, queue_name: str) -> None:
 
     def _queue_exists(self, queue_name: str) -> bool:
         """Return whether the queue already exists in PostgreSQL."""
-        return queue_name in {queue.queue_name for queue in self.client.list_queues()}
+        queues = self.client.list_queues(conn=self._current_connection)
+        return queue_name in {queue.queue_name for queue in queues}
 
     @override
     def close(self) -> None:
@@ -216,14 +217,12 @@ def consume(self, queue_name: str, prefetch: int = 1, timeout: int = 30000) -> "
     @override
     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.
+        Raises:
+          ValueError: If ``queue_name`` cannot be used as a SQL identifier.
         """
         if queue_name in self.queues:
             return
+        assert_valid_queue_name(queue_name)
         with self.tx():
             if self._current_connection is None:
                 raise ValueError("cannot be None we are inside a tx")
@@ -244,29 +243,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.
 
@@ -352,7 +335,7 @@ def flush_all(self) -> None:
             self.flush(queue_name)
 
     def _count_enqueued_messages(self, queue_name: str) -> int:
-        """Count every message stored in the queue."""
+        """Count every message stored in the queue, on a connection of its own."""
         return self.client.metrics(queue_name).queue_length
 
     @override
@@ -774,15 +757,15 @@ def _archive_message(self, message: "MessageProxy") -> None:
         visible again once its visibility timeout expires and is redelivered,
         which is the broker's at-least-once guarantee.
         """
-        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)
+        self._unregister_heartbeat_message_id(message.delivery_id)
         try:
-            self.client.archive(self.queue_name, message._postgres_message.msg_id)
+            self.client.archive(self.queue_name, message.delivery_id)
         except Exception:
             self.broker.logger.error(
                 "Failed to archive message %s on queue %s; it will be redelivered after its visibility timeout.",
-                message._postgres_message.msg_id,
+                message.delivery_id,
                 self.queue_name,
                 exc_info=True,
             )
@@ -800,17 +783,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.delivery_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 +831,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 +847,9 @@ 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
+
+    @property
+    @override
+    def delivery_id(self) -> int:
+        """This delivery's PGMQ ``msg_id``, which a retry does not keep."""
+        return self._postgres_message.msg_id

remoulade/helpers/postgres_client.py

@@ -0,0 +1,109 @@
+# This file is a part of Remoulade.
+#
+# Copyright (C) 2026 WIREMIND SAS <dev@wiremind.io>
+#
+# Remoulade is free software; you can redistribute it and/or modify it
+# under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or (at
+# your option) any later version.
+#
+# Remoulade is distributed in the hope that it will be useful, but WITHOUT
+# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
+# FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
+# License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""Remoulade's hand-written SQL on top of the PGMQ client.
+
+Every statement remoulade writes itself lives here, so ``PostgresBroker`` and the
+pgmq state backend never build SQL of their own.
+"""
+
+import json
+from collections.abc import Iterator
+from contextlib import contextmanager
+from typing import Any
+
+from pgmq import SQLAlchemyPGMQueue
+from sqlalchemy import Connection, text
+
+from ..actor import QUEUE_NAME_PATTERN
+
+#: pgmq's own limit on a queue name. It also keeps the longest identifier remoulade
+#: derives from one (``q_<queue>_msg_id_idx``) under PostgreSQL's 63-byte cap, past
+#: which PostgreSQL truncates -- and two long queue names would collide on one index.
+QUEUE_NAME_MAX_LENGTH = 47
+
+
+def assert_valid_queue_name(queue_name: str) -> None:
+    """Check that ``queue_name`` is safe to interpolate as a SQL identifier.
+
+    The character set is remoulade's own (:data:`~remoulade.actor.QUEUE_NAME_PATTERN`,
+    already enforced on every actor declaration), which holds nothing needing quotes
+    or escaping; only the length bound is specific to PostgreSQL.
+
+    Parameters:
+      queue_name(str): The name to check.
+
+    Raises:
+      ValueError: If the name does not match :data:`~remoulade.actor.QUEUE_NAME_PATTERN`
+        or is longer than :data:`QUEUE_NAME_MAX_LENGTH`.
+    """
+    if not QUEUE_NAME_PATTERN.fullmatch(queue_name) or len(queue_name) > QUEUE_NAME_MAX_LENGTH:
+        raise ValueError(
+            f"{queue_name!r} is not a usable queue name for a PostgresBroker: it becomes a SQL identifier, so it "
+            f"must start with a letter or an underscore, hold only letters, digits, dashes, dots and underscores, "
+            f"and be at most {QUEUE_NAME_MAX_LENGTH} characters long."
+        )
+
+
+class RemouladePostgresClient(SQLAlchemyPGMQueue):
+    """A PGMQ client that also knows how to patch a remoulade message's headers.
+
+    Inherits the whole PGMQ surface (``send``, ``read``, ``archive``,
+    ``set_vt``, ``metrics``, ...) unchanged and adds the statements remoulade
+    needs on top of it.
+    """
+
+    def create_indexes(self, queue_name: str, conn: Connection | None = None) -> None:
+        """Ensure the queue table carries the indexes remoulade needs.
+        Raises:
+          ValueError: If ``queue_name`` cannot be used as a SQL identifier.
+        """
+        assert_valid_queue_name(queue_name)
+        with self._connection(conn) as connection:
+            connection.execute(
+                text(f'CREATE INDEX IF NOT EXISTS "q_{queue_name}_msg_id_idx" ON pgmq."q_{queue_name}" (msg_id)')
+            )
+
+    def patch_headers(self, queue: str, msg_id: int, patch: dict[str, Any], conn: Connection | None = None) -> bool:
+        """Merge ``patch`` into an enqueued message's headers, key by key.
+
+        Only reaches a message still in ``pgmq.q_<queue>``; once archived, its
+        headers are out of reach. ``pgmq.archive`` carries them over.
+
+        Returns:
+          bool: Whether a row was patched.
+        """
+        assert_valid_queue_name(queue)
+        statement = text(f"""
+            UPDATE pgmq."q_{queue}"
+            SET headers = coalesce(headers, '{{}}'::jsonb) || CAST(:patch AS jsonb)
+            WHERE msg_id = :msg_id
+        """)  # noqa: S608
+        return self._run(statement, {"msg_id": msg_id, "patch": json.dumps(patch)}, conn) > 0
+
+    @contextmanager
+    def _connection(self, conn: Connection | None) -> Iterator[Connection]:
+        """Yield the caller's connection, or open a transaction of our own."""
+        if conn is not None:
+            yield conn
+            return
+        with self.engine.begin() as connection:
+            yield connection
+
+    def _run(self, statement: Any, params: dict[str, Any], conn: Connection | None) -> int:
+        """Execute a write statement, returning the number of affected rows."""
+        with self._connection(conn) as connection:
+            return connection.execute(statement, params).rowcount