termAt doesn't return Term.initial() for LogEntryIndex.initial() after ancestor is changed

[negokaz]

Fixes #105

Locations that uses termAt

publishAppendEntries

  private[this] def publishAppendEntries(): Unit = {
    resetHeartbeatTimeoutTimer()
    otherMemberIndexes.foreach { memberIndex =>
      val nextIndex    = currentData.nextIndexFor(memberIndex)
      val prevLogIndex = nextIndex.prev()
      val prevLogTerm  = currentData.replicatedLog.termAt(prevLogIndex)

akka-entity-replication/Leader.scala at v2.0.0 · lerna-stack/akka-entity-replication

affected part of #105

hasMatchLogEntry

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

akka-entity-replication/RaftMemberData.scala at v2.0.0 · lerna-stack/akka-entity-replication

No impact, because termAt doesn't take prevLogIndex if it is LogEntryIndex.initial().

resolveSnapshotTargets

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

akka-entity-replication/RaftMemberData.scala at v2.0.0 · lerna-stack/akka-entity-replication

No impact, because resoveSnapshotTargets isn't call when lastApplied is LogEntryIndex.initial().
hasLogEntriesThatCanBeCompacted should return false in the situation.

    if (
      currentData.replicatedLog.entries.size >= settings.compactionLogSizeThreshold
      && currentData.hasLogEntriesThatCanBeCompacted
    ) {
      val (term, logEntryIndex, entityIds) = currentData.resolveSnapshotTargets

akka-entity-replication/RaftActor.scala at v2.0.0 · lerna-stack/akka-entity-replication

  def hasLogEntriesThatCanBeCompacted: Boolean = {
    replicatedLog.sliceEntriesFromHead(lastApplied).nonEmpty
  }

akka-entity-replication/RaftMemberData.scala at v2.0.0 · lerna-stack/akka-entity-replication

sliceEntriesFromHead passed following test:

      val logEntries = Seq(
        LogEntry(LogEntryIndex(1), EntityEvent(None, "a"), Term(1)),
        LogEntry(LogEntryIndex(2), EntityEvent(None, "b"), Term(1)),
        LogEntry(LogEntryIndex(3), EntityEvent(None, "c"), Term(1)),
      )

      val log = new ReplicatedLog(logEntries)

      log.sliceEntriesFromHead(to = LogEntryIndex.initial()) should be(empty)

[xirc]

Great 👍 I left some comments.

[xirc]

Great! LGTM 👍