From 15dd846b4e620178f224867a5541a54ac1597fc3 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 17:46:02 -0600 Subject: [PATCH 1/7] Export symphony metrics from the standalone server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit symphony's counters were only reachable through the napi `metrics()` call, so a deployment running `symphony-server` as its own process — which is how host-manager supervises it — had no way to read them at all. The per-listener counters were worse off: they were incremented on every hot path and then never read by anything. Adds an out-of-process export path, and broadens what there is to export. Counters (`src/metrics.rs`): - Per-listener counters are now surfaced through `metrics()`, one entry per configured listener with its address and mode. - `total_errors` was a single bucket for TLS handshake failure, upstream connect failure, idle timeout and copy error — four different alerts collapsed into one number. Errors and blocks are now broken out by reason, using a fixed-size array indexed by enum discriminant so the hot path stays allocation-free. A new `protection::BlockReason` variant fails to compile until it is mapped to a `BlockKind`, so a new check can't land in an unlabeled bucket. - Byte counters, via a `CountingStream` wrapper on the client side. Counting `copy_bidirectional`'s return value instead would lose every byte of any session that ends by idle timeout or reset — i.e. most long-lived ones. - Suspended connections now record how they ended (resolved vs timed out), and the route table exposes its live and cert-failing route counts. - `forward()` reports an idle timeout as its own kind rather than synthesising an `io::ErrorKind::TimedOut` for the caller to infer from — a peer's kernel-level ETIMEDOUT produces the same kind and would have been misfiled. - A listener-level `maxConnections` rejection now increments the proxy-wide blocked total as well as the per-listener one, so the proxy total is once again the sum across its listeners. Export (`ts/admin.ts`): - An optional `admin` block in the config file exposes `GET /metrics` (Prometheus text), `/metrics.json` and `/health` over a Unix socket, a loopback TCP port, or both. Omitted by default — nothing is exposed unless configured. - Blocked/error counts are emitted only under their `reason` label; the labelled series sum to the total by construction, so a separate unlabelled metric would be a second representation of the same number. - The endpoint never affects proxying: a bind failure is logged and retried on a timer rather than aborting a reconcile. This is load-bearing during a version upgrade, where the incumbent still holds the socket while the successor is already serving traffic through SO_REUSEPORT. - A socket file left by a SIGKILLed process is reclaimed, but only after a connect probe proves nobody is listening — unlinking unconditionally would let a starting process silently steal the endpoint from the running one, the same failure mode the status.json ownership guard exists to prevent. Tests cover the per-listener breakdown and exact byte counts, each error classification, the Prometheus output shape, the endpoint over both transports, and stale-socket reclaim after a SIGKILL. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 27 +- README.md | 104 +++++++- __test__/metrics.spec.ts | 523 +++++++++++++++++++++++++++++++++++++++ package.json | 2 +- src/http_listener.rs | 49 ++-- src/listener.rs | 6 +- src/metrics.rs | 267 +++++++++++++++++++- src/proxy.rs | 73 ++++++ src/proxy_conn.rs | 65 +++-- src/router.rs | 11 + ts/addon.d.ts | 30 +++ ts/admin.ts | 406 ++++++++++++++++++++++++++++++ ts/index.ts | 4 + ts/proxy.ts | 16 ++ ts/server.ts | 30 ++- ts/types.ts | 32 +++ 16 files changed, 1587 insertions(+), 58 deletions(-) create mode 100644 __test__/metrics.spec.ts create mode 100644 ts/admin.ts diff --git a/CLAUDE.md b/CLAUDE.md index 71434e5..882a602 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -12,6 +12,18 @@ symphony is a **napi-rs cdylib** loaded by Node.js. The tokio multi-thread runti For consumers that want symphony as its own OS process rather than embedded in their Node app, the package ships a `symphony-server` bin. It reads a JSON config file (`{ version, proxies: [{ listeners, routes }] }` — one entry per port-set, since the route table is per-proxy), constructs a `SymphonyProxy` per entry, and **watches the config file** to hot-reload (route change → `updateConfig`; listener change → recreate that proxy). Cert material may be given inline (`certChain`/`privateKey`) or by path (`certChainFile`/`privateKeyFile`) — the path form is resolved in `server.ts` only, so the napi `CertConfig` stays inline-only. It writes a `status.json` (`{ pid, version, ports, ... }`) for supervisors, and handles `SIGHUP` (reload) / `SIGTERM`/`SIGINT` (graceful stop). host-manager uses this to supervise symphony out-of-process. +### Admin/metrics endpoint (`ts/admin.ts`) + +An optional `admin` block in the config file (`{ socketPath?, socketMode?, port?, host? }`) makes `symphony-server` expose `GET /metrics` (Prometheus text), `/metrics.json`, and `/health` over a Unix socket, a loopback TCP port, or both. It exists because an out-of-process symphony has no reachable napi `metrics()` — the endpoint is the only export path for that deployment. + +Three properties are load-bearing and easy to regress: + +- **It must never affect proxying.** A bind failure is logged and retried on a 5s timer instead of throwing out of `doReconcile()`. This is not defensiveness for its own sake: during a version upgrade host-manager runs both processes concurrently (the Rust listeners overlap via `SO_REUSEPORT`, which a Node HTTP server has no equivalent for), so the successor *will* lose the admin bind for a few seconds and must keep serving traffic anyway. +- **A stale Unix socket is reclaimed, a live one is not.** `EADDRINUSE` on a socket path triggers a connect probe; only `ECONNREFUSED` (nobody listening) justifies the `unlink`. Unlinking unconditionally would let a starting process silently steal the endpoint from the running one — the same class of bug as the `status.json` ownership guard. +- **Counters are read per request**, not cached at reconcile, so a scrape never serves numbers frozen at the last config reload. + +Prometheus shape: blocked/error counts are emitted **only** under their `reason` label (they sum to the unlabeled total, so a separate total would be a second representation of the same number), and the proxy-wide active gauge is `sum without(listener)`. `renderPrometheus` is exported from the package for embedded consumers. + The server also **watches the cert/key files referenced by the config** (grouped by parent dir, deduped, re-derived on every reconcile so watchers don't leak) → a debounced `reconcile()` on change, so an on-disk cert renewal is picked up live without a `config.json` write or restart. Two details make a listener-level cert rotation actually apply: the per-proxy `listenerSig` is computed over the *resolved* listeners (cert contents included), so a rotated `defaultCert`/mTLS file changes the signature and forces a recreate rather than a route-only hot-swap against the frozen `default_listener_tls`. Basename-filtered dir watching handles in-place / rename rotation (what host-manager does); k8s projected-volume `..data` symlink swaps are not yet covered. Cert-failure resilience lives in `router.rs::build_route_table`: a route whose cert can't be built (e.g. rustls `KeyMismatch` from a rotated key vs a stale inlined chain) is isolated — one bad tenant cert never aborts the whole table; on a hot-swap the last-good route is carried forward for that SNI (mid-rotation the old cert is still valid), and on initial build the SNI is simply dropped. ### Data flow @@ -46,7 +58,7 @@ TCP accept (SO_REUSEPORT per worker thread) | `src/proxy_conn.rs` | Per-connection handler: the full 7-step flow | | `src/protection.rs` | IP rate limiting, concurrency, CIDR lists, JA3 blocking | | `src/suspended.rs` | Pending-connection registry (DashMap + oneshot channels) | -| `src/metrics.rs` | AtomicU64 counters: active, accepted, errors, blocked | +| `src/metrics.rs` | AtomicU64 counters (active, accepted, bytes, per-reason blocks/errors) + `CountingStream` | | `src/error.rs` | SymphonyError enum → napi::Error conversion | --- @@ -151,6 +163,18 @@ napi `Buffer` contains raw pointers (`*mut napi_env__`, `*mut napi_ref__`) that 4. Add a new `kind` case in `parse_upstream_spec()` in `proxy.rs` 5. Add a test +### Adding a new metric + +1. Add the counter to `ListenerMetrics`/`GlobalMetrics` in `metrics.rs`, or a variant to the + `labeled_enum!` block for `BlockKind`/`ErrorKind` — the variant list drives the counter array, + the label, and the export, so there is no second list to update. +2. Increment it at the call site. A new `protection::BlockReason` variant will fail to compile + until `From<&BlockReason> for BlockKind` maps it — that is deliberate, so a new protection + check can't land in an unlabeled bucket. +3. Surface it in `JsProxyMetrics`/`JsListenerMetrics` (`proxy.rs`), `ts/types.ts`, and the + mapping in `ts/proxy.ts`. +4. Add the sample to `renderPrometheus` in `ts/admin.ts`, and a case in `__test__/metrics.spec.ts`. + ### Adding a new napi method 1. Implement in `proxy.rs` with `#[napi]` @@ -169,6 +193,7 @@ Tests live in `__test__/` and use Node's built-in `node:test` runner. - **`protection.spec.ts`** — rate limit token bucket exhaustion, CIDR blocklist in `blockedIps()` - **`suspended.spec.ts`** — hold → resolve → proxy, hold → null → close, hold → timeout → drop - **`mtls.spec.ts`** — mTLS termination + PROXY v2 TLV forwarding of the client cert chain (0xE2, SSL TLV 0x20); skips without openssl +- **`metrics.spec.ts`** — per-listener breakdown and byte counting, `renderPrometheus` output shape, the admin endpoint over UDS + TCP, and stale-socket reclaim after a `SIGKILL` Build and run: ```bash diff --git a/README.md b/README.md index 7b563ee..e1314fc 100644 --- a/README.md +++ b/README.md @@ -599,21 +599,113 @@ attestation roadmap, and the shared-responsibility split for DDoS — see ## Metrics & monitoring +### In-process (`proxy.metrics()`) + ```typescript const m = proxy.metrics(); -// m.activeConnections — connections being proxied right now -// m.blockedConnections — total blocked since start -// m.pendingSuspended — connections currently held waiting for resolveConnection() +// Proxy-wide +// m.activeConnections — connections being proxied right now +// m.blockedConnections — total rejected since start (protection + maxConnections) +// m.pendingSuspended — connections held waiting for resolveConnection() +// m.suspendedResolved — suspended connections that were resolved with a route +// m.suspendedUnresolved — suspended connections that timed out or were rejected +// m.routes — routes in the live table, including the default route +// m.failingRoutes — routes whose cert failed to build (see "Per-route certificates") + +// Per listener, in configuration order +for (const l of m.listeners) { + // l.address, l.mode ('tls' | 'http') + // l.activeConnections, l.accepted + // l.bytesReceived — bytes read from clients (client → upstream) + // l.bytesSent — bytes written to clients (upstream → client) + // Counted where the proxy sees the bytes: plaintext on a terminated-TLS route, wire bytes + // on a passthrough route. The handshake is not included in either. + // l.blockedByReason — [{ reason: 'rate_limited', count: 12 }, ...] + // l.errorsByReason — [{ reason: 'upstream_connect', count: 3 }, ...] +} const blocked = proxy.blockedIps(); // blocked.rateLimited — IPs with a depleted per-second or sustained token bucket // blocked.concurrencyLimited — IPs at their maxConcurrentPerIp limit // blocked.cidrBlocklist — the configured static CIDR blocklist // blocked.penaltyBoxed — IPs currently in the penalty box +``` + +Every reason is reported on every call, including reasons still at zero, so a dashboard series +exists before the first incident rather than appearing mid-outage. The per-reason counts sum to +`l.blocked` / `l.errors` by construction. + +**Block reasons:** `max_connections`, `cidr_blocked`, `ja3_blocked`, `ja4_blocked`, +`incomplete_handshake`, `no_sni`, `rate_limited`, `too_many_connections`, `penalty_boxed`. + +**Error reasons:** `no_route`, `route_rate_limited`, `suspend_unresolved`, `tls_handshake`, +`tls_missing_cert`, `upstream_connect`, `idle_timeout`, `stream`, `http_header`. + +### Out-of-process (`symphony-server` admin endpoint) + +When symphony runs as its own process there is no JS API to call, so the server bin can expose +the same numbers over HTTP. Add an `admin` block to the config file: + +```json +{ + "version": 1, + "admin": { + "socketPath": "/run/symphony/admin.sock", + "socketMode": 432, + "port": 9095, + "host": "127.0.0.1" + }, + "proxies": [ ... ] +} +``` + +Both bindings are optional; give either or both. Omit the `admin` block entirely and nothing is +exposed. `socketPath` may be relative to the config file's directory, and is chmodded to +`socketMode` (default `0o660`) after bind. `host` defaults to `127.0.0.1` — metrics carry no +tenant identifiers, but there is still no reason to publish them off-box. + +| Route | Response | +|---|---| +| `GET /metrics` | Prometheus text exposition (v0.0.4) | +| `GET /metrics.json` | the same snapshot as JSON | +| `GET /health` | `{ ok, pid, version, ports }` | + +``` +$ curl --unix-socket /run/symphony/admin.sock http://localhost/metrics +# HELP symphony_build_info Always 1; the version is carried in the label. +# TYPE symphony_build_info gauge +symphony_build_info{version="0.5.0"} 1 +... +symphony_listener_accepted_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 148213 +symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="rate_limited"} 27 +symphony_listener_errors_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="upstream_connect"} 4 +``` + +The `proxy` label is the port-set of the proxy entry the listener belongs to (each config entry +gets its own route table). Blocked and error counts are only ever emitted with their `reason` +label — the labelled series sum to the total, so use `sum without(reason)` rather than looking +for a separate unlabelled metric. Likewise the proxy-wide active-connection gauge is +`sum without(listener) (symphony_listener_active_connections)`. + +The endpoint is strictly read-only and best-effort: it never blocks proxying, and a bind failure +is logged and retried every 5s rather than aborting startup. That matters during a version +upgrade, where the incumbent still holds the socket while the replacement is already serving +traffic through `SO_REUSEPORT` — the successor picks up the admin endpoint once the old process +exits. A socket file left behind by a `SIGKILL`ed process is reclaimed automatically, but only +after a connect probe proves nobody is listening on it. -setInterval(() => { - console.log('active:', proxy.metrics().activeConnections); -}, 10_000); +The renderer is exported for consumers that want the same output from an embedded proxy: + +```typescript +import { renderPrometheus } from '@harperfast/symphony'; + +const text = renderPrometheus({ + pid: process.pid, + version: '0.5.0', + startedAt, + reloadedAt, + proxies: [{ ports: '80,443', metrics: proxy.metrics() }], +}); ``` --- diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts new file mode 100644 index 0000000..2eec88e --- /dev/null +++ b/__test__/metrics.spec.ts @@ -0,0 +1,523 @@ +/** + * Metrics coverage: the in-process `metrics()` breakdown, the Prometheus renderer, and the + * standalone server's admin endpoint over both a Unix socket and a loopback port. + * + * Requires the native addon to be built (npm run build:debug). + */ + +import assert from 'node:assert/strict'; +import { after, before, describe, it } from 'node:test'; +import { spawn, ChildProcess } from 'node:child_process'; +import * as fs from 'node:fs'; +import * as http from 'node:http'; +import * as os from 'node:os'; +import * as path from 'node:path'; +import * as tls from 'node:tls'; +import { SymphonyProxy, renderPrometheus, type MetricsSnapshot } from '../ts/index.js'; +import { generateSelfSignedCert, getFreePort, startEchoServer, tlsRoundTrip, sleep } from './util.js'; + +const SERVER_JS = path.join(__dirname, '..', 'ts', 'server.js'); + +async function waitFor(predicate: () => boolean | Promise, timeoutMs = 5000, stepMs = 50): Promise { + const deadline = Date.now() + timeoutMs; + while (Date.now() < deadline) { + if (await predicate()) return; + await sleep(stepMs); + } + throw new Error('waitFor: timed out'); +} + +/** GET over a Unix socket or a TCP port. */ +function get( + target: { socketPath: string } | { port: number }, + urlPath: string, + method = 'GET' +): Promise<{ status: number; body: string; contentType: string }> { + return new Promise((resolve, reject) => { + const req = http.request({ ...target, path: urlPath, method }, (res) => { + let body = ''; + res.setEncoding('utf8'); + res.on('data', (c) => (body += c)); + res.on('end', () => + resolve({ status: res.statusCode ?? 0, body, contentType: String(res.headers['content-type'] ?? '') }) + ); + }); + req.on('error', reject); + req.end(); + }); +} + +function reasonCount(reasons: Array<{ reason: string; count: number }>, reason: string): number { + const match = reasons.find((r) => r.reason === reason); + assert.ok(match, `expected a '${reason}' entry; got ${reasons.map((r) => r.reason).join(', ')}`); + return match.count; +} + +describe('proxy.metrics()', () => { + const cert = generateSelfSignedCert('localhost'); + let echo: Awaited>; + let proxy: SymphonyProxy; + let tlsPort: number; + let httpPort: number; + + before(async () => { + echo = await startEchoServer(); + tlsPort = await getFreePort(); + httpPort = await getFreePort(); + proxy = new SymphonyProxy({ + listeners: [ + { host: '127.0.0.1', port: tlsPort }, + { host: '127.0.0.1', port: httpPort, mode: 'http' }, + ], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }); + await proxy.start(); + }); + + after(async () => { + await proxy.stop().catch(() => {}); + await echo.close().catch(() => {}); + }); + + it('reports one entry per configured listener, in order, with its mode', () => { + const m = proxy.metrics(); + assert.equal(m.listeners.length, 2); + assert.deepEqual( + m.listeners.map((l) => [l.address, l.mode]), + [ + [`127.0.0.1:${tlsPort}`, 'tls'], + [`127.0.0.1:${httpPort}`, 'http'], + ] + ); + }); + + it('reports the live route count', () => { + assert.equal(proxy.metrics().routes, 1); + assert.equal(proxy.metrics().failingRoutes, 0); + }); + + it('emits every block and error reason, including the ones still at zero', () => { + const listener = proxy.metrics().listeners[0]; + // A reason that only ever fires under protection config must still have a series. + assert.equal(reasonCount(listener.blockedByReason, 'rate_limited'), 0); + assert.equal(reasonCount(listener.errorsByReason, 'upstream_connect'), 0); + assert.ok(listener.blockedByReason.some((r) => r.reason === 'max_connections')); + }); + + it('counts bytes in both directions across a proxied session', async () => { + const before = proxy.metrics().listeners[0]; + const payload = Buffer.from('x'.repeat(4096)); + const echoed = await tlsRoundTrip({ port: tlsPort, servername: 'localhost', caCert: cert.cert, data: payload }); + assert.equal(echoed.length, payload.length); + + const after = proxy.metrics().listeners[0]; + assert.equal(after.accepted, before.accepted + 1); + // The counter wraps the client *after* the handshake, so on a terminated-TLS route it + // sees exactly the plaintext payload each way — no handshake and no record framing. + assert.equal(after.bytesReceived, before.bytesReceived + payload.length); + assert.equal(after.bytesSent, before.bytesSent + payload.length); + }); + + it('classifies an unroutable SNI as no_route rather than a generic error', async () => { + const before = proxy.metrics().listeners[0]; + // No route for this SNI and no default route → symphony drops the connection. + await tlsRoundTrip({ port: tlsPort, servername: 'nope.example.com', caCert: cert.cert, data: 'ping' }).catch( + () => undefined + ); + + await waitFor(() => proxy.metrics().listeners[0].errors > before.errors); + const after = proxy.metrics().listeners[0]; + assert.equal(reasonCount(after.errorsByReason, 'no_route'), reasonCount(before.errorsByReason, 'no_route') + 1); + // The per-reason series always sum to the unlabeled total. + assert.equal( + after.errorsByReason.reduce((sum, r) => sum + r.count, 0), + after.errors + ); + }); +}); + +// The three session outcomes are told apart by where the failure is raised, not by inspecting an +// io::ErrorKind — these lock that in, since a misclassification is invisible until an incident. +describe('proxy.metrics() error classification', () => { + const cert = generateSelfSignedCert('localhost'); + let echo: Awaited>; + let proxy: SymphonyProxy; + let tlsPort: number; + let deadPort: number; + + before(async () => { + echo = await startEchoServer(); + tlsPort = await getFreePort(); + // Reserved and then released, so a connect here is refused rather than hanging. + deadPort = await getFreePort(); + proxy = new SymphonyProxy({ + listeners: [{ host: '127.0.0.1', port: tlsPort, idleTimeoutMs: 300 }], + routes: [ + { + sni: 'idle.test', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + { + sni: 'dead.test', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: deadPort }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }); + await proxy.start(); + }); + + after(async () => { + await proxy.stop().catch(() => {}); + await echo.close().catch(() => {}); + }); + + it('records an unreachable upstream as upstream_connect', async () => { + const before = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'upstream_connect'); + await tlsRoundTrip({ port: tlsPort, servername: 'dead.test', caCert: cert.cert, data: 'ping' }).catch( + () => undefined + ); + await waitFor(() => reasonCount(proxy.metrics().listeners[0].errorsByReason, 'upstream_connect') > before); + }); + + it('records a session that goes quiet as idle_timeout, not a stream error', async () => { + const beforeIdle = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'idle_timeout'); + const beforeStream = reasonCount(proxy.metrics().listeners[0].errorsByReason, 'stream'); + + // Connect, complete the handshake, then send nothing until the 300ms idle timeout fires. + const socket = tls.connect({ + port: tlsPort, + host: '127.0.0.1', + servername: 'idle.test', + ca: cert.cert, + rejectUnauthorized: false, + }); + try { + await new Promise((resolve, reject) => { + socket.once('secureConnect', resolve); + socket.once('error', reject); + }); + socket.on('error', () => {}); // the proxy dropping us is the expected outcome + await waitFor(() => reasonCount(proxy.metrics().listeners[0].errorsByReason, 'idle_timeout') > beforeIdle, 5000); + } finally { + socket.destroy(); + } + + assert.equal( + reasonCount(proxy.metrics().listeners[0].errorsByReason, 'stream'), + beforeStream, + 'an idle timeout must not also be counted as a stream error' + ); + }); +}); + +describe('renderPrometheus', () => { + const snapshot: MetricsSnapshot = { + pid: 42, + version: '9.9.9', + startedAt: '2026-01-01T00:00:00.000Z', + reloadedAt: '2026-01-01T00:01:00.000Z', + proxies: [ + { + ports: '80,443', + metrics: { + activeConnections: 3, + blockedConnections: 2, + pendingSuspended: 1, + suspendedResolved: 5, + suspendedUnresolved: 4, + routes: 7, + failingRoutes: 1, + listeners: [ + { + address: '0.0.0.0:443', + mode: 'tls', + activeConnections: 3, + accepted: 10, + blocked: 2, + errors: 1, + bytesReceived: 1024, + bytesSent: 2048, + blockedByReason: [ + { reason: 'rate_limited', count: 2 }, + { reason: 'no_sni', count: 0 }, + ], + errorsByReason: [{ reason: 'upstream_connect', count: 1 }], + }, + ], + }, + }, + ], + }; + + const output = renderPrometheus(snapshot); + const lines = output.split('\n'); + + it('declares HELP and TYPE exactly once per metric name', () => { + const typeLines = lines.filter((l) => l.startsWith('# TYPE ')); + const names = typeLines.map((l) => l.split(' ')[2]); + assert.deepEqual([...new Set(names)].length, names.length, `duplicate TYPE declarations in:\n${output}`); + // Both `outcome` samples share one declaration. + assert.equal(typeLines.filter((l) => l.includes('symphony_suspended_total')).length, 1); + }); + + it('labels every proxy-scoped and listener-scoped sample', () => { + assert.ok(lines.includes('symphony_routes{proxy="80,443"} 7')); + assert.ok(lines.includes('symphony_routes_failing{proxy="80,443"} 1')); + assert.ok(lines.includes('symphony_suspended_total{proxy="80,443",outcome="resolved"} 5')); + assert.ok(lines.includes('symphony_suspended_total{proxy="80,443",outcome="unresolved"} 4')); + assert.ok(lines.includes('symphony_listener_accepted_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 10')); + assert.ok( + lines.includes('symphony_listener_bytes_received_total{proxy="80,443",listener="0.0.0.0:443",mode="tls"} 1024') + ); + }); + + it('emits blocked/error counts only under their reason label', () => { + assert.ok( + lines.includes( + 'symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="rate_limited"} 2' + ) + ); + // A zero-valued reason still gets a series, so it exists before the first incident. + assert.ok( + lines.includes( + 'symphony_listener_blocked_total{proxy="80,443",listener="0.0.0.0:443",mode="tls",reason="no_sni"} 0' + ) + ); + // No unlabeled duplicate of the same number. + assert.ok(!lines.some((l) => /^symphony_listener_blocked_total\{[^}]*\}\s/.test(l) && !l.includes('reason='))); + }); + + it('carries the version in build_info and timestamps in seconds', () => { + assert.ok(lines.includes('symphony_build_info{version="9.9.9"} 1')); + assert.ok(lines.includes(`symphony_start_time_seconds ${Date.parse(snapshot.startedAt) / 1000}`)); + assert.ok(lines.includes(`symphony_config_reload_time_seconds ${Date.parse(snapshot.reloadedAt) / 1000}`)); + }); + + it('escapes label values', () => { + const escaped = renderPrometheus({ + ...snapshot, + version: 'a"b\\c', + proxies: [], + }); + assert.ok(escaped.includes('symphony_build_info{version="a\\"b\\\\c"} 1')); + }); +}); + +describe('symphony-server admin endpoint', () => { + const cert = generateSelfSignedCert('localhost'); + let dir: string; + let configPath: string; + let statusPath: string; + let socketPath: string; + let adminPort: number; + let proxyPort: number; + let echo: Awaited>; + let child: ChildProcess; + let stderr = ''; + let shuttingDown = false; + + before(async () => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'symphony-metrics-test-')); + configPath = path.join(dir, 'config.json'); + statusPath = path.join(dir, 'status.json'); + socketPath = path.join(dir, 'admin.sock'); + echo = await startEchoServer(); + proxyPort = await getFreePort(); + adminPort = await getFreePort(); + + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath, port: adminPort, host: '127.0.0.1' }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + + child = spawn(process.execPath, [SERVER_JS, '--config', configPath], { stdio: ['ignore', 'pipe', 'pipe'] }); + child.stderr?.on('data', (d) => (stderr += d.toString())); + child.on('exit', (code, sig) => { + if (!shuttingDown) stderr += `\n[child exited early code=${code} sig=${sig}]`; + }); + await waitFor(() => fs.existsSync(statusPath) && fs.existsSync(socketPath), 8000); + }); + + after(async () => { + shuttingDown = true; + if (child && child.exitCode === null) { + child.kill('SIGTERM'); + await waitFor(() => child.exitCode !== null, 3000).catch(() => child.kill('SIGKILL')); + } + await echo.close().catch(() => {}); + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('serves Prometheus text over the unix socket', async () => { + const res = await get({ socketPath }, '/metrics'); + assert.equal(res.status, 200, stderr); + assert.match(res.contentType, /text\/plain; version=0\.0\.4/); + assert.match(res.body, /^# HELP symphony_build_info /m); + assert.match( + res.body, + new RegExp( + `symphony_listener_accepted_total\\{proxy="${proxyPort}",listener="127\\.0\\.0\\.1:${proxyPort}",mode="tls"\\} \\d+` + ) + ); + }); + + it('serves the same snapshot as JSON over the loopback port', async () => { + const res = await get({ port: adminPort }, '/metrics.json'); + assert.equal(res.status, 200, stderr); + assert.match(res.contentType, /application\/json/); + const snapshot = JSON.parse(res.body) as MetricsSnapshot; + assert.equal(snapshot.proxies.length, 1); + assert.equal(snapshot.proxies[0].ports, String(proxyPort)); + assert.equal(snapshot.proxies[0].metrics.listeners[0].address, `127.0.0.1:${proxyPort}`); + }); + + it('reflects live traffic on the next scrape', async () => { + const before = JSON.parse((await get({ socketPath }, '/metrics.json')).body) as MetricsSnapshot; + await tlsRoundTrip({ port: proxyPort, servername: 'localhost', caCert: cert.cert, data: 'hello' }); + const after = JSON.parse((await get({ socketPath }, '/metrics.json')).body) as MetricsSnapshot; + + assert.equal(after.proxies[0].metrics.listeners[0].accepted, before.proxies[0].metrics.listeners[0].accepted + 1); + assert.ok(after.proxies[0].metrics.listeners[0].bytesSent > before.proxies[0].metrics.listeners[0].bytesSent); + }); + + it('restricts the unix socket to owner and group', () => { + assert.equal(fs.statSync(socketPath).mode & 0o777, 0o660); + }); + + it('answers /health with the pid and served ports', async () => { + const res = await get({ port: adminPort }, '/health'); + assert.equal(res.status, 200); + const health = JSON.parse(res.body) as { ok: boolean; pid: number; ports: number[] }; + assert.equal(health.ok, true); + assert.equal(health.pid, child.pid); + assert.deepEqual(health.ports, [proxyPort]); + }); + + it('404s an unknown path and 405s a non-GET', async () => { + assert.equal((await get({ port: adminPort }, '/nope')).status, 404); + assert.equal((await get({ port: adminPort }, '/metrics', 'POST')).status, 405); + }); + + it('ignores a query string on /metrics', async () => { + const res = await get({ port: adminPort }, '/metrics?foo=bar'); + assert.equal(res.status, 200); + }); + + it('removes the unix socket on shutdown', async () => { + shuttingDown = true; + child.kill('SIGTERM'); + await waitFor(() => child.exitCode !== null, 5000); + assert.equal(fs.existsSync(socketPath), false, 'the admin socket must not be left behind'); + }); +}); + +// A SIGKILLed process never runs its shutdown path, so the socket file survives it. The next +// process must reclaim it — but only after proving nobody is listening, since unlinking a live +// socket would silently steal the endpoint from a running symphony. +describe('symphony-server admin endpoint (stale socket recovery)', () => { + const cert = generateSelfSignedCert('localhost'); + let dir: string; + let socketPath: string; + let survivor: ChildProcess | null = null; + + function writeConfig(configPath: string, proxyPort: number, echoPort: number): void { + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: proxyPort }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echoPort }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + } + + before(() => { + dir = fs.mkdtempSync(path.join(os.tmpdir(), 'symphony-stale-sock-')); + socketPath = path.join(dir, 'admin.sock'); + }); + + after(async () => { + if (survivor && survivor.exitCode === null) { + survivor.kill('SIGKILL'); + await waitFor(() => survivor!.exitCode !== null, 3000).catch(() => {}); + } + fs.rmSync(dir, { recursive: true, force: true }); + }); + + it('reclaims a socket left behind by a SIGKILLed process', async () => { + const echo = await startEchoServer(); + try { + const firstConfig = path.join(dir, 'first.json'); + const firstStatus = path.join(dir, 'first-status.json'); + writeConfig(firstConfig, await getFreePort(), echo.port); + const first = spawn(process.execPath, [SERVER_JS, '--config', firstConfig, '--status', firstStatus], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + await waitFor(() => fs.existsSync(socketPath), 8000); + + first.kill('SIGKILL'); + await waitFor(() => first.exitCode !== null || first.signalCode !== null, 5000); + assert.ok(fs.existsSync(socketPath), 'SIGKILL should leave the socket file behind'); + + // Second process: same socket path, now stale. + const secondConfig = path.join(dir, 'second.json'); + const secondStatus = path.join(dir, 'second-status.json'); + writeConfig(secondConfig, await getFreePort(), echo.port); + let stderr = ''; + survivor = spawn(process.execPath, [SERVER_JS, '--config', secondConfig, '--status', secondStatus], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + survivor.stderr?.on('data', (d) => (stderr += d.toString())); + await waitFor(() => fs.existsSync(secondStatus), 8000); + + // The bind is retried on a timer if it loses the race, so allow a couple of cycles. + await waitFor(async () => { + const res = await get({ socketPath }, '/health').catch(() => null); + return res?.status === 200; + }, 15000); + const health = JSON.parse((await get({ socketPath }, '/health')).body) as { pid: number }; + assert.equal(health.pid, survivor.pid, `the new process must own the socket; stderr:\n${stderr}`); + } finally { + await echo.close().catch(() => {}); + } + }); +}); diff --git a/package.json b/package.json index 3143905..08e5b35 100644 --- a/package.json +++ b/package.json @@ -31,7 +31,7 @@ "artifacts": "napi artifacts", "prepublishOnly": "napi prepublish -t npm", "version": "napi version", - "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/proxy-protocol-v2.spec.js dist-test/__test__/mtls.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js", + "test": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/protection.spec.js dist-test/__test__/proxy.spec.js dist-test/__test__/proxy-protocol-v2.spec.js dist-test/__test__/mtls.spec.js dist-test/__test__/suspended.spec.js dist-test/__test__/http-listener.spec.js dist-test/__test__/server.spec.js dist-test/__test__/h2-dispatch.spec.js dist-test/__test__/metrics.spec.js", "test:integration": "tsc -p tsconfig.test.json && node --test --test-force-exit dist-test/__test__/harper-integration.spec.js", "benchmark": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark.js", "benchmark:throughput": "tsc -p tsconfig.test.json && node dist-test/__test__/benchmark-throughput.js", diff --git a/src/http_listener.rs b/src/http_listener.rs index 4fe0770..b563df4 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -17,6 +17,7 @@ use crate::http_proxy::{ host_header, read_http_headers, request_target, strip_body_framing, with_connection_close, }; use crate::listener::{make_reuseport_socket, set_rlimit_nofile}; +use crate::metrics::{BlockKind, CountingStream, ErrorKind}; use crate::proxy_conn::ConnContext; use crate::upstream::{self, UpstreamStream}; use std::net::SocketAddr; @@ -91,7 +92,10 @@ async fn accept_loop( let active = ctx.global_metrics.active_connections.load(std::sync::atomic::Ordering::Relaxed); if active >= max_connections as u64 { drop(stream); - ctx.listener_metrics.inc_blocked(); + // See listener.rs: keep the proxy-level blocked total equal to the + // sum across its listeners. + ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); + ctx.global_metrics.inc_blocked(); continue; } } @@ -125,12 +129,12 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc pair, Ok(Err(e)) => { tracing::debug!("http :80 header read error from {}: {e}", peer_addr.ip()); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::HttpHeader); return; } Err(_) => { tracing::debug!("http :80 header read timeout from {}", peer_addr.ip()); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::HttpHeader); return; } }; @@ -145,8 +149,8 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { let table = ctx.route_table.0.load(); let Some(route) = table.resolve(Some(host)) else { // No matching route — answer 404 so the ACME client gets a definitive answer // rather than a hung connection. let _ = write_simple_response(client, b"HTTP/1.1 404 Not Found\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await; - return Ok(()); + return Err(ErrorKind::NoRoute); }; // Honour the route's global rate limit the same way the TLS path does, so // a flood of /.well-known/acme-challenge/ requests can't bypass the cap. if let Some(rl) = &route.rate_limiter { if !rl.try_acquire() { - ctx.listener_metrics.inc_error(); - return Ok(()); + return Err(ErrorKind::RouteRateLimited); } } let upstream = upstream::connect(&route.destination, Some(peer_addr.ip()), UPSTREAM_CONNECT_TIMEOUT) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| { + tracing::debug!("acme upstream connect failed for {host}: {e}"); + ErrorKind::UpstreamConnect + })?; // ACME HTTP-01 challenges are GET requests with no body. Strip any // Content-Length / Transfer-Encoding headers so a client that lies about a @@ -210,19 +216,23 @@ async fn proxy_acme( // redirect path. let forwarded = with_connection_close(&strip_body_framing(headers)); - let result = match upstream { - UpstreamStream::Tcp(mut up) => proxy_one_shot(client, &mut up, &forwarded).await, - UpstreamStream::Uds { mut stream, _guard } => { - let r = proxy_one_shot(client, &mut stream, &forwarded).await; - drop(_guard); - r + // Scoped so the byte counter releases its borrow of `client` before the shutdown below. + let result = { + let mut counted = CountingStream::new(&mut *client, &ctx.listener_metrics); + match upstream { + UpstreamStream::Tcp(mut up) => proxy_one_shot(&mut counted, &mut up, &forwarded).await, + UpstreamStream::Uds { mut stream, _guard } => { + let r = proxy_one_shot(&mut counted, &mut stream, &forwarded).await; + drop(_guard); + r + } } }; // Always close the client socket after one request/response, regardless of // the upstream outcome. The HTTP-mode listener never reuses connections. let _ = client.shutdown().await; - result + result.map_err(|_| ErrorKind::Stream) } /// Send the (sanitized, body-less) request `headers` to `upstream`, then copy @@ -230,12 +240,13 @@ async fn proxy_acme( /// the request bytes flush — any bytes the client already pipelined past the /// header boundary are silently dropped, so no pipelined non-ACME payload can /// reach the backend. -async fn proxy_one_shot( - client: &mut TcpStream, +async fn proxy_one_shot( + client: &mut C, upstream: &mut U, headers: &[u8], ) -> std::io::Result<()> where + C: tokio::io::AsyncWrite + Unpin, U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, { upstream.write_all(headers).await?; diff --git a/src/listener.rs b/src/listener.rs index a90fe65..933f391 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -1,3 +1,4 @@ +use crate::metrics::BlockKind; use crate::proxy_conn::{ConnContext, handle}; use socket2::{Domain, Protocol, Socket, Type}; use std::net::SocketAddr; @@ -68,7 +69,10 @@ async fn accept_loop( if active >= max_connections as u64 { // Drop the stream — OS will send RST drop(stream); - ctx.listener_metrics.inc_blocked(); + // Both counters, so the proxy-level blocked total stays equal to the + // sum of its listeners' — the protection path already does both. + ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); + ctx.global_metrics.inc_blocked(); continue; } } diff --git a/src/metrics.rs b/src/metrics.rs index b4b29a0..7a4d4ed 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -1,11 +1,114 @@ +use crate::protection::BlockReason; +use std::pin::Pin; use std::sync::atomic::{AtomicU64, Ordering}; +use std::task::{Context, Poll}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; -#[derive(Default)] +/// Declares a small closed enum plus the label text each variant carries into the exported +/// metrics. `ALL` drives both the per-variant counter array and the export, so a new variant +/// is automatically counted and exported — there is no separate list to keep in sync. +macro_rules! labeled_enum { + ($name:ident { $($(#[$doc:meta])* $variant:ident => $label:literal),+ $(,)? }) => { + #[derive(Clone, Copy, Debug, PartialEq, Eq)] + pub enum $name { + $($(#[$doc])* $variant),+ + } + + impl $name { + pub const ALL: &'static [$name] = &[$($name::$variant),+]; + pub const COUNT: usize = Self::ALL.len(); + + pub fn as_str(self) -> &'static str { + match self { + $(Self::$variant => $label),+ + } + } + } + }; +} + +labeled_enum!(BlockKind { + /// Listener-level `maxConnections` cap — refused in the accept loop, before protection runs. + MaxConnections => "max_connections", + CidrBlocked => "cidr_blocked", + Ja3Blocked => "ja3_blocked", + Ja4Blocked => "ja4_blocked", + IncompleteHandshake => "incomplete_handshake", + NoSni => "no_sni", + RateLimited => "rate_limited", + TooManyConnections => "too_many_connections", + PenaltyBoxed => "penalty_boxed", +}); + +// Exhaustive by construction: adding a BlockReason variant fails to compile until it is given +// a BlockKind, so a new protection check can never silently land in an unlabeled bucket. +impl From<&BlockReason> for BlockKind { + fn from(reason: &BlockReason) -> Self { + match reason { + BlockReason::CidrBlocked => Self::CidrBlocked, + BlockReason::Ja3Blocked => Self::Ja3Blocked, + BlockReason::Ja4Blocked => Self::Ja4Blocked, + BlockReason::IncompleteHandshake => Self::IncompleteHandshake, + BlockReason::NoSni => Self::NoSni, + BlockReason::RateLimited => Self::RateLimited, + BlockReason::TooManyConnections => Self::TooManyConnections, + BlockReason::PenaltyBoxed => Self::PenaltyBoxed, + } + } +} + +labeled_enum!(ErrorKind { + /// SNI matched no route and the listener has no default route. + NoRoute => "no_route", + /// The route's own rate limiter rejected the connection. + RouteRateLimited => "route_rate_limited", + /// A suspended connection was never resolved — timed out or rejected by JS. + SuspendUnresolved => "suspend_unresolved", + /// TLS handshake failed or timed out. + TlsHandshake => "tls_handshake", + /// Route asks for TLS termination but has no usable cert (e.g. cert build failed). + TlsMissingCert => "tls_missing_cert", + /// Could not establish the upstream connection. + UpstreamConnect => "upstream_connect", + /// The proxied session hit the idle timeout. + IdleTimeout => "idle_timeout", + /// I/O error while proxying an established session. + Stream => "stream", + /// HTTP-mode listener could not read the request head. + HttpHeader => "http_header", +}); + +/// Per-listener counters. Every field is `Relaxed` — these are monotonic counters and gauges +/// read out of band by `metrics()`, never used to make a decision that needs ordering. pub struct ListenerMetrics { pub active_connections: AtomicU64, pub total_accepted: AtomicU64, pub total_blocked: AtomicU64, pub total_errors: AtomicU64, + /// Bytes read from clients on this listener (client → upstream). Counted at the point the + /// proxy sees them, so a terminated-TLS route counts plaintext and a passthrough route + /// counts wire bytes; neither includes the handshake, which precedes the counter. + pub bytes_in: AtomicU64, + /// Bytes written to clients on this listener (upstream → client). Same framing caveat as + /// `bytes_in`. + pub bytes_out: AtomicU64, + blocked_by_kind: [AtomicU64; BlockKind::COUNT], + errors_by_kind: [AtomicU64; ErrorKind::COUNT], +} + +impl Default for ListenerMetrics { + fn default() -> Self { + Self { + active_connections: AtomicU64::new(0), + total_accepted: AtomicU64::new(0), + total_blocked: AtomicU64::new(0), + total_errors: AtomicU64::new(0), + bytes_in: AtomicU64::new(0), + bytes_out: AtomicU64::new(0), + blocked_by_kind: std::array::from_fn(|_| AtomicU64::new(0)), + errors_by_kind: std::array::from_fn(|_| AtomicU64::new(0)), + } + } } impl ListenerMetrics { @@ -18,12 +121,31 @@ impl ListenerMetrics { self.active_connections.fetch_sub(1, Ordering::Relaxed); } - pub fn inc_blocked(&self) { + pub fn inc_blocked(&self, kind: BlockKind) { self.total_blocked.fetch_add(1, Ordering::Relaxed); + self.blocked_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); } - pub fn inc_error(&self) { + pub fn inc_error(&self, kind: ErrorKind) { self.total_errors.fetch_add(1, Ordering::Relaxed); + self.errors_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); + } + + /// Per-reason block counts, in `BlockKind::ALL` order. Zero-valued reasons are included + /// so an exported series exists from the first scrape rather than appearing mid-incident. + pub fn blocked_by_reason(&self) -> Vec<(&'static str, u64)> { + BlockKind::ALL + .iter() + .map(|k| (k.as_str(), self.blocked_by_kind[*k as usize].load(Ordering::Relaxed))) + .collect() + } + + /// Per-reason error counts, in `ErrorKind::ALL` order. See `blocked_by_reason`. + pub fn errors_by_reason(&self) -> Vec<(&'static str, u64)> { + ErrorKind::ALL + .iter() + .map(|k| (k.as_str(), self.errors_by_kind[*k as usize].load(Ordering::Relaxed))) + .collect() } } @@ -32,6 +154,10 @@ pub struct GlobalMetrics { pub active_connections: AtomicU64, pub total_blocked: AtomicU64, pub pending_suspended: AtomicU64, + /// Suspended connections that JS resolved with a route. + pub suspended_resolved: AtomicU64, + /// Suspended connections that timed out or were rejected with a null route. + pub suspended_unresolved: AtomicU64, } impl GlobalMetrics { @@ -51,7 +177,140 @@ impl GlobalMetrics { self.pending_suspended.fetch_add(1, Ordering::Relaxed); } - pub fn dec_suspended(&self) { + /// Leave the pending gauge, recording how the suspension ended. + pub fn dec_suspended(&self, resolved: bool) { self.pending_suspended.fetch_sub(1, Ordering::Relaxed); + if resolved { + self.suspended_resolved.fetch_add(1, Ordering::Relaxed); + } else { + self.suspended_unresolved.fetch_add(1, Ordering::Relaxed); + } + } +} + +/// Wraps the *client* side of a proxied session so byte counts accrue as the copy runs rather +/// than at completion. Counting `copy_bidirectional`'s return value instead would lose every +/// byte of any session that ends by idle timeout or reset — i.e. most long-lived ones. +/// +/// Because it wraps the client, the direction naming is from the proxy's point of view: bytes +/// read here came *from* the client, bytes written here go *to* the client. +pub struct CountingStream<'a, S> { + inner: S, + metrics: &'a ListenerMetrics, +} + +impl<'a, S> CountingStream<'a, S> { + pub fn new(inner: S, metrics: &'a ListenerMetrics) -> Self { + Self { inner, metrics } + } +} + +impl AsyncRead for CountingStream<'_, S> { + fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { + let this = self.get_mut(); + let before = buf.filled().len(); + let result = Pin::new(&mut this.inner).poll_read(cx, buf); + if matches!(result, Poll::Ready(Ok(()))) { + let read = buf.filled().len().saturating_sub(before); + if read > 0 { + this.metrics.bytes_in.fetch_add(read as u64, Ordering::Relaxed); + } + } + result + } +} + +impl AsyncWrite for CountingStream<'_, S> { + fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll> { + let this = self.get_mut(); + let result = Pin::new(&mut this.inner).poll_write(cx, buf); + if let Poll::Ready(Ok(written)) = &result { + this.metrics.bytes_out.fetch_add(*written as u64, Ordering::Relaxed); + } + result + } + + fn poll_write_vectored( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + bufs: &[std::io::IoSlice<'_>], + ) -> Poll> { + let this = self.get_mut(); + let result = Pin::new(&mut this.inner).poll_write_vectored(cx, bufs); + if let Poll::Ready(Ok(written)) = &result { + this.metrics.bytes_out.fetch_add(*written as u64, Ordering::Relaxed); + } + result + } + + fn is_write_vectored(&self) -> bool { + self.inner.is_write_vectored() + } + + fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_flush(cx) + } + + fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + Pin::new(&mut self.get_mut().inner).poll_shutdown(cx) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn block_and_error_kinds_have_distinct_labels() { + let mut labels: Vec<&str> = BlockKind::ALL.iter().map(|k| k.as_str()).collect(); + labels.sort_unstable(); + let count = labels.len(); + labels.dedup(); + assert_eq!(labels.len(), count, "BlockKind labels must be unique"); + + let mut labels: Vec<&str> = ErrorKind::ALL.iter().map(|k| k.as_str()).collect(); + labels.sort_unstable(); + let count = labels.len(); + labels.dedup(); + assert_eq!(labels.len(), count, "ErrorKind labels must be unique"); + } + + #[test] + fn per_reason_counts_sum_to_the_total() { + let m = ListenerMetrics::default(); + m.inc_blocked(BlockKind::RateLimited); + m.inc_blocked(BlockKind::RateLimited); + m.inc_blocked(BlockKind::NoSni); + m.inc_error(ErrorKind::UpstreamConnect); + + let blocked: u64 = m.blocked_by_reason().iter().map(|(_, v)| v).sum(); + assert_eq!(blocked, m.total_blocked.load(Ordering::Relaxed)); + assert_eq!(blocked, 3); + + let errors: u64 = m.errors_by_reason().iter().map(|(_, v)| v).sum(); + assert_eq!(errors, m.total_errors.load(Ordering::Relaxed)); + assert_eq!(errors, 1); + + // Every reason is exported, including the ones still at zero. + assert_eq!(m.blocked_by_reason().len(), BlockKind::COUNT); + assert_eq!(m.errors_by_reason().len(), ErrorKind::COUNT); + } + + #[tokio::test] + async fn counting_stream_records_both_directions() { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + + let metrics = ListenerMetrics::default(); + // duplex gives a paired in-memory stream; write into `peer` to be read through the counter. + let (client, mut peer) = tokio::io::duplex(64); + let mut counted = CountingStream::new(client, &metrics); + + peer.write_all(b"hello").await.unwrap(); + let mut buf = [0u8; 5]; + counted.read_exact(&mut buf).await.unwrap(); + counted.write_all(b"world!").await.unwrap(); + + assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 5); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 6); } } diff --git a/src/proxy.rs b/src/proxy.rs index 85d8ca3..aadee80 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -147,11 +147,47 @@ pub struct JsHotConfig { pub protection: Option>, } +/// A single labelled counter — one entry per block/error reason. +#[napi(object)] +pub struct JsLabeledCount { + pub reason: String, + pub count: f64, +} + +#[napi(object)] +pub struct JsListenerMetrics { + /// "host:port" — matches the `listener` field on emitted events. + pub address: String, + /// "tls" or "http". + pub mode: String, + pub active_connections: f64, + pub accepted: f64, + pub blocked: f64, + pub errors: f64, + /// Bytes read from clients (client → upstream). + pub bytes_received: f64, + /// Bytes written to clients (upstream → client). + pub bytes_sent: f64, + pub blocked_by_reason: Vec, + pub errors_by_reason: Vec, +} + +// Counters are reported as f64 because napi maps u64 to BigInt, which JSON.stringify cannot +// serialise. f64 is exact to 2^53, so a byte counter stays exact past 9 PB per listener. #[napi(object)] pub struct JsProxyMetrics { pub active_connections: f64, pub blocked_connections: f64, pub pending_suspended: f64, + /// Suspended connections that JS resolved with a route. + pub suspended_resolved: f64, + /// Suspended connections that timed out or were rejected. + pub suspended_unresolved: f64, + /// Routes currently in the live table, including the default route. + pub routes: f64, + /// Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. + pub failing_routes: f64, + pub listeners: Vec, } #[napi(object)] @@ -190,6 +226,13 @@ struct InternalListener { mode: ListenerMode, } +fn labeled_counts(counts: Vec<(&'static str, u64)>) -> Vec { + counts + .into_iter() + .map(|(reason, count)| JsLabeledCount { reason: reason.to_string(), count: count as f64 }) + .collect() +} + /// Per-listener runtime state. struct ListenerState { addr: String, @@ -553,10 +596,40 @@ impl SymphonyProxyWrap { #[napi] pub fn metrics(&self) -> JsProxyMetrics { + let table = self.route_table.0.load(); + + // `listeners` and `listener_states` are built in lockstep in the constructor and never + // mutated, so index i refers to the same listener in both. + let listeners = self + .listeners + .iter() + .zip(self.listener_states.iter()) + .map(|(listener, state)| JsListenerMetrics { + address: state.addr.clone(), + mode: match listener.mode { + ListenerMode::Tls => "tls".to_string(), + ListenerMode::Http => "http".to_string(), + }, + active_connections: state.metrics.active_connections.load(Ordering::Relaxed) as f64, + accepted: state.metrics.total_accepted.load(Ordering::Relaxed) as f64, + blocked: state.metrics.total_blocked.load(Ordering::Relaxed) as f64, + errors: state.metrics.total_errors.load(Ordering::Relaxed) as f64, + bytes_received: state.metrics.bytes_in.load(Ordering::Relaxed) as f64, + bytes_sent: state.metrics.bytes_out.load(Ordering::Relaxed) as f64, + blocked_by_reason: labeled_counts(state.metrics.blocked_by_reason()), + errors_by_reason: labeled_counts(state.metrics.errors_by_reason()), + }) + .collect(); + JsProxyMetrics { active_connections: self.global_metrics.active_connections.load(Ordering::Relaxed) as f64, blocked_connections: self.global_metrics.total_blocked.load(Ordering::Relaxed) as f64, pending_suspended: self.global_metrics.pending_suspended.load(Ordering::Relaxed) as f64, + suspended_resolved: self.global_metrics.suspended_resolved.load(Ordering::Relaxed) as f64, + suspended_unresolved: self.global_metrics.suspended_unresolved.load(Ordering::Relaxed) as f64, + routes: table.route_count() as f64, + failing_routes: table.failing_route_count() as f64, + listeners, } } diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index 21757c1..9c3d5a7 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -1,4 +1,4 @@ -use crate::metrics::{GlobalMetrics, ListenerMetrics}; +use crate::metrics::{BlockKind, CountingStream, ErrorKind, GlobalMetrics, ListenerMetrics}; use crate::protection::{IpState, ProtectionState}; use crate::router::{Destination, ForwardFingerprint, LiveRouteTable, SourceAddressMode}; use crate::sni; @@ -84,7 +84,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { - ctx.listener_metrics.inc_blocked(); + ctx.listener_metrics.inc_blocked(BlockKind::from(&reason)); ctx.global_metrics.inc_blocked(); emit(&ctx.js_emit, JsEvent::Blocked { ip: peer_ip.to_string(), @@ -110,7 +110,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc r.clone(), None => { - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::NoRoute); return; // No route and no default — drop } }; @@ -118,7 +118,7 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc r, _ => { ctx.suspended_registry.remove(id); - ctx.global_metrics.dec_suspended(); + ctx.global_metrics.dec_suspended(false); + ctx.listener_metrics.inc_error(ErrorKind::SuspendUnresolved); return; // Timed out or rejected } }; - ctx.global_metrics.dec_suspended(); + ctx.global_metrics.dec_suspended(true); EffectiveRoute { destination: resolved.destination, @@ -203,17 +204,17 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { tracing::debug!("TLS handshake error from {peer_ip}: {e}"); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsHandshake); return; } Err(_) => { tracing::debug!("TLS handshake timeout from {peer_ip}"); - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsHandshake); return; } } } else { - ctx.listener_metrics.inc_error(); + ctx.listener_metrics.inc_error(ErrorKind::TlsMissingCert); return; } } else { @@ -221,33 +222,39 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc, + client: tokio_rustls::server::TlsStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, -) -> std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { // TLS facts (incl. the verified mTLS client cert chain) forwarded via PROXY v2 // TLVs; only collected on routes that can carry them. let tls_forward = matches!(sf.mode, SourceAddressMode::ProxyProtocolV2) .then(|| collect_tls_forward(client.get_ref().1)); let sf = SourceForwarding { tls: tls_forward.as_ref(), ..sf }; - let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) - .await - .map_err(|e| std::io::Error::other(e.to_string()))?; - // HTTP-header injection is only valid for a plaintext HTTP/1 upstream. An h2-negotiated // upstream receives binary frames, so text header insertion would corrupt them. + // Read before wrapping — the counter has no view of the TLS session. let l7_http1 = client.get_ref().1.alpn_protocol() != Some(b"h2".as_ref()); + let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) + .await + .map_err(|e| { + tracing::debug!("upstream connect failed for {}: {e}", sf.peer_addr.ip()); + ErrorKind::UpstreamConnect + })?; + + let mut client = CountingStream::new(client, &ctx.listener_metrics); + match &mut upstream { UpstreamStream::Tcp(ref mut up) => forward(&mut client, up, &sf, l7_http1, ctx.idle_timeout).await, UpstreamStream::Uds { ref mut stream, .. } => { @@ -257,14 +264,19 @@ async fn proxy_via_tls( } async fn proxy_raw( - mut client: TcpStream, + client: TcpStream, dest: &Destination, sf: SourceForwarding<'_>, ctx: &ConnContext, -) -> std::io::Result<()> { +) -> std::result::Result<(), ErrorKind> { let mut upstream = upstream::connect(dest, Some(sf.peer_addr.ip()), ctx.upstream_connect_timeout) .await - .map_err(|e| std::io::Error::other(e.to_string()))?; + .map_err(|e| { + tracing::debug!("upstream connect failed for {}: {e}", sf.peer_addr.ip()); + ErrorKind::UpstreamConnect + })?; + + let mut client = CountingStream::new(client, &ctx.listener_metrics); // Passthrough forwards raw TLS bytes — never a plaintext HTTP/1 stream, so header injection // is disabled (only PROXY protocol carriers apply here). @@ -287,7 +299,7 @@ async fn forward( sf: &SourceForwarding<'_>, l7_http1: bool, idle: Duration, -) -> std::io::Result<()> +) -> std::result::Result<(), ErrorKind> where C: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, U: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin, @@ -301,12 +313,15 @@ where crate::http_proxy::proxy_http1_rewriting(client, upstream, &rewrites).await } }; + // The idle timeout is reported as its own kind rather than inferred from an + // `io::ErrorKind::TimedOut`, which a peer's kernel-level ETIMEDOUT would also produce. if idle.is_zero() { - body.await + body.await.map_err(|_| ErrorKind::Stream) } else { - timeout(idle, body) - .await - .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "idle timeout"))? + match timeout(idle, body).await { + Ok(result) => result.map_err(|_| ErrorKind::Stream), + Err(_) => Err(ErrorKind::IdleTimeout), + } } } diff --git a/src/router.rs b/src/router.rs index f4a01e4..f9eccbf 100644 --- a/src/router.rs +++ b/src/router.rs @@ -164,6 +164,17 @@ pub struct RouteTable { } impl RouteTable { + /// Number of routes serving traffic, including the default route if one is configured. + pub fn route_count(&self) -> usize { + self.exact.len() + self.wildcard.len() + usize::from(self.default.is_some()) + } + + /// Number of SNIs whose cert failed to build in this table — either dropped, or serving a + /// carried-forward last-good cert. Non-zero means a rotation needs attention. + pub fn failing_route_count(&self) -> usize { + self.failing_snis.len() + } + pub fn resolve(&self, sni: Option<&str>) -> Option<&Route> { let Some(sni) = sni else { return self.default.as_ref(); diff --git a/ts/addon.d.ts b/ts/addon.d.ts index 9c0a5eb..702adf3 100644 --- a/ts/addon.d.ts +++ b/ts/addon.d.ts @@ -118,10 +118,40 @@ export interface JsHotConfig { */ protection?: Array } +/** A single labelled counter — one entry per block/error reason. */ +export interface JsLabeledCount { + reason: string + count: number +} +export interface JsListenerMetrics { + /** "host:port" — matches the `listener` field on emitted events. */ + address: string + /** "tls" or "http". */ + mode: string + activeConnections: number + accepted: number + blocked: number + errors: number + /** Bytes read from clients (client → upstream). */ + bytesReceived: number + /** Bytes written to clients (upstream → client). */ + bytesSent: number + blockedByReason: Array + errorsByReason: Array +} export interface JsProxyMetrics { activeConnections: number blockedConnections: number pendingSuspended: number + /** Suspended connections that JS resolved with a route. */ + suspendedResolved: number + /** Suspended connections that timed out or were rejected. */ + suspendedUnresolved: number + /** Routes currently in the live table, including the default route. */ + routes: number + /** Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. */ + failingRoutes: number + listeners: Array } export interface JsBlockedIpsInfo { rateLimited: Array diff --git a/ts/admin.ts b/ts/admin.ts new file mode 100644 index 0000000..b2c2dc5 --- /dev/null +++ b/ts/admin.ts @@ -0,0 +1,406 @@ +//! Read-only admin/metrics endpoint for the standalone `symphony-server`. +//! +//! Consumers that run symphony out-of-process have no access to the napi `metrics()` call, so +//! the server exposes the same numbers over HTTP: +//! +//! GET /metrics Prometheus text exposition (v0.0.4) +//! GET /metrics.json the same snapshot as JSON +//! GET /health liveness (`{ ok, pid, version, ports }`) +//! +//! It binds a Unix socket, a loopback TCP port, or both. Everything here is best-effort: a +//! bind failure or a handler throw must never affect proxying, so failures are logged and +//! retried rather than propagated. + +import { createServer, type Server, type IncomingMessage, type ServerResponse } from 'node:http'; +import { connect } from 'node:net'; +import { chmodSync, unlinkSync } from 'node:fs'; +import type { ProxyMetrics } from './types.js'; + +export interface AdminConfig { + /** Unix socket path to listen on. Relative paths resolve against the config file's directory. */ + socketPath?: string; + /** Permissions applied to `socketPath` after bind. Default 0o660. */ + socketMode?: number; + /** TCP port to listen on. */ + port?: number; + /** Interface for `port`. Default 127.0.0.1 — do not expose metrics off-box without a reason. */ + host?: string; +} + +/** One running proxy's identity and current counters. */ +export interface ProxySnapshot { + /** Sorted listener ports for this proxy, as configured ("80,443"). */ + ports: string; + metrics: ProxyMetrics; +} + +export interface MetricsSnapshot { + pid: number; + version: string; + startedAt: string; + reloadedAt: string; + proxies: ProxySnapshot[]; +} + +const RETRY_MS = 5_000; + +// ── Prometheus rendering ────────────────────────────────────────────────────── + +// Only `\`, `"` and newline are special in a label value (exposition format v0.0.4). +function escapeLabel(value: string): string { + return value.replace(/\\/g, '\\\\').replace(/"/g, '\\"').replace(/\n/g, '\\n'); +} + +function labels(pairs: Record): string { + const rendered = Object.entries(pairs) + .map(([k, v]) => `${k}="${escapeLabel(v)}"`) + .join(','); + return rendered ? `{${rendered}}` : ''; +} + +class Exposition { + private readonly lines: string[] = []; + private declared = new Set(); + + /** Emit HELP/TYPE once per metric name, then a sample. */ + sample( + name: string, + type: 'counter' | 'gauge', + help: string, + value: number, + tags: Record = {} + ): void { + if (!this.declared.has(name)) { + this.declared.add(name); + this.lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`); + } + this.lines.push(`${name}${labels(tags)} ${value}`); + } + + toString(): string { + return `${this.lines.join('\n')}\n`; + } +} + +/** + * Render a snapshot as Prometheus text. + * + * Blocked and error counts are only ever emitted with their `reason` label — the per-reason + * series sum to the unlabeled total by construction, so a separate total would be a second + * representation of the same number. Use `sum without(reason)` for the total. Likewise the + * proxy-level active-connection gauge is `sum without(listener)` of the per-listener one. + */ +export function renderPrometheus(snapshot: MetricsSnapshot): string { + const out = new Exposition(); + + out.sample('symphony_build_info', 'gauge', 'Always 1; the version is carried in the label.', 1, { + version: snapshot.version, + }); + out.sample( + 'symphony_start_time_seconds', + 'gauge', + 'Unix time the server process started.', + Date.parse(snapshot.startedAt) / 1000 + ); + out.sample( + 'symphony_config_reload_time_seconds', + 'gauge', + 'Unix time of the last successful config reconcile.', + Date.parse(snapshot.reloadedAt) / 1000 + ); + + for (const { ports, metrics } of snapshot.proxies) { + const proxy = { proxy: ports }; + + out.sample( + 'symphony_routes', + 'gauge', + 'Routes in the live table, including the default route.', + metrics.routes, + proxy + ); + out.sample( + 'symphony_routes_failing', + 'gauge', + 'Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert.', + metrics.failingRoutes, + proxy + ); + out.sample( + 'symphony_suspended_pending', + 'gauge', + 'Connections held awaiting resolveConnection().', + metrics.pendingSuspended, + proxy + ); + out.sample( + 'symphony_suspended_total', + 'counter', + 'Suspended connections by how they ended.', + metrics.suspendedResolved, + { + ...proxy, + outcome: 'resolved', + } + ); + out.sample( + 'symphony_suspended_total', + 'counter', + 'Suspended connections by how they ended.', + metrics.suspendedUnresolved, + { + ...proxy, + outcome: 'unresolved', + } + ); + + for (const l of metrics.listeners) { + const tags = { ...proxy, listener: l.address, mode: l.mode }; + + out.sample( + 'symphony_listener_active_connections', + 'gauge', + 'Connections currently being proxied.', + l.activeConnections, + tags + ); + out.sample('symphony_listener_accepted_total', 'counter', 'Connections accepted for proxying.', l.accepted, tags); + out.sample( + 'symphony_listener_bytes_received_total', + 'counter', + 'Bytes read from clients (client → upstream).', + l.bytesReceived, + tags + ); + out.sample( + 'symphony_listener_bytes_sent_total', + 'counter', + 'Bytes written to clients (upstream → client).', + l.bytesSent, + tags + ); + for (const { reason, count } of l.blockedByReason) { + out.sample( + 'symphony_listener_blocked_total', + 'counter', + 'Connections rejected before proxying, by reason.', + count, + { + ...tags, + reason, + } + ); + } + for (const { reason, count } of l.errorsByReason) { + out.sample('symphony_listener_errors_total', 'counter', 'Connections that failed, by reason.', count, { + ...tags, + reason, + }); + } + } + } + + return out.toString(); +} + +// ── Server ──────────────────────────────────────────────────────────────────── + +/** True if something is actively listening on `socketPath` (as opposed to a stale socket file). */ +function socketIsLive(socketPath: string): Promise { + return new Promise((resolve) => { + const probe = connect(socketPath); + const done = (live: boolean) => { + probe.destroy(); + resolve(live); + }; + probe.once('connect', () => done(true)); + probe.once('error', () => done(false)); + }); +} + +/** + * Owns the admin listeners for the lifetime of the process. `update()` is idempotent: it is + * called on every reconcile and only rebuilds when the admin config actually changed. + */ +export class AdminServer { + private readonly snapshot: () => MetricsSnapshot; + private readonly log: (msg: string, ...rest: unknown[]) => void; + private readonly logErr: (msg: string, ...rest: unknown[]) => void; + private servers: Server[] = []; + private signature = ''; + private config: AdminConfig | null = null; + private retryTimer: NodeJS.Timeout | null = null; + private stopped = false; + + constructor( + snapshot: () => MetricsSnapshot, + log: (msg: string, ...rest: unknown[]) => void, + logErr: (msg: string, ...rest: unknown[]) => void + ) { + this.snapshot = snapshot; + this.log = log; + this.logErr = logErr; + } + + async update(config: AdminConfig | undefined): Promise { + if (this.stopped) return; + const signature = JSON.stringify(config ?? null); + if (signature === this.signature) return; + this.signature = signature; + this.config = config ?? null; + await this.closeServers(); + if (config && (config.socketPath || config.port !== undefined)) await this.bind(); + } + + async stop(): Promise { + this.stopped = true; + this.clearRetry(); + await this.closeServers(); + } + + private clearRetry(): void { + if (this.retryTimer) { + clearTimeout(this.retryTimer); + this.retryTimer = null; + } + } + + // A failed bind is retried rather than thrown: during a version upgrade the incumbent still + // holds the admin socket/port (the proxy listeners overlap via SO_REUSEPORT, but a Node HTTP + // server has no such luxury), so the successor binds a few seconds later once it exits. + private scheduleRetry(): void { + if (this.stopped || this.retryTimer) return; + this.retryTimer = setTimeout(() => { + this.retryTimer = null; + void this.bind(); + }, RETRY_MS); + this.retryTimer.unref(); + } + + private async bind(): Promise { + const config = this.config; + if (this.stopped || !config) return; + await this.closeServers(); + + const targets: Array<{ describe: string; listen: (server: Server) => void; onBound?: () => void }> = []; + if (config.socketPath) { + const path = config.socketPath; + targets.push({ + describe: path, + listen: (server) => server.listen(path), + onBound: () => chmodSync(path, config.socketMode ?? 0o660), + }); + } + if (config.port !== undefined) { + const host = config.host ?? '127.0.0.1'; + targets.push({ describe: `${host}:${config.port}`, listen: (server) => server.listen(config.port, host) }); + } + + for (const target of targets) { + try { + const server = await this.listenOne(target, config); + this.servers.push(server); + this.log(`admin endpoint listening on ${target.describe}`); + } catch (err) { + this.logErr(`could not bind admin endpoint on ${target.describe} (retrying):`, (err as Error).message); + // Drop any sibling that did bind so a retry starts from a clean slate and can't + // double-bind the one that succeeded. + await this.closeServers(); + this.scheduleRetry(); + return; + } + } + } + + private async listenOne( + target: { describe: string; listen: (server: Server) => void; onBound?: () => void }, + config: AdminConfig + ): Promise { + const server = createServer((req, res) => this.handle(req, res)); + // Metrics scrapes are short; don't hold sockets open between them. + server.keepAliveTimeout = 0; + + const attempt = (): Promise => + new Promise((resolve, reject) => { + const onError = (err: NodeJS.ErrnoException) => { + server.removeListener('listening', onListening); + reject(err); + }; + const onListening = () => { + server.removeListener('error', onError); + try { + target.onBound?.(); + } catch (err) { + this.logErr(`could not set permissions on ${target.describe}:`, (err as Error).message); + } + // Post-bind errors must not crash the process. + server.on('error', (err) => this.logErr(`admin endpoint error (${target.describe}):`, err)); + resolve(server); + }; + server.once('error', onError); + server.once('listening', onListening); + target.listen(server); + }); + + try { + return await attempt(); + } catch (err) { + // A Unix socket left behind by a process that died without cleaning up blocks the + // bind forever. Only remove it once a connect probe proves nobody is listening — + // unlinking a live socket would silently steal the endpoint from a running process. + const code = (err as NodeJS.ErrnoException).code; + if (code === 'EADDRINUSE' && config.socketPath && target.describe === config.socketPath) { + if (await socketIsLive(config.socketPath)) throw err; + this.log(`removing stale admin socket ${config.socketPath}`); + unlinkSync(config.socketPath); + return attempt(); + } + throw err; + } + } + + private handle(req: IncomingMessage, res: ServerResponse): void { + // Strip any query string; the endpoints take no parameters. + const path = (req.url ?? '/').split('?')[0]; + if (req.method !== 'GET' && req.method !== 'HEAD') { + res.writeHead(405, { allow: 'GET, HEAD' }).end(); + return; + } + + try { + const snapshot = this.snapshot(); + if (path === '/metrics') { + const body = renderPrometheus(snapshot); + res.writeHead(200, { 'content-type': 'text/plain; version=0.0.4; charset=utf-8' }).end(body); + } else if (path === '/metrics.json') { + res.writeHead(200, { 'content-type': 'application/json' }).end(JSON.stringify(snapshot)); + } else if (path === '/health') { + const ports = snapshot.proxies.flatMap((p) => p.ports.split(',').map(Number)); + res + .writeHead(200, { 'content-type': 'application/json' }) + .end(JSON.stringify({ ok: true, pid: snapshot.pid, version: snapshot.version, ports })); + } else { + res.writeHead(404, { 'content-type': 'text/plain' }).end('not found\n'); + } + } catch (err) { + this.logErr('admin request failed:', (err as Error).message); + res.writeHead(500, { 'content-type': 'text/plain' }).end('internal error\n'); + } + } + + private async closeServers(): Promise { + const servers = this.servers; + this.servers = []; + await Promise.all( + servers.map( + (server) => + new Promise((resolve) => { + server.close(() => resolve()); + // close() waits for in-flight requests; a scrape holding the socket must not + // stall shutdown. + server.closeAllConnections(); + }) + ) + ); + } +} diff --git a/ts/index.ts b/ts/index.ts index 5c31b80..cc0502c 100644 --- a/ts/index.ts +++ b/ts/index.ts @@ -12,6 +12,8 @@ export type { RateLimitConfig, HotConfig, ProxyMetrics, + ListenerMetrics, + LabeledCount, BlockedIpsInfo, SuspendedConnection, ResolveRoute, @@ -20,3 +22,5 @@ export type { SuspendedEvent, ErrorEvent, } from './types.js'; +export { renderPrometheus } from './admin.js'; +export type { AdminConfig, MetricsSnapshot, ProxySnapshot } from './admin.js'; diff --git a/ts/proxy.ts b/ts/proxy.ts index c44adbe..f8da7ca 100644 --- a/ts/proxy.ts +++ b/ts/proxy.ts @@ -239,6 +239,22 @@ export class SymphonyProxy extends EventEmitter { activeConnections: m.activeConnections, blockedConnections: m.blockedConnections, pendingSuspended: m.pendingSuspended, + suspendedResolved: m.suspendedResolved, + suspendedUnresolved: m.suspendedUnresolved, + routes: m.routes, + failingRoutes: m.failingRoutes, + listeners: m.listeners.map((l) => ({ + address: l.address, + mode: l.mode as 'tls' | 'http', + activeConnections: l.activeConnections, + accepted: l.accepted, + blocked: l.blocked, + errors: l.errors, + bytesReceived: l.bytesReceived, + bytesSent: l.bytesSent, + blockedByReason: l.blockedByReason.map((c) => ({ reason: c.reason, count: c.count })), + errorsByReason: l.errorsByReason.map((c) => ({ reason: c.reason, count: c.count })), + })), }; } diff --git a/ts/server.ts b/ts/server.ts index 4649cd3..26d5f93 100644 --- a/ts/server.ts +++ b/ts/server.ts @@ -3,6 +3,7 @@ import { readFileSync, writeFileSync, renameSync, unlinkSync, watch } from 'node import { dirname, isAbsolute, join, basename } from 'node:path'; import { SymphonyProxy } from './index.js'; import type { ProxyConfig, ListenerConfig, RouteConfig, CertConfig, MtlsConfig } from './index.js'; +import { AdminServer, type AdminConfig, type MetricsSnapshot } from './admin.js'; // package.json sits at the package root, which is 1 level above dist/server.js (production // layout) or 2 levels above dist-test/ts/server.js (test layout) — try both, like loadAddon. @@ -56,6 +57,8 @@ interface FileProxyConfig { interface ConfigFile { version?: number; + /** Optional read-only metrics/health endpoint. Omit to expose nothing. */ + admin?: AdminConfig; proxies: FileProxyConfig[]; } @@ -145,7 +148,9 @@ class ServerState { private readonly statusPath: string; private readonly baseDir: string; private readonly active = new Map(); + private readonly admin: AdminServer; private startedAt = ''; + private reloadedAt = ''; private reloading: Promise = Promise.resolve(); private watcher: ReturnType | null = null; // Cert/key files referenced by the current config, watched for rotation. Keyed by @@ -159,6 +164,19 @@ class ServerState { this.configPath = configPath; this.statusPath = statusPath; this.baseDir = dirname(configPath); + this.admin = new AdminServer(() => this.metricsSnapshot(), log, logErr); + } + + // Read live on each request rather than cached, so a scrape never serves counters frozen at + // the last reconcile. + private metricsSnapshot(): MetricsSnapshot { + return { + pid: process.pid, + version: pkg.version, + startedAt: this.startedAt, + reloadedAt: this.reloadedAt, + proxies: [...this.active].map(([ports, entry]) => ({ ports, metrics: entry.proxy.metrics() })), + }; } private readConfig(): ConfigFile | null { @@ -337,6 +355,15 @@ class ServerState { // renewal on disk (no config.json write) triggers a reconcile and live reload. this.updateCertWatchers(config); + // Rebinds only if the admin block changed. Never throws — the endpoint is + // observability, and losing it must not abort a reconcile that already applied routes. + const admin = config.admin && { + ...config.admin, + socketPath: config.admin.socketPath ? resolvePath(config.admin.socketPath, this.baseDir) : undefined, + }; + await this.admin.update(admin).catch((err) => logErr('admin endpoint update failed:', (err as Error).message)); + + this.reloadedAt = new Date().toISOString(); this.writeStatus(); } @@ -346,7 +373,7 @@ class ServerState { pid: process.pid, version: pkg.version, startedAt: this.startedAt, - reloadedAt: new Date().toISOString(), + reloadedAt: this.reloadedAt, configPath: this.configPath, ports, }; @@ -386,6 +413,7 @@ class ServerState { } for (const w of this.certWatchers.values()) w.close(); this.certWatchers.clear(); + await this.admin.stop().catch((err) => logErr('stopping admin endpoint:', (err as Error).message)); await this.reloading.catch(() => {}); for (const [key, entry] of this.active) { await entry.proxy.stop().catch((err) => logErr(`stopping proxy [${key}]:`, err)); diff --git a/ts/types.ts b/ts/types.ts index b741791..438afa3 100644 --- a/ts/types.ts +++ b/ts/types.ts @@ -268,6 +268,28 @@ export interface HotConfig { // ── Metrics ─────────────────────────────────────────────────────────────────── +/** A counter broken out by reason. Every reason is reported, including those still at zero. */ +export interface LabeledCount { + reason: string; + count: number; +} + +export interface ListenerMetrics { + /** "host:port" — matches the `listener` field on emitted events. */ + address: string; + mode: 'tls' | 'http'; + activeConnections: number; + accepted: number; + blocked: number; + errors: number; + /** Bytes read from clients (client → upstream). */ + bytesReceived: number; + /** Bytes written to clients (upstream → client). */ + bytesSent: number; + blockedByReason: LabeledCount[]; + errorsByReason: LabeledCount[]; +} + export interface ProxyMetrics { /** Number of connections currently being proxied. */ activeConnections: number; @@ -275,6 +297,16 @@ export interface ProxyMetrics { blockedConnections: number; /** Connections currently held waiting for resolveConnection(). */ pendingSuspended: number; + /** Suspended connections that were resolved with a route. */ + suspendedResolved: number; + /** Suspended connections that timed out or were rejected. */ + suspendedUnresolved: number; + /** Routes in the live table, including the default route. */ + routes: number; + /** Routes whose cert failed to build — dropped, or serving a carried-forward last-good cert. */ + failingRoutes: number; + /** Per-listener breakdown, in configuration order. */ + listeners: ListenerMetrics[]; } export interface BlockedIpsInfo { From 479070d46845ff40ce46b5b353e56b354f8a08b3 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 18:14:47 -0600 Subject: [PATCH 2/7] Address cross-model review: batch byte counters, group Prometheus samples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three findings from the Gemini review leg. Byte counting did a `fetch_add` on a listener-shared atomic for every chunk read or written — putting one cache line in the path of every 8 KiB of proxied traffic and ping-ponging it across every core, which is the cross-core contention SO_REUSEPORT per worker exists to avoid. Counts now accumulate per connection and publish every 256 KiB and on drop, so a saturated connection touches the shared line ~32x less often while a scrape still sees a busy connection's traffic promptly. Drop covers the aborted-at-shutdown path. The Prometheus renderer emitted samples in proxy -> listener call order, so with more than one proxy configured a metric name's samples were split across the output under a single HELP/TYPE pair. Strict parsers reject or drop the split group. Exposition now accumulates by name and renders grouped, leaving callers in the natural iteration order. AdminServer.update() left a pending bind retry armed, so a config reload during a retry backoff would tear down the listeners it had just bound. Co-Authored-By: Claude Opus 5 --- __test__/metrics.spec.ts | 25 ++++++++++++ src/metrics.rs | 86 +++++++++++++++++++++++++++++++++++++--- ts/admin.ts | 30 +++++++++----- 3 files changed, 127 insertions(+), 14 deletions(-) diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts index 2eec88e..72c6212 100644 --- a/__test__/metrics.spec.ts +++ b/__test__/metrics.spec.ts @@ -271,6 +271,31 @@ describe('renderPrometheus', () => { assert.equal(typeLines.filter((l) => l.includes('symphony_suspended_total')).length, 1); }); + // Samples of one metric name must be contiguous under a single HELP/TYPE pair. Emitting in + // proxy → listener call order would interleave them once a second proxy exists, and strict + // parsers reject or drop the split group. + it('keeps every metric name contiguous across multiple proxies', () => { + const second = structuredClone(snapshot.proxies[0]); + second.ports = '8443'; + second.metrics.listeners[0].address = '0.0.0.0:8443'; + const multi = renderPrometheus({ ...snapshot, proxies: [snapshot.proxies[0], second] }).split('\n'); + + const seen = new Set(); + let previous = ''; + for (const line of multi) { + if (!line || line.startsWith('#')) continue; + const name = line.slice(0, Math.min(...[line.indexOf('{'), line.indexOf(' ')].filter((i) => i >= 0))); + if (name !== previous) { + assert.ok(!seen.has(name), `samples for ${name} are split across the output`); + seen.add(name); + previous = name; + } + } + // Both proxies really are present, so the check above wasn't vacuous. + assert.ok(multi.includes('symphony_routes{proxy="80,443"} 7')); + assert.ok(multi.includes('symphony_routes{proxy="8443"} 7')); + }); + it('labels every proxy-scoped and listener-scoped sample', () => { assert.ok(lines.includes('symphony_routes{proxy="80,443"} 7')); assert.ok(lines.includes('symphony_routes_failing{proxy="80,443"} 1')); diff --git a/src/metrics.rs b/src/metrics.rs index 7a4d4ed..597bc11 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -188,20 +188,63 @@ impl GlobalMetrics { } } +/// Bytes a connection may accumulate locally before publishing to the shared listener counters. +/// The whole point of the local buffer is to keep the shared cache line off the per-chunk path +/// (see `CountingStream`), so this wants to be well above `copy_bidirectional`'s 8 KiB buffer — +/// at 256 KiB a saturated connection publishes ~32× less often than it would per chunk, while a +/// scrape still sees a busy connection's traffic within a fraction of a second. +const COUNTER_FLUSH_BYTES: u64 = 256 * 1024; + /// Wraps the *client* side of a proxied session so byte counts accrue as the copy runs rather /// than at completion. Counting `copy_bidirectional`'s return value instead would lose every /// byte of any session that ends by idle timeout or reset — i.e. most long-lived ones. /// +/// Counts are accumulated per connection and published to the shared `ListenerMetrics` only +/// every `COUNTER_FLUSH_BYTES` and on drop. A `fetch_add` per chunk would put a single shared +/// cache line in the path of every 8 KiB of proxied traffic, ping-ponging it across every core — +/// exactly the cross-core contention `SO_REUSEPORT` per worker exists to avoid. +/// /// Because it wraps the client, the direction naming is from the proxy's point of view: bytes /// read here came *from* the client, bytes written here go *to* the client. pub struct CountingStream<'a, S> { inner: S, metrics: &'a ListenerMetrics, + pending_in: u64, + pending_out: u64, } impl<'a, S> CountingStream<'a, S> { pub fn new(inner: S, metrics: &'a ListenerMetrics) -> Self { - Self { inner, metrics } + Self { inner, metrics, pending_in: 0, pending_out: 0 } + } + + fn record_in(&mut self, bytes: u64) { + self.pending_in += bytes; + if self.pending_in >= COUNTER_FLUSH_BYTES { + self.metrics.bytes_in.fetch_add(self.pending_in, Ordering::Relaxed); + self.pending_in = 0; + } + } + + fn record_out(&mut self, bytes: u64) { + self.pending_out += bytes; + if self.pending_out >= COUNTER_FLUSH_BYTES { + self.metrics.bytes_out.fetch_add(self.pending_out, Ordering::Relaxed); + self.pending_out = 0; + } + } +} + +// Publishes whatever is left when the session ends — including when the connection task is +// aborted at shutdown, since that drops the future and with it this stream. +impl Drop for CountingStream<'_, S> { + fn drop(&mut self) { + if self.pending_in > 0 { + self.metrics.bytes_in.fetch_add(self.pending_in, Ordering::Relaxed); + } + if self.pending_out > 0 { + self.metrics.bytes_out.fetch_add(self.pending_out, Ordering::Relaxed); + } } } @@ -213,7 +256,7 @@ impl AsyncRead for CountingStream<'_, S> { if matches!(result, Poll::Ready(Ok(()))) { let read = buf.filled().len().saturating_sub(before); if read > 0 { - this.metrics.bytes_in.fetch_add(read as u64, Ordering::Relaxed); + this.record_in(read as u64); } } result @@ -225,7 +268,7 @@ impl AsyncWrite for CountingStream<'_, S> { let this = self.get_mut(); let result = Pin::new(&mut this.inner).poll_write(cx, buf); if let Poll::Ready(Ok(written)) = &result { - this.metrics.bytes_out.fetch_add(*written as u64, Ordering::Relaxed); + this.record_out(*written as u64); } result } @@ -238,7 +281,7 @@ impl AsyncWrite for CountingStream<'_, S> { let this = self.get_mut(); let result = Pin::new(&mut this.inner).poll_write_vectored(cx, bufs); if let Poll::Ready(Ok(written)) = &result { - this.metrics.bytes_out.fetch_add(*written as u64, Ordering::Relaxed); + this.record_out(*written as u64); } result } @@ -297,7 +340,7 @@ mod tests { } #[tokio::test] - async fn counting_stream_records_both_directions() { + async fn counting_stream_publishes_both_directions_on_drop() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; let metrics = ListenerMetrics::default(); @@ -310,7 +353,40 @@ mod tests { counted.read_exact(&mut buf).await.unwrap(); counted.write_all(b"world!").await.unwrap(); + // Below the flush threshold, so the shared counters stay untouched until the session ends. + assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 0); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 0); + + drop(counted); assert_eq!(metrics.bytes_in.load(Ordering::Relaxed), 5); assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), 6); } + + // A long-lived connection must not withhold its traffic from scrapes until it closes. + #[tokio::test] + async fn counting_stream_publishes_once_past_the_flush_threshold() { + use tokio::io::AsyncWriteExt; + + let metrics = ListenerMetrics::default(); + let (client, mut peer) = tokio::io::duplex(COUNTER_FLUSH_BYTES as usize * 4); + let mut counted = CountingStream::new(client, &metrics); + + let chunk = vec![0u8; 8 * 1024]; + let mut written = 0u64; + while written < COUNTER_FLUSH_BYTES { + counted.write_all(&chunk).await.unwrap(); + written += chunk.len() as u64; + } + + assert_eq!( + metrics.bytes_out.load(Ordering::Relaxed), + written, + "crossing the threshold must publish everything accumulated so far" + ); + + // Drain so the duplex peer doesn't hold the buffer, then confirm drop double-counts nothing. + drop(counted); + peer.shutdown().await.ok(); + assert_eq!(metrics.bytes_out.load(Ordering::Relaxed), written); + } } diff --git a/ts/admin.ts b/ts/admin.ts index b2c2dc5..f5d5cff 100644 --- a/ts/admin.ts +++ b/ts/admin.ts @@ -58,11 +58,18 @@ function labels(pairs: Record): string { return rendered ? `{${rendered}}` : ''; } +/** + * Accumulates samples grouped by metric name. + * + * The exposition format requires every sample of a metric name to be contiguous, under a single + * HELP/TYPE pair. Emitting in call order would interleave them — with more than one proxy + * configured, `symphony_routes` for the second proxy would land after the first proxy's listener + * samples, and strict parsers reject or drop the split group. Grouping here means the callers + * below can stay in the natural proxy → listener iteration order. + */ class Exposition { - private readonly lines: string[] = []; - private declared = new Set(); + private readonly groups = new Map(); - /** Emit HELP/TYPE once per metric name, then a sample. */ sample( name: string, type: 'counter' | 'gauge', @@ -70,15 +77,17 @@ class Exposition { value: number, tags: Record = {} ): void { - if (!this.declared.has(name)) { - this.declared.add(name); - this.lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`); - } - this.lines.push(`${name}${labels(tags)} ${value}`); + let group = this.groups.get(name); + if (!group) this.groups.set(name, (group = { type, help, samples: [] })); + group.samples.push(`${name}${labels(tags)} ${value}`); } toString(): string { - return `${this.lines.join('\n')}\n`; + const lines: string[] = []; + for (const [name, { type, help, samples }] of this.groups) { + lines.push(`# HELP ${name} ${help}`, `# TYPE ${name} ${type}`, ...samples); + } + return `${lines.join('\n')}\n`; } } @@ -248,6 +257,9 @@ export class AdminServer { if (signature === this.signature) return; this.signature = signature; this.config = config ?? null; + // Drop a pending retry from the previous config — otherwise it fires seconds later and + // tears down the listeners this call is about to bind. + this.clearRetry(); await this.closeServers(); if (config && (config.socketPath || config.port !== undefined)) await this.bind(); } From 3c295e839289354fbd522290d656acba288b8ac9 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 18:25:10 -0600 Subject: [PATCH 3/7] Pin the allowed metric label set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Consumers bind this endpoint on loopback, which any local process can reach. That is only acceptable while the labels stay free of tenant identifiers — a per-SNI or per-route label would turn aggregate proxy health into a list of which customers a host fronts. Asserting the allowed set makes adding one a deliberate, visible decision rather than an incidental one. Co-Authored-By: Claude Opus 5 --- __test__/metrics.spec.ts | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts index 72c6212..326ab26 100644 --- a/__test__/metrics.spec.ts +++ b/__test__/metrics.spec.ts @@ -323,6 +323,23 @@ describe('renderPrometheus', () => { assert.ok(!lines.some((l) => /^symphony_listener_blocked_total\{[^}]*\}\s/.test(l) && !l.includes('reason='))); }); + // Consumers bind this endpoint on loopback, which any local process can reach. That is only + // acceptable while the labels stay free of tenant identifiers — a per-SNI or per-route label + // would turn aggregate proxy health into a list of which customers a host fronts. This pins + // the allowed label set so adding one is a deliberate, visible decision. + it('never labels a sample with a tenant identifier', () => { + const allowed = new Set(['version', 'proxy', 'listener', 'mode', 'reason', 'outcome']); + for (const line of lines) { + const labelSet = line.match(/\{(.*)\}/)?.[1]; + if (!labelSet) continue; + // Keys sit at the start or after a comma; a label *value* may itself contain a comma + // (proxy="80,443"), so the boundary has to be matched rather than split on. + for (const [, , key] of labelSet.matchAll(/(^|,)([a-z_]+)="/g)) { + assert.ok(allowed.has(key), `unexpected metric label '${key}' — is it tenant-identifying?`); + } + } + }); + it('carries the version in build_info and timestamps in seconds', () => { assert.ok(lines.includes('symphony_build_info{version="9.9.9"} 1')); assert.ok(lines.includes(`symphony_start_time_seconds ${Date.parse(snapshot.startedAt) / 1000}`)); From 3cd0e18a13d4c6b883813b575ae62ae2891e7c55 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 18:58:16 -0600 Subject: [PATCH 4/7] Address Codex review: socket reclaim safety, derived totals, HTTP byte accounting MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two blockers and four significant findings from the Codex review leg. Stale-socket reclamation could delete a live endpoint. The probe treated every connect error as "nobody is listening", so a live socket that refuses the probe with EACCES — restrictive permissions, or another uid — would be unlinked. It now reclaims only on ECONNREFUSED, and only when the inode is actually a socket: a socketPath misconfigured onto a regular file (status.json, say) would otherwise have been deleted and replaced. Reclamation also had a check/unlink race: two processes could both find a path stale, and the second one's unlink would remove the socket the first had already bound. The bind now happens on a pid-unique temp path that is renamed into place, which is atomic and leaves no window. The precheck stays, so an upgrade overlap still loses and retries rather than stealing a live incumbent. Since Node unlinks the path it bound — the temp one — shutdown now removes the published path itself, guarded by the inode it published there. The exported totals couldn't satisfy their own sum invariant under traffic: total_blocked and its reason counter were two non-atomic writes, so a scrape between them saw a total that disagreed with its breakdown. The invariant held only while the proxy was idle, which is exactly when nobody is looking. Totals are now derived from the same values reported alongside them, which also removes an atomic from the block/error paths and deletes the proxy-wide blocked counter outright — a better fix than the second increment added earlier in this branch. The HTTP-mode listener consumed the request head before any wrapper could see it and wrote its redirects directly, so bytesReceived never moved for an ACME request and no redirect response was counted at all. It now accounts for both explicitly. keepAliveTimeout = 0 disables the timeout rather than keep-alive, so idle scrape connections were never reaped — they could accumulate against the same fd budget the proxy listeners draw from. Set to 5s, with a connection cap. Corrects the byte-counting docs: passthrough routes do count handshake records, because to a passthrough route they are simply part of the forwarded stream. Codex independently flagged the per-chunk atomic on the byte counters, already fixed in 479070d from the Gemini leg. Co-Authored-By: Claude Opus 5 --- CLAUDE.md | 3 +- README.md | 12 ++- __test__/metrics.spec.ts | 62 ++++++++++++++++ src/http_listener.rs | 23 ++++-- src/listener.rs | 3 - src/metrics.rs | 57 ++++++++------ src/proxy.rs | 46 +++++++----- src/proxy_conn.rs | 1 - ts/admin.ts | 156 +++++++++++++++++++++++++++++---------- 9 files changed, 268 insertions(+), 95 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index 882a602..6ce1d70 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -19,8 +19,9 @@ An optional `admin` block in the config file (`{ socketPath?, socketMode?, port? Three properties are load-bearing and easy to regress: - **It must never affect proxying.** A bind failure is logged and retried on a 5s timer instead of throwing out of `doReconcile()`. This is not defensiveness for its own sake: during a version upgrade host-manager runs both processes concurrently (the Rust listeners overlap via `SO_REUSEPORT`, which a Node HTTP server has no equivalent for), so the successor *will* lose the admin bind for a few seconds and must keep serving traffic anyway. -- **A stale Unix socket is reclaimed, a live one is not.** `EADDRINUSE` on a socket path triggers a connect probe; only `ECONNREFUSED` (nobody listening) justifies the `unlink`. Unlinking unconditionally would let a starting process silently steal the endpoint from the running one — the same class of bug as the `status.json` ownership guard. +- **A stale Unix socket is reclaimed, a live one is not.** Three things protect this, and each was a real hole once: the probe counts a path reclaimable only on `ECONNREFUSED` (an `EACCES` from a restrictively-permissioned live socket is not evidence nobody is listening); the inode must actually be a socket (a `socketPath` misconfigured onto a regular file would otherwise be deleted); and the bind happens on a pid-unique temp path that is `rename`d into place, so there is no probe→unlink→bind window for a second process to delete a socket the first has already bound. On shutdown the published path is unlinked only while its inode is still the one we put there. Same family as the `status.json` ownership guard. - **Counters are read per request**, not cached at reconcile, so a scrape never serves numbers frozen at the last config reload. +- **Totals are derived, never maintained alongside their parts.** `blocked`/`errors` are summed from the per-reason values in the same snapshot, and the proxy-wide blocked total from the listener values. A separate `total_blocked` incremented next to its reason counter is two non-atomic writes: a scrape landing between them sees a total that disagrees with its own breakdown, so the invariant would hold only while the proxy is idle — precisely when nobody is reading it. Prometheus shape: blocked/error counts are emitted **only** under their `reason` label (they sum to the unlabeled total, so a separate total would be a second representation of the same number), and the proxy-wide active gauge is `sum without(listener)`. `renderPrometheus` is exported from the package for embedded consumers. diff --git a/README.md b/README.md index e1314fc..1c7baff 100644 --- a/README.md +++ b/README.md @@ -618,8 +618,10 @@ for (const l of m.listeners) { // l.activeConnections, l.accepted // l.bytesReceived — bytes read from clients (client → upstream) // l.bytesSent — bytes written to clients (upstream → client) - // Counted where the proxy sees the bytes: plaintext on a terminated-TLS route, wire bytes - // on a passthrough route. The handshake is not included in either. + // Counted where the proxy sees the bytes. On a terminated-TLS route that is the plaintext + // stream, and the handshake — which precedes the counter — is excluded. On a passthrough + // route the proxy has no plaintext view and forwards wire bytes, so the handshake records + // are part of the stream and are counted. // l.blockedByReason — [{ reason: 'rate_limited', count: 12 }, ...] // l.errorsByReason — [{ reason: 'upstream_connect', count: 3 }, ...] } @@ -632,8 +634,10 @@ const blocked = proxy.blockedIps(); ``` Every reason is reported on every call, including reasons still at zero, so a dashboard series -exists before the first incident rather than appearing mid-outage. The per-reason counts sum to -`l.blocked` / `l.errors` by construction. +exists before the first incident rather than appearing mid-outage. `l.blocked` / `l.errors` are +summed from the very reason values reported alongside them, and `m.blockedConnections` from the +listener values in the same snapshot — so a reading taken mid-traffic is internally consistent +rather than only adding up while the proxy is idle. **Block reasons:** `max_connections`, `cidr_blocked`, `ja3_blocked`, `ja4_blocked`, `incomplete_handshake`, `no_sni`, `rate_limited`, `too_many_connections`, `penalty_boxed`. diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts index 326ab26..3ac30f0 100644 --- a/__test__/metrics.spec.ts +++ b/__test__/metrics.spec.ts @@ -12,6 +12,7 @@ import * as fs from 'node:fs'; import * as http from 'node:http'; import * as os from 'node:os'; import * as path from 'node:path'; +import * as net from 'node:net'; import * as tls from 'node:tls'; import { SymphonyProxy, renderPrometheus, type MetricsSnapshot } from '../ts/index.js'; import { generateSelfSignedCert, getFreePort, startEchoServer, tlsRoundTrip, sleep } from './util.js'; @@ -562,4 +563,65 @@ describe('symphony-server admin endpoint (stale socket recovery)', () => { await echo.close().catch(() => {}); } }); + + // The inverse, and the one that actually costs something to get wrong: a socket that is still + // being served must survive a would-be reclaimer, whatever the probe happens to return. A + // permission-denied probe is not evidence that nobody is listening. + it('leaves a live socket alone even when the probe cannot connect to it', async () => { + const echo = await startEchoServer(); + const livePath = path.join(dir, 'live.sock'); + // Stand in for a running symphony: a real listening socket the reclaimer must not delete. + const incumbent = net.createServer(); + await new Promise((resolve) => incumbent.listen(livePath, resolve)); + const liveIno = fs.statSync(livePath).ino; + + try { + // 0o000 makes connect() fail with EACCES rather than ECONNREFUSED. Treating any probe + // error as "stale" would unlink this live socket. + fs.chmodSync(livePath, 0o000); + + const configPath = path.join(dir, 'contend.json'); + const statusPath = path.join(dir, 'contend-status.json'); + fs.writeFileSync( + configPath, + JSON.stringify({ + version: 1, + admin: { socketPath: livePath }, + proxies: [ + { + listeners: [{ host: '127.0.0.1', port: await getFreePort() }], + routes: [ + { + sni: 'localhost', + upstreams: [{ kind: 'tcp', host: '127.0.0.1', port: echo.port }], + terminateTls: true, + cert: { certChain: cert.cert, privateKey: cert.key }, + }, + ], + }, + ], + }) + ); + + const contender = spawn(process.execPath, [SERVER_JS, '--config', configPath, '--status', statusPath], { + stdio: ['ignore', 'pipe', 'pipe'], + }); + try { + // It boots and proxies regardless — losing the admin bind must never block startup. + await waitFor(() => fs.existsSync(statusPath), 8000); + // Give the retry timer a cycle to (wrongly) reclaim, if it were going to. + await sleep(1000); + + assert.ok(fs.existsSync(livePath), 'the live socket must not have been unlinked'); + assert.equal(fs.statSync(livePath).ino, liveIno, 'the live socket must not have been replaced'); + } finally { + contender.kill('SIGKILL'); + await waitFor(() => contender.exitCode !== null || contender.signalCode !== null, 3000).catch(() => {}); + } + } finally { + fs.chmodSync(livePath, 0o660); + await new Promise((resolve) => incumbent.close(() => resolve())); + await echo.close().catch(() => {}); + } + }); }); diff --git a/src/http_listener.rs b/src/http_listener.rs index b563df4..40fa037 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -92,10 +92,7 @@ async fn accept_loop( let active = ctx.global_metrics.active_connections.load(std::sync::atomic::Ordering::Relaxed); if active >= max_connections as u64 { drop(stream); - // See listener.rs: keep the proxy-level blocked total equal to the - // sum across its listeners. ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); - ctx.global_metrics.inc_blocked(); continue; } } @@ -144,6 +141,12 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc. let Some(host) = host else { - let _ = write_simple_response(&mut stream, b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n").await; + let _ = write_simple_response(&mut stream, b"HTTP/1.1 400 Bad Request\r\nContent-Length: 0\r\nConnection: close\r\n\r\n", &ctx.listener_metrics).await; return; }; let target_str = std::str::from_utf8(target).unwrap_or("/"); @@ -170,6 +173,7 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc std::io::Result<()> { +async fn write_simple_response( + stream: &mut TcpStream, + response: &[u8], + metrics: &crate::metrics::ListenerMetrics, +) -> std::io::Result<()> { + metrics.add_bytes_out(response.len() as u64); stream.write_all(response).await?; stream.shutdown().await } diff --git a/src/listener.rs b/src/listener.rs index 933f391..13142c2 100644 --- a/src/listener.rs +++ b/src/listener.rs @@ -69,10 +69,7 @@ async fn accept_loop( if active >= max_connections as u64 { // Drop the stream — OS will send RST drop(stream); - // Both counters, so the proxy-level blocked total stays equal to the - // sum of its listeners' — the protection path already does both. ctx.listener_metrics.inc_blocked(BlockKind::MaxConnections); - ctx.global_metrics.inc_blocked(); continue; } } diff --git a/src/metrics.rs b/src/metrics.rs index 597bc11..6d374f4 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -83,14 +83,13 @@ labeled_enum!(ErrorKind { pub struct ListenerMetrics { pub active_connections: AtomicU64, pub total_accepted: AtomicU64, - pub total_blocked: AtomicU64, - pub total_errors: AtomicU64, - /// Bytes read from clients on this listener (client → upstream). Counted at the point the - /// proxy sees them, so a terminated-TLS route counts plaintext and a passthrough route - /// counts wire bytes; neither includes the handshake, which precedes the counter. + /// Bytes read from clients on this listener (client → upstream), counted where the proxy + /// sees them. On a terminated-TLS route that is the plaintext stream — the handshake happens + /// before the counter is installed and is excluded. On a passthrough route the proxy has no + /// plaintext view and simply forwards wire bytes, so the handshake records are part of the + /// stream and are counted. pub bytes_in: AtomicU64, - /// Bytes written to clients on this listener (upstream → client). Same framing caveat as - /// `bytes_in`. + /// Bytes written to clients on this listener (upstream → client). Same framing as `bytes_in`. pub bytes_out: AtomicU64, blocked_by_kind: [AtomicU64; BlockKind::COUNT], errors_by_kind: [AtomicU64; ErrorKind::COUNT], @@ -101,8 +100,6 @@ impl Default for ListenerMetrics { Self { active_connections: AtomicU64::new(0), total_accepted: AtomicU64::new(0), - total_blocked: AtomicU64::new(0), - total_errors: AtomicU64::new(0), bytes_in: AtomicU64::new(0), bytes_out: AtomicU64::new(0), blocked_by_kind: std::array::from_fn(|_| AtomicU64::new(0)), @@ -122,12 +119,21 @@ impl ListenerMetrics { } pub fn inc_blocked(&self, kind: BlockKind) { - self.total_blocked.fetch_add(1, Ordering::Relaxed); self.blocked_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); } + /// Direct byte accounting for paths that handle a whole message at once (the HTTP-mode + /// listener), where `CountingStream`'s per-connection batching would buy nothing — these + /// fire once or twice per connection, not per chunk. + pub fn add_bytes_in(&self, bytes: u64) { + self.bytes_in.fetch_add(bytes, Ordering::Relaxed); + } + + pub fn add_bytes_out(&self, bytes: u64) { + self.bytes_out.fetch_add(bytes, Ordering::Relaxed); + } + pub fn inc_error(&self, kind: ErrorKind) { - self.total_errors.fetch_add(1, Ordering::Relaxed); self.errors_by_kind[kind as usize].fetch_add(1, Ordering::Relaxed); } @@ -149,10 +155,24 @@ impl ListenerMetrics { } } +/// Sums a per-reason breakdown into its total. +/// +/// The exported totals are derived from the same values the breakdown reports rather than kept +/// as separate counters. A standalone `total_blocked` incremented next to its reason counter is +/// two non-atomic writes, so a scrape landing between them observes a total that does not equal +/// the sum of its parts — the invariant would hold only while the proxy is idle, which is +/// exactly when nobody is looking. Deriving makes it structural, and removes an atomic from the +/// block/error paths. +pub fn total_of(counts: &[(&'static str, u64)]) -> u64 { + counts.iter().map(|(_, count)| count).sum() +} + +// No proxy-wide blocked counter: it is derived from the listeners in `metrics()` for the same +// reason the per-listener totals are derived from their reasons — a separately incremented copy +// can disagree with its parts under traffic. #[derive(Default)] pub struct GlobalMetrics { pub active_connections: AtomicU64, - pub total_blocked: AtomicU64, pub pending_suspended: AtomicU64, /// Suspended connections that JS resolved with a route. pub suspended_resolved: AtomicU64, @@ -169,10 +189,6 @@ impl GlobalMetrics { self.active_connections.fetch_sub(1, Ordering::Relaxed); } - pub fn inc_blocked(&self) { - self.total_blocked.fetch_add(1, Ordering::Relaxed); - } - pub fn inc_suspended(&self) { self.pending_suspended.fetch_add(1, Ordering::Relaxed); } @@ -326,13 +342,8 @@ mod tests { m.inc_blocked(BlockKind::NoSni); m.inc_error(ErrorKind::UpstreamConnect); - let blocked: u64 = m.blocked_by_reason().iter().map(|(_, v)| v).sum(); - assert_eq!(blocked, m.total_blocked.load(Ordering::Relaxed)); - assert_eq!(blocked, 3); - - let errors: u64 = m.errors_by_reason().iter().map(|(_, v)| v).sum(); - assert_eq!(errors, m.total_errors.load(Ordering::Relaxed)); - assert_eq!(errors, 1); + assert_eq!(total_of(&m.blocked_by_reason()), 3); + assert_eq!(total_of(&m.errors_by_reason()), 1); // Every reason is exported, including the ones still at zero. assert_eq!(m.blocked_by_reason().len(), BlockKind::COUNT); diff --git a/src/proxy.rs b/src/proxy.rs index aadee80..144efea 100644 --- a/src/proxy.rs +++ b/src/proxy.rs @@ -1,6 +1,6 @@ use crate::http_listener::spawn_http_listeners; use crate::listener::spawn_listeners; -use crate::metrics::{GlobalMetrics, ListenerMetrics}; +use crate::metrics::{total_of, GlobalMetrics, ListenerMetrics}; use crate::protection::ProtectionState; use crate::proxy_conn::{ConnContext, JsEvent}; use crate::router::{ @@ -600,30 +600,42 @@ impl SymphonyProxyWrap { // `listeners` and `listener_states` are built in lockstep in the constructor and never // mutated, so index i refers to the same listener in both. - let listeners = self + // + // Each listener's totals are summed from the very reason values it reports, so a scrape + // taken mid-traffic is internally consistent — a separately maintained total would be a + // second write racing the first and could disagree with its own breakdown. + let listeners: Vec = self .listeners .iter() .zip(self.listener_states.iter()) - .map(|(listener, state)| JsListenerMetrics { - address: state.addr.clone(), - mode: match listener.mode { - ListenerMode::Tls => "tls".to_string(), - ListenerMode::Http => "http".to_string(), - }, - active_connections: state.metrics.active_connections.load(Ordering::Relaxed) as f64, - accepted: state.metrics.total_accepted.load(Ordering::Relaxed) as f64, - blocked: state.metrics.total_blocked.load(Ordering::Relaxed) as f64, - errors: state.metrics.total_errors.load(Ordering::Relaxed) as f64, - bytes_received: state.metrics.bytes_in.load(Ordering::Relaxed) as f64, - bytes_sent: state.metrics.bytes_out.load(Ordering::Relaxed) as f64, - blocked_by_reason: labeled_counts(state.metrics.blocked_by_reason()), - errors_by_reason: labeled_counts(state.metrics.errors_by_reason()), + .map(|(listener, state)| { + let blocked_by_reason = state.metrics.blocked_by_reason(); + let errors_by_reason = state.metrics.errors_by_reason(); + JsListenerMetrics { + address: state.addr.clone(), + mode: match listener.mode { + ListenerMode::Tls => "tls".to_string(), + ListenerMode::Http => "http".to_string(), + }, + active_connections: state.metrics.active_connections.load(Ordering::Relaxed) as f64, + accepted: state.metrics.total_accepted.load(Ordering::Relaxed) as f64, + blocked: total_of(&blocked_by_reason) as f64, + errors: total_of(&errors_by_reason) as f64, + bytes_received: state.metrics.bytes_in.load(Ordering::Relaxed) as f64, + bytes_sent: state.metrics.bytes_out.load(Ordering::Relaxed) as f64, + blocked_by_reason: labeled_counts(blocked_by_reason), + errors_by_reason: labeled_counts(errors_by_reason), + } }) .collect(); + // Likewise derived, so the proxy-wide total always equals the sum of the listener values + // in this same snapshot. + let blocked_connections = listeners.iter().map(|l| l.blocked).sum(); + JsProxyMetrics { active_connections: self.global_metrics.active_connections.load(Ordering::Relaxed) as f64, - blocked_connections: self.global_metrics.total_blocked.load(Ordering::Relaxed) as f64, + blocked_connections, pending_suspended: self.global_metrics.pending_suspended.load(Ordering::Relaxed) as f64, suspended_resolved: self.global_metrics.suspended_resolved.load(Ordering::Relaxed) as f64, suspended_unresolved: self.global_metrics.suspended_unresolved.load(Ordering::Relaxed) as f64, diff --git a/src/proxy_conn.rs b/src/proxy_conn.rs index 9c3d5a7..8d10803 100644 --- a/src/proxy_conn.rs +++ b/src/proxy_conn.rs @@ -85,7 +85,6 @@ pub async fn handle(stream: TcpStream, peer_addr: SocketAddr, ctx: Arc { ctx.listener_metrics.inc_blocked(BlockKind::from(&reason)); - ctx.global_metrics.inc_blocked(); emit(&ctx.js_emit, JsEvent::Blocked { ip: peer_ip.to_string(), reason: reason.as_str().to_string(), diff --git a/ts/admin.ts b/ts/admin.ts index f5d5cff..530f86c 100644 --- a/ts/admin.ts +++ b/ts/admin.ts @@ -13,9 +13,20 @@ import { createServer, type Server, type IncomingMessage, type ServerResponse } from 'node:http'; import { connect } from 'node:net'; -import { chmodSync, unlinkSync } from 'node:fs'; +import { chmodSync, existsSync, lstatSync, renameSync, unlinkSync } from 'node:fs'; import type { ProxyMetrics } from './types.js'; +/** One thing to listen on. `precheck` may reject the bind before a server is created. */ +interface BindTarget { + describe: string; + listen: (server: Server) => void; + precheck?: () => Promise; + /** Runs once bound; throwing fails the bind (the caller closes the server and retries). */ + onBound?: () => void; + /** Best-effort removal of anything `listen`/`onBound` left behind on a failed attempt. */ + cleanup?: () => void; +} + export interface AdminConfig { /** Unix socket path to listen on. Relative paths resolve against the config file's directory. */ socketPath?: string; @@ -43,6 +54,9 @@ export interface MetricsSnapshot { } const RETRY_MS = 5_000; +const KEEP_ALIVE_MS = 5_000; +/** Hard ceiling on concurrent admin connections — a scrape endpoint needs a handful at most. */ +const MAX_ADMIN_CONNECTIONS = 16; // ── Prometheus rendering ────────────────────────────────────────────────────── @@ -214,16 +228,37 @@ export function renderPrometheus(snapshot: MetricsSnapshot): string { // ── Server ──────────────────────────────────────────────────────────────────── -/** True if something is actively listening on `socketPath` (as opposed to a stale socket file). */ -function socketIsLive(socketPath: string): Promise { +/** + * Whether `socketPath` is safe to remove and rebind. + * + * Reclaimable means two things, and both must hold — the cost of getting this wrong is deleting + * a *live* endpoint out from under a running process, which is the failure mode symphony's + * status.json ownership guard exists to prevent: + * + * - `ECONNREFUSED` specifically, not merely "the probe failed". A live socket with restrictive + * permissions (or one owned by another uid) refuses the probe with `EACCES`; treating every + * error as stale would unlink it. Anything that isn't a definitive "nobody is listening" + * leaves the path alone and the bind is retried instead. + * - The inode is actually a socket. Left to `EADDRINUSE` alone, a `socketPath` misconfigured + * onto a regular file (say, status.json) would see the connect fail and delete that file. + */ +function socketIsReclaimable(socketPath: string): Promise { return new Promise((resolve) => { + let stats; + try { + stats = lstatSync(socketPath); + } catch { + return resolve(false); // vanished under us — let the bind retry decide + } + if (!stats.isSocket()) return resolve(false); + const probe = connect(socketPath); - const done = (live: boolean) => { + const done = (reclaimable: boolean) => { probe.destroy(); - resolve(live); + resolve(reclaimable); }; - probe.once('connect', () => done(true)); - probe.once('error', () => done(false)); + probe.once('connect', () => done(false)); + probe.once('error', (err: NodeJS.ErrnoException) => done(err.code === 'ECONNREFUSED')); }); } @@ -240,6 +275,8 @@ export class AdminServer { private config: AdminConfig | null = null; private retryTimer: NodeJS.Timeout | null = null; private stopped = false; + /** The socket path this process published, and the inode it published there. */ + private publishedSocket: { path: string; ino: number } | null = null; constructor( snapshot: () => MetricsSnapshot, @@ -294,15 +331,8 @@ export class AdminServer { if (this.stopped || !config) return; await this.closeServers(); - const targets: Array<{ describe: string; listen: (server: Server) => void; onBound?: () => void }> = []; - if (config.socketPath) { - const path = config.socketPath; - targets.push({ - describe: path, - listen: (server) => server.listen(path), - onBound: () => chmodSync(path, config.socketMode ?? 0o660), - }); - } + const targets: BindTarget[] = []; + if (config.socketPath) targets.push(this.socketTarget(config.socketPath, config.socketMode ?? 0o660)); if (config.port !== undefined) { const host = config.host ?? '127.0.0.1'; targets.push({ describe: `${host}:${config.port}`, listen: (server) => server.listen(config.port, host) }); @@ -310,7 +340,7 @@ export class AdminServer { for (const target of targets) { try { - const server = await this.listenOne(target, config); + const server = await this.listenOne(target); this.servers.push(server); this.log(`admin endpoint listening on ${target.describe}`); } catch (err) { @@ -324,16 +354,59 @@ export class AdminServer { } } - private async listenOne( - target: { describe: string; listen: (server: Server) => void; onBound?: () => void }, - config: AdminConfig - ): Promise { + /** + * Bind the Unix socket by listening on a pid-unique temporary path and `rename`-ing it onto + * the real one. + * + * The obvious shape — probe, `unlink`, `listen` — has a window between the probe and the + * unlink. Two processes can both find the path stale, and the second one's unlink then + * deletes the socket the first has already bound and is serving. `rename` is atomic and + * replaces the target in one step, so the path always names a socket somebody is listening + * on. The precheck stays, because rename would otherwise happily clobber a *live* incumbent + * during an upgrade overlap — there we want to lose and retry, not steal the endpoint. + */ + private socketTarget(path: string, mode: number): BindTarget { + const tempPath = `${path}.${process.pid}`; + return { + describe: path, + precheck: async () => { + if (existsSync(path) && !(await socketIsReclaimable(path))) { + throw new Error(`${path} is in use by another process`); + } + }, + listen: (server) => server.listen(tempPath), + onBound: () => { + chmodSync(tempPath, mode); + renameSync(tempPath, path); + // server.close() unlinks the path it bound — the temp one — so the published path + // is ours to clean up. Record the inode to do that safely: if another process has + // since renamed its own socket over this path, the inode differs and we leave it + // alone rather than deleting a live endpoint (cf. the status.json ownership guard). + this.publishedSocket = { path, ino: lstatSync(path).ino }; + }, + cleanup: () => { + try { + unlinkSync(tempPath); + } catch { + // never created, or already renamed into place + } + }, + }; + } + + private async listenOne(target: BindTarget): Promise { + await target.precheck?.(); + const server = createServer((req, res) => this.handle(req, res)); - // Metrics scrapes are short; don't hold sockets open between them. - server.keepAliveTimeout = 0; + // Scrapes are short and infrequent, so reap idle connections quickly. Note this is a + // timeout, not a switch: setting it to 0 would disable the reaping, letting a client park + // arbitrarily many idle connections against the same process-wide fd budget the proxy + // listeners draw from. maxConnections caps that regardless of client behaviour. + server.keepAliveTimeout = KEEP_ALIVE_MS; + server.maxConnections = MAX_ADMIN_CONNECTIONS; - const attempt = (): Promise => - new Promise((resolve, reject) => { + try { + return await new Promise((resolve, reject) => { const onError = (err: NodeJS.ErrnoException) => { server.removeListener('listening', onListening); reject(err); @@ -343,7 +416,11 @@ export class AdminServer { try { target.onBound?.(); } catch (err) { - this.logErr(`could not set permissions on ${target.describe}:`, (err as Error).message); + // The socket is bound but not reachable at its published path — useless, and + // it would leak an fd and a temp inode. Fail so the retry starts clean. + server.close(); + reject(err); + return; } // Post-bind errors must not crash the process. server.on('error', (err) => this.logErr(`admin endpoint error (${target.describe}):`, err)); @@ -353,20 +430,8 @@ export class AdminServer { server.once('listening', onListening); target.listen(server); }); - - try { - return await attempt(); } catch (err) { - // A Unix socket left behind by a process that died without cleaning up blocks the - // bind forever. Only remove it once a connect probe proves nobody is listening — - // unlinking a live socket would silently steal the endpoint from a running process. - const code = (err as NodeJS.ErrnoException).code; - if (code === 'EADDRINUSE' && config.socketPath && target.describe === config.socketPath) { - if (await socketIsLive(config.socketPath)) throw err; - this.log(`removing stale admin socket ${config.socketPath}`); - unlinkSync(config.socketPath); - return attempt(); - } + target.cleanup?.(); throw err; } } @@ -414,5 +479,18 @@ export class AdminServer { }) ) ); + this.unpublishSocket(); + } + + /** Remove the published socket path, but only while it still names the inode we put there. */ + private unpublishSocket(): void { + const published = this.publishedSocket; + this.publishedSocket = null; + if (!published) return; + try { + if (lstatSync(published.path).ino === published.ino) unlinkSync(published.path); + } catch { + // already gone, or replaced by a successor — either way, not ours to remove + } } } From 3b8089e9462186ad329957cbe6e7b27567acae3a Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 19:03:14 -0600 Subject: [PATCH 5/7] Final-artifact pass: count response bytes after the write, document the unlink race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two low-severity residuals from the final review pass. The HTTP listener counted response bytes before writing them, so a partial or failed write reported bytes the client never received. Counted after a successful write instead. Documents that unpublishSocket()'s inode guard narrows the shutdown-vs-rename race rather than closing it — the check and the unlink are two syscalls. Closing it would need file locking, and the exposure is one observability endpoint that self-heals on the next reconcile, unlike the probe->unlink->bind window it replaced, which could strand a live endpoint indefinitely. Co-Authored-By: Claude Opus 5 --- src/http_listener.rs | 9 ++++++--- ts/admin.ts | 12 +++++++++++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/src/http_listener.rs b/src/http_listener.rs index 40fa037..0442141 100644 --- a/src/http_listener.rs +++ b/src/http_listener.rs @@ -173,8 +173,11 @@ async fn handle_http(mut stream: TcpStream, peer_addr: SocketAddr, ctx: Arc std::io::Result<()> { - metrics.add_bytes_out(response.len() as u64); stream.write_all(response).await?; + metrics.add_bytes_out(response.len() as u64); stream.shutdown().await } diff --git a/ts/admin.ts b/ts/admin.ts index 530f86c..ed88e20 100644 --- a/ts/admin.ts +++ b/ts/admin.ts @@ -482,7 +482,17 @@ export class AdminServer { this.unpublishSocket(); } - /** Remove the published socket path, but only while it still names the inode we put there. */ + /** + * Remove the published socket path, but only while it still names the inode we put there. + * + * The check and the unlink are two syscalls, so this narrows the race rather than closing it: + * a successor that renames its socket into place between them still loses its published path + * (it keeps serving an unlinked inode until its next reconcile republishes). Closing that + * would need file locking, and the exposure is one observability endpoint that self-heals — + * unlike the probe→unlink→bind window this replaced, which could strand a live endpoint + * indefinitely. The same residual applies if two *fresh* publishers race with no live + * incumbent; host-manager's single-successor upgrade model doesn't produce that. + */ private unpublishSocket(): void { const published = this.publishedSocket; this.publishedSocket = null; From 976b135cd3a437e618b75509889ce526aace470e Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 22:15:14 -0600 Subject: [PATCH 6/7] Address draft review: stop unlinking the published socket, serialize admin ops MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three findings from the draft review plus three from the CI review bot. Shutdown no longer unlinks the published socket path. The inode guard narrowed the check-then-unlink race but could not close it, and the loss it risked is not self-repairing as the previous comment claimed: update() returns early on an unchanged signature, so a successor whose socket got unlinked would keep serving an unreachable path until its config changed or it restarted. Leaving the pathname costs one inode in a 0o700 directory, and the next binder replaces it atomically after proving it stale. The review bot separately suggested reading the inode before the rename to make the guard exact; that fixes a different variant of the same hazard, and removing the unlink subsumes both. update(), stop(), and the retry timer all mutate the same listeners across awaits, and the retry ran fire-and-forget. A retry that was mid-bind could publish its server after a later update() or stop() had finished, and with a UDS target could leave two paths live. They now run through one operation chain. renderPrometheus and MetricsSnapshot are no longer exported from the package root. A snapshot carries the standalone server's pid, timestamps, and port-set grouping, which an embedded consumer would have to synthesise; exporting it made that an API commitment with no caller asking for one. The reclaimability probe now has a timeout — a frozen owner or a full accept backlog would otherwise hang it and stall the reconcile that called it. Test teardowns checked only exitCode, which stays null for a signal-terminated child, so every cleanup block waited out its 3s timeout. Co-Authored-By: Claude Opus 5 --- README.md | 18 +++----- __test__/metrics.spec.ts | 28 +++++++++---- ts/admin.ts | 89 ++++++++++++++++++++-------------------- ts/index.ts | 8 +++- 4 files changed, 76 insertions(+), 67 deletions(-) diff --git a/README.md b/README.md index 1c7baff..e01c312 100644 --- a/README.md +++ b/README.md @@ -698,19 +698,11 @@ traffic through `SO_REUSEPORT` — the successor picks up the admin endpoint onc exits. A socket file left behind by a `SIGKILL`ed process is reclaimed automatically, but only after a connect probe proves nobody is listening on it. -The renderer is exported for consumers that want the same output from an embedded proxy: - -```typescript -import { renderPrometheus } from '@harperfast/symphony'; - -const text = renderPrometheus({ - pid: process.pid, - version: '0.5.0', - startedAt, - reloadedAt, - proxies: [{ ports: '80,443', metrics: proxy.metrics() }], -}); -``` +The Prometheus renderer is internal to the standalone server and is not exported from the +package root: a snapshot carries that process's pid, timestamps, and port-set grouping, which an +embedded consumer would have to synthesise. An embedded proxy has `proxy.metrics()` directly. If +a caller genuinely needs Prometheus text from an embedded proxy, open an issue — the right shape +is a `ProxyMetrics`-based renderer, not this one. --- diff --git a/__test__/metrics.spec.ts b/__test__/metrics.spec.ts index 3ac30f0..0f4dc8d 100644 --- a/__test__/metrics.spec.ts +++ b/__test__/metrics.spec.ts @@ -14,7 +14,9 @@ import * as os from 'node:os'; import * as path from 'node:path'; import * as net from 'node:net'; import * as tls from 'node:tls'; -import { SymphonyProxy, renderPrometheus, type MetricsSnapshot } from '../ts/index.js'; +import { SymphonyProxy } from '../ts/index.js'; +// Not exported from the package root — the admin endpoint owns this shape (see ts/index.ts). +import { renderPrometheus, type MetricsSnapshot } from '../ts/admin.js'; import { generateSelfSignedCert, getFreePort, startEchoServer, tlsRoundTrip, sleep } from './util.js'; const SERVER_JS = path.join(__dirname, '..', 'ts', 'server.js'); @@ -410,9 +412,11 @@ describe('symphony-server admin endpoint', () => { after(async () => { shuttingDown = true; - if (child && child.exitCode === null) { + if (child && child.exitCode === null && child.signalCode === null) { child.kill('SIGTERM'); - await waitFor(() => child.exitCode !== null, 3000).catch(() => child.kill('SIGKILL')); + await waitFor(() => child.exitCode !== null || child.signalCode !== null, 3000).catch(() => + child.kill('SIGKILL') + ); } await echo.close().catch(() => {}); fs.rmSync(dir, { recursive: true, force: true }); @@ -473,11 +477,19 @@ describe('symphony-server admin endpoint', () => { assert.equal(res.status, 200); }); - it('removes the unix socket on shutdown', async () => { + // Shutdown deliberately does NOT unlink the published path — any check-then-unlink can delete + // a successor's live socket in the window between the two syscalls. What has to hold is that + // the endpoint stops answering; reclaiming the pathname is the next binder's job, atomically. + it('stops serving on shutdown and leaves the path safely reclaimable', async () => { shuttingDown = true; child.kill('SIGTERM'); - await waitFor(() => child.exitCode !== null, 5000); - assert.equal(fs.existsSync(socketPath), false, 'the admin socket must not be left behind'); + await waitFor(() => child.exitCode !== null || child.signalCode !== null, 5000); + + await assert.rejects( + () => get({ socketPath }, '/health'), + (err: NodeJS.ErrnoException) => err.code === 'ECONNREFUSED', + 'a dead endpoint must refuse connections, not hang or answer' + ); }); }); @@ -519,9 +531,9 @@ describe('symphony-server admin endpoint (stale socket recovery)', () => { }); after(async () => { - if (survivor && survivor.exitCode === null) { + if (survivor && survivor.exitCode === null && survivor.signalCode === null) { survivor.kill('SIGKILL'); - await waitFor(() => survivor!.exitCode !== null, 3000).catch(() => {}); + await waitFor(() => survivor!.exitCode !== null || survivor!.signalCode !== null, 3000).catch(() => {}); } fs.rmSync(dir, { recursive: true, force: true }); }); diff --git a/ts/admin.ts b/ts/admin.ts index ed88e20..8efe315 100644 --- a/ts/admin.ts +++ b/ts/admin.ts @@ -55,6 +55,7 @@ export interface MetricsSnapshot { const RETRY_MS = 5_000; const KEEP_ALIVE_MS = 5_000; +const PROBE_TIMEOUT_MS = 1_000; /** Hard ceiling on concurrent admin connections — a scrape endpoint needs a handful at most. */ const MAX_ADMIN_CONNECTIONS = 16; @@ -253,10 +254,19 @@ function socketIsReclaimable(socketPath: string): Promise { if (!stats.isSocket()) return resolve(false); const probe = connect(socketPath); + let settled = false; const done = (reclaimable: boolean) => { + if (settled) return; + settled = true; + clearTimeout(timer); probe.destroy(); resolve(reclaimable); }; + // A frozen owner or a full accept backlog can leave the connect hanging, which would + // stall the reconcile that called us. Treat a hang as "not reclaimable" — something is + // there, and refusing to touch it is the safe reading. + const timer = setTimeout(() => done(false), PROBE_TIMEOUT_MS); + timer.unref(); probe.once('connect', () => done(false)); probe.once('error', (err: NodeJS.ErrnoException) => done(err.code === 'ECONNREFUSED')); }); @@ -275,8 +285,10 @@ export class AdminServer { private config: AdminConfig | null = null; private retryTimer: NodeJS.Timeout | null = null; private stopped = false; - /** The socket path this process published, and the inode it published there. */ - private publishedSocket: { path: string; ino: number } | null = null; + // update(), stop(), and the retry timer all mutate the same listeners, and each of them + // awaits. Run them through one chain so a retry that is mid-bind can't publish its server + // after a later update() or stop() has already finished tearing things down. + private queue: Promise = Promise.resolve(); constructor( snapshot: () => MetricsSnapshot, @@ -288,23 +300,32 @@ export class AdminServer { this.logErr = logErr; } - async update(config: AdminConfig | undefined): Promise { - if (this.stopped) return; - const signature = JSON.stringify(config ?? null); - if (signature === this.signature) return; - this.signature = signature; - this.config = config ?? null; - // Drop a pending retry from the previous config — otherwise it fires seconds later and - // tears down the listeners this call is about to bind. - this.clearRetry(); - await this.closeServers(); - if (config && (config.socketPath || config.port !== undefined)) await this.bind(); + /** Serialize a mutation of the listener set; failures never break the chain. */ + private enqueue(operation: () => Promise): Promise { + this.queue = this.queue.then(operation, operation); + return this.queue; + } + + update(config: AdminConfig | undefined): Promise { + return this.enqueue(async () => { + if (this.stopped) return; + const signature = JSON.stringify(config ?? null); + if (signature === this.signature) return; + this.signature = signature; + this.config = config ?? null; + // Drop a pending retry from the previous config — otherwise it fires seconds later and + // tears down the listeners this call is about to bind. + this.clearRetry(); + await this.closeServers(); + if (config && (config.socketPath || config.port !== undefined)) await this.bind(); + }); } - async stop(): Promise { + stop(): Promise { + // Set before queueing, so an operation already waiting its turn bails instead of binding. this.stopped = true; this.clearRetry(); - await this.closeServers(); + return this.enqueue(() => this.closeServers()); } private clearRetry(): void { @@ -321,7 +342,8 @@ export class AdminServer { if (this.stopped || this.retryTimer) return; this.retryTimer = setTimeout(() => { this.retryTimer = null; - void this.bind(); + // Queued like everything else, so a retry can't interleave with an update or a stop. + void this.enqueue(() => this.bind()); }, RETRY_MS); this.retryTimer.unref(); } @@ -364,6 +386,13 @@ export class AdminServer { * replaces the target in one step, so the path always names a socket somebody is listening * on. The precheck stays, because rename would otherwise happily clobber a *live* incumbent * during an upgrade overlap — there we want to lose and retry, not steal the endpoint. + * + * Nothing unlinks the published path, including on a clean shutdown. Any check-then-unlink + * can delete a *successor's* socket in the window between the two syscalls, and that loss is + * not self-repairing: `update()` returns early on an unchanged signature, so the successor + * would keep serving an unreachable socket until its config changed or it restarted. Leaving + * a stale pathname behind costs one inode in a 0o700 directory, and the next binder replaces + * it atomically after proving it stale. Cheap litter beats a silently dead endpoint. */ private socketTarget(path: string, mode: number): BindTarget { const tempPath = `${path}.${process.pid}`; @@ -378,11 +407,6 @@ export class AdminServer { onBound: () => { chmodSync(tempPath, mode); renameSync(tempPath, path); - // server.close() unlinks the path it bound — the temp one — so the published path - // is ours to clean up. Record the inode to do that safely: if another process has - // since renamed its own socket over this path, the inode differs and we leave it - // alone rather than deleting a live endpoint (cf. the status.json ownership guard). - this.publishedSocket = { path, ino: lstatSync(path).ino }; }, cleanup: () => { try { @@ -479,28 +503,5 @@ export class AdminServer { }) ) ); - this.unpublishSocket(); - } - - /** - * Remove the published socket path, but only while it still names the inode we put there. - * - * The check and the unlink are two syscalls, so this narrows the race rather than closing it: - * a successor that renames its socket into place between them still loses its published path - * (it keeps serving an unlinked inode until its next reconcile republishes). Closing that - * would need file locking, and the exposure is one observability endpoint that self-heals — - * unlike the probe→unlink→bind window this replaced, which could strand a live endpoint - * indefinitely. The same residual applies if two *fresh* publishers race with no live - * incumbent; host-manager's single-successor upgrade model doesn't produce that. - */ - private unpublishSocket(): void { - const published = this.publishedSocket; - this.publishedSocket = null; - if (!published) return; - try { - if (lstatSync(published.path).ino === published.ino) unlinkSync(published.path); - } catch { - // already gone, or replaced by a successor — either way, not ours to remove - } } } diff --git a/ts/index.ts b/ts/index.ts index cc0502c..a128ef7 100644 --- a/ts/index.ts +++ b/ts/index.ts @@ -22,5 +22,9 @@ export type { SuspendedEvent, ErrorEvent, } from './types.js'; -export { renderPrometheus } from './admin.js'; -export type { AdminConfig, MetricsSnapshot, ProxySnapshot } from './admin.js'; +// `renderPrometheus` and the standalone server's snapshot shape are deliberately NOT exported +// from the package root. They are an implementation detail of the symphony-server admin +// endpoint — a snapshot carries that process's pid, timestamps, and port-set grouping, which an +// embedded consumer would have to synthesise. Exporting them would make that shape a +// compatibility commitment with no caller asking for it; add a `ProxyMetrics`-based renderer +// interface if one ever does. From 1d8ef00745fa3279894495a312d4937d81c3aeb6 Mon Sep 17 00:00:00 2001 From: Kris Zyp Date: Mon, 27 Jul 2026 22:16:08 -0600 Subject: [PATCH 7/7] Document that idle_timeout currently counts a duration cap The review bot flagged that forward()'s tokio::time::timeout is a hard deadline that does not reset on I/O, so idleTimeoutMs terminates busy connections at their total duration rather than after silence. That is pre-existing on main and out of scope here, but this branch is what names the metric, so the label would mislead without saying so. Filed as #34; noted where the reason is defined and in the README's reason list. Co-Authored-By: Claude Opus 5 --- README.md | 4 ++++ src/metrics.rs | 5 ++++- 2 files changed, 8 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index e01c312..5ed115a 100644 --- a/README.md +++ b/README.md @@ -645,6 +645,10 @@ rather than only adding up while the proxy is idle. **Error reasons:** `no_route`, `route_rate_limited`, `suspend_unresolved`, `tls_handshake`, `tls_missing_cert`, `upstream_connect`, `idle_timeout`, `stream`, `http_header`. +> `idle_timeout` counts terminations by `idleTimeoutMs`, which today is a *total duration* cap +> rather than an idleness one — see [#34](https://github.com/HarperFast/symphony/issues/34). Busy +> long-lived connections land in this bucket, not just quiet ones. + ### Out-of-process (`symphony-server` admin endpoint) When symphony runs as its own process there is no JS API to call, so the server bin can expose diff --git a/src/metrics.rs b/src/metrics.rs index 6d374f4..c416154 100644 --- a/src/metrics.rs +++ b/src/metrics.rs @@ -70,7 +70,10 @@ labeled_enum!(ErrorKind { TlsMissingCert => "tls_missing_cert", /// Could not establish the upstream connection. UpstreamConnect => "upstream_connect", - /// The proxied session hit the idle timeout. + /// The proxied session hit `idleTimeoutMs`. Note that today this fires on *total* duration, + /// not idleness — `forward()` wraps the copy in a hard `tokio::time::timeout` that does not + /// reset on activity (issue #34, pre-existing). Until that is fixed this counts busy + /// connections cut at the deadline, not quiet ones. IdleTimeout => "idle_timeout", /// I/O error while proxying an established session. Stream => "stream",