/// Reads a super-block and then replays journaled records.
Reads a super-block and then replays journaled records.
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 { item }, 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(
            root_parent
                .open_object(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() {
                                log::debug!("REPLAY {}", checkpoint.file_offset);
                                for mutation in mutations {
                                    self.apply_mutation(mutation.0, &checkpoint, mutation.1, true)
                                        .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 checkpoint =
                JournalCheckpoint::new(reader.read_offset(), reader.last_read_checksum());
            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.
            writer.seek_to_checkpoint(checkpoint, !end_block);
        }
        log::info!("replay done");
        Ok(())
    }
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 ! ( " " 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 { item } , 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 ( root_parent . open_object ( 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 ( ) { log :: debug ! ( " " checkpoint file_offset ) ; for mutation in mutations { self . apply_mutation ( mutation . 0 , & checkpoint , mutation . 1 , true ) . 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 checkpoint = JournalCheckpoint :: new ( reader . read_offset ( ) , reader . last_read_checksum ( ) ) ; 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. writer . seek_to_checkpoint ( checkpoint , ! end_block ) ; } log :: info ! ( " " ) ; Ok ( ( ) ) }
Reads a super-block and then replays journaled records.
Reads a super - block and then replays journaled records .
&self
filesystem
Arc<dyn Filesystem>