Avoid dropping physical slots too early (#3244)

Consider a situation: there is a permanent logical slot and primary and replica are temporary down.
When Patroni is started on the former primary it starts Postgres in a standby mode, what leads to removal of physical replication slot for the replica because it has xmin.

We should postpone removal of such physical slots:
- on replica until there will be a leader in the cluster
- on primary until Postgres is promoted

patroni/postgresql/slots.py

@@ -365,13 +365,15 @@ class SlotsHandler:
                     logger.error("Failed to drop replication slot '%s'", name)
                     self._schedule_load_slots = True
 
-    def _ensure_physical_slots(self, slots: Dict[str, Any]) -> None:
+    def _ensure_physical_slots(self, slots: Dict[str, Any], clean_inactive_physical_slots: bool) -> None:
         """Create or advance 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``.
+        :param clean_inactive_physical_slots: whether replication slots with ``xmin`` and not expected
+                                              to be active should be dropped.
         """
         immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else ''
         for name, value in slots.items():
@@ -386,7 +388,7 @@ class SlotsHandler:
             if self._postgresql.can_advance_slots and name in self._replication_slots and\
                     self._replication_slots[name]['type'] == 'physical':
                 self._copy_items(self._replication_slots[name], value, ('restart_lsn', 'xmin'))
-                if value.get('expected_active') is False and value['xmin']:
+                if clean_inactive_physical_slots and value.get('expected_active') is False and value['xmin']:
                     logger.warning('Dropping physical replication slot %s because of its xmin value %s',
                                    name, value['xmin'])
                     active, dropped = self.drop_replication_slot(name)
@@ -548,7 +550,12 @@ class SlotsHandler:
 
                 self._drop_incorrect_slots(cluster, slots)
 
-                self._ensure_physical_slots(slots)
+                # We don't want to clean physical replication slots with xmin feedback if:
+                # - cluster has no leader
+                # - current node is a leader, but still running as a standby
+                clean_inactive_physical_slots = not cluster.is_unlocked() and \
+                    (cluster.leader and cluster.leader.name != self._postgresql.name or self._postgresql.is_primary())
+                self._ensure_physical_slots(slots, clean_inactive_physical_slots)
 
                 if self._postgresql.is_primary():
                     self._logical_slots_processing_queue.clear()

tests/test_slots.py

@@ -355,6 +355,13 @@ class TestSlotsHandler(BaseTestPostgresql):
                 patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(False, False))):
             self.s.sync_replication_slots(cluster, self.tags)
 
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
+                                                                      None, None, None)], Exception])), \
+                patch.object(Cluster, 'is_unlocked', Mock(return_value=True)), \
+                patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
+            self.s.sync_replication_slots(cluster, self.tags)
+            mock_drop.assert_not_called()
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     @patch.object(Postgresql, 'role', PropertyMock(return_value='replica'))
     @patch.object(TestTags, 'tags', PropertyMock(return_value={'nofailover': True}))