From 4a86e40fd42da551f38fe951367161d728be2a20 Mon Sep 17 00:00:00 2001 From: Xie Zhuoyang <179989892+lite-tx@users.noreply.github.com> Date: Tue, 28 Jul 2026 01:07:16 +0800 Subject: [PATCH] Fix worker start retry and pool return safety --- CHANGELOG.md | 8 +++ CMakeLists.txt | 6 +++ docs/architecture.md | 9 ++++ docs/failure-model.md | 1 + docs/verification.md | 44 ++++++++++++++++ src/internal/libpq.cpp | 1 + src/internal/libpq.hpp | 5 ++ src/internal/worker_test_hooks.hpp | 17 +++++++ src/worker.cpp | 82 ++++++++++++++++++++++++++++++ tests/CMakeLists.txt | 6 ++- tests/integration_worker_tests.cpp | 62 ++++++++++++++++++++++ tests/unit_tests.cpp | 9 ++++ 12 files changed, 249 insertions(+), 1 deletion(-) create mode 100644 src/internal/worker_test_hooks.hpp diff --git a/CHANGELOG.md b/CHANGELOG.md index b8ceb38..baf4bb0 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CMakeLists.txt b/CMakeLists.txt index de2496d..1775f62 100644 --- a/CMakeLists.txt +++ b/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( $ 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) diff --git a/docs/architecture.md b/docs/architecture.md index 7f0975a..983c111 100644 --- a/docs/architecture.md +++ b/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: diff --git a/docs/failure-model.md b/docs/failure-model.md index 48ceac4..2672cde 100644 --- a/docs/failure-model.md +++ b/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 | diff --git a/docs/verification.md b/docs/verification.md index a3b1545..70fda2c 100644 --- a/docs/verification.md +++ b/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. diff --git a/src/internal/libpq.cpp b/src/internal/libpq.cpp index a2d8c09..430979d 100644 --- a/src/internal/libpq.cpp +++ b/src/internal/libpq.cpp @@ -291,6 +291,7 @@ ConnectionPool::ConnectionPool(ClientOptions options) {}, "create_client"}; } + idle_.reserve(options_.pool_size); } ConnectionPool::Lease ConnectionPool::acquire() { diff --git a/src/internal/libpq.hpp b/src/internal/libpq.hpp index fd0abc9..1c79f75 100644 --- a/src/internal/libpq.hpp +++ b/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) noexcept; diff --git a/src/internal/worker_test_hooks.hpp b/src/internal/worker_test_hooks.hpp new file mode 100644 index 0000000..c0c7f6a --- /dev/null +++ b/src/internal/worker_test_hooks.hpp @@ -0,0 +1,17 @@ +#pragma once + +#ifdef PGMQ_CPP_ENABLE_TEST_HOOKS + +#include + +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 diff --git a/src/worker.cpp b/src/worker.cpp index 73eda7b..42ad85b 100644 --- a/src/worker.cpp +++ b/src/worker.cpp @@ -8,18 +8,21 @@ #include #include #include +#include #include #include #include #include #include #include +#include #include #include #include #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 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::numeric_limits::max()); + worker_thread_launches_before_failure.store(static_cast( + 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> fresh_queues; + fresh_queues.reserve(queues.size()); + for (const auto& state : queues) { + fresh_queues.push_back(std::make_unique(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 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 started{}; std::atomic stopping{}; std::atomic shutdown_finished{}; + bool reset_before_start{}; ShutdownResult last_shutdown{}; }; diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt index cce50c6..305619f 100644 --- a/tests/CMakeLists.txt +++ b/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) diff --git a/tests/integration_worker_tests.cpp b/tests/integration_worker_tests.cpp index d28a283..ac917a9 100644 --- a/tests/integration_worker_tests.cpp +++ b/tests/integration_worker_tests.cpp @@ -9,9 +9,11 @@ #include #include #include +#include #include #include +#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 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(); } diff --git a/tests/unit_tests.cpp b/tests/unit_tests.cpp index 3b4d732..4a02a97 100644 --- a/tests/unit_tests.cpp +++ b/tests/unit_tests.cpp @@ -13,6 +13,7 @@ #include #include "internal/epoch.hpp" +#include "internal/libpq.hpp" #include "internal/metrics_access.hpp" #include "internal/worker_retry.hpp" #include "pgmq/client.hpp" @@ -349,6 +350,14 @@ TEST_CASE("worker lifecycle validates empty start and pre-start shutdown") { CHECK_EQ(shutdown.remaining_in_flight, std::size_t{0}); } +TEST_CASE("connection pool preallocates every nonthrowing return slot") { + pgmq::ClientOptions options{"postgresql://unused", 7}; + const auto pool = + std::make_shared(std::move(options)); + CHECK_EQ(pool->capacity(), std::size_t{7}); + CHECK(pool->idle_storage_capacity_for_testing() >= pool->capacity()); +} + TEST_CASE("metrics registry exports a valid empty OpenMetrics document") { pgmq::MetricsRegistry metrics; const auto snapshot = metrics.snapshot();