From 5c58a939d6cb3fbfeee25e1ac3dae6a0b51510d2 Mon Sep 17 00:00:00 2001 From: forhappy Date: Thu, 15 Oct 2026 12:41:57 -0700 Subject: [PATCH] fix(storage): retry corrupt immutable readbacks Share bounded hash verification and its existing corruption retry with acknowledged immutable writes. Retry only the GET and retain both identity fields from the response whose bytes passed verification. Persistent corruption still fails after two reads; exact and logical writes issue one PUT. Keep checksum-qualified and occupied-key contracts unchanged. --- crates/crab-storage/README.md | 7 ++ crates/crab-storage/src/store.rs | 127 ++++++++++++++++++++++++------- 2 files changed, 108 insertions(+), 26 deletions(-) diff --git a/crates/crab-storage/README.md b/crates/crab-storage/README.md index f82e41126..c9726cc4d 100644 --- a/crates/crab-storage/README.md +++ b/crates/crab-storage/README.md @@ -52,6 +52,13 @@ observe another object version. Occupied-key bytes still require caller validati `Store::verify_bounded` combines that content-hash contract with pre-body size admission and streamed length checks. It uses one GET for a successful read; the limit applies to buffered encoded bytes, not the caller's decoded objects. +Readback-required immutable writes use that same bounded verifier and corruption +retry: one corrupt hash response permits one more GET, while persistent corruption +still fails. An acknowledged PUT is not replayed during verification. A successful +logical create carries the ETag and version from the hash-verified GET, including +when that response follows a corrupt read. Checksum-qualified writes still skip +readback, and occupied logical content addresses still return their body for +caller validation rather than requiring the candidate's physical encoding. The admission token cancels pending admission, response headers, streamed body reads and pending listings without retrying the cancellation. diff --git a/crates/crab-storage/src/store.rs b/crates/crab-storage/src/store.rs index edfbb3d96..4c9619df6 100644 --- a/crates/crab-storage/src/store.rs +++ b/crates/crab-storage/src/store.rs @@ -662,18 +662,8 @@ impl Store { return Ok(created); } let readback_path = self.write_path(path); - let (stored, _) = self.get_with_etag_bounded(&readback_path, maximum).await?; - let actual_hash = *blake3::hash(&stored).as_bytes(); - if actual_hash != expected_hash { - return Err(StorageError::CorruptObject { - path: path.to_string(), - reason: format!( - "expected blake3 {}, got {}", - hex_lower(&expected_hash), - hex_lower(&actual_hash) - ), - }); - } + self.verify_bounded(&readback_path, &expected_hash, maximum) + .await?; Ok(created) } @@ -746,18 +736,7 @@ impl Store { }; if self.immutable_write_verification == ImmutableWriteVerification::ReadbackRequired { let maximum = u64::try_from(bytes.len()).unwrap_or(u64::MAX); - let (stored, observed_identity) = self.get_with_etag_bounded(path, maximum).await?; - let actual_hash = *blake3::hash(&stored).as_bytes(); - if actual_hash != expected_hash { - return Err(StorageError::CorruptObject { - path: path.to_string(), - reason: format!( - "expected blake3 {}, got {}", - hex_lower(&expected_hash), - hex_lower(&actual_hash) - ), - }); - } + let (_, observed_identity) = self.verified_body(path, &expected_hash, maximum).await?; // Bind proof to the object version whose bytes were verified. A // later HEAD may observe a replacement with a different encoding. identity = observed_identity; @@ -1486,13 +1465,27 @@ impl Store { expected_hash: &[u8; 32], max_bytes: u64, ) -> Result { + self.verified_body(path, expected_hash, max_bytes) + .await + .map(|(bytes, _)| bytes) + } + + // Readback and standalone verification share the corruption retry. Replay + // only the GET: an acknowledged immutable PUT must not be repeated, and + // its receipt must name the response whose bytes passed verification. + async fn verified_body( + &self, + path: &Path, + expected_hash: &[u8; 32], + max_bytes: u64, + ) -> Result<(Bytes, ETag)> { retry(&self.retry, || { let path = path.clone(); async move { - let (bytes, _etag) = self.get_with_etag_bounded(&path, max_bytes).await?; + let (bytes, identity) = self.get_with_etag_bounded(&path, max_bytes).await?; let actual = *blake3::hash(&bytes).as_bytes(); if actual == *expected_hash { - Ok(bytes) + Ok((bytes, identity)) } else { Err(StorageError::CorruptObject { path: path.to_string(), @@ -3767,6 +3760,7 @@ mod tests { BodyError, Range(Range), DistinctIdentities, + CorruptReads { bad_reads: u64, attempts: AtomicU64 }, } #[derive(Debug)] @@ -3853,6 +3847,20 @@ mod tests { response.meta.e_tag = Some("get-etag".to_owned()); response.meta.version = Some("get-version".to_owned()); } + ReadFault::CorruptReads { + bad_reads, + attempts, + } => { + let attempt = attempts.fetch_add(1, Ordering::Relaxed) + 1; + response.meta.e_tag = Some(format!("get-{attempt}-etag")); + response.meta.version = Some(format!("get-{attempt}-version")); + if attempt <= *bad_reads { + response.payload = object_store::GetResultPayload::Stream( + futures_util::stream::iter([Ok(Bytes::from_static(b"changed!!"))]) + .boxed(), + ); + } + } } } Ok(response) @@ -4281,6 +4289,73 @@ mod tests { assert!(matches!(error, StorageError::CorruptObject { .. })); } + #[tokio::test] + async fn logical_immutable_create_retries_one_corrupt_readback() { + let inner = Arc::new(ReadFaultStore { + inner: Arc::new(InMemory::new()), + fault: ReadFault::CorruptReads { + bad_reads: 1, + attempts: AtomicU64::new(0), + }, + }); + let store = memory_store_with_inner(inner); + let outcome = store + .create_or_read_immutable( + &Path::from("blobs/transient-corrupt-readback"), + Bytes::from_static(b"candidate"), + 1024, + ) + .await + .unwrap(); + assert_eq!( + outcome, + ImmutableCreateOutcome::Created(ETag { + e_tag: Some("get-2-etag".to_owned()), + version: Some("get-2-version".to_owned()), + }) + ); + } + + #[tokio::test] + async fn immutable_readback_retries_only_reads_and_still_rejects_persistent_corruption() { + for logical in [false, true] { + for bad_reads in 0..=2 { + let inner = Arc::new(crate::test_support::CountingObjectStore::new(Arc::new( + ReadFaultStore { + inner: Arc::new(InMemory::new()), + fault: ReadFault::CorruptReads { + bad_reads, + attempts: AtomicU64::new(0), + }, + }, + ))); + let store = memory_store_with_inner(inner.clone()); + let path = Path::from("blobs/verified-create"); + let body = Bytes::from_static(b"candidate"); + let outcome = if logical { + store + .create_or_read_immutable(&path, body, 1024) + .await + .map(|_| ()) + } else { + store.put_if_absent_verified(&path, body).await.map(|_| ()) + }; + let success = outcome.is_ok(); + let corrupt = matches!(outcome, Err(StorageError::CorruptObject { .. })); + assert_eq!( + (success, corrupt, inner.put_requests(), inner.counts().full), + ( + bad_reads < 2, + bad_reads == 2, + 1, + if bad_reads == 0 { 1 } else { 2 } + ), + "logical={logical}, bad_reads={bad_reads}" + ); + } + } + } + #[tokio::test] async fn logical_immutable_create_rejects_oversized_existing_body() { let store = memory_store();