added health checker

Author: g2px1

CMakeLists.txt

@@ -38,7 +38,9 @@ file(GLOB_RECURSE UPQ_SOURCES CONFIGURE_DEPENDS
 )
 
 add_library(upq ${UPQ_SOURCES} ${UPQ_HEADERS}
-        include/upq/PgNotificationMultiplexer.h)
+        include/upq/PgNotificationMultiplexer.h
+        include/upq/PgHealthChecker.h
+        src/upq/PgHealthChecker.cpp)
 add_library(usub::upq ALIAS upq)
 
 target_include_directories(upq

docs/pool.md

@@ -23,31 +23,94 @@ It guarantees **no blocking**, **no busy-waiting**, and **safe use from multiple
 
 ## Initialization
 
+`PgPool` now optionally supports a background health checker for periodic connection validation.
+
 ```cpp
 usub::pg::PgPool::init_global(
     host, port, user, db, password,
-    max_pool_size, queue_capacity
+    max_pool_size, queue_capacity,
+    usub::pg::PgHealthConfig{
+        .enabled = true,
+        .interval_ms = 3000
+    }
 );
 ```
 
-Example:
+When `.enabled = true`, the pool automatically spawns an internal coroutine that periodically runs lightweight
+`SELECT 1;` probes on temporary connections to ensure the database is reachable.
+
+If `.enabled = false` (default), the health checker is disabled.
+
+You can access runtime statistics at any time via:
 
 ```cpp
-usub::pg::PgPool::init_global(
-    "localhost",
-    "5432",
-    "postgres",
-    "mydb",
-    "password",
-    32,  // max_pool_size
-    64   // queue_capacity
-);
+auto& hc = usub::pg::PgPool::instance().health_checker();
+auto& stats = hc.stats();
+
+std::cout
+    << "checks=" << stats.iterations.load()
+    << " ok=" << stats.ok_checks.load()
+    << " failed=" << stats.failed_checks.load()
+    << std::endl;
+```
+
+---
+
+## Health Checker
+
+`PgHealthChecker` is a lightweight background monitor built into the pool.
+It ensures early detection of connection failures without impacting performance.
+
+| Field           | Type               | Description                                     |
+|-----------------|--------------------|-------------------------------------------------|
+| `enabled`       | `bool`             | Whether the health checker coroutine should run |
+| `interval_ms`   | `uint64_t`         | Delay between health probes                     |
+| `iterations`    | `atomic<uint64_t>` | Total number of health iterations               |
+| `ok_checks`     | `atomic<uint64_t>` | Successful `SELECT 1` responses                 |
+| `failed_checks` | `atomic<uint64_t>` | Failed or unreachable checks                    |
+
+### Behavior
 
-auto& pool = usub::pg::PgPool::instance();
+* When enabled, a coroutine runs forever in the background.
+* Every interval:
+
+    1. Acquires a fresh connection from the pool.
+    2. Executes `SELECT 1;` using non-blocking I/O.
+    3. Updates internal counters.
+    4. Returns the connection back to the pool.
+* Failures are silent — they do not throw or disrupt normal queries.
+* Intended for observability and proactive reconnection handling.
+
+### Example
+
+```cpp
+task::Awaitable<void> print_pg_health()
+{
+    auto& pool = usub::pg::PgPool::instance();
+    auto& hc = pool.health_checker();
+
+    for (;;)
+    {
+        auto& s = hc.stats();
+        std::cout
+            << "[Health] iter=" << s.iterations.load()
+            << " ok=" << s.ok_checks.load()
+            << " fail=" << s.failed_checks.load()
+            << std::endl;
+
+        co_await usub::uvent::system::this_coroutine::sleep_for(
+            std::chrono::seconds(5)
+        );
+    }
+}
 ```
 
-This must be called **once** during startup.
-`instance()` returns the global singleton thereafter.
+### Notes
+
+* Uses existing event loop and coroutine scheduler (`uvent`).
+* Generates negligible load on the database (`SELECT 1`).
+* Default interval: **600 000 ms**.
+* Can be disabled completely if unnecessary.
 
 ---
 
@@ -201,10 +264,6 @@ This provides natural backpressure and keeps memory footprint predictable.
 
 ---
 
-## Error transparency (since v1.0.1)
-
-All queries and internal operations now report structured diagnostic codes.
-
 ### Example: Connection loss
 
 ```cpp
@@ -244,25 +303,26 @@ if (!res.ok && res.code == PgErrorCode::ServerError)
 
 ## Best practices
 
-- ✅ Initialize pool early (before spawning coroutines).
-- ✅ Use `query_awaitable()` for short-lived queries.
-- ✅ Use manual `acquire_connection()` for long-running listeners or transactions.
-- ✅ Check `QueryResult.code` and `ok` for every query.
-- ✅ Release connections when finished — don’t hold them across awaits.
-- ❌ Don’t share one connection across threads.
-- ❌ Don’t assume immediate connection creation — first queries may take longer.
+* ✅ Initialize pool early (before spawning coroutines).
+* ✅ Use `query_awaitable()` for short-lived queries.
+* ✅ Use manual `acquire_connection()` for long-running listeners or transactions.
+* ✅ Check `QueryResult.code` and `ok` for every query.
+* ✅ Release connections when finished — don’t hold them across awaits.
+* ❌ Don’t share one connection across threads.
+* ❌ Don’t assume immediate connection creation — first queries may take longer.
 
 ---
 
 ## Summary
 
-| Feature      | Description                               |
-|--------------|-------------------------------------------|
-| Asynchronous | Non-blocking coroutine API                |
-| Safe         | Lock-free concurrent access               |
-| Lazy         | Creates connections only as needed        |
-| Bounded      | Max live connections controlled by config |
-| Transparent  | Every error is classified and traceable   |
+| Feature        | Description                               |
+|----------------|-------------------------------------------|
+| Asynchronous   | Non-blocking coroutine API                |
+| Safe           | Lock-free concurrent access               |
+| Lazy           | Creates connections only as needed        |
+| Bounded        | Max live connections controlled by config |
+| Transparent    | Every error is classified and traceable   |
+| Health Checker | Background connection heartbeat loop      |
 
 `PgPool` is the backbone of the upq runtime — it provides scalable, coroutine-friendly database access with
 deterministic failure semantics.

examples/main.cpp

@@ -204,7 +204,11 @@ int main()
         "postgres", // db
         "password", // password
         /*max_pool_size*/ 32,
-        /*queue_capacity*/ 64
+        /*queue_capacity*/ 64,
+        usub::pg::PgPoolHealthConfig{
+            .enabled = true,
+            .interval_ms = 3000
+        }
     );
 
     uvent.for_each_thread([&](int threadIndex, thread::ThreadLocalStorage* tls)

include/upq/PgHealthChecker.h

@@ -0,0 +1,43 @@
+#ifndef PGHEALTHCHECKER_H
+#define PGHEALTHCHECKER_H
+
+#include <cstdint>
+#include <atomic>
+#include <memory>
+#include <chrono>
+
+#include "uvent/Uvent.h"
+#include "upq/PgPool.h"
+#include "upq/PgConnection.h"
+#include "upq/PgTypes.h"
+
+namespace usub::pg
+{
+    struct PgHealthStats
+    {
+        std::atomic<uint64_t> iterations{0};
+        std::atomic<uint64_t> ok_checks{0};
+        std::atomic<uint64_t> failed_checks{0};
+    };
+
+    class PgHealthChecker
+    {
+    public:
+        PgHealthChecker(PgPool& pool, PgPoolHealthConfig cfg = {});
+
+        PgPoolHealthConfig& config();
+        const PgPoolHealthConfig& config() const;
+
+        PgHealthStats& stats();
+        const PgHealthStats& stats() const;
+
+        usub::uvent::task::Awaitable<void> run();
+
+    private:
+        PgPool& pool_;
+        PgPoolHealthConfig cfg_;
+        PgHealthStats stats_;
+    };
+} // namespace usub::pg
+
+#endif // PGHEALTHCHECKER_H

include/upq/PgPool.h

@@ -16,6 +16,14 @@
 
 namespace usub::pg
 {
+    struct PgPoolHealthConfig
+    {
+        bool enabled{false};
+        uint64_t interval_ms{600000};
+    };
+
+    class PgHealthChecker;
+
     class PgPool
     {
     public:
@@ -25,15 +33,17 @@ namespace usub::pg
                std::string db,
                std::string password,
                size_t max_pool_size = 32,
-               size_t queue_capacity = 64);
+               size_t queue_capacity = 64,
+               PgPoolHealthConfig health_cfg = {});
 
         static void init_global(const std::string& host,
                                 const std::string& port,
                                 const std::string& user,
                                 const std::string& db,
                                 const std::string& password,
                                 size_t max_pool_size = 32,
-                                size_t queue_capacity = 64);
+                                size_t queue_capacity = 64,
+                                PgPoolHealthConfig health_cfg = {});
 
         static PgPool& instance();
 
@@ -52,30 +62,26 @@ namespace usub::pg
         usub::uvent::task::Awaitable<QueryResult>
         query_awaitable(const std::string& sql, Args&&... args);
 
-        inline std::string host()
-        {
-            return this->host_;
-        }
+        inline std::string host() { return this->host_; }
+        inline std::string port() { return this->port_; }
+        inline std::string user() { return this->user_; }
+        inline std::string db() { return this->db_; }
+        inline std::string password() { return this->password_; }
 
-        inline std::string port()
-        {
-            return this->port_;
-        }
+        inline PgPoolHealthConfig health_cfg() const { return this->health_cfg_; }
 
-        inline std::string user()
-        {
-            return this->user_;
-        }
+        PgHealthChecker& health_checker();
 
-        inline std::string db()
-        {
-            return this->db_;
-        }
+        void mark_dead(std::shared_ptr<PgConnectionLibpq> const& conn);
 
-        inline std::string password()
+        struct HealthStats
         {
-            return this->password_;
-        }
+            std::atomic<uint64_t> checked{0};
+            std::atomic<uint64_t> alive{0};
+            std::atomic<uint64_t> reconnected{0};
+        };
+
+        inline HealthStats& health_stats() { return stats_; }
 
     private:
         std::string host_;
@@ -89,6 +95,12 @@ namespace usub::pg
         size_t max_pool_;
         std::atomic<size_t> live_count_;
 
+        PgPoolHealthConfig health_cfg_;
+
+        HealthStats stats_;
+
+        std::unique_ptr<PgHealthChecker> health_checker_;
+
         static std::unique_ptr<PgPool> instance_;
     };
 
@@ -138,7 +150,6 @@ namespace usub::pg
         release_connection(conn);
         co_return qr;
     }
-
 } // namespace usub::pg
 
 #endif // PGPOOL_H

src/upq/PgHealthChecker.cpp

@@ -0,0 +1,64 @@
+#include "upq/PgHealthChecker.h"
+#include "uvent/system/SystemContext.h"
+
+namespace usub::pg
+{
+    PgHealthChecker::PgHealthChecker(PgPool& pool, PgPoolHealthConfig cfg)
+        : pool_(pool), cfg_(std::move(cfg))
+    {
+    }
+
+    PgPoolHealthConfig& PgHealthChecker::config() { return this->cfg_; }
+    const PgPoolHealthConfig& PgHealthChecker::config() const { return this->cfg_; }
+
+    PgHealthStats& PgHealthChecker::stats() { return this->stats_; }
+    const PgHealthStats& PgHealthChecker::stats() const { return this->stats_; }
+
+    usub::uvent::task::Awaitable<void> PgHealthChecker::run()
+    {
+        for (;;)
+        {
+            PgPoolHealthConfig cur_cfg = this->cfg_;
+
+            if (!cur_cfg.enabled)
+            {
+                co_await usub::uvent::system::this_coroutine::sleep_for(
+                    std::chrono::milliseconds(1000)
+                );
+                continue;
+            }
+
+            uint64_t interval = cur_cfg.interval_ms;
+            if (interval == 0)
+                interval = 1000;
+
+            this->stats_.iterations.fetch_add(1, std::memory_order_relaxed);
+
+            auto conn = co_await this->pool_.acquire_connection();
+
+            bool alive = false;
+            if (conn && conn->connected())
+            {
+                QueryResult ping = co_await conn->exec_simple_query_nonblocking("SELECT 1;");
+                if (ping.ok)
+                {
+                    alive = true;
+                    this->stats_.ok_checks.fetch_add(1, std::memory_order_relaxed);
+                }
+            }
+
+            if (!alive)
+            {
+                this->stats_.failed_checks.fetch_add(1, std::memory_order_relaxed);
+            }
+
+            this->pool_.release_connection(conn);
+
+            co_await usub::uvent::system::this_coroutine::sleep_for(
+                std::chrono::milliseconds(interval)
+            );
+        }
+
+        co_return;
+    }
+} // namespace usub::pg