join support added (b076e887) · Commits · tuxist / authdb

src/cluster.cpp

+94 −33

Changes for src/cluster.cpp: 94 added lines, 33 removed lines.
Original line number	Diff line number	Diff line
		@@ -39,6 +39,7 @@
		#include <crypto/sha.h>

		#include <cstring>
		#include <fstream>
		#include <iostream>
		#include <algorithm>
		#include <map>
		@@ -278,9 +279,77 @@ namespace fenrirds {
		stop();
		}

		std::string Cluster::peersFile() const {
		return cfg_.store_path.empty() ? "" : cfg_.store_path + "/.peers";
		}

		std::string Cluster::peerList() const {
		std::vector<std::string> v;
		for (const auto &p : cfg_.peers)
		v.push_back(p.address + ":" + std::to_string(p.port));
		std::sort(v.begin(), v.end());
		std::string out;
		for (const auto &s : v)
		out += s + "\n";
		return out;
		}

		void Cluster::rebalanceAsync(bool afterPeerChange) {
		std::thread([this, afterPeerChange](){
		std::unique_lock<std::mutex> lk(scrub_mutex_, std::try_to_lock);
		if (!lk.owns_lock() \|\| !scrub_client_ \|\| !scrub_client_->try_scrub_lock()) {
		if (afterPeerChange) peers_changed_ = true; // next health cycle
		return;
		}
		scrub_running_ = true;
		if (server_) server_->set_rebuilding(true);
		bool ok = false, moved = false;
		try {
		auto rb = scrub_client_->rebalance();
		moved = rb.rebalanced > 0;
		if (moved) {
		std::cerr << "Cluster: " << (afterPeerChange ? "rebalance after peer change" : "periodic rebalance")
		<< " moved " << rb.rebalanced << " group(s)" << std::endl;
		scrub_client_->vacuum_all_nodes();
		} else if (afterPeerChange) {
		std::cerr << "Cluster: rebalance after peer change: nothing to move" << std::endl;
		}
		ok = true;
		} catch (const std::exception &e) {
		std::cerr << "Cluster: rebalance failed: " << e.what() << std::endl;
		} catch (...) {}
		scrub_running_ = false;
		if (server_) server_->set_rebuilding(false);
		scrub_client_->release_scrub_lock();
		if (moved)
		recovery_epoch_.fetch_add(1); // backends re-fetch moved data
		if (afterPeerChange) {
		if (!ok) {
		peers_changed_ = true;
		} else if (!peersFile().empty()) {
		std::ofstream f(peersFile(), std::ios::trunc);
		f << peerList();
		}
		}
		}).detach();
		}

		void Cluster::init(const ClusterConfig &cfg) {
		cfg_ = cfg;

		// a node joined (or left) since the last run: the data has to be
		// spread over the new peer list. The first run of a node counts too,
		// it may be the one that just joined.
		if (!peersFile().empty()) {
		std::ifstream f(peersFile());
		std::string last((std::istreambuf_iterator<char>(f)), std::istreambuf_iterator<char>());
		if (last != peerList()) {
		peers_changed_ = true;
		std::cerr << "Cluster: peer list changed since the last run, "
		"rebalancing once all peers are online" << std::endl;
		}
		}

		// Setup local block store — file-backed if path configured, otherwise memory
		if (!cfg_.store_path.empty()) {
		std::cerr << "Cluster: using file_block_store at " << cfg_.store_path << std::endl;
		@@ -673,29 +742,21 @@ namespace fenrirds {
		degraded_ = false;
		critical_ = false;

		// After a peer change: as soon as every peer (the new one
		// included) is reachable and the domains are loaded
		if (peers_changed_ && domains_ready_.load()
		&& health.nodes_online >= health.nodes_total) {
		peers_changed_ = false; // set again if it can't run yet
		healthy_cycles = 0;
		rebalanceAsync(true);
		}

		// Periodic rebalance while healthy: fixes misplaced blocks
		// from transient store_stripe timeouts that didn't trigger
		// a full degraded→recovered transition.
		if (++healthy_cycles >= REBALANCE_INTERVAL) {
		healthy_cycles = 0;
		std::thread([this](){
		std::unique_lock<std::mutex> lk(scrub_mutex_, std::try_to_lock);
		if (!lk.owns_lock()) return;
		if (!scrub_client_ \|\| !scrub_client_->try_scrub_lock()) return;
		scrub_running_ = true;
		if (server_) server_->set_rebuilding(true);
		try {
		auto rb = scrub_client_->rebalance();
		if (rb.rebalanced > 0) {
		std::cerr << "Cluster: periodic rebalance moved "
		<< rb.rebalanced << " group(s)" << std::endl;
		scrub_client_->vacuum_all_nodes();
		}
		} catch (...) {}
		scrub_running_ = false;
		if (server_) server_->set_rebuilding(false);
		scrub_client_->release_scrub_lock();
		}).detach();
		rebalanceAsync(false);
		}
		} else if (health.nodes_online >= k) {
		if (was_critical) {
		@@ -1209,7 +1270,9 @@ namespace fenrirds {
		return result;
		}

		size_t n = cfg_.data_blocks + cfg_.parity_blocks;
		// every peer: with more peers than data+parity blocks each group
		// lives on its own subset of them (paritypp::client::node_for())
		size_t n = cfg_.peers.size();

		// With multi-store support, session data lives in store_id=1 and
		// scrub_client_ uses store_id=0 (default), so list_groups_on_node()
		@@ -1256,19 +1319,17 @@ namespace fenrirds {
		<< " data groups across " << n << " nodes"
		<< " (removed " << orphaned_groups.size() << " orphaned)" << std::endl;

		// 2. Repair under-replicated AND rebalance misplaced in one pass
		for (auto &[gid, present_nodes] : group_nodes) {
		result.groups_checked++;

		if (present_nodes.size() >= n) {
		// All nodes have this group — OK
		continue;
		}
		// 2. Repair groups missing blocks on the nodes that own them. The
		// check looks at every block (presence of the group id on a node
		// says nothing about which nodes should have it); copies left on
		// other nodes are removed by rebalance(), not here.
		auto check = scrub_client_->check_groups();
		result.groups_checked = group_nodes.size();
		for (uint64_t gid : check.incomplete) {
		if (!group_nodes.count(gid))
		continue; // orphaned (handled above) or the lock group

		// Group is under-replicated or misplaced: retrieve and re-store
		std::cerr << "[SCRUB] group " << gid << " present on "
		<< present_nodes.size() << "/" << n
		<< " nodes — repairing" << std::endl;
		std::cerr << "[SCRUB] group " << gid << " incomplete — repairing" << std::endl;

		try {
		auto data = scrub_client_->retrieve(gid);
		@@ -1300,8 +1361,8 @@ namespace fenrirds {
		continue;
		}

		// Data is valid — re-store onto the canonical nodes
		// (block_index % n == node_index). store() alone fixes
		// Data is valid — re-store onto the owning nodes
		// (paritypp::client::node_for()). store() alone fixes
		// both under-replication and misplacement — it writes
		// every block to its correct node. Do NOT remove() the
		// group first: paritypp::client::remove() is best-effort

src/cluster.h

+10 −0

Changes for src/cluster.h: 10 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -280,6 +280,16 @@ namespace fenrirds {
		std::mutex scrub_mutex_; // prevents concurrent scrub / rebalance
		std::atomic<bool> scrub_running_{false};

		// PEERS differs from the list of the last run (a node joined): the
		// health monitor rebalances once every peer is online
		std::atomic<bool> peers_changed_{false};
		std::string peersFile() const;
		std::string peerList() const;
		// runs scrub_client_->rebalance() in the background unless a scrub or
		// rebalance is running; afterPeerChange retries on the next health
		// cycle if it couldn't run and records the peer list once it did
		void rebalanceAsync(bool afterPeerChange);

		struct PendingPush {
		std::vector<uint8_t> data;
		uint64_t sid_gid;

src/nodejoin.cpp

+1 −1

Changes for src/nodejoin.cpp: 1 added line, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -481,7 +481,7 @@ namespace fenrirds::nodejoin {
		}
		restartSelf = true;
		json_object_array_add(warnings, json_object_new_string(
		"run Rebalance in the Cluster view once all nodes are back"));
		"the cluster rebalances by itself once all nodes are back online"));
		}

		put(result, "status", "ok");