fix: resolve issues #246, #247, #248, #249

.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.

src/index.js

@@ -98,6 +98,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();
@@ -136,6 +137,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: {

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';
   }

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,
+};

src/services/webhookDispatcher.js

@@ -124,14 +124,15 @@ 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,
     'X-SmartDrop-Event': eventType,
     '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 };

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<function>} 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;