Commit 51198080 authored by jan.koester's avatar jan.koester
Browse files

test

parent 9b64150d

src/backends/cluster.cpp

+3 −17
Changes for src/backends/cluster.cpp: 3 added lines, 17 removed lines.
Original line number Diff line number Diff line
@@ -235,24 +235,10 @@ namespace authdb { 
        if (_Dirty) {

            newRevesion();

            uint64_t dgid = domainGroupId();

            // Push in a detached background thread so unlock() never blocks

            // the HTTP response on cluster timeouts.

            auto buf_copy = std::make_shared<std::vector<uint8_t>>(_Buffer);

            std::thread([dgid, buf_copy]() {

                if (!g_Cluster || !g_Cluster->isRunning()) return;

                auto &client = g_Cluster->getClient();

                if (!client) return;

                try {

                    auto &cmtx = g_Cluster->getClientMutex();

                    std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::seconds(5));

                    if (lock.owns_lock()) {

                        client->store(dgid, buf_copy->data(), buf_copy->size());

                    }

                } catch (const std::exception &e) {

                    std::cerr << "Cluster replicate (bg): " << e.what() << std::endl;

            // Queue for background push via the single worker thread

            if (g_Cluster && g_Cluster->isRunning()) {

                g_Cluster->queueDataPush(dgid, _Buffer);

            }

            }).detach();

            // Optimistically clear dirty — data is captured in bg thread

            _Dirty = false;

        }

    }

src/cluster.cpp

+41 −2
Changes for src/cluster.cpp: 41 added lines, 2 removed lines.
Original line number Diff line number Diff line
@@ -629,17 +629,56 @@ namespace authdb { 
        push_cv_.notify_one();

    }



    void Cluster::queueDataPush(uint64_t dgid, std::vector<uint8_t> buf) {

        std::lock_guard<std::mutex> pguard(push_mutex_);

        // Replace any pending push for the same domain group (only latest matters)

        for (auto &dp : data_push_queue_) {

            if (dp.dgid == dgid) {

                dp.buf = std::move(buf);

                push_cv_.notify_one();

                return;

            }

        }

        data_push_queue_.push_back({dgid, std::move(buf)});

        push_cv_.notify_one();

    }



    void Cluster::push_worker_loop() {

        while (running_) {

            std::unique_lock<std::mutex> lk(push_mutex_);

            push_cv_.wait_for(lk, std::chrono::seconds(5),

                              [this]{ return !push_queue_.empty() || !running_.load(); });

                              [this]{ return !push_queue_.empty() || !data_push_queue_.empty() || !running_.load(); });

            if (!running_) break;

            if (push_queue_.empty()) continue;



            // Don't try to push while critical

            if (critical_) continue;



            // Drain data pushes first (coalesce: only latest per domain)

            if (!data_push_queue_.empty()) {

                PendingDataPush dp = std::move(data_push_queue_.front());

                data_push_queue_.pop_front();

                lk.unlock();



                try {

                    std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(3));

                    if (lock.owns_lock()) {

                        pclient_->store(dp.dgid, dp.buf.data(), dp.buf.size());

                    } else {

                        // Re-queue (will be coalesced if newer push arrives)

                        std::lock_guard<std::mutex> pguard(push_mutex_);

                        data_push_queue_.push_back(std::move(dp));

                    }

                } catch (const std::exception &e) {

                    std::cerr << "[CLUSTER] push_worker: data push failed: " << e.what() << std::endl;

                    std::lock_guard<std::mutex> pguard(push_mutex_);

                    data_push_queue_.push_back(std::move(dp));

                }

                continue;

            }



            // Session pushes

            if (push_queue_.empty()) continue;



            // Take one item

            PendingPush item = std::move(push_queue_.front());

            push_queue_.pop_front();

src/cluster.h

+9 −0
Changes for src/cluster.h: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -161,6 +161,9 @@ namespace authdb { 
        // Queue a session for background push (used when degraded or push fails)

        void queuePush(const SessionData &sess);



        // Queue a backend data push for background replication

        void queueDataPush(uint64_t dgid, std::vector<uint8_t> buf);



        // Try to fetch a session from the cluster by uid+did

        // Returns true if found, fills out parameters

        bool fetchSession(const uuid::uuid &uid, const uuid::uuid &did,
@@ -232,6 +235,12 @@ namespace authdb { 
        };

        std::deque<PendingPush> push_queue_;



        struct PendingDataPush {

            uint64_t dgid;

            std::vector<uint8_t> buf;

        };

        std::deque<PendingDataPush> data_push_queue_;



        std::shared_ptr<paritypp::block_store> store_;

        std::shared_ptr<paritypp::memory_block_store> session_store_; // sessions only

        std::shared_ptr<paritypp::auth_store> auth_;