fn recover_queue_imp<M: ReplayMachine, FA: Factory<M>>(
        file_system: Arc<F>,
        recovery_cfg: RecoveryConfig,
        files: &mut Vec<FileToRecover<F>>,
        replay_machine_factory: &FA,
    ) -> Result<M> {
        if recovery_cfg.concurrency == 0 || files.is_empty() {
            return Ok(replay_machine_factory.new_target());
        }
        let queue = recovery_cfg.queue;
        let concurrency = recovery_cfg.concurrency;
        let recovery_mode = recovery_cfg.mode;
        let recovery_read_block_size = recovery_cfg.read_block_size as usize;

        let max_chunk_size = std::cmp::max((files.len() + concurrency - 1) / concurrency, 1);
        let chunks = files.par_chunks_mut(max_chunk_size);
        let chunk_count = chunks.len();
        debug_assert!(chunk_count <= concurrency);
        let sequential_replay_machine = chunks
            .enumerate()
            .map(|(index, chunk)| {
                let mut reader =
                    LogItemBatchFileReader::new(recovery_read_block_size);
                let mut sequential_replay_machine = replay_machine_factory.new_target();
                let file_count = chunk.len();
                for (i, f) in chunk.iter_mut().enumerate() {
                    let is_last_file = index == chunk_count - 1 && i == file_count - 1;
                    match build_file_reader(file_system.as_ref(), f.handle.clone(), None) {
                        Err(e) => {
                            let is_local_tail = f.handle.file_size()? <= LogFileFormat::len();
                            if recovery_mode == RecoveryMode::TolerateAnyCorruption
                              || recovery_mode == RecoveryMode::TolerateTailCorruption
                                && is_last_file && is_local_tail {
                                warn!(
                                    "File header is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(0)?;
                                f.version = Some(Version::default());
                                continue;
                            } else {
                                error!(
                                    "Failed to open log file due to broken header: {:?}:{}",
                                    queue, f.seq
                                );
                                return Err(e);
                            }
                        },
                        Ok(file_reader) => {
                            reader.open(FileId { queue, seq: f.seq }, file_reader)?;
                        }
                    }
                    // Update file version of each log file.
                    f.version = Some(reader.file_format().unwrap().version());
                    loop {
                        match reader.next() {
                            Ok(Some(item_batch)) => {
                                sequential_replay_machine
                                    .replay(item_batch, FileId { queue, seq: f.seq })?;
                            }
                            Ok(None) => break,
                            Err(e)
                                if recovery_mode == RecoveryMode::TolerateTailCorruption
                                    && is_last_file =>
                            {
                                warn!(
                                    "The last log file is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(reader.valid_offset())?;
                                break;
                            }
                            Err(e) if recovery_mode == RecoveryMode::TolerateAnyCorruption => {
                                warn!(
                                    "File is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(reader.valid_offset())?;
                                break;
                            }
                            Err(e) => {
                                error!(
                                    "Failed to open log file due to broken entry: {:?}:{} offset={}",
                                    queue, f.seq, reader.valid_offset()
                                );
                                return Err(e);
                            }
                        }
                    }
                }
                Ok(sequential_replay_machine)
            })
            .try_reduce(
                || replay_machine_factory.new_target(),
                |mut sequential_replay_machine_left, sequential_replay_machine_right| {
                    sequential_replay_machine_left.merge(sequential_replay_machine_right, queue)?;
                    Ok(sequential_replay_machine_left)
                },
            )?;

        Ok(sequential_replay_machine)
    }
/// Manually reads through log items in all available log files of the
/// specified queue, and replays them to specific [`ReplayMachine`]s
/// that can be constructed via `machine_factory`.
Manually reads through log items in all available log files of the
specified queue, and replays them to specific [`ReplayMachine`]s
that can be constructed via `machine_factory`.
Manually reads through log items in all available log files of the specified queue and replays them to specific [ ` ReplayMachine ` ] s that can be constructed via ` machine_factory ` .
fn recover_queue_imp<M: ReplayMachine, FA: Factory<M>>(
        file_system: Arc<F>,
        recovery_cfg: RecoveryConfig,
        files: &mut Vec<FileToRecover<F>>,
        replay_machine_factory: &FA,
    ) -> Result<M> {
        if recovery_cfg.concurrency == 0 || files.is_empty() {
            return Ok(replay_machine_factory.new_target());
        }
        let queue = recovery_cfg.queue;
        let concurrency = recovery_cfg.concurrency;
        let recovery_mode = recovery_cfg.mode;
        let recovery_read_block_size = recovery_cfg.read_block_size as usize;
        let max_chunk_size = std::cmp::max((files.len() + concurrency - 1) / concurrency, 1);
        let chunks = files.par_chunks_mut(max_chunk_size);
        let chunk_count = chunks.len();
        debug_assert!(chunk_count <= concurrency);
        let sequential_replay_machine = chunks
            .enumerate()
            .map(|(index, chunk)| {
                let mut reader =
                    LogItemBatchFileReader::new(recovery_read_block_size);
                let mut sequential_replay_machine = replay_machine_factory.new_target();
                let file_count = chunk.len();
                for (i, f) in chunk.iter_mut().enumerate() {
                    let is_last_file = index == chunk_count - 1 && i == file_count - 1;
                    match build_file_reader(file_system.as_ref(), f.handle.clone(), None) {
                        Err(e) => {
                            let is_local_tail = f.handle.file_size()? <= LogFileFormat::len();
                            if recovery_mode == RecoveryMode::TolerateAnyCorruption
                              || recovery_mode == RecoveryMode::TolerateTailCorruption
                                && is_last_file && is_local_tail {
                                warn!(
                                    "File header is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(0)?;
                                f.version = Some(Version::default());
                                continue;
                            } else {
                                error!(
                                    "Failed to open log file due to broken header: {:?}:{}",
                                    queue, f.seq
                                );
                                return Err(e);
                            }
                        },
                        Ok(file_reader) => {
                            reader.open(FileId { queue, seq: f.seq }, file_reader)?;
                        }
                    }
                    // Update file version of each log file.
                    f.version = Some(reader.file_format().unwrap().version());
                    loop {
                        match reader.next() {
                            Ok(Some(item_batch)) => {
                                sequential_replay_machine
                                    .replay(item_batch, FileId { queue, seq: f.seq })?;
                            }
                            Ok(None) => break,
                            Err(e)
                                if recovery_mode == RecoveryMode::TolerateTailCorruption
                                    && is_last_file =>
                            {
                                warn!(
                                    "The last log file is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(reader.valid_offset())?;
                                break;
                            }
                            Err(e) if recovery_mode == RecoveryMode::TolerateAnyCorruption => {
                                warn!(
                                    "File is corrupted but ignored: {:?}:{}, {}",
                                    queue, f.seq, e
                                );
                                f.handle.truncate(reader.valid_offset())?;
                                break;
                            }
                            Err(e) => {
                                error!(
                                    "Failed to open log file due to broken entry: {:?}:{} offset={}",
                                    queue, f.seq, reader.valid_offset()
                                );
                                return Err(e);
                            }
                        }
                    }
                }
                Ok(sequential_replay_machine)
            })
            .try_reduce(
                || replay_machine_factory.new_target(),
                |mut sequential_replay_machine_left, sequential_replay_machine_right| {
                    sequential_replay_machine_left.merge(sequential_replay_machine_right, queue)?;
                    Ok(sequential_replay_machine_left)
                },
            )?;
        Ok(sequential_replay_machine)
    }
fn recover_queue_imp < M : ReplayMachine , FA : Factory < M > > ( file_system : Arc < F > , recovery_cfg : RecoveryConfig , files : & mut Vec < FileToRecover < F > > , replay_machine_factory : & FA , ) -> Result < M > { if recovery_cfg . concurrency == 0 || files . is_empty ( ) { return Ok ( replay_machine_factory . new_target ( ) ) ; } let queue = recovery_cfg . queue ; let concurrency = recovery_cfg . concurrency ; let recovery_mode = recovery_cfg . mode ; let recovery_read_block_size = recovery_cfg . read_block_size as usize ; let max_chunk_size = std :: cmp :: max ( ( files . len ( ) + concurrency - 1 ) / concurrency , 1 ) ; let chunks = files . par_chunks_mut ( max_chunk_size ) ; let chunk_count = chunks . len ( ) ; debug_assert ! ( chunk_count concurrency ) ; let sequential_replay_machine = chunks . enumerate ( ) . map ( | ( index , chunk ) | { let mut reader = LogItemBatchFileReader :: new ( recovery_read_block_size ) ; let mut sequential_replay_machine = replay_machine_factory . new_target ( ) ; let file_count = chunk . len ( ) ; for ( i , f ) in chunk . iter_mut ( ) . enumerate ( ) { let is_last_file = index == chunk_count - 1 && i == file_count - 1 ; match build_file_reader ( file_system . as_ref ( ) , f . handle . clone ( ) , None ) { Err ( e ) => { let is_local_tail = f . handle . file_size ( ) ? <= LogFileFormat :: len ( ) ; if recovery_mode == RecoveryMode :: TolerateAnyCorruption || recovery_mode == RecoveryMode :: TolerateTailCorruption && is_last_file && is_local_tail { warn ! ( " " queue f seq e ) ; f . handle . truncate ( 0 ) ? ; f . version = Some ( Version :: default ( ) ) ; continue ; } else { error ! ( " " queue f seq ) ; return Err ( e ) ; } } , Ok ( file_reader ) => { reader . open ( FileId { queue , seq : f . seq } , file_reader ) ? ; } } // Update file version of each log file. f . version = Some ( reader . file_format ( ) . unwrap ( ) . version ( ) ) ; loop { match reader . next ( ) { Ok ( Some ( item_batch ) ) => { sequential_replay_machine . replay ( item_batch , FileId { queue , seq : f . seq } ) ? ; } Ok ( None ) => break , Err ( e ) if recovery_mode == RecoveryMode :: TolerateTailCorruption && is_last_file => { warn ! ( " " queue f seq e ) ; f . handle . truncate ( reader . valid_offset ( ) ) ? ; break ; } Err ( e ) if recovery_mode == RecoveryMode :: TolerateAnyCorruption => { warn ! ( " " queue f seq e ) ; f . handle . truncate ( reader . valid_offset ( ) ) ? ; break ; } Err ( e ) => { error ! ( " " queue f seq reader valid_offset ( ) ) ; return Err ( e ) ; } } } } Ok ( sequential_replay_machine ) } ) . try_reduce ( | | replay_machine_factory . new_target ( ) , | mut sequential_replay_machine_left , sequential_replay_machine_right | { sequential_replay_machine_left . merge ( sequential_replay_machine_right , queue ) ? ; Ok ( sequential_replay_machine_left ) } , ) ? ; Ok ( sequential_replay_machine ) }
Manually reads through log items in all available log files of the
specified queue, and replays them to specific [`ReplayMachine`]s
that can be constructed via `machine_factory`.
Manually reads through log items in all available log files of the specified queue and replays them to specific [ ` ReplayMachine ` ] s that can be constructed via ` machine_factory ` .
file_system
Arc<F>
recovery_cfg
RecoveryConfig
files
&mut Vec<FileToRecover<F>>
replay_machine_factory
&FA