def copy_logical_slots(self, cluster: Cluster, tags: Tags, create_slots: List[str]) -> None:
        """Create logical replication slots on standby nodes.

        :param cluster: object containing stateful information for the cluster.
        :param tags: reference to an object implementing :class:`Tags` interface.
        :param create_slots: list of slot names to copy from the primary.
        """
        leader = cluster.leader
        if not leader:
            return
        slots = cluster.get_replication_slots(self._postgresql, tags, role=PostgresqlRole.REPLICA)
        copy_slots: Dict[str, Dict[str, Any]] = {}
        with self._get_leader_connection_cursor(leader) as cur:
            try:
                filter_failover = ' NOT failover AND' if self._postgresql.major_version >= 170000 else ''
                cur.execute("SELECT slot_name, slot_type, datname, plugin, catalog_xmin, "
                            "pg_catalog.pg_wal_lsn_diff(confirmed_flush_lsn, '0/0')::bigint, "
                            "pg_catalog.pg_read_binary_file('pg_replslot/' || slot_name || '/state')"
                            " FROM pg_catalog.pg_get_replication_slots() JOIN pg_catalog.pg_database ON datoid = oid"
                            f" WHERE{filter_failover} NOT pg_catalog.pg_is_in_recovery()"
                            " AND slot_name = ANY(%s)", (create_slots,))

                for r in cur:
                    if r[0] in slots:  # slot_name is defined in the global configuration
                        slot = {'type': r[1], 'database': r[2], 'plugin': r[3],
                                'catalog_xmin': r[4], 'confirmed_flush_lsn': r[5], 'data': r[6]}
                        if compare_slots(slot, slots[r[0]]):
                            copy_slots[r[0]] = slot
                        else:
                            logger.warning('Will not copy the logical slot "%s" due to the configuration mismatch: '
                                           'configuration=%s, slot on the primary=%s', r[0], slots[r[0]], slot)
            except Exception as e:
                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)
                slot_tmp_dir = slot_dir + '.tmp'
                if os.path.exists(slot_tmp_dir):
                    shutil.rmtree(slot_tmp_dir)
                os.makedirs(slot_tmp_dir)
                os.chmod(slot_tmp_dir, pg_perm.dir_create_mode)
                fsync_dir(slot_tmp_dir)
                slot_filename = os.path.join(slot_tmp_dir, 'state')
                with open(slot_filename, 'wb') as f:
                    os.chmod(slot_filename, pg_perm.file_create_mode)
                    f.write(value['data'])
                    f.flush()
                    os.fsync(f.fileno())
                if os.path.exists(slot_dir):
                    shutil.rmtree(slot_dir)
                os.rename(slot_tmp_dir, slot_dir)
                os.chmod(slot_dir, pg_perm.dir_create_mode)
                fsync_dir(slot_dir)
                self._logical_slots_processing_queue[name] = None
            fsync_dir(self.pg_replslot_dir)
            self._postgresql.start()
Create logical replication slots on standby nodes.

        :param cluster: object containing stateful information for the cluster.
        :param tags: reference to an object implementing :class:`Tags` interface.
        :param create_slots: list of slot names to copy from the primary.
        
patroni/postgresql/slots.py