pgmq queues

What it is

PGMQ is a message queue implemented inside PostgreSQL. It stores messages in database tables and lets workers reserve them with a visibility timeout, without adding a separate queue service.

What it provides

• Queues whose messages participate in PostgreSQL durability and backups
• Visibility timeouts so a worker can claim a message before acknowledging it
• Archive and replay workflows for jobs that need an audit trail
• Batched publication and reads for background workers

Use it for background jobs, webhook delivery, transactional outboxes, email dispatch, document processing, or any application that wants its queue and relational state in the same operational boundary.

Use it with Rent

extension pgmq {
  name = "pgmq"
}

The RSL declaration makes installation and migration intent part of the same reviewed schema as application models. Register the typed pack at runtime to call its APIs:

use std::time::Duration;

use rent_ext_pgmq::{Pgmq, PgmqClientExt};
use serde::{Deserialize, Serialize};

#[derive(Serialize, Deserialize)]
struct PostPublished {
    event_id: String,
    post_id: i64,
    author_id: i64,
}

let pgmq = extensions.register_pack::<Pgmq>()?;
let queues = pgmq.client(&client);
let events = queues
    .queue("post_events")
    .typed::<PostPublished>();

events.create().await?;
let event = PostPublished {
    event_id: "post-42-published-v1".to_owned(),
    post_id: 42,
    author_id: 7,
};
let id = events.send(&event).await?;

let messages = events
    .read(Duration::from_secs(30), 10)
    .await?;

for message in messages {
    println!(
        "Post {}: delivery {}",
        message.payload.post_id,
        message.read_count,
    );

    events.archive(message.id).await?;
}

Queue names and payloads are bound parameters. MessageId keeps message identifiers distinct from application IDs; use id.value() when storing an idempotency key. Each message includes id, payload, read_count, enqueued_at, visible_at, and optional headers. Unknown additive server metadata is accepted.

Publish atomically with application data

Bind the pack to the transaction client to commit the application write and queue message together:

client
    .transaction_app(async |tx| -> anyhow::Result<()> {
        let post = tx
            .create_post()
            .id(42)
            .title("Hello")
            .body("My first published post.")
            .published(true)
            .author_id(7)
            .save()
            .await?;

        pgmq.client(tx)
            .queue("post_events")
            .typed::<PostPublished>()
            .send(PostPublished {
                event_id: format!("post-{}-published-v1", post.id),
                post_id: post.id,
                author_id: post.author_id,
            })
            .await?;

        Ok(())
    })
    .await?;

Create the queue during application setup. An error from the transaction closure rolls back both writes. Binding to the outer pool inside that closure would instead execute outside the transaction.

Batches and delivery guarantees

let batch = [
    PostPublished {
        event_id: "post-42-published-v1".to_owned(),
        post_id: 42,
        author_id: 7,
    },
    PostPublished {
        event_id: "post-43-published-v1".to_owned(),
        post_id: 43,
        author_id: 7,
    },
];

let ids = events.send_many(&batch).await?;

send_many sends one statement and returns one ID per payload in input order. Empty batches perform no SQL. read(timeout, quantity) returns up to the requested number of currently visible messages. A positive fractional-second timeout rounds up; zero allows immediate redelivery. Counts must be positive and both count and rounded timeout must fit PostgreSQL's signed 32-bit integer range.

Delivery is at least once, not exactly once. A crash before acknowledgement or an expired timeout can cause redelivery. Parallel workers can finish out of order. Make effects idempotent using a unique event/message key. For effects stored in the same database, a worker transaction can insert that key, apply the effect only when the key is new, and archive the message together. For external effects such as email or payment requests, use the external service's idempotency mechanism; a database transaction cannot undo an external request.

• archive(id) removes an active message and retains an archive copy.
• delete(id) removes it without retaining an archive copy.
• Both return false if the active message is already absent.

Build an idempotent worker

The event's event_id identifies a business action. It stays the same if a producer publishes that action again; the queue assigns a new MessageId to each publication. Use the business ID to deduplicate both redelivery and duplicate publication. A real application should include the publication version or another stable event key so two intentional publications are not mistaken for one.

The tutorial's worker records one system comment on the published post. Its receipt table has a unique key:

CREATE TABLE processed_events (
    event_id TEXT PRIMARY KEY,
    message_id BIGINT NOT NULL
);

Claim the key, write the comment, and archive the delivery in one transaction:

use rent::driver::StatementExecutor;
use rent::sql::Dialect;
use rent::sql::builder::{ConflictClause, Insert, InsertValue};

client
    .transaction_app(async |tx| -> anyhow::Result<()> {
        let claim = Insert::new(Dialect::Postgres, "processed_events")
            .columns(["event_id", "message_id"])
            .values([
                InsertValue::from(message.payload.event_id.clone()),
                InsertValue::from(message.id.value()),
            ])
            .on_conflict(ConflictClause::do_nothing().columns(["event_id"]))
            .render()?;

        let claim_result = tx.execute_statement(&claim).await?;

        if claim_result.rows_affected == 1 {
            let post = tx
                .post()
                .id_eq(message.payload.post_id)
                .author_id_eq(message.payload.author_id)
                .only()
                .await?;

            tx.create_comment()
                .id(message.id.value())
                .body("Publication recorded by the background worker.")
                .author_id(worker.id)
                .connect_post(&post)
                .save()
                .await?;
        }

        pgmq.client(tx)
            .queue("post_events")
            .archive(message.id)
            .await?;

        Ok(())
    })
    .await?;

The unique constraint arbitrates simultaneous claims; do not replace it with an existence query followed by an insert. If the worker fails before commit, its receipt, comment, and archive all roll back. The message remains eligible for redelivery after its visibility timeout. A duplicate is acknowledged without writing another comment.

The receipt insert uses Rent's lower-level statement executor to distinguish insertion from conflict without an extra query. It participates in the transaction but does not infer entity policies. Keep this table internal to the worker; scope keys and permissions appropriately in a multi-tenant application. Generated post/comment operations still run their normal policies. This pattern covers database effects, not external service calls.

Flexible payloads and failures

Omit .typed::<PostPublished>() to use serde_json::Value payloads while keeping typed delivery metadata. For custom SQL/record processing, queues.query_json(rent_ext_pgmq::read(...)) is the lower-level escape hatch.

Payload decoding errors include the queue name and message ID when available. A failed decode does not acknowledge the message: its visibility timeout still applies. Decide whether to retry, archive for inspection, or quarantine malformed events. Invalid timeout/count options fail before SQL execution.

Run the application

Use a disposable PostgreSQL database with pgmq installed. Export DATABASE_URL in the shell running the example:

cargo run -p rent --example extension_01_pgmq

The application reuses the generated User/Post/Comment client from the relationships chapter. It creates temporary application tables and a uniquely named queue, then removes that queue and closes the connection. It does not reset your existing application tables or queues. No DATABASE_URL or a missing extension is an error, not an offline success.

The progression executes and checks:

1. Create an author and worker account.
2. Roll back a publication and verify that neither the post nor event remains.
3. Commit a post and event together, then read the typed payload and delivery metadata.
4. Fail a worker after writing its receipt/comment and archiving; verify all three roll back.
5. Redeliver the same message and commit one comment successfully.
6. Republish the same business event with a new message ID; acknowledge it without duplicating the comment.
7. Eager-load the post's comments and verify the single effect.

The executable implementation is crates/rent/examples/extension_01_pgmq/workflow.rs; its standalone extension declaration is crates/rent/examples/extension_01_pgmq/schema.rsl. Its generated application models come from crates/rent/examples/tutorial_02_relationships/schema.rsl.

Run bash scripts/test-extension-matrix.sh to provision the pinned providers and execute this same workflow through nextest, plus the batch, visibility, concurrent-consumer, and archive/delete contracts. Ordinary nextest runs keep a separate offline API/schema check; they do not substitute for the required live workflow.