Check that synchronous_standby_names contains expected value (patroni#3370)

* Change interface of SyncHandler.current_state()

Make it always return nodes from synchronous_standby_names and
sync_confirmed instead of numsync_confirmed.

* Check that synchronous_standby_names contains expected value

Current mechanism implementing state machine for non-quorum synchronous
replication didn't check actual value of synchronous_standby_names, what
resulted in stale value of synchronous_standby_names being used when
pg_stat_replication is a subset of synchronous_standby_names.

Such situation is mainly possible when standby nodes with replicatefrom
tag connect to the primary because cascading replicas are temporary not
accessible.

To prevent it from happening we implement following measures:
1. Check the actual value of synchronous_standby_names when deceding
   whether we should continue to run state machine
2. Take into account `replicatefrom` tag on nodes that are streaming
   from the primary and don't consider them to be synchronous candidates
   if some other node in a chain already streaming from the primary.

Close patroni#3369

Author: CyberDem0n

patroni/ha.py

@@ -860,7 +860,7 @@ def _check_timeout(offset: float = 0) -> bool:
                                                                       voters=sync.voters,
                                                                       numsync=sync_state.numsync,
                                                                       sync=sync_state.sync,
-                                                                      numsync_confirmed=sync_state.numsync_confirmed,
+                                                                      numsync_confirmed=len(sync_state.sync_confirmed),
                                                                       active=sync_state.active,
                                                                       sync_wanted=sync_wanted,
                                                                       leader_wanted=self.state_handler.name):
@@ -906,7 +906,7 @@ def _process_multisync_replication(self) -> None:
 
         current_state = self.state_handler.sync_handler.current_state(self.cluster)
         picked = current_state.active
-        allow_promote = current_state.sync
+        allow_promote = current_state.sync_confirmed
         voters = CaseInsensitiveSet(sync.voters)
 
         if picked == voters and voters != allow_promote:
@@ -917,7 +917,7 @@ def _process_multisync_replication(self) -> None:
                 return logger.warning("Updating sync state failed")
             voters = CaseInsensitiveSet(sync.voters)
 
-        if picked == voters:
+        if picked == voters == current_state.sync and current_state.numsync == len(picked):
             return
 
         # update synchronous standby list in dcs temporarily to point to common nodes in current and picked
@@ -941,7 +941,7 @@ def _process_multisync_replication(self) -> None:
         if picked and picked != CaseInsensitiveSet('*') and allow_promote != picked:
             # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
             time.sleep(2)
-            allow_promote = self.state_handler.sync_handler.current_state(self.cluster).sync
+            allow_promote = self.state_handler.sync_handler.current_state(self.cluster).sync_confirmed
 
         if allow_promote and allow_promote != sync_common:
             if self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, version=sync.version):

patroni/postgresql/sync.py

@@ -7,7 +7,7 @@
 
 from .. import global_config
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
-from ..dcs import Cluster
+from ..dcs import Cluster, Member
 from ..psycopg import quote_ident
 from .misc import PostgresqlState
 
@@ -167,17 +167,16 @@ class _SyncState(NamedTuple):
     :ivar sync_type: possible values: ``off``, ``priority``, ``quorum``
     :ivar numsync: how many nodes are required to be synchronous (according to ``synchronous_standby_names``).
                    Is ``0`` if ``synchronous_standby_names`` value is invalid or contains ``*``.
-    :ivar numsync_confirmed: how many nodes are known to be synchronous according to the ``pg_stat_replication`` view.
-                             Only nodes that caught up with the :attr:`SyncHandler._primary_flush_lsn` are counted.
-    :ivar sync: collection of synchronous node names. In case of quorum commit all nodes listed
-                in ``synchronous_standby_names``, otherwise nodes that are confirmed to be synchronous according
-                to the ``pg_stat_replication`` view.
+    :ivar sync: collection of synchronous node names from ``synchronous_standby_names``.
+    :ivar sync_confirmed: collection of synchronous node names from ``synchronous_standby_names`` that are
+                          confirmed to be synchronous according to the ``pg_stat_replication`` view.
+                          Only nodes that caught up with the :attr:`SyncHandler._primary_flush_lsn` are counted.
     :ivar active: collection of node names that are streaming and have no restrictions to become synchronous.
     """
     sync_type: str
     numsync: int
-    numsync_confirmed: int
     sync: CaseInsensitiveSet
+    sync_confirmed: CaseInsensitiveSet
     active: CaseInsensitiveSet
 
 
@@ -229,16 +228,20 @@ def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
             'remote_write': 'write'
         }.get(postgresql.synchronous_commit(), 'flush') + '_lsn'
 
-        members = CaseInsensitiveDict({m.name: m for m in cluster.members if m.name.lower() != postgresql.name.lower()})
-        for row in postgresql.pg_stat_replication():
+        members = CaseInsensitiveDict({m.name: m for m in cluster.members
+                                       if m.is_running and not m.nosync and m.name.lower() != postgresql.name.lower()})
+        replication = CaseInsensitiveDict({row['application_name']: row for row in postgresql.pg_stat_replication()
+                                           if row[sort_col] is not None and row['application_name'] in members})
+        for row in replication.values():
             member = members.get(row['application_name'])
 
             # We want to consider only rows from ``pg_stat_replication` that:
             # 1. are known to be streaming (write/flush/replay LSN are not NULL).
             # 2. can be mapped to a ``Member`` of the ``Cluster``:
             #   a. ``Member`` doesn't have ``nosync`` tag set;
             #   b. PostgreSQL on the member is known to be running and accepting client connections.
-            if member and row[sort_col] is not None and member.is_running and not member.nosync:
+            #   c. ``Member`` isn't supposed to stream from another standby (``replicatefrom`` tag).
+            if member and row[sort_col] is not None and not self._should_cascade(members, replication, member):
                 self.append(_Replica(row['pid'], row['application_name'],
                                      row['sync_state'], row[sort_col],
                                      bool(member.nofailover), member.sync_priority))
@@ -251,6 +254,27 @@ def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
         # up-to-date replica otherwise with cluster LSN if there is only one replica.
         self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation()
 
+    @staticmethod
+    def _should_cascade(members: CaseInsensitiveDict, replication: CaseInsensitiveDict, member: Member) -> bool:
+        """Check whether *member* is supposed to cascade from another standby node.
+
+        :param members: members that are eligible to stream (state=running and don't have nosync tag)
+        :param replication: state of ``pg_stat_replication``, already filtered by member names from *members*
+        :param member: member that we want to check
+
+        :returns: ``True`` if provided member should stream from other standby node in
+                  the cluster (according to ``replicatefrom`` tag), because some standbys
+                  in a chain already streaming from the primary, otherwise ``False``
+        """
+        if not member.replicatefrom or member.replicatefrom not in members:
+            return False
+
+        member = members[member.replicatefrom]
+        if not member.replicatefrom:
+            return member.name in replication
+
+        return _ReplicaList._should_cascade(members, replication, member)
+
 
 class SyncHandler(object):
     """Class responsible for working with the `synchronous_standby_names`.
@@ -350,8 +374,7 @@ def current_state(self, cluster: Cluster) -> _SyncState:
         self._process_replica_readiness(cluster, replica_list)
 
         active = CaseInsensitiveSet()
-        sync_nodes = CaseInsensitiveSet()
-        numsync_confirmed = 0
+        sync_confirmed = CaseInsensitiveSet()
 
         sync_node_count = global_config.synchronous_node_count if self._postgresql.supports_multiple_sync else 1
         sync_node_maxlag = global_config.maximum_lag_on_syncnode
@@ -365,24 +388,20 @@ def current_state(self, cluster: Cluster) -> _SyncState:
                     # there is a chance that a non-promotable node is ahead of a promotable one.
                     if not replica.nofailover or len(active) < sync_node_count:
                         if replica.application_name in self._ready_replicas:
-                            numsync_confirmed += 1
+                            sync_confirmed.add(replica.application_name)
                         active.add(replica.application_name)
                 else:
                     active.add(replica.application_name)
                     if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas:
-                        sync_nodes.add(replica.application_name)
-                        numsync_confirmed += 1
+                        sync_confirmed.add(replica.application_name)
                     if len(active) >= sync_node_count:
                         break
 
-        if global_config.is_quorum_commit_mode:
-            sync_nodes = CaseInsensitiveSet() if self._ssn_data.has_star else self._ssn_data.members
-
         return _SyncState(
             self._ssn_data.sync_type,
-            0 if self._ssn_data.has_star else self._ssn_data.num,
-            numsync_confirmed,
-            sync_nodes,
+            self._ssn_data.num,
+            CaseInsensitiveSet() if self._ssn_data.has_star else self._ssn_data.members,
+            sync_confirmed,
             active)
 
     def set_synchronous_standby_names(self, sync: Collection[str], num: Optional[int] = None) -> None:

tests/test_ha.py

@@ -979,7 +979,6 @@ def test_manual_failover_process_no_leader_in_synchronous_mode(self):
         with patch('patroni.ha.logger.warning') as mock_warning:
             self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'other', None),
                                                                      sync=('leader1', 'postgresql0'))
-            self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
             self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
             self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
             self.assertEqual(mock_warning.call_args_list[0][0],
@@ -990,8 +989,6 @@ def test_manual_failover_process_no_leader_in_synchronous_mode(self):
         self.p.set_role(PostgresqlRole.REPLICA)
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
                                                                  sync=('leader1', 'other'))
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['leader1']),
-                                                               CaseInsensitiveSet(['leader1'])))
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
     def test_manual_switchover_process_no_leader_in_synchronous_mode(self):
@@ -1366,7 +1363,8 @@ def test__process_multisync_replication(self):
         self.ha.is_synchronous_mode = true
 
         # Test sync standby not touched when picking the same node
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1,
+                                                                         CaseInsensitiveSet(['other']),
                                                                          CaseInsensitiveSet(['other']),
                                                                          CaseInsensitiveSet(['other'])))
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
@@ -1377,15 +1375,17 @@ def test__process_multisync_replication(self):
         mock_cfg_set_sync.reset_mock()
 
         # Test sync standby is replaced when switching standbys
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, CaseInsensitiveSet(),
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, CaseInsensitiveSet(),
+                                                                         CaseInsensitiveSet(),
                                                                          CaseInsensitiveSet(['other2'])))
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(CaseInsensitiveSet(['other2']))
         mock_cfg_set_sync.assert_not_called()
 
         # Test sync standby is replaced when new standby is joined
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1,
+                                                                         CaseInsensitiveSet(['other2']),
                                                                          CaseInsensitiveSet(['other2']),
                                                                          CaseInsensitiveSet(['other2', 'other3'])))
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
@@ -1408,7 +1408,8 @@ def test__process_multisync_replication(self):
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         self.ha.dcs.get_cluster = Mock(return_value=get_cluster_initialized_with_leader(sync=('leader', 'other')))
         # self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1,
+                                                                         CaseInsensitiveSet(['other2']),
                                                                          CaseInsensitiveSet(['other2']),
                                                                          CaseInsensitiveSet(['other2'])))
         self.ha.run_cycle()
@@ -1433,8 +1434,8 @@ def test__process_multisync_replication(self):
         # Test sync set to '*' when synchronous_mode_strict is enabled
         mock_set_sync.reset_mock()
         mock_cfg_set_sync.reset_mock()
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, CaseInsensitiveSet(),
-                                                                         CaseInsensitiveSet()))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, CaseInsensitiveSet(),
+                                                                         CaseInsensitiveSet(), CaseInsensitiveSet()))
         with patch.object(global_config.__class__, 'is_synchronous_mode_strict', PropertyMock(return_value=True)):
             self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(CaseInsensitiveSet('*'))
@@ -1545,7 +1546,7 @@ def test_enable_synchronous_mode(self):
         self.ha.is_synchronous_mode = true
         self.ha.has_lock = true
         self.p.name = 'leader'
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, CaseInsensitiveSet(),
                                                                          CaseInsensitiveSet(), CaseInsensitiveSet()))
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         with patch('patroni.ha.logger.info') as mock_logger:
@@ -1562,7 +1563,7 @@ def test_inconsistent_synchronous_state(self):
         self.ha.has_lock = true
         self.p.name = 'leader'
         self.ha.cluster = get_cluster_initialized_without_leader(sync=('leader', 'a'))
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, CaseInsensitiveSet(),
                                                                          CaseInsensitiveSet(), CaseInsensitiveSet('a')))
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
@@ -1790,7 +1791,8 @@ def test__process_quorum_replication(self):
 
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=None)
         # Test /sync key is attempted to set and failed when missing or invalid
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('quorum', 1, 1, CaseInsensitiveSet(['other']),
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('quorum', 1, CaseInsensitiveSet(['other']),
+                                                                         CaseInsensitiveSet(['other']),
                                                                          CaseInsensitiveSet(['other'])))
         self.ha.run_cycle()
         self.assertEqual(mock_write_sync.call_count, 1)
@@ -1810,9 +1812,11 @@ def test__process_quorum_replication(self):
         self.assertEqual(mock_write_sync.call_args_list[1][1], {'version': None})
         self.assertEqual(mock_set_sync.call_count, 0)
 
-        self.p.sync_handler.current_state = Mock(side_effect=[_SyncState('quorum', 1, 0, CaseInsensitiveSet(['foo']),
+        self.p.sync_handler.current_state = Mock(side_effect=[_SyncState('quorum', 1, CaseInsensitiveSet(['foo']),
+                                                                         CaseInsensitiveSet(),
                                                                          CaseInsensitiveSet(['other'])),
-                                                              _SyncState('quorum', 1, 1, CaseInsensitiveSet(['foo']),
+                                                              _SyncState('quorum', 1, CaseInsensitiveSet(['foo']),
+                                                                         CaseInsensitiveSet(['foo']),
                                                                          CaseInsensitiveSet(['foo']))])
         mock_write_sync = self.ha.dcs.write_sync_state = Mock(return_value=SyncState(1, 'leader', 'foo', 0))
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'foo'))
@@ -1827,8 +1831,9 @@ def test__process_quorum_replication(self):
         self.assertEqual(mock_set_sync.call_args_list[0][0], ('ANY 1 (other)',))
 
         # Test ANY 1 (*) when synchronous_mode_strict and no nodes available
-        self.p.sync_handler.current_state = Mock(return_value=_SyncState('quorum', 1, 0,
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('quorum', 1,
                                                                          CaseInsensitiveSet(['other', 'foo']),
+                                                                         CaseInsensitiveSet(),
                                                                          CaseInsensitiveSet()))
         mock_write_sync.reset_mock()
         mock_set_sync.reset_mock()