add graceful shutdown to http servers (#1728)

Author: tim-smart

.changeset/sweet-donuts-bet.md

@@ -0,0 +1,5 @@
+---
+"effect": patch
+---
+
+add graceful shutdown to http servers

packages/effect/src/unstable/http/HttpServer.ts

@@ -36,6 +36,7 @@ export class HttpServer extends ServiceMap.Service<HttpServer, {
       Exclude<R, HttpServerRequest> | Scope.Scope
     >
   }
+
   readonly address: Address
 }>()("effect/http/HttpServer") {}
 
@@ -115,10 +116,7 @@ export const serve: {
   HttpServer | Exclude<Effect.Services<App>, HttpServerRequest | Scope.Scope>
 > =>
   Layer.effectDiscard(
-    Effect.flatMap(
-      HttpServer.asEffect(),
-      (server) => server.serve(effect, middleware!)
-    )
+    HttpServer.use((server) => server.serve(effect, middleware!))
   ) as any)
 
 /**
@@ -156,11 +154,7 @@ export const serveEffect: {
   void,
   never,
   Scope.Scope | HttpServer | Exclude<Effect.Services<App>, HttpServerRequest>
-> =>
-  Effect.flatMap(
-    HttpServer.asEffect(),
-    (server) => server.serve(effect, middleware!)
-  ) as any)
+> => HttpServer.use((server) => server.serve(effect, middleware!)) as any)
 
 /**
  * @since 4.0.0

packages/platform-bun/src/BunHttpServer.ts

@@ -5,6 +5,7 @@ import type { Server as BunServer, ServerWebSocket } from "bun"
 import * as Config from "effect/Config"
 import type { ConfigError } from "effect/Config"
 import * as Deferred from "effect/Deferred"
+import * as Duration from "effect/Duration"
 import * as Effect from "effect/Effect"
 import * as Exit from "effect/Exit"
 import * as Fiber from "effect/Fiber"
@@ -17,7 +18,7 @@ import * as Option from "effect/Option"
 import type * as Path from "effect/Path"
 import type * as Record from "effect/Record"
 import type * as Schema from "effect/Schema"
-import type * as Scope from "effect/Scope"
+import * as Scope from "effect/Scope"
 import * as Semaphore from "effect/Semaphore"
 import * as ServiceMap from "effect/ServiceMap"
 import * as Stream from "effect/Stream"
@@ -59,8 +60,12 @@ export type ServeOptions<R extends string> =
  */
 export const make = Effect.fnUntraced(
   function*<R extends string>(
-    options: ServeOptions<R>
+    options: ServeOptions<R> & {
+      readonly disablePreemptiveShutdown?: boolean | undefined
+      readonly gracefulShutdownTimeout?: Duration.Input | undefined
+    }
   ) {
+    const scope = yield* Effect.scope
     const handlerStack: Array<(request: Request, server: BunServer<WebSocketContext>) => Response | Promise<Response>> =
       [
         function(_request, _server) {
@@ -92,14 +97,24 @@ export const make = Effect.fnUntraced(
       }
     })
 
-    yield* Effect.addFinalizer(() => Effect.promise(() => server.stop()))
+    const shutdown = yield* Effect.promise(() => server.stop()).pipe(
+      Effect.cached
+    )
+    const preemptiveShutdown = options.disablePreemptiveShutdown ? Effect.void : Effect.timeoutOrElse(shutdown, {
+      duration: options.gracefulShutdownTimeout ?? Duration.seconds(20),
+      onTimeout: () => Effect.void
+    })
+
+    yield* Scope.addFinalizer(scope, shutdown)
 
     return Server.make({
       address: { _tag: "TcpAddress", port: server.port!, hostname: server.hostname! },
       serve: Effect.fnUntraced(function*(httpApp, middleware) {
         const parent = yield* Effect.fiber
-        const scope = yield* Effect.scope
         const services = parent.services
+        const serveScope = ServiceMap.getUnsafe(services, Scope.Scope)
+        const scope = Scope.forkUnsafe(serveScope, "parallel")
+
         const httpEffect = HttpEffect.toHandled(httpApp, (request, response) =>
           Effect.sync(() => {
             ;(request as BunServerRequest).resolve(makeResponse(request, response, services, scope))
@@ -119,17 +134,13 @@ export const make = Effect.fnUntraced(
           })
         }
 
-        yield* Effect.acquireRelease(
-          Effect.sync(() => {
-            handlerStack.push(handler)
-            server.reload({ fetch: handler })
-          }),
-          () =>
-            Effect.sync(() => {
-              handlerStack.pop()
-              server.reload({ fetch: handlerStack[handlerStack.length - 1] })
-            })
-        )
+        yield* Scope.addFinalizerExit(serveScope, () => {
+          handlerStack.pop()
+          server.reload({ fetch: handlerStack[handlerStack.length - 1] })
+          return preemptiveShutdown
+        })
+        handlerStack.push(handler)
+        server.reload({ fetch: handler })
       })
     })
   }
@@ -202,7 +213,10 @@ const makeResponse = (
  * @category Layers
  */
 export const layerServer: <R extends string>(
-  options: ServeOptions<R>
+  options: ServeOptions<R> & {
+    readonly disablePreemptiveShutdown?: boolean | undefined
+    readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  }
 ) => Layer.Layer<Server.HttpServer> = flow(make, Layer.effect(Server.HttpServer)) as any
 
 /**
@@ -224,7 +238,10 @@ export const layerHttpServices: Layer.Layer<
  * @category Layers
  */
 export const layer = <R extends string>(
-  options: ServeOptions<R>
+  options: ServeOptions<R> & {
+    readonly disablePreemptiveShutdown?: boolean | undefined
+    readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  }
 ): Layer.Layer<
   | Server.HttpServer
   | HttpPlatform
@@ -250,7 +267,12 @@ export const layerTest: Layer.Layer<
  * @category Layers
  */
 export const layerConfig = <R extends string>(
-  options: Config.Wrap<ServeOptions<R>>
+  options: Config.Wrap<
+    ServeOptions<R> & {
+      readonly disablePreemptiveShutdown?: boolean | undefined
+      readonly gracefulShutdownTimeout?: Duration.Input | undefined
+    }
+  >
 ): Layer.Layer<
   Server.HttpServer | HttpPlatform | FileSystem.FileSystem | Etag.Generator | Path.Path,
   ConfigError

packages/platform-node/src/NodeHttpServer.ts

@@ -3,6 +3,7 @@
  */
 import * as Cause from "effect/Cause"
 import * as Config from "effect/Config"
+import * as Duration from "effect/Duration"
 import * as Effect from "effect/Effect"
 import * as Fiber from "effect/Fiber"
 import type * as FileSystem from "effect/FileSystem"
@@ -56,25 +57,35 @@ import { NodeWS } from "./NodeSocket.ts"
  */
 export const make = Effect.fnUntraced(function*(
   evaluate: LazyArg<Http.Server>,
-  options: Net.ListenOptions
+  options: Net.ListenOptions & {
+    readonly disablePreemptiveShutdown?: boolean | undefined
+    readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  }
 ) {
   const scope = yield* Effect.scope
   const server = evaluate()
-  yield* Scope.addFinalizer(
-    scope,
-    Effect.callback<void>((resume) => {
-      if (!server.listening) {
-        return resume(Effect.void)
+
+  const shutdown = yield* Effect.callback<void>((resume) => {
+    if (!server.listening) {
+      return resume(Effect.void)
+    }
+    server.close((error) => {
+      if (error) {
+        resume(Effect.die(error))
+      } else {
+        resume(Effect.void)
       }
-      server.close((error) => {
-        if (error) {
-          resume(Effect.die(error))
-        } else {
-          resume(Effect.void)
-        }
-      })
     })
-  )
+  }).pipe(Effect.cached)
+
+  const preemptiveShutdown = options.disablePreemptiveShutdown ?
+    Effect.void :
+    Effect.timeoutOrElse(shutdown, {
+      duration: options.gracefulShutdownTimeout ?? Duration.seconds(20),
+      onTimeout: () => Effect.void
+    })
+
+  yield* Scope.addFinalizer(scope, shutdown)
 
   yield* Effect.callback<void, ServeError>((resume) => {
     function onError(cause: Error) {
@@ -112,7 +123,8 @@ export const make = Effect.fnUntraced(function*(
         port: address.port
       },
     serve: Effect.fnUntraced(function*(httpApp, middleware) {
-      const scope = yield* Effect.scope
+      const serveScope = yield* Effect.scope
+      const scope = Scope.forkUnsafe(serveScope, "parallel")
       const handler = yield* (makeHandler(httpApp, {
         middleware: middleware as any,
         scope
@@ -121,12 +133,11 @@ export const make = Effect.fnUntraced(function*(
         middleware: middleware as any,
         scope
       })
-      yield* Effect.addFinalizer(() =>
-        Effect.sync(() => {
-          server.off("request", handler)
-          server.off("upgrade", upgradeHandler)
-        })
-      )
+      yield* Scope.addFinalizerExit(serveScope, () => {
+        server.off("request", handler)
+        server.off("upgrade", upgradeHandler)
+        return preemptiveShutdown
+      })
       server.on("request", handler)
       server.on("upgrade", upgradeHandler)
     })
@@ -360,7 +371,10 @@ class ServerRequestImpl extends NodeHttpIncomingMessage<HttpServerError> impleme
  */
 export const layerServer: (
   evaluate: LazyArg<Http.Server<typeof Http.IncomingMessage, typeof Http.ServerResponse>>,
-  options: Net.ListenOptions
+  options: Net.ListenOptions & {
+    readonly disablePreemptiveShutdown?: boolean | undefined
+    readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  }
 ) => Layer.Layer<HttpServer.HttpServer, ServeError> = flow(make, Layer.effect(HttpServer.HttpServer))
 
 /**
@@ -381,7 +395,10 @@ export const layerHttpServices: Layer.Layer<
  */
 export const layer = (
   evaluate: LazyArg<Http.Server>,
-  options: Net.ListenOptions
+  options: Net.ListenOptions & {
+    readonly disablePreemptiveShutdown?: boolean | undefined
+    readonly gracefulShutdownTimeout?: Duration.Input | undefined
+  }
 ): Layer.Layer<
   HttpServer.HttpServer | NodeServices.NodeServices | HttpPlatform.HttpPlatform | Etag.Generator,
   ServeError
@@ -397,7 +414,12 @@ export const layer = (
  */
 export const layerConfig = (
   evaluate: LazyArg<Http.Server>,
-  options: Config.Wrap<Net.ListenOptions>
+  options: Config.Wrap<
+    Net.ListenOptions & {
+      readonly disablePreemptiveShutdown?: boolean | undefined
+      readonly gracefulShutdownTimeout?: Duration.Input | undefined
+    }
+  >
 ): Layer.Layer<
   HttpServer.HttpServer | FileSystem.FileSystem | Path.Path | HttpPlatform.HttpPlatform | Etag.Generator,
   ServeError | Config.ConfigError