feat(medusa,medusa-core-utils): graceful shutdown server (#3408)

* feat: graceful shutdown

.changeset/slimy-brooms-flash.md

@@ -0,0 +1,7 @@
+---
+"medusa-core-utils": minor
+"medusa-react": minor
+"@medusajs/medusa": minor
+---
+
+Http Server Graceful Shutdown

packages/medusa-core-utils/src/__tests__/graceful-shutdown-server.ts

@@ -0,0 +1,186 @@
+import { GracefulShutdownServer } from "../graceful-shutdown-server"
+
+describe("GracefulShutdownServer", () => {
+  beforeEach(() => {
+    jest.clearAllTimers()
+    jest.clearAllMocks()
+  })
+  afterEach(() => {})
+
+  it('should add "isShuttingDown" property to the existing server', () => {
+    const server = GracefulShutdownServer.create({ on: jest.fn() } as any)
+    expect(server).toHaveProperty("isShuttingDown")
+    expect(server.isShuttingDown).toEqual(false)
+  })
+
+  it("should listen for client connections and store reference to them", async () => {
+    const onEventMock = jest.fn()
+
+    GracefulShutdownServer.create({ on: onEventMock } as any)
+
+    expect(onEventMock).toBeCalledTimes(3)
+
+    expect(onEventMock.mock.calls[2][0]).toEqual("request")
+
+    const connectEvent: (socket) => any = onEventMock.mock.calls[0][1]
+
+    const onSocketClose = jest.fn()
+    const socket = { on: onSocketClose }
+    connectEvent(socket)
+    expect(socket).toEqual(
+      expect.objectContaining({
+        _idle: true,
+        _connectionId: 1,
+      })
+    )
+
+    const socket2 = { on: onSocketClose }
+    connectEvent(socket2)
+    expect(socket2).toEqual(
+      expect.objectContaining({
+        _idle: true,
+        _connectionId: 2,
+      })
+    )
+
+    const requestMock = onEventMock.mock.calls[2][1]
+    expect(typeof requestMock).toEqual("function")
+
+    const socket3 = { on: onSocketClose }
+    const req = { socket: socket3, on: jest.fn() }
+    const res = { on: jest.fn() }
+    connectEvent(socket3)
+    requestMock(req, res)
+
+    const finishRequestMock = res.on.mock.calls[0][1]
+
+    expect(socket3).toEqual(
+      expect.objectContaining({
+        _idle: false,
+        _connectionId: 3,
+      })
+    )
+    finishRequestMock()
+    expect(socket3).toEqual(
+      expect.objectContaining({
+        _idle: true,
+        _connectionId: 3,
+      })
+    )
+
+    expect(onSocketClose).toBeCalledTimes(3)
+    expect(onSocketClose.mock.calls[0][0]).toEqual("close")
+  })
+
+  it("waits requests to complete before shutting the server down", (done: Function) => {
+    jest.useFakeTimers()
+
+    const onEventMock = jest.fn()
+    const setIntervalSpy = jest.spyOn(global, "setInterval")
+    const setTimeoutSpy = jest.spyOn(global, "setTimeout")
+    const clearIntervalSpy = jest.spyOn(global, "clearInterval")
+
+    const waitTime = 200
+    let closeServerCallback: Function
+    const server = GracefulShutdownServer.create(
+      {
+        close: (callback) => {
+          closeServerCallback = callback
+        },
+        on: onEventMock,
+      } as any,
+      waitTime
+    )
+
+    const requestMock = onEventMock.mock.calls[2][1]
+    const connectEvent: (socket) => any = onEventMock.mock.calls[0][1]
+
+    expect(typeof requestMock).toEqual("function")
+
+    const socket = { on: jest.fn(), destroy: jest.fn() }
+    const req = { socket, on: jest.fn() }
+    const res = { on: jest.fn() }
+    connectEvent(socket)
+
+    requestMock(req, res)
+
+    const finishRequestMock = res.on.mock.calls[0][1]
+
+    server.shutdown().then(() => {
+      done()
+    })
+
+    expect(setTimeoutSpy).toBeCalledTimes(0)
+    expect(setIntervalSpy).toBeCalledTimes(1)
+    expect(setIntervalSpy.mock.calls[0][1]).toEqual(waitTime)
+    expect(clearIntervalSpy).toBeCalledTimes(0)
+    expect(socket.destroy).toBeCalledTimes(0)
+
+    jest.advanceTimersByTime(200)
+
+    expect(socket.destroy).toBeCalledTimes(0)
+
+    finishRequestMock()
+
+    expect(socket.destroy).toBeCalledTimes(0)
+
+    jest.advanceTimersByTime(waitTime)
+
+    expect(socket.destroy).toBeCalledTimes(1)
+
+    closeServerCallback!()
+  })
+
+  it("should force close all connections after the timeout is reached", (done: Function) => {
+    jest.useFakeTimers()
+
+    const onEventMock = jest.fn()
+    const setIntervalSpy = jest.spyOn(global, "setInterval")
+    const setTimeoutSpy = jest.spyOn(global, "setTimeout")
+    const clearIntervalSpy = jest.spyOn(global, "clearInterval")
+
+    const waitTime = 300
+    let closeServerCallback: Function
+    const server = GracefulShutdownServer.create(
+      {
+        close: (callback) => {
+          closeServerCallback = callback
+        },
+        on: onEventMock,
+      } as any,
+      waitTime
+    )
+
+    const requestMock = onEventMock.mock.calls[2][1]
+    const connectEvent: (socket) => any = onEventMock.mock.calls[0][1]
+
+    expect(typeof requestMock).toEqual("function")
+
+    const socket = { on: jest.fn(), destroy: jest.fn() }
+    const req = { socket, on: jest.fn() }
+    const res = { on: jest.fn() }
+    connectEvent(socket)
+
+    requestMock(req, res) // pending request
+
+    const forceTimeout = 600
+    server.shutdown(forceTimeout).then(() => {
+      done()
+    })
+
+    expect(setTimeoutSpy).toBeCalledTimes(1)
+    expect(setTimeoutSpy.mock.calls[0][1]).toEqual(forceTimeout)
+    expect(setIntervalSpy).toBeCalledTimes(1)
+    expect(setIntervalSpy.mock.calls[0][1]).toEqual(waitTime)
+    expect(clearIntervalSpy).toBeCalledTimes(0)
+    expect(socket.destroy).toBeCalledTimes(0)
+
+    jest.advanceTimersByTime(waitTime)
+    expect(socket.destroy).toBeCalledTimes(0)
+
+    jest.advanceTimersByTime(forceTimeout)
+    expect(socket.destroy).toBeCalledTimes(1)
+
+    closeServerCallback!()
+  })
+})

packages/medusa-core-utils/src/graceful-shutdown-server.ts

@@ -0,0 +1,96 @@
+import { Server } from "http"
+import { Socket } from "net"
+import Timer = NodeJS.Timer
+
+interface SocketState extends Socket {
+  _idle: boolean
+  _connectionId: number
+}
+
+export abstract class GracefulShutdownServer {
+  public isShuttingDown: boolean
+  public abstract shutdown(timeout?: number): Promise<void>
+  public static create<T extends Server>(
+    originalServer: T,
+    waitingResponseTime: number = 200
+  ): T & GracefulShutdownServer {
+    let connectionId = 0
+    let shutdownPromise: Promise<void>
+
+    const allSockets: { [id: number]: SocketState } = {}
+
+    const server = originalServer as T & GracefulShutdownServer
+    server.isShuttingDown = false
+    server.shutdown = async (timeout: number = 0): Promise<void> => {
+      if (server.isShuttingDown) {
+        return shutdownPromise
+      }
+
+      server.isShuttingDown = true
+
+      shutdownPromise = new Promise((ok, nok) => {
+        let forceQuit = false
+        let cleanInterval: Timer
+
+        try {
+          // stop accepting new incoming connections
+          server.close(() => {
+            clearInterval(cleanInterval)
+            ok()
+          })
+
+          if (+timeout > 0) {
+            setTimeout(() => {
+              forceQuit = true
+            }, timeout).unref()
+          }
+
+          cleanInterval = setInterval(() => {
+            if (!Object.keys(allSockets).length) {
+              clearInterval(cleanInterval)
+            }
+
+            for (const key of Object.keys(allSockets)) {
+              const socketId = +key
+              if (forceQuit || allSockets[socketId]._idle) {
+                allSockets[socketId].destroy()
+                delete allSockets[socketId]
+              }
+            }
+          }, waitingResponseTime)
+        } catch (error) {
+          clearInterval(cleanInterval!)
+          return nok(error)
+        }
+      })
+
+      return shutdownPromise
+    }
+
+    const onConnect = (originalSocket) => {
+      connectionId++
+      const socket = originalSocket as SocketState
+      socket._idle = true
+      socket._connectionId = connectionId
+      allSockets[connectionId] = socket
+
+      socket.on("close", () => {
+        delete allSockets[socket._connectionId]
+      })
+    }
+
+    server.on("connection", onConnect)
+    server.on("secureConnection", onConnect)
+
+    server.on("request", (req, res) => {
+      const customSocket = req.socket as SocketState
+      customSocket._idle = false
+
+      res.on("finish", () => {
+        customSocket._idle = true
+      })
+    })
+
+    return server
+  }
+}

packages/medusa-core-utils/src/index.ts

@@ -4,11 +4,11 @@ export { countries, isoCountryLookup } from "./countries"
 export { default as createRequireFromPath } from "./create-require-from-path"
 export { default as MedusaError } from "./errors"
 export { default as getConfigFile } from "./get-config-file"
+export * from "./graceful-shutdown-server"
 export { default as humanizeAmount } from "./humanize-amount"
 export { indexTypes } from "./index-types"
+export * from "./is-defined"
 export { parseCorsOrigins } from "./parse-cors-origins"
 export { transformIdableFields } from "./transform-idable-fields"
 export { default as Validator } from "./validator"
 export { default as zeroDecimalCurrencies } from "./zero-decimal-currencies"
-export * from "./is-defined"
-

packages/medusa-react/package.json

@@ -35,9 +35,7 @@
   "devDependencies": {
     "@babel/core": "^7.16.0",
     "@medusajs/medusa": "^1.7.6",
-    "@storybook/addon-contexts": "^5.3.21",
     "@storybook/addon-essentials": "^6.3.12",
-    "@storybook/addon-info": "^5.3.21",
     "@storybook/addon-links": "^6.3.12",
     "@storybook/addons": "^6.3.12",
     "@storybook/react": "^6.3.12",

packages/medusa/src/commands/start.js

@@ -2,6 +2,7 @@ import "core-js/stable"
 import "regenerator-runtime/runtime"
 
 import express from "express"
+import { GracefulShutdownServer } from "medusa-core-utils"
 import { track } from "medusa-telemetry"
 import { scheduleJob } from "node-schedule"
 
@@ -19,13 +20,31 @@ export default async function ({ port, directory }) {
 
     const { dbConnection } = await loaders({ directory, expressApp: app })
     const serverActivity = Logger.activity(`Creating server`)
-    const server = app.listen(port, (err) => {
+    const server = GracefulShutdownServer.create(
-      if (err) {
+      app.listen(port, (err) => {
-        return
+        if (err) {
-      }
+          return
-      Logger.success(serverActivity, `Server is ready on port: ${port}`)
+        }
-      track("CLI_START_COMPLETED")
+        Logger.success(serverActivity, `Server is ready on port: ${port}`)
-    })
+        track("CLI_START_COMPLETED")
+      })
+    )
+
+    // Handle graceful shutdown
+    const gracefulShutDown = () => {
+      server
+        .shutdown()
+        .then(() => {
+          Logger.info("Gracefully stopping the server.")
+          process.exit(0)
+        })
+        .catch((e) => {
+          Logger.error("Error received when shutting down the server.", e)
+          process.exit(1)
+        })
+    }
+    process.on("SIGTERM", gracefulShutDown)
+    process.on("SIGINT", gracefulShutDown)
 
     scheduleJob(CRON_SCHEDULE, () => {
       track("PING")

packages/medusa/src/utils/transaction/distributed-transaction.ts

@@ -1,4 +1,3 @@
-import e from "express"
 import { TransactionFlow, TransactionHandlerType, TransactionState } from "."
 
 /**