class RecoveryProtocol {

    State run() throws IOException, DamagedLogStorageException, DisposedException
    {
        State state = new State();
        SortedMap<Long,File> files = fileNames.getAllFiles( fileSystem, log );

        if ( files.entrySet().isEmpty() )
        {
            state.segments = new Segments( fileSystem, fileNames, readerPool, emptyList(), contentMarshal, logProvider, -1 );
            state.segments.rotate( -1, -1, -1 );
            state.terms = new Terms( -1, -1 );
            return state;
        }

        List<SegmentFile> segmentFiles = new ArrayList<>();
        SegmentFile segment = null;

        long expectedVersion = files.firstKey();
        boolean mustRecoverLastHeader = false;
        boolean skip = true; // the first file is treated the same as a skip

        for ( Map.Entry<Long,File> entry : files.entrySet() )
        {
            long fileNameVersion = entry.getKey();
            File file = entry.getValue();
            SegmentHeader header;

            checkVersionSequence( fileNameVersion, expectedVersion );

            try
            {
                header = loadHeader( fileSystem, file );
                checkVersionMatches( header.version(), fileNameVersion );
            }
            catch ( EndOfStreamException e )
            {
                if ( files.lastKey() != fileNameVersion )
                {
                    throw new DamagedLogStorageException( e, "Intermediate file with incomplete or no header found: %s", file );
                }
                else if ( files.size() == 1 )
                {
                    throw new DamagedLogStorageException( e, "Single file with incomplete or no header found: %s", file );
                }

                /* Last file header must be recovered by scanning next-to-last file and writing a new header based on that. */
                mustRecoverLastHeader = true;
                break;
            }

            segment = new SegmentFile( fileSystem, file, readerPool, fileNameVersion, contentMarshal, logProvider, header );
            segmentFiles.add( segment );

            if ( segment.header().prevIndex() != segment.header().prevFileLastIndex() )
            {
                log.info( format( "Skipping from index %d to %d.", segment.header().prevFileLastIndex(),
                        segment.header().prevIndex() + 1 ) );
                skip = true;
            }

            if ( skip )
            {
                state.prevIndex = segment.header().prevIndex();
                state.prevTerm = segment.header().prevTerm();
                skip = false;
            }

            expectedVersion++;
        }

        assert segment != null;

        state.appendIndex = segment.header().prevIndex();
        state.terms = new Terms( segment.header().prevIndex(), segment.header().prevTerm() );

        try ( IOCursor<EntryRecord> cursor = segment.getCursor( segment.header().prevIndex() + 1 ) )
        {
            while ( cursor.next() )
            {
                EntryRecord entry = cursor.get();
                state.appendIndex = entry.logIndex();
                state.terms.append( state.appendIndex, entry.logEntry().term() );
            }
        }

        if ( mustRecoverLastHeader )
        {
            SegmentHeader header = new SegmentHeader( state.appendIndex, expectedVersion, state.appendIndex, state.terms.latest() );
            log.warn( "Recovering last file based on next-to-last file. " + header );

            File file = fileNames.getForVersion( expectedVersion );
            writeHeader( fileSystem, file, header );

            segment = new SegmentFile( fileSystem, file, readerPool, expectedVersion, contentMarshal, logProvider, header );
            segmentFiles.add( segment );
        }

        state.segments = new Segments( fileSystem, fileNames, readerPool, segmentFiles, contentMarshal, logProvider,
                segment.header().version() );

        return state;
    }

      RecoveryProtocol( FileSystemAbstraction fileSystem, FileNames fileNames, ReaderPool readerPool,
            ChannelMarshal<ReplicatedContent> contentMarshal, LogProvider logProvider );

}