Provide a Scope per server call (#3197) (#3276)

* Provide a `Scope` per server call (#3197)

* Scope handling for handlers (#3197)

* Scope handling for handlers (#3197)

docs/reference/aop/middleware.md

@@ -195,14 +195,17 @@ val urlRewrite: Middleware[Any] =
   new Middleware[Any] {
     override def apply[Env1 <: Any, Err](routes: Routes[Env1, Err]): Routes[Env1, Err] =
       routes.transform { handler =>
-        Handler.fromFunctionZIO { request =>
-          handler(
-            request.updateURL(url =>
-              if (url.path.startsWith(Path("/api")))
-                url.copy(path = Path("/v1") ++ url.path)
-              else url,
-            ),
-          )
+        Handler.scoped[Env1] {
+          Handler.fromFunctionZIO { request =>
+            handler(
+              request.updateURL(
+                url =>
+                  if (url.path.startsWith(Path("/api")))
+                    url.copy(path = Path("/v1") ++ url.path)
+                  else url,
+                ),
+              )
+          }
         }
       }
   }

docs/reference/aop/protocol-stack.md

@@ -88,7 +88,7 @@ The behavior of the `handler` remains the same. Let's test it:
 ```scala mdoc
 Unsafe.unsafe{ implicit unsafe =>
   Runtime.default.unsafe.run(
-    handler("Hello World!")
+    ZIO.scoped(handler("Hello World!"))
   )
 }
 ```
@@ -112,7 +112,7 @@ val newHandler: Handler[Any, (Response, Int), Request, (Response, Int)] =
 
 Unsafe.unsafe { implicit unsafe =>
   Runtime.default.unsafe.run(
-    newHandler("Hello World! "),
+    ZIO.scoped(newHandler("Hello World! ")),
   )
 }
 ```
@@ -141,7 +141,7 @@ Now, let's apply the `uppercase` handler to the `anotherTrimAndLength` stack:
 ```scala mdoc
 Unsafe.unsafe { implicit unsafe =>
   Runtime.default.unsafe.run(
-    anotherTrimAndLength(uppercase).apply("Hello World!"),
+    ZIO.scoped(anotherTrimAndLength(uppercase).apply("Hello World!")),
   )
 }
 ```
@@ -210,7 +210,7 @@ Now, we are ready to test the `responseTime` protocol stack by applying the `han
 ```scala mdoc
 Unsafe.unsafe { implicit unsafe =>
   Runtime.default.unsafe.run(
-    responseTime(handler).apply("Hello, World!").debug("Response along with its latency"),
+    ZIO.scoped(responseTime(handler).apply("Hello, World!").debug("Response along with its latency")),
   )
 }
 ```

docs/reference/routing/routes.md

@@ -141,7 +141,7 @@ val routes: Routes[Any, Response] = ???
 
 routes.transform[Any] { handle =>
    handler { (request: Request) => 
-     ZIO.sleep(1.second) *> handle(request)
+     ZIO.sleep(1.second) *> ZIO.scoped(handle(request))
    }
 }
 ```

project/MimaSettings.scala

@@ -25,6 +25,7 @@ object MimaSettings {
         exclude[Problem]("zio.http.endpoint.openapi.OpenAPIGen#AtomizedMetaCodecs.copy"),
         exclude[IncompatibleMethTypeProblem]("zio.http.Middleware.addHeader"),
         exclude[IncompatibleMethTypeProblem]("zio.http.HandlerAspect.addHeader"),
+        ProblemFilters.exclude[ReversedMissingMethodProblem]("zio.http.Server.installInternal")
       ),
       mimaFailOnProblem := failOnProblem,
     )

zio-http-benchmarks/src/main/scala-2.13/zio/http/benchmarks/EndpointBenchmark.scala

@@ -674,15 +674,15 @@ class EndpointBenchmark {
       http4sRequestFromString(url),
     )
 
-  def unsafeRun[E, A](zio: ZIO[Any, E, A]): Unit = Unsafe.unsafe { implicit unsafe =>
+  def unsafeRun[E, A](z: ZIO[zio.Scope, E, A]): Unit = Unsafe.unsafe { implicit unsafe =>
     Runtime.default.unsafe
-      .run(zio.unit)
+      .run(ZIO.scoped(z.unit))
       .getOrThrowFiberFailure()
   }
 
-  private def unsafeRunResult[E, A](zio: ZIO[Any, E, A]): A = Unsafe.unsafe { implicit unsafe =>
+  private def unsafeRunResult[E, A](z: ZIO[zio.Scope, E, A]): A = Unsafe.unsafe { implicit unsafe =>
     Runtime.default.unsafe
-      .run(zio)
+      .run(ZIO.scoped(z))
       .getOrThrowFiberFailure()
   }
 

zio-http-benchmarks/src/main/scala/zhttp.benchmarks/HttpRouteTextPerf.scala

@@ -18,7 +18,7 @@ class HttpRouteTextPerf {
   private val res          = Response.text("HELLO WORLD")
   private val app          = Handler.succeed(res)
   private val req: Request = Request.get(URL(Path.root))
-  private val httpProgram  = ZIO.foreachDiscard(0 to 1000) { _ => app(req) }
+  private val httpProgram  = ZIO.scoped(ZIO.foreachDiscard(0 to 1000) { _ => app(req) })
   private val UIOProgram   = ZIO.foreachDiscard(0 to 1000) { _ => ZIO.succeed(res) }
 
   @Benchmark

zio-http-example/src/main/scala/example/middleware/CounterProtocolStackExample.scala

@@ -28,5 +28,5 @@ object CounterProtocolStackExample extends ZIOAppDefault {
     _ <- handler("What is ZIO?").debug
   } yield ()
 
-  def run = app.provide(ZLayer.fromZIO(Ref.make(0L)))
+  def run = app.provide(Scope.default, ZLayer.fromZIO(Ref.make(0L)))
 }

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

@@ -119,7 +119,7 @@ final case class TestClient(
     body: Body,
     sslConfig: Option[zio.http.ClientSSLConfig],
     proxy: Option[Proxy],
-  )(implicit trace: Trace): ZIO[Any, Throwable, Response] = {
+  )(implicit trace: Trace): ZIO[Scope, Throwable, Response] = {
     for {
       currentBehavior <- behavior.get.map(_ :+ Method.ANY / trailing -> handler(Response.notFound))
       request = Request(

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 private[http] def installInternal[R](routes: Routes[R, Response])(implicit
     trace: zio.Trace,
     tag: EnvironmentTag[R],
   ): URIO[R, Unit] =

zio-http-testkit/src/test/scala/zio/http/SocketContractSpec.scala

@@ -129,7 +129,7 @@ object SocketContractSpec extends ZIOHttpSpec {
     ZIO.serviceWithZIO[Server](server =>
       for {
         p    <- Promise.make[Throwable, Unit]
-        _    <- server.install(serverApp(p).toRoutes)
+        _    <- server.installInternal(serverApp(p).toRoutes)
         port <- server.port
       } yield (port, p),
     )

zio-http/js/src/test/scala/zio/http/JSClientSpec.scala

@@ -61,5 +61,5 @@ object JSClientSpec extends ZIOSpecDefault {
 //          } yield assertTrue(consoleMessages.contains("Server: Hello, World!"))
 //        }.provideSome[Scope & Client](ZLayer(Queue.bounded[String](100))),
 //      ),
-    )
+    ) @@ flaky
 }

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

@@ -206,7 +206,7 @@ private[zio] final case class ServerInboundHandler(
   private def attemptImmediateWrite(
     ctx: ChannelHandlerContext,
     method: Method,
-    exit: ZIO[Any, Response, Response],
+    exit: ZIO[Scope, Response, Response],
   ): Boolean = {
     exit match {
       case Exit.Success(response) if response ne null =>
@@ -282,7 +282,7 @@ private[zio] final case class ServerInboundHandler(
         ZIO.suspend {
           val nettyChannel     = NettyChannel.make[JWebSocketFrame](ctx.channel())
           val webSocketChannel = WebSocketChannel.make(nettyChannel, queue, handshakeCompleted)
-          webSocketApp.handler.runZIO(webSocketChannel).ignoreLogged.forkDaemon
+          ZIO.scoped(webSocketApp.handler.runZIO(webSocketChannel)).ignoreLogged.forkDaemon
         }
       }
     _                  <- ZIO.attempt {
@@ -311,7 +311,7 @@ private[zio] final case class ServerInboundHandler(
   private def writeResponse(
     ctx: ChannelHandlerContext,
     runtime: NettyRuntime,
-    exit: ZIO[Any, Response, Response],
+    exit: ZIO[Scope, Response, Response],
     req: Request,
   )(ensured: () => Unit): Unit = {
 
@@ -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/ClientConnectionSpec.scala

@@ -41,6 +41,7 @@ object ClientConnectionSpec extends RoutesRunnableSpec {
       serve.as(tests)
     }
       .provideShared(
+        Scope.default,
         DynamicServer.live,
         serverTestLayer,
         Client.live,

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

@@ -135,6 +135,6 @@ object ClientSpec extends RoutesRunnableSpec {
   override def spec = {
     suite("Client") {
       serve.as(List(clientSpec))
-    }.provideShared(DynamicServer.live, serverTestLayer, Client.default) @@ sequential @@ withLiveClock
+    }.provideShared(Scope.default, DynamicServer.live, serverTestLayer, Client.default) @@ sequential @@ withLiveClock
   }
 }

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

@@ -76,6 +76,6 @@ object ContentTypeSpec extends RoutesRunnableSpec {
   override def spec = {
     suite("Content-type") {
       serve.as(List(contentSpec))
-    }.provideShared(DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
+    }.provideShared(Scope.default, DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
   }
 }

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

@@ -133,7 +133,7 @@ object FlashSpec extends ZIOHttpSpec {
           )
           bodyString <- response2.body.asString
         } yield assertTrue(bodyString.contains("successfully") && bodyString.contains("green"))
-      }.provideLayer(Flash.Backend.inMemory),
+      }.provide(Scope.default, Flash.Backend.inMemory),
     )
 
 }

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

@@ -23,7 +23,7 @@ object HandlerAspectSpec extends ZIOSpecDefault {
           //can't be infix because of Scala 3
         }.@@[Boolean](handlerAspect)
         for {
-          response   <- handler0(Request(headers = Headers("accept", "*"))).provideEnvironment(ZEnvironment(true))
+          response   <- ZIO.scoped(handler0(Request(headers = Headers("accept", "*")))).provideEnvironment(ZEnvironment(true))
           bodyString <- response.body.asString
         } yield assertTrue(bodyString == "1")
       },
@@ -41,7 +41,7 @@ object HandlerAspectSpec extends ZIOSpecDefault {
           //can't be infix because of Scala 3
         }.@@[Boolean](handlerAspect)
         for {
-          response   <- handler0(Request(headers = Headers("accept", "*"))).provideEnvironment(ZEnvironment(true) ++ ZEnvironment("test"))
+          response   <- ZIO.scoped(handler0(Request(headers = Headers("accept", "*")))).provideEnvironment(ZEnvironment(true) ++ ZEnvironment("test"))
           bodyString <- response.body.asString
         } yield assertTrue(bodyString == "1 test")
       },

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

@@ -73,6 +73,7 @@ object HybridRequestStreamingServerSpec extends RoutesRunnableSpec {
         appWithHybridReqStreaming.as(List(requestBodySpec, hybridStreamingServerSpec))
       }
     }.provideShared(
+      Scope.default,
       DynamicServer.live,
       ZLayer.succeed(configAppWithHybridRequestStreaming),
       Server.customized,

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

@@ -66,7 +66,7 @@ object KeepAliveSpec extends RoutesRunnableSpec {
   override def spec: Spec[Any, Throwable] = {
     suite("KeepAliveSpec") {
       keepAliveSpec
-    }.provide(DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
+    }.provide(Scope.default, DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
   }
 
 }

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

@@ -87,7 +87,7 @@ object ResponseCompressionSpec extends ZIOHttpSpec {
         for {
           server       <- ZIO.service[Server]
           client       <- ZIO.service[Client]
-          _            <- server.install(app)
+          _            <- server.installInternal(app)
           port         <- server.port
           response     <- client.batched(
             Request(
@@ -110,7 +110,7 @@ object ResponseCompressionSpec extends ZIOHttpSpec {
         for {
           server       <- ZIO.service[Server]
           client       <- ZIO.service[Client]
-          _            <- server.install(app)
+          _            <- server.installInternal(app)
           port         <- server.port
           response     <- client.batched(
             Request(
@@ -134,7 +134,7 @@ object ResponseCompressionSpec extends ZIOHttpSpec {
     for {
       server       <- ZIO.service[Server]
       client       <- ZIO.service[Client]
-      _            <- server.install(app)
+      _            <- server.installInternal(app)
       port         <- server.port
       response     <- client.batched(
         Request(

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

@@ -77,5 +77,5 @@ object RoutesMiddlewareSpec extends ZIOHttpSpec with ExitAssertion {
           result2 <- ref.get
         } yield assertTrue(result1 == 1, result2 == 1)
       },
-    )
+    ).provide(Scope.default)
 }

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/ServerSpec.scala

@@ -524,6 +524,7 @@ object ServerSpec extends RoutesRunnableSpec {
       val spec = dynamicAppSpec + responseSpec + requestSpec + requestBodySpec + serverErrorSpec
       suite("app without request streaming") { app.as(List(spec)) }
     }.provideShared(
+      Scope.default,
       DynamicServer.live,
       ZLayer.succeed(configApp),
       Server.customized,

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

@@ -46,7 +46,7 @@ object StaticFileServerSpec extends RoutesRunnableSpec {
 
   override def spec = suite("StaticFileServerSpec") {
     serve.as(List(staticSpec))
-  }.provideShared(DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
+  }.provideShared(Scope.default, DynamicServer.live, serverTestLayer, Client.default) @@ withLiveClock @@ sequential
 
   private def staticSpec = suite("Static RandomAccessFile Server")(
     suite("fromResource")(

zio-http/jvm/src/test/scala/zio/http/endpoint/EndpointSpec.scala

@@ -53,14 +53,14 @@ object EndpointSpec extends ZIOHttpSpec {
   def testEndpoint[R](service: Routes[R, Nothing])(
     url: String,
     expected: String,
-  ): ZIO[R, Response, TestResult] =
+  ): ZIO[Scope & R, Response, TestResult] =
     testEndpointWithHeaders(service)(url, headers = List.empty, expected)
 
   def testEndpointWithHeaders[R](service: Routes[R, Nothing])(
     url: String,
     headers: List[(String, String)],
     expected: String,
-  ): ZIO[R, Response, TestResult] = {
+  ): ZIO[Scope & R, Response, TestResult] = {
     val request = Request
       .get(url = URL.decode(url).toOption.get)
       .addHeaders(headers.foldLeft(Headers.empty) { case (hs, (k, v)) => hs ++ Headers(k, v) })