Functional approach to transactions on Scala or write your own useful monad

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) // (1)
  .flatMap { file =>
    addFileRef(documentId, file) // (2)
      .flatMap { result =>
        addFileRef(fileRegistry, file) // (3)
          .flatMap { result =>
            ??? // (4, 5, ...)
          }
          .handleErrorWith { error =>
            // revert (2)
            removeFileRef(documentId, file).attempt >> IO.raiseError(error)
          }
      }
      .handleErrorWith { error =>
        // revert (1)
        removeFile(file).attempt >> IO.raiseError(error)
      }
  }

? Pyramid of doom.

! , .

, , . ( ) "" .

— :

• —

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' . , :

2. —
   

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) => ???
}

— , :

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, ).

transaction.compile(restOfTransaction) 
=== 
action1.compile(t1 => 
  action2.compile(t2 =>
    action3.compile(t3 => ...
      restOfTransaction(tn))))

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))) []

?

:

1. ( )
2. ,
   
3. ???
4. 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 catsyou need to explicitly declare evidence that Transactionthis 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)

    //    cats-effects,       
    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