带宽控制AsyncRead包装导致无理由连接关闭的问题排查求助

我最近在给AsyncRead实现一个带宽控制的包装流BandwidthControlStream，但遇到了一个奇怪的问题，希望各位能帮忙分析一下。

我的设计逻辑是这样的：先从内部流读取数据到buf，再尝试从上游的令牌桶获取对应数据量的令牌。如果能成功获取令牌，就返回Ready；如果令牌不足，就返回Poll::Pending等待。原本预期这个逻辑能正常限流，但现在发现当需要等待令牌（触发Poll::Pending）时，连接会无理由断开，而如果能一次性拿到足够令牌的话，整个流程就完全正常。

客户端用curl请求时返回的错误信息如下：

• upload completely sent off: 18 bytes
• Empty reply from server
• shutting down connection #0
  curl: (52) Empty reply from server

以下是我实现AsyncRead的核心代码：

impl<T> AsyncRead for BandwidthControlStream<T>
where T: AsyncStream,
{
    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut ReadBuf<'_>,
    ) -> Poll<io::Result<()>> {
        let this = self.project();

        if !*this.has_acquire_upstream {
            ready!(this.inner.poll_read(cx, buf))?;
            *this.has_acquire_upstream = true;
        }

        loop {
            if let Some(acquire) = this.upstream_acquire {
                match Pin::new(acquire).poll(cx) {
                    Poll::Ready(_) => {
                        *this.upstream_acquire = None;
                        *this.has_acquire_upstream = false;
                        return Poll::Ready(Ok(()));
                    },
                    Poll::Pending => {
                        counter!(*METRIC_BLOCK_BAND_UP).increment(1);
                        return Poll::Pending;
                    },
                }
            } else {
                let size = buf.filled().len();
                if size <= 0 {
                    *this.has_acquire_upstream = false;
                    return Poll::Ready(Ok(()));
                }
                *this.upstream_acquire = Some(Box::pin(this.upstream_control.clone().acquire_owned(size)));
                continue;
            }
        }
    }
}

我的困惑点主要有这几个：

• 为什么只有当需要等待令牌返回Pending时，连接才会无理由断开？
• 是不是我的poll_read逻辑中，在返回Pending前的buf处理有问题？比如已经填充了数据但没有正确保留状态，导致后续恢复时出现异常？

麻烦各位帮忙看看问题出在哪里，非常感谢！

内容来源于stack exchange