I wrote a test for  `PART MOVE TO SHARD` runs multiple times and received some unexpected behavior.

**Test description:**
1) Create a table
2) Make 2 rows insert on one shard and 1 row insert on some other
3) Stop Merges
4) Create Distributed table
5) Receive part UUID
6) Move this part multiple times
7) Make concurrent check on the number of rows

**How to reproduce**
1) CREATE TABLE IF NOT EXISTS {table_name} on CLUSTER {cluster_name}
                                                (v UInt64) "
                                                ENGINE = {table_engine}('/clickhouse/tables/replicated/{shard}/{table_name}', '{replica}') 
                                                ORDER BY tuple() 
                                                SETTINGS assign_part_uuids=1,
                                                part_moves_between_shards_enable=1,
                                                part_moves_between_shards_delay_seconds=2;

2) INSERT INTO {table_name} VALUES ({value}) - 2 rows  on one shard and 1 row on some other

3) SYSTEM STOP MERGES {table_name} - on all nodes

4) CREATE TABLE IF NOT EXISTS {table_name_d} as {table_name} 
                                                    ENGINE = Distributed({cluster_name}, currentDatabase(), {table_name})
5) SELECT uuid FROM system.parts where name = 'all_0_0_0'

6) SELECT name FROM system.parts where uuid = '{part_uuid}'
    ALTER TABLE {table_name} MOVE PART name TO SHARD '/clickhouse/tables/replicated/{shard1}/{table_name}' - multiple times

7) select count() from {table_name_d}

* ClickHouse server version to use
22.1.2
* Queries to run that lead to an unexpected result
select count() from {table_name_d}

Result: 4

**Expected behavior**
Result: 3

**Additional information**
If make retries some time result becomes correct. The test gives the correct result if put part_moves_between_shards_delay_seconds=0.

std::make_unique<BackupEntryFromImmutableFile>(disk, hardlink_filepath, checksum.file_size, file_hash, temp_dir_owner));
        }
        for (const auto & filepath : part->getFileNamesWithoutChecksums())
        {
            String relative_filepath = fs::path(part->relative_path) / filepath;
            backup_entries.emplace_back(relative_filepath, std::make_unique<BackupEntryFromSmallFile>(disk, part_dir / filepath));
        }
    }
    return backup_entries;
}
RestoreDataTasks MergeTreeData::restoreDataPartsFromBackup(const BackupPtr & backup, const String & data_path_in_backup,
                                                           const std::unordered_set<String> & partition_ids,
                                                           SimpleIncrement * increment)
{
    RestoreDataTasks restore_tasks;
    Strings part_names = backup->listFiles(data_path_in_backup);
    for (const String & part_name : part_names)
    {
        const auto part_info = MergeTreePartInfo::tryParsePartName(part_name, format_version);
        if (!part_info)
            continue;
        if (!partition_ids.empty() && !partition_ids.contains(part_info->partition_id))
            continue;
        UInt64 total_size_of_part = 0;
        Strings filenames = backup->listFiles(data_path_in_backup + part_name + "/", "");
        for (const String & filename : filenames)
            total_size_of_part += backup->getFileSize(data_path_in_backup + part_name + "/" + filename);
        std::shared_ptr<IReservation> reservation = getStoragePolicy()->reserveAndCheck(total_size_of_part);
        auto restore_task = [this,
                             backup,