Follow up on patroni#3148 (patroni#3167)

the original fix didn't address same problem with with permanent slots,
but only the part with member slots being retained due to
`member_slots_ttl`.

Author: CyberDem0n

patroni/dcs/__init__.py

@@ -1028,16 +1028,16 @@ def get_replication_slots(self, postgresql: 'Postgresql', member: Tags, *,
         permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role)
 
         disabled_permanent_logical_slots: List[str] = self._merge_permanent_slots(
-            slots, permanent_slots, name, postgresql.can_advance_slots)
+            slots, permanent_slots, name, role, postgresql.can_advance_slots)
 
         if disabled_permanent_logical_slots and show_error:
             logger.error("Permanent logical replication slots supported by Patroni only starting from PostgreSQL 11. "
                          "Following slots will not be created: %s.", disabled_permanent_logical_slots)
 
         return slots
 
-    def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slots: Dict[str, Any], name: str,
-                               can_advance_slots: bool) -> List[str]:
+    def _merge_permanent_slots(self, slots: Dict[str, Dict[str, Any]], permanent_slots: Dict[str, Any],
+                               name: str, role: str, can_advance_slots: bool) -> List[str]:
         """Merge replication *slots* for members with *permanent_slots*.
 
         Perform validation of configured permanent slot name, skipping invalid names.
@@ -1047,12 +1047,17 @@ def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slo
 
         :param slots: Slot names with existing attributes if known.
         :param name: name of this node.
+        :param role: role of the node -- ``primary``, ``standby_leader`` or ``replica``.
         :param permanent_slots: dictionary containing slot name key and slot information values.
         :param can_advance_slots: ``True`` if ``pg_replication_slot_advance()`` function is available,
                                   ``False`` otherwise.
 
         :returns: List of disabled permanent, logical slot names, if postgresql version < 11.
         """
+        name = slot_name_from_member_name(name)
+        topology = {slot_name_from_member_name(m.name): m.replicatefrom and slot_name_from_member_name(m.replicatefrom)
+                    for m in self.members}
+
         disabled_permanent_logical_slots: List[str] = []
 
         for slot_name, value in permanent_slots.items():
@@ -1068,8 +1073,14 @@ def _merge_permanent_slots(self, slots: Dict[str, Dict[str, str]], permanent_slo
 
                 if value['type'] == 'physical':
                     # Don't try to create permanent physical replication slot for yourself
-                    if slot_name not in slots and slot_name != slot_name_from_member_name(name):
-                        slots[slot_name] = value
+                    if slot_name not in slots and slot_name != name:
+                        # On the leader we expected to have permanent slots active, except the case when it is a slot
+                        # for a cascading replica. Lets consider a configuration with C being a permanent slot. In this
+                        # case we should have the following: A(B: active, C: inactive) <- B (C: active) <- C
+                        # We don't consider the same situation on node B, because if node C doesn't exists, we will not
+                        # be able to know its `replicatefrom` tag value.
+                        expected_active = not topology.get(slot_name) and role in ('primary', 'standby_leader')
+                        slots[slot_name] = {**value, 'expected_active': expected_active}
                     continue
 
                 if self.is_logical_slot(value):
@@ -1228,7 +1239,7 @@ def has_permanent_slots(self, postgresql: 'Postgresql', member: Tags) -> bool:
                                                                            postgresql.can_advance_slots)
         permanent_slots: Dict[str, Any] = self._get_permanent_slots(postgresql, member, role)
         slots = deepcopy(members_slots)
-        self._merge_permanent_slots(slots, permanent_slots, postgresql.name, postgresql.can_advance_slots)
+        self._merge_permanent_slots(slots, permanent_slots, postgresql.name, role, postgresql.can_advance_slots)
         return len(slots) > len(members_slots) or any(self.is_physical_slot(v) for v in permanent_slots.values())
 
     def maybe_filter_permanent_slots(self, postgresql: 'Postgresql', slots: Dict[str, int]) -> Dict[str, int]:

tests/test_slots.py

@@ -295,7 +295,33 @@ def test_slots_advance_thread(self):
             self.s.schedule_advance_slots({'foo': {'bar': 100}})
             self.s._advance.sync_slots()
 
+    def test_advance_physical_primary(self):
+        self.p.name = self.me.name
+        config = ClusterConfig(1, {'member_slots_ttl': 0, 'slots': {'test_1': {'type': 'physical'}}}, 1)
+        cluster = Cluster(True, config, self.leader, Status(0, {}, []),
+                          [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
+        self.other.data['xlog_location'] = 12346
+        global_config.update(cluster)
+
+        # Should advance permanent physical slot on the primary for a node that is cascading from the other node
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', None, 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, self.tags)
+            self.assertEqual(mock_query.call_args[0],
+                             ("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", "test_1", '0/303A'))
+            self.assertEqual(mock_error.call_args[0][0],
+                             "Error while advancing replication slot %s to position '%s': %r")
+
+        # Should drop permanent physical slot on the primary for a node
+        # that is cascading from the other node if given slot has xmin set
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
+                                                                      None, None, None)], Exception])) as mock_query:
+            self.s.sync_replication_slots(cluster, self.tags)
+            self.assertTrue(mock_query.call_args[0][0].startswith('WITH slots AS (SELECT slot_name, active'))
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
+    @patch.object(Postgresql, 'role', PropertyMock(return_value='replica'))
     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}, []),