feat(fx-engine): add cross-rate batch-ID validation

services/fx-engine/src/index.ts

@@ -67,10 +67,14 @@ const SUPPORTED_CURRENCIES = Object.keys(FALLBACK_RATES);
 
 interface RateCache {
   rates: Record<string, number>;
+  batchIds: Record<string, string>;
   cachedAt: number; // Unix ms timestamp
 }
 
+const initialBatchId = randomUUID();
 let cache: RateCache = {
+  rates:    { ...FALLBACK_RATES },
+  batchIds: Object.fromEntries(Object.keys(FALLBACK_RATES).map((c) => [c, initialBatchId])),
   rates: { ...FALLBACK_RATES },
   cachedAt: Date.now(),
 };
@@ -111,6 +115,23 @@ function getOrComputeRate(from: string, to: string): number {
     return entry.rate;
   }
 
+  const fromBatch = cache.batchIds[from];
+  const toBatch   = cache.batchIds[to];
+
+  if (!fromBatch) {
+    throw new Error(`No rate batch information for ${from}`);
+  }
+  if (!toBatch) {
+    throw new Error(`No rate batch information for ${to}`);
+  }
+
+  if (fromBatch !== toBatch) {
+    fastify.log.warn(
+      { from, to, fromBatch, toBatch },
+      'Cross-rate computed with rates from different fetch cycles',
+    );
+  }
+
   const rate = computeRate(from, to, cache.rates);
   computedRateCache.set(key, { rate, computedAt: now });
   return rate;
@@ -140,10 +161,14 @@ async function storeRateSnapshot(rates: Record<string, number>): Promise<void> {
     .exec();
 }
 
-function updateBaseRates(newRates: Record<string, number>): void {
-  cache = { rates: newRates, cachedAt: Date.now() };
+function updateBaseRates(updated: Record<string, number>, batchId: string): void {
+  for (const [currency, value] of Object.entries(updated)) {
+    cache.rates[currency] = value;
+    cache.batchIds[currency] = batchId;
+  }
+  cache.cachedAt = Date.now();
   computedRateCache.clear();
-  storeRateSnapshot(newRates).catch(() => {}); // Redis errors are non-fatal
+  storeRateSnapshot({ ...cache.rates }).catch(() => {}); // Redis errors are non-fatal
 }
 
 // ── Live rate refresh loop (issue #251) ────────────────────────────────────
@@ -395,6 +420,11 @@ async function refreshTick(): Promise<void> {
       try {
         const fetched = await fetchBaseRates();
         if (fetched) {
+          const batchId = randomUUID();
+          updateBaseRates(fetched, batchId);
+          fastify.log.info(
+            { durationMs: lastRefresh?.durationMs, assets: Object.keys(fetched), rateBatchId: batchId },
+            'FX rates refreshed',
           const maxDeviationBps = env.MAX_DEVIATION_BPS;
           const merged: Record<string, number> = { ...cache.rates };
           const rejected: string[] = [];
@@ -465,6 +495,11 @@ async function refreshTick(): Promise<void> {
     fastify.log.warn("Stampede poll timed out; falling back to direct fetch");
     const fetched = await fetchBaseRates();
     if (fetched) {
+      const batchId = randomUUID();
+      updateBaseRates(fetched, batchId);
+    } else if (fallbackStartTime === null) {
+      fallbackStartTime = Date.now();
+      fastify.log.warn('Entering fallback FX rate mode');
       const maxDeviationBps = env.MAX_DEVIATION_BPS;
       const merged: Record<string, number> = { ...cache.rates };
       const rejected: string[] = [];
@@ -575,6 +610,8 @@ async function warmupCacheFromRedis(): Promise<void> {
       return;
     }
 
+    const snapshot = JSON.parse(members[0]) as { ts: number; rates: Record<string, number> };
+    updateBaseRates(snapshot.rates, randomUUID());
     // If all rates for all currencies were discarded, trigger immediate fetch
     if (Object.keys(validatedRates).length === 0) {
       fastify.log.warn(
@@ -702,6 +739,8 @@ interface StoredQuote {
   result: string;
   rate: string;
   slippageBps: number;
+  expiresAt:   number; // Unix ms — quote validity cutoff
+  rateBatchId: string;
   expiresAt: number; // Unix ms — quote validity cutoff
 }
 
@@ -1173,6 +1212,7 @@ fastify.get(
     // Store quote so it can be verified later. If Redis is unavailable the
     // quote is still returned — clients just won't be able to call /verify.
     let quoteId: string | null = null;
+    const rateBatchId = cache.batchIds[from] ?? '';
     try {
       quoteId = randomUUID();
       const stored: StoredQuote = {
@@ -1184,6 +1224,7 @@ fastify.get(
         rate: exchangeRate.toFixed(8),
         slippageBps: effectiveBps,
         expiresAt,
+        rateBatchId,
       };
       await redis.set(
         `${QUOTE_KEY_PREFIX}${quoteId}`,
@@ -1215,6 +1256,9 @@ fastify.get(
       rate: exchangeRate.toFixed(8),
       slippageBps: effectiveBps,
       slippageLimit,
+      cachedAt:      new Date(cache.cachedAt).toISOString(),
+      expiresAt:     new Date(expiresAt).toISOString(),
+      rateBatchId,
       cachedAt: new Date(cache.cachedAt).toISOString(),
       expiresAt: new Date(expiresAt).toISOString(),
     };
@@ -1412,6 +1456,7 @@ fastify.post<{ Body: VerifyQuoteRouteBody }>(
 
     const currentRate = getOrComputeRate(stored.from, stored.to);
     const slippageBps = stored.slippageBps ?? env.DEFAULT_SLIPPAGE_BPS;
+    const rateBatchId = cache.batchIds[stored.from] ?? '';
     const quotedRate = parseFloat(stored.rate);
 
     // Fail-open: if market rate is unavailable (fallback mode), accept by expiry
@@ -1434,6 +1479,8 @@ fastify.post<{ Body: VerifyQuoteRouteBody }>(
       currentRate: currentRate.toFixed(8),
       slippageBps,
       slippageLimit: (slippageBps / 10_000).toFixed(4),
+      expiresAt:     new Date(stored.expiresAt).toISOString(),
+      rateBatchId,
       expiresAt: new Date(stored.expiresAt).toISOString(),
     };
   },

services/fx-engine/src/rate-refresh.test.ts

@@ -241,6 +241,98 @@ test('setInterval-driven loop: refetches at the configured interval', async (t)
   t.end();
 });
 
+// ── Cross-rate batch-ID validation (issue #??? ) ──────────────────────────
+
+interface RateEntry {
+  value: number;
+  rateBatchId: string;
+}
+
+function computeCrossRate(
+  from: string,
+  to: string,
+  rates: Record<string, RateEntry>,
+  warn: (obj: Record<string, unknown>, msg: string) => void,
+): number {
+  const fromEntry = rates[from];
+  const toEntry   = rates[to];
+
+  if (!fromEntry) {
+    throw new Error(`No rate batch information for ${from}`);
+  }
+  if (!toEntry) {
+    throw new Error(`No rate batch information for ${to}`);
+  }
+
+  if (fromEntry.rateBatchId !== toEntry.rateBatchId) {
+    warn(
+      { from, to, fromBatch: fromEntry.rateBatchId, toBatch: toEntry.rateBatchId },
+      'Cross-rate computed with rates from different fetch cycles',
+    );
+  }
+
+  return fromEntry.value / toEntry.value;
+}
+
+test('cross-rate: same batch — computed without warning', (t) => {
+  const batchId = '550e8400-e29b-41d4-a716-446655440000';
+  const rates: Record<string, RateEntry> = {
+    USDC: { value: 1500, rateBatchId: batchId },
+    EURT: { value: 1700, rateBatchId: batchId },
+    NGN:  { value: 1,    rateBatchId: batchId },
+  };
+  const warnings: Array<{ obj: Record<string, unknown>; msg: string }> = [];
+  const logger = { warn: (obj: Record<string, unknown>, msg: string) => warnings.push({ obj, msg }) };
+
+  const result = computeCrossRate('USDC', 'EURT', rates, logger.warn);
+
+  t.equal(result, 1500 / 1700, 'cross-rate computed correctly (USDC/EURT)');
+  t.equal(result, 0.8823529411764706, 'cross-rate matches expected value');
+  t.equal(warnings.length, 0, 'no warning logged for same batch');
+  t.end();
+});
+
+test('cross-rate: different batches — computed with warning logged', (t) => {
+  const rates: Record<string, RateEntry> = {
+    USDC: { value: 1500, rateBatchId: 'batch-a-0000-0000-0000-000000000001' },
+    EURT: { value: 1700, rateBatchId: 'batch-b-0000-0000-0000-000000000002' },
+    NGN:  { value: 1,    rateBatchId: 'batch-a-0000-0000-0000-000000000001' },
+  };
+  const warnings: Array<{ obj: Record<string, unknown>; msg: string }> = [];
+  const logger = { warn: (obj: Record<string, unknown>, msg: string) => warnings.push({ obj, msg }) };
+
+  const result = computeCrossRate('USDC', 'EURT', rates, logger.warn);
+
+  t.equal(result, 1500 / 1700, 'cross-rate still computed correctly with mixed batches');
+  t.equal(warnings.length, 1, 'exactly one warning logged');
+  if (warnings.length > 0) {
+    t.equal(warnings[0].obj.from, 'USDC', 'warning obj includes from currency');
+    t.equal(warnings[0].obj.to, 'EURT', 'warning obj includes to currency');
+    t.ok(
+      warnings[0].msg.includes('different fetch cycles'),
+      'warning message mentions different fetch cycles',
+    );
+  }
+  t.end();
+});
+
+test('cross-rate: only one rate available — error thrown', (t) => {
+  const rates: Record<string, RateEntry> = {
+    USDC: { value: 1500, rateBatchId: 'batch-a-0000-0000-0000-000000000001' },
+    // EURT is missing entirely
+    NGN:  { value: 1,    rateBatchId: 'batch-a-0000-0000-0000-000000000001' },
+  };
+  const warnings: Array<{ obj: Record<string, unknown>; msg: string }> = [];
+  const logger = { warn: (obj: Record<string, unknown>, msg: string) => warnings.push({ obj, msg }) };
+
+  try {
+    computeCrossRate('USDC', 'EURT', rates, logger.warn);
+    t.fail('expected error for missing EURT rate');
+  } catch (err) {
+    const e = err as Error;
+    t.ok(e.message.includes('EURT'), `error mentions the missing currency: ${e.message}`);
+    t.equal(warnings.length, 0, 'no warning logged when one rate is missing');
+  }
 test('jitter: delays are within ±25% range and mean approximates the base interval', (t) => {
   const BASE_INTERVAL = 100;
   const SAMPLES = 1000;

shared/validation/schemas.ts

@@ -140,6 +140,8 @@ export const fxQuoteSchema = z.object({
   fromCurrency: CurrencyCode,
   toCurrency: CurrencyCode,
   rate: z.string(),
+  expiresAt: isoDateString,
+  rateBatchId: z.string().uuid()
   slippageBps: z.number().int().min(0).optional(),
   expiresAt: isoDateString
 });