test (456528dc) · Commits · tuxist / authdb

src/admin.cpp

+32 −0

Original line number	Diff line number	Diff line
		@@ -681,6 +681,36 @@ namespace authdb {
		json_object_put(jobj);
		}

		/* ---- JSON API: Cluster Scrub ---- */
		void apiClusterScrub(libhttppp::HttpRequest &curreq) {
		json_object *jobj = json_object_new_object();

		if (!g_Cluster \|\| !g_Cluster->isRunning()) {
		json_object_object_add(jobj, "status", json_object_new_string("error"));
		json_object_object_add(jobj, "message", json_object_new_string("cluster not running"));
		sendJson(curreq, jobj);
		json_object_put(jobj);
		return;
		}

		if (g_Cluster->isCritical()) {
		json_object_object_add(jobj, "status", json_object_new_string("error"));
		json_object_object_add(jobj, "message", json_object_new_string("cluster is critical — need more nodes online"));
		sendJson(curreq, jobj);
		json_object_put(jobj);
		return;
		}

		auto result = g_Cluster->scrub();

		json_object_object_add(jobj, "status", json_object_new_string("ok"));
		json_object_object_add(jobj, "groups_checked", json_object_new_int(static_cast<int>(result.groups_checked)));
		json_object_object_add(jobj, "groups_repaired", json_object_new_int(static_cast<int>(result.groups_repaired)));
		json_object_object_add(jobj, "groups_failed", json_object_new_int(static_cast<int>(result.groups_failed)));
		sendJson(curreq, jobj);
		json_object_put(jobj);
		}

		/* ---- JSON API: List Domains ---- */
		void apiListDomains(libhttppp::HttpRequest &curreq) {
		json_object *jobj = json_object_new_object();
		@@ -2112,6 +2142,8 @@ namespace authdb {
		apiClusterStatus(curreq);
		} else if (strcmp(apiurl, "clustersync") == 0) {
		apiClusterSync(curreq);
		} else if (strcmp(apiurl, "clusterscrub") == 0) {
		apiClusterScrub(curreq);
		} else {
		sendJsonError(curreq, "Unknown API endpoint", 404);
		}

src/cluster.cpp

+78 −1

Original line number	Diff line number	Diff line
		@@ -40,6 +40,7 @@
		#include <cstring>
		#include <iostream>
		#include <algorithm>
		#include <map>
		#include <set>
		#include <unordered_map>
		#include <future>
		@@ -628,6 +629,7 @@ namespace authdb {
		}

		bool was_critical = critical_;
		bool was_degraded = degraded_;
		if (!store_ok) {
		degraded_ = true;
		critical_ = true;
		@@ -635,8 +637,12 @@ namespace authdb {
		if (degraded_) {
		std::cerr << "Cluster: recovered — all "
		<< health.nodes_online << "/" << n
		<< " nodes online" << std::endl;
		<< " nodes online, starting scrub" << std::endl;
		recovery_epoch_.fetch_add(1);
		// Auto-scrub after recovery to rebalance shards
		std::thread([this](){
		try { scrub(); } catch (...) {}
		}).detach();
		}
		degraded_ = false;
		critical_ = false;
		@@ -1041,6 +1047,77 @@ namespace authdb {
		return true;
		}

		// --- Scrub: verify and repair all data groups ---

		Cluster::ScrubResult Cluster::scrub() {
		ScrubResult result;
		if (!pclient_) return result;

		size_t n = cfg_.data_blocks + cfg_.parity_blocks;

		// 1. Collect all group IDs from every node
		std::map<uint64_t, std::set<size_t>> group_nodes; // group_id -> set of node indices that have it
		for (size_t ni = 0; ni < n; ++ni) {
		std::vector<uint64_t> node_groups;
		if (pclient_->list_groups_on_node(ni, node_groups)) {
		for (uint64_t gid : node_groups)
		group_nodes[gid].insert(ni);
		}
		}

		std::cerr << "[SCRUB] found " << group_nodes.size()
		<< " groups across " << n << " nodes" << std::endl;

		// 2. For each group, check block distribution and repair
		for (auto &[gid, present_nodes] : group_nodes) {
		result.groups_checked++;

		if (present_nodes.size() >= n) {
		// All nodes have this group — OK
		continue;
		}

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

		try {
		// Retrieve reconstructs from available shards (needs >= k)
		auto data = pclient_->retrieve(gid);
		if (data.empty()) {
		std::cerr << "[SCRUB] group " << gid
		<< " — retrieve returned empty, skipping" << std::endl;
		result.groups_failed++;
		continue;
		}

		// Re-store redistributes to all nodes (overwrites existing shards,
		// fills missing ones)
		pclient_->store(gid, data.data(), data.size());
		result.groups_repaired++;

		std::cerr << "[SCRUB] group " << gid << " — repaired" << std::endl;
		} catch (const std::exception &e) {
		std::cerr << "[SCRUB] group " << gid
		<< " — repair failed: " << e.what() << std::endl;
		result.groups_failed++;
		}
		}

		// 3. Vacuum all nodes to reclaim space from overwritten blocks
		if (result.groups_repaired > 0) {
		pclient_->vacuum_all_nodes();
		recovery_epoch_.fetch_add(1);
		}

		std::cerr << "[SCRUB] complete: checked=" << result.groups_checked
		<< " repaired=" << result.groups_repaired
		<< " failed=" << result.groups_failed << std::endl;

		return result;
		}

		// --- Data replication ---

		void Cluster::replicateOp(OpType op, const std::string &domain,

src/cluster.h

+11 −0

Original line number	Diff line number	Diff line
		@@ -212,6 +212,17 @@ namespace authdb {
		// Force all domain backends to re-fetch from cluster on next access
		void forceSync() { recovery_epoch_.fetch_add(1); }

		// Scrub: verify and repair all data groups across the cluster.
		// For each group, checks that every node has its blocks. Missing blocks
		// are fetched from a peer that has them and re-replicated.
		// Returns {groups_checked, groups_repaired, groups_failed}.
		struct ScrubResult {
		size_t groups_checked = 0;
		size_t groups_repaired = 0;
		size_t groups_failed = 0;
		};
		ScrubResult scrub();

		private:
		ClusterConfig cfg_;
		std::atomic<bool> running_{false};