Add replica-aware migration planning

/*

  Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain

  Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany

  This file is part of GekkoFS.

  SPDX-License-Identifier: GPL-3.0-or-later

*/

#ifndef GKFS_RPC_REPLICA_MIGRATOR_HPP

#define GKFS_RPC_REPLICA_MIGRATOR_HPP

#include <common/rpc/distributor.hpp>

#include <algorithm>

#include <stdexcept>

#include <string>

#include <vector>

namespace gkfs::rpc {

struct replica_migration_job {

    std::string path;

    chunkid_t chunk_id{};

    int source_copy{};

    int target_copy{};

    host_t source_node{};

    host_t target_node{};

};

struct logical_chunk_replica_plan {

    std::vector<host_t> old_hosts;

    std::vector<host_t> new_hosts;

    std::vector<replica_migration_job> jobs;

};

/**

 * Build a migration plan for one logical chunk and all configured copies.

 *

 * The plan does not perform I/O. A target already occupied by an old copy is

 * treated as materialized, even when its copy index changes. At most one job

 * is emitted for a target host, so one source read can be reused by the

 * executor for duplicate logical targets.

 */

inline logical_chunk_replica_plan

plan_replica_migration(const Distributor& old_distributor,

                       const Distributor& new_distributor,

                       const std::string& path, const chunkid_t chunk_id,

                       const int replica_count) {

    if(replica_count < 0) {

        throw std::invalid_argument("replica count must not be negative");

    }

    logical_chunk_replica_plan plan;

    plan.old_hosts.reserve(static_cast<std::size_t>(replica_count) + 1);

    plan.new_hosts.reserve(static_cast<std::size_t>(replica_count) + 1);

    for(int copy = 0; copy <= replica_count; ++copy) {

        plan.old_hosts.push_back(

                old_distributor.locate_data(path, chunk_id, copy));

        plan.new_hosts.push_back(

                new_distributor.locate_data(path, chunk_id, copy));

    }

    for(std::size_t target_copy = 0; target_copy < plan.new_hosts.size();

        ++target_copy) {

        const auto target_node = plan.new_hosts[target_copy];

        if(plan.old_hosts[target_copy] == target_node) {

            continue;

        }

        const auto existing_copy = std::find(plan.old_hosts.begin(),

                                             plan.old_hosts.end(), target_node);

        if(existing_copy != plan.old_hosts.end()) {

            continue;

        }

        const auto duplicate_target =

                std::find_if(plan.jobs.begin(), plan.jobs.end(),

                             [target_node](const auto& job) {

                                 return job.target_node == target_node;

                             });

        if(duplicate_target != plan.jobs.end()) {

            continue;

        }

        auto source_copy = std::find_first_of(

                plan.old_hosts.begin(), plan.old_hosts.end(),

                plan.new_hosts.begin(), plan.new_hosts.end());

        if(source_copy == plan.old_hosts.end() || *source_copy == target_node) {

            source_copy = plan.old_hosts.begin();

        }

        const auto source_copy_index = static_cast<int>(

                std::distance(plan.old_hosts.begin(), source_copy));

        plan.jobs.push_back({path, chunk_id, source_copy_index,

                             static_cast<int>(target_copy), *source_copy,

                             target_node});

    }

    return plan;

}

} // namespace gkfs::rpc

#endif // GKFS_RPC_REPLICA_MIGRATOR_HPP

 No newline at end of file

    ${CMAKE_CURRENT_LIST_DIR}/test_common_path.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_distributor.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_placement.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_replica_migrator.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_repair_queue.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_helpers.cpp

    ${CMAKE_CURRENT_LIST_DIR}/test_random_slicing_distributor.cpp

/*

  Copyright 2018-2025, Barcelona Supercomputing Center (BSC), Spain

  Copyright 2015-2025, Johannes Gutenberg Universitaet Mainz, Germany

  This file is part of GekkoFS.

  SPDX-License-Identifier: GPL-3.0-or-later

*/

#include <common/rpc/replica_migrator.hpp>

#include <catch2/catch_all.hpp>

#include <algorithm>

#include <set>

namespace {

class fixed_distributor final : public gkfs::rpc::Distributor {

public:

    explicit fixed_distributor(std::vector<gkfs::rpc::host_t> hosts)

        : hosts_(std::move(hosts)) {}

    gkfs::rpc::host_t

    localhost() const override {

        return hosts_.front();

    }

    gkfs::rpc::host_t

    locate_data(const std::string&, const gkfs::rpc::chunkid_t&, int copy) const

            override {

        return hosts_.at(static_cast<std::size_t>(copy));

    }

    unsigned int

    hosts_size() const override {

        return static_cast<unsigned int>(hosts_.size());

    }

    void

    hosts_size(unsigned int) override {}

    gkfs::rpc::host_t

    locate_data(const std::string&, const gkfs::rpc::chunkid_t&,

                unsigned int, int copy) override {

        return hosts_.at(static_cast<std::size_t>(copy));

    }

    gkfs::rpc::host_t

    locate_file_metadata(const std::string&, int copy) const override {

        return hosts_.at(static_cast<std::size_t>(copy));

    }

    std::vector<gkfs::rpc::host_t>

    locate_directory_metadata() const override {

        return hosts_;

    }

private:

    std::vector<gkfs::rpc::host_t> hosts_;

};

} // namespace

TEST_CASE("replica migration plan preserves logical copy placement",

          "[replication][mutation]") {

    const fixed_distributor old_distributor({1, 2, 3});

    const fixed_distributor new_distributor({4, 5, 6});

    const auto plan = gkfs::rpc::plan_replica_migration(

            old_distributor, new_distributor, "/file", 7, 2);

    REQUIRE(plan.old_hosts == std::vector<gkfs::rpc::host_t>{1, 2, 3});

    REQUIRE(plan.new_hosts == std::vector<gkfs::rpc::host_t>{4, 5, 6});

    REQUIRE(plan.jobs.size() == 3);

    for(const auto& job : plan.jobs) {

        REQUIRE(job.path == "/file");

        REQUIRE(job.chunk_id == 7);

        REQUIRE(job.source_copy == 0);

        REQUIRE(job.source_node == 1);

    }

    REQUIRE(plan.jobs[0].target_copy == 0);

    REQUIRE(plan.jobs[1].target_copy == 1);

    REQUIRE(plan.jobs[2].target_copy == 2);

}

TEST_CASE("replica migration plan reuses existing hosts and deduplicates targets",

          "[replication][mutation]") {

    const fixed_distributor old_distributor({1, 2, 3});

    const fixed_distributor new_distributor({2, 4, 4});

    const auto plan = gkfs::rpc::plan_replica_migration(

            old_distributor, new_distributor, "/file", 9, 2);

    REQUIRE(plan.jobs.size() == 1);

    REQUIRE(plan.jobs.front().source_copy == 1);

    REQUIRE(plan.jobs.front().target_copy == 1);

    REQUIRE(plan.jobs.front().source_node == 2);

    REQUIRE(plan.jobs.front().target_node == 4);

    REQUIRE(std::set<gkfs::rpc::host_t>{plan.jobs.front().target_node}.size() ==

            1);

}

TEST_CASE("replica migration plan rejects negative replica counts",

          "[replication][mutation]") {

    const fixed_distributor distributor({1});

    REQUIRE_THROWS_WITH(

            gkfs::rpc::plan_replica_migration(

                    distributor, distributor, "/file", 0, -1),

            "replica count must not be negative");

}

TEST_CASE("replica migration plan falls back to primary without survivors",

          "[replication][mutation]") {

    const fixed_distributor old_distributor({1, 2, 3});

    const fixed_distributor new_distributor({4, 5, 6});

    const auto plan = gkfs::rpc::plan_replica_migration(

            old_distributor, new_distributor, "/file", 9, 2);

    REQUIRE(plan.jobs.size() == 3);

    for(const auto& job : plan.jobs) {

        REQUIRE(job.source_copy == 0);

        REQUIRE(job.source_node == 1);

    }

}

 No newline at end of file