diff --git a/.changeset/exchange-rate-warm-up-and-failure-cooldown.md b/.changeset/exchange-rate-warm-up-and-failure-cooldown.md new file mode 100644 index 000000000..1828a6c18 --- /dev/null +++ b/.changeset/exchange-rate-warm-up-and-failure-cooldown.md @@ -0,0 +1,20 @@ +--- +'@openora/core': minor +--- + +The exchange-rate module now keeps its configured currencies warm instead of leaving a +hot-path caller (the RG limit gate, rank accrual, a social transfer, or anything else +converting inside a locked transaction) to hit the vendor synchronously once a rate goes +hard-stale. A new schedule refreshes every crypto currency the operator lists plus every +displayed fiat currency roughly every 30 seconds, so a request only ever reads a fresh row +from the cache. + +A failed vendor call previously cooled down for exactly `providerTimeoutMs`, so a vendor +outage added a full timeout stall to every call for as long as it lasted. The cooldown is +now its own `exchangeRate.failureCooldownMs` config knob, defaulting to 30 seconds - +independent of and longer than the timeout, so an outage fails fast from the last-known +row instead of waiting out the vendor on every request. + +`useExchangeRate`/`useExchangeRates` now set `staleTime: 60_000` and +`refetchOnWindowFocus: false`, matching the reader's own fresh window - a remount or a tab +refocus inside that window no longer re-fetches a rate that hasn't changed server-side. diff --git a/packages/core/src/contracts/schemas/platform-config.ts b/packages/core/src/contracts/schemas/platform-config.ts index fa8c0dfc7..482f29d93 100644 --- a/packages/core/src/contracts/schemas/platform-config.ts +++ b/packages/core/src/contracts/schemas/platform-config.ts @@ -161,6 +161,14 @@ export const ExchangeRateConfigSchema = z .positive() .default(15 * 60_000), providerTimeoutMs: z.number().int().positive().default(2_000), + /** + * How long a currency stays refused after a failed vendor call, before another + * caller is allowed to retry it. Deliberately independent of `providerTimeoutMs` + * (default well above it) - otherwise a vendor outage adds a full timeout stall to + * every hot-path call for as long as the outage lasts, instead of failing fast from + * cache. + */ + failureCooldownMs: z.number().int().positive().default(30_000), }) .strict(); diff --git a/packages/core/src/fx/exchange-rate/__tests__/exchange-rate-reader.service.int.test.ts b/packages/core/src/fx/exchange-rate/__tests__/exchange-rate-reader.service.int.test.ts index a11570381..3e9de6d95 100644 --- a/packages/core/src/fx/exchange-rate/__tests__/exchange-rate-reader.service.int.test.ts +++ b/packages/core/src/fx/exchange-rate/__tests__/exchange-rate-reader.service.int.test.ts @@ -88,6 +88,7 @@ function baseDeps( freshTtlMs: 60_000, hardMaxAgeMs: 120_000, providerTimeoutMs: 150, + failureCooldownMs: 150, ...over, }; } @@ -252,6 +253,41 @@ describe('ExchangeRateReaderService.getRate - age bands', () => { }); }); +describe('ExchangeRateReaderService.getRate - failure cooldown', () => { + function failingProvider() { + const getRate = vi.fn(async (): Promise => { + throw new Error('vendor unreachable'); + }); + return { provider: mock({ getRate }), getRate }; + } + + it('refuses a failed currency for failureCooldownMs, not providerTimeoutMs, without calling the vendor again', async () => { + const { provider, getRate } = failingProvider(); + const reader = new ExchangeRateReaderService( + baseDeps({ fiatProvider: provider, providerTimeoutMs: 10, failureCooldownMs: 60_000 }), + ); + + expect(await reader.getRate('EUR', 'USD')).toBeNull(); + await wait(50); + expect(await reader.getRate('EUR', 'USD')).toBeNull(); + + expect(getRate).toHaveBeenCalledTimes(1); + }); + + it('calls the vendor again once the cooldown has passed', async () => { + const { provider, getRate } = failingProvider(); + const reader = new ExchangeRateReaderService( + baseDeps({ fiatProvider: provider, failureCooldownMs: 30 }), + ); + + await reader.getRate('EUR', 'USD'); + await wait(80); + await reader.getRate('EUR', 'USD'); + + expect(getRate).toHaveBeenCalledTimes(2); + }); +}); + describe('ExchangeRateReaderService.getRate - single-flight', () => { it('collapses concurrent hard-stale callers for the same leg into one provider call', async () => { const providerAsOf = agedIso(0); diff --git a/packages/core/src/fx/exchange-rate/adapters/exchange-rate-reader.service.ts b/packages/core/src/fx/exchange-rate/adapters/exchange-rate-reader.service.ts index d3963ebf7..120e63248 100644 --- a/packages/core/src/fx/exchange-rate/adapters/exchange-rate-reader.service.ts +++ b/packages/core/src/fx/exchange-rate/adapters/exchange-rate-reader.service.ts @@ -53,6 +53,8 @@ export type ExchangeRateReaderServiceDeps = { freshTtlMs: number; hardMaxAgeMs: number; providerTimeoutMs: number; + /** See `ExchangeRateConfigSchema.failureCooldownMs`. */ + failureCooldownMs: number; }; /** @@ -81,6 +83,7 @@ export class ExchangeRateReaderService implements ExchangeRateReader { private readonly freshTtlMs: number; private readonly hardMaxAgeMs: number; private readonly providerTimeoutMs: number; + private readonly failureCooldownMs: number; private readonly inFlight = new Map>(); private readonly legResolution = new Map>(); private readonly failedUntil = new Map(); @@ -94,6 +97,7 @@ export class ExchangeRateReaderService implements ExchangeRateReader { this.freshTtlMs = deps.freshTtlMs; this.hardMaxAgeMs = deps.hardMaxAgeMs; this.providerTimeoutMs = deps.providerTimeoutMs; + this.failureCooldownMs = deps.failureCooldownMs; } async getRate(from: string, to: string): Promise { @@ -199,7 +203,7 @@ export class ExchangeRateReaderService implements ExchangeRateReader { } this.failedUntil.delete(oldest); } - this.failedUntil.set(currency, now + this.providerTimeoutMs); + this.failedUntil.set(currency, now + this.failureCooldownMs); } private async readRow(currency: string): Promise<{ rate: string; providerAsOf: Date } | null> { diff --git a/packages/core/src/fx/exchange-rate/plugin.ts b/packages/core/src/fx/exchange-rate/plugin.ts index ae8224b24..40d9a28c5 100644 --- a/packages/core/src/fx/exchange-rate/plugin.ts +++ b/packages/core/src/fx/exchange-rate/plugin.ts @@ -1,28 +1,50 @@ -import { DRIZZLE } from '@openora/core/server'; +import { DRIZZLE, createLogger, mapConcurrent } from '@openora/core/server'; import type { CoreTokenCatalog, Plugin, TypedContainer } from '@openora/core/server'; import { CRYPTO_EXCHANGE_RATE_PROVIDER, FIAT_EXCHANGE_RATE_PROVIDER, EXCHANGE_RATE_READER, PLATFORM_CONFIG, + JOB_QUEUE, + DEFAULT_CRYPTO_CURRENCIES, resolveDisplayCurrencies, resolveExchangeRatePivot, + queue, } from '@openora/core/contracts'; +import * as z from 'zod'; import { ExchangeRateService } from './service/exchange-rate.service.js'; import { createExchangeRateRouter } from './router/index.js'; import { ExchangeRateReaderService } from './adapters/exchange-rate-reader.service.js'; +const logger = createLogger('exchange-rate'); + const DEFAULT_FRESH_TTL_MS = 60_000; const DEFAULT_HARD_MAX_AGE_MS = 15 * 60_000; const DEFAULT_PROVIDER_TIMEOUT_MS = 2_000; +const DEFAULT_FAILURE_COOLDOWN_MS = 30_000; + +const RATE_WARM_QUEUE = queue('exchange-rate-warm'); +const RateWarmJobSchema = z.object({}); +// Well inside freshTtlMs (60s default) so a warmed coin never crosses into soft-stale +// between ticks, and its hot-path readers never fall through to a synchronous vendor call. +const WARM_INTERVAL_MS = 30_000; +const WARM_CONCURRENCY = 4; export default { id: 'exchange-rate', register(ctx) { + // Set inside the router factory below (needs container access) and read by the warm + // job worker, same lazily-constructed-singleton shape as wallet's sweep/reconciliation + // services - the worker registers before the router factory runs, but is only + // invoked once the schedule fires, by which point this is set. + let reader: ExchangeRateReaderService | null = null; + let warmCurrencies: readonly string[] = []; + let warmPivot = ''; + ctx.provide(EXCHANGE_RATE_READER, (c: TypedContainer) => { const platformConfig = c.get(PLATFORM_CONFIG); const exchangeRateConfig = platformConfig.exchangeRate; - return new ExchangeRateReaderService({ + const service = new ExchangeRateReaderService({ drizzle: c.get(DRIZZLE), pivot: resolveExchangeRatePivot(exchangeRateConfig), cryptoProvider: c.has(CRYPTO_EXCHANGE_RATE_PROVIDER) @@ -35,7 +57,45 @@ export default { freshTtlMs: exchangeRateConfig?.freshTtlMs ?? DEFAULT_FRESH_TTL_MS, hardMaxAgeMs: exchangeRateConfig?.hardMaxAgeMs ?? DEFAULT_HARD_MAX_AGE_MS, providerTimeoutMs: exchangeRateConfig?.providerTimeoutMs ?? DEFAULT_PROVIDER_TIMEOUT_MS, + failureCooldownMs: exchangeRateConfig?.failureCooldownMs ?? DEFAULT_FAILURE_COOLDOWN_MS, }); + reader = service; + const pivot = resolveExchangeRatePivot(exchangeRateConfig); + warmPivot = pivot; + // The launch catalog: every crypto currency the operator actually offers, plus + // whatever fiat it displays. Kept warm so a hot path (RG gate, rank accrual, a + // social transfer) never waits on the vendor for a coin players actually use - + // an unlisted currency simply falls back to the existing on-demand fetch. + warmCurrencies = [ + ...new Set( + [ + ...(platformConfig.wallet?.cryptoCurrencies ?? DEFAULT_CRYPTO_CURRENCIES), + ...resolveDisplayCurrencies(platformConfig.displayCurrencies), + ] + .map((code) => code.toUpperCase()) + .filter((code) => code !== pivot), + ), + ]; + return service; + }); + + ctx.jobs.worker({ + queue: RATE_WARM_QUEUE, + schema: RateWarmJobSchema, + handler: async () => { + if (!reader) { + throw new Error('exchange rate warm-up: reader not constructed yet'); + } + const svc = reader; + const pivot = warmPivot; + await mapConcurrent(warmCurrencies, WARM_CONCURRENCY, async (currency) => { + try { + await svc.getRate(currency, pivot); + } catch (err) { + logger.warn({ err, currency }, 'exchange rate warm-up failed'); + } + }); + }, }); ctx.routers.add('exchangeRate', (c) => { @@ -46,6 +106,14 @@ export default { ...resolveDisplayCurrencies(platformConfig.displayCurrencies), resolveExchangeRatePivot(platformConfig.exchangeRate), ]; + + const jobQueue = c.get(JOB_QUEUE); + // Idempotent schedule (keyed by scheduleId) - see wallet's custody-sweep cron for + // the same pattern. + void jobQueue + .schedule(RATE_WARM_QUEUE, 'exchange-rate-warm.cron', {}, { everyMs: WARM_INTERVAL_MS }) + .catch((err) => logger.error({ err }, 'exchange-rate-warm schedule failed')); + return createExchangeRateRouter( new ExchangeRateService(c.get(EXCHANGE_RATE_READER), supported), ); diff --git a/packages/core/src/fx/exchange-rate/react/exchange-rate.ts b/packages/core/src/fx/exchange-rate/react/exchange-rate.ts index f2c7c88cb..507879fc6 100644 --- a/packages/core/src/fx/exchange-rate/react/exchange-rate.ts +++ b/packages/core/src/fx/exchange-rate/react/exchange-rate.ts @@ -4,9 +4,19 @@ import { useQuery } from '@tanstack/react-query'; import { useOrpcQueryUtils } from '@openora/core/react'; import { exchangeRateContract } from '../contract/index.js'; +// Matches the reader's own freshTtlMs default - a mount or window refocus inside that +// window is still fresh server-side, so refetching only adds vendor-adjacent load for a +// value that hasn't changed. +const STALE_TIME_MS = 60_000; + export function useExchangeRate(from: string, to: string) { const utils = useOrpcQueryUtils(exchangeRateContract); - return useQuery({ ...utils.getRate.queryOptions({ input: { from, to } }), retry: false }); + return useQuery({ + ...utils.getRate.queryOptions({ input: { from, to } }), + retry: false, + staleTime: STALE_TIME_MS, + refetchOnWindowFocus: false, + }); } export function useExchangeRates(to: string, from: readonly string[]) { @@ -15,5 +25,7 @@ export function useExchangeRates(to: string, from: readonly string[]) { ...utils.getRates.queryOptions({ input: { to, from: [...from] } }), enabled: from.length > 0, retry: false, + staleTime: STALE_TIME_MS, + refetchOnWindowFocus: false, }); } diff --git a/packages/testing/src/__tests__/exchange-rate-warm.e2e.test.ts b/packages/testing/src/__tests__/exchange-rate-warm.e2e.test.ts new file mode 100644 index 000000000..78f50f793 --- /dev/null +++ b/packages/testing/src/__tests__/exchange-rate-warm.e2e.test.ts @@ -0,0 +1,58 @@ +import { afterAll, beforeAll, describe, expect, it, vi } from 'vitest'; +import { fileURLToPath } from 'node:url'; +import { DRIZZLE, loadExtensions } from '@openora/core/server'; +import { DEFAULT_CRYPTO_CURRENCIES, JOB_QUEUE, queue } from '@openora/core/contracts'; +import { exchangeRateQuote } from '@openora/core/fx/schema/exchange-rate'; +import { bootTestApp, setupTestDb, type TestApp, type TestDb } from '../index.js'; +import { TEST_EXCHANGE_RATE } from '../test-exchange-rate-provider-plugin.js'; + +const rateProviderPluginPath = fileURLToPath( + new URL('../test-exchange-rate-provider-plugin.ts', import.meta.url), +); + +let db: TestDb; +let testApp: TestApp; + +beforeAll(async () => { + process.env['BETTER_AUTH_SECRET'] ??= 'e2e-test-better-auth-secret-please-change-000000'; + process.env['AUTH_SECRET'] ??= process.env['BETTER_AUTH_SECRET']; + process.env['WITHDRAWAL_PIN_HMAC_SECRET'] ??= 'e2e-test-withdrawal-pin-hmac-secret-000000'; + process.env['NODE_ENV'] ??= 'test'; + + db = await setupTestDb(); + testApp = await bootTestApp({ + plugins: [ + ...(await loadExtensions()), + { id: 'testing-exchange-rate-provider', path: rateProviderPluginPath }, + ], + databaseUrl: db.url, + }); +}, 60_000); + +afterAll(async () => { + await testApp?.close(); + await db?.dispose(); +}); + +const storedRates = async () => + testApp.container + .get(DRIZZLE) + .db.select({ base: exchangeRateQuote.baseCurrency, rate: exchangeRateQuote.rate }) + .from(exchangeRateQuote); + +describe('exchange rate warm-up job', () => { + it('stores a rate for every configured crypto currency without any caller asking for one', async () => { + await testApp.container.get(JOB_QUEUE).enqueue(queue('exchange-rate-warm'), {}); + + await vi.waitFor( + async () => { + const rows = await storedRates(); + expect(rows.map((row) => row.base)).toEqual( + expect.arrayContaining([...DEFAULT_CRYPTO_CURRENCIES]), + ); + expect(rows.every((row) => row.rate === TEST_EXCHANGE_RATE)).toBe(true); + }, + { timeout: 10_000 }, + ); + }); +}); diff --git a/packages/testing/src/test-exchange-rate-provider-plugin.ts b/packages/testing/src/test-exchange-rate-provider-plugin.ts new file mode 100644 index 000000000..7d396925d --- /dev/null +++ b/packages/testing/src/test-exchange-rate-provider-plugin.ts @@ -0,0 +1,26 @@ +import type { CoreTokenCatalog, Plugin } from '@openora/core/server'; +import { CRYPTO_EXCHANGE_RATE_PROVIDER, type ExchangeRateProvider } from '@openora/core/contracts'; + +/** What the provider quotes for every currency, against any pivot. */ +export const TEST_EXCHANGE_RATE = '2.000000000000000000'; + +/** + * Binds a crypto rate provider that quotes `TEST_EXCHANGE_RATE` for every currency, so a test + * can prove a rate was fetched and stored without a vendor. Opt-in only - pass it in + * `config.plugins`. + */ +export default { + id: 'testing-exchange-rate-provider', + dependsOn: ['exchange-rate'], + register(ctx) { + ctx.provide( + CRYPTO_EXCHANGE_RATE_PROVIDER, + () => + ({ + async getRate() { + return { rate: TEST_EXCHANGE_RATE, asOf: new Date().toISOString() }; + }, + }) satisfies ExchangeRateProvider, + ); + }, +} satisfies Plugin;