Wake up the main thread after checkpoint is done (#1524)

Replicas are waiting for checkpoint indication via member key of the leader in DCS. The key is normally updated only one time per HA loop.
Without waking the main thread up replicas will have to wait up to `loop_wait` seconds longer than necessary.

patroni/ha.py

@@ -945,7 +945,7 @@ class Ha(object):
                     return msg
 
                 # check if the node is ready to be used by pg_rewind
-                self._rewind.ensure_checkpoint_after_promote()
+                self._rewind.ensure_checkpoint_after_promote(self.wakeup)
 
                 if self.is_standby_cluster():
                     # in case of standby cluster we don't really need to

patroni/postgresql/rewind.py

@@ -135,15 +135,17 @@ class Rewind(object):
             self._check_timeline_and_lsn(leader)
         return leader and leader.conn_url and self._state == REWIND_STATUS.NEED
 
-    def __checkpoint(self, task):
+    def __checkpoint(self, task, wakeup):
         try:
             result = self._postgresql.checkpoint()
         except Exception as e:
             result = 'Exception: ' + str(e)
         with task:
             task.complete(not bool(result))
+            if task.result:
+                wakeup()
 
-    def ensure_checkpoint_after_promote(self):
+    def ensure_checkpoint_after_promote(self, wakeup):
         """After promote issue a CHECKPOINT from a new thread and asynchronously check the result.
         In case if CHECKPOINT failed, just check that timeline in pg_control was updated."""
 
@@ -157,7 +159,7 @@ class Rewind(object):
                             return
                 else:
                     self._checkpoint_task = CriticalTask()
-                    return Thread(target=self.__checkpoint, args=(self._checkpoint_task,)).start()
+                    return Thread(target=self.__checkpoint, args=(self._checkpoint_task, wakeup)).start()
 
             if self._postgresql.get_master_timeline() == self._postgresql.pg_control_timeline():
                 self._state = REWIND_STATUS.CHECKPOINT

tests/test_rewind.py

@@ -117,16 +117,16 @@ class TestRewind(BaseTestPostgresql):
     @patch.object(Postgresql, 'checkpoint')
     def test_ensure_checkpoint_after_promote(self, mock_checkpoint, mock_controldata):
         mock_checkpoint.return_value = None
-        self.r.ensure_checkpoint_after_promote()
-        self.r.ensure_checkpoint_after_promote()
+        self.r.ensure_checkpoint_after_promote(Mock())
+        self.r.ensure_checkpoint_after_promote(Mock())
 
         self.r.reset_state()
         mock_controldata.return_value = {"Latest checkpoint's TimeLineID": 1}
         mock_checkpoint.side_effect = Exception
-        self.r.ensure_checkpoint_after_promote()
-        self.r.ensure_checkpoint_after_promote()
+        self.r.ensure_checkpoint_after_promote(Mock())
+        self.r.ensure_checkpoint_after_promote(Mock())
 
         self.r.reset_state()
         mock_controldata.side_effect = TypeError
-        self.r.ensure_checkpoint_after_promote()
-        self.r.ensure_checkpoint_after_promote()
+        self.r.ensure_checkpoint_after_promote(Mock())
+        self.r.ensure_checkpoint_after_promote(Mock())