test (d259c1f2) · Commits · tuxist / authdb

src/cluster.cpp

+57 −10

Changes for src/cluster.cpp: 57 added lines, 10 removed lines.
Original line number	Diff line number	Diff line
		@@ -245,15 +245,19 @@ namespace authdb {
		sess.getUid(uid);
		sess.getDid(did);

		uint64_t gid = sessionGroupId(uid, did);
		uint64_t sgid = sessionGroupId(uid, did);

		// Also store locally in memory_block_store for peers to fetch
		store_->store(gid, 0, data.data(), data.size());
		store_->store(sgid, 0, data.data(), data.size());

		// Distribute to peers with parity encoding
		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		pclient_->store(data);
		uint64_t pgid = pclient_->store(data);

		// Track the mapping so we can retrieve later
		std::lock_guard<std::mutex> ilock(index_mutex_);
		session_index_[sgid] = {pgid, data.size()};
		} catch (const std::exception &e) {
		// Log but don't fail - local session still valid
		std::cerr << "Cluster push failed: " << e.what() << std::endl;
		@@ -263,13 +267,38 @@ namespace authdb {
		bool Cluster::fetchSession(const uuid::uuid &uid, const uuid::uuid &did,
		uuid::uuid &sid, std::vector<uuid::uuid> &members,
		std::string &username) {
		uint64_t gid = sessionGroupId(uid, did);
		uint64_t sgid = sessionGroupId(uid, did);

		// Check local store first
		std::vector<uint8_t> data;
		if (store_->fetch(gid, 0, data)) {
		if (store_->fetch(sgid, 0, data)) {
		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username)) {
		return true;
		}
		}

		// Try to retrieve from cluster peers via parity client
		if (pclient_) {
		StoredSession info;
		{
		std::lock_guard<std::mutex> ilock(index_mutex_);
		auto it = session_index_.find(sgid);
		if (it == session_index_.end()) return false;
		info = it->second;
		}

		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		data = pclient_->retrieve(info.parity_group_id, info.data_size);
		} catch (...) {
		return false;
		}

		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username)) {
		// Cache locally for next time
		store_->store(sgid, 0, data.data(), data.size());
		return true;
		}
		}
		@@ -278,24 +307,42 @@ namespace authdb {
		}

		void Cluster::removeSession(const uuid::uuid &uid, const uuid::uuid &did) {
		uint64_t gid = sessionGroupId(uid, did);
		uint64_t sgid = sessionGroupId(uid, did);

		// Remove from local store
		store_->remove_group(gid);
		store_->remove_group(sgid);

		// Remove from peers
		if (pclient_) {
		StoredSession info;
		{
		std::lock_guard<std::mutex> ilock(index_mutex_);
		auto it = session_index_.find(sgid);
		if (it != session_index_.end()) {
		info = it->second;
		session_index_.erase(it);
		} else {
		return;
		}
		}

		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		pclient_->remove(gid);
		pclient_->remove(info.parity_group_id);
		} catch (...) {}
		}
		}

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

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

		// Check index
		std::lock_guard<std::mutex> ilock(index_mutex_);
		return session_index_.count(sgid) > 0;
		}

		};

src/cluster.h

+8 −0

Changes for src/cluster.h: 8 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -120,6 +120,14 @@ namespace authdb {
		std::unique_ptr<paritypp::client> pclient_;
		std::mutex client_mutex_;

		// Maps sessionGroupId → {paritypp_group_id, original_data_size}
		struct StoredSession {
		uint64_t parity_group_id;
		size_t data_size;
		};
		std::map<uint64_t, StoredSession> session_index_;
		std::mutex index_mutex_;

		void server_loop();
		};

src/session.cpp

+13 −0

Changes for src/session.cpp: 13 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -436,6 +436,11 @@ void authdb::Session::relSession(AuthBackend &backend,uuid::uuid sessionid){
		cursess->_members.emplace_back(cursess->_uid);
		}
		}

		// Update cluster with refreshed session data
		if (g_Cluster && g_Cluster->isRunning()) {
		g_Cluster->pushSession(*cursess);
		}
		}

		const authdb::SessionData *authdb::Session::firstData() const{
		@@ -450,6 +455,14 @@ void authdb::Session::cleanupExpired(){
		while(cur){
		auto age=std::chrono::duration_cast<std::chrono::hours>(now - cur->_created).count();
		if(age >= 24){
		// Remove from cluster
		if (g_Cluster && g_Cluster->isRunning()) {
		uuid::uuid uid, did;
		cur->getUid(uid);
		cur->getDid(did);
		g_Cluster->removeSession(uid, did);
		}

		SessionData *next=cur->_next;
		if(prev)
		prev->_next=next;