RaftActor might delete committed entries

[xirc]

The following errors happened when some failure-injection tests (like one AZ down):

ERROR   akka.actor.OneForOneStrategy    xxx-akka.actor.internal-dispatcher-4  akka://xxx/system/sharding/raft-shard-xxx-replica-group-2/1/1 -       Term not found at lastApplied: 7317     java.lang.IllegalStateException: Term not found at lastApplied: 7317
    at lerna.akka.entityreplication.raft.RaftMemberData.resolveSnapshotTargets(RaftMemberData.scala:416)
    at lerna.akka.entityreplication.raft.RaftMemberData.resolveSnapshotTargets$(RaftMemberData.scala:405)
    at lerna.akka.entityreplication.raft.RaftMemberDataImpl.resolveSnapshotTargets(RaftMemberData.scala:539)
    at lerna.akka.entityreplication.raft.RaftActor.handleSnapshotTick(RaftActor.scala:496)
    at lerna.akka.entityreplication.raft.Follower$$anonfun$followerBehavior$1.applyOrElse(Follower.scala:35)
    at akka.actor.Actor.aroundReceive(Actor.scala:537)
    at akka.actor.Actor.aroundReceive$(Actor.scala:535)
    at lerna.akka.entityreplication.raft.RaftActor.akka$persistence$Eventsourced$$super$aroundReceive(RaftActor.scala:124)
    at akka.persistence.Eventsourced$$anon$4.stateReceive(Eventsourced.scala:923)
    at akka.persistence.Eventsourced.aroundReceive$$original(Eventsourced.scala:251)
    at akka.persistence.Eventsourced.aroundReceive(Eventsourced.scala:148)
    at akka.persistence.Eventsourced.aroundReceive$(Eventsourced.scala:250)
    at lerna.akka.entityreplication.raft.RaftActor.aroundReceive(RaftActor.scala:124)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
    at akka.actor.ActorCell.invoke$$original(ActorCell.scala:548)
    at akka.actor.ActorCell.invoke(ActorCell.scala:61)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
    at akka.dispatch.Mailbox.run$$original(Mailbox.scala:231)
    at akka.dispatch.Mailbox.run(Mailbox.scala:32)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
    at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)

ERROR   akka.actor.OneForOneStrategy    xxx-akka.actor.internal-dispatcher-14 akka://xxx/system/sharding/raft-shard-xxx-replica-group-3/2/2      -       requirement failed: Cannot select the entries (1-7610) unless RaftActor have applied the entries to the entities (lastApplied: 0)       java.lang.IllegalArgumentException: requirement failed: Cannot select the entries (1-7610) unless RaftActor have applied the entries to the entities (lastApplied: 0)
    at scala.Predef$.require(Predef.scala:337)
    at lerna.akka.entityreplication.raft.RaftMemberData.selectEntityEntries(RaftMemberData.scala:375)
    at lerna.akka.entityreplication.raft.RaftMemberData.selectEntityEntries$(RaftMemberData.scala:368)
    at lerna.akka.entityreplication.raft.RaftMemberDataImpl.selectEntityEntries(RaftMemberData.scala:539)
    at lerna.akka.entityreplication.raft.RaftActor.receiveFetchEntityEvents(RaftActor.scala:164)
    at lerna.akka.entityreplication.raft.Follower$$anonfun$followerBehavior$1.applyOrElse(Follower.scala:32)
    at akka.actor.Actor.aroundReceive(Actor.scala:537)
    at akka.actor.Actor.aroundReceive$(Actor.scala:535)
    at lerna.akka.entityreplication.raft.RaftActor.akka$persistence$Eventsourced$$super$aroundReceive(RaftActor.scala:124)
    at akka.persistence.Eventsourced$$anon$4.stateReceive(Eventsourced.scala:923)
    at akka.persistence.Eventsourced.aroundReceive$$original(Eventsourced.scala:251)
    at akka.persistence.Eventsourced.aroundReceive(Eventsourced.scala:148)
    at akka.persistence.Eventsourced.aroundReceive$(Eventsourced.scala:250)
    at lerna.akka.entityreplication.raft.RaftActor.aroundReceive(RaftActor.scala:124)
    at akka.actor.ActorCell.receiveMessage(ActorCell.scala:580)
    at akka.actor.ActorCell.invoke$$original(ActorCell.scala:548)
    at akka.actor.ActorCell.invoke(ActorCell.scala:61)
    at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:270)
    at akka.dispatch.Mailbox.run$$original(Mailbox.scala:231)
    at akka.dispatch.Mailbox.run(Mailbox.scala:32)
    at akka.dispatch.Mailbox.exec(Mailbox.scala:243)
    at java.base/java.util.concurrent.ForkJoinTask.doExec(ForkJoinTask.java:290)
    at java.base/java.util.concurrent.ForkJoinPool$WorkQueue.topLevelExec(ForkJoinPool.java:1020)
    at java.base/java.util.concurrent.ForkJoinPool.scan(ForkJoinPool.java:1656)
    at java.base/java.util.concurrent.ForkJoinPool.runWorker(ForkJoinPool.java:1594)
    at java.base/java.util.concurrent.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:183)

[xirc]

Related codes are the following:

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala

Lines 368 to 378 in abe3af9

	def selectEntityEntries(
	entityId: NormalizedEntityId,
	from: LogEntryIndex,
	to: LogEntryIndex,
	): Seq[LogEntry] = {
	require(
	to <= lastApplied,
	s"Cannot select the entries (${from}-${to}) unless RaftActor have applied the entries to the entities (lastApplied: ${lastApplied})",
	)
	replicatedLog.sliceEntries(from, to).filter(_.event.entityId.contains(entityId))
	}

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala

Lines 405 to 418 in abe3af9

	def resolveSnapshotTargets(): (Term, LogEntryIndex, Set[NormalizedEntityId]) = {
	replicatedLog.termAt(lastApplied) match {
	case Some(lastAppliedTerm) =>
	val entityIds =
	replicatedLog
	.sliceEntries(lastSnapshotStatus.snapshotLastLogIndex.next(), lastApplied)
	.flatMap(_.event.entityId.toSeq)
	.toSet
	(lastAppliedTerm, lastApplied, entityIds)
	case None =>
	// This exception is not thrown unless there is a bug
	throw new IllegalStateException(s"Term not found at lastApplied: $lastApplied")
	}
	}

[xirc]

We've added some debug logs and diagnosed one RaftActor state change.

xxx:36:20.707 === [Follower] Applying event [FollowedLeaderCommit]. state diff: [commitIndex: 6955 -> 6966, lastApplied: 6955 -> 6966] ===
xxx:36:20.781 === [Follower] Applying event [AppendedEntries]. state diff: [replicatedLog: ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1099 entries with indices Some(5825)...Some(6923)) -> ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1115 entries with indices Some(5825)...Some(6939))] ===
xxx:36:20.781 === [Follower] Applying event [FollowedLeaderCommit]. state diff: - ===
xxx:36:20.781 === [Follower] Applying event [BecameFollower]. state diff: - ===
xxx:36:20.795 === [Follower] Applying event [AppendedEntries]. state diff: [replicatedLog: ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1115 entries with indices Some(5825)...Some(6939)) -> ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1131 entries with indices Some(5825)...Some(6955))] ===
xxx:36:20.795 === [Follower] Applying event [FollowedLeaderCommit]. state diff: - ===
xxx:36:20.795 === [Follower] Applying event [BecameFollower]. state diff: - ===
xxx:36:20.802 === [Follower] Applying event [AppendedEntries]. state diff: [replicatedLog: ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1131 entries with indices Some(5825)...Some(6955)) -> ReplicatedLog(ancestorTerm=Term(2), ancestorIndex=5824, 1019 entries with indices Some(5825)...Some(6843))] ===

As the above logs, RaftActor deleted its log entries (including committed entries).

[xirc]

RaftActor saves entire AppendEntries.entries as AppendedEntries event (not touched, which might truncate existing (committed) entries).

Instead, RaftActor should

• truncate existing entries only if the received entries contain conflict with existing ones
• truncate no entries if the received entries only contain existing ones

or AppendedEntries event replay should handle such things.

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/Follower.scala

Lines 95 to 121 in f77939c

	case appendEntries: AppendEntries =>
	if (currentData.hasMatchLogEntry(appendEntries.prevLogIndex, appendEntries.prevLogTerm)) {
	if (log.isDebugEnabled) log.debug("=== [Follower] append {} ===", appendEntries)
	cancelElectionTimeoutTimer()
	if (appendEntries.entries.isEmpty && appendEntries.term == currentData.currentTerm) {
	// do not persist event when no need
	applyDomainEvent(FollowedLeaderCommit(appendEntries.leader, appendEntries.leaderCommit)) { _ =>
	sender() ! AppendEntriesSucceeded(
	appendEntries.term,
	currentData.replicatedLog.lastLogIndex,
	selfMemberIndex,
	)
	become(Follower)
	}
	} else {
	applyDomainEvent(AppendedEntries(appendEntries.term, appendEntries.entries, appendEntries.prevLogIndex)) {
	domainEvent =>
	applyDomainEvent(FollowedLeaderCommit(appendEntries.leader, appendEntries.leaderCommit)) { _ =>
	sender() ! AppendEntriesSucceeded(
	domainEvent.term,
	currentData.replicatedLog.lastLogIndex,
	selfMemberIndex,
	)
	become(Follower)
	}
	}
	}

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala

Lines 388 to 393 in f77939c

	def hasMatchLogEntry(prevLogIndex: LogEntryIndex, prevLogTerm: Term): Boolean = {
	// リーダーにログが無い場合は LogEntryIndex.initial が送られてくる。
	// そのケースでは AppendEntries が成功したとみなしたいので、
	// prevLogIndex が LogEntryIndex.initial の場合はマッチするログが存在するとみなす
	prevLogIndex == LogEntryIndex.initial() || replicatedLog.termAt(prevLogIndex).contains(prevLogTerm)
	}

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftActor.scala

Lines 210 to 213 in f77939c

	case AppendedEntries(term, logEntries, prevLogIndex) =>
	currentData
	.syncTerm(term)
	.appendEntries(logEntries, prevLogIndex)

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/RaftMemberData.scala

Lines 107 to 111 in f77939c

	def appendEntries(logEntries: Seq[LogEntry], prevLogIndex: LogEntryIndex): RaftMemberData = {
	updatePersistentState(
	replicatedLog = replicatedLog.merge(logEntries, prevLogIndex),
	)
	}

akka-entity-replication/src/main/scala/lerna/akka/entityreplication/raft/model/ReplicatedLog.scala

Lines 82 to 85 in f77939c

	def merge(thatEntries: Seq[LogEntry], prevLogIndex: LogEntryIndex): ReplicatedLog = {
	val newEntries = this.entries.takeWhile(_.index <= prevLogIndex) ++ thatEntries
	copy(newEntries)
	}