test (1975ef5c) · Commits · tuxist / authdb

src/cluster.cpp

+76 −2

Changes for src/cluster.cpp: 76 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -430,6 +430,7 @@ namespace authdb {
		running_ = true;
		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);
		}

		size_t Cluster::waitForPeers(int timeout_s) {
		@@ -473,10 +474,14 @@ namespace authdb {
		if (!running_) return;
		running_ = false;
		monitor_cv_.notify_all();
		push_cv_.notify_all();
		if (server_) server_->stop();
		if (health_monitor_thread_.joinable()) {
		health_monitor_thread_.join();
		}
		if (push_worker_thread_.joinable()) {
		push_worker_thread_.join();
		}
		if (server_thread_.joinable()) {
		server_thread_.join();
		}
		@@ -577,7 +582,11 @@ namespace authdb {
		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 (3s)" << std::endl;
		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();
		@@ -596,7 +605,72 @@ namespace authdb {
		<< "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() << std::endl;
		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) {
		if (!pclient_) return;
		auto data = SessionBlock::serialize(sess);
		uuid::uuid uid, did, sid;
		sess.getSid(sid);
		sess.getUid(uid);
		sess.getDid(did);
		uint64_t sgid = sessionGroupId(uid, did);
		uint64_t sid_gid = sidGroupId(sid);
		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::push_worker_loop() {
		while (running_) {
		std::unique_lock<std::mutex> lk(push_mutex_);
		push_cv_.wait_for(lk, std::chrono::seconds(5),
		[this]{ return !push_queue_.empty() \|\| !running_.load(); });
		if (!running_) break;
		if (push_queue_.empty()) continue;

		// Don't try to push while critical
		if (critical_) continue;

		// Take one item
		PendingPush item = std::move(push_queue_.front());
		push_queue_.pop_front();
		lk.unlock();

		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(3));
		if (!lock.owns_lock()) {
		// Re-queue
		item.retries++;
		if (item.retries < 5) {
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(item));
		} else {
		std::cerr << "[CLUSTER] push_worker: dropped after 5 retries" << std::endl;
		}
		continue;
		}
		pclient_->store(item.sgid, item.data.data(), item.data.size());
		pclient_->store(item.sid_gid, item.data.data(), item.data.size());
		std::cerr << "[CLUSTER] push_worker: pushed queued session" << std::endl;
		} catch (const std::exception &e) {
		item.retries++;
		if (item.retries < 5) {
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(item));
		} else {
		std::cerr << "[CLUSTER] push_worker: dropped after 5 retries: "
		<< e.what() << std::endl;
		}
		}
		}
		}

src/cluster.h

+3 −0

Changes for src/cluster.h: 3 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -158,6 +158,9 @@ namespace authdb {
		// Push a session to the cluster (after local addSession)
		void pushSession(const SessionData &sess);

		// Queue a session for background push (used when degraded or push fails)
		void queuePush(const SessionData &sess);

		// Try to fetch a session from the cluster by uid+did
		// Returns true if found, fills out parameters
		bool fetchSession(const uuid::uuid &uid, const uuid::uuid &did,

src/session.cpp

+8 −6

Changes for src/session.cpp: 8 added lines, 6 removed lines.
Original line number	Diff line number	Diff line
		@@ -464,13 +464,14 @@ const authdb::SessionData *authdb::ClusterSession::addSession(AuthBackend &backe
		appendLocal(newSession);
		}

		// Push to cluster for replication (skip when degraded — session is in local cache)
		// Push to cluster for replication
		if (g_Cluster && g_Cluster->isRunning() && !g_Cluster->isDegraded()) {
		std::cerr << "[SESSION] addSession: pushing to cluster" << std::endl;
		g_Cluster->pushSession(*newSession);
		std::cerr << "[SESSION] addSession: push done" << std::endl;
		} else if (g_Cluster && g_Cluster->isDegraded()) {
		std::cerr << "[SESSION] addSession: skipping cluster push (degraded)" << std::endl;
		} else if (g_Cluster && g_Cluster->isRunning() && g_Cluster->isDegraded()) {
		std::cerr << "[SESSION] addSession: queuing push (degraded)" << std::endl;
		g_Cluster->queuePush(*newSession);
		}

		return newSession;
		@@ -504,9 +505,10 @@ const authdb::SessionData *authdb::ClusterSession::getSession(uuid::uuid session
		cur->setGPOResults(c_gpo);
		cur->_lastClusterSync = std::chrono::steady_clock::now();
		} else {
		// Session was removed from cluster — invalidate locally
		removeLocal(sessionid);
		return nullptr;
		// Session not in cluster — re-push it instead of removing
		std::cerr << "[SESSION] getSession: session not in cluster, re-pushing" << std::endl;
		g_Cluster->pushSession(*cur);
		cur->_lastClusterSync = std::chrono::steady_clock::now();
		}
		} catch (const std::exception &e) {
		// Cluster fetch error — keep local session, retry later