test (bd8a330f) · Commits · tuxist / authdb

src/cluster.cpp

+77 −6

Changes for src/cluster.cpp: 77 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -88,9 +88,9 @@ namespace authdb {
		std::vector<std::pair<uuid::uuid, bool>> gpo_results;
		sess.getGPOResults(gpo_results);

		// Magic(4) + Header: 16+16+16 + 4 + members16 + 4 + username + 4 + gpos17
		// Magic(4) + Header: 16+16+16 + 4 + members16 + 4 + username + 4 + gpos17 + 8 timestamp
		size_t total = 4 + 48 + 4 + (members.size() * 16) + 4 + username_len
		+ 4 + (gpo_results.size() * 17);
		+ 4 + (gpo_results.size() * 17) + 8;

		std::vector<uint8_t> buf(total);
		size_t pos = 0;
		@@ -144,6 +144,12 @@ namespace authdb {
		pos += 1;
		}

		// Timestamp (seconds since epoch)
		int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
		std::chrono::system_clock::now().time_since_epoch()).count();
		std::memcpy(buf.data() + pos, &now, 8);
		pos += 8;

		return buf;
		}

		@@ -151,7 +157,9 @@ namespace authdb {
		uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did,
		std::vector<uuid::uuid> &members,
		std::string &username,
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results) {
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results,
		int64_t &created_at) {
		created_at = 0;
		if (data.size() < 60) return false; // 4 magic + 48 uuids + 4 member_count + 4 username_len

		// Check magic
		@@ -208,6 +216,12 @@ namespace authdb {
		}
		}

		// Parse timestamp (may be absent in older format)
		if (pos + 8 <= data.size()) {
		std::memcpy(&created_at, data.data() + pos, 8);
		pos += 8;
		}

		return true;
		}

		@@ -418,6 +432,7 @@ namespace authdb {
		server_thread_ = std::thread(&Cluster::server_loop, this);
		health_monitor_thread_ = std::thread(&Cluster::health_monitor_loop, this);
		push_worker_thread_ = std::thread(&Cluster::push_worker_loop, this);
		session_cleanup_thread_ = std::thread(&Cluster::session_cleanup_loop, this);
		}

		size_t Cluster::waitForPeers(int timeout_s) {
		@@ -475,6 +490,9 @@ namespace authdb {
		if (push_worker_thread_.joinable()) {
		push_worker_thread_.join();
		}
		if (session_cleanup_thread_.joinable()) {
		session_cleanup_thread_.join();
		}
		if (server_thread_.joinable()) {
		server_thread_.join();
		}
		@@ -678,6 +696,56 @@ namespace authdb {
		}
		}

		void Cluster::session_cleanup_loop() {
		while (running_) {
		// Sleep 5 minutes between cleanup runs
		for (int i = 0; i < 300 && running_; ++i)
		std::this_thread::sleep_for(std::chrono::seconds(1));
		if (!running_) break;
		if (critical_) continue;
		if (!session_client_) continue;

		try {
		int64_t now = std::chrono::duration_cast<std::chrono::seconds>(
		std::chrono::system_clock::now().time_since_epoch()).count();

		std::set<uint64_t> all_groups;
		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 (session_client_->list_groups_on_node(i, node_groups))
		all_groups.insert(node_groups.begin(), node_groups.end());
		}

		int removed = 0;
		for (uint64_t gid : all_groups) {
		try {
		auto data = session_client_->retrieve(gid);
		if (data.empty()) continue;
		uuid::uuid sid, uid, did;
		std::vector<uuid::uuid> members;
		std::string username;
		std::vector<std::pair<uuid::uuid, bool>> gpo_results;
		int64_t created_at = 0;
		if (SessionBlock::deserialize(data, sid, uid, did, members, username,
		gpo_results, created_at)) {
		if (created_at > 0 && (now - created_at) > SESSION_TTL_SECONDS) {
		session_client_->remove(gid);
		removed++;
		}
		}
		} catch (...) {}
		}
		if (removed > 0) {
		std::cerr << "[CLUSTER-SESSION] cleanup: removed " << removed
		<< " expired sessions" << std::endl;
		}
		} catch (const std::exception &e) {
		std::cerr << "[CLUSTER-SESSION] cleanup error: " << e.what() << std::endl;
		}
		}
		}

		bool Cluster::fetchSessionLocalBySid(const uuid::uuid &session_id,
		uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did,
		std::vector<uuid::uuid> &members,
		@@ -688,7 +756,8 @@ namespace authdb {
		if (session_client_) {
		auto data = session_client_->retrieve(sid_gid);
		if (!data.empty()) {
		return SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results);
		int64_t created_at;
		return SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results, created_at);
		}
		}
		return false;
		@@ -722,7 +791,8 @@ namespace authdb {
		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)) {
		int64_t created_at;
		if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results, created_at)) {
		std::cerr << "[CLUSTER-SESSION] fetchBySid: deserialized OK, sid=" << sid.c_str() << std::endl;
		return true;
		}
		@@ -762,7 +832,8 @@ namespace authdb {
		if (data.empty()) continue;
		SessionInfo info;
		if (SessionBlock::deserialize(data, info.sid, info.uid, info.did,
		info.members, info.username, info.gpo_results)) {
		info.members, info.username, info.gpo_results,
		info.created_at)) {
		std::string key = info.sid.c_str();
		if (!info.sid.empty() && seen_sids.find(key) == seen_sids.end()) {
		seen_sids.insert(key);

src/cluster.h

+7 −1

Changes for src/cluster.h: 7 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -98,7 +98,8 @@ namespace authdb {
		uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did,
		std::vector<uuid::uuid> &members,
		std::string &username,
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results);
		std::vector<std::pair<uuid::uuid, bool>> &gpo_results,
		int64_t &created_at);
		};

		uint64_t sidGroupId(const uuid::uuid &sid);
		@@ -193,6 +194,7 @@ namespace authdb {
		std::vector<uuid::uuid> members;
		std::string username;
		std::vector<std::pair<uuid::uuid, bool>> gpo_results;
		int64_t created_at{0};
		};
		void listAllSessions(std::vector<SessionInfo> &out);

		@@ -242,6 +244,10 @@ namespace authdb {
		void server_loop();
		void health_monitor_loop();
		void push_worker_loop();
		void session_cleanup_loop();

		std::thread session_cleanup_thread_;
		static constexpr int64_t SESSION_TTL_SECONDS = 24 * 3600; // 24h
		};

		// Global cluster instance