You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: shard_transfer.rs
    path of file: ./repos/qdrant/lib/collection/src/collection
    
    the code of the file until where you have to start completion: use std::future::Future;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use common::defaults;
use parking_lot::Mutex;

use super::Collection;
use crate::operations::types::{CollectionError, CollectionResult};
use crate::shards::local_shard::LocalShard;
use crate::shards::replica_set::ReplicaState;
use crate::shards::shard::{PeerId, ShardId};
use crate::shards::shard_holder::ShardHolder;
use crate::shards::transfer;
use crate::shards::transfer::transfer_tasks_pool::{
    TaskResult, TransferTaskItem, TransferTaskProgress,
};
use crate::shards::transfer::{
    ShardTransfer, ShardTransferConsensus, ShardTransferKey, ShardTransferMethod,
};

impl Collection {
    pub async fn get_outgoing_transfers(&self, current_peer_id: &PeerId) -> Vec<ShardTransfer> {
        self.shards_holder
            .read()
            .await
            .get_outgoing_transfers(current_peer_id)
    }

    pub async fn check_transfer_exists(&self, transfer_key: &ShardTransferKey) -> bool {
        self.s
    
    pub async fn finish_shard_transfer(&self, transfer: ShardTransfer) -> CollectionResult<()> {
        let transfer_finished = self
            .transfer_tasks
            .lock()
            .await
            .stop_task(&transfer.key())
            .await
            == Some(TaskResult::Finished);
        log::debug!("transfer_finished: {transfer_finished}");

        let shards_holder_guard = self.shards_holder.read().await;

        // Should happen on transfer side
        // Unwrap forward proxy into local shard, or replace it with remote shard
        // depending on the `sync` flag.
        if self.this_peer_id == transfer.from {
            let proxy_promoted = transfer::driver::handle_transferred_shard_proxy(
                &shards_holder_guard,
                transfer.shard_id,
                transfer.to,
                transfer.sync,
            )
            .await?;
            log::debug!("proxy_promoted: {proxy_promoted}");
        }

        // Should happen on receiving side
        // Promote partial shard to active shard
        if self.this_peer_id == transfer.to {
            let shard_promoted =
                transfer::driver::finalize_partial_shard(&shards_holder_guard, transfer.shard_id)
                    .await?;
            log::debug!(
                "shard_promoted: {shard_promoted}, shard_id: {}, peer_id: {}",
                transfer.shard_id,
                self.this_peer_id,
            );
        }

        // Should happen on a third-party side
        // Change direction of the remote shards or add a new remote shard
        if self.this_peer_id != transfer.from {
            let remote_shard_rerouted = transfer::driver::change_remote_shard_route(
                &shards_holder_guard,
                transfer.shard_id,
                transfer.from,
                transfer.to,
                transfer.sync,
            )
            .await?;
            log::debug!("remote_shard_rerouted: {remote_shard_rerouted}");
        }
        let finish_was_registered =
            shards_holder_guard.register_finish_transfer(&transfer.key())?;
        log::debug!("finish_was_registered: {finish_was_registered}");
        Ok(())
    }