/Users/deen/code/yugabyte-db/src/yb/tablet/transaction_participant.cc

Source (jump to first uncovered line)
//
// Copyright (c) YugaByte, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except
// in compliance with the License.  You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied.  See the License for the specific language governing permissions and limitations
// under the License.
//
//

#include "yb/tablet/transaction_participant.h"

#include <queue>

#include <boost/multi_index/hashed_index.hpp>
#include <boost/multi_index/mem_fun.hpp>
#include <boost/multi_index/ordered_index.hpp>

#include "yb/client/transaction_rpc.h"

#include "yb/common/pgsql_error.h"
#include "yb/common/transaction_error.h"

#include "yb/consensus/consensus_util.h"

#include "yb/docdb/docdb_rocksdb_util.h"
#include "yb/docdb/transaction_dump.h"

#include "yb/rpc/poller.h"

#include "yb/server/clock.h"

#include "yb/tablet/cleanup_aborts_task.h"
#include "yb/tablet/cleanup_intents_task.h"
#include "yb/tablet/operations/update_txn_operation.h"
#include "yb/tablet/remove_intents_task.h"
#include "yb/tablet/running_transaction.h"
#include "yb/tablet/running_transaction_context.h"
#include "yb/tablet/transaction_loader.h"
#include "yb/tablet/transaction_participant_context.h"
#include "yb/tablet/transaction_status_resolver.h"

#include "yb/tserver/tserver_service.pb.h"

#include "yb/util/countdown_latch.h"
#include "yb/util/debug-util.h"
#include "yb/util/flag_tags.h"
#include "yb/util/format.h"
#include "yb/util/logging.h"
#include "yb/util/lru_cache.h"
#include "yb/util/metrics.h"
#include "yb/util/operation_counter.h"
#include "yb/util/scope_exit.h"
#include "yb/util/status_format.h"
#include "yb/util/status_log.h"
#include "yb/util/tsan_util.h"

using namespace std::literals;
using namespace std::placeholders;

DEFINE_uint64(transaction_min_running_check_delay_ms, 50,
              "When transaction with minimal start hybrid time is updated at transaction "
              "participant, we wait at least this number of milliseconds before checking its "
              "status at transaction coordinator. Used for the optimization that deletes "
              "provisional records RocksDB SSTable files.");

DEFINE_uint64(transaction_min_running_check_interval_ms, 250,
              "While transaction with minimal start hybrid time remains the same, we will try "
              "to check its status at transaction coordinator at regular intervals this "
              "long (ms). Used for the optimization that deletes "
              "provisional records RocksDB SSTable files.");

DEFINE_test_flag(double, transaction_ignore_applying_probability, 0,
                 "Probability to ignore APPLYING update in tests.");
DEFINE_test_flag(bool, fail_in_apply_if_no_metadata, false,
                 "Fail when applying intents if metadata is not found.");

DEFINE_int32(max_transactions_in_status_request, 128,
             "Request status for at most specified number of transactions at once. "
                 "0 disables load time transaction status resolution.");

DEFINE_uint64(transactions_cleanup_cache_size, 256, "Transactions cleanup cache size.");

DEFINE_uint64(transactions_status_poll_interval_ms, 500 * yb::kTimeMultiplier,
              "Transactions poll interval.");

DEFINE_bool(transactions_poll_check_aborted, true, "Check aborted transactions during poll.");

DECLARE_int64(transaction_abort_check_timeout_ms);

METRIC_DEFINE_simple_counter(
    tablet, transaction_not_found, "Total number of missing transactions during load",
    yb::MetricUnit::kTransactions);
METRIC_DEFINE_simple_gauge_uint64(
    tablet, transactions_running, "Total number of transactions running in participant",
    yb::MetricUnit::kTransactions);

DEFINE_test_flag(int32, txn_participant_inject_latency_on_apply_update_txn_ms, 0,
                 "How much latency to inject when a update txn operation is applied.");

namespace yb {
namespace tablet {

namespace {

YB_STRONGLY_TYPED_BOOL(PostApplyCleanup);

} // namespace

std::string TransactionApplyData::ToString() const {
  return YB_STRUCT_TO_STRING(
      leader_term, transaction_id, op_id, commit_ht, log_ht, sealed, status_tablet, apply_state);
}

class TransactionParticipant::Impl
    : public RunningTransactionContext, public TransactionLoaderContext {
 public:
  Impl(TransactionParticipantContext* context, TransactionIntentApplier* applier,
       const scoped_refptr<MetricEntity>& entity)
      : RunningTransactionContext(context, applier),
        log_prefix_(context->LogPrefix()),
        loader_(this, entity),
        poller_(log_prefix_, std::bind(&Impl::Poll, this)) {
    LOG_WITH_PREFIX(INFO) << "Create";
    metric_transactions_running_ = METRIC_transactions_running.Instantiate(entity, 0);
    metric_transaction_not_found_ = METRIC_transaction_not_found.Instantiate(entity);
  }

  ~Impl() {
    if (StartShutdown()) {
      CompleteShutdown();
    } else {
      LOG_IF_WITH_PREFIX(DFATAL, !shutdown_done_.load(std::memory_order_acquire))
          << "Destroying transaction participant that did not complete shutdown";
    }
  }

  bool StartShutdown() {
    bool expected = false;
    if (!closing_.compare_exchange_strong(expected, true)) {
      return false;
    }

    poller_.Shutdown();

    if (start_latch_.count()) {
      start_latch_.CountDown();
    }

    LOG_WITH_PREFIX(INFO) << "Shutdown";
    return true;
  }

  void CompleteShutdown() {
    LOG_IF_WITH_PREFIX(DFATAL, !closing_.load()) << __func__ << " w/o StartShutdown";

    decltype(status_resolvers_) status_resolvers;
    {
      MinRunningNotifier min_running_notifier(nullptr /* applier */);
      std::lock_guard<std::mutex> lock(mutex_);
      transactions_.clear();
      TransactionsModifiedUnlocked(&min_running_notifier);
      status_resolvers.swap(status_resolvers_);
    }

    rpcs_.Shutdown();
    loader_.Shutdown();
    for (auto& resolver : status_resolvers) {
      resolver.Shutdown();
    }
    shutdown_done_.store(true, std::memory_order_release);
  }

  bool Closing() const override {
    return closing_.load(std::memory_order_acquire);
  }

  void Start() {
    LOG_WITH_PREFIX(INFO) << "Start";
    start_latch_.CountDown();
  }

  // Adds new running transaction.
  Result<bool> Add(const TransactionMetadata& metadata) {
    loader_.WaitLoaded(metadata.transaction_id);

    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(metadata.transaction_id);
    if (it != transactions_.end()) {
      return false;
    }
    if (WasTransactionRecentlyRemoved(metadata.transaction_id) ||
        cleanup_cache_.Erase(metadata.transaction_id) != 0) {
      auto status = STATUS_EC_FORMAT(
          TryAgain, PgsqlError(YBPgErrorCode::YB_PG_T_R_SERIALIZATION_FAILURE),
          "Transaction was recently aborted: $0", metadata.transaction_id);
      return status.CloneAndAddErrorCode(TransactionError(TransactionErrorCode::kAborted));
    }
    VLOG_WITH_PREFIX(4) << "Create new transaction: " << metadata.transaction_id;
    transactions_.insert(std::make_shared<RunningTransaction>(
        metadata, TransactionalBatchData(), OneWayBitmap(), metadata.start_time, this));
    TransactionsModifiedUnlocked(&min_running_notifier);
    return true;
  }

  HybridTime LocalCommitTime(const TransactionId& id) {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(id);
    if (it == transactions_.end()) {
      return HybridTime::kInvalid;
    }
    return (**it).local_commit_time();
  }

  boost::optional<CommitMetadata> LocalCommitData(const TransactionId& id) {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(id);
    if (it == transactions_.end()) {
      return boost::none;
    }
    return boost::make_optional<CommitMetadata>({
      .commit_ht = (**it).local_commit_time(),
      .aborted_subtxn_set = (**it).local_commit_aborted_subtxn_set(),
    });
  }

  std::pair<size_t, size_t> TEST_CountIntents() {
    {
      MinRunningNotifier min_running_notifier(&applier_);
      std::lock_guard<std::mutex> lock(mutex_);
      ProcessRemoveQueueUnlocked(&min_running_notifier);
    }

    std::pair<size_t, size_t> result(0, 0);
    auto iter = docdb::CreateRocksDBIterator(db_.intents,
                                             key_bounds_,
                                             docdb::BloomFilterMode::DONT_USE_BLOOM_FILTER,
                                             boost::none,
                                             rocksdb::kDefaultQueryId);
    for (iter.SeekToFirst(); iter.Valid(); iter.Next()) {
      ++result.first;
      // Count number of transaction, by counting metadata records.
      if (iter.key().size() == TransactionId::StaticSize() + 1) {
        ++result.second;
        auto key = iter.key();
        key.remove_prefix(1);
        auto id = CHECK_RESULT(FullyDecodeTransactionId(key));
        LOG_WITH_PREFIX(INFO) << "Stored txn meta: " << id;
      }
    }

    return result;
  }

  Result<TransactionMetadata> PrepareMetadata(const TransactionMetadataPB& pb) {
    if (pb.has_isolation()) {
      auto metadata = VERIFY_RESULT(TransactionMetadata::FromPB(pb));
      std::unique_lock<std::mutex> lock(mutex_);
      auto it = transactions_.find(metadata.transaction_id);
      if (it != transactions_.end()) {
        RETURN_NOT_OK((**it).CheckAborted());
      } else if (WasTransactionRecentlyRemoved(metadata.transaction_id)) {
        return MakeAbortedStatus(metadata.transaction_id);
      }
      return metadata;
    }

    auto id = VERIFY_RESULT(FullyDecodeTransactionId(pb.transaction_id()));

    // We are not trying to cleanup intents here because we don't know whether this transaction
    // has intents or not.
    auto lock_and_iterator = LockAndFind(
        id, "metadata"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
    if (!lock_and_iterator.found()) {
      return STATUS(TryAgain,
                    Format("Unknown transaction, could be recently aborted: $0", id), Slice(),
                    PgsqlError(YBPgErrorCode::YB_PG_T_R_SERIALIZATION_FAILURE));
    }
    RETURN_NOT_OK(lock_and_iterator.transaction().CheckAborted());
    return lock_and_iterator.transaction().metadata();
  }

  boost::optional<std::pair<IsolationLevel, TransactionalBatchData>> PrepareBatchData(
      const TransactionId& id, size_t batch_idx,
      boost::container::small_vector_base<uint8_t>* encoded_replicated_batches) {
    // We are not trying to cleanup intents here because we don't know whether this transaction
    // has intents of not.
    auto lock_and_iterator = LockAndFind(
        id, "metadata with write id"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
    if (!lock_and_iterator.found()) {
      return boost::none;
    }
    auto& transaction = lock_and_iterator.transaction();
    transaction.AddReplicatedBatch(batch_idx, encoded_replicated_batches);
    return std::make_pair(transaction.metadata().isolation, transaction.last_batch_data());
  }

  void BatchReplicated(const TransactionId& id, const TransactionalBatchData& data) {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(id);
    if (it == transactions_.end()) {
      LOG_IF_WITH_PREFIX(DFATAL, !WasTransactionRecentlyRemoved(id))
          << "Update last write id for unknown transaction: " << id;
      return;
    }
    (**it).BatchReplicated(data);
  }

  void RequestStatusAt(const StatusRequest& request) {
    auto lock_and_iterator = LockAndFind(*request.id, *request.reason, request.flags);
    if (!lock_and_iterator.found()) {
      request.callback(
          STATUS_FORMAT(NotFound, "Request status of unknown transaction: $0", *request.id));
      return;
    }
    lock_and_iterator.transaction().RequestStatusAt(request, &lock_and_iterator.lock);
  }

  // Registers a request, giving it a newly allocated id and returning this id.
  int64_t RegisterRequest() {
    std::lock_guard<std::mutex> lock(mutex_);
    auto result = NextRequestIdUnlocked();
    running_requests_.push_back(result);
    return result;
  }

  // Unregisters a previously registered request.
  void UnregisterRequest(int64_t request) {
    MinRunningNotifier min_running_notifier(&applier_);
    {
      std::lock_guard<std::mutex> lock(mutex_);
      DCHECK(!running_requests_.empty());
      if (running_requests_.front() != request) {
        complete_requests_.push(request);
        return;
      }
      running_requests_.pop_front();
      while (!complete_requests_.empty() && complete_requests_.top() == running_requests_.front()) {
        complete_requests_.pop();
        running_requests_.pop_front();
      }

      CleanTransactionsUnlocked(&min_running_notifier);
    }
  }

  // Cleans transactions that are requested and now is safe to clean.
  // See RemoveUnlocked for details.
  void CleanTransactionsUnlocked(MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) {
    ProcessRemoveQueueUnlocked(min_running_notifier);

    CleanTransactionsQueue(&immediate_cleanup_queue_, min_running_notifier);
    CleanTransactionsQueue(&graceful_cleanup_queue_, min_running_notifier);
  }

  template <class Queue>
  void CleanTransactionsQueue(
      Queue* queue, MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) {
    int64_t min_request = running_requests_.empty() ? std::numeric_limits<int64_t>::max()
                                                    : running_requests_.front();
    HybridTime safe_time;
    while (!queue->empty()) {
      const auto& front = queue->front();
      if (front.request_id >= min_request) {
        break;
      }
      if (!front.Ready(&participant_context_, &safe_time)) {
        break;
      }
      const auto& id = front.transaction_id;
      RemoveIntentsData checkpoint;
      auto it = transactions_.find(id);

      if (it != transactions_.end() && !(**it).ProcessingApply()) {
        OpId op_id = (**it).GetOpId();
        participant_context_.GetLastCDCedData(&checkpoint);
        VLOG_WITH_PREFIX(2) << "Cleaning tx opid is " << op_id.ToString()
                            << " checkpoint opid is " << checkpoint.op_id.ToString();

        if (checkpoint.op_id < op_id) {
          break;
        }
        (**it).ScheduleRemoveIntents(*it);
        RemoveTransaction(it, front.reason, min_running_notifier);
      }
      VLOG_WITH_PREFIX(2) << "Cleaned from queue: " << id;
      queue->pop_front();
    }
  }

  void Abort(const TransactionId& id, TransactionStatusCallback callback) {
    // We are not trying to cleanup intents here because we don't know whether this transaction
    // has intents of not.
    auto lock_and_iterator = LockAndFind(
        id, "abort"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
    if (!lock_and_iterator.found()) {
      callback(STATUS_FORMAT(NotFound, "Abort of unknown transaction: $0", id));
      return;
    }
    auto client_result = client();
    if (!client_result.ok()) {
      callback(client_result.status());
      return;
    }
    lock_and_iterator.transaction().Abort(
        *client_result, std::move(callback), &lock_and_iterator.lock);
  }

  CHECKED_STATUS CheckAborted(const TransactionId& id) {
    // We are not trying to cleanup intents here because we don't know whether this transaction
    // has intents of not.
    auto lock_and_iterator = LockAndFind(id, "check aborted"s, TransactionLoadFlags{});
    if (!lock_and_iterator.found()) {
      return MakeAbortedStatus(id);
    }
    return lock_and_iterator.transaction().CheckAborted();
  }

  void FillPriorities(
      boost::container::small_vector_base<std::pair<TransactionId, uint64_t>>* inout) {
    // TODO(dtxn) optimize locking
    for (auto& pair : *inout) {
      auto lock_and_iterator = LockAndFind(
          pair.first, "fill priorities"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
      if (!lock_and_iterator.found() || lock_and_iterator.transaction().WasAborted()) {
        pair.second = 0; // Minimal priority for already aborted transactions
      } else {
        pair.second = lock_and_iterator.transaction().metadata().priority;
      }
    }
  }

  void Handle(std::unique_ptr<tablet::UpdateTxnOperation> operation, int64_t term) {
    auto txn_status = operation->request()->status();
    if (txn_status == TransactionStatus::APPLYING) {
      HandleApplying(std::move(operation), term);
      return;
    }

    if (txn_status == TransactionStatus::IMMEDIATE_CLEANUP ||
        txn_status == TransactionStatus::GRACEFUL_CLEANUP) {
      auto cleanup_type = txn_status == TransactionStatus::IMMEDIATE_CLEANUP
          ? CleanupType::kImmediate
          : CleanupType::kGraceful;
      HandleCleanup(std::move(operation), term, cleanup_type);
      return;
    }

    auto error_status = STATUS_FORMAT(
        InvalidArgument, "Unexpected status in transaction participant Handle: $0", *operation);
    LOG_WITH_PREFIX(DFATAL) << error_status;
    operation->CompleteWithStatus(error_status);
  }

  CHECKED_STATUS ProcessReplicated(const ReplicatedData& data) {
    if (FLAGS_TEST_txn_participant_inject_latency_on_apply_update_txn_ms > 0) {
      SleepFor(1ms * FLAGS_TEST_txn_participant_inject_latency_on_apply_update_txn_ms);
    }

    auto id = FullyDecodeTransactionId(data.state.transaction_id());
    if (!id.ok()) {
      return id.status();
    }

    if (data.state.status() == TransactionStatus::APPLYING) {
      return ReplicatedApplying(*id, data);
    } else if (data.state.status() == TransactionStatus::ABORTED) {
      return ReplicatedAborted(*id, data);
    }

    auto status = STATUS_FORMAT(
        InvalidArgument, "Unexpected status in transaction participant ProcessReplicated: $0, $1",
        data.op_id, data.state);
    LOG_WITH_PREFIX(DFATAL) << status;
    return status;
  }

  void Cleanup(TransactionIdSet&& set, TransactionStatusManager* status_manager) {
    auto cleanup_aborts_task = std::make_shared<CleanupAbortsTask>(
        &applier_, std::move(set), &participant_context_, status_manager, LogPrefix());
    cleanup_aborts_task->Prepare(cleanup_aborts_task);
    participant_context_.StrandEnqueue(cleanup_aborts_task.get());
  }

  CHECKED_STATUS ProcessApply(const TransactionApplyData& data) {
    VLOG_WITH_PREFIX(2) << "Apply: " << data.ToString();

    loader_.WaitLoaded(data.transaction_id);

    ScopedRWOperation operation(pending_op_counter_);
    if (!operation.ok()) {
      LOG_WITH_PREFIX(WARNING) << "Process apply rejected";
      return Status::OK();
    }

    bool was_applied = false;

    {
      // It is our last chance to load transaction metadata, if missing.
      // Because it will be deleted when intents are applied.
      // We are not trying to cleanup intents here because we don't know whether this transaction
      // has intents of not.
      auto lock_and_iterator = LockAndFind(
          data.transaction_id, "pre apply"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
      if (!lock_and_iterator.found()) {
        // This situation is normal and could be caused by 2 scenarios:
        // 1) Write batch failed, but originator doesn't know that.
        // 2) Failed to notify status tablet that we applied transaction.
        YB_LOG_WITH_PREFIX_EVERY_N_SECS(WARNING, 1)
            << Format("Apply of unknown transaction: $0", data);
        NotifyApplied(data);
        CHECK(!FLAGS_TEST_fail_in_apply_if_no_metadata);
        return Status::OK();
      }

      auto existing_commit_ht = lock_and_iterator.transaction().local_commit_time();
      if (existing_commit_ht) {
        was_applied = true;
        LOG_WITH_PREFIX(INFO) << "Transaction already applied: " << data.transaction_id;
        LOG_IF_WITH_PREFIX(DFATAL, data.commit_ht != existing_commit_ht)
            << "Transaction was previously applied with another commit ht: " << existing_commit_ht
            << ", new commit ht: " << data.commit_ht;
      } else {
        transactions_.modify(lock_and_iterator.iterator, [&data](auto& txn) {
          txn->SetLocalCommitData(data.commit_ht, data.aborted);
        });

        LOG_IF_WITH_PREFIX(DFATAL, data.log_ht < last_safe_time_)
            << "Apply transaction before last safe time " << data.transaction_id
            << ": " << data.log_ht << " vs " << last_safe_time_;
      }
    }

    if (!was_applied) {
      auto apply_state = CHECK_RESULT(applier_.ApplyIntents(data));

      VLOG_WITH_PREFIX(4) << "TXN: " << data.transaction_id << ": apply state: "
                          << apply_state.ToString();

      UpdateAppliedTransaction(data, apply_state, &operation);
    }

    NotifyApplied(data);
    return Status::OK();
  }

  void UpdateAppliedTransaction(
       const TransactionApplyData& data,
       const docdb::ApplyTransactionState& apply_state,
       ScopedRWOperation* operation) NO_THREAD_SAFETY_ANALYSIS {
    MinRunningNotifier min_running_notifier(&applier_);
    // We are not trying to cleanup intents here because we don't know whether this transaction
    // has intents or not.
    auto lock_and_iterator = LockAndFind(
        data.transaction_id, "apply"s, TransactionLoadFlags{TransactionLoadFlag::kMustExist});
    if (lock_and_iterator.found()) {
      lock_and_iterator.transaction().SetOpId(data.op_id);
      if (!apply_state.active()) {
        RemoveUnlocked(lock_and_iterator.iterator, RemoveReason::kApplied, &min_running_notifier);
      } else {
        lock_and_iterator.transaction().SetApplyData(apply_state, &data, operation);
      }
    }
  }

  void NotifyApplied(const TransactionApplyData& data) {
    VLOG_WITH_PREFIX(4) << Format("NotifyApplied($0)", data);

    if (data.leader_term != OpId::kUnknownTerm) {
      tserver::UpdateTransactionRequestPB req;
      req.set_tablet_id(data.status_tablet);
      req.set_propagated_hybrid_time(participant_context_.Now().ToUint64());
      auto& state = *req.mutable_state();
      state.set_transaction_id(data.transaction_id.data(), data.transaction_id.size());
      state.set_status(TransactionStatus::APPLIED_IN_ONE_OF_INVOLVED_TABLETS);
      state.add_tablets(participant_context_.tablet_id());
      auto client_result = client();
      if (!client_result.ok()) {
        LOG_WITH_PREFIX(WARNING) << "Get client failed: " << client_result.status();
        return;
      }

      auto handle = rpcs_.Prepare();
      if (handle != rpcs_.InvalidHandle()) {
        *handle = UpdateTransaction(
            TransactionRpcDeadline(),
            nullptr /* remote_tablet */,
            *client_result,
            &req,
            [this, handle](const Status& status,
                           const tserver::UpdateTransactionRequestPB& req,
                           const tserver::UpdateTransactionResponsePB& resp) {
              client::UpdateClock(resp, &participant_context_);
              rpcs_.Unregister(handle);
              LOG_IF_WITH_PREFIX(WARNING, !status.ok()) << "Failed to send applied: " << status;
            });
        (**handle).SendRpc();
      }
    }
  }

  CHECKED_STATUS ProcessCleanup(const TransactionApplyData& data, CleanupType cleanup_type) {
    VLOG_WITH_PREFIX_AND_FUNC(4) << AsString(data) << ", " << AsString(cleanup_type);

    loader_.WaitLoaded(data.transaction_id);

    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(data.transaction_id);
    if (it == transactions_.end()) {
      if (cleanup_type == CleanupType::kImmediate) {
        cleanup_cache_.Insert(data.transaction_id);
        return Status::OK();
      }
    } else if ((**it).ProcessingApply()) {
      VLOG_WITH_PREFIX(2) << "Don't cleanup transaction because it is applying intents: "
                          << data.transaction_id;
      return Status::OK();
    }

    if (cleanup_type == CleanupType::kGraceful) {
      graceful_cleanup_queue_.push_back(GracefulCleanupQueueEntry{
        .request_id = request_serial_,
        .transaction_id = data.transaction_id,
        .reason = RemoveReason::kProcessCleanup,
        .required_safe_time = participant_context_.Now(),
      });
      return Status::OK();
    }

    if (!RemoveUnlocked(it, RemoveReason::kProcessCleanup, &min_running_notifier)) {
      VLOG_WITH_PREFIX(2) << "Have added aborted txn to cleanup queue: "
                          << data.transaction_id;
    }

    return Status::OK();
  }

  void SetDB(
      const docdb::DocDB& db, const docdb::KeyBounds* key_bounds,
      RWOperationCounter* pending_op_counter) {
    bool had_db = db_.intents != nullptr;
    db_ = db;
    key_bounds_ = key_bounds;
    pending_op_counter_ = pending_op_counter;

    // We should only load transactions on the initial call to SetDB (when opening the tablet), not
    // in case of truncate/restore.
    if (!had_db) {
      loader_.Start(pending_op_counter, db_);
      return;
    }

    loader_.WaitAllLoaded();
    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    transactions_.clear();
    TransactionsModifiedUnlocked(&min_running_notifier);
  }

  void GetStatus(
      const TransactionId& transaction_id,
      size_t required_num_replicated_batches,
      int64_t term,
      tserver::GetTransactionStatusAtParticipantResponsePB* response,
      rpc::RpcContext* context) {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(transaction_id);
    if (it == transactions_.end()) {
      response->set_num_replicated_batches(0);
      response->set_status_hybrid_time(0);
    } else {
      if ((**it).WasAborted()) {
        response->set_aborted(true);
        return;
      }
      response->set_num_replicated_batches((**it).num_replicated_batches());
      response->set_status_hybrid_time((**it).last_batch_data().hybrid_time.ToUint64());
    }
  }

  TransactionParticipantContext* participant_context() const {
    return &participant_context_;
  }

  HybridTime MinRunningHybridTime() {
    auto result = min_running_ht_.load(std::memory_order_acquire);
    if (result == HybridTime::kMax || result == HybridTime::kInvalid) {
      return result;
    }
    auto now = CoarseMonoClock::now();
    auto current_next_check_min_running = next_check_min_running_.load(std::memory_order_relaxed);
    if (now >= current_next_check_min_running) {
      if (next_check_min_running_.compare_exchange_strong(
              current_next_check_min_running,
              now + 1ms * FLAGS_transaction_min_running_check_interval_ms,
              std::memory_order_acq_rel)) {
        std::unique_lock<std::mutex> lock(mutex_);
        if (transactions_.empty()) {
          return HybridTime::kMax;
        }
        auto& first_txn = **transactions_.get<StartTimeTag>().begin();
        VLOG_WITH_PREFIX(1) << "Checking status of long running min txn " << first_txn.id()
                            << ": " << first_txn.WasAborted();
        static const std::string kRequestReason = "min running check"s;
        // Get transaction status
        auto now_ht = participant_context_.Now();
        StatusRequest status_request = {
            .id = &first_txn.id(),
            .read_ht = now_ht,
            .global_limit_ht = now_ht,
            // Could use 0 here, because read_ht == global_limit_ht.
            // So we cannot accept status with time >= read_ht and < global_limit_ht.
            .serial_no = 0,
            .reason = &kRequestReason,
            .flags = TransactionLoadFlags{},
            .callback = [this, id = first_txn.id()](Result<TransactionStatusResult> result) {
              // Aborted status will result in cleanup of intents.
              VLOG_WITH_PREFIX(1) << "Min running status " << id << ": " << result;
            }
        };
        first_txn.RequestStatusAt(status_request, &lock);
      }
    }
    return result;
  }

  void WaitMinRunningHybridTime(HybridTime ht) {
    MinRunningNotifier min_running_notifier(&applier_);
    std::unique_lock<std::mutex> lock(mutex_);
    waiting_for_min_running_ht_ = ht;
    CheckMinRunningHybridTimeSatisfiedUnlocked(&min_running_notifier);
  }

  CHECKED_STATUS ResolveIntents(HybridTime resolve_at, CoarseTimePoint deadline) {
    RETURN_NOT_OK(WaitUntil(participant_context_.clock_ptr().get(), resolve_at, deadline));

    if (FLAGS_max_transactions_in_status_request == 0) {
      return STATUS(
          IllegalState,
          "Cannot resolve intents when FLAGS_max_transactions_in_status_request is zero");
    }

    std::vector<TransactionId> recheck_ids, committed_ids;

    // Maintain a set of transactions, check their statuses, and remove them as they get
    // committed/applied, aborted or we realize that transaction was not committed at
    // resolve_at.
    for (;;) {
      TransactionStatusResolver resolver(
          &participant_context_, &rpcs_, FLAGS_max_transactions_in_status_request,
          [this, resolve_at, &recheck_ids, &committed_ids](
              const std::vector <TransactionStatusInfo>& status_infos) {
            std::vector<TransactionId> aborted;
            for (const auto& info : status_infos) {
              VLOG_WITH_PREFIX(4) << "Transaction status: " << info.ToString();
              if (info.status == TransactionStatus::COMMITTED) {
                if (info.status_ht <= resolve_at) {
                  // Transaction was committed, but not yet applied.
                  // So rely on filtering recheck_ids before next phase.
                  committed_ids.push_back(info.transaction_id);
                }
              } else if (info.status == TransactionStatus::ABORTED) {
                aborted.push_back(info.transaction_id);
              } else {
                LOG_IF_WITH_PREFIX(DFATAL, info.status != TransactionStatus::PENDING)
                    << "Transaction is in unexpected state: " << info.ToString();
                if (info.status_ht <= resolve_at) {
                  recheck_ids.push_back(info.transaction_id);
                }
              }
            }
            if (!aborted.empty()) {
              MinRunningNotifier min_running_notifier(&applier_);
              std::lock_guard<std::mutex> lock(mutex_);
              for (const auto& id : aborted) {
                EnqueueRemoveUnlocked(id, RemoveReason::kStatusReceived, &min_running_notifier);
              }
            }
          });
      auto se = ScopeExit([&resolver] {
        resolver.Shutdown();
      });
      {
        std::lock_guard <std::mutex> lock(mutex_);
        if (recheck_ids.empty() && committed_ids.empty()) {
          // First step, check all transactions.
          for (const auto& transaction : transactions_) {
            if (!transaction->local_commit_time().is_valid()) {
              resolver.Add(transaction->metadata().status_tablet, transaction->id());
            }
          }
        } else {
          for (const auto& id : recheck_ids) {
            auto it = transactions_.find(id);
            if (it == transactions_.end() || (**it).local_commit_time().is_valid()) {
              continue;
            }
            resolver.Add((**it).metadata().status_tablet, id);
          }
          auto filter = [this](const TransactionId& id) {
            auto it = transactions_.find(id);
            return it == transactions_.end() || (**it).local_commit_time().is_valid();
          };
          committed_ids.erase(std::remove_if(committed_ids.begin(), committed_ids.end(), filter),
                              committed_ids.end());
        }
      }

      recheck_ids.clear();
      resolver.Start(deadline);

      RETURN_NOT_OK(resolver.ResultFuture().get());

      if (recheck_ids.empty()) {
        if (committed_ids.empty()) {
          break;
        } else {
          // We are waiting only for committed transactions to be applied.
          // So just add some delay.
          std::this_thread::sleep_for(10ms * std::min<size_t>(10, committed_ids.size()));
        }
      }
    }

    return Status::OK();
  }

  size_t TEST_GetNumRunningTransactions() {
    std::lock_guard<std::mutex> lock(mutex_);
    auto txn_to_id = [](const RunningTransactionPtr& txn) {
      return txn->id();
    };
    VLOG_WITH_PREFIX(4) << "Transactions: " << AsString(transactions_, txn_to_id)
                        << ", requests: " << AsString(running_requests_);
    return transactions_.size();
  }

  OneWayBitmap TEST_TransactionReplicatedBatches(const TransactionId& id) {
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(id);
    return it != transactions_.end() ? (**it).replicated_batches() : OneWayBitmap();
  }

  std::string DumpTransactions() {
    std::string result;
    std::lock_guard<std::mutex> lock(mutex_);

    result += Format(
        "{ safe_time_for_participant: $0 remove_queue_size: $1 ",
        participant_context_.SafeTimeForTransactionParticipant(), remove_queue_.size());
    if (!remove_queue_.empty()) {
      result += "remove_queue_front: " + AsString(remove_queue_.front());
    }
    if (!running_requests_.empty()) {
      result += "running_requests_front: " + AsString(running_requests_.front());
    }
    result += "}\n";

    for (const auto& txn : transactions_.get<StartTimeTag>()) {
      result += txn->ToString();
      result += "\n";
    }
    return result;
  }

  CHECKED_STATUS StopActiveTxnsPriorTo(
      HybridTime cutoff, CoarseTimePoint deadline, TransactionId* exclude_txn_id) {
    vector<TransactionId> ids_to_abort;
    {
      std::lock_guard<std::mutex> lock(mutex_);
      for (const auto& txn : transactions_.get<StartTimeTag>()) {
        if (txn->start_ht() > cutoff ||
            (exclude_txn_id != nullptr && txn->id() == *exclude_txn_id)) {
          break;
        }
        if (!txn->WasAborted()) {
          ids_to_abort.push_back(txn->id());
        }
      }
    }

    if (ids_to_abort.empty()) {
      return Status::OK();
    }

    // It is ok to attempt to abort txns that have committed. We don't care
    // if our request succeeds or not.
    CountDownLatch latch(ids_to_abort.size());
    std::atomic<bool> failed{false};
    Status return_status = Status::OK();
    for (const auto& id : ids_to_abort) {
      Abort(
          id, [this, id, &failed, &return_status, &latch](Result<TransactionStatusResult> result) {
            VLOG_WITH_PREFIX(2) << "Aborting " << id << " got " << result;
            if (!result ||
                (result->status != TransactionStatus::COMMITTED && result->status != ABORTED)) {
              LOG(INFO) << "Could not abort " << id << " got " << result;

              bool expected = false;
              if (failed.compare_exchange_strong(expected, true)) {
                if (!result) {
                  return_status = result.status();
                } else {
                  return_status =
                      STATUS_FORMAT(IllegalState, "Wrong status after abort: $0", result->status);
                }
              }
            }
            latch.CountDown();
          });
    }

    return latch.WaitUntil(deadline) ? return_status
                                     : STATUS(TimedOut, "TimedOut while aborting old transactions");
  }

  Result<HybridTime> WaitForSafeTime(HybridTime safe_time, CoarseTimePoint deadline) {
    return participant_context_.WaitForSafeTime(safe_time, deadline);
  }

  void IgnoreAllTransactionsStartedBefore(HybridTime limit) {
    std::lock_guard<std::mutex> lock(mutex_);
    ignore_all_transactions_started_before_ =
        std::max(ignore_all_transactions_started_before_, limit);
  }

 private:
  class AbortCheckTimeTag;
  class StartTimeTag;

  typedef boost::multi_index_container<RunningTransactionPtr,
      boost::multi_index::indexed_by <
          boost::multi_index::hashed_unique <
              boost::multi_index::const_mem_fun <
                  RunningTransaction, const TransactionId&, &RunningTransaction::id>
          >,
          boost::multi_index::ordered_non_unique <
              boost::multi_index::tag<StartTimeTag>,
              boost::multi_index::const_mem_fun <
                  RunningTransaction, HybridTime, &RunningTransaction::start_ht>
          >,
          boost::multi_index::ordered_non_unique <
              boost::multi_index::tag<AbortCheckTimeTag>,
              boost::multi_index::const_mem_fun <
                  RunningTransaction, HybridTime, &RunningTransaction::abort_check_ht>
          >
      >
  > Transactions;

  void CompleteLoad(const std::function<void()>& functor) override {
    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    functor();
    TransactionsModifiedUnlocked(&min_running_notifier);
  }

  void LoadFinished(const ApplyStatesMap& pending_applies) override {
    start_latch_.Wait();
    std::vector<ScopedRWOperation> operations;
    operations.reserve(pending_applies.size());
    for (;;) {
      if (closing_.load(std::memory_order_acquire)) {
        LOG_WITH_PREFIX(INFO)
            << __func__ << ": closing, not starting transaction status resolution";
        return;
      }
      while (operations.size() < pending_applies.size()) {
        ScopedRWOperation operation(pending_op_counter_);
        if (!operation.ok()) {
          break;
        }
        operations.push_back(std::move(operation));
      }
      if (operations.size() == pending_applies.size()) {
        break;
      }
      operations.clear();
      YB_LOG_WITH_PREFIX_EVERY_N_SECS(INFO, 5)
          << __func__ << ": unable to start scoped RW operation";
      std::this_thread::sleep_for(10ms);
    }

    if (!pending_applies.empty()) {
      LOG_WITH_PREFIX(INFO)
          << __func__ << ": starting " << pending_applies.size() << " pending applies";
      std::lock_guard<std::mutex> lock(mutex_);
      size_t idx = 0;
      for (const auto& p : pending_applies) {
        auto it = transactions_.find(p.first);
        if (it == transactions_.end()) {
          LOG_WITH_PREFIX(INFO) << "Unknown transaction for pending apply: " << AsString(p.first);
          continue;
        }

        TransactionApplyData apply_data;
        apply_data.transaction_id = p.first;
        apply_data.commit_ht = p.second.commit_ht;
        (**it).SetApplyData(p.second.state, &apply_data, &operations[idx]);
        ++idx;
      }
    }

    {
      LOG_WITH_PREFIX(INFO) << __func__ << ": starting transaction status resolution";
      std::lock_guard<std::mutex> lock(status_resolvers_mutex_);
      for (auto& status_resolver : status_resolvers_) {
        status_resolver.Start(CoarseTimePoint::max());
      }
    }

    poller_.Start(
        &participant_context_.scheduler(), 1ms * FLAGS_transactions_status_poll_interval_ms);
  }

  void TransactionsModifiedUnlocked(MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) {
    metric_transactions_running_->set_value(transactions_.size());
    if (!loader_.complete()) {
      return;
    }

    if (transactions_.empty()) {
      min_running_ht_.store(HybridTime::kMax, std::memory_order_release);
      CheckMinRunningHybridTimeSatisfiedUnlocked(min_running_notifier);
      return;
    }

    auto& first_txn = **transactions_.get<StartTimeTag>().begin();
    if (first_txn.start_ht() != min_running_ht_.load(std::memory_order_relaxed)) {
      min_running_ht_.store(first_txn.start_ht(), std::memory_order_release);
      next_check_min_running_.store(
          CoarseMonoClock::now() + 1ms * FLAGS_transaction_min_running_check_delay_ms,
          std::memory_order_release);
      CheckMinRunningHybridTimeSatisfiedUnlocked(min_running_notifier);
      return;
    }
  }

  void EnqueueRemoveUnlocked(
      const TransactionId& id, RemoveReason reason,
      MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) override {
    auto now = participant_context_.Now();
    VLOG_WITH_PREFIX_AND_FUNC(4) << id << " at " << now << ", reason: " << AsString(reason);
    remove_queue_.emplace_back(RemoveQueueEntry{
      .id = id,
      .time = now,
      .reason = reason,
    });
    ProcessRemoveQueueUnlocked(min_running_notifier);
  }

  void ProcessRemoveQueueUnlocked(MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) {
    if (!remove_queue_.empty()) {
      // When a transaction participant receives an "aborted" response from the coordinator,
      // it puts this transaction into a "remove queue", also storing the current hybrid
      // time. Then queue entries where time is less than current safe time are removed.
      //
      // This is correct because, from a transaction participant's point of view:
      //
      // (1) After we receive a response for a transaction status request, and
      // learn that the transaction is unknown to the coordinator, our local
      // hybrid time is at least as high as the local hybrid time on the
      // transaction status coordinator at the time the transaction was deleted
      // from the coordinator, due to hybrid time propagation on RPC response.
      //
      // (2) If our safe time is greater than the hybrid time when the
      // transaction was deleted from the coordinator, then we have already
      // applied this transaction's provisional records if the transaction was
      // committed.
      auto safe_time = participant_context_.SafeTimeForTransactionParticipant();
      if (!safe_time.is_valid()) {
        VLOG_WITH_PREFIX(3) << "Unable to obtain safe time to check remove queue";
        return;
      }
      VLOG_WITH_PREFIX(3) << "Checking remove queue: " << safe_time << ", "
                          << remove_queue_.front().ToString();
      LOG_IF_WITH_PREFIX(DFATAL, safe_time < last_safe_time_)
          << "Safe time decreased: " << safe_time << " vs " << last_safe_time_;
      last_safe_time_ = safe_time;
      while (!remove_queue_.empty()) {
        auto& front = remove_queue_.front();
        auto it = transactions_.find(front.id);
        if (it == transactions_.end() || (**it).local_commit_time().is_valid()) {
          // It is regular case, since the coordinator returns ABORTED for already applied
          // transaction. But this particular tablet could not yet apply it, so
          // it would add such transaction to remove queue.
          // And it is the main reason why we are waiting for safe time, before removing intents.
          VLOG_WITH_PREFIX(4) << "Evicting txn from remove queue, w/o removing intents: "
                              << front.ToString();
          remove_queue_.pop_front();
          continue;
        }
        if (safe_time <= front.time) {
          break;
        }
        VLOG_WITH_PREFIX(4) << "Removing from remove queue: " << front.ToString();
        RemoveUnlocked(front.id, front.reason, min_running_notifier);
        remove_queue_.pop_front();
      }
    }
  }

  // Tries to remove transaction with specified id.
  // Returns true if transaction is not exists after call to this method, otherwise returns false.
  // Which means that transaction will be removed later.
  bool RemoveUnlocked(
      const TransactionId& id, RemoveReason reason,
      MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) override {
    auto it = transactions_.find(id);
    if (it == transactions_.end()) {
      return true;
    }
    return RemoveUnlocked(it, reason, min_running_notifier);
  }

  bool RemoveUnlocked(
      const Transactions::iterator& it, RemoveReason reason,
      MinRunningNotifier* min_running_notifier) REQUIRES(mutex_) {
    TransactionId txn_id = (**it).id();
    RemoveIntentsData checkpoint;
    auto itr = transactions_.find(txn_id);
    OpId op_id = (**itr).GetOpId();
    participant_context_.GetLastCDCedData(&checkpoint);

    VLOG_WITH_PREFIX(2) << "Cleaning tx, data opid is " << op_id.ToString()
              << " checkpoint opid is " << checkpoint.op_id.ToString();

    if (running_requests_.empty() &&
        (op_id < checkpoint.op_id)) {
      (**it).ScheduleRemoveIntents(*it);
      RemoveTransaction(it, reason, min_running_notifier);
      VLOG_WITH_PREFIX(2) << "Cleaned transaction: " << txn_id << ", reason: " << reason
                          << ", left: " << transactions_.size();
      return true;
    }

    // We cannot remove the transaction at this point, because there are running requests
    // that are reading the provisional DB and could request status of this transaction.
    // So we store transaction in a queue and wait when all requests that we launched before our
    // attempt to remove this transaction are completed.
    // Since we try to remove the transaction after all its records are removed from the provisional
    // DB, it is safe to complete removal at this point, because it means that there will be no more
    // queries to status of this transactions.
    immediate_cleanup_queue_.push_back(ImmediateCleanupQueueEntry{
      .request_id = request_serial_,
      .transaction_id = (**it).id(),
      .reason = reason,
    });
    VLOG_WITH_PREFIX(2) << "Queued for cleanup: " << (**it).id() << ", reason: " << reason;
    return false;
  }

  struct LockAndFindResult {
    static Transactions::const_iterator UninitializedIterator() {
      static const Transactions empty_transactions;
      return empty_transactions.end();
    }

    std::unique_lock<std::mutex> lock;
    Transactions::const_iterator iterator = UninitializedIterator();
    bool recently_removed = false;

    bool found() const {
      return lock.owns_lock();
    }

    RunningTransaction& transaction() const {
      return **iterator;
    }
  };

  LockAndFindResult LockAndFind(
      const TransactionId& id, const std::string& reason, TransactionLoadFlags flags) {
    loader_.WaitLoaded(id);
    bool recently_removed;
    {
      std::unique_lock<std::mutex> lock(mutex_);
      auto it = transactions_.find(id);
      if (it != transactions_.end()) {
        if ((**it).start_ht() <= ignore_all_transactions_started_before_) {
          YB_LOG_WITH_PREFIX_EVERY_N_SECS(INFO, 1)
              << "Ignore transaction for '" << reason << "' because of limit: "
              << ignore_all_transactions_started_before_ << ", txn: " << AsString(**it);
          return LockAndFindResult{};
        }
        return LockAndFindResult{ std::move(lock), it };
      }
      recently_removed = WasTransactionRecentlyRemoved(id);
    }
    if (recently_removed) {
      VLOG_WITH_PREFIX(1)
          << "Attempt to load recently removed transaction: " << id << ", for: " << reason;
      LockAndFindResult result;
      result.recently_removed = true;
      return result;
    }
    metric_transaction_not_found_->Increment();
    if (flags.Test(TransactionLoadFlag::kMustExist)) {
      YB_LOG_WITH_PREFIX_EVERY_N_SECS(WARNING, 1)
          << "Transaction not found: " << id << ", for: " << reason;
    } else {
      YB_LOG_WITH_PREFIX_EVERY_N_SECS(INFO, 1)
          << "Transaction not found: " << id << ", for: " << reason;
    }
    if (flags.Test(TransactionLoadFlag::kCleanup)) {
      VLOG_WITH_PREFIX(2) << "Schedule cleanup for: " << id;
      auto cleanup_task = std::make_shared<CleanupIntentsTask>(
          &participant_context_, &applier_, id);
      cleanup_task->Prepare(cleanup_task);
      participant_context_.StrandEnqueue(cleanup_task.get());
    }
    return LockAndFindResult{};
  }

  void LoadTransaction(
      TransactionMetadata&& metadata,
      TransactionalBatchData&& last_batch_data,
      OneWayBitmap&& replicated_batches,
      const ApplyStateWithCommitHt* pending_apply) override {
    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    auto txn = std::make_shared<RunningTransaction>(
        std::move(metadata), std::move(last_batch_data), std::move(replicated_batches),
        participant_context_.Now().AddDelta(1ms * FLAGS_transaction_abort_check_timeout_ms), this);
    if (pending_apply) {
      VLOG_WITH_PREFIX(4) << "Apply state found for " << txn->id() << ": "
                          << pending_apply->ToString();
      txn->SetLocalCommitData(pending_apply->commit_ht, pending_apply->state.aborted);
      txn->SetApplyData(pending_apply->state);
    }
    transactions_.insert(txn);
    TransactionsModifiedUnlocked(&min_running_notifier);
  }

  Result<client::YBClient*> client() const {
    auto cached_value = client_cache_.load(std::memory_order_acquire);
    if (cached_value != nullptr) {
      return cached_value;
    }
    auto future_status = participant_context_.client_future().wait_for(
        TransactionRpcTimeout().ToSteadyDuration());
    if (future_status != std::future_status::ready) {
      return STATUS(TimedOut, "Client not ready");
    }
    auto result = participant_context_.client_future().get();
    client_cache_.store(result, std::memory_order_release);
    return result;
  }

  const std::string& LogPrefix() const override {
    return log_prefix_;
  }

  void RemoveTransaction(Transactions::iterator it, RemoveReason reason,
                         MinRunningNotifier* min_running_notifier)
      REQUIRES(mutex_) {
    auto now = CoarseMonoClock::now();
    CleanupRecentlyRemovedTransactions(now);
    auto& transaction = **it;
    YB_TRANSACTION_DUMP(
        Remove, participant_context_.tablet_id(), transaction.id(), participant_context_.Now(),
        static_cast<uint8_t>(reason));
    recently_removed_transactions_cleanup_queue_.push_back({transaction.id(), now + 15s});
    LOG_IF_WITH_PREFIX(DFATAL, !recently_removed_transactions_.insert(transaction.id()).second)
        << "Transaction removed twice: " << transaction.id();
    VLOG_WITH_PREFIX(4) << "Remove transaction: " << transaction.id();
    transactions_.erase(it);
    TransactionsModifiedUnlocked(min_running_notifier);
  }

  void CleanupRecentlyRemovedTransactions(CoarseTimePoint now) {
    while (!recently_removed_transactions_cleanup_queue_.empty() &&
           recently_removed_transactions_cleanup_queue_.front().time <= now) {
      recently_removed_transactions_.erase(recently_removed_transactions_cleanup_queue_.front().id);
      recently_removed_transactions_cleanup_queue_.pop_front();
    }
  }

  bool WasTransactionRecentlyRemoved(const TransactionId& id) {
    CleanupRecentlyRemovedTransactions(CoarseMonoClock::now());
    return recently_removed_transactions_.count(id) != 0;
  }

  void CheckMinRunningHybridTimeSatisfiedUnlocked(
      MinRunningNotifier* min_running_notifier) {
    if (min_running_ht_.load(std::memory_order_acquire) <= waiting_for_min_running_ht_) {
      return;
    }
    waiting_for_min_running_ht_ = HybridTime::kMax;
    min_running_notifier->Satisfied();
  }

  void TransactionsStatus(
      const std::vector<TransactionStatusInfo>& status_infos) {
    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    HybridTime now = participant_context_.Now();
    for (const auto& info : status_infos) {
      auto it = transactions_.find(info.transaction_id);
      if (it == transactions_.end()) {
        continue;
      }
      if ((**it).UpdateStatus(
          info.status, info.status_ht, info.coordinator_safe_time, info.aborted_subtxn_set)) {
        EnqueueRemoveUnlocked(
            info.transaction_id, RemoveReason::kStatusReceived, &min_running_notifier);
      } else {
        transactions_.modify(it, [now](const auto& txn) {
          txn->UpdateAbortCheckHT(now, UpdateAbortCheckHTMode::kStatusResponseReceived);
        });
      }
    }
  }

  void HandleApplying(std::unique_ptr<tablet::UpdateTxnOperation> operation, int64_t term) {
    if (RandomActWithProbability(GetAtomicFlag(
        &FLAGS_TEST_transaction_ignore_applying_probability))) {
      VLOG_WITH_PREFIX(2)
          << "TEST: Rejected apply: "
          << FullyDecodeTransactionId(operation->request()->transaction_id());
      operation->CompleteWithStatus(Status::OK());
      return;
    }
    participant_context_.SubmitUpdateTransaction(std::move(operation), term);
  }

  void HandleCleanup(
      std::unique_ptr<tablet::UpdateTxnOperation> operation, int64_t term,
      CleanupType cleanup_type) {
    VLOG_WITH_PREFIX(3) << "Cleanup";
    auto id = FullyDecodeTransactionId(operation->request()->transaction_id());
    if (!id.ok()) {
      operation->CompleteWithStatus(id.status());
      return;
    }

    TransactionApplyData data = {
        .leader_term = term,
        .transaction_id = *id,
        .aborted = AbortedSubTransactionSet(),
        .op_id = OpId(),
        .commit_ht = HybridTime(),
        .log_ht = HybridTime(),
        .sealed = operation->request()->sealed(),
        .status_tablet = std::string()
    };
    WARN_NOT_OK(ProcessCleanup(data, cleanup_type), "Process cleanup failed");
    operation->CompleteWithStatus(Status::OK());
  }

  CHECKED_STATUS ReplicatedApplying(const TransactionId& id, const ReplicatedData& data) {
    // data.state.tablets contains only status tablet.
    if (data.state.tablets_size() != 1) {
      return STATUS_FORMAT(InvalidArgument,
                           "Expected only one table during APPLYING, state received: $0",
                           data.state);
    }
    HybridTime commit_time(data.state.commit_hybrid_time());
    TransactionApplyData apply_data = {
        .leader_term = data.leader_term,
        .transaction_id = id,
        .aborted = VERIFY_RESULT(AbortedSubTransactionSet::FromPB(data.state.aborted().set())),
        .op_id = data.op_id,
        .commit_ht = commit_time,
        .log_ht = data.hybrid_time,
        .sealed = data.sealed,
        .status_tablet = data.state.tablets(0)
      };
    if (!data.already_applied_to_regular_db) {
      return ProcessApply(apply_data);
    }
    if (!data.sealed) {
      return ProcessCleanup(apply_data, CleanupType::kImmediate);
    }
    return Status::OK();
  }

  CHECKED_STATUS ReplicatedAborted(const TransactionId& id, const ReplicatedData& data) {
    MinRunningNotifier min_running_notifier(&applier_);
    std::lock_guard<std::mutex> lock(mutex_);
    auto it = transactions_.find(id);
    if (it == transactions_.end()) {
      TransactionMetadata metadata = {
        .transaction_id = id,
        .isolation = IsolationLevel::NON_TRANSACTIONAL,
        .status_tablet = TabletId(),
        .priority = 0
      };
      it = transactions_.insert(std::make_shared<RunningTransaction>(
          metadata, TransactionalBatchData(), OneWayBitmap(), HybridTime::kMax, this)).first;
      TransactionsModifiedUnlocked(&min_running_notifier);
    }

    // TODO(dtxn) store this fact to rocksdb.
    (**it).Aborted();

    return Status::OK();
  }

  void Poll() {
    {
      MinRunningNotifier min_running_notifier(&applier_);
      std::lock_guard<std::mutex> lock(mutex_);

      ProcessRemoveQueueUnlocked(&min_running_notifier);
      if (ANNOTATE_UNPROTECTED_READ(FLAGS_transactions_poll_check_aborted)) {
        CheckForAbortedTransactions();
      }
      CleanTransactionsQueue(&graceful_cleanup_queue_, &min_running_notifier);
    }
    CleanupStatusResolvers();
  }

  void CheckForAbortedTransactions() REQUIRES(mutex_) {
    if (transactions_.empty()) {
      return;
    }
    auto now = participant_context_.Now();
    auto& index = transactions_.get<AbortCheckTimeTag>();
    TransactionStatusResolver* resolver = nullptr;
    for (;;) {
      auto& txn = **index.begin();
      if (txn.abort_check_ht() > now) {
        break;
      }
      if (!resolver) {
        resolver = &AddStatusResolver();
      }
      const auto& metadata = txn.metadata();
      VLOG_WITH_PREFIX(4)
          << "Check aborted: " << metadata.status_tablet << ", " << metadata.transaction_id;
      resolver->Add(metadata.status_tablet, metadata.transaction_id);
      index.modify(index.begin(), [now](const auto& txn) {
        txn->UpdateAbortCheckHT(now, UpdateAbortCheckHTMode::kStatusRequestSent);
      });
    }

    // We don't introduce limit on number of status resolutions here, because we cannot predict
    // transactions throughput. And we rely the logic that we cannot start multiple resolutions
    // for single transaction because we set abort check hybrid time to the same value as
    // status resolution deadline.
    if (resolver) {
      resolver->Start(CoarseMonoClock::now() + 1ms * FLAGS_transaction_abort_check_timeout_ms);
    }
  }

  void CleanupStatusResolvers() EXCLUDES(status_resolvers_mutex_) {
    std::lock_guard<std::mutex> lock(status_resolvers_mutex_);
    while (!status_resolvers_.empty() && !status_resolvers_.front().Running()) {
      status_resolvers_.front().Shutdown();
      status_resolvers_.pop_front();
    }
  }

  TransactionStatusResolver& AddStatusResolver() override EXCLUDES(status_resolvers_mutex_) {
    std::lock_guard<std::mutex> lock(status_resolvers_mutex_);
    status_resolvers_.emplace_back(
        &participant_context_, &rpcs_, FLAGS_max_transactions_in_status_request,
        std::bind(&Impl::TransactionsStatus, this, _1));
    return status_resolvers_.back();
  }

  struct ImmediateCleanupQueueEntry {
    int64_t request_id;
    TransactionId transaction_id;
    RemoveReason reason;

    bool Ready(TransactionParticipantContext* participant_context, HybridTime* safe_time) const {
      return true;
    }
  };

  struct GracefulCleanupQueueEntry {
    int64_t request_id;
    TransactionId transaction_id;
    RemoveReason reason;
    HybridTime required_safe_time;

    bool Ready(TransactionParticipantContext* participant_context, HybridTime* safe_time) const {
      if (!*safe_time) {
        *safe_time = participant_context->SafeTimeForTransactionParticipant();
      }
      return *safe_time >= required_safe_time;
    }
  };

  std::string log_prefix_;

  docdb::DocDB db_;
  const docdb::KeyBounds* key_bounds_;
  // Owned externally, should be guaranteed that would not be destroyed before this.
  RWOperationCounter* pending_op_counter_ = nullptr;

  Transactions transactions_;
  // Ids of running requests, stored in increasing order.
  std::deque<int64_t> running_requests_;
  // Ids of complete requests, minimal request is on top.
  // Contains only ids greater than first running request id, otherwise entry is removed
  // from both collections.
  std::priority_queue<int64_t, std::vector<int64_t>, std::greater<void>> complete_requests_;

  // Queues of transaction ids that should be cleaned, paired with request that should be completed
  // in order to be able to do clean.
  // Immediate cleanup is performed as soon as possible.
  // Graceful cleanup is performed after safe time becomes greater than cleanup request hybrid time.
  std::deque<ImmediateCleanupQueueEntry> immediate_cleanup_queue_ GUARDED_BY(mutex_);
  std::deque<GracefulCleanupQueueEntry> graceful_cleanup_queue_ GUARDED_BY(mutex_);

  // Remove queue maintains transactions that could be cleaned when safe time for follower reaches
  // appropriate time for an entry.
  // Since we add entries with increasing time, this queue is ordered by time.
  struct RemoveQueueEntry {
    TransactionId id;
    HybridTime time;
    RemoveReason reason;

    std::string ToString() const {
      return YB_STRUCT_TO_STRING(id, time, reason);
    }
  };

  // Guarded by RunningTransactionContext::mutex_
  std::deque<RemoveQueueEntry> remove_queue_;

  // Guarded by RunningTransactionContext::mutex_
  HybridTime last_safe_time_ = HybridTime::kMin;

  HybridTime ignore_all_transactions_started_before_ GUARDED_BY(mutex_) = HybridTime::kMin;

  std::unordered_set<TransactionId, TransactionIdHash> recently_removed_transactions_;
  struct RecentlyRemovedTransaction {
    TransactionId id;
    CoarseTimePoint time;
  };
  std::deque<RecentlyRemovedTransaction> recently_removed_transactions_cleanup_queue_;

  std::mutex status_resolvers_mutex_;
  std::deque<TransactionStatusResolver> status_resolvers_ GUARDED_BY(status_resolvers_mutex_);

  scoped_refptr<AtomicGauge<uint64_t>> metric_transactions_running_;
  scoped_refptr<Counter> metric_transaction_not_found_;

  TransactionLoader loader_;
  std::atomic<bool> closing_{false};
  CountDownLatch start_latch_{1};

  std::atomic<HybridTime> min_running_ht_{HybridTime::kInvalid};
  std::atomic<CoarseTimePoint> next_check_min_running_{CoarseTimePoint()};
  HybridTime waiting_for_min_running_ht_ = HybridTime::kMax;
  std::atomic<bool> shutdown_done_{false};

  mutable std::atomic<client::YBClient*> client_cache_{nullptr};

  LRUCache<TransactionId> cleanup_cache_{FLAGS_transactions_cleanup_cache_size};

  rpc::Poller poller_;
};

TransactionParticipant::TransactionParticipant(
    TransactionParticipantContext* context, TransactionIntentApplier* applier,
    const scoped_refptr<MetricEntity>& entity)
    : impl_(new Impl(context, applier, entity)) {
}

TransactionParticipant::~TransactionParticipant() {
}

void TransactionParticipant::Start() {
  impl_->Start();
}

Result<bool> TransactionParticipant::Add(const TransactionMetadata& metadata) {
  return impl_->Add(metadata);
}

Result<TransactionMetadata> TransactionParticipant::PrepareMetadata(
    const TransactionMetadataPB& pb) {
  return impl_->PrepareMetadata(pb);
}

boost::optional<std::pair<IsolationLevel, TransactionalBatchData>>
    TransactionParticipant::PrepareBatchData(
    const TransactionId& id, size_t batch_idx,
    boost::container::small_vector_base<uint8_t>* encoded_replicated_batches) {
  return impl_->PrepareBatchData(id, batch_idx, encoded_replicated_batches);
}

void TransactionParticipant::BatchReplicated(
    const TransactionId& id, const TransactionalBatchData& data) {
  return impl_->BatchReplicated(id, data);
}

HybridTime TransactionParticipant::LocalCommitTime(const TransactionId& id) {
  return impl_->LocalCommitTime(id);
}

boost::optional<CommitMetadata> TransactionParticipant::LocalCommitData(const TransactionId& id) {
  return impl_->LocalCommitData(id);
}

std::pair<size_t, size_t> TransactionParticipant::TEST_CountIntents() const {
  return impl_->TEST_CountIntents();
}

void TransactionParticipant::RequestStatusAt(const StatusRequest& request) {
  return impl_->RequestStatusAt(request);
}

int64_t TransactionParticipant::RegisterRequest() {
  return impl_->RegisterRequest();
}

void TransactionParticipant::UnregisterRequest(int64_t request) {
  impl_->UnregisterRequest(request);
}

void TransactionParticipant::Abort(const TransactionId& id,
                                   TransactionStatusCallback callback) {
  return impl_->Abort(id, std::move(callback));
}

void TransactionParticipant::Handle(
    std::unique_ptr<tablet::UpdateTxnOperation> request, int64_t term) {
  impl_->Handle(std::move(request), term);
}

void TransactionParticipant::Cleanup(TransactionIdSet&& set) {
  return impl_->Cleanup(std::move(set), this);
}

Status TransactionParticipant::ProcessReplicated(const ReplicatedData& data) {
  return impl_->ProcessReplicated(data);
}

Status TransactionParticipant::CheckAborted(const TransactionId& id) {
  return impl_->CheckAborted(id);
}

void TransactionParticipant::FillPriorities(
    boost::container::small_vector_base<std::pair<TransactionId, uint64_t>>* inout) {
  return impl_->FillPriorities(inout);
}

void TransactionParticipant::SetDB(
    const docdb::DocDB& db, const docdb::KeyBounds* key_bounds,
    RWOperationCounter* pending_op_counter) {
  impl_->SetDB(db, key_bounds, pending_op_counter);
}

void TransactionParticipant::GetStatus(
    const TransactionId& transaction_id,
    size_t required_num_replicated_batches,
    int64_t term,
    tserver::GetTransactionStatusAtParticipantResponsePB* response,
    rpc::RpcContext* context) {
  impl_->GetStatus(transaction_id, required_num_replicated_batches, term, response, context);
}

TransactionParticipantContext* TransactionParticipant::context() const {
  return impl_->participant_context();
}

HybridTime TransactionParticipant::MinRunningHybridTime() const {
  return impl_->MinRunningHybridTime();
}

void TransactionParticipant::WaitMinRunningHybridTime(HybridTime ht) {
  impl_->WaitMinRunningHybridTime(ht);
}

Status TransactionParticipant::ResolveIntents(HybridTime resolve_at, CoarseTimePoint deadline) {
  return impl_->ResolveIntents(resolve_at, deadline);
}

size_t TransactionParticipant::TEST_GetNumRunningTransactions() const {
  return impl_->TEST_GetNumRunningTransactions();
}

OneWayBitmap TransactionParticipant::TEST_TransactionReplicatedBatches(
    const TransactionId& id) const {
  return impl_->TEST_TransactionReplicatedBatches(id);
}

std::string TransactionParticipant::ReplicatedData::ToString() const {
  return YB_STRUCT_TO_STRING(leader_term, state, op_id, hybrid_time, already_applied_to_regular_db);
}

void TransactionParticipant::StartShutdown() {
  impl_->StartShutdown();
}

void TransactionParticipant::CompleteShutdown() {
  impl_->CompleteShutdown();
}

std::string TransactionParticipant::DumpTransactions() const {
  return impl_->DumpTransactions();
}

Status TransactionParticipant::StopActiveTxnsPriorTo(
    HybridTime cutoff, CoarseTimePoint deadline, TransactionId* exclude_txn_id) {
  return impl_->StopActiveTxnsPriorTo(cutoff, deadline, exclude_txn_id);
}

Result<HybridTime> TransactionParticipant::WaitForSafeTime(
    HybridTime safe_time, CoarseTimePoint deadline) {
  return impl_->WaitForSafeTime(safe_time, deadline);
}

void TransactionParticipant::IgnoreAllTransactionsStartedBefore(HybridTime limit) {
  impl_->IgnoreAllTransactionsStartedBefore(limit);
}

const TabletId& TransactionParticipant::tablet_id() const {
  return impl_->participant_context()->tablet_id();
}

std::string TransactionParticipantContext::LogPrefix() const {
  return consensus::MakeTabletLogPrefix(tablet_id(), permanent_uuid());
}

HybridTime TransactionParticipantContext::Now() {
  return clock_ptr()->Now();
}

} // namespace tablet
} // namespace yb

YugabyteDB (2.13.0.0-b42, bfc6a6643e7399ac8a0e81d06a3ee6d6571b33ab)

Coverage Report

Created: 2022-03-09 17:30