Occasional flow control problem occurs on H2

Version

hyper = "=0.14.24"

Platform

windows

Description

I use a hyper-implemented client to send requests to the hyper-implemented server. When the first request is sent, traffic control errors occasionally occur. I collected the output log
bug.txt

pub const POOL_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
pub const MAX_POOL_IDLE_CONN: usize = 50;

pub const H1_HEADER_READ_TIMEOUT: Duration = Duration::from_secs(10);
pub const H1_MAX_BUFFER: usize = 32 * 1024;

pub const H2_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(10);
pub const H2_KEEP_ALIVE_TIMEOUT: Duration = Duration::from_secs(30);

pub const H2_MAX_CONCURRENT_STREAMS: u32 = 100;
pub const H2_CONNECTION_WINDOW_SIZE: u32 = 1024 * 1024 * 2;
pub const H2_STREAM_WINDOW_SIZE: u32 = 32 * 1024;

pub const H2_MAX_FRAME_SIZE: u32 = 1024 * 16;
pub const H2_MAX_SEND_BUF_SIZE: usize = 32 * 1024;
pub const H2_MAX_HEADER_LIST_SIZE: u32 = 32 * 1024;

 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: h2::codec::framed_read: poll
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: tokio_util::codec::framed_impl: attempting to decode a frame
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: tokio_util::codec::framed_impl: attempting to decode a frame
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: tokio_util::codec::framed_impl: attempting to decode a frame
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: tokio_util::codec::framed_impl: attempting to decode a frame
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: tokio_util::codec::framed_impl: frame decoded from buffer
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: h2::codec::framed_read: read.bytes=16393
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: [1mFramedRead::decode_frame [0m [1m{ [0moffset=16393 [1m} [0m: h2::codec::framed_read: decoding frame from 16393B
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: [1mFramedRead::decode_frame [0m [1m{ [0moffset=16393 [1m} [0m: h2::codec::framed_read: frame.kind=Data
 [2mSep 22 16:00:00.342 [0m  [34mDEBUG [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: [1mFramedRead::poll_next [0m: h2::codec::framed_read: received frame=Data { stream_id: StreamId(1) }
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::connection: recv DATA frame=Data { stream_id: StreamId(1) }
 [2mSep 22 16:00:00.342 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::streams::recv: recv_data; size=16384; connection=32767; stream=0
 [2mSep 22 16:00:00.343 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::streams::flow_control: send_data; sz=16384; window=32767; available=2064384
 [2mSep 22 16:00:00.343 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::streams::recv: release_connection_capacity; size=16384, connection in_flight_data=49152
 [2mSep 22 16:00:00.343 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::streams::send: send_reset(..., reason=FLOW_CONTROL_ERROR, initiator=Library, stream=StreamId(1), ..., is_reset=false; is_closed=false; pending_send.is_empty=true; state=State { inner: Open { local: AwaitingHeaders, remote: Streaming } } 
 [2mSep 22 16:00:00.343 [0m  [35mTRACE [0m  [1mserver_handshake [0m: [1mConnection [0m [1m{ [0mpeer=Server [1m} [0m: [1mpoll [0m: h2::proto::streams::send: send_reset -- queueing; frame=Reset { stream_id: StreamId(1), error_code: FLOW_CONTROL_ERROR }

[short summary of the bug]

The code I use is similar to this, but the server and client use the simple encapsulation component of Hyper (only the corresponding parameters are set for Hyper) and the TLS used by the network protocol. The probability of this problem is low. The problem is more likely to be reproduced in the debug version than in the release version.

use hyper::body::HttpBody;
use hyper::{Body, Request};
use std::net::SocketAddr;
use std::time::Duration;

#[cfg(windows)]
pub mod windows;

#[tokio::main]
async fn main() {
    use tracing::Level;
    use tracing_subscriber::FmtSubscriber;
    let subscriber = FmtSubscriber::builder()
        .with_max_level(Level::INFO)
        .finish();
    tracing::subscriber::set_global_default(subscriber).expect("setting default subscriber failed");
    use hyper::service::{make_service_fn, service_fn};
    use hyper::{Body, Error, Response, Server};
    let addr = ([127, 0, 0, 1], 3000).into();

    let make_svc = make_service_fn(|_| async {
        Ok::<_, Error>(service_fn(|req: Request<Body>| async move {
            println!("{:?}", req);
            let mut body = req.into_body();
            while let Some(_rt) = body.data().await {}
            Ok::<_, Error>(Response::new(Body::from("Hello World")))
        }))
    });
    tokio::spawn(async move {
        if let Err(err) = Server::bind(&addr)
            .http2_initial_stream_window_size(1024 * 16)
            .http2_initial_connection_window_size(1024 * 64)
            .http2_max_header_list_size(1024 * 32)
            .http2_max_send_buf_size(1024 * 32)
            .http2_only(true)
            .serve(make_svc)
            .await
        {
            eprintln!("server error: {}", err);
        }
    });
    tokio::time::sleep(Duration::from_secs(1)).await;
    send_req(addr).await;
}

async fn send_req(addr: SocketAddr) {
    let url = format!("http://{}/xxx", addr);
    let mut client = hyper::client::Client::builder()
        .http2_initial_stream_window_size(1024 * 32)
        .http2_initial_connection_window_size(1024 * 64)
        .http2_max_send_buf_size(1024 * 32)
        .http2_only(true)
        .build_http();
    let mut i = 0;
    loop {
        let req = Request::builder()
            .uri(&url)
            .body(Body::from(vec![1u8; 1024 * 1024 * 2]))
            .unwrap();
        let rt = client.request(req).await;
        println!("{:?}", rt);

        tokio::time::sleep(Duration::from_secs(1)).await;
        i += 1;
        if i % 3 == 0 {
            client = hyper::client::Client::builder()
                .http2_initial_stream_window_size(1024 * 32)
                .http2_initial_connection_window_size(1024 * 64)
                .http2_max_send_buf_size(1024 * 32)
                .http2_only(true)
                .build_http();
        }
    }
}

Start with the supplied Rust reproducer, its HTTP/2 client and server window settings, and the attached flow-control trace on Windows. Compare the trace's h2 flow-control transitions with the request lifecycle and determine why the first large request is reset; done means the supplied scenario no longer intermittently produces FLOW_CONTROL_ERROR.