fix: don't trigger heartbeats from idle backtracking (#630)

akka · Jan 9, 2025 · b3f1fca · b3f1fca
1 parent b1c2070
commit b3f1fca
Show file tree

Hide file tree

Showing 2 changed files with 25 additions and 6 deletions.
diff --git a/core/src/main/scala/akka/persistence/r2dbc/internal/BySliceQuery.scala b/core/src/main/scala/akka/persistence/r2dbc/internal/BySliceQuery.scala
@@ -50,7 +50,8 @@ import org.slf4j.Logger
         startTimestamp = Instant.EPOCH,
         startWallClock = Instant.EPOCH,
         currentQueryWallClock = Instant.EPOCH,
-        previousQueryWallClock = Instant.EPOCH)
+        previousQueryWallClock = Instant.EPOCH,
+        idleCountBeforeHeartbeat = 0)
   }
 
   final case class QueryState(
@@ -69,7 +70,8 @@ import org.slf4j.Logger
       startTimestamp: Instant,
       startWallClock: Instant,
       currentQueryWallClock: Instant,
-      previousQueryWallClock: Instant) {
+      previousQueryWallClock: Instant,
+      idleCountBeforeHeartbeat: Long) {
 
     def backtracking: Boolean = backtrackingCount > 0
 
@@ -435,6 +437,10 @@ import org.slf4j.Logger
 
     def nextQuery(state: QueryState): (QueryState, Option[Source[Envelope, NotUsed]]) = {
       val newIdleCount = if (state.rowCount == 0) state.idleCount + 1 else 0
+      val newIdleCountBeforeHeartbeat =
+        if (state.backtracking) state.idleCountBeforeHeartbeat
+        else if (state.rowCount == 0) state.idleCountBeforeHeartbeat + 1
+        else 0
       // only start tracking query wall clock (for heartbeats) after initial backtracking query
       val newQueryWallClock = if (state.latestBacktracking != TimestampOffset.Zero) clock.instant() else Instant.EPOCH
       val newState =
@@ -455,7 +461,8 @@ import org.slf4j.Logger
             latestBacktracking = fromOffset,
             backtrackingExpectFiltered = state.latestBacktrackingSeenCount,
             currentQueryWallClock = newQueryWallClock,
-            previousQueryWallClock = state.currentQueryWallClock)
+            previousQueryWallClock = state.currentQueryWallClock,
+            idleCountBeforeHeartbeat = newIdleCountBeforeHeartbeat)
         } else if (switchFromBacktracking(state)) {
           // switching from backtracking
           state.copy(
@@ -465,7 +472,8 @@ import org.slf4j.Logger
             idleCount = newIdleCount,
             backtrackingCount = 0,
             currentQueryWallClock = newQueryWallClock,
-            previousQueryWallClock = state.currentQueryWallClock)
+            previousQueryWallClock = state.currentQueryWallClock,
+            idleCountBeforeHeartbeat = newIdleCountBeforeHeartbeat)
         } else {
           // continue
           val newBacktrackingCount = if (state.backtracking) state.backtrackingCount + 1 else 0
@@ -477,7 +485,8 @@ import org.slf4j.Logger
             backtrackingCount = newBacktrackingCount,
             backtrackingExpectFiltered = state.latestBacktrackingSeenCount,
             currentQueryWallClock = newQueryWallClock,
-            previousQueryWallClock = state.currentQueryWallClock)
+            previousQueryWallClock = state.currentQueryWallClock,
+            idleCountBeforeHeartbeat = newIdleCountBeforeHeartbeat)
         }
 
       val behindCurrentTime =
@@ -533,7 +542,7 @@ import org.slf4j.Logger
     }
 
     def heartbeat(state: QueryState): Option[Envelope] = {
-      if (state.idleCount >= 1 && state.previousQueryWallClock != Instant.EPOCH) {
+      if (state.idleCountBeforeHeartbeat >= 2 && state.previousQueryWallClock != Instant.EPOCH) {
         // using wall clock to measure duration since the start time (database timestamp) up to idle backtracking limit
         val timestamp = state.startTimestamp.plus(
           JDuration.between(state.startWallClock, state.previousQueryWallClock.minus(backtrackingBehindCurrentTime)))

diff --git a/core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala b/core/src/main/scala/akka/persistence/r2dbc/query/scaladsl/R2dbcReadJournal.scala
@@ -624,6 +624,16 @@ final class R2dbcReadJournal(system: ExtendedActorSystem, config: Config, cfgPat
                     env.sequenceNr)
                   Nil
                 } else {
+                  if (log.isDebugEnabled()) {
+                    if (latestBacktracking.isAfter(t.timestamp))
+                      log.debug(
+                        "Event from query for persistenceId [{}] seqNr [{}] timestamp [{}]" +
+                        " was before latest timestamp from backtracking or heartbeat [{}].",
+                        env.persistenceId,
+                        env.sequenceNr,
+                        t.timestamp,
+                        latestBacktracking)
+                  }
                   env :: Nil
                 }
               case _ =>