test

        // This prevents a node with stale/empty data from wiping out freshly

        // imported or updated domain data on other nodes.

        try {

            auto &read_cli = cluster->getReadClient();

            auto &read_cli = cluster->getReadClient(domainGroupId());

            if (read_cli) {

                auto cluster_manifest = read_cli->retrieve(domainGroupId());

                if (cluster_manifest.size() >= sizeof(AuthHeader)) {

        DBG_LOG("[CLUSTER-BE] fetchFromCluster domain=" << _Domain

                  << " buf_size=" << _Buffer.size() << " dirty=" << _Dirty.load() << "\n");

        auto &client = cluster->getReadClient();

        auto &client = cluster->getReadClient(dgid);

        if (!client) return;

        // Single retrieve with timeout: manifest contains header + magic + entity keys + full record data

                if (self_idx != SIZE_MAX) {

                    pclient_->set_local_node(self_idx, store_);

                }

                // Dedicated read client — separate QUIC connections, never blocked

                // by push_worker's store() holding pclient_->mutex_

                read_client_ = std::make_unique<paritypp::client>(

                // Pool of dedicated read clients — separate QUIC connections,

                // never blocked by push_worker's store() holding pclient_->mutex_,

                // and spread across READ_CLIENT_POOL_SIZE instances so a slow

                // fetch for one domain (its client's mutex_ held for seconds)

                // doesn't stall unrelated domains hashed to a different instance.

                read_clients_.clear();

                for (size_t ci = 0; ci < READ_CLIENT_POOL_SIZE; ++ci) {

                    auto rc = std::make_unique<paritypp::client>(

                        cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds);

                    if (self_idx != SIZE_MAX) {

                    read_client_->set_local_node(self_idx, store_);

                        rc->set_local_node(self_idx, store_);

                    }

                    read_clients_.push_back(std::move(rc));

                }

                // Dedicated session write client — used only by push_worker

                session_client_ = std::make_unique<paritypp::client>(

                // Pre-establish QUIC connections on all clients to avoid

                // first-request handshake + auth latency

                pclient_->warmup();

                read_client_->warmup();

                for (auto &rc : read_clients_) rc->warmup();

                session_client_->warmup();

                session_read_client_->warmup();

                scrub_client_->warmup();

                            session_client_->mark_node_dead(i);

                            if (session_read_client_)

                                session_read_client_->mark_node_dead(i);

                            if (read_client_)

                                read_client_->mark_node_dead(i);

                            for (auto &rc : read_clients_)

                                if (rc) rc->mark_node_dead(i);

                            if (scrub_client_)

                                scrub_client_->mark_node_dead(i);

                        }

        // Accessors for backend use

        paritypp::block_store &getStore() { return *store_; }

        std::unique_ptr<paritypp::client> &getClient() { return pclient_; }

        std::unique_ptr<paritypp::client> &getReadClient() { return read_client_; }

        // Domain reads all share paritypp::client's single internal mutex_,

        // so on one client instance a slow/timed-out fetch for one domain

        // stalls every other domain's concurrent fetch behind it. Route by

        // domain_gid across a small pool of independent client instances

        // (separate connections + mutex each) so unrelated domains stop

        // blocking each other.

        std::unique_ptr<paritypp::client> &getReadClient(uint64_t domain_gid) {

            static std::unique_ptr<paritypp::client> null_client;

            if (read_clients_.empty()) return null_client;

            return read_clients_[domain_gid % read_clients_.size()];

        }

        std::unique_ptr<paritypp::client> &getSessionReadClient() { return session_read_client_; }

        // Vacuum all cluster nodes (acquires cluster-wide scrub lock)

        // (used by push_worker and health monitor) during long scrub operations

        std::unique_ptr<paritypp::client> scrub_client_;

        // Dedicated client for domain data reads — avoids contention with

        // pclient_ which is used by push_worker (store) and health monitor

        std::unique_ptr<paritypp::client> read_client_;

        // Pool of dedicated clients for domain data reads — avoids contention

        // with pclient_ (used by push_worker/health monitor), and pooling

        // across READ_CLIENT_POOL_SIZE instances (selected by domain_gid)

        // keeps unrelated domains' concurrent fetches from serializing behind

        // each other's mutex_.

        static constexpr size_t READ_CLIENT_POOL_SIZE = 4;

        std::vector<std::unique_ptr<paritypp::client>> read_clients_;

        // Dedicated client for session writes — used by push_worker only

        std::unique_ptr<paritypp::client> session_client_;