test (1d337d9e) · Commits · tuxist / authdb

src/cluster.cpp

+25 −113

Changes for src/cluster.cpp: 25 added lines, 113 removed lines.
Original line number	Diff line number	Diff line
		@@ -270,23 +270,17 @@ namespace authdb {

		uint64_t sgid = sessionGroupId(uid, did);

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

		// Distribute to peers with parity encoding
		// Replicate raw data to all peer nodes
		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		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()};
		pclient_->replicate_to_peers(sgid, 0, data.data(), data.size());
		} catch (const netplus::NetException &e) {
		// Log but don't fail - local session still valid
		std::cerr << "Cluster push failed: " << e.what() << std::endl;
		std::cerr << "Cluster push session failed: " << e.what() << std::endl;
		} catch (const std::exception &e) {
		// Log but don't fail - local session still valid
		std::cerr << "Cluster push failed: " << e.what() << std::endl;
		std::cerr << "Cluster push session failed: " << e.what() << std::endl;
		}
		}

		@@ -304,23 +298,11 @@ namespace authdb {
		}
		}

		// Try to retrieve from cluster peers via parity client
		// Try to fetch from peer nodes
		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;
		}

		if (pclient_->fetch_from_peers(sgid, 0, data) && !data.empty()) {
		uuid::uuid d_uid, d_did;
		if (SessionBlock::deserialize(data, sid, d_uid, d_did, members, username)) {
		// Cache locally for next time
		@@ -328,6 +310,10 @@ namespace authdb {
		return true;
		}
		}
		} catch (...) {
		return false;
		}
		}

		return false;
		}
		@@ -338,23 +324,11 @@ namespace authdb {
		// Remove from local store
		store_->remove_group(sgid);

		// Remove from peers
		// Remove from all peer nodes
		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(info.parity_group_id);
		pclient_->remove(sgid);
		} catch (...) {}
		}
		}
		@@ -366,9 +340,16 @@ namespace authdb {
		std::vector<uint8_t> 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;
		// Check peer nodes
		if (pclient_) {
		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		if (pclient_->fetch_from_peers(sgid, 0, data) && !data.empty())
		return true;
		} catch (...) {}
		}

		return false;
		}

		// --- DataOpBlock serialization ---
		@@ -501,13 +482,10 @@ namespace authdb {
		for (int i = 0; i < 8; i++) localId = (localId << 8) \| hash[i];
		store_->store(localId, 0, data.data(), data.size());

		// Distribute to peers
		// Replicate to all peer nodes
		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		uint64_t pgid = pclient_->store(data);

		std::lock_guard<std::mutex> olock(ops_mutex_);
		pending_ops_.push_back({pgid, data.size()});
		pclient_->replicate_to_peers(localId, 0, data.data(), data.size());
		} catch (const netplus::NetException &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;
		} catch (const std::exception &e) {
		@@ -515,72 +493,6 @@ namespace authdb {
		}
		}

		void Cluster::applyPendingOps(AuthBackend &backend) {
		std::vector<StoredOp> ops;
		{
		std::lock_guard<std::mutex> olock(ops_mutex_);
		ops.swap(pending_ops_);
		}

		for (const auto &sop : ops) {
		std::vector<uint8_t> data;
		try {
		std::lock_guard<std::mutex> lock(client_mutex_);
		data = pclient_->retrieve(sop.parity_group_id, sop.data_size);
		} catch (...) {
		continue;
		}

		OpType op;
		std::string domain;
		uuid::uuid uid;
		int dataType;
		std::vector<AuthData::Record> records;

		if (!DataOpBlock::deserialize(data, op, domain, uid, dataType, records))
		continue;

		// Only apply if domain matches this backend
		if (domain != backend.getDomain())
		continue;

		AuthBackendLock lck(backend);

		switch (op) {
		case OpType::Create: {
		AuthData rec(uid);
		for (auto &r : records) {
		r.type = dataType;
		std::memcpy(r.ruid, uid.value, 16);
		rec.append(r);
		}
		createRecord(backend, rec, dataType);
		break;
		}
		case OpType::Edit: {
		AuthData rec(uid);
		for (auto &r : records) {
		r.type = dataType;
		std::memcpy(r.ruid, uid.value, 16);
		rec.append(r);
		}
		editRecord(backend, rec, dataType);
		break;
		}
		case OpType::Delete: {
		delRecord(backend, uid, dataType);
		break;
		}
		}

		backend.newRevesion();

		// Clean up allocated data
		for (auto &r : records) {
		delete[] r.data;
		r.data = nullptr;
		}
		}
		}

		};

src/cluster.h

+0 −19

Changes for src/cluster.h: 0 added lines, 19 removed lines.
Original line number	Diff line number	Diff line
		@@ -142,9 +142,6 @@ namespace authdb {
		const uuid::uuid &uid, int dataType,
		const AuthData *records = nullptr);

		// Apply pending operations received from peers to a backend
		void applyPendingOps(AuthBackend &backend);

		private:
		ClusterConfig cfg_;
		std::atomic<bool> running_{false};
		@@ -157,22 +154,6 @@ 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_;

		// Data operation replication tracking
		struct StoredOp {
		uint64_t parity_group_id;
		size_t data_size;
		};
		std::vector<StoredOp> pending_ops_;
		std::mutex ops_mutex_;

		void server_loop();
		};