From 4a70a3c748d2efcee892b5746c7e2ac1ce9411c0 Mon Sep 17 00:00:00 2001 From: smartdev Date: Thu, 27 Aug 2026 15:38:33 +0100 Subject: [PATCH] fix: resolve issues #246, #247, #248, #249 - #246: Add FORBIDDEN to AppError codes so CORS origin validation returns detailed error messages with the rejected origin - #247: Add X-SmartDrop-Sequence header to webhook deliveries for ordering guarantees; sequence is persisted and propagated through retries - #248: Enhance WebSocket graceful shutdown with pre-close warning broadcast, reject new connections during drain, and track drain statistics - #249: Add Redis concurrency semaphore and backpressure to prevent connection pool exhaustion; expose queue depth and concurrency stats in /health endpoint --- .env.example | 12 ++++ src/errors/AppError.js | 1 + src/index.js | 7 ++ src/middleware/errorHandler.js | 2 +- src/services/cache.js | 108 +++++++++++++++++++++++------ src/services/webhookDispatcher.js | 24 ++++--- src/utils/semaphore.js | 70 +++++++++++++++++++ src/ws/PriceSubscriptionManager.js | 50 +++++++++++-- test/health.test.js | 6 ++ test/helpers/cacheMock.js | 2 + 10 files changed, 242 insertions(+), 40 deletions(-) create mode 100644 src/utils/semaphore.js diff --git a/.env.example b/.env.example index c597c29..61ad1c5 100644 --- a/.env.example +++ b/.env.example @@ -99,6 +99,14 @@ PRICE_RATELIMIT_WINDOW=60 # PRICE_RATELIMIT_MAX: max price requests per IP per window. Default: 30. PRICE_RATELIMIT_MAX=30 +# Redis concurrency and backpressure (issue #249) +# REDIS_MAX_CONCURRENT_OPS: max concurrent Redis operations via semaphore. Default: 50. +REDIS_MAX_CONCURRENT_OPS=50 +# REDIS_COMMAND_QUEUE_WARN_THRESHOLD: log warning when command queue exceeds this. Default: 100. +REDIS_COMMAND_QUEUE_WARN_THRESHOLD=100 +# REDIS_COMMAND_QUEUE_BACKPRESSURE_THRESHOLD: log error and apply backpressure above this. Default: 500. +REDIS_COMMAND_QUEUE_BACKPRESSURE_THRESHOLD=500 + # Rate limiting (Redis-backed, per API key) # Each authenticated key is metered in its own bucket sized by its tier, so # one abusive key cannot exhaust the shared per-IP bucket for everybody else. @@ -113,3 +121,7 @@ API_KEY_RATELIMIT_ADMIN_MAX=10000 # CORS CORS_ALLOWED_ORIGINS=http://localhost:4000,http://localhost:3001 + +# WebSocket +WS_MAX_CONNECTIONS=100 +WS_MAX_CONNECTIONS_PER_IP=5 diff --git a/src/errors/AppError.js b/src/errors/AppError.js index 35c636b..fd6d3ec 100644 --- a/src/errors/AppError.js +++ b/src/errors/AppError.js @@ -3,6 +3,7 @@ const ERROR_CODES = Object.freeze({ VALIDATION_ERROR: { statusCode: 400 }, UNAUTHORIZED: { statusCode: 401 }, + FORBIDDEN: { statusCode: 403 }, NOT_FOUND: { statusCode: 404 }, PAYLOAD_TOO_LARGE: { statusCode: 413 }, RATE_LIMITED: { statusCode: 429 }, diff --git a/src/index.js b/src/index.js index 6131043..36e42d1 100644 --- a/src/index.js +++ b/src/index.js @@ -96,6 +96,7 @@ async function readWebhookRetryQueueStats() { app.get('/health', async (req, res) => { const redisConnected = cache.isConnected(); const redisQueueDepth = cache.getCommandQueueLength(); + const redisConcurrency = cache.getConcurrencyStats(); const priceRefreshHealth = wrappedPriceRefreshJob.getHealth(); const webhookWorkerHealth = wrappedWebhookRetryWorker.getHealth(); const airdropExpiryHealth = wrappedAirdropExpiryJob.getHealth(); @@ -134,6 +135,12 @@ app.get('/health', async (req, res) => { redis: { connected: redisConnected, command_queue_depth: redisQueueDepth, + concurrency: redisConcurrency, + }, + websocket: { + connections: subscriptionManager.connectionCount, + draining: subscriptionManager.isDraining, + drain_stats: subscriptionManager.drainStats, }, jobs: { price_refresh: { diff --git a/src/middleware/errorHandler.js b/src/middleware/errorHandler.js index c614f21..0021f96 100644 --- a/src/middleware/errorHandler.js +++ b/src/middleware/errorHandler.js @@ -25,7 +25,7 @@ function errorHandler(err, req, res, _next) { message = 'Request body is too large'; } else if (err.status || err.statusCode) { status = err.status || err.statusCode; - const STATUS_CODES = { 400: 'VALIDATION_ERROR', 401: 'UNAUTHORIZED', 403: 'FORBIDDEN', 404: 'NOT_FOUND', 429: 'RATE_LIMITED' }; + const STATUS_CODES = { 400: 'VALIDATION_ERROR', 401: 'UNAUTHORIZED', 403: 'FORBIDDEN', 404: 'NOT_FOUND', 413: 'PAYLOAD_TOO_LARGE', 429: 'RATE_LIMITED' }; code = STATUS_CODES[status] || 'INTERNAL_ERROR'; message = err.message || 'Request rejected'; } diff --git a/src/services/cache.js b/src/services/cache.js index b5c19d9..f7b4a51 100644 --- a/src/services/cache.js +++ b/src/services/cache.js @@ -1,16 +1,57 @@ const Redis = require('ioredis'); const config = require('../config'); const logger = require('../logger'); +const Semaphore = require('../utils/semaphore'); const MAX_RETRIES = 10; const RETRY_DELAY_MS = 1000; const CONNECT_TIMEOUT_MS = 5000; const COMMAND_TIMEOUT_MS = 3000; const COMMAND_QUEUE_WARN_THRESHOLD = parseInt(process.env.REDIS_COMMAND_QUEUE_WARN_THRESHOLD, 10) || 100; +const COMMAND_QUEUE_BACKPRESSURE_THRESHOLD = parseInt(process.env.REDIS_COMMAND_QUEUE_BACKPRESSURE_THRESHOLD, 10) || 500; let client = null; let reconnectAttempts = 0; +// Concurrency limiter to prevent Redis connection pool exhaustion (issue #249). +// Limits concurrent in-flight Redis commands to prevent queue buildup. +const MAX_CONCURRENT_OPS = parseInt(process.env.REDIS_MAX_CONCURRENT_OPS, 10) || 50; +const operationSemaphore = new Semaphore(MAX_CONCURRENT_OPS); + +let consecutiveQueueWarnings = 0; + +function _checkQueueBackpressure(caller) { + const queueLen = getCommandQueueLength(); + if (queueLen > COMMAND_QUEUE_BACKPRESSURE_THRESHOLD) { + consecutiveQueueWarnings++; + if (consecutiveQueueWarnings % 10 === 1) { + logger.error('Redis command queue critically deep — backpressure active', { + queue_length: queueLen, + threshold: COMMAND_QUEUE_BACKPRESSURE_THRESHOLD, + caller, + consecutive_warnings: consecutiveQueueWarnings, + }); + } + return true; + } + if (queueLen > COMMAND_QUEUE_WARN_THRESHOLD) { + consecutiveQueueWarnings++; + if (consecutiveQueueWarnings % 5 === 1) { + logger.warn('Redis command queue depth high', { + queue_length: queueLen, + threshold: COMMAND_QUEUE_WARN_THRESHOLD, + caller, + }); + } + return false; + } + if (consecutiveQueueWarnings > 0) { + logger.info('Redis command queue depth recovered', { queue_length: queueLen, caller }); + consecutiveQueueWarnings = 0; + } + return false; +} + function getClient() { if (!client) { client = new Redis(config.redis.url, { @@ -59,38 +100,56 @@ function getCommandQueueLength() { return client.commandQueue ? client.commandQueue.length : 0; } +function getConcurrencyStats() { + return { + active: operationSemaphore.active, + waiting: operationSemaphore.waiting, + available: operationSemaphore.available, + max: MAX_CONCURRENT_OPS, + }; +} + async function get(key) { - const redis = getClient(); - const queueLen = getCommandQueueLength(); - if (queueLen > COMMAND_QUEUE_WARN_THRESHOLD) { - logger.warn('Redis command queue depth high', { queue_length: queueLen, threshold: COMMAND_QUEUE_WARN_THRESHOLD }); - } - const data = await redis.get(key); - if (!data) return null; + const release = await operationSemaphore.acquire(5000); try { - return JSON.parse(data); - } catch { - return data; + _checkQueueBackpressure('get'); + const redis = getClient(); + const data = await redis.get(key); + if (!data) return null; + try { + return JSON.parse(data); + } catch { + return data; + } + } finally { + release(); } } async function set(key, value, ttlSeconds) { - const redis = getClient(); - const queueLen = getCommandQueueLength(); - if (queueLen > COMMAND_QUEUE_WARN_THRESHOLD) { - logger.warn('Redis command queue depth high', { queue_length: queueLen, threshold: COMMAND_QUEUE_WARN_THRESHOLD }); - } - const serialized = JSON.stringify(value); - if (ttlSeconds) { - await redis.setex(key, ttlSeconds, serialized); - } else { - await redis.set(key, serialized); + const release = await operationSemaphore.acquire(5000); + try { + _checkQueueBackpressure('set'); + const redis = getClient(); + const serialized = JSON.stringify(value); + if (ttlSeconds) { + await redis.setex(key, ttlSeconds, serialized); + } else { + await redis.set(key, serialized); + } + } finally { + release(); } } async function del(key) { - const redis = getClient(); - await redis.del(key); + const release = await operationSemaphore.acquire(5000); + try { + const redis = getClient(); + await redis.del(key); + } finally { + release(); + } } async function disconnect() { @@ -100,4 +159,7 @@ async function disconnect() { } } -module.exports = { get, set, del, disconnect, getClient, isConnected, getCommandQueueLength }; +module.exports = { + get, set, del, disconnect, getClient, isConnected, + getCommandQueueLength, getConcurrencyStats, +}; diff --git a/src/services/webhookDispatcher.js b/src/services/webhookDispatcher.js index f6f3d6f..7bf0b7a 100644 --- a/src/services/webhookDispatcher.js +++ b/src/services/webhookDispatcher.js @@ -124,7 +124,7 @@ function shouldRetry(responseStatus, networkError) { return false; } -function buildHeaders(secret, body, eventType, deliveryId, requestId) { +function buildHeaders(secret, body, eventType, deliveryId, requestId, sequence) { const headers = { 'Content-Type': 'application/json', 'User-Agent': USER_AGENT, @@ -132,6 +132,7 @@ function buildHeaders(secret, body, eventType, deliveryId, requestId) { 'X-SmartDrop-Delivery': deliveryId, 'X-SmartDrop-Signature': signature.sign(secret, body), }; + if (sequence != null) headers['X-SmartDrop-Sequence'] = String(sequence); // Lets receivers correlate a delivery with the API request that caused // it when reporting problems back to us (issue #250). if (requestId) headers['X-Request-Id'] = requestId; @@ -175,7 +176,7 @@ async function postOnce(url, headers, body, timeoutMs) { }); } -async function attempt(deliveryId) { +async function attempt(deliveryId, sequence) { const delivery = await deliveryRepo.findById(deliveryId); if (!delivery) { logger.warn('Delivery missing, dropping retry', { delivery_id: deliveryId }); @@ -206,7 +207,8 @@ async function attempt(deliveryId) { occurred_at: delivery.created_at, }; const body = JSON.stringify(payload); - const headers = buildHeaders(webhook.secret, body, delivery.event_type, delivery.id, delivery.request_id); + const seq = sequence ?? delivery.sequence; + const headers = buildHeaders(webhook.secret, body, delivery.event_type, delivery.id, delivery.request_id, seq); const attempts = delivery.attempts + 1; let responseStatus = null; @@ -296,7 +298,7 @@ async function attempt(deliveryId) { }); } -async function deliverToWebhook(webhook, eventType, eventId, payload) { +async function deliverToWebhook(webhook, eventType, eventId, payload, sequence) { // Propagate the originating request's id onto the delivery record so a // webhook that fires hours later on a retry is still traceable back to // the API call that caused it (issue #250). @@ -307,19 +309,19 @@ async function deliverToWebhook(webhook, eventType, eventId, payload) { event_type: eventType, request_id: requestId && requestId !== 'system' ? requestId : null, }); - await deliveryRepo.update(delivery.id, { payload }); - return attempt(delivery.id); + await deliveryRepo.update(delivery.id, { payload, sequence }); + return attempt(delivery.id, sequence); } const DISPATCH_CONCURRENCY = parseInt(process.env.WEBHOOK_DISPATCH_CONCURRENCY, 10) || 10; const ORDERED_DELIVERY = process.env.WEBHOOK_ORDERED_DELIVERY === 'true'; -async function processBatch(batch, eventType, eventId, payload) { +async function processBatch(batch, eventType, eventId, payload, sequence) { if (ORDERED_DELIVERY) { const results = []; for (const webhook of batch) { try { - const value = await deliverToWebhook(webhook, eventType, eventId, payload); + const value = await deliverToWebhook(webhook, eventType, eventId, payload, sequence); results.push({ status: 'fulfilled', value }); } catch (reason) { results.push({ status: 'rejected', reason }); @@ -328,7 +330,7 @@ async function processBatch(batch, eventType, eventId, payload) { return results; } return Promise.allSettled( - batch.map((webhook) => deliverToWebhook(webhook, eventType, eventId, payload)) + batch.map((webhook) => deliverToWebhook(webhook, eventType, eventId, payload, sequence)) ); } @@ -368,7 +370,7 @@ async function dispatch({ event_type: eventType, event_id: eventId, data }) { const allResults = []; for (let i = 0; i < targets.length; i += DISPATCH_CONCURRENCY) { const batch = targets.slice(i, i + DISPATCH_CONCURRENCY); - const batchResults = await processBatch(batch, eventType, eventId, payload); + const batchResults = await processBatch(batch, eventType, eventId, payload, sequence); allResults.push(...batchResults); } @@ -394,7 +396,7 @@ async function sendTest(webhookId) { occurred_at: new Date().toISOString(), data: { test: true, message: 'This is a test delivery from SmartDrop' }, }; - return deliverToWebhook(webhook, eventType, payload.event_id, payload); + return deliverToWebhook(webhook, eventType, payload.event_id, payload, null); } module.exports = { dispatch, attempt, sendTest, backoffMs, shouldRetry, getMetrics, getInFlightCount }; diff --git a/src/utils/semaphore.js b/src/utils/semaphore.js new file mode 100644 index 0000000..a216713 --- /dev/null +++ b/src/utils/semaphore.js @@ -0,0 +1,70 @@ +'use strict'; + +/** + * A simple counting semaphore for limiting concurrent access to a resource. + * Used to prevent Redis connection pool exhaustion under high load (issue #249). + */ +class Semaphore { + constructor(maxConcurrency) { + this._maxConcurrency = maxConcurrency; + this._current = 0; + this._queue = []; + } + + get available() { + return this._maxConcurrency - this._current; + } + + get waiting() { + return this._queue.length; + } + + get active() { + return this._current; + } + + /** + * Acquire a permit. Resolves when a permit is available, or immediately + * if one is already available. Call release() when done. + * @param {number} [timeoutMs] - Max time to wait for a permit. 0 = no wait. + * @returns {Promise} A release function. + */ + acquire(timeoutMs = 0) { + if (this._current < this._maxConcurrency) { + this._current++; + return Promise.resolve(() => this._release()); + } + + if (timeoutMs === 0) { + return Promise.reject(new Error('Semaphore: no permits available')); + } + + return new Promise((resolve, reject) => { + const entry = { resolve: () => { + this._current++; + resolve(() => this._release()); + }, reject }; + + if (timeoutMs > 0) { + entry.timer = setTimeout(() => { + const idx = this._queue.indexOf(entry); + if (idx !== -1) this._queue.splice(idx, 1); + reject(new Error(`Semaphore: timed out after ${timeoutMs}ms`)); + }, timeoutMs); + } + + this._queue.push(entry); + }); + } + + _release() { + this._current--; + if (this._queue.length > 0) { + const next = this._queue.shift(); + if (next.timer) clearTimeout(next.timer); + next.resolve(); + } + } +} + +module.exports = Semaphore; diff --git a/src/ws/PriceSubscriptionManager.js b/src/ws/PriceSubscriptionManager.js index e166f0d..9d8869d 100644 --- a/src/ws/PriceSubscriptionManager.js +++ b/src/ws/PriceSubscriptionManager.js @@ -39,6 +39,8 @@ class PriceSubscriptionManager { this._connectionsByIp = new Map(); // ip → number this._previousPrices = new Map(); // assetKey → number this._pingTimer = null; + this._draining = false; + this._drainStats = { warned: 0, closed: 0, forceClosed: 0 }; } _getClientIp(req) { @@ -53,8 +55,13 @@ class PriceSubscriptionManager { return socket?.remoteAddress || 'unknown'; } - /** Register a new WebSocket connection. Returns false when at capacity. */ + /** Register a new WebSocket connection. Returns false when at capacity or draining. */ add(ws, req = {}) { + if (this._draining) { + ws.close(1013, 'Server shutting down'); + return false; + } + const clientIp = this._getClientIp(req).replace(/^::ffff:/, ''); const currentByIp = this._connectionsByIp.get(clientIp) || 0; @@ -208,34 +215,59 @@ class PriceSubscriptionManager { /** * Gracefully drain all connected clients during server shutdown. - * Sends a close frame with a warning reason, then force-closes any - * connections still open after `drainTimeoutMs` (issue #248). + * Broadcasts a shutdown warning, then sends close frames, and force-closes + * any connections still open after `drainTimeoutMs` (issue #248). */ drain(drainTimeoutMs = 5000) { this.stopHeartbeat(); + this._draining = true; const clientCount = this._clients.size; if (clientCount === 0) return Promise.resolve(); + this._drainStats = { warned: clientCount, closed: 0, forceClosed: 0 }; logger.info('Draining WebSocket connections', { count: clientCount, drain_timeout_ms: drainTimeoutMs }); + // Phase 1: Broadcast shutdown warning so clients can prepare for (const [ws] of this._clients) { try { - ws.close(1001, 'Server shutting down'); + this._send(ws, { type: 'server_shutdown', message: 'Server is shutting down', drain_timeout_ms: drainTimeoutMs }); } catch { // already closed or errored — ignore } } + // Phase 2: After a brief grace period for clients to finish in-flight work, + // send close frames to initiate orderly disconnection + const closeDelayMs = Math.min(1000, drainTimeoutMs / 2); return new Promise((resolve) => { + const closeTimer = setTimeout(() => { + for (const [ws] of this._clients) { + try { + ws.close(1001, 'Server shutting down'); + this._drainStats.closed++; + } catch { + // already closed or errored — ignore + } + } + }, closeDelayMs); + + closeTimer.unref(); + const deadline = setTimeout(() => { + const remaining = this._clients.size; for (const [ws] of this._clients) { try { ws.terminate(); } catch { /* ignore */ } + this._drainStats.forceClosed++; } this._clients.clear(); this._clientIpBySocket.clear(); this._connectionsByIp.clear(); updateGauge(-clientCount); - logger.info('WebSocket drain complete (force-closed remaining)', { force_closed: clientCount }); + logger.info('WebSocket drain complete', { + total: clientCount, + gracefully_closed: this._drainStats.closed, + force_closed: remaining, + }); resolve(); }, drainTimeoutMs); @@ -246,6 +278,14 @@ class PriceSubscriptionManager { get connectionCount() { return this._clients.size; } + + get isDraining() { + return this._draining; + } + + get drainStats() { + return { ...this._drainStats }; + } } module.exports = new PriceSubscriptionManager(); diff --git a/test/health.test.js b/test/health.test.js index 14fbd37..c320d94 100644 --- a/test/health.test.js +++ b/test/health.test.js @@ -5,6 +5,7 @@ const request = require('supertest'); jest.mock('../src/services/cache', () => ({ isConnected: jest.fn(() => false), disconnect: jest.fn(), + getConcurrencyStats: jest.fn(() => ({ active: 0, waiting: 0, available: 50, max: 50 })), })); jest.mock('../src/services/priceOracle', () => ({ @@ -145,6 +146,7 @@ describe('GET /health – status computation', () => { jest.mock('../src/services/cache', () => ({ isConnected: () => true, disconnect: jest.fn(), + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), })); jest.mock('../src/jobs/priceRefresh', () => ({ start: jest.fn(), @@ -170,6 +172,7 @@ describe('GET /health – status computation', () => { jest.mock('../src/services/cache', () => ({ isConnected: () => false, disconnect: jest.fn(), + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), })); jest.mock('../src/jobs/priceRefresh', () => ({ start: jest.fn(), @@ -196,6 +199,7 @@ describe('GET /health – status computation', () => { jest.mock('../src/services/cache', () => ({ isConnected: () => true, disconnect: jest.fn(), + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), })); jest.mock('../src/jobs/priceRefresh', () => ({ start: jest.fn(), @@ -223,6 +227,7 @@ describe('GET /health – status computation', () => { jest.mock('../src/services/cache', () => ({ isConnected: () => true, disconnect: jest.fn(), + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), })); jest.mock('../src/jobs/priceRefresh', () => ({ start: jest.fn(), @@ -249,6 +254,7 @@ describe('GET /health – status computation', () => { jest.mock('../src/services/cache', () => ({ isConnected: () => false, disconnect: jest.fn(), + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), })); jest.mock('../src/jobs/priceRefresh', () => ({ start: jest.fn(), diff --git a/test/helpers/cacheMock.js b/test/helpers/cacheMock.js index 0727344..402311c 100644 --- a/test/helpers/cacheMock.js +++ b/test/helpers/cacheMock.js @@ -177,6 +177,8 @@ function createCacheMock() { const cacheMock = { getClient: () => redis, isConnected: () => true, + getCommandQueueLength: () => 0, + getConcurrencyStats: () => ({ active: 0, waiting: 0, available: 50, max: 50 }), get: jest.fn(async (key) => { const v = store.get(key); return v !== undefined ? JSON.parse(JSON.stringify(v)) : null;