fix: improve cartographer-to-lighthouse Redis connectivity

Author: otsybizov

ops/mainnet/staging/core/main.tf

@@ -542,6 +542,7 @@ module "lighthouse_queue_cache" {
   public_redis                  = true
   transit_encryption_enabled    = true
   auth_token                    = var.lighthouse_queue_redis_auth_token
+  maxmemory_policy              = "noeviction"
 }
 
 # Expose the lighthouse queue Redis over PrivateLink so the cartographer-handler

ops/modules/redis/main.tf

@@ -3,14 +3,32 @@ locals {
   use_replication_group = var.transit_encryption_enabled
 }
 
+# BullMQ requires maxmemory-policy = noeviction. The default ElastiCache
+# parameter group uses volatile-lru, which can evict keys BullMQ depends on.
+resource "aws_elasticache_parameter_group" "redis" {
+  count  = var.maxmemory_policy != null ? 1 : 0
+  name   = "redis-params-${var.environment}-${var.stage}-${var.family}"
+  family = var.parameter_group_family
+
+  parameter {
+    name  = "maxmemory-policy"
+    value = var.maxmemory_policy
+  }
+
+  tags = {
+    Stage       = var.stage
+    Environment = var.environment
+  }
+}
+
 resource "aws_elasticache_cluster" "redis" {
   count           = local.use_replication_group ? 0 : 1
   cluster_id      = "redis-cluster-${var.environment}-${var.stage}-${var.family}"
   engine          = "redis"
   node_type       = var.node_type
   num_cache_nodes = 1
-  # parameter_group_name = "default.redis6.x"
-  # engine_version           = "6.x"
+  parameter_group_name     = var.maxmemory_policy != null ? aws_elasticache_parameter_group.redis[0].name : null
+  engine_version  = var.engine_version
   port                     = 6379
   subnet_group_name        = aws_elasticache_subnet_group.default.name
   security_group_ids       = [aws_security_group.redis.id]
@@ -28,6 +46,8 @@ resource "aws_elasticache_replication_group" "redis" {
   description                = "Redis replication group for ${var.family} (${var.environment}-${var.stage})"
   node_type                  = var.node_type
   num_cache_clusters         = 1
+  parameter_group_name       = var.maxmemory_policy != null ? aws_elasticache_parameter_group.redis[0].name : null
+  engine_version             = var.engine_version
   port                       = 6379
   subnet_group_name          = aws_elasticache_subnet_group.default.name
   security_group_ids         = [aws_security_group.redis.id]

ops/modules/redis/variables.tf

@@ -49,3 +49,21 @@ variable "transit_encryption_enabled" {
   default     = false
   type        = bool
 }
+
+variable "maxmemory_policy" {
+  description = "Redis maxmemory-policy. Set to 'noeviction' for BullMQ workloads. When null, the default parameter group is used."
+  default     = null
+  type        = string
+}
+
+variable "parameter_group_family" {
+  description = "ElastiCache parameter group family (e.g. redis7, redis6.x). Must match the engine version."
+  default     = "redis7"
+  type        = string
+}
+
+variable "engine_version" {
+  description = "Redis OSS engine_version for ElastiCache (e.g. 7.1). Must align with parameter_group_family (use redis7 for 7.x)."
+  default     = "7.1"
+  type        = string
+}

packages/adapters/mqclient/.nycrc.json

@@ -0,0 +1,10 @@
+{
+  "extends": "@istanbuljs/nyc-config-typescript",
+  "all": true,
+  "check-coverage": true,
+  "branches": 60,
+  "lines": 50,
+  "functions": 60,
+  "statements": 50,
+  "exclude": ["dist/**", "test/**"]
+}

packages/adapters/mqclient/package.json

@@ -13,14 +13,21 @@
     "lint:fix": "yarn lint:src --fix",
     "clean": "rimraf ./dist ./tsconfig.tsBuildInfo",
     "build": "tsc --build ./tsconfig.json",
+    "test": "nyc ts-mocha --require 'test/globalTestHook.ts' --check-leaks --exit --timeout 60000 'test/**/*.spec.ts'",
+    "test:unit": "nyc ts-mocha --require 'test/globalTestHook.ts' --check-leaks --exit --timeout 60000 'test/**/*.spec.ts'",
     "purge": "yarn clean && rimraf ./node_modules"
   },
   "dependencies": {
+    "@chimera-monorepo/utils": "workspace:*",
     "bullmq": "5.34.8"
   },
   "devDependencies": {
+    "@istanbuljs/nyc-config-typescript": "1.0.2",
     "eslint": "8.34.0",
+    "nyc": "15.1.0",
     "rimraf": "5.0.1",
+    "sinon": "15.0.1",
+    "ts-mocha": "10.0.0",
     "typescript": "5.9.2"
   }
 }

packages/adapters/mqclient/src/index.ts

@@ -1,4 +1,5 @@
 import { Queue, Worker, Job, ConnectionOptions, QueueOptions, WorkerOptions } from 'bullmq';
+import { jsonifyError, Logger } from '@chimera-monorepo/utils';
 
 export { Queue, Worker, Job } from 'bullmq';
 
@@ -24,15 +25,38 @@ export const parseRedisUrl = (redisUrl: string): ConnectionOptions => {
     ...(url.username ? { username: url.username } : {}),
     ...(url.protocol === 'rediss:' ? { tls: tlsServername ? { servername: tlsServername } : {} } : {}),
     connectTimeout: 17_000,
-    maxRetriesPerRequest: 4,
-    retryStrategy: (times: number) => Math.min(times * 30, 1000),
+    maxRetriesPerRequest: null,
+    retryStrategy: (times: number) => Math.min(times * 50, 5_000),
     keepAlive: 30_000,
   };
 };
 
-export const createProducer = (redisUrl: string, queueName: string, opts?: Partial<QueueOptions>): Queue => {
+export const pingRedis = async (queueOrWorker: Queue | Worker, timeoutMs = 3_000): Promise<boolean> => {
+  try {
+    const client = await Promise.race([
+      queueOrWorker.client,
+      new Promise<never>((_, reject) => setTimeout(() => reject(new Error('timeout')), timeoutMs)),
+    ]);
+    if (client.status !== 'ready') return false;
+
+    const result = await Promise.race([
+      client.ping(),
+      new Promise<never>((_, reject) => setTimeout(() => reject(new Error('timeout')), timeoutMs)),
+    ]);
+    return result === 'PONG';
+  } catch {
+    return false;
+  }
+};
+
+export const createProducer = (
+  redisUrl: string,
+  queueName: string,
+  logger: Logger,
+  opts?: Partial<QueueOptions>,
+): Queue => {
   const connection = parseRedisUrl(redisUrl);
-  return new Queue(queueName, {
+  const queue = new Queue(queueName, {
     connection,
     defaultJobOptions: {
       attempts: 3,
@@ -42,19 +66,40 @@ export const createProducer = (redisUrl: string, queueName: string, opts?: Parti
     },
     ...opts,
   });
+  queue.on('error', (err) =>
+    logger.error(`Queue "${queueName}" Redis connection error`, undefined, undefined, jsonifyError(err), {
+      queueName,
+      role: 'producer',
+    }),
+  );
+  return queue;
 };
 
 export const createWorker = (
   redisUrl: string,
   queueName: string,
   processor: (job: Job) => Promise<void>,
+  logger: Logger,
   opts?: Partial<WorkerOptions>,
 ): Worker => {
   const connection = parseRedisUrl(redisUrl);
-  return new Worker(queueName, processor, {
+  const worker = new Worker(queueName, processor, {
     connection,
     concurrency: 1,
     lockDuration: 300_000, // 5 minutes — prevents stale-lock re-processing for long tasks
     ...opts,
   });
+  worker.on('ready', () =>
+    logger.info(`Worker "${queueName}" Redis connected`, undefined, undefined, {
+      queueName,
+      role: 'worker',
+    }),
+  );
+  worker.on('error', (err) =>
+    logger.error(`Worker "${queueName}" Redis connection error`, undefined, undefined, jsonifyError(err), {
+      queueName,
+      role: 'worker',
+    }),
+  );
+  return worker;
 };

packages/adapters/mqclient/test/globalTestHook.ts

@@ -0,0 +1,8 @@
+import { restore, reset } from 'sinon';
+
+export const mochaHooks = {
+  afterEach() {
+    restore();
+    reset();
+  },
+};