Recreate (permanent) physical slot when it doesn't reserve WAL (patroni#3388)

It could happen that (permanent) physical slot was created outside of Patroni without reserving WAL.
For such slots pg_replication_slot_advance() function throws an error:
```sql
ERROR:  replication slot "foo" cannot be advanced
DETAIL:  This slot has never previously reserved WAL, or it has been invalidated.
```

To mitigate the problem we will drop (permanent) physical slots without `restart_lsn` and let state machine to recreate them on the next cycle.

Author: barthisrael

patroni/postgresql/slots.py

@@ -332,6 +332,26 @@ def drop_replication_slot(self, name: str) -> Tuple[bool, bool]:
                             ' FULL OUTER JOIN dropped ON true'), name)
         return (rows[0][0], rows[0][1]) if rows else (False, False)
 
+    def _drop_physical_slot(self, name: str) -> None:
+        """Drop a physical replication slot by name.
+
+        .. note::
+            If not able to drop the slot, it will log a message and set the flag to reload slots.
+
+        :param name: name of the slot to be dropped.
+        """
+        active, dropped = self.drop_replication_slot(name)
+        if dropped:
+            logger.info("Dropped physical replication slot '%s'", name)
+            if name in self._replication_slots:
+                del self._replication_slots[name]
+        else:
+            self._schedule_load_slots = True
+            if active:
+                logger.warning("Unable to drop replication slot '%s', slot is active", name)
+            else:
+                logger.error("Failed to drop replication slot '%s'", name)
+
     def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any]) -> None:
         """Compare required slots and configured as permanent slots with those found, dropping extraneous ones.
 
@@ -347,15 +367,8 @@ def _drop_incorrect_slots(self, cluster: Cluster, slots: Dict[str, Any]) -> None
         # drop old replication slots which are not presented in desired slots.
         for name in set(self._replication_slots) - set(slots):
             if not global_config.is_paused and not self.ignore_replication_slot(cluster, name):
-                active, dropped = self.drop_replication_slot(name)
-                if dropped:
-                    logger.info("Dropped unknown replication slot '%s'", name)
-                else:
-                    self._schedule_load_slots = True
-                    if active:
-                        logger.debug("Unable to drop unknown replication slot '%s', slot is still active", name)
-                    else:
-                        logger.error("Failed to drop replication slot '%s'", name)
+                logger.info("Trying to drop unknown replication slot '%s'", name)
+                self._drop_physical_slot(name)
 
         # drop slots with matching names but attributes that do not match, e.g. `plugin` or `database`.
         for name, value in slots.items():
@@ -394,15 +407,7 @@ def _ensure_physical_slots(self, slots: Dict[str, Any], clean_inactive_physical_
                 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)
-                    if dropped:
-                        self._replication_slots.pop(name)
-                    else:
-                        self._schedule_load_slots = True
-                        if active:
-                            logger.warning("Unable to drop replication slot '%s', slot is active", name)
-                        else:
-                            logger.error("Failed to drop replication slot '%s'", name)
+                    self._drop_physical_slot(name)
 
             # Now we will create physical replication slots that are missing.
             if name not in self._replication_slots:
@@ -417,8 +422,19 @@ def _ensure_physical_slots(self, slots: Dict[str, Any], clean_inactive_physical_
             # And advance restart_lsn on physical replication slots that are not expected to be active.
             elif self._postgresql.can_advance_slots and self._replication_slots[name]['type'] == 'physical' and\
                     value.get('expected_active') is not True and not value['xmin']:
+                restart_lsn = value.get('restart_lsn')
+                if not restart_lsn:
+                    # This slot either belongs to a member or was configured as a permanent slot. However, for some
+                    # reason the slot was created by an external agent instead of by Patroni, and it was created without
+                    # reserving the LSN. We drop the slot here, as we cannot advance it, and let Patroni recreate and
+                    # manage it in the next cycle.
+                    logger.warning('Dropping physical replication slot %s because it has no restart_lsn. '
+                                   'This slot was probably not created by Patroni, but by an external agent.',
+                                   name)
+                    self._drop_physical_slot(name)
+                    continue
                 lsn = value.get('lsn')
-                if lsn and lsn > value['restart_lsn']:  # The slot has feedback in DCS and needs to be advanced
+                if lsn and lsn > 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)

tests/test_slots.py

@@ -55,13 +55,13 @@ def test_sync_replication_slots(self):
         self.p.set_role(PostgresqlRole.STANDBY_LEADER)
         with patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(True, False))), \
                 patch.object(global_config.__class__, 'is_standby_cluster', PropertyMock(return_value=True)), \
-                patch('patroni.postgresql.slots.logger.debug') as mock_debug:
+                patch('patroni.postgresql.slots.logger.warning') as mock_warning:
             self.s.sync_replication_slots(cluster, self.tags)
-            mock_debug.assert_called_once()
+            mock_warning.assert_called_once_with("Unable to drop replication slot '%s', slot is active", 'foobar')
         self.p.set_role(PostgresqlRole.REPLICA)
         with patch.object(Postgresql, 'is_primary', Mock(return_value=False)), \
                 patch.object(global_config.__class__, 'is_paused', PropertyMock(return_value=True)), \
-                patch.object(SlotsHandler, 'drop_replication_slot') as mock_drop:
+                patch.object(SlotsHandler, '_drop_physical_slot') as mock_drop:
             config.data['slots'].pop('ls')
             self.s.sync_replication_slots(cluster, self.tags)
             mock_drop.assert_not_called()
@@ -343,6 +343,7 @@ def test_advance_physical_slots(self):
                           [self.me, self.other, self.leadermem], None, SyncState.empty(), None, None)
         global_config.update(cluster)
         self.s.sync_replication_slots(cluster, self.tags)
+
         with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', None, 12345, None, None,
                                                                       None, None, None)], Exception])) as mock_query, \
                 patch('patroni.postgresql.slots.logger.error') as mock_error:
@@ -354,7 +355,7 @@ def test_advance_physical_slots(self):
 
         with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
                                                                       None, None, None)], Exception])), \
-                patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(False, True))):
+                patch.object(SlotsHandler, '_drop_physical_slot', Mock(return_value=(True))):
             self.s.sync_replication_slots(cluster, self.tags)
 
         with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
@@ -367,16 +368,31 @@ def test_advance_physical_slots(self):
 
         with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('test_1', 'physical', 1, 12345, None, None,
                                                                       None, None, None)], Exception])), \
-                patch.object(SlotsHandler, 'drop_replication_slot', Mock(return_value=(False, False))):
+                patch.object(SlotsHandler, '_drop_physical_slot', Mock(return_value=(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:
+                patch.object(SlotsHandler, '_drop_physical_slot') as mock_drop:
             self.s.sync_replication_slots(cluster, self.tags)
             mock_drop.assert_not_called()
 
+        # If the slot has no restart_lsn, we should not try to advance it, and only warn the user that this is not an
+        # expected situation.
+        with patch.object(SlotsHandler, '_query', Mock(side_effect=[[('blabla', 'physical', None, None, None, None,
+                                                                      None, None, None)], Exception])) as mock_query, \
+                patch('patroni.postgresql.slots.logger.warning') as mock_warning, \
+                patch.object(SlotsHandler, '_drop_physical_slot') as mock_drop:
+            self.s.sync_replication_slots(cluster, self.tags)
+            for mock_call in mock_query.call_args_list:
+                self.assertNotIn("pg_catalog.pg_replication_slot_advance", mock_call[0][0])
+            self.assertEqual(mock_warning.call_args[0][0],
+                             'Dropping physical replication slot %s because it has no restart_lsn. '
+                             'This slot was probably not created by Patroni, but by an external agent.')
+            self.assertEqual(mock_warning.call_args[0][1], 'blabla')
+            mock_drop.assert_called_once_with('blabla')
+
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     @patch.object(Postgresql, 'role', PropertyMock(return_value=PostgresqlRole.REPLICA))
     @patch.object(TestTags, 'tags', PropertyMock(return_value={'nofailover': True}))
@@ -390,3 +406,50 @@ def test_slots_nofailover_tag(self):
                                                                       None, None, None)], Exception])) as mock_query:
             self.s.sync_replication_slots(cluster, self.tags)
             self.assertTrue(mock_query.call_args[0][0].startswith('SELECT slot_name, slot_type, xmin, '))
+
+    def test__drop_physical_slot(self):
+        """Test the :meth:~SlotsHandler._drop_physical_slot` method."""
+        # Should log info and remove the slot from the list when the slot is dropped
+        self.s._replication_slots['testslot'] = {'type': 'physical'}
+        self.s._schedule_load_slots = False
+        with patch.object(self.s, 'drop_replication_slot', return_value=(False, True)) as mock_drop, \
+                patch('patroni.postgresql.slots.logger.info') as mock_info, \
+                patch('patroni.postgresql.slots.logger.warning') as mock_warning, \
+                patch('patroni.postgresql.slots.logger.error') as mock_error:
+            self.s._drop_physical_slot('testslot')
+            mock_drop.assert_called_once_with('testslot')
+            mock_info.assert_called_once_with("Dropped physical replication slot '%s'", 'testslot')
+            mock_warning.assert_not_called()
+            mock_error.assert_not_called()
+            self.assertFalse(self.s._schedule_load_slots)
+            self.assertNotIn('testslot', self.s._replication_slots)
+
+        # Should log warning and keep slot in the list when the slot is active and not dropped
+        self.s._replication_slots['testslot'] = {'type': 'physical'}
+        self.s._schedule_load_slots = False
+        with patch.object(self.s, 'drop_replication_slot', return_value=(True, False)) as mock_drop, \
+                patch('patroni.postgresql.slots.logger.info') as mock_info, \
+                patch('patroni.postgresql.slots.logger.warning') as mock_warning, \
+                patch('patroni.postgresql.slots.logger.error') as mock_error:
+            self.s._drop_physical_slot('testslot')
+            mock_drop.assert_called_once_with('testslot')
+            mock_info.assert_not_called()
+            mock_warning.assert_called_once_with("Unable to drop replication slot '%s', slot is active", 'testslot')
+            mock_error.assert_not_called()
+            self.assertTrue(self.s._schedule_load_slots)
+            self.assertIn('testslot', self.s._replication_slots)
+
+        # Should log error and keep the slot in the list when the slot is not active and not dropped
+        self.s._replication_slots['testslot'] = {'type': 'physical'}
+        self.s._schedule_load_slots = False
+        with patch.object(self.s, 'drop_replication_slot', return_value=(False, False)) as mock_drop, \
+                patch('patroni.postgresql.slots.logger.info') as mock_info, \
+                patch('patroni.postgresql.slots.logger.warning') as mock_warning, \
+                patch('patroni.postgresql.slots.logger.error') as mock_error:
+            self.s._drop_physical_slot('testslot')
+            mock_drop.assert_called_once_with('testslot')
+            mock_info.assert_not_called()
+            mock_warning.assert_not_called()
+            mock_error.assert_called_once_with("Failed to drop replication slot '%s'", 'testslot')
+            self.assertTrue(self.s._schedule_load_slots)
+            self.assertIn('testslot', self.s._replication_slots)