Validate replica sequence number before updating seenPerReplica

Add defensive validation in handleExternalReplicatedEventPersist to warn
when receiving events with unexpected sequence numbers from replicas.
This catches bugs in external replication transport implementations.

Resolves FIXME referencing akka/akka-core#29259.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: He-Pin

persistence-typed/src/main/scala/org/apache/pekko/persistence/typed/internal/Running.scala

@@ -508,7 +508,26 @@ private[pekko] object Running {
         OptionVal.Some(
           ReplicatedEventMetadata(event.originReplica, event.originSequenceNr, updatedVersion, isConcurrent)))
       val shouldSnapshotAfterPersist = setup.shouldSnapshot(newState2.state, event.event, newState2.seqNr)
-      // FIXME validate this is the correct sequence nr from that replica https://github.com/akka/akka/issues/29259
+
+      // Validate that the sequence number from the replica is the expected next one
+      val expectedSeqNr = newState2.seenPerReplica.getOrElse(event.originReplica, 0L) + 1
+      if (event.originSequenceNr < expectedSeqNr) {
+        setup.internalLogger.warnN(
+          "Received replicated event from [{}] with sequence number [{}] that has already been seen, expected [{}]. " +
+          "This may indicate a bug in the replication transport.",
+          event.originReplica,
+          event.originSequenceNr,
+          expectedSeqNr)
+      } else if (event.originSequenceNr > expectedSeqNr) {
+        setup.internalLogger.warnN(
+          "Received replicated event from [{}] with unexpected sequence number [{}], expected [{}]. " +
+          "Events may have been lost or arrived out of order. " +
+          "This may indicate a bug in the replication transport.",
+          event.originReplica,
+          event.originSequenceNr,
+          expectedSeqNr)
+      }
+
       val updatedSeen = newState2.seenPerReplica.updated(event.originReplica, event.originSequenceNr)
       persistingEvents(
         newState2.copy(seenPerReplica = updatedSeen, version = updatedVersion),