Fix race condition with logical slot advance and copy (#3098)

The `SlotsAdvanceThread` is asynchronously calling
pg_replication_slot_advance() and providing feedback about logical
replication slots that must be reinitialized by copying from the
primary. That is, the parent thread will learn about slots to be copied
only when scheduling the next pg_replication_slot_advance() call.
As a result it was possible situation when logical slot was copied with
PostgreSQL restart more than once.

To improve it we implement following measures:
1. do not schedule slot sync if it is in the list to be copied
2. remove to be copued slots from the `self._scheduled` structure
3. clean state of `SlotsAdvanceThread` when slot files are copied.

patroni/postgresql/slots.py

@@ -95,8 +95,8 @@ class SlotsAdvanceThread(Thread):
                 self._failed = True
 
             new_lsn = self._scheduled.get(database, {}).get(slot, 0)
-            # remove slot from the self._scheduled structure only if it wasn't changed
-            if new_lsn == lsn and database in self._scheduled:
+            # remove slot from the self._scheduled structure if it is to be copied or if it wasn't changed
+            if copy or (new_lsn == lsn and database in self._scheduled):
                 self._scheduled[database].pop(slot)
                 if not self._scheduled[database]:
                     self._scheduled.pop(database)
@@ -152,7 +152,10 @@ class SlotsAdvanceThread(Thread):
         """
         with self._condition:
             for database, values in advance_slots.items():
-                self._scheduled[database].update(values)
+                for name, value in values.items():
+                    # Don't schedule sync for slots that just failed to be advanced and scheduled to be copied
+                    if name not in self._copy_slots:
+                        self._scheduled[database][name] = value
             ret = (self._failed, self._copy_slots)
             self._copy_slots = []
             self._failed = False
@@ -160,7 +163,7 @@ class SlotsAdvanceThread(Thread):
 
         return ret
 
-    def on_promote(self) -> None:
+    def clean(self) -> None:
         """Reset state of the daemon."""
         with self._condition:
             self._scheduled.clear()
@@ -675,6 +678,8 @@ class SlotsHandler:
                 logger.error("Failed to copy logical slots from the %s via postgresql connection: %r", leader.name, e)
 
         if copy_slots and self._postgresql.stop():
+            if self._advance:
+                self._advance.clean()
             pg_perm.set_permissions_from_data_directory(self._postgresql.data_dir)
             for name, value in copy_slots.items():
                 slot_dir = os.path.join(self.pg_replslot_dir, name)
@@ -718,7 +723,7 @@ class SlotsHandler:
 
         """
         if self._advance:
-            self._advance.on_promote()
+            self._advance.clean()
 
         if self._logical_slots_processing_queue:
             logger.warning('Logical replication slots that might be unsafe to use after promote: %s',

tests/test_slots.py

@@ -274,6 +274,10 @@ class TestSlotsHandler(BaseTestPostgresql):
                 self.s.schedule_advance_slots({'foo': {'bar': 100}})
                 self.s._advance.sync_slots()
                 self.assertEqual(self.s._advance._copy_slots, ["bar"])
+                # we don't want to make attempts to advance slots that are to be copied
+                self.s.schedule_advance_slots({'foo': {'bar': 101}})
+                self.assertEqual(self.s._advance._scheduled, {})
+                self.s._advance.clean()
 
         with patch.object(SlotsAdvanceThread, 'sync_slots', Mock(side_effect=Exception)):
             self.s._advance._condition.wait = Mock()