fix: Snapshots as starting point for eventsBySlices, take 2 #717
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,2 @@ | ||
| # internals | ||
| ProblemFilters.exclude[MissingClassProblem]("akka.persistence.r2dbc.internal.postgres.YugabyteSnapshotDao") |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -6,99 +6,212 @@ package akka.persistence.r2dbc.internal | |
|
|
||
| import java.time.Instant | ||
|
|
||
| import scala.concurrent.ExecutionContext | ||
| import scala.concurrent.Future | ||
| import scala.util.Failure | ||
| import scala.util.Success | ||
| import scala.util.Try | ||
|
|
||
| import akka.annotation.InternalApi | ||
| import akka.persistence.query.TimestampOffset | ||
| import akka.persistence.query.typed.EventEnvelope | ||
| import akka.persistence.r2dbc.internal.SnapshotDao.SerializedSnapshotRow | ||
| import akka.stream.Attributes | ||
| import akka.stream.FlowShape | ||
| import akka.stream.Inlet | ||
| import akka.stream.Outlet | ||
| import akka.stream.stage.GraphStage | ||
| import akka.stream.stage.GraphStageLogic | ||
| import akka.stream.stage.InHandler | ||
| import akka.stream.stage.OutHandler | ||
| import akka.util.RecencyList | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[r2dbc] object StartingFromSnapshotStage { | ||
| private case class SnapshotState(seqNr: Long, emitted: Boolean) | ||
| } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| */ | ||
| @InternalApi private[r2dbc] class StartingFromSnapshotStage[Event]( | ||
| cacheCapacity: Int, | ||
| sequenceNumberOfSnapshot: String => Future[Option[Long]], | ||
| loadSnapshot: String => Future[Option[SnapshotDao.SerializedSnapshotRow]], | ||
| createEnvelope: (SerializedSnapshotRow, TimestampOffset) => EventEnvelope[Event], | ||
| heartbeatAfter: Int, | ||
| createHeartbeat: Instant => EventEnvelope[Event]) | ||
| extends GraphStage[FlowShape[EventEnvelope[Event], EventEnvelope[Event]]] { | ||
| import StartingFromSnapshotStage._ | ||
|
|
||
| val in: Inlet[EventEnvelope[Event]] = Inlet("in") | ||
| val out: Outlet[EventEnvelope[Event]] = Outlet("out") | ||
|
|
||
| override val shape: FlowShape[EventEnvelope[Event], EventEnvelope[Event]] = | ||
| FlowShape(in, out) | ||
|
|
||
| override def createLogic(inheritedAttributes: Attributes): GraphStageLogic = | ||
| new GraphStageLogic(shape) with InHandler with OutHandler { self => | ||
| private implicit def ec: ExecutionContext = materializer.executionContext | ||
|
|
||
| private var snapshotState = Map.empty[String, SnapshotState] | ||
| private val recency = RecencyList.emptyWithNanoClock[String] | ||
| private var inProgress = false | ||
|
|
||
| // for emitting heartbeat events | ||
| private var filterCount = 0L | ||
| private var latestTimestamp = Instant.EPOCH | ||
|
|
||
| private def updateState(persistenceId: String, seqNr: Long, emitted: Boolean): Unit = { | ||
| snapshotState = snapshotState.updated(persistenceId, SnapshotState(seqNr, emitted)) | ||
| recency.update(persistenceId) | ||
| if (recency.size > cacheCapacity) | ||
| recency.removeLeastRecent().foreach { pid => | ||
| snapshotState -= pid | ||
| } | ||
| } | ||
|
|
||
| private val seqNrOfSnapshotCallback = getAsyncCallback[Try[(EventEnvelope[Event], Option[Long])]] { | ||
| case Success((env, Some(seqNr))) => | ||
| inProgress = false | ||
| if (env.sequenceNr == seqNr) { | ||
| // snapshot should be emitted, load full snapshot | ||
| loadCorrespondingSnapshot(env) | ||
| } else if (env.sequenceNr > seqNr) { | ||
| // event is ahead of snapshot, emit event | ||
| updateState(env.persistenceId, seqNr, emitted = false) | ||
| push(env) | ||
| } else if (filterCount >= heartbeatAfter) { | ||
| pushHeartbeat() | ||
|
||
| } else { | ||
| // snapshot will be emitted later, ignore event | ||
| updateState(env.persistenceId, seqNr, emitted = false) | ||
| ignore(env) | ||
| } | ||
|
||
|
|
||
| case Success((env, None)) => | ||
| inProgress = false | ||
| // no snapshot, emit event | ||
| updateState(env.persistenceId, 0L, emitted = true) | ||
| push(env) | ||
|
|
||
| case Failure(exc) => | ||
| inProgress = false | ||
| failStage(exc) | ||
| } | ||
|
|
||
| private val loadSnapshotCallback = getAsyncCallback[Try[(EventEnvelope[Event], Option[SerializedSnapshotRow])]] { | ||
| case Success((env, Some(snap))) => | ||
| inProgress = false | ||
| if (env.sequenceNr == snap.seqNr) { | ||
| push(createEnvelope(snap, env.offset.asInstanceOf[TimestampOffset])) | ||
| updateState(snap.persistenceId, snap.seqNr, emitted = true) | ||
| } else if (env.sequenceNr > snap.seqNr) { | ||
| // event is ahead of snapshot, emit event | ||
| updateState(snap.persistenceId, snap.seqNr, emitted = false) | ||
| push(env) | ||
| } else if (filterCount >= heartbeatAfter) { | ||
| pushHeartbeat() | ||
| } else { | ||
| // snapshot will be emitted later, ignore event | ||
| updateState(snap.persistenceId, snap.seqNr, emitted = false) | ||
| ignore(env) | ||
| } | ||
|
|
||
| case Success((env, None)) => | ||
| inProgress = false | ||
| // no snapshot, emit event | ||
| updateState(env.persistenceId, 0L, emitted = true) | ||
| push(env) | ||
|
|
||
| case Failure(exc) => | ||
| inProgress = false | ||
| failStage(exc) | ||
| } | ||
|
|
||
| override def onPush(): Unit = { | ||
| val env = grab(in) | ||
| snapshotState.get(env.persistenceId) match { | ||
| case Some(s) => | ||
| val eventIsAfterSnapshot = env.sequenceNr > s.seqNr | ||
|
|
||
| if (eventIsAfterSnapshot) { | ||
| // event is after snapshot, emit event | ||
| // we can't ignore it when snapshot is not emitted, because it might have been emitted in | ||
| // previous incarnation, but then the stream was restarted | ||
| push(env) | ||
| } else if (!s.emitted && env.sequenceNr == s.seqNr) { | ||
|
Contributor
There was a problem hiding this comment. How is it with event deletion, we only delete up to before sequence number, not the sequence nr of the snapshot? (Or else this will not work together with deletion)
Contributor
Author
There was a problem hiding this comment. We as in Akka runtime/sdk don't delete events, until the whole entity is deleted, and then all events and snapshots are deleted, much later. I'm not sure if snapshotting and retention strategies in EventSourcedBehavior can be setup to delete event seqNr == snapshot seqNr. However, deleting events combined with projections requires considerations anyway, and we have that documented somewhere.
Contributor
There was a problem hiding this comment. Alright, as long as we warn a bit about it somewhere that might be good enough. |
||
| // trigger emit of snapshot | ||
| loadCorrespondingSnapshot(env) | ||
| } else if (filterCount >= heartbeatAfter) { | ||
| pushHeartbeat() | ||
| } else { | ||
| // event is before (or same as) snapshot, ignore | ||
| ignore(env) | ||
| } | ||
|
|
||
| case None => | ||
| seqNrOfCorrespondingSnapshot(env) | ||
| } | ||
| } | ||
|
|
||
| override def onUpstreamFinish(): Unit = { | ||
| if (!inProgress) | ||
| super.onUpstreamFinish() | ||
| } | ||
|
|
||
| private def tryPullOrComplete(): Unit = { | ||
| if (isClosed(in)) | ||
| completeStage() | ||
| else | ||
| pull(in) | ||
| } | ||
|
|
||
| private def push(env: EventEnvelope[Event]): Unit = { | ||
| filterCount = 0L | ||
| push(out, env) | ||
| } | ||
|
|
||
| private def pushHeartbeat(): Unit = { | ||
| filterCount = 1L | ||
| push(out, createHeartbeat(latestTimestamp)) | ||
| } | ||
|
|
||
| private def ignore(env: EventEnvelope[Event]): Unit = { | ||
| filterCount += 1 | ||
| updateLatestTimestamp(env) | ||
| tryPullOrComplete() | ||
| } | ||
|
Comment on lines
+171
to
+179
Member
There was a problem hiding this comment. Could be simpler to remove the private def ignore(env: EventEnvelope[Event]): Unit = {
updateLatestTimestamp(env)
if (filterCount >= heartbeatAfter) {
filterCount = 1L
push(out, createHeartbeat(latestTimestamp))
} else {
filterCount += 1
tryPullOrComplete()
}
} |
||
|
|
||
| private def updateLatestTimestamp(env: EventEnvelope[Event]): Unit = { | ||
| val timestamp = env.offset.asInstanceOf[TimestampOffset].timestamp | ||
| if (timestamp.isAfter(latestTimestamp)) | ||
| latestTimestamp = timestamp | ||
| } | ||
|
|
||
| private def seqNrOfCorrespondingSnapshot(env: EventEnvelope[Event]): Unit = { | ||
| inProgress = true | ||
| sequenceNumberOfSnapshot(env.persistenceId) | ||
| .map(result => (env, result))(ExecutionContext.parasitic) | ||
| .onComplete(seqNrOfSnapshotCallback.invoke) | ||
| } | ||
|
|
||
| private def loadCorrespondingSnapshot(env: EventEnvelope[Event]): Unit = { | ||
| inProgress = true | ||
| loadSnapshot(env.persistenceId) | ||
| .map(result => (env, result))(ExecutionContext.parasitic) | ||
| .onComplete(loadSnapshotCallback.invoke) | ||
| } | ||
|
|
||
| override def onPull(): Unit = { | ||
| tryPullOrComplete() | ||
| } | ||
|
|
||
| setHandler(in, this) | ||
| setHandler(out, this) | ||
|
|
||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.