Skip to content

Commit e742ca7

Browse files
prodbycorneprodbycorne
authored
Merge pull request #64 from vku2018/grantfox-11-price-source-circuit-breakers
[codex] Add price source circuit breakers
2 parents f8b7e74 + e3ffd07 commit e742ca7

9 files changed

Lines changed: 348 additions & 1 deletion

File tree

‎README.md‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -155,6 +155,9 @@ The application reads configurations from the `.env` file at the root.
155155
| `PRICE_REFRESH_INTERVAL_SECONDS` | Refresh interval in seconds | 30 | No |
156156
| `PRICE_STALE_THRESHOLD_MINUTES` | Stale threshold in minutes | 5 | No |
157157
| `PRICE_ANOMALY_THRESHOLD_PCT` | Anomaly detection threshold % | 20 | No |
158+
| `CIRCUIT_BREAKER_FAILURE_THRESHOLD` | Source failures before opening a price-source circuit | 3 | No |
159+
| `CIRCUIT_BREAKER_SUCCESS_THRESHOLD` | Half-open successes required to close a circuit | 1 | No |
160+
| `CIRCUIT_BREAKER_TIMEOUT_MS` | Open-circuit cool-down before a half-open probe | 30000 | No |
158161
| `ADMIN_API_KEY` | Bootstrap admin bearer token for API key management | empty | Yes, for protected endpoints |
159162
| `AIRDROP_CSV_MAX_BYTES` | Maximum recipient CSV upload size in bytes | 5242880 (5 MiB) | No |
160163
| `AIRDROP_JSON_MAX_BYTES` | Maximum JSON request body size; 2 MiB accommodates 10,000 inline recipients | 2097152 (2 MiB) | No |

‎src/config.js‎

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,9 @@ const env = cleanEnv(rawEnv, {
8484
PRICE_REFRESH_INTERVAL_SECONDS: num({ default: 30 }),
8585
PRICE_STALE_THRESHOLD_MINUTES: num({ default: 5 }),
8686
PRICE_ANOMALY_THRESHOLD_PCT: num({ default: 20 }),
87+
CIRCUIT_BREAKER_FAILURE_THRESHOLD: num({ default: 3 }),
88+
CIRCUIT_BREAKER_SUCCESS_THRESHOLD: num({ default: 1 }),
89+
CIRCUIT_BREAKER_TIMEOUT_MS: num({ default: 30000 }),
8790
PRICE_SOURCE_CIRCUIT_COOLDOWN_MS: num({ default: 15 * 60 * 1000 }),
8891
PRICE_SOURCE_CIRCUIT_REMINDER_MS: num({ default: 5 * 60 * 1000 }),
8992
AIRDROP_EXPIRY_CHECK_INTERVAL_SECONDS: num({ default: 60 }),
@@ -129,6 +132,11 @@ module.exports = {
129132
refreshInterval: env.PRICE_REFRESH_INTERVAL_SECONDS,
130133
staleThresholdMinutes: env.PRICE_STALE_THRESHOLD_MINUTES,
131134
anomalyThresholdPercent: env.PRICE_ANOMALY_THRESHOLD_PCT,
135+
circuitBreaker: {
136+
failureThreshold: env.CIRCUIT_BREAKER_FAILURE_THRESHOLD,
137+
successThreshold: env.CIRCUIT_BREAKER_SUCCESS_THRESHOLD,
138+
timeoutMs: env.CIRCUIT_BREAKER_TIMEOUT_MS,
139+
},
132140
},
133141
priceSources: {
134142
// How long a source's circuit stays open after a nonRetryable (e.g. 401)

‎src/index.js‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -59,6 +59,9 @@ app.get('/health', (req, res) => {
5959
res.json({
6060
status,
6161
timestamp: new Date().toISOString(),
62+
redis_connected: redisConnected,
63+
redis_unavailable: !redisConnected,
64+
circuits: priceOracle.getCircuitStates(),
6265
redis: {
6366
connected: redisConnected,
6467
},

‎src/services/priceOracle.js‎

Lines changed: 33 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,10 +4,27 @@ const coingecko = require('./sources/coingecko');
44
const coinmarketcap = require('./sources/coinmarketcap');
55
const config = require('../config');
66
const logger = require('../logger');
7+
const { CircuitBreaker } = require('../utils/circuitBreaker');
78

89
const CACHE_PREFIX = 'price:';
910
const HISTORY_PREFIX = 'price:history:';
11+
const breakerOptions = config.price.circuitBreaker;
1012
const SOURCES = [
13+
{
14+
name: 'stellar_dex',
15+
fetch: stellarDex.fetchPrice,
16+
breaker: new CircuitBreaker('stellar_dex', breakerOptions),
17+
},
18+
{
19+
name: 'coingecko',
20+
fetch: coingecko.fetchPrice,
21+
breaker: new CircuitBreaker('coingecko', breakerOptions),
22+
},
23+
{
24+
name: 'coinmarketcap',
25+
fetch: coinmarketcap.fetchPrice,
26+
breaker: new CircuitBreaker('coinmarketcap', breakerOptions),
27+
},
1128
{ name: 'stellar_dex', fetch: stellarDex.fetchPrice },
1229
{ name: 'coingecko', fetch: coingecko.fetchPrice, getCircuitState: coingecko.getCircuitState },
1330
{ name: 'coinmarketcap', fetch: coinmarketcap.fetchPrice, getCircuitState: coinmarketcap.getCircuitState },
@@ -91,7 +108,7 @@ async function fetchFromAllSources(assetCode, issuer) {
91108

92109
for (const source of SOURCES) {
93110
try {
94-
const price = await source.fetch(assetCode, issuer);
111+
const price = await source.breaker.call(() => source.fetch(assetCode, issuer));
95112
if (price !== null && price > 0) {
96113
results.push({ source: source.name, price });
97114
}
@@ -103,6 +120,19 @@ async function fetchFromAllSources(assetCode, issuer) {
103120
return results;
104121
}
105122

123+
function getCircuitStates() {
124+
return SOURCES.reduce((states, source) => {
125+
states[source.name] = source.breaker.getState();
126+
return states;
127+
}, {});
128+
}
129+
130+
function resetCircuitBreakers() {
131+
for (const source of SOURCES) {
132+
source.breaker.reset();
133+
}
134+
}
135+
106136
async function getPrice(assetCode, issuer = null) {
107137
const cacheKey = buildCacheKey(assetCode, issuer);
108138
let redisUnavailable = false;
@@ -247,6 +277,8 @@ async function refreshAllCachedPrices() {
247277
module.exports = {
248278
getPrice,
249279
fetchFreshPrice,
280+
getCircuitStates,
281+
resetCircuitBreakers,
250282
refreshAllCachedPrices,
251283
// Internal helpers exported for unit testing.
252284
median,

‎src/utils/circuitBreaker.js‎

Lines changed: 143 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,143 @@
1+
'use strict';
2+
3+
const logger = require('../logger');
4+
5+
const STATES = Object.freeze({
6+
CLOSED: 'closed',
7+
OPEN: 'open',
8+
HALF_OPEN: 'half-open',
9+
});
10+
11+
class CircuitBreaker {
12+
constructor(name, options = {}) {
13+
this.name = name;
14+
this.failureThreshold = Math.max(1, options.failureThreshold ?? 3);
15+
this.successThreshold = Math.max(1, options.successThreshold ?? 1);
16+
this.timeoutMs = Math.max(1, options.timeoutMs ?? 30000);
17+
this._now = options.now || Date.now;
18+
this._logger = options.logger || logger;
19+
20+
this.state = STATES.CLOSED;
21+
this.failureCount = 0;
22+
this.successCount = 0;
23+
this.openedAt = null;
24+
this.halfOpenInFlight = false;
25+
}
26+
27+
getState() {
28+
this._moveToHalfOpenIfReady();
29+
return this.state;
30+
}
31+
32+
isOpen() {
33+
return this.getState() === STATES.OPEN;
34+
}
35+
36+
async call(fn) {
37+
this._moveToHalfOpenIfReady();
38+
39+
if (this.state === STATES.OPEN) {
40+
this._logger.info('Circuit breaker open, skipping source call', {
41+
source: this.name,
42+
state: this.state,
43+
});
44+
return null;
45+
}
46+
47+
if (this.state === STATES.HALF_OPEN && this.halfOpenInFlight) {
48+
this._logger.info('Circuit breaker half-open probe already in flight, skipping source call', {
49+
source: this.name,
50+
state: this.state,
51+
});
52+
return null;
53+
}
54+
55+
const probing = this.state === STATES.HALF_OPEN;
56+
if (probing) {
57+
this.halfOpenInFlight = true;
58+
}
59+
60+
try {
61+
const result = await fn();
62+
if (result === null || result === undefined) {
63+
this.recordFailure();
64+
} else {
65+
this.recordSuccess();
66+
}
67+
return result ?? null;
68+
} catch (err) {
69+
this.recordFailure();
70+
throw err;
71+
} finally {
72+
if (probing) {
73+
this.halfOpenInFlight = false;
74+
}
75+
}
76+
}
77+
78+
recordSuccess() {
79+
if (this.state === STATES.HALF_OPEN) {
80+
this.successCount += 1;
81+
if (this.successCount >= this.successThreshold) {
82+
this._transitionTo(STATES.CLOSED, { reason: 'success-threshold' });
83+
}
84+
return;
85+
}
86+
87+
if (this.state === STATES.CLOSED) {
88+
this.failureCount = 0;
89+
}
90+
}
91+
92+
recordFailure() {
93+
if (this.state === STATES.HALF_OPEN) {
94+
this._transitionTo(STATES.OPEN, { reason: 'half-open-failure' });
95+
return;
96+
}
97+
98+
if (this.state === STATES.CLOSED) {
99+
this.failureCount += 1;
100+
if (this.failureCount >= this.failureThreshold) {
101+
this._transitionTo(STATES.OPEN, { reason: 'failure-threshold' });
102+
}
103+
}
104+
}
105+
106+
reset() {
107+
this._transitionTo(STATES.CLOSED, { reason: 'manual-reset' });
108+
}
109+
110+
_moveToHalfOpenIfReady() {
111+
if (this.state !== STATES.OPEN || this.openedAt === null) {
112+
return;
113+
}
114+
115+
if (this._now() - this.openedAt >= this.timeoutMs) {
116+
this._transitionTo(STATES.HALF_OPEN, { reason: 'cooldown-elapsed' });
117+
}
118+
}
119+
120+
_transitionTo(nextState, metadata = {}) {
121+
if (this.state === nextState) {
122+
return;
123+
}
124+
125+
const previousState = this.state;
126+
this.state = nextState;
127+
this.failureCount = 0;
128+
this.successCount = 0;
129+
this.openedAt = nextState === STATES.OPEN ? this._now() : null;
130+
131+
this._logger.info('Circuit breaker state changed', {
132+
source: this.name,
133+
from: previousState,
134+
to: nextState,
135+
...metadata,
136+
});
137+
}
138+
}
139+
140+
module.exports = {
141+
CircuitBreaker,
142+
STATES,
143+
};

‎test/circuitBreaker.test.js‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,90 @@
11
'use strict';
22

3+
const { CircuitBreaker, STATES } = require('../src/utils/circuitBreaker');
4+
5+
function buildBreaker(options = {}) {
6+
let now = 1000;
7+
const logger = {
8+
info: jest.fn(),
9+
};
10+
11+
const breaker = new CircuitBreaker('coingecko', {
12+
failureThreshold: 2,
13+
successThreshold: 1,
14+
timeoutMs: 100,
15+
now: () => now,
16+
logger,
17+
...options,
18+
});
19+
20+
return {
21+
breaker,
22+
logger,
23+
advance(ms) {
24+
now += ms;
25+
},
26+
};
27+
}
28+
29+
describe('CircuitBreaker', () => {
30+
test('opens after repeated failures and skips calls while cooling down', async () => {
31+
const { breaker, logger } = buildBreaker();
32+
33+
await expect(breaker.call(async () => null)).resolves.toBeNull();
34+
await expect(breaker.call(async () => null)).resolves.toBeNull();
35+
36+
expect(breaker.getState()).toBe(STATES.OPEN);
37+
38+
const sourceFetch = jest.fn(async () => 0.12);
39+
await expect(breaker.call(sourceFetch)).resolves.toBeNull();
40+
41+
expect(sourceFetch).not.toHaveBeenCalled();
42+
expect(logger.info).toHaveBeenCalledWith(
43+
'Circuit breaker state changed',
44+
expect.objectContaining({
45+
source: 'coingecko',
46+
from: STATES.CLOSED,
47+
to: STATES.OPEN,
48+
reason: 'failure-threshold',
49+
})
50+
);
51+
});
52+
53+
test('moves to half-open after cooldown and closes on a successful probe', async () => {
54+
const { breaker, advance } = buildBreaker();
55+
56+
await breaker.call(async () => null);
57+
await breaker.call(async () => null);
58+
59+
advance(100);
60+
expect(breaker.getState()).toBe(STATES.HALF_OPEN);
61+
62+
await expect(breaker.call(async () => 0.12)).resolves.toBe(0.12);
63+
64+
expect(breaker.getState()).toBe(STATES.CLOSED);
65+
});
66+
67+
test('reopens when the half-open probe fails', async () => {
68+
const { breaker, advance } = buildBreaker();
69+
70+
await breaker.call(async () => null);
71+
await breaker.call(async () => null);
72+
73+
advance(100);
74+
await expect(breaker.call(async () => null)).resolves.toBeNull();
75+
76+
expect(breaker.getState()).toBe(STATES.OPEN);
77+
});
78+
79+
test('records thrown source errors as failures and rethrows them', async () => {
80+
const { breaker } = buildBreaker();
81+
const error = new Error('rate limited');
82+
83+
await expect(breaker.call(async () => {
84+
throw error;
85+
})).rejects.toThrow('rate limited');
86+
87+
expect(breaker.getState()).toBe(STATES.CLOSED);
388
const mockLogger = {
489
info: jest.fn(),
590
warn: jest.fn(),

‎test/config.test.js‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -71,6 +71,11 @@ describe('configuration validation', () => {
7171
refreshInterval: 30,
7272
staleThresholdMinutes: 5,
7373
anomalyThresholdPercent: 20,
74+
circuitBreaker: {
75+
failureThreshold: 3,
76+
successThreshold: 1,
77+
timeoutMs: 30000,
78+
},
7479
},
7580
watchedAssets: [],
7681
airdrops: {

0 commit comments

Comments
 (0)