git.ozlabs.org Git - ccan/blob - ccan/tdb2/transaction.c

   1  /*
   2    Unix SMB/CIFS implementation.
   3
   4    trivial database library
   5
   6    Copyright (C) Andrew Tridgell              2005
   7    Copyright (C) Rusty Russell                2010
   8
   9      ** NOTE! The following LGPL license applies to the tdb
  10      ** library. This does NOT imply that all of Samba is released
  11      ** under the LGPL
  12
  13    This library is free software; you can redistribute it and/or
  14    modify it under the terms of the GNU Lesser General Public
  15    License as published by the Free Software Foundation; either
  16    version 3 of the License, or (at your option) any later version.
  17
  18    This library is distributed in the hope that it will be useful,
  19    but WITHOUT ANY WARRANTY; without even the implied warranty of
  20    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  21    Lesser General Public License for more details.
  22
  23    You should have received a copy of the GNU Lesser General Public
  24    License along with this library; if not, see <http://www.gnu.org/licenses/>.
  25 */
  26
  27 #include "private.h"
  28 #define SAFE_FREE(x) do { if ((x) != NULL) {free(x); (x)=NULL;} } while(0)
  29
  30 /*
  31   transaction design:
  32
  33   - only allow a single transaction at a time per database. This makes
  34     using the transaction API simpler, as otherwise the caller would
  35     have to cope with temporary failures in transactions that conflict
  36     with other current transactions
  37
  38   - keep the transaction recovery information in the same file as the
  39     database, using a special 'transaction recovery' record pointed at
  40     by the header. This removes the need for extra journal files as
  41     used by some other databases
  42
  43   - dynamically allocated the transaction recover record, re-using it
  44     for subsequent transactions. If a larger record is needed then
  45     tdb_free() the old record to place it on the normal tdb freelist
  46     before allocating the new record
  47
  48   - during transactions, keep a linked list of writes all that have
  49     been performed by intercepting all tdb_write() calls. The hooked
  50     transaction versions of tdb_read() and tdb_write() check this
  51     linked list and try to use the elements of the list in preference
  52     to the real database.
  53
  54   - don't allow any locks to be held when a transaction starts,
  55     otherwise we can end up with deadlock (plus lack of lock nesting
  56     in POSIX locks would mean the lock is lost)
  57
  58   - if the caller gains a lock during the transaction but doesn't
  59     release it then fail the commit
  60
  61   - allow for nested calls to tdb_transaction_start(), re-using the
  62     existing transaction record. If the inner transaction is canceled
  63     then a subsequent commit will fail
  64
  65   - keep a mirrored copy of the tdb hash chain heads to allow for the
  66     fast hash heads scan on traverse, updating the mirrored copy in
  67     the transaction version of tdb_write
  68
  69   - allow callers to mix transaction and non-transaction use of tdb,
  70     although once a transaction is started then an exclusive lock is
  71     gained until the transaction is committed or canceled
  72
  73   - the commit stategy involves first saving away all modified data
  74     into a linearised buffer in the transaction recovery area, then
  75     marking the transaction recovery area with a magic value to
  76     indicate a valid recovery record. In total 4 fsync/msync calls are
  77     needed per commit to prevent race conditions. It might be possible
  78     to reduce this to 3 or even 2 with some more work.
  79
  80   - check for a valid recovery record on open of the tdb, while the
  81     open lock is held. Automatically recover from the transaction
  82     recovery area if needed, then continue with the open as
  83     usual. This allows for smooth crash recovery with no administrator
  84     intervention.
  85
  86   - if TDB_NOSYNC is passed to flags in tdb_open then transactions are
  87     still available, but no transaction recovery area is used and no
  88     fsync/msync calls are made.
  89 */
  90
  91 /*
  92   hold the context of any current transaction
  93 */
  94 struct tdb_transaction {
  95         /* the original io methods - used to do IOs to the real db */
  96         const struct tdb_methods *io_methods;
  97
  98         /* the list of transaction blocks. When a block is first
  99            written to, it gets created in this list */
 100         uint8_t **blocks;
 101         size_t num_blocks;
 102         size_t last_block_size; /* number of valid bytes in the last block */
 103
 104         /* non-zero when an internal transaction error has
 105            occurred. All write operations will then fail until the
 106            transaction is ended */
 107         int transaction_error;
 108
 109         /* when inside a transaction we need to keep track of any
 110            nested tdb_transaction_start() calls, as these are allowed,
 111            but don't create a new transaction */
 112         unsigned int nesting;
 113
 114         /* set when a prepare has already occurred */
 115         bool prepared;
 116         tdb_off_t magic_offset;
 117
 118         /* old file size before transaction */
 119         tdb_len_t old_map_size;
 120 };
 121
 122 /* This doesn't really need to be pagesize, but we use it for similar reasons. */
 123 #define PAGESIZE 65536
 124
 125 /*
 126   read while in a transaction. We need to check first if the data is in our list
 127   of transaction elements, then if not do a real read
 128 */
 129 static enum TDB_ERROR transaction_read(struct tdb_context *tdb, tdb_off_t off,
 130                                        void *buf, tdb_len_t len)
 131 {
 132         size_t blk;
 133         enum TDB_ERROR ecode;
 134
 135         /* break it down into block sized ops */
 136         while (len + (off % PAGESIZE) > PAGESIZE) {
 137                 tdb_len_t len2 = PAGESIZE - (off % PAGESIZE);
 138                 ecode = transaction_read(tdb, off, buf, len2);
 139                 if (ecode != TDB_SUCCESS) {
 140                         return ecode;
 141                 }
 142                 len -= len2;
 143                 off += len2;
 144                 buf = (void *)(len2 + (char *)buf);
 145         }
 146
 147         if (len == 0) {
 148                 return TDB_SUCCESS;
 149         }
 150
 151         blk = off / PAGESIZE;
 152
 153         /* see if we have it in the block list */
 154         if (tdb->transaction->num_blocks <= blk ||
 155             tdb->transaction->blocks[blk] == NULL) {
 156                 /* nope, do a real read */
 157                 ecode = tdb->transaction->io_methods->tread(tdb, off, buf, len);
 158                 if (ecode != TDB_SUCCESS) {
 159                         goto fail;
 160                 }
 161                 return 0;
 162         }
 163
 164         /* it is in the block list. Now check for the last block */
 165         if (blk == tdb->transaction->num_blocks-1) {
 166                 if (len > tdb->transaction->last_block_size) {
 167                         ecode = TDB_ERR_IO;
 168                         goto fail;
 169                 }
 170         }
 171
 172         /* now copy it out of this block */
 173         memcpy(buf, tdb->transaction->blocks[blk] + (off % PAGESIZE), len);
 174         return TDB_SUCCESS;
 175
 176 fail:
 177         tdb->transaction->transaction_error = 1;
 178         return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 179                           "transaction_read: failed at off=%zu len=%zu",
 180                           (size_t)off, (size_t)len);
 181 }
 182
 183
 184 /*
 185   write while in a transaction
 186 */
 187 static enum TDB_ERROR transaction_write(struct tdb_context *tdb, tdb_off_t off,
 188                                         const void *buf, tdb_len_t len)
 189 {
 190         size_t blk;
 191         enum TDB_ERROR ecode;
 192
 193         /* Only a commit is allowed on a prepared transaction */
 194         if (tdb->transaction->prepared) {
 195                 ecode = tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_ERROR,
 196                                    "transaction_write: transaction already"
 197                                    " prepared, write not allowed");
 198                 goto fail;
 199         }
 200
 201         /* break it up into block sized chunks */
 202         while (len + (off % PAGESIZE) > PAGESIZE) {
 203                 tdb_len_t len2 = PAGESIZE - (off % PAGESIZE);
 204                 ecode = transaction_write(tdb, off, buf, len2);
 205                 if (ecode != TDB_SUCCESS) {
 206                         return -1;
 207                 }
 208                 len -= len2;
 209                 off += len2;
 210                 if (buf != NULL) {
 211                         buf = (const void *)(len2 + (const char *)buf);
 212                 }
 213         }
 214
 215         if (len == 0) {
 216                 return TDB_SUCCESS;
 217         }
 218
 219         blk = off / PAGESIZE;
 220         off = off % PAGESIZE;
 221
 222         if (tdb->transaction->num_blocks <= blk) {
 223                 uint8_t **new_blocks;
 224                 /* expand the blocks array */
 225                 if (tdb->transaction->blocks == NULL) {
 226                         new_blocks = (uint8_t **)malloc(
 227                                 (blk+1)*sizeof(uint8_t *));
 228                 } else {
 229                         new_blocks = (uint8_t **)realloc(
 230                                 tdb->transaction->blocks,
 231                                 (blk+1)*sizeof(uint8_t *));
 232                 }
 233                 if (new_blocks == NULL) {
 234                         ecode = tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR,
 235                                            "transaction_write:"
 236                                            " failed to allocate");
 237                         goto fail;
 238                 }
 239                 memset(&new_blocks[tdb->transaction->num_blocks], 0,
 240                        (1+(blk - tdb->transaction->num_blocks))*sizeof(uint8_t *));
 241                 tdb->transaction->blocks = new_blocks;
 242                 tdb->transaction->num_blocks = blk+1;
 243                 tdb->transaction->last_block_size = 0;
 244         }
 245
 246         /* allocate and fill a block? */
 247         if (tdb->transaction->blocks[blk] == NULL) {
 248                 tdb->transaction->blocks[blk] = (uint8_t *)calloc(PAGESIZE, 1);
 249                 if (tdb->transaction->blocks[blk] == NULL) {
 250                         ecode = tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR,
 251                                            "transaction_write:"
 252                                            " failed to allocate");
 253                         goto fail;
 254                 }
 255                 if (tdb->transaction->old_map_size > blk * PAGESIZE) {
 256                         tdb_len_t len2 = PAGESIZE;
 257                         if (len2 + (blk * PAGESIZE) > tdb->transaction->old_map_size) {
 258                                 len2 = tdb->transaction->old_map_size - (blk * PAGESIZE);
 259                         }
 260                         ecode = tdb->transaction->io_methods->tread(tdb,
 261                                         blk * PAGESIZE,
 262                                         tdb->transaction->blocks[blk],
 263                                         len2);
 264                         if (ecode != TDB_SUCCESS) {
 265                                 ecode = tdb_logerr(tdb, ecode,
 266                                                    TDB_LOG_ERROR,
 267                                                    "transaction_write:"
 268                                                    " failed to"
 269                                                    " read old block: %s",
 270                                                    strerror(errno));
 271                                 SAFE_FREE(tdb->transaction->blocks[blk]);
 272                                 goto fail;
 273                         }
 274                         if (blk == tdb->transaction->num_blocks-1) {
 275                                 tdb->transaction->last_block_size = len2;
 276                         }
 277                 }
 278         }
 279
 280         /* overwrite part of an existing block */
 281         if (buf == NULL) {
 282                 memset(tdb->transaction->blocks[blk] + off, 0, len);
 283         } else {
 284                 memcpy(tdb->transaction->blocks[blk] + off, buf, len);
 285         }
 286         if (blk == tdb->transaction->num_blocks-1) {
 287                 if (len + off > tdb->transaction->last_block_size) {
 288                         tdb->transaction->last_block_size = len + off;
 289                 }
 290         }
 291
 292         return TDB_SUCCESS;
 293
 294 fail:
 295         tdb->transaction->transaction_error = 1;
 296         return ecode;
 297 }
 298
 299
 300 /*
 301   write while in a transaction - this variant never expands the transaction blocks, it only
 302   updates existing blocks. This means it cannot change the recovery size
 303 */
 304 static void transaction_write_existing(struct tdb_context *tdb, tdb_off_t off,
 305                                        const void *buf, tdb_len_t len)
 306 {
 307         size_t blk;
 308
 309         /* break it up into block sized chunks */
 310         while (len + (off % PAGESIZE) > PAGESIZE) {
 311                 tdb_len_t len2 = PAGESIZE - (off % PAGESIZE);
 312                 transaction_write_existing(tdb, off, buf, len2);
 313                 len -= len2;
 314                 off += len2;
 315                 if (buf != NULL) {
 316                         buf = (const void *)(len2 + (const char *)buf);
 317                 }
 318         }
 319
 320         if (len == 0) {
 321                 return;
 322         }
 323
 324         blk = off / PAGESIZE;
 325         off = off % PAGESIZE;
 326
 327         if (tdb->transaction->num_blocks <= blk ||
 328             tdb->transaction->blocks[blk] == NULL) {
 329                 return;
 330         }
 331
 332         if (blk == tdb->transaction->num_blocks-1 &&
 333             off + len > tdb->transaction->last_block_size) {
 334                 if (off >= tdb->transaction->last_block_size) {
 335                         return;
 336                 }
 337                 len = tdb->transaction->last_block_size - off;
 338         }
 339
 340         /* overwrite part of an existing block */
 341         memcpy(tdb->transaction->blocks[blk] + off, buf, len);
 342 }
 343
 344
 345 /*
 346   out of bounds check during a transaction
 347 */
 348 static enum TDB_ERROR transaction_oob(struct tdb_context *tdb, tdb_off_t len,
 349                                       bool probe)
 350 {
 351         if (len <= tdb->file->map_size) {
 352                 return TDB_SUCCESS;
 353         }
 354         if (!probe) {
 355                 tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR,
 356                            "tdb_oob len %lld beyond transaction size %lld",
 357                            (long long)len,
 358                            (long long)tdb->file->map_size);
 359         }
 360         return TDB_ERR_IO;
 361 }
 362
 363 /*
 364   transaction version of tdb_expand().
 365 */
 366 static enum TDB_ERROR transaction_expand_file(struct tdb_context *tdb,
 367                                               tdb_off_t addition)
 368 {
 369         enum TDB_ERROR ecode;
 370
 371         /* add a write to the transaction elements, so subsequent
 372            reads see the zero data */
 373         ecode = transaction_write(tdb, tdb->file->map_size, NULL, addition);
 374         if (ecode == TDB_SUCCESS) {
 375                 tdb->file->map_size += addition;
 376         }
 377         return ecode;
 378 }
 379
 380 static void *transaction_direct(struct tdb_context *tdb, tdb_off_t off,
 381                                 size_t len, bool write_mode)
 382 {
 383         size_t blk = off / PAGESIZE, end_blk;
 384
 385         /* This is wrong for zero-length blocks, but will fail gracefully */
 386         end_blk = (off + len - 1) / PAGESIZE;
 387
 388         /* Can only do direct if in single block and we've already copied. */
 389         if (write_mode) {
 390                 if (blk != end_blk)
 391                         return NULL;
 392                 if (blk >= tdb->transaction->num_blocks)
 393                         return NULL;
 394                 if (tdb->transaction->blocks[blk] == NULL)
 395                         return NULL;
 396                 return tdb->transaction->blocks[blk] + off % PAGESIZE;
 397         }
 398
 399         /* Single which we have copied? */
 400         if (blk == end_blk
 401             && blk < tdb->transaction->num_blocks
 402             && tdb->transaction->blocks[blk])
 403                 return tdb->transaction->blocks[blk] + off % PAGESIZE;
 404
 405         /* Otherwise must be all not copied. */
 406         while (blk <= end_blk) {
 407                 if (blk >= tdb->transaction->num_blocks)
 408                         break;
 409                 if (tdb->transaction->blocks[blk])
 410                         return NULL;
 411                 blk++;
 412         }
 413         return tdb->transaction->io_methods->direct(tdb, off, len, false);
 414 }
 415
 416 static const struct tdb_methods transaction_methods = {
 417         transaction_read,
 418         transaction_write,
 419         transaction_oob,
 420         transaction_expand_file,
 421         transaction_direct,
 422 };
 423
 424 /*
 425   sync to disk
 426 */
 427 static enum TDB_ERROR transaction_sync(struct tdb_context *tdb,
 428                                        tdb_off_t offset, tdb_len_t length)
 429 {
 430         if (tdb->flags & TDB_NOSYNC) {
 431                 return TDB_SUCCESS;
 432         }
 433
 434         if (fsync(tdb->file->fd) != 0) {
 435                 return tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR,
 436                                   "tdb_transaction: fsync failed: %s",
 437                                   strerror(errno));
 438         }
 439 #ifdef MS_SYNC
 440         if (tdb->file->map_ptr) {
 441                 tdb_off_t moffset = offset & ~(PAGESIZE-1);
 442                 if (msync(moffset + (char *)tdb->file->map_ptr,
 443                           length + (offset - moffset), MS_SYNC) != 0) {
 444                         return tdb_logerr(tdb, TDB_ERR_IO, TDB_LOG_ERROR,
 445                                           "tdb_transaction: msync failed: %s",
 446                                           strerror(errno));
 447                 }
 448         }
 449 #endif
 450         return TDB_SUCCESS;
 451 }
 452
 453
 454 static void _tdb_transaction_cancel(struct tdb_context *tdb)
 455 {
 456         int i;
 457         enum TDB_ERROR ecode;
 458
 459         if (tdb->transaction == NULL) {
 460                 tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
 461                            "tdb_transaction_cancel: no transaction");
 462                 return;
 463         }
 464
 465         if (tdb->transaction->nesting != 0) {
 466                 tdb->transaction->transaction_error = 1;
 467                 tdb->transaction->nesting--;
 468                 return;
 469         }
 470
 471         tdb->file->map_size = tdb->transaction->old_map_size;
 472
 473         /* free all the transaction blocks */
 474         for (i=0;i<tdb->transaction->num_blocks;i++) {
 475                 if (tdb->transaction->blocks[i] != NULL) {
 476                         free(tdb->transaction->blocks[i]);
 477                 }
 478         }
 479         SAFE_FREE(tdb->transaction->blocks);
 480
 481         if (tdb->transaction->magic_offset) {
 482                 const struct tdb_methods *methods = tdb->transaction->io_methods;
 483                 uint64_t invalid = TDB_RECOVERY_INVALID_MAGIC;
 484
 485                 /* remove the recovery marker */
 486                 ecode = methods->twrite(tdb, tdb->transaction->magic_offset,
 487                                         &invalid, sizeof(invalid));
 488                 if (ecode == TDB_SUCCESS)
 489                         ecode = transaction_sync(tdb,
 490                                                  tdb->transaction->magic_offset,
 491                                                  sizeof(invalid));
 492                 if (ecode != TDB_SUCCESS) {
 493                         tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 494                                    "tdb_transaction_cancel: failed to remove"
 495                                    " recovery magic");
 496                 }
 497         }
 498
 499         if (tdb->file->allrecord_lock.count)
 500                 tdb_allrecord_unlock(tdb, tdb->file->allrecord_lock.ltype);
 501
 502         /* restore the normal io methods */
 503         tdb->methods = tdb->transaction->io_methods;
 504
 505         tdb_transaction_unlock(tdb, F_WRLCK);
 506
 507         if (tdb_has_open_lock(tdb))
 508                 tdb_unlock_open(tdb);
 509
 510         SAFE_FREE(tdb->transaction);
 511 }
 512
 513 /*
 514   start a tdb transaction. No token is returned, as only a single
 515   transaction is allowed to be pending per tdb_context
 516 */
 517 enum TDB_ERROR tdb_transaction_start(struct tdb_context *tdb)
 518 {
 519         enum TDB_ERROR ecode;
 520
 521         /* some sanity checks */
 522         if (tdb->read_only || (tdb->flags & TDB_INTERNAL)) {
 523                 return tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
 524                                                     TDB_LOG_USE_ERROR,
 525                                                     "tdb_transaction_start:"
 526                                                     " cannot start a"
 527                                                     " transaction on a "
 528                                                     "read-only or internal db");
 529         }
 530
 531         /* cope with nested tdb_transaction_start() calls */
 532         if (tdb->transaction != NULL) {
 533                 if (!(tdb->flags & TDB_ALLOW_NESTING)) {
 534                         return tdb->last_error
 535                                 = tdb_logerr(tdb, TDB_ERR_IO,
 536                                              TDB_LOG_USE_ERROR,
 537                                              "tdb_transaction_start:"
 538                                              " already inside transaction");
 539                 }
 540                 tdb->transaction->nesting++;
 541                 return 0;
 542         }
 543
 544         if (tdb_has_hash_locks(tdb)) {
 545                 /* the caller must not have any locks when starting a
 546                    transaction as otherwise we'll be screwed by lack
 547                    of nested locks in POSIX */
 548                 return tdb->last_error = tdb_logerr(tdb, TDB_ERR_LOCK,
 549                                                     TDB_LOG_USE_ERROR,
 550                                                     "tdb_transaction_start:"
 551                                                     " cannot start a"
 552                                                     " transaction with locks"
 553                                                     " held");
 554         }
 555
 556         tdb->transaction = (struct tdb_transaction *)
 557                 calloc(sizeof(struct tdb_transaction), 1);
 558         if (tdb->transaction == NULL) {
 559                 return tdb->last_error = tdb_logerr(tdb, TDB_ERR_OOM,
 560                                                     TDB_LOG_ERROR,
 561                                                     "tdb_transaction_start:"
 562                                                     " cannot allocate");
 563         }
 564
 565         /* get the transaction write lock. This is a blocking lock. As
 566            discussed with Volker, there are a number of ways we could
 567            make this async, which we will probably do in the future */
 568         ecode = tdb_transaction_lock(tdb, F_WRLCK);
 569         if (ecode != TDB_SUCCESS) {
 570                 SAFE_FREE(tdb->transaction->blocks);
 571                 SAFE_FREE(tdb->transaction);
 572                 return tdb->last_error = ecode;
 573         }
 574
 575         /* get a read lock over entire file. This is upgraded to a write
 576            lock during the commit */
 577         ecode = tdb_allrecord_lock(tdb, F_RDLCK, TDB_LOCK_WAIT, true);
 578         if (ecode != TDB_SUCCESS) {
 579                 goto fail_allrecord_lock;
 580         }
 581
 582         /* make sure we know about any file expansions already done by
 583            anyone else */
 584         tdb->methods->oob(tdb, tdb->file->map_size + 1, true);
 585         tdb->transaction->old_map_size = tdb->file->map_size;
 586
 587         /* finally hook the io methods, replacing them with
 588            transaction specific methods */
 589         tdb->transaction->io_methods = tdb->methods;
 590         tdb->methods = &transaction_methods;
 591         return tdb->last_error = TDB_SUCCESS;
 592
 593 fail_allrecord_lock:
 594         tdb_transaction_unlock(tdb, F_WRLCK);
 595         SAFE_FREE(tdb->transaction->blocks);
 596         SAFE_FREE(tdb->transaction);
 597         return tdb->last_error = ecode;
 598 }
 599
 600
 601 /*
 602   cancel the current transaction
 603 */
 604 void tdb_transaction_cancel(struct tdb_context *tdb)
 605 {
 606         _tdb_transaction_cancel(tdb);
 607 }
 608
 609 /*
 610   work out how much space the linearised recovery data will consume (worst case)
 611 */
 612 static tdb_len_t tdb_recovery_size(struct tdb_context *tdb)
 613 {
 614         tdb_len_t recovery_size = 0;
 615         int i;
 616
 617         recovery_size = 0;
 618         for (i=0;i<tdb->transaction->num_blocks;i++) {
 619                 if (i * PAGESIZE >= tdb->transaction->old_map_size) {
 620                         break;
 621                 }
 622                 if (tdb->transaction->blocks[i] == NULL) {
 623                         continue;
 624                 }
 625                 recovery_size += 2*sizeof(tdb_off_t);
 626                 if (i == tdb->transaction->num_blocks-1) {
 627                         recovery_size += tdb->transaction->last_block_size;
 628                 } else {
 629                         recovery_size += PAGESIZE;
 630                 }
 631         }
 632
 633         return recovery_size;
 634 }
 635
 636 static enum TDB_ERROR tdb_recovery_area(struct tdb_context *tdb,
 637                                         const struct tdb_methods *methods,
 638                                         tdb_off_t *recovery_offset,
 639                                         struct tdb_recovery_record *rec)
 640 {
 641         enum TDB_ERROR ecode;
 642
 643         *recovery_offset = tdb_read_off(tdb,
 644                                         offsetof(struct tdb_header, recovery));
 645         if (TDB_OFF_IS_ERR(*recovery_offset)) {
 646                 return *recovery_offset;
 647         }
 648
 649         if (*recovery_offset == 0) {
 650                 rec->max_len = 0;
 651                 return TDB_SUCCESS;
 652         }
 653
 654         ecode = methods->tread(tdb, *recovery_offset, rec, sizeof(*rec));
 655         if (ecode != TDB_SUCCESS)
 656                 return ecode;
 657
 658         tdb_convert(tdb, rec, sizeof(*rec));
 659         /* ignore invalid recovery regions: can happen in crash */
 660         if (rec->magic != TDB_RECOVERY_MAGIC &&
 661             rec->magic != TDB_RECOVERY_INVALID_MAGIC) {
 662                 *recovery_offset = 0;
 663                 rec->max_len = 0;
 664         }
 665         return TDB_SUCCESS;
 666 }
 667
 668 static unsigned int same(const unsigned char *new,
 669                          const unsigned char *old,
 670                          unsigned int length)
 671 {
 672         unsigned int i;
 673
 674         for (i = 0; i < length; i++) {
 675                 if (new[i] != old[i])
 676                         break;
 677         }
 678         return i;
 679 }
 680
 681 static unsigned int different(const unsigned char *new,
 682                               const unsigned char *old,
 683                               unsigned int length,
 684                               unsigned int min_same,
 685                               unsigned int *samelen)
 686 {
 687         unsigned int i;
 688
 689         *samelen = 0;
 690         for (i = 0; i < length; i++) {
 691                 if (new[i] == old[i]) {
 692                         (*samelen)++;
 693                 } else {
 694                         if (*samelen >= min_same) {
 695                                 return i - *samelen;
 696                         }
 697                         *samelen = 0;
 698                 }
 699         }
 700
 701         if (*samelen < min_same)
 702                 *samelen = 0;
 703         return length - *samelen;
 704 }
 705
 706 /* Allocates recovery blob, without tdb_recovery_record at head set up. */
 707 static struct tdb_recovery_record *alloc_recovery(struct tdb_context *tdb,
 708                                                   tdb_len_t *len)
 709 {
 710         struct tdb_recovery_record *rec;
 711         size_t i;
 712         enum TDB_ERROR ecode;
 713         unsigned char *p;
 714         const struct tdb_methods *old_methods = tdb->methods;
 715
 716         rec = malloc(sizeof(*rec) + tdb_recovery_size(tdb));
 717         if (!rec) {
 718                 tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR,
 719                            "transaction_setup_recovery:"
 720                            " cannot allocate");
 721                 return TDB_ERR_PTR(TDB_ERR_OOM);
 722         }
 723
 724         /* We temporarily revert to the old I/O methods, so we can use
 725          * tdb_access_read */
 726         tdb->methods = tdb->transaction->io_methods;
 727
 728         /* build the recovery data into a single blob to allow us to do a single
 729            large write, which should be more efficient */
 730         p = (unsigned char *)(rec + 1);
 731         for (i=0;i<tdb->transaction->num_blocks;i++) {
 732                 tdb_off_t offset;
 733                 tdb_len_t length;
 734                 unsigned int off;
 735                 const unsigned char *buffer;
 736
 737                 if (tdb->transaction->blocks[i] == NULL) {
 738                         continue;
 739                 }
 740
 741                 offset = i * PAGESIZE;
 742                 length = PAGESIZE;
 743                 if (i == tdb->transaction->num_blocks-1) {
 744                         length = tdb->transaction->last_block_size;
 745                 }
 746
 747                 if (offset >= tdb->transaction->old_map_size) {
 748                         continue;
 749                 }
 750
 751                 if (offset + length > tdb->file->map_size) {
 752                         ecode = tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR,
 753                                            "tdb_transaction_setup_recovery:"
 754                                            " transaction data over new region"
 755                                            " boundary");
 756                         goto fail;
 757                 }
 758                 if (offset + length > tdb->transaction->old_map_size) {
 759                         /* Short read at EOF. */
 760                         length = tdb->transaction->old_map_size - offset;
 761                 }
 762                 buffer = tdb_access_read(tdb, offset, length, false);
 763                 if (TDB_PTR_IS_ERR(buffer)) {
 764                         ecode = TDB_PTR_ERR(buffer);
 765                         goto fail;
 766                 }
 767
 768                 /* Skip over anything the same at the start. */
 769                 off = same(tdb->transaction->blocks[i], buffer, length);
 770                 offset += off;
 771
 772                 while (off < length) {
 773                         tdb_len_t len;
 774                         unsigned int samelen;
 775
 776                         len = different(tdb->transaction->blocks[i] + off,
 777                                         buffer + off, length - off,
 778                                         sizeof(offset) + sizeof(len) + 1,
 779                                         &samelen);
 780
 781                         memcpy(p, &offset, sizeof(offset));
 782                         memcpy(p + sizeof(offset), &len, sizeof(len));
 783                         tdb_convert(tdb, p, sizeof(offset) + sizeof(len));
 784                         p += sizeof(offset) + sizeof(len);
 785                         memcpy(p, buffer + off, len);
 786                         p += len;
 787                         off += len + samelen;
 788                         offset += len + samelen;
 789                 }
 790                 tdb_access_release(tdb, buffer);
 791         }
 792
 793         *len = p - (unsigned char *)(rec + 1);
 794         tdb->methods = old_methods;
 795         return rec;
 796
 797 fail:
 798         free(rec);
 799         tdb->methods = old_methods;
 800         return TDB_ERR_PTR(ecode);
 801 }
 802
 803 static tdb_off_t create_recovery_area(struct tdb_context *tdb,
 804                                       tdb_len_t rec_length,
 805                                       struct tdb_recovery_record *rec)
 806 {
 807         tdb_off_t off, recovery_off;
 808         tdb_len_t addition;
 809         enum TDB_ERROR ecode;
 810         const struct tdb_methods *methods = tdb->transaction->io_methods;
 811
 812         /* round up to a multiple of page size. Overallocate, since each
 813          * such allocation forces us to expand the file. */
 814         rec->max_len
 815                 = (((sizeof(*rec) + rec_length + rec_length / 2)
 816                     + PAGESIZE-1) & ~(PAGESIZE-1))
 817                 - sizeof(*rec);
 818         off = tdb->file->map_size;
 819
 820         /* Restore ->map_size before calling underlying expand_file.
 821            Also so that we don't try to expand the file again in the
 822            transaction commit, which would destroy the recovery
 823            area */
 824         addition = (tdb->file->map_size - tdb->transaction->old_map_size) +
 825                 sizeof(*rec) + rec->max_len;
 826         tdb->file->map_size = tdb->transaction->old_map_size;
 827         ecode = methods->expand_file(tdb, addition);
 828         if (ecode != TDB_SUCCESS) {
 829                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 830                                   "tdb_recovery_allocate:"
 831                                   " failed to create recovery area");
 832         }
 833
 834         /* we have to reset the old map size so that we don't try to
 835            expand the file again in the transaction commit, which
 836            would destroy the recovery area */
 837         tdb->transaction->old_map_size = tdb->file->map_size;
 838
 839         /* write the recovery header offset and sync - we can sync without a race here
 840            as the magic ptr in the recovery record has not been set */
 841         recovery_off = off;
 842         tdb_convert(tdb, &recovery_off, sizeof(recovery_off));
 843         ecode = methods->twrite(tdb, offsetof(struct tdb_header, recovery),
 844                                 &recovery_off, sizeof(tdb_off_t));
 845         if (ecode != TDB_SUCCESS) {
 846                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 847                                   "tdb_recovery_allocate:"
 848                                   " failed to write recovery head");
 849         }
 850         transaction_write_existing(tdb, offsetof(struct tdb_header, recovery),
 851                                    &recovery_off,
 852                                    sizeof(tdb_off_t));
 853         return off;
 854 }
 855
 856 /*
 857   setup the recovery data that will be used on a crash during commit
 858 */
 859 static enum TDB_ERROR transaction_setup_recovery(struct tdb_context *tdb)
 860 {
 861         tdb_len_t recovery_size = 0;
 862         tdb_off_t recovery_off = 0;
 863         tdb_off_t old_map_size = tdb->transaction->old_map_size;
 864         struct tdb_recovery_record *recovery;
 865         const struct tdb_methods *methods = tdb->transaction->io_methods;
 866         uint64_t magic;
 867         enum TDB_ERROR ecode;
 868
 869         recovery = alloc_recovery(tdb, &recovery_size);
 870         if (TDB_PTR_IS_ERR(recovery))
 871                 return TDB_PTR_ERR(recovery);
 872
 873         ecode = tdb_recovery_area(tdb, methods, &recovery_off, recovery);
 874         if (ecode) {
 875                 free(recovery);
 876                 return ecode;
 877         }
 878
 879         if (recovery->max_len < recovery_size) {
 880                 /* Not large enough. Free up old recovery area. */
 881                 if (recovery_off) {
 882                         tdb->stats.frees++;
 883                         ecode = add_free_record(tdb, recovery_off,
 884                                                 sizeof(*recovery)
 885                                                 + recovery->max_len,
 886                                                 TDB_LOCK_WAIT, true);
 887                         free(recovery);
 888                         if (ecode != TDB_SUCCESS) {
 889                                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 890                                                   "tdb_recovery_allocate:"
 891                                                   " failed to free previous"
 892                                                   " recovery area");
 893                         }
 894
 895                         /* Refresh recovery after add_free_record above. */
 896                         recovery = alloc_recovery(tdb, &recovery_size);
 897                         if (TDB_PTR_IS_ERR(recovery))
 898                                 return TDB_PTR_ERR(recovery);
 899                 }
 900
 901                 recovery_off = create_recovery_area(tdb, recovery_size,
 902                                                     recovery);
 903                 if (TDB_OFF_IS_ERR(recovery_off)) {
 904                         free(recovery);
 905                         return recovery_off;
 906                 }
 907         }
 908
 909         /* Now we know size, convert rec header. */
 910         recovery->magic = TDB_RECOVERY_INVALID_MAGIC;
 911         recovery->len = recovery_size;
 912         recovery->eof = old_map_size;
 913         tdb_convert(tdb, recovery, sizeof(*recovery));
 914
 915         /* write the recovery data to the recovery area */
 916         ecode = methods->twrite(tdb, recovery_off, recovery, recovery_size);
 917         if (ecode != TDB_SUCCESS) {
 918                 free(recovery);
 919                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 920                                   "tdb_transaction_setup_recovery:"
 921                                   " failed to write recovery data");
 922         }
 923         transaction_write_existing(tdb, recovery_off, recovery, recovery_size);
 924
 925         free(recovery);
 926
 927         /* as we don't have ordered writes, we have to sync the recovery
 928            data before we update the magic to indicate that the recovery
 929            data is present */
 930         ecode = transaction_sync(tdb, recovery_off, recovery_size);
 931         if (ecode != TDB_SUCCESS)
 932                 return ecode;
 933
 934         magic = TDB_RECOVERY_MAGIC;
 935         tdb_convert(tdb, &magic, sizeof(magic));
 936
 937         tdb->transaction->magic_offset
 938                 = recovery_off + offsetof(struct tdb_recovery_record, magic);
 939
 940         ecode = methods->twrite(tdb, tdb->transaction->magic_offset,
 941                                 &magic, sizeof(magic));
 942         if (ecode != TDB_SUCCESS) {
 943                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
 944                                   "tdb_transaction_setup_recovery:"
 945                                   " failed to write recovery magic");
 946         }
 947         transaction_write_existing(tdb, tdb->transaction->magic_offset,
 948                                    &magic, sizeof(magic));
 949
 950         /* ensure the recovery magic marker is on disk */
 951         return transaction_sync(tdb, tdb->transaction->magic_offset,
 952                                 sizeof(magic));
 953 }
 954
 955 static enum TDB_ERROR _tdb_transaction_prepare_commit(struct tdb_context *tdb)
 956 {
 957         const struct tdb_methods *methods;
 958         enum TDB_ERROR ecode;
 959
 960         if (tdb->transaction == NULL) {
 961                 return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
 962                                   "tdb_transaction_prepare_commit:"
 963                                   " no transaction");
 964         }
 965
 966         if (tdb->transaction->prepared) {
 967                 _tdb_transaction_cancel(tdb);
 968                 return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_USE_ERROR,
 969                                   "tdb_transaction_prepare_commit:"
 970                                   " transaction already prepared");
 971         }
 972
 973         if (tdb->transaction->transaction_error) {
 974                 _tdb_transaction_cancel(tdb);
 975                 return tdb_logerr(tdb, TDB_ERR_EINVAL, TDB_LOG_ERROR,
 976                                   "tdb_transaction_prepare_commit:"
 977                                   " transaction error pending");
 978         }
 979
 980
 981         if (tdb->transaction->nesting != 0) {
 982                 return TDB_SUCCESS;
 983         }
 984
 985         /* check for a null transaction */
 986         if (tdb->transaction->blocks == NULL) {
 987                 return TDB_SUCCESS;
 988         }
 989
 990         methods = tdb->transaction->io_methods;
 991
 992         /* upgrade the main transaction lock region to a write lock */
 993         ecode = tdb_allrecord_upgrade(tdb);
 994         if (ecode != TDB_SUCCESS) {
 995                 return ecode;
 996         }
 997
 998         /* get the open lock - this prevents new users attaching to the database
 999            during the commit */
1000         ecode = tdb_lock_open(tdb, TDB_LOCK_WAIT|TDB_LOCK_NOCHECK);
1001         if (ecode != TDB_SUCCESS) {
1002                 return ecode;
1003         }
1004
1005         /* Since we have whole db locked, we don't need the expansion lock. */
1006         if (!(tdb->flags & TDB_NOSYNC)) {
1007                 /* Sets up tdb->transaction->recovery and
1008                  * tdb->transaction->magic_offset. */
1009                 ecode = transaction_setup_recovery(tdb);
1010                 if (ecode != TDB_SUCCESS) {
1011                         return ecode;
1012                 }
1013         }
1014
1015         tdb->transaction->prepared = true;
1016
1017         /* expand the file to the new size if needed */
1018         if (tdb->file->map_size != tdb->transaction->old_map_size) {
1019                 tdb_len_t add;
1020
1021                 add = tdb->file->map_size - tdb->transaction->old_map_size;
1022                 /* Restore original map size for tdb_expand_file */
1023                 tdb->file->map_size = tdb->transaction->old_map_size;
1024                 ecode = methods->expand_file(tdb, add);
1025                 if (ecode != TDB_SUCCESS) {
1026                         return ecode;
1027                 }
1028         }
1029
1030         /* Keep the open lock until the actual commit */
1031         return TDB_SUCCESS;
1032 }
1033
1034 /*
1035    prepare to commit the current transaction
1036 */
1037 enum TDB_ERROR tdb_transaction_prepare_commit(struct tdb_context *tdb)
1038 {
1039         return _tdb_transaction_prepare_commit(tdb);
1040 }
1041
1042 /*
1043   commit the current transaction
1044 */
1045 enum TDB_ERROR tdb_transaction_commit(struct tdb_context *tdb)
1046 {
1047         const struct tdb_methods *methods;
1048         int i;
1049         enum TDB_ERROR ecode;
1050
1051         if (tdb->transaction == NULL) {
1052                 return tdb->last_error = tdb_logerr(tdb, TDB_ERR_EINVAL,
1053                                                     TDB_LOG_USE_ERROR,
1054                                                     "tdb_transaction_commit:"
1055                                                     " no transaction");
1056         }
1057
1058         tdb_trace(tdb, "tdb_transaction_commit");
1059
1060         if (tdb->transaction->nesting != 0) {
1061                 tdb->transaction->nesting--;
1062                 return tdb->last_error = TDB_SUCCESS;
1063         }
1064
1065         /* check for a null transaction */
1066         if (tdb->transaction->blocks == NULL) {
1067                 _tdb_transaction_cancel(tdb);
1068                 return tdb->last_error = TDB_SUCCESS;
1069         }
1070
1071         if (!tdb->transaction->prepared) {
1072                 ecode = _tdb_transaction_prepare_commit(tdb);
1073                 if (ecode != TDB_SUCCESS) {
1074                         _tdb_transaction_cancel(tdb);
1075                         return tdb->last_error = ecode;
1076                 }
1077         }
1078
1079         methods = tdb->transaction->io_methods;
1080
1081         /* perform all the writes */
1082         for (i=0;i<tdb->transaction->num_blocks;i++) {
1083                 tdb_off_t offset;
1084                 tdb_len_t length;
1085
1086                 if (tdb->transaction->blocks[i] == NULL) {
1087                         continue;
1088                 }
1089
1090                 offset = i * PAGESIZE;
1091                 length = PAGESIZE;
1092                 if (i == tdb->transaction->num_blocks-1) {
1093                         length = tdb->transaction->last_block_size;
1094                 }
1095
1096                 ecode = methods->twrite(tdb, offset,
1097                                         tdb->transaction->blocks[i], length);
1098                 if (ecode != TDB_SUCCESS) {
1099                         /* we've overwritten part of the data and
1100                            possibly expanded the file, so we need to
1101                            run the crash recovery code */
1102                         tdb->methods = methods;
1103                         tdb_transaction_recover(tdb);
1104
1105                         _tdb_transaction_cancel(tdb);
1106
1107                         return tdb->last_error = ecode;
1108                 }
1109                 SAFE_FREE(tdb->transaction->blocks[i]);
1110         }
1111
1112         SAFE_FREE(tdb->transaction->blocks);
1113         tdb->transaction->num_blocks = 0;
1114
1115         /* ensure the new data is on disk */
1116         ecode = transaction_sync(tdb, 0, tdb->file->map_size);
1117         if (ecode != TDB_SUCCESS) {
1118                 return tdb->last_error = ecode;
1119         }
1120
1121         /*
1122           TODO: maybe write to some dummy hdr field, or write to magic
1123           offset without mmap, before the last sync, instead of the
1124           utime() call
1125         */
1126
1127         /* on some systems (like Linux 2.6.x) changes via mmap/msync
1128            don't change the mtime of the file, this means the file may
1129            not be backed up (as tdb rounding to block sizes means that
1130            file size changes are quite rare too). The following forces
1131            mtime changes when a transaction completes */
1132 #if HAVE_UTIME
1133         utime(tdb->name, NULL);
1134 #endif
1135
1136         /* use a transaction cancel to free memory and remove the
1137            transaction locks: it "restores" map_size, too. */
1138         tdb->transaction->old_map_size = tdb->file->map_size;
1139         _tdb_transaction_cancel(tdb);
1140
1141         return tdb->last_error = TDB_SUCCESS;
1142 }
1143
1144
1145 /*
1146   recover from an aborted transaction. Must be called with exclusive
1147   database write access already established (including the open
1148   lock to prevent new processes attaching)
1149 */
1150 enum TDB_ERROR tdb_transaction_recover(struct tdb_context *tdb)
1151 {
1152         tdb_off_t recovery_head, recovery_eof;
1153         unsigned char *data, *p;
1154         struct tdb_recovery_record rec;
1155         enum TDB_ERROR ecode;
1156
1157         /* find the recovery area */
1158         recovery_head = tdb_read_off(tdb, offsetof(struct tdb_header,recovery));
1159         if (TDB_OFF_IS_ERR(recovery_head)) {
1160                 return tdb_logerr(tdb, recovery_head, TDB_LOG_ERROR,
1161                                   "tdb_transaction_recover:"
1162                                   " failed to read recovery head");
1163         }
1164
1165         if (recovery_head == 0) {
1166                 /* we have never allocated a recovery record */
1167                 return TDB_SUCCESS;
1168         }
1169
1170         /* read the recovery record */
1171         ecode = tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec));
1172         if (ecode != TDB_SUCCESS) {
1173                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1174                                   "tdb_transaction_recover:"
1175                                   " failed to read recovery record");
1176         }
1177
1178         if (rec.magic != TDB_RECOVERY_MAGIC) {
1179                 /* there is no valid recovery data */
1180                 return TDB_SUCCESS;
1181         }
1182
1183         if (tdb->read_only) {
1184                 return tdb_logerr(tdb, TDB_ERR_CORRUPT, TDB_LOG_ERROR,
1185                                   "tdb_transaction_recover:"
1186                                   " attempt to recover read only database");
1187         }
1188
1189         recovery_eof = rec.eof;
1190
1191         data = (unsigned char *)malloc(rec.len);
1192         if (data == NULL) {
1193                 return tdb_logerr(tdb, TDB_ERR_OOM, TDB_LOG_ERROR,
1194                                   "tdb_transaction_recover:"
1195                                   " failed to allocate recovery data");
1196         }
1197
1198         /* read the full recovery data */
1199         ecode = tdb->methods->tread(tdb, recovery_head + sizeof(rec), data,
1200                                     rec.len);
1201         if (ecode != TDB_SUCCESS) {
1202                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1203                                   "tdb_transaction_recover:"
1204                                   " failed to read recovery data");
1205         }
1206
1207         /* recover the file data */
1208         p = data;
1209         while (p+sizeof(tdb_off_t)+sizeof(tdb_len_t) < data + rec.len) {
1210                 tdb_off_t ofs;
1211                 tdb_len_t len;
1212                 tdb_convert(tdb, p, sizeof(ofs) + sizeof(len));
1213                 memcpy(&ofs, p, sizeof(ofs));
1214                 memcpy(&len, p + sizeof(ofs), sizeof(len));
1215                 p += sizeof(ofs) + sizeof(len);
1216
1217                 ecode = tdb->methods->twrite(tdb, ofs, p, len);
1218                 if (ecode != TDB_SUCCESS) {
1219                         free(data);
1220                         return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1221                                           "tdb_transaction_recover:"
1222                                           " failed to recover %zu bytes"
1223                                           " at offset %zu",
1224                                           (size_t)len, (size_t)ofs);
1225                 }
1226                 p += len;
1227         }
1228
1229         free(data);
1230
1231         ecode = transaction_sync(tdb, 0, tdb->file->map_size);
1232         if (ecode != TDB_SUCCESS) {
1233                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1234                                   "tdb_transaction_recover:"
1235                                   " failed to sync recovery");
1236         }
1237
1238         /* if the recovery area is after the recovered eof then remove it */
1239         if (recovery_eof <= recovery_head) {
1240                 ecode = tdb_write_off(tdb, offsetof(struct tdb_header,
1241                                                     recovery),
1242                                       0);
1243                 if (ecode != TDB_SUCCESS) {
1244                         return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1245                                           "tdb_transaction_recover:"
1246                                           " failed to remove recovery head");
1247                 }
1248         }
1249
1250         /* remove the recovery magic */
1251         ecode = tdb_write_off(tdb,
1252                               recovery_head
1253                               + offsetof(struct tdb_recovery_record, magic),
1254                               TDB_RECOVERY_INVALID_MAGIC);
1255         if (ecode != TDB_SUCCESS) {
1256                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1257                                   "tdb_transaction_recover:"
1258                                   " failed to remove recovery magic");
1259         }
1260
1261         ecode = transaction_sync(tdb, 0, recovery_eof);
1262         if (ecode != TDB_SUCCESS) {
1263                 return tdb_logerr(tdb, ecode, TDB_LOG_ERROR,
1264                                   "tdb_transaction_recover:"
1265                                   " failed to sync2 recovery");
1266         }
1267
1268         tdb_logerr(tdb, TDB_SUCCESS, TDB_LOG_WARNING,
1269                    "tdb_transaction_recover: recovered %zu byte database",
1270                    (size_t)recovery_eof);
1271
1272         /* all done */
1273         return TDB_SUCCESS;
1274 }
1275
1276 tdb_bool_err tdb_needs_recovery(struct tdb_context *tdb)
1277 {
1278         tdb_off_t recovery_head;
1279         struct tdb_recovery_record rec;
1280         enum TDB_ERROR ecode;
1281
1282         /* find the recovery area */
1283         recovery_head = tdb_read_off(tdb, offsetof(struct tdb_header,recovery));
1284         if (TDB_OFF_IS_ERR(recovery_head)) {
1285                 return recovery_head;
1286         }
1287
1288         if (recovery_head == 0) {
1289                 /* we have never allocated a recovery record */
1290                 return false;
1291         }
1292
1293         /* read the recovery record */
1294         ecode = tdb_read_convert(tdb, recovery_head, &rec, sizeof(rec));
1295         if (ecode != TDB_SUCCESS) {
1296                 return ecode;
1297         }
1298
1299         return (rec.magic == TDB_RECOVERY_MAGIC);
1300 }