I edited the original question cause I found a simpler reproduction.

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let ln = TcpListener::bind(("0.0.0.0", 6789)).await?;

    // start an echo client
    tokio::spawn(async move {
        let mut connector = TcpStream::connect(("127.0.0.1", 6789)).await.unwrap();
        connector.write(b"a").await.unwrap();

        let mut buf = [0; 10];
        loop {
            // consume
            if let Ok(n) = connector.read(&mut buf).await {
                connector.write(&buf[..n]).await.unwrap();
            } else {
                break;
            }
        }
    });


    loop {
        let (io, _) = ln.accept().await?;

        tokio::spawn(async move {
            if let Err(e) = handle(io).await {
                println!("Connection error: {}", e);
            }
        });
    }
}

async fn handle(io: TcpStream) -> anyhow::Result<()> {
    let mut io = io;
    let mut buf = [0; 1];

    io.read(&mut buf).await?;
    println!("server received: {:?}", buf); // a
    io.write(b"b").await?;

    // io.read_exact(&mut buf).await?;
    io.peek(&mut buf).await?;
    println!("server received: {:?}", buf); // expect b but blocked

    Ok(())
}

It will blocked when call peek.

But read can work.

    // ...
    io.read_exact(&mut buf).await?;
    // io.peek(&mut buf).await?;
    println!("server received: {:?}", buf); // print b
    // ...