pgmq-cpp

pgmq-cpp is a C++20 SDK and embeddable concurrent worker runtime for Postgres Message Queue (PGMQ). It lets a C++ service enqueue, claim, execute, retry, dead-letter, and replay background jobs without introducing a separate broker.

Important

This is an independent community project. It is not maintained, sponsored, endorsed, or supported by the PGMQ project or its maintainers. “PGMQ” is used only to identify the upstream PostgreSQL extension with which this library interoperates.

中文文档

What it provides

• A typed, exception-based SDK over the public PGMQ SQL API.
• Single and batch send/read/pop/delete/archive/visibility operations, including targeted reads and relative or absolute delivery times.
• Delayed messages, absolute delivery times, message headers, conditional reads, FIFO message groups, topic routing, and queue metrics.
• Caller-owned PostgreSQL transactions. The SDK never commits a transaction that the caller created.
• A bounded, multi-queue worker runtime with backpressure, retry policies, dead-letter queues, automatic lease renewal, graceful shutdown, and in-process metrics/events.
• Two processing modes:
  • regular for long-running or external work, with at-least-once processing;
  • transactional for short work confined to the same PostgreSQL database, with claim, business SQL, acknowledgement, and commit in one transaction.
• PGMQ LISTEN/NOTIFY wake-ups with periodic polling as the authoritative fallback.
• Installable CMake targets pgmq::client and pgmq::worker.

pgmq-cpp is a library, not a broker, workflow engine, daemon, HTTP service, or web console.

Build, install, and run

With a current vcpkg checkout and the pinned manifest in this repository, a minimal Linux build is:

git clone https://github.com/lite-tx/pgmq-cpp.git
cd pgmq-cpp
export VCPKG_ROOT=/path/to/vcpkg
cmake --preset linux-debug -DPGMQ_CPP_REGISTER_DOCKER_TESTS=OFF
cmake --build --preset linux-debug
cmake --install build/linux-debug --prefix out/pgmq-cpp
./out/pgmq-cpp/bin/pgmq-cppctl --version

The final command runs without a database and prints the installed CLI/library version. Database operations require PostgreSQL with PGMQ 1.12.0; continue with the Database setup and SDK quick start sections below. Windows, direct-CMake, static/shared, and external-consumer instructions are in the Build section.

Compatibility

The v1.0 full-support baseline is:

Component	Supported baseline
C++	C++20
CMake	3.25 or newer
PostgreSQL	14 through 18
PGMQ	1.12.0 installed as a PostgreSQL extension
Upstream reference	tag v1.12.0, commit 08ace4087dbf00e51704c5a3d9df2e15fd566127
Database client	libpq
JSON	nlohmann/json

PGMQ's API grew across several releases. Client::capabilities() detects individual SQL functions and result fields instead of guessing from the version string. However, the complete v1.0 SDK and worker runtime are supported as a unit only against PGMQ 1.12.0. The real-database SDK and worker suites passed on PostgreSQL 14, 15, 16, 17, and 18 with PGMQ 1.12.0 and pg_partman 5.1.0. Older PGMQ servers are not a supported full configuration. See the compatibility matrix.

The upstream SQL-only installation is not currently supported: capability detection requires an entry for pgmq in pg_extension.

Delivery semantics, in one minute

Regular workers use a visibility lease. A successful read increments PGMQ's read_ct; the runtime treats (queue, msg_id, read_ct) as the claim token. Renewal and acknowledgement additionally require an unexpired vt. If that guard fails, the runtime emits lease_lost, requests cooperative handler stopping, and refuses to acknowledge the message.

This prevents a stale worker from deleting a message after a newer claim has taken ownership. It does not turn external side effects into exactly-once effects. A process can crash after an API call, file write, or email succeeds but before the PostgreSQL acknowledgement commits. The message will then be delivered again. Use a stable business idempotency key.

Transactional workers keep the PGMQ claim, business SQL, and delete/archive in one PostgreSQL transaction. Before commit, a failure rolls everything back. This atomicity applies only to changes in that same PostgreSQL transaction; it does not cover remote services or other databases.

Regular lease deadlines and transactional retry delays are derived from the database clock, so they do not require the application host clock to match PostgreSQL. Caller-supplied absolute enqueue or visibility TimePoint values still represent wall-clock instants; the caller is responsible for supplying the intended time.

Retry durations in either worker mode must be in the inclusive range 0..9,007,199,254,740 ms (about 285 years). The upper bound keeps PostgreSQL's float8-backed interval calculation at or below 2^53 internal microseconds, preserving exact integer milliseconds. Explicit handler delays, configured retry policies, and both runtime scheduling paths reject negative or larger values with pgmq::Error / invalid_input before issuing visibility SQL.

Transactional handlers receive WorkerTransaction, a narrowed view without direct commit, rollback, acknowledgement, or visibility methods. This prevents accidental use of those convenience APIs. Its execute() method is a trusted arbitrary-SQL escape hatch: the handler contract forbids transaction-control SQL and SQL that acknowledges or changes the current task. The atomicity guarantee assumes that contract is respected.

execute() returns QueryResult: SQL NULL remains std::nullopt, and row or column lookup by integral index or name is bounds-checked. Invalid numeric indices and unknown names throw std::out_of_range; a null C-string name throws std::invalid_argument.

Read reliability semantics, the failure model, and transactional mode before production use.

Build

Prerequisites:

• a C++20 compiler;
• CMake 3.25+;
• Ninja for the supplied Linux presets, or another generator for a direct configure;
• PostgreSQL client headers and libpq;
• nlohmann/json;
• optional: vcpkg, Docker, Doxygen, and clang-tidy.

The repository includes a vcpkg manifest pinned to builtin baseline 40f3c709db80acf154ac4b17a1f83c564ebd022e:

export VCPKG_ROOT=/path/to/vcpkg
cmake --preset linux-debug
cmake --build --preset linux-debug

The supplied windows-debug, windows-release, and linux-debug presets explicitly register the real Docker fixture and therefore require Docker Compose. To use one of those presets without Docker, append -DPGMQ_CPP_REGISTER_DOCKER_TESTS=OFF to the configure command. A direct CMake configure leaves Docker-backed tests off by default; unit and install-consumer tests remain available when BUILD_TESTING=ON. The linux-sanitize preset also leaves Docker tests off. Without vcpkg, use a direct configure and provide discoverable system packages for libpq and nlohmann/json.

The manifest's public dependencies are libpq and nlohmann-json. A clean bootstrap of the pinned vcpkg baseline installed them successfully; local checkout or cache directories such as _tools/ are not project dependencies.

Configuration compile-links the C++20 stop-token, stop-callback, jthread, and stop-aware condition-variable interfaces used by the public API and runtime. On Clang/AppleClang standard libraries that expose those interfaces only with -fexperimental-library, CMake verifies that mode and propagates the required compile and link option through the installed targets. Configuration fails clearly if neither mode is usable. This supports source builds; it is not a precompiled cross-toolchain ABI promise.

On Visual Studio 2022:

$env:VCPKG_ROOT = 'C:\path\to\vcpkg'
cmake --preset windows-release
cmake --build --preset windows-release

Libraries are static by default. A direct configure can select either form:

cmake -S . -B build/static -DBUILD_SHARED_LIBS=OFF
cmake -S . -B build/shared -DBUILD_SHARED_LIBS=ON

To install:

cmake --install build/linux-debug --prefix /desired/prefix

An external consumer can then use:

find_package(pgmq-cpp 1 CONFIG REQUIRED)

add_executable(my_worker main.cpp)
target_link_libraries(my_worker PRIVATE pgmq::worker)
target_compile_features(my_worker PRIVATE cxx_std_20)

Link pgmq::client when the worker runtime is not needed. pgmq::worker transitively links the client API.

For an installed Windows shared build, the consumer process must be able to find the pgmq DLLs from the install prefix's bin directory and the libpq / OpenSSL runtime DLLs from the selected vcpkg triplet's bin directory. The installed-consumer test copies those runtime files beside its executable before running it. cmake --install does not bundle third-party runtime DLLs.

Database setup

Install PGMQ 1.12.0 using the upstream instructions, then enable it in the target database:

CREATE EXTENSION IF NOT EXISTS pgmq VERSION '1.12.0';

Use a dedicated application role and grant only the database privileges that its SDK and worker operations require. Partitioned queues additionally require pg_partman as described by PGMQ.

Queue names are checked before use. The v1.0 boundary is 1–47 lowercase ASCII letters, digits, or underscores ([a-z0-9_]+). This is an intentional conservative subset of PGMQ's public validation: PGMQ 1.12 preserves spelling in some metadata/notification configuration while lowercasing physical tables and trigger channels. Rejecting uppercase avoids case aliases and silent notification mismatches. Existing uppercase queue names are not supported by v1.0.

SDK quick start

#include <chrono>
#include <cstdlib>
#include <memory>
#include <optional>
#include <stdexcept>
#include <string>
#include <utility>

#include <pgmq/client.hpp>
#include <pgmq/worker.hpp>

int main() {
  const char* database_url = std::getenv("DATABASE_URL");
  if (database_url == nullptr) {
    throw std::runtime_error{"DATABASE_URL is required"};
  }

  auto client = std::make_shared<pgmq::Client>(pgmq::ClientOptions{
      .connection_string = std::string{database_url},
      .pool_size = 6,
      .statement_timeout = std::chrono::seconds{30},
      .pool_acquire_timeout = std::chrono::seconds{30},
  });

  const pgmq::QueueName queue{"thumbnail_jobs"};
  client->create_queue(queue);
  const auto sent = client->send(
      queue,
      pgmq::Json{{"image_id", 42}},
      std::optional<pgmq::Json>{
          pgmq::Json{{"trace_id", "01J..."}}},
      std::chrono::seconds{5});
  (void)sent;

  pgmq::QueueWorkerConfig config{queue};
  config.concurrency = 2;
  config.claim_batch_size = 4;
  config.max_in_flight = 8;
  config.visibility_timeout = std::chrono::seconds{30};
  config.lease_renewal_interval = std::chrono::seconds{10};
  config.handler =
      [](const pgmq::Task& task, pgmq::HandlerContext& context) {
        if (context.stop_token.stop_requested()) {
          return pgmq::HandlerResult::abandon("shutdown requested");
        }
        process_thumbnail(task.message.body);  // your idempotent operation
        return pgmq::HandlerResult::success_delete();
      };

  pgmq::WorkerRuntime runtime{client};
  runtime.add_queue(std::move(config));
  runtime.start();

  run_until_signal();

  const auto stopped = runtime.shutdown(std::chrono::seconds{30});
  return stopped.clean ? 0 : 2;
}

process_thumbnail and run_until_signal represent application code. Complete buildable programs live in examples/.

Caller-controlled transaction

auto transaction = client->begin_transaction();
const auto updated = transaction.execute(
    "UPDATE orders SET state = $1 WHERE order_id = $2::bigint",
    {std::string{"paid"}, std::string{"42"}});
const auto event_id = transaction.send(
    pgmq::QueueName{"order_events"}, pgmq::Json{{"order_id", 42}});
(void)updated;
(void)event_id;
transaction.commit();

Destroying an active Transaction attempts a rollback. Calls made through a Transaction do not reconnect or silently retry, because doing so would break the transaction boundary. A successful explicit commit() or rollback() returns the pinned connection to the pool immediately, even if the Transaction object remains in scope. The per-connection statement_timeout also applies to long-poll reads and transactional handlers; size it for the longest intended statement, or set it to zero only when the application provides another bounded cancellation policy. pool_acquire_timeout bounds how long a call waits for a free pooled connection and defaults to 30 seconds.

Worker results

A handler returns exactly one HandlerResult:

Result	Effect
success_delete()	Delete the owned source message
success_archive()	Move it to the PGMQ archive
retry(error, delay)	Make it visible after the selected backoff
dead_letter(error)	Atomically send an envelope to the DLQ and delete the source
abandon(reason)	Do not acknowledge; regular mode waits for lease expiry, transactional mode rolls back

For source names up to 43 bytes, the default dead-letter queue is <source>_dlq. Longer names use a truncated prefix plus _dlq_ and a stable 16-hex-digit hash, staying within 47 bytes without silently merging long-name defaults. The runtime creates the destination at startup and rejects an explicit DLQ equal to its source. Its message body uses schema pgmq-cpp.dead-letter.v1 and preserves the source queue, id, read_ct, body, headers, error summary, and dead-letter time.

Notifications are hints

PGMQ notifications contain no message body and can be throttled. PostgreSQL delivers them only after the producer transaction commits, may coalesce identical channel/payload notifications within one transaction, and does not retain them for disconnected listeners. A listener can also start during a race. Therefore:

• the runtime uses a notification only to wake a dispatcher;
• every wake-up still executes a PGMQ read;
• periodic polling always remains enabled;
• reconnecting the listener reissues LISTEN and restores notification configuration;
• notification setup failure degrades to polling.

Do not build correctness around notification counts.

Observability

MetricsRegistry::snapshot() exposes counters, in-flight gauges, and latency summaries. MetricsRegistry::openmetrics() emits text suitable for adaptation to a Prometheus/OpenMetrics endpoint owned by the host service. Runtime events include claim, success, retry, dead-letter, lease renewal/loss, reconnect, notification setup failure, notification/polling wake-up, and shutdown outcomes.

The callback runs in runtime threads; keep it non-blocking. Exceptions thrown by the callback are swallowed so observability cannot change task semantics. Metrics are process-local and reset on restart.

Operations and examples

• pgmq-cppctl: capabilities, queues, queue metrics, DLQ summary, and atomic replay.
• basic_producer_worker: normal concurrent processing.
• transactional_orders: commit and rollback in transactional mode.
• crash_recovery: lease expiry and processing after restart.

The command shape is:

pgmq-cppctl [--connection URI] [--pool-size N] [--json] capabilities
pgmq-cppctl [--connection URI] [--pool-size N] [--json] list
pgmq-cppctl [--connection URI] [--pool-size N] [--json] metrics [QUEUE]
pgmq-cppctl [--connection URI] [--pool-size N] [--json] dlq-summary DLQ [--sample N]
pgmq-cppctl [--connection URI] [--pool-size N] [--json] dlq-replay DLQ --id ID [--dry-run]
pgmq-cppctl [--connection URI] [--pool-size N] [--json] dlq-replay DLQ --batch N [--dry-run]

Without --connection, it reads DATABASE_URL. Replay selects only visible dead letters. The destination sends and DLQ deletes occur in one transaction: all selected terminal changes commit, or none do. Selection itself is a prior PGMQ read and therefore increments each selected DLQ entry's read_ct, even for a dry run or a later replay error; the CLI makes a best-effort release of those claims. Run pgmq-cppctl --help for details. It is an administrative client, not a server.

The examples accept:

basic_producer_worker [--connection URI] [--count N] [--concurrency N]
transactional_orders [--connection URI]
crash_recovery [--connection URI] [--visibility-seconds N]

They also fall back to DATABASE_URL and create run-specific resources that they remove on a successful run.

Verification and benchmarks

The repository distinguishes test code from test evidence. A CI file, Docker Compose file, or benchmark executable is not itself proof that an environment passed.

• Verification ledger
• Benchmark protocol and results policy
• Contributing and local checks

Environment rows that have not actually run are explicitly marked not run. No performance number should be quoted without the raw command, hardware, build type, PostgreSQL/PGMQ versions, and parameters.

The current source completed the recorded v1.0.0 acceptance suite on 2026-07-26:

• Windows MSVC 19.41 Debug and Release, both static and shared, with /W4 /WX;
• Ubuntu 24.04 GCC 13.3 Debug and Clang 18.1 Release builds and installed consumers;
• real PGMQ 1.12.0 SDK and worker tests on PostgreSQL 14 through 18;
• GCC and Clang ASan/UBSan, strict clang-tidy, Doxygen, CLI, examples, crash recovery, multi-process competition, and a real-database benchmark smoke run.

The public main branch is hosted on GitHub. Its first hosted run exposed Linux documentation, Windows shell, and macOS C++20 library portability defects. After those defects were fixed, follow-up run 30200789701 passed all five jobs on source commit 58edaa86976ba510a38206f290f571705e813645: Linux real PGMQ, Linux ASan/UBSan, strict clang-tidy, Windows build/package, and macOS build/package. The macOS job used macOS 15.7.7 ARM64, Xcode 16.4, and Apple Clang 17.0.0; its verified libc++ path propagated -fexperimental-library to the installed consumer. GCC and Clang sanitizer runs also passed locally after the missing Clang runtime was installed in the validation image. See the verification ledger for exact commands, job URLs, initial failures, and proof boundaries. When the v1.0.0 GitHub Release is present, it contains source archives only, not precompiled binaries. The GitHub Releases page is the canonical publication record.

Documentation

• Problem and scope
• Compatibility
• Architecture
• Reliability semantics
• Failure model
• Transactional mode
• API design examples
• ADRs

Generate local API documentation with:

doxygen Doxyfile

Doxygen 1.9.8 generated the current reference with zero warnings. The configuration fails on structural, parser, and broken-reference warnings; it does not claim that every public member already has prose documentation.

License

pgmq-cpp is available under the MIT License. PGMQ itself is a separate project under the PostgreSQL License; installing or redistributing it is governed by its own license.