From a202c4c292bd7b775f65dbbcb9d36a482b6984ed Mon Sep 17 00:00:00 2001 From: b7r6 Date: Wed, 30 Sep 2026 19:02:30 -0400 Subject: [PATCH 1/2] store(dedup): reject torn content-chunk reads instead of serving them as success DedupStore::get_part trusted the per-chunk sizes recorded in the index and streamed each content chunk without checking it actually held as many bytes as its index entry declared. A short content chunk (a torn/partial read or an index/content size disagreement) would either panic the worker on the bytes_to_skip assert, or silently under-deliver and send EOF, serving a truncated read to the caller as success. Add a per-chunk size-consistency check: a fetched content chunk whose length does not equal the size its index entry declares now yields a NotFound error. The check is per-chunk (not a post-loop total), so a legitimate bounded request that overshoots the end of the blob still returns the available bytes. Regression tests: get_part_short_chunk_is_error_test (fails without the fix: "got Ok with 10 bytes") and get_part_over_length_request_still_ok_test (guards against over-correction). Co-Authored-By: Claude Opus 4.8 --- nativelink-store/src/dedup_store.rs | 17 +++++ nativelink-store/tests/dedup_store_test.rs | 83 +++++++++++++++++++++- 2 files changed, 99 insertions(+), 1 deletion(-) diff --git a/nativelink-store/src/dedup_store.rs b/nativelink-store/src/dedup_store.rs index e3b437ca00a..105be62d076 100644 --- a/nativelink-store/src/dedup_store.rs +++ b/nativelink-store/src/dedup_store.rs @@ -352,6 +352,23 @@ impl StoreDriver for DedupStore { .await .err_tip(|| "Failed to get_part in content_store in dedup_store")?; + // Content chunks are content-addressed and whole: a chunk must + // contain exactly the number of bytes its index entry declares. + // If the content store returns a short chunk (a torn/partial + // read, or an index/content size disagreement) the streaming + // loop below would either panic on the `bytes_to_skip` assert or + // silently under-deliver and send EOF as success. Reject the + // mismatch here so it surfaces as an error instead. + let declared = index_entry.size_bytes(); + let actual = data.len() as u64; + if actual != declared { + return Err(make_err!( + Code::NotFound, + "DedupStore content chunk {} is {actual} bytes but its index entry declares {declared}; refusing to serve a torn read as success", + index_entry, + )); + } + Result::<_, Error>::Ok(data) }) .buffered(self.max_concurrent_fetch_per_get); diff --git a/nativelink-store/tests/dedup_store_test.rs b/nativelink-store/tests/dedup_store_test.rs index e72f2905a62..f45f693ce0d 100644 --- a/nativelink-store/tests/dedup_store_test.rs +++ b/nativelink-store/tests/dedup_store_test.rs @@ -16,7 +16,8 @@ use nativelink_config::stores::{DedupSpec, MemorySpec, StoreSpec}; use nativelink_error::{Code, Error, ResultExt}; use nativelink_macro::nativelink_test; use nativelink_store::cas_utils::ZERO_BYTE_DIGESTS; -use nativelink_store::dedup_store::DedupStore; +use nativelink_store::compression_store::WincodeConfig; +use nativelink_store::dedup_store::{DedupIndex, DedupStore}; use nativelink_store::memory_store::MemoryStore; use nativelink_util::common::DigestInfo; use nativelink_util::store_trait::{Store, StoreLike}; @@ -445,3 +446,83 @@ async fn empty_cas_upload_does_not_publish_dedup_index() -> Result<(), Error> { assert_eq!(content.len_for_test(), 0); Ok(()) } + +// Regression: a content chunk must contain exactly the number of bytes its +// index entry declares. If the content store returns a short chunk (a torn or +// partial read, or an index/content size disagreement), `DedupStore::get_part` +// previously either panicked on the `bytes_to_skip` assert or silently +// under-delivered and sent EOF, serving a truncated read as success. This test +// hand-builds an index whose single entry declares more bytes than the stored +// content and asserts the read now surfaces an error. +#[nativelink_test] +async fn get_part_short_chunk_is_error_test() -> Result<(), Error> { + const DECLARED_SIZE: usize = 100; + const ACTUAL_SIZE: usize = 10; + + let index_store = MemoryStore::new(&MemorySpec::default()); + let content_store = MemoryStore::new(&MemorySpec::default()); + let store = DedupStore::new( + &make_default_config(), + Store::new(index_store.clone()), + Store::new(content_store.clone()), + )?; + + let chunk_digest = DigestInfo::try_new(VALID_HASH2, DECLARED_SIZE).unwrap(); + content_store + .update_oneshot(chunk_digest, vec![7u8; ACTUAL_SIZE].into()) + .await + .err_tip(|| "Failed to write short content chunk")?; + + let serialized = wincode::config::serialize( + &DedupIndex { + entries: vec![chunk_digest], + }, + WincodeConfig::new(), + ) + .unwrap(); + let top_digest = DigestInfo::try_new(VALID_HASH1, DECLARED_SIZE).unwrap(); + Store::new(index_store.clone()) + .update_oneshot(top_digest, serialized.into()) + .await + .err_tip(|| "Failed to write index entry")?; + + let result = store + .get_part_unchunked(top_digest, 0, Some(DECLARED_SIZE as u64)) + .await; + assert!( + result.is_err(), + "Expected short-chunk read to error, got Ok with {} bytes", + result.map_or(0, |d| d.len()) + ); + Ok(()) +} + +// Guard against over-correction: a bounded request for MORE bytes than the blob +// actually contains must still succeed and return the available bytes, matching +// the clamp-to-available semantics of the underlying stores. The +// size-consistency check must not turn a legitimate over-length request into an +// error. +#[nativelink_test] +async fn get_part_over_length_request_still_ok_test() -> Result<(), Error> { + const DATA_SIZE: usize = 64 * 1024; + + let store = DedupStore::new( + &make_default_config(), + Store::new(MemoryStore::new(&MemorySpec::default())), + Store::new(MemoryStore::new(&MemorySpec::default())), + )?; + + let original_data = make_random_data(DATA_SIZE); + let digest = DigestInfo::try_new(VALID_HASH1, DATA_SIZE).unwrap(); + store + .update_oneshot(digest, original_data.clone().into()) + .await + .err_tip(|| "Failed to write data to dedup store")?; + + let rt = store + .get_part_unchunked(digest, 0, Some((DATA_SIZE as u64) * 2)) + .await + .err_tip(|| "Over-length bounded read should still succeed")?; + assert_eq!(rt, original_data, "Expected available bytes to be returned"); + Ok(()) +} From c3c9ae163239e505ac6234e02b6c0df50061f450 Mon Sep 17 00:00:00 2001 From: b7r6 Date: Thu, 15 Oct 2026 12:23:46 -0400 Subject: [PATCH 2/2] store(dedup): drop a content chunk whose size does not match its digest Rejecting a wrong-size chunk with NotFound stopped the truncated read, but left the bad chunk in place. `has()` sums the sizes the content store reports, so the blob stayed "present", and `update()` skips any chunk the content store already has, so the client's re-upload kept the bad copy. The blob could never be read again. On a size mismatch, `get_part` now makes a best-effort `remove` of the chunk from the content store, as `VerifyStore` does for a copy that fails verification. The blob then reads as missing and the next upload writes the chunk again. The error stays NotFound, not DataLoss, so a `FastSlowStore` above can still fall back to its slow tier. Tests now cover the error code, the removal, a read that starts inside a short chunk (the old assert panic), a chunk longer than declared, and an end-to-end truncate / fail / re-upload / read cycle. The removal, has() and repair assertions fail without this change. Co-Authored-By: Claude Opus 5.5 --- nativelink-store/src/dedup_store.rs | 42 +++++- nativelink-store/tests/dedup_store_test.rs | 144 +++++++++++++++++---- 2 files changed, 157 insertions(+), 29 deletions(-) diff --git a/nativelink-store/src/dedup_store.rs b/nativelink-store/src/dedup_store.rs index 105be62d076..c2f4a6307c9 100644 --- a/nativelink-store/src/dedup_store.rs +++ b/nativelink-store/src/dedup_store.rs @@ -354,14 +354,46 @@ impl StoreDriver for DedupStore { // Content chunks are content-addressed and whole: a chunk must // contain exactly the number of bytes its index entry declares. - // If the content store returns a short chunk (a torn/partial - // read, or an index/content size disagreement) the streaming - // loop below would either panic on the `bytes_to_skip` assert or - // silently under-deliver and send EOF as success. Reject the - // mismatch here so it surfaces as an error instead. + // If the content store returns a chunk of any other size (a + // torn or partial read, or an index/content size disagreement) + // the streaming loop below would either panic on the + // `bytes_to_skip` assert or silently under-deliver and send EOF + // as success. Reject the mismatch here so it surfaces as an + // error instead. let declared = index_entry.size_bytes(); let actual = data.len() as u64; if actual != declared { + // Every blob that references this chunk is unreadable while + // it stays, and nothing else would notice: `has()` sums the + // sizes the content store reports, and `update()` skips any + // chunk the content store already has, so a re-upload would + // keep the bad copy. Drop it, as `VerifyStore` drops a copy + // that fails verification, so `has()` reports those blobs + // missing and the next upload writes the chunk again. This + // is best effort: a content store that does not support + // `remove` keeps the chunk. + match self + .content_store + .as_store_driver_pin() + .remove(index_entry.into()) + .await + { + Ok(removed) => warn!( + %index_entry, + actual, + declared, + removed, + "Dropped a dedup content chunk whose size does not match its digest" + ), + Err(err) => warn!( + %index_entry, + ?err, + "Could not drop a dedup content chunk whose size does not match its digest" + ), + } + // NotFound rather than the DataLoss `VerifyStore` returns: a + // `FastSlowStore` above falls back to its slow tier only on + // NotFound, and the slow tier may hold a good copy. return Err(make_err!( Code::NotFound, "DedupStore content chunk {} is {actual} bytes but its index entry declares {declared}; refusing to serve a torn read as success", diff --git a/nativelink-store/tests/dedup_store_test.rs b/nativelink-store/tests/dedup_store_test.rs index f45f693ce0d..22850e323f4 100644 --- a/nativelink-store/tests/dedup_store_test.rs +++ b/nativelink-store/tests/dedup_store_test.rs @@ -12,6 +12,8 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::sync::Arc; + use nativelink_config::stores::{DedupSpec, MemorySpec, StoreSpec}; use nativelink_error::{Code, Error, ResultExt}; use nativelink_macro::nativelink_test; @@ -447,18 +449,14 @@ async fn empty_cas_upload_does_not_publish_dedup_index() -> Result<(), Error> { Ok(()) } -// Regression: a content chunk must contain exactly the number of bytes its -// index entry declares. If the content store returns a short chunk (a torn or -// partial read, or an index/content size disagreement), `DedupStore::get_part` -// previously either panicked on the `bytes_to_skip` assert or silently -// under-delivered and sent EOF, serving a truncated read as success. This test -// hand-builds an index whose single entry declares more bytes than the stored -// content and asserts the read now surfaces an error. -#[nativelink_test] -async fn get_part_short_chunk_is_error_test() -> Result<(), Error> { - const DECLARED_SIZE: usize = 100; - const ACTUAL_SIZE: usize = 10; - +/// Builds a `DedupStore` whose index for `VALID_HASH1` names one content chunk +/// of `declared_size` bytes, and stores `stored` under that chunk's digest. A +/// `stored` of any other length is a chunk whose bytes disagree with its index +/// entry, as a torn write or a misbehaving content store would leave it. +async fn store_with_one_chunk( + declared_size: usize, + stored: Vec, +) -> Result<(Arc, Arc, DigestInfo, DigestInfo), Error> { let index_store = MemoryStore::new(&MemorySpec::default()); let content_store = MemoryStore::new(&MemorySpec::default()); let store = DedupStore::new( @@ -467,11 +465,11 @@ async fn get_part_short_chunk_is_error_test() -> Result<(), Error> { Store::new(content_store.clone()), )?; - let chunk_digest = DigestInfo::try_new(VALID_HASH2, DECLARED_SIZE).unwrap(); + let chunk_digest = DigestInfo::try_new(VALID_HASH2, declared_size).unwrap(); content_store - .update_oneshot(chunk_digest, vec![7u8; ACTUAL_SIZE].into()) + .update_oneshot(chunk_digest, stored.into()) .await - .err_tip(|| "Failed to write short content chunk")?; + .err_tip(|| "Failed to write content chunk")?; let serialized = wincode::config::serialize( &DedupIndex { @@ -480,20 +478,118 @@ async fn get_part_short_chunk_is_error_test() -> Result<(), Error> { WincodeConfig::new(), ) .unwrap(); - let top_digest = DigestInfo::try_new(VALID_HASH1, DECLARED_SIZE).unwrap(); - Store::new(index_store.clone()) + let top_digest = DigestInfo::try_new(VALID_HASH1, declared_size).unwrap(); + index_store .update_oneshot(top_digest, serialized.into()) .await .err_tip(|| "Failed to write index entry")?; + Ok((store, content_store, top_digest, chunk_digest)) +} - let result = store - .get_part_unchunked(top_digest, 0, Some(DECLARED_SIZE as u64)) - .await; - assert!( - result.is_err(), - "Expected short-chunk read to error, got Ok with {} bytes", - result.map_or(0, |d| d.len()) +// A content chunk must contain exactly the number of bytes its index entry +// declares. A short chunk used to be served as a truncated read that ended in a +// normal EOF. The read must fail with NotFound, and the bad chunk must be gone +// afterwards so `has()` stops reporting the blob as present. +#[nativelink_test] +async fn get_part_short_chunk_is_error_test() -> Result<(), Error> { + let (store, content_store, top_digest, chunk_digest) = + store_with_one_chunk(100, vec![7u8; 10]).await?; + + let result = store.get_part_unchunked(top_digest, 0, Some(100)).await; + let err = result.expect_err("a short chunk must not be served as a successful read"); + assert_eq!(err.code, Code::NotFound, "unexpected error: {err:?}"); + + assert_eq!( + content_store.has(chunk_digest).await?, + None, + "the short chunk must be dropped from the content store" ); + assert_eq!( + store.has(top_digest).await?, + None, + "a blob whose chunk was dropped must no longer be reported as present" + ); + Ok(()) +} + +// A read that starts inside a short chunk used to trip the `bytes_to_skip` +// assert and panic. It must fail like any other read of that chunk. +#[nativelink_test] +async fn get_part_offset_inside_short_chunk_is_error_test() -> Result<(), Error> { + let (store, _content_store, top_digest, _chunk_digest) = + store_with_one_chunk(100, vec![7u8; 10]).await?; + + let result = store.get_part_unchunked(top_digest, 50, Some(10)).await; + let err = result.expect_err("a read inside a short chunk must fail, not panic"); + assert_eq!(err.code, Code::NotFound, "unexpected error: {err:?}"); + Ok(()) +} + +// A chunk longer than its index entry declares is just as wrong as a short one. +#[nativelink_test] +async fn get_part_long_chunk_is_error_test() -> Result<(), Error> { + let (store, content_store, top_digest, chunk_digest) = + store_with_one_chunk(100, vec![7u8; 150]).await?; + + let result = store.get_part_unchunked(top_digest, 0, None).await; + let err = result.expect_err("a long chunk must not be served as a successful read"); + assert_eq!(err.code, Code::NotFound, "unexpected error: {err:?}"); + assert_eq!(content_store.has(chunk_digest).await?, None); + Ok(()) +} + +// End to end: a blob with one corrupted chunk fails to read, is then reported +// missing, and the client's re-upload makes it readable again. Before the bad +// chunk was dropped, `update()` skipped it because the content store still had +// a value under its digest, so the blob stayed unreadable forever. +#[nativelink_test] +async fn short_chunk_is_rewritten_by_the_next_upload_test() -> Result<(), Error> { + let index_store = MemoryStore::new(&MemorySpec::default()); + let content_store = MemoryStore::new(&MemorySpec::default()); + let store = DedupStore::new( + &make_default_config(), + Store::new(index_store.clone()), + Store::new(content_store.clone()), + )?; + + let original_data = make_random_data(MEGABYTE_SZ); + let digest = DigestInfo::try_new(VALID_HASH1, MEGABYTE_SZ).unwrap(); + store + .update_oneshot(digest, original_data.clone().into()) + .await + .err_tip(|| "Failed to write data to dedup store")?; + + // Truncate one chunk from the middle of the blob in place. + let index_bytes = index_store.get_part_unchunked(digest, 0, None).await?; + let index: DedupIndex = + wincode::config::deserialize(&index_bytes, WincodeConfig::new()).unwrap(); + assert!(index.entries.len() > 2, "expected several chunks"); + let victim = index.entries[index.entries.len() / 2]; + let victim_bytes = content_store.get_part_unchunked(victim, 0, None).await?; + content_store + .update_oneshot(victim, victim_bytes.slice(..victim_bytes.len() / 2)) + .await?; + + let err = store + .get_part_unchunked(digest, 0, None) + .await + .expect_err("a blob with a truncated chunk must not read back"); + assert_eq!(err.code, Code::NotFound, "unexpected error: {err:?}"); + assert_eq!( + store.has(digest).await?, + None, + "the blob must now be missing" + ); + + store + .update_oneshot(digest, original_data.clone().into()) + .await + .err_tip(|| "Failed to re-upload data to dedup store")?; + let rt_data = store + .get_part_unchunked(digest, 0, None) + .await + .err_tip(|| "The re-uploaded blob must read back")?; + assert_eq!(rt_data, original_data, "Expected round trip data to match"); Ok(()) }