replica_2pc.cpp

/*
 * The MIT License (MIT)
 *
 * Copyright (c) 2015 Microsoft Corporation
 *
 * -=- Robust Distributed System Nucleus (rDSN) -=-
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
 * THE SOFTWARE.
 */

#include "replica.h"
#include "mutation.h"
#include "mutation_log.h"
#include "replica_stub.h"
#include "bulk_load/replica_bulk_loader.h"
#include "split/replica_split_manager.h"
#include "runtime/security/access_controller.h"
#include <dsn/utils/latency_tracer.h>
#include <dsn/dist/replication/replication_app_base.h>
#include <dsn/dist/fmt_logging.h>

namespace dsn {
namespace replication {

DSN_DEFINE_bool("replication",
                reject_write_when_disk_insufficient,
                true,
                "reject client write requests if disk status is space insufficient");
DSN_TAG_VARIABLE(reject_write_when_disk_insufficient, FT_MUTABLE);

void replica::on_client_write(dsn::message_ex *request, bool ignore_throttling)
{
    _checker.only_one_thread_access();

    if (!_access_controller->allowed(request)) {
        response_client_write(request, ERR_ACL_DENY);
        return;
    }

    if (_deny_client.write) {
        if (_deny_client.reconfig) {
            // return ERR_INVALID_STATE will trigger client update config immediately
            response_client_write(request, ERR_INVALID_STATE);
            return;
        }
        // Do not reply any message to the peer client to let it timeout, it's OK coz some users
        // may retry immediately when they got a not success code which will make the server side
        // pressure more and more heavy.
        return;
    }

    if (dsn_unlikely(_stub->_max_allowed_write_size &&
                     request->body_size() > _stub->_max_allowed_write_size)) {
        std::string request_info = _app->dump_write_request(request);
        dwarn_replica("client from {} write request body size exceed threshold, request = [{}], "
                      "request_body_size "
                      "= {}, max_allowed_write_size = {}, it will be rejected!",
                      request->header->from_address.to_string(),
                      request_info,
                      request->body_size(),
                      _stub->_max_allowed_write_size);
        _stub->_counter_recent_write_size_exceed_threshold_count->increment();
        response_client_write(request, ERR_INVALID_DATA);
        return;
    }

    task_spec *spec = task_spec::get(request->rpc_code());
    if (dsn_unlikely(nullptr == spec || request->rpc_code() == TASK_CODE_INVALID)) {
        derror_f("recv message with unhandled rpc name {} from {}, trace_id = {}",
                 request->rpc_code().to_string(),
                 request->header->from_address.to_string(),
                 request->header->trace_id);
        response_client_write(request, ERR_HANDLER_NOT_FOUND);
        return;
    }

    if (is_duplication_master() && !spec->rpc_request_is_write_idempotent) {
        // Ignore non-idempotent write, because duplication provides no guarantee of atomicity to
        // make this write produce the same result on multiple clusters.
        _counter_dup_disabled_non_idempotent_write_count->increment();
        response_client_write(request, ERR_OPERATION_DISABLED);
        return;
    }

    CHECK_REQUEST_IF_SPLITTING(write)

    if (partition_status::PS_PRIMARY != status()) {
        response_client_write(request, ERR_INVALID_STATE);
        return;
    }

    if (FLAGS_reject_write_when_disk_insufficient &&
        (disk_space_insufficient() || _primary_states.secondary_disk_space_insufficient())) {
        response_client_write(request, ERR_DISK_INSUFFICIENT);
        return;
    }

    if (_is_bulk_load_ingestion) {
        if (request->rpc_code() != dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
            // reject write requests during ingestion
            _counter_recent_write_bulk_load_ingestion_reject_count->increment();
            response_client_write(request, ERR_OPERATION_DISABLED);
        } else {
            response_client_write(request, ERR_NO_NEED_OPERATE);
        }
        return;
    }

    if (request->rpc_code() == dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
        auto cur_bulk_load_status = _bulk_loader->get_bulk_load_status();
        if (cur_bulk_load_status != bulk_load_status::BLS_DOWNLOADED &&
            cur_bulk_load_status != bulk_load_status::BLS_INGESTING) {
            derror_replica("receive bulk load ingestion request with wrong status({})",
                           enum_to_string(cur_bulk_load_status));
            response_client_write(request, ERR_INVALID_STATE);
            return;
        }
        ddebug_replica("receive bulk load ingestion request");

        // bulk load ingestion request requires that all secondaries should be alive
        if (static_cast<int>(_primary_states.membership.secondaries.size()) + 1 <
            _primary_states.membership.max_replica_count) {
            response_client_write(request, ERR_NOT_ENOUGH_MEMBER);
            return;
        }
        _is_bulk_load_ingestion = true;
        _bulk_load_ingestion_start_time_ms = dsn_now_ms();
    }

    if (static_cast<int>(_primary_states.membership.secondaries.size()) + 1 <
        _options->mutation_2pc_min_replica_count) {
        response_client_write(request, ERR_NOT_ENOUGH_MEMBER);
        return;
    }

    if (!ignore_throttling && throttle_write_request(request)) {
        return;
    }

    dinfo("%s: got write request from %s", name(), request->header->from_address.to_string());
    auto mu = _primary_states.write_queue.add_work(request->rpc_code(), request, this);
    if (mu) {
        init_prepare(mu, false);
    }
}

void replica::init_prepare(mutation_ptr &mu, bool reconciliation, bool pop_all_committed_mutations)
{
    dassert(partition_status::PS_PRIMARY == status(),
            "invalid partition_status, status = %s",
            enum_to_string(status()));

    mu->_tracer->set_description("primary");
    ADD_POINT(mu->_tracer);

    error_code err = ERR_OK;
    uint8_t count = 0;
    const auto request_count = mu->client_requests.size();
    mu->data.header.last_committed_decree = last_committed_decree();

    dsn_log_level_t level = LOG_LEVEL_INFORMATION;
    if (mu->data.header.decree == invalid_decree) {
        mu->set_id(get_ballot(), _prepare_list->max_decree() + 1);
        // print a debug log if necessary
        if (_options->prepare_decree_gap_for_debug_logging > 0 &&
            mu->get_decree() % _options->prepare_decree_gap_for_debug_logging == 0)
            level = LOG_LEVEL_DEBUG;
        mu->set_timestamp(_uniq_timestamp_us.next());
    } else {
        mu->set_id(get_ballot(), mu->data.header.decree);
    }

    mu->_tracer->set_name(fmt::format("mutation[{}]", mu->name()));
    dlog(level,
         "%s: mutation %s init_prepare, mutation_tid=%" PRIu64,
         name(),
         mu->name(),
         mu->tid());

    // child should prepare mutation synchronously
    mu->set_is_sync_to_child(_primary_states.sync_send_write_request);

    // check bounded staleness
    if (mu->data.header.decree > last_committed_decree() + _options->staleness_for_commit) {
        err = ERR_CAPACITY_EXCEEDED;
        goto ErrOut;
    }

    // stop prepare bulk load ingestion if there are secondaries unalive
    for (auto i = 0; i < request_count; ++i) {
        const mutation_update &update = mu->data.updates[i];
        if (update.code != dsn::apps::RPC_RRDB_RRDB_BULK_LOAD) {
            break;
        }
        ddebug_replica("try to prepare bulk load mutation({})", mu->name());
        if (static_cast<int>(_primary_states.membership.secondaries.size()) + 1 <
            _primary_states.membership.max_replica_count) {
            err = ERR_NOT_ENOUGH_MEMBER;
            break;
        }
    }
    if (err != ERR_OK) {
        goto ErrOut;
    }

    // stop prepare if there are too few replicas unless it's a reconciliation
    // for reconciliation, we should ensure every prepared mutation to be committed
    // please refer to PacificA paper
    if (static_cast<int>(_primary_states.membership.secondaries.size()) + 1 <
            _options->mutation_2pc_min_replica_count &&
        !reconciliation) {
        err = ERR_NOT_ENOUGH_MEMBER;
        goto ErrOut;
    }

    dassert(mu->data.header.decree > last_committed_decree(),
            "%" PRId64 " VS %" PRId64 "",
            mu->data.header.decree,
            last_committed_decree());

    // local prepare
    err = _prepare_list->prepare(mu, partition_status::PS_PRIMARY, pop_all_committed_mutations);
    if (err != ERR_OK) {
        goto ErrOut;
    }

    // remote prepare
    mu->set_prepare_ts();
    mu->set_left_secondary_ack_count((unsigned int)_primary_states.membership.secondaries.size());
    for (auto it = _primary_states.membership.secondaries.begin();
         it != _primary_states.membership.secondaries.end();
         ++it) {
        send_prepare_message(*it,
                             partition_status::PS_SECONDARY,
                             mu,
                             _options->prepare_timeout_ms_for_secondaries,
                             pop_all_committed_mutations);
    }

    count = 0;
    for (auto it = _primary_states.learners.begin(); it != _primary_states.learners.end(); ++it) {
        if (it->second.prepare_start_decree != invalid_decree &&
            mu->data.header.decree >= it->second.prepare_start_decree) {
            send_prepare_message(it->first,
                                 partition_status::PS_POTENTIAL_SECONDARY,
                                 mu,
                                 _options->prepare_timeout_ms_for_potential_secondaries,
                                 pop_all_committed_mutations,
                                 it->second.signature);
            count++;
        }
    }
    mu->set_left_potential_secondary_ack_count(count);

    if (_split_mgr->is_splitting()) {
        _split_mgr->copy_mutation(mu);
    }

    if (mu->is_logged()) {
        do_possible_commit_on_primary(mu);
    } else {
        dassert(mu->data.header.log_offset == invalid_offset,
                "invalid log offset, offset = %" PRId64,
                mu->data.header.log_offset);
        dassert(mu->log_task() == nullptr, "");
        int64_t pending_size;
        mu->log_task() = _private_log->append(mu,
                                              LPC_WRITE_REPLICATION_LOG,
                                              &_tracker,
                                              std::bind(&replica::on_append_log_completed,
                                                        this,
                                                        mu,
                                                        std::placeholders::_1,
                                                        std::placeholders::_2),
                                              get_gpid().thread_hash(),
                                              &pending_size);
        dassert(nullptr != mu->log_task(), "");
        if (_options->log_shared_pending_size_throttling_threshold_kb > 0 &&
            _options->log_shared_pending_size_throttling_delay_ms > 0 &&
            pending_size >= _options->log_shared_pending_size_throttling_threshold_kb * 1024) {
            int delay_ms = _options->log_shared_pending_size_throttling_delay_ms;
            for (dsn::message_ex *r : mu->client_requests) {
                if (r && r->io_session->delay_recv(delay_ms)) {
                    dwarn("too large pending shared log (%" PRId64 "), "
                          "delay traffic from %s for %d milliseconds",
                          pending_size,
                          r->header->from_address.to_string(),
                          delay_ms);
                }
            }
        }
    }

    _primary_states.last_prepare_ts_ms = mu->prepare_ts_ms();
    return;

ErrOut:
    for (auto &r : mu->client_requests) {
        response_client_write(r, err);
    }
    return;
}

void replica::send_prepare_message(::dsn::rpc_address addr,
                                   partition_status::type status,
                                   const mutation_ptr &mu,
                                   int timeout_milliseconds,
                                   bool pop_all_committed_mutations,
                                   int64_t learn_signature)
{
    mu->_tracer->add_sub_tracer(addr.to_string());
    ADD_POINT(mu->_tracer->sub_tracer(addr.to_string()));

    dsn::message_ex *msg = dsn::message_ex::create_request(
        RPC_PREPARE, timeout_milliseconds, get_gpid().thread_hash());
    replica_configuration rconfig;
    _primary_states.get_replica_config(status, rconfig, learn_signature);
    rconfig.__set_pop_all(pop_all_committed_mutations);
    if (status == partition_status::PS_SECONDARY && _primary_states.sync_send_write_request) {
        rconfig.__set_split_sync_to_child(true);
    }

    {
        rpc_write_stream writer(msg);
        marshall(writer, get_gpid(), DSF_THRIFT_BINARY);
        marshall(writer, rconfig, DSF_THRIFT_BINARY);
        mu->write_to(writer, msg);
    }

    mu->remote_tasks()[addr] =
        rpc::call(addr,
                  msg,
                  &_tracker,
                  [=](error_code err, dsn::message_ex *request, dsn::message_ex *reply) {
                      on_prepare_reply(std::make_pair(mu, rconfig.status), err, request, reply);
                  },
                  get_gpid().thread_hash());

    dinfo("%s: mutation %s send_prepare_message to %s as %s",
          name(),
          mu->name(),
          addr.to_string(),
          enum_to_string(rconfig.status));
}

void replica::do_possible_commit_on_primary(mutation_ptr &mu)
{
    dassert(_config.ballot == mu->data.header.ballot,
            "invalid mutation ballot, %" PRId64 " VS %" PRId64 "",
            _config.ballot,
            mu->data.header.ballot);
    dassert(partition_status::PS_PRIMARY == status(),
            "invalid partition_status, status = %s",
            enum_to_string(status()));

    if (mu->is_ready_for_commit()) {
        _prepare_list->commit(mu->data.header.decree, COMMIT_ALL_READY);
    }
}

void replica::on_prepare(dsn::message_ex *request)
{
    _checker.only_one_thread_access();

    replica_configuration rconfig;
    mutation_ptr mu;
    bool pop_all_committed_mutations = false;

    {
        rpc_read_stream reader(request);
        unmarshall(reader, rconfig, DSF_THRIFT_BINARY);
        mu = mutation::read_from(reader, request);
        mu->set_is_sync_to_child(rconfig.split_sync_to_child);
        pop_all_committed_mutations = rconfig.pop_all;
        rconfig.split_sync_to_child = false;
        rconfig.pop_all = false;
    }

    decree decree = mu->data.header.decree;

    dinfo("%s: mutation %s on_prepare", name(), mu->name());
    mu->_tracer->set_name(fmt::format("mutation[{}]", mu->name()));
    mu->_tracer->set_description("secondary");
    ADD_POINT(mu->_tracer);

    dassert(mu->data.header.pid == rconfig.pid,
            "(%d.%d) VS (%d.%d)",
            mu->data.header.pid.get_app_id(),
            mu->data.header.pid.get_partition_index(),
            rconfig.pid.get_app_id(),
            rconfig.pid.get_partition_index());
    dassert(mu->data.header.ballot == rconfig.ballot,
            "invalid mutation ballot, %" PRId64 " VS %" PRId64 "",
            mu->data.header.ballot,
            rconfig.ballot);

    if (mu->data.header.ballot < get_ballot()) {
        derror("%s: mutation %s on_prepare skipped due to old view", name(), mu->name());
        // no need response because the rpc should have been cancelled on primary in this case
        return;
    }

    // update configuration when necessary
    else if (rconfig.ballot > get_ballot()) {
        if (!update_local_configuration(rconfig)) {
            derror("%s: mutation %s on_prepare failed as update local configuration failed, state "
                   "= %s",
                   name(),
                   mu->name(),
                   enum_to_string(status()));
            ack_prepare_message(ERR_INVALID_STATE, mu);
            return;
        }
    }

    if (partition_status::PS_INACTIVE == status() || partition_status::PS_ERROR == status()) {
        derror("%s: mutation %s on_prepare failed as invalid replica state, state = %s",
               name(),
               mu->name(),
               enum_to_string(status()));
        ack_prepare_message((partition_status::PS_INACTIVE == status() && _inactive_is_transient)
                                ? ERR_INACTIVE_STATE
                                : ERR_INVALID_STATE,
                            mu);
        return;
    } else if (partition_status::PS_POTENTIAL_SECONDARY == status()) {
        // new learning process
        if (rconfig.learner_signature != _potential_secondary_states.learning_version) {
            derror("%s: mutation %s on_prepare failed as unmatched learning signature, state = %s"
                   ", old_signature[%016" PRIx64 "] vs new_signature[%016" PRIx64 "]",
                   name(),
                   mu->name(),
                   enum_to_string(status()),
                   _potential_secondary_states.learning_version,
                   rconfig.learner_signature);
            handle_learning_error(ERR_INVALID_STATE, false);
            ack_prepare_message(ERR_INVALID_STATE, mu);
            return;
        }

        auto learning_status = _potential_secondary_states.learning_status;
        if (learning_status != learner_status::LearningWithPrepare &&
            learning_status != learner_status::LearningSucceeded) {
            // if prepare requests are received when learning status is changing from
            // LearningWithoutPrepare to LearningWithPrepare, we ack ERR_TRY_AGAIN.
            error_code ack_code =
                (learning_status == learner_status::LearningWithoutPrepare ? ERR_TRY_AGAIN
                                                                           : ERR_INVALID_STATE);
            derror("%s: mutation %s on_prepare skipped as invalid learning status, state = %s, "
                   "learning_status = %s, ack %s",
                   name(),
                   mu->name(),
                   enum_to_string(status()),
                   enum_to_string(learning_status),
                   ack_code.to_string());
            ack_prepare_message(ack_code, mu);
            return;
        }
    }

    dassert(rconfig.status == status(),
            "invalid status, %s VS %s",
            enum_to_string(rconfig.status),
            enum_to_string(status()));
    if (decree <= last_committed_decree()) {
        ack_prepare_message(ERR_OK, mu);
        return;
    }

    // real prepare start
    _uniq_timestamp_us.try_update(mu->data.header.timestamp);
    auto mu2 = _prepare_list->get_mutation_by_decree(decree);
    if (mu2 != nullptr && mu2->data.header.ballot == mu->data.header.ballot) {
        if (mu2->is_logged()) {
            // already logged, just response ERR_OK
            ack_prepare_message(ERR_OK, mu);
        } else {
            // not logged, combine duplicate request to old mutation
            mu2->add_prepare_request(request);
        }
        return;
    }

    error_code err = _prepare_list->prepare(mu, status(), pop_all_committed_mutations);
    dassert(err == ERR_OK, "prepare mutation failed, err = %s", err.to_string());

    if (partition_status::PS_POTENTIAL_SECONDARY == status() ||
        partition_status::PS_SECONDARY == status()) {
        dassert(mu->data.header.decree <=
                    last_committed_decree() + _options->max_mutation_count_in_prepare_list,
                "%" PRId64 " VS %" PRId64 "(%" PRId64 " + %d)",
                mu->data.header.decree,
                last_committed_decree() + _options->max_mutation_count_in_prepare_list,
                last_committed_decree(),
                _options->max_mutation_count_in_prepare_list);
    } else {
        derror("%s: mutation %s on_prepare failed as invalid replica state, state = %s",
               name(),
               mu->name(),
               enum_to_string(status()));
        ack_prepare_message(ERR_INVALID_STATE, mu);
        return;
    }

    if (_split_mgr->is_splitting()) {
        _split_mgr->copy_mutation(mu);
    }

    dassert(mu->log_task() == nullptr, "");
    mu->log_task() = _private_log->append(mu,
                                          LPC_WRITE_REPLICATION_LOG,
                                          &_tracker,
                                          std::bind(&replica::on_append_log_completed,
                                                    this,
                                                    mu,
                                                    std::placeholders::_1,
                                                    std::placeholders::_2),
                                          get_gpid().thread_hash());
    dassert(nullptr != mu->log_task(), "");
}

void replica::on_append_log_completed(mutation_ptr &mu, error_code err, size_t size)
{
    _checker.only_one_thread_access();

    dinfo("%s: append shared log completed for mutation %s, size = %u, err = %s",
          name(),
          mu->name(),
          size,
          err.to_string());

    ADD_POINT(mu->_tracer);

    if (err == ERR_OK) {
        mu->set_logged();
    } else {
        derror("%s: append shared log failed for mutation %s, err = %s",
               name(),
               mu->name(),
               err.to_string());
    }

    // skip old mutations
    if (mu->data.header.ballot >= get_ballot() && status() != partition_status::PS_INACTIVE) {
        switch (status()) {
        case partition_status::PS_PRIMARY:
            if (err == ERR_OK) {
                do_possible_commit_on_primary(mu);
            } else {
                handle_local_failure(err);
            }
            break;
        case partition_status::PS_SECONDARY:
        case partition_status::PS_POTENTIAL_SECONDARY:
            if (err != ERR_OK) {
                handle_local_failure(err);
            }
            // always ack
            ack_prepare_message(err, mu);
            // all mutations with lower decree must be ready
            _prepare_list->commit(mu->data.header.last_committed_decree, COMMIT_TO_DECREE_HARD);
            break;
        case partition_status::PS_PARTITION_SPLIT:
            if (err != ERR_OK) {
                handle_local_failure(err);
            }
            _split_mgr->ack_parent(err, mu);
            break;
        case partition_status::PS_ERROR:
            break;
        default:
            dassert(false, "invalid partition_status, status = %s", enum_to_string(status()));
            break;
        }
    }

    if (err != ERR_OK) {
        // mutation log failure, propagate to all replicas
        _stub->handle_log_failure(err);
    }
}

void replica::on_prepare_reply(std::pair<mutation_ptr, partition_status::type> pr,
                               error_code err,
                               dsn::message_ex *request,
                               dsn::message_ex *reply)
{
    _checker.only_one_thread_access();

    mutation_ptr mu = pr.first;
    partition_status::type target_status = pr.second;

    // skip callback for old mutations
    if (partition_status::PS_PRIMARY != status() || mu->data.header.ballot < get_ballot() ||
        mu->get_decree() <= last_committed_decree())
        return;

    dassert(mu->data.header.ballot == get_ballot(),
            "%s: invalid mutation ballot, %" PRId64 " VS %" PRId64 "",
            mu->name(),
            mu->data.header.ballot,
            get_ballot());

    ::dsn::rpc_address node = request->to_address;
    partition_status::type st = _primary_states.get_node_status(node);

    // handle reply
    prepare_ack resp;

    // handle error
    if (err != ERR_OK) {
        resp.err = err;
    } else {
        ::dsn::unmarshall(reply, resp);
    }

    auto send_prepare_tracer = mu->_tracer->sub_tracer(request->to_address.to_string());
    APPEND_EXTERN_POINT(send_prepare_tracer, resp.receive_timestamp, "remote_receive");
    APPEND_EXTERN_POINT(send_prepare_tracer, resp.response_timestamp, "remote_reply");
    ADD_CUSTOM_POINT(send_prepare_tracer, resp.err.to_string());

    if (resp.err == ERR_OK) {
        dinfo("%s: mutation %s on_prepare_reply from %s, appro_data_bytes = %d, "
              "target_status = %s, err = %s",
              name(),
              mu->name(),
              node.to_string(),
              mu->appro_data_bytes(),
              enum_to_string(target_status),
              resp.err.to_string());
    } else {
        derror("%s: mutation %s on_prepare_reply from %s, appro_data_bytes = %d, "
               "target_status = %s, err = %s",
               name(),
               mu->name(),
               node.to_string(),
               mu->appro_data_bytes(),
               enum_to_string(target_status),
               resp.err.to_string());
    }

    if (resp.err == ERR_OK) {
        dassert(resp.ballot == get_ballot(),
                "invalid response ballot, %" PRId64 " VS %" PRId64 "",
                resp.ballot,
                get_ballot());
        dassert(resp.decree == mu->data.header.decree,
                "invalid response decree, %" PRId64 " VS %" PRId64 "",
                resp.decree,
                mu->data.header.decree);

        switch (target_status) {
        case partition_status::PS_SECONDARY:
            dassert(_primary_states.check_exist(node, partition_status::PS_SECONDARY),
                    "invalid secondary node address, address = %s",
                    node.to_string());
            dassert(mu->left_secondary_ack_count() > 0, "%u", mu->left_secondary_ack_count());
            if (0 == mu->decrease_left_secondary_ack_count()) {
                do_possible_commit_on_primary(mu);
            }
            break;
        case partition_status::PS_POTENTIAL_SECONDARY:
            dassert(mu->left_potential_secondary_ack_count() > 0,
                    "%u",
                    mu->left_potential_secondary_ack_count());
            if (0 == mu->decrease_left_potential_secondary_ack_count()) {
                do_possible_commit_on_primary(mu);
            }
            break;
        default:
            dwarn("%s: mutation %s prepare ack skipped coz the node is now inactive",
                  name(),
                  mu->name());
            break;
        }
    }

    // failure handling
    else {
        // retry for INACTIVE or TRY_AGAIN if there is still time.
        if (resp.err == ERR_INACTIVE_STATE || resp.err == ERR_TRY_AGAIN) {
            int prepare_timeout_ms = (target_status == partition_status::PS_SECONDARY
                                          ? _options->prepare_timeout_ms_for_secondaries
                                          : _options->prepare_timeout_ms_for_potential_secondaries);
            int delay_time_ms = 5; // delay some time before retry to avoid sending too frequently
            if (mu->is_prepare_close_to_timeout(delay_time_ms + 2, prepare_timeout_ms)) {
                derror("%s: mutation %s do not retry prepare to %s for no enought time left, "
                       "prepare_ts_ms = %" PRIu64 ", prepare_timeout_ms = %d, now_ms = %" PRIu64,
                       name(),
                       mu->name(),
                       node.to_string(),
                       mu->prepare_ts_ms(),
                       prepare_timeout_ms,
                       dsn_now_ms());
            } else {
                ddebug("%s: mutation %s retry prepare to %s after %d ms",
                       name(),
                       mu->name(),
                       node.to_string(),
                       delay_time_ms);
                int64_t learn_signature = invalid_signature;
                if (target_status == partition_status::PS_POTENTIAL_SECONDARY) {
                    auto it = _primary_states.learners.find(node);
                    if (it != _primary_states.learners.end()) {
                        learn_signature = it->second.signature;
                    }
                }
                tasking::enqueue(
                    LPC_DELAY_PREPARE,
                    &_tracker,
                    [this, node, target_status, mu, prepare_timeout_ms, learn_signature] {
                        // need to check status/ballot/decree before sending prepare message,
                        // because the config may have been changed or the mutation may have been
                        // committed during the delay time.
                        if (status() == partition_status::PS_PRIMARY &&
                            get_ballot() == mu->data.header.ballot &&
                            mu->get_decree() > last_committed_decree()) {
                            send_prepare_message(node,
                                                 target_status,
                                                 mu,
                                                 prepare_timeout_ms,
                                                 false,
                                                 learn_signature);
                        }
                    },
                    get_gpid().thread_hash(),
                    std::chrono::milliseconds(delay_time_ms));
                return;
            }
        }

        _stub->_counter_replicas_recent_prepare_fail_count->increment();

        // make sure this is before any later commit ops
        // because now commit ops may lead to new prepare ops
        // due to replication throttling
        handle_remote_failure(st, node, resp.err, "prepare");

        // note targetStatus and (curent) status may diff
        if (target_status == partition_status::PS_POTENTIAL_SECONDARY) {
            dassert(mu->left_potential_secondary_ack_count() > 0,
                    "%u",
                    mu->left_potential_secondary_ack_count());
            if (0 == mu->decrease_left_potential_secondary_ack_count()) {
                do_possible_commit_on_primary(mu);
            }
        }
    }
}

void replica::ack_prepare_message(error_code err, mutation_ptr &mu)
{
    ADD_POINT(mu->_tracer);
    prepare_ack resp;
    resp.pid = get_gpid();
    resp.err = err;
    resp.ballot = get_ballot();
    resp.decree = mu->data.header.decree;

    resp.__set_receive_timestamp(mu->_tracer->start_time());
    resp.__set_response_timestamp(dsn_now_ns());

    // for partition_status::PS_POTENTIAL_SECONDARY ONLY
    resp.last_committed_decree_in_app = _app->last_committed_decree();
    resp.last_committed_decree_in_prepare_list = last_committed_decree();

    const std::vector<dsn::message_ex *> &prepare_requests = mu->prepare_requests();
    dassert(!prepare_requests.empty(), "mutation = %s", mu->name());

    if (err == ERR_OK) {
        if (mu->is_child_acked()) {
            dinfo_replica("mutation {} ack_prepare_message, err = {}", mu->name(), err);
            for (auto &request : prepare_requests) {
                reply(request, resp);
            }
        }
        return;
    }
    // only happened when prepare failed during partition split child copy mutation synchronously
    if (mu->is_error_acked()) {
        dwarn_replica("mutation {} has been ack_prepare_message, err = {}", mu->name(), err);
        return;
    }

    dwarn_replica("mutation {} ack_prepare_message, err = {}", mu->name(), err);
    if (mu->is_sync_to_child()) {
        mu->set_error_acked();
    }
    for (auto &request : prepare_requests) {
        reply(request, resp);
    }
}

void replica::cleanup_preparing_mutations(bool wait)
{
    decree start = last_committed_decree() + 1;
    decree end = _prepare_list->max_decree();

    for (decree decree = start; decree <= end; decree++) {
        mutation_ptr mu = _prepare_list->get_mutation_by_decree(decree);
        if (mu != nullptr) {
            mu->clear_prepare_or_commit_tasks();

            //
            // make sure the buffers from mutations are valid for underlying aio
            //
            if (wait) {
                if (dsn_unlikely(_private_log != nullptr)) {
                    _private_log->flush();
                }
                mu->wait_log_task();
            }
        }
    }
}
} // namespace replication
} // namespace dsn