turbo_persistence/
write_batch.rs

1use std::{
2    cell::SyncUnsafeCell,
3    io::Write,
4    mem::{replace, take},
5    path::PathBuf,
6    sync::atomic::{AtomicU32, AtomicU64, Ordering},
7};
8
9use anyhow::{Context, Result, bail};
10use byteorder::{BE, WriteBytesExt};
11use either::Either;
12use fs_err::File;
13use parking_lot::Mutex;
14use smallvec::SmallVec;
15use thread_local::ThreadLocal;
16
17use crate::{
18    FamilyConfig, FamilyKind, ValueBuffer,
19    collector::Collector,
20    collector_entry::CollectorEntry,
21    compression::{Compressor, checksum_block},
22    constants::{MAX_INLINE_VALUE_SIZE, MAX_MEDIUM_VALUE_SIZE, THREAD_LOCAL_SIZE_SHIFT},
23    db::WriteOperationGuard,
24    key::StoreKey,
25    meta_file::MetaEntryFlags,
26    meta_file_builder::MetaFileBuilder,
27    parallel_scheduler::ParallelScheduler,
28    static_sorted_file_builder::{StaticSortedFileBuilderMeta, write_static_stored_file},
29};
30
31/// A newly created database file (meta, SST, or blob), carrying its on-disk size so commit can sum
32/// written bytes without stat'ing the files afterwards.
33pub(crate) struct NewFile {
34    pub(crate) seq: u32,
35    pub(crate) file: File,
36    /// On-disk size in bytes.
37    pub(crate) size: u64,
38}
39
40/// The thread local state of a `WriteBatch`. `FAMILIES` should fit within a `u32`.
41//
42// NOTE: This type *must* use `usize`, even though the real type used in storage is `u32` because
43// there's no way to cast a `u32` to `usize` when declaring an array without the nightly
44// `min_generic_const_args` feature.
45struct ThreadLocalState<K: StoreKey + Send, const FAMILIES: usize> {
46    /// The collectors for each family.
47    collectors: [Option<Collector<K, THREAD_LOCAL_SIZE_SHIFT>>; FAMILIES],
48    /// The list of new blob files that have been created.
49    new_blob_files: Vec<NewFile>,
50}
51
52const COLLECTOR_SHARDS: usize = 4;
53const COLLECTOR_SHARD_SHIFT: usize =
54    u64::BITS as usize - COLLECTOR_SHARDS.trailing_zeros() as usize;
55
56/// The result of a `WriteBatch::finish` operation.
57pub(crate) struct FinishResult {
58    pub(crate) sequence_number: u32,
59    pub(crate) new_meta_files: Vec<NewFile>,
60    pub(crate) new_sst_files: Vec<NewFile>,
61    pub(crate) new_blob_files: Vec<NewFile>,
62    /// Number of keys written in this batch.
63    pub(crate) keys_written: u64,
64}
65
66enum GlobalCollectorState<K: StoreKey + Send> {
67    /// Initial state. Single collector. Once the collector is full, we switch to sharded mode.
68    Unsharded(Collector<K>),
69    /// Sharded mode.
70    /// We use multiple collectors, and select one based on the first bits of the key hash.
71    Sharded([Collector<K>; COLLECTOR_SHARDS]),
72}
73
74/// A write batch.
75pub struct WriteBatch<'db, K: StoreKey + Send, S: ParallelScheduler, const FAMILIES: usize> {
76    /// RAII guard that releases the write-operation slot (and rolls back on failure) on drop.
77    _write_guard: WriteOperationGuard<'db>,
78    /// Parallel scheduler
79    parallel_scheduler: S,
80    /// The database path
81    db_path: PathBuf,
82    /// Per-family storage configuration.
83    #[cfg_attr(not(feature = "verify_sst_content"), allow(dead_code))]
84    family_configs: [FamilyConfig; FAMILIES],
85    /// The current sequence number counter. Increased for every new SST file or blob file.
86    current_sequence_number: AtomicU32,
87    /// The thread local state.
88    thread_locals: ThreadLocal<SyncUnsafeCell<ThreadLocalState<K, FAMILIES>>>,
89    /// Collectors in use. The thread local collectors flush into these when they are full.
90    collectors: [Mutex<GlobalCollectorState<K>>; FAMILIES],
91    /// Meta file builders for each family.
92    meta_collectors: [Mutex<Vec<(u32, StaticSortedFileBuilderMeta<'static>)>>; FAMILIES],
93    /// The list of new SST files that have been created.
94    new_sst_files: Mutex<Vec<NewFile>>,
95}
96
97impl<'db, K: StoreKey + Send + Sync, S: ParallelScheduler, const FAMILIES: usize>
98    WriteBatch<'db, K, S, FAMILIES>
99{
100    /// Creates a new write batch for a database with per-family configuration.
101    pub(crate) fn new(
102        write_guard: WriteOperationGuard<'db>,
103        path: PathBuf,
104        current: u32,
105        parallel_scheduler: S,
106        family_configs: [FamilyConfig; FAMILIES],
107    ) -> Self {
108        const {
109            assert!(FAMILIES <= usize_from_u32(u32::MAX));
110        };
111        Self {
112            _write_guard: write_guard,
113            parallel_scheduler,
114            db_path: path,
115            family_configs,
116            current_sequence_number: AtomicU32::new(current),
117            thread_locals: ThreadLocal::new(),
118            collectors: [(); FAMILIES]
119                .map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new()))),
120            meta_collectors: [(); FAMILIES].map(|_| Mutex::new(Vec::new())),
121            new_sst_files: Mutex::new(Vec::new()),
122        }
123    }
124
125    /// Marks the write operation as successfully completed.
126    ///
127    /// Must be called before dropping the `WriteBatch` to skip the rollback in the guard's `Drop`
128    /// impl. Typically called by `TurboPersistence::commit_write_batch` after a successful commit.
129    pub(crate) fn mark_succeeded(&mut self) {
130        self._write_guard.success();
131    }
132
133    /// Returns the thread local state for the current thread.
134    #[allow(clippy::mut_from_ref)]
135    fn thread_local_state(&self) -> &mut ThreadLocalState<K, FAMILIES> {
136        let cell = self.thread_locals.get_or(|| {
137            SyncUnsafeCell::new(ThreadLocalState {
138                collectors: [const { None }; FAMILIES],
139                new_blob_files: Vec::new(),
140            })
141        });
142        // Safety: We know that the cell is only accessed from the current thread.
143        unsafe { &mut *cell.get() }
144    }
145
146    /// Returns the collector for a family for the current thread.
147    fn thread_local_collector_mut<'l>(
148        &self,
149        state: &'l mut ThreadLocalState<K, FAMILIES>,
150        family: u32,
151    ) -> Result<&'l mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>> {
152        debug_assert!(usize_from_u32(family) < FAMILIES);
153        let collector =
154            state.collectors[usize_from_u32(family)].get_or_insert_with(|| Collector::new());
155        if collector.is_full() {
156            self.flush_thread_local_collector(family, collector)?;
157        }
158        Ok(collector)
159    }
160
161    #[tracing::instrument(level = "trace", skip(self, collector), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
162    fn flush_thread_local_collector(
163        &self,
164        family: u32,
165        collector: &mut Collector<K, THREAD_LOCAL_SIZE_SHIFT>,
166    ) -> Result<()> {
167        let mut full_collectors = SmallVec::<[_; 2]>::new();
168        {
169            let mut global_collector_state = self.collectors[usize_from_u32(family)].lock();
170            for entry in collector.drain() {
171                match &mut *global_collector_state {
172                    GlobalCollectorState::Unsharded(collector) => {
173                        collector.add_entry(entry);
174                        if collector.is_full() {
175                            // When full, split the entries into shards.
176                            let mut shards: [Collector<K>; 4] =
177                                [(); COLLECTOR_SHARDS].map(|_| Collector::new());
178                            for entry in collector.drain() {
179                                let shard = (entry.key.hash >> COLLECTOR_SHARD_SHIFT) as usize;
180                                shards[shard].add_entry(entry);
181                            }
182                            // There is a rare edge case where all entries are in the same shard,
183                            // and the collector is full after the split.
184                            for collector in shards.iter_mut() {
185                                if collector.is_full() {
186                                    full_collectors
187                                        .push(replace(&mut *collector, Collector::new()));
188                                }
189                            }
190                            *global_collector_state = GlobalCollectorState::Sharded(shards);
191                        }
192                    }
193                    GlobalCollectorState::Sharded(shards) => {
194                        let shard = (entry.key.hash >> COLLECTOR_SHARD_SHIFT) as usize;
195                        let collector = &mut shards[shard];
196                        collector.add_entry(entry);
197                        if collector.is_full() {
198                            full_collectors.push(replace(&mut *collector, Collector::new()));
199                        }
200                    }
201                }
202            }
203        }
204        // After flushing write all the full global collectors to disk.
205        // TODO: This can distribute work unfairly
206        // * a thread could fill up multiple global collectors and then get stuck writing them all
207        //   out, if multiple threads could work on it we could take care of spare IO parallism
208        // * we can also have too much IO parallism with many threads concurrently writing files.
209        //
210        // Ideally we would limit the amount of data buffered in memory and control the amount of IO
211        // parallism.  Consider:
212        // * store full-buffers as a field on WireBatch (queued writes)
213        // * each thread will attempt to poll and flush a full buffer after flushing its local
214        //   buffer.
215        // This will distribute the writing work more fairly, but now we have the problem of to
216        // many concurrent writes contending for filesystem locks.  So we could also use a semaphore
217        // to restrict how many concurrent writes occur.  But then we would accumulate 'fullBuffers'
218        // leading to too much memory consumption.  So really we also need to slow down the threads
219        // submitting work data.  To do this we could simply use a tokio semaphore and make all
220        // these operations async, or we could integrate with the parallel::map operation that is
221        // driving the work to slow down task submission in this case.
222        for mut global_collector in full_collectors {
223            // When the global collector is full, we create a new SST file.
224            let sst = self.create_sst_file(
225                family,
226                global_collector.sorted(self.family_configs[usize_from_u32(family)].kind),
227            )?;
228            self.new_sst_files.lock().push(sst);
229            drop(global_collector);
230        }
231        Ok(())
232    }
233
234    /// Puts a key-value pair into the write batch.
235    pub fn put(&self, family: u32, key: K, value: ValueBuffer<'_>) -> Result<()> {
236        let state = self.thread_local_state();
237        let collector = self.thread_local_collector_mut(state, family)?;
238        if value.len() <= MAX_MEDIUM_VALUE_SIZE {
239            collector.put(key, value);
240        } else {
241            let blob = self.create_blob(family, &value)?;
242            collector.put_blob(key, blob.seq);
243            state.new_blob_files.push(blob);
244        }
245        Ok(())
246    }
247
248    /// Puts a delete operation into the write batch. This deletes *all* values for `key`.
249    ///
250    /// Combining this with a [`WriteBatch::put`] of the same key in the same batch is **not
251    /// supported**: which one wins is undefined, and callers are expected to resolve the intent
252    /// themselves before writing.
253    pub fn delete(&self, family: u32, key: K) -> Result<()> {
254        let state = self.thread_local_state();
255        let collector = self.thread_local_collector_mut(state, family)?;
256        collector.delete(key);
257        Ok(())
258    }
259
260    /// Deletes a single key-value pair, leaving any other values for `key` intact.
261    ///
262    /// Only valid for [`FamilyKind::MultiValue`] families: in a `SingleValue` family a key has one
263    /// value and [`WriteBatch::delete`] already removes it exactly.
264    ///
265    /// Deleting a pair that is written in the same batch — by this or any other operation on the
266    /// key — is **not supported**, for the reason given on [`WriteBatch::delete`]: which one wins
267    /// is undefined, and it is the caller's job to resolve that before writing.
268    ///
269    /// Only values of at most [`MAX_INLINE_VALUE_SIZE`] bytes can be deleted this way.  This is a
270    /// simplifying limitation that could be relaxed if needed. Of course in general the storage
271    /// overhead of deleting large values by value makes it apriori inefficient.
272    pub fn delete_value(&self, family: u32, key: K, value: ValueBuffer<'_>) -> Result<()> {
273        let family_config = &self.family_configs[usize_from_u32(family)];
274        if family_config.kind != FamilyKind::MultiValue {
275            bail!(
276                "delete_value is only valid for MultiValue families, but family {} is SingleValue",
277                family_config.name
278            );
279        }
280        if value.len() > MAX_INLINE_VALUE_SIZE {
281            bail!(
282                "delete_value only supports values of at most {MAX_INLINE_VALUE_SIZE} bytes, got \
283                 {} bytes",
284                value.len()
285            );
286        }
287        let state = self.thread_local_state();
288        let collector = self.thread_local_collector_mut(state, family)?;
289        collector.delete_value(key, &value);
290        Ok(())
291    }
292
293    /// Flushes a family of the write batch, reducing the amount of buffered memory used.
294    /// Does not commit any data persistently.
295    ///
296    /// # Safety
297    ///
298    /// Caller must ensure that no concurrent put or delete operation is happening on the flushed
299    /// family.
300    #[tracing::instrument(level = "trace", skip(self), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
301    pub unsafe fn flush(&self, family: u32) -> Result<()> {
302        // Flush the thread local collectors to the global collector.
303        let mut collectors = Vec::new();
304        for cell in self.thread_locals.iter() {
305            let state = unsafe { &mut *cell.get() };
306            if let Some(collector) = state.collectors[usize_from_u32(family)].take()
307                && !collector.is_empty()
308            {
309                collectors.push(collector);
310            }
311        }
312
313        self.parallel_scheduler
314            .try_parallel_for_each_owned(collectors, |mut collector| {
315                self.flush_thread_local_collector(family, &mut collector)?;
316                drop(collector);
317                anyhow::Ok(())
318            })?;
319
320        // Now we flush the global collector(s).
321        let family_usize = usize_from_u32(family);
322        let mut collector_state = self.collectors[family_usize].lock();
323        let family_config = self.family_configs[family_usize];
324        match &mut *collector_state {
325            GlobalCollectorState::Unsharded(collector) => {
326                if !collector.is_empty() {
327                    let sst = self.create_sst_file(family, collector.sorted(family_config.kind))?;
328                    collector.clear();
329                    self.new_sst_files.lock().push(sst);
330                }
331            }
332            GlobalCollectorState::Sharded(_) => {
333                let GlobalCollectorState::Sharded(mut shards) = replace(
334                    &mut *collector_state,
335                    GlobalCollectorState::Unsharded(Collector::new()),
336                ) else {
337                    unreachable!();
338                };
339                self.parallel_scheduler
340                    .try_parallel_for_each_mut(&mut shards, |collector| {
341                        if !collector.is_empty() {
342                            let sst =
343                                self.create_sst_file(family, collector.sorted(family_config.kind))?;
344                            collector.clear();
345                            self.new_sst_files.lock().push(sst);
346                            collector.drop_contents();
347                        }
348                        anyhow::Ok(())
349                    })?;
350            }
351        }
352
353        Ok(())
354    }
355
356    /// Finishes the write batch by returning the new sequence number and the new SST files. This
357    /// writes all outstanding thread local data to disk.
358    #[tracing::instrument(level = "trace", skip_all)]
359    pub(crate) fn finish(
360        &mut self,
361        get_accessed_key_hashes: impl Fn(u32) -> qfilter::Filter + Send + Sync,
362    ) -> Result<FinishResult> {
363        let mut new_blob_files = Vec::new();
364
365        // First, we flush all thread local collectors to the global collectors.
366        {
367            let _span = tracing::trace_span!("flush thread local collectors").entered();
368            let mut collectors = [const { Vec::new() }; FAMILIES];
369            for cell in self.thread_locals.iter_mut() {
370                let state = cell.get_mut();
371                new_blob_files.append(&mut state.new_blob_files);
372                for (family, thread_local_collector) in state.collectors.iter_mut().enumerate() {
373                    if let Some(collector) = thread_local_collector.take()
374                        && !collector.is_empty()
375                    {
376                        collectors[family].push(collector);
377                    }
378                }
379            }
380            let to_flush = collectors
381                .into_iter()
382                .enumerate()
383                .flat_map(|(family, collector)| {
384                    collector
385                        .into_iter()
386                        .map(move |collector| (family as u32, collector))
387                })
388                .collect::<Vec<_>>();
389            self.parallel_scheduler.try_parallel_for_each_owned(
390                to_flush,
391                |(family, mut collector)| {
392                    self.flush_thread_local_collector(family, &mut collector)?;
393                    drop(collector);
394                    anyhow::Ok(())
395                },
396            )?;
397        }
398
399        let _span = tracing::trace_span!("flush collectors").entered();
400
401        // Now we reduce the global collectors in parallel
402        let mut new_sst_files = take(self.new_sst_files.get_mut());
403        let shared_new_sst_files = Mutex::new(&mut new_sst_files);
404
405        let new_collectors =
406            [(); FAMILIES].map(|_| Mutex::new(GlobalCollectorState::Unsharded(Collector::new())));
407        let collectors = replace(&mut self.collectors, new_collectors);
408        let collectors = collectors
409            .into_iter()
410            .enumerate()
411            .flat_map(|(family, state)| {
412                let collector = state.into_inner();
413                match collector {
414                    GlobalCollectorState::Unsharded(collector) => {
415                        Either::Left([(family, collector)].into_iter())
416                    }
417                    GlobalCollectorState::Sharded(shards) => {
418                        Either::Right(shards.into_iter().map(move |collector| (family, collector)))
419                    }
420                }
421            })
422            .collect::<Vec<_>>();
423        self.parallel_scheduler.try_parallel_for_each_owned(
424            collectors,
425            |(family, mut collector)| {
426                let family = family as u32;
427                if !collector.is_empty() {
428                    let sst = self.create_sst_file(
429                        family,
430                        collector.sorted(self.family_configs[usize_from_u32(family)].kind),
431                    )?;
432                    collector.clear();
433                    drop(collector);
434                    shared_new_sst_files.lock().push(sst);
435                }
436                anyhow::Ok(())
437            },
438        )?;
439
440        // Now we need to write the new meta files.
441        let new_meta_collectors = [(); FAMILIES].map(|_| Mutex::new(Vec::new()));
442        let meta_collectors = replace(&mut self.meta_collectors, new_meta_collectors);
443        let keys_written = AtomicU64::new(0);
444        let file_to_write = meta_collectors
445            .into_iter()
446            .map(|mutex| mutex.into_inner())
447            .enumerate()
448            .filter(|(_, sst_files)| !sst_files.is_empty())
449            .collect::<Vec<_>>();
450        let new_meta_files = self
451            .parallel_scheduler
452            .parallel_map_collect_owned::<_, _, Result<Vec<_>>>(
453                file_to_write,
454                |(family, sst_files)| {
455                    let family = family as u32;
456                    let mut entries = 0;
457                    let mut builder = MetaFileBuilder::new(
458                        family,
459                        self.family_configs[usize_from_u32(family)].compression,
460                    );
461                    for (seq, sst) in sst_files {
462                        entries += sst.entries;
463                        builder.add(seq, sst);
464                    }
465                    keys_written.fetch_add(entries, Ordering::Relaxed);
466                    let accessed_key_hashes = get_accessed_key_hashes(family);
467                    builder.set_used_key_hashes_amqf(accessed_key_hashes);
468                    let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
469                    let (file, size) = builder.write(&self.db_path, seq)?;
470                    Ok(NewFile { seq, file, size })
471                },
472            )?;
473
474        // Finally we return the new files and sequence number.
475        let seq = self.current_sequence_number.load(Ordering::SeqCst);
476        Ok(FinishResult {
477            sequence_number: seq,
478            new_meta_files,
479            new_sst_files,
480            new_blob_files,
481            keys_written: keys_written.into_inner(),
482        })
483    }
484
485    /// Creates a new blob file with the given value.
486    #[tracing::instrument(level = "trace", skip(self, value), fields(value_len = value.len()))]
487    fn create_blob(&self, family: u32, value: &[u8]) -> Result<NewFile> {
488        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
489        let mut compressed = Vec::new();
490        let compression = self.family_configs[usize_from_u32(family)].compression;
491        Compressor::new(compression)?
492            .compress_into_buffer(value, &mut compressed)
493            .context("Compression of value for blob file failed")?;
494
495        let mut buffer = Vec::with_capacity(8 + compressed.len());
496        buffer.write_u32::<BE>(value.len() as u32)?;
497        buffer.write_u32::<BE>(checksum_block(&compressed))?;
498        buffer.extend_from_slice(&compressed);
499
500        let size = buffer.len() as u64;
501        let file = self.db_path.join(format!("{seq:08}.blob"));
502        let mut file = File::create(&file)?;
503        file.write_all(&buffer)?;
504        file.flush()?;
505        Ok(NewFile { seq, file, size })
506    }
507
508    /// Creates a new SST file with the given collector data.
509    #[tracing::instrument(level = "trace", skip(self, collector_data), fields(family_name = self.family_configs[usize_from_u32(family)].name))]
510    fn create_sst_file(
511        &self,
512        family: u32,
513        collector_data: (&[CollectorEntry<K>], usize),
514    ) -> Result<NewFile> {
515        let (entries, _total_key_size) = collector_data;
516        let seq = self.current_sequence_number.fetch_add(1, Ordering::SeqCst) + 1;
517
518        let path = self.db_path.join(format!("{seq:08}.sst"));
519        let (meta, file) = self
520            .parallel_scheduler
521            .block_in_place(|| {
522                write_static_stored_file(
523                    entries,
524                    &path,
525                    MetaEntryFlags::FRESH,
526                    self.family_configs[usize_from_u32(family)].compression,
527                )
528            })
529            .with_context(|| format!("Unable to write SST file {seq:08}.sst"))?;
530
531        #[cfg(feature = "verify_sst_content")]
532        {
533            use core::panic;
534
535            use crate::{
536                collector_entry::CollectorEntryValue,
537                key::hash_key,
538                lookup_entry::LookupValue,
539                static_sorted_file::{
540                    BlockCache, SstLookupResult, StaticSortedFile, StaticSortedFileMetaData,
541                },
542                static_sorted_file_builder::Entry,
543            };
544
545            file.sync_all()?;
546            let sst = StaticSortedFile::open(
547                &self.db_path,
548                StaticSortedFileMetaData {
549                    sequence_number: seq,
550                    block_count: meta.block_count,
551                },
552                self.family_configs[usize_from_u32(family)].compression,
553                crate::mmap_access_mode(),
554            )?;
555            let cache2 = BlockCache::with(
556                10,
557                u64::MAX,
558                Default::default(),
559                Default::default(),
560                Default::default(),
561            );
562            let cache3 = BlockCache::with(
563                10,
564                u64::MAX,
565                Default::default(),
566                Default::default(),
567                Default::default(),
568            );
569            let mut key_buf = Vec::new();
570            let family_config = self.family_configs[usize_from_u32(family)].kind;
571            for entry in entries {
572                entry.write_key_to(&mut key_buf);
573                let result = sst
574                    .lookup::<_, true>(hash_key(&key_buf), &key_buf, &cache2, &cache3)
575                    .expect("key found");
576                key_buf.clear();
577                match result {
578                    SstLookupResult::Found(values) => {
579                        if values.len() > 1 {
580                            use crate::FamilyKind;
581
582                            assert!(
583                                values.len() == 1 || family_config == FamilyKind::MultiValue,
584                                "only multi-value tables can have more than one value, got {} \
585                                 values",
586                                values.len()
587                            )
588                        }
589                        match &entry.value {
590                            CollectorEntryValue::Large { blob } => {
591                                assert!(
592                                    values.contains(&LookupValue::Blob {
593                                        sequence_number: *blob
594                                    }),
595                                    "we wrote a blob but did not read it"
596                                );
597                            }
598                            // Key tombstones sort last within a key group, so a same-batch
599                            // `put(K, v); delete(K)` reads back as [v, KeyDeleted].
600                            CollectorEntryValue::KeyDeleted => assert!(
601                                values.last() == Some(&LookupValue::KeyDeleted),
602                                "we wrote a key tombstone but it was not last in results"
603                            ),
604                            CollectorEntryValue::KeyValueDeleted { value, len } => {
605                                let expected = &value[..*len as usize];
606                                assert!(
607                                    values.iter().any(|lv| matches!(
608                                        lv,
609                                        LookupValue::KeyValueDeleted { value } if &**value == expected
610                                    )),
611                                    "we wrote a key-value tombstone but did not read it back"
612                                )
613                            }
614                            v => {
615                                assert!(
616                                    values.into_iter().any(|lv| {
617                                        if let LookupValue::Slice { value } = lv {
618                                            &*value == v.as_bytes().unwrap()
619                                        } else {
620                                            false
621                                        }
622                                    }),
623                                    "we wrote a slice of bytes but did not read it"
624                                )
625                            }
626                        }
627                    }
628                    SstLookupResult::NotFound => panic!("All keys must exist"),
629                }
630            }
631        }
632
633        let size = meta.size;
634        self.meta_collectors[usize_from_u32(family)]
635            .lock()
636            .push((seq, meta));
637
638        Ok(NewFile { seq, file, size })
639    }
640}
641
642#[inline(always)]
643const fn usize_from_u32(value: u32) -> usize {
644    // This should always be true, as we assume at least a 32-bit width architecture for Turbopack.
645    // Since this is a const expression, we expect it to be compiled away.
646    const {
647        assert!(u32::BITS <= usize::BITS);
648    };
649    value as usize
650}
turbo_persistence/write_batch.rs

turbo_persistence/
write_batch.rs
