diff --git a/CHANGELOG.md b/CHANGELOG.md index d3f762a..6fb8102 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,19 @@ All notable changes to this project are documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and releases follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +## [0.1.1] - 2026-08-14 + +### Changed + +- Reduced authenticated XHTTP hot-path allocation, locking, address-resolution, + reference-count, and session teardown overhead using profile-guided changes. +- Cancelled orphan-session grace timers promptly and skipped them entirely for the + normal download-first request order, substantially reducing transient RSS growth. +- Added a sustained PID-scoped `perf` driver, focused allocation reference + microbenchmarks, and a bilingual hotspot optimization report. + ## [0.1.0] - 2026-08-13 ### Added @@ -26,3 +39,5 @@ follow [Semantic Versioning](https://semver.org/spec/v2.0.0.html). - Added bounded request/session/target controls and fail-closed memory accounting. [0.1.0]: https://github.com/jacek4yang/rust-xhttp/releases/tag/v0.1.0 +[0.1.1]: https://github.com/jacek4yang/rust-xhttp/compare/v0.1.0...v0.1.1 +[Unreleased]: https://github.com/jacek4yang/rust-xhttp/compare/v0.1.1...HEAD diff --git a/Cargo.lock b/Cargo.lock index 273ccc9..2e2a99b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1389,7 +1389,7 @@ dependencies = [ [[package]] name = "rust-xhttp" -version = "0.1.0" +version = "0.1.1" dependencies = [ "aes", "aes-gcm", diff --git a/Cargo.toml b/Cargo.toml index 4961d87..d27b692 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "rust-xhttp" -version = "0.1.0" +version = "0.1.1" edition = "2024" rust-version = "1.88" description = "Pure-Rust XHTTP/VLESS server wire-compatible with the official Xray-core client (XHTTP packet-up + VLESS + VLESS-Encryption + Vision + XUDP)" diff --git a/README.md b/README.md index e387f68..da5600b 100644 --- a/README.md +++ b/README.md @@ -139,4 +139,5 @@ welcome under [`CONTRIBUTING.md`](CONTRIBUTING.md). | Configuration and deployment | [English](docs/configuration.md) | [简体中文](docs/configuration.zh-CN.md) | | Benchmarks and evidence | [English](docs/benchmarks.md) | [简体中文](docs/benchmarks.zh-CN.md) | | Performance and availability | [English](docs/performance-and-availability.md) | [简体中文](docs/performance-and-availability.zh-CN.md) | +| Hotspot optimization report | [English](docs/performance-hotspots.md) | [简体中文](docs/performance-hotspots.zh-CN.md) | | Security policy | [English](SECURITY.md) | [简体中文](SECURITY.zh-CN.md) | diff --git a/README.zh-CN.md b/README.zh-CN.md index de96228..05fda11 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -102,6 +102,7 @@ VLESS 协议。不支持 stream-up/stream-one，也不声称“不可检测” | 配置与部署 | [English](docs/configuration.md) | [简体中文](docs/configuration.zh-CN.md) | | Benchmark 与证据 | [English](docs/benchmarks.md) | [简体中文](docs/benchmarks.zh-CN.md) | | 性能与可用性 | [English](docs/performance-and-availability.md) | [简体中文](docs/performance-and-availability.zh-CN.md) | +| 热点优化报告 | [English](docs/performance-hotspots.md) | [简体中文](docs/performance-hotspots.zh-CN.md) | | 安全政策 | [English](SECURITY.md) | [简体中文](SECURITY.zh-CN.md) | ## 许可证 diff --git a/benches/geo.rs b/benches/geo.rs index b23a3f2..0942799 100644 --- a/benches/geo.rs +++ b/benches/geo.rs @@ -1,24 +1,35 @@ use criterion::{Criterion, criterion_group, criterion_main}; use http::{HeaderMap, Method, Uri}; -use rust_xhttp::xhttp::{Meta, classify, extract_meta_from_path, host_matches, path_matches}; +use rust_xhttp::vless::{User, Validator, process_uuid}; +use rust_xhttp::xhttp::{ + BorrowedMeta, ResponsePadding, classify_borrowed, extract_meta_from_path_borrowed, + extract_padding, extract_padding_len, generate_response_padding, host_matches, + is_padding_len_valid, is_padding_valid, path_matches, +}; +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +use subtle::ConstantTimeEq; fn bench_xhttp_path_classification(c: &mut Criterion) { let uri: Uri = "/xhttp/session-123/184467440737095516".parse().unwrap(); c.bench_function("xhttp path meta classify", |b| { b.iter(|| { - let meta = extract_meta_from_path("/xhttp/", &uri); - classify(&Method::POST, &meta) + let meta = extract_meta_from_path_borrowed("/xhttp/", &uri); + classify_borrowed(&Method::POST, &meta) }) }); + c.bench_function("xhttp allocating path reference", |b| { + b.iter(|| allocating_path_reference("/xhttp/", &uri)) + }); } fn bench_xhttp_host_and_path(c: &mut Criterion) { let uri: Uri = "https://example.com/xhttp/session-123".parse().unwrap(); let mut headers = HeaderMap::new(); headers.insert(http::header::HOST, "example.com:443".parse().unwrap()); - let meta = Meta { - session_id: "session-123".into(), - seq_str: String::new(), + let meta = BorrowedMeta { + session_id: "session-123", + seq_str: "", }; c.bench_function("xhttp host path download classify", |b| { @@ -26,15 +37,82 @@ fn bench_xhttp_host_and_path(c: &mut Criterion) { ( path_matches("/xhttp/", &uri), host_matches("example.com", &headers, &uri), - classify(&Method::GET, &meta), + classify_borrowed(&Method::GET, &meta), ) }) }); } +fn bench_xhttp_padding(c: &mut Criterion) { + let uri: Uri = "/xhttp/session-123/0".parse().unwrap(); + let mut headers = HeaderMap::new(); + headers.insert( + http::header::REFERER, + format!("https://example.com/?x_padding={}", "X".repeat(100)) + .parse() + .unwrap(), + ); + c.bench_function("xhttp request padding validate", |b| { + b.iter(|| is_padding_len_valid(extract_padding_len(&headers, &uri), 100, 1000)) + }); + c.bench_function("xhttp allocating padding reference", |b| { + b.iter(|| { + let padding = extract_padding(&headers, &uri); + is_padding_valid(&padding, 100, 1000) + }) + }); + + let response_padding = ResponsePadding::new(100, 1000); + c.bench_function("xhttp cached response padding", |b| { + b.iter(|| response_padding.header_value()) + }); + c.bench_function("xhttp allocating response padding reference", |b| { + b.iter(|| { + let value = generate_response_padding(100, 1000); + value.parse::().unwrap() + }) + }); +} + +fn bench_vless_user_lookup(c: &mut Criterion) { + let id = [7u8; 16]; + let validator = Validator::new([User { + id, + email: "bench@example.com".into(), + flow: String::new(), + }]); + c.bench_function("vless lock-free user lookup", |b| { + b.iter(|| validator.get_shared(&id).unwrap()) + }); + + let mut users = HashMap::new(); + users.insert(process_uuid(id), validator.get(&id).unwrap()); + let locked = RwLock::new(Arc::new(users)); + c.bench_function("vless rwlock cloning lookup reference", |b| { + b.iter(|| { + let key = process_uuid(id); + let users = locked.read().unwrap().clone(); + let candidate = users.get(&key).unwrap(); + assert!(bool::from(candidate.id.ct_eq(&key))); + candidate.clone() + }) + }); +} + +fn allocating_path_reference(base: &str, uri: &Uri) -> (String, u64) { + let rest = uri.path().strip_prefix(base).unwrap_or(""); + let mut segments = rest.split('/'); + let session_id = segments.next().unwrap_or("").to_string(); + let seq_string = segments.next().unwrap_or("").to_string(); + let seq = seq_string.parse().unwrap_or_default(); + (session_id.clone(), seq) +} + criterion_group!( benches, bench_xhttp_path_classification, - bench_xhttp_host_and_path + bench_xhttp_host_and_path, + bench_xhttp_padding, + bench_vless_user_lookup ); criterion_main!(benches); diff --git a/docs/index.md b/docs/index.md index 9aea773..8ae0a54 100644 --- a/docs/index.md +++ b/docs/index.md @@ -14,6 +14,7 @@ VLESS-Encryption, verification, and troubleshooting. | Configuration and deployment | [English](configuration.md) | [简体中文](configuration.zh-CN.md) | | Benchmarks and raw evidence | [English](benchmarks.md) | [简体中文](benchmarks.zh-CN.md) | | Performance and availability | [English](performance-and-availability.md) | [简体中文](performance-and-availability.zh-CN.md) | +| Hotspot optimization report | [English](performance-hotspots.md) | [简体中文](performance-hotspots.zh-CN.md) | | Security policy | [English](../SECURITY.md) | [简体中文](../SECURITY.zh-CN.md) | ## Engineering notes diff --git a/docs/index.zh-CN.md b/docs/index.zh-CN.md index efd9dc0..3c962ef 100644 --- a/docs/index.zh-CN.md +++ b/docs/index.zh-CN.md @@ -13,6 +13,7 @@ ACME/手动证书、网站 fallback、VLESS-Encryption、验证和排错。 | 配置与部署 | [English](configuration.md) | [简体中文](configuration.zh-CN.md) | | Benchmark 与原始证据 | [English](benchmarks.md) | [简体中文](benchmarks.zh-CN.md) | | 性能与可用性 | [English](performance-and-availability.md) | [简体中文](performance-and-availability.zh-CN.md) | +| 热点优化报告 | [English](performance-hotspots.md) | [简体中文](performance-hotspots.zh-CN.md) | | 安全政策 | [English](../SECURITY.md) | [简体中文](../SECURITY.zh-CN.md) | ## 工程说明 diff --git a/docs/performance-and-availability.md b/docs/performance-and-availability.md index ae7fb1b..a45f4f9 100644 --- a/docs/performance-and-availability.md +++ b/docs/performance-and-availability.md @@ -5,6 +5,8 @@ This note explains the current hot path, resource model, failure behavior, and measurement limits. It complements the committed Xray comparison in [Benchmarks](benchmarks.md). +The sampling method, allocation changes, and current microbenchmark evidence are +recorded in the [hotspot optimization report](performance-hotspots.md). ## Hot-path design @@ -15,6 +17,12 @@ measurement limits. It complements the committed Xray comparison in precomputed. Conditional GETs return 304 without reading a file. - The session table is sharded, and counters are relaxed atomics. Target concurrency uses a semaphore rather than unbounded task creation. +- Request path/session metadata is borrowed from parsed HTTP values. Single-frame + body uploads remain reference-counted `Bytes`, response padding is lazily cached, + and user-table reads use `ArcSwap` rather than a read lock. +- Download-created sessions skip orphan grace timers entirely; timers created for + upload-first sessions are cancelled as soon as the download opens or the session + ends. This prevents completed sessions from retaining timer tasks for the full TTL. - Packet reorder queues and per-session/global byte budgets reserve capacity before accepting payload memory. Oversized work fails early. - TCP_NODELAY, keepalive, a 4096 listen backlog, and `SO_REUSEPORT` are enabled diff --git a/docs/performance-and-availability.zh-CN.md b/docs/performance-and-availability.zh-CN.md index f6b32bb..56fb759 100644 --- a/docs/performance-and-availability.zh-CN.md +++ b/docs/performance-and-availability.zh-CN.md @@ -4,6 +4,7 @@ 本文说明当前热路径、资源模型、故障行为和测量边界；Xray 对比原始证据见 [Benchmark](benchmarks.zh-CN.md)。 +采样方法、分配优化和当前微基准证据见[热点优化报告](performance-hotspots.zh-CN.md)。 ## 热路径设计 @@ -13,6 +14,10 @@ Last-Modified 和路由别名预先计算；条件 GET 无需读盘即可返回 304； - session table 分片，计数器使用 relaxed atomic；目标并发通过 semaphore 限制， 不会无限创建任务； +- request path/session 元数据直接借用已解析的 HTTP 值；单 frame body upload 保持为 + 引用计数 `Bytes`，响应 padding 延迟缓存，用户表读取通过 `ArcSwap` 避免读锁； +- download 先到达时完全不创建孤立 session grace timer；upload 先到达时创建的 timer + 会在 download 打开或 session 结束时取消，已完成 session 不再滞留整个 TTL； - packet 乱序队列和单 session/全局字节预算在接受 payload 内存前预留容量； - 默认启用 TCP_NODELAY、keepalive、4096 listen backlog，以及受支持 Linux 上的 `SO_REUSEPORT`； diff --git a/docs/performance-hotspots.md b/docs/performance-hotspots.md new file mode 100644 index 0000000..d13c329 --- /dev/null +++ b/docs/performance-hotspots.md @@ -0,0 +1,94 @@ +# Hotspot Optimization Report + +English | [简体中文](performance-hotspots.zh-CN.md) + +This report records the 2026-08-14 profile-guided optimization pass. It separates +repeatable function-level evidence from end-to-end measurements that are sensitive to +host load. + +## Reproduction + +Build a release binary with symbols, then attach `perf` only to the temporary +rust-xhttp child process: + +```bash +CARGO_PROFILE_RELEASE_DEBUG=1 \ +CARGO_PROFILE_RELEASE_STRIP=false \ +cargo build --release --locked + +DURATION=15 CONCURRENCY=64 PAYLOAD_BYTES=4096 scripts/profile.sh +``` + +The sustained driver verifies every VLESS/XHTTP echo response. `profile.sh` writes +ignored local artifacts under `docs/profile/`: workload JSON, `perf.data`, and a flat +top-symbol report. It uses `sudo -n perf record --pid ` because this +host's `perf_event_paranoid` setting blocks unprivileged attachment; it never samples +the entire host. + +Run the focused Criterion suite with: + +```bash +cargo bench --bench geo -- --noplot +``` + +## Observed hotspots + +The initial 15-second raw XHTTP sample attributed substantial aggregate cost to +allocation/free, Hyper/Tokio connection processing, session insertion/removal, response +padding construction, query parsing, and timer-wheel work. Network syscalls dominate +short HTTP/1.1 connections, so the application changes target repeated fixed costs rather +than claiming those syscalls can be removed. + +The pass made these changes: + +- XHTTP path metadata now borrows URI slices instead of allocating two strings and + cloning the session ID during classification. +- Padding validation counts decoded bytes without constructing the padding string. +- Response padding keeps Xray's random uniform length selection but lazily caches valid + `HeaderValue` instances for ordinary ranges. +- A single-frame Hyper upload body is forwarded as its existing `Bytes`; only fragmented + or mixed header/cookie/body placement needs concatenation. +- VLESS user snapshots use `ArcSwap`; the server hot-path lookup returns `Arc` without + taking an `RwLock` or cloning email/flow strings, while the original public owned lookup + remains compatible. +- IPv4/IPv6 targets connect directly as `SocketAddr`, avoiding address formatting and a + redundant resolver path. +- New sessions acquire their shard once rather than using a redundant double-checked + lock. The precomputed session hash is reused during download teardown. +- The normal download-first request order creates no grace task. Upload-first grace tasks + are aborted immediately after the download opens or the session is removed. +- Origin and production Dispatcher tasks share one outer `Arc`; connection/session + creation no longer clones every Arc-backed field separately. + +## Focused results + +On this four-core host, Criterion reported the following same-process comparisons. The +reference functions reproduce the replaced allocating/locking operations, which avoids +cross-run frequency and background-load bias. + +| Kernel | Optimized | Replaced reference | Change | +| --- | ---: | ---: | ---: | +| Path extraction + classification | 62.7 ns | 103.2 ns | -39% | +| Request padding extraction + validation | 149.3 ns | 383.7 ns | -61% | +| Random response padding HeaderValue | 23.0 ns | 118.2 ns | -81% | +| VLESS user lookup | 73.2 ns | 90.3 ns | -19% | + +An idle-window alternating A/B run made before the final outer-Arc reduction showed a +4.6% lower mean server CPU/op and an 81% reduction in workload-window RSS growth. The +Python driver was already the throughput bottleneck, so median throughput moved only +0.4%; this is not presented as a capacity result. The final macro rerun was rejected +because an unrelated release build began consuming the shared host during measurement. + +## Interpretation and next work + +The microbenchmarks support the local fixed-cost changes; they do not replace the +official Xray-client comparison in [Benchmarks](benchmarks.md). A publishable capacity +result still requires an otherwise idle pinned host, at least five repetitions, and the +same TLS/encryption/client mode for both candidates. + +The remaining flat sample is dominated by allocator, Hyper/Tokio polling, socket setup, +and kernel TCP work from deliberately short HTTP/1.1 sessions. The next useful pass +should profile a long-lived official Xray HTTP/2 client separately, then evaluate buffer +reuse only if allocation stacks remain material there. A pool should not be introduced +solely from this short-connection workload because pool contention can regress the real +H2 path. diff --git a/docs/performance-hotspots.zh-CN.md b/docs/performance-hotspots.zh-CN.md new file mode 100644 index 0000000..955ace0 --- /dev/null +++ b/docs/performance-hotspots.zh-CN.md @@ -0,0 +1,82 @@ +# 热点优化报告 + +[English](performance-hotspots.md) | 简体中文 + +本文记录 2026-08-14 的 profile-guided 优化，并把可重复的函数级证据与容易受主机 +负载影响的端到端数字分开。 + +## 复现方法 + +先构建带符号的 release binary，再让 `perf` 只附加到临时启动的 rust-xhttp 子进程： + +```bash +CARGO_PROFILE_RELEASE_DEBUG=1 \ +CARGO_PROFILE_RELEASE_STRIP=false \ +cargo build --release --locked + +DURATION=15 CONCURRENCY=64 PAYLOAD_BYTES=4096 scripts/profile.sh +``` + +持续负载器逐字节验证每个 VLESS/XHTTP echo 响应。`profile.sh` 在已忽略的 +`docs/profile/` 下写入 workload JSON、`perf.data` 和 flat top-symbol 报告。由于本机 +`perf_event_paranoid` 禁止普通用户 attach，它使用 +`sudo -n perf record --pid `；不会进行整机采样。 + +函数级 Criterion 测试： + +```bash +cargo bench --bench geo -- --noplot +``` + +## 观测到的热点 + +最初 15 秒 raw XHTTP 采样的聚合热点包括 allocation/free、Hyper/Tokio 连接处理、 +session 插入/删除、响应 padding 构造、query 解析与 timer wheel。短 HTTP/1.1 连接 +天然以网络 syscall 为主，因此本轮针对可重复固定成本，不声称能消除这些 syscall。 + +本轮改动： + +- XHTTP path 元数据直接借用 URI slice，不再分配两个 String，也不在分类时克隆 + session ID； +- padding 校验直接计算 decode 后字节长度，不构造 padding String； +- 响应 padding 保持 Xray 的均匀随机长度选择，普通范围的有效 `HeaderValue` 延迟 + 缓存； +- 单 frame Hyper upload body 沿用原有 `Bytes`；只有 fragmented 或混合 + header/cookie/body placement 才做拼接； +- VLESS 用户快照改用 `ArcSwap`，服务端热路径 lookup 返回 `Arc`，不再取得 + `RwLock` 并克隆 email/flow String；原有公开 owned lookup 接口保持兼容； +- IPv4/IPv6 target 直接以 `SocketAddr` connect，避免地址格式化和多余 resolver 路径； +- 新 session 只获取一次 shard lock，并在 download teardown 复用已计算的 session + hash； +- 常见的 download-first 顺序不创建 grace task；upload-first timer 会在 download + 打开或 session 删除时立即 abort； +- Origin 与生产 Dispatcher task 共享单个外层 `Arc`，创建连接/session 时不再逐个 + clone 内部所有 `Arc` 字段。 + +## 函数级结果 + +在这台四核主机上，Criterion 得到以下同进程对比。reference 函数复现被替换的 +allocation/locking 操作，因此基本排除了跨进程频率和后台负载偏差。 + +| 内核 | 优化后 | 被替换 reference | 变化 | +| --- | ---: | ---: | ---: | +| Path 提取与分类 | 62.7 ns | 103.2 ns | -39% | +| Request padding 提取与校验 | 149.3 ns | 383.7 ns | -61% | +| 随机响应 padding HeaderValue | 23.0 ns | 118.2 ns | -81% | +| VLESS 用户查找 | 73.2 ns | 90.3 ns | -19% | + +在最终外层 `Arc` 优化之前，一组空闲窗口交替 A/B 显示 mean server CPU/op 降低 +4.6%，workload 窗口 RSS 增长降低 81%。Python driver 已成为吞吐瓶颈，因此 median +throughput 只变化 0.4%，本文不把它当作容量结果。最终宏基准复测期间，共享主机上 +无关的 release build 开始占用 CPU，所以该组结果已作废。 + +## 解释与后续工作 + +微基准支持这些局部固定成本改造，但不能替代 [Benchmark](benchmarks.zh-CN.md) 中的 +官方 Xray 客户端对比。可发布的容量结论仍需要空闲并固定 CPU 的主机、至少五次 +重复，以及双方完全相同的 TLS/encryption/client 模式。 + +剩余 flat sample 主要是 allocator、Hyper/Tokio poll、socket 建立，以及短 HTTP/1.1 +连接产生的内核 TCP 工作。下一步应单独 profile 长连接官方 Xray HTTP/2 客户端；仅当 +其中 allocation stack 仍明显时再评估 buffer reuse。不能只凭当前短连接负载引入 +buffer pool，因为 pool contention 可能让真实 H2 路径退化。 diff --git a/scripts/check_docs.py b/scripts/check_docs.py index 36243af..931352b 100644 --- a/scripts/check_docs.py +++ b/scripts/check_docs.py @@ -21,6 +21,7 @@ "docs/performance-and-availability.md", "docs/performance-and-availability.zh-CN.md", ), + ("docs/performance-hotspots.md", "docs/performance-hotspots.zh-CN.md"), ) diff --git a/scripts/hotspot_profile.py b/scripts/hotspot_profile.py new file mode 100755 index 0000000..0944831 --- /dev/null +++ b/scripts/hotspot_profile.py @@ -0,0 +1,178 @@ +#!/usr/bin/env python3 +"""Drive a sustained rust-xhttp raw XHTTP workload and optionally sample it with perf. + +This intentionally reuses the protocol-correct workload from docker_xray_perf.py, but +starts only rust-xhttp. perf is attached to that exact child PID; it never samples the +whole host. +""" + +from __future__ import annotations + +import argparse +import concurrent.futures +import itertools +import json +import os +import subprocess +import tempfile +import threading +import time + +import docker_xray_perf as workload + + +def run_for_duration( + candidate: workload.Candidate, + echo_port: int, + payload: bytes, + duration: float, + concurrency: int, + timeout: float, +) -> tuple[int, list[float]]: + deadline = time.monotonic() + duration + operation_ids = itertools.count() + latencies: list[float] = [] + errors: list[str] = [] + lock = threading.Lock() + + def worker() -> None: + local_latencies: list[float] = [] + local_errors: list[str] = [] + while time.monotonic() < deadline: + op_id = next(operation_ids) + try: + local_latencies.append( + workload.roundtrip( + candidate.port, echo_port, payload, op_id, timeout + ) + ) + except Exception as error: # noqa: BLE001 - report workload failures + local_errors.append(str(error)) + break + with lock: + latencies.extend(local_latencies) + errors.extend(local_errors) + + with concurrent.futures.ThreadPoolExecutor(max_workers=concurrency) as pool: + futures = [pool.submit(worker) for _ in range(concurrency)] + for future in futures: + future.result() + + if errors: + raise RuntimeError("; ".join(errors[:5])) + return len(latencies), latencies + + +def main() -> None: + parser = argparse.ArgumentParser() + parser.add_argument("--rust-bin", default="target/release/rust-xhttp") + parser.add_argument("--duration", type=float, default=10.0) + parser.add_argument("--concurrency", type=int, default=64) + parser.add_argument("--payload-bytes", type=int, default=4096) + parser.add_argument("--warmup", type=int, default=256) + parser.add_argument("--timeout", type=float, default=15.0) + parser.add_argument("--perf-data", help="record perf samples to this file") + parser.add_argument("--perf-frequency", type=int, default=499) + args = parser.parse_args() + + if args.duration <= 0 or args.concurrency <= 0 or args.payload_bytes <= 0: + raise SystemExit("duration, concurrency, and payload-bytes must be positive") + rust_bin = os.path.abspath(args.rust_bin) + if not os.path.isfile(rust_bin): + raise SystemExit(f"missing rust-xhttp binary: {rust_bin}") + + payload = bytes((index % 251 for index in range(args.payload_bytes))) + echo = workload.EchoServer(args.payload_bytes).start() + perf: subprocess.Popen[str] | None = None + with tempfile.TemporaryDirectory(prefix="rxhttp-hotspot-") as tmp: + candidate = workload.start_rust(tmp, rust_bin, args.payload_bytes + 128) + try: + if args.warmup: + workload.run_load( + candidate, + echo.port, + payload, + args.warmup, + min(args.concurrency, args.warmup), + args.timeout, + 1_000_000, + ) + + if args.perf_data: + perf_data = os.path.abspath(args.perf_data) + os.makedirs(os.path.dirname(perf_data), exist_ok=True) + perf = subprocess.Popen( + [ + "sudo", + "-n", + "perf", + "record", + "--quiet", + "--freq", + str(args.perf_frequency), + "--call-graph", + "dwarf,16384", + "--pid", + str(candidate.process.pid), + "--output", + perf_data, + "--", + "sleep", + str(args.duration), + ], + text=True, + ) + # Give perf time to attach before the measured interval begins. + time.sleep(0.25) + + cpu_before = workload.proc_cpu_seconds(candidate.process.pid) + rss_before = workload.proc_rss_kib(candidate.process.pid) + wall_start = time.perf_counter() + completed, latencies = run_for_duration( + candidate, + echo.port, + payload, + args.duration, + args.concurrency, + args.timeout, + ) + wall = time.perf_counter() - wall_start + cpu_seconds = max( + 0.0, workload.proc_cpu_seconds(candidate.process.pid) - cpu_before + ) + rss_after = workload.proc_rss_kib(candidate.process.pid) + + if perf is not None and perf.wait(timeout=args.duration + 10) != 0: + raise RuntimeError("perf record failed") + + report = { + "operations": completed, + "concurrency": args.concurrency, + "payload_bytes": args.payload_bytes, + "wall_seconds": wall, + "ops_per_second": completed / wall, + "server_cpu_seconds": cpu_seconds, + "server_cpu_ms_per_op": cpu_seconds * 1000 / completed, + "latency_ms": { + "mean": sum(latencies) * 1000 / completed, + "p50": workload.percentile(latencies, 50) * 1000, + "p90": workload.percentile(latencies, 90) * 1000, + "p99": workload.percentile(latencies, 99) * 1000, + "max": max(latencies) * 1000, + }, + "rss_kib": {"before": rss_before, "after": rss_after}, + "perf_data": os.path.abspath(args.perf_data) + if args.perf_data + else None, + } + print(json.dumps(report, indent=2, sort_keys=True)) + finally: + if perf is not None and perf.poll() is None: + perf.terminate() + perf.wait(timeout=5) + workload.terminate(candidate.process) + echo.close() + + +if __name__ == "__main__": + main() diff --git a/scripts/profile.sh b/scripts/profile.sh index db9f382..e20ed90 100755 --- a/scripts/profile.sh +++ b/scripts/profile.sh @@ -1,18 +1,28 @@ #!/usr/bin/env bash -# Profile the local rust-xhttp E2E smoke with perf when available. +# Profile a sustained local rust-xhttp XHTTP workload with perf. set -euo pipefail ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" OUTDIR="${OUTDIR:-$ROOT/docs/profile}" -DURATION_NOTE="m7 local XHTTP E2E" mkdir -p "$OUTDIR" if command -v perf >/dev/null 2>&1; then - echo "recording perf for: $DURATION_NOTE" - perf record -g -o "$OUTDIR/perf.data" -- "$ROOT/scripts/m7_e2e.sh" - perf report -i "$OUTDIR/perf.data" --stdio 2>/dev/null | grep -vE '^\s*#' | head -40 > "$OUTDIR/perf-top.txt" || true + echo "recording perf for sustained local XHTTP traffic" + python3 "$ROOT/scripts/hotspot_profile.py" \ + --rust-bin "$ROOT/target/release/rust-xhttp" \ + --duration "${DURATION:-10}" \ + --concurrency "${CONCURRENCY:-64}" \ + --payload-bytes "${PAYLOAD_BYTES:-4096}" \ + --perf-data "$OUTDIR/perf.data" | tee "$OUTDIR/workload.json" + sudo -n perf report -i "$OUTDIR/perf.data" --stdio --no-children \ + --call-graph none --sort=overhead,symbol \ + 2>/dev/null | grep -vE '^\s*#' | head -60 > "$OUTDIR/perf-top.txt" || true echo "perf data: $OUTDIR/perf.data" echo "top symbols: $OUTDIR/perf-top.txt" else - echo "perf not found; running local E2E without sampling" - "$ROOT/scripts/m7_e2e.sh" + echo "perf not found; running sustained workload without sampling" + python3 "$ROOT/scripts/hotspot_profile.py" \ + --rust-bin "$ROOT/target/release/rust-xhttp" \ + --duration "${DURATION:-10}" \ + --concurrency "${CONCURRENCY:-64}" \ + --payload-bytes "${PAYLOAD_BYTES:-4096}" fi diff --git a/src/dispatcher.rs b/src/dispatcher.rs index 6b4453e..9e1f2e1 100644 --- a/src/dispatcher.rs +++ b/src/dispatcher.rs @@ -3,7 +3,7 @@ use crate::metrics::Metrics; use crate::session::{DownlinkSink, SessionConn, UplinkReader}; use crate::vless::{ - Addons, Command, Validator, XRV, decode_request_header, encode_response_header, + Addons, Command, Validator, XRV, decode_request_header_shared, encode_response_header, }; use std::collections::HashMap; use std::io; @@ -72,6 +72,22 @@ impl Dispatcher { pub fn spawn(&self, conn: SessionConn) { let this = self.clone(); + Self::spawn_owned(this, conn); + } + + /// Spawn from the shared production dispatcher with one Arc clone instead of cloning + /// every Arc-backed field for each session. + pub fn spawn_shared(self: &Arc, conn: SessionConn) { + let this = Arc::clone(self); + let id_hash = conn.id_hash; + tokio::spawn(async move { + if let Err(error) = this.serve(conn).await { + tracing::debug!(session = id_hash, %error, "session ended"); + } + }); + } + + fn spawn_owned(this: Self, conn: SessionConn) { let id_hash = conn.id_hash; tokio::spawn(async move { if let Err(error) = this.serve(conn).await { @@ -128,7 +144,7 @@ impl Dispatcher { }; let (user, header, addons) = match tokio::time::timeout( self.handshake_timeout, - decode_request_header(&mut reader, &self.validator), + decode_request_header_shared(&mut reader, &self.validator), ) .await { @@ -157,21 +173,15 @@ impl Dispatcher { match header.command { Command::Tcp => { let address = header.address.ok_or(DispatchError::MissingAddress)?; - self.serve_tcp( - reader, - writer, - address.connect_target(header.port), - vision_uuid, - ) - .await + self.serve_tcp(reader, writer, address, header.port, vision_uuid) + .await } Command::Udp => { if vision_uuid.is_some() { return Err(DispatchError::VisionUdpUnsupported); } let address = header.address.ok_or(DispatchError::MissingAddress)?; - self.serve_udp(reader, writer, address.connect_target(header.port)) - .await + self.serve_udp(reader, writer, address, header.port).await } Command::Mux if vision_uuid.is_none() => self.serve_xudp(reader, writer).await, Command::Mux => Err(DispatchError::VisionXudpUnavailable), @@ -183,7 +193,8 @@ impl Dispatcher { &self, mut reader: ClientReader, writer: ProtocolWriter, - target: String, + address: crate::vless::Address, + port: u16, vision_uuid: Option<[u8; 16]>, ) -> Result<(), DispatchError> { let _permit = self @@ -193,7 +204,9 @@ impl Dispatcher { .await .map_err(|_| DispatchError::ShuttingDown)?; let mut stream = - match tokio::time::timeout(self.connect_timeout, TcpStream::connect(target)).await { + match tokio::time::timeout(self.connect_timeout, connect_tcp_target(&address, port)) + .await + { Ok(Ok(stream)) => stream, Ok(Err(error)) => { self.metrics @@ -245,7 +258,8 @@ impl Dispatcher { &self, mut reader: ClientReader, writer: ProtocolWriter, - target: String, + address: crate::vless::Address, + port: u16, ) -> Result<(), DispatchError> { let _permit = self .targets @@ -253,11 +267,9 @@ impl Dispatcher { .acquire_owned() .await .map_err(|_| DispatchError::ShuttingDown)?; - let mut resolved = - tokio::time::timeout(self.connect_timeout, tokio::net::lookup_host(target)) - .await - .map_err(|_| DispatchError::ConnectTimeout)??; - let target = resolved.next().ok_or(DispatchError::MissingAddress)?; + let target = tokio::time::timeout(self.connect_timeout, resolve_address(&address, port)) + .await + .map_err(|_| DispatchError::ConnectTimeout)??; let bind = if target.is_ipv6() { "[::]:0" } else { @@ -518,8 +530,33 @@ async fn spawn_association( } async fn resolve_target(target: &crate::xudp::Target) -> Result { - let mut addresses = tokio::net::lookup_host(target.address.connect_target(target.port)).await?; - addresses.next().ok_or(DispatchError::MissingAddress) + resolve_address(&target.address, target.port).await +} + +async fn connect_tcp_target(address: &crate::vless::Address, port: u16) -> io::Result { + match address { + crate::vless::Address::Ipv4(ip) => { + TcpStream::connect(SocketAddr::new((*ip).into(), port)).await + } + crate::vless::Address::Ipv6(ip) => { + TcpStream::connect(SocketAddr::new((*ip).into(), port)).await + } + crate::vless::Address::Domain(domain) => TcpStream::connect((domain.as_str(), port)).await, + } +} + +async fn resolve_address( + address: &crate::vless::Address, + port: u16, +) -> Result { + match address { + crate::vless::Address::Ipv4(ip) => Ok(SocketAddr::new((*ip).into(), port)), + crate::vless::Address::Ipv6(ip) => Ok(SocketAddr::new((*ip).into(), port)), + crate::vless::Address::Domain(domain) => tokio::net::lookup_host((domain.as_str(), port)) + .await? + .next() + .ok_or(DispatchError::MissingAddress), + } } fn socket_target(address: SocketAddr) -> crate::xudp::Target { diff --git a/src/origin.rs b/src/origin.rs index 0fc1173..76d208f 100644 --- a/src/origin.rs +++ b/src/origin.rs @@ -4,10 +4,10 @@ use crate::config::{FallbackConfig, TlsConfig, UplinkDataPlacement, XhttpConfig} use crate::metrics::Metrics; use crate::session::{OpenDownload, PushResult, SessionTable}; use crate::site; -use crate::xhttp::{RequestKind, path_matches}; +use crate::xhttp::{BorrowedRequestKind, path_matches}; use crate::xhttp::{ - classify, extract_meta_from_path, extract_padding, generate_response_padding, host_matches, - is_padding_valid, + ResponsePadding, classify_borrowed, extract_meta_from_path_borrowed, extract_padding_len, + host_matches, is_padding_len_valid, }; use base64::Engine; use base64::engine::general_purpose::URL_SAFE_NO_PAD; @@ -33,6 +33,7 @@ pub struct Origin { sessions: Arc, metrics: Arc, site: Arc, + response_padding: Arc, tls: Option, tcp_nodelay: bool, tcp_keepalive: Option, @@ -52,11 +53,13 @@ impl Origin { ) -> Result { let tls = tls.map(crate::tls::Server::from_config).transpose()?; let site = Arc::new(site::StaticSite::from_config(fallback)?); + let response_padding = Arc::new(ResponsePadding::new(xhttp.padding_from, xhttp.padding_to)); Ok(Self { xhttp: Arc::new(xhttp), sessions, metrics, site, + response_padding, tls, tcp_nodelay, tcp_keepalive, @@ -80,6 +83,7 @@ impl Origin { } pub async fn serve(self, listener: TcpListener) -> Result<(), OriginError> { + let this = Arc::new(self); let mut shutdown = Box::pin(shutdown_signal()); let mut connections = JoinSet::new(); loop { @@ -103,7 +107,7 @@ impl Origin { } Err(error) => return Err(OriginError::Io(error)), }; - let this = self.clone(); + let this = this.clone(); connections.spawn(async move { if let Err(error) = this.serve_connection(stream).await { tracing::debug!(%error, "origin connection ended"); @@ -114,7 +118,7 @@ impl Origin { } drop(listener); - let drained = tokio::time::timeout(self.graceful_shutdown, async { + let drained = tokio::time::timeout(this.graceful_shutdown, async { while connections.join_next().await.is_some() {} }) .await; @@ -130,7 +134,7 @@ impl Origin { Ok(()) } - async fn serve_connection(&self, stream: TcpStream) -> Result<(), OriginError> { + async fn serve_connection(self: Arc, stream: TcpStream) -> Result<(), OriginError> { crate::net::tune_stream(&stream, self.tcp_nodelay, self.tcp_keepalive); let stream = match &self.tls { Some(tls) => tokio::time::timeout(self.handshake_timeout, tls.accept(stream)) @@ -141,16 +145,15 @@ impl Origin { self.serve_io(TokioIo::new(stream)).await } - async fn serve_io(&self, io: TokioIo) -> Result<(), OriginError> + async fn serve_io(self: Arc, io: TokioIo) -> Result<(), OriginError> where I: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin + Send + 'static, { - let this = self.clone(); hyper_util::server::conn::auto::Builder::new(TokioExecutor::new()) .serve_connection( io, service_fn(move |request| { - let this = this.clone(); + let this = self.clone(); async move { Ok::<_, Infallible>(this.handle(request).await) } }), ) @@ -187,19 +190,22 @@ impl Origin { self.add_xhttp_response_padding(&mut response); return response; } - let padding = extract_padding(request.headers(), request.uri()); - if !is_padding_valid(&padding, self.xhttp.padding_from, self.xhttp.padding_to) { + let padding_len = extract_padding_len(request.headers(), request.uri()); + if !is_padding_len_valid(padding_len, self.xhttp.padding_from, self.xhttp.padding_to) { return self.site_response(&request); } - let meta = extract_meta_from_path(&self.xhttp.path, request.uri()); - let mut response = match classify(request.method(), &meta) { - RequestKind::PacketUpload { session_id, seq } => { - self.upload(request, &session_id, seq).await + let (parts, body) = request.into_parts(); + let meta = extract_meta_from_path_borrowed(&self.xhttp.path, &parts.uri); + let mut response = match classify_borrowed(&parts.method, &meta) { + BorrowedRequestKind::PacketUpload { session_id, seq } => { + self.upload(&parts.headers, body, session_id, seq).await + } + BorrowedRequestKind::StreamDownload { session_id } => self.download(session_id), + BorrowedRequestKind::Unsupported => empty(StatusCode::INTERNAL_SERVER_ERROR), + BorrowedRequestKind::StreamUp { .. } | BorrowedRequestKind::StreamOne => { + empty(StatusCode::BAD_REQUEST) } - RequestKind::StreamDownload { session_id } => self.download(&session_id), - RequestKind::Unsupported => empty(StatusCode::INTERNAL_SERVER_ERROR), - RequestKind::StreamUp { .. } | RequestKind::StreamOne => empty(StatusCode::BAD_REQUEST), - RequestKind::Options => empty(StatusCode::OK), + BorrowedRequestKind::Options => empty(StatusCode::OK), }; self.add_xhttp_response_padding(&mut response); response @@ -207,18 +213,18 @@ impl Origin { async fn upload( &self, - request: Request, + headers: &http::HeaderMap, + mut body: Incoming, session_id: &str, seq: u64, ) -> Response { - let (parts, mut body) = request.into_parts(); let mut payload = BytesMut::new(); - match self.decode_header_payload(&parts.headers) { + match self.decode_header_payload(headers) { Ok(header_payload) => payload.extend_from_slice(&header_payload), Err(UploadPayloadError::InvalidBase64) => return empty(StatusCode::BAD_REQUEST), Err(UploadPayloadError::TooLarge) => return self.reject_upload_too_large(), } - match self.decode_cookie_payload(&parts.headers) { + match self.decode_cookie_payload(headers) { Ok(cookie_payload) => payload.extend_from_slice(&cookie_payload), Err(UploadPayloadError::InvalidBase64) => return empty(StatusCode::BAD_REQUEST), Err(UploadPayloadError::TooLarge) => return self.reject_upload_too_large(), @@ -236,19 +242,22 @@ impl Origin { Err(UploadPayloadError::InvalidBase64) => return empty(StatusCode::BAD_REQUEST), Err(UploadPayloadError::TooLarge) => return self.reject_upload_too_large(), }; + if payload.is_empty() { + return self.finish_upload(session_id, seq, body_payload).await; + } payload.extend_from_slice(&body_payload); } + self.finish_upload(session_id, seq, payload.freeze()).await + } + + async fn finish_upload(&self, session_id: &str, seq: u64, payload: Bytes) -> Response { if payload.len() > self.xhttp.max_each_post_bytes { return self.reject_upload_too_large(); } self.metrics .upload_bytes .fetch_add(payload.len() as u64, Ordering::Relaxed); - match self - .sessions - .push_uplink(session_id, seq, payload.freeze()) - .await - { + match self.sessions.push_uplink(session_id, seq, payload).await { Some(PushResult::Accepted | PushResult::Duplicate) => { let mut response = empty(StatusCode::OK); if self.xhttp.uplink_data_placement != UplinkDataPlacement::Body { @@ -278,23 +287,38 @@ impl Origin { } } - async fn read_body_payload(&self, body: &mut Incoming) -> Result { - let mut payload = BytesMut::new(); + async fn read_body_payload(&self, body: &mut Incoming) -> Result { + let mut first: Option = None; + let mut combined: Option = None; + let mut total = 0usize; while let Some(frame) = body.frame().await { let Ok(frame) = frame else { return Err(UploadPayloadError::InvalidBase64); }; if let Ok(data) = frame.into_data() { - let Some(total) = payload.len().checked_add(data.len()) else { + if data.is_empty() { + continue; + } + let Some(next_total) = total.checked_add(data.len()) else { return Err(UploadPayloadError::TooLarge); }; - if total > self.xhttp.max_each_post_bytes { + if next_total > self.xhttp.max_each_post_bytes { return Err(UploadPayloadError::TooLarge); } - payload.extend_from_slice(&data); + total = next_total; + if let Some(payload) = combined.as_mut() { + payload.extend_from_slice(&data); + } else if let Some(initial) = first.take() { + let mut payload = BytesMut::with_capacity(total); + payload.extend_from_slice(&initial); + payload.extend_from_slice(&data); + combined = Some(payload); + } else { + first = Some(data); + } } } - Ok(payload) + Ok(combined.map_or_else(|| first.unwrap_or_default(), BytesMut::freeze)) } fn decode_header_payload( @@ -357,15 +381,19 @@ impl Origin { } fn download(&self, session_id: &str) -> Response { - let reader = match self.sessions.open_download(session_id) { + let mut reader = match self.sessions.open_download(session_id) { OpenDownload::Opened(reader) => reader, OpenDownload::Conflict => return empty(StatusCode::CONFLICT), OpenDownload::Capacity => return empty(StatusCode::SERVICE_UNAVAILABLE), }; + let (session_id, id_hash) = reader + .take_session_key() + .expect("session table attaches a download cleanup key"); let state = DownloadState { reader, sessions: self.sessions.clone(), - session_id: session_id.to_string(), + session_id, + id_hash, }; let body = StreamBody::new(stream::unfold(state, |mut state| async move { state @@ -432,10 +460,10 @@ impl Origin { } fn add_xhttp_response_padding(&self, response: &mut Response) { - let padding = generate_response_padding(self.xhttp.padding_from, self.xhttp.padding_to); - response - .headers_mut() - .insert("x-padding", padding.parse().unwrap()); + response.headers_mut().insert( + http::header::HeaderName::from_static("x-padding"), + self.response_padding.header_value(), + ); } } @@ -525,12 +553,13 @@ fn cookie_value(headers: &http::HeaderMap, key: &str) -> Option { struct DownloadState { reader: crate::session::DownlinkReader, sessions: Arc, - session_id: String, + session_id: Arc, + id_hash: u64, } impl Drop for DownloadState { fn drop(&mut self) { - self.sessions.remove(&self.session_id); + self.sessions.remove_hashed(&self.session_id, self.id_hash); } } @@ -539,10 +568,12 @@ fn empty(status: StatusCode) -> Response { } fn response(status: StatusCode, body: Body) -> Response { - let mut response = Response::builder().status(status).body(body).unwrap(); - response - .headers_mut() - .insert(http::header::SERVER, "nginx".parse().unwrap()); + let mut response = Response::new(body); + *response.status_mut() = status; + response.headers_mut().insert( + http::header::SERVER, + http::HeaderValue::from_static("nginx"), + ); response } diff --git a/src/runtime.rs b/src/runtime.rs index 6babee9..9ad7e39 100644 --- a/src/runtime.rs +++ b/src/runtime.rs @@ -48,17 +48,19 @@ pub async fn serve(cfg: Arc) -> Result<(), Box> { )), }; - let dispatcher = Dispatcher::new( - validator, - metrics.clone(), - cfg.limits.max_concurrent_target_conns, - cfg.limits.target_connect(), - cfg.limits.udp_idle(), - ) - .with_handshake_timeout(cfg.limits.handshake_timeout()) - .with_tcp_tuning(cfg.listen.tcp_nodelay, cfg.listen.tcp_keepalive()) - .with_encryption(encryption); - let handler: Handler = Arc::new(move |conn| dispatcher.spawn(conn)); + let dispatcher = Arc::new( + Dispatcher::new( + validator, + metrics.clone(), + cfg.limits.max_concurrent_target_conns, + cfg.limits.target_connect(), + cfg.limits.udp_idle(), + ) + .with_handshake_timeout(cfg.limits.handshake_timeout()) + .with_tcp_tuning(cfg.listen.tcp_nodelay, cfg.listen.tcp_keepalive()) + .with_encryption(encryption), + ); + let handler: Handler = Arc::new(move |conn| dispatcher.spawn_shared(conn)); let sessions = SessionTable::new( SessionConfig { diff --git a/src/session/downlink.rs b/src/session/downlink.rs index eb6b9c1..2869d90 100644 --- a/src/session/downlink.rs +++ b/src/session/downlink.rs @@ -6,6 +6,7 @@ //! the bytes are the raw protocol stream (the SSE *content-type* is only a middlebox hint). use bytes::Bytes; +use std::sync::Arc; use tokio::sync::mpsc; /// Writer side, held by the dispatcher / VLESS writer. @@ -17,12 +18,19 @@ pub struct DownlinkSink { /// Reader side, taken once by the GET (download) handler. pub struct DownlinkReader { rx: mpsc::Receiver, + session_key: Option<(Arc, u64)>, } /// `capacity` is the number of in-flight chunks before the writer blocks. pub fn channel(capacity: usize) -> (DownlinkSink, DownlinkReader) { let (tx, rx) = mpsc::channel(capacity.max(1)); - (DownlinkSink { tx }, DownlinkReader { rx }) + ( + DownlinkSink { tx }, + DownlinkReader { + rx, + session_key: None, + }, + ) } impl DownlinkSink { @@ -41,6 +49,14 @@ impl DownlinkSink { } impl DownlinkReader { + pub(crate) fn set_session_key(&mut self, id: Arc, id_hash: u64) { + self.session_key = Some((id, id_hash)); + } + + pub(crate) fn take_session_key(&mut self) -> Option<(Arc, u64)> { + self.session_key.take() + } + /// Next chunk to flush, or None when the writer side is dropped (target EOF/close). pub async fn recv(&mut self) -> Option { self.rx.recv().await diff --git a/src/session/mod.rs b/src/session/mod.rs index 0d44e58..fc79449 100644 --- a/src/session/mod.rs +++ b/src/session/mod.rs @@ -35,9 +35,12 @@ pub struct SessionConn { pub type Handler = Arc; struct Session { + id: Arc, + id_hash: u64, uplink: UplinkSink, downlink_reader: Mutex>, fully_connected: AtomicBool, + grace_reaper: Mutex>, } pub struct SessionConfig { @@ -65,7 +68,7 @@ impl Default for SessionConfig { } struct Shard { - map: Mutex>>, + map: Mutex, Arc>>, } pub struct SessionTable { @@ -115,11 +118,6 @@ impl SessionTable { }) } - fn shard_for(&self, id: &str) -> &Shard { - let idx = (fnv1a(id.as_bytes()) as usize) % self.shards.len(); - &self.shards[idx] - } - /// Push an uplink packet for `session_id`, creating the session if needed. /// Returns the push result; `None` if the global session cap is hit. pub async fn push_uplink( @@ -128,7 +126,7 @@ impl SessionTable { seq: u64, payload: bytes::Bytes, ) -> Option { - let session = self.upsert(session_id)?; + let session = self.upsert(session_id, false)?; let r = session.uplink.push(seq, payload).await; // keep gauges roughly current (best-effort; exact accounting in reader) self.metrics @@ -141,22 +139,50 @@ impl SessionTable { /// (cancels grace reaping). Returns None if there is no such session or the GET already /// took it. pub fn open_download(self: &Arc, session_id: &str) -> OpenDownload { - let Some(session) = self.upsert(session_id) else { + let Some(session) = self.upsert(session_id, true) else { return OpenDownload::Capacity; }; session.fully_connected.store(true, Ordering::Release); + if let Some(reaper) = session.grace_reaper.lock().unwrap().take() { + reaper.abort(); + } let reader = session.downlink_reader.lock().unwrap().take(); match reader { - Some(reader) => OpenDownload::Opened(reader), + Some(mut reader) => { + reader.set_session_key(session.id.clone(), session.id_hash); + OpenDownload::Opened(reader) + } None => OpenDownload::Conflict, } } /// Remove a session (called when the download GET ends, or on tear-down). pub fn remove(self: &Arc, session_id: &str) { - let shard = self.shard_for(session_id); - let removed = shard.map.lock().unwrap().remove(session_id); + self.remove_inner(session_id, fnv1a(session_id.as_bytes()), None, true); + } + + fn remove_inner( + &self, + session_id: &str, + id_hash: u64, + expected: Option<&Arc>, + cancel_reaper: bool, + ) { + let shard = &self.shards[(id_hash as usize) % self.shards.len()]; + let removed = { + let mut map = shard.map.lock().unwrap(); + if expected.is_some_and(|expected| { + map.get(session_id) + .is_none_or(|current| !Arc::ptr_eq(current, expected)) + }) { + return; + } + map.remove(session_id) + }; if let Some(s) = removed { + if cancel_reaper && let Some(reaper) = s.grace_reaper.lock().unwrap().take() { + reaper.abort(); + } s.uplink.close(); self.active.fetch_sub(1, Ordering::AcqRel); crate::metrics::Metrics::add_gauge(&self.metrics.active_sessions, -1); @@ -167,16 +193,19 @@ impl SessionTable { self.active.load(Ordering::Relaxed) } - fn upsert(self: &Arc, session_id: &str) -> Option> { - // fast path - { - let map = self.shard_for(session_id).map.lock().unwrap(); - if let Some(s) = map.get(session_id) { - return Some(s.clone()); - } - } - // slow path - let shard = self.shard_for(session_id); + pub(crate) fn remove_hashed(self: &Arc, session_id: &str, id_hash: u64) { + self.remove_inner(session_id, id_hash, None, true); + } + + fn upsert( + self: &Arc, + session_id: &str, + fully_connected_on_create: bool, + ) -> Option> { + let id_hash = fnv1a(session_id.as_bytes()); + let shard = &self.shards[(id_hash as usize) % self.shards.len()]; + // A single lookup covers both existing and new sessions. Acquiring the same lock + // twice added pure overhead for every new session. let mut map = shard.map.lock().unwrap(); if let Some(s) = map.get(session_id) { return Some(s.clone()); @@ -195,12 +224,16 @@ impl SessionTable { self.cfg.global_buffer_budget.clone(), ); let (dl_sink, dl_reader) = downlink::channel(self.cfg.downlink_capacity); + let id: Arc = Arc::from(session_id); let session = Arc::new(Session { + id: id.clone(), + id_hash, uplink: sink, downlink_reader: Mutex::new(Some(dl_reader)), - fully_connected: AtomicBool::new(false), + fully_connected: AtomicBool::new(fully_connected_on_create), + grace_reaper: Mutex::new(None), }); - map.insert(session_id.to_string(), session.clone()); + map.insert(id.clone(), session.clone()); self.active.fetch_add(1, Ordering::AcqRel); crate::metrics::Metrics::add_gauge(&self.metrics.active_sessions, 1); drop(map); @@ -209,15 +242,20 @@ impl SessionTable { (self.handler)(SessionConn { reader, writer: dl_sink, - id_hash: fnv1a(session_id.as_bytes()), + id_hash: session.id_hash, }); - // grace reaper: if the GET never opens within `grace`, evict. + // A download-created session is already fully connected and never needs a grace + // timer. This is the normal Xray request order and avoids one task/timer per session. + if fully_connected_on_create { + return Some(session); + } + + // Upload-created sessions need a grace reaper until their GET opens. let table = self.clone(); - let id = session_id.to_string(); let weak = Arc::downgrade(&session); let grace = self.cfg.grace; - tokio::spawn(async move { + let task = tokio::spawn(async move { tokio::time::sleep(grace).await; if let Some(s) = weak.upgrade() && !s.fully_connected.load(Ordering::Acquire) @@ -226,10 +264,76 @@ impl SessionTable { .metrics .session_timeouts .fetch_add(1, Ordering::Relaxed); - table.remove(&id); + table.remove_inner(&id, s.id_hash, Some(&s), false); } }); + let reaper = task.abort_handle(); + *session.grace_reaper.lock().unwrap() = Some(reaper.clone()); + if session.fully_connected.load(Ordering::Acquire) + || shard + .map + .lock() + .unwrap() + .get(session_id) + .is_none_or(|current| !Arc::ptr_eq(current, &session)) + { + reaper.abort(); + session.grace_reaper.lock().unwrap().take(); + } Some(session) } } + +#[cfg(test)] +mod tests { + use super::*; + + fn table(grace: Duration) -> (Arc, Arc) { + let metrics = crate::metrics::Metrics::new(); + let handler: Handler = Arc::new(|_| {}); + let table = SessionTable::new( + SessionConfig { + grace, + ..SessionConfig::default() + }, + handler, + metrics.clone(), + ); + (table, metrics) + } + + #[tokio::test(start_paused = true)] + async fn grace_reaper_expires_unconnected_session() { + let (table, metrics) = table(Duration::from_secs(30)); + table + .push_uplink("expires", 0, bytes::Bytes::from_static(b"x")) + .await; + assert_eq!(table.active_sessions(), 1); + + tokio::task::yield_now().await; + tokio::time::advance(Duration::from_secs(31)).await; + tokio::task::yield_now().await; + assert_eq!(table.active_sessions(), 0); + assert_eq!(metrics.session_timeouts.load(Ordering::Relaxed), 1); + } + + #[tokio::test(start_paused = true)] + async fn opening_download_cancels_grace_reaper() { + let (table, metrics) = table(Duration::from_secs(30)); + table + .push_uplink("connected", 0, bytes::Bytes::from_static(b"x")) + .await; + assert!(matches!( + table.open_download("connected"), + OpenDownload::Opened(_) + )); + + tokio::time::advance(Duration::from_secs(31)).await; + tokio::task::yield_now().await; + assert_eq!(table.active_sessions(), 1); + assert_eq!(metrics.session_timeouts.load(Ordering::Relaxed), 0); + table.remove("connected"); + assert_eq!(table.active_sessions(), 0); + } +} diff --git a/src/vless/header.rs b/src/vless/header.rs index 820347d..1086685 100644 --- a/src/vless/header.rs +++ b/src/vless/header.rs @@ -7,8 +7,9 @@ //! `version(1) | addonsLen(1) | addons(body)` — the server sends empty addons (`00 00`). use super::addons::{Addons, AddonsError, decode_addons_body, encode_addons}; -use super::address::{AddrError, Address, parse_port_then_address}; +use super::address::{AddrError, Address}; use super::validator::{User, Validator}; +use std::sync::Arc; use tokio::io::{AsyncRead, AsyncReadExt}; pub const VERSION: u8 = 0; @@ -70,6 +71,15 @@ pub async fn decode_request_header( reader: &mut R, validator: &Validator, ) -> Result<(User, RequestHeader, Addons), HeaderError> { + let (user, header, addons) = decode_request_header_shared(reader, validator).await?; + Ok((user.as_ref().clone(), header, addons)) +} + +/// Hot-path decoder that shares immutable user metadata instead of cloning its strings. +pub async fn decode_request_header_shared( + reader: &mut R, + validator: &Validator, +) -> Result<(Arc, RequestHeader, Addons), HeaderError> { let version = reader.read_u8().await?; if version != VERSION { return Err(HeaderError::BadVersion(version)); @@ -77,7 +87,9 @@ pub async fn decode_request_header( let mut raw_id = [0u8; 16]; reader.read_exact(&mut raw_id).await?; - let user = validator.get(&raw_id).ok_or(HeaderError::AuthFailed)?; + let user = validator + .get_shared(&raw_id) + .ok_or(HeaderError::AuthFailed)?; let addons_len = reader.read_u8().await? as usize; let addons = if addons_len != 0 { @@ -123,30 +135,32 @@ async fn read_address_port( // port(2) + type(1) let mut head = [0u8; 3]; reader.read_exact(&mut head).await?; - let extra = match head[2] { - super::address::ADDR_TYPE_IPV4 => 4, - super::address::ADDR_TYPE_IPV6 => 16, + let port = u16::from_be_bytes([head[0], head[1]]); + let address = match head[2] { + super::address::ADDR_TYPE_IPV4 => { + let mut octets = [0u8; 4]; + reader.read_exact(&mut octets).await?; + Address::Ipv4(octets.into()) + } + super::address::ADDR_TYPE_IPV6 => { + let mut octets = [0u8; 16]; + reader.read_exact(&mut octets).await?; + Address::Ipv6(octets.into()) + } super::address::ADDR_TYPE_DOMAIN => { let dlen = reader.read_u8().await? as usize; - // re-buffer: we already consumed the domain length byte; read domain then assemble + if dlen == 0 { + return Err(HeaderError::Address(AddrError::EmptyDomain)); + } let mut domain = vec![0u8; dlen]; reader.read_exact(&mut domain).await?; - let mut full = Vec::with_capacity(3 + 1 + dlen); - full.extend_from_slice(&head); - full.push(dlen as u8); - full.extend_from_slice(&domain); - let (port, addr, _n) = parse_port_then_address(&full)?; - return Ok((port, addr)); + let domain = String::from_utf8(domain) + .map_err(|_| HeaderError::Address(AddrError::BadDomainUtf8))?; + Address::Domain(domain) } other => return Err(HeaderError::Address(AddrError::UnknownType(other))), }; - let mut rest = vec![0u8; extra]; - reader.read_exact(&mut rest).await?; - let mut full = Vec::with_capacity(3 + extra); - full.extend_from_slice(&head); - full.extend_from_slice(&rest); - let (port, addr, _n) = parse_port_then_address(&full)?; - Ok((port, addr)) + Ok((port, address)) } /// Encode the VLESS response header: `version | addons`. The server always sends empty addons. diff --git a/src/vless/mod.rs b/src/vless/mod.rs index edfae0a..c9c089a 100644 --- a/src/vless/mod.rs +++ b/src/vless/mod.rs @@ -16,6 +16,7 @@ pub mod vision; pub use addons::{Addons, XRV}; pub use address::Address; pub use header::{ - Command, HeaderError, RequestHeader, decode_request_header, encode_response_header, + Command, HeaderError, RequestHeader, decode_request_header, decode_request_header_shared, + encode_response_header, }; pub use validator::{User, Validator, process_uuid}; diff --git a/src/vless/validator.rs b/src/vless/validator.rs index cdf6ac2..483cc53 100644 --- a/src/vless/validator.rs +++ b/src/vless/validator.rs @@ -7,8 +7,9 @@ //! processed id to avoid leaking near-miss timing. The table is swapped atomically on reload, //! so a config reload never tears down in-flight sessions and never takes a data-path lock. +use arc_swap::ArcSwap; use std::collections::HashMap; -use std::sync::{Arc, RwLock}; +use std::sync::Arc; use subtle::ConstantTimeEq; #[derive(Debug, Clone)] @@ -19,12 +20,12 @@ pub struct User { } struct Inner { - by_id: HashMap<[u8; 16], User>, + by_id: HashMap<[u8; 16], Arc>, } #[derive(Clone)] pub struct Validator { - inner: Arc>>, + inner: Arc>, } /// Zero bytes [6] and [7] (Xray `ProcessUUID`). @@ -39,10 +40,10 @@ impl Validator { let mut by_id = HashMap::new(); for mut u in users { u.id = process_uuid(u.id); - by_id.insert(u.id, u); + by_id.insert(u.id, Arc::new(u)); } Self { - inner: Arc::new(RwLock::new(Arc::new(Inner { by_id }))), + inner: Arc::new(ArcSwap::from_pointee(Inner { by_id })), } } @@ -51,26 +52,32 @@ impl Validator { let mut by_id = HashMap::new(); for mut u in users { u.id = process_uuid(u.id); - by_id.insert(u.id, u); + by_id.insert(u.id, Arc::new(u)); } - *self.inner.write().unwrap() = Arc::new(Inner { by_id }); + self.inner.store(Arc::new(Inner { by_id })); } - /// Look up a user by the *raw* request UUID. Returns the matched user or None. + /// Look up a user by the *raw* request UUID. Retained as an owned result for API + /// compatibility; the server hot path uses [`Self::get_shared`]. pub fn get(&self, raw_id: &[u8; 16]) -> Option { + self.get_shared(raw_id).map(|user| user.as_ref().clone()) + } + + /// Lock-free lookup that shares immutable user metadata without cloning strings. + pub fn get_shared(&self, raw_id: &[u8; 16]) -> Option> { let key = process_uuid(*raw_id); - let snap = self.inner.read().unwrap().clone(); + let snap = self.inner.load(); let candidate = snap.by_id.get(&key)?; // constant-time confirm on the processed id if candidate.id.ct_eq(&key).into() { - Some(candidate.clone()) + Some(Arc::clone(candidate)) } else { None } } pub fn len(&self) -> usize { - self.inner.read().unwrap().by_id.len() + self.inner.load().by_id.len() } pub fn is_empty(&self) -> bool { diff --git a/src/xhttp/mod.rs b/src/xhttp/mod.rs index bbfb5cd..b2eed00 100644 --- a/src/xhttp/mod.rs +++ b/src/xhttp/mod.rs @@ -12,7 +12,10 @@ use http::{HeaderMap, Method, Uri}; mod padding; -pub use padding::{extract_padding, generate_response_padding, is_padding_valid}; +pub use padding::{ + ResponsePadding, extract_padding, extract_padding_len, generate_response_padding, + is_padding_len_valid, is_padding_valid, +}; /// What the server should do with a request, after host/path/padding validation. #[derive(Debug, Clone, PartialEq, Eq)] @@ -41,12 +44,27 @@ pub struct Meta { /// Extract session id and seq from the path, given the normalized base `path` (ends with '/'). /// Default = both in the path (`///`). pub fn extract_meta_from_path(base: &str, uri: &Uri) -> Meta { + let meta = extract_meta_from_path_borrowed(base, uri); + Meta { + session_id: meta.session_id.to_owned(), + seq_str: meta.seq_str.to_owned(), + } +} + +/// Borrowed hot-path metadata. This avoids allocating session and sequence strings while +/// retaining the original owned [`Meta`] API for library callers. +pub struct BorrowedMeta<'a> { + pub session_id: &'a str, + pub seq_str: &'a str, +} + +pub fn extract_meta_from_path_borrowed<'a>(base: &str, uri: &'a Uri) -> BorrowedMeta<'a> { let full = uri.path(); let rest = full.strip_prefix(base).unwrap_or(""); let mut segs = rest.split('/'); - let session_id = segs.next().unwrap_or("").to_string(); - let seq_str = segs.next().unwrap_or("").to_string(); - Meta { + let session_id = segs.next().unwrap_or(""); + let seq_str = segs.next().unwrap_or(""); + BorrowedMeta { session_id, seq_str, } @@ -79,8 +97,42 @@ pub fn host_matches(configured: &str, headers: &HeaderMap, uri: &Uri) -> bool { /// * GET → uplink iff seq present; otherwise download (with session) or stream-one (no session) /// * other methods → uplink request pub fn classify(method: &Method, meta: &Meta) -> RequestKind { + match classify_borrowed( + method, + &BorrowedMeta { + session_id: &meta.session_id, + seq_str: &meta.seq_str, + }, + ) { + BorrowedRequestKind::PacketUpload { session_id, seq } => RequestKind::PacketUpload { + session_id: session_id.to_owned(), + seq, + }, + BorrowedRequestKind::StreamDownload { session_id } => RequestKind::StreamDownload { + session_id: session_id.to_owned(), + }, + BorrowedRequestKind::StreamUp { session_id } => RequestKind::StreamUp { + session_id: session_id.to_owned(), + }, + BorrowedRequestKind::StreamOne => RequestKind::StreamOne, + BorrowedRequestKind::Options => RequestKind::Options, + BorrowedRequestKind::Unsupported => RequestKind::Unsupported, + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum BorrowedRequestKind<'a> { + PacketUpload { session_id: &'a str, seq: u64 }, + StreamDownload { session_id: &'a str }, + StreamUp { session_id: &'a str }, + StreamOne, + Options, + Unsupported, +} + +pub fn classify_borrowed<'a>(method: &Method, meta: &BorrowedMeta<'a>) -> BorrowedRequestKind<'a> { if method == Method::OPTIONS { - return RequestKind::Options; + return BorrowedRequestKind::Options; } let has_session = !meta.session_id.is_empty(); let has_seq = !meta.seq_str.is_empty(); @@ -90,28 +142,28 @@ pub fn classify(method: &Method, meta: &Meta) -> RequestKind { if is_uplink && has_session { if has_seq { match meta.seq_str.parse::() { - Ok(seq) => RequestKind::PacketUpload { - session_id: meta.session_id.clone(), + Ok(seq) => BorrowedRequestKind::PacketUpload { + session_id: meta.session_id, seq, }, // Go returns 500 on ParseUint failure; surface as Unsupported→ caller maps to 500. - Err(_) => RequestKind::Unsupported, + Err(_) => BorrowedRequestKind::Unsupported, } } else { - RequestKind::StreamUp { - session_id: meta.session_id.clone(), + BorrowedRequestKind::StreamUp { + session_id: meta.session_id, } } } else if method == Method::GET || !has_session { if has_session { - RequestKind::StreamDownload { - session_id: meta.session_id.clone(), + BorrowedRequestKind::StreamDownload { + session_id: meta.session_id, } } else { - RequestKind::StreamOne + BorrowedRequestKind::StreamOne } } else { - RequestKind::Unsupported + BorrowedRequestKind::Unsupported } } @@ -125,14 +177,16 @@ mod tests { #[test] fn meta_from_path_default() { - let m = extract_meta_from_path("/yourpath/", &uri("/yourpath/SESSION123/7")); + let target = uri("/yourpath/SESSION123/7"); + let m = extract_meta_from_path("/yourpath/", &target); assert_eq!(m.session_id, "SESSION123"); assert_eq!(m.seq_str, "7"); } #[test] fn meta_download_no_seq() { - let m = extract_meta_from_path("/yourpath/", &uri("/yourpath/SESSION123")); + let target = uri("/yourpath/SESSION123"); + let m = extract_meta_from_path("/yourpath/", &target); assert_eq!(m.session_id, "SESSION123"); assert_eq!(m.seq_str, ""); } diff --git a/src/xhttp/padding.rs b/src/xhttp/padding.rs index 11beee9..5c7c42b 100644 --- a/src/xhttp/padding.rs +++ b/src/xhttp/padding.rs @@ -8,8 +8,44 @@ //! Validation: empty → invalid (Go returns 400). Default method compares the raw character //! length against the configured `[from, to]` byte range (default 100..=1000). -use http::{HeaderMap, Uri}; +use bytes::Bytes; +use http::{HeaderMap, HeaderValue, Uri}; use rand::Rng; +use std::sync::OnceLock; + +const MAX_CACHED_LENGTHS: usize = 2048; + +/// Lazily cached response-padding header values. +/// +/// The default XHTTP range has 901 possible lengths. Caching each value after its first +/// use removes both the padding allocation and HeaderValue parsing from the request path, +/// while retaining Xray's uniform random length distribution. Very wide custom ranges use +/// a one-allocation fallback to keep startup and resident memory bounded. +pub struct ResponsePadding { + from: u32, + to: u32, + cache: Option]>>, +} + +impl ResponsePadding { + pub fn new(from: u32, to: u32) -> Self { + let width = to.saturating_sub(from) as usize + 1; + let cache = + (width <= MAX_CACHED_LENGTHS).then(|| (0..width).map(|_| OnceLock::new()).collect()); + Self { from, to, cache } + } + + #[inline] + pub fn header_value(&self) -> HeaderValue { + let len = random_length(self.from, self.to); + if let Some(cache) = &self.cache { + return cache[(len - self.from) as usize] + .get_or_init(|| padding_header_value(len)) + .clone(); + } + padding_header_value(len) + } +} /// Pull the padding value the client sent, mirroring the non-obfs branch. /// Returns the value (possibly empty) — the caller validates length. @@ -28,6 +64,21 @@ pub fn extract_padding(headers: &HeaderMap, uri: &Uri) -> String { String::new() } +/// Return the percent-decoded padding byte length without allocating the padding value. +/// The request path only needs this length for validation. +pub fn extract_padding_len(headers: &HeaderMap, uri: &Uri) -> Option { + if let Some(referer) = headers + .get(http::header::REFERER) + .and_then(|value| value.to_str().ok()) + && !referer.is_empty() + { + return query_value_ref(referer, "x_padding").map(percent_decoded_len); + } + uri.query() + .and_then(|query| query_param_ref(query, "x_padding")) + .map(percent_decoded_len) +} + /// Default-method validity: non-empty and raw length within `[from, to]`. pub fn is_padding_valid(value: &str, from: u32, to: u32) -> bool { if value.is_empty() { @@ -37,36 +88,87 @@ pub fn is_padding_valid(value: &str, from: u32, to: u32) -> bool { n >= from && n <= to } +#[inline] +pub fn is_padding_len_valid(value: Option, from: u32, to: u32) -> bool { + value.is_some_and(|len| len != 0 && len >= from as usize && len <= to as usize) +} + /// Generate default response padding, mirroring Xray's non-obfs `X-Padding` /// response header placement and repeat-x padding method. pub fn generate_response_padding(from: u32, to: u32) -> String { - let len = if from >= to { + let len = random_length(from, to); + "X".repeat(len as usize) +} + +#[inline] +fn random_length(from: u32, to: u32) -> u32 { + if from >= to { from } else { rand::thread_rng().gen_range(from..=to) - }; - "X".repeat(len as usize) + } +} + +fn padding_header_value(len: u32) -> HeaderValue { + let bytes = Bytes::from(vec![b'X'; len as usize]); + HeaderValue::from_maybe_shared(bytes).expect("X padding is always a valid header value") } /// Parse a full URL string and return the value of query `key`. fn query_value(url: &str, key: &str) -> Option { - let q = url.split_once('?').map(|(_, q)| q)?; - // strip a possible fragment - let q = q.split('#').next().unwrap_or(q); - query_param(q, key) + query_value_ref(url, key).map(percent_decode) } /// Find `key` in a raw `a=b&c=d` query string, with minimal percent-decoding. fn query_param(query: &str, key: &str) -> Option { + query_param_ref(query, key).map(percent_decode) +} + +fn query_value_ref<'a>(url: &'a str, key: &str) -> Option<&'a str> { + let query = url.split_once('?').map(|(_, query)| query)?; + let query = query.split('#').next().unwrap_or(query); + query_param_ref(query, key) +} + +fn query_param_ref<'a>(query: &'a str, key: &str) -> Option<&'a str> { for pair in query.split('&') { let (k, v) = pair.split_once('=').unwrap_or((pair, "")); if k == key { - return Some(percent_decode(v)); + return Some(v); } } None } +fn percent_decoded_len(value: &str) -> usize { + let bytes = value.as_bytes(); + let mut decoded_len = 0; + let mut index = 0; + while index < bytes.len() { + if bytes[index] == b'%' + && index + 2 < bytes.len() + && hex_value(bytes[index + 1]).is_some() + && hex_value(bytes[index + 2]).is_some() + { + index += 3; + } else { + index += 1; + } + decoded_len += 1; + } + decoded_len +} + +#[inline] +fn hex_value(value: u8) -> Option { + match value { + b'0'..=b'9' => Some(value - b'0'), + b'a'..=b'f' => Some(value - b'a' + 10), + b'A'..=b'F' => Some(value - b'A' + 10), + _ => None, + } +} + /// Minimal percent-decode (enough for padding values, which are X/Z/base62 + maybe '%'). fn percent_decode(s: &str) -> String { let bytes = s.as_bytes(); @@ -119,6 +221,7 @@ mod tests { assert_eq!(v, "XXXXXXXXXX"); assert!(is_padding_valid(&v, 5, 20)); assert!(!is_padding_valid(&v, 50, 100)); + assert_eq!(extract_padding_len(&h, &uri("/p/s/1")), Some(10)); } #[test] @@ -142,4 +245,24 @@ mod tests { } assert_eq!(generate_response_padding(5, 5), "XXXXX"); } + + #[test] + fn padding_length_decodes_without_materializing_value() { + let h = HeaderMap::new(); + let target = uri("/p/s/1?before=x&x_padding=X%58+Z&after=y"); + assert_eq!(extract_padding_len(&h, &target), Some(4)); + assert!(is_padding_len_valid(Some(4), 4, 4)); + assert!(!is_padding_len_valid(None, 1, 4)); + assert!(!is_padding_len_valid(Some(0), 0, 4)); + } + + #[test] + fn response_padding_cache_preserves_range() { + let padding = ResponsePadding::new(4, 8); + for _ in 0..64 { + let value = padding.header_value(); + assert!((4..=8).contains(&value.as_bytes().len())); + assert!(value.as_bytes().iter().all(|byte| *byte == b'X')); + } + } }