ctdb/server/ctdb_recovery_helper.c

   1 /*
   2    ctdb parallel database recovery
   3
   4    Copyright (C) Amitay Isaacs  2015
   5
   6    This program is free software; you can redistribute it and/or modify
   7    it under the terms of the GNU General Public License as published by
   8    the Free Software Foundation; either version 3 of the License, or
   9    (at your option) any later version.
  10
  11    This program is distributed in the hope that it will be useful,
  12    but WITHOUT ANY WARRANTY; without even the implied warranty of
  13    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  14    GNU General Public License for more details.
  15
  16    You should have received a copy of the GNU General Public License
  17    along with this program; if not, see <http://www.gnu.org/licenses/>.
  18 */
  19
  20 #include "replace.h"
  21 #include "system/network.h"
  22 #include "system/filesys.h"
  23
  24 #include <talloc.h>
  25 #include <tevent.h>
  26 #include <tdb.h>
  27 #include <libgen.h>
  28
  29 #include "lib/tdb_wrap/tdb_wrap.h"
  30 #include "lib/util/sys_rw.h"
  31 #include "lib/util/time.h"
  32 #include "lib/util/tevent_unix.h"
  33
  34 #include "protocol/protocol.h"
  35 #include "protocol/protocol_api.h"
  36 #include "client/client.h"
  37
  38 #include "common/logging.h"
  39
  40 static int recover_timeout = 30;
  41
  42 #define NUM_RETRIES     3
  43
  44 #define TIMEOUT()       timeval_current_ofs(recover_timeout, 0)
  45
  46 /*
  47  * Utility functions
  48  */
  49
  50 static bool generic_recv(struct tevent_req *req, int *perr)
  51 {
  52         int err;
  53
  54         if (tevent_req_is_unix_error(req, &err)) {
  55                 if (perr != NULL) {
  56                         *perr = err;
  57                 }
  58                 return false;
  59         }
  60
  61         return true;
  62 }
  63
  64 static uint64_t rec_srvid = CTDB_SRVID_RECOVERY;
  65
  66 static uint64_t srvid_next(void)
  67 {
  68         rec_srvid += 1;
  69         return rec_srvid;
  70 }
  71
  72 /*
  73  * Recovery database functions
  74  */
  75
  76 struct recdb_context {
  77         uint32_t db_id;
  78         const char *db_name;
  79         const char *db_path;
  80         struct tdb_wrap *db;
  81         bool persistent;
  82 };
  83
  84 static struct recdb_context *recdb_create(TALLOC_CTX *mem_ctx, uint32_t db_id,
  85                                           const char *db_name,
  86                                           const char *db_path,
  87                                           uint32_t hash_size, bool persistent)
  88 {
  89         static char *db_dir_state = NULL;
  90         struct recdb_context *recdb;
  91         unsigned int tdb_flags;
  92
  93         recdb = talloc(mem_ctx, struct recdb_context);
  94         if (recdb == NULL) {
  95                 return NULL;
  96         }
  97
  98         if (db_dir_state == NULL) {
  99                 db_dir_state = getenv("CTDB_DBDIR_STATE");
 100         }
 101
 102         recdb->db_name = db_name;
 103         recdb->db_id = db_id;
 104         recdb->db_path = talloc_asprintf(recdb, "%s/recdb.%s",
 105                                          db_dir_state != NULL ?
 106                                             db_dir_state :
 107                                             dirname(discard_const(db_path)),
 108                                          db_name);
 109         if (recdb->db_path == NULL) {
 110                 talloc_free(recdb);
 111                 return NULL;
 112         }
 113         unlink(recdb->db_path);
 114
 115         tdb_flags = TDB_NOLOCK | TDB_INCOMPATIBLE_HASH | TDB_DISALLOW_NESTING;
 116         recdb->db = tdb_wrap_open(mem_ctx, recdb->db_path, hash_size,
 117                                   tdb_flags, O_RDWR|O_CREAT|O_EXCL, 0600);
 118         if (recdb->db == NULL) {
 119                 talloc_free(recdb);
 120                 D_ERR("failed to create recovery db %s\n", recdb->db_path);
 121                 return NULL;
 122         }
 123
 124         recdb->persistent = persistent;
 125
 126         return recdb;
 127 }
 128
 129 static uint32_t recdb_id(struct recdb_context *recdb)
 130 {
 131         return recdb->db_id;
 132 }
 133
 134 static const char *recdb_name(struct recdb_context *recdb)
 135 {
 136         return recdb->db_name;
 137 }
 138
 139 static const char *recdb_path(struct recdb_context *recdb)
 140 {
 141         return recdb->db_path;
 142 }
 143
 144 static struct tdb_context *recdb_tdb(struct recdb_context *recdb)
 145 {
 146         return recdb->db->tdb;
 147 }
 148
 149 static bool recdb_persistent(struct recdb_context *recdb)
 150 {
 151         return recdb->persistent;
 152 }
 153
 154 struct recdb_add_traverse_state {
 155         struct recdb_context *recdb;
 156         int mypnn;
 157 };
 158
 159 static int recdb_add_traverse(uint32_t reqid, struct ctdb_ltdb_header *header,
 160                               TDB_DATA key, TDB_DATA data,
 161                               void *private_data)
 162 {
 163         struct recdb_add_traverse_state *state =
 164                 (struct recdb_add_traverse_state *)private_data;
 165         struct ctdb_ltdb_header *hdr;
 166         TDB_DATA prev_data;
 167         int ret;
 168
 169         /* header is not marshalled separately in the pulldb control */
 170         if (data.dsize < sizeof(struct ctdb_ltdb_header)) {
 171                 return -1;
 172         }
 173
 174         hdr = (struct ctdb_ltdb_header *)data.dptr;
 175
 176         /* fetch the existing record, if any */
 177         prev_data = tdb_fetch(recdb_tdb(state->recdb), key);
 178
 179         if (prev_data.dptr != NULL) {
 180                 struct ctdb_ltdb_header prev_hdr;
 181
 182                 prev_hdr = *(struct ctdb_ltdb_header *)prev_data.dptr;
 183                 free(prev_data.dptr);
 184                 if (hdr->rsn < prev_hdr.rsn ||
 185                     (hdr->rsn == prev_hdr.rsn &&
 186                      prev_hdr.dmaster != state->mypnn)) {
 187                         return 0;
 188                 }
 189         }
 190
 191         ret = tdb_store(recdb_tdb(state->recdb), key, data, TDB_REPLACE);
 192         if (ret != 0) {
 193                 return -1;
 194         }
 195         return 0;
 196 }
 197
 198 static bool recdb_add(struct recdb_context *recdb, int mypnn,
 199                       struct ctdb_rec_buffer *recbuf)
 200 {
 201         struct recdb_add_traverse_state state;
 202         int ret;
 203
 204         state.recdb = recdb;
 205         state.mypnn = mypnn;
 206
 207         ret = ctdb_rec_buffer_traverse(recbuf, recdb_add_traverse, &state);
 208         if (ret != 0) {
 209                 return false;
 210         }
 211
 212         return true;
 213 }
 214
 215 /* This function decides which records from recdb are retained */
 216 static int recbuf_filter_add(struct ctdb_rec_buffer *recbuf, bool persistent,
 217                              uint32_t reqid, uint32_t dmaster,
 218                              TDB_DATA key, TDB_DATA data)
 219 {
 220         struct ctdb_ltdb_header *header;
 221         int ret;
 222
 223         /* Skip empty records */
 224         if (data.dsize <= sizeof(struct ctdb_ltdb_header)) {
 225                 return 0;
 226         }
 227
 228         /* update the dmaster field to point to us */
 229         header = (struct ctdb_ltdb_header *)data.dptr;
 230         if (!persistent) {
 231                 header->dmaster = dmaster;
 232                 header->flags |= CTDB_REC_FLAG_MIGRATED_WITH_DATA;
 233         }
 234
 235         ret = ctdb_rec_buffer_add(recbuf, recbuf, reqid, NULL, key, data);
 236         if (ret != 0) {
 237                 return ret;
 238         }
 239
 240         return 0;
 241 }
 242
 243 struct recdb_records_traverse_state {
 244         struct ctdb_rec_buffer *recbuf;
 245         uint32_t dmaster;
 246         uint32_t reqid;
 247         bool persistent;
 248         bool failed;
 249 };
 250
 251 static int recdb_records_traverse(struct tdb_context *tdb,
 252                                   TDB_DATA key, TDB_DATA data,
 253                                   void *private_data)
 254 {
 255         struct recdb_records_traverse_state *state =
 256                 (struct recdb_records_traverse_state *)private_data;
 257         int ret;
 258
 259         ret = recbuf_filter_add(state->recbuf, state->persistent,
 260                                 state->reqid, state->dmaster, key, data);
 261         if (ret != 0) {
 262                 state->failed = true;
 263                 return ret;
 264         }
 265
 266         return 0;
 267 }
 268
 269 static struct ctdb_rec_buffer *recdb_records(struct recdb_context *recdb,
 270                                              TALLOC_CTX *mem_ctx,
 271                                              uint32_t dmaster)
 272 {
 273         struct recdb_records_traverse_state state;
 274         int ret;
 275
 276         state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
 277         if (state.recbuf == NULL) {
 278                 return NULL;
 279         }
 280         state.dmaster = dmaster;
 281         state.reqid = 0;
 282         state.persistent = recdb_persistent(recdb);
 283         state.failed = false;
 284
 285         ret = tdb_traverse_read(recdb_tdb(recdb), recdb_records_traverse,
 286                                 &state);
 287         if (ret == -1 || state.failed) {
 288                 D_ERR("Failed to marshall recovery records for %s\n",
 289                       recdb_name(recdb));
 290                 TALLOC_FREE(state.recbuf);
 291                 return NULL;
 292         }
 293
 294         return state.recbuf;
 295 }
 296
 297 struct recdb_file_traverse_state {
 298         struct ctdb_rec_buffer *recbuf;
 299         struct recdb_context *recdb;
 300         TALLOC_CTX *mem_ctx;
 301         uint32_t dmaster;
 302         uint32_t reqid;
 303         bool persistent;
 304         bool failed;
 305         int fd;
 306         int max_size;
 307         int num_buffers;
 308 };
 309
 310 static int recdb_file_traverse(struct tdb_context *tdb,
 311                                TDB_DATA key, TDB_DATA data,
 312                                void *private_data)
 313 {
 314         struct recdb_file_traverse_state *state =
 315                 (struct recdb_file_traverse_state *)private_data;
 316         int ret;
 317
 318         ret = recbuf_filter_add(state->recbuf, state->persistent,
 319                                 state->reqid, state->dmaster, key, data);
 320         if (ret != 0) {
 321                 state->failed = true;
 322                 return ret;
 323         }
 324
 325         if (ctdb_rec_buffer_len(state->recbuf) > state->max_size) {
 326                 ret = ctdb_rec_buffer_write(state->recbuf, state->fd);
 327                 if (ret != 0) {
 328                         D_ERR("Failed to collect recovery records for %s\n",
 329                               recdb_name(state->recdb));
 330                         state->failed = true;
 331                         return ret;
 332                 }
 333
 334                 state->num_buffers += 1;
 335
 336                 TALLOC_FREE(state->recbuf);
 337                 state->recbuf = ctdb_rec_buffer_init(state->mem_ctx,
 338                                                      recdb_id(state->recdb));
 339                 if (state->recbuf == NULL) {
 340                         state->failed = true;
 341                         return ENOMEM;
 342                 }
 343         }
 344
 345         return 0;
 346 }
 347
 348 static int recdb_file(struct recdb_context *recdb, TALLOC_CTX *mem_ctx,
 349                       uint32_t dmaster, int fd, int max_size)
 350 {
 351         struct recdb_file_traverse_state state;
 352         int ret;
 353
 354         state.recbuf = ctdb_rec_buffer_init(mem_ctx, recdb_id(recdb));
 355         if (state.recbuf == NULL) {
 356                 return -1;
 357         }
 358         state.recdb = recdb;
 359         state.mem_ctx = mem_ctx;
 360         state.dmaster = dmaster;
 361         state.reqid = 0;
 362         state.persistent = recdb_persistent(recdb);
 363         state.failed = false;
 364         state.fd = fd;
 365         state.max_size = max_size;
 366         state.num_buffers = 0;
 367
 368         ret = tdb_traverse_read(recdb_tdb(recdb), recdb_file_traverse, &state);
 369         if (ret == -1 || state.failed) {
 370                 TALLOC_FREE(state.recbuf);
 371                 return -1;
 372         }
 373
 374         ret = ctdb_rec_buffer_write(state.recbuf, fd);
 375         if (ret != 0) {
 376                 D_ERR("Failed to collect recovery records for %s\n",
 377                       recdb_name(recdb));
 378                 TALLOC_FREE(state.recbuf);
 379                 return -1;
 380         }
 381         state.num_buffers += 1;
 382
 383         D_DEBUG("Wrote %d buffers of recovery records for %s\n",
 384                 state.num_buffers, recdb_name(recdb));
 385
 386         return state.num_buffers;
 387 }
 388
 389 /*
 390  * Pull database from a single node
 391  */
 392
 393 struct pull_database_state {
 394         struct tevent_context *ev;
 395         struct ctdb_client_context *client;
 396         struct recdb_context *recdb;
 397         uint32_t pnn;
 398         uint64_t srvid;
 399         int num_records;
 400 };
 401
 402 static void pull_database_handler(uint64_t srvid, TDB_DATA data,
 403                                   void *private_data);
 404 static void pull_database_register_done(struct tevent_req *subreq);
 405 static void pull_database_old_done(struct tevent_req *subreq);
 406 static void pull_database_unregister_done(struct tevent_req *subreq);
 407 static void pull_database_new_done(struct tevent_req *subreq);
 408
 409 static struct tevent_req *pull_database_send(
 410                         TALLOC_CTX *mem_ctx,
 411                         struct tevent_context *ev,
 412                         struct ctdb_client_context *client,
 413                         uint32_t pnn, uint32_t caps,
 414                         struct recdb_context *recdb)
 415 {
 416         struct tevent_req *req, *subreq;
 417         struct pull_database_state *state;
 418         struct ctdb_req_control request;
 419
 420         req = tevent_req_create(mem_ctx, &state, struct pull_database_state);
 421         if (req == NULL) {
 422                 return NULL;
 423         }
 424
 425         state->ev = ev;
 426         state->client = client;
 427         state->recdb = recdb;
 428         state->pnn = pnn;
 429         state->srvid = srvid_next();
 430
 431         if (caps & CTDB_CAP_FRAGMENTED_CONTROLS) {
 432                 subreq = ctdb_client_set_message_handler_send(
 433                                         state, state->ev, state->client,
 434                                         state->srvid, pull_database_handler,
 435                                         req);
 436                 if (tevent_req_nomem(subreq, req)) {
 437                         return tevent_req_post(req, ev);
 438                 }
 439
 440                 tevent_req_set_callback(subreq, pull_database_register_done,
 441                                         req);
 442
 443         } else {
 444                 struct ctdb_pulldb pulldb;
 445
 446                 pulldb.db_id = recdb_id(recdb);
 447                 pulldb.lmaster = CTDB_LMASTER_ANY;
 448
 449                 ctdb_req_control_pull_db(&request, &pulldb);
 450                 subreq = ctdb_client_control_send(state, state->ev,
 451                                                   state->client,
 452                                                   pnn, TIMEOUT(),
 453                                                   &request);
 454                 if (tevent_req_nomem(subreq, req)) {
 455                         return tevent_req_post(req, ev);
 456                 }
 457                 tevent_req_set_callback(subreq, pull_database_old_done, req);
 458         }
 459
 460         return req;
 461 }
 462
 463 static void pull_database_handler(uint64_t srvid, TDB_DATA data,
 464                                   void *private_data)
 465 {
 466         struct tevent_req *req = talloc_get_type_abort(
 467                 private_data, struct tevent_req);
 468         struct pull_database_state *state = tevent_req_data(
 469                 req, struct pull_database_state);
 470         struct ctdb_rec_buffer *recbuf;
 471         size_t np;
 472         int ret;
 473         bool status;
 474
 475         if (srvid != state->srvid) {
 476                 return;
 477         }
 478
 479         ret = ctdb_rec_buffer_pull(data.dptr, data.dsize, state, &recbuf, &np);
 480         if (ret != 0) {
 481                 D_ERR("Invalid data received for DB_PULL messages\n");
 482                 return;
 483         }
 484
 485         if (recbuf->db_id != recdb_id(state->recdb)) {
 486                 talloc_free(recbuf);
 487                 D_ERR("Invalid dbid:%08x for DB_PULL messages for %s\n",
 488                       recbuf->db_id, recdb_name(state->recdb));
 489                 return;
 490         }
 491
 492         status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
 493                            recbuf);
 494         if (! status) {
 495                 talloc_free(recbuf);
 496                 D_ERR("Failed to add records to recdb for %s\n",
 497                       recdb_name(state->recdb));
 498                 return;
 499         }
 500
 501         state->num_records += recbuf->count;
 502         talloc_free(recbuf);
 503 }
 504
 505 static void pull_database_register_done(struct tevent_req *subreq)
 506 {
 507         struct tevent_req *req = tevent_req_callback_data(
 508                 subreq, struct tevent_req);
 509         struct pull_database_state *state = tevent_req_data(
 510                 req, struct pull_database_state);
 511         struct ctdb_req_control request;
 512         struct ctdb_pulldb_ext pulldb_ext;
 513         int ret;
 514         bool status;
 515
 516         status = ctdb_client_set_message_handler_recv(subreq, &ret);
 517         TALLOC_FREE(subreq);
 518         if (! status) {
 519                 D_ERR("Failed to set message handler for DB_PULL for %s\n",
 520                       recdb_name(state->recdb));
 521                 tevent_req_error(req, ret);
 522                 return;
 523         }
 524
 525         pulldb_ext.db_id = recdb_id(state->recdb);
 526         pulldb_ext.lmaster = CTDB_LMASTER_ANY;
 527         pulldb_ext.srvid = state->srvid;
 528
 529         ctdb_req_control_db_pull(&request, &pulldb_ext);
 530         subreq = ctdb_client_control_send(state, state->ev, state->client,
 531                                           state->pnn, TIMEOUT(), &request);
 532         if (tevent_req_nomem(subreq, req)) {
 533                 return;
 534         }
 535         tevent_req_set_callback(subreq, pull_database_new_done, req);
 536 }
 537
 538 static void pull_database_old_done(struct tevent_req *subreq)
 539 {
 540         struct tevent_req *req = tevent_req_callback_data(
 541                 subreq, struct tevent_req);
 542         struct pull_database_state *state = tevent_req_data(
 543                 req, struct pull_database_state);
 544         struct ctdb_reply_control *reply;
 545         struct ctdb_rec_buffer *recbuf;
 546         int ret;
 547         bool status;
 548
 549         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 550         TALLOC_FREE(subreq);
 551         if (! status) {
 552                 D_ERR("control PULL_DB failed for %s on node %u, ret=%d\n",
 553                       recdb_name(state->recdb), state->pnn, ret);
 554                 tevent_req_error(req, ret);
 555                 return;
 556         }
 557
 558         ret = ctdb_reply_control_pull_db(reply, state, &recbuf);
 559         talloc_free(reply);
 560         if (ret != 0) {
 561                 tevent_req_error(req, ret);
 562                 return;
 563         }
 564
 565         status = recdb_add(state->recdb, ctdb_client_pnn(state->client),
 566                            recbuf);
 567         if (! status) {
 568                 talloc_free(recbuf);
 569                 tevent_req_error(req, EIO);
 570                 return;
 571         }
 572
 573         state->num_records = recbuf->count;
 574         talloc_free(recbuf);
 575
 576         D_INFO("Pulled %d records for db %s from node %d\n",
 577                state->num_records, recdb_name(state->recdb), state->pnn);
 578
 579         tevent_req_done(req);
 580 }
 581
 582 static void pull_database_new_done(struct tevent_req *subreq)
 583 {
 584         struct tevent_req *req = tevent_req_callback_data(
 585                 subreq, struct tevent_req);
 586         struct pull_database_state *state = tevent_req_data(
 587                 req, struct pull_database_state);
 588         struct ctdb_reply_control *reply;
 589         uint32_t num_records;
 590         int ret;
 591         bool status;
 592
 593         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
 594         TALLOC_FREE(subreq);
 595         if (! status) {
 596                 D_ERR("control DB_PULL failed for %s on node %u, ret=%d\n",
 597                       recdb_name(state->recdb), state->pnn, ret);
 598                 tevent_req_error(req, ret);
 599                 return;
 600         }
 601
 602         ret = ctdb_reply_control_db_pull(reply, &num_records);
 603         talloc_free(reply);
 604         if (num_records != state->num_records) {
 605                 D_ERR("mismatch (%u != %u) in DB_PULL records for db %s\n",
 606                       num_records, state->num_records,
 607                       recdb_name(state->recdb));
 608                 tevent_req_error(req, EIO);
 609                 return;
 610         }
 611
 612         D_INFO("Pulled %d records for db %s from node %d\n",
 613                state->num_records, recdb_name(state->recdb), state->pnn);
 614
 615         subreq = ctdb_client_remove_message_handler_send(
 616                                         state, state->ev, state->client,
 617                                         state->srvid, req);
 618         if (tevent_req_nomem(subreq, req)) {
 619                 return;
 620         }
 621         tevent_req_set_callback(subreq, pull_database_unregister_done, req);
 622 }
 623
 624 static void pull_database_unregister_done(struct tevent_req *subreq)
 625 {
 626         struct tevent_req *req = tevent_req_callback_data(
 627                 subreq, struct tevent_req);
 628         struct pull_database_state *state = tevent_req_data(
 629                 req, struct pull_database_state);
 630         int ret;
 631         bool status;
 632
 633         status = ctdb_client_remove_message_handler_recv(subreq, &ret);
 634         TALLOC_FREE(subreq);
 635         if (! status) {
 636                 D_ERR("failed to remove message handler for DB_PULL for db %s\n",
 637                       recdb_name(state->recdb));
 638                 tevent_req_error(req, ret);
 639                 return;
 640         }
 641
 642         tevent_req_done(req);
 643 }
 644
 645 static bool pull_database_recv(struct tevent_req *req, int *perr)
 646 {
 647         return generic_recv(req, perr);
 648 }
 649
 650 /*
 651  * Push database to specified nodes (old style)
 652  */
 653
 654 struct push_database_old_state {
 655         struct tevent_context *ev;
 656         struct ctdb_client_context *client;
 657         struct recdb_context *recdb;
 658         uint32_t *pnn_list;
 659         int count;
 660         struct ctdb_rec_buffer *recbuf;
 661         int index;
 662 };
 663
 664 static void push_database_old_push_done(struct tevent_req *subreq);
 665
 666 static struct tevent_req *push_database_old_send(
 667                         TALLOC_CTX *mem_ctx,
 668                         struct tevent_context *ev,
 669                         struct ctdb_client_context *client,
 670                         uint32_t *pnn_list, int count,
 671                         struct recdb_context *recdb)
 672 {
 673         struct tevent_req *req, *subreq;
 674         struct push_database_old_state *state;
 675         struct ctdb_req_control request;
 676         uint32_t pnn;
 677
 678         req = tevent_req_create(mem_ctx, &state,
 679                                 struct push_database_old_state);
 680         if (req == NULL) {
 681                 return NULL;
 682         }
 683
 684         state->ev = ev;
 685         state->client = client;
 686         state->recdb = recdb;
 687         state->pnn_list = pnn_list;
 688         state->count = count;
 689         state->index = 0;
 690
 691         state->recbuf = recdb_records(recdb, state,
 692                                       ctdb_client_pnn(client));
 693         if (tevent_req_nomem(state->recbuf, req)) {
 694                 return tevent_req_post(req, ev);
 695         }
 696
 697         pnn = state->pnn_list[state->index];
 698
 699         ctdb_req_control_push_db(&request, state->recbuf);
 700         subreq = ctdb_client_control_send(state, ev, client, pnn,
 701                                           TIMEOUT(), &request);
 702         if (tevent_req_nomem(subreq, req)) {
 703                 return tevent_req_post(req, ev);
 704         }
 705         tevent_req_set_callback(subreq, push_database_old_push_done, req);
 706
 707         return req;
 708 }
 709
 710 static void push_database_old_push_done(struct tevent_req *subreq)
 711 {
 712         struct tevent_req *req = tevent_req_callback_data(
 713                 subreq, struct tevent_req);
 714         struct push_database_old_state *state = tevent_req_data(
 715                 req, struct push_database_old_state);
 716         struct ctdb_req_control request;
 717         uint32_t pnn;
 718         int ret;
 719         bool status;
 720
 721         status = ctdb_client_control_recv(subreq, &ret, NULL, NULL);
 722         TALLOC_FREE(subreq);
 723         if (! status) {
 724                 D_ERR("control PUSH_DB failed for db %s on node %u, ret=%d\n",
 725                       recdb_name(state->recdb), state->pnn_list[state->index],
 726                       ret);
 727                 tevent_req_error(req, ret);
 728                 return;
 729         }
 730
 731         state->index += 1;
 732         if (state->index == state->count) {
 733                 TALLOC_FREE(state->recbuf);
 734                 tevent_req_done(req);
 735                 return;
 736         }
 737
 738         pnn = state->pnn_list[state->index];
 739
 740         ctdb_req_control_push_db(&request, state->recbuf);
 741         subreq = ctdb_client_control_send(state, state->ev, state->client,
 742                                           pnn, TIMEOUT(), &request);
 743         if (tevent_req_nomem(subreq, req)) {
 744                 return;
 745         }
 746         tevent_req_set_callback(subreq, push_database_old_push_done, req);
 747 }
 748
 749 static bool push_database_old_recv(struct tevent_req *req, int *perr)
 750 {
 751         return generic_recv(req, perr);
 752 }
 753
 754 /*
 755  * Push database to specified nodes (new style)
 756  */
 757
 758 struct push_database_new_state {
 759         struct tevent_context *ev;
 760         struct ctdb_client_context *client;
 761         struct recdb_context *recdb;
 762         uint32_t *pnn_list;
 763         int count;
 764         uint64_t srvid;
 765         uint32_t dmaster;
 766         int fd;
 767         int num_buffers;
 768         int num_buffers_sent;
 769         int num_records;
 770 };
 771
 772 static void push_database_new_started(struct tevent_req *subreq);
 773 static void push_database_new_send_msg(struct tevent_req *req);
 774 static void push_database_new_send_done(struct tevent_req *subreq);
 775 static void push_database_new_confirmed(struct tevent_req *subreq);
 776
 777 static struct tevent_req *push_database_new_send(
 778                         TALLOC_CTX *mem_ctx,
 779                         struct tevent_context *ev,
 780                         struct ctdb_client_context *client,
 781                         uint32_t *pnn_list, int count,
 782                         struct recdb_context *recdb,
 783                         int max_size)
 784 {
 785         struct tevent_req *req, *subreq;
 786         struct push_database_new_state *state;
 787         struct ctdb_req_control request;
 788         struct ctdb_pulldb_ext pulldb_ext;
 789         char *filename;
 790         off_t offset;
 791
 792         req = tevent_req_create(mem_ctx, &state,
 793                                 struct push_database_new_state);
 794         if (req == NULL) {
 795                 return NULL;
 796         }
 797
 798         state->ev = ev;
 799         state->client = client;
 800         state->recdb = recdb;
 801         state->pnn_list = pnn_list;
 802         state->count = count;
 803
 804         state->srvid = srvid_next();
 805         state->dmaster = ctdb_client_pnn(client);
 806         state->num_buffers_sent = 0;
 807         state->num_records = 0;
 808
 809         filename = talloc_asprintf(state, "%s.dat", recdb_path(recdb));
 810         if (tevent_req_nomem(filename, req)) {
 811                 return tevent_req_post(req, ev);
 812         }
 813
 814         state->fd = open(filename, O_RDWR|O_CREAT, 0644);
 815         if (state->fd == -1) {
 816                 tevent_req_error(req, errno);
 817                 return tevent_req_post(req, ev);
 818         }
 819         unlink(filename);
 820         talloc_free(filename);
 821
 822         state->num_buffers = recdb_file(recdb, state, state->dmaster,
 823                                         state->fd, max_size);
 824         if (state->num_buffers == -1) {
 825                 tevent_req_error(req, ENOMEM);
 826                 return tevent_req_post(req, ev);
 827         }
 828
 829         offset = lseek(state->fd, 0, SEEK_SET);
 830         if (offset != 0) {
 831                 tevent_req_error(req, EIO);
 832                 return tevent_req_post(req, ev);
 833         }
 834
 835         pulldb_ext.db_id = recdb_id(recdb);
 836         pulldb_ext.srvid = state->srvid;
 837
 838         ctdb_req_control_db_push_start(&request, &pulldb_ext);
 839         subreq = ctdb_client_control_multi_send(state, ev, client,
 840                                                 pnn_list, count,
 841                                                 TIMEOUT(), &request);
 842         if (tevent_req_nomem(subreq, req)) {
 843                 return tevent_req_post(req, ev);
 844         }
 845         tevent_req_set_callback(subreq, push_database_new_started, req);
 846
 847         return req;
 848 }
 849
 850 static void push_database_new_started(struct tevent_req *subreq)
 851 {
 852         struct tevent_req *req = tevent_req_callback_data(
 853                 subreq, struct tevent_req);
 854         struct push_database_new_state *state = tevent_req_data(
 855                 req, struct push_database_new_state);
 856         int *err_list;
 857         int ret;
 858         bool status;
 859
 860         status = ctdb_client_control_multi_recv(subreq, &ret, state,
 861                                                 &err_list, NULL);
 862         TALLOC_FREE(subreq);
 863         if (! status) {
 864                 int ret2;
 865                 uint32_t pnn;
 866
 867                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
 868                                                        state->count,
 869                                                        err_list, &pnn);
 870                 if (ret2 != 0) {
 871                         D_ERR("control DB_PUSH_START failed for db %s"
 872                               " on node %u, ret=%d\n",
 873                               recdb_name(state->recdb), pnn, ret2);
 874                 } else {
 875                         D_ERR("control DB_PUSH_START failed for db %s,"
 876                               " ret=%d\n",
 877                               recdb_name(state->recdb), ret);
 878                 }
 879                 talloc_free(err_list);
 880
 881                 tevent_req_error(req, ret);
 882                 return;
 883         }
 884
 885         push_database_new_send_msg(req);
 886 }
 887
 888 static void push_database_new_send_msg(struct tevent_req *req)
 889 {
 890         struct push_database_new_state *state = tevent_req_data(
 891                 req, struct push_database_new_state);
 892         struct tevent_req *subreq;
 893         struct ctdb_rec_buffer *recbuf;
 894         struct ctdb_req_message message;
 895         TDB_DATA data;
 896         size_t np;
 897         int ret;
 898
 899         if (state->num_buffers_sent == state->num_buffers) {
 900                 struct ctdb_req_control request;
 901
 902                 ctdb_req_control_db_push_confirm(&request,
 903                                                  recdb_id(state->recdb));
 904                 subreq = ctdb_client_control_multi_send(state, state->ev,
 905                                                         state->client,
 906                                                         state->pnn_list,
 907                                                         state->count,
 908                                                         TIMEOUT(), &request);
 909                 if (tevent_req_nomem(subreq, req)) {
 910                         return;
 911                 }
 912                 tevent_req_set_callback(subreq, push_database_new_confirmed,
 913                                         req);
 914                 return;
 915         }
 916
 917         ret = ctdb_rec_buffer_read(state->fd, state, &recbuf);
 918         if (ret != 0) {
 919                 tevent_req_error(req, ret);
 920                 return;
 921         }
 922
 923         data.dsize = ctdb_rec_buffer_len(recbuf);
 924         data.dptr = talloc_size(state, data.dsize);
 925         if (tevent_req_nomem(data.dptr, req)) {
 926                 return;
 927         }
 928
 929         ctdb_rec_buffer_push(recbuf, data.dptr, &np);
 930
 931         message.srvid = state->srvid;
 932         message.data.data = data;
 933
 934         D_DEBUG("Pushing buffer %d with %d records for db %s\n",
 935                 state->num_buffers_sent, recbuf->count,
 936                 recdb_name(state->recdb));
 937
 938         subreq = ctdb_client_message_multi_send(state, state->ev,
 939                                                 state->client,
 940                                                 state->pnn_list, state->count,
 941                                                 &message);
 942         if (tevent_req_nomem(subreq, req)) {
 943                 return;
 944         }
 945         tevent_req_set_callback(subreq, push_database_new_send_done, req);
 946
 947         state->num_records += recbuf->count;
 948
 949         talloc_free(data.dptr);
 950         talloc_free(recbuf);
 951 }
 952
 953 static void push_database_new_send_done(struct tevent_req *subreq)
 954 {
 955         struct tevent_req *req = tevent_req_callback_data(
 956                 subreq, struct tevent_req);
 957         struct push_database_new_state *state = tevent_req_data(
 958                 req, struct push_database_new_state);
 959         bool status;
 960         int ret;
 961
 962         status = ctdb_client_message_multi_recv(subreq, &ret, NULL, NULL);
 963         TALLOC_FREE(subreq);
 964         if (! status) {
 965                 D_ERR("Sending recovery records failed for %s\n",
 966                       recdb_name(state->recdb));
 967                 tevent_req_error(req, ret);
 968                 return;
 969         }
 970
 971         state->num_buffers_sent += 1;
 972
 973         push_database_new_send_msg(req);
 974 }
 975
 976 static void push_database_new_confirmed(struct tevent_req *subreq)
 977 {
 978         struct tevent_req *req = tevent_req_callback_data(
 979                 subreq, struct tevent_req);
 980         struct push_database_new_state *state = tevent_req_data(
 981                 req, struct push_database_new_state);
 982         struct ctdb_reply_control **reply;
 983         int *err_list;
 984         bool status;
 985         int ret, i;
 986         uint32_t num_records;
 987
 988         status = ctdb_client_control_multi_recv(subreq, &ret, state,
 989                                                 &err_list, &reply);
 990         TALLOC_FREE(subreq);
 991         if (! status) {
 992                 int ret2;
 993                 uint32_t pnn;
 994
 995                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
 996                                                        state->count, err_list,
 997                                                        &pnn);
 998                 if (ret2 != 0) {
 999                         D_ERR("control DB_PUSH_CONFIRM failed for db %s"
1000                               " on node %u, ret=%d\n",
1001                               recdb_name(state->recdb), pnn, ret2);
1002                 } else {
1003                         D_ERR("control DB_PUSH_CONFIRM failed for db %s,"
1004                               " ret=%d\n",
1005                               recdb_name(state->recdb), ret);
1006                 }
1007                 tevent_req_error(req, ret);
1008                 return;
1009         }
1010
1011         for (i=0; i<state->count; i++) {
1012                 ret = ctdb_reply_control_db_push_confirm(reply[i],
1013                                                          &num_records);
1014                 if (ret != 0) {
1015                         tevent_req_error(req, EPROTO);
1016                         return;
1017                 }
1018
1019                 if (num_records != state->num_records) {
1020                         D_ERR("Node %u received %d of %d records for %s\n",
1021                               state->pnn_list[i], num_records,
1022                               state->num_records, recdb_name(state->recdb));
1023                         tevent_req_error(req, EPROTO);
1024                         return;
1025                 }
1026         }
1027
1028         talloc_free(reply);
1029
1030         D_INFO("Pushed %d records for db %s\n",
1031                state->num_records, recdb_name(state->recdb));
1032
1033         tevent_req_done(req);
1034 }
1035
1036 static bool push_database_new_recv(struct tevent_req *req, int *perr)
1037 {
1038         return generic_recv(req, perr);
1039 }
1040
1041 /*
1042  * wrapper for push_database_old and push_database_new
1043  */
1044
1045 struct push_database_state {
1046         bool old_done, new_done;
1047 };
1048
1049 static void push_database_old_done(struct tevent_req *subreq);
1050 static void push_database_new_done(struct tevent_req *subreq);
1051
1052 static struct tevent_req *push_database_send(
1053                         TALLOC_CTX *mem_ctx,
1054                         struct tevent_context *ev,
1055                         struct ctdb_client_context *client,
1056                         uint32_t *pnn_list, int count, uint32_t *caps,
1057                         struct ctdb_tunable_list *tun_list,
1058                         struct recdb_context *recdb)
1059 {
1060         struct tevent_req *req, *subreq;
1061         struct push_database_state *state;
1062         uint32_t *old_list, *new_list;
1063         int old_count, new_count;
1064         int i;
1065
1066         req = tevent_req_create(mem_ctx, &state, struct push_database_state);
1067         if (req == NULL) {
1068                 return NULL;
1069         }
1070
1071         state->old_done = false;
1072         state->new_done = false;
1073
1074         old_count = 0;
1075         new_count = 0;
1076         old_list = talloc_array(state, uint32_t, count);
1077         new_list = talloc_array(state, uint32_t, count);
1078         if (tevent_req_nomem(old_list, req) ||
1079             tevent_req_nomem(new_list,req)) {
1080                 return tevent_req_post(req, ev);
1081         }
1082
1083         for (i=0; i<count; i++) {
1084                 uint32_t pnn = pnn_list[i];
1085
1086                 if (caps[pnn] & CTDB_CAP_FRAGMENTED_CONTROLS) {
1087                         new_list[new_count] = pnn;
1088                         new_count += 1;
1089                 } else {
1090                         old_list[old_count] = pnn;
1091                         old_count += 1;
1092                 }
1093         }
1094
1095         if (old_count > 0) {
1096                 subreq = push_database_old_send(state, ev, client,
1097                                                 old_list, old_count, recdb);
1098                 if (tevent_req_nomem(subreq, req)) {
1099                         return tevent_req_post(req, ev);
1100                 }
1101                 tevent_req_set_callback(subreq, push_database_old_done, req);
1102         } else {
1103                 state->old_done = true;
1104         }
1105
1106         if (new_count > 0) {
1107                 subreq = push_database_new_send(state, ev, client,
1108                                                 new_list, new_count, recdb,
1109                                                 tun_list->rec_buffer_size_limit);
1110                 if (tevent_req_nomem(subreq, req)) {
1111                         return tevent_req_post(req, ev);
1112                 }
1113                 tevent_req_set_callback(subreq, push_database_new_done, req);
1114         } else {
1115                 state->new_done = true;
1116         }
1117
1118         return req;
1119 }
1120
1121 static void push_database_old_done(struct tevent_req *subreq)
1122 {
1123         struct tevent_req *req = tevent_req_callback_data(
1124                 subreq, struct tevent_req);
1125         struct push_database_state *state = tevent_req_data(
1126                 req, struct push_database_state);
1127         bool status;
1128         int ret;
1129
1130         status = push_database_old_recv(subreq, &ret);
1131         if (! status) {
1132                 tevent_req_error(req, ret);
1133                 return;
1134         }
1135
1136         state->old_done = true;
1137
1138         if (state->old_done && state->new_done) {
1139                 tevent_req_done(req);
1140         }
1141 }
1142
1143 static void push_database_new_done(struct tevent_req *subreq)
1144 {
1145         struct tevent_req *req = tevent_req_callback_data(
1146                 subreq, struct tevent_req);
1147         struct push_database_state *state = tevent_req_data(
1148                 req, struct push_database_state);
1149         bool status;
1150         int ret;
1151
1152         status = push_database_new_recv(subreq, &ret);
1153         if (! status) {
1154                 tevent_req_error(req, ret);
1155                 return;
1156         }
1157
1158         state->new_done = true;
1159
1160         if (state->old_done && state->new_done) {
1161                 tevent_req_done(req);
1162         }
1163 }
1164
1165 static bool push_database_recv(struct tevent_req *req, int *perr)
1166 {
1167         return generic_recv(req, perr);
1168 }
1169
1170 /*
1171  * Collect databases using highest sequence number
1172  */
1173
1174 struct collect_highseqnum_db_state {
1175         struct tevent_context *ev;
1176         struct ctdb_client_context *client;
1177         uint32_t *pnn_list;
1178         int count;
1179         uint32_t *caps;
1180         uint32_t *ban_credits;
1181         uint32_t db_id;
1182         struct recdb_context *recdb;
1183         uint32_t max_pnn;
1184 };
1185
1186 static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq);
1187 static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq);
1188
1189 static struct tevent_req *collect_highseqnum_db_send(
1190                         TALLOC_CTX *mem_ctx,
1191                         struct tevent_context *ev,
1192                         struct ctdb_client_context *client,
1193                         uint32_t *pnn_list, int count, uint32_t *caps,
1194                         uint32_t *ban_credits, uint32_t db_id,
1195                         struct recdb_context *recdb)
1196 {
1197         struct tevent_req *req, *subreq;
1198         struct collect_highseqnum_db_state *state;
1199         struct ctdb_req_control request;
1200
1201         req = tevent_req_create(mem_ctx, &state,
1202                                 struct collect_highseqnum_db_state);
1203         if (req == NULL) {
1204                 return NULL;
1205         }
1206
1207         state->ev = ev;
1208         state->client = client;
1209         state->pnn_list = pnn_list;
1210         state->count = count;
1211         state->caps = caps;
1212         state->ban_credits = ban_credits;
1213         state->db_id = db_id;
1214         state->recdb = recdb;
1215
1216         ctdb_req_control_get_db_seqnum(&request, db_id);
1217         subreq = ctdb_client_control_multi_send(mem_ctx, ev, client,
1218                                                 state->pnn_list, state->count,
1219                                                 TIMEOUT(), &request);
1220         if (tevent_req_nomem(subreq, req)) {
1221                 return tevent_req_post(req, ev);
1222         }
1223         tevent_req_set_callback(subreq, collect_highseqnum_db_seqnum_done,
1224                                 req);
1225
1226         return req;
1227 }
1228
1229 static void collect_highseqnum_db_seqnum_done(struct tevent_req *subreq)
1230 {
1231         struct tevent_req *req = tevent_req_callback_data(
1232                 subreq, struct tevent_req);
1233         struct collect_highseqnum_db_state *state = tevent_req_data(
1234                 req, struct collect_highseqnum_db_state);
1235         struct ctdb_reply_control **reply;
1236         int *err_list;
1237         bool status;
1238         int ret, i;
1239         uint64_t seqnum, max_seqnum;
1240
1241         status = ctdb_client_control_multi_recv(subreq, &ret, state,
1242                                                 &err_list, &reply);
1243         TALLOC_FREE(subreq);
1244         if (! status) {
1245                 int ret2;
1246                 uint32_t pnn;
1247
1248                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1249                                                        state->count, err_list,
1250                                                        &pnn);
1251                 if (ret2 != 0) {
1252                         D_ERR("control GET_DB_SEQNUM failed for db %s"
1253                               " on node %u, ret=%d\n",
1254                               recdb_name(state->recdb), pnn, ret2);
1255                 } else {
1256                         D_ERR("control GET_DB_SEQNUM failed for db %s,"
1257                               " ret=%d\n",
1258                               recdb_name(state->recdb), ret);
1259                 }
1260                 tevent_req_error(req, ret);
1261                 return;
1262         }
1263
1264         max_seqnum = 0;
1265         state->max_pnn = state->pnn_list[0];
1266         for (i=0; i<state->count; i++) {
1267                 ret = ctdb_reply_control_get_db_seqnum(reply[i], &seqnum);
1268                 if (ret != 0) {
1269                         tevent_req_error(req, EPROTO);
1270                         return;
1271                 }
1272
1273                 if (max_seqnum < seqnum) {
1274                         max_seqnum = seqnum;
1275                         state->max_pnn = state->pnn_list[i];
1276                 }
1277         }
1278
1279         talloc_free(reply);
1280
1281         D_INFO("Pull persistent db %s from node %d with seqnum 0x%"PRIx64"\n",
1282                recdb_name(state->recdb), state->max_pnn, max_seqnum);
1283
1284         subreq = pull_database_send(state, state->ev, state->client,
1285                                     state->max_pnn,
1286                                     state->caps[state->max_pnn],
1287                                     state->recdb);
1288         if (tevent_req_nomem(subreq, req)) {
1289                 return;
1290         }
1291         tevent_req_set_callback(subreq, collect_highseqnum_db_pulldb_done,
1292                                 req);
1293 }
1294
1295 static void collect_highseqnum_db_pulldb_done(struct tevent_req *subreq)
1296 {
1297         struct tevent_req *req = tevent_req_callback_data(
1298                 subreq, struct tevent_req);
1299         struct collect_highseqnum_db_state *state = tevent_req_data(
1300                 req, struct collect_highseqnum_db_state);
1301         int ret;
1302         bool status;
1303
1304         status = pull_database_recv(subreq, &ret);
1305         TALLOC_FREE(subreq);
1306         if (! status) {
1307                 state->ban_credits[state->max_pnn] += 1;
1308                 tevent_req_error(req, ret);
1309                 return;
1310         }
1311
1312         tevent_req_done(req);
1313 }
1314
1315 static bool collect_highseqnum_db_recv(struct tevent_req *req, int *perr)
1316 {
1317         return generic_recv(req, perr);
1318 }
1319
1320 /*
1321  * Collect all databases
1322  */
1323
1324 struct collect_all_db_state {
1325         struct tevent_context *ev;
1326         struct ctdb_client_context *client;
1327         uint32_t *pnn_list;
1328         int count;
1329         uint32_t *caps;
1330         uint32_t *ban_credits;
1331         uint32_t db_id;
1332         struct recdb_context *recdb;
1333         struct ctdb_pulldb pulldb;
1334         int index;
1335 };
1336
1337 static void collect_all_db_pulldb_done(struct tevent_req *subreq);
1338
1339 static struct tevent_req *collect_all_db_send(
1340                         TALLOC_CTX *mem_ctx,
1341                         struct tevent_context *ev,
1342                         struct ctdb_client_context *client,
1343                         uint32_t *pnn_list, int count, uint32_t *caps,
1344                         uint32_t *ban_credits, uint32_t db_id,
1345                         struct recdb_context *recdb)
1346 {
1347         struct tevent_req *req, *subreq;
1348         struct collect_all_db_state *state;
1349         uint32_t pnn;
1350
1351         req = tevent_req_create(mem_ctx, &state,
1352                                 struct collect_all_db_state);
1353         if (req == NULL) {
1354                 return NULL;
1355         }
1356
1357         state->ev = ev;
1358         state->client = client;
1359         state->pnn_list = pnn_list;
1360         state->count = count;
1361         state->caps = caps;
1362         state->ban_credits = ban_credits;
1363         state->db_id = db_id;
1364         state->recdb = recdb;
1365         state->index = 0;
1366
1367         pnn = state->pnn_list[state->index];
1368
1369         subreq = pull_database_send(state, ev, client, pnn, caps[pnn], recdb);
1370         if (tevent_req_nomem(subreq, req)) {
1371                 return tevent_req_post(req, ev);
1372         }
1373         tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
1374
1375         return req;
1376 }
1377
1378 static void collect_all_db_pulldb_done(struct tevent_req *subreq)
1379 {
1380         struct tevent_req *req = tevent_req_callback_data(
1381                 subreq, struct tevent_req);
1382         struct collect_all_db_state *state = tevent_req_data(
1383                 req, struct collect_all_db_state);
1384         uint32_t pnn;
1385         int ret;
1386         bool status;
1387
1388         status = pull_database_recv(subreq, &ret);
1389         TALLOC_FREE(subreq);
1390         if (! status) {
1391                 pnn = state->pnn_list[state->index];
1392                 state->ban_credits[pnn] += 1;
1393                 tevent_req_error(req, ret);
1394                 return;
1395         }
1396
1397         state->index += 1;
1398         if (state->index == state->count) {
1399                 tevent_req_done(req);
1400                 return;
1401         }
1402
1403         pnn = state->pnn_list[state->index];
1404         subreq = pull_database_send(state, state->ev, state->client,
1405                                     pnn, state->caps[pnn], state->recdb);
1406         if (tevent_req_nomem(subreq, req)) {
1407                 return;
1408         }
1409         tevent_req_set_callback(subreq, collect_all_db_pulldb_done, req);
1410 }
1411
1412 static bool collect_all_db_recv(struct tevent_req *req, int *perr)
1413 {
1414         return generic_recv(req, perr);
1415 }
1416
1417
1418 /**
1419  * For each database do the following:
1420  *  - Get DB name
1421  *  - Get DB path
1422  *  - Freeze database on all nodes
1423  *  - Start transaction on all nodes
1424  *  - Collect database from all nodes
1425  *  - Wipe database on all nodes
1426  *  - Push database to all nodes
1427  *  - Commit transaction on all nodes
1428  *  - Thaw database on all nodes
1429  */
1430
1431 struct recover_db_state {
1432         struct tevent_context *ev;
1433         struct ctdb_client_context *client;
1434         struct ctdb_tunable_list *tun_list;
1435         uint32_t *pnn_list;
1436         int count;
1437         uint32_t *caps;
1438         uint32_t *ban_credits;
1439         uint32_t db_id;
1440         uint8_t db_flags;
1441
1442         uint32_t destnode;
1443         struct ctdb_transdb transdb;
1444
1445         const char *db_name, *db_path;
1446         struct recdb_context *recdb;
1447 };
1448
1449 static void recover_db_name_done(struct tevent_req *subreq);
1450 static void recover_db_path_done(struct tevent_req *subreq);
1451 static void recover_db_freeze_done(struct tevent_req *subreq);
1452 static void recover_db_transaction_started(struct tevent_req *subreq);
1453 static void recover_db_collect_done(struct tevent_req *subreq);
1454 static void recover_db_wipedb_done(struct tevent_req *subreq);
1455 static void recover_db_pushdb_done(struct tevent_req *subreq);
1456 static void recover_db_transaction_committed(struct tevent_req *subreq);
1457 static void recover_db_thaw_done(struct tevent_req *subreq);
1458
1459 static struct tevent_req *recover_db_send(TALLOC_CTX *mem_ctx,
1460                                           struct tevent_context *ev,
1461                                           struct ctdb_client_context *client,
1462                                           struct ctdb_tunable_list *tun_list,
1463                                           uint32_t *pnn_list, int count,
1464                                           uint32_t *caps,
1465                                           uint32_t *ban_credits,
1466                                           uint32_t generation,
1467                                           uint32_t db_id, uint8_t db_flags)
1468 {
1469         struct tevent_req *req, *subreq;
1470         struct recover_db_state *state;
1471         struct ctdb_req_control request;
1472
1473         req = tevent_req_create(mem_ctx, &state, struct recover_db_state);
1474         if (req == NULL) {
1475                 return NULL;
1476         }
1477
1478         state->ev = ev;
1479         state->client = client;
1480         state->tun_list = tun_list;
1481         state->pnn_list = pnn_list;
1482         state->count = count;
1483         state->caps = caps;
1484         state->ban_credits = ban_credits;
1485         state->db_id = db_id;
1486         state->db_flags = db_flags;
1487
1488         state->destnode = ctdb_client_pnn(client);
1489         state->transdb.db_id = db_id;
1490         state->transdb.tid = generation;
1491
1492         ctdb_req_control_get_dbname(&request, db_id);
1493         subreq = ctdb_client_control_send(state, ev, client, state->destnode,
1494                                           TIMEOUT(), &request);
1495         if (tevent_req_nomem(subreq, req)) {
1496                 return tevent_req_post(req, ev);
1497         }
1498         tevent_req_set_callback(subreq, recover_db_name_done, req);
1499
1500         return req;
1501 }
1502
1503 static void recover_db_name_done(struct tevent_req *subreq)
1504 {
1505         struct tevent_req *req = tevent_req_callback_data(
1506                 subreq, struct tevent_req);
1507         struct recover_db_state *state = tevent_req_data(
1508                 req, struct recover_db_state);
1509         struct ctdb_reply_control *reply;
1510         struct ctdb_req_control request;
1511         int ret;
1512         bool status;
1513
1514         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
1515         TALLOC_FREE(subreq);
1516         if (! status) {
1517                 D_ERR("control GET_DBNAME failed for db=0x%x, ret=%d\n",
1518                       state->db_id, ret);
1519                 tevent_req_error(req, ret);
1520                 return;
1521         }
1522
1523         ret = ctdb_reply_control_get_dbname(reply, state, &state->db_name);
1524         if (ret != 0) {
1525                 D_ERR("control GET_DBNAME failed for db=0x%x, ret=%d\n",
1526                       state->db_id, ret);
1527                 tevent_req_error(req, EPROTO);
1528                 return;
1529         }
1530
1531         talloc_free(reply);
1532
1533         ctdb_req_control_getdbpath(&request, state->db_id);
1534         subreq = ctdb_client_control_send(state, state->ev, state->client,
1535                                           state->destnode, TIMEOUT(),
1536                                           &request);
1537         if (tevent_req_nomem(subreq, req)) {
1538                 return;
1539         }
1540         tevent_req_set_callback(subreq, recover_db_path_done, req);
1541 }
1542
1543 static void recover_db_path_done(struct tevent_req *subreq)
1544 {
1545         struct tevent_req *req = tevent_req_callback_data(
1546                 subreq, struct tevent_req);
1547         struct recover_db_state *state = tevent_req_data(
1548                 req, struct recover_db_state);
1549         struct ctdb_reply_control *reply;
1550         struct ctdb_req_control request;
1551         int ret;
1552         bool status;
1553
1554         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
1555         TALLOC_FREE(subreq);
1556         if (! status) {
1557                 D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
1558                       state->db_name, ret);
1559                 tevent_req_error(req, ret);
1560                 return;
1561         }
1562
1563         ret = ctdb_reply_control_getdbpath(reply, state, &state->db_path);
1564         if (ret != 0) {
1565                 D_ERR("control GETDBPATH failed for db %s, ret=%d\n",
1566                       state->db_name, ret);
1567                 tevent_req_error(req, EPROTO);
1568                 return;
1569         }
1570
1571         talloc_free(reply);
1572
1573         ctdb_req_control_db_freeze(&request, state->db_id);
1574         subreq = ctdb_client_control_multi_send(state, state->ev,
1575                                                 state->client,
1576                                                 state->pnn_list, state->count,
1577                                                 TIMEOUT(), &request);
1578         if (tevent_req_nomem(subreq, req)) {
1579                 return;
1580         }
1581         tevent_req_set_callback(subreq, recover_db_freeze_done, req);
1582 }
1583
1584 static void recover_db_freeze_done(struct tevent_req *subreq)
1585 {
1586         struct tevent_req *req = tevent_req_callback_data(
1587                 subreq, struct tevent_req);
1588         struct recover_db_state *state = tevent_req_data(
1589                 req, struct recover_db_state);
1590         struct ctdb_req_control request;
1591         int *err_list;
1592         int ret;
1593         bool status;
1594
1595         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1596                                                 NULL);
1597         TALLOC_FREE(subreq);
1598         if (! status) {
1599                 int ret2;
1600                 uint32_t pnn;
1601
1602                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1603                                                        state->count, err_list,
1604                                                        &pnn);
1605                 if (ret2 != 0) {
1606                         D_ERR("control FREEZE_DB failed for db %s"
1607                               " on node %u, ret=%d\n",
1608                               state->db_name, pnn, ret2);
1609                         state->ban_credits[pnn] += 1;
1610                 } else {
1611                         D_ERR("control FREEZE_DB failed for db %s, ret=%d\n",
1612                               state->db_name, ret);
1613                 }
1614                 tevent_req_error(req, ret);
1615                 return;
1616         }
1617
1618         ctdb_req_control_db_transaction_start(&request, &state->transdb);
1619         subreq = ctdb_client_control_multi_send(state, state->ev,
1620                                                 state->client,
1621                                                 state->pnn_list, state->count,
1622                                                 TIMEOUT(), &request);
1623         if (tevent_req_nomem(subreq, req)) {
1624                 return;
1625         }
1626         tevent_req_set_callback(subreq, recover_db_transaction_started, req);
1627 }
1628
1629 static void recover_db_transaction_started(struct tevent_req *subreq)
1630 {
1631         struct tevent_req *req = tevent_req_callback_data(
1632                 subreq, struct tevent_req);
1633         struct recover_db_state *state = tevent_req_data(
1634                 req, struct recover_db_state);
1635         int *err_list;
1636         int ret;
1637         bool status;
1638
1639         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1640                                                 NULL);
1641         TALLOC_FREE(subreq);
1642         if (! status) {
1643                 int ret2;
1644                 uint32_t pnn;
1645
1646                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1647                                                        state->count,
1648                                                        err_list, &pnn);
1649                 if (ret2 != 0) {
1650                         D_ERR("control TRANSACTION_DB failed for db=%s"
1651                               " on node %u, ret=%d\n",
1652                               state->db_name, pnn, ret2);
1653                 } else {
1654                         D_ERR("control TRANSACTION_DB failed for db=%s,"
1655                               " ret=%d\n", state->db_name, ret);
1656                 }
1657                 tevent_req_error(req, ret);
1658                 return;
1659         }
1660
1661         state->recdb = recdb_create(state, state->db_id, state->db_name,
1662                                     state->db_path,
1663                                     state->tun_list->database_hash_size,
1664                                     state->db_flags & CTDB_DB_FLAGS_PERSISTENT);
1665         if (tevent_req_nomem(state->recdb, req)) {
1666                 return;
1667         }
1668
1669         if ((state->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
1670             (state->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
1671                 subreq = collect_highseqnum_db_send(
1672                                 state, state->ev, state->client,
1673                                 state->pnn_list, state->count, state->caps,
1674                                 state->ban_credits, state->db_id,
1675                                 state->recdb);
1676         } else {
1677                 subreq = collect_all_db_send(
1678                                 state, state->ev, state->client,
1679                                 state->pnn_list, state->count, state->caps,
1680                                 state->ban_credits, state->db_id,
1681                                 state->recdb);
1682         }
1683         if (tevent_req_nomem(subreq, req)) {
1684                 return;
1685         }
1686         tevent_req_set_callback(subreq, recover_db_collect_done, req);
1687 }
1688
1689 static void recover_db_collect_done(struct tevent_req *subreq)
1690 {
1691         struct tevent_req *req = tevent_req_callback_data(
1692                 subreq, struct tevent_req);
1693         struct recover_db_state *state = tevent_req_data(
1694                 req, struct recover_db_state);
1695         struct ctdb_req_control request;
1696         int ret;
1697         bool status;
1698
1699         if ((state->db_flags & CTDB_DB_FLAGS_PERSISTENT) ||
1700             (state->db_flags & CTDB_DB_FLAGS_REPLICATED)) {
1701                 status = collect_highseqnum_db_recv(subreq, &ret);
1702         } else {
1703                 status = collect_all_db_recv(subreq, &ret);
1704         }
1705         TALLOC_FREE(subreq);
1706         if (! status) {
1707                 tevent_req_error(req, ret);
1708                 return;
1709         }
1710
1711         ctdb_req_control_wipe_database(&request, &state->transdb);
1712         subreq = ctdb_client_control_multi_send(state, state->ev,
1713                                                 state->client,
1714                                                 state->pnn_list, state->count,
1715                                                 TIMEOUT(), &request);
1716         if (tevent_req_nomem(subreq, req)) {
1717                 return;
1718         }
1719         tevent_req_set_callback(subreq, recover_db_wipedb_done, req);
1720 }
1721
1722 static void recover_db_wipedb_done(struct tevent_req *subreq)
1723 {
1724         struct tevent_req *req = tevent_req_callback_data(
1725                 subreq, struct tevent_req);
1726         struct recover_db_state *state = tevent_req_data(
1727                 req, struct recover_db_state);
1728         int *err_list;
1729         int ret;
1730         bool status;
1731
1732         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1733                                                 NULL);
1734         TALLOC_FREE(subreq);
1735         if (! status) {
1736                 int ret2;
1737                 uint32_t pnn;
1738
1739                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1740                                                        state->count,
1741                                                        err_list, &pnn);
1742                 if (ret2 != 0) {
1743                         D_ERR("control WIPEDB failed for db %s on node %u,"
1744                               " ret=%d\n", state->db_name, pnn, ret2);
1745                 } else {
1746                         D_ERR("control WIPEDB failed for db %s, ret=%d\n",
1747                               state->db_name, ret);
1748                 }
1749                 tevent_req_error(req, ret);
1750                 return;
1751         }
1752
1753         subreq = push_database_send(state, state->ev, state->client,
1754                                     state->pnn_list, state->count,
1755                                     state->caps, state->tun_list,
1756                                     state->recdb);
1757         if (tevent_req_nomem(subreq, req)) {
1758                 return;
1759         }
1760         tevent_req_set_callback(subreq, recover_db_pushdb_done, req);
1761 }
1762
1763 static void recover_db_pushdb_done(struct tevent_req *subreq)
1764 {
1765         struct tevent_req *req = tevent_req_callback_data(
1766                 subreq, struct tevent_req);
1767         struct recover_db_state *state = tevent_req_data(
1768                 req, struct recover_db_state);
1769         struct ctdb_req_control request;
1770         int ret;
1771         bool status;
1772
1773         status = push_database_recv(subreq, &ret);
1774         TALLOC_FREE(subreq);
1775         if (! status) {
1776                 tevent_req_error(req, ret);
1777                 return;
1778         }
1779
1780         TALLOC_FREE(state->recdb);
1781
1782         ctdb_req_control_db_transaction_commit(&request, &state->transdb);
1783         subreq = ctdb_client_control_multi_send(state, state->ev,
1784                                                 state->client,
1785                                                 state->pnn_list, state->count,
1786                                                 TIMEOUT(), &request);
1787         if (tevent_req_nomem(subreq, req)) {
1788                 return;
1789         }
1790         tevent_req_set_callback(subreq, recover_db_transaction_committed, req);
1791 }
1792
1793 static void recover_db_transaction_committed(struct tevent_req *subreq)
1794 {
1795         struct tevent_req *req = tevent_req_callback_data(
1796                 subreq, struct tevent_req);
1797         struct recover_db_state *state = tevent_req_data(
1798                 req, struct recover_db_state);
1799         struct ctdb_req_control request;
1800         int *err_list;
1801         int ret;
1802         bool status;
1803
1804         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1805                                                 NULL);
1806         TALLOC_FREE(subreq);
1807         if (! status) {
1808                 int ret2;
1809                 uint32_t pnn;
1810
1811                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1812                                                        state->count,
1813                                                        err_list, &pnn);
1814                 if (ret2 != 0) {
1815                         D_ERR("control DB_TRANSACTION_COMMIT failed for db %s"
1816                               " on node %u, ret=%d\n",
1817                               state->db_name, pnn, ret2);
1818                 } else {
1819                         D_ERR("control DB_TRANSACTION_COMMIT failed for db %s,"
1820                               " ret=%d\n", state->db_name, ret);
1821                 }
1822                 tevent_req_error(req, ret);
1823                 return;
1824         }
1825
1826         ctdb_req_control_db_thaw(&request, state->db_id);
1827         subreq = ctdb_client_control_multi_send(state, state->ev,
1828                                                 state->client,
1829                                                 state->pnn_list, state->count,
1830                                                 TIMEOUT(), &request);
1831         if (tevent_req_nomem(subreq, req)) {
1832                 return;
1833         }
1834         tevent_req_set_callback(subreq, recover_db_thaw_done, req);
1835 }
1836
1837 static void recover_db_thaw_done(struct tevent_req *subreq)
1838 {
1839         struct tevent_req *req = tevent_req_callback_data(
1840                 subreq, struct tevent_req);
1841         struct recover_db_state *state = tevent_req_data(
1842                 req, struct recover_db_state);
1843         int *err_list;
1844         int ret;
1845         bool status;
1846
1847         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
1848                                                 NULL);
1849         TALLOC_FREE(subreq);
1850         if (! status) {
1851                 int ret2;
1852                 uint32_t pnn;
1853
1854                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
1855                                                        state->count,
1856                                                        err_list, &pnn);
1857                 if (ret2 != 0) {
1858                         D_ERR("control DB_THAW failed for db %s on node %u,"
1859                               " ret=%d\n", state->db_name, pnn, ret2);
1860                 } else {
1861                         D_ERR("control DB_THAW failed for db %s, ret=%d\n",
1862                               state->db_name, ret);
1863                 }
1864                 tevent_req_error(req, ret);
1865                 return;
1866         }
1867
1868         tevent_req_done(req);
1869 }
1870
1871 static bool recover_db_recv(struct tevent_req *req)
1872 {
1873         return generic_recv(req, NULL);
1874 }
1875
1876
1877 /*
1878  * Start database recovery for each database
1879  *
1880  * Try to recover each database 5 times before failing recovery.
1881  */
1882
1883 struct db_recovery_state {
1884         struct tevent_context *ev;
1885         struct ctdb_dbid_map *dbmap;
1886         int num_replies;
1887         int num_failed;
1888 };
1889
1890 struct db_recovery_one_state {
1891         struct tevent_req *req;
1892         struct ctdb_client_context *client;
1893         struct ctdb_dbid_map *dbmap;
1894         struct ctdb_tunable_list *tun_list;
1895         uint32_t *pnn_list;
1896         int count;
1897         uint32_t *caps;
1898         uint32_t *ban_credits;
1899         uint32_t generation;
1900         uint32_t db_id;
1901         uint8_t db_flags;
1902         int num_fails;
1903 };
1904
1905 static void db_recovery_one_done(struct tevent_req *subreq);
1906
1907 static struct tevent_req *db_recovery_send(TALLOC_CTX *mem_ctx,
1908                                            struct tevent_context *ev,
1909                                            struct ctdb_client_context *client,
1910                                            struct ctdb_dbid_map *dbmap,
1911                                            struct ctdb_tunable_list *tun_list,
1912                                            uint32_t *pnn_list, int count,
1913                                            uint32_t *caps,
1914                                            uint32_t *ban_credits,
1915                                            uint32_t generation)
1916 {
1917         struct tevent_req *req, *subreq;
1918         struct db_recovery_state *state;
1919         int i;
1920
1921         req = tevent_req_create(mem_ctx, &state, struct db_recovery_state);
1922         if (req == NULL) {
1923                 return NULL;
1924         }
1925
1926         state->ev = ev;
1927         state->dbmap = dbmap;
1928         state->num_replies = 0;
1929         state->num_failed = 0;
1930
1931         if (dbmap->num == 0) {
1932                 tevent_req_done(req);
1933                 return tevent_req_post(req, ev);
1934         }
1935
1936         for (i=0; i<dbmap->num; i++) {
1937                 struct db_recovery_one_state *substate;
1938
1939                 substate = talloc_zero(state, struct db_recovery_one_state);
1940                 if (tevent_req_nomem(substate, req)) {
1941                         return tevent_req_post(req, ev);
1942                 }
1943
1944                 substate->req = req;
1945                 substate->client = client;
1946                 substate->dbmap = dbmap;
1947                 substate->tun_list = tun_list;
1948                 substate->pnn_list = pnn_list;
1949                 substate->count = count;
1950                 substate->caps = caps;
1951                 substate->ban_credits = ban_credits;
1952                 substate->generation = generation;
1953                 substate->db_id = dbmap->dbs[i].db_id;
1954                 substate->db_flags = dbmap->dbs[i].flags;
1955
1956                 subreq = recover_db_send(state, ev, client, tun_list,
1957                                          pnn_list, count, caps, ban_credits,
1958                                          generation, substate->db_id,
1959                                          substate->db_flags);
1960                 if (tevent_req_nomem(subreq, req)) {
1961                         return tevent_req_post(req, ev);
1962                 }
1963                 tevent_req_set_callback(subreq, db_recovery_one_done,
1964                                         substate);
1965                 D_NOTICE("recover database 0x%08x\n", substate->db_id);
1966         }
1967
1968         return req;
1969 }
1970
1971 static void db_recovery_one_done(struct tevent_req *subreq)
1972 {
1973         struct db_recovery_one_state *substate = tevent_req_callback_data(
1974                 subreq, struct db_recovery_one_state);
1975         struct tevent_req *req = substate->req;
1976         struct db_recovery_state *state = tevent_req_data(
1977                 req, struct db_recovery_state);
1978         bool status;
1979
1980         status = recover_db_recv(subreq);
1981         TALLOC_FREE(subreq);
1982
1983         if (status) {
1984                 talloc_free(substate);
1985                 goto done;
1986         }
1987
1988         substate->num_fails += 1;
1989         if (substate->num_fails < NUM_RETRIES) {
1990                 subreq = recover_db_send(state, state->ev, substate->client,
1991                                          substate->tun_list,
1992                                          substate->pnn_list, substate->count,
1993                                          substate->caps, substate->ban_credits,
1994                                          substate->generation, substate->db_id,
1995                                          substate->db_flags);
1996                 if (tevent_req_nomem(subreq, req)) {
1997                         goto failed;
1998                 }
1999                 tevent_req_set_callback(subreq, db_recovery_one_done, substate);
2000                 D_NOTICE("recover database 0x%08x, attempt %d\n",
2001                          substate->db_id, substate->num_fails+1);
2002                 return;
2003         }
2004
2005 failed:
2006         state->num_failed += 1;
2007
2008 done:
2009         state->num_replies += 1;
2010
2011         if (state->num_replies == state->dbmap->num) {
2012                 tevent_req_done(req);
2013         }
2014 }
2015
2016 static bool db_recovery_recv(struct tevent_req *req, int *count)
2017 {
2018         struct db_recovery_state *state = tevent_req_data(
2019                 req, struct db_recovery_state);
2020         int err;
2021
2022         if (tevent_req_is_unix_error(req, &err)) {
2023                 *count = 0;
2024                 return false;
2025         }
2026
2027         *count = state->num_replies - state->num_failed;
2028
2029         if (state->num_failed > 0) {
2030                 return false;
2031         }
2032
2033         return true;
2034 }
2035
2036
2037 /*
2038  * Run the parallel database recovery
2039  *
2040  * - Get tunables
2041  * - Get nodemap
2042  * - Get vnnmap
2043  * - Get capabilities from all nodes
2044  * - Get dbmap
2045  * - Set RECOVERY_ACTIVE
2046  * - Send START_RECOVERY
2047  * - Update vnnmap on all nodes
2048  * - Run database recovery
2049  * - Set RECOVERY_NORMAL
2050  * - Send END_RECOVERY
2051  */
2052
2053 struct recovery_state {
2054         struct tevent_context *ev;
2055         struct ctdb_client_context *client;
2056         uint32_t generation;
2057         uint32_t *pnn_list;
2058         int count;
2059         uint32_t destnode;
2060         struct ctdb_node_map *nodemap;
2061         uint32_t *caps;
2062         uint32_t *ban_credits;
2063         struct ctdb_tunable_list *tun_list;
2064         struct ctdb_vnn_map *vnnmap;
2065         struct ctdb_dbid_map *dbmap;
2066 };
2067
2068 static void recovery_tunables_done(struct tevent_req *subreq);
2069 static void recovery_nodemap_done(struct tevent_req *subreq);
2070 static void recovery_vnnmap_done(struct tevent_req *subreq);
2071 static void recovery_capabilities_done(struct tevent_req *subreq);
2072 static void recovery_dbmap_done(struct tevent_req *subreq);
2073 static void recovery_active_done(struct tevent_req *subreq);
2074 static void recovery_start_recovery_done(struct tevent_req *subreq);
2075 static void recovery_vnnmap_update_done(struct tevent_req *subreq);
2076 static void recovery_db_recovery_done(struct tevent_req *subreq);
2077 static void recovery_failed_done(struct tevent_req *subreq);
2078 static void recovery_normal_done(struct tevent_req *subreq);
2079 static void recovery_end_recovery_done(struct tevent_req *subreq);
2080
2081 static struct tevent_req *recovery_send(TALLOC_CTX *mem_ctx,
2082                                         struct tevent_context *ev,
2083                                         struct ctdb_client_context *client,
2084                                         uint32_t generation)
2085 {
2086         struct tevent_req *req, *subreq;
2087         struct recovery_state *state;
2088         struct ctdb_req_control request;
2089
2090         req = tevent_req_create(mem_ctx, &state, struct recovery_state);
2091         if (req == NULL) {
2092                 return NULL;
2093         }
2094
2095         state->ev = ev;
2096         state->client = client;
2097         state->generation = generation;
2098         state->destnode = ctdb_client_pnn(client);
2099
2100         ctdb_req_control_get_all_tunables(&request);
2101         subreq = ctdb_client_control_send(state, state->ev, state->client,
2102                                           state->destnode, TIMEOUT(),
2103                                           &request);
2104         if (tevent_req_nomem(subreq, req)) {
2105                 return tevent_req_post(req, ev);
2106         }
2107         tevent_req_set_callback(subreq, recovery_tunables_done, req);
2108
2109         return req;
2110 }
2111
2112 static void recovery_tunables_done(struct tevent_req *subreq)
2113 {
2114         struct tevent_req *req = tevent_req_callback_data(
2115                 subreq, struct tevent_req);
2116         struct recovery_state *state = tevent_req_data(
2117                 req, struct recovery_state);
2118         struct ctdb_reply_control *reply;
2119         struct ctdb_req_control request;
2120         int ret;
2121         bool status;
2122
2123         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2124         TALLOC_FREE(subreq);
2125         if (! status) {
2126                 D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
2127                 tevent_req_error(req, ret);
2128                 return;
2129         }
2130
2131         ret = ctdb_reply_control_get_all_tunables(reply, state,
2132                                                   &state->tun_list);
2133         if (ret != 0) {
2134                 D_ERR("control GET_ALL_TUNABLES failed, ret=%d\n", ret);
2135                 tevent_req_error(req, EPROTO);
2136                 return;
2137         }
2138
2139         talloc_free(reply);
2140
2141         recover_timeout = state->tun_list->recover_timeout;
2142
2143         ctdb_req_control_get_nodemap(&request);
2144         subreq = ctdb_client_control_send(state, state->ev, state->client,
2145                                           state->destnode, TIMEOUT(),
2146                                           &request);
2147         if (tevent_req_nomem(subreq, req)) {
2148                 return;
2149         }
2150         tevent_req_set_callback(subreq, recovery_nodemap_done, req);
2151 }
2152
2153 static void recovery_nodemap_done(struct tevent_req *subreq)
2154 {
2155         struct tevent_req *req = tevent_req_callback_data(
2156                 subreq, struct tevent_req);
2157         struct recovery_state *state = tevent_req_data(
2158                 req, struct recovery_state);
2159         struct ctdb_reply_control *reply;
2160         struct ctdb_req_control request;
2161         bool status;
2162         int ret;
2163
2164         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2165         TALLOC_FREE(subreq);
2166         if (! status) {
2167                 D_ERR("control GET_NODEMAP failed to node %u, ret=%d\n",
2168                       state->destnode, ret);
2169                 tevent_req_error(req, ret);
2170                 return;
2171         }
2172
2173         ret = ctdb_reply_control_get_nodemap(reply, state, &state->nodemap);
2174         if (ret != 0) {
2175                 D_ERR("control GET_NODEMAP failed, ret=%d\n", ret);
2176                 tevent_req_error(req, ret);
2177                 return;
2178         }
2179
2180         state->count = list_of_active_nodes(state->nodemap, CTDB_UNKNOWN_PNN,
2181                                             state, &state->pnn_list);
2182         if (state->count <= 0) {
2183                 tevent_req_error(req, ENOMEM);
2184                 return;
2185         }
2186
2187         state->ban_credits = talloc_zero_array(state, uint32_t,
2188                                                state->nodemap->num);
2189         if (tevent_req_nomem(state->ban_credits, req)) {
2190                 return;
2191         }
2192
2193         ctdb_req_control_getvnnmap(&request);
2194         subreq = ctdb_client_control_send(state, state->ev, state->client,
2195                                           state->destnode, TIMEOUT(),
2196                                           &request);
2197         if (tevent_req_nomem(subreq, req)) {
2198                 return;
2199         }
2200         tevent_req_set_callback(subreq, recovery_vnnmap_done, req);
2201 }
2202
2203 static void recovery_vnnmap_done(struct tevent_req *subreq)
2204 {
2205         struct tevent_req *req = tevent_req_callback_data(
2206                 subreq, struct tevent_req);
2207         struct recovery_state *state = tevent_req_data(
2208                 req, struct recovery_state);
2209         struct ctdb_reply_control *reply;
2210         struct ctdb_req_control request;
2211         bool status;
2212         int ret;
2213
2214         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2215         TALLOC_FREE(subreq);
2216         if (! status) {
2217                 D_ERR("control GETVNNMAP failed to node %u, ret=%d\n",
2218                       state->destnode, ret);
2219                 tevent_req_error(req, ret);
2220                 return;
2221         }
2222
2223         ret = ctdb_reply_control_getvnnmap(reply, state, &state->vnnmap);
2224         if (ret != 0) {
2225                 D_ERR("control GETVNNMAP failed, ret=%d\n", ret);
2226                 tevent_req_error(req, ret);
2227                 return;
2228         }
2229
2230         ctdb_req_control_get_capabilities(&request);
2231         subreq = ctdb_client_control_multi_send(state, state->ev,
2232                                                 state->client,
2233                                                 state->pnn_list, state->count,
2234                                                 TIMEOUT(), &request);
2235         if (tevent_req_nomem(subreq, req)) {
2236                 return;
2237         }
2238         tevent_req_set_callback(subreq, recovery_capabilities_done, req);
2239 }
2240
2241 static void recovery_capabilities_done(struct tevent_req *subreq)
2242 {
2243         struct tevent_req *req = tevent_req_callback_data(
2244                 subreq, struct tevent_req);
2245         struct recovery_state *state = tevent_req_data(
2246                 req, struct recovery_state);
2247         struct ctdb_reply_control **reply;
2248         struct ctdb_req_control request;
2249         int *err_list;
2250         int ret, i;
2251         bool status;
2252
2253         status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2254                                                 &reply);
2255         TALLOC_FREE(subreq);
2256         if (! status) {
2257                 int ret2;
2258                 uint32_t pnn;
2259
2260                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2261                                                        state->count,
2262                                                        err_list, &pnn);
2263                 if (ret2 != 0) {
2264                         D_ERR("control GET_CAPABILITIES failed on node %u,"
2265                               " ret=%d\n", pnn, ret2);
2266                 } else {
2267                         D_ERR("control GET_CAPABILITIES failed, ret=%d\n",
2268                               ret);
2269                 }
2270                 tevent_req_error(req, ret);
2271                 return;
2272         }
2273
2274         /* Make the array size same as nodemap */
2275         state->caps = talloc_zero_array(state, uint32_t,
2276                                         state->nodemap->num);
2277         if (tevent_req_nomem(state->caps, req)) {
2278                 return;
2279         }
2280
2281         for (i=0; i<state->count; i++) {
2282                 uint32_t pnn;
2283
2284                 pnn = state->pnn_list[i];
2285                 ret = ctdb_reply_control_get_capabilities(reply[i],
2286                                                           &state->caps[pnn]);
2287                 if (ret != 0) {
2288                         D_ERR("control GET_CAPABILITIES failed on node %u\n",
2289                               pnn);
2290                         tevent_req_error(req, EPROTO);
2291                         return;
2292                 }
2293         }
2294
2295         talloc_free(reply);
2296
2297         ctdb_req_control_get_dbmap(&request);
2298         subreq = ctdb_client_control_send(state, state->ev, state->client,
2299                                           state->destnode, TIMEOUT(),
2300                                           &request);
2301         if (tevent_req_nomem(subreq, req)) {
2302                 return;
2303         }
2304         tevent_req_set_callback(subreq, recovery_dbmap_done, req);
2305 }
2306
2307 static void recovery_dbmap_done(struct tevent_req *subreq)
2308 {
2309         struct tevent_req *req = tevent_req_callback_data(
2310                 subreq, struct tevent_req);
2311         struct recovery_state *state = tevent_req_data(
2312                 req, struct recovery_state);
2313         struct ctdb_reply_control *reply;
2314         struct ctdb_req_control request;
2315         int ret;
2316         bool status;
2317
2318         status = ctdb_client_control_recv(subreq, &ret, state, &reply);
2319         TALLOC_FREE(subreq);
2320         if (! status) {
2321                 D_ERR("control GET_DBMAP failed to node %u, ret=%d\n",
2322                       state->destnode, ret);
2323                 tevent_req_error(req, ret);
2324                 return;
2325         }
2326
2327         ret = ctdb_reply_control_get_dbmap(reply, state, &state->dbmap);
2328         if (ret != 0) {
2329                 D_ERR("control GET_DBMAP failed, ret=%d\n", ret);
2330                 tevent_req_error(req, ret);
2331                 return;
2332         }
2333
2334         ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_ACTIVE);
2335         subreq = ctdb_client_control_multi_send(state, state->ev,
2336                                                 state->client,
2337                                                 state->pnn_list, state->count,
2338                                                 TIMEOUT(), &request);
2339         if (tevent_req_nomem(subreq, req)) {
2340                 return;
2341         }
2342         tevent_req_set_callback(subreq, recovery_active_done, req);
2343 }
2344
2345 static void recovery_active_done(struct tevent_req *subreq)
2346 {
2347         struct tevent_req *req = tevent_req_callback_data(
2348                 subreq, struct tevent_req);
2349         struct recovery_state *state = tevent_req_data(
2350                 req, struct recovery_state);
2351         struct ctdb_req_control request;
2352         struct ctdb_vnn_map *vnnmap;
2353         int *err_list;
2354         int ret, count, i;
2355         bool status;
2356
2357         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2358                                                 NULL);
2359         TALLOC_FREE(subreq);
2360         if (! status) {
2361                 int ret2;
2362                 uint32_t pnn;
2363
2364                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2365                                                        state->count,
2366                                                        err_list, &pnn);
2367                 if (ret2 != 0) {
2368                         D_ERR("failed to set recovery mode ACTIVE on node %u,"
2369                               " ret=%d\n", pnn, ret2);
2370                 } else {
2371                         D_ERR("failed to set recovery mode ACTIVE, ret=%d\n",
2372                               ret);
2373                 }
2374                 tevent_req_error(req, ret);
2375                 return;
2376         }
2377
2378         D_ERR("Set recovery mode to ACTIVE\n");
2379
2380         /* Calculate new VNNMAP */
2381         count = 0;
2382         for (i=0; i<state->nodemap->num; i++) {
2383                 if (state->nodemap->node[i].flags & NODE_FLAGS_INACTIVE) {
2384                         continue;
2385                 }
2386                 if (!(state->caps[i] & CTDB_CAP_LMASTER)) {
2387                         continue;
2388                 }
2389                 count += 1;
2390         }
2391
2392         if (count == 0) {
2393                 D_WARNING("No active lmasters found. Adding recmaster anyway\n");
2394         }
2395
2396         vnnmap = talloc_zero(state, struct ctdb_vnn_map);
2397         if (tevent_req_nomem(vnnmap, req)) {
2398                 return;
2399         }
2400
2401         vnnmap->size = (count == 0 ? 1 : count);
2402         vnnmap->map = talloc_array(vnnmap, uint32_t, vnnmap->size);
2403         if (tevent_req_nomem(vnnmap->map, req)) {
2404                 return;
2405         }
2406
2407         if (count == 0) {
2408                 vnnmap->map[0] = state->destnode;
2409         } else {
2410                 count = 0;
2411                 for (i=0; i<state->nodemap->num; i++) {
2412                         if (state->nodemap->node[i].flags &
2413                             NODE_FLAGS_INACTIVE) {
2414                                 continue;
2415                         }
2416                         if (!(state->caps[i] & CTDB_CAP_LMASTER)) {
2417                                 continue;
2418                         }
2419
2420                         vnnmap->map[count] = state->nodemap->node[i].pnn;
2421                         count += 1;
2422                 }
2423         }
2424
2425         vnnmap->generation = state->generation;
2426
2427         talloc_free(state->vnnmap);
2428         state->vnnmap = vnnmap;
2429
2430         ctdb_req_control_start_recovery(&request);
2431         subreq = ctdb_client_control_multi_send(state, state->ev,
2432                                                 state->client,
2433                                                 state->pnn_list, state->count,
2434                                                 TIMEOUT(), &request);
2435         if (tevent_req_nomem(subreq, req)) {
2436                 return;
2437         }
2438         tevent_req_set_callback(subreq, recovery_start_recovery_done, req);
2439 }
2440
2441 static void recovery_start_recovery_done(struct tevent_req *subreq)
2442 {
2443         struct tevent_req *req = tevent_req_callback_data(
2444                 subreq, struct tevent_req);
2445         struct recovery_state *state = tevent_req_data(
2446                 req, struct recovery_state);
2447         struct ctdb_req_control request;
2448         int *err_list;
2449         int ret;
2450         bool status;
2451
2452         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2453                                                 NULL);
2454         TALLOC_FREE(subreq);
2455         if (! status) {
2456                 int ret2;
2457                 uint32_t pnn;
2458
2459                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2460                                                        state->count,
2461                                                        err_list, &pnn);
2462                 if (ret2 != 0) {
2463                         D_ERR("failed to run start_recovery event on node %u,"
2464                               " ret=%d\n", pnn, ret2);
2465                 } else {
2466                         D_ERR("failed to run start_recovery event, ret=%d\n",
2467                               ret);
2468                 }
2469                 tevent_req_error(req, ret);
2470                 return;
2471         }
2472
2473         D_ERR("start_recovery event finished\n");
2474
2475         ctdb_req_control_setvnnmap(&request, state->vnnmap);
2476         subreq = ctdb_client_control_multi_send(state, state->ev,
2477                                                 state->client,
2478                                                 state->pnn_list, state->count,
2479                                                 TIMEOUT(), &request);
2480         if (tevent_req_nomem(subreq, req)) {
2481                 return;
2482         }
2483         tevent_req_set_callback(subreq, recovery_vnnmap_update_done, req);
2484 }
2485
2486 static void recovery_vnnmap_update_done(struct tevent_req *subreq)
2487 {
2488         struct tevent_req *req = tevent_req_callback_data(
2489                 subreq, struct tevent_req);
2490         struct recovery_state *state = tevent_req_data(
2491                 req, struct recovery_state);
2492         int *err_list;
2493         int ret;
2494         bool status;
2495
2496         status = ctdb_client_control_multi_recv(subreq, &ret, NULL, &err_list,
2497                                                 NULL);
2498         TALLOC_FREE(subreq);
2499         if (! status) {
2500                 int ret2;
2501                 uint32_t pnn;
2502
2503                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2504                                                        state->count,
2505                                                        err_list, &pnn);
2506                 if (ret2 != 0) {
2507                         D_ERR("failed to update VNNMAP on node %u, ret=%d\n",
2508                               pnn, ret2);
2509                 } else {
2510                         D_ERR("failed to update VNNMAP, ret=%d\n", ret);
2511                 }
2512                 tevent_req_error(req, ret);
2513                 return;
2514         }
2515
2516         D_NOTICE("updated VNNMAP\n");
2517
2518         subreq = db_recovery_send(state, state->ev, state->client,
2519                                   state->dbmap, state->tun_list,
2520                                   state->pnn_list, state->count,
2521                                   state->caps, state->ban_credits,
2522                                   state->vnnmap->generation);
2523         if (tevent_req_nomem(subreq, req)) {
2524                 return;
2525         }
2526         tevent_req_set_callback(subreq, recovery_db_recovery_done, req);
2527 }
2528
2529 static void recovery_db_recovery_done(struct tevent_req *subreq)
2530 {
2531         struct tevent_req *req = tevent_req_callback_data(
2532                 subreq, struct tevent_req);
2533         struct recovery_state *state = tevent_req_data(
2534                 req, struct recovery_state);
2535         struct ctdb_req_control request;
2536         bool status;
2537         int count;
2538
2539         status = db_recovery_recv(subreq, &count);
2540         TALLOC_FREE(subreq);
2541
2542         D_ERR("%d of %d databases recovered\n", count, state->dbmap->num);
2543
2544         if (! status) {
2545                 uint32_t max_pnn = CTDB_UNKNOWN_PNN, max_credits = 0;
2546                 int i;
2547
2548                 /* Bans are not enabled */
2549                 if (state->tun_list->enable_bans == 0) {
2550                         tevent_req_error(req, EIO);
2551                         return;
2552                 }
2553
2554                 for (i=0; i<state->count; i++) {
2555                         uint32_t pnn;
2556                         pnn = state->pnn_list[i];
2557                         if (state->ban_credits[pnn] > max_credits) {
2558                                 max_pnn = pnn;
2559                                 max_credits = state->ban_credits[pnn];
2560                         }
2561                 }
2562
2563                 /* If pulling database fails multiple times */
2564                 if (max_credits >= NUM_RETRIES) {
2565                         struct ctdb_req_message message;
2566
2567                         D_ERR("Assigning banning credits to node %u\n",
2568                               max_pnn);
2569
2570                         message.srvid = CTDB_SRVID_BANNING;
2571                         message.data.pnn = max_pnn;
2572
2573                         subreq = ctdb_client_message_send(
2574                                         state, state->ev, state->client,
2575                                         ctdb_client_pnn(state->client),
2576                                         &message);
2577                         if (tevent_req_nomem(subreq, req)) {
2578                                 return;
2579                         }
2580                         tevent_req_set_callback(subreq, recovery_failed_done,
2581                                                 req);
2582                 } else {
2583                         tevent_req_error(req, EIO);
2584                 }
2585                 return;
2586         }
2587
2588         ctdb_req_control_set_recmode(&request, CTDB_RECOVERY_NORMAL);
2589         subreq = ctdb_client_control_multi_send(state, state->ev,
2590                                                 state->client,
2591                                                 state->pnn_list, state->count,
2592                                                 TIMEOUT(), &request);
2593         if (tevent_req_nomem(subreq, req)) {
2594                 return;
2595         }
2596         tevent_req_set_callback(subreq, recovery_normal_done, req);
2597 }
2598
2599 static void recovery_failed_done(struct tevent_req *subreq)
2600 {
2601         struct tevent_req *req = tevent_req_callback_data(
2602                 subreq, struct tevent_req);
2603         int ret;
2604         bool status;
2605
2606         status = ctdb_client_message_recv(subreq, &ret);
2607         TALLOC_FREE(subreq);
2608         if (! status) {
2609                 D_ERR("failed to assign banning credits, ret=%d\n", ret);
2610         }
2611
2612         tevent_req_error(req, EIO);
2613 }
2614
2615 static void recovery_normal_done(struct tevent_req *subreq)
2616 {
2617         struct tevent_req *req = tevent_req_callback_data(
2618                 subreq, struct tevent_req);
2619         struct recovery_state *state = tevent_req_data(
2620                 req, struct recovery_state);
2621         struct ctdb_req_control request;
2622         int *err_list;
2623         int ret;
2624         bool status;
2625
2626         status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2627                                                 NULL);
2628         TALLOC_FREE(subreq);
2629         if (! status) {
2630                 int ret2;
2631                 uint32_t pnn;
2632
2633                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2634                                                        state->count,
2635                                                        err_list, &pnn);
2636                 if (ret2 != 0) {
2637                         D_ERR("failed to set recovery mode NORMAL on node %u,"
2638                               " ret=%d\n", pnn, ret2);
2639                 } else {
2640                         D_ERR("failed to set recovery mode NORMAL, ret=%d\n",
2641                               ret);
2642                 }
2643                 tevent_req_error(req, ret);
2644                 return;
2645         }
2646
2647         D_ERR("Set recovery mode to NORMAL\n");
2648
2649         ctdb_req_control_end_recovery(&request);
2650         subreq = ctdb_client_control_multi_send(state, state->ev,
2651                                                 state->client,
2652                                                 state->pnn_list, state->count,
2653                                                 TIMEOUT(), &request);
2654         if (tevent_req_nomem(subreq, req)) {
2655                 return;
2656         }
2657         tevent_req_set_callback(subreq, recovery_end_recovery_done, req);
2658 }
2659
2660 static void recovery_end_recovery_done(struct tevent_req *subreq)
2661 {
2662         struct tevent_req *req = tevent_req_callback_data(
2663                 subreq, struct tevent_req);
2664         struct recovery_state *state = tevent_req_data(
2665                 req, struct recovery_state);
2666         int *err_list;
2667         int ret;
2668         bool status;
2669
2670         status = ctdb_client_control_multi_recv(subreq, &ret, state, &err_list,
2671                                                 NULL);
2672         TALLOC_FREE(subreq);
2673         if (! status) {
2674                 int ret2;
2675                 uint32_t pnn;
2676
2677                 ret2 = ctdb_client_control_multi_error(state->pnn_list,
2678                                                        state->count,
2679                                                        err_list, &pnn);
2680                 if (ret2 != 0) {
2681                         D_ERR("failed to run recovered event on node %u,"
2682                               " ret=%d\n", pnn, ret2);
2683                 } else {
2684                         D_ERR("failed to run recovered event, ret=%d\n", ret);
2685                 }
2686                 tevent_req_error(req, ret);
2687                 return;
2688         }
2689
2690         D_ERR("recovered event finished\n");
2691
2692         tevent_req_done(req);
2693 }
2694
2695 static void recovery_recv(struct tevent_req *req, int *perr)
2696 {
2697         generic_recv(req, perr);
2698 }
2699
2700 static void usage(const char *progname)
2701 {
2702         fprintf(stderr, "\nUsage: %s <output-fd> <ctdb-socket-path> <generation>\n",
2703                 progname);
2704 }
2705
2706
2707 /*
2708  * Arguments - log fd, write fd, socket path, generation
2709  */
2710 int main(int argc, char *argv[])
2711 {
2712         int write_fd;
2713         const char *sockpath;
2714         TALLOC_CTX *mem_ctx;
2715         struct tevent_context *ev;
2716         struct ctdb_client_context *client;
2717         int ret;
2718         struct tevent_req *req;
2719         uint32_t generation;
2720
2721         if (argc != 4) {
2722                 usage(argv[0]);
2723                 exit(1);
2724         }
2725
2726         write_fd = atoi(argv[1]);
2727         sockpath = argv[2];
2728         generation = (uint32_t)strtoul(argv[3], NULL, 0);
2729
2730         mem_ctx = talloc_new(NULL);
2731         if (mem_ctx == NULL) {
2732                 fprintf(stderr, "recovery: talloc_new() failed\n");
2733                 goto failed;
2734         }
2735
2736         ret = logging_init(mem_ctx, NULL, NULL, "ctdb-recovery");
2737         if (ret != 0) {
2738                 fprintf(stderr, "recovery: Unable to initialize logging\n");
2739                 goto failed;
2740         }
2741
2742         ev = tevent_context_init(mem_ctx);
2743         if (ev == NULL) {
2744                 D_ERR("tevent_context_init() failed\n");
2745                 goto failed;
2746         }
2747
2748         ret = ctdb_client_init(mem_ctx, ev, sockpath, &client);
2749         if (ret != 0) {
2750                 D_ERR("ctdb_client_init() failed, ret=%d\n", ret);
2751                 goto failed;
2752         }
2753
2754         req = recovery_send(mem_ctx, ev, client, generation);
2755         if (req == NULL) {
2756                 D_ERR("database_recover_send() failed\n");
2757                 goto failed;
2758         }
2759
2760         if (! tevent_req_poll(req, ev)) {
2761                 D_ERR("tevent_req_poll() failed\n");
2762                 goto failed;
2763         }
2764
2765         recovery_recv(req, &ret);
2766         TALLOC_FREE(req);
2767         if (ret != 0) {
2768                 D_ERR("database recovery failed, ret=%d\n", ret);
2769                 goto failed;
2770         }
2771
2772         sys_write(write_fd, &ret, sizeof(ret));
2773         return 0;
2774
2775 failed:
2776         TALLOC_FREE(mem_ctx);
2777         return 1;
2778 }