Repository: qdrant/qdrant
File: lib/storage/src/content_manager/snapshots/download_tar.rs (Rust)

Code before:
    }
}

/// Download and unpack a tar file in streaming fashion without saving to disk first.
///
/// This function streams the HTTP response directly into the tar extractor,
/// avoiding the need to store the entire tar file on disk before extraction.
///
/// # Cancel safety
///
/// This function is cancel safe. If cancelled, the cancellation token will be triggered
/// and the download will be interrupted at the next read operation.
///
/// # Arguments
///
/// * `client` - The reqwest HTTP client to use for the download
/// * `url` - The URL to download the tar file from
/// * `target_dir` - The directory to extract the tar contents into
/// * `compute_checksum` - If true, compute and return the SHA-256 hash of the downloaded data
///
/// # Returns
///
/// Returns `Ok(Some(hash))` if `compute_checksum` is true, `Ok(None)` otherwise.
/// Returns a `StorageError` if the download or extraction fails.
pub async fn download_and_unpack_tar(
    client: &reqwest::Client,
    url: &Url,
    target_dir: &Path,
    compute_checksum: bool,
) -> Result<Option<String>, StorageError> {
    log::debug!(
        "Streaming tar download from {url} to {}",
        target_dir.display()
    );

    let response = client.get(url.clone()).send().await?;

    if !response.status().is_success() {
        return Err(StorageError::bad_input(format!(
            "Failed to download tar from {url}: status - {}",
            response.status()
        )));
    }

    // Convert the response body stream into an AsyncRead
    let stream = response.bytes_stream().map_err(std::io::Error::other);
    let async_reader = StreamReader::new(stream);

    let target_dir = target_dir.to_path_buf();
    let target_dir_for_log = target_dir.clone();


Review comment (bug):
This change is the most important in the PR.

Turns out, the `Interrupted` is a very poor choice, as it is interpreted as transient error and encouraged to be retried. From docs:

>    This operation was interrupted.
>     Interrupted operations can typically be retried.


Which is the opposite of cancelled. This caused Tar Unpack process to stuck in infinite loop, even after cancellation token was properly triggered.

Code after:
                        std::io::ErrorKind::TimedOut,
                        "stream read timeout: no data received within timeout period",
                    )));
                }
                Poll::Pending
            }
        }
    }
}

/// A sync Read wrapper that checks a cancellation token before each read.
struct CancellableReader<R> {
    inner: R,
    cancel: CancellationToken,
}

impl<R> CancellableReader<R> {
    fn new(inner: R, cancel: CancellationToken) -> Self {
        Self { inner, cancel }
    }
}

impl<R: Read> Read for CancellableReader<R> {
    fn read(&mut self, buf: &mut [u8]) -> std::io::Result<usize> {
        if self.cancel.is_cancelled() {
            return Err(std::io::Error::other("download cancelled"));
        }
        self.inner.read(buf)
    }
}

/// A sync Read wrapper that computes SHA-256 hash of the data as it's read.
struct HashingReader<R> {
    inner: R,
    hasher: Option<Sha256>,
}

impl<R> HashingReader<R> {
    fn new(inner: R, compute_hash: bool) -> Self {
        Self {
            inner,
            hasher: compute_hash.then(Sha256::new),
        }
    }

    /// Consume the reader and return the computed hash as a hex string.
    /// Returns None if hashing was not enabled.
    fn finalize(self) -> Option<String> {
        self.hasher.map(|h| {
            let hash = h.finalize();
            format!("{hash:x}")