Commit 9593db78 authored by Ramon Nou's avatar Ramon Nou
Browse files

Guard replica migration plan execution

parent ea3eaf1d

include/common/rpc/replica_migrator.hpp

+50 −7
Changes for include/common/rpc/replica_migrator.hpp: 50 added lines, 7 removed lines.
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@ 


#include <algorithm>

#include <cstdint>

#include <functional>

#include <stdexcept>

#include <string>

#include <vector>
@@ -39,6 +40,13 @@ struct logical_chunk_replica_plan { 
    uint64_t new_placement_generation{};

};



enum class replica_migration_plan_error {

    none,

    stale_generation,

    invalid_layout,

    execution_failed,

};



/**

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

 *
@@ -108,15 +116,17 @@ plan_replica_migration(const Distributor& old_distributor, 
    return plan;

}



inline bool

validate_replica_migration_plan(const logical_chunk_replica_plan& plan,

inline replica_migration_plan_error

classify_replica_migration_plan(const logical_chunk_replica_plan& plan,

                                const uint64_t expected_old_generation,

                                const uint64_t expected_new_generation) {

    if(plan.old_placement_generation != expected_old_generation ||

       plan.new_placement_generation != expected_new_generation ||

       expected_old_generation >= expected_new_generation ||

       plan.old_hosts.empty() || plan.new_hosts.empty()) {

        return false;

       expected_old_generation >= expected_new_generation) {

        return replica_migration_plan_error::stale_generation;

    }

    if(plan.old_hosts.empty() || plan.new_hosts.empty()) {

        return replica_migration_plan_error::invalid_layout;

    }



    std::vector<int> target_copies;
@@ -131,11 +141,44 @@ validate_replica_migration_plan(const logical_chunk_replica_plan& plan, 
           plan.new_hosts[job.target_copy] != job.target_node ||

           std::find(target_copies.begin(), target_copies.end(),

                     job.target_copy) != target_copies.end()) {

            return false;

            return replica_migration_plan_error::invalid_layout;

        }

        target_copies.push_back(job.target_copy);

    }

    return true;

    return replica_migration_plan_error::none;

}



inline bool

validate_replica_migration_plan(const logical_chunk_replica_plan& plan,

                                const uint64_t expected_old_generation,

                                const uint64_t expected_new_generation) {

    return classify_replica_migration_plan(plan, expected_old_generation,

                                           expected_new_generation) ==

           replica_migration_plan_error::none;

}



using replica_migration_handler =

        std::function<bool(const replica_migration_job&)>;



inline replica_migration_plan_error

execute_replica_migration_plan(const logical_chunk_replica_plan& plan,

                               const uint64_t expected_old_generation,

                               const uint64_t expected_new_generation,

                               const replica_migration_handler& handler) {

    const auto validation = classify_replica_migration_plan(

            plan, expected_old_generation, expected_new_generation);

    if(validation != replica_migration_plan_error::none || !handler) {

        return validation == replica_migration_plan_error::none

                       ? replica_migration_plan_error::invalid_layout

                       : validation;

    }



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

        if(!handler(job)) {

            return replica_migration_plan_error::execution_failed;

        }

    }

    return replica_migration_plan_error::none;

}



} // namespace gkfs::rpc

tests/unit/test_replica_migrator.cpp

+34 −0
Changes for tests/unit/test_replica_migrator.cpp: 34 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -114,11 +114,15 @@ TEST_CASE("replica migration plan validates placement generations and hosts", 


    REQUIRE(gkfs::rpc::validate_replica_migration_plan(plan, 10, 11));

    REQUIRE_FALSE(gkfs::rpc::validate_replica_migration_plan(plan, 9, 11));

    REQUIRE(gkfs::rpc::classify_replica_migration_plan(plan, 9, 11) ==

            gkfs::rpc::replica_migration_plan_error::stale_generation);



    auto malformed = plan;

    malformed.jobs.front().target_node = 99;

    REQUIRE_FALSE(

            gkfs::rpc::validate_replica_migration_plan(malformed, 10, 11));

    REQUIRE(gkfs::rpc::classify_replica_migration_plan(malformed, 10, 11) ==

            gkfs::rpc::replica_migration_plan_error::invalid_layout);

}



TEST_CASE("replica migration plan rejects duplicate target copies",
@@ -132,6 +136,36 @@ TEST_CASE("replica migration plan rejects duplicate target copies", 
    REQUIRE_FALSE(gkfs::rpc::validate_replica_migration_plan(plan, 1, 2));

}



TEST_CASE("replica migration execution validates before invoking handler",

          "[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, 1, 2);



    int calls = 0;

    const auto stale = gkfs::rpc::execute_replica_migration_plan(

            plan, 0, 2, [&](const auto&) {

                ++calls;

                return true;

            });

    REQUIRE(stale == gkfs::rpc::replica_migration_plan_error::stale_generation);

    REQUIRE(calls == 0);



    const auto success = gkfs::rpc::execute_replica_migration_plan(

            plan, 1, 2, [&](const auto& job) {

                ++calls;

                return job.target_node >= 4;

            });

    REQUIRE(success == gkfs::rpc::replica_migration_plan_error::none);

    REQUIRE(calls == static_cast<int>(plan.jobs.size()));



    const auto failed = gkfs::rpc::execute_replica_migration_plan(

            plan, 1, 2, [](const auto&) { return false; });

    REQUIRE(failed ==

            gkfs::rpc::replica_migration_plan_error::execution_failed);

}



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

          "[replication][mutation]") {

    const fixed_distributor distributor({1});