Transactional mode

July 26, 2026 ยท View on GitHub

Transactional mode is for short handlers whose durable effects are entirely in the same PostgreSQL database as PGMQ.

Atomic unit

One worker thread performs:

BEGIN
  PGMQ read/claim
  SAVEPOINT pgmq_cpp_handler
  business SQL through WorkerTransaction on the same transaction
  PGMQ delete or archive
COMMIT

The read updates PGMQ visibility and read_ct inside the open transaction and holds the relevant row lock. A competing PGMQ consumer skips that locked row. There is no gap in which business SQL is committed but acknowledgement is not, or vice versa.

Example

#include <cstdint>
#include <memory>
#include <string>
#include <utility>

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

void add_order_worker(pgmq::WorkerRuntime& runtime) {
  pgmq::QueueWorkerConfig config{pgmq::QueueName{"order_jobs"}};
  config.mode = pgmq::ProcessingMode::transactional;
  config.concurrency = 4;
  config.max_in_flight = 4;
  config.transactional_handler =
      [](const pgmq::Task& task,
         pgmq::WorkerTransaction& transaction,
         pgmq::HandlerContext& context) {
        if (context.stop_token.stop_requested()) {
          return pgmq::HandlerResult::abandon("shutdown requested");
        }

        const auto order_id =
            task.message.body.at("order_id").get<std::int64_t>();
        const auto result = transaction.execute(
            "INSERT INTO fulfilled_orders(order_id, source_msg_id) "
            "VALUES (\$1::bigint, \$2::bigint) "
            "ON CONFLICT (order_id) DO NOTHING",
            {std::to_string(order_id),
             std::to_string(task.message.id.value)});
        (void)result;
        return pgmq::HandlerResult::success_delete();
      };

  runtime.add_queue(std::move(config));
}

The handler receives a non-owning WorkerTransaction&. It can execute parameterized business SQL and send messages in the same transaction, but the type does not expose commit, rollback, delete, archive, or visibility methods. This prevents accidental use of those convenience APIs. execute() remains an arbitrary-SQL escape hatch for trusted application code, not a SQL sandbox. The handler contract forbids transaction-control SQL and SQL that acknowledges or changes the current task. Subject to that contract, the handler returns a disposition and the runtime alone finishes the claim and transaction.

Result behavior

The runtime creates a savepoint after the claim so failed business SQL can be discarded without discarding the claim needed to schedule a retry or make a dead-letter move.

Handler outcomeBusiness SQL after savepointQueue action
success_deletepreserveddelete, then commit
success_archivepreservedarchive, then commit
retryrolled back to savepointset future visibility, then commit
dead_letterrolled back to savepointsend DLQ envelope + delete source, then commit
abandonentire transaction rolled backoriginal claim also rolls back
throwrolled back to savepointretry or DLQ according to classification
cancellationentire transaction rolled backoriginal claim also rolls back

Transactional retry sends a validated integer millisecond duration to PostgreSQL. The transaction computes the new visibility timestamp as clock_timestamp() + delay, so application/database wall-clock skew does not change the backoff. This differs from caller-supplied absolute enqueue or visibility TimePoint values, whose meaning remains the caller's responsibility.

The accepted retry-duration range is 0..9,007,199,254,740 ms inclusive (about 285 years). The upper bound keeps PostgreSQL's float8-backed interval calculation at or below 2^53 internal microseconds and therefore preserves exact integer milliseconds. HandlerResult::retry, queue registration, and the runtime's final scheduling guard reject a negative or larger duration as invalid_input with operation worker_retry before calling pgmq.set_vt. If the runtime receives a manually constructed invalid retry result, it emits handler_failed, rolls handler business work back to the savepoint, and follows the non-retryable dead-letter path exactly once; it does not issue visibility SQL with that duration or repeatedly re-enter the handler.

Process and connection failure

Before commit, loss of the worker process or database session causes PostgreSQL to roll back:

  • the claim/read update;
  • all handler SQL;
  • delete/archive;
  • retry or DLQ work if it had not committed.

After successful commit, all those database changes exist together.

A connection failure during the commit response can leave the client uncertain whether PostgreSQL committed. This is an observation problem, not a partial commit. Use a unique operation key and reconcile database state after reconnecting.

Suitable work

Good candidates:

  • update an order row and delete its queue item;
  • insert into an inbox table and archive the item;
  • write an outbox record that another regular worker sends externally;
  • maintain database-only projections or counters with bounded execution time.

Poor candidates:

  • HTTP/RPC calls;
  • sending email;
  • object-store or filesystem writes;
  • work in another database;
  • CPU-heavy or long-running computation;
  • code that waits on user input or an unbounded lock.

For external work, use regular mode and idempotency, or use transactional mode only to write a local outbox that a regular worker later delivers.

Why handlers must be short

The transaction keeps a PostgreSQL connection and row lock for the full handler duration. Long transactions:

  • delay vacuum cleanup;
  • increase lock and connection pressure;
  • make shutdown slower;
  • expand the cost of rollback;
  • reduce worker throughput.

Transactional workers claim one message per transaction. Tune concurrency instead of doing long batches inside one transaction.

Pool sizing

At runtime start:

minimum pool_size =
  sum(transactional concurrency across all registered queues) + 2

The runtime rejects a smaller pool. This is a deadlock-avoidance minimum, not a throughput recommendation. Add capacity for producers, administrative calls, regular queue dispatchers, lease renewal, and application transactions that share the client.

The notification listener has its own dedicated connection outside this pool.

Isolation and application constraints

The runtime uses PostgreSQL's transaction behavior through libpq; it does not silently change the caller's server-wide isolation defaults. Business SQL should:

  • bind data through the params vector;
  • use unique constraints for idempotency;
  • impose statement/lock timeouts appropriate to the service;
  • avoid DDL and unbounded scans in the handler;
  • never issue transaction-control SQL or acknowledge/change the current task through WorkerTransaction::execute();
  • handle serialization/deadlock retries as repeat execution;
  • never retain WorkerTransaction& beyond the handler call.

QueryResult exposes rows as strings or SQL NULL. It supports checked row access and checked column lookup by integral index, std::string_view, or C string. Negative/out-of-range numeric indices and unknown columns throw std::out_of_range; a null C-string column name throws std::invalid_argument. Domain conversion remains application code.

Exactly-once wording

It is accurate to say:

The claim, same-database business changes, and acknowledgement commit atomically in one PostgreSQL transaction, provided the trusted handler follows the WorkerTransaction contract.

It is not accurate to say:

The job has exactly-once effects everywhere.

Even a transactional handler can be invoked again after a pre-commit crash, and an external call made inside it cannot be rolled back. Keep the scope of the atomicity statement explicit.