// 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.
//
// The super-block contains the starting offset and checksum for the journal file and sufficient
// information to locate the initial extents for the journal.  The super-block is written using the
// same per-block checksum that is used for the journal file.

mod reader;
pub mod super_block;
mod writer;

use {
    crate::{
        errors::FxfsError,
        object_handle::ObjectHandle,
        object_store::{
            allocator::{Allocator, SimpleAllocator},
            constants::SUPER_BLOCK_OBJECT_ID,
            directory::Directory,
            filesystem::{Filesystem, Mutations, ObjectFlush, ObjectManager, SyncOptions},
            graveyard::Graveyard,
            journal::{
                reader::{JournalReader, ReadResult},
                super_block::SuperBlock,
                writer::JournalWriter,
            },
            record::{ObjectItem, ObjectKey},
            transaction::{
                AssociatedObject, Mutation, ObjectStoreMutation, Transaction, TxnMutation,
            },
            HandleOptions, ObjectStore, StoreObjectHandle,
        },
    },
    anyhow::{anyhow, Context, Error},
    bincode::serialize_into,
    byteorder::{ByteOrder, LittleEndian},
    rand::Rng,
    serde::{Deserialize, Serialize},
    std::{
        clone::Clone,
        iter::IntoIterator,
        sync::{
            atomic::{self, AtomicBool},
            Arc, Mutex,
        },
        vec::Vec,
    },
};

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

// 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.
const RECLAIM_SIZE: u64 = 262_144;

// After replaying the journal, it's possible that the stream doesn't end cleanly, in which case the
// next journal block needs to indicate this.  This is done by pretending the previous block's
// checksum is xored with this value, and using that as the seed for the next journal block.
const RESET_XOR: u64 = 0xffffffffffffffff;

type Checksum = u64;

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

impl JournalCheckpoint {
    fn new(file_offset: u64, checksum: Checksum) -> JournalCheckpoint {
        JournalCheckpoint { file_offset, checksum }
    }
}

// All journal blocks are covered by a fletcher64 checksum as the last 8 bytes in a block.
fn fletcher64(buf: &[u8], previous: u64) -> u64 {
    assert!(buf.len() % 4 == 0);
    let mut lo = previous as u32;
    let mut hi = (previous >> 32) as u32;
    for chunk in buf.chunks(4) {
        lo = lo.wrapping_add(LittleEndian::read_u32(chunk));
        hi = hi.wrapping_add(lo);
    }
    (hi as u64) << 32 | lo as u64
}

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

fn journal_handle_options() -> HandleOptions {
    HandleOptions { overwrite: true, ..Default::default() }
}

fn clone_mutations<'a>(transaction: &Transaction<'_>) -> Vec<TxnMutation<'a>> {
    transaction
        .mutations
        .iter()
        .map(|m| TxnMutation {
            object_id: m.object_id,
            mutation: m.mutation.clone(),
            associated_object: None,
        })
        .collect()
}

/// 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>,
    writer: futures::lock::Mutex<JournalWriter<StoreObjectHandle<ObjectStore>>>,
    inner: Mutex<Inner>,
    trace: AtomicBool,
}

struct Inner {
    needs_super_block: bool,
    should_flush: bool,
    super_block: SuperBlock,
}

impl Journal {
    pub fn new(objects: Arc<ObjectManager>) -> Journal {
        let starting_checksum = rand::thread_rng().gen();
        Journal {
            objects: objects,
            writer: futures::lock::Mutex::new(JournalWriter::new(
                None,
                BLOCK_SIZE as usize,
                starting_checksum,
            )),
            inner: Mutex::new(Inner {
                needs_super_block: true,
                super_block: SuperBlock::default(),
                should_flush: false,
            }),
            trace: AtomicBool::new(false),
        }
    }

    pub fn set_trace(&self, v: bool) {
        self.trace.store(v, atomic::Ordering::Relaxed);
    }

    /// Reads a super-block and then replays journaled records.
    pub async fn replay(&self, filesystem: Arc<dyn Filesystem>) -> Result<(), Error> {
        let device = filesystem.device();
        let (super_block, mut reader) = SuperBlock::read(device).await?;
        log::info!("replaying journal, superblock: {:?}", super_block);

        let allocator = Arc::new(SimpleAllocator::new(
            filesystem.clone(),
            super_block.allocator_object_id,
            false,
        ));
        self.objects.set_allocator(allocator.clone());

        let root_parent =
            ObjectStore::new_empty(None, super_block.root_parent_store_object_id, filesystem);

        while let Some(item) = reader.next_item().await? {
            root_parent.apply_mutation(Mutation::insert_object(item.key, item.value), true).await;
        }

        {
            let mut inner = self.inner.lock().unwrap();
            inner.needs_super_block = false;
            inner.super_block = super_block.clone();
        }
        self.objects.set_root_parent_store_object_id(root_parent.store_object_id());
        let mut mutations = Vec::new();
        let mut journal_file_checkpoint = None;
        let mut end_block = false;
        root_parent.lazy_open_store(super_block.root_store_object_id);
        self.objects.set_root_store_object_id(super_block.root_store_object_id);
        let mut reader = JournalReader::new(
            ObjectStore::open_object(
                &root_parent,
                super_block.journal_object_id,
                journal_handle_options(),
            )
            .await?,
            self.block_size(),
            &super_block.journal_checkpoint,
        );
        loop {
            let current_checkpoint = Some(reader.journal_file_checkpoint());
            match reader.deserialize().await? {
                ReadResult::Reset => mutations.clear(), // Discard pending mutations
                ReadResult::Some(record) => {
                    end_block = false;
                    match record {
                        JournalRecord::EndBlock => {
                            reader.skip_to_end_of_block();
                            end_block = true;
                        }
                        JournalRecord::Mutation { object_id, mutation } => {
                            if mutations.len() == 0 {
                                journal_file_checkpoint = current_checkpoint;
                            }
                            mutations.push((object_id, mutation));
                        }
                        JournalRecord::Commit => {
                            if let Some(checkpoint) = journal_file_checkpoint.take() {
                                if self.trace.load(atomic::Ordering::Relaxed) {
                                    log::info!("REPLAY {}", checkpoint.file_offset);
                                }
                                for (object_id, mutation) in mutations {
                                    // Snoop the mutations for any that might apply to the journal
                                    // file to ensure that we accurately track changes in size.
                                    let associated_object = match (object_id, &mutation) {
                                        (
                                            store_object_id,
                                            Mutation::ObjectStore(ObjectStoreMutation {
                                                item:
                                                    ObjectItem {
                                                        key: ObjectKey { object_id, .. }, ..
                                                    },
                                                ..
                                            }),
                                        ) if store_object_id
                                            == super_block.root_parent_store_object_id
                                            && *object_id == super_block.journal_object_id =>
                                        {
                                            Some(reader.handle() as &_)
                                        }
                                        _ => None,
                                    };

                                    self.apply_mutation(
                                        object_id,
                                        &checkpoint,
                                        mutation,
                                        true,
                                        associated_object,
                                    )
                                    .await;
                                }
                                mutations = Vec::new();
                            }
                        }
                    }
                }
                // This is expected when we reach the end of the journal stream.
                ReadResult::ChecksumMismatch => break,
            }
        }
        // Configure the journal writer so that we can continue.
        {
            let mut checkpoint =
                JournalCheckpoint::new(reader.read_offset(), reader.last_read_checksum());
            if 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 {}",
                    checkpoint.file_offset, super_block.super_block_journal_file_offset
                )));
            }
            let mut writer = self.writer.lock().await;
            writer.set_handle(reader.take_handle());
            // If the last entry wasn't an end_block, then we need to reset the stream.
            if !end_block {
                checkpoint.checksum ^= RESET_XOR;
            }
            writer.seek_to_checkpoint(checkpoint);
        }

        let root_store = self.objects.root_store();
        root_store.ensure_open().await?;
        self.objects.register_graveyard(Arc::new(
            Graveyard::open(&self.objects.root_store(), root_store.graveyard_directory_object_id())
                .await
                .context(format!(
                    "failed to open graveyard (object_id: {})",
                    root_store.graveyard_directory_object_id()
                ))?,
        ));

        log::info!("replay done");
        Ok(())
    }

    /// Creates an empty filesystem with the minimum viable objects (including a root parent and
    /// root store but no further child stores).  Nothing is written to the device until sync is
    /// called.
    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 = 2;
        const INIT_ROOT_STORE_OBJECT_ID: u64 = 3;
        const INIT_ALLOCATOR_OBJECT_ID: u64 = 4;

        let checkpoint = self.writer.lock().await.journal_file_checkpoint();

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

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

        let journal_handle;
        let super_block_handle;
        let root_store;
        let mut transaction = filesystem.new_transaction(&[]).await?;
        root_store = root_parent
            .create_child_store_with_id(&mut transaction, INIT_ROOT_STORE_OBJECT_ID)
            .await
            .context("create root store")?;
        self.objects.set_root_store_object_id(root_store.store_object_id());

        // Create the super-block object...
        super_block_handle = ObjectStore::create_object_with_id(
            &root_store,
            &mut transaction,
            SUPER_BLOCK_OBJECT_ID,
            HandleOptions { overwrite: true, ..Default::default() },
        )
        .await
        .context("create super block")?;
        super_block_handle
            .extend(&mut transaction, super_block::first_extent())
            .await
            .context("extend super block")?;

        // the journal object...
        journal_handle =
            ObjectStore::create_object(&root_parent, &mut transaction, journal_handle_options())
                .await
                .context("create journal")?;
        journal_handle
            .preallocate_range(&mut transaction, 0..self.chunk_size())
            .await
            .context("preallocate journal")?;

        // the root store's graveyard and root directory...
        let graveyard = Arc::new(Graveyard::create(&mut transaction, &root_store).await?);
        root_store.set_graveyard_directory_object_id(&mut transaction, graveyard.object_id());
        self.objects.register_graveyard(graveyard);

        let root_directory = Directory::create(&mut transaction, &root_store)
            .await
            .context("create root directory")?;
        root_store.set_root_directory_object_id(&mut transaction, root_directory.object_id());

        self.commit(transaction).await;

        // Cache the super-block.
        self.inner.lock().unwrap().super_block = SuperBlock::new(
            root_parent.store_object_id(),
            root_store.store_object_id(),
            allocator.object_id(),
            journal_handle.object_id(),
            checkpoint,
        );

        // Initialize the journal writer.
        self.writer.lock().await.set_handle(journal_handle);
        Ok(())
    }

    /// Commits a transaction.
    pub async fn commit(&self, mut transaction: Transaction<'_>) {
        if transaction.is_empty() {
            return;
        }
        let mut writer = self.writer.lock().await;
        // TODO(csuter): handle the case where we are unable to extend the journal file.
        self.maybe_extend_journal_file(&mut writer).await.unwrap();
        // TODO(csuter): writing to the journal here can be asynchronous.
        let journal_file_checkpoint = writer.journal_file_checkpoint();
        writer.write_mutations(transaction.mutations.iter().cloned());
        if let Err(e) = writer.maybe_flush_buffer().await {
            // TODO(csuter): if writes to the journal start failing then we should prevent the
            // creation of new transactions.
            log::warn!("journal write failed: {}", e);
        }
        self.apply_mutations(std::mem::take(&mut transaction.mutations), journal_file_checkpoint)
            .await;
        let mut inner = self.inner.lock().unwrap();
        // 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.
        inner.should_flush = writer.journal_file_checkpoint().file_offset
            - inner.super_block.journal_checkpoint.file_offset
            > RECLAIM_SIZE / 2;
    }

    async fn maybe_extend_journal_file(
        &self,
        writer: &mut JournalWriter<StoreObjectHandle<ObjectStore>>,
    ) -> Result<(), Error> {
        // TODO(csuter): this currently assumes that a transaction can fit in CHUNK_SIZE.
        let file_offset = writer.journal_file_checkpoint().file_offset;
        let handle = match writer.handle() {
            None => return Ok(()),
            Some(handle) => handle,
        };
        let size = handle.get_size();
        if file_offset + self.chunk_size() <= size {
            return Ok(());
        }
        let mut transaction = handle.new_transaction().await?;
        handle.preallocate_range(&mut transaction, size..size + self.chunk_size()).await?;
        let journal_file_checkpoint = writer.journal_file_checkpoint();

        // We have to apply the mutations before writing them because we borrowed the writer for the
        // transaction.  First we clone the mutations without the associated objects since that's
        // where the handle is borrowed.
        let cloned_mutations = clone_mutations(&transaction);

        self.apply_mutations(std::mem::take(&mut transaction.mutations), journal_file_checkpoint)
            .await;

        std::mem::drop(transaction);
        writer.write_mutations(cloned_mutations);

        // We need to be sure that any journal records that arose from preallocation can fit in
        // within the old preallocated range.  If this situation arose (it shouldn't, so it would be
        // a bug if it did), then it could be fixed (e.g. by fsck) by forcing a sync of the root
        // store.
        assert!(writer.journal_file_checkpoint().file_offset <= size);
        let file_offset = writer.journal_file_checkpoint().file_offset;
        let handle = writer.handle().unwrap();
        assert!(file_offset + self.chunk_size() <= handle.get_size());
        Ok(())
    }

    async fn apply_mutations(
        &self,
        mutations: impl IntoIterator<Item = TxnMutation<'_>>,
        journal_file_checkpoint: JournalCheckpoint,
    ) {
        if self.trace.load(atomic::Ordering::Relaxed) {
            log::info!("BEGIN TXN {}", journal_file_checkpoint.file_offset);
        }
        for TxnMutation { object_id, mutation, associated_object } in mutations {
            self.apply_mutation(
                object_id,
                &journal_file_checkpoint,
                mutation,
                false,
                associated_object,
            )
            .await;
        }
        if self.trace.load(atomic::Ordering::Relaxed) {
            log::info!("END TXN");
        }
    }

    // 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.
    // Regardless, we want to keep track of the earliest mutation in the journal for a given object.
    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 apply_mutation(
        &self,
        object_id: u64,
        journal_file_checkpoint: &JournalCheckpoint,
        mutation: Mutation,
        filter: bool,
        object: Option<&dyn AssociatedObject>,
    ) {
        if !filter || self.should_apply(object_id, journal_file_checkpoint) {
            if self.trace.load(atomic::Ordering::Relaxed) {
                log::info!("applying mutation: {}: {:?}, filter: {}", object_id, mutation, filter);
            }
            self.objects
                .apply_mutation(object_id, mutation, filter, journal_file_checkpoint, object)
                .await;
        } else {
            if self.trace.load(atomic::Ordering::Relaxed) {
                log::info!("ignoring mutation: {}, {:?}", object_id, mutation);
            }
        }
    }

    pub async fn write_super_block(&self) -> Result<(), Error> {
        let root_parent_store = self.objects.root_parent_store();

        // First we must lock the root parent store so that no new entries are written to it.
        let sync = ObjectFlush::new(self.objects.clone(), root_parent_store.store_object_id());
        let mutable_layer = root_parent_store.tree().mutable_layer();
        let guard = mutable_layer.lock();

        // After locking, we need to flush the journal because it might have records that a new
        // super-block would refer to.
        let journal_file_checkpoint = {
            let mut writer = self.writer.lock().await;

            // We are holding the appropriate locks now (no new transaction can be applied whilst we
            // are holding the writer lock, so we can call ObjectFlush::begin for the root parent
            // object store.
            sync.begin();

            serialize_into(&mut *writer, &JournalRecord::EndBlock)?;
            writer.pad_to_block()?;
            writer.maybe_flush_buffer().await?;
            writer.journal_file_checkpoint()
        };

        // We need to flush previous writes to the device since the new super-block we are writing
        // relies on written data being observable.
        root_parent_store.device().flush().await?;

        // TODO(csuter): Here is the point where we should notify the allocator that it can now use
        // pending deallocations so long as they've been written to the journal.

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

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

        new_super_block.super_block_journal_file_offset = journal_file_checkpoint.file_offset;
        new_super_block.journal_checkpoint = min_checkpoint.unwrap_or(journal_file_checkpoint);
        new_super_block.journal_file_offsets = journal_file_offsets;

        new_super_block
            .write(
                &root_parent_store,
                ObjectStore::open_object(
                    &self.objects.root_store(),
                    SUPER_BLOCK_OBJECT_ID,
                    journal_handle_options(),
                )
                .await?,
            )
            .await?;

        {
            let mut inner = self.inner.lock().unwrap();
            inner.super_block = new_super_block;
            inner.needs_super_block = false;
        }

        sync.commit();
        std::mem::drop(guard);

        // The previous super-block is now guaranteed to be persisted (because we flushed the device
        // above), so we can free all journal space that it doesn't need.
        {
            let mut writer = self.writer.lock().await;

            if old_checkpoint_offset >= BLOCK_SIZE {
                let handle = writer.handle().unwrap();
                let mut transaction = handle.new_transaction().await?;
                let mut offset = old_checkpoint_offset;
                offset -= offset % BLOCK_SIZE;
                handle.zero(&mut transaction, 0..offset).await?;
                let cloned_mutations = clone_mutations(&transaction);
                self.apply_mutations(
                    std::mem::take(&mut transaction.mutations),
                    writer.journal_file_checkpoint(),
                )
                .await;
                std::mem::drop(transaction);
                writer.write_mutations(cloned_mutations);
            }
        }

        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.
    pub async fn sync(&self, _options: SyncOptions) -> Result<(), Error> {
        // TODO(csuter): There needs to be some kind of locking here.
        let needs_super_block = self.inner.lock().unwrap().needs_super_block;
        if needs_super_block {
            self.write_super_block().await?;
        }
        let mut writer = self.writer.lock().await;
        serialize_into(&mut *writer, &JournalRecord::EndBlock)?;
        writer.pad_to_block()?;
        writer.maybe_flush_buffer().await?;
        Ok(())
    }

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

    /// Returns whether or not a flush should be performed.  This is only updated after committing a
    /// transaction.
    pub fn should_flush(&self) -> bool {
        self.inner.lock().unwrap().should_flush
    }

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

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

impl<OH> JournalWriter<OH> {
    // Extends JournalWriter to write a transaction.
    fn write_mutations<'a>(&mut self, mutations: impl IntoIterator<Item = TxnMutation<'a>>) {
        for TxnMutation { object_id, mutation, .. } in mutations {
            self.write_record(&JournalRecord::Mutation { object_id, mutation });
        }
        self.write_record(&JournalRecord::Commit);
    }
}

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

    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(2048, 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(&[]).await.expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test")
                .await
                .expect("create_child_file failed");

            transaction.commit().await;
            let mut buf = handle.allocate_buffer(TEST_DATA.len());
            buf.as_mut_slice().copy_from_slice(TEST_DATA);
            handle.write(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()
        };

        {
            let fs = FxFilesystem::open(fs.take_device().await).await.expect("open failed");
            let handle =
                ObjectStore::open_object(&fs.root_store(), object_id, HandleOptions::default())
                    .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).await.expect("fsck failed");
        }
    }

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

        let device = DeviceHolder::new(FakeDevice::new(6144, 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(&[]).await.expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test")
                .await
                .expect("create_child_file failed");
            transaction.commit().await;
            let mut buf = handle.allocate_buffer(TEST_DATA.len());
            buf.as_mut_slice().copy_from_slice(TEST_DATA);
            handle.write(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(&[]).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;
                let mut buf = handle.allocate_buffer(TEST_DATA.len());
                buf.as_mut_slice().copy_from_slice(TEST_DATA);
                handle.write(0, buf.as_ref()).await.expect("write failed");
                object_ids.push(handle.object_id());
            }
        }

        let fs = FxFilesystem::open(fs.take_device().await).await.expect("open failed");
        fsck(&fs).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())
                        .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(&[]).await.expect("new_transaction failed");
            let handle = root_directory
                .create_child_file(&mut transaction, "test2")
                .await
                .expect("create_child_file failed");
            transaction.commit().await;
            let mut buf = handle.allocate_buffer(TEST_DATA.len());
            buf.as_mut_slice().copy_from_slice(TEST_DATA);
            handle.write(0, buf.as_ref()).await.expect("write failed");
            fs.sync(SyncOptions::default()).await.expect("sync failed");
            object_ids.push(handle.object_id());
        }

        let fs = FxFilesystem::open_with_trace(fs.take_device().await, false)
            .await
            .expect("open failed");
        {
            fsck(&fs).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())
                        .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);
            }
        }
    }
}