diff --git a/packages/beacon-node/src/network/core/networkCore.ts b/packages/beacon-node/src/network/core/networkCore.ts index 6e78e1acc3dc..c2ed2159bc6a 100644 --- a/packages/beacon-node/src/network/core/networkCore.ts +++ b/packages/beacon-node/src/network/core/networkCore.ts @@ -70,6 +70,15 @@ export type BaseNetworkInit = { initialCustodyGroupCount: number; }; +/** Counts of what is still keeping this thread's event loop alive, e.g. `TCPWrap=3,Timeout=1` */ +function formatActiveResources(): string { + const counts = new Map(); + for (const resource of process.getActiveResourcesInfo()) { + counts.set(resource, (counts.get(resource) ?? 0) + 1); + } + return Array.from(counts, ([name, count]) => `${name}=${count}`).join(","); +} + /** * This class is meant to work both: * - In a libp2p worker @@ -289,7 +298,10 @@ export class NetworkCore implements INetworkCore { this.attnetsService.close(); this.syncnetsService.close(); await this.libp2p.stop(); - this.logger.debug("network lib2p closed"); + // Diagnostic for the shutdown hang, this thread can spin in `Environment::CleanupHandles()` on + // a handle that never closes and `Worker.terminate()` then never resolves. Diffing this list + // between a clean and a stuck shutdown should narrow down which handle it is + this.logger.debug("network lib2p closed", {activeResources: formatActiveResources()}); this.closed = true; } diff --git a/packages/beacon-node/src/network/core/networkCoreWorkerHandler.ts b/packages/beacon-node/src/network/core/networkCoreWorkerHandler.ts index 508aec90a08e..72a9c4379611 100644 --- a/packages/beacon-node/src/network/core/networkCoreWorkerHandler.ts +++ b/packages/beacon-node/src/network/core/networkCoreWorkerHandler.ts @@ -10,6 +10,7 @@ import {BeaconConfig, chainConfigToJson} from "@lodestar/config"; import type {LoggerNode} from "@lodestar/logger/node"; import {ResponseIncoming, ResponseOutgoing} from "@lodestar/reqresp"; import {Status} from "@lodestar/types"; +import {withTimeout} from "@lodestar/utils"; import {Metrics} from "../../metrics/index.js"; import {AsyncIterableBridgeCaller, AsyncIterableBridgeHandler} from "../../util/asyncIterableToEvents.js"; import {PeerIdStr, peerIdFromString} from "../../util/peerId.js"; @@ -58,6 +59,10 @@ type WorkerNetworkCoreModules = WorkerNetworkCoreInitModules & { const NETWORK_WORKER_EXIT_TIMEOUT_MS = 1000; const NETWORK_WORKER_EXIT_RETRY_COUNT = 3; +/** `getApi().close()` is an unbounded RPC into the worker and runs before `BeaconNode.close()` + * archives chain state to disk, so it is bounded here. Set well above the ~2-4s a close takes on + * mainnet, tripping it leaves libp2p half torn down */ +const NETWORK_CORE_CLOSE_TIMEOUT_MS = 5000; /** * NetworkCore implementation using a Worker thread @@ -165,15 +170,25 @@ export class WorkerNetworkCore implements INetworkCore { async close(): Promise { this.modules.logger.debug("closing network core running in network worker"); - await this.getApi().close(); + try { + await withTimeout(async () => this.getApi().close(), NETWORK_CORE_CLOSE_TIMEOUT_MS); + } catch (e) { + this.modules.logger.debug("Error closing network core", {}, e as Error); + } this.modules.logger.debug("terminating network worker"); - await terminateWorkerThread({ + const terminated = await terminateWorkerThread({ worker: this.getApi(), retryCount: NETWORK_WORKER_EXIT_RETRY_COUNT, retryMs: NETWORK_WORKER_EXIT_TIMEOUT_MS, logger: this.modules.logger, }); - this.modules.logger.debug("terminated network worker"); + if (terminated) { + this.modules.logger.debug("terminated network worker"); + } else { + // Nothing more to do, `process.exit()` joins the thread regardless so the process still waits + // on it. What matters is returning, so the caller can archive state and close the db + this.modules.logger.debug("Network worker did not terminate in time, continuing shutdown without it"); + } } async test(): Promise { diff --git a/packages/beacon-node/src/util/workerEvents.ts b/packages/beacon-node/src/util/workerEvents.ts index d67017089763..b5815b14fffc 100644 --- a/packages/beacon-node/src/util/workerEvents.ts +++ b/packages/beacon-node/src/util/workerEvents.ts @@ -111,6 +111,12 @@ export function wireEventsOnMainThread( } } +/** + * Terminate a worker thread, bounded to `retryCount * retryMs`. + * + * @returns `false` if it could not be terminated, the worker thread is then still running and the + * caller has to decide how to proceed without it. + */ export async function terminateWorkerThread({ worker, retryMs, @@ -121,7 +127,7 @@ export async function terminateWorkerThread({ retryMs: number; retryCount: number; logger?: Logger; -}): Promise { +}): Promise { const terminated = new Promise((resolve) => { Thread.events(worker).subscribe((event) => { if (event.type === "termination") { @@ -131,13 +137,19 @@ export async function terminateWorkerThread({ }); for (let i = 0; i < retryCount; i++) { - await Thread.terminate(worker); - const result = await Promise.race([terminated, sleep(retryMs).then(() => false)]); + // `Worker.terminate()` resolves only once the worker thread exits, and the thread can spin + // forever in `Environment::CleanupHandles()` when a libuv handle on its loop never closes, so + // it has to be raced too, otherwise `retryCount * retryMs` never applies. + const result = await Promise.race([ + Thread.terminate(worker).then(() => terminated), + sleep(retryMs).then(() => false), + ]); - if (result) return; + if (result) return true; - logger?.warn("Worker thread failed to terminate, retrying..."); + logger?.debug("Worker thread failed to terminate, retrying..."); } - throw new Error(`Worker thread failed to terminate in ${retryCount * retryMs}ms.`); + logger?.debug(`Worker thread failed to terminate in ${retryCount * retryMs}ms`); + return false; } diff --git a/packages/beacon-node/test/unit/util/workerEvents.test.ts b/packages/beacon-node/test/unit/util/workerEvents.test.ts new file mode 100644 index 000000000000..21d33bbc2a7b --- /dev/null +++ b/packages/beacon-node/test/unit/util/workerEvents.test.ts @@ -0,0 +1,54 @@ +import {afterEach, beforeEach, describe, expect, it, vi} from "vitest"; +import {Thread} from "@chainsafe/threads"; +import {terminateWorkerThread} from "../../../src/util/workerEvents.js"; + +vi.mock("@chainsafe/threads", () => ({ + Thread: { + terminate: vi.fn(), + events: vi.fn(), + }, +})); + +describe("util / workerEvents / terminateWorkerThread", () => { + const retryMs = 20; + const retryCount = 3; + const worker = {} as Thread; + + function mockEvents(events: Array<{type: string}>): void { + vi.mocked(Thread.events).mockReturnValue({ + subscribe: (cb: (event: {type: string}) => void) => { + for (const event of events) cb(event); + return {unsubscribe: () => {}}; + }, + } as unknown as ReturnType); + } + + beforeEach(() => { + vi.clearAllMocks(); + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("returns true when the worker terminates and emits a termination event", async () => { + mockEvents([{type: "termination"}]); + vi.mocked(Thread.terminate).mockResolvedValue(undefined as never); + + await expect(terminateWorkerThread({worker, retryMs, retryCount})).resolves.toBe(true); + expect(Thread.terminate).toHaveBeenCalledTimes(1); + }); + + it("returns false in bounded time when Thread.terminate() never resolves (does not hang)", async () => { + // Worker that never exits, terminate() never resolves + mockEvents([]); + vi.mocked(Thread.terminate).mockReturnValue(new Promise(() => {}) as never); + + const promise = terminateWorkerThread({worker, retryMs, retryCount}); + await vi.advanceTimersByTimeAsync(retryMs * retryCount); + + await expect(promise).resolves.toBe(false); + expect(Thread.terminate).toHaveBeenCalledTimes(retryCount); + }); +});