Merge remote-tracking branch 'origin/main-iroh' into matheus23/fix-path-status-send

This commit is contained in:
Philipp Krüger
2025-12-02 11:31:27 +01:00
45 changed files with 1470 additions and 461 deletions
+2 -2
View File
@@ -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
+92 -113
View File
@@ -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}}"
Generated
+500 -19
View File
@@ -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]]
+2 -4
View File
@@ -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
+1 -1
View File
@@ -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,
+7 -1
View File
@@ -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 }
+27 -23
View File
@@ -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);
}
}
+4 -4
View File
@@ -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<crypto::Keys, crypto::UnsupportedVersion> {
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)
}
+11 -14
View File
@@ -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::<Result<_, _>>()
.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::<Result<_, _>>()
.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()];
+7 -3
View File
@@ -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]
+3 -3
View File
@@ -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();
}
}
+7
View File
@@ -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);
+1
View File
@@ -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;
+77 -50
View File
@@ -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<Instant>,
}
/// 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<CubicConfig>,
/// 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<Instant>,
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<State>,
}
impl Cubic {
/// Construct a state using the given `config` and current time `now`
pub fn new(config: Arc<CubicConfig>, _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,
}
}
+1
View File
@@ -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 {
+73 -18
View File
@@ -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<u64> = 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<Instant> = 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.
+17 -8
View File
@@ -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);
}
+49 -33
View File
@@ -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(
+8 -5
View File
@@ -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);
+13
View File
@@ -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<u64, SentPacket>,
/// Packets that were deemed lost
// Older packets are regularly removed in `Connection::drain_lost_packets`.
pub(super) lost_packets: BTreeMap<u64, LostPacket>,
/// 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)]
+5 -8
View File
@@ -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<KeyPair<Box<dyn PacketKey>>>;
/// 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<Keys, UnsupportedVersion>;
fn initial_keys(&self, version: u32, dst_cid: ConnectionId)
-> Result<Keys, UnsupportedVersion>;
/// 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
///
+20 -8
View File
@@ -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<Box<dyn Any>> {
@@ -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<String>,
/// 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<Keys, UnsupportedVersion> {
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);
+23 -24
View File
@@ -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<u8>,
server_config: Option<Arc<ServerConfig>>,
) -> Result<(ConnectionHandle, Connection), AcceptError> {
) -> Result<(ConnectionHandle, Connection), Box<AcceptError>> {
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<u8>,
) -> 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<RouteDatagramTo> {
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
}
}
+7 -7
View File
@@ -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,
}
}
+143 -3
View File
@@ -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<F: FnOnce(&mut ServerConfig)>(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<F: FnOnce(&mut ServerConfig)>(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);
}
+16
View File
@@ -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 {
+15 -8
View File
@@ -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<RangeFrom<u16>> = Mutex::new(4433..);
pub static ref CLIENT_PORTS: Mutex<RangeFrom<u16>> = Mutex::new(44433..);
pub(crate) static ref CERTIFIED_KEY: rcgen::CertifiedKey<rcgen::KeyPair> =
rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
}
pub(crate) static SERVER_PORTS: LazyLock<Mutex<RangeFrom<u16>>> =
LazyLock::new(|| Mutex::new(4433..));
pub(crate) static CLIENT_PORTS: LazyLock<Mutex<RangeFrom<u16>>> =
LazyLock::new(|| Mutex::new(44433..));
pub(crate) static CERTIFIED_KEY: LazyLock<rcgen::CertifiedKey<rcgen::KeyPair>> =
LazyLock::new(|| rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap());
#[derive(Default)]
struct SimpleTokenLog(Mutex<HashSet<u128>>);
+1 -2
View File
@@ -247,9 +247,8 @@ impl Token {
fn decode(key: &dyn HandshakeTokenKey, raw_token_bytes: &[u8]) -> Option<Self> {
// Decrypt
// MSRV: split_at_checked requires 1.80.0
let nonce_slice_start = raw_token_bytes.len().checked_sub(size_of::<u128>())?;
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());
+31 -2
View File
@@ -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<frame::FrameType>,
/// Human-readable explanation of the reason
pub reason: String,
/// An underlying crypto (e.g. TLS) layer error
pub crypto: Option<Arc<dyn std::error::Error + Send + Sync>>,
}
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<Code> 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,
}
}
)*
+4 -5
View File
@@ -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]
+2 -1
View File
@@ -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: {
+1
View File
@@ -72,6 +72,7 @@ impl UdpSocketState {
addr: addr.as_socket().unwrap(),
ecn: None,
dst_ip: None,
interface_index: None,
};
Ok(1)
}
+10 -6
View File
@@ -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<IpAddr>,
/// The interface index of the interface on which the datagram was received
pub interface_index: Option<u32>,
}
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<Instant>,
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<Instant>, _: impl core::fmt::Debug, _: &Transmit) {}
/// A borrowed UDP socket
+41 -11
View File
@@ -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<usize> {
let mut names = [MaybeUninit::<libc::sockaddr_storage>::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<usize> {
let mut name = MaybeUninit::<libc::sockaddr_storage>::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<RecvMeta> {
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::<u8, libc::cmsghdr>(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::<u8, libc::cmsghdr>(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::<libc::in6_pktinfo, libc::cmsghdr>(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))]
+7 -5
View File
@@ -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::<WinSock::IN6_PKTINFO, WinSock::CMSGHDR>(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<WinSock::LPFN_WSARECVMSG> = Lazy::new(|| {
static WSARECVMSG_PTR: LazyLock<WinSock::LPFN_WSARECVMSG> = 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<WinSock::LPFN_WSARECVMSG> = Lazy::new(|| {
wsa_recvmsg_ptr
});
static MAX_GSO_SEGMENTS: Lazy<usize> = Lazy::new(|| {
static MAX_GSO_SEGMENTS: LazyLock<usize> = LazyLock::new(|| {
let socket = match std::net::UdpSocket::bind("[::]:0")
.or_else(|_| std::net::UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)))
{
+14 -6
View File
@@ -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"]
+12 -9
View File
@@ -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::<Result<_, _>>()
.context("invalid PEM-encoded certificate")?
};
+44 -5
View File
@@ -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<oneshot::Sender<()>>,
on_connected: Option<oneshot::Sender<bool>>,
connected: bool,
handshake_confirmed: bool,
timer: Option<Pin<Box<dyn AsyncTimer>>>,
timer_deadline: Option<Instant>,
conn_events: mpsc::UnboundedReceiver<ConnectionEvent>,
@@ -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();
+20 -4
View File
@@ -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<Self> {
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<Self> {
let socket = std::net::UdpSocket::bind(addr)?;
let runtime =
+1 -1
View File
@@ -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()
}
}
+3 -1
View File
@@ -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)]
@@ -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<Box<dyn AsyncTimer>>;
/// Drive `future` to completion in the background
#[track_caller]
fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + 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<Arc<dyn Runtime>> {
#[cfg(feature = "runtime-tokio")]
@@ -238,12 +241,10 @@ pub fn default_runtime() -> Option<Arc<dyn Runtime>> {
#[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::*;
@@ -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<dyn AsyncTimer>> {
Box::pin(Timer::at(t))
}
/// A Quinn runtime for smol
#[derive(Debug)]
pub struct SmolRuntime;
fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
::smol::spawn(future).detach();
}
impl Runtime for SmolRuntime {
fn new_timer(&self, t: Instant) -> Pin<Box<dyn AsyncTimer>> {
Box::pin(Timer::at(t))
}
fn spawn(&self, future: Pin<Box<dyn Future<Output = ()> + Send>>) {
::smol::spawn(future).detach();
}
fn wrap_udp_socket(
&self,
sock: std::net::UdpSocket,
) -> io::Result<Box<dyn AsyncUdpSocket>> {
Ok(Box::new(UdpSocket::new(sock)?))
}
fn wrap_udp_socket(&self, sock: std::net::UdpSocket) -> io::Result<Box<dyn AsyncUdpSocket>> {
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<Async<std::net::UdpSocket>>,
inner: Arc<udp::UdpSocketState>,
}
#[cfg(feature = "runtime-smol")]
impl UdpSocket {
fn new(sock: std::net::UdpSocket) -> io::Result<Self> {
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<dyn UdpSender>> {
Box::pin(UdpSenderHelper::new(self.clone(), |socket: &Self| {
+4 -6
View File
@@ -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();
+124
View File
@@ -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::<HandshakeData>()
.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<Endpoint, Box<dyn Error + Send + Sync + 'static>> {
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<dyn Error + Send + Sync + 'static>> {
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))
}