From c3e5dbb092907b9bfd69799d082f9d8fd91878d1 Mon Sep 17 00:00:00 2001 From: Deeproute Date: Fri, 23 Oct 2026 21:01:05 -0700 Subject: [PATCH] Frame a block, and decode one, with one copy of the payload Three places on the tiered paths copied the payload for the sake of copying it again. Framing a block owned its payload first -- `value.to_vec()` -- and then copied that buffer into the block. Three of the four arms store the value as it is, so three of the four were paying two payload-sized allocations and two copies to produce one buffer that anybody keeps. A read that finds the entry below memory hands the bytes to two holders: the tier it refills and the caller. It decoded into a `Vec`, cloned the `Vec` for the tier, and let the tier build an `Arc` from the clone. Three allocations for two holders. The batch read did the same, once a key, and then cloned again for every position that had asked for that key. A decode now has two forms and the read picks the one it needs. A read that refills decodes straight into the handle the tier keeps, and copies once for the caller. A read that cannot refill -- an SSD-only instance never puts anything back into memory -- decodes into the value it is going to return and moves it out, so that configuration does not pay for a handle it has no use for. allocations bytes get, promoting from a lower tier 13.00 -> 12.00 25,334.0 -> 24,310.0 put, reaching the SSD tier, Rust 28.11 -> 27.11 59,247.8 -> 58,223.8 put, reaching the SSD tier, with the other store backend 12.11 -> 11.11 55,938.2 -> 54,914.2 One allocation and one kibibyte each, which is the payload: these are figures for a 1 KiB value, and the copies removed grow with it. The batch promoting read had no row at all -- every read row in the report was a memory hit until the previous change, and that one measured a single read. It has one now, with a bound per arm, because the batched store read is the backend's own: 59.80 allocations for eight keys on the pure-Rust store, 73.47 through the other, within 1,200 bytes of each other. Its per-key cost is dominated by work this change does not touch -- the store read, the admission, the eviction -- so it is bounded absolutely rather than held to the bookkeeping-per-key bound the resident batch rows are held to. Both regime checks now ask for nine reads in ten to have reached a lower tier rather than allowing a fixed sixteen exceptions. The batch row starts with a tier full of keys it is about to ask for, so a count of exceptions was a guess; a share of the reads is what the check is actually about, and a fixture that stopped demoting reads near zero either way. Tested: the two decodes are asserted to answer the same for twelve shapes of block -- raw, compressed, empty, shorter than the header, no magic, no bytes, a header cut short, a payload cut short, an unknown version, an unknown codec, a lying original length, a lying payload length -- accepting the same bytes or rejecting with the same message. The header checks exist twice now, and two copies of a validation drift. And the framing is asserted field by field: magic, version, codec, both lengths, the payload, the total. A block on a device outlives the process that wrote it, so the framing is a durable format, and the copy was only removable because the bytes written do not change. Planted both, one at a time, and each test catches its own defect and not the other's: dropping the original-length check from the shared decode the agreement test FAILS, the framing test passes writing the two length fields the other way round the framing test FAILS, the agreement test passes The second of those is why the framing test checks a compressed block as well. Planted against the raw block alone it passed: a raw block stores the value as it is, so its two lengths are equal by construction and swapping them writes the same bytes. The field that tells the two apart had to be checked on a block where they differ. The owning refill is gone with its last caller. Co-Authored-By: Claude Opus 5 --- examples/allocation_report.rs | 80 ++++++++++++-- src/runtime/builder_and_gc.rs | 165 +++++++++++++++++++++++++--- src/runtime/multilayer_cache.rs | 56 ++++++---- src/tests/mod.rs | 189 ++++++++++++++++++++++++++++++++ 4 files changed, 447 insertions(+), 43 deletions(-) diff --git a/examples/allocation_report.rs b/examples/allocation_report.rs index 2255490..d6a6d59 100755 --- a/examples/allocation_report.rs +++ b/examples/allocation_report.rs @@ -280,6 +280,40 @@ fn main() { // assumed from the fixture. Checked below. let promote_disk_hits = promoting.stats().disk_hits; + // The same read asked for eight keys at once. The batch path reads the + // blocks together, decodes each one once, and hands the bytes to every + // position that asked for that key -- so what it must not do is a copy per + // position on top of the one each position gets. + // + // The groups are built outside the measured closure, as the resident batch + // row builds its key list outside: a batch row that formats eight keys a + // call measures `format!`. + let promote_groups: Vec> = (0..PROMOTE_KEYS / BATCH) + .map(|group| { + ((group * BATCH)..(group * BATCH + BATCH)) + .map(key) + .collect() + }) + .collect(); + let mut promote_group_cursor = 0usize; + let batch_promotion_hits_before = promoting.stats().disk_hits; + rows.push(measure( + "get_batch, promoting from a lower tier", + BATCH, + || { + promote_group_cursor += 1; + let group = &promote_groups[promote_group_cursor % promote_groups.len()]; + let found = promoting + .get_batch(group) + .expect("a batch read of keys the fixture wrote must succeed"); + assert!( + found.iter().all(|slot| slot.is_some()), + "the fixture has to still hold every key of the group" + ); + }, + )); + let batch_promotion_hits = promoting.stats().disk_hits - batch_promotion_hits_before; + let mut reaching_ssd = 2_000_000usize; rows.push(measure("put, reaching the SSD tier", 1, || { reaching_ssd += 1; @@ -385,6 +419,7 @@ fn main() { // pinned batch of 8 15 -> 7 allocations // a write that evicts 11.48 -> 5.93 allocations, 44,140.8 -> 19,537.3 bytes // a batch of 8 shared 2 allocations for 8 keys + // a promoting read 13.00 -> 12.00 allocations, 25,334.0 -> 24,310.0 bytes // // Raising one is a deliberate act, and the number that replaced it goes // in the table beside it. @@ -397,7 +432,7 @@ fn main() { ("acquire_batch, 2 duplicated", 9.5, 4_300.0), ("put, overwriting", 2.5, 2_200.0), ("put, evicting", 6.5, 20_500.0), - ("get, promoting from a lower tier", 15.0, 27_500.0), + ("get, promoting from a lower tier", 13.0, 25_500.0), ] { match find(path) { Some(row) if row.allocations <= max_allocations && row.bytes <= max_bytes => {} @@ -424,18 +459,39 @@ fn main() { // RocksDB before 16.11 58,193.6 // RocksDB after 12.11 55,951.6 // + // And again when framing a block stopped copying the payload into a buffer + // of its own first: + // + // backend allocations bytes + // Rust after 27.11 58,223.8 + // RocksDB after 11.11 54,914.2 + // // Per arm, because the two are not comparable: this allocator sees Rust // only, and with RocksDB the store's own bytes are allocated elsewhere. // Three runs of the same binary moved the figures by 0.01 allocations and // 8 bytes, so the margin here is wide rather than lucky. #[cfg(not(feature = "rocksdb-ssd"))] - const SSD_WRITE_ALLOCATIONS: f64 = 30.0; + const BATCH_PROMOTION_ALLOCATIONS: f64 = 65.0; + #[cfg(feature = "rocksdb-ssd")] + const BATCH_PROMOTION_ALLOCATIONS: f64 = 80.0; + match find("get_batch, promoting from a lower tier") { + Some(row) if row.allocations <= BATCH_PROMOTION_ALLOCATIONS && row.bytes <= 190_000.0 => {} + Some(row) => failures.push(format!( + "a batch of {BATCH} promoting reads allocates {:.2} times and {:.1} bytes, against \ + {BATCH_PROMOTION_ALLOCATIONS:.0} and 190,000", + row.allocations, row.bytes + )), + None => failures.push("the promoting batch read is missing from the table".to_string()), + } + #[cfg(not(feature = "rocksdb-ssd"))] - const SSD_WRITE_BYTES: f64 = 60_500.0; + const SSD_WRITE_ALLOCATIONS: f64 = 28.5; + #[cfg(not(feature = "rocksdb-ssd"))] + const SSD_WRITE_BYTES: f64 = 59_000.0; #[cfg(feature = "rocksdb-ssd")] - const SSD_WRITE_ALLOCATIONS: f64 = 14.0; + const SSD_WRITE_ALLOCATIONS: f64 = 12.5; #[cfg(feature = "rocksdb-ssd")] - const SSD_WRITE_BYTES: f64 = 57_000.0; + const SSD_WRITE_BYTES: f64 = 55_700.0; match find("put, reaching the SSD tier") { Some(row) if row.allocations <= SSD_WRITE_ALLOCATIONS && row.bytes <= SSD_WRITE_BYTES => {} Some(row) => failures.push(format!( @@ -475,9 +531,19 @@ fn main() { // a larger memory capacity, a placement change -- and the row would go on // reporting the cost of a memory hit under the promotion label. let promote_reads = REPEATS as u64 + 1; - if promote_disk_hits + 16 < promote_reads { + if promote_disk_hits * 10 < promote_reads * 9 { + failures.push(format!( + "the promoting row made {promote_reads} reads and only {promote_disk_hits} of them reached a lower tier, so it is not measuring a promotion" + )); + } + + // The same check for the batch row, over its own reads: eight keys asked + // for on each of two hundred and one calls, all but a handful of which + // have to come off the SSD tier. + let batch_promotion_reads = (REPEATS as u64 + 1) * BATCH as u64; + if batch_promotion_hits * 10 < batch_promotion_reads * 9 { failures.push(format!( - "the promoting row made {promote_reads} reads and only {promote_disk_hits} of them reached a lower tier, so it is not measuring a promotion" + "the promoting batch row asked for {batch_promotion_reads} keys and only {batch_promotion_hits} came off a lower tier" )); } diff --git a/src/runtime/builder_and_gc.rs b/src/runtime/builder_and_gc.rs index ece8ba5..25befb0 100644 --- a/src/runtime/builder_and_gc.rs +++ b/src/runtime/builder_and_gc.rs @@ -287,21 +287,34 @@ const CACHE_CODEC_RAW: u8 = 0; const CACHE_CODEC_ZSTD: u8 = 1; const CACHE_HEADER_LEN: usize = 8 + 1 + 1 + 8 + 8; +/// Frame a value as a block, with one copy of the payload. +/// +/// Every arm of this used to `value.to_vec()` the payload into an owned buffer +/// and then copy that buffer into the block -- two payload-sized allocations +/// and two copies per block written, where the block itself is the only buffer +/// anyone keeps. Three of the four arms store the value raw, so three of the +/// four were paying for a copy that existed only to be copied again. +/// +/// The compressed payload still needs a buffer of its own, because whether it +/// is used at all is decided by how big it turns out to be. The raw payload +/// does not: it is already in the caller's slice. fn encode_cache_block(value: &[u8], options: CacheBlockOptions) -> Result, CacheError> { - let (codec, payload) = match options.compression { - CacheCompression::None if value.len() >= options.min_compress_bytes => { - (CACHE_CODEC_RAW, value.to_vec()) - } - CacheCompression::None => (CACHE_CODEC_RAW, value.to_vec()), + let compressed = match options.compression { CacheCompression::Zstd { level } if value.len() >= options.min_compress_bytes => { let compressed = zstd::stream::encode_all(value, level)?; + // Kept only if framing it is smaller than storing the value as it + // is, which is the same comparison as before. if CACHE_HEADER_LEN + compressed.len() < value.len() { - (CACHE_CODEC_ZSTD, compressed) + Some(compressed) } else { - (CACHE_CODEC_RAW, value.to_vec()) + None } } - CacheCompression::Zstd { .. } => (CACHE_CODEC_RAW, value.to_vec()), + CacheCompression::Zstd { .. } | CacheCompression::None => None, + }; + let (codec, payload): (u8, &[u8]) = match compressed.as_deref() { + Some(compressed) => (CACHE_CODEC_ZSTD, compressed), + None => (CACHE_CODEC_RAW, value), }; let mut block = Vec::with_capacity(CACHE_HEADER_LEN + payload.len()); block.extend_from_slice(CACHE_BLOCK_MAGIC); @@ -309,7 +322,7 @@ fn encode_cache_block(value: &[u8], options: CacheBlockOptions) -> Result std::io::Result<()> { Ok(()) } +/// Decoded block bytes, in the form the read that asked for them needs. +/// +/// A read that refills hands the bytes to **two** holders: the tier it refills +/// and the caller. Decoding into the handle the tier wants costs one +/// allocation for the tier and one copy for the caller. +/// +/// A read that cannot refill -- an SSD-only instance, which never puts +/// anything back into memory -- has one holder, and a handle would have to be +/// copied again to hand back an owned value. That read decodes into the value +/// it is going to return. +/// +/// Both of those used to be the same thing: decode into a `Vec`, clone the +/// `Vec` for the tier, and let the tier build an `Arc` from the clone. Three +/// payload-sized allocations to serve two holders, and the extra one grew with +/// the payload. +enum CacheDecodedBlock { + /// Shared with the tier this read refills. + Shared(Arc<[u8]>), + /// Owned by the read, because nothing else wants these bytes. + Owned(Vec), +} + +impl CacheDecodedBlock { + fn decode(block: &[u8], refilling: bool) -> Result { + if refilling { + Ok(Self::Shared(decode_cache_block_shared(block)?)) + } else { + Ok(Self::Owned(decode_cache_block(block)?)) + } + } + + fn len(&self) -> usize { + match self { + Self::Shared(bytes) => bytes.len(), + Self::Owned(bytes) => bytes.len(), + } + } + + /// A handle to refill a tier with, or `None` from the form that has none. + /// + /// A read that asked for the owning form is a read that said it would not + /// refill, so there is nothing to hand over. + fn to_refill(&self) -> Option> { + match self { + Self::Shared(bytes) => Some(Arc::clone(bytes)), + Self::Owned(_) => None, + } + } + + /// One owned copy, for a read that needs several. + fn to_vec(&self) -> Vec { + match self { + Self::Shared(bytes) => bytes.to_vec(), + Self::Owned(bytes) => bytes.clone(), + } + } + + /// The owned copy, for a read that needs one. + /// + /// The owning form moves rather than copies here, which is why the two + /// forms are worth telling apart at all. + fn into_vec(self) -> Vec { + match self { + Self::Shared(bytes) => bytes.to_vec(), + Self::Owned(bytes) => bytes, + } + } +} + +/// Decode a block into a handle the tiers and the caller share. +/// +/// Seven read paths wrote `Arc::<[u8]>::from(decode_cache_block(..)?)`, which +/// for an uncompressed block is two payload-sized allocations and two copies: +/// one into the `Vec` the decode owns, one into the `Arc` that `Vec` is thrown +/// away for. An uncompressed payload is already a slice of the block, so the +/// handle can be built straight from it. +/// +/// A compressed payload still decompresses into a `Vec` first -- building the +/// handle in place without one would need the unsafe this crate forbids. +fn decode_cache_block_shared(block: &[u8]) -> Result, CacheError> { + if let Some(payload) = uncompressed_block_payload(block)? { + return Ok(Arc::<[u8]>::from(payload)); + } + Ok(Arc::<[u8]>::from(decode_cache_block(block)?)) +} + +/// The payload of a block that needs no decompressing, or `None` when it does. +/// +/// The header checks are the ones [`decode_cache_block`] makes, in the same +/// order, so a block either passes both or fails both. +fn uncompressed_block_payload(block: &[u8]) -> Result, CacheError> { + if !block.starts_with(CACHE_BLOCK_MAGIC) { + return Ok(Some(block)); + } + if block.len() < CACHE_HEADER_LEN { + return Err(CacheError::CorruptBlock("short header".to_string())); + } + let version = block[8]; + if version != CACHE_BLOCK_VERSION { + return Err(CacheError::CorruptBlock(format!( + "unsupported version {version}" + ))); + } + if block[9] != CACHE_CODEC_RAW { + return Ok(None); + } + let original_len = u64::from_le_bytes( + block[10..18] + .try_into() + .expect("cache block original length slice"), + ) as usize; + let payload_len = u64::from_le_bytes( + block[18..26] + .try_into() + .expect("cache block payload length slice"), + ) as usize; + if block.len() != CACHE_HEADER_LEN + payload_len { + return Err(CacheError::CorruptBlock( + "payload length mismatch".to_string(), + )); + } + let payload = &block[CACHE_HEADER_LEN..]; + if payload.len() != original_len { + return Err(CacheError::CorruptBlock( + "original length mismatch".to_string(), + )); + } + Ok(Some(payload)) +} + fn decode_cache_block(block: &[u8]) -> Result, CacheError> { if !block.starts_with(CACHE_BLOCK_MAGIC) { return Ok(block.to_vec()); @@ -2039,10 +2182,6 @@ impl CacheInner { self.put_ssd_bypass_if_absent(key, value) } - fn refill_from_ssd(&mut self, key: &CacheKey, value: Vec) -> bool { - self.refill_from_ssd_shared(key, Arc::<[u8]>::from(value)) - } - fn refill_from_ssd_shared(&mut self, key: &CacheKey, value: Arc<[u8]>) -> bool { if matches!( self.tiering_policy.data_placement, diff --git a/src/runtime/multilayer_cache.rs b/src/runtime/multilayer_cache.rs index ceaedcf..628c4e7 100644 --- a/src/runtime/multilayer_cache.rs +++ b/src/runtime/multilayer_cache.rs @@ -3417,7 +3417,7 @@ impl MultiLayerCache { // four payload-sized allocations, and a caller holding a // different copy from the one the cache kept. The whole point // of a shared read is that there is one. - let value = Arc::<[u8]>::from(decode_cache_block(&block)?); + let value = decode_cache_block_shared(&block)?; let mut inner = self.inner.write().expect("cache lock poisoned"); inner.read_counters.disk_hits.fetch_add(1, Ordering::Relaxed); inner @@ -3626,7 +3626,7 @@ impl MultiLayerCache { for ((unique_position, started), block) in ssd_candidates.into_iter().zip(blocks) { let decoded_block = match block { Some(block) => Some(( - Arc::<[u8]>::from(decode_cache_block(&block)?), + decode_cache_block_shared(&block)?, is_encoded_compressed_block(&block), )), None => None, @@ -4112,22 +4112,26 @@ impl MultiLayerCache { } let refill_started = Instant::now(); - let block = { + // The flag is read here rather than after the decode because it + // decides which decode is the cheap one, and this lock is already + // being taken. + let (block, refill_wanted) = { let inner = self.inner.read().expect("cache lock poisoned"); - inner.read_ssd_block(key)? + (inner.read_ssd_block(key)?, !inner.ssd_instance_only) }; match block { Some(block) => { - let decoded = decode_cache_block(&block)?; + let decoded = CacheDecodedBlock::decode(&block, refill_wanted)?; let mut inner = self.inner.write().expect("cache lock poisoned"); inner.read_counters.disk_hits.fetch_add(1, Ordering::Relaxed); if is_encoded_compressed_block(&block) { inner.stats.compressed_hits += 1; } inner.record_hit(key, decoded.len()); - if !inner.ssd_instance_only && !inner.refill_from_ssd(key, decoded.clone()) - { - inner.stats.refill_failures += 1; + if let Some(refill) = decoded.to_refill() { + if !inner.refill_from_ssd_shared(key, refill) { + inner.stats.refill_failures += 1; + } } inner.record_get_latency(started); inner.record_read_through_latency(started); @@ -4135,7 +4139,7 @@ impl MultiLayerCache { drop(inner); self.drain_eviction_records(); Ok(Some(CacheReadResult { - value: decoded, + value: decoded.into_vec(), tier: CacheReadTier::Ssd, })) } @@ -4332,15 +4336,21 @@ impl MultiLayerCache { .map(|(key, _, _)| *key) .collect::>(); let refill_started = Instant::now(); - let blocks = { + // The flag is read with the blocks rather than later because it + // decides which decode is the cheap one, and this lock is already + // being taken. + let (blocks, refill_wanted) = { let inner = self.inner.read().expect("cache lock poisoned"); - inner.read_ssd_blocks(&candidate_keys)? + ( + inner.read_ssd_blocks(&candidate_keys)?, + !inner.ssd_instance_only, + ) }; let mut ssd_reads = Vec::with_capacity(ssd_candidates.len()); for ((key, positions, started), block) in ssd_candidates.into_iter().zip(blocks) { let decoded = match block { Some(block) => Some(( - decode_cache_block(&block)?, + CacheDecodedBlock::decode(&block, refill_wanted)?, is_encoded_compressed_block(&block), )), None => None, @@ -4356,11 +4366,11 @@ impl MultiLayerCache { for (key, positions, started, refill_started, decoded) in ssd_reads { match decoded { Some((value, compressed)) => { - if !inner.ssd_instance_only - && !inner.refill_from_ssd(key, value.clone()) - { - inner.stats.refill_failures = - inner.stats.refill_failures.saturating_add(1); + if let Some(refill) = value.to_refill() { + if !inner.refill_from_ssd_shared(key, refill) { + inner.stats.refill_failures = + inner.stats.refill_failures.saturating_add(1); + } } let occurrences = positions.len(); inner @@ -4377,7 +4387,7 @@ impl MultiLayerCache { inner.record_read_through_latency_micros_many(micros, occurrences); inner.record_refill_latency(refill_started); for index in positions { - results[index] = Some(value.clone()); + results[index] = Some(value.to_vec()); } needs_eviction_drain = true; } @@ -4587,7 +4597,7 @@ impl MultiLayerCache { }; match block { Some(block) => { - let decoded = Arc::<[u8]>::from(decode_cache_block(&block)?); + let decoded = decode_cache_block_shared(&block)?; let mut inner = self.inner.write().expect("cache lock poisoned"); inner.increment_pin_for_handle(key, decoded.len()); if !inner.ssd_instance_only @@ -4681,7 +4691,7 @@ impl MultiLayerCache { let Some(block) = block else { return Ok(None); }; - let value = Arc::<[u8]>::from(decode_cache_block(&block)?); + let value = decode_cache_block_shared(&block)?; // Shared: taking the pin changes only the pin state, which has its own // lock. let inner = self.inner.read().expect("cache lock poisoned"); @@ -4789,7 +4799,7 @@ impl MultiLayerCache { let Some(block) = block else { continue; }; - decoded_hits.push((key, positions, Arc::<[u8]>::from(decode_cache_block(&block)?))); + decoded_hits.push((key, positions, decode_cache_block_shared(&block)?)); } if decoded_hits.is_empty() { return Ok(results); @@ -5048,7 +5058,7 @@ impl MultiLayerCache { for ((key, positions, started), block) in ssd_candidates.into_iter().zip(blocks) { let value = match block { Some(block) => Some(( - Arc::<[u8]>::from(decode_cache_block(&block)?), + decode_cache_block_shared(&block)?, is_encoded_compressed_block(&block), )), None => None, @@ -5479,7 +5489,7 @@ impl MultiLayerCache { } CacheTier::Ssd => match inner.read_ssd_block(key)? { Some(block) => { - let value = Arc::<[u8]>::from(decode_cache_block(&block)?); + let value = decode_cache_block_shared(&block)?; inner.increment_pin_with_size(key, value.len()); Ok(Some(CachePinnedHandle { key: key.clone(), diff --git a/src/tests/mod.rs b/src/tests/mod.rs index acfdb9d..449182d 100644 --- a/src/tests/mod.rs +++ b/src/tests/mod.rs @@ -24749,3 +24749,192 @@ fn a_restart_into_less_room_stays_inside_the_new_bound() { ); assert!(restarted.stop(), "the cache did not stop cleanly"); } + +/// The shared block decode agrees with the owning one, block for block. +/// +/// `decode_cache_block_shared` skips the `Vec` for an uncompressed block by +/// reading the header itself, which means the header checks exist twice. Two +/// copies of a validation drift: one gains a case, or loses one, and the +/// cache either rejects a block it used to serve or serves one it used to +/// reject -- quietly, on whichever of the two paths was not updated. +/// +/// So this asserts the only thing that matters about the duplicate: for every +/// block either shape of it can be handed, the two answer the same. Accepted +/// with the same bytes, or rejected -- and a rejection is checked by its +/// message, because `CorruptBlock` carries the reason and a decode that +/// rejects for the wrong reason has still diverged. +#[test] +fn the_shared_block_decode_answers_exactly_what_the_owning_one_does() { + let raw_options = CacheBlockOptions { + compression: CacheCompression::None, + min_compress_bytes: 128, + }; + let zstd_options = CacheBlockOptions { + compression: CacheCompression::Zstd { level: 1 }, + min_compress_bytes: 128, + }; + + let compressible = vec![7u8; 4096]; + let mut fill = 0x2545_f491_4f6c_dd1du64; + let incompressible: Vec = (0..4096) + .map(|_| { + fill ^= fill << 13; + fill ^= fill >> 7; + fill ^= fill << 17; + (fill & 0xff) as u8 + }) + .collect(); + + let compressed_block = + encode_cache_block(&compressible, zstd_options).expect("a compressible value must encode"); + assert!( + is_encoded_compressed_block(&compressed_block), + "the fixture has to produce a compressed block for the compressed case to be covered" + ); + let raw_block = + encode_cache_block(&incompressible, zstd_options).expect("a value must encode"); + assert!( + !is_encoded_compressed_block(&raw_block), + "an incompressible value has to come back raw, or the raw case is not covered" + ); + let empty_block = encode_cache_block(&[], raw_options).expect("an empty value must encode"); + let short_value_block = + encode_cache_block(&[1, 2, 3], raw_options).expect("a short value must encode"); + + let mut truncated_payload = raw_block.clone(); + truncated_payload.truncate(truncated_payload.len() - 1); + let mut bad_version = raw_block.clone(); + bad_version[8] = CACHE_BLOCK_VERSION.wrapping_add(1); + let mut unsupported_codec = raw_block.clone(); + unsupported_codec[9] = 99; + let mut lying_original_len = raw_block.clone(); + lying_original_len[10..18].copy_from_slice(&7u64.to_le_bytes()); + let mut lying_payload_len = raw_block.clone(); + lying_payload_len[18..26].copy_from_slice(&7u64.to_le_bytes()); + + let blocks: Vec<(&str, Vec)> = vec![ + ("a raw block", raw_block.clone()), + ("a compressed block", compressed_block), + ("an empty value", empty_block), + ("a value shorter than the header", short_value_block), + ("bytes with no magic at all", b"not a cache block".to_vec()), + ("no bytes at all", Vec::new()), + ("a header cut short", raw_block[..12].to_vec()), + ("a payload cut short", truncated_payload), + ("an unknown version", bad_version), + ("an unknown codec", unsupported_codec), + ("a lying original length", lying_original_len), + ("a lying payload length", lying_payload_len), + ]; + + for (what, block) in blocks { + let owning = decode_cache_block(&block); + let shared = decode_cache_block_shared(&block); + match (owning, shared) { + (Ok(owning), Ok(shared)) => assert_eq!( + owning.as_slice(), + shared.as_ref(), + "{what}: the two decodes accepted it with different bytes" + ), + (Err(owning), Err(shared)) => assert_eq!( + owning.to_string(), + shared.to_string(), + "{what}: the two decodes rejected it for different reasons" + ), + (Ok(owning), Err(shared)) => panic!( + "{what}: the owning decode accepted {} bytes and the shared one rejected it: {shared}", + owning.len() + ), + (Err(owning), Ok(shared)) => panic!( + "{what}: the shared decode accepted {} bytes and the owning one rejected it: {owning}", + shared.len() + ), + } + } +} + +/// A raw block is framed with one copy of the payload, not two. +/// +/// Every arm of `encode_cache_block` used to own its payload before the block +/// was built, so a raw block cost two payload-sized allocations and two +/// copies. The copy is not observable from outside, but the block is: this +/// asserts the framing is byte for byte what it was, which is what makes the +/// copy removable at all. A block on a device outlives the process that wrote +/// it, so this is a durable format. +#[test] +fn a_raw_block_is_framed_exactly_as_it_always_was() { + let options = CacheBlockOptions { + compression: CacheCompression::None, + min_compress_bytes: 128, + }; + let value: Vec = (0..300u32).map(|byte| (byte & 0xff) as u8).collect(); + let block = encode_cache_block(&value, options).expect("a value must encode"); + + assert_eq!(&block[..8], CACHE_BLOCK_MAGIC, "the magic moved"); + assert_eq!(block[8], CACHE_BLOCK_VERSION, "the version moved"); + assert_eq!(block[9], CACHE_CODEC_RAW, "the codec byte moved"); + assert_eq!( + u64::from_le_bytes(block[10..18].try_into().unwrap()), + value.len() as u64, + "the original length moved" + ); + assert_eq!( + u64::from_le_bytes(block[18..26].try_into().unwrap()), + value.len() as u64, + "the payload length moved" + ); + assert_eq!(&block[26..], value.as_slice(), "the payload moved"); + assert_eq!( + block.len(), + CACHE_HEADER_LEN + value.len(), + "the block grew" + ); + assert_eq!( + decode_cache_block(&block).expect("the block must decode"), + value, + "the block does not decode to what was written" + ); + + // And a compressed block, where the two lengths are **different**. + // + // Planted the two length fields written the other way round: every + // assertion above still passed, because a raw block stores the value as + // it is and its two lengths are equal by construction. A reader handed a + // compressed block with those fields swapped would size the decompression + // by the compressed length and reject the block. So the field that tells + // the two apart has to be checked on a block where they differ. + let compressible = vec![7u8; 4096]; + let compressed = encode_cache_block( + &compressible, + CacheBlockOptions { + compression: CacheCompression::Zstd { level: 1 }, + min_compress_bytes: 128, + }, + ) + .expect("a compressible value must encode"); + assert!( + is_encoded_compressed_block(&compressed), + "the fixture has to produce a compressed block for the two lengths to differ" + ); + let original_len = u64::from_le_bytes(compressed[10..18].try_into().unwrap()) as usize; + let payload_len = u64::from_le_bytes(compressed[18..26].try_into().unwrap()) as usize; + assert_eq!( + original_len, + compressible.len(), + "the first length is the value's, before compression" + ); + assert_eq!( + payload_len, + compressed.len() - CACHE_HEADER_LEN, + "the second length is the stored payload's" + ); + assert!( + payload_len < original_len, + "a compressed block that did not get smaller is not a compressed block" + ); + assert_eq!( + decode_cache_block(&compressed).expect("the block must decode"), + compressible, + "the compressed block does not decode to what was written" + ); +}