diff --git a/backend/config/__tests__/database.test.ts b/backend/config/__tests__/database.test.ts index 63e022d4..10418edc 100644 --- a/backend/config/__tests__/database.test.ts +++ b/backend/config/__tests__/database.test.ts @@ -71,4 +71,26 @@ describe('database config', () => { expect(poolConfig.max).toBe(25); expect(poolConfig.database).toBe(base.database); }); + + it('loads primary from DATABASE_URL', () => { + const config = loadDatabaseConfig({ + DATABASE_URL: 'postgresql://app:secret@db.internal:6543/subtrackr?sslmode=require', + }); + expect(config.primary.host).toBe('db.internal'); + expect(config.primary.port).toBe(6543); + expect(config.primary.user).toBe('app'); + expect(config.primary.password).toBe('secret'); + expect(config.primary.ssl).toEqual({ rejectUnauthorized: true }); + }); + + it('loads replicas from DATABASE_READ_URLS', () => { + const config = loadDatabaseConfig({ + DATABASE_READ_URLS: + 'postgresql://app:p@r1.internal:5432/subtrackr,postgresql://app:p@r2.internal:5433/subtrackr', + }); + expect(config.replicas).toEqual([ + { name: 'replica-1', host: 'r1.internal', port: 5432 }, + { name: 'replica-2', host: 'r2.internal', port: 5433 }, + ]); + }); }); diff --git a/backend/config/database.ts b/backend/config/database.ts index a539a9af..bbd82379 100644 --- a/backend/config/database.ts +++ b/backend/config/database.ts @@ -3,9 +3,11 @@ * * Environment variables (primary): * DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD, DB_SSL + * DATABASE_URL – optional postgres:// URL (overrides discrete DB_* host fields) * - * Read replicas (optional — comma-separated host:port pairs): + * Read replicas (optional — comma-separated host:port pairs OR URLs): * DB_READ_REPLICAS – e.g. "replica-1.internal:6432,replica-2.internal:6433" + * DATABASE_READ_URLS – e.g. "postgres://u:p@r1:5432/db,postgres://u:p@r2:5432/db" * DB_REPLICA_POOL_SIZE – PgBouncer pool size per replica (default: 25) * * Replication lag thresholds (milliseconds): @@ -17,6 +19,10 @@ */ import type { PoolConfig } from '../shared/db/connectionPool'; +import { + ConnectionStringRotator, + type ParsedConnectionString, +} from '../shared/db/connectionStringRotation'; export interface ReplicaEndpoint { /** Logical name used in metrics labels (replica-1, replica-2, …). */ @@ -77,26 +83,70 @@ function parseReplicaEndpoints(raw: string | undefined): ReplicaEndpoint[] { }); } +function poolDefaults(env: NodeJS.ProcessEnv): Pick< + Required, + 'max' | 'idleTimeoutMillis' | 'connectionTimeoutMillis' | 'statementTimeout' +> { + return { + max: parsePositiveInt(env.DB_POOL_MAX, 20), + idleTimeoutMillis: parsePositiveInt(env.DB_IDLE_TIMEOUT_MS, 10_000), + connectionTimeoutMillis: parsePositiveInt(env.DB_CONNECTION_TIMEOUT_MS, 30_000), + statementTimeout: parsePositiveInt(env.DB_STATEMENT_TIMEOUT_MS, 30_000), + }; +} + +function primaryFromParsed( + parsed: ParsedConnectionString, + env: NodeJS.ProcessEnv, +): Required { + return { + ...poolDefaults(env), + host: parsed.host, + port: parsed.port, + database: parsed.database, + user: parsed.user, + password: parsed.password, + ssl: (parsed.ssl || env.DB_SSL === 'true') ? { rejectUnauthorized: true } : false, + }; +} + function buildPrimaryConfig(env: NodeJS.ProcessEnv): Required { + if (env.DATABASE_URL?.trim()) { + const rotator = new ConnectionStringRotator({ primaryUrl: env.DATABASE_URL }); + const primary = rotator.getPrimary(); + if (primary) { + return primaryFromParsed(primary, env); + } + } + return { + ...poolDefaults(env), host: env.DB_HOST?.trim() || 'localhost', port: parsePositiveInt(env.DB_PORT, 5432), database: env.DB_NAME?.trim() || 'subtrackr', user: env.DB_USER?.trim() || 'postgres', password: env.DB_PASSWORD ?? '', - max: parsePositiveInt(env.DB_POOL_MAX, 20), - idleTimeoutMillis: parsePositiveInt(env.DB_IDLE_TIMEOUT_MS, 10_000), - connectionTimeoutMillis: parsePositiveInt(env.DB_CONNECTION_TIMEOUT_MS, 30_000), - statementTimeout: parsePositiveInt(env.DB_STATEMENT_TIMEOUT_MS, 30_000), ssl: env.DB_SSL === 'true' ? { rejectUnauthorized: true } : false, }; } +function resolveReplicas(env: NodeJS.ProcessEnv): ReplicaEndpoint[] { + if (env.DATABASE_READ_URLS?.trim()) { + const rotator = ConnectionStringRotator.fromEnv(env); + return rotator.getReplicas().map((r) => ({ + name: r.name, + host: r.host, + port: r.port, + })); + } + return parseReplicaEndpoints(env.DB_READ_REPLICAS); +} + /** Load database configuration from environment variables. */ export function loadDatabaseConfig(env: NodeJS.ProcessEnv = process.env): DatabaseConfig { return { primary: buildPrimaryConfig(env), - replicas: parseReplicaEndpoints(env.DB_READ_REPLICAS), + replicas: resolveReplicas(env), replicaPoolSize: parsePositiveInt( env.DB_REPLICA_POOL_SIZE, DEFAULT_DATABASE_CONFIG.replicaPoolSize, diff --git a/backend/elasticsearch/__tests__/replicaRouter.test.ts b/backend/elasticsearch/__tests__/replicaRouter.test.ts new file mode 100644 index 00000000..0f01a9ad --- /dev/null +++ b/backend/elasticsearch/__tests__/replicaRouter.test.ts @@ -0,0 +1,69 @@ +import { + ElasticsearchReplicaRouter, + createElasticsearchReplicaRouter, +} from '../replicaRouter'; +import { loadElasticsearchConfig, loadElasticsearchNodes } from '../config'; + +describe('elasticsearch replica config', () => { + it('loads primary and replica nodes from env', () => { + const nodes = loadElasticsearchNodes({ + ES_PRIMARY_URL: 'https://es-primary:9200', + ES_READ_REPLICA_URLS: 'https://es-r1:9200,https://es-r2:9200', + }); + expect(nodes).toEqual([ + { name: 'es-primary', url: 'https://es-primary:9200', role: 'primary' }, + { name: 'es-replica-1', url: 'https://es-r1:9200', role: 'replica' }, + { name: 'es-replica-2', url: 'https://es-r2:9200', role: 'replica' }, + ]); + }); + + it('defaults to in-process when no nodes configured', () => { + const config = loadElasticsearchConfig({}); + expect(config.nodes).toEqual([]); + expect(config.readWriteSplitting).toBe(true); + expect(config.automaticFailover).toBe(true); + }); +}); + +describe('ElasticsearchReplicaRouter', () => { + it('routes writes to primary and reads to replicas', () => { + const router = createElasticsearchReplicaRouter({ + ES_PRIMARY_URL: 'https://es-primary:9200', + ES_READ_REPLICA_URLS: 'https://es-r1:9200,https://es-r2:9200', + }); + + expect(router.route('write').route).toBe('primary'); + expect(router.route('read').route).toBe('replica:es-replica-1'); + expect(router.route('read').route).toBe('replica:es-replica-2'); + }); + + it('fails over reads to primary when replicas are down', () => { + const router = new ElasticsearchReplicaRouter( + loadElasticsearchConfig({ + ES_PRIMARY_URL: 'https://es-primary:9200', + ES_READ_REPLICA_URLS: 'https://es-r1:9200', + }), + ); + + router.markFailed('es-replica-1'); + const result = router.route('read'); + expect(result.route).toBe('failover-primary'); + expect(result.failedOver).toBe(true); + expect(result.node?.name).toBe('es-primary'); + }); + + it('supports connection string / node URL rotation', () => { + const router = createElasticsearchReplicaRouter({ + ES_PRIMARY_URL: 'https://es-primary:9200', + ES_READ_REPLICA_URLS: 'https://es-r1:9200', + }); + + const rotated = router.rotateNodes([ + { name: 'es-primary', url: 'https://es-primary-new:9200', role: 'primary' }, + { name: 'es-replica-1', url: 'https://es-r1-new:9200', role: 'replica' }, + ]); + + expect(rotated.getPrimary()?.url).toBe('https://es-primary-new:9200'); + expect(rotated.route('read').node?.url).toBe('https://es-r1-new:9200'); + }); +}); diff --git a/backend/elasticsearch/config.ts b/backend/elasticsearch/config.ts index 9a57901e..0b302bb3 100644 --- a/backend/elasticsearch/config.ts +++ b/backend/elasticsearch/config.ts @@ -3,8 +3,21 @@ * In this mobile-first architecture the "cluster" is an in-process index * backed by AsyncStorage, mirroring a real ES setup so the service layer * can be swapped for a remote cluster without changing callers. + * + * When remote nodes are configured, reads prefer replica nodes and writes + * go to the primary, with automatic failover when a replica is marked down. */ +export type ElasticsearchNodeRole = 'primary' | 'replica'; + +export interface ElasticsearchNode { + /** Logical name used in routing / metrics (es-primary, es-replica-1, …). */ + name: string; + /** Node HTTP URL, e.g. https://es-replica-1.internal:9200 */ + url: string; + role: ElasticsearchNodeRole; +} + export interface ElasticsearchConfig { indexName: string; fuzzyMaxEdits: number; @@ -20,6 +33,57 @@ export interface ElasticsearchConfig { minQueryLength: number; /** Whether to highlight matching terms in result fields */ highlightEnabled: boolean; + /** + * Optional remote cluster nodes. When empty the in-process index is used. + * Env: ES_PRIMARY_URL + ES_READ_REPLICA_URLS (comma-separated). + */ + nodes: ElasticsearchNode[]; + /** Route search/read requests to replica nodes when available. */ + readWriteSplitting: boolean; + /** Fail reads over to primary when no healthy replica remains. */ + automaticFailover: boolean; +} + +function parsePositiveInt(value: string | undefined, fallback: number): number { + if (value === undefined || value === '') return fallback; + const parsed = Number.parseInt(value, 10); + return Number.isFinite(parsed) && parsed >= 0 ? parsed : fallback; +} + +/** Build ES node list from environment variables. */ +export function loadElasticsearchNodes(env: NodeJS.ProcessEnv = process.env): ElasticsearchNode[] { + const nodes: ElasticsearchNode[] = []; + const primaryUrl = env.ES_PRIMARY_URL?.trim(); + if (primaryUrl) { + nodes.push({ name: 'es-primary', url: primaryUrl, role: 'primary' }); + } + + const replicaRaw = env.ES_READ_REPLICA_URLS?.trim(); + if (replicaRaw) { + replicaRaw + .split(',') + .map((u) => u.trim()) + .filter(Boolean) + .forEach((url, index) => { + nodes.push({ name: `es-replica-${index + 1}`, url, role: 'replica' }); + }); + } + + return nodes; +} + +export function loadElasticsearchConfig( + env: NodeJS.ProcessEnv = process.env, + overrides: Partial = {}, +): ElasticsearchConfig { + return { + ...DEFAULT_ES_CONFIG, + nodes: loadElasticsearchNodes(env), + readWriteSplitting: env.ES_READ_WRITE_SPLITTING !== 'false', + automaticFailover: env.ES_AUTOMATIC_FAILOVER !== 'false', + maxResults: parsePositiveInt(env.ES_MAX_RESULTS, DEFAULT_ES_CONFIG.maxResults), + ...overrides, + }; } export const DEFAULT_ES_CONFIG: ElasticsearchConfig = { @@ -38,6 +102,9 @@ export const DEFAULT_ES_CONFIG: ElasticsearchConfig = { analyticsBufferSize: 500, minQueryLength: 1, highlightEnabled: true, + nodes: [], + readWriteSplitting: true, + automaticFailover: true, }; export interface FieldMapping { diff --git a/backend/elasticsearch/replicaRouter.ts b/backend/elasticsearch/replicaRouter.ts new file mode 100644 index 00000000..36c884b4 --- /dev/null +++ b/backend/elasticsearch/replicaRouter.ts @@ -0,0 +1,107 @@ +/** + * Elasticsearch read/write node router with automatic failover. + * + * Writes always target the primary node. Reads round-robin across healthy + * replicas and fall back to primary when replicas are unavailable. + */ + +import type { ElasticsearchConfig, ElasticsearchNode } from './config'; +import { loadElasticsearchConfig } from './config'; + +export type ElasticsearchRouteKind = 'read' | 'write'; + +export interface ElasticsearchRouteResult { + node: ElasticsearchNode | null; + /** in-process | primary | replica: | failover-primary */ + route: string; + failedOver: boolean; +} + +export class ElasticsearchReplicaRouter { + private readonly config: ElasticsearchConfig; + private readonly failed = new Set(); + private replicaIndex = 0; + + constructor(config: ElasticsearchConfig = loadElasticsearchConfig()) { + this.config = config; + } + + getConfig(): ElasticsearchConfig { + return this.config; + } + + getPrimary(): ElasticsearchNode | null { + return this.config.nodes.find((n) => n.role === 'primary') ?? null; + } + + getReplicas(): ElasticsearchNode[] { + return this.config.nodes.filter((n) => n.role === 'replica'); + } + + getHealthyReplicas(): ElasticsearchNode[] { + return this.getReplicas().filter((n) => !this.failed.has(n.name)); + } + + markFailed(name: string): void { + this.failed.add(name); + } + + markHealthy(name: string): void { + this.failed.delete(name); + } + + /** + * Select a node for the operation. Returns null when no remote nodes are + * configured (callers should use the in-process index). + */ + route(kind: ElasticsearchRouteKind): ElasticsearchRouteResult { + if (this.config.nodes.length === 0) { + return { node: null, route: 'in-process', failedOver: false }; + } + + if (kind === 'write' || !this.config.readWriteSplitting) { + const primary = this.getPrimary(); + return { + node: primary, + route: primary ? 'primary' : 'in-process', + failedOver: false, + }; + } + + const healthy = this.getHealthyReplicas(); + if (healthy.length > 0) { + const selected = healthy[this.replicaIndex % healthy.length]!; + this.replicaIndex = (this.replicaIndex + 1) % healthy.length; + return { + node: selected, + route: `replica:${selected.name}`, + failedOver: false, + }; + } + + if (this.config.automaticFailover) { + const primary = this.getPrimary(); + return { + node: primary, + route: primary ? 'failover-primary' : 'in-process', + failedOver: true, + }; + } + + return { node: null, route: 'in-process', failedOver: true }; + } + + /** Rotate remote node URLs (connection string rotation for ES). */ + rotateNodes(nodes: ElasticsearchNode[]): ElasticsearchReplicaRouter { + return new ElasticsearchReplicaRouter({ + ...this.config, + nodes: [...nodes], + }); + } +} + +export function createElasticsearchReplicaRouter( + env: NodeJS.ProcessEnv = process.env, +): ElasticsearchReplicaRouter { + return new ElasticsearchReplicaRouter(loadElasticsearchConfig(env)); +} diff --git a/backend/services/index.ts b/backend/services/index.ts index a1a2b04c..d914870c 100644 --- a/backend/services/index.ts +++ b/backend/services/index.ts @@ -62,6 +62,18 @@ export { AuditService, auditService } from './shared/auditService'; export { RateLimitingService, rateLimitingService } from './shared/rateLimitingService'; export { MonitoringService, monitoringService } from './shared/monitoring'; export { apiClient } from './shared/apiClient'; +export { + DatabaseService, + getDatabaseService, + resetDatabaseService, + ConnectionStringRotator, +} from './shared/databaseService'; +export type { + DatabaseFailoverStatus, + DatabaseServiceOptions, + ConnectionStringRotationOptions, + ParsedConnectionString, +} from './shared/databaseService'; // ── Upstream additions ──────────────────────────────────────────────────────── export { ExportService, exportService } from './exportService'; diff --git a/backend/services/shared/__tests__/databaseService.test.ts b/backend/services/shared/__tests__/databaseService.test.ts new file mode 100644 index 00000000..dc0a942f --- /dev/null +++ b/backend/services/shared/__tests__/databaseService.test.ts @@ -0,0 +1,162 @@ +import type { DatabaseConfig } from '../../../config/database'; +import type { Pool, QueryResult } from '../../../shared/db/connectionPool'; +import { ReadWritePool } from '../../../shared/db/readWriteRouter'; +import { ConnectionStringRotator } from '../../../shared/db/connectionStringRotation'; +import { DatabaseService } from '../databaseService'; + +function makeMockPool(label: string, lagMs = 100): Pool { + return { + query: jest.fn(async (sql: string) => { + if (sql.includes('pg_last_xact_replay_timestamp')) { + return { rows: [{ lag_ms: lagMs }], rowCount: 1 }; + } + return { rows: [{ source: label }], rowCount: 1 } as QueryResult<{ source: string }>; + }), + connect: jest.fn(), + end: jest.fn(async () => undefined), + on: jest.fn(), + totalCount: 5, + idleCount: 2, + waitingCount: 0, + } as unknown as Pool; +} + +function makeConfig(overrides: Partial = {}): DatabaseConfig { + return { + primary: { + host: 'primary', + port: 5432, + database: 'subtrackr', + user: 'postgres', + password: '', + max: 20, + idleTimeoutMillis: 10_000, + connectionTimeoutMillis: 30_000, + statementTimeout: 30_000, + ssl: false, + }, + replicas: [ + { name: 'replica-1', host: 'r1', port: 6433 }, + { name: 'replica-2', host: 'r2', port: 6434 }, + ], + replicaPoolSize: 25, + replicationLagP99AlarmMs: 1_000, + replicationLagFailoverMs: 5_000, + staleReadDefaultSeconds: 30, + lagPollIntervalMs: 60_000, + ...overrides, + }; +} + +describe('DatabaseService', () => { + it('splits reads to replicas and writes to primary', async () => { + const primary = makeMockPool('primary'); + const replica1 = makeMockPool('replica-1'); + const config = makeConfig({ + replicas: [{ name: 'replica-1', host: 'r1', port: 6433 }], + }); + const pool = new ReadWritePool( + primary, + new Map([['replica-1', replica1]]), + config.replicas, + config, + ); + await pool.pollReplicationLag(); + + const service = new DatabaseService({ config, pool }); + await service.initialize(); + + await service.read('SELECT 1'); + expect(replica1.query).toHaveBeenCalled(); + + await service.write('INSERT INTO plans (id) VALUES ($1)', ['p1']); + expect(primary.query).toHaveBeenCalledWith('INSERT INTO plans (id) VALUES ($1)', ['p1']); + }); + + it('fails over reads to primary and recovers', async () => { + const primary = makeMockPool('primary'); + const replica1 = makeMockPool('replica-1'); + const config = makeConfig({ + replicas: [{ name: 'replica-1', host: 'r1', port: 6433 }], + }); + const pool = new ReadWritePool( + primary, + new Map([['replica-1', replica1]]), + config.replicas, + config, + ); + await pool.pollReplicationLag(); + + const service = new DatabaseService({ + config, + pool, + rotator: new ConnectionStringRotator({ + primaryUrl: 'postgresql://u:p@primary:5432/subtrackr', + replicaUrls: 'postgresql://u:p@r1:6433/subtrackr', + }), + }); + + const before = service.getFailoverStatus(); + expect(before.mode).toBe('replicas-active'); + + const failed = await service.failoverToPrimary('replica-lag'); + expect(failed.mode).toBe('failover-primary'); + expect(failed.healthyReplicas).toBe(0); + + await service.read('SELECT 1'); + expect(primary.query).toHaveBeenCalledWith('SELECT 1', undefined); + + const recovered = await service.recoverFromFailover(); + expect(recovered.mode).toBe('replicas-active'); + }); + + it('rotates connection strings and bumps generation', async () => { + const primary = makeMockPool('primary'); + const config = makeConfig({ replicas: [] }); + const pool = new ReadWritePool(primary, new Map(), [], config); + const rotator = new ConnectionStringRotator({ + primaryUrl: 'postgresql://u:old@primary:5432/subtrackr', + replicaUrls: 'postgresql://u:old@r1:5432/subtrackr', + }); + + const service = new DatabaseService({ config, pool, rotator }); + await service.initialize(); + + // Avoid creating real pg pools during rotation — stub rebuild via empty replicas. + const result = await service.rotateConnectionStrings({ + primaryUrl: 'postgresql://u:new@primary:5432/subtrackr', + replicaUrls: [], + }); + + expect(result.generation).toBe(1); + expect(service.getFailoverStatus().connectionGeneration).toBe(1); + expect(service.getConnectionSnapshot()[0]?.url).toContain('***'); + expect(service.getConfig().primary.password).toBe('new'); + }); + + it('falls back to primary when replica lag exceeds threshold', async () => { + const primary = makeMockPool('primary'); + const replica1 = makeMockPool('replica-1', 8_000); + const config = makeConfig({ + replicas: [{ name: 'replica-1', host: 'r1', port: 6433 }], + replicationLagFailoverMs: 5_000, + }); + const pool = new ReadWritePool( + primary, + new Map([['replica-1', replica1]]), + config.replicas, + config, + ); + await pool.pollReplicationLag(); + + const service = new DatabaseService({ config, pool }); + const headers = new Map(); + await service.runWithRoutingContext({ responseHeaders: headers }, async () => { + await service.query('SELECT * FROM subscriptions'); + }); + + expect(primary.query).toHaveBeenCalledWith('SELECT * FROM subscriptions', undefined); + expect(headers.get('X-DB-Route')).toBe('primary'); + expect(headers.get('X-DB-Route-Warning')).toBe('replication-lag-fallback-primary'); + }); +}); diff --git a/backend/services/shared/databaseService.ts b/backend/services/shared/databaseService.ts new file mode 100644 index 00000000..c1038b43 --- /dev/null +++ b/backend/services/shared/databaseService.ts @@ -0,0 +1,302 @@ +/** + * Database service — read/write splitting, replica lag monitoring, + * automatic failover, and connection-string rotation. + * + * Wraps the shared ReadWritePool so the service layer can depend on a + * stable facade without knowing about pool internals. + */ + +import { + type DatabaseConfig, + type ReplicaEndpoint, + loadDatabaseConfig, + replicaPoolConfig, +} from '../../config/database'; +import { + ConnectionStringRotator, + type ConnectionStringRotationOptions, + type ParsedConnectionString, +} from '../../shared/db/connectionStringRotation'; +import { + closePool, + createPool, + getPool, + type Pool, + type QueryResult, +} from '../../shared/db/connectionPool'; +import { + type ReadWritePool, + type ReplicaLagState, + createReadWritePool, + runWithQueryRoutingContext, +} from '../../shared/db/readWriteRouter'; +import { isReadQuery } from '../../shared/db/queryClassifier'; + +export interface DatabaseFailoverStatus { + mode: 'primary-only' | 'replicas-active' | 'failover-primary'; + healthyReplicas: number; + totalReplicas: number; + lagStates: ReplicaLagState[]; + connectionGeneration: number; +} + +export interface DatabaseServiceOptions { + config?: DatabaseConfig; + rotator?: ConnectionStringRotator; + /** Injected pool for tests. */ + pool?: Pool | ReadWritePool; +} + +export class DatabaseService { + private config: DatabaseConfig; + private readonly rotator: ConnectionStringRotator; + private pool: Pool | ReadWritePool | null; + private forcedPrimaryFailover = false; + private initialized = false; + + constructor(options: DatabaseServiceOptions = {}) { + this.config = options.config ?? loadDatabaseConfig(); + this.rotator = + options.rotator ?? + new ConnectionStringRotator({ + primaryUrl: process.env.DATABASE_URL, + replicaUrls: process.env.DATABASE_READ_URLS, + }); + this.pool = options.pool ?? null; + } + + /** Initialise the underlying pool (idempotent). */ + async initialize(): Promise { + if (this.initialized && this.pool) return; + if (!this.pool) { + if (this.config.replicas.length > 0) { + this.pool = await createReadWritePool({ config: this.config }); + } else { + this.pool = await getPool(); + } + } + this.initialized = true; + } + + /** Route a query: SELECTs to replicas, writes to primary. */ + async query(sql: string, params?: unknown[]): Promise> { + await this.initialize(); + const pool = this.requirePool(); + + if (this.forcedPrimaryFailover || !isReadQuery(sql)) { + return this.queryPrimary(sql, params); + } + + return pool.query(sql, params); + } + + /** Force a write against the primary. */ + async write(sql: string, params?: unknown[]): Promise> { + return this.queryPrimary(sql, params); + } + + /** Prefer a read replica (falls back to primary on lag / failure). */ + async read(sql: string, params?: unknown[]): Promise> { + await this.initialize(); + if (this.forcedPrimaryFailover) { + return this.queryPrimary(sql, params); + } + return this.requirePool().query(sql, params); + } + + getLagStates(): ReplicaLagState[] { + const rw = this.asReadWritePool(); + return rw ? rw.getLagStates() : []; + } + + getFailoverStatus(): DatabaseFailoverStatus { + const lagStates = this.getLagStates(); + const healthyReplicas = lagStates.filter((s) => s.available).length; + const totalReplicas = this.config.replicas.length; + + let mode: DatabaseFailoverStatus['mode'] = 'primary-only'; + if (this.forcedPrimaryFailover) { + mode = 'failover-primary'; + } else if (totalReplicas > 0 && healthyReplicas > 0) { + mode = 'replicas-active'; + } else if (totalReplicas > 0) { + mode = 'failover-primary'; + } + + return { + mode, + healthyReplicas, + totalReplicas, + lagStates, + connectionGeneration: this.rotator.getGeneration(), + }; + } + + /** + * Force all reads onto the primary (automatic / operator failover). + * Replicas stay pooled so `recoverFromFailover()` can restore them. + */ + async failoverToPrimary(reason = 'manual'): Promise { + await this.initialize(); + this.forcedPrimaryFailover = true; + const rw = this.asReadWritePool(); + if (rw) { + for (const state of rw.getLagStates()) { + rw.markReplicaUnavailable(state.name); + this.rotator.markFailed(state.name); + } + } + console.warn(`[DatabaseService] Failover to primary (${reason})`); + return this.getFailoverStatus(); + } + + /** Clear forced failover and re-enable healthy replicas. */ + async recoverFromFailover(): Promise { + this.forcedPrimaryFailover = false; + const rw = this.asReadWritePool(); + if (rw) { + for (const state of rw.getLagStates()) { + rw.markReplicaAvailable(state.name); + this.rotator.markHealthy(state.name); + } + await rw.pollReplicationLag(); + } + return this.getFailoverStatus(); + } + + /** + * Rotate connection strings (credential / endpoint hot-swap) and rebuild + * replica pools against the new URLs. + */ + async rotateConnectionStrings( + options: ConnectionStringRotationOptions, + ): Promise<{ generation: number; replicas: ParsedConnectionString[] }> { + const generation = this.rotator.rotate(options); + const primary = this.rotator.getPrimary(); + const replicas = this.rotator.getReplicas(); + + if (primary) { + this.config = { + ...this.config, + primary: { + ...this.config.primary, + host: primary.host, + port: primary.port, + database: primary.database, + user: primary.user, + password: primary.password, + ssl: primary.ssl ? { rejectUnauthorized: true } : this.config.primary.ssl, + }, + replicas: replicas.map((r) => ({ name: r.name, host: r.host, port: r.port })), + }; + } else { + this.config = { + ...this.config, + replicas: replicas.map((r) => ({ name: r.name, host: r.host, port: r.port })), + }; + } + + await this.rebuildReplicaPools(this.config.replicas); + this.forcedPrimaryFailover = false; + + return { generation, replicas }; + } + + getConnectionSnapshot() { + return this.rotator.toSafeSnapshot(); + } + + getConfig(): DatabaseConfig { + return this.config; + } + + getRotator(): ConnectionStringRotator { + return this.rotator; + } + + /** Run a callback with routing context (stale-accept / response headers). */ + runWithRoutingContext( + context: { staleAcceptSeconds?: number; responseHeaders?: Map }, + fn: () => T | Promise, + ): T | Promise { + return runWithQueryRoutingContext(context, fn); + } + + async shutdown(): Promise { + const rw = this.asReadWritePool(); + if (rw) { + rw.stopLagMonitoring(); + await rw.end(); + } else if (this.pool) { + await this.pool.end(); + } else { + await closePool(); + } + this.pool = null; + this.initialized = false; + } + + private async queryPrimary(sql: string, params?: unknown[]): Promise> { + await this.initialize(); + const pool = this.requirePool(); + const rw = this.asReadWritePool(); + if (rw) { + return rw.primary.query(sql, params); + } + return pool.query(sql, params); + } + + private async rebuildReplicaPools(endpoints: ReplicaEndpoint[]): Promise { + await this.initialize(); + const rw = this.asReadWritePool(); + + const nextPools = new Map(); + for (const endpoint of endpoints) { + const poolConfig = replicaPoolConfig(endpoint, this.config.primary, this.config.replicaPoolSize); + nextPools.set(endpoint.name, await createPool(poolConfig)); + } + + if (rw) { + await rw.replaceReplicaPools(nextPools, endpoints); + return; + } + + if (endpoints.length > 0) { + this.pool = await createReadWritePool({ + config: { ...this.config, replicas: endpoints }, + primaryPool: this.pool ?? undefined, + replicaPools: nextPools, + }); + } + } + + private requirePool(): Pool | ReadWritePool { + if (!this.pool) { + throw new Error('DatabaseService is not initialised'); + } + return this.pool; + } + + private asReadWritePool(): ReadWritePool | null { + if (this.pool && 'getLagStates' in this.pool) { + return this.pool as ReadWritePool; + } + return null; + } +} + +let _databaseService: DatabaseService | null = null; + +export function getDatabaseService(): DatabaseService { + if (!_databaseService) { + _databaseService = new DatabaseService(); + } + return _databaseService; +} + +export function resetDatabaseService(): void { + _databaseService = null; +} + +export { ConnectionStringRotator }; +export type { ConnectionStringRotationOptions, ParsedConnectionString }; diff --git a/backend/services/shared/index.ts b/backend/services/shared/index.ts index b94ed9d1..d0be4442 100644 --- a/backend/services/shared/index.ts +++ b/backend/services/shared/index.ts @@ -29,6 +29,18 @@ export type { ClassificationLevel, PiiPattern, ClassifyResult, RedactOptions } f export { redactResponse, createPiiRedactionMiddleware } from './apiResponse'; export { RateLimitingService, rateLimitingService } from './rateLimitingService'; export type { BypassConfig, CustomLimits } from './rateLimitingService'; +export { + DatabaseService, + getDatabaseService, + resetDatabaseService, + ConnectionStringRotator, +} from './databaseService'; +export type { + DatabaseFailoverStatus, + DatabaseServiceOptions, + ConnectionStringRotationOptions, + ParsedConnectionString, +} from './databaseService'; export { createRateLimitMiddleware, createRateLimitStatusMiddleware, diff --git a/backend/shared/db/__tests__/connectionStringRotation.test.ts b/backend/shared/db/__tests__/connectionStringRotation.test.ts new file mode 100644 index 00000000..2327e37e --- /dev/null +++ b/backend/shared/db/__tests__/connectionStringRotation.test.ts @@ -0,0 +1,114 @@ +import { + ConnectionStringRotator, + parseConnectionString, + redactConnectionString, +} from '../connectionStringRotation'; + +describe('connectionStringRotation', () => { + describe('parseConnectionString', () => { + it('parses a postgres URL', () => { + const parsed = parseConnectionString( + 'postgresql://app:s3cret@db.internal:6432/subtrackr?sslmode=require', + 'primary', + 'primary', + ); + expect(parsed.host).toBe('db.internal'); + expect(parsed.port).toBe(6432); + expect(parsed.database).toBe('subtrackr'); + expect(parsed.user).toBe('app'); + expect(parsed.password).toBe('s3cret'); + expect(parsed.ssl).toBe(true); + expect(parsed.role).toBe('primary'); + }); + + it('defaults port and database', () => { + const parsed = parseConnectionString('postgres://user@localhost/', 'replica', 'replica-1'); + expect(parsed.port).toBe(5432); + expect(parsed.database).toBe('subtrackr'); + expect(parsed.password).toBe(''); + }); + + it('rejects empty and invalid URLs', () => { + expect(() => parseConnectionString('', 'primary', 'primary')).toThrow(/Empty/); + expect(() => parseConnectionString('not-a-url', 'primary', 'primary')).toThrow(/Invalid/); + expect(() => parseConnectionString('mysql://localhost/db', 'primary', 'primary')).toThrow( + /Unsupported protocol/, + ); + }); + }); + + describe('redactConnectionString', () => { + it('masks passwords', () => { + expect(redactConnectionString('postgresql://app:s3cret@db:5432/subtrackr')).toContain( + '***', + ); + expect(redactConnectionString('postgresql://app:s3cret@db:5432/subtrackr')).not.toContain( + 's3cret', + ); + }); + }); + + describe('ConnectionStringRotator', () => { + it('loads primary and replica URLs', () => { + const rotator = new ConnectionStringRotator({ + primaryUrl: 'postgresql://u:p@primary:5432/subtrackr', + replicaUrls: + 'postgresql://u:p@r1:5432/subtrackr,postgresql://u:p@r2:5433/subtrackr', + }); + + expect(rotator.getPrimary()?.host).toBe('primary'); + expect(rotator.getReplicas()).toHaveLength(2); + expect(rotator.getReplicas()[0]?.name).toBe('replica-1'); + expect(rotator.getReplicas()[1]?.port).toBe(5433); + }); + + it('round-robins healthy replicas and skips failed ones', () => { + const rotator = new ConnectionStringRotator({ + replicaUrls: [ + 'postgresql://u:p@r1:5432/db', + 'postgresql://u:p@r2:5432/db', + ], + }); + + expect(rotator.nextReplica()?.name).toBe('replica-1'); + expect(rotator.nextReplica()?.name).toBe('replica-2'); + expect(rotator.nextReplica()?.name).toBe('replica-1'); + + rotator.markFailed('replica-1'); + expect(rotator.nextReplica()?.name).toBe('replica-2'); + expect(rotator.nextReplica()?.name).toBe('replica-2'); + + rotator.markFailed('replica-2'); + expect(rotator.nextReplica()).toBeNull(); + }); + + it('rotates connection strings and bumps generation', () => { + const rotator = new ConnectionStringRotator({ + primaryUrl: 'postgresql://u:old@primary:5432/db', + replicaUrls: 'postgresql://u:old@r1:5432/db', + }); + + rotator.markFailed('replica-1'); + const generation = rotator.rotate({ + primaryUrl: 'postgresql://u:new@primary:5432/db', + replicaUrls: 'postgresql://u:new@r1:5432/db,postgresql://u:new@r2:5432/db', + }); + + expect(generation).toBe(1); + expect(rotator.getGeneration()).toBe(1); + expect(rotator.getPrimary()?.password).toBe('new'); + expect(rotator.getReplicas()).toHaveLength(2); + expect(rotator.isFailed('replica-1')).toBe(false); + expect(rotator.toSafeSnapshot()[0]?.url).toContain('***'); + }); + + it('builds from env DATABASE_URL / DATABASE_READ_URLS', () => { + const rotator = ConnectionStringRotator.fromEnv({ + DATABASE_URL: 'postgresql://u:p@primary:5432/subtrackr', + DATABASE_READ_URLS: 'postgresql://u:p@replica:6432/subtrackr', + }); + expect(rotator.getPrimary()?.host).toBe('primary'); + expect(rotator.getReplicas()[0]?.port).toBe(6432); + }); + }); +}); diff --git a/backend/shared/db/__tests__/readWriteRouter.test.ts b/backend/shared/db/__tests__/readWriteRouter.test.ts index 8803ec29..86e36365 100644 --- a/backend/shared/db/__tests__/readWriteRouter.test.ts +++ b/backend/shared/db/__tests__/readWriteRouter.test.ts @@ -278,6 +278,32 @@ describe('readWriteRouter', () => { expect(state?.lagMs).toBe(300); expect(state?.lagP99Ms).toBe(300); }); + + it('replaces replica pools during connection string rotation', async () => { + const primary = makeMockPool('primary'); + const oldReplica = makeMockPool('replica-1'); + const newReplica = makeMockPool('replica-2'); + const pool = new ReadWritePool( + primary, + new Map([['replica-1', oldReplica]]), + [{ name: 'replica-1', host: 'r1', port: 6433 }], + makeConfig({ replicas: [{ name: 'replica-1', host: 'r1', port: 6433 }] }), + ); + await pool.pollReplicationLag(); + + await pool.replaceReplicaPools( + new Map([['replica-2', newReplica]]), + [{ name: 'replica-2', host: 'r2', port: 6434 }], + ); + await pool.pollReplicationLag(); + + expect(oldReplica.end).toHaveBeenCalled(); + const responseHeaders = new Map(); + await runWithQueryRoutingContext({ responseHeaders }, async () => { + await pool.query('SELECT 1'); + }); + expect(responseHeaders.get('X-DB-Route')).toBe('replica:replica-2'); + }); }); describe('attachRoutingHeaderInterceptor', () => { diff --git a/backend/shared/db/connectionStringRotation.ts b/backend/shared/db/connectionStringRotation.ts new file mode 100644 index 00000000..901e640f --- /dev/null +++ b/backend/shared/db/connectionStringRotation.ts @@ -0,0 +1,204 @@ +/** + * Connection string rotation for primary + read-replica endpoints. + * + * Supports hot-swapping DATABASE_URL / DATABASE_READ_URLS style connection + * strings (credential rotation) and round-robin selection across healthy + * endpoints during failover. + */ + +export type ConnectionRole = 'primary' | 'replica'; + +export interface ParsedConnectionString { + /** Original connection string (password redacted in toSafeString). */ + connectionString: string; + host: string; + port: number; + database: string; + user: string; + password: string; + ssl: boolean; + role: ConnectionRole; + /** Logical name used in metrics / routing (primary, replica-1, …). */ + name: string; +} + +export interface ConnectionStringRotationOptions { + /** Primary connection string (DATABASE_URL). */ + primaryUrl?: string; + /** Comma-separated or array of replica URLs (DATABASE_READ_URLS). */ + replicaUrls?: string | string[]; +} + +/** Parse a postgres connection URL into structured fields. */ +export function parseConnectionString( + raw: string, + role: ConnectionRole, + name: string, +): ParsedConnectionString { + const trimmed = raw.trim(); + if (!trimmed) { + throw new Error(`Empty connection string for ${name}`); + } + + let url: URL; + try { + url = new URL(trimmed); + } catch { + throw new Error(`Invalid connection string for ${name}`); + } + + if (!['postgres:', 'postgresql:'].includes(url.protocol)) { + throw new Error(`Unsupported protocol in connection string for ${name}: ${url.protocol}`); + } + + const sslMode = url.searchParams.get('sslmode'); + const ssl = + sslMode === 'require' || + sslMode === 'verify-ca' || + sslMode === 'verify-full' || + url.searchParams.get('ssl') === 'true'; + + return { + connectionString: trimmed, + host: url.hostname || 'localhost', + port: url.port ? Number.parseInt(url.port, 10) : 5432, + database: decodeURIComponent((url.pathname || '/subtrackr').replace(/^\//, '') || 'subtrackr'), + user: decodeURIComponent(url.username || 'postgres'), + password: decodeURIComponent(url.password || ''), + ssl, + role, + name, + }; +} + +function splitReplicaUrls(raw: string | string[] | undefined): string[] { + if (!raw) return []; + if (Array.isArray(raw)) { + return raw.map((u) => u.trim()).filter(Boolean); + } + return raw + .split(',') + .map((u) => u.trim()) + .filter(Boolean); +} + +/** Redact password from a connection string for logs / diagnostics. */ +export function redactConnectionString(connectionString: string): string { + try { + const url = new URL(connectionString); + if (url.password) { + url.password = '***'; + } + return url.toString(); + } catch { + return '[invalid-connection-string]'; + } +} + +/** + * Rotates through healthy connection strings for primary and read replicas. + * Call `rotate()` when secrets/credentials change to hot-swap endpoints + * without restarting the process. + */ +export class ConnectionStringRotator { + private primary: ParsedConnectionString | null = null; + private replicas: ParsedConnectionString[] = []; + private failed = new Set(); + private replicaIndex = 0; + private rotationGeneration = 0; + + constructor(options: ConnectionStringRotationOptions = {}) { + this.applyOptions(options); + } + + /** Load endpoints from env (DATABASE_URL + DATABASE_READ_URLS). */ + static fromEnv(env: NodeJS.ProcessEnv = process.env): ConnectionStringRotator { + return new ConnectionStringRotator({ + primaryUrl: env.DATABASE_URL, + replicaUrls: env.DATABASE_READ_URLS, + }); + } + + getGeneration(): number { + return this.rotationGeneration; + } + + getPrimary(): ParsedConnectionString | null { + return this.primary; + } + + getReplicas(): ParsedConnectionString[] { + return [...this.replicas]; + } + + getHealthyReplicas(): ParsedConnectionString[] { + return this.replicas.filter((r) => !this.failed.has(r.name)); + } + + /** Round-robin next healthy replica connection string. */ + nextReplica(): ParsedConnectionString | null { + const healthy = this.getHealthyReplicas(); + if (healthy.length === 0) return null; + const selected = healthy[this.replicaIndex % healthy.length]!; + this.replicaIndex = (this.replicaIndex + 1) % healthy.length; + return selected; + } + + markFailed(name: string): void { + this.failed.add(name); + } + + markHealthy(name: string): void { + this.failed.delete(name); + } + + isFailed(name: string): boolean { + return this.failed.has(name); + } + + /** + * Hot-swap connection strings (credential / endpoint rotation). + * Clears failure state and bumps generation so pool rebuilders can react. + */ + rotate(options: ConnectionStringRotationOptions): number { + this.applyOptions(options); + this.failed.clear(); + this.replicaIndex = 0; + this.rotationGeneration += 1; + return this.rotationGeneration; + } + + /** Snapshot of all configured endpoints (password redacted). */ + toSafeSnapshot(): Array<{ name: string; role: ConnectionRole; url: string; failed: boolean }> { + const entries: Array<{ name: string; role: ConnectionRole; url: string; failed: boolean }> = []; + if (this.primary) { + entries.push({ + name: this.primary.name, + role: 'primary', + url: redactConnectionString(this.primary.connectionString), + failed: this.failed.has(this.primary.name), + }); + } + for (const replica of this.replicas) { + entries.push({ + name: replica.name, + role: 'replica', + url: redactConnectionString(replica.connectionString), + failed: this.failed.has(replica.name), + }); + } + return entries; + } + + private applyOptions(options: ConnectionStringRotationOptions): void { + if (options.primaryUrl?.trim()) { + this.primary = parseConnectionString(options.primaryUrl, 'primary', 'primary'); + } else { + this.primary = null; + } + + this.replicas = splitReplicaUrls(options.replicaUrls).map((url, index) => + parseConnectionString(url, 'replica', `replica-${index + 1}`), + ); + } +} diff --git a/backend/shared/db/readWriteRouter.ts b/backend/shared/db/readWriteRouter.ts index 687e7ef2..0d7264fd 100644 --- a/backend/shared/db/readWriteRouter.ts +++ b/backend/shared/db/readWriteRouter.ts @@ -95,8 +95,8 @@ export interface ReadWritePoolOptions { export class ReadWritePool implements Pool { readonly primary: Pool; private readonly replicas: Map; - private readonly replicaEndpoints: ReplicaEndpoint[]; - private readonly config: DatabaseConfig; + private replicaEndpoints: ReplicaEndpoint[]; + private config: DatabaseConfig; private readonly lagState: Map = new Map(); private readonly queryStats: Map = new Map(); private readonly lagSamples: Map = new Map(); @@ -111,7 +111,7 @@ export class ReadWritePool implements Pool { ) { this.primary = primary; this.replicas = replicas; - this.replicaEndpoints = endpoints; + this.replicaEndpoints = [...endpoints]; this.config = config; for (const endpoint of endpoints) { @@ -328,6 +328,75 @@ export class ReadWritePool implements Pool { await pool.end(); } } + + /** + * Hot-swap replica pools after connection-string rotation. + * Closes previous replica pools and reseeds lag/query state. + */ + async replaceReplicaPools( + nextReplicas: Map, + endpoints: ReplicaEndpoint[], + ): Promise { + const previous = [...this.replicas.entries()]; + this.replicas.clear(); + this.replicaEndpoints.length = 0; + this.replicaEndpoints.push(...endpoints); + this.lagState.clear(); + this.queryStats.clear(); + this.lagSamples.clear(); + this.roundRobinIndex = 0; + + for (const endpoint of endpoints) { + const pool = nextReplicas.get(endpoint.name); + if (pool) { + this.replicas.set(endpoint.name, pool); + } + this.lagState.set(endpoint.name, { + name: endpoint.name, + lagMs: 0, + lagP99Ms: 0, + available: true, + lastCheckedAt: 0, + }); + this.lagSamples.set(endpoint.name, []); + this.queryStats.set(endpoint.name, { + name: endpoint.name, + queryCount: 0, + totalLatencyMs: 0, + lastLatencyMs: 0, + errors: 0, + }); + } + + this.config = { ...this.config, replicas: [...endpoints] }; + + for (const [, pool] of previous) { + try { + await pool.end(); + } catch { + // Ignore close errors during rotation + } + } + + if (this.replicas.size > 0 && !this.lagPollTimer) { + this.startLagMonitoring(); + } + if (this.replicas.size === 0) { + this.stopLagMonitoring(); + } + } + + /** Mark a replica unavailable (manual / forced failover). */ + markReplicaUnavailable(name: string): void { + const state = this.lagState.get(name); + if (state) state.available = false; + } + + /** Restore a replica to the candidate set. */ + markReplicaAvailable(name: string): void { + const state = this.lagState.get(name); + if (state) state.available = true; + } } // ── Factory ─────────────────────────────────────────────────────────────────── diff --git a/docs/database-read-replicas.md b/docs/database-read-replicas.md new file mode 100644 index 00000000..0c2e588d --- /dev/null +++ b/docs/database-read-replicas.md @@ -0,0 +1,201 @@ +# Database Read Replicas & Failover + +SubTrackr routes read-heavy SQL to PostgreSQL read replicas and writes to the +primary. Replica lag is monitored continuously; when lag or availability crosses +thresholds, reads automatically fail over to the primary. + +## Architecture + +``` +Application / DatabaseService + │ + ├── WRITE (INSERT/UPDATE/DELETE/DDL) ──► Primary + │ + └── READ (SELECT / WITH) ──► Replica round-robin + │ + ├─ lag OK ──► replica-N + └─ lag / down ──► Primary (failover) +``` + +Elasticsearch follows the same pattern when remote nodes are configured: +writes → `ES_PRIMARY_URL`, reads → `ES_READ_REPLICA_URLS`, with failover to +primary when replicas are marked failed. + +## Files + +| File | Purpose | +|---|---| +| `backend/config/database.ts` | Primary + replica endpoint config | +| `backend/shared/db/readWriteRouter.ts` | Read/write splitting + lag failover | +| `backend/shared/db/connectionStringRotation.ts` | Connection string parse / rotate | +| `backend/shared/db/queryClassifier.ts` | SELECT vs write classification | +| `backend/services/shared/databaseService.ts` | Service-layer facade | +| `backend/monitoring/replicationLagExporter.ts` | Prometheus lag metrics | +| `backend/elasticsearch/config.ts` | ES replica node configuration | +| `backend/elasticsearch/replicaRouter.ts` | ES read/write + failover routing | + +## Configuration + +### Discrete host variables + +```bash +DB_HOST=primary.internal +DB_PORT=5432 +DB_NAME=subtrackr +DB_USER=app +DB_PASSWORD=secret +DB_READ_REPLICAS=replica-1.internal:6432,replica-2.internal:6433 +DB_REPLICA_POOL_SIZE=25 +DB_REPLICATION_LAG_P99_ALARM_MS=1000 +DB_REPLICATION_LAG_FAILOVER_MS=5000 +DB_STALE_READ_DEFAULT_SECONDS=30 +DB_LAG_POLL_INTERVAL_MS=5000 +``` + +### Connection string style (supports rotation) + +```bash +DATABASE_URL=postgresql://app:secret@primary.internal:5432/subtrackr +DATABASE_READ_URLS=postgresql://app:secret@r1:5432/subtrackr,postgresql://app:secret@r2:5432/subtrackr +``` + +### Elasticsearch replicas + +```bash +ES_PRIMARY_URL=https://es-primary:9200 +ES_READ_REPLICA_URLS=https://es-r1:9200,https://es-r2:9200 +ES_READ_WRITE_SPLITTING=true +ES_AUTOMATIC_FAILOVER=true +``` + +## Read / write splitting + +`ReadWritePool` (and `DatabaseService.query`) classifies SQL: + +- **Read** — `SELECT` / non-mutating `WITH` → healthy replica (round-robin) +- **Write** — `INSERT` / `UPDATE` / `DELETE` / DDL / `SELECT … FOR UPDATE` → primary +- **Transactions** — `connect()` always uses the primary + +Response headers (when routing context is attached): + +| Header | Example | +|---|---| +| `X-DB-Route` | `replica:replica-1` or `primary` | +| `X-DB-Route-Reason` | `no-replicas` / `lag-or-unavailable` | +| `X-DB-Route-Warning` | `replication-lag-fallback-primary` | +| `X-DB-Replication-Lag-Ms` | `240` | + +Analytics may raise the lag budget via `X-Stale-Accept: `. + +## Automatic failover + +Failover triggers (reads move to primary): + +1. **Replica lag** above `DB_REPLICATION_LAG_FAILOVER_MS` (default 5s) +2. **Replica query error** — replica marked unavailable, next read uses primary +3. **Operator / service failover** — `databaseService.failoverToPrimary(reason)` + +Recovery: + +```ts +import { getDatabaseService } from '../backend/services/shared/databaseService'; + +const db = getDatabaseService(); +await db.failoverToPrimary('replica-lag-p99-alarm'); +// … investigate / repair replicas … +await db.recoverFromFailover(); +``` + +Status snapshot: + +```ts +const status = db.getFailoverStatus(); +// { mode: 'failover-primary' | 'replicas-active' | 'primary-only', +// healthyReplicas, totalReplicas, lagStates, connectionGeneration } +``` + +## Replica lag monitoring + +Background polling (`DB_LAG_POLL_INTERVAL_MS`) runs: + +```sql +SELECT COALESCE( + EXTRACT(EPOCH FROM (now() - pg_last_xact_replay_timestamp())) * 1000, + 0 +)::float AS lag_ms +``` + +Prometheus metrics (via `replicationLagExporter`): + +- `subtrackr_replication_lag_ms{replica=…}` +- `subtrackr_replication_lag_p99_ms{replica=…}` +- `subtrackr_replication_lag_p99_alarm_ms` +- `subtrackr_replication_lag_failover_ms` +- `subtrackr_replica_available{replica=…}` +- `subtrackr_replica_pool_*` / `subtrackr_replica_query_*` + +Alert when P99 lag exceeds `DB_REPLICATION_LAG_P99_ALARM_MS` (default 1s). + +## Connection string rotation + +Rotate credentials or endpoints without process restart: + +```ts +await db.rotateConnectionStrings({ + primaryUrl: 'postgresql://app:new-secret@primary:5432/subtrackr', + replicaUrls: [ + 'postgresql://app:new-secret@r1:5432/subtrackr', + 'postgresql://app:new-secret@r2:5432/subtrackr', + ], +}); +``` + +Effects: + +1. Parses and stores the new URLs (`ConnectionStringRotator`) +2. Rebuilds replica pools against the new hosts +3. Clears failure state and bumps `connectionGeneration` +4. Safe snapshot via `db.getConnectionSnapshot()` (passwords redacted) + +For Elasticsearch node URL rotation use `ElasticsearchReplicaRouter.rotateNodes()`. + +## Failover testing + +Backend Jest suite covers: + +| Test | Location | +|---|---| +| Lag-based failover to primary | `backend/shared/db/__tests__/readWriteRouter.test.ts` | +| Replica query failure fallback | same | +| Connection string parse / rotate | `backend/shared/db/__tests__/connectionStringRotation.test.ts` | +| Service failover + recovery | `backend/services/shared/__tests__/databaseService.test.ts` | +| ES replica failover + rotation | `backend/elasticsearch/__tests__/replicaRouter.test.ts` | + +Run: + +```bash +npx jest -c jest.backend.config.js \ + backend/shared/db/__tests__/ \ + backend/services/shared/__tests__/databaseService.test.ts \ + backend/elasticsearch/__tests__/replicaRouter.test.ts \ + backend/config/__tests__/database.test.ts \ + backend/monitoring/__tests__/replicationLagExporter.test.ts +``` + +Manual checklist: + +1. Start primary + two replicas; set `DB_READ_REPLICAS` / `DATABASE_READ_URLS` +2. Confirm `X-DB-Route` shows `replica:*` on a SELECT endpoint +3. Stop one replica — traffic moves to the healthy replica / primary +4. Inject lag (pause WAL replay) above 5s — confirm primary fallback header +5. Rotate `DATABASE_URL` credentials via `rotateConnectionStrings` — generation increments, queries succeed +6. Call `failoverToPrimary` then `recoverFromFailover` — mode flips correctly + +## Operational runbook + +| Symptom | Action | +|---|---| +| P99 lag alarm | Check replica apply rate / network; raise stale budget only for analytics | +| All replicas unavailable | Automatic primary failover; page on-call if sustained | +| Credential rotation | Call `rotateConnectionStrings` then verify `connectionGeneration` | +| Split-brain risk | Never promote a replica from the app layer — use your cloud/DB provider promote flow, then rotate URLs |