void commit()

in src/consensus/aft/raft.h [2131:2306]

• 150 lines of code
• 38 McCabe index (conditional complexity)

    void commit(Index idx)
    {
      if (idx > state->last_idx)
      {
        throw std::logic_error(fmt::format(
          "Tried to commit {} but last_idx is {}", idx, state->last_idx));
      }

      LOG_DEBUG_FMT("Starting commit");

      // This could happen if a follower becomes the leader when it
      // has committed fewer log entries, although it has them available.
      if (idx <= state->commit_idx)
        return;

      state->commit_idx = idx;
      if (
        is_retired() && retirement_phase == kv::RetirementPhase::Signed &&
        retirement_committable_idx.has_value() &&
        idx >= retirement_committable_idx.value())
      {
        become_retired(idx, kv::RetirementPhase::Completed);
      }

      LOG_DEBUG_FMT("Compacting...");
      // Snapshots are not yet supported with BFT
      snapshotter->commit(
        idx,
        leadership_state == kv::LeadershipState::Leader &&
          consensus_type == ConsensusType::CFT);

      store->compact(idx);
      ledger->commit(idx);

      LOG_DEBUG_FMT("Commit on {}: {}", state->my_node_id, idx);

      // Examine all configurations that are followed by a globally committed
      // configuration.
      bool changed = false;

      while (true)
      {
        auto conf = configurations.begin();
        if (conf == configurations.end())
          break;

        auto next = std::next(conf);
        if (next == configurations.end())
          break;

        if (idx < next->idx)
          break;

        if (require_identity_for_reconfig)
        {
          assert(resharing_tracker);
          auto rr = resharing_tracker->find_reconfiguration(next->nodes);
          if (
            !rr.has_value() ||
            !resharing_tracker->have_resharing_result_for(rr.value(), idx))
          {
            LOG_TRACE_FMT(
              "Configurations: not switching to next configuration, resharing "
              "not completed yet.");
            break;
          }
        }

        if (reconfiguration_type == ReconfigurationType::ONE_TRANSACTION)
        {
          configurations.pop_front();
          changed = true;

          if (retired_node_cleanup && is_primary())
          {
            retired_node_cleanup->cleanup();
          }
        }
        else
        {
          if (
            !is_retired() &&
            conf->nodes.find(state->my_node_id) != conf->nodes.end() &&
            next->nodes.find(state->my_node_id) == next->nodes.end())
          {
            if (!is_retiring())
            {
              become_retiring();
            }
            else
            {
              become_retired(idx, kv::RetirementPhase::Committed);
            }
          }

          size_t num_trusted_nodes = num_trusted(*next);
          size_t num_retired_nodes = num_retired(*conf, *next);
          size_t num_required_retired_nodes =
            num_required_retirements(*conf, *next);
          if (
            num_trusted_nodes == next->nodes.size() &&
            num_retired_nodes == num_required_retired_nodes)
          {
            LOG_TRACE_FMT(
              "Configurations: all nodes trusted ({}) or retired ({}), "
              "switching to configuration #{}",
              num_trusted_nodes,
              num_retired_nodes,
              next->rid);

            if (
              is_learner() &&
              next->nodes.find(state->my_node_id) != next->nodes.end())
            {
              LOG_INFO_FMT(
                "Becoming follower {}: {}",
                state->my_node_id,
                state->current_view);
              leadership_state = kv::LeadershipState::Follower;
              membership_state = kv::MembershipState::Active;
            }

            for (auto& [nid, _] : next->nodes)
            {
              learner_nodes.erase(nid);
            }

            for (auto& [nid, _] : conf->nodes)
            {
              if (next->nodes.find(nid) == next->nodes.end())
              {
                retired_nodes.erase(nid);
              }
            }

            if (retired_node_cleanup && is_primary())
            {
              retired_node_cleanup->cleanup();
            }

            configurations.pop_front();
          }
          else
          {
            LOG_TRACE_FMT(
              "Configurations: not enough trusted or retired nodes for "
              "configuration #{} ({}/{} trusted, {}/{} retired)",
              next->rid,
              num_trusted_nodes,
              next->nodes.size(),
              num_retired_nodes,
              num_required_retired_nodes);
            if (
              node_client && !is_learner() &&
              (next->nodes.find(state->my_node_id) != next->nodes.end() ||
               (is_retiring() &&
                next->nodes.find(state->my_node_id) == next->nodes.end())))
            {
              schedule_submit_orc(
                node_client, state->my_node_id, next->rid, 2 * request_timeout);
            }
            break;
          }
        }
      }

      if (resharing_tracker)
      {
        resharing_tracker->compact(idx);
      }

      if (changed)
      {
        create_and_remove_node_state();
      }
    }