Repository: BlueBrain/nexus (scala)

Issue:
Tagging a resource may fail if the event has not been propagated to the replica
**Describe the bug**
In some cases when tagging a resource, the last events have not yet been propagated to the replica so replaying the events up to the events may miss the latest events and fail.

**Expected behavior**
Replaying the history must be made from the main node where the write operation happened.

Pull request:
Fixes #4948

Patch:
diff --git a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLog.scala b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLog.scala
index eda2440165..b838f649e0 100644
--- a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLog.scala
+++ b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLog.scala
@@ -81,7 +81,7 @@ object ScopedEventLog {
   ): ScopedEventLog[Id, S, Command, E, Rejection] =
     apply(
       definition.tpe,
-      ScopedEventStore(definition.tpe, definition.eventSerializer, config.queryConfig, xas),
+      ScopedEventStore(definition.tpe, definition.eventSerializer, config.queryConfig),
       ScopedStateStore(definition.tpe, definition.stateSerializer, config.queryConfig, xas),
       definition.stateMachine,
       definition.onUniqueViolation,
@@ -127,7 +127,7 @@ object ScopedEventLog {
           notFound: => R,
           invalidRevision: (Int, Int) => R
       ): IO[S] =
-        stateMachine.computeState(eventStore.history(ref, id, rev)).flatMap {
+        stateMachine.computeState(eventStore.history(ref, id, rev).transact(xas.read)).flatMap {
           case Some(s) if s.rev == rev => IO.pure(s)
           case Some(s)                 => IO.raiseError(invalidRevision(rev, s.rev))
           case None                    => IO.raiseError(notFound)
@@ -141,7 +141,7 @@ object ScopedEventLog {
               IO.some(tag -> state)
             case Some((tag, rev))                     =>
               stateMachine
-                .computeState(eventStore.history(ref, id, Some(rev)))
+                .computeState(eventStore.history(ref, id, Some(rev)).transact(xas.write))
                 .flatTap {
                   case stateOpt if !stateOpt.exists(_.rev == rev) =>
                     IO.raiseError(EvaluationTagFailure(command, stateOpt.map(_.rev)))
diff --git a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStore.scala b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStore.scala
index 6c9ac3950e..789a79178e 100644
--- a/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStore.scala
+++ b/delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStore.scala
@@ -1,11 +1,10 @@
 package ch.epfl.bluebrain.nexus.delta.sourcing.event
 
-import cats.effect.IO
 import cats.syntax.all._
 import ch.epfl.bluebrain.nexus.delta.sourcing.config.QueryConfig
 import ch.epfl.bluebrain.nexus.delta.sourcing.event.Event.ScopedEvent
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.{EntityType, ProjectRef}
-import ch.epfl.bluebrain.nexus.delta.sourcing.{Execute, PartitionInit, Serializer, Transactors}
+import ch.epfl.bluebrain.nexus.delta.sourcing.{Execute, PartitionInit, Serializer}
 import doobie._
 import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._
 
@@ -38,17 +37,17 @@ trait ScopedEventStore[Id, E <: ScopedEvent] {
   /**
     * Fetches the history for the event up to the provided revision
     */
-  def history(ref: ProjectRef, id: Id, to: Option[Int]): Stream[IO, E]
+  def history(ref: ProjectRef, id: Id, to: Option[Int]): Stream[ConnectionIO, E]
 
   /**
     * Fetches the history for the event up to the provided revision
     */
-  def history(ref: ProjectRef, id: Id, to: Int): Stream[IO, E] = history(ref, id, Some(to))
+  def history(ref: ProjectRef, id: Id, to: Int): Stream[ConnectionIO, E] = history(ref, id, Some(to))
 
   /**
     * Fetches the history for the global event up to the last existing revision
     */
-  def history(ref: ProjectRef, id: Id): Stream[IO, E] = history(ref, id, None)
+  def history(ref: ProjectRef, id: Id): Stream[ConnectionIO, E] = history(ref, id, None)
 }
 
 object ScopedEventStore {
@@ -56,8 +55,7 @@ object ScopedEventStore {
   def apply[Id, E <: ScopedEvent](
       tpe: EntityType,
       serializer: Serializer[Id, E],
-      config: QueryConfig,
-      xas: Transactors
+      config: QueryConfig
   ): ScopedEventStore[Id, E] =
     new ScopedEventStore[Id, E] {
       implicit val putId: Put[Id]   = serializer.putId
@@ -89,7 +87,7 @@ object ScopedEventStore {
       override def save(event: E, init: PartitionInit): doobie.ConnectionIO[Unit] =
         init.initializePartition("scoped_events") >> insertEvent(event)
 
-      override def history(ref: ProjectRef, id: Id, to: Option[Int]): Stream[IO, E] = {
+      override def history(ref: ProjectRef, id: Id, to: Option[Int]): Stream[ConnectionIO, E] = {
         val select =
           fr"SELECT value FROM scoped_events" ++
             Fragments.whereAndOpt(
@@ -101,7 +99,7 @@ object ScopedEventStore {
             ) ++
             fr"ORDER BY rev"
 
-        select.query[E].streamWithChunkSize(config.batchSize).transact(xas.read)
+        select.query[E].streamWithChunkSize(config.batchSize)
       }
     }
 }


Test patch:
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
index 587f193817..35e9dd90d7 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/PullRequest.scala
@@ -38,8 +38,7 @@ object PullRequest {
     val store = ScopedEventStore[Iri, PullRequestEvent](
       PullRequest.entityType,
       PullRequestEvent.serializer,
-      QueryConfig(10, RefreshStrategy.Stop),
-      xas
+      QueryConfig(10, RefreshStrategy.Stop)
     )
     populateWith.traverse(store.unsafeSave).transact(xas.write).as(store)
   }
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLogSuite.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLogSuite.scala
index 87648fc46f..e6b68ee292 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLogSuite.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLogSuite.scala
@@ -40,8 +40,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
   private lazy val eventStore = ScopedEventStore(
     PullRequest.entityType,
     PullRequestEvent.serializer,
-    queryConfig,
-    xas
+    queryConfig
   )
 
   private lazy val stateStore = ScopedStateStore(
@@ -102,7 +101,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
     val expectedDependencies                        = Set(DependsOn(proj, id2))
     for {
       _        <- eventLog.evaluate(proj, id, Create(id, proj)).assertEquals((opened, state1))
-      _        <- eventStore.history(proj, id).assert(opened)
+      _        <- eventStore.history(proj, id).transact(xas.read).assert(opened)
       _        <- eventLog.stateOr(proj, id, NotFound).assertEquals(state1)
       // Check dependency on id2
       _        <- EntityDependencyStore.directDependencies(proj, id, xas).assertEquals(expectedDependencies)
@@ -127,7 +126,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
   test("Tag and check that the state has also been successfully tagged as well") {
     for {
       _ <- eventLog.evaluate(proj, id, TagPR(id, proj, 2, 1)).assertEquals((tagged, state2))
-      _ <- eventStore.history(proj, id).assert(opened, tagged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state2)
       _ <- eventLog.stateOr(proj, id, tag, NotFound, TagNotFound).assertEquals(state1)
     } yield ()
@@ -141,7 +140,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
   test("Dry run successfully a command without persisting anything") {
     for {
       _ <- eventLog.dryRun(proj, id, Merge(id, proj, 3)).assertEquals((merged, state3))
-      _ <- eventStore.history(proj, id).assert(opened, tagged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state2)
     } yield ()
   }
@@ -149,7 +148,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
   test("Evaluate successfully merge command and store both event and state for an initial state") {
     for {
       _ <- eventLog.evaluate(proj, id, Merge(id, proj, 3)).assertEquals((merged, state3))
-      _ <- eventStore.history(proj, id).assert(opened, tagged, merged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged, merged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state3)
     } yield ()
   }
@@ -169,7 +168,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
   test("Reject a command and persist nothing") {
     for {
       _ <- eventLog.evaluate(proj, id, Update(id, proj, 3)).interceptEquals(PullRequestAlreadyClosed(id, proj))
-      _ <- eventStore.history(proj, id).assert(opened, tagged, merged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged, merged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state3)
     } yield ()
   }
@@ -178,7 +177,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
     val boom = Boom(id, proj, "fail")
     for {
       _ <- eventLog.evaluate(proj, id, boom).intercept[RuntimeException]
-      _ <- eventStore.history(proj, id).assert(opened, tagged, merged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged, merged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state3)
     } yield ()
   }
@@ -187,7 +186,7 @@ class ScopedEventLogSuite extends NexusSuite with Doobie.Fixture {
     val never = Never(id, proj)
     for {
       _ <- eventLog.evaluate(proj, id, never).interceptEquals(EvaluationTimeout(never, maxDuration))
-      _ <- eventStore.history(proj, id).assert(opened, tagged, merged)
+      _ <- eventStore.history(proj, id).transact(xas.read).assert(opened, tagged, merged)
       _ <- eventLog.stateOr(proj, id, NotFound).assertEquals(state3)
     } yield ()
   }
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/EventStreamingSuite.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/EventStreamingSuite.scala
index c11203b853..82e3244e05 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/EventStreamingSuite.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/EventStreamingSuite.scala
@@ -31,15 +31,13 @@ class EventStreamingSuite extends NexusSuite with Doobie.Fixture with Doobie.Ass
   private lazy val gitlabPrStore = ScopedEventStore[Iri, PullRequestEvent](
     PullRequest.entityType,
     PullRequestEvent.serializer,
-    queryConfig,
-    xas
+    queryConfig
   )
 
   private lazy val githubPrStore = ScopedEventStore[Iri, PullRequestEvent](
     EntityType("github"),
     PullRequestEvent.serializer,
-    queryConfig,
-    xas
+    queryConfig
   )
 
   private val id1 = nxv + "id1"
diff --git a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
index 5a8f79ea54..525caf63f0 100644
--- a/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
+++ b/delta/sourcing-psql/src/test/scala/ch/epfl/bluebrain/nexus/delta/sourcing/event/ScopedEventStoreSuite.scala
@@ -51,14 +51,14 @@ class ScopedEventStoreSuite extends NexusSuite with Doobie.Fixture with Doobie.A
   }
 
   test("Fetch all events for a given id") {
-    store.history(project1, id1).assert(event1, event2, event3)
+    store.history(project1, id1).transact(xas.read).assert(event1, event2, event3)
   }
 
   test("Fetch all events for a given id up to revision 2") {
-    store.history(project1, id1, 2).assert(event1, event2)
+    store.history(project1, id1, 2).transact(xas.read).assert(event1, event2)
   }
 
   test("Get an empty stream for an unknown (project, id)") {
-    store.history(project2, id2, 2).assertEmpty
+    store.history(project2, id2, 2).transact(xas.read).assertEmpty
   }
 }