Files
autofirmer-expanded/lib/outbound-proxy.ts
T
Brandon LiandClaude Opus 5.5 b932452e98 Self-healing broker clients, per-firm scheduler isolation, eager boot
Why the autotrader sometimes broke after a weekend until `pm2 restart`:
`next start` only logs uncaught errors, so nothing ever crashed, and a few
paths could leave a client dead for the life of the process.

Tradovate: a closed socket scheduled exactly one re-login, with no catch —
one failure during weekend maintenance (or a rate-limit penalty returned as
a 200 without a token) and the client never came back. connectAndAuth is now
a single-flight loop that never throws: jittered backoff capped at 5 min,
penalty tickets waited out, captcha demands backed off. A 15s watchdog
abandons silent, never-authorized or stuck-closing sockets and stale syncs;
renewal falls back to a fresh login; orders fail fast (OrderNotPlacedError)
on a dead socket and time out instead of hanging; a bad sync answer no
longer wipes accounts and positions.

TopstepX: hub start/stop/invoke and every REST call are bounded; a wedged
hub is discarded and rebuilt; a stuck sync is superseded after 3 min;
re-login no longer tears down the sync loop; a watchdog re-logs when syncs
go stale; a no-fill timeout cancels the order.

Scheduler: one unready client used to hold every firm forever. It now holds
the tick for a 10-minute grace, then trades the other firms without it.
Per-firm try/catch in batch collection, one batch at a time, accounts marked
in flight, legs un-counted only when the entry never reached the broker, and
the scheduler starts before the first batch fires.

Boot and supervision: instrumentation.ts boots the process at server start
(it used to wait for the first dashboard request); a supervisor rebuilds a
client that has been unhealthy for 20 min (not a TopstepX one holding a
position); broker HTTP calls default to a 60s timeout; GET /api/health shows
each client's liveness and the scheduler's last decision. Shared state the
eager boot would otherwise split across bundles lives on globalThis. Logs no
longer carry access tokens or whole AxiosErrors (which include the encoded
Tradovate password).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-09-23 20:18:33 -05:00

200 lines
9.3 KiB
TypeScript

/**
* Routes BROKER traffic through the operator's proxy — and nothing else.
*
* Scope is deliberate: TopstepX REST + its SignalR hub, Tradovate REST + its
* websocket, and the Yahoo quote lookups go through the proxy so each instance
* can present its own IP to the firms. The master link, reporter, catalog sync
* and backup push stay direct — the control channel must keep working when the
* proxy dies, and there is no setGlobalDispatcher here for exactly that
* reason. (Corollary: never call the master through axios — axios defaults
* carry the proxy agents. Master calls use fetch, which stays untouched.)
*
* Configured by the `proxy_url` setting: http://, https://, socks5:// or
* socks5h:// with optional user:pass. Read once per process — changing it
* requires a restart, which the settings UI says out loud. Blank = direct.
*
* Also home of the broker HTTP timeout (applyBrokerHttpDefaults): axios waits
* forever by default, and one broker request that never answers — a
* maintenance window that accepts the connection and goes silent — used to
* wedge a re-login or sync await for the life of the process. Same scoping
* argument as the proxy: every axios call here is broker/quote traffic, so the
* process default is the scoped choice. It covers axios only; websockets and
* the SignalR hub keep their own liveness handling in the clients.
*
* Both defaults are applied when this module loads. Every broker client
* imports it, and that matters: each server bundle (instrumentation.ts, the
* route handlers) can carry its own copy of axios, and a copy that never had
* the defaults applied would dial the brokers direct, with no timeout.
*/
import axios from 'axios';
import { ProxyAgent, type Dispatcher } from 'undici';
import { socksDispatcher } from 'fetch-socks';
import { HttpProxyAgent } from 'http-proxy-agent';
import { HttpsProxyAgent } from 'https-proxy-agent';
import { SocksProxyAgent } from 'socks-proxy-agent';
import type { Agent } from 'node:http';
import { getSetting } from './db';
const SCHEMES = ['http:', 'https:', 'socks5:', 'socks5h:'] as const;
interface ProxyState {
url: URL | null;
dispatcher: Dispatcher | null;
httpAgent: Agent | null;
httpsAgent: Agent | null;
}
// HMR-safe, and the "read once per process" contract lives here.
const g = globalThis as typeof globalThis & { __outboundProxy?: ProxyState };
function state(): ProxyState {
if (g.__outboundProxy) return g.__outboundProxy;
const s: ProxyState = { url: null, dispatcher: null, httpAgent: null, httpsAgent: null };
g.__outboundProxy = s;
const raw = getSetting('proxy_url')?.trim();
if (!raw) return s;
let url: URL;
try {
url = new URL(raw);
} catch {
// Not echoed: the value carries the proxy's user:password.
console.error('[proxy] proxy_url is not a URL — trading DIRECT');
return s;
}
if (!SCHEMES.includes(url.protocol as (typeof SCHEMES)[number])) {
console.error(`[proxy] proxy_url scheme must be one of ${SCHEMES.join(' ')} — trading DIRECT: ${url.protocol}`);
return s;
}
// A literal '%' in the user or password (one not starting a %XX escape)
// makes decodeURIComponent throw — here, and inside every proxy library
// below. Escape it so the password is used as typed. This now runs when
// the module loads, where a throw would fail every route that imports it.
const escapeStray = (part: string) => part.replace(/%(?![0-9A-Fa-f]{2})/g, '%25');
url.username = escapeStray(url.username);
url.password = escapeStray(url.password);
try {
const socks = url.protocol.startsWith('socks');
if (socks) {
const dispatcher = socksDispatcher({
type: 5,
host: url.hostname,
port: Number(url.port || 1080),
userId: url.username ? decodeURIComponent(url.username) : undefined,
password: url.password ? decodeURIComponent(url.password) : undefined,
});
const agent = new SocksProxyAgent(url);
s.dispatcher = dispatcher;
s.httpAgent = agent;
s.httpsAgent = agent;
} else {
const dispatcher = new ProxyAgent(url.href);
const httpAgent = new HttpProxyAgent(url);
const httpsAgent = new HttpsProxyAgent(url);
s.dispatcher = dispatcher;
s.httpAgent = httpAgent;
s.httpsAgent = httpsAgent;
}
} catch (err) {
// All or nothing — nothing is kept until every piece is built: half a
// proxy (REST proxied, sockets direct) would be worse than the
// documented bad-setting behavior.
console.error(`[proxy] proxy_url could not be set up — trading DIRECT: ${err instanceof Error ? err.message : String(err)}`);
return s;
}
s.url = url;
console.log(`[proxy] broker traffic will use ${url.protocol}//${url.host} (master link stays direct)`);
return s;
}
export function getProxyUrl(): string | null {
const { url } = state();
return url ? url.href : null;
}
/** Puts the proxy under every axios call. All axios usage in this app is
* broker/quote traffic (TopstepX, Tradovate REST, contract resolver), so the
* process-wide default is the scoped choice, not a global one. Idempotent;
* runs when this module loads and again first in the getClients() bootstrap.
*
* "Already applied" is judged on the axios defaults themselves, not a
* process-wide flag: the proxy state is shared through globalThis, but each
* bundle's copy of axios has its own defaults, and a global flag set by the
* first copy would leave every other one dialling direct. */
export function applyBrokerProxy(): void {
const s = state();
if (!s.url || axios.defaults.httpsAgent === s.httpsAgent) return;
axios.defaults.proxy = false; // agents own the routing; ignore env HTTP(S)_PROXY handling
axios.defaults.httpAgent = s.httpAgent!;
axios.defaults.httpsAgent = s.httpsAgent!;
}
/** Idle limit for one broker HTTP request. axios applies it as a socket
* inactivity timeout, so it bounds a server that has gone silent, not a
* big report that is still streaming. Nothing legitimate comes close: the
* Tradovate report endpoints answer with a p-ticket and are polled (each
* poll its own short request, already capped at 60s per call), logins and
* REST calls answer in seconds. A per-call `timeout` still overrides it.
*
* The clock starts once the socket has connected — a connect that never
* completes is left to the OS connect timeout. Calls that can't wait that
* long (the Tradovate login and renewal) add an AbortSignal of their own. */
export const BROKER_HTTP_TIMEOUT_MS = 60_000;
/** Gives axios a timeout unless one is already set (axios's own default, 0,
* means wait forever). Idempotent; runs when this module loads and at the
* top of the getClients() bootstrap. Timeouts surface as ECONNABORTED, which
* the Tradovate report fetch already retries as transient. */
export function applyBrokerHttpDefaults(): void {
if (!axios.defaults.timeout) axios.defaults.timeout = BROKER_HTTP_TIMEOUT_MS;
}
// At load, so no broker request from this bundle can go out before them. The
// proxy is skipped during `next build`: the build imports route modules to
// collect page data, and it dials nothing that would need one. Nothing may
// throw out of here: a module that fails to load takes every route that
// imports it (the whole trading side) down with it.
applyBrokerHttpDefaults();
try {
if (process.env.NEXT_PHASE !== 'phase-production-build') applyBrokerProxy();
} catch (err) {
console.error(`[proxy] could not apply proxy_url at load: ${err instanceof Error ? err.message : String(err)}`);
}
/** A websocket to a BROKER endpoint: the built-in (undici) WebSocket, dialled
* through the proxy when one is set. The master link keeps using plain
* `new WebSocket(...)` — do not use this for it. */
export function brokerWebSocket(wsUrl: string): WebSocket {
const { dispatcher } = state();
if (!dispatcher) return new WebSocket(wsUrl);
// The runtime accepts { dispatcher } fine; only the DOM lib type doesn't.
return new (WebSocket as unknown as new (u: string, o: object) => WebSocket)(wsUrl, { dispatcher });
}
/** A `ws`-package constructor with the proxy agent baked in, for SignalR's
* `withUrl({ WebSocket })` option. undefined = no proxy, let SignalR use its
* own require('ws'). */
export function signalrWebSocket(): unknown {
const s = state();
if (!s.httpsAgent) return undefined;
const agent = s.httpsAgent;
// eslint-disable-next-line @typescript-eslint/no-require-imports
const WS = require('ws') as new (address: string, protocols?: unknown, options?: object) => unknown;
return class ProxiedWebSocket extends (WS as new (...args: never[]) => object) {
constructor(address: string, protocols?: unknown, options?: Record<string, unknown>) {
// ws also allows (address, options) — normalize before merging.
if (protocols && typeof protocols === 'object' && !Array.isArray(protocols)) {
options = protocols as Record<string, unknown>;
protocols = undefined;
}
super(...[address, protocols, { ...(options ?? {}), agent }] as never[]);
}
};
}