test (b48d513a) · Commits · tuxist / authdb

src/cluster.cpp

+61 −8

Changes for src/cluster.cpp: 61 added lines, 8 removed lines.
Original line number	Diff line number	Diff line
		@@ -272,27 +272,66 @@ namespace authdb {
		std::shared_ptr<paritypp::memory_block_store> session_store)
		: inner_(std::move(inner)), session_store_(std::move(session_store)) {}

		void session_intercepting_store::mark_session_group(uint64_t group_id) {
		std::lock_guard<std::mutex> lock(sg_mutex_);
		session_groups_.insert(group_id);
		}

		bool session_intercepting_store::is_session_group(uint64_t group_id) {
		std::lock_guard<std::mutex> lock(sg_mutex_);
		return session_groups_.count(group_id) > 0;
		}

		bool session_intercepting_store::store(uint64_t group_id, uint32_t block_index,
		const uint8_t* data, size_t len) {
		if (is_session_group(group_id)) {
		// Session data goes ONLY to the in-memory session store
		return session_store_->store(group_id, block_index, data, len);
		}
		return inner_->store(group_id, block_index, data, len);
		}

		bool session_intercepting_store::fetch(uint64_t group_id, uint32_t block_index,
		std::vector<uint8_t>& out) {
		if (is_session_group(group_id)) {
		return session_store_->fetch(group_id, block_index, out);
		}
		return inner_->fetch(group_id, block_index, out);
		}

		bool session_intercepting_store::remove_group(uint64_t group_id) {
		if (is_session_group(group_id)) {
		std::lock_guard<std::mutex> lock(sg_mutex_);
		session_groups_.erase(group_id);
		return session_store_->remove_group(group_id);
		}
		return inner_->remove_group(group_id);
		}

		std::vector<uint32_t> session_intercepting_store::list_blocks(uint64_t group_id) {
		return inner_->list_blocks(group_id);
		// Merge blocks from both stores
		auto inner_blocks = inner_->list_blocks(group_id);
		auto sess_blocks = session_store_->list_blocks(group_id);
		if (sess_blocks.empty()) return inner_blocks;
		if (inner_blocks.empty()) return sess_blocks;
		std::set<uint32_t> merged(inner_blocks.begin(), inner_blocks.end());
		merged.insert(sess_blocks.begin(), sess_blocks.end());
		return std::vector<uint32_t>(merged.begin(), merged.end());
		}

		uint32_t session_intercepting_store::total_blocks() { return inner_->total_blocks(); }
		uint32_t session_intercepting_store::total_groups() { return inner_->total_groups(); }
		std::vector<uint64_t> session_intercepting_store::list_groups() { return inner_->list_groups(); }
		uint32_t session_intercepting_store::total_blocks() {
		return inner_->total_blocks() + session_store_->total_blocks();
		}
		uint32_t session_intercepting_store::total_groups() {
		return inner_->total_groups() + session_store_->total_groups();
		}
		std::vector<uint64_t> session_intercepting_store::list_groups() {
		auto ig = inner_->list_groups();
		auto sg = session_store_->list_groups();
		std::set<uint64_t> merged(ig.begin(), ig.end());
		merged.insert(sg.begin(), sg.end());
		return std::vector<uint64_t>(merged.begin(), merged.end());
		}

		// --- Cluster ---

		@@ -334,9 +373,9 @@ namespace authdb {
		scfg.require_auth = !cfg_.client_name.empty();

		// Wrap the store with session interceptor so incoming replicated sessions
		// are automatically mirrored into the dedicated session_store_
		auto intercepted = std::make_shared<session_intercepting_store>(store_, session_store_);
		server_ = std::make_unique<paritypp::server>(scfg, intercepted, auth_);
		// are automatically routed to the dedicated session_store_
		interceptor_ = std::make_shared<session_intercepting_store>(store_, session_store_);
		server_ = std::make_unique<paritypp::server>(scfg, interceptor_, auth_);

		// Setup client for talking to peers — filter out self
		size_t peer_count = 0;
		@@ -411,7 +450,7 @@ namespace authdb {
		session_client_ = std::make_unique<paritypp::client>(
		cfg_.data_blocks, cfg_.parity_blocks, nodes, pcreds);
		if (self_idx != SIZE_MAX) {
		session_client_->set_local_node(self_idx, store_);
		session_client_->set_local_node(self_idx, session_store_);
		}
		} else {
		std::cerr << "Cluster: no peers configured" << std::endl;
		@@ -429,6 +468,15 @@ namespace authdb {
		void Cluster::start() {
		if (running_) return;
		running_ = true;

		// Vacuum file store at startup to reclaim space from old session data
		// that was previously written to blocks.bin instead of session_store_
		if (store_) {
		if (store_->vacuum()) {
		std::cerr << "Cluster: startup vacuum completed" << std::endl;
		}
		}

		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);
		@@ -587,6 +635,8 @@ namespace authdb {
		sess.getSid(sid);
		uint64_t sid_gid = sidGroupId(sid);

		if (interceptor_) interceptor_->mark_session_group(sid_gid);

		for (int attempt = 0; attempt < 3; ++attempt) {
		try {
		session_client_->store(sid_gid, data.data(), data.size());
		@@ -605,6 +655,7 @@ namespace authdb {
		uuid::uuid sid;
		sess.getSid(sid);
		uint64_t sid_gid = sidGroupId(sid);
		if (interceptor_) interceptor_->mark_session_group(sid_gid);
		PendingPush pp{data, sid_gid, 0};
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(pp));
		@@ -724,6 +775,8 @@ namespace authdb {

		int removed = 0;
		for (uint64_t gid : all_groups) {
		// Register this group as a session so the interceptor routes it correctly
		if (interceptor_) interceptor_->mark_session_group(gid);
		try {
		auto data = session_client_->retrieve(gid);
		if (data.empty()) continue;

src/cluster.h

+10 −4

Changes for src/cluster.h: 10 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -30,6 +30,7 @@
		#include <string>
		#include <vector>
		#include <deque>
		#include <set>
		#include <memory>
		#include <mutex>
		#include <condition_variable>
		@@ -53,15 +54,17 @@ namespace authdb {
		int port;
		};

		// Wraps a real block_store and mirrors session blocks into a separate session store.
		// When a peer replicates a session to us via the parity server, the server
		// calls store() on this wrapper — if the data deserializes as a valid session,
		// it is also written to session_store_.
		// Wraps a real block_store and routes session blocks to a separate in-memory
		// session store, while domain data goes to the file-backed inner store.
		// Session group IDs must be registered via mark_session_group() before use.
		class session_intercepting_store : public paritypp::block_store {
		public:
		session_intercepting_store(std::shared_ptr<paritypp::block_store> inner,
		std::shared_ptr<paritypp::memory_block_store> session_store);

		void mark_session_group(uint64_t group_id);
		bool is_session_group(uint64_t group_id);

		bool store(uint64_t group_id, uint32_t block_index,
		const uint8_t* data, size_t len) override;
		bool fetch(uint64_t group_id, uint32_t block_index,
		@@ -75,6 +78,8 @@ namespace authdb {
		private:
		std::shared_ptr<paritypp::block_store> inner_;
		std::shared_ptr<paritypp::memory_block_store> session_store_;
		std::mutex sg_mutex_;
		std::set<uint64_t> session_groups_;
		};

		struct ClusterConfig {
		@@ -233,6 +238,7 @@ namespace authdb {

		std::shared_ptr<paritypp::block_store> store_;
		std::shared_ptr<paritypp::memory_block_store> session_store_; // sessions only
		std::shared_ptr<session_intercepting_store> interceptor_; // server store wrapper
		std::shared_ptr<paritypp::auth_store> auth_;
		std::unique_ptr<paritypp::server> server_;