<commit_before>/*
 * Copyright (C) 2015-present ScyllaDB
 */

/*
 * This file is part of Scylla.
 *
 * Scylla is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Scylla is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "repair.hh"
#include "repair/row_level.hh"

#include "atomic_cell_hash.hh"
#include "dht/sharder.hh"
#include "streaming/stream_reason.hh"
#include "gms/inet_address.hh"
#include "utils/fb_utilities.hh"
#include "gms/gossiper.hh"
#include "service/priority_manager.hh"
#include "message/messaging_service.hh"
#include "sstables/sstables.hh"
#include "replica/database.hh"
#include "db/config.hh"
#include "hashers.hh"
#include "locator/network_topology_strategy.hh"
#include "utils/bit_cast.hh"
#include "service/migration_manager.hh"
#include "partition_range_compat.hh"

#include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/cxx11/any_of.hpp>
#include <boost/range/algorithm.hpp>
#include <boost/range/algorithm_ext.hpp>
#include <boost/range/adaptor/map.hpp>

#include <seastar/core/gate.hh>
#include <seastar/util/defer.hh>
#include <seastar/core/metrics_registration.hh>
#include <seastar/core/coroutine.hh>
#include <seastar/core/sleep.hh>

#include <cfloat>

#include "idl/partition_checksum.dist.hh"

logging::logger rlogger("repair");

void node_ops_info::check_abort() {
    if (abort) {
        auto msg = format("Node operation with ops_uuid={} is aborted", ops_uuid);
        rlogger.warn("{}", msg);
        throw std::runtime_error(msg);
    }
}

class node_ops_metrics {
public:
    node_ops_metrics() {
        namespace sm = seastar::metrics;
        auto ops_label_type = sm::label("ops");
        _metrics.add_group("node_ops", {
            sm::make_gauge("finished_percentage", [this] { return bootstrap_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("bootstrap")}),
            sm::make_gauge("finished_percentage", [this] { return replace_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("replace")}),
            sm::make_gauge("finished_percentage", [this] { return rebuild_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("rebuild")}),
            sm::make_gauge("finished_percentage", [this] { return decommission_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("decommission")}),
            sm::make_gauge("finished_percentage", [this] { return removenode_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("removenode")}),
            sm::make_gauge("finished_percentage", [this] { return repair_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("repair")}),
        });
    }
    uint64_t bootstrap_total_ranges{0};
    uint64_t bootstrap_finished_ranges{0};
    uint64_t replace_total_ranges{0};
    uint64_t replace_finished_ranges{0};
    uint64_t rebuild_total_ranges{0};
    uint64_t rebuild_finished_ranges{0};
    uint64_t decommission_total_ranges{0};
    uint64_t decommission_finished_ranges{0};
    uint64_t removenode_total_ranges{0};
    uint64_t removenode_finished_ranges{0};
    uint64_t repair_total_ranges_sum{0};
    uint64_t repair_finished_ranges_sum{0};
private:
    seastar::metrics::metric_groups _metrics;
    float bootstrap_finished_percentage() {
        return bootstrap_total_ranges == 0 ? 1 : float(bootstrap_finished_ranges) / float(bootstrap_total_ranges);
    }
    float replace_finished_percentage() {
        return replace_total_ranges == 0 ? 1 : float(replace_finished_ranges) / float(replace_total_ranges);
    }
    float rebuild_finished_percentage() {
        return rebuild_total_ranges == 0 ? 1 : float(rebuild_finished_ranges) / float(rebuild_total_ranges);
    }
    float decommission_finished_percentage() {
        return decommission_total_ranges == 0 ? 1 : float(decommission_finished_ranges) / float(decommission_total_ranges);
    }
    float removenode_finished_percentage() {
        return removenode_total_ranges == 0 ? 1 : float(removenode_finished_ranges) / float(removenode_total_ranges);
    }
    float repair_finished_percentage();
public:
    void init() {
        // Dummy function to call during startup to initialize the thread local
        // variable node_ops_metrics _node_ops_metrics below.
    }
};

static thread_local node_ops_metrics _node_ops_metrics;

template <typename T1, typename T2>
inline
static std::ostream& operator<<(std::ostream& os, const std::unordered_map<T1, T2>& v) {
    bool first = true;
    os << "{";
    for (auto&& elem : v) {
        if (!first) {
            os << ", ";
        } else {
            first = false;
        }
        os << elem.first << "=" << elem.second;
    }
    os << "}";
    return os;
}

std::ostream& operator<<(std::ostream& out, row_level_diff_detect_algorithm algo) {
    switch (algo) {
    case row_level_diff_detect_algorithm::send_full_set:
        return out << "send_full_set";
    case row_level_diff_detect_algorithm::send_full_set_rpc_stream:
        return out << "send_full_set_rpc_stream";
    };
    return out << "unknown";
}

static std::vector<sstring> list_column_families(const replica::database& db, const sstring& keyspace) {
    std::vector<sstring> ret;
    for (auto &&e : db.get_column_families_mapping()) {
        if (e.first.first == keyspace) {
            ret.push_back(e.first.second);
        }
    }
    return ret;
}

std::ostream& operator<<(std::ostream& os, const repair_uniq_id& x) {
    return os << format("[id={}, uuid={}]", x.id, x.uuid);
}

// Must run inside a seastar thread
static std::vector<utils::UUID> get_table_ids(const replica::database& db, const sstring& keyspace, const std::vector<sstring>& tables) {
    std::vector<utils::UUID> table_ids;
    table_ids.reserve(tables.size());
    for (auto& table : tables) {
        thread::maybe_yield();
        table_ids.push_back(db.find_uuid(keyspace, table));
    }
    return table_ids;
}

static std::vector<sstring> get_table_names(const replica::database& db, const std::vector<utils::UUID>& table_ids) {
    std::vector<sstring> table_names;
    table_names.reserve(table_ids.size());
    for (auto& table_id : table_ids) {
        table_names.push_back(db.find_column_family(table_id).schema()->cf_name());
    }
    return table_names;
}

template<typename Collection, typename T>
void remove_item(Collection& c, T& item) {
    auto it = std::find(c.begin(), c.end(), item);
    if (it != c.end()) {
        c.erase(it);
    }
}

// Return all of the neighbors with whom we share the provided range.
static std::vector<gms::inet_address> get_neighbors(replica::database& db,
        const sstring& ksname, query::range<dht::token> range,
        const std::vector<sstring>& data_centers,
        const std::vector<sstring>& hosts,
        const std::unordered_set<gms::inet_address>& ignore_nodes) {

    replica::keyspace& ks = db.find_keyspace(ksname);
    auto erm = ks.get_effective_replication_map();

    dht::token tok = range.end() ? range.end()->value() : dht::maximum_token();
    auto ret = erm->get_natural_endpoints(tok);
    remove_item(ret, utils::fb_utilities::get_broadcast_address());

    if (!data_centers.empty()) {
        auto dc_endpoints_map = db.get_token_metadata().get_topology().get_datacenter_endpoints();
        std::unordered_set<gms::inet_address> dc_endpoints;
        for (const sstring& dc : data_centers) {
            auto it = dc_endpoints_map.find(dc);
            if (it == dc_endpoints_map.end()) {
                std::vector<sstring> dcs;
                for (const auto& e : dc_endpoints_map) {
                    dcs.push_back(e.first);
                }
                throw std::runtime_error(fmt::format("Unknown data center '{}'. "
                        "Known data centers: {}", dc, dcs));
            }
            for (const auto& endpoint : it->second) {
                dc_endpoints.insert(endpoint);
            }
        }
        // We require, like Cassandra does, that the current host must also
        // be part of the repair
        if (!dc_endpoints.contains(utils::fb_utilities::get_broadcast_address())) {
            throw std::runtime_error("The current host must be part of the repair");
        }
        // The resulting list of nodes is the intersection of the nodes in the
        // listed data centers, and the (range-dependent) list of neighbors.
        std::unordered_set<gms::inet_address> neighbor_set(ret.begin(), ret.end());
        ret.clear();
        for (const auto& endpoint : dc_endpoints) {
            if (neighbor_set.contains(endpoint)) {
                ret.push_back(endpoint);
            }
        }
    } else if (!hosts.empty()) {
        bool found_me = false;
        std::unordered_set<gms::inet_address> neighbor_set(ret.begin(), ret.end());
        ret.clear();
        for (const sstring& host : hosts) {
            gms::inet_address endpoint;
            try {
                endpoint = gms::inet_address(host);
            } catch(...) {
                throw std::runtime_error(format("Unknown host specified: {}", host));
            }
            if (endpoint == utils::fb_utilities::get_broadcast_address()) {
                found_me = true;
            } else if (neighbor_set.contains(endpoint)) {
                ret.push_back(endpoint);
                // If same host is listed twice, don't add it again later
                neighbor_set.erase(endpoint);
            }
            // Nodes which aren't neighbors for this range are ignored.
            // This allows the user to give a list of "good" nodes, where
            // for each different range, only the subset of nodes actually
            // holding a replica of the given range is used. This,
            // however, means the user is never warned if one of the nodes
            // on the list isn't even part of the cluster.
        }
        // We require, like Cassandra does, that the current host must also
        // be listed on the "-hosts" option - even those we don't want it in
        // the returned list:
        if (!found_me) {
            throw std::runtime_error("The current host must be part of the repair");
        }
        if (ret.size() < 1) {
            auto me = utils::fb_utilities::get_broadcast_address();
            auto others = erm->get_natural_endpoints(tok);
            remove_item(others, me);
            throw std::runtime_error(fmt::format("Repair requires at least two "
                    "endpoints that are neighbors before it can continue, "
                    "the endpoint used for this repair is {}, other "
                    "available neighbors are {} but these neighbors were not "
                    "part of the supplied list of hosts to use during the "
                    "repair ({}).", me, others, hosts));
        }
    } else if (!ignore_nodes.empty()) {
        auto it = std::remove_if(ret.begin(), ret.end(), [&ignore_nodes] (const gms::inet_address& node) {
            return ignore_nodes.contains(node);
        });
        ret.erase(it, ret.end());
    }

    return boost::copy_range<std::vector<gms::inet_address>>(std::move(ret));

#if 0
    // Origin's ActiveRepairService.getNeighbors() also verifies that the
    // requested range fits into a local range
        StorageService ss = StorageService.instance;
        Map<Range<Token>, List<InetAddress>> replicaSets = ss.getRangeToAddressMap(keyspaceName);
        Range<Token> rangeSuperSet = null;
        for (Range<Token> range : ss.getLocalRanges(keyspaceName))
        {
            if (range.contains(toRepair))
            {
                rangeSuperSet = range;
                break;
            }
            else if (range.intersects(toRepair))
            {
                throw new IllegalArgumentException("Requested range intersects a local range but is not fully contained in one; this would lead to imprecise repair");
            }
        }
        if (rangeSuperSet == null || !replicaSets.containsKey(rangeSuperSet))
            return Collections.emptySet();

#endif
}

static future<std::list<gms::inet_address>> get_hosts_participating_in_repair(replica::database& db,
        const sstring& ksname,
        const dht::token_range_vector& ranges,
        const std::vector<sstring>& data_centers,
        const std::vector<sstring>& hosts,
        const std::unordered_set<gms::inet_address>& ignore_nodes) {

    std::unordered_set<gms::inet_address> participating_hosts;

    // Repair coordinator must participate in repair, but it is never
    // returned by get_neighbors - add it here
    participating_hosts.insert(utils::fb_utilities::get_broadcast_address());

    co_await do_for_each(ranges, [&] (const dht::token_range& range) {
        const auto nbs = get_neighbors(db, ksname, range, data_centers, hosts, ignore_nodes);
        for (const auto& nb : nbs) {
            participating_hosts.insert(nb);
        }
    });

    co_return std::list<gms::inet_address>(participating_hosts.begin(), participating_hosts.end());
}

static thread_local tracker* _the_tracker = nullptr;

static tracker& the_repair_tracker() {
    if (_the_tracker) {
        return *_the_tracker;
    } else {
        throw std::runtime_error("The repair tracker is not initialized yet");
    }
}

float node_ops_metrics::repair_finished_percentage() {
    if (_the_tracker) {
        return the_repair_tracker().report_progress(streaming::stream_reason::repair);
    }
    return 1;
}

tracker::tracker(size_t max_repair_memory)
    : _shutdown(false)
    , _range_parallelism_semaphore(std::max(size_t(1), size_t(max_repair_memory / max_repair_memory_per_range() / 4)),
            named_semaphore_exception_factory{"repair range parallelism"})
{
    auto nr = _range_parallelism_semaphore.available_units();
    rlogger.info("Setting max_repair_memory={}, max_repair_memory_per_range={}, max_repair_ranges_in_parallel={}",
        max_repair_memory, max_repair_memory_per_range(), nr);
    _the_tracker = this;
}

tracker::~tracker() {
    _the_tracker = nullptr;
}

void tracker::start(repair_uniq_id id) {
    _status[id.id] = repair_status::RUNNING;
}

void tracker::done(repair_uniq_id id, bool succeeded) {
    if (succeeded) {
        _status.erase(id.id);
    } else {
        _status[id.id] = repair_status::FAILED;
    }
    _done_cond.broadcast();
}
repair_status tracker::get(int id) const {
    if (id >= _next_repair_command) {
        throw std::runtime_error(format("unknown repair id {}", id));
    }
    auto it = _status.find(id);
    if (it == _status.end()) {
        return repair_status::SUCCESSFUL;
    } else {
        return it->second;
    }
}

future<repair_status> tracker::repair_await_completion(int id, std::chrono::steady_clock::time_point timeout) {
    return seastar::with_gate(_gate, [this, id, timeout] {
        if (id >= _next_repair_command) {
            return make_exception_future<repair_status>(std::runtime_error(format("unknown repair id {}", id)));
        }
        return repeat_until_value([this, id, timeout] {
            auto it = _status.find(id);
            if (it == _status.end()) {
                return make_ready_future<std::optional<repair_status>>(repair_status::SUCCESSFUL);
            } else {
                if (it->second == repair_status::FAILED) {
                    return make_ready_future<std::optional<repair_status>>(repair_status::FAILED);
                } else {
                    return _done_cond.wait(timeout).then([] {
                        return make_ready_future<std::optional<repair_status>>(std::nullopt);
                    }).handle_exception_type([] (condition_variable_timed_out&) {
                        return make_ready_future<std::optional<repair_status>>(repair_status::RUNNING);
                    });
                }
            }
        });
    });
}

repair_uniq_id tracker::next_repair_command() {
    return repair_uniq_id{_next_repair_command++, utils::make_random_uuid()};
}

future<> tracker::shutdown() {
    _shutdown.store(true, std::memory_order_relaxed);
    return _gate.close();
}

void tracker::check_in_shutdown() {
    if (_shutdown.load(std::memory_order_relaxed)) {
        throw std::runtime_error(format("Repair service is being shutdown"));
    }
}

void tracker::add_repair_info(int id, lw_shared_ptr<repair_info> ri) {
    _repairs.emplace(id, ri);
}

void tracker::remove_repair_info(int id) {
    _repairs.erase(id);
}

lw_shared_ptr<repair_info> tracker::get_repair_info(int id) {
    auto it = _repairs.find(id);
    if (it != _repairs.end()) {
        return it->second;
    }
    return {};
}

std::vector<int> tracker::get_active() const {
    std::vector<int> res;
    boost::push_back(res, _status | boost::adaptors::filtered([] (auto& x) {
        return x.second == repair_status::RUNNING;
    }) | boost::adaptors::map_keys);
    return res;
}

size_t tracker::nr_running_repair_jobs() {
    size_t count = 0;
    if (this_shard_id() != 0) {
        return count;
    }
    for (auto& x : _status) {
        auto& status = x.second;
        if (status == repair_status::RUNNING) {
            count++;
        }
    }
    return count;
}

void tracker::abort_all_repairs() {
    size_t count = nr_running_repair_jobs();
    for (auto& x : _repairs) {
        auto& ri = x.second;
        ri->abort();
    }
    rlogger.info0("Aborted {} repair job(s)", count);
}

void tracker::abort_repair_node_ops(utils::UUID ops_uuid) {
    for (auto& x : _repairs) {
        auto& ri = x.second;
        if (ri->ops_uuid() && ri->ops_uuid().value() == ops_uuid) {
            rlogger.info0("Aborted repair jobs for ops_uuid={}", ops_uuid);
            ri->abort();
        }
    }
}

float tracker::report_progress(streaming::stream_reason reason) {
    uint64_t nr_ranges_finished = 0;
    uint64_t nr_ranges_total = 0;
    for (auto& x : _repairs) {
        auto& ri = x.second;
        if (ri->reason == reason) {
            nr_ranges_total += ri->nr_ranges_total;
            nr_ranges_finished += ri->nr_ranges_finished;
        }
    }
    return nr_ranges_total == 0 ? 1 : float(nr_ranges_finished) / float(nr_ranges_total);
}

named_semaphore& tracker::range_parallelism_semaphore() {
    return _range_parallelism_semaphore;
}

future<> tracker::run(repair_uniq_id id, std::function<void ()> func) {
    return seastar::with_gate(_gate, [this, id, func =std::move(func)] {
        start(id);
        return seastar::async([func = std::move(func)] { func(); }).then([this, id] {
            rlogger.info("repair id {} completed successfully", id);
            done(id, true);
        }).handle_exception([this, id] (std::exception_ptr ep) {
            done(id, false);
            return make_exception_future(std::move(ep));
        });
    });
}

void check_in_shutdown() {
    the_repair_tracker().check_in_shutdown();
}

future<uint64_t> estimate_partitions(seastar::sharded<replica::database>& db, const sstring& keyspace,
        const sstring& cf, const dht::token_range& range) {
    return db.map_reduce0(
        [keyspace, cf, range] (auto& db) {
            // FIXME: column_family should have a method to estimate the number of
            // partitions (and of course it should use cardinality estimation bitmaps,
            // not trivial sum). We shouldn't have this ugly code here...
            // FIXME: If sstables are shared, they will be accounted more than
            // once. However, shared sstables should exist for a short-time only.
            auto sstables = db.find_column_family(keyspace, cf).get_sstables();
            return boost::accumulate(*sstables, uint64_t(0),
                [&range] (uint64_t x, auto&& sst) { return x + sst->estimated_keys_for_range(range); });
        },
        uint64_t(0),
        std::plus<uint64_t>()
    );
}

static
const dht::sharder&
get_sharder_for_tables(seastar::sharded<replica::database>& db, const sstring& keyspace, const std::vector<utils::UUID>& table_ids) {
    schema_ptr last_s;
    for (size_t idx = 0 ; idx < table_ids.size(); idx++) {
        schema_ptr s;
        try {
            s = db.local().find_column_family(table_ids[idx]).schema();
        } catch(...) {
            throw replica::no_such_column_family(keyspace, table_ids[idx]);
        }
        if (last_s && last_s->get_sharder() != s->get_sharder()) {
            throw std::runtime_error(
                    format("All tables repaired together have to have the same sharding logic. "
                        "Different sharding logic found: {} (for table {}) and {} (for table {})",
                        last_s->get_sharder(), last_s->cf_name(),
                        s->get_sharder(), s->cf_name()));
        }
        last_s = std::move(s);
    }
    if (!last_s) {
        throw std::runtime_error(format("Failed to find sharder for keyspace={}, tables={}, no table in this keyspace",
                keyspace, table_ids));
    }
    return last_s->get_sharder();
}

repair_info::repair_info(repair_service& repair,
    const sstring& keyspace_,
    const dht::token_range_vector& ranges_,
    std::vector<utils::UUID> table_ids_,
    repair_uniq_id id_,
    const std::vector<sstring>& data_centers_,
    const std::vector<sstring>& hosts_,
    const std::unordered_set<gms::inet_address>& ignore_nodes_,
    streaming::stream_reason reason_,
    std::optional<utils::UUID> ops_uuid,
    bool hints_batchlog_flushed)
    : rs(repair)
    , db(repair.get_db())
    , messaging(repair.get_messaging().container())
    , sys_dist_ks(repair.get_sys_dist_ks())
    , view_update_generator(repair.get_view_update_generator())
    , mm(repair.get_migration_manager())
    , gossiper(repair.get_gossiper())
    , sharder(get_sharder_for_tables(db, keyspace_, table_ids_))
    , keyspace(keyspace_)
    , ranges(ranges_)
    , cfs(get_table_names(db.local(), table_ids_))
    , table_ids(std::move(table_ids_))
    , id(id_)
    , shard(this_shard_id())
    , data_centers(data_centers_)
    , hosts(hosts_)
    , ignore_nodes(ignore_nodes_)
    , reason(reason_)
    , total_rf(db.local().find_keyspace(keyspace).get_effective_replication_map()->get_replication_factor())
    , nr_ranges_total(ranges.size())
    , _ops_uuid(std::move(ops_uuid))
    , _hints_batchlog_flushed(std::move(hints_batchlog_flushed)) {
}

void repair_info::check_failed_ranges() {
    rlogger.info("repair id {} on shard {} stats: repair_reason={}, keyspace={}, tables={}, ranges_nr={}, {}",
        id, shard, reason, keyspace, table_names(), ranges.size(), _stats.get_stats());
    if (nr_failed_ranges) {
        rlogger.warn("repair id {} on shard {} failed - {} out of {} ranges failed", id, shard, nr_failed_ranges, ranges.size());
        throw std::runtime_error(format("repair id {} on shard {} failed to repair {} out of {} ranges", id, shard, nr_failed_ranges, ranges.size()));
    } else {
        if (dropped_tables.size()) {
            rlogger.warn("repair id {} on shard {} completed successfully, keyspace={}, ignoring dropped tables={}", id, shard, keyspace, dropped_tables);
        } else {
            rlogger.info("repair id {} on shard {} completed successfully, keyspace={}", id, shard, keyspace);
        }
    }
}

void repair_info::abort() {
    aborted = true;
}

void repair_info::check_in_abort() {
    if (aborted) {
        throw std::runtime_error(format("repair id {} is aborted on shard {}", id, shard));
    }
}

repair_neighbors repair_info::get_repair_neighbors(const dht::token_range& range) {
    return neighbors.empty() ?
        repair_neighbors(get_neighbors(db.local(), keyspace, range, data_centers, hosts, ignore_nodes)) :
        neighbors[range];
}

// Repair a single local range, multiple column families.
// Comparable to RepairSession in Origin
future<> repair_info::repair_range(const dht::token_range& range) {
    check_in_shutdown();
    check_in_abort();
    ranges_index++;
    repair_neighbors neighbors = get_repair_neighbors(range);
    return do_with(std::move(neighbors.all), std::move(neighbors.mandatory), [this, range] (auto& neighbors, auto& mandatory_neighbors) {
      auto live_neighbors = boost::copy_range<std::vector<gms::inet_address>>(neighbors |
                    boost::adaptors::filtered([this] (const gms::inet_address& node) { return gossiper.is_alive(node); }));
      for (auto& node : mandatory_neighbors) {
           auto it = std::find(live_neighbors.begin(), live_neighbors.end(), node);
           if (it == live_neighbors.end()) {
                nr_failed_ranges++;
                auto status = format("failed: mandatory neighbor={} is not alive", node);
                rlogger.error("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
                    ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
                abort();
                return make_exception_future<>(std::runtime_error(format("Repair mandatory neighbor={} is not alive, keyspace={}, mandatory_neighbors={}",
                    node, keyspace, mandatory_neighbors)));
           }
      }
      if (live_neighbors.size() != neighbors.size()) {
            nr_failed_ranges++;
            auto status = live_neighbors.empty() ? "skipped" : "partial";
            rlogger.warn("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
            if (live_neighbors.empty()) {
                return make_ready_future<>();
            }
            neighbors.swap(live_neighbors);
      }
      if (neighbors.empty()) {
            auto status = "skipped_no_followers";
            rlogger.warn("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
            return make_ready_future<>();
      }
      rlogger.info("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors);
      return mm.sync_schema(db.local(), neighbors).then([this, &neighbors, range] {
        return do_for_each(table_ids.begin(), table_ids.end(), [this, &neighbors, range] (utils::UUID table_id) {
            sstring cf;
            try {
                cf = db.local().find_column_family(table_id).schema()->cf_name();
            } catch (replica::no_such_column_family&) {
                return make_ready_future<>();
            }
            // Row level repair
            if (dropped_tables.contains(cf)) {
                return make_ready_future<>();
            }
            return repair_cf_range_row_level(*this, cf, table_id, range, neighbors).handle_exception_type([this, cf] (replica::no_such_column_family&) mutable {
                dropped_tables.insert(cf);
                return make_ready_future<>();
            }).handle_exception([this] (std::exception_ptr ep) mutable {
                nr_failed_ranges++;
                return make_exception_future<>(std::move(ep));
            });
        });
      });
    });
}

static dht::token_range_vector get_primary_ranges_for_endpoint(
        replica::database& db, sstring keyspace, gms::inet_address ep) {
    return db.find_keyspace(keyspace).get_effective_replication_map()->get_primary_ranges(ep);
}

static dht::token_range_vector get_primary_ranges(
        replica::database& db, sstring keyspace) {
    return get_primary_ranges_for_endpoint(db, keyspace,
            utils::fb_utilities::get_broadcast_address());
}

// get_primary_ranges_within_dc() is similar to get_primary_ranges(),
// but instead of each range being assigned just one primary owner
// across the entire cluster, here each range is assigned a primary
// owner in each of the clusters.
static dht::token_range_vector get_primary_ranges_within_dc(
        replica::database& db, sstring keyspace) {
    return db.find_keyspace(keyspace).get_effective_replication_map()->get_primary_ranges_within_dc(utils::fb_utilities::get_broadcast_address());
}

static sstring get_local_dc() {
    return locator::i_endpoint_snitch::get_local_snitch_ptr()->get_datacenter(
            utils::fb_utilities::get_broadcast_address());
}

void repair_stats::add(const repair_stats& o) {
    round_nr += o.round_nr;
    round_nr_fast_path_already_synced += o.round_nr_fast_path_already_synced;
    round_nr_fast_path_same_combined_hashes += o.round_nr_fast_path_same_combined_hashes;
    round_nr_slow_path += o.round_nr_slow_path;
    rpc_call_nr += o.rpc_call_nr;
    tx_hashes_nr += o.tx_hashes_nr;
    rx_hashes_nr += o.rx_hashes_nr;
    tx_row_nr += o.tx_row_nr;
    rx_row_nr += o.rx_row_nr;
    tx_row_bytes += o.tx_row_bytes;
    rx_row_bytes += o.rx_row_bytes;
    auto add_map = [] (auto& target, auto& src) {
         for (const auto& [k, v] : src) {
             target[k] += v;
         }
    };
    add_map(row_from_disk_bytes, o.row_from_disk_bytes);
    add_map(row_from_disk_nr, o.row_from_disk_nr);
    add_map(tx_row_nr_peer, o.tx_row_nr_peer);
    add_map(rx_row_nr_peer, o.rx_row_nr_peer);
}

sstring repair_stats::get_stats() {
    std::map<gms::inet_address, float> row_from_disk_bytes_per_sec;
    std::map<gms::inet_address, float> row_from_disk_rows_per_sec;
    auto duration = std::chrono::duration_cast<std::chrono::duration<float>>(lowres_clock::now() - start_time).count();
    for (auto& x : row_from_disk_bytes) {
        if (std::fabs(duration) > FLT_EPSILON) {
            row_from_disk_bytes_per_sec[x.first] = x.second / duration / 1024 / 1024;
        } else {
            row_from_disk_bytes_per_sec[x.first] = 0;
        }
    }
    for (auto& x : row_from_disk_nr) {
        if (std::fabs(duration) > FLT_EPSILON) {
            row_from_disk_rows_per_sec[x.first] = x.second / duration;
        } else {
            row_from_disk_rows_per_sec[x.first] = 0;
        }
    }
    return format("round_nr={}, round_nr_fast_path_already_synced={}, round_nr_fast_path_same_combined_hashes={}, round_nr_slow_path={}, rpc_call_nr={}, tx_hashes_nr={}, rx_hashes_nr={}, duration={} seconds, tx_row_nr={}, rx_row_nr={}, tx_row_bytes={}, rx_row_bytes={}, row_from_disk_bytes={}, row_from_disk_nr={}, row_from_disk_bytes_per_sec={} MiB/s, row_from_disk_rows_per_sec={} Rows/s, tx_row_nr_peer={}, rx_row_nr_peer={}",
            round_nr,
            round_nr_fast_path_already_synced,
            round_nr_fast_path_same_combined_hashes,
            round_nr_slow_path,
            rpc_call_nr,
            tx_hashes_nr,
            rx_hashes_nr,
            duration,
            tx_row_nr,
            rx_row_nr,
            tx_row_bytes,
            rx_row_bytes,
            row_from_disk_bytes,
            row_from_disk_nr,
            row_from_disk_bytes_per_sec,
            row_from_disk_rows_per_sec,
            tx_row_nr_peer,
            rx_row_nr_peer);
}

struct repair_options {
    // If primary_range is true, we should perform repair only on this node's
    // primary ranges. The default of false means perform repair on all ranges
    // held by the node. primary_range=true is useful if the user plans to
    // repair all nodes.
    bool primary_range = false;
    // If ranges is not empty, it overrides the repair's default heuristics
    // for determining the list of ranges to repair. In particular, "ranges"
    // overrides the setting of "primary_range".
    dht::token_range_vector ranges;
    // If start_token and end_token are set, they define a range which is
    // intersected with the ranges actually held by this node to decide what
    // to repair.
    sstring start_token;
    sstring end_token;
    // column_families is the list of column families to repair in the given
    // keyspace. If this list is empty (the default), all the column families
    // in this keyspace are repaired
    std::vector<sstring> column_families;
    // hosts specifies the list of known good hosts to repair with this host
    // (note that this host is required to also be on this list). For each
    // range repaired, only the relevant subset of the hosts (holding a
    // replica of this range) is used.
    std::vector<sstring> hosts;
    // The ignore_nodes specifies the list of nodes to ignore in this repair,
    // e.g., the user knows a node is down and wants to run repair without this
    // specific node.
    std::vector<sstring> ignore_nodes;
    // data_centers is used to restrict the repair to the local data center.
    // The node starting the repair must be in the data center; Issuing a
    // repair to a data center other than the named one returns an error.
    std::vector<sstring> data_centers;

    repair_options(std::unordered_map<sstring, sstring> options) {
        bool_opt(primary_range, options, PRIMARY_RANGE_KEY);
        ranges_opt(ranges, options, RANGES_KEY);
        list_opt(column_families, options, COLUMNFAMILIES_KEY);
        list_opt(hosts, options, HOSTS_KEY);
        list_opt(ignore_nodes, options, IGNORE_NODES_KEY);
        list_opt(data_centers, options, DATACENTERS_KEY);
        // We currently do not support incremental repair. We could probably
        // ignore this option as it is just an optimization, but for now,
        // let's make it an error.
        bool incremental = false;
        bool_opt(incremental, options, INCREMENTAL_KEY);
        if (incremental) {
            throw std::runtime_error("unsupported incremental repair");
        }
        // We do not currently support the distinction between "parallel" and
        // "sequential" repair, and operate the same for both.
        // We don't currently support "dc parallel" parallelism.
        int parallelism = PARALLEL;
        int_opt(parallelism, options, PARALLELISM_KEY);
        if (parallelism != PARALLEL && parallelism != SEQUENTIAL) {
            throw std::runtime_error(format("unsupported repair parallelism: {}", parallelism));
        }
        string_opt(start_token, options, START_TOKEN);
        string_opt(end_token, options, END_TOKEN);

        bool trace = false;
        bool_opt(trace, options, TRACE_KEY);
        if (trace) {
            throw std::runtime_error("unsupported trace");
        }
        // Consume, ignore.
        int job_threads;
        int_opt(job_threads, options, JOB_THREADS_KEY);

        // The parsing code above removed from the map options we have parsed.
        // If anything is left there in the end, it's an unsupported option.
        if (!options.empty()) {
            throw std::runtime_error(format("unsupported repair options: {}",
                    options));
        }
    }

    static constexpr const char* PRIMARY_RANGE_KEY = "primaryRange";
    static constexpr const char* PARALLELISM_KEY = "parallelism";
    static constexpr const char* INCREMENTAL_KEY = "incremental";
    static constexpr const char* JOB_THREADS_KEY = "jobThreads";
    static constexpr const char* RANGES_KEY = "ranges";
    static constexpr const char* COLUMNFAMILIES_KEY = "columnFamilies";
    static constexpr const char* DATACENTERS_KEY = "dataCenters";
    static constexpr const char* HOSTS_KEY = "hosts";
    static constexpr const char* IGNORE_NODES_KEY = "ignore_nodes";
    static constexpr const char* TRACE_KEY = "trace";
    static constexpr const char* START_TOKEN = "startToken";
    static constexpr const char* END_TOKEN = "endToken";

    // Settings of "parallelism" option. Numbers must match Cassandra's
    // RepairParallelism enum, which is used by the caller.
    enum repair_parallelism {
        SEQUENTIAL=0, PARALLEL=1, DATACENTER_AWARE=2
    };

private:
    static void bool_opt(bool& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            // Same parsing as Boolean.parseBoolean does:
            if (boost::algorithm::iequals(it->second, "true")) {
                var = true;
            } else {
                var = false;
            }
            options.erase(it);
        }
    }

    static void int_opt(int& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            errno = 0;
            var = strtol(it->second.c_str(), nullptr, 10);
            if (errno) {
                throw(std::runtime_error(format("cannot parse integer: '{}'", it->second)));
            }
            options.erase(it);
        }
    }

    static void string_opt(sstring& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            var = it->second;
            options.erase(it);
        }
    }

    // A range is expressed as start_token:end token and multiple ranges can
    // be given as comma separated ranges(e.g. aaa:bbb,ccc:ddd).
    static void ranges_opt(dht::token_range_vector& var,
            std::unordered_map<sstring, sstring>& options,
                        const sstring& key) {
        auto it = options.find(key);
        if (it == options.end()) {
            return;
        }
        std::vector<sstring> range_strings;
        boost::split(range_strings, it->second, boost::algorithm::is_any_of(","));
        for (auto range : range_strings) {
            std::vector<sstring> token_strings;
            boost::split(token_strings, range, boost::algorithm::is_any_of(":"));
            if (token_strings.size() != 2) {
                throw(std::runtime_error("range must have two components "
                        "separated by ':', got '" + range + "'"));
            }
            auto tok_start = dht::token::from_sstring(token_strings[0]);
            auto tok_end = dht::token::from_sstring(token_strings[1]);
            auto rng = wrapping_range<dht::token>(
                    ::range<dht::token>::bound(tok_start, false),
                    ::range<dht::token>::bound(tok_end, true));
            ::compat::unwrap_into(std::move(rng), dht::token_comparator(), [&] (dht::token_range&& x) {
                var.push_back(std::move(x));
            });
        }
        options.erase(it);
    }

    // A comma-separate list of strings
    static void list_opt(std::vector<sstring>& var,
            std::unordered_map<sstring, sstring>& options,
                        const sstring& key) {
        auto it = options.find(key);
        if (it == options.end()) {
            return;
        }
        std::vector<sstring> range_strings;
        boost::split(var, it->second, boost::algorithm::is_any_of(","));
        options.erase(it);
    }
};


static future<> do_repair_ranges(lw_shared_ptr<repair_info> ri) {
    // repair all the ranges in limited parallelism
    return parallel_for_each(ri->ranges, [ri] (auto&& range) {
        return with_semaphore(the_repair_tracker().range_parallelism_semaphore(), 1, [ri, &range] {
            return ri->repair_range(range).then([ri] {
                if (ri->reason == streaming::stream_reason::bootstrap) {
                    _node_ops_metrics.bootstrap_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::replace) {
                    _node_ops_metrics.replace_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::rebuild) {
                    _node_ops_metrics.rebuild_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::decommission) {
                    _node_ops_metrics.decommission_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::removenode) {
                    _node_ops_metrics.removenode_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::repair) {
                    _node_ops_metrics.repair_finished_ranges_sum++;
                    ri->nr_ranges_finished++;
                }
            });
        });
    });
}

// repair_ranges repairs a list of token ranges, each assumed to be a token
// range for which this node holds a replica, and, importantly, each range
// is assumed to be a indivisible in the sense that all the tokens in has the
// same nodes as replicas.
static future<> repair_ranges(lw_shared_ptr<repair_info> ri) {
    the_repair_tracker().add_repair_info(ri->id.id, ri);
    return do_repair_ranges(ri).then([ri] {
        ri->check_failed_ranges();
        the_repair_tracker().remove_repair_info(ri->id.id);
        return make_ready_future<>();
    }).handle_exception([ri] (std::exception_ptr eptr) {
        the_repair_tracker().remove_repair_info(ri->id.id);
        return make_exception_future<>(std::move(eptr));<commit_msg>repair: use repair_info to get to the repair tracker<commit_after>/*
 * Copyright (C) 2015-present ScyllaDB
 */

/*
 * This file is part of Scylla.
 *
 * Scylla is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Affero General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * Scylla is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Scylla.  If not, see <http://www.gnu.org/licenses/>.
 */

#include "repair.hh"
#include "repair/row_level.hh"

#include "atomic_cell_hash.hh"
#include "dht/sharder.hh"
#include "streaming/stream_reason.hh"
#include "gms/inet_address.hh"
#include "utils/fb_utilities.hh"
#include "gms/gossiper.hh"
#include "service/priority_manager.hh"
#include "message/messaging_service.hh"
#include "sstables/sstables.hh"
#include "replica/database.hh"
#include "db/config.hh"
#include "hashers.hh"
#include "locator/network_topology_strategy.hh"
#include "utils/bit_cast.hh"
#include "service/migration_manager.hh"
#include "partition_range_compat.hh"

#include <boost/algorithm/string/predicate.hpp>
#include <boost/algorithm/string/split.hpp>
#include <boost/algorithm/string/classification.hpp>
#include <boost/algorithm/cxx11/any_of.hpp>
#include <boost/range/algorithm.hpp>
#include <boost/range/algorithm_ext.hpp>
#include <boost/range/adaptor/map.hpp>

#include <seastar/core/gate.hh>
#include <seastar/util/defer.hh>
#include <seastar/core/metrics_registration.hh>
#include <seastar/core/coroutine.hh>
#include <seastar/core/sleep.hh>

#include <cfloat>

#include "idl/partition_checksum.dist.hh"

logging::logger rlogger("repair");

void node_ops_info::check_abort() {
    if (abort) {
        auto msg = format("Node operation with ops_uuid={} is aborted", ops_uuid);
        rlogger.warn("{}", msg);
        throw std::runtime_error(msg);
    }
}

class node_ops_metrics {
public:
    node_ops_metrics() {
        namespace sm = seastar::metrics;
        auto ops_label_type = sm::label("ops");
        _metrics.add_group("node_ops", {
            sm::make_gauge("finished_percentage", [this] { return bootstrap_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("bootstrap")}),
            sm::make_gauge("finished_percentage", [this] { return replace_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("replace")}),
            sm::make_gauge("finished_percentage", [this] { return rebuild_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("rebuild")}),
            sm::make_gauge("finished_percentage", [this] { return decommission_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("decommission")}),
            sm::make_gauge("finished_percentage", [this] { return removenode_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("removenode")}),
            sm::make_gauge("finished_percentage", [this] { return repair_finished_percentage(); },
                    sm::description("Finished percentage of node operation on this shard"), {ops_label_type("repair")}),
        });
    }
    uint64_t bootstrap_total_ranges{0};
    uint64_t bootstrap_finished_ranges{0};
    uint64_t replace_total_ranges{0};
    uint64_t replace_finished_ranges{0};
    uint64_t rebuild_total_ranges{0};
    uint64_t rebuild_finished_ranges{0};
    uint64_t decommission_total_ranges{0};
    uint64_t decommission_finished_ranges{0};
    uint64_t removenode_total_ranges{0};
    uint64_t removenode_finished_ranges{0};
    uint64_t repair_total_ranges_sum{0};
    uint64_t repair_finished_ranges_sum{0};
private:
    seastar::metrics::metric_groups _metrics;
    float bootstrap_finished_percentage() {
        return bootstrap_total_ranges == 0 ? 1 : float(bootstrap_finished_ranges) / float(bootstrap_total_ranges);
    }
    float replace_finished_percentage() {
        return replace_total_ranges == 0 ? 1 : float(replace_finished_ranges) / float(replace_total_ranges);
    }
    float rebuild_finished_percentage() {
        return rebuild_total_ranges == 0 ? 1 : float(rebuild_finished_ranges) / float(rebuild_total_ranges);
    }
    float decommission_finished_percentage() {
        return decommission_total_ranges == 0 ? 1 : float(decommission_finished_ranges) / float(decommission_total_ranges);
    }
    float removenode_finished_percentage() {
        return removenode_total_ranges == 0 ? 1 : float(removenode_finished_ranges) / float(removenode_total_ranges);
    }
    float repair_finished_percentage();
public:
    void init() {
        // Dummy function to call during startup to initialize the thread local
        // variable node_ops_metrics _node_ops_metrics below.
    }
};

static thread_local node_ops_metrics _node_ops_metrics;

template <typename T1, typename T2>
inline
static std::ostream& operator<<(std::ostream& os, const std::unordered_map<T1, T2>& v) {
    bool first = true;
    os << "{";
    for (auto&& elem : v) {
        if (!first) {
            os << ", ";
        } else {
            first = false;
        }
        os << elem.first << "=" << elem.second;
    }
    os << "}";
    return os;
}

std::ostream& operator<<(std::ostream& out, row_level_diff_detect_algorithm algo) {
    switch (algo) {
    case row_level_diff_detect_algorithm::send_full_set:
        return out << "send_full_set";
    case row_level_diff_detect_algorithm::send_full_set_rpc_stream:
        return out << "send_full_set_rpc_stream";
    };
    return out << "unknown";
}

static std::vector<sstring> list_column_families(const replica::database& db, const sstring& keyspace) {
    std::vector<sstring> ret;
    for (auto &&e : db.get_column_families_mapping()) {
        if (e.first.first == keyspace) {
            ret.push_back(e.first.second);
        }
    }
    return ret;
}

std::ostream& operator<<(std::ostream& os, const repair_uniq_id& x) {
    return os << format("[id={}, uuid={}]", x.id, x.uuid);
}

// Must run inside a seastar thread
static std::vector<utils::UUID> get_table_ids(const replica::database& db, const sstring& keyspace, const std::vector<sstring>& tables) {
    std::vector<utils::UUID> table_ids;
    table_ids.reserve(tables.size());
    for (auto& table : tables) {
        thread::maybe_yield();
        table_ids.push_back(db.find_uuid(keyspace, table));
    }
    return table_ids;
}

static std::vector<sstring> get_table_names(const replica::database& db, const std::vector<utils::UUID>& table_ids) {
    std::vector<sstring> table_names;
    table_names.reserve(table_ids.size());
    for (auto& table_id : table_ids) {
        table_names.push_back(db.find_column_family(table_id).schema()->cf_name());
    }
    return table_names;
}

template<typename Collection, typename T>
void remove_item(Collection& c, T& item) {
    auto it = std::find(c.begin(), c.end(), item);
    if (it != c.end()) {
        c.erase(it);
    }
}

// Return all of the neighbors with whom we share the provided range.
static std::vector<gms::inet_address> get_neighbors(replica::database& db,
        const sstring& ksname, query::range<dht::token> range,
        const std::vector<sstring>& data_centers,
        const std::vector<sstring>& hosts,
        const std::unordered_set<gms::inet_address>& ignore_nodes) {

    replica::keyspace& ks = db.find_keyspace(ksname);
    auto erm = ks.get_effective_replication_map();

    dht::token tok = range.end() ? range.end()->value() : dht::maximum_token();
    auto ret = erm->get_natural_endpoints(tok);
    remove_item(ret, utils::fb_utilities::get_broadcast_address());

    if (!data_centers.empty()) {
        auto dc_endpoints_map = db.get_token_metadata().get_topology().get_datacenter_endpoints();
        std::unordered_set<gms::inet_address> dc_endpoints;
        for (const sstring& dc : data_centers) {
            auto it = dc_endpoints_map.find(dc);
            if (it == dc_endpoints_map.end()) {
                std::vector<sstring> dcs;
                for (const auto& e : dc_endpoints_map) {
                    dcs.push_back(e.first);
                }
                throw std::runtime_error(fmt::format("Unknown data center '{}'. "
                        "Known data centers: {}", dc, dcs));
            }
            for (const auto& endpoint : it->second) {
                dc_endpoints.insert(endpoint);
            }
        }
        // We require, like Cassandra does, that the current host must also
        // be part of the repair
        if (!dc_endpoints.contains(utils::fb_utilities::get_broadcast_address())) {
            throw std::runtime_error("The current host must be part of the repair");
        }
        // The resulting list of nodes is the intersection of the nodes in the
        // listed data centers, and the (range-dependent) list of neighbors.
        std::unordered_set<gms::inet_address> neighbor_set(ret.begin(), ret.end());
        ret.clear();
        for (const auto& endpoint : dc_endpoints) {
            if (neighbor_set.contains(endpoint)) {
                ret.push_back(endpoint);
            }
        }
    } else if (!hosts.empty()) {
        bool found_me = false;
        std::unordered_set<gms::inet_address> neighbor_set(ret.begin(), ret.end());
        ret.clear();
        for (const sstring& host : hosts) {
            gms::inet_address endpoint;
            try {
                endpoint = gms::inet_address(host);
            } catch(...) {
                throw std::runtime_error(format("Unknown host specified: {}", host));
            }
            if (endpoint == utils::fb_utilities::get_broadcast_address()) {
                found_me = true;
            } else if (neighbor_set.contains(endpoint)) {
                ret.push_back(endpoint);
                // If same host is listed twice, don't add it again later
                neighbor_set.erase(endpoint);
            }
            // Nodes which aren't neighbors for this range are ignored.
            // This allows the user to give a list of "good" nodes, where
            // for each different range, only the subset of nodes actually
            // holding a replica of the given range is used. This,
            // however, means the user is never warned if one of the nodes
            // on the list isn't even part of the cluster.
        }
        // We require, like Cassandra does, that the current host must also
        // be listed on the "-hosts" option - even those we don't want it in
        // the returned list:
        if (!found_me) {
            throw std::runtime_error("The current host must be part of the repair");
        }
        if (ret.size() < 1) {
            auto me = utils::fb_utilities::get_broadcast_address();
            auto others = erm->get_natural_endpoints(tok);
            remove_item(others, me);
            throw std::runtime_error(fmt::format("Repair requires at least two "
                    "endpoints that are neighbors before it can continue, "
                    "the endpoint used for this repair is {}, other "
                    "available neighbors are {} but these neighbors were not "
                    "part of the supplied list of hosts to use during the "
                    "repair ({}).", me, others, hosts));
        }
    } else if (!ignore_nodes.empty()) {
        auto it = std::remove_if(ret.begin(), ret.end(), [&ignore_nodes] (const gms::inet_address& node) {
            return ignore_nodes.contains(node);
        });
        ret.erase(it, ret.end());
    }

    return boost::copy_range<std::vector<gms::inet_address>>(std::move(ret));

#if 0
    // Origin's ActiveRepairService.getNeighbors() also verifies that the
    // requested range fits into a local range
        StorageService ss = StorageService.instance;
        Map<Range<Token>, List<InetAddress>> replicaSets = ss.getRangeToAddressMap(keyspaceName);
        Range<Token> rangeSuperSet = null;
        for (Range<Token> range : ss.getLocalRanges(keyspaceName))
        {
            if (range.contains(toRepair))
            {
                rangeSuperSet = range;
                break;
            }
            else if (range.intersects(toRepair))
            {
                throw new IllegalArgumentException("Requested range intersects a local range but is not fully contained in one; this would lead to imprecise repair");
            }
        }
        if (rangeSuperSet == null || !replicaSets.containsKey(rangeSuperSet))
            return Collections.emptySet();

#endif
}

static future<std::list<gms::inet_address>> get_hosts_participating_in_repair(replica::database& db,
        const sstring& ksname,
        const dht::token_range_vector& ranges,
        const std::vector<sstring>& data_centers,
        const std::vector<sstring>& hosts,
        const std::unordered_set<gms::inet_address>& ignore_nodes) {

    std::unordered_set<gms::inet_address> participating_hosts;

    // Repair coordinator must participate in repair, but it is never
    // returned by get_neighbors - add it here
    participating_hosts.insert(utils::fb_utilities::get_broadcast_address());

    co_await do_for_each(ranges, [&] (const dht::token_range& range) {
        const auto nbs = get_neighbors(db, ksname, range, data_centers, hosts, ignore_nodes);
        for (const auto& nb : nbs) {
            participating_hosts.insert(nb);
        }
    });

    co_return std::list<gms::inet_address>(participating_hosts.begin(), participating_hosts.end());
}

static thread_local tracker* _the_tracker = nullptr;

static tracker& the_repair_tracker() {
    if (_the_tracker) {
        return *_the_tracker;
    } else {
        throw std::runtime_error("The repair tracker is not initialized yet");
    }
}

float node_ops_metrics::repair_finished_percentage() {
    if (_the_tracker) {
        return the_repair_tracker().report_progress(streaming::stream_reason::repair);
    }
    return 1;
}

tracker::tracker(size_t max_repair_memory)
    : _shutdown(false)
    , _range_parallelism_semaphore(std::max(size_t(1), size_t(max_repair_memory / max_repair_memory_per_range() / 4)),
            named_semaphore_exception_factory{"repair range parallelism"})
{
    auto nr = _range_parallelism_semaphore.available_units();
    rlogger.info("Setting max_repair_memory={}, max_repair_memory_per_range={}, max_repair_ranges_in_parallel={}",
        max_repair_memory, max_repair_memory_per_range(), nr);
    _the_tracker = this;
}

tracker::~tracker() {
    _the_tracker = nullptr;
}

void tracker::start(repair_uniq_id id) {
    _status[id.id] = repair_status::RUNNING;
}

void tracker::done(repair_uniq_id id, bool succeeded) {
    if (succeeded) {
        _status.erase(id.id);
    } else {
        _status[id.id] = repair_status::FAILED;
    }
    _done_cond.broadcast();
}
repair_status tracker::get(int id) const {
    if (id >= _next_repair_command) {
        throw std::runtime_error(format("unknown repair id {}", id));
    }
    auto it = _status.find(id);
    if (it == _status.end()) {
        return repair_status::SUCCESSFUL;
    } else {
        return it->second;
    }
}

future<repair_status> tracker::repair_await_completion(int id, std::chrono::steady_clock::time_point timeout) {
    return seastar::with_gate(_gate, [this, id, timeout] {
        if (id >= _next_repair_command) {
            return make_exception_future<repair_status>(std::runtime_error(format("unknown repair id {}", id)));
        }
        return repeat_until_value([this, id, timeout] {
            auto it = _status.find(id);
            if (it == _status.end()) {
                return make_ready_future<std::optional<repair_status>>(repair_status::SUCCESSFUL);
            } else {
                if (it->second == repair_status::FAILED) {
                    return make_ready_future<std::optional<repair_status>>(repair_status::FAILED);
                } else {
                    return _done_cond.wait(timeout).then([] {
                        return make_ready_future<std::optional<repair_status>>(std::nullopt);
                    }).handle_exception_type([] (condition_variable_timed_out&) {
                        return make_ready_future<std::optional<repair_status>>(repair_status::RUNNING);
                    });
                }
            }
        });
    });
}

repair_uniq_id tracker::next_repair_command() {
    return repair_uniq_id{_next_repair_command++, utils::make_random_uuid()};
}

future<> tracker::shutdown() {
    _shutdown.store(true, std::memory_order_relaxed);
    return _gate.close();
}

void tracker::check_in_shutdown() {
    if (_shutdown.load(std::memory_order_relaxed)) {
        throw std::runtime_error(format("Repair service is being shutdown"));
    }
}

void tracker::add_repair_info(int id, lw_shared_ptr<repair_info> ri) {
    _repairs.emplace(id, ri);
}

void tracker::remove_repair_info(int id) {
    _repairs.erase(id);
}

lw_shared_ptr<repair_info> tracker::get_repair_info(int id) {
    auto it = _repairs.find(id);
    if (it != _repairs.end()) {
        return it->second;
    }
    return {};
}

std::vector<int> tracker::get_active() const {
    std::vector<int> res;
    boost::push_back(res, _status | boost::adaptors::filtered([] (auto& x) {
        return x.second == repair_status::RUNNING;
    }) | boost::adaptors::map_keys);
    return res;
}

size_t tracker::nr_running_repair_jobs() {
    size_t count = 0;
    if (this_shard_id() != 0) {
        return count;
    }
    for (auto& x : _status) {
        auto& status = x.second;
        if (status == repair_status::RUNNING) {
            count++;
        }
    }
    return count;
}

void tracker::abort_all_repairs() {
    size_t count = nr_running_repair_jobs();
    for (auto& x : _repairs) {
        auto& ri = x.second;
        ri->abort();
    }
    rlogger.info0("Aborted {} repair job(s)", count);
}

void tracker::abort_repair_node_ops(utils::UUID ops_uuid) {
    for (auto& x : _repairs) {
        auto& ri = x.second;
        if (ri->ops_uuid() && ri->ops_uuid().value() == ops_uuid) {
            rlogger.info0("Aborted repair jobs for ops_uuid={}", ops_uuid);
            ri->abort();
        }
    }
}

float tracker::report_progress(streaming::stream_reason reason) {
    uint64_t nr_ranges_finished = 0;
    uint64_t nr_ranges_total = 0;
    for (auto& x : _repairs) {
        auto& ri = x.second;
        if (ri->reason == reason) {
            nr_ranges_total += ri->nr_ranges_total;
            nr_ranges_finished += ri->nr_ranges_finished;
        }
    }
    return nr_ranges_total == 0 ? 1 : float(nr_ranges_finished) / float(nr_ranges_total);
}

named_semaphore& tracker::range_parallelism_semaphore() {
    return _range_parallelism_semaphore;
}

future<> tracker::run(repair_uniq_id id, std::function<void ()> func) {
    return seastar::with_gate(_gate, [this, id, func =std::move(func)] {
        start(id);
        return seastar::async([func = std::move(func)] { func(); }).then([this, id] {
            rlogger.info("repair id {} completed successfully", id);
            done(id, true);
        }).handle_exception([this, id] (std::exception_ptr ep) {
            done(id, false);
            return make_exception_future(std::move(ep));
        });
    });
}

void check_in_shutdown() {
    the_repair_tracker().check_in_shutdown();
}

future<uint64_t> estimate_partitions(seastar::sharded<replica::database>& db, const sstring& keyspace,
        const sstring& cf, const dht::token_range& range) {
    return db.map_reduce0(
        [keyspace, cf, range] (auto& db) {
            // FIXME: column_family should have a method to estimate the number of
            // partitions (and of course it should use cardinality estimation bitmaps,
            // not trivial sum). We shouldn't have this ugly code here...
            // FIXME: If sstables are shared, they will be accounted more than
            // once. However, shared sstables should exist for a short-time only.
            auto sstables = db.find_column_family(keyspace, cf).get_sstables();
            return boost::accumulate(*sstables, uint64_t(0),
                [&range] (uint64_t x, auto&& sst) { return x + sst->estimated_keys_for_range(range); });
        },
        uint64_t(0),
        std::plus<uint64_t>()
    );
}

static
const dht::sharder&
get_sharder_for_tables(seastar::sharded<replica::database>& db, const sstring& keyspace, const std::vector<utils::UUID>& table_ids) {
    schema_ptr last_s;
    for (size_t idx = 0 ; idx < table_ids.size(); idx++) {
        schema_ptr s;
        try {
            s = db.local().find_column_family(table_ids[idx]).schema();
        } catch(...) {
            throw replica::no_such_column_family(keyspace, table_ids[idx]);
        }
        if (last_s && last_s->get_sharder() != s->get_sharder()) {
            throw std::runtime_error(
                    format("All tables repaired together have to have the same sharding logic. "
                        "Different sharding logic found: {} (for table {}) and {} (for table {})",
                        last_s->get_sharder(), last_s->cf_name(),
                        s->get_sharder(), s->cf_name()));
        }
        last_s = std::move(s);
    }
    if (!last_s) {
        throw std::runtime_error(format("Failed to find sharder for keyspace={}, tables={}, no table in this keyspace",
                keyspace, table_ids));
    }
    return last_s->get_sharder();
}

repair_info::repair_info(repair_service& repair,
    const sstring& keyspace_,
    const dht::token_range_vector& ranges_,
    std::vector<utils::UUID> table_ids_,
    repair_uniq_id id_,
    const std::vector<sstring>& data_centers_,
    const std::vector<sstring>& hosts_,
    const std::unordered_set<gms::inet_address>& ignore_nodes_,
    streaming::stream_reason reason_,
    std::optional<utils::UUID> ops_uuid,
    bool hints_batchlog_flushed)
    : rs(repair)
    , db(repair.get_db())
    , messaging(repair.get_messaging().container())
    , sys_dist_ks(repair.get_sys_dist_ks())
    , view_update_generator(repair.get_view_update_generator())
    , mm(repair.get_migration_manager())
    , gossiper(repair.get_gossiper())
    , sharder(get_sharder_for_tables(db, keyspace_, table_ids_))
    , keyspace(keyspace_)
    , ranges(ranges_)
    , cfs(get_table_names(db.local(), table_ids_))
    , table_ids(std::move(table_ids_))
    , id(id_)
    , shard(this_shard_id())
    , data_centers(data_centers_)
    , hosts(hosts_)
    , ignore_nodes(ignore_nodes_)
    , reason(reason_)
    , total_rf(db.local().find_keyspace(keyspace).get_effective_replication_map()->get_replication_factor())
    , nr_ranges_total(ranges.size())
    , _ops_uuid(std::move(ops_uuid))
    , _hints_batchlog_flushed(std::move(hints_batchlog_flushed)) {
}

void repair_info::check_failed_ranges() {
    rlogger.info("repair id {} on shard {} stats: repair_reason={}, keyspace={}, tables={}, ranges_nr={}, {}",
        id, shard, reason, keyspace, table_names(), ranges.size(), _stats.get_stats());
    if (nr_failed_ranges) {
        rlogger.warn("repair id {} on shard {} failed - {} out of {} ranges failed", id, shard, nr_failed_ranges, ranges.size());
        throw std::runtime_error(format("repair id {} on shard {} failed to repair {} out of {} ranges", id, shard, nr_failed_ranges, ranges.size()));
    } else {
        if (dropped_tables.size()) {
            rlogger.warn("repair id {} on shard {} completed successfully, keyspace={}, ignoring dropped tables={}", id, shard, keyspace, dropped_tables);
        } else {
            rlogger.info("repair id {} on shard {} completed successfully, keyspace={}", id, shard, keyspace);
        }
    }
}

void repair_info::abort() {
    aborted = true;
}

void repair_info::check_in_abort() {
    if (aborted) {
        throw std::runtime_error(format("repair id {} is aborted on shard {}", id, shard));
    }
}

repair_neighbors repair_info::get_repair_neighbors(const dht::token_range& range) {
    return neighbors.empty() ?
        repair_neighbors(get_neighbors(db.local(), keyspace, range, data_centers, hosts, ignore_nodes)) :
        neighbors[range];
}

// Repair a single local range, multiple column families.
// Comparable to RepairSession in Origin
future<> repair_info::repair_range(const dht::token_range& range) {
    check_in_shutdown();
    check_in_abort();
    ranges_index++;
    repair_neighbors neighbors = get_repair_neighbors(range);
    return do_with(std::move(neighbors.all), std::move(neighbors.mandatory), [this, range] (auto& neighbors, auto& mandatory_neighbors) {
      auto live_neighbors = boost::copy_range<std::vector<gms::inet_address>>(neighbors |
                    boost::adaptors::filtered([this] (const gms::inet_address& node) { return gossiper.is_alive(node); }));
      for (auto& node : mandatory_neighbors) {
           auto it = std::find(live_neighbors.begin(), live_neighbors.end(), node);
           if (it == live_neighbors.end()) {
                nr_failed_ranges++;
                auto status = format("failed: mandatory neighbor={} is not alive", node);
                rlogger.error("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
                    ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
                abort();
                return make_exception_future<>(std::runtime_error(format("Repair mandatory neighbor={} is not alive, keyspace={}, mandatory_neighbors={}",
                    node, keyspace, mandatory_neighbors)));
           }
      }
      if (live_neighbors.size() != neighbors.size()) {
            nr_failed_ranges++;
            auto status = live_neighbors.empty() ? "skipped" : "partial";
            rlogger.warn("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
            if (live_neighbors.empty()) {
                return make_ready_future<>();
            }
            neighbors.swap(live_neighbors);
      }
      if (neighbors.empty()) {
            auto status = "skipped_no_followers";
            rlogger.warn("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}, status={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors, status);
            return make_ready_future<>();
      }
      rlogger.info("Repair {} out of {} ranges, id={}, shard={}, keyspace={}, table={}, range={}, peers={}, live_peers={}",
            ranges_index, ranges.size(), id, shard, keyspace, table_names(), range, neighbors, live_neighbors);
      return mm.sync_schema(db.local(), neighbors).then([this, &neighbors, range] {
        return do_for_each(table_ids.begin(), table_ids.end(), [this, &neighbors, range] (utils::UUID table_id) {
            sstring cf;
            try {
                cf = db.local().find_column_family(table_id).schema()->cf_name();
            } catch (replica::no_such_column_family&) {
                return make_ready_future<>();
            }
            // Row level repair
            if (dropped_tables.contains(cf)) {
                return make_ready_future<>();
            }
            return repair_cf_range_row_level(*this, cf, table_id, range, neighbors).handle_exception_type([this, cf] (replica::no_such_column_family&) mutable {
                dropped_tables.insert(cf);
                return make_ready_future<>();
            }).handle_exception([this] (std::exception_ptr ep) mutable {
                nr_failed_ranges++;
                return make_exception_future<>(std::move(ep));
            });
        });
      });
    });
}

static dht::token_range_vector get_primary_ranges_for_endpoint(
        replica::database& db, sstring keyspace, gms::inet_address ep) {
    return db.find_keyspace(keyspace).get_effective_replication_map()->get_primary_ranges(ep);
}

static dht::token_range_vector get_primary_ranges(
        replica::database& db, sstring keyspace) {
    return get_primary_ranges_for_endpoint(db, keyspace,
            utils::fb_utilities::get_broadcast_address());
}

// get_primary_ranges_within_dc() is similar to get_primary_ranges(),
// but instead of each range being assigned just one primary owner
// across the entire cluster, here each range is assigned a primary
// owner in each of the clusters.
static dht::token_range_vector get_primary_ranges_within_dc(
        replica::database& db, sstring keyspace) {
    return db.find_keyspace(keyspace).get_effective_replication_map()->get_primary_ranges_within_dc(utils::fb_utilities::get_broadcast_address());
}

static sstring get_local_dc() {
    return locator::i_endpoint_snitch::get_local_snitch_ptr()->get_datacenter(
            utils::fb_utilities::get_broadcast_address());
}

void repair_stats::add(const repair_stats& o) {
    round_nr += o.round_nr;
    round_nr_fast_path_already_synced += o.round_nr_fast_path_already_synced;
    round_nr_fast_path_same_combined_hashes += o.round_nr_fast_path_same_combined_hashes;
    round_nr_slow_path += o.round_nr_slow_path;
    rpc_call_nr += o.rpc_call_nr;
    tx_hashes_nr += o.tx_hashes_nr;
    rx_hashes_nr += o.rx_hashes_nr;
    tx_row_nr += o.tx_row_nr;
    rx_row_nr += o.rx_row_nr;
    tx_row_bytes += o.tx_row_bytes;
    rx_row_bytes += o.rx_row_bytes;
    auto add_map = [] (auto& target, auto& src) {
         for (const auto& [k, v] : src) {
             target[k] += v;
         }
    };
    add_map(row_from_disk_bytes, o.row_from_disk_bytes);
    add_map(row_from_disk_nr, o.row_from_disk_nr);
    add_map(tx_row_nr_peer, o.tx_row_nr_peer);
    add_map(rx_row_nr_peer, o.rx_row_nr_peer);
}

sstring repair_stats::get_stats() {
    std::map<gms::inet_address, float> row_from_disk_bytes_per_sec;
    std::map<gms::inet_address, float> row_from_disk_rows_per_sec;
    auto duration = std::chrono::duration_cast<std::chrono::duration<float>>(lowres_clock::now() - start_time).count();
    for (auto& x : row_from_disk_bytes) {
        if (std::fabs(duration) > FLT_EPSILON) {
            row_from_disk_bytes_per_sec[x.first] = x.second / duration / 1024 / 1024;
        } else {
            row_from_disk_bytes_per_sec[x.first] = 0;
        }
    }
    for (auto& x : row_from_disk_nr) {
        if (std::fabs(duration) > FLT_EPSILON) {
            row_from_disk_rows_per_sec[x.first] = x.second / duration;
        } else {
            row_from_disk_rows_per_sec[x.first] = 0;
        }
    }
    return format("round_nr={}, round_nr_fast_path_already_synced={}, round_nr_fast_path_same_combined_hashes={}, round_nr_slow_path={}, rpc_call_nr={}, tx_hashes_nr={}, rx_hashes_nr={}, duration={} seconds, tx_row_nr={}, rx_row_nr={}, tx_row_bytes={}, rx_row_bytes={}, row_from_disk_bytes={}, row_from_disk_nr={}, row_from_disk_bytes_per_sec={} MiB/s, row_from_disk_rows_per_sec={} Rows/s, tx_row_nr_peer={}, rx_row_nr_peer={}",
            round_nr,
            round_nr_fast_path_already_synced,
            round_nr_fast_path_same_combined_hashes,
            round_nr_slow_path,
            rpc_call_nr,
            tx_hashes_nr,
            rx_hashes_nr,
            duration,
            tx_row_nr,
            rx_row_nr,
            tx_row_bytes,
            rx_row_bytes,
            row_from_disk_bytes,
            row_from_disk_nr,
            row_from_disk_bytes_per_sec,
            row_from_disk_rows_per_sec,
            tx_row_nr_peer,
            rx_row_nr_peer);
}

struct repair_options {
    // If primary_range is true, we should perform repair only on this node's
    // primary ranges. The default of false means perform repair on all ranges
    // held by the node. primary_range=true is useful if the user plans to
    // repair all nodes.
    bool primary_range = false;
    // If ranges is not empty, it overrides the repair's default heuristics
    // for determining the list of ranges to repair. In particular, "ranges"
    // overrides the setting of "primary_range".
    dht::token_range_vector ranges;
    // If start_token and end_token are set, they define a range which is
    // intersected with the ranges actually held by this node to decide what
    // to repair.
    sstring start_token;
    sstring end_token;
    // column_families is the list of column families to repair in the given
    // keyspace. If this list is empty (the default), all the column families
    // in this keyspace are repaired
    std::vector<sstring> column_families;
    // hosts specifies the list of known good hosts to repair with this host
    // (note that this host is required to also be on this list). For each
    // range repaired, only the relevant subset of the hosts (holding a
    // replica of this range) is used.
    std::vector<sstring> hosts;
    // The ignore_nodes specifies the list of nodes to ignore in this repair,
    // e.g., the user knows a node is down and wants to run repair without this
    // specific node.
    std::vector<sstring> ignore_nodes;
    // data_centers is used to restrict the repair to the local data center.
    // The node starting the repair must be in the data center; Issuing a
    // repair to a data center other than the named one returns an error.
    std::vector<sstring> data_centers;

    repair_options(std::unordered_map<sstring, sstring> options) {
        bool_opt(primary_range, options, PRIMARY_RANGE_KEY);
        ranges_opt(ranges, options, RANGES_KEY);
        list_opt(column_families, options, COLUMNFAMILIES_KEY);
        list_opt(hosts, options, HOSTS_KEY);
        list_opt(ignore_nodes, options, IGNORE_NODES_KEY);
        list_opt(data_centers, options, DATACENTERS_KEY);
        // We currently do not support incremental repair. We could probably
        // ignore this option as it is just an optimization, but for now,
        // let's make it an error.
        bool incremental = false;
        bool_opt(incremental, options, INCREMENTAL_KEY);
        if (incremental) {
            throw std::runtime_error("unsupported incremental repair");
        }
        // We do not currently support the distinction between "parallel" and
        // "sequential" repair, and operate the same for both.
        // We don't currently support "dc parallel" parallelism.
        int parallelism = PARALLEL;
        int_opt(parallelism, options, PARALLELISM_KEY);
        if (parallelism != PARALLEL && parallelism != SEQUENTIAL) {
            throw std::runtime_error(format("unsupported repair parallelism: {}", parallelism));
        }
        string_opt(start_token, options, START_TOKEN);
        string_opt(end_token, options, END_TOKEN);

        bool trace = false;
        bool_opt(trace, options, TRACE_KEY);
        if (trace) {
            throw std::runtime_error("unsupported trace");
        }
        // Consume, ignore.
        int job_threads;
        int_opt(job_threads, options, JOB_THREADS_KEY);

        // The parsing code above removed from the map options we have parsed.
        // If anything is left there in the end, it's an unsupported option.
        if (!options.empty()) {
            throw std::runtime_error(format("unsupported repair options: {}",
                    options));
        }
    }

    static constexpr const char* PRIMARY_RANGE_KEY = "primaryRange";
    static constexpr const char* PARALLELISM_KEY = "parallelism";
    static constexpr const char* INCREMENTAL_KEY = "incremental";
    static constexpr const char* JOB_THREADS_KEY = "jobThreads";
    static constexpr const char* RANGES_KEY = "ranges";
    static constexpr const char* COLUMNFAMILIES_KEY = "columnFamilies";
    static constexpr const char* DATACENTERS_KEY = "dataCenters";
    static constexpr const char* HOSTS_KEY = "hosts";
    static constexpr const char* IGNORE_NODES_KEY = "ignore_nodes";
    static constexpr const char* TRACE_KEY = "trace";
    static constexpr const char* START_TOKEN = "startToken";
    static constexpr const char* END_TOKEN = "endToken";

    // Settings of "parallelism" option. Numbers must match Cassandra's
    // RepairParallelism enum, which is used by the caller.
    enum repair_parallelism {
        SEQUENTIAL=0, PARALLEL=1, DATACENTER_AWARE=2
    };

private:
    static void bool_opt(bool& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            // Same parsing as Boolean.parseBoolean does:
            if (boost::algorithm::iequals(it->second, "true")) {
                var = true;
            } else {
                var = false;
            }
            options.erase(it);
        }
    }

    static void int_opt(int& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            errno = 0;
            var = strtol(it->second.c_str(), nullptr, 10);
            if (errno) {
                throw(std::runtime_error(format("cannot parse integer: '{}'", it->second)));
            }
            options.erase(it);
        }
    }

    static void string_opt(sstring& var,
            std::unordered_map<sstring, sstring>& options,
            const sstring& key) {
        auto it = options.find(key);
        if (it != options.end()) {
            var = it->second;
            options.erase(it);
        }
    }

    // A range is expressed as start_token:end token and multiple ranges can
    // be given as comma separated ranges(e.g. aaa:bbb,ccc:ddd).
    static void ranges_opt(dht::token_range_vector& var,
            std::unordered_map<sstring, sstring>& options,
                        const sstring& key) {
        auto it = options.find(key);
        if (it == options.end()) {
            return;
        }
        std::vector<sstring> range_strings;
        boost::split(range_strings, it->second, boost::algorithm::is_any_of(","));
        for (auto range : range_strings) {
            std::vector<sstring> token_strings;
            boost::split(token_strings, range, boost::algorithm::is_any_of(":"));
            if (token_strings.size() != 2) {
                throw(std::runtime_error("range must have two components "
                        "separated by ':', got '" + range + "'"));
            }
            auto tok_start = dht::token::from_sstring(token_strings[0]);
            auto tok_end = dht::token::from_sstring(token_strings[1]);
            auto rng = wrapping_range<dht::token>(
                    ::range<dht::token>::bound(tok_start, false),
                    ::range<dht::token>::bound(tok_end, true));
            ::compat::unwrap_into(std::move(rng), dht::token_comparator(), [&] (dht::token_range&& x) {
                var.push_back(std::move(x));
            });
        }
        options.erase(it);
    }

    // A comma-separate list of strings
    static void list_opt(std::vector<sstring>& var,
            std::unordered_map<sstring, sstring>& options,
                        const sstring& key) {
        auto it = options.find(key);
        if (it == options.end()) {
            return;
        }
        std::vector<sstring> range_strings;
        boost::split(var, it->second, boost::algorithm::is_any_of(","));
        options.erase(it);
    }
};


static future<> do_repair_ranges(lw_shared_ptr<repair_info> ri) {
    // repair all the ranges in limited parallelism
    return parallel_for_each(ri->ranges, [ri] (auto&& range) {
        return with_semaphore(ri->rs.repair_tracker().range_parallelism_semaphore(), 1, [ri, &range] {
            return ri->repair_range(range).then([ri] {
                if (ri->reason == streaming::stream_reason::bootstrap) {
                    _node_ops_metrics.bootstrap_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::replace) {
                    _node_ops_metrics.replace_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::rebuild) {
                    _node_ops_metrics.rebuild_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::decommission) {
                    _node_ops_metrics.decommission_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::removenode) {
                    _node_ops_metrics.removenode_finished_ranges++;
                } else if (ri->reason == streaming::stream_reason::repair) {
                    _node_ops_metrics.repair_finished_ranges_sum++;
                    ri->nr_ranges_finished++;
                }
            });
        });
    });
}

// repair_ranges repairs a list of token ranges, each assumed to be a token
// range for which this node holds a replica, and, importantly, each range
// is assumed to be a indivisible in the sense that all the tokens in has the
// same nodes as replicas.
static future<> repair_ranges(lw_shared_ptr<repair_info> ri) {
    ri->rs.repair_tracker().add_repair_info(ri->id.id, ri);
    return do_repair_ranges(ri).then([ri] {
        ri->check_failed_ranges();
        ri->rs.repair_tracker().remove_repair_info(ri->id.id);
        return make_ready_future<>();
    }).handle_exception([ri] (std::exception_ptr eptr) {
        ri->rs.repair_tracker().remove_repair_info(ri->id.id);
        return make_exception_future<>(std::move(eptr));
{'fraction_non_alphanumeric': 0.12239734894824986, 'fraction_numerical': 0.002565533951753709, 'mean_word_length': 2.9333800513898622, 'pattern_counts': {'":': 2, '<': 283, '<?xml version=': 0, '>': 333, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '182271', 'n_tokens_mistral': 26478, 'n_tokens_neox': 25918, 'n_words': 5817}