Repository: vercel/next.js
Pull request: turbo-persistence: StreamingSstWriter performance and readability improvements

## Summary

Performance and readability improvements to `StreamingSstWriter` in turbo-persistence, based on a performance audit of the streaming SST file builder.

### Pre-allocate Vec fields (33eb3e8f)
- Pre-allocate `compress_buffer`, `pending_small_value_block`, `key_buffer`, `block_offsets`, and `key_block_boundaries` with capacity estimates derived from `max_entry_count`
- Eliminates ~10-15 reallocations during early writes (e.g. for 1M-entry compaction files, the pre-allocation is ~89 KB across all Vecs)

### Readability cleanup (3de7a45a)
- Extract `KeyBlockAccumulator` struct from four loose `current_key_block_*` fields, with `should_flush`/`add`/`reset` methods
- Add architectural diagram comment for `pending_keys` queue layout showing resolved vs. unresolved regions
- Name magic constant `BLOCK_INDEX_CAPACITY_BUFFER` (was bare `16`)
- Simplify `flush_remaining_key_blocks()` using local `KeyBlockAccumulator` and remove dead state resets after `close()`
- Remove `try_flush_key_block()` — its logic is now split between `KeyBlockAccumulator::should_flush()` and inline flush calls, which is clearer
- Deduplicate `TestEntry` constructors via shared `new()` helper

### Derive value block estimate from constants (580e5d4c)
- Replace magic number `40` (entries per value block) with `MIN_SMALL_VALUE_BLOCK_SIZE / AVG_SMALL_VALUE_SIZE` (= 128)
- Add `AVG_SMALL_VALUE_SIZE = 64` constant

## Test Plan
- All 14 existing `static_sorted_file_builder` tests pass: `cargo test -p turbo-persistence --lib -- static_sorted_file_builder`
- No behavioral changes — all changes are internal to the writer

Changed files:
filename: turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs
patch: @@ -12,7 +12,7 @@ use turbo_bincode::turbo_bincode_encode;
 
 use crate::{
     compression::compress_into_buffer,
-    constants::{MAX_INLINE_VALUE_SIZE, MIN_SMALL_VALUE_BLOCK_SIZE},
+    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,
@@ -37,6 +37,74 @@ const MAX_KEY_BLOCK_SIZE: usize = 16 * 1024;
 const KEY_BLOCK_ENTRY_META_OVERHEAD: usize = 20;
 /// The aimed false positive rate for the AMQF
 const AMQF_FALSE_POSITIVE_RATE: f64 = 0.01;
+/// Assumed average small value size for pre-allocation estimates.
+/// Intentionally conservative (small values range from MAX_INLINE_VALUE_SIZE+1 to
+/// MAX_SMALL_VALUE_SIZE = 4096): a low estimate over-counts value blocks, which is
+/// preferable to under-allocating vectors.
+const AVG_SMALL_VALUE_SIZE: usize = 64;
+
+/// Safety margin for block index capacity estimation in
+/// [`StreamingSstWriter::has_block_index_capacity`]. Accounts for rounding in the entry-count and
+/// byte-size based estimates of pending key blocks.
+const BLOCK_INDEX_CAPACITY_BUFFER: usize = 16;
+
+/// Tracks the accumulated state of the current incomplete key block.
+///
+/// During streaming, this sits on [`StreamingSstWriter`] and tracks the tail of the resolved
+/// prefix. Entries are added one at a time; when [`should_flush`](Self::should_flush) returns
+/// `true`, the caller should flush the block and call [`reset`](Self::reset).
+struct KeyBlockAccumulator {
+    /// Accumulated byte size (keys + per-entry overhead) of entries in this block.
+    size: usize,
+    /// Number of entries accumulated so far.
+    entry_count: usize,
+    /// Maximum key length among accumulated entries (determines whether hashes are stored).
+    max_key_len: usize,
+    /// Hash of the most recently added entry (used to avoid splitting entries with equal hashes
+    /// across blocks).
+    last_hash: u64,
+}
+
+impl KeyBlockAccumulator {
+    fn new() -> Self {
+        Self {
+            size: 0,
+            entry_count: 0,
+            max_key_len: 0,
+            last_hash: 0,
+        }
+    }
+
+    /// Records a new entry in the accumulator.
+    fn add(&mut self, key_len: usize, key_hash: u64) {
+        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;
+    }
+
+    /// Returns `true` if the block should be flushed before adding an entry with the given key
+    /// length and hash. Returns `false` for empty blocks and when the next entry shares its hash
+    /// with the current last entry (to avoid splitting equal-hash runs).
+    fn should_flush(&self, next_key_len: usize, next_key_hash: u64) -> bool {
+        if self.entry_count == 0 {
+            return false;
+        }
+        let would_exceed_size =
+            self.size + next_key_len + KEY_BLOCK_ENTRY_META_OVERHEAD > MAX_KEY_BLOCK_SIZE;
+        let would_exceed_entries = self.entry_count >= MAX_KEY_BLOCK_ENTRIES;
+        // Never split entries with the same hash across blocks.
+        (would_exceed_size || would_exceed_entries) && self.last_hash != next_key_hash
+    }
+
+    /// Resets the accumulator for a new key block.
+    fn reset(&mut self) {
+        self.size = 0;
+        self.entry_count = 0;
+        self.max_key_len = 0;
+        // last_hash is intentionally not reset -- it is overwritten on the next add() call.
+    }
+}
 
 /// Determines whether to store the hash per entry based on max key length.
 fn use_hash(max_key_len: usize) -> bool {
@@ -246,13 +314,26 @@ pub struct StreamingSstWriter<E: Entry> {
 
     /// Pending key entries waiting to be flushed as key blocks.
     ///
-    /// Entries are appended at the back and flushed from the front. The front entries up to
-    /// `first_pending_small_index` are fully resolved and eligible for key block flushing.
+    /// Entries are appended at the back and drained from the front once flushed.
     ///
-    /// **Note:** This queue is effectively unbounded. In a pathological case -- a small number of
-    /// small values followed by a large number of medium/inline values -- the queue can grow large
-    /// because the front entries reference an unflushed small value block while the back keeps
-    /// accepting resolved entries.
+    /// ```text
+    ///  Resolved entries              Unresolved entries
+    ///  (value block index known)     (PendingSmall references)
+    /// |------------------------------|--------------------------|
+    /// 0                     first_pending_small_index         len()
+    ///
+    ///  ^-- current_key_block tracks      ^-- these wait for
+    ///      the incomplete tail block         flush_small_value_block()
+    ///      within this region                to resolve them
+    /// ```
+    ///
+    /// [`advance_boundary_to`](Self::advance_boundary_to) scans the resolved prefix, flushes
+    /// complete key blocks from the front, and drains them. When a small value block is flushed,
+    /// all `PendingSmall` entries are resolved in-place and the boundary advances to `len()`.
+    ///
+    /// **Unbounded growth note:** If a small number of small values appear early, followed by
+    /// many medium/inline values, the queue grows because the front entries block on the
+    /// unflushed small value block while the back keeps accepting resolved entries.
     pending_keys: VecDeque<PendingEntry<E>>,
 
     /// Index into `pending_keys` of the first entry that has a `PendingSmall` reference for the
@@ -290,15 +371,8 @@ pub struct StreamingSstWriter<E: Entry> {
     /// Total byte size of keys in `pending_keys` (for block capacity estimation).
     pending_key_total_size: usize,
 
-    /// Accumulated byte size of the current incomplete key block at the tail
-    /// of the resolved prefix.
-    current_key_block_size: usize,
-    /// Entry count in the current incomplete key block.
-    current_key_block_entry_count: usize,
-    /// Maximum key length in the current incomplete key block.
-    current_key_block_max_key_len: usize,
-    /// Hash of the last entry added to the current incomplete key block.
-    current_key_block_last_hash: u64,
+    /// State of the current incomplete key block at the tail of the resolved prefix.
+    current_key_block: KeyBlockAccumulator,
 
     /// Set to `true` by `close()` so the Drop guard can detect writers dropped without closing.
     #[cfg(debug_assertions)]
@@ -316,29 +390,41 @@ impl<E: Entry> StreamingSstWriter<E> {
         let filter = qfilter::Filter::new(max_entry_count.max(1), AMQF_FALSE_POSITIVE_RATE)
             .expect("Filter can't be constructed");
 
+        // Estimate number of key blocks based on max entry count.
+        // Each key block holds up to MAX_KEY_BLOCK_ENTRIES entries.
+        let estimated_key_blocks = (max_entry_count as usize)
+            .div_ceil(MAX_KEY_BLOCK_ENTRIES)
+            .max(1);
+        // Estimate value blocks assuming all entries are small values of average size.
+        // Each small value block holds ~MIN_SMALL_VALUE_BLOCK_SIZE / AVG_SMALL_VALUE_SIZE entries.
+        let entries_per_value_block = MIN_SMALL_VALUE_BLOCK_SIZE / AVG_SMALL_VALUE_SIZE;
+        let estimated_value_blocks = (max_entry_count as usize)
+            .div_ceil(entries_per_value_block)
+            .max(1);
+        let estimated_total_blocks = estimated_key_blocks + estimated_value_blocks + 1;
+
         Ok(Self {
             file: Some(file),
-            compress_buffer: Vec::new(),
-            block_offsets: Vec::new(),
-            pending_keys: VecDeque::new(),
+            compress_buffer: Vec::with_capacity(MIN_SMALL_VALUE_BLOCK_SIZE + MAX_SMALL_VALUE_SIZE),
+            block_offsets: Vec::with_capacity(estimated_total_blocks),
+            pending_keys: VecDeque::with_capacity(entries_per_value_block),
             first_pending_small_index: 0,
             #[cfg(debug_assertions)]
             current_small_block_id: 0,
-            pending_small_value_block: Vec::new(),
-            key_buffer: Vec::new(),
+            pending_small_value_block: Vec::with_capacity(
+                MIN_SMALL_VALUE_BLOCK_SIZE + MAX_SMALL_VALUE_SIZE,
+            ),
+            key_buffer: Vec::with_capacity(MAX_KEY_BLOCK_SIZE),
             filter: Some(filter),
-            key_block_boundaries: Vec::new(),
+            key_block_boundaries: Vec::with_capacity(estimated_key_blocks),
             min_hash: u64::MAX,
             max_hash: 0,
             entry_count: 0,
             flags,
             total_key_size: 0,
             total_value_size: 0,
             pending_key_total_size: 0,
-            current_key_block_size: 0,
-            current_key_block_entry_count: 0,
-            current_key_block_max_key_len: 0,
-            current_key_block_last_hash: 0,
+            current_key_block: KeyBlockAccumulator::new(),
             #[cfg(debug_assertions)]
             finished: false,
         })
@@ -370,8 +456,7 @@ impl<E: Entry> StreamingSstWriter<E> {
             .max(self.pending_key_total_size.div_ceil(MAX_KEY_BLOCK_SIZE))
             .max(1);
         let index_block = 1;
-        // Add some buffer, just to be safe...
-        let buffer = 16;
+        let buffer = BLOCK_INDEX_CAPACITY_BUFFER;
         blocks_written + pending_small_block + pending_key_blocks + index_block + buffer
             < u16::MAX as usize
     }
@@ -516,16 +601,20 @@ impl<E: Entry> StreamingSstWriter<E> {
             let key_len = entry.entry.key_len();
             let key_hash = entry.entry.key_hash();
 
-            if self.try_flush_key_block(last_flushed_end, key_len, key_hash)? {
+            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,
+                )?;
                 flushed_key_size = cumulative_key_size;
-                last_flushed_end = i;
+                last_flushed_end = block_end;
+                self.current_key_block.reset();
             }
 
             cumulative_key_size += key_len;
-            self.current_key_block_size += key_len + KEY_BLOCK_ENTRY_META_OVERHEAD;
-            self.current_key_block_max_key_len = self.current_key_block_max_key_len.max(key_len);
-            self.current_key_block_entry_count += 1;
-            self.current_key_block_last_hash = key_hash;
+            self.current_key_block.add(key_len, key_hash);
         }
 
         if last_flushed_end > 0 {
@@ -597,42 +686,10 @@ impl<E: Entry> StreamingSstWriter<E> {
         Ok(())
     }
 
-    /// Flushes the current key block if it is full (considering the next entry), then resets
-    /// the block-tracking state. Returns `true` if a block was flushed.
-    ///
-    /// `block_start` is the index into `pending_keys` where the current key block begins.
-    /// `next_key_len` / `next_key_hash` describe the entry about to be added.
-    fn try_flush_key_block(
-        &mut self,
-        block_start: usize,
-        next_key_len: usize,
-        next_key_hash: u64,
-    ) -> Result<bool> {
-        if self.current_key_block_entry_count == 0 {
-            return Ok(false);
-        }
-        let would_exceed_size =
-            self.current_key_block_size + next_key_len + KEY_BLOCK_ENTRY_META_OVERHEAD
-                > MAX_KEY_BLOCK_SIZE;
-        let would_exceed_entries = self.current_key_block_entry_count >= MAX_KEY_BLOCK_ENTRIES;
-        if !(would_exceed_size || would_exceed_entries)
-            || self.current_key_block_last_hash == next_key_hash
-        {
-            return Ok(false);
-        }
-
-        let block_end = block_start + self.current_key_block_entry_count;
-        self.flush_key_block(block_start, block_end)?;
-        self.current_key_block_size = 0;
-        self.current_key_block_entry_count = 0;
-        self.current_key_block_max_key_len = 0;
-        Ok(true)
-    }
-
     /// Flushes a single key block from `pending_keys[start..end]`.
-    fn flush_key_block(&mut self, start: usize, end: usize) -> Result<()> {
+    fn flush_key_block(&mut self, start: usize, end: usize, max_key_len: usize) -> Result<()> {
         let entry_count = end - start;
-        let has_hash = use_hash(self.current_key_block_max_key_len);
+        let has_hash = use_hash(max_key_len);
 
         self.key_buffer.clear();
         let mut builder = KeyBlockBuilder::new(&mut self.key_buffer, entry_count as u32, has_hash);
@@ -774,45 +831,50 @@ impl<E: Entry> StreamingSstWriter<E> {
 
     /// Flushes all remaining entries as key blocks. Called from `close()` after all small value
     /// blocks have been flushed, so all PendingSmall entries are resolved.
+    ///
+    /// This loop mirrors [`advance_boundary_to`], but uses a local accumulator (since the
+    /// `self.current_key_block` state is stale) and flushes the final incomplete block
+    /// (unlike `advance_boundary_to`, which keeps it for more entries during streaming).
     fn flush_remaining_key_blocks(&mut self) -> Result<()> {
         if self.pending_keys.is_empty() {
             return Ok(());
         }
 
+        // After flush_small_value_block() in close(), no PendingSmall entries should remain.
+        // first_pending_small_index may be non-zero (when all entries are medium/inline/etc
+        // and advance_boundary_to was never called), but it must equal pending_keys.len(),
+        // meaning no entries after the boundary exist.
+        debug_assert_eq!(
+            self.first_pending_small_index,
+            self.pending_keys.len(),
+            "expected no unresolved PendingSmall entries after flush_small_value_block"
+        );
+
         let total = self.pending_keys.len();
         let mut block_start = 0;
-
-        // Reset current block state so we iterate from scratch. The partially-tracked
-        // block from previous add() calls is irrelevant here since we re-scan all entries.
-        self.current_key_block_size = 0;
-        self.current_key_block_entry_count = 0;
-        self.current_key_block_max_key_len = 0;
+        let mut acc = KeyBlockAccumulator::new();
 
         for i in 0..total {
             let entry = &self.pending_keys[i];
             let key_len = entry.entry.key_len();
             let key_hash = entry.entry.key_hash();
 
-            if self.try_flush_key_block(block_start, key_len, key_hash)? {
+            if acc.should_flush(key_len, key_hash) {
+                self.flush_key_block(block_start, i, acc.max_key_len)?;
                 block_start = i;
+                acc.reset();
             }
 
-            self.current_key_block_size += key_len + KEY_BLOCK_ENTRY_META_OVERHEAD;
-            self.current_key_block_max_key_len = self.current_key_block_max_key_len.max(key_len);
-            self.current_key_block_entry_count += 1;
-            self.current_key_block_last_hash = key_hash;
+            acc.add(key_len, key_hash);
         }
 
         // Flush the final block
         if block_start < total {
-            self.flush_key_block(block_start, total)?;
+            self.flush_key_block(block_start, total, acc.max_key_len)?;
         }
 
+        // Free VecDeque memory. Numeric fields are not reset because close() consumes self.
         self.pending_keys.clear();
-        self.pending_key_total_size = 0;
-        self.current_key_block_size = 0;
-        self.current_key_block_entry_count = 0;
-        self.current_key_block_max_key_len = 0;
         Ok(())
     }
 }
@@ -1026,66 +1088,39 @@ mod tests {
     }
 
     impl TestEntry {
-        fn small(key: &[u8], value: &[u8]) -> Self {
+        fn new(key: &[u8], value_kind: TestValueKind) -> Self {
             let key = key.to_vec();
             let hash = hash_key(&key);
             Self {
                 key,
                 hash,
-                value_kind: TestValueKind::Small(value.to_vec()),
+                value_kind,
             }
         }
 
+        fn small(key: &[u8], value: &[u8]) -> Self {
+            Self::new(key, TestValueKind::Small(value.to_vec()))
+        }
+
         fn inline(key: &[u8], value: &[u8]) -> Self {
             debug_assert!(value.len() <= MAX_INLINE_VALUE_SIZE);
-            let key = key.to_vec();
-            let hash = hash_key(&key);
-            Self {
-                key,
-                hash,
-                value_kind: TestValueKind::Inline(value.to_vec()),
-            }
+            Self::new(key, TestValueKind::Inline(value.to_vec()))
         }
 
         fn medium(key: &[u8], value: &[u8]) -> Self {
-            let key = key.to_vec();
-            let hash = hash_key(&key);
-            Self {
-                key,
-                hash,
-                value_kind: TestValueKind::Medium(value.to_vec()),
-            }
+            Self::new(key, TestValueKind::Medium(value.to_vec()))
         }
 
         fn blob(key: &[u8], blob_id: u32) -> Self {
-            let key = key.to_vec();
-            let hash = hash_key(&key);
-            Self {
-                key,
-                hash,
-                value_kind: TestValueKind::Blob(blob_id),
-            }
+            Self::new(key, TestValueKind::Blob(blob_id))
         }
 
         fn deleted(key: &[u8]) -> Self {
-            let key = key.to_vec();
-            let hash = hash_key(&key);
-            Self {
-                key,
-                hash,
-                value_kind: TestValueKind::Deleted,
-            }
+            Self::new(key, TestValueKind::Deleted)
         }
 
         fn medium_raw(key: &[u8], value: &[u8]) -> Self {
-            let key = key.to_vec();
-            let hash = hash_key(&key);
-            Self {
-                key,
-                hash,
-                // Store as uncompressed raw block (uncompressed_size = 0 means "not compressed").
-                value_kind: TestValueKind::MediumRaw(value.to_vec()),
-            }
+            Self::new(key, TestValueKind::MediumRaw(value.to_vec()))
         }
 
         fn expected_value(&self) -> Option<&[u8]> {