Consider replay_lsn when advancing slots (patroni#3356)

Do not try to advance slots past replay_lsn
Advance slot to replay_lsn position if it is already past confirmed_flush_lsn of
this slot on the replica but the replica has still not replayed the
actual LSN which this slot is at on the primary (the position stored in
the DCS).

Author: hughcapet

patroni/postgresql/slots.py

@@ -491,8 +491,8 @@ def _ensure_logical_slots_replica(self, slots: Dict[str, Any]) -> List[str]:
         """Update logical *slots* on replicas.
 
         If the logical slot already exists, copy state information into the replication slots structure stored in the
-        class instance. Slots that exist are also advanced if their ``confirmed_flush_lsn`` is greater than the stored
-        state of the slot.
+        class instance. Slots that exist are also advanced if their ``confirmed_flush_lsn`` is smaller than the state
+        of the slot stored in DCS or at least the ``replay_lsn`` of this replica.
 
         As logical slots can only be created when the primary is available, pass the list of slots that need to be
         copied back to the caller. They will be created on replicas with :meth:`SlotsHandler.copy_logical_slots`.
@@ -513,9 +513,14 @@ class instance. Slots that exist are also advanced if their ``confirmed_flush_ls
             # If the logical already exists, copy some information about it into the original structure
             if name in self._replication_slots and compare_slots(value, self._replication_slots[name]):
                 self._copy_items(self._replication_slots[name], value)
-                if 'lsn' in value and value['confirmed_flush_lsn'] < value['lsn']:  # The slot has feedback in DCS
+
+                # The slot has feedback in DCS
+                if 'lsn' in value:
+                    # we can not advance past replay_lsn
+                    advance_value = min(value['lsn'], self._postgresql.replay_lsn())
                     # Skip slots that don't need to be advanced
-                    advance_slots[value['database']][name] = value['lsn']
+                    if value['confirmed_flush_lsn'] < advance_value:
+                        advance_slots[value['database']][name] = advance_value
             elif name not in self._replication_slots and 'lsn' in value:
                 # We want to copy only slots with feedback in a DCS
                 create_slots.append(name)

tests/test_slots.py

@@ -221,18 +221,32 @@ def test_should_enforce_hot_standby_feedback(self):
     @patch.object(Postgresql, 'is_primary', Mock(return_value=False))
     def test__ensure_logical_slots_replica(self):
         self.p.set_role(PostgresqlRole.REPLICA)
-        self.cluster.status.slots['ls'] = 12346
+
+        self.cluster.status.slots['ls'] = 800
         with patch.object(SlotsHandler, 'check_logical_slots_readiness', Mock(return_value=False)):
             self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
+
         with patch.object(SlotsHandler, '_query', Mock(return_value=[('ls', 'logical', 1, 499, 'b',
                                                                       'a', 5, 100, 500)])), \
                 patch.object(MockCursor, 'execute', Mock(side_effect=psycopg.OperationalError)), \
-                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))), \
-                patch.object(psycopg.OperationalError, 'diag') as mock_diag:
-            type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
-            self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
+                patch.object(SlotsAdvanceThread, 'schedule', Mock(return_value=(True, ['ls']))):
+            # copy invalidated slot
+            with patch.object(psycopg.OperationalError, 'diag') as mock_diag:
+                type(mock_diag).sqlstate = PropertyMock(return_value='58P01')
+                self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), ['ls'])
+            # advance slots based on the replay lsn value
+            with patch.object(Postgresql, 'replay_lsn', Mock(side_effect=[200, 700, 900])), \
+                 patch.object(SlotsHandler, 'schedule_advance_slots') as advance_mock:
+                self.s.sync_replication_slots(self.cluster, self.tags)
+                advance_mock.assert_called_with(dict())
+                self.s.sync_replication_slots(self.cluster, self.tags)
+                advance_mock.assert_called_with({'a': {'ls': 700}})
+                self.s.sync_replication_slots(self.cluster, self.tags)
+                advance_mock.assert_called_with({'a': {'ls': 800}})
+
         self.cluster.status.slots['ls'] = 'a'
         self.assertEqual(self.s.sync_replication_slots(self.cluster, self.tags), [])
+
         self.cluster.config.data['slots']['ls']['database'] = 'b'
         self.cluster.status.slots['ls'] = '500'
         with patch.object(MockCursor, 'rowcount', PropertyMock(return_value=1), create=True):