http3 0.1.0

// identity_op: we write out how test values are computed
#![allow(clippy::identity_op)]

use std::{borrow::BorrowMut, time::Duration};

use assert_matches::assert_matches;
use bytes::{Buf, Bytes, BytesMut};
use futures_util::future;
use http::{Request, Response, StatusCode};
use tokio::sync::oneshot::{self};

use super::{Pair, http3_quinn, init_tracing};
use crate::{
    client,
    client::SendRequest,
    error::{Code, ConnectionError, LocalError, StreamError},
    proto::{
        coding::{Decode as _, Encode as _},
        frame::{Frame, Settings},
        push::PushId,
        stream::StreamType,
        varint::VarInt,
    },
    qpack,
    quic::{self, ConnectionErrorIncoming, SendStream},
    server,
    shared_state::ConnectionState,
    tests::get_stream_blocking,
};

#[tokio::test]
async fn connect() {
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut drive, _client) = client::new(pair.client().await).await.expect("client init");
        assert_matches!(
            future::poll_fn(|cx| drive.poll_close(cx)).await,
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value()
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let _server = server::Connection::new(conn).await.unwrap();
    };

    tokio::select!(() = server_fut => (), () = client_fut => panic!("client resolved first"));
}

#[tokio::test]
async fn accept_request_end_on_client_close() {
    let mut pair = Pair::default();
    let mut server = pair.server();
    let client = pair.client();
    let (tx, rx) = oneshot::channel::<()>();
    let client_fut = async move {
        let client = client.await;
        let (mut driver, client) = client::new(client).await.expect("client init");
        let driver = async move {
            let _ = future::poll_fn(|cx: &mut std::task::Context<'_>| driver.poll_close(cx)).await;
        };

        let client_fut = async move {
            // wait for the server to accept the connection
            rx.await.unwrap();
            // client is dropped, it will send H3_NO_ERROR
            drop(client);
        };
        tokio::join!(driver, client_fut);
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        tx.send(()).unwrap();
        assert_matches!(
            incoming.accept().await.err().unwrap(),
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{error_code: code, ..})
            if code == Code::H3_NO_ERROR.value()
        );
    };
    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn server_drop_close() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let server_fut = async {
        let conn = server.next().await;
        let _ = server::Connection::new(conn).await.unwrap();
    };

    let client_fut = async {
        let (mut conn, mut send) = client::new(pair.client().await).await.expect("client init");
        let request_fut = async move {
            let mut request_stream = send
                .send_request(Request::get("http://no.way").body(()).unwrap())
                .await
                .unwrap();
            let response = request_stream.recv_response().await;

            assert_matches!(
                response.unwrap_err(),
                StreamError::ConnectionError(ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                    error_code: code,
                    ..
                }))
                if code == Code::H3_NO_ERROR.value()
            );
        };

        let drive_fut = async {
            let drive = future::poll_fn(|cx| conn.poll_close(cx)).await;
            assert_matches!(drive, ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value());
        };
        tokio::join! {request_fut,drive_fut}
    };
    tokio::join!(server_fut, client_fut);
}

// In this test the client calls send_data() without doing a finish(),
// i.e client keeps the body stream open. And client expects server to
// read_data() and send a response
#[tokio::test]
async fn server_send_data_without_finish() {
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (_driver, mut send_request) = client::new(pair.client().await).await.unwrap();

        let mut req = send_request
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();
        let data = vec![0; 100];
        req.send_data(bytes::Bytes::copy_from_slice(&data))
            .await
            .unwrap();
        let _ = req.recv_response().await.unwrap();
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        let request_resolver = incoming.accept().await.unwrap().unwrap();
        let (_, mut stream) = request_resolver.resolve_request().await.unwrap();
        let mut data = stream.recv_data().await.unwrap().unwrap();
        let data = data.copy_to_bytes(data.remaining());
        assert_eq!(data.len(), 100);
        response(stream).await;
        server.endpoint.wait_idle().await;
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn client_close_only_on_last_sender_drop() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();

        let (_, mut stream) = incoming
            .accept()
            .await
            .unwrap()
            .unwrap()
            .resolve_request()
            .await
            .unwrap();
        stream.stop_stream(Code::H3_REQUEST_CANCELLED);

        let (_, mut stream) = incoming
            .accept()
            .await
            .unwrap()
            .unwrap()
            .resolve_request()
            .await
            .unwrap();
        stream.stop_stream(Code::H3_REQUEST_CANCELLED);

        assert_matches!(
            incoming.accept().await.err().unwrap(),
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value()
        );
    };

    let client_fut = async {
        let (mut conn, mut send1) = client::new(pair.client().await).await.expect("client init");
        let mut send2 = send1.clone();
        let mut request_stream_1 = send1
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();

        assert_matches!(
            request_stream_1.recv_response().await,
            Err(StreamError::RemoteTerminate{
                code
            }) if code == Code::H3_REQUEST_CANCELLED.value()
        );

        request_stream_1.finish().await.unwrap();

        let mut request_stream_2 = send2
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();

        assert_matches!(
            request_stream_2.recv_response().await,
            Err(StreamError::RemoteTerminate{
                code
            }) if code == Code::H3_REQUEST_CANCELLED.value()
        );
        request_stream_2.finish().await.unwrap();

        drop(send1);
        drop(send2);

        let drive = future::poll_fn(|cx| conn.poll_close(cx)).await;
        assert_matches!(
            drive,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_NO_ERROR,
                    ..
                }
            }
        );
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn settings_exchange_client() {
    //= https://www.rfc-editor.org/rfc/rfc9114#section-3.2
    //= type=test
    //# After the QUIC connection is
    //# established, a SETTINGS frame MUST be sent by each endpoint as the
    //# initial frame of their respective HTTP control stream.

    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut conn, client) = client::new(pair.client().await).await.expect("client init");
        let settings_change = async {
            for _ in 0..10 {
                if client.settings().max_field_section_size == 12 {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(2)).await;
            }
            panic!("peer's max_field_section_size didn't change");
        };

        let drive = async move {
            assert_matches!(future::poll_fn(|cx| conn.poll_close(cx)).await,
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value());
        };

        tokio::select! { _ = settings_change => (), _ = drive => panic!("driver resolved first") };
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::builder()
            .max_field_section_size(12)
            .build(conn)
            .await
            .unwrap();
        incoming.accept().await.unwrap()
    };

    tokio::select! { _ = server_fut => panic!("server resolved first"), _ = client_fut => () };
}

#[tokio::test]
async fn settings_exchange_server() {
    //= https://www.rfc-editor.org/rfc/rfc9114#section-3.2
    //= type=test
    //# After the QUIC connection is
    //# established, a SETTINGS frame MUST be sent by each endpoint as the
    //# initial frame of their respective HTTP control stream.

    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut conn, _client) = client::builder()
            .max_field_section_size(12)
            .build::<_, _, Bytes>(pair.client().await)
            .await
            .expect("client init");
        let drive = async move {
            assert_matches!(
                future::poll_fn(|cx| conn.poll_close(cx)).await,
                ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                    error_code: code,
                    ..
                }) if code == Code::H3_NO_ERROR.value()
            );
        };

        drive.await;
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();

        let state = incoming.inner.shared.clone();
        let accept = async { incoming.accept().await.unwrap() };

        let settings_change = async {
            for _ in 0..10 {
                if state.settings().max_field_section_size == 12 {
                    return;
                }
                tokio::time::sleep(Duration::from_millis(2)).await;
            }
            panic!("peer's max_field_section_size didn't change");
        };
        tokio::select! { _ = accept => panic!("server resolved first"), _ = settings_change => () };
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn client_error_on_bidi_recv() {
    let mut pair = Pair::default();
    let server = pair.server();

    let client_fut = async {
        let (mut conn, mut send) = client::new(pair.client().await).await.expect("client init");

        //= https://www.rfc-editor.org/rfc/rfc9114#section-6.1
        //= type=test
        //# Clients MUST treat
        //# receipt of a server-initiated bidirectional stream as a connection
        //# error of type H3_STREAM_CREATION_ERROR unless such an extension has
        //# been negotiated.
        let driver = future::poll_fn(|cx| conn.poll_close(cx));
        assert_matches!(
            driver.await,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_STREAM_CREATION_ERROR,
                    reason: reason_string
                }
            } if reason_string.starts_with("client received a server-initiated bidirectional stream")
        );
        assert_matches!(send.send_request(Request::get("http://no.way").body(()).unwrap())
            .await.map(|_| ()).unwrap_err(),
            StreamError::ConnectionError(
                ConnectionError::Local { error: LocalError::Application { code: Code::H3_STREAM_CREATION_ERROR, reason: reason_string } }
            )
            if reason_string.starts_with("client received a server-initiated bidirectional stream")
        );
    };

    let server_fut = async {
        let connection = server.endpoint.accept().await.unwrap().await.unwrap();
        let (mut send, _recv) = connection.open_bi().await.unwrap();
        for _ in 0..100 {
            match send.write(b"I'm not really a server").await {
                Err(quinn::WriteError::ConnectionLost(
                    quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose {
                        error_code,
                        ..
                    }),
                )) if Code::H3_STREAM_CREATION_ERROR == error_code.into_inner() => return,
                Err(e) => panic!("got err: {}", e),
                Ok(_) => (),
            }
            tokio::time::sleep(Duration::from_millis(10)).await;
        }
        panic!("did not get the expected error");
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn client_accepts_late_control_streams_after_driver_wake() {
    use std::{
        sync::Arc,
        task::{Context, Poll, Wake, Waker},
    };

    use tokio::sync::Notify;

    struct DriverWake(Notify);

    impl Wake for DriverWake {
        fn wake(self: Arc<Self>) {
            self.0.notify_one();
        }
    }

    let mut pair = Pair::default();
    let server = pair.server();
    let (ready_tx, ready_rx) = oneshot::channel();
    let (settings_tx, settings_rx) = oneshot::channel();

    let client_fut = async {
        let (mut driver, _sender) = client::builder()
            .send_grease(false)
            .build::<_, _, Bytes>(pair.client().await)
            .await
            .unwrap();
        let wake = Arc::new(DriverWake(Notify::new()));
        let waker = Waker::from(wake.clone());
        assert!(
            driver
                .poll_close(&mut Context::from_waker(&waker))
                .is_pending()
        );
        ready_tx.send(()).unwrap();

        // Only the registered driver waker may trigger another poll. The peer
        // creates its first control stream after the initial accept was Pending.
        loop {
            wake.0.notified().await;
            assert!(
                driver
                    .poll_close(&mut Context::from_waker(&waker))
                    .is_pending()
            );
            if driver.settings().max_field_section_size == 12 {
                break;
            }
        }
        settings_tx.send(()).unwrap();

        let error = loop {
            wake.0.notified().await;
            if let Poll::Ready(error) = driver.poll_close(&mut Context::from_waker(&waker)) {
                break error;
            }
        };
        assert_matches!(
            error,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_STREAM_CREATION_ERROR,
                    ..
                }
            }
        );
    };

    let server_fut = async {
        let conn = server.endpoint.accept().await.unwrap().await.unwrap();
        ready_rx.await.unwrap();
        let mut control = conn.open_uni().await.unwrap();
        let mut settings = Settings::default();
        settings
            .insert(crate::proto::frame::SettingId::MAX_HEADER_LIST_SIZE, 12)
            .unwrap();
        let mut encoded = BytesMut::new();
        StreamType::CONTROL.encode(&mut encoded);
        Frame::<Bytes>::Settings(settings).encode(&mut encoded);
        control.write_all(&encoded).await.unwrap();

        settings_rx.await.unwrap();
        let mut duplicate = conn.open_uni().await.unwrap();
        encoded.clear();
        StreamType::CONTROL.encode(&mut encoded);
        duplicate.write_all(&encoded).await.unwrap();
        // Keep both critical streams open: rejection must be for the duplicate,
        // not an incidental FIN or reset.
        // https://www.rfc-editor.org/rfc/rfc9114.html#section-6.2.1
        conn.closed().await;
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(server_fut, client_fut);
    })
    .await
    .expect("late control streams did not wake the client driver");
}

#[tokio::test]
async fn two_control_streams() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let connection = pair.client_inner().await;

        //= https://www.rfc-editor.org/rfc/rfc9114#section-6.2.1
        //= type=test
        //# Only one control stream per peer is permitted;
        //# receipt of a second stream claiming to be a control stream MUST be
        //# treated as a connection error of type H3_STREAM_CREATION_ERROR.
        for _ in 0..=1 {
            let mut control_stream = connection.open_uni().await.unwrap();
            let mut buf = BytesMut::new();
            StreamType::CONTROL.encode(&mut buf);
            control_stream.write_all(&buf[..]).await.unwrap();
        }

        tokio::time::sleep(Duration::from_secs(10)).await;
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_STREAM_CREATION_ERROR,
                    ..
                }
            }
        );
    };

    tokio::select! { _ = server_fut => (), _ = client_fut => panic!("client resolved first") };
}

#[tokio::test]
async fn server_rejects_invalid_qpack_decoder_stream_instruction() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();
    let (done_send, done_recv) = oneshot::channel();

    let client_fut = async {
        let connection = pair.client_inner().await;

        let mut control_stream = connection.open_uni().await.unwrap();
        let mut control = BytesMut::new();
        StreamType::CONTROL.encode(&mut control);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut control);
        control_stream.write_all(&control).await.unwrap();

        let mut decoder_stream = connection.open_uni().await.unwrap();
        let mut instruction = BytesMut::new();
        StreamType::DECODER.encode(&mut instruction);
        // No dynamic field section is outstanding on stream 0, so this Section
        // Acknowledgment is a decoder-stream error.
        // https://www.rfc-editor.org/rfc/rfc9204.html#section-4.4.1
        instruction.extend_from_slice(&[0x80]);
        decoder_stream.write_all(&instruction).await.unwrap();

        let _ = done_recv.await;
        drop(decoder_stream);
        drop(control_stream);
        drop(connection);
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::QPACK_DECODER_STREAM_ERROR,
                    ..
                }
            }
        );
        done_send.send(()).unwrap();
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("QPACK decoder-stream error was not observed");
}

#[tokio::test]
async fn server_rejects_invalid_qpack_encoder_stream_instruction() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();
    let (done_send, done_recv) = oneshot::channel();

    let client_fut = async {
        let connection = pair.client_inner().await;

        let mut control_stream = connection.open_uni().await.unwrap();
        let mut control = BytesMut::new();
        StreamType::CONTROL.encode(&mut control);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut control);
        control_stream.write_all(&control).await.unwrap();

        let mut encoder_stream = connection.open_uni().await.unwrap();
        let mut instruction = BytesMut::new();
        StreamType::ENCODER.encode(&mut instruction);
        // The default server setting permits no dynamic table capacity. A Set
        // Dynamic Table Capacity instruction for 1 therefore exceeds the limit.
        // https://www.rfc-editor.org/rfc/rfc9204.html#section-4.3.1
        instruction.extend_from_slice(&[0x21]);
        encoder_stream.write_all(&instruction).await.unwrap();

        let _ = done_recv.await;
        drop(encoder_stream);
        drop(control_stream);
        drop(connection);
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::QPACK_ENCODER_STREAM_ERROR,
                    ..
                }
            }
        );
        done_send.send(()).unwrap();
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("QPACK encoder-stream error was not observed");
}

#[tokio::test]
async fn server_rejects_oversized_qpack_encoder_stream_string() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();
    let (done_send, done_recv) = oneshot::channel();

    let client_fut = async {
        let connection = pair.client_inner().await;

        let mut control_stream = connection.open_uni().await.unwrap();
        let mut control = BytesMut::new();
        StreamType::CONTROL.encode(&mut control);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut control);
        control_stream.write_all(&control).await.unwrap();

        let mut encoder_stream = connection.open_uni().await.unwrap();
        let mut instruction = BytesMut::new();
        StreamType::ENCODER.encode(&mut instruction);
        // Insert with Literal Name, followed by an encoded name length of two.
        // The server's one-byte local decode budget rejects the instruction
        // before waiting for either payload byte.
        // https://www.rfc-editor.org/rfc/rfc9204.html#section-7.4
        instruction.extend_from_slice(&[0b0100_0010]);
        encoder_stream.write_all(&instruction).await.unwrap();

        let _ = done_recv.await;
        drop(encoder_stream);
        drop(control_stream);
        drop(connection);
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut builder = server::builder();
        builder.max_qpack_decode_buffer_size(1);
        let mut incoming = builder.build(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::QPACK_ENCODER_STREAM_ERROR,
                    ..
                }
            }
        );
        done_send.send(()).unwrap();
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("QPACK encoder-stream string limit was not enforced");
}

#[tokio::test]
async fn server_rejects_closed_qpack_encoder_stream() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();
    let (done_send, done_recv) = oneshot::channel();

    let client_fut = async {
        let connection = pair.client_inner().await;

        let mut control_stream = connection.open_uni().await.unwrap();
        let mut control = BytesMut::new();
        StreamType::CONTROL.encode(&mut control);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut control);
        control_stream.write_all(&control).await.unwrap();

        let mut encoder_stream = connection.open_uni().await.unwrap();
        let mut stream_type = BytesMut::new();
        StreamType::ENCODER.encode(&mut stream_type);
        encoder_stream.write_all(&stream_type).await.unwrap();

        // FIN closes a critical QPACK stream even when no dynamic entries were sent.
        // https://www.rfc-editor.org/rfc/rfc9204.html#section-4.2
        encoder_stream.finish().unwrap();

        let _ = done_recv.await;
        drop(encoder_stream);
        drop(control_stream);
        drop(connection);
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_CLOSED_CRITICAL_STREAM,
                    ..
                }
            }
        );
        done_send.send(()).unwrap();
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("closed QPACK encoder stream was not observed");
}

#[tokio::test]
async fn client_rejects_closed_qpack_decoder_stream() {
    init_tracing();
    let mut pair = Pair::default();
    let server = pair.server();
    let (done_send, done_recv) = oneshot::channel();

    let client_fut = async {
        let (mut connection, _send_request) =
            client::new(pair.client().await).await.expect("client init");

        assert_matches!(
            future::poll_fn(|cx| connection.poll_close(cx)).await,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_CLOSED_CRITICAL_STREAM,
                    ..
                }
            }
        );
        done_send.send(()).unwrap();
    };

    let server_fut = async {
        let connection = server.endpoint.accept().await.unwrap().await.unwrap();
        let mut decoder_stream = connection.open_uni().await.unwrap();
        let mut stream_type = BytesMut::new();
        StreamType::DECODER.encode(&mut stream_type);
        decoder_stream.write_all(&stream_type).await.unwrap();

        // Both FIN and reset close a critical QPACK stream.
        // https://www.rfc-editor.org/rfc/rfc9204.html#section-4.2
        decoder_stream.finish().unwrap();

        let _ = done_recv.await;
        drop(decoder_stream);
        drop(connection);
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("closed QPACK decoder stream was not observed");
}

#[tokio::test]
async fn client_flushes_insert_count_increment_without_a_blocked_request() {
    init_tracing();
    let mut pair = Pair::default();
    let server = pair.server();

    let client_fut = async {
        let mut builder = client::builder();
        builder.qpack_max_table_capacity(128);
        builder.send_grease(false);
        let (mut connection, _send_request) = builder
            .build::<_, _, Bytes>(pair.client().await)
            .await
            .expect("client init");

        assert_matches!(
            future::poll_fn(|cx| connection.poll_close(cx)).await,
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose {
                error_code,
            }) if error_code == Code::H3_NO_ERROR.value()
        );
    };

    let server_fut = async {
        let connection = server.endpoint.accept().await.unwrap().await.unwrap();

        let mut control_stream = connection.open_uni().await.unwrap();
        let mut control = BytesMut::new();
        StreamType::CONTROL.encode(&mut control);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut control);
        control_stream.write_all(&control).await.unwrap();

        let mut decoder_stream = connection.open_uni().await.unwrap();
        let mut stream_type = BytesMut::new();
        StreamType::DECODER.encode(&mut stream_type);
        decoder_stream.write_all(&stream_type).await.unwrap();

        let mut encoder_stream = connection.open_uni().await.unwrap();
        let mut instructions = BytesMut::new();
        StreamType::ENCODER.encode(&mut instructions);
        qpack::DynamicTableSizeUpdate(128).encode(&mut instructions);
        qpack::InsertWithoutNameRef::new("name", "value")
            .encode(&mut instructions)
            .unwrap();
        encoder_stream.write_all(&instructions).await.unwrap();

        let mut other_streams = Vec::new();
        loop {
            let mut stream = connection.accept_uni().await.unwrap();
            let mut received = BytesMut::new();
            while received.is_empty() {
                let chunk = stream
                    .read_chunk(usize::MAX, true)
                    .await
                    .unwrap()
                    .expect("client critical stream closed");
                received.extend_from_slice(&chunk.bytes);
            }

            let stream_type = StreamType::decode(&mut received).unwrap();
            if stream_type != StreamType::DECODER {
                other_streams.push(stream);
                continue;
            }

            while received.is_empty() {
                let chunk = stream
                    .read_chunk(usize::MAX, true)
                    .await
                    .unwrap()
                    .expect("client decoder stream closed");
                received.extend_from_slice(&chunk.bytes);
            }

            assert_eq!(
                qpack::InsertCountIncrement::decode(&mut received),
                Ok(Some(qpack::InsertCountIncrement(1)))
            );
            assert!(!received.has_remaining());
            other_streams.push(stream);
            break;
        }

        connection.close(
            quinn::VarInt::from_u64(Code::H3_NO_ERROR.value()).unwrap(),
            b"test complete",
        );

        drop((
            control_stream,
            decoder_stream,
            encoder_stream,
            other_streams,
        ));
    };

    tokio::time::timeout(Duration::from_secs(5), async {
        tokio::join!(client_fut, server_fut);
    })
    .await
    .expect("Insert Count Increment was not received");
}

#[tokio::test]
async fn control_close_send_error() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let connection = pair.client_inner().await;
        let mut control_stream = connection.open_uni().await.unwrap();

        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();

        //= https://www.rfc-editor.org/rfc/rfc9114#section-6.2.1
        //= type=test
        //# If either control
        //# stream is closed at any point, this MUST be treated as a connection
        //# error of type H3_CLOSED_CRITICAL_STREAM.
        control_stream.finish().unwrap(); // close the client control stream immediately

        // create the Connection manually, so it does not open a second Control stream

        let connection_error = loop {
            let accepted = connection.accept_bi().await;
            match accepted {
                // do nothing with the stream
                Ok(_) => continue,
                Err(err) => break err,
            }
        };

        let err_code = match connection_error {
            quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose {
                error_code,
                ..
            }) => error_code.into_inner(),
            e => panic!("unexpected error: {:?}", e),
        };
        assert_eq!(err_code, Code::H3_CLOSED_CRITICAL_STREAM.value());
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        // Driver detects that the receiving side of the control stream has been closed
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_CLOSED_CRITICAL_STREAM,
                    reason: reason_string
                }
            }
            if reason_string.starts_with("control stream was closed"));
        // Poll it once again returns the previously stored error
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_CLOSED_CRITICAL_STREAM,
                    reason: reason_string
                }
            }
            if reason_string.starts_with("control stream was closed"));
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn missing_settings() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let connection = pair.client_inner().await;
        let mut control_stream = connection.open_uni().await.unwrap();

        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);

        //= https://www.rfc-editor.org/rfc/rfc9114#section-6.2.1
        //= type=test
        //# If the first frame of the control stream is any other frame
        //# type, this MUST be treated as a connection error of type
        //# H3_MISSING_SETTINGS.
        Frame::<Bytes>::CancelPush(PushId(0)).encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();

        tokio::time::sleep(Duration::from_secs(10)).await;
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_MISSING_SETTINGS,
                    ..
                }
            }
        );
    };

    tokio::select! { _ = server_fut => (), _ = client_fut => panic!("client resolved first") };
}

#[tokio::test]
async fn control_stream_frame_unexpected() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let connection = pair.client_inner().await;
        let mut control_stream = connection.open_uni().await.unwrap();

        // Send a Settings frame or we get a H3_MISSING_SETTINGS instead of H3_FRAME_UNEXPECTED
        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);
        Frame::Settings::<Bytes>(Settings::default()).encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();

        //= https://www.rfc-editor.org/rfc/rfc9114#section-7.2.1
        //= type=test
        //# If
        //# a DATA frame is received on a control stream, the recipient MUST
        //# respond with a connection error of type H3_FRAME_UNEXPECTED.
        let mut buf = BytesMut::new();
        Frame::Data(Bytes::from("")).encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();
        tokio::time::sleep(Duration::from_secs(10)).await;
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_FRAME_UNEXPECTED,
                    ..
                }
            }
        );
    };

    tokio::select! { _ = server_fut => (), _ = client_fut => panic!("client resolved first") };
}

#[tokio::test]
async fn timeout_on_control_frame_read() {
    init_tracing();
    let mut pair = Pair::default();
    pair.with_timeout(Duration::from_millis(10));

    let mut server = pair.server();

    let client_fut = async {
        let (mut driver, _send_request) = client::new(pair.client().await).await.unwrap();
        let _ = future::poll_fn(|cx| driver.poll_close(cx)).await;
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert_matches!(
            incoming.accept().await.map(|_| ()).unwrap_err(),
            ConnectionError::Timeout
        );
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn goaway_from_server_not_request_id() {
    init_tracing();
    let mut pair = Pair::default();
    let server = pair.server_inner();

    let client_fut = async {
        let connection = pair.client_inner().await;
        let mut control_stream = connection.open_uni().await.unwrap();

        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();
        control_stream.finish().unwrap(); // close the client control stream immediately

        let (mut driver, _send) = client::new(http3_quinn::Connection::new(connection))
            .await
            .unwrap();

        assert_matches!(
            future::poll_fn(|cx| driver.poll_close(cx)).await,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_ID_ERROR,
                    ..
                }
            }
        )
    };

    let server_fut = async {
        let conn = server.accept().await.unwrap().await.unwrap();
        let mut control_stream = conn.open_uni().await.unwrap();

        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);
        Frame::<Bytes>::Settings(Settings::default()).encode(&mut buf);

        //= https://www.rfc-editor.org/rfc/rfc9114#section-7.2.6
        //= type=test
        //# A client MUST treat receipt of a GOAWAY frame containing a stream ID
        //# of any other type as a connection error of type H3_ID_ERROR.

        // StreamId(index=0 << 2 | dir=Uni << 1 | initiator=Server as u64)
        Frame::<Bytes>::Goaway(VarInt(0u64 << 2 | 0 << 1 | 1)).encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();

        tokio::time::sleep(Duration::from_secs(10)).await;
    };

    tokio::select! { _ = server_fut => panic!("client resolved first"), _ = client_fut => () };
}

#[tokio::test]
async fn graceful_shutdown_server_rejects() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (_driver, mut send_request) = client::new(pair.client().await).await.unwrap();

        let mut first = send_request
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();
        let mut rejected = send_request
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();
        let first = first.recv_response().await;
        let rejected = rejected.recv_response().await;

        assert_matches!(first, Ok(_));
        assert_matches!(
            rejected.unwrap_err(),
            StreamError::RemoteTerminate {
                code: Code::H3_REQUEST_REJECTED
            }
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        let request_resolver = incoming.accept().await.unwrap().unwrap();
        let (_, stream) = request_resolver.resolve_request().await.unwrap();
        response(stream).await;
        incoming.shutdown(0).await.unwrap();
        assert_matches!(incoming.accept().await.map(|x| x.map(|_| ())), Ok(None));
        server.endpoint.wait_idle().await;
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn graceful_shutdown_grace_interval() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut driver, mut send_request) = client::new(pair.client().await).await.unwrap();

        // Sent as the connection is not shutting down
        let mut first = send_request
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();
        // Sent as the connection is shutting down, but GoAway has not been received yet
        let mut in_flight = send_request
            .send_request(Request::get("http://no.way").body(()).unwrap())
            .await
            .unwrap();
        let first = first.recv_response().await;
        let in_flight = in_flight.recv_response().await;

        // Will not be sent as client's driver already received the GoAway
        let too_late = async move {
            tokio::time::sleep(Duration::from_millis(15)).await;
            request(send_request).await
        };
        let driver = future::poll_fn(|cx| driver.poll_close(cx));

        let (too_late, driver) = tokio::join!(too_late, driver);
        assert_matches!(first, Ok(_));
        assert_matches!(in_flight, Ok(_));
        assert_matches!(too_late.unwrap_err(), StreamError::RemoteClosing);
        assert_matches!(
            driver,
            ConnectionError::Local {
                error: LocalError::Application {
                    code: Code::H3_NO_ERROR,
                    ..
                }
            }
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        let (_, first) = get_stream_blocking(&mut incoming).await.unwrap();
        incoming.shutdown(1).await.unwrap();
        let (_, in_flight) = get_stream_blocking(&mut incoming).await.unwrap();
        response(first).await;
        response(in_flight).await;

        while let Some((_, stream)) = get_stream_blocking(&mut incoming).await {
            response(stream).await;
        }

        // Ensure `too_late` request is executed as the connection is still
        // closing (no QUIC `Close` frame has been fired yet)
        tokio::time::sleep(Duration::from_millis(50)).await;
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
async fn graceful_shutdown_closes_when_idle() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut driver, mut send_request) = client::new(pair.client().await).await.unwrap();

        // Make continuous requests, ignoring GoAway because the connection is not driven
        while request(&mut send_request).await.is_ok() {
            tokio::task::yield_now().await;
        }
        assert_matches!(
            future::poll_fn(|cx| { driver.poll_close(cx) }).await,
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value()
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();

        let mut count = 0;

        while let Some(resolver) = incoming.accept().await.expect("accept during shutdown") {
            let (_, stream) = resolver.resolve_request().await.expect("resolve request");
            count += 1;
            if count == 4 {
                incoming.shutdown(2).await.unwrap();
            }

            response(stream).await;
        }
        assert_eq!(count, 6, "shutdown must finish the two allowed requests");
    };

    // Real QUIC handshakes and CI scheduling are not bounded to 100 ms.
    // Wait for both peers so an early completion cannot skip the other side's
    // shutdown assertions. The timeout only guards against a stalled driver.
    tokio::time::timeout(Duration::from_secs(10), async {
        tokio::join!(server_fut, client_fut);
    })
    .await
    .expect("graceful shutdown stalled");
}

#[tokio::test]
async fn graceful_shutdown_client() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        let (mut driver, mut _send_request) = client::new(pair.client().await).await.unwrap();
        driver.shutdown(0).await.unwrap();
        assert_matches!(
            future::poll_fn(|cx| { driver.poll_close(cx) }).await,
            ConnectionError::Remote(ConnectionErrorIncoming::ApplicationClose{
                error_code: code,
                ..
            }) if code == Code::H3_NO_ERROR.value()
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        assert!(incoming.accept().await.unwrap().is_none());
    };

    tokio::join!(server_fut, client_fut);
}

#[tokio::test]
// This test is to ensure that the server does still process requests even if a stream is started
// but has not sent any data
async fn server_not_blocking_on_idle_request() {
    init_tracing();
    let mut pair = Pair::default();
    let mut server = pair.server();

    let client_fut = async {
        // create a Connection
        let connection = pair.client_inner().await;
        let mut control_stream = connection.open_uni().await.unwrap();

        let mut buf = BytesMut::new();
        StreamType::CONTROL.encode(&mut buf);

        Frame::<Bytes>::Settings(Settings::default()).encode(&mut buf);
        control_stream.write_all(&buf[..]).await.unwrap();

        let mut control_recv = connection.accept_uni().await.unwrap();
        // create a Request stream which is idle
        let mut request_stream = connection.open_bi().await.unwrap();

        let mut buf = BytesMut::new();
        Frame::<Bytes>::headers(Bytes::from("test")).encode(&mut buf);
        request_stream.0.write_all(&buf[..]).await.unwrap();

        let mut buf = BytesMut::new();
        // send a wrong frame to control stream
        Frame::<Bytes>::Data(Bytes::from(
            "this frame should cause the server to respond with an error",
        ))
        .encode(&mut buf);
        tokio::time::sleep(Duration::from_millis(10)).await;

        control_stream.write_all(&buf[..]).await.unwrap();

        let mut buf2 = BytesMut::new();
        control_recv.read(buf2.as_mut()).await.unwrap();

        // no bidirectional stream is started by the server
        // this will fail when server sends the error
        let err = connection
            .accept_bi()
            .await
            .expect_err("connection should error after sending wrong data on control stream");

        assert_matches!(err,
        quinn::ConnectionError::ApplicationClosed(quinn::ApplicationClose { error_code, .. })
            if error_code.into_inner() == Code::H3_FRAME_UNEXPECTED.value()
        );
    };

    let server_fut = async {
        let conn = server.next().await;
        let mut incoming = server::Connection::new(conn).await.unwrap();
        let resolver = incoming.accept().await.unwrap().unwrap();
        let req1 = async move {
            let _ = resolver
                .resolve_request()
                .await
                .err()
                .expect("server should close connection");
        };

        let server = async move {
            let err = incoming.accept().await.err().expect("Connection Error");
            assert_matches!(
                err,
                ConnectionError::Local {
                    error: LocalError::Application {
                        code: Code::H3_FRAME_UNEXPECTED,
                        ..
                    }
                }
            );
        };

        tokio::join!(req1, server);
    };

    let join = async {
        tokio::join!(server_fut, client_fut);
    };

    tokio::select!(
        _ = join => (),
         _ = tokio::time::sleep(Duration::from_secs(100)) => panic!("timeout")
    );
}
async fn request<T, O, B>(mut send_request: T) -> Result<Response<()>, StreamError>
where
    T: BorrowMut<SendRequest<O, B>>,
    O: quic::OpenStreams<B>,
    B: Buf,
{
    let mut request_stream = send_request
        .borrow_mut()
        .send_request(Request::get("http://no.way").body(()).unwrap())
        .await?;
    request_stream.recv_response().await
}

async fn response<S, B>(mut stream: server::RequestStream<S, B>)
where
    S: quic::RecvStream + SendStream<B>,
    B: Buf,
{
    stream
        .send_response(
            Response::builder()
                .status(StatusCode::IM_A_TEAPOT)
                .body(())
                .unwrap(),
        )
        .await
        .unwrap();
    stream.finish().await.unwrap();
}