test (9cf3e659) · Commits · tuxist / authdb

src/session.cpp

+39 −104

Changes for src/session.cpp: 39 added lines, 104 removed lines.
Original line number	Diff line number	Diff line
		@@ -418,124 +418,72 @@ authdb::ClusterSession::ClusterSession() : LocalSession() {}
		authdb::ClusterSession::~ClusterSession() {}

		const authdb::SessionData *authdb::ClusterSession::addSession(AuthBackend &backend,uuid::uuid domainid,uuid::uuid userid){
		std::cerr << "[SESSION] addSession: checking local cache" << std::endl;
		// Check local cache first
		{
		std::lock_guard<std::mutex> guard(_lock);
		cleanupExpired();
		for (auto &[k, cur] : _sessions) {
		uuid::uuid cur_uid, cur_did;
		cur->getUid(cur_uid);
		cur->getDid(cur_did);
		if (cur_uid == userid && cur_did == domainid) {
		std::cerr << "[SESSION] addSession: found in local cache" << std::endl;
		return cur;
		}
		}
		}

		// Check cluster for existing session (skip only when critical)
		if (g_Cluster && g_Cluster->isRunning() && !g_Cluster->isCritical()) {
		std::cerr << "[SESSION] addSession: fetching from cluster" << std::endl;
		// Check cluster for existing session
		if (g_Cluster && g_Cluster->isRunning()) {
		uuid::uuid cluster_sid;
		std::vector<uuid::uuid> cluster_members;
		std::string cluster_username;
		std::vector<std::pair<uuid::uuid, bool>> cluster_gpo;
		if (g_Cluster->fetchSession(userid, domainid, cluster_sid,
		cluster_members, cluster_username, cluster_gpo)) {
		std::cerr << "[SESSION] addSession: imported from cluster" << std::endl;
		SessionData *imported = new SessionData(cluster_sid, userid, domainid, cluster_members);
		imported->_username = cluster_username;
		imported->setGPOResults(cluster_gpo);

		std::lock_guard<std::mutex> guard(_lock);
		// Replace any stale cache entry
		removeLocal(cluster_sid);
		appendLocal(imported);
		return imported;
		}
		std::cerr << "[SESSION] addSession: not found in cluster" << std::endl;
		} else if (g_Cluster && g_Cluster->isCritical()) {
		std::cerr << "[SESSION] addSession: skipping cluster fetch (critical)" << std::endl;
		}

		// Create new session
		std::cerr << "[SESSION] addSession: creating new session" << std::endl;
		SessionData *newSession = createSessionData(backend, domainid, userid);

		// Push to cluster with retry (login path — allowed to wait)
		bool pushed = false;
		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);
		}
		}

		{
		std::lock_guard<std::mutex> guard(_lock);
		appendLocal(newSession);
		}

		// Push to cluster for replication
		if (g_Cluster && g_Cluster->isRunning() && !g_Cluster->isCritical()) {
		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->isRunning() && g_Cluster->isCritical()) {
		std::cerr << "[SESSION] addSession: queuing push (critical)" << std::endl;
		g_Cluster->queuePush(*newSession);
		}

		return newSession;
		}

		const authdb::SessionData *authdb::ClusterSession::getSession(uuid::uuid sessionid){
		SessionData *cur = nullptr;
		bool needsSync = false;

		// Check temp cache first (TTL 5 seconds)
		{
		std::lock_guard<std::mutex> guard(_lock);
		cleanupExpired();
		cur = findLocal(sessionid);
		std::cerr << "[SESSION] getSession: findLocal=" << (cur ? "found" : "null")
		<< " map_size=" << _sessions.size() << std::endl;
		if (cur && g_Cluster && g_Cluster->isRunning() && !g_Cluster->isCritical()) {
		SessionData *cur = findLocal(sessionid);
		if (cur) {
		auto age = std::chrono::duration_cast<std::chrono::seconds>(
		std::chrono::steady_clock::now() - cur->_lastClusterSync).count();
		std::cerr << "[SESSION] getSession: cluster sync age=" << age
		<< "s critical=" << g_Cluster->isCritical() << std::endl;
		needsSync = (age >= 30);
		}
		}

		// Cluster sync OUTSIDE the session lock to avoid blocking other requests
		if (cur && needsSync) {
		uuid::uuid c_sid, c_uid, c_did;
		std::vector<uuid::uuid> c_members;
		std::string c_username;
		std::vector<std::pair<uuid::uuid, bool>> c_gpo;
		try {
		bool localOk = g_Cluster->fetchSessionLocalBySid(sessionid, c_sid, c_uid, c_did,
		c_members, c_username, c_gpo);
		std::cerr << "[SESSION] getSession: fetchLocalBySid=" << localOk << std::endl;
		std::lock_guard<std::mutex> guard(_lock);
		if (localOk) {
		cur->_members = c_members;
		cur->_username = c_username;
		cur->setGPOResults(c_gpo);
		cur->_lastClusterSync = std::chrono::steady_clock::now();
		} else if (!g_Cluster->isDegraded() && !g_Cluster->isCritical()) {
		// Only remove if cluster is fully healthy — an empty retrieve
		// on a degraded cluster doesn't prove the session was deleted.
		std::cerr << "[SESSION] getSession: session removed from cluster, removing locally" << std::endl;
		if (age < 5)
		return cur;
		// Cache expired — remove and re-fetch below
		removeLocal(sessionid);
		return nullptr;
		} else {
		// Cluster degraded/critical: keep local session, retry later
		std::cerr << "[SESSION] getSession: cluster degraded, keeping local session" << std::endl;
		cur->_lastClusterSync = std::chrono::steady_clock::now();
		}
		} catch (const std::exception &e) {
		std::cerr << "[SESSION] getSession: cluster sync error: " << e.what()
		<< ", keeping local session" << std::endl;
		std::lock_guard<std::mutex> guard(_lock);
		cur->_lastClusterSync = std::chrono::steady_clock::now();
		}
		}

		if (cur) return cur;

		// Not found locally — fetch from cluster peers
		// Fetch from cluster
		if (g_Cluster && g_Cluster->isRunning()) {
		uuid::uuid sid, uid, did;
		std::vector<uuid::uuid> members;
		@@ -545,6 +493,7 @@ const authdb::SessionData *authdb::ClusterSession::getSession(uuid::uuid session
		SessionData *imported = new SessionData(sid, uid, did, members);
		imported->_username = username;
		imported->setGPOResults(gpo_results);
		imported->_lastClusterSync = std::chrono::steady_clock::now();

		std::lock_guard<std::mutex> guard(_lock);
		appendLocal(imported);
		@@ -556,27 +505,22 @@ const authdb::SessionData *authdb::ClusterSession::getSession(uuid::uuid session
		}

		void authdb::ClusterSession::remSession(uuid::uuid sessionid){
		// Remove from cluster on ALL nodes
		// Remove from cluster
		if (g_Cluster && g_Cluster->isRunning()) {
		try {
		g_Cluster->removeSessionBySid(sessionid);
		} catch (...) {}
		}
		// Remove from local cache
		// Remove from temp cache
		std::lock_guard<std::mutex> guard(_lock);
		removeLocal(sessionid);
		}

		void authdb::ClusterSession::relSession(AuthBackend &backend,uuid::uuid sessionid){
		// Fetch current session from cluster
		SessionData *cursess = nullptr;

		{
		std::lock_guard<std::mutex> guard(_lock);
		cursess = findLocal(sessionid);
		}

		// Not found locally — try to import from cluster (outside lock)
		if (!cursess && g_Cluster && g_Cluster->isRunning()) {
		if (g_Cluster && g_Cluster->isRunning()) {
		uuid::uuid sid, uid, did;
		std::vector<uuid::uuid> members;
		std::string c_username;
		@@ -586,6 +530,7 @@ void authdb::ClusterSession::relSession(AuthBackend &backend,uuid::uuid sessioni
		cursess->_username = c_username;
		cursess->setGPOResults(gpo_results);
		std::lock_guard<std::mutex> guard(_lock);
		removeLocal(sessionid);
		appendLocal(cursess);
		}
		}
		@@ -598,16 +543,18 @@ void authdb::ClusterSession::relSession(AuthBackend &backend,uuid::uuid sessioni
		reloadSessionData(backend, cursess);
		}

		// Push updated data to cluster (outside lock)
		// Push updated data back to cluster
		if (g_Cluster && g_Cluster->isRunning()) {
		try {
		g_Cluster->pushSession(*cursess);
		} catch (...) {}
		}
		}

		void authdb::ClusterSession::listSessions(std::vector<SessionInfo> &out){
		out.clear();

		// Use cluster store as single source of truth
		// Always from cluster — single source of truth
		if (g_Cluster && g_Cluster->isRunning()) {
		std::vector<Cluster::SessionInfo> csessions;
		g_Cluster->listAllSessions(csessions);
		@@ -619,17 +566,5 @@ void authdb::ClusterSession::listSessions(std::vector<SessionInfo> &out){
		info.username = cs.username;
		out.push_back(std::move(info));
		}
		} else {
		// Fallback: local sessions only
		std::lock_guard<std::mutex> guard(_lock);
		cleanupExpired();
		for (auto &[k, cur] : _sessions) {
		SessionInfo info;
		cur->getSid(info.sid);
		cur->getUid(info.uid);
		cur->getDid(info.did);
		info.username = cur->getUsername();
		out.push_back(std::move(info));
		}
		}
		}