fix(websocket): bound what a slow client can cost the process

.changeset/ws-backpressure.md

@@ -0,0 +1,29 @@
+---
+'lens': minor
+---
+
+Bound what a slow websocket client can cost the process.
+
+`WebSocket#send()` neither blocks nor throws when a peer is slow, it queues.
+Nothing inspected `bufferedAmount`, so one subscriber that stopped reading — a
+suspended mobile client, a wedged TCP window — accumulated price updates in its
+socket buffer without limit, and the message was counted as sent the moment it
+was queued. With the feed pushing continuously, one stuck subscriber was enough
+to grow the process until it was OOM-killed, taking every other subscriber and
+every HTTP client with it.
+
+`FanOutManager` now applies a per-client policy in
+[`src/ws/backpressure.ts`](https://github.com/Divinity-Labs/Lens/blob/main/src/ws/backpressure.ts):
+past `WS_SLOW_CLIENT_BUFFER_BYTES` (256 KiB) updates are coalesced to the newest
+one, past `WS_MAX_BUFFERED_BYTES` (1 MiB) — or congested for longer than
+`WS_MAX_CONGESTED_MS` (30s) — the client is closed with `1013` and counted in
+`ws_clients_dropped_total`. Superseded updates count in
+`ws_messages_dropped_total`, so a truncated feed is visible rather than silent,
+and `ws_buffered_bytes` reports total queued bytes. All the limits are
+configurable, and none of them can slow down delivery to any other client.
+
+`ws_messages_sent_total` now increments when the socket reports the write
+completed rather than when `send()` returned, so it counts what was written
+rather than what was queued. The websocket metrics also moved from
+prom-client's global default registry onto the one `/metrics` renders, which is
+where they were supposed to be visible all along.

.env.example

@@ -126,6 +126,25 @@ AMM_PAGE_SIZE=200
 # How often to poll Soroswap pool reserves in milliseconds (default: 60000 = 60s)
 SOROSWAP_POLL_INTERVAL_MS=60000
 
+# --- WebSocket backpressure (/ws) ---
+# See docs/websocket-backpressure.md for the full policy.
+# Max price updates per second per client (default: 100). 0 disables the window.
+WS_COALESCE_INTERVAL_MS=100
+# Bytes a client socket may have queued before updates are coalesced to the
+# newest one instead of sent (default: 262144 = 256 KiB). Clamped to half of
+# WS_MAX_BUFFERED_BYTES if it is set higher.
+WS_SLOW_CLIENT_BUFFER_BYTES=262144
+# Bytes a client socket may have queued before it is disconnected with close
+# code 1013 (default: 1048576 = 1 MiB).
+WS_MAX_BUFFERED_BYTES=1048576
+# How long a client may stay congested before it is disconnected anyway
+# (default: 30000). Coalescing bounds the memory a slow client costs; this
+# bounds the time. 0 keeps congested clients indefinitely.
+WS_MAX_CONGESTED_MS=30000
+# How often the aggregate ws_buffered_bytes gauge is refreshed in milliseconds
+# (default: 5000).
+WS_BUFFER_SWEEP_INTERVAL_MS=5000
+
 # --- Auth & Security ---
 # Secret key required for admin endpoints (e.g., adding/removing pairs)
 ADMIN_API_KEY=super-secret-admin-key

README.md

@@ -137,6 +137,11 @@ histogram_quantile(0.95,
 See [`docs/http-metrics.md`](docs/http-metrics.md) for the full label reference,
 the bucket rationale and suggested alerting rules.
 
+The websocket stream carries its own signals — `ws_clients_total`,
+`ws_messages_sent_total`, `ws_messages_dropped_total`, `ws_clients_dropped_total`
+and `ws_buffered_bytes` — which exist to make slow subscribers visible instead
+of silent. See [`docs/websocket-backpressure.md`](docs/websocket-backpressure.md).
+
 The ingestion metrics are labelled by `network` as well, so a dual-network
 deployment reports each network separately:
 
@@ -227,6 +232,15 @@ spending warning, and worked example before running.
 
 Full documentation: [examples/mcp-server/README.md](examples/mcp-server/README.md)
 
+### WebSocket backpressure
+
+`/ws` counts a message as sent only once the socket reports the write completed,
+and bounds what a client that stops reading can cost: past 256 KiB of queued
+bytes updates are coalesced to the newest one, past 1 MiB — or congested for 30s
+— the client is closed with `1013` and counted. A truncated feed is visible in
+`ws_messages_dropped_total` rather than silent, and one slow client cannot affect
+any other. See [`docs/websocket-backpressure.md`](docs/websocket-backpressure.md).
+
 ## Usage Examples
 
 Lens gates `/price`, `/pools`, and `/candles` behind x402 micropayments on Stellar (testnet by default). The `/status` endpoint is free.
@@ -350,6 +364,8 @@ Or interactively at [http://localhost:3002/graphiql](http://localhost:3002/graph
 Detailed system design and data flow diagrams can be found in the [Architecture Overview](docs/architecture.md).
 The API specification is available in [OpenAPI 3.0 format](openapi.yaml) and is auto-published to GitHub Pages at https://miracle656.github.io/lens/openapi.json.
 
+Operational docs: [HTTP metrics](docs/http-metrics.md), [ingest metrics](docs/ingest-metrics.md), [websocket backpressure](docs/websocket-backpressure.md), [GraphQL](docs/graphql.md).
+
 ## Examples
 
 The [oracle relay example](examples/oracle-relay/README.md) shows a minimal Soroban contract plus a Node relay that reads Lens prices and pushes them on chain.

docs/http-metrics.md

@@ -58,10 +58,14 @@ So:
   need for an exact code appears, add it then — it is easy to add a label and
   painful to remove one dashboards depend on.
 - **Websocket pairs use fixed buckets.** `ws_clients_total`,
-  `ws_messages_sent_total` and `ws_messages_dropped_total` retain their `pair`
-  label, but pair-specific subscriptions are recorded as `other`; `*` remains
-  the aggregate across all subscriptions. Arbitrary pair names and issuer
-  addresses stay in fanout state only and cannot create Prometheus series.
+  `ws_messages_sent_total`, `ws_messages_dropped_total` and
+  `ws_clients_dropped_total` retain their `pair` label, but pair-specific
+  subscriptions are recorded as `other`; `*` remains the aggregate across all
+  subscriptions. Arbitrary pair names and issuer addresses stay in fanout state
+  only and cannot create Prometheus series. `ws_buffered_bytes` is aggregate-only
+  (`*`): one series per connected client would be unbounded cardinality driven by
+  caller-supplied subscription strings. See
+  [`websocket-backpressure.md`](websocket-backpressure.md).
 
 ## Bucket choice
 

docs/websocket-backpressure.md

@@ -0,0 +1,122 @@
+# WebSocket backpressure
+
+Lens streams live prices over `/ws`. One property of that stream decides how the
+server has to behave: **`send()` does not block and does not throw when a peer
+is slow.** It queues. A client on a bad mobile connection, or one whose process
+has been suspended and whose TCP window has filled, accumulates payloads in its
+socket's internal buffer. Nothing inspected `bufferedAmount`, so that buffer had
+no upper bound, and the message was counted as sent the moment it was queued.
+
+With a price feed pushing updates continuously, one stuck subscriber was enough
+to grow the process until it was OOM-killed — taking every other subscriber, and
+every HTTP client, with it.
+
+The fix is a policy over the socket's `bufferedAmount`, applied per client. It
+lives in [`src/ws/fanout.ts`](../src/ws/fanout.ts) (`FanOutManager.deliver`);
+the limits and their defaults are in
+[`src/ws/backpressure.ts`](../src/ws/backpressure.ts).
+
+## The policy
+
+Checked per client, per update, in this order:
+
+| Tier | Condition | Action |
+|---|---|---|
+| 1. Evict (dead peer) | `bufferedAmount >= WS_MAX_BUFFERED_BYTES` | Close with **1013**, unregister, count `ws_clients_dropped_total`. |
+| 2. Evict (stuck peer) | congested continuously for `WS_MAX_CONGESTED_MS` | Close with **1013**, unregister, count `ws_clients_dropped_total`. |
+| 3. Coalesce | `bufferedAmount >= WS_SLOW_CLIENT_BUFFER_BYTES`, or inside the coalescing window | Keep only the newest update; count every superseded one in `ws_messages_dropped_total`. |
+| 4. Send | otherwise | Send, and count `ws_messages_sent_total` when the socket reports the write completed. |
+
+**Why 1013.** RFC 6455 §7.4.1 reserves 1013 for "Try Again Later": the server is
+refusing to keep serving this client right now. The condition here is the
+client's rather than the server's, but the meaning to a client is the one it
+already handles for a throttled feed — back off and reconnect, do not treat the
+close as fatal. The close reason is one of `buffer limit exceeded` or
+`congested too long`, kept short because RFC 6455 caps a close reason at 123
+bytes.
+
+**Why coalescing is the primary tier.** For a price feed, an intermediate tick
+that arrives late is worthless: the newest tick has already superseded it, so
+truncating the backlog costs the client nothing it could still use — and costs
+the server one payload per client instead of one per tick. Tier 3 also covers
+the ordinary coalescing window (`WS_COALESCE_INTERVAL_MS`, 10 updates/sec by
+default), which existed before and is the same behaviour for the same reason.
+
+**Why evict at all.** Coalescing bounds the memory a slow client costs. It does
+not bound the *time*: a client that has stopped reading entirely would sit
+there forever, holding a slot, its buffered bytes and a `ws_clients_total`
+series, with no metric saying why. Tier 1 catches a peer wedged past the hard
+limit; tier 2 catches one wedged below it.
+
+**Why the soft limit is clamped.** `WS_SLOW_CLIENT_BUFFER_BYTES` is forced below
+`WS_MAX_BUFFERED_BYTES` (at half of it). Set at or above, every client past the
+soft limit would be disconnected on its first update over it and tier 3 — the
+one that keeps a merely-slow client connected — would never run.
+
+## Defaults
+
+| Variable | Default | Meaning |
+|---|---|---|
+| `WS_COALESCE_INTERVAL_MS` | `100` | Max updates per second per client. `0` disables the window. |
+| `WS_SLOW_CLIENT_BUFFER_BYTES` | `262144` (256 KiB) | Queued bytes above which updates are coalesced. Over a thousand ticks of backlog. |
+| `WS_MAX_BUFFERED_BYTES` | `1048576` (1 MiB) | Queued bytes above which the client is closed. |
+| `WS_MAX_CONGESTED_MS` | `30000` | How long a client may stay congested. `0` keeps them indefinitely. |
+| `WS_BUFFER_SWEEP_INTERVAL_MS` | `5000` | How often `ws_buffered_bytes` is refreshed. |
+
+## Counting, honestly
+
+- `ws_messages_sent_total` increments **when the socket reports the write
+  completed**, not when `send()` was called. `send()` returns once the payload is
+  queued, so counting there reported messages as delivered when they were only in
+  a buffer — including for a client that had stopped reading and would never see
+  them.
+- `ws_messages_dropped_total` counts updates discarded because a newer one
+  superseded them, plus the retained update dropped when a client is evicted.
+- `ws_clients_dropped_total` counts evicted clients.
+- `ws_buffered_bytes` is the aggregate of every client's queued bytes, sampled on
+  a timer. Aggregate only: one series per client would be unbounded cardinality
+  driven by caller-supplied subscription strings.
+
+All four keep the existing bounded `pair` label: pair-specific subscriptions are
+recorded as `other`, `*` is the aggregate. See
+[`http-metrics.md`](http-metrics.md).
+
+## Queries
+
+```promql
+# Clients evicted for backpressure, per second
+sum(rate(ws_clients_dropped_total[5m]))
+
+# Updates coalesced away, as a fraction of everything we tried to send
+sum(rate(ws_messages_dropped_total[5m]))
+  / sum(rate(ws_messages_sent_total[5m])) + sum(rate(ws_messages_dropped_total[5m]))
+
+# Total queued bytes across every client — the thing that used to grow unbounded
+ws_buffered_bytes{pair="*"}
+
+# Connected clients
+sum(ws_clients_total{pair="*"})
+```
+
+## Suggested alert
+
+```yaml
+groups:
+  - name: lens-websocket
+    rules:
+      - alert: LensWebsocketClientsDropped
+        expr: sum(rate(ws_clients_dropped_total[10m])) > 0.1
+        for: 10m
+        labels: { severity: warning }
+        annotations:
+          summary: "Lens is disconnecting websocket clients for falling behind"
+          description: >-
+            ws_clients_dropped_total is rising: clients are not draining their
+            socket buffers fast enough. Check ws_buffered_bytes and the network
+            between clients and this instance before raising the thresholds.
+```
+
+Tune against observed traffic before paging anyone. A steady trickle of
+evictions on a mobile-heavy client base is a capacity signal, not an incident —
+which is exactly why these clients are counted and dropped rather than allowed
+to accumulate.

src/__tests__/fanout.test.ts

@@ -50,6 +50,15 @@ vi.mock('prom-client', () => {
   }
   return {
     register: { metrics: vi.fn().mockResolvedValue('') },
+    // src/metrics.ts builds its own Registry; src/ws/fanout.ts registers the
+    // websocket metrics on it so /metrics actually exports them.
+    Registry: class {
+      setDefaultLabels() {}
+      registerMetric() {}
+      metrics() { return Promise.resolve('') }
+      getSingleMetric() { return undefined }
+      resetMetrics() { return Promise.resolve('') }
+    },
     Gauge: MockMetric,
     Counter: MockMetric,
     Histogram: MockMetric,