feat: harden the Redis connection for distributed runs

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@factorialco/specbandit",
-  "version": "1.0.1",
+  "version": "1.1.0",
   "description": "Distributed test runner using Redis as a work queue. Push file paths to a Redis list, then multiple CI runners atomically steal batches and execute them via a configurable command.",
   "author": "Ferran Basora",
   "license": "MIT",

src/cli.ts

@@ -1,6 +1,6 @@
 import { Configuration, SpecbanditError, VERSION } from './configuration.js'
 import { Publisher } from './publisher.js'
-import { RedisQueue } from './redisQueue.js'
+import { RedisQueue, type RedisQueueOptions } from './redisQueue.js'
 import { Worker } from './worker.js'
 import { CliAdapter } from './cliAdapter.js'
 import { JestAdapter } from './jestAdapter.js'
@@ -53,6 +53,16 @@ function parseArgs(argv: string[]): { flags: Record<string, string>; positional:
   return { flags, positional }
 }
 
+/** Build the ioredis resilience options from a resolved Configuration. */
+function redisOptions(config: Configuration): RedisQueueOptions {
+  return {
+    maxAttempts: config.redisMaxAttempts,
+    connectTimeout: config.redisConnectTimeout,
+    commandTimeout: config.redisCommandTimeout,
+    reconnectAttempts: config.redisReconnectAttempts,
+  }
+}
+
 async function runPush(argv: string[]): Promise<number> {
   const { flags, positional } = parseArgs(argv)
 
@@ -72,10 +82,11 @@ Options:
     key: flags.key,
     redisUrl: flags['redis-url'],
     keyTtl: flags['key-ttl'] ? parseInt(flags['key-ttl'], 10) : undefined,
+    redisMaxAttempts: flags['redis-max-attempts'] ? parseInt(flags['redis-max-attempts'], 10) : undefined,
   })
   config.validate()
 
-  const queue = new RedisQueue(config.redisUrl)
+  const queue = new RedisQueue(config.redisUrl, redisOptions(config))
   try {
     const publisher = new Publisher({
       key: config.key!,
@@ -155,6 +166,7 @@ Options:
   --project-root PATH    Project root directory (for jest/cypress adapters, default: cwd)
   --batch-size N         Files per batch (default: 5)
   --redis-url URL        Redis URL (default: redis://localhost:6379)
+  --redis-max-attempts N Redis connection retry attempts (default: 5)
   --key-rerun KEY        Per-runner rerun key for re-run support
   --key-failed KEY       Redis key to store failed test file paths for later review
   --key-ttl SECONDS      TTL for all Redis keys (default: 604800 / 1 week)
@@ -192,6 +204,7 @@ Adapters:
     keyFailed: flags['key-failed'],
     keyTtl: flags['key-ttl'] ? parseInt(flags['key-ttl'], 10) : undefined,
     verbose: flags.verbose === 'true' ? true : undefined,
+    redisMaxAttempts: flags['redis-max-attempts'] ? parseInt(flags['redis-max-attempts'], 10) : undefined,
   })
 
   // Only validate command requirement for CLI adapter
@@ -204,7 +217,7 @@ Adapters:
   }
 
   const adapter = buildAdapter(flags, config)
-  const queue = new RedisQueue(config.redisUrl)
+  const queue = new RedisQueue(config.redisUrl, redisOptions(config))
 
   try {
     const worker = new Worker({
@@ -276,6 +289,10 @@ Environment variables:
   SPECBANDIT_KEY_TTL          TTL for all Redis keys in seconds (default: 604800)
   SPECBANDIT_VERBOSE          Enable verbose output (1/true/yes)
   SPECBANDIT_REPORT           Path to write JSON report file
+  SPECBANDIT_REDIS_MAX_ATTEMPTS      Redis connection retry attempts (default: 5)
+  SPECBANDIT_REDIS_CONNECT_TIMEOUT   Redis connect timeout, seconds (default: 3)
+  SPECBANDIT_REDIS_TIMEOUT           Redis command timeout, seconds (default: 5)
+  SPECBANDIT_REDIS_RECONNECT_ATTEMPTS Redis reconnect/per-request retries (default: 3)
 
 File input priority for push:
   1. stdin (piped)     echo "test/a.test.ts" | specbandit push --key KEY

src/configuration.ts

@@ -17,12 +17,26 @@ export interface ConfigurationOptions {
   keyFailed?: string | null
   keyTtl?: number
   verbose?: boolean
+  redisMaxAttempts?: number
+  redisConnectTimeout?: number
+  redisCommandTimeout?: number
+  redisReconnectAttempts?: number
 }
 
 const DEFAULT_REDIS_URL = 'redis://localhost:6379'
 const DEFAULT_BATCH_SIZE = 5
 const DEFAULT_KEY_TTL = 604_800 // 1 week in seconds
 
+// Redis connection resilience. Redis is a best-effort coordination store for a
+// distributed test run, and CI runners can sit a WAN hop away from it (e.g. a
+// cross-datacenter mesh), so a transient blip must not red the build. Timeouts
+// are expressed in SECONDS in the env (matching the Ruby gem) and converted to
+// milliseconds for ioredis.
+const DEFAULT_REDIS_MAX_ATTEMPTS = 5
+const DEFAULT_REDIS_CONNECT_TIMEOUT_S = 3
+const DEFAULT_REDIS_COMMAND_TIMEOUT_S = 5
+const DEFAULT_REDIS_RECONNECT_ATTEMPTS = 3
+
 function envTruthy(name: string): boolean {
   const val = process.env[name]?.toLowerCase() ?? ''
   return ['1', 'true', 'yes'].includes(val)
@@ -33,6 +47,21 @@ function parseCommandOpts(opts: string | undefined | null): string[] {
   return opts.split(/\s+/)
 }
 
+function envInt(name: string, fallback: number): number {
+  const raw = process.env[name]
+  if (raw === undefined || raw.trim() === '') return fallback
+  const n = parseInt(raw, 10)
+  return Number.isNaN(n) ? fallback : n
+}
+
+function envFloat(name: string, fallback: number): number {
+  const raw = process.env[name]
+  if (raw === undefined || raw.trim() === '') return fallback
+  const n = parseFloat(raw)
+  return Number.isNaN(n) ? fallback : n
+}
+
+
 export class Configuration {
   redisUrl: string
   batchSize: number
@@ -43,6 +72,14 @@ export class Configuration {
   keyFailed: string | null
   keyTtl: number
   verbose: boolean
+  /** Application-level retry attempts on a Redis connection failure. */
+  redisMaxAttempts: number
+  /** ioredis connect timeout, in milliseconds. */
+  redisConnectTimeout: number
+  /** ioredis per-command timeout, in milliseconds. */
+  redisCommandTimeout: number
+  /** ioredis reconnect / per-request retry budget. */
+  redisReconnectAttempts: number
 
   constructor(options: ConfigurationOptions = {}) {
     this.redisUrl = options.redisUrl ?? process.env.SPECBANDIT_REDIS_URL ?? DEFAULT_REDIS_URL
@@ -54,6 +91,17 @@ export class Configuration {
     this.keyFailed = options.keyFailed ?? process.env.SPECBANDIT_KEY_FAILED ?? null
     this.keyTtl = options.keyTtl ?? parseInt(process.env.SPECBANDIT_KEY_TTL ?? String(DEFAULT_KEY_TTL), 10)
     this.verbose = options.verbose ?? envTruthy('SPECBANDIT_VERBOSE')
+
+    this.redisMaxAttempts =
+      options.redisMaxAttempts ?? envInt('SPECBANDIT_REDIS_MAX_ATTEMPTS', DEFAULT_REDIS_MAX_ATTEMPTS)
+    this.redisConnectTimeout =
+      options.redisConnectTimeout ??
+      Math.round(envFloat('SPECBANDIT_REDIS_CONNECT_TIMEOUT', DEFAULT_REDIS_CONNECT_TIMEOUT_S) * 1000)
+    this.redisCommandTimeout =
+      options.redisCommandTimeout ??
+      Math.round(envFloat('SPECBANDIT_REDIS_TIMEOUT', DEFAULT_REDIS_COMMAND_TIMEOUT_S) * 1000)
+    this.redisReconnectAttempts =
+      options.redisReconnectAttempts ?? envInt('SPECBANDIT_REDIS_RECONNECT_ATTEMPTS', DEFAULT_REDIS_RECONNECT_ATTEMPTS)
   }
 
   validate(): void {
@@ -66,6 +114,9 @@ export class Configuration {
     if (!Number.isInteger(this.keyTtl) || this.keyTtl <= 0) {
       throw new SpecbanditError('key_ttl must be a positive integer')
     }
+    if (!Number.isInteger(this.redisMaxAttempts) || this.redisMaxAttempts <= 0) {
+      throw new SpecbanditError('redis_max_attempts must be a positive integer')
+    }
   }
 
   validateForWork(): void {

src/redisQueue.ts

@@ -1,7 +1,25 @@
 import Redis from 'ioredis'
 
-const MAX_RETRIES = 3
+const DEFAULT_MAX_RETRIES = 5
 const BASE_DELAY_MS = 1000
+// Cap the exponential backoff so a real outage degrades/fails within a bounded
+// window instead of sleeping for minutes on the last attempts.
+const MAX_BACKOFF_MS = 10_000
+
+const DEFAULT_CONNECT_TIMEOUT_MS = 3000
+const DEFAULT_COMMAND_TIMEOUT_MS = 5000
+const DEFAULT_RECONNECT_ATTEMPTS = 3
+
+export interface RedisQueueOptions {
+  /** Application-level retry attempts on a connection failure. */
+  maxAttempts?: number
+  /** ioredis connect timeout, in milliseconds. */
+  connectTimeout?: number
+  /** ioredis per-command timeout, in milliseconds. */
+  commandTimeout?: number
+  /** ioredis reconnect / per-request retry budget. */
+  reconnectAttempts?: number
+}
 
 /** Companion marker key signalling that work was published for `key`. */
 function publishedMarkerKey(key: string): string {
@@ -10,11 +28,19 @@ function publishedMarkerKey(key: string): string {
 
 export class RedisQueue {
   readonly redis: Redis
+  private readonly maxAttempts: number
 
-  constructor(redisUrl: string = 'redis://localhost:6379') {
+  constructor(redisUrl: string = 'redis://localhost:6379', options: RedisQueueOptions = {}) {
+    this.maxAttempts = options.maxAttempts ?? DEFAULT_MAX_RETRIES
+    const reconnectAttempts = options.reconnectAttempts ?? DEFAULT_RECONNECT_ATTEMPTS
     this.redis = new Redis(redisUrl, {
       lazyConnect: true,
-      maxRetriesPerRequest: 3,
+      connectTimeout: options.connectTimeout ?? DEFAULT_CONNECT_TIMEOUT_MS,
+      commandTimeout: options.commandTimeout ?? DEFAULT_COMMAND_TIMEOUT_MS,
+      maxRetriesPerRequest: reconnectAttempts,
+      // Bounded reconnection backoff; give up (null) after the reconnect budget
+      // so a dead endpoint surfaces as an error the caller can degrade on.
+      retryStrategy: (times: number) => (times > reconnectAttempts ? null : Math.min(times * 200, 2000)),
     })
   }
 
@@ -96,13 +122,13 @@ export class RedisQueue {
   }
 
   private async withRetries<T>(operation: string, fn: () => Promise<T>): Promise<T> {
-    for (let attempt = 1; attempt <= MAX_RETRIES; attempt++) {
+    for (let attempt = 1; attempt <= this.maxAttempts; attempt++) {
       try {
         return await fn()
       } catch (error) {
-        if (attempt === MAX_RETRIES) throw error
-        const delay = BASE_DELAY_MS * Math.pow(2, attempt - 1)
-        console.warn(`[specbandit] Redis ${operation} failed (attempt ${attempt}/${MAX_RETRIES}), retrying in ${delay}ms: ${error}`)
+        if (attempt === this.maxAttempts) throw error
+        const delay = Math.min(BASE_DELAY_MS * Math.pow(2, attempt - 1), MAX_BACKOFF_MS)
+        console.warn(`[specbandit] Redis ${operation} failed (attempt ${attempt}/${this.maxAttempts}), retrying in ${delay}ms: ${error}`)
         await new Promise(resolve => setTimeout(resolve, delay))
       }
     }

test/configuration.test.ts

@@ -13,6 +13,10 @@ describe('Configuration', () => {
     'SPECBANDIT_KEY_FAILED',
     'SPECBANDIT_KEY_TTL',
     'SPECBANDIT_VERBOSE',
+    'SPECBANDIT_REDIS_MAX_ATTEMPTS',
+    'SPECBANDIT_REDIS_CONNECT_TIMEOUT',
+    'SPECBANDIT_REDIS_TIMEOUT',
+    'SPECBANDIT_REDIS_RECONNECT_ATTEMPTS',
   ]
   let savedEnv: Record<string, string | undefined>
 
@@ -190,6 +194,33 @@ describe('Configuration', () => {
     })
   })
 
+  describe('resilience settings', () => {
+    it('has sensible resilience defaults', () => {
+      const config = new Configuration()
+      expect(config.redisMaxAttempts).toBe(5)
+      expect(config.redisConnectTimeout).toBe(3000)
+      expect(config.redisCommandTimeout).toBe(5000)
+      expect(config.redisReconnectAttempts).toBe(3)
+    })
+
+    it('reads resilience settings from the environment (timeouts in seconds → ms)', () => {
+      process.env.SPECBANDIT_REDIS_MAX_ATTEMPTS = '8'
+      process.env.SPECBANDIT_REDIS_CONNECT_TIMEOUT = '2.5'
+      process.env.SPECBANDIT_REDIS_TIMEOUT = '7'
+      process.env.SPECBANDIT_REDIS_RECONNECT_ATTEMPTS = '1'
+      const config = new Configuration()
+      expect(config.redisMaxAttempts).toBe(8)
+      expect(config.redisConnectTimeout).toBe(2500)
+      expect(config.redisCommandTimeout).toBe(7000)
+      expect(config.redisReconnectAttempts).toBe(1)
+    })
+
+    it('throws when redis_max_attempts is not positive', () => {
+      const config = new Configuration({ key: 'valid-key', redisMaxAttempts: 0 })
+      expect(() => config.validate()).toThrow(/redis_max_attempts must be a positive integer/)
+    })
+  })
+
   describe('validateForWork()', () => {
     it('throws when command is null', () => {
       const config = new Configuration({ key: 'valid-key' })

test/redisQueue.test.ts

@@ -184,7 +184,7 @@ describe('RedisQueue', () => {
       expect(result).toBe(5)
       expect(mockRedis.llen).toHaveBeenCalledTimes(2)
       expect(warnSpy).toHaveBeenCalledTimes(1)
-      expect(warnSpy.mock.calls[0][0]).toMatch(/Redis length failed \(attempt 1\/3\)/)
+      expect(warnSpy.mock.calls[0][0]).toMatch(/Redis length failed \(attempt 1\/5\)/)
 
       warnSpy.mockRestore()
     })
@@ -209,24 +209,40 @@ describe('RedisQueue', () => {
       warnSpy.mockRestore()
     })
 
-    it('throws after exhausting all retries', async () => {
+    it('throws after exhausting all retries (default 5 attempts)', async () => {
       const error = new Error('persistent failure')
-      mockRedis.llen
-        .mockRejectedValueOnce(error)
-        .mockRejectedValueOnce(error)
-        .mockRejectedValueOnce(error)
+      mockRedis.llen.mockRejectedValue(error)
 
       const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
 
       const promise = queue.length('my-key').catch((e: Error) => e)
+      // Capped exponential backoff between the 5 attempts: 1s, 2s, 4s, 8s.
       await vi.advanceTimersByTimeAsync(1000)
       await vi.advanceTimersByTimeAsync(2000)
+      await vi.advanceTimersByTimeAsync(4000)
+      await vi.advanceTimersByTimeAsync(8000)
 
       const result = await promise
       expect(result).toBeInstanceOf(Error)
       expect((result as Error).message).toBe('persistent failure')
-      expect(mockRedis.llen).toHaveBeenCalledTimes(3)
-      expect(warnSpy).toHaveBeenCalledTimes(2)
+      expect(mockRedis.llen).toHaveBeenCalledTimes(5)
+      expect(warnSpy).toHaveBeenCalledTimes(4)
+
+      warnSpy.mockRestore()
+    })
+
+    it('honours a custom maxAttempts', async () => {
+      const q = new RedisQueue('redis://localhost:6379', { maxAttempts: 2 })
+      const local = { llen: vi.fn().mockRejectedValue(new Error('down')) }
+      ;(q as any).redis = local
+      const warnSpy = vi.spyOn(console, 'warn').mockImplementation(() => {})
+
+      const promise = q.length('my-key').catch((e: Error) => e)
+      await vi.advanceTimersByTimeAsync(1000)
+
+      const result = await promise
+      expect(result).toBeInstanceOf(Error)
+      expect(local.llen).toHaveBeenCalledTimes(2)
 
       warnSpy.mockRestore()
     })