From a6dd73a41f4d7d2a23d90ff2039216782c2f1612 Mon Sep 17 00:00:00 2001 From: Alex Emmet <111742636+Alex-Emmet@users.noreply.github.com> Date: Tue, 14 Jul 2026 01:55:31 +0200 Subject: [PATCH] [add] ping jitter, receiver backpressure, stream frame limits --- Cargo.lock | 270 ++++++++++----------------------- client/Cargo.toml | 6 + client/src/lib.rs | 19 ++- client/tests/ping.rs | 80 ++++++++++ transport/Cargo.toml | 1 + transport/src/connection.rs | 118 +++++++++++++- transport/tests/integration.rs | 146 ++++++++++++++++++ 7 files changed, 450 insertions(+), 190 deletions(-) create mode 100644 client/tests/ping.rs diff --git a/Cargo.lock b/Cargo.lock index a507421..d84909c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -176,9 +176,9 @@ checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" [[package]] name = "bytes" -version = "1.12.0" +version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8ae3f5d315924270530207e2a68396c3cc547f6dca3fbdca317cfb1a51edb593" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "cast" @@ -188,9 +188,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5" [[package]] name = "cc" -version = "1.2.65" +version = "1.2.67" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e228eec9be7c17ccb640b59b36a5cd805ea2a564a4c5e162c2f659fea30d3b96" +checksum = "e17dd265a7d0f31ef544e1b20e03add05d3b45b491b633b10d67145d2acc1a38" dependencies = [ "find-msvc-tools", "jobserver", @@ -221,6 +221,17 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chacha20poly1305" version = "0.10.1" @@ -228,7 +239,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "10cd79432192d1c0f4e1a0fef9527696cc039165d729fb41b3f4f4f354c2dc35" dependencies = [ "aead", - "chacha20", + "chacha20 0.9.1", "cipher", "poly1305", "zeroize", @@ -402,9 +413,9 @@ dependencies = [ [[package]] name = "der" -version = "0.8.0" +version = "0.8.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "71fd89660b2dc699704064e59e9dba0147b903e85319429e131620d022be411b" +checksum = "a69dedd701da44b0536442edf09c81a64b0ab97a7a4a5e3d1971f00027cbc63d" dependencies = [ "const-oid 0.10.2", "zeroize", @@ -592,20 +603,6 @@ dependencies = [ "wasm-bindgen", ] -[[package]] -name = "getrandom" -version = "0.3.4" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "899def5c37c4fd7b2664648c28120ecec138e4d395b459e5ca34f9cce2dd77fd" -dependencies = [ - "cfg-if", - "js-sys", - "libc", - "r-efi 5.3.0", - "wasip2", - "wasm-bindgen", -] - [[package]] name = "getrandom" version = "0.4.3" @@ -615,7 +612,7 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi 6.0.0", + "r-efi", "rand_core 0.10.1", "wasm-bindgen", ] @@ -806,11 +803,11 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" [[package]] name = "jobserver" -version = "0.1.34" +version = "0.1.35" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +checksum = "1c00acbd29eabad4a2392fa0e921c874934dbbf4194312ad20f04a0ed67a3cb3" dependencies = [ - "getrandom 0.3.4", + "getrandom 0.4.3", "libc", ] @@ -891,9 +888,9 @@ checksum = "112b39cec0b298b6c1999fee3e31427f74f676e4cb9879ed1a121b43661a4154" [[package]] name = "memchr" -version = "2.8.2" +version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "88904434abc2901f197fe8cc55f0445e7ded921dba5911dad2e2b39b48e663c4" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" [[package]] name = "minicov" @@ -913,9 +910,9 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a" [[package]] name = "mio" -version = "1.2.1" +version = "1.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "02bd0af71c67b473010cbbc60715ee815645a4dc942899111f494b4b737d6fda" +checksum = "30d65c71f1ce40ab09135ce117d742b9f8a19ff91a41a8b57ed50bc2de59c427" dependencies = [ "libc", "wasi", @@ -987,7 +984,7 @@ dependencies = [ "mtp-host", "mtp-transport", "mtp-type-map", - "rand 0.8.6", + "rand 0.8.7", "rcgen", "tokio", ] @@ -999,9 +996,12 @@ dependencies = [ "mtp-codec", "mtp-common", "mtp-crypto", + "mtp-host", "mtp-transport", - "rand 0.8.6", + "rand 0.8.7", + "rcgen", "tokio", + "tracing", ] [[package]] @@ -1013,7 +1013,7 @@ dependencies = [ "mtp-common", "mtp-crypto", "mtp-type-map", - "rand 0.8.6", + "rand 0.8.7", ] [[package]] @@ -1061,7 +1061,7 @@ dependencies = [ "mtp-common", "mtp-crypto", "mtp-transport", - "rand 0.8.6", + "rand 0.8.7", "tokio", ] @@ -1076,6 +1076,7 @@ dependencies = [ "rustls", "rustls-native-certs", "tokio", + "tracing", "wtransport", ] @@ -1269,7 +1270,7 @@ version = "0.11.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "451913da69c775a56034ea8d9003d27ee8948e12443eae7c038ba100a4f21cb7" dependencies = [ - "der 0.8.0", + "der 0.8.1", "spki 0.8.0", ] @@ -1357,15 +1358,16 @@ dependencies = [ [[package]] name = "quinn-proto" -version = "0.11.15" +version = "0.11.16" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4fcb935c5bec503c2f0e306bdd3e58bb9029dcb14fa8d9ac76e3a5256ac0763e" +checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" dependencies = [ "aws-lc-rs", "bytes", - "getrandom 0.3.4", + "getrandom 0.4.3", "lru-slab", - "rand 0.9.4", + "rand 0.10.2", + "rand_pcg", "ring", "rustc-hash", "rustls", @@ -1379,16 +1381,16 @@ dependencies = [ [[package]] name = "quinn-udp" -version = "0.5.14" +version = "0.5.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "addec6a0dcad8a8d96a771f815f0eaf55f9d1805756410b39f5fa81332574cbd" +checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694" dependencies = [ "cfg_aliases", "libc", "once_cell", "socket2", "tracing", - "windows-sys 0.60.2", + "windows-sys 0.61.2", ] [[package]] @@ -1400,12 +1402,6 @@ dependencies = [ "proc-macro2", ] -[[package]] -name = "r-efi" -version = "5.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" - [[package]] name = "r-efi" version = "6.0.0" @@ -1414,23 +1410,24 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" [[package]] name = "rand" -version = "0.8.6" +version = "0.8.7" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a" +checksum = "22f6172bdec972074665ed81ed53b71da00bfc44b65a753cfde883ec4c702a1a" dependencies = [ "libc", - "rand_chacha 0.3.1", + "rand_chacha", "rand_core 0.6.4", ] [[package]] name = "rand" -version = "0.9.4" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "44c5af06bb1b7d3216d91932aed5265164bf384dc89cd6ba05cf59a35f5f76ea" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" dependencies = [ - "rand_chacha 0.9.0", - "rand_core 0.9.5", + "chacha20 0.10.1", + "getrandom 0.4.3", + "rand_core 0.10.1", ] [[package]] @@ -1443,16 +1440,6 @@ dependencies = [ "rand_core 0.6.4", ] -[[package]] -name = "rand_chacha" -version = "0.9.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d3022b5f1df60f26e1ffddd6c66e8aa15de382ae63b3a0c1bfc0e4d3e3f325cb" -dependencies = [ - "ppv-lite86", - "rand_core 0.9.5", -] - [[package]] name = "rand_core" version = "0.6.4" @@ -1462,21 +1449,21 @@ dependencies = [ "getrandom 0.2.17", ] -[[package]] -name = "rand_core" -version = "0.9.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "76afc826de14238e6e8c374ddcc1fa19e374fd8dd986b0d2af0d02377261d83c" -dependencies = [ - "getrandom 0.3.4", -] - [[package]] name = "rand_core" version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rand_pcg" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "caa0f4137e1c0a72f4c651489402276c8e8e1cf081f3b0ba156d2cbeef09e86a" +dependencies = [ + "rand_core 0.10.1", +] + [[package]] name = "rcgen" version = "0.14.8" @@ -1541,9 +1528,9 @@ dependencies = [ [[package]] name = "rustls" -version = "0.23.41" +version = "0.23.42" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "6b92b125634d9b795e7beca796cc790df15a7fb38323bf3196fda83292d06b1f" +checksum = "3c54fcab019b409d04215d3a17cb438fd7fbf192ee61461f20f4fe18704bc138" dependencies = [ "aws-lc-rs", "log", @@ -1591,9 +1578,9 @@ dependencies = [ [[package]] name = "rustversion" -version = "1.0.22" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" +checksum = "cf54715a573b99ac80df0bc206da022bcd442c974952c7b9720069370852e21f" [[package]] name = "ryu" @@ -1802,9 +1789,9 @@ checksum = "8ed6a63f02c8539c91a8685a86f4099661ba3da017932f6ebbea6de3f0fa7c90" [[package]] name = "socket2" -version = "0.6.4" +version = "0.6.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "52d1cfed4120b4d927bf7c0f86d2087a4a7d6027c906d9f9d525a80573b9be51" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" dependencies = [ "libc", "windows-sys 0.61.2", @@ -1827,7 +1814,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1d9efca8738c78ee9484207732f728b1ef517bbb1833d6fc0879ca898a522f6f" dependencies = [ "base64ct", - "der 0.8.0", + "der 0.8.1", ] [[package]] @@ -1952,9 +1939,9 @@ dependencies = [ [[package]] name = "tinyvec" -version = "1.11.0" +version = "1.12.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "3e61e67053d25a4e82c844e8424039d9745781b3fc4f32b8d55ed50f5f667ef3" +checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f" dependencies = [ "tinyvec_macros", ] @@ -2104,15 +2091,6 @@ version = "0.11.1+wasi-snapshot-preview1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" -[[package]] -name = "wasip2" -version = "1.0.4+wasi-0.2.12" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b67efb37e106e55ce722a510d6b5f9c17f083e5fc79afc2badeb12cc313d9487" -dependencies = [ - "wit-bindgen", -] - [[package]] name = "wasm-bindgen" version = "0.2.126" @@ -2238,16 +2216,7 @@ version = "0.52.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d" dependencies = [ - "windows-targets 0.52.6", -] - -[[package]] -name = "windows-sys" -version = "0.60.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "f2f500e4d28234f72040990ec9d39e3a6b950f9f22d3dba18416c35882612bcb" -dependencies = [ - "windows-targets 0.53.5", + "windows-targets", ] [[package]] @@ -2265,31 +2234,14 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9b724f72796e036ab90c1021d4780d4d3d648aca59e491e6b98e725b84e99973" dependencies = [ - "windows_aarch64_gnullvm 0.52.6", - "windows_aarch64_msvc 0.52.6", - "windows_i686_gnu 0.52.6", - "windows_i686_gnullvm 0.52.6", - "windows_i686_msvc 0.52.6", - "windows_x86_64_gnu 0.52.6", - "windows_x86_64_gnullvm 0.52.6", - "windows_x86_64_msvc 0.52.6", -] - -[[package]] -name = "windows-targets" -version = "0.53.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4945f9f551b88e0d65f3db0bc25c33b8acea4d9e41163edf90dcd0b19f9069f3" -dependencies = [ - "windows-link", - "windows_aarch64_gnullvm 0.53.1", - "windows_aarch64_msvc 0.53.1", - "windows_i686_gnu 0.53.1", - "windows_i686_gnullvm 0.53.1", - "windows_i686_msvc 0.53.1", - "windows_x86_64_gnu 0.53.1", - "windows_x86_64_gnullvm 0.53.1", - "windows_x86_64_msvc 0.53.1", + "windows_aarch64_gnullvm", + "windows_aarch64_msvc", + "windows_i686_gnu", + "windows_i686_gnullvm", + "windows_i686_msvc", + "windows_x86_64_gnu", + "windows_x86_64_gnullvm", + "windows_x86_64_msvc", ] [[package]] @@ -2298,102 +2250,48 @@ version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32a4622180e7a0ec044bb555404c800bc9fd9ec262ec147edd5989ccd0c02cd3" -[[package]] -name = "windows_aarch64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a9d8416fa8b42f5c947f8482c43e7d89e73a173cead56d044f6a56104a6d1b53" - [[package]] name = "windows_aarch64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "09ec2a7bb152e2252b53fa7803150007879548bc709c039df7627cabbd05d469" -[[package]] -name = "windows_aarch64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b9d782e804c2f632e395708e99a94275910eb9100b2114651e04744e9b125006" - [[package]] name = "windows_i686_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8e9b5ad5ab802e97eb8e295ac6720e509ee4c243f69d781394014ebfe8bbfa0b" -[[package]] -name = "windows_i686_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "960e6da069d81e09becb0ca57a65220ddff016ff2d6af6a223cf372a506593a3" - [[package]] name = "windows_i686_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0eee52d38c090b3caa76c563b86c3a4bd71ef1a819287c19d586d7334ae8ed66" -[[package]] -name = "windows_i686_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "fa7359d10048f68ab8b09fa71c3daccfb0e9b559aed648a8f95469c27057180c" - [[package]] name = "windows_i686_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "240948bc05c5e7c6dabba28bf89d89ffce3e303022809e73deaefe4f6ec56c66" -[[package]] -name = "windows_i686_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1e7ac75179f18232fe9c285163565a57ef8d3c89254a30685b57d83a38d326c2" - [[package]] name = "windows_x86_64_gnu" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "147a5c80aabfbf0c7d901cb5895d1de30ef2907eb21fbbab29ca94c5b08b1a78" -[[package]] -name = "windows_x86_64_gnu" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "9c3842cdd74a865a8066ab39c8a7a473c0778a3f29370b5fd6b4b9aa7df4a499" - [[package]] name = "windows_x86_64_gnullvm" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "24d5b23dc417412679681396f2b49f3de8c1473deb516bd34410872eff51ed0d" -[[package]] -name = "windows_x86_64_gnullvm" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0ffa179e2d07eee8ad8f57493436566c7cc30ac536a3379fdf008f47f6bb7ae1" - [[package]] name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" -[[package]] -name = "windows_x86_64_msvc" -version = "0.53.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6bbff5f0aada427a1e5a6da5f1f98158182f26556f345ac9e04d36d0ebed650" - -[[package]] -name = "wit-bindgen" -version = "0.57.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e" - [[package]] name = "writeable" version = "0.6.3" @@ -2502,18 +2400,18 @@ dependencies = [ [[package]] name = "zerocopy" -version = "0.8.52" +version = "0.8.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "ce1022995ff5ff5d841ad7d994facc23098cd40152f2c1d11cd607c6f530653f" +checksum = "b7cbbc0a705a0fd05cc3676525980d2bf5a9bc4adac6d6475209a7887cf59d19" dependencies = [ "zerocopy-derive", ] [[package]] name = "zerocopy-derive" -version = "0.8.52" +version = "0.8.54" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "1ae7f38b72ec2a254e2b87ef277cf2cd4fb97cbebf944faa6f33354da0867930" +checksum = "e2e817b7b52d0c7358d3246da9d69935ebb18116b2b102b4230dac079b4862f5" dependencies = [ "proc-macro2", "quote", @@ -2596,6 +2494,6 @@ dependencies = [ [[package]] name = "zmij" -version = "1.0.21" +version = "1.0.23" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/client/Cargo.toml b/client/Cargo.toml index b913851..693378e 100644 --- a/client/Cargo.toml +++ b/client/Cargo.toml @@ -10,6 +10,12 @@ mtp-transport = { version = "0.1.0", path = "../transport" } mtp-crypto = { version = "0.1.0", path = "../crypto", optional = true } rand = "0.8" tokio = { version = "1", features = ["rt", "sync", "time"] } +tracing = "0.1" + +[dev-dependencies] +mtp-host = { version = "0.1.0", path = "../host" } +mtp-transport = { version = "0.1.0", path = "../transport", features = ["host"] } +rcgen = "0.14" [features] crypto = ["dep:mtp-crypto", "mtp-codec/crypto"] diff --git a/client/src/lib.rs b/client/src/lib.rs index 9c8b49b..860ae85 100644 --- a/client/src/lib.rs +++ b/client/src/lib.rs @@ -2,14 +2,14 @@ use mtp_codec::{CommunicationValue, DataType, DataValue, PROTOCOL_VERSION, Versi use mtp_common::CommunicationError; use std::collections::HashMap; use std::sync::Arc; -use tokio::sync::{mpsc, Mutex}; +use rand::Rng; +use tokio::sync::{Mutex, mpsc}; use tokio::time::{Duration, Instant}; pub use MTPClient as Client; pub use MTPConnection as Connection; pub use mtp_transport::Policy; pub use mtp_transport::Receiver; -#[cfg(feature = "streaming")] pub use mtp_transport::SendMode; pub use mtp_transport::Sender; @@ -34,6 +34,7 @@ pub struct ClientConfig { pub description: Option, pub policy: Policy, pub ping_interval: Duration, + pub ping_jitter: Option, pub max_missed_pings: usize, pub ping_timestamp: bool, #[cfg(feature = "crypto")] @@ -55,6 +56,7 @@ impl ClientConfig { description: None, policy: Policy::default(), ping_interval: Duration::ZERO, + ping_jitter: None, max_missed_pings: 3, ping_timestamp: true, #[cfg(feature = "crypto")] @@ -91,6 +93,11 @@ impl ClientConfig { self } + pub fn with_ping_jitter(mut self, jitter: Option) -> Self { + self.ping_jitter = jitter; + self + } + pub fn with_max_missed_pings(mut self, max_missed_pings: usize) -> Self { self.max_missed_pings = max_missed_pings; self @@ -209,6 +216,7 @@ fn start_ping_session( let last_ping = Arc::new(Mutex::new(None)); let ping_state = last_ping.clone(); let interval = config.ping_interval; + let ping_jitter = config.ping_jitter; let max_missed_pings = config.max_missed_pings; let ping_timestamp = config.ping_timestamp; let mut close_rx = receiver.handle().subscribe_close(); @@ -231,6 +239,12 @@ fn start_ping_session( break; } + if let Some(jitter) = ping_jitter && !jitter.is_zero() { + let max_ms = jitter.as_millis() as u64; + let extra = rand::thread_rng().gen_range(0..=max_ms); + tokio::time::sleep(Duration::from_millis(extra)).await; + } + let mut ping = CommunicationValue::new(mtp_codec::CommunicationType::Ping); if ping_timestamp { let sent_at = std::time::SystemTime::now() @@ -817,6 +831,7 @@ mod tests { .with_max_missed_pings(2) .with_ping_timestamp(false); assert_eq!(config.ping_interval, Duration::from_secs(5)); + assert_eq!(config.ping_jitter, None); assert_eq!(config.max_missed_pings, 2); assert!(!config.ping_timestamp); } diff --git a/client/tests/ping.rs b/client/tests/ping.rs new file mode 100644 index 0000000..b071864 --- /dev/null +++ b/client/tests/ping.rs @@ -0,0 +1,80 @@ +use std::net::{IpAddr, Ipv4Addr}; + +use mtp_client::{ClientConfig, MTPClient}; +use mtp_host::{HostConfig, MTPHost}; + +fn generate_self_signed_cert() -> (Vec, Vec) { + let key_pair = rcgen::KeyPair::generate().unwrap(); + let params = + rcgen::CertificateParams::new(vec!["localhost".into(), "127.0.0.1".into()]).unwrap(); + let cert = params.self_signed(&key_pair).unwrap(); + let cert_pem = cert.pem(); + let key_pem = key_pair.serialize_pem(); + (cert_pem.into_bytes(), key_pem.into_bytes()) +} + +async fn start_host(send_pongs: bool) -> (MTPHost, Vec) { + let (cert_pem, key_pem) = generate_self_signed_cert(); + let host = MTPHost::new( + HostConfig::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 0, cert_pem.clone(), key_pem) + .with_pongs(send_pongs), + ) + .await + .unwrap(); + (host, cert_pem) +} + +#[tokio::test] +async fn test_ping_rtt_and_missed_ping_teardown() { + let (mut host, cert_pem) = start_host(true).await; + let url = format!("https://127.0.0.1:{}", host.local_addr().port()); + + let client = MTPClient::connect( + ClientConfig::new(url) + .with_pinned_pem(cert_pem) + .with_ping_interval(std::time::Duration::from_millis(25)) + .with_max_missed_pings(3), + ) + .await + .unwrap(); + + let _accepted = host.accept().await.unwrap().unwrap(); + + let ping = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if let Some(ping) = client.get_ping() { + return ping; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await + .unwrap(); + + assert!(ping > std::time::Duration::ZERO); + + let (mut silent_host, silent_cert_pem) = start_host(false).await; + let silent_url = format!("https://127.0.0.1:{}", silent_host.local_addr().port()); + let silent_client = MTPClient::connect( + ClientConfig::new(silent_url) + .with_pinned_pem(silent_cert_pem) + .with_ping_interval(std::time::Duration::from_millis(25)) + .with_max_missed_pings(2), + ) + .await + .unwrap(); + + let _accepted = silent_host.accept().await.unwrap().unwrap(); + + let closed = tokio::time::timeout(std::time::Duration::from_secs(5), async { + loop { + if silent_client.sender.is_closed() { + return; + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } + }) + .await; + + assert!(closed.is_ok(), "client should close after missed pings"); +} diff --git a/transport/Cargo.toml b/transport/Cargo.toml index 69c85b4..6a72928 100644 --- a/transport/Cargo.toml +++ b/transport/Cargo.toml @@ -16,6 +16,7 @@ tokio = { version = "1", features = ["full"] } rustls-native-certs = "0.8.4" log = "0.4" rcgen = "0.14" +tracing = "0.1" [dev-dependencies] diff --git a/transport/src/connection.rs b/transport/src/connection.rs index b4d6e8a..a726a74 100644 --- a/transport/src/connection.rs +++ b/transport/src/connection.rs @@ -2,9 +2,10 @@ use crate::ConnectionHandle; use mtp_codec::CommunicationValue; use mtp_common::CommunicationError; use std::sync::Arc; -use tokio::sync::{mpsc, Mutex, RwLock, Semaphore}; +use tokio::sync::{mpsc, Mutex, Notify, RwLock, Semaphore}; use tokio::time::{Duration, sleep, timeout}; use wtransport::Connection; +use tracing::{debug, info, instrument, trace}; const APPLICATION_CLOSE_REASON: &str = "mtp-close"; @@ -33,6 +34,7 @@ pub struct Policy { pub persistent_stream_retry_backoff: Duration, pub receiver_queue_capacity: usize, pub max_concurrent_stream_tasks: usize, + pub max_frames_per_stream: Option, } impl Default for Policy { @@ -55,6 +57,7 @@ impl Default for Policy { persistent_stream_retry_backoff: Duration::from_millis(20), receiver_queue_capacity: 1000, max_concurrent_stream_tasks: 128, + max_frames_per_stream: None, } } } @@ -114,6 +117,11 @@ impl Policy { self.max_concurrent_stream_tasks = max_concurrent_stream_tasks; self } + + pub fn with_max_frames_per_stream(mut self, max_frames_per_stream: Option) -> Self { + self.max_frames_per_stream = max_frames_per_stream; + self + } } enum ReceivedFrame { @@ -142,6 +150,7 @@ impl Sender { } } + #[instrument(skip(stream, data, policy), level = "trace")] async fn write_frame( stream: &mut wtransport::SendStream, data: &CommunicationValue, @@ -189,6 +198,7 @@ impl Sender { } } + #[instrument(skip(conn, policy), level = "trace")] async fn open_uni_stream( conn: &Connection, policy: &Policy, @@ -221,6 +231,7 @@ impl Sender { } } + #[instrument(skip(conn, stream_opt, data, policy), level = "trace")] async fn send_on_persistent_stream( conn: &Connection, stream_opt: &mut Option, @@ -254,6 +265,7 @@ impl Sender { } } + #[instrument(skip(conn, data, policy), level = "trace")] async fn send_on_single_stream( conn: &Connection, data: &CommunicationValue, @@ -279,6 +291,7 @@ impl Sender { } } + #[instrument(skip(conn, policy), level = "trace")] async fn send_close_frame( conn: &Connection, policy: &Policy, @@ -319,6 +332,7 @@ impl Sender { Ok(()) } + #[instrument(skip(self, data), level = "trace")] pub async fn send(&self, data: &CommunicationValue) -> Result<(), CommunicationError> { if self.handle.is_closed() { return Err(self @@ -374,6 +388,7 @@ impl Sender { } } + #[instrument(skip(self), level = "trace")] pub async fn finish_stream(&self) -> Result<(), CommunicationError> { let _send_lock = self.send_guard.lock().await; let mut stream_opt = self.stream_guard.lock().await; @@ -403,7 +418,9 @@ impl Sender { &self.handle } + #[instrument(skip(self), level = "trace")] pub fn close(&self) { + info!(target = "mtp.transport", "fire-and-forget close requested"); let connection = self.connection.clone(); let handle = self.handle.clone(); let policy = self.policy.clone(); @@ -431,6 +448,7 @@ impl Sender { let _ = Self::send_close_frame(&connection, &policy).await; handle.close(Some(CommunicationError::StreamClosed)); + info!(target = "mtp.transport", "connection closed"); sleep(policy.force_close_delay).await; if connection.quic_connection().close_reason().is_none() { @@ -442,6 +460,51 @@ impl Sender { }); } + #[instrument(skip(self), level = "trace")] + /// Initiate a best-effort graceful close and wait for the configured force-close delay. + pub async fn close_and_wait(&self) { + info!(target = "mtp.transport", "graceful close initiated"); + let connection = self.connection.clone(); + let handle = self.handle.clone(); + let policy = self.policy.clone(); + let mut stream_opt = self.stream_guard.lock().await; + + if connection.quic_connection().close_reason().is_some() || handle.is_closed() { + handle.close(Some(CommunicationError::StreamClosed)); + return; + } + + if let Some(mut stream) = stream_opt.take() { + let close_bytes = policy.close_frame_len.to_be_bytes(); + let close_write = async { + stream.write_all(&close_bytes).await?; + stream.finish().await + }; + + match timeout(policy.write_timeout, close_write).await { + Ok(Ok(())) => {} + Ok(Err(wtransport::error::StreamWriteError::Stopped(code))) => log::warn!( + "[Sender] close_and_wait failed: peer sent STOP_SENDING (error code {code})" + ), + Ok(Err(e)) => log::warn!("[Sender] close_and_wait failed: {e}"), + Err(_) => log::warn!("[Sender] close_and_wait timed out"), + } + } else { + let _ = Self::send_close_frame(&connection, &policy).await; + } + + handle.close(Some(CommunicationError::StreamClosed)); + info!(target = "mtp.transport", "connection closed"); + + sleep(policy.force_close_delay).await; + if connection.quic_connection().close_reason().is_none() { + connection.quic_connection().close( + policy.application_close_code.into(), + APPLICATION_CLOSE_REASON.as_bytes(), + ); + } + } + pub fn is_open(&self) -> bool { self.handle.is_open() } @@ -455,11 +518,17 @@ impl Sender { } } +/// Single-consumer framed message receiver. +/// +/// `receive()` is intended to be driven by one task at a time. Internally the +/// underlying `mpsc::Receiver` is protected by a mutex so the type remains +/// `Sync`, but it is not a multi-consumer queue. pub struct Receiver { rx: Mutex>>, _accept_task: tokio::task::JoinHandle<()>, handle: Arc, ping_control: Arc>, + queue_notify: Arc, } #[derive(Clone, Default)] @@ -491,13 +560,40 @@ impl Receiver { let accept_policy = policy.clone(); let ping_control = Arc::new(RwLock::new(PingControl::default())); let accept_ping_control = ping_control.clone(); + let queue_notify = Arc::new(Notify::new()); + let accept_queue_notify = queue_notify.clone(); let stream_limit = Arc::new(Semaphore::new(policy.max_concurrent_stream_tasks.max(1))); let accept_stream_limit = stream_limit.clone(); + debug!( + target = "mtp.transport", + max_concurrent_stream_tasks = policy.max_concurrent_stream_tasks, + receiver_queue_capacity = policy.receiver_queue_capacity, + "receiver accept loop started" + ); + info!( + target = "mtp.transport", + max_concurrent_stream_tasks = policy.max_concurrent_stream_tasks, + receiver_queue_capacity = policy.receiver_queue_capacity, + "connection accepted" + ); let accept_task = tokio::spawn(async move { let mut close_rx = conn_handle.subscribe_close(); loop { + if tx.capacity() == 0 { + trace!(target = "mtp.transport", "accept loop paused: receiver queue full"); + tokio::select! { + _ = close_rx.changed() => { + if close_rx.borrow().is_some() { + break; + } + } + _ = accept_queue_notify.notified() => {} + } + continue; + } + tokio::select! { _ = close_rx.changed() => { if close_rx.borrow().is_some() { @@ -523,9 +619,20 @@ impl Receiver { tokio::spawn(async move { let _permit = permit; let mut s = stream; + let mut frame_count = 0usize; loop { + if let Some(max_frames) = stream_policy.max_frames_per_stream + && frame_count >= max_frames + { + let close_error = CommunicationError::StreamError; + let _ = tx_stream.send(Err(close_error.clone())).await; + stream_handle.close(Some(close_error)); + break; + } + match Self::read_one_frame(&mut s, &stream_policy).await { Ok(ReceivedFrame::Message(msg)) => { + frame_count += 1; let ping_type = mtp_codec::CommunicationType::Ping .to_id(&mtp_codec::TypeMap::latest()); let pong_type = mtp_codec::CommunicationType::Pong @@ -630,6 +737,7 @@ impl Receiver { _accept_task: accept_task, handle, ping_control, + queue_notify, } } @@ -651,6 +759,7 @@ impl Receiver { } } + #[instrument(skip(stream, policy), level = "trace")] async fn read_one_frame( stream: &mut wtransport::RecvStream, policy: &Policy, @@ -726,6 +835,7 @@ impl Receiver { Ok(ReceivedFrame::Message(message)) } + #[instrument(skip(self), level = "trace")] pub async fn receive(&self) -> Result { if self.handle.is_closed() { return Err(self @@ -736,7 +846,10 @@ impl Receiver { let mut rx = self.rx.lock().await; match rx.recv().await { - Some(result) => result, + Some(result) => { + self.queue_notify.notify_one(); + result + } _ => Err(self .handle .close_reason() @@ -796,6 +909,7 @@ mod tests { assert_eq!(p.persistent_stream_retry_backoff, Duration::from_millis(20)); assert_eq!(p.receiver_queue_capacity, 1000); assert_eq!(p.max_concurrent_stream_tasks, 128); + assert_eq!(p.max_frames_per_stream, None); } #[test] diff --git a/transport/tests/integration.rs b/transport/tests/integration.rs index 1099b9e..276c975 100644 --- a/transport/tests/integration.rs +++ b/transport/tests/integration.rs @@ -199,3 +199,149 @@ async fn test_drop_receiver_keeps_sender_alive() { client_tx.close(); host_tx.close(); } + +#[tokio::test] +async fn test_persistent_stream_reopens_after_local_finish() { + let (_h, client_tx, client_rx, host_tx, host_rx) = connected_pair().await; + + let tm = TypeMap::latest(); + let msg1 = numbered_message(CommunicationType::Ping, 11, &tm); + client_tx.send(&msg1).await.unwrap(); + let received1 = host_rx.receive().await.unwrap(); + assert_numbered_message(&received1, CommunicationType::Ping, 11, &tm); + + client_tx.finish_stream().await.unwrap(); + + let msg2 = numbered_message(CommunicationType::Pong, 22, &tm); + client_tx.send(&msg2).await.unwrap(); + let received2 = host_rx.receive().await.unwrap(); + assert_numbered_message(&received2, CommunicationType::Pong, 22, &tm); + + client_tx.close(); + host_tx.close(); + drop(client_rx); +} + +#[tokio::test] +async fn test_receiver_backpressure_with_small_queue() { + let (cert_pem, key_pem) = generate_self_signed_cert(); + let mut h = start_test_host(cert_pem.clone(), key_pem).await; + let url = format!("https://127.0.0.1:{}", h.local_addr().port()); + let policy = Policy::default().with_receiver_queue_capacity(1); + let (client_tx, client_rx) = connect(&url, Some(cert_pem), policy.clone()) + .await + .unwrap(); + let (_host_tx, host_rx) = h.next().await.unwrap(); + + let tm = TypeMap::latest(); + for i in 0..8u128 { + client_tx + .send(&numbered_message(CommunicationType::Ping, i, &tm)) + .await + .unwrap(); + } + + for i in 0..8u128 { + let received = tokio::time::timeout( + std::time::Duration::from_secs(5), + host_rx.receive(), + ) + .await + .unwrap() + .unwrap(); + assert_numbered_message(&received, CommunicationType::Ping, i, &tm); + } + + client_tx.close(); + drop(client_rx); + h.shutdown(); +} + +#[tokio::test] +async fn test_max_frames_per_stream_enforced() { + let (cert_pem, key_pem) = generate_self_signed_cert(); + let policy = Policy::default().with_max_frames_per_stream(Some(1)); + let mut h = host( + IpAddr::V4(Ipv4Addr::LOCALHOST), + 0, + cert_pem.clone(), + key_pem, + policy, + ) + .await + .unwrap(); + let url = format!("https://127.0.0.1:{}", h.local_addr().port()); + let (client_tx, _client_rx) = connect(&url, Some(cert_pem), Policy::default()) + .await + .unwrap(); + let (_host_tx, host_rx) = h.next().await.unwrap(); + + let tm = TypeMap::latest(); + client_tx + .send(&numbered_message(CommunicationType::Ping, 1, &tm)) + .await + .unwrap(); + let first = host_rx.receive().await.unwrap(); + assert_numbered_message(&first, CommunicationType::Ping, 1, &tm); + + client_tx + .send(&numbered_message(CommunicationType::Ping, 2, &tm)) + .await + .unwrap(); + let second = host_rx.receive().await; + assert!(second.is_err(), "stream should be closed after frame limit"); + + client_tx.close(); + h.shutdown(); +} + +#[tokio::test] +async fn test_semaphore_saturation_with_concurrent_streams() { + let (cert_pem, key_pem) = generate_self_signed_cert(); + let policy = Policy::default() + .with_send_mode(mtp_transport::SendMode::SingleStreamPerMessage) + .with_receiver_queue_capacity(1) + .with_max_concurrent_stream_tasks(1); + let mut h = host( + IpAddr::V4(Ipv4Addr::LOCALHOST), + 0, + cert_pem.clone(), + key_pem, + policy, + ) + .await + .unwrap(); + let url = format!("https://127.0.0.1:{}", h.local_addr().port()); + let (client_tx, _client_rx) = connect(&url, Some(cert_pem), Policy::default()) + .await + .unwrap(); + let (_host_tx, host_rx) = h.next().await.unwrap(); + + let tm = TypeMap::latest(); + let mut joins = Vec::new(); + for i in 0..6u128 { + let tx = client_tx.clone(); + let msg = numbered_message(CommunicationType::Ping, i, &tm); + joins.push(tokio::spawn(async move { tx.send(&msg).await })); + } + + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + + for join in joins { + join.await.unwrap().unwrap(); + } + + for i in 0..6u128 { + let received = tokio::time::timeout( + std::time::Duration::from_secs(5), + host_rx.receive(), + ) + .await + .unwrap() + .unwrap(); + assert_numbered_message(&received, CommunicationType::Ping, i, &tm); + } + + client_tx.close(); + h.shutdown(); +}