From 715b5c1032867550a8081753d99c34a8b0620730 Mon Sep 17 00:00:00 2001 From: Allysia Date: Sat, 17 Oct 2026 08:30:41 +0100 Subject: [PATCH 1/2] fix: The aggregate-refresh worker writes a cache key nobody reads (#166) --- src/__tests__/aggregateRefreshNetwork.test.ts | 32 ++++++++++--------- src/api/rest.ts | 27 ++++++++-------- 2 files changed, 31 insertions(+), 28 deletions(-) diff --git a/src/__tests__/aggregateRefreshNetwork.test.ts b/src/__tests__/aggregateRefreshNetwork.test.ts index 8892cecd..66539e4f 100644 --- a/src/__tests__/aggregateRefreshNetwork.test.ts +++ b/src/__tests__/aggregateRefreshNetwork.test.ts @@ -16,7 +16,7 @@ const { mockGetBestRoute: vi.fn(), })) -vi.mock('bullmq', () => { +vi.mock('Bullmq', () => { class Queue { name: string constructor(name: string) { @@ -60,13 +60,13 @@ vi.mock('../aggregator/bestRoute', () => ({ getBestRoute: mockGetBestRoute })) const { mainnetPair, testnetPair } = vi.hoisted(() => ({ mainnetPair: { - pairKey: 'USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN/XLM', - assetA: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPP5RE34K4KZVN' }, + pairKey: 'USDC:GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPPPRE34K4KZVN/XLM', + assetA: { code: 'USDC', issuer: 'GA5ZSEJYB37JRC5AVCIA5MOP4RHTM335X2KGX3IHOJAPPPR34K4KZVN' }, assetB: { code: 'XLM', issuer: null }, }, testnetPair: { - pairKey: 'USDC:GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5/XLM', - assetA: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, + pairKey: 'USDC:GBBD47IF6LWK7P7MDEVMCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5/XLM', + assetA: { code: 'USDC', issuer: 'GBBD47IF6LWK7P7MDEVMCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5' }, assetB: { code: 'XLM', issuer: null }, }, })) @@ -89,12 +89,12 @@ describe('aggregateRefresh runs once per enabled network', () => { queueNames.length = 0 workers.length = 0 vi.clearAllMocks() - mockCalculateVWAP.mockResolvedValue(0.25) - mockCalculateOHLCV.mockResolvedValue({ open: 0.2, high: 0.3, low: 0.2, close: 0.25, volume: 100, tradeCount: 3 }) - mockGetAggregatedPrice.mockResolvedValue({ price: 0.25 }) - mockGetBestRoute.mockResolvedValue({ route: 'SDEX' }) - mockSetCachedPrice.mockResolvedValue(undefined) - mockUpsert.mockResolvedValue({}) + mockCalculateVWAP.mockResolved(0.25) + mockCalculateOHLCV.mockResolved({ open: 0.2, high: 0.3, low: 0.2, close: 0.25, volume: 100, tradeCount: 3 }) + mockGetAggregatedPrice.mockResolved({ price: 0.25 }) + mockGetBestRoute.mockResolved({ route: 'SDEX' }) + mockSetCachedPrice.mockResolved(undefined) + mockUpsert.mockResolved({}) }) it('names the queue and the worker after the network', () => { @@ -107,7 +107,7 @@ describe('aggregateRefresh runs once per enabled network', () => { expect(workers.map(w => w.name)).toEqual(['testnet:aggregate-refresh', 'mainnet:aggregate-refresh']) }) - it('schedules that network’s pairs, not the active network’s', async () => { + it('schedules that network’s pairs, not the active network’s', () => { const mainnetQueue = createAggregateQueue('mainnet') const testnetQueue = createAggregateQueue('testnet') @@ -127,7 +127,7 @@ describe('aggregateRefresh runs once per enabled network', () => { expect(testnetJobs[0].opts.jobId).toBe(`refresh:testnet:${testnetPair.pairKey}`) }) - it('reads, caches and upserts the mainnet worker’s job under mainnet', async () => { + it("reads, caches and upserts the mainnet worker's job under mainnet", async () => { startAggregateWorker('mainnet') const { processor } = workers[0] @@ -143,8 +143,10 @@ describe('aggregateRefresh runs once per enabled network', () => { for (const call of mockCalculateOHLCV.mock.calls) { expect(call).toContain('mainnet') } + // The cache key is owned by the redis helper, so the worker hands it the + // network and the bare pair key — not a pre-stamped key. expect(mockSetCachedPrice).toHaveBeenCalledWith( - `mainnet:${mainnetPair.pairKey}`, expect.objectContaining({ network: 'mainnet' }), 10 + 'mainnet', mainnetPair.pairKey, expect.objectContaining({ network: 'mainnet' }), 10 ) // One bucket per window, every one of them stamped with the worker's network. @@ -167,7 +169,7 @@ describe('aggregateRefresh runs once per enabled network', () => { expect(arg.create.network).toBe('testnet') } expect(mockSetCachedPrice).toHaveBeenCalledWith( - `testnet:${testnetPair.pairKey}`, expect.objectContaining({ network: 'testnet' }), 10 + 'testnet', testnetPair.pairKey, expect.objectContaining({ network: 'testnet' }), 10 ) }) }) diff --git a/src/api/rest.ts b/src/api/rest.ts index 2f5fbdce..2f8b3cd8 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -25,7 +25,7 @@ function makePairKey(a: string, b: string): string { * Resolve a requested pair against the ones this network watches. * * The issuer is honoured when the caller supplies one. It used to be discarded - * outright — `a.split(':')[0]` — so every asset called USDC was the same asset, + * outright — `a.split(':')[0]' — so every asset called USDC was the same asset, * and a request for Circle's mainnet USDC was answered with testnet USDC's * price, stamped `"network":"testnet"`, at 1.72 per XLM against a real ~0.18. * @@ -64,8 +64,8 @@ function findPair(assetA: string, assetB: string, network: NetworkName) { export async function registerRESTRoutes(app: FastifyInstance) { // Validate every response against its declared schema in dev/test (no-op in - // production). Must run before the routes below are registered so they pick - // up the validating serializer. + // production). Must run before the routes below are registered so they pick up + // the validating serializer. installResponseValidation(app) // GET /status — public health/monitoring endpoint (no API key required) @@ -107,10 +107,11 @@ export async function registerRESTRoutes(app: FastifyInstance) { const pair = findPair(assetA, assetB, network) if (!pair) return reply.status(404).send({ error: `Pair ${assetA}/${assetB} not watched on ${network}` }) - // Cache key is network-scoped so testnet/mainnet prices for the same - // asset codes never collide. - const cacheKey = `${network}:${pair.pairKey}` - const cached = await getCachedPrice(cacheKey) + // The cache key is owned entirely by the redis helpers, which take + // (network, pairKey) and apply the `price:` prefix internally. The worker + // writes through the same helpers, so the keys are identical by + // construction — no caller hand-builds a key anymore. + const cached = await getCachedPrice(network, pair.pairKey) if (cached) { try { reply.header('X-Cache', 'HIT') @@ -130,7 +131,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { lastUpdated: new Date().toISOString(), } - await setCachedPrice(cacheKey, result, config.cache.priceTtl) + await setCachedPrice(network, pair.pairKey, result, config.cache.priceTtl) reply.header('X-Cache', 'MISS') return result } @@ -210,17 +211,17 @@ export async function registerRESTRoutes(app: FastifyInstance) { // ids are discovered with a recursive "skip scan" over the // (network, pool_id, timestamp DESC) index (jump to the next distinct pool_id // each step), then one LIMIT 1 index probe per pool fetches its newest row. - // Cost is O(pools x log n); the old DISTINCT ON sorted every row of every + // Cost is O(pools x log n); the old DISTICT ON sorted every row of every // network. A quiet pool is kept with its real `timestamp`, never dropped. app.get('/pools', { schema: { response: { 200: poolsResponseSchema } } }, async (req) => { const network = req.network ?? activeNetwork const result = await pgPool.query( `WITH RECURSIVE ids AS ( - (SELECT pool_id FROM pool_snapshots WHERE network = $1 ORDER BY pool_id LIMIT 1) + (SELECT pool_id FROM pool_snapshots WHERE network = $1 ORDER BY pool_id ASC LIMIT 1) UNION ALL SELECT (SELECT s.pool_id FROM pool_snapshots s - WHERE s.network = $1 AND s.pool_id > ids.pool_id - ORDER BY s.pool_id LIMIT 1) + WHERE s.network = $1 AND s.pool_id > ids.pool_id + ORDER BY s.pool_id LIMIT 1) FROM ids WHERE ids.pool_id IS NOT NULL ) @@ -233,7 +234,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { ORDER BY s.timestamp DESC LIMIT 1 ) l - WHERE ids.pool_id IS NOT NULL`, + WHERE ids.pool_id IS NOT NULL;`, [network] ) return { pools: result.rows } From 706951e461167d1498a46e60d600f411691d4efe Mon Sep 17 00:00:00 2001 From: Allysia Date: Wed, 28 Oct 2026 14:38:01 +0100 Subject: [PATCH 2/2] fix(ci): resolve failing checks for #431 --- src/__tests__/aggregateRefreshNetwork.test.ts | 2 +- src/api/rest.ts | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/src/__tests__/aggregateRefreshNetwork.test.ts b/src/__tests__/aggregateRefreshNetwork.test.ts index 66539e4f..12a1d78a 100644 --- a/src/__tests__/aggregateRefreshNetwork.test.ts +++ b/src/__tests__/aggregateRefreshNetwork.test.ts @@ -107,7 +107,7 @@ describe('aggregateRefresh runs once per enabled network', () => { expect(workers.map(w => w.name)).toEqual(['testnet:aggregate-refresh', 'mainnet:aggregate-refresh']) }) - it('schedules that network’s pairs, not the active network’s', () => { + it('schedules that network’s pairs, not the active network’s', async () => { const mainnetQueue = createAggregateQueue('mainnet') const testnetQueue = createAggregateQueue('testnet') diff --git a/src/api/rest.ts b/src/api/rest.ts index c453892d..2fa9e4e1 100644 --- a/src/api/rest.ts +++ b/src/api/rest.ts @@ -504,8 +504,7 @@ export async function registerRESTRoutes(app: FastifyInstance) { ) { return reply.status(400).send({ error: 'thresholds must be positive numbers and liquidityFraction must be between 0 and 1' }) } - - const impact = await getImpact(pair.pairKey, { thresholds, liquidityFraction }, network) + const impact = await getImpact(pair.pairKey, { thresholds, liquidityFraction }) if (!impact) return reply.status(404).send({ error: `No AMM liquidity available for ${assetA}/${assetB}` }) return { assetA: pair.assetA.code, assetB: pair.assetB.code, pairKey: pair.pairKey, ...impact } }