Public API design from real usage

The public API was derived from concrete producer and consumer call sites before implementation. The snippets below match the public headers. Full programs are under examples/.

Client construction and capabilities

auto client = std::make_shared<pgmq::Client>(pgmq::ClientOptions{
    .connection_string = database_url,
    .pool_size = 8,
    .connect_timeout = std::chrono::seconds{5},
    .application_name = "thumbnail-service",
    .statement_timeout = std::chrono::seconds{30},
    .pool_acquire_timeout = std::chrono::seconds{30},
});

const auto capabilities = client->capabilities();
if (!capabilities.notifications) {
  log_warning("PGMQ insert notifications unavailable; worker will poll");
}

Client is movable but not copyable. It owns a lazy bounded pool and may be shared safely between application components. A checked QueueName is used where an operation can derive a physical queue identifier. Exhausted-pool waits are bounded by pool_acquire_timeout; expiry throws a timeout-category pgmq::Error.

Producer and regular worker

const pgmq::QueueName queue{"thumbnail_jobs"};
client->create_queue(queue);

const auto message_id = client->send(
    queue,
    pgmq::Json{{"image_id", 42}},
    std::optional<pgmq::Json>{
        pgmq::Json{{"trace_id", "abc"}}},
    std::chrono::seconds{5});
(void)message_id;

pgmq::QueueWorkerConfig config{queue};
config.concurrency = 4;
config.claim_batch_size = 8;
config.max_in_flight = 16;
config.visibility_timeout = std::chrono::seconds{30};
config.lease_renewal_interval = std::chrono::seconds{10};
config.handler =
    [](const pgmq::Task& task, pgmq::HandlerContext& context) {
      render_thumbnail(task.message.body, context.stop_token);
      return pgmq::HandlerResult::success_delete();
    };

pgmq::WorkerRuntime runtime{client};
runtime.add_queue(std::move(config));
runtime.start();
const auto stopped = runtime.shutdown(std::chrono::seconds{30});

QueueWorkerConfig has an explicit QueueName constructor and is not a designated aggregate. max_in_flight bounds claimed pending plus executing tasks.

Read variants

pgmq::ReadOptions options;
options.visibility_timeout = std::chrono::seconds{30};
options.quantity = 10;
options.strategy = pgmq::ReadStrategy::grouped_heads;
options.max_poll_time = std::chrono::seconds{5};
options.poll_interval = std::chrono::milliseconds{100};

const auto messages = client->read(pgmq::QueueName{"ordered_jobs"}, options);

For standard reads, options.conditional carries the experimental upstream JSON predicate. It cannot be combined with grouped strategies.

Each returned Message carries both visible_at and database_observed_at. The latter is PostgreSQL clock_timestamp() projected by the same query, so consumers can interpret the visibility timestamp in the database clock domain instead of subtracting the application host's wall clock.

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();

No SDK method commits a caller-owned transaction. If an active Transaction leaves scope, its destructor attempts rollback. A successful explicit commit() or rollback() returns the pinned connection to the pool immediately, so later client calls in the same C++ scope do not wait for the transaction object's destructor.

Transactional worker

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&) {
      const auto inserted = transaction.execute(
          "INSERT INTO fulfilled_orders(order_id) VALUES (\$1::bigint) "
          "ON CONFLICT DO NOTHING",
          {std::to_string(
              task.message.body.at("order_id").get<long long>())});
      (void)inserted;
      return pgmq::HandlerResult::success_delete();
    };

The claim, business statement, acknowledgement, and commit use one PostgreSQL transaction. This mode is intentionally for short database-only work. The handler receives a non-owning WorkerTransaction view with only execute/send methods; direct commit, rollback, delete, archive, and visibility methods are deliberately absent to prevent accidental misuse. execute() still accepts trusted arbitrary SQL. The handler contract forbids transaction-control statements and SQL that acknowledges or changes the current task; this API is not a SQL security sandbox.

Error handling

try {
  const auto metrics = client->metrics(queue);
  consume(metrics);
} catch (const pgmq::Error& error) {
  log_database_error(pgmq::to_string(error.category()),
                     error.sqlstate(),
                     error.operation(),
                     error.detail());
  if (error.retryable()) {
    schedule_controlled_retry();
  }
}

SDK validation and operational failures use pgmq::Error; PostgreSQL SQLSTATE and operation context are retained when available. Invalid QueryResult row/column access uses std::out_of_range, and application code that directly uses nlohmann/json retains that library's exception behavior.