Method upload_change_streamed

atomic-remote/src/http/upload.rs:437–504

Upload a change file by streaming from disk. Uses reqwest's streaming body support to avoid loading the entire file into memory. The `Content-Length` header is set from file metadata so the server can validate the upload without buffering. # Arguments `hash` - The base32-encoded hash of the change. `view` - The target view name. `path` - Path to the `.change` file on disk. # Errors Returns a

(
        &self,
        hash: &str,
        view: &str,
        path: &std::path::Path,
    )

    /// Returns an error if the file can't be opened, the upload fails, or
    /// the server rejects the change.
    pub async fn upload_change_streamed(
        &self,
        hash: &str,
        view: &str,
        path: &std::path::Path,
    ) -> RemoteResult<()> {
        let url = format!("{}?insert={}&view={}", self.base_url, hash, view);
        debug!("POST insert (streamed): {} from {:?}", url, path);

        let file = tokio::fs::File::open(path).await.map_err(|e| {
            RemoteError::other(format!("Failed to open change file {:?}: {}", path, e))
        })?;

        let file_size = file
            .metadata()
            .await
            .map_err(|e| {
                RemoteError::other(format!("Failed to read metadata for {:?}: {}", path, e))
            })?
            .len();

        info!("Streaming upload of change {} ({} bytes)", hash, file_size);

        let stream = ReaderStream::new(file);
        let body = reqwest::Body::wrap_stream(stream);

        let response = self
            .client
            .post(&url)
            .header(CONTENT_TYPE, "application/octet-stream")
            .header(CONTENT_LENGTH, file_size)
            .body(body)
            .send()
            .await
            .map_err(|e| RemoteError::connection_failed(&url, e))?;

        crate::check_min_version_header(response.headers());
        let status = response.status();

        match status {
            StatusCode::OK => {
                debug!("Successfully streamed change {} from file", hash);
                Ok(())
            }
            StatusCode::NOT_FOUND => Err(RemoteError::repo_not_found(&url)),
            StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => {
                let msg = response.text().await.unwrap_or_default();
                Err(RemoteError::auth_failed(&url, msg))
            }
            StatusCode::BAD_REQUEST | StatusCode::INTERNAL_SERVER_ERROR => {
                let msg = response.text().await.unwrap_or_default();
                if msg.contains("missing") && msg.contains("dependenc") {
                    let hashes = extract_change_hashes(&msg);
                    if hashes.is_empty() {
                        Err(RemoteError::http(status.as_u16(), msg))
                    } else {
                        Err(RemoteError::missing_deps(hashes))
                    }