// 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 checksum_list;
mod reader;
pub mod super_block;
mod writer;

use {
    crate::{
        checksum::Checksum,
        debug_assert_not_too_long,
        errors::FxfsError,
        filesystem::{ApplyContext, ApplyMode, Filesystem, SyncOptions},
        log::*,
        object_handle::{BootstrapObjectHandle, ObjectHandle},
        object_store::{
            allocator::{Allocator, SimpleAllocator},
            extent_record::{Checksums, ExtentKey, ExtentValue, DEFAULT_DATA_ATTRIBUTE_ID},
            graveyard::Graveyard,
            journal::{
                checksum_list::ChecksumList,
                reader::{JournalReader, ReadResult},
                super_block::{SuperBlockInstance, SuperBlockManager},
                writer::JournalWriter,
            },
            object_manager::ObjectManager,
            object_record::{AttributeKey, ObjectKey, ObjectKeyData, ObjectValue},
            transaction::{
                AllocatorMutation, Mutation, ObjectStoreMutation, Options, Transaction,
                TxnMutation, TRANSACTION_MAX_JOURNAL_USAGE,
            },
            HandleOptions, Item, ItemRef, LastObjectId, LockState, NewChildStoreOptions,
            ObjectStore, StoreObjectHandle, INVALID_OBJECT_ID,
        },
        range::RangeExt,
        round::{round_down, round_up},
        serialized_types::{Version, Versioned, LATEST_VERSION},
        trace_duration,
    },
    anyhow::{anyhow, bail, Context, Error},
    async_utils::event::Event,
    futures::{self, future::poll_fn, FutureExt},
    once_cell::sync::OnceCell,
    rand::Rng,
    serde::{Deserialize, Serialize},
    static_assertions::const_assert,
    std::{
        clone::Clone,
        collections::HashMap,
        convert::TryFrom as _,
        ops::Bound,
        sync::{
            atomic::{AtomicBool, Ordering},
            Arc, Mutex,
        },
        task::{Poll, Waker},
        vec::Vec,
    },
    storage_device::buffer::Buffer,
};

// Exposed for serialized_types.
pub use super_block::{SuperBlock, SuperBlockRecord};

// The journal file is written to in blocks of this size.
const BLOCK_SIZE: u64 = 8192;

// 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 = 262_144;

// 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.
#[derive(Clone, Debug, Default, Deserialize, Eq, PartialEq, Serialize)]
pub struct JournalCheckpoint {
    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,
}

#[derive(Clone, Debug, Serialize, Deserialize, Versioned)]
pub enum JournalRecord {
    // Indicates no more records in this block.
    EndBlock,
    // Mutation for a particular object.  object_id here is for the collection i.e. the store or
    // allocator.
    Mutation { object_id: u64, mutation: Mutation },
    // 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),
}

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>,
    handle: OnceCell<StoreObjectHandle<ObjectStore>>,
    super_block_manager: SuperBlockManager,
    inner: Mutex<Inner>,
    commit_mutex: futures::lock::Mutex<()>,
    writer_mutex: futures::lock::Mutex<()>,
    sync_mutex: futures::lock::Mutex<()>,
    trace: AtomicBool,
}

struct Inner {
    super_block: SuperBlock,

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

    // 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,

    // Disable compactions.
    disable_compactions: bool,

    // True if compactions are running.
    compaction_running: bool,

    // 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,

    // 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,
}

impl Inner {
    fn terminate(&mut self) {
        self.terminate = true;
        if let Some(waker) = self.flush_waker.take() {
            waker.wake();
        }
        if let Some(waker) = self.sync_waker.take() {
            waker.wake();
        }
        self.reclaim_event = None;
    }
}

#[derive(Clone, Debug)]
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,
}

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

impl Journal {
    pub fn new(objects: Arc<ObjectManager>, options: JournalOptions) -> Journal {
        let starting_checksum = rand::thread_rng().gen();
        Journal {
            objects: objects,
            handle: OnceCell::new(),
            super_block_manager: SuperBlockManager::new(),
            inner: Mutex::new(Inner {
                super_block: SuperBlock::default(),
                reclaim_event: None,
                zero_offset: None,
                device_flushed_offset: 0,
                needs_did_flush_device: false,
                writer: JournalWriter::new(BLOCK_SIZE as usize, starting_checksum),
                output_reset_version: false,
                flush_waker: None,
                terminate: false,
                disable_compactions: false,
                compaction_running: false,
                sync_waker: None,
                flushed_offset: 0,
                discard_offset: None,
                reclaim_size: options.reclaim_size,
            }),
            commit_mutex: futures::lock::Mutex::new(()),
            writer_mutex: futures::lock::Mutex::new(()),
            sync_mutex: futures::lock::Mutex::new(()),
            trace: AtomicBool::new(false),
        }
    }

    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 journal_file_offset(&self) -> u64 {
        self.inner.lock().unwrap().super_block.super_block_journal_file_offset
    }

    /// 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) -> Result<bool, Error> {
        match mutation {
            Mutation::ObjectStore(ObjectStoreMutation {
                item:
                    Item {
                        key:
                            ObjectKey {
                                data:
                                    ObjectKeyData::Attribute(
                                        _,
                                        AttributeKey::Extent(ExtentKey { range }),
                                    ),
                                ..
                            },
                        value:
                            ObjectValue::Extent(ExtentValue::Some {
                                checksums: Checksums::Fletcher(checksums),
                                ..
                            }),
                        ..
                    },
                ..
            }) => {
                if checksums.len() == 0 {
                    return Ok(false);
                }
                let len = if let Ok(l) = usize::try_from(range.length()?) {
                    l
                } else {
                    return Ok(false);
                };
                if len % checksums.len() != 0 {
                    return Ok(false);
                }
                if (len / checksums.len()) % 4 != 0 {
                    return Ok(false);
                }
            }
            Mutation::ObjectStore(_) => {}
            Mutation::EncryptedObjectStore(_) => {}
            Mutation::Allocator(AllocatorMutation::Allocate { device_range, owner_object_id }) => {
                return Ok(device_range.valid() && *owner_object_id != INVALID_OBJECT_ID);
            }
            Mutation::Allocator(AllocatorMutation::Deallocate {
                device_range,
                owner_object_id,
            }) => {
                return Ok(device_range.valid() && *owner_object_id != INVALID_OBJECT_ID);
            }
            Mutation::Allocator(AllocatorMutation::MarkForDeletion(owner_object_id)) => {
                return Ok(*owner_object_id != INVALID_OBJECT_ID);
            }
            Mutation::BeginFlush => {}
            Mutation::EndFlush => {}
            Mutation::DeleteVolume => {}
            Mutation::UpdateBorrowed(_) => {}
            Mutation::UpdateMutationsKey(_) => {}
        }
        Ok(true)
    }

    fn update_checksum_list(
        &self,
        journal_offset: u64,
        owner_object_id: u64,
        mutation: &Mutation,
        checksum_list: &mut ChecksumList,
    ) -> Result<(), Error> {
        match mutation {
            Mutation::ObjectStore(ObjectStoreMutation {
                item:
                    Item {
                        key:
                            ObjectKey {
                                data:
                                    ObjectKeyData::Attribute(
                                        _,
                                        AttributeKey::Extent(ExtentKey { range }),
                                    ),
                                ..
                            },
                        value:
                            ObjectValue::Extent(ExtentValue::Some {
                                device_offset,
                                checksums: Checksums::Fletcher(checksums),
                                ..
                            }),
                        ..
                    },
                ..
            }) => {
                checksum_list.push(
                    journal_offset,
                    owner_object_id,
                    *device_offset..*device_offset + range.length()?,
                    checksums,
                );
            }
            Mutation::ObjectStore(_) => {}
            Mutation::Allocator(AllocatorMutation::Deallocate { device_range, .. }) => {
                checksum_list.mark_deallocated(journal_offset, device_range.clone());
            }
            _ => {}
        }
        Ok(())
    }

    /// Reads the latest super-block, and then replays journaled records.
    pub async fn replay(
        &self,
        filesystem: Arc<dyn Filesystem>,
        on_new_allocator: Option<Box<dyn Fn(Arc<dyn Allocator>) + Send + Sync>>,
    ) -> Result<(), Error> {
        trace_duration!("Journal::replay");
        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(SimpleAllocator::new(filesystem.clone(), super_block.allocator_object_id));
        if let Some(on_new_allocator) = on_new_allocator {
            on_new_allocator(allocator.clone() as Arc<dyn Allocator>);
        }
        self.objects.set_allocator(allocator.clone());
        self.objects.set_borrowed_metadata_space(super_block.borrowed_metadata_space);
        self.objects.set_last_end_offset(super_block.super_block_journal_file_offset);
        {
            let mut inner = self.inner.lock().unwrap();
            inner.super_block = super_block.clone();
        }
        let root_store = ObjectStore::new(
            Some(root_parent.clone()),
            super_block.root_store_object_id,
            filesystem.clone(),
            None,
            None,
            LockState::Unencrypted,
            LastObjectId::default(),
        );
        self.objects.set_root_store(root_store);

        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,
                    DEFAULT_DATA_ATTRIBUTE_ID,
                    AttributeKey::Extent(ExtentKey::search_key_from_offset(round_down(
                        super_block.journal_checkpoint.file_offset,
                        BLOCK_SIZE,
                    ))),
                )))
                .await?;
            let start_offset = if let Some(ItemRef {
                key:
                    ObjectKey {
                        data:
                            ObjectKeyData::Attribute(
                                DEFAULT_DATA_ATTRIBUTE_ID,
                                AttributeKey::Extent(ExtentKey { range }),
                            ),
                        ..
                    },
                ..
            }) = iter.get()
            {
                range.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,
                        DEFAULT_DATA_ATTRIBUTE_ID,
                        ExtentKey { range },
                        ExtentValue::Some { device_offset, .. },
                    )) if object_id == super_block.journal_object_id => {
                        if range.start != start_offset + handle.get_size() {
                            bail!(anyhow!(FxfsError::Inconsistent).context(format!(
                                "Unexpected journal extent {:?}, expected start: {}",
                                item,
                                handle.get_size()
                            )));
                        }
                        handle.push_extent(*device_offset..*device_offset + range.length()?);
                        true
                    }
                    _ => false,
                } {
                    break;
                }
                iter.advance().await?;
            }
        }
        let mut reader =
            JournalReader::new(handle, self.block_size(), &super_block.journal_checkpoint);
        let mut transactions = Vec::new();
        let mut current_transaction = None;
        let mut device_flushed_offset = super_block.super_block_journal_file_offset;
        // Maps from owner_object_id to offset at which it was deleted. Required for checksum
        // validation.
        let mut marked_for_deletion: HashMap<u64, u64> = HashMap::new();
        loop {
            // Cache the checkpoint before we deserialize a record.
            let checkpoint = reader.journal_file_checkpoint();
            let result = reader.deserialize().await?;
            match result {
                ReadResult::Reset => {
                    let version;
                    reader.consume({
                        let mut cursor = std::io::Cursor::new(reader.buffer());
                        version = Version::deserialize_from(&mut cursor)?;
                        cursor.position() as usize
                    });
                    reader.set_version(version);
                    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 current_transaction = match current_transaction.as_mut() {
                                None => {
                                    transactions.push((checkpoint, Vec::new(), 0));
                                    current_transaction = transactions.last_mut();
                                    current_transaction.as_mut().unwrap()
                                }
                                Some(transaction) => transaction,
                            };
                            // If this mutation doesn't need to be applied, don't bother adding it
                            // to the transaction.
                            if self.should_apply(object_id, &current_transaction.0) {
                                current_transaction.1.push((object_id, mutation));
                            }
                        }
                        JournalRecord::Commit => {
                            if let Some((_checkpoint, mutations, ref mut end_offset)) =
                                current_transaction.take()
                            {
                                for (object_id, mutation) in 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 *object_id == super_block.root_parent_store_object_id {
                                        if let Mutation::ObjectStore(ObjectStoreMutation {
                                            item:
                                                Item {
                                                    key:
                                                        ObjectKey {
                                                            object_id,
                                                            data:
                                                                ObjectKeyData::Attribute(
                                                                    DEFAULT_DATA_ATTRIBUTE_ID,
                                                                    AttributeKey::Extent(
                                                                        ExtentKey { range },
                                                                    ),
                                                                ),
                                                            ..
                                                        },
                                                    value:
                                                        ObjectValue::Extent(ExtentValue::Some {
                                                            device_offset,
                                                            ..
                                                        }),
                                                    ..
                                                },
                                            ..
                                        }) = mutation
                                        {
                                            let handle = &mut reader.handle();
                                            if *object_id != handle.object_id() {
                                                continue;
                                            }
                                            if range.start
                                                != handle.start_offset() + handle.get_size()
                                            {
                                                bail!(anyhow!(FxfsError::Inconsistent).context(
                                                    format!(
                                                        "Unexpected journal extent {:?} -> {}, \
                                                        expected start: {}",
                                                        range,
                                                        device_offset,
                                                        handle.get_size()
                                                    )
                                                ));
                                            }
                                            handle.push_extent(
                                                *device_offset..*device_offset + range.length()?,
                                            );
                                        }
                                    }
                                    // If a MarkForDeletion mutation is found, we want to skip
                                    // checksum validation of prior writes.
                                    if let Mutation::Allocator(
                                        AllocatorMutation::MarkForDeletion(owner_object_id),
                                    ) = mutation
                                    {
                                        marked_for_deletion.insert(
                                            *owner_object_id,
                                            reader.journal_file_checkpoint().file_offset,
                                        );
                                    }
                                }
                                *end_offset = reader.journal_file_checkpoint().file_offset;
                            }
                        }
                        JournalRecord::Discard(offset) => {
                            if let Some(transaction) = current_transaction.as_ref() {
                                if transaction.0.file_offset < offset {
                                    // Odd, but OK.
                                    continue;
                                }
                            }
                            current_transaction = None;
                            while let Some(transaction) = transactions.last() {
                                if transaction.0.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();
        }

        // Validate all the mutations.
        let mut checksum_list = ChecksumList::new(device_flushed_offset);
        let mut valid_to = reader.journal_file_checkpoint().file_offset;
        'bad_replay: for (checkpoint, mutations, _) in &transactions {
            for (object_id, mutation) in mutations {
                if !self.validate_mutation(&mutation)? {
                    info!(?mutation, "Stopping replay at bad mutation");
                    valid_to = checkpoint.file_offset;
                    break 'bad_replay;
                }
                self.update_checksum_list(
                    checkpoint.file_offset,
                    *object_id,
                    &mutation,
                    &mut checksum_list,
                )?;
            }
        }

        // Validate the checksums.
        let valid_to = checksum_list
            .verify(device.as_ref(), marked_for_deletion, valid_to)
            .await
            .context("Failed to validate checksums")?;

        // Apply the mutations.
        let last_checkpoint = if transactions.is_empty() {
            super_block.journal_checkpoint.clone()
        } else {
            // Loop used here in place of for {} else {}
            #[allow(clippy::never_loop)]
            'outer: loop {
                for (checkpoint, mutations, end_offset) in transactions {
                    if checkpoint.file_offset >= valid_to {
                        break 'outer checkpoint;
                    }
                    self.objects
                        .replay_mutations(
                            mutations,
                            &ApplyContext { mode: ApplyMode::Replay, checkpoint },
                            end_offset,
                        )
                        .await;
                }
                break reader.journal_file_checkpoint();
            }
        };

        let root_store = self.objects.root_store();
        root_store.on_replay_complete().await?;
        allocator.open().await?;

        // 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
            .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().unwrap();
            let mut reader_checkpoint = reader.journal_file_checkpoint();
            // Reset the stream to indicate that we've remounted the journal.
            reader_checkpoint.checksum ^= RESET_XOR;
            reader_checkpoint.version = LATEST_VERSION;
            inner.device_flushed_offset = device_flushed_offset;
            let mut writer_checkpoint = reader_checkpoint.clone();
            writer_checkpoint.file_offset =
                round_up(writer_checkpoint.file_offset, BLOCK_SIZE).unwrap();
            inner.flushed_offset = writer_checkpoint.file_offset;
            inner.writer.seek(writer_checkpoint);
            inner.output_reset_version = true;
            if last_checkpoint.file_offset < inner.flushed_offset {
                inner.discard_offset = Some(last_checkpoint.file_offset);
            }
        }

        self.objects.on_replay_complete().await?;

        if last_checkpoint.file_offset != reader.journal_file_checkpoint().file_offset {
            info!(
                checkpoint = last_checkpoint.file_offset,
                discarded_to = reader.journal_file_checkpoint().file_offset,
                "replay complete"
            );
        } else {
            info!(checkpoint = reader.journal_file_checkpoint().file_offset, "replay complete",);
        }
        Ok(())
    }

    /// 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<dyn Filesystem>) -> 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().unwrap().writer.journal_file_checkpoint()
        };

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

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

        let journal_handle;
        let super_block_a_handle;
        let super_block_b_handle;
        let root_store;
        let mut transaction = filesystem
            .clone()
            .new_transaction(&[], 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() },
            )
            .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,
            SuperBlockInstance::A.object_id(),
            HandleOptions::default(),
            None,
        )
        .await
        .context("create super block")?;
        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,
            SuperBlockInstance::B.object_id(),
            HandleOptions::default(),
            None,
        )
        .await
        .context("create super block")?;
        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")?;
        journal_handle
            .preallocate_range(&mut transaction, 0..self.chunk_size())
            .await
            .context("preallocate journal")?;

        // 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));

        transaction.commit().await?;

        self.inner.lock().unwrap().super_block = SuperBlock::new(
            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);
        self.write_super_block().await?;
        SuperBlock::shred(super_block_b_handle).await
    }

    /// Commits a transaction.
    pub async fn commit(&self, transaction: &mut Transaction<'_>) -> Result<u64, Error> {
        if transaction.is_empty() {
            return Ok(self.inner.lock().unwrap().writer.journal_file_checkpoint().file_offset);
        }

        let _guard = debug_assert_not_too_long!(self.commit_mutex.lock());
        self.pre_commit().await?;
        Ok(debug_assert_not_too_long!(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) -> Result<(), Error> {
        let handle;

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

            // 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() {
                return Ok(());
            }

            (size, inner.zero_offset)
        };

        let mut transaction = handle
            .new_transaction_with_options(Options {
                skip_journal_checks: true,
                borrow_metadata_space: true,
                allocator_reservation: Some(self.objects.metadata_reservation()),
                ..Default::default()
            })
            .await?;
        if let Some(size) = size {
            handle.preallocate_range(&mut transaction, 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).await;

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

        // 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 = &self.inner.lock().unwrap().super_block;
        let offset = super_block
            .journal_file_offsets
            .get(&object_id)
            .cloned()
            .unwrap_or(super_block.super_block_journal_file_offset);
        journal_file_checkpoint.file_offset >= offset
    }

    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 _write_guard;
        let (checkpoint, borrowed) = {
            let _sync_guard = debug_assert_not_too_long!(self.sync_mutex.lock());
            _write_guard = debug_assert_not_too_long!(self.writer_mutex.lock());
            let result = self.pad_to_block()?;
            self.flush_device(result.0.file_offset).await?;
            result
        };

        let mut new_super_block = self.inner.lock().unwrap().super_block.clone();

        let old_super_block_offset = new_super_block.journal_checkpoint.file_offset;

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

        new_super_block.generation =
            new_super_block.generation.checked_add(1).ok_or(FxfsError::Inconsistent)?;
        new_super_block.super_block_journal_file_offset = checkpoint.file_offset;
        new_super_block.journal_checkpoint = min_checkpoint.unwrap_or(checkpoint);
        new_super_block.journal_checkpoint.version = LATEST_VERSION;
        new_super_block.journal_file_offsets = journal_file_offsets;
        new_super_block.borrowed_metadata_space = borrowed;

        self.super_block_manager.save(&new_super_block, root_parent_store).await?;

        {
            let mut inner = self.inner.lock().unwrap();
            inner.super_block = new_super_block;
            inner.zero_offset = Some(round_down(old_super_block_offset, BLOCK_SIZE));
        }

        Ok(())
    }

    /// Flushes any buffered journal data to the device.  Note that this does not flush the device
    /// so it still does not guarantee data will 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 and the amount of borrowed metadata
    /// space at the point it was flushed.
    pub async fn sync(
        &self,
        options: SyncOptions<'_>,
    ) -> Result<Option<(JournalCheckpoint, u64)>, Error> {
        let _guard = debug_assert_not_too_long!(self.sync_mutex.lock());

        let (checkpoint, borrowed) = {
            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 = debug_assert_not_too_long!(self.writer_mutex.lock());

            self.pad_to_block()?
        };

        if options.flush_device {
            self.flush_device(checkpoint.file_offset).await?;
        }

        Ok(Some((checkpoint, borrowed)))
    }

    // 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, u64), Error> {
        let mut inner = self.inner.lock().unwrap();
        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, self.objects.borrowed_metadata_space()))
    }

    async fn flush_device(&self, checkpoint_offset: u64) -> Result<(), Error> {
        debug_assert_not_too_long!(poll_fn(|ctx| {
            let mut inner = self.inner.lock().unwrap();
            if inner.flushed_offset >= checkpoint_offset {
                Poll::Ready(Ok(()))
            } else if inner.terminate {
                Poll::Ready(Err(FxfsError::JournalFlushError))
            } else {
                inner.sync_waker = Some(ctx.waker().clone());
                Poll::Pending
            }
        }))?;

        let needs_flush = self.inner.lock().unwrap().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 to 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().unwrap();
                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).await;
            if trace {
                info!("J: did flush device");
            }
        }

        Ok(())
    }

    /// Returns a copy of the super-block.
    pub fn super_block(&self) -> SuperBlock {
        self.inner.lock().unwrap().super_block.clone()
    }

    /// Waits for there to be sufficient space in the journal.
    pub async fn check_journal_space(&self) -> Result<(), Error> {
        loop {
            let _ = debug_assert_not_too_long!({
                let mut inner = self.inner.lock().unwrap();
                if inner.terminate {
                    // If the flush error is set, this will never make progress, since we can't
                    // extend the journal any more.
                    break Err(anyhow!(FxfsError::JournalFlushError).context("Journal closed"));
                }
                if self.objects.last_end_offset() - inner.super_block.journal_checkpoint.file_offset
                    < inner.reclaim_size
                {
                    break Ok(());
                }
                if inner.disable_compactions {
                    break Err(anyhow!(FxfsError::JournalFlushError).context("Journal closed"));
                }
                let event = inner.reclaim_event.get_or_insert_with(|| Event::new());
                event.wait_or_dropped()
            });
        }
    }

    fn block_size(&self) -> u64 {
        BLOCK_SIZE
    }

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

    async fn write_and_apply_mutations(&self, transaction: &mut Transaction<'_>) -> u64 {
        let checkpoint_before;
        let checkpoint_after;
        {
            let _guard = debug_assert_not_too_long!(self.writer_mutex.lock());
            checkpoint_before = {
                let mut inner = self.inner.lock().unwrap();
                let checkpoint = inner.writer.journal_file_checkpoint();
                for TxnMutation { object_id, mutation, .. } in &transaction.mutations {
                    let mutation = if let Some(mutation) =
                        self.objects.encrypt_mutation(*object_id, mutation)
                    {
                        mutation
                    } else {
                        mutation.clone()
                    };
                    inner
                        .writer
                        .write_record(&JournalRecord::Mutation { object_id: *object_id, mutation });
                }
                checkpoint
            };
            // The call here isn't drop-safe.  If the future gets dropped, it will leave things in a
            // bad state and subsequent threads might try and commit another transaction which has
            // the potential to fire assertions.  With that said, this should only occur if there
            // has been another panic, since we take care not to drop futures at other other times.
            let maybe_mutation =
                self.objects.apply_transaction(transaction, &checkpoint_before).await;
            checkpoint_after = {
                let mut inner = self.inner.lock().unwrap();
                if let Some(mutation) = maybe_mutation {
                    inner.writer.write_record(&JournalRecord::Mutation { object_id: 0, mutation });
                }
                inner.writer.write_record(&JournalRecord::Commit);

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

        if let Some(waker) = self.inner.lock().unwrap().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) {
        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().unwrap();
                if flush_fut.is_none() && !flush_error {
                    if let Some(handle) = self.handle.get() {
                        if let Some((offset, buf)) = inner.writer.take_buffer(handle) {
                            flush_fut = Some(self.flush(offset, buf).boxed());
                        }
                    }
                }
                if inner.terminate && flush_fut.is_none() && compact_fut.is_none() {
                    return Poll::Ready(());
                }
                // 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.terminate
                    && !inner.disable_compactions
                    && self.objects.last_end_offset()
                        - inner.super_block.journal_checkpoint.file_offset
                        > inner.reclaim_size / 2
                {
                    compact_fut = Some(self.compact().boxed());
                    inner.compaction_running = true;
                }
                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 {
                        info!(error = e.as_value(), "Flush error");
                        self.inner.lock().unwrap().terminate();
                        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().unwrap();
                    if let Err(e) = result {
                        info!(error = e.as_value(), "Compaction error");
                        inner.terminate();
                    }
                    compact_fut = None;
                    inner.compaction_running = false;
                    inner.reclaim_event = None;
                    pending = false;
                }
            }
            if pending {
                return Poll::Pending;
            }
        })
        .await;
    }

    async fn flush(&self, offset: u64, mut buf: Buffer<'_>) -> Result<(), Error> {
        let len = buf.len() as u64;
        self.handle.get().unwrap().overwrite(offset, buf.as_mut()).await?;
        let mut inner = self.inner.lock().unwrap();
        if let Some(waker) = inner.sync_waker.take() {
            waker.wake();
        }
        inner.flushed_offset = offset + len;
        Ok(())
    }

    async fn compact(&self) -> Result<(), Error> {
        let trace = self.trace.load(Ordering::Relaxed);
        debug!("Compaction starting");
        if trace {
            info!("J: start compaction");
        }
        trace_duration!("Journal::compact");
        let earliest_version = self.objects.flush().await?;
        self.inner.lock().unwrap().super_block.earliest_version = earliest_version;
        self.write_super_block().await?;
        if trace {
            info!("J: end compaction");
        }
        debug!("Compaction finished");
        Ok(())
    }

    pub async fn stop_compactions(&self) {
        loop {
            let _ = debug_assert_not_too_long!({
                let mut inner = self.inner.lock().unwrap();
                inner.disable_compactions = true;
                if !inner.compaction_running {
                    return;
                }
                let event = inner.reclaim_event.get_or_insert_with(|| Event::new());
                event.wait_or_dropped()
            });
        }
    }

    /// Terminate all journal activity.
    pub fn terminate(&self) {
        self.inner.lock().unwrap().terminate();
    }
}

#[cfg(test)]
mod tests {
    use {
        crate::{
            filesystem::{Filesystem, FxFilesystem, SyncOptions},
            fsck::fsck,
            object_handle::{ObjectHandle, ReadObjectHandle, WriteObjectHandle},
            object_store::{
                directory::Directory,
                transaction::{Options, TransactionHandler},
                HandleOptions, ObjectStore,
            },
        },
        fuchsia_async as fasync,
        storage_device::{fake_device::FakeDevice, DeviceHolder},
    };

    const TEST_DEVICE_BLOCK_SIZE: u32 = 512;

    #[fasync::run_singlethreaded(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
                .clone()
                .new_transaction(&[], 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());
            buf.as_mut_slice().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();
            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(TEST_DEVICE_BLOCK_SIZE as usize);
            assert_eq!(handle.read(0, buf.as_mut()).await.expect("read failed"), TEST_DATA.len());
            assert_eq!(&buf.as_slice()[..TEST_DATA.len()], TEST_DATA);
            fsck(&fs, None).await.expect("fsck failed");
            fs.close().await.expect("Close failed");
        }
    }

    #[fasync::run_singlethreaded(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
                .clone()
                .new_transaction(&[], 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());
            buf.as_mut_slice().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
                    .clone()
                    .new_transaction(&[], 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());
                buf.as_mut_slice().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();
        let fs = FxFilesystem::open(device).await.expect("open failed");
        fsck(&fs, None).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(TEST_DEVICE_BLOCK_SIZE as usize);
                assert_eq!(
                    handle.read(0, buf.as_mut()).await.expect("read failed"),
                    TEST_DATA.len()
                );
                assert_eq!(&buf.as_slice()[..TEST_DATA.len()], 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
                .clone()
                .new_transaction(&[], 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());
            buf.as_mut_slice().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();
        let fs = FxFilesystem::open(device).await.expect("open failed");
        {
            fsck(&fs, None).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
                .expect(&format!("open_object failed (object_id: {})", object_id));
                let mut buf = handle.allocate_buffer(TEST_DEVICE_BLOCK_SIZE as usize);
                assert_eq!(
                    handle.read(0, buf.as_mut()).await.expect("read failed"),
                    TEST_DATA.len()
                );
                assert_eq!(&buf.as_slice()[..TEST_DATA.len()], TEST_DATA);
            }
        }
        fs.close().await.expect("close failed");
    }
}