citadel/database.c

   1 // This is a data store backend for the Citadel server which uses Berkeley DB.
   2 //
   3 // Copyright (c) 1987-2021 by the citadel.org team
   4 //
   5 // This program is open source software; you can redistribute it and/or
   6 // modify it under the terms of the GNU General Public License version 3.
   7 //
   8 // This program is distributed in the hope that it will be useful,
   9 // but WITHOUT ANY WARRANTY; without even the implied warranty of
  10 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  11 // GNU General Public License for more details.
  12
  13 /*****************************************************************************
  14        Tunable configuration parameters for the Berkeley DB back end
  15  *****************************************************************************/
  16
  17 /* Citadel will checkpoint the db at the end of every session, but only if
  18  * the specified number of kilobytes has been written, or if the specified
  19  * number of minutes has passed, since the last checkpoint.
  20  */
  21 #define MAX_CHECKPOINT_KBYTES   256
  22 #define MAX_CHECKPOINT_MINUTES  15
  23
  24 /*****************************************************************************/
  25
  26 #include "sysdep.h"
  27 #include <stdlib.h>
  28 #include <unistd.h>
  29 #include <sys/stat.h>
  30 #include <stdio.h>
  31 #include <dirent.h>
  32 #include <zlib.h>
  33 #include <db.h>
  34
  35 #if DB_VERSION_MAJOR < 5
  36 #error Citadel requires Berkeley DB v5.0 or newer.  Please upgrade.
  37 #endif
  38
  39 #include <libcitadel.h>
  40 #include "ctdl_module.h"
  41 #include "control.h"
  42 #include "citserver.h"
  43 #include "config.h"
  44
  45 static DB *dbp[MAXCDB];         /* One DB handle for each Citadel database */
  46 static DB_ENV *dbenv;           /* The DB environment (global) */
  47
  48
  49 void cdb_abort(void) {
  50         syslog(LOG_DEBUG, "db: citserver is stopping in order to prevent data loss. uid=%d gid=%d euid=%d egid=%d",
  51                 getuid(), getgid(), geteuid(), getegid()
  52         );
  53         exit(CTDLEXIT_DB);
  54 }
  55
  56
  57 /* Verbose logging callback */
  58 void cdb_verbose_log(const DB_ENV * dbenv, const char *msg) {
  59         if (!IsEmptyStr(msg)) {
  60                 syslog(LOG_DEBUG, "db: %s", msg);
  61         }
  62 }
  63
  64
  65 /* Verbose logging callback */
  66 void cdb_verbose_err(const DB_ENV * dbenv, const char *errpfx, const char *msg) {
  67         syslog(LOG_ERR, "db: %s", msg);
  68 }
  69
  70
  71 /* wrapper for txn_abort() that logs/aborts on error */
  72 static void txabort(DB_TXN *tid) {
  73         int ret;
  74
  75         ret = tid->abort(tid);
  76
  77         if (ret) {
  78                 syslog(LOG_ERR, "db: txn_abort: %s", db_strerror(ret));
  79                 cdb_abort();
  80         }
  81 }
  82
  83
  84 /* wrapper for txn_commit() that logs/aborts on error */
  85 static void txcommit(DB_TXN *tid) {
  86         int ret;
  87
  88         ret = tid->commit(tid, 0);
  89
  90         if (ret) {
  91                 syslog(LOG_ERR, "db: txn_commit: %s", db_strerror(ret));
  92                 cdb_abort();
  93         }
  94 }
  95
  96
  97 /* wrapper for txn_begin() that logs/aborts on error */
  98 static void txbegin(DB_TXN **tid) {
  99         int ret;
 100
 101         ret = dbenv->txn_begin(dbenv, NULL, tid, 0);
 102
 103         if (ret) {
 104                 syslog(LOG_ERR, "db: txn_begin: %s", db_strerror(ret));
 105                 cdb_abort();
 106         }
 107 }
 108
 109
 110 /* panic callback */
 111 static void dbpanic(DB_ENV * env, int errval) {
 112         syslog(LOG_ERR, "db: PANIC: %s", db_strerror(errval));
 113 }
 114
 115
 116 static void cclose(DBC * cursor) {
 117         int ret;
 118
 119         if ((ret = cursor->c_close(cursor))) {
 120                 syslog(LOG_ERR, "db: c_close: %s", db_strerror(ret));
 121                 cdb_abort();
 122         }
 123 }
 124
 125
 126 static void bailIfCursor(DBC ** cursors, const char *msg) {
 127         int i;
 128
 129         for (i = 0; i < MAXCDB; i++)
 130                 if (cursors[i] != NULL) {
 131                         syslog(LOG_ERR, "db: cursor still in progress on cdb %02x: %s", i, msg);
 132                         cdb_abort();
 133                 }
 134 }
 135
 136
 137 void cdb_check_handles(void) {
 138         bailIfCursor(TSD->cursors, "in check_handles");
 139
 140         if (TSD->tid != NULL) {
 141                 syslog(LOG_ERR, "db: transaction still in progress!");
 142                 cdb_abort();
 143         }
 144 }
 145
 146
 147 /*
 148  * Request a checkpoint of the database.  Called once per minute by the thread manager.
 149  */
 150 void cdb_checkpoint(void) {
 151         int ret;
 152
 153         syslog(LOG_DEBUG, "db: -- checkpoint --");
 154         ret = dbenv->txn_checkpoint(dbenv, MAX_CHECKPOINT_KBYTES, MAX_CHECKPOINT_MINUTES, 0);
 155
 156         if (ret != 0) {
 157                 syslog(LOG_ERR, "db: cdb_checkpoint() txn_checkpoint: %s", db_strerror(ret));
 158                 cdb_abort();
 159         }
 160
 161         /* After a successful checkpoint, we can cull the unused logs */
 162         if (CtdlGetConfigInt("c_auto_cull")) {
 163                 ret = dbenv->log_set_config(dbenv, DB_LOG_AUTO_REMOVE, 1);
 164         }
 165         else {
 166                 ret = dbenv->log_set_config(dbenv, DB_LOG_AUTO_REMOVE, 0);
 167         }
 168 }
 169
 170
 171 /*
 172  * Open the various databases we'll be using.  Any database which
 173  * does not exist should be created.  Note that we don't need a
 174  * critical section here, because there aren't any active threads
 175  * manipulating the database yet.
 176  */
 177 void open_databases(void) {
 178         int ret;
 179         int i;
 180         char dbfilename[32];
 181         u_int32_t flags = 0;
 182         int dbversion_major, dbversion_minor, dbversion_patch;
 183
 184         syslog(LOG_DEBUG, "db: open_databases() starting");
 185         syslog(LOG_DEBUG, "db: Compiled libdb: %s", DB_VERSION_STRING);
 186         syslog(LOG_DEBUG, "db:   Linked libdb: %s", db_version(&dbversion_major, &dbversion_minor, &dbversion_patch));
 187         syslog(LOG_DEBUG, "db:    Linked zlib: %s", zlibVersion());
 188
 189         /*
 190          * Silently try to create the database subdirectory.  If it's already there, no problem.
 191          */
 192         if ((mkdir(ctdl_db_dir, 0700) != 0) && (errno != EEXIST)) {
 193                 syslog(LOG_ERR, "db: unable to create database directory [%s]: %m", ctdl_db_dir);
 194         }
 195         if (chmod(ctdl_db_dir, 0700) != 0) {
 196                 syslog(LOG_ERR, "db: unable to set database directory permissions [%s]: %m", ctdl_db_dir);
 197         }
 198         if (chown(ctdl_db_dir, CTDLUID, (-1)) != 0) {
 199                 syslog(LOG_ERR, "db: unable to set the owner for [%s]: %m", ctdl_db_dir);
 200         }
 201         syslog(LOG_DEBUG, "db: Setting up DB environment");
 202         // db_env_set_func_yield((int (*)(u_long,  u_long))sched_yield);
 203         ret = db_env_create(&dbenv, 0);
 204         if (ret) {
 205                 syslog(LOG_ERR, "db: db_env_create: %s", db_strerror(ret));
 206                 syslog(LOG_ERR, "db: exit code %d", ret);
 207                 exit(CTDLEXIT_DB);
 208         }
 209         dbenv->set_errpfx(dbenv, "citserver");
 210         dbenv->set_paniccall(dbenv, dbpanic);
 211         dbenv->set_errcall(dbenv, cdb_verbose_err);
 212         dbenv->set_errpfx(dbenv, "ctdl");
 213         dbenv->set_msgcall(dbenv, cdb_verbose_log);
 214         dbenv->set_verbose(dbenv, DB_VERB_DEADLOCK, 1);
 215         dbenv->set_verbose(dbenv, DB_VERB_RECOVERY, 1);
 216
 217         /*
 218          * We want to specify the shared memory buffer pool cachesize,
 219          * but everything else is the default.
 220          */
 221         ret = dbenv->set_cachesize(dbenv, 0, 64 * 1024, 0);
 222         if (ret) {
 223                 syslog(LOG_ERR, "db: set_cachesize: %s", db_strerror(ret));
 224                 dbenv->close(dbenv, 0);
 225                 syslog(LOG_ERR, "db: exit code %d", ret);
 226                 exit(CTDLEXIT_DB);
 227         }
 228
 229         if ((ret = dbenv->set_lk_detect(dbenv, DB_LOCK_DEFAULT))) {
 230                 syslog(LOG_ERR, "db: set_lk_detect: %s", db_strerror(ret));
 231                 dbenv->close(dbenv, 0);
 232                 syslog(LOG_ERR, "db: exit code %d", ret);
 233                 exit(CTDLEXIT_DB);
 234         }
 235
 236         flags = DB_CREATE | DB_INIT_MPOOL | DB_PRIVATE | DB_INIT_TXN | DB_INIT_LOCK | DB_THREAD | DB_INIT_LOG;
 237         syslog(LOG_DEBUG, "db: dbenv->open(dbenv, %s, %d, 0)", ctdl_db_dir, flags);
 238         ret = dbenv->open(dbenv, ctdl_db_dir, flags, 0);                                // try opening the database cleanly
 239         if (ret == DB_RUNRECOVERY) {
 240                 syslog(LOG_ERR, "db: dbenv->open: %s", db_strerror(ret));
 241                 syslog(LOG_ERR, "db: attempting recovery...");
 242                 flags |= DB_RECOVER;
 243                 ret = dbenv->open(dbenv, ctdl_db_dir, flags, 0);                        // try recovery
 244         }
 245         if (ret == DB_RUNRECOVERY) {
 246                 syslog(LOG_ERR, "db: dbenv->open: %s", db_strerror(ret));
 247                 syslog(LOG_ERR, "db: attempting catastrophic recovery...");
 248                 flags &= ~DB_RECOVER;
 249                 flags |= DB_RECOVER_FATAL;
 250                 ret = dbenv->open(dbenv, ctdl_db_dir, flags, 0);                        // try catastrophic recovery
 251         }
 252         if (ret) {
 253                 syslog(LOG_ERR, "db: dbenv->open: %s", db_strerror(ret));
 254                 dbenv->close(dbenv, 0);
 255                 syslog(LOG_ERR, "db: exit code %d", ret);
 256                 exit(CTDLEXIT_DB);
 257         }
 258
 259         syslog(LOG_INFO, "db: mounting databases");
 260         for (i = 0; i < MAXCDB; ++i) {
 261                 ret = db_create(&dbp[i], dbenv, 0);                                     // Create a database handle
 262                 if (ret) {
 263                         syslog(LOG_ERR, "db: db_create: %s", db_strerror(ret));
 264                         syslog(LOG_ERR, "db: exit code %d", ret);
 265                         exit(CTDLEXIT_DB);
 266                 }
 267
 268                 snprintf(dbfilename, sizeof dbfilename, "cdb.%02x", i);                 // table names by number
 269                 ret = dbp[i]->open(dbp[i], NULL, dbfilename, NULL, DB_BTREE, DB_CREATE | DB_AUTO_COMMIT | DB_THREAD, 0600);
 270                 if (ret) {
 271                         syslog(LOG_ERR, "db: db_open[%02x]: %s", i, db_strerror(ret));
 272                         if (ret == ENOMEM) {
 273                                 syslog(LOG_ERR, "db: You may need to tune your database; please read http://www.citadel.org/doku.php?id=faq:troubleshooting:out_of_lock_entries for more information.");
 274                         }
 275                         syslog(LOG_ERR, "db: exit code %d", ret);
 276                         exit(CTDLEXIT_DB);
 277                 }
 278         }
 279 }
 280
 281
 282 /*
 283  * Make sure we own all the files, because in a few milliseconds we're going to drop root privs.
 284  */
 285 void cdb_chmod_data(void) {
 286         DIR *dp;
 287         struct dirent *d;
 288         char filename[PATH_MAX];
 289
 290         dp = opendir(ctdl_db_dir);
 291         if (dp != NULL) {
 292                 while (d = readdir(dp), d != NULL) {
 293                         if (d->d_name[0] != '.') {
 294                                 snprintf(filename, sizeof filename, "%s/%s", ctdl_db_dir, d->d_name);
 295                                 syslog(LOG_DEBUG, "db: chmod(%s, 0600) returned %d", filename, chmod(filename, 0600));
 296                                 syslog(LOG_DEBUG, "db: chown(%s, CTDLUID, -1) returned %d",
 297                                         filename, chown(filename, CTDLUID, (-1))
 298                                 );
 299                         }
 300                 }
 301                 closedir(dp);
 302         }
 303 }
 304
 305
 306 /*
 307  * Close all of the db database files we've opened.  This can be done
 308  * in a loop, since it's just a bunch of closes.
 309  */
 310 void close_databases(void) {
 311         int i;
 312         int ret;
 313
 314         syslog(LOG_INFO, "db: performing final checkpoint");
 315         if ((ret = dbenv->txn_checkpoint(dbenv, 0, 0, 0))) {
 316                 syslog(LOG_ERR, "db: txn_checkpoint: %s", db_strerror(ret));
 317         }
 318
 319         syslog(LOG_INFO, "db: flushing the database logs");
 320         if ((ret = dbenv->log_flush(dbenv, NULL))) {
 321                 syslog(LOG_ERR, "db: log_flush: %s", db_strerror(ret));
 322         }
 323
 324         /* close the tables */
 325         syslog(LOG_INFO, "db: closing databases");
 326         for (i = 0; i < MAXCDB; ++i) {
 327                 syslog(LOG_INFO, "db: closing database %02x", i);
 328                 ret = dbp[i]->close(dbp[i], 0);
 329                 if (ret) {
 330                         syslog(LOG_ERR, "db: db_close: %s", db_strerror(ret));
 331                 }
 332
 333         }
 334
 335         // This seemed nifty at the time but did anyone really look at it?
 336         // #ifdef DB_STAT_ALL
 337         // /* print some statistics... */
 338         // dbenv->lock_stat_print(dbenv, DB_STAT_ALL);
 339         // #endif
 340
 341         /* Close the handle. */
 342         ret = dbenv->close(dbenv, 0);
 343         if (ret) {
 344                 syslog(LOG_ERR, "db: DBENV->close: %s", db_strerror(ret));
 345         }
 346 }
 347
 348
 349 /*
 350  * Decompress a database item if it was compressed on disk
 351  */
 352 void cdb_decompress_if_necessary(struct cdbdata *cdb) {
 353         static int magic = COMPRESS_MAGIC;
 354
 355         if ((cdb == NULL) || (cdb->ptr == NULL) || (cdb->len < sizeof(magic)) || (memcmp(cdb->ptr, &magic, sizeof(magic)))) {
 356                 return;
 357         }
 358
 359         /* At this point we know we're looking at a compressed item. */
 360
 361         struct CtdlCompressHeader zheader;
 362         char *uncompressed_data;
 363         char *compressed_data;
 364         uLongf destLen, sourceLen;
 365         size_t cplen;
 366
 367         memset(&zheader, 0, sizeof(struct CtdlCompressHeader));
 368         cplen = sizeof(struct CtdlCompressHeader);
 369         if (sizeof(struct CtdlCompressHeader) > cdb->len) {
 370                 cplen = cdb->len;
 371         }
 372         memcpy(&zheader, cdb->ptr, cplen);
 373
 374         compressed_data = cdb->ptr;
 375         compressed_data += sizeof(struct CtdlCompressHeader);
 376
 377         sourceLen = (uLongf) zheader.compressed_len;
 378         destLen = (uLongf) zheader.uncompressed_len;
 379         uncompressed_data = malloc(zheader.uncompressed_len);
 380
 381         if (uncompress((Bytef *) uncompressed_data,
 382                        (uLongf *) & destLen, (const Bytef *) compressed_data, (uLong) sourceLen) != Z_OK) {
 383                 syslog(LOG_ERR, "db: uncompress() error");
 384                 cdb_abort();
 385         }
 386
 387         free(cdb->ptr);
 388         cdb->len = (size_t) destLen;
 389         cdb->ptr = uncompressed_data;
 390 }
 391
 392
 393 /*
 394  * Store a piece of data.  Returns 0 if the operation was successful.  If a
 395  * key already exists it should be overwritten.
 396  */
 397 int cdb_store(int cdb, const void *ckey, int ckeylen, void *cdata, int cdatalen) {
 398
 399         DBT dkey, ddata;
 400         DB_TXN *tid = NULL;
 401         int ret = 0;
 402         struct CtdlCompressHeader zheader;
 403         char *compressed_data = NULL;
 404         int compressing = 0;
 405         size_t buffer_len = 0;
 406         uLongf destLen = 0;
 407
 408         memset(&dkey, 0, sizeof(DBT));
 409         memset(&ddata, 0, sizeof(DBT));
 410         dkey.size = ckeylen;
 411         dkey.data = (void *) ckey;
 412         ddata.size = cdatalen;
 413         ddata.data = cdata;
 414
 415         /* Only compress Visit and UseTable records.  Everything else is uncompressed. */
 416         if ((cdb == CDB_VISIT) || (cdb == CDB_USETABLE)) {
 417                 compressing = 1;
 418                 zheader.magic = COMPRESS_MAGIC;
 419                 zheader.uncompressed_len = cdatalen;
 420                 buffer_len = ((cdatalen * 101) / 100) + 100 + sizeof(struct CtdlCompressHeader);
 421                 destLen = (uLongf) buffer_len;
 422                 compressed_data = malloc(buffer_len);
 423                 if (compress2((Bytef *) (compressed_data + sizeof(struct CtdlCompressHeader)),
 424                               &destLen, (Bytef *) cdata, (uLongf) cdatalen, 1) != Z_OK) {
 425                         syslog(LOG_ERR, "db: compress2() error");
 426                         cdb_abort();
 427                 }
 428                 zheader.compressed_len = (size_t) destLen;
 429                 memcpy(compressed_data, &zheader, sizeof(struct CtdlCompressHeader));
 430                 ddata.size = (size_t) (sizeof(struct CtdlCompressHeader) + zheader.compressed_len);
 431                 ddata.data = compressed_data;
 432         }
 433
 434         if (TSD->tid != NULL) {
 435                 ret = dbp[cdb]->put(dbp[cdb],   // db
 436                                     TSD->tid,   // transaction ID
 437                                     &dkey,      // key
 438                                     &ddata,     // data
 439                                     0           // flags
 440                 );
 441                 if (ret) {
 442                         syslog(LOG_ERR, "db: cdb_store(%d): %s", cdb, db_strerror(ret));
 443                         cdb_abort();
 444                 }
 445                 if (compressing) {
 446                         free(compressed_data);
 447                 }
 448                 return ret;
 449         } else {
 450                 bailIfCursor(TSD->cursors, "attempt to write during r/o cursor");
 451
 452               retry:
 453                 txbegin(&tid);
 454
 455                 if ((ret = dbp[cdb]->put(dbp[cdb],      // db
 456                                          tid,           // transaction ID
 457                                          &dkey,         // key
 458                                          &ddata,        // data
 459                                          0))) {         // flags
 460                         if (ret == DB_LOCK_DEADLOCK) {
 461                                 txabort(tid);
 462                                 goto retry;
 463                         } else {
 464                                 syslog(LOG_ERR, "db: cdb_store(%d): %s", cdb, db_strerror(ret));
 465                                 cdb_abort();
 466                         }
 467                 } else {
 468                         txcommit(tid);
 469                         if (compressing) {
 470                                 free(compressed_data);
 471                         }
 472                         return ret;
 473                 }
 474         }
 475         return ret;
 476 }
 477
 478
 479 /*
 480  * Delete a piece of data.  Returns 0 if the operation was successful.
 481  */
 482 int cdb_delete(int cdb, void *key, int keylen) {
 483         DBT dkey;
 484         DB_TXN *tid;
 485         int ret;
 486
 487         memset(&dkey, 0, sizeof dkey);
 488         dkey.size = keylen;
 489         dkey.data = key;
 490
 491         if (TSD->tid != NULL) {
 492                 ret = dbp[cdb]->del(dbp[cdb], TSD->tid, &dkey, 0);
 493                 if (ret) {
 494                         syslog(LOG_ERR, "db: cdb_delete(%d): %s", cdb, db_strerror(ret));
 495                         if (ret != DB_NOTFOUND) {
 496                                 cdb_abort();
 497                         }
 498                 }
 499         } else {
 500                 bailIfCursor(TSD->cursors, "attempt to delete during r/o cursor");
 501
 502               retry:
 503                 txbegin(&tid);
 504
 505                 if ((ret = dbp[cdb]->del(dbp[cdb], tid, &dkey, 0)) && ret != DB_NOTFOUND) {
 506                         if (ret == DB_LOCK_DEADLOCK) {
 507                                 txabort(tid);
 508                                 goto retry;
 509                         } else {
 510                                 syslog(LOG_ERR, "db: cdb_delete(%d): %s", cdb, db_strerror(ret));
 511                                 cdb_abort();
 512                         }
 513                 } else {
 514                         txcommit(tid);
 515                 }
 516         }
 517         return ret;
 518 }
 519
 520
 521 static DBC *localcursor(int cdb) {
 522         int ret;
 523         DBC *curs;
 524
 525         if (TSD->cursors[cdb] == NULL) {
 526                 ret = dbp[cdb]->cursor(dbp[cdb], TSD->tid, &curs, 0);
 527         }
 528         else {
 529                 ret = TSD->cursors[cdb]->c_dup(TSD->cursors[cdb], &curs, DB_POSITION);
 530         }
 531
 532         if (ret) {
 533                 syslog(LOG_ERR, "db: localcursor: %s", db_strerror(ret));
 534                 cdb_abort();
 535         }
 536
 537         return curs;
 538 }
 539
 540
 541 /*
 542  * Fetch a piece of data.  If not found, returns NULL.  Otherwise, it returns
 543  * a struct cdbdata which it is the caller's responsibility to free later on
 544  * using the cdb_free() routine.
 545  */
 546 struct cdbdata *cdb_fetch(int cdb, const void *key, int keylen) {
 547
 548         if (keylen == 0) {              // key length zero is impossible
 549                 return(NULL);
 550         }
 551
 552         struct cdbdata *tempcdb;
 553         DBT dkey, dret;
 554         int ret;
 555
 556         memset(&dkey, 0, sizeof(DBT));
 557         dkey.size = keylen;
 558         dkey.data = (void *) key;
 559
 560         if (TSD->tid != NULL) {
 561                 memset(&dret, 0, sizeof(DBT));
 562                 dret.flags = DB_DBT_MALLOC;
 563                 ret = dbp[cdb]->get(dbp[cdb], TSD->tid, &dkey, &dret, 0);
 564         }
 565         else {
 566                 DBC *curs;
 567
 568                 do {
 569                         memset(&dret, 0, sizeof(DBT));
 570                         dret.flags = DB_DBT_MALLOC;
 571                         curs = localcursor(cdb);
 572                         ret = curs->c_get(curs, &dkey, &dret, DB_SET);
 573                         cclose(curs);
 574                 } while (ret == DB_LOCK_DEADLOCK);
 575         }
 576
 577         if ((ret != 0) && (ret != DB_NOTFOUND)) {
 578                 syslog(LOG_ERR, "db: cdb_fetch(%d): %s", cdb, db_strerror(ret));
 579                 cdb_abort();
 580         }
 581
 582         if (ret != 0) {
 583                 return NULL;
 584         }
 585
 586         tempcdb = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 587         if (tempcdb == NULL) {
 588                 syslog(LOG_ERR, "db: cdb_fetch() cannot allocate memory for tempcdb: %m");
 589                 cdb_abort();
 590                 return NULL;    /* make it easier for static analysis... */
 591         }
 592         else {
 593                 tempcdb->len = dret.size;
 594                 tempcdb->ptr = dret.data;
 595                 cdb_decompress_if_necessary(tempcdb);
 596                 return (tempcdb);
 597         }
 598 }
 599
 600
 601 /*
 602  * Free a cdbdata item.
 603  *
 604  * Note that we only free the 'ptr' portion if it is not NULL.  This allows
 605  * other code to assume ownership of that memory simply by storing the
 606  * pointer elsewhere and then setting 'ptr' to NULL.  cdb_free() will then
 607  * avoid freeing it.
 608  */
 609 void cdb_free(struct cdbdata *cdb) {
 610         if (cdb->ptr) {
 611                 free(cdb->ptr);
 612         }
 613         free(cdb);
 614 }
 615
 616
 617 void cdb_close_cursor(int cdb) {
 618         if (TSD->cursors[cdb] != NULL) {
 619                 cclose(TSD->cursors[cdb]);
 620         }
 621
 622         TSD->cursors[cdb] = NULL;
 623 }
 624
 625
 626 /*
 627  * Prepare for a sequential search of an entire database.
 628  * (There is guaranteed to be no more than one traversal in
 629  * progress per thread at any given time.)
 630  */
 631 void cdb_rewind(int cdb) {
 632         int ret = 0;
 633
 634         if (TSD->cursors[cdb] != NULL) {
 635                 syslog(LOG_ERR, "db: cdb_rewind: must close cursor on database %d before reopening", cdb);
 636                 cdb_abort();
 637                 /* cclose(TSD->cursors[cdb]); */
 638         }
 639
 640         /*
 641          * Now initialize the cursor
 642          */
 643         ret = dbp[cdb]->cursor(dbp[cdb], TSD->tid, &TSD->cursors[cdb], 0);
 644         if (ret) {
 645                 syslog(LOG_ERR, "db: cdb_rewind: db_cursor: %s", db_strerror(ret));
 646                 cdb_abort();
 647         }
 648 }
 649
 650
 651 /*
 652  * Fetch the next item in a sequential search.  Returns a pointer to a
 653  * cdbdata structure, or NULL if we've hit the end.
 654  */
 655 struct cdbdata *cdb_next_item(int cdb) {
 656         DBT key, data;
 657         struct cdbdata *cdbret;
 658         int ret = 0;
 659
 660         /* Initialize the key/data pair so the flags aren't set. */
 661         memset(&key, 0, sizeof(key));
 662         memset(&data, 0, sizeof(data));
 663         data.flags = DB_DBT_MALLOC;
 664
 665         ret = TSD->cursors[cdb]->c_get(TSD->cursors[cdb], &key, &data, DB_NEXT);
 666
 667         if (ret) {
 668                 if (ret != DB_NOTFOUND) {
 669                         syslog(LOG_ERR, "db: cdb_next_item(%d): %s", cdb, db_strerror(ret));
 670                         cdb_abort();
 671                 }
 672                 cdb_close_cursor(cdb);
 673                 return NULL;    /* presumably, end of file */
 674         }
 675
 676         cdbret = (struct cdbdata *) malloc(sizeof(struct cdbdata));
 677         cdbret->len = data.size;
 678         cdbret->ptr = data.data;
 679         cdb_decompress_if_necessary(cdbret);
 680
 681         return (cdbret);
 682 }
 683
 684
 685 /*
 686  * Transaction-based stuff.  I'm writing this as I bake cookies...
 687  */
 688 void cdb_begin_transaction(void) {
 689         bailIfCursor(TSD->cursors, "can't begin transaction during r/o cursor");
 690
 691         if (TSD->tid != NULL) {
 692                 syslog(LOG_ERR, "db: cdb_begin_transaction: ERROR: nested transaction");
 693                 cdb_abort();
 694         }
 695
 696         txbegin(&TSD->tid);
 697 }
 698
 699
 700 void cdb_end_transaction(void) {
 701         int i;
 702
 703         for (i = 0; i < MAXCDB; i++) {
 704                 if (TSD->cursors[i] != NULL) {
 705                         syslog(LOG_WARNING, "db: cdb_end_transaction: WARNING: cursor %d still open at transaction end", i);
 706                         cclose(TSD->cursors[i]);
 707                         TSD->cursors[i] = NULL;
 708                 }
 709         }
 710
 711         if (TSD->tid == NULL) {
 712                 syslog(LOG_ERR, "db: cdb_end_transaction: ERROR: txcommit(NULL) !!");
 713                 cdb_abort();
 714         }
 715         else {
 716                 txcommit(TSD->tid);
 717         }
 718
 719         TSD->tid = NULL;
 720 }
 721
 722
 723 /*
 724  * Truncate (delete every record)
 725  */
 726 void cdb_trunc(int cdb) {
 727         /* DB_TXN *tid; */
 728         int ret;
 729         u_int32_t count;
 730
 731         if (TSD->tid != NULL) {
 732                 syslog(LOG_ERR, "db: cdb_trunc must not be called in a transaction.");
 733                 cdb_abort();
 734         }
 735         else {
 736                 bailIfCursor(TSD->cursors, "attempt to write during r/o cursor");
 737
 738               retry:
 739                 /* txbegin(&tid); */
 740
 741                 if ((ret = dbp[cdb]->truncate(dbp[cdb], /* db */
 742                                               NULL,     /* transaction ID */
 743                                               &count,   /* #rows deleted */
 744                                               0))) {    /* flags */
 745                         if (ret == DB_LOCK_DEADLOCK) {
 746                                 /* txabort(tid); */
 747                                 goto retry;
 748                         } else {
 749                                 syslog(LOG_ERR, "db: cdb_truncate(%d): %s", cdb, db_strerror(ret));
 750                                 if (ret == ENOMEM) {
 751                                         syslog(LOG_ERR, "db: You may need to tune your database; please read http://www.citadel.org/doku.php?id=faq:troubleshooting:out_of_lock_entries for more information.");
 752                                 }
 753                                 exit(CTDLEXIT_DB);
 754                         }
 755                 }
 756                 else {
 757                         /* txcommit(tid); */
 758                 }
 759         }
 760 }
 761
 762
 763 // compact (defragment) the database, possibly returning space back to the underlying filesystem
 764 void cdb_compact(void) {
 765         int ret;
 766         int i;
 767
 768         syslog(LOG_DEBUG, "db: cdb_compact() started");
 769         for (i = 0; i < MAXCDB; i++) {
 770                 syslog(LOG_DEBUG, "db: compacting database %d", i);
 771                 ret = dbp[i]->compact(dbp[i], NULL, NULL, NULL, NULL, DB_FREE_SPACE, NULL);
 772                 if (ret) {
 773                         syslog(LOG_ERR, "db: compact: %s", db_strerror(ret));
 774                 }
 775         }
 776         syslog(LOG_DEBUG, "db: cdb_compact() finished");
 777 }
 778
 779
 780 // Has an item already been seen (is it in the CDB_USETABLE) ?
 781 // Returns 0 if it hasn't, 1 if it has
 782 // In either case, writes the item to the database for next time.
 783 int CheckIfAlreadySeen(StrBuf *guid) {
 784         int found = 0;
 785         struct UseTable ut;
 786         struct cdbdata *cdbut;
 787
 788         syslog(LOG_DEBUG, "db: CheckIfAlreadySeen(%s)", ChrPtr(guid));
 789         cdbut = cdb_fetch(CDB_USETABLE, SKEY(guid));
 790         if (cdbut != NULL) {
 791                 found = 1;
 792                 cdb_free(cdbut);
 793         }
 794
 795         /* (Re)write the record, to update the timestamp.  Zeroing it out makes it compress better. */
 796         memset(&ut, 0, sizeof(struct UseTable));
 797         memcpy(ut.ut_msgid, SKEY(guid));
 798         ut.ut_timestamp = time(NULL);
 799         cdb_store(CDB_USETABLE, SKEY(guid), &ut, sizeof(struct UseTable));
 800         return (found);
 801 }
 802
 803
 804 CTDL_MODULE_INIT(database)
 805 {
 806         if (!threading) {
 807                 // nothing to do here
 808         }
 809
 810         /* return our module id for the log */
 811         return "database";
 812 }