diff --git a/.github/workflows/codecov.yml b/.github/workflows/codecov.yml index 8a9ddab56..6ff4e30ed 100644 --- a/.github/workflows/codecov.yml +++ b/.github/workflows/codecov.yml @@ -11,7 +11,7 @@ jobs: os: [ubuntu-latest, macos-latest, windows-latest] runs-on: ${{ matrix.os }} steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@stable - uses: taiki-e/install-action@cargo-llvm-cov - shell: bash @@ -27,7 +27,7 @@ jobs: # | paste -sd ',' - run: | cargo llvm-cov \ - --features="arbitrary,async-io,aws-lc-rs,bloom,direct-log,fast-apple-datapath,futures-io,json-output,lock_tracking,log,platform-verifier,qlog,ring,runtime-smol,runtime-tokio,rustls,rustls-aws-lc-rs,rustls-log,rustls-ring,serde,serde_json,smol,tracing" \ + --features="arbitrary,async-io,aws-lc-rs,bloom,log,fast-apple-datapath,futures-io,json-output,lock_tracking,tracing-log,platform-verifier,qlog,ring,runtime-smol,runtime-tokio,rustls,rustls-aws-lc-rs,rustls-log,rustls-ring,serde,serde_json,smol,tracing" \ --workspace --lcov --output-path lcov.info - name: Upload coverage to Codecov uses: codecov/codecov-action@v5 diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 04a34d4eb..70272e547 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -6,99 +6,74 @@ on: pull_request: jobs: -# test-freebsd: -# # see https://github.com/actions/runner/issues/385 -# # use https://github.com/vmactions/freebsd-vm for now -# name: test on freebsd -# runs-on: ubuntu-latest -# steps: -# - uses: actions/checkout@v5 -# - name: test on freebsd -# uses: vmactions/freebsd-vm@v1 -# with: -# usesh: true -# mem: 4096 -# copyback: false -# prepare: | -# pkg install -y curl -# curl https://sh.rustup.rs -sSf --output rustup.sh -# sh rustup.sh -y --profile minimal --default-toolchain stable -# run: | -# export PATH="$HOME/.cargo/bin:$PATH" -# echo "===== rustc --version =====" -# rustc --version -# echo "===== freebsd-version =====" -# freebsd-version + test-freebsd: + # see https://github.com/actions/runner/issues/385 + # use https://github.com/vmactions/freebsd-vm for now + name: test on freebsd + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - name: test on freebsd + uses: vmactions/freebsd-vm@v1 + with: + usesh: true + mem: 4096 + copyback: false + prepare: | + pkg install -y curl + curl https://sh.rustup.rs -sSf --output rustup.sh + sh rustup.sh -y --profile minimal --default-toolchain stable + run: | + export PATH="$HOME/.cargo/bin:$PATH" + echo "===== rustc --version =====" + rustc --version + echo "===== freebsd-version =====" + freebsd-version # cargo build --locked --all-targets && cargo test --locked && cargo test --locked -- --ignored stress && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p iroh-quinn-udp --benches - # test-netbsd: - # name: test on netbsd - # runs-on: ubuntu-latest - # steps: - # - uses: actions/checkout@v5 - # - name: test on netbsd - # uses: vmactions/netbsd-vm@v1 - # with: - # usesh: true - # mem: 4096 - # copyback: false - # prepare: | - # export PATH="/usr/sbin:/sbin:$PATH" - # pkg_add curl - # curl https://sh.rustup.rs -sSf --output rustup.sh - # sh rustup.sh -y --profile minimal --default-toolchain stable - # run: | - # export PATH="$HOME/.cargo/bin:$PATH" - # echo "===== rustc --version =====" - # rustc --version - # echo "===== uname -a =====" - # uname -a + test-solaris: + name: test on solaris + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - name: test on Solaris + uses: vmactions/solaris-vm@v1 + with: + release: "11.4-gcc" + usesh: true + mem: 4096 + copyback: false + prepare: | + source <(curl -s https://raw.githubusercontent.com/psumbera/solaris-rust/refs/heads/main/sh.rust-web-install) + echo "~~~~ rustc --version ~~~~" + rustc --version + echo "~~~~ Solaris-version ~~~~" + uname -a + # Unlike others, don't un-ignore stress tests, because they hang on Solaris + run: | + export PATH=$HOME/.rust_solaris/bin:$PATH + # Workaround for https://github.com/quinn-rs/quinn/issues/2218 + export CARGO_HTTP_MULTIPLEXING=false + cargo build --locked --all-targets && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p iroh-quinn-udp --benches - # cargo build --locked --all-targets && cargo test --locked && cargo test --locked -- --ignored stress && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p iroh-quinn-udp --benches - - # test-solaris: - # name: test on solaris - # runs-on: ubuntu-latest - # steps: - # - uses: actions/checkout@v5 - # - name: test on Solaris - # uses: vmactions/solaris-vm@v1 - # with: - # release: "11.4-gcc" - # usesh: true - # mem: 4096 - # copyback: false - # prepare: | - # source <(curl -s https://raw.githubusercontent.com/psumbera/solaris-rust/refs/heads/main/sh.rust-web-install) - # echo "~~~~ rustc --version ~~~~" - # rustc --version - # echo "~~~~ Solaris-version ~~~~" - # uname -a - # # Unlike others, don't un-ignore stress tests, because they hang on Solaris - # run: | - # export PATH=$HOME/.rust_solaris/bin:$PATH - # # Workaround for https://github.com/quinn-rs/quinn/issues/2218 - # export CARGO_HTTP_MULTIPLEXING=false - # cargo build --locked --all-targets && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p quinn-udp --benches - # - # test-illumos: - # name: test on illumos - # runs-on: ubuntu-latest - # steps: - # - uses: actions/checkout@v5 - # - name: test on Illumos - # uses: vmactions/omnios-vm@v1 - # with: - # usesh: true - # mem: 4096 - # copyback: false - # prepare: | - # pkg install gcc14 curl pkg-config glib2 - # curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal - # run: | - # . "$HOME/.cargo/env" - # cargo build --locked --all-targets && cargo test --locked && cargo test --locked -- --ignored stress && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p iroh-quinn-udp --benches + test-illumos: + name: test on illumos + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v6 + - name: test on Illumos + uses: vmactions/omnios-vm@v1 + with: + usesh: true + mem: 4096 + copyback: false + prepare: | + pkg install gcc14 curl pkg-config glib2 + curl --proto '=https' --tlsv1.2 -sSf https://sh.rustup.rs | sh -s -- -y --profile minimal + run: | + . "$HOME/.cargo/env" + cargo build --locked --all-targets && cargo test --locked && cargo test --locked -- --ignored stress && cargo test --locked --manifest-path fuzz/Cargo.toml && cargo test --locked -p iroh-quinn-udp --benches test: strategy: @@ -119,7 +94,7 @@ jobs: SCCACHE_GHA_ENABLED: "on" steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: mozilla-actions/sccache-action@v0.0.9 - uses: dtolnay/rust-toolchain@master with: @@ -139,24 +114,26 @@ jobs: test-aws-lc-rs: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 # Prevent feature unification from selecting *ring* as the crypto provider - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn-proto/Cargo.toml --no-default-features --features rustls-aws-lc-rs - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn/Cargo.toml --no-default-features --features rustls-aws-lc-rs,runtime-tokio + - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn/Cargo.toml --no-default-features --features rustls-aws-lc-rs,runtime-tokio,__rustls-post-quantum-test # FIPS - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn-proto/Cargo.toml --no-default-features --features rustls-aws-lc-rs-fips - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn/Cargo.toml --no-default-features --features rustls-aws-lc-rs-fips,runtime-tokio + - run: RUST_BACKTRACE=1 cargo test --locked --manifest-path quinn/Cargo.toml --no-default-features --features rustls-aws-lc-rs-fips,__rustls-post-quantum-test,runtime-tokio wasm_test: name: test wasm32-unknown-unknown runs-on: ubuntu-latest steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: dtolnay/rust-toolchain@stable - run: rustup target add wasm32-unknown-unknown - - uses: actions/setup-node@v4 + - uses: actions/setup-node@v6 with: node-version: 20 - uses: bytecodealliance/actions/wasm-tools/setup@v1 @@ -164,7 +141,7 @@ jobs: - run: cargo test --locked -p iroh-quinn-proto --target wasm32-unknown-unknown --no-run - run: cargo check --locked -p iroh-quinn-udp --target wasm32-unknown-unknown --no-default-features --features=tracing,log - - run: cargo rustc --locked -p iroh-quinn --target wasm32-unknown-unknown --no-default-features --features=log,platform-verifier,rustls-ring --crate-type=cdylib + - run: cargo rustc --locked -p iroh-quinn --target wasm32-unknown-unknown --no-default-features --features=tracing-log,platform-verifier,rustls-ring --crate-type=cdylib # If the Wasm file contains any 'import "env"' declarations, then # some non-Wasm-compatible code made it into the final code. @@ -175,7 +152,8 @@ jobs: run: | ! wasm-tools print --skeleton target/wasm32-unknown-unknown/debug/iroh_quinn.wasm | grep 'import "env"' - - run: cargo binstall wasm-bindgen-cli --locked --no-confirm + # Match the version of wasm-bindgen used in `Cargo.lock` + - run: cargo binstall wasm-bindgen-cli@0.2.106 --locked --no-confirm - run: cargo test --locked -p iroh-quinn-proto --target wasm32-unknown-unknown msrv: @@ -184,7 +162,7 @@ jobs: RUSTC_WRAPPER: "sccache" SCCACHE_GHA_ENABLED: "on" steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: mozilla-actions/sccache-action@v0.0.9 # Note that we must also update the README when changing the MSRV - uses: dtolnay/rust-toolchain@1.83 @@ -197,7 +175,7 @@ jobs: RUSTC_WRAPPER: "sccache" SCCACHE_GHA_ENABLED: "on" steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: mozilla-actions/sccache-action@v0.0.9 - uses: dtolnay/rust-toolchain@stable with: @@ -220,7 +198,7 @@ jobs: audit: runs-on: ubuntu-latest steps: - - uses: actions/checkout@v5 + - uses: actions/checkout@v6 - uses: EmbarkStudios/cargo-deny-action@v2 test-android: @@ -242,7 +220,7 @@ jobs: run: echo "API_LEVEL=${{ matrix.api-level }}" >> $GITHUB_ENV - name: Checkout code - uses: actions/checkout@v5 + uses: actions/checkout@v6 - name: Install JDK uses: actions/setup-java@v5 @@ -284,17 +262,18 @@ jobs: api-level: ${{ matrix.api-level }} arch: ${{ matrix.emulator-arch }} script: .github/workflows/rust-android-run-tests-on-emulator.sh - # features: - # strategy: - # matrix: - # os: [ubuntu-latest, macos-latest, windows-latest] - # runs-on: ${{ matrix.os }} - # env: - # RUSTFLAGS: -Dwarnings - # # skip FIPS features outside of Linux - # SKIP_FEATURES: ${{ matrix.os != 'ubuntu-latest' && 'rustls-aws-lc-rs-fips,aws-lc-rs-fips' || '' }} - # steps: - # - uses: actions/checkout@v5 - # - uses: dtolnay/rust-toolchain@stable - # - uses: taiki-e/install-action@cargo-hack - # - run: cargo hack check --feature-powerset --depth 3 --optional-deps --no-dev-deps --ignore-private --skip "${{env.SKIP_FEATURES}}" + + features: + strategy: + matrix: + os: [ubuntu-latest, macos-latest, windows-latest] + runs-on: ${{ matrix.os }} + env: + RUSTFLAGS: -Dwarnings + # skip FIPS features outside of Linux + SKIP_FEATURES: ${{ matrix.os != 'ubuntu-latest' && 'rustls-aws-lc-rs-fips,aws-lc-rs-fips,__rustls-post-quantum-test' || '' }} + steps: + - uses: actions/checkout@v6 + - uses: dtolnay/rust-toolchain@stable + - uses: taiki-e/install-action@cargo-hack + - run: cargo hack check --feature-powerset --depth 3 --optional-deps --no-dev-deps --ignore-private --skip "${{env.SKIP_FEATURES}}" diff --git a/Cargo.lock b/Cargo.lock index 2994a25fd..4f59ee49e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -300,6 +300,49 @@ dependencies = [ "libloading", ] +[[package]] +name = "axum" +version = "0.8.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5b098575ebe77cb6d14fc7f32749631a6e44edbef6b796f89b020e99ba20d425" +dependencies = [ + "axum-core", + "bytes", + "futures-util", + "http", + "http-body", + "http-body-util", + "itoa", + "matchit", + "memchr", + "mime", + "percent-encoding", + "pin-project-lite", + "serde_core", + "sync_wrapper", + "tower", + "tower-layer", + "tower-service", +] + +[[package]] +name = "axum-core" +version = "0.5.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "59446ce19cd142f8833f856eb31f3eb097812d1479ab224f54d72428ca21ea22" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "http-body-util", + "mime", + "pin-project-lite", + "sync_wrapper", + "tower-layer", + "tower-service", +] + [[package]] name = "backtrace" version = "0.3.76" @@ -315,6 +358,12 @@ dependencies = [ "windows-link", ] +[[package]] +name = "base64" +version = "0.21.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9d297deb1925b89f2ccc13d7635fa0714f12c87adce1c75356b39ca9b7178567" + [[package]] name = "base64" version = "0.22.1" @@ -568,6 +617,47 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "console-api" +version = "0.9.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e8599749b6667e2f0c910c1d0dff6901163ff698a52d5a39720f61b5be4b20d3" +dependencies = [ + "futures-core", + "prost", + "prost-types", + "tonic", + "tonic-prost", + "tracing-core", +] + +[[package]] +name = "console-subscriber" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb4915b7d8dd960457a1b6c380114c2944f728e7c65294ab247ae6b6f1f37592" +dependencies = [ + "console-api", + "crossbeam-channel", + "crossbeam-utils", + "futures-task", + "hdrhistogram", + "humantime", + "hyper-util", + "parking_lot", + "prost", + "prost-types", + "serde", + "serde_json", + "thread_local", + "tokio", + "tokio-stream", + "tonic", + "tracing", + "tracing-core", + "tracing-subscriber", +] + [[package]] name = "core-foundation" version = "0.10.1" @@ -599,6 +689,15 @@ version = "2.4.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "19d374276b40fb8bbdee95aef7c7fa6b5316ec764510eb64b8dd0e2ed0d7e7f5" +[[package]] +name = "crc32fast" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9481c1c90cbf2ac953f07c8d4a58aa3945c425b7185c9154d67a65e4230da511" +dependencies = [ + "cfg-if", +] + [[package]] name = "criterion" version = "0.7.0" @@ -631,6 +730,15 @@ dependencies = [ "itertools", ] +[[package]] +name = "crossbeam-channel" +version = "0.5.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "82b8f8f868b36967f9606790d1903570de9ceaf870a7bf9fbbd3016d636a2cb2" +dependencies = [ + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.21" @@ -755,7 +863,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb" dependencies = [ "libc", - "windows-sys 0.52.0", + "windows-sys 0.61.1", ] [[package]] @@ -803,6 +911,16 @@ version = "0.1.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1ced73b1dacfc750a6db6c0a0c3a3853c8b41997e2e2c563dc90804ae6867959" +[[package]] +name = "flate2" +version = "1.1.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfe33edd8e85a12a67454e37f8c75e730830d83e313556ab9ebf9ee7fbeb3bfb" +dependencies = [ + "crc32fast", + "miniz_oxide", +] + [[package]] name = "fnv" version = "1.0.7" @@ -824,6 +942,15 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "42703706b716c37f96a77aea830392ad231f44c9e9a67872fa5548707e11b11c" +[[package]] +name = "futures-channel" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2dff15bf788c671c1934e366d07e30c1814a8ef514e1af724a602e8a2fbe1b10" +dependencies = [ + "futures-core", +] + [[package]] name = "futures-core" version = "0.3.31" @@ -855,6 +982,24 @@ version = "0.3.31" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e575fab7d1e0dcb8d0c7bcf9a63ee213816ab51902e6d244a95819acacf1d4f7" +[[package]] +name = "futures-task" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f90f7dce0722e95104fcb095585910c0977252f286e354b5e3bd38902cd99988" + +[[package]] +name = "futures-util" +version = "0.3.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9fa08315bb612088cc391249efdc3bc77536f16c91f6cf495e6fbe85b20a4a81" +dependencies = [ + "futures-core", + "futures-task", + "pin-project-lite", + "pin-utils", +] + [[package]] name = "fuzz" version = "0.1.0" @@ -903,6 +1048,25 @@ version = "0.3.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0cc23270f6e1808e30a928bdc84dea0b9b4136a8bc82338574f23baf47bbd280" +[[package]] +name = "h2" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f3c0b69cfcb4e1b9f1bf2f53f95f766e4661169728ec61cd3fe5a0166f2d1386" +dependencies = [ + "atomic-waker", + "bytes", + "fnv", + "futures-core", + "futures-sink", + "http", + "indexmap", + "slab", + "tokio", + "tokio-util", + "tracing", +] + [[package]] name = "half" version = "2.6.0" @@ -925,7 +1089,10 @@ version = "7.5.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "765c9198f173dd59ce26ff9f95ef0aafd0a0fe01fb9d72841bc5066a4c06511d" dependencies = [ + "base64 0.21.7", "byteorder", + "flate2", + "nom", "num-traits", ] @@ -947,6 +1114,114 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bcaaec4551594c969335c98c903c1397853d4198408ea609190f420500f6be71" +[[package]] +name = "http" +version = "1.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e3ba2a386d7f85a81f119ad7498ebe444d2e22c2af0b86b069416ace48b3311a" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "http-body" +version = "1.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1efedce1fb8e6913f23e0c92de8e62cd5b772a67e7b3946df930a62566c93184" +dependencies = [ + "bytes", + "http", +] + +[[package]] +name = "http-body-util" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b021d93e26becf5dc7e1b75b1bed1fd93124b374ceb73f43d4d4eafec896a64a" +dependencies = [ + "bytes", + "futures-core", + "http", + "http-body", + "pin-project-lite", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "httpdate" +version = "1.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df3b46402a9d5adb4c86a0cf463f42e19994e3ee891101b1841f30a545cb49a9" + +[[package]] +name = "humantime" +version = "2.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "135b12329e5e3ce057a9f972339ea52bc954fe1e9358ef27f95e89716fbc5424" + +[[package]] +name = "hyper" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ab2d4f250c3d7b1c9fcdff1cece94ea4e2dfbec68614f7b87cb205f24ca9d11" +dependencies = [ + "atomic-waker", + "bytes", + "futures-channel", + "futures-core", + "h2", + "http", + "http-body", + "httparse", + "httpdate", + "itoa", + "pin-project-lite", + "pin-utils", + "smallvec", + "tokio", + "want", +] + +[[package]] +name = "hyper-timeout" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2b90d566bffbce6a75bd8b09a05aa8c2cb1fabb6cb348f8840c9e4c90a0d83b0" +dependencies = [ + "hyper", + "hyper-util", + "pin-project-lite", + "tokio", + "tower-service", +] + +[[package]] +name = "hyper-util" +version = "0.1.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "52e9a2a24dc5c6821e71a7030e1e14b7b632acac55c40e9d2e082c621261bb56" +dependencies = [ + "bytes", + "futures-channel", + "futures-core", + "futures-util", + "http", + "http-body", + "hyper", + "libc", + "pin-project-lite", + "socket2", + "tokio", + "tower-service", + "tracing", +] + [[package]] name = "icu_collections" version = "2.0.0" @@ -1110,7 +1385,6 @@ dependencies = [ "rcgen", "rustc-hash", "rustls", - "rustls-pemfile", "smol", "socket2", "thiserror 2.0.17", @@ -1135,7 +1409,6 @@ dependencies = [ "getrandom 0.3.3", "hex-literal", "identity-hash", - "lazy_static", "lru-slab", "qlog", "rand", @@ -1157,17 +1430,16 @@ dependencies = [ [[package]] name = "iroh-quinn-udp" -version = "0.5.12" +version = "0.6.0" dependencies = [ "cfg_aliases", "criterion", "libc", "log", - "once_cell", "socket2", "tokio", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.61.1", ] [[package]] @@ -1293,6 +1565,15 @@ version = "0.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "241eaef5fd12c88705a01fc1066c48c4b36e0dd4377dcdc7ec3942cea7a69956" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.28" @@ -1314,12 +1595,24 @@ dependencies = [ "regex-automata", ] +[[package]] +name = "matchit" +version = "0.8.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "47e1ffaa40ddd1f3ed91f717a33c8c0ee23fff369e3aa8772b9605cc1d22f4c3" + [[package]] name = "memchr" version = "2.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f52b00d39961fc5b2736ea853c9cc86238e165017a493d1d5c8eac6bdc4cc273" +[[package]] +name = "mime" +version = "0.3.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6877bb514081ee2a7ff5ef9de3281f14a4dd4bceac4c09388074a6b5df8a139a" + [[package]] name = "minicov" version = "0.3.7" @@ -1343,6 +1636,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fa76a2c86f704bdb222d66965fb3d63269ce38518b83cb0575fca855ebb6316" dependencies = [ "adler2", + "simd-adler32", ] [[package]] @@ -1439,13 +1733,36 @@ version = "2.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba" +[[package]] +name = "parking_lot" +version = "0.12.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a" +dependencies = [ + "lock_api", + "parking_lot_core", +] + +[[package]] +name = "parking_lot_core" +version = "0.9.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1" +dependencies = [ + "cfg-if", + "libc", + "redox_syscall", + "smallvec", + "windows-link", +] + [[package]] name = "pem" version = "3.0.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "38af38e8470ac9dee3ce1bae1af9c1671fffc44ddfd8bd1d0a3445bf349a8ef3" dependencies = [ - "base64", + "base64 0.22.1", "serde", ] @@ -1462,12 +1779,12 @@ dependencies = [ "anyhow", "bytes", "clap", + "console-subscriber", "hdrhistogram", "iroh-quinn", "iroh-quinn-proto", "rcgen", "rustls", - "rustls-pemfile", "serde", "serde_json", "socket2", @@ -1502,6 +1819,12 @@ version = "0.2.16" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "3b3cff922bd51709b605d9ead9aa71031d81447142d828eb4a6eba76fe619f9b" +[[package]] +name = "pin-utils" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184" + [[package]] name = "piper" version = "0.2.4" @@ -1570,6 +1893,38 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "prost" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "101fec8d036f8d9d4a1e8ebf90d566d1d798f3b1aa379d2576a54a0d9acea5bd" +dependencies = [ + "bytes", + "prost-derive", +] + +[[package]] +name = "prost-derive" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2d93e596a829ebe00afa41c3a056e6308d6b8a4c7d869edf184e2c91b1ba564" +dependencies = [ + "anyhow", + "itertools", + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "prost-types" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f5d7b7346e150de32340ae3390b8b3ffa37ad93ec31fb5dad86afe817619e4e7" +dependencies = [ + "prost", +] + [[package]] name = "qlog" version = "0.15.2" @@ -1648,6 +2003,15 @@ dependencies = [ "yasna", ] +[[package]] +name = "redox_syscall" +version = "0.5.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d" +dependencies = [ + "bitflags", +] + [[package]] name = "redox_users" version = "0.4.6" @@ -1724,7 +2088,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys", - "windows-sys 0.52.0", + "windows-sys 0.61.1", ] [[package]] @@ -1755,15 +2119,6 @@ dependencies = [ "security-framework", ] -[[package]] -name = "rustls-pemfile" -version = "2.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "dce314e5fee3f39953d46bb63bb8a46d40c2f8fb7cc5a3b6cab2bde9721d6e50" -dependencies = [ - "rustls-pki-types", -] - [[package]] name = "rustls-pki-types" version = "1.12.0" @@ -1843,6 +2198,12 @@ dependencies = [ "windows-sys 0.61.1", ] +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "security-framework" version = "3.5.1" @@ -1957,6 +2318,12 @@ dependencies = [ "libc", ] +[[package]] +name = "simd-adler32" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d66dc143e6b11c1eddc06d5c423cfc97062865baf299914ab64caa38182078fe" + [[package]] name = "siphasher" version = "1.0.1" @@ -2034,6 +2401,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "sync_wrapper" +version = "1.0.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0bf256ce5efdfa370213c1dabab5935a12e49f2c58d15e9eac2870d3b4f27263" + [[package]] name = "synstructure" version = "0.13.2" @@ -2169,6 +2542,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "89e49afdadebb872d3145a5638b59eb0691ea23e46ca484037cfab3b76b95038" dependencies = [ "backtrace", + "bytes", "io-uring", "libc", "mio", @@ -2177,6 +2551,7 @@ dependencies = [ "slab", "socket2", "tokio-macros", + "tracing", "windows-sys 0.59.0", ] @@ -2216,6 +2591,77 @@ dependencies = [ "tokio", ] +[[package]] +name = "tonic" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "eb7613188ce9f7df5bfe185db26c5814347d110db17920415cf2fbcad85e7203" +dependencies = [ + "async-trait", + "axum", + "base64 0.22.1", + "bytes", + "h2", + "http", + "http-body", + "http-body-util", + "hyper", + "hyper-timeout", + "hyper-util", + "percent-encoding", + "pin-project", + "socket2", + "sync_wrapper", + "tokio", + "tokio-stream", + "tower", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tonic-prost" +version = "0.14.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "66bd50ad6ce1252d87ef024b3d64fe4c3cf54a86fb9ef4c631fdd0ded7aeaa67" +dependencies = [ + "bytes", + "prost", + "tonic", +] + +[[package]] +name = "tower" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d039ad9159c98b70ecfd540b2573b97f7f52c3e8d9f8ad57a24b916a536975f9" +dependencies = [ + "futures-core", + "futures-util", + "indexmap", + "pin-project-lite", + "slab", + "sync_wrapper", + "tokio", + "tokio-util", + "tower-layer", + "tower-service", + "tracing", +] + +[[package]] +name = "tower-layer" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "121c2a6cda46980bb0fcd1647ffaf6cd3fc79a013de288782836f6df9c48780e" + +[[package]] +name = "tower-service" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8df9b6e13f2d32c91b9bd719c00d1958837bc7dec474d94952798cc8e69eeec3" + [[package]] name = "tracing" version = "0.1.41" @@ -2224,9 +2670,21 @@ checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0" dependencies = [ "log", "pin-project-lite", + "tracing-attributes", "tracing-core", ] +[[package]] +name = "tracing-attributes" +version = "0.1.31" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7490cfa5ec963746568740651ac6781f701c9c5ea257c58e057f3ba8cf69e8da" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "tracing-core" version = "0.1.34" @@ -2234,6 +2692,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678" dependencies = [ "once_cell", + "valuable", ] [[package]] @@ -2255,6 +2714,7 @@ dependencies = [ "matchers", "nu-ansi-term", "once_cell", + "parking_lot", "regex-automata", "sharded-slab", "thread_local", @@ -2263,6 +2723,12 @@ dependencies = [ "tracing-core", ] +[[package]] +name = "try-lock" +version = "0.2.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e421abadd41a4225275504ea4d6566923418b7f05506fbc9c0fe86ba7396114b" + [[package]] name = "unicode-ident" version = "1.0.19" @@ -2299,6 +2765,12 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "valuable" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" + [[package]] name = "walkdir" version = "2.5.0" @@ -2309,6 +2781,15 @@ dependencies = [ "winapi-util", ] +[[package]] +name = "want" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa7760aed19e106de2c7c0b581b509f2f25d3dacaf737cb82ac61bc6d760b0e" +dependencies = [ + "try-lock", +] + [[package]] name = "wasi" version = "0.11.1+wasi-snapshot-preview1" @@ -2474,7 +2955,7 @@ version = "0.1.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22" dependencies = [ - "windows-sys 0.52.0", + "windows-sys 0.61.1", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index f04499ce2..99b8b9695 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,9 +29,7 @@ hdrhistogram = { version = "7.2", default-features = false } hex-literal = "1" identity-hash = "0.1.0" lru-slab = "0.1.2" -lazy_static = "1" -log = "0.4.22" -once_cell = "1.19" +log = "0.4" pin-project-lite = "0.2" qlog = "0.15.2" rand = "0.9" @@ -56,7 +54,7 @@ tracing-subscriber = { version = "0.3.1", default-features = false, features = [ url = "2" wasm-bindgen-test = { version = "0.3.45" } web-time = "1" -windows-sys = { version = ">=0.52, <=0.60", features = ["Win32_Foundation", "Win32_System_IO", "Win32_Networking_WinSock"] } +windows-sys = { version = ">=0.52, <=0.61", features = ["Win32_Foundation", "Win32_System_IO", "Win32_Networking_WinSock"] } cfg_aliases = "0.2" # Fix minimal dependencies for indirect deps diff --git a/README.md b/README.md index 7ce8ce170..82e9038ed 100644 --- a/README.md +++ b/README.md @@ -10,7 +10,7 @@ Quinn is a pure-rust, async-compatible implementation of the IETF [rustls][rustls] and [*ring*][ring] - Application-layer datagrams for small, unreliable messages - Future-based async API -- Minimum supported Rust version of 1.83 +- Minimum supported Rust version of 1.83.0 This is a fork incorporating some changes for use in iroh. The aim is to contribute back any generally useful changes into upstream Quinn, diff --git a/perf/Cargo.toml b/perf/Cargo.toml index b8182680c..3d4c7e281 100644 --- a/perf/Cargo.toml +++ b/perf/Cargo.toml @@ -17,17 +17,23 @@ default = ["json-output", "qlog"] json-output = ["serde", "serde_json"] # Enable qlog support qlog = ["quinn/qlog"] +# Enable tokio-console https://github.com/tokio-rs/console +# See https://docs.rs/console-subscriber/latest/console_subscriber/struct.Builder.html#method.with_default_env +# for useful environment variables in particular to change listen addr and port +# WARNING: it requires special compilation flags +# RUSTFLAGS="--cfg tokio_unstable" cargo build -r -p perf -F tokio-console +tokio-console = ["console-subscriber"] [dependencies] anyhow = { workspace = true } bytes = { workspace = true } clap = { workspace = true } +console-subscriber = { version = "0.5.0", features = ["parking_lot"], optional = true } hdrhistogram = { workspace = true } quinn = { package = "iroh-quinn", path = "../quinn" } quinn-proto = { package = "iroh-quinn-proto", path = "../quinn-proto" } rcgen = { workspace = true } rustls = { workspace = true } -rustls-pemfile = { workspace = true } serde = { workspace = true, optional = true } serde_json = { workspace = true, optional = true } socket2 = { workspace = true } diff --git a/perf/src/bin/perf.rs b/perf/src/bin/perf.rs index 54c6ff797..221cc12aa 100644 --- a/perf/src/bin/perf.rs +++ b/perf/src/bin/perf.rs @@ -1,9 +1,35 @@ use clap::{Parser, Subcommand}; use tracing::error; -use tracing_subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt}; +use tracing_subscriber::{EnvFilter, Layer, fmt, layer::SubscriberExt, util::SubscriberInitExt}; use perf::{client, server}; +#[tokio::main(flavor = "current_thread")] +async fn main() { + let opt = Cli::parse(); + + let registry = tracing_subscriber::registry(); + #[cfg(feature = "tokio-console")] + let registry = registry.with(console_subscriber::spawn()); + registry + .with( + fmt::layer().with_filter( + EnvFilter::try_from_default_env() + .or_else(|_| EnvFilter::try_new("warn")) + .unwrap(), + ), + ) + .init(); + + let r = match opt.command { + Commands::Server(opt) => server::run(opt).await, + Commands::Client(opt) => client::run(opt).await, + }; + if let Err(e) = r { + error!("{:#}", e); + } +} + #[derive(Parser)] #[clap(long_about = None)] struct Cli { @@ -18,25 +44,3 @@ enum Commands { /// Run as a perf client Client(client::Opt), } - -#[tokio::main(flavor = "current_thread")] -async fn main() { - let opt = Cli::parse(); - - tracing_subscriber::registry() - .with( - EnvFilter::try_from_default_env() - .or_else(|_| EnvFilter::try_new("warn")) - .unwrap(), - ) - .with(fmt::layer()) - .init(); - - let r = match opt.command { - Commands::Server(opt) => server::run(opt).await, - Commands::Client(opt) => client::run(opt).await, - }; - if let Err(e) = r { - error!("{:#}", e); - } -} diff --git a/perf/src/noprotection.rs b/perf/src/noprotection.rs index 862b3eccc..a74cbff87 100644 --- a/perf/src/noprotection.rs +++ b/perf/src/noprotection.rs @@ -64,7 +64,7 @@ impl NoProtectionServerConfig { // forward all calls to inner except those related to packet encryption/decryption impl crypto::Session for NoProtectionSession { - fn initial_keys(&self, dst_cid: &ConnectionId, side: Side) -> crypto::Keys { + fn initial_keys(&self, dst_cid: ConnectionId, side: Side) -> crypto::Keys { self.inner.initial_keys(dst_cid, side) } @@ -115,7 +115,7 @@ impl crypto::Session for NoProtectionSession { Some(Self::wrap_packet_keys(keys)) } - fn is_valid_retry(&self, orig_dst_cid: &ConnectionId, header: &[u8], payload: &[u8]) -> bool { + fn is_valid_retry(&self, orig_dst_cid: ConnectionId, header: &[u8], payload: &[u8]) -> bool { self.inner.is_valid_retry(orig_dst_cid, header, payload) } @@ -149,12 +149,12 @@ impl crypto::ServerConfig for NoProtectionServerConfig { fn initial_keys( &self, version: u32, - dst_cid: &ConnectionId, + dst_cid: ConnectionId, ) -> Result { self.inner.initial_keys(version, dst_cid) } - fn retry_tag(&self, version: u32, orig_dst_cid: &ConnectionId, packet: &[u8]) -> [u8; 16] { + fn retry_tag(&self, version: u32, orig_dst_cid: ConnectionId, packet: &[u8]) -> [u8; 16] { self.inner.retry_tag(version, orig_dst_cid, packet) } diff --git a/perf/src/server.rs b/perf/src/server.rs index 040b942f7..18d1a4d8e 100644 --- a/perf/src/server.rs +++ b/perf/src/server.rs @@ -1,10 +1,10 @@ -use std::{fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; +use std::{net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; use anyhow::{Context, Result}; use bytes::Bytes; use clap::Parser; use quinn::{TokioRuntime, crypto::rustls::QuicServerConfig}; -use rustls::pki_types::{CertificateDer, PrivatePkcs8KeyDer}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject}; use tracing::{debug, error, info}; use crate::{CommonOpt, PERF_CIPHER_SUITES, noprotection::NoProtectionServerConfig}; @@ -28,20 +28,17 @@ pub struct Opt { pub async fn run(opt: Opt) -> Result<()> { let (key, cert) = match (&opt.key, &opt.cert) { - (Some(key), Some(cert)) => { - let key = fs::read(key).context("reading key")?; - let cert = fs::read(cert).expect("reading cert"); - ( - PrivatePkcs8KeyDer::from(key), - rustls_pemfile::certs(&mut cert.as_ref()) - .collect::>() - .context("parsing cert")?, - ) - } + (Some(key), Some(cert)) => ( + PrivateKeyDer::from_pem_file(key).context("reading private key")?, + CertificateDer::pem_file_iter(cert) + .context("reading certificate chain file")? + .collect::>() + .context("reading certificate chain")?, + ), _ => { let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); ( - PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()), + PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()).into(), vec![CertificateDer::from(cert.cert)], ) } @@ -57,7 +54,7 @@ pub async fn run(opt: Opt) -> Result<()> { .with_protocol_versions(&[&rustls::version::TLS13]) .unwrap() .with_no_client_auth() - .with_single_cert(cert, key.into()) + .with_single_cert(cert, key) .unwrap(); crypto.alpn_protocols = vec![b"perf".to_vec()]; diff --git a/quinn-proto/Cargo.toml b/quinn-proto/Cargo.toml index 9653f7ea7..0fb36ab5c 100644 --- a/quinn-proto/Cargo.toml +++ b/quinn-proto/Cargo.toml @@ -13,7 +13,7 @@ workspace = ".." [features] # NOTE: Please keep this in sync with the feature list in `.github/workflows/codecov.yml`, see # comment in that file for more information. -default = ["rustls-ring", "log", "bloom"] +default = ["rustls-ring", "tracing-log", "bloom"] aws-lc-rs = ["dep:aws-lc-rs", "aws-lc-rs?/aws-lc-sys", "aws-lc-rs?/prebuilt-nasm"] aws-lc-rs-fips = ["aws-lc-rs", "aws-lc-rs?/fips"] # Enables BloomTokenLog, and uses it by default @@ -30,12 +30,17 @@ ring = ["dep:ring"] # Provides `ClientConfig::with_platform_verifier()` convenience method platform-verifier = ["dep:rustls-platform-verifier"] # Configure `tracing` to log events via `log` if no `tracing` subscriber exists. -log = ["tracing/log"] +tracing-log = ["tracing/log"] # Enable rustls logging rustls-log = ["rustls?/logging"] # Enable qlog support qlog = ["dep:qlog"] +# Internal (PRIVATE!) features used to aid testing. +# Don't rely on these whatsoever. They may disappear at any time. + +__rustls-post-quantum-test = [] + [dependencies] arbitrary = { workspace = true, optional = true } aws-lc-rs = { workspace = true, optional = true } @@ -68,7 +73,6 @@ hex-literal = { workspace = true } rand_pcg = "0.9" rcgen = { workspace = true } tracing-subscriber = { workspace = true } -lazy_static = "1" wasm-bindgen-test = { workspace = true } [lints.rust] diff --git a/quinn-proto/src/cid_generator.rs b/quinn-proto/src/cid_generator.rs index e62415e8b..7af4eca84 100644 --- a/quinn-proto/src/cid_generator.rs +++ b/quinn-proto/src/cid_generator.rs @@ -20,7 +20,7 @@ pub trait ConnectionIdGenerator: Send + Sync { /// Quickly determine whether `cid` could have been generated by this generator /// /// False positives are permitted, but increase the cost of handling invalid packets. - fn validate(&self, _cid: &ConnectionId) -> Result<(), InvalidCid> { + fn validate(&self, _cid: ConnectionId) -> Result<(), InvalidCid> { Ok(()) } @@ -143,7 +143,7 @@ impl ConnectionIdGenerator for HashedConnectionIdGenerator { ConnectionId::new(&bytes_arr) } - fn validate(&self, cid: &ConnectionId) -> Result<(), InvalidCid> { + fn validate(&self, cid: ConnectionId) -> Result<(), InvalidCid> { let (nonce, signature) = cid.split_at(NONCE_LEN); let mut hasher = rustc_hash::FxHasher::default(); hasher.write_u64(self.key); @@ -175,6 +175,6 @@ mod tests { fn validate_keyed_cid() { let mut generator = HashedConnectionIdGenerator::new(); let cid = generator.generate_cid(); - generator.validate(&cid).unwrap(); + generator.validate(cid).unwrap(); } } diff --git a/quinn-proto/src/congestion.rs b/quinn-proto/src/congestion.rs index 391e4d19c..6b34e2854 100644 --- a/quinn-proto/src/congestion.rs +++ b/quinn-proto/src/congestion.rs @@ -56,9 +56,16 @@ pub trait Controller: Send + Sync { now: Instant, sent: Instant, is_persistent_congestion: bool, + is_ecn: bool, lost_bytes: u64, ); + /// Packets were incorrectly deemed lost + /// + /// This function is called when all packets that were deemed lost (for instance because + /// of packet reordering) are acknowledged after the congestion event was raised. + fn on_spurious_congestion_event(&mut self) {} + /// The known MTU for the current network path has been updated fn on_mtu_update(&mut self, new_mtu: u16); diff --git a/quinn-proto/src/congestion/bbr/mod.rs b/quinn-proto/src/congestion/bbr/mod.rs index 97fb3bffc..458716230 100644 --- a/quinn-proto/src/congestion/bbr/mod.rs +++ b/quinn-proto/src/congestion/bbr/mod.rs @@ -465,6 +465,7 @@ impl Controller for Bbr { _now: Instant, _sent: Instant, _is_persistent_congestion: bool, + _is_ecn: bool, lost_bytes: u64, ) { self.loss_state.lost_bytes += lost_bytes; diff --git a/quinn-proto/src/congestion/cubic.rs b/quinn-proto/src/congestion/cubic.rs index 1bd9c69de..08db3ddad 100644 --- a/quinn-proto/src/congestion/cubic.rs +++ b/quinn-proto/src/congestion/cubic.rs @@ -19,12 +19,27 @@ const C: f64 = 0.4; /// k, w_max are described in the RFC. #[derive(Debug, Default, Clone)] pub(super) struct State { + /// Time period that the cubic function takes to increase the window size to W_max. k: f64, + /// Congestion window size when the last congestion event occurred. w_max: f64, - // Store cwnd increment during congestion avoidance. + /// Congestion window increment stored during congestion avoidance. cwnd_inc: u64, + + /// Maximum number of bytes in flight that may be sent. + window: u64, + + /// Slow start threshold in bytes. + /// + /// When the congestion window is below ssthresh, the mode is slow start + /// and the window grows by the number of bytes acknowledged. + ssthresh: u64, + + /// The time when QUIC first detects a loss, causing it to enter recovery. When a packet sent + /// after this time is acknowledged, QUIC exits recovery. + recovery_start_time: Option, } /// CUBIC Functions. @@ -39,7 +54,7 @@ impl State { (w_max * (1.0 - BETA_CUBIC) / C).cbrt() } - // W_cubic(t) = C * (t - K)^3 - w_max (Eq. 1) + // W_cubic(t) = C * (t - K)^3 + w_max (Eq. 1) fn w_cubic(&self, t: Duration, max_datagram_size: u64) -> f64 { let w_max = self.w_max / max_datagram_size as f64; @@ -60,28 +75,24 @@ impl State { #[derive(Debug, Clone)] pub struct Cubic { config: Arc, - /// Maximum number of bytes in flight that may be sent. - window: u64, - /// Slow start threshold in bytes. When the congestion window is below ssthresh, the mode is - /// slow start and the window grows by the number of bytes acknowledged. - ssthresh: u64, - /// The time when QUIC first detects a loss, causing it to enter recovery. When a packet sent - /// after this time is acknowledged, QUIC exits recovery. - recovery_start_time: Option, - cubic_state: State, current_mtu: u64, + state: State, + /// Copy of the controller state to restore when a spurious congestion event is detected. + pre_congestion_state: Option, } impl Cubic { /// Construct a state using the given `config` and current time `now` pub fn new(config: Arc, _now: Instant, current_mtu: u16) -> Self { Self { - window: config.initial_window, - ssthresh: u64::MAX, - recovery_start_time: None, - config, - cubic_state: Default::default(), + state: State { + window: config.initial_window, + ssthresh: u64::MAX, + ..Default::default() + }, current_mtu: current_mtu as u64, + pre_congestion_state: None, + config, } } @@ -101,6 +112,7 @@ impl Controller for Cubic { ) { if app_limited || self + .state .recovery_start_time .map(|recovery_start_time| sent <= recovery_start_time) .unwrap_or(false) @@ -108,35 +120,35 @@ impl Controller for Cubic { return; } - if self.window < self.ssthresh { + if self.state.window < self.state.ssthresh { // Slow start - self.window += bytes; + self.state.window += bytes; } else { // Congestion avoidance. let ca_start_time; - match self.recovery_start_time { + match self.state.recovery_start_time { Some(t) => ca_start_time = t, None => { // When we come here without congestion_event() triggered, // initialize congestion_recovery_start_time, w_max and k. ca_start_time = now; - self.recovery_start_time = Some(now); + self.state.recovery_start_time = Some(now); - self.cubic_state.w_max = self.window as f64; - self.cubic_state.k = 0.0; + self.state.w_max = self.state.window as f64; + self.state.k = 0.0; } } let t = now - ca_start_time; // w_cubic(t + rtt) - let w_cubic = self.cubic_state.w_cubic(t + rtt.get(), self.current_mtu); + let w_cubic = self.state.w_cubic(t + rtt.get(), self.current_mtu); // w_est(t) - let w_est = self.cubic_state.w_est(t, rtt.get(), self.current_mtu); + let w_est = self.state.w_est(t, rtt.get(), self.current_mtu); - let mut cubic_cwnd = self.window; + let mut cubic_cwnd = self.state.window; if w_cubic < w_est { // TCP friendly region. @@ -150,14 +162,14 @@ impl Controller for Cubic { } // Update the increment and increase cwnd by MSS. - self.cubic_state.cwnd_inc += cubic_cwnd - self.window; + self.state.cwnd_inc += cubic_cwnd - self.state.window; // cwnd_inc can be more than 1 MSS in the late stage of max probing. // however RFC9002 §7.3.3 (Congestion Avoidance) limits // the increase of cwnd to 1 max_datagram_size per cwnd acknowledged. - if self.cubic_state.cwnd_inc >= self.current_mtu { - self.window += self.current_mtu; - self.cubic_state.cwnd_inc = 0; + if self.state.cwnd_inc >= self.current_mtu { + self.state.window += self.current_mtu; + self.state.cwnd_inc = 0; } } } @@ -167,9 +179,11 @@ impl Controller for Cubic { now: Instant, sent: Instant, is_persistent_congestion: bool, + is_ecn: bool, _lost_bytes: u64, ) { if self + .state .recovery_start_time .map(|recovery_start_time| sent <= recovery_start_time) .unwrap_or(false) @@ -177,53 +191,66 @@ impl Controller for Cubic { return; } - self.recovery_start_time = Some(now); - - // Fast convergence - if (self.window as f64) < self.cubic_state.w_max { - self.cubic_state.w_max = self.window as f64 * (1.0 + BETA_CUBIC) / 2.0; - } else { - self.cubic_state.w_max = self.window as f64; + // Save state in case this event ends up being spurious + if !is_ecn { + self.pre_congestion_state = Some(self.state.clone()); } - self.ssthresh = cmp::max( - (self.cubic_state.w_max * BETA_CUBIC) as u64, + self.state.recovery_start_time = Some(now); + + // Fast convergence + if (self.state.window as f64) < self.state.w_max { + self.state.w_max = self.state.window as f64 * (1.0 + BETA_CUBIC) / 2.0; + } else { + self.state.w_max = self.state.window as f64; + } + + self.state.ssthresh = cmp::max( + (self.state.w_max * BETA_CUBIC) as u64, self.minimum_window(), ); - self.window = self.ssthresh; - self.cubic_state.k = self.cubic_state.cubic_k(self.current_mtu); + self.state.window = self.state.ssthresh; + self.state.k = self.state.cubic_k(self.current_mtu); - self.cubic_state.cwnd_inc = (self.cubic_state.cwnd_inc as f64 * BETA_CUBIC) as u64; + self.state.cwnd_inc = (self.state.cwnd_inc as f64 * BETA_CUBIC) as u64; if is_persistent_congestion { - self.recovery_start_time = None; - self.cubic_state.w_max = self.window as f64; + self.state.recovery_start_time = None; + self.state.w_max = self.state.window as f64; // 4.7 Timeout - reduce ssthresh based on BETA_CUBIC - self.ssthresh = cmp::max( - (self.window as f64 * BETA_CUBIC) as u64, + self.state.ssthresh = cmp::max( + (self.state.window as f64 * BETA_CUBIC) as u64, self.minimum_window(), ); - self.cubic_state.cwnd_inc = 0; + self.state.cwnd_inc = 0; - self.window = self.minimum_window(); + self.state.window = self.minimum_window(); + } + } + + fn on_spurious_congestion_event(&mut self) { + if let Some(prior_state) = self.pre_congestion_state.take() { + if self.state.window < prior_state.window { + self.state = prior_state; + } } } fn on_mtu_update(&mut self, new_mtu: u16) { self.current_mtu = new_mtu as u64; - self.window = self.window.max(self.minimum_window()); + self.state.window = self.state.window.max(self.minimum_window()); } fn window(&self) -> u64 { - self.window + self.state.window } fn metrics(&self) -> super::ControllerMetrics { super::ControllerMetrics { congestion_window: self.window(), - ssthresh: Some(self.ssthresh), + ssthresh: Some(self.state.ssthresh), pacing_rate: None, } } diff --git a/quinn-proto/src/congestion/new_reno.rs b/quinn-proto/src/congestion/new_reno.rs index 7bc61c666..90a6fa2c5 100644 --- a/quinn-proto/src/congestion/new_reno.rs +++ b/quinn-proto/src/congestion/new_reno.rs @@ -87,6 +87,7 @@ impl Controller for NewReno { now: Instant, sent: Instant, is_persistent_congestion: bool, + _is_ecn: bool, _lost_bytes: u64, ) { if sent <= self.recovery_start_time { diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs index aed6d4ef6..b1160a3bc 100644 --- a/quinn-proto/src/connection/mod.rs +++ b/quinn-proto/src/connection/mod.rs @@ -25,6 +25,7 @@ use crate::{ coding::BufMutExt, config::{ServerConfig, TransportConfig}, congestion::Controller, + connection::spaces::LostPacket, connection::timer::{ConnTimer, PathTimer}, crypto::{self, KeyPair, Keys, PacketKey}, frame::{self, Close, Datagram, FrameStruct, NewToken, ObservedAddr}, @@ -337,7 +338,7 @@ impl Connection { let mut rng = StdRng::from_seed(rng_seed); let initial_space = { let mut space = PacketSpace::new(now, SpaceId::Initial, &mut rng); - space.crypto = Some(crypto.initial_keys(&init_cid, side)); + space.crypto = Some(crypto.initial_keys(init_cid, side)); space }; let handshake_space = PacketSpace::new(now, SpaceId::Handshake, &mut rng); @@ -1074,7 +1075,7 @@ impl Connection { trace!( ?space_id, %path_id, - ?next_path_id, + %next_path_id, "nothing to send on path" ); path_id = *next_path_id; @@ -2288,6 +2289,12 @@ impl Connection { } }; + if self.detect_spurious_loss(&ack, space, path) { + self.path_data_mut(path) + .congestion + .on_spurious_congestion_event(); + } + // Avoid DoS from unreasonably huge ack ranges by filtering out just the new acks. let mut newly_acked = ArrayRangeSet::new(); for range in ack.iter() { @@ -2396,6 +2403,43 @@ impl Connection { Ok(()) } + fn detect_spurious_loss(&mut self, ack: &frame::Ack, space: SpaceId, path: PathId) -> bool { + let lost_packets = &mut self.spaces[space].for_path(path).lost_packets; + + if lost_packets.is_empty() { + return false; + } + + for range in ack.iter() { + let spurious_losses: Vec = lost_packets + .range(range.clone()) + .map(|(pn, _info)| pn) + .copied() + .collect(); + + for pn in spurious_losses { + lost_packets.remove(&pn); + } + } + + // If this ACK frame acknowledged all deemed lost packets, + // then we have raised a spurious congestion event in the past. + // We cannot conclude when there are remaining packets, + // but future ACK frames might indicate a spurious loss detection. + lost_packets.is_empty() + } + + /// Drain lost packets that we reasonably think will never arrive + /// + /// The current criterion is copied from `msquic`: + /// discard packets that were sent earlier than 2 probe timeouts ago. + fn drain_lost_packets(&mut self, now: Instant, space: SpaceId, path: PathId) { + let two_pto = 2 * self.path_data(path).rtt.pto_base(); + + let lost_packets = &mut self.spaces[space].for_path(path).lost_packets; + lost_packets.retain(|_pn, info| now.saturating_duration_since(info.time_sent) <= two_pto); + } + /// Process a new ECN block from an in-order ACK fn process_ecn( &mut self, @@ -2425,6 +2469,7 @@ impl Connection { now, largest_sent_time, false, + true, 0, ); } @@ -2581,8 +2626,9 @@ impl Connection { // InPersistentCongestion: Determine if all packets in the time period before the newest // lost packet, including the edges, are marked lost. PTO computation must always // include max ACK delay, i.e. operate as if in Data space (see RFC9001 §7.6.1). - let congestion_period = - self.pto(SpaceId::Data, path_id) * self.config.persistent_congestion_threshold; + let congestion_period = self + .pto(SpaceId::Data, path_id) + .saturating_mul(self.config.persistent_congestion_threshold); let mut persistent_congestion_start: Option = None; let mut prev_packet = None; let space = self.spaces[pn_space].for_path(path_id); @@ -2716,6 +2762,9 @@ impl Connection { }, "lost_packets must be sorted" ); + + self.drain_lost_packets(now, pn_space, path_id); + // OnPacketsLost if let Some(largest_lost) = lost_packets.last().cloned() { let old_bytes_in_flight = self.path_data_mut(path_id).in_flight.bytes; @@ -2731,17 +2780,6 @@ impl Connection { "packets lost", ); - // Packets sent before this time are deemed lost. - // We avoid computing this value above, since it's possible for this to panic - // if the `loss_delay` value internally stores a bigger `Duration` than the - // `Duration` that's stored inside the `Instant`, because some platforms may - // implement the `Instant` with a counter relative to system or even process - // startup (Wasm is one such case). - // If we're at this point, then it must be possible to have instants that are - // longer ago than `loss_delay` (see the `packet_too_old` computation - // above). - let lost_send_time = now.checked_sub(loss_delay).unwrap(); - for &packet in &lost_packets { let Some(info) = self.spaces[pn_space].for_path(path_id).take(packet) else { continue; @@ -2749,7 +2787,7 @@ impl Connection { self.config.qlog_sink.emit_packet_lost( packet, &info, - lost_send_time, + loss_delay, pn_space, now, self.orig_rem_cid, @@ -2758,6 +2796,7 @@ impl Connection { .get_mut(&path_id) .unwrap() .remove_in_flight(&info); + for frame in info.stream_frames { self.streams.retransmit(frame); } @@ -2765,6 +2804,13 @@ impl Connection { self.path_data_mut(path_id) .mtud .on_non_probe_lost(packet, info.size); + + self.spaces[pn_space].for_path(path_id).lost_packets.insert( + packet, + LostPacket { + time_sent: info.time_sent, + }, + ); } let path = self.path_data_mut(path_id); @@ -2792,6 +2838,7 @@ impl Connection { now, largest_lost_sent, in_persistent_congestion, + false, size_of_lost_packets, ); } @@ -3656,7 +3703,7 @@ impl Connection { .map(|cids| cids.active()) .map(|orig_dst_cid| { self.crypto.is_valid_retry( - &orig_dst_cid, + orig_dst_cid, &packet.header_data, &packet.payload, ) @@ -3695,7 +3742,7 @@ impl Connection { // any retransmitted Initials self.spaces[SpaceId::Initial] = { let mut space = PacketSpace::new(now, SpaceId::Initial, &mut self.rng); - space.crypto = Some(self.crypto.initial_keys(&rem_cid, self.side.side())); + space.crypto = Some(self.crypto.initial_keys(rem_cid, self.side.side())); space.crypto_offset = client_hello.len() as u64; space.for_path(path_id).next_packet_number = self.spaces[SpaceId::Initial] .for_path(path_id) @@ -3771,6 +3818,7 @@ impl Connection { code: TransportErrorCode::crypto(0x6d), frame: None, reason: "transport parameters missing".into(), + crypto: None, })?; if self.has_0rtt() { @@ -3809,6 +3857,8 @@ impl Connection { // Server-only self.spaces[SpaceId::Data].pending.handshake_done = true; self.discard_space(now, SpaceId::Handshake); + self.events.push_back(Event::HandshakeConfirmed); + trace!("handshake confirmed"); } self.events.push_back(Event::Connected); @@ -3859,6 +3909,7 @@ impl Connection { code: TransportErrorCode::crypto(0x6d), frame: None, reason: "transport parameters missing".into(), + crypto: None, })?; self.handle_peer_params(params, loc_cid, rem_cid)?; self.issue_first_cids(now); @@ -4332,6 +4383,8 @@ impl Connection { if self.spaces[SpaceId::Handshake].crypto.is_some() { self.discard_space(now, SpaceId::Handshake); } + self.events.push_back(Event::HandshakeConfirmed); + trace!("handshake confirmed"); } Frame::ObservedAddr(observed) => { // check if params allows the peer to send report and this node to receive it @@ -6301,6 +6354,8 @@ pub enum Event { HandshakeDataReady, /// The connection was successfully established Connected, + /// The TLS handshake was confirmed + HandshakeConfirmed, /// The connection was lost /// /// Emitted if the peer closes the connection or an error is encountered. diff --git a/quinn-proto/src/connection/mtud.rs b/quinn-proto/src/connection/mtud.rs index b690731b4..7e6fc1d8a 100644 --- a/quinn-proto/src/connection/mtud.rs +++ b/quinn-proto/src/connection/mtud.rs @@ -83,9 +83,14 @@ impl MtuDiscovery { self.current_mtu = self.current_mtu.min(peer_max_udp_payload_size); if let Some(state) = self.state.as_mut() { - // MTUD is only active after the connection has been fully established, so it is - // guaranteed we will receive the peer's transport parameters before we start probing - debug_assert!(matches!(state.phase, Phase::Initial)); + // It is possible for black hole detection to trigger before the connection has been + // fully established, if the initial MTU is greater the minimum MTU. We should never + // send probes before the connection has been fully established and we have received + // the peer's transport parameters though. + debug_assert!( + !matches!(state.phase, Phase::Searching(_)), + "Transport parameters received after MTU probing started" + ); state.peer_max_udp_payload_size = peer_max_udp_payload_size; } } @@ -450,9 +455,9 @@ impl BlackHoleDetector { }; // If a loss burst contains a packet smaller than the minimum MTU or a more recently // transmitted packet, it is not suspicious. - if burst.smallest_packet_size < self.min_mtu + if burst.smallest_packet_size <= self.min_mtu || (burst.latest_non_probe < self.largest_post_loss_packet - && burst.smallest_packet_size < self.acked_mtu) + && burst.smallest_packet_size <= self.acked_mtu) { return; } @@ -733,10 +738,14 @@ mod tests { #[cfg(debug_assertions)] #[test] - #[should_panic] - fn mtu_discovery_with_peer_max_udp_payload_size_after_search_panics() { + #[should_panic(expected = "Transport parameters received after MTU probing started")] + fn mtu_discovery_with_peer_max_udp_payload_size_during_search_panics() { let mut mtud = default_mtud(); - drive_to_completion(&mut mtud, Instant::now(), 1500); + assert!(mtud.poll_transmit(Instant::now(), 0).is_some()); + assert!(matches!( + mtud.state.as_ref().unwrap().phase, + Phase::Searching(_) + )); mtud.on_peer_max_udp_payload_size_received(1300); } diff --git a/quinn-proto/src/connection/pacing.rs b/quinn-proto/src/connection/pacing.rs index 2e469948c..495f0954f 100644 --- a/quinn-proto/src/connection/pacing.rs +++ b/quinn-proto/src/connection/pacing.rs @@ -89,13 +89,14 @@ impl Pacer { } let elapsed_rtts = time_elapsed.as_secs_f64() / smoothed_rtt.as_secs_f64(); - let new_tokens = window as f64 * 1.25 * elapsed_rtts; - self.tokens = self - .tokens - .saturating_add(new_tokens as _) - .min(self.capacity); + let new_tokens = (window as f64 * 1.25 * elapsed_rtts).round() as u64; + self.tokens = self.tokens.saturating_add(new_tokens).min(self.capacity); - self.prev = now; + // In the unlikely event that we're getting polled faster than tokens are generated, ensure + // that `elapsed_rtts` can grow until we make progress. + if new_tokens > 0 { + self.prev = now; + } // if we can already send a packet, there is no need for delay if self.tokens >= bytes_to_send { @@ -109,7 +110,7 @@ impl Pacer { // divisions come before multiplications to prevent overflow // this is the time at which the pacing window becomes empty - Some(self.prev + (unscaled_delay / 5) * 4) + Some(now + (unscaled_delay / 5) * 4) } } @@ -128,23 +129,34 @@ impl Pacer { /// Too long burst intervals make pacing less effective. fn optimal_capacity(smoothed_rtt: Duration, window: u64, mtu: u16) -> u64 { let rtt = smoothed_rtt.as_nanos().max(1); + let mtu = u64::from(mtu); - let capacity = ((window as u128 * BURST_INTERVAL_NANOS) / rtt) as u64; + let target_capacity = ((window as u128 * TARGET_BURST_INTERVAL.as_nanos()) / rtt) as u64; + // Never restrict capacity below one MTU. + let max_capacity = Ord::max( + ((window as u128 * MAX_BURST_INTERVAL.as_nanos()) / rtt) as u64, + mtu, + ); - // Small bursts are less efficient (no GSO), could increase latency and don't effectively - // use the channel's buffer capacity. Large bursts might block the connection on sending. - capacity.clamp(MIN_BURST_SIZE * mtu as u64, MAX_BURST_SIZE * mtu as u64) + // Batch the greater of `TARGET_BURST_INTERVAL` or `MIN_BURST_SIZE` worth of traffic at a + // time. To avoid inducing excessive latency, limit that result to at most `MAX_BURST_INTERVAL` + // worth of traffic. + Ord::min( + max_capacity, + target_capacity.clamp(MIN_BURST_SIZE * mtu, MAX_BURST_SIZE * mtu), + ) } -/// The burst interval -/// -/// The capacity will we refilled in 4/5 of that time. -/// 2ms is chosen here since framework timers might have 1ms precision. -/// If kernel-level pacing is supported later a higher time here might be -/// more applicable. -const BURST_INTERVAL_NANOS: u128 = 2_000_000; // 2ms +/// Period of traffic to batch together on a reasonably fast connection +const TARGET_BURST_INTERVAL: Duration = Duration::from_millis(2); -/// Allows some usage of GSO, and doesn't slow down the handshake. +/// Maximum period of traffic to batch together on a slow connection +/// +/// Takes precedence over [`MIN_BURST_SIZE`]. +const MAX_BURST_INTERVAL: Duration = Duration::from_millis(10); + +/// Minimum number of datagrams to batch together, so long as we won't have to wait for more than +/// [`MAX_BURST_INTERVAL`] const MIN_BURST_SIZE: u64 = 10; /// Creating 256 packets took 1ms in a benchmark, so larger bursts don't make sense. @@ -187,7 +199,7 @@ mod tests { let pacer = Pacer::new(rtt, window, mtu, now); assert_eq!( pacer.capacity, - (window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + (window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64 ); assert_eq!(pacer.tokens, pacer.capacity); @@ -196,7 +208,7 @@ mod tests { assert_eq!(pacer.tokens, pacer.capacity); let pacer = Pacer::new(rtt, 1, mtu, now); - assert_eq!(pacer.capacity, MIN_BURST_SIZE * mtu as u64); + assert_eq!(pacer.capacity, mtu as u64); assert_eq!(pacer.tokens, pacer.capacity); } @@ -210,7 +222,7 @@ mod tests { let mut pacer = Pacer::new(rtt, window, mtu, now); assert_eq!( pacer.capacity, - (window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + (window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64 ); assert_eq!(pacer.tokens, pacer.capacity); let initial_tokens = pacer.tokens; @@ -218,21 +230,21 @@ mod tests { pacer.delay(rtt, mtu as u64, mtu, window * 2, now); assert_eq!( pacer.capacity, - (2 * window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + (2 * window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64 ); assert_eq!(pacer.tokens, initial_tokens); pacer.delay(rtt, mtu as u64, mtu, window / 2, now); assert_eq!( pacer.capacity, - (window as u128 / 2 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + (window as u128 / 2 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64 ); assert_eq!(pacer.tokens, initial_tokens / 2); pacer.delay(rtt, mtu as u64, mtu * 2, window, now); assert_eq!( pacer.capacity, - (window as u128 * BURST_INTERVAL_NANOS / rtt.as_nanos()) as u64 + (window as u128 * TARGET_BURST_INTERVAL.as_nanos() / rtt.as_nanos()) as u64 ); pacer.delay(rtt, mtu as u64, 20_000, window, now); @@ -259,16 +271,20 @@ mod tests { pacer.on_transmit(mtu); } - let pace_duration = Duration::from_nanos((BURST_INTERVAL_NANOS * 4 / 5) as u64); + let pace_duration = Duration::from_nanos((TARGET_BURST_INTERVAL.as_nanos() * 4 / 5) as u64); - assert_eq!( - pacer - .delay(rtt, mtu as u64, mtu, window, old_instant) - .expect("Send must be delayed") - .duration_since(old_instant), - pace_duration + let actual_delay = pacer + .delay(rtt, mtu as u64, mtu, window, old_instant) + .expect("Send must be delayed") + .duration_since(old_instant); + + let diff = actual_delay.abs_diff(pace_duration); + + // Allow up to 2ns difference due to rounding + assert!( + diff < Duration::from_nanos(2), + "expected ≈ {pace_duration:?}, got {actual_delay:?} (diff {diff:?})" ); - // Refill half of the tokens assert_eq!( pacer.delay( diff --git a/quinn-proto/src/connection/qlog.rs b/quinn-proto/src/connection/qlog.rs index a324746f1..dafba3c47 100644 --- a/quinn-proto/src/connection/qlog.rs +++ b/quinn-proto/src/connection/qlog.rs @@ -3,6 +3,7 @@ #[cfg(feature = "qlog")] use std::sync::{Arc, Mutex}; +use std::time::Duration; #[cfg(feature = "qlog")] use qlog::{ @@ -86,7 +87,7 @@ impl QlogSink { &self, pn: u64, info: &SentPacket, - lost_send_time: Instant, + loss_delay: Duration, space: SpaceId, now: Instant, orig_rem_cid: ConnectionId, @@ -105,10 +106,12 @@ impl QlogSink { ..Default::default() }), frames: None, - trigger: Some(match info.time_sent <= lost_send_time { - true => PacketLostTrigger::TimeThreshold, - false => PacketLostTrigger::ReorderingThreshold, - }), + trigger: Some( + match info.time_sent.saturating_duration_since(now) >= loss_delay { + true => PacketLostTrigger::TimeThreshold, + false => PacketLostTrigger::ReorderingThreshold, + }, + ), }; stream.emit_event(orig_rem_cid, EventData::PacketLost(event), now); diff --git a/quinn-proto/src/connection/spaces.rs b/quinn-proto/src/connection/spaces.rs index e7f888e4f..9691149dd 100644 --- a/quinn-proto/src/connection/spaces.rs +++ b/quinn-proto/src/connection/spaces.rs @@ -222,6 +222,9 @@ pub(super) struct PacketNumberSpace { /// Transmitted but not acked // We use a BTreeMap here so we can efficiently query by range on ACK and for loss detection pub(super) sent_packets: BTreeMap, + /// Packets that were deemed lost + // Older packets are regularly removed in `Connection::drain_lost_packets`. + pub(super) lost_packets: BTreeMap, /// Number of explicit congestion notification codepoints seen on incoming packets pub(super) ecn_counters: frame::EcnCounts, /// Recent ECN counters sent by the peer in ACK frames @@ -274,6 +277,7 @@ impl PacketNumberSpace { largest_ack_eliciting_sent: 0, unacked_non_ack_eliciting_tail: 0, sent_packets: BTreeMap::new(), + lost_packets: BTreeMap::new(), ecn_counters: frame::EcnCounts::ZERO, ecn_feedback: frame::EcnCounts::ZERO, sent_with_keys: 0, @@ -302,6 +306,7 @@ impl PacketNumberSpace { largest_ack_eliciting_sent: 0, unacked_non_ack_eliciting_tail: 0, sent_packets: BTreeMap::new(), + lost_packets: BTreeMap::new(), ecn_counters: frame::EcnCounts::ZERO, ecn_feedback: frame::EcnCounts::ZERO, sent_with_keys: 0, @@ -331,6 +336,7 @@ impl PacketNumberSpace { largest_ack_eliciting_sent: 0, unacked_non_ack_eliciting_tail: 0, sent_packets: BTreeMap::new(), + lost_packets: BTreeMap::new(), ecn_counters: frame::EcnCounts::ZERO, ecn_feedback: frame::EcnCounts::ZERO, sent_with_keys: 0, @@ -512,6 +518,13 @@ pub(super) struct SentPacket { pub(super) stream_frames: frame::StreamMetaVec, } +/// Represents one or more packets that are deemed lost. +#[derive(Debug)] +pub(super) struct LostPacket { + /// The time the packet was sent. + pub(super) time_sent: Instant, +} + /// Retransmittable data queue #[allow(unreachable_pub)] // fuzzing only #[derive(Debug, Default, Clone)] diff --git a/quinn-proto/src/crypto.rs b/quinn-proto/src/crypto.rs index 8d79abc2b..bef848f3f 100644 --- a/quinn-proto/src/crypto.rs +++ b/quinn-proto/src/crypto.rs @@ -27,7 +27,7 @@ pub mod rustls; /// A cryptographic session (commonly TLS) pub trait Session: Send + Sync + 'static { /// Create the initial set of keys given the client's initial destination ConnectionId - fn initial_keys(&self, dst_cid: &ConnectionId, side: Side) -> Keys; + fn initial_keys(&self, dst_cid: ConnectionId, side: Side) -> Keys; /// Get data negotiated during the handshake, if available /// @@ -77,7 +77,7 @@ pub trait Session: Send + Sync + 'static { fn next_1rtt_keys(&mut self) -> Option>>; /// Verify the integrity of a retry packet - fn is_valid_retry(&self, orig_dst_cid: &ConnectionId, header: &[u8], payload: &[u8]) -> bool; + fn is_valid_retry(&self, orig_dst_cid: ConnectionId, header: &[u8], payload: &[u8]) -> bool; /// Fill `output` with `output.len()` bytes of keying material derived /// from the [Session]'s secrets, using `label` and `context` for domain @@ -123,16 +123,13 @@ pub trait ClientConfig: Send + Sync { /// Server-side configuration for the crypto protocol pub trait ServerConfig: Send + Sync { /// Create the initial set of keys given the client's initial destination ConnectionId - fn initial_keys( - &self, - version: u32, - dst_cid: &ConnectionId, - ) -> Result; + fn initial_keys(&self, version: u32, dst_cid: ConnectionId) + -> Result; /// Generate the integrity tag for a retry packet /// /// Never called if `initial_keys` rejected `version`. - fn retry_tag(&self, version: u32, orig_dst_cid: &ConnectionId, packet: &[u8]) -> [u8; 16]; + fn retry_tag(&self, version: u32, orig_dst_cid: ConnectionId, packet: &[u8]) -> [u8; 16]; /// Start a server session with this configuration /// diff --git a/quinn-proto/src/crypto/rustls.rs b/quinn-proto/src/crypto/rustls.rs index 4fd9d7ad8..25fa59ded 100644 --- a/quinn-proto/src/crypto/rustls.rs +++ b/quinn-proto/src/crypto/rustls.rs @@ -6,6 +6,8 @@ use bytes::BytesMut; #[cfg(feature = "ring")] use ring::aead; pub use rustls::Error; +#[cfg(feature = "__rustls-post-quantum-test")] +use rustls::NamedGroup; use rustls::{ self, CipherSuite, client::danger::ServerCertVerifier, @@ -51,8 +53,8 @@ impl TlsSession { } impl crypto::Session for TlsSession { - fn initial_keys(&self, dst_cid: &ConnectionId, side: Side) -> Keys { - initial_keys(self.version, *dst_cid, side, &self.suite) + fn initial_keys(&self, dst_cid: ConnectionId, side: Side) -> Keys { + initial_keys(self.version, dst_cid, side, &self.suite) } fn handshake_data(&self) -> Option> { @@ -65,6 +67,12 @@ impl crypto::Session for TlsSession { Connection::Client(_) => None, Connection::Server(ref session) => session.server_name().map(|x| x.into()), }, + #[cfg(feature = "__rustls-post-quantum-test")] + negotiated_key_exchange_group: self + .inner + .negotiated_key_exchange_group() + .expect("key exchange group is negotiated") + .name(), })) } @@ -102,6 +110,7 @@ impl crypto::Session for TlsSession { code: TransportErrorCode::crypto(alert.into()), frame: None, reason: e.to_string(), + crypto: Some(Arc::new(e)), } } else { TransportError::PROTOCOL_VIOLATION(format!("TLS error: {e}")) @@ -163,7 +172,7 @@ impl crypto::Session for TlsSession { }) } - fn is_valid_retry(&self, orig_dst_cid: &ConnectionId, header: &[u8], payload: &[u8]) -> bool { + fn is_valid_retry(&self, orig_dst_cid: ConnectionId, header: &[u8], payload: &[u8]) -> bool { let tag_start = match payload.len().checked_sub(16) { Some(x) => x, None => return false, @@ -172,7 +181,7 @@ impl crypto::Session for TlsSession { let mut pseudo_packet = Vec::with_capacity(header.len() + payload.len() + orig_dst_cid.len() + 1); pseudo_packet.push(orig_dst_cid.len() as u8); - pseudo_packet.extend_from_slice(orig_dst_cid); + pseudo_packet.extend_from_slice(&orig_dst_cid); pseudo_packet.extend_from_slice(header); let tag_start = tag_start + pseudo_packet.len(); pseudo_packet.extend_from_slice(payload); @@ -257,6 +266,9 @@ pub struct HandshakeData { /// /// Always `None` for outgoing connections pub server_name: Option, + /// The key exchange group negotiated with the peer + #[cfg(feature = "__rustls-post-quantum-test")] + pub negotiated_key_exchange_group: NamedGroup, } /// A QUIC-compatible TLS client configuration @@ -521,13 +533,13 @@ impl crypto::ServerConfig for QuicServerConfig { fn initial_keys( &self, version: u32, - dst_cid: &ConnectionId, + dst_cid: ConnectionId, ) -> Result { let version = interpret_version(version)?; - Ok(initial_keys(version, *dst_cid, Side::Server, &self.initial)) + Ok(initial_keys(version, dst_cid, Side::Server, &self.initial)) } - fn retry_tag(&self, version: u32, orig_dst_cid: &ConnectionId, packet: &[u8]) -> [u8; 16] { + fn retry_tag(&self, version: u32, orig_dst_cid: ConnectionId, packet: &[u8]) -> [u8; 16] { // Safe: `start_session()` is never called if `initial_keys()` rejected `version` let version = interpret_version(version).unwrap(); let (nonce, key) = match version { @@ -538,7 +550,7 @@ impl crypto::ServerConfig for QuicServerConfig { let mut pseudo_packet = Vec::with_capacity(packet.len() + orig_dst_cid.len() + 1); pseudo_packet.push(orig_dst_cid.len() as u8); - pseudo_packet.extend_from_slice(orig_dst_cid); + pseudo_packet.extend_from_slice(&orig_dst_cid); pseudo_packet.extend_from_slice(packet); let nonce = aead::Nonce::assume_unique_for_key(nonce); diff --git a/quinn-proto/src/endpoint.rs b/quinn-proto/src/endpoint.rs index eacbb7825..7b3e9b6bb 100644 --- a/quinn-proto/src/endpoint.rs +++ b/quinn-proto/src/endpoint.rs @@ -268,7 +268,7 @@ impl Endpoint { } else { // If we got this far, we're receiving a seemingly valid packet for an unknown // connection. Send a stateless reset if possible. - self.stateless_reset(now, datagram_len, addresses, *dst_cid, buf) + self.stateless_reset(now, datagram_len, addresses, dst_cid, buf) .map(DatagramEvent::Response) } } @@ -451,7 +451,7 @@ impl Endpoint { let Some(server_config) = &self.server_config else { debug!("packet for unrecognized connection {}", dst_cid); return self - .stateless_reset(event.now, datagram_len, addresses, *dst_cid, buf) + .stateless_reset(event.now, datagram_len, addresses, dst_cid, buf) .map(DatagramEvent::Response); }; @@ -478,7 +478,7 @@ impl Endpoint { header.version, addresses, &crypto, - &header.src_cid, + header.src_cid, reason, buf, ))); @@ -511,7 +511,7 @@ impl Endpoint { header.version, addresses, &crypto, - &header.src_cid, + header.src_cid, TransportError::INVALID_TOKEN(""), buf, ))); @@ -540,15 +540,14 @@ impl Endpoint { } /// Attempt to accept this incoming connection (an error may still occur) - // AcceptError cannot be made smaller without semver breakage - #[allow(clippy::result_large_err)] + // box err to avoid clippy::result_large_err pub fn accept( &mut self, mut incoming: Incoming, now: Instant, buf: &mut Vec, server_config: Option>, - ) -> Result<(ConnectionHandle, Connection), AcceptError> { + ) -> Result<(ConnectionHandle, Connection), Box> { let remote_address_validated = incoming.remote_address_validated(); incoming.improper_drop_warner.dismiss(); let incoming_buffer = self.incoming_buffers.remove(incoming.incoming_idx); @@ -573,26 +572,26 @@ impl Endpoint { { debug!("abandoning accept of stale initial"); self.index.remove_initial(dst_cid); - return Err(AcceptError { + return Err(Box::new(AcceptError { cause: ConnectionError::TimedOut, response: None, - }); + })); } if self.cids_exhausted() { debug!("refusing connection"); self.index.remove_initial(dst_cid); - return Err(AcceptError { + return Err(Box::new(AcceptError { cause: ConnectionError::CidsExhausted, response: Some(self.initial_close( version, incoming.addresses, &incoming.crypto, - &src_cid, + src_cid, TransportError::CONNECTION_REFUSED(""), buf, )), - }); + })); } if incoming @@ -609,10 +608,10 @@ impl Endpoint { { debug!(packet_number, "failed to authenticate initial packet"); self.index.remove_initial(dst_cid); - return Err(AcceptError { + return Err(Box::new(AcceptError { cause: TransportError::PROTOCOL_VIOLATION("authentication failed").into(), response: None, - }); + })); }; let ch = ConnectionHandle(self.connections.vacant_key()); @@ -685,13 +684,13 @@ impl Endpoint { version, incoming.addresses, &incoming.crypto, - &src_cid, + src_cid, e.clone(), buf, )), _ => None, }; - Err(AcceptError { cause: e, response }) + Err(Box::new(AcceptError { cause: e, response })) } } } @@ -735,7 +734,7 @@ impl Endpoint { incoming.packet.header.version, incoming.addresses, &incoming.crypto, - &incoming.packet.header.src_cid, + incoming.packet.header.src_cid, TransportError::CONNECTION_REFUSED(""), buf, ) @@ -779,7 +778,7 @@ impl Endpoint { buf.put_slice(&token); buf.extend_from_slice(&server_config.crypto.retry_tag( incoming.packet.header.version, - &incoming.packet.header.dst_cid, + incoming.packet.header.dst_cid, buf, )); encode.finish(buf, &*incoming.crypto.header.local, None); @@ -872,7 +871,7 @@ impl Endpoint { version: u32, addresses: FourTuple, crypto: &Keys, - remote_id: &ConnectionId, + remote_id: ConnectionId, reason: TransportError, buf: &mut Vec, ) -> Transmit { @@ -882,7 +881,7 @@ impl Endpoint { let local_id = self.local_cid_generator.generate_cid(); let number = PacketNumber::U8(0); let header = Header::Initial(InitialHeader { - dst_cid: *remote_id, + dst_cid: remote_id, src_cid: local_id, number, token: Bytes::new(), @@ -1100,12 +1099,12 @@ impl ConnectionIndex { /// Find the existing connection that `datagram` should be routed to, if any fn get(&self, addresses: &FourTuple, datagram: &PartialDecode) -> Option { if !datagram.dst_cid().is_empty() { - if let Some(&(ch, path_id)) = self.connection_ids.get(datagram.dst_cid()) { + if let Some(&(ch, path_id)) = self.connection_ids.get(&datagram.dst_cid()) { return Some(RouteDatagramTo::Connection(ch, path_id)); } } if datagram.is_initial() || datagram.is_0rtt() { - if let Some(&ch) = self.connection_ids_initial.get(datagram.dst_cid()) { + if let Some(&ch) = self.connection_ids_initial.get(&datagram.dst_cid()) { return Some(ch); } } @@ -1245,8 +1244,8 @@ impl Incoming { } /// The original destination connection ID sent by the client - pub fn orig_dst_cid(&self) -> &ConnectionId { - &self.token.orig_dst_cid + pub fn orig_dst_cid(&self) -> ConnectionId { + self.token.orig_dst_cid } } diff --git a/quinn-proto/src/packet.rs b/quinn-proto/src/packet.rs index 28b925742..6fff11930 100644 --- a/quinn-proto/src/packet.rs +++ b/quinn-proto/src/packet.rs @@ -99,7 +99,7 @@ impl PartialDecode { } /// The destination connection ID of the packet - pub fn dst_cid(&self) -> &ConnectionId { + pub fn dst_cid(&self) -> ConnectionId { self.plain_header.dst_cid() } @@ -572,14 +572,14 @@ impl ProtectedHeader { } /// The destination Connection ID of the packet - pub fn dst_cid(&self) -> &ConnectionId { + pub fn dst_cid(&self) -> ConnectionId { use ProtectedHeader::*; match self { - Initial(header) => &header.dst_cid, - Long { dst_cid, .. } => dst_cid, - Retry { dst_cid, .. } => dst_cid, - Short { dst_cid, .. } => dst_cid, - VersionNegotiate { dst_cid, .. } => dst_cid, + Initial(header) => header.dst_cid, + &Long { dst_cid, .. } => dst_cid, + &Retry { dst_cid, .. } => dst_cid, + &Short { dst_cid, .. } => dst_cid, + &VersionNegotiate { dst_cid, .. } => dst_cid, } } diff --git a/quinn-proto/src/tests/mod.rs b/quinn-proto/src/tests/mod.rs index c9fe23c9a..983e95748 100644 --- a/quinn-proto/src/tests/mod.rs +++ b/quinn-proto/src/tests/mod.rs @@ -580,6 +580,10 @@ fn zero_rtt_happypath() { pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); // We don't currently preserve stream event order wrt. connection events assert_matches!( pair.server_conn_mut(server_ch).poll(), @@ -625,6 +629,10 @@ fn zero_rtt_rejection() { pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); assert_matches!( pair.server_conn_mut(server_ch).poll(), Some(Event::Connected) @@ -666,6 +674,10 @@ fn zero_rtt_rejection() { pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); assert_matches!( pair.server_conn_mut(server_ch).poll(), Some(Event::Connected) @@ -698,10 +710,17 @@ fn test_zero_rtt_incoming_limit(configure_server: const EXPECTED_DROPPED: u64 = 4; let _guard = subscribe(); + + let mut transport = TransportConfig::default(); + // Assume a low-latency connection so pacing doesn't interfere with the test + transport.initial_rtt(Duration::from_millis(10)); + let transport = Arc::new(transport); + let mut server_config = server_config(); configure_server(&mut server_config); let mut pair = Pair::new(Arc::new(EndpointConfig::default()), server_config); - let config = client_config(); + let mut config = client_config(); + config.transport_config(transport); // Establish normal connection let client_ch = pair.begin_connect(config.clone()); @@ -748,6 +767,10 @@ fn test_zero_rtt_incoming_limit(configure_server: pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); // We don't currently preserve stream event order wrt. connection events assert_matches!( pair.server_conn_mut(server_ch).poll(), @@ -821,6 +844,10 @@ fn alpn_success() { pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); assert_matches!( pair.server_conn_mut(server_ch).poll(), Some(Event::Connected) @@ -2110,6 +2137,10 @@ fn large_initial() { pair.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + pair.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); assert_matches!( pair.server_conn_mut(server_ch).poll(), Some(Event::Connected) @@ -2305,9 +2336,14 @@ fn handshake_anti_deadlock_probe() { #[test] fn server_can_send_3_inital_packets() { let _guard = subscribe(); + let mut transport = TransportConfig::default(); + // Assume a low-latency connection so pacing doesn't interfere with the test + transport.initial_rtt(Duration::from_millis(10)); + let transport = Arc::new(transport); let (cert, key) = big_cert_and_key(); - let server = server_config_with_cert(cert.clone(), key); + let mut server = server_config_with_cert(cert.clone(), key); + server.transport_config(transport); let client = client_config_with_certs(vec![cert]); let mut pair = Pair::new(Default::default(), server); @@ -2859,7 +2895,8 @@ fn setup_ack_frequency_test(max_ack_delay: Duration) -> (Pair, ConnectionHandle, Arc::get_mut(&mut client_config.transport) .unwrap() .ack_frequency_config(Some(ack_freq_config)) - .mtu_discovery_config(None); // To keep traffic cleaner + .mtu_discovery_config(None) // To keep traffic cleaner + .initial_rtt(Duration::from_millis(10)); // To avoid delays from pacing let mut pair = Pair::default_with_deterministic_pns(); pair.latency = Duration::from_millis(10); // Need latency to avoid an RTT = 0 @@ -3515,6 +3552,7 @@ fn address_discovery() { let conn = pair.client_conn_mut(conn_handle); assert_matches!(conn.poll(), Some(Event::HandshakeDataReady)); assert_matches!(conn.poll(), Some(Event::Connected)); + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Path(PathEvent::ObservedAddr{id: PathId::ZERO, addr})) if addr == expected_addr); assert_matches!(conn.poll(), None); @@ -3523,6 +3561,7 @@ fn address_discovery() { let expected_addr = pair.server.addr; let conn = pair.server_conn_mut(conn_handle); assert_matches!(conn.poll(), Some(Event::HandshakeDataReady)); + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Connected)); assert_matches!(conn.poll(), Some(Event::Path(PathEvent::ObservedAddr{id: PathId::ZERO, addr})) if addr == expected_addr); assert_matches!(conn.poll(), None); @@ -3593,6 +3632,7 @@ fn address_discovery_zero_rtt_accepted() { let conn = pair.server_conn_mut(server_ch); assert_matches!(conn.poll(), Some(Event::HandshakeDataReady)); // We don't currently preserve stream event order wrt. connection events + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Connected)); assert_matches!( conn.poll(), @@ -3652,6 +3692,7 @@ fn address_discovery_zero_rtt_rejection() { let server_ch = pair.server.assert_accept(); let conn = pair.server_conn_mut(server_ch); assert_matches!(conn.poll(), Some(Event::HandshakeDataReady)); + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Connected)); assert_matches!(conn.poll(), None); pair.client @@ -3693,6 +3734,8 @@ fn address_discovery_retransmission() { let server = ServerConfig { transport: Arc::new(TransportConfig { address_discovery_role: crate::address_discovery::Role::Both, + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..server_config() @@ -3701,6 +3744,8 @@ fn address_discovery_retransmission() { let client_config = ClientConfig { transport: Arc::new(TransportConfig { address_discovery_role: crate::address_discovery::Role::Both, + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..client_config() @@ -3718,6 +3763,7 @@ fn address_discovery_retransmission() { pair.drive(); let conn = pair.client_conn_mut(client_ch); + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Path(PathEvent::ObservedAddr{id: PathId::ZERO, addr})) if addr == pair.client.addr); } @@ -3728,6 +3774,8 @@ fn address_discovery_rebind_retransmission() { let server = ServerConfig { transport: Arc::new(TransportConfig { address_discovery_role: crate::address_discovery::Role::Both, + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..server_config() @@ -3736,6 +3784,8 @@ fn address_discovery_rebind_retransmission() { let client_config = ClientConfig { transport: Arc::new(TransportConfig { address_discovery_role: crate::address_discovery::Role::Both, + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..client_config() @@ -3760,6 +3810,7 @@ fn address_discovery_rebind_retransmission() { pair.drive(); let conn = pair.client_conn_mut(client_ch); + assert_matches!(conn.poll(), Some(Event::HandshakeConfirmed)); assert_matches!(conn.poll(), Some(Event::Path(PathEvent::ObservedAddr{id: PathId::ZERO, addr})) if addr == pair.client.addr); } @@ -3795,3 +3846,92 @@ fn preferred_address() { let mut pair = Pair::new(Arc::new(EndpointConfig::default()), server_config); pair.connect(); } + +#[test] +fn handshake_sequence() { + let _guard = subscribe(); + + let mut pair = Pair::default(); + let ch = pair.begin_connect(client_config()); + + pair.step(); + assert_matches!(pair.client_conn_mut(ch).poll(), None); + let sh = pair.server.assert_accept(); + assert_matches!( + pair.server_conn_mut(sh).poll(), + Some(Event::HandshakeDataReady) + ); + assert_matches!(pair.server_conn_mut(sh).poll(), None); + + pair.step(); + assert_matches!( + pair.client_conn_mut(ch).poll(), + Some(Event::HandshakeDataReady) + ); + assert_matches!(pair.client_conn_mut(ch).poll(), Some(Event::Connected)); + assert_matches!(pair.client_conn_mut(ch).poll(), None); + assert_matches!( + pair.server_conn_mut(sh).poll(), + Some(Event::HandshakeConfirmed) + ); + assert_matches!(pair.server_conn_mut(sh).poll(), Some(Event::Connected)); + assert_matches!(pair.server_conn_mut(sh).poll(), None); + + pair.drive_client(); + assert_matches!( + pair.client_conn_mut(ch).poll(), + Some(Event::HandshakeConfirmed) + ); + assert_matches!(pair.client_conn_mut(ch).poll(), None); +} + +#[test] +fn handshake_confirmation_no_resumption_shortcut() { + let _guard = subscribe(); + + // Initial connection + let mut pair = Pair::default(); + let config = client_config(); + let (ch, _) = pair.connect_with(config.clone()); + pair.client + .connections + .get_mut(&ch) + .unwrap() + .close(pair.time, VarInt(0), [][..].into()); + pair.drive(); + + // Resumed connection + info!("resuming session"); + let ch = pair.begin_connect(config); + assert!(pair.client_conn_mut(ch).has_0rtt()); + + pair.step(); + assert_matches!(pair.client_conn_mut(ch).poll(), None); + let sh = pair.server.assert_accept(); + assert_matches!( + pair.server_conn_mut(sh).poll(), + Some(Event::HandshakeDataReady) + ); + assert_matches!(pair.server_conn_mut(sh).poll(), None); + + pair.step(); + assert_matches!( + pair.client_conn_mut(ch).poll(), + Some(Event::HandshakeDataReady) + ); + assert_matches!(pair.client_conn_mut(ch).poll(), Some(Event::Connected)); + assert_matches!(pair.client_conn_mut(ch).poll(), None); + assert_matches!( + pair.server_conn_mut(sh).poll(), + Some(Event::HandshakeConfirmed) + ); + assert_matches!(pair.server_conn_mut(sh).poll(), Some(Event::Connected)); + assert_matches!(pair.server_conn_mut(sh).poll(), None); + + pair.drive_client(); + assert_matches!( + pair.client_conn_mut(ch).poll(), + Some(Event::HandshakeConfirmed) + ); + assert_matches!(pair.client_conn_mut(ch).poll(), None); +} diff --git a/quinn-proto/src/tests/multipath.rs b/quinn-proto/src/tests/multipath.rs index b7bb2b1d2..4cd4f4170 100644 --- a/quinn-proto/src/tests/multipath.rs +++ b/quinn-proto/src/tests/multipath.rs @@ -25,6 +25,8 @@ const MAX_PATHS: u32 = 3; fn multipath_pair() -> (Pair, ConnectionHandle, ConnectionHandle) { let multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(MAX_PATHS), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let server_cfg = Arc::new(ServerConfig { @@ -50,6 +52,8 @@ fn non_zero_length_cids() { let _guard = subscribe(); let multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(3 as _), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let server_cfg = Arc::new(ServerConfig { @@ -197,6 +201,8 @@ fn multipath_cid_rotation() { let server_cfg = ServerConfig { transport: Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(MAX_PATHS), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..server_config() @@ -217,6 +223,8 @@ fn multipath_cid_rotation() { let client_cfg = ClientConfig { transport: Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(MAX_PATHS), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }), ..client_config() @@ -306,6 +314,8 @@ fn issue_max_path_id() { // We enable multipath but initially do not allow any paths to be opened. let multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(1), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let server_cfg = Arc::new(ServerConfig { @@ -320,6 +330,8 @@ fn issue_max_path_id() { // The client is allowed to create more paths immediately. let client_multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(MAX_PATHS), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let client_cfg = ClientConfig { @@ -374,6 +386,8 @@ fn issue_max_path_id_reordered() { // We enable multipath but initially do not allow any paths to be opened. let multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(1), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let server_cfg = Arc::new(ServerConfig { @@ -388,6 +402,8 @@ fn issue_max_path_id_reordered() { // The client is allowed to create more paths immediately. let client_multipath_transport_cfg = Arc::new(TransportConfig { max_concurrent_multipath_paths: NonZeroU32::new(MAX_PATHS), + // Assume a low-latency connection so pacing doesn't interfere with the test + initial_rtt: Duration::from_millis(10), ..TransportConfig::default() }); let client_cfg = ClientConfig { diff --git a/quinn-proto/src/tests/util.rs b/quinn-proto/src/tests/util.rs index 372693b6d..b409adb01 100644 --- a/quinn-proto/src/tests/util.rs +++ b/quinn-proto/src/tests/util.rs @@ -7,12 +7,11 @@ use std::{ net::{Ipv6Addr, SocketAddr, UdpSocket}, ops::RangeFrom, str, - sync::{Arc, Mutex}, + sync::{Arc, LazyLock, Mutex}, }; use assert_matches::assert_matches; use bytes::BytesMut; -use lazy_static::lazy_static; use rustls::{ KeyLogFile, client::WebPkiServerVerifier, @@ -246,10 +245,18 @@ impl Pair { self.server_conn_mut(server_ch).poll(), Some(Event::HandshakeDataReady) ); + assert_matches!( + self.server_conn_mut(server_ch).poll(), + Some(Event::HandshakeConfirmed) + ); assert_matches!( self.server_conn_mut(server_ch).poll(), Some(Event::Connected) ); + assert_matches!( + self.client_conn_mut(client_ch).poll(), + Some(Event::HandshakeConfirmed) + ); } pub(super) fn client_conn_mut(&mut self, ch: ConnectionHandle) -> &mut Connection { @@ -742,12 +749,12 @@ fn set_congestion_experienced( }) } -lazy_static! { - pub static ref SERVER_PORTS: Mutex> = Mutex::new(4433..); - pub static ref CLIENT_PORTS: Mutex> = Mutex::new(44433..); - pub(crate) static ref CERTIFIED_KEY: rcgen::CertifiedKey = - rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); -} +pub(crate) static SERVER_PORTS: LazyLock>> = + LazyLock::new(|| Mutex::new(4433..)); +pub(crate) static CLIENT_PORTS: LazyLock>> = + LazyLock::new(|| Mutex::new(44433..)); +pub(crate) static CERTIFIED_KEY: LazyLock> = + LazyLock::new(|| rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap()); #[derive(Default)] struct SimpleTokenLog(Mutex>); diff --git a/quinn-proto/src/token.rs b/quinn-proto/src/token.rs index e4ab5ea52..c128d0603 100644 --- a/quinn-proto/src/token.rs +++ b/quinn-proto/src/token.rs @@ -247,9 +247,8 @@ impl Token { fn decode(key: &dyn HandshakeTokenKey, raw_token_bytes: &[u8]) -> Option { // Decrypt - // MSRV: split_at_checked requires 1.80.0 let nonce_slice_start = raw_token_bytes.len().checked_sub(size_of::())?; - let (sealed_token, nonce_bytes) = raw_token_bytes.split_at(nonce_slice_start); + let (sealed_token, nonce_bytes) = raw_token_bytes.split_at_checked(nonce_slice_start)?; let nonce = u128::from_le_bytes(nonce_bytes.try_into().unwrap()); diff --git a/quinn-proto/src/transport_error.rs b/quinn-proto/src/transport_error.rs index cc82b0f61..be9d2cd9d 100644 --- a/quinn-proto/src/transport_error.rs +++ b/quinn-proto/src/transport_error.rs @@ -1,4 +1,4 @@ -use std::fmt; +use std::{fmt, sync::Arc}; use bytes::{Buf, BufMut}; @@ -9,7 +9,12 @@ use crate::{ }; /// Transport-level errors occur when a peer violates the protocol specification -#[derive(Debug, Clone, Eq, PartialEq)] +/// +/// # Note +/// +/// The `PartialEq` implementation for this type performs comparison on the `code` field only +#[derive(Debug, Clone)] +#[non_exhaustive] pub struct Error { /// Type of error pub code: Code, @@ -17,8 +22,30 @@ pub struct Error { pub frame: Option, /// Human-readable explanation of the reason pub reason: String, + /// An underlying crypto (e.g. TLS) layer error + pub crypto: Option>, } +impl Error { + /// Construct an error with a code and a reason + pub fn new(code: Code, reason: String) -> Self { + Self { + code, + frame: None, + reason, + crypto: None, + } + } +} + +impl PartialEq for Error { + fn eq(&self, other: &Self) -> bool { + self.code == other.code + } +} + +impl Eq for Error {} + impl fmt::Display for Error { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { self.code.fmt(f)?; @@ -40,6 +67,7 @@ impl From for Error { code: x, frame: None, reason: "".to_string(), + crypto: None, } } } @@ -92,6 +120,7 @@ macro_rules! errors { code: Code::$name, frame: None, reason: reason.into(), + crypto: None, } } )* diff --git a/quinn-udp/Cargo.toml b/quinn-udp/Cargo.toml index b5658dbd6..33f2fbcce 100644 --- a/quinn-udp/Cargo.toml +++ b/quinn-udp/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "iroh-quinn-udp" -version = "0.5.12" +version = "0.6.0" edition.workspace = true rust-version.workspace = true license.workspace = true @@ -13,10 +13,10 @@ workspace = ".." [features] # NOTE: Please keep this in sync with the feature list in `.github/workflows/codecov.yml`, see # comment in that file for more information. -default = ["tracing", "log"] +default = ["tracing", "tracing-log"] # Configure `tracing` to log events via `log` if no `tracing` subscriber exists. -log = ["tracing/log"] -direct-log = ["dep:log"] +tracing-log = ["tracing/log"] +log = ["dep:log"] # Use private Apple APIs to send multiple packets in a single syscall. fast-apple-datapath = [] @@ -29,7 +29,6 @@ tracing = { workspace = true, optional = true } socket2 = { workspace = true } [target.'cfg(windows)'.dependencies] -once_cell = { workspace = true } windows-sys = { workspace = true } [dev-dependencies] diff --git a/quinn-udp/build.rs b/quinn-udp/build.rs index c43c0aa34..ed26e5e64 100644 --- a/quinn-udp/build.rs +++ b/quinn-udp/build.rs @@ -9,7 +9,8 @@ fn main() { target_os = "macos", target_os = "ios", target_os = "tvos", - target_os = "visionos" + target_os = "visionos", + target_os = "watchos" ) }, bsd: { diff --git a/quinn-udp/src/fallback.rs b/quinn-udp/src/fallback.rs index 4444bed81..ee5762090 100644 --- a/quinn-udp/src/fallback.rs +++ b/quinn-udp/src/fallback.rs @@ -72,6 +72,7 @@ impl UdpSocketState { addr: addr.as_socket().unwrap(), ecn: None, dst_ip: None, + interface_index: None, }; Ok(1) } diff --git a/quinn-udp/src/lib.rs b/quinn-udp/src/lib.rs index 0f69070f6..eaceaf521 100644 --- a/quinn-udp/src/lib.rs +++ b/quinn-udp/src/lib.rs @@ -56,13 +56,13 @@ mod imp; #[allow(unused_imports, unused_macros)] mod log { - #[cfg(all(feature = "direct-log", not(feature = "tracing")))] + #[cfg(all(feature = "log", not(feature = "tracing-log")))] pub(crate) use log::{debug, error, info, trace, warn}; - #[cfg(feature = "tracing")] + #[cfg(feature = "tracing-log")] pub(crate) use tracing::{debug, error, info, trace, warn}; - #[cfg(not(any(feature = "direct-log", feature = "tracing")))] + #[cfg(not(any(feature = "log", feature = "tracing-log")))] mod no_op { macro_rules! trace ( ($($tt:tt)*) => {{}} ); macro_rules! debug ( ($($tt:tt)*) => {{}} ); @@ -73,7 +73,7 @@ mod log { pub(crate) use {debug, error, info, log_warn as warn, trace}; } - #[cfg(not(any(feature = "direct-log", feature = "tracing")))] + #[cfg(not(any(feature = "log", feature = "tracing-log")))] pub(crate) use no_op::*; } @@ -93,6 +93,7 @@ pub const BATCH_SIZE: usize = 1; /// /// [`stride`]: RecvMeta::stride #[derive(Debug, Copy, Clone)] +#[non_exhaustive] pub struct RecvMeta { /// The source address of the datagram(s) contained in the buffer pub addr: SocketAddr, @@ -115,6 +116,8 @@ pub struct RecvMeta { /// Populated on platforms: Windows, Linux, Android (API level > 25), /// FreeBSD, OpenBSD, NetBSD, macOS, and iOS. pub dst_ip: Option, + /// The interface index of the interface on which the datagram was received + pub interface_index: Option, } impl Default for RecvMeta { @@ -126,6 +129,7 @@ impl Default for RecvMeta { stride: 0, ecn: None, dst_ip: None, + interface_index: None, } } } @@ -154,7 +158,7 @@ const IO_ERROR_LOG_INTERVAL: Duration = std::time::Duration::from_secs(60); /// /// Logging will only be performed if at least [`IO_ERROR_LOG_INTERVAL`] /// has elapsed since the last error was logged. -#[cfg(all(not(wasm_browser), any(feature = "tracing", feature = "direct-log")))] +#[cfg(all(not(wasm_browser), any(feature = "tracing-log", feature = "log")))] fn log_sendmsg_error( last_send_error: &Mutex, err: impl core::fmt::Debug, @@ -177,7 +181,7 @@ fn log_sendmsg_error( } // No-op -#[cfg(not(any(wasm_browser, feature = "tracing", feature = "direct-log")))] +#[cfg(not(any(wasm_browser, feature = "tracing-log", feature = "log")))] fn log_sendmsg_error(_: &Mutex, _: impl core::fmt::Debug, _: &Transmit) {} /// A borrowed UDP socket diff --git a/quinn-udp/src/unix.rs b/quinn-udp/src/unix.rs index 6557eac56..376b4afa2 100644 --- a/quinn-udp/src/unix.rs +++ b/quinn-udp/src/unix.rs @@ -113,7 +113,12 @@ impl UdpSocketState { // mac and ios do not support IP_RECVTOS on dual-stack sockets :( // older macos versions also don't have the flag and will error out if we don't ignore it - #[cfg(not(any(target_os = "openbsd", target_os = "netbsd", solarish)))] + #[cfg(not(any( + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish + )))] if is_ipv4 || !io.only_v6()? { if let Err(_err) = set_socket_option(&*io, libc::IPPROTO_IP, libc::IP_RECVTOS, OPTION_ON) @@ -461,7 +466,13 @@ fn send(state: &UdpSocketState, io: SockRef<'_>, transmit: &Transmit<'_>) -> io: } } -#[cfg(not(any(apple, target_os = "openbsd", target_os = "netbsd", solarish)))] +#[cfg(not(any( + apple, + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish +)))] fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result { let mut names = [MaybeUninit::::uninit(); BATCH_SIZE]; let mut ctrls = [cmsg::Aligned(MaybeUninit::<[u8; CMSG_LEN]>::uninit()); BATCH_SIZE]; @@ -498,7 +509,7 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> } }; for i in 0..(msg_count as usize) { - meta[i] = decode_recv(&names[i], &hdrs[i].msg_hdr, hdrs[i].msg_len as usize); + meta[i] = decode_recv(&names[i], &hdrs[i].msg_hdr, hdrs[i].msg_len as usize)?; } Ok(msg_count as usize) } @@ -533,12 +544,18 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> } }; for i in 0..(msg_count as usize) { - meta[i] = decode_recv(&names[i], &hdrs[i], hdrs[i].msg_datalen as usize); + meta[i] = decode_recv(&names[i], &hdrs[i], hdrs[i].msg_datalen as usize)?; } Ok(msg_count as usize) } -#[cfg(any(target_os = "openbsd", target_os = "netbsd", solarish, apple_slow))] +#[cfg(any( + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish, + apple_slow +))] fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> io::Result { let mut name = MaybeUninit::::uninit(); let mut ctrl = cmsg::Aligned(MaybeUninit::<[u8; CMSG_LEN]>::uninit()); @@ -562,7 +579,7 @@ fn recv(io: SockRef<'_>, bufs: &mut [IoSliceMut<'_>], meta: &mut [RecvMeta]) -> _ => return Err(e), } }; - meta[0] = decode_recv(&name, &hdr, n as usize); + meta[0] = decode_recv(&name, &hdr, n as usize)?; Ok(1) } @@ -700,10 +717,11 @@ fn decode_recv( #[cfg(not(apple_fast))] hdr: &libc::msghdr, #[cfg(apple_fast)] hdr: &msghdr_x, len: usize, -) -> RecvMeta { +) -> io::Result { let name = unsafe { name.assume_init() }; let mut ecn_bits = 0; let mut dst_ip = None; + let mut interface_index = None; #[allow(unused_mut)] // only mutable on Linux let mut stride = len; @@ -714,7 +732,12 @@ fn decode_recv( ecn_bits = cmsg::decode::(cmsg); }, // FreeBSD uses IP_RECVTOS here, and we can be liberal because cmsgs are opt-in. - #[cfg(not(any(target_os = "openbsd", target_os = "netbsd", solarish)))] + #[cfg(not(any( + target_os = "openbsd", + target_os = "netbsd", + target_os = "dragonfly", + solarish + )))] (libc::IPPROTO_IP, libc::IP_RECVTOS) => unsafe { ecn_bits = cmsg::decode::(cmsg); }, @@ -736,6 +759,7 @@ fn decode_recv( dst_ip = Some(IpAddr::V4(Ipv4Addr::from( pktinfo.ipi_addr.s_addr.to_ne_bytes(), ))); + interface_index = Some(pktinfo.ipi_ifindex as u32); } #[cfg(any(bsd, apple))] (libc::IPPROTO_IP, libc::IP_RECVDSTADDR) => { @@ -745,6 +769,7 @@ fn decode_recv( (libc::IPPROTO_IPV6, libc::IPV6_PKTINFO) => { let pktinfo = unsafe { cmsg::decode::(cmsg) }; dst_ip = Some(IpAddr::V6(Ipv6Addr::from(pktinfo.ipi6_addr.s6_addr))); + interface_index = Some(pktinfo.ipi6_ifindex as u32); } #[cfg(any(target_os = "linux", target_os = "android"))] (libc::SOL_UDP, gro::UDP_GRO) => unsafe { @@ -775,16 +800,21 @@ fn decode_recv( addr.sin6_scope_id, )) } - _ => unreachable!(), + f => { + return Err(io::Error::other(format!( + "expected AF_INET or AF_INET6, got {f} in decode_recv" + ))); + } }; - RecvMeta { + Ok(RecvMeta { len, stride, addr, ecn: EcnCodepoint::from_bits(ecn_bits), dst_ip, - } + interface_index, + }) } #[cfg(not(apple_slow))] diff --git a/quinn-udp/src/windows.rs b/quinn-udp/src/windows.rs index 1e0410722..ae1fb5033 100644 --- a/quinn-udp/src/windows.rs +++ b/quinn-udp/src/windows.rs @@ -4,12 +4,11 @@ use std::{ net::{IpAddr, Ipv4Addr}, os::windows::io::AsRawSocket, ptr, - sync::Mutex, + sync::{LazyLock, Mutex}, time::Instant, }; use libc::{c_int, c_uint}; -use once_cell::sync::Lazy; use windows_sys::Win32::Networking::WinSock; use crate::{ @@ -225,6 +224,7 @@ impl UdpSocketState { // Decode control messages (PKTINFO and ECN) let mut ecn_bits = 0; let mut dst_ip = None; + let mut interface_index = None; let mut stride = len; let cmsg_iter = unsafe { cmsg::Iter::new(&wsa_msg) }; @@ -238,12 +238,14 @@ impl UdpSocketState { // Addr is stored in big endian format let ip4 = Ipv4Addr::from(u32::from_be(unsafe { pktinfo.ipi_addr.S_un.S_addr })); dst_ip = Some(ip4.into()); + interface_index = Some(pktinfo.ipi_ifindex); } (WinSock::IPPROTO_IPV6, WinSock::IPV6_PKTINFO) => { let pktinfo = unsafe { cmsg::decode::(cmsg) }; // Addr is stored in big endian format dst_ip = Some(IpAddr::from(unsafe { pktinfo.ipi6_addr.u.Byte })); + interface_index = Some(pktinfo.ipi6_ifindex); } (WinSock::IPPROTO_IP, WinSock::IP_ECN) => { // ECN is a C integer https://learn.microsoft.com/en-us/windows/win32/winsock/winsock-ecn @@ -268,6 +270,7 @@ impl UdpSocketState { addr: addr.unwrap(), ecn: EcnCodepoint::from_bits(ecn_bits as u8), dst_ip, + interface_index, }; Ok(1) } @@ -443,8 +446,7 @@ pub(crate) const BATCH_SIZE: usize = 1; const CMSG_LEN: usize = 128; const OPTION_ON: u32 = 1; -// FIXME this could use [`std::sync::OnceLock`] once the MSRV is bumped to 1.70 and upper -static WSARECVMSG_PTR: Lazy = Lazy::new(|| { +static WSARECVMSG_PTR: LazyLock = LazyLock::new(|| { let s = unsafe { WinSock::socket(WinSock::AF_INET as _, WinSock::SOCK_DGRAM as _, 0) }; if s == WinSock::INVALID_SOCKET { debug!( @@ -492,7 +494,7 @@ static WSARECVMSG_PTR: Lazy = Lazy::new(|| { wsa_recvmsg_ptr }); -static MAX_GSO_SEGMENTS: Lazy = Lazy::new(|| { +static MAX_GSO_SEGMENTS: LazyLock = LazyLock::new(|| { let socket = match std::net::UdpSocket::bind("[::]:0") .or_else(|_| std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0))) { diff --git a/quinn/Cargo.toml b/quinn/Cargo.toml index 7b4f6a294..5d4a49024 100644 --- a/quinn/Cargo.toml +++ b/quinn/Cargo.toml @@ -15,7 +15,7 @@ rust-version.workspace = true [features] # NOTE: Please keep this in sync with the feature list in `.github/workflows/codecov.yml`, see # comment in that file for more information. -default = ["log", "platform-verifier", "runtime-tokio", "rustls-ring", "bloom"] +default = ["tracing-log", "platform-verifier", "runtime-tokio", "rustls-ring", "bloom"] # Enables `Endpoint::client` and `Endpoint::server` conveniences aws-lc-rs = ["proto/aws-lc-rs"] aws-lc-rs-fips = ["proto/aws-lc-rs-fips"] @@ -36,15 +36,20 @@ rustls-ring = ["dep:rustls", "ring", "proto/rustls-ring", "proto/ring"] # Outside wasm*-unknown-unknown targets, this enables `Endpoint::client` and `Endpoint::server` conveniences. ring = ["proto/ring"] runtime-tokio = ["tokio/time", "tokio/rt", "tokio/net"] -runtime-smol = ["async-io", "smol"] +runtime-smol = ["dep:async-io", "dep:smol"] # Configure `tracing` to log events via `log` if no `tracing` subscriber exists. -log = ["tracing/log", "proto/log", "udp/log"] +tracing-log = ["tracing/log", "proto/tracing-log", "udp/tracing-log"] # Enable rustls logging rustls-log = ["rustls?/logging"] # Enable qlog support qlog = ["proto/qlog"] +# Internal (PRIVATE!) features used to aid testing. +# Don't rely on these whatsoever. They may disappear at any time. + +__rustls-post-quantum-test = ["rustls/prefer-post-quantum", "rustls-aws-lc-rs", "proto/__rustls-post-quantum-test"] + [dependencies] async-io = { workspace = true, optional = true } bytes = { workspace = true } @@ -58,7 +63,7 @@ smol = { workspace = true, optional = true } thiserror = { workspace = true } tracing = { workspace = true } tokio = { workspace = true } -udp = { package = "iroh-quinn-udp", path = "../quinn-udp", version = "0.5", default-features = false, features = ["tracing"] } +udp = { package = "iroh-quinn-udp", path = "../quinn-udp", version = "0.6", default-features = false, features = ["tracing"] } # Fix minimal dependencies for indirect deps async-global-executor = { workspace = true, optional = true } @@ -79,7 +84,6 @@ bencher = { workspace = true } directories-next = { workspace = true } rand = { workspace = true } rcgen = { workspace = true } -rustls-pemfile = { workspace = true } clap = { workspace = true } tokio = { workspace = true, features = ["rt", "rt-multi-thread", "time", "macros"] } tracing-subscriber = { workspace = true } @@ -110,6 +114,10 @@ required-features = ["rustls-ring"] name = "connection" required-features = ["rustls-ring"] +[[test]] +name = "post_quantum" +required-features = ["__rustls-post-quantum-test"] + [[bench]] name = "bench" harness = false @@ -117,4 +125,4 @@ required-features = ["rustls-ring"] [package.metadata.docs.rs] # all non-default features except fips (cannot build on docs.rs environment) -features = ["lock_tracking", "rustls-aws-lc-rs", "rustls-ring", "runtime-tokio", "runtime-smol", "log", "rustls-log"] +features = ["lock_tracking", "rustls-aws-lc-rs", "rustls-ring", "runtime-tokio", "runtime-smol", "tracing-log", "rustls-log"] diff --git a/quinn/examples/server.rs b/quinn/examples/server.rs index 8fd954e4d..31abf4c5c 100644 --- a/quinn/examples/server.rs +++ b/quinn/examples/server.rs @@ -14,7 +14,7 @@ use anyhow::{Context, Result, anyhow, bail}; use clap::Parser; use iroh_quinn as quinn; use proto::crypto::rustls::QuicServerConfig; -use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer, pem::PemObject}; use tracing::{error, info, info_span}; use tracing_futures::Instrument as _; @@ -70,19 +70,22 @@ fn main() { #[tokio::main] async fn run(options: Opt) -> Result<()> { let (certs, key) = if let (Some(key_path), Some(cert_path)) = (&options.key, &options.cert) { - let key = fs::read(key_path).context("failed to read private key")?; let key = if key_path.extension().is_some_and(|x| x == "der") { - PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(key)) + PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from( + fs::read(key_path).context("failed to read private key file")?, + )) } else { - rustls_pemfile::private_key(&mut &*key) - .context("malformed PKCS #1 private key")? - .ok_or_else(|| anyhow::Error::msg("no private keys found"))? + PrivateKeyDer::from_pem_file(key_path) + .context("failed to read PEM from private key file")? }; - let cert_chain = fs::read(cert_path).context("failed to read certificate chain")?; + let cert_chain = if cert_path.extension().is_some_and(|x| x == "der") { - vec![CertificateDer::from(cert_chain)] + vec![CertificateDer::from( + fs::read(cert_path).context("failed to read certificate chain file")?, + )] } else { - rustls_pemfile::certs(&mut &*cert_chain) + CertificateDer::pem_file_iter(cert_path) + .context("failed to read PEM from certificate chain file")? .collect::>() .context("invalid PEM-encoded certificate")? }; diff --git a/quinn/src/connection.rs b/quinn/src/connection.rs index a89665736..976fcc84b 100644 --- a/quinn/src/connection.rs +++ b/quinn/src/connection.rs @@ -588,6 +588,38 @@ impl Connection { conn.close(error_code, Bytes::copy_from_slice(reason), &self.0.shared); } + /// Wait for the handshake to be confirmed. + /// + /// As a server, who must be authenticated by clients, + /// this happens when the handshake completes + /// upon receiving a TLS Finished message from the client. + /// In return, the server send a HANDSHAKE_DONE frame. + /// + /// As a client, this happens when receiving a HANDSHAKE_DONE frame. + /// At this point, the server has either accepted our authentication, + /// or, if client authentication is not required, accepted our lack of authentication. + pub async fn handshake_confirmed(&self) -> Result<(), ConnectionError> { + { + let conn = self.0.state.lock("handshake_confirmed"); + if let Some(error) = conn.error.as_ref() { + return Err(error.clone()); + } + if conn.handshake_confirmed { + return Ok(()); + } + // Construct the future while the lock is held to ensure we can't miss a wakeup if + // the `Notify` is signaled immediately after we release the lock. `await` it after + // the lock guard is out of scope. + self.0.shared.handshake_confirmed.notified() + } + .await; + if let Some(error) = self.0.state.lock("handshake_confirmed").error.as_ref() { + Err(error.clone()) + } else { + Ok(()) + } + } + /// Transmit `data` as an unreliable, unordered application datagram /// /// Application datagrams are a low-level primitive. They may be lost or delivered out of order, @@ -1179,6 +1211,7 @@ impl ConnectionRef { on_handshake_data: Some(on_handshake_data), on_connected: Some(on_connected), connected: false, + handshake_confirmed: false, timer: None, timer_deadline: None, conn_events, @@ -1295,6 +1328,7 @@ impl WeakConnectionHandle { #[derive(Debug, Default)] pub(crate) struct Shared { + handshake_confirmed: Notify, /// Notified when new streams may be locally initiated due to an increase in stream ID flow /// control budget stream_budget_available: [Notify; 2], @@ -1312,6 +1346,7 @@ pub(crate) struct State { on_handshake_data: Option>, on_connected: Option>, connected: bool, + handshake_confirmed: bool, timer: Option>>, timer_deadline: Option, conn_events: mpsc::UnboundedReceiver, @@ -1424,11 +1459,10 @@ impl State { self.close(error_code, reason, shared); } Poll::Ready(None) => { - return Err(ConnectionError::TransportError(proto::TransportError { - code: proto::TransportErrorCode::INTERNAL_ERROR, - frame: None, - reason: "endpoint driver future was dropped".to_string(), - })); + return Err(ConnectionError::TransportError(proto::TransportError::new( + proto::TransportErrorCode::INTERNAL_ERROR, + "endpoint driver future was dropped".to_string(), + ))); } Poll::Pending => { return Ok(()); @@ -1460,6 +1494,10 @@ impl State { wake_all_notify(&mut self.stopped); } } + HandshakeConfirmed => { + self.handshake_confirmed = true; + shared.handshake_confirmed.notify_waiters(); + } ConnectionLost { reason } => { self.terminate(reason, shared); } @@ -1599,6 +1637,7 @@ impl State { if let Some(x) = self.on_connected.take() { let _ = x.send(false); } + shared.handshake_confirmed.notify_waiters(); wake_all_notify(&mut self.stopped); shared.closed.notify_waiters(); diff --git a/quinn/src/endpoint.rs b/quinn/src/endpoint.rs index 0cb8ee13b..d1a9d5cb9 100644 --- a/quinn/src/endpoint.rs +++ b/quinn/src/endpoint.rs @@ -11,7 +11,11 @@ use std::{ task::{Context, Poll, RawWaker, RawWakerVTable, Waker}, }; -#[cfg(all(not(wasm_browser), any(feature = "aws-lc-rs", feature = "ring")))] +#[cfg(all( + not(wasm_browser), + any(feature = "runtime-tokio", feature = "runtime-smol"), + any(feature = "aws-lc-rs", feature = "ring"), +))] use crate::runtime::default_runtime; use crate::{ Instant, @@ -25,7 +29,11 @@ use proto::{ EndpointEvent, ServerConfig, }; use rustc_hash::FxHashMap; -#[cfg(all(not(wasm_browser), any(feature = "aws-lc-rs", feature = "ring"),))] +#[cfg(all( + not(wasm_browser), + any(feature = "runtime-tokio", feature = "runtime-smol"), + any(feature = "aws-lc-rs", feature = "ring"), +))] use socket2::{Domain, Protocol, Socket, Type}; use tokio::sync::{Notify, futures::Notified, mpsc}; use tracing::{Instrument, Span}; @@ -67,7 +75,11 @@ impl Endpoint { /// /// Some environments may not allow creation of dual-stack sockets, in which case an IPv6 /// client will only be able to connect to IPv6 servers. An IPv4 client is never dual-stack. - #[cfg(all(not(wasm_browser), any(feature = "aws-lc-rs", feature = "ring")))] // `EndpointConfig::default()` is only available with these + #[cfg(all( + not(wasm_browser), + any(feature = "runtime-tokio", feature = "runtime-smol"), + any(feature = "aws-lc-rs", feature = "ring"), // `EndpointConfig::default()` is only available with these + ))] pub fn client(addr: SocketAddr) -> io::Result { let socket = Socket::new(Domain::for_address(addr), Type::DGRAM, Some(Protocol::UDP))?; if addr.is_ipv6() { @@ -97,7 +109,11 @@ impl Endpoint { /// IPv6 address on Windows will not by default be able to communicate with IPv4 /// addresses. Portable applications should bind an address that matches the family they wish to /// communicate within. - #[cfg(all(not(wasm_browser), any(feature = "aws-lc-rs", feature = "ring")))] // `EndpointConfig::default()` is only available with these + #[cfg(all( + not(wasm_browser), + any(feature = "runtime-tokio", feature = "runtime-smol"), + any(feature = "aws-lc-rs", feature = "ring"), // `EndpointConfig::default()` is only available with these + ))] pub fn server(config: ServerConfig, addr: SocketAddr) -> io::Result { let socket = std::net::UdpSocket::bind(addr)?; let runtime = diff --git a/quinn/src/incoming.rs b/quinn/src/incoming.rs index 8eced8cdc..f11a76e9b 100644 --- a/quinn/src/incoming.rs +++ b/quinn/src/incoming.rs @@ -96,7 +96,7 @@ impl Incoming { /// The original destination CID when initiating the connection pub fn orig_dst_cid(&self) -> ConnectionId { - *self.0.as_ref().unwrap().inner.orig_dst_cid() + self.0.as_ref().unwrap().inner.orig_dst_cid() } } diff --git a/quinn/src/lib.rs b/quinn/src/lib.rs index e17a0dec4..5bdfac607 100644 --- a/quinn/src/lib.rs +++ b/quinn/src/lib.rs @@ -87,7 +87,9 @@ pub use crate::recv_stream::{ReadError, ReadExactError, ReadToEndError, RecvStre pub use crate::runtime::SmolRuntime; #[cfg(feature = "runtime-tokio")] pub use crate::runtime::TokioRuntime; -pub use crate::runtime::{AsyncTimer, AsyncUdpSocket, Runtime, UdpSender, default_runtime}; +#[cfg(any(feature = "runtime-tokio", feature = "runtime-smol"))] +pub use crate::runtime::default_runtime; +pub use crate::runtime::{AsyncTimer, AsyncUdpSocket, Runtime, UdpSender}; pub use crate::send_stream::{SendStream, StoppedError, WriteError}; #[cfg(test)] diff --git a/quinn/src/runtime.rs b/quinn/src/runtime/mod.rs similarity index 97% rename from quinn/src/runtime.rs rename to quinn/src/runtime/mod.rs index a6bb185ef..15a7a1112 100644 --- a/quinn/src/runtime.rs +++ b/quinn/src/runtime/mod.rs @@ -1,10 +1,11 @@ +#[cfg(any(feature = "runtime-tokio", feature = "runtime-smol"))] +use std::sync::Arc; use std::{ fmt::{self, Debug}, future::Future, io::{self, IoSliceMut}, net::SocketAddr, pin::Pin, - sync::Arc, task::{Context, Poll}, }; @@ -17,6 +18,7 @@ pub trait Runtime: Send + Sync + Debug + 'static { /// Construct a timer that will expire at `i` fn new_timer(&self, i: Instant) -> Pin>; /// Drive `future` to completion in the background + #[track_caller] fn spawn(&self, future: Pin + Send>>); /// Convert `t` into the socket type used by this runtime #[cfg(not(wasm_browser))] @@ -218,6 +220,7 @@ trait UdpSenderHelperSocket: Send + Sync + 'static { /// If `runtime-tokio` is enabled and this function is called from within a Tokio runtime context, /// then `TokioRuntime` is returned. Otherwise, if `runtime-smol` is enabled, `SmolRuntime` is /// returned. Otherwise, `None` is returned. +#[cfg(any(feature = "runtime-tokio", feature = "runtime-smol"))] #[allow(clippy::needless_return)] // Be sure we return the right thing pub fn default_runtime() -> Option> { #[cfg(feature = "runtime-tokio")] @@ -238,12 +241,10 @@ pub fn default_runtime() -> Option> { #[cfg(feature = "runtime-tokio")] mod tokio; -// Due to MSRV, we must specify `self::` where there's crate/module ambiguity #[cfg(feature = "runtime-tokio")] -pub use self::tokio::TokioRuntime; +pub use tokio::TokioRuntime; -#[cfg(feature = "async-io")] -mod async_io; -// Due to MSRV, we must specify `self::` where there's crate/module ambiguity #[cfg(feature = "runtime-smol")] -pub use self::async_io::*; +mod smol; +#[cfg(feature = "runtime-smol")] +pub use smol::*; diff --git a/quinn/src/runtime/async_io.rs b/quinn/src/runtime/smol.rs similarity index 67% rename from quinn/src/runtime/async_io.rs rename to quinn/src/runtime/smol.rs index eb5380665..f9bff2194 100644 --- a/quinn/src/runtime/async_io.rs +++ b/quinn/src/runtime/smol.rs @@ -4,44 +4,29 @@ use std::{ task::{Context, Poll}, time::Instant, }; -#[cfg(feature = "runtime-smol")] use std::{io, sync::Arc, task::ready}; -#[cfg(feature = "runtime-smol")] use async_io::Async; use async_io::Timer; use super::AsyncTimer; -#[cfg(feature = "runtime-smol")] use super::{AsyncUdpSocket, Runtime, UdpSender, UdpSenderHelper, UdpSenderHelperSocket}; -#[cfg(feature = "runtime-smol")] -// Due to MSRV, we must specify `self::` where there's crate/module ambiguity -pub use self::smol::SmolRuntime; +/// A Quinn runtime for smol +#[derive(Debug)] +pub struct SmolRuntime; -#[cfg(feature = "runtime-smol")] -mod smol { - use super::*; +impl Runtime for SmolRuntime { + fn new_timer(&self, t: Instant) -> Pin> { + Box::pin(Timer::at(t)) + } - /// A Quinn runtime for smol - #[derive(Debug)] - pub struct SmolRuntime; + fn spawn(&self, future: Pin + Send>>) { + ::smol::spawn(future).detach(); + } - impl Runtime for SmolRuntime { - fn new_timer(&self, t: Instant) -> Pin> { - Box::pin(Timer::at(t)) - } - - fn spawn(&self, future: Pin + Send>>) { - ::smol::spawn(future).detach(); - } - - fn wrap_udp_socket( - &self, - sock: std::net::UdpSocket, - ) -> io::Result> { - Ok(Box::new(UdpSocket::new(sock)?)) - } + fn wrap_udp_socket(&self, sock: std::net::UdpSocket) -> io::Result> { + Ok(Box::new(UdpSocket::new(sock)?)) } } @@ -55,14 +40,12 @@ impl AsyncTimer for Timer { } } -#[cfg(any(feature = "runtime-smol"))] #[derive(Debug, Clone)] struct UdpSocket { io: Arc>, inner: Arc, } -#[cfg(feature = "runtime-smol")] impl UdpSocket { fn new(sock: std::net::UdpSocket) -> io::Result { Ok(Self { @@ -72,7 +55,6 @@ impl UdpSocket { } } -#[cfg(feature = "runtime-smol")] impl UdpSenderHelperSocket for UdpSocket { fn max_transmit_segments(&self) -> usize { self.inner.max_gso_segments() @@ -83,7 +65,6 @@ impl UdpSenderHelperSocket for UdpSocket { } } -#[cfg(feature = "runtime-smol")] impl AsyncUdpSocket for UdpSocket { fn create_sender(&self) -> Pin> { Box::pin(UdpSenderHelper::new(self.clone(), |socket: &Self| { diff --git a/quinn/tests/many_connections.rs b/quinn/tests/many_connections.rs index c54aec1f8..2a3f16931 100644 --- a/quinn/tests/many_connections.rs +++ b/quinn/tests/many_connections.rs @@ -21,12 +21,10 @@ struct Shared { #[test] #[ignore] fn connect_n_nodes_to_1_and_send_1mb_data() { - tracing::subscriber::set_global_default( - tracing_subscriber::FmtSubscriber::builder() - .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) - .finish(), - ) - .unwrap(); + let _ = tracing_subscriber::FmtSubscriber::builder() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .try_init(); let runtime = Builder::new_current_thread().enable_all().build().unwrap(); let _guard = runtime.enter(); diff --git a/quinn/tests/post_quantum.rs b/quinn/tests/post_quantum.rs new file mode 100644 index 000000000..421e97219 --- /dev/null +++ b/quinn/tests/post_quantum.rs @@ -0,0 +1,124 @@ +#![cfg(feature = "rustls-aws-lc-rs")] + +use std::{ + error::Error, + net::{Ipv4Addr, SocketAddr}, + sync::Arc, +}; + +use rustls::{ + NamedGroup, + pki_types::{CertificateDer, PrivatePkcs8KeyDer}, +}; +use tracing::info; + +use iroh_quinn::{ + Endpoint, + crypto::rustls::{HandshakeData, QuicClientConfig, QuicServerConfig}, +}; + +#[tokio::test] +async fn post_quantum_key_worst_case_header() { + check_post_quantum_key_exchange(1274).await; +} + +#[tokio::test] +async fn post_quantum_key_exchange_large_mtu() { + check_post_quantum_key_exchange(1433).await; +} + +async fn check_post_quantum_key_exchange(min_mtu: u16) { + let _ = tracing_subscriber::FmtSubscriber::builder() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .with_test_writer() + .try_init(); + + let server_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, 0)); + + let (endpoint, server_cert) = make_server_endpoint(server_addr, min_mtu).unwrap(); + let server_addr = endpoint.local_addr().unwrap(); + // accept a single connection + let jh = tokio::spawn(async move { + let incoming_conn = endpoint.accept().await.unwrap(); + let conn = incoming_conn.await.unwrap(); + info!( + "[server] connection accepted: addr={}", + conn.remote_address() + ); + assert_eq!( + conn.handshake_data() + .unwrap() + .downcast::() + .unwrap() + .negotiated_key_exchange_group, + NamedGroup::X25519MLKEM768 + ) + }); + + let endpoint = + make_client_endpoint(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)), server_cert).unwrap(); + // connect to server + let connection = endpoint + .connect(server_addr, "localhost") + .unwrap() + .await + .unwrap(); + info!("[client] connected: addr={}", connection.remote_address()); + + // Waiting for a stream will complete with an error when the server closes the connection + let _ = connection.accept_uni().await; + + // Make sure the server has a chance to clean up + endpoint.wait_idle().await; + jh.await.unwrap(); +} + +fn make_client_endpoint( + bind_addr: SocketAddr, + server_cert: CertificateDer<'static>, +) -> Result> { + let mut certs = rustls::RootCertStore::empty(); + certs.add(server_cert)?; + let rustls_config = rustls::ClientConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_root_certificates(certs) + .with_no_client_auth(); + + let client_cfg = + iroh_quinn::ClientConfig::new(Arc::new(QuicClientConfig::try_from(rustls_config).unwrap())); + let mut endpoint = Endpoint::client(bind_addr)?; + endpoint.set_default_client_config(client_cfg); + Ok(endpoint) +} + +fn make_server_endpoint( + bind_addr: SocketAddr, + min_mtu: u16, +) -> Result<(Endpoint, CertificateDer<'static>), Box> { + let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap(); + let key = PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der()); + let cert = CertificateDer::from(cert.cert); + let mut server_config = iroh_quinn::ServerConfig::with_crypto(Arc::new( + QuicServerConfig::try_from( + rustls::ServerConfig::builder_with_provider(Arc::new( + rustls::crypto::aws_lc_rs::default_provider(), + )) + .with_safe_default_protocol_versions() + .unwrap() + .with_no_client_auth() + .with_single_cert(vec![cert.clone()], key.into()) + .unwrap(), + ) + .unwrap(), + )); + + let transport_config = Arc::get_mut(&mut server_config.transport).unwrap(); + transport_config.max_concurrent_uni_streams(0_u8.into()); + transport_config.min_mtu(min_mtu); + + let endpoint = Endpoint::server(server_config, bind_addr)?; + Ok((endpoint, cert)) +}