From 789a75abfa2af0af39616c69575882e5db2b6f07 Mon Sep 17 00:00:00 2001
From: Martin Schwenke <martin@meltin.net>
Date: Mon, 16 Mar 2020 16:07:26 +1100
Subject: [PATCH] ctdb-recoverd: Process leader broadcasts

Signed-off-by: Martin Schwenke <martin@meltin.net>
Reviewed-by: Amitay Isaacs <amitay@gmail.com>
---
 ctdb/server/ctdb_recoverd.c | 43 +++++++++++++++++++++++++++++++++++++
 1 file changed, 43 insertions(+)

diff --git a/ctdb/server/ctdb_recoverd.c b/ctdb/server/ctdb_recoverd.c
index ab8074c8efd..2425ac5433a 100644
--- a/ctdb/server/ctdb_recoverd.c
+++ b/ctdb/server/ctdb_recoverd.c
@@ -38,6 +38,8 @@
 #include "ctdb_private.h"
 #include "ctdb_client.h"
 
+#include "protocol/protocol_basic.h"
+
 #include "common/system_socket.h"
 #include "common/common.h"
 #include "common/logging.h"
@@ -1845,6 +1847,42 @@ static void push_flags_handler(uint64_t srvid, TDB_DATA data,
 	talloc_free(tmp_ctx);
 }
 
+static void leader_handler(uint64_t srvid, TDB_DATA data, void *private_data)
+{
+	struct ctdb_recoverd *rec = talloc_get_type_abort(
+		private_data, struct ctdb_recoverd);
+	struct ctdb_context *ctdb = rec->ctdb;
+	uint32_t pnn;
+	size_t npull;
+	int ret;
+
+	ret = ctdb_uint32_pull(data.dptr, data.dsize, &pnn, &npull);
+	if (ret != 0) {
+		DBG_WARNING("Unable to parse leader broadcast, ret=%d\n", ret);
+		return;
+	}
+
+	if (pnn == rec->leader) {
+		return;
+	}
+
+	if (pnn == CTDB_UNKNOWN_PNN) {
+		return;
+	}
+
+	D_NOTICE("Received leader broadcast, leader=%"PRIu32"\n", pnn);
+
+	ret = ctdb_ctrl_setrecmaster(ctdb,
+				     CONTROL_TIMEOUT(),
+				     CTDB_CURRENT_NODE,
+				     pnn);
+	if (ret != 0) {
+		DBG_WARNING("Failed to set leader\n");
+		return;
+	}
+
+	rec->leader = pnn;
+}
 
 struct verify_recmode_normal_data {
 	uint32_t count;
@@ -3012,6 +3050,11 @@ static void monitor_cluster(struct ctdb_context *ctdb)
 					CTDB_SRVID_DISABLE_RECOVERIES,
 					disable_recoveries_handler, rec);
 
+	ctdb_client_set_message_handler(ctdb,
+					CTDB_SRVID_LEADER,
+					leader_handler,
+					rec);
+
 	for (;;) {
 		TALLOC_CTX *mem_ctx = talloc_new(ctdb);
 		struct timeval start;