rubbathabrahband · Oct 17, 2026
diff --git a/.env.example b/.env.example
@@ -81,6 +81,50 @@ TRUST_INBOUND_REQUEST_ID=true
 INDEXER_ENABLED=true
 INDEXER_POLL_MS=5000
 
+# ── Socket.io throttling & slow-consumer guard (Issue #96) ──────────────────
+# Price tick coalescing interval (ms); at most one MARKET_PRICE emitted per market per tick.
+SOCKET_PRICE_TICK_MS=1000
+# Deliver TRADE_EXECUTED as volatile (skip stalled/congested clients).
+SOCKET_TRADE_VOLATILE=true
+# Max ledger lag to classify an event as live vs historical catch-up.
+SOCKET_LIVE_LEDGER_LAG=12
+# Buffered bytes threshold to degrade socket (stop sending trades, keep price ticks).
+SOCKET_SLOW_WARN_BYTES=262144
+# Buffered bytes threshold to disconnect stalled consumer socket.
+SOCKET_SLOW_DROP_BYTES=2097152
+# Check interval for slow-consumer buffer inspection (ms).
+SOCKET_SLOW_CHECK_MS=2000
+
+# ── Socket.io hardening & connection limits (Issue #97) ─────────────────────
+# Maximum concurrent connections per instance.
+SOCKET_MAX_CONNECTIONS=5000
+# Maximum concurrent connections per client IP.
+SOCKET_MAX_PER_IP=20
+# Connection rate limit (new connections per IP per minute).
+SOCKET_CONNECT_RATE_PER_MIN=60
+# Max inbound frame/buffer size (bytes).
+SOCKET_MAX_BUFFER_BYTES=16384
+# Allowed transports: websocket,polling or websocket.
+SOCKET_TRANSPORTS=websocket,polling
+# Max inbound application events per socket per 10 seconds.
+SOCKET_INBOUND_EVENTS_PER_10S=20
+# Fraction of socket connections/disconnections logged at info level (0.0 to 1.0).
+SOCKET_LOG_SAMPLE_RATE=0.01
+# Comma-separated CIDRs of trusted reverse proxies for X-Forwarded-For parsing.
+TRUSTED_PROXIES=
+# Comma-separated CIDRs exempt from per-IP connection limits (NAT / monitoring).
+SOCKET_IP_EXEMPT_CIDRS=
+
+# ── Chain read cache (Issue #98) ────────────────────────────────────────────
+# Read-through cache for live Soroban contract simulation reads.
+CHAIN_CACHE_ENABLED=true
+# TTL for cached chain reads (ms).
+CHAIN_CACHE_TTL_MS=5000
+# Negative TTL for contract errors such as MarketNotFound (ms).
+CHAIN_CACHE_NEG_TTL_MS=2000
+# Max wait time for another node's single-flight lock before direct simulation (ms).
+CHAIN_CACHE_WAIT_MS=1500
+
 # Health probes (#111): /livez (process only) and /readyz (dependencies).
 # HEALTH_CRITICAL: comma list of database,redis,rpc,indexer that fail readiness.
 HEALTH_CRITICAL=database,redis

diff --git a/docs/api.md b/docs/api.md
@@ -288,7 +288,9 @@ string; `404 MARKET_NOT_FOUND`.
   "sets_minted": "1599000000",
   "status": { "tag": "Resolved", "outcome": "Yes" },
   "yesBps": 5319,
-  "source": "chain"
+  "source": "chain",
+  "ledger": 123456,
+  "cached": true
 }
 ```
 
@@ -462,9 +464,9 @@ Not verified: the contracts repo has no known x402-on-Stellar facilitator URL
 #### `GET /portfolio`
 
 The agent's (treasury's) portfolio. Response:
-`{ totalUsdc, deployedCapital, availableCapital, openPositions, totalPnl, dailyPnl, builderFeesEarned, correlationRisk, availableCapitalRaw }`.
+`{ totalUsdc, deployedCapital, availableCapital, openPositions, totalPnl, dailyPnl, builderFeesEarned, correlationRisk, availableCapitalRaw, availableCapitalLedger, latestReport }`.
 `availableCapitalRaw` is `treasury.available_capital()` read live (`null` if
-the RPC read fails). The float fields are the DB's view.
+the RPC read fails), and `availableCapitalLedger` is the ledger it was read at. The float fields are the DB's view.
 
 #### `GET /portfolio/positions`
 
@@ -648,18 +650,48 @@ The response now includes `contentSource: "cache" | "gateway"`: `cache` means
 the bytes came from the local verified content store and both gateways could
 be down. The on-chain `trace_hash` remains the authority.
 
+### Oracle resolution
+
+#### `GET /oracle/markets/:marketId/resolution`
+
+Returns resolution status directly from the resolver contract and market-core:
+`{ marketId, onChainMarketId, resolverState, marketStatus, resolutionHash, resolutionSpec, dbStatus, ledger, cached }`.
+`ledger` is the minimum ledger of the live contract reads; `cached` indicates whether values were served from the Redis chain cache.
+
 ## Realtime (socket.io)
 
-Same origin as the API without `/api/v1`. The frontend listens to these
-events; event names are unchanged.
+Same origin as the API without `/api/v1`. The frontend listens to these events.
+
+### Realtime limits & security
+
+- **Allowed origins:** Checked via the shared CORS allowlist (`CORS_ORIGINS`). Websocket upgrades from disallowed origins are rejected (`ORIGIN_NOT_ALLOWED`). Requests without an `Origin` header (non-browser clients) are accepted.
+- **Connection limits:**
+  - `SOCKET_MAX_CONNECTIONS` (default 5000): global connection cap across instances.
+  - `SOCKET_MAX_PER_IP` (default 20): maximum concurrent connections per client IP.
+  - `SOCKET_CONNECT_RATE_PER_MIN` (default 60): maximum new connections per IP per minute.
+  - Rejections emit `CONNECTION_LIMIT` error with reason (`GLOBAL`, `PER_IP`, `RATE`).
+- **Buffers & frames:** `maxHttpBufferSize` is set to `SOCKET_MAX_BUFFER_BYTES` (default 16 KB). Oversized frames close the connection.
+- **Inbound events:** Clients send no application events. Inbound events not in the allowlist are rejected, and more than 3 violations within 60 s trigger immediate disconnection for abuse.
+- **Slow consumers:** Sockets buffering more than `SOCKET_SLOW_WARN_BYTES` (default 256 KB) are degraded (removed from `realtime:full` room, receiving only `MARKET_PRICE` ticks). Sockets buffering more than `SOCKET_SLOW_DROP_BYTES` (default 2 MB) are disconnected.
+
+### `MARKET_PRICE`
 
-**Finding:** before this port the backend never emitted either event (there
-is no `io.emit` anywhere in `src/`), so these listeners received nothing. The
-indexer now emits them.
+Emitted for price updates, coalesced latest-wins per market and emitted at most once per `SOCKET_PRICE_TICK_MS` (default 1000 ms). Delivered reliably (non-volatile) to all connected clients.
+
+```json
+{
+  "onChainMarketId": "0",
+  "marketId": "uuid or null",
+  "priceYesBps": 6421,
+  "ledger": 123456,
+  "eventId": "0000…-0000000001",
+  "ts": 1700000000000
+}
+```
 
 ### `TRADE_EXECUTED`
 
-Emitted when the indexer stores a market-core `trade` event.
+Emitted when the indexer processes a live trade event. Delivered as **volatile** (`SOCKET_TRADE_VOLATILE=true`) to clients in the `realtime:full` room. Clients that cannot take it skip it without building unbounded server buffers. Historical events during indexer catch-up are suppressed.
 
 ```json
 {
@@ -679,21 +711,11 @@ Emitted when the indexer stores a market-core `trade` event.
 }
 ```
 
-The old payload had `amount` (float USDC), `price` (0–1) and `probChange`
-(string). Frontends derive display values from `collateralRaw` and
-`priceYesBps`.
-
-> A trace's `verified` flag means its `trace_hash` was confirmed on-chain: set
-> by the indexer when it sees the matching `TracePublished` event (or by
-> `POST /traces/verify`), never from a write result alone and never in dry-run.
+> **Note:** Clients requiring complete, reliable trade histories must query the REST API (`GET /markets/:id` or indexed history) rather than relying on the volatile socket stream.
 
 ### `REASONING_PUBLISHED`
 
 Emitted when the indexer stores a reasoning-registry publish event.
-`action` is one of the backend's `create_market`, `buy_yes`, `buy_no`
-(`src/lib/trace-actions.ts`); traces published by other agents may use other
-strings, such as `buy` from the contracts repo's TypeScript agents.
-
 ```json
 {
   "onChainTraceId": "3",
@@ -708,6 +730,60 @@ strings, such as `buy` from the contracts repo's TypeScript agents.
 }
 ```
 
+### `CATCHUP_COMPLETE`
+
+Emitted once when the indexer finishes catching up from an older ledger cursor:
+```json
+{
+  "fromLedger": 123000,
+  "toLedger": 123456,
+  "events": 42
+}
+```
+Clients should use this event as a signal to refetch REST state.
+
+## Caching & Conditional Requests
+
+Every endpoint defines an explicit cache policy and strong ETag derivation. Express's default weak ETags are disabled.
+
+### Policy Table
+
+| Route | Policy | Cache-Control | Validator / ETag source |
+|---|---|---|---|
+| `GET /markets` | `shared` | `public, max-age=5, stale-while-revalidate=30` | `max(updatedAt)`, count, pagination |
+| `GET /markets/:id` (active) | `shared` | `public, max-age=5, stale-while-revalidate=30` | `market.updatedAt`, latest trace `updatedAt`, counts |
+| `GET /markets/:id` (resolved/cancelled) | `immutable` | `public, max-age=86400, immutable` | `market.updatedAt`, status, counts |
+| `GET /markets/on-chain/:id` | `shared` / `immutable` | same as `GET /markets/:id` | same as `GET /markets/:id` |
+| `GET /markets/on-chain/:id/state` | `shared` | `public, max-age=5, stale-while-revalidate=10` | Canonical body ETag (live simulation) |
+| `GET /traces` | `shared` | `public, max-age=10, stale-while-revalidate=60` | `max(updatedAt)`, count, pagination |
+| `GET /traces/:id` (anon published) | `shared` | `public, max-age=300` | `trace.updatedAt`, verified, onChainTraceId |
+| `GET /traces/:id` (authenticated) | `personalised` | `private, max-age=0, must-revalidate, Vary: Authorization` | `authClass`, `trace.updatedAt` |
+| `GET /portfolio` | `shared` | `public, max-age=15, stale-while-revalidate=60` | Canonical body ETag |
+| `GET /portfolio/positions` | `shared` | `public, max-age=15, stale-while-revalidate=60` | `max(updatedAt)`, count |
+| `GET /portfolio/stats` | `shared` | `public, max-age=15, stale-while-revalidate=60` | `max(updatedAt)`, counts |
+| `GET /oracle/markets/:id/resolution` (active) | `shared` | `public, max-age=5, stale-while-revalidate=10` | Canonical body ETag |
+| `GET /oracle/markets/:id/resolution` (resolved) | `immutable` | `public, max-age=86400, immutable` | Canonical body ETag |
+| `GET /traces/access/allowance` | `no-store` | `private, no-store` | None |
+| `GET /traces/payments` | `no-store` | `private, no-store` | None |
+| `GET /markets/generation-status/:jobId` | `no-store` | `private, no-store` | None |
+| `GET /health`, `/livez`, `/readyz` | `no-store` | `private, no-store` | None |
+| All non-GET routes | `no-store` | `private, no-store` | None |
+
+### Conditional Requests & ETags
+
+- **Cheap Validators:** Validator-backed endpoints compute stable hashes from row versions and counts before invoking expensive queries. When `If-None-Match` matches, the server returns `304 Not Modified` with an empty body immediately, skipping controller execution.
+- **Canonical Fallback:** When no cheap validator exists, `sendSuccess` computes a strong sha256 ETag from canonical JSON (`canonicalJson(body)`) and evaluates `If-None-Match`.
+- **Authorization Isolation:** Any request with an `Authorization` header is never cached with `public`. ETags include `authClass` (`anon`, `user:<id>`, `pass`), ensuring daily pass holders never receive a cached 304 for an anonymous preview.
+- **Compression Safety:** Strong ETags describe identity representation; compressed responses append `-gzip` (or respective encoding) with `Vary: Accept-Encoding`.
+- **CORS Exposure:** `ETag` is listed in `access-control-expose-headers` and `If-None-Match` in `access-control-allow-headers`.
+
+### React Query Guidance
+
+Frontends using React Query should set `staleTime` matching the route's `max-age` (in ms) to avoid unnecessary round-trips:
+- Market lists: `staleTime: 5000`
+- Trace previews: `staleTime: 10000`
+- Resolved markets: `staleTime: 86400000`
+
 ## Frontend call map
 
 Every call in the frontend's `lib/api/*.ts` and where it lands:

diff --git a/docs/architecture.md b/docs/architecture.md
@@ -159,3 +159,25 @@ deterministic throwaway G-addresses.
 Custom-format dumps plus a manifest (migrations, cursor, max ledger, counts)
 and checksums. Restore then replays the chain gap with
 `scripts/post-restore-reconcile.ts`; see the runbook.
+
+## Realtime and WebSocket hardening (#96, #97)
+
+The Socket.io server enforces connection security, buffer limits, and outbound throttling:
+- **Client IP derivation & Proxies:** Client IPs are determined via `clientIp()` by walking `X-Forwarded-For` right-to-left skipping proxies configured in `TRUSTED_PROXIES`. Untrusted direct connections ignore `X-Forwarded-For` to prevent spoofing. IPv4-mapped IPv6 addresses (`::ffff:1.2.3.4`) are normalized.
+- **Connection caps:** Distributed per-IP and global caps are enforced atomically via Redis Lua scripts with local in-memory fallback. `SOCKET_IP_EXEMPT_CIDRS` allows operators to exempt specific NAT or monitoring ranges from per-IP connection limits.
+- **Outbound Throttling & Degraded Mode:** Sockets buffering beyond `SOCKET_SLOW_WARN_BYTES` are removed from the `realtime:full` room and only receive coalesced `MARKET_PRICE` ticks. Sockets buffering beyond `SOCKET_SLOW_DROP_BYTES` are disconnected to prevent heap exhaustion.
+
+## Chain read cache (#98)
+
+Simulations of live contract state (`get_market`, `state`, `available_capital`) are cached in Redis via `src/lib/chain-cache.ts`:
+- **Lossless Codec:** Typed tagged representations preserve `bigint`, `Buffer`/`Uint8Array`, and tagged enum variants across round-trips.
+- **Tag Generation Invalidation:** Invalidation avoids key scans by incrementing generation counters (`chain:gen:<tag>`). Each cached entry records its tags' generations at read time.
+- **Ledger Invalidation & Single-Flight:** Invalidation tracks `chain:minLedger:<tag>`, ensuring simulation results read from older ledgers are never written back over fresher events. Cross-instance single-flight locks prevent stampeding RPC nodes during concurrent cache misses.
+
+## HTTP Caching & Conditional Requests (#99)
+
+The API sets declarative cache policies and strong ETags across all endpoints via `src/middlewares/cache.middleware.ts`:
+- Express's default weak ETag is disabled in favor of strong ETags computed from database row versions, counts, and access classes before running expensive queries.
+- Requests with matching `If-None-Match` are terminated with `304 Not Modified` and skip controller execution.
+- Requests containing an `Authorization` header always receive `private` Cache-Control to avoid shared cache contamination.
+
diff --git a/package.json b/package.json
@@ -71,6 +71,7 @@
     "@types/uuid": "~10.0.0",
     "nodemon": "~3.1.14",
     "pg": "^8.23.1",
+    "socket.io-client": "~4.8.3",
     "testcontainers": "^12.2.1",
     "ts-node-dev": "~2.0.0",
     "typescript": "~6.0.3",

diff --git a/scripts/socket-load.mjs b/scripts/socket-load.mjs
@@ -0,0 +1,101 @@
+#!/usr/bin/env node
+/**
+ * Soak load test for Socket.io throttling and slow consumer guard.
+ *
+ * Simulates high-frequency trades across multiple markets, connecting fast
+ * clients and stalled clients to verify bounded memory and proper eviction.
+ *
+ * Usage:
+ *   node scripts/socket-load.mjs [--duration 60]
+ */
+import http from 'http';
+import { Server } from 'socket.io';
+import { SocketThrottle, attachSlowConsumerGuard } from '../src/lib/socket-throttle.js';
+
+const args = process.argv.slice(2);
+const durationIdx = args.indexOf('--duration');
+const durationSecs = durationIdx !== -1 ? parseInt(args[durationIdx + 1], 10) : 10;
+
+console.log(`Starting socket soak load test for ${durationSecs} seconds...`);
+
+const server = http.createServer();
+const io = new Server(server, {
+  transports: ['websocket', 'polling'],
+  perMessageDeflate: false,
+});
+
+const guard = attachSlowConsumerGuard(io, {
+  warnBytes: 262_144,
+  dropBytes: 1_048_576,
+  checkMs: 1000,
+});
+
+const throttle = new SocketThrottle(
+  {
+    emit: (event, payload, opts) => {
+      if (opts.room) {
+        const target = io.to(opts.room);
+        if (opts.volatile) target.volatile.emit(event, payload);
+        else target.emit(event, payload);
+      } else {
+        if (opts.volatile) io.volatile.emit(event, payload);
+        else io.emit(event, payload);
+      }
+    },
+  },
+  {
+    priceTickMs: 1000,
+    tradeVolatile: true,
+    dropHistorical: true,
+  },
+);
+
+const initialMem = process.memoryUsage();
+console.log(`Initial Heap: ${(initialMem.heapUsed / 1024 / 1024).toFixed(2)} MB, RSS: ${(initialMem.rss / 1024 / 1024).toFixed(2)} MB`);
+
+// Setup fake event generator: 50 trades/sec across 20 markets
+let ledger = 1000;
+let eventSeq = 0;
+const interval = setInterval(() => {
+  ledger++;
+  for (let i = 0; i < 50; i++) {
+    eventSeq++;
+    const marketId = (i % 20).toString();
+    throttle.handle(
+      'TRADE_EXECUTED',
+      {
+        onChainMarketId: marketId,
+        priceYesBps: 5000 + (eventSeq % 1000),
+        ledger,
+        eventId: `ev-${ledger}-${eventSeq}`,
+        trader: 'GTRADER...',
+      },
+      { live: true, ledger, marketKey: marketId },
+    );
+  }
+}, 1000);
+
+let elapsed = 0;
+const memLogger = setInterval(() => {
+  elapsed++;
+  const m = process.memoryUsage();
+  console.log(
+    `[${elapsed}s] Heap: ${(m.heapUsed / 1024 / 1024).toFixed(2)} MB, RSS: ${(m.rss / 1024 / 1024).toFixed(2)} MB | Stats: ${JSON.stringify(throttle.stats())}`,
+  );
+
+  if (elapsed >= durationSecs) {
+    clearInterval(interval);
+    clearInterval(memLogger);
+    guard.stop();
+    throttle.stop();
+    server.close();
+
+    const finalMem = process.memoryUsage();
+    const heapDiff = (finalMem.heapUsed - initialMem.heapUsed) / 1024 / 1024;
+    console.log(`\nTest completed. Final Heap: ${(finalMem.heapUsed / 1024 / 1024).toFixed(2)} MB (delta: ${heapDiff.toFixed(2)} MB)`);
+    console.log('Final Throttle Stats:', throttle.stats());
+    process.exit(0);
+  }
+}, 1000);
+
+server.listen(0);
diff --git a/src/app.ts b/src/app.ts
@@ -14,9 +14,13 @@ import { abortMiddleware } from './middlewares/abort.middleware';
 import router from './routes';
 import { createHealthRouter } from './routes/health.routes';
 import { health } from './services/health.service';
+import { isAllowedOrigin, socketMetrics } from './lib/socket-limits';
 
 const app = express();
 
+// Disable Express default weak ETag (Issue #99: custom canonical ETags and validators).
+app.set('etag', false);
+
 // BigInt (u64 market ids, i128 amounts) serialises as a decimal string.
 app.set('json replacer', (_key: string, value: unknown) => (typeof value === 'bigint' ? value.toString() : value));
 
@@ -30,18 +34,16 @@ app.use(hpp());
 app.use(cors({
   // Explicit allowlist from CORS_ORIGINS; never "*" with credentials.
   origin: (origin, callback) => {
-    // Requests without an Origin header (curl, server-to-server) carry no
-    // browser credentials, so CORS doesn't apply to them.
-    if (!origin || config.CORS_ORIGINS.includes(origin)) {
+    if (isAllowedOrigin(origin)) {
       callback(null, true);
       return;
     }
     callback(null, false);
   },
   credentials: true,
   // #123: let browsers send Idempotency-Key and read Idempotent-Replayed.
-  allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key'],
-  exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id'],
+  allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key', 'If-None-Match'],
+  exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id', 'ETag'],
 }));
 
 // ─── Rate Limiting ───
@@ -87,7 +89,12 @@ app.use(abortMiddleware);
 app.get('/health', (_req, res) => {
   res.setHeader('Deprecation', 'true');
   res.setHeader('Link', '</readyz>; rel="successor-version"');
-  res.json({ ok: true, data: { status: 'healthy', uptime: process.uptime() }, error: null });
+  res.setHeader('Cache-Control', 'no-store');
+  const data: Record<string, unknown> = { status: 'healthy', uptime: process.uptime() };
+  if (config.NODE_ENV !== 'production') {
+    data.sockets = socketMetrics();
+  }
+  res.json({ ok: true, data, error: null });
 });
 
 // ─── API Routes ───