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" + ); +}