test (96bd5c5e) · Commits · tuxist / authdb

src/cluster.cpp

+12 −53

Changes for src/cluster.cpp: 12 added lines, 53 removed lines.
Original line number	Diff line number	Diff line
		@@ -563,7 +563,6 @@ namespace authdb {
		void Cluster::pushSession(const SessionData &sess) {
		if (!pclient_) return;

		auto t0 = std::chrono::steady_clock::now();
		auto data = SessionBlock::serialize(sess);

		uuid::uuid uid, did, sid;
		@@ -574,40 +573,11 @@ namespace authdb {
		uint64_t sgid = sessionGroupId(uid, did);
		uint64_t sid_gid = sidGroupId(sid);

		try {
		std::cerr << "[CLUSTER] pushSession: acquiring lock" << std::endl;
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(3));
		if (!lock.owns_lock()) {
		std::cerr << "[CLUSTER] pushSession: lock timeout, queuing" << std::endl;
		PendingPush pp{data, sgid, sid_gid, 0};
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(pp));
		push_cv_.notify_one();
		return;
		}
		auto t1 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] pushSession: lock acquired in "
		<< std::chrono::duration_cast<std::chrono::milliseconds>(t1-t0).count()
		<< "ms, storing sgid" << std::endl;
		// Synchronous push — block until lock acquired, throw on failure.
		// The caller (addSession) retries, so no queuing needed here.
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		pclient_->store(sgid, data.data(), data.size());
		auto t2 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] pushSession: store(sgid) took "
		<< std::chrono::duration_cast<std::chrono::milliseconds>(t2-t1).count()
		<< "ms, storing sid_gid" << std::endl;
		pclient_->store(sid_gid, data.data(), data.size());
		auto t3 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] pushSession: store(sid_gid) took "
		<< std::chrono::duration_cast<std::chrono::milliseconds>(t3-t2).count()
		<< "ms, total=" << std::chrono::duration_cast<std::chrono::milliseconds>(t3-t0).count()
		<< "ms" << std::endl;
		} catch (const std::exception &e) {
		std::cerr << "[CLUSTER] pushSession: exception: " << e.what()
		<< ", queuing for retry" << std::endl;
		PendingPush pp{data, sgid, sid_gid, 0};
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(pp));
		push_cv_.notify_one();
		}
		}

		void Cluster::queuePush(const SessionData &sess) {
		@@ -720,17 +690,8 @@ namespace authdb {
		if (pclient_) {
		try {
		auto t0 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] fetchSession: acquiring lock" << std::endl;
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(2)); if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		auto t1 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] fetchSession: lock in "
		<< std::chrono::duration_cast<std::chrono::milliseconds>(t1-t0).count()
		<< "ms, retrieving" << std::endl;
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		data = pclient_->retrieve(sgid);
		auto t2 = std::chrono::steady_clock::now();
		std::cerr << "[CLUSTER] fetchSession: retrieve took "
		<< std::chrono::duration_cast<std::chrono::milliseconds>(t2-t1).count()
		<< "ms, size=" << data.size() << std::endl;
		if (!data.empty()) {
		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username, gpo_results)) {
		@@ -751,7 +712,7 @@ namespace authdb {
		// Fetch session to get the SID for removing the SID-keyed entry
		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		std::vector<uint8_t> data = pclient_->retrieve(sgid);
		if (!data.empty()) {
		uuid::uuid sid, d_uid, d_did;
		@@ -774,7 +735,7 @@ namespace authdb {
		// Fetch session data to get uid+did key
		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		std::vector<uint8_t> data = pclient_->retrieve(sid_gid);
		if (!data.empty()) {
		uuid::uuid sid, uid, did;
		@@ -796,7 +757,7 @@ namespace authdb {

		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		auto data = pclient_->retrieve(sgid);
		if (!data.empty()) return true;
		} catch (...) {}
		@@ -815,8 +776,7 @@ namespace authdb {
		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(5));
		if (!lock.owns_lock()) throw std::runtime_error("cluster client lock timeout");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		auto data = pclient_->retrieve(sid_gid);
		if (!data.empty()) {
		return SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results);
		@@ -830,7 +790,7 @@ namespace authdb {

		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		auto data = pclient_->retrieve(sid_gid);
		if (!data.empty()) return true;
		} catch (...) {}
		@@ -848,7 +808,7 @@ namespace authdb {

		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		auto data = pclient_->retrieve(sid_gid);
		if (!data.empty()) {
		if (SessionBlock::deserialize(data, sid, uid, did, members, username, gpo_results)) {
		@@ -871,8 +831,7 @@ namespace authdb {
		// 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(5));
		if (!lock.owns_lock()) return;
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		size_t n = pclient_->get_cluster_status().nodes_total;
		for (size_t i = 0; i < n; i++) {
		std::vector<uint64_t> node_groups;
		@@ -1030,7 +989,7 @@ namespace authdb {

		// Distribute erasure-coded shards to peer nodes
		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");
		std::unique_lock<std::timed_mutex> lock(client_mutex_);
		pclient_->store(localId, data.data(), data.size());
		} catch (const netplus::NetException &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;

src/session.cpp

+1 −14

Changes for src/session.cpp: 1 added line, 14 removed lines.
Original line number	Diff line number	Diff line
		@@ -442,22 +442,9 @@ const authdb::SessionData *authdb::ClusterSession::addSession(AuthBackend &backe
		// Create new session
		SessionData *newSession = createSessionData(backend, domainid, userid);

		// Push to cluster with retry (login path — allowed to wait)
		bool pushed = false;
		// Synchronous push to cluster — blocks until stored
		if (g_Cluster && g_Cluster->isRunning()) {
		for (int attempt = 0; attempt < 3 && !pushed; ++attempt) {
		try {
		g_Cluster->pushSession(*newSession);
		pushed = true;
		} catch (const std::exception &e) {
		std::cerr << "[SESSION] addSession: push attempt " << (attempt+1)
		<< " failed: " << e.what() << std::endl;
		}
		}
		if (!pushed) {
		// Last resort: queue for background push
		g_Cluster->queuePush(*newSession);
		}
		}

		{