core-d08ae9eff/libpromises/dbm_lmdb.c

/*
  Copyright 2021 Northern.tech AS

  This file is part of CFEngine 3 - written and maintained by Northern.tech AS.

  This program is free software; you can redistribute it and/or modify it
  under the terms of the GNU General Public License as published by the
  Free Software Foundation; version 3.

  This program is distributed in the hope that it will be useful,
  but WITHOUT ANY WARRANTY; without even the implied warranty of
  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  GNU General Public License for more details.

  You should have received a copy of the GNU General Public License
  along with this program; if not, write to the Free Software
  Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA

  To the extent this program is licensed as part of the Enterprise
  versions of CFEngine, the applicable Commercial Open Source License
  (COSL) may apply to this file if you as a licensee so wish it. See
  included file COSL.txt.
*/

/*
 * Implementation using LMDB API.
 */

#include <cf3.defs.h>
#include <dbm_priv.h>
#include <logging.h>
#include <string_lib.h>
#include <misc_lib.h>
#include <file_lib.h>
#include <known_dirs.h>
#include <bootstrap.h>

#ifdef LMDB

#include <lmdb.h>
#include <repair.h>
#include <global_mutex.h> /* cf_db_corruption_lock */
#include <mutex.h>
#include <time.h>               /* time() */

// Shared between threads.
struct DBPriv_
{
    MDB_env *env;
    MDB_dbi dbi;
    // Used to keep track of transactions.
    // We set this to the transaction address when a thread creates a
    // transaction, and back to 0x0 when it is destroyed.
    pthread_key_t txn_key;
};

// Not shared between threads.
typedef struct DBTxn_
{
    MDB_txn *txn;
    // Whether txn is a read/write (true) or read-only (false) transaction.
    bool rw_txn;
    bool cursor_open;
} DBTxn;

struct DBCursorPriv_
{
    DBPriv *db;
    MDB_cursor *mc;
    MDB_val delkey;
    void *curkv;
    bool pending_delete;
};

static int DB_MAX_READERS = -1;

#define N_LMDB_EINVAL_RETRIES 5

/******************************************************************************/

static void HandleLMDBCorruption(MDB_env *env, const char *msg);

static inline void CheckLMDBCorrupted(int rc, MDB_env *env)
{
    if (rc == MDB_CORRUPTED)
    {
        HandleLMDBCorruption(env, "");
    }
}

static int GetReadTransaction(DBPriv *const db, DBTxn **const txn)
{
    assert(db != NULL);
    assert(txn != NULL);

    DBTxn *db_txn = pthread_getspecific(db->txn_key);
    int rc = MDB_SUCCESS;

    if (db_txn == NULL)
    {
        db_txn = xcalloc(1, sizeof(DBTxn));
        pthread_setspecific(db->txn_key, db_txn);
    }

    if (db_txn->txn == NULL)
    {
        rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &db_txn->txn);
        if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Unable to open read transaction in '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        }
    }

    *txn = db_txn;

    return rc;
}

static int GetWriteTransaction(DBPriv *const db, DBTxn **const txn)
{
    assert(db != NULL);
    assert(txn != NULL);

    DBTxn *db_txn = pthread_getspecific(db->txn_key);
    int rc = MDB_SUCCESS;

    if (db_txn == NULL)
    {
        db_txn = xcalloc(1, sizeof(DBTxn));
        pthread_setspecific(db->txn_key, db_txn);
    }

    if (db_txn->txn != NULL && !db_txn->rw_txn)
    {
        rc = mdb_txn_commit(db_txn->txn);
        CheckLMDBCorrupted(rc, db->env);
        if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Unable to close read-only transaction in '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        }
        db_txn->txn = NULL;
    }

    if (db_txn->txn == NULL)
    {
        rc = mdb_txn_begin(db->env, NULL, 0, &db_txn->txn);
        CheckLMDBCorrupted(rc, db->env);
        if (rc == MDB_SUCCESS)
        {
            db_txn->rw_txn = true;
        }
        else
        {
            Log(LOG_LEVEL_ERR, "Unable to open write transaction in '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        }
    }

    *txn = db_txn;

    return rc;
}

static void AbortTransaction(DBPriv *const db)
{
    assert(db != NULL);

    DBTxn *db_txn = pthread_getspecific(db->txn_key);
    if (db_txn != NULL)
    {
        if (db_txn->txn != NULL)
        {
            mdb_txn_abort(db_txn->txn);
        }

        pthread_setspecific(db->txn_key, NULL);
        free(db_txn);
    }
}

static void DestroyTransaction(void *const ptr)
{
    DBTxn *const db_txn = (DBTxn *)ptr;
    UnexpectedError("Transaction object still exists when terminating thread");
    if (db_txn->txn)
    {
        UnexpectedError("Transaction still open when terminating thread!");
        mdb_txn_abort(db_txn->txn);
    }
    free(db_txn);
}

const char *DBPrivGetFileExtension(void)
{
    return "lmdb";
}

#ifndef LMDB_MAXSIZE
#define LMDB_MAXSIZE    104857600
#endif

void DBPrivSetMaximumConcurrentTransactions(const int max_txn)
{
    DB_MAX_READERS = max_txn;
}

static int LmdbEnvOpen(
    MDB_env *const env,
    const char *const path,
    const unsigned int flags,
    const mdb_mode_t mode)
{
    assert(env != NULL);  // dereferenced in lmdb (mdb_env_open)
    assert(path != NULL); // dereferenced (strlen) in lmdb (mdb_env_open)
    assert(mdb_env_get_maxkeysize(env) == 511); // Search for 511 in locks.c

    /* There is a race condition in LMDB that will fail to open the database
     * environment if another process is opening it at the exact same time. This
     * condition is signaled by returning ENOENT, which we should never get
     * otherwise. This can lead to error messages on a heavily loaded machine,
     * so try to open it again after allowing other threads to finish their
     * opening process. */
    int attempts = 5;
    while (attempts-- > 0)
    {
        int rc = mdb_env_open(env, path, flags, mode);
        if (rc != ENOENT)
        {
            return rc;
        }

#if HAVE_DECL_SCHED_YIELD && defined(HAVE_SCHED_YIELD)
        // Not required for this to work, but makes it less likely that the race
        // condition will persist.
        sched_yield();
#endif
    }

    // Return EBUSY for an error message slightly more related to reality.
    return EBUSY;
}

/**
 * @warning Expects @fd_stamp to be locked.
 */
static bool RepairedAfterOpen(const char *lmdb_file, int fd_tstamp)
{
    time_t repaired_tstamp = -1;
    ssize_t n_read = read(fd_tstamp, &repaired_tstamp, sizeof(time_t));
    lseek(fd_tstamp, 0, SEEK_SET);

    if (n_read < 0)
    {
        Log(LOG_LEVEL_ERR, "Failed to read %s: %s", lmdb_file, GetErrorStr());
    }
    else if (n_read == 0)
    {
        /* EOF (empty file) => never repaired */
        Log(LOG_LEVEL_VERBOSE, "DB '%s' never repaired before", lmdb_file);
    }
    else if ((size_t) n_read < sizeof(time_t))
    {
        /* error */
        Log(LOG_LEVEL_ERR, "Failed to read the timestamp of repair of the '%s' DB",
            lmdb_file);
    }
    else
    {
        /* read the timestamp => Check if the LMDB file was repaired after
         * we opened it last time. Or, IOW, if this is a new corruption or
         * an already-handled one. */
        DBHandle *handle = GetDBHandleFromFilename(lmdb_file);
        if (repaired_tstamp > GetDBOpenTimestamp(handle))
        {
            return true;
        }
    }
    return false;
}

static void HandleLMDBCorruption(MDB_env *env, const char *msg)
{
    const char *lmdb_file = mdb_env_get_userctx(env);
    Log(LOG_LEVEL_CRIT, "Corruption in the '%s' DB detected! %s",
        lmdb_file, msg);

    /* Freeze the DB ASAP. This also makes the call to exit() safe regarding
     * this particular DB because exit handlers will ignore it. */
    DBHandle *handle = GetDBHandleFromFilename(lmdb_file);
    FreezeDB(handle);

#ifdef __MINGW32__
    /* Not much we can do on Windows because there is no fork() and file locking
     * is also not so nice. */
    Log(LOG_LEVEL_WARNING, "Removing the corrupted DB file '%s'",
        lmdb_file);
    if (unlink(lmdb_file) != 0)
    {
        Log(LOG_LEVEL_CRIT, "Failed to remove the corrupted DB file '%s'",
            lmdb_file);
        exit(EC_CORRUPTION_REPAIR_FAILED);
    }
    exit(EC_CORRUPTION_REPAIRED);
#else
    /* Try to handle the corruption gracefully by repairing the LMDB file
     * (replacing it with a new LMDB file with all the data we managed to read
     * from the corrupted one). */

    /* To avoid two processes acting on the same corrupted file at once, file
     * locks are involved. Looking at OpenDBInstance() and DBPathLock()
     * in libpromises/db_api.c might also be useful.*/

    /* Only allow one thread at a time to handle DB corruption. File locks are
     * *process* specific so threads could step on each others toes. */
    ThreadLock(cft_db_corruption_lock);

    char *tstamp_file = StringFormat("%s.repaired", lmdb_file);
    char *db_lock_file = StringFormat("%s.lock", lmdb_file);

    int fd_tstamp = safe_open(tstamp_file, O_CREAT|O_RDWR);
    if (fd_tstamp == -1)
    {
        Log(LOG_LEVEL_CRIT, "Failed to open the '%s' DB repair timestamp file",
            lmdb_file);
        ThreadUnlock(cft_db_corruption_lock);
        free(db_lock_file);
        free(tstamp_file);

        exit(EC_CORRUPTION_REPAIR_FAILED);
    }
    FileLock tstamp_lock = { .fd = fd_tstamp };

    int fd_db_lock = safe_open(db_lock_file, O_CREAT|O_RDWR);
    if (fd_db_lock == -1)
    {
        Log(LOG_LEVEL_CRIT, "Failed to open the '%s' DB lock file",
            lmdb_file);
        ThreadUnlock(cft_db_corruption_lock);
        close(fd_tstamp);
        free(db_lock_file);
        free(tstamp_file);

        exit(EC_CORRUPTION_REPAIR_FAILED);
    }
    FileLock db_lock = { .fd = fd_db_lock };

    int ret;
    bool handle_corruption = true;

    /* Make sure we are not holding the DB's lock (potentially needed by some
     * other process for the repair) to avoid deadlocks. */
    Log(LOG_LEVEL_DEBUG, "Releasing lock on the '%s' DB", lmdb_file);
    ExclusiveFileUnlock(&db_lock, false); /* close=false */

    ret = SharedFileLock(&tstamp_lock, true);
    if (ret == 0)
    {
        if (RepairedAfterOpen(lmdb_file, fd_tstamp))
        {
            /* The corruption has already been handled. This process should
             * just die because we have no way to return to the point where
             * it would just open the new (repaired) LMDB file. */
            handle_corruption = false;
        }
        SharedFileUnlock(&tstamp_lock, false);
    }
    else
    {
        /* should never happen (we tried to wait), but if it does, just log an
         * error and keep going */
        Log(LOG_LEVEL_ERR,
            "Failed to get shared lock for the repair timestamp of the '%s' DB",
            lmdb_file);
    }

    if (!handle_corruption)
    {
        /* Just clean after ourselves and terminate the process. */
        ThreadUnlock(cft_db_corruption_lock);
        close(fd_db_lock);
        close(fd_tstamp);
        free(db_lock_file);
        free(tstamp_file);

        exit(EC_CORRUPTION_REPAIRED);
    }

    /* HERE is a window for some other process to do the repair between when we
     * checked the timestamp using the shared lock above and the attempt to get
     * the exclusive lock right below. However, this is detected by checking the
     * contents of the timestamp file again below, while holding the EXCLUSIVE
     * lock. */

    ret = ExclusiveFileLock(&tstamp_lock, true);
    if (ret != 0)
    {
        /* should never happen (we tried to wait), but if it does, just
         * terminate because doing the repair without the lock could be
         * disasterous */
        Log(LOG_LEVEL_ERR,
            "Failed to get shared lock for the repair timestamp of the '%s' DB",
            lmdb_file);

        ThreadUnlock(cft_db_corruption_lock);
        close(fd_db_lock);
        close(fd_tstamp);
        free(db_lock_file);
        free(tstamp_file);

        exit(EC_CORRUPTION_REPAIR_FAILED);
    }

    /* Cleared to resolve the corruption. */

    /* 1. Acquire the lock for the DB to prevent more processes trying to use
     *    it while it is corrupted (wait till the lock is available). */
    while (ExclusiveFileLock(&db_lock, false) == -1)
    {
        /* busy wait to do the logging */
        Log(LOG_LEVEL_INFO, "Waiting for the lock on the '%s' DB",
            lmdb_file);
        sleep(1);
    }

    /* 2. Check the last repair timestamp again (see the big "HERE..." comment
     *    above) */
    if (RepairedAfterOpen(lmdb_file, fd_tstamp))
    {
        /* Some other process repaired the DB since we checked last time,
         * nothing more to do here. */
        ThreadUnlock(cft_db_corruption_lock);
        close(fd_db_lock);      /* releases locks */
        close(fd_tstamp);       /* releases locks */
        free(db_lock_file);
        free(tstamp_file);

        exit(EC_CORRUPTION_REPAIRED);
    }

    /* 3. Repair the DB or at least move it out of the way. */
    /* repair_lmdb_file() forks so it is safe (file locks are not
     * inherited). */
    ret = repair_lmdb_file(lmdb_file, fd_tstamp);

    /* repair_lmdb_file returns -1 in case of error, 0 in case of successfull
     * repair, >0 in case of failed repair, but successful remove */
    bool repair_successful = (ret != -1);
    if (repair_successful)
    {
        Log(LOG_LEVEL_NOTICE, "DB '%s' successfully repaired",
            lmdb_file);
    }
    else
    {
        Log(LOG_LEVEL_CRIT, "Failed to repair DB '%s'", lmdb_file);
    }

    /* 4. Make the repaired DB available for others. Also release the locks
     *    in the opposite order in which they were acquired to avoid
     *    deadlocks. */
    if (ExclusiveFileUnlock(&db_lock, true) != 0)
    {
        Log(LOG_LEVEL_ERR, "Failed to release the acquired lock for '%s'",
            db_lock_file);
    }

    /* 5. Signal that the repair is done (also closes fd_tstamp). */
    if (ExclusiveFileUnlock(&tstamp_lock, true) != 0)
    {
        Log(LOG_LEVEL_ERR, "Failed to release the acquired lock for '%s'",
            tstamp_file);
    }

    ThreadUnlock(cft_db_corruption_lock);
    free(db_lock_file);
    free(tstamp_file);
    /* fd_db_lock and fd_tstamp are already closed by the calls to
     * ExclusiveFileUnlock above. */

    if (repair_successful)
    {
        exit(EC_CORRUPTION_REPAIRED);
    }
    else
    {
        exit(EC_CORRUPTION_REPAIR_FAILED);
    }
#endif  /* __MINGW32__ */
}

DBPriv *DBPrivOpenDB(const char *const dbpath, const dbid id)
{
    DBPriv *const db = xcalloc(1, sizeof(DBPriv));
    MDB_txn *txn = NULL;

    int rc = pthread_key_create(&db->txn_key, &DestroyTransaction);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not create transaction key. (pthread_key_create: '%s')",
            GetErrorStrFromCode(rc));
        free(db);
        return NULL;
    }

    rc = mdb_env_create(&db->env);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not create handle for database %s: %s",
              dbpath, mdb_strerror(rc));
        goto err;
    }
    rc = mdb_env_set_userctx(db->env, xstrdup(dbpath));
    if (rc != MDB_SUCCESS)
    {
        Log(LOG_LEVEL_WARNING, "Could not store DB file path (%s) in the DB context",
            dbpath);
    }
    rc = mdb_env_set_assert(db->env, (MDB_assert_func*) HandleLMDBCorruption);
    if (rc != MDB_SUCCESS)
    {
        Log(LOG_LEVEL_WARNING, "Could not set the corruption handler for '%s'",
            dbpath);
    }
    rc = mdb_env_set_mapsize(db->env, LMDB_MAXSIZE);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not set mapsize for database %s: %s",
              dbpath, mdb_strerror(rc));
        goto err;
    }
    if (DB_MAX_READERS > 0)
    {
        rc = mdb_env_set_maxreaders(db->env, DB_MAX_READERS);
        if (rc)
        {
            Log(LOG_LEVEL_ERR, "Could not set maxreaders for database %s: %s",
                dbpath, mdb_strerror(rc));
            goto err;
        }
    }

    unsigned int open_flags = MDB_NOSUBDIR;
#if !defined(_AIX) && !defined(__sun)
    /* The locks and lastseen (on hubs) DBs are heavily used and using
     * MDB_NOSYNC increases performance. However, AIX and Solaris often suffer
     * from some serious issues with consistency (ENT-4002) so it's better to
     * sacrifice some performance there in favor of stability. */
    if (id == dbid_locks || (GetAmPolicyHub() && id == dbid_lastseen))
    {
        open_flags |= MDB_NOSYNC;
    }
#endif

#ifdef __hpux
    /*
     * On HP-UX, a unified file cache was not introduced until version 11.31.
     * This means that on 11.23 there are separate file caches for mmap()'ed
     * files and open()'ed files. When these two are mixed, changes made using
     * one mode won't be immediately seen by the other mode, which is an
     * assumption LMDB is relying on. The MDB_WRITEMAP flag causes LMDB to use
     * mmap() only, so that we stay within one file cache.
     */
    open_flags |= MDB_WRITEMAP;
#endif

    rc = LmdbEnvOpen(db->env, dbpath, open_flags, CF_PERMS_DEFAULT);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not open database %s: %s",
              dbpath, mdb_strerror(rc));
        if (rc == MDB_CORRUPTED || rc == MDB_INVALID)
        {
            HandleLMDBCorruption(db->env, mdb_strerror(rc));
        }
        goto err;
    }
    if (DB_MAX_READERS > 0)
    {
        int max_readers;
        rc = mdb_env_get_maxreaders(db->env, &max_readers);
        if (rc)
        {
            Log(LOG_LEVEL_ERR, "Could not get maxreaders for database %s: %s",
                dbpath, mdb_strerror(rc));
            goto err;
        }
        if (max_readers < DB_MAX_READERS)
        {
            // LMDB will only reinitialize maxreaders if no database handles are
            // open, including in other processes, which is how we might end up
            // here.
            Log(LOG_LEVEL_VERBOSE, "Failed to set LMDB max reader limit on database '%s', "
                "consider restarting CFEngine",
                dbpath);
        }
    }

    /* There seems to be a race condition causing mdb_txn_begin() return
     * EINVAL. We do a couple retries before giving up. */
    rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
    int attempts = N_LMDB_EINVAL_RETRIES;
    while ((rc != 0) && (attempts-- > 0))
    {
        CheckLMDBCorrupted(rc, db->env);
        if (rc != EINVAL)
        {
            Log(LOG_LEVEL_ERR, "Could not open database txn %s: %s",
                dbpath, mdb_strerror(rc));
            goto err;
        }
#if HAVE_DECL_SCHED_YIELD && defined(HAVE_SCHED_YIELD)
        // Not required for this to work, but makes it less likely that the race
        // condition will persist.
        sched_yield();
#endif
        rc = mdb_txn_begin(db->env, NULL, MDB_RDONLY, &txn);
    }
    if (rc != 0)
    {
        Log(LOG_LEVEL_ERR, "Could not open database txn %s: %s",
            dbpath, mdb_strerror(rc));
        goto err;
    }
    rc = mdb_open(txn, NULL, 0, &db->dbi);
    CheckLMDBCorrupted(rc, db->env);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not open database dbi %s: %s",
              dbpath, mdb_strerror(rc));
        mdb_txn_abort(txn);
        goto err;
    }
    rc = mdb_txn_commit(txn);
    CheckLMDBCorrupted(rc, db->env);
    if (rc)
    {
        Log(LOG_LEVEL_ERR, "Could not commit database dbi %s: %s",
              dbpath, mdb_strerror(rc));
        goto err;
    }

    return db;

err:
    if (db->env)
    {
        mdb_env_close(db->env);
    }
    pthread_key_delete(db->txn_key);
    free(db);
    if (rc == MDB_INVALID)
    {
        return DB_PRIV_DATABASE_BROKEN;
    }
    return NULL;
}

void DBPrivCloseDB(DBPriv *db)
{
    assert(db != NULL);

    /* Abort LMDB transaction of the current thread. There should only be some
     * transaction open when the signal handler or atexit() hook is called. */
    AbortTransaction(db);

    char *db_path = mdb_env_get_userctx(db->env);
    if (db_path)
    {
        free(db_path);
    }
    if (db->env)
    {
        mdb_env_close(db->env);
    }

    pthread_key_delete(db->txn_key);
    free(db);
}

#define EMPTY_DB 0

bool DBPrivClean(DBPriv *db)
{
    assert(db != NULL);

    DBTxn *txn;
    const int rc = GetWriteTransaction(db, &txn);

    if (rc != MDB_SUCCESS)
    {
        Log(LOG_LEVEL_ERR, "Unable to get write transaction for '%s': %s",
            (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        return false;
    }
    assert(txn != NULL);
    assert(!txn->cursor_open);

    return (mdb_drop(txn->txn, db->dbi, EMPTY_DB) != 0);
}

void DBPrivCommit(DBPriv *db)
{
    assert(db != NULL);

    DBTxn *db_txn = pthread_getspecific(db->txn_key);
    if (db_txn != NULL && db_txn->txn != NULL)
    {
        assert(!db_txn->cursor_open);
        const int rc = mdb_txn_commit(db_txn->txn);
        CheckLMDBCorrupted(rc, db->env);
        if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Could not commit database transaction to '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        }
    }
    pthread_setspecific(db->txn_key, NULL);
    free(db_txn);
}

bool DBPrivHasKey(DBPriv *db, const void *key, int key_size)
{
    assert(db != NULL);

    MDB_val mkey, data;
    DBTxn *txn;
    // FIXME: distinguish between "entry not found" and "error occurred"

    int rc = GetReadTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        assert(!txn->cursor_open);
        mkey.mv_data = (void *) key;
        mkey.mv_size = key_size;
        rc = mdb_get(txn->txn, db->dbi, &mkey, &data);
        CheckLMDBCorrupted(rc, db->env);
        if (rc != 0 && rc != MDB_NOTFOUND)
        {
            Log(LOG_LEVEL_ERR, "Could not read database entry from '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
    }

    return (rc == MDB_SUCCESS);
}

int DBPrivGetValueSize(DBPriv *const db, const void *const key, const int key_size)
{
    assert(db != NULL);
    assert(key_size >= 0);

    MDB_val mkey, data;
    DBTxn *txn;

    data.mv_size = 0;

    int rc = GetReadTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        assert(!txn->cursor_open);
        mkey.mv_data = (void *) key;
        mkey.mv_size = key_size;
        rc = mdb_get(txn->txn, db->dbi, &mkey, &data);
        CheckLMDBCorrupted(rc, db->env);
        if (rc && rc != MDB_NOTFOUND)
        {
            Log(LOG_LEVEL_ERR, "Could not read database entry from '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
    }

    size_t ret = data.mv_size;
    assert(ret <= INT_MAX);
    return ret;
}

bool DBPrivRead(
    DBPriv *const db,
    const void *const key,
    const int key_size,
    void *const dest,
    size_t dest_size)
{
    assert(db != NULL);
    assert(key_size >= 0);

    DBTxn *txn;
    bool ret = false;

    int rc = GetReadTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        MDB_val mkey, data;
        assert(txn != NULL);
        assert(!txn->cursor_open);
        mkey.mv_data = (void *) key;
        mkey.mv_size = key_size;
        rc = mdb_get(txn->txn, db->dbi, &mkey, &data);
        CheckLMDBCorrupted(rc, db->env);
        if (rc == MDB_SUCCESS)
        {
            if (dest_size > data.mv_size)
            {
                dest_size = data.mv_size;
            }
            memcpy(dest, data.mv_data, dest_size);
            ret = true;
        }
        else if (rc != MDB_NOTFOUND)
        {
            Log(LOG_LEVEL_ERR, "Could not read database entry from '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
    }
    return ret;
}

bool DBPrivWrite(
    DBPriv *const db,
    const void *const key,
    const int key_size,
    const void *const value,
    const int value_size)
{
    assert(db != NULL);
    assert(key_size >= 0);

    DBTxn *txn;
    int rc = GetWriteTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        MDB_val mkey, data;
        assert(txn != NULL);
        assert(!txn->cursor_open);
        mkey.mv_data = (void *) key;
        mkey.mv_size = key_size;
        data.mv_data = (void *)value;
        data.mv_size = value_size;
        rc = mdb_put(txn->txn, db->dbi, &mkey, &data, 0);
        CheckLMDBCorrupted(rc, db->env);
        if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Could not write database entry to '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
    }
    return (rc == MDB_SUCCESS);
}

bool DBPrivOverwrite(DBPriv *db, const char *key, int key_size, const void *value, size_t value_size,
                     OverwriteCondition Condition, void *data)
{
    assert(db != NULL);
    assert(key_size >= 0);
    DBTxn *txn;
    int rc = GetWriteTransaction(db, &txn);

    if (rc != MDB_SUCCESS)
    {
        return false;
    }

    assert(txn != NULL);
    assert(!txn->cursor_open);

    MDB_val mkey, orig_data;
    mkey.mv_data = (void *) key;
    mkey.mv_size = key_size;
    rc = mdb_get(txn->txn, db->dbi, &mkey, &orig_data);
    CheckLMDBCorrupted(rc, db->env);
    if ((rc != MDB_SUCCESS) && (rc != MDB_NOTFOUND))
    {
        Log(LOG_LEVEL_ERR, "Could not read database entry from '%s': %s",
            (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        AbortTransaction(db);
        return false;
    }

    if (Condition != NULL)
    {
        if (rc == MDB_SUCCESS)
        {
            assert(orig_data.mv_size > 0);

            /* We have to copy the data because orig_data.mv_data is a pointer to
             * the mmap()-ed area which can potentially have bad alignment causing
             * a SIGBUS on some architectures. */
            unsigned char cur_val[orig_data.mv_size];
            memcpy(cur_val, orig_data.mv_data, orig_data.mv_size);
            if (!Condition(cur_val, orig_data.mv_size, data))
            {
                AbortTransaction(db);
                return false;
            }
        }
        else
        {
            assert(rc == MDB_NOTFOUND);
            if (!Condition(NULL, 0, data))
            {
                AbortTransaction(db);
                return false;
            }
        }
    }

    MDB_val new_data;
    new_data.mv_data = (void *)value;
    new_data.mv_size = value_size;
    rc = mdb_put(txn->txn, db->dbi, &mkey, &new_data, 0);
    CheckLMDBCorrupted(rc, db->env);
    if (rc != MDB_SUCCESS)
    {
        Log(LOG_LEVEL_ERR, "Could not write database entry to '%s': %s",
            (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        AbortTransaction(db);
        return false;
    }
    DBPrivCommit(db);
    return true;
}

bool DBPrivDelete(DBPriv *const db, const void *const key, const int key_size)
{
    assert(key_size >= 0);
    assert(db != NULL);

    MDB_val mkey;
    DBTxn *txn;
    int rc = GetWriteTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        assert(!txn->cursor_open);
        mkey.mv_data = (void *) key;
        mkey.mv_size = key_size;
        rc = mdb_del(txn->txn, db->dbi, &mkey, NULL);
        CheckLMDBCorrupted(rc, db->env);
        if (rc == MDB_NOTFOUND)
        {
            Log(LOG_LEVEL_DEBUG, "Entry not found in '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
        }
        else if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Could not delete from '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
    }
    return (rc == MDB_SUCCESS);
}

DBCursorPriv *DBPrivOpenCursor(DBPriv *const db)
{
    assert(db != NULL);

    DBCursorPriv *cursor = NULL;
    DBTxn *txn;
    MDB_cursor *mc;

    int rc = GetWriteTransaction(db, &txn);
    if (rc == MDB_SUCCESS)
    {
        assert(!txn->cursor_open);
        rc = mdb_cursor_open(txn->txn, db->dbi, &mc);
        CheckLMDBCorrupted(rc, db->env);
        if (rc == MDB_SUCCESS)
        {
            cursor = xcalloc(1, sizeof(DBCursorPriv));
            cursor->db = db;
            cursor->mc = mc;
            txn->cursor_open = true;
        }
        else
        {
            Log(LOG_LEVEL_ERR, "Could not open cursor in '%s': %s",
                (char *) mdb_env_get_userctx(db->env), mdb_strerror(rc));
            AbortTransaction(db);
        }
        /* txn remains with cursor */
    }

    return cursor;
}

bool DBPrivAdvanceCursor(
    DBCursorPriv *const cursor,
    void **const key,
    int *const key_size,
    void **const value,
    int *const value_size)
{
    assert(cursor != NULL);
    assert(cursor->db != NULL);

    MDB_val mkey, data;
    bool retval = false;

    if (cursor->curkv != NULL)
    {
        free(cursor->curkv);
        cursor->curkv = NULL;
    }

    int rc = mdb_cursor_get(cursor->mc, &mkey, &data, MDB_NEXT);
    CheckLMDBCorrupted(rc, cursor->db->env);
    if (rc == MDB_SUCCESS)
    {
        // Align second buffer to 64-bit boundary, to avoid alignment errors on
        // certain platforms.
        size_t keybuf_size = mkey.mv_size;
        if (keybuf_size & 0x7)
        {
            keybuf_size += 8 - (keybuf_size % 8);
        }
        cursor->curkv = xmalloc(keybuf_size + data.mv_size);
        memcpy(cursor->curkv, mkey.mv_data, mkey.mv_size);
        *key = cursor->curkv;
        *key_size = mkey.mv_size;
        *value_size = data.mv_size;
        memcpy((char *) cursor->curkv + keybuf_size, data.mv_data, data.mv_size);
        *value = ((char *) cursor->curkv + keybuf_size);
        retval = true;
    }
    else if (rc != MDB_NOTFOUND)
    {
        Log(LOG_LEVEL_ERR, "Could not advance cursor in '%s': %s",
            (char *) mdb_env_get_userctx(cursor->db->env), mdb_strerror(rc));
    }
    if (cursor->pending_delete)
    {
        int r2;
        /* Position on key to delete */
        r2 = mdb_cursor_get(cursor->mc, &cursor->delkey, NULL, MDB_SET);
        if (r2 == MDB_SUCCESS)
        {
            r2 = mdb_cursor_del(cursor->mc, 0);
            // TODO: Should the return value be checked?
        }
        /* Reposition the cursor if it was valid before */
        if (rc == MDB_SUCCESS)
        {
            mkey.mv_data = *key;
            rc = mdb_cursor_get(cursor->mc, &mkey, NULL, MDB_SET);
            CheckLMDBCorrupted(rc, cursor->db->env);
            // TODO: Should the return value be checked?
        }
        cursor->pending_delete = false;
    }
    return retval;
}

bool DBPrivDeleteCursorEntry(DBCursorPriv *const cursor)
{
    assert(cursor != NULL);

    int rc = mdb_cursor_get(cursor->mc, &cursor->delkey, NULL, MDB_GET_CURRENT);
    CheckLMDBCorrupted(rc, cursor->db->env);
    if (rc == MDB_SUCCESS)
    {
        cursor->pending_delete = true;
    }
    return (rc == MDB_SUCCESS);
}

bool DBPrivWriteCursorEntry(
    DBCursorPriv *const cursor, const void *const value, const int value_size)
{
    assert(cursor != NULL);
    assert(cursor->db != NULL);

    MDB_val data;
    int rc;

    cursor->pending_delete = false;
    data.mv_data = (void *) value;
    data.mv_size = value_size;

    if (cursor->curkv)
    {
        MDB_val curkey;
        curkey.mv_data = cursor->curkv;
        curkey.mv_size = sizeof(cursor->curkv);

        rc = mdb_cursor_put(cursor->mc, &curkey, &data, MDB_CURRENT);
        CheckLMDBCorrupted(rc, cursor->db->env);
        if (rc != MDB_SUCCESS)
        {
            Log(LOG_LEVEL_ERR, "Could not write cursor entry to '%s': %s",
                (char *) mdb_env_get_userctx(cursor->db->env), mdb_strerror(rc));
        }
    }
    else
    {
        Log(LOG_LEVEL_ERR, "Could not write cursor entry to '%s': cannot find current key",
            (char *) mdb_env_get_userctx(cursor->db->env));
        rc = MDB_INVALID;
    }
    return (rc == MDB_SUCCESS);
}

void DBPrivCloseCursor(DBCursorPriv *const cursor)
{
    assert(cursor != NULL);
    assert(cursor->db != NULL);

    DBTxn *txn;
    const int rc = GetWriteTransaction(cursor->db, &txn);
    CF_ASSERT(rc == MDB_SUCCESS, "Could not get write transaction");
    CF_ASSERT(txn->cursor_open, "Transaction not open");
    txn->cursor_open = false;

    if (cursor->curkv)
    {
        free(cursor->curkv);
    }

    if (cursor->pending_delete)
    {
        mdb_cursor_del(cursor->mc, 0);
    }

    mdb_cursor_close(cursor->mc);
    free(cursor);
}

char *DBPrivDiagnose(const char *const dbpath)
{
    return StringFormat("Unable to diagnose LMDB file (not implemented) for '%s'", dbpath);
}
#endif