Doobie, Zio & transaction scoping

The problem

Two tables: session and clickcollect

Either we want both inserts to succeed or none

ZIO


def insertSession(userId: ClickCollectUserId, state: State): UIO[(Session, ClickCollectRecord[State])] =
	for
		now <- Clock.instant
		(initialSession, record) <- insertClickCollectQuery(userId, state, now).transaction
		_ <- ZIO.foreachDiscard(initialSession.existingStartedAt)(time =>
			ZIO.logWarning(s"Still existing session when initiating new session. Existing session started at ${time.toString}")
		)
	yield (initialSession.session, record)

ConnectionIO


private def insertClickCollectQuery(userId: ClickCollectUserId, state: State, now: Instant): ConnectionIO[(InitResult, ClickCollectRecord[State])] = for
	initialSession <- SessionRepository.insertQuery(userId, now)
	record <- insertQuery(initialSession.session.id, state, now).unique
yield (initialSession, record)

trait SessionRepository{
  def insertQuery(userId: UserId, now: Instant): ConnectionIO[InitResult] = for
    maybeActive <- SessionRepository.findActiveQuery(userId).option
    _ <- maybeActive.traverse(active => SessionRepository.setEndedAtQuery(active.id, now).run)
    session = Session.initial(userId, now)
    _ <- SessionRepository.insertQuery(session).run
  yield InitResult(session, maybeActive.map(_.createdAt))
}
In ZIO code the 'pieces' are mostly ZIO[R,E,A] but the Doobie sql parts are ConnectionIO

Two worlds

So what can we do?
  • Have them in separate zio's. No shared transaction
    • Easy to do unwittingly
  • Have ConnectionIO methods separate from ZIO methods that need a transactor
    • So that you can reuse them in different combinations
  • Parametrize the repo over the effect F[_]
  • ...
  • Allow marking of arbitrary zio code as transactional
    • One monad to rule them all: zio
Like in Kotlin / Java 

@Transactional
public void createCourse(Course course) {
    courseDao.create(course);
    throw new DataIntegrityViolationException("Throwing exception for demoing Rollback!!!");
}

How do we do that?

  • using ZIO R environment
  • like in java: store in fiber context
  • with scala 3 context functions

How does Doobie ConnectionIO work?

Transactor

A Doobie Transactor contains
  • a source of jdbc Connection
  • an interpreter
  • a Strategy which knows what to do before, after, onError

Doobie sql evaluation

sql program ⇒ interpreter ⇒ connection ⇒ result

Sql program

ConnectionIO[A]

is a

Free[ConnectionOp, A]

a program of ConnectionOp that can be interpreted and will return A

a ConnectionOp is something you can do with a jdbc transaction. Do an insert, set a savepoint, rollback etc

Interpreter

ConnectionOp ~> Kleisli[Task, Connection, *]
  • ~> is a FunctionK
    • Transform monad A[_] to B[_]
    • Like ZIO.fromTry transforms a Try[A] to a Task[A]
  • Kleisli[Task, A, B] is a function A => Task[B]
  • so we give the interpreter a sql program and it gives us a function that needs a connection to give the result
'a function that needs a connection to give the result'

this is something we can do as a zio... 💡

3 solutions for arbitrary scoping

Standard Doobie Zio

A simple sql program. ZIO code and transactions live in separate worlds.

(for comparison)

See demo.DoobieZio0.scala

Typed

Transactional scoping of our zio code that we see in the types

See demo.DoobieZio1.scala

Untyped

Transactional scoping of our zio code that we don't see in the types

See demo.DoobieZio2.scala

Scala 3

Transactional scoping of our zio code with scala 3 context functions

See demo.DoobieZio3.scala
Other approach: lift zio into ConnectionIO

But then you still have two separate worlds: zio and ConnectionIO

Discussion: how to use transactional scoping

Some use cases and pitfalls

  • All tasks should succeed or none
  • Prevent concurrent updates (pessimistic locking)
  • Simple batch runner. Multiple service instances but only one job runs
  • Be careful with how long tasks take in the transactional scope
    • If they sometimes take, say, seconds it can deplete your connection pool
    • So put timeout guards and/or use separate connection pool

Fin