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 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`.

        :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot
                      if the value is a dictionary with the key ``type`` and a value of ``logical``.

        :returns: list of slots to be copied from the primary.
        """
        # Group logical slots to be advanced by database name
        advance_slots: Dict[str, Dict[str, int]] = defaultdict(dict)
        create_slots: List[str] = []  # Collect logical slots to be created on the replica

        for name, value in slots.items():
            if value['type'] != 'logical':
                continue

            # 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)

                # 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
                    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)

        # Slots to be copied from the primary should be removed from the *slots* structure,
        # otherwise Patroni falsely assumes that they already exist.
        for name in create_slots:
            slots.pop(name)

        error, copy_slots = self.schedule_advance_slots(advance_slots)
        if error:
            self._schedule_load_slots = True
        return create_slots + copy_slots
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 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`.

        :param slots: A dictionary mapping slot name to slot attributes. This method only considers a slot
                      if the value is a dictionary with the key ``type`` and a value of ``logical``.

        :returns: list of slots to be copied from the primary.
        
patroni/postgresql/slots.py