Handle logical slots invalidation on a standby (#3097)

Since PG16 logical replication slots on a standby can be invalidated due
to horizon. In this case, pg_replication_slot_advance() will fail with
ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE. We should force slot copy
(i.e., recreation) of such slots.

patroni/postgresql/slots.py

@@ -86,7 +86,8 @@ class SlotsAdvanceThread(Thread):
         except Exception as e:
             logger.error("Failed to advance logical replication slot '%s': %r", slot, e)
             failed = True
-            copy = isinstance(e, OperationalError) and e.diag.sqlstate == '58P01'  # WAL file is gone
+            # WAL file is gone or slot is invalidated
+            copy = isinstance(e, OperationalError) and e.diag.sqlstate in ('58P01', '55000')
         with self._condition:
             if self._scheduled and failed:
                 if copy and slot not in self._copy_slots:

tests/test_slots.py

@@ -269,9 +269,11 @@ class TestSlotsHandler(BaseTestPostgresql):
     def test_slots_advance_thread(self):
         with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
                 patch.object(psycopg.OperationalError, 'diag') as mock_diag:
-            type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-            self.s.schedule_advance_slots({'foo': {'bar': 100}})
-            self.s._advance.sync_slots()
+            for err in ('58P01', '55000'):
+                type(mock_diag).sqlstate = PropertyMock(return_value=err)
+                self.s.schedule_advance_slots({'foo': {'bar': 100}})
+                self.s._advance.sync_slots()
+                self.assertEqual(self.s._advance._copy_slots, ["bar"])
 
         with patch.object(SlotsAdvanceThread, 'sync_slots', Mock(side_effect=Exception)):
             self.s._advance._condition.wait = Mock()