test (21cc8513) · Commits · tuxist / authdb

src/backends/cluster.cpp

+1 −2

Changes for src/backends/cluster.cpp: 1 added line, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -265,9 +265,8 @@ namespace authdb {
		if (client) {
		try {
		auto &cmtx = g_Cluster->getClientMutex();
		// Retry up to 3 times to avoid silently dropping replication
		for (int attempt = 0; attempt < 3; ++attempt) {
		std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::seconds(3));
		std::unique_lock<std::timed_mutex> lock(cmtx, std::chrono::seconds(1));
		if (lock.owns_lock()) {
		client->store(dgid, buf.data(), buf.size());
		return true;

src/cluster.cpp

+36 −21

Changes for src/cluster.cpp: 36 added lines, 21 removed lines.
Original line number	Diff line number	Diff line
		@@ -657,18 +657,34 @@ namespace authdb {
		// Don't try to push while critical
		if (critical_) continue;

		// Drain data pushes first (coalesce: only latest per domain)
		// Grab pending items
		PendingDataPush dp_item;
		bool have_dp = false;
		if (!data_push_queue_.empty()) {
		PendingDataPush dp = std::move(data_push_queue_.front());
		dp_item = std::move(data_push_queue_.front());
		data_push_queue_.pop_front();
		have_dp = true;
		}

		PendingPush sess_item;
		bool have_sess = false;
		if (!push_queue_.empty()) {
		sess_item = std::move(push_queue_.front());
		push_queue_.pop_front();
		have_sess = true;
		}

		lk.unlock();

		// Launch data push and session push in parallel
		std::future<void> data_future;
		if (have_dp) {
		data_future = std::async(std::launch::async, [this, dp = std::move(dp_item)]() mutable {
		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(3));
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(1));
		if (lock.owns_lock()) {
		pclient_->store(dp.dgid, dp.buf.data(), dp.buf.size());
		} else {
		// Re-queue (will be coalesced if newer push arrives)
		std::lock_guard<std::mutex> pguard(push_mutex_);
		data_push_queue_.push_back(std::move(dp));
		}
		@@ -677,43 +693,42 @@ namespace authdb {
		std::lock_guard<std::mutex> pguard(push_mutex_);
		data_push_queue_.push_back(std::move(dp));
		}
		continue;
		});
		}

		// Session pushes
		if (push_queue_.empty()) continue;

		// Take one item
		PendingPush item = std::move(push_queue_.front());
		push_queue_.pop_front();
		lk.unlock();

		std::future<void> sess_future;
		if (have_sess) {
		sess_future = std::async(std::launch::async, [this, item = std::move(sess_item)]() mutable {
		try {
		std::unique_lock<std::timed_mutex> lock(client_mutex_, std::chrono::seconds(3));
		std::unique_lock<std::timed_mutex> lock(session_client_mutex_, std::chrono::seconds(1));
		if (!lock.owns_lock()) {
		// Re-queue
		item.retries++;
		if (item.retries < 5) {
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(item));
		} else {
		std::cerr << "[CLUSTER] push_worker: dropped after 5 retries" << std::endl;
		std::cerr << "[CLUSTER] push_worker: dropped session after 5 retries" << std::endl;
		}
		continue;
		return;
		}
		pclient_->store(item.sgid, item.data.data(), item.data.size());
		pclient_->store(item.sid_gid, item.data.data(), item.data.size());
		std::cerr << "[CLUSTER] push_worker: pushed queued session" << std::endl;
		session_client_->store(item.sgid, item.data.data(), item.data.size());
		session_client_->store(item.sid_gid, item.data.data(), item.data.size());
		} catch (const std::exception &e) {
		item.retries++;
		if (item.retries < 5) {
		std::lock_guard<std::mutex> pguard(push_mutex_);
		push_queue_.push_back(std::move(item));
		} else {
		std::cerr << "[CLUSTER] push_worker: dropped after 5 retries: "
		std::cerr << "[CLUSTER] push_worker: dropped session after 5 retries: "
		<< e.what() << std::endl;
		}
		}
		});
		}

		// Wait for both to finish
		if (data_future.valid()) data_future.get();
		if (sess_future.valid()) sess_future.get();
		}
		}