Async Utilities

Use ReaderT for safely converting Future-based traits to cats-effect

You have a higher-kinded trait like this:

trait FooService[F[_]] {
  def foo(i: Int): F[Unit]
}

and an implementation in Future:

class FutureFoo extends FooService[Future] {
  def foo(i: Int): Future[Unit] = Future(println(i))
}

(Perhaps the implementation was automatically generated by a tool like Twitter Scrooge, in which case, scroll down for Twitter Future and Finagle support!)

With cats-tagless, you can auto-derive a FunctorK[FooService] instance and an implementation in ReaderT[Future, FooService[Future], *]:

object FooService {
  import cats.tagless._

  implicit def FooServiceReaderT[F[_]]: FooService[ReaderT[F, FooService[F], *]] = 
    Derive.readerT[FooService, F]
  implicit val FooServiceFunctorK: FunctorK[FooService] = Derive.functorK
}

These implicit instances can be automatically added to the code generated by Scrooge by running the AddCatsTaglessInstances scalafix rule! See below for more detail.

Now you can safely convert your Future-based implementation into one in F[_] : Async:

import cats.tagless.syntax.all._
import com.dwolla.util.async.stdlib._

val futureFoo: FooService[Future] = new FutureFoo

val fooService: FooService[IO] = futureFoo.asyncMapK[IO]

Artifacts

The Group ID for each artifact is "com.dwolla". All artifacts are published to Maven Central.

Artifact	Description	Scala 2.12	Scala 2.13	Scala.js	Scala Native
"async-utils-core"	Shared definition of AsyncFunctorK and supporting code	✅	✅	✅	✅
"async-utils"	Implementation for stdlib Scala Future	✅	✅	✅	✅
"async-utils-twitter"	Implementation for Twitter Future	✅	✅	❌	❌
"async-utils-finagle"	Safely create Thrift clients and servers using cats-effect as the effect type	✅	✅	❌	❌
"async-utils-finagle-natchez"	Bridge between Natchez tracing and Finagle's built-in Zipkin support	✅	✅	❌	❌
"async-utils-finagle-otel4s"	Trace Finagle Thrift servers with otel4s, continuing traces received by Finagle	❌	✅	❌	❌
"async-utils-http4s"	Gracefully shut down an http4s server behind a health-checked registry or load balancer	✅	✅	❌	✅
"finagle-tagless-scalafix"	Automatically adds implicit instances needed by `asyncMapK` to the companion objects of Finagle services generated by Scrooge	✅	✅	❌	❌

Twitter Ecosystem Releases

Twitter uses a {YEAR}-{MONTH} version scheme, where the combination of the two forms a major version.

The "current" version of the artifacts published by this project intend to track the latest Twitter ecosystem release, with previously supported versions published as supplemental artifacts with the supported release version appended to the artifact name.

For example, the latest Twitter ecosystem version is 22.7.0, so the latest version of com.dwolla::async-utils-twitter depends on com.twitter::util-core:22.7.0. In addition, we publish artifacts named like com.dwolla::async-utils-twitter-22.4.0 for each of the previously supported Twitter ecosystem releases.

Twitter Futures

import cats.data.ReaderT
import cats.tagless.{Derive, FunctorK}
import com.twitter.util.Closable

// generated by twitter-scrooge
trait FooScroogeService[F[_]] {
  def foo(i: Int): F[Unit]

  def asClosable: Closable = Closable.nop
}

object FooScroogeService {
  // Let Scalafix generate these instances for you! 
  // Follow the instructions in the `Scalafix Rule` section below.
  implicit def FooScroogeServiceReaderT[F[_]]: FooScroogeService[ReaderT[F, FooScroogeService[F], *]] =
    Derive.readerT[FooScroogeService, F]
  implicit val FooScroogeServiceFunctorK: FunctorK[FooScroogeService] = Derive.functorK[FooScroogeService]
}

Finagle Clients

Safely create a Finagle client in IO from an implementation in Twitter Future:

import com.dwolla.util.async.finagle.ThriftClient
import com.dwolla.util.async.twitter._

val fooClient: Resource[IO, FooScroogeService[IO]] = ThriftClient[FooScroogeService]("destination")

Finagle Servers

Safely create a Finagle server in Twitter Future from an implementation in IO:

import com.dwolla.util.async.finagle.ThriftServer
import com.dwolla.util.async.twitter._
import com.twitter.finagle.ListeningServer

val fooImpl: FooScroogeService[IO] = new FooScroogeService[IO] {
  def foo(i: Int): IO[Unit] = IO(println(i))
}

val thriftServer: Resource[IO, ListeningServer] = ThriftServer("address", fooImpl)

Releasing the server's Resource shuts it down gracefully. The server refuses new connections at once and closes idle ones, while requests already in flight get up to ThriftServer.DefaultShutdownGracePeriod (15 seconds) to finish; release completes as soon as they do. A request still running when the grace period ends is interrupted, which cancels the effect handling it. The traced servers in async-utils-finagle-natchez and async-utils-finagle-otel4s shut down the same way.

Finagle Servers Traced with otel4s

Using async-utils-finagle-otel4s, safely create a Finagle server whose requests are each wrapped in an otel4s SpanKind.Server span named like FooScroogeService/foo:

import com.comcast.ip4s._
import com.dwolla.util.async.finagle.otel4s.TracedThriftServer
import org.typelevel.otel4s.trace.TracerProvider

implicit val tracerProvider: TracerProvider[IO] = ??? // e.g. from otel4s-oteljava

val thriftServer: Resource[IO, ListeningServer] =
  TracedThriftServer(SocketAddress(ip"0.0.0.0", port"8080"), "foo-service", fooImpl)

The Thrift algebra needs a cats.tagless.aop.Instrument instance, which the AddCatsTaglessInstances Scalafix rule generates. An overload accepts a SpanBuilder[F] => SpanBuilder[F] function to customize each request's span.

Each span is named {service}/{method} and carries the attributes rpc.system.name (thrift, a custom value because the OpenTelemetry RPC conventions have no Thrift entry) and rpc.method (the same {service}/{method} as the span name). The customization function runs after these are set, so it can override them.

When a Finagle client calls the server, Finagle receives the caller's TraceId. The server continues that trace by making the caller's span the remote parent of the server span, so no otel4s propagator (B3 or otherwise) needs to be configured. The trace is sampled unless Finagle's TraceId explicitly says it isn't (an undecided TraceId is continued as sampled). Requests without a Finagle TraceId start new root spans. Finagle's own tracing is disabled on this server (NullTracer), so a Finagle Tracer registered through LoadService doesn't record a duplicate server span.

Releasing the server's Resource gives requests in flight a grace period to finish, like ThriftServer (see Finagle Servers above).

otel4s isn't published for Scala 2.12, so this module is only available for Scala 2.13.

http4s Graceful Shutdown

Available since 1.6.0 for Scala 2.12, 2.13, and 3 on the JVM, and since 1.6.1 for Scala Native:

libraryDependencies += "com.dwolla" %% "async-utils-http4s" % "1.6.0"

// or, in a project cross-built for Scala Native
libraryDependencies += "com.dwolla" %%% "async-utils-http4s" % "1.6.1"

async-utils-http4s helps an http4s server stop receiving traffic before it stops serving it, when a health-checked service registry or load balancer decides where requests go. Once the server begins shutting down, GET /health answers 503 (it answers 204 No Content while serving) and every response carries Connection: close. The server is released only after a drain period, giving the registry time to notice and stop routing to it. Requests already running when the drain period ends get the server's shutdown timeout to finish.

Load-balancer health checks must accept 204 (or any 2xx), because some default to matching only 200. The example also needs "org.http4s" %% "http4s-ember-server", because this module itself depends only on http4s-core.

import cats.effect._
import cats.syntax.all._
import com.comcast.ip4s._
import com.dwolla.util.async.http4s._
import org.http4s._
import org.http4s.ember.server.EmberServerBuilder
import org.http4s.server.Server
import scala.concurrent.duration._

def server(routes: HttpRoutes[IO]): Resource[IO, Server] =
  DrainSignal[IO].toResource.flatMap { signal =>
    GracefulShutdown.drainBeforeRelease(signal, GracefulShutdown.DefaultDrainPeriod) {
      EmberServerBuilder
        .default[IO]
        .withHost(host"0.0.0.0")
        .withPort(port"8080")
        .withShutdownTimeout(5.seconds)
        .withHttpApp(
          GracefulShutdown.closeConnectionsWhileDraining(signal)(
            (GracefulShutdown.healthRoutes(signal) <+> routes).orNotFound
          )
        )
        .build
    }
  }

DrainSignal is the one-way switch the pieces share. Releasing the Resource returned by drainBeforeRelease starts draining, waits for the drain period, and then releases the server. If the server fails to start, nothing waits and the signal is left alone.

Choose the timings to satisfy two rules:

1. The drain period must be longer than the time the registry or load balancer takes to stop routing to the instance: the health-check interval, plus its timeout, plus the time the change takes to propagate.
2. The drain period, plus the server's shutdown timeout, plus the time any other finalizers take, must fit within the orchestrator's stop timeout (the time between SIGTERM and SIGKILL).

The example's DefaultDrainPeriod of 10 seconds plus a 5 second shutdown timeout takes 15 seconds, which fits within a typical 30 second stop timeout. A keep-alive connection that is idle through the whole drain period is closed when the shutdown timeout ends.

A few limits to know about:

• Connection: close is added while draining, and only matters for HTTP/1.x. Ember strips connection-specific headers on HTTP/2, so h2 connections get no close signal and are closed at the server's shutdown timeout.
• Errors the app raises, rather than returns as a response, are answered by the server's error handler outside the middleware, so they don't get Connection: close. To cover them, put http4s's org.http4s.server.middleware.ErrorHandling (or equivalent) inside closeConnectionsWhileDraining.
• A service that runs more than one server, such as an HTTP server and a Thrift server, should compose them in parallel ((http, thrift).parTupled or Resource.both) so they drain concurrently. Composed sequentially, their release times add up.
• Releasing waits the full drain period even on Ctrl-C, so local or test runs should pass a short drain period.

Scalafix Rule

Add Scalafix to your project's build by following the instructions:

1. Add the Scalafix plugin to the project by adding this to project/plugins.sbt:

   addSbtPlugin("ch.epfl.scala" % "sbt-scalafix" % "0.12.1")

2. Enable SemanticDB by adding this to build.sbt:

   ThisBuild / semanticdbEnabled := true
   ThisBuild / semanticdbVersion := scalafixSemanticdb.revision
   ThisBuild / scalafixScalaBinaryVersion := CrossVersion.binaryScalaVersion(scalaVersion.value)
   ThisBuild / scalafixDependencies += "com.dwolla" %% "finagle-tagless-scalafix" % "1.1.1"

3. Run the Scalafix rule automatically after generating the Thrift sources by adding this to build.sbt:

   Compile / scalafix / unmanagedSources := (Compile / sources).value
   scalafixOnCompile := true
   libraryDependencies ++= {
     val catsTaglessV = "0.14.0"
     Seq(
       "org.typelevel" %% "cats-tagless-core" % catsTaglessV,
       "org.typelevel" %% "cats-tagless-macros" % catsTaglessV,
     )
   }

   and adding this to .scalafix.conf:

   triggered.rules = [ AddCatsTaglessInstances ]

AddCatsTaglessInstances

The AddCatsTaglessInstances rule finds generated Thrift service traits and adds implicit instances of ThriftService[Kleisli[F, ThriftService[Future], *]] and FunctorK[ThriftService] to each service's companion object.

Twitter's Scrooge project changed the way it generates code for Thrift services, removing the higher-kinded service trait used by this library, leaving only the MethodPerEndpoint trait that used to extend the higher-kinded service trait, setting the type parameter to com.twitter.util.Future. The AddCatsTaglessInstances rule now addresses this as well, rewriting MethodPerEndpoint to {Name}Service and reintroducing the type parameter. (A new MethodPerEndpoint is also added, going back to how it used to extend {Name}Service[Future].)

This Scalafix rule should be idempotent, so it can be rerun many times.

AdaptHigherKindedThriftCode

Because the AddCatsTaglessInstances rewrite rule couldn't easily move the new {Name}Service trait up to the same level as the {Name}Service object, the new traits must be addressed differently. In other words, instead of finding the trait at com.example.ThriftService, it will now be at com.example.ThriftService.ThriftService.

The AdaptHigherKindedThriftCode rule exists to adapt existing code to the new location. It will find references to traits that extend com.twitter.finagle.thrift.ThriftService and have a type parameter of the correct shape, and add the object name before the trait name (i.e., rewriting ThriftService to ThriftService.ThriftService or com.example.ThriftService to com.example.ThriftService.ThriftService).

This rule is not idempotent, but it will typically only be executed once per codebase.

The order in which the rule is executed matters. Follow these steps:

1. Add Scalafix to your project by following steps 1 and 2 under "Scalafix Rule" above.

2. Look at your project's sbt project graph. Because the rule is a semantic rule, it depends on the compiler being able to compile the code it will modify. This means the leaves of the project graph need to be updated before the nodes that depend on each leaf.

   For example, run Test/scalafix AdaptHigherKindedThriftCode before running Compile/scalafix AdaptHigherKindedThriftCode.

3. Only after running the AdaptHigherKindedThriftCode rule should you update the Scrooge and Finagle version being used in the project. Once this is updated, you can run the AddCatsTaglessInstances rule on the updated generated code.

Credits

Thanks to Georgi Krastev and the cats-tagless project for the idea to use ReaderT in this way.