<commit_before>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2012 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "config.h"

#ifdef _MSC_VER
#include <direct.h>
#define mkdir(a, b) _mkdir(a)
#endif

#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>

#include <algorithm>
#include <cctype>
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <list>
#include <map>
#include <string>
#include <utility>
#include <vector>
#include <platform/dirutils.h>
#include <cJSON.h>

#include "common.h"
#include "couch-kvstore/couch-kvstore.h"
#define STATWRITER_NAMESPACE couchstore_engine
#include "statwriter.h"
#undef STATWRITER_NAMESPACE
#include "tools/JSON_checker.h"

using namespace CouchbaseDirectoryUtilities;

static const int MUTATION_FAILED = -1;
static const int DOC_NOT_FOUND = 0;
static const int MUTATION_SUCCESS = 1;

static const int MAX_OPEN_DB_RETRY = 10;

static const uint32_t DEFAULT_META_LEN = 16;

class NoLookupCallback : public Callback<CacheLookup> {
public:
    NoLookupCallback() {}
    ~NoLookupCallback() {}
    void callback(CacheLookup&) {}
};

extern "C" {
    static int recordDbDumpC(Db *db, DocInfo *docinfo, void *ctx)
    {
        return CouchKVStore::recordDbDump(db, docinfo, ctx);
    }
}

extern "C" {
    static int getMultiCbC(Db *db, DocInfo *docinfo, void *ctx)
    {
        return CouchKVStore::getMultiCb(db, docinfo, ctx);
    }
}

static std::string getStrError(Db *db) {
    const size_t max_msg_len = 256;
    char msg[max_msg_len];
    couchstore_last_os_error(db, msg, max_msg_len);
    std::string errorStr(msg);
    return errorStr;
}

static std::string getSystemStrerror(void) {
    std::stringstream ss;
#ifdef WIN32
    char* win_msg = NULL;
    DWORD err = GetLastError();
    FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER |
                   FORMAT_MESSAGE_FROM_SYSTEM |
                   FORMAT_MESSAGE_IGNORE_INSERTS,
                   NULL, err,
                   MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
                   (LPTSTR) &win_msg,
                   0, NULL);
    ss << "errno = " << err << ": '" << win_msg << "'";
    LocalFree(win_msg);
#else
    ss << "errno = " << errno << ": '" << strerror(errno) << "'";
#endif

    return ss.str();
}


static bool isJSON(const value_t &value)
{
    const int len = value->vlength();
    const unsigned char *data = (unsigned char*) value->getData();
    return checkUTF8JSON(data, len);
}

static const std::string getJSONObjString(const cJSON *i)
{
    if (i == NULL) {
        return "";
    }
    if (i->type != cJSON_String) {
        abort();
    }
    return i->valuestring;
}

static bool endWithCompact(const std::string &filename)
{
    size_t pos = filename.find(".compact");
    if (pos == std::string::npos || (filename.size() - sizeof(".compact")) != pos) {
        return false;
    }
    return true;
}

static void discoverDbFiles(const std::string &dir, std::vector<std::string> &v)
{
    std::vector<std::string> files = findFilesContaining(dir, ".couch");
    std::vector<std::string>::iterator ii;
    for (ii = files.begin(); ii != files.end(); ++ii) {
        if (!endWithCompact(*ii)) {
            v.push_back(*ii);
        }
    }
}

static uint64_t computeMaxDeletedSeqNum(DocInfo **docinfos, const int numdocs)
{
    uint64_t max = 0;
    for (int idx = 0; idx < numdocs; idx++) {
        if (docinfos[idx]->deleted) {
            // check seq number only from a deleted file
            uint64_t seqNum = docinfos[idx]->rev_seq;
            max = std::max(seqNum, max);
        }
    }
    return max;
}

static int getMutationStatus(couchstore_error_t errCode)
{
    switch (errCode) {
    case COUCHSTORE_SUCCESS:
        return MUTATION_SUCCESS;
    case COUCHSTORE_ERROR_NO_HEADER:
    case COUCHSTORE_ERROR_NO_SUCH_FILE:
    case COUCHSTORE_ERROR_DOC_NOT_FOUND:
        // this return causes ep engine to drop the failed flush
        // of an item since it does not know about the itme any longer
        return DOC_NOT_FOUND;
    default:
        // this return causes ep engine to keep requeuing the failed
        // flush of an item
        return MUTATION_FAILED;
    }
}

static bool allDigit(std::string &input)
{
    size_t numchar = input.length();
    for(size_t i = 0; i < numchar; ++i) {
        if (!isdigit(input[i])) {
            return false;
        }
    }
    return true;
}

static std::string couchkvstore_strerrno(Db *db, couchstore_error_t err) {
    return (err == COUCHSTORE_ERROR_OPEN_FILE ||
            err == COUCHSTORE_ERROR_READ ||
            err == COUCHSTORE_ERROR_WRITE) ? getStrError(db) : "none";
}

struct GetMultiCbCtx {
    GetMultiCbCtx(CouchKVStore &c, uint16_t v, vb_bgfetch_queue_t &f) :
        cks(c), vbId(v), fetches(f) {}

    CouchKVStore &cks;
    uint16_t vbId;
    vb_bgfetch_queue_t &fetches;
};

struct StatResponseCtx {
public:
    StatResponseCtx(std::map<std::pair<uint16_t, uint16_t>, vbucket_state> &sm,
                    uint16_t vb) : statMap(sm), vbId(vb) {
        /* EMPTY */
    }

    std::map<std::pair<uint16_t, uint16_t>, vbucket_state> &statMap;
    uint16_t vbId;
};

struct LoadResponseCtx {
    shared_ptr<Callback<GetValue> > callback;
    shared_ptr<Callback<CacheLookup> > lookup;
    uint16_t vbucketId;
    uint64_t endSeqno;
    bool keysonly;
    EPStats *stats;
};

struct AllKeysCtx {
    AllKeysCtx(AllKeysCB *callback, uint32_t cnt) :
        cb(callback), count(cnt) { }

    AllKeysCB *cb;
    uint32_t count;
};

CouchRequest::CouchRequest(const Item &it, uint64_t rev,
                           CouchRequestCallback &cb, bool del) :
    value(it.getValue()), vbucketId(it.getVBucketId()), fileRevNum(rev),
    key(it.getKey()), deleteItem(del)
{
    bool isjson = false;
    uint64_t cas = htonll(it.getCas());
    uint32_t flags = it.getFlags();
    uint32_t vlen = it.getNBytes();
    uint32_t exptime = it.getExptime();

    // Datatype used to determine whether document requires compression or not
    uint8_t datatype;

    // Save time of deletion in expiry time field of deleted item's metadata.
    if (del) {
        exptime = ep_real_time();
    }
    exptime = htonl(exptime);

    dbDoc.id.buf = const_cast<char *>(key.c_str());
    dbDoc.id.size = it.getNKey();
    if (vlen) {
        isjson = isJSON(value);
        dbDoc.data.buf = const_cast<char *>(value->getData());
        dbDoc.data.size = vlen;
        datatype = it.getDataType();
    } else {
        dbDoc.data.buf = NULL;
        dbDoc.data.size = 0;
        datatype = 0x00;
    }

    memset(meta, 0, sizeof(meta));
    memcpy(meta, &cas, 8);
    memcpy(meta + 8, &exptime, 4);
    memcpy(meta + 12, &flags, 4);
    *(meta + DEFAULT_META_LEN) = FLEX_META_CODE;
    memcpy(meta + DEFAULT_META_LEN + FLEX_DATA_OFFSET, it.getExtMeta(),
           it.getExtMetaLen());

    dbDocInfo.db_seq = it.getBySeqno();
    dbDocInfo.rev_meta.buf = reinterpret_cast<char *>(meta);
    dbDocInfo.rev_meta.size = COUCHSTORE_METADATA_SIZE;
    dbDocInfo.rev_seq = it.getRevSeqno();
    dbDocInfo.size = dbDoc.data.size;
    if (del) {
        dbDocInfo.deleted =  1;
        callback.delCb = cb.delCb;
    } else {
        dbDocInfo.deleted = 0;
        callback.setCb = cb.setCb;
    }
    dbDocInfo.id = dbDoc.id;
    dbDocInfo.content_meta = isjson ? COUCH_DOC_IS_JSON : COUCH_DOC_NON_JSON_MODE;

    //Compress only those documents that aren't already compressed.
    if (dbDoc.data.size > 0 && !deleteItem) {
        if (datatype == PROTOCOL_BINARY_RAW_BYTES ||
                datatype == PROTOCOL_BINARY_DATATYPE_JSON) {
            dbDocInfo.content_meta |= COUCH_DOC_IS_COMPRESSED;
        }
    }
    start = gethrtime();
}

CouchKVStore::CouchKVStore(EPStats &stats, Configuration &config, bool read_only) :
    KVStore(read_only), epStats(stats), configuration(config),
    dbname(configuration.getDbname()), couchNotifier(NULL),
    intransaction(false), dbFileRevMapPopulated(false)
{
    open();
    statCollectingFileOps = getCouchstoreStatsOps(&st.fsStats);

    // init db file map with default revision number, 1
    numDbFiles = static_cast<uint16_t>(configuration.getMaxVbuckets());
    for (uint16_t i = 0; i < numDbFiles; i++) {
        dbFileRevMap.push_back(1);
    }
}

CouchKVStore::CouchKVStore(const CouchKVStore &copyFrom) :
    KVStore(copyFrom), epStats(copyFrom.epStats),
    configuration(copyFrom.configuration),
    dbname(copyFrom.dbname),
    couchNotifier(NULL), dbFileRevMap(copyFrom.dbFileRevMap),
    numDbFiles(copyFrom.numDbFiles),
    intransaction(false), dbFileRevMapPopulated(true)
{
    open();
    statCollectingFileOps = getCouchstoreStatsOps(&st.fsStats);
}

void CouchKVStore::reset(bool notify)
{
    assert(!isReadOnly());
    // TODO CouchKVStore::flush() when couchstore api ready

    if (notify) {
        RememberingCallback<bool> cb;
        couchNotifier->flush(cb);
        cb.waitForValue();
    }

    vbucket_map_t::iterator itor = cachedVBStates.begin();
    for (; itor != cachedVBStates.end(); ++itor) {
        uint16_t vbucket = itor->first;
        itor->second.checkpointId = 0;
        itor->second.maxDeletedSeqno = 0;
        resetVBucket(vbucket, itor->second);
        updateDbFileMap(vbucket, 1);
    }
}

void CouchKVStore::set(const Item &itm, Callback<mutation_result> &cb)
{
    assert(!isReadOnly());
    assert(intransaction);
    bool deleteItem = false;
    CouchRequestCallback requestcb;
    std::string dbFile;
    uint64_t fileRev = dbFileRevMap[itm.getVBucketId()];

    // each req will be de-allocated after commit
    requestcb.setCb = &cb;
    CouchRequest *req = new CouchRequest(itm, fileRev, requestcb, deleteItem);
    pendingReqsQ.push_back(req);
}

void CouchKVStore::get(const std::string &key, uint64_t, uint16_t vb,
                       Callback<GetValue> &cb)
{
    Db *db = NULL;
    std::string dbFile;
    GetValue rv;
    uint64_t fileRev = dbFileRevMap[vb];

    couchstore_error_t errCode = openDB(vb, fileRev, &db,
                                        COUCHSTORE_OPEN_FLAG_RDONLY);
    if (errCode != COUCHSTORE_SUCCESS) {
        ++st.numGetFailure;
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to open database to retrieve data "
            "from vBucketId = %d, key = %s, file = %s\n",
            vb, key.c_str(), dbFile.c_str());
        rv.setStatus(couchErr2EngineErr(errCode));
        cb.callback(rv);
        return;
    }

    getWithHeader(db, key, vb, cb);
    closeDatabaseHandle(db);
}

void CouchKVStore::getWithHeader(void *dbHandle, const std::string &key,
                                 uint16_t vb, Callback<GetValue> &cb) {

    Db *db = (Db *)dbHandle;
    hrtime_t start = gethrtime();
    RememberingCallback<GetValue> *rc = dynamic_cast<RememberingCallback<GetValue> *>(&cb);
    bool getMetaOnly = rc && rc->val.isPartial();
    DocInfo *docInfo = NULL;
    sized_buf id;
    GetValue rv;
    std::string dbFile;

    id.size = key.size();
    id.buf = const_cast<char *>(key.c_str());
    couchstore_error_t errCode = couchstore_docinfo_by_id(db, (uint8_t *)id.buf,
                                                          id.size, &docInfo);
    if (errCode != COUCHSTORE_SUCCESS) {
        if (!getMetaOnly) {
            // log error only if this is non-xdcr case
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to retrieve doc info from "
                "database, name=%s key=%s error=%s [%s]\n",
                dbFile.c_str(), id.buf, couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str());
        }
    } else {
        assert(docInfo);
        errCode = fetchDoc(db, docInfo, rv, vb, getMetaOnly);
        if (errCode != COUCHSTORE_SUCCESS) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to retrieve key value from "
                "database, name=%s key=%s error=%s [%s] "
                "deleted=%s", dbFile.c_str(), id.buf,
                couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str(),
                docInfo->deleted ? "yes" : "no");
        }

        // record stats
        st.readTimeHisto.add((gethrtime() - start) / 1000);
        if (errCode == COUCHSTORE_SUCCESS) {
            st.readSizeHisto.add(key.length() + rv.getValue()->getNBytes());
        }
    }

    if(errCode != COUCHSTORE_SUCCESS) {
        ++st.numGetFailure;
    }

    couchstore_free_docinfo(docInfo);
    rv.setStatus(couchErr2EngineErr(errCode));
    cb.callback(rv);
}

void CouchKVStore::getMulti(uint16_t vb, vb_bgfetch_queue_t &itms)
{
    std::string dbFile;
    int numItems = itms.size();
    uint64_t fileRev = dbFileRevMap[vb];

    Db *db = NULL;
    couchstore_error_t errCode = openDB(vb, fileRev, &db,
                                        COUCHSTORE_OPEN_FLAG_RDONLY);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to open database for data fetch, "
            "vBucketId = %d file = %s numDocs = %d\n",
            vb, dbFile.c_str(), numItems);
        st.numGetFailure.fetch_add(numItems);
        vb_bgfetch_queue_t::iterator itr = itms.begin();
        for (; itr != itms.end(); ++itr) {
            std::list<VBucketBGFetchItem *> &fetches = (*itr).second;
            std::list<VBucketBGFetchItem *>::iterator fitr = fetches.begin();
            for (; fitr != fetches.end(); ++fitr) {
                (*fitr)->value.setStatus(ENGINE_NOT_MY_VBUCKET);
            }
        }
        return;
    }

    size_t idx = 0;
    sized_buf *ids = new sized_buf[itms.size()];
    vb_bgfetch_queue_t::iterator itr = itms.begin();
    for (; itr != itms.end(); ++itr) {
        ids[idx].size = itr->first.size();
        ids[idx].buf = const_cast<char *>(itr->first.c_str());
        ++idx;
    }

    GetMultiCbCtx ctx(*this, vb, itms);
    errCode = couchstore_docinfos_by_id(db, ids, itms.size(),
                                        0, getMultiCbC, &ctx);
    if (errCode != COUCHSTORE_SUCCESS) {
        st.numGetFailure.fetch_add(numItems);
        for (itr = itms.begin(); itr != itms.end(); ++itr) {
            LOG(EXTENSION_LOG_WARNING, "Warning: failed to read database by"
                " vBucketId = %d key = %s file = %s error = %s [%s]\n",
                vb, (*itr).first.c_str(),
                dbFile.c_str(), couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str());
            std::list<VBucketBGFetchItem *> &fetches = (*itr).second;
            std::list<VBucketBGFetchItem *>::iterator fitr = fetches.begin();
            for (; fitr != fetches.end(); ++fitr) {
                (*fitr)->value.setStatus(couchErr2EngineErr(errCode));
            }
        }
    }
    closeDatabaseHandle(db);
    delete []ids;
}

void CouchKVStore::del(const Item &itm,
                       Callback<int> &cb)
{
    assert(!isReadOnly());
    assert(intransaction);
    uint16_t fileRev = dbFileRevMap[itm.getVBucketId()];
    CouchRequestCallback requestcb;
    requestcb.delCb = &cb;
    CouchRequest *req = new CouchRequest(itm, fileRev, requestcb, true);
    pendingReqsQ.push_back(req);
}

bool CouchKVStore::delVBucket(uint16_t vbucket, bool recreate)
{
    assert(!isReadOnly());
    assert(couchNotifier);
    RememberingCallback<bool> cb;

    couchNotifier->delVBucket(vbucket, cb);
    cb.waitForValue();

    if (recreate) {
        vbucket_state vbstate(vbucket_state_dead, 0, 0, 0);
        vbucket_map_t::iterator it = cachedVBStates.find(vbucket);
        if (it != cachedVBStates.end()) {
            vbstate.state = it->second.state;
        }
        cachedVBStates[vbucket] = vbstate;
        resetVBucket(vbucket, vbstate);
    } else {
        cachedVBStates.erase(vbucket);
    }
    updateDbFileMap(vbucket, 1);
    return cb.val;
}

vbucket_map_t CouchKVStore::listPersistedVbuckets()
{
    std::vector<std::string> files;

    if (!dbFileRevMapPopulated) {
        // warmup, first discover db files from local directory
        discoverDbFiles(dbname, files);
        populateFileNameMap(files);
    }

    if (!cachedVBStates.empty()) {
        cachedVBStates.clear();
    }

    Db *db = NULL;
    couchstore_error_t errorCode;
    for (uint16_t id = 0; id < numDbFiles; id++) {
        uint64_t rev = dbFileRevMap[id];
        errorCode = openDB(id, rev, &db, COUCHSTORE_OPEN_FLAG_RDONLY);
        if (errorCode != COUCHSTORE_SUCCESS) {
            std::stringstream revnum, vbid;
            revnum  << rev;
            vbid << id;
            std::string fileName =
                dbname + "/" + vbid.str() + ".couch." + revnum.str();
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database file, name=%s\n",
                fileName.c_str());
            remVBucketFromDbFileMap(id);
        } else {
            vbucket_state vb_state;

            /* read state of VBucket from db file */
            readVBState(db, id, vb_state);
            /* insert populated state to the array to return to the caller */
            cachedVBStates[id] = vb_state;
            /* update stat */
            ++st.numLoadedVb;
            closeDatabaseHandle(db);
        }
        db = NULL;
    }
    return cachedVBStates;
}

void CouchKVStore::getPersistedStats(std::map<std::string, std::string> &stats)
{
    char *buffer = NULL;
    std::string fname = dbname + "/stats.json";
    if (access(fname.c_str(), R_OK) == -1) {
        return ;
    }

    std::ifstream session_stats;
    session_stats.exceptions (session_stats.failbit | session_stats.badbit);
    try {
        session_stats.open(fname.c_str(), std::ios::binary);
        session_stats.seekg(0, std::ios::end);
        int flen = session_stats.tellg();
        if (flen < 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: error in session stats ifstream!!!");
            session_stats.close();
            return;
        }
        session_stats.seekg(0, std::ios::beg);
        buffer = new char[flen + 1];
        session_stats.read(buffer, flen);
        session_stats.close();
        buffer[flen] = '\0';

        cJSON *json_obj = cJSON_Parse(buffer);
        if (!json_obj) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to parse the session stats json doc!!!");
            delete[] buffer;
            return;
        }

        int json_arr_size = cJSON_GetArraySize(json_obj);
        for (int i = 0; i < json_arr_size; ++i) {
            cJSON *obj = cJSON_GetArrayItem(json_obj, i);
            if (obj) {
                stats[obj->string] = obj->valuestring ? obj->valuestring : "";
            }
        }
        cJSON_Delete(json_obj);

    } catch (const std::ifstream::failure &e) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to load the engine session stats "
            " due to IO exception \"%s\"", e.what());
    } catch (...) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to load the engine session stats "
            " due to IO exception");
    }

    delete[] buffer;
}

static std::string getDBFileName(const std::string &dbname,
                                 uint16_t vbid,
                                 uint64_t rev)
{
    std::stringstream ss;
    ss << dbname << "/" << vbid << ".couch." << rev;
    return ss.str();
}

static int time_purge_hook(Db* d, DocInfo* info, void* ctx_p) {
    compaction_ctx* ctx = (compaction_ctx*) ctx_p;

    //Compaction finished
    if (info == NULL) {
        return couchstore_set_purge_seq(d, ctx->max_purged_seq);
    }

    if (info->rev_meta.size >= DEFAULT_META_LEN) {
        uint32_t exptime;
        memcpy(&exptime, info->rev_meta.buf + 8, 4);
        exptime = ntohl(exptime);
        if (info->deleted) {
            if (exptime < ctx->purge_before_ts &&
                    (!ctx->purge_before_seq ||
                     info->db_seq <= ctx->purge_before_seq)) {
                if (ctx->max_purged_seq < info->db_seq) {
                    ctx->max_purged_seq = info->db_seq;
                }
                return COUCHSTORE_COMPACT_DROP_ITEM;
            }
        } else if (exptime && exptime < ctx->curr_time) {
            std::string keyStr(info->id.buf, info->id.size);
            expiredItemCtx expItem = { info->rev_seq, keyStr };
            ctx->expiredItems.push_back(expItem);
        }
    }

    return COUCHSTORE_COMPACT_KEEP_ITEM;
}

bool CouchKVStore::compactVBucket(const uint16_t vbid,
                                  compaction_ctx *hook_ctx,
                                  Callback<compaction_ctx> &cb,
                                  Callback<kvstats_ctx> &kvcb) {
    couchstore_compact_flags     flags = hook_ctx->drop_deletes ?
                                         COUCHSTORE_COMPACT_FLAG_DROP_DELETES :
                                         0;
    couchstore_compact_hook       hook = time_purge_hook;
    const couch_file_ops     *def_iops = couchstore_get_default_file_ops();
    Db                      *compactdb = NULL;
    Db                       *targetDb = NULL;
    uint64_t                   fileRev = dbFileRevMap[vbid];
    uint64_t                   new_rev = fileRev + 1;
    couchstore_error_t         errCode = COUCHSTORE_SUCCESS;
    hrtime_t                     start = gethrtime();
    uint64_t              newHeaderPos = 0;
    std::string                 dbfile;
    std::string           compact_file;
    std::string               new_file;
    kvstats_ctx                  kvctx;
    DbInfo                        info;

    // Open the source VBucket database file ...
    errCode = openDB(vbid, fileRev, &compactdb,
                     (uint64_t)COUCHSTORE_OPEN_FLAG_RDONLY, NULL);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database, vbucketId = %d "
                "fileRev = %llu", vbid, fileRev);
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Build the temporary vbucket.compact file name
    dbfile       = getDBFileName(dbname, vbid, fileRev);
    compact_file = dbfile + ".compact";

    // Perform COMPACTION of vbucket.couch.rev into vbucket.couch.rev.compact
    errCode = couchstore_compact_db_ex(compactdb, compact_file.c_str(), flags,
                                       hook, hook_ctx, def_iops);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to compact database with name=%s "
            "error=%s errno=%s",
            dbfile.c_str(),
            couchstore_strerror(errCode),
            couchkvstore_strerrno(compactdb, errCode).c_str());
        closeDatabaseHandle(compactdb);
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Close the source Database File once compaction is done
    closeDatabaseHandle(compactdb);

    // Rename the .compact file to one with the next revision number
    new_file = getDBFileName(dbname, vbid, new_rev);
    if (rename(compact_file.c_str(), new_file.c_str()) != 0) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to rename '%s' to '%s': %s",
            compact_file.c_str(), new_file.c_str(), getSystemStrerror().c_str());
        if (remove(compact_file.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: Failed to remove '%s': %s",
                compact_file.c_str(), getSystemStrerror().c_str());
        }
        return notifyCompaction(vbid, new_rev, VB_COMPACT_RENAME_ERROR, 0);
    }

    // Open the newly compacted VBucket database file ...
    errCode = openDB(vbid, new_rev, &targetDb,
                     (uint64_t)COUCHSTORE_OPEN_FLAG_RDONLY, NULL);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open compacted database file %s "
                "fileRev = %llu", new_file.c_str(), new_rev);
        if (remove(new_file.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING, NULL,
                "Warning: Failed to remove '%s': %s",
                new_file.c_str(), getSystemStrerror().c_str());
        }
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Update the global VBucket file map so all operations use the new file
    updateDbFileMap(vbid, new_rev);

    LOG(EXTENSION_LOG_INFO,
            "INFO: created new couch db file, name=%s rev=%llu",
            new_file.c_str(), new_rev);

    // Update stats to caller
    kvctx.vbucket = vbid;
    couchstore_db_info(targetDb, &info);
    kvctx.fileSpaceUsed = info.space_used;
    kvctx.fileSize = info.file_size;
    kvcb.callback(kvctx);

    // Notify MCCouch that compaction is Done...
    newHeaderPos = couchstore_get_header_position(targetDb);
    closeDatabaseHandle(targetDb);

    bool retVal = notifyCompaction(vbid, new_rev, VB_COMPACTION_DONE,
                                   newHeaderPos);

    if (hook_ctx->expiredItems.size()) {
        cb.callback(*hook_ctx);
    }

    st.compactHisto.add((gethrtime() - start) / 1000);
    return retVal;
}

bool CouchKVStore::notifyCompaction(const uint16_t vbid, uint64_t new_rev,
                                    uint32_t result, uint64_t header_pos) {
    RememberingCallback<uint16_t> lcb;

    VBStateNotification vbs(0, 0, result, vbid);

    couchNotifier->notify_update(vbs, new_rev, header_pos, lcb);
    if (lcb.val != PROTOCOL_BINARY_RESPONSE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: compactor failed to notify mccouch on vbucket "
                "%d. err %d", vbid, lcb.val);
        return false;
    }
    return true;
}

bool CouchKVStore::snapshotVBuckets(const vbucket_map_t &vbstates,
                                    Callback<kvstats_ctx> *cb)
{
    assert(!isReadOnly());
    bool success = true;

    vbucket_map_t::const_reverse_iterator iter = vbstates.rbegin();
    for (; iter != vbstates.rend(); ++iter) {
        bool notify = false;
        uint16_t vbucketId = iter->first;
        vbucket_state vbstate = iter->second;
        vbucket_map_t::iterator it = cachedVBStates.find(vbucketId);
        uint32_t vb_change_type = VB_NO_CHANGE;
        if (it != cachedVBStates.end()) {
            if (it->second.state != vbstate.state) {
                vb_change_type |= VB_STATE_CHANGED;
                notify = true;
            }
            if (it->second.checkpointId != vbstate.checkpointId) {
                vb_change_type |= VB_CHECKPOINT_CHANGED;
                notify = true;
            }

            if (it->second.failovers.compare(vbstate.failovers) == 0 &&
                vb_change_type == VB_NO_CHANGE) {
                continue; // no changes
            }
            it->second.state = vbstate.state;
            it->second.checkpointId = vbstate.checkpointId;
            it->second.failovers = vbstate.failovers;
            // Note that the max deleted seq number is maintained within CouchKVStore
            vbstate.maxDeletedSeqno = it->second.maxDeletedSeqno;
        } else {
            vb_change_type = VB_STATE_CHANGED;
            cachedVBStates[vbucketId] = vbstate;
            notify = true;
        }

        success = setVBucketState(vbucketId, vbstate, vb_change_type, cb,
                                  notify);
        if (!success) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to set new state, %s, for vbucket %d\n",
                VBucket::toString(vbstate.state), vbucketId);
            break;
        }
    }
    return success;
}

bool CouchKVStore::snapshotStats(const std::map<std::string, std::string> &stats)
{
    assert(!isReadOnly());
    size_t count = 0;
    size_t size = stats.size();
    std::stringstream stats_buf;
    stats_buf << "{";
    std::map<std::string, std::string>::const_iterator it = stats.begin();
    for (; it != stats.end(); ++it) {
        stats_buf << "\"" << it->first << "\": \"" << it->second << "\"";
        ++count;
        if (count < size) {
            stats_buf << ", ";
        }
    }
    stats_buf << "}";

    // TODO: This stats json should be written into the master database. However,
    // we don't support the write synchronization between CouchKVStore in C++ and
    // compaction manager in the erlang side for the master database yet. At this time,
    // we simply log the engine stats into a separate json file. As part of futhre work,
    // we need to get rid of the tight coupling between those two components.
    bool rv = true;
    std::string next_fname = dbname + "/stats.json.new";
    std::ofstream new_stats;
    new_stats.exceptions (new_stats.failbit | new_stats.badbit);
    try {
        new_stats.open(next_fname.c_str());
        new_stats << stats_buf.str().c_str() << std::endl;
        new_stats.flush();
        new_stats.close();
    } catch (const std::ofstream::failure& e) {
        LOG(EXTENSION_LOG_WARNING, "Warning: failed to log the engine stats due"
            " to IO exception \"%s\"", e.what());
        rv = false;
    }

    if (rv) {
        std::string old_fname = dbname + "/stats.json.old";
        std::string stats_fname = dbname + "/stats.json";
        if (access(old_fname.c_str(), F_OK) == 0 && remove(old_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING, "FATAL: Failed to remove '%s': %s",
                old_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        } else if (access(stats_fname.c_str(), F_OK) == 0 &&
                   rename(stats_fname.c_str(), old_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to rename '%s' to '%s': %s",
                stats_fname.c_str(), old_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        } else if (rename(next_fname.c_str(), stats_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to rename '%s' to '%s': %s",
                next_fname.c_str(), stats_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        }
    }

    return rv;
}

bool CouchKVStore::setVBucketState(uint16_t vbucketId, vbucket_state &vbstate,
                                   uint32_t vb_change_type,
                                   Callback<kvstats_ctx> *kvcb,
                                   bool notify)
{
    Db *db = NULL;
    uint64_t fileRev, newFileRev;
    std::stringstream id;
    std::string dbFileName;
    std::map<uint16_t, uint64_t>::iterator mapItr;
    kvstats_ctx kvctx;
    kvctx.vbucket = vbucketId;

    id << vbucketId;
    dbFileName = dbname + "/" + id.str() + ".couch." + id.str();
    fileRev = dbFileRevMap[vbucketId];

    couchstore_error_t errorCode;
    bool retry = true;
    while (retry) {
        retry = false;
        errorCode = openDB(vbucketId, fileRev, &db,
                           (uint64_t)COUCHSTORE_OPEN_FLAG_CREATE, &newFileRev);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database, name=%s",
                dbFileName.c_str());
            return false;
        }

        fileRev = newFileRev;
        errorCode = saveVBState(db, vbstate);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to save local doc, name=%s",
                dbFileName.c_str());
            closeDatabaseHandle(db);
            return false;
        }

        errorCode = couchstore_commit(db);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: commit failed, vbid=%u rev=%llu error=%s [%s]",
                vbucketId, fileRev, couchstore_strerror(errorCode),
                couchkvstore_strerrno(db, errorCode).c_str());
            closeDatabaseHandle(db);
            return false;
        } else {
            if (notify) {
                uint64_t newHeaderPos = couchstore_get_header_position(db);
                RememberingCallback<uint16_t> lcb;

                VBStateNotification vbs(vbstate.checkpointId, vbstate.state,
                        vb_change_type, vbucketId);

                couchNotifier->notify_update(vbs, fileRev, newHeaderPos, lcb);
                if (lcb.val != PROTOCOL_BINARY_RESPONSE_SUCCESS) {
                    if (lcb.val == PROTOCOL_BINARY_RESPONSE_ETMPFAIL) {
                        LOG(EXTENSION_LOG_WARNING,
                                "Retry notify CouchDB of update, "
                                "vbid=%u rev=%llu\n", vbucketId, fileRev);
                        retry = true;
                    } else {
                        LOG(EXTENSION_LOG_WARNING,
                                "Warning: failed to notify CouchDB of update, "
                                "vbid=%u rev=%llu error=0x%x\n",
                                vbucketId, fileRev, lcb.val);
                        if (!epStats.isShutdown) {
                            closeDatabaseHandle(db);
                            return false;
                        }
                    }
                }
            }
        }
        if (kvcb) {
            DbInfo info;
            couchstore_db_info(db, &info);
            kvctx.fileSpaceUsed = info.space_used;
            kvctx.fileSize = info.file_size;
            kvcb->callback(kvctx);
        }
        closeDatabaseHandle(db);
    }

    return true;
}

void CouchKVStore::dump(std::vector<uint16_t> &vbids,
                        shared_ptr<Callback<GetValue> > cb,
                        shared_ptr<Callback<CacheLookup> > cl)
{
    std::vector<uint16_t>::iterator itr = vbids.begin();
    for (; itr != vbids.end(); ++itr) {
        loadDB(cb, cl, false, *itr, 0, std::numeric_limits<uint64_t>::max(),
               COUCHSTORE_NO_DELETES);
    }
}

void CouchKVStore::dump(uint16_t vb, uint64_t stSeqno, uint64_t enSeqno,
                        shared_ptr<Callback<GetValue> > cb,
                        shared_ptr<Callback<CacheLookup> > cl)
{
    std::vector<uint16_t> vbids;
    vbids.push_back(vb);
    loadDB(cb, cl, false, vb, stSeqno, enSeqno);
}

void CouchKVStore::dumpKeys(std::vector<uint16_t> &vbids,  shared_ptr<Callback<GetValue> > cb)
{
    shared_ptr<Callback<CacheLookup> > cl(new NoLookupCallback());
    std::vector<uint16_t>::iterator itr = vbids.begin();
    for (; itr != vbids.end(); ++itr) {
        loadDB(cb, cl, true, *itr, 0, std::numeric_limits<uint64_t>::max(),
               COUCHSTORE_NO_DELETES);
    }
}

void CouchKVStore::dumpDeleted(uint16_t vb, uint64_t stSeqno, uint64_t enSeqno,
                               shared_ptr<Callback<GetValue> > cb)
{
    std::vector<uint16_t> vbids;
    vbids.push_back(vb);
    shared_ptr<Callback<CacheLookup> > cl(new NoLookupCallback());
    loadDB(cb, cl, true, vb, stSeqno, enSeqno, COUCHSTORE_DELETES_ONLY);
}

StorageProperties CouchKVStore::getStorageProperties()
{
    StorageProperties rv(true, true, true, true);
    return rv;
}

bool CouchKVStore::commit(Callback<kvstats_ctx> *cb)
{
    assert(!isReadOnly());
    if (intransaction) {
        intransaction = commit2couchstore(cb) ? false : true;
    }
    return !intransaction;

}

void CouchKVStore::addStats(const std::string &prefix,
                            ADD_STAT add_stat,
                            const void *c)
{
    const char *prefix_str = prefix.c_str();

    /* stats for both read-only and read-write threads */
    addStat(prefix_str, "backend_type",   "couchstore",       add_stat, c);
    addStat(prefix_str, "open",           st.numOpen,         add_stat, c);
    addStat(prefix_str, "close",          st.numClose,        add_stat, c);
    addStat(prefix_str, "readTime",       st.readTimeHisto,   add_stat, c);
    addStat(prefix_str, "readSize",       st.readSizeHisto,   add_stat, c);
    addStat(prefix_str, "numLoadedVb",    st.numLoadedVb,     add_stat, c);

    // failure stats
    addStat(prefix_str, "failure_open",   st.numOpenFailure, add_stat, c);
    addStat(prefix_str, "failure_get",    st.numGetFailure,  add_stat, c);

    if (!isReadOnly()) {
        addStat(prefix_str, "failure_set",   st.numSetFailure,   add_stat, c);
        addStat(prefix_str, "failure_del",   st.numDelFailure,   add_stat, c);
        addStat(prefix_str, "failure_vbset", st.numVbSetFailure, add_stat, c);
        addStat(prefix_str, "lastCommDocs",  st.docsCommitted,   add_stat, c);
        addStat(prefix_str, "numCommitRetry", st.numCommitRetry, add_stat, c);

        // stats for CouchNotifier
        couchNotifier->addStats(prefix, add_stat, c);
    }
}

void CouchKVStore::addTimingStats(const std::string &prefix,
                                  ADD_STAT add_stat, const void *c) {
    if (isReadOnly()) {
        return;
    }
    const char *prefix_str = prefix.c_str();
    addStat(prefix_str, "commit",      st.commitHisto,      add_stat, c);
    addStat(prefix_str, "commitRetry", st.commitRetryHisto, add_stat, c);
    addStat(prefix_str, "delete",      st.delTimeHisto,     add_stat, c);
    addStat(prefix_str, "save_documents", st.saveDocsHisto, add_stat, c);
    addStat(prefix_str, "writeTime",   st.writeTimeHisto,   add_stat, c);
    addStat(prefix_str, "writeSize",   st.writeSizeHisto,   add_stat, c);
    addStat(prefix_str, "bulkSize",    st.batchSize,        add_stat, c);

    // Couchstore file ops stats
    addStat(prefix_str, "fsReadTime",  st.fsStats.readTimeHisto,  add_stat, c);
    addStat(prefix_str, "fsWriteTime", st.fsStats.writeTimeHisto, add_stat, c);
    addStat(prefix_str, "fsSyncTime",  st.fsStats.syncTimeHisto,  add_stat, c);<commit_msg>Add compaction times histogram to kvtimings<commit_after>/* -*- Mode: C++; tab-width: 4; c-basic-offset: 4; indent-tabs-mode: nil -*- */
/*
 *     Copyright 2012 Couchbase, Inc
 *
 *   Licensed under the Apache License, Version 2.0 (the "License");
 *   you may not use this file except in compliance with the License.
 *   You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *   Unless required by applicable law or agreed to in writing, software
 *   distributed under the License is distributed on an "AS IS" BASIS,
 *   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *   See the License for the specific language governing permissions and
 *   limitations under the License.
 */

#include "config.h"

#ifdef _MSC_VER
#include <direct.h>
#define mkdir(a, b) _mkdir(a)
#endif

#include <fcntl.h>
#include <stdio.h>
#include <string.h>
#include <sys/stat.h>
#include <sys/types.h>

#include <algorithm>
#include <cctype>
#include <cstdlib>
#include <fstream>
#include <iostream>
#include <list>
#include <map>
#include <string>
#include <utility>
#include <vector>
#include <platform/dirutils.h>
#include <cJSON.h>

#include "common.h"
#include "couch-kvstore/couch-kvstore.h"
#define STATWRITER_NAMESPACE couchstore_engine
#include "statwriter.h"
#undef STATWRITER_NAMESPACE
#include "tools/JSON_checker.h"

using namespace CouchbaseDirectoryUtilities;

static const int MUTATION_FAILED = -1;
static const int DOC_NOT_FOUND = 0;
static const int MUTATION_SUCCESS = 1;

static const int MAX_OPEN_DB_RETRY = 10;

static const uint32_t DEFAULT_META_LEN = 16;

class NoLookupCallback : public Callback<CacheLookup> {
public:
    NoLookupCallback() {}
    ~NoLookupCallback() {}
    void callback(CacheLookup&) {}
};

extern "C" {
    static int recordDbDumpC(Db *db, DocInfo *docinfo, void *ctx)
    {
        return CouchKVStore::recordDbDump(db, docinfo, ctx);
    }
}

extern "C" {
    static int getMultiCbC(Db *db, DocInfo *docinfo, void *ctx)
    {
        return CouchKVStore::getMultiCb(db, docinfo, ctx);
    }
}

static std::string getStrError(Db *db) {
    const size_t max_msg_len = 256;
    char msg[max_msg_len];
    couchstore_last_os_error(db, msg, max_msg_len);
    std::string errorStr(msg);
    return errorStr;
}

static std::string getSystemStrerror(void) {
    std::stringstream ss;
#ifdef WIN32
    char* win_msg = NULL;
    DWORD err = GetLastError();
    FormatMessageA(FORMAT_MESSAGE_ALLOCATE_BUFFER |
                   FORMAT_MESSAGE_FROM_SYSTEM |
                   FORMAT_MESSAGE_IGNORE_INSERTS,
                   NULL, err,
                   MAKELANGID(LANG_NEUTRAL, SUBLANG_DEFAULT),
                   (LPTSTR) &win_msg,
                   0, NULL);
    ss << "errno = " << err << ": '" << win_msg << "'";
    LocalFree(win_msg);
#else
    ss << "errno = " << errno << ": '" << strerror(errno) << "'";
#endif

    return ss.str();
}


static bool isJSON(const value_t &value)
{
    const int len = value->vlength();
    const unsigned char *data = (unsigned char*) value->getData();
    return checkUTF8JSON(data, len);
}

static const std::string getJSONObjString(const cJSON *i)
{
    if (i == NULL) {
        return "";
    }
    if (i->type != cJSON_String) {
        abort();
    }
    return i->valuestring;
}

static bool endWithCompact(const std::string &filename)
{
    size_t pos = filename.find(".compact");
    if (pos == std::string::npos || (filename.size() - sizeof(".compact")) != pos) {
        return false;
    }
    return true;
}

static void discoverDbFiles(const std::string &dir, std::vector<std::string> &v)
{
    std::vector<std::string> files = findFilesContaining(dir, ".couch");
    std::vector<std::string>::iterator ii;
    for (ii = files.begin(); ii != files.end(); ++ii) {
        if (!endWithCompact(*ii)) {
            v.push_back(*ii);
        }
    }
}

static uint64_t computeMaxDeletedSeqNum(DocInfo **docinfos, const int numdocs)
{
    uint64_t max = 0;
    for (int idx = 0; idx < numdocs; idx++) {
        if (docinfos[idx]->deleted) {
            // check seq number only from a deleted file
            uint64_t seqNum = docinfos[idx]->rev_seq;
            max = std::max(seqNum, max);
        }
    }
    return max;
}

static int getMutationStatus(couchstore_error_t errCode)
{
    switch (errCode) {
    case COUCHSTORE_SUCCESS:
        return MUTATION_SUCCESS;
    case COUCHSTORE_ERROR_NO_HEADER:
    case COUCHSTORE_ERROR_NO_SUCH_FILE:
    case COUCHSTORE_ERROR_DOC_NOT_FOUND:
        // this return causes ep engine to drop the failed flush
        // of an item since it does not know about the itme any longer
        return DOC_NOT_FOUND;
    default:
        // this return causes ep engine to keep requeuing the failed
        // flush of an item
        return MUTATION_FAILED;
    }
}

static bool allDigit(std::string &input)
{
    size_t numchar = input.length();
    for(size_t i = 0; i < numchar; ++i) {
        if (!isdigit(input[i])) {
            return false;
        }
    }
    return true;
}

static std::string couchkvstore_strerrno(Db *db, couchstore_error_t err) {
    return (err == COUCHSTORE_ERROR_OPEN_FILE ||
            err == COUCHSTORE_ERROR_READ ||
            err == COUCHSTORE_ERROR_WRITE) ? getStrError(db) : "none";
}

struct GetMultiCbCtx {
    GetMultiCbCtx(CouchKVStore &c, uint16_t v, vb_bgfetch_queue_t &f) :
        cks(c), vbId(v), fetches(f) {}

    CouchKVStore &cks;
    uint16_t vbId;
    vb_bgfetch_queue_t &fetches;
};

struct StatResponseCtx {
public:
    StatResponseCtx(std::map<std::pair<uint16_t, uint16_t>, vbucket_state> &sm,
                    uint16_t vb) : statMap(sm), vbId(vb) {
        /* EMPTY */
    }

    std::map<std::pair<uint16_t, uint16_t>, vbucket_state> &statMap;
    uint16_t vbId;
};

struct LoadResponseCtx {
    shared_ptr<Callback<GetValue> > callback;
    shared_ptr<Callback<CacheLookup> > lookup;
    uint16_t vbucketId;
    uint64_t endSeqno;
    bool keysonly;
    EPStats *stats;
};

struct AllKeysCtx {
    AllKeysCtx(AllKeysCB *callback, uint32_t cnt) :
        cb(callback), count(cnt) { }

    AllKeysCB *cb;
    uint32_t count;
};

CouchRequest::CouchRequest(const Item &it, uint64_t rev,
                           CouchRequestCallback &cb, bool del) :
    value(it.getValue()), vbucketId(it.getVBucketId()), fileRevNum(rev),
    key(it.getKey()), deleteItem(del)
{
    bool isjson = false;
    uint64_t cas = htonll(it.getCas());
    uint32_t flags = it.getFlags();
    uint32_t vlen = it.getNBytes();
    uint32_t exptime = it.getExptime();

    // Datatype used to determine whether document requires compression or not
    uint8_t datatype;

    // Save time of deletion in expiry time field of deleted item's metadata.
    if (del) {
        exptime = ep_real_time();
    }
    exptime = htonl(exptime);

    dbDoc.id.buf = const_cast<char *>(key.c_str());
    dbDoc.id.size = it.getNKey();
    if (vlen) {
        isjson = isJSON(value);
        dbDoc.data.buf = const_cast<char *>(value->getData());
        dbDoc.data.size = vlen;
        datatype = it.getDataType();
    } else {
        dbDoc.data.buf = NULL;
        dbDoc.data.size = 0;
        datatype = 0x00;
    }

    memset(meta, 0, sizeof(meta));
    memcpy(meta, &cas, 8);
    memcpy(meta + 8, &exptime, 4);
    memcpy(meta + 12, &flags, 4);
    *(meta + DEFAULT_META_LEN) = FLEX_META_CODE;
    memcpy(meta + DEFAULT_META_LEN + FLEX_DATA_OFFSET, it.getExtMeta(),
           it.getExtMetaLen());

    dbDocInfo.db_seq = it.getBySeqno();
    dbDocInfo.rev_meta.buf = reinterpret_cast<char *>(meta);
    dbDocInfo.rev_meta.size = COUCHSTORE_METADATA_SIZE;
    dbDocInfo.rev_seq = it.getRevSeqno();
    dbDocInfo.size = dbDoc.data.size;
    if (del) {
        dbDocInfo.deleted =  1;
        callback.delCb = cb.delCb;
    } else {
        dbDocInfo.deleted = 0;
        callback.setCb = cb.setCb;
    }
    dbDocInfo.id = dbDoc.id;
    dbDocInfo.content_meta = isjson ? COUCH_DOC_IS_JSON : COUCH_DOC_NON_JSON_MODE;

    //Compress only those documents that aren't already compressed.
    if (dbDoc.data.size > 0 && !deleteItem) {
        if (datatype == PROTOCOL_BINARY_RAW_BYTES ||
                datatype == PROTOCOL_BINARY_DATATYPE_JSON) {
            dbDocInfo.content_meta |= COUCH_DOC_IS_COMPRESSED;
        }
    }
    start = gethrtime();
}

CouchKVStore::CouchKVStore(EPStats &stats, Configuration &config, bool read_only) :
    KVStore(read_only), epStats(stats), configuration(config),
    dbname(configuration.getDbname()), couchNotifier(NULL),
    intransaction(false), dbFileRevMapPopulated(false)
{
    open();
    statCollectingFileOps = getCouchstoreStatsOps(&st.fsStats);

    // init db file map with default revision number, 1
    numDbFiles = static_cast<uint16_t>(configuration.getMaxVbuckets());
    for (uint16_t i = 0; i < numDbFiles; i++) {
        dbFileRevMap.push_back(1);
    }
}

CouchKVStore::CouchKVStore(const CouchKVStore &copyFrom) :
    KVStore(copyFrom), epStats(copyFrom.epStats),
    configuration(copyFrom.configuration),
    dbname(copyFrom.dbname),
    couchNotifier(NULL), dbFileRevMap(copyFrom.dbFileRevMap),
    numDbFiles(copyFrom.numDbFiles),
    intransaction(false), dbFileRevMapPopulated(true)
{
    open();
    statCollectingFileOps = getCouchstoreStatsOps(&st.fsStats);
}

void CouchKVStore::reset(bool notify)
{
    assert(!isReadOnly());
    // TODO CouchKVStore::flush() when couchstore api ready

    if (notify) {
        RememberingCallback<bool> cb;
        couchNotifier->flush(cb);
        cb.waitForValue();
    }

    vbucket_map_t::iterator itor = cachedVBStates.begin();
    for (; itor != cachedVBStates.end(); ++itor) {
        uint16_t vbucket = itor->first;
        itor->second.checkpointId = 0;
        itor->second.maxDeletedSeqno = 0;
        resetVBucket(vbucket, itor->second);
        updateDbFileMap(vbucket, 1);
    }
}

void CouchKVStore::set(const Item &itm, Callback<mutation_result> &cb)
{
    assert(!isReadOnly());
    assert(intransaction);
    bool deleteItem = false;
    CouchRequestCallback requestcb;
    std::string dbFile;
    uint64_t fileRev = dbFileRevMap[itm.getVBucketId()];

    // each req will be de-allocated after commit
    requestcb.setCb = &cb;
    CouchRequest *req = new CouchRequest(itm, fileRev, requestcb, deleteItem);
    pendingReqsQ.push_back(req);
}

void CouchKVStore::get(const std::string &key, uint64_t, uint16_t vb,
                       Callback<GetValue> &cb)
{
    Db *db = NULL;
    std::string dbFile;
    GetValue rv;
    uint64_t fileRev = dbFileRevMap[vb];

    couchstore_error_t errCode = openDB(vb, fileRev, &db,
                                        COUCHSTORE_OPEN_FLAG_RDONLY);
    if (errCode != COUCHSTORE_SUCCESS) {
        ++st.numGetFailure;
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to open database to retrieve data "
            "from vBucketId = %d, key = %s, file = %s\n",
            vb, key.c_str(), dbFile.c_str());
        rv.setStatus(couchErr2EngineErr(errCode));
        cb.callback(rv);
        return;
    }

    getWithHeader(db, key, vb, cb);
    closeDatabaseHandle(db);
}

void CouchKVStore::getWithHeader(void *dbHandle, const std::string &key,
                                 uint16_t vb, Callback<GetValue> &cb) {

    Db *db = (Db *)dbHandle;
    hrtime_t start = gethrtime();
    RememberingCallback<GetValue> *rc = dynamic_cast<RememberingCallback<GetValue> *>(&cb);
    bool getMetaOnly = rc && rc->val.isPartial();
    DocInfo *docInfo = NULL;
    sized_buf id;
    GetValue rv;
    std::string dbFile;

    id.size = key.size();
    id.buf = const_cast<char *>(key.c_str());
    couchstore_error_t errCode = couchstore_docinfo_by_id(db, (uint8_t *)id.buf,
                                                          id.size, &docInfo);
    if (errCode != COUCHSTORE_SUCCESS) {
        if (!getMetaOnly) {
            // log error only if this is non-xdcr case
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to retrieve doc info from "
                "database, name=%s key=%s error=%s [%s]\n",
                dbFile.c_str(), id.buf, couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str());
        }
    } else {
        assert(docInfo);
        errCode = fetchDoc(db, docInfo, rv, vb, getMetaOnly);
        if (errCode != COUCHSTORE_SUCCESS) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to retrieve key value from "
                "database, name=%s key=%s error=%s [%s] "
                "deleted=%s", dbFile.c_str(), id.buf,
                couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str(),
                docInfo->deleted ? "yes" : "no");
        }

        // record stats
        st.readTimeHisto.add((gethrtime() - start) / 1000);
        if (errCode == COUCHSTORE_SUCCESS) {
            st.readSizeHisto.add(key.length() + rv.getValue()->getNBytes());
        }
    }

    if(errCode != COUCHSTORE_SUCCESS) {
        ++st.numGetFailure;
    }

    couchstore_free_docinfo(docInfo);
    rv.setStatus(couchErr2EngineErr(errCode));
    cb.callback(rv);
}

void CouchKVStore::getMulti(uint16_t vb, vb_bgfetch_queue_t &itms)
{
    std::string dbFile;
    int numItems = itms.size();
    uint64_t fileRev = dbFileRevMap[vb];

    Db *db = NULL;
    couchstore_error_t errCode = openDB(vb, fileRev, &db,
                                        COUCHSTORE_OPEN_FLAG_RDONLY);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to open database for data fetch, "
            "vBucketId = %d file = %s numDocs = %d\n",
            vb, dbFile.c_str(), numItems);
        st.numGetFailure.fetch_add(numItems);
        vb_bgfetch_queue_t::iterator itr = itms.begin();
        for (; itr != itms.end(); ++itr) {
            std::list<VBucketBGFetchItem *> &fetches = (*itr).second;
            std::list<VBucketBGFetchItem *>::iterator fitr = fetches.begin();
            for (; fitr != fetches.end(); ++fitr) {
                (*fitr)->value.setStatus(ENGINE_NOT_MY_VBUCKET);
            }
        }
        return;
    }

    size_t idx = 0;
    sized_buf *ids = new sized_buf[itms.size()];
    vb_bgfetch_queue_t::iterator itr = itms.begin();
    for (; itr != itms.end(); ++itr) {
        ids[idx].size = itr->first.size();
        ids[idx].buf = const_cast<char *>(itr->first.c_str());
        ++idx;
    }

    GetMultiCbCtx ctx(*this, vb, itms);
    errCode = couchstore_docinfos_by_id(db, ids, itms.size(),
                                        0, getMultiCbC, &ctx);
    if (errCode != COUCHSTORE_SUCCESS) {
        st.numGetFailure.fetch_add(numItems);
        for (itr = itms.begin(); itr != itms.end(); ++itr) {
            LOG(EXTENSION_LOG_WARNING, "Warning: failed to read database by"
                " vBucketId = %d key = %s file = %s error = %s [%s]\n",
                vb, (*itr).first.c_str(),
                dbFile.c_str(), couchstore_strerror(errCode),
                couchkvstore_strerrno(db, errCode).c_str());
            std::list<VBucketBGFetchItem *> &fetches = (*itr).second;
            std::list<VBucketBGFetchItem *>::iterator fitr = fetches.begin();
            for (; fitr != fetches.end(); ++fitr) {
                (*fitr)->value.setStatus(couchErr2EngineErr(errCode));
            }
        }
    }
    closeDatabaseHandle(db);
    delete []ids;
}

void CouchKVStore::del(const Item &itm,
                       Callback<int> &cb)
{
    assert(!isReadOnly());
    assert(intransaction);
    uint16_t fileRev = dbFileRevMap[itm.getVBucketId()];
    CouchRequestCallback requestcb;
    requestcb.delCb = &cb;
    CouchRequest *req = new CouchRequest(itm, fileRev, requestcb, true);
    pendingReqsQ.push_back(req);
}

bool CouchKVStore::delVBucket(uint16_t vbucket, bool recreate)
{
    assert(!isReadOnly());
    assert(couchNotifier);
    RememberingCallback<bool> cb;

    couchNotifier->delVBucket(vbucket, cb);
    cb.waitForValue();

    if (recreate) {
        vbucket_state vbstate(vbucket_state_dead, 0, 0, 0);
        vbucket_map_t::iterator it = cachedVBStates.find(vbucket);
        if (it != cachedVBStates.end()) {
            vbstate.state = it->second.state;
        }
        cachedVBStates[vbucket] = vbstate;
        resetVBucket(vbucket, vbstate);
    } else {
        cachedVBStates.erase(vbucket);
    }
    updateDbFileMap(vbucket, 1);
    return cb.val;
}

vbucket_map_t CouchKVStore::listPersistedVbuckets()
{
    std::vector<std::string> files;

    if (!dbFileRevMapPopulated) {
        // warmup, first discover db files from local directory
        discoverDbFiles(dbname, files);
        populateFileNameMap(files);
    }

    if (!cachedVBStates.empty()) {
        cachedVBStates.clear();
    }

    Db *db = NULL;
    couchstore_error_t errorCode;
    for (uint16_t id = 0; id < numDbFiles; id++) {
        uint64_t rev = dbFileRevMap[id];
        errorCode = openDB(id, rev, &db, COUCHSTORE_OPEN_FLAG_RDONLY);
        if (errorCode != COUCHSTORE_SUCCESS) {
            std::stringstream revnum, vbid;
            revnum  << rev;
            vbid << id;
            std::string fileName =
                dbname + "/" + vbid.str() + ".couch." + revnum.str();
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database file, name=%s\n",
                fileName.c_str());
            remVBucketFromDbFileMap(id);
        } else {
            vbucket_state vb_state;

            /* read state of VBucket from db file */
            readVBState(db, id, vb_state);
            /* insert populated state to the array to return to the caller */
            cachedVBStates[id] = vb_state;
            /* update stat */
            ++st.numLoadedVb;
            closeDatabaseHandle(db);
        }
        db = NULL;
    }
    return cachedVBStates;
}

void CouchKVStore::getPersistedStats(std::map<std::string, std::string> &stats)
{
    char *buffer = NULL;
    std::string fname = dbname + "/stats.json";
    if (access(fname.c_str(), R_OK) == -1) {
        return ;
    }

    std::ifstream session_stats;
    session_stats.exceptions (session_stats.failbit | session_stats.badbit);
    try {
        session_stats.open(fname.c_str(), std::ios::binary);
        session_stats.seekg(0, std::ios::end);
        int flen = session_stats.tellg();
        if (flen < 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: error in session stats ifstream!!!");
            session_stats.close();
            return;
        }
        session_stats.seekg(0, std::ios::beg);
        buffer = new char[flen + 1];
        session_stats.read(buffer, flen);
        session_stats.close();
        buffer[flen] = '\0';

        cJSON *json_obj = cJSON_Parse(buffer);
        if (!json_obj) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to parse the session stats json doc!!!");
            delete[] buffer;
            return;
        }

        int json_arr_size = cJSON_GetArraySize(json_obj);
        for (int i = 0; i < json_arr_size; ++i) {
            cJSON *obj = cJSON_GetArrayItem(json_obj, i);
            if (obj) {
                stats[obj->string] = obj->valuestring ? obj->valuestring : "";
            }
        }
        cJSON_Delete(json_obj);

    } catch (const std::ifstream::failure &e) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to load the engine session stats "
            " due to IO exception \"%s\"", e.what());
    } catch (...) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to load the engine session stats "
            " due to IO exception");
    }

    delete[] buffer;
}

static std::string getDBFileName(const std::string &dbname,
                                 uint16_t vbid,
                                 uint64_t rev)
{
    std::stringstream ss;
    ss << dbname << "/" << vbid << ".couch." << rev;
    return ss.str();
}

static int time_purge_hook(Db* d, DocInfo* info, void* ctx_p) {
    compaction_ctx* ctx = (compaction_ctx*) ctx_p;

    //Compaction finished
    if (info == NULL) {
        return couchstore_set_purge_seq(d, ctx->max_purged_seq);
    }

    if (info->rev_meta.size >= DEFAULT_META_LEN) {
        uint32_t exptime;
        memcpy(&exptime, info->rev_meta.buf + 8, 4);
        exptime = ntohl(exptime);
        if (info->deleted) {
            if (exptime < ctx->purge_before_ts &&
                    (!ctx->purge_before_seq ||
                     info->db_seq <= ctx->purge_before_seq)) {
                if (ctx->max_purged_seq < info->db_seq) {
                    ctx->max_purged_seq = info->db_seq;
                }
                return COUCHSTORE_COMPACT_DROP_ITEM;
            }
        } else if (exptime && exptime < ctx->curr_time) {
            std::string keyStr(info->id.buf, info->id.size);
            expiredItemCtx expItem = { info->rev_seq, keyStr };
            ctx->expiredItems.push_back(expItem);
        }
    }

    return COUCHSTORE_COMPACT_KEEP_ITEM;
}

bool CouchKVStore::compactVBucket(const uint16_t vbid,
                                  compaction_ctx *hook_ctx,
                                  Callback<compaction_ctx> &cb,
                                  Callback<kvstats_ctx> &kvcb) {
    couchstore_compact_flags     flags = hook_ctx->drop_deletes ?
                                         COUCHSTORE_COMPACT_FLAG_DROP_DELETES :
                                         0;
    couchstore_compact_hook       hook = time_purge_hook;
    const couch_file_ops     *def_iops = couchstore_get_default_file_ops();
    Db                      *compactdb = NULL;
    Db                       *targetDb = NULL;
    uint64_t                   fileRev = dbFileRevMap[vbid];
    uint64_t                   new_rev = fileRev + 1;
    couchstore_error_t         errCode = COUCHSTORE_SUCCESS;
    hrtime_t                     start = gethrtime();
    uint64_t              newHeaderPos = 0;
    std::string                 dbfile;
    std::string           compact_file;
    std::string               new_file;
    kvstats_ctx                  kvctx;
    DbInfo                        info;

    // Open the source VBucket database file ...
    errCode = openDB(vbid, fileRev, &compactdb,
                     (uint64_t)COUCHSTORE_OPEN_FLAG_RDONLY, NULL);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database, vbucketId = %d "
                "fileRev = %llu", vbid, fileRev);
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Build the temporary vbucket.compact file name
    dbfile       = getDBFileName(dbname, vbid, fileRev);
    compact_file = dbfile + ".compact";

    // Perform COMPACTION of vbucket.couch.rev into vbucket.couch.rev.compact
    errCode = couchstore_compact_db_ex(compactdb, compact_file.c_str(), flags,
                                       hook, hook_ctx, def_iops);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to compact database with name=%s "
            "error=%s errno=%s",
            dbfile.c_str(),
            couchstore_strerror(errCode),
            couchkvstore_strerrno(compactdb, errCode).c_str());
        closeDatabaseHandle(compactdb);
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Close the source Database File once compaction is done
    closeDatabaseHandle(compactdb);

    // Rename the .compact file to one with the next revision number
    new_file = getDBFileName(dbname, vbid, new_rev);
    if (rename(compact_file.c_str(), new_file.c_str()) != 0) {
        LOG(EXTENSION_LOG_WARNING,
            "Warning: failed to rename '%s' to '%s': %s",
            compact_file.c_str(), new_file.c_str(), getSystemStrerror().c_str());
        if (remove(compact_file.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: Failed to remove '%s': %s",
                compact_file.c_str(), getSystemStrerror().c_str());
        }
        return notifyCompaction(vbid, new_rev, VB_COMPACT_RENAME_ERROR, 0);
    }

    // Open the newly compacted VBucket database file ...
    errCode = openDB(vbid, new_rev, &targetDb,
                     (uint64_t)COUCHSTORE_OPEN_FLAG_RDONLY, NULL);
    if (errCode != COUCHSTORE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open compacted database file %s "
                "fileRev = %llu", new_file.c_str(), new_rev);
        if (remove(new_file.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING, NULL,
                "Warning: Failed to remove '%s': %s",
                new_file.c_str(), getSystemStrerror().c_str());
        }
        return notifyCompaction(vbid, new_rev, VB_COMPACT_OPENDB_ERROR, 0);
    }

    // Update the global VBucket file map so all operations use the new file
    updateDbFileMap(vbid, new_rev);

    LOG(EXTENSION_LOG_INFO,
            "INFO: created new couch db file, name=%s rev=%llu",
            new_file.c_str(), new_rev);

    // Update stats to caller
    kvctx.vbucket = vbid;
    couchstore_db_info(targetDb, &info);
    kvctx.fileSpaceUsed = info.space_used;
    kvctx.fileSize = info.file_size;
    kvcb.callback(kvctx);

    // Notify MCCouch that compaction is Done...
    newHeaderPos = couchstore_get_header_position(targetDb);
    closeDatabaseHandle(targetDb);

    bool retVal = notifyCompaction(vbid, new_rev, VB_COMPACTION_DONE,
                                   newHeaderPos);

    if (hook_ctx->expiredItems.size()) {
        cb.callback(*hook_ctx);
    }

    st.compactHisto.add((gethrtime() - start) / 1000);
    return retVal;
}

bool CouchKVStore::notifyCompaction(const uint16_t vbid, uint64_t new_rev,
                                    uint32_t result, uint64_t header_pos) {
    RememberingCallback<uint16_t> lcb;

    VBStateNotification vbs(0, 0, result, vbid);

    couchNotifier->notify_update(vbs, new_rev, header_pos, lcb);
    if (lcb.val != PROTOCOL_BINARY_RESPONSE_SUCCESS) {
        LOG(EXTENSION_LOG_WARNING,
                "Warning: compactor failed to notify mccouch on vbucket "
                "%d. err %d", vbid, lcb.val);
        return false;
    }
    return true;
}

bool CouchKVStore::snapshotVBuckets(const vbucket_map_t &vbstates,
                                    Callback<kvstats_ctx> *cb)
{
    assert(!isReadOnly());
    bool success = true;

    vbucket_map_t::const_reverse_iterator iter = vbstates.rbegin();
    for (; iter != vbstates.rend(); ++iter) {
        bool notify = false;
        uint16_t vbucketId = iter->first;
        vbucket_state vbstate = iter->second;
        vbucket_map_t::iterator it = cachedVBStates.find(vbucketId);
        uint32_t vb_change_type = VB_NO_CHANGE;
        if (it != cachedVBStates.end()) {
            if (it->second.state != vbstate.state) {
                vb_change_type |= VB_STATE_CHANGED;
                notify = true;
            }
            if (it->second.checkpointId != vbstate.checkpointId) {
                vb_change_type |= VB_CHECKPOINT_CHANGED;
                notify = true;
            }

            if (it->second.failovers.compare(vbstate.failovers) == 0 &&
                vb_change_type == VB_NO_CHANGE) {
                continue; // no changes
            }
            it->second.state = vbstate.state;
            it->second.checkpointId = vbstate.checkpointId;
            it->second.failovers = vbstate.failovers;
            // Note that the max deleted seq number is maintained within CouchKVStore
            vbstate.maxDeletedSeqno = it->second.maxDeletedSeqno;
        } else {
            vb_change_type = VB_STATE_CHANGED;
            cachedVBStates[vbucketId] = vbstate;
            notify = true;
        }

        success = setVBucketState(vbucketId, vbstate, vb_change_type, cb,
                                  notify);
        if (!success) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to set new state, %s, for vbucket %d\n",
                VBucket::toString(vbstate.state), vbucketId);
            break;
        }
    }
    return success;
}

bool CouchKVStore::snapshotStats(const std::map<std::string, std::string> &stats)
{
    assert(!isReadOnly());
    size_t count = 0;
    size_t size = stats.size();
    std::stringstream stats_buf;
    stats_buf << "{";
    std::map<std::string, std::string>::const_iterator it = stats.begin();
    for (; it != stats.end(); ++it) {
        stats_buf << "\"" << it->first << "\": \"" << it->second << "\"";
        ++count;
        if (count < size) {
            stats_buf << ", ";
        }
    }
    stats_buf << "}";

    // TODO: This stats json should be written into the master database. However,
    // we don't support the write synchronization between CouchKVStore in C++ and
    // compaction manager in the erlang side for the master database yet. At this time,
    // we simply log the engine stats into a separate json file. As part of futhre work,
    // we need to get rid of the tight coupling between those two components.
    bool rv = true;
    std::string next_fname = dbname + "/stats.json.new";
    std::ofstream new_stats;
    new_stats.exceptions (new_stats.failbit | new_stats.badbit);
    try {
        new_stats.open(next_fname.c_str());
        new_stats << stats_buf.str().c_str() << std::endl;
        new_stats.flush();
        new_stats.close();
    } catch (const std::ofstream::failure& e) {
        LOG(EXTENSION_LOG_WARNING, "Warning: failed to log the engine stats due"
            " to IO exception \"%s\"", e.what());
        rv = false;
    }

    if (rv) {
        std::string old_fname = dbname + "/stats.json.old";
        std::string stats_fname = dbname + "/stats.json";
        if (access(old_fname.c_str(), F_OK) == 0 && remove(old_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING, "FATAL: Failed to remove '%s': %s",
                old_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        } else if (access(stats_fname.c_str(), F_OK) == 0 &&
                   rename(stats_fname.c_str(), old_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to rename '%s' to '%s': %s",
                stats_fname.c_str(), old_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        } else if (rename(next_fname.c_str(), stats_fname.c_str()) != 0) {
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to rename '%s' to '%s': %s",
                next_fname.c_str(), stats_fname.c_str(), strerror(errno));
            remove(next_fname.c_str());
            rv = false;
        }
    }

    return rv;
}

bool CouchKVStore::setVBucketState(uint16_t vbucketId, vbucket_state &vbstate,
                                   uint32_t vb_change_type,
                                   Callback<kvstats_ctx> *kvcb,
                                   bool notify)
{
    Db *db = NULL;
    uint64_t fileRev, newFileRev;
    std::stringstream id;
    std::string dbFileName;
    std::map<uint16_t, uint64_t>::iterator mapItr;
    kvstats_ctx kvctx;
    kvctx.vbucket = vbucketId;

    id << vbucketId;
    dbFileName = dbname + "/" + id.str() + ".couch." + id.str();
    fileRev = dbFileRevMap[vbucketId];

    couchstore_error_t errorCode;
    bool retry = true;
    while (retry) {
        retry = false;
        errorCode = openDB(vbucketId, fileRev, &db,
                           (uint64_t)COUCHSTORE_OPEN_FLAG_CREATE, &newFileRev);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to open database, name=%s",
                dbFileName.c_str());
            return false;
        }

        fileRev = newFileRev;
        errorCode = saveVBState(db, vbstate);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: failed to save local doc, name=%s",
                dbFileName.c_str());
            closeDatabaseHandle(db);
            return false;
        }

        errorCode = couchstore_commit(db);
        if (errorCode != COUCHSTORE_SUCCESS) {
            ++st.numVbSetFailure;
            LOG(EXTENSION_LOG_WARNING,
                "Warning: commit failed, vbid=%u rev=%llu error=%s [%s]",
                vbucketId, fileRev, couchstore_strerror(errorCode),
                couchkvstore_strerrno(db, errorCode).c_str());
            closeDatabaseHandle(db);
            return false;
        } else {
            if (notify) {
                uint64_t newHeaderPos = couchstore_get_header_position(db);
                RememberingCallback<uint16_t> lcb;

                VBStateNotification vbs(vbstate.checkpointId, vbstate.state,
                        vb_change_type, vbucketId);

                couchNotifier->notify_update(vbs, fileRev, newHeaderPos, lcb);
                if (lcb.val != PROTOCOL_BINARY_RESPONSE_SUCCESS) {
                    if (lcb.val == PROTOCOL_BINARY_RESPONSE_ETMPFAIL) {
                        LOG(EXTENSION_LOG_WARNING,
                                "Retry notify CouchDB of update, "
                                "vbid=%u rev=%llu\n", vbucketId, fileRev);
                        retry = true;
                    } else {
                        LOG(EXTENSION_LOG_WARNING,
                                "Warning: failed to notify CouchDB of update, "
                                "vbid=%u rev=%llu error=0x%x\n",
                                vbucketId, fileRev, lcb.val);
                        if (!epStats.isShutdown) {
                            closeDatabaseHandle(db);
                            return false;
                        }
                    }
                }
            }
        }
        if (kvcb) {
            DbInfo info;
            couchstore_db_info(db, &info);
            kvctx.fileSpaceUsed = info.space_used;
            kvctx.fileSize = info.file_size;
            kvcb->callback(kvctx);
        }
        closeDatabaseHandle(db);
    }

    return true;
}

void CouchKVStore::dump(std::vector<uint16_t> &vbids,
                        shared_ptr<Callback<GetValue> > cb,
                        shared_ptr<Callback<CacheLookup> > cl)
{
    std::vector<uint16_t>::iterator itr = vbids.begin();
    for (; itr != vbids.end(); ++itr) {
        loadDB(cb, cl, false, *itr, 0, std::numeric_limits<uint64_t>::max(),
               COUCHSTORE_NO_DELETES);
    }
}

void CouchKVStore::dump(uint16_t vb, uint64_t stSeqno, uint64_t enSeqno,
                        shared_ptr<Callback<GetValue> > cb,
                        shared_ptr<Callback<CacheLookup> > cl)
{
    std::vector<uint16_t> vbids;
    vbids.push_back(vb);
    loadDB(cb, cl, false, vb, stSeqno, enSeqno);
}

void CouchKVStore::dumpKeys(std::vector<uint16_t> &vbids,  shared_ptr<Callback<GetValue> > cb)
{
    shared_ptr<Callback<CacheLookup> > cl(new NoLookupCallback());
    std::vector<uint16_t>::iterator itr = vbids.begin();
    for (; itr != vbids.end(); ++itr) {
        loadDB(cb, cl, true, *itr, 0, std::numeric_limits<uint64_t>::max(),
               COUCHSTORE_NO_DELETES);
    }
}

void CouchKVStore::dumpDeleted(uint16_t vb, uint64_t stSeqno, uint64_t enSeqno,
                               shared_ptr<Callback<GetValue> > cb)
{
    std::vector<uint16_t> vbids;
    vbids.push_back(vb);
    shared_ptr<Callback<CacheLookup> > cl(new NoLookupCallback());
    loadDB(cb, cl, true, vb, stSeqno, enSeqno, COUCHSTORE_DELETES_ONLY);
}

StorageProperties CouchKVStore::getStorageProperties()
{
    StorageProperties rv(true, true, true, true);
    return rv;
}

bool CouchKVStore::commit(Callback<kvstats_ctx> *cb)
{
    assert(!isReadOnly());
    if (intransaction) {
        intransaction = commit2couchstore(cb) ? false : true;
    }
    return !intransaction;

}

void CouchKVStore::addStats(const std::string &prefix,
                            ADD_STAT add_stat,
                            const void *c)
{
    const char *prefix_str = prefix.c_str();

    /* stats for both read-only and read-write threads */
    addStat(prefix_str, "backend_type",   "couchstore",       add_stat, c);
    addStat(prefix_str, "open",           st.numOpen,         add_stat, c);
    addStat(prefix_str, "close",          st.numClose,        add_stat, c);
    addStat(prefix_str, "readTime",       st.readTimeHisto,   add_stat, c);
    addStat(prefix_str, "readSize",       st.readSizeHisto,   add_stat, c);
    addStat(prefix_str, "numLoadedVb",    st.numLoadedVb,     add_stat, c);

    // failure stats
    addStat(prefix_str, "failure_open",   st.numOpenFailure, add_stat, c);
    addStat(prefix_str, "failure_get",    st.numGetFailure,  add_stat, c);

    if (!isReadOnly()) {
        addStat(prefix_str, "failure_set",   st.numSetFailure,   add_stat, c);
        addStat(prefix_str, "failure_del",   st.numDelFailure,   add_stat, c);
        addStat(prefix_str, "failure_vbset", st.numVbSetFailure, add_stat, c);
        addStat(prefix_str, "lastCommDocs",  st.docsCommitted,   add_stat, c);
        addStat(prefix_str, "numCommitRetry", st.numCommitRetry, add_stat, c);

        // stats for CouchNotifier
        couchNotifier->addStats(prefix, add_stat, c);
    }
}

void CouchKVStore::addTimingStats(const std::string &prefix,
                                  ADD_STAT add_stat, const void *c) {
    if (isReadOnly()) {
        return;
    }
    const char *prefix_str = prefix.c_str();
    addStat(prefix_str, "commit",      st.commitHisto,      add_stat, c);
    addStat(prefix_str, "compact",     st.compactHisto,     add_stat, c);
    addStat(prefix_str, "commitRetry", st.commitRetryHisto, add_stat, c);
    addStat(prefix_str, "delete",      st.delTimeHisto,     add_stat, c);
    addStat(prefix_str, "save_documents", st.saveDocsHisto, add_stat, c);
    addStat(prefix_str, "writeTime",   st.writeTimeHisto,   add_stat, c);
    addStat(prefix_str, "writeSize",   st.writeSizeHisto,   add_stat, c);
    addStat(prefix_str, "bulkSize",    st.batchSize,        add_stat, c);

    // Couchstore file ops stats
    addStat(prefix_str, "fsReadTime",  st.fsStats.readTimeHisto,  add_stat, c);
    addStat(prefix_str, "fsWriteTime", st.fsStats.writeTimeHisto, add_stat, c);
    addStat(prefix_str, "fsSyncTime",  st.fsStats.syncTimeHisto,  add_stat, c);
{'fraction_non_alphanumeric': 0.11919894160678617, 'fraction_numerical': 0.006355547485019067, 'mean_word_length': 2.3115282192251523, 'pattern_counts': {'":': 6, '<': 307, '<?xml version=': 0, '>': 293, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 48, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '954416', 'n_tokens_mistral': 25734, 'n_tokens_neox': 25020, 'n_words': 5211}