Add upstream error classification and optional persistent cache for finance connectors

.gitignore

@@ -3,3 +3,4 @@ npm-debug.log*
 .env
 .DS_Store
 coverage/
+.cache/

README.md

@@ -132,7 +132,9 @@ The running server loads connector settings from the environment via
 | `PORTFOLIO_WATCHER_API_KEY` | Portfolio-Watcher credential placeholder | empty |
 | `TAX_BREAK_ENDPOINT` | tax-break endpoint placeholder | `mock://tax-break` |
 | `TAX_BREAK_API_KEY` | tax-break credential placeholder | empty |
-| `FINANCE_CACHE_TTL_MS` | Minimal resolver cache TTL | `1000` |
+| `FINANCE_CACHE_TTL_MS` | Resolver cache TTL | `1000` |
+| `FINANCE_CACHE_STORE` | Cache strategy: `memory` or `file` (persistent) | `memory` |
+| `FINANCE_CACHE_FILE` | Cache file used when `FINANCE_CACHE_STORE=file` | `.cache/finance-cache.json` |
 | `FINANCE_HTTP_TIMEOUT_MS` | Per-request upstream timeout | `5000` |
 | `FINANCE_HTTP_MAX_RETRIES` | Retries for timeouts, 429s, and 5xx responses | `2` |
 | `FINANCE_DEFAULT_PAGE_SIZE` | Default page size when `limit` is omitted | `25` |
@@ -247,14 +249,34 @@ query RecentSells {
   operation name, duration, outcome, and error codes.
 - **Metrics**: `src/observability/metrics.js` records GraphQL operation
   counts/latency, connector call counts/latency per source and operation,
-  upstream retry failures, and cache hit/miss/coalesced counters. Scrape them at
-  `GET /metrics`.
+  upstream retry failures, classified failures
+  (`finance_upstream_errors_total{source,operation,category,retryable}`), and
+  cache hit/miss/coalesced counters labelled with the active store. Scrape them
+  at `GET /metrics`.
+- **Error classification**: `src/observability/errors.js` maps every upstream
+  failure to a stable `category` (`AUTH`, `RATE_LIMIT`, `TIMEOUT`, `NETWORK`,
+  `UPSTREAM_CLIENT_ERROR`, `UPSTREAM_SERVER_ERROR`, `UNKNOWN`) plus `status` and
+  `retryable`. Those fields are returned on every payload's
+  `errors { source code category status retryable message }`, so a partial
+  response still explains what failed and whether retrying helps.
 - **Health**: `GET /health` is a liveness probe; `GET /ready` calls each
   connector's health check and returns `503` when any upstream is degraded.
 
+### Caching
+
+Upstream reads go through a TTL cache selected by `FINANCE_CACHE_STORE`
+(`src/cache/index.js`):
+
+- `memory` (default): in-process, fastest, cleared on restart.
+- `file`: the same TTL semantics mirrored to `FINANCE_CACHE_FILE`, so a
+  restarted process serves warm upstream data instead of refetching everything.
+
+Concurrent resolvers asking for the same key share one in-flight request, and
+payloads containing upstream errors are never cached so a transient outage is
+not pinned for the whole TTL.
+
 Follow-up production tasks:
 
-- Add persisted caching/batching if upstream latency becomes significant.
 - Move from offset pagination to cursor pagination if upstream APIs expose
   stable cursors.
 

src/cache/index.js

@@ -0,0 +1,124 @@
+import { readFileSync } from 'node:fs';
+import { mkdir, rename, writeFile } from 'node:fs/promises';
+import { dirname } from 'node:path';
+
+import { logger as defaultLogger } from '../observability/logger.js';
+
+/** Default in-process cache: fastest option, lost on restart. */
+export function createMemoryCacheStore() {
+  const entries = new Map();
+
+  return {
+    kind: 'memory',
+    get(key) {
+      const hit = entries.get(key);
+      if (!hit) return undefined;
+      if (hit.expiresAt <= Date.now()) {
+        entries.delete(key);
+        return undefined;
+      }
+      return hit.value;
+    },
+    set(key, value, ttlMs) {
+      entries.set(key, { value, expiresAt: Date.now() + ttlMs });
+    },
+    clear() {
+      entries.clear();
+    },
+  };
+}
+
+/**
+ * Optional persistent cache: entries are mirrored to a JSON file so a restarted
+ * process can serve warm upstream data instead of re-fetching everything.
+ * Values must be JSON-serialisable, which holds for all normalized finance
+ * payloads.
+ */
+export function createFileCacheStore({ file, logger = defaultLogger } = {}) {
+  if (!file) throw new Error('file cache store requires a file path');
+
+  const memory = new Map();
+  let writeChain = Promise.resolve();
+  let writing = false;
+  let dirty = false;
+
+  function load() {
+    try {
+      const parsed = JSON.parse(readFileSync(file, 'utf8'));
+      for (const [key, entry] of Object.entries(parsed)) {
+        if (entry && typeof entry.expiresAt === 'number' && entry.expiresAt > Date.now()) {
+          memory.set(key, entry);
+        }
+      }
+    } catch (error) {
+      // A missing or corrupt cache file is never fatal: start cold instead.
+      if (error?.code !== 'ENOENT') {
+        logger.warn('persistent cache could not be read', { file, error: error?.message ?? String(error) });
+      }
+    }
+  }
+
+  function persist() {
+    dirty = true;
+    if (writing) return writeChain;
+
+    writing = true;
+    writeChain = (async () => {
+      try {
+        // Coalesce every update queued while a write is in flight into one pass.
+        while (dirty) {
+          dirty = false;
+          const payload = JSON.stringify(Object.fromEntries(memory));
+          await mkdir(dirname(file), { recursive: true });
+          const temporaryFile = `${file}.${process.pid}.tmp`;
+          await writeFile(temporaryFile, payload);
+          await rename(temporaryFile, file);
+        }
+      } catch (error) {
+        logger.warn('persistent cache could not be written', { file, error: error?.message ?? String(error) });
+      } finally {
+        writing = false;
+      }
+    })();
+
+    return writeChain;
+  }
+
+  load();
+
+  return {
+    kind: 'file',
+    file,
+    get(key) {
+      const hit = memory.get(key);
+      if (!hit) return undefined;
+      if (hit.expiresAt <= Date.now()) {
+        memory.delete(key);
+        persist();
+        return undefined;
+      }
+      return hit.value;
+    },
+    set(key, value, ttlMs) {
+      memory.set(key, { value, expiresAt: Date.now() + ttlMs });
+      persist();
+    },
+    clear() {
+      memory.clear();
+      persist();
+    },
+    /** Resolves once every queued write has been flushed to disk. */
+    flush() {
+      return writeChain;
+    },
+  };
+}
+
+/** Selects a cache store from configuration (`memory` by default). */
+export function createCacheStore({ store = 'memory', file, logger = defaultLogger } = {}) {
+  if (store === 'file') return createFileCacheStore({ file, logger });
+  if (store !== 'memory') {
+    logger.warn('unknown cache store, falling back to memory', { store });
+  }
+  return createMemoryCacheStore();
+}

src/config/finance.js

@@ -27,6 +27,10 @@ export function loadFinanceConfig(env = process.env) {
       maxRetries,
     },
     cacheTtlMs: parseNonNegativeInt(env.FINANCE_CACHE_TTL_MS, 1000),
+    // `memory` keeps the cache in-process; `file` persists it so restarts and
+    // short-lived workers can reuse warm upstream data.
+    cacheStore: env.FINANCE_CACHE_STORE ?? 'memory',
+    cacheFile: env.FINANCE_CACHE_FILE ?? '.cache/finance-cache.json',
     defaultPageSize: Number(env.FINANCE_DEFAULT_PAGE_SIZE ?? 25),
     maxPageSize: Number(env.FINANCE_MAX_PAGE_SIZE ?? 100),
   };

src/connectors/httpClient.js

@@ -3,12 +3,15 @@ import { metrics as defaultMetrics } from '../observability/metrics.js';
 
 /** Error carrying the upstream source and HTTP status for actionable messages. */
 export class UpstreamHttpError extends Error {
-  constructor(source, message, { status = null, retryable = false } = {}) {
+  constructor(source, message, { status = null, retryable = false, kind = 'http' } = {}) {
     super(message);
     this.name = 'UpstreamHttpError';
     this.source = source;
     this.status = status;
     this.retryable = retryable;
+    // `kind` distinguishes transport failures (timeout/network) from HTTP
+    // responses so they can be classified without parsing the message.
+    this.kind = kind;
   }
 }
 
@@ -75,11 +78,18 @@ export function createHttpClient({
         try {
           return await once(path, { ...options, signal: controller.signal });
         } catch (error) {
-          lastError = error instanceof UpstreamHttpError
-            ? error
-            : new UpstreamHttpError(source, error?.name === 'AbortError' ? `request to ${path} timed out after ${timeoutMs}ms` : String(error?.message ?? error), { retryable: true });
-
-          metrics.increment('finance_upstream_attempt_failures_total', { source, path });
+          if (error instanceof UpstreamHttpError) {
+            lastError = error;
+          } else {
+            const timedOut = error?.name === 'AbortError';
+            lastError = new UpstreamHttpError(
+              source,
+              timedOut ? `request to ${path} timed out after ${timeoutMs}ms` : String(error?.message ?? error),
+              { retryable: true, kind: timedOut ? 'timeout' : 'network' }
+            );
+          }
+
+          metrics.increment('finance_upstream_attempt_failures_total', { source, path, kind: lastError.kind });
           logger.warn('upstream request failed', {
             source,
             path,

src/observability/errors.js

@@ -0,0 +1,58 @@
+/**
+ * Upstream failure classification.
+ *
+ * Resolvers, metrics and clients all benefit from a stable category instead of
+ * a raw message: dashboards can alert on `AUTH` or `RATE_LIMIT` separately from
+ * generic outages, and callers can decide whether a retry is worthwhile.
+ */
+export const ERROR_CATEGORIES = {
+  AUTH: 'AUTH',
+  RATE_LIMIT: 'RATE_LIMIT',
+  TIMEOUT: 'TIMEOUT',
+  NETWORK: 'NETWORK',
+  UPSTREAM_CLIENT_ERROR: 'UPSTREAM_CLIENT_ERROR',
+  UPSTREAM_SERVER_ERROR: 'UPSTREAM_SERVER_ERROR',
+  UNKNOWN: 'UNKNOWN',
+};
+
+function categoryFromStatus(status) {
+  if (status === 401 || status === 403) return ERROR_CATEGORIES.AUTH;
+  if (status === 429) return ERROR_CATEGORIES.RATE_LIMIT;
+  if (status === 408 || status === 504) return ERROR_CATEGORIES.TIMEOUT;
+  if (status >= 500) return ERROR_CATEGORIES.UPSTREAM_SERVER_ERROR;
+  if (status >= 400) return ERROR_CATEGORIES.UPSTREAM_CLIENT_ERROR;
+  return ERROR_CATEGORIES.UNKNOWN;
+}
+
+function categoryFromError(error) {
+  if (error?.kind === 'timeout' || error?.name === 'AbortError') return ERROR_CATEGORIES.TIMEOUT;
+  if (error?.kind === 'network') return ERROR_CATEGORIES.NETWORK;
+  return ERROR_CATEGORIES.UNKNOWN;
+}
+
+/** Categories that are transient and therefore worth retrying. */
+const RETRYABLE = new Set([
+  ERROR_CATEGORIES.RATE_LIMIT,
+  ERROR_CATEGORIES.TIMEOUT,
+  ERROR_CATEGORIES.NETWORK,
+  ERROR_CATEGORIES.UPSTREAM_SERVER_ERROR,
+]);
+
+/**
+ * Converts any thrown value into the `FinanceUpstreamError` shape exposed by
+ * the schema.
+ */
+export function classifyUpstreamError(source, error) {
+  const message = error instanceof Error ? error.message : String(error);
+  const status = Number.isInteger(error?.status) ? error.status : null;
+  const category = status === null ? categoryFromError(error) : categoryFromStatus(status);
+
+  return {
+    source,
+    code: status === null ? `UPSTREAM_${category}` : `UPSTREAM_HTTP_${status}`,
+    category,
+    status,
+    retryable: typeof error?.retryable === 'boolean' ? error.retryable : RETRYABLE.has(category),
+    message: `${source} connector failed: ${message}`,
+  };
+}

src/schema.js

@@ -103,6 +103,12 @@ export const typeDefs = /* GraphQL */ `
   type FinanceUpstreamError {
     source: String!
     code: String!
+    "Stable failure class: AUTH, RATE_LIMIT, TIMEOUT, NETWORK, UPSTREAM_CLIENT_ERROR, UPSTREAM_SERVER_ERROR or UNKNOWN."
+    category: String!
+    "HTTP status when the failure came from an upstream response."
+    status: Int
+    "True when retrying the request may succeed."
+    retryable: Boolean!
     message: String!
   }
 

src/services/financeService.js

@@ -1,5 +1,7 @@
+import { createCacheStore } from '../cache/index.js';
 import { createConnectors } from '../connectors/index.js';
 import { loadFinanceConfig } from '../config/finance.js';
+import { classifyUpstreamError } from '../observability/errors.js';
 import { logger as defaultLogger } from '../observability/logger.js';
 import { metrics as defaultMetrics } from '../observability/metrics.js';
 import {
@@ -18,25 +20,17 @@ import {
   paginate,
 } from '../domain/finance.js';
 
-function upstreamError(source, error) {
-  const message = error instanceof Error ? error.message : String(error);
-  return {
-    source,
-    code: error?.status ? `UPSTREAM_HTTP_${error.status}` : 'UPSTREAM_UNAVAILABLE',
-    message: `${source} connector failed: ${message}`,
-  };
-}
-
 export function createFinanceService({
   config = loadFinanceConfig(),
   connectors,
   cacheTtlMs = config.cacheTtlMs,
+  cacheStore,
   logger = defaultLogger,
   metrics = defaultMetrics,
 } = {}) {
   const upstreams = connectors ?? createConnectors({ config, logger, metrics });
   const pageLimits = { defaultLimit: config.defaultPageSize ?? 25, maxLimit: config.maxPageSize ?? 100 };
-  const cache = new Map();
+  const cache = cacheStore ?? createCacheStore({ store: config.cacheStore, file: config.cacheFile, logger });
   // Concurrent resolvers asking for the same upstream data share a single
   // in-flight request instead of fanning out duplicate connector calls.
   const inFlight = new Map();
@@ -47,31 +41,46 @@ export function createFinanceService({
       const data = await metrics.time('finance_connector_call', { source, operation }, task);
       return { data, error: null };
     } catch (error) {
-      logger.error('finance connector call failed', { source, operation, error: error?.message ?? String(error) });
-      return { data: [], error: upstreamError(source, error) };
+      const classified = classifyUpstreamError(source, error);
+      metrics.increment('finance_upstream_errors_total', {
+        source,
+        operation,
+        category: classified.category,
+        retryable: classified.retryable,
+      });
+      logger.error('finance connector call failed', {
+        source,
+        operation,
+        category: classified.category,
+        status: classified.status,
+        retryable: classified.retryable,
+        error: error?.message ?? String(error),
+      });
+      return { data: [], error: classified };
     }
   }
 
   async function cached(key, load) {
-    const now = Date.now();
     const hit = cache.get(key);
-    if (hit && hit.expiresAt > now) {
-      metrics.increment('finance_cache_total', { key, outcome: 'hit' });
-      return hit.value;
+    if (hit !== undefined) {
+      metrics.increment('finance_cache_total', { key, outcome: 'hit', store: cache.kind });
+      return hit;
     }
 
     const pending = inFlight.get(key);
     if (pending) {
-      metrics.increment('finance_cache_total', { key, outcome: 'coalesced' });
+      metrics.increment('finance_cache_total', { key, outcome: 'coalesced', store: cache.kind });
       return pending;
     }
 
-    metrics.increment('finance_cache_total', { key, outcome: 'miss' });
+    metrics.increment('finance_cache_total', { key, outcome: 'miss', store: cache.kind });
 
     const request = (async () => {
       try {
         const value = await load();
-        cache.set(key, { value, expiresAt: Date.now() + cacheTtlMs });
+        // Failed upstream reads are not cached: a transient outage should not
+        // pin an empty payload for the whole TTL.
+        if ((value.errors ?? []).length === 0) cache.set(key, value, cacheTtlMs);
         return value;
       } finally {
         inFlight.delete(key);