<commit_before>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.aeron.cluster.service;

import io.aeron.archive.client.AeronArchive;
import io.aeron.cluster.codecs.RecoveryPlanDecoder;
import io.aeron.cluster.codecs.RecoveryPlanEncoder;
import org.agrona.BitUtil;
import org.agrona.CloseHelper;
import org.agrona.LangUtil;
import org.agrona.concurrent.UnsafeBuffer;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

import static io.aeron.archive.client.AeronArchive.NULL_POSITION;
import static java.nio.file.StandardOpenOption.*;
import static org.agrona.BitUtil.*;

/**
 * A log of recordings that make up the history of a Raft log. Entries are in order.
 * <p>
 * The log is made up of entries of log terms or snapshots to roll up state as of a log position and leadership term.
 * <p>
 * The latest state is made up of a the latest snapshot followed by any term logs which follow. It is possible that
 * the a snapshot is taken mid term and therefore the latest state is the snapshot plus the log of messages which
 * begin before the snapshot but continues after it.
 * <p>
 * Record layout as follows:
 * <pre>
 *   0                   1                   2                   3
 *   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 *  |                        Recording ID                           |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                     Leadership Term ID                        |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |             Log Position at beginning of term                 |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                    Term Position/Length                       |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |   Timestamp at beginning of term or when snapshot was taken   |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                       Member ID vote                          |
 *  +---------------------------------------------------------------+
 *  |                 Entry Type (Log or Snapshot)                  |
 *  +---------------------------------------------------------------+
 *  |                                                               |
 *  |                                                              ...
 * ...                Repeats to the end of the log                 |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 * </pre>
 */
public class RecordingLog
{
    /**
     * A copy of the entry in the log.
     */
    public static final class Entry
    {
        public final long recordingId;
        public final long leadershipTermId;
        public final long logPosition;
        public final long termPosition;
        public final long timestamp;
        public final int memberIdVote;
        public final int type;
        public final int entryIndex;

        /**
         * A new entry in the recording log.
         *
         * @param recordingId      of the entry in an archive.
         * @param leadershipTermId of this entry.
         * @param logPosition      accumulated position of the log over leadership terms for the beginning of the term.
         * @param termPosition     position reached within the current leadership term, same at leadership term length.
         * @param timestamp        of this entry.
         * @param memberIdVote     which member this node voted for in the election.
         * @param type             of the entry as a log of a term or a snapshot.
         * @param entryIndex       of the entry on disk.
         */
        public Entry(
            final long recordingId,
            final long leadershipTermId,
            final long logPosition,
            final long termPosition,
            final long timestamp,
            final int memberIdVote,
            final int type,
            final int entryIndex)
        {
            this.recordingId = recordingId;
            this.leadershipTermId = leadershipTermId;
            this.logPosition = logPosition;
            this.termPosition = termPosition;
            this.timestamp = timestamp;
            this.memberIdVote = memberIdVote;
            this.type = type;
            this.entryIndex = entryIndex;
        }

        public Entry(final RecoveryPlanDecoder.StepsDecoder decoder)
        {
            this.recordingId = decoder.entryRecordingId();
            this.leadershipTermId = decoder.entryLeadershipTermId();
            this.logPosition = decoder.entryLogPosition();
            this.termPosition = decoder.entryTermPosition();
            this.timestamp = decoder.entryTimestamp();
            this.memberIdVote = decoder.entryMemberIdVote();
            this.type = decoder.entryType();
            this.entryIndex = decoder.entryIndex();
        }

        public void encode(final RecoveryPlanEncoder.StepsEncoder encoder)
        {
            encoder
                .entryRecordingId(recordingId)
                .entryLeadershipTermId(leadershipTermId)
                .entryLogPosition(logPosition)
                .entryTermPosition(termPosition)
                .entryTimestamp(timestamp)
                .entryMemberIdVote(memberIdVote)
                .entryType(type)
                .entryIndex(entryIndex);
        }

        public String toString()
        {
            return "Entry{" +
                "recordingId=" + recordingId +
                ", leadershipTermId=" + leadershipTermId +
                ", logPosition=" + logPosition +
                ", termPosition=" + termPosition +
                ", timestamp=" + timestamp +
                ", memberIdVote=" + memberIdVote +
                ", type=" + type +
                ", entryIndex=" + entryIndex +
                '}';
        }
    }

    /**
     * Steps in a recovery plan.
     */
    public static class ReplayStep
    {
        public final long recordingStartPosition;
        public final long recordingStopPosition;
        public final Entry entry;

        public ReplayStep(final long recordingStartPosition, final long recordingStopPosition, final Entry entry)
        {
            this.recordingStartPosition = recordingStartPosition;
            this.recordingStopPosition = recordingStopPosition;
            this.entry = entry;
        }

        public ReplayStep(final RecoveryPlanDecoder.StepsDecoder decoder)
        {
            this.recordingStartPosition = decoder.recordingStartPosition();
            this.recordingStopPosition = decoder.recordingStopPosition();
            entry = new Entry(decoder);
        }

        public void encode(final RecoveryPlanEncoder.StepsEncoder encoder)
        {
            encoder
                .recordingStartPosition(recordingStartPosition)
                .recordingStopPosition(recordingStopPosition);
            entry.encode(encoder);
        }

        public String toString()
        {
            return "ReplayStep{" +
                "recordingStartPosition=" + recordingStartPosition +
                ", recordingStopPosition=" + recordingStopPosition +
                ", entry=" + entry +
                '}';
        }
    }

    /**
     * The snapshot and steps to recover the state of a cluster.
     */
    public static class RecoveryPlan
    {
        public final long lastLeadershipTermId;
        public final long lastLogPosition;
        public final long lastTermPositionCommitted;
        public final long lastTermPositionAppended;
        public final ReplayStep snapshotStep;
        public final ArrayList<ReplayStep> termSteps;
        public final RecoveryPlanEncoder encoder = new RecoveryPlanEncoder();
        public final RecoveryPlanDecoder decoder = new RecoveryPlanDecoder();
        public final UnsafeBuffer unsafeBuffer = new UnsafeBuffer();

        public RecoveryPlan(
            final long lastLeadershipTermId,
            final long lastLogPosition,
            final long lastTermPositionCommitted,
            final long lastTermPositionAppended,
            final ReplayStep snapshotStep,
            final ArrayList<ReplayStep> termSteps)
        {
            this.lastLeadershipTermId = lastLeadershipTermId;
            this.lastLogPosition = lastLogPosition;
            this.lastTermPositionCommitted = lastTermPositionCommitted;
            this.lastTermPositionAppended = lastTermPositionAppended;
            this.snapshotStep = snapshotStep;
            this.termSteps = termSteps;
        }

        public RecoveryPlan(final ByteBuffer byteBuffer)
        {
            this(byteBuffer.array());
        }

        public RecoveryPlan(final byte[] decodingBuffer)
        {
            unsafeBuffer.wrap(decodingBuffer);
            decoder.wrap(unsafeBuffer, 0, RecoveryPlanDecoder.BLOCK_LENGTH, RecoveryPlanDecoder.SCHEMA_VERSION);

            this.lastLeadershipTermId = decoder.lastLeadershipTermId();
            this.lastLogPosition = decoder.lastLogPosition();
            this.lastTermPositionCommitted = decoder.lastTermPositionCommitted();
            this.lastTermPositionAppended = decoder.lastTermPositionAppended();

            ReplayStep snapshot = null;
            termSteps = new ArrayList<>();
            int stepNumber = 0;

            for (final RecoveryPlanDecoder.StepsDecoder stepDecoder : decoder.steps())
            {
                if (0 == stepNumber && stepDecoder.entryType() == RecordingLog.ENTRY_TYPE_SNAPSHOT)
                {
                    snapshot = new ReplayStep(stepDecoder);
                }
                else
                {
                    termSteps.add(new ReplayStep(stepDecoder));
                }

                stepNumber++;
            }

            this.snapshotStep = snapshot;
        }

        public byte[] encode()
        {
            final int numberOfSteps = termSteps.size() + (null != snapshotStep ? 1 : 0);
            final int length = RecoveryPlanEncoder.BLOCK_LENGTH +
                RecoveryPlanEncoder.StepsEncoder.sbeHeaderSize() +
                numberOfSteps * RecoveryPlanEncoder.StepsEncoder.sbeBlockLength();
            final byte[] bytes = new byte[length];

            unsafeBuffer.wrap(bytes);
            encoder.wrap(unsafeBuffer, 0);

            encoder
                .lastLeadershipTermId(lastLeadershipTermId)
                .lastLogPosition(lastLogPosition)
                .lastTermPositionCommitted(lastTermPositionCommitted)
                .lastTermPositionAppended(lastTermPositionAppended);

            final RecoveryPlanEncoder.StepsEncoder stepEncoder = encoder.stepsCount(numberOfSteps);

            if (null != snapshotStep)
            {
                snapshotStep.encode(stepEncoder);
                stepEncoder.next();
            }

            for (final ReplayStep step : termSteps)
            {
                step.encode(stepEncoder);
                stepEncoder.next();
            }

            return bytes;
        }

        public String toString()
        {
            return "RecoveryPlan{" +
                "lastLeadershipTermId=" + lastLeadershipTermId +
                ", lastLogPosition=" + lastLogPosition +
                ", lastTermPositionCommitted=" + lastTermPositionCommitted +
                ", lastTermPositionAppended=" + lastTermPositionAppended +
                ", snapshotStep=" + snapshotStep +
                ", termSteps=" + termSteps +
                '}';
        }
    }

    /**
     * Filename for the history of log terms and snapshot recordings.
     */
    public static final String RECORDING_LOG_FILE_NAME = "recordings.log";

    /**
     * Represents a value that is not set or invalid.
     */
    public static final int NULL_VALUE = -1;

    /**
     * The index entry is for a recording of messages within a term to the consensus log.
     */
    public static final int ENTRY_TYPE_TERM = 0;

    /**
     * The index entry is for a recording of a snapshot of state taken as of a position in the log.
     */
    public static final int ENTRY_TYPE_SNAPSHOT = 1;<commit_msg>[Java] Correct file name for recording log.<commit_after>/*
 * Copyright 2014-2018 Real Logic Ltd.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.aeron.cluster.service;

import io.aeron.archive.client.AeronArchive;
import io.aeron.cluster.codecs.RecoveryPlanDecoder;
import io.aeron.cluster.codecs.RecoveryPlanEncoder;
import org.agrona.BitUtil;
import org.agrona.CloseHelper;
import org.agrona.LangUtil;
import org.agrona.concurrent.UnsafeBuffer;

import java.io.File;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

import static io.aeron.archive.client.AeronArchive.NULL_POSITION;
import static java.nio.file.StandardOpenOption.*;
import static org.agrona.BitUtil.*;

/**
 * A log of recordings that make up the history of a Raft log. Entries are in order.
 * <p>
 * The log is made up of entries of log terms or snapshots to roll up state as of a log position and leadership term.
 * <p>
 * The latest state is made up of a the latest snapshot followed by any term logs which follow. It is possible that
 * the a snapshot is taken mid term and therefore the latest state is the snapshot plus the log of messages which
 * begin before the snapshot but continues after it.
 * <p>
 * Record layout as follows:
 * <pre>
 *   0                   1                   2                   3
 *   0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
 *  +-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
 *  |                        Recording ID                           |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                     Leadership Term ID                        |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |             Log Position at beginning of term                 |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                    Term Position/Length                       |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |   Timestamp at beginning of term or when snapshot was taken   |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 *  |                       Member ID vote                          |
 *  +---------------------------------------------------------------+
 *  |                 Entry Type (Log or Snapshot)                  |
 *  +---------------------------------------------------------------+
 *  |                                                               |
 *  |                                                              ...
 * ...                Repeats to the end of the log                 |
 *  |                                                               |
 *  +---------------------------------------------------------------+
 * </pre>
 */
public class RecordingLog
{
    /**
     * A copy of the entry in the log.
     */
    public static final class Entry
    {
        public final long recordingId;
        public final long leadershipTermId;
        public final long logPosition;
        public final long termPosition;
        public final long timestamp;
        public final int memberIdVote;
        public final int type;
        public final int entryIndex;

        /**
         * A new entry in the recording log.
         *
         * @param recordingId      of the entry in an archive.
         * @param leadershipTermId of this entry.
         * @param logPosition      accumulated position of the log over leadership terms for the beginning of the term.
         * @param termPosition     position reached within the current leadership term, same at leadership term length.
         * @param timestamp        of this entry.
         * @param memberIdVote     which member this node voted for in the election.
         * @param type             of the entry as a log of a term or a snapshot.
         * @param entryIndex       of the entry on disk.
         */
        public Entry(
            final long recordingId,
            final long leadershipTermId,
            final long logPosition,
            final long termPosition,
            final long timestamp,
            final int memberIdVote,
            final int type,
            final int entryIndex)
        {
            this.recordingId = recordingId;
            this.leadershipTermId = leadershipTermId;
            this.logPosition = logPosition;
            this.termPosition = termPosition;
            this.timestamp = timestamp;
            this.memberIdVote = memberIdVote;
            this.type = type;
            this.entryIndex = entryIndex;
        }

        public Entry(final RecoveryPlanDecoder.StepsDecoder decoder)
        {
            this.recordingId = decoder.entryRecordingId();
            this.leadershipTermId = decoder.entryLeadershipTermId();
            this.logPosition = decoder.entryLogPosition();
            this.termPosition = decoder.entryTermPosition();
            this.timestamp = decoder.entryTimestamp();
            this.memberIdVote = decoder.entryMemberIdVote();
            this.type = decoder.entryType();
            this.entryIndex = decoder.entryIndex();
        }

        public void encode(final RecoveryPlanEncoder.StepsEncoder encoder)
        {
            encoder
                .entryRecordingId(recordingId)
                .entryLeadershipTermId(leadershipTermId)
                .entryLogPosition(logPosition)
                .entryTermPosition(termPosition)
                .entryTimestamp(timestamp)
                .entryMemberIdVote(memberIdVote)
                .entryType(type)
                .entryIndex(entryIndex);
        }

        public String toString()
        {
            return "Entry{" +
                "recordingId=" + recordingId +
                ", leadershipTermId=" + leadershipTermId +
                ", logPosition=" + logPosition +
                ", termPosition=" + termPosition +
                ", timestamp=" + timestamp +
                ", memberIdVote=" + memberIdVote +
                ", type=" + type +
                ", entryIndex=" + entryIndex +
                '}';
        }
    }

    /**
     * Steps in a recovery plan.
     */
    public static class ReplayStep
    {
        public final long recordingStartPosition;
        public final long recordingStopPosition;
        public final Entry entry;

        public ReplayStep(final long recordingStartPosition, final long recordingStopPosition, final Entry entry)
        {
            this.recordingStartPosition = recordingStartPosition;
            this.recordingStopPosition = recordingStopPosition;
            this.entry = entry;
        }

        public ReplayStep(final RecoveryPlanDecoder.StepsDecoder decoder)
        {
            this.recordingStartPosition = decoder.recordingStartPosition();
            this.recordingStopPosition = decoder.recordingStopPosition();
            entry = new Entry(decoder);
        }

        public void encode(final RecoveryPlanEncoder.StepsEncoder encoder)
        {
            encoder
                .recordingStartPosition(recordingStartPosition)
                .recordingStopPosition(recordingStopPosition);
            entry.encode(encoder);
        }

        public String toString()
        {
            return "ReplayStep{" +
                "recordingStartPosition=" + recordingStartPosition +
                ", recordingStopPosition=" + recordingStopPosition +
                ", entry=" + entry +
                '}';
        }
    }

    /**
     * The snapshot and steps to recover the state of a cluster.
     */
    public static class RecoveryPlan
    {
        public final long lastLeadershipTermId;
        public final long lastLogPosition;
        public final long lastTermPositionCommitted;
        public final long lastTermPositionAppended;
        public final ReplayStep snapshotStep;
        public final ArrayList<ReplayStep> termSteps;
        public final RecoveryPlanEncoder encoder = new RecoveryPlanEncoder();
        public final RecoveryPlanDecoder decoder = new RecoveryPlanDecoder();
        public final UnsafeBuffer unsafeBuffer = new UnsafeBuffer();

        public RecoveryPlan(
            final long lastLeadershipTermId,
            final long lastLogPosition,
            final long lastTermPositionCommitted,
            final long lastTermPositionAppended,
            final ReplayStep snapshotStep,
            final ArrayList<ReplayStep> termSteps)
        {
            this.lastLeadershipTermId = lastLeadershipTermId;
            this.lastLogPosition = lastLogPosition;
            this.lastTermPositionCommitted = lastTermPositionCommitted;
            this.lastTermPositionAppended = lastTermPositionAppended;
            this.snapshotStep = snapshotStep;
            this.termSteps = termSteps;
        }

        public RecoveryPlan(final ByteBuffer byteBuffer)
        {
            this(byteBuffer.array());
        }

        public RecoveryPlan(final byte[] decodingBuffer)
        {
            unsafeBuffer.wrap(decodingBuffer);
            decoder.wrap(unsafeBuffer, 0, RecoveryPlanDecoder.BLOCK_LENGTH, RecoveryPlanDecoder.SCHEMA_VERSION);

            this.lastLeadershipTermId = decoder.lastLeadershipTermId();
            this.lastLogPosition = decoder.lastLogPosition();
            this.lastTermPositionCommitted = decoder.lastTermPositionCommitted();
            this.lastTermPositionAppended = decoder.lastTermPositionAppended();

            ReplayStep snapshot = null;
            termSteps = new ArrayList<>();
            int stepNumber = 0;

            for (final RecoveryPlanDecoder.StepsDecoder stepDecoder : decoder.steps())
            {
                if (0 == stepNumber && stepDecoder.entryType() == RecordingLog.ENTRY_TYPE_SNAPSHOT)
                {
                    snapshot = new ReplayStep(stepDecoder);
                }
                else
                {
                    termSteps.add(new ReplayStep(stepDecoder));
                }

                stepNumber++;
            }

            this.snapshotStep = snapshot;
        }

        public byte[] encode()
        {
            final int numberOfSteps = termSteps.size() + (null != snapshotStep ? 1 : 0);
            final int length = RecoveryPlanEncoder.BLOCK_LENGTH +
                RecoveryPlanEncoder.StepsEncoder.sbeHeaderSize() +
                numberOfSteps * RecoveryPlanEncoder.StepsEncoder.sbeBlockLength();
            final byte[] bytes = new byte[length];

            unsafeBuffer.wrap(bytes);
            encoder.wrap(unsafeBuffer, 0);

            encoder
                .lastLeadershipTermId(lastLeadershipTermId)
                .lastLogPosition(lastLogPosition)
                .lastTermPositionCommitted(lastTermPositionCommitted)
                .lastTermPositionAppended(lastTermPositionAppended);

            final RecoveryPlanEncoder.StepsEncoder stepEncoder = encoder.stepsCount(numberOfSteps);

            if (null != snapshotStep)
            {
                snapshotStep.encode(stepEncoder);
                stepEncoder.next();
            }

            for (final ReplayStep step : termSteps)
            {
                step.encode(stepEncoder);
                stepEncoder.next();
            }

            return bytes;
        }

        public String toString()
        {
            return "RecoveryPlan{" +
                "lastLeadershipTermId=" + lastLeadershipTermId +
                ", lastLogPosition=" + lastLogPosition +
                ", lastTermPositionCommitted=" + lastTermPositionCommitted +
                ", lastTermPositionAppended=" + lastTermPositionAppended +
                ", snapshotStep=" + snapshotStep +
                ", termSteps=" + termSteps +
                '}';
        }
    }

    /**
     * Filename for the history of log terms and snapshot recordings.
     */
    public static final String RECORDING_LOG_FILE_NAME = "recording.log";

    /**
     * Represents a value that is not set or invalid.
     */
    public static final int NULL_VALUE = -1;

    /**
     * The index entry is for a recording of messages within a term to the consensus log.
     */
    public static final int ENTRY_TYPE_TERM = 0;

    /**
     * The index entry is for a recording of a snapshot of state taken as of a position in the log.
     */
    public static final int ENTRY_TYPE_SNAPSHOT = 1;
{'fraction_non_alphanumeric': 0.10988722790453712, 'fraction_numerical': 0.00427110261886029, 'mean_word_length': 1.50487987987988, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 44, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4333960', 'n_tokens_mistral': 6376, 'n_tokens_neox': 5942, 'n_words': 1769}