Fix race conditions in async actions (#1215)

Specifically, there was a chance that `patronictl reinit --force` was overwritten by recover and we end up in a situation when Patroni was trying to start the postgres while basebackup still running.

patroni/async_executor.py

@@ -110,6 +110,12 @@ class AsyncExecutor(object):
     def run_async(self, func, args=()):
         Thread(target=self.run, args=(func, args)).start()
 
+    def try_run_async(self, action, func, args=()):
+        prev = self.schedule(action)
+        if prev is None:
+            return self.run_async(func, args)
+        return 'Failed to run {0}, {1} is already in progress'.format(action, prev)
+
     def cancel(self):
         with self:
             with self._scheduled_action_lock:

patroni/ha.py

@@ -230,9 +230,8 @@ class Ha(object):
             clone_member = self.cluster.get_clone_member(self.state_handler.name)
             member_role = 'leader' if clone_member == self.cluster.leader else 'replica'
             msg = "from {0} '{1}'".format(member_role, clone_member.name)
-            self._async_executor.schedule('bootstrap {0}'.format(msg))
-            self._async_executor.run_async(self.clone, args=(clone_member, msg))
-            return 'trying to bootstrap {0}'.format(msg)
+            ret = self._async_executor.try_run_async('bootstrap {0}'.format(msg), self.clone, args=(clone_member, msg))
+            return ret or 'trying to bootstrap {0}'.format(msg)
 
         # no initialize key and node is allowed to be master and has 'bootstrap' section in a configuration file
         elif self.cluster.initialize is None and not self.patroni.nofailover and 'bootstrap' in self.patroni.config:
@@ -241,16 +240,12 @@ class Ha(object):
                 self._post_bootstrap_task = CriticalTask()
 
                 if self.is_standby_cluster():
-                    self._async_executor.schedule('bootstrap_standby_leader')
-                    self._async_executor.run_async(self.bootstrap_standby_leader)
-                    return 'trying to bootstrap a new standby leader'
+                    ret = self._async_executor.try_run_async('bootstrap_standby_leader', self.bootstrap_standby_leader)
+                    return ret or 'trying to bootstrap a new standby leader'
                 else:
-                    self._async_executor.schedule('bootstrap')
-                    self._async_executor.run_async(
-                        self.state_handler.bootstrap.bootstrap,
-                        args=(self.patroni.config['bootstrap'],)
-                    )
-                    return 'trying to bootstrap a new cluster'
+                    ret = self._async_executor.try_run_async('bootstrap', self.state_handler.bootstrap.bootstrap,
+                                                             args=(self.patroni.config['bootstrap'],))
+                    return ret or 'trying to bootstrap a new cluster'
             else:
                 return 'failed to acquire initialize lock'
         else:
@@ -258,9 +253,7 @@ class Ha(object):
                                      if self.is_standby_cluster() else None
             if self.state_handler.can_create_replica_without_replication_connection(create_replica_methods):
                 msg = 'bootstrap (without leader)'
-                self._async_executor.schedule(msg)
-                self._async_executor.run_async(self.clone)
-                return 'trying to ' + msg
+                return self._async_executor.try_run_async(msg, self.clone) or 'trying to ' + msg
             return 'waiting for {0}leader to bootstrap'.format('standby_' if self.is_standby_cluster() else '')
 
     def bootstrap_standby_leader(self):
@@ -283,15 +276,13 @@ class Ha(object):
             return None
 
         if self._rewind.can_rewind:
-            self._async_executor.schedule('running pg_rewind from ' + leader.name)
-            self._async_executor.run_async(self._rewind.execute, (leader,))
-            return True
+            msg = 'running pg_rewind from ' + leader.name
+            return self._async_executor.try_run_async(msg, self._rewind.execute, args=(leader,)) or msg
 
         # remove_data_directory_on_diverged_timelines is set
         if not self.is_standby_cluster():
-            self._async_executor.schedule('reinitializing due to diverged timelines')
-            self._async_executor.run_async(self._do_reinitialize, args=(self.cluster, ))
-            return True
+            msg = 'reinitializing due to diverged timelines'
+            return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg
 
     def recover(self):
         # Postgres is not running and we will restart in standby mode. Watchdog is not needed until we promote.
@@ -318,9 +309,8 @@ class Ha(object):
                 and not self._crash_recovery_executed and \
                 (self.cluster.is_unlocked() or self._rewind.can_rewind):
             self._crash_recovery_executed = True
-            self._async_executor.schedule('doing crash recovery in a single user mode')
-            self._async_executor.run_async(self.state_handler.fix_cluster_state)
-            return self._async_executor.scheduled_action
+            msg = 'doing crash recovery in a single user mode'
+            return self._async_executor.try_run_async(msg, self.state_handler.fix_cluster_state) or msg
 
         self.load_cluster_from_dcs()
 
@@ -328,8 +318,9 @@ class Ha(object):
         if self.is_standby_cluster() or not self.has_lock():
             if not self._rewind.executed:
                 self._rewind.trigger_check_diverged_lsn()
-            if self._handle_rewind_or_reinitialize():
-                return self._async_executor.scheduled_action
+            msg = self._handle_rewind_or_reinitialize()
+            if msg:
+                return msg
 
             if self.has_lock():  # in standby cluster
                 msg = "starting as a standby leader because i had the session lock"
@@ -345,10 +336,9 @@ class Ha(object):
             msg = "starting as readonly because i had the session lock"
             node_to_follow = None
 
-        self.recovering = True
-
-        self._async_executor.schedule('restarting after failure')
-        self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role, timeout))
+        if self._async_executor.try_run_async('restarting after failure', self.state_handler.follow,
+                                              args=(node_to_follow, role, timeout)) is None:
+            self.recovering = True
         return msg
 
     def _get_node_to_follow(self, cluster):
@@ -383,8 +373,9 @@ class Ha(object):
             self.demote('immediate-nolock')
             return demote_reason
 
-        if self._handle_rewind_or_reinitialize():
-            return self._async_executor.scheduled_action
+        msg = self._handle_rewind_or_reinitialize()
+        if msg:
+            return msg
 
         role = 'standby_leader' if isinstance(node_to_follow, RemoteMember) and self.has_lock(False) else 'replica'
         # It might happen that leader key in the standby cluster references non-exiting member.
@@ -392,8 +383,8 @@ class Ha(object):
         if self.is_standby_cluster() and role == 'replica' and not (node_to_follow and node_to_follow.conn_url):
             return 'continue following the old known standby leader'
         elif not self.state_handler.config.check_recovery_conf(node_to_follow):
-            self._async_executor.schedule('changing primary_conninfo and restarting')
-            self._async_executor.run_async(self.state_handler.follow, (node_to_follow, role))
+            self._async_executor.try_run_async('changing primary_conninfo and restarting',
+                                               self.state_handler.follow, args=(node_to_follow, role))
         elif role == 'standby_leader' and self.state_handler.role != role:
             self.state_handler.set_role(role)
             self.state_handler.call_nowait(ACTION_ON_ROLE_CHANGE)
@@ -552,9 +543,9 @@ class Ha(object):
                     self._rewind.reset_state()
                     logger.info("cleared rewind state after becoming the leader")
 
-                self._async_executor.schedule('promote')
-                self._async_executor.run_async(self.state_handler.promote,
-                                               args=(self.dcs.loop_wait, on_success, self._leader_access_is_restricted))
+                self._async_executor.try_run_async('promote', self.state_handler.promote,
+                                                   args=(self.dcs.loop_wait, on_success,
+                                                         self._leader_access_is_restricted))
             return promote_message
 
     def fetch_node_status(self, member):
@@ -772,8 +763,7 @@ class Ha(object):
         # there could be an async action already running, calling follow from here will lead
         # to racy state handler state updates.
         if mode_control['async_req']:
-            self._async_executor.schedule('starting after demotion')
-            self._async_executor.run_async(self.state_handler.follow, (node_to_follow,))
+            self._async_executor.try_run_async('starting after demotion', self.state_handler.follow, (node_to_follow,))
         else:
             if self.is_synchronous_mode():
                 self.state_handler.config.set_synchronous_standby(None)
@@ -846,9 +836,8 @@ class Ha(object):
                         members = [m for m in self.cluster.members
                                    if not failover.candidate or m.name == failover.candidate]
                     if self.is_failover_possible(members):  # check that there are healthy members
-                        self._async_executor.schedule('manual failover: demote')
-                        self._async_executor.run_async(self.demote, ('graceful',))
-                        return 'manual failover: demoting myself'
+                        ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',))
+                        return ret or 'manual failover: demoting myself'
                     else:
                         logger.warning('manual failover: no healthy members found, failover is not possible')
             else:
@@ -1151,10 +1140,9 @@ class Ha(object):
                 return 'waiting for end of recovery after bootstrap'
 
             self.state_handler.set_role('master')
-            self._async_executor.schedule('post_bootstrap')
-            self._async_executor.run_async(self.state_handler.bootstrap.post_bootstrap,
-                                           args=(self.patroni.config['bootstrap'], self._post_bootstrap_task))
-            return 'running post_bootstrap'
+            ret = self._async_executor.try_run_async('post_bootstrap', self.state_handler.bootstrap.post_bootstrap,
+                                                     args=(self.patroni.config['bootstrap'], self._post_bootstrap_task))
+            return ret or 'running post_bootstrap'
 
         self.state_handler.bootstrapping = False
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))

tests/test_ha.py

@@ -141,7 +141,11 @@ zookeeper:
 
 
 def run_async(self, func, args=()):
-    return func(*args) if args else func()
+    self.reset_scheduled_action()
+    if args:
+        func(*args)
+    else:
+        func()
 
 
 @patch.object(Postgresql, 'is_running', Mock(return_value=MockPostmaster()))
@@ -426,7 +430,7 @@ class TestHa(PostgresInit):
 
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertIsNone(self.ha.reinitialize(True))
-
+        self.ha._async_executor.schedule('reinitialize')
         self.assertIsNotNone(self.ha.reinitialize())
 
         self.ha.state_handler.name = self.ha.cluster.leader.name
@@ -440,7 +444,7 @@ class TestHa(PostgresInit):
         self.p.restart = false
         self.assertEqual(self.ha.restart({}), (False, 'restart failed'))
         self.ha.cluster = get_cluster_initialized_with_leader()
-        self.ha.reinitialize()
+        self.ha._async_executor.schedule('reinitialize')
         self.assertEqual(self.ha.restart({}), (False, 'reinitialize already in progress'))
         with patch.object(self.ha, "restart_matches", return_value=False):
             self.assertEqual(self.ha.restart({'foo': 'bar'}), (False, "restart conditions are not satisfied"))
@@ -448,7 +452,7 @@ class TestHa(PostgresInit):
     @patch('os.kill', Mock())
     def test_restart_in_progress(self):
         with patch('patroni.async_executor.AsyncExecutor.busy', PropertyMock(return_value=True)):
-            self.ha.restart({}, run_async=True)
+            self.ha._async_executor.schedule('restart')
             self.assertTrue(self.ha.restart_scheduled())
             self.assertEqual(self.ha.run_cycle(), 'restart in progress')