impl(storage): fine grained known checksum tracking (#2805)

With this change we track whether the CRC32C checksum, the MD5 hash, or
both are set to known values. Once one of them is set to a known value,
the value cannot be changed: the corresponding `with_known_*()` function
is not available. It is also not possible to revert from known values
to computed values. The `compute_*()` functions are not available.

As a side effect: the CRC32C checksum is always present, either known
or computed. It cannot be removed from the upload.

Renamed the (crate private) types from `Precomputed` to `Known*`.

Author: coryan

src/storage/src/storage/checksum.rs

@@ -15,6 +15,13 @@
 use crate::model::ObjectChecksums;
 use crate::storage::ChecksumMismatch;
 
+pub fn update(known: &mut ObjectChecksums, computed: ObjectChecksums) {
+    known.crc32c = known.crc32c.or(computed.crc32c);
+    if known.md5_hash.is_empty() {
+        known.md5_hash = computed.md5_hash;
+    }
+}
+
 /// Compare the received object checksums vs. the computed checksums.
 ///
 /// If the `crc32c` field is `None`, or the `md5_mash` field is empty, they do
@@ -84,13 +91,40 @@ impl ChecksumEngine for Null {
     }
 }
 
-/// Assumes the checksums are provided as part of the object metadata.
+/// Assumes the CRC32C checksum is known and included in the object metadata.
+#[derive(Clone, Debug)]
+pub struct KnownCrc32c;
+
+impl sealed::ChecksumEngine for KnownCrc32c {}
+
+impl ChecksumEngine for KnownCrc32c {
+    fn update(&mut self, _offset: u64, _data: &bytes::Bytes) {}
+    fn finalize(&self) -> ObjectChecksums {
+        ObjectChecksums::new()
+    }
+}
+
+/// Assumes the MD5 hash is known and included in the object metadata.
+#[derive(Clone, Debug)]
+pub struct KnownMd5;
+
+impl sealed::ChecksumEngine for KnownMd5 {}
+
+impl ChecksumEngine for KnownMd5 {
+    fn update(&mut self, _offset: u64, _data: &bytes::Bytes) {}
+    fn finalize(&self) -> ObjectChecksums {
+        ObjectChecksums::new()
+    }
+}
+
+/// Assumes both the CRC32C checksum, and the MD5 checksums are known and
+/// included in the object metadata.
 #[derive(Clone, Debug)]
-pub struct Precomputed;
+pub struct Known;
 
-impl sealed::ChecksumEngine for Precomputed {}
+impl sealed::ChecksumEngine for Known {}
 
-impl ChecksumEngine for Precomputed {
+impl ChecksumEngine for Known {
     fn update(&mut self, _offset: u64, _data: &bytes::Bytes) {}
     fn finalize(&self) -> ObjectChecksums {
         ObjectChecksums::new()
@@ -359,7 +393,7 @@ mod tests {
 
     #[test]
     fn precomputed() {
-        let mut engine = Precomputed;
+        let mut engine = Known;
         engine.update(0, &data());
         assert_eq!(engine.finalize(), ObjectChecksums::new());
     }

src/storage/src/storage/perform_upload.rs

@@ -15,7 +15,7 @@
 use super::client::{StorageInner, apply_customer_supplied_encryption_headers};
 use crate::model::Object;
 use crate::retry_policy::ContinueOn308;
-use crate::storage::checksum::{ChecksumEngine, ChecksummedSource, Precomputed};
+use crate::storage::checksum::{ChecksumEngine, ChecksummedSource, Known};
 use crate::storage::client::enc;
 use crate::storage::client::info::X_GOOG_API_CLIENT_HEADER;
 use crate::storage::v1;

src/storage/src/storage/perform_upload/buffered.rs

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use super::{
-    ChecksumEngine, ChecksummedSource, ContinueOn308, Error, IterSource, Object, PerformUpload,
-    Precomputed, Result, ResumableUploadStatus, StreamingSource, X_GOOG_API_CLIENT_HEADER,
+    ChecksumEngine, ChecksummedSource, ContinueOn308, Error, IterSource, Known, Object,
+    PerformUpload, Result, ResumableUploadStatus, StreamingSource, X_GOOG_API_CLIENT_HEADER,
     apply_customer_supplied_encryption_headers,
 };
 use crate::storage::checksum::validate as validate_checksum;
@@ -142,20 +142,15 @@ where
         // Use the computed checksum, if any, and if the spec does not have a
         // checksum already.
         let computed = stream.final_checksum();
-        let has = self
+        let current = self
             .spec
             .resource
             .get_or_insert_default()
             .checksums
             .get_or_insert_default();
-        if has.crc32c.is_none() {
-            has.crc32c = computed.crc32c;
-        }
-        if has.md5_hash.is_empty() {
-            has.md5_hash = computed.md5_hash;
-        }
+        crate::storage::checksum::update(current, computed);
         let upload = PerformUpload {
-            payload: Arc::new(Mutex::new(ChecksummedSource::new(Precomputed, source))),
+            payload: Arc::new(Mutex::new(ChecksummedSource::new(Known, source))),
             inner: self.inner,
             spec: self.spec,
             params: self.params,