i/Aether
1
0
Fork 0
mirror of https://github.com/fawney19/Aether.git synced 2026-09-02 01:10:23 +08:00
Code Issues Packages Projects Releases Wiki Activity

perf(transport): 全链路传输压缩优化

Browse Source 
- 上游请求启用 HTTP/2 (HPACK 头部压缩 + 多路复用)，添加 h2 依赖
- 上游请求体超过阈值时自动 gzip 压缩，使用紧凑 JSON 序列化
- 添加 brotli 依赖，Accept-Encoding 支持 gzip/deflate/br
- 隧道帧压缩: Rust 端响应帧和 Python 端请求/响应帧均支持 gzip
- Rust 端压缩/解压逻辑统一提取到 protocol.rs
- Rust 端请求头构建改用 .headers() 替换 reqwest 默认值
- 新增 ENABLE_HTTP2、ENABLE_REQUEST_COMPRESSION 等环境变量配置
This commit is contained in:
fawney19
2026-02-28 20:53:11 +08:00
parent 3ff67fec2f
commit a2d1cff3b0
10 changed files with 279 additions and 59 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
aether-proxy/src/tunnel/dispatcher.rs
View File

@@ -14,7 +14,7 @@ use tracing::{debug, error, info, warn};
use crate::state::{AppState, ServerContext};
use super::heartbeat::HeartbeatHandle;
use super::protocol::{Frame, MsgType, RequestMeta};
use super::protocol::{decompress_if_gzip, Frame, MsgType, RequestMeta};
use super::stream_handler;
use super::writer::FrameSender;
@@ -92,8 +92,15 @@ where
match frame.msg_type {
MsgType::RequestHeaders => {
// Parse request metadata
let meta: RequestMeta = match serde_json::from_slice(&frame.payload) {
// Decompress if the frame is gzip-compressed, then parse metadata
let payload = match decompress_if_gzip(&frame) {
Ok(p) => p,
Err(e) => {
warn!(stream_id = frame.stream_id, error = %e, "frame decompress failed");
continue;
}
};
let meta: RequestMeta = match serde_json::from_slice(&payload) {
Ok(m) => m,
Err(e) => {
warn!(stream_id = frame.stream_id, error = %e, "invalid request metadata");

50
aether-proxy/src/tunnel/protocol.rs
View File

@@ -159,3 +159,53 @@ pub struct ResponseMeta {
/// Header list preserving duplicates (e.g. multiple Set-Cookie).
pub headers: Vec<(String, String)>,
}
// ---------------------------------------------------------------------------
// Tunnel frame compression helpers
// ---------------------------------------------------------------------------
/// Minimum payload size to attempt gzip compression (bytes).
const COMPRESS_MIN_SIZE: usize = 512;
/// If the frame has the GZIP_COMPRESSED flag, decompress the payload; otherwise
/// return a clone of the raw payload bytes.
pub fn decompress_if_gzip(frame: &Frame) -> Result<Bytes, std::io::Error> {
if frame.is_gzip() {
decompress_gzip(&frame.payload)
} else {
Ok(frame.payload.clone())
}
}
/// Gzip-compress `data` if it is large enough and compression actually shrinks
/// the payload. Returns `(payload, extra_flags)` where `extra_flags` contains
/// `GZIP_COMPRESSED` when compression was applied.
pub fn compress_payload(data: Bytes) -> (Bytes, u8) {
if data.len() >= COMPRESS_MIN_SIZE {
if let Ok(compressed) = compress_gzip(&data) {
if compressed.len() < data.len() {
return (compressed, flags::GZIP_COMPRESSED);
}
}
}
(data, 0)
}
fn decompress_gzip(data: &[u8]) -> Result<Bytes, std::io::Error> {
use flate2::read::GzDecoder;
use std::io::Read;
let mut decoder = GzDecoder::new(data);
let mut buf = Vec::new();
decoder.read_to_end(&mut buf)?;
Ok(Bytes::from(buf))
}
fn compress_gzip(data: &[u8]) -> Result<Bytes, std::io::Error> {
use flate2::write::GzEncoder;
use flate2::Compression;
use std::io::Write;
let mut encoder = GzEncoder::new(Vec::new(), Compression::fast());
encoder.write_all(data)?;
let compressed = encoder.finish()?;
Ok(Bytes::from(compressed))
}

71
aether-proxy/src/tunnel/stream_handler.rs
View File

@@ -15,7 +15,9 @@ use tracing::{debug, warn};
use crate::state::{AppState, ServerContext};
use crate::target_filter;
use super::protocol::{flags, Frame, MsgType, RequestMeta, ResponseMeta};
use super::protocol::{
compress_payload, decompress_if_gzip, flags, Frame, MsgType, RequestMeta, ResponseMeta,
};
use super::writer::FrameSender;
/// Maximum response body chunk size per frame (32 KB).
@@ -95,21 +97,17 @@ async fn handle_stream_inner(
match body_rx.recv().await {
Some(frame) => {
if frame.msg_type == MsgType::RequestBody {
let payload = if frame.is_gzip() {
match decompress_gzip(&frame.payload) {
Ok(d) => d,
Err(e) => {
send_error(
frame_tx,
stream_id,
&format!("gzip decompress failed: {e}"),
)
.await;
return;
}
let payload = match decompress_if_gzip(&frame) {
Ok(d) => d,
Err(e) => {
send_error(
frame_tx,
stream_id,
&format!("gzip decompress failed: {e}"),
)
.await;
return;
}
} else {
frame.payload.clone()
};
if !payload.is_empty() {
body_parts.push(payload);
@@ -194,21 +192,23 @@ async fn handle_stream_inner(
let timeout = Duration::from_secs(meta.timeout.clamp(MIN_TIMEOUT_SECS, MAX_TIMEOUT_SECS));
let method: reqwest::Method = meta.method.parse().unwrap_or(reqwest::Method::GET);
let mut req = client.request(method, &meta.url);
// Build a complete HeaderMap from tunnel headers, then set it all at once
// via .headers() which *replaces* reqwest defaults (e.g. Accept: */*),
// ensuring upstream sees exactly what Aether server intended.
let mut header_map = reqwest::header::HeaderMap::with_capacity(meta.headers.len());
for (k, v) in &meta.headers {
let k_lower = k.to_ascii_lowercase();
// Skip hop-by-hop and security-sensitive headers
if BLOCKED_HEADERS.contains(&k_lower.as_str()) {
continue;
}
// Validate header name/value are valid HTTP
if let (Ok(name), Ok(value)) = (
reqwest::header::HeaderName::from_bytes(k.as_bytes()),
reqwest::header::HeaderValue::from_str(v),
) {
req = req.header(name, value);
header_map.insert(name, value);
}
}
let mut req = client.request(method, &meta.url).headers(header_map);
let body_size = body.len();
if !body.is_empty() {
req = req.body(body);
@@ -258,39 +258,51 @@ async fn handle_stream_inner(
status,
headers: resp_headers,
};
let meta_json = serde_json::to_vec(&resp_meta).unwrap_or_default();
let meta_json: Bytes = serde_json::to_vec(&resp_meta).unwrap_or_default().into();
let (meta_payload, meta_flags) = compress_payload(meta_json);
if !send_frame(
frame_tx,
Frame::new(stream_id, MsgType::ResponseHeaders, 0, meta_json),
Frame::new(
stream_id,
MsgType::ResponseHeaders,
meta_flags,
meta_payload,
),
)
.await
{
return;
}
// Stream response body
// Stream response body — relay upstream bytes through the tunnel.
// Apply tunnel-level frame compression for chunks that benefit from it
// (e.g. uncompressed SSE text). Already-compressed data (gzip/br from
// upstream Content-Encoding) won't shrink further and will be sent as-is
// thanks to the size check in compress_payload().
let mut stream = response.bytes_stream();
while let Some(chunk_result) = stream.next().await {
match chunk_result {
Ok(chunk) => {
if chunk.len() <= MAX_CHUNK_SIZE {
let (payload, extra_flags) = compress_payload(chunk);
if !send_frame(
frame_tx,
Frame::new(stream_id, MsgType::ResponseBody, 0, chunk),
Frame::new(stream_id, MsgType::ResponseBody, extra_flags, payload),
)
.await
{
return;
}
} else {
// Split oversized chunks
// Split oversized chunks, compress each slice
let mut offset = 0;
while offset < chunk.len() {
let end = (offset + MAX_CHUNK_SIZE).min(chunk.len());
let slice = chunk.slice(offset..end);
let (payload, extra_flags) = compress_payload(slice);
if !send_frame(
frame_tx,
Frame::new(stream_id, MsgType::ResponseBody, 0, slice),
Frame::new(stream_id, MsgType::ResponseBody, extra_flags, payload),
)
.await
{
@@ -337,12 +349,3 @@ async fn send_error(tx: &FrameSender, stream_id: u32, msg: &str) {
)
.await;
}
fn decompress_gzip(data: &[u8]) -> Result<Bytes, std::io::Error> {
use flate2::read::GzDecoder;
use std::io::Read;
let mut decoder = GzDecoder::new(data);
let mut buf = Vec::new();
decoder.read_to_end(&mut buf)?;
Ok(Bytes::from(buf))
}