Loading src/cluster.cpp +48 −1 Changes for src/cluster.cpp: 48 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -232,6 +232,50 @@ namespace authdb { return id; } // --- session_intercepting_store --- session_intercepting_store::session_intercepting_store( std::shared_ptr<paritypp::block_store> inner, std::shared_ptr<paritypp::memory_block_store> session_store) : inner_(std::move(inner)), session_store_(std::move(session_store)) {} bool session_intercepting_store::store(uint64_t group_id, uint32_t block_index, const uint8_t* data, size_t len) { bool ok = inner_->store(group_id, block_index, data, len); // Try to detect if this is a session block and mirror it if (ok && len >= 56) { std::vector<uint8_t> buf(data, data + len); uuid::uuid sid, uid, did; std::vector<uuid::uuid> members; std::string username; std::vector<std::pair<uuid::uuid, bool>> gpo; if (SessionBlock::deserialize(buf, sid, uid, did, members, username, gpo) && !sid.empty() && !uid.empty()) { session_store_->store(group_id, block_index, data, len); } } return ok; } bool session_intercepting_store::fetch(uint64_t group_id, uint32_t block_index, std::vector<uint8_t>& out) { return inner_->fetch(group_id, block_index, out); } bool session_intercepting_store::remove_group(uint64_t group_id) { 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); } 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(); } // --- Cluster --- Cluster::Cluster() {} Loading Loading @@ -271,7 +315,10 @@ namespace authdb { scfg.max_connections = 64; scfg.require_auth = !cfg_.client_name.empty(); server_ = std::make_unique<paritypp::server>(scfg, store_, auth_); // 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_); // Setup client for talking to peers — filter out self size_t peer_count = 0; Loading src/cluster.h +24 −0 Changes for src/cluster.h: 24 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -51,6 +51,30 @@ 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_. 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); 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, std::vector<uint8_t>& out) override; bool remove_group(uint64_t group_id) override; std::vector<uint32_t> list_blocks(uint64_t group_id) override; uint32_t total_blocks() override; uint32_t total_groups() override; std::vector<uint64_t> list_groups() override; private: std::shared_ptr<paritypp::block_store> inner_; std::shared_ptr<paritypp::memory_block_store> session_store_; }; struct ClusterConfig { std::vector<ClusterNode> peers; std::string bind_address = "0.0.0.0"; Loading Loading
src/cluster.cpp +48 −1 Changes for src/cluster.cpp: 48 added lines, 1 removed line. Original line number Diff line number Diff line Loading @@ -232,6 +232,50 @@ namespace authdb { return id; } // --- session_intercepting_store --- session_intercepting_store::session_intercepting_store( std::shared_ptr<paritypp::block_store> inner, std::shared_ptr<paritypp::memory_block_store> session_store) : inner_(std::move(inner)), session_store_(std::move(session_store)) {} bool session_intercepting_store::store(uint64_t group_id, uint32_t block_index, const uint8_t* data, size_t len) { bool ok = inner_->store(group_id, block_index, data, len); // Try to detect if this is a session block and mirror it if (ok && len >= 56) { std::vector<uint8_t> buf(data, data + len); uuid::uuid sid, uid, did; std::vector<uuid::uuid> members; std::string username; std::vector<std::pair<uuid::uuid, bool>> gpo; if (SessionBlock::deserialize(buf, sid, uid, did, members, username, gpo) && !sid.empty() && !uid.empty()) { session_store_->store(group_id, block_index, data, len); } } return ok; } bool session_intercepting_store::fetch(uint64_t group_id, uint32_t block_index, std::vector<uint8_t>& out) { return inner_->fetch(group_id, block_index, out); } bool session_intercepting_store::remove_group(uint64_t group_id) { 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); } 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(); } // --- Cluster --- Cluster::Cluster() {} Loading Loading @@ -271,7 +315,10 @@ namespace authdb { scfg.max_connections = 64; scfg.require_auth = !cfg_.client_name.empty(); server_ = std::make_unique<paritypp::server>(scfg, store_, auth_); // 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_); // Setup client for talking to peers — filter out self size_t peer_count = 0; Loading
src/cluster.h +24 −0 Changes for src/cluster.h: 24 added lines, 0 removed lines. Original line number Diff line number Diff line Loading @@ -51,6 +51,30 @@ 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_. 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); 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, std::vector<uint8_t>& out) override; bool remove_group(uint64_t group_id) override; std::vector<uint32_t> list_blocks(uint64_t group_id) override; uint32_t total_blocks() override; uint32_t total_groups() override; std::vector<uint64_t> list_groups() override; private: std::shared_ptr<paritypp::block_store> inner_; std::shared_ptr<paritypp::memory_block_store> session_store_; }; struct ClusterConfig { std::vector<ClusterNode> peers; std::string bind_address = "0.0.0.0"; Loading