Move next state logic in Finish function

Author: mkaruza

src/server/cluster/cluster_family.cc

@@ -835,7 +835,7 @@ ClusterFamily::TakeOutOutgoingMigrations(shared_ptr<ClusterConfig> new_config,
     removed_slots.Merge(slots);
     LOG(INFO) << "Outgoing migration cancelled: slots " << slots.ToString() << " to "
               << migration.GetHostIp() << ":" << migration.GetPort();
-    migration.Finish(MigrationState::C_FINISHED);
+    migration.Finish();
     res.migrations.push_back(std::move(*it));
     outgoing_migration_jobs_.erase(it);
   }

src/server/cluster/outgoing_slot_migration.cc

@@ -37,8 +37,7 @@ class OutgoingMigration::SliceSlotMigration : private ProtocolClient {
   SliceSlotMigration(DbSlice* slice, ServerContext server_context, SlotSet slots,
                      journal::Journal* journal, OutgoingMigration* om)
       : ProtocolClient(server_context), streamer_(slice, std::move(slots), journal, &exec_st_) {
-    exec_st_.SwitchErrorHandler(
-        [om](auto ge) { om->Finish(MigrationState::C_ERROR, std::move(ge)); });
+    exec_st_.SwitchErrorHandler([om](auto ge) { om->Finish(std::move(ge)); });
   }
 
   ~SliceSlotMigration() {
@@ -143,8 +142,15 @@ void OutgoingMigration::OnAllShards(
   });
 }
 
-void OutgoingMigration::Finish(MigrationState next_state, GenericError error) {
+void OutgoingMigration::Finish(GenericError error) {
+  auto next_state = MigrationState::C_FINISHED;
   if (error) {
+    // If OOM error move to FATAL, non-recoverable  state
+    if (error == errc::not_enough_memory) {
+      next_state = MigrationState::C_FATAL;
+    } else {
+      next_state = MigrationState::C_ERROR;
+    }
     LOG(WARNING) << "Finish outgoing migration for " << cf_->MyID() << ": "
                  << migration_info_.node_info.id << " with error: " << error.Format();
     exec_st_.ReportError(std::move(error));
@@ -380,7 +386,8 @@ bool OutgoingMigration::FinalizeMigration(long attempt) {
 
     // Check OOM from incoming slot migration on ACK request
     if (CheckRespSimpleError(kIncomingMigrationOOM)) {
-      Finish(MigrationState::C_FATAL, std::string(kIncomingMigrationOOM));
+      Finish(GenericError{std::make_error_code(errc::not_enough_memory),
+                          std::string(kIncomingMigrationOOM)});
       return false;
     }
 
@@ -400,7 +407,7 @@ bool OutgoingMigration::FinalizeMigration(long attempt) {
   }
 
   if (!exec_st_.GetError()) {
-    Finish(MigrationState::C_FINISHED);
+    Finish();
     keys_number_ = cluster::GetKeyCount(migration_info_.slot_ranges);
     cf_->ApplyMigrationSlotRangeToConfig(migration_info_.node_info.id, migration_info_.slot_ranges,
                                          false);

src/server/cluster/outgoing_slot_migration.h

@@ -30,7 +30,7 @@ class OutgoingMigration : private ProtocolClient {
   // if is_error = false mark migration as FINISHED and cancel migration if it's not finished yet
   // can be called from any thread, but only after Start()
   // if is_error = true and migration is in progress it will be restarted otherwise nothing happens
-  void Finish(MigrationState next_state, GenericError error = {}) ABSL_LOCKS_EXCLUDED(state_mu_);
+  void Finish(GenericError error = {}) ABSL_LOCKS_EXCLUDED(state_mu_);
 
   MigrationState GetState() const ABSL_LOCKS_EXCLUDED(state_mu_);
 

tests/dragonfly/cluster_test.py

@@ -3283,7 +3283,7 @@ async def test_slot_migration_oom(df_factory):
     # Direction
     assert status[0][0] == "out"
     # Error message
-    assert status[0][4] == "INCOMING_MIGRATION_OOM"
+    assert status[0][4] == "Cannot allocate memory: INCOMING_MIGRATION_OOM"
 
     # Node_1 slot-migration-status
     status = await nodes[1].admin_client.execute_command(