More use of CaseInsensitiveSet (#2631)

1. make `SyncHandler.current_state()` return `CaseInsensitiveSet` instead of `list` objects.
2. take `sync_node_count` and `sync_node_maxlag` from the `Postgresql._global_config` instead of passing them as arguments.
3. Make `AbstractDCS.write_sync_state()` accept any `Collection`-like objects.

patroni/collections.py

@@ -35,7 +35,7 @@ class CaseInsensitiveSet(MutableSet):
     def discard(self, value: str) -> None:
         self._values.pop(value.lower(), None)
 
-    def issubset(self, other: 'CaseInsensitiveSet'):
+    def issubset(self, other: 'CaseInsensitiveSet') -> bool:
         return self <= other
 
 

patroni/config.py

@@ -107,10 +107,15 @@ class GlobalConfig(object):
         ret = parse_int(self.get(name))
         return default if ret is None else ret
 
+    @property
+    def min_synchronous_nodes(self) -> int:
+        """:returns: the minimal number of synchronous nodes based on whether strict mode is requested or not."""
+        return 1 if self.is_synchronous_mode_strict else 0
+
     @property
     def synchronous_node_count(self) -> int:
         """:returns: currently configured value from the global configuration or 1 if it is not set or invalid."""
-        return self.get_int('synchronous_node_count', 0)
+        return max(self.get_int('synchronous_node_count', 1), self.min_synchronous_nodes)
 
     @property
     def maximum_lag_on_failover(self) -> int:

patroni/dcs/__init__.py

@@ -14,7 +14,7 @@ from collections import defaultdict, namedtuple
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
-from typing import Any, Dict, List, Optional, Union
+from typing import Any, Collection, Dict, List, Optional, Union
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 from ..exceptions import PatroniFatalException
@@ -1024,13 +1024,24 @@ class AbstractDCS(abc.ABC):
         """Delete cluster from DCS"""
 
     @staticmethod
-    def sync_state(leader, sync_standby):
-        """Build sync_state dict
-           sync_standby dictionary key being kept for backward compatibility
+    def sync_state(leader: Union[str, None], sync_standby: Union[Collection[str], None]) -> Dict[str, Any]:
+        """Build sync_state dict.
+        The sync_standby key being kept for backward compatibility.
+        :param leader: name of the leader node that manages /sync key
+        :param sync_standby: collection of currently known synchronous standby node names
+        :returns: dictionary that later could be serialized to JSON or saved directly to DCS
         """
-        return {'leader': leader, 'sync_standby': sync_standby and ','.join(sorted(sync_standby)) or None}
+        return {'leader': leader, 'sync_standby': ','.join(sorted(sync_standby)) if sync_standby else None}
 
-    def write_sync_state(self, leader, sync_standby, index=None):
+    def write_sync_state(self, leader: Union[str, None], sync_standby: Union[Collection[str], None],
+                         index: Optional[Union[int, str]] = None) -> bool:
+        """Write the new synchronous state to DCS.
+        Calls :func:`sync_state` method to build a dict and than calls DCS specific :func:`set_sync_state_value` method.
+        :param leader: name of the leader node that manages /sync key
+        :param sync_standby: collection of currently known synchronous standby node names
+        :param index: for conditional update of the key/object
+        :returns: `True` if /sync key was successfully updated
+        """
         sync_value = self.sync_state(leader, sync_standby)
         return self.set_sync_state_value(json.dumps(sync_value, separators=(',', ':')), index)
 
@@ -1039,8 +1050,13 @@ class AbstractDCS(abc.ABC):
         """"""
 
     @abc.abstractmethod
-    def set_sync_state_value(self, value, index=None):
-        """"""
+    def set_sync_state_value(self, value: str, index: Optional[Union[int, str]] = None) -> bool:
+        """Set synchronous state in DCS, should be implemented in the child class.
+
+        :param value: the new value of /sync key
+        :param index: for conditional update of the key/object
+        :returns: `True` if key/object was successfully updated
+        """
 
     @abc.abstractmethod
     def delete_sync_state(self, index=None):

patroni/dcs/kubernetes.py

@@ -16,7 +16,7 @@ from collections import defaultdict
 from copy import deepcopy
 from http.client import HTTPException
 from threading import Condition, Lock, Thread
-from typing import Any, Dict, List, Optional
+from typing import Any, Collection, Dict, List, Optional, Union
 from urllib3.exceptions import HTTPError
 
 from . import AbstractDCS, Cluster, ClusterConfig, Failover, Leader, Member, SyncState,\
@@ -1267,11 +1267,26 @@ class Kubernetes(AbstractDCS):
     def set_sync_state_value(self, value, index=None):
         """Unused"""
 
-    def write_sync_state(self, leader, sync_standby, index=None):
-        return self.patch_or_create(self.sync_path, self.sync_state(leader, sync_standby), index, False)
+    def write_sync_state(self, leader: Union[str, None], sync_standby: Union[Collection[str], None],
+                         index: Optional[Union[int, str]] = None) -> bool:
+        """Prepare and write annotations to $SCOPE-sync Endpoint or ConfigMap.
 
-    def delete_sync_state(self, index=None):
-        return self.write_sync_state(None, None, index)
+        :param leader: name of the leader node that manages /sync key
+        :param sync_standby: collection of currently known synchronous standby node names
+        :param index: last known `resource_version` for conditional update of the object
+        :returns: `True` if update was successful
+        """
+        sync_state = self.sync_state(leader, sync_standby)
+        return self.patch_or_create(self.sync_path, sync_state, index, False)
+
+    def delete_sync_state(self, index: Optional[str] = None) -> bool:
+        """Patch annotations of $SCOPE-sync Endpoint or ConfigMap with empty values.
+
+        Effectively it removes "leader" and "sync_standby" annotations from the object.
+        :param index: last known `resource_version` for conditional update of the object
+        :returns: `True` if "delete" was successful
+        """
+        return self.write_sync_state(None, None, index=index)
 
     def watch(self, leader_index, timeout):
         if self.__do_not_watch:

patroni/ha.py

@@ -13,6 +13,7 @@ from typing import List, Optional, Union
 
 from . import psycopg
 from .async_executor import AsyncExecutor, CriticalTask
+from .collections import CaseInsensitiveSet
 from .exceptions import DCSError, PostgresConnectionException, PatroniFatalException
 from .postgresql.callback_executor import CallbackAction
 from .postgresql.misc import postgres_version_to_int
@@ -472,7 +473,7 @@ class Ha(object):
                 node_to_follow = self._get_node_to_follow(self.cluster)
 
             if self.is_synchronous_mode():
-                self.state_handler.sync_handler.set_synchronous_standby_names([])
+                self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet())
         elif self.has_lock():
             msg = "starting as readonly because i had the session lock"
             node_to_follow = None
@@ -568,7 +569,7 @@ class Ha(object):
         """:returns: `True` if failsafe_mode is enabled in global configuration."""
         return self.global_config.check_mode('failsafe_mode')
 
-    def process_sync_replication(self):
+    def process_sync_replication(self) -> None:
         """Process synchronous standby beahvior.
 
         Synchronous standbys are registered in two places postgresql.conf and DCS. The order of updating them must
@@ -578,36 +579,34 @@ class Ha(object):
         promoting standbys that were guaranteed to be replicating synchronously.
         """
         if self.is_synchronous_mode():
-            sync_node_count = self.global_config.synchronous_node_count
-            sync_node_maxlag = self.global_config.maximum_lag_on_syncnode
-            current = [] if self.cluster.sync.is_empty else self.cluster.sync.members
-            picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count,
-                                                                                  sync_node_maxlag)
-            if set(picked) != set(current):
+            current = CaseInsensitiveSet(self.cluster.sync.members)
+            picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster)
+
+            if picked != current:
                 # update synchronous standby list in dcs temporarily to point to common nodes in current and picked
-                sync_common = list(set(current).intersection(set(allow_promote)))
-                if set(sync_common) != set(current):
-                    logger.info("Updating synchronous privilege temporarily from %s to %s", current, sync_common)
-                    if not self.dcs.write_sync_state(self.state_handler.name,
-                                                     sync_common or None,
+                sync_common = current & allow_promote
+                if sync_common != current:
+                    logger.info("Updating synchronous privilege temporarily from %s to %s",
+                                list(current), list(sync_common))
+                    if not self.dcs.write_sync_state(self.state_handler.name, sync_common,
                                                      index=self.cluster.sync.index):
                         logger.info('Synchronous replication key updated by someone else.')
                         return
 
-                # Update  db param and wait for x secs
+                # When strict mode and no suitable replication connections put "*" to synchronous_standby_names
                 if self.global_config.is_synchronous_mode_strict and not picked:
-                    picked = ['*']
+                    picked = CaseInsensitiveSet('*')
                     logger.warning("No standbys available!")
 
+                # Update postgresql.conf and wait 2 secs for changes to become active
                 logger.info("Assigning synchronous standby status to %s", picked)
                 self.state_handler.sync_handler.set_synchronous_standby_names(picked)
 
-                if picked and picked[0] != '*' and set(allow_promote) != set(picked) and not allow_promote:
+                if picked and picked != CaseInsensitiveSet('*') and allow_promote != picked and not allow_promote:
                     # 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_node_count,
-                                                                                     sync_node_maxlag)
-                if allow_promote and set(allow_promote) != set(sync_common):
+                    _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster)
+                if allow_promote and allow_promote != sync_common:
                     try:
                         cluster = self.dcs.get_cluster()
                     except DCSError:
@@ -618,11 +617,11 @@ class Ha(object):
                     if not self.dcs.write_sync_state(self.state_handler.name, allow_promote, index=cluster.sync.index):
                         logger.info("Synchronous replication key updated by someone else")
                         return
-                    logger.info("Synchronous standby status assigned to %s", allow_promote)
+                    logger.info("Synchronous standby status assigned to %s", list(allow_promote))
         else:
             if not self.cluster.sync.is_empty and self.dcs.delete_sync_state(index=self.cluster.sync.index):
                 logger.info("Disabled synchronous replication")
-            self.state_handler.sync_handler.set_synchronous_standby_names([])
+            self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet())
 
     def is_sync_standby(self, cluster: Cluster) -> bool:
         """:returns: `True` if the current node is a synchronous standby."""
@@ -734,7 +733,7 @@ class Ha(object):
                     # promotion until next cycle. TODO: trigger immediate retry of run_cycle
                     return 'Postponing promotion because synchronous replication state was updated by somebody else'
                 self.state_handler.sync_handler.set_synchronous_standby_names(
-                    ['*'] if self.global_config.is_synchronous_mode_strict else [])
+                    CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
             if self.state_handler.role not in ('master', 'promoted', 'primary'):
                 def on_success():
                     self._rewind.reset_state()
@@ -1088,7 +1087,7 @@ class Ha(object):
                 node_to_follow, leader = None, None
 
         if self.is_synchronous_mode():
-            self.state_handler.sync_handler.set_synchronous_standby_names([])
+            self.state_handler.sync_handler.set_synchronous_standby_names(CaseInsensitiveSet())
 
         # FIXME: with mode offline called from DCS exception handler and handle_long_action_in_progress
         # there could be an async action already running, calling follow from here will lead

patroni/postgresql/__init__.py

@@ -12,7 +12,7 @@ from datetime import datetime
 from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
-from typing import Optional, Union, TYPE_CHECKING
+from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING
 
 from .bootstrap import Bootstrap
 from .callback_executor import CallbackAction, CallbackExecutor
@@ -30,7 +30,7 @@ from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
-    from .config import GlobalConfig
+    from ..config import GlobalConfig
 
 logger = logging.getLogger(__name__)
 
@@ -157,6 +157,11 @@ class Postgresql(object):
     def lsn_name(self):
         return 'lsn' if self._major_version >= 100000 else 'location'
 
+    @property
+    def supports_multiple_sync(self) -> bool:
+        """:returns: `True` if Postgres version supports more than one synchronous node."""
+        return self._major_version >= 90600
+
     @property
     def cluster_info_query(self):
         """Returns the monitoring query with a fixed number of fields.
@@ -419,13 +424,16 @@ class Postgresql(object):
     def received_timeline(self):
         return self._cluster_info_state_get('received_tli')
 
-    def synchronous_commit(self):
+    def synchronous_commit(self) -> str:
+        """:returns: "synchronous_commit" GUC value."""
         return self._cluster_info_state_get('synchronous_commit')
 
-    def synchronous_standby_names(self):
+    def synchronous_standby_names(self) -> str:
+        """:returns: "synchronous_standby_names" GUC value."""
         return self._cluster_info_state_get('synchronous_standby_names')
 
-    def pg_stat_replication(self):
+    def pg_stat_replication(self) -> List[Dict[str, Any]]:
+        """:returns: a result set of 'SELECT * FROM pg_stat_replication'."""
         return self._cluster_info_state_get('pg_stat_replication') or []
 
     def is_leader(self):
@@ -924,7 +932,8 @@ class Postgresql(object):
             self._cached_replica_timeline = self.get_replica_timeline()
         return self._cached_replica_timeline
 
-    def get_primary_timeline(self):
+    def get_primary_timeline(self) -> int:
+        """:returns: current timeline if postgres is running as a primary or 0."""
         return self._cluster_info_state_get('timeline')
 
     def get_history(self, timeline):

patroni/postgresql/sync.py

@@ -3,9 +3,13 @@ import re
 import time
 
 from copy import deepcopy
+from typing import Any, Collection, Dict, Tuple, TYPE_CHECKING
 
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
+from ..dcs import Cluster
 from ..psycopg import quote_ident as _quote_ident
+if TYPE_CHECKING:  # pragma: no cover
+    from . import Postgresql
 
 logger = logging.getLogger(__name__)
 
@@ -26,12 +30,12 @@ SYNC_REP_PARSER_RE = re.compile(r"""
 _EMPTY_SSN = {'type': 'off', 'num': 0, 'members': CaseInsensitiveSet()}
 
 
-def quote_ident(value):
-    """Very simplified version of quote_ident"""
+def quote_ident(value: str) -> str:
+    """Very simplified version of `psycopg` :func:`quote_ident` function."""
     return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value)
 
 
-def parse_sync_standby_names(value):
+def parse_sync_standby_names(value: str) -> Dict[str, Any]:
     """Parse postgresql synchronous_standby_names to constituent parts.
     Returns dict with the following keys:
     * type: 'quorum'|'priority'
@@ -137,7 +141,7 @@ class SyncHandler(object):
     and the `current_state()` method will count newly added names as "sync" only when
     they reached memorized LSN and also reported as "sync" by `pg_stat_replication`"""
 
-    def __init__(self, postgresql):
+    def __init__(self, postgresql: 'Postgresql') -> None:
         self._postgresql = postgresql
         self._synchronous_standby_names = ''  # last known value of synchronous_standby_names
         self._ssn_data = deepcopy(_EMPTY_SSN)
@@ -169,18 +173,21 @@ 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, sync_node_count=1, sync_node_maxlag=-1):
+    def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
         """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
         synchronous standby any longer.
-        Parameter sync_node_maxlag(maximum_lag_on_syncnode) would help swapping unhealthy sync replica in case
-        if it stops responding (or hung). Please set the value high enough so it won't unncessarily swap sync
-        standbys during high loads. Any less or equal of 0 value keep the behavior backward compatible and
-        will not swap. Please note that it will not also swap sync standbys in case where all replicas are hung.
 
-        :returns: tuple of candidates list and synchronous standby list."""
+        Standbys are selected based on values from the global configuration:
+        - `maximum_lag_on_syncnode`: would help swapping unhealthy sync replica in case if it stops
+          responding (or hung). Please set the value high enough so it won't unncessarily swap sync
+          standbys during high loads. Any value less or equal of 0 keeps the behavior backward compatible.
+          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`.
+        """
         self._handle_synchronous_standby_names_change()
 
         # Pick candidates based on who has higher replay/remote_write/flush lsn.
@@ -209,11 +216,13 @@ class SyncHandler(object):
         max_lsn = max(replica_list, key=lambda x: x[3])[3]\
             if len(replica_list) > 1 else self._postgresql.last_operation()
 
-        if self._postgresql.major_version < 90600:
-            sync_node_count = 1
+        assert self._postgresql._global_config is not None
+        sync_node_count = self._postgresql._global_config.synchronous_node_count\
+            if self._postgresql.supports_multiple_sync else 1
+        sync_node_maxlag = self._postgresql._global_config.maximum_lag_on_syncnode
 
-        candidates = []
-        sync_nodes = []
+        candidates = CaseInsensitiveSet()
+        sync_nodes = CaseInsensitiveSet()
         # 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
@@ -223,28 +232,32 @@ class SyncHandler(object):
                 self._ready_replicas[app_name] = pid
 
             if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
-                candidates.append(app_name)
+                candidates.add(app_name)
                 if sync_state == 'sync' and app_name in self._ready_replicas:
-                    sync_nodes.append(app_name)
+                    sync_nodes.add(app_name)
             if len(candidates) >= sync_node_count:
                 break
 
         return candidates, sync_nodes
 
-    def set_synchronous_standby_names(self, value):
-        """Constructs and sets `synchronous_standby_names` value.
+    def set_synchronous_standby_names(self, sync: Collection[str]) -> None:
+        """Constructs and sets "synchronous_standby_names" GUC value.
 
-        :param value: list[str] - the list of wanted sync members"""
-        if value and value != ['*']:
-            value = [quote_ident(x) for x in value]
-
-        if self._postgresql.major_version >= 90600 and len(value) > 1:
-            sync_param = '{0} ({1})'.format(len(value), ','.join(value))
+        :param sync: set of nodes to sync to
+        """
+        has_asterisk = '*' in sync
+        if has_asterisk:
+            sync = ['*']
         else:
-            sync_param = next(iter(value), None)
+            sync = [quote_ident(x) for x in sync]
+
+        if self._postgresql.supports_multiple_sync and len(sync) > 1:
+            sync_param = '{0} ({1})'.format(len(sync), ','.join(sync))
+        else:
+            sync_param = next(iter(sync), None)
 
         if not (self._postgresql.config.set_synchronous_standby_names(sync_param)
-                and self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']:
+                and self._postgresql.state == 'running' and self._postgresql.is_leader()) or has_asterisk:
             return
 
         time.sleep(0.1)  # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms

tests/test_ha.py

@@ -4,6 +4,7 @@ import os
 import sys
 
 from mock import Mock, MagicMock, PropertyMock, patch, mock_open
+from patroni.collections import CaseInsensitiveSet
 from patroni.config import Config
 from patroni.dcs import Cluster, ClusterConfig, Failover, Leader, Member, get_dcs, SyncState, TimelineHistory
 from patroni.dcs.etcd import AbstractEtcdClientWithFailover
@@ -795,7 +796,7 @@ class TestHa(PostgresInit):
         # manual failover when the `other` node isn't available but our name is in the /sync key
         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=([], []))
+        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(), CaseInsensitiveSet()))
         self.ha.dcs.write_sync_state = true
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
@@ -811,7 +812,8 @@ class TestHa(PostgresInit):
         self.ha.cluster = get_cluster_initialized_without_leader(failover=Failover(0, '', 'postgresql0', None),
                                                                  sync=('leader1', 'other'))
         self.p.set_role('replica')
-        self.p.sync_handler.current_state = Mock(return_value=(['leader1'], ['leader1']))
+        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_failover_process_no_leader_in_pause(self):
@@ -1112,7 +1114,7 @@ class TestHa(PostgresInit):
         with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync:
             self.ha.run_cycle()
             mock_delete_sync.assert_called_once()
-            mock_set_sync.assert_called_once_with([])
+            mock_set_sync.assert_called_once_with(CaseInsensitiveSet())
 
         mock_set_sync.reset_mock()
         # Test sync key not touched when not there
@@ -1120,14 +1122,15 @@ class TestHa(PostgresInit):
         with patch.object(self.ha.dcs, 'delete_sync_state') as mock_delete_sync:
             self.ha.run_cycle()
             mock_delete_sync.assert_not_called()
-            mock_set_sync.assert_called_once_with([])
+            mock_set_sync.assert_called_once_with(CaseInsensitiveSet())
 
         mock_set_sync.reset_mock()
 
         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=(['other'], ['other']))
+        self.p.sync_handler.current_state = Mock(return_value=(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,17 +1138,18 @@ 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=(['other2'], []))
+        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2']), CaseInsensitiveSet()))
         self.ha.dcs.write_sync_state = Mock(return_value=True)
         self.ha.run_cycle()
-        mock_set_sync.assert_called_once_with(['other2'])
+        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=(['other2', 'other3'], ['other2']))
+        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2', 'other3']),
+                                                               CaseInsensitiveSet(['other2'])))
         self.ha.dcs.write_sync_state = Mock(return_value=True)
         self.ha.run_cycle()
-        self.assertEqual(mock_set_sync.call_args_list[0][0], (['other2'],))
-        self.assertEqual(mock_set_sync.call_args_list[1][0], (['other2', 'other3'],))
+        self.assertEqual(mock_set_sync.call_args_list[0][0], (CaseInsensitiveSet(['other2']),))
+        self.assertEqual(mock_set_sync.call_args_list[1][0], (CaseInsensitiveSet(['other2', 'other3']),))
 
         mock_set_sync.reset_mock()
         # Test sync standby is not disabled when updating dcs fails
@@ -1158,7 +1162,8 @@ class TestHa(PostgresInit):
         self.ha.dcs.write_sync_state = Mock(return_value=True)
         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=(['other2'], ['other2']))
+        self.p.sync_handler.current_state = Mock(return_value=(CaseInsensitiveSet(['other2']),
+                                                               CaseInsensitiveSet(['other2'])))
         self.ha.run_cycle()
         self.ha.dcs.get_cluster.assert_called_once()
         self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2)
@@ -1180,10 +1185,10 @@ 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=([], []))
+        self.p.sync_handler.current_state = Mock(return_value=(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(['*'])
+        mock_set_sync.assert_called_once_with(CaseInsensitiveSet('*'))
 
     def test_sync_replication_become_primary(self):
         self.ha.is_synchronous_mode = true
@@ -1198,7 +1203,7 @@ class TestHa(PostgresInit):
 
         # When we just became primary nobody is sync
         self.assertEqual(self.ha.enforce_primary_role('msg', 'promote msg'), 'promote msg')
-        mock_set_sync.assert_called_once_with([])
+        mock_set_sync.assert_called_once_with(CaseInsensitiveSet())
         mock_write_sync.assert_called_once_with('leader', None, index=0)
 
         mock_set_sync.reset_mock()

tests/test_sync.py

@@ -2,6 +2,8 @@ import os
 
 from mock import Mock, patch
 
+from patroni.collections import CaseInsensitiveSet
+from patroni.config import GlobalConfig
 from patroni.dcs import Cluster, SyncState
 from patroni.postgresql import Postgresql
 
@@ -20,6 +22,7 @@ class TestSync(BaseTestPostgresql):
     def setUp(self):
         super(TestSync, self).setUp()
         self.p.config.write_postgresql_conf()
+        self.p._global_config = GlobalConfig({'synchronous_mode': True})
         self.s = self.p.sync_handler
 
     @patch.object(Postgresql, 'last_operation', Mock(return_value=1))
@@ -35,32 +38,34 @@ 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), ([self.leadermem.name], [self.leadermem.name]))
+            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]),
+                                                             CaseInsensitiveSet([self.leadermem.name])))
 
         # prefer node with sync_state='potential', even if it is slightly behind of async
         pg_stat_replication[0]['sync_state'] = 'potential'
         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), ([self.leadermem.name], []))
+            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.leadermem.name]),
+                                                             CaseInsensitiveSet()))
 
         # 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), ([self.me.name], []))
+            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet()))
 
         # 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), ([self.me.name], []))
+            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet([self.me.name]), CaseInsensitiveSet()))
 
         # 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), ([], []))
+            self.assertEqual(self.s.current_state(cluster), (CaseInsensitiveSet(), CaseInsensitiveSet()))
 
     def test_set_sync_standby(self):
         def value_in_conf():
@@ -70,20 +75,26 @@ class TestSync(BaseTestPostgresql):
                         return line.strip()
 
         mock_reload = self.p.reload = Mock()
-        self.s.set_synchronous_standby_names(['n1'])
+        self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1']))
         self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
         mock_reload.assert_called()
 
         mock_reload.reset_mock()
-        self.s.set_synchronous_standby_names(['n1'])
+        self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1']))
         mock_reload.assert_not_called()
         self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
 
-        self.s.set_synchronous_standby_names(['n1', 'n2'])
+        self.s.set_synchronous_standby_names(CaseInsensitiveSet(['n1', 'n2']))
         mock_reload.assert_called()
         self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'")
 
         mock_reload.reset_mock()
-        self.s.set_synchronous_standby_names([])
+        self.s.set_synchronous_standby_names(CaseInsensitiveSet([]))
         mock_reload.assert_called()
         self.assertEqual(value_in_conf(), None)
+
+        mock_reload.reset_mock()
+        self.p._global_config = GlobalConfig({'synchronous_mode': True})
+        self.s.set_synchronous_standby_names(CaseInsensitiveSet('*'))
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = '*'")