Repository: zalando/patroni
Pull request #2747 (2023-07-11 16:23:00)

Refactor get_dcs
Now uses generators instead of for loops and implements importing modules once.

Split from #2743 


Patch:
diff --git a/Dockerfile b/Dockerfile
index 4aa0ce5c..c5b927ee 100644
--- a/Dockerfile
+++ b/Dockerfile
@@ -25,8 +25,7 @@ RUN set -ex \
             | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
             | xargs apt-get install -y vim curl less jq locales haproxy sudo \
                             python3-etcd python3-kazoo python3-pip busybox \
-                            net-tools iputils-ping --fix-missing \
-    && pip3 install dumb-init \
+                            net-tools iputils-ping dumb-init --fix-missing \
 \
     # Cleanup all locales but en_US.UTF-8
     && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \
@@ -71,7 +70,7 @@ RUN set -ex \
     # Clean up all useless packages and some files
     && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
                 libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
-                exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
+                exim4-config gnupg-agent dirmngr \
                 git make \
     && apt-get autoremove -y \
     && apt-get clean -y \
diff --git a/Dockerfile.citus b/Dockerfile.citus
index 8693b504..7e6ec18c 100644
--- a/Dockerfile.citus
+++ b/Dockerfile.citus
@@ -25,7 +25,7 @@ RUN set -ex \
             | grep -Ev '^python3-(sphinx|etcd|consul|kazoo|kubernetes)' \
             | xargs apt-get install -y vim curl less jq locales haproxy sudo \
                             python3-etcd python3-kazoo python3-pip busybox \
-                            net-tools iputils-ping lsb-release --fix-missing \
+                            net-tools iputils-ping lsb-release dumb-init --fix-missing \
     && if [ $(dpkg --print-architecture) = 'arm64' ]; then \
         apt-get install -y postgresql-server-dev-$PG_MAJOR \
                            git gcc make autoconf \
@@ -42,7 +42,6 @@ RUN set -ex \
         && apt-get update -y \
         && apt-get -y install postgresql-$PG_MAJOR-citus-11.3; \
     fi \
-    && pip3 install dumb-init \
 \
     # Cleanup all locales but en_US.UTF-8
     && find /usr/share/i18n/charmaps/ -type f ! -name UTF-8.gz -delete \
@@ -88,7 +87,7 @@ RUN set -ex \
     # Clean up all useless packages and some files
     && apt-get purge -y --allow-remove-essential python3-pip gzip bzip2 util-linux e2fsprogs \
                 libmagic1 bsdmainutils login ncurses-bin libmagic-mgc e2fslibs bsdutils \
-                exim4-config gnupg-agent dirmngr libpython2.7-stdlib libpython2.7-minimal \
+                exim4-config gnupg-agent dirmngr \
                 postgresql-server-dev-$PG_MAJOR git gcc make autoconf \
                 libc6-dev flex libicu-dev libkrb5-dev liblz4-dev \
                 libpam0g-dev libreadline-dev libselinux1-dev libssl-dev libxslt1-dev libzstd-dev uuid-dev \
diff --git a/docs/releases.rst b/docs/releases.rst
index b9a3eae7..48cad683 100644
--- a/docs/releases.rst
+++ b/docs/releases.rst
@@ -3,6 +3,50 @@
 Release notes
 =============
 
+Version 3.0.4
+-------------
+
+**New features**
+
+- Make the replication status of standby nodes visible (Alexander Kukushkin)
+
+  For PostgreSQL 9.6+ Patroni will report the replication state as ``streaming`` when the standby is streaming from the other node or ``in archive recovery`` when there is no replication connection and ``restore_command`` is set. The state is visible in ``member`` keys in DCS, in the REST API, and in ``patronictl list`` output.
+
+
+**Improvements**
+
+- Improved error messages with Etcd v3 (Alexander Kukushkin)
+
+  When Etcd v3 cluster isn't accessible Patroni was reporting that it can't access ``/v2`` endpoints.
+
+- Use quorum read in ``patronictl`` if it is possible (Alexander Kukushkin)
+
+  Etcd or Consul clusters could be degraded to read-only, but from the ``patronictl`` view everything was fine. Now it will fail with the error.
+
+- Prevent splitbrain from duplicate names in configuration (Mark Pekala)
+
+  When starting Patroni will check if node with the same name is registered in DCS, and try to query its REST API. If REST API is accessible Patroni exits with an error. It will help to protect from the human error.
+
+- Start Postgres not in recovery if it crashed while Patroni is running (Alexander Kukushkin)
+
+  It may reduce recovery time and will help from unnecessary timeline increments.
+
+
+**Bugfixes**
+
+- REST API SSL certificate were not reloaded upon receiving a SIGHUP (Israel Barth Rubio)
+
+  Regression was introduced in 3.0.3.
+
+- Fixed integer GUCs validation for parameters like ``max_connections`` (Feike Steenbergen)
+
+  Patroni didn't like quoted numeric values. Regression was introduced in 3.0.3.
+
+- Fix issue with ``synchronous_mode`` (Alexander Kukushkin)
+
+  Execute ``txid_current()`` with ``synchronous_commit=off`` so it doesn't accidentally wait for absent synchronous standbys when ``synchronous_mode_strict`` is enabled.
+
+
 Version 3.0.3
 -------------
 
diff --git a/docs/rest_api.rst b/docs/rest_api.rst
index 7bde651f..9deb3f5e 100644
--- a/docs/rest_api.rst
+++ b/docs/rest_api.rst
@@ -141,8 +141,8 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# TYPE patroni_replica gauge
 	patroni_replica{scope="batman"} 0
 	# HELP patroni_sync_standby Value is 1 if this node is a sync standby replica, 0 otherwise.
-        # TYPE patroni_sync_standby gauge
-        patroni_sync_standby{scope="batman"} 0
+	# TYPE patroni_sync_standby gauge
+	patroni_sync_standby{scope="batman"} 0
 	# HELP patroni_xlog_received_location Current location of the received Postgres transaction log, 0 if this node is not a replica.
 	# TYPE patroni_xlog_received_location counter
 	patroni_xlog_received_location{scope="batman"} 0
@@ -155,6 +155,12 @@ Retrieve the Patroni metrics in Prometheus format through the ``GET /metrics`` e
 	# HELP patroni_xlog_paused Value is 1 if the Postgres xlog is paused, 0 otherwise.
 	# TYPE patroni_xlog_paused gauge
 	patroni_xlog_paused{scope="batman"} 0
+	# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.
+	# TYPE patroni_postgres_streaming gauge
+	patroni_postgres_streaming{scope="batman"} 1
+	# HELP patroni_postgres_in_archive_recovery Value is 1 if Postgres is replicating from archive, 0 otherwise.
+	# TYPE patroni_postgres_in_archive_recovery gauge
+	patroni_postgres_in_archive_recovery{scope="batman"} 0
 	# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.
 	# TYPE patroni_postgres_server_version gauge
 	patroni_postgres_server_version {scope="batman"} 140004
diff --git a/features/basic_replication.feature b/features/basic_replication.feature
index 4a0eda68..b58520bd 100644
--- a/features/basic_replication.feature
+++ b/features/basic_replication.feature
@@ -72,18 +72,14 @@ Feature: basic replication
     Then table bar is present on postgres1 after 20 seconds
     And Response on GET http://127.0.0.1:8010/config contains master_start_timeout after 10 seconds
 
-  Scenario: check immediate failover when master_start_timeout=0
-    Given I kill postmaster on postgres2
-    Then postgres1 is a leader after 10 seconds
-    And postgres1 role is the primary after 10 seconds
-
   Scenario: check rejoin of the former primary with pg_rewind
     Given I add the table splitbrain to postgres0
     And I start postgres0
     Then postgres0 role is the secondary after 20 seconds
-    When I add the table buz to postgres1
+    When I add the table buz to postgres2
     Then table buz is present on postgres0 after 20 seconds
 
+  @reject-duplicate-name
   Scenario: check graceful rejection when two nodes have the same name
     Given I start duplicate postgres0 on port 8011
     Then there is a "Can't start; there is already a node named 'postgres0' running" CRITICAL in the dup-postgres0 patroni log
diff --git a/features/environment.py b/features/environment.py
index f7690ad2..5906becd 100644
--- a/features/environment.py
+++ b/features/environment.py
@@ -1144,3 +1144,5 @@ def before_scenario(context, scenario):
                 break
     if 'dcs-failsafe' in scenario.effective_tags and not context.dcs_ctl._handle:
         scenario.skip('it is not possible to control state of {0} from tests'.format(context.dcs_ctl.name()))
+    if 'reject-duplicate-name' in scenario.effective_tags and context.dcs_ctl.name() == 'raft':
+        scenario.skip('Flaky test with Raft')
diff --git a/features/recovery.feature b/features/recovery.feature
new file mode 100644
index 00000000..809f7fb9
--- /dev/null
+++ b/features/recovery.feature
@@ -0,0 +1,24 @@
+Feature: recovery
+  We want to check that crashed postgres is started back
+
+  Scenario: check that timeline is not incremented when primary is started after crash
+    Given I start postgres0
+    Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
+    When I start postgres1
+    And I add the table foo to postgres0
+    Then table foo is present on postgres1 after 20 seconds
+    When I kill postmaster on postgres0
+    Then postgres0 role is the primary after 10 seconds
+    When I issue a GET request to http://127.0.0.1:8008/
+    Then I receive a response code 200
+    And I receive a response role master
+    And I receive a response timeline 1
+
+  Scenario: check immediate failover when master_start_timeout=0
+    Given I issue a PATCH request to http://127.0.0.1:8008/config with {"master_start_timeout": 0}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/config contains master_start_timeout after 10 seconds
+    When I kill postmaster on postgres0
+    Then postgres1 is a leader after 10 seconds
+    And postgres1 role is the primary after 10 seconds
diff --git a/features/standby_cluster.feature b/features/standby_cluster.feature
index 4e3bd5f0..850c7970 100644
--- a/features/standby_cluster.feature
+++ b/features/standby_cluster.feature
@@ -13,6 +13,10 @@ Feature: standby cluster
     When I start postgres0
     Then "members/postgres0" key in DCS has state=running after 10 seconds
     And replication works from postgres1 to postgres0 after 15 seconds
+    When I issue a GET request to http://127.0.0.1:8008/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
+    And "members/postgres0" key in DCS has replication_state=streaming after 10 seconds
 
   @slot-advance
   Scenario: check permanent logical slots are synced to the replica
@@ -34,6 +38,9 @@ Feature: standby cluster
     Then postgres1 is a leader of batman1 after 10 seconds
     When I add the table foo to postgres0
     Then table foo is present on postgres1 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8009/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And I sleep for 3 seconds
     When I issue a GET request to http://127.0.0.1:8009/primary
     Then I receive a response code 503
@@ -44,6 +51,9 @@ Feature: standby cluster
     When I start postgres2 in a cluster batman1
     Then postgres2 role is the replica after 24 seconds
     And table foo is present on postgres2 after 20 seconds
+    When I issue a GET request to http://127.0.0.1:8010/patroni
+    Then I receive a response code 200
+    And I receive a response replication_state streaming
     And postgres1 does not have a logical replication slot named test_logical
 
   Scenario: check failover
diff --git a/patroni/api.py b/patroni/api.py
index 7be2a049..8d991d5f 100644
--- a/patroni/api.py
+++ b/patroni/api.py
@@ -463,6 +463,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
         * ``patroni_dcs_last_seen``: epoch timestamp when DCS was last contacted successfully;
         * ``patroni_pending_restart``: ``1`` if this PostgreSQL node is pending a restart, else ``0``;
         * ``patroni_is_paused``: ``1`` if Patroni is in maintenance node, else ``0``.
+
+        For PostgreSQL v9.6+ the response will also have the following:
+
+        * ``patroni_postgres_streaming``: 1 if Postgres is streaming from another node, else ``0``;
+        * ``patroni_postgres_in_archive_recovery``: ``1`` if Postgres isn't streaming and
+                                                    there is ``restore_command`` available, else ``0``.
         """
         postgres = self.get_postgresql_status(True)
         patroni = self.server.patroni
@@ -535,6 +541,18 @@ class RestApiHandler(BaseHTTPRequestHandler):
         metrics.append("patroni_xlog_paused{0} {1}"
                        .format(scope_label, int(postgres.get('xlog', {}).get('paused', False) is True)))
 
+        if postgres.get('server_version', 0) >= 90600:
+            metrics.append("# HELP patroni_postgres_streaming Value is 1 if Postgres is streaming, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_streaming gauge")
+            metrics.append("patroni_postgres_streaming{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'streaming')))
+
+            metrics.append("# HELP patroni_postgres_in_archive_recovery Value is 1"
+                           " if Postgres is replicating from archive, 0 otherwise.")
+            metrics.append("# TYPE patroni_postgres_in_archive_recovery gauge")
+            metrics.append("patroni_postgres_in_archive_recovery{0} {1}"
+                           .format(scope_label, int(postgres.get('replication_state') == 'in archive recovery')))
+
         metrics.append("# HELP patroni_postgres_server_version Version of Postgres (if running), 0 otherwise.")
         metrics.append("# TYPE patroni_postgres_server_version gauge")
         metrics.append("patroni_postgres_server_version {0} {1}".format(scope_label, postgres.get('server_version', 0)))
@@ -1151,8 +1169,11 @@ class RestApiHandler(BaseHTTPRequestHandler):
 
             if postgresql.state not in ('running', 'restarting', 'starting'):
                 raise RetryFailedError('')
+            replication_state = ('(pg_catalog.pg_stat_get_wal_receiver()).status'
+                                 if postgresql.major_version >= 90600 else 'NULL') + ", " +\
+                ("pg_catalog.current_setting('restore_command')" if postgresql.major_version >= 120000 else "NULL")
             stmt = ("SELECT " + postgresql.POSTMASTER_START_TIME + ", " + postgresql.TL_LSN + ","
-                    " pg_catalog.pg_last_xact_replay_timestamp(),"
+                    " pg_catalog.pg_last_xact_replay_timestamp(), " + replication_state + ","
                     " pg_catalog.array_to_json(pg_catalog.array_agg(pg_catalog.row_to_json(ri))) "
                     "FROM (SELECT (SELECT rolname FROM pg_catalog.pg_authid WHERE oid = usesysid) AS usename,"
                     " application_name, client_addr, w.state, sync_state, sync_priority"
@@ -1188,8 +1209,12 @@ class RestApiHandler(BaseHTTPRequestHandler):
                     if not cluster or cluster.is_unlocked() or not cluster.leader else cluster.leader.timeline
                 result['timeline'] = postgresql.replica_cached_timeline(leader_timeline)
 
-            if row[7]:
-                result['replication'] = row[7]
+            replication_state = postgresql.replication_state_from_parameters(row[1] > 0, row[7], row[8])
+            if replication_state:
+                result['replication_state'] = replication_state
+
+            if row[9]:
+                result['replication'] = row[9]
 
         except (psycopg.Error, RetryFailedError, PostgresConnectionException):
             state = postgresql.state
diff --git a/patroni/ctl.py b/patroni/ctl.py
index 12463aad..55a054c1 100644
--- a/patroni/ctl.py
+++ b/patroni/ctl.py
@@ -1490,7 +1490,8 @@ def output_members(obj: Dict[str, Any], cluster: Cluster, name: str,
         * ``Role``: ``Leader``, ``Standby Leader``, ``Sync Standby`` or ``Replica``;
         * ``State``: ``stopping``, ``stopped``, ``stop failed``, ``crashed``, ``running``, ``starting``,
           ``start failed``, ``restarting``, ``restart failed``, ``initializing new cluster``, ``initdb failed``,
-          ``running custom bootstrap script``, ``custom bootstrap failed``, or ``creating replica``, and so on;
+          ``running custom bootstrap script``, ``custom bootstrap failed``, ``creating replica``, ``streaming``,
+          ``in archive recovery``, and so on;
         * ``TL``: current timeline in Postgres;
           ``Lag in MB``: replication lag.
 
diff --git a/patroni/dcs/__init__.py b/patroni/dcs/__init__.py
index a61c6b59..f07ce6cd 100644
--- a/patroni/dcs/__init__.py
+++ b/patroni/dcs/__init__.py
@@ -15,7 +15,9 @@ from collections import defaultdict
 from copy import deepcopy
 from random import randint
 from threading import Event, Lock
-from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING
+from types import ModuleType
+from typing import Any, Callable, Collection, Dict, List, NamedTuple, Optional, Set, Tuple, Union, TYPE_CHECKING, \
+    Type, Iterator
 from urllib.parse import urlparse, urlunparse, parse_qsl
 
 from ..exceptions import PatroniFatalException
@@ -85,38 +87,83 @@ def dcs_modules() -> List[str]:
         return [module_prefix + name for _, name, is_pkg in pkgutil.iter_modules([dcs_dirname]) if not is_pkg]
 
 
-def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
-    modules = dcs_modules()
+def iter_dcs_classes(
+        config: Optional[Union['Config', Dict[str, Any]]] = None
+) -> Iterator[Tuple[str, Type['AbstractDCS']]]:
+    """Attempt to import DCS modules that are present in the given configuration.
+
+    .. note::
+            If a module successfully imports we can assume that all its requirements are installed.
+
+    :param config: configuration information with possible DCS names as keys. If given, only attempt to import DCS
+                   modules defined in the configuration. Else, if ``None``, attempt to import any supported DCS module.
+
+    :yields: a tuple containing the module ``name`` and the imported DCS class object.
+    """
+    for mod_name in dcs_modules():
+        name = mod_name.rpartition('.')[2]
+        if config is None or name in config:
 
-    for module_name in modules:
-        name = module_name.split('.')[-1]
-        if name in config:  # we will try to import only modules which have configuration section in the config file
             try:
-                module = importlib.import_module(module_name)
-                for key, item in module.__dict__.items():  # iterate through the module content
-                    # try to find implementation of AbstractDCS interface, class name must match with module_name
-                    if key.lower() == name and inspect.isclass(item) and issubclass(item, AbstractDCS):
-                        # propagate some parameters
-                        config[name].update({p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
-                                             'patronictl', 'ttl', 'retry_timeout') if p in config})
-                        # From citus section we only need "group" parameter, but will propagate everything just in case.
-                        if isinstance(config.get('citus'), dict):
-                            config[name].update(config['citus'])
-                        return item(config[name])
+                module = importlib.import_module(mod_name)
+                dcs_module = find_dcs_class_in_module(module)
+                if dcs_module:
+                    yield name, dcs_module
+
             except ImportError:
-                logger.debug('Failed to import %s', module_name)
+                logger.log(logging.DEBUG if config is not None else logging.INFO,
+                           'Failed to import %s', mod_name)
 
-    available_implementations: List[str] = []
-    for module_name in modules:
-        name = module_name.split('.')[-1]
-        try:
-            module = importlib.import_module(module_name)
-            available_implementations.extend(name for key, item in module.__dict__.items() if key.lower() == name
-                                             and inspect.isclass(item) and issubclass(item, AbstractDCS))
-        except ImportError:
-            logger.info('Failed to import %s', module_name)
-    raise PatroniFatalException("""Can not find suitable configuration of distributed configuration store
-Available implementations: """ + ', '.join(sorted(set(available_implementations))))
+
+def find_dcs_class_in_module(module: ModuleType) -> Optional[Type['AbstractDCS']]:
+    """Try to find the implementation of :class:`AbstractDCS` interface in *module* matching the *module* name.
+
+    :param module: Imported DCS module.
+
+    :returns: class with a name matching the name of *module* that implements :class:`AbstractDCS` or ``None`` if not
+              found.
+    """
+    module_name = module.__name__.rpartition('.')[2]
+    return next(
+        (obj for obj_name, obj in module.__dict__.items()
+         if (obj_name.lower() == module_name
+             and inspect.isclass(obj) and issubclass(obj, AbstractDCS))),
+        None)
+
+
+def get_dcs(config: Union['Config', Dict[str, Any]]) -> 'AbstractDCS':
+    """Attempt to load a Distributed Configuration Store from known available implementations.
+
+    .. note::
+        Using the list of available DCS modules returned by :func:`iter_dcs_modules` attempt to dynamically import and
+        instantiate the class that implements a DCS using the abstract class :class:`AbstractDCS`.
+
+        Basic top-level configuration parameters retrieved from *config* are propagated to the DCS specific config
+        before being passed to the module DCS class.
+
+        If no module is found to satisfy configuration then report and log an error. This will cause Patroni to exit.
+
+    :raises :exc:`PatroniFatalException`: if a load of all available DCS modules have been tried and none succeeded.
+
+    :param config: object or dictionary with Patroni configuration. This is normally a representation of the main
+                   Patroni
+
+    :returns: The first successfully loaded DCS module which is an implementation of :class:`AbstractDCS`.
+    """
+    for name, dcs_class in iter_dcs_classes(config):
+        # Propagate some parameters from top level of config if defined to the DCS specific config section.
+        config[name].update({
+            p: config[p] for p in ('namespace', 'name', 'scope', 'loop_wait',
+                                   'patronictl', 'ttl', 'retry_timeout')
+            if p in config})
+        # From citus section we only need "group" parameter, but will propagate everything just in case.
+        if isinstance(config.get('citus'), dict):
+            config[name].update(config['citus'])
+        return dcs_class(config[name])
+
+    raise PatroniFatalException(
+        f"Can not find suitable configuration of distributed configuration store\n"
+        f"Available implementations: {', '.join(sorted([n for n, _ in iter_dcs_classes()]))}")
 
 
 _Version = Union[int, str]
@@ -239,23 +286,26 @@ class Member(NamedTuple):
 
 
 class RemoteMember(Member):
-    """Represents a remote member (typically a primary) for a standby cluster"""
+    """Represents a remote member (typically a primary) for a standby cluster.
+
+    :cvar ALLOWED_KEYS: Controls access to relevant key names that could be in stored :attr:`~RemoteMember.data`.
+    """
+
+    ALLOWED_KEYS: Tuple[str, ...] = (
+        'primary_slot_name',
+        'create_replica_methods',
+        'restore_command',
+        'archive_cleanup_command',
+        'recovery_min_apply_delay',
+        'no_replication_slot'
+    )
 
     @classmethod
     def from_name_and_data(cls, name: str, data: Dict[str, Any]) -> 'RemoteMember':
         return super(RemoteMember, cls).__new__(cls, -1, name, None, data)
 
-    @staticmethod
-    def allowed_keys() -> Tuple[str, ...]:
-        return ('primary_slot_name',
-                'create_replica_methods',
-                'restore_command',
-                'archive_cleanup_command',
-                'recovery_min_apply_delay',
-                'no_replication_slot')
-
     def __getattr__(self, name: str) -> Any:
-        if name in RemoteMember.allowed_keys():
+        if name in RemoteMember.ALLOWED_KEYS:
             return self.data.get(name)
 
 
@@ -583,24 +633,25 @@ class Cluster(NamedTuple):
 
     def get_replication_slots(self, my_name: str, role: str, nofailover: bool,
                               major_version: int, show_error: bool = False) -> Dict[str, Dict[str, Any]]:
-        # if the replicatefrom tag is set on the member - we should not create the replication slot for it on
-        # the current primary, because that member would replicate from elsewhere. We still create the slot if
-        # the replicatefrom destination member is currently not a member of the cluster (fallback to the
-        # primary), or if replicatefrom destination member happens to be the current primary
-        use_slots = self.use_slots
-        if role in ('master', 'primary', 'standby_leader'):
-            slot_members = [m.name for m in self.members if use_slots and m.name != my_name
-                            and (m.replicatefrom is None or m.replicatefrom == my_name
-                                 or not self.has_member(m.replicatefrom))]
-            permanent_slots = self.__permanent_slots if use_slots and \
-                role in ('master', 'primary') else self.__permanent_physical_slots
-        else:
-            # only manage slots for replicas that replicate from this one, except for the leader among them
-            slot_members = [m.name for m in self.members if use_slots
-                            and m.replicatefrom == my_name and m.name != self.leader_name]
-            permanent_slots = self.__permanent_logical_slots if use_slots and not nofailover else {}
+        """Lookup configured slot names in the DCS, report issues found and merge with permanent slots.
+
+        Will log an error if:
+
+            * Conflicting slot names between members are found
+            * Any logical slots are disabled, due to version compatibility, and *show_error* is ``True``.
+
+        :param my_name: name of this node.
+        :param role: role of this node.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param major_version: postgresql major version.
+        :param show_error: if ``True`` report error if any disabled logical slots or conflicting slot names are found.
+
+        :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
+        """
+        slot_members: List[str] = self._get_slot_members(my_name, role) if self.use_slots else []
 
-        slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
+        slots: Dict[str, Dict[str, str]] = {slot_name_from_member_name(name): {'type': 'physical'}
+                                            for name in slot_members}
 
         if len(slots) < len(slot_members):
             # Find which names are conflicting for a nicer error message
@@ -608,11 +659,38 @@ class Cluster(NamedTuple):
             for name in slot_members:
                 slot_conflicts[slot_name_from_member_name(name)].append(name)
             logger.error("Following cluster members share a replication slot name: %s",
-                         "; ".join("{} map to {}".format(", ".join(v), k)
+                         "; ".join(f"{', '.join(v)} map to {k}"
                                    for k, v in slot_conflicts.items() if len(v) > 1))
 
-        # "merge" replication slots for members with permanent_replication_slots
+        permanent_slots: dict[str, Any] = self._get_permanent_slots(role, nofailover) if self.use_slots else {}
+        disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
+            slots, permanent_slots, my_name, major_version)
+
+        if disabled_permanent_logical_slots and show_error:
+            logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
+                         "Following slots will not be created: %s.", disabled_permanent_logical_slots)
+
+        return slots
+
+    @staticmethod
+    def _merge_permanent_slots(slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], my_name: str,
+                               major_version: int) -> List[str]:
+        """Merge replication *slots* for members with *permanent_slots*.
+
+        Perform validation of configured permanent slot name, skipping invalid names.
+
+        Will update *slots* in-line based on ``type`` of slot, ``physical`` or ``logical``, and name of node.
+        Type is assumed to be ``physical`` if there are no attributes stored as the slot value.
+
+        :param slots: Slot names with existing attributes if known.
+        :param my_name: name of this node.
+        :param permanent_slots: dictionary containing slot name key and slot information values.
+        :param major_version: postgresql major version.
+
+        :returns: List of disabled permanent, logical slot names, if postgresql version < 11.
+        """
         disabled_permanent_logical_slots: List[str] = []
+
         for name, value in permanent_slots.items():
             if not slot_name_re.match(name):
                 logger.error("Invalid permanent replication slot name '%s'", name)
@@ -629,7 +707,8 @@ class Cluster(NamedTuple):
                     if name != slot_name_from_member_name(my_name):
                         slots[name] = value
                     continue
-                elif value['type'] == 'logical' and value.get('database') and value.get('plugin'):
+
+                if value['type'] == 'logical' and value.get('database') and value.get('plugin'):
                     if major_version < 110000:
                         disabled_permanent_logical_slots.append(name)
                     elif name in slots:
@@ -640,12 +719,62 @@ class Cluster(NamedTuple):
                     continue
 
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
+        return disabled_permanent_logical_slots
 
-        if disabled_permanent_logical_slots and show_error:
-            logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
-                         "Following slots will not be created: %s.", disabled_permanent_logical_slots)
+    def _get_permanent_slots(self, role: str, nofailover: bool) -> Dict[str, Any]:
+        """Get configured permanent slot names.
 
-        return slots
+        .. note::
+            Permanent logical replication slots are only considered if ``use_slots`` configuration is enabled. Also,
+            only considered if *role* is ``primary`` or if it is a promotable ``replica`` -- what excludes a
+            ``standby_leader`` or ``replica`` with ``nofailover`` tag enabled. That combination is used for failing
+            over logical replication slots, and the latter nodes are not eligible for such task.
+
+            Permanent physical slots are only considered if *role* is ``primary`` or ``standby_leader``, independently
+            if ``use_slots`` is enabled or not. That is done that way because even if Patroni itself is not using slots
+            to replicate among its members when ``use_slots`` is disabled, the user may still have configured Patroni to
+            keep permanent physical slots used out of Patroni.
+
+        :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
+                     or logical slots being consumed.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+
+        :returns: dictionary of permanent slot names mapped to attributes.
+        """
+        if role in ('master', 'primary', 'standby_leader'):
+            permanent_slots = (self.__permanent_slots
+                               if role in ('master', 'primary')
+                               else self.__permanent_physical_slots)
+        else:
+            permanent_slots = self.__permanent_logical_slots if not nofailover else {}
+        return permanent_slots
+
+    def _get_slot_members(self, my_name: str, role: str) -> List[str]:
+        """Get a list of member names that have replication slots sourcing from this node.
+
+        If the ``replicatefrom`` tag is set on the member - we should not create the replication slot for it on
+        the current primary, because that member would replicate from elsewhere. We still create the slot if
+        the ``replicatefrom`` destination member is currently not a member of the cluster (fallback to the
+        primary), or if ``replicatefrom`` destination member happens to be the current primary.
+
+        :param my_name: name of this node.
+        :param role: role of this node, if this is a ``primary`` or ``standby_leader`` return list of members
+                     replicating from this node. If not then return a list of members replicating as cascaded
+                     replicas from this node.
+
+        :returns: list of member names.
+        """
+        if role in ('master', 'primary', 'standby_leader'):
+            slot_members = [m.name for m in self.members
+                            if m.name != my_name
+                            and (m.replicatefrom is None
+                                 or m.replicatefrom == my_name
+                                 or not self.has_member(m.replicatefrom))]
+        else:
+            # only manage slots for replicas that replicate from this one, except for the leader among them
+            slot_members = [m.name for m in self.members
+                            if m.replicatefrom == my_name and m.name != self.leader_name]
+        return slot_members
 
     def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool:
         if major_version < 110000:
diff --git a/patroni/ha.py b/patroni/ha.py
index 3108f787..5894824c 100644
--- a/patroni/ha.py
+++ b/patroni/ha.py
@@ -149,7 +149,6 @@ class Ha(object):
         self._leader_timeline = None
         self.recovering = False
         self._async_response = CriticalTask()
-        self._crash_recovery_executed = False
         self._crash_recovery_started = 0
         self._start_timeout = None
         self._async_executor = AsyncExecutor(self.state_handler.cancellable, self.wakeup)
@@ -307,10 +306,13 @@ class Ha(object):
             if self._async_executor.scheduled_action in (None, 'promote') \
                     and data['state'] in ['running', 'restarting', 'starting']:
                 try:
-                    timeline: Optional[int]
                     timeline, wal_position, pg_control_timeline = self.state_handler.timeline_wal_position()
                     data['xlog_location'] = wal_position
-                    if not timeline:  # try pg_stat_wal_receiver to get the timeline
+                    if not timeline:  # running as a standby
+                        replication_state = self.state_handler.replication_state()
+                        if replication_state:
+                            data['replication_state'] = replication_state
+                        # try pg_stat_wal_receiver to get the timeline
                         timeline = self.state_handler.received_timeline()
                     if not timeline:
                         # So far the only way to get the current timeline on the standby is from
@@ -411,8 +413,7 @@ class Ha(object):
         return result
 
     def _handle_crash_recovery(self) -> Optional[str]:
-        if not self._crash_recovery_executed and (self.cluster.is_unlocked() or self._rewind.can_rewind):
-            self._crash_recovery_executed = True
+        if self._crash_recovery_started == 0 and (self.cluster.is_unlocked() or self._rewind.can_rewind):
             self._crash_recovery_started = time.time()
             msg = 'doing crash recovery in a single user mode'
             return self._async_executor.try_run_async(msg, self._rewind.ensure_clean_shutdown) or msg
@@ -438,15 +439,29 @@ class Ha(object):
             return self._async_executor.try_run_async(msg, self._do_reinitialize, args=(self.cluster,)) or msg
 
     def recover(self) -> str:
-        # Postgres is not running and we will restart in standby mode. Watchdog is not needed until we promote.
-        self.watchdog.disable()
-
+        """Handle the case when postgres isn't running.
+
+        Depending on the state of Patroni, DCS cluster view, and pg_controldata the following could happen:
+        - if ``primary_start_timeout`` is 0 and this node owns the leader lock, the lock
+          will be voluntarily released if there are healthy replicas to take it over.
+        - if postgres was running as a ``primary`` and this node owns the leader lock, postgres is started as primary.
+        - crash recover in a single-user mode is executed in the following cases:
+          - postgres was running as ``primary`` wasn't ``shut down`` cleanly and there is no leader in DCS
+          - postgres was running as ``replica`` wasn't ``shut down in recovery`` (cleanly)
+            and we need to run ``pg_rewind`` to join back to the cluster.
+        - ``pg_rewind`` is executed if it is necessary, or optinally, the data directory could
+           be removed if it is allowed by configuration.
+        - after ``crash recovery`` and/or ``pg_rewind`` are executed, postgres is started in recovery.
+
+        :returns: action message, describing what was performed.
+        """
         if self.has_lock() and self.update_lock():
             timeout = self.global_config.primary_start_timeout
             if timeout == 0:
                 # We are requested to prefer failing over to restarting primary. But see first if there
                 # is anyone to fail over to.
                 if self.is_failover_possible(self.cluster.members):
+                    self.watchdog.disable()
                     logger.info("Primary crashed. Failing over.")
                     self.demote('immediate')
                     return 'stopped PostgreSQL to fail over after a crash'
@@ -455,6 +470,24 @@ class Ha(object):
 
         data = self.state_handler.controldata()
         logger.info('pg_controldata:\n%s\n', '\n'.join('  {0}: {1}'.format(k, v) for k, v in data.items()))
+
+        # timeout > 0 indicates that we still have the leader lock, and it was just updated
+        if timeout\
+                and data.get('Database cluster state') in ('in production', 'in crash recovery',
+                                                           'shutting down', 'shut down')\
+                and self.state_handler.state == 'crashed'\
+                and self.state_handler.role in ('primary', 'master')\
+                and not self.state_handler.config.recovery_conf_exists():
+            # We know 100% that we were running as a primary a few moments ago, therefore could just start postgres
+            msg = 'starting primary after failure'
+            if self._async_executor.try_run_async(msg, self.state_handler.start,
+                                                  args=(timeout, self._async_executor.critical_task)) is None:
+                self.recovering = True
+                return msg
+
+        # Postgres is not running, and we will restart in standby mode. Watchdog is not needed until we promote.
+        self.watchdog.disable()
+
         if data.get('Database cluster state') in ('in production', 'shutting down', 'in crash recovery'):
             msg = self._handle_crash_recovery()
             if msg:
@@ -965,9 +998,6 @@ class Ha(object):
             if ret is not None:  # continue if we just deleted the stale failover key as a leader
                 return ret
 
-        if self.state_handler.is_starting():  # postgresql still starting up is unhealthy
-            return False
-
         if self.state_handler.is_leader():
             # in pause leader is the healthiest only when no initialize or sysid matches with initialize!
             return not self.is_paused() or not self.cluster.initialize\
@@ -1055,7 +1085,9 @@ class Ha(object):
             # It could happen if Postgres is still archiving the backlog of WAL files.
             # If we know that there are replicas that received the shutdown checkpoint
             # location, we can remove the leader key and allow them to start leader race.
-            if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location):
+
+            # for a manual failover/switchover with a candidate, we should check the requested candidate only
+            if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location):
                 self.state_handler.set_role('demoted')
                 with self._async_executor:
                     self.release_leader_key_voluntarily(checkpoint_location)
@@ -1159,15 +1191,12 @@ class Ha(object):
                     logger.warning('Failover is possible only to a specific candidate in a paused state')
                 else:
                     if self.is_synchronous_mode():
-                        if failover.candidate and not self.cluster.sync.matches(failover.candidate):
+                        members = self.get_failover_candidates(check_sync=True)
+                        if failover.candidate and not members:
                             logger.warning('Failover candidate=%s does not match with sync_standbys=%s',
                                            failover.candidate, self.cluster.sync.sync_standby)
-                            members = []
-                        else:
-                            members = [m for m in self.cluster.members if self.cluster.sync.matches(m.name)]
                     else:
-                        members = [m for m in self.cluster.members
-                                   if not failover.candidate or m.name == failover.candidate]
+                        members = self.get_failover_candidates()
                     if self.is_failover_possible(members, False):  # check that there are healthy members
                         ret = self._async_executor.try_run_async('manual failover: demote', self.demote, ('graceful',))
                         return ret or 'manual failover: demoting myself'
@@ -1451,7 +1480,7 @@ class Ha(object):
 
             if self.state_handler.role in ('master', 'primary'):
                 logger.info('Demoting primary during %s', self._async_executor.scheduled_action)
-                if self._async_executor.scheduled_action == 'restart':
+                if self._async_executor.scheduled_action in ('restart', 'starting primary after failure'):
                     # Restart needs a special interlocking cancel because postmaster may be just started in a
                     # background thread and has not even written a pid file yet.
                     with self._async_executor.critical_task as task:
@@ -1515,6 +1544,9 @@ class Ha(object):
         self.dcs.set_config_value(json.dumps(self.patroni.config.dynamic_configuration, separators=(',', ':')))
         self.dcs.take_leader()
         self.set_is_leader(True)
+        if self.is_synchronous_mode():
+            self.state_handler.sync_handler.set_synchronous_standby_names(
+                CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
         self.state_handler.call_nowait(CallbackAction.ON_START)
         self.load_cluster_from_dcs()
 
@@ -1616,7 +1648,7 @@ class Ha(object):
                         return msg
 
                 # Reset some states after postgres successfully started up
-                self._crash_recovery_executed = False
+                self._crash_recovery_started = 0
                 if self._rewind.executed and not self._rewind.failed:
                     self._rewind.reset_state()
 
@@ -1708,16 +1740,21 @@ class Ha(object):
                 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 primary)
-            if not self._async_executor.busy and not self.state_handler.is_starting():
+            # We might not have a valid PostgreSQL connection here if AsyncExecutor is doing
+            # something with PostgreSQL. Therefore we will sync replication slots only if no
+            # asynchronous processes are running or we know that this is a standby being promoted.
+            # But, we don't want to run pg_rewind checks or copy logical slots from itself,
+            # therefore we have a couple additional `not is_promoting` checks.
+            is_promoting = self._async_executor.scheduled_action == 'promote'
+            if (not self._async_executor.busy or is_promoting) 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():
+                    if not is_promoting and not self.state_handler.is_leader():
                         self._rewind.trigger_check_diverged_lsn()
                     self.state_handler.call_nowait(CallbackAction.ON_START)
-                if create_slots and self.cluster.leader:
+
+                if not is_promoting and 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))
@@ -1811,7 +1848,9 @@ class Ha(object):
                     # It could happen if Postgres is still archiving the backlog of WAL files.
                     # If we know that there are replicas that received the shutdown checkpoint
                     # location, we can remove the leader key and allow them to start leader race.
-                    if self.is_failover_possible(self.cluster.members, cluster_lsn=checkpoint_location):
+
+                    # for a manual failover/switchover with a candidate, we should check the requested candidate only
+                    if self.is_failover_possible(self.get_failover_candidates(), cluster_lsn=checkpoint_location):
                         self.dcs.delete_leader(checkpoint_location)
                         status['deleted'] = True
                     else:
@@ -1862,7 +1901,7 @@ class Ha(object):
         cluster_params = self.global_config.get_standby_cluster_config()
 
         if cluster_params:
-            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.allowed_keys()})
+            data.update({k: v for k, v in cluster_params.items() if k in RemoteMember.ALLOWED_KEYS})
             data['no_replication_slot'] = 'primary_slot_name' not in cluster_params
             conn_kwargs = member.conn_kwargs() if member else \
                 {k: cluster_params[k] for k in ('host', 'port') if k in cluster_params}
@@ -1871,3 +1910,24 @@ class Ha(object):
 
         name = member.name if member else 'remote_member:{}'.format(uuid.uuid1())
         return RemoteMember.from_name_and_data(name, data)
+
+    def get_failover_candidates(self, check_sync: bool = False) -> List[Member]:
+        """Return list of candidates for either manual or automatic failover.
+
+        Mainly used to later be passed to ``Ha.is_failover_possible()``.
+
+        :param check_sync: if ``True``, also check against the sync key members
+
+        :returns: a list of ``Member`` ojects or an empty list if there is no candidate available
+        """
+        failover = self.cluster.failover
+        if check_sync:
+            # TODO: allow manual failover (=no leader specified) to async node
+            # every sync_standby or the candidate specified if is in sync_standbys
+            return [m for m in self.cluster.members
+                    if self.cluster.sync.matches(m.name)
+                    and (not failover or not failover.candidate or m.name == failover.candidate)]
+        else:
+            # every member or the candidate specified
+            return [m for m in self.cluster.members
+                    if not failover or not failover.candidate or m.name == failover.candidate]
diff --git a/patroni/postgresql/__init__.py b/patroni/postgresql/__init__.py
index b2423c89..632cbee7 100644
--- a/patroni/postgresql/__init__.py
+++ b/patroni/postgresql/__init__.py
@@ -197,18 +197,19 @@ class Postgresql(object):
                         and self.role in ('master', 'primary', 'promoted') else "'on', '', NULL")
 
         if self._major_version >= 90600:
-            extra = ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
-                     "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
-                     " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
-                     if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
+            extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
+                ", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
+                        "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
+                        " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
-                     " slot_name, conninfo, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
+                     " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
             if self.role == 'standby_leader':
                 extra = "timeline_id" + extra + ", pg_catalog.pg_control_checkpoint()"
             else:
                 extra = "0" + extra
         else:
-            extra = "0, NULL, NULL, NULL, NULL" + extra
+            extra = "0, NULL, NULL, NULL, NULL, NULL, NULL" + extra
 
         return ("SELECT " + self.TL_LSN + ", {2}").format(self.wal_name, self.lsn_name, extra)
 
@@ -426,7 +427,8 @@ class Postgresql(object):
                 result = self._is_leader_retry(self._query, self.cluster_info_query).fetchone()
                 cluster_info_state = dict(zip(['timeline', 'wal_position', 'replayed_location',
                                                'received_location', 'replay_paused', 'pg_control_timeline',
-                                               'received_tli', 'slot_name', 'conninfo', 'slots', 'synchronous_commit',
+                                               'received_tli', 'slot_name', 'conninfo', 'receiver_state',
+                                               'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
                 if self._has_permanent_logical_slots:
                     cluster_info_state['slots'] =\
@@ -472,6 +474,41 @@ class Postgresql(object):
         """:returns: a result set of 'SELECT * FROM pg_stat_replication'."""
         return self._cluster_info_state_get('pg_stat_replication') or []
 
+    def replication_state_from_parameters(self, is_leader: bool, receiver_state: Optional[str],
+                                          restore_command: Optional[str]) -> Optional[str]:
+        """Figure out the replication state from input parameters.
+
+        .. note::
+            This method could be only called when Postgres is up, running and queries are successfuly executed.
+
+        :is_leader: `True` is postgres is not running in recovery
+        :receiver_state: value from `pg_stat_get_wal_receiver.state` or None if Postgres is older than 9.6
+        :restore_command: value of ``restore_command`` GUC for PostgreSQL 12+ or
+                          `postgresql.recovery_conf.restore_command` if it is set in Patroni configuration
+
+        :returns: - `None` for the primary and for Postgres older than 9.6;
+                  - 'streaming' if replica is streaming according to the `pg_stat_wal_receiver` view;
+                  - 'in archive recovery' if replica isn't streaming and there is a `restore_command`
+        """
+        if self._major_version >= 90600 and not is_leader:
+            if receiver_state == 'streaming':
+                return 'streaming'
+            # For Postgres older than 12 we get `restore_command` from Patroni config, otherwise we check GUC
+            if self._major_version < 120000 and self.config.restore_command() or restore_command:
+                return 'in archive recovery'
+
+    def replication_state(self) -> Optional[str]:
+        """Checks replication state from `pg_stat_get_wal_receiver()`.
+
+        .. note::
+            Available only since 9.6
+
+        :returns: ``streaming``, ``in archive recovery``, or ``None``
+        """
+        return self.replication_state_from_parameters(self.is_leader(),
+                                                      self._cluster_info_state_get('receiver_state'),
+                                                      self._cluster_info_state_get('restore_command'))
+
     def is_leader(self) -> bool:
         try:
             return bool(self._cluster_info_state_get('timeline'))
diff --git a/patroni/postgresql/bootstrap.py b/patroni/postgresql/bootstrap.py
index a76f9f5f..6e25012b 100644
--- a/patroni/postgresql/bootstrap.py
+++ b/patroni/postgresql/bootstrap.py
@@ -185,7 +185,7 @@ class Bootstrap(object):
                 r['host'] = 'localhost'  # set it to localhost to write into pgpass
 
             env = self._postgresql.config.write_pgpass(r)
-            env['PGOPTIONS'] = '-c synchronous_commit=local'
+            env['PGOPTIONS'] = '-c synchronous_commit=local -c statement_timeout=0'
 
             try:
                 ret = self._postgresql.cancellable.call(shlex.split(cmd) + [connstring], env=env)
diff --git a/patroni/postgresql/config.py b/patroni/postgresql/config.py
index ef3f5c36..9461714f 100644
--- a/patroni/postgresql/config.py
+++ b/patroni/postgresql/config.py
@@ -1177,3 +1177,6 @@ class ConfigHandler(object):
 
     def get(self, key: str, default: Optional[Any] = None) -> Optional[Any]:
         return self._config.get(key, default)
+
+    def restore_command(self) -> Optional[str]:
+        return (self.get('recovery_conf') or {}).get('restore_command')
diff --git a/patroni/postgresql/rewind.py b/patroni/postgresql/rewind.py
index f70ad065..ff6fc751 100644
--- a/patroni/postgresql/rewind.py
+++ b/patroni/postgresql/rewind.py
@@ -370,7 +370,7 @@ class Rewind(object):
                 # it is the author of archive_command, who is responsible
                 # for not overriding the WALs already present in archive
                 logger.info('Trying to archive %s: %s', wal, cmd)
-                if self._postgresql.cancellable.call(shlex.split(cmd)) == 0:
+                if self._postgresql.cancellable.call([cmd], shell=True) == 0:
                     new_name = os.path.join(status_dir, wal + '.done')
                     try:
                         shutil.move(old_name, new_name)
diff --git a/patroni/postgresql/slots.py b/patroni/postgresql/slots.py
index b0a53561..fde81a7a 100644
--- a/patroni/postgresql/slots.py
+++ b/patroni/postgresql/slots.py
@@ -21,8 +21,24 @@ logger = logging.getLogger(__name__)
 
 
 def compare_slots(s1: Dict[str, Any], s2: Dict[str, Any], dbid: str = 'database') -> bool:
-    return s1['type'] == s2['type'] and (s1['type'] == 'physical'
-                                         or s1.get(dbid) == s2.get(dbid) and s1['plugin'] == s2['plugin'])
+    """Compare 2 replication slot objects for equality.
+
+    ..note ::
+        If the first argument is a ``physical`` replication slot then only the `type` of the second slot is compared.
+        If the first argument is another ``type`` (e.g. ``logical``) then *dbid* and ``plugin`` are compared.
+
+    :param s1: First slot dictionary to be compared.
+    :param s2: Second slot dictionary to be compared.
+    :param dbid: Optional attribute to be compared when comparing ``logical`` replication slots.
+
+    :return: ``True`` if the slot ``type`` of *s1* and *s2* is matches, and the ``type`` of *s1* is ``physical``,
+             OR the ``types`` match AND the *dbid* and ``plugin`` attributes are equal.
+
+    """
+    return (s1['type'] == s2['type']
+            and (s1['type'] == 'physical'
+                 or s1.get(dbid) == s2.get(dbid)
+                 and s1['plugin'] == s2['plugin']))
 
 
 class SlotsAdvanceThread(Thread):
@@ -114,7 +130,7 @@ class SlotsHandler(object):
         self._postgresql = postgresql
         self._advance = None
         self._replication_slots: Dict[str, Dict[str, Any]] = {}  # already existing replication slots
-        self._unready_logical_slots: Dict[str, Optional[int]] = {}
+        self._logical_slots_processing_queue: Dict[str, Optional[int]] = {}
         self.pg_replslot_dir = os.path.join(self._postgresql.data_dir, 'pg_replslot')
         self.schedule()
 
@@ -174,7 +190,8 @@ class SlotsHandler(object):
             self._replication_slots = replication_slots
             self._schedule_load_slots = False
             if self._force_readiness_check:
-                self._unready_logical_slots = {n: None for n, v in replication_slots.items() if v['type'] == 'logical'}
+                self._logical_slots_processing_queue = {n: None for n, v in replication_slots.items()
+                                                        if v['type'] == 'logical'}
                 self._force_readiness_check = False
 
     def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool:
@@ -311,10 +328,10 @@ class SlotsHandler(object):
                 self._ensure_physical_slots(slots)
 
                 if self._postgresql.is_leader():
-                    self._unready_logical_slots.clear()
+                    self._logical_slots_processing_queue.clear()
                     self._ensure_logical_slots_primary(slots)
                 elif cluster.slots and slots:
-                    self.check_logical_slots_readiness(cluster, nofailover, replicatefrom)
+                    self.check_logical_slots_readiness(cluster, replicatefrom)
 
                     ret = self._ensure_logical_slots_replica(cluster, slots)
 
@@ -331,48 +348,100 @@ class SlotsHandler(object):
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
             yield cur
 
-    def check_logical_slots_readiness(self, cluster: Cluster, nofailover: bool, replicatefrom: Optional[str]) -> None:
+    def check_logical_slots_readiness(self, cluster: Cluster, replicatefrom: Optional[str]) -> bool:
+        """Determine whether all known logical slots are synchronised from the leader.
+
+         1) Retrieve the current ``catalog_xmin`` value for the physical slot from the cluster leader, and
+         2) using previously stored list of "unready" logical slots, those which have yet to be checked hence have no
+            stored slot attributes,
+         3) store logical slot ``catalog_xmin`` when the physical slot ``catalog_xmin`` becomes valid.
+
+         :param cluster: object containing stateful information for the cluster.
+         :param replicatefrom: name of the member that should be used to replicate from.
+
+         :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise.
+         """
         catalog_xmin = None
-        if self._unready_logical_slots and cluster.leader:
+        if self._logical_slots_processing_queue and cluster.leader:
             slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
             try:
                 with self._get_leader_connection_cursor(cluster.leader) as cur:
                     cur.execute("SELECT slot_name, catalog_xmin FROM pg_catalog.pg_get_replication_slots()"
                                 " WHERE NOT pg_catalog.pg_is_in_recovery() AND slot_name = ANY(%s)",
-                                ([n for n, v in self._unready_logical_slots.items() if v is None] + [slot_name],))
+                                ([n for n, v in self._logical_slots_processing_queue.items()
+                                  if v is None] + [slot_name],))
                     slots = {row[0]: row[1] for row in cur}
                     if slot_name not in slots:
-                        return logger.warning('Physical slot %s does not exist on the primary', slot_name)
+                        logger.warning('Physical slot %s does not exist on the primary', slot_name)
+                        return False
                     catalog_xmin = slots.pop(slot_name)
             except Exception as e:
-                return logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
-            # Remember catalog_xmin of logical slots on the primary when catalog_xmin of
-            # the physical slot became valid. Logical slots on replica will be safe to use after
-            # promote when catalog_xmin of the physical slot overtakes these values.
-            if catalog_xmin is not None:
-                for name, value in slots.items():
-                    self._unready_logical_slots[name] = value
-            else:  # Replica isn't streaming or the hot_standby_feedback isn't enabled
-                try:
-                    cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
-                    row = cur.fetchone()
-                    if row and not row[0]:
-                        logger.error('Logical slot failover requires "hot_standby_feedback".'
-                                     ' Please check postgresql.auto.conf')
-                except Exception as e:
-                    logger.error('Failed to check the hot_standby_feedback setting: %r', e)
-                return  # since `catalog_xmin` isn't valid further checks don't make any sense
-
-        for name in list(self._unready_logical_slots):
-            value = self._replication_slots.get(name)
-            # The logical slot on a replica is safe to use when the physical replica slot on the primary:
-            # 1. has a nonzero/non-null catalog_xmin
-            # 2. has a catalog_xmin that is not newer (greater) than the catalog_xmin of any slot on the standby
-            # 3. overtook the catalog_xmin of remembered values of logical slots on the primary.
-            if not value or catalog_xmin is not None and\
-                    self._unready_logical_slots[name] <= catalog_xmin <= value['catalog_xmin']:
-                del self._unready_logical_slots[name]
-                if value:
+                logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+                return False
+
+            if not self._update_pending_logical_slot_primary(slots, catalog_xmin):
+                return False  # since `catalog_xmin` isn't valid further checks don't make any sense
+
+        self._ready_logical_slots(catalog_xmin)
+        return True
+
+    def _update_pending_logical_slot_primary(self, slots: Dict[str, Any], catalog_xmin: Optional[int] = None) -> bool:
+        """Store pending logical slot information for ``catalog_xmin`` on the primary.
+
+        Remember ``catalog_xmin`` of logical slots on the primary when ``catalog_xmin`` of the physical slot became
+        valid. Logical slots on replica will be safe to use after promote when ``catalog_xmin`` of the physical slot
+        overtakes these values.
+
+        :param slots: dictionary of slot information from the primary
+        :param catalog_xmin: ``catalog_xmin`` of the physical slot used by this replica to stream changes from primary.
+
+        :returns: ``False`` if any issue was faced while processing, ``True`` otherwise.
+        """
+        if catalog_xmin is not None:
+            for name, value in slots.items():
+                self._logical_slots_processing_queue[name] = value
+            return True
+
+        # Replica isn't streaming or the hot_standby_feedback isn't enabled
+        try:
+            cur = self._query("SELECT pg_catalog.current_setting('hot_standby_feedback')::boolean")
+            row = cur.fetchone()
+            if row and not row[0]:
+                logger.error('Logical slot failover requires "hot_standby_feedback".'
+                             ' Please check postgresql.auto.conf')
+        except Exception as e:
+            logger.error('Failed to check the hot_standby_feedback setting: %r', e)
+        return False
+
+    def _ready_logical_slots(self, primary_physical_catalog_xmin: Optional[int] = None) -> None:
+        """Ready logical slots by comparing primary physical slot ``catalog_xmin`` to logical ``catalog_xmin``.
+
+        The logical slot on a replica is safe to use when the physical replica slot on the primary:
+
+            1. has a nonzero/non-null ``catalog_xmin`` represented by ``primary_physical_xmin``.
+            2. has a ``catalog_xmin`` that is not newer (greater) than the ``catalog_xmin`` of any slot on the standby
+            3. overtook the ``catalog_xmin`` of remembered values of logical slots on the primary.
+
+        :param primary_physical_catalog_xmin: is the value retrieved from ``pg_catalog.pg_get_replication_slots()`` for
+                                              the physical replication slot on the primary.
+        """
+        # Make a copy of processing queue keys as a list as the queue dictionary is modified inside the loop.
+        for name in list(self._logical_slots_processing_queue):
+            primary_logical_catalog_xmin = self._logical_slots_processing_queue[name]
+            standby_logical_slot = self._replication_slots.get(name, {})
+            standby_logical_catalog_xmin = standby_logical_slot.get('catalog_xmin', 0)
+            if TYPE_CHECKING:  # pragma: no cover
+                assert primary_logical_catalog_xmin is not None
+
+            if (
+                    not standby_logical_slot
+                    or primary_physical_catalog_xmin is not None
+                    and primary_logical_catalog_xmin <= primary_physical_catalog_xmin <= standby_logical_catalog_xmin
+            ):
+
+                del self._logical_slots_processing_queue[name]
+
+                if standby_logical_slot:
                     logger.info('Logical slot %s is safe to be used after a failover', name)
 
     def copy_logical_slots(self, cluster: Cluster, create_slots: List[str]) -> None:
@@ -417,7 +486,7 @@ class SlotsHandler(object):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
-                self._unready_logical_slots[name] = None
+                self._logical_slots_processing_queue[name] = None
             fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
             self._postgresql.start()
 
@@ -430,6 +499,6 @@ class SlotsHandler(object):
         if self._advance:
             self._advance.on_promote()
 
-        if self._unready_logical_slots:
+        if self._logical_slots_processing_queue:
             logger.warning('Logical replication slots that might be unsafe to use after promote: %s',
-                           set(self._unready_logical_slots))
+                           set(self._logical_slots_processing_queue))
diff --git a/patroni/postgresql/sync.py b/patroni/postgresql/sync.py
index c56bdbcd..d0f28586 100644
--- a/patroni/postgresql/sync.py
+++ b/patroni/postgresql/sync.py
@@ -193,7 +193,12 @@ class SyncHandler(object):
 
         # 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
+        # Ensure some WAL traffic to move replication
+        self._postgresql.query("""DO $$
+BEGIN
+    SET local synchronous_commit = 'off';
+    PERFORM * FROM pg_catalog.txid_current();
+END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
@@ -289,6 +294,6 @@ class SyncHandler(object):
         # 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
+        # timeline == 0 -- indicates that this is the replica
         if self._postgresql.get_primary_timeline() > 0:
             self._handle_synchronous_standby_names_change()
diff --git a/patroni/utils.py b/patroni/utils.py
index ee9a3d21..b7694d17 100644
--- a/patroni/utils.py
+++ b/patroni/utils.py
@@ -773,7 +773,8 @@ def cluster_as_json(cluster: 'Cluster', global_config: Optional['GlobalConfig']
         else:
             role = 'replica'
 
-        member = {'name': m.name, 'role': role, 'state': m.data.get('state', ''), 'api_url': m.api_url}
+        state = (m.data.get('replication_state', '') if role != 'leader' else '') or m.data.get('state', '')
+        member = {'name': m.name, 'role': role, 'state': state, 'api_url': m.api_url}
         conn_kwargs = m.conn_kwargs()
         if conn_kwargs.get('host'):
             member['host'] = conn_kwargs['host']
diff --git a/patroni/validator.py b/patroni/validator.py
index fc6cdde5..e9afcc13 100644
--- a/patroni/validator.py
+++ b/patroni/validator.py
@@ -11,11 +11,12 @@ import shutil
 import socket
 import subprocess
 
-from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, TYPE_CHECKING
+from typing import Any, Dict, Union, Iterator, List, Optional as OptionalType, Tuple, TYPE_CHECKING
 
-from .utils import parse_int, split_host_port, data_directory_is_empty
+from .collections import CaseInsensitiveSet
 from .dcs import dcs_modules
 from .exceptions import ConfigParseError
+from .utils import parse_int, split_host_port, data_directory_is_empty
 
 
 def data_directory_empty(data_dir: str) -> bool:
@@ -802,6 +803,39 @@ class IntValidator(object):
         return ret
 
 
+class EnumValidator(object):
+    """Validate enum setting
+
+    :ivar allowed_values: a ``set`` or ``CaseInsensitiveSet`` object with allowed enum values.
+    :ivar raise_assert: if an ``assert`` call should be performed regarding expected type and valid range.
+    """
+
+    def __init__(self, allowed_values: Tuple[str, ...],
+                 case_sensitive: bool = False, raise_assert: bool = False) -> None:
+        """Create an :class:`EnumValidator` object with given allowed values.
+
+        :param allowed_values: a tuple with allowed enum values
+        :param case_sensitive: set to ``True`` to do case sensitive comparisons
+        :param raise_assert: if an ``assert`` call should be performed regarding expected values.
+        """
+        self.allowed_values = set(allowed_values) if case_sensitive else CaseInsensitiveSet(allowed_values)
+        self.raise_assert = raise_assert
+
+    def __call__(self, value: str) -> bool:
+        """Check if provided *value* could be found within *allowed_values*.
+
+        .. note::
+            If ``raise_assert`` is ``True`` and *value* is not valid, then an ``AssertionError`` will be triggered.
+        :param value: value to be checked.
+        :returns: ``True`` if *value* could be found within *allowed_values*.
+        """
+        ret = value in self.allowed_values
+
+        if self.raise_assert:
+            assert_(ret)
+        return ret
+
+
 def validate_watchdog_mode(value: Any) -> None:
     """Validate ``watchdog.mode`` configuration option.
 
@@ -827,15 +861,44 @@ validate_etcd = {
         "srv": str,
         "srv_suffix": str,
         "url": str,
-        "proxy": str})
+        "proxy": str
+    }),
+    Optional("protocol"): str,
+    Optional("username"): str,
+    Optional("password"): str,
+    Optional("cacert"): str,
+    Optional("cert"): str,
+    Optional("key"): str
 }
 
 schema = Schema({
     "name": str,
     "scope": str,
+    Optional("ctl"): {
+        Optional("insecure"): bool,
+        Optional("cacert"): str,
+        Optional("certfile"): str,
+        Optional("keyfile"): str,
+        Optional("keyfile_password"): str
+    },
     "restapi": {
         "listen": validate_host_port_listen,
         "connect_address": validate_connect_address,
+        Optional("authentication"): {
+            "username": str,
+            "password": str
+        },
+        Optional("certfile"): str,
+        Optional("keyfile"): str,
+        Optional("keyfile_password"): str,
+        Optional("cafile"): str,
+        Optional("ciphers"): str,
+        Optional("verify_client"): EnumValidator(("none", "optional", "required"),
+                                                 case_sensitive=True, raise_assert=True),
+        Optional("allowlist"): [str],
+        Optional("allowlist_include_members"): bool,
+        Optional("http_extra_headers"): dict,
+        Optional("https_extra_headers"): dict,
         Optional("request_queue_size"): IntValidator(min=0, max=4096, raise_assert=True)
     },
     Optional("bootstrap"): {
@@ -843,15 +906,64 @@ schema = Schema({
             Optional("ttl"): int,
             Optional("loop_wait"): int,
             Optional("retry_timeout"): int,
-            Optional("maximum_lag_on_failover"): int
+            Optional("maximum_lag_on_failover"): int,
+            Optional("maximum_lag_on_syncnode"): int,
+            Optional("postgresql"): {
+                Optional("parameters"): {
+                    Optional("max_connections"): int,
+                    Optional("max_locks_per_transaction"): int,
+                    Optional("max_prepared_transactions"): int,
+                    Optional("max_replication_slots"): int,
+                    Optional("max_wal_senders"): int,
+                    Optional("max_worker_processes"): int
+                },
+                Optional("use_pg_rewind"): bool,
+                Optional("pg_hba"): [str],
+                Optional("pg_ident"): [str],
+                Optional("pg_ctl_timeout"): int,
+                Optional("use_slots"): bool,
+            },
+            Optional("primary_start_timeout"): int,
+            Optional("primary_stop_timeout"): int,
+            Optional("standby_cluster"): {
+                Or("host", "port", "restore_command"): Case({
+                    "host": str,
+                    "port": int,
+                    "restore_command": str
+                }),
+                Optional("primary_slot_name"): str,
+                Optional("create_replica_methods"): [str],
+                Optional("archive_cleanup_command"): str,
+                Optional("recovery_min_apply_delay"): str
+            },
+            Optional("synchronous_mode"): bool,
+            Optional("synchronous_mode_strict"): bool,
+            Optional("synchronous_node_count"): int
         },
-        Optional("initdb"): [Or(str, dict)]
+        Optional("initdb"): [Or(str, dict)],
+        Optional("method"): str
     },
     Or(*available_dcs): Case({
         "consul": {
             Or("host", "url"): Case({
                 "host": validate_host_port,
-                "url": str})
+                "url": str
+            }),
+            Optional("port"): int,
+            Optional("scheme"): str,
+            Optional("token"): str,
+            Optional("verify"): bool,
+            Optional("cacert"): str,
+            Optional("cert"): str,
+            Optional("key"): str,
+            Optional("dc"): str,
+            Optional("checks"): [str],
+            Optional("register_service"): bool,
+            Optional("service_tags"): [str],
+            Optional("service_check_interval"): str,
+            Optional("service_check_tls_server_name"): str,
+            Optional("consistency"): EnumValidator(('default', 'consistent', 'stale'),
+                                                   case_sensitive=True, raise_assert=True)
         },
         "etcd": validate_etcd,
         "etcd3": validate_etcd,
@@ -869,15 +981,24 @@ schema = Schema({
         },
         "zookeeper": {
             "hosts": Or(comma_separated_host_port, [validate_host_port]),
+            Optional("use_ssl"): bool,
+            Optional("cacert"): str,
+            Optional("cert"): str,
+            Optional("key"): str,
+            Optional("key_password"): str,
+            Optional("verify"): bool,
+            Optional("set_acls"): dict
         },
         "kubernetes": {
             "labels": {},
+            Optional("bypass_api_service"): bool,
             Optional("namespace"): str,
             Optional("scope_label"): str,
             Optional("role_label"): str,
             Optional("use_endpoints"): bool,
             Optional("pod_ip"): Or(is_ipv4_address, is_ipv6_address),
             Optional("ports"): [{"name": str, "port": int}],
+            Optional("cacert"): str,
             Optional("retriable_http_codes"): Or(int, [int]),
         },
     }),
@@ -915,7 +1036,8 @@ schema = Schema({
     },
     Optional("watchdog"): {
         Optional("mode"): validate_watchdog_mode,
-        Optional("device"): str
+        Optional("device"): str,
+        Optional("safety_margin"): int
     },
     Optional("tags"): {
         Optional("nofailover"): bool,
diff --git a/patroni/version.py b/patroni/version.py
index 96c68e77..8653a3bc 100644
--- a/patroni/version.py
+++ b/patroni/version.py
@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.0.3'
+__version__ = '3.0.4'
diff --git a/postgres0.yml b/postgres0.yml
index 2e83c7e1..0605c322 100644
--- a/postgres0.yml
+++ b/postgres0.yml
@@ -93,7 +93,7 @@ bootstrap:
   # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
 # post_init: /usr/local/bin/setup_cluster.sh
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%
diff --git a/postgres1.yml b/postgres1.yml
index e8b2806d..89dcca1a 100644
--- a/postgres1.yml
+++ b/postgres1.yml
@@ -87,7 +87,7 @@ bootstrap:
   # Additional script to be launched after initial cluster creation (will be passed the connection URL as parameter)
 # post_init: /usr/local/bin/setup_cluster.sh
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%
diff --git a/postgres2.yml b/postgres2.yml
index 8272e3ba..c77e734a 100644
--- a/postgres2.yml
+++ b/postgres2.yml
@@ -84,7 +84,7 @@ bootstrap:
   - encoding: UTF8
   - data-checksums
 
-  # Some additional users users which needs to be created after initializing new cluster
+  # Some additional users which needs to be created after initializing new cluster
   users:
     admin:
       password: admin%


Test patch:
diff --git a/.github/workflows/tests.yaml b/.github/workflows/tests.yaml
index 59a53c6e..355ea927 100644
--- a/.github/workflows/tests.yaml
+++ b/.github/workflows/tests.yaml
@@ -173,4 +173,4 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.316
+        version: 1.1.317
diff --git a/tests/__init__.py b/tests/__init__.py
index 03598ae1..dcf9ed9e 100644
--- a/tests/__init__.py
+++ b/tests/__init__.py
@@ -108,7 +108,7 @@ class MockCursor(object):
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)] if self.rowcount == 1 else [None]
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
-            self.results = [(1, 2, 1, 0, False, 1, 1, None, None,
+            self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '',
                              [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
                              'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
@@ -117,7 +117,7 @@ class MockCursor(object):
             replication_info = '[{"application_name":"walreceiver","client_addr":"1.2.3.4",' +\
                                '"state":"streaming","sync_state":"async","sync_priority":0}]'
             now = datetime.datetime.now(tzutc)
-            self.results = [(now, 0, '', 0, '', False, now, replication_info)]
+            self.results = [(now, 0, '', 0, '', False, now, 'streaming', None, replication_info)]
         elif sql.startswith('SELECT name, setting'):
             self.results = [('wal_segment_size', '2048', '8kB', 'integer', 'internal'),
                             ('wal_block_size', '8192', None, 'integer', 'internal'),
diff --git a/tests/test_api.py b/tests/test_api.py
index 25342dca..f9d0b1c1 100644
--- a/tests/test_api.py
+++ b/tests/test_api.py
@@ -29,7 +29,8 @@ class MockPostgresql(object):
     name = 'test'
     state = 'running'
     role = 'primary'
-    server_version = '999999'
+    server_version = 90625
+    major_version = 90600
     sysid = 'dummysysid'
     scope = 'dummy'
     pending_restart = True
@@ -55,6 +56,10 @@ class MockPostgresql(object):
     def is_running():
         return True
 
+    @staticmethod
+    def replication_state_from_parameters(*args):
+        return 'streaming'
+
 
 class MockWatchdog(object):
     is_healthy = False
@@ -219,7 +224,7 @@ class TestRestApiHandler(unittest.TestCase):
         with patch.object(MockHa, 'restart_scheduled', Mock(return_value=True)):
             MockRestApiServer(RestApiHandler, 'GET /primary')
         self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /primary'))
-        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, '')])):
+        with patch.object(RestApiServer, 'query', Mock(return_value=[('', 1, '', '', '', '', False, None, None, '')])):
             self.assertIsNotNone(MockRestApiServer(RestApiHandler, 'GET /patroni'))
         with patch.object(GlobalConfig, 'is_standby_cluster', Mock(return_value=True)),\
                 patch.object(GlobalConfig, 'is_paused', Mock(return_value=True)):
diff --git a/tests/test_ha.py b/tests/test_ha.py
index 8495a1f0..005e922e 100644
--- a/tests/test_ha.py
+++ b/tests/test_ha.py
@@ -223,9 +223,12 @@ class TestHa(PostgresInit):
 
     @patch.object(Postgresql, 'received_timeline', Mock(return_value=None))
     def test_touch_member(self):
+        self.p._major_version = 110000
+        self.p.is_leader = false
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 0))
         self.p.replica_cached_timeline = Mock(side_effect=Exception)
-        self.ha.touch_member()
+        with patch.object(Postgresql, '_cluster_info_state_get', Mock(return_value='streaming')):
+            self.ha.touch_member()
         self.p.timeline_wal_position = Mock(return_value=(0, 1, 1))
         self.p.set_role('standby_leader')
         self.ha.touch_member()
@@ -275,18 +278,29 @@ class TestHa(PostgresInit):
         self.p.follow = true
         self.assertEqual(self.ha.run_cycle(), 'starting as a secondary')
         self.p.is_running = true
+        ha_dcs_orig_name = self.ha.dcs.__class__.__name__
         self.ha.dcs.__class__.__name__ = 'Raft'
         self.assertEqual(self.ha.run_cycle(), 'started as a secondary')
+        self.ha.dcs.__class__.__name__ = ha_dcs_orig_name
 
     def test_recover_former_primary(self):
         self.p.follow = false
         self.p.is_running = false
         self.p.name = 'leader'
-        self.p.set_role('primary')
+        self.p.set_role('demoted')
         self.p.controldata = lambda: {'Database cluster state': 'shut down', 'Database system identifier': SYSID}
         self.ha.cluster = get_cluster_initialized_with_leader()
         self.assertEqual(self.ha.run_cycle(), 'starting as readonly because i had the session lock')
 
+    def test_start_primary_after_failure(self):
+        self.p.start = false
+        self.p.is_running = false
+        self.p.name = 'leader'
+        self.p.set_role('primary')
+        self.p.controldata = lambda: {'Database cluster state': 'in production', 'Database system identifier': SYSID}
+        self.ha.cluster = get_cluster_initialized_with_leader()
+        self.assertEqual(self.ha.run_cycle(), 'starting primary after failure')
+
     @patch.object(Rewind, 'ensure_clean_shutdown', Mock())
     def test_crash_recovery(self):
         self.ha.has_lock = true
@@ -574,6 +588,7 @@ class TestHa(PostgresInit):
         self.p.is_leader = false
         self.assertEqual(self.ha.run_cycle(), 'waiting for end of recovery after bootstrap')
         self.p.is_leader = true
+        self.ha.is_synchronous_mode = true
         self.assertEqual(self.ha.run_cycle(), 'running post_bootstrap')
         self.assertEqual(self.ha.run_cycle(), 'initialized a new cluster')
 
@@ -837,8 +852,6 @@ class TestHa(PostgresInit):
         self.ha.dcs._last_failsafe = None
         with patch.object(Watchdog, 'is_healthy', PropertyMock(return_value=False)):
             self.assertFalse(self.ha.is_healthiest_node())
-        with patch('patroni.postgresql.Postgresql.is_starting', return_value=True):
-            self.assertFalse(self.ha.is_healthiest_node())
         self.ha.is_paused = true
         self.assertFalse(self.ha.is_healthiest_node())
 
diff --git a/tests/test_rewind.py b/tests/test_rewind.py
index 3d921621..a4fafcff 100644
--- a/tests/test_rewind.py
+++ b/tests/test_rewind.py
@@ -91,9 +91,10 @@ class TestRewind(BaseTestPostgresql):
                                              'Latest checkpoint location': '0/'})):
             self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
-        with patch.object(Postgresql, 'is_running', Mock(return_value=True)):
-            with patch.object(MockCursor, 'fetchone', Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None), Exception])):
-                self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
+        with patch.object(Postgresql, 'is_running', Mock(return_value=True)),\
+                patch.object(MockCursor, 'fetchone',
+                             Mock(side_effect=[(0, 0, 1, 1, 0, 0, 0, 0, 0, None, None, None), Exception])):
+            self.r.rewind_or_reinitialize_needed_and_possible(self.leader)
 
     @patch.object(CancellableSubprocess, 'call', mock_cancellable_call)
     @patch.object(Postgresql, 'checkpoint', side_effect=['', '1'],)
@@ -238,7 +239,7 @@ class TestRewind(BaseTestPostgresql):
 
         with patch('os.listdir', Mock(return_value=['000000000000000000000000.ready'])):
             # successful archive_command call
-            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)):
+            with patch.object(CancellableSubprocess, 'call', Mock(return_value=0)) as mock_subprocess_call:
                 get_guc_value_res = [
                     'on', 'command %f',
                     'always', 'command %f',
@@ -251,6 +252,10 @@ class TestRewind(BaseTestPostgresql):
                                           '000000000000000000000000', 'command 000000000000000000000000'),
                                          mock_logger_info.call_args[0])
                         mock_logger_info.reset_mock()
+                        mock_subprocess_call.assert_called_once()
+                        self.assertEqual(mock_subprocess_call.call_args.args[0], ['command 000000000000000000000000'])
+                        self.assertEqual(mock_subprocess_call.call_args.kwargs['shell'], True)
+                        mock_subprocess_call.reset_mock()
 
             # failed archive_command call
             with patch.object(CancellableSubprocess, 'call', Mock(return_value=1)):
diff --git a/tests/test_slots.py b/tests/test_slots.py
index bdb4bc5d..3f21998f 100644
--- a/tests/test_slots.py
+++ b/tests/test_slots.py
@@ -77,14 +77,14 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(Postgresql, '_query') as mock_query:
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {'ls': 12345})
 
             self.p.reset_cluster_info_state(None)
             mock_query.return_value.fetchone.return_value = (
-                1, 0, 0, 0, 0, 0, 0, 0, 0,
+                1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 6, "plugin": "b",
                   "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])
             self.assertEqual(self.p.slots(), {})
@@ -93,7 +93,7 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test__ensure_logical_slots_replica(self):
         self.p.set_role('replica')
         self.cluster.slots['ls'] = 12346
-        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock()):
+        with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
         self.s._schedule_load_slots = False
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)),\
@@ -121,12 +121,12 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.s.copy_logical_slots(self.cluster, ['ls'])
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
                 patch.object(MockCursor, 'fetchone', Mock(side_effect=Exception)):
-            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('postgresql0', None)]))),\
                 patch.object(MockCursor, 'fetchone', Mock(return_value=(False,))):
-            self.assertIsNone(self.s.check_logical_slots_readiness(self.cluster, False, None))
+            self.assertFalse(self.s.check_logical_slots_readiness(self.cluster, None))
         with patch.object(MockCursor, '__iter__', Mock(return_value=iter([('ls', 100)]))):
-            self.s.check_logical_slots_readiness(self.cluster, False, None)
+            self.s.check_logical_slots_readiness(self.cluster, None)
 
     @patch.object(Postgresql, 'stop', Mock(return_value=True))
     @patch.object(Postgresql, 'start', Mock(return_value=True))
diff --git a/tests/test_validator.py b/tests/test_validator.py
index 4c195a0e..24379c6b 100644
--- a/tests/test_validator.py
+++ b/tests/test_validator.py
@@ -15,7 +15,8 @@ config = {
     "scope": "string",
     "restapi": {
         "listen": "127.0.0.2:800",
-        "connect_address": "127.0.0.2:800"
+        "connect_address": "127.0.0.2:800",
+        "verify_client": 'none'
     },
     "bootstrap": {
         "dcs": {