def _ensure_physical_slots(self, slots: Dict[str, Any], clean_inactive_physical_slots: bool) -> None:
        """Create or advance physical replication *slots*.

        Any failures are logged and do not interrupt creation of all *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 ``physical``.
        :param clean_inactive_physical_slots: whether replication slots with ``xmin`` and not expected
                                              to be active should be dropped.
        """
        immediately_reserve = ', true' if self._postgresql.major_version >= 90600 else ''
        for name, value in slots.items():
            if value['type'] != 'physical':
                continue
            # First we want to detect physical replication slots that are not
            # expected to be active but have NOT NULL xmin value and drop them.
            # As the slot is not expected to be active, nothing would be consuming this
            # slot, consequently no hot-standby feedback messages would be received
            # by Postgres regarding this slot. In that case, the `xmin` value would never
            # change, which would prevent Postgres from advancing the xmin horizon.
            if self._postgresql.can_advance_slots and name in self._replication_slots and\
                    self._replication_slots[name]['type'] == 'physical':
                self._copy_items(self._replication_slots[name], value, ('restart_lsn', 'xmin'))
                if clean_inactive_physical_slots and value.get('expected_active') is False and value['xmin']:
                    logger.warning('Dropping physical replication slot %s because of its xmin value %s',
                                   name, value['xmin'])
                    active, dropped = self.drop_replication_slot(name)
                    if dropped:
                        self._replication_slots.pop(name)
                    else:
                        self._schedule_load_slots = True
                        if active:
                            logger.warning("Unable to drop replication slot '%s', slot is active", name)
                        else:
                            logger.error("Failed to drop replication slot '%s'", name)

            # Now we will create physical replication slots that are missing.
            if name not in self._replication_slots:
                try:
                    self._query(f"SELECT pg_catalog.pg_create_physical_replication_slot(%s{immediately_reserve})"
                                f" WHERE NOT EXISTS (SELECT 1 FROM pg_catalog.pg_replication_slots"
                                f" WHERE slot_type = 'physical' AND slot_name = %s)",
                                name, name)
                except Exception:
                    logger.exception("Failed to create physical replication slot '%s'", name)
                self._schedule_load_slots = True
            # And advance restart_lsn on physical replication slots that are not expected to be active.
            elif self._postgresql.can_advance_slots and self._replication_slots[name]['type'] == 'physical' and\
                    value.get('expected_active') is not True and not value['xmin']:
                lsn = value.get('lsn')
                if lsn and lsn > value['restart_lsn']:  # The slot has feedback in DCS and needs to be advanced
                    try:
                        lsn = format_lsn(lsn)
                        self._query("SELECT pg_catalog.pg_replication_slot_advance(%s, %s)", name, lsn)
                    except Exception as exc:
                        logger.error("Error while advancing replication slot %s to position '%s': %r", name, lsn, exc)
Create or advance physical replication *slots*.

        Any failures are logged and do not interrupt creation of all *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 ``physical``.
        :param clean_inactive_physical_slots: whether replication slots with ``xmin`` and not expected
                                              to be active should be dropped.
        
patroni/postgresql/slots.py