diff --git a/services/tts/package-lock.json b/services/tts/package-lock.json index d48bbc31..96586457 100644 --- a/services/tts/package-lock.json +++ b/services/tts/package-lock.json @@ -18,6 +18,7 @@ "@opentelemetry/semantic-conventions": "^1.43.0", "express": "^5.2.1", "express-rate-limit": "^8.6.1", + "ioredis": "^5.11.1", "opossum": "^10.0.0" }, "devDependencies": { @@ -651,6 +652,12 @@ "node": ">=6" } }, + "node_modules/@ioredis/commands": { + "version": "1.10.0", + "resolved": "https://registry.npmjs.org/@ioredis/commands/-/commands-1.10.0.tgz", + "integrity": "sha512-UmeW7z4LfctwoQ5wkhVzgq8tXkreED2xZGpX+Bg+zA+WJFZCT6c062AfCK/Dfk81xZnnwdhJCUMkitihRaoC2Q==", + "license": "MIT" + }, "node_modules/@isaacs/cliui": { "version": "8.0.2", "resolved": "https://registry.npmjs.org/@isaacs/cliui/-/cliui-8.0.2.tgz", @@ -3386,6 +3393,15 @@ "url": "https://github.com/chalk/wrap-ansi?sponsor=1" } }, + "node_modules/cluster-key-slot": { + "version": "1.1.1", + "resolved": "https://registry.npmjs.org/cluster-key-slot/-/cluster-key-slot-1.1.1.tgz", + "integrity": "sha512-rwHwUfXL40Chm1r08yrhU3qpUvdVlgkKNeyeGPOxnW8/SyVDvgRaed/Uz54AqWNaTCAThlj6QAs3TZcKI0xDEw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10.0" + } + }, "node_modules/co": { "version": "4.6.0", "resolved": "https://registry.npmjs.org/co/-/co-4.6.0.tgz", @@ -3548,6 +3564,15 @@ "node": ">=0.10.0" } }, + "node_modules/denque": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/denque/-/denque-2.1.0.tgz", + "integrity": "sha512-HVQE3AAb/pxF8fQAoiqpvg9i3evqug3hoiwakOyZAwJm+6vZehbkYXZ0l4JxS+I3QxM97v5aaRNhj8v5oBhekw==", + "license": "Apache-2.0", + "engines": { + "node": ">=0.10" + } + }, "node_modules/depd": { "version": "2.0.0", "resolved": "https://registry.npmjs.org/depd/-/depd-2.0.0.tgz", @@ -4434,6 +4459,28 @@ "integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ==", "license": "ISC" }, + "node_modules/ioredis": { + "version": "5.11.1", + "resolved": "https://registry.npmjs.org/ioredis/-/ioredis-5.11.1.tgz", + "integrity": "sha512-ehuGcf94bQXhfagULNXrJdfnWO38v070jxSx/qE87Kjzmu2fU7ro5EFAb+OPituLqgfyuQaym5DlrNydW2sJ9A==", + "license": "MIT", + "dependencies": { + "@ioredis/commands": "1.10.0", + "cluster-key-slot": "1.1.1", + "debug": "4.4.3", + "denque": "2.1.0", + "redis-errors": "1.2.0", + "redis-parser": "3.0.0", + "standard-as-callback": "2.1.0" + }, + "engines": { + "node": ">=12.22.0" + }, + "funding": { + "type": "opencollective", + "url": "https://opencollective.com/ioredis" + } + }, "node_modules/ip-address": { "version": "10.2.0", "resolved": "https://registry.npmjs.org/ip-address/-/ip-address-10.2.0.tgz", @@ -6001,6 +6048,27 @@ "node": ">= 6" } }, + "node_modules/redis-errors": { + "version": "1.2.0", + "resolved": "https://registry.npmjs.org/redis-errors/-/redis-errors-1.2.0.tgz", + "integrity": "sha512-1qny3OExCf0UvUV/5wpYKf2YwPcOqXzkwKKSmKHiE6ZMQs5heeE/c8eXK+PNllPvmjgAbfnsbpkGZWy8cBpn9w==", + "license": "MIT", + "engines": { + "node": ">=4" + } + }, + "node_modules/redis-parser": { + "version": "3.0.0", + "resolved": "https://registry.npmjs.org/redis-parser/-/redis-parser-3.0.0.tgz", + "integrity": "sha512-DJnGAeenTdpMEH6uAJRK/uiyEIH9WVsUmoLwzudwGJUwZPp80PDBWPHXSAGNPwNvIXAbe7MSUB1zQFugFml66A==", + "license": "MIT", + "dependencies": { + "redis-errors": "^1.0.0" + }, + "engines": { + "node": ">=4" + } + }, "node_modules/require-directory": { "version": "2.1.1", "resolved": "https://registry.npmjs.org/require-directory/-/require-directory-2.1.1.tgz", @@ -6336,6 +6404,12 @@ "node": ">=10" } }, + "node_modules/standard-as-callback": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/standard-as-callback/-/standard-as-callback-2.1.0.tgz", + "integrity": "sha512-qoRRSyROncaz1z0mvYqIE4lCd9p2R90i6GxW3uZv5ucSu8tU7B5HXUP1gG8pVZsYNVaXjk8ClXHPttLyxAL48A==", + "license": "MIT" + }, "node_modules/statuses": { "version": "2.0.2", "resolved": "https://registry.npmjs.org/statuses/-/statuses-2.0.2.tgz", diff --git a/services/tts/package.json b/services/tts/package.json index d831bbbe..b308b5a2 100644 --- a/services/tts/package.json +++ b/services/tts/package.json @@ -29,6 +29,7 @@ "@opentelemetry/semantic-conventions": "^1.43.0", "express": "^5.2.1", "express-rate-limit": "^8.6.1", + "ioredis": "^5.11.1", "opossum": "^10.0.0" }, "devDependencies": { diff --git a/services/tts/src/SharedStore.ts b/services/tts/src/SharedStore.ts new file mode 100644 index 00000000..6d016abf --- /dev/null +++ b/services/tts/src/SharedStore.ts @@ -0,0 +1,123 @@ +/** + * Shared, cross-replica state backing for job status, rate limiting, and the + * audio cache (issue #1133). + * + * jobStore, RateLimiter, and AudioCache in TTSService.ts are process-local + * Maps. That's fine for a single instance, but breaks under horizontal + * scaling: GET /tts/job/:id 404s when the polling request lands on a + * different pod than the one that processed the job, rate limits are only + * enforced per-replica (multiplying the effective global limit by replica + * count), and cache hit rate collapses across replicas. + * + * These interfaces let TTSService optionally delegate to a shared backend + * (e.g. Redis) instead of process-local memory. Omit `sharedStore` from + * TTSConfig to keep the existing in-memory, single-instance behavior. + */ + +import type { Redis as RedisClient } from "ioredis"; +import type { TTSJob } from "./TTSService"; + +export interface SharedJobStore { + getJob(id: string): Promise; + setJob(job: TTSJob): Promise; + listJobs(status?: TTSJob["status"]): Promise; +} + +export interface SharedRateLimitStore { + /** + * Atomically increment the counter for `key` within `windowMs` and return + * the new count. The first increment in a window sets its expiry. + */ + incr(key: string, windowMs: number): Promise; +} + +export interface SharedCacheStore { + get(key: string): Promise; + set(key: string, buffer: Buffer, ttlMs: number): Promise; +} + +export interface SharedStoreConfig { + jobStore?: SharedJobStore; + rateLimitStore?: SharedRateLimitStore; + cacheStore?: SharedCacheStore; +} + +function serializeJob(job: TTSJob): string { + return JSON.stringify(job); +} + +function deserializeJob(raw: string): TTSJob { + const parsed = JSON.parse(raw); + return { ...parsed, createdAt: new Date(parsed.createdAt), updatedAt: new Date(parsed.updatedAt) }; +} + +/** Redis-backed job store. Jobs expire after `ttlMs` (default 24h) so completed jobs don't accumulate forever. */ +export class RedisJobStore implements SharedJobStore { + constructor( + private redis: RedisClient, + private prefix = "tts:job:", + private ttlMs = 86_400_000 + ) {} + + async getJob(id: string): Promise { + const raw = await this.redis.get(this.prefix + id); + return raw ? deserializeJob(raw) : undefined; + } + + async setJob(job: TTSJob): Promise { + await this.redis.set(this.prefix + job.id, serializeJob(job), "PX", this.ttlMs); + } + + async listJobs(status?: TTSJob["status"]): Promise { + const keys = await this.redis.keys(`${this.prefix}*`); + if (keys.length === 0) return []; + const raws = await this.redis.mget(...keys); + const jobs = raws.filter((r): r is string => !!r).map(deserializeJob); + return status ? jobs.filter((j) => j.status === status) : jobs; + } +} + +/** Redis-backed rate limit counter using atomic INCR + PEXPIRE-on-first-hit. */ +export class RedisRateLimitStore implements SharedRateLimitStore { + constructor(private redis: RedisClient, private prefix = "tts:rl:") {} + + async incr(key: string, windowMs: number): Promise { + const redisKey = this.prefix + key; + const count = await this.redis.incr(redisKey); + if (count === 1) { + await this.redis.pexpire(redisKey, windowMs); + } + return count; + } +} + +/** Redis-backed audio cache, storing raw buffers with a per-entry TTL. */ +export class RedisCacheStore implements SharedCacheStore { + constructor(private redis: RedisClient, private prefix = "tts:cache:") {} + + async get(key: string): Promise { + const raw = await this.redis.getBuffer(this.prefix + key); + return raw ?? undefined; + } + + async set(key: string, buffer: Buffer, ttlMs: number): Promise { + await this.redis.set(this.prefix + key, buffer, "PX", ttlMs); + } +} + +/** Lazily constructs an ioredis client. Kept out of the module-level import graph so ioredis stays optional at runtime for single-instance deployments. */ +export function createRedisClient(url: string): RedisClient { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const IORedis = require("ioredis"); + return new IORedis(url); +} + +/** Builds a full SharedStoreConfig backed by a single Redis connection. */ +export function createRedisSharedStore(url: string): SharedStoreConfig { + const redis = createRedisClient(url); + return { + jobStore: new RedisJobStore(redis), + rateLimitStore: new RedisRateLimitStore(redis), + cacheStore: new RedisCacheStore(redis), + }; +} diff --git a/services/tts/src/TTSService.ts b/services/tts/src/TTSService.ts index 34f9c9ca..aeaf3479 100644 --- a/services/tts/src/TTSService.ts +++ b/services/tts/src/TTSService.ts @@ -19,6 +19,7 @@ import path from "path"; import { createHash } from "crypto"; import { trace, SpanStatusCode, Span } from "@opentelemetry/api"; import CircuitBreaker from "opossum"; +import type { SharedStoreConfig } from "./SharedStore"; // --------------------------------------------------------------------------- // Types @@ -388,6 +389,12 @@ export interface TTSConfig { circuitBreaker?: CircuitBreakerConfig; /** Retry config for transient provider errors — omit to use defaults */ retry?: RetryConfig; + /** + * Shared, cross-replica backing (e.g. Redis) for job status, rate + * limiting, and cache (issue #1133). Omit to keep process-local in-memory + * state, which only gives correct results for a single instance. + */ + sharedStore?: SharedStoreConfig; /** * TTL in milliseconds for completed/errored jobs before they're evicted * from the in-memory job store. Omit to use the default (1 hour). @@ -769,6 +776,14 @@ export class TTSService { // Per-call timeout (counted as a failure). Bounds the whole retried // request, since the action below includes the retry/backoff loop. timeout: cbCfg.timeoutMs, + // Issue #1135: 4xx TTSProviderErrors (bad voice, invalid credential, ...) + // are client/config mistakes, not upstream provider health signals. + // Excluding them from breaker accounting stops a handful of bad + // requests from tripping the breaker for every other user of the + // provider. Genuine 5xx/network failures (no statusCode, or >= 500) + // still count as failures. + errorFilter: (err: unknown) => + err instanceof TTSProviderError && err.statusCode >= 400 && err.statusCode < 500, }; if (this.config.elevenlabs) { @@ -833,6 +848,11 @@ export class TTSService { /** * Enqueue a TTS job and return its ID immediately. + * + * Rate limiting and job visibility are process-local here — use + * `enqueueAsync` when `config.sharedStore` is configured so multiple + * replicas behind a load balancer see consistent state (issue #1133). + * * @param credential API key or JWT Bearer token (required when auth is configured). * @param rateLimitKey IP address or user ID for rate limiting (e.g. "ip:1.2.3.4" or "user:abc"). * @param bypassCache If true, skip cache lookup and always generate fresh audio. @@ -852,34 +872,49 @@ export class TTSService { this.rateLimiter.check(rateLimitKey); } - // Input sanitization - const sanitized = sanitizeInput(text); + const job = this._createJob(text, voice, provider, bypassCache, owner); + jobStore.set(job.id, job); + this._process(job).catch((err) => this._handleProcessError(job.id, err)); + return job.id; + } - const id = makeId(); - const job: TTSJob = { - id, - text: sanitized, - voice, - provider: provider ?? this.config.provider, - status: "pending", - createdAt: new Date(), - updatedAt: new Date(), - bypassCache: bypassCache || false, - owner, - }; - jobStore.set(id, job); - - this._process(job).catch((err) => { - const j = jobStore.get(id); - if (j) { - j.status = "error"; - j.error = err instanceof Error ? err.message : String(err); - j.updatedAt = new Date(); - console.error(`[TTSService] Job ${id} failed: ${j.error}`); + /** + * Async equivalent of `enqueue`. When `config.sharedStore` is configured, + * rate limiting is enforced against the shared counter (consistent across + * replicas) and the job is written to the shared job store immediately so + * `getJobAsync` on another replica can see it right away, before + * processing even completes (issue #1133). + */ + async enqueueAsync( + text: string, + voice: TTSVoice, + provider?: TTSProvider, + credential?: string, + rateLimitKey?: string, + bypassCache?: boolean + ): Promise { + const owner = this._resolveOwner(credential); + + if (rateLimitKey) { + const sharedRateLimit = this.config.sharedStore?.rateLimitStore; + if (sharedRateLimit && this.config.rateLimit) { + const count = await sharedRateLimit.incr(rateLimitKey, this.config.rateLimit.windowMs); + if (count > this.config.rateLimit.maxRequests) { + throw new RateLimitError( + `Rate limit exceeded for key "${rateLimitKey}": ${count}/${this.config.rateLimit.maxRequests} in ${this.config.rateLimit.windowMs}ms` + ); + } + } else if (this.rateLimiter) { + this.rateLimiter.check(rateLimitKey); } - }); + } - return id; + const job = this._createJob(text, voice, provider, bypassCache, owner); + jobStore.set(job.id, job); + await this._persistJob(job); + + this._process(job).catch((err) => this._handleProcessError(job.id, err)); + return job.id; } /** @@ -894,6 +929,8 @@ export class TTSService { /** * Look up a job by ID, scoped to the requesting credential's tenant. + * Process-local lookup only — use `getJobAsync` under `config.sharedStore` + * so polling succeeds regardless of which replica handled the job (issue #1133). * Returns `undefined` both when the job doesn't exist and when it exists * but belongs to a different tenant — the two cases are indistinguishable * to the caller so a credential cannot probe for other tenants' job IDs. @@ -905,6 +942,22 @@ export class TTSService { return job; } + /** + * Looks up a job locally first (fast path for the replica that processed + * it), falling back to the shared store so polling succeeds regardless of + * which replica originally handled the job (issue #1133). Scoped to the + * requesting credential's tenant like `getJob`, on both the local and + * shared-store paths. + */ + async getJobAsync(id: string, credential?: string): Promise { + const owner = this._resolveOwner(credential); + const local = jobStore.get(id); + if (local) return local.owner === owner ? local : undefined; + const remote = await this.config.sharedStore?.jobStore?.getJob(id); + if (!remote || remote.owner !== owner) return undefined; + return remote; + } + /** List jobs belonging to the requesting credential's tenant, optionally filtered by status. */ listJobs(status?: TTSJob["status"], credential?: string): TTSJob[] { const owner = this._resolveOwner(credential); @@ -912,6 +965,20 @@ export class TTSService { return status ? all.filter((j) => j.status === status) : all; } + /** + * Async equivalent of `listJobs`, preferring the shared store when + * configured so listings are consistent across replicas (issue #1133). + * Scoped to the requesting credential's tenant like `listJobs`. + */ + async listJobsAsync(status?: TTSJob["status"], credential?: string): Promise { + const owner = this._resolveOwner(credential); + if (this.config.sharedStore?.jobStore) { + const all = await this.config.sharedStore.jobStore.listJobs(status); + return all.filter((j) => j.owner === owner); + } + return this.listJobs(status, credential); + } + /** * Returns jobs across *all* tenants, optionally filtered by status. * For internal operational use only (health checks, queue-depth metrics) @@ -925,6 +992,8 @@ export class TTSService { /** * Synchronous generation — awaits completion and returns the output path. + * Uses `enqueueAsync` internally so rate limiting is consistent across + * replicas when `config.sharedStore` is configured. * @param rateLimitKey IP address or user ID for rate limiting. * @param bypassCache If true, skip cache lookup and always generate fresh audio. */ @@ -936,7 +1005,7 @@ export class TTSService { rateLimitKey?: string, bypassCache?: boolean ): Promise { - const id = this.enqueue(text, voice, provider, credential, rateLimitKey, bypassCache); + const id = await this.enqueueAsync(text, voice, provider, credential, rateLimitKey, bypassCache); return this._waitForJob(id); } @@ -965,36 +1034,97 @@ export class TTSService { // Private // --------------------------------------------------------------------------- + /** Builds a pending TTSJob. Shared by `enqueue` and `enqueueAsync`. */ + private _createJob( + text: string, + voice: TTSVoice, + provider: TTSProvider | undefined, + bypassCache: boolean | undefined, + owner: string + ): TTSJob { + const sanitized = sanitizeInput(text); + return { + id: makeId(), + text: sanitized, + voice, + provider: provider ?? this.config.provider, + status: "pending", + createdAt: new Date(), + updatedAt: new Date(), + bypassCache: bypassCache || false, + owner, + }; + } + + /** Writes a job to the shared store, if one is configured (issue #1133). No-op otherwise. */ + private async _persistJob(job: TTSJob): Promise { + if (this.config.sharedStore?.jobStore) { + await this.config.sharedStore.jobStore.setJob(job); + } + } + + private _handleProcessError(id: string, err: unknown): void { + const job = jobStore.get(id); + if (!job) return; + job.status = "error"; + job.error = err instanceof Error ? err.message : String(err); + job.updatedAt = new Date(); + console.error(`[TTSService] Job ${id} failed: ${job.error}`); + this._persistJob(job).catch((persistErr) => + console.error(`[TTSService] Failed to persist job ${id} error state: ${persistErr}`) + ); + } + + /** Prefers the shared cache store when configured so cache hit rate isn't split across replicas (issue #1133). */ + private async _cacheGet(key: string): Promise { + const shared = this.config.sharedStore?.cacheStore; + if (shared) return shared.get(key); + return this.cache?.get(key); + } + + private async _cacheSet(key: string, buffer: Buffer): Promise { + const shared = this.config.sharedStore?.cacheStore; + if (shared) { + await shared.set(key, buffer, this.config.cache?.ttlMs ?? 86_400_000); + return; + } + this.cache?.set(key, buffer); + } + private async _process(job: TTSJob): Promise { job.status = "processing"; job.updatedAt = new Date(); + await this._persistJob(job); - const cacheKey = this.cache && !job.bypassCache + const cachingEnabled = (this.cache || this.config.sharedStore?.cacheStore) && !job.bypassCache; + const cacheKey = cachingEnabled ? AudioCache.key(job.text, job.voice.voiceId, job.provider) : null; // Cache hit — write cached buffer to disk and skip provider call - if (cacheKey && this.cache) { - const cached = this.cache.get(cacheKey); + if (cacheKey) { + const cached = await this._cacheGet(cacheKey); if (cached) { const outputPath = await saveAudio(cached, this.config.outputDir, job.id); job.outputPath = outputPath; job.status = "done"; job.updatedAt = new Date(); + await this._persistJob(job); return; } } const buffer = await this._generateWithFallback(job); - if (cacheKey && this.cache) { - this.cache.set(cacheKey, buffer); + if (cacheKey) { + await this._cacheSet(cacheKey, buffer); } const outputPath = await saveAudio(buffer, this.config.outputDir, job.id); job.outputPath = outputPath; job.status = "done"; job.updatedAt = new Date(); + await this._persistJob(job); } /** diff --git a/services/tts/src/__tests__/circuit-breaker.test.ts b/services/tts/src/__tests__/circuit-breaker.test.ts index bb9aa946..6e44d5c6 100644 --- a/services/tts/src/__tests__/circuit-breaker.test.ts +++ b/services/tts/src/__tests__/circuit-breaker.test.ts @@ -150,6 +150,38 @@ describe("TTSService — circuit breaker", () => { expect(["open", "halfOpen"]).toContain(stateAfterProbe); }, 15_000); + it("does not open the breaker for repeated 4xx client errors (issue #1135)", async () => { + // A 400 response (e.g. invalid voice parameter) should be a non-retriable + // TTSProviderError, but must NOT count toward tripping the breaker — + // it's a client mistake, not upstream provider health degradation. + (global as any).fetch = jest.fn().mockResolvedValue({ + ok: false, + status: 400, + statusText: "Bad Request", + text: async () => "invalid voice parameter", + }); + + const svc = new TTSService({ + provider: "elevenlabs", + elevenlabs: { apiKey: "test-key" }, + outputDir: "/tmp/tts-cb-4xx-test", + circuitBreaker: { + openThreshold: 2, + rollingWindowMs: 30_000, + halfOpenIntervalMs: 60_000, + timeoutMs: 500, + }, + }); + + // Fire well more than the failure threshold worth of 4xx errors. + for (let i = 0; i < 6; i++) { + await expect(svc.generate("hello", VOICE, "elevenlabs")).rejects.toThrow(); + } + + const states = svc.getCircuitBreakerStates(); + expect(states["elevenlabs"].state).toBe("closed"); + }, 15_000); + it("getCircuitBreakerStates returns empty object when no providers are configured", () => { // No elevenlabs or google config → no breakers initialised const svc = new TTSService({ diff --git a/services/tts/src/__tests__/rate-limit-key.test.ts b/services/tts/src/__tests__/rate-limit-key.test.ts new file mode 100644 index 00000000..fe86282f --- /dev/null +++ b/services/tts/src/__tests__/rate-limit-key.test.ts @@ -0,0 +1,87 @@ +/** + * Test for issue #1132: rate limiter bypass via a rotating Authorization header. + * + * The bucket key must be derived from the caller's IP address only — never + * from the (unvalidated, optional) Authorization header — otherwise a caller + * can rotate a fake credential on every request to get a fresh rate-limit + * bucket each time. + */ + +import { Request } from "express"; +import rateLimit from "express-rate-limit"; +import { rateLimitKeyGenerator } from "../rateLimitKey"; + +function mockRequest(ip: string, authHeader?: string): Request { + return { + ip, + headers: authHeader ? { authorization: authHeader } : {}, + } as unknown as Request; +} + +describe("rateLimitKeyGenerator", () => { + it("keys on IP regardless of Authorization header presence", () => { + const withAuth = mockRequest("1.2.3.4", "Bearer fake-token"); + const withoutAuth = mockRequest("1.2.3.4"); + expect(rateLimitKeyGenerator(withAuth)).toBe(rateLimitKeyGenerator(withoutAuth)); + }); + + it("returns the same key across a rotating fake Authorization header", () => { + const keys = new Set(); + for (let i = 0; i < 20; i++) { + keys.add(rateLimitKeyGenerator(mockRequest("9.8.7.6", `Bearer fake-token-${i}`))); + } + // Every request should collapse to the same bucket key. + expect(keys.size).toBe(1); + expect([...keys][0]).toBe("ip:9.8.7.6"); + }); + + it("isolates buckets by IP", () => { + expect(rateLimitKeyGenerator(mockRequest("1.1.1.1"))).not.toBe( + rateLimitKeyGenerator(mockRequest("2.2.2.2")) + ); + }); +}); + +describe("Express rate limiter — rotating Authorization header", () => { + it("still enforces the limit when the caller rotates a fake Authorization header per request", async () => { + const limiter = rateLimit({ + windowMs: 60_000, + max: 3, + keyGenerator: rateLimitKeyGenerator, + standardHeaders: false, + legacyHeaders: false, + }); + + const ip = "5.5.5.5"; + let blocked = 0; + let allowed = 0; + + for (let i = 0; i < 6; i++) { + const req = mockRequest(ip, `Bearer rotating-fake-${i}`); + const res: any = { + setHeader: jest.fn(), + getHeader: jest.fn(), + removeHeader: jest.fn(), + status: jest.fn().mockReturnThis(), + json: jest.fn(), + send: jest.fn(), + end: jest.fn(), + headersSent: false, + }; + let nextCalled = false; + await new Promise((resolve) => { + limiter(req, res, () => { + nextCalled = true; + resolve(); + }); + // express-rate-limit's handler path doesn't call next(); resolve for that case too. + if (res.status.mock.calls.length > 0) resolve(); + }); + if (nextCalled) allowed++; + else blocked++; + } + + expect(allowed).toBe(3); + expect(blocked).toBe(3); + }); +}); diff --git a/services/tts/src/__tests__/shared-store-multi-instance.test.ts b/services/tts/src/__tests__/shared-store-multi-instance.test.ts new file mode 100644 index 00000000..36066f76 --- /dev/null +++ b/services/tts/src/__tests__/shared-store-multi-instance.test.ts @@ -0,0 +1,135 @@ +/** + * Tests for issue #1133: in-memory job/rate-limit/cache state breaks + * correctness under horizontal scaling. + * + * jobStore, RateLimiter, and AudioCache are process-local Maps with no + * shared backing store. Simulates two replicas (two independent TTSService + * instances) pointed at the same shared store and asserts: + * - a job enqueued on one replica is visible on the other (job polling + * doesn't 404 just because a different pod processed it) + * - rate limits are enforced globally across replicas, not per-replica + * - the audio cache is shared across replicas + * + * Uses a small in-memory fake that implements the SharedStore interfaces + * rather than a real Redis connection, so this runs without any external + * infrastructure. RedisJobStore/RedisRateLimitStore/RedisCacheStore in + * SharedStore.ts implement the same interfaces against a real Redis + * connection for production multi-replica deployments (set REDIS_URL). + */ + +import { TTSService, VOICES, RateLimitError, TTSJob } from "../TTSService"; +import type { SharedJobStore, SharedRateLimitStore, SharedCacheStore, SharedStoreConfig } from "../SharedStore"; + +const VOICE = VOICES["el-rachel-en"]; + +class FakeSharedJobStore implements SharedJobStore { + private store = new Map(); + async getJob(id: string) { + return this.store.get(id); + } + async setJob(job: TTSJob) { + this.store.set(job.id, { ...job }); + } + async listJobs(status?: TTSJob["status"]) { + const all = Array.from(this.store.values()); + return status ? all.filter((j) => j.status === status) : all; + } +} + +class FakeSharedRateLimitStore implements SharedRateLimitStore { + private counts = new Map(); + async incr(key: string) { + const next = (this.counts.get(key) ?? 0) + 1; + this.counts.set(key, next); + return next; + } +} + +class FakeSharedCacheStore implements SharedCacheStore { + private store = new Map(); + async get(key: string) { + return this.store.get(key); + } + async set(key: string, buffer: Buffer) { + this.store.set(key, buffer); + } +} + +function makeSharedStore(): SharedStoreConfig { + return { + jobStore: new FakeSharedJobStore(), + rateLimitStore: new FakeSharedRateLimitStore(), + cacheStore: new FakeSharedCacheStore(), + }; +} + +describe("Shared store — multi-instance simulation (issue #1133)", () => { + it("job enqueued on one replica is visible on another replica", async () => { + const sharedStore = makeSharedStore(); + + const replicaA = new TTSService({ + provider: "elevenlabs", + elevenlabs: { apiKey: "k" }, + outputDir: "/tmp/tts-shared-a", + sharedStore, + }); + const replicaB = new TTSService({ + provider: "elevenlabs", + elevenlabs: { apiKey: "k" }, + outputDir: "/tmp/tts-shared-b", + sharedStore, + }); + + const jobId = await replicaA.enqueueAsync("hello", VOICE); + + // replicaB never processed this job locally — without a shared job + // store this would 404. With it, the job (at least in "pending" or + // later status) is visible immediately. + const seenByB = await replicaB.getJobAsync(jobId); + expect(seenByB).toBeDefined(); + expect(seenByB!.id).toBe(jobId); + }); + + it("rate limits are enforced globally across replicas, not per-replica", async () => { + const sharedStore = makeSharedStore(); + const rateLimit = { maxRequests: 2, windowMs: 60_000 }; + + const replicaA = new TTSService({ + provider: "elevenlabs", + elevenlabs: { apiKey: "k" }, + outputDir: "/tmp/tts-shared-rl-a", + rateLimit, + sharedStore, + }); + const replicaB = new TTSService({ + provider: "elevenlabs", + elevenlabs: { apiKey: "k" }, + outputDir: "/tmp/tts-shared-rl-b", + rateLimit, + sharedStore, + }); + + const key = "ip:1.2.3.4"; + + // Without a shared store, each replica would allow 2 requests of its + // own (4 total) — the bug this issue describes. With the shared + // counter, the limit of 2 applies across both replicas combined. + await replicaA.enqueueAsync("hello", VOICE, undefined, undefined, key); + await replicaB.enqueueAsync("hello", VOICE, undefined, undefined, key); + + await expect(replicaA.enqueueAsync("hello", VOICE, undefined, undefined, key)).rejects.toThrow( + RateLimitError + ); + }); + + it("audio cache is shared across replicas", async () => { + const sharedStore = makeSharedStore(); + const cacheStore = sharedStore.cacheStore!; + + const buf = Buffer.from("cached-audio"); + await cacheStore.set("shared-key", buf, 60_000); + + const readBack = await cacheStore.get("shared-key"); + expect(readBack).toEqual(buf); + }); +}); diff --git a/services/tts/src/__tests__/tracing.test.ts b/services/tts/src/__tests__/tracing.test.ts new file mode 100644 index 00000000..d764daed --- /dev/null +++ b/services/tts/src/__tests__/tracing.test.ts @@ -0,0 +1,69 @@ +/** + * Tests for issue #1134: OpenTelemetry tracing SDK is never initialized. + * + * initTracing() previously existed but was never invoked anywhere, so every + * tracer.startActiveSpan(...) call across TTSService/HealthCheck ran against + * the default no-op tracer provider — spans were created and discarded, + * never exported. + */ + +const startMock = jest.fn(); +const shutdownMock = jest.fn().mockResolvedValue(undefined); + +jest.mock("@opentelemetry/sdk-node", () => ({ + NodeSDK: jest.fn().mockImplementation(() => ({ + start: startMock, + shutdown: shutdownMock, + })), +})); + +jest.mock("@opentelemetry/exporter-trace-otlp-grpc", () => ({ + OTLPTraceExporter: jest.fn().mockImplementation(() => ({})), +})); + +jest.mock("@opentelemetry/instrumentation-http", () => ({ + HttpInstrumentation: jest.fn().mockImplementation(() => ({})), +})); + +describe("initTracing", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + it("starts the NodeSDK", () => { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { initTracing } = require("../tracing"); + initTracing(); + expect(startMock).toHaveBeenCalledTimes(1); + }); + + it("configures the OTLP exporter and HTTP instrumentation", () => { + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { NodeSDK } = require("@opentelemetry/sdk-node"); + // eslint-disable-next-line @typescript-eslint/no-var-requires + const { initTracing } = require("../tracing"); + initTracing(); + expect(NodeSDK).toHaveBeenCalledTimes(1); + const sdkArgs = (NodeSDK as jest.Mock).mock.calls[0][0]; + expect(sdkArgs.traceExporter).toBeDefined(); + expect(sdkArgs.instrumentations).toHaveLength(1); + }); +}); + +describe("server startup — tracing wiring (issue #1134)", () => { + beforeEach(() => { + jest.resetModules(); + process.env.TTS_OUTPUT_DIR = "/tmp/tts-tracing-test"; + }); + + it("calls initTracing() exactly once at module load, before request handling", () => { + const initTracingMock = jest.fn(); + jest.doMock("../tracing", () => ({ initTracing: initTracingMock })); + + // require.main !== module in Jest, so server.ts's app.listen() is skipped — + // importing it here only exercises the module's top-level wiring. + require("../server"); + + expect(initTracingMock).toHaveBeenCalledTimes(1); + }); +}); diff --git a/services/tts/src/rateLimitKey.ts b/services/tts/src/rateLimitKey.ts new file mode 100644 index 00000000..5d3b898d --- /dev/null +++ b/services/tts/src/rateLimitKey.ts @@ -0,0 +1,19 @@ +/** + * Rate-limit bucket key derivation for the Express-level limiter (issue #995). + * + * Split into its own module so it can be unit-tested without booting the + * full server (issue #1132). + */ + +import { Request } from "express"; + +/** + * Key the limiter on the caller's IP address only. The Authorization header + * MUST NOT be used here: it is unvalidated at this point in the middleware + * chain (auth runs after rate limiting, and is optional entirely), so a + * caller could otherwise send a different fake credential on every request + * to get a fresh bucket each time, bypassing the throttle. + */ +export function rateLimitKeyGenerator(req: Request): string { + return `ip:${req.ip ?? "unknown"}`; +} diff --git a/services/tts/src/server.ts b/services/tts/src/server.ts index b0aa7785..a0ab6925 100644 --- a/services/tts/src/server.ts +++ b/services/tts/src/server.ts @@ -27,6 +27,19 @@ import { } from "./HealthCheck"; import { W3CTraceContextPropagator } from "@opentelemetry/core"; import { trace, context } from "@opentelemetry/api"; +import { rateLimitKeyGenerator } from "./rateLimitKey"; +import { initTracing } from "./tracing"; +import { createRedisSharedStore } from "./SharedStore"; + +// --------------------------------------------------------------------------- +// Tracing +// --------------------------------------------------------------------------- + +// Issue #1134: initTracing() was defined but never called anywhere, so every +// tracer.startActiveSpan(...) call ran against the default no-op global +// tracer provider. Must run once, before any request handling, so the +// HttpInstrumentation and OTLP exporter are wired up in time. +initTracing(); // --------------------------------------------------------------------------- // Configuration @@ -64,6 +77,13 @@ const config: TTSConfig = { maxRetries: parseInt(process.env.TTS_MAX_RETRIES || "3", 10), maxDelayMs: parseInt(process.env.TTS_MAX_DELAY_MS || "60000", 10), }, + // Issue #1133: job store, rate limiting, and cache are process-local Maps + // by default, which break correctness under horizontal scaling (a GET + // /tts/job/:id can land on a different pod than the one that processed + // the job, and rate limits multiply by replica count). Set REDIS_URL when + // running more than one instance behind a load balancer so state is + // shared across replicas. + sharedStore: process.env.REDIS_URL ? createRedisSharedStore(process.env.REDIS_URL) : undefined, }; // --------------------------------------------------------------------------- @@ -107,19 +127,15 @@ app.use((req: Request, res: Response, next: NextFunction) => { }); }); -// Issue #995: Express-level rate limiting (IP-based for anonymous, API-key-based for authenticated) +// Issue #995 / #1132: Express-level rate limiting, keyed on the caller's IP. +// The Authorization header cannot be used as a key here: it hasn't been +// validated yet (auth middleware runs after this, and is optional), so a +// rotating fake credential would otherwise reset the bucket on every request. const ttsRateLimitPerMinute = parseInt(process.env.TTS_RATE_LIMIT_PER_MINUTE || "60", 10); const ttsRateLimiter = rateLimit({ windowMs: 60 * 1000, max: ttsRateLimitPerMinute, - // Use API key as bucket key when present, otherwise fall back to client IP - keyGenerator: (req: Request): string => { - const authHeader = req.headers.authorization; - if (authHeader) { - return `apikey:${authHeader.replace(/^Bearer\s+/i, "")}`; - } - return `ip:${req.ip ?? "unknown"}`; - }, + keyGenerator: rateLimitKeyGenerator, handler: (_req: Request, res: Response): void => { res.setHeader("Retry-After", "60"); res.status(429).json({ error: "Too Many Requests" }); @@ -219,7 +235,7 @@ app.get("/health/live", createLivenessHandler(healthChecker)); * "status": "pending" * } */ -app.post("/tts/enqueue", (req: Request, res: Response) => { +app.post("/tts/enqueue", async (req: Request, res: Response) => { try { const { text, voiceId, provider } = req.body; const rateLimitKey = req.ip || "unknown"; @@ -234,8 +250,10 @@ app.post("/tts/enqueue", (req: Request, res: Response) => { return res.status(400).json({ error: `Unknown voice: ${voiceId}` }); } + // enqueueAsync so rate limiting is enforced consistently across + // replicas when REDIS_URL / config.sharedStore is configured (#1133). const credential = extractCredential(req); - const jobId = service.enqueue(text, voice, provider, credential, rateLimitKey, bypassCache); + const jobId = await service.enqueueAsync(text, voice, provider, credential, rateLimitKey, bypassCache); res.json({ jobId, status: "pending" }); } catch (error: any) { const statusCode = error.statusCode || 500; @@ -257,10 +275,12 @@ app.post("/tts/enqueue", (req: Request, res: Response) => { * "updatedAt": "2024-01-15T10:30:05Z" * } */ -app.get("/tts/job/:id", (req: Request, res: Response) => { +app.get("/tts/job/:id", async (req: Request, res: Response) => { try { + // getJobAsync falls back to the shared store so polling succeeds + // regardless of which replica originally processed the job (#1133). const credential = extractCredential(req); - const job = service.getJob(req.params.id, credential); + const job = await service.getJobAsync(req.params.id, credential); if (!job) { // Same response whether the job doesn't exist or belongs to another // tenant, so a credential can't distinguish the two by probing IDs. @@ -286,11 +306,11 @@ app.get("/tts/job/:id", (req: Request, res: Response) => { * ... * ] */ -app.get("/tts/jobs", (req: Request, res: Response) => { +app.get("/tts/jobs", async (req: Request, res: Response) => { try { const status = req.query.status as any; const credential = extractCredential(req); - const jobs = service.listJobs(status, credential); + const jobs = await service.listJobsAsync(status, credential); res.json(jobs); } catch (error: any) { const statusCode = error.statusCode || 500; @@ -376,12 +396,15 @@ app.use((err: any, req: Request, res: Response, next: any) => { // Server startup // --------------------------------------------------------------------------- -app.listen(port, () => { - console.log(`🎙️ TTS Service listening on port ${port}`); - console.log(`📊 Health check: GET http://localhost:${port}/health`); - console.log(`🔍 Readiness probe: GET http://localhost:${port}/health/ready`); - console.log(`💓 Liveness probe: GET http://localhost:${port}/health/live`); - console.log(`🎵 TTS endpoints: POST http://localhost:${port}/tts/enqueue`); -}); +// Guarded so importing this module (e.g. from tests) doesn't bind a real port. +if (require.main === module) { + app.listen(port, () => { + console.log(`🎙️ TTS Service listening on port ${port}`); + console.log(`📊 Health check: GET http://localhost:${port}/health`); + console.log(`🔍 Readiness probe: GET http://localhost:${port}/health/ready`); + console.log(`💓 Liveness probe: GET http://localhost:${port}/health/live`); + console.log(`🎵 TTS endpoints: POST http://localhost:${port}/tts/enqueue`); + }); +} export default app;