Fix race condition causing a spurious promote during a global DCS outage

Fallback leader observation mechanism was using a non-quorum read that
can see a stale value of multisite status. For purposes of rewinding
stadbys this is fine, but if timing was wrong it caused the main HA loop
to observe the stale value and promote.

Fix this by only running the leader observation fallback while the node
is not a leader. If the node is leader the regular heartbeat will take
care of updating the view.

Observing a stale value during startup is not a problem because
promoting to local leader will force a write to global DCS via
resolve_leader().

Author: ants

patroni/ha.py

@@ -300,6 +300,7 @@ def set_is_leader(self, value: bool) -> None:
 
         :param value: is the current node the leader.
         """
+        self.patroni.multisite.set_is_local_leader(value)
         with self._leader_expiry_lock:
             self._leader_expiry = time.time() + self.dcs.ttl if value else 0
             if not value:

patroni/multisite.py

@@ -64,6 +64,10 @@ def status(self) -> Dict[str, Any]:
     def should_failover(self) -> bool:
         return False
 
+    def set_is_local_leader(self, value: bool):
+        """Update multisite mechanisms view if this node is running as a site leader."""
+        pass
+
     def on_shutdown(self, checkpoint_location: int, prev_location: int):
         pass
 
@@ -111,6 +115,8 @@ def __init__(self, config: 'Config', on_change: Optional[Callable[..., None]] =
 
         self._dcs_error = None
 
+        self._is_local_leader = False
+
     @staticmethod
     def get_dcs_config(config: 'Config') -> Tuple[Dict[str, Any], AbstractDCS]:
         msconfig = config['multisite']
@@ -419,6 +425,10 @@ def touch_member(self):
         logger.info(f"Touching member {self.name} with {data!r}")
         self.dcs.touch_member(data)
 
+    def set_is_local_leader(self, value: bool):
+        # Assumes setting a boolean flag from other threads without a lock is atomic
+        self._is_local_leader = value
+
     def run(self):
         self._observe_leader()
         while not self._heartbeat.wait(self.config['observe_interval']):
@@ -431,7 +441,8 @@ def run(self):
             if self._state_updater:
                 self._state_updater.store_updates()
             while not self._heartbeat.wait(self.config['observe_interval']):
-                self._observe_leader()
+                if not self._is_local_leader:
+                    self._observe_leader()
 
     def shutdown(self):
         self.stop_requested = True