Merge branch 'master' of github.com:zalando/patroni into feature/quorum-commit

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 \

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 \

features/basic_replication.feature

@@ -79,6 +79,7 @@ Feature: basic replication
     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

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')

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])
-            except ImportError:
-                logger.debug('Failed to import %s', module_name)
+                module = importlib.import_module(mod_name)
+                dcs_module = find_dcs_class_in_module(module)
+                if dcs_module:
+                    yield name, dcs_module
 
-    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))))
+            except ImportError:
+                logger.log(logging.DEBUG if config is not None else logging.INFO,
+                           'Failed to import %s', mod_name)
+
+
+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]
@@ -596,24 +643,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.
 
-        slots = {slot_name_from_member_name(name): {'type': 'physical'} for name in slot_members}
+        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: 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
@@ -621,11 +669,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)
@@ -642,7 +717,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:
@@ -653,12 +729,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:

patroni/ha.py

@@ -1267,7 +1267,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)
@@ -1371,15 +1373,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'
@@ -2031,7 +2030,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:
@@ -2091,3 +2092,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]

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)

patroni/postgresql/slots.py

@@ -130,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()
 
@@ -190,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:
@@ -327,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)
 
@@ -347,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
+                logger.error("Failed to check %s physical slot on the primary: %r", slot_name, e)
+                return False
 
-        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:
+            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:
@@ -433,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()
 
@@ -446,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))

tests/test_ha.py

@@ -280,8 +280,10 @@ 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

tests/test_rewind.py

@@ -239,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',
@@ -252,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)):

tests/test_slots.py

@@ -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))