diff --git a/README.md b/README.md index 4e69aa5..d41a1f5 100644 --- a/README.md +++ b/README.md @@ -133,6 +133,9 @@ The application reads configurations from the `.env` file at the root. | `PRICE_REFRESH_INTERVAL_SECONDS` | Refresh interval in seconds | 30 | No | | `PRICE_STALE_THRESHOLD_MINUTES` | Stale threshold in minutes | 5 | No | | `PRICE_ANOMALY_THRESHOLD_PCT` | Anomaly detection threshold % | 20 | No | +| `CIRCUIT_BREAKER_FAILURE_THRESHOLD` | Source failures before opening a price-source circuit | 3 | No | +| `CIRCUIT_BREAKER_SUCCESS_THRESHOLD` | Half-open successes required to close a circuit | 1 | No | +| `CIRCUIT_BREAKER_TIMEOUT_MS` | Open-circuit cool-down before a half-open probe | 30000 | No | | `ADMIN_API_KEY` | Bootstrap admin bearer token for API key management | empty | Yes, for protected endpoints | | `LOG_LEVEL` | Logging level: `debug`, `info`, `warn`, or `error` | info | No | diff --git a/src/config.js b/src/config.js index 80f1547..5d46892 100644 --- a/src/config.js +++ b/src/config.js @@ -38,6 +38,9 @@ const env = cleanEnv(rawEnv, { PRICE_REFRESH_INTERVAL_SECONDS: num({ default: 30 }), PRICE_STALE_THRESHOLD_MINUTES: num({ default: 5 }), PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }), + CIRCUIT_BREAKER_FAILURE_THRESHOLD: num({ default: 3 }), + CIRCUIT_BREAKER_SUCCESS_THRESHOLD: num({ default: 1 }), + CIRCUIT_BREAKER_TIMEOUT_MS: num({ default: 30000 }), LOG_LEVEL: str({ default: 'info', choices: ['debug', 'info', 'warn', 'error'], @@ -74,6 +77,11 @@ module.exports = { refreshInterval: env.PRICE_REFRESH_INTERVAL_SECONDS, staleThresholdMinutes: env.PRICE_STALE_THRESHOLD_MINUTES, anomalyThresholdPercent: env.PRICE_ANOMALY_THRESHOLD_PCT, + circuitBreaker: { + failureThreshold: env.CIRCUIT_BREAKER_FAILURE_THRESHOLD, + successThreshold: env.CIRCUIT_BREAKER_SUCCESS_THRESHOLD, + timeoutMs: env.CIRCUIT_BREAKER_TIMEOUT_MS, + }, }, auth: { adminApiKey: env.ADMIN_API_KEY, diff --git a/src/index.js b/src/index.js index f7a5edf..3f05fde 100644 --- a/src/index.js +++ b/src/index.js @@ -5,6 +5,7 @@ const helmet = require('helmet'); const config = require('./config'); const logger = require('./logger'); const cache = require('./services/cache'); +const priceOracle = require('./services/priceOracle'); const priceRefreshJob = require('./jobs/priceRefresh'); const webhookRetryWorker = require('./jobs/webhookRetryWorker'); const buildCorsMiddleware = require('./middleware/cors'); @@ -35,6 +36,7 @@ app.get('/health', (req, res) => { timestamp: new Date().toISOString(), redis_connected: redisConnected, redis_unavailable: !redisConnected, + circuits: priceOracle.getCircuitStates(), }); }); diff --git a/src/services/priceOracle.js b/src/services/priceOracle.js index 7d03f27..b73cdda 100644 --- a/src/services/priceOracle.js +++ b/src/services/priceOracle.js @@ -4,13 +4,27 @@ const coingecko = require('./sources/coingecko'); const coinmarketcap = require('./sources/coinmarketcap'); const config = require('../config'); const logger = require('../logger'); +const { CircuitBreaker } = require('../utils/circuitBreaker'); const CACHE_PREFIX = 'price:'; const HISTORY_PREFIX = 'price:history:'; +const breakerOptions = config.price.circuitBreaker; const SOURCES = [ - { name: 'stellar_dex', fetch: stellarDex.fetchPrice }, - { name: 'coingecko', fetch: coingecko.fetchPrice }, - { name: 'coinmarketcap', fetch: coinmarketcap.fetchPrice }, + { + name: 'stellar_dex', + fetch: stellarDex.fetchPrice, + breaker: new CircuitBreaker('stellar_dex', breakerOptions), + }, + { + name: 'coingecko', + fetch: coingecko.fetchPrice, + breaker: new CircuitBreaker('coingecko', breakerOptions), + }, + { + name: 'coinmarketcap', + fetch: coinmarketcap.fetchPrice, + breaker: new CircuitBreaker('coinmarketcap', breakerOptions), + }, ]; function median(values) { @@ -79,7 +93,7 @@ async function fetchFromAllSources(assetCode, issuer) { for (const source of SOURCES) { try { - const price = await source.fetch(assetCode, issuer); + const price = await source.breaker.call(() => source.fetch(assetCode, issuer)); if (price !== null && price > 0) { results.push({ source: source.name, price }); } @@ -91,6 +105,19 @@ async function fetchFromAllSources(assetCode, issuer) { return results; } +function getCircuitStates() { + return SOURCES.reduce((states, source) => { + states[source.name] = source.breaker.getState(); + return states; + }, {}); +} + +function resetCircuitBreakers() { + for (const source of SOURCES) { + source.breaker.reset(); + } +} + async function getPrice(assetCode, issuer = null) { const cacheKey = buildCacheKey(assetCode, issuer); let redisUnavailable = false; @@ -126,7 +153,7 @@ async function getPrice(assetCode, issuer = null) { async function fetchFreshPrice(assetCode, issuer = null, redisUnavailable = false) { const sourceResults = await fetchFromAllSources(assetCode, issuer); - const sourcesAttempted = sourceResults.map((r) => r.name); + const sourcesAttempted = sourceResults.map((r) => r.source); const prices = sourceResults.map((r) => r.price); const aggregatedPrice = median(prices); @@ -235,5 +262,7 @@ async function refreshAllCachedPrices() { module.exports = { getPrice, fetchFreshPrice, + getCircuitStates, + resetCircuitBreakers, refreshAllCachedPrices, }; diff --git a/src/utils/circuitBreaker.js b/src/utils/circuitBreaker.js new file mode 100644 index 0000000..03fef49 --- /dev/null +++ b/src/utils/circuitBreaker.js @@ -0,0 +1,143 @@ +'use strict'; + +const logger = require('../logger'); + +const STATES = Object.freeze({ + CLOSED: 'closed', + OPEN: 'open', + HALF_OPEN: 'half-open', +}); + +class CircuitBreaker { + constructor(name, options = {}) { + this.name = name; + this.failureThreshold = Math.max(1, options.failureThreshold ?? 3); + this.successThreshold = Math.max(1, options.successThreshold ?? 1); + this.timeoutMs = Math.max(1, options.timeoutMs ?? 30000); + this._now = options.now || Date.now; + this._logger = options.logger || logger; + + this.state = STATES.CLOSED; + this.failureCount = 0; + this.successCount = 0; + this.openedAt = null; + this.halfOpenInFlight = false; + } + + getState() { + this._moveToHalfOpenIfReady(); + return this.state; + } + + isOpen() { + return this.getState() === STATES.OPEN; + } + + async call(fn) { + this._moveToHalfOpenIfReady(); + + if (this.state === STATES.OPEN) { + this._logger.info('Circuit breaker open, skipping source call', { + source: this.name, + state: this.state, + }); + return null; + } + + if (this.state === STATES.HALF_OPEN && this.halfOpenInFlight) { + this._logger.info('Circuit breaker half-open probe already in flight, skipping source call', { + source: this.name, + state: this.state, + }); + return null; + } + + const probing = this.state === STATES.HALF_OPEN; + if (probing) { + this.halfOpenInFlight = true; + } + + try { + const result = await fn(); + if (result === null || result === undefined) { + this.recordFailure(); + } else { + this.recordSuccess(); + } + return result ?? null; + } catch (err) { + this.recordFailure(); + throw err; + } finally { + if (probing) { + this.halfOpenInFlight = false; + } + } + } + + recordSuccess() { + if (this.state === STATES.HALF_OPEN) { + this.successCount += 1; + if (this.successCount >= this.successThreshold) { + this._transitionTo(STATES.CLOSED, { reason: 'success-threshold' }); + } + return; + } + + if (this.state === STATES.CLOSED) { + this.failureCount = 0; + } + } + + recordFailure() { + if (this.state === STATES.HALF_OPEN) { + this._transitionTo(STATES.OPEN, { reason: 'half-open-failure' }); + return; + } + + if (this.state === STATES.CLOSED) { + this.failureCount += 1; + if (this.failureCount >= this.failureThreshold) { + this._transitionTo(STATES.OPEN, { reason: 'failure-threshold' }); + } + } + } + + reset() { + this._transitionTo(STATES.CLOSED, { reason: 'manual-reset' }); + } + + _moveToHalfOpenIfReady() { + if (this.state !== STATES.OPEN || this.openedAt === null) { + return; + } + + if (this._now() - this.openedAt >= this.timeoutMs) { + this._transitionTo(STATES.HALF_OPEN, { reason: 'cooldown-elapsed' }); + } + } + + _transitionTo(nextState, metadata = {}) { + if (this.state === nextState) { + return; + } + + const previousState = this.state; + this.state = nextState; + this.failureCount = 0; + this.successCount = 0; + this.openedAt = nextState === STATES.OPEN ? this._now() : null; + + this._logger.info('Circuit breaker state changed', { + source: this.name, + from: previousState, + to: nextState, + ...metadata, + }); + } +} + +module.exports = { + CircuitBreaker, + STATES, +}; diff --git a/test/circuitBreaker.test.js b/test/circuitBreaker.test.js new file mode 100644 index 0000000..e7f336f --- /dev/null +++ b/test/circuitBreaker.test.js @@ -0,0 +1,89 @@ +'use strict'; + +const { CircuitBreaker, STATES } = require('../src/utils/circuitBreaker'); + +function buildBreaker(options = {}) { + let now = 1000; + const logger = { + info: jest.fn(), + }; + + const breaker = new CircuitBreaker('coingecko', { + failureThreshold: 2, + successThreshold: 1, + timeoutMs: 100, + now: () => now, + logger, + ...options, + }); + + return { + breaker, + logger, + advance(ms) { + now += ms; + }, + }; +} + +describe('CircuitBreaker', () => { + test('opens after repeated failures and skips calls while cooling down', async () => { + const { breaker, logger } = buildBreaker(); + + await expect(breaker.call(async () => null)).resolves.toBeNull(); + await expect(breaker.call(async () => null)).resolves.toBeNull(); + + expect(breaker.getState()).toBe(STATES.OPEN); + + const sourceFetch = jest.fn(async () => 0.12); + await expect(breaker.call(sourceFetch)).resolves.toBeNull(); + + expect(sourceFetch).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith( + 'Circuit breaker state changed', + expect.objectContaining({ + source: 'coingecko', + from: STATES.CLOSED, + to: STATES.OPEN, + reason: 'failure-threshold', + }) + ); + }); + + test('moves to half-open after cooldown and closes on a successful probe', async () => { + const { breaker, advance } = buildBreaker(); + + await breaker.call(async () => null); + await breaker.call(async () => null); + + advance(100); + expect(breaker.getState()).toBe(STATES.HALF_OPEN); + + await expect(breaker.call(async () => 0.12)).resolves.toBe(0.12); + + expect(breaker.getState()).toBe(STATES.CLOSED); + }); + + test('reopens when the half-open probe fails', async () => { + const { breaker, advance } = buildBreaker(); + + await breaker.call(async () => null); + await breaker.call(async () => null); + + advance(100); + await expect(breaker.call(async () => null)).resolves.toBeNull(); + + expect(breaker.getState()).toBe(STATES.OPEN); + }); + + test('records thrown source errors as failures and rethrows them', async () => { + const { breaker } = buildBreaker(); + const error = new Error('rate limited'); + + await expect(breaker.call(async () => { + throw error; + })).rejects.toThrow('rate limited'); + + expect(breaker.getState()).toBe(STATES.CLOSED); + }); +}); diff --git a/test/config.test.js b/test/config.test.js index ee0c74a..170eaec 100644 --- a/test/config.test.js +++ b/test/config.test.js @@ -69,6 +69,11 @@ describe('configuration validation', () => { refreshInterval: 30, staleThresholdMinutes: 5, anomalyThresholdPercent: 20, + circuitBreaker: { + failureThreshold: 3, + successThreshold: 1, + timeoutMs: 30000, + }, }, }); }); diff --git a/test/health.test.js b/test/health.test.js new file mode 100644 index 0000000..cac0bf3 --- /dev/null +++ b/test/health.test.js @@ -0,0 +1,47 @@ +'use strict'; + +const request = require('supertest'); + +jest.mock('../src/services/cache', () => ({ + isConnected: jest.fn(() => false), + disconnect: jest.fn(), +})); + +jest.mock('../src/services/priceOracle', () => ({ + getCircuitStates: jest.fn(() => ({ + coingecko: 'closed', + coinmarketcap: 'open', + stellar_dex: 'half-open', + })), + refreshAllCachedPrices: jest.fn(), +})); + +jest.mock('../src/jobs/priceRefresh', () => ({ + start: jest.fn(), + stop: jest.fn(), +})); + +jest.mock('../src/jobs/webhookRetryWorker', () => ({ + start: jest.fn(), + stop: jest.fn(), +})); + +jest.mock('../src/ws/priceWebSocket', () => ({ + attach: jest.fn(), +})); + +describe('health endpoint', () => { + test('exposes price source circuit states', async () => { + jest.resetModules(); + const { app } = require('../src/index'); + + const res = await request(app).get('/health'); + + expect(res.status).toBe(200); + expect(res.body.circuits).toEqual({ + coingecko: 'closed', + coinmarketcap: 'open', + stellar_dex: 'half-open', + }); + }); +}); diff --git a/test/resilience.test.js b/test/resilience.test.js index a16d318..d1cb442 100644 --- a/test/resilience.test.js +++ b/test/resilience.test.js @@ -37,6 +37,8 @@ beforeEach(() => { mockStellarFetch.mockReset(); mockCoingeckoFetch.mockReset(); mockCmcFetch.mockReset(); + priceOracle.resetCircuitBreakers(); + logger.info.mockClear(); logger.warn.mockClear(); logger.error.mockClear(); @@ -156,6 +158,35 @@ describe('all sources unavailable during Redis outage', () => { }); }); +describe('price source circuit breakers', () => { + test('opens after repeated source failures and skips the failing source', async () => { + mockCacheGet.mockResolvedValue(null); + mockCacheSet.mockResolvedValue(undefined); + mockStellarFetch.mockResolvedValue(null); + mockCoingeckoFetch.mockResolvedValue(null); + mockCmcFetch.mockResolvedValue(null); + + await priceOracle.fetchFreshPrice('XLM'); + await priceOracle.fetchFreshPrice('XLM'); + await priceOracle.fetchFreshPrice('XLM'); + + expect(priceOracle.getCircuitStates()).toMatchObject({ + stellar_dex: 'open', + coingecko: 'open', + coinmarketcap: 'open', + }); + + mockStellarFetch.mockClear(); + await priceOracle.fetchFreshPrice('XLM'); + + expect(mockStellarFetch).not.toHaveBeenCalled(); + expect(logger.info).toHaveBeenCalledWith( + 'Circuit breaker open, skipping source call', + expect.objectContaining({ source: 'stellar_dex', state: 'open' }) + ); + }); +}); + describe('refreshAllCachedPrices when Redis is down', () => { test('skips refresh cycle when isConnected returns false', async () => { mockIsConnected.mockReturnValue(false);