Architecture

Overview

pgmq-cpp is an in-process library. PostgreSQL and the PGMQ extension remain the queue authority; there is no pgmq-cpp server.

flowchart LR
  A["C++ application"] --> C["pgmq::Client"]
  A --> W["pgmq::WorkerRuntime"]
  W --> C
  C --> P["bounded libpq connection pool"]
  W --> N["dedicated notification listener"]
  P --> D["PostgreSQL 14-18<br/>PGMQ 1.12.0"]
  N --> D
  W --> M["MetricsRegistry<br/>event callback"]

The public headers contain C++ values, std::chrono types, nlohmann/json, and exceptions. libpq handles and implementation synchronization remain private.

SDK layer

Client and pool

Client owns a bounded lazy connection pool. A normal SDK call leases one connection for the duration of the call and returns it only when healthy. No libpq connection is used concurrently by multiple threads.

Calls on one stable Client instance may run concurrently. Moving or destroying that client while another thread is using it is still a caller error, as with ordinary C++ object lifetime. A Transaction pins one connection and is single-threaded; callers must serialize its methods. Likewise, one NotificationListener is intended to have one active waiter.

Before an operation, an unhealthy idle connection can be reset. An operation that loses its connection after it has been sent is not transparently replayed: the commit outcome may be unknown and automatic replay could duplicate a write.

Each connection sets the configured PostgreSQL statement_timeout (30 seconds by default). It applies to SDK SQL, long-poll functions, and business SQL in a transaction. Zero disables the server-side statement timeout; callers should retain another bounded cancellation policy if they choose zero.

connect_timeout bounds a new PostgreSQL connection attempt; it is not a pool acquisition timeout. When every pooled connection is leased, another SDK call waits up to pool_acquire_timeout (30 seconds by default) and then throws a timeout-category pgmq::Error. Long-poll reads and transactions retain their lease for their full duration, so production pool sizing must account for them.

Every SQL value is passed through PQexecParams. The small number of dynamic queue-table identifiers used by the worker are derived from a validated lowercase QueueName and quoted as PostgreSQL identifiers. Restricting v1.0 queue names to [a-z0-9_]+ avoids PGMQ 1.12 case aliases between metadata, physical tables, and notification channels.

Transactions

Client::begin_transaction() leases and pins one connection and executes BEGIN. Transaction exposes parameterized business SQL plus the queue operations needed for atomic producer and consumer flows.

• commit() and rollback() are explicit.
• A successful commit() or rollback() releases the pinned pool connection immediately; release does not wait for the C++ object to leave scope.
• Destroying an active transaction attempts ROLLBACK.
• Transaction operations disable reconnect and retry.
• The SDK never commits a transaction on the caller's behalf.
• The caller must not use one transaction concurrently from multiple threads.

Capabilities

Client::capabilities() reads PostgreSQL and PGMQ versions and probes exact function signatures/result fields in the catalogs. The result is cached within the client. The full supported configuration is still PGMQ 1.12.0; capability flags are not a blanket compatibility promise for older record layouts.

Errors

Operational validation, connection, database, protocol, timeout, and unsupported-capability failures use pgmq::Error, which contains:

• ErrorCategory;
• PostgreSQL SQLSTATE when available;
• operation name;
• server detail;
• retryable() classification.

There is no parallel Result<T> error convention. Ordinary value-access APIs retain normal C++ behavior: invalid QueryResult::row()/at() access throws std::out_of_range, and direct nlohmann/json operations can throw that library's exceptions.

Regular worker mode

Each regular-mode queue has:

• one dispatcher;
• concurrency handler std::jthreads;
• one bounded pending queue;
• an in-flight counter covering claimed pending and executing messages.

The process also has one centralized lease-renewal thread and one notification thread shared across queues.

flowchart TD
  R["PGMQ read<br/>claim_batch_size"] --> G{"in-flight capacity"}
  G -->|available| B["bounded pending queue"]
  B --> H1["handler 1"]
  B --> H2["handler N"]
  H1 --> O["guarded terminal action"]
  H2 --> O
  L["central lease scheduler"] -->|"batch UPDATE per queue"| R
  Q["NOTIFY or polling deadline"] --> R

The dispatcher never claims more than min(claim_batch_size, max_in_flight - current_in_flight). The bounded queue and counter prevent unlimited threads or buffered tasks. Claimed tasks waiting locally are included in centralized renewal.

Claim ownership

PGMQ does not return an SQS-style opaque receipt handle. A regular claim stores the msg_id, the read_ct returned by this read, and the lease deadline. Every message projection also returns PostgreSQL clock_timestamp() as database_observed_at. The runtime computes the server-clock remaining lease as visible_at - database_observed_at, clamps it to the configured visibility timeout, and maps that duration onto steady_clock. Client/database wall-clock offset and later wall-clock jumps therefore do not move the in-process renewal deadline. Response transit after the database observation is not included in that duration; the guarded vt > clock_timestamp() renewal and terminal SQL remain authoritative and fail closed if ownership has already expired.

Renewal is one conditional update per due queue group:

WHERE q.msg_id = claim.msg_id
  AND q.read_ct = claim.read_ct
  AND q.vt > clock_timestamp()

Delete, archive, retry scheduling, and dead-letter moves first acquire a row lock and test the same ownership predicate in the transaction that performs the terminal operation. A failed predicate means the claim is stale: acknowledgement is disabled, lease_lost is emitted, and the handler's stop_token is requested.

The rationale is recorded in ADR 0002.

Completion, retry, and dead-letter

Handler results are mapped to guarded database transactions:

• delete or archive the source;
• update vt for retry;
• or send a dead-letter envelope and delete the source atomically.

Automatic retry delay is exponential, capped, and jittered. Clock and random sources can be injected for deterministic tests. read_ct is the attempt number. A retry at or beyond max_attempts becomes a dead letter.

For source names up to 43 bytes the destination defaults to <source>_dlq. Longer names use a truncated prefix plus _dlq_ and a stable 16-hex-digit hash, producing a 47-byte deterministic name. It is created during runtime startup as an ordinary PGMQ queue; an explicit destination equal to the source is rejected.

Transactional worker mode

Transactional mode has concurrency worker threads per queue. Each thread:

1. begins a transaction;
2. reads exactly one message;
3. establishes a savepoint after the claim;
4. invokes the handler with a non-owning WorkerTransaction view over the same transaction;
5. preserves handler SQL only on success;
6. deletes or archives the message;
7. commits.

For retry or dead-letter, the runtime rolls back to the post-claim savepoint before scheduling the retry or moving the message. For abandon, cancellation, or pre-commit process death, the outer transaction rolls back.

WorkerTransaction exposes parameterized business SQL and producer sends, but not direct commit, rollback, delete, archive, or visibility methods. This narrows the accidental-misuse surface; it is not a security boundary. execute() accepts arbitrary SQL from a trusted handler. The handler contract forbids transaction-control statements and SQL that acknowledges or changes the current task. Atomicity claims assume that contract is respected.

The row remains transactionally locked, so this mode does not use the regular lease-renewal scheduler. It is intended only for short, same-database work.

At startup, the pool must have at least:

sum(transactional concurrency across queues) + 2

connections when queues are registered. The two shared slots cover dispatcher and lease/client work. A larger pool is usually appropriate when producers, regular queues, or administrative calls share the client.

Wake-up path

PGMQ notification triggers emit on pgmq.q_<lowercase_queue>.INSERT. The listener uses a dedicated connection so waiting for notifications cannot occupy a client-pool lease.

On a notification, the matching queue's condition variable is signalled. The dispatcher still calls PGMQ read; the notification contains no work item. Every queue also has a positive polling_interval. Listener disconnects cause reconnect and LISTEN restoration; configuration/listening failure leaves polling active.

See ADR 0003.

Startup

WorkerRuntime::start() performs database/capability setup before launching the runtime threads. If a thread launch fails after earlier threads have started, the runtime requests their stop and joins them before returning the exception. A later start() on the same object is supported: it first rebuilds the per-run queues, counters, and thread holders so no closed local queue from the failed attempt can be reused.

Shutdown

Shutdown:

1. stops notification waits and new claims;
2. closes and drains pending local queues without acknowledging them;
3. requests each active handler's stop_token;
4. waits up to the caller's grace period;
5. joins runtime threads on a clean drain.

If the grace period expires, the result contains timed_out=true and the remaining in-flight count. The runtime disables their acknowledgements and stops renewing them. It cannot safely kill C++ handler code. The owning application decides when to terminate the process and must keep referenced state alive until its handlers actually return.

Observability

MetricsRegistry records:

• claimed, succeeded, retried, dead-lettered, and abandoned;
• in-flight;
• handler and queue-wait latency;
• lease renewal success/failure/loss;
• database reconnects;
• notification setup failures and notification/polling wake-ups;
• shutdown duration/timeouts.

Snapshots and OpenMetrics text are in-process adapters; the library does not open a port. Event callbacks run from runtime threads and are exception isolated from task processing.

Packaging boundaries

• pgmq::client: SDK, types, errors, libpq implementation.
• pgmq::worker: runtime and metrics; publicly depends on pgmq::client.
• pgmq-cppctl: optional operational executable.
• examples, tests, and benchmarks: never part of the library ABI.

The installed package exports the two namespaced targets and public headers without exposing private libpq wrappers.