test (c4bb9cb0) · Commits · tuxist / authdb

src/cluster.cpp

+101 −18

Changes for src/cluster.cpp: 101 added lines, 18 removed lines.
Original line number	Diff line number	Diff line
		@@ -40,6 +40,7 @@
		#include <cstring>
		#include <iostream>
		#include <algorithm>
		#include <set>

		#ifdef _WIN32
		#include <winsock2.h>
		@@ -387,9 +388,9 @@ namespace authdb {
		uint64_t sgid = sessionGroupId(uid, did);
		uint64_t sid_gid = sidGroupId(sid);

		// Always store locally so hasSessionBySid() can find it
		store_->store(sgid, 0, data.data(), data.size());
		store_->store(sid_gid, 0, data.data(), data.size());
		// Store in dedicated session store (memory-only, never mixed with data)
		session_store_->store(sgid, 0, data.data(), data.size());
		session_store_->store(sid_gid, 0, data.data(), data.size());

		if (!pclient_) return;

		@@ -411,9 +412,9 @@ namespace authdb {
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results) {
		uint64_t sgid = sessionGroupId(uid, did);

		// Check local store first
		// Check local session store first
		std::vector<uint8_t> data;
		if (store_->fetch(sgid, 0, data)) {
		if (session_store_->fetch(sgid, 0, data)) {
		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) {
		return true;
		@@ -427,8 +428,8 @@ namespace authdb {
		if (pclient_->fetch_from_peers(sgid, 0, data) && !data.empty()) {
		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) {
		// Cache locally for next time
		store_->store(sgid, 0, data.data(), data.size());
		// Cache in session store
		session_store_->store(sgid, 0, data.data(), data.size());
		return true;
		}
		}
		@@ -445,14 +446,14 @@ namespace authdb {

		// Also remove the SID-keyed entry: fetch data first to get the SID
		std::vector<uint8_t> data;
		if (store_->fetch(sgid, 0, data)) {
		if (session_store_->fetch(sgid, 0, data)) {
		uuid::uuid sid, d_uid, d_did;
		std::vector<uuid::uuid> members;
		std::string username;
		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);
		store_->remove_group(sid_gid);
		session_store_->remove_group(sid_gid);
		if (pclient_) {
		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5)); if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		@@ -463,7 +464,7 @@ namespace authdb {
		}

		// Remove uid+did keyed entry
		store_->remove_group(sgid);
		session_store_->remove_group(sgid);

		if (pclient_) {
		try {
		@@ -473,12 +474,44 @@ namespace authdb {
		}
		}

		void Cluster::removeSessionBySid(const uuid::uuid &session_id) {
		uint64_t sid_gid = sidGroupId(session_id);

		// Fetch session data to get uid+did key as well
		std::vector<uint8_t> data;
		if (session_store_->fetch(sid_gid, 0, data)) {
		uuid::uuid sid, uid, did;
		std::vector<uuid::uuid> members;
		std::string username;
		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);
		session_store_->remove_group(sgid);
		if (pclient_) {
		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5)); if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		pclient_->remove(sgid);
		} catch (...) {}
		}
		}
		}

		session_store_->remove_group(sid_gid);

		if (pclient_) {
		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5)); if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		pclient_->remove(sid_gid);
		} catch (...) {}
		}
		}

		bool Cluster::hasSession(const uuid::uuid &uid, const uuid::uuid &did) {
		uint64_t sgid = sessionGroupId(uid, did);

		// Check local store
		// Check local session store
		std::vector<uint8_t> data;
		if (store_->fetch(sgid, 0, data)) return true;
		if (session_store_->fetch(sgid, 0, data)) return true;

		// Check peer nodes
		if (pclient_) {
		@@ -495,9 +528,9 @@ namespace authdb {
		bool Cluster::hasSessionBySid(const uuid::uuid &session_id) {
		uint64_t sid_gid = sidGroupId(session_id);

		// Check local store
		// Check local session store
		std::vector<uint8_t> data;
		if (store_->fetch(sid_gid, 0, data)) return true;
		if (session_store_->fetch(sid_gid, 0, data)) return true;

		// Check peer nodes
		if (pclient_) {
		@@ -518,9 +551,9 @@ namespace authdb {
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results) {
		uint64_t sid_gid = sidGroupId(session_id);

		// Check local store first
		// Check local session store first
		std::vector<uint8_t> data;
		if (store_->fetch(sid_gid, 0, data)) {
		if (session_store_->fetch(sid_gid, 0, data)) {
		if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) {
		return true;
		}
		@@ -532,8 +565,8 @@ namespace authdb {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5)); if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		if (pclient_->fetch_from_peers(sid_gid, 0, data) && !data.empty()) {
		if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) {
		// Cache locally
		store_->store(sid_gid, 0, data.data(), data.size());
		// Cache in session store
		session_store_->store(sid_gid, 0, data.data(), data.size());
		return true;
		}
		}
		@@ -545,6 +578,56 @@ namespace authdb {
		return false;
		}

		void Cluster::listAllSessions(std::vector<SessionInfo> &out) {
		out.clear();
		std::set<std::string> seen_sids;

		auto tryDeserialize = [&](const std::vector<uint8_t> &data) {
		SessionInfo info;
		if (SessionBlock::deserialize(data, info.sid, info.uid, info.did,
		info.members, info.username, info.gpo_results)) {
		std::string key = info.sid.c_str();
		if (!info.sid.empty() && seen_sids.find(key) == seen_sids.end()) {
		seen_sids.insert(key);
		out.push_back(std::move(info));
		}
		}
		};

		// Enumerate dedicated session store (contains only sessions)
		auto groups = session_store_->list_groups();
		for (uint64_t gid : groups) {
		std::vector<uint8_t> data;
		if (session_store_->fetch(gid, 0, data)) {
		tryDeserialize(data);
		}
		}

		// Enumerate each peer node for sessions we haven't seen
		if (pclient_) {
		for (size_t ni = 0; ni < cfg_.peers.size(); ni++) {
		try {
		std::vector<uint64_t> pgroups;
		{
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5));
		if (!lock.owns_lock()) continue;
		if (!pclient_->list_groups_on_node(ni, pgroups)) continue;
		}
		for (uint64_t gid : pgroups) {
		std::vector<uint8_t> data;
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(5));
		if (!lock.owns_lock()) break;
		if (pclient_->fetch_from_peers(gid, 0, data) && !data.empty()) {
		tryDeserialize(data);
		}
		}
		} catch (...) {
		continue;
		}
		}
		}
		}

		// --- DataOpBlock serialization ---

		std::vector<uint8_t> DataOpBlock::serialize(OpType op, const std::string &domain,

src/cluster.h

+3 −0

Changes for src/cluster.h: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -138,6 +138,9 @@ namespace authdb {
		// Remove a session from the cluster
		void removeSession(const uuid::uuid &uid, const uuid::uuid &did);

		// Remove a session from the cluster by SID
		void removeSessionBySid(const uuid::uuid &session_id);

		// Check if a session already exists in the cluster for uid+did
		bool hasSession(const uuid::uuid &uid, const uuid::uuid &did);