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

import io.aeron.*;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.cluster.client.ClusterException;
import io.aeron.cluster.codecs.*;
import io.aeron.cluster.service.Cluster;
import io.aeron.cluster.service.ClusterMarkFile;
import io.aeron.cluster.service.CommitPos;
import io.aeron.cluster.service.RecoveryState;
import io.aeron.exceptions.TimeoutException;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import io.aeron.protocol.DataHeaderFlyweight;
import io.aeron.security.Authenticator;
import io.aeron.status.ReadableCounter;
import org.agrona.BitUtil;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.MutableDirectBuffer;
import org.agrona.collections.ArrayListUtil;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.collections.LongHashSet;
import org.agrona.concurrent.Agent;
import org.agrona.concurrent.AgentInvoker;
import org.agrona.concurrent.EpochClock;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.status.CountersReader;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.ChannelUri.SPY_QUALIFIER;
import static io.aeron.CommonContext.*;
import static io.aeron.archive.client.AeronArchive.NULL_LENGTH;
import static io.aeron.archive.codecs.SourceLocation.LOCAL;
import static io.aeron.cluster.ClusterSession.State.*;
import static io.aeron.cluster.ConsensusModule.Configuration.*;
import static io.aeron.cluster.ServiceAck.hasReachedPosition;
import static io.aeron.cluster.ServiceAck.newArray;
import static io.aeron.logbuffer.FrameDescriptor.FRAME_ALIGNMENT;
import static java.lang.Long.MAX_VALUE;

class ConsensusModuleAgent implements Agent, MemberStatusListener
{
    private final int memberId;
    private final long sessionTimeoutMs;
    private final long leaderHeartbeatIntervalMs;
    private final long leaderHeartbeatTimeoutMs;
    private final long serviceHeartbeatTimeoutMs;
    private long nextSessionId = 1;
    private long leadershipTermId = NULL_VALUE;
    private long expectedAckPosition = 0;
    private long serviceAckId = 0;
    private long lastAppendedPosition = 0;
    private long followerCommitPosition = 0;
    private long timeOfLastLogUpdateMs = 0;
    private long cachedTimeMs;
    private long clusterTimeMs = NULL_VALUE;
    private long lastRecordingId = RecordingPos.NULL_RECORDING_ID;
    private int logPublicationInitialTermId = NULL_VALUE;
    private int logPublicationTermBufferLength = NULL_VALUE;
    private int logPublicationMtuLength = NULL_VALUE;
    private ReadableCounter appendedPosition;
    private Counter commitPosition;
    private ConsensusModule.State state = ConsensusModule.State.INIT;
    private Cluster.Role role;
    private ClusterMember[] clusterMembers;
    private ClusterMember leaderMember;
    private final ClusterMember thisMember;
    private long[] rankedPositions;
    private final ServiceAck[] serviceAcks;
    private final Counter clusterRoleCounter;
    private final ClusterMarkFile markFile;
    private final AgentInvoker aeronClientInvoker;
    private final EpochClock epochClock;
    private final Counter moduleState;
    private final Counter controlToggle;
    private final TimerService timerService;
    private final ConsensusModuleAdapter consensusModuleAdapter;
    private final ServiceProxy serviceProxy;
    private final IngressAdapter ingressAdapter;
    private final EgressPublisher egressPublisher;
    private final LogPublisher logPublisher;
    private LogAdapter logAdapter;
    private final MemberStatusAdapter memberStatusAdapter;
    private final MemberStatusPublisher memberStatusPublisher = new MemberStatusPublisher();
    private final Long2ObjectHashMap<ClusterSession> sessionByIdMap = new Long2ObjectHashMap<>();
    private final ArrayList<ClusterSession> pendingSessions = new ArrayList<>();
    private final ArrayList<ClusterSession> rejectedSessions = new ArrayList<>();
    private final LongHashSet missedTimersSet = new LongHashSet();
    private final Authenticator authenticator;
    private final ClusterSessionProxy sessionProxy;
    private final Aeron aeron;
    private AeronArchive archive;
    private final ConsensusModule.Context ctx;
    private final MutableDirectBuffer tempBuffer;
    private final Counter[] serviceHeartbeats;
    private final IdleStrategy idleStrategy;
    private final RecordingLog recordingLog;
    private RecordingLog.RecoveryPlan recoveryPlan;
    private Election election;
    private String logRecordingChannel;
    private String liveLogDestination;
    private String clientFacingEndpoints;

    ConsensusModuleAgent(final ConsensusModule.Context ctx)
    {
        this.ctx = ctx;
        this.aeron = ctx.aeron();
        this.epochClock = ctx.epochClock();
        this.sessionTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.sessionTimeoutNs());
        this.leaderHeartbeatIntervalMs = TimeUnit.NANOSECONDS.toMillis(ctx.leaderHeartbeatIntervalNs());
        this.leaderHeartbeatTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.leaderHeartbeatTimeoutNs());
        this.serviceHeartbeatTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.serviceHeartbeatTimeoutNs());
        this.egressPublisher = ctx.egressPublisher();
        this.moduleState = ctx.moduleStateCounter();
        this.controlToggle = ctx.controlToggleCounter();
        this.logPublisher = ctx.logPublisher();
        this.idleStrategy = ctx.idleStrategy();
        this.timerService = new TimerService(this);
        this.clusterMembers = ClusterMember.parse(ctx.clusterMembers());
        this.sessionProxy = new ClusterSessionProxy(egressPublisher);
        this.memberId = ctx.clusterMemberId();
        this.clusterRoleCounter = ctx.clusterNodeCounter();
        this.markFile = ctx.clusterMarkFile();
        this.recordingLog = ctx.recordingLog();
        this.tempBuffer = ctx.tempBuffer();
        this.serviceHeartbeats = ctx.serviceHeartbeatCounters();
        this.serviceAcks = newArray(ctx.serviceCount());

        aeronClientInvoker = aeron.conductorAgentInvoker();
        aeronClientInvoker.invoke();

        rankedPositions = new long[ClusterMember.quorumThreshold(clusterMembers.length)];
        role(Cluster.Role.FOLLOWER);

        thisMember = clusterMembers[memberId];
        leaderMember = thisMember;
        final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
        memberStatusUri.put(ENDPOINT_PARAM_NAME, thisMember.memberFacingEndpoint());

        final int statusStreamId = ctx.memberStatusStreamId();
        memberStatusAdapter = new MemberStatusAdapter(
            aeron.addSubscription(memberStatusUri.toString(), statusStreamId), this);

        ClusterMember.addMemberStatusPublications(clusterMembers, thisMember, memberStatusUri, statusStreamId, aeron);

        ingressAdapter = new IngressAdapter(this, ctx.invalidRequestCounter());

        final ChannelUri archiveUri = ChannelUri.parse(ctx.archiveContext().controlRequestChannel());
        ClusterMember.checkArchiveEndpoint(thisMember, archiveUri);
        archiveUri.put(ENDPOINT_PARAM_NAME, thisMember.archiveEndpoint());
        ctx.archiveContext().controlRequestChannel(archiveUri.toString());

        consensusModuleAdapter = new ConsensusModuleAdapter(
            aeron.addSubscription(ctx.serviceControlChannel(), ctx.consensusModuleStreamId()), this);
        serviceProxy = new ServiceProxy(
            aeron.addPublication(ctx.serviceControlChannel(), ctx.serviceStreamId()));

        authenticator = ctx.authenticatorSupplier().get();
    }

    public void onClose()
    {
        CloseHelper.close(archive);

        if (!ctx.ownsAeronClient())
        {
            for (final ClusterSession session : sessionByIdMap.values())
            {
                session.close();
            }

            CloseHelper.close(memberStatusAdapter);
            ClusterMember.closeMemberPublications(clusterMembers);

            logPublisher.disconnect();
            CloseHelper.close(ingressAdapter);
            CloseHelper.close(serviceProxy);
            CloseHelper.close(consensusModuleAdapter);
        }
    }

    public void onStart()
    {
        archive = AeronArchive.connect(ctx.archiveContext().clone());
        recoveryPlan = recordingLog.createRecoveryPlan(archive, ctx.serviceCount());

        try (Counter ignore = addRecoveryStateCounter(recoveryPlan))
        {
            if (!recoveryPlan.snapshots.isEmpty())
            {
                recoverFromSnapshot(recoveryPlan.snapshots.get(0), archive);
            }

            awaitServiceAcks(expectedAckPosition);
        }

        if (ConsensusModule.State.SUSPENDED != state)
        {
            state(ConsensusModule.State.ACTIVE);
        }

        timeOfLastLogUpdateMs = cachedTimeMs = epochClock.time();
        leadershipTermId = recoveryPlan.lastLeadershipTermId;

        election = new Election(
            true,
            leadershipTermId,
            recoveryPlan.appendedLogPosition,
            clusterMembers,
            thisMember,
            memberStatusAdapter,
            memberStatusPublisher,
            ctx,
            this);
    }

    public int doWork()
    {
        int workCount = 0;

        boolean isSlowTickCycle = false;
        final long nowMs = epochClock.time();
        if (cachedTimeMs != nowMs)
        {
            cachedTimeMs = nowMs;
            if (Cluster.Role.LEADER == role)
            {
                clusterTimeMs(nowMs);
            }

            isSlowTickCycle = true;
        }

        if (null != election)
        {
            workCount += election.doWork(nowMs);
        }
        else
        {
            if (Cluster.Role.LEADER == role && ConsensusModule.State.ACTIVE == state)
            {
                workCount += ingressAdapter.poll();
                workCount += timerService.poll(nowMs);
            }
            else if (Cluster.Role.FOLLOWER == role &&
                (ConsensusModule.State.ACTIVE == state || ConsensusModule.State.SUSPENDED == state))
            {
                final int count = logAdapter.poll(followerCommitPosition);
                if (0 == count && logAdapter.isImageClosed())
                {
                    enterElection(nowMs);

                    return 1;
                }

                workCount += count;
            }

            workCount += memberStatusAdapter.poll();
            workCount += updateMemberPosition(nowMs);
            workCount += consensusModuleAdapter.poll();
        }

        if (isSlowTickCycle)
        {
            workCount += slowTickCycle(nowMs);
        }

        return workCount;
    }

    public String roleName()
    {
        return "consensus-module";
    }

    public void onSessionConnect(
        final long correlationId,
        final int responseStreamId,
        final String responseChannel,
        final byte[] encodedCredentials)
    {
        final long sessionId = nextSessionId++;
        final ClusterSession session = new ClusterSession(sessionId, responseStreamId, responseChannel);
        session.connect(aeron);
        session.lastActivity(clusterTimeMs, correlationId);

        if (pendingSessions.size() + sessionByIdMap.size() < ctx.maxConcurrentSessions())
        {
            authenticator.onConnectRequest(sessionId, encodedCredentials, clusterTimeMs);
            pendingSessions.add(session);
        }
        else
        {
            rejectedSessions.add(session);
        }
    }

    public void onSessionClose(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session)
        {
            session.close(CloseReason.CLIENT_ACTION);

            if (logPublisher.appendSessionClose(session, clusterTimeMs))
            {
                sessionByIdMap.remove(clusterSessionId);
            }
        }
    }

    public ControlledFragmentAssembler.Action onSessionMessage(
        final DirectBuffer buffer,
        final int offset,
        final int length,
        final long clusterSessionId,
        final long correlationId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null == session || session.state() == CLOSED)
        {
            return ControlledFragmentHandler.Action.CONTINUE;
        }

        if (session.state() == OPEN && logPublisher.appendMessage(buffer, offset, length, clusterTimeMs))
        {
            session.lastActivity(clusterTimeMs, correlationId);
            return ControlledFragmentHandler.Action.CONTINUE;
        }

        return ControlledFragmentHandler.Action.ABORT;
    }

    public void onSessionKeepAlive(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session && session.state() == OPEN)
        {
            session.timeOfLastActivityMs(clusterTimeMs);
        }
    }

    public void onChallengeResponse(
        final long correlationId, final long clusterSessionId, final byte[] encodedCredentials)
    {
        for (int lastIndex = pendingSessions.size() - 1, i = lastIndex; i >= 0; i--)
        {
            final ClusterSession session = pendingSessions.get(i);

            if (session.id() == clusterSessionId && session.state() == CHALLENGED)
            {
                session.lastActivity(clusterTimeMs, correlationId);
                authenticator.onChallengeResponse(clusterSessionId, encodedCredentials, clusterTimeMs);
                break;
            }
        }
    }

    public boolean onTimerEvent(final long correlationId, final long nowMs)
    {
        return Cluster.Role.LEADER != role || logPublisher.appendTimer(correlationId, nowMs);
    }

    public void onCanvassPosition(final long logLeadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (null != election)
        {
            election.onCanvassPosition(logLeadershipTermId, logPosition, followerMemberId);
        }
        else if (Cluster.Role.LEADER == role)
        {
            memberStatusPublisher.newLeadershipTerm(
                clusterMembers[followerMemberId].publication(),
                this.leadershipTermId,
                recordingLog.getTermEntry(this.leadershipTermId).termBaseLogPosition,
                this.leadershipTermId,
                thisMember.id(),
                logPublisher.sessionId());
        }
    }

    public void onRequestVote(
        final long logLeadershipTermId, final long logPosition, final long candidateTermId, final int candidateId)
    {
        if (null != election)
        {
            election.onRequestVote(logLeadershipTermId, logPosition, candidateTermId, candidateId);
        }
        else if (candidateTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
            election.onRequestVote(logLeadershipTermId, logPosition, candidateTermId, candidateId);
        }
    }

    public void onVote(
        final long candidateTermId,
        final long logLeadershipTermId,
        final long logPosition,
        final int candidateMemberId,
        final int followerMemberId,
        final boolean vote)
    {
        if (null != election)
        {
            election.onVote(
                candidateTermId, logLeadershipTermId, logPosition, candidateMemberId, followerMemberId, vote);
        }
    }

    public void onNewLeadershipTerm(
        final long logLeadershipTermId,
        final long logPosition,
        final long leadershipTermId,
        final int leaderMemberId,
        final int logSessionId)
    {
        if (null != election)
        {
            election.onNewLeadershipTerm(
                logLeadershipTermId, logPosition, leadershipTermId, leaderMemberId, logSessionId);
        }
        else if (leadershipTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
        }
    }

    public void onAppendedPosition(final long leadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (null != election)
        {
            election.onAppendedPosition(leadershipTermId, logPosition, followerMemberId);
        }
        else if (Cluster.Role.LEADER == role && leadershipTermId == this.leadershipTermId)
        {
            clusterMembers[followerMemberId].logPosition(logPosition);
        }
    }

    public void onCommitPosition(final long leadershipTermId, final long logPosition, final int leaderMemberId)
    {
        if (null != election)
        {
            election.onCommitPosition(leadershipTermId, logPosition, leaderMemberId);
        }
        else if (Cluster.Role.FOLLOWER == role && leadershipTermId == this.leadershipTermId)
        {
            timeOfLastLogUpdateMs = cachedTimeMs;
            followerCommitPosition = logPosition;
        }
        else if (leadershipTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
        }
    }

    public void onCatchupPosition(final long leadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (Cluster.Role.LEADER == role && leadershipTermId == this.leadershipTermId)
        {
            final String replayChannel = new ChannelUriStringBuilder()
                .media(CommonContext.UDP_MEDIA)
                .endpoint(clusterMembers[followerMemberId].transferEndpoint())
                .isSessionIdTagged(true)
                .sessionId(ConsensusModule.Configuration.LOG_PUBLICATION_SESSION_ID_TAG)
                .build();

            final ClusterMember member = clusterMembers[followerMemberId];

            if (member.catchupReplaySessionId() == Aeron.NULL_VALUE)
            {
                member.catchupReplaySessionId(archive.startReplay(
                    logRecordingId(), logPosition, Long.MAX_VALUE, replayChannel, ctx.logStreamId()));
            }
        }
    }

    public void onStopCatchup(final int replaySessionId, final int followerMemberId)
    {
        final ClusterMember member = clusterMembers[followerMemberId];

        if (member.catchupReplaySessionId() == replaySessionId)
        {
            archive.stopReplay(member.catchupReplaySessionId());
            member.catchupReplaySessionId(Aeron.NULL_VALUE);
        }
    }

    public void onRecoveryPlanQuery(final long correlationId, final int requestMemberId, final int leaderMemberId)
    {
        if (leaderMemberId == memberId)
        {
            memberStatusPublisher.recoveryPlan(
                clusterMembers[requestMemberId].publication(),
                correlationId,
                requestMemberId,
                leaderMemberId,
                recoveryPlan);
        }
    }

    public void onRecoveryPlan(final RecoveryPlanDecoder recoveryPlanDecoder)
    {
    }

    public void onRecordingLogQuery(
        final long correlationId,
        final int requestMemberId,
        final int leaderMemberId,
        final long fromLeadershipTermId,
        final int count,
        final boolean includeSnapshots)
    {
        if (leaderMemberId == memberId)
        {
            memberStatusPublisher.recordingLog(
                clusterMembers[requestMemberId].publication(),
                correlationId,
                requestMemberId,
                leaderMemberId,
                recordingLog,
                fromLeadershipTermId,
                count,
                includeSnapshots);
        }
    }

    public void onRecordingLog(final RecordingLogDecoder recordingLogDecoder)
    {
    }

    void state(final ConsensusModule.State state)
    {
        this.state = state;
        moduleState.set(state.code());
    }

    void role(final Cluster.Role role)
    {
        this.role = role;
        clusterRoleCounter.setOrdered(role.code());
    }

    Cluster.Role role()
    {
        return role;
    }

    long prepareForElection(final long logPosition)
    {
        final RecordingExtent recordingExtent = new RecordingExtent();

        long recordingId = RecordingPos.getRecordingId(aeron.countersReader(), appendedPosition.counterId());
        if (RecordingPos.NULL_RECORDING_ID == recordingId)
        {
            recordingId = recordingLog.getTermEntry(leadershipTermId).recordingId;
        }

        stopLogRecording();

        idleStrategy.reset();
        while (true)
        {
            if (0 == archive.listRecording(recordingId, recordingExtent))
            {
                throw new ClusterException("recording not found id=" + recordingId);
            }

            if (AeronArchive.NULL_POSITION != recordingExtent.stopPosition)
            {
                break;
            }

            idle();
        }

        if (recordingExtent.stopPosition > logPosition)
        {
            archive.truncateRecording(recordingId, logPosition);
        }

        lastAppendedPosition = recordingExtent.stopPosition;
        followerCommitPosition = recordingExtent.stopPosition;

        lastRecordingId = recordingId;
        logPublicationInitialTermId = recordingExtent.initialTermId;
        logPublicationTermBufferLength = recordingExtent.termBufferLength;
        logPublicationMtuLength = recordingExtent.mtuLength;

        commitPosition.setOrdered(recordingExtent.stopPosition);
        clearSessionsAfter(recordingExtent.stopPosition);

        return recordingExtent.stopPosition;
    }

    void stopLogRecording()
    {
        if (null != logRecordingChannel)
        {
            archive.stopRecording(logRecordingChannel, ctx.logStreamId());
            logRecordingChannel = null;
        }

        if (null != logAdapter && null != liveLogDestination)
        {
            logAdapter.removeDestination(liveLogDestination);
            liveLogDestination = null;
        }
    }

    void appendedPositionCounter(final ReadableCounter appendedPositionCounter)
    {
        this.appendedPosition = appendedPositionCounter;
    }

    void commitPositionCounter(final Counter commitPositionCounter)
    {
        this.commitPosition = commitPositionCounter;
    }

    void clearSessionsAfter(final long logPosition)
    {
        for (final Iterator<ClusterSession> i = sessionByIdMap.values().iterator(); i.hasNext(); )
        {
            final ClusterSession session = i.next();
            if (session.openedLogPosition() >= logPosition)
            {
                i.remove();
                session.close();
            }
        }

        for (final ClusterSession session : pendingSessions)
        {
            session.close();
        }

        pendingSessions.clear();
    }

    void onServiceCloseSession(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session)
        {
            session.close(CloseReason.SERVICE_ACTION);

            if (Cluster.Role.LEADER == role && logPublisher.appendSessionClose(session, clusterTimeMs))
            {
                sessionByIdMap.remove(clusterSessionId);
            }
        }
    }

    void onScheduleTimer(final long correlationId, final long deadlineMs)
    {
        timerService.scheduleTimer(correlationId, deadlineMs);
    }

    void onCancelTimer(final long correlationId)
    {
        timerService.cancelTimer(correlationId);
    }

    void onServiceAck(final long logPosition, final long ackId, final long relevantId, final int serviceId)
    {
        validateServiceAck(logPosition, ackId, serviceId);
        serviceAcks[serviceId].logPosition(logPosition).ackId(ackId).relevantId(relevantId);

        if (hasReachedPosition(logPosition, serviceAckId, serviceAcks))
        {
            switch (state)
            {
                case SNAPSHOT:
                    ++serviceAckId;
                    takeSnapshot(clusterTimeMs, logPosition);
                    state(ConsensusModule.State.ACTIVE);
                    ClusterControl.ToggleState.reset(controlToggle);
                    for (final ClusterSession session : sessionByIdMap.values())
                    {
                        session.timeOfLastActivityMs(clusterTimeMs);
                    }
                    break;

                case SHUTDOWN:
                    takeSnapshot(clusterTimeMs, logPosition);
                    recordingLog.commitLogPosition(leadershipTermId, logPosition);
                    state(ConsensusModule.State.CLOSED);
                    ctx.terminationHook().run();
                    break;

                case ABORT:
                    recordingLog.commitLogPosition(leadershipTermId, logPosition);
                    state(ConsensusModule.State.CLOSED);
                    ctx.terminationHook().run();
                    break;
            }
        }
    }

    @SuppressWarnings("unused")
    void onReplaySessionMessage(
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final DirectBuffer buffer,
        final int offset,
        final int length,
        final Header header)
    {
        clusterTimeMs(timestamp);
        sessionByIdMap.get(clusterSessionId).lastActivity(timestamp, correlationId);
    }

    void onReplayTimerEvent(final long correlationId, final long timestamp)
    {
        clusterTimeMs(timestamp);

        if (!timerService.cancelTimer(correlationId))
        {
            missedTimersSet.add(correlationId);
        }
    }

    void onReplaySessionOpen(
        final long logPosition,
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final int responseStreamId,
        final String responseChannel)
    {
        clusterTimeMs(timestamp);

        final ClusterSession session = new ClusterSession(clusterSessionId, responseStreamId, responseChannel);
        session.open(logPosition);
        session.lastActivity(timestamp, correlationId);

        sessionByIdMap.put(clusterSessionId, session);
        if (clusterSessionId >= nextSessionId)
        {
            nextSessionId = clusterSessionId + 1;
        }
    }

    void onLoadSession(
        final long logPosition,
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final CloseReason closeReason,
        final int responseStreamId,
        final String responseChannel)
    {
        sessionByIdMap.put(clusterSessionId, new ClusterSession(
            clusterSessionId, responseStreamId, responseChannel, logPosition, timestamp, correlationId, closeReason));

        if (clusterSessionId >= nextSessionId)
        {
            nextSessionId = clusterSessionId + 1;
        }
    }

    @SuppressWarnings("unused")
    void onReplaySessionClose(
        final long correlationId, final long clusterSessionId, final long timestamp, final CloseReason closeReason)
    {
        clusterTimeMs(timestamp);
        sessionByIdMap.remove(clusterSessionId).close();
    }

    @SuppressWarnings("unused")
    void onReplayClusterAction(
        final long leadershipTermId, final long logPosition, final long timestamp, final ClusterAction action)
    {
        clusterTimeMs(timestamp);

        switch (action)
        {
            case SUSPEND:
                state(ConsensusModule.State.SUSPENDED);
                break;

            case RESUME:
                state(ConsensusModule.State.ACTIVE);
                break;

            case SNAPSHOT:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.SNAPSHOT);
                break;

            case SHUTDOWN:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.SHUTDOWN);
                break;

            case ABORT:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.ABORT);
                break;
        }
    }

    @SuppressWarnings("unused")
    void onReplayNewLeadershipTermEvent(
        final long leadershipTermId,
        final long logPosition,
        final long timestamp,
        final int leaderMemberId,
        final int logSessionId)
    {
        clusterTimeMs(timestamp);
        this.leadershipTermId = leadershipTermId;

        if (null != election && null != appendedPosition)
        {
            final long recordingId = RecordingPos.getRecordingId(aeron.countersReader(), appendedPosition.counterId());
            election.onReplayNewLeadershipTermEvent(recordingId, leadershipTermId, logPosition, cachedTimeMs);
        }
    }

    void onReloadState(final long nextSessionId)
    {
        this.nextSessionId = nextSessionId;
    }

    Publication addNewLogPublication()
    {
        closeExistingLog();

        final ChannelUri channelUri = ChannelUri.parse(ctx.logChannel());
        final Publication publication = createLogPublication(channelUri, recoveryPlan, election.logPosition());

        logPublisher.connect(publication);

        return publication;
    }

    void becomeLeader(final long leadershipTermId, final int logSessionId)
    {
        this.leadershipTermId = leadershipTermId;

        final ChannelUri channelUri = ChannelUri.parse(ctx.logChannel());
        channelUri.put(CommonContext.SESSION_ID_PARAM_NAME, Integer.toString(logSessionId));
        startLogRecording(channelUri.toString(), SourceLocation.LOCAL);
        createAppendPosition(logSessionId);
        commitPosition = CommitPos.allocate(aeron, tempBuffer, leadershipTermId, election.logPosition(), MAX_VALUE);
        awaitServicesReady(channelUri, logSessionId, election.logPosition());

        final ChannelUri ingressUri = ChannelUri.parse(ctx.ingressChannel());
        if (!ingressUri.containsKey(ENDPOINT_PARAM_NAME))
        {
            ingressUri.put(ENDPOINT_PARAM_NAME, thisMember.clientFacingEndpoint());
        }

        ingressAdapter.subscription(aeron.addSubscription(
            ingressUri.toString(), ctx.ingressStreamId(), null, this::onUnavailableIngressImage));

        for (final ClusterSession session : sessionByIdMap.values())
        {
            if (session.state() != CLOSED)
            {
                session.connect(aeron);
            }
        }<commit_msg>[Java] Add missing update of member details when a member becomes leader in the cluster.<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;

import io.aeron.*;
import io.aeron.archive.client.AeronArchive;
import io.aeron.archive.codecs.SourceLocation;
import io.aeron.archive.status.RecordingPos;
import io.aeron.cluster.client.ClusterException;
import io.aeron.cluster.codecs.*;
import io.aeron.cluster.service.Cluster;
import io.aeron.cluster.service.ClusterMarkFile;
import io.aeron.cluster.service.CommitPos;
import io.aeron.cluster.service.RecoveryState;
import io.aeron.exceptions.TimeoutException;
import io.aeron.logbuffer.ControlledFragmentHandler;
import io.aeron.logbuffer.Header;
import io.aeron.protocol.DataHeaderFlyweight;
import io.aeron.security.Authenticator;
import io.aeron.status.ReadableCounter;
import org.agrona.BitUtil;
import org.agrona.CloseHelper;
import org.agrona.DirectBuffer;
import org.agrona.MutableDirectBuffer;
import org.agrona.collections.ArrayListUtil;
import org.agrona.collections.Long2ObjectHashMap;
import org.agrona.collections.LongHashSet;
import org.agrona.concurrent.Agent;
import org.agrona.concurrent.AgentInvoker;
import org.agrona.concurrent.EpochClock;
import org.agrona.concurrent.IdleStrategy;
import org.agrona.concurrent.status.CountersReader;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

import static io.aeron.Aeron.NULL_VALUE;
import static io.aeron.ChannelUri.SPY_QUALIFIER;
import static io.aeron.CommonContext.*;
import static io.aeron.archive.client.AeronArchive.NULL_LENGTH;
import static io.aeron.archive.codecs.SourceLocation.LOCAL;
import static io.aeron.cluster.ClusterSession.State.*;
import static io.aeron.cluster.ConsensusModule.Configuration.*;
import static io.aeron.cluster.ServiceAck.hasReachedPosition;
import static io.aeron.cluster.ServiceAck.newArray;
import static io.aeron.logbuffer.FrameDescriptor.FRAME_ALIGNMENT;
import static java.lang.Long.MAX_VALUE;

class ConsensusModuleAgent implements Agent, MemberStatusListener
{
    private final int memberId;
    private final long sessionTimeoutMs;
    private final long leaderHeartbeatIntervalMs;
    private final long leaderHeartbeatTimeoutMs;
    private final long serviceHeartbeatTimeoutMs;
    private long nextSessionId = 1;
    private long leadershipTermId = NULL_VALUE;
    private long expectedAckPosition = 0;
    private long serviceAckId = 0;
    private long lastAppendedPosition = 0;
    private long followerCommitPosition = 0;
    private long timeOfLastLogUpdateMs = 0;
    private long cachedTimeMs;
    private long clusterTimeMs = NULL_VALUE;
    private long lastRecordingId = RecordingPos.NULL_RECORDING_ID;
    private int logPublicationInitialTermId = NULL_VALUE;
    private int logPublicationTermBufferLength = NULL_VALUE;
    private int logPublicationMtuLength = NULL_VALUE;
    private ReadableCounter appendedPosition;
    private Counter commitPosition;
    private ConsensusModule.State state = ConsensusModule.State.INIT;
    private Cluster.Role role;
    private ClusterMember[] clusterMembers;
    private ClusterMember leaderMember;
    private final ClusterMember thisMember;
    private long[] rankedPositions;
    private final ServiceAck[] serviceAcks;
    private final Counter clusterRoleCounter;
    private final ClusterMarkFile markFile;
    private final AgentInvoker aeronClientInvoker;
    private final EpochClock epochClock;
    private final Counter moduleState;
    private final Counter controlToggle;
    private final TimerService timerService;
    private final ConsensusModuleAdapter consensusModuleAdapter;
    private final ServiceProxy serviceProxy;
    private final IngressAdapter ingressAdapter;
    private final EgressPublisher egressPublisher;
    private final LogPublisher logPublisher;
    private LogAdapter logAdapter;
    private final MemberStatusAdapter memberStatusAdapter;
    private final MemberStatusPublisher memberStatusPublisher = new MemberStatusPublisher();
    private final Long2ObjectHashMap<ClusterSession> sessionByIdMap = new Long2ObjectHashMap<>();
    private final ArrayList<ClusterSession> pendingSessions = new ArrayList<>();
    private final ArrayList<ClusterSession> rejectedSessions = new ArrayList<>();
    private final LongHashSet missedTimersSet = new LongHashSet();
    private final Authenticator authenticator;
    private final ClusterSessionProxy sessionProxy;
    private final Aeron aeron;
    private AeronArchive archive;
    private final ConsensusModule.Context ctx;
    private final MutableDirectBuffer tempBuffer;
    private final Counter[] serviceHeartbeats;
    private final IdleStrategy idleStrategy;
    private final RecordingLog recordingLog;
    private RecordingLog.RecoveryPlan recoveryPlan;
    private Election election;
    private String logRecordingChannel;
    private String liveLogDestination;
    private String clientFacingEndpoints;

    ConsensusModuleAgent(final ConsensusModule.Context ctx)
    {
        this.ctx = ctx;
        this.aeron = ctx.aeron();
        this.epochClock = ctx.epochClock();
        this.sessionTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.sessionTimeoutNs());
        this.leaderHeartbeatIntervalMs = TimeUnit.NANOSECONDS.toMillis(ctx.leaderHeartbeatIntervalNs());
        this.leaderHeartbeatTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.leaderHeartbeatTimeoutNs());
        this.serviceHeartbeatTimeoutMs = TimeUnit.NANOSECONDS.toMillis(ctx.serviceHeartbeatTimeoutNs());
        this.egressPublisher = ctx.egressPublisher();
        this.moduleState = ctx.moduleStateCounter();
        this.controlToggle = ctx.controlToggleCounter();
        this.logPublisher = ctx.logPublisher();
        this.idleStrategy = ctx.idleStrategy();
        this.timerService = new TimerService(this);
        this.clusterMembers = ClusterMember.parse(ctx.clusterMembers());
        this.sessionProxy = new ClusterSessionProxy(egressPublisher);
        this.memberId = ctx.clusterMemberId();
        this.clusterRoleCounter = ctx.clusterNodeCounter();
        this.markFile = ctx.clusterMarkFile();
        this.recordingLog = ctx.recordingLog();
        this.tempBuffer = ctx.tempBuffer();
        this.serviceHeartbeats = ctx.serviceHeartbeatCounters();
        this.serviceAcks = newArray(ctx.serviceCount());

        aeronClientInvoker = aeron.conductorAgentInvoker();
        aeronClientInvoker.invoke();

        rankedPositions = new long[ClusterMember.quorumThreshold(clusterMembers.length)];
        role(Cluster.Role.FOLLOWER);

        thisMember = clusterMembers[memberId];
        leaderMember = thisMember;
        final ChannelUri memberStatusUri = ChannelUri.parse(ctx.memberStatusChannel());
        memberStatusUri.put(ENDPOINT_PARAM_NAME, thisMember.memberFacingEndpoint());

        final int statusStreamId = ctx.memberStatusStreamId();
        memberStatusAdapter = new MemberStatusAdapter(
            aeron.addSubscription(memberStatusUri.toString(), statusStreamId), this);

        ClusterMember.addMemberStatusPublications(clusterMembers, thisMember, memberStatusUri, statusStreamId, aeron);

        ingressAdapter = new IngressAdapter(this, ctx.invalidRequestCounter());

        final ChannelUri archiveUri = ChannelUri.parse(ctx.archiveContext().controlRequestChannel());
        ClusterMember.checkArchiveEndpoint(thisMember, archiveUri);
        archiveUri.put(ENDPOINT_PARAM_NAME, thisMember.archiveEndpoint());
        ctx.archiveContext().controlRequestChannel(archiveUri.toString());

        consensusModuleAdapter = new ConsensusModuleAdapter(
            aeron.addSubscription(ctx.serviceControlChannel(), ctx.consensusModuleStreamId()), this);
        serviceProxy = new ServiceProxy(
            aeron.addPublication(ctx.serviceControlChannel(), ctx.serviceStreamId()));

        authenticator = ctx.authenticatorSupplier().get();
    }

    public void onClose()
    {
        CloseHelper.close(archive);

        if (!ctx.ownsAeronClient())
        {
            for (final ClusterSession session : sessionByIdMap.values())
            {
                session.close();
            }

            CloseHelper.close(memberStatusAdapter);
            ClusterMember.closeMemberPublications(clusterMembers);

            logPublisher.disconnect();
            CloseHelper.close(ingressAdapter);
            CloseHelper.close(serviceProxy);
            CloseHelper.close(consensusModuleAdapter);
        }
    }

    public void onStart()
    {
        archive = AeronArchive.connect(ctx.archiveContext().clone());
        recoveryPlan = recordingLog.createRecoveryPlan(archive, ctx.serviceCount());

        try (Counter ignore = addRecoveryStateCounter(recoveryPlan))
        {
            if (!recoveryPlan.snapshots.isEmpty())
            {
                recoverFromSnapshot(recoveryPlan.snapshots.get(0), archive);
            }

            awaitServiceAcks(expectedAckPosition);
        }

        if (ConsensusModule.State.SUSPENDED != state)
        {
            state(ConsensusModule.State.ACTIVE);
        }

        timeOfLastLogUpdateMs = cachedTimeMs = epochClock.time();
        leadershipTermId = recoveryPlan.lastLeadershipTermId;

        election = new Election(
            true,
            leadershipTermId,
            recoveryPlan.appendedLogPosition,
            clusterMembers,
            thisMember,
            memberStatusAdapter,
            memberStatusPublisher,
            ctx,
            this);
    }

    public int doWork()
    {
        int workCount = 0;

        boolean isSlowTickCycle = false;
        final long nowMs = epochClock.time();
        if (cachedTimeMs != nowMs)
        {
            cachedTimeMs = nowMs;
            if (Cluster.Role.LEADER == role)
            {
                clusterTimeMs(nowMs);
            }

            isSlowTickCycle = true;
        }

        if (null != election)
        {
            workCount += election.doWork(nowMs);
        }
        else
        {
            if (Cluster.Role.LEADER == role && ConsensusModule.State.ACTIVE == state)
            {
                workCount += ingressAdapter.poll();
                workCount += timerService.poll(nowMs);
            }
            else if (Cluster.Role.FOLLOWER == role &&
                (ConsensusModule.State.ACTIVE == state || ConsensusModule.State.SUSPENDED == state))
            {
                final int count = logAdapter.poll(followerCommitPosition);
                if (0 == count && logAdapter.isImageClosed())
                {
                    enterElection(nowMs);

                    return 1;
                }

                workCount += count;
            }

            workCount += memberStatusAdapter.poll();
            workCount += updateMemberPosition(nowMs);
            workCount += consensusModuleAdapter.poll();
        }

        if (isSlowTickCycle)
        {
            workCount += slowTickCycle(nowMs);
        }

        return workCount;
    }

    public String roleName()
    {
        return "consensus-module";
    }

    public void onSessionConnect(
        final long correlationId,
        final int responseStreamId,
        final String responseChannel,
        final byte[] encodedCredentials)
    {
        final long sessionId = nextSessionId++;
        final ClusterSession session = new ClusterSession(sessionId, responseStreamId, responseChannel);
        session.connect(aeron);
        session.lastActivity(clusterTimeMs, correlationId);

        if (pendingSessions.size() + sessionByIdMap.size() < ctx.maxConcurrentSessions())
        {
            authenticator.onConnectRequest(sessionId, encodedCredentials, clusterTimeMs);
            pendingSessions.add(session);
        }
        else
        {
            rejectedSessions.add(session);
        }
    }

    public void onSessionClose(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session)
        {
            session.close(CloseReason.CLIENT_ACTION);

            if (logPublisher.appendSessionClose(session, clusterTimeMs))
            {
                sessionByIdMap.remove(clusterSessionId);
            }
        }
    }

    public ControlledFragmentAssembler.Action onSessionMessage(
        final DirectBuffer buffer,
        final int offset,
        final int length,
        final long clusterSessionId,
        final long correlationId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null == session || session.state() == CLOSED)
        {
            return ControlledFragmentHandler.Action.CONTINUE;
        }

        if (session.state() == OPEN && logPublisher.appendMessage(buffer, offset, length, clusterTimeMs))
        {
            session.lastActivity(clusterTimeMs, correlationId);
            return ControlledFragmentHandler.Action.CONTINUE;
        }

        return ControlledFragmentHandler.Action.ABORT;
    }

    public void onSessionKeepAlive(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session && session.state() == OPEN)
        {
            session.timeOfLastActivityMs(clusterTimeMs);
        }
    }

    public void onChallengeResponse(
        final long correlationId, final long clusterSessionId, final byte[] encodedCredentials)
    {
        for (int lastIndex = pendingSessions.size() - 1, i = lastIndex; i >= 0; i--)
        {
            final ClusterSession session = pendingSessions.get(i);

            if (session.id() == clusterSessionId && session.state() == CHALLENGED)
            {
                session.lastActivity(clusterTimeMs, correlationId);
                authenticator.onChallengeResponse(clusterSessionId, encodedCredentials, clusterTimeMs);
                break;
            }
        }
    }

    public boolean onTimerEvent(final long correlationId, final long nowMs)
    {
        return Cluster.Role.LEADER != role || logPublisher.appendTimer(correlationId, nowMs);
    }

    public void onCanvassPosition(final long logLeadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (null != election)
        {
            election.onCanvassPosition(logLeadershipTermId, logPosition, followerMemberId);
        }
        else if (Cluster.Role.LEADER == role)
        {
            memberStatusPublisher.newLeadershipTerm(
                clusterMembers[followerMemberId].publication(),
                this.leadershipTermId,
                recordingLog.getTermEntry(this.leadershipTermId).termBaseLogPosition,
                this.leadershipTermId,
                thisMember.id(),
                logPublisher.sessionId());
        }
    }

    public void onRequestVote(
        final long logLeadershipTermId, final long logPosition, final long candidateTermId, final int candidateId)
    {
        if (null != election)
        {
            election.onRequestVote(logLeadershipTermId, logPosition, candidateTermId, candidateId);
        }
        else if (candidateTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
            election.onRequestVote(logLeadershipTermId, logPosition, candidateTermId, candidateId);
        }
    }

    public void onVote(
        final long candidateTermId,
        final long logLeadershipTermId,
        final long logPosition,
        final int candidateMemberId,
        final int followerMemberId,
        final boolean vote)
    {
        if (null != election)
        {
            election.onVote(
                candidateTermId, logLeadershipTermId, logPosition, candidateMemberId, followerMemberId, vote);
        }
    }

    public void onNewLeadershipTerm(
        final long logLeadershipTermId,
        final long logPosition,
        final long leadershipTermId,
        final int leaderMemberId,
        final int logSessionId)
    {
        if (null != election)
        {
            election.onNewLeadershipTerm(
                logLeadershipTermId, logPosition, leadershipTermId, leaderMemberId, logSessionId);
        }
        else if (leadershipTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
        }
    }

    public void onAppendedPosition(final long leadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (null != election)
        {
            election.onAppendedPosition(leadershipTermId, logPosition, followerMemberId);
        }
        else if (Cluster.Role.LEADER == role && leadershipTermId == this.leadershipTermId)
        {
            clusterMembers[followerMemberId].logPosition(logPosition);
        }
    }

    public void onCommitPosition(final long leadershipTermId, final long logPosition, final int leaderMemberId)
    {
        if (null != election)
        {
            election.onCommitPosition(leadershipTermId, logPosition, leaderMemberId);
        }
        else if (Cluster.Role.FOLLOWER == role && leadershipTermId == this.leadershipTermId)
        {
            timeOfLastLogUpdateMs = cachedTimeMs;
            followerCommitPosition = logPosition;
        }
        else if (leadershipTermId > this.leadershipTermId)
        {
            enterElection(cachedTimeMs);
        }
    }

    public void onCatchupPosition(final long leadershipTermId, final long logPosition, final int followerMemberId)
    {
        if (Cluster.Role.LEADER == role && leadershipTermId == this.leadershipTermId)
        {
            final String replayChannel = new ChannelUriStringBuilder()
                .media(CommonContext.UDP_MEDIA)
                .endpoint(clusterMembers[followerMemberId].transferEndpoint())
                .isSessionIdTagged(true)
                .sessionId(ConsensusModule.Configuration.LOG_PUBLICATION_SESSION_ID_TAG)
                .build();

            final ClusterMember member = clusterMembers[followerMemberId];

            if (member.catchupReplaySessionId() == Aeron.NULL_VALUE)
            {
                member.catchupReplaySessionId(archive.startReplay(
                    logRecordingId(), logPosition, Long.MAX_VALUE, replayChannel, ctx.logStreamId()));
            }
        }
    }

    public void onStopCatchup(final int replaySessionId, final int followerMemberId)
    {
        final ClusterMember member = clusterMembers[followerMemberId];

        if (member.catchupReplaySessionId() == replaySessionId)
        {
            archive.stopReplay(member.catchupReplaySessionId());
            member.catchupReplaySessionId(Aeron.NULL_VALUE);
        }
    }

    public void onRecoveryPlanQuery(final long correlationId, final int requestMemberId, final int leaderMemberId)
    {
        if (leaderMemberId == memberId)
        {
            memberStatusPublisher.recoveryPlan(
                clusterMembers[requestMemberId].publication(),
                correlationId,
                requestMemberId,
                leaderMemberId,
                recoveryPlan);
        }
    }

    public void onRecoveryPlan(final RecoveryPlanDecoder recoveryPlanDecoder)
    {
    }

    public void onRecordingLogQuery(
        final long correlationId,
        final int requestMemberId,
        final int leaderMemberId,
        final long fromLeadershipTermId,
        final int count,
        final boolean includeSnapshots)
    {
        if (leaderMemberId == memberId)
        {
            memberStatusPublisher.recordingLog(
                clusterMembers[requestMemberId].publication(),
                correlationId,
                requestMemberId,
                leaderMemberId,
                recordingLog,
                fromLeadershipTermId,
                count,
                includeSnapshots);
        }
    }

    public void onRecordingLog(final RecordingLogDecoder recordingLogDecoder)
    {
    }

    void state(final ConsensusModule.State state)
    {
        this.state = state;
        moduleState.set(state.code());
    }

    void role(final Cluster.Role role)
    {
        this.role = role;
        clusterRoleCounter.setOrdered(role.code());
    }

    Cluster.Role role()
    {
        return role;
    }

    long prepareForElection(final long logPosition)
    {
        final RecordingExtent recordingExtent = new RecordingExtent();

        long recordingId = RecordingPos.getRecordingId(aeron.countersReader(), appendedPosition.counterId());
        if (RecordingPos.NULL_RECORDING_ID == recordingId)
        {
            recordingId = recordingLog.getTermEntry(leadershipTermId).recordingId;
        }

        stopLogRecording();

        idleStrategy.reset();
        while (true)
        {
            if (0 == archive.listRecording(recordingId, recordingExtent))
            {
                throw new ClusterException("recording not found id=" + recordingId);
            }

            if (AeronArchive.NULL_POSITION != recordingExtent.stopPosition)
            {
                break;
            }

            idle();
        }

        if (recordingExtent.stopPosition > logPosition)
        {
            archive.truncateRecording(recordingId, logPosition);
        }

        lastAppendedPosition = recordingExtent.stopPosition;
        followerCommitPosition = recordingExtent.stopPosition;

        lastRecordingId = recordingId;
        logPublicationInitialTermId = recordingExtent.initialTermId;
        logPublicationTermBufferLength = recordingExtent.termBufferLength;
        logPublicationMtuLength = recordingExtent.mtuLength;

        commitPosition.setOrdered(recordingExtent.stopPosition);
        clearSessionsAfter(recordingExtent.stopPosition);

        return recordingExtent.stopPosition;
    }

    void stopLogRecording()
    {
        if (null != logRecordingChannel)
        {
            archive.stopRecording(logRecordingChannel, ctx.logStreamId());
            logRecordingChannel = null;
        }

        if (null != logAdapter && null != liveLogDestination)
        {
            logAdapter.removeDestination(liveLogDestination);
            liveLogDestination = null;
        }
    }

    void appendedPositionCounter(final ReadableCounter appendedPositionCounter)
    {
        this.appendedPosition = appendedPositionCounter;
    }

    void commitPositionCounter(final Counter commitPositionCounter)
    {
        this.commitPosition = commitPositionCounter;
    }

    void clearSessionsAfter(final long logPosition)
    {
        for (final Iterator<ClusterSession> i = sessionByIdMap.values().iterator(); i.hasNext(); )
        {
            final ClusterSession session = i.next();
            if (session.openedLogPosition() >= logPosition)
            {
                i.remove();
                session.close();
            }
        }

        for (final ClusterSession session : pendingSessions)
        {
            session.close();
        }

        pendingSessions.clear();
    }

    void onServiceCloseSession(final long clusterSessionId)
    {
        final ClusterSession session = sessionByIdMap.get(clusterSessionId);
        if (null != session)
        {
            session.close(CloseReason.SERVICE_ACTION);

            if (Cluster.Role.LEADER == role && logPublisher.appendSessionClose(session, clusterTimeMs))
            {
                sessionByIdMap.remove(clusterSessionId);
            }
        }
    }

    void onScheduleTimer(final long correlationId, final long deadlineMs)
    {
        timerService.scheduleTimer(correlationId, deadlineMs);
    }

    void onCancelTimer(final long correlationId)
    {
        timerService.cancelTimer(correlationId);
    }

    void onServiceAck(final long logPosition, final long ackId, final long relevantId, final int serviceId)
    {
        validateServiceAck(logPosition, ackId, serviceId);
        serviceAcks[serviceId].logPosition(logPosition).ackId(ackId).relevantId(relevantId);

        if (hasReachedPosition(logPosition, serviceAckId, serviceAcks))
        {
            switch (state)
            {
                case SNAPSHOT:
                    ++serviceAckId;
                    takeSnapshot(clusterTimeMs, logPosition);
                    state(ConsensusModule.State.ACTIVE);
                    ClusterControl.ToggleState.reset(controlToggle);
                    for (final ClusterSession session : sessionByIdMap.values())
                    {
                        session.timeOfLastActivityMs(clusterTimeMs);
                    }
                    break;

                case SHUTDOWN:
                    takeSnapshot(clusterTimeMs, logPosition);
                    recordingLog.commitLogPosition(leadershipTermId, logPosition);
                    state(ConsensusModule.State.CLOSED);
                    ctx.terminationHook().run();
                    break;

                case ABORT:
                    recordingLog.commitLogPosition(leadershipTermId, logPosition);
                    state(ConsensusModule.State.CLOSED);
                    ctx.terminationHook().run();
                    break;
            }
        }
    }

    @SuppressWarnings("unused")
    void onReplaySessionMessage(
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final DirectBuffer buffer,
        final int offset,
        final int length,
        final Header header)
    {
        clusterTimeMs(timestamp);
        sessionByIdMap.get(clusterSessionId).lastActivity(timestamp, correlationId);
    }

    void onReplayTimerEvent(final long correlationId, final long timestamp)
    {
        clusterTimeMs(timestamp);

        if (!timerService.cancelTimer(correlationId))
        {
            missedTimersSet.add(correlationId);
        }
    }

    void onReplaySessionOpen(
        final long logPosition,
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final int responseStreamId,
        final String responseChannel)
    {
        clusterTimeMs(timestamp);

        final ClusterSession session = new ClusterSession(clusterSessionId, responseStreamId, responseChannel);
        session.open(logPosition);
        session.lastActivity(timestamp, correlationId);

        sessionByIdMap.put(clusterSessionId, session);
        if (clusterSessionId >= nextSessionId)
        {
            nextSessionId = clusterSessionId + 1;
        }
    }

    void onLoadSession(
        final long logPosition,
        final long correlationId,
        final long clusterSessionId,
        final long timestamp,
        final CloseReason closeReason,
        final int responseStreamId,
        final String responseChannel)
    {
        sessionByIdMap.put(clusterSessionId, new ClusterSession(
            clusterSessionId, responseStreamId, responseChannel, logPosition, timestamp, correlationId, closeReason));

        if (clusterSessionId >= nextSessionId)
        {
            nextSessionId = clusterSessionId + 1;
        }
    }

    @SuppressWarnings("unused")
    void onReplaySessionClose(
        final long correlationId, final long clusterSessionId, final long timestamp, final CloseReason closeReason)
    {
        clusterTimeMs(timestamp);
        sessionByIdMap.remove(clusterSessionId).close();
    }

    @SuppressWarnings("unused")
    void onReplayClusterAction(
        final long leadershipTermId, final long logPosition, final long timestamp, final ClusterAction action)
    {
        clusterTimeMs(timestamp);

        switch (action)
        {
            case SUSPEND:
                state(ConsensusModule.State.SUSPENDED);
                break;

            case RESUME:
                state(ConsensusModule.State.ACTIVE);
                break;

            case SNAPSHOT:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.SNAPSHOT);
                break;

            case SHUTDOWN:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.SHUTDOWN);
                break;

            case ABORT:
                replayClusterAction(leadershipTermId, logPosition, ConsensusModule.State.ABORT);
                break;
        }
    }

    @SuppressWarnings("unused")
    void onReplayNewLeadershipTermEvent(
        final long leadershipTermId,
        final long logPosition,
        final long timestamp,
        final int leaderMemberId,
        final int logSessionId)
    {
        clusterTimeMs(timestamp);
        this.leadershipTermId = leadershipTermId;

        if (null != election && null != appendedPosition)
        {
            final long recordingId = RecordingPos.getRecordingId(aeron.countersReader(), appendedPosition.counterId());
            election.onReplayNewLeadershipTermEvent(recordingId, leadershipTermId, logPosition, cachedTimeMs);
        }
    }

    void onReloadState(final long nextSessionId)
    {
        this.nextSessionId = nextSessionId;
    }

    Publication addNewLogPublication()
    {
        closeExistingLog();

        final ChannelUri channelUri = ChannelUri.parse(ctx.logChannel());
        final Publication publication = createLogPublication(channelUri, recoveryPlan, election.logPosition());

        logPublisher.connect(publication);

        return publication;
    }

    void becomeLeader(final long leadershipTermId, final int logSessionId)
    {
        this.leadershipTermId = leadershipTermId;
        updateMemberDetails(election);

        final ChannelUri channelUri = ChannelUri.parse(ctx.logChannel());
        channelUri.put(CommonContext.SESSION_ID_PARAM_NAME, Integer.toString(logSessionId));
        startLogRecording(channelUri.toString(), SourceLocation.LOCAL);
        createAppendPosition(logSessionId);
        commitPosition = CommitPos.allocate(aeron, tempBuffer, leadershipTermId, election.logPosition(), MAX_VALUE);
        awaitServicesReady(channelUri, logSessionId, election.logPosition());

        final ChannelUri ingressUri = ChannelUri.parse(ctx.ingressChannel());
        if (!ingressUri.containsKey(ENDPOINT_PARAM_NAME))
        {
            ingressUri.put(ENDPOINT_PARAM_NAME, thisMember.clientFacingEndpoint());
        }

        ingressAdapter.subscription(aeron.addSubscription(
            ingressUri.toString(), ctx.ingressStreamId(), null, this::onUnavailableIngressImage));

        for (final ClusterSession session : sessionByIdMap.values())
        {
            if (session.state() != CLOSED)
            {
                session.connect(aeron);
            }
        }
{'fraction_non_alphanumeric': 0.07762967721214067, 'fraction_numerical': 0.0009635458487233017, 'mean_word_length': 2.527102803738318, 'pattern_counts': {'":': 0, '<': 19, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1374142', 'n_tokens_mistral': 16265, 'n_tokens_neox': 15572, 'n_words': 3259}