Fix worker start retry and pool return safety

CHANGELOG.md

@@ -6,6 +6,14 @@ follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ## Unreleased
 
+### Fixed
+
+- Rebuild per-run worker state before retrying `WorkerRuntime::start()` after a
+  partial thread-launch failure, so accepted retries cannot inherit closed
+  local work queues from the failed attempt.
+- Preallocate every idle connection-pool return slot during pool construction,
+  keeping healthy lease return non-allocating inside its `noexcept` path.
+
 ## [1.0.0] - 2026-07-26
 
 This is the prepared v1.0.0 release entry, not a claim that publication had

CMakeLists.txt

@@ -146,6 +146,9 @@ if(WIN32)
   target_link_libraries(pgmq_client PRIVATE ws2_32)
   target_compile_definitions(pgmq_client PRIVATE NOMINMAX WIN32_LEAN_AND_MEAN)
 endif()
+if(BUILD_TESTING)
+  target_compile_definitions(pgmq_client PRIVATE PGMQ_CPP_ENABLE_TEST_HOOKS)
+endif()
 pgmq_cpp_set_warnings(pgmq_client)
 pgmq_cpp_enable_sanitizers(pgmq_client)
 
@@ -165,6 +168,9 @@ target_include_directories(
          $<INSTALL_INTERFACE:${CMAKE_INSTALL_INCLUDEDIR}>
   PRIVATE ${CMAKE_CURRENT_SOURCE_DIR}/src)
 target_link_libraries(pgmq_worker PUBLIC pgmq_client Threads::Threads)
+if(BUILD_TESTING)
+  target_compile_definitions(pgmq_worker PRIVATE PGMQ_CPP_ENABLE_TEST_HOOKS)
+endif()
 pgmq_cpp_set_warnings(pgmq_worker)
 pgmq_cpp_enable_sanitizers(pgmq_worker)
 

docs/architecture.md

@@ -222,6 +222,15 @@ polling active.
 
 See [ADR 0003](adr/0003-notifications-are-advisory.md).
 
+## 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:

docs/failure-model.md

@@ -46,6 +46,7 @@ behavior is not a substitute for a run.
 | DB disconnect during one autocommit statement | libpq error | server outcome can be unknown; SDK does not replay blindly | Producer write may have committed | Reconcile/idempotently retry |
 | DB disconnect inside caller transaction | error; reconnect disabled | connection loss causes server rollback unless commit already completed | Commit outcome can be unknown | Reconcile durable state |
 | Connection pool exhausted beyond `pool_acquire_timeout` | timeout-category `pgmq::Error` | operation was not sent | None from that operation | Increase capacity, shorten leases, or reduce contention |
+| Runtime thread creation fails during `WorkerRuntime::start()` | thread constructor exception | already launched runtime threads are stopped and joined; a later `start()` rebuilds all per-run state | Work claimed during a partially completed multi-queue start can become visible again after its lease | Log the failure, relieve the process resource limit, then retry `start()` |
 | Invalid `QueryResult` row/column lookup | `std::out_of_range`, or `std::invalid_argument` for a null C-string name | query has already completed; result access fails without another database operation | None | Correct the index/name or check result shape |
 | Notification lost/coalesced/throttled | usually not directly knowable | polling later finds row | None from notification alone | Keep polling enabled |
 | Listener disconnect | listener reconnect counter/event | `LISTEN` and configuration restored; race covered by poll | None from notification alone | Monitor reconnect rate |

docs/verification.md

@@ -653,3 +653,47 @@ analysis, Doxygen-probe, dependency-bootstrap, and validation-image build
 outputs were removed. The temporary `pgmq-cpp-validation:ubuntu24` image and
 drive substitution were also removed. Earlier build/cache directories that
 predated this goal were left untouched.
+
+## Post-release-candidate worker-start and pool-return follow-up
+
+On 2026-07-27, a source review identified two exceptional resource-failure
+paths that the release-candidate suite did not distinguish.
+
+For partial worker startup, a build-test-only one-shot thread-launch failpoint
+was added. With the old lifecycle logic still present, failure after one
+successful launch followed by a second `start()` produced:
+
+```text
+[FAIL] worker start retry rebuilds per-run state after launch failure:
+retried runtime did not dispatch a task
+0/1 tests passed
+```
+
+The repair rebuilds all per-run queue and thread state before an accepted
+retry. The same integration test then passed at two different partial-launch
+indices (after one and after three successful launches). The test starts the
+same runtime again, sends a real PGMQ message, requires exactly one handler
+call, observes an empty queue, and requires a clean shutdown.
+
+The connection-pool follow-up reserves `pool_size` idle slots during
+construction. Its unit test checks that the storage capacity covers every
+possible healthy connection return, so `release()` cannot allocate while
+executing through its `noexcept` lease-destruction path.
+
+Fresh Windows MSVC 19.41 Release evidence used both static
+`build/next-step-pgmq` and shared `build/next-step-pgmq-shared` builds with
+`/W4 /WX`. Each complete CTest run passed **8/8**, including the installed
+consumer, real PostgreSQL 18 SDK and Worker suites, deterministic crash
+recovery, multiprocess behavior, and fixture cleanup. The shared build also
+proved that the test-only launch seam links correctly across the DLL boundary.
+
+A Linux GCC 13.4 Debug build then ran with ASan and UBSan, leak detection,
+strict string checks, stack traces, and halt-on-error enabled. The first
+attempt correctly failed at compile time because the new internal pool test
+did not inherit the private PostgreSQL header directory on Linux. After the
+test target declared that include dependency, the sanitizer build passed, the
+unit executable passed **14/14**, and the real-PostgreSQL filtered
+partial-start retry test passed **1/1**. No sanitizer diagnostic was emitted.
+
+These results are local follow-up evidence. They do not claim that a new tag,
+GitHub Release, or hosted workflow was created or run.

src/internal/libpq.cpp

@@ -291,6 +291,7 @@ ConnectionPool::ConnectionPool(ClientOptions options)
         {},
         "create_client"};
   }
+  idle_.reserve(options_.pool_size);
 }
 
 ConnectionPool::Lease ConnectionPool::acquire() {

src/internal/libpq.hpp

@@ -103,6 +103,11 @@ class ConnectionPool final
   [[nodiscard]] Lease acquire();
   [[nodiscard]] std::size_t capacity() const noexcept { return options_.pool_size; }
   [[nodiscard]] const ClientOptions& options() const noexcept { return options_; }
+#ifdef PGMQ_CPP_ENABLE_TEST_HOOKS
+  [[nodiscard]] std::size_t idle_storage_capacity_for_testing() const noexcept {
+    return idle_.capacity();
+  }
+#endif
 
  private:
   void release(std::unique_ptr<Connection> connection) noexcept;

src/internal/worker_test_hooks.hpp

@@ -0,0 +1,17 @@
+#pragma once
+
+#ifdef PGMQ_CPP_ENABLE_TEST_HOOKS
+
+#include <cstddef>
+
+namespace pgmq::detail {
+
+void fail_worker_thread_launch_after_for_testing(
+    std::size_t successful_launches) noexcept;
+void clear_worker_thread_launch_failure_for_testing() noexcept;
+[[nodiscard]] bool
+consume_worker_thread_launch_failure_for_testing() noexcept;
+
+}  // namespace pgmq::detail
+
+#endif

src/worker.cpp

@@ -8,18 +8,21 @@
 #include <cstddef>
 #include <deque>
 #include <exception>
+#include <limits>
 #include <map>
 #include <memory>
 #include <mutex>
 #include <random>
 #include <set>
 #include <string>
+#include <system_error>
 #include <thread>
 #include <unordered_map>
 #include <utility>
 
 #include "internal/metrics_access.hpp"
 #include "internal/sql.hpp"
+#include "internal/worker_test_hooks.hpp"
 #include "internal/worker_retry.hpp"
 #include "pgmq/error.hpp"
 
@@ -262,6 +265,47 @@ class AtomicFlagReset final {
 
 }  // namespace
 
+#ifdef PGMQ_CPP_ENABLE_TEST_HOOKS
+
+namespace detail {
+namespace {
+
+std::atomic<std::ptrdiff_t> worker_thread_launches_before_failure{-1};
+
+}  // namespace
+
+void fail_worker_thread_launch_after_for_testing(
+    std::size_t successful_launches) noexcept {
+  const auto maximum =
+      static_cast<std::size_t>(std::numeric_limits<std::ptrdiff_t>::max());
+  worker_thread_launches_before_failure.store(static_cast<std::ptrdiff_t>(
+      std::min(successful_launches, maximum)));
+}
+
+void clear_worker_thread_launch_failure_for_testing() noexcept {
+  worker_thread_launches_before_failure.store(-1);
+}
+
+bool consume_worker_thread_launch_failure_for_testing() noexcept {
+  auto remaining = worker_thread_launches_before_failure.load();
+  while (remaining >= 0) {
+    if (remaining == 0) {
+      if (worker_thread_launches_before_failure.compare_exchange_weak(
+              remaining, -1)) {
+        return true;
+      }
+    } else if (worker_thread_launches_before_failure.compare_exchange_weak(
+                   remaining, remaining - 1)) {
+      return false;
+    }
+  }
+  return false;
+}
+
+}  // namespace detail
+
+#endif
+
 HandlerResult HandlerResult::success_delete() {
   return {Disposition::success_delete, {}, std::nullopt};
 }
@@ -645,6 +689,32 @@ struct WorkerRuntime::Impl {
     thread_condition.notify_all();
   }
 
+  void rebuild_per_run_state_after_failed_start() {
+    std::vector<std::unique_ptr<QueueState>> fresh_queues;
+    fresh_queues.reserve(queues.size());
+    for (const auto& state : queues) {
+      fresh_queues.push_back(std::make_unique<QueueState>(state->config));
+    }
+
+    {
+      std::scoped_lock lock{claims_mutex};
+      claims.clear();
+    }
+    {
+      std::scoped_lock lock{transactional_stops_mutex};
+      transactional_stops.clear();
+    }
+    lease_thread = {};
+    notification_thread = {};
+    queues.swap(fresh_queues);
+    total_in_flight.store(0);
+    active_threads.store(0);
+    stopping.store(false);
+    shutdown_finished.store(false);
+    last_shutdown = {};
+    reset_before_start = false;
+  }
+
   void change_in_flight(QueueState& state, std::ptrdiff_t delta) {
     std::size_t updated{};
     if (delta > 0) {
@@ -1488,6 +1558,13 @@ struct WorkerRuntime::Impl {
   template <typename Function>
   std::jthread launch_thread(const QueueName* queue, std::string role,
                              Function function) {
+#ifdef PGMQ_CPP_ENABLE_TEST_HOOKS
+    if (detail::consume_worker_thread_launch_failure_for_testing()) {
+      throw std::system_error{
+          std::make_error_code(std::errc::resource_unavailable_try_again),
+          "injected worker thread launch failure"};
+    }
+#endif
     active_threads.fetch_add(1);
     try {
       return std::jthread(
@@ -1522,6 +1599,9 @@ struct WorkerRuntime::Impl {
                   {},
                   "worker_start"};
     }
+    if (reset_before_start) {
+      rebuild_per_run_state_after_failed_start();
+    }
     if (queues.empty()) {
       throw Error{ErrorCategory::invalid_input,
                   "WorkerRuntime has no queues",
@@ -1625,6 +1705,7 @@ struct WorkerRuntime::Impl {
     } catch (...) {
       request_threads_stop(true);
       join_completed_threads();
+      reset_before_start = true;
       started.store(false);
       throw;
     }
@@ -1782,6 +1863,7 @@ struct WorkerRuntime::Impl {
   std::atomic<bool> started{};
   std::atomic<bool> stopping{};
   std::atomic<bool> shutdown_finished{};
+  bool reset_before_start{};
   ShutdownResult last_shutdown{};
 };
 

tests/CMakeLists.txt

@@ -30,7 +30,9 @@ endfunction()
 
 add_executable(pgmq_unit_tests unit_tests.cpp)
 target_link_libraries(pgmq_unit_tests PRIVATE pgmq::worker)
-target_include_directories(pgmq_unit_tests PRIVATE "${PROJECT_SOURCE_DIR}/src")
+target_include_directories(
+  pgmq_unit_tests PRIVATE "${PROJECT_SOURCE_DIR}/src" ${PostgreSQL_INCLUDE_DIRS})
+target_compile_definitions(pgmq_unit_tests PRIVATE PGMQ_CPP_ENABLE_TEST_HOOKS)
 pgmq_cpp_test_target(pgmq_unit_tests)
 add_test(NAME pgmq_unit COMMAND pgmq_unit_tests)
 set_tests_properties(pgmq_unit PROPERTIES LABELS "unit" TIMEOUT 30)
@@ -43,6 +45,8 @@ add_executable(pgmq_integration_worker_tests integration_worker_tests.cpp)
 target_link_libraries(pgmq_integration_worker_tests PRIVATE pgmq::worker)
 target_include_directories(
   pgmq_integration_worker_tests PRIVATE "${PROJECT_SOURCE_DIR}/src")
+target_compile_definitions(
+  pgmq_integration_worker_tests PRIVATE PGMQ_CPP_ENABLE_TEST_HOOKS)
 pgmq_cpp_test_target(pgmq_integration_worker_tests)
 
 add_executable(pgmq_crash_worker crash_worker.cpp)

tests/integration_worker_tests.cpp

@@ -9,9 +9,11 @@
 #include <set>
 #include <stdexcept>
 #include <string>
+#include <system_error>
 #include <thread>
 #include <vector>
 
+#include "internal/worker_test_hooks.hpp"
 #include "internal/worker_retry.hpp"
 #include "pgmq/client.hpp"
 #include "pgmq/worker.hpp"
@@ -62,6 +64,22 @@ class FutureWallClock final : public pgmq::Clock {
   }
 };
 
+class WorkerThreadLaunchFailure final {
+ public:
+  explicit WorkerThreadLaunchFailure(std::size_t successful_launches) noexcept {
+    pgmq::detail::fail_worker_thread_launch_after_for_testing(
+        successful_launches);
+  }
+
+  ~WorkerThreadLaunchFailure() {
+    pgmq::detail::clear_worker_thread_launch_failure_for_testing();
+  }
+
+  WorkerThreadLaunchFailure(const WorkerThreadLaunchFailure&) = delete;
+  WorkerThreadLaunchFailure& operator=(const WorkerThreadLaunchFailure&) =
+      delete;
+};
+
 }  // namespace
 
 TEST_CASE("regular workers process a bounded batch concurrently once") {
@@ -1035,4 +1053,48 @@ TEST_CASE(
   CHECK_EQ(client->pop(queue, 1).size(), std::size_t{1});
 }
 
+TEST_CASE("worker start retry rebuilds per-run state after launch failure") {
+  for (const std::size_t successful_launches : {std::size_t{1},
+                                                std::size_t{3}}) {
+    auto client = pgmq_test::make_client("worker-start-retry", 8);
+    const auto queue = pgmq_test::unique_queue(
+        "worker_start_retry_" + std::to_string(successful_launches));
+    const auto dlq = dlq_for(queue);
+    client->create_queue(queue);
+    pgmq_test::QueueGuard queue_guard{client, queue};
+    pgmq_test::QueueGuard dlq_guard{client, dlq};
+
+    std::atomic<std::size_t> calls{};
+    pgmq::WorkerRuntime runtime{client};
+    pgmq::QueueWorkerConfig config{queue};
+    config.enable_notifications = false;
+    config.polling_interval = 25ms;
+    config.visibility_timeout = 2s;
+    config.lease_renewal_interval = 1s;
+    config.handler = [&](const pgmq::Task&, pgmq::HandlerContext&) {
+      calls.fetch_add(1);
+      return pgmq::HandlerResult::success_delete();
+    };
+    runtime.add_queue(std::move(config));
+
+    {
+      const WorkerThreadLaunchFailure failure{successful_launches};
+      CHECK_THROWS_AS(runtime.start(), std::system_error);
+    }
+    CHECK(!runtime.running());
+
+    runtime.start();
+    (void)client->send(queue, {{"attempt", successful_launches}});
+    CHECK_MSG(pgmq_test::eventually(3s, [&] { return calls.load() == 1; }),
+              "retried runtime did not dispatch a task");
+    CHECK_MSG(
+        pgmq_test::eventually(
+            3s, [&] { return client->metrics(queue).queue_length == 0; }),
+        "retried runtime did not acknowledge its task");
+    const auto shutdown = runtime.shutdown(2s);
+    CHECK(shutdown.clean);
+    CHECK_EQ(calls.load(), std::size_t{1});
+  }
+}
+
 int main() { return pgmq_test::run_all(); }