Avoid retries when syncing replication slots. (#282)

* Avoid retries when syncing replication slots.

Do not retry postgres queries that fetch, create and drop slots at the end of
the HA cycle. The complete run_cycle routine executes with the async_executor
lock. This lock is also used with scheduling operations like reinit or restart
in different threads. Looks like CPython threading class has fairness issues
when multiple threads try to acquire the same lock and one of them executes
long-running actions while holding it: the others have little chances of
acquiring the lock in order. To get around this issue, the long action (i.e.
retrying the query) is removed.

Investigation by Ants Aasma and Alexander Kukushkin.

patroni/__init__.py

@@ -84,6 +84,10 @@ class Patroni(object):
         nap_time = self.next_run - current_time
         if nap_time <= 0:
             self.next_run = current_time
+            # Release the GIL so we don't starve anyone waiting on async_executor lock
+            time.sleep(0.001)
+            # Warn user that Patroni is not keeping up
+            logger.warning("Loop time exceeded, rescheduling immediately.")
         elif self.dcs.watch(nap_time):
             self.next_run = time.time()
 

patroni/ha.py

@@ -571,9 +571,7 @@ class Ha(object):
 
             # try to start dead postgres
             if not self.state_handler.is_healthy():
-                msg = self.recover()
-                if msg is not None:
-                    return msg
+                return self.recover()
 
             try:
                 if self.cluster.is_unlocked():

patroni/postgresql.py

@@ -890,7 +890,7 @@ $$""".format(name, ' '.join(options)), name, password, password)
 
     def load_replication_slots(self):
         if self.use_slots and self._schedule_load_slots:
-            cursor = self.query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
+            cursor = self._query("SELECT slot_name FROM pg_replication_slots WHERE slot_type='physical'")
             self._replication_slots = [r[0] for r in cursor]
             self._schedule_load_slots = False
 
@@ -930,19 +930,20 @@ $$""".format(name, ' '.join(options)), name, password, password)
 
                 # drop unused slots
                 for slot in set(self._replication_slots) - slots:
-                    self.query("""SELECT pg_drop_replication_slot(%s)
-                                   WHERE EXISTS(SELECT 1 FROM pg_replication_slots
-                                   WHERE slot_name = %s AND NOT active)""", slot, slot)
+                    self._query("""SELECT pg_drop_replication_slot(%s)
+                                    WHERE EXISTS(SELECT 1 FROM pg_replication_slots
+                                    WHERE slot_name = %s AND NOT active)""", slot, slot)
 
                 # create new slots
                 for slot in slots - set(self._replication_slots):
-                    self.query("""SELECT pg_create_physical_replication_slot(%s)
-                                   WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
-                                   WHERE slot_name = %s)""", slot, slot)
+                    self._query("""SELECT pg_create_physical_replication_slot(%s)
+                                    WHERE NOT EXISTS (SELECT 1 FROM pg_replication_slots
+                                    WHERE slot_name = %s)""", slot, slot)
 
                 self._replication_slots = slots
-            except psycopg2.Error:
+            except Exception:
                 logger.exception('Exception when changing replication slots')
+                self._schedule_load_slots = True
 
     def last_operation(self):
         return str(self.xlog_position())

tests/test_postgresql.py

@@ -315,11 +315,9 @@ class TestPostgresql(unittest.TestCase):
     def test_sync_replication_slots(self):
         self.p.start()
         cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None)
+        with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg2.OperationalError)):
+            self.p.sync_replication_slots(cluster)
         self.p.sync_replication_slots(cluster)
-        self.p.query = Mock(side_effect=psycopg2.OperationalError)
-        self.p.schedule_load_slots = True
-        self.p.sync_replication_slots(cluster)
-        self.p.schedule_load_slots = False
         with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
             self.p.sync_replication_slots(cluster)
         with mock.patch('patroni.postgresql.logger.error', new_callable=Mock()) as errorlog_mock: