Provide a Scope per server call (zio#3197)

project/Dependencies.scala

@@ -5,7 +5,7 @@ object Dependencies {
   val NettyVersion                  = "4.1.117.Final"
   val NettyIncubatorVersion         = "0.0.25.Final"
   val ScalaCompactCollectionVersion = "2.12.0"
-  val ZioVersion                    = "2.1.11"
+  val ZioVersion                    = "2.1.14"
   val ZioCliVersion                 = "0.5.0"
   val ZioJsonVersion                = "0.7.1"
   val ZioParserVersion              = "0.1.10"
@@ -16,8 +16,8 @@ object Dependencies {
   val `jwt-core`                 = "com.github.jwt-scala"   %% "jwt-core"                % JwtCoreVersion
   val `scala-compact-collection` = "org.scala-lang.modules" %% "scala-collection-compat" % ScalaCompactCollectionVersion
 
-  val scalafmt         = "org.scalameta" %% "scalafmt-dynamic" % "3.8.1"
-  val scalametaParsers = "org.scalameta" %% "parsers"          % "4.9.9"
+  val scalafmt         = "org.scalameta" %% "scalafmt-dynamic" % "3.8.5"
+  val scalametaParsers = "org.scalameta" %% "parsers"          % "4.12.6"
 
   val netty =
     Seq(

zio-http-testkit/src/main/scala/zio/http/TestServer.scala

@@ -98,7 +98,7 @@ final case class TestServer(driver: Driver, bindPort: Int) extends Server {
       _ <- driver.addApp(provided, r)
     } yield ()
 
-  override def install[R](routes: Routes[R, Response])(implicit
+  override def install[R](routes: Routes[Scope & R, Response])(implicit
     trace: zio.Trace,
     tag: EnvironmentTag[R],
   ): URIO[R, Unit] =

zio-http/jvm/src/main/scala/zio/http/netty/server/NettyDriver.scala

@@ -61,7 +61,7 @@ private[zio] final case class NettyDriver(
       )
     } yield StartResult(port, serverInboundHandler.inFlightRequests)
 
-  def addApp[R](newApp: Routes[R, Response], env: ZEnvironment[R])(implicit trace: Trace): UIO[Unit] =
+  override def addApp[R](newApp: Routes[Scope & R, Response], env: ZEnvironment[R])(implicit trace: Trace): UIO[Unit] =
     ZIO.fiberId.map { fiberId =>
       var loop = true
       while (loop) {

zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala

@@ -334,13 +334,16 @@ private[zio] final case class ServerInboundHandler(
         )
       }
 
-    val program = exit.foldCauseZIO(
-      _.failureOrCause match {
-        case Left(resp)                      => writeResponse(resp)
-        case Right(c) if c.isInterruptedOnly => closeChannel()
-        case Right(c)                        => writeResponse(withDefaultErrorResponse(FiberFailure(c)))
-      },
-      writeResponse,
+    val scope   = Scope.unsafe.make
+    val program = scope.use(
+      exit.foldCauseZIO(
+        _.failureOrCause match {
+          case Left(resp)                      => writeResponse(resp)
+          case Right(c) if c.isInterruptedOnly => closeChannel()
+          case Right(c)                        => writeResponse(withDefaultErrorResponse(FiberFailure(c)))
+        },
+        writeResponse,
+      ),
     )
 
     runtime.run(ctx, ensured, preferOnCurrentThread = avoidCtxSwitching)(program)

zio-http/jvm/src/test/scala/zio/http/ServerRuntimeSpec.scala

@@ -20,6 +20,9 @@ import zio._
 import zio.test.TestAspect._
 import zio.test._
 
+import zio.stream.ZStream
+
+import zio.http.codec.HttpContentCodec._
 import zio.http.internal.{DynamicServer, RoutesRunnableSpec}
 import zio.http.netty.NettyConfig
 
@@ -75,10 +78,48 @@ object ServerRuntimeSpec extends RoutesRunnableSpec {
             .scoped(serve[Foo](server))
             .zipRight(server.deploy.body.run(path = Path.root / "test", method = Method.GET))
             .flatMap(_.asString(Charsets.Utf8))
-            .map(b => assertTrue(b == "1"))
+            .map(b => assertTrue(b == "2")) // one extra for Scope
+        } +
+        test("with scope") {
+          val ref    = Ref.unsafe.make(0)(zio.Unsafe)
+          val routes = Routes(
+            Method.GET / "test" -> handler(
+              ZIO.addFinalizer(ref.set(1)).as(Response.text("ok")),
+            ),
+          )
+          serve(routes)
+            .zipRight(routes.deploy.body.run(path = Path.root / "test", method = Method.GET))
+            .flatMap(_.asString(Charsets.Utf8))
+            .map(b => assertTrue(b == "ok")) *> ref.get.map { v => assertTrue(v == 1) }
+        } +
+        test("with scope streaming") {
+          val ref    = Ref.unsafe.make(0)(zio.Unsafe)
+          val routes = Routes(
+            Method.GET / "test" -> handler(
+              Body
+                .fromStreamEnv(
+                  ZStream.fromZIO(
+                    ZIO.addFinalizer(ref.set(1)) *> ref.get.flatMap(v =>
+                      if (v == 0) Exit.succeed(Chunk.fromArray("ok".getBytes)) else Exit.fail(new Exception("error")),
+                    ),
+                  ),
+                )
+                .map(body =>
+                  Response(
+                    body = body,
+                  ),
+                )
+                .orDie,
+            ),
+          )
+          serve(routes)
+            .zipRight(routes.deploy.body.run(path = Path.root / "test", method = Method.GET))
+            .flatMap(_.asString(Charsets.Utf8))
+            .map(b => assertTrue(b == "ok")) *> ref.get.map { v => assertTrue(v == 1) }
         }
     }
       .provide(
+        Scope.default,
         DynamicServer.live,
         Server.customized,
         ZLayer.succeed(Server.Config.default),

zio-http/jvm/src/test/scala/zio/http/internal/RoutesRunnableSpec.scala

@@ -16,7 +16,7 @@
 
 package zio.http.internal
 
-import zio.{EnvironmentTag, Scope, ZIO}
+import zio._
 
 import zio.http.URL.Location
 import zio.http._
@@ -111,7 +111,9 @@ abstract class RoutesRunnableSpec extends ZIOHttpSpec { self =>
       _    <- DynamicServer.setStart(server)
     } yield port
 
-  def serve[R: EnvironmentTag](routes: Routes[R, Response]): ZIO[R with DynamicServer with Server, Nothing, Int] =
+  def serve[R: EnvironmentTag](
+    routes: Routes[Scope & R, Response],
+  ): ZIO[R with DynamicServer with Server, Nothing, Int] =
     for {
       server <- ZIO.service[Server]
       port   <- Server.install(routes)

zio-http/shared/src/main/scala/zio/http/Driver.scala

@@ -26,7 +26,7 @@ import zio.http.Driver.StartResult
 trait Driver {
   def start(implicit trace: Trace): RIO[Scope, StartResult]
 
-  def addApp[R](newRoutes: Routes[R, Response], env: ZEnvironment[R])(implicit trace: Trace): UIO[Unit]
+  def addApp[R](newRoutes: Routes[Scope & R, Response], env: ZEnvironment[R])(implicit trace: Trace): UIO[Unit]
 
   def createClientDriver()(implicit trace: Trace): ZIO[Scope, Throwable, ClientDriver]
 }

zio-http/shared/src/main/scala/zio/http/Server.scala

@@ -32,7 +32,7 @@ trait Server {
   /**
    * Installs the given HTTP application into the server.
    */
-  def install[R](routes: Routes[R, Response])(implicit trace: Trace, tag: EnvironmentTag[R]): URIO[R, Unit]
+  def install[R](routes: Routes[Scope & R, Response])(implicit trace: Trace, tag: EnvironmentTag[R]): URIO[R, Unit]
 
   /**
    * The port on which the server is listening.
@@ -435,7 +435,7 @@ object Server extends ServerPlatformSpecific {
   }
 
   def serve[R](
-    routes: Routes[R, Response],
+    routes: Routes[Scope & R, Response],
   )(implicit trace: Trace, tag: EnvironmentTag[R]): URIO[R with Server, Nothing] = {
     ZIO.logInfo("Starting the server...") *>
       ZIO.serviceWithZIO[Server](_.install[R](routes)) *>
@@ -444,14 +444,14 @@ object Server extends ServerPlatformSpecific {
   }
 
   def serve[R](
-    route: Route[R, Response],
-    routes: Route[R, Response]*,
+    route: Route[Scope & R, Response],
+    routes: Route[Scope & R, Response]*,
   )(implicit trace: Trace, tag: EnvironmentTag[R]): URIO[R with Server, Nothing] = {
     serve(Routes(route, routes: _*))
   }
 
   def install[R](
-    routes: Routes[R, Response],
+    routes: Routes[Scope & R, Response],
   )(implicit trace: Trace, tag: EnvironmentTag[R]): URIO[R with Server, Int] = {
     ZIO.serviceWithZIO[Server](_.install[R](routes)) *> ZIO.serviceWithZIO[Server](_.port)
   }
@@ -523,7 +523,7 @@ object Server extends ServerPlatformSpecific {
     // or a throwable if starting the driver failed for any reason.
     private val serverStarted: Promise[Throwable, Int],
   ) extends Server {
-    override def install[R](routes: Routes[R, Response])(implicit
+    override def install[R](routes: Routes[Scope & R, Response])(implicit
       trace: Trace,
       tag: EnvironmentTag[R],
     ): URIO[R, Unit] =

zio-http/shared/src/main/scala/zio/http/codec/HttpContentCodec.scala

@@ -373,7 +373,7 @@ object HttpContentCodec {
 
   }
 
-  private val ByteChunkBinaryCodec: BinaryCodec[Chunk[Byte]] = new BinaryCodec[Chunk[Byte]] {
+  private[http] implicit val ByteChunkBinaryCodec: BinaryCodec[Chunk[Byte]] = new BinaryCodec[Chunk[Byte]] {
 
     override def encode(value: Chunk[Byte]): Chunk[Byte] = value
 

zio-http/shared/src/main/scala/zio/http/endpoint/Endpoint.scala

@@ -233,22 +233,22 @@ final case class Endpoint[PathInput, Input, Err, Output, Auth <: AuthType](
   def implementEither(f: Input => Either[Err, Output])(implicit
     trace: Trace,
   ): Route[Any, Nothing] =
-    implementHandler[Any](Handler.fromFunctionHandler[Input](in => Handler.fromEither(f(in))))
+    implementHandler[Any](Handler.fromFunctionEither[Input](f))
 
   def implementPurely(f: Input => Output)(implicit
     trace: Trace,
   ): Route[Any, Nothing] =
-    implementHandler[Any](Handler.fromFunctionHandler[Input](in => Handler.succeed(f(in))))
+    implementHandler[Any](Handler.fromFunctionExit[Input](in => Exit.succeed(f(in))))
 
   def implementAs(output: Output)(implicit
     trace: Trace,
   ): Route[Any, Nothing] =
     implementHandler[Any](Handler.succeed(output))
 
-  def implementAsZIO(output: ZIO[Any, Err, Output])(implicit
+  def implementAsZIO[Env](output: ZIO[Env, Err, Output])(implicit
     trace: Trace,
-  ): Route[Any, Nothing] =
-    implementHandler[Any](Handler.fromZIO(output))
+  ): Route[Env, Nothing] =
+    implementHandler(Handler.fromZIO(output))
 
   def implementAsError(err: Err)(implicit
     trace: Trace,