From 4fb08e4ea9c618d2bed1c0cdd48e5cb51da00e5f Mon Sep 17 00:00:00 2001 From: Nimrod Teich Date: Tue, 2 Jun 2026 15:54:10 +0300 Subject: [PATCH] fix(health-probe): bound gRPC clientPool to stop memory leak MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The health-probe clientPool created one gRPC client per provider endpoint and never closed or evicted any of them. Each client owns a live HTTP/2 channel (socket, keepalive timers, native buffers, file descriptors), and provider endpoints churn continuously (restarts, re-IPs, stake changes), so the pool grew without bound. On the 512MB info-mainnet Fargate task this manifested as a monotonic heap climb (~40%→97% over ~36h, no GC recovery) ending in repeated OOM-kills and task restarts. Bound the pool with an LRU cap (HEALTH_PROBE_CLIENT_POOL_MAX, default 256) and close the least-recently-used channel on overflow; a cache hit re-inserts the key so active endpoints stay warm. Client creation is now injectable so the eviction/close behaviour is unit-tested without opening a real channel. --- .../api/src/__tests__/grpc-probe-pool.test.ts | 126 ++++++++++++++++++ apps/api/src/services/grpc-probe.ts | 84 +++++++++++- 2 files changed, 204 insertions(+), 6 deletions(-) create mode 100644 apps/api/src/__tests__/grpc-probe-pool.test.ts diff --git a/apps/api/src/__tests__/grpc-probe-pool.test.ts b/apps/api/src/__tests__/grpc-probe-pool.test.ts new file mode 100644 index 0000000..78ec187 --- /dev/null +++ b/apps/api/src/__tests__/grpc-probe-pool.test.ts @@ -0,0 +1,126 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + CLIENT_POOL_LIMIT, + __clientPoolSize, + __resetClientPool, + getClient, +} from "../services/grpc-probe.js"; + +/** + * Regression test for the gRPC clientPool memory leak: the pool created one + * client per endpoint and never closed/evicted any, so churning provider + * endpoints accumulated dead HTTP/2 channels until the task OOM-killed. + * + * These tests exercise the eviction + close logic through the injectable + * createClient factory — no real network/gRPC channel is opened. + */ + +interface FakeClient { + endpoint: string; + close: ReturnType; +} + +function makeFactory() { + const created: FakeClient[] = []; + const create = (endpoint: string) => { + const client = { endpoint, close: vi.fn() }; + created.push(client); + // The cast matches the real grpc ServiceClient instance shape closely + // enough for the pool, which only ever calls .close() on it here. + return client as unknown as InstanceType< + typeof import("@grpc/grpc-js").ServiceClientConstructor + >; + }; + return { created, create }; +} + +afterEach(() => { + __resetClientPool(); +}); + +describe("grpc-probe clientPool", () => { + it("reuses the same client for a repeated endpoint (no new channel)", () => { + const { created, create } = makeFactory(); + + const a = getClient("provider-1.lava.build:443", create); + const b = getClient("provider-1.lava.build:443", create); + + expect(a).toBe(b); + expect(created).toHaveLength(1); + expect(__clientPoolSize()).toBe(1); + }); + + it("creates distinct clients for distinct endpoints", () => { + const { created, create } = makeFactory(); + + getClient("provider-1.lava.build:443", create); + getClient("provider-2.lava.build:443", create); + + expect(created).toHaveLength(2); + expect(__clientPoolSize()).toBe(2); + }); + + it("is bounded — never exceeds CLIENT_POOL_LIMIT entries", () => { + const { create } = makeFactory(); + + // Push well past the cap with unique endpoints (the leak scenario). + for (let i = 0; i < CLIENT_POOL_LIMIT + 50; i++) { + getClient(`provider-${i}.lava.build:443`, create); + } + + expect(__clientPoolSize()).toBe(CLIENT_POOL_LIMIT); + }); + + it("closes the channel of every evicted client", () => { + const { created, create } = makeFactory(); + + const overflow = 50; + for (let i = 0; i < CLIENT_POOL_LIMIT + overflow; i++) { + getClient(`provider-${i}.lava.build:443`, create); + } + + // The first `overflow` endpoints are the least-recently-used and must have + // been evicted AND closed. + const closed = created.filter((c) => c.close.mock.calls.length > 0); + expect(closed).toHaveLength(overflow); + for (let i = 0; i < overflow; i++) { + expect(created[i]!.close).toHaveBeenCalledTimes(1); + } + }); + + it("keeps recently-used endpoints warm (LRU touch survives eviction)", () => { + const { create } = makeFactory(); + + // Fill the pool exactly to the cap. + for (let i = 0; i < CLIENT_POOL_LIMIT; i++) { + getClient(`provider-${i}.lava.build:443`, create); + } + + // Touch endpoint 0 so it becomes most-recently-used. + const warm = getClient("provider-0.lava.build:443", create); + + // One more unique endpoint forces a single eviction — it must NOT be + // endpoint 0 (which we just touched); endpoint 1 is now the LRU. + getClient("provider-new.lava.build:443", create); + + // Endpoint 0 still resolves to the same live client (no re-create). + const stillWarm = getClient("provider-0.lava.build:443", create); + expect(stillWarm).toBe(warm); + expect(__clientPoolSize()).toBe(CLIENT_POOL_LIMIT); + }); + + it("__resetClientPool closes all channels and empties the pool", () => { + const { created, create } = makeFactory(); + + getClient("provider-1.lava.build:443", create); + getClient("provider-2.lava.build:443", create); + expect(__clientPoolSize()).toBe(2); + + __resetClientPool(); + + expect(__clientPoolSize()).toBe(0); + for (const c of created) { + expect(c.close).toHaveBeenCalledTimes(1); + } + }); +}); diff --git a/apps/api/src/services/grpc-probe.ts b/apps/api/src/services/grpc-probe.ts index 6e1eeff..877dab4 100644 --- a/apps/api/src/services/grpc-probe.ts +++ b/apps/api/src/services/grpc-probe.ts @@ -27,18 +27,90 @@ const proto = grpc.loadPackageDefinition(packageDef) as unknown as { const RelayerClient = proto.lavanet.lava.pairing.Relayer; -// Reuse gRPC clients per endpoint — avoids TLS handshake per probe call -const clientPool = new Map>(); +type RelayerClientInstance = InstanceType; -function getClient(endpoint: string): InstanceType { - let client = clientPool.get(endpoint); - if (client) return client; +/** + * Max number of gRPC clients kept warm at once. Each entry owns a live HTTP/2 + * channel (socket + keepalive timers + native buffers), so an unbounded pool + * leaks memory and file descriptors as provider endpoints churn (restarts, + * re-IPs, stake changes) over the process lifetime. We bound the pool and + * close the least-recently-used channel on overflow. + * + * Override via HEALTH_PROBE_CLIENT_POOL_MAX. The provider universe in flight at + * any moment is a few hundred endpoints; 256 keeps the active set warm while + * capping retained channels. + */ +const CLIENT_POOL_MAX = (() => { + const raw = process.env.HEALTH_PROBE_CLIENT_POOL_MAX; + const n = raw !== undefined ? parseInt(raw, 10) : NaN; + return Number.isFinite(n) && n > 0 ? n : 256; +})(); - client = new RelayerClient(endpoint, grpc.credentials.createSsl()); +// Reuse gRPC clients per endpoint — avoids TLS handshake per probe call. +// Map insertion order gives us LRU ordering: a hit re-inserts the key to the +// tail, so the least-recently-used entry is always Map's first key. +const clientPool = new Map(); + +/** + * Build a real SSL-backed Relayer client. Injectable so the pool's + * eviction/closing behaviour can be unit-tested without network I/O. + */ +function defaultCreateClient(endpoint: string): RelayerClientInstance { + return new RelayerClient(endpoint, grpc.credentials.createSsl()); +} + +function closeClient(client: RelayerClientInstance): void { + try { + // grpc-js clients expose close() to tear down the channel; guard in case a + // test double or future client shape omits it. + (client as { close?: () => void }).close?.(); + } catch { + // A failed close must not crash the probe loop — the channel will be GC'd + // once unreferenced regardless. + } +} + +export function getClient( + endpoint: string, + createClient: (endpoint: string) => RelayerClientInstance = defaultCreateClient, +): RelayerClientInstance { + const existing = clientPool.get(endpoint); + if (existing) { + // LRU touch: move to tail so active endpoints survive eviction. + clientPool.delete(endpoint); + clientPool.set(endpoint, existing); + return existing; + } + + // Evict least-recently-used entries (and close their channels) until there's + // room for the newcomer. + while (clientPool.size >= CLIENT_POOL_MAX) { + const lruKey = clientPool.keys().next().value; + if (lruKey === undefined) break; + const lru = clientPool.get(lruKey); + clientPool.delete(lruKey); + if (lru) closeClient(lru); + } + + const client = createClient(endpoint); clientPool.set(endpoint, client); return client; } +/** Test hook: close every pooled channel and empty the pool. */ +export function __resetClientPool(): void { + for (const client of clientPool.values()) closeClient(client); + clientPool.clear(); +} + +/** Test hook: current pool occupancy. */ +export function __clientPoolSize(): number { + return clientPool.size; +} + +/** The configured pool cap — exported for tests/observability. */ +export const CLIENT_POOL_LIMIT = CLIENT_POOL_MAX; + export interface ProbeResult { latestBlock: number; lavaEpoch: number;