Функциональный подход к транзакциям на Scala или пишем свою полезную монаду

Если вы работаете с одной базой данных которая поддерживает транзакции вы даже не задумываетесь о консистентности — база все делает за вас. Если же у вас несколько баз, распределенная система или даже к примеру MongoDB до 4 версии — все не так радужно.

Рассмотрим пример — мы хотим сохранить файл в хранилище и добавить ссылку на него в два документа. Конечно же мы хотим атомарности — либо файл сохранен и добавлен в документы либо ни то ни другое (тут и далее используется cats-effects IO):

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'ов нам нужно рассмотреть два случая. Если мы хотим выполнить атомарный шаг, то логика следующая:

1. выполнить действие
2. если позникла ошибка ничего делать не надо — действие не удалось и не требует
   компенсации
3. если действие удалось нужно продолжать транзакцию до конца
4. по окончанию транзакции нужно подтвердить дейтсвие или же компенсировать его

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

Вот теперь мы имеем чистый код, явно описывающий намерение — последовательно совершить несколько действий. При этом побочный эффект — транзакционность полностью скрыт, но гарантируется.

Для cats нужно явно объявить доказательство того, что Transaction это монада — экземпляр тайпкласса:

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

Что это нам дает? Возможность использовать готовые методы из cats, например:

val transaction = dataChunks.traverse_ { data =>
  saveDataAndAddFileRefs(data, documentId) // выполняет действие для всех элементов списка
}
transaction.compile // собирает все действия в одну транзакцию

Весь код доступен тут: https://github.com/atamurius/scala-transactions