双向转发

主线位置：请求主线的「upstream_request_filter → proxy_h1 双向转发请求体和响应体 → 响应钩子」。拿到上游连接之后，请求头先发出去，然后请求体从下游流向上游、响应从上游流向下游，两个方向同时进行，直到两边都结束。这一页讲 HTTP/1 上游的 proxy_h1.rs（1271 行）和 HTTP/2 上游的 proxy_h2.rs（1391 行），共用的状态机在 proxy_common.rs。

flowchart TB
  subgraph down["下游半边 proxy_handle_downstream"]
    rb[读请求体] --> rbf[request_body_filter]
    put[处理上游任务] --> wr[写给下游]
  end
  subgraph up["上游半边 proxy_handle_upstream"]
    rr[读响应] --> tx2[发进管道]
    wb[写请求体]
  end
  rbf -->|"Body 任务"| p1[["管道 容量4"]]
  p1 --> wb
  tx2 --> p2[["管道 容量4"]]
  p2 --> put

两半各是一个 select! 循环，被 tokio::try_join! 放在同一个任务里并发 poll；它们之间只靠两个 mpsc 管道传 HttpTask。实验在 exp-pingora/proxy-lab（M2 Max、macOS 27、rustc 1.99.0，release，tokio 1.53.2）；0.8.0 的对照在 exp-pingora/proxy-08（pingora 0.8.0，tokio 1.52.3，debug）。

发请求头之前：proxy_1to1

proxy_1to1 先把下游请求头克隆一份，改成要发给上游的样子，顺序是：

1. 版本改成 HTTP/1.1；下游是 HTTP/2 时，从 :authority 合成 Host。
2. sanitize_h1_upstream_request：按 peer 上的 HttpUpstreamRequestPolicy 删逐跳头（hop-by-hop header）和 Connection 点名的头，处理 Upgrade。0.9.0 起默认就删，规则见 HTTP/1 服务端。
3. 开了缓存时，pingora_cache::filters::upstream::request_filter 加条件请求头（重新验证用）。
4. 用户的 upstream_request_filter。
5. 过滤器可能改了 Host、URI，重新对齐 authority 并校验；有请求体却既没有 Content-Length 也没有 Transfer-Encoding 时，补 Transfer-Encoding: chunked（finalize_h1_upstream_request_framing）。
6. 记下「发给上游的是不是 upgrade 请求」，upstream_compression.request_filter 记下 Accept-Encoding。
7. write_request_header。

然后建管道、打开重试缓冲，开始双向转发：

        let (tx_upstream, rx_upstream) = mpsc::channel::<HttpTask>(TASK_BUFFER_SIZE);
        let (tx_downstream, rx_downstream) = mpsc::channel::<HttpTask>(TASK_BUFFER_SIZE);

        session.as_mut().enable_retry_buffering();
        // ...
        // start bi-directional streaming
        let ret = tokio::try_join!(
            self.proxy_handle_downstream(
                session,
                tx_downstream,
                rx_upstream,
                ctx,
                &mut downstream_custom_message_writer,
                &mut downstream_custom_message_reader,
                pipe_state.clone(),
            ),
            self.proxy_handle_upstream(client_session, tx_upstream, rx_downstream, pipe_state),
        );

TASK_BUFFER_SIZE 是 4。try_join! 的含义是：两半都 Ok 才算完，任何一半返回 Err，另一半就被丢掉。try_join! 不 spawn 任务，两半跑在当前请求的任务里，见 tokio 卷的 select! 与 join!。重试缓冲（enable_retry_buffering）把读过的前 64 KiB 请求体（BODY_BUF_LIMIT）留一份，换一个上游重试时先把它发出去；超过 64 KiB 就截断，截断后不再重试，见 错误、重试与故障转移。

管道里流的是 HttpTask：

pub enum HttpTask {
    /// the response header and the boolean end of response flag
    Header(Box<pingora_http::ResponseHeader>, bool),
    /// A piece of request or response body and the end of request/response boolean flag.
    Body(Option<bytes::Bytes>, bool),
    /// Request or response body bytes that have been upgraded on H1.1, and EOF bool flag.
    UpgradedBody(Option<bytes::Bytes>, bool),
    /// HTTP response trailer
    Trailer(Option<Box<http::HeaderMap>>),
    /// Signal that the response is already finished
    Done,
    /// Signal that the reading of the response encountered errors.
    Failed(pingora_error::BError),
}

下游往上游只发 Body 和 UpgradedBody；上游往下游什么都可能发。上游读出错时不返回 Err，而是发一个 Failed 进管道，让下游半边在处理完前面的任务之后再看到错误。

上游半边：proxy_handle_upstream

proxy_handle_upstream 只有两个分支：

        while !request_done || !response_done {
            tokio::select! {
                res = client_session.read_response_task(), if !response_done => {
                    // ... Ok(task): 发进 tx；Err(e): 发 Failed 然后 return Ok(false)
                },

                body = rx.recv(), if !request_done => {
                    match send_body_to1(client_session, body).await {
                        // ...
                    }
                },

读一个响应任务（头、一块体、结束），或者从管道取一块请求体写给上游。两件事都在 client_session 上，所以不能拆成两个任务，只能在一个 select! 里轮流做。返回值是「上游连接能否复用」：响应读出错、请求体写到一半出错、或者下游提前结束导致响应没读完，都是 false。

select! 每一轮只让一个分支完成，另一个分支的 future 被丢掉，下一轮重新创建。rx.recv() 被丢掉没关系（它是 cancel safe 的），read_response_task() 被丢掉就要看它内部有没有把读了一半的数据存在 future 里。0.9.0 之前是存在 future 里的：

实验：响应头分两次到达

tools/split_header_upstream.py 读完请求头后，先发 HTTP/1.1 200 OK\r\nContent-，一秒后再发 Length: 3\r\n\r\nok\n；tools/drip_client.py 同时以每 100 ms 1000 字节的速度发一个 20000 字节的 POST 请求体。这一秒里，上游半边的 select! 不断因为 rx.recv() 有数据而丢掉正在读响应头的 future。pingora 0.8.0：

upstream_peer #1 POST
upstream: got request head
upstream: sent first half of response header
upstream: sent second half
[2026-10-29T04:59:35Z ERROR pingora_proxy] Fail to proxy: Upstream InvalidHTTPHeader context: Peer: addr: 127.0.0.1:18503, scheme: HTTP cause:  context: buf: Length: 3\r\n\r\nok\n cause: invalid HTTP version, status: 502, tries: 1, retry: false, POST /drip, Host: x
logging attempts=1 status=502
client got: b'HTTP/1.1 502 Bad Gateway'

前半个头随着被丢掉的 future 一起没了，后来的解析从 Length: 3 开始，httparse 报「invalid HTTP version」。同样的程序在 0.9.0 上：

upstream: got request head
upstream: sent first half of response header
upstream: sent second half
client got: b'HTTP/1.1 200 OK'
[2026-10-29T04:59:31Z ERROR pingora_proxy] Fail to proxy: Downstream ConnectionClosed context: Peer: addr: 127.0.0.1:18505, scheme: HTTP cause:  context: Peer prematurely closed connection with 10000 bytes of body remaining to read, status: 0, tries: 1, retry: false, POST /drip, Host: x

客户端拿到了 200。后面那条 Downstream ConnectionClosed 是实验客户端自己造成的：它读到响应头就退出了，请求体还剩 10000 字节没发。

管道关闭

上游半边往 tx 发任务失败，说明下游半边已经结束、丢掉了接收端。这可能是正常的：比如下游按 Content-Length 或 range 已经写完了响应，上游的流还没读到结束。0.9.0 用一个共享的 PipeState（AtomicU8）区分：下游半边在正常返回之前存 DownstreamComplete，上游半边发送失败时读它，是就当作良性关闭，否则返回 InternalError。这是 7142ad4（「handle upstream task-pipe closure on early downstream finish」，0.9.0）加的，之前这种情况会报一个假的「channel closed」错误并断开上游连接。良性关闭时如果响应已经读完，上游半边继续循环把请求体发完，因为连接能不能进池要看两个方向是不是都完整。

下游半边：proxy_handle_downstream

proxy_handle_downstream 复杂得多，因为缓存、自定义消息都在这里汇合。它的状态是两个小状态机：

• DownstreamStateMachine：Reading、ReadingFinished、Errored，Errored 是终态，进去之后下游连接不能复用。
• ResponseStateMachine：两个布尔，上游响应完了没有、缓存响应完了没有（没有缓存响应时默认是完）。

进循环之前，如果有重试缓冲（这是一次重试）或者请求体为空，先往管道里送一块：重试时是缓冲的内容，体为空时是 Body(None, true)。所以没有请求体的 GET 也会调一次 request_body_filter（ProxyHttp 的实测日志里那行 len=0 eos=true）。

循环的条件是「请求没读完，或者响应没写完，或者还有自定义消息要转」。select! 有七个分支，不开缓存、不用自定义协议时，真正参与的是三个：

            let send_permit = tx
                .try_reserve()
                .or_err(InternalError, "try_reserve() body pipe for upstream");
            // ...
            tokio::select! {
                // only try to send to pipe if there is capacity to avoid deadlock
                // Otherwise deadlock could happen if both upstream and downstream are blocked
                // on sending to their corresponding pipes which are both full.
                body = session.downstream_session.read_body_or_idle(downstream_state.is_done()),
                    if downstream_state.can_poll() && send_permit.is_ok() => {
                    // ... send_body_to_pipe(session, body, is_body_done, send_permit.unwrap(), ctx)
                },

                _ = tx.reserve(), if downstream_state.is_reading() && send_permit.is_err() => {
                    // If tx is closed, the upstream has already finished its job.
                    downstream_state.maybe_finished(tx.is_closed());
                    // ...
                },
                // ...
                task = rx.recv(), if !response_state.upstream_done() && next_upstream_task.is_none() => {
                    // ... process_upstream_tasks(...)
                },

• 读请求体只在已经拿到管道空位时进行。读到的一块经过下游模块和 request_body_filter（send_body_to_pipe），用预留的位置发出去，发送本身不会阻塞。
• 管道满了，就等一个空位，不读下游。
• 从 rx 收上游任务，交给 process_upstream_tasks。

请求体结束后，read_body_or_idle(true) 不再读体，而是等下游关闭连接（「idle」），这样客户端提前断开时能尽快发现。

每个上游任务经过什么

process_upstream_tasks 先把管道里已经到了的任务一次取完：

        // tokio::task::unconstrained because now_or_never may yield None when the future is ready
        while let Some(maybe_task) = tokio::task::unconstrained(rx.recv()).now_or_never() {

now_or_never 只 poll 一次；tokio 的协作调度预算（coop budget）用完时，rx.recv() 即使有数据也会返回 Pending，unconstrained 让这次 poll 不受预算限制，见 tokio 卷的 协作调度预算。然后对每个任务依次：

1. revalidate_or_stale：缓存重新验证收到 304 时，改为从缓存出响应，后面的过滤都跳过；
2. adjust_upstream_modules 和上游模块（upstream_modules feature），见 上游模块与自定义转发；
3. upstream_compression.response_filter：按需解压或压缩；
4. h1_response_filter：upstream_response_* 钩子 → 写缓存 → range / 条件请求（只在缓存改过请求头时）→ 没有长度信息的响应改成 chunked → response_* 钩子；
5. 整批交给 session.write_response_tasks，跑下游模块，写给客户端。

第 5 步是 await 在下游的 socket 写上。下游写不动时，下游半边就停在这里，不再收 rx；rx 满了，上游半边停在 tx.send 上，不再读上游。背压就是这么从下游一路传到上游的。请求方向是对称的：上游不读，上游半边的 send_body_to1 卡住，rx 不再被收，管道满，下游半边拿不到空位，不读下游。

实验：上游慢读时下游的背压

tools/slow_upstream.py 每 50 ms 读 16 KiB；tools/fast_client.py 发一个 Content-Length: 67108864（64 MiB）的 POST，用另一个线程尽快 send，每 0.5 s 报一次累计写出多少；duplex 代理在 request_body_filter 里累计「pingora 从下游拉了多少」。三个进程各自打印毫秒时间戳，跑 8 秒（uptime 的负载 18.87，机器上有其他写手在编译）。按时间排序后每 2 秒取一组：

原始输出在 proxy-lab/bp-client.log、bp-proxy.log、bp-up.log。上游的读速大约 275 KB/s，pingora 拉取的速度和它一样，每 0.5 s 前进 262144 字节（4 次 request_body_filter，每次 65536）。8 秒里客户端只写出了 64 MiB 中的 5.7 MB，其余都被挡在它自己的 send 上。

两列差值都不随时间增长，这就是背压。「pingora − 上游」约 1.1 MB，是 pingora 手里和它到上游之间的量：管道里最多 4 块（每块不超过 64 KiB），加上正在写的一块，加上代理到上游这条 TCP 连接两端的内核缓冲（上游 socket 报告的 SO_RCVBUF 是 342972）。「客户端 − pingora」约 2.2 MB，全在客户端到代理这条连接的内核缓冲里（客户端 SO_SNDBUF 报告 146988，代理一侧的接收缓冲由 macOS 自动调整，无法从外面读到）。pingora 自己的内存占用只是管道里那几块。

结束时代理打出一条 Upstream WriteError ... Broken pipe (os error 32), status: 502，那是 8 秒后 runq 杀掉慢上游造成的。

0.9.0 的 proxy task：只在从缓存出响应时用

CHANGELOG 0.9.0 的一条是「Add cancel-safe body and header writer primitives (proxy tasks) to prevent stalled cache misses from applying backpressure」。看名字容易以为整个转发都换成了 cancel-safe 写，代码里不是。

v1 服务端会话多了一个任务队列：send_proxy_task 只入队，write_proxy_tasks 把队列写出去，写到一半被取消，下次从断点继续（当前写到哪记在 proxy_task_state.current_writer 里，而不是 future 里）。下游半边用它的地方只有两个：serve_from_cache.next_http_task 那个分支（从缓存读一块就入队），和一个「闸门」分支：

                _ = std::future::ready(()),
                    if !downstream_state.is_errored()
                        && session.has_pending_downstream_tasks()
                        && next_upstream_task.is_none() => {
                    tokio::select! {
                        // Try to write downstream proxy tasks (cancel-safe)
                        write_result = session.write_downstream_proxy_tasks() => {
                            // ...
                        }

                        // Also poll for upstream tasks - if we get one, cancel the write and handle it.
                        // Only poll if there is no buffered task already waiting to be processed.
                        upstream_task = rx.recv(), if !response_state.upstream_done() && serve_from_cache.is_on() && next_upstream_task.is_none() => {
                            if let Some(t) = upstream_task {
                                // Store this upstream task to be processed next iteration
                                next_upstream_task = Some(t);
                                continue;

缓存未命中、边从上游收边写缓存、同时从缓存读出来发给下游时，如果下游很慢，旧的写法会让整个循环停在下游写上，上游的数据也就停了，缓存跟着填不满。新写法在写下游的同时也收上游任务，收到就取消写、先去处理（写缓存），下一轮再接着写下游。普通的不走缓存的转发仍然是上面第 5 步的 write_response_tasks，背压照旧。

而且这个 API 默认是关的：v1 会话的 proxy_tasks_enabled 初始值是 false，HTTP/2 下游的 supports_proxy_task_api 恒为 false。要用就在请求钩子里调 session.set_proxy_tasks_enabled(true)。缓存这一侧的细节见 缓存状态机。

Upgrade 与 WebSocket

HTTP/1 的 Upgrade（WebSocket 是最常见的一种）在这两个循环里是一次「换模式」：

• 上游半边读到 101 响应后 client_session.was_upgraded() 变真，它把 request_done 重置为 false，因为升级之后请求方向又有数据了。
• 下游半边收到 101 头并写出去之后，把 DownstreamStateMachine 重置回 Reading，继续读下游。之后两个方向的数据都是 UpgradedBody，不再按 HTTP 分帧。
• 任一方向结束（下游读到 None，或者上游写完），另一方向也算结束。

0.9.0 加了一道检查：下游和上游必须对「这是不是 upgrade 请求」意见一致，101 才算数（reject_mismatched_h1_upgrade_101）。注释给的理由：否则一边换了协议、另一边还在按 HTTP 处理，隧道里的流量可以绕过请求处理，或者把连接状态搞乱。response_filter 把状态改成 101 的情况也会被拦下（proxy_h1.rs 末尾的两个测试）。

哪些 upgrade 会被转发，由 peer 的 HttpUpstreamRequestPolicy::h1_upgrade 决定，默认 H1UpgradePolicy::WebSocketOnly：只有 HTTP/1.1 下游、Upgrade: websocket 的请求，会以规范化的 Connection: Upgrade、Upgrade: websocket 转发；别的 Upgrade 连同 Connection 一起删掉。tools/ws_upstream.py 打印收到的请求头，有 Upgrade 就回 101 然后回显；tools/ws_client.py 发一次 websocket、一次 foo：

client got: HTTP/1.1 101 Switching Protocols | Upgrade: websocket | Connection: Upgrade | body: b''
client got: b'echo:ping-1'
client got: b'echo:ping-2'
client got: HTTP/1.1 200 OK | Content-Length: 3 | Date: Thu, 29 Oct 2026 04:56:20 GMT | Connection: keep-alive | body: b'no\n'
upstream got: GET /ws HTTP/1.1 | Host: x | Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ== | Sec-WebSocket-Version: 13 | Connection: Upgrade | upgrade: websocket
upstream got: GET /ws HTTP/1.1 | Host: x | Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ== | Sec-WebSocket-Version: 13

WebSocket 隧道通了，两条消息都回显了。上游收到的 Connection: Upgrade 和 upgrade: websocket 被挪到了最后，upgrade 变成小写：它们是删掉之后重新插入的（sanitize_h1_upstream_request 的 WebSocketOnly 分支）。Upgrade: foo 的请求到上游时只剩一个普通 GET，上游回了 200。要转发任意 upgrade，得把 peer 的策略设成 Preserve（文档注释说它「RFC-non-compliant」，只为兼容旧行为）。hyper 处理 upgrade 的方式是把连接整个交还给用户，见 hyper 卷的 Upgrade。

proxy_h2：一个管道就够

HTTP/2 上游的结构一样是 try_join! 两半，但只有一个管道（上游到下游）：

• proxy_down_to_up 发请求头之前删 h1 专有的头（Transfer-Encoding、Connection、Upgrade、Keep-Alive、Proxy-Connection），把 Host 拿掉、填进 :authority；请求体为空且可以的话，在 HEADERS 帧上直接带 END_STREAM。
• 上游半边是 pipe_up_to_down_response，只读响应、发进管道，不碰请求体。
• 请求体由下游半边 bidirection_down_to_up 直接写进 h2 的 SendStream。注释说「we don't need to do the same because the h2 client_body pipe is unbounded (never block)」：h2 的发送是先进流的缓冲，不用像 h1 那样预留管道空位。真正的限制是 HTTP/2 的流量控制窗口，写被窗口卡住时，send_body_to2 会同时盯着下游 h2 流有没有被 RST_STREAM，免得一个已经取消的下游流一直占着窗口额度（b79ce4f，0.9.0）。

几处 h2 特有的处理：上游读超时或下游出错时，给上游流发 RST_STREAM(CANCEL)，读超时还把整个 h2 连接标成待关闭（可能已经卡死）；上游 HEADERS 帧带了 END_STREAM 却声明了非零 Content-Length，当作上游错误；下游是 h1 时，响应没有 Content-Length 就补 Transfer-Encoding: chunked；收到 UpgradedBody 直接 panic!，因为 h2 客户端会话不可能产生它。上游半边读响应体时还和 tx.closed() 竞争：h1 下游按 Content-Length 已经写完、退出了，就不必等 h2 流的 END_STREAM（proxy_h2.rs 里那段注释说，不这样的话要等到尾部或读超时）。

小结

• h1 转发是 try_join! 里的两个 select! 循环，中间两个容量 4 的 HttpTask 管道；上游半边独占 client_session，在读响应和写请求体之间轮流。
• 下游半边先预留管道空位再读下游，避免两半都卡在 send 上死锁；下游写不动、上游读不动，背压都会经过管道传到另一端。实测上游 275 KB/s 慢读时，pingora 和上游之间稳定在约 1.1 MB，客户端 8 秒只写出 64 MiB 中的 5.7 MB。
• select! 会丢掉没完成的分支；0.8.0 的 h1 响应头读取不是 cancel safe 的，请求体持续到达时分两次到的响应头会被读坏成 502，0.9.0 把半个头存进会话修好了。
• 0.9.0 的 proxy task 只用于从缓存出响应的路径，默认关闭；普通转发的写仍然阻塞在下游。
• 默认只转发 WebSocket 的 upgrade，其他 Upgrade 被删；101 要求上下游对 upgrade 的判断一致。
• h2 上游只有一个管道，请求体直接写进 h2 流，靠流量控制而不是管道限速。