From 8275f29c1b146397f2a03c42c66be9594cd82932 Mon Sep 17 00:00:00 2001 From: "phamtobao@gmail.com" Date: Thu, 30 Jul 2026 22:51:17 +0400 Subject: [PATCH 1/2] feat: add memory leak detection --- lib/index.js | 2 + lib/logger.js | 9 ++ lib/memory-watch.js | 258 +++++++++++++++++++++++++++++++++++++++ lib/test/memory-watch.js | 47 +++++++ package.json | 1 + 5 files changed, 317 insertions(+) create mode 100644 lib/memory-watch.js create mode 100644 lib/test/memory-watch.js diff --git a/lib/index.js b/lib/index.js index c90bc88..8632bdf 100644 --- a/lib/index.js +++ b/lib/index.js @@ -6,6 +6,7 @@ const dbConf = require("./configs"); const Events = require("./lex/event"); const Logger = require("./logger"); const Mariadb = require("./mariadb"); +const MemoryWatch = require("./memory-watch"); const Messenger = require("./messenger"); const Network = require("./network"); const Offline = require("./offline"); @@ -67,6 +68,7 @@ module.exports = { getUiInfo, Logger, Mariadb, + MemoryWatch, MessageBus: RedisStore, Messenger, Network, diff --git a/lib/logger.js b/lib/logger.js index 98d3bdd..450658f 100644 --- a/lib/logger.js +++ b/lib/logger.js @@ -25,6 +25,7 @@ const { existsSync, mkdirSync, readFileSync } = require('fs'); const { resolve, normalize, join, dirname } = require('path'); const Cache = require("./cache"); +const MemoryWatch = require("./memory-watch"); const LEVEL = { ERROR: 0, @@ -79,6 +80,13 @@ class logger extends Backbone.Model { return LEVEL; } + /** + * + */ + preinitialize() { + MemoryWatch.track(this); + } + /** * * @param {...any} args @@ -412,6 +420,7 @@ class logger extends Backbone.Model { timer = null; }, 11000); this._stopping = 1; + MemoryWatch.markStopped(this); }; diff --git a/lib/memory-watch.js b/lib/memory-watch.js new file mode 100644 index 0000000..823e503 --- /dev/null +++ b/lib/memory-watch.js @@ -0,0 +1,258 @@ + +// ================================ * +// Copyright Xialia.com 2013-2026 * +// FILE : memory-watch.js +// TYPE : module +// ================================ * + +const { env } = process; + +const MB = 1024 * 1024; +const STOPPED_CAP = 10000; + +/** + * + * @param {*} name + * @param {*} fallback + * @returns + */ +function intEnv(name, fallback) { + let v = parseInt(env[name]); + if (isNaN(v) || v < 0) return fallback; + return v; +} + +/** + * + * @returns + */ +function now() { + return new Date().toISOString(); +} + +class MemoryWatch { + constructor() { + this.enabled = env.MEMWATCH !== '0'; + this.tracking = this.enabled + && env.MEMWATCH_TRACK !== '0' + && typeof FinalizationRegistry === 'function' + && typeof WeakRef === 'function'; + this.interval = intEnv('MEMWATCH_INTERVAL', 60); + this.window = Math.max(intEnv('MEMWATCH_WINDOW', 15), 3); + this.growthMb = intEnv('MEMWATCH_GROWTH_MB', 64); + this.rssMaxMb = intEnv('MEMWATCH_RSS_MAX_MB', 1024); + this.zombieAge = intEnv('MEMWATCH_ZOMBIE_AGE', 300); + this.zombieMax = intEnv('MEMWATCH_ZOMBIE_MAX', 200); + this.cooldown = intEnv('MEMWATCH_COOLDOWN', 1800); + + this.samples = []; + this.classes = new Map(); + this.stopped = new Set(); + this.lastWarn = {}; + this.timer = null; + if (this.tracking) { + this.registry = new FinalizationRegistry((name) => { + let c = this.classes.get(name); + if (c) c.finalized++; + }); + } + } + + /** + * + * @param {*} obj + * @returns + */ + track(obj) { + if (!this.tracking) return; + let name = obj.constructor.name || 'anonymous'; + let c = this.classes.get(name); + if (!c) { + c = { created: 0, finalized: 0 }; + this.classes.set(name, c); + } + c.created++; + this.registry.register(obj, name); + } + + /** + * + * @param {*} obj + * @returns + */ + markStopped(obj) { + if (!this.tracking) return; + if (this.stopped.size >= STOPPED_CAP) { + let oldest = this.stopped.values().next().value; + this.stopped.delete(oldest); + } + this.stopped.add({ + ref: new WeakRef(obj), + name: obj.constructor.name || 'anonymous', + time: Date.now(), + }); + } + + /** + * + * @param {*} limit + * @returns + */ + topLive(limit = 5) { + let rows = []; + for (let [name, c] of this.classes) { + let live = c.created - c.finalized; + if (live > 0) rows.push({ name, live }); + } + rows.sort((a, b) => b.live - a.live); + return rows.slice(0, limit); + } + + /** + * + * @returns + */ + zombies() { + let res = new Map(); + let limit = Date.now() - this.zombieAge * 1000; + for (let item of this.stopped) { + if (!item.ref.deref()) { + this.stopped.delete(item); + continue; + } + if (item.time < limit) { + res.set(item.name, (res.get(item.name) || 0) + 1); + } + } + return res; + } + + /** + * + * @param {*} kind + * @param {...any} args + * @returns + */ + warn(kind, ...args) { + let last = this.lastWarn[kind] || 0; + if (Date.now() - last < this.cooldown * 1000) return; + this.lastWarn[kind] = Date.now(); + console.warn(`[${now()}] MemoryWatch[WARN]:`, ...args); + } + + /** + * + * @returns + */ + sample() { + let m = process.memoryUsage(); + let s = { + time: Date.now(), + rss: m.rss, + heapUsed: m.heapUsed, + heapTotal: m.heapTotal, + external: m.external, + used: m.heapUsed + m.external, + }; + this.samples.push(s); + if (this.samples.length > this.window) this.samples.shift(); + + if (this.rssMaxMb && s.rss > this.rssMaxMb * MB) { + this.warn('rss', + `rss ${(s.rss / MB).toFixed(1)}MB exceeds limit ${this.rssMaxMb}MB;`, + `top live:`, this.format(this.topLive()) + ); + } + + if (this.samples.length === this.window) { + let first = this.samples[0]; + let growth = s.used - first.used; + let rising = 0; + for (let i = 1; i < this.samples.length; i++) { + if (this.samples[i].used > this.samples[i - 1].used) rising++; + } + let ratio = rising / (this.samples.length - 1); + if (growth > this.growthMb * MB && ratio >= 0.85) { + let hours = (s.time - first.time) / 3600000; + let rate = hours > 0 ? (growth / MB / hours).toFixed(1) : 'n/a'; + this.warn('leak', + `probable memory leak: heap+external grew ${(growth / MB).toFixed(1)}MB`, + `over last ${this.samples.length} samples (~${rate}MB/h),`, + `heapUsed=${(s.heapUsed / MB).toFixed(1)}MB`, + `external=${(s.external / MB).toFixed(1)}MB`, + `rss=${(s.rss / MB).toFixed(1)}MB;`, + `top live:`, this.format(this.topLive()) + ); + this.samples = [s]; + } + } + + if (this.tracking) { + let z = this.zombies(); + let total = 0; + for (let n of z.values()) total += n; + if (total > this.zombieMax) { + let rows = [...z.entries()] + .sort((a, b) => b[1] - a[1]).slice(0, 5) + .map(([name, n]) => `${name}:${n}`).join(' '); + this.warn('zombie', + `${total} components still retained ${this.zombieAge}s after stop();`, + `top:`, rows + ); + } + } + return s; + } + + /** + * + * @param {*} rows + * @returns + */ + format(rows) { + if (!rows.length) return 'n/a'; + return rows.map((r) => `${r.name}:${r.live}`).join(' '); + } + + /** + * + * @returns + */ + report() { + let m = process.memoryUsage(); + return { + enabled: this.enabled, + tracking: this.tracking, + rssMb: +(m.rss / MB).toFixed(1), + heapUsedMb: +(m.heapUsed / MB).toFixed(1), + samples: this.samples.length, + topLive: this.topLive(10), + stoppedPending: this.stopped.size, + }; + } + + /** + * + * @returns + */ + start() { + if (!this.enabled || this.timer) return; + this.timer = setInterval(() => this.sample(), this.interval * 1000); + if (this.timer.unref) this.timer.unref(); + } + + /** + * + * @returns + */ + stop() { + if (!this.timer) return; + clearInterval(this.timer); + this.timer = null; + } +} + +const instance = new MemoryWatch(); +instance.start(); + +module.exports = instance; diff --git a/lib/test/memory-watch.js b/lib/test/memory-watch.js new file mode 100644 index 0000000..38f8bb8 --- /dev/null +++ b/lib/test/memory-watch.js @@ -0,0 +1,47 @@ + +// ================================ * +// Copyright Xialia.com 2013-2026 * +// FILE : test/memory-watch.js +// TYPE : test +// ================================ * + +process.env.MEMWATCH_INTERVAL = '1'; +process.env.MEMWATCH_WINDOW = '5'; +process.env.MEMWATCH_GROWTH_MB = '10'; +process.env.MEMWATCH_ZOMBIE_AGE = '2'; +process.env.MEMWATCH_ZOMBIE_MAX = '3'; +process.env.MEMWATCH_COOLDOWN = '5'; + +const MemoryWatch = require('../memory-watch'); +const Logger = require('../logger'); + +class LeakyComponent extends Logger { } + +const retained = []; +const hoard = []; + +console.log('--- initial report:', JSON.stringify(MemoryWatch.report())); + +for (let i = 0; i < 10; i++) { + let c = new LeakyComponent(); + c.stop(); + retained.push(c); +} + +let leaker = setInterval(() => { + hoard.push(Buffer.alloc(4 * 1024 * 1024, 1)); +}, 1000); + +setTimeout(() => { + clearInterval(leaker); + let report = MemoryWatch.report(); + console.log('--- final report:', JSON.stringify(report, null, 2)); + let ok = report.enabled && report.tracking + && report.topLive.some((r) => r.name === 'LeakyComponent'); + if (ok) { + console.log('memory-watch: OK'); + process.exit(0); + } + console.error('memory-watch: FAILED'); + process.exit(1); +}, 15000); diff --git a/package.json b/package.json index fa1e93e..e175142 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "test:email": "node lib/test/email.js", "test:template": "node lib/test/template.js", "test:db": "node lib/test/db.js", + "test:memwatch": "node lib/test/memory-watch.js", "test:modules": "node lib/test/cache.js", "release": "git push && npm publish --access public && npm version patch" }, From 475ac46c9c7503dd4c7de923415797b10203d53d Mon Sep 17 00:00:00 2001 From: "phamtobao@gmail.com" Date: Thu, 6 Aug 2026 23:11:47 +0400 Subject: [PATCH 2/2] fix: use real connection pool and drop process.exit on connection errors --- lib/mariadb.js | 113 ++++++++++++++++++++++++++++++++++--------------- 1 file changed, 79 insertions(+), 34 deletions(-) diff --git a/lib/mariadb.js b/lib/mariadb.js index 866fa1e..8d16562 100644 --- a/lib/mariadb.js +++ b/lib/mariadb.js @@ -59,32 +59,32 @@ class mariadb_stub extends Logger { if (!Mariadb) { Mariadb = await import('mariadb'); } - return new Promise(async (resolve, reject) => { - if (this._connection && this._connection.isValid()) { - return resolve(this._connection); - } - if (this.get(Attr.limit) > 1) { - let pool = Mariadb.createPool(this.configs); - this._pool = pool; - try { - this._connection = await Mariadb.createConnection(this.configs) - resolve(this._connection); - } catch (err) { - console.trace() - this.warn("[FATAL]: not connected due to error ", err); - reject(err) - } - } else { - try { - this._connection = await Mariadb.createConnection(this.configs) - resolve(this._connection); - } catch (err) { - console.trace() - this.warn("[FATAL]: not connected due to error ", err); - reject(err) - } - } - }) + if (this._connection && this._connection.isValid()) { + return this._connection; + } + try { + this._connection = await Mariadb.createConnection(this.configs) + return this._connection; + } catch (err) { + console.trace() + this.warn("[FATAL]: not connected due to error ", err); + throw err; + } + } + + /** + * Lazily creates the connection pool. Only used when limit > 1; + * instances with limit 1 keep the legacy single-connection behavior. + * @returns + */ + async getPool() { + if (!Mariadb) { + Mariadb = await import('mariadb'); + } + if (!this._pool) { + this._pool = Mariadb.createPool({ ...this.configs, minimumIdle: 1 }); + } + return this._pool; } /** @@ -112,14 +112,17 @@ class mariadb_stub extends Logger { } if (e.fatal) { - this.warn("FATAL ERROR RAISED -- EXITING", e); + this.warn("FATAL ERROR RAISED -- RESETTING CONNECTION", e); + this._connection = null; throw e; } switch (e.code) { case 'ER_CMD_CONNECTION_CLOSED': case 'ER_CONNECTION_TIMEOUT': - this.warn("CONNECTION ERROR -- FORCE RELOAD", e); - process.exit(1); + this.warn("CONNECTION ERROR -- RESETTING CONNECTION", e); + this._connection = null; + this.trigger(ERROR, err); + return; case 'ER_LOCK_DEADLOCK': return; default: @@ -235,6 +238,9 @@ class mariadb_stub extends Logger { sql = args.shift(); } this.log(sql, args); + if (this.get(Attr.limit) > 1) { + return this._runPooled(sql, args, handler); + } let res = []; let c = null; try { @@ -244,19 +250,56 @@ class mariadb_stub extends Logger { this.warn('Attempt to run after connection close'); return { failed: 1, code: "CONNECTION_ALREADY_CLOSED" } } - this.warn('Some abnormal errors occurred. Exitng...'); + this.warn('Some abnormal errors occurred. Connection dropped'); this.debug({ db: this._dbname, sql, args }); - process.exit(1); + this._connection = null; + return { failed: 1, code: "CONNECTION_LOST" } } } catch (e) { this.warn('Failed to get DB connection', e); - process.exit(1); + return { failed: 1, code: "CONNECTION_FAILED" } } return c.beginTransaction() .then(() => { return c.query(sql, args); }) .then((rows) => { + if (rows) { + try { + res = rows.get_rows(); + } catch (e) { + res = rows; + } + } + c.commit(); + if (isFunction(handler)) { + return handler(res); + } + return res; + }) + .catch(this._handleError); + } + + /** + * Runs a statement on a pooled connection. The pool acquires and + * releases the connection around each statement (autocommit), so a + * long-running call never blocks the other requests of the process. + * @param {*} sql + * @param {*} args + * @param {*} handler + * @returns + */ + async _runPooled(sql, args, handler) { + let pool; + try { + pool = await this.getPool(); + } catch (e) { + this.warn('Failed to get DB pool', e); + return { failed: 1, code: "CONNECTION_FAILED" } + } + return pool.query(sql, args) + .then((rows) => { + let res = []; if (rows) { try { res = rows.get_rows(); @@ -267,7 +310,6 @@ class mariadb_stub extends Logger { if (isFunction(handler)) { return handler(res); } - c.commit(); return res; }) .catch(this._handleError); @@ -351,6 +393,9 @@ class mariadb_stub extends Logger { * @returns */ async await_run(sql, args) { + if (this.get(Attr.limit) > 1) { + return this._runPooled(sql, args); + } let c = await this.getConnection(); return c.beginTransaction() .then(() => { @@ -380,7 +425,7 @@ class mariadb_stub extends Logger { this._closed = 1; try { this.silly("STOPING DB", this._dbname, this.isValid()); - if (!this.isValid()) { + if (!this.isValid() && !this._pool) { this.stop(); return; }