If you work with one database that supports transactions, you donβt even think about consistency - the database does everything for you. If you have several databases, a distributed system, or even, for example, MongoDB up to version 4, everything is not so rosy.
Consider an example - we want to save the file in the repository and add a link to it in two documents. Of course, we want atomicity - either the file is saved and added to the documents, or neither (the cats-effects IO are used hereinafter):
saveDataToFile(data)
.flatMap { file =>
addFileRef(documentId, file)
.flatMap { result =>
addFileRef(fileRegistry, file)
.flatMap { result =>
???
}
.handleErrorWith { error =>
removeFileRef(documentId, file).attempt >> IO.raiseError(error)
}
}
.handleErrorWith { error =>
removeFile(file).attempt >> IO.raiseError(error)
}
}
? Pyramid of doom.
! , .
, , . ( ) "" .
- Saga, . / , . , . β Scala .
β :
final case class Action(
perform: ???,
commit: ???,
compensate: ???
)
? - () => Try[T]
, IO
β , cats.
final case class Action(
perform: IO[Unit],
commit: IO[Unit],
compensate: IO[Unit]
)
commit
perform
, compensate
perform
?
:
final case class Action[T](
perform: IO[T],
commit: T => IO[Unit],
compensate: T => IO[Unit]
)
, T
β .. perform
.
:
def saveDataToFile(data: Data): IO[File] = ???
def removeFile(file: File): IO[Unit] = ???
def saveDataAction(data: Data): Action[File] = Action(
perform = saveDataToFile(data),
compensate = removeFile
)
β
, , ? Seq[Action[_]]
β . β , .
β - ? .
:
final case class ActionChain[A, B](
first: Action[A],
next: A => Action[B]
)
, β . :
sealed trait Transaction[T]
final case class Action[T](...) extends Transaction[T]
final case class ActionChain[A, B](
first: Transaction[A],
next: A => Transaction[B]
) extends Transaction[B]
! , :
ActionChain(
saveDataAction(data),
{ file =>
ActionChain(
addFileRefAction(documentId, file),
{ _ =>
addFileRefAction(fileRegistry, file)
}
)
}
)
, .
, β , ?
, , IO
.
, . "" β .
Transaction
case
' . , :
- β
private def compile[R](restOfTransaction: T => IO[R]): IO[R] = this match {
case Action(perform, commit, compensate) => perform.flatMap { t =>
restOfTransaction(t).redeemWith(
bind = commit(t).attempt >> IO.pure(_),
recover = compensate(t).attempt >> IO.raiseError(_)
)
}
case ActionChain(first, next) => ???
}
, CatsredeemWith
, attempt
/ ( , ), >>
" ", IO.pure
IO.raiseError
continue(t)
β .
β , :
private def compile[R](restOfTransaction: T => IO[R]): IO[R] = this match {
case Action(perform, commit, compensate) => ...
case ActionChain(first, next) =>
first.compile { a =>
next(a).compile { t =>
restOfTransaction(t)
}
}
}
, :
sealed trait Transaction[T] {
def compile: IO[T] = compile(IO.pure)
}
IO
, commit
/compensate
, ( / IO
, ).
, . , ? , .
, . :
-, Action.compile(restOfTransaction)
:
perform
, restOfTransaction
( T
, perform
)compile
: perform
, restOfTransaction
( T
)perform
, commit
compensate
restOfTransaction
( redeemWith
)
ActionChain.compile(restOfTransaction)
, Action
' , compile
':
transaction.compile(restOfTransaction)
===
action1.compile(t1 =>
action2.compile(t2 =>
action3.compile(t3 => ...
restOfTransaction(tn))))
, ActionChain.first
ActionChain
:
ActionChain(ActionChain(action1, t1 => action2), t2 => action3).compile(restOfTransaction) >>
ActionChain(action1, t1 => action2).compile(t2 => action3.compile(restOfTransaction)) >>
action1.compile(t1 => action2.compile(t2 => action3.compile(restOfTransaction))) []
Action.compile
:
commit
compensate
?
:
- ( )
- ,
- ???
- PROFIT!
, :
ActionChain(
saveDataAction(data),
{ file =>
ActionChain(
addFileRefAction(documentId, file),
{ _ =>
addFileRefAction(fileRegistry, file)
}
)
}
).compile
chain
ActionChain
, :
sealed trait Transaction[T] {
def chain[R](f: T => Transaction[R]): Transaction[R] = ActionChain(this, f)
}
saveDataAction(data).chain { file =>
addFileRefAction(documentId, file).chain { _ =>
addFileRefAction(fileRegistry, file)
}
}.compile
chain
flatMap
, , () !
, Scala, for
cats
, ( ).
, β !
sealed trait Transaction[T] {
def flatMap[R](f: T => Transaction[R]): Transaction[R] = ActionChain(this, f)
def map[R](f: T => R): Transaction[R] = flatMap { t => Action(IO(f(t))) }
}
Scala, :
def saveDataAndAddFileRefs(data: Data, documentId: ID): Transaction[Unit] = for {
file <- saveDataAction(data)
_ <- addFileRefAction(documentId, file)
_ <- addFileRefAction(fileRegistry, file)
} yield ()
saveDataAndAddFileRefs(data, documentId).compile
, β . β , .
For cats
you need to explicitly declare evidence that Transaction
this is a monad - an instance of a typclass:
object Transaction {
implicit object MonadInstance extends Monad[Transaction] {
override def pure[A](x: A): Transaction[A] = Action(IO.pure(x))
override def flatMap[A, B](fa: Transaction[A])(f: A => Transaction[B]): Transaction[B] = fa.flatMap(f)
override def tailRecM[A, B](a: A)(f: A => Transaction[Either[A, B]]): Transaction[B] = f(a).flatMap {
case Left(a) => tailRecM(a)(f)
case Right(b) => pure(b)
}
}
}
What does this give us? The ability to use ready-made methods from cats
, for example:
val transaction = dataChunks.traverse_ { data =>
saveDataAndAddFileRefs(data, documentId)
}
transaction.compile
All code is available here: https://github.com/atamurius/scala-transactions