Copy the logical slot over if advance failed due to the missing WAL (#1946)

It could happen that the replica for some reason is missing the WAL file required by the replication slot.
The nature of this phenomenon is a bit unclear, it might be that the WAL was recycled short before we copied the slot file, but, we still need a solution to this problem. If the `pg_replication_slot_advance()` fails with the `UndefinedFile` exception (requested WAL segment pg_wal/... has already been removed), the logical slot on the replica must be recreated.

patroni/postgresql/slots.py

@@ -5,6 +5,7 @@ import shutil
 
 from collections import defaultdict
 from contextlib import contextmanager
+from psycopg2.errors import UndefinedFile
 
 from .connection import get_connection_cursor
 from .misc import format_lsn
@@ -198,7 +199,9 @@ class SlotsHandler(object):
                         cur.execute("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)",
                                     (name, format_lsn(int(cluster.slots[name]))))
                     except Exception as e:
-                        logger.exception("Failed to advance logical replication slot '%s': %r", name, e)
+                        logger.error("Failed to advance logical replication slot '%s': %r", name, e)
+                        if isinstance(e, UndefinedFile):
+                            create_slots.append(name)
                         self._schedule_load_slots = True
         return create_slots
 
@@ -280,6 +283,8 @@ class SlotsHandler(object):
                     f.write(value['data'])
                     f.flush()
                     os.fsync(f.fileno())
+                if os.path.exists(slot_dir):
+                    shutil.rmtree(slot_dir)
                 os.rename(slot_tmp_dir, slot_dir)
                 fsync_dir(slot_dir)
                 self._unready_logical_slots.add(name)

tests/test_slots.py

@@ -86,8 +86,8 @@ class TestSlotsHandler(BaseTestPostgresql):
                           [self.me, self.other, self.leadermem], None, None, None, {'ls': 12346})
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         self.s._schedule_load_slots = False
-        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.OperationalError)):
-            self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
+        with patch.object(MockCursor, 'execute', Mock(side_effect=psycopg2.errors.UndefinedFile)):
+            self.assertEqual(self.s.sync_replication_slots(cluster, False), ['ls'])
         cluster.slots['ls'] = 'a'
         self.assertEqual(self.s.sync_replication_slots(cluster, False), [])
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):