replica_backup.cpp

// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you 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 <boost/cstdint.hpp>
// IWYU pragma: no_include <boost/detail/basic_pointerbuf.hpp>
#include <boost/lexical_cast.hpp>
// IWYU pragma: no_include <ext/alloc_traits.h>
#include <inttypes.h>
#include <stdio.h>
#include <algorithm>
#include <chrono>
#include <cstdint>
#include <ios>
#include <map>
#include <memory>
#include <string>
#include <type_traits>
#include <utility>
#include <vector>

#include "backup/cold_backup_context.h"
#include "backup/replica_backup_manager.h"
#include "backup_types.h"
#include "block_service/block_service_manager.h"
#include "common/gpid.h"
#include "common/replication.codes.h"
#include "common/replication_enums.h"
#include "common/replication_other_types.h"
#include "dsn.layer2_types.h"
#include "metadata_types.h"
#include "replica.h"
#include "replica/replica_context.h"
#include "replica/replication_app_base.h"
#include "replica_stub.h"
#include "runtime/api_layer1.h"
#include "task/async_calls.h"
#include "utils/autoref_ptr.h"
#include "utils/env.h"
#include "utils/error_code.h"
#include "utils/filesystem.h"
#include "utils/flags.h"
#include "utils/fmt_logging.h"
#include "utils/metrics.h"
#include "utils/strings.h"
#include "utils/thread_access_checker.h"
#include "utils/time_utils.h"

DSN_DEFINE_uint64(replication,
                  max_concurrent_uploading_file_count,
                  10,
                  "concurrent uploading file count to block service");

DSN_DECLARE_string(cold_backup_root);

namespace dsn {
namespace dist {
namespace block_service {
class block_filesystem;
} // namespace block_service
} // namespace dist

namespace replication {
void replica::on_cold_backup(const backup_request &request, /*out*/ backup_response &response)
{
    _checker.only_one_thread_access();

    const std::string &policy_name = request.policy.policy_name;
    auto backup_id = request.backup_id;
    cold_backup_context_ptr new_context(
        new cold_backup_context(this, request, FLAGS_max_concurrent_uploading_file_count));

    LOG_INFO_PREFIX("{}: received cold backup request, partition_status = {}",
                    new_context->name,
                    enum_to_string(status()));

    if (status() == partition_status::type::PS_PRIMARY ||
        status() == partition_status::type::PS_SECONDARY) {
        cold_backup_context_ptr backup_context = nullptr;
        auto find = _cold_backup_contexts.find(policy_name);
        if (find != _cold_backup_contexts.end()) {
            backup_context = find->second;
        } else {
            /// TODO: policy may change provider
            dist::block_service::block_filesystem *block_service =
                _stub->_block_service_manager.get_or_create_block_filesystem(
                    request.policy.backup_provider_type);
            if (block_service == nullptr) {
                LOG_ERROR(
                    "{}: create cold backup block service failed, provider_type = {}, response "
                    "ERR_INVALID_PARAMETERS",
                    new_context->name,
                    request.policy.backup_provider_type);
                response.err = ERR_INVALID_PARAMETERS;
                return;
            }
            auto r = _cold_backup_contexts.insert(std::make_pair(policy_name, new_context));
            CHECK(r.second, "");
            backup_context = r.first->second;
            backup_context->block_service = block_service;
            backup_context->backup_root = request.__isset.backup_path
                                              ? dsn::utils::filesystem::path_combine(
                                                    request.backup_path, FLAGS_cold_backup_root)
                                              : FLAGS_cold_backup_root;
        }

        CHECK_EQ_PREFIX(backup_context->request.policy.policy_name, policy_name);
        cold_backup_status backup_status = backup_context->status();

        if (backup_context->request.backup_id < backup_id || backup_status == ColdBackupCanceled) {
            if (backup_status == ColdBackupCheckpointing) {
                LOG_INFO("{}: delay clearing obsoleted cold backup context, cause backup_status == "
                         "ColdBackupCheckpointing",
                         new_context->name);
                tasking::enqueue(
                    LPC_REPLICATION_COLD_BACKUP,
                    &_tracker,
                    [this, request]() {
                        backup_response response;
                        on_cold_backup(request, response);
                    },
                    get_gpid().thread_hash(),
                    std::chrono::seconds(100));
            } else {
                // TODO(wutao1): deleting cold backup context should be
                //               extracted as a function like try_delete_cold_backup_context;
                // clear obsoleted backup context firstly
                LOG_INFO("{}: clear obsoleted cold backup context, old_backup_id = {}, "
                         "old_backup_status = {}",
                         new_context->name,
                         backup_context->request.backup_id,
                         cold_backup_status_to_string(backup_status));
                backup_context->cancel();
                _cold_backup_contexts.erase(policy_name);
                // go to another round
                on_cold_backup(request, response);
            }
            return;
        }

        if (backup_context->request.backup_id > backup_id) {
            // backup_id is outdated
            LOG_ERROR("{}: request outdated cold backup, current_backup_id = {}, response "
                      "ERR_VERSION_OUTDATED",
                      new_context->name,
                      backup_context->request.backup_id);
            response.err = ERR_VERSION_OUTDATED;
            return;
        }

        // for secondary, request is already filtered by primary, so if
        //      request is repeated, so generate_backup_checkpoint is already running, we do
        //      nothing;
        //      request is new, we should call generate_backup_checkpoint;

        // TODO: if secondary's status have changed, how to process the _cold_backup_state,
        // and how to process the backup_status, cancel/pause
        if (status() == partition_status::PS_SECONDARY) {
            if (backup_status == ColdBackupInvalid) {
                // new backup_request, should set status to ColdBackupChecked to allow secondary
                // can start to checkpoint
                backup_context->start_check();
                backup_context->complete_check(false);
                if (backup_context->start_checkpoint()) {
                    METRIC_VAR_INCREMENT(backup_started_count);
                    tasking::enqueue(
                        LPC_BACKGROUND_COLD_BACKUP, &_tracker, [this, backup_context]() {
                            generate_backup_checkpoint(backup_context);
                        });
                }
            }
            return;
        }

        send_backup_request_to_secondary(request);

        if (backup_status == ColdBackupChecking || backup_status == ColdBackupCheckpointing ||
            backup_status == ColdBackupUploading) {
            // do nothing
            LOG_INFO("{}: backup is busy, status = {}, progress = {}, response ERR_BUSY",
                     backup_context->name,
                     cold_backup_status_to_string(backup_status),
                     backup_context->progress());
            response.err = ERR_BUSY;
        } else if (backup_status == ColdBackupInvalid && backup_context->start_check()) {
            METRIC_VAR_INCREMENT(backup_started_count);
            LOG_INFO("{}: start checking backup on remote, response ERR_BUSY",
                     backup_context->name);
            tasking::enqueue(LPC_BACKGROUND_COLD_BACKUP, nullptr, [backup_context]() {
                backup_context->check_backup_on_remote();
            });
            response.err = ERR_BUSY;
        } else if (backup_status == ColdBackupChecked && backup_context->start_checkpoint()) {
            // start generating checkpoint
            LOG_INFO("{}: start generating checkpoint, response ERR_BUSY", backup_context->name);
            tasking::enqueue(LPC_BACKGROUND_COLD_BACKUP, &_tracker, [this, backup_context]() {
                generate_backup_checkpoint(backup_context);
            });
            response.err = ERR_BUSY;
        } else if ((backup_status == ColdBackupCheckpointed || backup_status == ColdBackupPaused) &&
                   backup_context->start_upload()) {
            // start uploading checkpoint
            LOG_INFO("{}: start uploading checkpoint, response ERR_BUSY", backup_context->name);
            tasking::enqueue(LPC_BACKGROUND_COLD_BACKUP, nullptr, [backup_context]() {
                backup_context->upload_checkpoint_to_remote();
            });
            response.err = ERR_BUSY;
        } else if (backup_status == ColdBackupFailed) {
            LOG_ERROR("{}: upload checkpoint failed, reason = {}, response ERR_LOCAL_APP_FAILURE",
                      backup_context->name,
                      backup_context->reason());
            response.err = ERR_LOCAL_APP_FAILURE;
            backup_context->cancel();
            _cold_backup_contexts.erase(policy_name);
        } else if (backup_status == ColdBackupCompleted) {
            LOG_INFO("{}: upload checkpoint completed, response ERR_OK", backup_context->name);
            _backup_mgr->send_clear_request_to_secondaries(backup_context->request.pid,
                                                           policy_name);

            // clear local checkpoint dirs in background thread
            _backup_mgr->background_clear_backup_checkpoint(policy_name);
            response.err = ERR_OK;
        } else {
            LOG_WARNING(
                "{}: unhandled case, handle_status = {}, real_time_status = {}, response ERR_BUSY",
                backup_context->name,
                cold_backup_status_to_string(backup_status),
                cold_backup_status_to_string(backup_context->status()));
            response.err = ERR_BUSY;
        }

        response.progress = backup_context->progress();
        response.checkpoint_total_size = backup_context->get_checkpoint_total_size();
        LOG_INFO("{}: backup progress is {}", backup_context->name, response.progress);
    } else {
        LOG_ERROR(
            "{}: invalid state for cold backup, partition_status = {}, response ERR_INVALID_STATE",
            new_context->name,
            enum_to_string(status()));
        response.err = ERR_INVALID_STATE;
    }
}

void replica::send_backup_request_to_secondary(const backup_request &request)
{
    for (const auto &secondary : _primary_states.pc.secondaries) {
        // primary will send backup_request to secondary periodically
        // so, we shouldn't handle the response
        rpc::call_one_way_typed(secondary, RPC_COLD_BACKUP, request, get_gpid().thread_hash());
    }
}

// backup/backup.<policy_name>.<backup_id>.<decree>.<timestamp>
static std::string backup_get_dir_name(const std::string &policy_name,
                                       int64_t backup_id,
                                       int64_t decree,
                                       int64_t timestamp)
{
    char buffer[256];
    sprintf(buffer,
            "backup.%s.%" PRId64 ".%" PRId64 ".%" PRId64 "",
            policy_name.c_str(),
            backup_id,
            decree,
            timestamp);
    return std::string(buffer);
}

// backup/backup_tmp.<policy_name>.<backup_id>.<timestamp>
static std::string
backup_get_tmp_dir_name(const std::string &policy_name, int64_t backup_id, int64_t timestamp)
{
    char buffer[256];
    sprintf(
        buffer, "backup_tmp.%s.%" PRId64 ".%" PRId64 "", policy_name.c_str(), backup_id, timestamp);
    return std::string(buffer);
}

// returns:
//   0 : not related
//   1 : related (belong to this policy but not belong to this backup_context)
//   2 : valid (belong to this policy and belong to this backup_context)
static int is_related_or_valid_checkpoint(const std::string &chkpt_dirname,
                                          const cold_backup_context_ptr &backup_context)
{
    std::vector<std::string> strs;
    ::dsn::utils::split_args(chkpt_dirname.c_str(), strs, '.');
    if (strs.size() == 4 && strs[0] == std::string("backup_tmp") &&
        strs[1] == backup_context->request.policy.policy_name) {
        // backup_tmp.<policy_name>.<backup_id>.<timestamp>
        // refer to backup_get_tmp_dir_name().
        int64_t backup_id = boost::lexical_cast<int64_t>(strs[2]);
        if (backup_id < backup_context->request.backup_id) {
            // it belongs to old backup_context, we can remove it safely.
            return 1;
        }
    } else if (strs.size() == 5 && strs[0] == std::string("backup_tmp") &&
               strs[4] == std::string("tmp") &&
               strs[1] == backup_context->request.policy.policy_name) {
        // backup_tmp.<policy_name>.<backup_id>.<timestamp>.tmp
        // refer to CheckpointImpl::CreateCheckpointQuick().
        int64_t backup_id = boost::lexical_cast<int64_t>(strs[2]);
        if (backup_id < backup_context->request.backup_id) {
            // it belongs to old backup_context, we can remove it safely.
            return 1;
        }
    } else if (strs.size() == 5 && strs[0] == std::string("backup") &&
               strs[1] == backup_context->request.policy.policy_name) {
        // backup.<policy_name>.<backup_id>.<decree>.<timestamp>
        // refer to backup_get_dir_name().
        int64_t backup_id = boost::lexical_cast<int64_t>(strs[2]);
        // here, we only need policy_name and backup_id to verify whether chkpt_dirname belong
        // to this backup_context.
        if (backup_id == backup_context->request.backup_id) {
            // it belongs to this backup_context.
            return 2;
        } else if (backup_id < backup_context->request.backup_id) {
            // it belongs to old backup_context, we can remove it safely.
            return 1;
        }
    } else {
        // unknown dir, ignore it
        LOG_WARNING("{}: found a invalid checkpoint dir({})", backup_context->name, chkpt_dirname);
    }
    return 0;
}

// filter backup checkpoint under 'dir'
//  - find the valid backup checkpoint dir if exist
//  - find all the backup checkpoint belong to this policy, mainly obsolete backup checkpoint
static bool filter_checkpoint(const std::string &dir,
                              const cold_backup_context_ptr &backup_context,
                              /*out*/ std::vector<std::string> &related_chkpt_dirs,
                              /*out*/ std::string &valid_chkpt_dir)
{
    valid_chkpt_dir.clear();
    related_chkpt_dirs.clear();
    // list sub dirs
    std::vector<std::string> sub_dirs;
    if (!utils::filesystem::get_subdirectories(dir, sub_dirs, false)) {
        LOG_ERROR("{}: list sub dirs of dir {} failed", backup_context->name, dir);
        return false;
    }

    for (std::string &d : sub_dirs) {
        std::string dirname = utils::filesystem::get_file_name(d);
        int ret = is_related_or_valid_checkpoint(dirname, backup_context);
        if (ret == 1) {
            related_chkpt_dirs.emplace_back(std::move(dirname));
        } else if (ret == 2) {
            CHECK(valid_chkpt_dir.empty(),
                  "{}: there are two valid backup checkpoint dir, {} VS {}",
                  backup_context->name,
                  valid_chkpt_dir,
                  dirname);
            valid_chkpt_dir = dirname;
        }
    }
    return true;
}

static bool
statistic_file_infos_under_dir(const std::string &dir,
                               /*out*/ std::vector<std::pair<std::string, int64_t>> &file_infos,
                               /*out*/ int64_t &total_size)
{
    std::vector<std::string> sub_files;
    if (!utils::filesystem::get_subfiles(dir, sub_files, false)) {
        LOG_ERROR("list sub files of dir {} failed", dir);
        return false;
    }

    total_size = 0;
    file_infos.clear();

    // TODO(yingchun): check if there are any files that are not sensitive (not encrypted).
    for (std::string &file : sub_files) {
        std::pair<std::string, int64_t> file_info;

        if (!utils::filesystem::file_size(
                file, dsn::utils::FileDataType::kSensitive, file_info.second)) {
            LOG_ERROR("get file size of {} failed", file);
            return false;
        }
        file_info.first = utils::filesystem::get_file_name(file);
        total_size += file_info.second;

        file_infos.emplace_back(std::move(file_info));
    }
    return true;
}

static bool backup_parse_dir_name(const char *name,
                                  std::string &policy_name,
                                  int64_t &backup_id,
                                  int64_t &decree,
                                  int64_t &timestamp)
{
    std::vector<std::string> strs;
    ::dsn::utils::split_args(name, strs, '.');
    if (strs.size() < 5) {
        return false;
    } else {
        policy_name = strs[1];
        backup_id = boost::lexical_cast<int64_t>(strs[2]);
        decree = boost::lexical_cast<int64_t>(strs[3]);
        timestamp = boost::lexical_cast<int64_t>(strs[4]);
        return (std::string(name) ==
                backup_get_dir_name(policy_name, backup_id, decree, timestamp));
    }
}

// run in REPLICATION_LONG thread
// Effection:
// - may ignore_checkpoint() if in invalid status
// - may fail_checkpoint() if some error occurs
// - may complete_checkpoint() and schedule on_cold_backup() if backup checkpoint dir is already
// exist
// - may schedule trigger_async_checkpoint_for_backup() if backup checkpoint dir is not exist
void replica::generate_backup_checkpoint(cold_backup_context_ptr backup_context)
{
    if (backup_context->status() != ColdBackupCheckpointing) {
        LOG_INFO("{}: ignore generating backup checkpoint because backup_status = {}",
                 backup_context->name,
                 cold_backup_status_to_string(backup_context->status()));
        backup_context->ignore_checkpoint();
        return;
    }

    // prepare back dir
    auto backup_dir = _app->backup_dir();
    if (!utils::filesystem::directory_exists(backup_dir) &&
        !utils::filesystem::create_directory(backup_dir)) {
        LOG_ERROR("{}: create backup dir {} failed", backup_context->name, backup_dir);
        backup_context->fail_checkpoint("create backup dir failed");
        return;
    }

    std::vector<std::string> related_backup_chkpt_dirname;
    std::string valid_backup_chkpt_dirname;
    if (!filter_checkpoint(
            backup_dir, backup_context, related_backup_chkpt_dirname, valid_backup_chkpt_dirname)) {
        // encounter some error, just return
        backup_context->fail_checkpoint("list sub backup dir failed");
        return;
    }
    if (!valid_backup_chkpt_dirname.empty()) {
        std::vector<std::pair<std::string, int64_t>> file_infos;
        int64_t total_size = 0;
        std::string valid_chkpt_full_path =
            utils::filesystem::path_combine(backup_dir, valid_backup_chkpt_dirname);
        // parse checkpoint dirname
        std::string policy_name;
        int64_t backup_id = 0, decree = 0, timestamp = 0;
        CHECK(backup_parse_dir_name(
                  valid_backup_chkpt_dirname.c_str(), policy_name, backup_id, decree, timestamp),
              "{}: valid chekpoint dirname {}",
              backup_context->name,
              valid_backup_chkpt_dirname);

        if (statistic_file_infos_under_dir(valid_chkpt_full_path, file_infos, total_size)) {
            backup_context->checkpoint_decree = decree;
            backup_context->checkpoint_timestamp = timestamp;
            backup_context->checkpoint_dir = valid_chkpt_full_path;
            for (std::pair<std::string, int64_t> &p : file_infos) {
                backup_context->checkpoint_files.emplace_back(std::move(p.first));
                backup_context->checkpoint_file_sizes.emplace_back(std::move(p.second));
            }
            backup_context->checkpoint_file_total_size = total_size;
            backup_context->complete_checkpoint();

            LOG_INFO(
                "{}: backup checkpoint aleady exist, dir = {}, file_count = {}, total_size = {}",
                backup_context->name,
                backup_context->checkpoint_dir,
                file_infos.size(),
                total_size);
            // TODO: in primary, this will make the request send to secondary again
            tasking::enqueue(
                LPC_REPLICATION_COLD_BACKUP,
                &_tracker,
                [this, backup_context]() {
                    backup_response response;
                    on_cold_backup(backup_context->request, response);
                },
                get_gpid().thread_hash());
        } else {
            backup_context->fail_checkpoint("statistic file info under checkpoint failed");
            return;
        }
    } else {
        LOG_INFO("{}: backup checkpoint not exist, start to trigger async checkpoint",
                 backup_context->name);
        tasking::enqueue(
            LPC_REPLICATION_COLD_BACKUP,
            &_tracker,
            [this, backup_context]() { trigger_async_checkpoint_for_backup(backup_context); },
            get_gpid().thread_hash());
    }

    // clear related but not valid checkpoint
    for (const std::string &dirname : related_backup_chkpt_dirname) {
        std::string full_path = utils::filesystem::path_combine(backup_dir, dirname);
        LOG_INFO("{}: found obsolete backup checkpoint dir({}), remove it",
                 backup_context->name,
                 full_path);
        if (!utils::filesystem::remove_path(full_path)) {
            LOG_WARNING("{}: remove obsolete backup checkpoint dir({}) failed",
                        backup_context->name,
                        full_path);
        }
    }
}

// run in REPLICATION thread
// Effection:
// - may ignore_checkpoint() if in invalid status
// - may fail_checkpoint() if some error occurs
// - may trigger async checkpoint and invoke wait_async_checkpoint_for_backup()
void replica::trigger_async_checkpoint_for_backup(cold_backup_context_ptr backup_context)
{
    _checker.only_one_thread_access();

    if (backup_context->status() != ColdBackupCheckpointing) {
        LOG_INFO("{}: ignore triggering async checkpoint because backup_status = {}",
                 backup_context->name,
                 cold_backup_status_to_string(backup_context->status()));
        backup_context->ignore_checkpoint();
        return;
    }

    if (status() != partition_status::PS_PRIMARY && status() != partition_status::PS_SECONDARY) {
        LOG_INFO("{}: ignore triggering async checkpoint because partition_status = {}",
                 backup_context->name,
                 enum_to_string(status()));
        backup_context->ignore_checkpoint();
        return;
    }

    decree durable_decree = last_durable_decree();
    if (backup_context->checkpoint_decree > 0 &&
        durable_decree >= backup_context->checkpoint_decree) {
        // checkpoint done
    } else if (backup_context->checkpoint_decree > 0 &&
               backup_context->durable_decree_when_checkpoint == durable_decree) {
        // already triggered, just wait
        char time_buf[20];
        dsn::utils::time_ms_to_date_time(backup_context->checkpoint_timestamp, time_buf, 20);
        LOG_INFO("{}: do not trigger async checkpoint because it is already triggered, "
                 "checkpoint_decree = {}, checkpoint_timestamp = {} ({}), "
                 "durable_decree_when_checkpoint = {}",
                 backup_context->name,
                 backup_context->checkpoint_decree,
                 backup_context->checkpoint_timestamp,
                 time_buf,
                 backup_context->durable_decree_when_checkpoint);
    } else { // backup_context->checkpoint_decree == 0 ||
             // backup_context->durable_decree_when_checkpoint != durable_decree
        if (backup_context->checkpoint_decree == 0) {
            // first trigger
            backup_context->checkpoint_decree = last_committed_decree();
        } else { // backup_context->durable_decree_when_checkpoint != durable_decree
            // checkpoint generated, but is behind checkpoint_decree, need trigger again
            CHECK_LT(backup_context->durable_decree_when_checkpoint, durable_decree);
            LOG_INFO("{}: need trigger async checkpoint again", backup_context->name);
        }
        backup_context->checkpoint_timestamp = dsn_now_ms();
        backup_context->durable_decree_when_checkpoint = durable_decree;
        char time_buf[20];
        dsn::utils::time_ms_to_date_time(backup_context->checkpoint_timestamp, time_buf, 20);
        LOG_INFO("{}: trigger async checkpoint, "
                 "checkpoint_decree = {}, checkpoint_timestamp = {} ({}), "
                 "durable_decree_when_checkpoint = {}",
                 backup_context->name,
                 backup_context->checkpoint_decree,
                 backup_context->checkpoint_timestamp,
                 time_buf,
                 backup_context->durable_decree_when_checkpoint);
        init_checkpoint(true);
    }

    // after triggering init_checkpoint, we just wait until it finish
    wait_async_checkpoint_for_backup(backup_context);
}

// run in REPLICATION thread
// Effection:
// - may ignore_checkpoint() if in invalid status
// - may delay some time and schedule trigger_async_checkpoint_for_backup() if async checkpoint not
// completed
// - may schedule local_create_backup_checkpoint if async checkpoint completed
void replica::wait_async_checkpoint_for_backup(cold_backup_context_ptr backup_context)
{
    _checker.only_one_thread_access();

    if (backup_context->status() != ColdBackupCheckpointing) {
        LOG_INFO("{}: ignore waiting async checkpoint because backup_status = {}",
                 backup_context->name,
                 cold_backup_status_to_string(backup_context->status()));
        backup_context->ignore_checkpoint();
        return;
    }

    if (status() != partition_status::PS_PRIMARY && status() != partition_status::PS_SECONDARY) {
        LOG_INFO("{}: ignore waiting async checkpoint because partition_status = {}",
                 backup_context->name,
                 enum_to_string(status()));
        backup_context->ignore_checkpoint();
        return;
    }

    decree du = last_durable_decree();
    if (du < backup_context->checkpoint_decree) {
        LOG_INFO("{}: async checkpoint not done, we just wait it done, "
                 "last_durable_decree = {}, backup_checkpoint_decree = {}",
                 backup_context->name,
                 du,
                 backup_context->checkpoint_decree);
        tasking::enqueue(
            LPC_REPLICATION_COLD_BACKUP,
            &_tracker,
            [this, backup_context]() { trigger_async_checkpoint_for_backup(backup_context); },
            get_gpid().thread_hash(),
            std::chrono::seconds(10));
    } else {
        LOG_INFO("{}: async checkpoint done, last_durable_decree = {}, "
                 "backup_context->checkpoint_decree = {}",
                 backup_context->name,
                 du,
                 backup_context->checkpoint_decree);
        tasking::enqueue(LPC_BACKGROUND_COLD_BACKUP, &_tracker, [this, backup_context]() {
            local_create_backup_checkpoint(backup_context);
        });
    }
}

// run in REPLICATION_LONG thread
// Effection:
// - may ignore_checkpoint() if in invalid status
// - may fail_checkpoint() if some error occurs
// - may complete_checkpoint() and schedule on_cold_backup() if checkpoint dir is successfully
// copied
void replica::local_create_backup_checkpoint(cold_backup_context_ptr backup_context)
{
    if (backup_context->status() != ColdBackupCheckpointing) {
        LOG_INFO("{}: ignore generating backup checkpoint because backup_status = {}",
                 backup_context->name,
                 cold_backup_status_to_string(backup_context->status()));
        backup_context->ignore_checkpoint();
        return;
    }

    // the real checkpoint decree may be larger than backup_context->checkpoint_decree,
    // so we need copy checkpoint to backup_checkpoint_tmp_dir_path, and then rename it.
    std::string backup_checkpoint_tmp_dir_path = utils::filesystem::path_combine(
        _app->backup_dir(),
        backup_get_tmp_dir_name(backup_context->request.policy.policy_name,
                                backup_context->request.backup_id,
                                backup_context->checkpoint_timestamp));
    int64_t last_decree = 0;
    dsn::error_code err =
        _app->copy_checkpoint_to_dir(backup_checkpoint_tmp_dir_path.c_str(), &last_decree);
    if (err != ERR_OK) {
        // try local_create_backup_checkpoint 10s later
        LOG_INFO("{}: create backup checkpoint failed with err = {}, try call "
                 "local_create_backup_checkpoint 10s later",
                 backup_context->name,
                 err);
        utils::filesystem::remove_path(backup_checkpoint_tmp_dir_path);
        tasking::enqueue(
            LPC_BACKGROUND_COLD_BACKUP,
            &_tracker,
            [this, backup_context]() { local_create_backup_checkpoint(backup_context); },
            0,
            std::chrono::seconds(10));
    } else {
        CHECK_GE(last_decree, backup_context->checkpoint_decree);
        backup_context->checkpoint_decree = last_decree; // update to real decree
        std::string backup_checkpoint_dir_path = utils::filesystem::path_combine(
            _app->backup_dir(),
            backup_get_dir_name(backup_context->request.policy.policy_name,
                                backup_context->request.backup_id,
                                backup_context->checkpoint_decree,
                                backup_context->checkpoint_timestamp));
        if (!utils::filesystem::rename_path(backup_checkpoint_tmp_dir_path,
                                            backup_checkpoint_dir_path)) {
            LOG_ERROR("{}: rename checkpoint dir({}) to dir({}) failed",
                      backup_context->name,
                      backup_checkpoint_tmp_dir_path,
                      backup_checkpoint_dir_path);
            utils::filesystem::remove_path(backup_checkpoint_tmp_dir_path);
            utils::filesystem::remove_path(backup_checkpoint_dir_path);
            backup_context->fail_checkpoint("rename checkpoint dir failed");
            return;
        }

        std::vector<std::pair<std::string, int64_t>> file_infos;
        int64_t total_size = 0;
        if (!statistic_file_infos_under_dir(backup_checkpoint_dir_path, file_infos, total_size)) {
            LOG_ERROR("{}: statistic file info under dir({}) failed",
                      backup_context->name,
                      backup_checkpoint_dir_path);
            backup_context->fail_checkpoint("statistic file info under dir failed");
            return;
        }

        LOG_INFO(
            "{}: generate backup checkpoint succeed, dir = {}, file_count = {}, total_size = {}",
            backup_context->name,
            backup_checkpoint_dir_path,
            file_infos.size(),
            total_size);
        backup_context->checkpoint_dir = backup_checkpoint_dir_path;
        for (std::pair<std::string, int64_t> &pair : file_infos) {
            backup_context->checkpoint_files.emplace_back(std::move(pair.first));
            backup_context->checkpoint_file_sizes.emplace_back(std::move(pair.second));
        }
        backup_context->checkpoint_file_total_size = total_size;
        backup_context->complete_checkpoint();
        tasking::enqueue(
            LPC_REPLICATION_COLD_BACKUP,
            &_tracker,
            [this, backup_context]() {
                backup_response response;
                on_cold_backup(backup_context->request, response);
            },
            get_gpid().thread_hash());
    }
}

void replica::set_backup_context_cancel()
{
    for (auto &pair : _cold_backup_contexts) {
        pair.second->cancel();
        LOG_INFO_PREFIX("cancel backup progress, backup_request = {}",
                        boost::lexical_cast<std::string>(pair.second->request));
    }
}

void replica::clear_cold_backup_state() { _cold_backup_contexts.clear(); }
} // namespace replication
} // namespace dsn