test (b31252cf) · Commits · tuxist / authdb

src/cluster.cpp

+210 −0

Changes for src/cluster.cpp: 210 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -345,4 +345,214 @@ namespace authdb {
		return session_index_.count(sgid) > 0;
		}

		// --- DataOpBlock serialization ---

		std::vector<uint8_t> DataOpBlock::serialize(OpType op, const std::string &domain,
		const uuid::uuid &uid, int dataType,
		const AuthData *records) {
		std::vector<uint8_t> buf;

		// Op type
		buf.push_back(static_cast<uint8_t>(op));

		// Domain
		uint32_t dlen = static_cast<uint32_t>(domain.size());
		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&dlen), reinterpret_cast<const uint8_t>(&dlen) + 4);
		buf.insert(buf.end(), domain.begin(), domain.end());

		// UID
		buf.insert(buf.end(), uid.value, uid.value + 16);

		// Data type
		uint32_t dt = static_cast<uint32_t>(dataType);
		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&dt), reinterpret_cast<const uint8_t>(&dt) + 4);

		// Records (only for Create/Edit)
		if (op == OpType::Delete \|\| !records) {
		uint32_t rc = 0;
		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&rc), reinterpret_cast<const uint8_t>(&rc) + 4);
		} else {
		// Count records
		uint32_t rc = 0;
		for (const AuthData *cur = records; cur; cur = cur->next()) rc++;

		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&rc), reinterpret_cast<const uint8_t>(&rc) + 4);

		for (const AuthData *cur = records; cur; cur = cur->next()) {
		// fieldname (255 bytes fixed)
		buf.insert(buf.end(), reinterpret_cast<const uint8_t*>(cur->Data.fieldname),
		reinterpret_cast<const uint8_t*>(cur->Data.fieldname) + 255);
		// storage type
		uint32_t st = static_cast<uint32_t>(cur->Data.storage);
		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&st), reinterpret_cast<const uint8_t>(&st) + 4);
		// data size
		uint32_t ds = static_cast<uint32_t>(cur->Data.datasize);
		buf.insert(buf.end(), reinterpret_cast<const uint8_t>(&ds), reinterpret_cast<const uint8_t>(&ds) + 4);
		// data
		if (ds > 0 && cur->Data.data) {
		buf.insert(buf.end(), reinterpret_cast<const uint8_t*>(cur->Data.data),
		reinterpret_cast<const uint8_t*>(cur->Data.data) + ds);
		}
		}
		}

		return buf;
		}

		bool DataOpBlock::deserialize(const std::vector<uint8_t> &data,
		OpType &op, std::string &domain,
		uuid::uuid &uid, int &dataType,
		std::vector<AuthData::Record> &records) {
		if (data.size() < 1 + 4) return false;
		size_t pos = 0;

		// Op type
		op = static_cast<OpType>(data[pos++]);

		// Domain
		uint32_t dlen = 0;
		std::memcpy(&dlen, data.data() + pos, 4); pos += 4;
		if (pos + dlen > data.size()) return false;
		domain.assign(reinterpret_cast<const char*>(data.data() + pos), dlen);
		pos += dlen;

		// UID
		if (pos + 16 > data.size()) return false;
		std::memcpy(uid.value, data.data() + pos, 16); pos += 16;

		// Data type
		if (pos + 4 > data.size()) return false;
		uint32_t dt = 0;
		std::memcpy(&dt, data.data() + pos, 4); pos += 4;
		dataType = static_cast<int>(dt);

		// Record count
		if (pos + 4 > data.size()) return false;
		uint32_t rc = 0;
		std::memcpy(&rc, data.data() + pos, 4); pos += 4;

		records.clear();
		for (uint32_t i = 0; i < rc; i++) {
		if (pos + 255 + 4 + 4 > data.size()) return false;
		AuthData::Record rec;
		std::memcpy(rec.fieldname, data.data() + pos, 255); pos += 255;

		uint32_t st = 0;
		std::memcpy(&st, data.data() + pos, 4); pos += 4;
		rec.storage = static_cast<int>(st);

		uint32_t ds = 0;
		std::memcpy(&ds, data.data() + pos, 4); pos += 4;
		rec.datasize = ds;

		if (ds > 0) {
		if (pos + ds > data.size()) return false;
		rec.data = new char[ds];
		std::memcpy(rec.data, data.data() + pos, ds);
		pos += ds;
		}

		std::memcpy(rec.ruid, uid.value, 16);
		rec.type = dataType;
		records.push_back(rec);
		}

		return true;
		}

		// --- Data replication ---

		void Cluster::replicateOp(OpType op, const std::string &domain,
		const uuid::uuid &uid, int dataType,
		const AuthData *records) {
		if (!pclient_) return;

		auto data = DataOpBlock::serialize(op, domain, uid, dataType, records);

		// Store locally for peers to fetch
		auto hash = netplus::sha256_hash(data);
		uint64_t localId = 0;
		for (int i = 0; i < 8; i++) localId = (localId << 8) \| hash[i];
		store_->store(localId, 0, data.data(), data.size());

		// Distribute to peers
		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()});
		} catch (const std::exception &e) {
		std::cerr << "Cluster replicate failed: " << e.what() << std::endl;
		}
		}

		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

+46 −7

Changes for src/cluster.h: 46 added lines, 7 removed lines.
Original line number	Diff line number	Diff line
		@@ -40,6 +40,8 @@
		#include <paritypp/auth.h>

		#include "session.h"
		#include "backend.h"
		#include "authdb.h"

		namespace authdb {

		@@ -60,23 +62,44 @@ namespace authdb {
		size_t parity_blocks = 1; // m: parity shards (fault tolerance)
		};

		// Serialized session block for cluster transport
		// --- Session serialization ---

		struct SessionBlock {
		// Serialize a SessionData into a byte buffer
		static std::vector<uint8_t> serialize(const SessionData &sess);

		// Deserialize a byte buffer back into SessionData fields
		static bool deserialize(const std::vector<uint8_t> &data,
		uuid::uuid &sid, uuid::uuid &uid, uuid::uuid &did,
		std::vector<uuid::uuid> &members,
		std::string &username);
		};

		// Computes a deterministic session group_id from uid+did for deduplication.
		// Same user + same domain = same group_id across all nodes.
		uint64_t sessionGroupId(const uuid::uuid &uid, const uuid::uuid &did);

		// Cluster manager: distributes and retrieves sessions across peer nodes
		// --- Data operation replication ---

		enum class OpType : uint8_t {
		Create = 1,
		Edit = 2,
		Delete = 3
		};

		// Serialized data operation for cluster transport
		// Wire format:
		// [1B op_type][4B domain_len][domain_bytes]
		// [16B ruid][4B data_type]
		// [4B record_count]
		// for each record:
		// [255B fieldname][4B storage][4B datasize][datasize bytes]
		struct DataOpBlock {
		static std::vector<uint8_t> serialize(OpType op, const std::string &domain,
		const uuid::uuid &uid, int dataType,
		const AuthData *records);
		static bool deserialize(const std::vector<uint8_t> &data,
		OpType &op, std::string &domain,
		uuid::uuid &uid, int &dataType,
		std::vector<AuthData::Record> &records);
		};

		// Cluster manager: distributes sessions and data operations across peer nodes
		class Cluster {
		public:
		Cluster();
		@@ -108,6 +131,14 @@ namespace authdb {
		// Check if a session already exists in the cluster for uid+did
		bool hasSession(const uuid::uuid &uid, const uuid::uuid &did);

		// Replicate a data operation (create/edit/delete) to peers
		void replicateOp(OpType op, const std::string &domain,
		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};
		@@ -128,6 +159,14 @@ namespace authdb {
		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();
		};

src/data.cpp

+25 −2

Changes for src/data.cpp: 25 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@
		#include <iostream>

		#include "authdb.h"
		#include "cluster.h"

		namespace authdb {
		AuthData::AuthData(const uuid::uuid &id){
		@@ -111,6 +112,15 @@ namespace authdb {
		}

		void createRecord(authdb::AuthBackend &backend,AuthData &rec,int type){
		// If cluster is active, replicate only — don't write local file
		if (g_Cluster && g_Cluster->isRunning()) {
		for(AuthData* curec=&rec; curec; curec=curec->next())
		curec->Data.type=type;
		g_Cluster->replicateOp(OpType::Create, backend.getDomain(),
		uuid::uuid(rec.Data.ruid), type, &rec);
		return;
		}

		size_t wd=backend.end();

		for(AuthData* curec=&rec; curec; curec=curec->next()){
		@@ -123,6 +133,13 @@ namespace authdb {
		}

		bool editRecord(AuthBackend &backend,AuthData &rec,int type){
		// If cluster is active, replicate only — don't write local file
		if (g_Cluster && g_Cluster->isRunning()) {
		g_Cluster->replicateOp(OpType::Edit, backend.getDomain(),
		uuid::uuid(rec.Data.ruid), type, &rec);
		return true;
		}

		AuthData old(rec.Data.ruid);

		std::vector<AuthData::Record*> changemap; //use fieldname;
		@@ -205,11 +222,17 @@ namespace authdb {
		}

		void delRecord(AuthBackend &backend,const uuid::uuid &uid,int type){
		size_t end=backend.end(),rd=sizeof(AuthHeader);

		if(uid.empty())
		throw AuthBackendError("user info uid required!");

		// If cluster is active, replicate only — don't write local file
		if (g_Cluster && g_Cluster->isRunning()) {
		g_Cluster->replicateOp(OpType::Delete, backend.getDomain(), uid, type);
		return;
		}

		size_t end=backend.end(),rd=sizeof(AuthHeader);

		while(rd<end){
		std::shared_ptr<authdb::AuthData::Record> cur=std::make_shared<authdb::AuthData::Record>();
		backend.setPos(rd);