Add a test for ban expiration unban timing

Author: magec

tests/ruby/helpers/pg_instance.rb

@@ -6,6 +6,7 @@ class PgInstance
   attr_reader :username
   attr_reader :password
   attr_reader :database_name
+  attr_reader :original_port
 
   def self.mass_takedown(databases)
     raise StandardError "block missing" unless block_given?

tests/ruby/helpers/pgcat_helper.rb

@@ -119,7 +119,7 @@ def self.single_instance_setup(pool_name, pool_size, pool_mode="transaction", lb
       end
     end
 
-    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info", pool_settings={})
+    def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mode="random", log_level="info", pool_settings={}, general_settings = {})
       user = {
         "password" => "sharding_user",
         "pool_size" => pool_size,
@@ -163,6 +163,7 @@ def self.single_shard_setup(pool_name, pool_size, pool_mode="transaction", lb_mo
         "#{pool_name}" => pool_config,
       }
       pgcat_cfg["general"]["port"] = pgcat.port
+      pgcat_cfg["general"] = pgcat_cfg["general"].merge(general_settings)
       pgcat.update_config(pgcat_cfg)
       pgcat.start
       pgcat.wait_until_ready

tests/ruby/load_balancing_spec.rb

@@ -165,7 +165,7 @@
 
 describe "Candidate filtering based on `default_pool`" do
   let(:processes) {
-    Helpers::Pgcat.single_shard_setup("sharded_db", 5, "transaction", "random", "debug", pool_settings)
+    Helpers::Pgcat.single_shard_setup("sharded_db", 5, "transaction", "random", "debug", pool_settings, general_settings)
   }
 
   after do
@@ -175,12 +175,16 @@
 
   context("with default_pool set to replicas") do
     context("when all replicas are down ") do
+      let(:ban_time) { 60 }
+      let(:connect_timeout) { 1000 }
       let(:pool_settings) do
         {
           "default_role" => "replica",
-          "replica_to_primary_failover_enabled" => replica_to_primary_failover_enabled
+          "replica_to_primary_failover_enabled" => replica_to_primary_failover_enabled,
+          "connect_timeout" => connect_timeout,
         }
       end
+      let(:general_settings) { { "ban_time" => ban_time } }
 
       context("with `replica_to_primary_failover_enabled` set to false`") do
         let(:replica_to_primary_failover_enabled) { false }
@@ -249,6 +253,49 @@
 
           expect(failed_count).to(eq(number_of_replicas))
         end
+        context("when banned replicas are tested for availability because they expired the ban time") do
+          let(:ban_time) { 2 }
+          it "should be done in the background without interfering with traffic" do
+            select_server_port = "SELECT setting AS port FROM pg_settings WHERE name = 'port';"
+            conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+            primary_port = processes.primary.original_port;
+
+            response = conn.async_exec(select_server_port)
+            expect(response[0]["port"].to_i).not_to(eq(primary_port))
+
+            failed_count = 0
+            number_of_replicas = processes[:replicas].length
+
+            # Take down all replicas
+            processes[:replicas].each(&:take_down)
+
+            (number_of_replicas).times do |n|
+              response = conn.async_exec(select_server_port)
+              expect(response[0]["port"].to_i).to(eq(primary_port))
+            rescue
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              failed_count += 1
+            end
+            expect(failed_count).to(eq(number_of_replicas))
+            failed_count = 0
+
+            response = conn.async_exec(select_server_port)
+            expect(response[0]["port"].to_i).to(eq(primary_port))
+            sleep(ban_time + 1)
+
+            begin
+              time_before_query = Time.now
+              response = conn.async_exec(select_server_port)
+              expect(response[0]["port"].to_i).to(eq(primary_port))
+              expect(Time.now - time_before_query).to(be < connect_timeout / 1000.0)
+            rescue
+              conn = PG.connect(processes.pgcat.connection_string("sharded_db", "sharding_user"))
+              failed_count += 1
+            end
+            expect(response[0]["port"].to_i).to(eq(primary_port))
+            expect(failed_count).to(eq(0))
+          end
+        end
       end
     end
   end