def process_permanent_slots(self, slots: List[Dict[str, Any]]) -> Dict[str, int]:
        """Process replication slot information from the host and prepare information used in subsequent cluster tasks.

        .. note::
            This methods solves three problems.

            The ``cluster_info_query`` from :class:``Postgresql`` is executed every HA loop and returns information
            about all replication slots that exists on the current host.

            Based on this information perform the following actions:

            1. For the primary we want to expose to DCS permanent logical slots, therefore build (and return) a dict
               that maps permanent logical slot names to ``confirmed_flush_lsn``.
            2. detect if one of the previously known permanent slots is missing and schedule resync.
            3. Update the local cache with the fresh ``catalog_xmin`` and ``confirmed_flush_lsn`` for every known slot.

           This info is used when performing the check of logical slot readiness on standbys.

        :param slots: replication slot information that exists on the current host.

        :return: dictionary of logical slot names to ``confirmed_flush_lsn``.
        """
        ret: Dict[str, int] = {}

        slots_dict: Dict[str, Dict[str, Any]] = {slot['slot_name']: slot for slot in slots or []}
        for name, value in slots_dict.items():
            if name in self._replication_slots:
                if compare_slots(value, self._replication_slots[name], 'datoid'):
                    if value['type'] == 'logical':
                        ret[name] = value['confirmed_flush_lsn']
                        self._copy_items(value, self._replication_slots[name])
                    else:
                        ret[name] = value['restart_lsn']
                        self._copy_items(value, self._replication_slots[name], ('restart_lsn', 'xmin'))
                else:
                    self._schedule_load_slots = True

        # It could happen that the slot was deleted in the background, we want to detect this case
        if any(name not in slots_dict for name in self._replication_slots.keys()):
            self._schedule_load_slots = True

        return ret
Process replication slot information from the host and prepare information used in subsequent cluster tasks.

        .. note::
            This methods solves three problems.

            The ``cluster_info_query`` from :class:``Postgresql`` is executed every HA loop and returns information
            about all replication slots that exists on the current host.

            Based on this information perform the following actions:

            1. For the primary we want to expose to DCS permanent logical slots, therefore build (and return) a dict
               that maps permanent logical slot names to ``confirmed_flush_lsn``.
            2. detect if one of the previously known permanent slots is missing and schedule resync.
            3. Update the local cache with the fresh ``catalog_xmin`` and ``confirmed_flush_lsn`` for every known slot.

           This info is used when performing the check of logical slot readiness on standbys.

        :param slots: replication slot information that exists on the current host.

        :return: dictionary of logical slot names to ``confirmed_flush_lsn``.
        
patroni/postgresql/slots.py