fxfs/object_store/

journal.rs

// Copyright 2021 The Fuchsia Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

//! The journal is implemented as an ever extending file which contains variable length records
//! that describe mutations to be applied to various objects.  The journal file consists of
//! blocks, with a checksum at the end of each block, but otherwise it can be considered a
//! continuous stream.
//!
//! The checksum is seeded with the checksum from the previous block.  To free space in the
//! journal, records are replaced with sparse extents when it is known they are no longer
//! needed to mount.
//!
//! At mount time, the journal is replayed: the mutations are applied into memory.
//! Eventually, a checksum failure will indicate no more records exist to be replayed,
//! at which point the mount can continue and the journal will be extended from that point with
//! further mutations as required.

mod bootstrap_handle;
mod checksum_list;
mod reader;
pub mod super_block;
mod writer;

use crate::checksum::{Checksum, Checksums};
use crate::errors::FxfsError;
use crate::filesystem::{
    ApplyContext, ApplyMode, FlushReason, ForceMajor, FxFilesystem, SyncOptions,
};
use crate::hooks::HooksHandle;
use crate::log::*;
use crate::lsm_tree::types::LayerIterator;
use crate::object_handle::{ObjectHandle as _, ReadObjectHandle};
use crate::object_store::allocator::Allocator;
use crate::object_store::data_object_handle::OverwriteOptions;
use crate::object_store::extent_record::{ExtentMode, ExtentValue};
use crate::object_store::graveyard::Graveyard;
use crate::object_store::journal::bootstrap_handle::BootstrapObjectHandle;
use crate::object_store::journal::checksum_list::ChecksumList;
use crate::object_store::journal::reader::{JournalReader, ReadResult};
use crate::object_store::journal::super_block::{
    SuperBlockHeader, SuperBlockInstance, SuperBlockManager,
};
use crate::object_store::journal::writer::JournalWriter;
use crate::object_store::object_manager::ObjectManager;
use crate::object_store::object_record::{AttributeKey, ObjectKey, ObjectKeyData, ObjectValue};
use crate::object_store::transaction::{
    AllocatorMutation, LockKey, Mutation, MutationV56, MutationV57, MutationV59, MutationV61,
    MutationV62, ObjectMutationIterator, ObjectStoreMutation, Options, ReservationOptions,
    TRANSACTION_MAX_JOURNAL_USAGE, Transaction, lock_keys,
};
use crate::object_store::{
    AssocObj, AttributeId, DataObjectHandle, Extent, HandleOptions, HandleOwner, INVALID_OBJECT_ID,
    Item, ItemRef, NewChildStoreOptions, ObjectStore, ReservedId,
};
use crate::range::RangeExt;
use crate::round::round_div;
use crate::serialized_types::{LATEST_VERSION, Migrate, Version, Versioned, migrate_to_version};
use anyhow::{Context, Error, anyhow, bail, ensure};
use core::iter::Iterator;
use event_listener::Event;
use fprint::TypeFingerprint;
use fuchsia_inspect::NumericProperty;
use fuchsia_sync::Mutex;
use futures::FutureExt as _;
use futures::future::poll_fn;
use rustc_hash::FxHashMap as HashMap;
use serde::{Deserialize, Serialize};
use smallvec::SmallVec;
use static_assertions::const_assert;
use std::clone::Clone;
use std::collections::HashSet;
use std::num::NonZero;
use std::ops::{Bound, Range};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, OnceLock};
use std::task::{Poll, Waker};
use storage_device::Device;
use storage_units::BlockSize;

// The journal file is written to in blocks of this size.
pub const BLOCK_SIZE: BlockSize = BlockSize::SIZE_4KIB;

// The journal file is extended by this amount when necessary.
const CHUNK_SIZE: u64 = 131_072;
const_assert!(CHUNK_SIZE > TRANSACTION_MAX_JOURNAL_USAGE);

// See the comment for the `reclaim_size` member of Inner.
pub const DEFAULT_RECLAIM_SIZE: u64 = 524_288;

// Temporary space that should be reserved for the journal.  For example: space that is currently
// used in the journal file but cannot be deallocated yet because we are flushing.
pub const RESERVED_SPACE: u64 = 1_048_576;

// Whenever the journal is replayed (i.e. the system is unmounted and remounted), we reset the
// journal stream, at which point any half-complete transactions are discarded.  We indicate a
// journal reset by XORing the previous block's checksum with this mask, and using that value as a
// seed for the next journal block.
const RESET_XOR: u64 = 0xffffffffffffffff;

// To keep track of offsets within a journal file, we need both the file offset and the check-sum of
// the preceding block, since the check-sum of the preceding block is an input to the check-sum of
// every block.
pub type JournalCheckpoint = JournalCheckpointV32;

#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize, TypeFingerprint)]
pub struct JournalCheckpointV32 {
    pub file_offset: u64,

    // Starting check-sum for block that contains file_offset i.e. the checksum for the previous
    // block.
    pub checksum: Checksum,

    // If versioned, the version of elements stored in the journal. e.g. JournalRecord version.
    // This can change across reset events so we store it along with the offset and checksum to
    // know which version to deserialize.
    pub version: Version,
}

pub type JournalRecord = JournalRecordV62;

#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
#[cfg_attr(fuzz, derive(arbitrary::Arbitrary))]
pub enum JournalRecordV62 {
    EndBlock,
    Mutation {
        object_id: u64,
        mutation: MutationV62,
    },

    /// Commits records in the transaction.
    Commit,

    /// Discard all mutations with offsets greater than or equal to the given offset.
    Discard(u64),

    /// Indicates the device was flushed at the given journal offset.
    /// Note that this really means that at this point in the journal offset, we can be certain that
    /// there's no remaining buffered data in the block device; the buffers and the disk contents
    /// are consistent.
    /// We insert one of these records *after* a flush along with the *next* transaction to go
    /// through.  If that never comes (either due to graceful or hard shutdown), the journal reset
    /// on the next mount will serve the same purpose and count as a flush, although it is necessary
    /// to defensively flush the device before replaying the journal (if possible, i.e. not
    /// read-only) in case the block device connection was reused.
    DidFlushDevice(u64),

    /// Checksums for a data range written by this transaction, for a particular store identified by
    /// the u64 object id. A transaction is only valid if these checksums are right. The range is
    /// the device offset the checksums are for.
    ///
    /// A boolean indicates whether this range is being written to for the first time. For overwrite
    /// extents, we only check the checksums for a block if it has been written to for the first
    /// time since the last flush, because otherwise we can't roll it back anyway so it doesn't
    /// matter. For copy-on-write extents, the bool is always true.
    DataChecksums(u64, Range<u64>, crate::checksum::ChecksumsV38, bool),

    /// Commits records in a transaction that borrows metadata space, recording the total bytes
    /// currently held in `ObjectManager::metadata_reservation` (`M_total`).
    CommitBorrowed(u64),
}

#[allow(clippy::large_enum_variant)]
#[derive(Migrate, Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
#[migrate_to_version(JournalRecordV62)]
pub enum JournalRecordV61 {
    EndBlock,
    Mutation { object_id: u64, mutation: MutationV61 },
    Commit,
    Discard(u64),
    DidFlushDevice(u64),
    DataChecksums(u64, Range<u64>, crate::checksum::ChecksumsV38, bool),
    CommitBorrowed(u64),
}

#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
pub enum JournalRecordV60 {
    EndBlock,
    Mutation { object_id: u64, mutation: MutationV59 },
    Commit,
    Discard(u64),
    DidFlushDevice(u64),
    DataChecksums(u64, Range<u64>, crate::checksum::ChecksumsV38, bool),
}

impl From<JournalRecordV60> for JournalRecordV61 {
    fn from(record: JournalRecordV60) -> Self {
        match record {
            JournalRecordV60::EndBlock => Self::EndBlock,
            JournalRecordV60::Mutation { object_id, mutation } => {
                Self::Mutation { object_id, mutation: mutation.into() }
            }
            JournalRecordV60::Commit => Self::Commit,
            JournalRecordV60::Discard(offset) => Self::Discard(offset),
            JournalRecordV60::DidFlushDevice(offset) => Self::DidFlushDevice(offset),
            JournalRecordV60::DataChecksums(owner, range, checksums, first_write) => {
                Self::DataChecksums(owner, range, checksums, first_write)
            }
        }
    }
}

#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
pub enum JournalRecordV59 {
    EndBlock,
    Mutation { object_id: u64, mutation: MutationV59 },
    Commit,
    Discard(u64),
    DidFlushDevice(u64),
    DataChecksums(Range<u64>, crate::checksum::ChecksumsV38, bool),
}

impl From<JournalRecordV59> for JournalRecordV60 {
    fn from(record: JournalRecordV59) -> Self {
        match record {
            JournalRecordV59::EndBlock => Self::EndBlock,
            JournalRecordV59::Mutation { object_id, mutation } => {
                Self::Mutation { object_id, mutation }
            }
            JournalRecordV59::Commit => Self::Commit,
            JournalRecordV59::Discard(offset) => Self::Discard(offset),
            JournalRecordV59::DidFlushDevice(offset) => Self::DidFlushDevice(offset),
            JournalRecordV59::DataChecksums(range, checksums, first_write) => {
                // `JournalRecordV59::DataChecksums` did not record the owning store ID, so we map
                // it to `INVALID_OBJECT_ID`. During replay, `read_transactions` only permits
                // `INVALID_OBJECT_ID` when `checkpoint.version.major < 60` (rejecting it in V60+
                // records), and `ChecksumList::push` allows an existing `INVALID_OBJECT_ID` entry
                // to be overlapped by a subsequent write (adopting the new `owner_object_id`).
                Self::DataChecksums(INVALID_OBJECT_ID, range, checksums, first_write)
            }
        }
    }
}

#[allow(clippy::large_enum_variant)]
#[derive(Migrate, Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
#[migrate_to_version(JournalRecordV59)]
pub enum JournalRecordV57 {
    EndBlock,
    Mutation { object_id: u64, mutation: MutationV57 },
    Commit,
    Discard(u64),
    DidFlushDevice(u64),
    DataChecksums(Range<u64>, crate::checksum::ChecksumsV38, bool),
}

#[allow(clippy::large_enum_variant)]
#[derive(Migrate, Clone, Debug, Serialize, Deserialize, TypeFingerprint, Versioned)]
#[migrate_to_version(JournalRecordV57)]
pub enum JournalRecordV56 {
    EndBlock,
    Mutation { object_id: u64, mutation: MutationV56 },
    Commit,
    Discard(u64),
    DidFlushDevice(u64),
    DataChecksums(Range<u64>, crate::checksum::ChecksumsV38, bool),
}

pub(super) fn journal_handle_options() -> HandleOptions {
    HandleOptions { skip_journal_checks: true, ..Default::default() }
}

/// The journal records a stream of mutations that are to be applied to other objects.  At mount
/// time, these records can be replayed into memory.  It provides a way to quickly persist changes
/// without having to make a large number of writes; they can be deferred to a later time (e.g.
/// when a sufficient number have been queued).  It also provides support for transactions, the
/// ability to have mutations that are to be applied atomically together.
pub struct Journal {
    objects: Arc<ObjectManager>,
    hooks: Arc<HooksHandle>,
    handle: OnceLock<DataObjectHandle<ObjectStore>>,
    super_block_manager: SuperBlockManager,
    inner: Mutex<Inner>,
    writer_mutex: Mutex<()>,
    sync_mutex: futures::lock::Mutex<()>,
    trace: AtomicBool,

    // This event is used when we are waiting for a compaction to free up journal space.
    reclaim_event: Event,
}

#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum CompactionState {
    Idle,
    Running,
    Forced,
}

impl CompactionState {
    fn is_running(self) -> bool {
        !matches!(self, Self::Idle)
    }
}

struct Inner {
    super_block_header: SuperBlockHeader,

    // The offset that we can zero the journal up to now that it is no longer needed.
    zero_offset: Option<u64>,

    // The journal offset that we most recently flushed to the device.
    device_flushed_offset: u64,

    // If true, indicates a DidFlushDevice record is pending.
    needs_did_flush_device: bool,

    // The writer for the journal.
    writer: JournalWriter,

    // Set when a reset is encountered during a read.
    // Used at write pre_commit() time to ensure we write a version first thing after a reset.
    output_reset_version: bool,

    // Waker for the flush task.
    flush_waker: Option<Waker>,

    // Indicates the journal has been terminated.
    terminate: bool,

    // Latched error indicating reason for journal termination if not graceful.
    terminate_reason: Option<Error>,

    // Disable compactions.
    disable_compactions: bool,

    // When true, compactions are paused.
    compactions_paused: bool,

    // Current compaction state.
    compaction_state: CompactionState,

    // Waker for the sync task for when it's waiting for the flush task to finish.
    sync_waker: Option<Waker>,

    // The last offset we flushed to the journal file.
    flushed_offset: u64,

    // The last offset that should be considered valid in the journal file.  Most of the time, this
    // will be the same as `flushed_offset` but at mount time, this could be less and will only be
    // up to the end of the last valid transaction; it won't include transactions that follow that
    // have been discarded.
    valid_to: u64,

    // If, after replaying, we have to discard a number of mutations (because they don't validate),
    // this offset specifies where we need to discard back to.  This is so that when we next replay,
    // we ignore those mutations and continue with new good mutations.
    discard_offset: Option<u64>,

    // In the steady state, the journal should fluctuate between being approximately half of this
    // number and this number.  New super-blocks will be written every time about half of this
    // amount is written to the journal.
    reclaim_size: u64,

    image_builder_mode: Option<SuperBlockInstance>,

    // If true and `needs_barrier`, issue a pre-barrier on the first device write of each journal
    // write (which happens in multiples of `BLOCK_SIZE`). This ensures that all the corresponding
    // data writes make it to disk before the journal gets written to.
    barriers_enabled: bool,

    // If true, indicates that data write requests have been made to the device since the last
    // journal write.
    needs_barrier: bool,
}

impl Inner {
    fn terminate(&mut self, reason: Option<Error>) {
        self.terminate = true;

        if let Some(err) = reason {
            error!(error:? = err; "Terminating journal");
            // Log previous error if one was already set, otherwise latch the error.
            if let Some(prev_err) = self.terminate_reason.as_ref() {
                error!(error:? = prev_err; "Journal previously terminated");
            } else {
                self.terminate_reason = Some(err);
            }
        }

        if let Some(waker) = self.flush_waker.take() {
            waker.wake();
        }
        if let Some(waker) = self.sync_waker.take() {
            waker.wake();
        }
    }
}

pub struct JournalOptions {
    /// In the steady state, the journal should fluctuate between being approximately half of this
    /// number and this number.  New super-blocks will be written every time about half of this
    /// amount is written to the journal.
    pub reclaim_size: u64,

    // If true, issue a pre-barrier on the first device write of each journal write (which happens
    // in multiples of `BLOCK_SIZE`). This ensures that all the corresponding data writes make it
    // to disk before the journal gets written to.
    pub barriers_enabled: bool,
}

impl Default for JournalOptions {
    fn default() -> Self {
        JournalOptions { reclaim_size: DEFAULT_RECLAIM_SIZE, barriers_enabled: false }
    }
}

struct JournaledTransactions {
    transactions: Vec<JournaledTransaction>,
    device_flushed_offset: u64,
}

/// Indicates whether a replayed transaction borrowed metadata space.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub enum BorrowedMetadata {
    /// Funded transaction (`JournalRecord::Commit`) that adds `4 * txn_size` to `M_total`.
    #[default]
    None,

    /// V61+ transaction (`JournalRecord::CommitBorrowed(u64)`) that borrowed metadata space,
    /// carrying `M_total` (`metadata_reservation.amount()`).
    Borrowed(u64),

    /// Pre-V61 transaction that contained a `Mutation::UpdateBorrowed(u64)` record.
    LegacyBorrowed(u64),
}

#[derive(Debug, Default)]
pub struct JournaledTransaction {
    pub checkpoint: JournalCheckpoint,
    pub root_parent_mutations: Vec<Mutation>,
    pub root_mutations: Vec<Mutation>,
    /// List of (store_object_id, mutation).
    pub non_root_mutations: Vec<(u64, Mutation)>,
    pub end_offset: u64,
    pub checksums: Vec<JournaledChecksums>,

    /// Records offset + 1 of the matching begin_flush transaction. The +1 is because we want to
    /// ignore the begin flush transaction; we don't need or want to replay it.
    pub end_flush: Option<(/* store_id: */ u64, /* begin offset: */ u64)>,

    /// The volume which was deleted in this transaction, if any.
    pub volume_deleted: Option</* store_id: */ u64>,

    /// Whether this transaction borrowed metadata space.
    pub borrowed_metadata: BorrowedMetadata,

    /// Per-object journal usage (in bytes) for mutations in this transaction.
    pub object_journal_usage: SmallVec<[(u64, u64); 2]>,
}

impl JournaledTransaction {
    fn new(checkpoint: JournalCheckpoint) -> Self {
        Self { checkpoint, ..Default::default() }
    }
}

const VOLUME_DELETED: u64 = u64::MAX;

#[derive(Debug)]
pub struct JournaledChecksums {
    pub owner_object_id: u64,
    pub device_range: Range<u64>,
    pub checksums: Checksums,
    pub first_write: bool,
}

/// Handles for journal-like objects have some additional functionality to manage their extents,
/// since during replay we need to add extents as we find them.
pub trait JournalHandle: ReadObjectHandle {
    /// The end offset of the last extent in the JournalHandle.  Used only for validating extents
    /// (which will be skipped if None is returned).
    /// Note this is equivalent in value to ReadObjectHandle::get_size, when present.
    fn end_offset(&self) -> Option<u64>;
    /// Adds an extent to the current end of the journal stream.
    /// `added_offset` is the offset into the journal of the transaction which added this extent,
    /// used for discard_extents.
    fn push_extent(&mut self, added_offset: u64, device_range: Range<u64>);
    /// Discards all extents which were added in a transaction at offset >= |discard_offset|.
    fn discard_extents(&mut self, discard_offset: u64);
}

// Provide a stub implementation for DataObjectHandle so we can use it in
// Journal::read_transactions.  Manual extent management is a NOP (which is OK since presumably the
// DataObjectHandle already knows where its extents live).
impl<S: HandleOwner> JournalHandle for DataObjectHandle<S> {
    fn end_offset(&self) -> Option<u64> {
        None
    }
    fn push_extent(&mut self, _added_offset: u64, _device_range: Range<u64>) {
        // NOP
    }
    fn discard_extents(&mut self, _discard_offset: u64) {
        // NOP
    }
}

#[fxfs_trace::trace]
impl Journal {
    pub fn new(
        objects: Arc<ObjectManager>,
        options: JournalOptions,
        hooks: Arc<HooksHandle>,
    ) -> Journal {
        let starting_checksum = rand::random_range(1..u64::MAX);
        Journal {
            objects: objects,
            hooks,
            handle: OnceLock::new(),
            super_block_manager: SuperBlockManager::new(),
            inner: Mutex::new(Inner {
                super_block_header: SuperBlockHeader::default(),
                zero_offset: None,
                device_flushed_offset: 0,
                needs_did_flush_device: false,
                writer: JournalWriter::new(BLOCK_SIZE, starting_checksum),
                output_reset_version: false,
                flush_waker: None,
                terminate: false,
                terminate_reason: None,
                disable_compactions: false,
                compactions_paused: false,
                compaction_state: CompactionState::Idle,
                sync_waker: None,
                flushed_offset: 0,
                valid_to: 0,
                discard_offset: None,
                reclaim_size: options.reclaim_size,
                image_builder_mode: None,
                barriers_enabled: options.barriers_enabled,
                needs_barrier: false,
            }),
            writer_mutex: Mutex::new(()),
            sync_mutex: futures::lock::Mutex::new(()),
            trace: AtomicBool::new(false),
            reclaim_event: Event::new(),
        }
    }

    pub fn set_trace(&self, trace: bool) {
        let old_value = self.trace.swap(trace, Ordering::Relaxed);
        if trace != old_value {
            info!(trace; "J: trace");
        }
    }

    pub fn set_image_builder_mode(&self, mode: Option<SuperBlockInstance>) {
        self.inner.lock().image_builder_mode = mode;
        if let Some(instance) = mode {
            *self.super_block_manager.next_instance.lock() = instance;
        }
    }

    pub fn image_builder_mode(&self) -> Option<SuperBlockInstance> {
        self.inner.lock().image_builder_mode
    }

    #[cfg(feature = "migration")]
    pub fn set_filesystem_uuid(&self, uuid: &[u8; 16]) -> Result<(), Error> {
        ensure!(
            self.inner.lock().image_builder_mode.is_some(),
            "Can only set filesystem uuid in image builder mode."
        );
        self.inner.lock().super_block_header.guid.0 = uuid::Uuid::from_bytes(*uuid);
        Ok(())
    }

    pub(crate) async fn read_superblocks(
        &self,
        device: Arc<dyn Device>,
        block_size: BlockSize,
    ) -> Result<(SuperBlockHeader, ObjectStore), Error> {
        self.super_block_manager.load(device, block_size).await
    }

    /// Used during replay to validate a mutation.  This should return false if the mutation is not
    /// valid and should not be applied.  This could be for benign reasons: e.g. the device flushed
    /// data out-of-order, or because of a malicious actor.
    fn validate_mutation(
        &self,
        mutation: &Mutation,
        block_size: BlockSize,
        device_size: u64,
    ) -> bool {
        match mutation {
            Mutation::ObjectStore(ObjectStoreMutation {
                item:
                    Item {
                        key:
                            ObjectKey {
                                data: ObjectKeyData::Attribute(_, AttributeKey::Extent(extent)),
                                ..
                            },
                        value: ObjectValue::Extent(ExtentValue::Some { device_offset, mode, .. }),
                        ..
                    },
                ..
            }) => {
                if extent.is_empty() || !block_size.is_aligned(extent) {
                    return false;
                }
                let len = extent.length().unwrap();
                if let ExtentMode::Cow(checksums) = mode {
                    if checksums.len() > 0 {
                        if len % checksums.len() as u64 != 0 {
                            return false;
                        }
                        if !block_size.is_aligned(len / checksums.len() as u64) {
                            return false;
                        }
                    }
                }
                if !block_size.is_aligned(device_offset)
                    || *device_offset >= device_size
                    || device_size - *device_offset < len
                {
                    return false;
                }
            }
            Mutation::ObjectStore(_) => {}
            Mutation::EncryptedObjectStore(_) => {}
            Mutation::Allocator(AllocatorMutation::Allocate { device_range, owner_object_id }) => {
                return !device_range.is_empty()
                    && *owner_object_id != INVALID_OBJECT_ID
                    && device_range.end <= device_size;
            }
            Mutation::Allocator(AllocatorMutation::Deallocate {
                device_range,
                owner_object_id,
            }) => {
                return !device_range.is_empty()
                    && *owner_object_id != INVALID_OBJECT_ID
                    && device_range.end <= device_size;
            }
            Mutation::Allocator(AllocatorMutation::MarkForDeletion(owner_object_id)) => {
                return *owner_object_id != INVALID_OBJECT_ID;
            }
            Mutation::Allocator(AllocatorMutation::SetLimit { owner_object_id, .. }) => {
                return *owner_object_id != INVALID_OBJECT_ID;
            }
            Mutation::BeginFlush => {}
            Mutation::EndFlush(_) => {}
            Mutation::DeleteVolume => {}
            Mutation::UpdateBorrowed(_) => {}
            Mutation::UpdateMutationsKey(_) => {}
            Mutation::CreateInternalDir(owner_object_id) => {
                return *owner_object_id != INVALID_OBJECT_ID;
            }
        }
        true
    }

    // Assumes that `mutation` has been validated.
    fn update_checksum_list(
        &self,
        journal_offset: u64,
        mutation: &Mutation,
        checksum_list: &mut ChecksumList,
    ) -> Result<(), Error> {
        match mutation {
            Mutation::ObjectStore(_) => {}
            Mutation::Allocator(AllocatorMutation::Deallocate { device_range, .. }) => {
                checksum_list.mark_deallocated(journal_offset, device_range.clone().into());
            }
            Mutation::Allocator(AllocatorMutation::MarkForDeletion(owner_object_id)) => {
                checksum_list.mark_for_deletion(journal_offset, *owner_object_id);
            }
            _ => {}
        }
        Ok(())
    }

    /// Reads the latest super-block, and then replays journaled records.
    #[trace]
    pub async fn replay(
        &self,
        filesystem: Arc<FxFilesystem>,
        on_new_allocator: Option<Box<dyn Fn(Arc<Allocator>) + Send + Sync>>,
    ) -> Result<(), Error> {
        let block_size = filesystem.block_size();

        let (super_block, root_parent) =
            self.super_block_manager.load(filesystem.device(), block_size).await?;

        let root_parent = Arc::new(ObjectStore::attach_filesystem(root_parent, filesystem.clone()));

        self.objects.set_root_parent_store(root_parent.clone());
        let allocator =
            Arc::new(Allocator::new(filesystem.clone(), super_block.allocator_object_id));
        if let Some(on_new_allocator) = on_new_allocator {
            on_new_allocator(allocator.clone());
        }
        self.objects.set_allocator(allocator.clone());
        self.objects.init_replay_metadata_reservation(
            super_block.journal_checkpoint.version,
            &super_block.reserved_metadata_space,
        );
        self.objects.set_last_end_offset(super_block.super_block_journal_file_offset);
        {
            let mut inner = self.inner.lock();
            inner.super_block_header = super_block.clone();
        }

        let device = filesystem.device();

        let mut handle;
        {
            let root_parent_layer = root_parent.tree().mutable_layer();
            let mut iter = root_parent_layer.seek(Bound::Included(&ObjectKey::attribute(
                super_block.journal_object_id,
                AttributeId::DATA,
                AttributeKey::Extent(Extent::search_key_from_offset(
                    BLOCK_SIZE.align_down(super_block.journal_checkpoint.file_offset),
                )),
            )));
            let start_offset = if let Some(ItemRef {
                key:
                    ObjectKey {
                        data:
                            ObjectKeyData::Attribute(AttributeId::DATA, AttributeKey::Extent(extent)),
                        ..
                    },
                ..
            }) = iter.get()
            {
                extent.start
            } else {
                0
            };
            handle = BootstrapObjectHandle::new_with_start_offset(
                super_block.journal_object_id,
                device.clone(),
                start_offset,
            );
            while let Some(item) = iter.get() {
                if !match item.into() {
                    Some((
                        object_id,
                        AttributeId::DATA,
                        extent,
                        ExtentValue::Some { device_offset, .. },
                    )) if object_id == super_block.journal_object_id => {
                        if let Some(end_offset) = handle.end_offset() {
                            if extent.start != end_offset {
                                bail!(anyhow!(FxfsError::Inconsistent).context(format!(
                                    "Unexpected journal extent {:?}, expected start: {}",
                                    item, end_offset
                                )));
                            }
                        }
                        handle.push_extent(
                            0, // We never discard extents from the root parent store.
                            *device_offset
                                ..*device_offset + extent.length().context("Invalid extent")?,
                        );
                        true
                    }
                    _ => false,
                } {
                    break;
                }
                iter.advance().await.context("Failed to advance root parent store iterator")?;
            }
        }

        let mut reader = JournalReader::new(handle, &super_block.journal_checkpoint);
        let JournaledTransactions { mut transactions, device_flushed_offset } = self
            .read_transactions(&mut reader, None, INVALID_OBJECT_ID)
            .await
            .context("Reading transactions for replay")?;

        // Validate all the mutations.
        let mut checksum_list = ChecksumList::new(device_flushed_offset);
        let mut valid_to = reader.journal_file_checkpoint().file_offset;
        let device_size = device.size();
        'bad_replay: for JournaledTransaction {
            checkpoint,
            root_parent_mutations,
            root_mutations,
            non_root_mutations,
            checksums,
            ..
        } in &transactions
        {
            for JournaledChecksums { owner_object_id, device_range, checksums, first_write } in
                checksums
            {
                checksum_list
                    .push(
                        checkpoint.file_offset,
                        *owner_object_id,
                        device_range.clone(),
                        checksums.maybe_as_ref().context("Malformed checksums")?,
                        *first_write,
                    )
                    .context("Pushing journal checksum records to checksum list")?;
            }
            for mutation in root_parent_mutations
                .iter()
                .chain(root_mutations)
                .chain(non_root_mutations.iter().map(|(_, m)| m))
            {
                if !self.validate_mutation(mutation, block_size, device_size) {
                    info!(mutation:?; "Stopping replay at bad mutation");
                    valid_to = checkpoint.file_offset;
                    break 'bad_replay;
                }
                self.update_checksum_list(checkpoint.file_offset, &mutation, &mut checksum_list)?;
            }
        }

        // Validate the checksums. Note if barriers are enabled, there will be no checksums in
        // practice to verify.
        let valid_to = checksum_list
            .verify(device.as_ref(), valid_to)
            .await
            .context("Failed to validate checksums")?;

        // Apply the mutations...

        let mut last_checkpoint = reader.journal_file_checkpoint();
        let mut journal_offsets = super_block.journal_file_offsets.clone();

        // Start with the root-parent mutations, and also determine the journal offsets for all
        // other objects.
        for (
            index,
            JournaledTransaction {
                checkpoint,
                root_parent_mutations,
                end_flush,
                volume_deleted,
                ..
            },
        ) in transactions.iter_mut().enumerate()
        {
            if checkpoint.file_offset >= valid_to {
                last_checkpoint = checkpoint.clone();

                // Truncate the transactions so we don't need to worry about them on the next pass.
                transactions.truncate(index);
                break;
            }

            let context = ApplyContext { mode: ApplyMode::Replay, checkpoint: checkpoint.clone() };
            for mutation in root_parent_mutations.drain(..) {
                self.objects
                    .apply_mutation(
                        super_block.root_parent_store_object_id,
                        mutation,
                        &context,
                        AssocObj::None,
                    )
                    .context("Failed to replay root parent store mutations")?;
            }

            if let Some((object_id, journal_offset)) = end_flush {
                journal_offsets.insert(*object_id, *journal_offset);
            }

            if let Some(object_id) = volume_deleted {
                journal_offsets.insert(*object_id, VOLUME_DELETED);
            }
        }

        // Now we can open the root store.
        let root_store = ObjectStore::open(&root_parent, super_block.root_store_object_id, None)
            .await
            .context("Unable to open root store")?;

        ensure!(
            !root_store.is_encrypted(),
            anyhow!(FxfsError::Inconsistent).context("Root store is encrypted")
        );
        self.objects.set_root_store(root_store);

        let root_store_offset =
            journal_offsets.get(&super_block.root_store_object_id).copied().unwrap_or(0);

        // Now replay the root store mutations.
        for JournaledTransaction { checkpoint, root_mutations, .. } in &mut transactions {
            if checkpoint.file_offset < root_store_offset {
                continue;
            }

            let context = ApplyContext { mode: ApplyMode::Replay, checkpoint: checkpoint.clone() };
            for mutation in root_mutations.drain(..) {
                self.objects
                    .apply_mutation(
                        super_block.root_store_object_id,
                        mutation,
                        &context,
                        AssocObj::None,
                    )
                    .context("Failed to replay root store mutations")?;
            }
        }

        // Now we can open the allocator.
        allocator.open().await.context("Failed to open allocator")?;

        // Now replay all other mutations.
        for JournaledTransaction {
            checkpoint,
            non_root_mutations,
            end_offset,
            borrowed_metadata,
            object_journal_usage,
            ..
        } in transactions
        {
            self.objects
                .replay_mutations(
                    non_root_mutations,
                    &object_journal_usage,
                    &journal_offsets,
                    &ApplyContext { mode: ApplyMode::Replay, checkpoint },
                    end_offset,
                    borrowed_metadata,
                )
                .await
                .context("Failed to replay mutations")?;
        }

        allocator.on_replay_complete().await.context("Failed to complete replay for allocator")?;

        let discarded_to =
            if last_checkpoint.file_offset != reader.journal_file_checkpoint().file_offset {
                Some(reader.journal_file_checkpoint().file_offset)
            } else {
                None
            };

        // Configure the journal writer so that we can continue.
        {
            if last_checkpoint.file_offset < super_block.super_block_journal_file_offset {
                return Err(anyhow!(FxfsError::Inconsistent).context(format!(
                    "journal replay cut short; journal finishes at {}, but super-block was \
                     written at {}",
                    last_checkpoint.file_offset, super_block.super_block_journal_file_offset
                )));
            }
            let handle = ObjectStore::open_object(
                &root_parent,
                super_block.journal_object_id,
                journal_handle_options(),
                None,
            )
            .await
            .with_context(|| {
                format!(
                    "Failed to open journal file (object id: {})",
                    super_block.journal_object_id
                )
            })?;
            let _ = self.handle.set(handle);
            let mut inner = self.inner.lock();
            reader.skip_to_end_of_block();
            let mut writer_checkpoint = reader.journal_file_checkpoint();

            // Make sure we don't accidentally use the reader from now onwards.
            std::mem::drop(reader);

            // Reset the stream to indicate that we've remounted the journal.
            writer_checkpoint.checksum ^= RESET_XOR;
            writer_checkpoint.version = LATEST_VERSION;
            inner.flushed_offset = writer_checkpoint.file_offset;

            // When we open the filesystem as writable, we flush the device.
            inner.device_flushed_offset = inner.flushed_offset;

            inner.writer.seek(writer_checkpoint);
            inner.output_reset_version = true;
            inner.valid_to = last_checkpoint.file_offset;
            if last_checkpoint.file_offset < inner.flushed_offset {
                inner.discard_offset = Some(last_checkpoint.file_offset);
            }
        }

        self.objects
            .on_replay_complete()
            .await
            .context("Failed to complete replay for object manager")?;

        let required = self.objects.required_reservation();
        let borrowed = self.objects.borrowed_metadata_space();
        info!(
            checkpoint = last_checkpoint.file_offset,
            discarded_to,
            metadata_reservation:% =
                format_args!("{}({} - {})", required - borrowed, required, borrowed),
            store_borrowed:? = self.objects.stores_borrowed();
            "replay complete"
        );
        Ok(())
    }

    async fn read_transactions(
        &self,
        reader: &mut JournalReader,
        end_offset: Option<u64>,
        object_id_filter: u64,
    ) -> Result<JournaledTransactions, Error> {
        let mut transactions = Vec::new();
        let (mut device_flushed_offset, root_parent_store_object_id, root_store_object_id) = {
            let super_block = &self.inner.lock().super_block_header;
            (
                super_block.super_block_journal_file_offset,
                super_block.root_parent_store_object_id,
                super_block.root_store_object_id,
            )
        };
        let mut current_transaction = None;
        let mut begin_flush_offsets = HashMap::default();
        let mut stores_deleted = HashSet::new();
        loop {
            // Cache the checkpoint before we deserialize a record.
            let checkpoint = reader.journal_file_checkpoint();
            if let Some(end_offset) = end_offset {
                if checkpoint.file_offset >= end_offset {
                    break;
                }
            }
            let result = reader
                .deserialize::<JournalRecord>()
                .await
                .context("Failed to deserialize journal record")?;
            match result {
                ReadResult::Reset(_) => {
                    if current_transaction.is_some() {
                        current_transaction = None;
                        transactions.pop();
                    }
                    let offset = reader.journal_file_checkpoint().file_offset;
                    if offset > device_flushed_offset {
                        device_flushed_offset = offset;
                    }
                }
                ReadResult::Some(record) => match record {
                    JournalRecord::EndBlock => {
                        reader.skip_to_end_of_block();
                    }
                    JournalRecord::Mutation { object_id, mutation } => {
                        let record_start_offset = checkpoint.file_offset;
                        let current_transaction = match current_transaction.as_mut() {
                            None => {
                                transactions.push(JournaledTransaction::new(checkpoint));
                                current_transaction = transactions.last_mut();
                                current_transaction.as_mut().unwrap()
                            }
                            Some(transaction) => transaction,
                        };

                        if stores_deleted.contains(&object_id) {
                            bail!(
                                anyhow!(FxfsError::Inconsistent)
                                    .context("Encountered mutations for deleted store")
                            );
                        }

                        match &mutation {
                            Mutation::BeginFlush => {
                                begin_flush_offsets
                                    .insert(object_id, current_transaction.checkpoint.file_offset);
                            }
                            Mutation::EndFlush(_) => {
                                if let Some(offset) = begin_flush_offsets.remove(&object_id) {
                                    if let Some(deleted_volume) =
                                        &current_transaction.volume_deleted
                                    {
                                        if *deleted_volume == object_id {
                                            bail!(anyhow!(FxfsError::Inconsistent).context(
                                                "Multiple EndFlush/DeleteVolume mutations in a \
                                                single transaction for the same object"
                                            ));
                                        }
                                    }
                                    // `BeginFlush` seals the object's in-memory mutable layer so
                                    // all mutations prior to `BeginFlush` are written to the new
                                    // layer file referenced by `EndFlush`. Transactions committed
                                    // after `BeginFlush` (starting at `>= offset + 1`) are written
                                    // to the new mutable layer and must still be replayed; the
                                    // `BeginFlush` transaction itself at `offset` has no remaining
                                    // mutations to replay once `EndFlush` commits.
                                    if current_transaction
                                        .end_flush
                                        .replace((object_id, offset + 1))
                                        .is_some()
                                    {
                                        bail!(anyhow!(FxfsError::Inconsistent).context(
                                            "Multiple EndFlush mutations in a \
                                             single transaction"
                                        ));
                                    }
                                }
                            }
                            Mutation::DeleteVolume => {
                                if let Some((flushed_object, _)) = &current_transaction.end_flush {
                                    if *flushed_object == object_id {
                                        bail!(anyhow!(FxfsError::Inconsistent).context(
                                            "Multiple EndFlush/DeleteVolume mutations in a \
                                                single transaction for the same object"
                                        ));
                                    }
                                }
                                if current_transaction.volume_deleted.replace(object_id).is_some() {
                                    bail!(anyhow!(FxfsError::Inconsistent).context(
                                        "Multiple DeleteVolume mutations in a single \
                                         transaction"
                                    ));
                                }
                                stores_deleted.insert(object_id);
                            }
                            Mutation::UpdateBorrowed(borrowed) => {
                                ensure!(
                                    object_id == 0,
                                    anyhow!(FxfsError::Inconsistent)
                                        .context("Invalid object_id for UpdateBorrowed")
                                );
                                current_transaction.borrowed_metadata =
                                    BorrowedMetadata::LegacyBorrowed(*borrowed);
                                continue;
                            }
                            _ => {}
                        }

                        let record_len =
                            reader.journal_file_checkpoint().file_offset - record_start_offset;

                        // If this mutation doesn't need to be applied, don't bother adding it
                        // to the transaction.
                        if (object_id_filter == INVALID_OBJECT_ID || object_id_filter == object_id)
                            && self.should_apply(object_id, &current_transaction.checkpoint)
                        {
                            if object_id == root_parent_store_object_id {
                                current_transaction.root_parent_mutations.push(mutation);
                            } else if object_id == root_store_object_id {
                                current_transaction.root_mutations.push(mutation);
                            } else {
                                if mutation.counts_towards_object_journal_usage() {
                                    if let Some((last_id, usage)) =
                                        current_transaction.object_journal_usage.last_mut()
                                        && *last_id == object_id
                                    {
                                        *usage += record_len;
                                    } else if let Some((_, usage)) = current_transaction
                                        .object_journal_usage
                                        .iter_mut()
                                        .find(|(id, _)| *id == object_id)
                                    {
                                        *usage += record_len;
                                    } else {
                                        current_transaction
                                            .object_journal_usage
                                            .push((object_id, record_len));
                                    }
                                }
                                current_transaction.non_root_mutations.push((object_id, mutation));
                            }
                        }
                    }
                    JournalRecord::DataChecksums(
                        owner_object_id,
                        device_range,
                        checksums,
                        first_write,
                    ) => {
                        ensure!(
                            checkpoint.version.major < 60 || owner_object_id != INVALID_OBJECT_ID,
                            anyhow!(FxfsError::Inconsistent)
                                .context("Invalid owner_object_id in DataChecksums")
                        );
                        if stores_deleted.contains(&owner_object_id) {
                            bail!(
                                anyhow!(FxfsError::Inconsistent)
                                    .context("Encountered DataChecksums for deleted store")
                            );
                        }
                        let current_transaction = match current_transaction.as_mut() {
                            None => {
                                transactions.push(JournaledTransaction::new(checkpoint));
                                current_transaction = transactions.last_mut();
                                current_transaction.as_mut().unwrap()
                            }
                            Some(transaction) => transaction,
                        };
                        current_transaction.checksums.push(JournaledChecksums {
                            owner_object_id,
                            device_range,
                            checksums,
                            first_write,
                        });
                    }
                    JournalRecord::Commit | JournalRecord::CommitBorrowed(_) => {
                        // `pre_commit` can emit an empty `Borrowed` transaction
                        // (`CommitBorrowed` with no mutations) when `needs_borrow_for_journal`
                        // is true after a journal reset or discard. Recording it during full
                        // journal replay advances `last_end_offset` so that a subsequent funded
                        // transaction is not charged for the discarded/padded journal range.
                        if current_transaction.is_none()
                            && object_id_filter == INVALID_OBJECT_ID
                            && matches!(record, JournalRecord::CommitBorrowed(_))
                        {
                            transactions.push(JournaledTransaction::new(checkpoint));
                            current_transaction = transactions.last_mut();
                        }
                        if let Some(&mut JournaledTransaction {
                            ref checkpoint,
                            ref root_parent_mutations,
                            ref mut end_offset,
                            ref mut borrowed_metadata,
                            ..
                        }) = current_transaction.take()
                        {
                            for mutation in root_parent_mutations {
                                // Snoop the mutations for any that might apply to the journal
                                // file so that we can pass them to the reader so that it can
                                // read the journal file.
                                if let Mutation::ObjectStore(ObjectStoreMutation {
                                    item:
                                        Item {
                                            key:
                                                ObjectKey {
                                                    object_id,
                                                    data:
                                                        ObjectKeyData::Attribute(
                                                            AttributeId::DATA,
                                                            AttributeKey::Extent(extent),
                                                        ),
                                                    ..
                                                },
                                            value:
                                                ObjectValue::Extent(ExtentValue::Some {
                                                    device_offset,
                                                    ..
                                                }),
                                            ..
                                        },
                                    ..
                                }) = mutation
                                {
                                    // Add the journal extents we find on the way to our
                                    // reader.
                                    let handle = reader.handle();
                                    if *object_id != handle.object_id() {
                                        continue;
                                    }
                                    if let Some(end_offset) = handle.end_offset() {
                                        if extent.start != end_offset {
                                            bail!(anyhow!(FxfsError::Inconsistent).context(
                                                format!(
                                                    "Unexpected journal extent {:?} -> {}, \
                                                       expected start: {}",
                                                    *extent, device_offset, end_offset,
                                                )
                                            ));
                                        }
                                    }
                                    handle.push_extent(
                                        checkpoint.file_offset,
                                        *device_offset
                                            ..*device_offset
                                                + extent.length().context("Invalid extent")?,
                                    );
                                }
                            }
                            *end_offset = reader.journal_file_checkpoint().file_offset;
                            if let JournalRecord::CommitBorrowed(borrowed) = record {
                                *borrowed_metadata = BorrowedMetadata::Borrowed(borrowed);
                            }
                        }
                    }
                    JournalRecord::Discard(offset) => {
                        if offset == 0 {
                            bail!(
                                anyhow!(FxfsError::Inconsistent)
                                    .context("Invalid offset for Discard")
                            );
                        }
                        if let Some(transaction) = current_transaction.as_ref() {
                            if transaction.checkpoint.file_offset < offset {
                                // Odd, but OK.
                                continue;
                            }
                        }
                        current_transaction = None;
                        while let Some(transaction) = transactions.last() {
                            if transaction.checkpoint.file_offset < offset {
                                break;
                            }
                            transactions.pop();
                        }
                        reader.handle().discard_extents(offset);
                    }
                    JournalRecord::DidFlushDevice(offset) => {
                        if offset > device_flushed_offset {
                            device_flushed_offset = offset;
                        }
                    }
                },

                // This is expected when we reach the end of the journal stream.
                ReadResult::ChecksumMismatch => break,
            }
        }

        // Discard any uncommitted transaction.
        if current_transaction.is_some() {
            transactions.pop();
        }

        Ok(JournaledTransactions { transactions, device_flushed_offset })
    }

    /// Creates an empty filesystem with the minimum viable objects (including a root parent and
    /// root store but no further child stores).
    pub async fn init_empty(&self, filesystem: Arc<FxFilesystem>) -> Result<(), Error> {
        // The following constants are only used at format time. When mounting, the recorded values
        // in the superblock should be used.  The root parent store does not have a parent, but
        // needs an object ID to be registered with ObjectManager, so it cannot collide (i.e. have
        // the same object ID) with any objects in the root store that use the journal to track
        // mutations.
        const INIT_ROOT_PARENT_STORE_OBJECT_ID: u64 = 3;
        const INIT_ROOT_STORE_OBJECT_ID: u64 = 4;
        const INIT_ALLOCATOR_OBJECT_ID: u64 = 5;

        info!(device_size = filesystem.device().size(); "Formatting");

        let checkpoint = JournalCheckpoint {
            version: LATEST_VERSION,
            ..self.inner.lock().writer.journal_file_checkpoint()
        };

        let mut current_generation = 1;
        if filesystem.options().image_builder_mode.is_some() {
            // Note that in non-image_builder_mode we write both superblocks when we format
            // (in FxFilesystemBuilder::open). In image_builder_mode we only write once at the end
            // as part of finalize(), which is why we must make sure the generation we write is
            // newer than any existing generation.

            // Note: This should is the *filesystem* block size, not the device block size which
            // is currently always 4096 (https://fxbug.dev/42063349)
            let block_size = filesystem.block_size();
            match self.read_superblocks(filesystem.device(), block_size).await {
                Ok((super_block, _)) => {
                    log::info!(
                        "Found existing superblock with generation {}. Bumping by 1.",
                        super_block.generation
                    );
                    current_generation = super_block.generation.wrapping_add(1);
                }
                Err(_) => {
                    // TODO(https://fxbug.dev/463757813): It's not unusual to fail to read
                    // superblocks when we're formatting a new filesystem but we should probably
                    // fail the format if we get an IO error.
                }
            }
        }

        let root_parent = ObjectStore::new_empty(
            None,
            INIT_ROOT_PARENT_STORE_OBJECT_ID,
            filesystem.clone(),
            None,
        );
        self.objects.set_root_parent_store(root_parent.clone());

        let allocator = Arc::new(Allocator::new(filesystem.clone(), INIT_ALLOCATOR_OBJECT_ID));
        self.objects.set_allocator(allocator.clone());
        self.objects.init_metadata_reservation().await?;

        let journal_handle;
        let super_block_a_handle;
        let super_block_b_handle;
        let root_store;
        let mut transaction = root_parent
            .new_transaction(
                lock_keys![],
                Options { skip_journal_checks: true, ..Default::default() },
            )
            .await?;
        root_store = root_parent
            .new_child_store(
                &mut transaction,
                NewChildStoreOptions { object_id: INIT_ROOT_STORE_OBJECT_ID, ..Default::default() },
                None,
            )
            .await
            .context("new_child_store")?;
        self.objects.set_root_store(root_store.clone());

        allocator.create(&mut transaction).await?;

        // Create the super-block objects...
        super_block_a_handle = ObjectStore::create_object_with_id(
            &root_store,
            &mut transaction,
            ReservedId::new(&root_store, NonZero::new(SuperBlockInstance::A.object_id()).unwrap()),
            HandleOptions::default(),
            None,
        )
        .context("create super block")?;
        root_store.update_last_object_id(SuperBlockInstance::A.object_id());
        super_block_a_handle
            .extend(&mut transaction, SuperBlockInstance::A.first_extent())
            .await
            .context("extend super block")?;
        super_block_b_handle = ObjectStore::create_object_with_id(
            &root_store,
            &mut transaction,
            ReservedId::new(&root_store, NonZero::new(SuperBlockInstance::B.object_id()).unwrap()),
            HandleOptions::default(),
            None,
        )
        .context("create super block")?;
        root_store.update_last_object_id(SuperBlockInstance::B.object_id());
        super_block_b_handle
            .extend(&mut transaction, SuperBlockInstance::B.first_extent())
            .await
            .context("extend super block")?;

        // the journal object...
        journal_handle = ObjectStore::create_object(
            &root_parent,
            &mut transaction,
            journal_handle_options(),
            None,
        )
        .await
        .context("create journal")?;
        if self.inner.lock().image_builder_mode.is_none() {
            let mut file_range = 0..self.chunk_size();
            journal_handle
                .preallocate_range(&mut transaction, &mut file_range)
                .await
                .context("preallocate journal")?;
            if file_range.start < file_range.end {
                bail!("preallocate_range returned too little space");
            }
        }

        // Write the root store object info.
        root_store.create(&mut transaction).await?;

        // The root parent graveyard.
        root_parent.set_graveyard_directory_object_id(
            Graveyard::create(&mut transaction, &root_parent).await?,
        );

        transaction.commit().await?;

        self.inner.lock().super_block_header = SuperBlockHeader::new(
            current_generation,
            root_parent.store_object_id(),
            root_parent.graveyard_directory_object_id(),
            root_store.store_object_id(),
            allocator.object_id(),
            journal_handle.object_id(),
            checkpoint,
            /* earliest_version: */ LATEST_VERSION,
        );

        // Initialize the journal writer.
        let _ = self.handle.set(journal_handle);
        Ok(())
    }

    /// Finalizes a filesystem built in `image_builder_mode` by preallocating the journal, flushing
    /// all in-memory object store mutations to layer files, funding the metadata reservation for
    /// those layers, and writing the super-block.
    pub async fn finalize_image_builder(&self) -> Result<(), Error> {
        let handle = self.handle.get().unwrap();
        let mut transaction = handle
            .store()
            .new_transaction(
                lock_keys![LockKey::object(handle.store().store_object_id(), handle.object_id())],
                Options { skip_journal_checks: true, ..Default::default() },
            )
            .await?;
        let mut file_range = 0..self.chunk_size();
        handle
            .preallocate_range(&mut transaction, &mut file_range)
            .await
            .context("preallocate journal")?;
        if file_range.start < file_range.end {
            bail!("preallocate_range returned too little space");
        }
        transaction.commit().await?;

        {
            let mut inner = self.inner.lock();
            inner.image_builder_mode = None;
            inner.compaction_state = CompactionState::Forced;
        }
        scopeguard::defer! {
            let mut inner = self.inner.lock();
            inner.compaction_state = CompactionState::Idle;
            if let Some(waker) = inner.flush_waker.take() {
                waker.wake();
            }
            self.reclaim_event.notify(usize::MAX);
        }

        let earliest_version = self
            .objects
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .context("Failed to flush objects")?;
        self.inner.lock().super_block_header.earliest_version = earliest_version;

        // In image builder mode, transactions borrow metadata space and allocate directly from the
        // allocator rather than from `metadata_reservation`. Now that all stores have been flushed
        // and their layer sizes are known, top up `metadata_reservation` to
        // `required_reservation()` so that the finalized image starts with a fully funded metadata
        // reservation.
        let reservation = self.objects.metadata_reservation();
        let needed = self.objects.required_reservation().saturating_sub(reservation.amount());
        if needed > 0 {
            reservation.add(
                self.objects
                    .allocator()
                    .reserve(None, needed)
                    .ok_or(FxfsError::NoSpace)
                    .context("Failed to reserve metadata space in image builder mode")?
                    .forget(),
            );
        }

        self.write_super_block().await.context("Failed to write superblock")
    }

    pub async fn init_superblocks(&self) -> Result<(), Error> {
        // Overwrite both superblocks.
        for _ in 0..2 {
            self.write_super_block().await?;
        }
        Ok(())
    }

    /// Takes a snapshot of all journaled transactions which affect |object_id| since its last
    /// flush.
    /// The caller is responsible for locking; it must ensure that the journal is not trimmed during
    /// this call.  For example, a Flush lock could be held on the object in question (assuming that
    /// object has data to flush and is registered with ObjectManager).
    pub async fn read_transactions_for_object(
        &self,
        object_id: u64,
    ) -> Result<Vec<JournaledTransaction>, Error> {
        let handle = self.handle.get().expect("No journal handle");
        // Reopen the handle since JournalReader needs an owned handle.
        let handle = ObjectStore::open_object(
            handle.owner(),
            handle.object_id(),
            journal_handle_options(),
            None,
        )
        .await?;

        let checkpoint = match self.objects.journal_checkpoint(object_id) {
            Some(checkpoint) => checkpoint,
            None => return Ok(vec![]),
        };
        let mut reader = JournalReader::new(handle, &checkpoint);
        // Record the current end offset and only read to there, so we don't accidentally read any
        // partially flushed blocks.
        let end_offset = self.inner.lock().valid_to;
        Ok(self.read_transactions(&mut reader, Some(end_offset), object_id).await?.transactions)
    }

    /// Commits a transaction.  This is not thread safe; the caller must take appropriate locks.
    pub async fn commit(&self, transaction: &mut Transaction<'_>) -> Result<u64, Error> {
        if transaction.is_empty() {
            return Ok(self.inner.lock().writer.journal_file_checkpoint().file_offset);
        }

        self.pre_commit(transaction).await?;
        Ok(self.write_and_apply_mutations(transaction))
    }

    // Before we commit, we might need to extend the journal or write pending records to the
    // journal.
    async fn pre_commit(&self, _transaction: &Transaction<'_>) -> Result<(), Error> {
        let handle;

        let (size, zero_offset) = {
            let mut inner = self.inner.lock();

            // If this is the first write after a RESET, we need to output version first.
            if std::mem::take(&mut inner.output_reset_version) {
                LATEST_VERSION.serialize_into(&mut inner.writer)?;
            }

            if let Some(discard_offset) = inner.discard_offset {
                JournalRecord::Discard(discard_offset).serialize_into(&mut inner.writer)?;
                inner.discard_offset = None;
            }

            if inner.needs_did_flush_device {
                let offset = inner.device_flushed_offset;
                JournalRecord::DidFlushDevice(offset).serialize_into(&mut inner.writer)?;
                inner.needs_did_flush_device = false;
            }

            handle = match self.handle.get() {
                None => return Ok(()),
                Some(x) => x,
            };

            let file_offset = inner.writer.journal_file_checkpoint().file_offset;

            let size = handle.get_size();
            let size = if file_offset + self.chunk_size() > size { Some(size) } else { None };

            if size.is_none()
                && inner.zero_offset.is_none()
                && !self.objects.needs_borrow_for_journal(file_offset)
            {
                return Ok(());
            }

            (size, inner.zero_offset)
        };

        let mut transaction = handle
            .new_transaction_with_options(Options {
                skip_journal_checks: true,
                reservation: ReservationOptions::Borrowed,
                ..Default::default()
            })
            .await?;
        if let Some(size) = size {
            handle
                .preallocate_range(&mut transaction, &mut (size..size + self.chunk_size()))
                .await?;
        }
        if let Some(zero_offset) = zero_offset {
            handle.zero(&mut transaction, 0..zero_offset).await?;
        }

        // We can't use regular transaction commit, because that can cause re-entrancy issues, so
        // instead we just apply the transaction directly here.
        self.write_and_apply_mutations(&mut transaction);

        let mut inner = self.inner.lock();

        // Make sure the transaction to extend the journal made it to the journal within the old
        // size, since otherwise, it won't be possible to replay.
        if let Some(size) = size {
            assert!(inner.writer.journal_file_checkpoint().file_offset < size);
        }

        if inner.zero_offset == zero_offset {
            inner.zero_offset = None;
        }

        Ok(())
    }

    // Determines whether a mutation at the given checkpoint should be applied.  During replay, not
    // all records should be applied because the object store or allocator might already contain the
    // mutation.  After replay, that obviously isn't the case and we want to apply all mutations.
    fn should_apply(&self, object_id: u64, journal_file_checkpoint: &JournalCheckpoint) -> bool {
        let super_block_header = &self.inner.lock().super_block_header;
        let offset = super_block_header
            .journal_file_offsets
            .get(&object_id)
            .cloned()
            .unwrap_or(super_block_header.super_block_journal_file_offset);
        journal_file_checkpoint.file_offset >= offset
    }

    /// Flushes previous writes to the device and then writes out a new super-block.
    /// Callers must ensure that we do not make concurrent calls.
    async fn write_super_block(&self) -> Result<(), Error> {
        let root_parent_store = self.objects.root_parent_store();

        // We need to flush previous writes to the device since the new super-block we are writing
        // relies on written data being observable, and we also need to lock the root parent store
        // so that no new entries are written to it whilst we are writing the super-block, and for
        // that we use the write lock.
        let old_layers;
        let old_super_block_offset;
        let mut new_super_block_header;
        let checkpoint;
        let reserved_metadata_space;

        {
            let _sync_guard = debug_assert_not_too_long!(self.sync_mutex.lock());
            {
                let _write_guard = self.writer_mutex.lock();
                checkpoint = self.pad_to_block()?;
                reserved_metadata_space =
                    self.objects.reconcile_metadata_reservation(checkpoint.file_offset);
                old_layers = super_block::compact_root_parent(&*root_parent_store)?;
            }
            self.flush_device(checkpoint.file_offset)
                .await
                .context("flush failed when writing superblock")?;
        }

        new_super_block_header = self.inner.lock().super_block_header.clone();

        old_super_block_offset = new_super_block_header.journal_checkpoint.file_offset;

        let (journal_file_offsets, min_checkpoint) = self.objects.journal_file_offsets();

        new_super_block_header.generation = new_super_block_header.generation.wrapping_add(1);
        new_super_block_header.super_block_journal_file_offset = checkpoint.file_offset;
        new_super_block_header.journal_checkpoint = min_checkpoint.unwrap_or(checkpoint);
        new_super_block_header.journal_checkpoint.version = LATEST_VERSION;
        new_super_block_header.journal_file_offsets = journal_file_offsets;
        new_super_block_header.reserved_metadata_space = reserved_metadata_space;

        self.super_block_manager
            .save(
                new_super_block_header.clone(),
                self.objects.root_parent_store().filesystem(),
                old_layers,
            )
            .await?;
        {
            let mut inner = self.inner.lock();
            inner.super_block_header = new_super_block_header;
            inner.zero_offset = Some(BLOCK_SIZE.align_down(old_super_block_offset));
        }

        Ok(())
    }

    /// Flushes any buffered journal data to the device.  Note that this does not flush the device
    /// unless the flush_device option is set, in which case data should have been persisted to
    /// lower layers.  If a precondition is supplied, it is evaluated and the sync will be skipped
    /// if it returns false.  This allows callers to check a condition whilst a lock is held.  If a
    /// sync is performed, this function returns the checkpoint that was flushed.
    pub async fn sync(&self, options: SyncOptions<'_>) -> Result<Option<JournalCheckpoint>, Error> {
        let _guard = debug_assert_not_too_long!(self.sync_mutex.lock());

        let checkpoint = {
            if let Some(precondition) = options.precondition {
                if !precondition() {
                    return Ok(None);
                }
            }

            // This guard is required so that we don't insert an EndBlock record in the middle of a
            // transaction.
            let _guard = self.writer_mutex.lock();

            self.pad_to_block()?
        };

        if options.flush_device {
            self.flush_device(checkpoint.file_offset).await.context("sync: flush failed")?;
        }

        Ok(Some(checkpoint))
    }

    // Returns the checkpoint as it was prior to padding.  This is done because the super block
    // needs to record where the last transaction ends and it's the next transaction that pays the
    // price of the padding.
    fn pad_to_block(&self) -> Result<JournalCheckpoint, Error> {
        let mut inner = self.inner.lock();
        let checkpoint = inner.writer.journal_file_checkpoint();
        if checkpoint.file_offset % BLOCK_SIZE != 0 {
            JournalRecord::EndBlock.serialize_into(&mut inner.writer)?;
            inner.writer.pad_to_block()?;
            if let Some(waker) = inner.flush_waker.take() {
                waker.wake();
            }
        }
        Ok(checkpoint)
    }

    async fn flush_device(&self, checkpoint_offset: u64) -> Result<(), Error> {
        assert!(
            self.inner.lock().image_builder_mode.is_none(),
            "flush_device called in image builder mode"
        );
        debug_assert_not_too_long!(poll_fn(|ctx| {
            let mut inner = self.inner.lock();
            if inner.flushed_offset >= checkpoint_offset {
                Poll::Ready(Ok(()))
            } else if inner.terminate {
                let context = inner
                    .terminate_reason
                    .as_ref()
                    .map(|e| format!("Journal closed with error: {:?}", e))
                    .unwrap_or_else(|| "Journal closed".to_string());
                Poll::Ready(Err(anyhow!(FxfsError::JournalFlushError).context(context)))
            } else {
                inner.sync_waker = Some(ctx.waker().clone());
                Poll::Pending
            }
        }))?;

        let needs_flush = self.inner.lock().device_flushed_offset < checkpoint_offset;
        if needs_flush {
            let trace = self.trace.load(Ordering::Relaxed);
            if trace {
                info!("J: start flush device");
            }
            self.handle.get().unwrap().flush_device().await?;
            if trace {
                info!("J: end flush device");
            }

            // We need to write a DidFlushDevice record at some point, but if we are in the
            // process of shutting down the filesystem, we want to leave the journal clean to
            // avoid there being log messages complaining about unwritten journal data, so we
            // queue it up so that the next transaction will trigger this record to be written.
            // If we are shutting down, that will never happen but since the DidFlushDevice
            // message is purely advisory (it reduces the number of checksums we have to verify
            // during replay), it doesn't matter if it isn't written.
            {
                let mut inner = self.inner.lock();
                inner.device_flushed_offset = checkpoint_offset;
                inner.needs_did_flush_device = true;
            }

            // Tell the allocator that we flushed the device so that it can now start using
            // space that was deallocated.
            self.objects.allocator().did_flush_device(checkpoint_offset);
            if trace {
                info!("J: did flush device");
            }
        }

        Ok(())
    }

    /// Returns a copy of the super-block header.
    pub fn super_block_header(&self) -> SuperBlockHeader {
        self.inner.lock().super_block_header.clone()
    }

    /// Waits for there to be sufficient space in the journal.
    pub async fn check_journal_space(&self) -> Result<(), Error> {
        loop {
            let listener = {
                let inner = self.inner.lock();
                if inner.terminate {
                    // If the flush error is set, this will never make progress, since we can't
                    // extend the journal any more.
                    let context = inner
                        .terminate_reason
                        .as_ref()
                        .map(|e| format!("Journal closed with error: {:?}", e))
                        .unwrap_or_else(|| "Journal closed".to_string());
                    return Err(anyhow!(FxfsError::JournalFlushError).context(context));
                }
                if self.objects.last_end_offset()
                    - inner.super_block_header.journal_checkpoint.file_offset
                    < inner.reclaim_size
                {
                    return Ok(());
                }
                if inner.image_builder_mode.is_some() {
                    return Ok(());
                }
                if inner.disable_compactions {
                    return Err(
                        anyhow!(FxfsError::JournalFlushError).context("Compactions disabled")
                    );
                }
                self.reclaim_event.listen()
            };
            self.hooks.on_waiting_for_journal_space();
            debug_assert_not_too_long!(listener);
        }
    }

    fn chunk_size(&self) -> u64 {
        CHUNK_SIZE
    }

    fn write_and_apply_mutations(&self, transaction: &mut Transaction<'_>) -> u64 {
        let checkpoint_before;
        let checkpoint_after;
        let mut object_usages = SmallVec::<[(u64, u64); 4]>::new();
        {
            let _guard = self.writer_mutex.lock();
            checkpoint_before = {
                let mut inner = self.inner.lock();
                if transaction.includes_write() {
                    inner.needs_barrier = true;
                }
                let checkpoint = inner.writer.journal_file_checkpoint();
                let mut iter = transaction.mutations().iter().peekable();
                while let Some(store_iter) = ObjectMutationIterator::new(&mut iter) {
                    let object_id = store_iter.object_id();
                    let mut usage = 0;
                    self.objects.write_mutations(
                        object_id,
                        store_iter,
                        Writer { object_id, writer: &mut inner.writer, usage: &mut usage },
                    );
                    if usage > 0 {
                        object_usages.push((object_id, usage));
                    }
                }
                checkpoint
            };
            self.objects.apply_transaction(transaction, &checkpoint_before).expect(
                "apply_transaction should not fail in live mode; \
                 filesystem will be in an inconsistent state",
            );
            checkpoint_after = {
                let mut inner = self.inner.lock();
                for (owner_object_id, device_range, checksums, first_write) in
                    transaction.take_checksums().into_iter()
                {
                    inner
                        .writer
                        .write_record(&JournalRecord::DataChecksums(
                            owner_object_id,
                            device_range,
                            Checksums::fletcher(checksums),
                            first_write,
                        ))
                        .unwrap();
                }
                let commit_record = if transaction.metadata_reservation.is_borrowed() {
                    JournalRecord::CommitBorrowed(self.objects.metadata_reservation().amount())
                } else {
                    JournalRecord::Commit
                };
                inner.writer.write_record(&commit_record).unwrap();

                inner.writer.journal_file_checkpoint()
            };
        }
        self.objects.did_commit_transaction(
            transaction,
            checkpoint_after.file_offset,
            &object_usages,
        );

        if let Some(waker) = self.inner.lock().flush_waker.take() {
            waker.wake();
        }

        checkpoint_before.file_offset
    }

    /// This task will flush journal data to the device when there is data that needs flushing, and
    /// trigger compactions when short of journal space.  It will return after the terminate method
    /// has been called, or an error is encountered with either flushing or compaction.
    pub async fn flush_task(self: Arc<Self>) {
        let mut flush_fut = None;
        let mut compact_fut = None;
        let mut flush_error = false;
        poll_fn(|ctx| {
            loop {
                {
                    let mut inner = self.inner.lock();
                    if flush_fut.is_none() && !flush_error && self.handle.get().is_some() {
                        let flushable = inner.writer.flushable_bytes();
                        if flushable > 0 {
                            flush_fut = Some(Box::pin(self.flush(flushable)));
                        }
                    }
                    if inner.terminate && flush_fut.is_none() && compact_fut.is_none() {
                        return Poll::Ready(());
                    }
                    // `journal_bytes` refers to bytes in the journal that haven't yet been flushed
                    // to the layer files. It increases with each transaction and decreases when
                    // compactions complete. The flush task is woken whenever a transaction is
                    // committed and we should see this metric updated regularly.
                    let journal_bytes = self.objects.last_end_offset()
                        - inner.super_block_header.journal_checkpoint.file_offset;
                    fxfs_trace::counter!("journal-bytes", 0, "total" => journal_bytes);
                    // The / 2 is here because after compacting, we cannot reclaim the space until
                    // the _next_ time we flush the device since the super-block is not guaranteed
                    // to persist until then.
                    if compact_fut.is_none()
                        && !inner.compaction_state.is_running()
                        && !inner.terminate
                        && !inner.disable_compactions
                        && !inner.compactions_paused
                        && inner.image_builder_mode.is_none()
                        && journal_bytes > inner.reclaim_size / 2
                    {
                        compact_fut = Some(Box::pin(self.compact()));
                        inner.compaction_state = CompactionState::Running;
                    }
                    inner.flush_waker = Some(ctx.waker().clone());
                }
                let mut pending = true;
                if let Some(fut) = flush_fut.as_mut() {
                    if let Poll::Ready(result) = fut.poll_unpin(ctx) {
                        if let Err(e) = result {
                            self.inner.lock().terminate(Some(e.context("Flush error")));
                            self.reclaim_event.notify(usize::MAX);
                            flush_error = true;
                        }
                        flush_fut = None;
                        pending = false;
                    }
                }
                if let Some(fut) = compact_fut.as_mut() {
                    if let Poll::Ready(result) = fut.poll_unpin(ctx) {
                        let mut inner = self.inner.lock();
                        if let Err(e) = result {
                            inner.terminate(Some(e.context("Compaction error")));
                        }
                        compact_fut = None;
                        inner.compaction_state = CompactionState::Idle;
                        self.reclaim_event.notify(usize::MAX);
                        pending = false;
                        fxfs_trace::counter!(
                            "journal-bytes",
                            0,
                            "total" => self.objects.last_end_offset()
                                - inner.super_block_header.journal_checkpoint.file_offset
                        );
                    }
                }
                if pending {
                    return Poll::Pending;
                }
            }
        })
        .await;
    }

    /// Returns a yielder that can be used for compactions.
    pub fn get_compaction_yielder(&self) -> CompactionYielder<'_> {
        CompactionYielder::new(self)
    }

    async fn flush(&self, amount: usize) -> Result<(), Error> {
        let handle = self.handle.get().unwrap();
        let mut buf = handle.allocate_buffer(amount).await;
        let (offset, len, barrier_on_first_write) = {
            let mut inner = self.inner.lock();
            let offset = inner.writer.take_flushable(buf.as_mut());
            let barrier_on_first_write = inner.needs_barrier && inner.barriers_enabled;
            // Reset `needs_barrier` before instead of after the overwrite in case a txn commit
            // that contains data happens during the overwrite.
            inner.needs_barrier = false;
            (offset, buf.len() as u64, barrier_on_first_write)
        };
        self.handle
            .get()
            .unwrap()
            .overwrite(
                offset,
                buf.as_mut(),
                OverwriteOptions { barrier_on_first_write, ..Default::default() },
            )
            .await?;

        let mut inner = self.inner.lock();
        if let Some(waker) = inner.sync_waker.take() {
            waker.wake();
        }
        inner.flushed_offset = offset + len;
        inner.valid_to = inner.flushed_offset;
        Ok(())
    }

    #[trace]
    async fn compact(&self) -> Result<(), Error> {
        assert!(
            self.inner.lock().image_builder_mode.is_none(),
            "compact called in image builder mode"
        );
        let bytes_before = self.objects.compaction_bytes_written();
        let _measure = crate::metrics::DurationMeasureScope::new(
            &crate::metrics::lsm_tree_metrics().journal_compaction_time,
        );
        crate::metrics::lsm_tree_metrics().journal_compactions_total.add(1);
        let trace = self.trace.load(Ordering::Relaxed);
        debug!("Compaction starting");
        if trace {
            info!("J: start compaction");
        }
        let earliest_version = self
            .objects
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .context("Failed to flush objects")?;
        self.inner.lock().super_block_header.earliest_version = earliest_version;
        self.write_super_block().await.context("Failed to write superblock")?;
        if trace {
            info!("J: end compaction");
        }
        debug!("Compaction finished");
        let bytes_after = self.objects.compaction_bytes_written();
        crate::metrics::lsm_tree_metrics()
            .journal_compaction_bytes_written
            .add(bytes_after.saturating_sub(bytes_before));
        Ok(())
    }

    /// This should generally NOT be called externally. It is public to allow use by FIDL service
    /// fxfs.Debug.
    pub async fn force_compact(&self) -> Result<(), Error> {
        loop {
            debug_assert_not_too_long!({
                let mut inner = self.inner.lock();
                if !inner.compaction_state.is_running() {
                    inner.compaction_state = CompactionState::Forced;
                    break;
                }
                self.reclaim_event.listen()
            });
        }
        scopeguard::defer! {
            let mut inner = self.inner.lock();
            inner.compaction_state = CompactionState::Idle;
            if let Some(waker) = inner.flush_waker.take() {
                waker.wake();
            }
            self.reclaim_event.notify(usize::MAX);
        }
        self.compact().await
    }

    pub async fn stop_compactions(&self) {
        loop {
            debug_assert_not_too_long!({
                let mut inner = self.inner.lock();
                inner.disable_compactions = true;
                if !inner.compaction_state.is_running() {
                    return;
                }
                self.reclaim_event.listen()
            });
        }
    }

    pub async fn pause_compactions(&self) {
        loop {
            debug_assert_not_too_long!({
                let mut inner = self.inner.lock();
                inner.compactions_paused = true;
                if !inner.compaction_state.is_running() {
                    return;
                }
                self.reclaim_event.listen()
            });
        }
    }

    pub fn resume_compactions(&self) {
        let mut inner = self.inner.lock();
        inner.compactions_paused = false;
        if let Some(waker) = inner.flush_waker.take() {
            waker.wake();
        }
    }

    /// Creates a lazy inspect node named `str` under `parent` which will yield statistics for the
    /// journal when queried.
    pub fn track_statistics(self: &Arc<Self>, parent: &fuchsia_inspect::Node, name: &str) {
        let this = Arc::downgrade(self);
        parent.record_lazy_child(name, move || {
            let this_clone = this.clone();
            async move {
                let inspector = fuchsia_inspect::Inspector::default();
                if let Some(this) = this_clone.upgrade() {
                    let (journal_min, journal_max, journal_reclaim_size) = {
                        // TODO(https://fxbug.dev/42069513): Push-back or rate-limit to prevent DoS.
                        let inner = this.inner.lock();
                        (
                            BLOCK_SIZE.align_down(
                                inner.super_block_header.journal_checkpoint.file_offset,
                            ),
                            inner.flushed_offset,
                            inner.reclaim_size,
                        )
                    };
                    let root = inspector.root();
                    root.record_uint("journal_min_offset", journal_min);
                    root.record_uint("journal_max_offset", journal_max);
                    root.record_uint("journal_size", journal_max - journal_min);
                    root.record_uint("journal_reclaim_size", journal_reclaim_size);

                    // TODO(https://fxbug.dev/42068224): Post-compute rather than manually computing metrics.
                    if let Some(x) = round_div(
                        100 * (journal_max - journal_min),
                        this.objects.allocator().get_disk_bytes(),
                    ) {
                        root.record_uint("journal_size_to_disk_size_percent", x);
                    }
                }
                Ok(inspector)
            }
            .boxed()
        });
    }

    /// Terminate all journal activity.
    pub fn terminate(&self) {
        self.inner.lock().terminate(/*reason*/ None);
        self.reclaim_event.notify(usize::MAX);
    }
}

/// Wrapper to allow records to be written to the journal.
pub struct Writer<'a> {
    object_id: u64,
    writer: &'a mut JournalWriter,
    usage: &'a mut u64,
}

impl Writer<'_> {
    pub fn write(&mut self, mutation: Mutation) {
        let count_usage = mutation.counts_towards_object_journal_usage();
        let offset_before = self.writer.journal_file_checkpoint().file_offset;
        self.writer
            .write_record(&JournalRecord::Mutation { object_id: self.object_id, mutation })
            .unwrap();
        if count_usage {
            *self.usage += self.writer.journal_file_checkpoint().file_offset - offset_before;
        }
    }
}

#[cfg(target_os = "fuchsia")]
mod yielder {
    use super::{CompactionState, Journal};
    use crate::lsm_tree::Yielder;
    use fuchsia_async as fasync;

    /// CompactionYielder uses fuchsia-async to yield if other tasks are being polled, which should
    /// be a proxy for how busy the system is.  We can afford to delay compactions for a small
    /// amount of time but not so long that we end up blocking new transactions.
    pub struct CompactionYielder<'a> {
        journal: &'a Journal,
        low_priority_task: Option<fasync::LowPriorityTask>,
    }

    impl<'a> CompactionYielder<'a> {
        pub fn new(journal: &'a Journal) -> Self {
            Self { journal, low_priority_task: None }
        }
    }

    impl Yielder for CompactionYielder<'_> {
        async fn yield_now(&mut self) {
            // We will wait for the executor to be idle for 4ms, but no longer than 16ms.  We need
            // to cap the maximum amount of time we wait in case we've reached a point where
            // compaction is now urgent or else we could block new transactions.
            const IDLE_PERIOD: zx::MonotonicDuration = zx::MonotonicDuration::from_millis(4);
            const MAX_YIELD_DURATION: zx::MonotonicDuration =
                zx::MonotonicDuration::from_millis(16);

            {
                let inner = self.journal.inner.lock();
                if inner.compaction_state == CompactionState::Forced {
                    return;
                }
                let outstanding = self.journal.objects.last_end_offset()
                    - inner.super_block_header.journal_checkpoint.file_offset;
                let half_reclaim_size = inner.reclaim_size / 2;
                if outstanding
                    .checked_sub(half_reclaim_size)
                    .is_some_and(|x| x >= half_reclaim_size / 2)
                {
                    // If we have got to the point where we have used up 3/4 of reclaim size in the
                    // journal, do not delay any further.  If we continue to yield we will get to
                    // the point where we block new transactions.
                    self.low_priority_task = None;
                    return;
                }
            }

            self.low_priority_task
                .get_or_insert_with(|| fasync::LowPriorityTask::new())
                .wait_until_idle_for(
                    IDLE_PERIOD,
                    fasync::MonotonicInstant::after(MAX_YIELD_DURATION),
                )
                .await;
        }
    }
}

#[cfg(not(target_os = "fuchsia"))]
mod yielder {
    use super::Journal;
    use crate::lsm_tree::Yielder;

    #[expect(dead_code)]
    pub struct CompactionYielder<'a>(&'a Journal);

    impl<'a> CompactionYielder<'a> {
        pub fn new(journal: &'a Journal) -> Self {
            Self(journal)
        }
    }

    impl Yielder for CompactionYielder<'_> {
        async fn yield_now(&mut self) {}
    }
}

pub use yielder::*;

#[cfg(test)]
mod tests {
    use super::{JournalRecord, JournalRecordV59, SuperBlockInstance};
    use crate::checksum::Checksums;
    use crate::errors::FxfsError;
    use crate::filesystem::{
        FlushReason, ForceMajor, FxFilesystem, FxFilesystemBuilder, SyncOptions,
    };
    use crate::fsck::fsck;
    use crate::object_handle::{
        INVALID_OBJECT_ID, ObjectHandle, ReadObjectHandle, WriteObjectHandle,
    };
    use crate::object_store::directory::{Directory, replace_child};
    use crate::object_store::object_record::{ObjectKey, ObjectValue};
    use crate::object_store::transaction::{Mutation, Options, ReservationOptions};
    use crate::object_store::volume::root_volume;
    use crate::object_store::{
        HandleOptions, LockKey, NewChildStoreOptions, ObjectStore, StoreOptions, lock_keys,
    };
    use crate::serialized_types::Version;
    #[cfg(target_os = "fuchsia")]
    use fuchsia_async::TestExecutor;
    use fuchsia_async::{self as fasync, MonotonicDuration};
    use fxfs_insecure_crypto::new_insecure_crypt;
    use rustc_hash::FxHashMap as HashMap;
    use std::sync::Arc;
    use storage_device::DeviceHolder;
    use storage_device::fake_device::FakeDevice;

    const TEST_DEVICE_BLOCK_SIZE: u32 = 512;

    #[fuchsia::test]
    async fn test_replay() {
        const TEST_DATA: &[u8] = b"hello";

        let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));

        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");

        let object_id = {
            let root_store = fs.root_store();
            let root_directory =
                Directory::open(&root_store, root_store.root_directory_object_id())
                    .await
                    .expect("open failed");
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        root_store.store_object_id(),
                        root_store.root_directory_object_id(),
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test")
                .await
                .expect("create_child_file failed");

            transaction.commit().await.expect("commit failed");
            let mut buf = handle.allocate_buffer(TEST_DATA.len()).await;
            buf.copy_from_slice(TEST_DATA);
            handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
            // As this is the first sync, this will actually trigger a new super-block, but normally
            // this would not be the case.
            fs.sync(SyncOptions::default()).await.expect("sync failed");
            handle.object_id()
        };

        {
            fs.close().await.expect("Close failed");
            let device = fs.take_device().await;
            device.reopen(false);
            let fs = FxFilesystem::open(device).await.expect("open failed");
            let handle = ObjectStore::open_object(
                &fs.root_store(),
                object_id,
                HandleOptions::default(),
                None,
            )
            .await
            .expect("open_object failed");
            let mut buf = handle.allocate_buffer(handle.block_size().get() as usize).await;
            assert_eq!(
                handle.read_aligned(0, buf.as_mut()).await.expect("read failed"),
                TEST_DATA.len()
            );
            assert_eq!(buf.subslice(..TEST_DATA.len()).to_vec(), TEST_DATA);
            fsck(fs.clone()).await.expect("fsck failed");
            fs.close().await.expect("Close failed");
        }
    }

    #[fuchsia::test]
    async fn test_reset() {
        const TEST_DATA: &[u8] = b"hello";

        let device = DeviceHolder::new(FakeDevice::new(32768, TEST_DEVICE_BLOCK_SIZE));

        let mut object_ids = Vec::new();

        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        {
            let root_store = fs.root_store();
            let root_directory =
                Directory::open(&root_store, root_store.root_directory_object_id())
                    .await
                    .expect("open failed");
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        root_store.store_object_id(),
                        root_store.root_directory_object_id(),
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
            let mut buf = handle.allocate_buffer(TEST_DATA.len()).await;
            buf.copy_from_slice(TEST_DATA);
            handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
            fs.sync(SyncOptions::default()).await.expect("sync failed");
            object_ids.push(handle.object_id());

            // Create a lot of objects but don't sync at the end. This should leave the filesystem
            // with a half finished transaction that cannot be replayed.
            for i in 0..1000 {
                let mut transaction = fs
                    .root_store()
                    .new_transaction(
                        lock_keys![LockKey::object(
                            root_store.store_object_id(),
                            root_store.root_directory_object_id(),
                        )],
                        Options::default(),
                    )
                    .await
                    .expect("new_transaction failed");
                let handle = root_directory
                    .create_child_file(&mut transaction, &format!("{}", i))
                    .await
                    .expect("create_child_file failed");
                transaction.commit().await.expect("commit failed");
                let mut buf = handle.allocate_buffer(TEST_DATA.len()).await;
                buf.copy_from_slice(TEST_DATA);
                handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
                object_ids.push(handle.object_id());
            }
        }
        fs.close().await.expect("fs close failed");
        let device = fs.take_device().await;
        device.reopen(false);
        let fs = FxFilesystem::open(device).await.expect("open failed");
        fsck(fs.clone()).await.expect("fsck failed");
        {
            let root_store = fs.root_store();
            // Check the first two objects which should exist.
            for &object_id in &object_ids[0..1] {
                let handle = ObjectStore::open_object(
                    &root_store,
                    object_id,
                    HandleOptions::default(),
                    None,
                )
                .await
                .expect("open_object failed");
                let mut buf = handle.allocate_buffer(handle.block_size().get() as usize).await;
                assert_eq!(
                    handle.read_aligned(0, buf.as_mut()).await.expect("read failed"),
                    TEST_DATA.len()
                );
                let data = buf.subslice(..TEST_DATA.len()).to_vec();
                assert_eq!(data, TEST_DATA);
            }

            // Write one more object and sync.
            let root_directory =
                Directory::open(&root_store, root_store.root_directory_object_id())
                    .await
                    .expect("open failed");
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        root_store.store_object_id(),
                        root_store.root_directory_object_id(),
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test2")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
            let mut buf = handle.allocate_buffer(TEST_DATA.len()).await;
            buf.copy_from_slice(TEST_DATA);
            handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
            fs.sync(SyncOptions::default()).await.expect("sync failed");
            object_ids.push(handle.object_id());
        }

        fs.close().await.expect("close failed");
        let device = fs.take_device().await;
        device.reopen(false);
        let fs = FxFilesystem::open(device).await.expect("open failed");
        {
            fsck(fs.clone()).await.expect("fsck failed");

            // Check the first two and the last objects.
            for &object_id in object_ids[0..1].iter().chain(object_ids.last().cloned().iter()) {
                let handle = ObjectStore::open_object(
                    &fs.root_store(),
                    object_id,
                    HandleOptions::default(),
                    None,
                )
                .await
                .unwrap_or_else(|e| {
                    panic!("open_object failed (object_id: {}): {:?}", object_id, e)
                });
                let mut buf = handle.allocate_buffer(handle.block_size().get() as usize).await;
                assert_eq!(
                    handle.read_aligned(0, buf.as_mut()).await.expect("read failed"),
                    TEST_DATA.len()
                );
                let data = buf.subslice(..TEST_DATA.len()).to_vec();
                assert_eq!(data, TEST_DATA);
            }
        }
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_discard() {
        let device = {
            let device = DeviceHolder::new(FakeDevice::new(8192, 4096));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

            let store = root_volume
                .new_volume("test", NewChildStoreOptions::default())
                .await
                .expect("new_volume failed");
            let root_directory = Directory::open(&store, store.root_directory_object_id())
                .await
                .expect("open failed");

            // Create enough data so that another journal extent is used.
            let mut i = 0;
            loop {
                let mut transaction = fs
                    .root_store()
                    .new_transaction(
                        lock_keys![LockKey::object(
                            store.store_object_id(),
                            store.root_directory_object_id()
                        )],
                        Options::default(),
                    )
                    .await
                    .expect("new_transaction failed");
                root_directory
                    .create_child_file(&mut transaction, &format!("a {i}"))
                    .await
                    .expect("create_child_file failed");
                if transaction.commit().await.expect("commit failed") > super::CHUNK_SIZE {
                    break;
                }
                i += 1;
            }

            // Compact and then disable compactions.
            fs.journal().force_compact().await.expect("compact failed");
            fs.journal().stop_compactions().await;

            // Keep going until we need another journal extent.
            let mut i = 0;
            loop {
                let mut transaction = fs
                    .root_store()
                    .new_transaction(
                        lock_keys![LockKey::object(
                            store.store_object_id(),
                            store.root_directory_object_id()
                        )],
                        Options::default(),
                    )
                    .await
                    .expect("new_transaction failed");
                root_directory
                    .create_child_file(&mut transaction, &format!("b {i}"))
                    .await
                    .expect("create_child_file failed");
                if transaction.commit().await.expect("commit failed") > 2 * super::CHUNK_SIZE {
                    break;
                }
                i += 1;
            }

            // Allow the journal to flush, but we don't want to sync.
            fasync::Timer::new(MonotonicDuration::from_millis(10)).await;
            // Because we're not gracefully closing the filesystem, a Discard record will be
            // emitted.
            fs.device().snapshot().expect("snapshot failed")
        };

        let fs = FxFilesystem::open(device).await.expect("open failed");

        {
            let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

            let store =
                root_volume.volume("test", StoreOptions::default()).await.expect("volume failed");

            let root_directory = Directory::open(&store, store.root_directory_object_id())
                .await
                .expect("open failed");

            // Write one more transaction.
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        store.store_object_id(),
                        store.root_directory_object_id()
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            root_directory
                .create_child_file(&mut transaction, &format!("d"))
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
        }

        fs.close().await.expect("close failed");
        let device = fs.take_device().await;
        device.reopen(false);

        let fs = FxFilesystem::open(device).await.expect("open failed");
        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_use_existing_generation() {
        let device = DeviceHolder::new(FakeDevice::new(8192, 4096));

        // First format should be generation 1.
        let fs = FxFilesystemBuilder::new()
            .format(true)
            .image_builder_mode(Some(SuperBlockInstance::A))
            .open(device)
            .await
            .expect("open failed");
        fs.enable_allocations();
        let generation0 = fs.super_block_header().generation;
        assert_eq!(generation0, 1);
        fs.close().await.expect("close failed");
        let device = fs.take_device().await;
        device.reopen(false);

        // Format the device normally (again, generation 1).
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let generation1 = fs.super_block_header().generation;
        {
            let root_volume = crate::object_store::volume::root_volume(fs.clone())
                .await
                .expect("root_volume failed");
            root_volume
                .new_volume("test", crate::object_store::NewChildStoreOptions::default())
                .await
                .expect("new_volume failed");
        }
        fs.close().await.expect("close failed");
        let device = fs.take_device().await;
        device.reopen(false);

        // Format again with image_builder_mode.
        let fs = FxFilesystemBuilder::new()
            .format(true)
            .image_builder_mode(Some(SuperBlockInstance::A))
            .open(device)
            .await
            .expect("open failed");
        fs.enable_allocations();
        let generation2 = fs.super_block_header().generation;
        assert!(
            generation2 > generation1,
            "generation2 ({}) should be greater than generation1 ({})",
            generation2,
            generation1
        );
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_image_builder_mode_generation_bump_512_byte_block() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 512));

        // Format initial filesystem (generation 1)
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let generation1 = fs.super_block_header().generation;
        fs.close().await.expect("close failed");
        let device = fs.take_device().await;
        device.reopen(false);

        // Format again with image_builder_mode (should bump generation)
        let fs = FxFilesystemBuilder::new()
            .format(true)
            .image_builder_mode(Some(SuperBlockInstance::A))
            .open(device)
            .await
            .expect("open failed");

        fs.enable_allocations();
        let generation2 = fs.super_block_header().generation;
        assert!(
            generation2 > generation1,
            "Expected generation bump, got {} vs {}",
            generation2,
            generation1
        );
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    #[cfg(target_os = "fuchsia")]
    fn test_low_priority_compaction() {
        let mut executor = TestExecutor::new_with_fake_time();
        let mut fut = std::pin::pin!(async {
            use std::sync::Arc;
            use std::sync::atomic::{AtomicBool, Ordering};

            let device = DeviceHolder::new(FakeDevice::new(16384, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystemBuilder::new()
                .journal_options(super::JournalOptions {
                    reclaim_size: 65536,
                    ..Default::default()
                })
                .format(true)
                .open(device)
                .await
                .expect("open failed");

            let _low = fasync::LowPriorityTask::new();

            // Add some data to the tree.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                for i in 0..100 {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("test{}", i))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");
                }
            }

            // Spawn a task that polls every 1ms.
            let stop = Arc::new(AtomicBool::new(false));
            let stop_clone = stop.clone();
            let _normal_task = fasync::Task::spawn(async move {
                while !stop_clone.load(Ordering::Relaxed) {
                    fasync::Timer::new(fasync::MonotonicInstant::after(
                        MonotonicDuration::from_millis(1),
                    ))
                    .await;
                }
            });

            // Trigger journal compaction.
            // We can do this by writing more data until outstanding > reclaim_size / 2.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                let mut i = 0;
                loop {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("trigger{i}"))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");

                    if fs.journal().inner.lock().compaction_state.is_running() {
                        break;
                    }
                    TestExecutor::advance_to(fasync::MonotonicInstant::after(
                        MonotonicDuration::from_millis(1),
                    ))
                    .await;
                    i += 1;
                }
            }

            // Compaction should now be running. Because of our 1ms poller, it should be yielding.
            // It will yield for 16ms at each point.
            for _ in 0..10 {
                TestExecutor::advance_to(fasync::MonotonicInstant::after(
                    MonotonicDuration::from_millis(1),
                ))
                .await;
                assert!(fs.journal().inner.lock().compaction_state.is_running());
            }

            // Stop the normal task.
            stop.store(true, Ordering::Relaxed);
            TestExecutor::advance_to(fasync::MonotonicInstant::after(
                MonotonicDuration::from_millis(1),
            ))
            .await;

            // Compaction should still be running because it hasn't been 4 ms since the normal task
            // finished.
            assert!(fs.journal().inner.lock().compaction_state.is_running());

            // For the next 3ms, compaction should still be running.
            for _ in 0..3 {
                TestExecutor::advance_to(fasync::MonotonicInstant::after(
                    MonotonicDuration::from_millis(1),
                ))
                .await;
                assert!(fs.journal().inner.lock().compaction_state.is_running());
            }

            // 1 more ms and compaction should be unblocked.
            TestExecutor::advance_to(fasync::MonotonicInstant::after(
                MonotonicDuration::from_millis(1),
            ))
            .await;

            // When the executor next stalls, compaction should be done.
            let _ = TestExecutor::poll_until_stalled(std::future::pending::<()>()).await;
            assert!(!fs.journal().inner.lock().compaction_state.is_running());

            fs.close().await.expect("Close failed");
        });
        assert!(executor.run_until_stalled(&mut fut).is_ready());
    }

    #[fuchsia::test]
    #[cfg(target_os = "fuchsia")]
    fn test_low_priority_compaction_deadline() {
        let mut executor = TestExecutor::new_with_fake_time();
        let mut fut = std::pin::pin!(async {
            use std::sync::Arc;
            use std::sync::atomic::{AtomicBool, Ordering};

            let device = DeviceHolder::new(FakeDevice::new(216384, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystemBuilder::new()
                .journal_options(super::JournalOptions {
                    reclaim_size: 65536,
                    ..Default::default()
                })
                .format(true)
                .open(device)
                .await
                .expect("open failed");

            let _low = fasync::LowPriorityTask::new();

            // Add some data to the tree.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                for i in 0..10 {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("test{}", i))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");
                }
            }

            // Spawn a task that polls every 3ms.
            let stop = Arc::new(AtomicBool::new(false));
            let stop_clone = stop.clone();
            let _normal_task = fasync::Task::spawn(async move {
                while !stop_clone.load(Ordering::Relaxed) {
                    fasync::Timer::new(fasync::MonotonicInstant::after(
                        MonotonicDuration::from_millis(3),
                    ))
                    .await;
                }
            });

            // Trigger journal compaction.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                let mut i = 0;
                loop {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("trigger{i}"))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");

                    if fs.journal().inner.lock().compaction_state.is_running() {
                        break;
                    }
                    TestExecutor::advance_to(fasync::MonotonicInstant::after(
                        MonotonicDuration::from_millis(1),
                    ))
                    .await;
                    i += 1;
                }
            }

            // Advance time in 20ms increments. Each increment should allow compaction to make progress
            // on one item (since MAX_YIELD_DURATION is 16ms).
            let mut count = 0;
            for _ in 0..1000 {
                TestExecutor::advance_to(fasync::MonotonicInstant::after(
                    MonotonicDuration::from_millis(20),
                ))
                .await;
                if !fs.journal().inner.lock().compaction_state.is_running() {
                    break;
                }
                count += 1;
            }
            assert!(!fs.journal().inner.lock().compaction_state.is_running());

            // Make sure it took a few iterations to complete.  It's difficult to know what the exact
            // number should be.
            assert!(count > 200);

            stop.store(true, Ordering::Relaxed);
            fs.close().await.expect("Close failed");
        });
        assert!(executor.run_until_stalled(&mut fut).is_ready());
    }

    #[fuchsia::test]
    #[cfg(target_os = "fuchsia")]
    fn test_low_priority_compaction_no_yielding_when_full() {
        let mut executor = TestExecutor::new_with_fake_time();
        let mut fut = std::pin::pin!(async {
            use std::sync::Arc;
            use std::sync::atomic::{AtomicBool, Ordering};

            let reclaim_size = 65536;
            let device = DeviceHolder::new(FakeDevice::new(216384, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystemBuilder::new()
                .journal_options(super::JournalOptions { reclaim_size, ..Default::default() })
                .format(true)
                .open(device)
                .await
                .expect("open failed");

            let _low = fasync::LowPriorityTask::new();

            // Add some data to the tree.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                for i in 0..10 {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("test{}", i))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");
                }
            }

            // Spawn a task that polls every 3ms.
            let stop = Arc::new(AtomicBool::new(false));
            let stop_clone = stop.clone();
            let _normal_task = fasync::Task::spawn(async move {
                while !stop_clone.load(Ordering::Relaxed) {
                    fasync::Timer::new(fasync::MonotonicInstant::after(
                        MonotonicDuration::from_millis(3),
                    ))
                    .await;
                }
            });

            // Trigger journal compaction, but this time we fill it up to 3/4 full.
            {
                let root_store = fs.root_store();
                let root_directory =
                    Directory::open(&root_store, root_store.root_directory_object_id())
                        .await
                        .expect("open failed");
                let mut i = 0;
                loop {
                    let mut transaction = fs
                        .root_store()
                        .new_transaction(
                            lock_keys![LockKey::object(
                                root_store.store_object_id(),
                                root_store.root_directory_object_id(),
                            )],
                            Options::default(),
                        )
                        .await
                        .expect("new_transaction failed");
                    root_directory
                        .create_child_file(&mut transaction, &format!("trigger{i}"))
                        .await
                        .expect("create_child_file failed");
                    transaction.commit().await.expect("commit failed");

                    let outstanding = {
                        let inner = fs.journal().inner.lock();
                        fs.journal().objects.last_end_offset()
                            - inner.super_block_header.journal_checkpoint.file_offset
                    };
                    if outstanding >= reclaim_size * 7 / 8 {
                        break;
                    }
                    // We don't advance time here to try and reach 3/4 before compaction can yield.
                    i += 1;
                }
            }

            // Advancing by 4ms should be enough to wake compaction up.
            TestExecutor::advance_to(fasync::MonotonicInstant::after(
                MonotonicDuration::from_millis(4),
            ))
            .await;

            // When the executor next stalls, compaction should be done.
            let _ = TestExecutor::poll_until_stalled(std::future::pending::<()>()).await;
            assert!(!fs.journal().inner.lock().compaction_state.is_running());

            stop.store(true, Ordering::Relaxed);
            fs.close().await.expect("Close failed");
        });
        assert!(executor.run_until_stalled(&mut fut).is_ready());
    }

    #[fuchsia::test]
    async fn test_data_checksums_validation() {
        // 1. A < V60 DataChecksums record (which migrates with `owner_object_id =
        //    INVALID_OBJECT_ID`) is accepted during replay.
        {
            let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);

            let fs = FxFilesystem::open(device).await.expect("open failed");
            let _ = fs
                .root_store()
                .new_transaction(lock_keys![], Options::default())
                .await
                .expect("new_transaction failed");
            {
                let mut inner = fs.journal().inner.lock();
                assert!(std::mem::take(&mut inner.output_reset_version));
                Version { major: 59, minor: 0 }.serialize_into(&mut inner.writer).unwrap();
                inner
                    .writer
                    .write_record(&JournalRecordV59::DataChecksums(
                        512..1024,
                        Checksums::fletcher(vec![0]),
                        false,
                    ))
                    .unwrap();
                inner.writer.write_record(&JournalRecordV59::Commit).unwrap();
            }
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);

            let fs = FxFilesystem::open(device).await.expect("open V59 DataChecksums failed");
            fs.close().await.expect("close failed");
        }

        // 2. A >= V60 DataChecksums record with `owner_object_id == INVALID_OBJECT_ID` is rejected.
        {
            let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            {
                let mut inner = fs.journal().inner.lock();
                inner
                    .writer
                    .write_record(&JournalRecord::DataChecksums(
                        INVALID_OBJECT_ID,
                        512..1024,
                        Checksums::fletcher(vec![0]),
                        false,
                    ))
                    .unwrap();
                inner.writer.write_record(&JournalRecord::Commit).unwrap();
            }
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);

            let err = FxFilesystem::open(device)
                .await
                .err()
                .expect("expected INVALID_OBJECT_ID in V60+ DataChecksums to fail");
            assert!(FxfsError::Inconsistent.matches(&err), "unexpected error: {err:?}");
        }

        // 3. A DataChecksums record for a deleted store is rejected.
        {
            let device = DeviceHolder::new(FakeDevice::new(8192, TEST_DEVICE_BLOCK_SIZE));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            {
                let mut inner = fs.journal().inner.lock();
                inner
                    .writer
                    .write_record(&JournalRecord::Mutation {
                        object_id: 99,
                        mutation: Mutation::DeleteVolume,
                    })
                    .unwrap();
                inner.writer.write_record(&JournalRecord::Commit).unwrap();
                inner
                    .writer
                    .write_record(&JournalRecord::DataChecksums(
                        99,
                        512..1024,
                        Checksums::fletcher(vec![0]),
                        false,
                    ))
                    .unwrap();
                inner.writer.write_record(&JournalRecord::Commit).unwrap();
            }
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);

            let err = FxFilesystem::open(device)
                .await
                .err()
                .expect("expected DataChecksums for deleted store to fail");
            assert!(FxfsError::Inconsistent.matches(&err), "unexpected error: {err:?}");
        }
    }

    #[fuchsia::test]
    async fn test_delete_volume_reused_extents_before_did_flush_device() {
        let device = DeviceHolder::new(FakeDevice::new(4096, TEST_DEVICE_BLOCK_SIZE));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_vol = root_volume(fs.clone()).await.expect("root_volume failed");

        // Create "vol1" and "vol2" (with an empty file "bar" in "vol2") up front so no metadata
        // allocations occur between deleting "vol1" and writing to "bar".
        let vol1 =
            root_vol.new_volume("vol1", NewChildStoreOptions::default()).await.expect("new_volume");
        let vol2 =
            root_vol.new_volume("vol2", NewChildStoreOptions::default()).await.expect("new_volume");
        let vol2_root =
            Directory::open(&vol2, vol2.root_directory_object_id()).await.expect("open failed");
        let mut transaction = fs
            .root_store()
            .new_transaction(
                lock_keys![LockKey::object(
                    vol2.store_object_id(),
                    vol2.root_directory_object_id()
                )],
                Options::default(),
            )
            .await
            .expect("new_transaction failed");
        let bar =
            vol2_root.create_child_file(&mut transaction, "bar").await.expect("create_child_file");
        transaction.commit().await.expect("commit failed");

        // Write 64 KiB to a file in "vol1", emitting `DataChecksums` owned by "vol1".
        {
            let vol1_root =
                Directory::open(&vol1, vol1.root_directory_object_id()).await.expect("open failed");
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        vol1.store_object_id(),
                        vol1.root_directory_object_id()
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let file = vol1_root
                .create_child_file(&mut transaction, "file")
                .await
                .expect("create_child_file");
            transaction.commit().await.expect("commit failed");
            let mut buf = file.allocate_buffer(65536).await;
            buf.fill(0xaa);
            file.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
        }
        drop(vol1);

        // Delete "vol1", which emits `MarkForDeletion` for "vol1".
        let (_, transaction) = root_vol
            .acquire_transaction_for_remove_volume("vol1", [], false, Options::default())
            .await
            .expect("acquire_transaction_for_remove_volume failed");
        root_vol.delete_volume("vol1", transaction, || {}).await.expect("delete_volume failed");

        // Write 64 KiB to "bar" in "vol2". Allocating in "vol2" sees
        // `volumes_deleted_pending_sync`, syncs the journal and flushes the device (queuing
        // `DidFlushDevice` for the next journal write), and frees "vol1"'s 64 KiB extent. Writing
        // to "bar" reuses and overwrites those exact disk blocks with `0xbb`, while the
        // `DidFlushDevice` record remains buffered in memory in `JournalWriter`.
        let mut buf = bar.allocate_buffer(65536).await;
        buf.fill(0xbb);
        bar.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");

        // Simulate power loss before `DidFlushDevice` is flushed to the journal on disk.
        let snapshot = fs.device().snapshot().expect("snapshot failed");
        let fs2 = FxFilesystem::open(snapshot).await.expect("open snapshot failed");
        fsck(fs2.clone()).await.expect("fsck failed");
        let root_vol2 = root_volume(fs2.clone()).await.expect("root_volume failed");
        root_vol2
            .volume("vol1", StoreOptions::default())
            .await
            .err()
            .expect("vol1 should remain deleted");
        fs2.close().await.expect("close failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_metadata_reservation_replay_and_reserved_space_change() {
        let device = DeviceHolder::new(FakeDevice::new(8192, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");
        let store = root_volume
            .new_volume("test", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_id = store.store_object_id();
        let root_directory =
            Directory::open(&store, store.root_directory_object_id()).await.expect("open failed");

        let other_store = root_volume
            .new_volume("other", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let other_store_id = other_store.store_object_id();
        let other_root_directory =
            Directory::open(&other_store, other_store.root_directory_object_id())
                .await
                .expect("open failed");

        let object_id = {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(store_id, store.root_directory_object_id())],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "to_delete")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
            let mut buf = handle.allocate_buffer(4096).await;
            buf.fill(0xab);
            handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
            handle.object_id()
        };

        {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(
                        other_store_id,
                        other_store.root_directory_object_id()
                    )],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            other_root_directory
                .create_child_file(&mut transaction, "other_file")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
        }

        // Compact so the super-block records the current actual reserved metadata space.
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        let (store_borrowed_after_compact, store_req_after_compact) =
            fs.object_manager().store_borrowed(store_id).unwrap();
        assert_eq!(store_borrowed_after_compact, 0);
        assert!(store_req_after_compact > 0);
        let (other_borrowed_after_compact, other_req_after_compact) =
            fs.object_manager().store_borrowed(other_store_id).unwrap();
        assert_eq!(other_borrowed_after_compact, 0);
        assert!(other_req_after_compact > 0);

        // Commit a non-borrowing transaction followed by borrowing transactions (unlink +
        // tombstone) in `store` after the super-block, leaving `other_store` untouched.
        {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(store_id, store.root_directory_object_id())],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            root_directory
                .create_child_file(&mut transaction, "keep")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");

            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![
                        LockKey::object(store_id, store.root_directory_object_id()),
                        LockKey::object(store_id, object_id),
                    ],
                    Options { reservation: ReservationOptions::Borrowed, ..Default::default() },
                )
                .await
                .expect("new_transaction failed");
            replace_child(&mut transaction, None, (&root_directory, "to_delete"))
                .await
                .expect("replace_child failed");
            transaction.commit().await.expect("commit failed");

            fs.tombstone_object(store_id, object_id, None).await.expect("tombstone_object failed");
        }

        fs.sync(SyncOptions { flush_device: true, ..Default::default() })
            .await
            .expect("sync failed");

        let expected_reservation = fs.object_manager().metadata_reservation().amount();
        let expected_borrowed = fs.object_manager().borrowed_metadata_space();
        let expected_reserved_map = fs.object_manager().reserved_metadata_space();
        assert!(expected_borrowed > 0);

        let (expected_store_borrowed, expected_store_req) =
            fs.object_manager().store_borrowed(store_id).unwrap();
        assert!(expected_store_borrowed > 0 && expected_store_borrowed < expected_store_req);
        let (expected_other_borrowed, expected_other_req) =
            fs.object_manager().store_borrowed(other_store_id).unwrap();
        assert_eq!(expected_other_borrowed, 0);

        let snapshot1 = fs.device().snapshot().expect("snapshot failed");
        let snapshot2 = fs.device().snapshot().expect("snapshot failed");
        let snapshot3 = fs.device().snapshot().expect("snapshot failed");
        fs.close().await.expect("close failed");

        // 1. Remount with unchanged required_reservation: total and per-store borrowed amounts
        // must match live state across the system and all child stores.
        {
            let fs2 = FxFilesystem::open(snapshot1).await.expect("open failed");
            assert_eq!(fs2.object_manager().metadata_reservation().amount(), expected_reservation);
            assert_eq!(fs2.object_manager().borrowed_metadata_space(), expected_borrowed);
            assert_eq!(fs2.object_manager().reserved_metadata_space(), expected_reserved_map);
            assert_eq!(
                fs2.object_manager().store_borrowed(store_id),
                Some((expected_store_borrowed, expected_store_req))
            );
            assert_eq!(
                fs2.object_manager().store_borrowed(other_store_id),
                Some((expected_other_borrowed, expected_other_req))
            );
            fsck(fs2.clone()).await.expect("fsck failed");
            fs2.close().await.expect("close failed");
        }

        // 2. Remount with an increased required_reservation (simulating a version upgrade that
        // increases expected reservation): actual metadata_reservation must remain unchanged rather
        // than reserving more from the allocator at mount time, and fsck must succeed.
        {
            let extra = 65536;
            let fs2 = FxFilesystemBuilder::new()
                .on_new_store(move |store| {
                    store
                        .filesystem()
                        .ensure_local()
                        .object_manager()
                        .set_reserved_space(super::RESERVED_SPACE + extra);
                })
                .open(snapshot2)
                .await
                .expect("open failed");
            assert_eq!(fs2.object_manager().metadata_reservation().amount(), expected_reservation);
            assert_eq!(fs2.object_manager().borrowed_metadata_space(), expected_borrowed + extra);
            assert_eq!(fs2.object_manager().reserved_metadata_space(), expected_reserved_map);
            fsck(fs2.clone()).await.expect("fsck failed");
            fs2.close().await.expect("close failed");
        }

        // 3. Remount with a decreased required_reservation (smaller than expected_reservation):
        // metadata_reservation is capped at `required_reservation()`.
        {
            let fs2 = FxFilesystemBuilder::new()
                .on_new_store(move |store| {
                    store.filesystem().ensure_local().object_manager().set_reserved_space(0);
                })
                .open(snapshot3)
                .await
                .expect("open failed");
            assert_eq!(
                fs2.object_manager().metadata_reservation().amount(),
                fs2.object_manager().required_reservation()
            );
            assert_eq!(fs2.object_manager().borrowed_metadata_space(), 0);
            fsck(fs2.clone()).await.expect("fsck failed");
            fs2.close().await.expect("close failed");
        }
    }

    #[fuchsia::test]
    async fn test_borrowing_between_begin_and_end_flush_and_replay() {
        let device = DeviceHolder::new(FakeDevice::new(8192, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");
        let store_a = root_volume
            .new_volume("store_a", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();
        let dir_a = Directory::open(&store_a, store_a.root_directory_object_id())
            .await
            .expect("open failed");

        let store_b = root_volume
            .new_volume("store_b", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let dir_b = Directory::open(&store_b, store_b.root_directory_object_id())
            .await
            .expect("open failed");

        let mut file_ids = Vec::new();
        for name in ["file1", "file2"] {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, store_a.root_directory_object_id())],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            let handle = dir_a
                .create_child_file(&mut transaction, name)
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
            let mut buf = handle.allocate_buffer(4096).await;
            buf.fill(0xcd);
            handle.write_or_append(Some(0), buf.as_ref()).await.expect("write failed");
            file_ids.push(handle.object_id());
        }

        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // 1. Borrow in `store_a` before `BeginFlush` by deleting and tombstoning `file1`.
        {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![
                        LockKey::object(store_a_id, store_a.root_directory_object_id()),
                        LockKey::object(store_a_id, file_ids[0]),
                    ],
                    Options { reservation: ReservationOptions::Borrowed, ..Default::default() },
                )
                .await
                .expect("new_transaction failed");
            replace_child(&mut transaction, None, (&dir_a, "file1"))
                .await
                .expect("replace_child failed");
            transaction.commit().await.expect("commit failed");
            fs.tombstone_object(store_a_id, file_ids[0], None)
                .await
                .expect("tombstone_object failed");
        }

        let (pre_borrowed, _) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert!(pre_borrowed > 0);

        // 2. Hold the lock on `store_a`'s StoreInfo object in `root_store` so `store_a.flush()`
        // commits `BeginFlush`, writes the new layer file, and pauses right before `EndFlush`.
        let store_info_lock = fs
            .lock_manager()
            .write_lock(lock_keys![LockKey::object(
                fs.root_store().store_object_id(),
                store_a.store_info_handle_object_id().unwrap(),
            )])
            .await;

        let store_a_clone = store_a.clone();
        let flush_task = fasync::Task::spawn(async move {
            store_a_clone.flush().await.expect("store_a flush failed");
        });

        // Wait until `BeginFlush` has sealed the mutable layer of `store_a`.
        while !store_a.tree().immutable_layer_set().layers.iter().any(|l| l.handle().is_none()) {
            fasync::Timer::new(MonotonicDuration::from_millis(1)).await;
        }

        // 3. While `store_a` is between `BeginFlush` and `EndFlush`, borrow in `store_a` (deleting
        // `file2`) and commit a funded transaction in `store_b`.
        {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![
                        LockKey::object(store_a_id, store_a.root_directory_object_id()),
                        LockKey::object(store_a_id, file_ids[1]),
                    ],
                    Options { reservation: ReservationOptions::Borrowed, ..Default::default() },
                )
                .await
                .expect("new_transaction failed");
            replace_child(&mut transaction, None, (&dir_a, "file2"))
                .await
                .expect("replace_child failed");
            transaction.commit().await.expect("commit failed");
            fs.tombstone_object(store_a_id, file_ids[1], None)
                .await
                .expect("tombstone_object failed");

            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, store_b.root_directory_object_id())],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            dir_b
                .create_child_file(&mut transaction, "b_file")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
        }

        let (total_borrowed_mid, _) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        let post_borrowed = total_borrowed_mid - pre_borrowed;
        assert!(post_borrowed > 0);

        // Allow `EndFlush` to complete.
        std::mem::drop(store_info_lock);
        flush_task.await;

        // Pre-`BeginFlush` borrowed space must be cleared by the full compaction, while
        // post-`BeginFlush` borrowed space must remain borrowed.
        let (b_a_after, req_a_after) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert_eq!(b_a_after, post_borrowed);
        let (b_b_after, req_b_after) = fs.object_manager().store_borrowed(store_b_id).unwrap();
        assert_eq!(b_b_after, 0);

        // 4. Snapshot before a new super-block is written (so `EndFlush` is replayed after the
        // super-block) and verify replay produces the exact same reservations across all buckets.
        fs.sync(SyncOptions { flush_device: true, ..Default::default() })
            .await
            .expect("sync failed");
        let expected_map_before_sb = fs.object_manager().reserved_metadata_space();
        let snapshot_before_sb = fs.device().snapshot().expect("snapshot failed");

        // 5. Write a new super-block (where `journal_file_offsets[store_a_id]` points to the
        // post-`BeginFlush` transaction prior to `EndFlush`), commit one more transaction, and
        // snapshot again to verify replay when `EndFlush` precedes
        // `super_block_journal_file_offset`.
        fs.journal().write_super_block().await.expect("write_super_block failed");
        {
            let mut transaction = fs
                .root_store()
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, store_a.root_directory_object_id())],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            dir_a
                .create_child_file(&mut transaction, "after_sb")
                .await
                .expect("create_child_file failed");
            transaction.commit().await.expect("commit failed");
        }
        fs.sync(SyncOptions { flush_device: true, ..Default::default() })
            .await
            .expect("sync failed");
        let expected_map_after_sb = fs.object_manager().reserved_metadata_space();
        let expected_store_a_after_sb = fs.object_manager().store_borrowed(store_a_id);
        let expected_store_b_after_sb = fs.object_manager().store_borrowed(store_b_id);
        let snapshot_after_sb = fs.device().snapshot().expect("snapshot failed");
        fs.close().await.expect("close failed");

        {
            let fs2 = FxFilesystem::open(snapshot_before_sb).await.expect("open failed");
            assert_eq!(fs2.object_manager().reserved_metadata_space(), expected_map_before_sb);
            assert_eq!(
                fs2.object_manager().store_borrowed(store_a_id),
                Some((b_a_after, req_a_after))
            );
            assert_eq!(
                fs2.object_manager().store_borrowed(store_b_id),
                Some((b_b_after, req_b_after))
            );
            fsck(fs2.clone()).await.expect("fsck failed");
            fs2.close().await.expect("close failed");
        }

        {
            let fs3 = FxFilesystem::open(snapshot_after_sb).await.expect("open failed");
            assert_eq!(fs3.object_manager().reserved_metadata_space(), expected_map_after_sb);
            assert_eq!(fs3.object_manager().store_borrowed(store_a_id), expected_store_a_after_sb);
            assert_eq!(fs3.object_manager().store_borrowed(store_b_id), expected_store_b_after_sb);
            fsck(fs3.clone()).await.expect("fsck failed");
            fs3.close().await.expect("close failed");
        }
    }

    async fn populate_store_with_base_layer(store: &ObjectStore) {
        let store_id = store.store_object_id();
        let dir_id = store.root_directory_object_id();

        // Populate `store` with > 512 KiB (`DEFAULT_RECLAIM_SIZE`) of inline extended attributes
        // so minor compactions keep their base layer.
        for batch in 0..28 {
            let mut transaction = store
                .new_transaction(lock_keys![LockKey::object(store_id, dir_id)], Options::default())
                .await
                .expect("new_transaction failed");
            for i in 0..50 {
                let idx = batch * 50 + i;
                transaction.add(
                    store_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            dir_id,
                            format!("attr_{idx:04}_{:<200}", idx).into_bytes(),
                        ),
                        ObjectValue::inline_extended_attribute(vec![0x55; 200]),
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
        }
        store
            .flush_with_reason(FlushReason::Journal(ForceMajor::True))
            .await
            .expect("flush failed");
    }

    #[fuchsia::test]
    async fn test_per_store_selective_major_compaction_isolation() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let store_a = root_volume
            .new_volume("store_a", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();
        let root_dir_a = store_a.root_directory_object_id();

        let store_b = root_volume
            .new_volume("store_b", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let root_dir_b = store_b.root_directory_object_id();

        populate_store_with_base_layer(&store_a).await;
        populate_store_with_base_layer(&store_b).await;
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 1);

        // Add a second layer to `store_b` via a funded transaction and minor flush.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"extra".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0x77; 200]),
                ),
            );
            transaction.commit().await.expect("commit failed");
            store_b
                .flush_with_reason(FlushReason::Journal(ForceMajor::False))
                .await
                .expect("minor flush failed");
        }
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 2);
        let (b_b_before, req_b_before) = fs.object_manager().store_borrowed(store_b_id).unwrap();
        assert_eq!(b_b_before, 0);

        // Delete extended attributes in `store_a` using `Borrowed` until `B_A >= R_A / 2`.
        for batch in 0..28 {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 0..50 {
                let idx = batch * 50 + i;
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("attr_{idx:04}_{:<200}", idx).into_bytes(),
                        ),
                        ObjectValue::None,
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
            let (b_a, req_a) = fs.object_manager().store_borrowed(store_a_id).unwrap();
            if b_a >= req_a - req_a / 2 {
                break;
            }
        }
        let (b_a_depleted, req_a_depleted) =
            fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert!(b_a_depleted >= req_a_depleted - req_a_depleted / 2);

        // `store_b`'s borrowed space must remain completely isolated from `store_a`'s borrowing.
        assert_eq!(fs.object_manager().store_borrowed(store_b_id), Some((0, req_b_before)));

        // Flushing with `ForceMajor::False` must force a major compaction on `store_a` (resetting
        // `B_A == 0` and merging its layers into 1) while leaving `store_b` untouched (still 2
        // layers).
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);
        let (b_a_restored, _) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert_eq!(b_a_restored, 0);
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 2);

        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_shortfall_major_compaction_does_not_cycle() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let store_b = root_volume
            .new_volume("store_b", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let root_dir_b = store_b.root_directory_object_id();

        populate_store_with_base_layer(&store_b).await;
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // Add a second layer to `store_b` via a funded transaction and minor flush.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"extra".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0x77; 200]),
                ),
            );
            transaction.commit().await.expect("commit failed");
            store_b
                .flush_with_reason(FlushReason::Journal(ForceMajor::False))
                .await
                .expect("minor flush failed");
        }
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 2);
        let (_, req_b_before) = fs.object_manager().store_borrowed(store_b_id).unwrap();

        // Simulate a shortfall in `M_total` alongside `B_B >= R_B / 2` on `store_b` (which has 2
        // layers of live data and no tombstones, so compacting yields no space).
        let current_total = fs.object_manager().metadata_reservation().amount();
        let reduction = current_total / 2;
        fs.object_manager().metadata_reservation().give_back(reduction);
        fs.object_manager().set_store_borrowed(store_b_id, req_b_before);

        // 1) Flushing with `ForceMajor::False` must major-compact `store_b` down to 1 layer and
        // reset `B_B == 0` even though `M_total` is short and compacting yielded no space.
        fs.journal().force_compact().await.expect("force_compact failed");
        let info_b_after_major = store_b.load_store_info().await.unwrap();
        assert_eq!(info_b_after_major.layers.len(), 1);
        let (b_b_restored, _) = fs.object_manager().store_borrowed(store_b_id).unwrap();
        assert_eq!(b_b_restored, 0);

        // 2) Commit a single transaction on `store_b` while `M_total` is still short and flush
        // with `ForceMajor::False`. Because `B_B` stayed `0` across
        // `reconcile_metadata_reservation`, this flush must perform a minor compaction (creating a
        // 2nd layer) rather than cycling on major compactions.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"after_shortfall".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0x88; 200]),
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(
            store_b.load_store_info().await.unwrap().layers.len(),
            2,
            "Single transaction after shortfall major compaction must not trigger major compaction"
        );

        fs.allocator()
            .reserve(None, reduction)
            .unwrap()
            .move_to(fs.object_manager().metadata_reservation(), reduction);
        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_no_op_major_compaction_guard() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let store_a = root_volume
            .new_volume("store_a", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();

        populate_store_with_base_layer(&store_a).await;
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // If `B_A >= R_A / 2` when `store_a` already has only 1 layer and no unflushed journal
        // mutations, `flush(FlushReason::Journal(ForceMajor::False))` must not re-compact
        // `store_a`.
        let layer_oid_a_before = store_a.load_store_info().await.unwrap().layers[0];
        let (_, req_a) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        fs.object_manager().set_store_borrowed(store_a_id, req_a);
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers, vec![layer_oid_a_before]);

        fs.object_manager().set_store_borrowed(store_a_id, 0);
        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_minor_compaction_replay_and_delete_volume_prunes_borrowed() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let store_b = root_volume
            .new_volume("store_b", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let root_dir_b = store_b.root_directory_object_id();

        populate_store_with_base_layer(&store_b).await;
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // Commit a small borrowing transaction on `store_b` (which has 1 base layer > 512 KiB),
        // perform a minor flush (`layers.len() > 1`, which creates a new minor layer and preserves
        // `B_B > 0`), and create + delete a volume `store_c`. Verify that `DeleteVolume` does not
        // leave `store_c` in `store_borrowed` and that replaying the journal across the minor
        // `EndFlush` preserves `store_b`'s borrowed space.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"attr_0000_0".to_vec()),
                    ObjectValue::None,
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        let (b_b_minor, _) = fs.object_manager().store_borrowed(store_b_id).unwrap();
        assert!(b_b_minor > 0);
        store_b
            .flush_with_reason(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("minor flush failed");
        assert!(store_b.load_store_info().await.unwrap().layers.len() > 1);
        assert_eq!(fs.object_manager().store_borrowed(store_b_id).unwrap().0, b_b_minor);

        let store_c = root_volume
            .new_volume("store_c", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_c_id = store_c.store_object_id();
        std::mem::drop(store_c);
        let (_, delete_txn) = root_volume
            .acquire_transaction_for_remove_volume("store_c", [], false, Options::default())
            .await
            .expect("acquire_transaction_for_remove_volume failed");
        root_volume
            .delete_volume("store_c", delete_txn, || {})
            .await
            .expect("delete_volume failed");
        assert!(fs.object_manager().store_borrowed(store_c_id).is_none());

        fs.sync(SyncOptions { flush_device: true, ..Default::default() })
            .await
            .expect("sync failed");
        let expected_map = fs.object_manager().reserved_metadata_space();
        assert!(!expected_map.contains_key(&store_c_id));
        let snapshot = fs.device().snapshot().expect("snapshot failed");
        fs.close().await.expect("close failed");

        let fs2 = FxFilesystem::open(snapshot).await.expect("open failed");
        assert_eq!(fs2.object_manager().reserved_metadata_space(), expected_map);
        fsck(fs2.clone()).await.expect("fsck failed");
        fs2.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_legacy_borrowed_proportional_distribution() {
        let device = DeviceHolder::new(FakeDevice::new(8192, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let crypt_a = Arc::new(new_insecure_crypt());
        let store_a = root_volume
            .new_volume(
                "store_a",
                NewChildStoreOptions {
                    options: StoreOptions {
                        crypt: Some(crypt_a.clone()),
                        ..StoreOptions::default()
                    },
                    ..NewChildStoreOptions::default()
                },
            )
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();
        let root_dir_a = store_a.root_directory_object_id();

        let store_b = root_volume
            .new_volume("store_b", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let root_dir_b = store_b.root_directory_object_id();

        // Give `store_a` more metadata than `store_b`.
        for i in 0..16 {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_a_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_a, format!("attr_{i}").into_bytes()),
                    ObjectValue::inline_extended_attribute(vec![0xaa; 1024]),
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"attr_0".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0xbb; 1024]),
                ),
            );
            transaction.commit().await.expect("commit failed");
        }

        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // Write an additional mutation to `store_a`, lock `store_a`, and flush it while locked so
        // `store_a` has both layer files and a non-empty `encrypted_mutations` file.
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_a_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_a, b"locked_attr".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0xcc; 1024]),
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        store_a.lock().await.expect("lock failed");
        store_a.flush().await.expect("locked flush failed");
        fs.journal().write_super_block().await.expect("write_super_block failed");
        fs.journal().write_super_block().await.expect("write_super_block failed");

        // After the super-block, commit:
        // 1) A funded (`Commit`) transaction on `store_b`.
        // 2) A V60 borrowing (`CommitBorrowed`) transaction on `store_b`.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"funded_attr".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0xdd; 64]),
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        let offset_before_legacy_borrowed = fs.object_manager().last_end_offset();
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"legacy_borrowed_attr".to_vec()),
                    ObjectValue::None,
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        let offset_after_legacy_borrowed = fs.object_manager().last_end_offset();

        let (_, req_a) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        let (_, req_b) = fs.object_manager().store_borrowed(store_b_id).unwrap();
        assert!(req_a > req_b);

        // Compute `legacy_required` (`max_i(1.5 * L_i) + 4 * J_total + 1_048_576`), where locked
        // `store_a` includes `encrypted_mutations_size * 3`.
        let (_, min_checkpoint) = fs.object_manager().journal_file_offsets();
        let min_offset = min_checkpoint.unwrap().file_offset;
        let journal_usage = offset_after_legacy_borrowed - min_offset;
        let info_a = store_a.load_store_info().await.unwrap();
        let mut store_a_layer_size = 0;
        for &oid in &info_a.layers {
            store_a_layer_size += fs.root_store().get_file_size(oid).await.unwrap();
        }
        assert_ne!(info_a.encrypted_mutations_object_id, INVALID_OBJECT_ID);
        store_a_layer_size +=
            fs.root_store().get_file_size(info_a.encrypted_mutations_object_id).await.unwrap() * 3;
        let max_layer_size = [
            fs.allocator()
                .tree()
                .immutable_layer_set()
                .layers
                .iter()
                .map(|l| l.handle().map(ReadObjectHandle::get_size).unwrap_or(0))
                .sum::<u64>(),
            fs.root_store()
                .tree()
                .immutable_layer_set()
                .layers
                .iter()
                .map(|l| l.handle().map(ReadObjectHandle::get_size).unwrap_or(0))
                .sum::<u64>(),
            store_a_layer_size,
            store_b
                .tree()
                .immutable_layer_set()
                .layers
                .iter()
                .map(|l| l.handle().map(ReadObjectHandle::get_size).unwrap_or(0))
                .sum::<u64>(),
        ]
        .into_iter()
        .max()
        .unwrap();
        assert_eq!(max_layer_size, store_a_layer_size);
        let legacy_required = max_layer_size * 3 / 2 + journal_usage * 4 + 1_048_576;
        let new_required = fs.object_manager().required_reservation();
        assert!(new_required > legacy_required);

        let initial_legacy_borrowed = 1_048_576 / 4;
        let expected_final_legacy_borrowed = initial_legacy_borrowed
            + (offset_after_legacy_borrowed - offset_before_legacy_borrowed) * 4;

        fs.sync(SyncOptions { flush_device: true, ..Default::default() })
            .await
            .expect("sync failed");
        let snapshot_legacy = fs.device().snapshot().expect("snapshot failed");
        fs.close().await.expect("close failed");

        // Seed `ReplayMetadataReservation::Borrowed` when `store_a` is opened in
        // `on_replay_complete` (right before `init_metadata_reservation`) so
        // `init_metadata_reservation` runs in `< V61` migration mode with `store_b`'s unflushed
        // borrowing in `journal_checkpoints`.
        let fs2 = FxFilesystemBuilder::new()
            .on_new_store(move |store| {
                if store.store_object_id() == store_a_id {
                    store
                        .filesystem()
                        .ensure_local()
                        .object_manager()
                        .init_replay_metadata_reservation(
                            Version { major: 59, minor: 0 },
                            &HashMap::from_iter([(0, expected_final_legacy_borrowed)]),
                        );
                }
            })
            .open(snapshot_legacy)
            .await
            .expect("open failed");

        let expected_reservation = legacy_required - expected_final_legacy_borrowed;
        assert_eq!(fs2.object_manager().metadata_reservation().amount(), expected_reservation);
        assert_eq!(
            fs2.object_manager().borrowed_metadata_space(),
            new_required - expected_reservation
        );
        let (b_a, r_a) = fs2.object_manager().store_borrowed(store_a_id).unwrap();
        let (b_b, r_b) = fs2.object_manager().store_borrowed(store_b_id).unwrap();
        assert!(b_a > 0 && b_a < r_a);
        assert!(b_b > 0 && b_b < r_b);
        fsck(fs2.clone()).await.expect("fsck failed");
        fs2.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_edge_cases_shrink_minor_layer_both_checkpoints_and_system_compaction() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let store_a = root_volume
            .new_volume("store_a", NewChildStoreOptions::default())
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();
        let root_dir_a = store_a.root_directory_object_id();

        // Populate `store_a` with > 512 KiB so minor compactions keep its base layer.
        for batch in 0..28 {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            for i in 0..50 {
                let idx = batch * 50 + i;
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("base_{idx:04}_{:<200}", idx).into_bytes(),
                        ),
                        ObjectValue::inline_extended_attribute(vec![0x55; 200]),
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
        }
        store_a
            .flush_with_reason(FlushReason::Journal(ForceMajor::True))
            .await
            .expect("major flush failed");
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);

        // 1) Exercise `Checkpoints::Both(x, y)` merging into `Checkpoints::Old(x + y)` on repeated
        // `BeginFlush`: commit borrowing transaction `x`, then `Mutation::BeginFlush`, then
        // borrowing transaction `y`, and then minor-flush `store_a`.
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 0..20 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("minor_attr_{i:02}").into_bytes(),
                        ),
                        ObjectValue::inline_extended_attribute(vec![0xaa; 1024]),
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");

            let mut begin_txn = fs
                .root_store()
                .new_transaction(
                    lock_keys![],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            begin_txn.add(store_a_id, Mutation::BeginFlush);
            begin_txn.commit().await.expect("commit failed");

            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 20..40 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("minor_attr_{i:02}").into_bytes(),
                        ),
                        ObjectValue::inline_extended_attribute(vec![0xaa; 1024]),
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
        }
        store_a
            .flush_with_reason(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("minor flush failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 2);
        let (layer_borrowed_after_grow, _) =
            fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert!(layer_borrowed_after_grow > 0);

        // 2) Exercise pre-super-block partitioning of `B_A` into `store_borrowed` (flushed layer)
        // and `journal_checkpoints` (unflushed journal): commit an unflushed borrowing transaction,
        // write a super-block without flushing `store_a`, remount, and verify `store_borrowed`
        // matches before and after a minor flush.
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_a_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_a, b"unflushed_before_sb".to_vec()),
                    ObjectValue::None,
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        let (total_b_with_unflushed, _) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert!(total_b_with_unflushed > layer_borrowed_after_grow);
        fs.journal().write_super_block().await.expect("write_super_block failed");
        {
            let snapshot = fs.device().snapshot().expect("snapshot failed");
            let fs_replay = FxFilesystem::open(snapshot).await.expect("open failed");
            assert_eq!(
                fs_replay.object_manager().store_borrowed(store_a_id).unwrap().0,
                total_b_with_unflushed
            );
            fs_replay.close().await.expect("close failed");
        }

        // 3) Exercise minor compaction shrinking a store's layer size (`new_size < old_size` in
        // `store_borrowed_for_flush`): replace the 40 large (1024-byte) attributes in Layer 1 with
        // `ObjectValue::None` in a funded transaction and minor-flush `store_a`.
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            for i in 0..40 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("minor_attr_{i:02}").into_bytes(),
                        ),
                        ObjectValue::None,
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
        }
        let res_before_shrink = fs.object_manager().reservation(store_a_id).unwrap();
        store_a
            .flush_with_reason(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("minor flush failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 2);
        let res_after_shrink = fs.object_manager().reservation(store_a_id).unwrap();
        assert!(res_after_shrink < res_before_shrink);
        let (layer_borrowed_after_shrink, _) =
            fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert_eq!(
            layer_borrowed_after_shrink,
            layer_borrowed_after_grow.saturating_sub(res_before_shrink - res_after_shrink)
        );

        // 4) Exercise system bucket (`root_store`) forced major compaction when
        // `b_sys >= sys_headroom / 2`: reset `store_a` borrowed to 0 while keeping 2 layers on
        // `store_a`, commit a mutation on `root_store` so `root_store` can compact, simulate
        // system borrowing by giving back reservation space, and verify a journal flush forces
        // `root_store` to compact without compacting `store_a`.
        fs.object_manager().set_store_borrowed(store_a_id, 0);
        {
            let root_store = fs.root_store();
            let root_id = root_store.store_object_id();
            let graveyard_id = root_store.graveyard_directory_object_id();
            let mut transaction = root_store
                .new_transaction(
                    lock_keys![LockKey::object(root_id, graveyard_id)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                root_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(graveyard_id, b"sys_test".to_vec()),
                    ObjectValue::None,
                ),
            );
            transaction.commit().await.expect("commit failed");
        }
        let current_res = fs.object_manager().metadata_reservation().amount();
        fs.object_manager().metadata_reservation().give_back(current_res / 2);
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(
            store_a.load_store_info().await.unwrap().layers.len(),
            2,
            "System borrowing must not force major compaction on child store_a when B_A == 0"
        );
        let restored = fs.allocator().reserve(None, current_res / 2).unwrap();
        restored.move_to(fs.object_manager().metadata_reservation(), current_res / 2);
        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }

    #[fuchsia::test]
    async fn test_locked_store_isolation_and_minor_flush_reservation() {
        let device = DeviceHolder::new(FakeDevice::new(16384, 4096));
        let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
        let root_volume = root_volume(fs.clone()).await.expect("root_volume failed");

        let crypt_a = Arc::new(new_insecure_crypt());
        let store_a = root_volume
            .new_volume(
                "store_a",
                NewChildStoreOptions {
                    options: StoreOptions {
                        crypt: Some(crypt_a.clone()),
                        ..StoreOptions::default()
                    },
                    ..NewChildStoreOptions::default()
                },
            )
            .await
            .expect("new_volume failed");
        let store_a_id = store_a.store_object_id();
        let root_dir_a = store_a.root_directory_object_id();

        let crypt_b = Arc::new(new_insecure_crypt());
        let store_b = root_volume
            .new_volume(
                "store_b",
                NewChildStoreOptions {
                    options: StoreOptions {
                        crypt: Some(crypt_b.clone()),
                        ..StoreOptions::default()
                    },
                    ..NewChildStoreOptions::default()
                },
            )
            .await
            .expect("new_volume failed");
        let store_b_id = store_b.store_object_id();
        let root_dir_b = store_b.root_directory_object_id();

        // Populate both encrypted stores so each has a base layer > 512 KiB.
        for (store, store_id, dir_id) in
            [(&store_a, store_a_id, root_dir_a), (&store_b, store_b_id, root_dir_b)]
        {
            for batch in 0..28 {
                let mut transaction = store
                    .new_transaction(
                        lock_keys![LockKey::object(store_id, dir_id)],
                        Options::default(),
                    )
                    .await
                    .expect("new_transaction failed");
                for i in 0..50 {
                    let idx = batch * 50 + i;
                    transaction.add(
                        store_id,
                        Mutation::replace_or_insert_object(
                            ObjectKey::extended_attribute(
                                dir_id,
                                format!("attr_{idx:04}_{:<200}", idx).into_bytes(),
                            ),
                            ObjectValue::inline_extended_attribute(vec![0x55; 200]),
                        ),
                    );
                }
                transaction.commit().await.expect("commit failed");
            }
        }

        store_a
            .flush_with_reason(FlushReason::Journal(ForceMajor::True))
            .await
            .expect("flush failed");
        store_b
            .flush_with_reason(FlushReason::Journal(ForceMajor::True))
            .await
            .expect("flush failed");
        fs.journal().force_compact().await.expect("force_compact failed");
        fs.journal().stop_compactions().await;

        // Add a 2nd layer to `store_b` via a funded transaction and minor flush.
        {
            let mut transaction = store_b
                .new_transaction(
                    lock_keys![LockKey::object(store_b_id, root_dir_b)],
                    Options::default(),
                )
                .await
                .expect("new_transaction failed");
            transaction.add(
                store_b_id,
                Mutation::replace_or_insert_object(
                    ObjectKey::extended_attribute(root_dir_b, b"extra_1".to_vec()),
                    ObjectValue::inline_extended_attribute(vec![0x77; 200]),
                ),
            );
            transaction.commit().await.expect("commit failed");
            store_b
                .flush_with_reason(FlushReason::Journal(ForceMajor::False))
                .await
                .expect("minor flush failed");
        }
        fs.journal().force_compact().await.expect("force_compact failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 2);
        assert_eq!(fs.object_manager().borrowed_metadata_space(), 0);

        // 1) Minor flush reservation retention test:
        // Commit borrowing deletions on `store_a`, perform a minor flush (`ForceMajor::False`),
        // and compact the journal (`reconcile_metadata_reservation`). When `store_a` subsequently
        // performs a major compaction (`ForceMajor::True`) and compacts the journal, `M_total` must
        // be fully restored (`borrowed_metadata_space() == 0`).
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 0..50 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("temp_{i:04}_{:<200}", i).into_bytes(),
                        ),
                        ObjectValue::None,
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
            store_a
                .flush_with_reason(FlushReason::Journal(ForceMajor::False))
                .await
                .expect("minor flush failed");
            assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 2);
        }
        fs.journal().force_compact().await.expect("force_compact failed");
        store_a
            .flush_with_reason(FlushReason::Journal(ForceMajor::True))
            .await
            .expect("major flush failed");
        fs.journal().force_compact().await.expect("force_compact failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);
        assert_eq!(fs.object_manager().store_borrowed(store_a_id).unwrap().0, 0);
        assert_eq!(
            fs.object_manager().borrowed_metadata_space(),
            0,
            "reconcile_metadata_reservation must not give away M_total during a minor flush"
        );

        // 2) Locked store isolation test:
        // Commit borrowing deletions on `store_a`, flush the first batch as a minor compaction
        // while unlocked, and lock `store_a` while the second batch is still in the journal so that
        // `force_compact` flushes it via `flush_locked()`.
        {
            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 0..50 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("temp_{i:04}_{:<200}", i).into_bytes(),
                        ),
                        ObjectValue::None,
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
            store_a
                .flush_with_reason(FlushReason::Journal(ForceMajor::False))
                .await
                .expect("minor flush failed");
            assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 2);

            let mut transaction = store_a
                .new_transaction(
                    lock_keys![LockKey::object(store_a_id, root_dir_a)],
                    Options {
                        skip_journal_checks: true,
                        reservation: ReservationOptions::Borrowed,
                        ..Default::default()
                    },
                )
                .await
                .expect("new_transaction failed");
            for i in 50..100 {
                transaction.add(
                    store_a_id,
                    Mutation::replace_or_insert_object(
                        ObjectKey::extended_attribute(
                            root_dir_a,
                            format!("temp_{i:04}_{:<200}", i).into_bytes(),
                        ),
                        ObjectValue::None,
                    ),
                );
            }
            transaction.commit().await.expect("commit failed");
        }

        store_a.lock().await.expect("lock failed");
        fs.journal().force_compact().await.expect("force_compact failed");

        let (b_a_locked, req_a_locked) = fs.object_manager().store_borrowed(store_a_id).unwrap();
        assert!(b_a_locked > 0);
        // Simulate `store_a` having borrowed up to its full `req_A` budget before being locked.
        fs.object_manager().set_store_borrowed(store_a_id, req_a_locked);

        // While `store_a` is locked with maximal borrowed space, `store_b` (which has 2 layers)
        // must remain completely isolated and not be forced into a major compaction.
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(
            store_b.load_store_info().await.unwrap().layers.len(),
            2,
            "Locked store_a must not force major compactions on unlocked store_b"
        );
        assert_eq!(fs.object_manager().store_borrowed(store_b_id).unwrap().0, 0);

        // Unlock `store_a`: because `B_A >= req_A / 2`, a journal flush must now force a major
        // compaction on `store_a`, purging its transient attributes and resetting `B_A` to 0
        // without touching `store_b`'s 2 layers.
        store_a.unlock(crypt_a).await.expect("unlock failed");
        fs.object_manager()
            .flush(FlushReason::Journal(ForceMajor::False))
            .await
            .expect("flush failed");
        assert_eq!(store_a.load_store_info().await.unwrap().layers.len(), 1);
        assert_eq!(fs.object_manager().store_borrowed(store_a_id).unwrap().0, 0);
        assert_eq!(store_b.load_store_info().await.unwrap().layers.len(), 2);

        // Compact the journal and verify that all metadata reservation space borrowed by `store_a`
        // during its minor/locked flushes was retained and restored
        // (`borrowed_metadata_space == 0`).
        fs.journal().force_compact().await.expect("force_compact failed");
        assert_eq!(fs.object_manager().borrowed_metadata_space(), 0);

        fsck(fs.clone()).await.expect("fsck failed");
        fs.close().await.expect("close failed");
    }
}

#[cfg(fuzz)]
mod fuzz {
    use fuzz::fuzz;

    #[fuzz]
    fn fuzz_journal_bytes(input: Vec<u8>) {
        use crate::filesystem::FxFilesystem;
        use fuchsia_async as fasync;
        use std::io::Write;
        use storage_device::DeviceHolder;
        use storage_device::fake_device::FakeDevice;

        fasync::SendExecutorBuilder::new().num_threads(4).build().run(async move {
            let device = DeviceHolder::new(FakeDevice::new(32768, 512));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            fs.journal().inner.lock().writer.write_all(&input).expect("write failed");
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);
            if let Ok(fs) = FxFilesystem::open(device).await {
                // `close()` can fail if there were objects to be tombstoned. If the said object is
                // corrupted, there will be an error when we compact the journal.
                let _ = fs.close().await;
            }
        });
    }

    #[fuzz]
    fn fuzz_journal(input: Vec<super::JournalRecord>) {
        use crate::filesystem::FxFilesystem;
        use fuchsia_async as fasync;
        use storage_device::DeviceHolder;
        use storage_device::fake_device::FakeDevice;

        fasync::SendExecutorBuilder::new().num_threads(4).build().run(async move {
            let device = DeviceHolder::new(FakeDevice::new(32768, 512));
            let fs = FxFilesystem::new_empty(device).await.expect("new_empty failed");
            {
                let mut inner = fs.journal().inner.lock();
                for record in &input {
                    let _ = inner.writer.write_record(record);
                }
            }
            fs.close().await.expect("close failed");
            let device = fs.take_device().await;
            device.reopen(false);
            if let Ok(fs) = FxFilesystem::open(device).await {
                // `close()` can fail if there were objects to be tombstoned. If the said object is
                // corrupted, there will be an error when we compact the journal.
                let _ = fs.close().await;
            }
        });
    }
}