Enhanced sync connections check (#2524)

When `synchronous_standby_names` GUC is changed PostgreSQL nearly immediately starts reporting corresponding walsenders as synchronous, while in fact they maybe didn't reach this state yet. To mitigate this problem we memorize current flush lsn on the primary right after change of `synchronous_standby_names` got visible and use it as an additional check for walsenders.
The walsender will be counted as truly "sync" only when write/flush/replay_lsn on it reached memorized LSN and the `application_name` is known to be a part of `synchronous_standby_names`.

The size of PR mostly related to refactoring and moving the code responsible for working with `synchronous_standby_names` and `pg_stat_replication` to the dedicated file.
And `parse_sync_standby_names()` function was mostly copied from #672.

patroni/ha.py

@@ -458,7 +458,7 @@ class Ha(object):
                 node_to_follow = self._get_node_to_follow(self.cluster)
 
             if self.is_synchronous_mode():
-                self.state_handler.config.set_synchronous_standby([])
+                self.state_handler.sync_handler.set_synchronous_standby_names([])
         elif self.has_lock():
             msg = "starting as readonly because i had the session lock"
             node_to_follow = None
@@ -568,9 +568,9 @@ class Ha(object):
         if self.is_synchronous_mode():
             sync_node_count = self.patroni.config['synchronous_node_count']
             current = self.cluster.sync.leader and self.cluster.sync.members or []
-            picked, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster, sync_node_count,
-                                                                                self.patroni.config[
-                                                                                    'maximum_lag_on_syncnode'])
+            picked, allow_promote = self.state_handler.sync_handler.current_state(self.cluster, sync_node_count,
+                                                                                  self.patroni.config[
+                                                                                      'maximum_lag_on_syncnode'])
             if set(picked) != set(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)))
@@ -588,15 +588,15 @@ class Ha(object):
                     logger.warning("No standbys available!")
 
                 logger.info("Assigning synchronous standby status to %s", picked)
-                self.state_handler.config.set_synchronous_standby(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:
                     # Wait for PostgreSQL to enable synchronous mode and see if we can immediately set sync_standby
                     time.sleep(2)
-                    _, allow_promote = self.state_handler.pick_synchronous_standby(self.cluster,
-                                                                                   sync_node_count,
-                                                                                   self.patroni.config[
-                                                                                       'maximum_lag_on_syncnode'])
+                    _, allow_promote = self.state_handler.sync_handler.current_state(self.cluster,
+                                                                                     sync_node_count,
+                                                                                     self.patroni.config[
+                                                                                         'maximum_lag_on_syncnode'])
                 if allow_promote and set(allow_promote) != set(sync_common):
                     try:
                         cluster = self.dcs.get_cluster()
@@ -612,7 +612,7 @@ class Ha(object):
         else:
             if self.cluster.sync.leader and self.dcs.delete_sync_state(index=self.cluster.sync.index):
                 logger.info("Disabled synchronous replication")
-            self.state_handler.config.set_synchronous_standby([])
+            self.state_handler.sync_handler.set_synchronous_standby_names([])
 
     def is_sync_standby(self, cluster):
         return cluster.leader and cluster.sync.leader == cluster.leader.name \
@@ -721,7 +721,8 @@ class Ha(object):
                     # Somebody else updated sync state, it may be due to us losing the lock. To be safe, postpone
                     # 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.config.set_synchronous_standby(['*'] if self.is_synchronous_mode_strict() else [])
+                self.state_handler.sync_handler.set_synchronous_standby_names(
+                        ['*'] if self.is_synchronous_mode_strict() else [])
             if self.state_handler.role != 'master':
                 def on_success():
                     self._rewind.reset_state()
@@ -1044,7 +1045,7 @@ class Ha(object):
                 node_to_follow, leader = None, None
 
         if self.is_synchronous_mode():
-            self.state_handler.config.set_synchronous_standby([])
+            self.state_handler.sync_handler.set_synchronous_standby_names([])
 
         # 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

@@ -22,7 +22,7 @@ from .connection import Connection, get_connection_cursor
 from .misc import parse_history, parse_lsn, postgres_major_version_to_int
 from .postmaster import PostmasterProcess
 from .slots import SlotsHandler
-from .validator import CaseInsensitiveDict
+from .sync import SyncHandler
 from .. import psycopg
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
@@ -85,6 +85,7 @@ class Postgresql(object):
         self.__thread_ident = current_thread().ident
 
         self.slots_handler = SlotsHandler(self)
+        self.sync_handler = SyncHandler(self)
 
         self._callback_executor = CallbackExecutor()
         self.__cb_called = False
@@ -176,12 +177,12 @@ class Postgresql(object):
 
         extra = ", " + (("pg_catalog.current_setting('synchronous_commit'), " +
                          "pg_catalog.current_setting('synchronous_standby_names'), "
-                         "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT application_name, sync_state," +
+                         "(SELECT pg_catalog.json_agg(r.*) FROM (SELECT w.pid as pid, application_name, sync_state," +
                          " pg_catalog.pg_{0}_{1}_diff(write_{1}, '0/0')::bigint AS write_lsn," +
                          " pg_catalog.pg_{0}_{1}_diff(flush_{1}, '0/0')::bigint AS flush_lsn," +
                          " pg_catalog.pg_{0}_{1}_diff(replay_{1}, '0/0')::bigint AS replay_lsn " +
                          "FROM pg_catalog.pg_stat_get_wal_senders() w," +
-                         " pg_catalog.pg_stat_get_activity(pid)" +
+                         " pg_catalog.pg_stat_get_activity(w.pid)" +
                          " WHERE w.state = 'streaming') r)").format(self.wal_name, self.lsn_name)
                         if self._is_synchronous_mode and self.role == 'master' else "'on', '', NULL")
 
@@ -403,6 +404,15 @@ class Postgresql(object):
     def received_timeline(self):
         return self._cluster_info_state_get('received_tli')
 
+    def synchronous_commit(self):
+        return self._cluster_info_state_get('synchronous_commit')
+
+    def synchronous_standby_names(self):
+        return self._cluster_info_state_get('synchronous_standby_names')
+
+    def pg_stat_replication(self):
+        return self._cluster_info_state_get('pg_stat_replication') or []
+
     def is_leader(self):
         try:
             return bool(self._cluster_info_state_get('timeline'))
@@ -1112,54 +1122,6 @@ class Postgresql(object):
             logger.exception('Could not remove data directory %s', self._data_dir)
             self.move_data_directory()
 
-    def pick_synchronous_standby(self, cluster, sync_node_count=1, sync_node_maxlag=-1):
-        """Finds the best candidate to be the synchronous standby.
-
-        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.
-        """
-        if self._major_version < 90600:
-            sync_node_count = 1
-        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
-        candidates = []
-        sync_nodes = []
-        replica_list = []
-        # Pick candidates based on who has higher replay/remote_write/flush lsn.
-        synchronous_commit = self._cluster_info_state_get('synchronous_commit')
-        sort_col = {'remote_apply': 'replay', 'remote_write': 'write'}.get(synchronous_commit, 'flush') + '_lsn'
-        pg_stat_replication = [(r['application_name'], r['sync_state'], r[sort_col])
-                               for r in self._cluster_info_state_get('pg_stat_replication') or []
-                               if r[sort_col] is not None]
-        # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
-        # That is, alphabetically they are in the reversed order of priority.
-        # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
-        # consistent in case if a synchronous standby member is slowed down OR async node receiving
-        # changes faster than the sync member (very rare but possible).
-        # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
-        for app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: (r[1], r[2]), reverse=True):
-            member = members.get(app_name)
-            if member and member.is_running and not member.tags.get('nosync', False):
-                replica_list.append((member.name, sync_state, replica_lsn, bool(member.nofailover)))
-
-        max_lsn = max(replica_list, key=lambda x: x[2])[2] if len(replica_list) > 1 else self.last_operation()
-
-        # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
-        for app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[3]):
-            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
-                candidates.append(app_name)
-                if sync_state == 'sync':
-                    sync_nodes.append(app_name)
-            if len(candidates) >= sync_node_count:
-                break
-
-        return candidates, sync_nodes
-
     def schedule_sanity_checks_after_pause(self):
         """
             After coming out of pause we have to:

patroni/postgresql/config.py

@@ -12,21 +12,14 @@ from .validator import CaseInsensitiveDict, recovery_parameters,\
         transform_postgresql_parameter_value, transform_recovery_parameter_value
 from ..dcs import slot_name_from_member_name, RemoteMember
 from ..exceptions import PatroniFatalException
-from ..psycopg import quote_ident as _quote_ident
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, \
         validate_directory, is_subpath
 
 logger = logging.getLogger(__name__)
 
-SYNC_STANDBY_NAME_RE = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$')
 PARAMETER_RE = re.compile(r'([a-z_]+)\s*=\s*')
 
 
-def quote_ident(value):
-    """Very simplified version of quote_ident"""
-    return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value)
-
-
 def conninfo_uri_parse(dsn):
     ret = {}
     r = urlparse(dsn)
@@ -1044,23 +1037,19 @@ class ConfigHandler(object):
         else:
             logger.info('No PostgreSQL configuration items changed, nothing to reload.')
 
-    def set_synchronous_standby(self, sync_members):
-        """Sets a node to be synchronous standby and if changed does a reload for PostgreSQL."""
-        if sync_members and sync_members != ['*']:
-            sync_members = [quote_ident(x) for x in sync_members]
-        if self._postgresql.major_version >= 90600 and len(sync_members) > 1:
-            sync_param = '{0} ({1})'.format(len(sync_members), ','.join(sync_members))
-        else:
-            sync_param = next(iter(sync_members), None)
-        if sync_param != self._synchronous_standby_names:
-            if sync_param is None:
+    def set_synchronous_standby_names(self, value):
+        """Updates synchronous_standby_names and reloads if necessary.
+        :returns: True if value was updated."""
+        if value != self._synchronous_standby_names:
+            if value is None:
                 self._server_parameters.pop('synchronous_standby_names', None)
             else:
-                self._server_parameters['synchronous_standby_names'] = sync_param
-            self._synchronous_standby_names = sync_param
+                self._server_parameters['synchronous_standby_names'] = value
+            self._synchronous_standby_names = value
             if self._postgresql.state == 'running':
                 self.write_postgresql_conf()
                 self._postgresql.reload()
+            return True
 
     @property
     def effective_configuration(self):

patroni/postgresql/sync.py

@@ -0,0 +1,258 @@
+import logging
+import re
+import time
+
+from copy import deepcopy
+
+from .validator import CaseInsensitiveDict
+from ..psycopg import quote_ident as _quote_ident
+
+logger = logging.getLogger(__name__)
+
+SYNC_STANDBY_NAME_RE = re.compile(r'^[A-Za-z_][A-Za-z_0-9\$]*$')
+SYNC_REP_PARSER_RE = re.compile(r"""
+           (?P<first> [fF][iI][rR][sS][tT] )
+         | (?P<any> [aA][nN][yY] )
+         | (?P<space> \s+ )
+         | (?P<ident> [A-Za-z_][A-Za-z_0-9\$]* )
+         | (?P<dquot> " (?: [^"]+ | "" )* " )
+         | (?P<star> [*] )
+         | (?P<num> \d+ )
+         | (?P<comma> , )
+         | (?P<parenstart> \( )
+         | (?P<parenend> \) )
+         | (?P<JUNK> . )
+        """, re.X)
+_EMPTY_SSN = {'type': 'off', 'num': 0, 'members': CaseInsensitiveDict({})}
+
+
+def quote_ident(value):
+    """Very simplified version of quote_ident"""
+    return value if SYNC_STANDBY_NAME_RE.match(value) else _quote_ident(value)
+
+
+def parse_sync_standby_names(value):
+    """Parse postgresql synchronous_standby_names to constituent parts.
+    Returns dict with the following keys:
+    * type: 'quorum'|'priority'
+    * num: int
+    * members: CaseInsensitiveDict, with names as keys
+    * has_star: bool - Present if true
+    If the configuration value can not be parsed, raises a ValueError.
+
+    >>> parse_sync_standby_names('')['type']
+    'off'
+
+    >>> parse_sync_standby_names('FiRsT')['type']
+    'priority'
+
+    >>> parse_sync_standby_names('FiRsT')['members']
+    {'FiRsT': True}
+
+    >>> parse_sync_standby_names('"1"')['members']
+    {'1': True}
+
+    >>> parse_sync_standby_names(' a , b ')['members']
+    {'a': True, 'b': True}
+
+    >>> parse_sync_standby_names(' a , b ')['num']
+    1
+
+    >>> parse_sync_standby_names('ANY 4("a",*,b)')['has_star']
+    True
+
+    >>> parse_sync_standby_names('ANY 4("a",*,b)')['num']
+    4
+
+    >>> parse_sync_standby_names('1')  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError: Unparseable synchronous_standby_names value
+
+    >>> parse_sync_standby_names('a,')  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError: Unparseable synchronous_standby_names value
+
+    >>> parse_sync_standby_names('ANY 4("a" b,"c c")')  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError: Unparseable synchronous_standby_names value
+
+    >>> parse_sync_standby_names('FIRST 4("a",)')  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError: Unparseable synchronous_standby_names value
+
+    >>> parse_sync_standby_names('2 (,)')  # doctest: +IGNORE_EXCEPTION_DETAIL
+    Traceback (most recent call last):
+        ...
+    ValueError: Unparseable synchronous_standby_names value
+    """
+    tokens = [(m.lastgroup, m.group(0), m.start())
+              for m in SYNC_REP_PARSER_RE.finditer(value)
+              if m.lastgroup != 'space']
+    if not tokens:
+        return deepcopy(_EMPTY_SSN)
+
+    if [t[0] for t in tokens[0:3]] == ['any', 'num', 'parenstart'] and tokens[-1][0] == 'parenend':
+        result = {'type': 'quorum', 'num': int(tokens[1][1])}
+        synclist = tokens[3:-1]
+    elif [t[0] for t in tokens[0:3]] == ['first', 'num', 'parenstart'] and tokens[-1][0] == 'parenend':
+        result = {'type': 'priority', 'num': int(tokens[1][1])}
+        synclist = tokens[3:-1]
+    elif [t[0] for t in tokens[0:2]] == ['num', 'parenstart'] and tokens[-1][0] == 'parenend':
+        result = {'type': 'priority', 'num': int(tokens[0][1])}
+        synclist = tokens[2:-1]
+    else:
+        result = {'type': 'priority', 'num': 1}
+        synclist = tokens
+    result['members'] = CaseInsensitiveDict({})
+    for i, (a_type, a_value, a_pos) in enumerate(synclist):
+        if i % 2 == 1:  # odd elements are supposed to be commas
+            if len(synclist) == i + 1:  # except the last token
+                raise ValueError("Unparseable synchronous_standby_names value %r: Unexpected token %s %r at %d" %
+                                 (value, a_type, a_value, a_pos))
+            elif a_type != 'comma':
+                raise ValueError("Unparseable synchronous_standby_names value %r: ""Got token %s %r while"
+                                 " expecting comma at %d" % (value, a_type, a_value, a_pos))
+        elif a_type in {'ident', 'first', 'any'}:
+            result['members'][a_value] = True
+        elif a_type == 'star':
+            result['members'][a_value] = True
+            result['has_star'] = True
+        elif a_type == 'dquot':
+            result['members'][a_value[1:-1].replace('""', '"')] = True
+        else:
+            raise ValueError("Unparseable synchronous_standby_names value %r: Unexpected token %s %r at %d" %
+                             (value, a_type, a_value, a_pos))
+    return result
+
+
+class SyncHandler(object):
+    """Class responsible for working with the `synchronous_standby_names`.
+
+    Sync standbys are chosen based on their state in `pg_stat_replication`.
+    When `synchronous_standby_names` is changed we memorize the `_primary_flush_lsn`
+    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):
+        self._postgresql = postgresql
+        self._synchronous_standby_names = ''  # last known value of synchronous_standby_names
+        self._ssn_data = deepcopy(_EMPTY_SSN)
+        self._primary_flush_lsn = 0
+        # "sync" replication connections, that were verified to reach self._primary_flush_lsn at some point
+        self._ready_replicas = CaseInsensitiveDict({})  # keys: member names, values: connection pids
+
+    def _handle_synchronous_standby_names_change(self):
+        """If synchronous_standby_names has changed we need to check that newly added replicas
+        have reached self._primary_flush_lsn. Only after that they could be counted as sync."""
+        synchronous_standby_names = self._postgresql.synchronous_standby_names()
+        if synchronous_standby_names == self._synchronous_standby_names:
+            return False
+
+        self._synchronous_standby_names = synchronous_standby_names
+        try:
+            self._ssn_data = parse_sync_standby_names(synchronous_standby_names)
+        except ValueError as e:
+            logger.warning('%s', e)
+            self._ssn_data = deepcopy(_EMPTY_SSN)
+
+        # Invalidate cache of "sync" connections
+        for app_name in list(self._ready_replicas.keys()):
+            if app_name not in self._ssn_data['members']:
+                del self._ready_replicas[app_name]
+
+        # Newly connected replicas will be counted as sync only when reached self._primary_flush_lsn
+        self._primary_flush_lsn = self._postgresql.last_operation()
+        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):
+        """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."""
+
+        self._handle_synchronous_standby_names_change()
+
+        # Pick candidates based on who has higher replay/remote_write/flush lsn.
+        sort_col = {
+            'remote_apply': 'replay',
+            'remote_write': 'write'
+        }.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn'
+
+        pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col])
+                               for r in self._postgresql.pg_stat_replication()
+                               if r[sort_col] is not None]
+
+        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
+        replica_list = []
+        # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
+        # That is, alphabetically they are in the reversed order of priority.
+        # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
+        # consistent in case if a synchronous standby member is slowed down OR async node receiving
+        # changes faster than the sync member (very rare but possible).
+        # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
+        for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True):
+            member = members.get(app_name)
+            if member and member.is_running and not member.tags.get('nosync', False):
+                replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover)))
+
+        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
+
+        candidates = []
+        sync_nodes = []
+        # 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
+            # ig 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 and app_name in cluster.sync.members or
+                     sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn):
+                self._ready_replicas[app_name] = pid
+
+            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
+                candidates.append(app_name)
+                if sync_state == 'sync' and app_name in self._ready_replicas:
+                    sync_nodes.append(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.
+
+        :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))
+        else:
+            sync_param = next(iter(value), None)
+
+        if not (self._postgresql.config.set_synchronous_standby_names(sync_param) and
+                self._postgresql.state == 'running' and self._postgresql.is_leader()) or value == ['*']:
+            return
+
+        time.sleep(0.1)  # Usualy it takes 1ms to reload postgresql.conf, but we will give it 100ms
+
+        # Reset internal cache to query fresh values
+        self._postgresql.reset_cluster_info_state(None)
+
+        # timeline == 0 -- indicates that this is the replica, shoudn't ever happen
+        if self._postgresql.get_master_timeline() > 0:
+            self._handle_synchronous_standby_names_change()

tests/__init__.py

@@ -101,7 +101,9 @@ class MockCursor(object):
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2, 1, 0, False, 1, 1, None, None, [{"slot_name": "ls", "confirmed_flush_lsn": 12345}])]
+            self.results = [(1, 2, 1, 0, False, 1, 1, None, None,
+                             [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
+                             'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
         elif sql.startswith('SELECT pg_catalog.pg_postmaster_start_time'):

tests/test_ha.py

@@ -757,7 +757,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.pick_synchronous_standby = Mock(return_value=([], []))
+        self.p.sync_handler.current_state = Mock(return_value=([], []))
         self.ha.dcs.write_sync_state = true
         self.assertEqual(self.ha.run_cycle(), 'promoted self to leader by acquiring session lock')
 
@@ -773,7 +773,7 @@ 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.pick_synchronous_standby = Mock(return_value=(['leader1'], ['leader1']))
+        self.p.sync_handler.current_state = Mock(return_value=(['leader1'], ['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):
@@ -1062,7 +1062,7 @@ class TestHa(PostgresInit):
 
     def test_process_sync_replication(self):
         self.ha.has_lock = true
-        mock_set_sync = self.p.config.set_synchronous_standby = Mock()
+        mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
         self.p.name = 'leader'
 
         # Test sync key removed when sync mode disabled
@@ -1085,7 +1085,7 @@ class TestHa(PostgresInit):
         self.ha.is_synchronous_mode = true
 
         # Test sync standby not touched when picking the same node
-        self.p.pick_synchronous_standby = Mock(return_value=(['other'], ['other']))
+        self.p.sync_handler.current_state = Mock(return_value=(['other'], ['other']))
         self.ha.cluster = get_cluster_initialized_with_leader(sync=('leader', 'other'))
         self.ha.run_cycle()
         mock_set_sync.assert_not_called()
@@ -1093,13 +1093,13 @@ class TestHa(PostgresInit):
         mock_set_sync.reset_mock()
 
         # Test sync standby is replaced when switching standbys
-        self.p.pick_synchronous_standby = Mock(return_value=(['other2'], []))
+        self.p.sync_handler.current_state = Mock(return_value=(['other2'], []))
         self.ha.dcs.write_sync_state = Mock(return_value=True)
         self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(['other2'])
 
         # Test sync standby is replaced when new standby is joined
-        self.p.pick_synchronous_standby = Mock(return_value=(['other2', 'other3'], ['other2']))
+        self.p.sync_handler.current_state = Mock(return_value=(['other2', 'other3'], ['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'],))
@@ -1116,7 +1116,7 @@ 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.pick_synchronous_standby = Mock(return_value=(['other2'], ['other2']))
+        self.p.sync_handler.current_state = Mock(return_value=(['other2'], ['other2']))
         self.ha.run_cycle()
         self.ha.dcs.get_cluster.assert_called_once()
         self.assertEqual(self.ha.dcs.write_sync_state.call_count, 2)
@@ -1139,14 +1139,14 @@ class TestHa(PostgresInit):
         # Test sync set to '*' when synchronous_mode_strict is enabled
         mock_set_sync.reset_mock()
         self.ha.is_synchronous_mode_strict = true
-        self.p.pick_synchronous_standby = Mock(return_value=([], []))
+        self.p.sync_handler.current_state = Mock(return_value=([], []))
         self.ha.run_cycle()
         mock_set_sync.assert_called_once_with(['*'])
 
     def test_sync_replication_become_master(self):
         self.ha.is_synchronous_mode = true
 
-        mock_set_sync = self.p.config.set_synchronous_standby = Mock()
+        mock_set_sync = self.p.sync_handler.set_synchronous_standby_names = Mock()
         self.p.is_leader = false
         self.p.set_role('replica')
         self.ha.has_lock = true

tests/test_postgresql.py

@@ -10,7 +10,7 @@ from mock import Mock, MagicMock, PropertyMock, patch, mock_open
 import patroni.psycopg as psycopg
 
 from patroni.async_executor import CriticalTask
-from patroni.dcs import Cluster, RemoteMember, SyncState
+from patroni.dcs import RemoteMember
 from patroni.exceptions import PostgresConnectionException, PatroniException
 from patroni.postgresql import Postgresql, STATE_REJECT, STATE_NO_RESPONSE
 from patroni.postgresql.bootstrap import Bootstrap
@@ -640,79 +640,12 @@ class TestPostgresql(BaseTestPostgresql):
             self.p._state = 'starting'
             self.assertIsNone(self.p.wait_for_startup())
 
-    @patch.object(Postgresql, 'last_operation', Mock(return_value=2))
-    def test_pick_sync_standby(self):
-        cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
-                          SyncState(0, self.me.name, self.leadermem.name), None, None, None)
-
-        with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[
-                    'on',
-                    [{'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1},
-                     {'application_name': self.me.name, 'sync_state': 'async', 'flush_lsn': 2},
-                     {'application_name': self.other.name, 'sync_state': 'async', 'flush_lsn': 2}]
-                ]):
-
-            self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], [self.leadermem.name]))
-
-        with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[
-                    'remote_write',
-                    [{'application_name': self.leadermem.name, 'sync_state': 'potential', 'write_lsn': 1},
-                     {'application_name': self.me.name, 'sync_state': 'async', 'write_lsn': 2},
-                     {'application_name': self.other.name, 'sync_state': 'async', 'write_lsn': 2}]
-                ]):
-            self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.leadermem.name], []))
-
-        with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[
-                    'remote_apply',
-                    [{'application_name': self.me.name.upper(), 'sync_state': 'async', 'replay_lsn': 2},
-                     {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}]
-                ]):
-            self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))
-
-        with patch.object(Postgresql, "_cluster_info_state_get", side_effect=[
-                    'remote_apply',
-                    [{'application_name': 'missing', 'sync_state': 'sync', 'replay_lsn': 3},
-                     {'application_name': self.me.name, 'sync_state': 'async', 'replay_lsn': 2},
-                     {'application_name': self.other.name, 'sync_state': 'async', 'replay_lsn': 1}]
-                ]):
-            self.assertEqual(self.p.pick_synchronous_standby(cluster), ([self.me.name], []))
-
-        with patch.object(Postgresql, "_cluster_info_state_get", side_effect=['remote_apply', []]):
-            self.p._major_version = 90400
-            self.assertEqual(self.p.pick_synchronous_standby(cluster), ([], []))
-
-    def test_set_sync_standby(self):
-        def value_in_conf():
-            with open(os.path.join(self.p.data_dir, 'postgresql.conf')) as f:
-                for line in f:
-                    if line.startswith('synchronous_standby_names'):
-                        return line.strip()
-
-        mock_reload = self.p.reload = Mock()
-        self.p.config.set_synchronous_standby(['n1'])
-        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
-        mock_reload.assert_called()
-
-        mock_reload.reset_mock()
-        self.p.config.set_synchronous_standby(['n1'])
-        mock_reload.assert_not_called()
-        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
-
-        self.p.config.set_synchronous_standby(['n1', 'n2'])
-        mock_reload.assert_called()
-        self.assertEqual(value_in_conf(), "synchronous_standby_names = '2 (n1,n2)'")
-
-        mock_reload.reset_mock()
-        self.p.config.set_synchronous_standby([])
-        mock_reload.assert_called()
-        self.assertEqual(value_in_conf(), None)
-
     def test_get_server_parameters(self):
         config = {'synchronous_mode': True, 'parameters': {'wal_level': 'hot_standby'}, 'listen': '0'}
         self.p.config.get_server_parameters(config)
         config['synchronous_mode_strict'] = True
         self.p.config.get_server_parameters(config)
-        self.p.config.set_synchronous_standby('foo')
+        self.p.config.set_synchronous_standby_names('foo')
         self.assertTrue(str(self.p.config.get_server_parameters(config)).startswith('{'))
 
     @patch('time.sleep', Mock())

tests/test_sync.py

@@ -0,0 +1,89 @@
+import os
+
+from mock import Mock, patch
+
+from patroni.dcs import Cluster, SyncState
+from patroni.postgresql import Postgresql
+
+from . import BaseTestPostgresql, psycopg_connect
+
+
+@patch('subprocess.call', Mock(return_value=0))
+@patch('patroni.psycopg.connect', psycopg_connect)
+class TestSync(BaseTestPostgresql):
+
+    @patch('subprocess.call', Mock(return_value=0))
+    @patch('os.rename', Mock())
+    @patch('patroni.postgresql.CallbackExecutor', Mock())
+    @patch.object(Postgresql, 'get_major_version', Mock(return_value=140000))
+    @patch.object(Postgresql, 'is_running', Mock(return_value=True))
+    def setUp(self):
+        super(TestSync, self).setUp()
+        self.p.config.write_postgresql_conf()
+        self.s = self.p.sync_handler
+
+    @patch.object(Postgresql, 'last_operation', Mock(return_value=1))
+    def test_pick_sync_standby(self):
+        cluster = Cluster(True, None, self.leader, 0, [self.me, self.other, self.leadermem], None,
+                          SyncState(0, self.me.name, self.leadermem.name), None, None, None)
+
+        pg_stat_replication = [
+            {'pid': 100, 'application_name': self.leadermem.name, 'sync_state': 'sync', 'flush_lsn': 1},
+            {'pid': 101, 'application_name': self.me.name, 'sync_state': 'async', 'flush_lsn': 2},
+            {'pid': 102, 'application_name': self.other.name, 'sync_state': 'async', '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=[self.leadermem.name,
+                                                                              'on', pg_stat_replication]):
+            self.assertEqual(self.s.current_state(cluster), ([self.leadermem.name], [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], []))
+
+        # 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], []))
+
+        # 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], []))
+
+        # 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), ([], []))
+
+    def test_set_sync_standby(self):
+        def value_in_conf():
+            with open(os.path.join(self.p.data_dir, 'postgresql.conf')) as f:
+                for line in f:
+                    if line.startswith('synchronous_standby_names'):
+                        return line.strip()
+
+        mock_reload = self.p.reload = Mock()
+        self.s.set_synchronous_standby_names(['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'])
+        mock_reload.assert_not_called()
+        self.assertEqual(value_in_conf(), "synchronous_standby_names = 'n1'")
+
+        self.s.set_synchronous_standby_names(['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([])
+        mock_reload.assert_called()
+        self.assertEqual(value_in_conf(), None)