google-cloud-storage 1.20.0

// Copyright 2025 Google LLC
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     https://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use super::{
    ContinueOn308, Error, Object, PerformUpload, Result, ResumableUploadStatus, Seek, SizeHint,
    StreamingSource, X_GOOG_API_CLIENT_HEADER, apply_customer_supplied_encryption_headers,
    handle_object_response, v1,
};
use crate::storage::checksum::details::{Checksum, update as checksum_update};
use futures::stream::unfold;
use gaxi::attempt_info::AttemptInfo;
use gaxi::http::HttpRequestBuilder;
use gaxi::http::reqwest::{Body, HeaderValue, Method, multipart};
use google_cloud_gax::options::internal::{PathTemplate, RequestOptionsExt, ResourceName};
use std::sync::Arc;

impl<S> PerformUpload<S>
where
    S: StreamingSource + Seek + Send + Sync + 'static,
    <S as StreamingSource>::Error: std::error::Error + Send + Sync + 'static,
    <S as Seek>::Error: std::error::Error + Send + Sync + 'static,
{
    pub(crate) async fn send_unbuffered(self, checksum_precomputation: bool) -> Result<Object> {
        let hint = self
            .payload
            .lock()
            .await
            .size_hint()
            .await
            .map_err(Error::deser)?;
        let threshold = self.options.resumable_upload_threshold() as u64;
        if hint.upper().is_none_or(|max| max >= threshold) {
            self.send_unbuffered_resumable(hint, checksum_precomputation)
                .await
        } else {
            self.send_unbuffered_single_shot(hint).await
        }
    }

    async fn send_unbuffered_resumable(
        mut self,
        hint: SizeHint,
        checksum_precomputation: bool,
    ) -> Result<Object> {
        let has_upfront_crc32c = self
            .resource()
            .checksums
            .as_ref()
            .and_then(|c| c.crc32c)
            .is_some();

        // Checksum precomputation is only executed if:
        // 1. Precomputation is enabled (default `true`, or toggled via `with_checksum_precomputation`).
        // 2. An upfront CRC32C checksum is not already known (e.g. via `with_known_crc32c`).
        // If an upfront checksum is already present, recomputing it is redundant and skipped.
        if checksum_precomputation && !has_upfront_crc32c {
            // 1. Precompute checksum by consuming the stream through ChecksummedSource.
            let mut payload = self.payload.lock().await;
            payload.seek(0_u64).await.map_err(Error::ser)?;
            while payload
                .next()
                .await
                .transpose()
                .map_err(Error::ser)?
                .is_some()
            {}
            let computed = payload.final_checksum();

            // 2. Reset the hasher, rewind the stream to byte 0, and drop lock.
            payload.reset_checksum();
            payload.seek(0_u64).await.map_err(Error::ser)?;
            drop(payload);

            // 3. Put the precomputed checksum into object metadata for the start-upload request.
            let current = self.mut_resource().checksums.get_or_insert_default();
            checksum_update(current, computed);
            // Clear checksum options so streaming PUT does not redundantly re-hash.
            self.options.checksum = Checksum::default();
        }

        let mut upload_url = None;
        let throttler = self.options.retry_throttler.clone();
        let retry = Arc::new(ContinueOn308::new(self.options.retry_policy.clone()));
        let backoff = self.options.backoff_policy.clone();
        let mut count = 0_u32;
        let inner = async move |_| {
            let previous = count;
            count += 1;
            self.resumable_attempt(&mut upload_url, hint, previous)
                .await
        };
        google_cloud_gax::retry_loop_internal::retry_loop(
            inner,
            async |duration| tokio::time::sleep(duration).await,
            true,
            throttler,
            retry,
            backoff,
        )
        .await
    }

    async fn resumable_attempt(
        &self,
        url: &mut Option<String>,
        hint: SizeHint,
        attempt_count: u32,
    ) -> Result<Object> {
        let (offset, upload_url) = if let Some(upload_url) = url.as_deref() {
            match self
                .query_resumable_upload_attempt(upload_url, attempt_count)
                .await?
            {
                ResumableUploadStatus::Finalized(object) => {
                    return Ok(*object);
                }
                ResumableUploadStatus::Partial(offset) => (offset, upload_url),
            }
        } else {
            let upload_url = self.start_resumable_upload_attempt(attempt_count).await?;
            (0_u64, url.insert(upload_url).as_str())
        };

        let range = match (offset, hint.exact()) {
            (o, None) => format!("bytes {o}-*/*"),
            (_, Some(0)) => "bytes */0".to_string(),
            (o, Some(u)) => format!("bytes {o}-{}/{u}", u - 1),
        };
        let builder = self
            .inner
            .client
            .http_builder_with_url(Method::PUT, upload_url, crate::storage::DEFAULT_HOST)?
            .header("content-type", "application/octet-stream")
            .header("Content-Range", range)
            .header(
                "x-goog-api-client",
                HeaderValue::from_static(&X_GOOG_API_CLIENT_HEADER),
            );

        let builder = apply_customer_supplied_encryption_headers(builder, &self.params);

        self.payload
            .lock()
            .await
            .seek(offset)
            .await
            .map_err(Error::ser)?;
        let payload = self.payload_to_body().await?;
        let options = self
            .options
            .gax()
            .insert_extension(PathTemplate("/upload/storage/v1/b/{bucket}/o"))
            .insert_extension(ResourceName(format!(
                "//storage.googleapis.com/{}",
                self.resource().bucket
            )));
        let builder = builder.body(payload);
        // TODO(#4862) - maybe this should also use attempt_count ?
        let response = builder.send(options, AttemptInfo::new(0)).await?;
        let object = self::handle_object_response(response).await?;
        self.validate_response_object(object).await
    }

    pub(super) async fn send_unbuffered_single_shot(self, hint: SizeHint) -> Result<Object> {
        // Single shot uploads are idempotent only if they have pre-conditions.
        let idempotent = self.options.idempotency.unwrap_or(
            self.spec.if_generation_match.is_some() || self.spec.if_metageneration_match.is_some(),
        );
        let throttler = self.options.retry_throttler.clone();
        let retry = self.options.retry_policy.clone();
        let backoff = self.options.backoff_policy.clone();
        let mut count = 0;
        // TODO(#2044) - we need to apply any timeouts here.
        let inner = async move |_| {
            let previous = count;
            count += 1;
            self.single_shot_attempt(hint, previous).await
        };
        google_cloud_gax::retry_loop_internal::retry_loop(
            inner,
            async |duration| tokio::time::sleep(duration).await,
            idempotent,
            throttler,
            retry,
            backoff,
        )
        .await
    }

    async fn single_shot_attempt(&self, hint: SizeHint, attempt_count: u32) -> Result<Object> {
        let builder = self.single_shot_builder(hint).await?;
        let options = self
            .options
            .gax()
            .insert_extension(PathTemplate("/upload/storage/v1/b/{bucket}/o"))
            .insert_extension(ResourceName(format!(
                "//storage.googleapis.com/{}",
                self.resource().bucket
            )));
        let response = builder
            .send(options, AttemptInfo::new(attempt_count))
            .await?;
        let object = super::handle_object_response(response).await?;
        self.validate_response_object(object).await
    }

    async fn single_shot_builder(&self, hint: SizeHint) -> Result<HttpRequestBuilder> {
        let bucket = &self.resource().bucket;
        let bucket_id = bucket.strip_prefix("projects/_/buckets/").ok_or_else(|| {
            Error::binding(format!(
                "malformed bucket name, it must start with `projects/_/buckets/`: {bucket}"
            ))
        })?;
        let object = &self.resource().name;
        let builder = self
            .inner
            .client
            .http_builder(Method::POST, &format!("/upload/storage/v1/b/{bucket_id}/o"))
            .query("uploadType", "multipart")
            .query("name", object)
            .header(
                "x-goog-api-client",
                HeaderValue::from_static(&X_GOOG_API_CLIENT_HEADER),
            );

        let builder = self.apply_preconditions(builder);
        let builder = apply_customer_supplied_encryption_headers(builder, &self.params);

        // Multipart part 1: JSON object metadata
        let metadata = multipart::Part::text(v1::insert_body(self.resource()).to_string())
            .mime_str("application/json; charset=UTF-8")
            .map_err(Error::ser)?;
        self.payload
            .lock()
            .await
            .seek(0)
            .await
            .map_err(Error::ser)?;

        // Multipart part 2: Media payload stream
        let payload = self.payload_to_body().await?;
        let form = multipart::Form::new().part("metadata", metadata);
        let form = if let Some(exact) = hint.exact() {
            form.part("media", multipart::Part::stream_with_length(payload, exact))
        } else {
            form.part("media", multipart::Part::stream(payload))
        };

        // Multipart part 3: Trailing checksums metadata
        // We only append Part 3 trailing metadata when on-the-fly streaming calculation
        // is active and no upfront CRC32C checksum was already placed into Part 1
        // (e.g. from known/precomputed checksums or delegated buffered uploads).
        let has_on_the_fly =
            self.options.checksum.crc32c.is_some() || self.options.checksum.md5_hash.is_some();
        let has_upfront_crc32c = self
            .resource()
            .checksums
            .as_ref()
            .and_then(|c| c.crc32c)
            .is_some();

        let form = if has_on_the_fly && !has_upfront_crc32c {
            let checksums = self.trailing_checksums_to_body()?;
            let part = multipart::Part::stream(checksums)
                .mime_str("application/json; charset=UTF-8")
                .map_err(Error::ser)?;
            form.part("checksums", part)
        } else {
            form
        };

        let builder = builder.header(
            "content-type",
            format!("multipart/related; boundary={}", form.boundary()),
        );
        Ok(builder.body(Body::wrap_stream(form.into_stream())))
    }

    async fn payload_to_body(&self) -> Result<Body> {
        let payload = self.payload.clone();
        let stream = Box::pin(unfold(Some(payload), move |state| async move {
            if let Some(payload) = state {
                let mut guard = payload.lock().await;
                if let Some(next) = guard.next().await {
                    drop(guard);
                    return Some((next, Some(payload)));
                }
            }
            None
        }));
        Ok(Body::wrap_stream(stream))
    }

    fn trailing_checksums_to_body(&self) -> Result<Body> {
        let payload = self.payload.clone();
        let stream = Box::pin(unfold(Some(payload), move |state| async move {
            if let Some(payload) = state {
                let checksums = payload.lock().await.final_checksum();

                #[derive(serde::Serialize)]
                #[serde(rename_all = "camelCase")]
                struct TrailingMetadata {
                    #[serde(skip_serializing_if = "Option::is_none")]
                    crc32c: Option<String>,
                    #[serde(skip_serializing_if = "Option::is_none")]
                    md5_hash: Option<String>,
                }

                use base64::{Engine, prelude::BASE64_STANDARD};
                let trailing = TrailingMetadata {
                    crc32c: checksums
                        .crc32c
                        .map(|c| BASE64_STANDARD.encode(c.to_be_bytes())),
                    md5_hash: if checksums.md5_hash.is_empty() {
                        None
                    } else {
                        Some(BASE64_STANDARD.encode(&checksums.md5_hash))
                    },
                };

                let body = serde_json::to_vec(&trailing)
                    .map(bytes::Bytes::from)
                    .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e));
                return Some((body, None));
            }
            None
        }));
        Ok(Body::wrap_stream(stream))
    }
}

#[cfg(test)]
mod resumable_tests;

#[cfg(test)]
mod single_shot_tests;