上游模块与自定义转发

主线位置：请求主线的两头。上游模块和自定义转发在「upstream_response_filter / response_filter」那一段，处理从上游回来的每个 HttpTask；connection_filter 在最前面的「监听器 accept」，连 TLS 握手都还没开始。三样东西都是可选的，前两个藏在 feature 或 #[doc(hidden)] 后面，第三个要开 connection_filter feature。

flowchart TB
  up([上游响应任务]) --> adj[adjust_upstream_modules]
  adj --> um[上游模块]
  um --> comp[upstream_compression]
  comp --> urf["upstream_response_*"]
  urf --> cache[写缓存]
  cache --> rf["response_*"]
  rf --> dm[下游模块]
  dm --> down([写给客户端])

这张图是一个响应任务（头、一块体、尾）在 process_upstream_tasks 和 write_response_tasks 里经过的顺序，h2 上游（process_upstream_tasks_h2）和自定义协议上游一样。实验在 exp-pingora/proxy-lab（M2 Max、macOS 27、rustc 1.99.0，tokio 1.53.2；order_bug 同时跑 debug 和 release，其余 release）。

下游模块回顾

HttpModule（modules/http/mod.rs）是一组可选的过滤方法：request_header_filter、request_body_filter、response_header_filter、response_body_filter、response_trailer_filter、response_done_filter。HttpModuleBuilder 每个请求 init() 一个模块实例，order() 决定先后。HttpModules 在服务启动时装好，每个请求 build_ctx() 得到一个 HttpModuleCtx。下游模块在请求进来时看请求头和请求体，在写给客户端之前看响应；默认装的是关闭状态的响应压缩。详见 HTTP 模块与压缩。

上游模块：在解压之前

0.9.0 的 6ac51b3（「Add upstream module system」，说明是「This is analogous to the downstream modules but can apply prior to upstream compression」）加了第二组模块。要开 upstream_modules feature（pingora 门面 crate 的同名 feature 转发给 pingora-proxy），三处入口：

• init_upstream_modules：建服务时装模块，和 init_downstream_modules 一起在 handle_init_modules 里调用；
• Session::upstream_modules_ctx：每个请求一份；
• adjust_upstream_modules：上游响应头到达时、上游模块处理它之前调用，拿到的是 &ResponseHeader（只读），用来按响应配置模块（trait 文档举的例子是给字典压缩设字典）。1xx 响应头也会触发它，所以一个请求里可能调用多次。

驱动它的是 upstream_modules_filter_task：

    pub async fn upstream_modules_filter_task(&mut self, t: &mut HttpTask) -> Result<()> {
        match t {
            HttpTask::Header(header, eos) => {
                self.upstream_modules_ctx
                    .response_header_filter(header, *eos)
                    .await?;
            }
            HttpTask::Body(body, eos) | HttpTask::UpgradedBody(body, eos) => {
                self.upstream_modules_ctx.response_body_filter(body, *eos)?;
            }
            HttpTask::Trailer(trailers) => {
                if let Some(buf) = self
                    .upstream_modules_ctx
                    .response_trailer_filter(trailers)?
                {
                    *t = HttpTask::Body(Some(buf), true);
                }
            }
            HttpTask::Done => {
                if let Some(buf) = self.upstream_modules_ctx.response_done_filter()? {
                    *t = HttpTask::Body(Some(buf), true);
                }
            }
            HttpTask::Failed(_) => {}
        }
        Ok(())
    }

只有响应方向。upstream_modules_ctx 在整个 pingora-proxy 里只在这里被调用（git grep upstream_modules_ctx 0.9.0 -- pingora-proxy/src），模块的 request_header_filter、request_body_filter 装进上游模块后永远不会执行。

实验：上游模块看到的是压缩的字节

src/bin/modules.rs 给上下游各装一个打印用的 Probe 模块（同一个类型，两组 HttpModules 互不影响），在 request_filter 里 session.upstream_compression.adjust_decompression(true)，上游 tools/gzip_upstream.py 总是回 Content-Encoding: gzip。curl 不带 Accept-Encoding，所以 pingora 要替它解压：

DOWN request_header_filter /m
hook adjust_upstream_modules status=200 eos=false
UP response_header_filter status=200 content-encoding=Some("gzip")
hook upstream_response_filter content-encoding=None
hook response_filter content-encoding=None
UP response_body_filter len=54 first=Some("[1f, 8b, 08, 00]") end=true
DOWN response_header_filter status=200 content-encoding=None
DOWN response_body_filter len=1000 first=Some("[68, 65, 6c, 6c]") end=true

• 上游模块 UP 看到的头还带 Content-Encoding: gzip，看到的体是 54 字节、以 gzip 魔数 1f 8b 开头。
• upstream_response_filter 已经看不到 Content-Encoding 了：upstream_compression 排在上游模块之后、h1_response_filter 之前，解压时把这个头删了。这一点 phase.md 没写，名字里带「upstream」的钩子看到的并不是上游的原始响应。
• 下游模块 DOWN 看到的是解压后的 1000 字节（68 65 6c 6c 是 hell）。
• UP 没有 request_header_filter 那一行，DOWN 有。

curl 收到的头里多了一个 vary: accept-encoding，响应改成了 Transfer-Encoding: chunked（解压后长度变了，原来的 Content-Length 不能用）。

order() 取负的溢出

HttpModules::add_module 按 order() 从大到小排：

        self.modules.push(builder);
        // not the most efficient way but should be fine
        // largest order first
        self.modules.sort_by_key(|m| -m.order());

order() 返回 i16，-i16::MIN 溢出。src/bin/order_bug.rs 依次装 order 为 100、0、i16::MIN 的三个模块，再跑一次 response_header_filter：

== debug
added high(100), zero(0); adding min(i16::MIN)

thread 'main' (5419617) panicked at ~/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/pingora-core-0.9.0/src/modules/http/mod.rs:130:38:
attempt to negate with overflow
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
[exit 101]
== release
added high(100), zero(0); adding min(i16::MIN)
  runs: min (order -32768)
  runs: high (order 100)
  runs: zero (order 0)
[exit 0]

debug 下装模块时就 panic；release 下取负回绕成它自己，i16::MIN 被排到最前面，和「The lower the value, the later it runs」正好相反。想让一个模块最后跑，最自然的写法就是 i16::MIN。这行代码从 0.1.0 起就是这样，main 上（84f21ee）也没有改。上下游模块共用这段代码。

proxy_custom：自定义协议

proxy_custom.rs（1298 行）是 376f39b（「Add Custom Protocol Support to the Framework」，首次进入 0.7.0）加的，让下游、上游或两边都可以不是 HTTP，而是一种「HTTP 语义加自定义消息」的协议。相关的 trait 方法全都 #[doc(hidden)]，pingora-core 里的 connectors::http::custom::Connector 也是。本卷只读了源码，没有跑：0.9.0 里没有公开的自定义协议实现可以拿来连。

怎么进入。 上游侧：peer 的 options.alpn 是 ALPN::Custom(_) 时，连接器的 get_http_session 先找可复用的自定义会话，再找可复用的 h1 会话，最后让 custom::Connector 新建；协商结果不是自定义协议就退回 h1。下游侧：HttpServerOptions::force_custom，或者 TLS 协商出自定义 ALPN，apps/mod.rs 把连接交给 process_custom_session，也就是用户在 ProxyServiceBuilder::custom 里给的闭包。

怎么转发。 custom_proxy_down_to_up 的骨架和 h2 一样：一个 HttpTask 管道，请求体直接写；多出来的是两个方向的自定义消息。它把上游会话的消息读写端、下游会话的消息读写端都拿出来，各建一个注入通道（容量 CUSTOM_MESSAGE_QUEUE_SIZE），然后：

1. 调 custom_forwarding，把两个注入通道的发送端交给用户，用户可以自己往任一方向塞消息；
2. try_join! 四个 future：下游半边 custom_bidirection_down_to_up，上游响应的读取 custom_pipe_up_to_down_response，两个 CustomMessageForwarder（下游到上游、上游到下游各一个，从读端读、从注入通道收，写到另一边）；
3. 转发器不直接调钩子：它把读到的消息经一个通道交给下游半边，下游半边（手里有 session 和 ctx）调 downstream_custom_message_proxy_filter 或 upstream_custom_message_proxy_filter，结果经一个 oneshot 送回转发器，返回 None 就丢掉。用户经注入通道塞进来的消息不经过过滤器，直接写出。final_hop 参数表示另一侧是不是终点：在 h1、h2 上游的转发里，下游发来的自定义消息无处可去，所以 final_hop 是 true，过滤器是它的最后一站。

h1、h2 转发的下游半边里也有两个自定义消息分支（见 双向转发 那七个分支里的后两个），处理「下游是自定义协议、上游是普通 HTTP」的情况。0.9.0 里和自定义协议有关的提交不少，比如 7c04f54（「Restore custom message handles across upstream retries」）：重试时要把上一次拿出来的消息读写端还回去，否则第二次尝试就拿不到了，proxy_1to1 里「Keep the reader in this caller so it is restored even if retryable upstream errors make try_join! cancel the downstream future」那段注释说的就是它。

connection_filter：在 accept 循环里

ConnectionFilter 不是 pingora-proxy 的东西，而是 pingora-core 监听器的一部分，因为用户通常从代理服务上设置它，放在这一页讲。trait 只有一个方法（listeners/connection_filter.rs）：

#[async_trait]
pub trait ConnectionFilter: Debug + Send + Sync {
    // ...
    async fn should_accept(&self, _addr: Option<&SocketAddr>) -> bool {
        true
    }
}

文档注释里的两段示例写的签名是 should_accept(&self, addr: &SocketAddr)，和实际的 Option<&SocketAddr> 对不上，照抄编不过；模块注释里的依赖写法还是 pingora-core = { version = "0.5", ... }。没开 feature 时，Service::set_connection_filter 是一个什么都不做的空函数，Listeners 上也没有这个字段。

调用它的是 ListenerEndpoint::accept（listeners/l4.rs）：

            loop {
                let mut stream = self
                    .listener
                    .accept()
                    .await
                    .or_err(AcceptError, "Fail to accept()")?;
                // ...
                let should_accept = if let Some(digest) = stream.get_socket_digest() {
                    if let Some(peer_addr) = digest.peer_addr() {
                        self.connection_filter
                            .should_accept(peer_addr.as_inet())
                            .await
                    } else {
                        // No peer address available - accept by default
                        true
                    }
                // ...
                if !should_accept {
                    debug!("Connection rejected by filter");
                    drop(stream);
                    continue;
                }

这段在 accept 循环里 .await，而 accept 循环（services/listening.rs 的 run_endpoint）是每个监听地址一个、串行的：accept 一个，spawn 一个任务去握手，再 accept 下一个。过滤器 await 多久，这个端口就停止 accept 多久。被拒的连接直接 drop，此时 TCP 三次握手早已由内核完成，客户端看到的是连接被关闭。

BLOCK=1 时过滤器直接返回 false，两次运行里 curl 一次报 (56)（接收失败，连接被重置），一次报 (52) Empty reply from server：内核已经接受了连接、curl 已经发出了请求，pingora 关闭 socket 时请求还在接收缓冲里没被读，取决于时机，对端收到的是 RST 或 FIN。

不是 0.9.0 的 pre-TLS 回调

CHANGELOG 0.9.0 的「Add a pre-TLS callback for PROXY protocol support」说的是另一个东西：600c5c0（同步自 PR #799，首次进入 0.9.0）加的 PreTlsProcess，用 Listeners::set_pre_tls_callback 设置，不需要任何 feature。两者的区别：

PreTlsProcess 只对 TLS 端点生效，是那个 PR 有意为之：它的提交说明最后一条是「move pre-TLS callback inside TLS branch」。明文端口上的 PROXY protocol 头要在别处处理。这一项属于监听器，见 服务与监听。

动脑筋答案

curl3 code=200 connect=0.000920 total=1.009439
curl1 code=200 connect=0.000801 total=2.009040
curl2 code=200 connect=0.000832 total=3.012013
curl4 code=200 connect=0.000761 total=4.012979
[  986ms] should_accept Some(127.0.0.1:50329) begin
[ 1988ms] should_accept Some(127.0.0.1:50329) end
[ 1989ms] should_accept Some(127.0.0.1:50330) begin
[ 2991ms] should_accept Some(127.0.0.1:50330) end
[ 2991ms] should_accept Some(127.0.0.1:50331) begin
[ 3994ms] should_accept Some(127.0.0.1:50331) end
[ 3994ms] should_accept Some(127.0.0.1:50332) begin
[ 4996ms] should_accept Some(127.0.0.1:50332) end

• 四个 time_connect 都在 1 ms 以内：TCP 握手由内核在 listen 队列里完成，不经过 pingora。
• time_total 是 1、2、3、4 秒：过滤器一次只处理一个连接，上一个 end 之后下一个才 begin。第四个请求在过滤器里只待了 1 秒，却等了 4 秒。
• 按 curl 编号看，完成顺序是 3、1、2、4：过滤器按 accept 的先后处理，四个后台进程谁先连上不由 shell 里的启动顺序决定。
• 过滤器里要查远程服务（比如 IP 信誉），要么做成同步的本地查表，要么只在 accept 之后、按连接 spawn 的地方做（比如 PreTlsProcess，或者 request_filter）。

小结

• 上游模块（0.9.0，upstream_modules feature）在 upstream_compression 之前处理上游响应，能看到压缩的原始字节；只有响应方向，模块的请求过滤方法不会被调用。
• upstream_response_filter 排在解压之后，看到的已经不是上游的原始头；adjust_upstream_modules 只读，用来按响应配置模块。
• HttpModules 用 -m.order() 排序：order() 为 i16::MIN 时 debug panic、release 排到最前，main 上未修。
• proxy_custom 在 HTTP 转发之外再开两路自定义消息转发，入口是 ALPN::Custom 或 force_custom，相关 API 全部 #[doc(hidden)]。
• ConnectionFilter（0.7.0）在每个监听地址的 accept 循环里串行 await，慢过滤器会让整个端口排队；0.9.0 的 PreTlsProcess 是另一个东西，只对 TLS 端点、在握手任务里并发执行。