Adapt SyncHandler interfaces for quorum commit

patroni/config.py

@@ -77,10 +77,15 @@ class GlobalConfig(object):
         """:returns: `True` if cluster is in maintenance mode."""
         return self.check_mode('pause')
 
+    @property
+    def is_quorum_commit_mode(self) -> bool:
+        """:returns: `True` if quorum commit replication is requested"""
+        return str(self.get('synchronous_mode')).lower() == 'quorum'
+
     @property
     def is_synchronous_mode(self) -> bool:
         """:returns: `True` if synchronous replication is requested."""
-        return self.check_mode('synchronous_mode')
+        return self.check_mode('synchronous_mode') is True or self.is_quorum_commit_mode
 
     @property
     def is_synchronous_mode_strict(self) -> bool:

patroni/ha.py

@@ -586,7 +586,9 @@ class Ha(object):
         promoting standbys that were guaranteed to be replicating synchronously.
         """
         if self.is_synchronous_mode():
-            picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster)
+            current_state = self.state_handler.sync_handler.current_state(self.cluster)
+            picked = current_state.active
+            allow_promote = current_state.sync
             voters = CaseInsensitiveSet(self.cluster.sync.voters)
 
             if picked != voters:
@@ -612,7 +614,7 @@ class Ha(object):
                 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)
+                    allow_promote = self.state_handler.sync_handler.current_state(self.cluster).sync
                 if allow_promote and allow_promote != sync_common:
                     if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, 0, index=sync.index):
                         return logger.info("Synchronous replication key updated by someone else")

patroni/postgresql/__init__.py

@@ -162,6 +162,11 @@ class Postgresql(object):
     def lsn_name(self) -> str:
         return 'lsn' if self._major_version >= 100000 else 'location'
 
+    @property
+    def supports_quorum_commit(self) -> bool:
+        """:returns: `True` if quorum commit is supported by Postgres."""
+        return self._major_version >= 100000
+
     @property
     def supports_multiple_sync(self) -> bool:
         """:returns: `True` if Postgres version supports more than one synchronous node."""

patroni/postgresql/sync.py

@@ -3,7 +3,7 @@ import re
 import time
 
 from copy import deepcopy
-from typing import Collection, List, NamedTuple, Tuple, TYPE_CHECKING
+from typing import Collection, List, NamedTuple, Optional, Tuple, TYPE_CHECKING
 
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Cluster
@@ -153,6 +153,14 @@ def parse_sync_standby_names(value: str) -> _SSN:
     return _SSN(sync_type, has_star, num, members)
 
 
+class _SyncState(NamedTuple):
+    sync_type: str
+    numsync: int
+    numsync_confirmed: int
+    sync: CaseInsensitiveSet
+    active: CaseInsensitiveSet
+
+
 class SyncHandler(object):
     """Class responsible for working with the `synchronous_standby_names`.
 
@@ -196,7 +204,7 @@ class SyncHandler(object):
         self._postgresql.query('SELECT pg_catalog.txid_current()')  # Ensure some WAL traffic to move replication
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
-    def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
+    def current_state(self, cluster: Cluster) -> _SyncState:
         """Finds best candidates to be the synchronous standbys.
 
         Current synchronous standby is always preferred, unless it has disconnected or does not want to be a
@@ -209,7 +217,8 @@ class SyncHandler(object):
           Please note that it will not also swap sync standbys in case where all replicas are hung.
         - `synchronous_node_count`: controlls how many nodes should be set as synchronous.
 
-        :returns: tuple of candidates :class:`CaseInsensitiveSet` and synchronous standbys :class:`CaseInsensitiveSet`.
+        :param cluster: current cluster topology from DCS
+        :returns: current synchronous replication state as a :class:`_SyncState` object
         """
         self._handle_synchronous_standby_names_change()
 
@@ -245,41 +254,79 @@ class SyncHandler(object):
             if self._postgresql.supports_multiple_sync else 1
         sync_node_maxlag = self._postgresql._global_config.maximum_lag_on_syncnode
 
-        candidates = CaseInsensitiveSet()
+        active = CaseInsensitiveSet()
         sync_nodes = CaseInsensitiveSet()
+        numsync_confirmed = 0
         # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
-        for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]):
-            # if standby name is listed in the /sync key we can count it as synchronous, otherwice
-            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
-            if app_name not in self._ready_replicas and app_name in self._ssn_data.members and\
-                    (cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn):
-                self._ready_replicas[app_name] = pid
+        for pid, app_name, sync_state, replica_lsn, nofailover in sorted(replica_list, key=lambda x: x[4]):
+            if app_name not in self._ready_replicas and app_name in self._ssn_data.members:
+                if self._postgresql._global_config.is_quorum_commit_mode:
+                    # When quorum commit is enabled we can't check against cluster.sync because nodes
+                    # are written there when at least one of them caught up with _primary_flush_lsn.
+                    if replica_lsn >= self._primary_flush_lsn\
+                            and (sync_state == 'quorum' or (not self._postgresql.supports_quorum_commit
+                                                            and sync_state in ('sync', 'potential'))):
+                        self._ready_replicas[app_name] = pid
+                elif cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn:
+                    # if standby name is listed in the /sync key we can count it as synchronous, otherwise it becomes
+                    # "really" synchronous when sync_state = 'sync' and we known that it managed to catch up
+                    self._ready_replicas[app_name] = pid
 
             if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
-                candidates.add(app_name)
-                if sync_state == 'sync' and app_name in self._ready_replicas:
-                    sync_nodes.add(app_name)
-            if len(candidates) >= sync_node_count:
-                break
+                if self._postgresql._global_config.is_quorum_commit_mode:
+                    # add nodes with nofailover tag only to get enough "active" nodes
+                    if not nofailover or len(active) < sync_node_count:
+                        if app_name in self._ready_replicas:
+                            numsync_confirmed += 1
+                        active.add(app_name)
+                else:
+                    active.add(app_name)
+                    if sync_state == 'sync' and app_name in self._ready_replicas:
+                        sync_nodes.add(app_name)
+                        numsync_confirmed += 1
+                    if len(active) >= sync_node_count:
+                        break
 
-        return candidates, sync_nodes
+        if self._postgresql._global_config.is_quorum_commit_mode:
+            sync_nodes = CaseInsensitiveSet() if self._ssn_data.has_star else self._ssn_data.members
 
-    def set_synchronous_standby_names(self, sync: Collection[str]) -> None:
+        return _SyncState(
+            self._ssn_data.sync_type,
+            0 if self._ssn_data.has_star else self._ssn_data.num,
+            numsync_confirmed,
+            sync_nodes,
+            active)
+
+    def set_synchronous_standby_names(self, sync: Collection[str], num: Optional[int] = None) -> None:
         """Constructs and sets "synchronous_standby_names" GUC value.
 
         :param sync: set of nodes to sync to
+        :param num: specifies number of nodes to sync to. The *num* is set only in case if quorum commit is enabled
         """
-        has_asterisk = '*' in sync
+        # Special case. If sync nodes set is empty but requested num of sync nodes >= 1
+        # we want to set synchronous_standby_names to '*'
+        has_asterisk = '*' in sync or num and num >= 1 and not sync
         if has_asterisk:
             sync = ['*']
         else:
-            sync = [quote_ident(x) for x in sync]
+            sync = [quote_ident(x) for x in sorted(sync)]
 
         if self._postgresql.supports_multiple_sync and len(sync) > 1:
-            sync_param = '{0} ({1})'.format(len(sync), ','.join(sync))
+            if num is None:
+                num = len(sync)
+            sync_param = ','.join(sync)
         else:
             sync_param = next(iter(sync), None)
 
+        if TYPE_CHECKING:  # pragma: no cover
+            assert self._postgresql._global_config is not None
+
+        if self._postgresql._global_config.is_quorum_commit_mode and sync or\
+                self._postgresql.supports_multiple_sync and len(sync) > 1:
+            prefix = 'ANY ' if self._postgresql._global_config.is_quorum_commit_mode\
+                and self._postgresql.supports_quorum_commit else ''
+            sync_param = '{0}{1} ({2})'.format(prefix, num, sync_param)
+
         if not (self._postgresql.config.set_synchronous_standby_names(sync_param)
                 and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk:
             return

tests/test_ha.py

@@ -17,6 +17,7 @@ from patroni.postgresql.config import ConfigHandler
 from patroni.postgresql.postmaster import PostmasterProcess
 from patroni.postgresql.rewind import Rewind
 from patroni.postgresql.slots import SlotsHandler
+from patroni.postgresql.sync import _SyncState
 from patroni.utils import tzutc
 from patroni.watchdog import Watchdog
 
@@ -1126,8 +1127,9 @@ class TestHa(PostgresInit):
         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=(CaseInsensitiveSet(['other']),
-                                                               CaseInsensitiveSet(['other'])))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+                                                                         CaseInsensitiveSet(['other']),
+                                                                         CaseInsensitiveSet(['other'])))
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
         self.ha.run_cycle()
         mock_set_sync.assert_not_called()
@@ -1135,14 +1137,16 @@ class TestHa(PostgresInit):
         mock_set_sync.reset_mock()
 
         # Test sync standby is replaced when switching standbys
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2']), CaseInsensitiveSet()))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, 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']))
 
         # Test sync standby is replaced when new standby is joined
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2', 'other3']),
-                                                               CaseInsensitiveSet(['other2'])))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+                                                                         CaseInsensitiveSet(['other2']),
+                                                                         CaseInsensitiveSet(['other2', 'other3'])))
         self.ha.dcs.write_sync_state = Mock(return_value=SyncState.empty())
         self.ha.run_cycle()
         self.assertEqual(mock_set_sync.call_args_list[0][0], (CaseInsensitiveSet(['other2']),))
@@ -1159,8 +1163,9 @@ class TestHa(PostgresInit):
         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=(CaseInsensitiveSet(['other2']),
-                                                               CaseInsensitiveSet(['other2'])))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 1, 1,
+                                                                         CaseInsensitiveSet(['other2']),
+                                                                         CaseInsensitiveSet(['other2'])))
         self.ha.run_cycle()
         self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2)
 
@@ -1182,7 +1187,8 @@ class TestHa(PostgresInit):
 
         # Test sync set to '*' when synchronous_mode_strict is enabled
         mock_set_sync.reset_mock()
-        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
+        self.p.sync_handler.current_state = Mock(return_value=_SyncState('priority', 0, 0, CaseInsensitiveSet(),
+                                                                         CaseInsensitiveSet()))
         with patch('patroni.config.GlobalConfig.is_synchronous_mode_strict', PropertyMock(return_value=True)):
             self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(CaseInsensitiveSet('*'))

tests/test_sync.py

@@ -38,7 +38,8 @@ class TestSync(BaseTestPostgresql):
         # sync node is a bit behind of async, but we prefer it anyway
         with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[self.leadermem.name,
                                                                               'on', pg_stat_replication]):
-            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]),
+            self.assertEqual(self.s.current_state(cluster), ('priority', 1, 1,
+                                                             CaseInsensitiveSet([self.leadermem.name]),
                                                              CaseInsensitiveSet([self.leadermem.name])))
 
         # prefer node with sync_state='potential', even if it is slightly behind of async
@@ -46,26 +47,46 @@ class TestSync(BaseTestPostgresql):
         for r in pg_stat_replication:
             r['write_lsn'] = r.pop('flush_lsn')
         with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_write', pg_stat_replication]):
-            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]),
-                                                             CaseInsensitiveSet()))
+            self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(),
+                                                             CaseInsensitiveSet([self.leadermem.name])))
 
         # when there are no sync or potential candidates we pick async with the minimal replication lag
         for i, r in enumerate(pg_stat_replication):
             r.update(replay_lsn=3 - i, application_name=r['application_name'].upper())
         missing = pg_stat_replication.pop(0)
         with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_apply', pg_stat_replication]):
-            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet()))
+            self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(),
+                                                             CaseInsensitiveSet([self.me.name])))
 
         # unknown sync node is ignored
         missing.update(application_name='missing', sync_state='sync')
         pg_stat_replication.insert(0, missing)
         with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['', 'remote_apply', pg_stat_replication]):
-            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet()))
+            self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(),
+                                                             CaseInsensitiveSet([self.me.name])))
 
         # invalid synchronous_standby_names and empty pg_stat_replication
         with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['a b', 'remote_apply', None]):
             self.p._major_version = 90400
-            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet(), CaseInsensitiveSet()))
+            self.assertEqual(self.s.current_state(cluster), ('off', 0, 0, CaseInsensitiveSet(), CaseInsensitiveSet()))
+
+    @patch.object(Postgresql, 'last_operation', Mock(return_value=1))
+    def test_current_state_quorum(self):
+        self.p._global_config = GlobalConfig({'synchronous_mode': 'quorum'})
+        cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
+                          SyncState(0, self.me.name, self.leadermem.name, 0), None, None, None)
+
+        pg_stat_replication = [
+            {'pid': 100, 'application_name': self.leadermem.name, 'sync_state': 'quorum', 'flush_lsn': 1},
+            {'pid': 101, 'application_name': self.other.name, 'sync_state': 'quorum', 'flush_lsn': 2}]
+
+        # sync node is a bit behind of async, but we prefer it anyway
+        with patch.object(Postgresql, "_cluster_info_state_get",
+                          side_effect=['ANY 1 ({0},"{1}")'.format(self.leadermem.name, self.other.name),
+                                       'on', pg_stat_replication]):
+            self.assertEqual(self.s.current_state(cluster),
+                             ('quorum', 1, 2, CaseInsensitiveSet([self.other.name, self.leadermem.name]),
+                              CaseInsensitiveSet([self.leadermem.name, self.other.name])))
 
     def test_set_sync_standby(self):
         def value_in_conf():
@@ -84,6 +105,7 @@ class TestSync(BaseTestPostgresql):
         mock_reload.assert_not_called()
         self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
 
+        mock_reload.reset_mock()
         self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1', 'n2']))
         mock_reload.assert_called()
         self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'")
@@ -98,3 +120,35 @@ class TestSync(BaseTestPostgresql):
         self.s.set_synchronous_standby_names(CaseInsensitiveSet('*'))
         mock_reload.assert_called()
         self.assertEqual(value_in_conf(), "synchronous_standby_names = '*'")
+
+        self.p._global_config = GlobalConfig({'synchronous_mode': 'quorum'})
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names([], 1)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 1 (*)'")
+
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names(['a', 'b'], 1)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 1 (a,b)'")
+
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names(['a', 'b'], 3)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'ANY 3 (a,b)'")
+
+        self.p._major_version = 90601
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names([], 1)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = '1 (*)'")
+
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names(['a', 'b'], 1)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = '1 (a,b)'")
+
+        mock_reload.reset_mock()
+        self.s.set_synchronous_standby_names(['a', 'b'], 3)
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = '3 (a,b)'")