fix: lastSequenceNumber for initial stashed command (#32657)

* place currentSequenceNumber in mutable field in BehaviorSetup
* similar for currentMetadata

Author: patriknw

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/scaladsl/EventSourcedMetadataSpec.scala

@@ -109,14 +109,19 @@ class EventSourcedMetadataSpec
 
       // and during replay
       val ref2 = spawn(behavior(PersistenceId.ofUniqueId("ess-1"), probe.ref))
+
+      // This command will be handled after the replay.
+      // It will most likely be stashed during the replay and it should see seqNr 5 when handled.
+      // Reproducer of issue #32651
+      ref2 ! "cmd"
+
       probe.expectMessage("Some(meta) eventHandler evt")
       probe.expectMessage("Some(meta) eventHandler evt")
       probe.expectMessage("Some(meta-1) eventHandler evt1")
       probe.expectMessage("Some(meta-2) eventHandler evt2")
       probe.expectMessage("Some(meta-3) eventHandler evt3")
       probe.expectMessage("Some(meta-3) RecoveryCompleted")
 
-      ref2 ! "cmd"
       probe.expectMessage("None onCommand")
       probe.expectMessage("Some(meta) eventHandler evt")
       probe.expectMessage("None thenRun")

akka-persistence-typed-tests/src/test/scala/akka/persistence/typed/scaladsl/EventSourcedSequenceNumberSpec.scala

@@ -102,14 +102,19 @@ class EventSourcedSequenceNumberSpec
 
       // and during replay
       val ref2 = spawn(behavior(PersistenceId.ofUniqueId("ess-1"), probe.ref))
+
+      // This command will be handled after the replay.
+      // It will most likely be stashed during the replay and it should see seqNr 5 when handled.
+      // Reproducer of issue #32651
+      ref2 ! "cmd"
+
       probe.expectMessage("1 eventHandler evt")
       probe.expectMessage("2 eventHandler evt")
       probe.expectMessage("3 eventHandler evt1")
       probe.expectMessage("4 eventHandler evt2")
       probe.expectMessage("5 eventHandler evt3")
       probe.expectMessage("5 onRecoveryComplete")
 
-      ref2 ! "cmd"
       probe.expectMessage("5 onCommand")
       probe.expectMessage("6 eventHandler evt")
       probe.expectMessage("6 thenRun")

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/BehaviorSetup.scala

@@ -102,6 +102,12 @@ private[akka] final class BehaviorSetup[C, E, S](
 
   private var mdcPhase = PersistenceMdc.Initializing
 
+  // Needed for WithSeqNrAccessible
+  var currentSequenceNumber = 0L
+
+  // Needed for WithMetadataAccessible
+  var currentMetadata: Option[Any] = None
+
   if (isOnlyOneSnapshot) {
     retention match {
       case SnapshotCountRetentionCriteriaImpl(_, keepNSnapshots, _) if keepNSnapshots > 1 =>

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/ReplayingEvents.scala

@@ -65,8 +65,7 @@ private[akka] object ReplayingEvents {
       recoveryStartTime: Long,
       version: VersionVector,
       seenSeqNrPerReplica: Map[ReplicaId, Long],
-      eventsReplayed: Long,
-      metadata: Option[Any])
+      eventsReplayed: Long)
 
   def apply[C, E, S](setup: BehaviorSetup[C, E, S], state: ReplayingState[S]): Behavior[InternalProtocol] =
     Behaviors.setup { _ =>
@@ -142,8 +141,9 @@ private[akka] final class ReplayingEvents[C, E, S](
 
           def handleEvent(event: E): Unit = {
             eventForErrorReporting = OptionVal.Some(event)
-            state =
-              state.copy(seqNr = repr.sequenceNr, eventsReplayed = state.eventsReplayed + 1, metadata = repr.metadata)
+            state = state.copy(seqNr = repr.sequenceNr, eventsReplayed = state.eventsReplayed + 1)
+            setup.currentSequenceNumber = repr.sequenceNr
+            setup.currentMetadata = repr.metadata
 
             val replicatedMetaAndSelfReplica: Option[(ReplicatedEventMetadata, ReplicaId, ReplicationSetup)] =
               setup.replication match {
@@ -196,14 +196,17 @@ private[akka] final class ReplayingEvents[C, E, S](
           this
         } catch {
           case NonFatal(ex) =>
-            state = state.copy(seqNr = repr.sequenceNr, metadata = repr.metadata)
+            state = state.copy(seqNr = repr.sequenceNr)
+            setup.currentSequenceNumber = repr.sequenceNr
+            setup.currentMetadata = repr.metadata
             onRecoveryFailure(ex, eventForErrorReporting.toOption, "replaying-event")
         }
 
       case RecoverySuccess(highestJournalSeqNr) =>
         try {
           val highestSeqNr = Math.max(highestJournalSeqNr, state.seqNr)
           state = state.copy(seqNr = highestSeqNr)
+          setup.currentSequenceNumber = highestSeqNr
           setup.internalLogger.debug("Recovery successful, recovered until sequenceNr: [{}]", highestSeqNr)
           onRecoveryCompleted(state)
         } catch {
@@ -330,7 +333,7 @@ private[akka] final class ReplayingEvents[C, E, S](
         }
         setup.retention match {
           case criteria: SnapshotCountRetentionCriteriaImpl if criteria.snapshotEveryNEvents <= state.eventsReplayed =>
-            internalSaveSnapshot(initialRunningState, state.metadata)
+            internalSaveSnapshot(initialRunningState, setup.currentMetadata)
             new running.StoringSnapshot(initialRunningState, immutable.Seq.empty, SnapshotWithoutRetention)
           case _ =>
             tryUnstashOne(new running.HandlingCommands(initialRunningState))
@@ -342,10 +345,10 @@ private[akka] final class ReplayingEvents[C, E, S](
 
   // WithSeqNrAccessible
   override def currentSequenceNumber: Long =
-    state.seqNr
+    setup.currentSequenceNumber
 
   // WithMetadataAccessible
   override def metadata[M: ClassTag]: Option[M] =
-    CompositeMetadata.extract[M](state.metadata)
+    CompositeMetadata.extract[M](setup.currentMetadata)
 
 }

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/ReplayingSnapshot.scala

@@ -171,6 +171,9 @@ private[akka] class ReplayingSnapshot[C, E, S](override val setup: BehaviorSetup
 
       setup.cancelRecoveryTimer()
 
+      setup.currentSequenceNumber = seqNr
+      setup.currentMetadata = metadata
+
       ReplayingEvents[C, E, S](
         setup,
         ReplayingEvents.ReplayingState(
@@ -182,8 +185,7 @@ private[akka] class ReplayingSnapshot[C, E, S](override val setup: BehaviorSetup
           System.nanoTime(),
           version,
           seenPerReplica,
-          eventsReplayed = 0,
-          metadata = metadata))
+          eventsReplayed = 0))
     }
 
     response match {

akka-persistence-typed/src/main/scala/akka/persistence/typed/internal/Running.scala

@@ -261,30 +261,24 @@ private[akka] object Running {
   import Running.RunningState
   import Running.formatTimestamp
 
-  // Needed for WithSeqNrAccessible, when unstashing
-  private var _currentSequenceNumber = 0L
-
-  // Needed for WithMetadataAccessible
-  private var _currentMetadata: Option[Any] = None
-
   private var recursiveUnstashOne = 0
 
   private def updateMetadata(metadataEntries: Seq[Any]): Unit = {
     if (metadataEntries.isEmpty)
-      _currentMetadata = None
+      setup.currentMetadata = None
     else if (metadataEntries.size == 1)
-      _currentMetadata = Some(metadataEntries.head)
+      setup.currentMetadata = Some(metadataEntries.head)
     else
-      _currentMetadata = Some(CompositeMetadata(metadataEntries))
+      setup.currentMetadata = Some(CompositeMetadata(metadataEntries))
   }
 
   final class HandlingCommands(state: RunningState[S])
       extends AbstractBehavior[InternalProtocol](setup.context)
       with WithSeqNrAccessible
       with WithMetadataAccessible {
 
-    _currentSequenceNumber = state.seqNr
-    _currentMetadata = None
+    setup.currentSequenceNumber = state.seqNr
+    setup.currentMetadata = None
 
     private def alreadySeen(e: ReplicatedEvent[_]): Boolean = {
       e.originSequenceNr <= state.seenPerReplica.getOrElse(e.originReplica, 0L)
@@ -502,8 +496,8 @@ private[akka] object Running {
         replication: ReplicationSetup,
         event: ReplicatedEvent[E],
         ackToOnPersisted: Option[ActorRef[Done]]): Behavior[InternalProtocol] = {
-      _currentSequenceNumber = state.seqNr + 1
-      _currentMetadata = event.metadata
+      setup.currentSequenceNumber = state.seqNr + 1
+      setup.currentMetadata = event.metadata
       val isConcurrent: Boolean = event.originVersion <> state.version
       val updatedVersion = event.originVersion.merge(state.version)
 
@@ -569,7 +563,7 @@ private[akka] object Running {
         // apply the event before persist so that validation exception is handled before persisting
         // the invalid event, in case such validation is implemented in the event handler.
         // also, ensure that there is an event handler for each single event
-        _currentSequenceNumber = state.seqNr + 1
+        setup.currentSequenceNumber = state.seqNr + 1
         updateMetadata(metadataEntries)
 
         setup.replication.foreach(r => r.setContext(recoveryRunning = false, r.replicaId, concurrent = false))
@@ -580,9 +574,13 @@ private[akka] object Running {
 
         val newState2 = setup.replication match {
           case Some(replication) =>
-            val updatedVersion = stateAfterApply.version.updated(replication.replicaId.id, _currentSequenceNumber)
+            val updatedVersion = stateAfterApply.version.updated(replication.replicaId.id, setup.currentSequenceNumber)
             val replicatedEventMetadata =
-              ReplicatedEventMetadata(replication.replicaId, _currentSequenceNumber, updatedVersion, concurrent = false)
+              ReplicatedEventMetadata(
+                replication.replicaId,
+                setup.currentSequenceNumber,
+                updatedVersion,
+                concurrent = false)
             val r = internalPersist(
               OptionVal.Some(cmd),
               stateAfterApply,
@@ -626,7 +624,7 @@ private[akka] object Running {
           // apply the event before persist so that validation exception is handled before persisting
           // the invalid event, in case such validation is implemented in the event handler.
           // also, ensure that there is an event handler for each single event
-          _currentSequenceNumber = state.seqNr
+          setup.currentSequenceNumber = state.seqNr
 
           val replicatedEventMetadataTemplate: Option[ReplicatedEventMetadata] = setup.replication match {
             case Some(replication) =>
@@ -641,25 +639,26 @@ private[akka] object Running {
 
           eventsWithMetadata.foreach { evtWithMeta =>
             val event = evtWithMeta.event
-            _currentSequenceNumber += 1
+            setup.currentSequenceNumber += 1
             updateMetadata(evtWithMeta.metadataEntries)
             val evtManifest = setup.eventAdapter.manifest(event)
             val eventMetadata = replicatedEventMetadataTemplate match {
               case Some(template) =>
-                val updatedVersion = currentState.version.updated(template.originReplica.id, _currentSequenceNumber)
+                val updatedVersion =
+                  currentState.version.updated(template.originReplica.id, setup.currentSequenceNumber)
                 if (setup.internalLogger.isDebugEnabled)
                   setup.internalLogger.trace(
                     "Processing event [{}] with version vector [{}]",
                     Logging.simpleName(event.getClass),
                     updatedVersion)
                 currentState = currentState.copy(version = updatedVersion)
-                template.copy(originSequenceNr = _currentSequenceNumber, version = updatedVersion) +: evtWithMeta.metadataEntries
+                template.copy(originSequenceNr = setup.currentSequenceNumber, version = updatedVersion) +: evtWithMeta.metadataEntries
               case None => evtWithMeta.metadataEntries
             }
 
             currentState = currentState.applyEvent(setup, event)
             if (shouldSnapshotAfterPersist == NoSnapshot)
-              shouldSnapshotAfterPersist = setup.shouldSnapshot(currentState.state, event, _currentSequenceNumber)
+              shouldSnapshotAfterPersist = setup.shouldSnapshot(currentState.state, event, setup.currentSequenceNumber)
 
             val adaptedEvent = adaptEvent(currentState.state, event)
 
@@ -684,7 +683,7 @@ private[akka] object Running {
         }
       } else {
         // run side-effects even when no events are emitted
-        _currentMetadata = None
+        setup.currentMetadata = None
         (applySideEffects(sideEffects, state), true)
       }
     }
@@ -761,11 +760,11 @@ private[akka] object Running {
 
     // WithSeqNrAccessible
     override def currentSequenceNumber: Long =
-      _currentSequenceNumber
+      setup.currentSequenceNumber
 
     // WithMetadataAccessible
     override def metadata[M: ClassTag]: Option[M] =
-      CompositeMetadata.extract[M](_currentMetadata)
+      CompositeMetadata.extract[M](setup.currentMetadata)
 
   }
 
@@ -910,7 +909,7 @@ private[akka] object Running {
             onWriteDone(setup.context, p)
             if (shouldSnapshotAfterPersist == SnapshotWithRetention)
               setup.retentionProgressSaveSnapshotStarted(state2.seqNr)
-            internalSaveSnapshot(state2, _currentMetadata)
+            internalSaveSnapshot(state2, setup.currentMetadata)
             new StoringSnapshot(state2.clearInstrumentationContext, sideEffects, shouldSnapshotAfterPersist)
           }
         }
@@ -979,11 +978,11 @@ private[akka] object Running {
 
     // WithSeqNrAccessible
     override def currentSequenceNumber: Long =
-      _currentSequenceNumber
+      setup.currentSequenceNumber
 
     // WithMetadataAccessible
     override def metadata[M: ClassTag]: Option[M] =
-      CompositeMetadata.extract[M](_currentMetadata)
+      CompositeMetadata.extract[M](setup.currentMetadata)
   }
 
   // ===============================================
@@ -1072,11 +1071,11 @@ private[akka] object Running {
 
     // WithSeqNrAccessible
     override def currentSequenceNumber: Long =
-      _currentSequenceNumber
+      setup.currentSequenceNumber
 
     // WithMetadataAccessible
     override def metadata[M: ClassTag]: Option[M] =
-      CompositeMetadata.extract[M](_currentMetadata)
+      CompositeMetadata.extract[M](setup.currentMetadata)
   }
 
   // ===============================================
@@ -1240,11 +1239,11 @@ private[akka] object Running {
 
     // WithSeqNrAccessible
     override def currentSequenceNumber: Long =
-      _currentSequenceNumber
+      setup.currentSequenceNumber
 
     // WithMetadataAccessible
     override def metadata[M: ClassTag]: Option[M] =
-      CompositeMetadata.extract[M](_currentMetadata)
+      CompositeMetadata.extract[M](setup.currentMetadata)
   }
 
   // ===============================================
@@ -1291,11 +1290,11 @@ private[akka] object Running {
 
     // WithSeqNrAccessible
     override def currentSequenceNumber: Long =
-      _currentSequenceNumber
+      setup.currentSequenceNumber
 
     // WithMetadataAccessible
     override def metadata[M: ClassTag]: Option[M] =
-      CompositeMetadata.extract[M](_currentMetadata)
+      CompositeMetadata.extract[M](setup.currentMetadata)
   }
 
   // --------------------------