From b17b83049ba2a18b6a6591cc2bc719a0de9c0601 Mon Sep 17 00:00:00 2001 From: yash-pouranik Date: Fri, 31 Jul 2026 01:44:23 +0530 Subject: [PATCH 1/2] feat(common): implement LRU connection cache and circuit breakers --- .../common/src/utils/connection.manager.js | 41 +++++++++- packages/common/src/utils/registry.js | 76 ++++++++++++++++++- 2 files changed, 112 insertions(+), 5 deletions(-) diff --git a/packages/common/src/utils/connection.manager.js b/packages/common/src/utils/connection.manager.js index 38ea85494..f08dda15e 100644 --- a/packages/common/src/utils/connection.manager.js +++ b/packages/common/src/utils/connection.manager.js @@ -1,13 +1,46 @@ -const { registry } = require("./registry"); +const { registry, circuitBreakers } = require("./registry"); const { getPublicIp } = require("./network"); const Project = require("../models/Project"); const { decrypt } = require("./encryption"); const mongoose = require("mongoose"); const redis = require("../config/redis"); +const CB_MAX_FAILURES = 3; +const CB_TIMEOUT_MS = 30000; // 30 seconds + +function recordFailure(key) { + const state = circuitBreakers.get(key) || { failures: 0, openUntil: null }; + state.failures += 1; + if (state.failures >= CB_MAX_FAILURES) { + console.warn(`[CIRCUIT BREAKER] Tripped OPEN for project ${key} due to ${state.failures} failures.`); + state.openUntil = Date.now() + CB_TIMEOUT_MS; + } + circuitBreakers.set(key, state); +} + +function recordSuccess(key) { + if (circuitBreakers.has(key)) { + circuitBreakers.delete(key); + } +} + async function getConnection(projectId) { const key = projectId.toString(); + // 0. Circuit Breaker Check (Fast Fail) + const cbState = circuitBreakers.get(key); + if (cbState && cbState.openUntil) { + if (Date.now() < cbState.openUntil) { + const error = new Error(`Service Unavailable: Database for project ${key} is currently unreachable. Circuit breaker is OPEN.`); + error.status = 503; + throw error; + } else { + // Half-open: Timeout expired, allow retry + cbState.openUntil = null; + circuitBreakers.set(key, cbState); + } + } + // 1. Instant Cache Hit (Fastest Path for UX) if (registry.has(key)) { const cachedConn = registry.get(key); @@ -83,8 +116,11 @@ async function getConnection(projectId) { try { await connection.asPromise(); + recordSuccess(key); // Reset failures on successful connection } catch (err) { console.error("❌ Initial Connection Failed:", err.message); + recordFailure(key); // Increment circuit breaker failures + if (err.message.includes("Server selection timed out") || err.message.includes("Could not connect")) { const serverIp = await getPublicIp(); throw new Error(`Access Denied: Please whitelist Server IP [${serverIp}] in MongoDB Atlas.`); @@ -94,6 +130,7 @@ async function getConnection(projectId) { connection.on("error", (err) => { console.error(`❌ DB Connection Error for project ${projectId}:`, err); + recordFailure(key); registry.delete(key); // Clear bad connections immediately }); @@ -107,7 +144,7 @@ async function getConnection(projectId) { console.log(`🔌 Connection closed: ${key}`); }); - // Save back to your central in-memory store + // Save back to your central in-memory store (LRU handles the capacity limit internally) registry.set(key, connection); return connection; diff --git a/packages/common/src/utils/registry.js b/packages/common/src/utils/registry.js index 767114268..f26974ecd 100644 --- a/packages/common/src/utils/registry.js +++ b/packages/common/src/utils/registry.js @@ -1,3 +1,73 @@ -const registry = new Map(); -const storageRegistry = new Map(); -module.exports = { registry, storageRegistry }; +class LruConnectionMap { + constructor(maxSize, onEvict) { + this.map = new Map(); + this.maxSize = maxSize; + this.onEvict = onEvict; + } + + get(key) { + if (!this.map.has(key)) return undefined; + // Move to end to mark as recently used + const val = this.map.get(key); + this.map.delete(key); + this.map.set(key, val); + return val; + } + + set(key, value) { + if (this.map.has(key)) { + this.map.delete(key); + } else if (this.map.size >= this.maxSize) { + // Evict oldest (least recently used) + const oldestKey = this.map.keys().next().value; + const oldestVal = this.map.get(oldestKey); + this.map.delete(oldestKey); + if (this.onEvict) { + try { + this.onEvict(oldestKey, oldestVal); + } catch (e) { + console.error(`Error during LRU eviction for ${oldestKey}:`, e); + } + } + } + this.map.set(key, value); + } + + has(key) { + return this.map.has(key); + } + + delete(key) { + return this.map.delete(key); + } + + get size() { + return this.map.size; + } + + // Support Iterators for gc.js compatibility + *[Symbol.iterator]() { + yield* this.map[Symbol.iterator](); + } +} + +// Mongoose connection pool registry (limit 50 to protect Node.js RAM/sockets) +const registry = new LruConnectionMap(50, (key, conn) => { + console.log(`[LRU Eviction] Closing idle Mongoose connection for project ${key}`); + if (conn && typeof conn.close === 'function') { + conn.close().catch(err => console.error(`[LRU Eviction Error] Failed to close connection ${key}:`, err)); + } +}); + +// Storage registry (S3/R2 clients, lower overhead so limit is higher) +const storageRegistry = new LruConnectionMap(200, (key, client) => { + console.log(`[LRU Eviction] Removing idle storage client for project ${key}`); + if (client && typeof client.destroy === 'function') { + client.destroy(); + } +}); + +// Circuit Breaker State Registry (No eviction needed, small overhead, or we can use another Map) +const circuitBreakers = new Map(); + +module.exports = { registry, storageRegistry, circuitBreakers, LruConnectionMap }; From 4474f7009bed1e00a3193624fea0cd62df2ce5d8 Mon Sep 17 00:00:00 2001 From: yash-pouranik Date: Fri, 31 Jul 2026 02:44:08 +0530 Subject: [PATCH 2/2] fix(common): circuit breaker race conditions and bounded state registry --- .../common/src/utils/connection.manager.js | 30 +++++++++++++------ packages/common/src/utils/registry.js | 11 +++++-- 2 files changed, 30 insertions(+), 11 deletions(-) diff --git a/packages/common/src/utils/connection.manager.js b/packages/common/src/utils/connection.manager.js index f08dda15e..61a3db497 100644 --- a/packages/common/src/utils/connection.manager.js +++ b/packages/common/src/utils/connection.manager.js @@ -9,8 +9,9 @@ const CB_MAX_FAILURES = 3; const CB_TIMEOUT_MS = 30000; // 30 seconds function recordFailure(key) { - const state = circuitBreakers.get(key) || { failures: 0, openUntil: null }; + const state = circuitBreakers.get(key) || { failures: 0, openUntil: null, probeInFlight: false }; state.failures += 1; + state.probeInFlight = false; if (state.failures >= CB_MAX_FAILURES) { console.warn(`[CIRCUIT BREAKER] Tripped OPEN for project ${key} due to ${state.failures} failures.`); state.openUntil = Date.now() + CB_TIMEOUT_MS; @@ -20,6 +21,8 @@ function recordFailure(key) { function recordSuccess(key) { if (circuitBreakers.has(key)) { + const state = circuitBreakers.get(key); + state.probeInFlight = false; circuitBreakers.delete(key); } } @@ -29,14 +32,21 @@ async function getConnection(projectId) { // 0. Circuit Breaker Check (Fast Fail) const cbState = circuitBreakers.get(key); - if (cbState && cbState.openUntil) { - if (Date.now() < cbState.openUntil) { + if (cbState) { + if (cbState.openUntil && Date.now() < cbState.openUntil) { const error = new Error(`Service Unavailable: Database for project ${key} is currently unreachable. Circuit breaker is OPEN.`); error.status = 503; throw error; - } else { - // Half-open: Timeout expired, allow retry + } else if (cbState.openUntil && Date.now() >= cbState.openUntil) { + if (cbState.probeInFlight) { + // Reject concurrent callers while the single half-open probe is in flight + const error = new Error(`Service Unavailable: Database for project ${key} is currently unreachable. Circuit breaker is probing.`); + error.status = 503; + throw error; + } + // Half-open: Timeout expired, allow one retry probe cbState.openUntil = null; + cbState.probeInFlight = true; circuitBreakers.set(key, cbState); } } @@ -84,6 +94,7 @@ async function getConnection(projectId) { plan = projectDoc.plan || 'free'; } catch (err) { console.error("Decryption Error:", err); + // Treat decryption errors as hard failures, don't trip breaker for user config errors, but we can't connect anyway throw new Error("Invalid or corrupted external config"); } @@ -130,18 +141,19 @@ async function getConnection(projectId) { connection.on("error", (err) => { console.error(`❌ DB Connection Error for project ${projectId}:`, err); - recordFailure(key); - registry.delete(key); // Clear bad connections immediately + if (registry.deleteIfCurrent(key, connection)) { + recordFailure(key); // Only record failure if it was the active connection + } }); connection.on("disconnected", () => { console.log(`🔌 External DB disconnected: ${projectId}`); - registry.delete(key); + registry.deleteIfCurrent(key, connection); }); connection.on("close", () => { - registry.delete(key); console.log(`🔌 Connection closed: ${key}`); + registry.deleteIfCurrent(key, connection); }); // Save back to your central in-memory store (LRU handles the capacity limit internally) diff --git a/packages/common/src/utils/registry.js b/packages/common/src/utils/registry.js index f26974ecd..d6a0f6981 100644 --- a/packages/common/src/utils/registry.js +++ b/packages/common/src/utils/registry.js @@ -40,6 +40,13 @@ class LruConnectionMap { delete(key) { return this.map.delete(key); } + + deleteIfCurrent(key, expectedValue) { + if (this.map.has(key) && this.map.get(key) === expectedValue) { + return this.map.delete(key); + } + return false; + } get size() { return this.map.size; @@ -67,7 +74,7 @@ const storageRegistry = new LruConnectionMap(200, (key, client) => { } }); -// Circuit Breaker State Registry (No eviction needed, small overhead, or we can use another Map) -const circuitBreakers = new Map(); +// Circuit Breaker State Registry (Capped to prevent memory leak on inactive projects) +const circuitBreakers = new LruConnectionMap(5000); module.exports = { registry, storageRegistry, circuitBreakers, LruConnectionMap };