<commit_before>#include "RecoveryChecker.h"
#include "DistributeFileSyncMgr.h"
#include "DistributeFileSys.h"
#include "DistributeDriver.h"
#include "RequestLog.h"
#include "NodeManagerBase.h"
#include "DistributeTest.hpp"

#include <glog/logging.h>
#include <boost/filesystem.hpp>
#include <common/Keys.h>
#include <util/driver/Reader.h>
#include <util/driver/Writer.h>
#include <util/driver/writers/JsonWriter.h>
#include <util/driver/readers/JsonReader.h>

#define MAX_BACKUP_NUM 3
namespace bfs = boost::filesystem;

using namespace izenelib::driver;

namespace sf1r
{

class CopyGuard
{
public:
    CopyGuard(const std::string& guard_dir)
        : ok_(false), guard_file_(guard_dir + "/" + getGuardFileName())
    {
        // touch a flag
        std::ofstream ofs(guard_file_.c_str());
        ofs << std::flush;
        ofs.close();
    }
    ~CopyGuard()
    {
        // clear a flag
        if (ok_ && bfs::exists(guard_file_))
        {
            bfs::remove(guard_file_);
        }
    }
    void setOK()
    {
        ok_ = true;
    }
    static bool isDirCopyOK(const std::string& dir)
    {
        return !bfs::exists(dir + "/" + getGuardFileName());
    }
    static void safe_remove_all(const std::string& dir)
    {
        CopyGuard dir_guard(dir);
        try
        {
            // remove other files first and then remove flag.
            static bfs::directory_iterator end_it = bfs::directory_iterator();
            bfs::directory_iterator dir_it = bfs::directory_iterator(dir);
            while(dir_it != end_it)
            {
                bfs::path current(dir_it->path());
                if (bfs::is_regular_file(current) &&
                    current.filename().string() == getGuardFileName())
                {
                    ++dir_it;
                    continue;
                }
                DistributeTestSuit::testFail(Fail_At_CopyRemove_File);
                bfs::remove_all(current);
                ++dir_it;
            }
            bfs::remove(dir + "/" + getGuardFileName());
            bfs::remove_all(dir);
        }
        catch(const std::exception& e)
        {
            LOG(ERROR) << "safe remove error: " << e.what();
            return;
        }
        dir_guard.setOK();
    }

    static std::string getGuardFileName()
    {
        static std::string guard_flag("copy_guard_file.flag");
        return guard_flag;
    }

private:
    bool ok_;
    std::string guard_file_;
};

static void getBackupList(const bfs::path& backup_basepath, std::vector<uint32_t>& backup_req_incids)
{
    if (!bfs::exists(backup_basepath))
        return;
    // only keep the lastest 3 backups.
    static const bfs::directory_iterator end_dir = bfs::directory_iterator();
    bfs::directory_iterator backup_dirs_it = bfs::directory_iterator(backup_basepath);
    while(backup_dirs_it != end_dir)
    {
        if (!bfs::is_directory(*backup_dirs_it))
        {
            ++backup_dirs_it;
            continue;
        }
        try
        {
            uint32_t inc_id = boost::lexical_cast<uint32_t>((*backup_dirs_it).path().filename().string());
            backup_req_incids.push_back(inc_id);
        }
        catch(const boost::bad_lexical_cast& e)
        {
            LOG(INFO) << "not a backup dir : " << (*backup_dirs_it).path().filename();
        }
        ++backup_dirs_it;
    }
    std::sort(backup_req_incids.begin(), backup_req_incids.end(), std::greater<uint32_t>());
}

static void cleanUnnessesaryBackup(const bfs::path& backup_basepath)
{
    // only keep the lastest 3 backups.
    std::vector<uint32_t> backup_req_incids;
    getBackupList(backup_basepath, backup_req_incids);
    int keeped = 0;
    for (size_t i = 0; i < backup_req_incids.size(); ++i)
    {
        if (backup_req_incids[i] == 0)
        {
            // keep the first backup.
            continue;
        }
        bfs::path backup_dir = bfs::path(backup_basepath)/bfs::path(boost::lexical_cast<std::string>(backup_req_incids[i]));
        if (keeped >= MAX_BACKUP_NUM)
            CopyGuard::safe_remove_all(backup_dir.string());
        else
        {
            if (CopyGuard::isDirCopyOK(backup_dir.string()))
                ++keeped;
            else 
            {
                LOG(WARNING) << "a corrupted directory removed " << backup_dir;
                CopyGuard::safe_remove_all(backup_dir.string());
            }
        }
    }
}

static bool getLastBackup(const bfs::path& backup_basepath, uint32_t less_than, std::string& backup_path, uint32_t& backup_inc_id)
{
    std::vector<uint32_t> backup_req_incids;
    getBackupList(backup_basepath, backup_req_incids);
    if (backup_req_incids.empty())
        return false;
    LOG(INFO) << "getting last backup id which is less than : " << less_than;
    for (size_t i = 0; i < backup_req_incids.size(); ++i)
    {
        backup_inc_id = backup_req_incids[i];
        backup_path = (backup_basepath/bfs::path(boost::lexical_cast<std::string>(backup_inc_id))).string() + "/";
        if (CopyGuard::isDirCopyOK(backup_path))
        {
            if (less_than == 0 || backup_inc_id < less_than)
                return true;
        }
        else 
        {
            LOG(WARNING) << "a corrupted directory removed " << backup_path;
            CopyGuard::safe_remove_all(backup_path);
        }
    }
    return false;
}

static void copy_file_keep_modification(const bfs::path& from, const bfs::path& to)
{
    bfs::copy_file(from, to, bfs::copy_option::overwrite_if_exists);
    try
    {
        bfs::last_write_time(to, bfs::last_write_time(from));
    }catch(const std::exception& e){}
}

static void copyDir(bfs::path src, bfs::path dest)
{
    if( !bfs::exists(src) ||
        !bfs::is_directory(src) )
    {
        RecoveryChecker::forceExit("Source directory " + src.string() +
            " does not exist or is not a directory.");
    }
    if( !bfs::exists(dest) )
    {
        bfs::create_directory(dest);
    }
    // Iterate through the source directory
    static const bfs::directory_iterator end_it = bfs::directory_iterator();
    for( bfs::directory_iterator file(src); file != end_it; ++file )
    {
        bfs::path current(file->path());
        if (!bfs::exists(current))
        {
            LOG(WARNING) << "the file disappeared while coping: " << current;
            continue;
        }
        if(bfs::is_directory(current))
        {
            // copy recursion
            copyDir(current, dest / current.filename());
        }
        else
        {
            if (current.filename().string().find("_removed.rollback") != std::string::npos)
                continue;

            DistributeTestSuit::testFail(Fail_At_CopyRemove_File);
            // Found file: Copy
            copy_file_keep_modification(current, dest / current.filename());
            //LOG(INFO) << "copying : " << current << " to " << dest;
        }
    }
    // remove the file that exists in dest but not in src.
    for ( bfs::directory_iterator file(dest); file != end_it; ++file )
    {
        bfs::path current(file->path());
        if (!bfs::exists(src / current.filename()))
        {
            bfs::remove_all(current);
            LOG(INFO) << "path removed since not in src : " << current;
        }
    }
}

// copy_dir will keep src path filename
// src = xxx/xxx/src_name
// dest = xxx/xxx/xxx
// copy xxx/xxx/src_name to dest/src
// with keep_full_path = false copy xxx/xxx/src_name to dest/src_name
// if dest = xxx/xxx/xxx/src_name
// then copy src to dest
static void copy_dir(const bfs::path& src, const bfs::path& dest, bool keep_full_path = true)
{
    bfs::path dest_path = dest;
    if (src.filename() == dest_path.filename())
    {
    }
    else if (!keep_full_path)
    {
        dest_path /= src.filename();
    }
    else
    {
        dest_path /= src;
    }
    bfs::create_directories(dest_path);
    copyDir(src, dest_path);
}

void RecoveryChecker::getCollList(std::vector<std::string>& coll_list)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    CollInfoMapT::const_iterator cit = all_col_info_.begin();
    while (cit != all_col_info_.end())
    {
        coll_list.push_back(cit->first);
        ++cit;
    }
}

bool RecoveryChecker::getCollPath(const std::string& colname, CollectionPath& colpath)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    CollInfoMapT::const_iterator cit = all_col_info_.find(colname);
    if (cit == all_col_info_.end())
        return false;
    colpath = cit->second.first;
    return true;
}

void RecoveryChecker::addCollection(const std::string& colname, const CollectionPath& colpath, const std::string& configfile)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    LOG(INFO) << "collection added, updating in RecoveryChecker." << colname;
    all_col_info_[colname] = std::make_pair(colpath, configfile);
}

void RecoveryChecker::removeCollection(const std::string& colname)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    LOG(INFO) << "collection removed, updating in RecoveryChecker." << colname;
    all_col_info_.erase(colname);
}


bool RecoveryChecker::setRollbackFlag(uint32_t inc_id)
{
    if (bfs::exists(rollback_file_))
    {
        LOG(ERROR) << "rollback_file already exist, this mean last rollback is not finished. must exit.";
        forceExit("set rollback flag failed for existed flag.");
    }
    ofstream ofs(rollback_file_.c_str());
    if (ofs.good())
    {
        ofs.write((const char*)&inc_id, sizeof(inc_id));
        return true;
    }
    return false;
}

void RecoveryChecker::clearRollbackFlag()
{
    if (bfs::exists(rollback_file_))
        bfs::remove(rollback_file_);
}

bool RecoveryChecker::backup(bool force_remove)
{
    CollInfoMapT tmp_all_col_info;
    {
        boost::unique_lock<boost::mutex> lock(mutex_);
        if (!reqlog_mgr_)
        {
            LOG(ERROR) << "RecoveryChecker did not init!";
            return false;
        }
        need_backup_ = false;
        // find all collection and backup.
        tmp_all_col_info = all_col_info_;
    }

    // backup changeable data first, so that we can rollback if old data corrupt while process the request.
    // Ignore SCD files and any other files that will not change during processing.
    CollInfoMapT::const_iterator cit = tmp_all_col_info.begin();
    bfs::path dest_path = backup_basepath_ + "/" + boost::lexical_cast<std::string>(reqlog_mgr_->getLastSuccessReqId());
    bfs::path dest_coldata_backup = dest_path/bfs::path("backup_data");

    if (bfs::exists(dest_path))
    {
        if (CopyGuard::isDirCopyOK(dest_path.string()) && !force_remove)
        {
            LOG(INFO) << "backup already exists : " << dest_path;
            return true;
        }
        CopyGuard::safe_remove_all(dest_path.string());
    }
    bfs::create_directories(dest_path);

    {
        CopyGuard dir_guard(dest_path.string());

        bfs::create_directories(dest_coldata_backup);

        // set invalide flag before do copy and
        // clear invalide flag after copy. So we can
        // know whether copy is finished correctly by checking flag .
        //
        while(cit != tmp_all_col_info.end())
        {
            LOG(INFO) << "backing up the collection: " << cit->first;
            // flush collection to make sure all changes have been saved to disk.
            if (flush_col_)
                flush_col_(cit->first);

            LOG(INFO) << "flush the collection finished.";
            if(!backupColl(cit->second.first, dest_path))
            {
                return false;
            }
            ++cit;
        }
        try
        {
            copy_dir(request_log_basepath_, dest_path, false);
            copy_file_keep_modification(last_conf_file_, dest_path/bfs::path(last_conf_file_).filename());
        }
        catch(const std::exception& e)
        {
            // clear state.
            LOG(ERROR) << "backup request log failed. " << e.what() << std::endl;
            return false;
        }
        dir_guard.setOK();
    }

    LOG(INFO) << "back up finished.";
    cleanUnnessesaryBackup(backup_basepath_);
    return true;
}

void RecoveryChecker::replayLog(bool is_primary, const std::string& from_col,
    const std::string& to_col, std::vector<uint32_t>& replayed_id_list)
{
    // replay log, ignore any fail and do not write any log during replay.
    // read redo log and do request to current log.
    uint32_t start_id = 0;
    uint32_t end_id = reqlog_mgr_->getLastSuccessReqId();
    if (start_id > end_id)
    {
        LOG(ERROR) << "no log replay.";
        return;
    }
    ReqLogHead rethead;
    size_t redo_offset = 0;
    if(!reqlog_mgr_->getHeadOffset(start_id, rethead, redo_offset))
    {
        LOG(ERROR) << "get start head failed." << start_id;
        return;
    }

    LOG(INFO) << "replaying log is_primary: " << is_primary << ", from collection: " << from_col <<
        ", to collection : " << to_col;
    if (is_primary)
        replayed_id_list.reserve(end_id - start_id);
    else
    {
        if (replayed_id_list.empty())
        {
            LOG(INFO) << "nothing log need replay.";
            return;
        }
    }


    uint32_t replayed_num = 0;

    while(true)
    {
        std::string req_packed_data; 
        CommonReqData req_commondata;
        try
        {
            bool hasmore = reqlog_mgr_->getReqDataByHeadOffset(redo_offset, rethead, req_packed_data);
            if (!hasmore || rethead.inc_id > end_id)
            {
                if (rethead.inc_id < end_id)
                    end_id = rethead.inc_id;
                LOG(INFO) << "end at inc_id : " << end_id;
                break;
            }
            if (!is_primary)
            {
                // only replay the id that primary has replayed.
                if (rethead.inc_id != replayed_id_list[replayed_num])
                {
                    LOG(INFO) << "ignore replay : " << rethead.inc_id;
                    continue;
                }
            }

            if(!ReqLogMgr::unpackReqLogData(req_packed_data, req_commondata))
            {
                LOG(INFO) << "unpack replaying log data failed : " << req_packed_data;
                if (!is_primary)
                    break;
                continue;
            }

            if (req_commondata.reqtype == Req_UpdateConfig ||
                req_commondata.reqtype == Req_CronJob ||
                req_commondata.reqtype == Req_Index ||
                req_commondata.reqtype == Req_Callback)
            {
                //LOG(INFO) << "ignore replaying log type : " << req_commondata.reqtype;
                if (!is_primary)
                    break;
                continue;
            }

            // change collection .
            static izenelib::driver::JsonReader reader;
            Value requestValue;
            if(!reader.read(req_commondata.req_json_data, requestValue))
            {
                LOG(ERROR) << "read request json data error.";
                if (!is_primary)
                    break;
                continue;
            }
            if (requestValue.type() != Value::kObjectType)
            {
                LOG(ERROR) << "request value type error.";
                if (!is_primary)
                    break;
                continue;
            }
            Request request;
            request.assignTmp(requestValue);
            if (!ReqLogMgr::isReplayWriteReq(request.controller(), request.action()))
            {
                if (!is_primary)
                    break;
                continue;
            }
            std::string collection = asString(request[Keys::collection]);
            if (collection != from_col)
            {
                //LOG(INFO) << "not match collection : " << collection;
                if (!is_primary)
                    break;
                continue;
            }
            request[Keys::collection] = to_col;

            std::string replay_json;
            izenelib::driver::JsonWriter writer;
            writer.write(request.get(), replay_json);
            //LOG(INFO) << "replaying for request id : " << rethead.inc_id;
            CommonReqData new_common = req_commondata;
            new_common.req_json_data = replay_json;
            ReqLogMgr::replaceCommonReqData(req_commondata, new_common, req_packed_data);

            if(!DistributeDriver::get()->handleReqFromLog(req_commondata.reqtype, replay_json, req_packed_data))
            {
                LOG(INFO) << "ignore replaying from log failed: " << rethead.inc_id << ", json : " << req_commondata.req_json_data;
                if (!is_primary)
                    break;
                continue;
            }
            replayed_num++;
            if (is_primary)
                replayed_id_list.push_back(rethead.inc_id);
        }
        catch(const std::exception& e)
        {
            LOG(INFO) << "replay log exception: " << e.what() << ", req data: " << req_commondata.req_json_data;
            if (!is_primary)
                break;
        }
    }
    if (!is_primary && replayed_num != replayed_id_list.size())
    {
        LOG(ERROR) << "replay log error on secondary.";
        forceExit("Replay Log Failed!");
    }
    LOG(INFO) << "replay log finished, total replayed: " << replayed_num;
}

bool RecoveryChecker::redoLog(ReqLogMgr* redolog, uint32_t start_id, uint32_t end_id)
{
    bool ret = true;
    try
    {
        // read redo log and do request to current log.
        if (start_id > redolog->getLastSuccessReqId())
        {
            LOG(ERROR) << "backup id should not larger than last write id. " << start_id;
            LOG(ERROR) << "leaving old backup and do not redo log. restart and wait to sync to newest log";
            forceExit("redo start id is larger than last writed id.");
        }
        ReqLogHead rethead;
        size_t redo_offset = 0;
        if(!redolog->getHeadOffset(start_id, rethead, redo_offset))
        {
            assert(false);
            LOG(ERROR) << "get head failed. This should never happen.";
            forceExit("redo get head info failed.");
        }
        while(true)
        {
            std::string req_packed_data; 
            bool hasmore = redolog->getReqDataByHeadOffset(redo_offset, rethead, req_packed_data);
            if (!hasmore || rethead.inc_id >= end_id)
                break;
            LOG(INFO) << "redoing for request id : " << rethead.inc_id;
            CommonReqData req_commondata;
            ReqLogMgr::unpackReqLogData(req_packed_data, req_commondata);

            if(!DistributeDriver::get()->handleReqFromLog(req_commondata.reqtype, req_commondata.req_json_data, req_packed_data))
            {
                LOG(INFO) << "redoing from log failed: " << rethead.inc_id;
                forceExit("handleReqFromLog failed.");
            }
        }
    }
    catch(const std::exception& e)
    {
        LOG(INFO) << "redo log exception: " << e.what();
        ret = false;
    }
    bfs::remove_all(redo_log_basepath_);
    return ret;
}

bool RecoveryChecker::backupColl(const CollectionPath& colpath, const bfs::path& dest_path)
{
    // find all collection and backup.
    bfs::path dest_coldata_backup = dest_path/bfs::path("backup_data");

    if (!bfs::exists(dest_path))
    {
        bfs::create_directories(dest_path);
        bfs::create_directories(dest_coldata_backup);
    }

    bfs::path coldata_path(colpath.getCollectionDataPath());
    bfs::path querydata_path(colpath.getQueryDataPath());

    try
    {
        copy_dir(coldata_path, dest_coldata_backup);
        copy_dir(querydata_path, dest_coldata_backup);
    }
    catch(const std::exception& e)
    {
        LOG(ERROR) << "backup collection data failed. " << e.what() << std::endl;
        return false;
    }
    return true;
}

bool RecoveryChecker::hasAnyBackup(uint32_t& last_backup_id)
{
    std::string last_backup_path;
    last_backup_id = 0;
    bool has_backup = true;
    if (!getLastBackup(backup_basepath_, 0, last_backup_path, last_backup_id))
    {
        last_backup_id = 0;
        has_backup = false;
    }
    return has_backup;
}

bool RecoveryChecker::isNeedRollback(bool starting_up)
{
    if (!NodeManagerBase::get()->isDistributed())
        return false;
    if (bfs::exists(rollback_file_))
        return true;
    if (starting_up && !isLastNormalExit())
    {
        return true;
    }
    return false;
}

bool RecoveryChecker::checkAndRestoreBackupFile(const CollectionPath& colpath)
{
    LOG(INFO) << "check and restoring backup at startup.";

    if(!isNeedRollback(true))
    {
        return true;
    }

    uint32_t rollback_id = 0;
    ifstream ifs(rollback_file_.c_str());
    if (ifs.good())
    {
        ifs.read((char*)&rollback_id, sizeof(rollback_id));
    }

    std::string last_backup_path;
    uint32_t last_backup_id = 0;
    bool has_backup = true;
    if (!getLastBackup(backup_basepath_, rollback_id, last_backup_path, last_backup_id))
    {
        LOG(ERROR) << "no backup available while check startup.";
        LOG(ERROR) << "need restart to redo all log or get full data from primary.";
        last_backup_id = 0;
        has_backup = false;
        return false;
    }
    if (has_backup)
    {
        // copy backup data to replace current and reopen all file.
        try
        {
            bfs::path dest_coldata_backup(last_backup_path + "/backup_data");
            LOG(INFO) << "restoring the backup for the collection: " <<
                colpath.getCollectionDataPath() << ", backup id: " << last_backup_id;

            bfs::path coldata_path(colpath.getCollectionDataPath());
            bfs::path querydata_path(colpath.getQueryDataPath());

            copy_dir(dest_coldata_backup/coldata_path, coldata_path);
            copy_dir(dest_coldata_backup/querydata_path, querydata_path);
        }
        catch(const std::exception& e)
        {
            LOG(ERROR) << "copy backup to current working path failed." << e.what();
            return false;
        }
    }
    return true;
}

// note : if there are more than one collection we need backup
// all collection data. only in this way we can rollback and redo log to 
// rollback the data before last failed request.
bool RecoveryChecker::rollbackLastFail(bool starting_up)
{
    // rollback will only happen in recovery or abort request, both of them 
    // are protected by the lock of node state in NodeManagerBase. 
    // not all inc_id has a backup, so we just find the newest backup.<commit_msg>1)Remove check query-data directory<commit_after>#include "RecoveryChecker.h"
#include "DistributeFileSyncMgr.h"
#include "DistributeFileSys.h"
#include "DistributeDriver.h"
#include "RequestLog.h"
#include "NodeManagerBase.h"
#include "DistributeTest.hpp"

#include <glog/logging.h>
#include <boost/filesystem.hpp>
#include <common/Keys.h>
#include <util/driver/Reader.h>
#include <util/driver/Writer.h>
#include <util/driver/writers/JsonWriter.h>
#include <util/driver/readers/JsonReader.h>

#define MAX_BACKUP_NUM 3
namespace bfs = boost::filesystem;

using namespace izenelib::driver;

namespace sf1r
{

class CopyGuard
{
public:
    CopyGuard(const std::string& guard_dir)
        : ok_(false), guard_file_(guard_dir + "/" + getGuardFileName())
    {
        // touch a flag
        std::ofstream ofs(guard_file_.c_str());
        ofs << std::flush;
        ofs.close();
    }
    ~CopyGuard()
    {
        // clear a flag
        if (ok_ && bfs::exists(guard_file_))
        {
            bfs::remove(guard_file_);
        }
    }
    void setOK()
    {
        ok_ = true;
    }
    static bool isDirCopyOK(const std::string& dir)
    {
        return !bfs::exists(dir + "/" + getGuardFileName());
    }
    static void safe_remove_all(const std::string& dir)
    {
        CopyGuard dir_guard(dir);
        try
        {
            // remove other files first and then remove flag.
            static bfs::directory_iterator end_it = bfs::directory_iterator();
            bfs::directory_iterator dir_it = bfs::directory_iterator(dir);
            while(dir_it != end_it)
            {
                bfs::path current(dir_it->path());
                if (bfs::is_regular_file(current) &&
                    current.filename().string() == getGuardFileName())
                {
                    ++dir_it;
                    continue;
                }
                DistributeTestSuit::testFail(Fail_At_CopyRemove_File);
                bfs::remove_all(current);
                ++dir_it;
            }
            bfs::remove(dir + "/" + getGuardFileName());
            bfs::remove_all(dir);
        }
        catch(const std::exception& e)
        {
            LOG(ERROR) << "safe remove error: " << e.what();
            return;
        }
        dir_guard.setOK();
    }

    static std::string getGuardFileName()
    {
        static std::string guard_flag("copy_guard_file.flag");
        return guard_flag;
    }

private:
    bool ok_;
    std::string guard_file_;
};

static void getBackupList(const bfs::path& backup_basepath, std::vector<uint32_t>& backup_req_incids)
{
    if (!bfs::exists(backup_basepath))
        return;
    // only keep the lastest 3 backups.
    static const bfs::directory_iterator end_dir = bfs::directory_iterator();
    bfs::directory_iterator backup_dirs_it = bfs::directory_iterator(backup_basepath);
    while(backup_dirs_it != end_dir)
    {
        if (!bfs::is_directory(*backup_dirs_it))
        {
            ++backup_dirs_it;
            continue;
        }
        try
        {
            uint32_t inc_id = boost::lexical_cast<uint32_t>((*backup_dirs_it).path().filename().string());
            backup_req_incids.push_back(inc_id);
        }
        catch(const boost::bad_lexical_cast& e)
        {
            LOG(INFO) << "not a backup dir : " << (*backup_dirs_it).path().filename();
        }
        ++backup_dirs_it;
    }
    std::sort(backup_req_incids.begin(), backup_req_incids.end(), std::greater<uint32_t>());
}

static void cleanUnnessesaryBackup(const bfs::path& backup_basepath)
{
    // only keep the lastest 3 backups.
    std::vector<uint32_t> backup_req_incids;
    getBackupList(backup_basepath, backup_req_incids);
    int keeped = 0;
    for (size_t i = 0; i < backup_req_incids.size(); ++i)
    {
        if (backup_req_incids[i] == 0)
        {
            // keep the first backup.
            continue;
        }
        bfs::path backup_dir = bfs::path(backup_basepath)/bfs::path(boost::lexical_cast<std::string>(backup_req_incids[i]));
        if (keeped >= MAX_BACKUP_NUM)
            CopyGuard::safe_remove_all(backup_dir.string());
        else
        {
            if (CopyGuard::isDirCopyOK(backup_dir.string()))
                ++keeped;
            else 
            {
                LOG(WARNING) << "a corrupted directory removed " << backup_dir;
                CopyGuard::safe_remove_all(backup_dir.string());
            }
        }
    }
}

static bool getLastBackup(const bfs::path& backup_basepath, uint32_t less_than, std::string& backup_path, uint32_t& backup_inc_id)
{
    std::vector<uint32_t> backup_req_incids;
    getBackupList(backup_basepath, backup_req_incids);
    if (backup_req_incids.empty())
        return false;
    LOG(INFO) << "getting last backup id which is less than : " << less_than;
    for (size_t i = 0; i < backup_req_incids.size(); ++i)
    {
        backup_inc_id = backup_req_incids[i];
        backup_path = (backup_basepath/bfs::path(boost::lexical_cast<std::string>(backup_inc_id))).string() + "/";
        if (CopyGuard::isDirCopyOK(backup_path))
        {
            if (less_than == 0 || backup_inc_id < less_than)
                return true;
        }
        else 
        {
            LOG(WARNING) << "a corrupted directory removed " << backup_path;
            CopyGuard::safe_remove_all(backup_path);
        }
    }
    return false;
}

static void copy_file_keep_modification(const bfs::path& from, const bfs::path& to)
{
    bfs::copy_file(from, to, bfs::copy_option::overwrite_if_exists);
    try
    {
        bfs::last_write_time(to, bfs::last_write_time(from));
    }catch(const std::exception& e){}
}

static void copyDir(bfs::path src, bfs::path dest)
{
    if( !bfs::exists(src) ||
        !bfs::is_directory(src) )
    {
        RecoveryChecker::forceExit("Source directory " + src.string() +
            " does not exist or is not a directory.");
    }
    if( !bfs::exists(dest) )
    {
        bfs::create_directory(dest);
    }
    // Iterate through the source directory
    static const bfs::directory_iterator end_it = bfs::directory_iterator();
    for( bfs::directory_iterator file(src); file != end_it; ++file )
    {
        bfs::path current(file->path());
        if (!bfs::exists(current))
        {
            LOG(WARNING) << "the file disappeared while coping: " << current;
            continue;
        }
        if(bfs::is_directory(current))
        {
            // copy recursion
            copyDir(current, dest / current.filename());
        }
        else
        {
            if (current.filename().string().find("_removed.rollback") != std::string::npos)
                continue;

            DistributeTestSuit::testFail(Fail_At_CopyRemove_File);
            // Found file: Copy
            copy_file_keep_modification(current, dest / current.filename());
            //LOG(INFO) << "copying : " << current << " to " << dest;
        }
    }
    // remove the file that exists in dest but not in src.
    for ( bfs::directory_iterator file(dest); file != end_it; ++file )
    {
        bfs::path current(file->path());
        if (!bfs::exists(src / current.filename()))
        {
            bfs::remove_all(current);
            LOG(INFO) << "path removed since not in src : " << current;
        }
    }
}

// copy_dir will keep src path filename
// src = xxx/xxx/src_name
// dest = xxx/xxx/xxx
// copy xxx/xxx/src_name to dest/src
// with keep_full_path = false copy xxx/xxx/src_name to dest/src_name
// if dest = xxx/xxx/xxx/src_name
// then copy src to dest
static void copy_dir(const bfs::path& src, const bfs::path& dest, bool keep_full_path = true)
{
    bfs::path dest_path = dest;
    if (src.filename() == dest_path.filename())
    {
    }
    else if (!keep_full_path)
    {
        dest_path /= src.filename();
    }
    else
    {
        dest_path /= src;
    }
    bfs::create_directories(dest_path);
    copyDir(src, dest_path);
}

void RecoveryChecker::getCollList(std::vector<std::string>& coll_list)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    CollInfoMapT::const_iterator cit = all_col_info_.begin();
    while (cit != all_col_info_.end())
    {
        coll_list.push_back(cit->first);
        ++cit;
    }
}

bool RecoveryChecker::getCollPath(const std::string& colname, CollectionPath& colpath)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    CollInfoMapT::const_iterator cit = all_col_info_.find(colname);
    if (cit == all_col_info_.end())
        return false;
    colpath = cit->second.first;
    return true;
}

void RecoveryChecker::addCollection(const std::string& colname, const CollectionPath& colpath, const std::string& configfile)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    LOG(INFO) << "collection added, updating in RecoveryChecker." << colname;
    all_col_info_[colname] = std::make_pair(colpath, configfile);
}

void RecoveryChecker::removeCollection(const std::string& colname)
{
    boost::unique_lock<boost::mutex> lock(mutex_);
    LOG(INFO) << "collection removed, updating in RecoveryChecker." << colname;
    all_col_info_.erase(colname);
}


bool RecoveryChecker::setRollbackFlag(uint32_t inc_id)
{
    if (bfs::exists(rollback_file_))
    {
        LOG(ERROR) << "rollback_file already exist, this mean last rollback is not finished. must exit.";
        forceExit("set rollback flag failed for existed flag.");
    }
    ofstream ofs(rollback_file_.c_str());
    if (ofs.good())
    {
        ofs.write((const char*)&inc_id, sizeof(inc_id));
        return true;
    }
    return false;
}

void RecoveryChecker::clearRollbackFlag()
{
    if (bfs::exists(rollback_file_))
        bfs::remove(rollback_file_);
}

bool RecoveryChecker::backup(bool force_remove)
{
    CollInfoMapT tmp_all_col_info;
    {
        boost::unique_lock<boost::mutex> lock(mutex_);
        if (!reqlog_mgr_)
        {
            LOG(ERROR) << "RecoveryChecker did not init!";
            return false;
        }
        need_backup_ = false;
        // find all collection and backup.
        tmp_all_col_info = all_col_info_;
    }

    // backup changeable data first, so that we can rollback if old data corrupt while process the request.
    // Ignore SCD files and any other files that will not change during processing.
    CollInfoMapT::const_iterator cit = tmp_all_col_info.begin();
    bfs::path dest_path = backup_basepath_ + "/" + boost::lexical_cast<std::string>(reqlog_mgr_->getLastSuccessReqId());
    bfs::path dest_coldata_backup = dest_path/bfs::path("backup_data");

    if (bfs::exists(dest_path))
    {
        if (CopyGuard::isDirCopyOK(dest_path.string()) && !force_remove)
        {
            LOG(INFO) << "backup already exists : " << dest_path;
            return true;
        }
        CopyGuard::safe_remove_all(dest_path.string());
    }
    bfs::create_directories(dest_path);

    {
        CopyGuard dir_guard(dest_path.string());

        bfs::create_directories(dest_coldata_backup);

        // set invalide flag before do copy and
        // clear invalide flag after copy. So we can
        // know whether copy is finished correctly by checking flag .
        //
        while(cit != tmp_all_col_info.end())
        {
            LOG(INFO) << "backing up the collection: " << cit->first;
            // flush collection to make sure all changes have been saved to disk.
            if (flush_col_)
                flush_col_(cit->first);

            LOG(INFO) << "flush the collection finished.";
            if(!backupColl(cit->second.first, dest_path))
            {
                return false;
            }
            ++cit;
        }
        try
        {
            copy_dir(request_log_basepath_, dest_path, false);
            copy_file_keep_modification(last_conf_file_, dest_path/bfs::path(last_conf_file_).filename());
        }
        catch(const std::exception& e)
        {
            // clear state.
            LOG(ERROR) << "backup request log failed. " << e.what() << std::endl;
            return false;
        }
        dir_guard.setOK();
    }

    LOG(INFO) << "back up finished.";
    cleanUnnessesaryBackup(backup_basepath_);
    return true;
}

void RecoveryChecker::replayLog(bool is_primary, const std::string& from_col,
    const std::string& to_col, std::vector<uint32_t>& replayed_id_list)
{
    // replay log, ignore any fail and do not write any log during replay.
    // read redo log and do request to current log.
    uint32_t start_id = 0;
    uint32_t end_id = reqlog_mgr_->getLastSuccessReqId();
    if (start_id > end_id)
    {
        LOG(ERROR) << "no log replay.";
        return;
    }
    ReqLogHead rethead;
    size_t redo_offset = 0;
    if(!reqlog_mgr_->getHeadOffset(start_id, rethead, redo_offset))
    {
        LOG(ERROR) << "get start head failed." << start_id;
        return;
    }

    LOG(INFO) << "replaying log is_primary: " << is_primary << ", from collection: " << from_col <<
        ", to collection : " << to_col;
    if (is_primary)
        replayed_id_list.reserve(end_id - start_id);
    else
    {
        if (replayed_id_list.empty())
        {
            LOG(INFO) << "nothing log need replay.";
            return;
        }
    }


    uint32_t replayed_num = 0;

    while(true)
    {
        std::string req_packed_data; 
        CommonReqData req_commondata;
        try
        {
            bool hasmore = reqlog_mgr_->getReqDataByHeadOffset(redo_offset, rethead, req_packed_data);
            if (!hasmore || rethead.inc_id > end_id)
            {
                if (rethead.inc_id < end_id)
                    end_id = rethead.inc_id;
                LOG(INFO) << "end at inc_id : " << end_id;
                break;
            }
            if (!is_primary)
            {
                // only replay the id that primary has replayed.
                if (rethead.inc_id != replayed_id_list[replayed_num])
                {
                    LOG(INFO) << "ignore replay : " << rethead.inc_id;
                    continue;
                }
            }

            if(!ReqLogMgr::unpackReqLogData(req_packed_data, req_commondata))
            {
                LOG(INFO) << "unpack replaying log data failed : " << req_packed_data;
                if (!is_primary)
                    break;
                continue;
            }

            if (req_commondata.reqtype == Req_UpdateConfig ||
                req_commondata.reqtype == Req_CronJob ||
                req_commondata.reqtype == Req_Index ||
                req_commondata.reqtype == Req_Callback)
            {
                //LOG(INFO) << "ignore replaying log type : " << req_commondata.reqtype;
                if (!is_primary)
                    break;
                continue;
            }

            // change collection .
            static izenelib::driver::JsonReader reader;
            Value requestValue;
            if(!reader.read(req_commondata.req_json_data, requestValue))
            {
                LOG(ERROR) << "read request json data error.";
                if (!is_primary)
                    break;
                continue;
            }
            if (requestValue.type() != Value::kObjectType)
            {
                LOG(ERROR) << "request value type error.";
                if (!is_primary)
                    break;
                continue;
            }
            Request request;
            request.assignTmp(requestValue);
            if (!ReqLogMgr::isReplayWriteReq(request.controller(), request.action()))
            {
                if (!is_primary)
                    break;
                continue;
            }
            std::string collection = asString(request[Keys::collection]);
            if (collection != from_col)
            {
                //LOG(INFO) << "not match collection : " << collection;
                if (!is_primary)
                    break;
                continue;
            }
            request[Keys::collection] = to_col;

            std::string replay_json;
            izenelib::driver::JsonWriter writer;
            writer.write(request.get(), replay_json);
            //LOG(INFO) << "replaying for request id : " << rethead.inc_id;
            CommonReqData new_common = req_commondata;
            new_common.req_json_data = replay_json;
            ReqLogMgr::replaceCommonReqData(req_commondata, new_common, req_packed_data);

            if(!DistributeDriver::get()->handleReqFromLog(req_commondata.reqtype, replay_json, req_packed_data))
            {
                LOG(INFO) << "ignore replaying from log failed: " << rethead.inc_id << ", json : " << req_commondata.req_json_data;
                if (!is_primary)
                    break;
                continue;
            }
            replayed_num++;
            if (is_primary)
                replayed_id_list.push_back(rethead.inc_id);
        }
        catch(const std::exception& e)
        {
            LOG(INFO) << "replay log exception: " << e.what() << ", req data: " << req_commondata.req_json_data;
            if (!is_primary)
                break;
        }
    }
    if (!is_primary && replayed_num != replayed_id_list.size())
    {
        LOG(ERROR) << "replay log error on secondary.";
        forceExit("Replay Log Failed!");
    }
    LOG(INFO) << "replay log finished, total replayed: " << replayed_num;
}

bool RecoveryChecker::redoLog(ReqLogMgr* redolog, uint32_t start_id, uint32_t end_id)
{
    bool ret = true;
    try
    {
        // read redo log and do request to current log.
        if (start_id > redolog->getLastSuccessReqId())
        {
            LOG(ERROR) << "backup id should not larger than last write id. " << start_id;
            LOG(ERROR) << "leaving old backup and do not redo log. restart and wait to sync to newest log";
            forceExit("redo start id is larger than last writed id.");
        }
        ReqLogHead rethead;
        size_t redo_offset = 0;
        if(!redolog->getHeadOffset(start_id, rethead, redo_offset))
        {
            assert(false);
            LOG(ERROR) << "get head failed. This should never happen.";
            forceExit("redo get head info failed.");
        }
        while(true)
        {
            std::string req_packed_data; 
            bool hasmore = redolog->getReqDataByHeadOffset(redo_offset, rethead, req_packed_data);
            if (!hasmore || rethead.inc_id >= end_id)
                break;
            LOG(INFO) << "redoing for request id : " << rethead.inc_id;
            CommonReqData req_commondata;
            ReqLogMgr::unpackReqLogData(req_packed_data, req_commondata);

            if(!DistributeDriver::get()->handleReqFromLog(req_commondata.reqtype, req_commondata.req_json_data, req_packed_data))
            {
                LOG(INFO) << "redoing from log failed: " << rethead.inc_id;
                forceExit("handleReqFromLog failed.");
            }
        }
    }
    catch(const std::exception& e)
    {
        LOG(INFO) << "redo log exception: " << e.what();
        ret = false;
    }
    bfs::remove_all(redo_log_basepath_);
    return ret;
}

bool RecoveryChecker::backupColl(const CollectionPath& colpath, const bfs::path& dest_path)
{
    // find all collection and backup.
    bfs::path dest_coldata_backup = dest_path/bfs::path("backup_data");

    if (!bfs::exists(dest_path))
    {
        bfs::create_directories(dest_path);
        bfs::create_directories(dest_coldata_backup);
    }

    bfs::path coldata_path(colpath.getCollectionDataPath());
    //bfs::path querydata_path(colpath.getQueryDataPath());

    try
    {
        copy_dir(coldata_path, dest_coldata_backup);
        //copy_dir(querydata_path, dest_coldata_backup);
    }
    catch(const std::exception& e)
    {
        LOG(ERROR) << "backup collection data failed. " << e.what() << std::endl;
        return false;
    }
    return true;
}

bool RecoveryChecker::hasAnyBackup(uint32_t& last_backup_id)
{
    std::string last_backup_path;
    last_backup_id = 0;
    bool has_backup = true;
    if (!getLastBackup(backup_basepath_, 0, last_backup_path, last_backup_id))
    {
        last_backup_id = 0;
        has_backup = false;
    }
    return has_backup;
}

bool RecoveryChecker::isNeedRollback(bool starting_up)
{
    if (!NodeManagerBase::get()->isDistributed())
        return false;
    if (bfs::exists(rollback_file_))
        return true;
    if (starting_up && !isLastNormalExit())
    {
        return true;
    }
    return false;
}

bool RecoveryChecker::checkAndRestoreBackupFile(const CollectionPath& colpath)
{
    LOG(INFO) << "check and restoring backup at startup.";

    if(!isNeedRollback(true))
    {
        return true;
    }

    uint32_t rollback_id = 0;
    ifstream ifs(rollback_file_.c_str());
    if (ifs.good())
    {
        ifs.read((char*)&rollback_id, sizeof(rollback_id));
    }

    std::string last_backup_path;
    uint32_t last_backup_id = 0;
    bool has_backup = true;
    if (!getLastBackup(backup_basepath_, rollback_id, last_backup_path, last_backup_id))
    {
        LOG(ERROR) << "no backup available while check startup.";
        LOG(ERROR) << "need restart to redo all log or get full data from primary.";
        last_backup_id = 0;
        has_backup = false;
        return false;
    }
    if (has_backup)
    {
        // copy backup data to replace current and reopen all file.
        try
        {
            bfs::path dest_coldata_backup(last_backup_path + "/backup_data");
            LOG(INFO) << "restoring the backup for the collection: " <<
                colpath.getCollectionDataPath() << ", backup id: " << last_backup_id;

            bfs::path coldata_path(colpath.getCollectionDataPath());
            //bfs::path querydata_path(colpath.getQueryDataPath());

            copy_dir(dest_coldata_backup/coldata_path, coldata_path);
            //copy_dir(dest_coldata_backup/querydata_path, querydata_path);
        }
        catch(const std::exception& e)
        {
            LOG(ERROR) << "copy backup to current working path failed." << e.what();
            return false;
        }
    }
    return true;
}

// note : if there are more than one collection we need backup
// all collection data. only in this way we can rollback and redo log to 
// rollback the data before last failed request.
bool RecoveryChecker::rollbackLastFail(bool starting_up)
{
    // rollback will only happen in recovery or abort request, both of them 
    // are protected by the lock of node state in NodeManagerBase. 
    // not all inc_id has a backup, so we just find the newest backup.
{'fraction_non_alphanumeric': 0.11717055464565534, 'fraction_numerical': 0.002456730556416272, 'mean_word_length': 2.226204926811853, 'pattern_counts': {'":': 0, '<': 419, '<?xml version=': 0, '>': 93, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6878711', 'n_tokens_mistral': 14033, 'n_tokens_neox': 13682, 'n_words': 2888}