Repository: zalando/patroni
Pull request #2524 (2023-01-12 17:10:55)

Enhanced sync connections check
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.

coveralls: ## Pull Request Test Coverage Report for [Build 3904436501](https://coveralls.io/builds/55899045)

* **88** of **88**   **(100.0%)**  changed or added relevant lines in **4** files are covered.
* No unchanged relevant lines lost coverage.
* Overall coverage increased (+**0.0006%**) to **99.843%**

---



|  Totals | [![Coverage Status](https://coveralls.io/builds/55899045/badge)](https://coveralls.io/builds/55899045) |
| :-- | --: |
| Change from base [Build 3900371470](https://coveralls.io/builds/55879727): |  0.0006% |
| Covered Lines: | 11425 |
| Relevant Lines: | 11443 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/features/dcs_failsafe_mode.feature b/features/dcs_failsafe_mode.feature
index 52278041..1c250762 100644
--- a/features/dcs_failsafe_mode.feature
+++ b/features/dcs_failsafe_mode.feature
@@ -24,9 +24,9 @@ Feature: dcs failsafe mode
 
   @dcs-failsafe
   Scenario: check new replica isn't promoted when leader is down and DCS is up
+    Given DCS is up
     When I do a backup of postgres0
     And I shut down postgres0
-    And DCS is up
     When I start postgres1 in a cluster batman from backup with no_master
     And I sleep for 2 seconds
     Then postgres1 role is the replica after 12 seconds
@@ -47,31 +47,28 @@ Feature: dcs failsafe mode
   Scenario: check leader and replica are functioning while DCS is down
     Given logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
     And DCS is down
-    And I sleep for 12 seconds
+    Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds
     Then postgres0 role is the primary after 10 seconds
     And postgres1 role is the replica after 2 seconds
     And replication works from postgres0 to postgres1 after 10 seconds
     And I get all changes from logical slot dcs_slot_0 on postgres0
-    And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 10 seconds
+    And logical slot dcs_slot_0 is in sync between postgres0 and postgres1 after 20 seconds
 
   @dcs-failsafe
   Scenario: check master is demoted when one replica is shut down and DCS is down
     Given DCS is down
-    And I shut down postgres1
+    And I kill postgres1
+    And I kill postmaster on postgres1
     And I sleep for 2 seconds
     Then postgres0 role is the replica after 12 seconds
 
   @dcs-failsafe
   Scenario: check known replica is promoted when leader is down and DCS is up
-    Given DCS is up
-    Then postgres0 role is the primary after 22 seconds
+    Given I shut down postgres0
+    And DCS is up
     When I start postgres1
     Then "members/postgres1" key in DCS has state=running after 10 seconds
-    And Response on GET http://127.0.0.1:8009/failsafe contains postgres1 after 10 seconds
-    Given DCS is down
-    And I shut down postgres0
-    And DCS is up
-    Then postgres1 role is the primary after 22 seconds
+    And postgres1 role is the primary after 25 seconds
 
   @dcs-failsafe
   Scenario: check three-node cluster is functioning while DCS is down
@@ -80,8 +77,9 @@ Feature: dcs failsafe mode
     Then "members/postgres0" key in DCS has state=running after 10 seconds
     And "members/postgres2" key in DCS has state=running after 10 seconds
     And Response on GET http://127.0.0.1:8008/failsafe contains postgres2 after 10 seconds
+    And replication works from postgres1 to postgres0 after 10 seconds
     Given DCS is down
-    And I sleep for 12 seconds
+    Then Response on GET http://127.0.0.1:8008/primary contains failsafe_mode_is_active after 12 seconds
     Then postgres1 role is the primary after 10 seconds
     And postgres0 role is the replica after 2 seconds
     And postgres2 role is the replica after 2 seconds
diff --git a/features/environment.py b/features/environment.py
index b1a76f5a..318c9801 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -184,7 +184,9 @@ class PatroniController(AbstractController):
             config.pop('etcd', None)
 
         raft_port = os.environ.get('RAFT_PORT')
-        if raft_port:
+        # If patroni_raft_controller is suspended two Patroni members is enough to get a quorum,
+        # therefore we don't want Patroni to join as a voting member when testing dcs_failsafe_mode.
+        if raft_port and not self._output_dir.endswith('dcs_failsafe_mode'):
             os.environ['RAFT_PORT'] = str(int(raft_port) + 1)
             config['raft'] = {'data_dir': self._output_dir, 'self_addr': 'localhost:' + os.environ['RAFT_PORT']}
 
diff --git a/patroni/ha.py b/patroni/ha.py
index 62836a41..6b95c6d0 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -105,7 +105,6 @@ class Failsafe(object):
             cluster = list(cluster)
             # We rely on the strict order of fields in the namedtuple
             cluster[2] = leader
-            cluster[4].append(leader.member)
             cluster[8] = leader.member.data['slots']
             cluster = Cluster(*cluster)
         return cluster
@@ -459,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
@@ -569,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)))
@@ -589,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()
@@ -613,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 \
@@ -722,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()
@@ -1045,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
@@ -1641,31 +1641,27 @@ class Ha(object):
                 # try to start dead postgres
                 return self.recover()
 
-            try:
-                if self.cluster.is_unlocked():
-                    ret = self.process_unhealthy_cluster()
-                else:
-                    msg = self.process_healthy_cluster()
-                    ret = self.evaluate_scheduled_restart() or msg
-            finally:
-                # we might not have a valid PostgreSQL connection here if another thread
-                # stops PostgreSQL, therefore, we only reload replication slots if no
-                # asynchronous processes are running (should be always the case for the master)
-                if not self._async_executor.busy and not self.state_handler.is_starting():
-                    create_slots = self.state_handler.slots_handler.sync_replication_slots(self.cluster,
-                                                                                           self.patroni.nofailover,
-                                                                                           self.patroni.replicatefrom,
-                                                                                           self.is_paused())
-                    if not self.state_handler.cb_called:
-                        if not self.state_handler.is_leader():
-                            self._rewind.trigger_check_diverged_lsn()
-                        self.state_handler.call_nowait(ACTION_ON_START)
-                    if create_slots and self.cluster.leader:
-                        err = self._async_executor.try_run_async('copy_logical_slots',
-                                                                 self.state_handler.slots_handler.copy_logical_slots,
-                                                                 args=(self.cluster, create_slots))
-                        if not err:
-                            ret = 'Copying logical slots {0} from the primary'.format(create_slots)
+            if self.cluster.is_unlocked():
+                ret = self.process_unhealthy_cluster()
+            else:
+                msg = self.process_healthy_cluster()
+                ret = self.evaluate_scheduled_restart() or msg
+
+            # we might not have a valid PostgreSQL connection here if another thread
+            # stops PostgreSQL, therefore, we only reload replication slots if no
+            # asynchronous processes are running (should be always the case for the master)
+            if not self._async_executor.busy and not self.state_handler.is_starting():
+                create_slots = self._sync_replication_slots(False)
+                if not self.state_handler.cb_called:
+                    if not self.state_handler.is_leader():
+                        self._rewind.trigger_check_diverged_lsn()
+                    self.state_handler.call_nowait(ACTION_ON_START)
+                if create_slots and self.cluster.leader:
+                    err = self._async_executor.try_run_async('copy_logical_slots',
+                                                             self.state_handler.slots_handler.copy_logical_slots,
+                                                             args=(self.cluster, create_slots))
+                    if not err:
+                        ret = 'Copying logical slots {0} from the primary'.format(create_slots)
             return ret
         except DCSError:
             dcs_failed = True
@@ -1675,9 +1671,9 @@ class Ha(object):
             return 'Error communicating with PostgreSQL. Will try again later'
         finally:
             if not dcs_failed:
-                self.touch_member()
-                if self.state_handler.is_leader():
+                if self.is_leader():
                     self._failsafe.set_is_active(0)
+                self.touch_member()
 
     def _handle_dcs_error(self):
         if not self.is_paused() and self.state_handler.is_running():
@@ -1694,15 +1690,35 @@ class Ha(object):
                 logger.warning('AsyncExecutor is busy, demoting from the main thread')
                 self.demote('offline')
                 return 'demoted self because DCS is not accessible and I was a leader'
-            elif self.is_failsafe_mode():
-                cluster = self._failsafe.update_cluster(self.cluster)
-                if cluster:
-                    self.state_handler.slots_handler.sync_replication_slots(cluster,
+            else:
+                self._sync_replication_slots(True)
+        return 'DCS is not accessible'
+
+    def _sync_replication_slots(self, dcs_failed):
+        """Handles replication slots.
+
+        :param dcs_failed: bool, indicates that communication with DCS failed (get_cluster() or update_leader())
+        :returns: list[str], replication slots names that should be copied from the primary"""
+
+        slots = []
+
+        # If dcs_failed we don't want to touch replication slots on a leader or replicas if failsafe_mode isn't enabled.
+        if not self.cluster or dcs_failed and (self.is_leader() or not self.is_failsafe_mode()):
+            return slots
+
+        # It could be that DCS is read-only, or only the leader can't access it.
+        # Only the second one could be handled by `load_cluster_from_dcs()`.
+        # The first one affects advancing logical replication slots on replicas, therefore we rely on
+        # Failsafe.update_cluster(), that will return "modified" Cluster if failsafe mode is active.
+        cluster = self._failsafe.update_cluster(self.cluster)\
+            if self.is_failsafe_mode() and not self.is_leader() else self.cluster
+        if cluster:
+            slots = self.state_handler.slots_handler.sync_replication_slots(cluster,
                                                                             self.patroni.nofailover,
                                                                             self.patroni.replicatefrom,
                                                                             self.is_paused())
-
-        return 'DCS is not accessible'
+        # Don't copy replication slots if failsafe_mode is active
+        return [] if self.failsafe_is_active() else slots
 
     def run_cycle(self):
         with self._async_executor:
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index f3eae728..afd40e11 100644
--- a/patroni/postgresql/__init__.py
+++ b/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'))
@@ -1053,13 +1063,15 @@ class Postgresql(object):
     def move_data_directory(self):
         if os.path.isdir(self._data_dir) and not self.is_running():
             try:
-                postfix = time.strftime('%Y-%m-%d-%H-%M-%S')
+                postfix = 'failed'
 
                 # let's see if the wal directory is a symlink, in this case we
                 # should move the target
                 for (source, pg_wal_realpath) in self.pg_wal_realpath().items():
                     logger.info('renaming WAL directory and updating symlink: %s', pg_wal_realpath)
-                    new_name = '{0}_{1}'.format(pg_wal_realpath, postfix)
+                    new_name = '{0}.{1}'.format(pg_wal_realpath, postfix)
+                    if os.path.exists(new_name):
+                        shutil.rmtree(new_name)
                     os.rename(pg_wal_realpath, new_name)
                     os.unlink(source)
                     os.symlink(new_name, source)
@@ -1067,13 +1079,17 @@ class Postgresql(object):
                 # Move user defined tablespace directory
                 for (source, pg_tsp_rpath) in self.pg_tblspc_realpaths().items():
                     logger.info('renaming user defined tablespace directory and updating symlink: %s', pg_tsp_rpath)
-                    new_name = '{0}_{1}'.format(pg_tsp_rpath, postfix)
+                    new_name = '{0}.{1}'.format(pg_tsp_rpath, postfix)
+                    if os.path.exists(new_name):
+                        shutil.rmtree(new_name)
                     os.rename(pg_tsp_rpath, new_name)
                     os.unlink(source)
                     os.symlink(new_name, source)
 
-                new_name = '{0}_{1}'.format(self._data_dir, postfix)
+                new_name = '{0}.{1}'.format(self._data_dir, postfix)
                 logger.info('renaming data directory to %s', new_name)
+                if os.path.exists(new_name):
+                    shutil.rmtree(new_name)
                 os.rename(self._data_dir, new_name)
             except OSError:
                 logger.exception("Could not rename data directory %s", self._data_dir)
@@ -1106,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:
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index b732d01f..3a438c43 100644
--- a/patroni/postgresql/config.py
+++ b/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):
diff --git a/patroni/postgresql/misc.py b/patroni/postgresql/misc.py
index 3a6faa1c..798bed35 100644
--- a/patroni/postgresql/misc.py
+++ b/patroni/postgresql/misc.py
@@ -1,4 +1,6 @@
+import errno
 import logging
+import os
 
 from patroni.exceptions import PostgresException
 
@@ -73,3 +75,16 @@ def parse_history(data):
 def format_lsn(lsn, full=False):
     template = '{0:X}/{1:08X}' if full else '{0:X}/{1:X}'
     return template.format(lsn >> 32, lsn & 0xFFFFFFFF)
+
+
+def fsync_dir(path):
+    if os.name != 'nt':
+        fd = os.open(path, os.O_DIRECTORY)
+        try:
+            os.fsync(fd)
+        except OSError as e:
+            # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe.
+            if e.errno != errno.EINVAL:
+                raise
+        finally:
+            os.close(fd)
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index 595105ba..76e6a54f 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -9,7 +9,7 @@ import subprocess
 from threading import Lock, Thread
 
 from .connection import get_connection_cursor
-from .misc import format_lsn, parse_history, parse_lsn
+from .misc import format_lsn, fsync_dir, parse_history, parse_lsn
 from ..async_executor import CriticalTask
 from ..dcs import Leader
 
@@ -362,6 +362,18 @@ class Rewind(object):
                 else:
                     logger.info('Failed to archive WAL segment %s', wal)
 
+    def _maybe_clean_pg_replslot(self):
+        """Clean pg_replslot directory if pg version is less then 11
+        (pg_rewind deletes $PGDATA/pg_replslot content only since pg11)."""
+        if self._postgresql.major_version < 110000:
+            replslot_dir = self._postgresql.slots_handler.pg_replslot_dir
+            try:
+                for f in os.listdir(replslot_dir):
+                    shutil.rmtree(os.path.join(replslot_dir, f))
+                fsync_dir(replslot_dir)
+            except Exception as e:
+                logger.warning('Unable to clean %s: %r', replslot_dir, e)
+
     def pg_rewind(self, r):
         # prepare pg_rewind connection
         env = self._postgresql.config.write_pgpass(r)
@@ -439,6 +451,7 @@ class Rewind(object):
             return
 
         if self.pg_rewind(r):
+            self._maybe_clean_pg_replslot()
             self._state = REWIND_STATUS.SUCCESS
         else:
             if not self.check_leader_is_not_in_recovery(r):
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index 7abe2182..ea04015c 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -1,4 +1,3 @@
-import errno
 import logging
 import os
 import shutil
@@ -8,7 +7,7 @@ from contextlib import contextmanager
 from threading import Condition, Thread
 
 from .connection import get_connection_cursor
-from .misc import format_lsn
+from .misc import format_lsn, fsync_dir
 from ..psycopg import OperationalError
 
 logger = logging.getLogger(__name__)
@@ -19,19 +18,6 @@ def compare_slots(s1, s2, dbid='database'):
                                          s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
 
 
-def fsync_dir(path):
-    if os.name != 'nt':
-        fd = os.open(path, os.O_DIRECTORY)
-        try:
-            os.fsync(fd)
-        except OSError as e:
-            # Some filesystems don't like fsyncing directories and raise EINVAL. Ignoring it is usually safe.
-            if e.errno != errno.EINVAL:
-                raise
-        finally:
-            os.close(fd)
-
-
 class SlotsAdvanceThread(Thread):
 
     def __init__(self, slots_handler):
@@ -122,6 +108,7 @@ class SlotsHandler(object):
         self._advance = None
         self._replication_slots = {}  # already existing replication slots
         self._unready_logical_slots = {}
+        self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
         self.schedule()
 
     def _query(self, sql, *params):
@@ -192,31 +179,33 @@ class SlotsHandler(object):
         return False
 
     def drop_replication_slot(self, name):
-        cursor = self._query(('SELECT pg_catalog.pg_drop_replication_slot(%s) WHERE EXISTS (SELECT 1 ' +
-                              'FROM pg_catalog.pg_replication_slots WHERE slot_name = %s AND NOT active)'), name, name)
-        # In normal situation rowcount should be 1, otherwise either slot doesn't exists or it is still active
-        return cursor.rowcount == 1
+        """Returns a tuple(active, dropped)"""
+        cursor = self._query(('WITH slots AS (SELECT slot_name, active' +
+                              ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),' +
+                              ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),' +
+                              ' true AS dropped FROM slots WHERE not active) ' +
+                              'SELECT active, COALESCE(dropped, false) FROM slots' +
+                              ' FULL OUTER JOIN dropped ON true'), name)
+        return cursor.fetchone() if cursor.rowcount == 1 else (False, False)
 
     def _drop_incorrect_slots(self, cluster, slots, paused):
         # drop old replication slots which are not presented in desired slots
         for name in set(self._replication_slots) - set(slots):
             if not paused and not self.ignore_replication_slot(cluster, name):
-                if not self.drop_replication_slot(name):
+                active, dropped = self.drop_replication_slot(name)
+                if dropped:
+                    logger.info("Dropped unknown replication slot '%s'", name)
+                else:
                     self._schedule_load_slots = True
-                    # Check if slot is still active, this is not considered an error
-                    cursor = self._query(('SELECT 1 FROM pg_catalog.pg_replication_slots WHERE ' +
-                                          'slot_name = %s AND active'), name)
-                    if (cursor.rowcount == 1):
+                    if active:
                         logger.debug("Unable to drop unknown replication slot '%s', slot is still active", name)
                     else:
                         logger.error("Failed to drop replication slot '%s'", name)
-                else:
-                    logger.info("Dropped unknown replication slot '%s'", name)
         for name, value in slots.items():
             if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]):
                 logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s",
                             name, self._replication_slots[name], value)
-                if self.drop_replication_slot(name):
+                if self.drop_replication_slot(name) == (False, True):
                     self._replication_slots.pop(name)
                 else:
                     logger.error("Failed to drop replication slot '%s'", name)
@@ -395,9 +384,8 @@ class SlotsHandler(object):
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         if isinstance(create_slots, dict) and create_slots and self._postgresql.stop():
-            pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
             for name, value in create_slots.items():
-                slot_dir = os.path.join(pg_replslot_dir, name)
+                slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
                     shutil.rmtree(slot_tmp_dir)
@@ -412,7 +400,7 @@ class SlotsHandler(object):
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
                 self._unready_logical_slots[name] = None
-            fsync_dir(pg_replslot_dir)
+            fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
             self._postgresql.start()
 
     def schedule(self, value=None):
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
new file mode 100644
index 00000000..bbc30044
--- /dev/null
+++ b/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()


Test patch:
diff --git a/tests/__init__.py b/tests/__init__.py
index 0da270d7..91cc99df 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -98,8 +98,12 @@ class MockCursor(object):
             self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
             self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'a', 'b', 5, 100, 500)]
+        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'):
diff --git a/tests/test_ha.py b/tests/test_ha.py
index c9c2f129..c88fc774 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -310,6 +310,8 @@ class TestHa(PostgresInit):
 
     @patch.object(Rewind, 'rewind_or_reinitialize_needed_and_possible', Mock(return_value=True))
     @patch.object(Rewind, 'can_rewind', PropertyMock(return_value=True))
+    @patch('os.listdir', Mock(return_value=[]))
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
     def test_recover_with_rewind(self):
         self.p.is_running = false
         self.ha.cluster = get_cluster_initialized_with_leader()
@@ -498,6 +500,14 @@ class TestHa(PostgresInit):
         self.assertEqual(self.ha.run_cycle(),
                          'continue to run as a leader because failsafe mode is enabled and all members are accessible')
 
+    def test_readonly_dcs_primary_failsafe(self):
+        self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
+        self.ha.dcs.update_leader = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.dcs._last_failsafe = self.ha.cluster.failsafe
+        self.ha.state_handler.name = self.ha.cluster.leader.name
+        self.assertEqual(self.ha.run_cycle(),
+                         'continue to run as a leader because failsafe mode is enabled and all members are accessible')
+
     def test_no_dcs_connection_replica_failsafe(self):
         self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
         self.ha.cluster = get_cluster_initialized_with_leader_and_failsafe()
@@ -506,6 +516,14 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
 
+    def test_no_dcs_connection_replica_failsafe_not_enabled_but_active(self):
+        self.ha.load_cluster_from_dcs = Mock(side_effect=DCSError('Etcd is not responding properly'))
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.ha.update_failsafe({'name': 'leader', 'api_url': 'http://127.0.0.1:8008/patroni',
+                                 'conn_url': 'postgres://127.0.0.1:5432/postgres', 'slots': {'foo': 1000}})
+        self.p.is_leader = false
+        self.assertEqual(self.ha.run_cycle(), 'DCS is not accessible')
+
     def test_update_failsafe(self):
         self.assertRaises(Exception, self.ha.update_failsafe, {})
         self.p.set_role('master')
@@ -739,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')
 
@@ -755,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):
@@ -817,6 +835,8 @@ class TestHa(PostgresInit):
 
     @patch.object(Rewind, 'pg_rewind', true)
     @patch.object(Rewind, 'check_leader_is_not_in_recovery', true)
+    @patch('os.listdir', Mock(return_value=[]))
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
     def test_post_recover(self):
         self.p.is_running = false
         self.ha.has_lock = true
@@ -1042,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
@@ -1065,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()
@@ -1073,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'],))
@@ -1096,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)
@@ -1119,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
diff --git a/tests/test_postgresql.py b/tests/test_postgresql.py
index d5524bad..b8725218 100644
--- a/tests/test_postgresql.py
+++ b/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
@@ -438,6 +438,8 @@ class TestPostgresql(BaseTestPostgresql):
         self.assertTrue(self.p.stop())
 
     @patch('os.rename', Mock())
+    @patch('os.path.exists', Mock(return_value=True))
+    @patch('shutil.rmtree', Mock())
     @patch('os.path.isdir', Mock(return_value=True))
     @patch('os.unlink', Mock())
     @patch('os.symlink', Mock())
@@ -638,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())
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index b9398205..9da500e9 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -274,6 +274,19 @@ class TestRewind(BaseTestPostgresql):
             self.r._archive_ready_wals()
             mock_logger_info.assert_not_called()
 
+    @patch.object(Postgresql, 'major_version', PropertyMock(return_value=100000))
+    @patch('os.listdir', Mock(side_effect=[OSError, ['something', 'something_else']]))
+    @patch('shutil.rmtree', Mock())
+    @patch('patroni.postgresql.rewind.fsync_dir', Mock())
+    @patch('patroni.postgresql.rewind.logger.warning')
+    def test_maybe_clean_pg_replslot(self, mock_logger):
+        # failed to list pg_replslot/
+        self.assertIsNone(self.r._maybe_clean_pg_replslot())
+        mock_logger.assert_called_once()
+        mock_logger.reset_mock()
+
+        self.assertIsNone(self.r._maybe_clean_pg_replslot())
+
     @patch('os.unlink', Mock())
     @patch('os.listdir', Mock(return_value=[]))
     @patch('os.path.isfile', Mock(return_value=True))
diff --git a/tests/test_slots.py b/tests/test_slots.py
index 4b19fc4a..6fe1691c 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -9,7 +9,8 @@ from threading import Thread
 from patroni import psycopg
 from patroni.dcs import Cluster, ClusterConfig, Member
 from patroni.postgresql import Postgresql
-from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler, fsync_dir
+from patroni.postgresql.misc import fsync_dir
+from patroni.postgresql.slots import SlotsAdvanceThread, SlotsHandler
 
 from . import BaseTestPostgresql, psycopg_connect, MockCursor
 
@@ -42,7 +43,10 @@ class TestSlotsHandler(BaseTestPostgresql):
         with mock.patch('patroni.postgresql.Postgresql._query', Mock(side_effect=psycopg.OperationalError)):
             self.s.sync_replication_slots(cluster, False)
         self.p.set_role('standby_leader')
-        self.s.sync_replication_slots(cluster, False)
+        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))),\
+                patch('patroni.postgresql.slots.logger.debug') as mock_debug:
+            self.s.sync_replication_slots(cluster, False)
+            mock_debug.assert_called_once()
         self.p.set_role('replica')
         with patch.object(Postgresql, 'is_leader', Mock(return_value=False)),\
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
@@ -51,8 +55,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('master')
         with mock.patch('patroni.postgresql.Postgresql.role', new_callable=PropertyMock(return_value='replica')):
             self.s.sync_replication_slots(cluster, False)
-        with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=True)),\
-                patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
+        with patch('patroni.dcs.logger.error', new_callable=Mock()) as errorlog_mock:
             alias1 = Member(0, 'test-3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
             alias2 = Member(0, 'test.3', 28, {'conn_url': 'postgres://replicator:rep-pass@127.0.0.1:5436/postgres'})
             cluster.members.extend([alias1, alias2])
diff --git a/tests/test_sync.py b/tests/test_sync.py
new file mode 100644
index 00000000..7e6aea47
--- /dev/null
+++ b/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)