Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Provide API that allow user defined action in case of recovery failure #351

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

Provide API that allow user defined action in case of recovery failure #351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
4e9ad3f
Provide API that allow user defined action in case of recovery failure
Nov 19, 2024
60dc363
scalafix
Nov 21, 2024
c356ab1
pass DeleteTo in recovery failure instead of Journaller
Nov 21, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
145 changes: 83 additions & 62 deletions ...tence/src/main/scala/com/evolutiongaming/akkaeffect/persistence/EventSourcedActorOf.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,33 +91,47 @@ object EventSourcedActorOf {

replay = recovering.replay

seqNr <- replay.use { replay =>
// used to recover snapshot, i.e. the snapshot stored with [[snapSeqNr]] will be loaded, if any
val snapSeqNr = snapshot.map(_.metadata.seqNr).getOrElse(SeqNr.Min)
// used to recover events _following_ the snapshot OR if no snapshot available then [[SeqNr.Min]]
val fromSeqNr = snapshot.map(_.metadata.seqNr + 1).getOrElse(SeqNr.Min)
for {
_ <- log.debug(s"snapshot seqNr: $snapSeqNr, load events from seqNr: $fromSeqNr").allocated
events <- eventStore.events(fromSeqNr)
seqNrL <- events.foldWhileM(snapSeqNr) {
case (_, EventStore.Event(event, seqNr)) => replay(event, seqNr).as(seqNr.asLeft[Unit])
case (_, EventStore.HighestSeqNr(seqNr)) => seqNr.asLeft[Unit].pure[F]
}
seqNr <- seqNrL match {
case Left(seqNr) => seqNr.pure[F]
case Right(_) =>
// function used in events.foldWhileM always returns Left
// and sstream.Stream.foldWhileM returns Right only if passed function do so
// thus makes getting Right here impossible
new IllegalStateException("should never happened").raiseError[F, SeqNr]
}
} yield seqNr
}.toResource

_ <- log.debug(s"recovery completed with seqNr $seqNr")
journaller <- eventStore.asJournaller(actorCtx, seqNr).toResource
context = Recovering.RecoveryContext(seqNr, journaller, snapshotStore.asSnapshotter)
receive <- recovering.completed(context)
seqNr <- replay
.use { replay =>
// used to recover snapshot, i.e. the snapshot stored with [[snapSeqNr]] will be loaded, if any
val snapSeqNr = snapshot.map(_.metadata.seqNr).getOrElse(SeqNr.Min)
// used to recover events _following_ the snapshot OR if no snapshot available then [[SeqNr.Min]]
val fromSeqNr = snapshot.map(_.metadata.seqNr + 1).getOrElse(SeqNr.Min)
for {
_ <- log.debug(s"snapshot seqNr: $snapSeqNr, load events from seqNr: $fromSeqNr").allocated
events <- eventStore.events(fromSeqNr)
seqNrL <- events.foldWhileM(snapSeqNr) {
case (_, EventStore.Event(event, seqNr)) => replay(event, seqNr).as(seqNr.asLeft[Unit])
case (_, EventStore.HighestSeqNr(seqNr)) => seqNr.asLeft[Unit].pure[F]
}
seqNr <- seqNrL match {
case Left(seqNr) => seqNr.pure[F]
case Right(_) =>
// function used in events.foldWhileM always returns Left
// and sstream.Stream.foldWhileM returns Right only if passed function do so
// thus makes getting Right here impossible
new IllegalStateException("should never happened").raiseError[F, SeqNr]
}
} yield seqNr
}
.toResource
.attempt

receive <- seqNr match {
case Right(seqNr) =>
for {
_ <- log.debug(s"recovery completed with seqNr $seqNr")
journaller <- eventStore.asJournaller(actorCtx, seqNr).toResource
receive <- recovering.completed(seqNr, journaller, snapshotStore.asSnapshotter)
} yield receive

case Left(error) =>
for {
_ <- log.error(s"recovery failed", error)
_ <- recovering.failed(error, eventStore.asDeleteTo, snapshotStore.asSnapshotter)
receive <- error.raiseError[F, Receive[F, Envelope[Any], ActorOf.Stop]].toResource
} yield receive
}
} yield receive

receive.onError {
Expand Down Expand Up @@ -158,46 +172,53 @@ object EventSourcedActorOf {

implicit final private[evolutiongaming] class EventStoreOps[F[_], E](val store: EventStore[F, E]) extends AnyVal {

def asJournaller(actorCtx: ActorCtx[F], seqNr: SeqNr)(implicit F: Concurrent[F], log: Log[F]): F[Journaller[F, E]] =
for {
seqNrRef <- Ref[F].of(seqNr)
} yield new Journaller[F, E] {
val append = new Append[F, E] {

def apply(events: Events[E]): F[F[SeqNr]] =
seqNrRef
.modify { seqNr =>
events.mapAccumulate(seqNr) {
case (seqNr0, event) =>
val seqNr1 = seqNr0 + 1
seqNr1 -> EventStore.Event(event, seqNr1)
}
}
.flatMap { events =>
def handleError(err: Throwable) = {
val from = events.values.head.head.seqNr
val to = events.values.last.last.seqNr
stopActor(from, to, err)
}
store
.save(events)
.onError(handleError)
.flatTap(_.onError(handleError))
}
def asDeleteTo: DeleteEventsTo[F] = new DeleteEventsTo[F] {

private def stopActor(from: SeqNr, to: SeqNr, error: Throwable): F[Unit] =
for {
_ <- log.error(s"failed to append events with seqNr range [$from .. $to], stopping actor", error)
_ <- actorCtx.stop
} yield {}
def apply(seqNr: SeqNr): F[F[Unit]] = store.deleteTo(seqNr)

}
}

val deleteTo = new DeleteEventsTo[F] {
def asAppend(
actorCtx: ActorCtx[F],
seqNrRef: Ref[F, SeqNr],
)(implicit F: Concurrent[F], log: Log[F]): Append[F, E] =
new Append[F, E] {

def apply(events: Events[E]): F[F[SeqNr]] =
seqNrRef
.modify { seqNr =>
events.mapAccumulate(seqNr) {
case (seqNr0, event) =>
val seqNr1 = seqNr0 + 1
seqNr1 -> EventStore.Event(event, seqNr1)
}
}
.flatMap { events =>
def handleError(err: Throwable) = {
val from = events.values.head.head.seqNr
val to = events.values.last.last.seqNr
stopActor(from, to, err)
}
store
.save(events)
.onError(handleError)
.flatTap(_.onError(handleError))
}

private def stopActor(from: SeqNr, to: SeqNr, error: Throwable): F[Unit] =
for {
_ <- log.error(s"failed to append events with seqNr range [$from .. $to], stopping actor", error)
_ <- actorCtx.stop
} yield {}

def apply(seqNr: SeqNr): F[F[Unit]] = store.deleteTo(seqNr)
}

}
def asJournaller(actorCtx: ActorCtx[F], seqNr: SeqNr)(implicit F: Concurrent[F], log: Log[F]): F[Journaller[F, E]] =
for {
seqNrRef <- Ref[F].of(seqNr)
} yield new Journaller[F, E] {
val append = asAppend(actorCtx, seqNrRef)
val deleteTo = asDeleteTo
}
}
}
6 changes: 2 additions & 4 deletions persistence/src/main/scala/com/evolutiongaming/akkaeffect/persistence/Persistence.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -58,8 +58,7 @@ private[akkaeffect] object Persistence {
) = {
val receive = for {
recovering <- recoveryStarted(seqNr, none)
context = Recovering.RecoveryContext(seqNr, journaller, snapshotter)
receive <- recovering.completed(context)
receive <- recovering.completed(seqNr, journaller, snapshotter)
} yield Persistence.receive[F, S, E, C](receive)
receive.toReleasable
}
Expand Down Expand Up @@ -114,9 +113,8 @@ private[akkaeffect] object Persistence {
.foldMapM(_.release)
.toResource
.productR {
val context = Recovering.RecoveryContext(seqNr, journaller, snapshotter)
recovering
.completed(context)
.completed(seqNr, journaller, snapshotter)
.map(receive => Persistence.receive[F, S, E, C](receive))
}
.toReleasable
Expand Down
Loading