Repository: vercel/next.js
Pull request: [turbopack] Add support for fixed key blocks

## What?

Adds fixed-size key block types to turbo-persistence SST files. When all entries in a key block share the same key size and value type, the 4-byte-per-entry offset table is eliminated entirely. Entry positions become a direct arithmetic calculation: `header_size + index * stride`.

Also fixes a bug in `sst_inspect` where value blocks were misidentified as key blocks (value blocks have no type header, so their raw data could coincidentally match a key block type byte).

## Why?

Many turbo-persistence tables have uniform key and value sizes. For example, TaskCache stores 4-byte task ID keys with 4-byte inline values — every single entry is identical in structure. The existing variable-size key block format stores a 4-byte offset table entry per entry (1B type + 3B position) to support variable-length entries. For uniform blocks, this offset table is pure overhead in both space and read-path indirection (binary search must chase offset table pointers at each step).

## How?

**New block types 3 and 4** (parallel to existing types 1/2 for hash/no-hash):

```
Variable: [1B type][3B count][4B offset × count][entries...]
Fixed:    [1B type][3B count][1B key_size][1B value_type][entries...]
```

The 2 extra header bytes (key_size + value_type) are amortized across all entries since we save 4 bytes per entry from removing the offset table. Break-even at 1 entry.

**Writer changes:**
- `KeyBlockFormat` enum with state machine (`Unknown → Fixed → Variable`) tracks uniformity as entries are added to the accumulator
- `FixedKeyBlockBuilder` writes the compact 6-byte header + contiguous entries with no offset table
- Falls back to variable-size automatically when entries aren't uniform
- `ValueRef::write_value_to()` shared across both builder types to avoid duplication

**Reader changes:**
- `lookup_fixed_key_block()` — binary search using stride arithmetic (pure arithmetic, zero conditional branching per step)
- `get_fixed_key_entry()` — direct index calculation instead of offset table indirection
- Iterator refactored with `CurrentKeyBlockKind` enum (Variable vs Fixed variants)

**sst_inspect fix:** Reads the index block first to determine which block indices are key blocks, rather than guessing from the first byte of block data.

### Real-world impact (vercel-site .next cache, ~9.5M entries per family)

| Family | Fixed Blocks | Variable Blocks | Key Block Size | Notes |
|--------|-------------|----------------|---------------|-------|
| **TaskCache** | **16,274 (100%)** | 0 | **108.82 MB** (was 145 MB, **-25%**) | All inline 4B values |
| TaskMeta | 10,078 (72%) | 3,877 (28%) | 118.86 MB (was ~145 MB) | Variable blocks contain rare medium values |
| TaskData | 39 (0.3%) | 13,943 (99.7%) | 144.45 MB | Medium values spread across most blocks |
| Infra | 0 | 1 | 25 B | Mixed inline sizes |

This also saves 73M (2%) of the overall cache size

### Read-path benchmarks (`static_sorted_file_lookup`, 8B keys + 4B values)

| Benchmark | Canary | Fixed Blocks | Change |
|---|---|---|---|
| **1Ki hit/cached** | 534 ns | 510 ns | **-4.5%** |
| **10Ki hit/cached** | 559 ns | 556 ns | ~0% |
| **100Ki hit/cached** | 566 ns | 503 ns | **-11.1%** |
| **1Mi hit/cached** | 573 ns | 519 ns | **-9.4%** |
| 1Ki miss/cached | 571 ns | 571 ns | ~0% |
| 10Ki miss/cached | 613 ns | 556 ns | **-9.3%** |
| 100Ki miss/cached | 639 ns | 593 ns | **-7.2%** |
| 1Mi miss/cached | 791 ns | 702 ns | **-11.3%** |
| 1Ki hit/uncached | 4.07 µs | 3.81 µs | **-6.4%** |
| 10Ki hit/uncached | 5.33 µs | 5.03 µs | **-5.6%** |
| 100Ki hit/uncached | 5.55 µs | 5.24 µs | **-5.6%** |
| 1Mi hit/uncached | 8.47 µs | 8.31 µs | **-1.9%** |
| 1Ki miss/uncached | 3.37 µs | 3.19 µs | **-5.3%** |
| 10Ki miss/uncached | 3.87 µs | 3.41 µs | **-11.9%** |
| 100Ki miss/uncached | 4.15 µs | 3.70 µs | **-10.8%** |
| 1Mi miss/uncached | 6.98 µs | 6.66 µs | **-4.6%** |

Consistent **5-11% improvement** on cached lookups (the hot path in production), where the binary search dominates and offset table elimination matters most. Uncached lookups also improve 2-12% due to smaller blocks and faster binary search.

Changed files:
filename: turbopack/crates/turbo-persistence/README.md
patch: @@ -108,7 +108,7 @@ The hashes are sorted.
 
 `n` is `(block size + 1) / 10`
 
-#### Key Block
+#### Key Block (variable-size)
 
 - 1 byte block type (1: key block with hash, 2: key block without hash)
 - 3 bytes entry count
@@ -157,8 +157,20 @@ Depending on the `type` field entry has a different format:
 
 The entries are sorted by key hash and key.
 
-Future:
- * Some tables have fixed sized keys with consistent values (4 byte task ids).  We could optimize key block representation in this case by skipping offset tables.
+#### Key Block (fixed-size)
+
+Used when all entries in a block have the same key size and value type. Eliminates the per-entry offset table, enabling direct arithmetic indexing during binary search.
+
+- 1 byte block type (3: fixed-size with hash, 4: fixed-size without hash)
+- 3 bytes entry count
+- 1 byte key size (uniform across all entries)
+- 1 byte value type (shared by all entries, same encoding as variable-size type field)
+- foreach entry (packed at stride = hash_len + key_size + val_size):
+  - 8 bytes key hash (if block type 3)
+  - key data (key_size bytes)
+  - value data (size determined by value type)
+
+Entry position for index `i` is computed as `header_size + i * stride` with no indirection. The writer automatically selects fixed-size format when all entries in a block qualify; otherwise falls back to the variable-size format above.
 
 #### Value Block
 

filename: turbopack/crates/turbo-persistence/src/bin/sst_inspect.rs
patch: @@ -11,27 +11,29 @@
 //! - 8-255: Inline value where (type - 8) = value byte count
 
 use std::{
-    collections::BTreeMap,
+    collections::{BTreeMap, HashSet},
     fs::{self, File},
     path::{Path, PathBuf},
-    sync::Arc,
 };
 
 use anyhow::{Context, Result, bail};
 use byteorder::{BE, ReadBytesExt};
 use lzzzz::lz4::decompress;
 use memmap2::Mmap;
-// Import shared constants from the crate
-use turbo_persistence::static_sorted_file::{
-    BLOCK_TYPE_INDEX, BLOCK_TYPE_KEY_NO_HASH, BLOCK_TYPE_KEY_WITH_HASH, KEY_BLOCK_ENTRY_TYPE_BLOB,
-    KEY_BLOCK_ENTRY_TYPE_DELETED, KEY_BLOCK_ENTRY_TYPE_INLINE_MIN, KEY_BLOCK_ENTRY_TYPE_MEDIUM,
-    KEY_BLOCK_ENTRY_TYPE_SMALL,
-};
 use turbo_persistence::{
-    BLOCK_HEADER_SIZE, checksum_block, meta_file::MetaFile,
+    BLOCK_HEADER_SIZE, checksum_block,
+    meta_file::MetaFile,
     mmap_helper::advise_mmap_for_persistence,
+    static_sorted_file::{
+        BLOCK_TYPE_FIXED_KEY_NO_HASH, BLOCK_TYPE_FIXED_KEY_WITH_HASH, BLOCK_TYPE_KEY_NO_HASH,
+        BLOCK_TYPE_KEY_WITH_HASH, KEY_BLOCK_ENTRY_TYPE_BLOB, KEY_BLOCK_ENTRY_TYPE_DELETED,
+        KEY_BLOCK_ENTRY_TYPE_INLINE_MIN, KEY_BLOCK_ENTRY_TYPE_MEDIUM, KEY_BLOCK_ENTRY_TYPE_SMALL,
+    },
 };
 
+/// Size of the key block header (1B type + 3B entry count).
+const KEY_BLOCK_HEADER_SIZE: usize = 4;
+
 /// Block size information
 #[derive(Default, Debug, Clone)]
 struct BlockSizeInfo {
@@ -78,8 +80,12 @@ struct SstStats {
 
     /// Index block sizes
     index_blocks: BlockSizeInfo,
-    /// Key block sizes
+    /// Key block sizes (all types combined)
     key_blocks: BlockSizeInfo,
+    /// Variable-size key blocks (types 1/2)
+    variable_key_blocks: BlockSizeInfo,
+    /// Fixed-size key blocks (types 3/4)
+    fixed_key_blocks: BlockSizeInfo,
     /// Value block sizes (small values)
     value_blocks: BlockSizeInfo,
 
@@ -105,6 +111,8 @@ impl SstStats {
         self.total_entries += other.total_entries;
         self.index_blocks.merge(&other.index_blocks);
         self.key_blocks.merge(&other.key_blocks);
+        self.variable_key_blocks.merge(&other.variable_key_blocks);
+        self.fixed_key_blocks.merge(&other.fixed_key_blocks);
         self.value_blocks.merge(&other.value_blocks);
         self.block_directory_size += other.block_directory_size;
         self.inline_value_bytes += other.inline_value_bytes;
@@ -122,6 +130,32 @@ struct SstInfo {
     block_count: u16,
 }
 
+/// Accumulates statistics for a single entry of the given type.
+fn track_entry_type(stats: &mut SstStats, entry_type: u8) {
+    *stats.entry_type_counts.entry(entry_type).or_insert(0) += 1;
+    stats.total_entries += 1;
+
+    match entry_type {
+        KEY_BLOCK_ENTRY_TYPE_SMALL => {
+            stats.small_value_refs += 1;
+        }
+        KEY_BLOCK_ENTRY_TYPE_BLOB => {
+            stats.blob_refs += 1;
+        }
+        KEY_BLOCK_ENTRY_TYPE_DELETED => {
+            stats.deleted_count += 1;
+        }
+        KEY_BLOCK_ENTRY_TYPE_MEDIUM => {
+            stats.medium_value_refs += 1;
+        }
+        ty if ty >= KEY_BLOCK_ENTRY_TYPE_INLINE_MIN => {
+            let inline_size = (ty - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN) as u64;
+            stats.inline_value_bytes += inline_size;
+        }
+        _ => {}
+    }
+}
+
 fn entry_type_description(ty: u8) -> String {
     match ty {
         KEY_BLOCK_ENTRY_TYPE_SMALL => "small value (in value block)".to_string(),
@@ -208,21 +242,146 @@ fn collect_sst_info(db_path: &Path) -> Result<BTreeMap<u32, Vec<SstInfo>>> {
     Ok(family_sst_info)
 }
 
-/// Decompress a block, respecting the optional compression protocol.
-/// When uncompressed_length is 0, the block is stored uncompressed.
-fn decompress_block(compressed: &[u8], uncompressed_length: u32) -> Result<Arc<[u8]>> {
-    // Sentinel: uncompressed_length = 0 means block is stored uncompressed
-    if uncompressed_length == 0 {
-        return Ok(Arc::from(compressed));
+/// Information about a raw block read from disk.
+struct RawBlock {
+    data: Box<[u8]>,
+    compressed_size: u64,
+    actual_size: u64,
+    was_compressed: bool,
+}
+
+/// Reads, checksums, and decompresses a single block from the mmap.
+fn read_block(
+    mmap: &Mmap,
+    block_offsets_start: usize,
+    block_index: u16,
+    sequence_number: u32,
+) -> Result<RawBlock> {
+    let offset = block_offsets_start + block_index as usize * size_of::<u32>();
+
+    let block_start = if block_index == 0 {
+        0
+    } else {
+        (&mmap[offset - size_of::<u32>()..offset]).read_u32::<BE>()? as usize
+    };
+    let block_end = (&mmap[offset..offset + size_of::<u32>()]).read_u32::<BE>()? as usize;
+
+    let uncompressed_length =
+        (&mmap[block_start..block_start + size_of::<u32>()]).read_u32::<BE>()?;
+    let expected_checksum = (&mmap
+        [block_start + size_of::<u32>()..block_start + BLOCK_HEADER_SIZE])
+        .read_u32::<BE>()?;
+    let compressed_data = &mmap[block_start + BLOCK_HEADER_SIZE..block_end];
+    let compressed_size = compressed_data.len() as u64;
+
+    let was_compressed = uncompressed_length > 0;
+    let actual_size = if was_compressed {
+        uncompressed_length as u64
+    } else {
+        compressed_size
+    };
+
+    let actual_checksum = checksum_block(compressed_data);
+    if actual_checksum != expected_checksum {
+        bail!(
+            "Cache corruption detected: checksum mismatch in block {} of {:08}.sst (expected \
+             {:08x}, got {:08x})",
+            block_index,
+            sequence_number,
+            expected_checksum,
+            actual_checksum
+        );
     }
 
-    let mut buffer = vec![0u8; uncompressed_length as usize];
-    let bytes_written = decompress(compressed, &mut buffer)?;
-    assert_eq!(
-        bytes_written, uncompressed_length as usize,
-        "Decompressed length does not match expected"
-    );
-    Ok(Arc::from(buffer))
+    let data = if was_compressed {
+        let mut buffer = vec![0u8; uncompressed_length as usize];
+        let bytes_written = decompress(compressed_data, &mut buffer)?;
+        assert_eq!(
+            bytes_written, uncompressed_length as usize,
+            "Decompressed length does not match expected"
+        );
+        buffer.into_boxed_slice()
+    } else {
+        Box::from(compressed_data)
+    };
+
+    Ok(RawBlock {
+        data,
+        compressed_size,
+        actual_size,
+        was_compressed,
+    })
+}
+
+/// Parses an index block to extract all referenced key block indices.
+///
+/// Index block format: `[1B type][2B first_block][N * (8B hash + 2B block_index)]`.
+fn parse_key_block_indices(index_block: &[u8]) -> HashSet<u16> {
+    assert!(index_block.len() >= 4, "Index block too small");
+    let mut data = &index_block[1..]; // skip block type byte
+    let first_block = data.read_u16::<BE>().unwrap();
+    let mut indices = HashSet::new();
+    indices.insert(first_block);
+    const ENTRY_SIZE: usize = size_of::<u64>() + size_of::<u16>();
+    let entry_count = data.len() / ENTRY_SIZE;
+    for i in 0..entry_count {
+        let block_index = (&data[i * ENTRY_SIZE + 8..]).read_u16::<BE>().unwrap();
+        indices.insert(block_index);
+    }
+    indices
+}
+
+/// Parsed header of a key block.
+enum KeyBlockHeader {
+    Variable { entry_count: u32 },
+    Fixed { entry_count: u32, value_type: u8 },
+}
+
+/// Parses the header of a key block from the full decompressed block data.
+fn parse_key_block_header(block: &[u8]) -> Result<KeyBlockHeader> {
+    assert!(block.len() >= 4, "Key block too small");
+    let block_type = block[0];
+    let entry_count = ((block[1] as u32) << 16) | ((block[2] as u32) << 8) | (block[3] as u32);
+    match block_type {
+        BLOCK_TYPE_KEY_WITH_HASH | BLOCK_TYPE_KEY_NO_HASH => {
+            Ok(KeyBlockHeader::Variable { entry_count })
+        }
+        BLOCK_TYPE_FIXED_KEY_WITH_HASH | BLOCK_TYPE_FIXED_KEY_NO_HASH => {
+            assert!(block.len() >= 6, "Fixed key block header too small");
+            Ok(KeyBlockHeader::Fixed {
+                entry_count,
+                value_type: block[5],
+            })
+        }
+        _ => bail!("Invalid key block type: {block_type}"),
+    }
+}
+
+/// Iterates over entry type bytes in a key block.
+///
+/// For variable-size key blocks, reads byte 0 of each 4-byte offset table entry.
+/// For fixed-size key blocks, yields the single `value_type` repeated `entry_count` times.
+fn iter_key_block_entry_types(
+    header: KeyBlockHeader,
+    block: &[u8],
+) -> impl Iterator<Item = u8> + '_ {
+    let (entry_count, fixed_type) = match header {
+        KeyBlockHeader::Variable { entry_count } => (entry_count, None),
+        KeyBlockHeader::Fixed {
+            entry_count,
+            value_type,
+        } => (entry_count, Some(value_type)),
+    };
+    (0..entry_count).map(move |i| {
+        if let Some(vt) = fixed_type {
+            vt
+        } else {
+            // Variable block: offset table starts at byte 4 (after 1B type + 3B count),
+            // each entry is 4 bytes, first byte is the entry type.
+            let header_offset = KEY_BLOCK_HEADER_SIZE + i as usize * 4;
+            block[header_offset]
+        }
+    })
 }
 
 /// Analyze an SST file and return entry type statistics
@@ -236,160 +395,88 @@ fn analyze_sst_file(db_path: &Path, info: &SstInfo) -> Result<SstStats> {
     advise_mmap_for_persistence(&mmap)?;
 
     let mut stats = SstStats {
-        block_directory_size: info.block_count as u64 * 4,
+        block_directory_size: info.block_count as u64 * size_of::<u32>() as u64,
         file_size,
         ..Default::default()
     };
 
-    // Calculate offsets
-    let block_offsets_start = mmap.len() - (info.block_count as usize * 4);
-    let blocks_start = 0;
-
-    // Iterate through all blocks
-    for block_index in 0..info.block_count {
-        let offset = block_offsets_start + block_index as usize * 4;
-
-        let block_start = if block_index == 0 {
-            blocks_start
-        } else {
-            blocks_start + (&mmap[offset - 4..offset]).read_u32::<BE>()? as usize
-        };
-        let block_end = blocks_start + (&mmap[offset..offset + 4]).read_u32::<BE>()? as usize;
-
-        // Read block header (uncompressed length + checksum) and block data
-        let uncompressed_length = (&mmap[block_start..block_start + 4]).read_u32::<BE>()?;
-        let expected_checksum = (&mmap[block_start + 4..block_start + 8]).read_u32::<BE>()?;
-        let compressed_data = &mmap[block_start + BLOCK_HEADER_SIZE..block_end];
-        let compressed_size = compressed_data.len() as u64;
-
-        // Determine if block was compressed (uncompressed_length > 0 means it was compressed)
-        let was_compressed = uncompressed_length > 0;
-        // Actual size: if uncompressed_length is 0, use stored size (block wasn't compressed)
-        let actual_size = if uncompressed_length == 0 {
-            compressed_size
-        } else {
-            uncompressed_length as u64
-        };
-
-        // Verify checksum on the raw on-disk data before decompression.
-        let actual_checksum = checksum_block(compressed_data);
-        if actual_checksum != expected_checksum {
-            bail!(
-                "Cache corruption detected: checksum mismatch in block {} of {:08}.sst (expected \
-                 {:08x}, got {:08x})",
-                block_index,
-                info.sequence_number,
-                expected_checksum,
-                actual_checksum
-            );
-        }
+    let block_offsets_start = mmap.len() - (info.block_count as usize * size_of::<u32>());
+
+    // Read the index block (always the last block) first to learn which blocks are key blocks.
+    // Without this, we'd have to guess block types from their first byte, which is wrong for
+    // value blocks (they have no type header and their data can start with any byte).
+    let index_block_index = info.block_count - 1;
+    let index_raw = read_block(
+        &mmap,
+        block_offsets_start,
+        index_block_index,
+        info.sequence_number,
+    )?;
+    let key_block_indices = parse_key_block_indices(&index_raw.data);
+
+    stats.index_blocks.add(
+        index_raw.compressed_size,
+        index_raw.actual_size,
+        index_raw.was_compressed,
+    );
 
-        let decompressed = match decompress_block(compressed_data, uncompressed_length) {
-            Ok(data) => data,
+    // Now iterate through all blocks, using the key block set for classification.
+    for block_index in 0..index_block_index {
+        let raw = match read_block(
+            &mmap,
+            block_offsets_start,
+            block_index,
+            info.sequence_number,
+        ) {
+            Ok(raw) => raw,
             Err(e) => {
                 eprintln!(
-                    "Warning: Failed to decompress block {} in {:08}.sst: {}",
+                    "Warning: Failed to read block {} in {:08}.sst: {}",
                     block_index, info.sequence_number, e
                 );
                 continue;
             }
         };
 
-        let block = &decompressed[..];
-        if block.is_empty() {
+        if !key_block_indices.contains(&block_index) {
+            // Value block — no type header, just raw data.
+            stats
+                .value_blocks
+                .add(raw.compressed_size, raw.actual_size, raw.was_compressed);
             continue;
         }
 
-        let block_type = block[0];
-
-        // The index block is always the LAST block in the file
-        let is_last_block = block_index == info.block_count - 1;
-
-        match block_type {
-            BLOCK_TYPE_INDEX if is_last_block => {
-                // Validate index block structure: 1 byte type + 2 byte first_block + N*(10 bytes)
-                let content_len = block.len() - 3; // subtract header
-                if content_len % 10 == 0 {
-                    stats
-                        .index_blocks
-                        .add(compressed_size, actual_size, was_compressed);
-                } else {
-                    // Invalid structure, treat as value block
-                    stats
-                        .value_blocks
-                        .add(compressed_size, actual_size, was_compressed);
-                }
-            }
-            BLOCK_TYPE_KEY_WITH_HASH | BLOCK_TYPE_KEY_NO_HASH => {
-                // Key block - extract entry types
-                if block.len() < 4 {
-                    // Too small to be a valid key block, likely garbage from wrong decompression
-                    stats
-                        .value_blocks
-                        .add(compressed_size, actual_size, was_compressed);
-                    continue;
-                }
-
-                // Entry count is stored as 3 bytes after the block type
-                let entry_count =
-                    ((block[1] as u32) << 16) | ((block[2] as u32) << 8) | (block[3] as u32);
-
-                // Validate entry count - if it's unreasonably large or the block is too small
-                // to contain the headers, this is likely garbage from wrong decompression
-                let expected_header_size = 4 + entry_count as usize * 4;
-                if entry_count == 0 || entry_count > 100_000 || expected_header_size > block.len() {
-                    // Invalid key block structure, treat as value block
-                    stats
-                        .value_blocks
-                        .add(compressed_size, actual_size, was_compressed);
-                    continue;
-                }
-
-                stats
-                    .key_blocks
-                    .add(compressed_size, actual_size, was_compressed);
-
-                // Entry headers start at offset 4
-                // Each entry header is 4 bytes: 1 byte type + 3 bytes position
-                for i in 0..entry_count as usize {
-                    let header_offset = 4 + i * 4;
-                    if header_offset >= block.len() {
-                        break;
-                    }
-                    let entry_type = block[header_offset];
-
-                    *stats.entry_type_counts.entry(entry_type).or_insert(0) += 1;
-                    stats.total_entries += 1;
-
-                    // Track value statistics
-                    match entry_type {
-                        KEY_BLOCK_ENTRY_TYPE_SMALL => {
-                            stats.small_value_refs += 1;
-                        }
-                        KEY_BLOCK_ENTRY_TYPE_BLOB => {
-                            stats.blob_refs += 1;
-                        }
-                        KEY_BLOCK_ENTRY_TYPE_DELETED => {
-                            stats.deleted_count += 1;
-                        }
-                        KEY_BLOCK_ENTRY_TYPE_MEDIUM => {
-                            stats.medium_value_refs += 1;
-                        }
-                        ty if ty >= KEY_BLOCK_ENTRY_TYPE_INLINE_MIN => {
-                            let inline_size = (ty - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN) as u64;
-                            stats.inline_value_bytes += inline_size;
-                        }
-                        _ => {}
-                    }
-                }
+        let block: &[u8] = &raw.data;
+
+        stats
+            .key_blocks
+            .add(raw.compressed_size, raw.actual_size, raw.was_compressed);
+
+        let key_block_header = parse_key_block_header(block).with_context(|| {
+            format!(
+                "Warning: key block {} in {:08}.sst has unexpected block type {}",
+                block_index, info.sequence_number, block[0]
+            )
+        })?;
+        match key_block_header {
+            KeyBlockHeader::Variable { .. } => {
+                stats.variable_key_blocks.add(
+                    raw.compressed_size,
+                    raw.actual_size,
+                    raw.was_compressed,
+                );
             }
-            _ => {
-                // Unknown block type - might be a value block that happened to decompress with dict
-                // Try to identify it as a value block
-                stats
-                    .value_blocks
-                    .add(compressed_size, actual_size, was_compressed);
+            KeyBlockHeader::Fixed { .. } => {
+                stats.fixed_key_blocks.add(
+                    raw.compressed_size,
+                    raw.actual_size,
+                    raw.was_compressed,
+                );
             }
+        };
+
+        for entry_type in iter_key_block_entry_types(key_block_header, block) {
+            track_entry_type(&mut stats, entry_type);
         }
     }
 
@@ -567,6 +654,14 @@ fn print_sst_details(seq_num: u32, stats: &SstStats) {
     print_block_stats("Index blocks", &stats.index_blocks);
     print!("  │   ");
     print_block_stats("Key blocks", &stats.key_blocks);
+    if stats.variable_key_blocks.total_count() > 0 && stats.fixed_key_blocks.total_count() > 0 {
+        print!("  │       ");
+        print_block_stats("Variable", &stats.variable_key_blocks);
+        print!("  │       ");
+        print_block_stats("Fixed", &stats.fixed_key_blocks);
+    } else if stats.fixed_key_blocks.total_count() > 0 {
+        println!("  │       (all fixed-size)");
+    }
     print!("  │   ");
     print_block_stats("Value blocks", &stats.value_blocks);
 
@@ -643,6 +738,15 @@ fn print_family_summary(family: u32, sst_count: usize, stats: &SstStats) {
     print_block_stats("Index blocks", &stats.index_blocks);
     print!("  ");
     print_block_stats("Key blocks", &stats.key_blocks);
+    if stats.variable_key_blocks.total_count() > 0 && stats.fixed_key_blocks.total_count() > 0 {
+        // Only show breakdown when both types are present
+        print!("      ");
+        print_block_stats("Variable", &stats.variable_key_blocks);
+        print!("      ");
+        print_block_stats("Fixed", &stats.fixed_key_blocks);
+    } else if stats.fixed_key_blocks.total_count() > 0 {
+        println!("      (all fixed-size)");
+    }
     print!("  ");
     print_block_stats("Value blocks", &stats.value_blocks);
 

filename: turbopack/crates/turbo-persistence/src/static_sorted_file.rs
patch: @@ -29,6 +29,10 @@ pub const BLOCK_TYPE_INDEX: u8 = 0;
 pub const BLOCK_TYPE_KEY_WITH_HASH: u8 = 1;
 /// The block header for a key block without hash.
 pub const BLOCK_TYPE_KEY_NO_HASH: u8 = 2;
+/// The block header for a fixed-size key block with 8-byte hash per entry.
+pub const BLOCK_TYPE_FIXED_KEY_WITH_HASH: u8 = 3;
+/// The block header for a fixed-size key block without hash.
+pub const BLOCK_TYPE_FIXED_KEY_NO_HASH: u8 = 4;
 
 /// The tag for a small-sized value.
 pub const KEY_BLOCK_ENTRY_TYPE_SMALL: u8 = 0;
@@ -41,6 +45,15 @@ pub const KEY_BLOCK_ENTRY_TYPE_MEDIUM: u8 = 3;
 /// The minimum tag for inline values. The actual size is (tag - INLINE_MIN).
 pub const KEY_BLOCK_ENTRY_TYPE_INLINE_MIN: u8 = 8;
 
+/// Encoded size of a small value reference: 2B block index + 2B size + 4B offset.
+pub(crate) const SMALL_VALUE_REF_SIZE: usize = 8;
+/// Encoded size of a medium value reference: 2B block index.
+pub(crate) const MEDIUM_VALUE_REF_SIZE: usize = 2;
+/// Encoded size of a blob value reference: 4B blob id.
+pub(crate) const BLOB_VALUE_REF_SIZE: usize = 4;
+/// Encoded size of a deleted (tombstone) value reference.
+pub(crate) const DELETED_VALUE_REF_SIZE: usize = 0;
+
 // Static assertion: MAX_INLINE_VALUE_SIZE must fit in the key type encoding.
 // Key types 8-255 encode inline values of size 0-247, so max is 255 - 8 = 247.
 const _: () = assert!(
@@ -239,6 +252,16 @@ impl StaticSortedFile {
                         value_block_cache,
                     );
                 }
+                BLOCK_TYPE_FIXED_KEY_WITH_HASH | BLOCK_TYPE_FIXED_KEY_NO_HASH => {
+                    let has_hash = block_type == BLOCK_TYPE_FIXED_KEY_WITH_HASH;
+                    return self.lookup_fixed_key_block::<K, FIND_ALL>(
+                        key_block_arc,
+                        key_hash,
+                        key,
+                        has_hash,
+                        value_block_cache,
+                    );
+                }
                 _ => {
                     bail!("Invalid block type");
                 }
@@ -309,6 +332,63 @@ impl StaticSortedFile {
         let offsets = &block[..entry_count * 4];
         let entries = &block[entry_count * 4..];
 
+        self.lookup_block_inner::<K, FIND_ALL>(
+            &block,
+            entry_count,
+            key_hash,
+            key,
+            value_block_cache,
+            |i| get_key_entry(offsets, entries, entry_count, i, hash_len),
+        )
+    }
+
+    /// Looks up a key in a fixed-size key block.
+    ///
+    /// Fixed-size key blocks store entries at predictable offsets (no offset table),
+    /// enabling direct indexing during binary search.
+    fn lookup_fixed_key_block<K: QueryKey, const FIND_ALL: bool>(
+        &self,
+        mut block: ArcBytes,
+        key_hash: u64,
+        key: &K,
+        has_hash: bool,
+        value_block_cache: &BlockCache,
+    ) -> Result<SstLookupResult> {
+        let hash_len: u8 = if has_hash { 8 } else { 0 };
+        let entry_count = block.read_u24::<BE>()? as usize;
+        let key_size = block.read_u8()? as usize;
+        let value_type = block.read_u8()?;
+        let val_size = entry_val_size(value_type)?;
+        let stride = hash_len as usize + key_size + val_size;
+        let entries = &block[..];
+
+        self.lookup_block_inner::<K, FIND_ALL>(
+            &block,
+            entry_count,
+            key_hash,
+            key,
+            value_block_cache,
+            |i| {
+                Ok(get_fixed_key_entry(
+                    entries, i, hash_len, key_size, value_type, stride,
+                ))
+            },
+        )
+    }
+
+    /// Shared binary search + collection logic for both key block variants.
+    ///
+    /// The `get_entry` closure abstracts over the difference between variable-size
+    /// key blocks (offset table lookup) and fixed-size key blocks (stride-based indexing).
+    fn lookup_block_inner<'a, K: QueryKey, const FIND_ALL: bool>(
+        &self,
+        block: &ArcBytes,
+        entry_count: usize,
+        key_hash: u64,
+        key: &K,
+        value_block_cache: &BlockCache,
+        get_entry: impl Fn(usize) -> Result<GetKeyEntryResult<'a>>,
+    ) -> Result<SstLookupResult> {
         let mut l = 0;
         let mut r = entry_count;
         // binary search for a matching key
@@ -319,7 +399,7 @@ impl StaticSortedFile {
                 key: mid_key,
                 ty,
                 val,
-            } = get_key_entry(offsets, entries, entry_count, m, hash_len)?;
+            } = get_entry(m)?;
 
             let comparison = compare_hash_key(mid_hash, mid_key, key_hash, key);
 
@@ -329,7 +409,7 @@ impl StaticSortedFile {
                     if !FIND_ALL {
                         // SingleValue mode: each key has exactly one entry
                         // this is enforced when writing
-                        let result = self.handle_key_match(ty, val, &block, value_block_cache)?;
+                        let result = self.handle_key_match(ty, val, block, value_block_cache)?;
                         return Ok(SstLookupResult::Found(SmallVec::from_buf([result])));
                     }
                     // FIND_ALL (MultiValue) mode: collect all values for this key.
@@ -345,32 +425,33 @@ impl StaticSortedFile {
                             key: entry_key,
                             ty,
                             val,
-                        } = get_key_entry(offsets, entries, entry_count, i, hash_len)?;
+                        } = get_entry(i)?;
                         if compare_hash_key(hash, entry_key, key_hash, key) != Ordering::Equal {
                             break;
                         }
-                        results.push(self.handle_key_match(ty, val, &block, value_block_cache)?);
+                        results.push(self.handle_key_match(ty, val, block, value_block_cache)?);
                     }
-                    // Technically we could `.reverse()` the items collected by the backwards scan,
-                    // but the only ordering constraint we need to maintain for single sst
-                    // multivalue reads is that a deleted token, if it exists comes last.  Because
-                    // all the backwards scan items are strictly before the found item we know they
-                    // don't contain the _last_ item. So we don't care about their order
+                    // Technically we could `.reverse()` the items collected by the backwards
+                    // scan, but the only ordering constraint we need to maintain for single
+                    // sst multivalue reads is that a deleted token, if it exists comes last.
+                    // Because all the backwards scan items are strictly before the found item
+                    // we know they don't contain the _last_ item. So we don't care about
+                    // their order.
 
                     // Add the entry at `m`
-                    results.push(self.handle_key_match(ty, val, &block, value_block_cache)?);
+                    results.push(self.handle_key_match(ty, val, block, value_block_cache)?);
                     // Forward scan: collect remaining entries with the same key
                     for i in (m + 1)..entry_count {
                         let GetKeyEntryResult {
                             hash,
                             key: entry_key,
                             ty,
                             val,
-                        } = get_key_entry(offsets, entries, entry_count, i, hash_len)?;
+                        } = get_entry(i)?;
                         if compare_hash_key(hash, entry_key, key_hash, key) != Ordering::Equal {
                             break;
                         }
-                        results.push(self.handle_key_match(ty, val, &block, value_block_cache)?);
+                        results.push(self.handle_key_match(ty, val, block, value_block_cache)?);
                     }
                     return Ok(SstLookupResult::Found(results));
                 }
@@ -623,12 +704,23 @@ pub struct StaticSortedFileIter {
     value_block_cache: Option<(u16, ArcBytes)>,
 }
 
+enum CurrentKeyBlockKind {
+    /// Variable-size entries with an offset table for random access.
+    Variable { offsets: ArcBytes, hash_len: u8 },
+    /// Fixed-size entries with uniform key size and value type (no offset table).
+    Fixed {
+        hash_len: u8,
+        key_size: usize,
+        value_type: u8,
+        stride: usize,
+    },
+}
+
 struct CurrentKeyBlock {
-    offsets: ArcBytes,
+    kind: CurrentKeyBlockKind,
     entries: ArcBytes,
     entry_count: usize,
     index: usize,
-    hash_len: u8,
 }
 
 struct CurrentIndexBlock {
@@ -670,11 +762,33 @@ impl StaticSortedFileIter {
                 let offsets = block_arc.clone().slice(offsets_range);
                 let entries = block_arc.slice(entries_range);
                 self.current_key_block = Some(CurrentKeyBlock {
-                    offsets,
+                    kind: CurrentKeyBlockKind::Variable { offsets, hash_len },
+                    entries,
+                    entry_count,
+                    index: 0,
+                });
+            }
+            BLOCK_TYPE_FIXED_KEY_WITH_HASH | BLOCK_TYPE_FIXED_KEY_NO_HASH => {
+                let has_hash = block_type == BLOCK_TYPE_FIXED_KEY_WITH_HASH;
+                let hash_len = if has_hash { 8 } else { 0 };
+                let entry_count = block.read_u24::<BE>()? as usize;
+                let key_size = block.read_u8()? as usize;
+                let value_type = block.read_u8()?;
+                let val_size = entry_val_size(value_type)?;
+                let stride = hash_len as usize + key_size + val_size;
+                // Header is 6 bytes for fixed-size blocks
+                let entries_range = 6..block_arc.len();
+                let entries = block_arc.slice(entries_range);
+                self.current_key_block = Some(CurrentKeyBlock {
+                    kind: CurrentKeyBlockKind::Fixed {
+                        hash_len,
+                        key_size,
+                        value_type,
+                        stride,
+                    },
                     entries,
                     entry_count,
                     index: 0,
-                    hash_len,
                 });
             }
             _ => {
@@ -688,15 +802,30 @@ impl StaticSortedFileIter {
     fn next_internal(&mut self) -> Result<Option<LookupEntry>> {
         loop {
             if let Some(CurrentKeyBlock {
-                offsets,
+                kind,
                 entries,
                 entry_count,
                 index,
-                hash_len,
             }) = self.current_key_block.take()
             {
-                let GetKeyEntryResult { hash, key, ty, val } =
-                    get_key_entry(&offsets, &entries, entry_count, index, hash_len)?;
+                let GetKeyEntryResult { hash, key, ty, val } = match &kind {
+                    CurrentKeyBlockKind::Variable { offsets, hash_len } => {
+                        get_key_entry(offsets, &entries, entry_count, index, *hash_len)?
+                    }
+                    CurrentKeyBlockKind::Fixed {
+                        hash_len,
+                        key_size,
+                        value_type,
+                        stride,
+                    } => get_fixed_key_entry(
+                        &entries,
+                        index,
+                        *hash_len,
+                        *key_size,
+                        *value_type,
+                        *stride,
+                    ),
+                };
                 // Convert hash slice to u64, computing from key if no hash stored
                 let full_hash = if hash.is_empty() {
                     crate::key::hash_key(&key)
@@ -729,11 +858,10 @@ impl StaticSortedFileIter {
                 };
                 if index + 1 < entry_count {
                     self.current_key_block = Some(CurrentKeyBlock {
-                        offsets,
+                        kind,
                         entries,
                         entry_count,
                         index: index + 1,
-                        hash_len,
                     });
                 }
                 return Ok(Some(entry));
@@ -796,14 +924,14 @@ fn compare_hash_key<K: QueryKey>(
 /// Returns the byte size of the value portion for a given key block entry type.
 fn entry_val_size(ty: u8) -> Result<usize> {
     match ty {
-        KEY_BLOCK_ENTRY_TYPE_SMALL => Ok(8), // 2 bytes block index, 2 bytes size, 4 bytes position
-        KEY_BLOCK_ENTRY_TYPE_MEDIUM => Ok(2), // 2 bytes block index
-        KEY_BLOCK_ENTRY_TYPE_BLOB => Ok(4),  // 4 byte blob id
-        KEY_BLOCK_ENTRY_TYPE_DELETED => Ok(0), // no value
+        KEY_BLOCK_ENTRY_TYPE_SMALL => Ok(SMALL_VALUE_REF_SIZE),
+        KEY_BLOCK_ENTRY_TYPE_MEDIUM => Ok(MEDIUM_VALUE_REF_SIZE),
+        KEY_BLOCK_ENTRY_TYPE_BLOB => Ok(BLOB_VALUE_REF_SIZE),
+        KEY_BLOCK_ENTRY_TYPE_DELETED => Ok(DELETED_VALUE_REF_SIZE),
         ty if ty >= KEY_BLOCK_ENTRY_TYPE_INLINE_MIN => {
             Ok((ty - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN) as usize)
         }
-        _ => bail!("Invalid key block entry type"),
+        _ => bail!("Invalid key block entry type: {ty}"),
     }
 }
 
@@ -834,3 +962,25 @@ fn get_key_entry<'l>(
         val: &entries[end - val_size..end],
     })
 }
+
+/// Reads a key entry from a fixed-size key block by direct indexing.
+///
+/// All entries have the same key size and value type, so positions are computed
+/// arithmetically with no offset table indirection.
+fn get_fixed_key_entry<'l>(
+    entries: &'l [u8],
+    index: usize,
+    hash_len: u8,
+    key_size: usize,
+    value_type: u8,
+    stride: usize,
+) -> GetKeyEntryResult<'l> {
+    let hash_len_usize = hash_len as usize;
+    let start = index * stride;
+    GetKeyEntryResult {
+        hash: &entries[start..start + hash_len_usize],
+        key: &entries[start + hash_len_usize..start + hash_len_usize + key_size],
+        ty: value_type,
+        val: &entries[start + hash_len_usize + key_size..(index + 1) * stride],
+    }
+}

filename: turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs
patch: @@ -15,9 +15,11 @@ use crate::{
     constants::{MAX_INLINE_VALUE_SIZE, MAX_SMALL_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE},
     meta_file::{AmqfBincodeWrapper, MetaEntryFlags},
     static_sorted_file::{
-        BLOCK_TYPE_INDEX, BLOCK_TYPE_KEY_NO_HASH, BLOCK_TYPE_KEY_WITH_HASH,
+        BLOB_VALUE_REF_SIZE, BLOCK_TYPE_FIXED_KEY_NO_HASH, BLOCK_TYPE_FIXED_KEY_WITH_HASH,
+        BLOCK_TYPE_INDEX, BLOCK_TYPE_KEY_NO_HASH, BLOCK_TYPE_KEY_WITH_HASH, DELETED_VALUE_REF_SIZE,
         KEY_BLOCK_ENTRY_TYPE_BLOB, KEY_BLOCK_ENTRY_TYPE_DELETED, KEY_BLOCK_ENTRY_TYPE_INLINE_MIN,
-        KEY_BLOCK_ENTRY_TYPE_MEDIUM, KEY_BLOCK_ENTRY_TYPE_SMALL,
+        KEY_BLOCK_ENTRY_TYPE_MEDIUM, KEY_BLOCK_ENTRY_TYPE_SMALL, MEDIUM_VALUE_REF_SIZE,
+        SMALL_VALUE_REF_SIZE,
     },
 };
 
@@ -51,6 +53,72 @@ const AVG_SMALL_VALUE_SIZE: usize = 64;
 /// byte-size based estimates of pending key blocks.
 const BLOCK_INDEX_CAPACITY_BUFFER: usize = 16;
 
+/// Maximum key length that can use fixed-size key block layout.
+///
+/// The on-disk fixed-key header stores the key size as a single byte, so keys longer than this
+/// fall back to variable-size layout.
+const MAX_FIXED_KEY_LEN: usize = u8::MAX as usize;
+
+/// Newtype for the key block entry type byte.
+///
+/// This encodes what kind of value reference an entry has (small, medium, blob, deleted, or
+/// inline with embedded length). See `KEY_BLOCK_ENTRY_TYPE_*` constants.
+#[derive(Clone, Copy, PartialEq, Eq, Debug)]
+struct EntryType(u8);
+
+/// Tracks whether a key block's entries are uniform enough for fixed-size layout.
+///
+/// State transitions:
+/// - `Unknown` → first entry → `Fixed { key_len, value_type }`
+/// - `Fixed` + matching entry → stays `Fixed`
+/// - `Fixed` + mismatched key_len or value_type → `Variable`
+/// - `Variable` → stays `Variable`
+#[derive(Clone, Copy)]
+enum KeyBlockFormat {
+    /// No entries yet — format undetermined.
+    Unknown,
+    /// All entries so far have uniform key length and value type.
+    Fixed { key_len: u8, value_type: EntryType },
+    /// Entries have mixed key lengths or value types; must use offset table.
+    Variable,
+}
+
+impl KeyBlockFormat {
+    /// Updates the format after seeing an entry with the given key length and value type.
+    ///
+    /// A `Fixed` state is only reachable when all entries have matching key length and value type,
+    /// and the key length fits in a u8 (required by the on-disk header).
+    fn update(&mut self, key_len: usize, value_type: EntryType) {
+        *self = match *self {
+            KeyBlockFormat::Unknown => {
+                if key_len <= MAX_FIXED_KEY_LEN {
+                    KeyBlockFormat::Fixed {
+                        key_len: key_len as u8,
+                        value_type,
+                    }
+                } else {
+                    KeyBlockFormat::Variable
+                }
+            }
+            KeyBlockFormat::Fixed {
+                key_len: k,
+                value_type: v,
+            } if k as usize == key_len && v == value_type => KeyBlockFormat::Fixed {
+                key_len: k,
+                value_type: v,
+            },
+            KeyBlockFormat::Fixed { .. } | KeyBlockFormat::Variable => KeyBlockFormat::Variable,
+        };
+    }
+}
+
+/// Copy-able snapshot of the accumulator state needed by [`flush_key_block`].
+#[derive(Clone, Copy)]
+struct KeyBlockFlushInfo {
+    max_key_len: usize,
+    format: KeyBlockFormat,
+}
+
 /// Tracks the accumulated state of the current incomplete key block.
 ///
 /// During streaming, this sits on [`StreamingSstWriter`] and tracks the tail of the resolved
@@ -66,6 +134,8 @@ struct KeyBlockAccumulator {
     /// Hash of the most recently added entry (used to avoid splitting entries with equal hashes
     /// across blocks).
     last_hash: u64,
+    /// Whether the block qualifies for fixed-size layout.
+    format: KeyBlockFormat,
 }
 
 impl KeyBlockAccumulator {
@@ -75,15 +145,25 @@ impl KeyBlockAccumulator {
             entry_count: 0,
             max_key_len: 0,
             last_hash: 0,
+            format: KeyBlockFormat::Unknown,
         }
     }
 
     /// Records a new entry in the accumulator.
-    fn add(&mut self, key_len: usize, key_hash: u64) {
+    fn add(&mut self, key_len: usize, key_hash: u64, value_type: EntryType) {
         self.size += key_len + KEY_BLOCK_ENTRY_META_OVERHEAD;
         self.max_key_len = self.max_key_len.max(key_len);
         self.entry_count += 1;
         self.last_hash = key_hash;
+        self.format.update(key_len, value_type);
+    }
+
+    /// Snapshots the state needed by `flush_key_block`.
+    fn flush_info(&self) -> KeyBlockFlushInfo {
+        KeyBlockFlushInfo {
+            max_key_len: self.max_key_len,
+            format: self.format,
+        }
     }
 
     /// Returns `true` if the block should be flushed before adding an entry with the given key
@@ -105,6 +185,7 @@ impl KeyBlockAccumulator {
         self.size = 0;
         self.entry_count = 0;
         self.max_key_len = 0;
+        self.format = KeyBlockFormat::Unknown;
         // last_hash is intentionally not reset -- it is overwritten on the next add() call.
     }
 }
@@ -308,6 +389,56 @@ enum ValueRef {
     Deleted,
 }
 
+impl ValueRef {
+    /// Returns the key block entry type for this value reference.
+    fn entry_type(&self) -> EntryType {
+        EntryType(match self {
+            ValueRef::Small { .. } | ValueRef::PendingSmall { .. } => KEY_BLOCK_ENTRY_TYPE_SMALL,
+            ValueRef::Medium { .. } => KEY_BLOCK_ENTRY_TYPE_MEDIUM,
+            ValueRef::Inline { len, .. } => KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + *len,
+            ValueRef::Blob { .. } => KEY_BLOCK_ENTRY_TYPE_BLOB,
+            ValueRef::Deleted => KEY_BLOCK_ENTRY_TYPE_DELETED,
+        })
+    }
+
+    /// Writes the value bytes for this reference to a buffer.
+    ///
+    /// This is the shared serialization logic used by both variable-size and fixed-size key block
+    /// builders.
+    fn write_value_to(&self, buffer: &mut Vec<u8>) {
+        match self {
+            ValueRef::Small {
+                block_index,
+                offset,
+                size,
+            } => {
+                let mut scratch = [0u8; 8];
+                BE::write_u16(&mut scratch, *block_index);
+                BE::write_u16(&mut scratch[2..], *size);
+                BE::write_u32(&mut scratch[4..], *offset);
+                buffer.extend(&scratch);
+            }
+            ValueRef::Medium { block_index } => {
+                let mut scratch = [0u8; 2];
+                BE::write_u16(&mut scratch, *block_index);
+                buffer.extend(scratch);
+            }
+            ValueRef::Inline { data, len } => {
+                buffer.extend(&data[..*len as usize]);
+            }
+            ValueRef::Blob { blob_id } => {
+                let mut scratch = [0u8; 4];
+                BE::write_u32(&mut scratch, *blob_id);
+                buffer.extend(scratch);
+            }
+            ValueRef::Deleted => { /* no value bytes */ }
+            ValueRef::PendingSmall { .. } => {
+                unreachable!("PendingSmall should have been resolved");
+            }
+        }
+    }
+}
+
 struct PendingEntry<E> {
     entry: E,
     value_ref: ValueRef,
@@ -619,21 +750,19 @@ impl<E: Entry> StreamingSstWriter<E> {
             let entry = &self.pending_keys[i];
             let key_len = entry.entry.key_len();
             let key_hash = entry.entry.key_hash();
+            let value_type = entry.value_ref.entry_type();
 
             if self.current_key_block.should_flush(key_len, key_hash) {
                 let block_end = last_flushed_end + self.current_key_block.entry_count;
-                self.flush_key_block(
-                    last_flushed_end,
-                    block_end,
-                    self.current_key_block.max_key_len,
-                )?;
+                let info = self.current_key_block.flush_info();
+                self.flush_key_block(last_flushed_end, block_end, info)?;
                 flushed_key_size = cumulative_key_size;
                 last_flushed_end = block_end;
                 self.current_key_block.reset();
             }
 
             cumulative_key_size += key_len;
-            self.current_key_block.add(key_len, key_hash);
+            self.current_key_block.add(key_len, key_hash, value_type);
         }
 
         if last_flushed_end > 0 {
@@ -706,43 +835,40 @@ impl<E: Entry> StreamingSstWriter<E> {
     }
 
     /// Flushes a single key block from `pending_keys[start..end]`.
-    fn flush_key_block(&mut self, start: usize, end: usize, max_key_len: usize) -> Result<()> {
+    fn flush_key_block(&mut self, start: usize, end: usize, info: KeyBlockFlushInfo) -> Result<()> {
         let entry_count = end - start;
-        let has_hash = use_hash(max_key_len);
+        let has_hash = use_hash(info.max_key_len);
 
         self.key_buffer.clear();
-        let mut builder = KeyBlockBuilder::new(&mut self.key_buffer, entry_count as u32, has_hash);
 
-        for i in start..end {
-            let pending = &self.pending_keys[i];
-            match pending.value_ref {
-                ValueRef::Small {
-                    block_index,
-                    offset,
-                    size,
-                } => {
-                    builder.put_small(&pending.entry, block_index, offset, size, has_hash);
-                }
-                ValueRef::Medium { block_index } => {
-                    builder.put_medium(&pending.entry, block_index, has_hash);
-                }
-                ValueRef::Inline { data, len } => {
-                    builder.put_inline(&pending.entry, &data[..len as usize], has_hash);
-                }
-                ValueRef::Blob { blob_id } => {
-                    builder.put_blob(&pending.entry, blob_id, has_hash);
-                }
-                ValueRef::Deleted => {
-                    builder.delete(&pending.entry, has_hash);
-                }
-                ValueRef::PendingSmall { .. } => {
-                    unreachable!("PendingSmall should have been resolved");
-                }
+        if let KeyBlockFormat::Fixed {
+            key_len: key_size,
+            value_type,
+        } = info.format
+        {
+            let mut builder = FixedKeyBlockBuilder::new(
+                &mut self.key_buffer,
+                entry_count as u32,
+                has_hash,
+                key_size,
+                value_type,
+            );
+            for i in start..end {
+                let pending = &self.pending_keys[i];
+                builder.put(&pending.entry, &pending.value_ref, has_hash);
+            }
+            builder.finish();
+        } else {
+            let mut builder =
+                KeyBlockBuilder::new(&mut self.key_buffer, entry_count as u32, has_hash);
+
+            for i in start..end {
+                let pending = &self.pending_keys[i];
+                builder.put(&pending.entry, &pending.value_ref, has_hash);
             }
-        }
 
-        // Drop builder to release borrow on key_buffer before writing
-        builder.finish();
+            builder.finish();
+        }
 
         // Record boundary
         let first_hash = self.pending_keys[start].entry.key_hash();
@@ -874,19 +1000,20 @@ impl<E: Entry> StreamingSstWriter<E> {
             let entry = &self.pending_keys[i];
             let key_len = entry.entry.key_len();
             let key_hash = entry.entry.key_hash();
+            let value_type = entry.value_ref.entry_type();
 
             if acc.should_flush(key_len, key_hash) {
-                self.flush_key_block(block_start, i, acc.max_key_len)?;
+                self.flush_key_block(block_start, i, acc.flush_info())?;
                 block_start = i;
                 acc.reset();
             }
 
-            acc.add(key_len, key_hash);
+            acc.add(key_len, key_hash, value_type);
         }
 
         // Flush the final block
         if block_start < total {
-            self.flush_key_block(block_start, total, acc.max_key_len)?;
+            self.flush_key_block(block_start, total, acc.flush_info())?;
         }
 
         // Free VecDeque memory. Numeric fields are not reset because close() consumes self.
@@ -949,83 +1076,108 @@ impl<'l> KeyBlockBuilder<'l> {
         }
     }
 
-    /// Writes the 8-byte hash from a raw u64 if `has_hash` is true.
-    fn write_hash(&mut self, hash: u64, has_hash: bool) {
-        if has_hash {
-            self.buffer.extend_from_slice(&hash.to_be_bytes());
-        }
-    }
-
     /// Writes the entry header (position + type) for the current entry.
-    fn write_entry_header(&mut self, entry_type: u8) {
+    fn write_entry_header(&mut self, entry_type: EntryType) {
         let pos = self.buffer.len() - self.header_size;
         let header_offset = KEY_BLOCK_HEADER_SIZE + self.current_entry * 4;
-        let header = (pos as u32) | ((entry_type as u32) << 24);
+        let header = (pos as u32) | ((entry_type.0 as u32) << 24);
         BE::write_u32(&mut self.buffer[header_offset..header_offset + 4], header);
     }
 
-    /// Writes the hash and key from an entry.
-    fn write_entry_key<E: Entry>(&mut self, entry: &E, has_hash: bool) {
-        self.write_hash(entry.key_hash(), has_hash);
+    /// Writes a single entry (header + hash + key + value data) to the block.
+    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef, has_hash: bool) {
+        self.write_entry_header(value_ref.entry_type());
+        if has_hash {
+            self.buffer
+                .extend_from_slice(&entry.key_hash().to_be_bytes());
+        }
         entry.write_key_to(self.buffer);
-    }
-
-    /// Writes a small-sized value entry.
-    fn put_small<E: Entry>(
-        &mut self,
-        entry: &E,
-        value_block: u16,
-        value_offset: u32,
-        value_size: u16,
-        has_hash: bool,
-    ) {
-        self.write_entry_header(KEY_BLOCK_ENTRY_TYPE_SMALL);
-        self.write_entry_key(entry, has_hash);
-        self.buffer.write_u16::<BE>(value_block).unwrap();
-        self.buffer.write_u16::<BE>(value_size).unwrap();
-        self.buffer.write_u32::<BE>(value_offset).unwrap();
+        value_ref.write_value_to(self.buffer);
         self.current_entry += 1;
     }
 
-    /// Writes a medium-sized value entry.
-    fn put_medium<E: Entry>(&mut self, entry: &E, value_block: u16, has_hash: bool) {
-        self.write_entry_header(KEY_BLOCK_ENTRY_TYPE_MEDIUM);
-        self.write_entry_key(entry, has_hash);
-        self.buffer.write_u16::<BE>(value_block).unwrap();
-        self.current_entry += 1;
+    /// Returns the key block buffer.
+    fn finish(self) -> &'l mut Vec<u8> {
+        self.buffer
     }
+}
 
-    /// Writes a tombstone entry.
-    fn delete<E: Entry>(&mut self, entry: &E, has_hash: bool) {
-        self.write_entry_header(KEY_BLOCK_ENTRY_TYPE_DELETED);
-        self.write_entry_key(entry, has_hash);
-        self.current_entry += 1;
-    }
+// ---------------------------------------------------------------------------
+// FixedKeyBlockBuilder
+// ---------------------------------------------------------------------------
 
-    /// Writes a blob value entry.
-    fn put_blob<E: Entry>(&mut self, entry: &E, blob_id: u32, has_hash: bool) {
-        self.write_entry_header(KEY_BLOCK_ENTRY_TYPE_BLOB);
-        self.write_entry_key(entry, has_hash);
-        self.buffer.write_u32::<BE>(blob_id).unwrap();
-        self.current_entry += 1;
+/// The size of the fixed-size key block header (block type + entry count + key size + value type).
+const FIXED_KEY_BLOCK_HEADER_SIZE: usize = 6;
+
+/// Builder for a fixed-size key block where all entries share the same key size and value type.
+///
+/// No offset table is written — entry positions are computed arithmetically from the stride.
+struct FixedKeyBlockBuilder<'l> {
+    buffer: &'l mut Vec<u8>,
+}
+
+impl<'l> FixedKeyBlockBuilder<'l> {
+    fn new(
+        buffer: &'l mut Vec<u8>,
+        entry_count: u32,
+        has_hash: bool,
+        key_size: u8,
+        value_type: EntryType,
+    ) -> Self {
+        let hash_len: usize = if has_hash { 8 } else { 0 };
+        let val_size = value_type_val_size(value_type);
+        let stride = hash_len + key_size as usize + val_size;
+        buffer.reserve(FIXED_KEY_BLOCK_HEADER_SIZE + entry_count as usize * stride);
+
+        let block_type = if has_hash {
+            BLOCK_TYPE_FIXED_KEY_WITH_HASH
+        } else {
+            BLOCK_TYPE_FIXED_KEY_NO_HASH
+        };
+        buffer.extend_from_slice(&[
+            block_type,
+            (entry_count >> 16) as u8,
+            (entry_count >> 8) as u8,
+            entry_count as u8,
+            key_size,
+            value_type.0,
+        ]);
+
+        Self { buffer }
     }
 
-    /// Writes an inline value entry.
-    fn put_inline<E: Entry>(&mut self, entry: &E, value: &[u8], has_hash: bool) {
-        debug_assert!(value.len() <= MAX_INLINE_VALUE_SIZE);
-        let entry_type = KEY_BLOCK_ENTRY_TYPE_INLINE_MIN + value.len() as u8;
-        self.write_entry_header(entry_type);
-        self.write_entry_key(entry, has_hash);
-        self.buffer.extend_from_slice(value);
-        self.current_entry += 1;
+    /// Writes a single entry (hash + key + value data) to the block.
+    fn put<E: Entry>(&mut self, entry: &E, value_ref: &ValueRef, has_hash: bool) {
+        if has_hash {
+            self.buffer
+                .extend_from_slice(&entry.key_hash().to_be_bytes());
+        }
+        entry.write_key_to(self.buffer);
+        value_ref.write_value_to(self.buffer);
     }
 
-    /// Returns the key block buffer.
     fn finish(self) -> &'l mut Vec<u8> {
         self.buffer
     }
 }
 
+/// Returns the value size for a given entry type (builder-side, infallible).
+///
+/// This mirrors `entry_val_size` in the reader but panics on invalid types since the builder
+/// only produces valid types.
+fn value_type_val_size(ty: EntryType) -> usize {
+    match ty.0 {
+        KEY_BLOCK_ENTRY_TYPE_SMALL => SMALL_VALUE_REF_SIZE,
+        KEY_BLOCK_ENTRY_TYPE_MEDIUM => MEDIUM_VALUE_REF_SIZE,
+        KEY_BLOCK_ENTRY_TYPE_BLOB => BLOB_VALUE_REF_SIZE,
+        KEY_BLOCK_ENTRY_TYPE_DELETED => DELETED_VALUE_REF_SIZE,
+        ty if ty >= KEY_BLOCK_ENTRY_TYPE_INLINE_MIN => {
+            (ty - KEY_BLOCK_ENTRY_TYPE_INLINE_MIN) as usize
+        }
+        _ => panic!("Invalid key block entry type: {:?}", ty),
+    }
+}
+
 // ---------------------------------------------------------------------------
 // IndexBlockBuilder
 // ---------------------------------------------------------------------------