fix: R2dbcOffsetStore evict and delete per slice #1255

patriknw · 2024-11-19T15:56:33Z

evict time window for each slice
remove keep-number-of-entries and evict-interval
lazy loading of offsets
delete detached from time window, separate delete-after config

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

patriknw · 2024-11-19T16:01:44Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

@@ -269,6 +291,12 @@ private[projection] class R2dbcOffsetStore(
          system.executionContext))
    else None

+  private def scheduleNextDelete(): Unit = {
+    if (!settings.deleteInterval.isZero && !settings.deleteInterval.isNegative)
+      system.scheduler.scheduleOnce(settings.deleteInterval, () => deleteOldTimestampOffsets(), system.executionContext)


Those scheduled tasks is something I want to revisit in separate PR. Feels wrong that we never stop these scheduled tasks. Probably doesn't do much due to the idle flag but anyway.

👍🏼 good to make sure these are not continuing

patriknw · 2024-11-20T06:36:07Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

        this
+      } else {
+        // this will always keep at least one, latest per slice
+        val until = recordsSortedByTimestamp.last.timestamp.minus(timeWindow)


This idea will not fly. After scaling it may start at an earlier offset from some other slice than what is evicted/deleted .

Good news is that I see where the problems are coming from now. I think we should change to the lazy loading approach of offsets as we did for dynamodb, and decouple deletes from the time window.

incorporated true lazy loading of offsets, in same way as for DynamoDB, and thereby it's no problem to evict too much

patriknw · 2024-11-20T19:17:25Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

@@ -414,6 +451,46 @@ private[projection] class R2dbcOffsetStore(
    }
  }

+  def load(pid: Pid): Future[State] = {


incorporated lazy loading of offsets in same way as for DynamoDB

patriknw · 2024-11-20T19:19:24Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

+    val recordsWithKeyFut =
+      Source(minSlice to maxSlice)
+        .mapAsyncUnordered(offsetSliceReadParallelism) { slice =>
+          dao.readTimestampOffset(slice)


We need to read offsets to find the start offset, and also good to pre-populate the state with latest offsets for each slice. Rest will be lazy loaded when needed.

patriknw · 2024-11-20T19:19:34Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

-      val newState = State(recordsWithKey.map(_.record))
+      val newState = {
+        val s = State(recordsWithKey.map(_.record))
+        // FIXME shall we evict here, or how does that impact the logic for moreThanOneProjectionKey and foreignOffsets?


I think that should be ok, as that starting offset logic is based on latest by slice already.

patriknw · 2024-11-20T19:22:26Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

+    val currentState = getState()
+    if ((triggerDeletion == null || triggerDeletion == TRUE) && currentState.bySliceSorted.contains(slice)) {
+      val latest = currentState.bySliceSorted(slice).last
+      val until = latest.timestamp.minus(settings.deleteAfter)


The deletes are now much later than the time window (in memory). Not sure what would be a good/safe default? Picked one day without much thought.

Kept the deletes to be per slice. Could maybe go back to slice range, but maybe good to have smaller transactions?

patriknw · 2024-11-20T19:26:48Z

...-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/PostgresOffsetStoreDao.scala

-    SELECT projection_key, slice, persistence_id, seq_nr, timestamp_offset
-    FROM $timestampOffsetTable WHERE slice BETWEEN ? AND ? AND projection_name = ?"""
+    SELECT projection_key, persistence_id, seq_nr, timestamp_offset
+    FROM $timestampOffsetTable WHERE slice = ? AND projection_name = ? ORDER BY timestamp_offset DESC LIMIT ?"""


The intention is that this should make use of the primary key

PRIMARY KEY(slice, projection_name, timestamp_offset, persistence_id, seq_nr)

That, and the limit, is the reason for individual query per slice.

patriknw · 2024-11-20T19:32:02Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

-      val evictedNewState =
-        if (newState.size > settings.keepNumberOfEntries && evictThresholdReached && newState.window
-              .compareTo(evictWindow) > 0) {
-          val evictUntil = newState.latestTimestamp.minus(settings.timeWindow)


This could have been a problem after scaling, where latestTimestamp could be far ahead from received events from other slices.

patriknw · 2024-11-20T19:33:05Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

-        // it hasn't filled up the window yet
-        Future.successful(0)
-      } else {
-        val until = currentState.latestTimestamp.minus(settings.timeWindow)


This could have been a problem after scaling, where latestTimestamp could be far ahead from received events from other slices and then deleting too much.

* evict time window for each slice * remove keep-number-of-entries and evict-interval * delete per slice, so that we always keep offsets within time window for each slice, also after projection scaling

* delete much later, still based on latest by slice * delete-after config * increase delete-interval

* read from each slice, desc timestamp and limit

patriknw · 2024-11-26T14:50:26Z

@pvlugter I have a few FIXMEs remaining here, but before completing I'd like your first review

pvlugter

LGTM. Can't think of any new issues from these approaches. But would be useful to test an application with these changes as well.

pvlugter · 2024-11-26T23:22:16Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

+          val newerRecords = recordsSortedByTimestamp.rangeImpl(Some(untilRecord), None) // inclusive of until
+          val olderRecords = recordsSortedByTimestamp.rangeImpl(None, Some(untilRecord)) // exclusive of until


Suggested change

val newerRecords = recordsSortedByTimestamp.rangeImpl(Some(untilRecord), None) // inclusive of until

val olderRecords = recordsSortedByTimestamp.rangeImpl(None, Some(untilRecord)) // exclusive of until

val newerRecords = recordsSortedByTimestamp.rangeFrom(untilRecord) // inclusive of until

val olderRecords = recordsSortedByTimestamp.rangeUntil(untilRecord) // exclusive of until

pvlugter · 2024-11-26T23:23:52Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

@@ -269,6 +291,12 @@ private[projection] class R2dbcOffsetStore(
          system.executionContext))
    else None

+  private def scheduleNextDelete(): Unit = {
+    if (!settings.deleteInterval.isZero && !settings.deleteInterval.isNegative)
+      system.scheduler.scheduleOnce(settings.deleteInterval, () => deleteOldTimestampOffsets(), system.executionContext)


👍🏼 good to make sure these are not continuing

pvlugter · 2024-11-26T23:43:33Z

akka-projection-r2dbc/src/main/scala/akka/projection/r2dbc/internal/R2dbcOffsetStore.scala

-      val newState = State(recordsWithKey.map(_.record))
+      val newState = {
+        val s = State(recordsWithKey.map(_.record))
+        // FIXME shall we evict here, or how does that impact the logic for moreThanOneProjectionKey and foreignOffsets?


I think that should be ok, as that starting offset logic is based on latest by slice already.

patriknw · 2024-11-27T13:35:45Z

Thanks, I'll test this with a real service before we merge

patriknw commented Nov 19, 2024

View reviewed changes

patriknw commented Nov 20, 2024

View reviewed changes

patriknw force-pushed the wip-timestamp-validation-patriknw branch from bb9d82b to fa3ecfc Compare November 25, 2024 08:53

patriknw force-pushed the wip-delete-patriknw branch from 3cfb8e4 to 22c0c26 Compare November 25, 2024 09:16

patriknw mentioned this pull request Nov 25, 2024

fix: [DynamoDB] Ensure that inflight persistence IDs are not evicted #1269

Merged

Base automatically changed from wip-timestamp-validation-patriknw to main November 26, 2024 09:28

patriknw added 7 commits November 26, 2024 15:46

fix: R2dbcOffsetStore evict and delete per slice

be72e02

* evict time window for each slice * remove keep-number-of-entries and evict-interval * delete per slice, so that we always keep offsets within time window for each slice, also after projection scaling

only keep latest per pid

2ff013b

logPrefix

f83baa6

lazy loading of offsets

29568ed

detach delete from evict

0558e3b

* delete much later, still based on latest by slice * delete-after config * increase delete-interval

readTimestampOffset

4b03e0d

* read from each slice, desc timestamp and limit

fix: Ensure that inflight persistence IDs are not evicted

9a60b78

patriknw force-pushed the wip-delete-patriknw branch from 22c0c26 to 9a60b78 Compare November 26, 2024 14:49

patriknw marked this pull request as ready for review November 26, 2024 14:49

pvlugter reviewed Nov 26, 2024

View reviewed changes

patriknw marked this pull request as draft November 27, 2024 13:35

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

fix: R2dbcOffsetStore evict and delete per slice #1255

fix: R2dbcOffsetStore evict and delete per slice #1255

patriknw commented Nov 19, 2024 •

edited

Loading

patriknw Nov 19, 2024

pvlugter Nov 26, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

pvlugter Nov 26, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw Nov 20, 2024

patriknw commented Nov 26, 2024

pvlugter left a comment

pvlugter Nov 26, 2024

pvlugter Nov 26, 2024

pvlugter Nov 26, 2024

patriknw commented Nov 27, 2024

		val newerRecords = recordsSortedByTimestamp.rangeImpl(Some(untilRecord), None) // inclusive of until
		val olderRecords = recordsSortedByTimestamp.rangeImpl(None, Some(untilRecord)) // exclusive of until

fix: R2dbcOffsetStore evict and delete per slice #1255

Are you sure you want to change the base?

fix: R2dbcOffsetStore evict and delete per slice #1255

Conversation

patriknw commented Nov 19, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw commented Nov 26, 2024

pvlugter left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

patriknw commented Nov 27, 2024

patriknw commented Nov 19, 2024 •

edited

Loading