diff --git a/Cargo.lock b/Cargo.lock index 3aa2951..d16b578 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -39,6 +39,18 @@ dependencies = [ "cpufeatures 0.2.17", ] +[[package]] +name = "ahash" +version = "0.8.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5a15f179cd60c4584b8a8c596927aadc462e27f2ca70c04e0071964a73ba7a75" +dependencies = [ + "cfg-if", + "once_cell", + "version_check", + "zerocopy", +] + [[package]] name = "aho-corasick" version = "1.1.4" @@ -1646,6 +1658,18 @@ dependencies = [ "once_cell", ] +[[package]] +name = "fastbloom" +version = "0.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ef975e30683b2d965054bb0a836f8973857c4ebf6acf274fe46617cd285060d8" +dependencies = [ + "foldhash 0.2.0", + "libm", + "portable-atomic", + "siphasher", +] + [[package]] name = "fastrand" version = "2.4.1" @@ -1724,6 +1748,12 @@ version = "0.1.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d9c4f5dac5e15c24eb999c26181a6ca40b39fe946cbe4c263c7209467bc83af2" +[[package]] +name = "foldhash" +version = "0.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "77ce24cb58228fbb8aa041425bb1050850ac19177686ea6e0f41a70416f56fdb" + [[package]] name = "form_urlencoded" version = "1.2.2" @@ -2014,6 +2044,15 @@ dependencies = [ "tracing", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" +dependencies = [ + "ahash", +] + [[package]] name = "hashbrown" version = "0.15.5" @@ -2022,7 +2061,7 @@ checksum = "9229cfe53dfd69f0609a49f65461bd93001ea1ef889cd5529dd176593f5338a1" dependencies = [ "allocator-api2", "equivalent", - "foldhash", + "foldhash 0.1.5", ] [[package]] @@ -2040,6 +2079,15 @@ dependencies = [ "fxhash", ] +[[package]] +name = "hashlink" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ba4ff7128dee98c7dc9794b6a411377e1404dba1c97deb8d1a55297bd25d8af" +dependencies = [ + "hashbrown 0.14.5", +] + [[package]] name = "hashlink" version = "0.10.0" @@ -2754,10 +2802,17 @@ version = "0.2.186" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66" +[[package]] +name = "libm" +version = "0.2.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981" + [[package]] name = "libp2p" version = "0.56.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce71348bf5838e46449ae240631117b487073d5f347c06d434caddcb91dceb5a" dependencies = [ "bytes", "either", @@ -2791,7 +2846,8 @@ dependencies = [ [[package]] name = "libp2p-allow-block-list" version = "0.6.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d16ccf824ee859ca83df301e1c0205270206223fd4b1f2e512a693e1912a8f4a" dependencies = [ "libp2p-core", "libp2p-identity", @@ -2801,7 +2857,8 @@ dependencies = [ [[package]] name = "libp2p-autonat" version = "0.15.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fab5e25c49a7d48dac83d95d8f3bac0a290d8a5df717012f6e34ce9886396c0b" dependencies = [ "async-trait", "asynchronous-codec", @@ -2825,7 +2882,8 @@ dependencies = [ [[package]] name = "libp2p-connection-limits" version = "0.6.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a18b8b607cf3bfa2f8c57db9c7d8569a315d5cc0a282e6bfd5ebfc0a9840b2a0" dependencies = [ "libp2p-core", "libp2p-identity", @@ -2834,8 +2892,9 @@ dependencies = [ [[package]] name = "libp2p-core" -version = "0.43.1" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.43.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "249128cd37a2199aff30a7675dffa51caf073b51aa612d2f544b19932b9aebca" dependencies = [ "either", "fnv", @@ -2852,14 +2911,15 @@ dependencies = [ "rw-stream-sink", "thiserror 2.0.18", "tracing", - "unsigned-varint", + "unsigned-varint 0.8.0", "web-time", ] [[package]] name = "libp2p-dns" version = "0.44.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b770c1c8476736ca98c578cba4b505104ff8e842c2876b528925f9766379f9a" dependencies = [ "async-trait", "futures", @@ -2873,8 +2933,9 @@ dependencies = [ [[package]] name = "libp2p-gossipsub" -version = "0.50.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.49.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3573f3d8e30bd62cda336df5c7c1041a1caa40a648376b8e1e274d585c0ed25c" dependencies = [ "async-channel", "asynchronous-codec", @@ -2886,7 +2947,7 @@ dependencies = [ "futures", "futures-timer", "getrandom 0.2.17", - "hashlink", + "hashlink 0.9.1", "hex_fmt", "libp2p-core", "libp2p-identity", @@ -2904,7 +2965,8 @@ dependencies = [ [[package]] name = "libp2p-identify" version = "0.47.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8ab792a8b68fdef443a62155b01970c81c3aadab5e659621b063ef252a8e65e8" dependencies = [ "asynchronous-codec", "either", @@ -2923,14 +2985,15 @@ dependencies = [ [[package]] name = "libp2p-identity" -version = "0.2.12" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.2.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9525f3831544f7ae497bde79adf114ef127b0fbbb97edbbf692a80408636421c" dependencies = [ "bs58", "ed25519-dalek", "hkdf", "multihash", - "quick-protobuf", + "prost 0.14.4", "rand 0.8.6", "serde", "sha2", @@ -2941,8 +3004,9 @@ dependencies = [ [[package]] name = "libp2p-kad" -version = "0.49.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.48.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "13d3fd632a5872ec804d37e7413ceea20588f69d027a0fa3c46f82574f4dee60" dependencies = [ "asynchronous-codec", "bytes", @@ -2969,7 +3033,8 @@ dependencies = [ [[package]] name = "libp2p-mdns" version = "0.48.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c66872d0f1ffcded2788683f76931be1c52e27f343edb93bc6d0bcd8887be443" dependencies = [ "futures", "hickory-proto", @@ -2979,7 +3044,7 @@ dependencies = [ "libp2p-swarm", "rand 0.8.6", "smallvec", - "socket2 0.6.4", + "socket2 0.5.10", "tokio", "tracing", ] @@ -2987,7 +3052,8 @@ dependencies = [ [[package]] name = "libp2p-metrics" version = "0.17.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "805a555148522cb3414493a5153451910cb1a146c53ffbf4385708349baf62b7" dependencies = [ "futures", "libp2p-core", @@ -3006,7 +3072,8 @@ dependencies = [ [[package]] name = "libp2p-ping" version = "0.47.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "74bb7fcdfd9fead4144a3859da0b49576f171a8c8c7c0bfc7c541921d25e60d3" dependencies = [ "futures", "futures-timer", @@ -3021,7 +3088,8 @@ dependencies = [ [[package]] name = "libp2p-plaintext" version = "0.43.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7e659439578fc6d305da8303834beb9d62f155f40e7f5b9d81c9f2b2c69d1926" dependencies = [ "asynchronous-codec", "bytes", @@ -3035,8 +3103,9 @@ dependencies = [ [[package]] name = "libp2p-quic" -version = "0.13.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9dcc597d70bf7f6f30cbe07081802c836184e48416e89e9ce73a0ba2c56a319e" dependencies = [ "futures", "futures-timer", @@ -3045,10 +3114,11 @@ dependencies = [ "libp2p-identity", "libp2p-tls", "quinn", + "quinn-proto", "rand 0.8.6", "ring 0.17.14", "rustls 0.23.41", - "socket2 0.6.4", + "socket2 0.5.10", "thiserror 2.0.18", "tokio", "tracing", @@ -3056,8 +3126,9 @@ dependencies = [ [[package]] name = "libp2p-relay" -version = "0.21.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.21.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8b9b0392ed623243ad298326b9f806d51191829ac7585cc825c54c6c67b04d9" dependencies = [ "asynchronous-codec", "bytes", @@ -3080,7 +3151,8 @@ dependencies = [ [[package]] name = "libp2p-request-response" version = "0.29.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a9f1cca83488b90102abac7b67d5c36fc65bc02ed47620228af7ed002e6a1478" dependencies = [ "async-trait", "futures", @@ -3093,29 +3165,17 @@ dependencies = [ "tracing", ] -[[package]] -name = "libp2p-stream" -version = "0.4.0-alpha" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" -dependencies = [ - "futures", - "libp2p-core", - "libp2p-identity", - "libp2p-swarm", - "rand 0.8.6", - "tracing", -] - [[package]] name = "libp2p-swarm" -version = "0.47.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.47.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce88c6c4bf746c8482480345ea3edfd08301f49e026889d1cbccfa1808a9ed9e" dependencies = [ "either", "fnv", "futures", "futures-timer", - "hashlink", + "hashlink 0.10.0", "libp2p-core", "libp2p-identity", "multistream-select", @@ -3129,7 +3189,8 @@ dependencies = [ [[package]] name = "libp2p-swarm-derive" version = "0.35.1" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "dd297cf53f0cb3dee4d2620bb319ae47ef27c702684309f682bdb7e55a18ae9c" dependencies = [ "heck 0.5.0", "quote", @@ -3139,7 +3200,8 @@ dependencies = [ [[package]] name = "libp2p-swarm-test" version = "0.6.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7b149112570d507efe305838c7130835955a0b1147aa8051c1c3867a83175cf6" dependencies = [ "async-trait", "futures", @@ -3155,8 +3217,9 @@ dependencies = [ [[package]] name = "libp2p-tcp" -version = "0.44.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.44.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fb6585b9309699f58704ec9ab0bb102eca7a3777170fa91a8678d73ca9cafa93" dependencies = [ "futures", "futures-timer", @@ -3171,7 +3234,8 @@ dependencies = [ [[package]] name = "libp2p-tls" version = "0.6.2" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "96ff65a82e35375cbc31ebb99cacbbf28cb6c4fefe26bf13756ddcf708d40080" dependencies = [ "futures", "futures-rustls", @@ -3188,8 +3252,9 @@ dependencies = [ [[package]] name = "libp2p-upnp" -version = "0.5.1" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4757e65fe69399c1a243bbb90ec1ae5a2114b907467bf09f3575e899815bb8d3" dependencies = [ "futures", "futures-timer", @@ -3203,7 +3268,8 @@ dependencies = [ [[package]] name = "libp2p-yamux" version = "0.47.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f15df094914eb4af272acf9adaa9e287baa269943f32ea348ba29cfb9bfc60d8" dependencies = [ "either", "futures", @@ -3363,7 +3429,7 @@ dependencies = [ "percent-encoding", "serde", "static_assertions", - "unsigned-varint", + "unsigned-varint 0.8.0", "url", ] @@ -3387,7 +3453,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "577c63b00ad74d57e8c9aa870b5fccebf2fd64a308a5aee9f1bb88e4aea19447" dependencies = [ "serde", - "unsigned-varint", + "unsigned-varint 0.8.0", ] [[package]] @@ -3399,14 +3465,15 @@ checksum = "1d87ecb2933e8aeadb3e3a02b828fed80a7528047e68b4f424523a0981a3a084" [[package]] name = "multistream-select" version = "0.13.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea0df8e5eec2298a62b326ee4f0d7fe1a6b90a09dfcf9df37b38f947a8c42f19" dependencies = [ "bytes", "futures", + "log", "pin-project", "smallvec", - "tracing", - "unsigned-varint", + "unsigned-varint 0.7.2", ] [[package]] @@ -4077,6 +4144,16 @@ dependencies = [ "prost-derive 0.13.5", ] +[[package]] +name = "prost" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "528ac67416ff8646872a3c02cad9cc4ee5dc9f9540c9b10771855c95cb2e5ae1" +dependencies = [ + "bytes", + "prost-derive 0.14.4", +] + [[package]] name = "prost-build" version = "0.12.6" @@ -4124,6 +4201,19 @@ dependencies = [ "syn 2.0.118", ] +[[package]] +name = "prost-derive" +version = "0.14.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" +dependencies = [ + "anyhow", + "itertools 0.11.0", + "proc-macro2", + "quote", + "syn 2.0.118", +] + [[package]] name = "prost-types" version = "0.12.6" @@ -4145,13 +4235,14 @@ dependencies = [ [[package]] name = "quick-protobuf-codec" version = "0.3.1" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "15a0580ab32b169745d7a39db2ba969226ca16738931be152a3209b409de2474" dependencies = [ "asynchronous-codec", "bytes", "quick-protobuf", - "thiserror 2.0.18", - "unsigned-varint", + "thiserror 1.0.69", + "unsigned-varint 0.8.0", ] [[package]] @@ -4182,6 +4273,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560" dependencies = [ "bytes", + "fastbloom", "getrandom 0.4.3", "lru-slab", "rand 0.10.2", @@ -4667,7 +4759,8 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" [[package]] name = "rw-stream-sink" version = "0.4.0" -source = "git+https://github.com/kalabukdima/rust-libp2p.git?rev=fac3b0c9#fac3b0c97757d6d8d6042f5423d9467a94ea80b0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d8c9026ff5d2f23da5e45bbc283f156383001bfb09c4e44256d02c1a685fe9a1" dependencies = [ "futures", "pin-project", @@ -5112,7 +5205,9 @@ dependencies = [ "lazy_static", "libp2p", "libp2p-connection-limits", - "libp2p-stream", + "libp2p-core", + "libp2p-identity", + "libp2p-swarm", "libp2p-swarm-derive", "libp2p-swarm-test", "log", @@ -5129,6 +5224,7 @@ dependencies = [ "tokio", "tokio-stream", "tokio-util", + "tracing", ] [[package]] @@ -5782,6 +5878,12 @@ dependencies = [ "subtle", ] +[[package]] +name = "unsigned-varint" +version = "0.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6889a77d49f1f013504cec6bf97a2c730394adedaeb1deb5ea08949a50541105" + [[package]] name = "unsigned-varint" version = "0.8.0" diff --git a/Cargo.toml b/Cargo.toml index 2cb72d0..88f621f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -3,18 +3,13 @@ members = ["crates/*"] resolver = "2" [workspace.dependencies] -libp2p = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } +libp2p = { version = "0.56.0" } +libp2p-core = { version = "0.43.1" } libp2p-identity = { version = "0.2.12", features = ["peerid"] } -libp2p-connection-limits = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } -libp2p-swarm-derive = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } -libp2p-stream = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } -libp2p-swarm-test = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } - -# The fork builds its crates against its in-tree libp2p-identity (path = "identity"), so consumers -# must redirect the crates.io libp2p-identity to the same source, otherwise two incompatible copies -# of libp2p_identity are linked. See the fork's "Use libp2p-identity from the fork" commit. -[patch.crates-io] -libp2p-identity = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" } +libp2p-connection-limits = { version = "0.6.0" } +libp2p-swarm = { version = "0.47.0" } +libp2p-swarm-derive = { version = "0.35.1" } +libp2p-swarm-test = { version = "0.6.0" } [workspace.lints.rust] unsafe_code = "deny" diff --git a/crates/transport/Cargo.toml b/crates/transport/Cargo.toml index 19b8141..263e307 100644 --- a/crates/transport/Cargo.toml +++ b/crates/transport/Cargo.toml @@ -15,7 +15,14 @@ futures-core = "0.3" lazy_static = { version = "1" } libp2p = { workspace = true, features = ["dns", "tokio", "identify", "kad", "ping", "request-response", "serde", "autonat", "quic"] } libp2p-connection-limits = { workspace = true } -libp2p-stream = { workspace = true } +# libp2p-core, libp2p-identity, libp2p-swarm, rand and tracing are used by the vendored +# `libp2p_stream` module only — the rest of the crate goes through the `libp2p` facade. Bump +# these three together with `libp2p` itself: pointing them at a version the facade does not use +# links two copies of the swarm types and fails with "expected NetworkBehaviour, found +# NetworkBehaviour". +libp2p-core = { workspace = true } +libp2p-identity = { workspace = true } +libp2p-swarm = { workspace = true } libp2p-swarm-derive = { workspace = true } log = "0.4" lru = "0.12" @@ -29,16 +36,17 @@ thiserror = "1" tokio = { version = "1", features = ["fs", "macros", "sync"] } tokio-util = { version = "0.7", features = ["time"] } tokio-stream = "0.1.17" +tracing = { version = "0.1", optional = true } sqd-contract-client = { path = "../contract-client" } sqd-messages = { path = "../messages", features = ["signatures", "semver"] } [features] actors = ["behaviour"] -behaviour = [] +behaviour = ["dep:rand", "dep:tracing"] proto = [] pubsub = ["behaviour", "libp2p/gossipsub"] -noise = ["rand"] +noise = ["dep:rand"] request-server = [] stream-server = ["behaviour"] portal = ["actors"] diff --git a/crates/transport/src/behaviour/keep_alive.rs b/crates/transport/src/behaviour/keep_alive.rs index c23e5e5..da2ee34 100644 --- a/crates/transport/src/behaviour/keep_alive.rs +++ b/crates/transport/src/behaviour/keep_alive.rs @@ -9,7 +9,7 @@ use libp2p::{ }; use std::task::{Context, Poll}; -use crate::{protocol::KEEP_ALIVE_PROTOCOL, Multiaddr, PeerId}; +use crate::{libp2p_stream, protocol::KEEP_ALIVE_PROTOCOL, Multiaddr, PeerId}; /// A behaviour that allows keeping connections alive forever. /// diff --git a/crates/transport/src/behaviour/noise.rs b/crates/transport/src/behaviour/noise.rs index 7172f56..8de31c5 100644 --- a/crates/transport/src/behaviour/noise.rs +++ b/crates/transport/src/behaviour/noise.rs @@ -9,7 +9,7 @@ use libp2p::{ }; use std::task::{Context, Poll}; -use crate::{protocol::NOISE_PROTOCOL, Multiaddr, PeerId}; +use crate::{libp2p_stream, protocol::NOISE_PROTOCOL, Multiaddr, PeerId}; const BUFFER_SIZE: usize = 10 * 1024 * 1024; // 10 MB diff --git a/crates/transport/src/behaviour/stream_client.rs b/crates/transport/src/behaviour/stream_client.rs index efd0720..95dfc61 100644 --- a/crates/transport/src/behaviour/stream_client.rs +++ b/crates/transport/src/behaviour/stream_client.rs @@ -2,9 +2,12 @@ use std::{collections::HashMap, sync::Arc, time::Duration}; use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use libp2p::{PeerId, StreamProtocol}; -use libp2p_stream::OpenStreamError; -use crate::{util::StreamWithPayload, BehaviourWrapper}; +use crate::{ + libp2p_stream::{self, OpenStreamError}, + util::StreamWithPayload, + BehaviourWrapper, +}; #[derive(Debug, Clone, Copy)] pub struct ClientConfig { @@ -136,18 +139,13 @@ impl Clone for StreamClientHandle { } } +// Derived rather than written out: now that `libp2p_stream` lives in this crate, clippy sees +// that `Behaviour: Default` and flags the manual impl as derivable. +#[derive(Default)] pub struct ClientBehaviour { inner: libp2p_stream::Behaviour, } -impl Default for ClientBehaviour { - fn default() -> Self { - Self { - inner: libp2p_stream::Behaviour::new(), - } - } -} - impl ClientBehaviour { pub fn new_handle(&self, protocol: &'static str, config: ClientConfig) -> StreamClientHandle { let control = self.inner.new_control(); @@ -174,7 +172,6 @@ impl From for RequestError { match e { OpenStreamError::UnsupportedProtocol(_) => Self::UnsupportedProtocol, OpenStreamError::Io(e) => Self::Io(e), - _ => unreachable!(), } } } diff --git a/crates/transport/src/behaviour/stream_server.rs b/crates/transport/src/behaviour/stream_server.rs index 0a91609..72539e4 100644 --- a/crates/transport/src/behaviour/stream_server.rs +++ b/crates/transport/src/behaviour/stream_server.rs @@ -5,7 +5,10 @@ use futures::{AsyncReadExt, AsyncWriteExt, StreamExt}; use libp2p::{swarm::ToSwarm, PeerId, Stream, StreamProtocol}; use tokio::sync::mpsc; -use crate::behaviour::wrapped::{BehaviourWrapper, TToSwarm}; +use crate::{ + behaviour::wrapped::{BehaviourWrapper, TToSwarm}, + libp2p_stream, +}; #[derive(Debug, Clone, Copy)] pub struct ServerConfig { diff --git a/crates/transport/src/lib.rs b/crates/transport/src/lib.rs index ec18eab..eeb8685 100644 --- a/crates/transport/src/lib.rs +++ b/crates/transport/src/lib.rs @@ -40,6 +40,8 @@ mod builder; mod cli; #[cfg(feature = "proto")] mod codec; +#[cfg(feature = "behaviour")] +pub mod libp2p_stream; #[cfg(feature = "metrics")] pub mod metrics; pub mod protocol; diff --git a/crates/transport/src/libp2p_stream/LICENSE b/crates/transport/src/libp2p_stream/LICENSE new file mode 100644 index 0000000..da84696 --- /dev/null +++ b/crates/transport/src/libp2p_stream/LICENSE @@ -0,0 +1,18 @@ +Copyright 2017-2020 Parity Technologies (UK) Ltd. + +Permission is hereby granted, free of charge, to any person obtaining a copy of +this software and associated documentation files (the "Software"), to deal in +the Software without restriction, including without limitation the rights to +use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of +the Software, and to permit persons to whom the Software is furnished to do so, +subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS +FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS +OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, +WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN +CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. diff --git a/crates/transport/src/libp2p_stream/behaviour.rs b/crates/transport/src/libp2p_stream/behaviour.rs new file mode 100644 index 0000000..0dbedee --- /dev/null +++ b/crates/transport/src/libp2p_stream/behaviour.rs @@ -0,0 +1,146 @@ +use core::fmt; +use std::{ + sync::{Arc, Mutex}, + task::{Context, Poll}, +}; + +use futures::{channel::mpsc, StreamExt}; +use libp2p_core::{transport::PortUse, Endpoint, Multiaddr}; +use libp2p_identity::PeerId; +use libp2p_swarm::{ + self as swarm, dial_opts::DialOpts, ConnectionDenied, ConnectionId, FromSwarm, + NetworkBehaviour, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, +}; +use swarm::{ + behaviour::ConnectionEstablished, dial_opts::PeerCondition, ConnectionClosed, DialError, + DialFailure, +}; + +use super::{handler::Handler, shared::Shared, Control}; + +/// A generic behaviour for stream-oriented protocols. +pub struct Behaviour { + shared: Arc>, + dial_receiver: mpsc::UnboundedReceiver, +} + +impl Default for Behaviour { + fn default() -> Self { + Self::new() + } +} + +impl Behaviour { + pub fn new() -> Self { + let (dial_sender, dial_receiver) = mpsc::unbounded(); + + Self { + shared: Arc::new(Mutex::new(Shared::new(dial_sender))), + dial_receiver, + } + } + + /// Obtain a new [`Control`]. + pub fn new_control(&self) -> Control { + Control::new(self.shared.clone()) + } +} + +/// The protocol is already registered. +#[derive(Debug)] +pub struct AlreadyRegistered; + +impl fmt::Display for AlreadyRegistered { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "The protocol is already registered") + } +} + +impl std::error::Error for AlreadyRegistered {} + +impl NetworkBehaviour for Behaviour { + type ConnectionHandler = Handler; + type ToSwarm = (); + + fn handle_established_inbound_connection( + &mut self, + connection_id: ConnectionId, + peer: PeerId, + _: &Multiaddr, + _: &Multiaddr, + ) -> Result, ConnectionDenied> { + Ok(Handler::new( + peer, + self.shared.clone(), + Shared::lock(&self.shared).receiver(peer, connection_id), + )) + } + + fn handle_established_outbound_connection( + &mut self, + connection_id: ConnectionId, + peer: PeerId, + _: &Multiaddr, + _: Endpoint, + _: PortUse, + ) -> Result, ConnectionDenied> { + Ok(Handler::new( + peer, + self.shared.clone(), + Shared::lock(&self.shared).receiver(peer, connection_id), + )) + } + + fn on_swarm_event(&mut self, event: FromSwarm) { + match event { + FromSwarm::ConnectionEstablished(ConnectionEstablished { + peer_id, + connection_id, + .. + }) => Shared::lock(&self.shared).on_connection_established(connection_id, peer_id), + FromSwarm::ConnectionClosed(ConnectionClosed { connection_id, .. }) => { + Shared::lock(&self.shared).on_connection_closed(connection_id) + } + FromSwarm::DialFailure(DialFailure { + peer_id: Some(peer_id), + error: + error @ (DialError::LocalPeerId { .. } + | DialError::Transport(_) + | DialError::Denied { .. } + | DialError::NoAddresses + | DialError::Aborted + | DialError::WrongPeerId { .. }), + .. + }) => { + let reason = error.to_string(); // We can only forward the string repr but it is better than nothing. + + Shared::lock(&self.shared).on_dial_failure(peer_id, reason) + } + _ => {} + } + } + + fn on_connection_handler_event( + &mut self, + _peer_id: PeerId, + _connection_id: ConnectionId, + event: THandlerOutEvent, + ) { + libp2p_core::util::unreachable(event); + } + + fn poll( + &mut self, + cx: &mut Context<'_>, + ) -> Poll>> { + if let Poll::Ready(Some(peer)) = self.dial_receiver.poll_next_unpin(cx) { + return Poll::Ready(ToSwarm::Dial { + opts: DialOpts::peer_id(peer) + .condition(PeerCondition::DisconnectedAndNotDialing) + .build(), + }); + } + + Poll::Pending + } +} diff --git a/crates/transport/src/libp2p_stream/control.rs b/crates/transport/src/libp2p_stream/control.rs new file mode 100644 index 0000000..d640062 --- /dev/null +++ b/crates/transport/src/libp2p_stream/control.rs @@ -0,0 +1,154 @@ +use core::fmt; +use std::{ + io, + pin::Pin, + sync::{Arc, Mutex}, + task::{Context, Poll}, +}; + +use futures::{ + channel::{mpsc, oneshot}, + SinkExt as _, StreamExt as _, +}; +use libp2p_identity::PeerId; +use libp2p_swarm::{Stream, StreamProtocol}; + +use super::{handler::NewStream, shared::Shared, AlreadyRegistered}; + +/// A (remote) control for opening new streams and registration of inbound protocols. +/// +/// A [`Control`] can be cloned and thus allows for concurrent access. +#[derive(Clone)] +pub struct Control { + shared: Arc>, +} + +impl Control { + pub(crate) fn new(shared: Arc>) -> Self { + Self { shared } + } + + /// Attempt to open a new stream for the given protocol and peer. + /// + /// In case we are currently not connected to the peer, + /// we will attempt to make a new connection. + /// + /// ## Backpressure + /// + /// [`Control`]s support backpressure similarly to bounded channels: + /// Each [`Control`] has a guaranteed slot for internal messages. + /// A single control will always open one stream at a + /// time which is enforced by requiring `&mut self`. + /// + /// This backpressure mechanism breaks if you clone [`Control`]s excessively. + pub async fn open_stream( + &mut self, + peer: PeerId, + protocol: StreamProtocol, + ) -> Result { + tracing::debug!(%peer, "Requesting new stream"); + + let mut new_stream_sender = Shared::lock(&self.shared).sender(peer)?; + + let (sender, receiver) = oneshot::channel(); + + new_stream_sender + .send(NewStream { protocol, sender }) + .await + .map_err(|e| io::Error::new(io::ErrorKind::ConnectionReset, e))?; + + let stream = receiver + .await + .map_err(|e| io::Error::new(io::ErrorKind::ConnectionReset, e))??; + + Ok(stream) + } + + /// Accept inbound streams for the provided protocol. + /// + /// To stop accepting streams, simply drop the returned [`IncomingStreams`] handle. + /// + /// The returned [`IncomingStreams`] does not buffer any streams: an inbound stream is + /// dropped unless [`IncomingStreams`] is being polled at the moment it arrives. Use + /// [`Control::accept_with_capacity`] to buffer a burst of concurrent inbound streams. + pub fn accept( + &mut self, + protocol: StreamProtocol, + ) -> Result { + Shared::lock(&self.shared).accept(protocol, 0) + } + + /// Accept inbound streams for the provided protocol, buffering up to `capacity` streams. + /// + /// Unlike [`Control::accept`], this buffers inbound streams that arrive while the returned + /// [`IncomingStreams`] is not being polled, up to `capacity`. This prevents streams from a + /// burst of concurrent inbound connections being dropped. Streams arriving while the buffer + /// is full are still dropped. + /// + /// To stop accepting streams, simply drop the returned [`IncomingStreams`] handle. + pub fn accept_with_capacity( + &mut self, + protocol: StreamProtocol, + capacity: usize, + ) -> Result { + Shared::lock(&self.shared).accept(protocol, capacity) + } +} + +/// Errors while opening a new stream. +#[derive(Debug)] +#[non_exhaustive] +pub enum OpenStreamError { + /// The remote does not support the requested protocol. + UnsupportedProtocol(StreamProtocol), + /// IO Error that occurred during the protocol handshake. + Io(std::io::Error), +} + +impl From for OpenStreamError { + fn from(v: std::io::Error) -> Self { + Self::Io(v) + } +} + +impl fmt::Display for OpenStreamError { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + OpenStreamError::UnsupportedProtocol(p) => { + write!(f, "failed to open stream: remote peer does not support {p}") + } + OpenStreamError::Io(e) => { + write!(f, "failed to open stream: io error: {e}") + } + } + } +} + +impl std::error::Error for OpenStreamError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + Self::Io(error) => Some(error), + _ => None, + } + } +} + +/// A handle to inbound streams for a particular protocol. +#[must_use = "Streams do nothing unless polled."] +pub struct IncomingStreams { + receiver: mpsc::Receiver<(PeerId, Stream)>, +} + +impl IncomingStreams { + pub(crate) fn new(receiver: mpsc::Receiver<(PeerId, Stream)>) -> Self { + Self { receiver } + } +} + +impl futures::Stream for IncomingStreams { + type Item = (PeerId, Stream); + + fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.receiver.poll_next_unpin(cx) + } +} diff --git a/crates/transport/src/libp2p_stream/handler.rs b/crates/transport/src/libp2p_stream/handler.rs new file mode 100644 index 0000000..f264fbc --- /dev/null +++ b/crates/transport/src/libp2p_stream/handler.rs @@ -0,0 +1,145 @@ +use std::{ + convert::Infallible, + io, + sync::{Arc, Mutex}, + task::{Context, Poll}, +}; + +use futures::{ + channel::{mpsc, oneshot}, + StreamExt as _, +}; +use libp2p_identity::PeerId; +use libp2p_swarm::{ + self as swarm, + handler::{ConnectionEvent, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound}, + ConnectionHandler, Stream, StreamProtocol, +}; + +use super::{shared::Shared, upgrade::Upgrade, OpenStreamError}; + +pub struct Handler { + remote: PeerId, + shared: Arc>, + + receiver: mpsc::Receiver, + pending_upgrade: Option<(StreamProtocol, oneshot::Sender>)>, +} + +impl Handler { + pub(crate) fn new( + remote: PeerId, + shared: Arc>, + receiver: mpsc::Receiver, + ) -> Self { + Self { + shared, + receiver, + pending_upgrade: None, + remote, + } + } +} + +impl ConnectionHandler for Handler { + type FromBehaviour = Infallible; + type ToBehaviour = Infallible; + type InboundProtocol = Upgrade; + type OutboundProtocol = Upgrade; + type InboundOpenInfo = (); + type OutboundOpenInfo = (); + + fn listen_protocol(&self) -> swarm::SubstreamProtocol { + swarm::SubstreamProtocol::new( + Upgrade { + supported_protocols: Shared::lock(&self.shared).supported_inbound_protocols(), + }, + (), + ) + } + + fn poll( + &mut self, + cx: &mut Context<'_>, + ) -> Poll> { + if self.pending_upgrade.is_some() { + return Poll::Pending; + } + + match self.receiver.poll_next_unpin(cx) { + Poll::Ready(Some(new_stream)) => { + self.pending_upgrade = Some((new_stream.protocol.clone(), new_stream.sender)); + return Poll::Ready(swarm::ConnectionHandlerEvent::OutboundSubstreamRequest { + protocol: swarm::SubstreamProtocol::new( + Upgrade { + supported_protocols: vec![new_stream.protocol], + }, + (), + ), + }); + } + Poll::Ready(None) => {} // Sender is gone, no more work to do. + Poll::Pending => {} + } + + Poll::Pending + } + + fn on_behaviour_event(&mut self, event: Self::FromBehaviour) { + libp2p_core::util::unreachable(event) + } + + fn on_connection_event( + &mut self, + event: ConnectionEvent, + ) { + match event { + ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { + protocol: (stream, protocol), + info: (), + }) => { + Shared::lock(&self.shared).on_inbound_stream(self.remote, stream, protocol); + } + ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { + protocol: (stream, actual_protocol), + info: (), + }) => { + let Some((expected_protocol, sender)) = self.pending_upgrade.take() else { + debug_assert!(false, "Negotiated an outbound stream without a back channel"); + return; + }; + debug_assert_eq!(expected_protocol, actual_protocol); + + let _ = sender.send(Ok(stream)); + } + ConnectionEvent::DialUpgradeError(DialUpgradeError { error, info: () }) => { + let Some((p, sender)) = self.pending_upgrade.take() else { + debug_assert!(false, "Received a `DialUpgradeError` without a back channel"); + return; + }; + + let error = match error { + swarm::StreamUpgradeError::Timeout => { + OpenStreamError::Io(io::Error::from(io::ErrorKind::TimedOut)) + } + swarm::StreamUpgradeError::Apply(v) => libp2p_core::util::unreachable(v), + swarm::StreamUpgradeError::NegotiationFailed => { + OpenStreamError::UnsupportedProtocol(p) + } + swarm::StreamUpgradeError::Io(io) => OpenStreamError::Io(io), + }; + + let _ = sender.send(Err(error)); + } + _ => {} + } + } +} + +/// Message from a [`Control`](super::Control) to +/// a [`ConnectionHandler`] to negotiate a new outbound stream. +#[derive(Debug)] +pub(crate) struct NewStream { + pub(crate) protocol: StreamProtocol, + pub(crate) sender: oneshot::Sender>, +} diff --git a/crates/transport/src/libp2p_stream/mod.rs b/crates/transport/src/libp2p_stream/mod.rs new file mode 100644 index 0000000..6559425 --- /dev/null +++ b/crates/transport/src/libp2p_stream/mod.rs @@ -0,0 +1,44 @@ +//! A vendored copy of the `libp2p-stream` crate, carrying local reliability patches. +//! +//! Upstream source: , `protocols/stream` at commit +//! `aefbfbdcef203f9d48af864335fe737b4445a042` (`libp2p-stream` v0.4.0-alpha). MIT licensed — +//! see the `LICENSE` file next to this module. +//! +//! # Local patches +//! +//! The first two patches came from the original fork and fix streams being silently dropped under +//! concurrency: +//! +//! * The dial request channel in [`Behaviour::new`] is unbounded. Upstream uses +//! `mpsc::channel(0)` and discards the `try_send` error in `Shared::sender`, so only one dial +//! request fits between two swarm polls: a concurrent `open_stream` for a second, not yet +//! connected peer is dropped and its caller waits for a dial that never happens. Requests are +//! de-duplicated per peer so the unbounded channel cannot accumulate duplicate dials. +//! * [`Control::accept_with_capacity`] is added. Upstream's [`Control::accept`] registers an +//! `mpsc::channel(0)`, so an inbound stream is dropped unless [`IncomingStreams`] happens to +//! be polled at that moment, losing bursts of concurrent requests. `accept` keeps the +//! unbuffered behaviour. +//! +//! The vendored version also propagates every terminal dial failure, reports an error when a +//! [`Control`] outlives its [`Behaviour`], and removes per-connection senders when connections +//! close. These prevent hung `open_stream` calls and an unbounded connection-churn leak. +//! +//! Apart from the changes listed above, the module paths (`crate::` -> `super::`), +//! `rand::thread_rng()` -> `rand::rng()` (renamed in rand 0.9) and this repository's rustfmt +//! settings, the code is upstream's. Keep it that way: refreshing this copy should stay a matter +//! of taking the newer upstream files, re-applying the patches and running `cargo fmt`. + +// Vendored code is exempt from this workspace's lints: it should stay as close to upstream as +// possible rather than follow local style. Applies to the child modules as well. +#![allow(deprecated, clippy::all, clippy::pedantic, clippy::nursery)] + +mod behaviour; +mod control; +mod handler; +mod shared; +#[cfg(test)] +mod tests; +mod upgrade; + +pub use behaviour::{AlreadyRegistered, Behaviour}; +pub use control::{Control, IncomingStreams, OpenStreamError}; diff --git a/crates/transport/src/libp2p_stream/shared.rs b/crates/transport/src/libp2p_stream/shared.rs new file mode 100644 index 0000000..1200313 --- /dev/null +++ b/crates/transport/src/libp2p_stream/shared.rs @@ -0,0 +1,202 @@ +use std::{ + collections::{hash_map::Entry, HashMap}, + io, + sync::{Arc, Mutex, MutexGuard}, +}; + +use futures::channel::mpsc; +use libp2p_identity::PeerId; +use libp2p_swarm::{ConnectionId, Stream, StreamProtocol}; +use rand::seq::IteratorRandom as _; + +use super::{handler::NewStream, AlreadyRegistered, IncomingStreams}; + +pub(crate) struct Shared { + /// Tracks the supported inbound protocols created via + /// [`Control::accept`](super::Control::accept). + /// + /// For each [`StreamProtocol`], we hold the [`mpsc::Sender`] corresponding to the + /// [`mpsc::Receiver`] in [`IncomingStreams`]. + supported_inbound_protocols: HashMap>, + + connections: HashMap, + senders: HashMap>, + + /// Tracks channel pairs for a peer whilst we are dialing them. + pending_channels: HashMap, mpsc::Receiver)>, + + /// Sender for peers we want to dial. + /// + /// We manage this through a channel to avoid locks as part of + /// [`NetworkBehaviour::poll`](libp2p_swarm::NetworkBehaviour::poll). + dial_sender: mpsc::UnboundedSender, +} + +impl Shared { + pub(crate) fn lock(shared: &Arc>) -> MutexGuard<'_, Shared> { + shared.lock().unwrap_or_else(|e| e.into_inner()) + } +} + +impl Shared { + pub(crate) fn new(dial_sender: mpsc::UnboundedSender) -> Self { + Self { + dial_sender, + connections: Default::default(), + senders: Default::default(), + pending_channels: Default::default(), + supported_inbound_protocols: Default::default(), + } + } + + pub(crate) fn accept( + &mut self, + protocol: StreamProtocol, + capacity: usize, + ) -> Result { + self.supported_inbound_protocols.retain(|_, sender| !sender.is_closed()); + + if self.supported_inbound_protocols.contains_key(&protocol) { + return Err(AlreadyRegistered); + } + + let (sender, receiver) = mpsc::channel(capacity); + self.supported_inbound_protocols.insert(protocol.clone(), sender); + + Ok(IncomingStreams::new(receiver)) + } + + /// Lists the protocols for which we have an active [`IncomingStreams`] instance. + pub(crate) fn supported_inbound_protocols(&mut self) -> Vec { + self.supported_inbound_protocols.retain(|_, sender| !sender.is_closed()); + + self.supported_inbound_protocols.keys().cloned().collect() + } + + pub(crate) fn on_inbound_stream( + &mut self, + remote: PeerId, + stream: Stream, + protocol: StreamProtocol, + ) { + match self.supported_inbound_protocols.entry(protocol.clone()) { + Entry::Occupied(mut entry) => match entry.get_mut().try_send((remote, stream)) { + Ok(()) => {} + Err(e) if e.is_full() => { + tracing::debug!(%protocol, "Channel is full, dropping inbound stream"); + } + Err(e) if e.is_disconnected() => { + tracing::debug!(%protocol, "Channel is gone, dropping inbound stream"); + entry.remove(); + } + _ => unreachable!(), + }, + Entry::Vacant(_) => { + tracing::debug!(%protocol, "channel is gone, dropping inbound stream"); + } + } + } + + pub(crate) fn on_connection_established(&mut self, conn: ConnectionId, peer: PeerId) { + self.connections.insert(conn, peer); + } + + pub(crate) fn on_connection_closed(&mut self, conn: ConnectionId) { + self.connections.remove(&conn); + self.senders.remove(&conn); + } + + pub(crate) fn on_dial_failure(&mut self, peer: PeerId, reason: String) { + let Some((_, mut receiver)) = self.pending_channels.remove(&peer) else { + return; + }; + + while let Ok(Some(new_stream)) = receiver.try_next() { + let _ = new_stream.sender.send(Err(super::OpenStreamError::Io(io::Error::new( + io::ErrorKind::NotConnected, + reason.clone(), + )))); + } + } + + pub(crate) fn sender(&mut self, peer: PeerId) -> io::Result> { + let maybe_sender = self + .connections + .iter() + .filter_map(|(c, p)| (p == &peer).then_some(c)) + .choose(&mut rand::rng()) + .and_then(|c| self.senders.get(c)); + + match maybe_sender { + Some(sender) => { + tracing::debug!("Returning sender to existing connection"); + + Ok(sender.clone()) + } + None => { + if let Some((sender, _)) = self.pending_channels.get(&peer) { + tracing::debug!(%peer, "Returning sender for pending connection"); + return Ok(sender.clone()); + } + + tracing::debug!(%peer, "Not connected to peer, initiating dial"); + + self.dial_sender.unbounded_send(peer).map_err(|_| { + io::Error::new( + io::ErrorKind::NotConnected, + "stream behaviour is no longer running", + ) + })?; + + let (sender, receiver) = mpsc::channel(0); + self.pending_channels.insert(peer, (sender.clone(), receiver)); + + Ok(sender) + } + } + } + + pub(crate) fn receiver( + &mut self, + peer: PeerId, + connection: ConnectionId, + ) -> mpsc::Receiver { + if let Some((sender, receiver)) = self.pending_channels.remove(&peer) { + tracing::debug!(%peer, %connection, "Returning existing pending receiver"); + + self.senders.insert(connection, sender); + return receiver; + } + + tracing::debug!(%peer, %connection, "Creating new channel pair"); + + let (sender, receiver) = mpsc::channel(0); + self.senders.insert(connection, sender); + + receiver + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn closing_connection_removes_its_sender() { + let (dial_sender, _dial_receiver) = mpsc::unbounded(); + let mut shared = Shared::new(dial_sender); + let connection = ConnectionId::new_unchecked(1); + let peer = PeerId::random(); + + let _receiver = shared.receiver(peer, connection); + shared.on_connection_established(connection, peer); + + assert!(shared.connections.contains_key(&connection)); + assert!(shared.senders.contains_key(&connection)); + + shared.on_connection_closed(connection); + + assert!(!shared.connections.contains_key(&connection)); + assert!(!shared.senders.contains_key(&connection)); + } +} diff --git a/crates/transport/src/libp2p_stream/tests.rs b/crates/transport/src/libp2p_stream/tests.rs new file mode 100644 index 0000000..07eb539 --- /dev/null +++ b/crates/transport/src/libp2p_stream/tests.rs @@ -0,0 +1,228 @@ +// The first two tests come from upstream's `protocols/stream/tests/lib.rs`; the remaining tests +// cover the local reliability patches described in `mod.rs`. + +use std::{ + future::Future as _, + io, + task::{Context, Poll}, + time::Duration, +}; + +use futures::{ + future::join_all, task::noop_waker, AsyncReadExt as _, AsyncWriteExt as _, StreamExt as _, +}; +use libp2p_identity::PeerId; +use libp2p_swarm::{ + behaviour::DialFailure, ConnectionId, DialError, FromSwarm, NetworkBehaviour, StreamProtocol, + Swarm, ToSwarm, +}; +use libp2p_swarm_test::SwarmExt as _; + +use crate::{libp2p_stream as stream, libp2p_stream::OpenStreamError}; + +const PROTOCOL: StreamProtocol = StreamProtocol::new("/test"); + +#[tokio::test] +async fn dropping_incoming_streams_deregisters() { + let mut swarm1 = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new()); + let mut swarm2 = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new()); + + let mut control = swarm1.behaviour().new_control(); + let mut incoming = swarm2.behaviour().new_control().accept(PROTOCOL).unwrap(); + + swarm2.listen().with_memory_addr_external().await; + swarm1.connect(&mut swarm2).await; + + let swarm2_peer_id = *swarm2.local_peer_id(); + + let handle = tokio::spawn(async move { + while let Some((_, mut stream)) = incoming.next().await { + stream.write_all(&[42]).await.unwrap(); + stream.close().await.unwrap(); + } + }); + tokio::spawn(swarm1.loop_on_next()); + tokio::spawn(swarm2.loop_on_next()); + + let mut stream = control.open_stream(swarm2_peer_id, PROTOCOL).await.unwrap(); + + let mut buf = [0u8; 1]; + stream.read_exact(&mut buf).await.unwrap(); + assert_eq!([42], buf); + + handle.abort(); + let _ = handle.await; + + let error = control.open_stream(swarm2_peer_id, PROTOCOL).await.unwrap_err(); + assert!(matches!(error, OpenStreamError::UnsupportedProtocol(_))); +} + +#[tokio::test] +async fn dial_errors_are_propagated() { + let swarm1 = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new()); + + let mut control = swarm1.behaviour().new_control(); + tokio::spawn(swarm1.loop_on_next()); + + let error = control.open_stream(PeerId::random(), PROTOCOL).await.unwrap_err(); + + let OpenStreamError::Io(e) = error else { + panic!("Unexpected error: {error}") + }; + + assert_eq!(e.kind(), io::ErrorKind::NotConnected); + assert_eq!("Dial error: no addresses for peer.", e.to_string()); +} + +#[test] +fn concurrent_open_streams_queue_every_distinct_dial() { + let mut behaviour = stream::Behaviour::new(); + let mut control1 = behaviour.new_control(); + let mut control2 = behaviour.new_control(); + let mut open1 = Box::pin(control1.open_stream(PeerId::random(), PROTOCOL)); + let mut open2 = Box::pin(control2.open_stream(PeerId::random(), PROTOCOL)); + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(open1.as_mut().poll(&mut cx).is_pending()); + assert!(open2.as_mut().poll(&mut cx).is_pending()); + + assert!(matches!( + NetworkBehaviour::poll(&mut behaviour, &mut cx), + Poll::Ready(ToSwarm::Dial { .. }) + )); + assert!(matches!( + NetworkBehaviour::poll(&mut behaviour, &mut cx), + Poll::Ready(ToSwarm::Dial { .. }) + )); + assert!(NetworkBehaviour::poll(&mut behaviour, &mut cx).is_pending()); +} + +#[test] +fn concurrent_open_streams_to_same_peer_queue_one_dial() { + let mut behaviour = stream::Behaviour::new(); + let peer = PeerId::random(); + let mut control1 = behaviour.new_control(); + let mut control2 = behaviour.new_control(); + let mut open1 = Box::pin(control1.open_stream(peer, PROTOCOL)); + let mut open2 = Box::pin(control2.open_stream(peer, PROTOCOL)); + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + + assert!(open1.as_mut().poll(&mut cx).is_pending()); + assert!(open2.as_mut().poll(&mut cx).is_pending()); + + assert!(matches!( + NetworkBehaviour::poll(&mut behaviour, &mut cx), + Poll::Ready(ToSwarm::Dial { .. }) + )); + assert!(NetworkBehaviour::poll(&mut behaviour, &mut cx).is_pending()); +} + +#[tokio::test] +async fn accept_with_capacity_buffers_unpolled_inbound_streams() { + const STREAM_COUNT: usize = 4; + + let mut client = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new()); + let mut server = Swarm::new_ephemeral_tokio(|_| stream::Behaviour::new()); + let client_peer = *client.local_peer_id(); + let server_peer = *server.local_peer_id(); + let control = client.behaviour().new_control(); + let mut incoming = server + .behaviour() + .new_control() + .accept_with_capacity(PROTOCOL, STREAM_COUNT) + .unwrap(); + + server.listen().with_memory_addr_external().await; + client.connect(&mut server).await; + + tokio::spawn(client.loop_on_next()); + tokio::spawn(server.loop_on_next()); + + let opens = (0..STREAM_COUNT).map(|_| { + let mut control = control.clone(); + async move { control.open_stream(server_peer, PROTOCOL).await } + }); + let opened = tokio::time::timeout(Duration::from_secs(5), join_all(opens)) + .await + .expect("opening buffered streams timed out"); + let _streams: Vec<_> = opened.into_iter().map(Result::unwrap).collect(); + + tokio::time::timeout(Duration::from_secs(5), async { + for _ in 0..STREAM_COUNT { + let (peer, _stream) = incoming.next().await.expect("incoming streams ended"); + assert_eq!(peer, client_peer); + } + }) + .await + .expect("buffered inbound streams were dropped"); +} + +#[tokio::test] +async fn aborted_dial_is_propagated() { + assert_terminal_dial_error_is_propagated(DialError::Aborted).await; +} + +#[tokio::test] +async fn local_peer_dial_error_is_propagated() { + assert_terminal_dial_error_is_propagated(DialError::LocalPeerId { + address: "/memory/1".parse().unwrap(), + }) + .await; +} + +#[tokio::test] +async fn control_outliving_behaviour_returns_error() { + let behaviour = stream::Behaviour::new(); + let mut control = behaviour.new_control(); + drop(behaviour); + + let error = tokio::time::timeout( + Duration::from_secs(1), + control.open_stream(PeerId::random(), PROTOCOL), + ) + .await + .expect("open_stream hung after its behaviour was dropped") + .unwrap_err(); + + let OpenStreamError::Io(error) = error else { + panic!("unexpected error: {error}") + }; + assert_eq!(error.kind(), io::ErrorKind::NotConnected); + assert_eq!(error.to_string(), "stream behaviour is no longer running"); +} + +async fn assert_terminal_dial_error_is_propagated(dial_error: DialError) { + let mut behaviour = stream::Behaviour::new(); + let peer = PeerId::random(); + let mut control = behaviour.new_control(); + let mut open = Box::pin(control.open_stream(peer, PROTOCOL)); + let expected_reason = dial_error.to_string(); + + { + let waker = noop_waker(); + let mut cx = Context::from_waker(&waker); + assert!(open.as_mut().poll(&mut cx).is_pending()); + } + + NetworkBehaviour::on_swarm_event( + &mut behaviour, + FromSwarm::DialFailure(DialFailure { + peer_id: Some(peer), + error: &dial_error, + connection_id: ConnectionId::new_unchecked(1), + }), + ); + + let error = tokio::time::timeout(Duration::from_secs(1), open) + .await + .expect("terminal dial error was not propagated") + .unwrap_err(); + let OpenStreamError::Io(error) = error else { + panic!("unexpected error: {error}") + }; + + assert_eq!(error.kind(), io::ErrorKind::NotConnected); + assert_eq!(error.to_string(), expected_reason); +} diff --git a/crates/transport/src/libp2p_stream/upgrade.rs b/crates/transport/src/libp2p_stream/upgrade.rs new file mode 100644 index 0000000..bbe679f --- /dev/null +++ b/crates/transport/src/libp2p_stream/upgrade.rs @@ -0,0 +1,45 @@ +use std::{ + convert::Infallible, + future::{ready, Ready}, +}; + +use libp2p_core::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}; +use libp2p_swarm::{Stream, StreamProtocol}; + +pub struct Upgrade { + pub(crate) supported_protocols: Vec, +} + +impl UpgradeInfo for Upgrade { + type Info = StreamProtocol; + + type InfoIter = std::vec::IntoIter; + + fn protocol_info(&self) -> Self::InfoIter { + self.supported_protocols.clone().into_iter() + } +} + +impl InboundUpgrade for Upgrade { + type Output = (Stream, StreamProtocol); + + type Error = Infallible; + + type Future = Ready>; + + fn upgrade_inbound(self, socket: Stream, info: Self::Info) -> Self::Future { + ready(Ok((socket, info))) + } +} + +impl OutboundUpgrade for Upgrade { + type Output = (Stream, StreamProtocol); + + type Error = Infallible; + + type Future = Ready>; + + fn upgrade_outbound(self, socket: Stream, info: Self::Info) -> Self::Future { + ready(Ok((socket, info))) + } +}