Commit 828fe132 authored by jan.koester's avatar jan.koester
Browse files

test

parent ac5bbd65

src/backends/cluster.cpp

+13 −6
Changes for src/backends/cluster.cpp: 13 added lines, 6 removed lines.
Original line number Diff line number Diff line
@@ -92,17 +92,26 @@ namespace authdb { 
                  << " buf_size=" << _Buffer.size() << " dirty=" << _Dirty.load() << "\n");



        // Only fetch from peers every 30 seconds — auth data changes rarely

        std::vector<uint8_t> peer_data;

        // Check BEFORE acquiring the mutex to avoid contention with session ops

        auto now = std::chrono::steady_clock::now();

        bool do_peer_fetch = (now - _LastPeerFetch) >= std::chrono::seconds(30);



        if (do_peer_fetch) {

        // Force immediate fetch if cluster recovered (new nodes available)

        uint64_t cur_epoch = g_Cluster->getRecoveryEpoch();

        if (cur_epoch != _lastRecoveryEpoch) {

            do_peer_fetch = true;

            _lastRecoveryEpoch = cur_epoch;

            std::cerr << "[CLUSTER-BE] recovery detected, forcing refresh for domain=" << _Domain << std::endl;

        }



        if (!do_peer_fetch)

            return;  // Nothing to do — don't touch client_mutex_



        std::vector<uint8_t> peer_data;

        auto &client = g_Cluster->getClient();

        if (client) {

            try {

                auto &cmtx = g_Cluster->getClientMutex();

                    // Brief wait (200ms) to let short operations finish, but

                    // not long enough to starve shared readers (5s timeout).

                std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::milliseconds(200));

                if (lock.owns_lock()) {

                    peer_data = client->retrieve(dgid);
@@ -111,14 +120,12 @@ namespace authdb { 
            } catch (const std::exception &e) {

                std::cerr << "ClusterBackend::fetchFromCluster: retrieve failed: "

                          << e.what() << std::endl;

                    // "not enough blocks" means data exists but too few peers are reachable

                if (std::string(e.what()).find("not enough blocks") != std::string::npos)

                    _ClusterDataExists.store(true);

            } catch (...) {

                std::cerr << "ClusterBackend::fetchFromCluster: retrieve failed (unknown)" << std::endl;

            }

        }

        }



        if (!bufferValid(peer_data))

            peer_data.clear();

src/backends/cluster.h

+4 −0
Changes for src/backends/cluster.h: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -66,6 +66,9 @@ namespace authdb { 


        bool clusterDataExists() const override { return _ClusterDataExists.load(); }



        // Force next fetchFromCluster to bypass the 30s throttle

        void forceRefresh() { _LastPeerFetch = {}; }



    private:

        uint64_t domainGroupId() const;

        void fetchFromCluster();
@@ -78,5 +81,6 @@ namespace authdb { 
        std::atomic<bool>        _Dirty;

        std::atomic<bool>        _ClusterDataExists{false};

        std::chrono::steady_clock::time_point _LastPeerFetch{};

        uint64_t _lastRecoveryEpoch{0};

    };

};

src/cluster.cpp

+1 −0
Changes for src/cluster.cpp: 1 added line, 0 removed lines.
Original line number Diff line number Diff line
@@ -528,6 +528,7 @@ namespace authdb { 
                        std::cerr << "Cluster: recovered — all "

                                  << health.nodes_online << "/" << n

                                  << " nodes online" << std::endl;

                        recovery_epoch_.fetch_add(1);

                    }

                    degraded_ = false;

                    critical_ = false;

src/cluster.h

+4 −0
Changes for src/cluster.h: 4 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -214,11 +214,15 @@ namespace authdb { 
        bool isDegraded() const { return degraded_; }

        bool isCritical() const { return critical_; }



        // Incremented on cluster recovery — backends check this to force refresh

        uint64_t getRecoveryEpoch() const { return recovery_epoch_.load(); }



    private:

        ClusterConfig cfg_;

        std::atomic<bool> running_{false};

        std::atomic<bool> degraded_{false};

        std::atomic<bool> critical_{false};

        std::atomic<uint64_t> recovery_epoch_{0};

        std::thread server_thread_;

        std::thread health_monitor_thread_;

        std::thread push_worker_thread_;