Loading src/backends/cluster.cpp +5 −0 Changes for src/backends/cluster.cpp: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -175,6 +175,11 @@ namespace authdb { ClusterBackend::ClusterBackend(const char *domain, size_t version, const char *filepath) : _Domain(domain), _Pos(0), _Dirty(false) { // Initialize recovery epoch to current cluster value so newly created // instances don't trigger a redundant forced refresh. if (g_Cluster) _lastRecoveryEpoch = g_Cluster->getRecoveryEpoch(); // Try to fetch existing data from cluster (single attempt, non-blocking). // If peers aren't reachable yet, lock() will fetch on first use. fetchFromCluster(); Loading src/cluster.cpp +41 −39 Changes for src/cluster.cpp: 41 added lines, 39 removed lines. Original line number Diff line number Diff line Loading @@ -411,6 +411,12 @@ namespace authdb { if (self_idx != SIZE_MAX) { pclient_->set_local_node(self_idx, store_); } // Dedicated session client — separate connections, no contention with domain fetches session_client_ = std::make_unique<paritypp::client>( cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds); if (self_idx != SIZE_MAX) { session_client_->set_local_node(self_idx, store_); } } else { std::cerr << "Cluster: no peers configured" << std::endl; } Loading Loading @@ -562,7 +568,7 @@ namespace authdb { } void Cluster::pushSession(const SessionData &sess) { if (!pclient_) return; if (!session_client_) return; auto data = SessionBlock::serialize(sess); Loading @@ -578,12 +584,12 @@ namespace authdb { << " sgid=" << sgid << " sid_gid=" << sid_gid << " data_size=" << data.size() << std::endl; // Synchronous push — 2s timeout to avoid blocking backend lock. std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); // Dedicated session client — no contention with domain fetches std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) throw AuthBackendError("cluster busy — session push timeout"); pclient_->store(sgid, data.data(), data.size()); pclient_->store(sid_gid, data.data(), data.size()); session_client_->store(sgid, data.data(), data.size()); session_client_->store(sid_gid, data.data(), data.size()); std::cerr << "[CLUSTER-SESSION] pushSession: stored OK" << std::endl; } Loading Loading @@ -693,13 +699,12 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> &gpo_results) { uint64_t sgid = sessionGroupId(uid, did); // Retrieve from peer nodes via erasure coding std::vector<uint8_t> data; if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; data = pclient_->retrieve(sgid); data = session_client_->retrieve(sgid); if (!data.empty()) { uuid::uuid d_uid, d_did; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) { Loading @@ -718,11 +723,11 @@ namespace authdb { uint64_t sgid = sessionGroupId(uid, did); // Fetch session to get the SID for removing the SID-keyed entry if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; std::vector<uint8_t> data = pclient_->retrieve(sgid); std::vector<uint8_t> data = session_client_->retrieve(sgid); if (!data.empty()) { uuid::uuid sid, d_uid, d_did; std::vector<uuid::uuid> members; Loading @@ -730,10 +735,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> gpo_results; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) { uint64_t sid_gid = sidGroupId(sid); pclient_->remove(sid_gid); session_client_->remove(sid_gid); } } pclient_->remove(sgid); session_client_->remove(sgid); } catch (...) {} } } Loading @@ -742,11 +747,11 @@ namespace authdb { uint64_t sid_gid = sidGroupId(session_id); // Fetch session data to get uid+did key if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; std::vector<uint8_t> data = pclient_->retrieve(sid_gid); std::vector<uint8_t> data = session_client_->retrieve(sid_gid); if (!data.empty()) { uuid::uuid sid, uid, did; std::vector<uuid::uuid> members; Loading @@ -754,10 +759,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> gpo_results; if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) { uint64_t sgid = sessionGroupId(uid, did); pclient_->remove(sgid); session_client_->remove(sgid); } } pclient_->remove(sid_gid); session_client_->remove(sid_gid); } catch (...) {} } } Loading @@ -765,11 +770,11 @@ namespace authdb { bool Cluster::hasSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t sgid = sessionGroupId(uid, did); if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sgid); auto data = session_client_->retrieve(sgid); if (!data.empty()) return true; } catch (...) {} } Loading @@ -784,12 +789,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> &gpo_results) { // No local store — retrieve via erasure coding uint64_t sid_gid = sidGroupId(session_id); if (pclient_) { // Let exceptions propagate so caller can distinguish // "not found" (return false) from "cluster error" (exception) std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); if (session_client_) { std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); if (!data.empty()) { return SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results); } Loading @@ -800,11 +803,11 @@ namespace authdb { bool Cluster::hasSessionBySid(const uuid::uuid &session_id) { uint64_t sid_gid = sidGroupId(session_id); if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); if (!data.empty()) return true; } catch (...) {} } Loading @@ -822,14 +825,14 @@ namespace authdb { std::cerr << "[CLUSTER-SESSION] fetchBySid: sid=" << session_id.c_str() << " sid_gid=" << sid_gid << std::endl; if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) { std::cerr << "[CLUSTER-SESSION] fetchBySid: client_mutex_ timeout" << std::endl; std::cerr << "[CLUSTER-SESSION] fetchBySid: session_client_mutex_ timeout" << std::endl; return false; } auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); std::cerr << "[CLUSTER-SESSION] fetchBySid: retrieved " << data.size() << " bytes" << std::endl; if (!data.empty()) { if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) { Loading @@ -852,18 +855,17 @@ namespace authdb { void Cluster::listAllSessions(std::vector<SessionInfo> &out) { out.clear(); if (!pclient_) return; if (!session_client_) return; std::set<std::string> seen_sids; // Collect all group IDs across all nodes std::set<uint64_t> all_groups; try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; size_t n = pclient_->get_cluster_status().nodes_total; size_t n = session_client_->get_cluster_status().nodes_total; for (size_t i = 0; i < n; i++) { std::vector<uint64_t> node_groups; if (pclient_->list_groups_on_node(i, node_groups)) { if (session_client_->list_groups_on_node(i, node_groups)) { all_groups.insert(node_groups.begin(), node_groups.end()); } } Loading src/cluster.h +4 −0 Changes for src/cluster.h: 4 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -253,6 +253,10 @@ namespace authdb { std::unique_ptr<paritypp::client> pclient_; std::timed_mutex client_mutex_; // Dedicated client for session ops — avoids contention with domain data fetches std::unique_ptr<paritypp::client> session_client_; std::timed_mutex session_client_mutex_; void server_loop(); void health_monitor_loop(); void push_worker_loop(); Loading Loading
src/backends/cluster.cpp +5 −0 Changes for src/backends/cluster.cpp: 5 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -175,6 +175,11 @@ namespace authdb { ClusterBackend::ClusterBackend(const char *domain, size_t version, const char *filepath) : _Domain(domain), _Pos(0), _Dirty(false) { // Initialize recovery epoch to current cluster value so newly created // instances don't trigger a redundant forced refresh. if (g_Cluster) _lastRecoveryEpoch = g_Cluster->getRecoveryEpoch(); // Try to fetch existing data from cluster (single attempt, non-blocking). // If peers aren't reachable yet, lock() will fetch on first use. fetchFromCluster(); Loading
src/cluster.cpp +41 −39 Changes for src/cluster.cpp: 41 added lines, 39 removed lines. Original line number Diff line number Diff line Loading @@ -411,6 +411,12 @@ namespace authdb { if (self_idx != SIZE_MAX) { pclient_->set_local_node(self_idx, store_); } // Dedicated session client — separate connections, no contention with domain fetches session_client_ = std::make_unique<paritypp::client>( cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds); if (self_idx != SIZE_MAX) { session_client_->set_local_node(self_idx, store_); } } else { std::cerr << "Cluster: no peers configured" << std::endl; } Loading Loading @@ -562,7 +568,7 @@ namespace authdb { } void Cluster::pushSession(const SessionData &sess) { if (!pclient_) return; if (!session_client_) return; auto data = SessionBlock::serialize(sess); Loading @@ -578,12 +584,12 @@ namespace authdb { << " sgid=" << sgid << " sid_gid=" << sid_gid << " data_size=" << data.size() << std::endl; // Synchronous push — 2s timeout to avoid blocking backend lock. std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); // Dedicated session client — no contention with domain fetches std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) throw AuthBackendError("cluster busy — session push timeout"); pclient_->store(sgid, data.data(), data.size()); pclient_->store(sid_gid, data.data(), data.size()); session_client_->store(sgid, data.data(), data.size()); session_client_->store(sid_gid, data.data(), data.size()); std::cerr << "[CLUSTER-SESSION] pushSession: stored OK" << std::endl; } Loading Loading @@ -693,13 +699,12 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> &gpo_results) { uint64_t sgid = sessionGroupId(uid, did); // Retrieve from peer nodes via erasure coding std::vector<uint8_t> data; if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; data = pclient_->retrieve(sgid); data = session_client_->retrieve(sgid); if (!data.empty()) { uuid::uuid d_uid, d_did; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) { Loading @@ -718,11 +723,11 @@ namespace authdb { uint64_t sgid = sessionGroupId(uid, did); // Fetch session to get the SID for removing the SID-keyed entry if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; std::vector<uint8_t> data = pclient_->retrieve(sgid); std::vector<uint8_t> data = session_client_->retrieve(sgid); if (!data.empty()) { uuid::uuid sid, d_uid, d_did; std::vector<uuid::uuid> members; Loading @@ -730,10 +735,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> gpo_results; if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) { uint64_t sid_gid = sidGroupId(sid); pclient_->remove(sid_gid); session_client_->remove(sid_gid); } } pclient_->remove(sgid); session_client_->remove(sgid); } catch (...) {} } } Loading @@ -742,11 +747,11 @@ namespace authdb { uint64_t sid_gid = sidGroupId(session_id); // Fetch session data to get uid+did key if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; std::vector<uint8_t> data = pclient_->retrieve(sid_gid); std::vector<uint8_t> data = session_client_->retrieve(sid_gid); if (!data.empty()) { uuid::uuid sid, uid, did; std::vector<uuid::uuid> members; Loading @@ -754,10 +759,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> gpo_results; if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) { uint64_t sgid = sessionGroupId(uid, did); pclient_->remove(sgid); session_client_->remove(sgid); } } pclient_->remove(sid_gid); session_client_->remove(sid_gid); } catch (...) {} } } Loading @@ -765,11 +770,11 @@ namespace authdb { bool Cluster::hasSession(const uuid::uuid &uid, const uuid::uuid &did) { uint64_t sgid = sessionGroupId(uid, did); if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sgid); auto data = session_client_->retrieve(sgid); if (!data.empty()) return true; } catch (...) {} } Loading @@ -784,12 +789,10 @@ namespace authdb { std::vector<std::pair<uuid::uuid, bool>> &gpo_results) { // No local store — retrieve via erasure coding uint64_t sid_gid = sidGroupId(session_id); if (pclient_) { // Let exceptions propagate so caller can distinguish // "not found" (return false) from "cluster error" (exception) std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); if (session_client_) { std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); if (!data.empty()) { return SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results); } Loading @@ -800,11 +803,11 @@ namespace authdb { bool Cluster::hasSessionBySid(const uuid::uuid &session_id) { uint64_t sid_gid = sidGroupId(session_id); if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return false; auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); if (!data.empty()) return true; } catch (...) {} } Loading @@ -822,14 +825,14 @@ namespace authdb { std::cerr << "[CLUSTER-SESSION] fetchBySid: sid=" << session_id.c_str() << " sid_gid=" << sid_gid << std::endl; if (pclient_) { if (session_client_) { try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) { std::cerr << "[CLUSTER-SESSION] fetchBySid: client_mutex_ timeout" << std::endl; std::cerr << "[CLUSTER-SESSION] fetchBySid: session_client_mutex_ timeout" << std::endl; return false; } auto data = pclient_->retrieve(sid_gid); auto data = session_client_->retrieve(sid_gid); std::cerr << "[CLUSTER-SESSION] fetchBySid: retrieved " << data.size() << " bytes" << std::endl; if (!data.empty()) { if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) { Loading @@ -852,18 +855,17 @@ namespace authdb { void Cluster::listAllSessions(std::vector<SessionInfo> &out) { out.clear(); if (!pclient_) return; if (!session_client_) return; std::set<std::string> seen_sids; // Collect all group IDs across all nodes std::set<uint64_t> all_groups; try { std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) return; size_t n = pclient_->get_cluster_status().nodes_total; size_t n = session_client_->get_cluster_status().nodes_total; for (size_t i = 0; i < n; i++) { std::vector<uint64_t> node_groups; if (pclient_->list_groups_on_node(i, node_groups)) { if (session_client_->list_groups_on_node(i, node_groups)) { all_groups.insert(node_groups.begin(), node_groups.end()); } } Loading
src/cluster.h +4 −0 Changes for src/cluster.h: 4 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -253,6 +253,10 @@ namespace authdb { std::unique_ptr<paritypp::client> pclient_; std::timed_mutex client_mutex_; // Dedicated client for session ops — avoids contention with domain data fetches std::unique_ptr<paritypp::client> session_client_; std::timed_mutex session_client_mutex_; void server_loop(); void health_monitor_loop(); void push_worker_loop(); Loading