From 1454eba754a8714ac404a49a545702fabf9b87be Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Fri, 21 Jul 2023 13:51:06 +0200 Subject: [PATCH 1/7] Gracefully terminate connections when closing http server --- .../beacon-node/src/api/rest/activeSockets.ts | 61 +++++++++++++++++-- packages/beacon-node/src/api/rest/base.ts | 8 ++- .../beacon-node/src/metrics/server/http.ts | 8 ++- packages/utils/src/index.ts | 1 + packages/utils/src/waitFor.ts | 55 +++++++++++++++++ packages/utils/test/unit/waitFor.test.ts | 37 +++++++++++ 6 files changed, 159 insertions(+), 11 deletions(-) create mode 100644 packages/utils/src/waitFor.ts create mode 100644 packages/utils/test/unit/waitFor.test.ts diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index b98db9b0fb9b..9859162f1b54 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -1,5 +1,6 @@ import http, {Server} from "node:http"; import {Socket} from "node:net"; +import {waitFor} from "@lodestar/utils"; import {IGauge} from "../../metrics/index.js"; export type SocketMetrics = { @@ -8,15 +9,21 @@ export type SocketMetrics = { socketsBytesWritten: IGauge; }; +// Use relatively short timeout to speed up shutdown +const GRACEFUL_TERMINATION_TIMEOUT = 1_000; + /** * From https://github.com/nodejs/node/blob/57bd715d527aba8dae56b975056961b0e429e91e/lib/_http_client.js#L363-L413 - * But exposes the count of sockets, and does not have a graceful period + * But exposes the count of sockets, and waits for connections to drain until timeout */ export class HttpActiveSocketsTracker { private sockets = new Set(); private terminated = false; - constructor(server: Server, metrics: SocketMetrics | null) { + constructor( + private readonly server: Server, + metrics: SocketMetrics | null + ) { server.on("connection", (socket) => { if (this.terminated) { socket.destroy(Error("Closing")); @@ -39,9 +46,24 @@ export class HttpActiveSocketsTracker { } } - destroyAll(): void { + /** + * Wait for all connections to drain, forcefully terminate any open connections after timeout + * + * From https://github.com/gajus/http-terminator/blob/aabca4751552e983f8a59ba896b7fb58ce3b4087/src/factories/createInternalHttpTerminator.ts#L78-L165 + * But only handles HTTP sockets and does not close server, immediately terminates api.eventstream connections + */ + async terminate(): Promise { + if (this.terminated) return; this.terminated = true; + // Inform new incoming requests (and keep-alive connections) that + // the connection will be closed after the current response + this.server.on("request", (_req, res) => { + if (!res.headersSent) { + res.setHeader("Connection", "close"); + } + }); + for (const socket of this.sockets) { // This is the HTTP CONNECT request socket. // @ts-expect-error Unclear if I am using wrong type or how else this should be handled. @@ -49,8 +71,37 @@ export class HttpActiveSocketsTracker { continue; } - socket.destroy(Error("Closing")); - this.sockets.delete(socket); + // @ts-expect-error Unclear if I am using wrong type or how else this should be handled. + const serverResponse = socket._httpMessage as http.ServerResponse | undefined; + + if (serverResponse == null) { + // Immediately destroy sockets without an attached HTTP request + this.destroySocket(socket); + } else if (serverResponse.getHeader("Content-Type") === "text/event-stream") { + // api.eventstream will never stop and must be forcefully terminated + this.destroySocket(socket); + } else if (!serverResponse.headersSent) { + // Inform existing keep-alive connections that they will be closed after the current response + serverResponse.setHeader("Connection", "close"); + } + } + + // Wait for all connections to drain, forcefully terminate after timeout + try { + await waitFor(() => this.sockets.size === 0, { + timeout: GRACEFUL_TERMINATION_TIMEOUT, + }); + } catch { + // Ignore timeout errors + } finally { + for (const socket of this.sockets) { + this.destroySocket(socket); + } } } + + private destroySocket(socket: Socket): void { + socket.destroy(Error("Closing")); + this.sockets.delete(socket); + } } diff --git a/packages/beacon-node/src/api/rest/base.ts b/packages/beacon-node/src/api/rest/base.ts index fda16fffc8ce..0a04139e9b55 100644 --- a/packages/beacon-node/src/api/rest/base.ts +++ b/packages/beacon-node/src/api/rest/base.ts @@ -153,12 +153,14 @@ export class RestApiServer { // In NodeJS land calling close() only causes new connections to be rejected. // Existing connections can prevent .close() from resolving for potentially forever. - // In Lodestar case when the BeaconNode wants to close we will just abruptly terminate - // all existing connections for a fast shutdown. + // In Lodestar case when the BeaconNode wants to close we will attempt to gracefully + // close all existing connections but forcefully terminate after timeout for a fast shutdown. // Inspired by https://github.com/gajus/http-terminator/ - this.activeSockets.destroyAll(); + await this.activeSockets.terminate(); await this.server.close(); + + this.logger.debug("REST API server closed"); } /** For child classes to override */ diff --git a/packages/beacon-node/src/metrics/server/http.ts b/packages/beacon-node/src/metrics/server/http.ts index ba17b743d112..b699471e07d5 100644 --- a/packages/beacon-node/src/metrics/server/http.ts +++ b/packages/beacon-node/src/metrics/server/http.ts @@ -90,10 +90,10 @@ export async function getHttpMetricsServer( async close(): Promise { // In NodeJS land calling close() only causes new connections to be rejected. // Existing connections can prevent .close() from resolving for potentially forever. - // In Lodestar case when the BeaconNode wants to close we will just abruptly terminate - // all existing connections for a fast shutdown. + // In Lodestar case when the BeaconNode wants to close we will attempt to gracefully + // close all existing connections but forcefully terminate after timeout for a fast shutdown. // Inspired by https://github.com/gajus/http-terminator/ - activeSockets.destroyAll(); + await activeSockets.terminate(); await new Promise((resolve, reject) => { server.close((err) => { @@ -101,6 +101,8 @@ export async function getHttpMetricsServer( else resolve(); }); }); + + logger.debug("Metrics HTTP server closed"); }, }; } diff --git a/packages/utils/src/index.ts b/packages/utils/src/index.ts index bcb0bf27109f..a09c615fbf2b 100644 --- a/packages/utils/src/index.ts +++ b/packages/utils/src/index.ts @@ -16,3 +16,4 @@ export * from "./timeout.js"; export {RecursivePartial, bnToNum} from "./types.js"; export * from "./verifyMerkleBranch.js"; export * from "./promise.js"; +export * from "./waitFor.js"; diff --git a/packages/utils/src/waitFor.ts b/packages/utils/src/waitFor.ts new file mode 100644 index 000000000000..532d66f254a3 --- /dev/null +++ b/packages/utils/src/waitFor.ts @@ -0,0 +1,55 @@ +import {ErrorAborted, TimeoutError} from "./errors.js"; + +export type WaitForOpts = { + /** Time in milliseconds between checking condition */ + interval?: number; + /** Time in milliseconds to wait before throwing TimeoutError */ + timeout?: number; + /** Signal to abort waiting for condition by throwing ErrorAborted */ + signal?: AbortSignal; +}; + +/** + * Wait for a condition to be true + * + * Simplified and abortable implementation of https://github.com/sindresorhus/p-wait-for + */ +export function waitFor(condition: () => boolean, opts: WaitForOpts = {}): Promise { + return new Promise((resolve, reject) => { + const {interval = 10, timeout = Infinity, signal} = opts; + + if (signal && signal.aborted) { + return reject(new ErrorAborted()); + } + + if (condition()) { + return resolve(); + } + + let onDone: () => void = () => {}; + + const timeoutId = setTimeout(() => { + onDone(); + reject(new TimeoutError()); + }, timeout); + + const intervalId = setInterval(() => { + if (condition()) { + onDone(); + resolve(); + } + }, interval); + + const onAbort = (): void => { + onDone(); + reject(new ErrorAborted()); + }; + if (signal) signal.addEventListener("abort", onAbort); + + onDone = () => { + clearTimeout(timeoutId); + clearInterval(intervalId); + if (signal) signal.removeEventListener("abort", onAbort); + }; + }); +} diff --git a/packages/utils/test/unit/waitFor.test.ts b/packages/utils/test/unit/waitFor.test.ts new file mode 100644 index 000000000000..1dd3dec766b7 --- /dev/null +++ b/packages/utils/test/unit/waitFor.test.ts @@ -0,0 +1,37 @@ +import "../setup.js"; +import {expect} from "chai"; +import {waitFor} from "../../src/waitFor.js"; +import {ErrorAborted, TimeoutError} from "../../src/errors.js"; + +describe("waitFor", () => { + const interval = 10; + const timeout = 20; + + it("Should resolve if condition is already true", async () => { + await expect(waitFor(() => true, {interval, timeout})).to.be.fulfilled; + }); + + it("Should resolve if condition becomes true within timeout", async () => { + let condition = false; + setTimeout(() => { + condition = true; + }, interval); + await waitFor(() => condition, {interval, timeout}); + }); + + it("Should reject with TimeoutError if condition does not become true within timeout", async () => { + await expect(waitFor(() => false, {interval, timeout})).to.be.rejectedWith(TimeoutError); + }); + + it("Should reject with ErrorAborted if aborted before condition becomes true", async () => { + const controller = new AbortController(); + setTimeout(() => controller.abort(), interval); + await expect(waitFor(() => false, {interval, timeout, signal: controller.signal})).to.be.rejectedWith(ErrorAborted); + }); + + it("Should reject with ErrorAborted if signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(); + await expect(waitFor(() => true, {interval, timeout, signal: controller.signal})).to.be.rejectedWith(ErrorAborted); + }); +}); From 7a1e4355224d74a4809d3a5a8260814b1b56a3d8 Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sat, 22 Jul 2023 14:52:41 +0200 Subject: [PATCH 2/7] Log unexpected errors when shutting down server --- packages/beacon-node/src/api/rest/base.ts | 17 +---------------- 1 file changed, 1 insertion(+), 16 deletions(-) diff --git a/packages/beacon-node/src/api/rest/base.ts b/packages/beacon-node/src/api/rest/base.ts index 0a04139e9b55..c9c6e4bfb0ef 100644 --- a/packages/beacon-node/src/api/rest/base.ts +++ b/packages/beacon-node/src/api/rest/base.ts @@ -29,11 +29,6 @@ export type RestApiServerMetrics = SocketMetrics & { errors: IGauge<"operationId">; }; -enum Status { - Listening = "listening", - Closed = "closed", -} - /** * REST API powered by `fastify` server. */ @@ -42,8 +37,6 @@ export class RestApiServer { protected readonly logger: Logger; private readonly activeSockets: HttpActiveSocketsTracker; - private status = Status.Closed; - constructor( private readonly opts: RestApiServerOpts, modules: RestApiServerModules @@ -106,8 +99,7 @@ export class RestApiServer { server.addHook("onError", async (req, _res, err) => { // Don't log ErrorAborted errors, they happen on node shutdown and are not useful // Don't log NodeISSyncing errors, they happen very frequently while syncing and the validator polls duties - // Don't log eventstream aborted errors if server instance is being closed on node shutdown - if (err instanceof ErrorAborted || err instanceof NodeIsSyncing || this.status === Status.Closed) return; + if (err instanceof ErrorAborted || err instanceof NodeIsSyncing) return; const {operationId} = req.routeConfig as RouteConfig; @@ -127,9 +119,6 @@ export class RestApiServer { * Start the REST API server. */ async listen(): Promise { - if (this.status === Status.Listening) return; - this.status = Status.Listening; - try { const host = this.opts.address; const address = await this.server.listen({port: this.opts.port, host}); @@ -139,7 +128,6 @@ export class RestApiServer { } } catch (e) { this.logger.error("Error starting REST api server", this.opts, e as Error); - this.status = Status.Closed; throw e; } } @@ -148,9 +136,6 @@ export class RestApiServer { * Close the server instance and terminate all existing connections. */ async close(): Promise { - if (this.status === Status.Closed) return; - this.status = Status.Closed; - // In NodeJS land calling close() only causes new connections to be rejected. // Existing connections can prevent .close() from resolving for potentially forever. // In Lodestar case when the BeaconNode wants to close we will attempt to gracefully From 485a996a1c2244015c40ab0c8f576ad790b79f7a Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sat, 22 Jul 2023 17:02:57 +0200 Subject: [PATCH 3/7] Fix code references in comments --- packages/beacon-node/src/api/rest/activeSockets.ts | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index 9859162f1b54..2dd3cc969936 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -13,8 +13,8 @@ export type SocketMetrics = { const GRACEFUL_TERMINATION_TIMEOUT = 1_000; /** - * From https://github.com/nodejs/node/blob/57bd715d527aba8dae56b975056961b0e429e91e/lib/_http_client.js#L363-L413 - * But exposes the count of sockets, and waits for connections to drain until timeout + * From https://github.com/gajus/http-terminator/blob/aabca4751552e983f8a59ba896b7fb58ce3b4087/src/factories/createInternalHttpTerminator.ts#L24-L61 + * But only handles HTTP sockets, exposes the count of sockets as metrics */ export class HttpActiveSocketsTracker { private sockets = new Set(); @@ -50,13 +50,13 @@ export class HttpActiveSocketsTracker { * Wait for all connections to drain, forcefully terminate any open connections after timeout * * From https://github.com/gajus/http-terminator/blob/aabca4751552e983f8a59ba896b7fb58ce3b4087/src/factories/createInternalHttpTerminator.ts#L78-L165 - * But only handles HTTP sockets and does not close server, immediately terminates api.eventstream connections + * But only handles HTTP sockets and does not close server, immediately terminates eventstream API connections */ async terminate(): Promise { if (this.terminated) return; this.terminated = true; - // Inform new incoming requests (and keep-alive connections) that + // Inform new incoming requests on keep-alive connections that // the connection will be closed after the current response this.server.on("request", (_req, res) => { if (!res.headersSent) { @@ -78,7 +78,7 @@ export class HttpActiveSocketsTracker { // Immediately destroy sockets without an attached HTTP request this.destroySocket(socket); } else if (serverResponse.getHeader("Content-Type") === "text/event-stream") { - // api.eventstream will never stop and must be forcefully terminated + // eventstream API will never stop and must be forcefully terminated this.destroySocket(socket); } else if (!serverResponse.headersSent) { // Inform existing keep-alive connections that they will be closed after the current response From 654bf05808e6cb123e3f6bc52c1357be702a3443 Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sun, 23 Jul 2023 11:18:38 +0200 Subject: [PATCH 4/7] More gracefully close eventstream api --- packages/beacon-node/src/api/rest/activeSockets.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index 2dd3cc969936..b04660185576 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -78,8 +78,8 @@ export class HttpActiveSocketsTracker { // Immediately destroy sockets without an attached HTTP request this.destroySocket(socket); } else if (serverResponse.getHeader("Content-Type") === "text/event-stream") { - // eventstream API will never stop and must be forcefully terminated - this.destroySocket(socket); + // eventstream API will never stop and must be forcefully closed + socket.end(); } else if (!serverResponse.headersSent) { // Inform existing keep-alive connections that they will be closed after the current response serverResponse.setHeader("Connection", "close"); From 31450a423372159dda6d40f9fb0caa6a8b7ec3b9 Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sun, 23 Jul 2023 11:37:19 +0200 Subject: [PATCH 5/7] Misc updates --- .../beacon-node/src/api/rest/activeSockets.ts | 22 +++++++++---------- packages/utils/src/waitFor.ts | 6 ++--- 2 files changed, 13 insertions(+), 15 deletions(-) diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index b04660185576..f84fcd78a352 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -18,14 +18,14 @@ const GRACEFUL_TERMINATION_TIMEOUT = 1_000; */ export class HttpActiveSocketsTracker { private sockets = new Set(); - private terminated = false; + private terminating = false; constructor( private readonly server: Server, metrics: SocketMetrics | null ) { server.on("connection", (socket) => { - if (this.terminated) { + if (this.terminating) { socket.destroy(Error("Closing")); } else { this.sockets.add(socket); @@ -50,11 +50,11 @@ export class HttpActiveSocketsTracker { * Wait for all connections to drain, forcefully terminate any open connections after timeout * * From https://github.com/gajus/http-terminator/blob/aabca4751552e983f8a59ba896b7fb58ce3b4087/src/factories/createInternalHttpTerminator.ts#L78-L165 - * But only handles HTTP sockets and does not close server, immediately terminates eventstream API connections + * But only handles HTTP sockets and does not close server, immediately closes eventstream API connections */ async terminate(): Promise { - if (this.terminated) return; - this.terminated = true; + if (this.terminating) return; + this.terminating = true; // Inform new incoming requests on keep-alive connections that // the connection will be closed after the current response @@ -72,17 +72,17 @@ export class HttpActiveSocketsTracker { } // @ts-expect-error Unclear if I am using wrong type or how else this should be handled. - const serverResponse = socket._httpMessage as http.ServerResponse | undefined; + const res = socket._httpMessage as http.ServerResponse | undefined; - if (serverResponse == null) { + if (res == null) { // Immediately destroy sockets without an attached HTTP request this.destroySocket(socket); - } else if (serverResponse.getHeader("Content-Type") === "text/event-stream") { + } else if (res.getHeader("Content-Type") === "text/event-stream") { // eventstream API will never stop and must be forcefully closed socket.end(); - } else if (!serverResponse.headersSent) { + } else if (!res.headersSent) { // Inform existing keep-alive connections that they will be closed after the current response - serverResponse.setHeader("Connection", "close"); + res.setHeader("Connection", "close"); } } @@ -92,7 +92,7 @@ export class HttpActiveSocketsTracker { timeout: GRACEFUL_TERMINATION_TIMEOUT, }); } catch { - // Ignore timeout errors + // Ignore timeout error } finally { for (const socket of this.sockets) { this.destroySocket(socket); diff --git a/packages/utils/src/waitFor.ts b/packages/utils/src/waitFor.ts index 532d66f254a3..91206267e6ca 100644 --- a/packages/utils/src/waitFor.ts +++ b/packages/utils/src/waitFor.ts @@ -5,20 +5,18 @@ export type WaitForOpts = { interval?: number; /** Time in milliseconds to wait before throwing TimeoutError */ timeout?: number; - /** Signal to abort waiting for condition by throwing ErrorAborted */ + /** Abort signal to stop waiting for condition by throwing ErrorAborted */ signal?: AbortSignal; }; /** * Wait for a condition to be true - * - * Simplified and abortable implementation of https://github.com/sindresorhus/p-wait-for */ export function waitFor(condition: () => boolean, opts: WaitForOpts = {}): Promise { return new Promise((resolve, reject) => { const {interval = 10, timeout = Infinity, signal} = opts; - if (signal && signal.aborted) { + if (signal?.aborted) { return reject(new ErrorAborted()); } From a0c170b59add84c398fad0fdfda46afaa4581852 Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sun, 23 Jul 2023 12:06:04 +0200 Subject: [PATCH 6/7] Close idle connections on server --- packages/beacon-node/src/api/rest/activeSockets.ts | 3 +++ 1 file changed, 3 insertions(+) diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index f84fcd78a352..600c3049471c 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -56,6 +56,9 @@ export class HttpActiveSocketsTracker { if (this.terminating) return; this.terminating = true; + // Can speed up shutdown by a few milliseconds + this.server.closeIdleConnections(); + // Inform new incoming requests on keep-alive connections that // the connection will be closed after the current response this.server.on("request", (_req, res) => { From ff7097238f619cfa3e9a805f67e5477f8e022677 Mon Sep 17 00:00:00 2001 From: Nico Flaig Date: Sun, 23 Jul 2023 12:46:16 +0200 Subject: [PATCH 7/7] Update ts-expect-error comments --- packages/beacon-node/src/api/rest/activeSockets.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/beacon-node/src/api/rest/activeSockets.ts b/packages/beacon-node/src/api/rest/activeSockets.ts index 600c3049471c..ba8a35c80119 100644 --- a/packages/beacon-node/src/api/rest/activeSockets.ts +++ b/packages/beacon-node/src/api/rest/activeSockets.ts @@ -69,12 +69,12 @@ export class HttpActiveSocketsTracker { for (const socket of this.sockets) { // This is the HTTP CONNECT request socket. - // @ts-expect-error Unclear if I am using wrong type or how else this should be handled. + // @ts-expect-error HTTP sockets have reference to server if (!(socket.server instanceof http.Server)) { continue; } - // @ts-expect-error Unclear if I am using wrong type or how else this should be handled. + // @ts-expect-error Internal property but only way to access response of socket const res = socket._httpMessage as http.ServerResponse | undefined; if (res == null) {