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 system = client.system();
let queues = pgmq.client(&system);
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.system().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

Add the extension declaration above to your application's schema. Confirm that your PostgreSQL provider offers the extension and that your migration role can install it, then apply the migration and run your application:

rent migrate dev --name enable_pgmq
cargo run

Create the queue before starting producers and workers. Use the same transaction-bound client when a message and its application record must commit together. Workers should make retries idempotent.

Complete example

Download this project, or copy the files below into an empty directory.

Replace RENT_REPOSITORY_URL in Cargo.toml with the URL supplied with your preview access. Use the matching Rent CLI release.

Use an empty PostgreSQL database with the extension binaries available on its server. Set DATABASE_URL to that database and follow any server setup requirements described above.

rent extension check --provider self-hosted
rent generate
rent migrate dev --name initial
cargo run

If your host preinstalls other extensions, declare those in RSL as well. For a database with existing application tables, baseline it first.

Cargo.toml

[package]
name = "rent-extension-pgmq"
version = "0.1.0"
edition = "2024"
[workspace]
[[bin]]
name = "rent-extension-pgmq"
path = "main.rs"
[dependencies]
rent = { git = "RENT_REPOSITORY_URL", tag = "v0.8.6", default-features = false, features = ["postgres", "extensions", "migrate"] }
rent-ext-pgmq = { git = "RENT_REPOSITORY_URL", tag = "v0.8.6" }
anyhow = "1.0.104"
tokio = { version = "1.53.1", features = ["full"] }
serde_json = "1.0.151"

rent.toml

schema = "schema.rsl"
generated = "generated"
migrations = "migrations"

[database]
dialect = "postgres"
backends = ["postgres"]

schema.rsl

datasource db {
  provider = "postgresql"
}

extension pgmq {
  name = "pgmq"
}

model Post {
  id    BigInt  @id  @default(autoincrement())
  title String
  body  String  @default("")
}

main.rs

use anyhow::Result;
use rent::{driver::DatabasePool, extension::ExtensionRegistry};
#[allow(dead_code, unused_imports)]
#[rustfmt::skip]
#[path = "generated/mod.rs"]
mod generated;
#[tokio::main]
async fn main() -> Result<()> {
    let pool = DatabasePool::connect_url(&std::env::var("DATABASE_URL")?).await?;
    let client = generated::client::Client::new(pool.clone());
    let mut extensions = ExtensionRegistry::default();
    use rent_ext_pgmq::{Pgmq, PgmqClientExt};
    let pack = extensions.register_pack::<Pgmq>()?;
    let queue_client = pack.client(&pool);
    let queue = queue_client.queue("post_events");
    queue.create().await?;
    let message = client
        .transaction_app(async |tx| -> anyhow::Result<i64> {
            let post = tx.create_post().title("New post").save().await?;
            let id = pack
                .client(&tx.system())
                .queue("post_events")
                .send(serde_json::json!({"post_id": post.id}))
                .await?;
            Ok(i64::from(id))
        })
        .await?;
    println!("Post and event {message} committed together.");
    Ok(())
}

pgmq queues

On this page