ExecutionContext[EC] phantom for control, Future[EC, A] (#7347)

* add phantom-tagged ExecutionContext and Future to scala-utils concurrent package

* many new operations for Futures

* Future, ExecutionContext combinators from porting ledger-on-sql

- picked from 546b84ab9cdf4de2d93ec5682bdee6cfd6b385f8

* move Future, ExecutionContext companions into normal package

* lots of new docs

* many new Future utilities

* working zipWith

* tests for ExecutionContext resolution, showing what will be picked under different scenarios

* even more tests for ExecutionContext resolution

* tests showing some well-typed and ill-typed Future combinator usage

* no changelog

CHANGELOG_BEGIN
CHANGELOG_END

* missed scalafmt

* one more doc note

* split concurrent package to concurrent library

Co-authored-by: Samir Talwar <samir.talwar@digitalasset.com>

libs-scala/concurrent/BUILD.bazel

@@ -0,0 +1,44 @@
+# Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+load(
+    "//bazel_tools:scala.bzl",
+    "da_scala_library",
+    "da_scala_test",
+    "lf_scalacopts",
+)
+
+scalacopts = lf_scalacopts + [
+    "-P:wartremover:traverser:org.wartremover.warts.NonUnitStatements",
+]
+
+da_scala_library(
+    name = "concurrent",
+    srcs = glob(["src/main/scala/**/*.scala"]),
+    plugins = [
+        "@maven//:org_spire_math_kind_projector_2_12",
+    ],
+    scalacopts = scalacopts,
+    tags = ["maven_coordinates=com.daml:concurrent:__VERSION__"],
+    visibility = [
+        "//visibility:public",
+    ],
+    deps = [
+        "@maven//:org_scalaz_scalaz_core_2_12",
+    ],
+)
+
+da_scala_test(
+    name = "test",
+    srcs = glob(["src/test/scala/**/*.scala"]),
+    plugins = [
+        "@maven//:com_github_ghik_silencer_plugin_2_12_11",
+    ],
+    scalacopts = scalacopts + ["-P:silencer:checkUnused"],
+    deps = [
+        ":concurrent",
+        "@maven//:com_chuusai_shapeless_2_12",
+        "@maven//:com_github_ghik_silencer_lib_2_12_11",
+        "@maven//:org_scalaz_scalaz_core_2_12",
+    ],
+)

libs-scala/concurrent/src/main/scala/concurrent/ExecutionContextOf.scala

@@ -0,0 +1,19 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.concurrent
+
+import scala.language.higherKinds
+import scala.{concurrent => sc}
+
+sealed abstract class ExecutionContextOf {
+  type T[+P] <: sc.ExecutionContext
+  private[concurrent] def subst[F[_], P](fe: F[sc.ExecutionContext]): F[T[P]]
+}
+
+object ExecutionContextOf {
+  val Instance: ExecutionContextOf = new ExecutionContextOf {
+    type T[+P] = sc.ExecutionContext
+    override private[concurrent] def subst[F[_], P](fe: F[sc.ExecutionContext]) = fe
+  }
+}

libs-scala/concurrent/src/main/scala/concurrent/FutureOf.scala

@@ -0,0 +1,165 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.concurrent
+
+import scala.language.{higherKinds, implicitConversions}
+import scala.{concurrent => sc}
+import scala.util.Try
+
+import scalaz.{Catchable, Cobind, Isomorphism, Leibniz, MonadError, Nondeterminism, Semigroup}
+import Isomorphism.<~>
+import Leibniz.===
+import scalaz.std.scalaFuture._
+
+sealed abstract class FutureOf {
+
+  /** We don't use [[sc.Future]] as the upper bound because it has methods that
+    * collide with the versions we want to use, i.e. those that preserve the
+    * phantom `EC` type parameter.  By contrast, [[sc.Awaitable]] has only the
+    * `ready` and `result` methods, which are mostly useless.
+    */
+  type T[-EC, +A] <: sc.Awaitable[A]
+  private[concurrent] def subst[F[_[+ _]], EC](ff: F[sc.Future]): F[T[EC, +?]]
+}
+
+/** Instances and methods for `FutureOf`. You should not import these; instead,
+  * enable `-Xsource:2.13` and they will always be available without import.
+  */
+object FutureOf {
+  val Instance: FutureOf = new FutureOf {
+    type T[-EC, +A] = sc.Future[A]
+    override private[concurrent] def subst[F[_[+ _]], EC](ff: F[sc.Future]) = ff
+  }
+
+  type ScalazF[F[+ _]] = Nondeterminism[F]
+    with Cobind[F]
+    with MonadError[F, Throwable]
+    with Catchable[F]
+
+  implicit def `future Instance`[EC: ExecutionContext]: ScalazF[Future[EC, +?]] =
+    Instance subst [ScalazF, EC] implicitly
+
+  implicit def `future Semigroup`[A: Semigroup, EC: ExecutionContext]: Semigroup[Future[EC, A]] = {
+    type K[T[+ _]] = Semigroup[T[A]]
+    Instance subst [K, EC] implicitly
+  }
+
+  implicit def `future is any type`[A]: sc.Future[A] === Future[Any, A] =
+    Instance subst [Lambda[`t[+_]` => sc.Future[A] === t[A]], Any] Leibniz.refl
+
+  /** A [[sc.Future]] converts to our [[Future]] with any choice of EC type. */
+  implicit def `future is any`[A](sf: sc.Future[A]): Future[Any, A] =
+    `future is any type`(sf)
+
+  private[this] def unsubstF[Arr[_, + _], A, B](f: A Arr Future[Nothing, B]): A Arr sc.Future[B] = {
+    type K[T[+ _]] = (A Arr T[B]) => A Arr sc.Future[B]
+    (Instance subst [K, Nothing] identity)(f)
+  }
+
+  def swapExecutionContext[L, R]: Future[L, ?] <~> Future[R, ?] =
+    Instance.subst[Lambda[`t[+_]` => t <~> Future[R, ?]], L](
+      Instance.subst[Lambda[`t[+_]` => sc.Future <~> t], R](implicitly[sc.Future <~> sc.Future]))
+
+  /** Common methods like `map` and `flatMap` are not provided directly; instead,
+    * import the appropriate Scalaz syntax for these; `scalaz.syntax.bind._`
+    * will give you `map`, `flatMap`, and most other common choices.  Only
+    * exotic Future-specific combinators are provided here.
+    */
+  implicit final class Ops[-EC, +A](private val self: Future[EC, A]) extends AnyVal {
+
+    /** `.require[NEC]` is a friendly alias for `: Future[NEC, A]`. */
+    def require[NEC <: EC]: Future[NEC, A] = self
+
+    def transform[B](f: Try[A] => Try[B])(implicit ec: ExecutionContext[EC]): Future[EC, B] =
+      self.removeExecutionContext transform f
+
+    // The rule of thumb is "EC determines what happens next". So `recoverWith`
+    // doesn't let the Future returned by pf control what EC it uses to *call* pf,
+    // because that happens "before".  Same with `transformWith`.  By contrast,
+    // zipWith's f gets called "after" the two futures feeding it arguments, so
+    // we allow both futures control over the EC used to invoke f.
+
+    def transformWith[LEC <: EC, B](f: Try[A] => Future[LEC, B])(
+        implicit ec: ExecutionContext[EC]): Future[LEC, B] =
+      self.removeExecutionContext transformWith unsubstF(f)
+
+    def collect[B](pf: A PartialFunction B)(implicit ec: ExecutionContext[EC]): Future[EC, B] =
+      self.removeExecutionContext collect pf
+
+    def failed: Future[EC, Throwable] = self.removeExecutionContext.failed
+
+    def fallbackTo[LEC <: EC, B >: A](that: Future[LEC, B]): Future[LEC, B] =
+      self.removeExecutionContext fallbackTo that.removeExecutionContext
+
+    def filter(p: A => Boolean)(implicit ec: ExecutionContext[EC]): Future[EC, A] =
+      self.removeExecutionContext filter p
+
+    def withFilter(p: A => Boolean)(implicit ec: ExecutionContext[EC]): Future[EC, A] =
+      self.removeExecutionContext withFilter p
+
+    def recover[B >: A](pf: Throwable PartialFunction B)(
+        implicit ec: ExecutionContext[EC]): Future[EC, B] =
+      self.removeExecutionContext recover pf
+
+    def recoverWith[LEC <: EC, B >: A](pf: Throwable PartialFunction Future[LEC, B])(
+        implicit ec: ExecutionContext[EC]): Future[EC, B] =
+      self.removeExecutionContext recoverWith unsubstF(pf)
+
+    def transform[B](s: A => B, f: Throwable => Throwable)(
+        implicit ec: ExecutionContext[EC]): Future[EC, B] =
+      self.removeExecutionContext transform (s, f)
+
+    def foreach[U](f: A => U)(implicit ec: ExecutionContext[EC]): Unit =
+      self.removeExecutionContext foreach f
+
+    def andThen[U](pf: Try[A] PartialFunction U)(implicit ec: ExecutionContext[EC]): Future[EC, A] =
+      self.removeExecutionContext andThen pf
+
+    def onComplete[U](f: Try[A] => U)(implicit ec: ExecutionContext[EC]): Unit =
+      self.removeExecutionContext onComplete f
+
+    def zip[LEC <: EC, B](that: Future[LEC, B]): Future[LEC, (A, B)] = {
+      type K[T[+ _]] = (T[A], T[B]) => T[(A, B)]
+      Instance.subst[K, LEC](_ zip _)(self, that)
+    }
+
+    def zipWith[LEC <: EC, B, C](that: Future[LEC, B])(f: (A, B) => C)(
+        implicit ec: ExecutionContext[LEC]): Future[LEC, C] = {
+      type K[T[+ _]] = (T[A], T[B]) => T[C]
+      Instance.subst[K, LEC](_.zipWith(_)(f))(self, that)
+    }
+  }
+
+  /** Operations that don't refer to an ExecutionContext. */
+  implicit final class NonEcOps[+A](private val self: Future[Nothing, A]) extends AnyVal {
+
+    /** Switch execution contexts for later operations.  This is not necessary if
+      * `NEC <: EC`, as the future will simply widen in those cases, or you can
+      * use `require` instead, which implies more safety.
+      */
+    def changeExecutionContext[NEC]: Future[NEC, A] =
+      swapExecutionContext[Nothing, NEC].to(self)
+
+    /** The "unsafe" conversion to Future.  Does nothing itself, but removes
+      * the control on which [[sc.ExecutionContext]] is used for later
+      * operations.
+      */
+    def removeExecutionContext: sc.Future[A] =
+      self.changeExecutionContext[Any].asScala
+
+    def isCompleted: Boolean = self.removeExecutionContext.isCompleted
+
+    def value: Option[Try[A]] = self.removeExecutionContext.value
+  }
+
+  /** Operations safe if the Future is set to any ExecutionContext. */
+  implicit final class AnyOps[+A](private val self: Future[Any, A]) extends AnyVal {
+
+    /** The "safe" conversion to Future.  `EC = Any` already means "use any
+      * ExecutionContext", so there is little harm in restating that by
+      * referring directly to [[sc.Future]].
+      */
+    def asScala: sc.Future[A] = `future is any type`[A].flip(self)
+  }
+}

libs-scala/concurrent/src/main/scala/concurrent/package.scala

@@ -0,0 +1,110 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml
+
+import scala.util.Try
+import scala.{concurrent => sc}
+
+import scalaz.Id.Id
+
+/** A compatible layer for `scala.concurrent` with extra type parameters to
+  * control `ExecutionContext`s.  Deliberately uses the same names as the
+  * equivalent concepts in `scala.concurrent`.
+  *
+  * The trouble with [[sc.ExecutionContext]] is that it is used incoherently.
+  * This leads to the problems described in
+  * https://failex.blogspot.com/2020/05/global-typeclass-coherence-principles-3.html
+  * .  The extension layer in this package adds a phantom type parameter to
+  * `ExecutionContext` and related types, so that types can be used to
+  * discriminate between ExecutionContexts at compile-time, and so Futures can
+  * declare which ExecutionContext their operations are in.
+  *
+  * For Scala 2.12, you must pass `-Xsource:2.13` to scalac for methods and
+  * conversions to be automatically found.  You must also `import
+  * scalaz.syntax.bind._` or similar for Future methods like `map`, `flatMap`,
+  * and so on.
+  *
+  * There are no constraints on the `EC` type variable; you need only declare
+  * types you wish to use for it that are sufficient for describing the domains
+  * in which you want ExecutionContexts to be discriminated.  These types will
+  * never be instantiated, so you can simply declare that they exist.  They can
+  * be totally separate, or have subtyping relationships; any subtyping
+  * relationships they have will be reflected in equivalent subtyping
+  * relationships between the resulting `ExecutionContext`s; if you declare
+  * `sealed trait Elephant extends Animal`, then automatically
+  * `ExecutionContext[Elephant] <: ExecutionContext[Animal]` with scalac
+  * preferring the former when available (because it is "more specific").  They
+  * can even be singleton types, so you might use `x.type` to suggest that the
+  * context is associated with the exact value of the `x` variable.
+  *
+  * If you want to, say, refer to both [[sc.Future]] and [[concurrent.Future]]
+  * in the same file, we recommend importing *the containing package* with an
+  * alias rather than renaming each individual class you import.  For example,
+  *
+  * {{{
+  *   import com.daml.concurrent._
+  *   import scala.{concurrent => sc}
+  *   // OR
+  *   import scala.concurrent._
+  *   import com.daml.{concurrent => dc}
+  * }}}
+  *
+  * The exact name isn't important, but you should pick a short one that is
+  * sufficiently suggestive for you.
+  *
+  * You should always be able to remove the substring `Of.Instance.T` from any
+  * inferred type; we strongly suggest doing this for clarity.
+  *
+  * Demonstrations of the typing behavior can be found in FutureSpec and
+  * ExecutionContextSpec.  This library has no interesting runtime
+  * characteristics; you should think of it as exactly like `scala.concurrent`
+  * in that regard.
+  */
+package object concurrent {
+
+  /** Like [[scala.concurrent.Future]] but with an extra type parameter indicating
+    * which [[ExecutionContext]] should be used for `map`, `flatMap` and other
+    * operations.
+    */
+  type Future[-EC, +A] = FutureOf.Instance.T[EC, A]
+
+  /** A subtype of [[sc.ExecutionContext]], more specific as `P` gets more
+    * specific.
+    */
+  type ExecutionContext[+P] = ExecutionContextOf.Instance.T[P]
+}
+
+// keeping the companions with the same-named type aliases in same file
+package concurrent {
+
+  object Future {
+
+    /** {{{
+      *  Future[MyECTag] { expr }
+      * }}}
+      *
+      * returns `Future[MyECTag, E]` where `E` is `expr`'s inferred type and
+      * `ExecutionContext[MyECTag]` is required implicitly.
+      */
+    def apply[EC]: apply[EC] =
+      new apply(())
+
+    final class apply[EC](private val ignore: Unit) extends AnyVal {
+      def apply[A](body: => A)(implicit ec: ExecutionContext[EC]): Future[EC, A] =
+        sc.Future(body)(ec)
+    }
+
+    def fromTry[EC, A](result: Try[A]): Future[EC, A] =
+      sc.Future.fromTry(result)
+  }
+
+  object ExecutionContext {
+
+    /** Explicitly tag an [[sc.ExecutionContext]], or replace the tag on an
+      * [[ExecutionContext]].
+      */
+    def apply[EC](ec: sc.ExecutionContext): ExecutionContext[EC] =
+      ExecutionContextOf.Instance.subst[Id, EC](ec)
+  }
+}

libs-scala/concurrent/src/test/scala/concurrent/ExecutionContextSpec.scala

@@ -0,0 +1,120 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.concurrent
+
+import scala.{concurrent => sc}
+
+import com.github.ghik.silencer.silent
+import org.scalatest.{WordSpec, Matchers}
+import shapeless.test.illTyped
+
+@SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements"))
+@silent("Unused import")
+class ExecutionContextSpec extends WordSpec with Matchers {
+  import ExecutionContextSpec._
+
+  // In these tests, you can think of the type argument to `theEC` as being like
+  // the EC on a Future whose ExecutionContext lookup behavior you are wondering
+  // about; the implicit resolution behaves the same.
+
+  "importing only untyped" should {
+    import TestImplicits.untyped
+
+    "disallow lookup" in {
+      illTyped("theEC[Animal]", "could not find implicit value.*")
+    }
+
+    "disallow lookup, even of Any" in {
+      illTyped("theEC[Any]", "could not find implicit value.*")
+    }
+
+    "allow lookup only for untyped" in {
+      implicitly[sc.ExecutionContext] should ===(untyped)
+    }
+  }
+
+  "importing supertype and subtype" should {
+    import TestImplicits.{animal1, Elephant}
+
+    "always prefer the subtype" in {
+      theEC[Any] should ===(Elephant)
+      theEC[Animal] should ===(Elephant)
+      theEC[Elephant] should ===(Elephant)
+    }
+
+    "refuse to resolve a separate subtype" in {
+      illTyped("theEC[Cat]", "could not find implicit value.*")
+    }
+  }
+
+  "importing everything" should {
+    import TestImplicits._
+
+    "always prefer Nothing" in {
+      theEC[Any] should ===(nothing)
+      theEC[Animal] should ===(nothing)
+    }
+  }
+
+  "importing two types with LUB, one lower in hierarchy" should {
+    import TestImplicits.{Elephant, Tabby}
+
+    "consider neither more specific" in {
+      illTyped("theEC[Animal]", "ambiguous implicit values.*")
+    }
+  }
+
+  "importing a type and a related singleton type" should {
+    import TestImplicits.{Tabby, chiefMouserEC}
+
+    "prefer the singleton" in {
+      theEC[Tabby] should ===(chiefMouserEC)
+      theEC[ChiefMouser.type] should ===(chiefMouserEC)
+    }
+  }
+
+  "using intersections" should {
+    import TestImplicits.{Elephant, Cat, cryptozoology}
+
+    "prefer the intersection" in {
+      theEC[Elephant] should ===(cryptozoology)
+      theEC[Cat] should ===(cryptozoology)
+    }
+
+    "be symmetric" in {
+      theEC[Elephant with Cat] should ===(cryptozoology)
+      theEC[Cat with Elephant] should ===(cryptozoology)
+    }
+  }
+}
+
+object ExecutionContextSpec {
+  def theEC[EC](implicit ec: ExecutionContext[EC]): ec.type = ec
+
+  def fakeEC[EC](name: String): ExecutionContext[EC] =
+    ExecutionContext(new sc.ExecutionContext {
+      override def toString = s"<the $name fakeEC>"
+      override def execute(runnable: Runnable) = sys.error("never use this")
+      override def reportFailure(cause: Throwable) = sys.error("could never have failed")
+    })
+
+  sealed trait Animal
+  sealed trait Elephant extends Animal
+  sealed trait Cat extends Animal
+  sealed trait Tabby extends Cat
+  val ChiefMouser: Tabby = new Tabby {}
+
+  object TestImplicits {
+    implicit val untyped: sc.ExecutionContext = fakeEC[Any]("untyped")
+    implicit val any: ExecutionContext[Any] = fakeEC[Any]("any")
+    implicit val animal1: ExecutionContext[Animal] = fakeEC("animal1")
+    implicit val animal2: ExecutionContext[Animal] = fakeEC("animal2")
+    implicit val Elephant: ExecutionContext[Elephant] = fakeEC("Elephant")
+    implicit val Cat: ExecutionContext[Cat] = fakeEC("Cat")
+    implicit val cryptozoology: ExecutionContext[Elephant with Cat] = fakeEC("cryptozoology")
+    implicit val Tabby: ExecutionContext[Tabby] = fakeEC("Tabby")
+    implicit val chiefMouserEC: ExecutionContext[ChiefMouser.type] = fakeEC("chiefMouserEC")
+    implicit val nothing: ExecutionContext[Nothing] = fakeEC("Nothing")
+  }
+}

libs-scala/concurrent/src/test/scala/concurrent/FutureSpec.scala

@@ -0,0 +1,81 @@
+// Copyright (c) 2020 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.concurrent
+
+import scala.{concurrent => sc}
+
+import com.github.ghik.silencer.silent
+import org.scalatest.{WordSpec, Matchers}
+import shapeless.test.illTyped
+
+@SuppressWarnings(Array("org.wartremover.warts.NonUnitStatements"))
+@silent("local method example")
+class FutureSpec extends WordSpec with Matchers {
+  import ExecutionContextSpec._
+
+  val elephantVal = 3000
+  val catVal = 9
+  val untypedVal = -1
+
+  val someElephantFuture: Future[Elephant, Int] = sc.Future successful elephantVal
+  val someCatFuture: Future[Cat, Int] = sc.Future successful catVal
+  val someUntypedFuture: sc.Future[Int] = sc.Future successful untypedVal
+
+  // we repeat imports below to show exactly what imports are needed for a given
+  // scenario.  Naturally, in real code, you would not be so repetitive.
+
+  "an untyped future" can {
+    "be flatmapped to by any future" in {
+      import scalaz.syntax.bind._, TestImplicits.Elephant
+      def example = someElephantFuture flatMap (_ => someUntypedFuture)
+    }
+
+    "simply become a typed future" in {
+      def example: Future[Cat, Int] = someUntypedFuture
+    }
+  }
+
+  "a well-typed future" should {
+    "not lose its type to conversion" in {
+      illTyped(
+        "someCatFuture: sc.Future[Int]",
+        "type mismatch.*found.*daml.concurrent.Future.*required: scala.concurrent.Future.*")
+    }
+  }
+
+  "two unrelated futures" should {
+    "mix their types if zipped" in {
+      // putting in Set (an invariant context) lets us check the inferred type
+      def example = Set(someElephantFuture zip someCatFuture)
+      example: Set[Future[Elephant with Cat, (Int, Int)]]
+    }
+
+    "disallow mixing in flatMap" in {
+      import scalaz.syntax.bind._, TestImplicits.Elephant
+      illTyped(
+        "someElephantFuture flatMap (_ => someCatFuture)",
+        "type mismatch.*found.*Cat.*required.*Elephant.*")
+    }
+
+    "allow mixing in flatMap if requirement changed first" in {
+      import scalaz.syntax.bind._, TestImplicits.Cat
+      def example =
+        someElephantFuture
+          .changeExecutionContext[Cat]
+          .flatMap(_ => someCatFuture)
+    }
+
+    "continue a chain after requirements tightened" in {
+      import scalaz.syntax.bind._, TestImplicits.cryptozoology
+      def example =
+        someElephantFuture
+          .require[Elephant with Cat]
+          .flatMap(_ => someCatFuture)
+    }
+
+    "disallow require on unrelated types" in {
+      illTyped("someElephantFuture.require[Cat]", "type arguments.*do not conform.*")
+    }
+  }
+}

release/artifacts.yaml

@@ -153,3 +153,5 @@
   type: jar-scala
 - target: //libs-scala/scala-utils:scala-utils
   type: jar-scala
+- target: //libs-scala/concurrent:concurrent
+  type: jar-scala