Changeset 989 for vendor/current/ctdb/server

Timestamp:

Nov 25, 2016, 8:04:54 PM (9 years ago)

Author:

Silvan Scherrer

Message:

Samba Server: update vendor to version 4.4.7

Location:

vendor/current/ctdb/server

Files:

• ctdb_control.c (modified) (1 diff)
• ctdb_daemon.c (modified) (9 diffs)
• ctdb_logging.c (modified) (1 diff)
• ctdb_recover.c (modified) (4 diffs)
• ctdb_recoverd.c (modified) (8 diffs)
• ctdb_recovery_helper.c (modified) (50 diffs)
• ctdb_takeover.c (modified) (26 diffs)
• ctdb_tunables.c (modified) (2 diffs)
• ctdbd.c (modified) (1 diff)
• eventscript.c (modified) (4 diffs)

vendor/current/ctdb/server/ctdb_control.c

@@ -716 +716 @@
                 return ctdb_control_db_transaction_cancel(ctdb, indata);
 
+        case CTDB_CONTROL_DB_PULL:
+                CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_pulldb_ext));
+                return ctdb_control_db_pull(ctdb, c, indata, outdata);
+
+        case CTDB_CONTROL_DB_PUSH_START:
+                CHECK_CONTROL_DATA_SIZE(sizeof(struct ctdb_pulldb_ext));
+                return ctdb_control_db_push_start(ctdb, indata);
+
+        case CTDB_CONTROL_DB_PUSH_CONFIRM:
+                CHECK_CONTROL_DATA_SIZE(sizeof(uint32_t));
+                return ctdb_control_db_push_confirm(ctdb, indata, outdata);
+
         default:
                 DEBUG(DEBUG_CRIT,(__location__ " Unknown CTDB control opcode %u\n", opcode));

vendor/current/ctdb/server/ctdb_daemon.c

@@ -44 +44 @@
 #include "common/common.h"
 #include "common/logging.h"
+#include "common/pidfile.h"
 
 struct ctdb_client_pid_list {
@@ -53 +54 @@
 
 const char *ctdbd_pidfile = NULL;
+static struct pidfile_context *ctdbd_pidfile_ctx = NULL;
 
 static void daemon_incoming_packet(void *, struct ctdb_req_header *);
@@ -984 +986 @@
 {
         struct sockaddr_un addr;
+        int ret;
 
         ctdb->daemon.sd = socket(AF_UNIX, SOCK_STREAM, 0);
@@ -994 +997 @@
         strncpy(addr.sun_path, ctdb->daemon.name, sizeof(addr.sun_path)-1);
 
-        /* First check if an old ctdbd might be running */
-        if (connect(ctdb->daemon.sd,
-                    (struct sockaddr *)&addr, sizeof(addr)) == 0) {
-                DEBUG(DEBUG_CRIT,
-                      ("Something is already listening on ctdb socket '%s'\n",
-                       ctdb->daemon.name));
-                goto failed;
-        }
-
         /* Remove any old socket */
-        unlink(ctdb->daemon.name);
+        ret = unlink(ctdb->daemon.name);
+        if (ret == 0) {
+                DEBUG(DEBUG_WARNING,
+                      ("Removed stale socket %s\n", ctdb->daemon.name));
+        } else if (errno != ENOENT) {
+                DEBUG(DEBUG_ERR,
+                      ("Failed to remove stale socket %s\n", ctdb->daemon.name));
+                return -1;
+        }
 
         set_close_on_exec(ctdb->daemon.sd);
@@ -1128 +1130 @@
 static void ctdb_remove_pidfile(void)
 {
-        /* Only the main ctdbd's PID matches the SID */
-        if (ctdbd_pidfile != NULL && getsid(0) == getpid()) {
-                if (unlink(ctdbd_pidfile) == 0) {
-                        DEBUG(DEBUG_NOTICE, ("Removed PID file %s\n",
-                                             ctdbd_pidfile));
-                } else {
-                        DEBUG(DEBUG_WARNING, ("Failed to Remove PID file %s\n",
-                                              ctdbd_pidfile));
-                }
-        }
-}
-
-static void ctdb_create_pidfile(pid_t pid)
+        TALLOC_FREE(ctdbd_pidfile_ctx);
+}
+
+static void ctdb_create_pidfile(TALLOC_CTX *mem_ctx)
 {
         if (ctdbd_pidfile != NULL) {
-                FILE *fp;
-
-                fp = fopen(ctdbd_pidfile, "w");
-                if (fp == NULL) {
-                        DEBUG(DEBUG_ALERT,
-                              ("Failed to open PID file %s\n", ctdbd_pidfile));
+                int ret = pidfile_create(mem_ctx, ctdbd_pidfile,
+                                         &ctdbd_pidfile_ctx);
+                if (ret != 0) {
+                        DEBUG(DEBUG_ERR,
+                              ("Failed to create PID file %s\n",
+                               ctdbd_pidfile));
                         exit(11);
                 }
 
-                fprintf(fp, "%d\n", pid);
-                fclose(fp);
                 DEBUG(DEBUG_NOTICE, ("Created PID file %s\n", ctdbd_pidfile));
                 atexit(ctdb_remove_pidfile);
@@ -1214 +1205 @@
         int res, ret = -1;
         struct tevent_fd *fde;
+
+        if (do_fork && fork()) {
+                return 0;
+        }
+
+        if (do_fork) {
+                if (setsid() == -1) {
+                        ctdb_die(ctdb, "Failed to setsid()\n");
+                }
+                close(0);
+                if (open("/dev/null", O_RDONLY) != 0) {
+                        DEBUG(DEBUG_ALERT,(__location__ " Failed to setup stdin on /dev/null\n"));
+                        exit(11);
+                }
+        }
+        ignore_signal(SIGPIPE);
+        ignore_signal(SIGUSR1);
+
+        ctdb->ctdbd_pid = getpid();
+        DEBUG(DEBUG_ERR, ("Starting CTDBD (Version %s) as PID: %u\n",
+                          CTDB_VERSION_STRING, ctdb->ctdbd_pid));
+        ctdb_create_pidfile(ctdb);
 
         /* create a unix domain stream socket to listen to */
@@ -1222 +1235 @@
         }
 
-        if (do_fork && fork()) {
-                return 0;
-        }
-
-        tdb_reopen_all(false);
-
-        if (do_fork) {
-                if (setsid() == -1) {
-                        ctdb_die(ctdb, "Failed to setsid()\n");
-                }
-                close(0);
-                if (open("/dev/null", O_RDONLY) != 0) {
-                        DEBUG(DEBUG_ALERT,(__location__ " Failed to setup stdin on /dev/null\n"));
-                        exit(11);
-                }
-        }
-        ignore_signal(SIGPIPE);
-        ignore_signal(SIGUSR1);
-
-        ctdb->ctdbd_pid = getpid();
-        DEBUG(DEBUG_ERR, ("Starting CTDBD (Version %s) as PID: %u\n",
-                          CTDB_VERSION_STRING, ctdb->ctdbd_pid));
-        ctdb_create_pidfile(ctdb->ctdbd_pid);
-
         /* Make sure we log something when the daemon terminates.
          * This must be the first exit handler to run (so the last to
@@ -1261 +1250 @@
 
         ctdb->ev = tevent_context_init(NULL);
+        if (ctdb->ev == NULL) {
+                DEBUG(DEBUG_ALERT,("tevent_context_init() failed\n"));
+                exit(1);
+        }
         tevent_loop_allow_nesting(ctdb->ev);
         tevent_set_trace_callback(ctdb->ev, ctdb_tevent_trace, ctdb);
@@ -1846 +1839 @@
         /* get a new event context */
         ctdb->ev = tevent_context_init(ctdb);
+        if (ctdb->ev == NULL) {
+                DEBUG(DEBUG_ALERT,("tevent_context_init() failed\n"));
+                exit(1);
+        }
         tevent_loop_allow_nesting(ctdb->ev);
 

vendor/current/ctdb/server/ctdb_logging.c

@@ -282 +282 @@
         /* We'll fail if stderr/stdout not already open; it's simpler. */
         old_stdout = dup(STDOUT_FILENO);
+        if (old_stdout < 0) {
+                DEBUG(DEBUG_ERR, ("Failed to dup stdout for child logging\n"));
+                return -1;
+        }
         old_stderr = dup(STDERR_FILENO);
-        if (old_stdout < 0 || old_stderr < 0) {
-                DEBUG(DEBUG_ERR, ("Failed to dup stdout/stderr for child logging\n"));
+        if (old_stderr < 0) {
+                DEBUG(DEBUG_ERR, ("Failed to dup stderr for child logging\n"));
+                close(old_stdout);
                 return -1;
         }

vendor/current/ctdb/server/ctdb_recover.c

@@ -314 +314 @@
 }
 
+struct db_pull_state {
+        struct ctdb_context *ctdb;
+        struct ctdb_db_context *ctdb_db;
+        struct ctdb_marshall_buffer *recs;
+        uint32_t pnn;
+        uint64_t srvid;
+        uint32_t num_records;
+};
+
+static int traverse_db_pull(struct tdb_context *tdb, TDB_DATA key,
+                            TDB_DATA data, void *private_data)
+{
+        struct db_pull_state *state = (struct db_pull_state *)private_data;
+        struct ctdb_marshall_buffer *recs;
+
+        recs = ctdb_marshall_add(state->ctdb, state->recs,
+                                 state->ctdb_db->db_id, 0, key, NULL, data);
+        if (recs == NULL) {
+                TALLOC_FREE(state->recs);
+                return -1;
+        }
+        state->recs = recs;
+
+        if (talloc_get_size(state->recs) >=
+                        state->ctdb->tunable.rec_buffer_size_limit) {
+                TDB_DATA buffer;
+                int ret;
+
+                buffer = ctdb_marshall_finish(state->recs);
+                ret = ctdb_daemon_send_message(state->ctdb, state->pnn,
+                                               state->srvid, buffer);
+                if (ret != 0) {
+                        TALLOC_FREE(state->recs);
+                        return -1;
+                }
+
+                state->num_records += state->recs->count;
+                TALLOC_FREE(state->recs);
+        }
+
+        return 0;
+}
+
+int32_t ctdb_control_db_pull(struct ctdb_context *ctdb,
+                             struct ctdb_req_control_old *c,
+                             TDB_DATA indata, TDB_DATA *outdata)
+{
+        struct ctdb_pulldb_ext *pulldb_ext;
+        struct ctdb_db_context *ctdb_db;
+        struct db_pull_state state;
+        int ret;
+
+        pulldb_ext = (struct ctdb_pulldb_ext *)indata.dptr;
+
+        ctdb_db = find_ctdb_db(ctdb, pulldb_ext->db_id);
+        if (ctdb_db == NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n",
+                                 pulldb_ext->db_id));
+                return -1;
+        }
+
+        if (!ctdb_db_frozen(ctdb_db)) {
+                DEBUG(DEBUG_ERR,
+                      ("rejecting ctdb_control_pull_db when not frozen\n"));
+                return -1;
+        }
+
+        if (ctdb_db->unhealthy_reason) {
+                /* this is just a warning, as the tdb should be empty anyway */
+                DEBUG(DEBUG_WARNING,
+                      ("db(%s) unhealty in ctdb_control_db_pull: %s\n",
+                       ctdb_db->db_name, ctdb_db->unhealthy_reason));
+        }
+
+        state.ctdb = ctdb;
+        state.ctdb_db = ctdb_db;
+        state.recs = NULL;
+        state.pnn = c->hdr.srcnode;
+        state.srvid = pulldb_ext->srvid;
+        state.num_records = 0;
+
+        if (ctdb_lockdb_mark(ctdb_db) != 0) {
+                DEBUG(DEBUG_ERR,
+                      (__location__ " Failed to get lock on entire db - failing\n"));
+                return -1;
+        }
+
+        ret = tdb_traverse_read(ctdb_db->ltdb->tdb, traverse_db_pull, &state);
+        if (ret == -1) {
+                DEBUG(DEBUG_ERR,
+                      (__location__ " Failed to get traverse db '%s'\n",
+                       ctdb_db->db_name));
+                ctdb_lockdb_unmark(ctdb_db);
+                return -1;
+        }
+
+        /* Last few records */
+        if (state.recs != NULL) {
+                TDB_DATA buffer;
+
+                buffer = ctdb_marshall_finish(state.recs);
+                ret = ctdb_daemon_send_message(state.ctdb, state.pnn,
+                                               state.srvid, buffer);
+                if (ret != 0) {
+                        TALLOC_FREE(state.recs);
+                        ctdb_lockdb_unmark(ctdb_db);
+                        return -1;
+                }
+
+                state.num_records += state.recs->count;
+                TALLOC_FREE(state.recs);
+        }
+
+        ctdb_lockdb_unmark(ctdb_db);
+
+        outdata->dptr = talloc_size(outdata, sizeof(uint32_t));
+        if (outdata->dptr == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " Memory allocation error\n"));
+                return -1;
+        }
+
+        memcpy(outdata->dptr, (uint8_t *)&state.num_records, sizeof(uint32_t));
+        outdata->dsize = sizeof(uint32_t);
+
+        return 0;
+}
+
 /*
   push a bunch of records into a ltdb, filtering by rsn
@@ -406 +533 @@
         ctdb_lockdb_unmark(ctdb_db);
         return -1;
+}
+
+struct db_push_state {
+        struct ctdb_context *ctdb;
+        struct ctdb_db_context *ctdb_db;
+        uint64_t srvid;
+        uint32_t num_records;
+        bool failed;
+};
+
+static void db_push_msg_handler(uint64_t srvid, TDB_DATA indata,
+                                void *private_data)
+{
+        struct db_push_state *state = talloc_get_type(
+                private_data, struct db_push_state);
+        struct ctdb_marshall_buffer *recs;
+        struct ctdb_rec_data_old *rec;
+        int i, ret;
+
+        if (state->failed) {
+                return;
+        }
+
+        recs = (struct ctdb_marshall_buffer *)indata.dptr;
+        rec = (struct ctdb_rec_data_old *)&recs->data[0];
+
+        DEBUG(DEBUG_INFO, ("starting push of %u records for dbid 0x%x\n",
+                           recs->count, recs->db_id));
+
+        for (i=0; i<recs->count; i++) {
+                TDB_DATA key, data;
+                struct ctdb_ltdb_header *hdr;
+
+                key.dptr = &rec->data[0];
+                key.dsize = rec->keylen;
+                data.dptr = &rec->data[key.dsize];
+                data.dsize = rec->datalen;
+
+                if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
+                        DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record\n"));
+                        goto failed;
+                }
+
+                hdr = (struct ctdb_ltdb_header *)data.dptr;
+                /* Strip off any read only record flags.
+                 * All readonly records are revoked implicitely by a recovery.
+                 */
+                hdr->flags &= ~CTDB_REC_RO_FLAGS;
+
+                data.dptr += sizeof(*hdr);
+                data.dsize -= sizeof(*hdr);
+
+                ret = ctdb_ltdb_store(state->ctdb_db, key, hdr, data);
+                if (ret != 0) {
+                        DEBUG(DEBUG_ERR,
+                              (__location__ " Unable to store record\n"));
+                        goto failed;
+                }
+
+                rec = (struct ctdb_rec_data_old *)(rec->length + (uint8_t *)rec);
+        }
+
+        DEBUG(DEBUG_DEBUG, ("finished push of %u records for dbid 0x%x\n",
+                            recs->count, recs->db_id));
+
+        state->num_records += recs->count;
+        return;
+
+failed:
+        state->failed = true;
+}
+
+int32_t ctdb_control_db_push_start(struct ctdb_context *ctdb, TDB_DATA indata)
+{
+        struct ctdb_pulldb_ext *pulldb_ext;
+        struct ctdb_db_context *ctdb_db;
+        struct db_push_state *state;
+        int ret;
+
+        pulldb_ext = (struct ctdb_pulldb_ext *)indata.dptr;
+
+        ctdb_db = find_ctdb_db(ctdb, pulldb_ext->db_id);
+        if (ctdb_db == NULL) {
+                DEBUG(DEBUG_ERR,
+                      (__location__ " Unknown db 0x%08x\n", pulldb_ext->db_id));
+                return -1;
+        }
+
+        if (!ctdb_db_frozen(ctdb_db)) {
+                DEBUG(DEBUG_ERR,
+                      ("rejecting ctdb_control_db_push_start when not frozen\n"));
+                return -1;
+        }
+
+        if (ctdb_db->push_started) {
+                DEBUG(DEBUG_WARNING,
+                      (__location__ " DB push already started for %s\n",
+                       ctdb_db->db_name));
+
+                /* De-register old state */
+                state = (struct db_push_state *)ctdb_db->push_state;
+                if (state != NULL) {
+                        srvid_deregister(ctdb->srv, state->srvid, state);
+                        talloc_free(state);
+                        ctdb_db->push_state = NULL;
+                }
+        }
+
+        state = talloc_zero(ctdb_db, struct db_push_state);
+        if (state == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " Memory allocation error\n"));
+                return -1;
+        }
+
+        state->ctdb = ctdb;
+        state->ctdb_db = ctdb_db;
+        state->srvid = pulldb_ext->srvid;
+        state->failed = false;
+
+        ret = srvid_register(ctdb->srv, state, state->srvid,
+                             db_push_msg_handler, state);
+        if (ret != 0) {
+                DEBUG(DEBUG_ERR,
+                      (__location__ " Failed to register srvid for db push\n"));
+                talloc_free(state);
+                return -1;
+        }
+
+        if (ctdb_lockdb_mark(ctdb_db) != 0) {
+                DEBUG(DEBUG_ERR,
+                      (__location__ " Failed to get lock on entire db - failing\n"));
+                srvid_deregister(ctdb->srv, state->srvid, state);
+                talloc_free(state);
+                return -1;
+        }
+
+        ctdb_db->push_started = true;
+        ctdb_db->push_state = state;
+
+        return 0;
+}
+
+int32_t ctdb_control_db_push_confirm(struct ctdb_context *ctdb,
+                                     TDB_DATA indata, TDB_DATA *outdata)
+{
+        uint32_t db_id;
+        struct ctdb_db_context *ctdb_db;
+        struct db_push_state *state;
+
+        db_id = *(uint32_t *)indata.dptr;
+
+        ctdb_db = find_ctdb_db(ctdb, db_id);
+        if (ctdb_db == NULL) {
+                DEBUG(DEBUG_ERR,(__location__ " Unknown db 0x%08x\n", db_id));
+                return -1;
+        }
+
+        if (!ctdb_db_frozen(ctdb_db)) {
+                DEBUG(DEBUG_ERR,
+                      ("rejecting ctdb_control_db_push_confirm when not frozen\n"));
+                return -1;
+        }
+
+        if (!ctdb_db->push_started) {
+                DEBUG(DEBUG_ERR, (__location__ " DB push not started\n"));
+                return -1;
+        }
+
+        if (ctdb_db->readonly) {
+                DEBUG(DEBUG_ERR,
+                      ("Clearing the tracking database for dbid 0x%x\n",
+                       ctdb_db->db_id));
+                if (tdb_wipe_all(ctdb_db->rottdb) != 0) {
+                        DEBUG(DEBUG_ERR,
+                              ("Failed to wipe tracking database for 0x%x."
+                               " Dropping read-only delegation support\n",
+                               ctdb_db->db_id));
+                        ctdb_db->readonly = false;
+                        tdb_close(ctdb_db->rottdb);
+                        ctdb_db->rottdb = NULL;
+                        ctdb_db->readonly = false;
+                }
+
+                while (ctdb_db->revokechild_active != NULL) {
+                        talloc_free(ctdb_db->revokechild_active);
+                }
+        }
+
+        ctdb_lockdb_unmark(ctdb_db);
+
+        state = (struct db_push_state *)ctdb_db->push_state;
+        if (state == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " Missing push db state\n"));
+                return -1;
+        }
+
+        srvid_deregister(ctdb->srv, state->srvid, state);
+
+        outdata->dptr = talloc_size(outdata, sizeof(uint32_t));
+        if (outdata->dptr == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " Memory allocation error\n"));
+                talloc_free(state);
+                ctdb_db->push_state = NULL;
+                return -1;
+        }
+
+        memcpy(outdata->dptr, (uint8_t *)&state->num_records, sizeof(uint32_t));
+        outdata->dsize = sizeof(uint32_t);
+
+        talloc_free(state);
+        ctdb_db->push_started = false;
+        ctdb_db->push_state = NULL;
+
+        return 0;
 }
 
@@ -1033 +1374 @@
                 if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
                         DEBUG(DEBUG_CRIT,(__location__ " bad ltdb record in indata\n"));
+                        talloc_free(records);
                         return -1;
                 }
@@ -1225 +1567 @@
                         DEBUG(DEBUG_CRIT, (__location__ " bad ltdb record "
                                            "in indata\n"));
+                        talloc_free(records);
                         return -1;
                 }

vendor/current/ctdb/server/ctdb_recoverd.c

@@ -253 +253 @@
         uint32_t *force_rebalance_nodes;
         struct ctdb_node_capabilities *caps;
+        bool frozen_on_inactive;
 };
 
@@ -1782 +1783 @@
         }
 
+        setenv("CTDB_DBDIR_STATE", rec->ctdb->db_directory_state, 1);
+
         if (!ctdb_vfork_with_logging(state, rec->ctdb, "recovery", prog, nargs,
                                      args, NULL, NULL, &state->pid)) {
@@ -1981 +1984 @@
 
         DEBUG(DEBUG_NOTICE, (__location__ " Recovery - disabled recovery mode\n"));
+
+        /* execute the "recovered" event script on all nodes */
+        ret = run_recovered_eventscript(rec, nodemap, "do_recovery");
+        if (ret!=0) {
+                DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'recovered' event on cluster. Recovery process failed.\n"));
+                return -1;
+        }
+
+        DEBUG(DEBUG_NOTICE, (__location__ " Recovery - finished the recovered event\n"));
 
         return 0;
@@ -2157 +2169 @@
         do_takeover_run(rec, nodemap, false);
 
-        /* execute the "recovered" event script on all nodes */
-        ret = run_recovered_eventscript(rec, nodemap, "do_recovery");
-        if (ret!=0) {
-                DEBUG(DEBUG_ERR, (__location__ " Unable to run the 'recovered' event on cluster. Recovery process failed.\n"));
-                goto fail;
-        }
-
-        DEBUG(DEBUG_NOTICE, (__location__ " Recovery - finished the recovered event\n"));
-
         /* send a message to all clients telling them that the cluster 
            has been reconfigured */
@@ -2659 +2662 @@
 }
 
+/*
+ * handler for assigning banning credits
+ */
+static void banning_handler(uint64_t srvid, TDB_DATA data, void *private_data)
+{
+        struct ctdb_recoverd *rec = talloc_get_type(
+                private_data, struct ctdb_recoverd);
+        uint32_t ban_pnn;
+
+        /* Ignore if we are not recmaster */
+        if (rec->ctdb->pnn != rec->recmaster) {
+                return;
+        }
+
+        if (data.dsize != sizeof(uint32_t)) {
+                DEBUG(DEBUG_ERR, (__location__ "invalid data size %zu\n",
+                                  data.dsize));
+                return;
+        }
+
+        ban_pnn = *(uint32_t *)data.dptr;
+
+        ctdb_set_culprit_count(rec, ban_pnn, rec->nodemap->num);
+}
 
 /*
@@ -3490 +3517 @@
                                 return;
                         }
-                        ret = ctdb_ctrl_freeze(ctdb, CONTROL_TIMEOUT(), CTDB_CURRENT_NODE);
+                }
+                if (! rec->frozen_on_inactive) {
+                        ret = ctdb_ctrl_freeze(ctdb, CONTROL_TIMEOUT(),
+                                               CTDB_CURRENT_NODE);
                         if (ret != 0) {
-                                DEBUG(DEBUG_ERR,(__location__ " Failed to freeze node in STOPPED or BANNED state\n"));
+                                DEBUG(DEBUG_ERR,
+                                      (__location__ " Failed to freeze node "
+                                       "in STOPPED or BANNED state\n"));
                                 return;
                         }
+
+                        rec->frozen_on_inactive = true;
                 }
 
@@ -3503 +3537 @@
                 return;
         }
+
+        rec->frozen_on_inactive = false;
 
         /* If we are not the recmaster then do some housekeeping */
@@ -3883 +3919 @@
 
         rec->priority_time = timeval_current();
+        rec->frozen_on_inactive = false;
 
         /* register a message port for sending memory dumps */
         ctdb_client_set_message_handler(ctdb, CTDB_SRVID_MEM_DUMP, mem_dump_handler, rec);
+
+        /* when a node is assigned banning credits */
+        ctdb_client_set_message_handler(ctdb, CTDB_SRVID_BANNING,
+                                        banning_handler, rec);
 
         /* register a message port for recovery elections */

vendor/current/ctdb/server/ctdb_recovery_helper.c

@@ -35 +35 @@
 #include "client/client.h"
 
-#define TIMEOUT()       timeval_current_ofs(10, 0)
+static int recover_timeout = 30;
+
+#define NUM_RETRIES     3
+
+#define TIMEOUT()       timeval_current_ofs(recover_timeout, 0)
 
 static void LOG(const char *fmt, ...)
@@ -64 +68 @@
 }
 
+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;
+}
+
 /*
  * Recovery database functions
@@ -81 +107 @@
                                           uint32_t hash_size, bool persistent)
 {
+        static char *db_dir_state = NULL;
         struct recdb_context *recdb;
         unsigned int tdb_flags;
@@ -89 +116 @@
         }
 
+        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",
-                                         dirname(discard_const(db_path)),
+                                         db_dir_state != NULL ?
+                                            db_dir_state :
+                                            dirname(discard_const(db_path)),
                                          db_name);
         if (recdb->db_path == NULL) {
@@ -113 +146 @@
 }
 
+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;
 }
 
@@ -141 +194 @@
 
         /* fetch the existing record, if any */
-        prev_data = tdb_fetch(state->recdb->db->tdb, key);
+        prev_data = tdb_fetch(recdb_tdb(state->recdb), key);
 
         if (prev_data.dptr != NULL) {
@@ -155 +208 @@
         }
 
-        ret = tdb_store(state->recdb->db->tdb, key, data, TDB_REPLACE);
+        ret = tdb_store(recdb_tdb(state->recdb), key, data, TDB_REPLACE);
         if (ret != 0) {
                 return -1;
@@ -179 +232 @@
 }
 
-struct recdb_traverse_state {
-        struct ctdb_rec_buffer *recbuf;
-        uint32_t pnn;
-        uint32_t reqid;
-        bool persistent;
-        bool failed;
-};
-
-static int recdb_traverse(struct tdb_context *tdb, TDB_DATA key, TDB_DATA data,
-                          void *private_data)
-{
-        struct recdb_traverse_state *state =
-                (struct recdb_traverse_state *)private_data;
+/* 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;
@@ -225 +270 @@
          * data structures built from the various tdb-level records.
          */
-        if (!state->persistent &&
-            data.dsize <= sizeof(struct ctdb_ltdb_header)) {
+        if (!persistent && data.dsize <= sizeof(struct ctdb_ltdb_header)) {
                 return 0;
         }
@@ -232 +276 @@
         /* update the dmaster field to point to us */
         header = (struct ctdb_ltdb_header *)data.dptr;
-        if (!state->persistent) {
-                header->dmaster = state->pnn;
+        if (!persistent) {
+                header->dmaster = dmaster;
                 header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
         }
 
-        ret = ctdb_rec_buffer_add(state->recbuf, state->recbuf, state->reqid,
-                                  NULL, key, data);
+        ret = ctdb_rec_buffer_add(recbuf, recbuf, reqid, NULL, key, data);
+        if (ret != 0) {
+                return ret;
+        }
+
+        return 0;
+}
+
+struct recdb_records_traverse_state {
+        struct ctdb_rec_buffer *recbuf;
+        uint32_t dmaster;
+        uint32_t reqid;
+        bool persistent;
+        bool failed;
+};
+
+static int recdb_records_traverse(struct tdb_context *tdb,
+                                  TDB_DATA key, TDB_DATA data,
+                                  void *private_data)
+{
+        struct recdb_records_traverse_state *state =
+                (struct recdb_records_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;
@@ -248 +316 @@
 
 static struct ctdb_rec_buffer *recdb_records(struct recdb_context *recdb,
-                                             TALLOC_CTX *mem_ctx, uint32_t pnn)
-{
-        struct recdb_traverse_state state;
-        int ret;
-
-        state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb->db_id);
+                                             TALLOC_CTX *mem_ctx,
+                                             uint32_t dmaster)
+{
+        struct recdb_records_traverse_state state;
+        int ret;
+
+        state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
         if (state.recbuf == NULL) {
                 return NULL;
         }
-        state.pnn = pnn;
+        state.dmaster = dmaster;
         state.reqid = 0;
-        state.persistent = recdb->persistent;
+        state.persistent = recdb_persistent(recdb);
         state.failed = false;
 
-        ret = tdb_traverse_read(recdb->db->tdb, recdb_traverse, &state);
+        ret = tdb_traverse_read(recdb_tdb(recdb), recdb_records_traverse,
+                                &state);
+        if (ret == -1 || state.failed) {
+                LOG("Failed to marshall recovery records for %s\n",
+                    recdb_name(recdb));
+                TALLOC_FREE(state.recbuf);
+                return NULL;
+        }
+
+        return state.recbuf;
+}
+
+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;
+        int max_size;
+        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) {
+                        LOG("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) {
+                LOG("Failed to collect recovery records for %s\n",
+                    recdb_name(recdb));
+                TALLOC_FREE(state.recbuf);
+                return -1;
+        }
+        state.num_buffers += 1;
+
+        LOG("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;
+        int num_records;
+};
+
+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_old_done(struct tevent_req *subreq);
+static void pull_database_unregister_done(struct tevent_req *subreq);
+static void pull_database_new_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, uint32_t caps,
+                        struct recdb_context *recdb)
+{
+        struct tevent_req *req, *subreq;
+        struct pull_database_state *state;
+        struct ctdb_req_control request;
+
+        req = tevent_req_create(mem_ctx, &state, struct pull_database_state);
+        if (req == NULL) {
                 return NULL;
         }
 
-        return state.recbuf;
+        state->ev = ev;
+        state->client = client;
+        state->recdb = recdb;
+        state->pnn = pnn;
+        state->srvid = srvid_next();
+
+        if (caps & CTDB_CAP_FRAGMENTED_CONTROLS) {
+                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);
+
+        } else {
+                struct ctdb_pulldb pulldb;
+
+                pulldb.db_id = recdb_id(recdb);
+                pulldb.lmaster = CTDB_LMASTER_ANY;
+
+                ctdb_req_control_pull_db(&request, &pulldb);
+                subreq = ctdb_client_control_send(state, state->ev,
+                                                  state->client,
+                                                  pnn, TIMEOUT(),
+                                                  &request);
+                if (tevent_req_nomem(subreq, req)) {
+                        return tevent_req_post(req, ev);
+                }
+                tevent_req_set_callback(subreq, pull_database_old_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;
+        int ret;
+        bool status;
+
+        if (srvid != state->srvid) {
+                return;
+        }
+
+        ret = ctdb_rec_buffer_pull(data.dptr, data.dsize, state, &recbuf);
+        if (ret != 0) {
+                LOG("Invalid data received for DB_PULL messages\n");
+                return;
+        }
+
+        if (recbuf->db_id != recdb_id(state->recdb)) {
+                talloc_free(recbuf);
+                LOG("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);
+                LOG("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) {
+                LOG("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_new_done, req);
+}
+
+static void pull_database_old_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;
+        struct ctdb_rec_buffer *recbuf;
+        int ret;
+        bool status;
+
+        status = ctdb_client_control_recv(subreq, &ret, state, &reply);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                LOG("control PULL_DB failed for %s on node %u, ret=%d\n",
+                    recdb_name(state->recdb), state->pnn, ret);
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        ret = ctdb_reply_control_pull_db(reply, state, &recbuf);
+        talloc_free(reply);
+        if (ret != 0) {
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
+                           recbuf);
+        if (! status) {
+                talloc_free(recbuf);
+                tevent_req_error(req, EIO);
+                return;
+        }
+
+        state->num_records = recbuf->count;
+        talloc_free(recbuf);
+
+        LOG("Pulled %d records for db %s from node %d\n",
+            state->num_records, recdb_name(state->recdb), state->pnn);
+
+        tevent_req_done(req);
+}
+
+static void pull_database_new_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) {
+                LOG("control DB_PULL failed for %s on node %u, ret=%d\n",
+                    recdb_name(state->recdb), state->pnn, ret);
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        ret = ctdb_reply_control_db_pull(reply, &num_records);
+        talloc_free(reply);
+        if (num_records != state->num_records) {
+                LOG("mismatch (%u != %u) in DB_PULL records for %s\n",
+                    num_records, state->num_records, recdb_name(state->recdb));
+                tevent_req_error(req, EIO);
+                return;
+        }
+
+        LOG("Pulled %d records for db %s from node %d\n",
+            state->num_records, recdb_name(state->recdb), state->pnn);
+
+        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) {
+                LOG("failed to remove message handler for DB_PULL for %s\n",
+                    recdb_name(state->recdb));
+                tevent_req_error(req, ret);
+                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 (old style)
+ */
+
+struct push_database_old_state {
+        struct tevent_context *ev;
+        struct ctdb_client_context *client;
+        struct recdb_context *recdb;
+        uint32_t *pnn_list;
+        int count;
+        struct ctdb_rec_buffer *recbuf;
+        int index;
+};
+
+static void push_database_old_push_done(struct tevent_req *subreq);
+
+static struct tevent_req *push_database_old_send(
+                        TALLOC_CTX *mem_ctx,
+                        struct tevent_context *ev,
+                        struct ctdb_client_context *client,
+                        uint32_t *pnn_list, int count,
+                        struct recdb_context *recdb)
+{
+        struct tevent_req *req, *subreq;
+        struct push_database_old_state *state;
+        struct ctdb_req_control request;
+        uint32_t pnn;
+
+        req = tevent_req_create(mem_ctx, &state,
+                                struct push_database_old_state);
+        if (req == NULL) {
+                return NULL;
+        }
+
+        state->ev = ev;
+        state->client = client;
+        state->recdb = recdb;
+        state->pnn_list = pnn_list;
+        state->count = count;
+        state->index = 0;
+
+        state->recbuf = recdb_records(recdb, state,
+                                      ctdb_client_pnn(client));
+        if (tevent_req_nomem(state->recbuf, req)) {
+                return tevent_req_post(req, ev);
+        }
+
+        pnn = state->pnn_list[state->index];
+
+        ctdb_req_control_push_db(&request, state->recbuf);
+        subreq = ctdb_client_control_send(state, ev, client, pnn,
+                                          TIMEOUT(), &request);
+        if (tevent_req_nomem(subreq, req)) {
+                return tevent_req_post(req, ev);
+        }
+        tevent_req_set_callback(subreq, push_database_old_push_done, req);
+
+        return req;
+}
+
+static void push_database_old_push_done(struct tevent_req *subreq)
+{
+        struct tevent_req *req = tevent_req_callback_data(
+                subreq, struct tevent_req);
+        struct push_database_old_state *state = tevent_req_data(
+                req, struct push_database_old_state);
+        struct ctdb_req_control request;
+        uint32_t pnn;
+        int ret;
+        bool status;
+
+        status = ctdb_client_control_recv(subreq, &ret, NULL, NULL);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                LOG("control PUSH_DB failed for db %s on node %u, ret=%d\n",
+                    recdb_name(state->recdb), state->pnn_list[state->index],
+                    ret);
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        state->index += 1;
+        if (state->index == state->count) {
+                TALLOC_FREE(state->recbuf);
+                tevent_req_done(req);
+                return;
+        }
+
+        pnn = state->pnn_list[state->index];
+
+        ctdb_req_control_push_db(&request, state->recbuf);
+        subreq = ctdb_client_control_send(state, state->ev, state->client,
+                                          pnn, TIMEOUT(), &request);
+        if (tevent_req_nomem(subreq, req)) {
+                return;
+        }
+        tevent_req_set_callback(subreq, push_database_old_push_done, req);
+}
+
+static bool push_database_old_recv(struct tevent_req *req, int *perr)
+{
+        return generic_recv(req, perr);
+}
+
+/*
+ * Push database to specified nodes (new style)
+ */
+
+struct push_database_new_state {
+        struct tevent_context *ev;
+        struct ctdb_client_context *client;
+        struct recdb_context *recdb;
+        uint32_t *pnn_list;
+        int count;
+        uint64_t srvid;
+        uint32_t dmaster;
+        int fd;
+        int num_buffers;
+        int num_buffers_sent;
+        int num_records;
+};
+
+static void push_database_new_started(struct tevent_req *subreq);
+static void push_database_new_send_msg(struct tevent_req *req);
+static void push_database_new_send_done(struct tevent_req *subreq);
+static void push_database_new_confirmed(struct tevent_req *subreq);
+
+static struct tevent_req *push_database_new_send(
+                        TALLOC_CTX *mem_ctx,
+                        struct tevent_context *ev,
+                        struct ctdb_client_context *client,
+                        uint32_t *pnn_list, int count,
+                        struct recdb_context *recdb,
+                        int max_size)
+{
+        struct tevent_req *req, *subreq;
+        struct push_database_new_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_new_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_new_started, req);
+
+        return req;
+}
+
+static void push_database_new_started(struct tevent_req *subreq)
+{
+        struct tevent_req *req = tevent_req_callback_data(
+                subreq, struct tevent_req);
+        struct push_database_new_state *state = tevent_req_data(
+                req, struct push_database_new_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) {
+                        LOG("control DB_PUSH_START failed for db %s "
+                            "on node %u, ret=%d\n",
+                            recdb_name(state->recdb), pnn, ret2);
+                } else {
+                        LOG("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_new_send_msg(req);
+}
+
+static void push_database_new_send_msg(struct tevent_req *req)
+{
+        struct push_database_new_state *state = tevent_req_data(
+                req, struct push_database_new_state);
+        struct tevent_req *subreq;
+        struct ctdb_rec_buffer *recbuf;
+        struct ctdb_req_message message;
+        TDB_DATA data;
+        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_new_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);
+
+        message.srvid = state->srvid;
+        message.data.data = data;
+
+        LOG("Pushing buffer %d with %d records for %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_new_send_done, req);
+
+        state->num_records += recbuf->count;
+
+        talloc_free(data.dptr);
+        talloc_free(recbuf);
+}
+
+static void push_database_new_send_done(struct tevent_req *subreq)
+{
+        struct tevent_req *req = tevent_req_callback_data(
+                subreq, struct tevent_req);
+        struct push_database_new_state *state = tevent_req_data(
+                req, struct push_database_new_state);
+        bool status;
+        int ret;
+
+        status = ctdb_client_message_multi_recv(subreq, &ret, NULL, NULL);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                LOG("Sending recovery records failed for %s\n",
+                    recdb_name(state->recdb));
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        state->num_buffers_sent += 1;
+
+        push_database_new_send_msg(req);
+}
+
+static void push_database_new_confirmed(struct tevent_req *subreq)
+{
+        struct tevent_req *req = tevent_req_callback_data(
+                subreq, struct tevent_req);
+        struct push_database_new_state *state = tevent_req_data(
+                req, struct push_database_new_state);
+        struct ctdb_reply_control **reply;
+        int *err_list;
+        bool status;
+        int ret, i;
+        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) {
+                        LOG("control DB_PUSH_CONFIRM failed for %s on node %u,"
+                            " ret=%d\n", recdb_name(state->recdb), pnn, ret2);
+                } else {
+                        LOG("control DB_PUSH_CONFIRM failed for %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) {
+                        LOG("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);
+
+        LOG("Pushed %d records for db %s\n",
+            state->num_records, recdb_name(state->recdb));
+
+        tevent_req_done(req);
+}
+
+static bool push_database_new_recv(struct tevent_req *req, int *perr)
+{
+        return generic_recv(req, perr);
+}
+
+/*
+ * wrapper for push_database_old and push_database_new
+ */
+
+struct push_database_state {
+        bool old_done, new_done;
+};
+
+static void push_database_old_done(struct tevent_req *subreq);
+static void push_database_new_done(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, int count, uint32_t *caps,
+                        struct ctdb_tunable_list *tun_list,
+                        struct recdb_context *recdb)
+{
+        struct tevent_req *req, *subreq;
+        struct push_database_state *state;
+        uint32_t *old_list, *new_list;
+        int old_count, new_count;
+        int i;
+
+        req = tevent_req_create(mem_ctx, &state, struct push_database_state);
+        if (req == NULL) {
+                return NULL;
+        }
+
+        state->old_done = false;
+        state->new_done = false;
+
+        old_count = 0;
+        new_count = 0;
+        old_list = talloc_array(state, uint32_t, count);
+        new_list = talloc_array(state, uint32_t, count);
+        if (tevent_req_nomem(old_list, req) ||
+            tevent_req_nomem(new_list,req)) {
+                return tevent_req_post(req, ev);
+        }
+
+        for (i=0; i<count; i++) {
+                uint32_t pnn = pnn_list[i];
+
+                if (caps[pnn] & CTDB_CAP_FRAGMENTED_CONTROLS) {
+                        new_list[new_count] = pnn;
+                        new_count += 1;
+                } else {
+                        old_list[old_count] = pnn;
+                        old_count += 1;
+                }
+        }
+
+        if (old_count > 0) {
+                subreq = push_database_old_send(state, ev, client,
+                                                old_list, old_count, recdb);
+                if (tevent_req_nomem(subreq, req)) {
+                        return tevent_req_post(req, ev);
+                }
+                tevent_req_set_callback(subreq, push_database_old_done, req);
+        } else {
+                state->old_done = true;
+        }
+
+        if (new_count > 0) {
+                subreq = push_database_new_send(state, ev, client,
+                                                new_list, new_count, recdb,
+                                                tun_list->rec_buffer_size_limit);
+                if (tevent_req_nomem(subreq, req)) {
+                        return tevent_req_post(req, ev);
+                }
+                tevent_req_set_callback(subreq, push_database_new_done, req);
+        } else {
+                state->new_done = true;
+        }
+
+        return req;
+}
+
+static void push_database_old_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 = push_database_old_recv(subreq, &ret);
+        if (! status) {
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        state->old_done = true;
+
+        if (state->old_done && state->new_done) {
+                tevent_req_done(req);
+        }
+}
+
+static void push_database_new_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 = push_database_new_recv(subreq, &ret);
+        if (! status) {
+                tevent_req_error(req, ret);
+                return;
+        }
+
+        state->new_done = true;
+
+        if (state->old_done && state->new_done) {
+                tevent_req_done(req);
+        }
+}
+
+static bool push_database_recv(struct tevent_req *req, int *perr)
+{
+        return generic_recv(req, perr);
 }
 
@@ -280 +1218 @@
         uint32_t *pnn_list;
         int count;
+        uint32_t *caps;
+        uint32_t *ban_credits;
         uint32_t db_id;
         struct recdb_context *recdb;
@@ -292 +1232 @@
                         struct tevent_context *ev,
                         struct ctdb_client_context *client,
-                        uint32_t *pnn_list, int count,
-                        uint32_t db_id, struct recdb_context *recdb)
+                        uint32_t *pnn_list, int count, uint32_t *caps,
+                        uint32_t *ban_credits, uint32_t db_id,
+                        struct recdb_context *recdb)
 {
         struct tevent_req *req, *subreq;
@@ -309 +1250 @@
         state->pnn_list = pnn_list;
         state->count = count;
+        state->caps = caps;
+        state->ban_credits = ban_credits;
         state->db_id = db_id;
         state->recdb = recdb;
@@ -332 +1275 @@
                 req, struct collect_highseqnum_db_state);
         struct ctdb_reply_control **reply;
-        struct ctdb_req_control request;
-        struct ctdb_pulldb pulldb;
         int *err_list;
         bool status;
@@ -380 +1321 @@
             recdb_name(state->recdb), state->max_pnn, max_seqnum);
 
-        pulldb.db_id = state->db_id;
-        pulldb.lmaster = CTDB_LMASTER_ANY;
-
-        ctdb_req_control_pull_db(&request, &pulldb);
-        subreq = ctdb_client_control_send(state, state->ev, state->client,
-                                          state->max_pnn, TIMEOUT(), &request);
+        subreq = pull_database_send(state, state->ev, state->client,
+                                    state->max_pnn,
+                                    state->caps[state->max_pnn],
+                                    state->recdb);
         if (tevent_req_nomem(subreq, req)) {
                 return;
@@ -399 +1338 @@
         struct collect_highseqnum_db_state *state = tevent_req_data(
                 req, struct collect_highseqnum_db_state);
-        struct ctdb_reply_control *reply;
-        struct ctdb_rec_buffer *recbuf;
-        int ret;
-        bool status;
-
-        status = ctdb_client_control_recv(subreq, &ret, state, &reply);
-        TALLOC_FREE(subreq);
-        if (! status) {
-                LOG("control PULL_DB failed for %s on node %u, ret=%d\n",
-                    recdb_name(state->recdb), state->max_pnn, ret);
-                tevent_req_error(req, ret);
-                return;
-        }
-
-        ret = ctdb_reply_control_pull_db(reply, state, &recbuf);
-        if (ret != 0) {
-                tevent_req_error(req, EPROTO);
-                return;
-        }
-
-        talloc_free(reply);
-
-        ret = recdb_add(state->recdb, ctdb_client_pnn(state->client), recbuf);
-        talloc_free(recbuf);
-        if (! ret) {
-                tevent_req_error(req, EIO);
+        int ret;
+        bool status;
+
+        status = pull_database_recv(subreq, &ret);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                state->ban_credits[state->max_pnn] += 1;
+                tevent_req_error(req, ret);
                 return;
         }
@@ -433 +1354 @@
 static bool collect_highseqnum_db_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;
+        return generic_recv(req, perr);
 }
 
@@ -454 +1366 @@
         uint32_t *pnn_list;
         int count;
+        uint32_t *caps;
+        uint32_t *ban_credits;
         uint32_t db_id;
         struct recdb_context *recdb;
@@ -466 +1380 @@
                         struct tevent_context *ev,
                         struct ctdb_client_context *client,
-                        uint32_t *pnn_list, int count,
-                        uint32_t db_id, struct recdb_context *recdb)
+                        uint32_t *pnn_list, int count, uint32_t *caps,
+                        uint32_t *ban_credits, uint32_t db_id,
+                        struct recdb_context *recdb)
 {
         struct tevent_req *req, *subreq;
         struct collect_all_db_state *state;
-        struct ctdb_req_control request;
+        uint32_t pnn;
 
         req = tevent_req_create(mem_ctx, &state,
@@ -483 +1398 @@
         state->pnn_list = pnn_list;
         state->count = count;
+        state->caps = caps;
+        state->ban_credits = ban_credits;
         state->db_id = db_id;
         state->recdb = recdb;
-
-        state->pulldb.db_id = db_id;
-        state->pulldb.lmaster = CTDB_LMASTER_ANY;
-
         state->index = 0;
 
-        ctdb_req_control_pull_db(&request, &state->pulldb);
-        subreq = ctdb_client_control_send(state, ev, client,
-                                          state->pnn_list[state->index],
-                                          TIMEOUT(), &request);
+        pnn = state->pnn_list[state->index];
+
+        subreq = pull_database_send(state, ev, client, pnn, caps[pnn], recdb);
         if (tevent_req_nomem(subreq, req)) {
                 return tevent_req_post(req, ev);
@@ -509 +1421 @@
         struct collect_all_db_state *state = tevent_req_data(
                 req, struct collect_all_db_state);
-        struct ctdb_reply_control *reply;
-        struct ctdb_req_control request;
-        struct ctdb_rec_buffer *recbuf;
-        int ret;
-        bool status;
-
-        status = ctdb_client_control_recv(subreq, &ret, state, &reply);
-        TALLOC_FREE(subreq);
-        if (! status) {
-                LOG("control PULL_DB failed for %s from node %u, ret=%d\n",
-                    recdb_name(state->recdb), state->pnn_list[state->index],
-                    ret);
-                tevent_req_error(req, ret);
-                return;
-        }
-
-        ret = ctdb_reply_control_pull_db(reply, state, &recbuf);
-        if (ret != 0) {
-                LOG("control PULL_DB failed for %s, ret=%d\n",
-                    recdb_name(state->recdb), ret);
-                tevent_req_error(req, EPROTO);
-                return;
-        }
-
-        talloc_free(reply);
-
-        status = recdb_add(state->recdb, ctdb_client_pnn(state->client), recbuf);
-        talloc_free(recbuf);
-        if (! status) {
-                tevent_req_error(req, EIO);
+        uint32_t pnn;
+        int ret;
+        bool status;
+
+        status = pull_database_recv(subreq, &ret);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                pnn = state->pnn_list[state->index];
+                state->ban_credits[pnn] += 1;
+                tevent_req_error(req, ret);
                 return;
         }
@@ -548 +1440 @@
         }
 
-        ctdb_req_control_pull_db(&request, &state->pulldb);
-        subreq = ctdb_client_control_send(state, state->ev, state->client,
-                                          state->pnn_list[state->index],
-                                          TIMEOUT(), &request);
+        pnn = state->pnn_list[state->index];
+        subreq = pull_database_send(state, state->ev, state->client,
+                                    pnn, state->caps[pnn], state->recdb);
         if (tevent_req_nomem(subreq, req)) {
                 return;
@@ -560 +1451 @@
 static bool collect_all_db_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;
+        return generic_recv(req, perr);
 }
 
@@ -592 +1474 @@
         uint32_t *pnn_list;
         int count;
+        uint32_t *caps;
+        uint32_t *ban_credits;
         uint32_t db_id;
         bool persistent;
@@ -600 +1484 @@
         const char *db_name, *db_path;
         struct recdb_context *recdb;
-        struct ctdb_rec_buffer *recbuf;
-
 };
 
@@ -619 +1501 @@
                                           struct ctdb_tunable_list *tun_list,
                                           uint32_t *pnn_list, int count,
+                                          uint32_t *caps,
+                                          uint32_t *ban_credits,
                                           uint32_t generation,
                                           uint32_t db_id, bool persistent)
@@ -636 +1520 @@
         state->pnn_list = pnn_list;
         state->count = count;
+        state->caps = caps;
+        state->ban_credits = ban_credits;
         state->db_id = db_id;
         state->persistent = persistent;
@@ -820 +1706 @@
                 subreq = collect_highseqnum_db_send(
                                 state, state->ev, state->client,
-                                state->pnn_list, state->count,
-                                state->db_id, state->recdb);
+                                state->pnn_list, state->count, state->caps,
+                                state->ban_credits, state->db_id,
+                                state->recdb);
         } else {
                 subreq = collect_all_db_send(
                                 state, state->ev, state->client,
-                                state->pnn_list, state->count,
-                                state->db_id, state->recdb);
+                                state->pnn_list, state->count, state->caps,
+                                state->ban_credits, state->db_id,
+                                state->recdb);
         }
         if (tevent_req_nomem(subreq, req)) {
@@ -872 +1760 @@
         struct recover_db_state *state = tevent_req_data(
                 req, struct recover_db_state);
-        struct ctdb_req_control request;
         int *err_list;
         int ret;
@@ -898 +1785 @@
         }
 
-        state->recbuf = recdb_records(state->recdb, state, state->destnode);
-        if (tevent_req_nomem(state->recbuf, req)) {
-                return;
-        }
-
-        TALLOC_FREE(state->recdb);
-
-        ctdb_req_control_push_db(&request, state->recbuf);
-        subreq = ctdb_client_control_multi_send(state, state->ev,
-                                                state->client,
-                                                state->pnn_list, state->count,
-                                                TIMEOUT(), &request);
+        subreq = push_database_send(state, state->ev, state->client,
+                                    state->pnn_list, state->count,
+                                    state->caps, state->tun_list,
+                                    state->recdb);
         if (tevent_req_nomem(subreq, req)) {
                 return;
@@ -923 +1802 @@
                 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->pnn_list,
-                                                       state->count,
-                                                       err_list, &pnn);
-                if (ret2 != 0) {
-                        LOG("control PUSHDB failed for db %s on node %u,"
-                            " ret=%d\n", state->db_name, pnn, ret2);
-                } else {
-                        LOG("control PUSHDB failed for db %s, ret=%d\n",
-                            state->db_name, ret);
-                }
-                tevent_req_error(req, ret);
-                return;
-        }
-
-        TALLOC_FREE(state->recbuf);
+        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);
@@ -1040 +1904 @@
 static bool recover_db_recv(struct tevent_req *req)
 {
-        int err;
-
-        if (tevent_req_is_unix_error(req, &err)) {
-                return false;
-        }
-
-        return true;
+        return generic_recv(req, NULL);
 }
 
@@ -1070 +1928 @@
         uint32_t *pnn_list;
         int count;
+        uint32_t *caps;
+        uint32_t *ban_credits;
         uint32_t generation;
         uint32_t db_id;
@@ -1084 +1944 @@
                                            struct ctdb_tunable_list *tun_list,
                                            uint32_t *pnn_list, int count,
+                                           uint32_t *caps,
+                                           uint32_t *ban_credits,
                                            uint32_t generation)
 {
@@ -1119 +1981 @@
                 substate->pnn_list = pnn_list;
                 substate->count = count;
+                substate->caps = caps;
+                substate->ban_credits = ban_credits;
                 substate->generation = generation;
                 substate->db_id = dbmap->dbs[i].db_id;
@@ -1125 +1989 @@
 
                 subreq = recover_db_send(state, ev, client, tun_list,
-                                         pnn_list, count, generation,
-                                         substate->db_id,
+                                         pnn_list, count, caps, ban_credits,
+                                         generation, substate->db_id,
                                          substate->persistent);
                 if (tevent_req_nomem(subreq, req)) {
@@ -1157 +2021 @@
 
         substate->num_fails += 1;
-        if (substate->num_fails < 5) {
+        if (substate->num_fails < NUM_RETRIES) {
                 subreq = recover_db_send(state, state->ev, substate->client,
                                          substate->tun_list,
                                          substate->pnn_list, substate->count,
+                                         substate->caps, substate->ban_credits,
                                          substate->generation, substate->db_id,
                                          substate->persistent);
@@ -1207 +2072 @@
  * Run the parallel database recovery
  *
+ * - Get tunables
  * - Get nodemap
  * - Get vnnmap
  * - Get capabilities from all nodes
- * - Get tunables from all nodes
  * - Get dbmap
  * - Set RECOVERY_ACTIVE
@@ -1229 +2094 @@
         struct ctdb_node_map *nodemap;
         uint32_t *caps;
+        uint32_t *ban_credits;
         struct ctdb_tunable_list *tun_list;
         struct ctdb_vnn_map *vnnmap;
@@ -1234 +2100 @@
 };
 
+static void recovery_tunables_done(struct tevent_req *subreq);
 static void recovery_nodemap_done(struct tevent_req *subreq);
 static void recovery_vnnmap_done(struct tevent_req *subreq);
 static void recovery_capabilities_done(struct tevent_req *subreq);
-static void recovery_tunables_done(struct tevent_req *subreq);
 static void recovery_dbmap_done(struct tevent_req *subreq);
 static void recovery_active_done(struct tevent_req *subreq);
@@ -1243 +2109 @@
 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);
@@ -1265 +2132 @@
         state->destnode = ctdb_client_pnn(client);
 
-        ctdb_req_control_get_nodemap(&request);
-        subreq = ctdb_client_control_send(mem_ctx, ev, client, state->destnode,
-                                          TIMEOUT(), &request);
+        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_nodemap_done, req);
+        tevent_req_set_callback(subreq, recovery_tunables_done, req);
 
         return req;
 }
 
-static void recovery_nodemap_done(struct tevent_req *subreq)
+static void recovery_tunables_done(struct tevent_req *subreq)
 {
         struct tevent_req *req = tevent_req_callback_data(
@@ -1284 +2152 @@
         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) {
+                LOG("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) {
+                LOG("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;
         bool status;
         int ret;
@@ -1307 +2216 @@
         if (state->count <= 0) {
                 tevent_req_error(req, ENOMEM);
+                return;
+        }
+
+        state->ban_credits = talloc_zero_array(state, uint32_t,
+                                               state->nodemap->num);
+        if (tevent_req_nomem(state->ban_credits, req)) {
                 return;
         }
@@ -1408 +2323 @@
                         return;
                 }
-        }
-
-        talloc_free(reply);
-
-        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_set_callback(subreq, recovery_tunables_done, 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) {
-                LOG("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) {
-                LOG("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
-                tevent_req_error(req, EPROTO);
-                return;
         }
 
@@ -1675 +2551 @@
                                   state->dbmap, state->tun_list,
                                   state->pnn_list, state->count,
+                                  state->caps, state->ban_credits,
                                   state->vnnmap->generation);
         if (tevent_req_nomem(subreq, req)) {
@@ -1695 +2572 @@
         TALLOC_FREE(subreq);
 
-        LOG("%d databases recovered\n", count);
-
-        if (! status) {
-                tevent_req_error(req, EIO);
+        LOG("%d of %d databases recovered\n", count, state->dbmap->num);
+
+        if (! status) {
+                uint32_t max_pnn = CTDB_UNKNOWN_PNN, max_credits = 0;
+                int i;
+
+                /* Bans are not enabled */
+                if (state->tun_list->enable_bans == 0) {
+                        tevent_req_error(req, EIO);
+                        return;
+                }
+
+                for (i=0; i<state->count; i++) {
+                        uint32_t pnn;
+                        pnn = state->pnn_list[i];
+                        if (state->ban_credits[pnn] > max_credits) {
+                                max_pnn = pnn;
+                                max_credits = state->ban_credits[pnn];
+                        }
+                }
+
+                /* If pulling database fails multiple times */
+                if (max_credits >= NUM_RETRIES) {
+                        struct ctdb_req_message message;
+
+                        LOG("Assigning banning credits to node %u\n", max_pnn);
+
+                        message.srvid = CTDB_SRVID_BANNING;
+                        message.data.pnn = max_pnn;
+
+                        subreq = ctdb_client_message_send(
+                                        state, state->ev, state->client,
+                                        ctdb_client_pnn(state->client),
+                                        &message);
+                        if (tevent_req_nomem(subreq, req)) {
+                                return;
+                        }
+                        tevent_req_set_callback(subreq, recovery_failed_done,
+                                                req);
+                } else {
+                        tevent_req_error(req, EIO);
+                }
                 return;
         }
@@ -1711 +2626 @@
         }
         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 = ctdb_client_message_recv(subreq, &ret);
+        TALLOC_FREE(subreq);
+        if (! status) {
+                LOG("failed to assign banning credits, ret=%d\n", ret);
+        }
+
+        tevent_req_error(req, EIO);
 }
 
@@ -1795 +2726 @@
 static void recovery_recv(struct tevent_req *req, int *perr)
 {
-        int err;
-
-        if (tevent_req_is_unix_error(req, &err)) {
-                if (perr != NULL) {
-                        *perr = err;
-                }
-                return;
-        }
+        generic_recv(req, perr);
 }
 

vendor/current/ctdb/server/ctdb_takeover.c

@@ -404 +404 @@
 }
 
-struct takeover_callback_state {
-        struct ctdb_req_control_old *c;
-        ctdb_sock_addr *addr;
-        struct ctdb_vnn *vnn;
-};
-
 struct ctdb_do_takeip_state {
         struct ctdb_req_control_old *c;
@@ -502 +496 @@
         CTDB_NO_MEMORY(ctdb, state);
 
-        state->c = talloc_steal(ctdb, c);
+        state->c = NULL;
         state->vnn   = vnn;
 
@@ -531 +525 @@
         }
 
+        state->c = talloc_steal(ctdb, c);
         return 0;
 }
@@ -639 +634 @@
         CTDB_NO_MEMORY(ctdb, state);
 
-        state->c = talloc_steal(ctdb, c);
+        state->c = NULL;
         state->old = old;
         state->vnn = vnn;
@@ -671 +666 @@
         }
 
+        state->c = talloc_steal(ctdb, c);
         return 0;
 }
@@ -816 +812 @@
 }
 
-/*
-  kill any clients that are registered with a IP that is being released
- */
-static void release_kill_clients(struct ctdb_context *ctdb, ctdb_sock_addr *addr)
-{
-        struct ctdb_client_ip *ip;
-
-        DEBUG(DEBUG_INFO,("release_kill_clients for ip %s\n",
-                ctdb_addr_to_str(addr)));
-
-        for (ip=ctdb->client_ip_list; ip; ip=ip->next) {
-                ctdb_sock_addr tmp_addr;
-
-                tmp_addr = ip->addr;
-                DEBUG(DEBUG_INFO,("checking for client %u with IP %s\n", 
-                        ip->client_id,
-                        ctdb_addr_to_str(&ip->addr)));
-
-                if (ctdb_same_ip(&tmp_addr, addr)) {
-                        struct ctdb_client *client = reqid_find(ctdb->idr,
-                                                                ip->client_id,
-                                                                struct ctdb_client);
-                        DEBUG(DEBUG_INFO,("matched client %u with IP %s and pid %u\n", 
-                                ip->client_id,
-                                ctdb_addr_to_str(&ip->addr),
-                                client->pid));
-
-                        if (client->pid != 0) {
-                                DEBUG(DEBUG_INFO,(__location__ " Killing client pid %u for IP %s on client_id %u\n",
-                                        (unsigned)client->pid,
-                                        ctdb_addr_to_str(addr),
-                                        ip->client_id));
-                                kill(client->pid, SIGKILL);
-                        }
-                }
-        }
-}
-
 static void do_delete_ip(struct ctdb_context *ctdb, struct ctdb_vnn *vnn)
 {
@@ -862 +820 @@
 }
 
+static struct ctdb_vnn *release_ip_post(struct ctdb_context *ctdb,
+                                        struct ctdb_vnn *vnn,
+                                        ctdb_sock_addr *addr)
+{
+        TDB_DATA data;
+
+        /* Send a message to all clients of this node telling them
+         * that the cluster has been reconfigured and they should
+         * close any connections on this IP address
+         */
+        data.dptr = (uint8_t *)ctdb_addr_to_str(addr);
+        data.dsize = strlen((char *)data.dptr)+1;
+        DEBUG(DEBUG_INFO, ("Sending RELEASE_IP message for %s\n", data.dptr));
+        ctdb_daemon_send_message(ctdb, ctdb->pnn, CTDB_SRVID_RELEASE_IP, data);
+
+        ctdb_vnn_unassign_iface(ctdb, vnn);
+
+        /* Process the IP if it has been marked for deletion */
+        if (vnn->delete_pending) {
+                do_delete_ip(ctdb, vnn);
+                return NULL;
+        }
+
+        return vnn;
+}
+
+struct release_ip_callback_state {
+        struct ctdb_req_control_old *c;
+        ctdb_sock_addr *addr;
+        struct ctdb_vnn *vnn;
+        uint32_t target_pnn;
+};
+
 /*
   called when releaseip event finishes
  */
-static void release_ip_callback(struct ctdb_context *ctdb, int status, 
+static void release_ip_callback(struct ctdb_context *ctdb, int status,
                                 void *private_data)
 {
-        struct takeover_callback_state *state = 
-                talloc_get_type(private_data, struct takeover_callback_state);
-        TDB_DATA data;
+        struct release_ip_callback_state *state =
+                talloc_get_type(private_data, struct release_ip_callback_state);
 
         if (status == -ETIME) {
@@ -888 +878 @@
         }
 
-        /* send a message to all clients of this node telling them
-           that the cluster has been reconfigured and they should
-           release any sockets on this IP */
-        data.dptr = (uint8_t *)talloc_strdup(state, ctdb_addr_to_str(state->addr));
-        CTDB_NO_MEMORY_VOID(ctdb, data.dptr);
-        data.dsize = strlen((char *)data.dptr)+1;
-
-        DEBUG(DEBUG_INFO,(__location__ " sending RELEASE_IP for '%s'\n", data.dptr));
-
-        ctdb_daemon_send_message(ctdb, ctdb->pnn, CTDB_SRVID_RELEASE_IP, data);
-
-        /* kill clients that have registered with this IP */
-        release_kill_clients(ctdb, state->addr);
-
-        ctdb_vnn_unassign_iface(ctdb, state->vnn);
-
-        /* Process the IP if it has been marked for deletion */
-        if (state->vnn->delete_pending) {
-                do_delete_ip(ctdb, state->vnn);
-                state->vnn = NULL;
-        }
+        state->vnn->pnn = state->target_pnn;
+        state->vnn = release_ip_post(ctdb, state->vnn, state->addr);
 
         /* the control succeeded */
@@ -915 +886 @@
 }
 
-static int ctdb_releaseip_destructor(struct takeover_callback_state *state)
+static int ctdb_releaseip_destructor(struct release_ip_callback_state *state)
 {
         if (state->vnn != NULL) {
@@ -932 +903 @@
 {
         int ret;
-        struct takeover_callback_state *state;
+        struct release_ip_callback_state *state;
         struct ctdb_public_ip *pip = (struct ctdb_public_ip *)indata.dptr;
         struct ctdb_vnn *vnn;
@@ -944 +915 @@
                 return 0;
         }
-        vnn->pnn = pip->pnn;
 
         /* stop any previous arps */
@@ -950 +920 @@
         vnn->takeover_ctx = NULL;
 
-        /* Some ctdb tool commands (e.g. moveip, rebalanceip) send
-         * lazy multicast to drop an IP from any node that isn't the
-         * intended new node.  The following causes makes ctdbd ignore
-         * a release for any address it doesn't host.
+        /* RELEASE_IP controls are sent to all nodes that should not
+         * be hosting a particular IP.  This serves 2 purposes.  The
+         * first is to help resolve any inconsistencies.  If a node
+         * does unexpectly host an IP then it will be released.  The
+         * 2nd is to use a "redundant release" to tell non-takeover
+         * nodes where an IP is moving to.  This is how "ctdb ip" can
+         * report the (likely) location of an IP by only asking the
+         * local node.  Redundant releases need to update the PNN but
+         * are otherwise ignored.
          */
         if (ctdb->tunable.disable_ip_failover == 0 && ctdb->do_checkpublicip) {
@@ -961 +936 @@
                                 vnn->public_netmask_bits,
                                 ctdb_vnn_iface_string(vnn)));
+                        vnn->pnn = pip->pnn;
                         ctdb_vnn_unassign_iface(ctdb, vnn);
                         return 0;
@@ -969 +945 @@
                                            ctdb_addr_to_str(&pip->addr),
                                            vnn->public_netmask_bits));
+                        vnn->pnn = pip->pnn;
                         return 0;
                 }
@@ -994 +971 @@
                 pip->pnn));
 
-        state = talloc(ctdb, struct takeover_callback_state);
+        state = talloc(ctdb, struct release_ip_callback_state);
         if (state == NULL) {
                 ctdb_set_error(ctdb, "Out of memory at %s:%d",
@@ -1002 +979 @@
         }
 
-        state->c = talloc_steal(state, c);
-        state->addr = talloc(state, ctdb_sock_addr);       
+        state->c = NULL;
+        state->addr = talloc(state, ctdb_sock_addr);
         if (state->addr == NULL) {
                 ctdb_set_error(ctdb, "Out of memory at %s:%d",
@@ -1012 +989 @@
         }
         *state->addr = pip->addr;
+        state->target_pnn = pip->pnn;
         state->vnn   = vnn;
 
@@ -1035 +1013 @@
         /* tell the control that we will be reply asynchronously */
         *async_reply = true;
+        state->c = talloc_steal(state, c);
         return 0;
 }
@@ -1776 +1755 @@
         bool can_host_ips;
 
+        /* Default timeout for early jump to IPREALLOCATED.  See below
+         * for explanation of 3 times... */
+        timeout = timeval_current_ofs(3 * ctdb->tunable.takeover_timeout, 0);
+
         /*
          * ip failover is completely disabled, just send out the 
@@ -1852 +1835 @@
 
         ZERO_STRUCT(ip); /* Avoid valgrind warnings for union */
+
+        /* Each of the following stages (RELEASE_IP, TAKEOVER_IP,
+         * IPREALLOCATED) notionally has a timeout of TakeoverTimeout
+         * seconds.  However, RELEASE_IP can take longer due to TCP
+         * connection killing, so sometimes needs more time.
+         * Therefore, use a cumulative timeout of TakeoverTimeout * 3
+         * seconds across all 3 stages.  No explicit expiry checks are
+         * needed before each stage because tevent is smart enough to
+         * fire the timeouts even if they are in the past.  Initialise
+         * this here so it explicitly covers the stages we're
+         * interested in but, in particular, not the time taken by the
+         * ipalloc().
+         */
+        timeout = timeval_current_ofs(3 * ctdb->tunable.takeover_timeout, 0);
 
         /* Send a RELEASE_IP to all nodes that should not be hosting
@@ -1875 +1872 @@
                         ip.addr = tmp_ip->addr;
 
-                        timeout = TAKEOVER_TIMEOUT();
                         data.dsize = sizeof(ip);
                         data.dptr  = (uint8_t *)&ip;
@@ -1918 +1914 @@
                 ip.addr = tmp_ip->addr;
 
-                timeout = TAKEOVER_TIMEOUT();
                 data.dsize = sizeof(ip);
                 data.dptr  = (uint8_t *)&ip;
@@ -1956 +1951 @@
         nodes = list_of_connected_nodes(ctdb, nodemap, tmp_ctx, true);
         ret = ctdb_client_async_control(ctdb, CTDB_CONTROL_IPREALLOCATED,
-                                        nodes, 0, TAKEOVER_TIMEOUT(),
+                                        nodes, 0, timeout,
                                         false, tdb_null,
                                         NULL, iprealloc_fail_callback,
@@ -2377 +2372 @@
 void ctdb_release_all_ips(struct ctdb_context *ctdb)
 {
-        struct ctdb_vnn *vnn;
+        struct ctdb_vnn *vnn, *next;
         int count = 0;
 
@@ -2384 +2379 @@
         }
 
-        for (vnn=ctdb->vnn;vnn;vnn=vnn->next) {
+        for (vnn = ctdb->vnn; vnn != NULL; vnn = next) {
+                /* vnn can be freed below in release_ip_post() */
+                next = vnn->next;
+
                 if (!ctdb_sys_have_ip(&vnn->public_address)) {
                         ctdb_vnn_unassign_iface(ctdb, vnn);
-                        continue;
-                }
-                if (!vnn->iface) {
                         continue;
                 }
@@ -2413 +2408 @@
 
                 ctdb_event_script_args(ctdb, CTDB_EVENT_RELEASE_IP, "%s %s %u",
-                                  ctdb_vnn_iface_string(vnn),
-                                  ctdb_addr_to_str(&vnn->public_address),
-                                  vnn->public_netmask_bits);
-                release_kill_clients(ctdb, &vnn->public_address);
-                ctdb_vnn_unassign_iface(ctdb, vnn);
-                vnn->update_in_flight = false;
+                                       ctdb_vnn_iface_string(vnn),
+                                       ctdb_addr_to_str(&vnn->public_address),
+                                       vnn->public_netmask_bits);
+                /* releaseip timeouts are converted to success, so to
+                 * detect failures just check if the IP address is
+                 * still there...
+                 */
+                if (ctdb_sys_have_ip(&vnn->public_address)) {
+                        DEBUG(DEBUG_ERR,
+                              (__location__
+                               " IP address %s not released\n",
+                               ctdb_addr_to_str(&vnn->public_address)));
+                        vnn->update_in_flight = false;
+                        continue;
+                }
+
+                vnn = release_ip_post(ctdb, vnn, &vnn->public_address);
+                if (vnn != NULL) {
+                        vnn->update_in_flight = false;
+                }
                 count++;
         }

vendor/current/ctdb/server/ctdb_tunables.c

@@ -42 +42 @@
         { "KeepaliveInterval",    5,  offsetof(struct ctdb_tunable_list, keepalive_interval), false },
         { "KeepaliveLimit",       5,  offsetof(struct ctdb_tunable_list, keepalive_limit), false },
-        { "RecoverTimeout",     120,  offsetof(struct ctdb_tunable_list, recover_timeout), false },
+        { "RecoverTimeout",      30,  offsetof(struct ctdb_tunable_list, recover_timeout), false },
         { "RecoverInterval",      1,  offsetof(struct ctdb_tunable_list, recover_interval), false },
         { "ElectionTimeout",      3,  offsetof(struct ctdb_tunable_list, election_timeout), false },
@@ -94 +94 @@
         { "TDBMutexEnabled", 0, offsetof(struct ctdb_tunable_list, mutex_enabled), false },
         { "LockProcessesPerDB", 200, offsetof(struct ctdb_tunable_list, lock_processes_per_db), false },
+        { "RecBufferSizeLimit", 1000000, offsetof(struct ctdb_tunable_list, rec_buffer_size_limit), false },
 };
 

vendor/current/ctdb/server/ctdbd.c

@@ -179 +179 @@
 
         ev = tevent_context_init(NULL);
+        if (ev == NULL) {
+                DEBUG(DEBUG_ALERT,("tevent_context_init() failed\n"));
+                exit(1);
+        }
         tevent_loop_allow_nesting(ev);
 

vendor/current/ctdb/server/eventscript.c

@@ -697 +697 @@
 {
         DLIST_REMOVE(callback->ctdb->script_callbacks, callback);
+        return 0;
+}
+
+struct schedule_callback_state {
+        struct ctdb_context *ctdb;
+        void (*callback)(struct ctdb_context *, int, void *);
+        void *private_data;
+        int status;
+        struct tevent_immediate *im;
+};
+
+static void schedule_callback_handler(struct tevent_context *ctx,
+                                      struct tevent_immediate *im,
+                                      void *private_data)
+{
+        struct schedule_callback_state *state =
+                talloc_get_type_abort(private_data,
+                                      struct schedule_callback_state);
+
+        if (state->callback != NULL) {
+                state->callback(state->ctdb, state->status,
+                                state->private_data);
+        }
+        talloc_free(state);
+}
+
+static int
+schedule_callback_immediate(struct ctdb_context *ctdb,
+                            void (*callback)(struct ctdb_context *,
+                                             int, void *),
+                            void *private_data,
+                            int status)
+{
+        struct schedule_callback_state *state;
+        struct tevent_immediate *im;
+
+        state = talloc_zero(ctdb, struct schedule_callback_state);
+        if (state == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " out of memory\n"));
+                return -1;
+        }
+        im = tevent_create_immediate(state);
+        if (im == NULL) {
+                DEBUG(DEBUG_ERR, (__location__ " out of memory\n"));
+                talloc_free(state);
+                return -1;
+        }
+
+        state->ctdb = ctdb;
+        state->callback = callback;
+        state->private_data = private_data;
+        state->status = status;
+        state->im = im;
+
+        tevent_schedule_immediate(im, ctdb->ev,
+                                  schedule_callback_handler, state);
         return 0;
 }
@@ -808 +864 @@
         state->child = 0;
 
+        /* Nothing to do? */
+        if (state->scripts->num_scripts == 0) {
+                int ret = schedule_callback_immediate(ctdb, callback,
+                                                      private_data, 0);
+                talloc_free(state);
+                if (ret != 0) {
+                        DEBUG(DEBUG_ERR,
+                              ("Unable to schedule callback for 0 scripts\n"));
+                        return 1;
+                }
+                return 0;
+        }
+
+        state->scripts->scripts[0].status = fork_child_for_script(ctdb, state);
+        if (state->scripts->scripts[0].status != 0) {
+                talloc_free(state);
+                return -1;
+        }
+
         if (call == CTDB_EVENT_MONITOR) {
                 ctdb->current_monitor = state;
         }
 
+        ctdb->active_events++;
+
         talloc_set_destructor(state, event_script_destructor);
-
-        ctdb->active_events++;
-
-        /* Nothing to do? */
-        if (state->scripts->num_scripts == 0) {
-                callback(ctdb, 0, private_data);
-                talloc_free(state);
-                return 0;
-        }
-
-        state->scripts->scripts[0].status = fork_child_for_script(ctdb, state);
-        if (state->scripts->scripts[0].status != 0) {
-                /* Callback is called from destructor, with fail result. */
-                talloc_free(state);
-                return 0;
-        }
 
         if (!timeval_is_zero(&state->timeout)) {
@@ -1008 +1069 @@
         CTDB_NO_MEMORY(ctdb, state);
 
-        state->c = talloc_steal(state, c);
+        state->c = NULL;
 
         DEBUG(DEBUG_NOTICE,("Running eventscripts with arguments %s\n", indata.dptr));
@@ -1024 +1085 @@
         /* tell ctdb_control.c that we will be replying asynchronously */
         *async_reply = true;
-
+        state->c = talloc_steal(state, c);
         return 0;
 }