Merge branch 'master' of github.com:zalando/patroni into feature/failover-switchover-definition

.github/workflows/tests.yaml

@@ -173,4 +173,4 @@ jobs:
 
     - uses: jakebailey/pyright-action@v1
       with:
-        version: 1.1.317
+        version: 1.1.320

docs/dynamic_configuration.rst

@@ -46,9 +46,9 @@ In order to change the dynamic configuration you can use either ``patronictl edi
    -  **archive\_cleanup\_command**: cleanup command for standby leader
    -  **recovery\_min\_apply\_delay**: how long to wait before actually apply WAL records on a standby leader
 
--  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. The physical slots are maintained only in the current primary. The logical slots are copied from the primary to a standby with restart, and after that their position advanced every **loop_wait** seconds (if necessary). Copying logical slot files performed via ``libpq`` connection and using either rewind or superuser credentials (see **postgresql.authentication** section). There is always a chance that the logical slot position on the replica is a bit behind the former primary, therefore application should be prepared that some messages could be received the second time after the failover. The easiest way of doing so - tracking ``confirmed_flush_lsn``. Enabling permanent logical replication slots requires **postgresql.use_slots** to be set and will also automatically enable the ``hot_standby_feedback``. Since the failover of logical replication slots is unsafe on PostgreSQL 9.6 and older and PostgreSQL version 10 is missing some important functions, the feature only works with PostgreSQL 11+.
 
-   -  **my\_slot\_name**: the name of replication slot. If the permanent slot name matches with the name of the current primary it will not be created. Everything else is the responsibility of the operator to make sure that there are no clashes in names between replication slots automatically created by Patroni for members and permanent replication slots.
+   -  **my\_slot\_name**: the name of the permanent replication slot. If the permanent slot name matches with the name of the current leader it will not be created. Please note that Patroni does not make checks for permanent slot names added to this configuration matching those that Patroni creates automatically for members. If those names are added, Patroni will ensure that any slots that were created are not removed even if the member becomes unresponsive, situation which would normally result in the slot's removal by Patroni. Although this can be useful in some situations, such as when importing existing members to a new Patroni cluster (see :ref:`Convert a Standalone to a Patroni Cluster <existing_data>` for details), caution should be exercised by the operator that these clashes in names are not persisted in the DCS due to its effect on normal functioning of Patroni.
 
       -  **type**: slot type. Could be ``physical`` or ``logical``. If the slot is logical, you have to additionally define ``database`` and ``plugin``.
       -  **database**: the database name where logical slots should be created.

docs/existing_data.rst

@@ -10,18 +10,58 @@ To deploy a Patroni cluster without using a pre-existing PostgreSQL instance, se
 Procedure
 ---------
 
-A Patroni cluster can be started with a data directory from a single-node PostgreSQL database. This is achieved by following closely these steps:
+You can find below an overview of steps for converting an existing Postgres cluster to a Patroni managed cluster. In the steps we assume all nodes that are part of the existing cluster are currently up and running, and that you *do not* intend to change Postgres configuration while the migration is ongoing. The steps:
 
-1. Manually start PostgreSQL daemon
-2. Create Patroni superuser and replication users as defined in the :ref:`authentication <postgresql_settings>` section of the Patroni configuration. If this user is created in SQL, the following queries achieve this:
+#. Create the Postgres users as explained for :ref:`authentication <postgresql_settings>` section of the Patroni configuration. You can find sample SQL commands to create the users in the code block below, in which you need to replace the usernames and passwords as per your environment. If you already have the relevant users, then you can skip this step.
 
-.. code-block:: sql
+   .. code-block:: sql
 
-   CREATE USER $PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD '$PATRONI_SUPERUSER_PASSWORD';
-   CREATE USER $PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD '$PATRONI_REPLICATION_PASSWORD';
+      -- Patroni superuser
+      -- Replace PATRONI_SUPERUSER_USERNAME and PATRONI_SUPERUSER_PASSWORD accordingly
+      CREATE USER PATRONI_SUPERUSER_USERNAME WITH SUPERUSER ENCRYPTED PASSWORD 'PATRONI_SUPERUSER_PASSWORD';
 
-3. Start Patroni (e.g. ``patroni /etc/patroni/patroni.yml``). It automatically detects that PostgreSQL daemon is already running but its configuration might be out-of-date.
-4. Ask Patroni to restart the node with ``patronictl restart cluster-name node-name``. This step is only required if PostgreSQL configuration is out-of-date.
+      -- Patroni replication user
+      -- Replace PATRONI_REPLICATION_USERNAME and PATRONI_REPLICATION_PASSWORD accordingly
+      CREATE USER PATRONI_REPLICATION_USERNAME WITH REPLICATION ENCRYPTED PASSWORD 'PATRONI_REPLICATION_PASSWORD';
+
+      -- Patroni rewind user, if you intend to enable use_pg_rewind in your Patroni configuration
+      -- Replace PATRONI_REWIND_USERNAME and PATRONI_REWIND_PASSWORD accordingly
+      CREATE USER PATRONI_REWIND_USERNAME WITH ENCRYPTED PASSWORD 'PATRONI_REWIND_PASSWORD';
+      GRANT EXECUTE ON function pg_catalog.pg_ls_dir(text, boolean, boolean) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_stat_file(text, boolean) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text) TO PATRONI_REWIND_USERNAME;
+      GRANT EXECUTE ON function pg_catalog.pg_read_binary_file(text, bigint, bigint, boolean) TO PATRONI_REWIND_USERNAME;
+
+#. Perform the following steps on all Postgres nodes. Perform all steps on one node before proceeding with the next node. Start with the primary node, then proceed with each standby node:
+
+   #. If you are running Postgres through systemd, then disable the Postgres systemd unit. This is performed as Patroni manages starting and stopping the Postgres daemon.
+
+   #. Create a YAML configuration file for Patroni.
+
+      * **Note (specific for the primary node):** If you have replication slots being used for replication between cluster members, then it is recommended that you enable ``use_slots`` and configure the existing replication slots as permanent via the ``slots`` configuration item. Be aware that Patroni automatically creates replication slots for replication between members, and drops replication slots that it does not recognize, when ``use_slots`` is enabled. The idea of using permanent slots here is to allow your existing slots to persist while the migration to Patroni is in progress. See :ref:`YAML Configuration Settings <yaml_configuration>` for details.
+
+   #. Start Patroni using the ``patroni`` systemd service unit. It automatically detects that Postgres is already running and starts monitoring the instance.
+
+#. Hand over Postgres "start up procedure" to Patroni. In order to do that you need to restart the cluster members through ``patronictl restart cluster-name member-name`` command. For minimal downtime you might want to split this step into:
+
+   #. Immediate restart of the standby nodes.
+   #. Scheduled restart of the primary node within a maintenance window.
+
+#. If you configured permanent slots in step ``1.2.``, then you should remove them from ``slots`` configuration through ``patronictl edit-config cluster-name member-name`` command once the ``restart_lsn`` of the slots created by Patroni is able to catch up with the ``restart_lsn`` of the original slots for the corresponding members. By removing the slots from ``slots`` configuration you will allow Patroni to drop the original slots from your cluster once they are not needed anymore. You can find below an example query to check the ``restart_lsn`` of a couple slots, so you can compare them:
+
+   .. code-block:: sql
+
+      -- Assume original_slot_for_member_x is the name of the slot in your original
+      -- cluster for replicating changes to member X, and slot_for_member_x is the
+      -- slot created by Patroni for that purpose. You need restart_lsn of
+      -- slot_for_member_x to be >= restart_lsn of original_slot_for_member_x
+      SELECT slot_name,
+             restart_lsn
+      FROM pg_replication_slots
+      WHERE slot_name IN (
+          'original_slot_for_member_x',
+          'slot_for_member_x'
+      )
 
 .. _major_upgrade:
 
@@ -30,14 +70,14 @@ Major Upgrade of PostgreSQL Version
 
 The only possible way to do a major upgrade currently is:
 
-1. Stop Patroni
-2. Upgrade PostgreSQL binaries and perform `pg_upgrade <https://www.postgresql.org/docs/current/pgupgrade.html>`_ on the primary node
-3. Update patroni.yml
-4. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove <cluster-name>``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier.
-5. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one.  It will help you to retain some PostgreSQL parameters you had set before.
-6. Start Patroni on the primary node.
-7. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes.
-8. Start Patroni on the standby nodes and wait for the replication to complete.
+#. Stop Patroni
+#. Upgrade PostgreSQL binaries and perform `pg_upgrade <https://www.postgresql.org/docs/current/pgupgrade.html>`_ on the primary node
+#. Update patroni.yml
+#. Remove the initialize key from DCS or wipe complete cluster state from DCS. The second one could be achieved by running ``patronictl remove <cluster-name>``. It is necessary because pg_upgrade runs initdb which actually creates a new database with a new PostgreSQL system identifier.
+#. If you wiped the cluster state in the previous step, you may wish to copy patroni.dynamic.json from old data dir to the new one.  It will help you to retain some PostgreSQL parameters you had set before.
+#. Start Patroni on the primary node.
+#. Upgrade PostgreSQL binaries, update patroni.yml and wipe the data_dir on standby nodes.
+#. Start Patroni on the standby nodes and wait for the replication to complete.
 
 Running pg_upgrade on standby nodes is not supported by PostgreSQL. If you know what you are doing, you can try the rsync procedure described in https://www.postgresql.org/docs/current/pgupgrade.html instead of wiping data_dir on standby nodes. The safest way is however to let Patroni replicate the data for you.
 

docs/kubernetes.rst

@@ -32,8 +32,11 @@ Configuration
 
 Patroni Kubernetes :ref:`settings <kubernetes_settings>` and :ref:`environment variables <kubernetes_environment>` are described in the general chapters of the documentation.
 
+.. _kubernetes_role_values:
+
 Customize role label
 ^^^^^^^^^^^^^^^^^^^^
+
 By default, Patroni will set corresponding labels on the pod it runs in based on node's role, such as ``role=master``.
 The key and value of label can be customized by `kubernetes.role_label`, `kubernetes.leader_label_value`, `kubernetes.follower_label_value` and `kubernetes.standby_leader_label_value`.
 

docs/releases.rst

@@ -3,6 +3,84 @@
 Release notes
 =============
 
+Version 3.1.0
+-------------
+
+**Breaking changes**
+
+- Changed semantic of ``restapi.keyfile`` and ``restapi.certfile`` (Alexander Kukushkin)
+
+  Previously Patroni was using ``restapi.keyfile`` and ``restapi.certfile`` as client certificates as a fallback if there were no respective configuration parameters in the ``ctl`` section.
+
+.. warning::
+    If you enabled client certificates validation (``restapi.verify_client`` is set to ``required``), you also **must** provide **valid client certificates** in the ``ctl.certfile``, ``ctl.keyfile``, ``ctl.keyfile_password``. If not provided, Patroni will not work correctly.
+
+
+**New features**
+
+- Make Pod role label configurable (Waynerv)
+
+  Values could be customized using ``kubernetes.leader_label_value``, ``kubernetes.follower_label_value`` and ``kubernetes.standby_leader_label_value`` parameters. This feature will be very useful when we change the ``master`` role to the ``primary``. You can read more about the feature and migration steps :ref:`here <kubernetes_role_values>`.
+
+
+**Improvements**
+
+- Various improvements of ``patroni --validate-config`` (Alexander Kukushkin)
+
+  Improved parameter validation for different DCS, ``bootstrap.dcs`` , ``ctl``, ``restapi``, and ``watchdog`` sections.
+
+- Start Postgres not in recovery if it crashed during recovery while Patroni is running (Alexander Kukushkin)
+
+  It may reduce recovery time and will help to prevent unnecessary timeline increments.
+
+- Avoid unnecessary updates of ``/status`` key (Alexander Kukushkin)
+
+  When there are no permanent logical slots Patroni was updating the ``/status`` on every heartbeat loop even when LSN on the primary didn't move forward.
+
+- Don't allow stale primary to win the leader race (Alexander Kukushkin)
+
+  If Patroni was hanging during a significant time due to lack of resources it will additionally check that no other nodes promoted Postgres before acquiring the leader lock.
+
+- Implemented visibility of certain PostgreSQL parameters validation (Alexander Kukushkin, Feike Steenbergen)
+
+  If validation of ``max_connections``, ``max_wal_senders``, ``max_prepared_transactions``, ``max_locks_per_transaction``, ``max_replication_slots``, or ``max_worker_processes`` failed Patroni was using some sane default value. Now in addition to that it will also show a warning.
+
+- Set permissions for files and directories created in ``PGDATA`` (Alexander Kukushkin)
+
+  All files created by Patroni had only owner read/write permissions. This behaviour was breaking backup tools that run under a different user and relying on group read permissions. Now Patroni honors permissions on ``PGDATA`` and correctly sets permissions on all directories and files it creates inside ``PGDATA``.
+
+
+**Bugfixes**
+
+- Run ``archive_command`` through shell (Waynerv)
+
+  Patroni might archive some WAL segments before doing crash recovery in a single-user mode or before ``pg_rewind``. If the archive_command contains some shell operators, like ``&&`` it didn't work with Patroni.
+
+- Fixed "on switchover" shutdown checks (Polina Bungina)
+
+  It was possible that specified candidate is still streaming and didn't received shut down checking but the leader key was removed because some other nodes were healthy.
+
+- Fixed "is primary" check (Alexander Kukushkin)
+
+  During the leader race replicas were not able to recognize that Postgres on the old leader is still running as a primary.
+
+- Fixed ``patronictl list`` (Alexander Kukushkin)
+
+  The Cluster name field was missing in ``tsv``, ``json``, and ``yaml`` output formats.
+
+- Fixed ``pg_rewind`` behaviour after pause (Alexander Kukushkin)
+
+  Under certain conditions, Patroni wasn't able to join the false primary back to the cluster with ``pg_rewind`` after coming out of maintenance mode.
+
+- Fixed bug in Etcd v3 implementation (Alexander Kukushkin)
+
+  Invalidate internal KV cache if key update performed using ``create_revision``/``mod_revision`` field due to revision mismatch.
+
+- Fixed behaviour of replicas in standby cluster in pause (Alexander Kukushkin)
+
+  When the leader key expires replicas in standby cluster will not follow the remote node but keep ``primary_conninfo`` as it is.
+
+
 Version 3.0.4
 -------------
 

patroni/__main__.py

@@ -74,7 +74,7 @@ class Patroni(AbstractPatroniDaemon):
         if not isinstance(member, Member):
             return
         try:
-            _ = self.request(member, endpoint="/liveness")
+            _ = self.request(member, endpoint="/liveness", timeout=3)
             logger.fatal("Can't start; there is already a node named '%s' running", self.config['name'])
             sys.exit(1)
         except Exception:

patroni/config.py

@@ -13,6 +13,7 @@ from . import PATRONI_ENV_PREFIX
 from .collections import CaseInsensitiveDict
 from .dcs import ClusterConfig, Cluster
 from .exceptions import ConfigParseError
+from .file_perm import pg_perm
 from .postgresql.config import ConfigHandler
 from .utils import deep_compare, parse_bool, parse_int, patch_config
 
@@ -275,11 +276,13 @@ class Config(object):
         if self._cache_needs_saving:
             tmpfile = fd = None
             try:
+                pg_perm.set_permissions_from_data_directory(self._data_dir)
                 (fd, tmpfile) = tempfile.mkstemp(prefix=self.__CACHE_FILENAME, dir=self._data_dir)
                 with os.fdopen(fd, 'w') as f:
                     fd = None
                     json.dump(self.dynamic_configuration, f)
                 tmpfile = shutil.move(tmpfile, self._cache_file)
+                os.chmod(self._cache_file, pg_perm.file_create_mode)
                 self._cache_needs_saving = False
             except Exception:
                 logger.exception('Exception when saving file: %s', self._cache_file)

patroni/ctl.py

@@ -36,7 +36,7 @@ from collections import defaultdict
 from contextlib import contextmanager
 from prettytable import ALL, FRAME, PrettyTable
 from urllib.parse import urlparse
-from typing import Any, Dict, Generator, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
     from psycopg2 import cursor
@@ -1824,7 +1824,7 @@ def resume(obj: Dict[str, Any], cluster_name: str, group: Optional[int], wait: b
 
 
 @contextmanager
-def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Generator[str, None, None]:
+def temporary_file(contents: bytes, suffix: str = '', prefix: str = 'tmp') -> Iterator[str]:
     """Create a temporary file with specified contents that persists for the context.
 
     :param contents: binary string that will be written to the file.

patroni/dcs/etcd3.py

@@ -228,7 +228,7 @@ class Etcd3Client(AbstractEtcdClientWithFailover):
         return self.http.urlopen
 
     def _handle_server_response(self, response: urllib3.response.HTTPResponse) -> Dict[str, Any]:
-        data: Union[bytes, str] = response.data
+        data = response.data
         try:
             data = data.decode('utf-8')
             ret: Dict[str, Any] = json.loads(data)

patroni/dcs/kubernetes.py

@@ -134,6 +134,8 @@ class K8sConfig(object):
             config: Dict[str, Any] = yaml.safe_load(f)
 
         context = context or config['current-context']
+        if TYPE_CHECKING:  # pragma: no cover
+            assert isinstance(context, str)
         context_value = self._get_by_name(config, 'context', context)
         if TYPE_CHECKING:  # pragma: no cover
             assert isinstance(context_value, dict)

patroni/file_perm.py

@@ -0,0 +1,95 @@
+"""Helper object that helps with figuring out file and directory permissions based on permissions of PGDATA.
+
+:var logger: logger of this module.
+:var pg_perm: instance of the :class:`__FilePermissions` object.
+"""
+import logging
+import os
+import stat
+
+logger = logging.getLogger(__name__)
+
+
+class __FilePermissions:
+    """Helper class for managing permissions of directories and files under PGDATA.
+
+    Execute :meth:`set_permissions_from_data_directory` to figure out which permissions should be used for files and
+    directories under PGDATA based on permissions of PGDATA root directory.
+    """
+
+    # Mode mask for data directory permissions that only allows the owner to
+    # read/write directories and files -- mask 077.
+    __PG_MODE_MASK_OWNER = stat.S_IRWXG | stat.S_IRWXO
+
+    # Mode mask for data directory permissions that also allows group read/execute -- mask 027.
+    __PG_MODE_MASK_GROUP = stat.S_IWGRP | stat.S_IRWXO
+
+    # Default mode for creating directories -- mode 700.
+    __PG_DIR_MODE_OWNER = stat.S_IRWXU
+
+    # Mode for creating directories that allows group read/execute -- mode 750.
+    __PG_DIR_MODE_GROUP = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+
+    # Default mode for creating files -- mode 600.
+    __PG_FILE_MODE_OWNER = stat.S_IRUSR | stat.S_IWUSR
+
+    # Mode for creating files that allows group read -- mode 640.
+    __PG_FILE_MODE_GROUP = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP
+
+    def __init__(self) -> None:
+        """Create a :class:`__FilePermissions` object and set default permissions."""
+        self.__set_owner_permissions()
+        self.__set_umask()
+
+    def __set_umask(self) -> None:
+        """Set umask value based on calculations.
+
+        .. note::
+            Should only be called once either :meth:`__set_owner_permissions`
+            or :meth:`__set_group_permissions` has been executed.
+        """
+        try:
+            os.umask(self.__pg_mode_mask)
+        except Exception as e:
+            logger.error('Can not set umask to %03o: %r', self.__pg_mode_mask, e)
+
+    def __set_owner_permissions(self) -> None:
+        """Make directories/files accessible only by the owner."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_OWNER
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_OWNER
+        self.__pg_mode_mask = self.__PG_MODE_MASK_OWNER
+
+    def __set_group_permissions(self) -> None:
+        """Make directories/files accessible by the owner and readable by group."""
+        self.__pg_dir_create_mode = self.__PG_DIR_MODE_GROUP
+        self.__pg_file_create_mode = self.__PG_FILE_MODE_GROUP
+        self.__pg_mode_mask = self.__PG_MODE_MASK_GROUP
+
+    def set_permissions_from_data_directory(self, data_dir: str) -> None:
+        """Set new permissions based on provided *data_dir*.
+
+        :param data_dir: reference to PGDATA to calculate permissions from.
+        """
+        try:
+            st = os.stat(data_dir)
+            if (st.st_mode & self.__PG_DIR_MODE_GROUP) == self.__PG_DIR_MODE_GROUP:
+                self.__set_group_permissions()
+            else:
+                self.__set_owner_permissions()
+        except Exception as e:
+            logger.error('Can not check permissions on %s: %r', data_dir, e)
+        else:
+            self.__set_umask()
+
+    @property
+    def dir_create_mode(self) -> int:
+        """Directory permissions."""
+        return self.__pg_dir_create_mode
+
+    @property
+    def file_create_mode(self) -> int:
+        """File permissions."""
+        return self.__pg_file_create_mode
+
+
+pg_perm = __FilePermissions()

patroni/ha.py

@@ -83,11 +83,7 @@ class Failsafe(object):
     def __init__(self, dcs: AbstractDCS) -> None:
         self._lock = RLock()
         self._dcs = dcs
-        self._last_update = 0
-        self._name = None
-        self._conn_url = None
-        self._api_url = None
-        self._slots = None
+        self._reset_state()
 
     def update(self, data: Dict[str, Any]) -> None:
         with self._lock:
@@ -97,13 +93,20 @@ class Failsafe(object):
             self._api_url = data['api_url']
             self._slots = data.get('slots')
 
+    def _reset_state(self) -> None:
+        self._last_update = 0
+        self._name = None
+        self._conn_url = None
+        self._api_url = None
+        self._slots = None
+
     @property
     def leader(self) -> Optional[Leader]:
         with self._lock:
             if self._last_update + self._dcs.ttl > time.time() and self._name:
-                return Leader('', '', RemoteMember.from_name_and_data(self._name, {'api_url': self._api_url,
-                                                                                   'conn_url': self._conn_url,
-                                                                                   'slots': self._slots}))
+                return Leader('', '', RemoteMember(self._name, {'api_url': self._api_url,
+                                                                'conn_url': self._conn_url,
+                                                                'slots': self._slots}))
 
     def update_cluster(self, cluster: Cluster) -> Cluster:
         # Enreach cluster with the real leader if there was a ping from it
@@ -130,6 +133,8 @@ class Failsafe(object):
     def set_is_active(self, value: float) -> None:
         with self._lock:
             self._last_update = value
+            if not value:
+                self._reset_state()
 
 
 class Ha(object):
@@ -195,6 +200,13 @@ class Ha(object):
         with self._is_leader_lock:
             self._is_leader = time.time() + self.dcs.ttl if value else 0
 
+    def sync_mode_is_active(self) -> bool:
+        """Check whether synchronous replication is requested and already active.
+
+        :returns: ``True`` if the primary already put its name into the ``/sync`` in DCS.
+        """
+        return self.is_synchronous_mode() and not self.cluster.sync.is_empty
+
     def load_cluster_from_dcs(self) -> None:
         cluster = self.dcs.get_cluster()
 
@@ -460,7 +472,7 @@ class Ha(object):
             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.get_failover_candidates()):
+                if self.is_failover_possible():
                     self.watchdog.disable()
                     logger.info("Primary crashed. Failing over.")
                     self.demote('immediate')
@@ -527,10 +539,17 @@ class Ha(object):
         return msg
 
     def _get_node_to_follow(self, cluster: Cluster) -> Union[Leader, Member, None]:
-        # determine the node to follow. If replicatefrom tag is set,
-        # try to follow the node mentioned there, otherwise, follow the leader.
-        if self.is_standby_cluster() and (self.cluster.is_unlocked() or self.has_lock(False)):
+        """Determine the node to follow.
+
+        :param cluster: the currently known cluster state from DCS.
+
+        :returns: the node which we should be replicating from.
+        """
+        # The standby leader or when there is no standby leader we want to follow
+        # the remote member, except when there is no standby leader in pause.
+        if self.is_standby_cluster() and (self.has_lock(False) or self.cluster.is_unlocked() and not self.is_paused()):
             node_to_follow = self.get_remote_member()
+        # If replicatefrom tag is set, try to follow the node mentioned there, otherwise, follow the leader.
         elif self.patroni.replicatefrom and self.patroni.replicatefrom != self.state_handler.name:
             node_to_follow = cluster.get_member(self.patroni.replicatefrom)
         else:
@@ -770,6 +789,9 @@ class Ha(object):
                 self.state_handler.sync_handler.set_synchronous_standby_names(
                     CaseInsensitiveSet('*') if self.global_config.is_synchronous_mode_strict else CaseInsensitiveSet())
             if self.state_handler.role not in ('master', 'promoted', 'primary'):
+                # reset failsafe state when promote
+                self._failsafe.set_is_active(0)
+
                 def before_promote():
                     self.notify_citus_coordinator('before_promote')
 
@@ -795,6 +817,8 @@ class Ha(object):
         return _MemberStatus.unknown(member)
 
     def fetch_nodes_statuses(self, members: List[Member]) -> List[_MemberStatus]:
+        if not members:
+            return []
         pool = ThreadPool(len(members))
         results = pool.map(self.fetch_node_status, members)  # Run API calls on members in parallel
         pool.close()
@@ -832,7 +856,7 @@ class Ha(object):
             data['slots'] = self.state_handler.slots()
         except Exception:
             logger.exception('Exception when called state_handler.slots()')
-        members = [RemoteMember.from_name_and_data(name, {'api_url': url})
+        members = [RemoteMember(name, {'api_url': url})
                    for name, url in failsafe.items() if name != self.state_handler.name]
         if not members:  # A sinlge node cluster
             return True
@@ -873,48 +897,52 @@ class Ha(object):
         # Prepare list of nodes to run check against
         members = [m for m in members if m.name != self.state_handler.name and not m.nofailover and m.api_url]
 
-        if members:
-            for st in self.fetch_nodes_statuses(members):
-                if st.failover_limitation() is None:
-                    if st.in_recovery is False:
-                        logger.warning('Primary (%s) is still alive', st.member.name)
+        for st in self.fetch_nodes_statuses(members):
+            if st.failover_limitation() is None:
+                if st.in_recovery is False:
+                    logger.warning('Primary (%s) is still alive', st.member.name)
+                    return False
+                if my_wal_position < st.wal_position:
+                    logger.info('Wal position of %s is ahead of my wal position', st.member.name)
+                    # In synchronous mode the former leader might be still accessible and even be ahead of us.
+                    # We should not disqualify himself from the leader race in such a situation.
+                    if not self.sync_mode_is_active() or not self.cluster.sync.leader_matches(st.member.name):
                         return False
-                    if my_wal_position < st.wal_position:
-                        logger.info('Wal position of %s is ahead of my wal position', st.member.name)
-                        # In synchronous mode the former leader might be still accessible and even be ahead of us.
-                        # We should not disqualify himself from the leader race in such a situation.
-                        if not self.is_synchronous_mode() or self.cluster.sync.is_empty\
-                                or not self.cluster.sync.leader_matches(st.member.name):
-                            return False
-                        logger.info('Ignoring the former leader being ahead of us')
+                    logger.info('Ignoring the former leader being ahead of us')
         return True
 
-    def is_failover_possible(self, members: List[Member], cluster_lsn: Optional[int] = 0) -> bool:
-        """Checks whether one of the members from the list is healthy enough and is allowed to promote.
+    def is_failover_possible(self, *, cluster_lsn: int = 0, exclude_failover_candidate: bool = False) -> bool:
+        """Checks whether any of the cluster members is allowed to promote and is healthy enough for that.
 
-        :param members: list of members to check
-        :param cluster_lsn: to calculate replication lag and exclude member if it is laggin
-        :returns: `True` if there are members eligible to be the new leader
+        :param cluster_lsn: to calculate replication lag and exclude member if it is lagging.
+        :param exclude_failover_candidate: if ``True``, exclude :attr:`failover.candidate` from the members
+                                           list against which the failover possibility checks are run.
+        :returns: `True` if there are members eligible to become the new leader.
         """
+        candidates = self.get_failover_candidates(exclude_failover_candidate)
+
+        action = 'switchover' if self.cluster.failover and self.cluster.failover.leader else 'failover'
+        if self.is_synchronous_mode() and self.cluster.failover and self.cluster.failover.candidate and not candidates:
+            logger.warning('%s candidate=%s does not match with sync_standbys=%s',
+                           action.title(), self.cluster.failover.candidate, self.cluster.sync.sync_standby)
+        elif not candidates:
+            logger.warning('%s%s: candidates list is empty', '' if not self.cluster.failover else 'manual ', action)
+            logger.warning('manual failover: candidates list is empty')
+
         ret = False
         cluster_timeline = self.cluster.timeline
-        members = [m for m in members if not m.nofailover and m.api_url]
-        if members:
-            for st in self.fetch_nodes_statuses(members):
-                not_allowed_reason = st.failover_limitation()
-                if not_allowed_reason:
-                    logger.info('Member %s is %s', st.member.name, not_allowed_reason)
-                elif cluster_lsn and st.wal_position < cluster_lsn or\
-                        not cluster_lsn and self.is_lagging(st.wal_position):
-                    logger.info('Member %s exceeds maximum replication lag', st.member.name)
-                elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
-                    logger.info('Timeline %s of member %s is behind the cluster timeline %s',
-                                st.timeline, st.member.name, cluster_timeline)
-                else:
-                    ret = True
-        else:
-            action = 'switchover' if self.cluster.failover and self.cluster.failover.leader else 'failover'
-            logger.warning('%s%s: members list is empty', '' if not self.cluster.failover else 'manual ', action)
+        for st in self.fetch_nodes_statuses(candidates):
+            not_allowed_reason = st.failover_limitation()
+            if not_allowed_reason:
+                logger.info('Member %s is %s', st.member.name, not_allowed_reason)
+            elif cluster_lsn and st.wal_position < cluster_lsn or \
+                    not cluster_lsn and self.is_lagging(st.wal_position):
+                logger.info('Member %s exceeds maximum replication lag', st.member.name)
+            elif self.check_timeline() and (not st.timeline or st.timeline < cluster_timeline):
+                logger.info('Timeline %s of member %s is behind the cluster timeline %s',
+                            st.timeline, st.member.name, cluster_timeline)
+            else:
+                ret = True
         return ret
 
     def manual_failover_process_no_leader(self) -> Optional[bool]:
@@ -966,9 +994,8 @@ class Ha(object):
         # try to pick some other members to switchover and check that they are healthy
         if failover.leader:
             if self.state_handler.name == failover.leader:  # I was the leader
-                # exclude me and desired member which is unhealthy (failover.candidate can be None)
-                members = [m for m in self.cluster.members if m.name not in (failover.candidate, failover.leader)]
-                if self.is_failover_possible(members):  # check that there are healthy members
+                # exclude desired member which is unhealthy if it was specified
+                if self.is_failover_possible(exclude_failover_candidate=bool(failover.candidate)):
                     return False
                 else:  # I was the leader and it looks like currently I am the only healthy member
                     return True
@@ -1021,8 +1048,8 @@ class Ha(object):
 
         if self.cluster.failover:
             # When doing a switchover in synchronous mode only synchronous nodes and former leader are allowed to race
-            if self.cluster.failover.leader and self.is_synchronous_mode() and\
-                    not self.cluster.sync.is_empty and not self.cluster.sync.matches(self.state_handler.name, True):
+            if self.cluster.failover.leader and self.sync_mode_is_active() \
+                    and not self.cluster.sync.matches(self.state_handler.name, True):
                 return False
             return self.manual_failover_process_no_leader() or False
 
@@ -1039,12 +1066,11 @@ class Ha(object):
                 if failsafe_members and self.state_handler.name not in failsafe_members:
                     return False
                 # Race among not only existing cluster members, but also all known members from the failsafe config
-                all_known_members += [RemoteMember.from_name_and_data(name, {'api_url': url})
-                                      for name, url in failsafe_members.items()]
+                all_known_members += [RemoteMember(name, {'api_url': url}) for name, url in failsafe_members.items()]
         all_known_members += self.cluster.members
 
         # When in sync mode, only last known primary and sync standby are allowed to promote automatically.
-        if self.is_synchronous_mode() and not self.cluster.sync.is_empty:
+        if self.sync_mode_is_active():
             if not self.cluster.sync.matches(self.state_handler.name, True):
                 return False
             # pick between synchronous candidates so we minimize unnecessary failovers/demotions
@@ -1095,8 +1121,7 @@ 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.get_failover_candidates(check_sync=False),
-                                         cluster_lsn=checkpoint_location):
+            if self.is_failover_possible(cluster_lsn=checkpoint_location):
                 self.state_handler.set_role('demoted')
                 with self._async_executor:
                     self.release_leader_key_voluntarily(checkpoint_location)
@@ -1204,16 +1229,11 @@ class Ha(object):
             if not failover.candidate or failover.candidate != self.state_handler.name:
                 if not failover.candidate and self.is_paused():
                     logger.warning('%s is possible only to a specific candidate in a paused state', action.title())
+                elif self.is_failover_possible():
+                    ret = self._async_executor.try_run_async(f'manual {action}: demote', self.demote, ('graceful',))
+                    return ret or f'manual {action}: demoting myself'
                 else:
-                    members = self.get_failover_candidates(check_sync=self.is_synchronous_mode())
-                    if failover.candidate and not members:
-                        logger.warning('%s candidate=%s does not match with sync_standbys=%s',
-                                       action.title(), failover.candidate, self.cluster.sync.sync_standby)
-                    if self.is_failover_possible(members):  # check that there are healthy members
-                        ret = self._async_executor.try_run_async(f'manual {action}: demote', self.demote, ('graceful',))
-                        return ret or f'manual {action}: demoting myself'
-                    else:
-                        logger.warning('manual %s: no healthy members found, %s is not possible', action, action)
+                    logger.warning('manual %s: no healthy members found, %s is not possible', action, action)
             else:
                 logger.warning('manual %s: I am already the leader, no need to %s', action, action)
         else:
@@ -1475,7 +1495,7 @@ class Ha(object):
         if self.has_lock() and self.update_lock():
             if self._async_executor.scheduled_action == 'doing crash recovery in a single user mode':
                 time_left = self.global_config.primary_start_timeout - (time.time() - self._crash_recovery_started)
-                if time_left <= 0 and self.is_failover_possible(self.get_failover_candidates()):
+                if time_left <= 0 and self.is_failover_possible():
                     logger.info("Demoting self because crash recovery is taking too long")
                     self.state_handler.cancellable.cancel(True)
                     self.demote('immediate')
@@ -1587,7 +1607,7 @@ class Ha(object):
             time_left = timeout - self.state_handler.time_in_state()
 
             if time_left <= 0:
-                if self.is_failover_possible(self.get_failover_candidates()):
+                if self.is_failover_possible():
                     logger.info("Demoting self because primary startup is taking too long")
                     self.demote('immediate')
                     return 'stopped PostgreSQL because of startup timeout'
@@ -1865,9 +1885,7 @@ class Ha(object):
                     # 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.
 
-                    # for a manual failover/switchover with a candidate, we should check the requested candidate only
-                    if self.is_failover_possible(self.get_failover_candidates(check_sync=False),
-                                                 cluster_lsn=checkpoint_location):
+                    if self.is_failover_possible(cluster_lsn=checkpoint_location):
                         self.dcs.delete_leader(checkpoint_location)
                         status['deleted'] = True
                     else:
@@ -1926,31 +1944,33 @@ class Ha(object):
                 data['conn_kwargs'] = conn_kwargs
 
         name = member.name if member else 'remote_member:{}'.format(uuid.uuid1())
-        return RemoteMember.from_name_and_data(name, data)
+        return RemoteMember(name, data)
 
-    def get_failover_candidates(self, check_sync: bool = True) -> List[Member]:
-        """Return list of candidates (except me) for either manual or automatic failover.
+    def get_failover_candidates(self, exclude_failover_candidate: bool) -> List[Member]:
+        """Return a list of candidates for either manual or automatic failover.
 
-        The result is later passed to ``Ha.is_failover_possible()`` to check if any member
-        is actually healthy enough and is allowed to poromote.
+        Exclude non-sync members when in synchronous mode, the current node (its checks are always performed earlier)
+        and the candidate if required. If failover candidate exclusion is not requested and a candidate is specified
+        in the /failover key, return the candidate only.
+        The result is further evaluated in the caller :func:`Ha.is_failover_possible` to check if any member is actually
+        healthy enough and is allowed to poromote.
 
-        :param check_sync: if ``True``, also check against the sync key members
+        :param exclude_failover_candidate: if ``True``, exclude :attr:`failover.candidate` from the candidates.
 
-        :returns: a list of ``Member`` ojects or an empty list if there is no candidate available.
-                  Never includes the current node, as its checks are always performed earlier.
+        :returns: a list of :class:`Member` ojects or an empty list if there is no candidate available.
         """
         failover = self.cluster.failover
-        if check_sync and self.is_synchronous_mode() and not self.cluster.sync.is_empty:
-            if failover and not failover.leader:
-                # manual *failover*, only check the candidate (even if not in sync members)
-                return [m for m in self.cluster.members if m.name == failover.candidate
-                        and m.name != self.state_handler.name]
-            else:
-                # the candidate if is in /sync members for a candidate failover, every /sync member otherwise
-                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)
-                        and m.name != self.state_handler.name]
-        # the candidate for a candidate failover, every cluster member otherwise
-        return [m for m in self.cluster.members
-                if (not failover or not failover.candidate or m.name == failover.candidate)
-                and m.name != self.state_handler.name]
+        exclude = [self.state_handler.name] + ([failover.candidate] if failover and exclude_failover_candidate else [])
+
+        def is_eligible(node: Member) -> bool:
+            # in synchronous mode we allow failover (not switchover!) to async node
+            if self.sync_mode_is_active() and not self.cluster.sync.matches(node.name)\
+                    and not (failover and not failover.leader):
+                return False
+            # Don't spend time on "nofailover" nodes checking.
+            # We also don't need nodes which we can't query with the api in the list.
+            return node.name not in exclude and \
+                not node.nofailover and bool(node.api_url) and \
+                (not failover or not failover.candidate or node.name == failover.candidate)
+
+        return list(filter(is_eligible, self.cluster.members))

patroni/postgresql/__init__.py

@@ -12,7 +12,7 @@ from datetime import datetime
 from dateutil import tz
 from psutil import TimeoutExpired
 from threading import current_thread, Lock
-from typing import Any, Callable, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Callable, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING
 
 from .bootstrap import Bootstrap
 from .callback_executor import CallbackAction, CallbackExecutor
@@ -999,7 +999,7 @@ class Postgresql(object):
 
     @contextmanager
     def get_replication_connection_cursor(self, host: Optional[str] = None, port: int = 5432,
-                                          **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+                                          **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
         conn_kwargs = self.config.replication.copy()
         conn_kwargs.update(host=host, port=int(port) if port else None, user=conn_kwargs.pop('username'),
                            connect_timeout=3, replication=1, options='-c statement_timeout=2000')

patroni/postgresql/config.py

@@ -6,14 +6,16 @@ import socket
 import stat
 import time
 
+from contextlib import contextmanager
 from urllib.parse import urlparse, parse_qsl, unquote
 from types import TracebackType
-from typing import Any, Collection, Dict, List, Optional, Union, Tuple, Type, TYPE_CHECKING
+from typing import Any, Collection, Dict, Iterator, List, Optional, Union, Tuple, Type, TYPE_CHECKING
 
 from .validator import recovery_parameters, transform_postgresql_parameter_value, transform_recovery_parameter_value
 from ..collections import CaseInsensitiveDict, CaseInsensitiveSet
 from ..dcs import Leader, Member, RemoteMember, slot_name_from_member_name
 from ..exceptions import PatroniFatalException
+from ..file_perm import pg_perm
 from ..utils import compare_values, parse_bool, parse_int, split_host_port, uri, validate_directory, is_subpath
 from ..validator import IntValidator, EnumValidator
 
@@ -367,6 +369,30 @@ class ConfigHandler(object):
             configuration.append('pg_ident.conf')
         return configuration
 
+    def set_file_permissions(self, filename: str) -> None:
+        """Set permissions of file *filename* according to the expected permissions if it resides under PGDATA.
+
+        .. note::
+            Do nothing if the file is not under PGDATA.
+
+        :param filename: path to a file which permissions might need to be adjusted.
+        """
+        if is_subpath(self._postgresql.data_dir, filename):
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
+            os.chmod(filename, pg_perm.file_create_mode)
+
+    @contextmanager
+    def config_writer(self, filename: str) -> Iterator[ConfigWriter]:
+        """Create :class:`ConfigWriter` object and set permissions on a *filename*.
+
+        :param filename: path to a config file.
+
+        :yields: :class:`ConfigWriter` object.
+        """
+        with ConfigWriter(filename) as writer:
+            yield writer
+        self.set_file_permissions(filename)
+
     def save_configuration_files(self, check_custom_bootstrap: bool = False) -> bool:
         """
             copy postgresql.conf to postgresql.conf.backup to be able to retrieve configuration files
@@ -380,6 +406,7 @@ class ConfigHandler(object):
                     backup_file = os.path.join(self._postgresql.data_dir, f + '.backup')
                     if os.path.isfile(config_file):
                         shutil.copy(config_file, backup_file)
+                        self.set_file_permissions(backup_file)
             except IOError:
                 logger.exception('unable to create backup copies of configuration files')
         return True
@@ -393,9 +420,11 @@ class ConfigHandler(object):
                 if not os.path.isfile(config_file):
                     if os.path.isfile(backup_file):
                         shutil.copy(backup_file, config_file)
+                        self.set_file_permissions(config_file)
                     # Previously we didn't backup pg_ident.conf, if file is missing just create empty
                     elif f == 'pg_ident.conf':
                         open(config_file, 'w').close()
+                        self.set_file_permissions(config_file)
         except IOError:
             logger.exception('unable to restore configuration files from backup')
 
@@ -409,7 +438,7 @@ class ConfigHandler(object):
         if self._postgresql.enforce_hot_standby_feedback:
             configuration['hot_standby_feedback'] = 'on'
 
-        with ConfigWriter(self._postgresql_conf) as f:
+        with self.config_writer(self._postgresql_conf) as f:
             include = self._config.get('custom_conf') or self._postgresql_base_conf_name
             f.writeline("include '{0}'\n".format(ConfigWriter.escape(include)))
             for name, value in sorted((configuration).items()):
@@ -439,6 +468,7 @@ class ConfigHandler(object):
         if not self.hba_file and not self._config.get('pg_hba'):
             with open(self._pg_hba_conf, 'a') as f:
                 f.write('\n{}\n'.format('\n'.join(config)))
+            self.set_file_permissions(self._pg_hba_conf)
         return True
 
     def replace_pg_hba(self) -> Optional[bool]:
@@ -458,14 +488,14 @@ class ConfigHandler(object):
                                   self.local_replication_address['host'], self.local_replication_address['port'],
                                   0, socket.SOCK_STREAM, socket.IPPROTO_TCP)})
 
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 for address, t in addresses.items():
                     f.writeline((
                         '{0}\treplication\t{1}\t{3}\ttrust\n'
                         '{0}\tall\t{2}\t{3}\ttrust'
                     ).format(t, self.replication['username'], self._superuser.get('username') or 'all', address))
         elif not self.hba_file and self._config.get('pg_hba'):
-            with ConfigWriter(self._pg_hba_conf) as f:
+            with self.config_writer(self._pg_hba_conf) as f:
                 f.writelines(self._config['pg_hba'])
             return True
 
@@ -478,7 +508,7 @@ class ConfigHandler(object):
         """
 
         if not self.ident_file and self._config.get('pg_ident'):
-            with ConfigWriter(self._pg_ident_conf) as f:
+            with self.config_writer(self._pg_ident_conf) as f:
                 f.writelines(self._config['pg_ident'])
             return True
 
@@ -800,9 +830,11 @@ class ConfigHandler(object):
         if self._postgresql.major_version >= 120000:
             if parse_bool(recovery_params.pop('standby_mode', None)):
                 open(self._standby_signal, 'w').close()
+                self.set_file_permissions(self._standby_signal)
             else:
                 self._remove_file_if_exists(self._standby_signal)
                 open(self._recovery_signal, 'w').close()
+                self.set_file_permissions(self._recovery_signal)
 
             def restart_required(name: str) -> bool:
                 if self._postgresql.major_version >= 140000:
@@ -813,8 +845,7 @@ class ConfigHandler(object):
             self._current_recovery_params = CaseInsensitiveDict({n: [v, restart_required(n), self._postgresql_conf]
                                                                  for n, v in recovery_params.items()})
         else:
-            with ConfigWriter(self._recovery_conf) as f:
-                os.chmod(self._recovery_conf, stat.S_IWRITE | stat.S_IREAD)
+            with self.config_writer(self._recovery_conf) as f:
                 self._write_recovery_params(f, recovery_params)
 
     def remove_recovery_conf(self) -> None:
@@ -843,6 +874,7 @@ class ConfigHandler(object):
         if overwrite:
             try:
                 with open(self._auto_conf, 'w') as f:
+                    self.set_file_permissions(self._auto_conf)
                     for raw_line in lines:
                         f.write(raw_line)
             except Exception:

patroni/postgresql/connection.py

@@ -2,7 +2,7 @@ import logging
 
 from contextlib import contextmanager
 from threading import Lock
-from typing import Any, Dict, Generator, Union, TYPE_CHECKING
+from typing import Any, Dict, Iterator, Union, TYPE_CHECKING
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Connection as Connection3, Cursor
     from psycopg2 import connection, cursor
@@ -44,7 +44,7 @@ class Connection(object):
 
 
 @contextmanager
-def get_connection_cursor(**kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+def get_connection_cursor(**kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
     conn = psycopg.connect(**kwargs)
     with conn.cursor() as cur:
         yield cur

patroni/postgresql/slots.py

@@ -1,15 +1,20 @@
+"""Replication slot handling.
+
+Provides classes for the creation, monitoring, management and synchronisation of PostgreSQL replication slots.
+"""
+
 import logging
 import os
 import shutil
-
 from collections import defaultdict
 from contextlib import contextmanager
 from threading import Condition, Thread
-from typing import Any, Dict, Generator, List, Optional, Union, Tuple, TYPE_CHECKING
+from typing import Any, Dict, Iterator, List, Optional, Union, Tuple, TYPE_CHECKING, Collection
 
 from .connection import get_connection_cursor
 from .misc import format_lsn, fsync_dir
 from ..dcs import Cluster, Leader
+from ..file_perm import pg_perm
 from ..psycopg import OperationalError
 
 if TYPE_CHECKING:  # pragma: no cover
@@ -42,9 +47,17 @@ def compare_slots(s1: Dict[str, Any], s2: Dict[str, Any], dbid: str = 'database'
 
 
 class SlotsAdvanceThread(Thread):
+    """Daemon process :class:``Thread`` object for advancing logical replication slots on replicas.
+
+    This ensures that slot advancing queries sent to postgres do not block the main loop.
+    """
 
     def __init__(self, slots_handler: 'SlotsHandler') -> None:
-        super(SlotsAdvanceThread, self).__init__()
+        """Create and start a new thread for handling slot advance queries.
+
+        :param slots_handler: The calling class instance for reference to slot information attributes.
+        """
+        super().__init__()
         self.daemon = True
         self._slots_handler = slots_handler
 
@@ -58,6 +71,13 @@ class SlotsAdvanceThread(Thread):
         self.start()
 
     def sync_slot(self, cur: Union['cursor', 'Cursor[Any]'], database: str, slot: str, lsn: int) -> None:
+        """Execute a ``pg_replication_slot_advance`` query and store success for scheduled synchronisation task.
+
+        :param cur: database connection cursor.
+        :param database: name of the database associated with the slot.
+        :param slot: name of the slot to be synchronised.
+        :param lsn: last known LSN position
+        """
         failed = copy = False
         try:
             cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", (slot, format_lsn(lsn)))
@@ -79,6 +99,11 @@ class SlotsAdvanceThread(Thread):
                     self._scheduled.pop(database)
 
     def sync_slots_in_database(self, database: str, slots: List[str]) -> None:
+        """Synchronise slots for a single database.
+
+        :param database: name of the database.
+        :param slots: list of slot names to synchronise.
+        """
         with self._slots_handler.get_local_connection_cursor(dbname=database, options='-c statement_timeout=0') as cur:
             for slot in slots:
                 with self._condition:
@@ -87,6 +112,7 @@ class SlotsAdvanceThread(Thread):
                     self.sync_slot(cur, database, slot, lsn)
 
     def sync_slots(self) -> None:
+        """Synchronise slots for all scheduled databases."""
         with self._condition:
             databases = list(self._scheduled.keys())
         for database in databases:
@@ -99,6 +125,12 @@ class SlotsAdvanceThread(Thread):
                     logger.error('Failed to advance replication slots in database %s: %r', database, e)
 
     def run(self) -> None:
+        """Thread main loop entrypoint.
+
+        .. note::
+            Thread will wait until a sync is scheduled from outside, normally triggered during the HA loop or a wakeup
+            call.
+        """
         while True:
             with self._condition:
                 if not self._scheduled:
@@ -107,6 +139,14 @@ class SlotsAdvanceThread(Thread):
             self.sync_slots()
 
     def schedule(self, advance_slots: Dict[str, Dict[str, int]]) -> Tuple[bool, List[str]]:
+        """Trigger a synchronisation of slots.
+
+        This is the main entrypoint for Patroni HA loop wakeup call.
+
+        :param advance_slots: dictionary containing slots that need to be advanced
+
+        :return: tuple of failure status and a list of slots to be copied
+        """
         with self._condition:
             for database, values in advance_slots.items():
                 self._scheduled[database].update(values)
@@ -118,15 +158,27 @@ class SlotsAdvanceThread(Thread):
         return ret
 
     def on_promote(self) -> None:
+        """Reset state of the daemon."""
         with self._condition:
             self._scheduled.clear()
             self._failed = False
             self._copy_slots = []
 
 
-class SlotsHandler(object):
+class SlotsHandler:
+    """Handler for managing and storing information on replication slots in PostgreSQL.
+
+    :ivar pg_replslot_dir: system location path of the PostgreSQL replication slots.
+    :ivar _logical_slots_processing_queue: yet to be processed logical replication slots on the primary
+    """
 
     def __init__(self, postgresql: 'Postgresql') -> None:
+        """Create an instance with storage attributes for replication slots and schedule the first synchronisation.
+
+        :param postgresql: Calling class instance providing interface to PostgreSQL.
+        """
+        self._force_readiness_check = False
+        self._schedule_load_slots = False
         self._postgresql = postgresql
         self._advance = None
         self._replication_slots: Dict[str, Dict[str, Any]] = {}  # already existing replication slots
@@ -135,23 +187,46 @@ class SlotsHandler(object):
         self.schedule()
 
     def _query(self, sql: str, *params: Any) -> Union['cursor', 'Cursor[Any]']:
+        """Helper method for :meth:`Postgresql.query`.
+
+        :param sql: SQL statement to execute.
+        :param params: parameters to pass through to :meth:`Postgresql.query`.
+
+        :returns: query response.
+        """
         return self._postgresql.query(sql, *params, retry=False)
 
     @staticmethod
-    def _copy_items(src: Dict[str, Any], dst: Dict[str, Any], keys: Optional[List[str]] = None) -> None:
+    def _copy_items(src: Dict[str, Any], dst: Dict[str, Any], keys: Optional[Collection[str]] = None) -> None:
+        """Select values from *src* dictionary to update in *dst* dictionary for optional supplied *keys*.
+
+        :param src: source dictionary that *keys* will be looked up from.
+        :param dst: destination dictionary to be updated.
+        :param keys: optional list of keys to be looked up in the source dictionary.
+        """
         dst.update({key: src[key] for key in keys or ('datoid', 'catalog_xmin', 'confirmed_flush_lsn')})
 
     def process_permanent_slots(self, slots: List[Dict[str, Any]]) -> Dict[str, int]:
-        """This methods solves three problems at once (I know, it is weird).
+        """Process replication slot information from the host and prepare information used in subsequent cluster tasks.
+
+        .. note::
+            This methods solves three problems.
+
+            The ``cluster_info_query`` from :class:``Postgresql`` is executed every HA loop and returns information
+            about all replication slots that exists on the current host.
+
+            Based on this information perform the following actions:
+
+            1. For the primary we want to expose to DCS permanent logical slots, therefore build (and return) a dict
+               that maps permanent logical slot names to ``confirmed_flush_lsn``.
+            2. detect if one of the previously known permanent slots is missing and schedule resync.
+            3. Update the local cache with the fresh ``catalog_xmin`` and ``confirmed_flush_lsn`` for every known slot.
 
-        The cluster_info_query from `Postgresql` is executed every HA loop and returns
-        information about all replication slots that exists on the current host.
-        Based on this information we perform the following actions:
-        1. For the primary we want to expose to DCS permanent logical slots, therefore the method
-           builds (and returns) a dict, that maps permanent logical slot names and confirmed_flush_lsns.
-        2. This method also detects if one of the previously known permanent slots got missing and schedules resync.
-        3. Updates the local cache with the fresh catalog_xmin and confirmed_flush_lsn for every known slot.
            This info is used when performing the check of logical slot readiness on standbys.
+
+        :param slots: replication slot information that exists on the current host.
+
+        :return: dictionary of logical slot names to ``confirmed_flush_lsn``.
         """
         ret: Dict[str, int] = {}
 
@@ -173,13 +248,23 @@ class SlotsHandler(object):
         return ret
 
     def load_replication_slots(self) -> None:
+        """Query replication slot information from the database and store it for processing by other tasks.
+
+        .. note::
+            Only supported from PostgreSQL version 9.4 onwards.
+
+        Store replication slot ``name``, ``type``, ``plugin``, ``database`` and ``datoid``.
+        If PostgreSQL version is 10 or newer also store ``catalog_xmin`` and ``confirmed_flush_lsn``.
+
+        When using logical slots, store information separately for slot synchronisation  on replica nodes.
+        """
         if self._postgresql.major_version >= 90400 and self._schedule_load_slots:
             replication_slots: Dict[str, Dict[str, Any]] = {}
-            extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"\
+            extra = ", catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint" \
                 if self._postgresql.major_version >= 100000 else ""
             skip_temp_slots = ' WHERE NOT temporary' if self._postgresql.major_version >= 100000 else ''
-            cursor = self._query('SELECT slot_name, slot_type, plugin, database, datoid'
-                                 '{0} FROM pg_catalog.pg_replication_slots{1}'.format(extra, skip_temp_slots))
+            cursor = self._query(f'SELECT slot_name, slot_type, plugin, database, datoid'
+                                 f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}')
             for r in cursor:
                 value = {'type': r[1]}
                 if r[1] == 'logical':
@@ -195,16 +280,34 @@ class SlotsHandler(object):
                 self._force_readiness_check = False
 
     def ignore_replication_slot(self, cluster: Cluster, name: str) -> bool:
+        """Check if slot *name* should not be managed by Patroni.
+
+        :param cluster: cluster state information object.
+        :param name: name of the slot to ignore
+
+        :returns: ``True`` if slot *name* matches any slot specified in ``ignore_slots`` configuration,
+                 otherwise will pass through and return result of :meth:`CitusHandler.ignore_replication_slot`.
+        """
         slot = self._replication_slots[name]
         if cluster.config:
             for matcher in cluster.config.ignore_slots_matchers:
-                if ((matcher.get("name") is None or matcher["name"] == name)
-                   and all(not matcher.get(a) or matcher[a] == slot.get(a) for a in ('database', 'plugin', 'type'))):
+                if (
+                        (matcher.get("name") is None or matcher["name"] == name)
+                        and all(not matcher.get(a) or matcher[a] == slot.get(a)
+                                for a in ('database', 'plugin', 'type'))
+                ):
                     return True
         return self._postgresql.citus_handler.ignore_replication_slot(slot)
 
     def drop_replication_slot(self, name: str) -> Tuple[bool, bool]:
-        """Returns a tuple(active, dropped)"""
+        """Drop a named slot from Postgres.
+
+        :param name: name of the slot to be dropped.
+
+        :returns: a tuple of ``active`` and ``dropped``. ``active`` is ``True`` if the slot is active,
+                  ``dropped`` is ``True`` if the slot was successfully dropped. If the slot was not found return
+                  ``False`` for both.
+        """
         cursor = self._query(('WITH slots AS (SELECT slot_name, active'
                               ' FROM pg_catalog.pg_replication_slots WHERE slot_name = %s),'
                               ' dropped AS (SELECT pg_catalog.pg_drop_replication_slot(slot_name),'
@@ -217,7 +320,19 @@ class SlotsHandler(object):
         return row
 
     def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any], paused: bool) -> None:
-        # drop old replication slots which are not presented in desired slots
+        """Compare required slots and configured as permanent slots with those found, dropping extraneous ones.
+
+        .. note::
+            Slots that are not contained in *slots* will be dropped.
+            Slots can be filtered out with ``ignore_slots`` configuration.
+
+            Slots that have matching names but do not match attributes in *slots* will also be dropped.
+
+        :param cluster: cluster state information object.
+        :param slots: dictionary of desired slot names as keys with slot attributes as a dictionary value, if known.
+        :param paused: ``True`` if the patroni cluster is currently in a paused state.
+        """
+        # drop old replication slots which are not presented in desired slots.
         for name in set(self._replication_slots) - set(slots):
             if not paused and not self.ignore_replication_slot(cluster, name):
                 active, dropped = self.drop_replication_slot(name)
@@ -229,6 +344,8 @@ class SlotsHandler(object):
                         logger.debug("Unable to drop unknown replication slot '%s', slot is still active", name)
                     else:
                         logger.error("Failed to drop replication slot '%s'", name)
+
+        # drop slots with matching names but attributes that do not match, e.g. `plugin` or `database`.
         for name, value in slots.items():
             if name in self._replication_slots and not compare_slots(value, self._replication_slots[name]):
                 logger.info("Trying to drop replication slot '%s' because value is changing from %s to %s",
@@ -240,31 +357,57 @@ class SlotsHandler(object):
                     self._schedule_load_slots = True
 
     def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None:
+        """Create any missing physical replication *slots*.
+
+        Any failures are logged and do not interrupt creation of all *slots*.
+
+        :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot
+                      if the value is a dictionary with the key ``type`` and a value of ``physical``.
+        """
         immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else ''
         for name, value in slots.items():
             if name not in self._replication_slots and value['type'] == 'physical':
                 try:
-                    self._query(("SELECT pg_catalog.pg_create_physical_replication_slot(%s{0})"
-                                 " WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
-                                 " WHERE slot_type = 'physical' AND slot_name = %s)").format(
-                                     immediately_reserve), name, name)
+                    self._query(f"SELECT pg_catalog.pg_create_physical_replication_slot(%s{immediately_reserve})"
+                                f" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
+                                f" WHERE slot_type = 'physical' AND slot_name = %s)",
+                                name, name)
                 except Exception:
                     logger.exception("Failed to create physical replication slot '%s'", name)
                 self._schedule_load_slots = True
 
     @contextmanager
-    def get_local_connection_cursor(self, **kwargs: Any) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
+        """Create a new database connection to local server.
+
+        Create a non-blocking connection cursor to avoid the situation where an execution of the query of
+        ``pg_replication_slot_advance`` takes longer than the timeout on a HA loop, which could cause a false
+        failure state.
+
+        :param kwargs: Any keyword arguments to pass to :func:`psycopg.connect`.
+
+        :yields: connection cursor object, note implementation varies depending on version of :mod:`psycopg`.
+        """
         conn_kwargs = self._postgresql.config.local_connect_kwargs
         conn_kwargs.update(kwargs)
         with get_connection_cursor(**conn_kwargs) as cur:
             yield cur
 
     def _ensure_logical_slots_primary(self, slots: Dict[str, Any]) -> None:
+        """Create any missing logical replication *slots* on the primary.
+
+        If the logical slot already exists, copy state information into the replication slots structure stored in the
+        class instance.
+
+        :param slots: Slots that should exist are supplied in a dictionary, mapping slot name to any attributes.
+                      The method will only consider slots that have a value that is a dictionary with a key ``type``
+                      with a value that is ``logical``.
+
+        """
         # Group logical slots to be created by database name
         logical_slots: Dict[str, Dict[str, Dict[str, Any]]] = defaultdict(dict)
         for name, value in slots.items():
             if value['type'] == 'logical':
-                # If the logical already exists, copy some information about it into the original structure
                 if self._replication_slots.get(name, {}).get('datoid'):
                     self._copy_items(self._replication_slots[name], value)
                 else:
@@ -286,27 +429,51 @@ class SlotsHandler(object):
                     self._schedule_load_slots = True
 
     def schedule_advance_slots(self, slots: Dict[str, Dict[str, int]]) -> Tuple[bool, List[str]]:
+        """Wrapper to ensure slots advance daemon thread is started if not already.
+
+        :param slots: dictionary containing slot information.
+
+        :return: tuple with the result of the scheduling of slot advancement: ``failed`` and list of slots to copy.
+        """
         if not self._advance:
             self._advance = SlotsAdvanceThread(self)
         return self._advance.schedule(slots)
 
     def _ensure_logical_slots_replica(self, cluster: Cluster, slots: Dict[str, Any]) -> List[str]:
+        """Update logical *slots* on replicas.
+
+        If the logical slot already exists, copy state information into the replication slots structure stored in the
+        class instance. Slots that exist are also advanced if their ``confirmed_flush_lsn`` is greater than the stored
+        state of the slot.
+
+        As logical slots can only be created when the primary is available, pass the list of slots that need to be
+        copied back to the caller. They will be created on replicas with :meth:`SlotsHandler.copy_logical_slots`.
+
+        :param cluster: object containing stateful information for the cluster.
+        :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot
+                      if the value is a dictionary with the key ``type`` and a value of ``logical``.
+
+        :returns: list of slots to be copied from the primary.
+        """
         # Group logical slots to be advanced by database name
         advance_slots: Dict[str, Dict[str, int]] = defaultdict(dict)
-        create_slots: List[str] = []  # And collect logical slots to be created on the replica
+        create_slots: List[str] = []  # Collect logical slots to be created on the replica
+
         for name, value in slots.items():
-            if value['type'] == 'logical':
-                # If the logical already exists, copy some information about it into the original structure
-                if self._replication_slots.get(name, {}).get('datoid'):
-                    self._copy_items(self._replication_slots[name], value)
-                    if cluster.slots and name in cluster.slots:
-                        try:  # Skip slots that doesn't need to be advanced
-                            if value['confirmed_flush_lsn'] < int(cluster.slots[name]):
-                                advance_slots[value['database']][name] = int(cluster.slots[name])
-                        except Exception as e:
-                            logger.error('Failed to parse "%s": %r', cluster.slots[name], e)
-                elif cluster.slots and name in cluster.slots:  # We want to copy only slots with feedback in a DCS
-                    create_slots.append(name)
+            if value['type'] != 'logical':
+                continue
+
+            # If the logical already exists, copy some information about it into the original structure
+            if self._replication_slots.get(name, {}).get('datoid'):
+                self._copy_items(self._replication_slots[name], value)
+                if cluster.slots and name in cluster.slots:
+                    try:  # Skip slots that don't need to be advanced
+                        if value['confirmed_flush_lsn'] < int(cluster.slots[name]):
+                            advance_slots[value['database']][name] = int(cluster.slots[name])
+                    except Exception as e:
+                        logger.error('Failed to parse "%s": %r', cluster.slots[name], e)
+            elif cluster.slots and name in cluster.slots:  # We want to copy only slots with feedback in a DCS
+                create_slots.append(name)
 
         error, copy_slots = self.schedule_advance_slots(advance_slots)
         if error:
@@ -315,6 +482,20 @@ class SlotsHandler(object):
 
     def sync_replication_slots(self, cluster: Cluster, nofailover: bool,
                                replicatefrom: Optional[str] = None, paused: bool = False) -> List[str]:
+        """During the HA loop read, check and alter replication slots found in the cluster.
+
+        Read physical and logical slots found on the primary, then compare to those configured in the DCS.
+        Drop any slots that do not match those required by configuration and are not configured as permanent.
+        Create any missing physical slots. If we are the leader then logical slots too, otherwise if logical slots
+        are known and active create them on replica nodes.
+
+        :param cluster: object containing stateful information for the cluster.
+        :param nofailover: ``True`` if this node has been tagged to not be a failover candidate.
+        :param replicatefrom: the tag containing the node to replicate from.
+        :param paused: ``True`` if the cluster is in maintenance mode.
+
+        :returns: list of logical replication slots names that should be copied from the primary.
+        """
         ret = []
         if self._postgresql.major_version >= 90400 and cluster.config:
             try:
@@ -342,7 +523,17 @@ class SlotsHandler(object):
         return ret
 
     @contextmanager
-    def _get_leader_connection_cursor(self, leader: Leader) -> Generator[Union['cursor', 'Cursor[Any]'], None, None]:
+    def _get_leader_connection_cursor(self, leader: Leader) -> Iterator[Union['cursor', 'Cursor[Any]']]:
+        """Create a new database connection to the leader.
+
+        .. note::
+            Uses rewind user credentials because it has enough permissions to read files from PGDATA.
+            Sets the options ``connect_timeout`` to ``3`` and ``statement_timeout`` to ``2000``.
+
+        :param leader: object with information on the leader
+
+        :yields: connection cursor object, note implementation varies depending on version of ``psycopg``.
+        """
         conn_kwargs = leader.conn_kwargs(self._postgresql.config.rewind_credentials)
         conn_kwargs['dbname'] = self._postgresql.database
         with get_connection_cursor(connect_timeout=3, options="-c statement_timeout=2000", **conn_kwargs) as cur:
@@ -351,16 +542,16 @@ class SlotsHandler(object):
     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.
+        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.
+        :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.
-         """
+        :returns: ``False`` if any issue while checking logical slots readiness, ``True`` otherwise.
+        """
         catalog_xmin = None
         if self._logical_slots_processing_queue and cluster.leader:
             slot_name = cluster.get_my_slot_name_on_primary(self._postgresql.name, replicatefrom)
@@ -445,6 +636,11 @@ class SlotsHandler(object):
                     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:
+        """Create logical replication slots on standby nodes.
+
+        :param cluster: object containing stateful information for the cluster.
+        :param create_slots: list of slot names to copy from the primary.
+        """
         leader = cluster.leader
         if not leader:
             return
@@ -471,31 +667,48 @@ class SlotsHandler(object):
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         if copy_slots and self._postgresql.stop():
+            pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
             for name, value in copy_slots.items():
-                slot_dir = os.path.join(self._postgresql.slots_handler.pg_replslot_dir, name)
+                slot_dir = os.path.join(self.pg_replslot_dir, name)
                 slot_tmp_dir = slot_dir + '.tmp'
                 if os.path.exists(slot_tmp_dir):
                     shutil.rmtree(slot_tmp_dir)
                 os.makedirs(slot_tmp_dir)
+                os.chmod(slot_tmp_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_tmp_dir)
-                with open(os.path.join(slot_tmp_dir, 'state'), 'wb') as f:
+                slot_filename = os.path.join(slot_tmp_dir, 'state')
+                with open(slot_filename, 'wb') as f:
+                    os.chmod(slot_filename, pg_perm.file_create_mode)
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
                 if os.path.exists(slot_dir):
                     shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
+                os.chmod(slot_dir, pg_perm.dir_create_mode)
                 fsync_dir(slot_dir)
                 self._logical_slots_processing_queue[name] = None
-            fsync_dir(self._postgresql.slots_handler.pg_replslot_dir)
+            fsync_dir(self.pg_replslot_dir)
             self._postgresql.start()
 
     def schedule(self, value: Optional[bool] = None) -> None:
+        """Schedule the loading of slot information from the database.
+
+        :param value: the optional value can be used to unschedule if set to ``False`` or force it to be ``True``.
+                      If it is omitted the value will be ``True`` if this PostgreSQL node supports slot replication.
+        """
         if value is None:
             value = self._postgresql.major_version >= 90400
         self._schedule_load_slots = self._force_readiness_check = value
 
     def on_promote(self) -> None:
+        """Entry point from HA cycle used when a standby node is to be promoted to primary.
+
+        .. note::
+            If logical replication slot synchronisation is enabled then slot advancement will be triggered.
+            If any logical slots that were copied are yet to be confirmed as ready a warning message will be logged.
+
+        """
         if self._advance:
             self._advance.on_promote()
 

patroni/postgresql/sync.py

@@ -153,6 +153,72 @@ def parse_sync_standby_names(value: str) -> _SSN:
     return _SSN(sync_type, has_star, num, members)
 
 
+class _Replica(NamedTuple):
+    """Class representing a single replica that is eligible to be synchronous.
+
+    Attributes are taken from ``pg_stat_replication`` view and respective ``Cluster.members``.
+
+    :ivar pid: PID of walsender process.
+    :ivar application_name: matches with the ``Member.name``.
+    :ivar sync_state: possible values are: ``async``, ``potential``, ``quorum``, and ``sync``.
+    :ivar lsn: ``write_lsn``, ``flush_lsn``, or ``replay_lsn``, depending on the value of ``synchronous_commit`` GUC.
+    :ivar nofailover: whether the corresponding member has ``nofailover`` tag set to ``True``.
+    """
+    pid: int
+    application_name: str
+    sync_state: str
+    lsn: int
+    nofailover: bool
+
+
+class _ReplicaList(List[_Replica]):
+    """A collection of :class:``_Replica`` objects.
+
+    Values are reverse ordered by ``_Replica.sync_state`` and ``_Replica.lsn``.
+    That is, first there will be replicas that have ``sync_state`` == ``sync``, even if they are not
+    the most up-to-date in term of write/flush/replay LSN. It helps to keep the result of chosing new
+    synchronous nodes consistent in case if a synchronous standby member is slowed down OR async node
+    is receiving changes faster than the sync member. Such cases would trigger sync standby member
+    swapping, but only if lag on this member is exceeding a threshold (``maximum_lag_on_syncnode``).
+
+    :ivar max_lsn: maximum value of ``_Replica.lsn`` among all values. In case if there is just one
+                   element in the list we take value of ``pg_current_wal_lsn()``.
+    """
+
+    def __init__(self, postgresql: 'Postgresql', cluster: Cluster) -> None:
+        """Create :class:``_ReplicaList`` object.
+
+        :param postgresql: reference to :class:``Postgresql`` object.
+        :param cluster: currently known cluster state from DCS.
+        """
+        super().__init__()
+
+        # We want to prioritize candidates based on `write_lsn``, ``flush_lsn``, or ``replay_lsn``.
+        # Which column exactly to pick depends on the values of ``synchronous_commit`` GUC.
+        sort_col = {
+            'remote_apply': 'replay',
+            'remote_write': 'write'
+        }.get(postgresql.synchronous_commit(), 'flush') + '_lsn'
+
+        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
+        for row in postgresql.pg_stat_replication():
+            member = members.get(row['application_name'])
+
+            # We want to consider only rows from ``pg_stat_replication` that:
+            # 1. are known to be streaming (write/flush/replay LSN are not NULL).
+            # 2. can be mapped to a ``Member`` of the ``Cluster``:
+            #   a. ``Member`` doesn't have ``nosync`` tag set;
+            #   b. PostgreSQL on the member is known to be running and accepting client connections.
+            if member and row[sort_col] is not None and member.is_running and not member.tags.get('nosync', False):
+                self.append(_Replica(row['pid'], row['application_name'],
+                                     row['sync_state'], row[sort_col], bool(member.nofailover)))
+
+        # Prefer replicas that are in state ``sync`` and with higher values of ``write``/``flush``/``replay`` LSN.
+        self.sort(key=lambda r: (r.sync_state, r.lsn), reverse=True)
+
+        self.max_lsn = max(self, key=lambda x: x.lsn).lsn if len(self) > 1 else postgresql.last_operation()
+
+
 class SyncHandler(object):
     """Class responsible for working with the `synchronous_standby_names`.
 
@@ -201,6 +267,21 @@ BEGIN
 END;$$""")
         self._postgresql.reset_cluster_info_state(None)  # Reset internal cache to query fresh values
 
+    def _process_replica_readiness(self, cluster: Cluster, replica_list: _ReplicaList) -> None:
+        """Flags replicas as truly "synchronous" when they have caught up with ``_primary_flush_lsn``.
+
+        :param cluster: current cluster topology from DCS
+        :param replica_list: collection of replicas that we want to evaluate.
+        """
+        for replica in replica_list:
+            # if standby name is listed in the /sync key we can count it as synchronous, otherwise
+            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
+            if replica.application_name not in self._ready_replicas\
+                    and replica.application_name in self._ssn_data.members\
+                    and (cluster.sync.matches(replica.application_name)
+                         or replica.sync_state == 'sync' and replica.lsn >= self._primary_flush_lsn):
+                self._ready_replicas[replica.application_name] = replica.pid
+
     def current_state(self, cluster: Cluster) -> Tuple[CaseInsensitiveSet, CaseInsensitiveSet]:
         """Finds best candidates to be the synchronous standbys.
 
@@ -218,31 +299,8 @@ END;$$""")
         """
         self._handle_synchronous_standby_names_change()
 
-        # Pick candidates based on who has higher replay/remote_write/flush lsn.
-        sort_col = {
-            'remote_apply': 'replay',
-            'remote_write': 'write'
-        }.get(self._postgresql.synchronous_commit(), 'flush') + '_lsn'
-
-        pg_stat_replication = [(r['pid'], r['application_name'], r['sync_state'], r[sort_col])
-                               for r in self._postgresql.pg_stat_replication()
-                               if r[sort_col] is not None]
-
-        members = CaseInsensitiveDict({m.name: m for m in cluster.members})
-        replica_list: List[Tuple[int, str, str, int, bool]] = []
-        # pg_stat_replication.sync_state has 4 possible states - async, potential, quorum, sync.
-        # That is, alphabetically they are in the reversed order of priority.
-        # Since we are doing reversed sort on (sync_state, lsn) tuples, it helps to keep the result
-        # consistent in case if a synchronous standby member is slowed down OR async node receiving
-        # changes faster than the sync member (very rare but possible).
-        # Such cases would trigger sync standby member swapping, but only if lag on a sync node exceeding a threshold.
-        for pid, app_name, sync_state, replica_lsn in sorted(pg_stat_replication, key=lambda r: r[2:4], reverse=True):
-            member = members.get(app_name)
-            if member and member.is_running and not member.tags.get('nosync', False):
-                replica_list.append((pid, member.name, sync_state, replica_lsn, bool(member.nofailover)))
-
-        max_lsn = max(replica_list, key=lambda x: x[3])[3]\
-            if len(replica_list) > 1 else self._postgresql.last_operation()
+        replica_list = _ReplicaList(self._postgresql, cluster)
+        self._process_replica_readiness(cluster, replica_list)
 
         if TYPE_CHECKING:  # pragma: no cover
             assert self._postgresql.global_config is not None
@@ -253,17 +311,11 @@ END;$$""")
         candidates = CaseInsensitiveSet()
         sync_nodes = CaseInsensitiveSet()
         # Prefer members without nofailover tag. We are relying on the fact that sorts are guaranteed to be stable.
-        for pid, app_name, sync_state, replica_lsn, _ in sorted(replica_list, key=lambda x: x[4]):
-            # if standby name is listed in the /sync key we can count it as synchronous, otherwice
-            # it becomes really synchronous when sync_state = 'sync' and it is known that it managed to catch up
-            if app_name not in self._ready_replicas and app_name in self._ssn_data.members and\
-                    (cluster.sync.matches(app_name) or sync_state == 'sync' and replica_lsn >= self._primary_flush_lsn):
-                self._ready_replicas[app_name] = pid
-
-            if sync_node_maxlag <= 0 or max_lsn - replica_lsn <= sync_node_maxlag:
-                candidates.add(app_name)
-                if sync_state == 'sync' and app_name in self._ready_replicas:
-                    sync_nodes.add(app_name)
+        for replica in sorted(replica_list, key=lambda x: x.nofailover):
+            if sync_node_maxlag <= 0 or replica_list.max_lsn - replica.lsn <= sync_node_maxlag:
+                candidates.add(replica.application_name)
+                if replica.sync_state == 'sync' and replica.application_name in self._ready_replicas:
+                    sync_nodes.add(replica.application_name)
             if len(candidates) >= sync_node_count:
                 break
 

patroni/version.py

@@ -2,4 +2,4 @@
 
 :var __version__: the current Patroni version.
 """
-__version__ = '3.0.4'
+__version__ = '3.1.0'

postgres2.yml

@@ -121,4 +121,4 @@ tags:
     nofailover: false
     noloadbalance: false
     clonefrom: false
-    replicatefrom: postgres1
+#    replicatefrom: postgresql1

tests/test_config.py

@@ -85,6 +85,7 @@ class TestConfig(unittest.TestCase):
     @patch('os.path.exists', Mock(return_value=True))
     @patch('os.remove', Mock(side_effect=IOError))
     @patch('os.close', Mock(side_effect=IOError))
+    @patch('os.chmod', Mock())
     @patch('shutil.move', Mock(return_value=None))
     @patch('json.dump', Mock())
     def test_save_cache(self):

tests/test_file_perm.py

@@ -0,0 +1,33 @@
+import unittest
+import stat
+
+from mock import Mock, patch
+
+from patroni.file_perm import pg_perm
+
+
+class TestFilePermissions(unittest.TestCase):
+
+    @patch('os.stat')
+    @patch('os.umask')
+    @patch('patroni.file_perm.logger.error')
+    def test_set_umask(self, mock_logger, mock_umask, mock_stat):
+        mock_umask.side_effect = Exception
+        mock_stat.return_value.st_mode = stat.S_IRWXU | stat.S_IRGRP | stat.S_IXGRP
+        pg_perm.set_permissions_from_data_directory('test')
+
+        # umask is called with PG_MODE_MASK_GROUP
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IWGRP | stat.S_IRWXO)
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not set umask to %03o: %r')
+
+        mock_umask.reset_mock()
+        mock_stat.return_value.st_mode = stat.S_IRWXU
+        pg_perm.set_permissions_from_data_directory('test')
+        # umask is called with PG_MODE_MASK_OWNER (permissions changed from group to owner)
+        self.assertEqual(mock_umask.call_args[0][0], stat.S_IRWXG | stat.S_IRWXO)
+
+    @patch('os.stat', Mock(side_effect=FileNotFoundError))
+    @patch('patroni.file_perm.logger.error')
+    def test_set_permissions_from_data_directory(self, mock_logger):
+        pg_perm.set_permissions_from_data_directory('test')
+        self.assertEqual(mock_logger.call_args[0][0], 'Can not check permissions on %s: %r')

tests/test_ha.py

@@ -1584,6 +1584,7 @@ class TestHa(PostgresInit):
     @patch('os.open', Mock())
     @patch('os.fsync', Mock())
     @patch('os.close', Mock())
+    @patch('os.chmod', Mock())
     @patch('os.rename', Mock())
     @patch('patroni.postgresql.Postgresql.is_starting', Mock(return_value=False))
     @patch('builtins.open', mock_open())

tests/test_kubernetes.py

@@ -306,19 +306,19 @@ class TestKubernetesConfigMaps(BaseTestKubernetes):
 
         self.k.touch_member({'state': 'running', 'role': 'replica'})
         mock_patch_namespaced_pod.assert_called()
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false')
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'replica')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'replica')
 
         self.k.touch_member({'state': 'running', 'role': 'standby-leader'})
         mock_patch_namespaced_pod.assert_called()
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'false')
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'standby-leader')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'false')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'standby-leader')
 
         self.k._name = 'p-0'
         self.k.touch_member({'role': 'primary'})
         mock_patch_namespaced_pod.assert_called()
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['isMaster'], 'true')
-        self.assertEqual(mock_patch_namespaced_pod.call_args.args[2].metadata.labels['tmp_role'], 'master')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['isMaster'], 'true')
+        self.assertEqual(mock_patch_namespaced_pod.call_args[0][2].metadata.labels['tmp_role'], 'master')
 
     def test_initialize(self):
         self.k.initialize()

tests/test_patroni.py

@@ -40,6 +40,7 @@ class MockFrozenImporter(object):
 @patch('time.sleep', Mock())
 @patch('subprocess.call', Mock(return_value=0))
 @patch('patroni.psycopg.connect', psycopg_connect)
+@patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
 @patch.object(ConfigHandler, 'append_pg_hba', Mock())
 @patch.object(ConfigHandler, 'write_postgresql_conf', Mock())
 @patch.object(ConfigHandler, 'write_recovery_conf', Mock())
@@ -63,6 +64,7 @@ class TestPatroni(unittest.TestCase):
             self.assertRaises(SystemExit, _main)
 
     @patch('pkgutil.iter_importers', Mock(return_value=[MockFrozenImporter()]))
+    @patch('urllib3.PoolManager.request', Mock(side_effect=Exception))
     @patch('sys.frozen', Mock(return_value=True), create=True)
     @patch.object(HTTPServer, '__init__', Mock())
     @patch.object(etcd.Client, 'read', etcd_read)

tests/test_postgresql.py

@@ -346,8 +346,7 @@ class TestPostgresql(BaseTestPostgresql):
     @patch.object(Postgresql, 'start', Mock())
     def test_follow(self):
         self.p.call_nowait(CallbackAction.ON_START)
-        m = RemoteMember.from_name_and_data('1', {'restore_command': '2', 'primary_slot_name': 'foo',
-                                                  'conn_kwargs': {'host': 'bar'}})
+        m = RemoteMember('1', {'restore_command': '2', 'primary_slot_name': 'foo', 'conn_kwargs': {'host': 'bar'}})
         self.p.follow(m)
         with patch.object(Postgresql, 'ensure_major_version_is_known', Mock(return_value=False)):
             self.assertIsNone(self.p.follow(m))
@@ -523,8 +522,9 @@ class TestPostgresql(BaseTestPostgresql):
     def test_save_configuration_files(self):
         self.p.config.save_configuration_files()
 
-    @patch('os.path.isfile', Mock(side_effect=[False, True]))
-    @patch('shutil.copy', Mock(side_effect=IOError))
+    @patch('os.path.isfile', Mock(side_effect=[False, True, False, True]))
+    @patch('shutil.copy', Mock(side_effect=[None, IOError]))
+    @patch('os.chmod', Mock())
     def test_restore_configuration_files(self):
         self.p.config.restore_configuration_files()
 

tests/test_rewind.py

@@ -253,8 +253,8 @@ class TestRewind(BaseTestPostgresql):
                                          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)
+                        self.assertEqual(mock_subprocess_call.call_args[0][0], ['command 000000000000000000000000'])
+                        self.assertEqual(mock_subprocess_call.call_args[1]['shell'], True)
                         mock_subprocess_call.reset_mock()
 
             # failed archive_command call

tests/test_slots.py

@@ -67,6 +67,23 @@ class TestSlotsHandler(BaseTestPostgresql):
             with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
                 self.s.sync_replication_slots(cluster, False)
 
+    def test_cascading_replica_sync_replication_slots(self):
+        """Test sync with a cascading replica so physical slots are present on a replica."""
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}}}, 1)
+        cascading_replica = Member(0, 'test-2', 28, {
+            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5436/postgres',
+            'tags': {'replicatefrom': 'postgresql0'}
+        })
+        cluster = Cluster(True, config, self.leader, 0,
+                          [self.me, self.other, self.leadermem, cascading_replica],
+                          None, SyncState.empty(), None, {'ls': 10}, None)
+        self.p.set_role('replica')
+        with patch.object(Postgresql, '_query') as mock_query, \
+                patch.object(Postgresql, 'is_leader', Mock(return_value=False)):
+            mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)]
+            ret = self.s.sync_replication_slots(cluster, False)
+        self.assertEqual(ret, [])
+
     def test_process_permanent_slots(self):
         config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)