feat: add flag pause_wait_timeout (#5328)

* feat: add flag pause_wait_timeout

Author: BorysTheDev

src/server/server_family.cc

@@ -112,6 +112,10 @@ ABSL_FLAG(int32_t, slowlog_log_slower_than, 10000,
           "microseconds and if it's negative - disables the slowlog.");
 ABSL_FLAG(uint32_t, slowlog_max_len, 20, "Slow log maximum length.");
 
+ABSL_FLAG(uint32_t, pause_wait_timeout, 1,
+          "Timeout in seconds, to set up the pause for all connections for CLIENT PAUSE command "
+          "and cluster slot migration finalization procedure.");
+
 ABSL_FLAG(string, s3_endpoint, "", "endpoint for s3 snapshots, default uses aws regional endpoint");
 ABSL_FLAG(bool, s3_use_https, true, "whether to use https for s3 endpoints");
 // Disable EC2 metadata by default, or if a users credentials are invalid the
@@ -751,7 +755,7 @@ std::optional<fb2::Fiber> Pause(std::vector<facade::Listener*> listeners, Namesp
 
   // Wait for all busy commands to finish running before replying to guarantee
   // that no more (write) operations will occur.
-  const absl::Duration kDispatchTimeout = absl::Seconds(1);
+  const absl::Duration kDispatchTimeout = absl::Seconds(absl::GetFlag(FLAGS_pause_wait_timeout));
   if (!tracker.Wait(kDispatchTimeout)) {
     LOG(WARNING) << "Couldn't wait for commands to finish dispatching in " << kDispatchTimeout;
     shard_set->pool()->AwaitBrief([pause_state](unsigned, util::ProactorBase*) {

tests/dragonfly/cluster_test.py

@@ -1810,8 +1810,7 @@ async def test_cluster_replication_migration(
     assert await seeder.compare(fake_capture, r1_node.instance.port)
 
 
-@pytest.mark.skip("Flaky test")
-@dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
+@dfly_args({"proactor_threads": 4, "cluster_mode": "yes", "pause_wait_timeout": 10})
 async def test_start_replication_during_migration(
     df_factory: DflyInstanceFactory, df_seeder_factory: DflySeederFactory
 ):
@@ -2561,8 +2560,7 @@ async def test_replicate_redis_cluster(redis_cluster, df_factory, df_seeder_fact
     assert await seeder.compare(capture, replica.port)
 
 
-@pytest.mark.skip("Flaky test")
-@dfly_args({"proactor_threads": 4})
+@dfly_args({"proactor_threads": 4, "pause_wait_timeout": 10})
 async def test_replicate_disconnect_redis_cluster(redis_cluster, df_factory, df_seeder_factory):
     """
     Create redis cluster of 3 nodes.
@@ -2859,11 +2857,10 @@ async def test_migration_one_after_another(df_factory: DflyInstanceFactory, df_s
 """
 
 
-@pytest.mark.skip("Flaky test")
 @pytest.mark.slow
 @pytest.mark.exclude_epoll
 @pytest.mark.asyncio
-@dfly_args({"proactor_threads": 4, "cluster_mode": "yes"})
+@dfly_args({"proactor_threads": 4, "cluster_mode": "yes", "pause_wait_timeout": 10})
 async def test_migration_rebalance_node(df_factory: DflyInstanceFactory, df_seeder_factory):
     # 1. Create cluster of 3 nodes with all slots allocated to first node.
     instances = [