diff --git a/Cargo.lock b/Cargo.lock
index f3bbc2faf296..a1d226de0cfb 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -178,9 +178,9 @@ dependencies = [
[[package]]
name = "aho-corasick"
-version = "1.0.4"
+version = "1.0.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6748e8def348ed4d14996fa801f4122cd763fff530258cdc03f64b25f89d3a5a"
+checksum = "0c378d78423fdad8089616f827526ee33c19f2fddbd5de1629152c9593ba4783"
dependencies = [
"memchr",
]
@@ -617,7 +617,7 @@ dependencies = [
"num-traits",
"rusticata-macros",
"thiserror",
- "time 0.3.27",
+ "time 0.3.28",
]
[[package]]
@@ -633,7 +633,7 @@ dependencies = [
"num-traits",
"rusticata-macros",
"thiserror",
- "time 0.3.27",
+ "time 0.3.28",
]
[[package]]
@@ -1149,7 +1149,7 @@ checksum = "cd56dd203fef61ac097dd65721a419ddccb106b2d2b70ba60a6b529f03961a51"
dependencies = [
"async-stream-impl",
"futures-core",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
]
[[package]]
@@ -1190,7 +1190,7 @@ dependencies = [
"futures-sink",
"futures-util",
"memchr",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
]
[[package]]
@@ -1285,9 +1285,9 @@ checksum = "9e1b586273c5702936fe7b7d6896644d8be71e6314cfe09d3167c95f712589e8"
[[package]]
name = "base64"
-version = "0.21.2"
+version = "0.21.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "604178f6c5c21f02dc555784810edfb88d34ac2c73b2eae109655649ee73ce3d"
+checksum = "414dcefbc63d77c526a76b3afcf6fbb9b5e2791c19c3aa2297733208750c6e53"
[[package]]
name = "base64ct"
@@ -2139,12 +2139,12 @@ dependencies = [
[[package]]
name = "bstr"
-version = "1.6.0"
+version = "1.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6798148dccfbff0fae41c7574d2fa8f1ef3492fba0face179de5d8d447d67b05"
+checksum = "8042c26c77e5bd6897a7358e0abb3ec412ed126d826988135653fc669263899d"
dependencies = [
"memchr",
- "regex-automata 0.3.6",
+ "regex-automata 0.3.7",
"serde",
]
@@ -3795,6 +3795,7 @@ dependencies = [
"cumulus-relay-chain-interface",
"cumulus-relay-chain-rpc-interface",
"futures",
+ "parking_lot 0.12.1",
"polkadot-availability-recovery",
"polkadot-collator-protocol",
"polkadot-core-primitives",
@@ -4085,9 +4086,9 @@ dependencies = [
[[package]]
name = "cxx"
-version = "1.0.106"
+version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28403c86fc49e3401fdf45499ba37fad6493d9329449d6449d7f0e10f4654d28"
+checksum = "bbe98ba1789d56fb3db3bee5e032774d4f421b685de7ba703643584ba24effbe"
dependencies = [
"cc",
"cxxbridge-flags",
@@ -4097,9 +4098,9 @@ dependencies = [
[[package]]
name = "cxx-build"
-version = "1.0.106"
+version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78da94fef01786dc3e0c76eafcd187abcaa9972c78e05ff4041e24fdf059c285"
+checksum = "c4ce20f6b8433da4841b1dadfb9468709868022d829d5ca1f2ffbda928455ea3"
dependencies = [
"cc",
"codespan-reporting",
@@ -4112,15 +4113,15 @@ dependencies = [
[[package]]
name = "cxxbridge-flags"
-version = "1.0.106"
+version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2a6f5e1dfb4b34292ad4ea1facbfdaa1824705b231610087b00b17008641809"
+checksum = "20888d9e1d2298e2ff473cee30efe7d5036e437857ab68bbfea84c74dba91da2"
[[package]]
name = "cxxbridge-macro"
-version = "1.0.106"
+version = "1.0.107"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50c49547d73ba8dcfd4ad7325d64c6d5391ff4224d498fc39a6f3f49825a530d"
+checksum = "2fa16a70dd58129e4dfffdff535fb1bce66673f7bbeec4a5a1765a504e1ccd84"
dependencies = [
"proc-macro2",
"quote",
@@ -4164,9 +4165,9 @@ dependencies = [
[[package]]
name = "dashmap"
-version = "5.5.1"
+version = "5.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "edd72493923899c6f10c641bdbdeddc7183d6396641d99c1a0d1597f37f92e28"
+checksum = "978747c1d849a7d2ee5e8adc0159961c48fb7e5db2f06af6723b80123bb53856"
dependencies = [
"cfg-if",
"hashbrown 0.14.0",
@@ -4782,9 +4783,9 @@ checksum = "5443807d6dff69373d433ab9ef5378ad8df50ca6298caf15de6e52e24aaf54d5"
[[package]]
name = "erased-serde"
-version = "0.3.30"
+version = "0.3.31"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "837c0466252947ada828b975e12daf82e18bb5444e4df87be6038d4469e2a3d2"
+checksum = "6c138974f9d5e7fe373eb04df7cae98833802ae4b11c24ac7039a21d5af4b26c"
dependencies = [
"serde",
]
@@ -4801,9 +4802,9 @@ dependencies = [
[[package]]
name = "errno"
-version = "0.3.2"
+version = "0.3.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6b30f669a7961ef1631673d2766cc92f52d64f7ef354d4fe0ddfd30ed52f0f4f"
+checksum = "136526188508e25c6fef639d7927dfb3e0e3084488bf202267829cf7fc23dbdd"
dependencies = [
"errno-dragonfly",
"libc",
@@ -5565,7 +5566,7 @@ version = "0.6.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2eeb4ed9e12f43b7fa0baae3f9cdda28352770132ef2e09a23760c29cae8bd47"
dependencies = [
- "rustix 0.38.8",
+ "rustix 0.38.10",
"windows-sys 0.48.0",
]
@@ -5641,7 +5642,7 @@ dependencies = [
"futures-io",
"memchr",
"parking",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"waker-fn",
]
@@ -5663,7 +5664,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "d2411eed028cdf8c8034eaf21f9915f956b6c3abec4d4c7949ee67f0721127bd"
dependencies = [
"futures-io",
- "rustls 0.20.8",
+ "rustls 0.20.9",
"webpki 0.22.0",
]
@@ -5698,7 +5699,7 @@ dependencies = [
"futures-sink",
"futures-task",
"memchr",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"pin-utils",
"slab",
]
@@ -6107,7 +6108,7 @@ checksum = "d5f38f16d184e36f2408a55281cd658ecbd3ca05cce6d6510a176eca393e26d1"
dependencies = [
"bytes",
"http",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
]
[[package]]
@@ -6150,7 +6151,7 @@ dependencies = [
"httparse",
"httpdate",
"itoa",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"socket2 0.4.9",
"tokio",
"tower-service",
@@ -6168,7 +6169,7 @@ dependencies = [
"http",
"hyper",
"log",
- "rustls 0.21.6",
+ "rustls 0.21.7",
"rustls-native-certs",
"tokio",
"tokio-rustls",
@@ -6513,7 +6514,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "cb0889898416213fab133e1d33a0e5858a48177452750691bde3666d0fdbaf8b"
dependencies = [
"hermit-abi 0.3.2",
- "rustix 0.38.8",
+ "rustix 0.38.10",
"windows-sys 0.48.0",
]
@@ -7261,7 +7262,7 @@ dependencies = [
"parking_lot 0.12.1",
"quinn-proto",
"rand 0.8.5",
- "rustls 0.20.8",
+ "rustls 0.20.9",
"thiserror",
"tokio",
]
@@ -7342,7 +7343,7 @@ dependencies = [
"libp2p-identity",
"rcgen 0.10.0",
"ring 0.16.20",
- "rustls 0.20.8",
+ "rustls 0.20.9",
"thiserror",
"webpki 0.22.0",
"x509-parser 0.14.0",
@@ -7526,9 +7527,9 @@ dependencies = [
[[package]]
name = "linregress"
-version = "0.5.2"
+version = "0.5.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4de0b5f52a9f84544d268f5fabb71b38962d6aa3c6600b8bcd27d44ccf9c9c45"
+checksum = "4de04dcecc58d366391f9920245b85ffa684558a5ef6e7736e754347c3aea9c2"
dependencies = [
"nalgebra",
]
@@ -7756,9 +7757,9 @@ dependencies = [
[[package]]
name = "memchr"
-version = "2.5.0"
+version = "2.6.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2dffe52ecf27772e601905b7522cb4ef790d2cc203488bbd0e2fe85fcb74566d"
+checksum = "f478948fd84d9f8e86967bf432640e46adfb5a4bd4f14ef7e864ab38220534ae"
[[package]]
name = "memfd"
@@ -8169,16 +8170,15 @@ dependencies = [
[[package]]
name = "nix"
-version = "0.26.2"
+version = "0.26.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bfdda3d196821d6af13126e40375cdf7da646a96114af134d5f417a9a1dc8e1a"
+checksum = "598beaf3cc6fdd9a5dfb1630c2800c7acd31df7aaf0f565796fba2b53ca1af1b"
dependencies = [
"bitflags 1.3.2",
"cfg-if",
"libc",
"memoffset 0.7.1",
"pin-utils",
- "static_assertions",
]
[[package]]
@@ -8239,7 +8239,7 @@ dependencies = [
"jsonrpsee",
"kitchensink-runtime",
"log",
- "nix 0.26.2",
+ "nix 0.26.4",
"node-executor",
"node-inspect",
"node-primitives",
@@ -11055,9 +11055,9 @@ dependencies = [
[[package]]
name = "parity-scale-codec"
-version = "3.6.4"
+version = "3.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dd8e946cc0cc711189c0b0249fb8b599cbeeab9784d83c415719368bb8d4ac64"
+checksum = "0dec8a8073036902368c2cdc0387e85ff9a37054d7e7c98e592145e0c92cd4fb"
dependencies = [
"arrayvec 0.7.4",
"bitvec",
@@ -11070,9 +11070,9 @@ dependencies = [
[[package]]
name = "parity-scale-codec-derive"
-version = "3.6.4"
+version = "3.6.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2a296c3079b5fefbc499e1de58dc26c09b1b9a5952d26694ee89f04a43ebbb3e"
+checksum = "312270ee71e1cd70289dacf597cab7b207aa107d2f28191c2ae45b2ece18a260"
dependencies = [
"proc-macro-crate",
"proc-macro2",
@@ -11385,9 +11385,9 @@ checksum = "257b64915a082f7811703966789728173279bdebb956b143dbcd23f6f970a777"
[[package]]
name = "pin-project-lite"
-version = "0.2.12"
+version = "0.2.13"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12cc1b0bf1727a77a54b6654e7b5f1af8604923edc8b81885f8ec92f9e3f0a05"
+checksum = "8afb450f006bf6385ca15ef45d71d2288452bc3683ce2e2cacc0d18e4be60b58"
[[package]]
name = "pin-utils"
@@ -11423,9 +11423,9 @@ checksum = "26072860ba924cbfa98ea39c8c19b4dd6a4a25423dbdf219c1eca91aa0cf6964"
[[package]]
name = "platforms"
-version = "3.0.2"
+version = "3.1.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e3d7ddaed09e0eb771a79ab0fd64609ba0afb0a8366421957936ad14cbd13630"
+checksum = "4503fa043bf02cee09a9582e9554b4c6403b2ef55e4612e96561d294419429f8"
[[package]]
name = "plotters"
@@ -11461,7 +11461,7 @@ version = "1.0.0"
dependencies = [
"assert_cmd",
"color-eyre",
- "nix 0.26.2",
+ "nix 0.26.4",
"polkadot-cli",
"polkadot-core-primitives",
"polkadot-node-core-pvf",
@@ -12429,7 +12429,7 @@ dependencies = [
"hex-literal 0.4.1",
"jsonrpsee",
"log",
- "nix 0.26.2",
+ "nix 0.26.4",
"pallet-transaction-payment-rpc",
"parachains-common",
"parity-scale-codec",
@@ -12841,6 +12841,7 @@ dependencies = [
"pallet-transaction-payment-rpc-runtime-api",
"parity-db",
"parity-scale-codec",
+ "parking_lot 0.12.1",
"polkadot-approval-distribution",
"polkadot-availability-bitfield-distribution",
"polkadot-availability-distribution",
@@ -13187,7 +13188,7 @@ dependencies = [
"concurrent-queue",
"libc",
"log",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"windows-sys 0.48.0",
]
@@ -13239,9 +13240,9 @@ dependencies = [
[[package]]
name = "portable-atomic"
-version = "1.4.2"
+version = "1.4.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f32154ba0af3a075eefa1eda8bb414ee928f62303a54ea85b8d6638ff1a6ee9e"
+checksum = "31114a898e107c51bb1609ffaf55a0e011cf6a4d7f1170d0015a165082c0338b"
[[package]]
name = "portpicker"
@@ -13263,7 +13264,7 @@ dependencies = [
"findshlibs",
"libc",
"log",
- "nix 0.26.2",
+ "nix 0.26.4",
"once_cell",
"parking_lot 0.12.1",
"smallvec",
@@ -13423,9 +13424,9 @@ checksum = "dc375e1527247fe1a97d8b7156678dfe7c1af2fc075c9a4db3690ecd2a148068"
[[package]]
name = "proc-macro-warning"
-version = "0.4.1"
+version = "0.4.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "70550716265d1ec349c41f70dd4f964b4fd88394efe4405f0c1da679c4799a07"
+checksum = "3d1eaa7fa0aa1929ffdf7eeb6eac234dde6268914a14ad44d23521ab6a9b258e"
dependencies = [
"proc-macro2",
"quote",
@@ -13641,7 +13642,7 @@ dependencies = [
"rand 0.8.5",
"ring 0.16.20",
"rustc-hash",
- "rustls 0.20.8",
+ "rustls 0.20.9",
"slab",
"thiserror",
"tinyvec",
@@ -13790,7 +13791,7 @@ checksum = "6413f3de1edee53342e6138e75b56d32e7bc6e332b3bd62d497b1929d4cfbcdd"
dependencies = [
"pem",
"ring 0.16.20",
- "time 0.3.27",
+ "time 0.3.28",
"x509-parser 0.13.2",
"yasna",
]
@@ -13803,7 +13804,7 @@ checksum = "ffbe84efe2f38dea12e9bfc1f65377fdf03e53a18cb3b995faedf7934c7e785b"
dependencies = [
"pem",
"ring 0.16.20",
- "time 0.3.27",
+ "time 0.3.28",
"yasna",
]
@@ -13883,14 +13884,14 @@ dependencies = [
[[package]]
name = "regex"
-version = "1.9.3"
+version = "1.9.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "81bc1d4caf89fac26a70747fe603c130093b53c773888797a6329091246d651a"
+checksum = "12de2eff854e5fa4b1295edd650e227e9d8fb0c9e90b12e7f36d6a6811791a29"
dependencies = [
"aho-corasick",
"memchr",
- "regex-automata 0.3.6",
- "regex-syntax 0.7.4",
+ "regex-automata 0.3.7",
+ "regex-syntax 0.7.5",
]
[[package]]
@@ -13904,13 +13905,13 @@ dependencies = [
[[package]]
name = "regex-automata"
-version = "0.3.6"
+version = "0.3.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fed1ceff11a1dddaee50c9dc8e4938bd106e9d89ae372f192311e7da498e3b69"
+checksum = "49530408a136e16e5b486e883fbb6ba058e8e4e8ae6621a77b048b314336e629"
dependencies = [
"aho-corasick",
"memchr",
- "regex-syntax 0.7.4",
+ "regex-syntax 0.7.5",
]
[[package]]
@@ -13921,9 +13922,9 @@ checksum = "f162c6dd7b008981e4d40210aca20b4bd0f9b60ca9271061b07f78537722f2e1"
[[package]]
name = "regex-syntax"
-version = "0.7.4"
+version = "0.7.5"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e5ea92a5b6195c6ef2a0295ea818b312502c6fc94dde986c5553242e18fd4ce2"
+checksum = "dbb5fb1acd8a1a18b3dd5be62d25485eb770e05afb408a9627d14d451bae12da"
[[package]]
name = "remote-ext-tests-bags-list"
@@ -13950,7 +13951,7 @@ version = "0.11.20"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "3e9ad3fe7488d7e34558a2033d45a0c90b72d97b4f80705666fea71472e2e6a1"
dependencies = [
- "base64 0.21.2",
+ "base64 0.21.3",
"bytes",
"encoding_rs",
"futures-core",
@@ -13966,8 +13967,8 @@ dependencies = [
"mime",
"once_cell",
"percent-encoding",
- "pin-project-lite 0.2.12",
- "rustls 0.21.6",
+ "pin-project-lite 0.2.13",
+ "rustls 0.21.7",
"rustls-pemfile",
"serde",
"serde_json",
@@ -14361,9 +14362,9 @@ dependencies = [
[[package]]
name = "rustix"
-version = "0.38.8"
+version = "0.38.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "19ed4fa021d81c8392ce04db050a3da9a60299050b7ae1cf482d862b54a7218f"
+checksum = "ed6248e1caa625eb708e266e06159f135e8c26f2bb7ceb72dc4b2766d0340964"
dependencies = [
"bitflags 2.4.0",
"errno",
@@ -14387,9 +14388,9 @@ dependencies = [
[[package]]
name = "rustls"
-version = "0.20.8"
+version = "0.20.9"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fff78fc74d175294f4e83b28343315ffcfb114b156f0185e9741cb5570f50e2f"
+checksum = "1b80e3dec595989ea8510028f30c408a4630db12c9cbb8de34203b89d6577e99"
dependencies = [
"log",
"ring 0.16.20",
@@ -14399,9 +14400,9 @@ dependencies = [
[[package]]
name = "rustls"
-version = "0.21.6"
+version = "0.21.7"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1d1feddffcfcc0b33f5c6ce9a29e341e4cd59c3f78e7ee45f4a40c038b1d6cbb"
+checksum = "cd8d6c9f025a446bc4d18ad9632e69aec8f287aa84499ee335599fabd20c3fd8"
dependencies = [
"log",
"ring 0.16.20",
@@ -14427,7 +14428,7 @@ version = "1.0.3"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "2d3987094b1d07b653b7dfdc3f70ce9a1da9c51ac18c1b06b662e4f9a0e9f4b2"
dependencies = [
- "base64 0.21.2",
+ "base64 0.21.3",
]
[[package]]
@@ -15245,6 +15246,7 @@ dependencies = [
"tempfile",
"thiserror",
"tokio",
+ "tokio-stream",
"tokio-test",
"tokio-util",
"unsigned-varint",
@@ -15300,10 +15302,12 @@ name = "sc-network-gossip"
version = "0.10.0-dev"
dependencies = [
"ahash 0.8.3",
+ "async-trait",
"futures",
"futures-timer",
"libp2p",
"log",
+ "parity-scale-codec",
"quickcheck",
"sc-network",
"sc-network-common",
@@ -16528,7 +16532,7 @@ dependencies = [
"arrayvec 0.7.4",
"async-lock",
"atomic-take",
- "base64 0.21.2",
+ "base64 0.21.3",
"bip39",
"blake2-rfc",
"bs58 0.5.0",
@@ -16581,7 +16585,7 @@ checksum = "256b5bad1d6b49045e95fe87492ce73d5af81545d8b4d8318a872d2007024c33"
dependencies = [
"async-channel",
"async-lock",
- "base64 0.21.2",
+ "base64 0.21.3",
"blake2-rfc",
"derive_more",
"either",
@@ -18077,7 +18081,7 @@ version = "0.1.0"
dependencies = [
"assert_cmd",
"futures",
- "nix 0.26.2",
+ "nix 0.26.4",
"node-cli",
"node-primitives",
"regex",
@@ -18515,7 +18519,7 @@ dependencies = [
"cfg-if",
"fastrand 2.0.0",
"redox_syscall 0.3.5",
- "rustix 0.38.8",
+ "rustix 0.38.10",
"windows-sys 0.48.0",
]
@@ -18775,9 +18779,9 @@ dependencies = [
[[package]]
name = "time"
-version = "0.3.27"
+version = "0.3.28"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0bb39ee79a6d8de55f48f2293a830e040392f1c5f16e336bdd1788cd0aadce07"
+checksum = "17f6bb557fd245c28e6411aa56b6403c689ad95061f50e4be16c274e70a17e48"
dependencies = [
"deranged",
"itoa",
@@ -18794,9 +18798,9 @@ checksum = "7300fbefb4dadc1af235a9cef3737cea692a9d97e1b9cbcd4ebdae6f8868e6fb"
[[package]]
name = "time-macros"
-version = "0.2.13"
+version = "0.2.14"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "733d258752e9303d392b94b75230d07b0b9c489350c69b851fc6c065fde3e8f9"
+checksum = "1a942f44339478ef67935ab2bbaec2fb0322496cf3cbe84b261e06ac3814c572"
dependencies = [
"time-core",
]
@@ -18866,7 +18870,7 @@ dependencies = [
"mio",
"num_cpus",
"parking_lot 0.12.1",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"signal-hook-registry",
"socket2 0.5.3",
"tokio-macros",
@@ -18901,7 +18905,7 @@ version = "0.24.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c28327cf380ac148141087fbfb9de9d7bd4e84ab5d2c28fbc911d753de8a7081"
dependencies = [
- "rustls 0.21.6",
+ "rustls 0.21.7",
"tokio",
]
@@ -18912,7 +18916,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "397c988d37662c7dda6d2208364a706264bf3d6138b11d436cbac0ad38832842"
dependencies = [
"futures-core",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"tokio",
"tokio-util",
]
@@ -18952,7 +18956,7 @@ dependencies = [
"futures-core",
"futures-io",
"futures-sink",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"tokio",
"tracing",
]
@@ -19024,7 +19028,7 @@ dependencies = [
"http",
"http-body",
"http-range-header",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"tower-layer",
"tower-service",
]
@@ -19049,7 +19053,7 @@ checksum = "8ce8c33a8d48bd45d624a6e523445fd21ec13d3653cd51f681abf67418f54eb8"
dependencies = [
"cfg-if",
"log",
- "pin-project-lite 0.2.12",
+ "pin-project-lite 0.2.13",
"tracing-attributes",
"tracing-core",
]
@@ -19480,9 +19484,9 @@ checksum = "a156c684c91ea7d62626509bce3cb4e1d9ed5c4d978f7b4352658f96a4c26b4a"
[[package]]
name = "url"
-version = "2.4.0"
+version = "2.4.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "50bff7831e19200a85b17131d085c25d7811bc4e186efdaf54bbd132994a88cb"
+checksum = "143b538f18257fac9cad154828a57c6bf5157e1aa604d4816b5995bf6de87ae5"
dependencies = [
"form_urlencoded",
"idna 0.4.0",
@@ -19725,9 +19729,9 @@ checksum = "ca6ad05a4870b2bf5fe995117d3728437bd27d7cd5f06f13c17443ef369775a1"
[[package]]
name = "wasm-encoder"
-version = "0.31.1"
+version = "0.32.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "41763f20eafed1399fff1afb466496d3a959f58241436cfdc17e3f5ca954de16"
+checksum = "1ba64e81215916eaeb48fee292f29401d69235d62d8b8fd92a7b2844ec5ae5f7"
dependencies = [
"leb128",
]
@@ -19899,7 +19903,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "c86437fa68626fe896e5afc69234bb2b5894949083586535f200385adfd71213"
dependencies = [
"anyhow",
- "base64 0.21.2",
+ "base64 0.21.3",
"bincode",
"directories-next",
"file-per-thread-logger",
@@ -20052,9 +20056,9 @@ dependencies = [
[[package]]
name = "wast"
-version = "63.0.0"
+version = "64.0.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2560471f60a48b77fccefaf40796fda61c97ce1e790b59dfcec9dc3995c9f63a"
+checksum = "a259b226fd6910225aa7baeba82f9d9933b6d00f2ce1b49b80fa4214328237cc"
dependencies = [
"leb128",
"memchr",
@@ -20064,9 +20068,9 @@ dependencies = [
[[package]]
name = "wat"
-version = "1.0.70"
+version = "1.0.71"
source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3bdc306c2c4c2f2bf2ba69e083731d0d2a77437fc6a350a19db139636e7e416c"
+checksum = "53253d920ab413fca1c7dc2161d601c79b4fdf631d0ba51dd4343bf9b556c3f6"
dependencies = [
"wast",
]
@@ -20151,7 +20155,7 @@ dependencies = [
"sha2 0.10.7",
"stun",
"thiserror",
- "time 0.3.27",
+ "time 0.3.28",
"tokio",
"turn",
"url",
@@ -20759,7 +20763,7 @@ dependencies = [
"ring 0.16.20",
"rusticata-macros",
"thiserror",
- "time 0.3.27",
+ "time 0.3.28",
]
[[package]]
@@ -20777,7 +20781,7 @@ dependencies = [
"oid-registry 0.6.1",
"rusticata-macros",
"thiserror",
- "time 0.3.27",
+ "time 0.3.28",
]
[[package]]
@@ -20947,7 +20951,7 @@ version = "0.5.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e17bb3549cc1321ae1296b9cdc2698e2b6cb1992adfa19a8c72e5b7a738f44cd"
dependencies = [
- "time 0.3.27",
+ "time 0.3.28",
]
[[package]]
diff --git a/cumulus/client/relay-chain-minimal-node/Cargo.toml b/cumulus/client/relay-chain-minimal-node/Cargo.toml
index 39056d6b6511..7ecc7ef15537 100644
--- a/cumulus/client/relay-chain-minimal-node/Cargo.toml
+++ b/cumulus/client/relay-chain-minimal-node/Cargo.toml
@@ -40,6 +40,7 @@ schnellru = "0.2.1"
tracing = "0.1.37"
async-trait = "0.1.73"
futures = "0.3.28"
+parking_lot = "0.12.1"
[features]
network-protocol-staging = [
diff --git a/cumulus/client/relay-chain-minimal-node/src/collator_overseer.rs b/cumulus/client/relay-chain-minimal-node/src/collator_overseer.rs
index 491758c1329a..c7d536d116ca 100644
--- a/cumulus/client/relay-chain-minimal-node/src/collator_overseer.rs
+++ b/cumulus/client/relay-chain-minimal-node/src/collator_overseer.rs
@@ -15,8 +15,9 @@
// along with Polkadot. If not, see .
use futures::{select, StreamExt};
+use parking_lot::Mutex;
use schnellru::{ByLength, LruMap};
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
use polkadot_availability_recovery::AvailabilityRecoverySubsystem;
use polkadot_collator_protocol::{CollatorProtocolSubsystem, ProtocolSide};
@@ -27,7 +28,7 @@ use polkadot_network_bridge::{
use polkadot_node_collation_generation::CollationGenerationSubsystem;
use polkadot_node_core_runtime_api::RuntimeApiSubsystem;
use polkadot_node_network_protocol::{
- peer_set::PeerSetProtocolNames,
+ peer_set::{PeerSet, PeerSetProtocolNames},
request_response::{
v1::{self, AvailableDataFetchingRequest},
vstaging, IncomingRequestReceiver, ReqProtocolNames,
@@ -41,7 +42,7 @@ use polkadot_overseer::{
use polkadot_primitives::CollatorPair;
use sc_authority_discovery::Service as AuthorityDiscoveryService;
-use sc_network::NetworkStateInfo;
+use sc_network::{NetworkStateInfo, NotificationService};
use sc_service::TaskManager;
use sp_runtime::traits::Block as BlockT;
@@ -77,6 +78,8 @@ pub(crate) struct CollatorOverseerGenArgs<'a> {
pub req_protocol_names: ReqProtocolNames,
/// Peerset protocols name mapping
pub peer_set_protocol_names: PeerSetProtocolNames,
+ /// Notification services for validation/collation protocols.
+ pub notification_services: HashMap>,
}
fn build_overseer(
@@ -94,6 +97,7 @@ fn build_overseer(
collator_pair,
req_protocol_names,
peer_set_protocol_names,
+ notification_services,
}: CollatorOverseerGenArgs<'_>,
) -> Result<
(Overseer, Arc>, OverseerHandle),
@@ -101,6 +105,8 @@ fn build_overseer(
> {
let spawner = SpawnGlue(spawner);
let network_bridge_metrics: NetworkBridgeMetrics = Metrics::register(registry)?;
+ let notification_sinks = Arc::new(Mutex::new(HashMap::new()));
+
let builder = Overseer::builder()
.availability_distribution(DummySubsystem)
.availability_recovery(AvailabilityRecoverySubsystem::with_availability_store_skip(
@@ -131,6 +137,8 @@ fn build_overseer(
sync_oracle,
network_bridge_metrics.clone(),
peer_set_protocol_names.clone(),
+ notification_services,
+ notification_sinks.clone(),
))
.network_bridge_tx(NetworkBridgeTxSubsystem::new(
network_service,
@@ -138,6 +146,7 @@ fn build_overseer(
network_bridge_metrics,
req_protocol_names,
peer_set_protocol_names,
+ notification_sinks,
))
.provisioner(DummySubsystem)
.runtime_api(RuntimeApiSubsystem::new(
diff --git a/cumulus/client/relay-chain-minimal-node/src/lib.rs b/cumulus/client/relay-chain-minimal-node/src/lib.rs
index 366d428eda70..cae093678ad6 100644
--- a/cumulus/client/relay-chain-minimal-node/src/lib.rs
+++ b/cumulus/client/relay-chain-minimal-node/src/lib.rs
@@ -21,7 +21,7 @@ use cumulus_relay_chain_rpc_interface::{RelayChainRpcClient, RelayChainRpcInterf
use network::build_collator_network;
use polkadot_network_bridge::{peer_sets_info, IsAuthority};
use polkadot_node_network_protocol::{
- peer_set::PeerSetProtocolNames,
+ peer_set::{PeerSet, PeerSetProtocolNames},
request_response::{
v1, vstaging, IncomingRequest, IncomingRequestReceiver, Protocol, ReqProtocolNames,
},
@@ -176,10 +176,13 @@ async fn new_minimal_relay_chain(
let peer_set_protocol_names =
PeerSetProtocolNames::new(genesis_hash, config.chain_spec.fork_id());
let is_authority = if role.is_authority() { IsAuthority::Yes } else { IsAuthority::No };
-
- for config in peer_sets_info(is_authority, &peer_set_protocol_names) {
- net_config.add_notification_protocol(config);
- }
+ let notification_services = peer_sets_info(is_authority, &peer_set_protocol_names)
+ .into_iter()
+ .map(|(config, (peerset, service))| {
+ net_config.add_notification_protocol(config);
+ (peerset, service)
+ })
+ .collect::>>();
let request_protocol_names = ReqProtocolNames::new(genesis_hash, config.chain_spec.fork_id());
let (collation_req_receiver_v1, collation_req_receiver_vstaging, available_data_req_receiver) =
@@ -219,6 +222,7 @@ async fn new_minimal_relay_chain(
collator_pair,
req_protocol_names: request_protocol_names,
peer_set_protocol_names,
+ notification_services,
};
let overseer_handle =
diff --git a/cumulus/client/relay-chain-minimal-node/src/network.rs b/cumulus/client/relay-chain-minimal-node/src/network.rs
index f39d7a26dd88..0258cf9ddd58 100644
--- a/cumulus/client/relay-chain-minimal-node/src/network.rs
+++ b/cumulus/client/relay-chain-minimal-node/src/network.rs
@@ -25,10 +25,9 @@ use sc_network::{
NetworkService,
};
-use sc_network::config::FullNetworkConfiguration;
+use sc_network::{config::FullNetworkConfiguration, NotificationService};
use sc_network_common::{role::Roles, sync::message::BlockAnnouncesHandshake};
use sc_service::{error::Error, Configuration, NetworkStarter, SpawnTaskHandle};
-use sc_utils::mpsc::tracing_unbounded;
use std::{iter, sync::Arc};
@@ -44,7 +43,7 @@ pub(crate) fn build_collator_network(
Error,
> {
let protocol_id = config.protocol_id();
- let block_announce_config = get_block_announce_proto_config::(
+ let (block_announce_config, _notification_service) = get_block_announce_proto_config::(
protocol_id.clone(),
&None,
Roles::from(&config.role),
@@ -64,8 +63,6 @@ pub(crate) fn build_collator_network(
let peer_store_handle = peer_store.handle();
spawn_handle.spawn("peer-store", Some("networking"), peer_store.run());
- // RX is not used for anything because syncing is not started for the minimal node
- let (tx, _rx) = tracing_unbounded("mpsc_syncing_engine_protocol", 100_000);
let network_params = sc_network::config::Params:: {
role: config.role.clone(),
executor: {
@@ -81,7 +78,6 @@ pub(crate) fn build_collator_network(
protocol_id,
metrics_registry: config.prometheus_config.as_ref().map(|config| config.registry.clone()),
block_announce_config,
- tx,
};
let network_worker = sc_network::NetworkWorker::new(network_params)?;
@@ -133,7 +129,7 @@ fn get_block_announce_proto_config(
best_number: NumberFor,
best_hash: B::Hash,
genesis_hash: B::Hash,
-) -> NonDefaultSetConfig {
+) -> (NonDefaultSetConfig, Box) {
let block_announces_protocol = {
let genesis_hash = genesis_hash.as_ref();
if let Some(ref fork_id) = fork_id {
@@ -143,12 +139,11 @@ fn get_block_announce_proto_config(
}
};
- NonDefaultSetConfig {
- notifications_protocol: block_announces_protocol.into(),
- fallback_names: iter::once(format!("/{}/block-announces/1", protocol_id.as_ref()).into())
- .collect(),
- max_notification_size: 1024 * 1024,
- handshake: Some(NotificationHandshake::new(BlockAnnouncesHandshake::::build(
+ NonDefaultSetConfig::new(
+ block_announces_protocol.into(),
+ iter::once(format!("/{}/block-announces/1", protocol_id.as_ref()).into()).collect(),
+ 1024 * 1024,
+ Some(NotificationHandshake::new(BlockAnnouncesHandshake::::build(
roles,
best_number,
best_hash,
@@ -156,11 +151,11 @@ fn get_block_announce_proto_config(
))),
// NOTE: `set_config` will be ignored by `protocol.rs` as the block announcement
// protocol is still hardcoded into the peerset.
- set_config: SetConfig {
+ SetConfig {
in_peers: 0,
out_peers: 0,
reserved_nodes: Vec::new(),
non_reserved_mode: NonReservedPeerMode::Deny,
},
- }
+ )
}
diff --git a/polkadot/node/network/bridge/src/lib.rs b/polkadot/node/network/bridge/src/lib.rs
index 46d4a00faace..ddce99d5c2a8 100644
--- a/polkadot/node/network/bridge/src/lib.rs
+++ b/polkadot/node/network/bridge/src/lib.rs
@@ -83,6 +83,7 @@ pub(crate) enum WireMessage {
ViewUpdate(View),
}
+#[derive(Debug)]
pub(crate) struct PeerData {
/// The Latest view sent by the peer.
view: View,
diff --git a/polkadot/node/network/bridge/src/network.rs b/polkadot/node/network/bridge/src/network.rs
index 4f21212dcb64..1bdc602dfe04 100644
--- a/polkadot/node/network/bridge/src/network.rs
+++ b/polkadot/node/network/bridge/src/network.rs
@@ -14,21 +14,24 @@
// You should have received a copy of the GNU General Public License
// along with Polkadot. If not, see .
-use std::{collections::HashSet, sync::Arc};
+use std::{
+ collections::{HashMap, HashSet},
+ sync::Arc,
+};
use async_trait::async_trait;
-use futures::{prelude::*, stream::BoxStream};
+use parking_lot::Mutex;
use parity_scale_codec::Encode;
use sc_network::{
- config::parse_addr, multiaddr::Multiaddr, types::ProtocolName, Event as NetworkEvent,
- IfDisconnected, NetworkEventStream, NetworkNotification, NetworkPeers, NetworkRequest,
- NetworkService, OutboundFailure, ReputationChange, RequestFailure,
+ config::parse_addr, multiaddr::Multiaddr, types::ProtocolName, IfDisconnected, MessageSink,
+ NetworkPeers, NetworkRequest, NetworkService, OutboundFailure, ReputationChange,
+ RequestFailure,
};
use polkadot_node_network_protocol::{
- peer_set::{PeerSet, PeerSetProtocolNames, ProtocolVersion},
+ peer_set::{PeerSet, ProtocolVersion},
request_response::{OutgoingRequest, Recipient, ReqProtocolNames, Requests},
PeerId,
};
@@ -45,51 +48,50 @@ const LOG_TARGET: &'static str = "parachain::network-bridge-net";
/// messages that are compatible with the passed peer set, as that is currently not enforced by
/// this function. These are messages of type `WireMessage` parameterized on the matching type.
pub(crate) fn send_message(
- net: &mut impl Network,
mut peers: Vec,
peer_set: PeerSet,
version: ProtocolVersion,
- protocol_names: &PeerSetProtocolNames,
message: M,
metrics: &super::Metrics,
+ network_notification_sinks: &Arc>>>,
) where
M: Encode + Clone,
{
if peers.is_empty() {
return
}
+
let message = {
let encoded = message.encode();
metrics.on_notification_sent(peer_set, version, encoded.len(), peers.len());
encoded
};
+ let notification_sinks = network_notification_sinks.lock();
+
// optimization: avoid cloning the message for the last peer in the
// list. The message payload can be quite large. If the underlying
// network used `Bytes` this would not be necessary.
+ //
+ // peer may have gotten disconnect by the time `send_message()` is called
+ // at which point the the sink is not available.
let last_peer = peers.pop();
-
- // We always send messages on the "main" name even when a negotiated
- // fallback is used. The libp2p implementation handles the fallback
- // under the hood.
- let protocol_name = protocol_names.get_main_name(peer_set);
peers.into_iter().for_each(|peer| {
- net.write_notification(peer, protocol_name.clone(), message.clone());
+ if let Some(sink) = notification_sinks.get(&(peer_set, peer)) {
+ sink.send_sync_notification(message.clone());
+ }
});
+
if let Some(peer) = last_peer {
- net.write_notification(peer, protocol_name, message);
+ if let Some(sink) = notification_sinks.get(&(peer_set, peer)) {
+ sink.send_sync_notification(message.clone());
+ }
}
}
/// An abstraction over networking for the purposes of this subsystem.
#[async_trait]
pub trait Network: Clone + Send + 'static {
- /// Get a stream of all events occurring on the network. This may include events unrelated
- /// to the Polkadot protocol - the user of this function should filter only for events related
- /// to the [`VALIDATION_PROTOCOL_NAME`](VALIDATION_PROTOCOL_NAME)
- /// or [`COLLATION_PROTOCOL_NAME`](COLLATION_PROTOCOL_NAME)
- fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent>;
-
/// Ask the network to keep a substream open with these nodes and not disconnect from them
/// until removed from the protocol's peer set.
/// Note that `out_peers` setting has no effect on this.
@@ -121,16 +123,12 @@ pub trait Network: Clone + Send + 'static {
/// Disconnect a given peer from the protocol specified without harming reputation.
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName);
- /// Write a notification to a peer on the given protocol.
- fn write_notification(&self, who: PeerId, protocol: ProtocolName, message: Vec);
+ /// Get peer role.
+ fn peer_role(&self, who: PeerId, handshake: Vec) -> Option;
}
#[async_trait]
impl Network for Arc> {
- fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
- NetworkService::event_stream(self, "polkadot-network-bridge").boxed()
- }
-
async fn set_reserved_peers(
&mut self,
protocol: ProtocolName,
@@ -155,10 +153,6 @@ impl Network for Arc> {
NetworkService::disconnect_peer(&**self, who, protocol);
}
- fn write_notification(&self, who: PeerId, protocol: ProtocolName, message: Vec) {
- NetworkService::write_notification(&**self, who, protocol, message);
- }
-
async fn start_request(
&self,
authority_discovery: &mut AD,
@@ -230,6 +224,10 @@ impl Network for Arc> {
if_disconnected,
);
}
+
+ fn peer_role(&self, who: PeerId, handshake: Vec) -> Option {
+ NetworkService::peer_role(self, who, handshake)
+ }
}
/// We assume one `peer_id` per `authority_id`.
diff --git a/polkadot/node/network/bridge/src/rx/mod.rs b/polkadot/node/network/bridge/src/rx/mod.rs
index 51d248ca2d49..378fbd3a1a2b 100644
--- a/polkadot/node/network/bridge/src/rx/mod.rs
+++ b/polkadot/node/network/bridge/src/rx/mod.rs
@@ -20,10 +20,13 @@ use super::*;
use always_assert::never;
use bytes::Bytes;
-use futures::stream::BoxStream;
use parity_scale_codec::{Decode, DecodeAll};
+use parking_lot::Mutex;
-use sc_network::Event as NetworkEvent;
+use sc_network::{
+ service::traits::{NotificationEvent, ValidationResult},
+ MessageSink, NotificationService,
+};
use sp_consensus::SyncOracle;
use polkadot_node_network_protocol::{
@@ -85,6 +88,9 @@ pub struct NetworkBridgeRx {
shared: Shared,
metrics: Metrics,
peerset_protocol_names: PeerSetProtocolNames,
+ validation_service: Box,
+ collation_service: Box,
+ notification_sinks: Arc>>>,
}
impl NetworkBridgeRx {
@@ -99,8 +105,18 @@ impl NetworkBridgeRx {
sync_oracle: Box,
metrics: Metrics,
peerset_protocol_names: PeerSetProtocolNames,
+ mut notification_services: HashMap>,
+ notification_sinks: Arc>>>,
) -> Self {
let shared = Shared::default();
+
+ let validation_service = notification_services
+ .remove(&PeerSet::Validation)
+ .expect("validation protocol was enabled so `NotificationService` must exist; qed");
+ let collation_service = notification_services
+ .remove(&PeerSet::Collation)
+ .expect("collation protocol was enabled so `NotificationService` must exist; qed");
+
Self {
network_service,
authority_discovery_service,
@@ -108,6 +124,9 @@ impl NetworkBridgeRx {
shared,
metrics,
peerset_protocol_names,
+ validation_service,
+ collation_service,
+ notification_sinks,
}
}
}
@@ -118,423 +137,542 @@ where
Net: Network + Sync,
AD: validator_discovery::AuthorityDiscovery + Clone + Sync,
{
- fn start(mut self, ctx: Context) -> SpawnedSubsystem {
- // The stream of networking events has to be created at initialization, otherwise the
- // networking might open connections before the stream of events has been grabbed.
- let network_stream = self.network_service.event_stream();
-
+ fn start(self, ctx: Context) -> SpawnedSubsystem {
// Swallow error because failure is fatal to the node and we log with more precision
// within `run_network`.
- let future = run_network_in(self, ctx, network_stream)
+ let future = run_network_in(self, ctx)
.map_err(|e| SubsystemError::with_origin("network-bridge", e))
.boxed();
SpawnedSubsystem { name: "network-bridge-rx-subsystem", future }
}
}
-async fn handle_network_messages(
- mut sender: impl overseer::NetworkBridgeRxSenderTrait,
- mut network_service: impl Network,
- network_stream: BoxStream<'static, NetworkEvent>,
- mut authority_discovery_service: AD,
- metrics: Metrics,
- shared: Shared,
- peerset_protocol_names: PeerSetProtocolNames,
-) -> Result<(), Error>
-where
+/// Handle notification event received over the validation protocol.
+async fn handle_validation_message(
+ event: NotificationEvent,
+ network_service: &mut impl Network,
+ sender: &mut impl overseer::NetworkBridgeRxSenderTrait,
+ authority_discovery_service: &mut AD,
+ metrics: &Metrics,
+ shared: &Shared,
+ peerset_protocol_names: &PeerSetProtocolNames,
+ notification_service: &mut Box,
+ notification_sinks: &mut Arc>>>,
+) where
AD: validator_discovery::AuthorityDiscovery + Send,
{
- let mut network_stream = network_stream.fuse();
- loop {
- match network_stream.next().await {
- None => return Err(Error::EventStreamConcluded),
- Some(NetworkEvent::Dht(_)) => {},
- Some(NetworkEvent::NotificationStreamOpened {
- remote: peer,
- protocol,
- role,
- negotiated_fallback,
- received_handshake: _,
- }) => {
- let role = ObservedRole::from(role);
- let (peer_set, version) = {
- let (peer_set, version) =
- match peerset_protocol_names.try_get_protocol(&protocol) {
- None => continue,
- Some(p) => p,
- };
-
- if let Some(fallback) = negotiated_fallback {
- match peerset_protocol_names.try_get_protocol(&fallback) {
- None => {
+ match event {
+ NotificationEvent::ValidateInboundSubstream { result_tx, .. } => {
+ let _ = result_tx.send(ValidationResult::Accept);
+ },
+ NotificationEvent::NotificationStreamOpened {
+ peer,
+ handshake,
+ negotiated_fallback,
+ ..
+ } => {
+ let role = match network_service.peer_role(peer, handshake) {
+ Some(role) => ObservedRole::from(role),
+ None => {
+ gum::debug!(
+ target: LOG_TARGET,
+ ?peer,
+ "Failed to determine peer role for validation protocol",
+ );
+ return
+ },
+ };
+
+ let (peer_set, version) = {
+ let (peer_set, version) =
+ (PeerSet::Validation, PeerSet::Validation.get_main_version());
+
+ if let Some(fallback) = negotiated_fallback {
+ match peerset_protocol_names.try_get_protocol(&fallback) {
+ None => {
+ gum::debug!(
+ target: LOG_TARGET,
+ fallback = &*fallback,
+ ?peer,
+ ?peer_set,
+ "Unknown fallback",
+ );
+
+ return
+ },
+ Some((p2, v2)) => {
+ if p2 != peer_set {
gum::debug!(
target: LOG_TARGET,
fallback = &*fallback,
- ?peer,
- ?peer_set,
- "Unknown fallback",
+ fallback_peerset = ?p2,
+ peerset = ?peer_set,
+ "Fallback mismatched peer-set",
);
- continue
- },
- Some((p2, v2)) => {
- if p2 != peer_set {
- gum::debug!(
- target: LOG_TARGET,
- fallback = &*fallback,
- fallback_peerset = ?p2,
- protocol = &*protocol,
- peerset = ?peer_set,
- "Fallback mismatched peer-set",
- );
-
- continue
- }
-
- (p2, v2)
- },
- }
- } else {
- (peer_set, version)
- }
- };
-
- gum::debug!(
- target: LOG_TARGET,
- action = "PeerConnected",
- peer_set = ?peer_set,
- version = %version,
- peer = ?peer,
- role = ?role
- );
-
- let local_view = {
- let mut shared = shared.0.lock();
- let peer_map = match peer_set {
- PeerSet::Validation => &mut shared.validation_peers,
- PeerSet::Collation => &mut shared.collation_peers,
- };
+ return
+ }
- match peer_map.entry(peer) {
- hash_map::Entry::Occupied(_) => continue,
- hash_map::Entry::Vacant(vacant) => {
- vacant.insert(PeerData { view: View::default(), version });
+ (p2, v2)
},
}
-
- metrics.on_peer_connected(peer_set, version);
- metrics.note_peer_count(peer_set, version, peer_map.len());
-
- shared.local_view.clone().unwrap_or(View::default())
- };
-
- let maybe_authority =
- authority_discovery_service.get_authority_ids_by_peer_id(peer).await;
-
- match peer_set {
- PeerSet::Validation => {
- dispatch_validation_events_to_all(
- vec![
- NetworkBridgeEvent::PeerConnected(
- peer,
- role,
- version,
- maybe_authority,
- ),
- NetworkBridgeEvent::PeerViewChange(peer, View::default()),
- ],
- &mut sender,
- )
- .await;
-
- match ValidationVersion::try_from(version)
- .expect("try_get_protocol has already checked version is known; qed")
- {
- ValidationVersion::V1 => send_message(
- &mut network_service,
- vec![peer],
- PeerSet::Validation,
- version,
- &peerset_protocol_names,
- WireMessage::::ViewUpdate(
- local_view,
- ),
- &metrics,
- ),
- ValidationVersion::VStaging => send_message(
- &mut network_service,
- vec![peer],
- PeerSet::Validation,
- version,
- &peerset_protocol_names,
- WireMessage::::ViewUpdate(
- local_view,
- ),
- &metrics,
- ),
- }
- },
- PeerSet::Collation => {
- dispatch_collation_events_to_all(
- vec![
- NetworkBridgeEvent::PeerConnected(
- peer,
- role,
- version,
- maybe_authority,
- ),
- NetworkBridgeEvent::PeerViewChange(peer, View::default()),
- ],
- &mut sender,
- )
- .await;
-
- match CollationVersion::try_from(version)
- .expect("try_get_protocol has already checked version is known; qed")
- {
- CollationVersion::V1 => send_message(
- &mut network_service,
- vec![peer],
- PeerSet::Collation,
- version,
- &peerset_protocol_names,
- WireMessage::::ViewUpdate(
- local_view,
- ),
- &metrics,
- ),
- CollationVersion::VStaging => send_message(
- &mut network_service,
- vec![peer],
- PeerSet::Collation,
- version,
- &peerset_protocol_names,
- WireMessage::::ViewUpdate(
- local_view,
- ),
- &metrics,
- ),
- }
+ } else {
+ (peer_set, version)
+ }
+ };
+ // store the notification sink to `notification_sinks` so both `NetworkBridgeRx`
+ // and `NetworkBridgeTx` can send messages to the peer.
+ match notification_service.message_sink(&peer) {
+ Some(sink) => {
+ notification_sinks.lock().insert((peer_set, peer), sink);
+ },
+ None => {
+ gum::warn!(
+ target: LOG_TARGET,
+ peer_set = ?peer_set,
+ version = %version,
+ peer = ?peer,
+ role = ?role,
+ "Message sink not available for peer",
+ );
+ return
+ },
+ }
+
+ gum::debug!(
+ target: LOG_TARGET,
+ action = "PeerConnected",
+ peer_set = ?peer_set,
+ version = %version,
+ peer = ?peer,
+ role = ?role
+ );
+
+ let local_view = {
+ let mut shared = shared.0.lock();
+ let peer_map = &mut shared.validation_peers;
+
+ match peer_map.entry(peer) {
+ hash_map::Entry::Occupied(_) => return,
+ hash_map::Entry::Vacant(vacant) => {
+ vacant.insert(PeerData { view: View::default(), version });
},
}
- },
- Some(NetworkEvent::NotificationStreamClosed { remote: peer, protocol }) => {
- let (peer_set, version) = match peerset_protocol_names.try_get_protocol(&protocol) {
- None => continue,
- Some(peer_set) => peer_set,
- };
- gum::debug!(
- target: LOG_TARGET,
- action = "PeerDisconnected",
- peer_set = ?peer_set,
- peer = ?peer
- );
-
- let was_connected = {
- let mut shared = shared.0.lock();
- let peer_map = match peer_set {
- PeerSet::Validation => &mut shared.validation_peers,
- PeerSet::Collation => &mut shared.collation_peers,
- };
+ metrics.on_peer_connected(peer_set, version);
+ metrics.note_peer_count(peer_set, version, peer_map.len());
- let w = peer_map.remove(&peer).is_some();
+ shared.local_view.clone().unwrap_or(View::default())
+ };
- metrics.on_peer_disconnected(peer_set, version);
- metrics.note_peer_count(peer_set, version, peer_map.len());
+ let maybe_authority =
+ authority_discovery_service.get_authority_ids_by_peer_id(peer).await;
- w
- };
+ dispatch_validation_events_to_all(
+ vec![
+ NetworkBridgeEvent::PeerConnected(peer, role, version, maybe_authority),
+ NetworkBridgeEvent::PeerViewChange(peer, View::default()),
+ ],
+ sender,
+ )
+ .await;
- if was_connected && version == peer_set.get_main_version() {
- match peer_set {
- PeerSet::Validation =>
- dispatch_validation_event_to_all(
- NetworkBridgeEvent::PeerDisconnected(peer),
- &mut sender,
- )
- .await,
- PeerSet::Collation =>
- dispatch_collation_event_to_all(
- NetworkBridgeEvent::PeerDisconnected(peer),
- &mut sender,
- )
- .await,
- }
+ match ValidationVersion::try_from(version)
+ .expect("try_get_protocol has already checked version is known; qed")
+ {
+ ValidationVersion::V1 => send_message(
+ vec![peer],
+ PeerSet::Validation,
+ version,
+ WireMessage::::ViewUpdate(local_view),
+ metrics,
+ notification_sinks,
+ ),
+ ValidationVersion::VStaging => send_message(
+ vec![peer],
+ PeerSet::Validation,
+ version,
+ WireMessage::::ViewUpdate(local_view),
+ metrics,
+ notification_sinks,
+ ),
+ }
+ },
+ NotificationEvent::NotificationStreamClosed { peer } => {
+ let (peer_set, version) = (PeerSet::Validation, PeerSet::Validation.get_main_version());
+
+ gum::debug!(
+ target: LOG_TARGET,
+ action = "PeerDisconnected",
+ ?peer_set,
+ ?peer
+ );
+
+ let was_connected = {
+ let mut shared = shared.0.lock();
+ let peer_map = &mut shared.validation_peers;
+
+ let w = peer_map.remove(&peer).is_some();
+
+ metrics.on_peer_disconnected(peer_set, version);
+ metrics.note_peer_count(peer_set, version, peer_map.len());
+
+ w
+ };
+
+ if was_connected && version == peer_set.get_main_version() {
+ dispatch_validation_event_to_all(
+ NetworkBridgeEvent::PeerDisconnected(peer),
+ sender,
+ )
+ .await;
+ }
+ },
+ NotificationEvent::NotificationReceived { peer, notification } => {
+ let expected_versions = {
+ let mut versions = PerPeerSet::>::default();
+ let shared = shared.0.lock();
+
+ if let Some(peer_data) = shared.validation_peers.get(&peer) {
+ versions[PeerSet::Validation] = Some(peer_data.version);
}
- },
- Some(NetworkEvent::NotificationsReceived { remote, messages }) => {
- let expected_versions = {
- let mut versions = PerPeerSet::>::default();
- let shared = shared.0.lock();
- if let Some(peer_data) = shared.validation_peers.get(&remote) {
- versions[PeerSet::Validation] = Some(peer_data.version);
- }
- if let Some(peer_data) = shared.collation_peers.get(&remote) {
- versions[PeerSet::Collation] = Some(peer_data.version);
- }
+ versions
+ };
+
+ gum::trace!(
+ target: LOG_TARGET,
+ action = "PeerMessage",
+ peerset = ?PeerSet::Validation,
+ ?peer,
+ );
+
+ let (events, reports) =
+ if expected_versions[PeerSet::Validation] == Some(ValidationVersion::V1.into()) {
+ handle_peer_messages::(
+ peer,
+ PeerSet::Validation,
+ &mut shared.0.lock().validation_peers,
+ vec![notification.into()],
+ metrics,
+ )
+ } else if expected_versions[PeerSet::Validation] ==
+ Some(ValidationVersion::VStaging.into())
+ {
+ handle_peer_messages::(
+ peer,
+ PeerSet::Validation,
+ &mut shared.0.lock().validation_peers,
+ vec![notification.into()],
+ metrics,
+ )
+ } else {
+ gum::warn!(
+ target: LOG_TARGET,
+ version = ?expected_versions[PeerSet::Validation],
+ "Major logic bug. Peer somehow has unsupported validation protocol version."
+ );
- versions
- };
+ never!("Only versions 1 and 2 are supported; peer set connection checked above; qed");
- // non-decoded, but version-checked validation messages.
- let v_messages: Result, _> = messages
- .iter()
- .filter_map(|(protocol, msg_bytes)| {
- // version doesn't matter because we always receive on the 'correct'
- // protocol name, not the negotiated fallback.
- let (peer_set, _version) =
- peerset_protocol_names.try_get_protocol(protocol)?;
- if peer_set == PeerSet::Validation {
- if expected_versions[PeerSet::Validation].is_none() {
- return Some(Err(UNCONNECTED_PEERSET_COST))
- }
+ // If a peer somehow triggers this, we'll disconnect them
+ // eventually.
+ (Vec::new(), vec![UNCONNECTED_PEERSET_COST])
+ };
- Some(Ok(msg_bytes.clone()))
- } else {
- None
- }
- })
- .collect();
+ for report in reports {
+ network_service.report_peer(peer, report.into());
+ }
- let v_messages = match v_messages {
- Err(rep) => {
- gum::debug!(target: LOG_TARGET, action = "ReportPeer");
- network_service.report_peer(remote, rep.into());
+ dispatch_validation_events_to_all(events, sender).await;
+ },
+ }
+}
- continue
- },
- Ok(v) => v,
- };
+/// Handle notification event received over the collation protocol.
+async fn handle_collation_message(
+ event: NotificationEvent,
+ network_service: &mut impl Network,
+ sender: &mut impl overseer::NetworkBridgeRxSenderTrait,
+ authority_discovery_service: &mut AD,
+ metrics: &Metrics,
+ shared: &Shared,
+ peerset_protocol_names: &PeerSetProtocolNames,
+ notification_service: &mut Box,
+ notification_sinks: &mut Arc>>>,
+) where
+ AD: validator_discovery::AuthorityDiscovery + Send,
+{
+ match event {
+ NotificationEvent::ValidateInboundSubstream { result_tx, .. } => {
+ let _ = result_tx.send(ValidationResult::Accept);
+ },
+ NotificationEvent::NotificationStreamOpened {
+ peer,
+ handshake,
+ negotiated_fallback,
+ ..
+ } => {
+ let role = match network_service.peer_role(peer, handshake) {
+ Some(role) => ObservedRole::from(role),
+ None => {
+ gum::debug!(
+ target: LOG_TARGET,
+ ?peer,
+ "Failed to determine peer role for validation protocol",
+ );
+ return
+ },
+ };
+
+ let (peer_set, version) = {
+ let (peer_set, version) =
+ (PeerSet::Collation, PeerSet::Collation.get_main_version());
+
+ if let Some(fallback) = negotiated_fallback {
+ match peerset_protocol_names.try_get_protocol(&fallback) {
+ None => {
+ gum::debug!(
+ target: LOG_TARGET,
+ fallback = &*fallback,
+ ?peer,
+ ?peer_set,
+ "Unknown fallback",
+ );
+
+ return
+ },
+ Some((p2, v2)) => {
+ if p2 != peer_set {
+ gum::debug!(
+ target: LOG_TARGET,
+ fallback = &*fallback,
+ fallback_peerset = ?p2,
+ peerset = ?peer_set,
+ "Fallback mismatched peer-set",
+ );
- // non-decoded, but version-checked collation messages.
- let c_messages: Result, _> = messages
- .iter()
- .filter_map(|(protocol, msg_bytes)| {
- // version doesn't matter because we always receive on the 'correct'
- // protocol name, not the negotiated fallback.
- let (peer_set, _version) =
- peerset_protocol_names.try_get_protocol(protocol)?;
-
- if peer_set == PeerSet::Collation {
- if expected_versions[PeerSet::Collation].is_none() {
- return Some(Err(UNCONNECTED_PEERSET_COST))
+ return
}
- Some(Ok(msg_bytes.clone()))
- } else {
- None
- }
- })
- .collect();
-
- let c_messages = match c_messages {
- Err(rep) => {
- gum::debug!(target: LOG_TARGET, action = "ReportPeer");
- network_service.report_peer(remote, rep.into());
-
- continue
+ (p2, v2)
+ },
+ }
+ } else {
+ (peer_set, version)
+ }
+ };
+
+ // store the notification sink to `notification_sinks` so both `NetworkBridgeRx`
+ // and `NetworkBridgeTx` can send messages to the peer.
+ match notification_service.message_sink(&peer) {
+ Some(sink) => {
+ notification_sinks.lock().insert((peer_set, peer), sink);
+ },
+ None => {
+ gum::warn!(
+ target: LOG_TARGET,
+ peer_set = ?peer_set,
+ version = %version,
+ peer = ?peer,
+ role = ?role,
+ "Message sink not available for peer",
+ );
+ return
+ },
+ }
+
+ gum::debug!(
+ target: LOG_TARGET,
+ action = "PeerConnected",
+ peer_set = ?peer_set,
+ version = %version,
+ peer = ?peer,
+ role = ?role
+ );
+
+ let local_view = {
+ let mut shared = shared.0.lock();
+ let peer_map = &mut shared.collation_peers;
+
+ match peer_map.entry(peer) {
+ hash_map::Entry::Occupied(_) => return,
+ hash_map::Entry::Vacant(vacant) => {
+ vacant.insert(PeerData { view: View::default(), version });
},
- Ok(v) => v,
- };
-
- if v_messages.is_empty() && c_messages.is_empty() {
- continue
}
- gum::trace!(
- target: LOG_TARGET,
- action = "PeerMessages",
- peer = ?remote,
- num_validation_messages = %v_messages.len(),
- num_collation_messages = %c_messages.len()
- );
-
- if !v_messages.is_empty() {
- let (events, reports) = if expected_versions[PeerSet::Validation] ==
- Some(ValidationVersion::V1.into())
- {
- handle_peer_messages::(
- remote,
- PeerSet::Validation,
- &mut shared.0.lock().validation_peers,
- v_messages,
- &metrics,
- )
- } else if expected_versions[PeerSet::Validation] ==
- Some(ValidationVersion::VStaging.into())
- {
- handle_peer_messages::(
- remote,
- PeerSet::Validation,
- &mut shared.0.lock().validation_peers,
- v_messages,
- &metrics,
- )
- } else {
- gum::warn!(
- target: LOG_TARGET,
- version = ?expected_versions[PeerSet::Validation],
- "Major logic bug. Peer somehow has unsupported validation protocol version."
- );
+ metrics.on_peer_connected(peer_set, version);
+ metrics.note_peer_count(peer_set, version, peer_map.len());
- never!("Only versions 1 and 2 are supported; peer set connection checked above; qed");
+ shared.local_view.clone().unwrap_or(View::default())
+ };
- // If a peer somehow triggers this, we'll disconnect them
- // eventually.
- (Vec::new(), vec![UNCONNECTED_PEERSET_COST])
- };
+ let maybe_authority =
+ authority_discovery_service.get_authority_ids_by_peer_id(peer).await;
- for report in reports {
- network_service.report_peer(remote, report.into());
- }
+ dispatch_collation_events_to_all(
+ vec![
+ NetworkBridgeEvent::PeerConnected(peer, role, version, maybe_authority),
+ NetworkBridgeEvent::PeerViewChange(peer, View::default()),
+ ],
+ sender,
+ )
+ .await;
- dispatch_validation_events_to_all(events, &mut sender).await;
+ match CollationVersion::try_from(version)
+ .expect("try_get_protocol has already checked version is known; qed")
+ {
+ CollationVersion::V1 => send_message(
+ vec![peer],
+ PeerSet::Collation,
+ version,
+ WireMessage::::ViewUpdate(local_view),
+ metrics,
+ notification_sinks,
+ ),
+ CollationVersion::VStaging => send_message(
+ vec![peer],
+ PeerSet::Collation,
+ version,
+ WireMessage::::ViewUpdate(local_view),
+ metrics,
+ notification_sinks,
+ ),
+ }
+ },
+ NotificationEvent::NotificationStreamClosed { peer } => {
+ let (peer_set, version) = (PeerSet::Collation, PeerSet::Collation.get_main_version());
+
+ gum::debug!(
+ target: LOG_TARGET,
+ action = "PeerDisconnected",
+ ?peer_set,
+ ?peer
+ );
+
+ let was_connected = {
+ let mut shared = shared.0.lock();
+ let peer_map = &mut shared.collation_peers;
+
+ let w = peer_map.remove(&peer).is_some();
+
+ metrics.on_peer_disconnected(peer_set, version);
+ metrics.note_peer_count(peer_set, version, peer_map.len());
+
+ w
+ };
+
+ if was_connected && version == peer_set.get_main_version() {
+ dispatch_collation_event_to_all(NetworkBridgeEvent::PeerDisconnected(peer), sender)
+ .await;
+ }
+ },
+ NotificationEvent::NotificationReceived { peer, notification } => {
+ let expected_versions = {
+ let mut versions = PerPeerSet::>::default();
+ let shared = shared.0.lock();
+
+ if let Some(peer_data) = shared.collation_peers.get(&peer) {
+ versions[PeerSet::Collation] = Some(peer_data.version);
}
- if !c_messages.is_empty() {
- let (events, reports) = if expected_versions[PeerSet::Collation] ==
- Some(CollationVersion::V1.into())
- {
- handle_peer_messages::(
- remote,
- PeerSet::Collation,
- &mut shared.0.lock().collation_peers,
- c_messages,
- &metrics,
- )
- } else if expected_versions[PeerSet::Collation] ==
- Some(CollationVersion::VStaging.into())
- {
- handle_peer_messages::(
- remote,
- PeerSet::Collation,
- &mut shared.0.lock().collation_peers,
- c_messages,
- &metrics,
- )
- } else {
- gum::warn!(
- target: LOG_TARGET,
- version = ?expected_versions[PeerSet::Collation],
- "Major logic bug. Peer somehow has unsupported collation protocol version."
- );
+ versions
+ };
+
+ gum::trace!(
+ target: LOG_TARGET,
+ action = "PeerMessage",
+ peerset = ?PeerSet::Collation,
+ ?peer,
+ );
+
+ let (events, reports) =
+ if expected_versions[PeerSet::Collation] == Some(CollationVersion::V1.into()) {
+ handle_peer_messages::(
+ peer,
+ PeerSet::Collation,
+ &mut shared.0.lock().collation_peers,
+ vec![notification.into()],
+ metrics,
+ )
+ } else if expected_versions[PeerSet::Collation] ==
+ Some(CollationVersion::VStaging.into())
+ {
+ handle_peer_messages::(
+ peer,
+ PeerSet::Collation,
+ &mut shared.0.lock().collation_peers,
+ vec![notification.into()],
+ metrics,
+ )
+ } else {
+ gum::warn!(
+ target: LOG_TARGET,
+ version = ?expected_versions[PeerSet::Collation],
+ "Major logic bug. Peer somehow has unsupported collation protocol version."
+ );
- never!("Only versions 1 and 2 are supported; peer set connection checked above; qed");
+ never!("Only versions 1 and 2 are supported; peer set connection checked above; qed");
- // If a peer somehow triggers this, we'll disconnect them
- // eventually.
- (Vec::new(), vec![UNCONNECTED_PEERSET_COST])
- };
+ // If a peer somehow triggers this, we'll disconnect them
+ // eventually.
+ (Vec::new(), vec![UNCONNECTED_PEERSET_COST])
+ };
- for report in reports {
- network_service.report_peer(remote, report.into());
- }
+ for report in reports {
+ network_service.report_peer(peer, report.into());
+ }
- dispatch_collation_events_to_all(events, &mut sender).await;
- }
+ dispatch_collation_events_to_all(events, sender).await;
+ },
+ }
+}
+
+async fn handle_network_messages(
+ mut sender: impl overseer::NetworkBridgeRxSenderTrait,
+ mut network_service: impl Network,
+ mut authority_discovery_service: AD,
+ metrics: Metrics,
+ shared: Shared,
+ peerset_protocol_names: PeerSetProtocolNames,
+ mut validation_service: Box,
+ mut collation_service: Box,
+ mut notification_sinks: Arc>>>,
+) -> Result<(), Error>
+where
+ AD: validator_discovery::AuthorityDiscovery + Send,
+{
+ loop {
+ futures::select! {
+ event = validation_service.next_event().fuse() => match event {
+ Some(event) => handle_validation_message(
+ event,
+ &mut network_service,
+ &mut sender,
+ &mut authority_discovery_service,
+ &metrics,
+ &shared,
+ &peerset_protocol_names,
+ &mut validation_service,
+ &mut notification_sinks,
+ ).await,
+ None => return Err(Error::EventStreamConcluded),
},
+ event = collation_service.next_event().fuse() => match event {
+ Some(event) => handle_collation_message(
+ event,
+ &mut network_service,
+ &mut sender,
+ &mut authority_discovery_service,
+ &metrics,
+ &shared,
+ &peerset_protocol_names,
+ &mut collation_service,
+ &mut notification_sinks,
+ ).await,
+ None => return Err(Error::EventStreamConcluded),
+ }
}
}
}
@@ -560,17 +698,15 @@ where
}
#[overseer::contextbounds(NetworkBridgeRx, prefix = self::overseer)]
-async fn run_incoming_orchestra_signals(
+async fn run_incoming_orchestra_signals(
mut ctx: Context,
- mut network_service: N,
mut authority_discovery_service: AD,
shared: Shared,
sync_oracle: Box,
metrics: Metrics,
- peerset_protocol_names: PeerSetProtocolNames,
+ notification_sinks: Arc>>>,
) -> Result<(), Error>
where
- N: Network,
AD: validator_discovery::AuthorityDiscovery + Clone,
{
// This is kept sorted, descending, by block number.
@@ -662,13 +798,12 @@ where
mode = Mode::Active;
update_our_view(
- &mut network_service,
&mut ctx,
&live_heads,
&shared,
finalized_number,
&metrics,
- &peerset_protocol_names,
+ ¬ification_sinks,
);
}
}
@@ -702,7 +837,6 @@ where
async fn run_network_in(
bridge: NetworkBridgeRx,
mut ctx: Context,
- network_stream: BoxStream<'static, NetworkEvent>,
) -> Result<(), Error>
where
N: Network,
@@ -715,16 +849,21 @@ where
sync_oracle,
shared,
peerset_protocol_names,
+ validation_service,
+ collation_service,
+ notification_sinks,
} = bridge;
let (task, network_event_handler) = handle_network_messages(
ctx.sender().clone(),
network_service.clone(),
- network_stream,
authority_discovery_service.clone(),
metrics.clone(),
shared.clone(),
peerset_protocol_names.clone(),
+ validation_service,
+ collation_service,
+ notification_sinks.clone(),
)
.remote_handle();
@@ -733,12 +872,11 @@ where
let orchestra_signal_handler = run_incoming_orchestra_signals(
ctx,
- network_service,
authority_discovery_service,
shared,
sync_oracle,
metrics,
- peerset_protocol_names,
+ notification_sinks,
);
futures::pin_mut!(orchestra_signal_handler);
@@ -758,17 +896,14 @@ fn construct_view(
}
#[overseer::contextbounds(NetworkBridgeRx, prefix = self::overseer)]
-fn update_our_view(
- net: &mut Net,
+fn update_our_view(
ctx: &mut Context,
live_heads: &[ActivatedLeaf],
shared: &Shared,
finalized_number: BlockNumber,
metrics: &Metrics,
- peerset_protocol_names: &PeerSetProtocolNames,
-) where
- Net: Network,
-{
+ notification_sinks: &Arc>>>,
+) {
let new_view = construct_view(live_heads.iter().map(|v| v.hash), finalized_number);
let (validation_peers, collation_peers) = {
@@ -817,35 +952,31 @@ fn update_our_view(
filter_by_version(&collation_peers, ValidationVersion::VStaging.into());
send_validation_message_v1(
- net,
v1_validation_peers,
- peerset_protocol_names,
WireMessage::ViewUpdate(new_view.clone()),
metrics,
+ notification_sinks,
);
send_collation_message_v1(
- net,
v1_collation_peers,
- peerset_protocol_names,
WireMessage::ViewUpdate(new_view.clone()),
metrics,
+ notification_sinks,
);
send_validation_message_vstaging(
- net,
vstaging_validation_peers,
- peerset_protocol_names,
WireMessage::ViewUpdate(new_view.clone()),
metrics,
+ notification_sinks,
);
send_collation_message_vstaging(
- net,
vstaging_collation_peers,
- peerset_protocol_names,
WireMessage::ViewUpdate(new_view),
metrics,
+ notification_sinks,
);
let our_view = OurView::new(
@@ -918,74 +1049,66 @@ fn handle_peer_messages>(
}
fn send_validation_message_v1(
- net: &mut impl Network,
peers: Vec,
- peerset_protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Validation,
ValidationVersion::V1.into(),
- peerset_protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_collation_message_v1(
- net: &mut impl Network,
peers: Vec,
- peerset_protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Collation,
CollationVersion::V1.into(),
- peerset_protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_validation_message_vstaging(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Validation,
ValidationVersion::VStaging.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_collation_message_vstaging(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Collation,
CollationVersion::VStaging.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
diff --git a/polkadot/node/network/bridge/src/rx/tests.rs b/polkadot/node/network/bridge/src/rx/tests.rs
index 88a4b247fdc6..3161801aa2c2 100644
--- a/polkadot/node/network/bridge/src/rx/tests.rs
+++ b/polkadot/node/network/bridge/src/rx/tests.rs
@@ -15,7 +15,7 @@
// along with Polkadot. If not, see .
use super::*;
-use futures::{channel::oneshot, executor, stream::BoxStream};
+use futures::{channel::oneshot, executor};
use polkadot_node_network_protocol::{self as net_protocol, OurView};
use polkadot_node_subsystem::{messages::NetworkBridgeEvent, ActivatedLeaf};
@@ -28,7 +28,11 @@ use std::{
task::Poll,
};
-use sc_network::{Event as NetworkEvent, IfDisconnected, ProtocolName, ReputationChange};
+use sc_network::{
+ service::traits::{Direction, MessageSink, NotificationService},
+ IfDisconnected, Multiaddr, ObservedRole as SubstrateObservedRole, ProtocolName,
+ ReputationChange, Roles,
+};
use polkadot_node_network_protocol::{
peer_set::PeerSetProtocolNames,
@@ -49,7 +53,6 @@ use polkadot_node_subsystem_test_helpers::{
use polkadot_node_subsystem_util::metered;
use polkadot_primitives::{AuthorityDiscoveryId, CandidateHash, Hash};
-use sc_network::Multiaddr;
use sp_keyring::Sr25519Keyring;
use crate::{network::Network, validator_discovery::AuthorityDiscovery};
@@ -64,10 +67,9 @@ pub enum NetworkAction {
WriteNotification(PeerId, PeerSet, Vec),
}
-// The subsystem's view of the network - only supports a single call to `event_stream`.
+// The subsystem's view of the network.
#[derive(Clone)]
struct TestNetwork {
- net_events: Arc>>>,
action_tx: Arc>>,
protocol_names: Arc,
}
@@ -79,37 +81,42 @@ struct TestAuthorityDiscovery;
// of `NetworkAction`s.
struct TestNetworkHandle {
action_rx: metered::UnboundedMeteredReceiver,
- net_tx: SingleItemSink,
- protocol_names: PeerSetProtocolNames,
+ validation_tx: SingleItemSink,
+ collation_tx: SingleItemSink,
}
fn new_test_network(
protocol_names: PeerSetProtocolNames,
-) -> (TestNetwork, TestNetworkHandle, TestAuthorityDiscovery) {
- let (net_tx, net_rx) = polkadot_node_subsystem_test_helpers::single_item_sink();
+) -> (
+ TestNetwork,
+ TestNetworkHandle,
+ TestAuthorityDiscovery,
+ Box,
+ Box,
+) {
let (action_tx, action_rx) = metered::unbounded();
+ let (validation_tx, validation_rx) = polkadot_node_subsystem_test_helpers::single_item_sink();
+ let (collation_tx, collation_rx) = polkadot_node_subsystem_test_helpers::single_item_sink();
+ let action_tx = Arc::new(Mutex::new(action_tx));
(
TestNetwork {
- net_events: Arc::new(Mutex::new(Some(net_rx))),
- action_tx: Arc::new(Mutex::new(action_tx)),
+ action_tx: action_tx.clone(),
protocol_names: Arc::new(protocol_names.clone()),
},
- TestNetworkHandle { action_rx, net_tx, protocol_names },
+ TestNetworkHandle { action_rx, validation_tx, collation_tx },
TestAuthorityDiscovery,
+ Box::new(TestNotificationService::new(
+ PeerSet::Validation,
+ action_tx.clone(),
+ validation_rx,
+ )),
+ Box::new(TestNotificationService::new(PeerSet::Collation, action_tx, collation_rx)),
)
}
#[async_trait]
impl Network for TestNetwork {
- fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
- self.net_events
- .lock()
- .take()
- .expect("Subsystem made more than one call to `event_stream`")
- .boxed()
- }
-
async fn set_reserved_peers(
&mut self,
_protocol: ProtocolName,
@@ -143,7 +150,8 @@ impl Network for TestNetwork {
}
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
- let (peer_set, _) = self.protocol_names.try_get_protocol(&protocol).unwrap();
+ let (peer_set, version) = self.protocol_names.try_get_protocol(&protocol).unwrap();
+ assert_eq!(version, peer_set.get_main_version());
self.action_tx
.lock()
@@ -151,13 +159,10 @@ impl Network for TestNetwork {
.unwrap();
}
- fn write_notification(&self, who: PeerId, protocol: ProtocolName, message: Vec) {
- let (peer_set, _) = self.protocol_names.try_get_protocol(&protocol).unwrap();
-
- self.action_tx
- .lock()
- .unbounded_send(NetworkAction::WriteNotification(who, peer_set, message))
- .unwrap();
+ fn peer_role(&self, _peer_id: PeerId, handshake: Vec) -> Option {
+ Roles::decode_all(&mut &handshake[..])
+ .ok()
+ .and_then(|role| Some(SubstrateObservedRole::from(role)))
}
}
@@ -194,42 +199,93 @@ impl TestNetworkHandle {
v
}
+ // async fn connect_peer(
+ // &mut self,
+ // peer: PeerId,
+ // protocol_version: ValidationVersion,
+ // peer_set: PeerSet,
+ // role: ObservedRole,
+ // ) {
+ // // let protocol_version = ProtocolVersion::from(protocol_version);
+ // // self.send_network_event(NetworkEvent::NotificationStreamOpened {
+ // // remote: peer,
+ // // protocol: self.protocol_names.get_name(peer_set, protocol_version),
+ // // negotiated_fallback: None,
+ // // role: role.into(),
+ // // received_handshake: vec![],
+ // // })
+ // // .await;
+ // }
+
async fn connect_peer(
&mut self,
peer: PeerId,
- protocol_version: ValidationVersion,
+ _protocol_version: ValidationVersion,
peer_set: PeerSet,
role: ObservedRole,
) {
- let protocol_version = ProtocolVersion::from(protocol_version);
- self.send_network_event(NetworkEvent::NotificationStreamOpened {
- remote: peer,
- protocol: self.protocol_names.get_name(peer_set, protocol_version),
- negotiated_fallback: None,
- role: role.into(),
- received_handshake: vec![],
- })
- .await;
+ fn observed_role_to_handshake(role: &ObservedRole) -> Vec {
+ match role {
+ &ObservedRole::Light => Roles::LIGHT.encode(),
+ &ObservedRole::Authority => Roles::AUTHORITY.encode(),
+ &ObservedRole::Full => Roles::FULL.encode(),
+ }
+ }
+
+ match peer_set {
+ PeerSet::Validation => {
+ self.validation_tx
+ .send(NotificationEvent::NotificationStreamOpened {
+ peer,
+ direction: Direction::Inbound,
+ handshake: observed_role_to_handshake(&role),
+ negotiated_fallback: None,
+ })
+ .await
+ .expect("subsystem concluded early");
+ },
+ PeerSet::Collation => {
+ self.collation_tx
+ .send(NotificationEvent::NotificationStreamOpened {
+ peer,
+ direction: Direction::Inbound,
+ handshake: observed_role_to_handshake(&role),
+ negotiated_fallback: None,
+ })
+ .await
+ .expect("subsystem concluded early");
+ },
+ }
}
async fn disconnect_peer(&mut self, peer: PeerId, peer_set: PeerSet) {
- self.send_network_event(NetworkEvent::NotificationStreamClosed {
- remote: peer,
- protocol: self.protocol_names.get_main_name(peer_set),
- })
- .await;
+ match peer_set {
+ PeerSet::Validation => self
+ .validation_tx
+ .send(NotificationEvent::NotificationStreamClosed { peer })
+ .await
+ .expect("subsystem concluded early"),
+ PeerSet::Collation => self
+ .collation_tx
+ .send(NotificationEvent::NotificationStreamClosed { peer })
+ .await
+ .expect("subsystem concluded early"),
+ }
}
async fn peer_message(&mut self, peer: PeerId, peer_set: PeerSet, message: Vec) {
- self.send_network_event(NetworkEvent::NotificationsReceived {
- remote: peer,
- messages: vec![(self.protocol_names.get_main_name(peer_set), message.into())],
- })
- .await;
- }
-
- async fn send_network_event(&mut self, event: NetworkEvent) {
- self.net_tx.send(event).await.expect("subsystem concluded early");
+ match peer_set {
+ PeerSet::Validation => self
+ .validation_tx
+ .send(NotificationEvent::NotificationReceived { peer, notification: message })
+ .await
+ .expect("subsystem concluded early"),
+ PeerSet::Collation => self
+ .collation_tx
+ .send(NotificationEvent::NotificationReceived { peer, notification: message })
+ .await
+ .expect("subsystem concluded early"),
+ }
}
}
@@ -240,6 +296,119 @@ fn assert_network_actions_contains(actions: &[NetworkAction], action: &NetworkAc
}
}
+struct TestNotificationService {
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+ rx: SingleItemStream,
+}
+
+impl std::fmt::Debug for TestNotificationService {
+ fn fmt(&self, _: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+ Ok(())
+ }
+}
+
+impl TestNotificationService {
+ pub fn new(
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+ rx: SingleItemStream,
+ ) -> Self {
+ Self { peer_set, action_tx, rx }
+ }
+}
+
+struct TestMessageSink {
+ peer: PeerId,
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+}
+
+impl TestMessageSink {
+ fn new(
+ peer: PeerId,
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+ ) -> TestMessageSink {
+ Self { peer, peer_set, action_tx }
+ }
+}
+
+#[async_trait::async_trait]
+impl MessageSink for TestMessageSink {
+ fn send_sync_notification(&self, notification: Vec) {
+ self.action_tx
+ .lock()
+ .unbounded_send(NetworkAction::WriteNotification(
+ self.peer,
+ self.peer_set,
+ notification,
+ ))
+ .unwrap();
+ }
+
+ async fn send_async_notification(
+ &self,
+ _notification: Vec,
+ ) -> Result<(), sc_network::error::Error> {
+ unimplemented!();
+ }
+}
+
+// TODO: how to implement this?
+#[async_trait::async_trait]
+impl NotificationService for TestNotificationService {
+ /// Instruct `Notifications` to open a new substream for `peer`.
+ async fn open_substream(&mut self, _peer: PeerId) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Instruct `Notifications` to close substream for `peer`.
+ async fn close_substream(&mut self, _peer: PeerId) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Send synchronous `notification` to `peer`.
+ fn send_sync_notification(&self, _peer: &PeerId, _notification: Vec) {
+ unimplemented!();
+ }
+
+ /// Send asynchronous `notification` to `peer`, allowing sender to exercise backpressure.
+ async fn send_async_notification(
+ &self,
+ _peer: &PeerId,
+ _notification: Vec,
+ ) -> Result<(), sc_network::error::Error> {
+ unimplemented!();
+ }
+
+ /// Set handshake for the notification protocol replacing the old handshake.
+ async fn set_handshake(&mut self, _handshake: Vec) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Get next event from the `Notifications` event stream.
+ async fn next_event(&mut self) -> Option {
+ self.rx.next().await
+ }
+
+ // Clone [`NotificationService`]
+ fn clone(&mut self) -> Result, ()> {
+ unimplemented!();
+ }
+
+ /// Get protocol name.
+ fn protocol(&self) -> &ProtocolName {
+ unimplemented!();
+ }
+
+ /// Get notification sink of the peer.
+ fn message_sink(&self, peer: &PeerId) -> Option> {
+ // TODO: verify that peer exists?
+ Some(Box::new(TestMessageSink::new(*peer, self.peer_set, self.action_tx.clone())))
+ }
+}
+
#[derive(Clone)]
struct TestSyncOracle {
is_major_syncing: Arc,
@@ -334,11 +503,14 @@ fn test_harness>(
let fork_id = None;
let peerset_protocol_names = PeerSetProtocolNames::new(genesis_hash, fork_id);
+ println!("protocol names: {peerset_protocol_names:#?}");
+
let pool = sp_core::testing::TaskExecutor::new();
- let (mut network, network_handle, discovery) = new_test_network(peerset_protocol_names.clone());
+ let (network, network_handle, discovery, validation_service, collation_service) =
+ new_test_network(peerset_protocol_names.clone());
let (context, virtual_overseer) =
polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
- let network_stream = network.event_stream();
+ let notification_sinks = Arc::new(Mutex::new(HashMap::new()));
let shared = Shared::default();
let bridge = NetworkBridgeRx {
@@ -348,9 +520,12 @@ fn test_harness>(
sync_oracle,
shared: shared.clone(),
peerset_protocol_names,
+ validation_service,
+ collation_service,
+ notification_sinks,
};
- let network_bridge = run_network_in(bridge, context, network_stream)
+ let network_bridge = run_network_in(bridge, context)
.map_err(|_| panic!("subsystem execution failed"))
.map(|_| ());
@@ -972,8 +1147,6 @@ fn relays_collation_protocol_messages() {
.await;
}
- // peer A gets reported for sending a collation message.
-
let collator_protocol_message = protocol_v1::CollatorProtocolMessage::Declare(
Sr25519Keyring::Alice.public().into(),
Default::default(),
@@ -983,19 +1156,23 @@ fn relays_collation_protocol_messages() {
let message_v1 =
protocol_v1::CollationProtocol::CollatorProtocol(collator_protocol_message.clone());
- network_handle
- .peer_message(
- peer_a,
- PeerSet::Collation,
- WireMessage::ProtocolMessage(message_v1.clone()).encode(),
- )
- .await;
-
- let actions = network_handle.next_network_actions(3).await;
- assert_network_actions_contains(
- &actions,
- &NetworkAction::ReputationChange(peer_a, UNCONNECTED_PEERSET_COST.into()),
- );
+ // peer A gets reported for sending a collation message.
+ // NOTE: this is not possible since peer A cannot send
+ // a collation message if it has not opened a collation protocol
+
+ // network_handle
+ // .peer_message(
+ // peer_a,
+ // PeerSet::Collation,
+ // WireMessage::ProtocolMessage(message_v1.clone()).encode(),
+ // )
+ // .await;
+
+ // let actions = network_handle.next_network_actions(3).await;
+ // assert_network_actions_contains(
+ // &actions,
+ // &NetworkAction::ReputationChange(peer_a, UNCONNECTED_PEERSET_COST.into()),
+ // );
// peer B has the message relayed.
@@ -1252,7 +1429,7 @@ fn our_view_updates_decreasing_order_and_limited_to_max() {
fn network_protocol_versioning_view_update() {
let (oracle, handle) = make_sync_oracle(false);
test_harness(Box::new(oracle), |test_harness| async move {
- let TestHarness { mut network_handle, mut virtual_overseer, .. } = test_harness;
+ let TestHarness { mut network_handle, mut virtual_overseer, shared } = test_harness;
let peer_ids: Vec<_> = (0..4).map(|_| PeerId::random()).collect();
let peers = [
@@ -1276,12 +1453,22 @@ fn network_protocol_versioning_view_update() {
handle.await_mode_switch().await;
+ let mut total_validation_peers = 0;
+ let mut total_collation_peers = 0;
+
for &(peer_id, peer_set, version) in &peers {
network_handle
.connect_peer(peer_id, version, peer_set, ObservedRole::Full)
.await;
+
+ match peer_set {
+ PeerSet::Validation => total_validation_peers += 1,
+ PeerSet::Collation => total_collation_peers += 1,
+ }
}
+ await_peer_connections(&shared, total_validation_peers, total_collation_peers).await;
+
let view = view![head];
let actions = network_handle.next_network_actions(4).await;
@@ -1304,11 +1491,13 @@ fn network_protocol_versioning_view_update() {
});
}
+// TODO: discuss with parachains engineering
#[test]
+#[cfg(feature = "network-protocol-staging")]
fn network_protocol_versioning_subsystem_msg() {
let (oracle, _handle) = make_sync_oracle(false);
test_harness(Box::new(oracle), |test_harness| async move {
- let TestHarness { mut network_handle, mut virtual_overseer, .. } = test_harness;
+ let TestHarness { mut network_handle, mut virtual_overseer, shared } = test_harness;
let peer = PeerId::random();
@@ -1320,6 +1509,7 @@ fn network_protocol_versioning_subsystem_msg() {
ObservedRole::Full,
)
.await;
+ await_peer_connections(&shared, 1, 0).await;
// bridge will inform about all connected peers.
{
diff --git a/polkadot/node/network/bridge/src/tx/mod.rs b/polkadot/node/network/bridge/src/tx/mod.rs
index 1b386ce1239b..2cbe7883d38a 100644
--- a/polkadot/node/network/bridge/src/tx/mod.rs
+++ b/polkadot/node/network/bridge/src/tx/mod.rs
@@ -33,7 +33,7 @@ use polkadot_node_subsystem::{
///
/// To be passed to [`FullNetworkConfiguration::add_notification_protocol`]().
pub use polkadot_node_network_protocol::peer_set::{peer_sets_info, IsAuthority};
-use sc_network::ReputationChange;
+use sc_network::{MessageSink, ReputationChange};
use crate::validator_discovery;
@@ -58,6 +58,7 @@ pub struct NetworkBridgeTx {
metrics: Metrics,
req_protocol_names: ReqProtocolNames,
peerset_protocol_names: PeerSetProtocolNames,
+ notification_sinks: Arc>>>,
}
impl NetworkBridgeTx {
@@ -72,6 +73,7 @@ impl NetworkBridgeTx {
metrics: Metrics,
req_protocol_names: ReqProtocolNames,
peerset_protocol_names: PeerSetProtocolNames,
+ notification_sinks: Arc>>>,
) -> Self {
Self {
network_service,
@@ -79,6 +81,7 @@ impl NetworkBridgeTx {
metrics,
req_protocol_names,
peerset_protocol_names,
+ notification_sinks,
}
}
}
@@ -105,6 +108,7 @@ async fn handle_subsystem_messages(
metrics: Metrics,
req_protocol_names: ReqProtocolNames,
peerset_protocol_names: PeerSetProtocolNames,
+ notification_sinks: Arc>>>,
) -> Result<(), Error>
where
N: Network,
@@ -128,6 +132,7 @@ where
&metrics,
&req_protocol_names,
&peerset_protocol_names,
+ ¬ification_sinks,
)
.await;
},
@@ -138,13 +143,14 @@ where
#[overseer::contextbounds(NetworkBridgeTx, prefix = self::overseer)]
async fn handle_incoming_subsystem_communication(
_ctx: &mut Context,
- mut network_service: N,
+ network_service: N,
validator_discovery: &mut validator_discovery::Service,
mut authority_discovery_service: AD,
msg: NetworkBridgeTxMessage,
metrics: &Metrics,
req_protocol_names: &ReqProtocolNames,
peerset_protocol_names: &PeerSetProtocolNames,
+ notification_sinks: &Arc>>>,
) -> (N, AD)
where
N: Network,
@@ -191,18 +197,16 @@ where
match msg {
Versioned::V1(msg) => send_validation_message_v1(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
Versioned::VStaging(msg) => send_validation_message_vstaging(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
}
},
@@ -216,18 +220,16 @@ where
for (peers, msg) in msgs {
match msg {
Versioned::V1(msg) => send_validation_message_v1(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
Versioned::VStaging(msg) => send_validation_message_vstaging(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
}
}
@@ -241,18 +243,16 @@ where
match msg {
Versioned::V1(msg) => send_collation_message_v1(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
Versioned::VStaging(msg) => send_collation_message_vstaging(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
}
},
@@ -266,18 +266,16 @@ where
for (peers, msg) in msgs {
match msg {
Versioned::V1(msg) => send_collation_message_v1(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
Versioned::VStaging(msg) => send_collation_message_vstaging(
- &mut network_service,
peers,
- peerset_protocol_names,
WireMessage::ProtocolMessage(msg),
&metrics,
+ notification_sinks,
),
}
}
@@ -359,6 +357,7 @@ where
metrics,
req_protocol_names,
peerset_protocol_names,
+ notification_sinks,
} = bridge;
handle_subsystem_messages(
@@ -368,6 +367,7 @@ where
metrics,
req_protocol_names,
peerset_protocol_names,
+ notification_sinks,
)
.await?;
@@ -375,73 +375,65 @@ where
}
fn send_validation_message_v1(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Validation,
ValidationVersion::V1.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_collation_message_v1(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Collation,
CollationVersion::V1.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_validation_message_vstaging(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Validation,
ValidationVersion::VStaging.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
fn send_collation_message_vstaging(
- net: &mut impl Network,
peers: Vec,
- protocol_names: &PeerSetProtocolNames,
message: WireMessage,
metrics: &Metrics,
+ notification_sinks: &Arc>>>,
) {
send_message(
- net,
peers,
PeerSet::Collation,
CollationVersion::VStaging.into(),
- protocol_names,
message,
metrics,
+ notification_sinks,
);
}
diff --git a/polkadot/node/network/bridge/src/tx/tests.rs b/polkadot/node/network/bridge/src/tx/tests.rs
index 21cd134c54f2..a19df51dc575 100644
--- a/polkadot/node/network/bridge/src/tx/tests.rs
+++ b/polkadot/node/network/bridge/src/tx/tests.rs
@@ -15,15 +15,18 @@
// along with Polkadot. If not, see .
use super::*;
-use futures::{executor, stream::BoxStream};
+use futures::executor;
use polkadot_node_subsystem_util::TimeoutExt;
use async_trait::async_trait;
use parking_lot::Mutex;
use std::collections::HashSet;
-use sc_network::{Event as NetworkEvent, IfDisconnected, ProtocolName, ReputationChange};
+use sc_network::{
+ IfDisconnected, ObservedRole as SubstrateObservedRole, ProtocolName, ReputationChange, Roles,
+};
+use parity_scale_codec::DecodeAll;
use polkadot_node_network_protocol::{
peer_set::PeerSetProtocolNames,
request_response::{outgoing::Requests, ReqProtocolNames},
@@ -51,10 +54,9 @@ pub enum NetworkAction {
WriteNotification(PeerId, PeerSet, Vec),
}
-// The subsystem's view of the network - only supports a single call to `event_stream`.
+// The subsystem's view of the network.
#[derive(Clone)]
struct TestNetwork {
- net_events: Arc>>>,
action_tx: Arc>>,
peerset_protocol_names: Arc,
}
@@ -66,37 +68,98 @@ struct TestAuthorityDiscovery;
// of `NetworkAction`s.
struct TestNetworkHandle {
action_rx: metered::UnboundedMeteredReceiver,
- net_tx: metered::MeteredSender,
- peerset_protocol_names: PeerSetProtocolNames,
+ _peerset_protocol_names: PeerSetProtocolNames,
+ notification_sinks: Arc>>>,
+ action_tx: Arc>>,
+}
+
+struct TestMessageSink {
+ peer: PeerId,
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+}
+
+impl TestMessageSink {
+ fn new(
+ peer: PeerId,
+ peer_set: PeerSet,
+ action_tx: Arc>>,
+ ) -> TestMessageSink {
+ Self { peer, peer_set, action_tx }
+ }
+}
+
+#[async_trait::async_trait]
+impl MessageSink for TestMessageSink {
+ fn send_sync_notification(&self, notification: Vec) {
+ self.action_tx
+ .lock()
+ .unbounded_send(NetworkAction::WriteNotification(
+ self.peer,
+ self.peer_set,
+ notification,
+ ))
+ .unwrap();
+ }
+
+ async fn send_async_notification(
+ &self,
+ _notification: Vec,
+ ) -> Result<(), sc_network::error::Error> {
+ unimplemented!();
+ }
}
fn new_test_network(
peerset_protocol_names: PeerSetProtocolNames,
-) -> (TestNetwork, TestNetworkHandle, TestAuthorityDiscovery) {
- let (net_tx, net_rx) = metered::channel(10);
+) -> (
+ TestNetwork,
+ TestNetworkHandle,
+ TestAuthorityDiscovery,
+ Arc>>>,
+) {
let (action_tx, action_rx) = metered::unbounded();
+ let notification_sinks = Arc::new(Mutex::new(HashMap::new()));
+ let action_tx = Arc::new(Mutex::new(action_tx));
(
TestNetwork {
- net_events: Arc::new(Mutex::new(Some(net_rx))),
- action_tx: Arc::new(Mutex::new(action_tx)),
+ action_tx: action_tx.clone(),
peerset_protocol_names: Arc::new(peerset_protocol_names.clone()),
},
- TestNetworkHandle { action_rx, net_tx, peerset_protocol_names },
+ TestNetworkHandle {
+ action_rx,
+ _peerset_protocol_names: peerset_protocol_names,
+ action_tx,
+ notification_sinks: notification_sinks.clone(),
+ },
TestAuthorityDiscovery,
+ notification_sinks,
)
}
-#[async_trait]
-impl Network for TestNetwork {
- fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
- self.net_events
- .lock()
- .take()
- .expect("Subsystem made more than one call to `event_stream`")
- .boxed()
+// wait until all needed validation and collation peers have connected.
+async fn await_peer_connections(
+ shared: &Shared,
+ num_validation_peers: usize,
+ num_collation_peers: usize,
+) {
+ loop {
+ {
+ let shared = shared.0.lock();
+ if shared.validation_peers.len() == num_validation_peers &&
+ shared.collation_peers.len() == num_collation_peers
+ {
+ break
+ }
+ }
+
+ futures_timer::Delay::new(std::time::Duration::from_millis(100)).await;
}
+}
+#[async_trait]
+impl Network for TestNetwork {
async fn set_reserved_peers(
&mut self,
_protocol: ProtocolName,
@@ -130,7 +193,8 @@ impl Network for TestNetwork {
}
fn disconnect_peer(&self, who: PeerId, protocol: ProtocolName) {
- let (peer_set, _) = self.peerset_protocol_names.try_get_protocol(&protocol).unwrap();
+ let (peer_set, version) = self.peerset_protocol_names.try_get_protocol(&protocol).unwrap();
+ assert_eq!(version, peer_set.get_main_version());
self.action_tx
.lock()
@@ -138,13 +202,10 @@ impl Network for TestNetwork {
.unwrap();
}
- fn write_notification(&self, who: PeerId, protocol: ProtocolName, message: Vec) {
- let (peer_set, _) = self.peerset_protocol_names.try_get_protocol(&protocol).unwrap();
-
- self.action_tx
- .lock()
- .unbounded_send(NetworkAction::WriteNotification(who, peer_set, message))
- .unwrap();
+ fn peer_role(&self, _peer_id: PeerId, handshake: Vec) -> Option {
+ Roles::decode_all(&mut &handshake[..])
+ .ok()
+ .and_then(|role| Some(SubstrateObservedRole::from(role)))
}
}
@@ -174,23 +235,14 @@ impl TestNetworkHandle {
async fn connect_peer(
&mut self,
peer: PeerId,
- protocol_version: ValidationVersion,
+ _protocol_version: ValidationVersion,
peer_set: PeerSet,
- role: ObservedRole,
+ _role: ObservedRole,
) {
- let protocol_version = ProtocolVersion::from(protocol_version);
- self.send_network_event(NetworkEvent::NotificationStreamOpened {
- remote: peer,
- protocol: self.peerset_protocol_names.get_name(peer_set, protocol_version),
- negotiated_fallback: None,
- role: role.into(),
- received_handshake: vec![],
- })
- .await;
- }
-
- async fn send_network_event(&mut self, event: NetworkEvent) {
- self.net_tx.send(event).await.expect("subsystem concluded early");
+ self.notification_sinks.lock().insert(
+ (peer_set, peer),
+ Box::new(TestMessageSink::new(peer, peer_set, self.action_tx.clone())),
+ );
}
}
@@ -208,7 +260,8 @@ fn test_harness>(test: impl FnOnce(TestHarne
let peerset_protocol_names = PeerSetProtocolNames::new(genesis_hash, fork_id);
let pool = sp_core::testing::TaskExecutor::new();
- let (network, network_handle, discovery) = new_test_network(peerset_protocol_names.clone());
+ let (network, network_handle, discovery, network_notification_sinks) =
+ new_test_network(peerset_protocol_names.clone());
let (context, virtual_overseer) =
polkadot_node_subsystem_test_helpers::make_subsystem_context(pool);
@@ -219,6 +272,7 @@ fn test_harness>(test: impl FnOnce(TestHarne
Metrics(None),
req_protocol_names,
peerset_protocol_names,
+ network_notification_sinks,
);
let network_bridge_out_fut = run_network_out(bridge_out, context)
@@ -334,7 +388,9 @@ fn send_messages_to_peers() {
});
}
+// TODO: discuss with parachains engineering
#[test]
+#[cfg(feature = "network-protocol-staging")]
fn network_protocol_versioning_send() {
test_harness(|test_harness| async move {
let TestHarness { mut network_handle, mut virtual_overseer } = test_harness;
diff --git a/polkadot/node/network/bridge/src/validator_discovery.rs b/polkadot/node/network/bridge/src/validator_discovery.rs
index 86e861fbc5b5..b11af8a8a089 100644
--- a/polkadot/node/network/bridge/src/validator_discovery.rs
+++ b/polkadot/node/network/bridge/src/validator_discovery.rs
@@ -169,13 +169,12 @@ mod tests {
use crate::network::Network;
use async_trait::async_trait;
- use futures::stream::BoxStream;
use polkadot_node_network_protocol::{
request_response::{outgoing::Requests, ReqProtocolNames},
PeerId,
};
use polkadot_primitives::Hash;
- use sc_network::{Event as NetworkEvent, IfDisconnected, ProtocolName, ReputationChange};
+ use sc_network::{IfDisconnected, ProtocolName, ReputationChange};
use sp_keyring::Sr25519Keyring;
use std::collections::{HashMap, HashSet};
@@ -224,10 +223,6 @@ mod tests {
#[async_trait]
impl Network for TestNetwork {
- fn event_stream(&mut self) -> BoxStream<'static, NetworkEvent> {
- panic!()
- }
-
async fn set_reserved_peers(
&mut self,
_protocol: ProtocolName,
@@ -263,7 +258,11 @@ mod tests {
panic!()
}
- fn write_notification(&self, _: PeerId, _: ProtocolName, _: Vec) {
+ fn peer_role(
+ &self,
+ _peer_id: PeerId,
+ _handshake: Vec,
+ ) -> Option {
panic!()
}
}
diff --git a/polkadot/node/network/protocol/src/peer_set.rs b/polkadot/node/network/protocol/src/peer_set.rs
index c2163783c2ce..9db5d4771d4b 100644
--- a/polkadot/node/network/protocol/src/peer_set.rs
+++ b/polkadot/node/network/protocol/src/peer_set.rs
@@ -21,6 +21,7 @@ use polkadot_primitives::Hash;
use sc_network::{
config::{NonDefaultSetConfig, SetConfig},
types::ProtocolName,
+ NotificationService,
};
use std::{
collections::{hash_map::Entry, HashMap},
@@ -68,7 +69,7 @@ impl PeerSet {
self,
is_authority: IsAuthority,
peerset_protocol_names: &PeerSetProtocolNames,
- ) -> NonDefaultSetConfig {
+ ) -> (NonDefaultSetConfig, (PeerSet, Box)) {
// Networking layer relies on `get_main_name()` being the main name of the protocol
// for peersets and connection management.
let protocol = peerset_protocol_names.get_main_name(self);
@@ -76,39 +77,47 @@ impl PeerSet {
let max_notification_size = self.get_max_notification_size(is_authority);
match self {
- PeerSet::Validation => NonDefaultSetConfig {
- notifications_protocol: protocol,
- fallback_names,
- max_notification_size,
- handshake: None,
- set_config: SetConfig {
- // we allow full nodes to connect to validators for gossip
- // to ensure any `MIN_GOSSIP_PEERS` always include reserved peers
- // we limit the amount of non-reserved slots to be less
- // than `MIN_GOSSIP_PEERS` in total
- in_peers: super::MIN_GOSSIP_PEERS as u32 / 2 - 1,
- out_peers: super::MIN_GOSSIP_PEERS as u32 / 2 - 1,
- reserved_nodes: Vec::new(),
- non_reserved_mode: sc_network::config::NonReservedPeerMode::Accept,
- },
+ PeerSet::Validation => {
+ let (config, notification_service) = NonDefaultSetConfig::new(
+ protocol,
+ fallback_names,
+ max_notification_size,
+ None,
+ SetConfig {
+ // we allow full nodes to connect to validators for gossip
+ // to ensure any `MIN_GOSSIP_PEERS` always include reserved peers
+ // we limit the amount of non-reserved slots to be less
+ // than `MIN_GOSSIP_PEERS` in total
+ in_peers: super::MIN_GOSSIP_PEERS as u32 / 2 - 1,
+ out_peers: super::MIN_GOSSIP_PEERS as u32 / 2 - 1,
+ reserved_nodes: Vec::new(),
+ non_reserved_mode: sc_network::config::NonReservedPeerMode::Accept,
+ },
+ );
+
+ (config, (PeerSet::Validation, notification_service))
},
- PeerSet::Collation => NonDefaultSetConfig {
- notifications_protocol: protocol,
- fallback_names,
- max_notification_size,
- handshake: None,
- set_config: SetConfig {
- // Non-authority nodes don't need to accept incoming connections on this peer
- // set:
- in_peers: if is_authority == IsAuthority::Yes { 100 } else { 0 },
- out_peers: 0,
- reserved_nodes: Vec::new(),
- non_reserved_mode: if is_authority == IsAuthority::Yes {
- sc_network::config::NonReservedPeerMode::Accept
- } else {
- sc_network::config::NonReservedPeerMode::Deny
+ PeerSet::Collation => {
+ let (config, notification_service) = NonDefaultSetConfig::new(
+ protocol,
+ fallback_names,
+ max_notification_size,
+ None,
+ SetConfig {
+ // Non-authority nodes don't need to accept incoming connections on this
+ // peer set:
+ in_peers: if is_authority == IsAuthority::Yes { 100 } else { 0 },
+ out_peers: 0,
+ reserved_nodes: Vec::new(),
+ non_reserved_mode: if is_authority == IsAuthority::Yes {
+ sc_network::config::NonReservedPeerMode::Accept
+ } else {
+ sc_network::config::NonReservedPeerMode::Deny
+ },
},
- },
+ );
+
+ (config, (PeerSet::Collation, notification_service))
},
}
}
@@ -202,7 +211,7 @@ impl IndexMut for PerPeerSet {
pub fn peer_sets_info(
is_authority: IsAuthority,
peerset_protocol_names: &PeerSetProtocolNames,
-) -> Vec {
+) -> Vec<(NonDefaultSetConfig, (PeerSet, Box))> {
PeerSet::iter()
.map(|s| s.get_info(is_authority, &peerset_protocol_names))
.collect()
@@ -281,7 +290,7 @@ impl From for ProtocolVersion {
}
/// On the wire protocol name to [`PeerSet`] mapping.
-#[derive(Clone)]
+#[derive(Debug, Clone)]
pub struct PeerSetProtocolNames {
protocols: HashMap,
names: HashMap<(PeerSet, ProtocolVersion), ProtocolName>,
diff --git a/polkadot/node/service/Cargo.toml b/polkadot/node/service/Cargo.toml
index 90881aa051a0..f9ebb03d497b 100644
--- a/polkadot/node/service/Cargo.toml
+++ b/polkadot/node/service/Cargo.toml
@@ -84,6 +84,7 @@ kvdb = "0.13.0"
kvdb-rocksdb = { version = "0.19.0", optional = true }
parity-db = { version = "0.4.8", optional = true }
codec = { package = "parity-scale-codec", version = "3.6.1" }
+parking_lot = "0.12.1"
async-trait = "0.1.57"
schnellru = "0.2.1"
diff --git a/polkadot/node/service/src/lib.rs b/polkadot/node/service/src/lib.rs
index 95c887947c98..7d10a212ae30 100644
--- a/polkadot/node/service/src/lib.rs
+++ b/polkadot/node/service/src/lib.rs
@@ -51,7 +51,8 @@ use {
},
polkadot_node_core_dispute_coordinator::Config as DisputeCoordinatorConfig,
polkadot_node_network_protocol::{
- peer_set::PeerSetProtocolNames, request_response::ReqProtocolNames,
+ peer_set::{PeerSet, PeerSetProtocolNames},
+ request_response::ReqProtocolNames,
},
sc_client_api::BlockBackend,
sc_transaction_pool_api::OffchainTransactionPoolFactory,
@@ -74,7 +75,7 @@ pub use {
#[cfg(feature = "full-node")]
use polkadot_node_subsystem::jaeger;
-use std::{path::PathBuf, sync::Arc, time::Duration};
+use std::{collections::HashMap, path::PathBuf, sync::Arc, time::Duration};
use prometheus_endpoint::Registry;
#[cfg(feature = "full-node")]
@@ -810,9 +811,9 @@ pub fn new_full(
// anything in terms of behaviour, but makes the logs more consistent with the other
// Substrate nodes.
let grandpa_protocol_name = grandpa::protocol_standard_name(&genesis_hash, &config.chain_spec);
- net_config.add_notification_protocol(grandpa::grandpa_peers_set_config(
- grandpa_protocol_name.clone(),
- ));
+ let (grandpa_protocol_config, grandpa_notification_service) =
+ grandpa::grandpa_peers_set_config(grandpa_protocol_name.clone());
+ net_config.add_notification_protocol(grandpa_protocol_config);
let beefy_gossip_proto_name =
beefy::gossip_protocol_name(&genesis_hash, config.chain_spec.fork_id());
@@ -825,12 +826,17 @@ pub fn new_full(
client.clone(),
prometheus_registry.clone(),
);
- if enable_beefy {
- net_config.add_notification_protocol(beefy::communication::beefy_peers_set_config(
- beefy_gossip_proto_name.clone(),
- ));
- net_config.add_request_response_protocol(beefy_req_resp_cfg);
- }
+ let beefy_notification_service = match enable_beefy {
+ false => None,
+ true => {
+ let (beefy_notification_config, beefy_notification_service) =
+ beefy::communication::beefy_peers_set_config(beefy_gossip_proto_name.clone());
+
+ net_config.add_notification_protocol(beefy_notification_config);
+ net_config.add_request_response_protocol(beefy_req_resp_cfg);
+ Some(beefy_notification_service)
+ },
+ };
// validation/collation protocols are enabled only if `Overseer` is enabled
let peerset_protocol_names =
@@ -841,13 +847,21 @@ pub fn new_full(
//
// Collators and parachain full nodes require the collator and validator networking to send
// collations and to be able to recover PoVs.
- if role.is_authority() || is_parachain_node.is_running_alongside_parachain_node() {
- use polkadot_network_bridge::{peer_sets_info, IsAuthority};
- let is_authority = if role.is_authority() { IsAuthority::Yes } else { IsAuthority::No };
- for config in peer_sets_info(is_authority, &peerset_protocol_names) {
- net_config.add_notification_protocol(config);
- }
- }
+ let notification_services =
+ if role.is_authority() || is_parachain_node.is_running_alongside_parachain_node() {
+ use polkadot_network_bridge::{peer_sets_info, IsAuthority};
+ let is_authority = if role.is_authority() { IsAuthority::Yes } else { IsAuthority::No };
+
+ peer_sets_info(is_authority, &peerset_protocol_names)
+ .into_iter()
+ .map(|(config, (peerset, service))| {
+ net_config.add_notification_protocol(config);
+ (peerset, service)
+ })
+ .collect::>>()
+ } else {
+ std::collections::HashMap::new()
+ };
let req_protocol_names = ReqProtocolNames::new(&genesis_hash, config.chain_spec.fork_id());
@@ -1074,6 +1088,7 @@ pub fn new_full(
offchain_transaction_pool_factory: OffchainTransactionPoolFactory::new(
transaction_pool.clone(),
),
+ notification_services,
},
)
.map_err(|e| {
@@ -1182,6 +1197,8 @@ pub fn new_full(
sync: sync_service.clone(),
gossip_protocol_name: beefy_gossip_proto_name,
justifications_protocol_name,
+ notification_service: beefy_notification_service
+ .expect("BEEFY was enabled so `NotificationService` exists; qed"),
_phantom: core::marker::PhantomData::,
};
let payload_provider = beefy_primitives::mmr::MmrRootProvider::new(client.clone());
@@ -1279,6 +1296,7 @@ pub fn new_full(
prometheus_registry: prometheus_registry.clone(),
shared_voter_state,
telemetry: telemetry.as_ref().map(|x| x.handle()),
+ notification_service: grandpa_notification_service,
offchain_tx_pool_factory: OffchainTransactionPoolFactory::new(transaction_pool.clone()),
};
diff --git a/polkadot/node/service/src/overseer.rs b/polkadot/node/service/src/overseer.rs
index 33127b638e5a..99265541c0bb 100644
--- a/polkadot/node/service/src/overseer.rs
+++ b/polkadot/node/service/src/overseer.rs
@@ -26,7 +26,7 @@ use polkadot_node_core_candidate_validation::Config as CandidateValidationConfig
use polkadot_node_core_chain_selection::Config as ChainSelectionConfig;
use polkadot_node_core_dispute_coordinator::Config as DisputeCoordinatorConfig;
use polkadot_node_network_protocol::{
- peer_set::PeerSetProtocolNames,
+ peer_set::{PeerSet, PeerSetProtocolNames},
request_response::{
v1 as request_v1, vstaging as request_vstaging, IncomingRequestReceiver, ReqProtocolNames,
},
@@ -42,15 +42,16 @@ use polkadot_overseer::{
};
use schnellru::{ByLength, LruMap};
+use parking_lot::Mutex;
use polkadot_primitives::runtime_api::ParachainHost;
use sc_authority_discovery::Service as AuthorityDiscoveryService;
use sc_client_api::AuxStore;
use sc_keystore::LocalKeystore;
-use sc_network::NetworkStateInfo;
+use sc_network::{NetworkStateInfo, NotificationService};
use sp_api::ProvideRuntimeApi;
use sp_blockchain::HeaderBackend;
use sp_consensus_babe::BabeApi;
-use std::sync::Arc;
+use std::{collections::HashMap, sync::Arc};
pub use polkadot_approval_distribution::ApprovalDistribution as ApprovalDistributionSubsystem;
pub use polkadot_availability_bitfield_distribution::BitfieldDistribution as BitfieldDistributionSubsystem;
@@ -143,6 +144,8 @@ where
pub peerset_protocol_names: PeerSetProtocolNames,
/// The offchain transaction pool factory.
pub offchain_transaction_pool_factory: OffchainTransactionPoolFactory,
+ /// Notification services for validation/collation protocols.
+ pub notification_services: HashMap>,
}
/// Obtain a prepared `OverseerBuilder`, that is initialized
@@ -176,6 +179,7 @@ pub fn prepared_overseer_builder(
req_protocol_names,
peerset_protocol_names,
offchain_transaction_pool_factory,
+ notification_services,
}: OverseerGenArgs,
) -> Result<
InitializedOverseerBuilder<
@@ -221,6 +225,7 @@ where
use polkadot_node_subsystem_util::metrics::Metrics;
let metrics = ::register(registry)?;
+ let notification_sinks = Arc::new(Mutex::new(HashMap::new()));
let spawner = SpawnGlue(spawner);
@@ -238,6 +243,7 @@ where
network_bridge_metrics.clone(),
req_protocol_names,
peerset_protocol_names.clone(),
+ notification_sinks.clone(),
))
.network_bridge_rx(NetworkBridgeRxSubsystem::new(
network_service.clone(),
@@ -245,6 +251,8 @@ where
Box::new(sync_service.clone()),
network_bridge_metrics,
peerset_protocol_names,
+ notification_services,
+ notification_sinks,
))
.availability_distribution(AvailabilityDistributionSubsystem::new(
keystore.clone(),
diff --git a/substrate/bin/node-template/node/src/service.rs b/substrate/bin/node-template/node/src/service.rs
index 7303f5cd6dd6..f80dd1bb73a6 100644
--- a/substrate/bin/node-template/node/src/service.rs
+++ b/substrate/bin/node-template/node/src/service.rs
@@ -163,9 +163,9 @@ pub fn new_full(config: Configuration) -> Result {
&client.block_hash(0).ok().flatten().expect("Genesis block exists; qed"),
&config.chain_spec,
);
- net_config.add_notification_protocol(sc_consensus_grandpa::grandpa_peers_set_config(
- grandpa_protocol_name.clone(),
- ));
+ let (grandpa_protocol_config, grandpa_notification_service) =
+ sc_consensus_grandpa::grandpa_peers_set_config(grandpa_protocol_name.clone());
+ net_config.add_notification_protocol(grandpa_protocol_config);
let warp_sync = Arc::new(sc_consensus_grandpa::warp_proof::NetworkProvider::new(
backend.clone(),
@@ -315,6 +315,7 @@ pub fn new_full(config: Configuration) -> Result {
link: grandpa_link,
network,
sync: Arc::new(sync_service),
+ notification_service: grandpa_notification_service,
voting_rule: sc_consensus_grandpa::VotingRulesBuilder::default().build(),
prometheus_registry,
shared_voter_state: SharedVoterState::empty(),
diff --git a/substrate/bin/node/cli/src/service.rs b/substrate/bin/node/cli/src/service.rs
index ecca5c60db51..6ed890c8a7b3 100644
--- a/substrate/bin/node/cli/src/service.rs
+++ b/substrate/bin/node/cli/src/service.rs
@@ -359,19 +359,20 @@ pub fn new_full_base(
&client.block_hash(0).ok().flatten().expect("Genesis block exists; qed"),
&config.chain_spec,
);
- net_config.add_notification_protocol(grandpa::grandpa_peers_set_config(
- grandpa_protocol_name.clone(),
- ));
-
- let statement_handler_proto = sc_network_statement::StatementHandlerPrototype::new(
- client
- .block_hash(0u32.into())
- .ok()
- .flatten()
- .expect("Genesis block exists; qed"),
- config.chain_spec.fork_id(),
- );
- net_config.add_notification_protocol(statement_handler_proto.set_config());
+ let (grandpa_protocol_config, grandpa_notification_service) =
+ grandpa::grandpa_peers_set_config(grandpa_protocol_name.clone());
+ net_config.add_notification_protocol(grandpa_protocol_config);
+
+ let (statement_handler_proto, statement_config) =
+ sc_network_statement::StatementHandlerPrototype::new(
+ client
+ .block_hash(0u32.into())
+ .ok()
+ .flatten()
+ .expect("Genesis block exists; qed"),
+ config.chain_spec.fork_id(),
+ );
+ net_config.add_notification_protocol(statement_config);
let warp_sync = Arc::new(grandpa::warp_proof::NetworkProvider::new(
backend.clone(),
@@ -551,6 +552,7 @@ pub fn new_full_base(
link: grandpa_link,
network: network.clone(),
sync: Arc::new(sync_service.clone()),
+ notification_service: grandpa_notification_service,
telemetry: telemetry.as_ref().map(|x| x.handle()),
voting_rule: grandpa::VotingRulesBuilder::default().build(),
prometheus_registry: prometheus_registry.clone(),
diff --git a/substrate/client/consensus/beefy/src/communication/mod.rs b/substrate/client/consensus/beefy/src/communication/mod.rs
index 7f9535bfc23f..d92d3be7d687 100644
--- a/substrate/client/consensus/beefy/src/communication/mod.rs
+++ b/substrate/client/consensus/beefy/src/communication/mod.rs
@@ -67,10 +67,16 @@ pub(crate) mod beefy_protocol_name {
/// For standard protocol name see [`beefy_protocol_name::gossip_protocol_name`].
pub fn beefy_peers_set_config(
gossip_protocol_name: sc_network::ProtocolName,
-) -> sc_network::config::NonDefaultSetConfig {
- let mut cfg = sc_network::config::NonDefaultSetConfig::new(gossip_protocol_name, 1024 * 1024);
+) -> (sc_network::config::NonDefaultSetConfig, Box) {
+ let (mut cfg, notification_service) = sc_network::config::NonDefaultSetConfig::new(
+ gossip_protocol_name,
+ Vec::new(),
+ 1024 * 1024,
+ None,
+ Default::default(),
+ );
cfg.allow_non_reserved(25, 25);
- cfg
+ (cfg, notification_service)
}
// cost scalars for reporting peers.
diff --git a/substrate/client/consensus/beefy/src/lib.rs b/substrate/client/consensus/beefy/src/lib.rs
index 72df3cab8550..2b62dec4a626 100644
--- a/substrate/client/consensus/beefy/src/lib.rs
+++ b/substrate/client/consensus/beefy/src/lib.rs
@@ -38,7 +38,7 @@ use parking_lot::Mutex;
use prometheus::Registry;
use sc_client_api::{Backend, BlockBackend, BlockchainEvents, FinalityNotifications, Finalizer};
use sc_consensus::BlockImport;
-use sc_network::{NetworkRequest, ProtocolName};
+use sc_network::{NetworkRequest, NotificationService, ProtocolName};
use sc_network_gossip::{GossipEngine, Network as GossipNetwork, Syncing as GossipSyncing};
use sp_api::{HeaderT, NumberFor, ProvideRuntimeApi};
use sp_blockchain::{
@@ -178,6 +178,8 @@ pub struct BeefyNetworkParams {
pub network: Arc,
/// Syncing service implementing a sync oracle and an event stream for peers.
pub sync: Arc,
+ /// Handle for receiving notification events.
+ pub notification_service: Box,
/// Chain specific BEEFY gossip protocol name. See
/// [`communication::beefy_protocol_name::gossip_protocol_name`].
pub gossip_protocol_name: ProtocolName,
@@ -243,6 +245,7 @@ pub async fn start_beefy_gadget(
let BeefyNetworkParams {
network,
sync,
+ notification_service,
gossip_protocol_name,
justifications_protocol_name,
..
@@ -264,6 +267,7 @@ pub async fn start_beefy_gadget(
let gossip_engine = GossipEngine::new(
network.clone(),
sync.clone(),
+ notification_service,
gossip_protocol_name.clone(),
gossip_validator.clone(),
None,
diff --git a/substrate/client/consensus/beefy/src/tests.rs b/substrate/client/consensus/beefy/src/tests.rs
index 3bb65e9d57f4..5bc183387599 100644
--- a/substrate/client/consensus/beefy/src/tests.rs
+++ b/substrate/client/consensus/beefy/src/tests.rs
@@ -71,7 +71,7 @@ use substrate_test_runtime_client::{BlockBuilderExt, ClientExt};
use tokio::time::Duration;
const GENESIS_HASH: H256 = H256::zero();
-fn beefy_gossip_proto_name() -> ProtocolName {
+pub(crate) fn beefy_gossip_proto_name() -> ProtocolName {
gossip_protocol_name(GENESIS_HASH, None)
}
@@ -370,6 +370,7 @@ async fn voter_init_setup(
let mut gossip_engine = sc_network_gossip::GossipEngine::new(
net.peer(0).network_service().clone(),
net.peer(0).sync_service().clone(),
+ net.peer(0).take_notification_service(&beefy_gossip_proto_name()).unwrap(),
"/beefy/whatever",
gossip_validator,
None,
@@ -391,6 +392,14 @@ where
{
let tasks = FuturesUnordered::new();
+ let mut notification_services = peers
+ .iter()
+ .map(|(peer_id, _, _)| {
+ let peer = &mut net.peers[*peer_id];
+ (*peer_id, peer.take_notification_service(&beefy_gossip_proto_name()).unwrap())
+ })
+ .collect::>();
+
for (peer_id, key, api) in peers.into_iter() {
let peer = &net.peers[peer_id];
@@ -408,6 +417,7 @@ where
let network_params = crate::BeefyNetworkParams {
network: peer.network_service().clone(),
sync: peer.sync_service().clone(),
+ notification_service: notification_services.remove(&peer_id).unwrap(),
gossip_protocol_name: beefy_gossip_proto_name(),
justifications_protocol_name: on_demand_justif_handler.protocol_name(),
_phantom: PhantomData,
@@ -1031,7 +1041,25 @@ async fn should_initialize_voter_at_custom_genesis() {
net.peer(0).client().as_client().finalize_block(hashes[8], None).unwrap();
// load persistent state - nothing in DB, should init at genesis
- let persisted_state = voter_init_setup(&mut net, &mut finality, &api).await.unwrap();
+ //
+ // NOTE: code from `voter_init_setup()` is moved here because the new network event system
+ // doesn't allow creating a new `GossipEngine` as the notification handle is consumed by the
+ // first `GossipEngine`
+ let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
+ let (gossip_validator, _) = GossipValidator::new(known_peers);
+ let gossip_validator = Arc::new(gossip_validator);
+ let mut gossip_engine = sc_network_gossip::GossipEngine::new(
+ net.peer(0).network_service().clone(),
+ net.peer(0).sync_service().clone(),
+ net.peer(0).take_notification_service(&beefy_gossip_proto_name()).unwrap(),
+ "/beefy/whatever",
+ gossip_validator,
+ None,
+ );
+ let (beefy_genesis, best_grandpa) =
+ wait_for_runtime_pallet(&api, &mut gossip_engine, &mut finality).await.unwrap();
+ let persisted_state =
+ load_or_init_voter_state(&*backend, &api, beefy_genesis, best_grandpa, 1).unwrap();
// Test initialization at session boundary.
// verify voter initialized with single session starting at block `custom_pallet_genesis` (7)
@@ -1061,7 +1089,11 @@ async fn should_initialize_voter_at_custom_genesis() {
net.peer(0).client().as_client().finalize_block(hashes[10], None).unwrap();
// load persistent state - state preset in DB, but with different pallet genesis
- let new_persisted_state = voter_init_setup(&mut net, &mut finality, &api).await.unwrap();
+ // the network state persists and uses the old `GossipEngine` initialized for `peer(0)`
+ let (beefy_genesis, best_grandpa) =
+ wait_for_runtime_pallet(&api, &mut gossip_engine, &mut finality).await.unwrap();
+ let new_persisted_state =
+ load_or_init_voter_state(&*backend, &api, beefy_genesis, best_grandpa, 1).unwrap();
// verify voter initialized with single session starting at block `new_pallet_genesis` (10)
let sessions = new_persisted_state.voting_oracle().sessions();
@@ -1309,7 +1341,7 @@ async fn gossipped_finality_proofs() {
let api = Arc::new(TestApi::with_validator_set(&validator_set));
let beefy_peers = peers.iter().enumerate().map(|(id, key)| (id, key, api.clone())).collect();
- let charlie = &net.peers[2];
+ let charlie = &mut net.peers[2];
let known_peers = Arc::new(Mutex::new(KnownPeers::::new()));
// Charlie will run just the gossip engine and not the full voter.
let (gossip_validator, _) = GossipValidator::new(known_peers);
@@ -1322,6 +1354,7 @@ async fn gossipped_finality_proofs() {
let mut charlie_gossip_engine = sc_network_gossip::GossipEngine::new(
charlie.network_service().clone(),
charlie.sync_service().clone(),
+ charlie.take_notification_service(&beefy_gossip_proto_name()).unwrap(),
beefy_gossip_proto_name(),
charlie_gossip_validator.clone(),
None,
diff --git a/substrate/client/consensus/beefy/src/worker.rs b/substrate/client/consensus/beefy/src/worker.rs
index a239e34030c2..a50fdab28ec4 100644
--- a/substrate/client/consensus/beefy/src/worker.rs
+++ b/substrate/client/consensus/beefy/src/worker.rs
@@ -1131,12 +1131,16 @@ pub(crate) mod tests {
let api = Arc::new(TestApi::with_validator_set(&genesis_validator_set));
let network = peer.network_service().clone();
let sync = peer.sync_service().clone();
+ let notification_service = peer
+ .take_notification_service(&crate::tests::beefy_gossip_proto_name())
+ .unwrap();
let known_peers = Arc::new(Mutex::new(KnownPeers::new()));
let (gossip_validator, gossip_report_stream) = GossipValidator::new(known_peers.clone());
let gossip_validator = Arc::new(gossip_validator);
let gossip_engine = GossipEngine::new(
network.clone(),
sync.clone(),
+ notification_service,
"/beefy/1",
gossip_validator.clone(),
None,
diff --git a/substrate/client/consensus/grandpa/src/communication/mod.rs b/substrate/client/consensus/grandpa/src/communication/mod.rs
index c0749858568f..22785639e1f6 100644
--- a/substrate/client/consensus/grandpa/src/communication/mod.rs
+++ b/substrate/client/consensus/grandpa/src/communication/mod.rs
@@ -46,7 +46,7 @@ use finality_grandpa::{
Message::{Precommit, Prevote, PrimaryPropose},
};
use parity_scale_codec::{Decode, DecodeAll, Encode};
-use sc_network::{NetworkBlock, NetworkSyncForkRequest, ReputationChange};
+use sc_network::{NetworkBlock, NetworkSyncForkRequest, NotificationService, ReputationChange};
use sc_network_gossip::{GossipEngine, Network as GossipNetwork};
use sc_telemetry::{telemetry, TelemetryHandle, CONSENSUS_DEBUG, CONSENSUS_INFO};
use sp_keystore::KeystorePtr;
@@ -247,6 +247,7 @@ impl, S: Syncing> NetworkBridge {
pub(crate) fn new(
service: N,
sync: S,
+ notification_service: Box,
config: crate::Config,
set_state: crate::environment::SharedVoterSetState,
prometheus_registry: Option<&Registry>,
@@ -260,6 +261,7 @@ impl, S: Syncing> NetworkBridge {
let gossip_engine = Arc::new(Mutex::new(GossipEngine::new(
service.clone(),
sync.clone(),
+ notification_service,
protocol,
validator.clone(),
prometheus_registry,
diff --git a/substrate/client/consensus/grandpa/src/communication/tests.rs b/substrate/client/consensus/grandpa/src/communication/tests.rs
index 10c4772fc76d..3fbc954d89cf 100644
--- a/substrate/client/consensus/grandpa/src/communication/tests.rs
+++ b/substrate/client/consensus/grandpa/src/communication/tests.rs
@@ -24,17 +24,18 @@ use super::{
};
use crate::{communication::grandpa_protocol_name, environment::SharedVoterSetState};
use futures::prelude::*;
-use parity_scale_codec::Encode;
+use parity_scale_codec::{DecodeAll, Encode};
use sc_network::{
config::{MultiaddrWithPeerId, Role},
event::Event as NetworkEvent,
+ service::traits::{Direction, MessageSink, NotificationEvent, NotificationService},
types::ProtocolName,
Multiaddr, NetworkBlock, NetworkEventStream, NetworkNotification, NetworkPeers,
NetworkSyncForkRequest, NotificationSenderError, NotificationSenderT as NotificationSender,
PeerId, ReputationChange,
};
use sc_network_common::{
- role::ObservedRole,
+ role::{ObservedRole, Roles},
sync::{SyncEvent as SyncStreamEvent, SyncEventStream},
};
use sc_network_gossip::Validator;
@@ -125,6 +126,12 @@ impl NetworkPeers for TestNetwork {
fn sync_num_connected(&self) -> usize {
unimplemented!();
}
+
+ fn peer_role(&self, _peer_id: PeerId, handshake: Vec) -> Option {
+ Roles::decode_all(&mut &handshake[..])
+ .ok()
+ .and_then(|role| Some(ObservedRole::from(role)))
+ }
}
impl NetworkEventStream for TestNetwork {
@@ -213,10 +220,65 @@ impl NetworkSyncForkRequest> for TestSync {
fn set_sync_fork_request(&self, _peers: Vec, _hash: Hash, _number: NumberFor) {}
}
+#[derive(Debug)]
+pub(crate) struct TestNotificationService {
+ rx: TracingUnboundedReceiver,
+}
+
+#[async_trait::async_trait]
+impl NotificationService for TestNotificationService {
+ /// Instruct `Notifications` to open a new substream for `peer`.
+ async fn open_substream(&mut self, _peer: PeerId) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Instruct `Notifications` to close substream for `peer`.
+ async fn close_substream(&mut self, _peer: PeerId) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Send synchronous `notification` to `peer`.
+ fn send_sync_notification(&self, _peer: &PeerId, _notification: Vec) {
+ unimplemented!();
+ }
+
+ /// Send asynchronous `notification` to `peer`, allowing sender to exercise backpressure.
+ async fn send_async_notification(
+ &self,
+ _peer: &PeerId,
+ _notification: Vec,
+ ) -> Result<(), sc_network::error::Error> {
+ unimplemented!();
+ }
+
+ /// Set handshake for the notification protocol replacing the old handshake.
+ async fn set_handshake(&mut self, _handshake: Vec) -> Result<(), ()> {
+ unimplemented!();
+ }
+
+ /// Get next event from the `Notifications` event stream.
+ async fn next_event(&mut self) -> Option {
+ self.rx.next().await
+ }
+
+ fn clone(&mut self) -> Result, ()> {
+ unimplemented!();
+ }
+
+ fn protocol(&self) -> &ProtocolName {
+ unimplemented!();
+ }
+
+ fn message_sink(&self, _peer: &PeerId) -> Option> {
+ unimplemented!();
+ }
+}
+
pub(crate) struct Tester {
pub(crate) net_handle: super::NetworkBridge,
gossip_validator: Arc>,
pub(crate) events: TracingUnboundedReceiver,
+ pub(crate) notification_tx: TracingUnboundedSender,
}
impl Tester {
@@ -281,6 +343,9 @@ fn voter_set_state() -> SharedVoterSetState {
// needs to run in a tokio runtime.
pub(crate) fn make_test_network() -> (impl Future