Fix Elasticsearch default views resetter

delta/plugins/elasticsearch/src/main/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/indexing/ElasticSearchDefaultViewsResetter.scala

@@ -12,6 +12,7 @@ import ch.epfl.bluebrain.nexus.delta.plugins.elasticsearch.model.{defaultViewId,
 import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
 import ch.epfl.bluebrain.nexus.delta.sdk.projects.Projects
 import ch.epfl.bluebrain.nexus.delta.sourcing.Transactors
+import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.Identity.Subject
 import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef
 import doobie.implicits._
@@ -92,15 +93,15 @@ object ElasticSearchDefaultViewsResetter {
  override def resetDefaultViews: IO[Unit] =
  resetTrigger.flatMap { triggered =>
  IO.whenA(triggered) {
- views.compile.toList
- .flatMap { _.traverse { resetView } }
- .flatMap { _ => logger.info("Completed resetting default elasticsearch views.") }
+ deleteDatabaseEntries >>
+ views.compile.toList
+ .flatMap { _.traverse { resetView } }
+ .flatMap { _ => logger.info("Completed resetting default elasticsearch views.") }
  }
  }
 
  override def resetView(view: ViewElement): IO[Unit] =
  deleteEsIndex(view) >>
- deleteEventsStatesOffsets(view.project).transact(xas.write) >>
  createDefaultView(view.project)
 
  private def deleteEsIndex(view: ViewElement) =
@@ -109,12 +110,12 @@ object ElasticSearchDefaultViewsResetter {
  case _: DeprecatedViewDef => IO.pure(true)
  }.void
 
- private def deleteEventsStatesOffsets(project: ProjectRef): doobie.ConnectionIO[Unit] =
+ private def deleteDatabaseEntries: IO[Unit] =
  sql"""
- DELETE FROM scoped_events WHERE type = 'elasticsearch' AND id = ${defaultViewId.toString} AND org = ${project.organization} AND project = ${project.project};
- DELETE FROM scoped_states WHERE type = 'elasticsearch' AND id = ${defaultViewId.toString} AND org = ${project.organization} AND project = ${project.project};
- DELETE FROM projection_offsets WHERE module = 'elasticsearch' AND resource_id = ${defaultViewId.toString} AND project = $project;
- """.stripMargin.update.run.void
+ DELETE FROM scoped_events WHERE type = 'elasticsearch' AND id = $defaultViewId;
+ DELETE FROM scoped_states WHERE type = 'elasticsearch' AND id = $defaultViewId;
+ DELETE FROM projection_offsets WHERE module = 'elasticsearch' AND resource_id = $defaultViewId;
+ """.stripMargin.update.run.void.transact(xas.write)
 
  private def createDefaultView(project: ProjectRef): IO[Unit] =
  createView(defaultViewId, project, newViewValue)

delta/plugins/elasticsearch/src/test/scala/ch/epfl/bluebrain/nexus/delta/plugins/elasticsearch/indexing/ElasticSearchDefaultViewsResetterSuite.scala

@@ -11,7 +11,7 @@ import ch.epfl.bluebrain.nexus.delta.rdf.IriOrBNode.Iri
 import ch.epfl.bluebrain.nexus.delta.sourcing.implicits._
 import ch.epfl.bluebrain.nexus.delta.rdf.syntax.iriStringContextSyntax
 import ch.epfl.bluebrain.nexus.delta.sdk.views.{IndexingRev, ViewRef}
-import ch.epfl.bluebrain.nexus.delta.sourcing.model.ProjectRef
+import ch.epfl.bluebrain.nexus.delta.sourcing.model.{Label, ProjectRef}
 import ch.epfl.bluebrain.nexus.delta.sourcing.postgres.Doobie
 import ch.epfl.bluebrain.nexus.delta.sourcing.query.SelectFilter
 import ch.epfl.bluebrain.nexus.testkit.mu.NexusSuite
@@ -46,59 +46,15 @@ class ElasticSearchDefaultViewsResetterSuite
  )
 
  private val customViewId = iri"https://other.id"
- private val customView = defaultView.copy(ref = ViewRef(project, iri"$customViewId"))
 
  private val existingDefaultView = ExistingView(defaultView)
  private val missingDefaultView = MissingView(project3)
- private val existingCustomView = ExistingView(customView)
 
  // TODO: Find how to move this to beforeAll
  test("Setup: partitions should be created") {
  initPartitions(xas, project, project2)
  }
 
- test("The resetter should delete scoped events for an existing default view") {
- insertViewEvent(defaultViewId, project) >>
- assertIO(eventsCount, 1) >>
- resetWithNoViewCreation.resetView(existingDefaultView) >>
- assertIO(eventsCount, 0)
- }
-
- test("The resetter should delete scoped states for an existing default view") {
- insertViewState(defaultViewId, project) >>
- assertIO(statesCount, 1) >>
- resetWithNoViewCreation.resetView(existingDefaultView) >>
- assertIO(statesCount, 0)
- }
-
- test("The resetter should delete projection offsets for an existing default view") {
- insertViewProjectionOffset(defaultViewId, project) >>
- assertIO(projectionOffsetCount, 1) >>
- resetWithNoViewCreation.resetView(existingDefaultView) >>
- assertIO(projectionOffsetCount, 0)
- }
-
- test("The resetter should not delete a scoped event for an existing custom view") {
- insertViewEvent(customViewId, project) >>
- assertIO(eventsCount, 1) >>
- resetWithNoViewCreation.resetView(existingCustomView) >>
- assertIO(eventsCount, 1)
- }
-
- test("The resetter should not delete a scoped state for an existing custom view") {
- insertViewState(customViewId, project) >>
- assertIO(statesCount, 1) >>
- resetWithNoViewCreation.resetView(existingCustomView) >>
- assertIO(statesCount, 1)
- }
-
- test("The resetter should not delete projection offsets for a custom view") {
- insertViewProjectionOffset(customViewId, project) >>
- assertIO(projectionOffsetCount, 1) >>
- resetWithNoViewCreation.resetView(existingCustomView) >>
- assertIO(projectionOffsetCount, 1)
- }
-
  test("The resetter should create a new view") {
  for {
  createdViewRef <- Ref.of[IO, Set[ViewRef]](Set.empty[ViewRef])
@@ -119,11 +75,18 @@ class ElasticSearchDefaultViewsResetterSuite
  for {
  createdViewRef <- Ref.of[IO, Set[ViewRef]](Set.empty[ViewRef])
  _ <- clearDB
- _ <- insertViewEvent(defaultViewId, project)
- _ <- insertViewEvent(defaultViewId, project2)
- _ <- assertIO(eventsCount, 2)
+ _ <- insertView(defaultViewId, project)
+ _ <- insertView(defaultViewId, project2)
+ _ <- insertView(customViewId, project)
+ expectedViews = Set(project -> defaultViewId, project2 -> defaultViewId, project -> customViewId)
+ _ <- assertIO(eventsId, expectedViews)
+ _ <- assertIO(statesId, expectedViews)
+ _ <- assertIO(projectionOffsetIds, expectedViews)
  _ <- resetWithViewCreation(createdViewRef).resetDefaultViews
- _ <- assertIO(eventsCount, 0)
+ remainingCustom = Set(project -> customViewId)
+ _ <- assertIO(eventsId, remainingCustom)
+ _ <- assertIO(statesId, remainingCustom)
+ _ <- assertIO(projectionOffsetIds, remainingCustom)
  expectedCreatedViews = Set(project, project2, project3).map(ViewRef(_, defaultViewId))
  _ <- assertIO(createdViewRef.get, expectedCreatedViews)
  } yield ()
@@ -146,15 +109,6 @@ class ElasticSearchDefaultViewsResetterSuite
  private val viewElem3 = missingDefaultView
  private val viewStream = Stream(viewElem1, viewElem2, viewElem3)
 
- private def resetWithNoViewCreation = ElasticSearchDefaultViewsResetter(
- viewStream,
- _ => IO(true),
- (_, _, _) => IO.unit,
- defaultViewValue,
- IO(true),
- xas
- )
-
  private def resetWithViewCreation(created: Ref[IO, Set[ViewRef]]) =
  ElasticSearchDefaultViewsResetter(
  viewStream,
@@ -175,6 +129,9 @@ class ElasticSearchDefaultViewsResetterSuite
  xas
  )
 
+ private def insertView(id: Iri, projectRef: ProjectRef) =
+ insertViewEvent(id, projectRef) >> insertViewState(id, projectRef) >> insertViewProjectionOffset(id, projectRef)
+
  private def insertViewEvent(id: Iri, projectRef: ProjectRef) =
  sql"""
  INSERT INTO scoped_events (type, org, project, id, rev, value, instant)
@@ -187,33 +144,37 @@ class ElasticSearchDefaultViewsResetterSuite
  VALUES ('elasticsearch', ${projectRef.organization}, ${projectRef.project}, $id, 'tag', 5, '{"nb": 1}', false, CURRENT_TIMESTAMP);
  """.stripMargin.update.run.transact(xas.write)
 
- private def insertViewProjectionOffset(id: Iri, projectRef: ProjectRef) =
+ private def insertViewProjectionOffset(id: Iri, projectRef: ProjectRef) = {
+ val projectionName = s"${projectRef}_$id"
  sql"""
  INSERT INTO projection_offsets (name, module, project, resource_id, ordering, processed, discarded, failed, created_at, updated_at)
- VALUES ('default', 'elasticsearch', $projectRef, $id, 123, 2, 1, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP);
+ VALUES ($projectionName, 'elasticsearch', $projectRef, $id, 123, 2, 1, 0, CURRENT_TIMESTAMP, CURRENT_TIMESTAMP);
  """.stripMargin.update.run.transact(xas.write)
+ }
 
  private def clearDB =
  sql"""
  DELETE FROM scoped_events; DELETE FROM scoped_states; DELETE FROM projection_offsets;
  """.stripMargin.update.run.transact(xas.write)
 
- private def eventsCount =
- sql"""SELECT count(*) FROM scoped_events;""".stripMargin
- .query[Int]
- .unique
+ private def eventsId: IO[Set[(ProjectRef, Iri)]] =
+ sql"""SELECT org, project, id FROM scoped_events;""".stripMargin
+ .query[(Label, Label, Iri)]
+ .map { case (org, proj, id) => ProjectRef(org, proj) -> id }
+ .to[Set]
  .transact(xas.read)
 
- private def statesCount =
- sql"""SELECT count(*) FROM scoped_states;""".stripMargin
- .query[Int]
- .unique
+ private def statesId: IO[Set[(ProjectRef, Iri)]] =
+ sql"""SELECT org, project, id FROM scoped_states;""".stripMargin
+ .query[(Label, Label, Iri)]
+ .map { case (org, proj, id) => ProjectRef(org, proj) -> id }
+ .to[Set]
  .transact(xas.read)
 
- private def projectionOffsetCount =
- sql"""SELECT count(*) FROM projection_offsets;""".stripMargin
- .query[Int]
- .unique
+ private def projectionOffsetIds: IO[Set[(ProjectRef, Iri)]] =
+ sql"""SELECT project, resource_id FROM projection_offsets;""".stripMargin
+ .query[(ProjectRef, Iri)]
+ .to[Set]
  .transact(xas.read)
 
 }

delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/GlobalEventLog.scala

@@ -134,7 +134,7 @@ object GlobalEventLog {
  }
 
  override def evaluate(id: Id, command: Command): IO[(E, S)] =
- stateStore.get(id).flatMap { current =>
+ stateStore.getWrite(id).flatMap { current =>
  stateMachine
  .evaluate(current, command, maxDuration)
  .flatTap { case (event, state) =>
@@ -148,7 +148,7 @@ object GlobalEventLog {
  }
 
  override def dryRun(id: Id, command: Command): IO[(E, S)] =
- stateStore.get(id).flatMap { current =>
+ stateStore.getWrite(id).flatMap { current =>
  stateMachine.evaluate(current, command, maxDuration)
  }
 

delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/ScopedEventLog.scala

@@ -288,7 +288,7 @@ object ScopedEventLog {
  }
 
  for {
- originalState <- stateStore.get(ref, id).redeem(_ => None, Some(_))
+ originalState <- stateStore.getWrite(ref, id).redeem(_ => None, Some(_))
  result <- stateMachine.evaluate(originalState, command, maxDuration)
  _ <- persist(result._1, originalState, result._2)
  } yield result
@@ -298,7 +298,7 @@ object ScopedEventLog {
  sql.getSQLState == sqlstate.class23.UNIQUE_VIOLATION.value
 
  override def dryRun(ref: ProjectRef, id: Id, command: Command): IO[(E, S)] =
- stateStore.get(ref, id).redeem(_ => None, Some(_)).flatMap { state =>
+ stateStore.getWrite(ref, id).redeem(_ => None, Some(_)).flatMap { state =>
  stateMachine.evaluate(state, command, maxDuration)
  }
 

delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/state/GlobalStateStore.scala

@@ -31,6 +31,12 @@ trait GlobalStateStore[Id, S <: GlobalState] {
  */
  def delete(id: Id): ConnectionIO[Unit]
 
+ /**
+ * Returns the latest state from the write nodes to get a stronger consistency when the Postgres works in a
+ * replicated fashion
+ */
+ def getWrite(id: Id): IO[Option[S]]
+
  /**
  * Returns the state
  */
@@ -102,7 +108,11 @@ object GlobalStateStore {
  override def delete(id: Id): ConnectionIO[Unit] =
  sql"""DELETE FROM global_states WHERE type = $tpe AND id = $id""".stripMargin.update.run.void
 
- override def get(id: Id): IO[Option[S]] = GlobalStateGet[Id, S](tpe, id).transact(xas.read)
+ override def getWrite(id: Id): IO[Option[S]] = get(id, xas.write)
+
+ override def get(id: Id): IO[Option[S]] = get(id, xas.read)
+
+ private def get(id: Id, xa: Transactor[IO]) = GlobalStateGet[Id, S](tpe, id).transact(xa)
 
  private def states(offset: Offset, strategy: RefreshStrategy): Stream[IO, S] =
  StreamingQuery[(S, Long)](

delta/sourcing-psql/src/main/scala/ch/epfl/bluebrain/nexus/delta/sourcing/state/ScopedStateStore.scala

@@ -60,6 +60,12 @@ trait ScopedStateStore[Id, S <: ScopedState] {
  */
  def delete(ref: ProjectRef, id: Id, tag: Tag): ConnectionIO[Unit]
 
+ /**
+ * Returns the latest state from the write nodes to get a stronger consistency when the Postgres works in a
+ * replicated fashion
+ */
+ def getWrite(ref: ProjectRef, id: Id): IO[S]
+
  /**
  * Returns the latest state
  */
@@ -208,8 +214,12 @@ object ScopedStateStore {
  .option
  .map(_.isDefined)
 
- override def get(ref: ProjectRef, id: Id): IO[S] =
- ScopedStateGet.latest[Id, S](tpe, ref, id).transact(xas.read).flatMap { s =>
+ override def getWrite(ref: ProjectRef, id: Id): IO[S] = get(ref, id, xas.write)
+
+ override def get(ref: ProjectRef, id: Id): IO[S] = get(ref, id, xas.read)
+
+ private def get(ref: ProjectRef, id: Id, xa: Transactor[IO]) =
+ ScopedStateGet.latest[Id, S](tpe, ref, id).transact(xa).flatMap { s =>
  IO.fromOption(s)(UnknownState)
  }