Turbopack: move block offsets from header to footer (#82047)

### What?

move block offset map to the end of the file to enable streaming write of the SST file.

Author: sokra

turbopack/crates/turbo-persistence/README.md

@@ -60,11 +60,11 @@ The SST file contains only data without any header.
 
 - serialized key Compression Dictionary
 - serialized value Compression Dictionary
-- foreach block
-  - 4 bytes end of block offset relative to start of all blocks
 - foreach block
   - 4 bytes uncompressed block length
   - compressed data
+- foreach block
+  - 4 bytes end of block offset relative to start of all blocks
 
 #### Index Block
 

turbopack/crates/turbo-persistence/src/static_sorted_file.rs

@@ -74,15 +74,15 @@ pub struct StaticSortedFileMetaData {
 }
 
 impl StaticSortedFileMetaData {
-    pub fn block_offsets_start(&self) -> usize {
-        let k: usize = self.key_compression_dictionary_length.into();
-        let v: usize = self.value_compression_dictionary_length.into();
-        k + v
+    pub fn block_offsets_start(&self, sst_len: usize) -> usize {
+        let bc: usize = self.block_count.into();
+        sst_len - (bc * size_of::<u32>())
     }
 
     pub fn blocks_start(&self) -> usize {
-        let bc: usize = self.block_count.into();
-        self.block_offsets_start() + bc * size_of::<u32>()
+        let k: usize = self.key_compression_dictionary_length.into();
+        let v: usize = self.value_compression_dictionary_length.into();
+        k + v
     }
 
     pub fn key_compression_dictionary_range(&self) -> Range<usize> {
@@ -347,11 +347,11 @@ impl StaticSortedFile {
                 self.meta.sequence_number,
                 block_index,
                 self.meta.block_count,
-                self.meta.block_offsets_start(),
+                self.meta.block_offsets_start(self.mmap.len()),
                 self.meta.blocks_start()
             );
         }
-        let offset = self.meta.block_offsets_start() + block_index as usize * 4;
+        let offset = self.meta.block_offsets_start(self.mmap.len()) + block_index as usize * 4;
         #[cfg(feature = "strict_checks")]
         if offset + 4 > self.mmap.len() {
             bail!(
@@ -361,7 +361,7 @@ impl StaticSortedFile {
                 block_index,
                 offset,
                 self.mmap.len(),
-                self.meta.block_offsets_start(),
+                self.meta.block_offsets_start(self.mmap.len()),
                 self.meta.blocks_start()
             );
         }
@@ -382,7 +382,7 @@ impl StaticSortedFile {
                 block_start,
                 block_end,
                 self.mmap.len(),
-                self.meta.block_offsets_start(),
+                self.meta.block_offsets_start(self.mmap.len()),
                 self.meta.blocks_start()
             );
         }

turbopack/crates/turbo-persistence/src/static_sorted_file_builder.rs

@@ -409,20 +409,24 @@ impl<'a> StaticSortedFileBuilder<'a> {
         file.write_all(&self.value_compression_dictionary)?;
 
         // Write the blocks
+        let block_count = self.blocks.len().try_into().unwrap();
+        let mut block_offsets = Vec::with_capacity(self.blocks.len());
         let mut offset = 0;
-        for (_, block) in &self.blocks {
+        for (uncompressed_size, block) in self.blocks {
             // Block length (including the uncompressed length field)
             let len = block.len() + 4;
             offset += len;
-            file.write_u32::<BE>(offset.try_into().unwrap())?;
-        }
-        let block_count = self.blocks.len().try_into().unwrap();
-        for (uncompressed_size, block) in self.blocks {
+            block_offsets.push(offset.try_into().unwrap());
             // Uncompressed size
             file.write_u32::<BE>(uncompressed_size)?;
             // Compressed block
             file.write_all(&block)?;
         }
+        // Write the block offsets
+        for offset in block_offsets {
+            file.write_u32::<BE>(offset)?;
+        }
+
         let meta = StaticSortedFileBuilderMeta {
             min_hash: self.min_hash,
             max_hash: self.max_hash,