coordinode-lsm-tree 7.4.2

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2024-present, fjall-rs
// Copyright (c) 2026-present, Dmitry Prudnikov

use super::writer::Writer;
use crate::fs::FsFile;
use crate::path::{Path, PathBuf};
use crate::{
    BlobFile, CompressionType, DescriptorTable, SeqNo, SequenceNumberCounter, TreeId,
    file_accessor::FileAccessor,
    fs::{Fs, SyncMode, WriteOrigin},
    vlog::{
        ValueHandle,
        blob_file::{Inner as BlobFileInner, Metadata},
    },
};
use alloc::sync::Arc;
#[cfg(not(feature = "std"))]
use alloc::{string::ToString, vec::Vec};
use core::sync::atomic::AtomicBool;

#[cfg(test)]
mod tests;

/// Blob file writer, may write multiple blob files
pub(crate) struct MultiWriter {
    fs: Arc<dyn Fs>,

    folder: PathBuf,
    target_size: u64,

    active_writer: Writer,

    /// The lifetime class the active writer's file holds.
    active_class: u8,

    /// Writers for the other outputs a write has met, each still filling its
    /// own file while the active one takes frames of its codec and class.
    ///
    /// A relocation merges its sources by key, so two compression generations
    /// over the same keys alternate on nearly every frame, and values of
    /// different lifetime classes interleave the same way in any write. A file
    /// records one codec and one class, and closing it on every switch would
    /// produce a file per value; keeping one open per pair leaves every file
    /// as full as the size target allows. At most one entry per codec the
    /// sources carry times [`Self::lifetime_groups`].
    parked: Vec<(CompressionType, u8, Writer)>,

    /// How many lifetime classes the files are split into.
    lifetime_groups: crate::config::LifetimeGroups,

    results: Vec<BlobFile>,

    id_generator: SequenceNumberCounter,

    compression: CompressionType,
    passthrough_compression: CompressionType,

    /// Durability level wired from `Config::sync_mode`, stamped on every
    /// rotated blob writer.
    sync_mode: SyncMode,

    /// Whether the files and their folder are synced at finish, stamped on
    /// every rotated blob writer.
    durability: crate::config::FlushDurability,

    /// `Config::writeback_bytes`, stamped on every rotated blob writer.
    writeback_bytes: u64,

    /// Dictionary for `ZstdDict` compression, shared across all rotated writers.
    #[cfg(zstd_any)]
    zstd_dictionary: Option<alloc::sync::Arc<crate::compression::ZstdDictionary>>,

    /// Whether zstd levels 19-22 run the `btultra2` two-pass seed, shared
    /// across all rotated writers so one run is uniform.
    #[cfg(zstd_any)]
    zstd_two_pass_seed: bool,

    /// The tree's dictionary set, used to PIN the dictionary each finished file
    /// records on its handle (see `blob_file::Inner::zstd_dictionary`).
    ///
    /// Not the same question as [`Self::zstd_dictionary`], which is what this
    /// writer COMPRESSES with. A relocation pass compresses with nothing (it
    /// copies frames verbatim) yet still records a codec, so the file it
    /// produces needs a dictionary its own descriptor names and the write slot
    /// cannot answer.
    #[cfg(zstd_any)]
    zstd_dictionaries: crate::compression::ZstdDictionaries,

    tree_id: TreeId,
    descriptor_table: Option<Arc<DescriptorTable>>,

    /// The tree's clock, handed to every blob file writer.
    clock: Option<Arc<dyn crate::Clock>>,

    /// Where a compaction learns of every blob file this writer creates, so
    /// it can remove the ones it never installs. `None` outside compaction.
    outputs: Option<crate::compaction::output_ledger::OutputLedger>,

    /// How many of `results`, the first ones, have landed in `outputs`.
    landed: usize,

    /// The key domains no blob file spans, the current one's, and the
    /// comparator they are intervals of.
    domain: Option<(crate::key_domain::DomainCursor, crate::SharedComparator)>,

    /// What the blob files are written for.
    origin: Option<WriteOrigin>,
}

impl MultiWriter {
    /// A blob file writer whose files carry no origin.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    #[cfg(test)]
    pub(crate) fn new<P: AsRef<Path>>(
        id_generator: SequenceNumberCounter,
        folder: P,
        tree_id: TreeId,
        descriptor_table: Option<Arc<DescriptorTable>>,
        fs: Arc<dyn Fs>,
    ) -> crate::Result<Self> {
        Self::open(id_generator, folder, tree_id, descriptor_table, fs, None)
    }

    /// Initializes a new blob file writer, its blob files written as
    /// `origin`'s.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    pub(crate) fn new_for<P: AsRef<Path>>(
        id_generator: SequenceNumberCounter,
        folder: P,
        tree_id: TreeId,
        descriptor_table: Option<Arc<DescriptorTable>>,
        fs: Arc<dyn Fs>,
        origin: WriteOrigin,
    ) -> crate::Result<Self> {
        Self::open(
            id_generator,
            folder,
            tree_id,
            descriptor_table,
            fs,
            Some(origin),
        )
    }

    fn open<P: AsRef<Path>>(
        id_generator: SequenceNumberCounter,
        folder: P,
        tree_id: TreeId,
        descriptor_table: Option<Arc<DescriptorTable>>,
        fs: Arc<dyn Fs>,
        origin: Option<WriteOrigin>,
    ) -> crate::Result<Self> {
        let folder = folder.as_ref();

        let blob_file_id = id_generator.next();
        let blob_file_path = folder.join(blob_file_id.to_string());

        Ok(Self {
            id_generator,
            folder: folder.into(),
            target_size: 64 * 1_024 * 1_024,

            active_writer: Writer::open(blob_file_path, blob_file_id, tree_id, &*fs, origin)?,
            origin,
            active_class: 0,
            parked: Vec::new(),
            lifetime_groups: crate::config::LifetimeGroups::ONE,

            results: Vec::new(),

            compression: CompressionType::None,
            passthrough_compression: CompressionType::None,
            sync_mode: SyncMode::Normal,
            durability: crate::config::FlushDurability::Synced,
            writeback_bytes: 0,

            #[cfg(zstd_any)]
            zstd_dictionary: None,
            #[cfg(zstd_any)]
            zstd_two_pass_seed: true,
            #[cfg(zstd_any)]
            zstd_dictionaries: crate::compression::ZstdDictionaries::new(),

            tree_id,
            descriptor_table,
            clock: None,
            fs,
            outputs: None,
            landed: 0,
            domain: None,
        })
    }

    /// Closes every file this writer holds open before a value whose key lies
    /// past the key domain of the values written so far, so no blob file holds
    /// values of two of `domains`, intervals of `comparator`'s order: expiring
    /// a domain leaves its blob files wholly dead (see [`crate::KeyDomains`]).
    #[must_use]
    pub(crate) fn use_key_domains(
        mut self,
        domains: Option<Arc<dyn crate::KeyDomains>>,
        comparator: crate::SharedComparator,
    ) -> Self {
        self.domain = domains.map(|d| (crate::key_domain::DomainCursor::new(d), comparator));
        self
    }

    /// Takes `key`, the key of the value about to be written, into the
    /// current domain, closing every open file first when it lies past it.
    fn enter_key(&mut self, key: &[u8]) -> crate::Result<()> {
        let Some((domain, comparator)) = &mut self.domain else {
            return Ok(());
        };
        let crossed = domain.crosses(key, comparator.as_ref()).is_some();
        domain.enter(key, crossed);
        if crossed {
            for (codec, _, writer) in core::mem::take(&mut self.parked) {
                self.land(writer, codec)?;
            }
            if self.active_writer.item_count > 0 {
                self.rotate()?;
            }
        }
        Ok(())
    }

    /// Records the blob file already open and every one this writer creates
    /// after it in `outputs`.
    #[must_use]
    pub(crate) fn use_output_ledger(
        mut self,
        outputs: crate::compaction::output_ledger::OutputLedger,
    ) -> Self {
        outputs.record_blob_file(
            self.active_writer.blob_file_id,
            self.active_writer.path.clone(),
            self.fs.clone(),
        );
        self.outputs = Some(outputs);
        self
    }

    /// Wires the tree's `Config::sync_mode` through to every blob file this
    /// writer finalizes (active + rotated).
    #[must_use]
    pub(crate) fn use_sync_mode(mut self, sync_mode: SyncMode) -> Self {
        self.sync_mode = sync_mode;
        self.active_writer.sync_mode = sync_mode;
        self
    }

    /// Leaves every blob file this writer finishes, and their folder,
    /// unsynced under
    /// [`FlushDurability::Deferred`](crate::config::FlushDurability::Deferred),
    /// for the install that persists them to sync.
    #[must_use]
    pub(crate) fn use_durability(mut self, durability: crate::config::FlushDurability) -> Self {
        self.durability = durability;
        self.active_writer.durability = durability;
        self
    }

    /// Reads the time on the tree's `clock` in every blob file written.
    #[must_use]
    pub(crate) fn use_clock(mut self, clock: Option<Arc<dyn crate::Clock>>) -> Self {
        self.active_writer = self.active_writer.use_clock(clock.clone());
        self.clock = clock;
        self
    }

    /// Wires the tree's `Config::writeback_bytes` through to every blob file
    /// this writer writes (active + rotated).
    #[must_use]
    pub(crate) fn use_writeback_bytes(mut self, bytes: u64) -> Self {
        self.writeback_bytes = bytes;
        self.active_writer = self.active_writer.use_writeback_bytes(bytes);
        self
    }

    /// Sets the blob file target size.
    #[must_use]
    pub(crate) fn use_target_size(mut self, bytes: u64) -> Self {
        self.target_size = bytes;
        self
    }

    /// Sets the compression method in blob file writer metadata, but does not actually compress blobs.
    ///
    /// This is used in garbage collection to pass through already-compressed blobs, but correctly
    /// set the compression type in the metadata.
    pub(crate) fn use_passthrough_compression(mut self, compression: CompressionType) -> Self {
        assert_eq!(self.compression, CompressionType::None);
        self.passthrough_compression = compression;
        // The bytes handed to `write_raw` are already compressed, so the writer
        // keeps `compression = None` (no re-compress) but must RECORD the real
        // codec in the file metadata or a reopened reader cannot decode it.
        self.active_writer.metadata_compression_override = Some(compression);
        self
    }

    /// Splits the files this writer produces into `groups` lifetime classes,
    /// the files written so far taking the class of a fresh value.
    #[must_use]
    pub(crate) fn use_lifetime_groups(mut self, groups: crate::config::LifetimeGroups) -> Self {
        self.lifetime_groups = groups;
        let class = groups.fresh();
        self.active_class = class;
        self.active_writer
            .set_lifetime_class(class, groups.write_lifetime(class));
        self
    }

    /// The lifetime classes this writer splits its files into.
    pub(crate) fn lifetime_groups(&self) -> crate::config::LifetimeGroups {
        self.lifetime_groups
    }

    /// Directs the next frames to a file of lifetime `class` that records the
    /// codec of the frames written so far. `class` past the last group goes to
    /// the last group.
    ///
    /// # Errors
    ///
    /// Propagates the creation of the first file for a class not met before.
    pub(crate) fn select_lifetime_class(&mut self, class: u8) -> crate::Result<()> {
        self.select_output(self.passthrough_compression, class)
    }

    /// Directs the next frames to a file that records `compression` and holds
    /// lifetime `class`.
    ///
    /// The relocation counterpart of [`Self::use_passthrough_compression`],
    /// which fixes one codec for the whole pass. A relocation copies frames
    /// VERBATIM out of sources that need not share a codec (the blob policy may
    /// have moved since they were written), while a blob file records exactly
    /// one. Each codec therefore gets its own output file, and so does each
    /// lifetime class; switching parks the current one rather than closing it
    /// (see [`Self::parked`]).
    ///
    /// # Errors
    ///
    /// Propagates the creation of the first file for a pair not met before.
    pub(crate) fn select_output(
        &mut self,
        compression: CompressionType,
        class: u8,
    ) -> crate::Result<()> {
        let class = self.lifetime_groups.cap(class);
        if self.passthrough_compression == compression && self.active_class == class {
            return Ok(());
        }
        let next = if let Some(at) = self
            .parked
            .iter()
            .position(|(c, k, _)| *c == compression && *k == class)
        {
            self.parked.swap_remove(at).2
        } else if self.active_writer.item_count == 0 {
            // Nothing written constrains an untouched writer: restamp it.
            // Only a codec switch restamps the codec: a class switch keeps what
            // the writer records, which for a compressing writer is its own.
            if compression != self.passthrough_compression {
                self.active_writer.metadata_compression_override = Some(compression);
            }
            self.active_writer
                .set_lifetime_class(class, self.lifetime_groups.write_lifetime(class));
            self.passthrough_compression = compression;
            self.active_class = class;
            return Ok(());
        } else {
            self.fresh_writer(compression, class)?
        };
        let previous = core::mem::replace(&mut self.active_writer, next);
        self.parked
            .push((self.passthrough_compression, self.active_class, previous));
        self.passthrough_compression = compression;
        self.active_class = class;
        Ok(())
    }

    /// A new writer for the next blob file of lifetime `class`, recording
    /// `passthrough` when this writer copies already-encoded frames.
    fn fresh_writer(&self, passthrough: CompressionType, class: u8) -> crate::Result<Writer> {
        let id = self.id_generator.next();
        let path = self.folder.join(id.to_string());
        let w = Writer::open(path, id, self.tree_id, &*self.fs, self.origin)?;
        // Recorded once it exists: a failed constructor created nothing ours.
        if let Some(outputs) = &self.outputs {
            outputs.record_blob_file(id, w.path.clone(), self.fs.clone());
        }
        let mut w = w
            .use_compression(self.compression)
            .use_sync_mode(self.sync_mode)
            .use_writeback_bytes(self.writeback_bytes)
            .use_clock(self.clock.clone());
        w.durability = self.durability;
        // Carry the passthrough metadata codec onto the new writer so every
        // file in a relocation records the real compression.
        if passthrough != CompressionType::None {
            w.metadata_compression_override = Some(passthrough);
        }
        w.set_lifetime_class(class, self.lifetime_groups.write_lifetime(class));
        #[cfg(zstd_any)]
        let w = w
            .use_zstd_dictionary(self.zstd_dictionary.clone())
            .use_zstd_two_pass_seed(self.zstd_two_pass_seed);
        Ok(w)
    }

    /// Sets the compression method.
    #[must_use]
    #[doc(hidden)]
    pub(crate) fn use_compression(mut self, compression: CompressionType) -> Self {
        self.compression.clone_from(&compression);
        self.active_writer.compression = compression;
        self
    }

    /// Provides the zstd dictionary for [`CompressionType::ZstdDict`] writes.
    ///
    /// The dictionary is propagated to every rotated writer so that all blob
    /// files produced by this `MultiWriter` use the same dictionary.
    #[cfg(zstd_any)]
    #[must_use]
    pub(crate) fn use_zstd_dictionary(
        mut self,
        dict: Option<alloc::sync::Arc<crate::compression::ZstdDictionary>>,
    ) -> Self {
        self.active_writer = self.active_writer.use_zstd_dictionary(dict.clone());
        self.zstd_dictionary = dict;
        self
    }

    /// Selects whether zstd levels 19-22 run the `btultra2` two-pass seed,
    /// propagated to every rotated writer so all blob files of one run are
    /// written under the same strategy.
    ///
    /// See [`crate::runtime_config::RuntimeConfig::zstd_two_pass_seed`].
    #[cfg(zstd_any)]
    #[must_use]
    pub(crate) fn use_zstd_two_pass_seed(mut self, enabled: bool) -> Self {
        self.active_writer = self.active_writer.use_zstd_two_pass_seed(enabled);
        self.zstd_two_pass_seed = enabled;
        self
    }

    /// Provides the tree's dictionary set, so each finished file can pin the
    /// dictionary its own recorded descriptor names.
    ///
    /// Required wherever the files this writer produces are `ZstdDict`, whether
    /// this writer compresses them itself or passes already-compressed frames
    /// through: the handle a reader gets must carry its own dictionary.
    #[cfg(zstd_any)]
    #[must_use]
    pub(crate) fn use_zstd_dictionaries(
        mut self,
        dicts: crate::compression::ZstdDictionaries,
    ) -> Self {
        self.zstd_dictionaries = dicts;
        self
    }

    /// Sets up a new writer for the next blob file.
    fn rotate(&mut self) -> crate::Result<()> {
        log::debug!("Rotating blob file writer");

        let new_writer = self.fresh_writer(self.passthrough_compression, self.active_class)?;

        let old_writer = core::mem::replace(&mut self.active_writer, new_writer);
        self.land(old_writer, self.passthrough_compression)
    }

    /// Finishes `writer`, whose file records `passthrough`, into the results.
    fn land(&mut self, writer: Writer, passthrough: CompressionType) -> crate::Result<()> {
        let blob_file = Self::consume_writer(
            writer,
            passthrough,
            self.descriptor_table.clone(),
            &self.fs,
            #[cfg(zstd_any)]
            &self.zstd_dictionaries,
        )?;
        // Before the next file takes more room.
        if let (Some(outputs), Some(file)) = (&self.outputs, &blob_file) {
            outputs.landed(&*self.fs, file.path())?;
            self.landed += 1;
        }
        self.results.extend(blob_file);

        Ok(())
    }

    fn consume_writer(
        writer: Writer,
        passthrough_compression: CompressionType,
        descriptor_table: Option<Arc<DescriptorTable>>,
        fs: &Arc<dyn Fs>,
        #[cfg(zstd_any)] zstd_dictionaries: &crate::compression::ZstdDictionaries,
    ) -> crate::Result<Option<BlobFile>> {
        if writer.item_count > 0 {
            let blob_file_id = writer.blob_file_id;
            let path = writer.path.clone();

            log::debug!(
                "Created blob file #{blob_file_id:?} ({} items, {} userdata bytes)",
                writer.item_count,
                writer.uncompressed_bytes,
            );

            let tree_id = writer.tree_id;
            // Taken before `finish` consumes the writer.
            #[cfg(zstd_any)]
            let writer_dictionary = writer.zstd_dictionary.clone();

            let (metadata, checksum) = writer.finish()?;

            // What the FILE will record, which is what its reader resolves
            // against: the passthrough codec when relocation is stamping the
            // source's own, else what this writer compressed with.
            let recorded_compression = if passthrough_compression == CompressionType::None {
                metadata.compression
            } else {
                passthrough_compression
            };

            // Resolved BEFORE the file is opened and its descriptor published,
            // because those are side effects only the finished handle's `Drop`
            // knows how to undo. Failing between them would strand both: an
            // orphan file no version names, and an open descriptor in the
            // shared table for the life of the process.
            //
            // The set answers first, then the dictionary this writer compressed
            // with when THAT is the one the file records: a writer holding the
            // matching dictionary needs no set to pin it, since it just used
            // those bytes. The set is what answers for a relocation, which
            // compresses nothing and records the source's codec.
            #[cfg(zstd_any)]
            let zstd_dictionary = match zstd_dictionaries.for_compression(recorded_compression) {
                Ok(dict) => dict,
                Err(e) => match (&recorded_compression, &writer_dictionary) {
                    (CompressionType::ZstdDict { dict_id, .. }, Some(d)) if d.id() == *dict_id => {
                        Some(alloc::sync::Arc::clone(d))
                    }
                    _ => {
                        // No handle exists yet, so no `Drop` will reclaim this;
                        // the unlink has to happen here.
                        if let Err(remove) = fs.remove_file(&path) {
                            log::warn!(
                                "Could not delete unusable blob file at {}: {remove:?}",
                                path.display(),
                            );
                        }
                        return Err(e);
                    }
                },
            };

            let file: Arc<dyn FsFile> =
                Arc::from(fs.open(&path, &crate::fs::FsOpenOptions::new().read(true))?);
            let file_accessor = if let Some(dt) = descriptor_table {
                FileAccessor::DescriptorTable {
                    table: dt,
                    fs: fs.clone(),
                }
            } else {
                FileAccessor::File(file.clone())
            };
            file_accessor.insert_for_blob_file((tree_id, blob_file_id).into(), file);

            let blob_file = BlobFile(Arc::new(BlobFileInner {
                checksum,
                tree_id,
                path,
                is_deleted: AtomicBool::new(false),
                punch_on_drop: portable_atomic::AtomicU64::new(u64::MAX),
                punched_through: once_cell::race::OnceBox::new(),
                // A freshly written file is whole: nothing has been reclaimed.
                live_data_start: 0,
                id: blob_file_id,
                file_accessor,
                #[cfg(zstd_any)]
                zstd_dictionary,
                meta: Metadata {
                    id: blob_file_id,
                    version: metadata.version,
                    created_at: metadata.created_at,
                    item_count: metadata.item_count,
                    total_compressed_bytes: metadata.total_compressed_bytes,
                    total_uncompressed_bytes: metadata.total_uncompressed_bytes,
                    key_range: metadata.key_range,

                    compression: recorded_compression,
                    lifetime_class: metadata.lifetime_class,
                },
                fs: fs.clone(),
                deletion_pause: once_cell::race::OnceBox::new(),

                #[cfg(feature = "std")]
                background_deleter: once_cell::race::OnceBox::new(),
            }));

            Ok(Some(blob_file))
        } else {
            log::debug!(
                "Blob file writer at {} has written no data, deleting empty blob file",
                writer.path.display(),
            );

            if let Err(e) = fs.remove_file(&writer.path) {
                log::warn!(
                    "Could not delete empty blob file at {}: {e:?}",
                    writer.path.display(),
                );
            }

            Ok(None)
        }
    }

    /// Writes an item.
    ///
    /// Returns the [`ValueHandle`] of the written blob.
    ///
    /// # Errors
    ///
    /// Will return `Err` if an IO error occurs.
    pub(crate) fn write(
        &mut self,
        key: &[u8],
        seqno: SeqNo,
        value: &[u8],
    ) -> crate::Result<ValueHandle> {
        self.enter_key(key)?;
        let target_size = self.target_size;

        // Write actual value into blob file
        let writer = &mut self.active_writer;

        let offset = writer.offset();
        let on_disk_value_len = writer.write(key, seqno, value)?;

        let handle = ValueHandle {
            blob_file_id: writer.blob_file_id(),
            offset,
            on_disk_size: on_disk_value_len,
        };

        // Check for blob file size target, maybe rotate to next writer
        if writer.offset() >= target_size {
            self.rotate()?;
        }

        Ok(handle)
    }

    pub(crate) fn write_raw(
        &mut self,
        key: &[u8],
        seqno: SeqNo,
        value: &[u8],
        uncompressed_len: u32,
    ) -> crate::Result<ValueHandle> {
        self.enter_key(key)?;
        let target_size = self.target_size;

        // Write actual value into blob file
        let writer = &mut self.active_writer;

        let offset = writer.offset();
        let on_disk_value_len = writer.write_raw(key, seqno, value, uncompressed_len)?;

        let handle = ValueHandle {
            blob_file_id: writer.blob_file_id(),
            offset,
            on_disk_size: on_disk_value_len,
        };

        // Check for blob file size target, maybe rotate to next writer
        if writer.offset() >= target_size {
            self.rotate()?;
        }

        Ok(handle)
    }

    /// Writes `value`, a value new to the value log, encoded under
    /// `compression` into a file of lifetime `class` that records that codec:
    /// the file a relocation copies frames of the same codec into, so a pass
    /// that both relocates and separates new values cuts one set of files.
    ///
    /// # Errors
    ///
    /// Propagates the encoding (a dictionary codec without the dictionary this
    /// writer holds) and the write.
    pub(crate) fn write_fresh(
        &mut self,
        key: &[u8],
        seqno: SeqNo,
        value: &[u8],
        compression: CompressionType,
        class: u8,
    ) -> crate::Result<ValueHandle> {
        super::writer::check_size_cap(value.len())?;
        let encoded = super::writer::encode_value(
            compression,
            #[cfg(zstd_any)]
            self.zstd_dictionary.as_deref(),
            #[cfg(zstd_any)]
            self.zstd_two_pass_seed,
            value,
        )?;
        self.enter_key(key)?;
        self.select_output(compression, class)?;

        let writer = &mut self.active_writer;
        let offset = writer.offset();
        #[expect(
            clippy::cast_possible_truncation,
            reason = "the size cap above bounds the value far below u32::MAX"
        )]
        let on_disk_size = writer.write_encoded(key, seqno, &encoded, value.len() as u32)?;
        let handle = ValueHandle {
            blob_file_id: writer.blob_file_id(),
            offset,
            on_disk_size,
        };
        if writer.offset() >= self.target_size {
            self.rotate()?;
        }
        Ok(handle)
    }

    pub(crate) fn finish(mut self) -> crate::Result<Vec<BlobFile>> {
        let mut writers = core::iter::once((self.passthrough_compression, self.active_writer))
            .chain(
                self.parked
                    .into_iter()
                    .map(|(codec, _, writer)| (codec, writer)),
            );
        while let Some((passthrough, writer)) = writers.next() {
            match Self::consume_writer(
                writer,
                passthrough,
                self.descriptor_table.clone(),
                &self.fs,
                #[cfg(zstd_any)]
                &self.zstd_dictionaries,
            ) {
                Ok(blob_file) => self.results.extend(blob_file),
                Err(e) => {
                    // The write as a whole failed, so no version will ever name
                    // the files finished so far: marked deleted, their handles'
                    // drop unlinks them and evicts their descriptors. The
                    // writers not reached yet never produced a handle, so their
                    // files are removed here.
                    for file in &self.results {
                        file.mark_as_deleted();
                    }
                    for (_, unfinished) in writers {
                        let path = unfinished.path.clone();
                        drop(unfinished);
                        if let Err(rm) = self.fs.remove_file(&path) {
                            log::warn!(
                                "Could not delete unfinished blob file at {}: {rm:?}",
                                path.display(),
                            );
                        }
                    }
                    return Err(e);
                }
            }
        }
        // A new file's directory entry is durable only once its directory is
        // synced (POSIX `fsync(2)`), and the manifest edit that names these
        // files follows: synced once here for every file the write made. The
        // manifest is in the folder above; on another device (a nested mount)
        // a barrier would not order this sync before it, so that is in full.
        // A deferred write leaves this to the install that persists it.
        let mode = crate::file::folder_sync_mode(&*self.fs, &self.folder, self.sync_mode);
        if !self.results.is_empty()
            && self.durability == crate::config::FlushDurability::Synced
            && let Err(e) = crate::file::fsync_directory(&self.folder, &*self.fs, mode)
        {
            // Not durable, so no version may name them.
            for file in &self.results {
                file.mark_as_deleted();
            }
            return Err(e.into());
        }
        // The files the rotations finished have landed already; the rest land
        // now. A run stopped here removes them with the rest of its outputs.
        if let Some(outputs) = &self.outputs {
            for file in self.results.iter().skip(self.landed) {
                outputs.landed(&*self.fs, file.path())?;
            }
        }
        Ok(self.results)
    }
}