Adding upstream version 2:4.17.12+dfsg.upstream/2%4.17.12+dfsgupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 3200 insertions, 0 deletions

diff --git a/ctdb/server/ctdb_recovery_helper.c b/ctdb/server/ctdb_recovery_helper.c
new file mode 100644
index 0000000..e0d3219
--- /dev/null
+++ b/ctdb/server/ctdb_recovery_helper.c
@@ -0,0 +1,3200 @@
+/*
+   ctdb parallel database recovery
+
+   Copyright (C) Amitay Isaacs  2015
+
+   This program is free software; you can redistribute it and/or modify
+   it under the terms of the GNU General Public License as published by
+   the Free Software Foundation; either version 3 of the License, or
+   (at your option) any later version.
+
+   This program is distributed in the hope that it will be useful,
+   but WITHOUT ANY WARRANTY; without even the implied warranty of
+   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+   GNU General Public License for more details.
+
+   You should have received a copy of the GNU General Public License
+   along with this program; if not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "replace.h"
+#include "system/network.h"
+#include "system/filesys.h"
+
+#include <talloc.h>
+#include <tevent.h>
+#include <tdb.h>
+#include <libgen.h>
+
+#include "lib/tdb_wrap/tdb_wrap.h"
+#include "lib/util/dlinklist.h"
+#include "lib/util/sys_rw.h"
+#include "lib/util/time.h"
+#include "lib/util/tevent_unix.h"
+#include "lib/util/util.h"
+#include "lib/util/smb_strtox.h"
+
+#include "protocol/protocol.h"
+#include "protocol/protocol_api.h"
+#include "client/client.h"
+
+#include "common/logging.h"
+
+static int recover_timeout = 30;
+
+#define NUM_RETRIES	3
+
+#define TIMEOUT()	timeval_current_ofs(recover_timeout, 0)
+
+/*
+ * Utility functions
+ */
+
+static bool generic_recv(struct tevent_req *req, int *perr)
+{
+	int err;
+
+	if (tevent_req_is_unix_error(req, &err)) {
+		if (perr != NULL) {
+			*perr = err;
+		}
+		return false;
+	}
+
+	return true;
+}
+
+static uint64_t rec_srvid = CTDB_SRVID_RECOVERY;
+
+static uint64_t srvid_next(void)
+{
+	rec_srvid += 1;
+	return rec_srvid;
+}
+
+/*
+ * Node related functions
+ */
+
+struct node_list {
+	uint32_t *pnn_list;
+	uint32_t *caps;
+	uint32_t *ban_credits;
+	unsigned int size;
+	unsigned int count;
+};
+
+static struct node_list *node_list_init(TALLOC_CTX *mem_ctx, unsigned int size)
+{
+	struct node_list *nlist;
+	unsigned int i;
+
+	nlist = talloc_zero(mem_ctx, struct node_list);
+	if (nlist == NULL) {
+		return NULL;
+	}
+
+	nlist->pnn_list = talloc_array(nlist, uint32_t, size);
+	nlist->caps = talloc_zero_array(nlist, uint32_t, size);
+	nlist->ban_credits = talloc_zero_array(nlist, uint32_t, size);
+
+	if (nlist->pnn_list == NULL ||
+	    nlist->caps == NULL ||
+	    nlist->ban_credits == NULL) {
+		talloc_free(nlist);
+		return NULL;
+	}
+	nlist->size = size;
+
+	for (i=0; i<nlist->size; i++) {
+		nlist->pnn_list[i] = CTDB_UNKNOWN_PNN;
+	}
+
+	return nlist;
+}
+
+static bool node_list_add(struct node_list *nlist, uint32_t pnn)
+{
+	unsigned int i;
+
+	if (nlist->count == nlist->size) {
+		return false;
+	}
+
+	for (i=0; i<nlist->count; i++) {
+		if (nlist->pnn_list[i] == pnn) {
+			return false;
+		}
+	}
+
+	nlist->pnn_list[nlist->count] = pnn;
+	nlist->count += 1;
+
+	return true;
+}
+
+static uint32_t *node_list_lmaster(struct node_list *nlist,
+				   TALLOC_CTX *mem_ctx,
+				   unsigned int *pnn_count)
+{
+	uint32_t *pnn_list;
+	unsigned int count, i;
+
+	pnn_list = talloc_zero_array(mem_ctx, uint32_t, nlist->count);
+	if (pnn_list == NULL) {
+		return NULL;
+	}
+
+	count = 0;
+	for (i=0; i<nlist->count; i++) {
+		if (!(nlist->caps[i] & CTDB_CAP_LMASTER)) {
+			continue;
+		}
+
+		pnn_list[count] = nlist->pnn_list[i];
+		count += 1;
+	}
+
+	*pnn_count = count;
+	return pnn_list;
+}
+
+static void node_list_ban_credits(struct node_list *nlist, uint32_t pnn)
+{
+	unsigned int i;
+
+	for (i=0; i<nlist->count; i++) {
+		if (nlist->pnn_list[i] == pnn) {
+			nlist->ban_credits[i] += 1;
+			break;
+		}
+	}
+}
+
+/*
+ * Database list functions
+ *
+ * Simple, naive implementation that could be updated to a db_hash or similar
+ */
+
+struct db {
+	struct db *prev, *next;
+
+	uint32_t db_id;
+	uint32_t db_flags;
+	uint32_t *pnn_list;
+	unsigned int num_nodes;
+};
+
+struct db_list {
+	unsigned int num_dbs;
+	struct db *db;
+	unsigned int num_nodes;
+};
+
+static struct db_list *db_list_init(TALLOC_CTX *mem_ctx, unsigned int num_nodes)
+{
+	struct db_list *l;
+
+	l = talloc_zero(mem_ctx, struct db_list);
+	l->num_nodes = num_nodes;
+
+	return l;
+}
+
+static struct db *db_list_find(struct db_list *dblist, uint32_t db_id)
+{
+	struct db *db;
+
+	if (dblist == NULL) {
+		return NULL;
+	}
+
+	db = dblist->db;
+	while (db != NULL && db->db_id != db_id) {
+		db = db->next;
+	}
+
+	return db;
+}
+
+static int db_list_add(struct db_list *dblist,
+		       uint32_t db_id,
+		       uint32_t db_flags,
+		       uint32_t node)
+{
+	struct db *db = NULL;
+
+	if (dblist == NULL) {
+		return EINVAL;
+	}
+
+	db = talloc_zero(dblist, struct db);
+	if (db == NULL) {
+		return ENOMEM;
+	}
+
+	db->db_id = db_id;
+	db->db_flags = db_flags;
+	db->pnn_list = talloc_zero_array(db, uint32_t, dblist->num_nodes);
+	if (db->pnn_list == NULL) {
+		talloc_free(db);
+		return ENOMEM;
+	}
+	db->pnn_list[0] = node;
+	db->num_nodes = 1;
+
+	DLIST_ADD_END(dblist->db, db);
+	dblist->num_dbs++;
+
+	return 0;
+}
+
+static int db_list_check_and_add(struct db_list *dblist,
+		       uint32_t db_id,
+		       uint32_t db_flags,
+		       uint32_t node)
+{
+	struct db *db = NULL;
+	int ret;
+
+	/*
+	 * These flags are masked out because they are only set on a
+	 * node when a client attaches to that node, so they might not
+	 * be set yet.  They can't be passed as part of the attch, so
+	 * they're no use here.
+	 */
+	db_flags &= ~(CTDB_DB_FLAGS_READONLY | CTDB_DB_FLAGS_STICKY);
+
+	if (dblist == NULL) {
+		return EINVAL;
+	}
+
+	db = db_list_find(dblist, db_id);
+	if (db == NULL) {
+		ret = db_list_add(dblist, db_id, db_flags, node);
+		return ret;
+	}
+
+	if (db->db_flags != db_flags) {
+		D_ERR("Incompatible database flags for 0x%"PRIx32" "
+		      "(0x%"PRIx32" != 0x%"PRIx32")\n",
+		      db_id,
+		      db_flags,
+		      db->db_flags);
+		return EINVAL;
+	}
+
+	if (db->num_nodes >= dblist->num_nodes) {
+		return EINVAL;
+	}
+
+	db->pnn_list[db->num_nodes] = node;
+	db->num_nodes++;
+
+	return 0;
+}
+
+/*
+ * Create database on nodes where it is missing
+ */
+
+struct db_create_missing_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+
+	struct node_list *nlist;
+
+	const char *db_name;
+	uint32_t *missing_pnn_list;
+	int missing_num_nodes;
+};
+
+static void db_create_missing_done(struct tevent_req *subreq);
+
+static struct tevent_req *db_create_missing_send(
+					TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct ctdb_client_context *client,
+					struct node_list *nlist,
+					const char *db_name,
+					struct db *db)
+{
+	struct tevent_req *req, *subreq;
+	struct db_create_missing_state *state;
+	struct ctdb_req_control request;
+	unsigned int i, j;
+
+	req = tevent_req_create(mem_ctx,
+				&state,
+				struct db_create_missing_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->nlist = nlist;
+	state->db_name = db_name;
+
+	if (nlist->count == db->num_nodes) {
+		tevent_req_done(req);
+		return tevent_req_post(req, ev);
+	}
+
+	state->missing_pnn_list = talloc_array(mem_ctx, uint32_t, nlist->count);
+	if (tevent_req_nomem(state->missing_pnn_list, req)) {
+		return tevent_req_post(req, ev);
+	}
+
+	for (i = 0; i < nlist->count; i++) {
+		uint32_t pnn = nlist->pnn_list[i] ;
+
+		for (j = 0; j < db->num_nodes; j++) {
+			if (pnn == db->pnn_list[j]) {
+				break;
+			}
+		}
+
+		if (j < db->num_nodes) {
+			continue;
+		}
+
+		DBG_INFO("Create database %s on node %u\n",
+			 state->db_name,
+			 pnn);
+		state->missing_pnn_list[state->missing_num_nodes] = pnn;
+		state->missing_num_nodes++;
+	}
+
+	if (db->db_flags & CTDB_DB_FLAGS_PERSISTENT) {
+		ctdb_req_control_db_attach_persistent(&request, db_name);
+	} else if (db->db_flags & CTDB_DB_FLAGS_REPLICATED) {
+		ctdb_req_control_db_attach_replicated(&request, db_name);
+	} else {
+		ctdb_req_control_db_attach(&request, db_name);
+	}
+	request.flags = CTDB_CTRL_FLAG_ATTACH_RECOVERY;
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->missing_pnn_list,
+						state->missing_num_nodes,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, db_create_missing_done, req);
+
+	return req;
+}
+
+static void db_create_missing_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct db_create_missing_state *state = tevent_req_data(
+		req, struct db_create_missing_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq,
+						&ret,
+						NULL,
+						&err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(
+						state->missing_pnn_list,
+						state->missing_num_nodes,
+						err_list,
+						&pnn);
+		if (ret2 != 0) {
+			D_ERR("control DB_ATTACH failed for db %s"
+			      " on node %u, ret=%d\n",
+			      state->db_name,
+			      pnn,
+			      ret2);
+			node_list_ban_credits(state->nlist, pnn);
+		} else {
+			D_ERR("control DB_ATTACH failed for db %s, ret=%d\n",
+			      state->db_name,
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static bool db_create_missing_recv(struct tevent_req *req, int *perr)
+{
+	return generic_recv(req, perr);
+}
+
+/*
+ * Recovery database functions
+ */
+
+struct recdb_context {
+	uint32_t db_id;
+	const char *db_name;
+	const char *db_path;
+	struct tdb_wrap *db;
+	bool persistent;
+};
+
+static struct recdb_context *recdb_create(TALLOC_CTX *mem_ctx, uint32_t db_id,
+					  const char *db_name,
+					  const char *db_path,
+					  uint32_t hash_size, bool persistent)
+{
+	static char *db_dir_state = NULL;
+	struct recdb_context *recdb;
+	unsigned int tdb_flags;
+
+	recdb = talloc(mem_ctx, struct recdb_context);
+	if (recdb == NULL) {
+		return NULL;
+	}
+
+	if (db_dir_state == NULL) {
+		db_dir_state = getenv("CTDB_DBDIR_STATE");
+	}
+
+	recdb->db_name = db_name;
+	recdb->db_id = db_id;
+	recdb->db_path = talloc_asprintf(recdb, "%s/recdb.%s",
+					 db_dir_state != NULL ?
+					    db_dir_state :
+					    dirname(discard_const(db_path)),
+					 db_name);
+	if (recdb->db_path == NULL) {
+		talloc_free(recdb);
+		return NULL;
+	}
+	unlink(recdb->db_path);
+
+	tdb_flags = TDB_NOLOCK | TDB_INCOMPATIBLE_HASH | TDB_DISALLOW_NESTING;
+	recdb->db = tdb_wrap_open(mem_ctx, recdb->db_path, hash_size,
+				  tdb_flags, O_RDWR|O_CREAT|O_EXCL, 0600);
+	if (recdb->db == NULL) {
+		talloc_free(recdb);
+		D_ERR("failed to create recovery db %s\n", recdb->db_path);
+		return NULL;
+	}
+
+	recdb->persistent = persistent;
+
+	return recdb;
+}
+
+static uint32_t recdb_id(struct recdb_context *recdb)
+{
+	return recdb->db_id;
+}
+
+static const char *recdb_name(struct recdb_context *recdb)
+{
+	return recdb->db_name;
+}
+
+static const char *recdb_path(struct recdb_context *recdb)
+{
+	return recdb->db_path;
+}
+
+static struct tdb_context *recdb_tdb(struct recdb_context *recdb)
+{
+	return recdb->db->tdb;
+}
+
+static bool recdb_persistent(struct recdb_context *recdb)
+{
+	return recdb->persistent;
+}
+
+struct recdb_add_traverse_state {
+	struct recdb_context *recdb;
+	uint32_t mypnn;
+};
+
+static int recdb_add_traverse(uint32_t reqid, struct ctdb_ltdb_header *header,
+			      TDB_DATA key, TDB_DATA data,
+			      void *private_data)
+{
+	struct recdb_add_traverse_state *state =
+		(struct recdb_add_traverse_state *)private_data;
+	struct ctdb_ltdb_header *hdr;
+	TDB_DATA prev_data;
+	int ret;
+
+	/* header is not marshalled separately in the pulldb control */
+	if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
+		return -1;
+	}
+
+	hdr = (struct ctdb_ltdb_header *)data.dptr;
+
+	/* fetch the existing record, if any */
+	prev_data = tdb_fetch(recdb_tdb(state->recdb), key);
+
+	if (prev_data.dptr != NULL) {
+		struct ctdb_ltdb_header prev_hdr;
+
+		prev_hdr = *(struct ctdb_ltdb_header *)prev_data.dptr;
+		free(prev_data.dptr);
+		if (hdr->rsn < prev_hdr.rsn ||
+		    (hdr->rsn == prev_hdr.rsn &&
+		     prev_hdr.dmaster != state->mypnn)) {
+			return 0;
+		}
+	}
+
+	ret = tdb_store(recdb_tdb(state->recdb), key, data, TDB_REPLACE);
+	if (ret != 0) {
+		return -1;
+	}
+	return 0;
+}
+
+static bool recdb_add(struct recdb_context *recdb, int mypnn,
+		      struct ctdb_rec_buffer *recbuf)
+{
+	struct recdb_add_traverse_state state;
+	int ret;
+
+	state.recdb = recdb;
+	state.mypnn = mypnn;
+
+	ret = ctdb_rec_buffer_traverse(recbuf, recdb_add_traverse, &state);
+	if (ret != 0) {
+		return false;
+	}
+
+	return true;
+}
+
+/* This function decides which records from recdb are retained */
+static int recbuf_filter_add(struct ctdb_rec_buffer *recbuf, bool persistent,
+			     uint32_t reqid, uint32_t dmaster,
+			     TDB_DATA key, TDB_DATA data)
+{
+	struct ctdb_ltdb_header *header;
+	int ret;
+
+	/* Skip empty records */
+	if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
+		return 0;
+	}
+
+	/* update the dmaster field to point to us */
+	header = (struct ctdb_ltdb_header *)data.dptr;
+	if (!persistent) {
+		header->dmaster = dmaster;
+		header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
+	}
+
+	ret = ctdb_rec_buffer_add(recbuf, recbuf, reqid, NULL, key, data);
+	if (ret != 0) {
+		return ret;
+	}
+
+	return 0;
+}
+
+struct recdb_file_traverse_state {
+	struct ctdb_rec_buffer *recbuf;
+	struct recdb_context *recdb;
+	TALLOC_CTX *mem_ctx;
+	uint32_t dmaster;
+	uint32_t reqid;
+	bool persistent;
+	bool failed;
+	int fd;
+	size_t max_size;
+	unsigned int num_buffers;
+};
+
+static int recdb_file_traverse(struct tdb_context *tdb,
+			       TDB_DATA key, TDB_DATA data,
+			       void *private_data)
+{
+	struct recdb_file_traverse_state *state =
+		(struct recdb_file_traverse_state *)private_data;
+	int ret;
+
+	ret = recbuf_filter_add(state->recbuf, state->persistent,
+				state->reqid, state->dmaster, key, data);
+	if (ret != 0) {
+		state->failed = true;
+		return ret;
+	}
+
+	if (ctdb_rec_buffer_len(state->recbuf) > state->max_size) {
+		ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
+		if (ret != 0) {
+			D_ERR("Failed to collect recovery records for %s\n",
+			      recdb_name(state->recdb));
+			state->failed = true;
+			return ret;
+		}
+
+		state->num_buffers += 1;
+
+		TALLOC_FREE(state->recbuf);
+		state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
+						     recdb_id(state->recdb));
+		if (state->recbuf == NULL) {
+			state->failed = true;
+			return ENOMEM;
+		}
+	}
+
+	return 0;
+}
+
+static int recdb_file(struct recdb_context *recdb, TALLOC_CTX *mem_ctx,
+		      uint32_t dmaster, int fd, int max_size)
+{
+	struct recdb_file_traverse_state state;
+	int ret;
+
+	state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
+	if (state.recbuf == NULL) {
+		return -1;
+	}
+	state.recdb = recdb;
+	state.mem_ctx = mem_ctx;
+	state.dmaster = dmaster;
+	state.reqid = 0;
+	state.persistent = recdb_persistent(recdb);
+	state.failed = false;
+	state.fd = fd;
+	state.max_size = max_size;
+	state.num_buffers = 0;
+
+	ret = tdb_traverse_read(recdb_tdb(recdb), recdb_file_traverse, &state);
+	if (ret == -1 || state.failed) {
+		TALLOC_FREE(state.recbuf);
+		return -1;
+	}
+
+	ret = ctdb_rec_buffer_write(state.recbuf, fd);
+	if (ret != 0) {
+		D_ERR("Failed to collect recovery records for %s\n",
+		      recdb_name(recdb));
+		TALLOC_FREE(state.recbuf);
+		return -1;
+	}
+	state.num_buffers += 1;
+
+	D_DEBUG("Wrote %d buffers of recovery records for %s\n",
+		state.num_buffers, recdb_name(recdb));
+
+	return state.num_buffers;
+}
+
+/*
+ * Pull database from a single node
+ */
+
+struct pull_database_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct recdb_context *recdb;
+	uint32_t pnn;
+	uint64_t srvid;
+	unsigned int num_records;
+	int result;
+};
+
+static void pull_database_handler(uint64_t srvid, TDB_DATA data,
+				  void *private_data);
+static void pull_database_register_done(struct tevent_req *subreq);
+static void pull_database_unregister_done(struct tevent_req *subreq);
+static void pull_database_done(struct tevent_req *subreq);
+
+static struct tevent_req *pull_database_send(
+			TALLOC_CTX *mem_ctx,
+			struct tevent_context *ev,
+			struct ctdb_client_context *client,
+			uint32_t pnn,
+			struct recdb_context *recdb)
+{
+	struct tevent_req *req, *subreq;
+	struct pull_database_state *state;
+
+	req = tevent_req_create(mem_ctx, &state, struct pull_database_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->recdb = recdb;
+	state->pnn = pnn;
+	state->srvid = srvid_next();
+
+	subreq = ctdb_client_set_message_handler_send(
+					state, state->ev, state->client,
+					state->srvid, pull_database_handler,
+					req);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+
+	tevent_req_set_callback(subreq, pull_database_register_done, req);
+
+	return req;
+}
+
+static void pull_database_handler(uint64_t srvid, TDB_DATA data,
+				  void *private_data)
+{
+	struct tevent_req *req = talloc_get_type_abort(
+		private_data, struct tevent_req);
+	struct pull_database_state *state = tevent_req_data(
+		req, struct pull_database_state);
+	struct ctdb_rec_buffer *recbuf;
+	size_t np;
+	int ret;
+	bool status;
+
+	if (srvid != state->srvid) {
+		return;
+	}
+
+	ret = ctdb_rec_buffer_pull(data.dptr, data.dsize, state, &recbuf, &np);
+	if (ret != 0) {
+		D_ERR("Invalid data received for DB_PULL messages\n");
+		return;
+	}
+
+	if (recbuf->db_id != recdb_id(state->recdb)) {
+		talloc_free(recbuf);
+		D_ERR("Invalid dbid:%08x for DB_PULL messages for %s\n",
+		      recbuf->db_id, recdb_name(state->recdb));
+		return;
+	}
+
+	status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
+			   recbuf);
+	if (! status) {
+		talloc_free(recbuf);
+		D_ERR("Failed to add records to recdb for %s\n",
+		      recdb_name(state->recdb));
+		return;
+	}
+
+	state->num_records += recbuf->count;
+	talloc_free(recbuf);
+}
+
+static void pull_database_register_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct pull_database_state *state = tevent_req_data(
+		req, struct pull_database_state);
+	struct ctdb_req_control request;
+	struct ctdb_pulldb_ext pulldb_ext;
+	int ret;
+	bool status;
+
+	status = ctdb_client_set_message_handler_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("Failed to set message handler for DB_PULL for %s\n",
+		      recdb_name(state->recdb));
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	pulldb_ext.db_id = recdb_id(state->recdb);
+	pulldb_ext.lmaster = CTDB_LMASTER_ANY;
+	pulldb_ext.srvid = state->srvid;
+
+	ctdb_req_control_db_pull(&request, &pulldb_ext);
+	subreq = ctdb_client_control_send(state, state->ev, state->client,
+					  state->pnn, TIMEOUT(), &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, pull_database_done, req);
+}
+
+static void pull_database_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct pull_database_state *state = tevent_req_data(
+		req, struct pull_database_state);
+	struct ctdb_reply_control *reply;
+	uint32_t num_records;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("control DB_PULL failed for %s on node %u, ret=%d\n",
+		      recdb_name(state->recdb), state->pnn, ret);
+		state->result = ret;
+		goto unregister;
+	}
+
+	ret = ctdb_reply_control_db_pull(reply, &num_records);
+	talloc_free(reply);
+	if (num_records != state->num_records) {
+		D_ERR("mismatch (%u != %u) in DB_PULL records for db %s\n",
+		      num_records, state->num_records,
+		      recdb_name(state->recdb));
+		state->result = EIO;
+		goto unregister;
+	}
+
+	D_INFO("Pulled %d records for db %s from node %d\n",
+	       state->num_records, recdb_name(state->recdb), state->pnn);
+
+unregister:
+
+	subreq = ctdb_client_remove_message_handler_send(
+					state, state->ev, state->client,
+					state->srvid, req);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, pull_database_unregister_done, req);
+}
+
+static void pull_database_unregister_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct pull_database_state *state = tevent_req_data(
+		req, struct pull_database_state);
+	int ret;
+	bool status;
+
+	status = ctdb_client_remove_message_handler_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("failed to remove message handler for DB_PULL for db %s\n",
+		      recdb_name(state->recdb));
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	if (state->result != 0) {
+		tevent_req_error(req, state->result);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static bool pull_database_recv(struct tevent_req *req, int *perr)
+{
+	return generic_recv(req, perr);
+}
+
+/*
+ * Push database to specified nodes (new style)
+ */
+
+struct push_database_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct recdb_context *recdb;
+	uint32_t *pnn_list;
+	unsigned int count;
+	uint64_t srvid;
+	uint32_t dmaster;
+	int fd;
+	int num_buffers;
+	int num_buffers_sent;
+	unsigned int num_records;
+};
+
+static void push_database_started(struct tevent_req *subreq);
+static void push_database_send_msg(struct tevent_req *req);
+static void push_database_send_done(struct tevent_req *subreq);
+static void push_database_confirmed(struct tevent_req *subreq);
+
+static struct tevent_req *push_database_send(
+			TALLOC_CTX *mem_ctx,
+			struct tevent_context *ev,
+			struct ctdb_client_context *client,
+			uint32_t *pnn_list,
+			unsigned int count,
+			struct recdb_context *recdb,
+			int max_size)
+{
+	struct tevent_req *req, *subreq;
+	struct push_database_state *state;
+	struct ctdb_req_control request;
+	struct ctdb_pulldb_ext pulldb_ext;
+	char *filename;
+	off_t offset;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct push_database_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->recdb = recdb;
+	state->pnn_list = pnn_list;
+	state->count = count;
+
+	state->srvid = srvid_next();
+	state->dmaster = ctdb_client_pnn(client);
+	state->num_buffers_sent = 0;
+	state->num_records = 0;
+
+	filename = talloc_asprintf(state, "%s.dat", recdb_path(recdb));
+	if (tevent_req_nomem(filename, req)) {
+		return tevent_req_post(req, ev);
+	}
+
+	state->fd = open(filename, O_RDWR|O_CREAT, 0644);
+	if (state->fd == -1) {
+		tevent_req_error(req, errno);
+		return tevent_req_post(req, ev);
+	}
+	unlink(filename);
+	talloc_free(filename);
+
+	state->num_buffers = recdb_file(recdb, state, state->dmaster,
+					state->fd, max_size);
+	if (state->num_buffers == -1) {
+		tevent_req_error(req, ENOMEM);
+		return tevent_req_post(req, ev);
+	}
+
+	offset = lseek(state->fd, 0, SEEK_SET);
+	if (offset != 0) {
+		tevent_req_error(req, EIO);
+		return tevent_req_post(req, ev);
+	}
+
+	pulldb_ext.db_id = recdb_id(recdb);
+	pulldb_ext.srvid = state->srvid;
+
+	ctdb_req_control_db_push_start(&request, &pulldb_ext);
+	subreq = ctdb_client_control_multi_send(state, ev, client,
+						pnn_list, count,
+						TIMEOUT(), &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, push_database_started, req);
+
+	return req;
+}
+
+static void push_database_started(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct push_database_state *state = tevent_req_data(
+		req, struct push_database_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state,
+						&err_list, NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->pnn_list,
+						       state->count,
+						       err_list, &pnn);
+		if (ret2 != 0) {
+			D_ERR("control DB_PUSH_START failed for db %s"
+			      " on node %u, ret=%d\n",
+			      recdb_name(state->recdb), pnn, ret2);
+		} else {
+			D_ERR("control DB_PUSH_START failed for db %s,"
+			      " ret=%d\n",
+			      recdb_name(state->recdb), ret);
+		}
+		talloc_free(err_list);
+
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	push_database_send_msg(req);
+}
+
+static void push_database_send_msg(struct tevent_req *req)
+{
+	struct push_database_state *state = tevent_req_data(
+		req, struct push_database_state);
+	struct tevent_req *subreq;
+	struct ctdb_rec_buffer *recbuf;
+	struct ctdb_req_message message;
+	TDB_DATA data;
+	size_t np;
+	int ret;
+
+	if (state->num_buffers_sent == state->num_buffers) {
+		struct ctdb_req_control request;
+
+		ctdb_req_control_db_push_confirm(&request,
+						 recdb_id(state->recdb));
+		subreq = ctdb_client_control_multi_send(state, state->ev,
+							state->client,
+							state->pnn_list,
+							state->count,
+							TIMEOUT(), &request);
+		if (tevent_req_nomem(subreq, req)) {
+			return;
+		}
+		tevent_req_set_callback(subreq, push_database_confirmed, req);
+		return;
+	}
+
+	ret = ctdb_rec_buffer_read(state->fd, state, &recbuf);
+	if (ret != 0) {
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	data.dsize = ctdb_rec_buffer_len(recbuf);
+	data.dptr = talloc_size(state, data.dsize);
+	if (tevent_req_nomem(data.dptr, req)) {
+		return;
+	}
+
+	ctdb_rec_buffer_push(recbuf, data.dptr, &np);
+
+	message.srvid = state->srvid;
+	message.data.data = data;
+
+	D_DEBUG("Pushing buffer %d with %d records for db %s\n",
+		state->num_buffers_sent, recbuf->count,
+		recdb_name(state->recdb));
+
+	subreq = ctdb_client_message_multi_send(state, state->ev,
+						state->client,
+						state->pnn_list, state->count,
+						&message);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, push_database_send_done, req);
+
+	state->num_records += recbuf->count;
+
+	talloc_free(data.dptr);
+	talloc_free(recbuf);
+}
+
+static void push_database_send_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct push_database_state *state = tevent_req_data(
+		req, struct push_database_state);
+	bool status;
+	int ret;
+
+	status = ctdb_client_message_multi_recv(subreq, &ret, NULL, NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("Sending recovery records failed for %s\n",
+		      recdb_name(state->recdb));
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	state->num_buffers_sent += 1;
+
+	push_database_send_msg(req);
+}
+
+static void push_database_confirmed(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct push_database_state *state = tevent_req_data(
+		req, struct push_database_state);
+	struct ctdb_reply_control **reply;
+	int *err_list;
+	bool status;
+	unsigned int i;
+	int ret;
+	uint32_t num_records;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state,
+						&err_list, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->pnn_list,
+						       state->count, err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control DB_PUSH_CONFIRM failed for db %s"
+			      " on node %u, ret=%d\n",
+			      recdb_name(state->recdb), pnn, ret2);
+		} else {
+			D_ERR("control DB_PUSH_CONFIRM failed for db %s,"
+			      " ret=%d\n",
+			      recdb_name(state->recdb), ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	for (i=0; i<state->count; i++) {
+		ret = ctdb_reply_control_db_push_confirm(reply[i],
+							 &num_records);
+		if (ret != 0) {
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		if (num_records != state->num_records) {
+			D_ERR("Node %u received %d of %d records for %s\n",
+			      state->pnn_list[i], num_records,
+			      state->num_records, recdb_name(state->recdb));
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+	}
+
+	talloc_free(reply);
+
+	D_INFO("Pushed %d records for db %s\n",
+	       state->num_records, recdb_name(state->recdb));
+
+	tevent_req_done(req);
+}
+
+static bool push_database_recv(struct tevent_req *req, int *perr)
+{
+	return generic_recv(req, perr);
+}
+
+/*
+ * Collect databases using highest sequence number
+ */
+
+struct collect_highseqnum_db_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct node_list *nlist;
+	uint32_t db_id;
+	struct recdb_context *recdb;
+
+	uint32_t max_pnn;
+};
+
+static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq);
+static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq);
+
+static struct tevent_req *collect_highseqnum_db_send(
+			TALLOC_CTX *mem_ctx,
+			struct tevent_context *ev,
+			struct ctdb_client_context *client,
+			struct node_list *nlist,
+			uint32_t db_id,
+			struct recdb_context *recdb)
+{
+	struct tevent_req *req, *subreq;
+	struct collect_highseqnum_db_state *state;
+	struct ctdb_req_control request;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct collect_highseqnum_db_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->nlist = nlist;
+	state->db_id = db_id;
+	state->recdb = recdb;
+
+	ctdb_req_control_get_db_seqnum(&request, db_id);
+	subreq = ctdb_client_control_multi_send(mem_ctx,
+						ev,
+						client,
+						nlist->pnn_list,
+						nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, collect_highseqnum_db_seqnum_done,
+				req);
+
+	return req;
+}
+
+static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct collect_highseqnum_db_state *state = tevent_req_data(
+		req, struct collect_highseqnum_db_state);
+	struct ctdb_reply_control **reply;
+	int *err_list;
+	bool status;
+	unsigned int i;
+	int ret;
+	uint64_t seqnum, max_seqnum;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state,
+						&err_list, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control GET_DB_SEQNUM failed for db %s"
+			      " on node %u, ret=%d\n",
+			      recdb_name(state->recdb), pnn, ret2);
+		} else {
+			D_ERR("control GET_DB_SEQNUM failed for db %s,"
+			      " ret=%d\n",
+			      recdb_name(state->recdb), ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	max_seqnum = 0;
+	state->max_pnn = state->nlist->pnn_list[0];
+	for (i=0; i<state->nlist->count; i++) {
+		ret = ctdb_reply_control_get_db_seqnum(reply[i], &seqnum);
+		if (ret != 0) {
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		if (max_seqnum < seqnum) {
+			max_seqnum = seqnum;
+			state->max_pnn = state->nlist->pnn_list[i];
+		}
+	}
+
+	talloc_free(reply);
+
+	D_INFO("Pull persistent db %s from node %d with seqnum 0x%"PRIx64"\n",
+	       recdb_name(state->recdb), state->max_pnn, max_seqnum);
+
+	subreq = pull_database_send(state,
+				    state->ev,
+				    state->client,
+				    state->max_pnn,
+				    state->recdb);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, collect_highseqnum_db_pulldb_done,
+				req);
+}
+
+static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct collect_highseqnum_db_state *state = tevent_req_data(
+		req, struct collect_highseqnum_db_state);
+	int ret;
+	bool status;
+
+	status = pull_database_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		node_list_ban_credits(state->nlist, state->max_pnn);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static bool collect_highseqnum_db_recv(struct tevent_req *req, int *perr)
+{
+	return generic_recv(req, perr);
+}
+
+/*
+ * Collect all databases
+ */
+
+struct collect_all_db_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct node_list *nlist;
+	uint32_t db_id;
+	struct recdb_context *recdb;
+
+	struct ctdb_pulldb pulldb;
+	unsigned int index;
+};
+
+static void collect_all_db_pulldb_done(struct tevent_req *subreq);
+
+static struct tevent_req *collect_all_db_send(
+			TALLOC_CTX *mem_ctx,
+			struct tevent_context *ev,
+			struct ctdb_client_context *client,
+			struct node_list *nlist,
+			uint32_t db_id,
+			struct recdb_context *recdb)
+{
+	struct tevent_req *req, *subreq;
+	struct collect_all_db_state *state;
+
+	req = tevent_req_create(mem_ctx, &state,
+				struct collect_all_db_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->nlist = nlist;
+	state->db_id = db_id;
+	state->recdb = recdb;
+	state->index = 0;
+
+	subreq = pull_database_send(state,
+				    ev,
+				    client,
+				    nlist->pnn_list[state->index],
+				    recdb);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
+
+	return req;
+}
+
+static void collect_all_db_pulldb_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct collect_all_db_state *state = tevent_req_data(
+		req, struct collect_all_db_state);
+	int ret;
+	bool status;
+
+	status = pull_database_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		node_list_ban_credits(state->nlist,
+				      state->nlist->pnn_list[state->index]);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	state->index += 1;
+	if (state->index == state->nlist->count) {
+		tevent_req_done(req);
+		return;
+	}
+
+	subreq = pull_database_send(state,
+				    state->ev,
+				    state->client,
+				    state->nlist->pnn_list[state->index],
+				    state->recdb);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
+}
+
+static bool collect_all_db_recv(struct tevent_req *req, int *perr)
+{
+	return generic_recv(req, perr);
+}
+
+
+/**
+ * For each database do the following:
+ *  - Get DB name from all nodes
+ *  - Attach database on missing nodes
+ *  - Get DB path
+ *  - Freeze database on all nodes
+ *  - Start transaction on all nodes
+ *  - Collect database from all nodes
+ *  - Wipe database on all nodes
+ *  - Push database to all nodes
+ *  - Commit transaction on all nodes
+ *  - Thaw database on all nodes
+ */
+
+struct recover_db_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct ctdb_tunable_list *tun_list;
+	struct node_list *nlist;
+	struct db *db;
+
+	uint32_t destnode;
+	struct ctdb_transdb transdb;
+
+	const char *db_name, *db_path;
+	struct recdb_context *recdb;
+};
+
+static void recover_db_name_done(struct tevent_req *subreq);
+static void recover_db_create_missing_done(struct tevent_req *subreq);
+static void recover_db_path_done(struct tevent_req *subreq);
+static void recover_db_freeze_done(struct tevent_req *subreq);
+static void recover_db_transaction_started(struct tevent_req *subreq);
+static void recover_db_collect_done(struct tevent_req *subreq);
+static void recover_db_wipedb_done(struct tevent_req *subreq);
+static void recover_db_pushdb_done(struct tevent_req *subreq);
+static void recover_db_transaction_committed(struct tevent_req *subreq);
+static void recover_db_thaw_done(struct tevent_req *subreq);
+
+static struct tevent_req *recover_db_send(TALLOC_CTX *mem_ctx,
+					  struct tevent_context *ev,
+					  struct ctdb_client_context *client,
+					  struct ctdb_tunable_list *tun_list,
+					  struct node_list *nlist,
+					  uint32_t generation,
+					  struct db *db)
+{
+	struct tevent_req *req, *subreq;
+	struct recover_db_state *state;
+	struct ctdb_req_control request;
+
+	req = tevent_req_create(mem_ctx, &state, struct recover_db_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->tun_list = tun_list;
+	state->nlist = nlist;
+	state->db = db;
+
+	state->destnode = ctdb_client_pnn(client);
+	state->transdb.db_id = db->db_id;
+	state->transdb.tid = generation;
+
+	ctdb_req_control_get_dbname(&request, db->db_id);
+	subreq = ctdb_client_control_multi_send(state,
+						ev,
+						client,
+						state->db->pnn_list,
+						state->db->num_nodes,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, recover_db_name_done, req);
+
+	return req;
+}
+
+static void recover_db_name_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_reply_control **reply;
+	int *err_list;
+	unsigned int i;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq,
+						&ret,
+						state,
+						&err_list,
+						&reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->db->pnn_list,
+						       state->db->num_nodes,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control GET_DBNAME failed on node %u,"
+			      " ret=%d\n",
+			      pnn,
+			      ret2);
+		} else {
+			D_ERR("control GET_DBNAME failed, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	for (i = 0; i < state->db->num_nodes; i++) {
+		const char *db_name;
+		uint32_t pnn;
+
+		pnn = state->nlist->pnn_list[i];
+
+		ret = ctdb_reply_control_get_dbname(reply[i],
+						    state,
+						    &db_name);
+		if (ret != 0) {
+			D_ERR("control GET_DBNAME failed on node %u "
+			      "for db=0x%x, ret=%d\n",
+			      pnn,
+			      state->db->db_id,
+			      ret);
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		if (state->db_name == NULL) {
+			state->db_name = db_name;
+			continue;
+		}
+
+		if (strcmp(state->db_name, db_name) != 0) {
+			D_ERR("Incompatible database name for 0x%"PRIx32" "
+			      "(%s != %s) on node %"PRIu32"\n",
+			      state->db->db_id,
+			      db_name,
+			      state->db_name,
+			      pnn);
+			node_list_ban_credits(state->nlist, pnn);
+			tevent_req_error(req, ret);
+			return;
+		}
+	}
+
+	talloc_free(reply);
+
+	subreq = db_create_missing_send(state,
+					state->ev,
+					state->client,
+					state->nlist,
+					state->db_name,
+					state->db);
+
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_create_missing_done, req);
+}
+
+static void recover_db_create_missing_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_req_control request;
+	int ret;
+	bool status;
+
+	/* Could sanity check the db_id here */
+	status = db_create_missing_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ctdb_req_control_getdbpath(&request, state->db->db_id);
+	subreq = ctdb_client_control_send(state, state->ev, state->client,
+					  state->destnode, TIMEOUT(),
+					  &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_path_done, req);
+}
+
+static void recover_db_path_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_reply_control *reply;
+	struct ctdb_req_control request;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
+		      state->db_name, ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ret = ctdb_reply_control_getdbpath(reply, state, &state->db_path);
+	if (ret != 0) {
+		D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
+		      state->db_name, ret);
+		tevent_req_error(req, EPROTO);
+		return;
+	}
+
+	talloc_free(reply);
+
+	ctdb_req_control_db_freeze(&request, state->db->db_id);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_freeze_done, req);
+}
+
+static void recover_db_freeze_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_req_control request;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control FREEZE_DB failed for db %s"
+			      " on node %u, ret=%d\n",
+			      state->db_name, pnn, ret2);
+
+			node_list_ban_credits(state->nlist, pnn);
+		} else {
+			D_ERR("control FREEZE_DB failed for db %s, ret=%d\n",
+			      state->db_name, ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ctdb_req_control_db_transaction_start(&request, &state->transdb);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_transaction_started, req);
+}
+
+static void recover_db_transaction_started(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	int *err_list;
+	uint32_t flags;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control TRANSACTION_DB failed for db=%s"
+			      " on node %u, ret=%d\n",
+			      state->db_name, pnn, ret2);
+		} else {
+			D_ERR("control TRANSACTION_DB failed for db=%s,"
+			      " ret=%d\n", state->db_name, ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	flags = state->db->db_flags;
+	state->recdb = recdb_create(state,
+				    state->db->db_id,
+				    state->db_name,
+				    state->db_path,
+				    state->tun_list->database_hash_size,
+				    flags & CTDB_DB_FLAGS_PERSISTENT);
+	if (tevent_req_nomem(state->recdb, req)) {
+		return;
+	}
+
+	if ((flags & CTDB_DB_FLAGS_PERSISTENT) ||
+	    (flags & CTDB_DB_FLAGS_REPLICATED)) {
+		subreq = collect_highseqnum_db_send(state,
+						    state->ev,
+						    state->client,
+						    state->nlist,
+						    state->db->db_id,
+						    state->recdb);
+	} else {
+		subreq = collect_all_db_send(state,
+					     state->ev,
+					     state->client,
+					     state->nlist,
+					     state->db->db_id,
+					     state->recdb);
+	}
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_collect_done, req);
+}
+
+static void recover_db_collect_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_req_control request;
+	int ret;
+	bool status;
+
+	if ((state->db->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
+	    (state->db->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
+		status = collect_highseqnum_db_recv(subreq, &ret);
+	} else {
+		status = collect_all_db_recv(subreq, &ret);
+	}
+	TALLOC_FREE(subreq);
+	if (! status) {
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ctdb_req_control_wipe_database(&request, &state->transdb);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_wipedb_done, req);
+}
+
+static void recover_db_wipedb_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control WIPEDB failed for db %s on node %u,"
+			      " ret=%d\n", state->db_name, pnn, ret2);
+		} else {
+			D_ERR("control WIPEDB failed for db %s, ret=%d\n",
+			      state->db_name, ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	subreq = push_database_send(state,
+				    state->ev,
+				    state->client,
+				    state->nlist->pnn_list,
+				    state->nlist->count,
+				    state->recdb,
+				    state->tun_list->rec_buffer_size_limit);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_pushdb_done, req);
+}
+
+static void recover_db_pushdb_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_req_control request;
+	int ret;
+	bool status;
+
+	status = push_database_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	TALLOC_FREE(state->recdb);
+
+	ctdb_req_control_db_transaction_commit(&request, &state->transdb);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_transaction_committed, req);
+}
+
+static void recover_db_transaction_committed(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	struct ctdb_req_control request;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control DB_TRANSACTION_COMMIT failed for db %s"
+			      " on node %u, ret=%d\n",
+			      state->db_name, pnn, ret2);
+		} else {
+			D_ERR("control DB_TRANSACTION_COMMIT failed for db %s,"
+			      " ret=%d\n", state->db_name, ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ctdb_req_control_db_thaw(&request, state->db->db_id);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recover_db_thaw_done, req);
+}
+
+static void recover_db_thaw_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recover_db_state *state = tevent_req_data(
+		req, struct recover_db_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control DB_THAW failed for db %s on node %u,"
+			      " ret=%d\n", state->db_name, pnn, ret2);
+		} else {
+			D_ERR("control DB_THAW failed for db %s, ret=%d\n",
+			      state->db_name, ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	tevent_req_done(req);
+}
+
+static bool recover_db_recv(struct tevent_req *req)
+{
+	return generic_recv(req, NULL);
+}
+
+
+/*
+ * Start database recovery for each database
+ *
+ * Try to recover each database 5 times before failing recovery.
+ */
+
+struct db_recovery_state {
+	struct tevent_context *ev;
+	struct db_list *dblist;
+	unsigned int num_replies;
+	unsigned int num_failed;
+};
+
+struct db_recovery_one_state {
+	struct tevent_req *req;
+	struct ctdb_client_context *client;
+	struct db_list *dblist;
+	struct ctdb_tunable_list *tun_list;
+	struct node_list *nlist;
+	uint32_t generation;
+	struct db *db;
+	int num_fails;
+};
+
+static void db_recovery_one_done(struct tevent_req *subreq);
+
+static struct tevent_req *db_recovery_send(TALLOC_CTX *mem_ctx,
+					   struct tevent_context *ev,
+					   struct ctdb_client_context *client,
+					   struct db_list *dblist,
+					   struct ctdb_tunable_list *tun_list,
+					   struct node_list *nlist,
+					   uint32_t generation)
+{
+	struct tevent_req *req, *subreq;
+	struct db_recovery_state *state;
+	struct db *db;
+
+	req = tevent_req_create(mem_ctx, &state, struct db_recovery_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->dblist = dblist;
+	state->num_replies = 0;
+	state->num_failed = 0;
+
+	if (dblist->num_dbs == 0) {
+		tevent_req_done(req);
+		return tevent_req_post(req, ev);
+	}
+
+	for (db = dblist->db; db != NULL; db = db->next) {
+		struct db_recovery_one_state *substate;
+
+		substate = talloc_zero(state, struct db_recovery_one_state);
+		if (tevent_req_nomem(substate, req)) {
+			return tevent_req_post(req, ev);
+		}
+
+		substate->req = req;
+		substate->client = client;
+		substate->dblist = dblist;
+		substate->tun_list = tun_list;
+		substate->nlist = nlist;
+		substate->generation = generation;
+		substate->db = db;
+
+		subreq = recover_db_send(state,
+					 ev,
+					 client,
+					 tun_list,
+					 nlist,
+					 generation,
+					 substate->db);
+		if (tevent_req_nomem(subreq, req)) {
+			return tevent_req_post(req, ev);
+		}
+		tevent_req_set_callback(subreq, db_recovery_one_done,
+					substate);
+		D_NOTICE("recover database 0x%08x\n", substate->db->db_id);
+	}
+
+	return req;
+}
+
+static void db_recovery_one_done(struct tevent_req *subreq)
+{
+	struct db_recovery_one_state *substate = tevent_req_callback_data(
+		subreq, struct db_recovery_one_state);
+	struct tevent_req *req = substate->req;
+	struct db_recovery_state *state = tevent_req_data(
+		req, struct db_recovery_state);
+	bool status;
+
+	status = recover_db_recv(subreq);
+	TALLOC_FREE(subreq);
+
+	if (status) {
+		talloc_free(substate);
+		goto done;
+	}
+
+	substate->num_fails += 1;
+	if (substate->num_fails < NUM_RETRIES) {
+		subreq = recover_db_send(state,
+					 state->ev,
+					 substate->client,
+					 substate->tun_list,
+					 substate->nlist,
+					 substate->generation,
+					 substate->db);
+		if (tevent_req_nomem(subreq, req)) {
+			goto failed;
+		}
+		tevent_req_set_callback(subreq, db_recovery_one_done, substate);
+		D_NOTICE("recover database 0x%08x, attempt %d\n",
+			 substate->db->db_id, substate->num_fails+1);
+		return;
+	}
+
+failed:
+	state->num_failed += 1;
+
+done:
+	state->num_replies += 1;
+
+	if (state->num_replies == state->dblist->num_dbs) {
+		tevent_req_done(req);
+	}
+}
+
+static bool db_recovery_recv(struct tevent_req *req, unsigned int *count)
+{
+	struct db_recovery_state *state = tevent_req_data(
+		req, struct db_recovery_state);
+	int err;
+
+	if (tevent_req_is_unix_error(req, &err)) {
+		*count = 0;
+		return false;
+	}
+
+	*count = state->num_replies - state->num_failed;
+
+	if (state->num_failed > 0) {
+		return false;
+	}
+
+	return true;
+}
+
+struct ban_node_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	struct ctdb_tunable_list *tun_list;
+	struct node_list *nlist;
+	uint32_t destnode;
+
+	uint32_t max_pnn;
+};
+
+static bool ban_node_check(struct tevent_req *req);
+static void ban_node_check_done(struct tevent_req *subreq);
+static void ban_node_done(struct tevent_req *subreq);
+
+static struct tevent_req *ban_node_send(TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct ctdb_client_context *client,
+					struct ctdb_tunable_list *tun_list,
+					struct node_list *nlist)
+{
+	struct tevent_req *req;
+	struct ban_node_state *state;
+	bool ok;
+
+	req = tevent_req_create(mem_ctx, &state, struct ban_node_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->tun_list = tun_list;
+	state->nlist = nlist;
+	state->destnode = ctdb_client_pnn(client);
+
+	/* Bans are not enabled */
+	if (state->tun_list->enable_bans == 0) {
+		D_ERR("Bans are not enabled\n");
+		tevent_req_done(req);
+		return tevent_req_post(req, ev);
+	}
+
+	ok = ban_node_check(req);
+	if (!ok) {
+		return tevent_req_post(req, ev);
+	}
+
+	return req;
+}
+
+static bool ban_node_check(struct tevent_req *req)
+{
+	struct tevent_req *subreq;
+	struct ban_node_state *state = tevent_req_data(
+		req, struct ban_node_state);
+	struct ctdb_req_control request;
+	unsigned max_credits = 0, i;
+
+	for (i=0; i<state->nlist->count; i++) {
+		if (state->nlist->ban_credits[i] > max_credits) {
+			state->max_pnn = state->nlist->pnn_list[i];
+			max_credits = state->nlist->ban_credits[i];
+		}
+	}
+
+	if (max_credits < NUM_RETRIES) {
+		tevent_req_done(req);
+		return false;
+	}
+
+	ctdb_req_control_get_nodemap(&request);
+	subreq = ctdb_client_control_send(state,
+					  state->ev,
+					  state->client,
+					  state->max_pnn,
+					  TIMEOUT(),
+					  &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return false;
+	}
+	tevent_req_set_callback(subreq, ban_node_check_done, req);
+
+	return true;
+}
+
+static void ban_node_check_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct ban_node_state *state = tevent_req_data(
+		req, struct ban_node_state);
+	struct ctdb_reply_control *reply;
+	struct ctdb_node_map *nodemap;
+	struct ctdb_req_control request;
+	struct ctdb_ban_state ban;
+	unsigned int i;
+	int ret;
+	bool ok;
+
+	ok = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (!ok) {
+		D_ERR("control GET_NODEMAP failed to node %u, ret=%d\n",
+		      state->max_pnn, ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ret = ctdb_reply_control_get_nodemap(reply, state, &nodemap);
+	if (ret != 0) {
+		D_ERR("control GET_NODEMAP failed, ret=%d\n", ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	for (i=0; i<nodemap->num; i++) {
+		if (nodemap->node[i].pnn != state->max_pnn) {
+			continue;
+		}
+
+		/* If the node became inactive, reset ban_credits */
+		if (nodemap->node[i].flags & NODE_FLAGS_INACTIVE) {
+			unsigned int j;
+
+			for (j=0; j<state->nlist->count; j++) {
+				if (state->nlist->pnn_list[j] ==
+						state->max_pnn) {
+					state->nlist->ban_credits[j] = 0;
+					break;
+				}
+			}
+			state->max_pnn = CTDB_UNKNOWN_PNN;
+		}
+	}
+
+	talloc_free(nodemap);
+	talloc_free(reply);
+
+	/* If node becames inactive during recovery, pick next */
+	if (state->max_pnn == CTDB_UNKNOWN_PNN) {
+		(void) ban_node_check(req);
+		return;
+	}
+
+	ban = (struct ctdb_ban_state) {
+		.pnn = state->max_pnn,
+		.time = state->tun_list->recovery_ban_period,
+	};
+
+	D_ERR("Banning node %u for %u seconds\n", ban.pnn, ban.time);
+
+	ctdb_req_control_set_ban_state(&request, &ban);
+	subreq = ctdb_client_control_send(state,
+					  state->ev,
+					  state->client,
+					  ban.pnn,
+					  TIMEOUT(),
+					  &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, ban_node_done, req);
+}
+
+static void ban_node_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct node_ban_state *state = tevent_req_data(
+		req, struct node_ban_state);
+	struct ctdb_reply_control *reply;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ret = ctdb_reply_control_set_ban_state(reply);
+	if (ret != 0) {
+		D_ERR("control SET_BAN_STATE failed, ret=%d\n", ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	talloc_free(reply);
+	tevent_req_done(req);
+}
+
+static bool ban_node_recv(struct tevent_req *req, int *perr)
+{
+	if (tevent_req_is_unix_error(req, perr)) {
+		return false;
+	}
+
+	return true;
+}
+
+/*
+ * Run the parallel database recovery
+ *
+ * - Get tunables
+ * - Get nodemap from all nodes
+ * - Get capabilities from all nodes
+ * - Get dbmap
+ * - Set RECOVERY_ACTIVE
+ * - Send START_RECOVERY
+ * - Update vnnmap on all nodes
+ * - Run database recovery
+ * - Set RECOVERY_NORMAL
+ * - Send END_RECOVERY
+ */
+
+struct recovery_state {
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	uint32_t generation;
+	uint32_t destnode;
+	struct node_list *nlist;
+	struct ctdb_tunable_list *tun_list;
+	struct ctdb_vnn_map *vnnmap;
+	struct db_list *dblist;
+};
+
+static void recovery_tunables_done(struct tevent_req *subreq);
+static void recovery_nodemap_done(struct tevent_req *subreq);
+static void recovery_nodemap_verify(struct tevent_req *subreq);
+static void recovery_capabilities_done(struct tevent_req *subreq);
+static void recovery_dbmap_done(struct tevent_req *subreq);
+static void recovery_active_done(struct tevent_req *subreq);
+static void recovery_start_recovery_done(struct tevent_req *subreq);
+static void recovery_vnnmap_update_done(struct tevent_req *subreq);
+static void recovery_db_recovery_done(struct tevent_req *subreq);
+static void recovery_failed_done(struct tevent_req *subreq);
+static void recovery_normal_done(struct tevent_req *subreq);
+static void recovery_end_recovery_done(struct tevent_req *subreq);
+
+static struct tevent_req *recovery_send(TALLOC_CTX *mem_ctx,
+					struct tevent_context *ev,
+					struct ctdb_client_context *client,
+					uint32_t generation)
+{
+	struct tevent_req *req, *subreq;
+	struct recovery_state *state;
+	struct ctdb_req_control request;
+
+	req = tevent_req_create(mem_ctx, &state, struct recovery_state);
+	if (req == NULL) {
+		return NULL;
+	}
+
+	state->ev = ev;
+	state->client = client;
+	state->generation = generation;
+	state->destnode = ctdb_client_pnn(client);
+
+	ctdb_req_control_get_all_tunables(&request);
+	subreq = ctdb_client_control_send(state, state->ev, state->client,
+					  state->destnode, TIMEOUT(),
+					  &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return tevent_req_post(req, ev);
+	}
+	tevent_req_set_callback(subreq, recovery_tunables_done, req);
+
+	return req;
+}
+
+static void recovery_tunables_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_reply_control *reply;
+	struct ctdb_req_control request;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ret = ctdb_reply_control_get_all_tunables(reply, state,
+						  &state->tun_list);
+	if (ret != 0) {
+		D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
+		tevent_req_error(req, EPROTO);
+		return;
+	}
+
+	talloc_free(reply);
+
+	recover_timeout = state->tun_list->recover_timeout;
+
+	ctdb_req_control_get_nodemap(&request);
+	subreq = ctdb_client_control_send(state, state->ev, state->client,
+					  state->destnode, TIMEOUT(),
+					  &request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_nodemap_done, req);
+}
+
+static void recovery_nodemap_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_reply_control *reply;
+	struct ctdb_req_control request;
+	struct ctdb_node_map *nodemap;
+	unsigned int i;
+	bool status;
+	int ret;
+
+	status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("control GET_NODEMAP failed to node %u, ret=%d\n",
+		      state->destnode, ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	ret = ctdb_reply_control_get_nodemap(reply, state, &nodemap);
+	if (ret != 0) {
+		D_ERR("control GET_NODEMAP failed, ret=%d\n", ret);
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	state->nlist = node_list_init(state, nodemap->num);
+	if (tevent_req_nomem(state->nlist, req)) {
+		return;
+	}
+
+	for (i=0; i<nodemap->num; i++) {
+		bool ok;
+
+		if (nodemap->node[i].flags & NODE_FLAGS_DISCONNECTED) {
+			continue;
+		}
+
+		ok = node_list_add(state->nlist, nodemap->node[i].pnn);
+		if (!ok) {
+			tevent_req_error(req, EINVAL);
+			return;
+		}
+	}
+
+	talloc_free(nodemap);
+	talloc_free(reply);
+
+	/* Verify flags by getting local node information from each node */
+	ctdb_req_control_get_nodemap(&request);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_nodemap_verify, req);
+}
+
+static void recovery_nodemap_verify(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_req_control request;
+	struct ctdb_reply_control **reply;
+	struct node_list *nlist;
+	unsigned int i;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq,
+						&ret,
+						state,
+						&err_list,
+						&reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control GET_NODEMAP failed on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("control GET_NODEMAP failed, ret=%d\n", ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	nlist = node_list_init(state, state->nlist->size);
+	if (tevent_req_nomem(nlist, req)) {
+		return;
+	}
+
+	for (i=0; i<state->nlist->count; i++) {
+		struct ctdb_node_map *nodemap = NULL;
+		uint32_t pnn, flags;
+		unsigned int j;
+		bool ok;
+
+		pnn = state->nlist->pnn_list[i];
+		ret = ctdb_reply_control_get_nodemap(reply[i],
+						     state,
+						     &nodemap);
+		if (ret != 0) {
+			D_ERR("control GET_NODEMAP failed on node %u\n", pnn);
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		flags = NODE_FLAGS_DISCONNECTED;
+		for (j=0; j<nodemap->num; j++) {
+			if (nodemap->node[j].pnn == pnn) {
+				flags = nodemap->node[j].flags;
+				break;
+			}
+		}
+
+		TALLOC_FREE(nodemap);
+
+		if (flags & NODE_FLAGS_INACTIVE) {
+			continue;
+		}
+
+		ok = node_list_add(nlist, pnn);
+		if (!ok) {
+			tevent_req_error(req, EINVAL);
+			return;
+		}
+	}
+
+	talloc_free(reply);
+
+	talloc_free(state->nlist);
+	state->nlist = nlist;
+
+	ctdb_req_control_get_capabilities(&request);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_capabilities_done, req);
+}
+
+static void recovery_capabilities_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_reply_control **reply;
+	struct ctdb_req_control request;
+	int *err_list;
+	unsigned int i;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
+						&reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control GET_CAPABILITIES failed on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("control GET_CAPABILITIES failed, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	for (i=0; i<state->nlist->count; i++) {
+		uint32_t caps;
+
+		ret = ctdb_reply_control_get_capabilities(reply[i], &caps);
+		if (ret != 0) {
+			D_ERR("control GET_CAPABILITIES failed on node %u\n",
+			      state->nlist->pnn_list[i]);
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		state->nlist->caps[i] = caps;
+	}
+
+	talloc_free(reply);
+
+	ctdb_req_control_get_dbmap(&request);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_dbmap_done, req);
+}
+
+static void recovery_dbmap_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_reply_control **reply;
+	struct ctdb_req_control request;
+	int *err_list;
+	unsigned int i, j;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq,
+						&ret,
+						state,
+						&err_list,
+						&reply);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("control GET_DBMAP failed on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("control GET_DBMAP failed, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	state->dblist = db_list_init(state, state->nlist->count);
+	if (tevent_req_nomem(state->dblist, req)) {
+		D_ERR("memory allocation error\n");
+		return;
+	}
+
+	for (i = 0; i < state->nlist->count; i++) {
+		struct ctdb_dbid_map *dbmap = NULL;
+		uint32_t pnn;
+
+		pnn = state->nlist->pnn_list[i];
+
+		ret = ctdb_reply_control_get_dbmap(reply[i], state, &dbmap);
+		if (ret != 0) {
+			D_ERR("control GET_DBMAP failed on node %u\n",
+			      pnn);
+			tevent_req_error(req, EPROTO);
+			return;
+		}
+
+		for (j = 0; j < dbmap->num; j++) {
+			ret = db_list_check_and_add(state->dblist,
+						    dbmap->dbs[j].db_id,
+						    dbmap->dbs[j].flags,
+						    pnn);
+			if (ret != 0) {
+				D_ERR("failed to add database list entry, "
+				      "ret=%d\n",
+				      ret);
+				tevent_req_error(req, ret);
+				return;
+			}
+		}
+
+		TALLOC_FREE(dbmap);
+	}
+
+	ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_ACTIVE);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_active_done, req);
+}
+
+static void recovery_active_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_req_control request;
+	struct ctdb_vnn_map *vnnmap;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("failed to set recovery mode ACTIVE on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("failed to set recovery mode ACTIVE, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	D_ERR("Set recovery mode to ACTIVE\n");
+
+	/* Calculate new VNNMAP */
+	vnnmap = talloc_zero(state, struct ctdb_vnn_map);
+	if (tevent_req_nomem(vnnmap, req)) {
+		return;
+	}
+
+	vnnmap->map = node_list_lmaster(state->nlist, vnnmap, &vnnmap->size);
+	if (tevent_req_nomem(vnnmap->map, req)) {
+		return;
+	}
+
+	if (vnnmap->size == 0) {
+		D_WARNING("No active lmasters found. Adding recmaster anyway\n");
+		vnnmap->map[0] = state->destnode;
+		vnnmap->size = 1;
+	}
+
+	vnnmap->generation = state->generation;
+
+	state->vnnmap = vnnmap;
+
+	ctdb_req_control_start_recovery(&request);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_start_recovery_done, req);
+}
+
+static void recovery_start_recovery_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_req_control request;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("failed to run start_recovery event on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("failed to run start_recovery event, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	D_ERR("start_recovery event finished\n");
+
+	ctdb_req_control_setvnnmap(&request, state->vnnmap);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_vnnmap_update_done, req);
+}
+
+static void recovery_vnnmap_update_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("failed to update VNNMAP on node %u, ret=%d\n",
+			      pnn, ret2);
+		} else {
+			D_ERR("failed to update VNNMAP, ret=%d\n", ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	D_NOTICE("updated VNNMAP\n");
+
+	subreq = db_recovery_send(state,
+				  state->ev,
+				  state->client,
+				  state->dblist,
+				  state->tun_list,
+				  state->nlist,
+				  state->vnnmap->generation);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_db_recovery_done, req);
+}
+
+static void recovery_db_recovery_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_req_control request;
+	bool status;
+	unsigned int count;
+
+	status = db_recovery_recv(subreq, &count);
+	TALLOC_FREE(subreq);
+
+	D_ERR("%d of %d databases recovered\n", count, state->dblist->num_dbs);
+
+	if (! status) {
+		subreq = ban_node_send(state,
+				       state->ev,
+				       state->client,
+				       state->tun_list,
+				       state->nlist);
+		if (tevent_req_nomem(subreq, req)) {
+			return;
+		}
+		tevent_req_set_callback(subreq, recovery_failed_done, req);
+		return;
+	}
+
+	ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_NORMAL);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_normal_done, req);
+}
+
+static void recovery_failed_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	int ret;
+	bool status;
+
+	status = ban_node_recv(subreq, &ret);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		D_ERR("failed to ban node, ret=%d\n", ret);
+	}
+
+	tevent_req_error(req, EIO);
+}
+
+static void recovery_normal_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	struct ctdb_req_control request;
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("failed to set recovery mode NORMAL on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("failed to set recovery mode NORMAL, ret=%d\n",
+			      ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	D_ERR("Set recovery mode to NORMAL\n");
+
+	ctdb_req_control_end_recovery(&request);
+	subreq = ctdb_client_control_multi_send(state,
+						state->ev,
+						state->client,
+						state->nlist->pnn_list,
+						state->nlist->count,
+						TIMEOUT(),
+						&request);
+	if (tevent_req_nomem(subreq, req)) {
+		return;
+	}
+	tevent_req_set_callback(subreq, recovery_end_recovery_done, req);
+}
+
+static void recovery_end_recovery_done(struct tevent_req *subreq)
+{
+	struct tevent_req *req = tevent_req_callback_data(
+		subreq, struct tevent_req);
+	struct recovery_state *state = tevent_req_data(
+		req, struct recovery_state);
+	int *err_list;
+	int ret;
+	bool status;
+
+	status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
+						NULL);
+	TALLOC_FREE(subreq);
+	if (! status) {
+		int ret2;
+		uint32_t pnn;
+
+		ret2 = ctdb_client_control_multi_error(state->nlist->pnn_list,
+						       state->nlist->count,
+						       err_list,
+						       &pnn);
+		if (ret2 != 0) {
+			D_ERR("failed to run recovered event on node %u,"
+			      " ret=%d\n", pnn, ret2);
+		} else {
+			D_ERR("failed to run recovered event, ret=%d\n", ret);
+		}
+		tevent_req_error(req, ret);
+		return;
+	}
+
+	D_ERR("recovered event finished\n");
+
+	tevent_req_done(req);
+}
+
+static void recovery_recv(struct tevent_req *req, int *perr)
+{
+	generic_recv(req, perr);
+}
+
+static void usage(const char *progname)
+{
+	fprintf(stderr, "\nUsage: %s <output-fd> <ctdb-socket-path> <generation>\n",
+		progname);
+}
+
+
+/*
+ * Arguments - log fd, write fd, socket path, generation
+ */
+int main(int argc, char *argv[])
+{
+	int write_fd;
+	const char *sockpath;
+	TALLOC_CTX *mem_ctx = NULL;
+	struct tevent_context *ev;
+	struct ctdb_client_context *client;
+	bool status;
+	int ret = 0;
+	struct tevent_req *req;
+	uint32_t generation;
+
+	if (argc != 4) {
+		usage(argv[0]);
+		exit(1);
+	}
+
+	write_fd = atoi(argv[1]);
+	sockpath = argv[2];
+	generation = (uint32_t)smb_strtoul(argv[3],
+					   NULL,
+					   0,
+					   &ret,
+					   SMB_STR_STANDARD);
+	if (ret != 0) {
+		fprintf(stderr, "recovery: unable to initialize generation\n");
+		goto failed;
+	}
+
+	mem_ctx = talloc_new(NULL);
+	if (mem_ctx == NULL) {
+		fprintf(stderr, "recovery: talloc_new() failed\n");
+		goto failed;
+	}
+
+	ret = logging_init(mem_ctx, NULL, NULL, "ctdb-recovery");
+	if (ret != 0) {
+		fprintf(stderr, "recovery: Unable to initialize logging\n");
+		goto failed;
+	}
+
+	ev = tevent_context_init(mem_ctx);
+	if (ev == NULL) {
+		D_ERR("tevent_context_init() failed\n");
+		goto failed;
+	}
+
+	status = logging_setup_sighup_handler(ev, mem_ctx, NULL, NULL);
+	if (!status) {
+		D_ERR("logging_setup_sighup_handler() failed\n");
+		goto failed;
+	}
+
+	ret = ctdb_client_init(mem_ctx, ev, sockpath, &client);
+	if (ret != 0) {
+		D_ERR("ctdb_client_init() failed, ret=%d\n", ret);
+		goto failed;
+	}
+
+	req = recovery_send(mem_ctx, ev, client, generation);
+	if (req == NULL) {
+		D_ERR("database_recover_send() failed\n");
+		goto failed;
+	}
+
+	if (! tevent_req_poll(req, ev)) {
+		D_ERR("tevent_req_poll() failed\n");
+		goto failed;
+	}
+
+	recovery_recv(req, &ret);
+	TALLOC_FREE(req);
+	if (ret != 0) {
+		D_ERR("database recovery failed, ret=%d\n", ret);
+		goto failed;
+	}
+
+	sys_write(write_fd, &ret, sizeof(ret));
+	return 0;
+
+failed:
+	TALLOC_FREE(mem_ctx);
+	return 1;
+}