Rust程序处理GCP告警请求时在epoll_wait处阻塞问题排查

问题场景 #

我定义了用于反序列化GCP告警请求的结构体：

• 顶层结构体实现FromRequest trait，用于Axum请求解析
• 嵌套结构体实现serde的Deserialize和Serialize trait

编写测试用例对比手动实例化的GcpAlert与Axum处理请求生成的实例时，程序在GcpBody的FromRequest实现中的body::to_bytes(body).await处冻结。调试发现：

• 存在两个活跃线程
• Data结构体被轮询两次，第二次本该无数据返回却持续等待

将代码改为take(1)时测试成功，take值大于1则复现阻塞问题。

核心原因：Request Body是消耗型Stream，不可重复读取 #

Axum基于hyper实现，hyper的Body本质是一个Stream<Item = Result<Bytes, Error>>，这类Stream有两个关键特性：

1. 数据只能被读取一次，读取完成后Stream就会被耗尽
2. 若尝试读取已经耗尽的Stream，且底层实现未正确返回Poll::Ready(None)（比如测试中用静态Bytes生成的Body），就会导致await永远挂起，表现为程序冻结

你的场景中，代码/测试逻辑存在两次读取同一个Request Body的行为：

• 第一次读取成功消耗了所有数据
• 第二次读取时，Stream已无数据，但底层未触发结束信号，导致body::to_bytes一直等待新数据，最终阻塞

take(1)能临时解决问题，是因为它强制只读取Stream的第一个数据块，读取完成后直接返回None结束Stream，避免了后续无意义的轮询等待，但这只是规避手段，并非根本解法。

正确解决思路 #

1. 杜绝重复读取同一个Request Body

   • 确保FromRequest的实现中只读取一次Body，不要在逻辑中重复调用body::to_bytes或其他读取方法
   • 测试时，每次测试都创建全新的Request实例，不要复用同一个Request（因为Body被消耗后无法恢复）

2. 缓存Body数据（如需多次使用）
   如果业务逻辑需要多次使用Body内容，可在第一次读取后将Bytes存储到请求扩展（Extension）中，后续从扩展中获取，而非再次读取Body：

   impl FromRequest<()> for GcpBody {
       type Rejection = StatusCode;
   
       async fn from_request(req: &mut RequestParts<()>) -> Result<Self, Self::Rejection> {
           // 优先从Extension获取已读取的Bytes
           if let Some(bytes) = req.extensions().get::<Bytes>() {
               let body: GcpBody = serde_json::from_slice(bytes).map_err(|_| StatusCode::BAD_REQUEST)?;
               return Ok(body);
           }
   
           // 第一次读取Body并缓存
           let bytes = body::to_bytes(req.body_mut()).await.map_err(|_| StatusCode::BAD_REQUEST)?;
           req.extensions_mut().insert(bytes.clone());
           let body: GcpBody = serde_json::from_slice(&bytes).map_err(|_| StatusCode::BAD_REQUEST)?;
           Ok(body)
       }
   }
   

3. 测试用可重置Body（可选）
   如果测试需要复用Request结构，可以用hyper::Body::wrap_stream包装可重复生成的Stream，比如用futures::stream::iter生成多次相同的Bytes流，但这种方式仅适用于测试场景。

内容的提问来源于stack exchange，提问作者heretomurimudamura