Permanent physical slots on standby nodes (#2852)

Create permanent physical replication slots on standby nodes and use `pg_replication_slot_advance()` function to move them forward.

The `restart_lsn` is advanced based on values stored in the `/status` key by the primary node.

When slot is created on a replica it could be ahead the same slot on the primary and therefore there is some period of time when it doesn't protect WAL files from being recycled.

docs/dynamic_configuration.rst

@@ -46,7 +46,7 @@ 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. 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+.
+-  **slots**: define permanent replication slots. These slots will be preserved during switchover/failover. Permanent slots that don't exist will be created by Patroni. With PostgreSQL 11 onwards permanent physical slots are created on all nodes and their position is advanced every **loop_wait** seconds. For PostgreSQL versions older than 11 permanent physical replication slots are maintained only on 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 replication slots requires **postgresql.use_slots** to be set to ``true``. If there are permanent logical replication slots defined Patroni will 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 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.
 
@@ -81,3 +81,16 @@ Note: **slots** is a hashmap while **ignore_slots** is an array. For example:
           - name: ignored_physical_slot_name
             type: physical
           ...
+
+Note: if cluster topology is static (fixed number of nodes that never change their names) you can configure permanent physical replication slots with names corresponding to names of nodes to avoid recycling of WAL files while replica is temporary down:
+
+.. code:: YAML
+
+        slots:
+          node_name1:
+            type: physical
+          node_name2:
+            type: physical
+          node_name3:
+            type: physical
+          ...

features/ignored_slots.feature

@@ -25,10 +25,10 @@ Feature: ignored slots
     # but Patroni can actually end up dropping them almost immediately, so it's helpful
     # to verify they exist before we begin testing whether they persist through failover
     # cycles.
-    Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    Then postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
 
     When I start postgres0
     Then "members/postgres0" key in DCS has role=replica after 10 seconds
@@ -46,16 +46,16 @@ Feature: ignored slots
     And "members/postgres1" key in DCS has role=replica after 10 seconds
     # give Patroni time to sync replication slots
     And I sleep for 2 seconds
-    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds
     And postgres1 does not have a logical replication slot named dummy_slot
 
     # 3. After a failover the server (now a primary) still has the slot.
     When I shut down postgres0
     Then "members/postgres1" key in DCS has role=master after 10 seconds
-    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin
-    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin
+    And postgres1 has a logical replication slot named unmanaged_slot_0 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_1 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_2 with the test_decoding plugin after 2 seconds
+    And postgres1 has a logical replication slot named unmanaged_slot_3 with the test_decoding plugin after 2 seconds

features/permanent_slots.feature

@@ -0,0 +1,34 @@
+Feature: permanent slots
+  Scenario: check that physical permanent slots are created
+    Given I start postgres0
+    Then postgres0 is a leader after 10 seconds
+    And there is a non empty initialize key in DCS after 15 seconds
+    When I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_physical": {"type": "physical"}}, "postgresql": {"parameters": {"wal_level": "logical"}}}
+    Then I receive a response code 200
+    And Response on GET http://127.0.0.1:8008/config contains slots after 10 seconds
+    Then postgres0 has a physical replication slot named test_physical after 10 seconds
+    And I start postgres1
+
+  @slot-advance
+  Scenario: check that logical permanent slots are created
+    Given I run patronictl.py restart batman postgres0 --force
+    And I issue a PATCH request to http://127.0.0.1:8008/config with {"slots": {"test_logical": {"type": "logical", "database": "postgres", "plugin": "test_decoding"}}}
+    Then postgres0 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
+
+  @slot-advance
+  Scenario: check that permanent slots are created on the replica
+    Given postgres1 has a logical replication slot named test_logical with the test_decoding plugin after 10 seconds
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+    And postgres1 has a physical replication slot named test_physical after 2 seconds
+
+  @slot-advance
+  Scenario: check that permanent slots are advanced on the replica
+    Given I add the table replicate_me to postgres0
+    And I get all changes from physical slot test_physical on postgres0
+    When I get all changes from logical slot test_logical on postgres0
+    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
+    And Physical slot test_physical is in sync between postgres0 and postgres1 after 10 seconds
+
+  Scenario: check permanent physical replication slot after failover
+    Given I shut down postgres0
+    Then postgres1 has a physical replication slot named test_physical after 10 seconds

features/standby_cluster.feature

@@ -22,9 +22,6 @@ Feature: standby cluster
   Scenario: check permanent logical slots are synced to the replica
     Given I run patronictl.py restart batman postgres1 --force
     Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
-    When I add the table replicate_me to postgres1
-    And I get all changes from logical slot test_logical on postgres1
-    Then Logical slot test_logical is in sync between postgres0 and postgres1 after 10 seconds
 
   Scenario: Detach exiting node from the cluster
     When I shut down postgres1

features/steps/slots.py

@@ -15,17 +15,25 @@ def create_logical_replication_slot(context, slot_name, pg_name, plugin):
         assert False, "Error creating slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
 
 
-@then('{pg_name:w} has a logical replication slot named {slot_name} with the {plugin:w} plugin')
-def has_logical_replication_slot(context, pg_name, slot_name, plugin):
-    try:
-        row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots"
-                                           " WHERE slot_name = '{0}'").format(slot_name)).fetchone()
-        assert row, "Couldn't find replication slot named {0}".format(slot_name)
-        assert row[0] == "logical", "Found replication slot named {0} but wasn't a logical slot".format(slot_name)
-        assert row[1] == plugin, ("Found replication slot named {0} but was using plugin "
-                                  "{1} rather than {2}").format(slot_name, row[1], plugin)
-    except pg.Error:
-        assert False, "Error looking for slot {0} on {1} with plugin {2}".format(slot_name, pg_name, plugin)
+@step('{pg_name:w} has a logical replication slot named {slot_name}'
+      ' with the {plugin:w} plugin after {time_limit:d} seconds')
+@then('{pg_name:w} has a logical replication slot named {slot_name}'
+      ' with the {plugin:w} plugin after {time_limit:d} seconds')
+def has_logical_replication_slot(context, pg_name, slot_name, plugin, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    while time.time() < max_time:
+        try:
+            row = context.pctl.query(pg_name, ("SELECT slot_type, plugin FROM pg_replication_slots"
+                                               f" WHERE slot_name = '{slot_name}'")).fetchone()
+            if row:
+                assert row[0] == "logical", f"Replication slot {slot_name} isn't a logical but {row[0]}"
+                assert row[1] == plugin, f"Replication slot {slot_name} using plugin {row[1]} rather than {plugin}"
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, f"Error looking for slot {slot_name} on {pg_name} with plugin {plugin}"
 
 
 @then('{pg_name:w} does not have a logical replication slot named {slot_name}')
@@ -38,13 +46,14 @@ def does_not_have_logical_replication_slot(context, pg_name, slot_name):
         assert False, "Error looking for slot {0} on {1}".format(slot_name, pg_name)
 
 
-@step('Logical slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds')
-def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit):
+@step('{slot_type:w} slot {slot_name:w} is in sync between {pg_name1:w} and {pg_name2:w} after {time_limit:d} seconds')
+def slots_in_sync(context, slot_type, slot_name, pg_name1, pg_name2, time_limit):
     time_limit *= context.timeout_multiplier
     max_time = time.time() + int(time_limit)
+    column = 'confirmed_flush_lsn' if slot_type.lower() == 'logical' else 'restart_lsn'
+    query = f"SELECT {column} FROM pg_replication_slots WHERE slot_name = '{slot_name}'"
     while time.time() < max_time:
         try:
-            query = "SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = '{0}'".format(slot_name)
             slot1 = context.pctl.query(pg_name1, query).fetchone()
             slot2 = context.pctl.query(pg_name2, query).fetchone()
             if slot1[0] == slot2[0]:
@@ -52,9 +61,31 @@ def logical_slots_in_sync(context, slot_name, pg_name1, pg_name2, time_limit):
         except Exception:
             pass
         time.sleep(1)
-    assert False, "Logical slot {0} is not in sync between {1} and {2}".format(slot_name, pg_name1, pg_name2)
+    assert False, \
+        f"{slot_type} slot {slot_name} is not in sync between {pg_name1} and {pg_name2} after {time_limit} seconds"
 
 
 @step('I get all changes from logical slot {slot_name:w} on {pg_name:w}')
 def logical_slot_get_changes(context, slot_name, pg_name):
     context.pctl.query(pg_name, "SELECT * FROM pg_logical_slot_get_changes('{0}', NULL, NULL)".format(slot_name))
+
+
+@step('I get all changes from physical slot {slot_name:w} on {pg_name:w}')
+def physical_slot_get_changes(context, slot_name, pg_name):
+    context.pctl.query(pg_name, f"SELECT * FROM pg_replication_slot_advance('{slot_name}', pg_current_wal_lsn())")
+
+
+@step('{pg_name:w} has a physical replication slot named {slot_name} after {time_limit:d} seconds')
+def has_physical_replication_slot(context, pg_name, slot_name, time_limit):
+    time_limit *= context.timeout_multiplier
+    max_time = time.time() + int(time_limit)
+    query = f"SELECT * FROM pg_catalog.pg_replication_slots WHERE slot_type = 'physical' AND slot_name = '{slot_name}'"
+    while time.time() < max_time:
+        try:
+            row = context.pctl.query(pg_name, query).fetchone()
+            if row:
+                return
+        except Exception:
+            pass
+        time.sleep(1)
+    assert False, f"Physical slot {slot_name} doesn't exist after {time_limit} seconds"

patroni/dcs/__init__.py

@@ -28,6 +28,7 @@ from ..tags import Tags
 if TYPE_CHECKING:  # pragma: no cover
     from ..config import Config
 
+SLOT_ADVANCE_AVAILABLE_VERSION = 110000
 CITUS_COORDINATOR_GROUP_ID = 0
 citus_group_re = re.compile('^(0|[1-9][0-9]*)$')
 slot_name_re = re.compile('^[a-z0-9_]{1,63}$')
@@ -350,6 +351,11 @@ class Member(Tags, NamedTuple('Member',
                 logger.debug('Failed to parse Patroni version %s', version)
         return None
 
+    @property
+    def lsn(self) -> Optional[int]:
+        """Current LSN (receive/flush/replay)."""
+        return self.data.get('xlog_location')
+
 
 class RemoteMember(Member):
     """Represents a remote member (typically a primary) for a standby cluster.
@@ -964,24 +970,42 @@ class Cluster(NamedTuple('Cluster',
         candidates = [m for m in self.members if m.clonefrom and m.is_running and m.name not in exclude]
         return candidates[randint(0, len(candidates) - 1)] if candidates else self.leader
 
+    @staticmethod
+    def is_physical_slot(value: Union[Any, Dict[str, Any]]) -> bool:
+        """Check whether provided configuration is for permanent physical replication slot.
+
+        :returns: ``True`` if this is a physical replication slot, otherwise ``False``.
+        """
+        return not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'
+
     @property
     def __permanent_slots(self) -> Dict[str, Union[Dict[str, Any], Any]]:
         """Dictionary of permanent replication slots with their known LSN."""
-        ret = deepcopy(self.config.permanent_slots if self.config else {})
-        # If primary reported flush LSN for permanent slots we want to enrich our structure with it
-        for name, lsn in (self.slots or {}).items():
-            if name in ret:
-                if not ret[name]:
-                    ret[name] = {}
-                if isinstance(ret[name], dict):
-                    ret[name]['lsn'] = lsn
+        leader = self.leader and self.leader.member
+        leader_name = slot_name_from_member_name(leader.name) if leader and leader.lsn else None
+
+        slots = self.slots or {}
+        ret: Dict[str, Union[Dict[str, Any], Any]] = deepcopy(self.config.permanent_slots if self.config else {})
+
+        for name, value in list(ret.items()):
+            if not value:
+                value = ret[name] = {}
+            if isinstance(value, dict):
+                if name in slots:
+                    # If primary reported flush LSN for permanent slots we want to enrich our structure with it
+                    value['lsn'] = slots[name]
+                elif self.is_physical_slot(value) and name == leader_name and leader and leader.lsn:
+                    # there is no slot on the leader for itself, use `lsn` from the member key.
+                    value['lsn'] = leader.lsn
+                else:
+                    # Don't let anyone set 'lsn' in the global configuration :)
+                    value.pop('lsn', None)
         return ret
 
     @property
     def __permanent_physical_slots(self) -> Dict[str, Any]:
         """Dictionary of permanent ``physical`` replication slots."""
-        return {name: value for name, value in self.__permanent_slots.items()
-                if not value or isinstance(value, dict) and value.get('type', 'physical') == 'physical'}
+        return {name: value for name, value in self.__permanent_slots.items() if self.is_physical_slot(value)}
 
     @property
     def __permanent_logical_slots(self) -> Dict[str, Any]:
@@ -1013,7 +1037,7 @@ class Cluster(NamedTuple('Cluster',
         :returns: final dictionary of slot names, after merging with permanent slots and performing sanity checks.
         """
         slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, role)
-        permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover)
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(is_standby_cluster, role, nofailover, major_version)
 
         disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
             slots, permanent_slots, my_name, major_version)
@@ -1061,7 +1085,7 @@ class Cluster(NamedTuple('Cluster',
                     continue
 
                 if value['type'] == 'logical' and value.get('database') and value.get('plugin'):
-                    if major_version < 110000:
+                    if major_version < SLOT_ADVANCE_AVAILABLE_VERSION:
                         disabled_permanent_logical_slots.append(name)
                     elif name in slots:
                         logger.error("Permanent logical replication slot {'%s': %s} is conflicting with"
@@ -1073,7 +1097,8 @@ class Cluster(NamedTuple('Cluster',
             logger.error("Bad value for slot '%s' in permanent_slots: %s", name, permanent_slots[name])
         return disabled_permanent_logical_slots
 
-    def _get_permanent_slots(self, is_standby_cluster: bool, role: str, nofailover: bool) -> Dict[str, Any]:
+    def _get_permanent_slots(self, is_standby_cluster: bool, role: str,
+                             nofailover: bool, major_version: int) -> Dict[str, Any]:
         """Get configured permanent replication slots.
 
         .. note::
@@ -1089,6 +1114,7 @@ class Cluster(NamedTuple('Cluster',
                                    the outside because we want to protect from the ``/config`` key removal.
         :param role: role of this node -- ``primary``, ``standby_leader`` or ``replica``.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+        :param major_version: postgresql major version.
 
         :returns: dictionary of permanent slot names mapped to attributes.
         """
@@ -1096,9 +1122,11 @@ class Cluster(NamedTuple('Cluster',
             return {}
 
         if is_standby_cluster:
-            return self.__permanent_physical_slots if role == 'standby_leader' else {}
+            return self.__permanent_physical_slots \
+                if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION or role == 'standby_leader' else {}
 
-        return self.__permanent_slots if role in ('master', 'primary') else self.__permanent_logical_slots
+        return self.__permanent_slots if major_version >= SLOT_ADVANCE_AVAILABLE_VERSION\
+            or role in ('master', 'primary') else self.__permanent_logical_slots
 
     def _get_members_slots(self, my_name: str, role: str) -> Dict[str, Dict[str, str]]:
         """Get physical replication slots configuration for members that sourcing from this node.
@@ -1143,21 +1171,33 @@ class Cluster(NamedTuple('Cluster',
                                    for k, v in slot_conflicts.items() if len(v) > 1))
         return slots
 
-    def has_permanent_logical_slots(self, my_name: str, nofailover: bool, major_version: int = 110000) -> bool:
+    def has_permanent_slots(self, my_name: str, nofailover: bool = False) -> bool:
+        """Check if the given member node has permanent replication slots configured.
+
+        :param my_name: name of the member node to check.
+        :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
+
+        :returns: ``True`` if there are permanent replication slots configured, otherwise ``False``.
+        """
+        members_slots: Dict[str, Dict[str, str]] = self._get_members_slots(my_name, 'replica')
+        permanent_slots: Dict[str, Any] = self._get_permanent_slots(nofailover, 'replica', False,
+                                                                    SLOT_ADVANCE_AVAILABLE_VERSION)
+        slots = deepcopy(members_slots)
+        self._merge_permanent_slots(slots, permanent_slots, my_name, SLOT_ADVANCE_AVAILABLE_VERSION)
+        return len(slots) > len(members_slots)
+
+    def _has_permanent_logical_slots(self, my_name: str, nofailover: bool) -> bool:
         """Check if the given member node has permanent ``logical`` replication slots configured.
 
         :param my_name: name of the member node to check.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
-        :param major_version: the PostgreSQL major version number.
 
-        :returns: ``False`` if PostgreSQL is < 11, ``True`` if any detected replications slots are ``logical``.
+        :returns: ``True`` if any detected replications slots are ``logical``, otherwise ``False``.
         """
-        if major_version < 110000:
-            return False
-        slots = self.get_replication_slots(my_name, 'replica', nofailover, major_version).values()
+        slots = self.get_replication_slots(my_name, 'replica', nofailover, SLOT_ADVANCE_AVAILABLE_VERSION).values()
         return any(v for v in slots if v.get("type") == "logical")
 
-    def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool, major_version: int) -> bool:
+    def should_enforce_hot_standby_feedback(self, my_name: str, nofailover: bool) -> bool:
         """Determine whether ``hot_standby_feedback`` should be enabled for the given member.
 
         The ``hot_standby_feedback`` must be enabled if the current replica has ``logical`` slots,
@@ -1165,20 +1205,16 @@ class Cluster(NamedTuple('Cluster',
 
         :param my_name: name of the member node to check.
         :param nofailover: ``True`` if this node is tagged to not be a failover candidate.
-        :param major_version: PostgreSQL major version number.
 
-        :returns: ``True`` if this node or any member replicating from this node has permanent logical slots.
-                 ``False`` if PostgreSQL major version is < 11.
+        :returns: ``True`` if this node or any member replicating from this node has
+                  permanent logical slots, otherwise ``False``.
         """
-        if major_version < 110000:
-            return False
-
-        if self.has_permanent_logical_slots(my_name, nofailover, major_version):
+        if self._has_permanent_logical_slots(my_name, nofailover):
             return True
 
         if self.use_slots:
             members = [m for m in self.members if m.replicatefrom == my_name and m.name != self.leader_name]
-            return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover, major_version) for m in members)
+            return any(self.should_enforce_hot_standby_feedback(m.name, m.nofailover) for m in members)
         return False
 
     def get_my_slot_name_on_primary(self, my_name: str, replicatefrom: Optional[str]) -> str:

patroni/dcs/zookeeper.py

@@ -301,7 +301,7 @@ class ZooKeeper(AbstractDCS):
                 raise ZooKeeperError('ZooKeeper in not responding properly')
         # The /status ZNode was updated or doesn't exist
         elif self._fetch_status and not self._fetch_cluster or not cluster.last_lsn \
-                or cluster.has_permanent_logical_slots(self._name, False) and not cluster.slots:
+                or cluster.has_permanent_slots(self._name) and not cluster.slots:
             # If current node is the leader just clear the event without fetching anything (we are updating the /status)
             if cluster.leader and cluster.leader.name == self._name:
                 self.event.clear()

patroni/postgresql/__init__.py

@@ -27,7 +27,7 @@ from .sync import SyncHandler
 from .. import psycopg
 from ..async_executor import CriticalTask
 from ..collections import CaseInsensitiveSet
-from ..dcs import Cluster, Leader, Member
+from ..dcs import Cluster, Leader, Member, SLOT_ADVANCE_AVAILABLE_VERSION
 from ..exceptions import PostgresConnectionException
 from ..utils import Retry, RetryFailedError, polling_loop, data_directory_is_empty, parse_int
 
@@ -112,7 +112,7 @@ class Postgresql(object):
         self._state_entry_timestamp = 0
 
         self._cluster_info_state = {}
-        self._has_permanent_logical_slots = True
+        self._has_permanent_slots = True
         self._enforce_hot_standby_feedback = False
         self._cached_replica_timeline = None
 
@@ -174,6 +174,11 @@ class Postgresql(object):
         """:returns: `True` if Postgres version supports more than one synchronous node."""
         return self._major_version >= 90600
 
+    @property
+    def can_advance_slots(self) -> bool:
+        """``True`` if :attr:``major_version`` is greater than 110000."""
+        return self.major_version >= SLOT_ADVANCE_AVAILABLE_VERSION
+
     @property
     def cluster_info_query(self) -> str:
         """Returns the monitoring query with a fixed number of fields.
@@ -208,8 +213,9 @@ class Postgresql(object):
             extra = ("pg_catalog.current_setting('restore_command')" if self._major_version >= 120000 else "NULL") +\
                 ", " + ("(SELECT pg_catalog.json_agg(s.*) FROM (SELECT slot_name, slot_type as type, datoid::bigint, "
                         "plugin, catalog_xmin, pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint"
-                        " AS confirmed_flush_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
-                        if self._has_permanent_logical_slots and self._major_version >= 110000 else "NULL") + extra
+                        " AS confirmed_flush_lsn, pg_catalog.pg_wal_lsn_diff(restart_lsn, '0/0')::bigint"
+                        " AS restart_lsn FROM pg_catalog.pg_get_replication_slots()) AS s)"
+                        if self._has_permanent_slots and self.can_advance_slots else "NULL") + extra
             extra = (", CASE WHEN latest_end_lsn IS NULL THEN NULL ELSE received_tli END,"
                      " slot_name, conninfo, status, {0} FROM pg_catalog.pg_stat_get_wal_receiver()").format(extra)
             if self.role == 'standby_leader':
@@ -434,18 +440,15 @@ class Postgresql(object):
             return
 
         if self._global_config.is_standby_cluster:
+            self._has_permanent_slots = False
             # Standby cluster can't have logical replication slots, and we don't need to enforce hot_standby_feedback
-            self._has_permanent_logical_slots = False
             self.set_enforce_hot_standby_feedback(False)
         elif cluster and cluster.config and cluster.config.modify_version:
-            self._has_permanent_logical_slots =\
-                cluster.has_permanent_logical_slots(self.name, nofailover, self.major_version)
-
+            self._has_permanent_slots = cluster.has_permanent_slots(self.name, nofailover)
             # We want to enable hot_standby_feedback if the replica is supposed
             # to have a logical slot or in case if it is the cascading replica.
             self.set_enforce_hot_standby_feedback(
-                self._has_permanent_logical_slots
-                or cluster.should_enforce_hot_standby_feedback(self.name, nofailover, self.major_version))
+                self.can_advance_slots and cluster.should_enforce_hot_standby_feedback(self.name, nofailover))
 
     def _cluster_info_state_get(self, name: str) -> Optional[Any]:
         if not self._cluster_info_state:
@@ -456,7 +459,7 @@ class Postgresql(object):
                                                'received_tli', 'slot_name', 'conninfo', 'receiver_state',
                                                'restore_command', 'slots', 'synchronous_commit',
                                                'synchronous_standby_names', 'pg_stat_replication'], result))
-                if self._has_permanent_logical_slots:
+                if self._has_permanent_slots and self.can_advance_slots:
                     cluster_info_state['slots'] =\
                         self.slots_handler.process_permanent_slots(cluster_info_state['slots'])
                 self._cluster_info_state = cluster_info_state

patroni/postgresql/slots.py

@@ -16,6 +16,7 @@ from .misc import format_lsn, fsync_dir
 from ..dcs import Cluster, Leader
 from ..file_perm import pg_perm
 from ..psycopg import OperationalError
+from ..utils import parse_int
 
 if TYPE_CHECKING:  # pragma: no cover
     from psycopg import Cursor
@@ -231,15 +232,16 @@ class SlotsHandler:
         ret: Dict[str, int] = {}
 
         slots_dict: Dict[str, Dict[str, Any]] = {slot['slot_name']: slot for slot in slots or []}
-        if slots_dict:
-            for name, value in slots_dict.items():
-                if name in self._replication_slots:
-                    if compare_slots(value, self._replication_slots[name], 'datoid'):
-                        if value['type'] == 'logical':
-                            ret[name] = value['confirmed_flush_lsn']
-                            self._copy_items(value, self._replication_slots[name])
+        for name, value in slots_dict.items():
+            if name in self._replication_slots:
+                if compare_slots(value, self._replication_slots[name], 'datoid'):
+                    if value['type'] == 'logical':
+                        ret[name] = value['confirmed_flush_lsn']
+                        self._copy_items(value, self._replication_slots[name])
                     else:
-                        self._schedule_load_slots = True
+                        self._replication_slots[name]['restart_lsn'] = ret[name] = value['restart_lsn']
+                else:
+                    self._schedule_load_slots = True
 
         # It could happen that the slot was deleted in the background, we want to detect this case
         if any(name not in slots_dict for name in self._replication_slots.keys()):
@@ -260,16 +262,19 @@ class SlotsHandler:
         """
         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" \
+            pg_wal_lsn_diff = f"pg_catalog.pg_{self._postgresql.wal_name}_{self._postgresql.lsn_name}_diff"
+            extra = f", catalog_xmin, {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 ''
-            for r in self._query('SELECT slot_name, slot_type, plugin, database, datoid'
-                                 f'{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}'):
+            for r in self._query(f"SELECT slot_name, slot_type, {pg_wal_lsn_diff}(restart_lsn, '0/0')::bigint, plugin,"
+                                 f" database, datoid{extra} FROM pg_catalog.pg_replication_slots{skip_temp_slots}"):
                 value = {'type': r[1]}
                 if r[1] == 'logical':
-                    value.update(plugin=r[2], database=r[3], datoid=r[4])
+                    value.update(plugin=r[3], database=r[4], datoid=r[5])
                     if self._postgresql.major_version >= 100000:
-                        value.update(catalog_xmin=r[5], confirmed_flush_lsn=r[6])
+                        value.update(catalog_xmin=r[6], confirmed_flush_lsn=r[7])
+                else:
+                    value['restart_lsn'] = r[2]
                 replication_slots[r[0]] = value
             self._replication_slots = replication_slots
             self._schedule_load_slots = False
@@ -353,7 +358,7 @@ class SlotsHandler:
                     self._schedule_load_slots = True
 
     def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None:
-        """Create any missing physical replication *slots*.
+        """Create or advance physical replication *slots*.
 
         Any failures are logged and do not interrupt creation of all *slots*.
 
@@ -362,7 +367,9 @@ class SlotsHandler:
         """
         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':
+            if value['type'] != 'physical':
+                continue
+            if name not in self._replication_slots:
                 try:
                     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"
@@ -371,6 +378,16 @@ class SlotsHandler:
                 except Exception:
                     logger.exception("Failed to create physical replication slot '%s'", name)
                 self._schedule_load_slots = True
+            elif not self._postgresql.is_primary() and self._postgresql.can_advance_slots \
+                    and self._replication_slots[name]['type'] == 'physical':
+                value['restart_lsn'] = self._replication_slots[name]['restart_lsn']
+                lsn = parse_int(value.get('lsn'))
+                if lsn and lsn > value['restart_lsn']:  # The slot has feedback in DCS and needs to be advanced
+                    try:
+                        lsn = format_lsn(lsn)
+                        self._query("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", name, lsn)
+                    except Exception as exc:
+                        logger.error("Error while advancing replication slot %s to position '%s': %r", name, lsn, exc)
 
     @contextmanager
     def get_local_connection_cursor(self, **kwargs: Any) -> Iterator[Union['cursor', 'Cursor[Any]']]:
@@ -484,10 +501,11 @@ class SlotsHandler:
                                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.
+        Read physical and logical slots from ``pg_replication_slots``, 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.
+        Create any missing physical slots, or advance their position according to feedback stored in DCS.
+        If we are the primary then create logical slots, otherwise if logical slots are known and active create
+        them on replica nodes by copying slot files from the primary.
 
         :param cluster: object containing stateful information for the cluster.
         :param nofailover: ``True`` if this node has been tagged to not be a failover candidate.

tests/__init__.py

@@ -104,12 +104,14 @@ class MockCursor(object):
         elif sql.startswith('SELECT slot_name, slot_type, datname, plugin, catalog_xmin'):
             self.results = [('ls', 'logical', 'a', 'b', 100, 500, b'123456')]
         elif sql.startswith('SELECT slot_name'):
-            self.results = [('blabla', 'physical'), ('foobar', 'physical'), ('ls', 'logical', 'b', 'a', 5, 100, 500)]
+            self.results = [('blabla', 'physical', 12345),
+                            ('foobar', 'physical', 12345),
+                            ('ls', 'logical', 499, 'b', 'a', 5, 100, 500)]
         elif sql.startswith('WITH slots AS (SELECT slot_name, active'):
             self.results = [(False, True)] if self.rowcount == 1 else []
         elif sql.startswith('SELECT CASE WHEN pg_catalog.pg_is_in_recovery()'):
             self.results = [(1, 2, 1, 0, False, 1, 1, None, None, 'streaming', '',
-                             [{"slot_name": "ls", "confirmed_flush_lsn": 12345}],
+                             [{"slot_name": "ls", "confirmed_flush_lsn": 12345, "restart_lsn": 12344}],
                              'on', 'n1', None)]
         elif sql.startswith('SELECT pg_catalog.pg_is_in_recovery()'):
             self.results = [(False, 2)]
@@ -252,8 +254,8 @@ class BaseTestPostgresql(PostgresInit):
         if not os.path.exists(self.p.data_dir):
             os.makedirs(self.p.data_dir)
 
-        self.leadermem = Member(0, 'leader', 28, {
-            'state': 'running', 'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
+        self.leadermem = Member(0, 'leader', 28, {'xlog_location': 100, 'state': 'running',
+                                                  'conn_url': 'postgres://replicator:[email protected]:5435/postgres'})
         self.leader = Leader(-1, 28, self.leadermem)
         self.other = Member(0, 'test-1', 28, {'conn_url': 'postgres://replicator:[email protected]:5433/postgres',
                                               'state': 'running', 'tags': {'replicatefrom': 'leader'}})

tests/test_slots.py

@@ -53,6 +53,7 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('replica')
         with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
+            config.data['slots'].pop('ls')
             self.s.sync_replication_slots(cluster, False, paused=True)
             mock_drop.assert_not_called()
         self.p.set_role('primary')
@@ -69,6 +70,8 @@ class TestSlotsHandler(BaseTestPostgresql):
             self.assertTrue("test.3" in ca, "non matching {0}".format(ca))
             with patch.object(Postgresql, 'major_version', PropertyMock(return_value=90618)):
                 self.s.sync_replication_slots(cluster, False)
+                self.p.set_role('replica')
+                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."""
@@ -82,12 +85,12 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.p.set_role('replica')
         with patch.object(Postgresql, '_query') as mock_query, \
                 patch.object(Postgresql, 'is_primary', Mock(return_value=False)):
-            mock_query.return_value = [('ls', 'logical', 'b', 'a', 5, 12345, 105)]
+            mock_query.return_value = [('ls', 'logical', 104, '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'}},
+        config = ClusterConfig(1, {'slots': {'ls': {'database': 'a', 'plugin': 'b'}, 'blabla': {'type': 'physical'}},
                                    'ignore_slots': [{'name': 'blabla'}]}, 1)
         cluster = Cluster(True, config, self.leader, Status.empty(), [self.me, self.other, self.leadermem],
                           None, SyncState.empty(), None, None)
@@ -98,8 +101,10 @@ class TestSlotsHandler(BaseTestPostgresql):
             mock_query.return_value = [(
                 1, 0, 0, 0, 0, 0, 0, 0, 0, None, None,
                 [{"slot_name": "ls", "type": "logical", "datoid": 5, "plugin": "b",
-                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105}])]
-            self.assertEqual(self.p.slots(), {'ls': 12345})
+                  "confirmed_flush_lsn": 12345, "catalog_xmin": 105, "restart_lsn": 12344},
+                 {"slot_name": "blabla", "type": "physical", "datoid": None, "plugin": None,
+                  "confirmed_flush_lsn": None, "catalog_xmin": 105, "restart_lsn": 12344}])]
+            self.assertEqual(self.p.slots(), {'ls': 12345, 'blabla': 12344})
 
             self.p.reset_cluster_info_state(None)
             mock_query.return_value = [(
@@ -114,8 +119,8 @@ class TestSlotsHandler(BaseTestPostgresql):
         self.cluster.slots['ls'] = 12346
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, False), [])
-        self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
+        with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 499, 'b', 'a', 5, 100, 500)])), \
+                patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
             type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
@@ -177,3 +182,18 @@ class TestSlotsHandler(BaseTestPostgresql):
         with patch.object(SlotsHandler, 'get_local_connection_cursor', Mock(side_effect=Exception)):
             self.s.schedule_advance_slots({'foo': {'bar': 100}})
             self.s._advance.sync_slots()
+
+    @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
+    def test_advance_physical_slots(self):
+        config = ClusterConfig(1, {'slots': {'blabla': {'type': 'physical'}, 'leader': None}}, 1)
+        cluster = Cluster(True, config, self.leader, Status(0, {'blabla': 12346}),
+                          [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
+        self.s.sync_replication_slots(cluster, False)
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', 12345, None, None, None,
+                                                                      None, None)], Exception])) as mock_query, \
+                patch('patroni.postgresql.slots.logger.error') as mock_error:
+            self.s.sync_replication_slots(cluster, False)
+            self.assertEqual(mock_query.call_args[0],
+                             ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "blabla", '0/303A'))
+            self.assertEqual(mock_error.call_args[0][0],
+                             "Error while advancing replication slot %s to position '%s': %r")