Compare commits

..

6 Commits

Author SHA1 Message Date
jkunz 200e86e311 v27.7.1
Default (tags) / security (push) Failing after 0s
Default (tags) / test (push) Failing after 0s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-04-14 00:54:12 +00:00
jkunz a53a2c4ca5 fix(rustproxy-http,rustproxy-metrics): fix domain-scoped request host detection and harden connection metrics cleanup 2026-04-14 00:54:12 +00:00
jkunz 6ee7237357 v27.7.0
Default (tags) / security (push) Failing after 0s
Default (tags) / test (push) Failing after 0s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-04-13 23:21:54 +00:00
jkunz b5b4c608f0 feat(smart-proxy): add typed Rust config serialization and regex header contract coverage 2026-04-13 23:21:54 +00:00
jkunz af132f40fc v27.6.0
Default (tags) / security (push) Failing after 0s
Default (tags) / test (push) Failing after 0s
Default (tags) / release (push) Has been skipped
Default (tags) / metadata (push) Has been skipped
2026-04-13 18:33:28 +00:00
jkunz 781634446a feat(metrics): track per-IP domain request metrics across HTTP and TCP passthrough traffic 2026-04-13 18:33:28 +00:00
31 changed files with 3379 additions and 724 deletions
+23
View File
@@ -1,5 +1,28 @@
# Changelog
## 2026-04-14 - 27.7.1 - fix(rustproxy-http,rustproxy-metrics)
fix domain-scoped request host detection and harden connection metrics cleanup
- use a shared request host extractor that falls back to URI authority so domain-scoped IP allow lists work for HTTP/2 and HTTP/3 requests without a Host header
- add request filter and host extraction tests covering domain-scoped ACL behavior
- prevent connection counters from underflowing during close handling and clean up per-IP metrics entries more safely
- normalize tracked domain keys in metrics to reduce duplicate entries caused by case or trailing-dot variations
## 2026-04-13 - 27.7.0 - feat(smart-proxy)
add typed Rust config serialization and regex header contract coverage
- serialize SmartProxy routes and top-level options into explicit Rust-safe types, including header regex literals, UDP field normalization, ACME, defaults, and proxy settings
- support JS-style regex header literals with flags in Rust header matching and add cross-contract tests for route preprocessing and config deserialization
- improve TypeScript safety for Rust bridge and metrics integration by replacing loose any-based payloads with dedicated Rust type definitions
## 2026-04-13 - 27.6.0 - feat(metrics)
track per-IP domain request metrics across HTTP and TCP passthrough traffic
- records domain request counts per frontend IP from HTTP Host headers and TCP SNI
- exposes per-IP domain maps and top IP-domain request pairs through the TypeScript metrics adapter
- bounds per-IP domain tracking and prunes stale entries to limit memory growth
- adds metrics system documentation covering architecture, collected data, and known gaps
## 2026-04-06 - 27.5.0 - feat(security)
add domain-scoped IP allow list support across HTTP and passthrough filtering
+1 -1
View File
@@ -1,6 +1,6 @@
{
"name": "@push.rocks/smartproxy",
"version": "27.5.0",
"version": "27.7.1",
"private": false,
"description": "A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.",
"main": "dist_ts/index.js",
+484
View File
@@ -0,0 +1,484 @@
# SmartProxy Metrics System
## Architecture
Two-tier design separating the data plane from the observation plane:
**Hot path (per-chunk, lock-free):** All recording in the proxy data plane touches only `AtomicU64` counters. No `Mutex` is ever acquired on the forwarding path. `CountingBody` batches flushes every 64KB to reduce DashMap shard contention.
**Cold path (1Hz sampling):** A background tokio task drains pending atomics into `ThroughputTracker` circular buffers (Mutex-guarded), producing per-second throughput history. Same task prunes orphaned entries and cleans up rate limiter state.
**Read path (on-demand):** `snapshot()` reads all atomics and locks ThroughputTrackers to build a serializable `Metrics` struct. TypeScript polls this at 1s intervals via IPC.
```
Data Plane (lock-free) Background (1Hz) Read Path
───────────────────── ────────────────── ─────────
record_bytes() ──> AtomicU64 ──┐
record_http_request() ──> AtomicU64 ──┤
connection_opened/closed() ──> AtomicU64 ──┤ sample_all() snapshot()
backend_*() ──> DashMap<AtomicU64> ──┤────> drain atomics ──────> Metrics struct
protocol_*() ──> AtomicU64 ──┤ feed ThroughputTrackers ──> JSON
datagram_*() ──> AtomicU64 ──┘ prune orphans ──> IPC stdout
──> TS cache
──> IMetrics API
```
### Key Types
| Type | Crate | Purpose |
|---|---|---|
| `MetricsCollector` | `rustproxy-metrics` | Central store. All DashMaps, atomics, and throughput trackers |
| `ThroughputTracker` | `rustproxy-metrics` | Circular buffer of 1Hz samples. Default 3600 capacity (1 hour) |
| `ForwardMetricsCtx` | `rustproxy-passthrough` | Carries `Arc<MetricsCollector>` + route_id + source_ip through TCP forwarding |
| `CountingBody` | `rustproxy-http` | Wraps HTTP bodies, batches byte recording per 64KB, flushes on drop |
| `ProtocolGuard` | `rustproxy-http` | RAII guard for frontend/backend protocol active/total counters |
| `ConnectionGuard` | `rustproxy-passthrough` | RAII guard calling `connection_closed()` on drop |
| `RustMetricsAdapter` | TypeScript | Polls Rust via IPC, implements `IMetrics` interface over cached JSON |
---
## What's Collected
### Global Counters
| Metric | Type | Updated by |
|---|---|---|
| Active connections | AtomicU64 | `connection_opened/closed` |
| Total connections (lifetime) | AtomicU64 | `connection_opened` |
| Total bytes in | AtomicU64 | `record_bytes` |
| Total bytes out | AtomicU64 | `record_bytes` |
| Total HTTP requests | AtomicU64 | `record_http_request` |
| Active UDP sessions | AtomicU64 | `udp_session_opened/closed` |
| Total UDP sessions | AtomicU64 | `udp_session_opened` |
| Total datagrams in | AtomicU64 | `record_datagram_in` |
| Total datagrams out | AtomicU64 | `record_datagram_out` |
### Per-Route Metrics (keyed by route ID string)
| Metric | Storage |
|---|---|
| Active connections | `DashMap<String, AtomicU64>` |
| Total connections | `DashMap<String, AtomicU64>` |
| Bytes in / out | `DashMap<String, AtomicU64>` |
| Pending throughput (in, out) | `DashMap<String, (AtomicU64, AtomicU64)>` |
| Throughput history | `DashMap<String, Mutex<ThroughputTracker>>` |
Entries are pruned via `retain_routes()` when routes are removed.
### Per-IP Metrics (keyed by IP string)
| Metric | Storage |
|---|---|
| Active connections | `DashMap<String, AtomicU64>` |
| Total connections | `DashMap<String, AtomicU64>` |
| Bytes in / out | `DashMap<String, AtomicU64>` |
| Pending throughput (in, out) | `DashMap<String, (AtomicU64, AtomicU64)>` |
| Throughput history | `DashMap<String, Mutex<ThroughputTracker>>` |
| Domain requests | `DashMap<String, DashMap<String, AtomicU64>>` (IP → domain → count) |
All seven maps for an IP are evicted when its active connection count drops to 0. Safety-net pruning in `sample_all()` catches entries orphaned by races. Snapshots cap at 100 IPs, sorted by active connections descending.
**Domain request tracking:** Records which domains each frontend IP has requested. Populated from HTTP Host headers (for HTTP/1.1, HTTP/2, HTTP/3 requests) and from SNI (for TLS passthrough connections). Capped at 256 domains per IP (`MAX_DOMAINS_PER_IP`) to prevent subdomain-spray abuse. Inner DashMap uses 2 shards to minimise base memory per IP (~200 bytes). Common case (IP + domain both known) is two DashMap reads + one atomic increment with zero allocation.
### Per-Backend Metrics (keyed by "host:port")
| Metric | Storage |
|---|---|
| Active connections | `DashMap<String, AtomicU64>` |
| Total connections | `DashMap<String, AtomicU64>` |
| Detected protocol (h1/h2/h3) | `DashMap<String, String>` |
| Connect errors | `DashMap<String, AtomicU64>` |
| Handshake errors | `DashMap<String, AtomicU64>` |
| Request errors | `DashMap<String, AtomicU64>` |
| Total connect time (microseconds) | `DashMap<String, AtomicU64>` |
| Connect count | `DashMap<String, AtomicU64>` |
| Pool hits | `DashMap<String, AtomicU64>` |
| Pool misses | `DashMap<String, AtomicU64>` |
| H2 failures (fallback to H1) | `DashMap<String, AtomicU64>` |
All per-backend maps are evicted when active count reaches 0. Pruned via `retain_backends()`.
### Frontend Protocol Distribution
Tracked via `ProtocolGuard` RAII guards and `FrontendProtocolTracker`. Five protocol categories, each with active + total counters (AtomicU64):
| Protocol | Where detected |
|---|---|
| h1 | `FrontendProtocolTracker` on first HTTP/1.x request |
| h2 | `FrontendProtocolTracker` on first HTTP/2 request |
| h3 | `ProtocolGuard::frontend("h3")` in H3ProxyService |
| ws | `ProtocolGuard::frontend("ws")` on WebSocket upgrade |
| other | Fallback (TCP passthrough without HTTP) |
Uses `fetch_update` for saturating decrements to prevent underflow races.
### Backend Protocol Distribution
Same five categories (h1/h2/h3/ws/other), tracked via `ProtocolGuard::backend()` at connection establishment. Backend h2 failures (fallback to h1) are separately counted.
### Throughput History
`ThroughputTracker` is a circular buffer storing `ThroughputSample { timestamp_ms, bytes_in, bytes_out }` at 1Hz.
- Global tracker: 1 instance, default 3600 capacity
- Per-route trackers: 1 per active route
- Per-IP trackers: 1 per connected IP (evicted with the IP)
- HTTP request tracker: reuses ThroughputTracker with bytes_in = request count, bytes_out = 0
Query methods:
- `instant()` — last 1 second average
- `recent()` — last 10 seconds average
- `throughput(N)` — last N seconds average
- `history(N)` — last N raw samples in chronological order
Snapshots return 60 samples of global throughput history.
### Protocol Detection Cache
Not part of MetricsCollector. Maintained by `HttpProxyService`'s protocol detection system. Injected into the metrics snapshot at read time by `get_metrics()`.
Each entry records: host, port, domain, detected protocol (h1/h2/h3), H3 port, age, last accessed, last probed, suppression flags, cooldown timers, consecutive failure counts.
---
## Instrumentation Points
### TCP Passthrough (`rustproxy-passthrough`)
**Connection lifecycle**`tcp_listener.rs`:
- Accept: `conn_tracker.connection_opened(&ip)` (rate limiter) + `ConnectionTrackerGuard` RAII
- Route match: `metrics.connection_opened(route_id, source_ip)` + `ConnectionGuard` RAII
- Close: Both guards call their respective `_closed()` methods on drop
**Byte recording**`forwarder.rs` (`forward_bidirectional_with_timeouts`):
- Initial peeked data recorded immediately
- Per-chunk in both directions: `record_bytes(n, 0, ...)` / `record_bytes(0, n, ...)`
- Same pattern in `forward_bidirectional_split_with_timeouts` (tcp_listener.rs) for TLS-terminated paths
### HTTP Proxy (`rustproxy-http`)
**Request counting**`proxy_service.rs`:
- `record_http_request()` called once per request after route matching succeeds
**Body byte counting**`counting_body.rs` wrapping:
- Request bodies: `CountingBody::new(body, ..., Direction::In)` — counts client-to-upstream bytes
- Response bodies: `CountingBody::new(body, ..., Direction::Out)` — counts upstream-to-client bytes
- Batched flush every 64KB (`BYTE_FLUSH_THRESHOLD = 65_536`), remainder flushed on drop
- Also updates `connection_activity` atomic (idle watchdog) and `active_requests` counter (streaming detection)
**Backend metrics**`proxy_service.rs`:
- `backend_connection_opened(key, connect_time)` — after TCP/TLS connect succeeds
- `backend_connection_closed(key)` — on teardown
- `backend_connect_error(key)` — TCP/TLS connect failure or timeout
- `backend_handshake_error(key)` — H1/H2 protocol handshake failure
- `backend_request_error(key)` — send_request failure
- `backend_h2_failure(key)` — H2 attempted, fell back to H1
- `backend_pool_hit(key)` / `backend_pool_miss(key)` — connection pool reuse
- `set_backend_protocol(key, proto)` — records detected protocol
**WebSocket**`proxy_service.rs`:
- Does NOT use CountingBody; records bytes directly per-chunk in both directions of the bidirectional copy loop
### QUIC (`rustproxy-passthrough`)
**Connection level**`quic_handler.rs`:
- `connection_opened` / `connection_closed` via `QuicConnGuard` RAII
- `conn_tracker.connection_opened/closed` for rate limiting
**Stream level**:
- For QUIC-to-TCP stream forwarding: `record_bytes(bytes_in, bytes_out, ...)` called once per stream at completion (post-hoc, not per-chunk)
- For HTTP/3: delegates to `HttpProxyService.handle_request()`, so all HTTP proxy metrics apply
**H3 specifics**`h3_service.rs`:
- `ProtocolGuard::frontend("h3")` tracks the H3 connection
- H3 request bodies: `record_bytes(data.len(), 0, ...)` called directly (not CountingBody) since H3 uses `stream.send_data()`
- H3 response bodies: wrapped in CountingBody like HTTP/1 and HTTP/2
### UDP (`rustproxy-passthrough`)
**Session lifecycle**`udp_listener.rs` / `udp_session.rs`:
- `udp_session_opened()` + `connection_opened(route_id, source_ip)` on new session
- `udp_session_closed()` + `connection_closed(route_id, source_ip)` on idle reap or port drain
**Datagram counting**`udp_listener.rs`:
- Inbound: `record_bytes(len, 0, ...)` + `record_datagram_in()`
- Outbound (backend reply): `record_bytes(0, len, ...)` + `record_datagram_out()`
---
## Sampling Loop
`lib.rs` spawns a tokio task at configurable interval (default 1000ms):
```rust
loop {
tokio::select! {
_ = cancel => break,
_ = interval.tick() => {
metrics.sample_all();
conn_tracker.cleanup_stale_timestamps();
http_proxy.cleanup_all_rate_limiters();
}
}
}
```
`sample_all()` performs in one pass:
1. Drains `global_pending_tp_in/out` into global ThroughputTracker, samples
2. Drains per-route pending counters into per-route trackers, samples each
3. Samples idle route trackers (no new data) to advance their window
4. Drains per-IP pending counters into per-IP trackers, samples each
5. Drains `pending_http_requests` into HTTP request throughput tracker
6. Prunes orphaned per-IP entries (bytes/throughput maps with no matching ip_connections key)
7. Prunes orphaned per-backend entries (error/stats maps with no matching active/total key)
---
## Data Flow: Rust to TypeScript
```
MetricsCollector.snapshot()
├── reads all AtomicU64 counters
├── iterates DashMaps (routes, IPs, backends)
├── locks ThroughputTrackers for instant/recent rates + history
└── produces Metrics struct
RustProxy::get_metrics()
├── calls snapshot()
├── enriches with detectedProtocols from HTTP proxy protocol cache
└── returns Metrics
management.rs "getMetrics" IPC command
├── calls get_metrics()
├── serde_json::to_value (camelCase)
└── writes JSON to stdout
RustProxyBridge (TypeScript)
├── reads JSON from Rust process stdout
└── returns parsed object
RustMetricsAdapter
├── setInterval polls bridge.getMetrics() every 1s
├── stores raw JSON in this.cache
└── IMetrics methods read synchronously from cache
SmartProxy.getMetrics()
└── returns the RustMetricsAdapter instance
```
### IPC JSON Shape (Metrics)
```json
{
"activeConnections": 42,
"totalConnections": 1000,
"bytesIn": 123456789,
"bytesOut": 987654321,
"throughputInBytesPerSec": 50000,
"throughputOutBytesPerSec": 80000,
"throughputRecentInBytesPerSec": 45000,
"throughputRecentOutBytesPerSec": 75000,
"routes": {
"<route-id>": {
"activeConnections": 5,
"totalConnections": 100,
"bytesIn": 0, "bytesOut": 0,
"throughputInBytesPerSec": 0, "throughputOutBytesPerSec": 0,
"throughputRecentInBytesPerSec": 0, "throughputRecentOutBytesPerSec": 0
}
},
"ips": {
"<ip>": {
"activeConnections": 2, "totalConnections": 10,
"bytesIn": 0, "bytesOut": 0,
"throughputInBytesPerSec": 0, "throughputOutBytesPerSec": 0,
"domainRequests": {
"example.com": 4821,
"api.example.com": 312
}
}
},
"backends": {
"<host:port>": {
"activeConnections": 3, "totalConnections": 50,
"protocol": "h2",
"connectErrors": 0, "handshakeErrors": 0, "requestErrors": 0,
"totalConnectTimeUs": 150000, "connectCount": 50,
"poolHits": 40, "poolMisses": 10, "h2Failures": 1
}
},
"throughputHistory": [
{ "timestampMs": 1713000000000, "bytesIn": 50000, "bytesOut": 80000 }
],
"totalHttpRequests": 5000,
"httpRequestsPerSec": 100,
"httpRequestsPerSecRecent": 95,
"activeUdpSessions": 0, "totalUdpSessions": 5,
"totalDatagramsIn": 1000, "totalDatagramsOut": 1000,
"frontendProtocols": {
"h1Active": 10, "h1Total": 500,
"h2Active": 5, "h2Total": 200,
"h3Active": 1, "h3Total": 50,
"wsActive": 2, "wsTotal": 30,
"otherActive": 0, "otherTotal": 0
},
"backendProtocols": { "...same shape..." },
"detectedProtocols": [
{
"host": "backend", "port": 443, "domain": "example.com",
"protocol": "h2", "h3Port": 443,
"ageSecs": 120, "lastAccessedSecs": 5, "lastProbedSecs": 120,
"h2Suppressed": false, "h3Suppressed": false,
"h2CooldownRemainingSecs": null, "h3CooldownRemainingSecs": null,
"h2ConsecutiveFailures": null, "h3ConsecutiveFailures": null
}
]
}
```
### IPC JSON Shape (Statistics)
Lightweight administrative summary, fetched on-demand (not polled):
```json
{
"activeConnections": 42,
"totalConnections": 1000,
"routesCount": 5,
"listeningPorts": [80, 443, 8443],
"uptimeSeconds": 86400
}
```
---
## TypeScript Consumer API
`SmartProxy.getMetrics()` returns an `IMetrics` object. All members are synchronous methods reading from the polled cache.
### connections
| Method | Return | Source |
|---|---|---|
| `active()` | `number` | `cache.activeConnections` |
| `total()` | `number` | `cache.totalConnections` |
| `byRoute()` | `Map<string, number>` | `cache.routes[name].activeConnections` |
| `byIP()` | `Map<string, number>` | `cache.ips[ip].activeConnections` |
| `topIPs(limit?)` | `Array<{ip, count}>` | `cache.ips` sorted by active desc, default 10 |
| `domainRequestsByIP()` | `Map<string, Map<string, number>>` | `cache.ips[ip].domainRequests` |
| `topDomainRequests(limit?)` | `Array<{ip, domain, count}>` | Flattened from all IPs, sorted by count desc, default 20 |
| `frontendProtocols()` | `IProtocolDistribution` | `cache.frontendProtocols.*` |
| `backendProtocols()` | `IProtocolDistribution` | `cache.backendProtocols.*` |
### throughput
| Method | Return | Source |
|---|---|---|
| `instant()` | `{in, out}` | `cache.throughputInBytesPerSec/Out` |
| `recent()` | `{in, out}` | `cache.throughputRecentInBytesPerSec/Out` |
| `average()` | `{in, out}` | Falls back to `instant()` (not wired to windowed average) |
| `custom(seconds)` | `{in, out}` | Falls back to `instant()` (not wired) |
| `history(seconds)` | `IThroughputHistoryPoint[]` | `cache.throughputHistory` sliced to last N entries |
| `byRoute(windowSeconds?)` | `Map<string, {in, out}>` | `cache.routes[name].throughputIn/OutBytesPerSec` |
| `byIP(windowSeconds?)` | `Map<string, {in, out}>` | `cache.ips[ip].throughputIn/OutBytesPerSec` |
### requests
| Method | Return | Source |
|---|---|---|
| `perSecond()` | `number` | `cache.httpRequestsPerSec` |
| `perMinute()` | `number` | `cache.httpRequestsPerSecRecent * 60` |
| `total()` | `number` | `cache.totalHttpRequests` (fallback: totalConnections) |
### totals
| Method | Return | Source |
|---|---|---|
| `bytesIn()` | `number` | `cache.bytesIn` |
| `bytesOut()` | `number` | `cache.bytesOut` |
| `connections()` | `number` | `cache.totalConnections` |
### backends
| Method | Return | Source |
|---|---|---|
| `byBackend()` | `Map<string, IBackendMetrics>` | `cache.backends[key].*` with computed `avgConnectTimeMs` and `poolHitRate` |
| `protocols()` | `Map<string, string>` | `cache.backends[key].protocol` |
| `topByErrors(limit?)` | `IBackendMetrics[]` | Sorted by total errors desc |
| `detectedProtocols()` | `IProtocolCacheEntry[]` | `cache.detectedProtocols` passthrough |
`IBackendMetrics`: `{ protocol, activeConnections, totalConnections, connectErrors, handshakeErrors, requestErrors, avgConnectTimeMs, poolHitRate, h2Failures }`
### udp
| Method | Return | Source |
|---|---|---|
| `activeSessions()` | `number` | `cache.activeUdpSessions` |
| `totalSessions()` | `number` | `cache.totalUdpSessions` |
| `datagramsIn()` | `number` | `cache.totalDatagramsIn` |
| `datagramsOut()` | `number` | `cache.totalDatagramsOut` |
### percentiles (stub)
`connectionDuration()` and `bytesTransferred()` always return zeros. Not implemented.
---
## Configuration
```typescript
interface IMetricsConfig {
enabled: boolean; // default true
sampleIntervalMs: number; // default 1000 (1Hz sampling + TS polling)
retentionSeconds: number; // default 3600 (ThroughputTracker capacity)
enableDetailedTracking: boolean;
enablePercentiles: boolean;
cacheResultsMs: number;
prometheusEnabled: boolean; // not wired
prometheusPath: string; // not wired
prometheusPrefix: string; // not wired
}
```
Rust-side config (`MetricsConfig` in `rustproxy-config`):
```rust
pub struct MetricsConfig {
pub enabled: Option<bool>,
pub sample_interval_ms: Option<u64>, // default 1000
pub retention_seconds: Option<u64>, // default 3600
}
```
---
## Design Decisions
**Lock-free hot path.** `record_bytes()` is the most frequently called method (per-chunk in TCP, per-64KB in HTTP). It only touches `AtomicU64` with `Relaxed` ordering and short-circuits zero-byte directions to skip DashMap lookups entirely.
**CountingBody batching.** HTTP body frames are typically 16KB. Flushing to MetricsCollector every 64KB reduces DashMap shard contention by ~4x compared to per-frame recording.
**RAII guards everywhere.** `ConnectionGuard`, `ConnectionTrackerGuard`, `QuicConnGuard`, `ProtocolGuard`, `FrontendProtocolTracker` all use Drop to guarantee counter cleanup on all exit paths including panics.
**Saturating decrements.** Protocol counters use `fetch_update` loops instead of `fetch_sub` to prevent underflow to `u64::MAX` from concurrent close races.
**Bounded memory.** Per-IP entries evicted on last connection close. Per-backend entries evicted on last connection close. Snapshot caps IPs and backends at 100 each. `sample_all()` prunes orphaned entries every second.
**Two-phase throughput.** Pending bytes accumulate in lock-free atomics. The 1Hz cold path drains them into Mutex-guarded ThroughputTrackers. This means the hot path never contends on a Mutex, while the cold path does minimal work (one drain + one sample per tracker).
---
## Known Gaps
| Gap | Status |
|---|---|
| `throughput.average()` / `throughput.custom(seconds)` | Fall back to `instant()`. Not wired to Rust windowed queries. |
| `percentiles.connectionDuration()` / `percentiles.bytesTransferred()` | Stub returning zeros. No histogram in Rust. |
| Prometheus export | Config fields exist but not wired to any exporter. |
| `LogDeduplicator` | Implemented in `rustproxy-metrics` but not connected to any call site. |
| Rate limit hit counters | Rate-limited requests return 429 but no counter is recorded in MetricsCollector. |
| QUIC stream byte counting | Post-hoc (per-stream totals after close), not per-chunk like TCP. |
| Throughput history in snapshot | Capped at 60 samples. TS `history(seconds)` cannot return more than 60 points regardless of `retentionSeconds`. |
| Per-route total connections / bytes | Available in Rust JSON but `IMetrics.connections.byRoute()` only exposes active connections. |
| Per-IP total connections / bytes | Available in Rust JSON but `IMetrics.connections.byIP()` only exposes active connections. |
| IPC response typing | `RustProxyBridge` declares `result: any` for both metrics commands. No type-safe response. |
+246 -23
View File
@@ -129,7 +129,6 @@ pub struct RustProxyOptions {
pub defaults: Option<DefaultConfig>,
// ─── Timeout Settings ────────────────────────────────────────────
/// Timeout for establishing connection to backend (ms), default: 30000
#[serde(skip_serializing_if = "Option::is_none")]
pub connection_timeout: Option<u64>,
@@ -159,7 +158,6 @@ pub struct RustProxyOptions {
pub graceful_shutdown_timeout: Option<u64>,
// ─── Socket Optimization ─────────────────────────────────────────
/// Disable Nagle's algorithm (default: true)
#[serde(skip_serializing_if = "Option::is_none")]
pub no_delay: Option<bool>,
@@ -177,7 +175,6 @@ pub struct RustProxyOptions {
pub max_pending_data_size: Option<u64>,
// ─── Enhanced Features ───────────────────────────────────────────
/// Disable inactivity checking entirely
#[serde(skip_serializing_if = "Option::is_none")]
pub disable_inactivity_check: Option<bool>,
@@ -199,7 +196,6 @@ pub struct RustProxyOptions {
pub enable_randomized_timeouts: Option<bool>,
// ─── Rate Limiting ───────────────────────────────────────────────
/// Maximum simultaneous connections from a single IP
#[serde(skip_serializing_if = "Option::is_none")]
pub max_connections_per_ip: Option<u64>,
@@ -213,7 +209,6 @@ pub struct RustProxyOptions {
pub max_connections: Option<u64>,
// ─── Keep-Alive Settings ─────────────────────────────────────────
/// How to treat keep-alive connections
#[serde(skip_serializing_if = "Option::is_none")]
pub keep_alive_treatment: Option<KeepAliveTreatment>,
@@ -227,7 +222,6 @@ pub struct RustProxyOptions {
pub extended_keep_alive_lifetime: Option<u64>,
// ─── HttpProxy Integration ───────────────────────────────────────
/// Array of ports to forward to HttpProxy
#[serde(skip_serializing_if = "Option::is_none")]
pub use_http_proxy: Option<Vec<u16>>,
@@ -237,13 +231,11 @@ pub struct RustProxyOptions {
pub http_proxy_port: Option<u16>,
// ─── Metrics ─────────────────────────────────────────────────────
/// Metrics configuration
#[serde(skip_serializing_if = "Option::is_none")]
pub metrics: Option<MetricsConfig>,
// ─── ACME ────────────────────────────────────────────────────────
/// Global ACME configuration
#[serde(skip_serializing_if = "Option::is_none")]
pub acme: Option<AcmeOptions>,
@@ -318,7 +310,8 @@ impl RustProxyOptions {
/// Get all unique ports that routes listen on.
pub fn all_listening_ports(&self) -> Vec<u16> {
let mut ports: Vec<u16> = self.routes
let mut ports: Vec<u16> = self
.routes
.iter()
.flat_map(|r| r.listening_ports())
.collect();
@@ -340,7 +333,12 @@ mod tests {
route_match: RouteMatch {
ports: PortRange::Single(listen_port),
domains: Some(DomainSpec::Single(domain.to_string())),
path: None, client_ip: None, transport: None, tls_version: None, headers: None, protocol: None,
path: None,
client_ip: None,
transport: None,
tls_version: None,
headers: None,
protocol: None,
},
action: RouteAction {
action_type: RouteActionType::Forward,
@@ -348,14 +346,30 @@ mod tests {
target_match: None,
host: HostSpec::Single(host.to_string()),
port: PortSpec::Fixed(port),
tls: None, websocket: None, load_balancing: None, send_proxy_protocol: None,
headers: None, advanced: None, backend_transport: None, priority: None,
tls: None,
websocket: None,
load_balancing: None,
send_proxy_protocol: None,
headers: None,
advanced: None,
backend_transport: None,
priority: None,
}]),
tls: None, websocket: None, load_balancing: None, advanced: None,
options: None, send_proxy_protocol: None, udp: None,
tls: None,
websocket: None,
load_balancing: None,
advanced: None,
options: None,
send_proxy_protocol: None,
udp: None,
},
headers: None, security: None, name: None, description: None,
priority: None, tags: None, enabled: None,
headers: None,
security: None,
name: None,
description: None,
priority: None,
tags: None,
enabled: None,
}
}
@@ -363,8 +377,12 @@ mod tests {
let mut route = make_route(domain, host, port, 443);
route.action.tls = Some(RouteTls {
mode: TlsMode::Passthrough,
certificate: None, acme: None, versions: None, ciphers: None,
honor_cipher_order: None, session_timeout: None,
certificate: None,
acme: None,
versions: None,
ciphers: None,
honor_cipher_order: None,
session_timeout: None,
});
route
}
@@ -410,6 +428,209 @@ mod tests {
assert_eq!(parsed.connection_timeout, Some(5000));
}
#[test]
fn test_deserialize_ts_contract_route_shapes() {
let value = serde_json::json!({
"routes": [{
"name": "contract-route",
"match": {
"ports": [443, { "from": 8443, "to": 8444 }],
"domains": ["api.example.com", "*.example.com"],
"transport": "udp",
"protocol": "http3",
"headers": {
"content-type": "/^application\\/json$/i"
}
},
"action": {
"type": "forward",
"targets": [{
"match": {
"ports": [443],
"path": "/api/*",
"method": ["GET"],
"headers": {
"x-env": "/^(prod|stage)$/"
}
},
"host": ["backend-a", "backend-b"],
"port": "preserve",
"sendProxyProtocol": true,
"backendTransport": "tcp"
}],
"tls": {
"mode": "terminate",
"certificate": "auto"
},
"sendProxyProtocol": true,
"udp": {
"maxSessionsPerIp": 321,
"quic": {
"enableHttp3": true
}
}
},
"security": {
"ipAllowList": [{
"ip": "10.0.0.0/8",
"domains": ["api.example.com"]
}]
}
}],
"preserveSourceIp": true,
"proxyIps": ["10.0.0.1"],
"acceptProxyProtocol": true,
"sendProxyProtocol": true,
"noDelay": true,
"keepAlive": true,
"keepAliveInitialDelay": 1500,
"maxPendingDataSize": 4096,
"disableInactivityCheck": true,
"enableKeepAliveProbes": true,
"enableDetailedLogging": true,
"enableTlsDebugLogging": true,
"enableRandomizedTimeouts": true,
"connectionTimeout": 5000,
"initialDataTimeout": 7000,
"socketTimeout": 9000,
"inactivityCheckInterval": 1100,
"maxConnectionLifetime": 13000,
"inactivityTimeout": 15000,
"gracefulShutdownTimeout": 17000,
"maxConnectionsPerIp": 20,
"connectionRateLimitPerMinute": 30,
"keepAliveTreatment": "extended",
"keepAliveInactivityMultiplier": 2.0,
"extendedKeepAliveLifetime": 19000,
"metrics": {
"enabled": true,
"sampleIntervalMs": 250,
"retentionSeconds": 60
},
"acme": {
"enabled": true,
"email": "ops@example.com",
"environment": "staging",
"useProduction": false,
"skipConfiguredCerts": true,
"renewThresholdDays": 14,
"renewCheckIntervalHours": 12,
"autoRenew": true,
"port": 80
}
});
let options: RustProxyOptions = serde_json::from_value(value).unwrap();
assert_eq!(options.routes.len(), 1);
assert_eq!(options.preserve_source_ip, Some(true));
assert_eq!(options.proxy_ips, Some(vec!["10.0.0.1".to_string()]));
assert_eq!(options.accept_proxy_protocol, Some(true));
assert_eq!(options.send_proxy_protocol, Some(true));
assert_eq!(options.no_delay, Some(true));
assert_eq!(options.keep_alive, Some(true));
assert_eq!(options.keep_alive_initial_delay, Some(1500));
assert_eq!(options.max_pending_data_size, Some(4096));
assert_eq!(options.disable_inactivity_check, Some(true));
assert_eq!(options.enable_keep_alive_probes, Some(true));
assert_eq!(options.enable_detailed_logging, Some(true));
assert_eq!(options.enable_tls_debug_logging, Some(true));
assert_eq!(options.enable_randomized_timeouts, Some(true));
assert_eq!(options.connection_timeout, Some(5000));
assert_eq!(options.initial_data_timeout, Some(7000));
assert_eq!(options.socket_timeout, Some(9000));
assert_eq!(options.inactivity_check_interval, Some(1100));
assert_eq!(options.max_connection_lifetime, Some(13000));
assert_eq!(options.inactivity_timeout, Some(15000));
assert_eq!(options.graceful_shutdown_timeout, Some(17000));
assert_eq!(options.max_connections_per_ip, Some(20));
assert_eq!(options.connection_rate_limit_per_minute, Some(30));
assert_eq!(
options.keep_alive_treatment,
Some(KeepAliveTreatment::Extended)
);
assert_eq!(options.keep_alive_inactivity_multiplier, Some(2.0));
assert_eq!(options.extended_keep_alive_lifetime, Some(19000));
let route = &options.routes[0];
assert_eq!(route.route_match.transport, Some(TransportProtocol::Udp));
assert_eq!(route.route_match.protocol.as_deref(), Some("http3"));
assert_eq!(
route
.route_match
.headers
.as_ref()
.unwrap()
.get("content-type")
.unwrap(),
"/^application\\/json$/i"
);
let target = &route.action.targets.as_ref().unwrap()[0];
assert!(matches!(target.host, HostSpec::List(_)));
assert!(matches!(target.port, PortSpec::Special(ref p) if p == "preserve"));
assert_eq!(target.backend_transport, Some(TransportProtocol::Tcp));
assert_eq!(target.send_proxy_protocol, Some(true));
assert_eq!(
target
.target_match
.as_ref()
.unwrap()
.headers
.as_ref()
.unwrap()
.get("x-env")
.unwrap(),
"/^(prod|stage)$/"
);
assert_eq!(route.action.send_proxy_protocol, Some(true));
assert_eq!(
route.action.udp.as_ref().unwrap().max_sessions_per_ip,
Some(321)
);
assert_eq!(
route
.action
.udp
.as_ref()
.unwrap()
.quic
.as_ref()
.unwrap()
.enable_http3,
Some(true)
);
let allow_list = route
.security
.as_ref()
.unwrap()
.ip_allow_list
.as_ref()
.unwrap();
assert!(matches!(
&allow_list[0],
crate::security_types::IpAllowEntry::DomainScoped { ip, domains }
if ip == "10.0.0.0/8" && domains == &vec!["api.example.com".to_string()]
));
let metrics = options.metrics.as_ref().unwrap();
assert_eq!(metrics.enabled, Some(true));
assert_eq!(metrics.sample_interval_ms, Some(250));
assert_eq!(metrics.retention_seconds, Some(60));
let acme = options.acme.as_ref().unwrap();
assert_eq!(acme.enabled, Some(true));
assert_eq!(acme.email.as_deref(), Some("ops@example.com"));
assert_eq!(acme.environment, Some(AcmeEnvironment::Staging));
assert_eq!(acme.use_production, Some(false));
assert_eq!(acme.skip_configured_certs, Some(true));
assert_eq!(acme.renew_threshold_days, Some(14));
assert_eq!(acme.renew_check_interval_hours, Some(12));
assert_eq!(acme.auto_renew, Some(true));
assert_eq!(acme.port, Some(80));
}
#[test]
fn test_default_timeouts() {
let options = RustProxyOptions::default();
@@ -438,9 +659,9 @@ mod tests {
fn test_all_listening_ports() {
let options = RustProxyOptions {
routes: vec![
make_route("a.com", "backend", 8080, 80), // port 80
make_route("a.com", "backend", 8080, 80), // port 80
make_passthrough_route("b.com", "backend", 443), // port 443
make_route("c.com", "backend", 9090, 80), // port 80 (duplicate)
make_route("c.com", "backend", 9090, 80), // port 80 (duplicate)
],
..Default::default()
};
@@ -464,9 +685,11 @@ mod tests {
#[test]
fn test_deserialize_example_json() {
let content = std::fs::read_to_string(
concat!(env!("CARGO_MANIFEST_DIR"), "/../../config/example.json")
).unwrap();
let content = std::fs::read_to_string(concat!(
env!("CARGO_MANIFEST_DIR"),
"/../../config/example.json"
))
.unwrap();
let options: RustProxyOptions = serde_json::from_str(&content).unwrap();
assert_eq!(options.routes.len(), 4);
let ports = options.all_listening_ports();
@@ -1,8 +1,8 @@
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use crate::tls_types::RouteTls;
use crate::security_types::RouteSecurity;
use crate::tls_types::RouteTls;
// ─── Port Range ──────────────────────────────────────────────────────
@@ -32,12 +32,13 @@ impl PortRange {
pub fn to_ports(&self) -> Vec<u16> {
match self {
PortRange::Single(p) => vec![*p],
PortRange::List(items) => {
items.iter().flat_map(|item| match item {
PortRange::List(items) => items
.iter()
.flat_map(|item| match item {
PortRangeItem::Port(p) => vec![*p],
PortRangeItem::Range(r) => (r.from..=r.to).collect(),
}).collect()
}
})
.collect(),
}
}
}
@@ -105,7 +106,8 @@ impl From<Vec<&str>> for DomainSpec {
}
/// Header match value: either exact string or regex pattern.
/// In JSON, all values come as strings. Regex patterns are prefixed with `/` and suffixed with `/`.
/// In JSON, all values come as strings. Regex patterns use JS-style literal syntax,
/// e.g. `/^application\/json$/` or `/^application\/json$/i`.
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(untagged)]
pub enum HeaderMatchValue {
@@ -3,8 +3,8 @@
//! Reuses idle keep-alive connections to avoid per-request TCP+TLS handshakes.
//! HTTP/2 and HTTP/3 connections are multiplexed (clone the sender / share the connection).
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
use bytes::Bytes;
@@ -105,13 +105,19 @@ impl ConnectionPool {
/// Try to check out an idle HTTP/1.1 sender for the given key.
/// Returns `None` if no usable idle connection exists.
pub fn checkout_h1(&self, key: &PoolKey) -> Option<http1::SendRequest<BoxBody<Bytes, hyper::Error>>> {
pub fn checkout_h1(
&self,
key: &PoolKey,
) -> Option<http1::SendRequest<BoxBody<Bytes, hyper::Error>>> {
let mut entry = self.h1_pool.get_mut(key)?;
let idles = entry.value_mut();
while let Some(idle) = idles.pop() {
// Check if the connection is still alive and ready
if idle.idle_since.elapsed() < IDLE_TIMEOUT && idle.sender.is_ready() && !idle.sender.is_closed() {
if idle.idle_since.elapsed() < IDLE_TIMEOUT
&& idle.sender.is_ready()
&& !idle.sender.is_closed()
{
// H1 pool hit — no logging on hot path
return Some(idle.sender);
}
@@ -128,7 +134,11 @@ impl ConnectionPool {
/// Return an HTTP/1.1 sender to the pool after the response body has been prepared.
/// The caller should NOT call this if the sender is closed or not ready.
pub fn checkin_h1(&self, key: PoolKey, sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>) {
pub fn checkin_h1(
&self,
key: PoolKey,
sender: http1::SendRequest<BoxBody<Bytes, hyper::Error>>,
) {
if sender.is_closed() || !sender.is_ready() {
return; // Don't pool broken connections
}
@@ -145,7 +155,10 @@ impl ConnectionPool {
/// Try to get a cloned HTTP/2 sender for the given key.
/// HTTP/2 senders are Clone-able (multiplexed), so we clone rather than remove.
pub fn checkout_h2(&self, key: &PoolKey) -> Option<(http2::SendRequest<BoxBody<Bytes, hyper::Error>>, Duration)> {
pub fn checkout_h2(
&self,
key: &PoolKey,
) -> Option<(http2::SendRequest<BoxBody<Bytes, hyper::Error>>, Duration)> {
let entry = self.h2_pool.get(key)?;
let pooled = entry.value();
let age = pooled.created_at.elapsed();
@@ -184,16 +197,23 @@ impl ConnectionPool {
/// Register an HTTP/2 sender in the pool. Returns the generation ID for this entry.
/// The caller should pass this generation to the connection driver so it can use
/// `remove_h2_if_generation` instead of `remove_h2` to avoid phantom eviction.
pub fn register_h2(&self, key: PoolKey, sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>) -> u64 {
pub fn register_h2(
&self,
key: PoolKey,
sender: http2::SendRequest<BoxBody<Bytes, hyper::Error>>,
) -> u64 {
let gen = self.h2_generation.fetch_add(1, Ordering::Relaxed);
if sender.is_closed() {
return gen;
}
self.h2_pool.insert(key, PooledH2 {
sender,
created_at: Instant::now(),
generation: gen,
});
self.h2_pool.insert(
key,
PooledH2 {
sender,
created_at: Instant::now(),
generation: gen,
},
);
gen
}
@@ -204,7 +224,11 @@ impl ConnectionPool {
pub fn checkout_h3(
&self,
key: &PoolKey,
) -> Option<(h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>, quinn::Connection, Duration)> {
) -> Option<(
h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>,
quinn::Connection,
Duration,
)> {
let entry = self.h3_pool.get(key)?;
let pooled = entry.value();
let age = pooled.created_at.elapsed();
@@ -234,12 +258,15 @@ impl ConnectionPool {
send_request: h3::client::SendRequest<h3_quinn::OpenStreams, Bytes>,
) -> u64 {
let gen = self.h2_generation.fetch_add(1, Ordering::Relaxed);
self.h3_pool.insert(key, PooledH3 {
send_request,
connection,
created_at: Instant::now(),
generation: gen,
});
self.h3_pool.insert(
key,
PooledH3 {
send_request,
connection,
created_at: Instant::now(),
generation: gen,
},
);
gen
}
@@ -280,7 +307,9 @@ impl ConnectionPool {
// Evict dead or aged-out H2 connections
let mut dead_h2 = Vec::new();
for entry in h2_pool.iter() {
if entry.value().sender.is_closed() || entry.value().created_at.elapsed() >= MAX_H2_AGE {
if entry.value().sender.is_closed()
|| entry.value().created_at.elapsed() >= MAX_H2_AGE
{
dead_h2.push(entry.key().clone());
}
}
@@ -1,8 +1,8 @@
//! A body wrapper that counts bytes flowing through and reports them to MetricsCollector.
use std::pin::Pin;
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::task::{Context, Poll};
use bytes::Bytes;
@@ -76,7 +76,11 @@ impl<B> CountingBody<B> {
/// Set the connection-level activity tracker. When set, each data frame
/// updates this timestamp to prevent the idle watchdog from killing the
/// connection during active body streaming.
pub fn with_connection_activity(mut self, activity: Arc<AtomicU64>, start: std::time::Instant) -> Self {
pub fn with_connection_activity(
mut self,
activity: Arc<AtomicU64>,
start: std::time::Instant,
) -> Self {
self.connection_activity = Some(activity);
self.activity_start = Some(start);
self
@@ -134,7 +138,9 @@ where
}
// Keep the connection-level idle watchdog alive on every frame
// (this is just one atomic store — cheap enough per-frame)
if let (Some(activity), Some(start)) = (&this.connection_activity, &this.activity_start) {
if let (Some(activity), Some(start)) =
(&this.connection_activity, &this.activity_start)
{
activity.store(start.elapsed().as_millis() as u64, Ordering::Relaxed);
}
}
+28 -10
View File
@@ -11,8 +11,8 @@ use std::task::{Context, Poll};
use bytes::{Buf, Bytes};
use http_body::Frame;
use http_body_util::BodyExt;
use http_body_util::combinators::BoxBody;
use http_body_util::BodyExt;
use tracing::{debug, warn};
use rustproxy_config::RouteConfig;
@@ -49,7 +49,8 @@ impl H3ProxyService {
debug!("HTTP/3 connection from {} on port {}", remote_addr, port);
// Track frontend H3 connection for the QUIC connection's lifetime.
let _frontend_h3_guard = ProtocolGuard::frontend(Arc::clone(self.http_proxy.metrics()), "h3");
let _frontend_h3_guard =
ProtocolGuard::frontend(Arc::clone(self.http_proxy.metrics()), "h3");
let mut h3_conn: h3::server::Connection<h3_quinn::Connection, Bytes> =
h3::server::builder()
@@ -92,8 +93,15 @@ impl H3ProxyService {
tokio::spawn(async move {
if let Err(e) = handle_h3_request(
request, stream, port, remote_addr, &http_proxy, request_cancel,
).await {
request,
stream,
port,
remote_addr,
&http_proxy,
request_cancel,
)
.await
{
debug!("HTTP/3 request error from {}: {}", remote_addr, e);
}
});
@@ -153,11 +161,14 @@ async fn handle_h3_request(
// Delegate to HttpProxyService — same backend path as TCP/HTTP:
// route matching, ALPN protocol detection, connection pool, H1/H2/H3 auto.
let conn_activity = ConnActivity::new_standalone();
let response = http_proxy.handle_request(req, peer_addr, port, cancel, conn_activity).await
let response = http_proxy
.handle_request(req, peer_addr, port, cancel, conn_activity)
.await
.map_err(|e| anyhow::anyhow!("Backend request failed: {}", e))?;
// Await the body reader to get the H3 stream back
let mut stream = body_reader.await
let mut stream = body_reader
.await
.map_err(|e| anyhow::anyhow!("Body reader task failed: {}", e))?;
// Send response headers over H3 (skip hop-by-hop headers)
@@ -170,10 +181,13 @@ async fn handle_h3_request(
}
h3_response = h3_response.header(name, value);
}
let h3_response = h3_response.body(())
let h3_response = h3_response
.body(())
.map_err(|e| anyhow::anyhow!("Failed to build H3 response: {}", e))?;
stream.send_response(h3_response).await
stream
.send_response(h3_response)
.await
.map_err(|e| anyhow::anyhow!("Failed to send H3 response: {}", e))?;
// Stream response body back over H3
@@ -182,7 +196,9 @@ async fn handle_h3_request(
match frame {
Ok(frame) => {
if let Ok(data) = frame.into_data() {
stream.send_data(data).await
stream
.send_data(data)
.await
.map_err(|e| anyhow::anyhow!("Failed to send H3 data: {}", e))?;
}
}
@@ -194,7 +210,9 @@ async fn handle_h3_request(
}
// Finish the H3 stream (send QUIC FIN)
stream.finish().await
stream
.finish()
.await
.map_err(|e| anyhow::anyhow!("Failed to finish H3 stream: {}", e))?;
Ok(())
+2 -1
View File
@@ -5,14 +5,15 @@
pub mod connection_pool;
pub mod counting_body;
pub mod h3_service;
pub mod protocol_cache;
pub mod proxy_service;
pub mod request_filter;
mod request_host;
pub mod response_filter;
pub mod shutdown_on_drop;
pub mod template;
pub mod upstream_selector;
pub mod h3_service;
pub use connection_pool::*;
pub use counting_body::*;
@@ -144,10 +144,14 @@ impl FailureState {
}
fn all_expired(&self) -> bool {
let h2_expired = self.h2.as_ref()
let h2_expired = self
.h2
.as_ref()
.map(|r| r.failed_at.elapsed() >= r.cooldown)
.unwrap_or(true);
let h3_expired = self.h3.as_ref()
let h3_expired = self
.h3
.as_ref()
.map(|r| r.failed_at.elapsed() >= r.cooldown)
.unwrap_or(true);
h2_expired && h3_expired
@@ -355,9 +359,13 @@ impl ProtocolCache {
let record = entry.get_mut(protocol);
let (consecutive, new_cooldown) = match record {
Some(existing) if existing.failed_at.elapsed() < existing.cooldown.saturating_mul(2) => {
Some(existing)
if existing.failed_at.elapsed() < existing.cooldown.saturating_mul(2) =>
{
// Still within the "recent" window — escalate
let c = existing.consecutive_failures.saturating_add(1)
let c = existing
.consecutive_failures
.saturating_add(1)
.min(PROTOCOL_FAILURE_ESCALATION_CAP);
(c, escalate_cooldown(c))
}
@@ -394,8 +402,13 @@ impl ProtocolCache {
if protocol == DetectedProtocol::H1 {
return false;
}
self.failures.get(key)
.and_then(|entry| entry.get(protocol).map(|r| r.failed_at.elapsed() < r.cooldown))
self.failures
.get(key)
.and_then(|entry| {
entry
.get(protocol)
.map(|r| r.failed_at.elapsed() < r.cooldown)
})
.unwrap_or(false)
}
@@ -464,19 +477,18 @@ impl ProtocolCache {
/// Snapshot all non-expired cache entries for metrics/UI display.
pub fn snapshot(&self) -> Vec<ProtocolCacheEntry> {
self.cache.iter()
self.cache
.iter()
.filter(|entry| entry.value().last_accessed_at.elapsed() < PROTOCOL_CACHE_TTL)
.map(|entry| {
let key = entry.key();
let val = entry.value();
let failure_info = self.failures.get(key);
let (h2_sup, h2_cd, h2_cons) = Self::suppression_info(
failure_info.as_deref().and_then(|f| f.h2.as_ref()),
);
let (h3_sup, h3_cd, h3_cons) = Self::suppression_info(
failure_info.as_deref().and_then(|f| f.h3.as_ref()),
);
let (h2_sup, h2_cd, h2_cons) =
Self::suppression_info(failure_info.as_deref().and_then(|f| f.h2.as_ref()));
let (h3_sup, h3_cd, h3_cons) =
Self::suppression_info(failure_info.as_deref().and_then(|f| f.h3.as_ref()));
ProtocolCacheEntry {
host: key.host.clone(),
@@ -507,7 +519,13 @@ impl ProtocolCache {
/// Insert a protocol detection result with an optional H3 port.
/// Logs protocol transitions when overwriting an existing entry.
/// No suppression check — callers must check before calling.
fn insert_internal(&self, key: ProtocolCacheKey, protocol: DetectedProtocol, h3_port: Option<u16>, reason: &str) {
fn insert_internal(
&self,
key: ProtocolCacheKey,
protocol: DetectedProtocol,
h3_port: Option<u16>,
reason: &str,
) {
// Check for existing entry to log protocol transitions
if let Some(existing) = self.cache.get(&key) {
if existing.protocol != protocol {
@@ -522,7 +540,9 @@ impl ProtocolCache {
// Evict oldest entry if at capacity
if self.cache.len() >= PROTOCOL_CACHE_MAX_ENTRIES && !self.cache.contains_key(&key) {
let oldest = self.cache.iter()
let oldest = self
.cache
.iter()
.min_by_key(|entry| entry.value().last_accessed_at)
.map(|entry| entry.key().clone());
if let Some(oldest_key) = oldest {
@@ -531,13 +551,16 @@ impl ProtocolCache {
}
let now = Instant::now();
self.cache.insert(key, CachedEntry {
protocol,
detected_at: now,
last_accessed_at: now,
last_probed_at: now,
h3_port,
});
self.cache.insert(
key,
CachedEntry {
protocol,
detected_at: now,
last_accessed_at: now,
last_probed_at: now,
h3_port,
},
);
}
/// Reduce a failure record's remaining cooldown to `target`, if it currently
@@ -582,26 +605,34 @@ impl ProtocolCache {
interval.tick().await;
// Clean expired cache entries (sliding TTL based on last_accessed_at)
let expired: Vec<ProtocolCacheKey> = cache.iter()
let expired: Vec<ProtocolCacheKey> = cache
.iter()
.filter(|entry| entry.value().last_accessed_at.elapsed() >= PROTOCOL_CACHE_TTL)
.map(|entry| entry.key().clone())
.collect();
if !expired.is_empty() {
debug!("Protocol cache cleanup: removing {} expired entries", expired.len());
debug!(
"Protocol cache cleanup: removing {} expired entries",
expired.len()
);
for key in expired {
cache.remove(&key);
}
}
// Clean fully-expired failure entries
let expired_failures: Vec<ProtocolCacheKey> = failures.iter()
let expired_failures: Vec<ProtocolCacheKey> = failures
.iter()
.filter(|entry| entry.value().all_expired())
.map(|entry| entry.key().clone())
.collect();
if !expired_failures.is_empty() {
debug!("Protocol cache cleanup: removing {} expired failure entries", expired_failures.len());
debug!(
"Protocol cache cleanup: removing {} expired failure entries",
expired_failures.len()
);
for key in expired_failures {
failures.remove(&key);
}
@@ -609,7 +640,8 @@ impl ProtocolCache {
// Safety net: cap failures map at 2× max entries
if failures.len() > PROTOCOL_CACHE_MAX_ENTRIES * 2 {
let oldest: Vec<ProtocolCacheKey> = failures.iter()
let oldest: Vec<ProtocolCacheKey> = failures
.iter()
.filter(|e| e.value().all_expired())
.map(|e| e.key().clone())
.take(failures.len() - PROTOCOL_CACHE_MAX_ENTRIES)
File diff suppressed because it is too large Load Diff
+141 -41
View File
@@ -4,13 +4,15 @@ use std::net::SocketAddr;
use std::sync::Arc;
use bytes::Bytes;
use http_body_util::Full;
use http_body_util::BodyExt;
use hyper::{Request, Response, StatusCode};
use http_body_util::combinators::BoxBody;
use http_body_util::BodyExt;
use http_body_util::Full;
use hyper::{Request, Response, StatusCode};
use rustproxy_config::RouteSecurity;
use rustproxy_security::{IpFilter, BasicAuthValidator, JwtValidator, RateLimiter};
use rustproxy_security::{BasicAuthValidator, IpFilter, JwtValidator, RateLimiter};
use crate::request_host::extract_request_host;
pub struct RequestFilter;
@@ -35,16 +37,13 @@ impl RequestFilter {
let client_ip = peer_addr.ip();
let request_path = req.uri().path();
// IP filter (domain-aware: extract Host header for domain-scoped entries)
// IP filter (domain-aware: use the same host extraction as route matching)
if security.ip_allow_list.is_some() || security.ip_block_list.is_some() {
let allow = security.ip_allow_list.as_deref().unwrap_or(&[]);
let block = security.ip_block_list.as_deref().unwrap_or(&[]);
let filter = IpFilter::new(allow, block);
let normalized = IpFilter::normalize_ip(&client_ip);
let host = req.headers()
.get("host")
.and_then(|v| v.to_str().ok())
.map(|h| h.split(':').next().unwrap_or(h));
let host = extract_request_host(req);
if !filter.is_allowed_for_domain(&normalized, host) {
return Some(error_response(StatusCode::FORBIDDEN, "Access denied"));
}
@@ -59,16 +58,15 @@ impl RequestFilter {
!limiter.check(&key)
} else {
// Create a per-check limiter (less ideal but works for non-shared case)
let limiter = RateLimiter::new(
rate_limit_config.max_requests,
rate_limit_config.window,
);
let limiter =
RateLimiter::new(rate_limit_config.max_requests, rate_limit_config.window);
let key = Self::rate_limit_key(rate_limit_config, req, peer_addr);
!limiter.check(&key)
};
if should_block {
let message = rate_limit_config.error_message
let message = rate_limit_config
.error_message
.as_deref()
.unwrap_or("Rate limit exceeded");
return Some(error_response(StatusCode::TOO_MANY_REQUESTS, message));
@@ -84,36 +82,48 @@ impl RequestFilter {
if let Some(ref basic_auth) = security.basic_auth {
if basic_auth.enabled {
// Check basic auth exclude paths
let skip_basic = basic_auth.exclude_paths.as_ref()
let skip_basic = basic_auth
.exclude_paths
.as_ref()
.map(|paths| Self::path_matches_any(request_path, paths))
.unwrap_or(false);
if !skip_basic {
let users: Vec<(String, String)> = basic_auth.users.iter()
let users: Vec<(String, String)> = basic_auth
.users
.iter()
.map(|c| (c.username.clone(), c.password.clone()))
.collect();
let validator = BasicAuthValidator::new(users, basic_auth.realm.clone());
let auth_header = req.headers()
let auth_header = req
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok());
match auth_header {
Some(header) => {
if validator.validate(header).is_none() {
return Some(Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header("WWW-Authenticate", validator.www_authenticate())
.body(boxed_body("Invalid credentials"))
.unwrap());
return Some(
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header(
"WWW-Authenticate",
validator.www_authenticate(),
)
.body(boxed_body("Invalid credentials"))
.unwrap(),
);
}
}
None => {
return Some(Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header("WWW-Authenticate", validator.www_authenticate())
.body(boxed_body("Authentication required"))
.unwrap());
return Some(
Response::builder()
.status(StatusCode::UNAUTHORIZED)
.header("WWW-Authenticate", validator.www_authenticate())
.body(boxed_body("Authentication required"))
.unwrap(),
);
}
}
}
@@ -124,7 +134,9 @@ impl RequestFilter {
if let Some(ref jwt_auth) = security.jwt_auth {
if jwt_auth.enabled {
// Check JWT auth exclude paths
let skip_jwt = jwt_auth.exclude_paths.as_ref()
let skip_jwt = jwt_auth
.exclude_paths
.as_ref()
.map(|paths| Self::path_matches_any(request_path, paths))
.unwrap_or(false);
@@ -136,18 +148,25 @@ impl RequestFilter {
jwt_auth.audience.as_deref(),
);
let auth_header = req.headers()
let auth_header = req
.headers()
.get("authorization")
.and_then(|v| v.to_str().ok());
match auth_header.and_then(JwtValidator::extract_token) {
Some(token) => {
if validator.validate(token).is_err() {
return Some(error_response(StatusCode::UNAUTHORIZED, "Invalid token"));
return Some(error_response(
StatusCode::UNAUTHORIZED,
"Invalid token",
));
}
}
None => {
return Some(error_response(StatusCode::UNAUTHORIZED, "Bearer token required"));
return Some(error_response(
StatusCode::UNAUTHORIZED,
"Bearer token required",
));
}
}
}
@@ -209,7 +228,11 @@ impl RequestFilter {
/// Check IP-based security (for use in passthrough / TCP-level connections).
/// `domain` is the SNI from the TLS handshake (if available) for domain-scoped filtering.
/// Returns true if allowed, false if blocked.
pub fn check_ip_security(security: &RouteSecurity, client_ip: &std::net::IpAddr, domain: Option<&str>) -> bool {
pub fn check_ip_security(
security: &RouteSecurity,
client_ip: &std::net::IpAddr,
domain: Option<&str>,
) -> bool {
if security.ip_allow_list.is_some() || security.ip_block_list.is_some() {
let allow = security.ip_allow_list.as_deref().unwrap_or(&[]);
let block = security.ip_block_list.as_deref().unwrap_or(&[]);
@@ -238,19 +261,28 @@ impl RequestFilter {
return None;
}
let origin = req.headers()
let origin = req
.headers()
.get("origin")
.and_then(|v| v.to_str().ok())
.unwrap_or("*");
Some(Response::builder()
.status(StatusCode::NO_CONTENT)
.header("Access-Control-Allow-Origin", origin)
.header("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, PATCH, OPTIONS")
.header("Access-Control-Allow-Headers", "Content-Type, Authorization, X-Requested-With")
.header("Access-Control-Max-Age", "86400")
.body(boxed_body(""))
.unwrap())
Some(
Response::builder()
.status(StatusCode::NO_CONTENT)
.header("Access-Control-Allow-Origin", origin)
.header(
"Access-Control-Allow-Methods",
"GET, POST, PUT, DELETE, PATCH, OPTIONS",
)
.header(
"Access-Control-Allow-Headers",
"Content-Type, Authorization, X-Requested-With",
)
.header("Access-Control-Max-Age", "86400")
.body(boxed_body(""))
.unwrap(),
)
}
}
@@ -265,3 +297,71 @@ fn error_response(status: StatusCode, message: &str) -> Response<BoxBody<Bytes,
fn boxed_body(data: &str) -> BoxBody<Bytes, hyper::Error> {
BoxBody::new(Full::new(Bytes::from(data.to_string())).map_err(|never| match never {}))
}
#[cfg(test)]
mod tests {
use bytes::Bytes;
use http_body_util::Empty;
use hyper::{Request, StatusCode, Version};
use rustproxy_config::{IpAllowEntry, RouteSecurity};
use super::RequestFilter;
fn domain_scoped_security() -> RouteSecurity {
RouteSecurity {
ip_allow_list: Some(vec![IpAllowEntry::DomainScoped {
ip: "10.8.0.2".to_string(),
domains: vec!["*.abc.xyz".to_string()],
}]),
ip_block_list: None,
max_connections: None,
authentication: None,
rate_limit: None,
basic_auth: None,
jwt_auth: None,
}
}
fn peer_addr() -> std::net::SocketAddr {
std::net::SocketAddr::from(([10, 8, 0, 2], 4242))
}
fn request(uri: &str, version: Version, host: Option<&str>) -> Request<Empty<Bytes>> {
let mut builder = Request::builder().uri(uri).version(version);
if let Some(host) = host {
builder = builder.header("host", host);
}
builder.body(Empty::<Bytes>::new()).unwrap()
}
#[test]
fn domain_scoped_acl_allows_uri_authority_without_host_header() {
let security = domain_scoped_security();
let req = request("https://outline.abc.xyz/", Version::HTTP_2, None);
assert!(RequestFilter::apply(&security, &req, &peer_addr()).is_none());
}
#[test]
fn domain_scoped_acl_allows_host_header_with_port() {
let security = domain_scoped_security();
let req = request(
"https://unrelated.invalid/",
Version::HTTP_11,
Some("outline.abc.xyz:443"),
);
assert!(RequestFilter::apply(&security, &req, &peer_addr()).is_none());
}
#[test]
fn domain_scoped_acl_denies_non_matching_uri_authority() {
let security = domain_scoped_security();
let req = request("https://outline.other.xyz/", Version::HTTP_2, None);
let response = RequestFilter::apply(&security, &req, &peer_addr())
.expect("non-matching domain should be denied");
assert_eq!(response.status(), StatusCode::FORBIDDEN);
}
}
@@ -0,0 +1,43 @@
use hyper::Request;
/// Extract the effective request host for routing and scoped ACL checks.
///
/// Prefer the explicit `Host` header when present, otherwise fall back to the
/// URI authority used by HTTP/2 and HTTP/3 requests.
pub(crate) fn extract_request_host<B>(req: &Request<B>) -> Option<&str> {
req.headers()
.get("host")
.and_then(|value| value.to_str().ok())
.map(|host| host.split(':').next().unwrap_or(host))
.or_else(|| req.uri().host())
}
#[cfg(test)]
mod tests {
use bytes::Bytes;
use http_body_util::Empty;
use hyper::Request;
use super::extract_request_host;
#[test]
fn extracts_host_header_before_uri_authority() {
let req = Request::builder()
.uri("https://uri.abc.xyz/test")
.header("host", "header.abc.xyz:443")
.body(Empty::<Bytes>::new())
.unwrap();
assert_eq!(extract_request_host(&req), Some("header.abc.xyz"));
}
#[test]
fn falls_back_to_uri_authority_when_host_header_missing() {
let req = Request::builder()
.uri("https://outline.abc.xyz/test")
.body(Empty::<Bytes>::new())
.unwrap();
assert_eq!(extract_request_host(&req), Some("outline.abc.xyz"));
}
}
@@ -3,7 +3,7 @@
use hyper::header::{HeaderMap, HeaderName, HeaderValue};
use rustproxy_config::RouteConfig;
use crate::template::{RequestContext, expand_template};
use crate::template::{expand_template, RequestContext};
pub struct ResponseFilter;
@@ -11,12 +11,17 @@ impl ResponseFilter {
/// Apply response headers from route config and CORS settings.
/// If a `RequestContext` is provided, template variables in header values will be expanded.
/// Also injects Alt-Svc header for routes with HTTP/3 enabled.
pub fn apply_headers(route: &RouteConfig, headers: &mut HeaderMap, req_ctx: Option<&RequestContext>) {
pub fn apply_headers(
route: &RouteConfig,
headers: &mut HeaderMap,
req_ctx: Option<&RequestContext>,
) {
// Inject Alt-Svc for HTTP/3 advertisement if QUIC/HTTP3 is enabled on this route
if let Some(ref udp) = route.action.udp {
if let Some(ref quic) = udp.quic {
if quic.enable_http3.unwrap_or(false) {
let port = quic.alt_svc_port
let port = quic
.alt_svc_port
.or_else(|| req_ctx.map(|c| c.port))
.unwrap_or(443);
let max_age = quic.alt_svc_max_age.unwrap_or(86400);
@@ -63,10 +68,7 @@ impl ResponseFilter {
headers.insert("access-control-allow-origin", val);
}
} else {
headers.insert(
"access-control-allow-origin",
HeaderValue::from_static("*"),
);
headers.insert("access-control-allow-origin", HeaderValue::from_static("*"));
}
// Allow-Methods
@@ -62,17 +62,11 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send + 'static> AsyncWrite for Shutdown
self.inner.as_ref().unwrap().is_write_vectored()
}
fn poll_flush(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<io::Result<()>> {
fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Pin::new(self.get_mut().inner.as_mut().unwrap()).poll_flush(cx)
}
fn poll_shutdown(
self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Poll<io::Result<()>> {
fn poll_shutdown(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let this = self.get_mut();
let result = Pin::new(this.inner.as_mut().unwrap()).poll_shutdown(cx);
if result.is_ready() {
@@ -93,7 +87,8 @@ impl<S: AsyncRead + AsyncWrite + Unpin + Send + 'static> Drop for ShutdownOnDrop
let _ = tokio::time::timeout(
std::time::Duration::from_secs(2),
tokio::io::AsyncWriteExt::shutdown(&mut stream),
).await;
)
.await;
// stream is dropped here — all resources freed
});
}
+6 -2
View File
@@ -39,7 +39,8 @@ pub fn expand_headers(
headers: &HashMap<String, String>,
ctx: &RequestContext,
) -> HashMap<String, String> {
headers.iter()
headers
.iter()
.map(|(k, v)| (k.clone(), expand_template(v, ctx)))
.collect()
}
@@ -150,7 +151,10 @@ mod tests {
let ctx = test_context();
let template = "{clientIp}|{domain}|{port}|{path}|{routeName}|{connectionId}";
let result = expand_template(template, &ctx);
assert_eq!(result, "192.168.1.100|example.com|443|/api/v1/users|api-route|42");
assert_eq!(
result,
"192.168.1.100|example.com|443|/api/v1/users|api-route|42"
);
}
#[test]
@@ -7,7 +7,7 @@ use std::sync::Arc;
use std::sync::Mutex;
use dashmap::DashMap;
use rustproxy_config::{RouteTarget, LoadBalancingAlgorithm};
use rustproxy_config::{LoadBalancingAlgorithm, RouteTarget};
/// Upstream selection result.
pub struct UpstreamSelection {
@@ -51,21 +51,19 @@ impl UpstreamSelector {
}
// Determine load balancing algorithm
let algorithm = target.load_balancing.as_ref()
let algorithm = target
.load_balancing
.as_ref()
.map(|lb| &lb.algorithm)
.unwrap_or(&LoadBalancingAlgorithm::RoundRobin);
let idx = match algorithm {
LoadBalancingAlgorithm::RoundRobin => {
self.round_robin_select(&hosts, port)
}
LoadBalancingAlgorithm::RoundRobin => self.round_robin_select(&hosts, port),
LoadBalancingAlgorithm::IpHash => {
let hash = Self::ip_hash(client_addr);
hash % hosts.len()
}
LoadBalancingAlgorithm::LeastConnections => {
self.least_connections_select(&hosts, port)
}
LoadBalancingAlgorithm::LeastConnections => self.least_connections_select(&hosts, port),
};
UpstreamSelection {
@@ -78,9 +76,7 @@ impl UpstreamSelector {
fn round_robin_select(&self, hosts: &[&str], port: u16) -> usize {
let key = format!("{}:{}", hosts[0], port);
let mut counters = self.round_robin.lock().unwrap();
let counter = counters
.entry(key)
.or_insert_with(|| AtomicUsize::new(0));
let counter = counters.entry(key).or_insert_with(|| AtomicUsize::new(0));
let idx = counter.fetch_add(1, Ordering::Relaxed);
idx % hosts.len()
}
@@ -91,7 +87,8 @@ impl UpstreamSelector {
for (i, host) in hosts.iter().enumerate() {
let key = format!("{}:{}", host, port);
let conns = self.active_connections
let conns = self
.active_connections
.get(&key)
.map(|entry| entry.value().load(Ordering::Relaxed))
.unwrap_or(0);
@@ -228,13 +225,21 @@ mod tests {
selector.connection_started("backend:8080");
selector.connection_started("backend:8080");
assert_eq!(
selector.active_connections.get("backend:8080").unwrap().load(Ordering::Relaxed),
selector
.active_connections
.get("backend:8080")
.unwrap()
.load(Ordering::Relaxed),
2
);
selector.connection_ended("backend:8080");
assert_eq!(
selector.active_connections.get("backend:8080").unwrap().load(Ordering::Relaxed),
selector
.active_connections
.get("backend:8080")
.unwrap()
.load(Ordering::Relaxed),
1
);
+537 -145
View File
@@ -1,6 +1,6 @@
use dashmap::DashMap;
use serde::{Deserialize, Serialize};
use std::collections::HashSet;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Mutex;
use std::time::Duration;
@@ -62,6 +62,8 @@ pub struct IpMetrics {
pub bytes_out: u64,
pub throughput_in_bytes_per_sec: u64,
pub throughput_out_bytes_per_sec: u64,
/// Per-domain request/connection counts for this IP.
pub domain_requests: HashMap<String, u64>,
}
/// Per-backend metrics (keyed by "host:port").
@@ -139,6 +141,18 @@ const MAX_IPS_IN_SNAPSHOT: usize = 100;
/// Maximum number of backends to include in a snapshot (top by total connections).
const MAX_BACKENDS_IN_SNAPSHOT: usize = 100;
/// Maximum number of distinct domains tracked per IP (prevents subdomain-spray abuse).
const MAX_DOMAINS_PER_IP: usize = 256;
fn canonicalize_domain_key(domain: &str) -> Option<String> {
let normalized = domain.trim().trim_end_matches('.').to_ascii_lowercase();
if normalized.is_empty() {
None
} else {
Some(normalized)
}
}
/// Metrics collector tracking connections and throughput.
///
/// Design: The hot path (`record_bytes`) is entirely lock-free — it only touches
@@ -165,6 +179,10 @@ pub struct MetricsCollector {
ip_bytes_out: DashMap<String, AtomicU64>,
ip_pending_tp: DashMap<String, (AtomicU64, AtomicU64)>,
ip_throughput: DashMap<String, Mutex<ThroughputTracker>>,
/// Per-IP domain request counts: IP → { domain → count }.
/// Tracks which domains each frontend IP has requested (via HTTP Host/SNI).
/// Inner DashMap uses 2 shards to minimise base memory per IP.
ip_domain_requests: DashMap<String, DashMap<String, AtomicU64>>,
// ── Per-backend tracking (keyed by "host:port") ──
backend_active: DashMap<String, AtomicU64>,
@@ -247,6 +265,7 @@ impl MetricsCollector {
ip_bytes_out: DashMap::new(),
ip_pending_tp: DashMap::new(),
ip_throughput: DashMap::new(),
ip_domain_requests: DashMap::new(),
backend_active: DashMap::new(),
backend_total: DashMap::new(),
backend_protocol: DashMap::new(),
@@ -324,25 +343,43 @@ impl MetricsCollector {
/// Record a connection closing.
pub fn connection_closed(&self, route_id: Option<&str>, source_ip: Option<&str>) {
self.active_connections.fetch_sub(1, Ordering::Relaxed);
self.active_connections
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
if let Some(route_id) = route_id {
if let Some(counter) = self.route_connections.get(route_id) {
let val = counter.load(Ordering::Relaxed);
if val > 0 {
counter.fetch_sub(1, Ordering::Relaxed);
}
counter
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
}
}
if let Some(ip) = source_ip {
if let Some(counter) = self.ip_connections.get(ip) {
let val = counter.load(Ordering::Relaxed);
if val > 0 {
counter.fetch_sub(1, Ordering::Relaxed);
}
let prev = counter
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
// Clean up zero-count entries to prevent memory growth
if val <= 1 {
if matches!(prev, Some(v) if v <= 1) {
drop(counter);
self.ip_connections.remove(ip);
// Evict all per-IP tracking data for this IP
@@ -351,6 +388,7 @@ impl MetricsCollector {
self.ip_bytes_out.remove(ip);
self.ip_pending_tp.remove(ip);
self.ip_throughput.remove(ip);
self.ip_domain_requests.remove(ip);
}
}
}
@@ -360,17 +398,25 @@ impl MetricsCollector {
///
/// Called per-chunk in the TCP copy loop. Only touches AtomicU64 counters —
/// no Mutex is taken. The throughput trackers are fed during `sample_all()`.
pub fn record_bytes(&self, bytes_in: u64, bytes_out: u64, route_id: Option<&str>, source_ip: Option<&str>) {
pub fn record_bytes(
&self,
bytes_in: u64,
bytes_out: u64,
route_id: Option<&str>,
source_ip: Option<&str>,
) {
// Short-circuit: only touch counters for the direction that has data.
// CountingBody always calls with one direction zero — skipping the zero
// direction avoids ~50% of DashMap shard-locked reads per call.
if bytes_in > 0 {
self.total_bytes_in.fetch_add(bytes_in, Ordering::Relaxed);
self.global_pending_tp_in.fetch_add(bytes_in, Ordering::Relaxed);
self.global_pending_tp_in
.fetch_add(bytes_in, Ordering::Relaxed);
}
if bytes_out > 0 {
self.total_bytes_out.fetch_add(bytes_out, Ordering::Relaxed);
self.global_pending_tp_out.fetch_add(bytes_out, Ordering::Relaxed);
self.global_pending_tp_out
.fetch_add(bytes_out, Ordering::Relaxed);
}
// Per-route tracking: use get() first (zero-alloc fast path for existing entries),
@@ -380,7 +426,8 @@ impl MetricsCollector {
if let Some(counter) = self.route_bytes_in.get(route_id) {
counter.fetch_add(bytes_in, Ordering::Relaxed);
} else {
self.route_bytes_in.entry(route_id.to_string())
self.route_bytes_in
.entry(route_id.to_string())
.or_insert_with(|| AtomicU64::new(0))
.fetch_add(bytes_in, Ordering::Relaxed);
}
@@ -389,7 +436,8 @@ impl MetricsCollector {
if let Some(counter) = self.route_bytes_out.get(route_id) {
counter.fetch_add(bytes_out, Ordering::Relaxed);
} else {
self.route_bytes_out.entry(route_id.to_string())
self.route_bytes_out
.entry(route_id.to_string())
.or_insert_with(|| AtomicU64::new(0))
.fetch_add(bytes_out, Ordering::Relaxed);
}
@@ -397,13 +445,23 @@ impl MetricsCollector {
// Accumulate into per-route pending throughput counters (lock-free)
if let Some(entry) = self.route_pending_tp.get(route_id) {
if bytes_in > 0 { entry.0.fetch_add(bytes_in, Ordering::Relaxed); }
if bytes_out > 0 { entry.1.fetch_add(bytes_out, Ordering::Relaxed); }
if bytes_in > 0 {
entry.0.fetch_add(bytes_in, Ordering::Relaxed);
}
if bytes_out > 0 {
entry.1.fetch_add(bytes_out, Ordering::Relaxed);
}
} else {
let entry = self.route_pending_tp.entry(route_id.to_string())
let entry = self
.route_pending_tp
.entry(route_id.to_string())
.or_insert_with(|| (AtomicU64::new(0), AtomicU64::new(0)));
if bytes_in > 0 { entry.0.fetch_add(bytes_in, Ordering::Relaxed); }
if bytes_out > 0 { entry.1.fetch_add(bytes_out, Ordering::Relaxed); }
if bytes_in > 0 {
entry.0.fetch_add(bytes_in, Ordering::Relaxed);
}
if bytes_out > 0 {
entry.1.fetch_add(bytes_out, Ordering::Relaxed);
}
}
}
@@ -417,7 +475,8 @@ impl MetricsCollector {
if let Some(counter) = self.ip_bytes_in.get(ip) {
counter.fetch_add(bytes_in, Ordering::Relaxed);
} else {
self.ip_bytes_in.entry(ip.to_string())
self.ip_bytes_in
.entry(ip.to_string())
.or_insert_with(|| AtomicU64::new(0))
.fetch_add(bytes_in, Ordering::Relaxed);
}
@@ -426,7 +485,8 @@ impl MetricsCollector {
if let Some(counter) = self.ip_bytes_out.get(ip) {
counter.fetch_add(bytes_out, Ordering::Relaxed);
} else {
self.ip_bytes_out.entry(ip.to_string())
self.ip_bytes_out
.entry(ip.to_string())
.or_insert_with(|| AtomicU64::new(0))
.fetch_add(bytes_out, Ordering::Relaxed);
}
@@ -434,13 +494,23 @@ impl MetricsCollector {
// Accumulate into per-IP pending throughput counters (lock-free)
if let Some(entry) = self.ip_pending_tp.get(ip) {
if bytes_in > 0 { entry.0.fetch_add(bytes_in, Ordering::Relaxed); }
if bytes_out > 0 { entry.1.fetch_add(bytes_out, Ordering::Relaxed); }
if bytes_in > 0 {
entry.0.fetch_add(bytes_in, Ordering::Relaxed);
}
if bytes_out > 0 {
entry.1.fetch_add(bytes_out, Ordering::Relaxed);
}
} else {
let entry = self.ip_pending_tp.entry(ip.to_string())
let entry = self
.ip_pending_tp
.entry(ip.to_string())
.or_insert_with(|| (AtomicU64::new(0), AtomicU64::new(0)));
if bytes_in > 0 { entry.0.fetch_add(bytes_in, Ordering::Relaxed); }
if bytes_out > 0 { entry.1.fetch_add(bytes_out, Ordering::Relaxed); }
if bytes_in > 0 {
entry.0.fetch_add(bytes_in, Ordering::Relaxed);
}
if bytes_out > 0 {
entry.1.fetch_add(bytes_out, Ordering::Relaxed);
}
}
}
}
@@ -452,6 +522,41 @@ impl MetricsCollector {
self.pending_http_requests.fetch_add(1, Ordering::Relaxed);
}
/// Record a domain request/connection for a frontend IP.
///
/// Called per HTTP request (with Host header) and per TCP passthrough
/// connection (with SNI domain). The common case (IP + domain both already
/// tracked) is two DashMap reads + one atomic increment — zero allocation.
pub fn record_ip_domain_request(&self, ip: &str, domain: &str) {
let Some(domain) = canonicalize_domain_key(domain) else {
return;
};
// Fast path: IP already tracked, domain already tracked
if let Some(domains) = self.ip_domain_requests.get(ip) {
if let Some(counter) = domains.get(domain.as_str()) {
counter.fetch_add(1, Ordering::Relaxed);
return;
}
// New domain for this IP — enforce cap
if domains.len() >= MAX_DOMAINS_PER_IP {
return;
}
domains
.entry(domain)
.or_insert_with(|| AtomicU64::new(0))
.fetch_add(1, Ordering::Relaxed);
return;
}
// New IP — only create if the IP has active connections
if !self.ip_connections.contains_key(ip) {
return;
}
let inner = DashMap::with_capacity_and_shard_amount(4, 2);
inner.insert(domain, AtomicU64::new(1));
self.ip_domain_requests.insert(ip.to_string(), inner);
}
// ── UDP session recording methods ──
/// Record a new UDP session opened.
@@ -462,7 +567,15 @@ impl MetricsCollector {
/// Record a UDP session closed.
pub fn udp_session_closed(&self) {
self.active_udp_sessions.fetch_sub(1, Ordering::Relaxed);
self.active_udp_sessions
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
}
/// Record a UDP datagram (inbound or outbound).
@@ -511,9 +624,15 @@ impl MetricsCollector {
let (active, _) = self.frontend_proto_counters(proto);
// Atomic saturating decrement — avoids TOCTOU race where concurrent
// closes could both read val=1, both subtract, wrapping to u64::MAX.
active.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 { Some(v - 1) } else { None }
}).ok();
active
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
}
/// Record a backend connection opened with a given protocol.
@@ -527,9 +646,15 @@ impl MetricsCollector {
pub fn backend_protocol_closed(&self, proto: &str) {
let (active, _) = self.backend_proto_counters(proto);
// Atomic saturating decrement — see frontend_protocol_closed for rationale.
active.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 { Some(v - 1) } else { None }
}).ok();
active
.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |v| {
if v > 0 {
Some(v - 1)
} else {
None
}
})
.ok();
}
// ── Per-backend recording methods ──
@@ -639,17 +764,28 @@ impl MetricsCollector {
/// Remove per-backend metrics for backends no longer in any route target.
pub fn retain_backends(&self, active_backends: &HashSet<String>) {
self.backend_active.retain(|k, _| active_backends.contains(k));
self.backend_total.retain(|k, _| active_backends.contains(k));
self.backend_protocol.retain(|k, _| active_backends.contains(k));
self.backend_connect_errors.retain(|k, _| active_backends.contains(k));
self.backend_handshake_errors.retain(|k, _| active_backends.contains(k));
self.backend_request_errors.retain(|k, _| active_backends.contains(k));
self.backend_connect_time_us.retain(|k, _| active_backends.contains(k));
self.backend_connect_count.retain(|k, _| active_backends.contains(k));
self.backend_pool_hits.retain(|k, _| active_backends.contains(k));
self.backend_pool_misses.retain(|k, _| active_backends.contains(k));
self.backend_h2_failures.retain(|k, _| active_backends.contains(k));
self.backend_active
.retain(|k, _| active_backends.contains(k));
self.backend_total
.retain(|k, _| active_backends.contains(k));
self.backend_protocol
.retain(|k, _| active_backends.contains(k));
self.backend_connect_errors
.retain(|k, _| active_backends.contains(k));
self.backend_handshake_errors
.retain(|k, _| active_backends.contains(k));
self.backend_request_errors
.retain(|k, _| active_backends.contains(k));
self.backend_connect_time_us
.retain(|k, _| active_backends.contains(k));
self.backend_connect_count
.retain(|k, _| active_backends.contains(k));
self.backend_pool_hits
.retain(|k, _| active_backends.contains(k));
self.backend_pool_misses
.retain(|k, _| active_backends.contains(k));
self.backend_h2_failures
.retain(|k, _| active_backends.contains(k));
}
/// Take a throughput sample on all trackers (cold path, call at 1Hz or configured interval).
@@ -740,40 +876,64 @@ impl MetricsCollector {
// Safety-net: prune orphaned per-IP entries that have no corresponding
// ip_connections entry. This catches any entries created by a race between
// record_bytes and connection_closed.
self.ip_bytes_in.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_bytes_out.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_pending_tp.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_throughput.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_total_connections.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_bytes_in
.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_bytes_out
.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_pending_tp
.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_throughput
.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_total_connections
.retain(|k, _| self.ip_connections.contains_key(k));
self.ip_domain_requests
.retain(|k, _| self.ip_connections.contains_key(k));
// Safety-net: prune orphaned backend error/stats entries for backends
// that have no active or total connections (error-only backends).
// These accumulate when backend_connect_error/backend_handshake_error
// create entries but backend_connection_opened is never called.
let known_backends: HashSet<String> = self.backend_active.iter()
let known_backends: HashSet<String> = self
.backend_active
.iter()
.map(|e| e.key().clone())
.chain(self.backend_total.iter().map(|e| e.key().clone()))
.collect();
self.backend_connect_errors.retain(|k, _| known_backends.contains(k));
self.backend_handshake_errors.retain(|k, _| known_backends.contains(k));
self.backend_request_errors.retain(|k, _| known_backends.contains(k));
self.backend_connect_time_us.retain(|k, _| known_backends.contains(k));
self.backend_connect_count.retain(|k, _| known_backends.contains(k));
self.backend_pool_hits.retain(|k, _| known_backends.contains(k));
self.backend_pool_misses.retain(|k, _| known_backends.contains(k));
self.backend_h2_failures.retain(|k, _| known_backends.contains(k));
self.backend_protocol.retain(|k, _| known_backends.contains(k));
self.backend_connect_errors
.retain(|k, _| known_backends.contains(k));
self.backend_handshake_errors
.retain(|k, _| known_backends.contains(k));
self.backend_request_errors
.retain(|k, _| known_backends.contains(k));
self.backend_connect_time_us
.retain(|k, _| known_backends.contains(k));
self.backend_connect_count
.retain(|k, _| known_backends.contains(k));
self.backend_pool_hits
.retain(|k, _| known_backends.contains(k));
self.backend_pool_misses
.retain(|k, _| known_backends.contains(k));
self.backend_h2_failures
.retain(|k, _| known_backends.contains(k));
self.backend_protocol
.retain(|k, _| known_backends.contains(k));
}
/// Remove per-route metrics for route IDs that are no longer active.
/// Call this after `update_routes()` to prune stale entries.
pub fn retain_routes(&self, active_route_ids: &HashSet<String>) {
self.route_connections.retain(|k, _| active_route_ids.contains(k));
self.route_total_connections.retain(|k, _| active_route_ids.contains(k));
self.route_bytes_in.retain(|k, _| active_route_ids.contains(k));
self.route_bytes_out.retain(|k, _| active_route_ids.contains(k));
self.route_pending_tp.retain(|k, _| active_route_ids.contains(k));
self.route_throughput.retain(|k, _| active_route_ids.contains(k));
self.route_connections
.retain(|k, _| active_route_ids.contains(k));
self.route_total_connections
.retain(|k, _| active_route_ids.contains(k));
self.route_bytes_in
.retain(|k, _| active_route_ids.contains(k));
self.route_bytes_out
.retain(|k, _| active_route_ids.contains(k));
self.route_pending_tp
.retain(|k, _| active_route_ids.contains(k));
self.route_throughput
.retain(|k, _| active_route_ids.contains(k));
}
/// Get current active connection count.
@@ -816,78 +976,116 @@ impl MetricsCollector {
for entry in self.route_total_connections.iter() {
let route_id = entry.key().clone();
let total = entry.value().load(Ordering::Relaxed);
let active = self.route_connections
let active = self
.route_connections
.get(&route_id)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let bytes_in = self.route_bytes_in
let bytes_in = self
.route_bytes_in
.get(&route_id)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let bytes_out = self.route_bytes_out
let bytes_out = self
.route_bytes_out
.get(&route_id)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let (route_tp_in, route_tp_out, route_recent_in, route_recent_out) = self.route_throughput
let (route_tp_in, route_tp_out, route_recent_in, route_recent_out) = self
.route_throughput
.get(&route_id)
.and_then(|entry| entry.value().lock().ok().map(|t| {
let (i_in, i_out) = t.instant();
let (r_in, r_out) = t.recent();
(i_in, i_out, r_in, r_out)
}))
.and_then(|entry| {
entry.value().lock().ok().map(|t| {
let (i_in, i_out) = t.instant();
let (r_in, r_out) = t.recent();
(i_in, i_out, r_in, r_out)
})
})
.unwrap_or((0, 0, 0, 0));
routes.insert(route_id, RouteMetrics {
active_connections: active,
total_connections: total,
bytes_in,
bytes_out,
throughput_in_bytes_per_sec: route_tp_in,
throughput_out_bytes_per_sec: route_tp_out,
throughput_recent_in_bytes_per_sec: route_recent_in,
throughput_recent_out_bytes_per_sec: route_recent_out,
});
routes.insert(
route_id,
RouteMetrics {
active_connections: active,
total_connections: total,
bytes_in,
bytes_out,
throughput_in_bytes_per_sec: route_tp_in,
throughput_out_bytes_per_sec: route_tp_out,
throughput_recent_in_bytes_per_sec: route_recent_in,
throughput_recent_out_bytes_per_sec: route_recent_out,
},
);
}
// Collect per-IP metrics — only IPs with active connections or total > 0,
// capped at top MAX_IPS_IN_SNAPSHOT sorted by active count
let mut ip_entries: Vec<(String, u64, u64, u64, u64, u64, u64)> = Vec::new();
let mut ip_entries: Vec<(String, u64, u64, u64, u64, u64, u64, HashMap<String, u64>)> =
Vec::new();
for entry in self.ip_total_connections.iter() {
let ip = entry.key().clone();
let total = entry.value().load(Ordering::Relaxed);
let active = self.ip_connections
let active = self
.ip_connections
.get(&ip)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let bytes_in = self.ip_bytes_in
let bytes_in = self
.ip_bytes_in
.get(&ip)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let bytes_out = self.ip_bytes_out
let bytes_out = self
.ip_bytes_out
.get(&ip)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let (tp_in, tp_out) = self.ip_throughput
let (tp_in, tp_out) = self
.ip_throughput
.get(&ip)
.and_then(|entry| entry.value().lock().ok().map(|t| t.instant()))
.unwrap_or((0, 0));
ip_entries.push((ip, active, total, bytes_in, bytes_out, tp_in, tp_out));
// Collect per-domain request counts for this IP
let domain_requests = self
.ip_domain_requests
.get(&ip)
.map(|domains| {
domains
.iter()
.map(|e| (e.key().clone(), e.value().load(Ordering::Relaxed)))
.collect()
})
.unwrap_or_default();
ip_entries.push((
ip,
active,
total,
bytes_in,
bytes_out,
tp_in,
tp_out,
domain_requests,
));
}
// Sort by active connections descending, then cap
ip_entries.sort_by(|a, b| b.1.cmp(&a.1));
ip_entries.truncate(MAX_IPS_IN_SNAPSHOT);
let mut ips = std::collections::HashMap::new();
for (ip, active, total, bytes_in, bytes_out, tp_in, tp_out) in ip_entries {
ips.insert(ip, IpMetrics {
active_connections: active,
total_connections: total,
bytes_in,
bytes_out,
throughput_in_bytes_per_sec: tp_in,
throughput_out_bytes_per_sec: tp_out,
});
for (ip, active, total, bytes_in, bytes_out, tp_in, tp_out, domain_requests) in ip_entries {
ips.insert(
ip,
IpMetrics {
active_connections: active,
total_connections: total,
bytes_in,
bytes_out,
throughput_in_bytes_per_sec: tp_in,
throughput_out_bytes_per_sec: tp_out,
domain_requests,
},
);
}
// Collect per-backend metrics, capped at top MAX_BACKENDS_IN_SNAPSHOT by total connections
@@ -895,69 +1093,84 @@ impl MetricsCollector {
for entry in self.backend_total.iter() {
let key = entry.key().clone();
let total = entry.value().load(Ordering::Relaxed);
let active = self.backend_active
let active = self
.backend_active
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let protocol = self.backend_protocol
let protocol = self
.backend_protocol
.get(&key)
.map(|v| v.value().clone())
.unwrap_or_else(|| "unknown".to_string());
let connect_errors = self.backend_connect_errors
let connect_errors = self
.backend_connect_errors
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let handshake_errors = self.backend_handshake_errors
let handshake_errors = self
.backend_handshake_errors
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let request_errors = self.backend_request_errors
let request_errors = self
.backend_request_errors
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let total_connect_time_us = self.backend_connect_time_us
let total_connect_time_us = self
.backend_connect_time_us
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let connect_count = self.backend_connect_count
let connect_count = self
.backend_connect_count
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let pool_hits = self.backend_pool_hits
let pool_hits = self
.backend_pool_hits
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let pool_misses = self.backend_pool_misses
let pool_misses = self
.backend_pool_misses
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
let h2_failures = self.backend_h2_failures
let h2_failures = self
.backend_h2_failures
.get(&key)
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0);
backend_entries.push((key, BackendMetrics {
active_connections: active,
total_connections: total,
protocol,
connect_errors,
handshake_errors,
request_errors,
total_connect_time_us,
connect_count,
pool_hits,
pool_misses,
h2_failures,
}));
backend_entries.push((
key,
BackendMetrics {
active_connections: active,
total_connections: total,
protocol,
connect_errors,
handshake_errors,
request_errors,
total_connect_time_us,
connect_count,
pool_hits,
pool_misses,
h2_failures,
},
));
}
// Sort by total connections descending, then cap
backend_entries.sort_by(|a, b| b.1.total_connections.cmp(&a.1.total_connections));
backend_entries.truncate(MAX_BACKENDS_IN_SNAPSHOT);
let backends: std::collections::HashMap<String, BackendMetrics> = backend_entries.into_iter().collect();
let backends: std::collections::HashMap<String, BackendMetrics> =
backend_entries.into_iter().collect();
// HTTP request rates
let (http_rps, http_rps_recent) = self.http_request_throughput
let (http_rps, http_rps_recent) = self
.http_request_throughput
.lock()
.map(|t| {
let (instant, _) = t.instant();
@@ -1132,11 +1345,19 @@ mod tests {
// Check IP active connections (drop DashMap refs immediately to avoid deadlock)
assert_eq!(
collector.ip_connections.get("1.2.3.4").unwrap().load(Ordering::Relaxed),
collector
.ip_connections
.get("1.2.3.4")
.unwrap()
.load(Ordering::Relaxed),
2
);
assert_eq!(
collector.ip_connections.get("5.6.7.8").unwrap().load(Ordering::Relaxed),
collector
.ip_connections
.get("5.6.7.8")
.unwrap()
.load(Ordering::Relaxed),
1
);
@@ -1154,7 +1375,11 @@ mod tests {
// Close connections
collector.connection_closed(Some("route-a"), Some("1.2.3.4"));
assert_eq!(
collector.ip_connections.get("1.2.3.4").unwrap().load(Ordering::Relaxed),
collector
.ip_connections
.get("1.2.3.4")
.unwrap()
.load(Ordering::Relaxed),
1
);
@@ -1199,6 +1424,79 @@ mod tests {
assert!(collector.ip_total_connections.get("10.0.0.2").is_some());
}
#[test]
fn test_connection_closed_saturates_active_gauges() {
let collector = MetricsCollector::new();
collector.connection_closed(Some("route-a"), Some("10.0.0.1"));
assert_eq!(collector.active_connections(), 0);
collector.connection_opened(Some("route-a"), Some("10.0.0.1"));
collector.connection_closed(Some("route-a"), Some("10.0.0.1"));
collector.connection_closed(Some("route-a"), Some("10.0.0.1"));
assert_eq!(collector.active_connections(), 0);
assert_eq!(
collector
.route_connections
.get("route-a")
.map(|c| c.load(Ordering::Relaxed))
.unwrap_or(0),
0
);
assert!(collector.ip_connections.get("10.0.0.1").is_none());
}
#[test]
fn test_udp_session_closed_saturates() {
let collector = MetricsCollector::new();
collector.udp_session_closed();
assert_eq!(collector.snapshot().active_udp_sessions, 0);
collector.udp_session_opened();
collector.udp_session_closed();
collector.udp_session_closed();
assert_eq!(collector.snapshot().active_udp_sessions, 0);
}
#[test]
fn test_ip_domain_requests_are_canonicalized() {
let collector = MetricsCollector::new();
collector.connection_opened(Some("route-a"), Some("10.0.0.1"));
collector.record_ip_domain_request("10.0.0.1", "Example.COM");
collector.record_ip_domain_request("10.0.0.1", "example.com.");
collector.record_ip_domain_request("10.0.0.1", " example.com ");
let snapshot = collector.snapshot();
let ip_metrics = snapshot.ips.get("10.0.0.1").unwrap();
assert_eq!(ip_metrics.domain_requests.len(), 1);
assert_eq!(ip_metrics.domain_requests.get("example.com"), Some(&3));
}
#[test]
fn test_protocol_metrics_appear_in_snapshot() {
let collector = MetricsCollector::new();
collector.frontend_protocol_opened("h2");
collector.frontend_protocol_opened("ws");
collector.backend_protocol_opened("h3");
collector.backend_protocol_opened("ws");
collector.frontend_protocol_closed("h2");
collector.backend_protocol_closed("h3");
let snapshot = collector.snapshot();
assert_eq!(snapshot.frontend_protocols.h2_active, 0);
assert_eq!(snapshot.frontend_protocols.h2_total, 1);
assert_eq!(snapshot.frontend_protocols.ws_active, 1);
assert_eq!(snapshot.frontend_protocols.ws_total, 1);
assert_eq!(snapshot.backend_protocols.h3_active, 0);
assert_eq!(snapshot.backend_protocols.h3_total, 1);
assert_eq!(snapshot.backend_protocols.ws_active, 1);
assert_eq!(snapshot.backend_protocols.ws_total, 1);
}
#[test]
fn test_http_request_tracking() {
let collector = MetricsCollector::with_retention(60);
@@ -1273,9 +1571,16 @@ mod tests {
let collector = MetricsCollector::with_retention(60);
// Manually insert orphaned entries (simulates the race before the guard)
collector.ip_bytes_in.insert("orphan-ip".to_string(), AtomicU64::new(100));
collector.ip_bytes_out.insert("orphan-ip".to_string(), AtomicU64::new(200));
collector.ip_pending_tp.insert("orphan-ip".to_string(), (AtomicU64::new(0), AtomicU64::new(0)));
collector
.ip_bytes_in
.insert("orphan-ip".to_string(), AtomicU64::new(100));
collector
.ip_bytes_out
.insert("orphan-ip".to_string(), AtomicU64::new(200));
collector.ip_pending_tp.insert(
"orphan-ip".to_string(),
(AtomicU64::new(0), AtomicU64::new(0)),
);
// No ip_connections entry for "orphan-ip"
assert!(collector.ip_connections.get("orphan-ip").is_none());
@@ -1313,17 +1618,59 @@ mod tests {
collector.backend_connection_opened(key, Duration::from_millis(15));
collector.backend_connection_opened(key, Duration::from_millis(25));
assert_eq!(collector.backend_active.get(key).unwrap().load(Ordering::Relaxed), 2);
assert_eq!(collector.backend_total.get(key).unwrap().load(Ordering::Relaxed), 2);
assert_eq!(collector.backend_connect_count.get(key).unwrap().load(Ordering::Relaxed), 2);
assert_eq!(
collector
.backend_active
.get(key)
.unwrap()
.load(Ordering::Relaxed),
2
);
assert_eq!(
collector
.backend_total
.get(key)
.unwrap()
.load(Ordering::Relaxed),
2
);
assert_eq!(
collector
.backend_connect_count
.get(key)
.unwrap()
.load(Ordering::Relaxed),
2
);
// 15ms + 25ms = 40ms = 40_000us
assert_eq!(collector.backend_connect_time_us.get(key).unwrap().load(Ordering::Relaxed), 40_000);
assert_eq!(
collector
.backend_connect_time_us
.get(key)
.unwrap()
.load(Ordering::Relaxed),
40_000
);
// Close one
collector.backend_connection_closed(key);
assert_eq!(collector.backend_active.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(
collector
.backend_active
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
// total stays
assert_eq!(collector.backend_total.get(key).unwrap().load(Ordering::Relaxed), 2);
assert_eq!(
collector
.backend_total
.get(key)
.unwrap()
.load(Ordering::Relaxed),
2
);
// Record errors
collector.backend_connect_error(key);
@@ -1334,12 +1681,54 @@ mod tests {
collector.backend_pool_hit(key);
collector.backend_pool_miss(key);
assert_eq!(collector.backend_connect_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(collector.backend_handshake_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(collector.backend_request_errors.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(collector.backend_h2_failures.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(collector.backend_pool_hits.get(key).unwrap().load(Ordering::Relaxed), 2);
assert_eq!(collector.backend_pool_misses.get(key).unwrap().load(Ordering::Relaxed), 1);
assert_eq!(
collector
.backend_connect_errors
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
assert_eq!(
collector
.backend_handshake_errors
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
assert_eq!(
collector
.backend_request_errors
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
assert_eq!(
collector
.backend_h2_failures
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
assert_eq!(
collector
.backend_pool_hits
.get(key)
.unwrap()
.load(Ordering::Relaxed),
2
);
assert_eq!(
collector
.backend_pool_misses
.get(key)
.unwrap()
.load(Ordering::Relaxed),
1
);
// Protocol
collector.set_backend_protocol(key, "h1");
@@ -1396,7 +1785,10 @@ mod tests {
assert!(collector.backend_total.get("stale:8080").is_none());
assert!(collector.backend_protocol.get("stale:8080").is_none());
assert!(collector.backend_connect_errors.get("stale:8080").is_none());
assert!(collector.backend_connect_time_us.get("stale:8080").is_none());
assert!(collector
.backend_connect_time_us
.get("stale:8080")
.is_none());
assert!(collector.backend_connect_count.get("stale:8080").is_none());
assert!(collector.backend_pool_hits.get("stale:8080").is_none());
assert!(collector.backend_pool_misses.get("stale:8080").is_none());
@@ -943,6 +943,9 @@ impl TcpListenerManager {
// Track connection in metrics — guard ensures connection_closed on all exit paths
metrics.connection_opened(route_id, Some(&ip_str));
if let Some(d) = effective_domain {
metrics.record_ip_domain_request(&ip_str, d);
}
let _conn_guard = ConnectionGuard::new(Arc::clone(&metrics), route_id, Some(&ip_str));
// Check if this is a socket-handler route that should be relayed to TypeScript
@@ -1,5 +1,42 @@
use std::collections::HashMap;
use regex::Regex;
use std::collections::HashMap;
fn compile_regex_pattern(pattern: &str) -> Option<Regex> {
if !pattern.starts_with('/') {
return None;
}
let last_slash = pattern.rfind('/')?;
if last_slash == 0 {
return None;
}
let regex_body = &pattern[1..last_slash];
let flags = &pattern[last_slash + 1..];
let mut inline_flags = String::new();
for flag in flags.chars() {
match flag {
'i' | 'm' | 's' | 'u' => {
if !inline_flags.contains(flag) {
inline_flags.push(flag);
}
}
'g' => {
// Global has no effect for single header matching.
}
_ => return None,
}
}
let compiled = if inline_flags.is_empty() {
regex_body.to_string()
} else {
format!("(?{}){}", inline_flags, regex_body)
};
Regex::new(&compiled).ok()
}
/// Match HTTP headers against a set of patterns.
///
@@ -24,16 +61,15 @@ pub fn headers_match(
None => return false, // Required header not present
};
// Check if pattern is a regex (surrounded by /)
if pattern.starts_with('/') && pattern.ends_with('/') && pattern.len() > 2 {
let regex_str = &pattern[1..pattern.len() - 1];
match Regex::new(regex_str) {
Ok(re) => {
// Check if pattern is a regex literal (/pattern/ or /pattern/flags)
if pattern.starts_with('/') && pattern.len() > 2 {
match compile_regex_pattern(pattern) {
Some(re) => {
if !re.is_match(header_value) {
return false;
}
}
Err(_) => {
None => {
// Invalid regex, fall back to exact match
if header_value != pattern {
return false;
@@ -85,6 +121,24 @@ mod tests {
assert!(headers_match(&patterns, &headers));
}
#[test]
fn test_regex_header_match_with_flags() {
let patterns: HashMap<String, String> = {
let mut m = HashMap::new();
m.insert(
"Content-Type".to_string(),
"/^application\\/json$/i".to_string(),
);
m
};
let headers: HashMap<String, String> = {
let mut m = HashMap::new();
m.insert("content-type".to_string(), "Application/JSON".to_string());
m
};
assert!(headers_match(&patterns, &headers));
}
#[test]
fn test_missing_header() {
let patterns: HashMap<String, String> = {
+25
View File
@@ -537,6 +537,31 @@ tap.test('Route Matching - routeMatchesHeaders', async () => {
'X-Custom-Header': 'value'
})).toBeFalse();
const regexHeaderRoute: IRouteConfig = {
match: {
domains: 'example.com',
ports: 80,
headers: {
'Content-Type': /^application\/(json|problem\+json)$/i,
}
},
action: {
type: 'forward',
targets: [{
host: 'localhost',
port: 3000
}]
}
};
expect(routeMatchesHeaders(regexHeaderRoute, {
'Content-Type': 'Application/Problem+Json',
})).toBeTrue();
expect(routeMatchesHeaders(regexHeaderRoute, {
'Content-Type': 'text/html',
})).toBeFalse();
// Route without header matching should match any headers
const noHeaderRoute: IRouteConfig = {
match: { ports: 80, domains: 'example.com' },
+192
View File
@@ -0,0 +1,192 @@
import { expect, tap } from '@git.zone/tstest/tapbundle';
import type { ISmartProxyOptions } from '../ts/proxies/smart-proxy/models/interfaces.js';
import type { IRouteConfig } from '../ts/proxies/smart-proxy/models/route-types.js';
import { RoutePreprocessor } from '../ts/proxies/smart-proxy/route-preprocessor.js';
import { buildRustProxyOptions } from '../ts/proxies/smart-proxy/utils/rust-config.js';
tap.test('Rust contract - preprocessor serializes regex headers for Rust', async () => {
const route: IRouteConfig = {
name: 'contract-route',
match: {
ports: [443, { from: 8443, to: 8444 }],
domains: ['api.example.com', '*.example.com'],
transport: 'udp',
protocol: 'http3',
headers: {
'Content-Type': /^application\/json$/i,
},
},
action: {
type: 'forward',
targets: [{
match: {
ports: [443],
path: '/api/*',
method: ['GET'],
headers: {
'X-Env': /^(prod|stage)$/,
},
},
host: ['backend-a', 'backend-b'],
port: 'preserve',
sendProxyProtocol: true,
backendTransport: 'tcp',
}],
tls: {
mode: 'terminate',
certificate: 'auto',
},
sendProxyProtocol: true,
udp: {
maxSessionsPerIP: 321,
quic: {
enableHttp3: true,
},
},
},
security: {
ipAllowList: [{
ip: '10.0.0.0/8',
domains: ['api.example.com'],
}],
},
};
const preprocessor = new RoutePreprocessor();
const [rustRoute] = preprocessor.preprocessForRust([route]);
expect(rustRoute.match.headers?.['Content-Type']).toEqual('/^application\\/json$/i');
expect(rustRoute.match.transport).toEqual('udp');
expect(rustRoute.match.protocol).toEqual('http3');
expect(rustRoute.action.targets?.[0].match?.headers?.['X-Env']).toEqual('/^(prod|stage)$/');
expect(rustRoute.action.targets?.[0].port).toEqual('preserve');
expect(rustRoute.action.targets?.[0].backendTransport).toEqual('tcp');
expect(rustRoute.action.sendProxyProtocol).toBeTrue();
expect(rustRoute.action.udp?.maxSessionsPerIp).toEqual(321);
});
tap.test('Rust contract - preprocessor converts dynamic targets to relay-safe payloads', async () => {
const route: IRouteConfig = {
name: 'dynamic-contract-route',
match: {
ports: 8080,
},
action: {
type: 'forward',
targets: [{
host: () => 'dynamic-backend.internal',
port: () => 9443,
}],
},
};
const preprocessor = new RoutePreprocessor();
const [rustRoute] = preprocessor.preprocessForRust([route]);
expect(rustRoute.action.type).toEqual('socket-handler');
expect(rustRoute.action.targets?.[0].host).toEqual('localhost');
expect(rustRoute.action.targets?.[0].port).toEqual(0);
expect(preprocessor.getOriginalRoute('dynamic-contract-route')).toEqual(route);
});
tap.test('Rust contract - top-level config keeps shared SmartProxy settings', async () => {
const settings: ISmartProxyOptions = {
routes: [{
name: 'top-level-contract-route',
match: {
ports: 443,
domains: 'api.example.com',
},
action: {
type: 'forward',
targets: [{
host: 'backend.internal',
port: 8443,
}],
tls: {
mode: 'terminate',
certificate: 'auto',
},
},
}],
preserveSourceIP: true,
proxyIPs: ['10.0.0.1'],
acceptProxyProtocol: true,
sendProxyProtocol: true,
noDelay: true,
keepAlive: true,
keepAliveInitialDelay: 1500,
maxPendingDataSize: 4096,
disableInactivityCheck: true,
enableKeepAliveProbes: true,
enableDetailedLogging: true,
enableTlsDebugLogging: true,
enableRandomizedTimeouts: true,
connectionTimeout: 5000,
initialDataTimeout: 7000,
socketTimeout: 9000,
inactivityCheckInterval: 1100,
maxConnectionLifetime: 13000,
inactivityTimeout: 15000,
gracefulShutdownTimeout: 17000,
maxConnectionsPerIP: 20,
connectionRateLimitPerMinute: 30,
keepAliveTreatment: 'extended',
keepAliveInactivityMultiplier: 2,
extendedKeepAliveLifetime: 19000,
metrics: {
enabled: true,
sampleIntervalMs: 250,
retentionSeconds: 60,
},
acme: {
enabled: true,
email: 'ops@example.com',
environment: 'staging',
useProduction: false,
skipConfiguredCerts: true,
renewThresholdDays: 14,
renewCheckIntervalHours: 12,
autoRenew: true,
port: 80,
},
};
const preprocessor = new RoutePreprocessor();
const routes = preprocessor.preprocessForRust(settings.routes);
const config = buildRustProxyOptions(settings, routes);
expect(config.preserveSourceIp).toBeTrue();
expect(config.proxyIps).toEqual(['10.0.0.1']);
expect(config.acceptProxyProtocol).toBeTrue();
expect(config.sendProxyProtocol).toBeTrue();
expect(config.noDelay).toBeTrue();
expect(config.keepAlive).toBeTrue();
expect(config.keepAliveInitialDelay).toEqual(1500);
expect(config.maxPendingDataSize).toEqual(4096);
expect(config.disableInactivityCheck).toBeTrue();
expect(config.enableKeepAliveProbes).toBeTrue();
expect(config.enableDetailedLogging).toBeTrue();
expect(config.enableTlsDebugLogging).toBeTrue();
expect(config.enableRandomizedTimeouts).toBeTrue();
expect(config.connectionTimeout).toEqual(5000);
expect(config.initialDataTimeout).toEqual(7000);
expect(config.socketTimeout).toEqual(9000);
expect(config.inactivityCheckInterval).toEqual(1100);
expect(config.maxConnectionLifetime).toEqual(13000);
expect(config.inactivityTimeout).toEqual(15000);
expect(config.gracefulShutdownTimeout).toEqual(17000);
expect(config.maxConnectionsPerIp).toEqual(20);
expect(config.connectionRateLimitPerMinute).toEqual(30);
expect(config.keepAliveTreatment).toEqual('extended');
expect(config.keepAliveInactivityMultiplier).toEqual(2);
expect(config.extendedKeepAliveLifetime).toEqual(19000);
expect(config.metrics?.sampleIntervalMs).toEqual(250);
expect(config.acme?.email).toEqual('ops@example.com');
expect(config.acme?.environment).toEqual('staging');
expect(config.acme?.skipConfiguredCerts).toBeTrue();
expect(config.acme?.renewThresholdDays).toEqual(14);
});
export default tap.start();
+1 -1
View File
@@ -3,6 +3,6 @@
*/
export const commitinfo = {
name: '@push.rocks/smartproxy',
version: '27.5.0',
version: '27.7.1',
description: 'A powerful proxy package with unified route-based configuration for high traffic management. Features include SSL/TLS support, flexible routing patterns, WebSocket handling, advanced security options, and automatic ACME certificate management.'
}
@@ -57,6 +57,10 @@ export interface IMetrics {
byRoute(): Map<string, number>;
byIP(): Map<string, number>;
topIPs(limit?: number): Array<{ ip: string; count: number }>;
/** Per-IP domain request counts: IP -> { domain -> count }. */
domainRequestsByIP(): Map<string, Map<string, number>>;
/** Top IP-domain pairs sorted by request count descending. */
topDomainRequests(limit?: number): Array<{ ip: string; domain: string; count: number }>;
frontendProtocols(): IProtocolDistribution;
backendProtocols(): IProtocolDistribution;
};
+160
View File
@@ -0,0 +1,160 @@
import type { IProtocolCacheEntry, IProtocolDistribution } from './metrics-types.js';
import type { IAcmeOptions, ISmartProxyOptions } from './interfaces.js';
import type {
IRouteAction,
IRouteConfig,
IRouteMatch,
IRouteTarget,
ITargetMatch,
IRouteUdp,
} from './route-types.js';
export type TRustHeaderMatchers = Record<string, string>;
export interface IRustRouteMatch extends Omit<IRouteMatch, 'headers'> {
headers?: TRustHeaderMatchers;
}
export interface IRustTargetMatch extends Omit<ITargetMatch, 'headers'> {
headers?: TRustHeaderMatchers;
}
export interface IRustRouteTarget extends Omit<IRouteTarget, 'host' | 'port' | 'match'> {
host: string | string[];
port: number | 'preserve';
match?: IRustTargetMatch;
}
export interface IRustRouteUdp extends Omit<IRouteUdp, 'maxSessionsPerIP'> {
maxSessionsPerIp?: number;
}
export interface IRustDefaultConfig extends Omit<NonNullable<ISmartProxyOptions['defaults']>, 'preserveSourceIP'> {
preserveSourceIp?: boolean;
}
export interface IRustRouteAction
extends Omit<IRouteAction, 'targets' | 'socketHandler' | 'datagramHandler' | 'forwardingEngine' | 'nftables' | 'udp'> {
targets?: IRustRouteTarget[];
udp?: IRustRouteUdp;
}
export interface IRustRouteConfig extends Omit<IRouteConfig, 'match' | 'action'> {
match: IRustRouteMatch;
action: IRustRouteAction;
}
export interface IRustAcmeOptions extends Omit<IAcmeOptions, 'routeForwards'> {}
export interface IRustProxyOptions {
routes: IRustRouteConfig[];
preserveSourceIp?: boolean;
proxyIps?: string[];
acceptProxyProtocol?: boolean;
sendProxyProtocol?: boolean;
defaults?: IRustDefaultConfig;
connectionTimeout?: number;
initialDataTimeout?: number;
socketTimeout?: number;
inactivityCheckInterval?: number;
maxConnectionLifetime?: number;
inactivityTimeout?: number;
gracefulShutdownTimeout?: number;
noDelay?: boolean;
keepAlive?: boolean;
keepAliveInitialDelay?: number;
maxPendingDataSize?: number;
disableInactivityCheck?: boolean;
enableKeepAliveProbes?: boolean;
enableDetailedLogging?: boolean;
enableTlsDebugLogging?: boolean;
enableRandomizedTimeouts?: boolean;
maxConnectionsPerIp?: number;
connectionRateLimitPerMinute?: number;
keepAliveTreatment?: ISmartProxyOptions['keepAliveTreatment'];
keepAliveInactivityMultiplier?: number;
extendedKeepAliveLifetime?: number;
metrics?: ISmartProxyOptions['metrics'];
acme?: IRustAcmeOptions;
}
export interface IRustStatistics {
activeConnections: number;
totalConnections: number;
routesCount: number;
listeningPorts: number[];
uptimeSeconds: number;
}
export interface IRustCertificateStatus {
domain: string;
source: string;
expiresAt: number;
isValid: boolean;
}
export interface IRustThroughputSample {
timestampMs: number;
bytesIn: number;
bytesOut: number;
}
export interface IRustRouteMetrics {
activeConnections: number;
totalConnections: number;
bytesIn: number;
bytesOut: number;
throughputInBytesPerSec: number;
throughputOutBytesPerSec: number;
throughputRecentInBytesPerSec: number;
throughputRecentOutBytesPerSec: number;
}
export interface IRustIpMetrics {
activeConnections: number;
totalConnections: number;
bytesIn: number;
bytesOut: number;
throughputInBytesPerSec: number;
throughputOutBytesPerSec: number;
domainRequests: Record<string, number>;
}
export interface IRustBackendMetrics {
activeConnections: number;
totalConnections: number;
protocol: string;
connectErrors: number;
handshakeErrors: number;
requestErrors: number;
totalConnectTimeUs: number;
connectCount: number;
poolHits: number;
poolMisses: number;
h2Failures: number;
}
export interface IRustMetricsSnapshot {
activeConnections: number;
totalConnections: number;
bytesIn: number;
bytesOut: number;
throughputInBytesPerSec: number;
throughputOutBytesPerSec: number;
throughputRecentInBytesPerSec: number;
throughputRecentOutBytesPerSec: number;
routes: Record<string, IRustRouteMetrics>;
ips: Record<string, IRustIpMetrics>;
backends: Record<string, IRustBackendMetrics>;
throughputHistory: IRustThroughputSample[];
totalHttpRequests: number;
httpRequestsPerSec: number;
httpRequestsPerSecRecent: number;
activeUdpSessions: number;
totalUdpSessions: number;
totalDatagramsIn: number;
totalDatagramsOut: number;
detectedProtocols: IProtocolCacheEntry[];
frontendProtocols: IProtocolDistribution;
backendProtocols: IProtocolDistribution;
}
+13 -11
View File
@@ -1,5 +1,6 @@
import type { IRouteConfig, IRouteAction, IRouteTarget } from './models/route-types.js';
import { logger } from '../../core/utils/logger.js';
import type { IRustRouteConfig } from './models/rust-types.js';
import { serializeRouteForRust } from './utils/rust-config.js';
/**
* Preprocesses routes before sending them to Rust.
@@ -24,7 +25,7 @@ export class RoutePreprocessor {
* - Non-serializable fields are stripped
* - Original routes are preserved in the local map for handler lookup
*/
public preprocessForRust(routes: IRouteConfig[]): IRouteConfig[] {
public preprocessForRust(routes: IRouteConfig[]): IRustRouteConfig[] {
this.originalRoutes.clear();
return routes.map((route, index) => this.preprocessRoute(route, index));
}
@@ -43,7 +44,7 @@ export class RoutePreprocessor {
return new Map(this.originalRoutes);
}
private preprocessRoute(route: IRouteConfig, index: number): IRouteConfig {
private preprocessRoute(route: IRouteConfig, index: number): IRustRouteConfig {
const routeKey = route.name || route.id || `route_${index}`;
// Check if this route needs TS-side handling
@@ -57,7 +58,7 @@ export class RoutePreprocessor {
// Create a clean copy for Rust
const cleanRoute: IRouteConfig = {
...route,
action: this.cleanAction(route.action, routeKey, needsTsHandling),
action: this.cleanAction(route.action, needsTsHandling),
};
// Ensure we have a name for handler lookup
@@ -65,7 +66,7 @@ export class RoutePreprocessor {
cleanRoute.name = routeKey;
}
return cleanRoute;
return serializeRouteForRust(cleanRoute);
}
private routeNeedsTsHandling(route: IRouteConfig): boolean {
@@ -91,15 +92,16 @@ export class RoutePreprocessor {
return false;
}
private cleanAction(action: IRouteAction, routeKey: string, needsTsHandling: boolean): IRouteAction {
const cleanAction: IRouteAction = { ...action };
private cleanAction(action: IRouteAction, needsTsHandling: boolean): IRouteAction {
let cleanAction: IRouteAction = { ...action };
if (needsTsHandling) {
// Convert to socket-handler type for Rust (Rust will relay back to TS)
cleanAction.type = 'socket-handler';
// Remove the JS handlers (not serializable)
delete (cleanAction as any).socketHandler;
delete (cleanAction as any).datagramHandler;
const { socketHandler: _socketHandler, datagramHandler: _datagramHandler, ...serializableAction } = cleanAction;
cleanAction = {
...serializableAction,
type: 'socket-handler',
};
}
// Clean targets - replace functions with static values
+64 -32
View File
@@ -1,5 +1,6 @@
import type { IMetrics, IBackendMetrics, IProtocolCacheEntry, IProtocolDistribution, IThroughputData, IThroughputHistoryPoint } from './models/metrics-types.js';
import type { RustProxyBridge } from './rust-proxy-bridge.js';
import type { IRustBackendMetrics, IRustIpMetrics, IRustMetricsSnapshot, IRustRouteMetrics } from './models/rust-types.js';
/**
* Adapts Rust JSON metrics to the IMetrics interface.
@@ -14,7 +15,7 @@ import type { RustProxyBridge } from './rust-proxy-bridge.js';
*/
export class RustMetricsAdapter implements IMetrics {
private bridge: RustProxyBridge;
private cache: any = null;
private cache: IRustMetricsSnapshot | null = null;
private pollTimer: ReturnType<typeof setInterval> | null = null;
private pollIntervalMs: number;
@@ -65,8 +66,8 @@ export class RustMetricsAdapter implements IMetrics {
byRoute: (): Map<string, number> => {
const result = new Map<string, number>();
if (this.cache?.routes) {
for (const [name, rm] of Object.entries(this.cache.routes)) {
result.set(name, (rm as any).activeConnections ?? 0);
for (const [name, rm] of Object.entries(this.cache.routes) as Array<[string, IRustRouteMetrics]>) {
result.set(name, rm.activeConnections ?? 0);
}
}
return result;
@@ -74,8 +75,8 @@ export class RustMetricsAdapter implements IMetrics {
byIP: (): Map<string, number> => {
const result = new Map<string, number>();
if (this.cache?.ips) {
for (const [ip, im] of Object.entries(this.cache.ips)) {
result.set(ip, (im as any).activeConnections ?? 0);
for (const [ip, im] of Object.entries(this.cache.ips) as Array<[string, IRustIpMetrics]>) {
result.set(ip, im.activeConnections ?? 0);
}
}
return result;
@@ -83,8 +84,41 @@ export class RustMetricsAdapter implements IMetrics {
topIPs: (limit: number = 10): Array<{ ip: string; count: number }> => {
const result: Array<{ ip: string; count: number }> = [];
if (this.cache?.ips) {
for (const [ip, im] of Object.entries(this.cache.ips)) {
result.push({ ip, count: (im as any).activeConnections ?? 0 });
for (const [ip, im] of Object.entries(this.cache.ips) as Array<[string, IRustIpMetrics]>) {
result.push({ ip, count: im.activeConnections ?? 0 });
}
}
result.sort((a, b) => b.count - a.count);
return result.slice(0, limit);
},
domainRequestsByIP: (): Map<string, Map<string, number>> => {
const result = new Map<string, Map<string, number>>();
if (this.cache?.ips) {
for (const [ip, im] of Object.entries(this.cache.ips) as Array<[string, IRustIpMetrics]>) {
const dr = im.domainRequests;
if (dr && typeof dr === 'object') {
const domainMap = new Map<string, number>();
for (const [domain, count] of Object.entries(dr)) {
domainMap.set(domain, count as number);
}
if (domainMap.size > 0) {
result.set(ip, domainMap);
}
}
}
}
return result;
},
topDomainRequests: (limit: number = 20): Array<{ ip: string; domain: string; count: number }> => {
const result: Array<{ ip: string; domain: string; count: number }> = [];
if (this.cache?.ips) {
for (const [ip, im] of Object.entries(this.cache.ips) as Array<[string, IRustIpMetrics]>) {
const dr = im.domainRequests;
if (dr && typeof dr === 'object') {
for (const [domain, count] of Object.entries(dr)) {
result.push({ ip, domain, count: count as number });
}
}
}
}
result.sort((a, b) => b.count - a.count);
@@ -143,7 +177,7 @@ export class RustMetricsAdapter implements IMetrics {
},
history: (seconds: number): Array<IThroughputHistoryPoint> => {
if (!this.cache?.throughputHistory) return [];
return this.cache.throughputHistory.slice(-seconds).map((p: any) => ({
return this.cache.throughputHistory.slice(-seconds).map((p) => ({
timestamp: p.timestampMs,
in: p.bytesIn,
out: p.bytesOut,
@@ -152,10 +186,10 @@ export class RustMetricsAdapter implements IMetrics {
byRoute: (_windowSeconds?: number): Map<string, IThroughputData> => {
const result = new Map<string, IThroughputData>();
if (this.cache?.routes) {
for (const [name, rm] of Object.entries(this.cache.routes)) {
for (const [name, rm] of Object.entries(this.cache.routes) as Array<[string, IRustRouteMetrics]>) {
result.set(name, {
in: (rm as any).throughputInBytesPerSec ?? 0,
out: (rm as any).throughputOutBytesPerSec ?? 0,
in: rm.throughputInBytesPerSec ?? 0,
out: rm.throughputOutBytesPerSec ?? 0,
});
}
}
@@ -164,10 +198,10 @@ export class RustMetricsAdapter implements IMetrics {
byIP: (_windowSeconds?: number): Map<string, IThroughputData> => {
const result = new Map<string, IThroughputData>();
if (this.cache?.ips) {
for (const [ip, im] of Object.entries(this.cache.ips)) {
for (const [ip, im] of Object.entries(this.cache.ips) as Array<[string, IRustIpMetrics]>) {
result.set(ip, {
in: (im as any).throughputInBytesPerSec ?? 0,
out: (im as any).throughputOutBytesPerSec ?? 0,
in: im.throughputInBytesPerSec ?? 0,
out: im.throughputOutBytesPerSec ?? 0,
});
}
}
@@ -203,23 +237,22 @@ export class RustMetricsAdapter implements IMetrics {
byBackend: (): Map<string, IBackendMetrics> => {
const result = new Map<string, IBackendMetrics>();
if (this.cache?.backends) {
for (const [key, bm] of Object.entries(this.cache.backends)) {
const m = bm as any;
const totalTimeUs = m.totalConnectTimeUs ?? 0;
const count = m.connectCount ?? 0;
const poolHits = m.poolHits ?? 0;
const poolMisses = m.poolMisses ?? 0;
for (const [key, bm] of Object.entries(this.cache.backends) as Array<[string, IRustBackendMetrics]>) {
const totalTimeUs = bm.totalConnectTimeUs ?? 0;
const count = bm.connectCount ?? 0;
const poolHits = bm.poolHits ?? 0;
const poolMisses = bm.poolMisses ?? 0;
const poolTotal = poolHits + poolMisses;
result.set(key, {
protocol: m.protocol ?? 'unknown',
activeConnections: m.activeConnections ?? 0,
totalConnections: m.totalConnections ?? 0,
connectErrors: m.connectErrors ?? 0,
handshakeErrors: m.handshakeErrors ?? 0,
requestErrors: m.requestErrors ?? 0,
protocol: bm.protocol ?? 'unknown',
activeConnections: bm.activeConnections ?? 0,
totalConnections: bm.totalConnections ?? 0,
connectErrors: bm.connectErrors ?? 0,
handshakeErrors: bm.handshakeErrors ?? 0,
requestErrors: bm.requestErrors ?? 0,
avgConnectTimeMs: count > 0 ? (totalTimeUs / count) / 1000 : 0,
poolHitRate: poolTotal > 0 ? poolHits / poolTotal : 0,
h2Failures: m.h2Failures ?? 0,
h2Failures: bm.h2Failures ?? 0,
});
}
}
@@ -228,8 +261,8 @@ export class RustMetricsAdapter implements IMetrics {
protocols: (): Map<string, string> => {
const result = new Map<string, string>();
if (this.cache?.backends) {
for (const [key, bm] of Object.entries(this.cache.backends)) {
result.set(key, (bm as any).protocol ?? 'unknown');
for (const [key, bm] of Object.entries(this.cache.backends) as Array<[string, IRustBackendMetrics]>) {
result.set(key, bm.protocol ?? 'unknown');
}
}
return result;
@@ -237,9 +270,8 @@ export class RustMetricsAdapter implements IMetrics {
topByErrors: (limit: number = 10): Array<{ backend: string; errors: number }> => {
const result: Array<{ backend: string; errors: number }> = [];
if (this.cache?.backends) {
for (const [key, bm] of Object.entries(this.cache.backends)) {
const m = bm as any;
const errors = (m.connectErrors ?? 0) + (m.handshakeErrors ?? 0) + (m.requestErrors ?? 0);
for (const [key, bm] of Object.entries(this.cache.backends) as Array<[string, IRustBackendMetrics]>) {
const errors = (bm.connectErrors ?? 0) + (bm.handshakeErrors ?? 0) + (bm.requestErrors ?? 0);
if (errors > 0) result.push({ backend: key, errors });
}
}
+24 -18
View File
@@ -1,23 +1,29 @@
import * as plugins from '../../plugins.js';
import { logger } from '../../core/utils/logger.js';
import type { IRouteConfig } from './models/route-types.js';
import type {
IRustCertificateStatus,
IRustMetricsSnapshot,
IRustProxyOptions,
IRustRouteConfig,
IRustStatistics,
} from './models/rust-types.js';
/**
* Type-safe command definitions for the Rust proxy IPC protocol.
*/
type TSmartProxyCommands = {
start: { params: { config: any }; result: void };
stop: { params: Record<string, never>; result: void };
updateRoutes: { params: { routes: IRouteConfig[] }; result: void };
getMetrics: { params: Record<string, never>; result: any };
getStatistics: { params: Record<string, never>; result: any };
provisionCertificate: { params: { routeName: string }; result: void };
renewCertificate: { params: { routeName: string }; result: void };
getCertificateStatus: { params: { routeName: string }; result: any };
getListeningPorts: { params: Record<string, never>; result: { ports: number[] } };
setSocketHandlerRelay: { params: { socketPath: string }; result: void };
addListeningPort: { params: { port: number }; result: void };
removeListeningPort: { params: { port: number }; result: void };
start: { params: { config: IRustProxyOptions }; result: void };
stop: { params: Record<string, never>; result: void };
updateRoutes: { params: { routes: IRustRouteConfig[] }; result: void };
getMetrics: { params: Record<string, never>; result: IRustMetricsSnapshot };
getStatistics: { params: Record<string, never>; result: IRustStatistics };
provisionCertificate: { params: { routeName: string }; result: void };
renewCertificate: { params: { routeName: string }; result: void };
getCertificateStatus: { params: { routeName: string }; result: IRustCertificateStatus | null };
getListeningPorts: { params: Record<string, never>; result: { ports: number[] } };
setSocketHandlerRelay: { params: { socketPath: string }; result: void };
addListeningPort: { params: { port: number }; result: void };
removeListeningPort: { params: { port: number }; result: void };
loadCertificate: { params: { domain: string; cert: string; key: string; ca?: string }; result: void };
setDatagramHandlerRelay: { params: { socketPath: string }; result: void };
};
@@ -121,7 +127,7 @@ export class RustProxyBridge extends plugins.EventEmitter {
// --- Convenience methods for each management command ---
public async startProxy(config: any): Promise<void> {
public async startProxy(config: IRustProxyOptions): Promise<void> {
await this.bridge.sendCommand('start', { config });
}
@@ -129,15 +135,15 @@ export class RustProxyBridge extends plugins.EventEmitter {
await this.bridge.sendCommand('stop', {} as Record<string, never>);
}
public async updateRoutes(routes: IRouteConfig[]): Promise<void> {
public async updateRoutes(routes: IRustRouteConfig[]): Promise<void> {
await this.bridge.sendCommand('updateRoutes', { routes });
}
public async getMetrics(): Promise<any> {
public async getMetrics(): Promise<IRustMetricsSnapshot> {
return this.bridge.sendCommand('getMetrics', {} as Record<string, never>);
}
public async getStatistics(): Promise<any> {
public async getStatistics(): Promise<IRustStatistics> {
return this.bridge.sendCommand('getStatistics', {} as Record<string, never>);
}
@@ -149,7 +155,7 @@ export class RustProxyBridge extends plugins.EventEmitter {
await this.bridge.sendCommand('renewCertificate', { routeName });
}
public async getCertificateStatus(routeName: string): Promise<any> {
public async getCertificateStatus(routeName: string): Promise<IRustCertificateStatus | null> {
return this.bridge.sendCommand('getCertificateStatus', { routeName });
}
+6 -33
View File
@@ -11,6 +11,7 @@ import { RustMetricsAdapter } from './rust-metrics-adapter.js';
// Route management
import { SharedRouteManager as RouteManager } from '../../core/routing/route-manager.js';
import { RouteValidator } from './utils/route-validator.js';
import { buildRustProxyOptions } from './utils/rust-config.js';
import { generateDefaultCertificate } from './utils/default-cert-generator.js';
import { Mutex } from './utils/mutex.js';
import { ConcurrencySemaphore } from './utils/concurrency-semaphore.js';
@@ -19,6 +20,7 @@ import { ConcurrencySemaphore } from './utils/concurrency-semaphore.js';
import type { ISmartProxyOptions, TSmartProxyCertProvisionObject, IAcmeOptions, ICertProvisionEventComms, ICertificateIssuedEvent, ICertificateFailedEvent } from './models/interfaces.js';
import type { IRouteConfig } from './models/route-types.js';
import type { IMetrics } from './models/metrics-types.js';
import type { IRustCertificateStatus, IRustProxyOptions, IRustStatistics } from './models/rust-types.js';
/**
* SmartProxy - Rust-backed proxy engine with TypeScript configuration API.
@@ -365,7 +367,7 @@ export class SmartProxy extends plugins.EventEmitter {
/**
* Get certificate status for a route (async - calls Rust).
*/
public async getCertificateStatus(routeName: string): Promise<any> {
public async getCertificateStatus(routeName: string): Promise<IRustCertificateStatus | null> {
return this.bridge.getCertificateStatus(routeName);
}
@@ -379,7 +381,7 @@ export class SmartProxy extends plugins.EventEmitter {
/**
* Get statistics (async - calls Rust).
*/
public async getStatistics(): Promise<any> {
public async getStatistics(): Promise<IRustStatistics> {
return this.bridge.getStatistics();
}
@@ -484,37 +486,8 @@ export class SmartProxy extends plugins.EventEmitter {
/**
* Build the Rust configuration object from TS settings.
*/
private buildRustConfig(routes: IRouteConfig[], acmeOverride?: IAcmeOptions): any {
const acme = acmeOverride !== undefined ? acmeOverride : this.settings.acme;
return {
routes,
defaults: this.settings.defaults,
acme: acme
? {
enabled: acme.enabled,
email: acme.email,
useProduction: acme.useProduction,
port: acme.port,
renewThresholdDays: acme.renewThresholdDays,
autoRenew: acme.autoRenew,
renewCheckIntervalHours: acme.renewCheckIntervalHours,
}
: undefined,
connectionTimeout: this.settings.connectionTimeout,
initialDataTimeout: this.settings.initialDataTimeout,
socketTimeout: this.settings.socketTimeout,
maxConnectionLifetime: this.settings.maxConnectionLifetime,
gracefulShutdownTimeout: this.settings.gracefulShutdownTimeout,
maxConnectionsPerIp: this.settings.maxConnectionsPerIP,
connectionRateLimitPerMinute: this.settings.connectionRateLimitPerMinute,
keepAliveTreatment: this.settings.keepAliveTreatment,
keepAliveInactivityMultiplier: this.settings.keepAliveInactivityMultiplier,
extendedKeepAliveLifetime: this.settings.extendedKeepAliveLifetime,
proxyIps: this.settings.proxyIPs,
acceptProxyProtocol: this.settings.acceptProxyProtocol,
sendProxyProtocol: this.settings.sendProxyProtocol,
metrics: this.settings.metrics,
};
private buildRustConfig(routes: IRustProxyOptions['routes'], acmeOverride?: IAcmeOptions): IRustProxyOptions {
return buildRustProxyOptions(this.settings, routes, acmeOverride);
}
/**
+22 -8
View File
@@ -168,14 +168,28 @@ export function routeMatchesHeaders(
if (!route.match?.headers || Object.keys(route.match.headers).length === 0) {
return true; // No headers specified means it matches any headers
}
// Convert RegExp patterns to strings for HeaderMatcher
const stringHeaders: Record<string, string> = {};
for (const [key, value] of Object.entries(route.match.headers)) {
stringHeaders[key] = value instanceof RegExp ? value.source : value;
for (const [headerName, expectedValue] of Object.entries(route.match.headers)) {
const actualKey = Object.keys(headers).find((key) => key.toLowerCase() === headerName.toLowerCase());
const actualValue = actualKey ? headers[actualKey] : undefined;
if (actualValue === undefined) {
return false;
}
if (expectedValue instanceof RegExp) {
if (!expectedValue.test(actualValue)) {
return false;
}
continue;
}
if (!HeaderMatcher.match(expectedValue, actualValue)) {
return false;
}
}
return HeaderMatcher.matchAll(stringHeaders, headers);
return true;
}
/**
@@ -283,4 +297,4 @@ export function generateRouteId(route: IRouteConfig): string {
*/
export function cloneRoute(route: IRouteConfig): IRouteConfig {
return JSON.parse(JSON.stringify(route));
}
}
+187
View File
@@ -0,0 +1,187 @@
import type { IAcmeOptions, ISmartProxyOptions } from '../models/interfaces.js';
import type { IRouteAction, IRouteConfig, IRouteMatch, IRouteTarget, ITargetMatch } from '../models/route-types.js';
import type {
IRustAcmeOptions,
IRustDefaultConfig,
IRustProxyOptions,
IRustRouteAction,
IRustRouteConfig,
IRustRouteMatch,
IRustRouteTarget,
IRustTargetMatch,
IRustRouteUdp,
TRustHeaderMatchers,
} from '../models/rust-types.js';
const SUPPORTED_REGEX_FLAGS = new Set(['i', 'm', 's', 'u', 'g']);
export function serializeHeaderMatchValue(value: string | RegExp): string {
if (typeof value === 'string') {
return value;
}
const unsupportedFlags = Array.from(new Set(value.flags)).filter((flag) => !SUPPORTED_REGEX_FLAGS.has(flag));
if (unsupportedFlags.length > 0) {
throw new Error(
`Header RegExp uses unsupported flags for Rust serialization: ${unsupportedFlags.join(', ')}`
);
}
return `/${value.source}/${value.flags}`;
}
export function serializeHeaderMatchers(headers?: Record<string, string | RegExp>): TRustHeaderMatchers | undefined {
if (!headers) {
return undefined;
}
return Object.fromEntries(
Object.entries(headers).map(([key, value]) => [key, serializeHeaderMatchValue(value)])
);
}
export function serializeTargetMatchForRust(match?: ITargetMatch): IRustTargetMatch | undefined {
if (!match) {
return undefined;
}
return {
...match,
headers: serializeHeaderMatchers(match.headers),
};
}
export function serializeRouteMatchForRust(match: IRouteMatch): IRustRouteMatch {
return {
...match,
headers: serializeHeaderMatchers(match.headers),
};
}
export function serializeRouteTargetForRust(target: IRouteTarget): IRustRouteTarget {
if (typeof target.host !== 'string' && !Array.isArray(target.host)) {
throw new Error('Route target host must be serialized before sending to Rust');
}
if (typeof target.port !== 'number' && target.port !== 'preserve') {
throw new Error('Route target port must be serialized before sending to Rust');
}
return {
...target,
host: target.host,
port: target.port,
match: serializeTargetMatchForRust(target.match),
};
}
function serializeUdpForRust(udp?: IRouteAction['udp']): IRustRouteUdp | undefined {
if (!udp) {
return undefined;
}
const { maxSessionsPerIP, ...rest } = udp;
return {
...rest,
maxSessionsPerIp: maxSessionsPerIP,
};
}
export function serializeRouteActionForRust(action: IRouteAction): IRustRouteAction {
const {
socketHandler: _socketHandler,
datagramHandler: _datagramHandler,
forwardingEngine: _forwardingEngine,
nftables: _nftables,
targets,
udp,
...rest
} = action;
return {
...rest,
targets: targets?.map((target) => serializeRouteTargetForRust(target)),
udp: serializeUdpForRust(udp),
};
}
export function serializeRouteForRust(route: IRouteConfig): IRustRouteConfig {
return {
...route,
match: serializeRouteMatchForRust(route.match),
action: serializeRouteActionForRust(route.action),
};
}
function serializeAcmeForRust(acme?: IAcmeOptions): IRustAcmeOptions | undefined {
if (!acme) {
return undefined;
}
return {
enabled: acme.enabled,
email: acme.email,
environment: acme.environment,
accountEmail: acme.accountEmail,
port: acme.port,
useProduction: acme.useProduction,
renewThresholdDays: acme.renewThresholdDays,
autoRenew: acme.autoRenew,
skipConfiguredCerts: acme.skipConfiguredCerts,
renewCheckIntervalHours: acme.renewCheckIntervalHours,
};
}
function serializeDefaultsForRust(defaults?: ISmartProxyOptions['defaults']): IRustDefaultConfig | undefined {
if (!defaults) {
return undefined;
}
const { preserveSourceIP, ...rest } = defaults;
return {
...rest,
preserveSourceIp: preserveSourceIP,
};
}
export function buildRustProxyOptions(
settings: ISmartProxyOptions,
routes: IRustRouteConfig[],
acmeOverride?: IAcmeOptions,
): IRustProxyOptions {
const acme = acmeOverride !== undefined ? acmeOverride : settings.acme;
return {
routes,
preserveSourceIp: settings.preserveSourceIP,
proxyIps: settings.proxyIPs,
acceptProxyProtocol: settings.acceptProxyProtocol,
sendProxyProtocol: settings.sendProxyProtocol,
defaults: serializeDefaultsForRust(settings.defaults),
connectionTimeout: settings.connectionTimeout,
initialDataTimeout: settings.initialDataTimeout,
socketTimeout: settings.socketTimeout,
inactivityCheckInterval: settings.inactivityCheckInterval,
maxConnectionLifetime: settings.maxConnectionLifetime,
inactivityTimeout: settings.inactivityTimeout,
gracefulShutdownTimeout: settings.gracefulShutdownTimeout,
noDelay: settings.noDelay,
keepAlive: settings.keepAlive,
keepAliveInitialDelay: settings.keepAliveInitialDelay,
maxPendingDataSize: settings.maxPendingDataSize,
disableInactivityCheck: settings.disableInactivityCheck,
enableKeepAliveProbes: settings.enableKeepAliveProbes,
enableDetailedLogging: settings.enableDetailedLogging,
enableTlsDebugLogging: settings.enableTlsDebugLogging,
enableRandomizedTimeouts: settings.enableRandomizedTimeouts,
maxConnectionsPerIp: settings.maxConnectionsPerIP,
connectionRateLimitPerMinute: settings.connectionRateLimitPerMinute,
keepAliveTreatment: settings.keepAliveTreatment,
keepAliveInactivityMultiplier: settings.keepAliveInactivityMultiplier,
extendedKeepAliveLifetime: settings.extendedKeepAliveLifetime,
metrics: settings.metrics,
acme: serializeAcmeForRust(acme),
};
}