<commit_before>#include "DistributeRequestHooker.h"
#include "DistributeDriver.h"
#include "RecoveryChecker.h"
#include "DistributeTest.hpp"

#include <boost/filesystem.hpp>
#include <util/driver/writers/JsonWriter.h>
#include <util/driver/readers/JsonReader.h>
#include <util/scheduler.h>
#include <bundles/index/IndexBundleConfiguration.h>
#include <node-manager/NodeManagerBase.h>

#include "RequestLog.h"

using namespace izenelib::driver;
using namespace izenelib;
namespace bfs = boost::filesystem;

namespace sf1r
{

std::set<ReqLogType> DistributeRequestHooker::need_backup_types_;

void DistributeRequestHooker::init()
{
    need_backup_types_.insert(Req_NoAdditionData_NeedBackup_Req);
    //need_backup_types_.insert(Req_CronJob);
    need_backup_types_.insert(Req_Index);
    // init callback for distribute request.
    NodeManagerBase::get()->setCallback(
        boost::bind(&DistributeRequestHooker::onElectingFinished, this),
        boost::bind(&DistributeRequestHooker::waitReplicasProcessCallback, this),
        boost::bind(&DistributeRequestHooker::waitReplicasLogCallback, this),
        boost::bind(&DistributeRequestHooker::waitPrimaryCallback, this),
        boost::bind(&DistributeRequestHooker::abortRequestCallback, this),
        boost::bind(&DistributeRequestHooker::waitReplicasAbortCallback, this),
        boost::bind(&DistributeRequestHooker::onRequestFromPrimary, this, _1, _2));
}

DistributeRequestHooker::DistributeRequestHooker()
    :type_(Req_None), hook_type_(0), chain_status_(NoChain)
{
}

bool DistributeRequestHooker::isNeedBackup(ReqLogType type)
{
    return need_backup_types_.find(type) != need_backup_types_.end();
}

bool DistributeRequestHooker::isValid()
{
    if (NodeManagerBase::get()->isDistributed() && hook_type_ == 0)
        return false;
    return true;
}

void DistributeRequestHooker::hookCurrentReq(const std::string& reqdata)
{
    if (req_log_mgr_)
    {
        // for the request that will shard to different node, 
        // it is normal to be hook twice. 
        // If no sharding it should be some wrong if hook again.
        LOG(INFO) << "hooked request again!!";
    }
    current_req_ = reqdata;
    req_log_mgr_ = RecoveryChecker::get()->getReqLogMgr();
    LOG(INFO) << "current request hooked: " << current_req_;
}

bool DistributeRequestHooker::onRequestFromPrimary(int type, const std::string& packed_reqdata)
{
    LOG(INFO) << "callback for new request from primary, packeddata len: " << packed_reqdata.size();
    CommonReqData reqloghead;
    if(!ReqLogMgr::unpackReqLogData(packed_reqdata, reqloghead))
    {
        LOG(ERROR) << "unpack request data from primary failed. data: " << packed_reqdata;
        // return false to abortRequest.
        return false;
    }
    bool ret = true;
    ret = DistributeDriver::get()->handleReqFromPrimary(reqloghead.reqtype, reqloghead.req_json_data, packed_reqdata);
    if (!ret)
    {
        LOG(ERROR) << "send request come from primary failed in replica. " << reqloghead.req_json_data;
        return false;
    }
    LOG(INFO) << "send the request come from primary success in replica.";
    return true;
}

void DistributeRequestHooker::setHook(int calltype, const std::string& addition_data)
{
    hook_type_ = calltype;
    // for request for primary master, the addition_data is the original request json data.
    // for request from primary worker to replicas, the addition_data is original request
    // json data plus the data used for this request.
    current_req_ = addition_data;
    LOG(INFO) << "setting hook : " << hook_type_ << ", data:" << current_req_;
    DistributeTestSuit::loadTestConf();
}

int  DistributeRequestHooker::getHookType()
{
    return hook_type_;
}

bool DistributeRequestHooker::isHooked()
{
    return (hook_type_ > 0) && !current_req_.empty();
}

bool DistributeRequestHooker::readPrevChainData(CommonReqData& reqlogdata)
{
    if (!isHooked())
        return true;
    if (chain_status_ == NoChain || chain_status_ == ChainBegin)
        return true;
    if (!ReqLogMgr::unpackReqLogData(current_req_, reqlogdata))
    {
        LOG(ERROR) << "unpack log data error while read from chain data.";
        return false;
    }
    return true;
}

bool DistributeRequestHooker::prepare(ReqLogType type, CommonReqData& prepared_req)
{
    if (!isHooked())
        return true;
    assert(req_log_mgr_);
    bool isprimary = (hook_type_ == Request::FromDistribute);
    if (isprimary)
    {
        if (chain_status_ == ChainBegin || chain_status_ == NoChain)
            prepared_req.req_json_data = current_req_;
    }
    else
    {
        // get addition data from primary
        if(!ReqLogMgr::unpackReqLogData(current_req_, prepared_req))
        {
            LOG(ERROR) << "unpack log data failed while prepare the data from primary.";
            forceExit();
        }
        if (type != (ReqLogType)prepared_req.reqtype)
        {
            LOG(ERROR) << "log type mismatch with primary while prepare the data from primary.";
            LOG(ERROR) << "It may happen when the code is not the same. Must exit.";
            forceExit();
        }
        LOG(INFO) << "got hooked write request, inc_id :" << prepared_req.inc_id;
    }
    prepared_req.reqtype = type;
    type_ = type;
    LOG(INFO) << "begin prepare log ";
    if (chain_status_ == NoChain || chain_status_ == ChainBegin)
    {
        if (!req_log_mgr_->prepareReqLog(prepared_req, isprimary))
        {
            LOG(ERROR) << "prepare request log failed.";
            if (!isprimary)
            {
                assert(false);
                forceExit();
            }
            finish(false);
            return false;
        }
    }
    
    if (isprimary)
    {
        bool needpack = true;
        if (type == Req_NoAdditionData_NeedBackup_Req || type == Req_NoAdditionDataReq ||
            type == Req_NoAdditionDataNoRollback)
        {
            if (chain_status_ != NoChain && chain_status_ != ChainBegin)
                needpack = false;
        }
        if (needpack)
        {
            // save primary prepared data to current_req_ and 
            // after primary finished, send it to replica.
            ReqLogMgr::packReqLogData(prepared_req, current_req_);
        }
    }

    if (chain_status_ != NoChain && chain_status_ != ChainBegin)
    {
        // only prepare at the begin of chain request.
        LOG(INFO) << "no need prepare for middle of request chain.";
        return true;
    }

    if (isNeedBackup(type))
    {
        LOG(INFO) << "begin backup";
        if(!RecoveryChecker::get()->backup())
        {
            LOG(ERROR) << "backup failed. Maybe not enough space.";
            if (!isprimary)
            {
                forceExit();
            }
            finish(false);
            return false;
        }
    }
    // set rollback flag.
    if(type != Req_NoAdditionDataNoRollback && !RecoveryChecker::get()->setRollbackFlag(prepared_req.inc_id))
    {
        LOG(ERROR) << "set rollback failed.";
        if (!isprimary)
        {
            forceExit();
        }
        finish(false);
        return false; 
    }
    if (isprimary)
        DistributeTestSuit::testFail(PrimaryFail_At_PrepareFinished);
    else
        DistributeTestSuit::testFail(ReplicaFail_At_PrepareFinished);
    return true;
}

void DistributeRequestHooker::processLocalBegin()
{
    if (!isHooked())
        return;
    if (hook_type_ == Request::FromLog)
        return;
    if (chain_status_ != NoChain && chain_status_ != ChainBegin)
    {
        // only need change node state at the begin of chain request.
        LOG(INFO) << "one of chain request begin local";
        return;
    }
    LOG(INFO) << "begin process request on worker";
    NodeManagerBase::get()->beginReqProcess();
}

bool DistributeRequestHooker::processFailedBeforePrepare()
{
    if (!isHooked())
        return false;
    if (hook_type_ == Request::FromLog)
    {
        LOG(ERROR) << "redo log failed finished, must exit.";
        forceExit();
    }
    static CommonReqData reqlog;
    if (hook_type_ == Request::FromDistribute && !req_log_mgr_->getPreparedReqLog(reqlog))
    {
        LOG(INFO) << "primary end request before prepared, request ignored.";
        NodeManagerBase::get()->notifyMasterReadyForNew();
        clearHook(true);
        return true;
    }
    return false;
}

void DistributeRequestHooker::processLocalFinished(bool finishsuccess, CommonReqData& updated_preparedata)
{
    // if the prepared data has changed during processing, 
    // we need update the current_req_ before send it to replicas. 
    ReqLogMgr::packReqLogData(updated_preparedata, current_req_);
    processLocalFinished(finishsuccess);
}

void DistributeRequestHooker::processLocalFinished(bool finishsuccess)
{
    if (!isHooked())
        return;
    //current_req_ = packed_req_data;
    if (!finishsuccess)
    {
        if (processFailedBeforePrepare())
            return;
        LOG(INFO) << "process finished fail.";
        abortRequest();
        return;
    }

    if (chain_status_ != NoChain && chain_status_ != ChainEnd)
    {
        LOG(INFO) << "request chain finished one of chain that not end.";
        return;
    }

    LOG(INFO) << "process request on local worker finished.";
    if (hook_type_ == Request::FromLog)
    {
        writeLocalLog();
        return;
    }
    // need set rollback flag after primary finished, even no rollback type.
    if(type_ == Req_NoAdditionDataNoRollback)
    {
	    CommonReqData reqlog;
	    req_log_mgr_->getPreparedReqLog(reqlog);
	    if(!RecoveryChecker::get()->setRollbackFlag(reqlog.inc_id))
	    {
		    LOG(ERROR) << "set rollback failed after finish local.this node need restore by human.";
		    forceExit();
	    }
    }
 
    LOG(INFO) << "send packed request data len from local. len: " << current_req_.size();
    NodeManagerBase::get()->finishLocalReqProcess(type_, current_req_);
}

void DistributeRequestHooker::waitReplicasProcessCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished the request. Begin write log";
}

void DistributeRequestHooker::waitPrimaryCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "got respond from primary while waiting write log after finished on replica.";
    writeLocalLog();
}

void DistributeRequestHooker::abortRequest()
{
    if (!isHooked())
        return;
    if (hook_type_ == Request::FromLog)
    {
        LOG(ERROR) << "redo log failed, must exit.";
        forceExit();
        return;
    }
    chain_status_ = ChainStop;
    LOG(INFO) << "abortRequest...";
    NodeManagerBase::get()->abortRequest();
}

void DistributeRequestHooker::abortRequestCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "got abort request.";
    finish(false);
}

void DistributeRequestHooker::waitReplicasAbortCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished the abort request.";
    finish(false);
}

void DistributeRequestHooker::writeLocalLog()
{
    if (!isHooked())
        return;

    assert(chain_status_ == ChainEnd || chain_status_ == NoChain);
    LOG(INFO) << "begin write request log to local.";

    bool ret = req_log_mgr_->appendReqData(current_req_);
    if (!ret)
    {
        LOG(ERROR) << "write local log failed. something wrong on this node, must exit.";
        forceExit();
    }
    finish(true);
}

void DistributeRequestHooker::waitReplicasLogCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished write request log to local.";
    writeLocalLog();
}

void DistributeRequestHooker::onElectingFinished()
{
    LOG(INFO) << "an electing has finished. notify ready to master.";
}

void DistributeRequestHooker::finish(bool success)
{
    assert(chain_status_ == ChainEnd || chain_status_ == NoChain);
    if (hook_type_ == Request::FromLog && !success)
    {
        LOG(ERROR) << "redo log failed. must exit";
        forceExit();
    }

    CommonReqData reqlog;
    req_log_mgr_->getPreparedReqLog(reqlog);

    req_log_mgr_->delPreparedReqLog();
    clearHook(true);
    if (success)
    {
        LOG(INFO) << "The request has finally finished both on primary and replicas.";
        RecoveryChecker::get()->clearRollbackFlag();
        DistributeTestSuit::updateMemoryState("Last_Success_Request", reqlog.inc_id);
    }
    else
    {
        DistributeTestSuit::updateMemoryState("Last_Failed_Request", reqlog.inc_id);
        LOG(INFO) << "The request failed to finish. rollback from backup.";
        // rollback from backup.
        // all the file need to be reopened to make effective.
        if (!RecoveryChecker::get()->rollbackLastFail())
        {
            LOG(ERROR) << "failed to rollback ! must exit.";
            forceExit();
        }
    }
    LOG(INFO) << DistributeTestSuit::getStatusReport();
}

void DistributeRequestHooker::clearHook(bool force)
{
    static CommonReqData reqlog;
    if (!force && req_log_mgr_->getPreparedReqLog(reqlog))
    {
        // if prepared , we will clear hook after finish .
        return;
    }
    current_req_.clear();
    req_log_mgr_.reset();
    hook_type_ = 0;
    chain_status_ = NoChain;
    LOG(INFO) << "request hook cleard.";
}

void DistributeRequestHooker::forceExit()
{
    RecoveryChecker::forceExit("force exit in DistributeRequestHooker");
}

}

<commit_msg>add test point for unpack  request data fail<commit_after>#include "DistributeRequestHooker.h"
#include "DistributeDriver.h"
#include "RecoveryChecker.h"
#include "DistributeTest.hpp"

#include <boost/filesystem.hpp>
#include <util/driver/writers/JsonWriter.h>
#include <util/driver/readers/JsonReader.h>
#include <util/scheduler.h>
#include <bundles/index/IndexBundleConfiguration.h>
#include <node-manager/NodeManagerBase.h>

#include "RequestLog.h"

using namespace izenelib::driver;
using namespace izenelib;
namespace bfs = boost::filesystem;

namespace sf1r
{

std::set<ReqLogType> DistributeRequestHooker::need_backup_types_;

void DistributeRequestHooker::init()
{
    need_backup_types_.insert(Req_NoAdditionData_NeedBackup_Req);
    //need_backup_types_.insert(Req_CronJob);
    need_backup_types_.insert(Req_Index);
    // init callback for distribute request.
    NodeManagerBase::get()->setCallback(
        boost::bind(&DistributeRequestHooker::onElectingFinished, this),
        boost::bind(&DistributeRequestHooker::waitReplicasProcessCallback, this),
        boost::bind(&DistributeRequestHooker::waitReplicasLogCallback, this),
        boost::bind(&DistributeRequestHooker::waitPrimaryCallback, this),
        boost::bind(&DistributeRequestHooker::abortRequestCallback, this),
        boost::bind(&DistributeRequestHooker::waitReplicasAbortCallback, this),
        boost::bind(&DistributeRequestHooker::onRequestFromPrimary, this, _1, _2));
}

DistributeRequestHooker::DistributeRequestHooker()
    :type_(Req_None), hook_type_(0), chain_status_(NoChain)
{
}

bool DistributeRequestHooker::isNeedBackup(ReqLogType type)
{
    return need_backup_types_.find(type) != need_backup_types_.end();
}

bool DistributeRequestHooker::isValid()
{
    if (NodeManagerBase::get()->isDistributed() && hook_type_ == 0)
        return false;
    return true;
}

void DistributeRequestHooker::hookCurrentReq(const std::string& reqdata)
{
    if (req_log_mgr_)
    {
        // for the request that will shard to different node, 
        // it is normal to be hook twice. 
        // If no sharding it should be some wrong if hook again.
        LOG(INFO) << "hooked request again!!";
    }
    current_req_ = reqdata;
    req_log_mgr_ = RecoveryChecker::get()->getReqLogMgr();
    LOG(INFO) << "current request hooked, data len: " << current_req_.size() << ", data:" << current_req_;
}

bool DistributeRequestHooker::onRequestFromPrimary(int type, const std::string& packed_reqdata)
{
    LOG(INFO) << "callback for new request from primary, packeddata len: " << packed_reqdata.size();
    CommonReqData reqloghead;
    if(!ReqLogMgr::unpackReqLogData(packed_reqdata, reqloghead))
    {
        LOG(ERROR) << "unpack request data from primary failed. data: " << packed_reqdata;
        // return false to abortRequest.
        return false;
    }
    bool ret = true;
    ret = DistributeDriver::get()->handleReqFromPrimary(reqloghead.reqtype, reqloghead.req_json_data, packed_reqdata);
    if (!ret)
    {
        LOG(ERROR) << "send request come from primary failed in replica. " << reqloghead.req_json_data;
        return false;
    }
    LOG(INFO) << "send the request come from primary success in replica.";
    return true;
}

void DistributeRequestHooker::setHook(int calltype, const std::string& addition_data)
{
    hook_type_ = calltype;
    // for request for primary master, the addition_data is the original request json data.
    // for request from primary worker to replicas, the addition_data is original request
    // json data plus the data used for this request.
    current_req_ = addition_data;
    LOG(INFO) << "setting hook : " << hook_type_ << ", data:" << current_req_;
    DistributeTestSuit::loadTestConf();
}

int  DistributeRequestHooker::getHookType()
{
    return hook_type_;
}

bool DistributeRequestHooker::isHooked()
{
    return (hook_type_ > 0) && !current_req_.empty();
}

bool DistributeRequestHooker::readPrevChainData(CommonReqData& reqlogdata)
{
    if (!isHooked())
        return true;
    if (chain_status_ == NoChain || chain_status_ == ChainBegin)
        return true;
    if (!ReqLogMgr::unpackReqLogData(current_req_, reqlogdata))
    {
        LOG(ERROR) << "unpack log data error while read from chain data.";
        return false;
    }
    return true;
}

bool DistributeRequestHooker::prepare(ReqLogType type, CommonReqData& prepared_req)
{
    if (!isHooked())
        return true;
    assert(req_log_mgr_);
    bool isprimary = (hook_type_ == Request::FromDistribute);
    if (isprimary)
    {
        if (chain_status_ == ChainBegin || chain_status_ == NoChain)
            prepared_req.req_json_data = current_req_;
    }
    else
    {
        DistributeTestSuit::testFail(ReplicaFail_At_UnpackPrimaryReq);
        // get addition data from primary
        if(!ReqLogMgr::unpackReqLogData(current_req_, prepared_req))
        {
            LOG(ERROR) << "unpack log data failed while prepare the data from primary.";
            forceExit();
        }
        if (type != (ReqLogType)prepared_req.reqtype)
        {
            LOG(ERROR) << "log type mismatch with primary while prepare the data from primary.";
            LOG(ERROR) << "It may happen when the code is not the same. Must exit.";
            forceExit();
        }
        LOG(INFO) << "got hooked write request, inc_id :" << prepared_req.inc_id;
    }
    prepared_req.reqtype = type;
    type_ = type;
    LOG(INFO) << "begin prepare log ";
    if (chain_status_ == NoChain || chain_status_ == ChainBegin)
    {
        if (!req_log_mgr_->prepareReqLog(prepared_req, isprimary))
        {
            LOG(ERROR) << "prepare request log failed.";
            if (!isprimary)
            {
                assert(false);
                forceExit();
            }
            finish(false);
            return false;
        }
    }
    
    if (isprimary)
    {
        bool needpack = true;
        if (type == Req_NoAdditionData_NeedBackup_Req || type == Req_NoAdditionDataReq ||
            type == Req_NoAdditionDataNoRollback)
        {
            if (chain_status_ != NoChain && chain_status_ != ChainBegin)
                needpack = false;
        }
        if (needpack)
        {
            // save primary prepared data to current_req_ and 
            // after primary finished, send it to replica.
            ReqLogMgr::packReqLogData(prepared_req, current_req_);
        }
    }

    if (chain_status_ != NoChain && chain_status_ != ChainBegin)
    {
        // only prepare at the begin of chain request.
        LOG(INFO) << "no need prepare for middle of request chain.";
        return true;
    }

    if (isNeedBackup(type))
    {
        LOG(INFO) << "begin backup";
        if(!RecoveryChecker::get()->backup())
        {
            LOG(ERROR) << "backup failed. Maybe not enough space.";
            if (!isprimary)
            {
                forceExit();
            }
            finish(false);
            return false;
        }
    }
    // set rollback flag.
    if(type != Req_NoAdditionDataNoRollback && !RecoveryChecker::get()->setRollbackFlag(prepared_req.inc_id))
    {
        LOG(ERROR) << "set rollback failed.";
        if (!isprimary)
        {
            forceExit();
        }
        finish(false);
        return false; 
    }
    if (isprimary)
        DistributeTestSuit::testFail(PrimaryFail_At_PrepareFinished);
    else
        DistributeTestSuit::testFail(ReplicaFail_At_PrepareFinished);
    return true;
}

void DistributeRequestHooker::processLocalBegin()
{
    if (!isHooked())
        return;
    if (hook_type_ == Request::FromLog)
        return;
    if (chain_status_ != NoChain && chain_status_ != ChainBegin)
    {
        // only need change node state at the begin of chain request.
        LOG(INFO) << "one of chain request begin local";
        return;
    }
    LOG(INFO) << "begin process request on worker";
    NodeManagerBase::get()->beginReqProcess();
}

bool DistributeRequestHooker::processFailedBeforePrepare()
{
    if (!isHooked())
        return false;
    if (hook_type_ == Request::FromLog)
    {
        LOG(ERROR) << "redo log failed finished, must exit.";
        forceExit();
    }
    static CommonReqData reqlog;
    if (hook_type_ == Request::FromDistribute && !req_log_mgr_->getPreparedReqLog(reqlog))
    {
        LOG(INFO) << "primary end request before prepared, request ignored.";
        NodeManagerBase::get()->notifyMasterReadyForNew();
        clearHook(true);
        return true;
    }
    return false;
}

void DistributeRequestHooker::processLocalFinished(bool finishsuccess, CommonReqData& updated_preparedata)
{
    // if the prepared data has changed during processing, 
    // we need update the current_req_ before send it to replicas. 
    ReqLogMgr::packReqLogData(updated_preparedata, current_req_);
    processLocalFinished(finishsuccess);
}

void DistributeRequestHooker::processLocalFinished(bool finishsuccess)
{
    if (!isHooked())
        return;
    //current_req_ = packed_req_data;
    if (!finishsuccess)
    {
        if (processFailedBeforePrepare())
            return;
        LOG(INFO) << "process finished fail.";
        abortRequest();
        return;
    }

    if (chain_status_ != NoChain && chain_status_ != ChainEnd)
    {
        LOG(INFO) << "request chain finished one of chain that not end.";
        return;
    }

    LOG(INFO) << "process request on local worker finished.";
    if (hook_type_ == Request::FromLog)
    {
        writeLocalLog();
        return;
    }
    // need set rollback flag after primary finished, even no rollback type.
    if(type_ == Req_NoAdditionDataNoRollback)
    {
	    CommonReqData reqlog;
	    req_log_mgr_->getPreparedReqLog(reqlog);
	    if(!RecoveryChecker::get()->setRollbackFlag(reqlog.inc_id))
	    {
		    LOG(ERROR) << "set rollback failed after finish local.this node need restore by human.";
		    forceExit();
	    }
    }
 
    LOG(INFO) << "send packed request data len from local. len: " << current_req_.size();
    NodeManagerBase::get()->finishLocalReqProcess(type_, current_req_);
}

void DistributeRequestHooker::waitReplicasProcessCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished the request. Begin write log";
}

void DistributeRequestHooker::waitPrimaryCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "got respond from primary while waiting write log after finished on replica.";
    writeLocalLog();
}

void DistributeRequestHooker::abortRequest()
{
    if (!isHooked())
        return;
    if (hook_type_ == Request::FromLog)
    {
        LOG(ERROR) << "redo log failed, must exit.";
        forceExit();
        return;
    }
    chain_status_ = ChainStop;
    LOG(INFO) << "abortRequest...";
    NodeManagerBase::get()->abortRequest();
}

void DistributeRequestHooker::abortRequestCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "got abort request.";
    finish(false);
}

void DistributeRequestHooker::waitReplicasAbortCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished the abort request.";
    finish(false);
}

void DistributeRequestHooker::writeLocalLog()
{
    if (!isHooked())
        return;

    assert(chain_status_ == ChainEnd || chain_status_ == NoChain);
    LOG(INFO) << "begin write request log to local.";

    bool ret = req_log_mgr_->appendReqData(current_req_);
    if (!ret)
    {
        LOG(ERROR) << "write local log failed. something wrong on this node, must exit.";
        forceExit();
    }
    finish(true);
}

void DistributeRequestHooker::waitReplicasLogCallback()
{
    if (!isHooked())
        return;
    LOG(INFO) << "all replicas finished write request log to local.";
    writeLocalLog();
}

void DistributeRequestHooker::onElectingFinished()
{
    LOG(INFO) << "an electing has finished. notify ready to master.";
}

void DistributeRequestHooker::finish(bool success)
{
    assert(chain_status_ == ChainEnd || chain_status_ == NoChain);
    if (hook_type_ == Request::FromLog && !success)
    {
        LOG(ERROR) << "redo log failed. must exit";
        forceExit();
    }

    CommonReqData reqlog;
    req_log_mgr_->getPreparedReqLog(reqlog);

    req_log_mgr_->delPreparedReqLog();
    clearHook(true);
    if (success)
    {
        LOG(INFO) << "The request has finally finished both on primary and replicas.";
        RecoveryChecker::get()->clearRollbackFlag();
        DistributeTestSuit::updateMemoryState("Last_Success_Request", reqlog.inc_id);
    }
    else
    {
        DistributeTestSuit::updateMemoryState("Last_Failed_Request", reqlog.inc_id);
        LOG(INFO) << "The request failed to finish. rollback from backup.";
        // rollback from backup.
        // all the file need to be reopened to make effective.
        if (!RecoveryChecker::get()->rollbackLastFail())
        {
            LOG(ERROR) << "failed to rollback ! must exit.";
            forceExit();
        }
    }
    LOG(INFO) << DistributeTestSuit::getStatusReport();
}

void DistributeRequestHooker::clearHook(bool force)
{
    static CommonReqData reqlog;
    if (!force && req_log_mgr_->getPreparedReqLog(reqlog))
    {
        // if prepared , we will clear hook after finish .
        return;
    }
    current_req_.clear();
    req_log_mgr_.reset();
    hook_type_ = 0;
    chain_status_ = NoChain;
    LOG(INFO) << "request hook cleard.";
}

void DistributeRequestHooker::forceExit()
{
    RecoveryChecker::forceExit("force exit in DistributeRequestHooker");
}

}


{'fraction_non_alphanumeric': 0.11470850022251891, 'fraction_numerical': 0.0005192107995846313, 'mean_word_length': 3.0114549241297235, 'pattern_counts': {'":': 0, '<': 229, '<?xml version=': 0, '>': 59, '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': '6609095', 'n_tokens_mistral': 8028, 'n_tokens_neox': 7978, 'n_words': 1888}