Skip to content
This repository has been archived by the owner on Nov 15, 2023. It is now read-only.

split NetworkBridge into two subsystems #5616

Merged
merged 29 commits into from
Jul 12, 2022
Merged

split NetworkBridge into two subsystems #5616

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
29 commits
Select commit Hold shift + click to select a range
4f84c52
foo
drahnr May 25, 2022
859d0b4
rolling session window
drahnr May 25, 2022
3fe17bc
fixup
drahnr May 30, 2022
7f50156
remove use statemetn
drahnr May 30, 2022
93bc459
fmt
drahnr May 30, 2022
db967f7
split NetworkBridge into two subsystems
drahnr May 31, 2022
858f99e
split
drahnr Jun 1, 2022
f279051
chore: reexport OrchestraError as OverseerError
drahnr Jun 1, 2022
d874fe3
chore: silence warnings
drahnr Jun 1, 2022
fdb637b
fixup tests
drahnr Jun 2, 2022
471b0de
chore: add default timenout of 30s to subsystem test helper ctx handle
drahnr Jun 2, 2022
3c92466
single item channel
drahnr Jun 2, 2022
a5e5a71
fixins
drahnr Jun 2, 2022
2076903
fmt
drahnr Jun 2, 2022
b8004a5
cleanup
drahnr Jun 2, 2022
1762559
remove dead code
drahnr Jun 2, 2022
653c461
remove sync bounds again
drahnr Jun 2, 2022
9ba9173
wire up shared state
drahnr Jun 2, 2022
5019318
deal with some FIXMEs
drahnr Jun 2, 2022
ad37193
use distinct tags
drahnr Jun 6, 2022
683b978
use tag
drahnr Jun 6, 2022
c7337c2
address naming
drahnr Jun 20, 2022
079a28c
remove unused sync oracle
drahnr Jun 20, 2022
6109018
remove unneeded state
drahnr Jun 20, 2022
4beba74
fix tests
drahnr Jun 22, 2022
7cdb033
chore: fmt
drahnr Jun 22, 2022
36253e2
do not try to register twice
drahnr Jun 23, 2022
28d0ae8
leak Metrics type
drahnr Jun 23, 2022
687ae9a
Merge branch 'master' into bernhard-split-network-bridge-into-two-sub…
ordian Jul 12, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

16 changes: 8 additions & 8 deletions node/network/approval-distribution/src/lib.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -32,7 +32,7 @@ use polkadot_node_primitives::approval::{
use polkadot_node_subsystem::{
messages::{
ApprovalCheckResult, ApprovalDistributionMessage, ApprovalVotingMessage,
AssignmentCheckResult, NetworkBridgeEvent, NetworkBridgeMessage,
AssignmentCheckResult, NetworkBridgeEvent, NetworkBridgeTxMessage,
},
overseer, ActiveLeavesUpdate, FromOrchestra, OverseerSignal, SpawnedSubsystem, SubsystemError,
};
Expand Down Expand Up @@ -938,7 +938,7 @@ impl State {
"Sending an assignment to peers",
);

ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
ctx.send_message(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments),
Expand Down Expand Up @@ -1200,7 +1200,7 @@ impl State {
"Sending an approval to peers",
);

ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
ctx.send_message(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(approvals),
Expand Down Expand Up @@ -1328,7 +1328,7 @@ impl State {
);

sender
.send_message(NetworkBridgeMessage::SendValidationMessage(
.send_message(NetworkBridgeTxMessage::SendValidationMessage(
vec![peer_id.clone()],
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments_to_send),
Expand All @@ -1346,7 +1346,7 @@ impl State {
);

sender
.send_message(NetworkBridgeMessage::SendValidationMessage(
.send_message(NetworkBridgeTxMessage::SendValidationMessage(
vec![peer_id],
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(approvals_to_send),
Expand Down Expand Up @@ -1549,7 +1549,7 @@ async fn adjust_required_routing_and_propagate<Context, BlockFilter, RoutingModi
// Send messages in accumulated packets, assignments preceding approvals.

for (peer, assignments_packet) in peer_assignments {
ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
ctx.send_message(NetworkBridgeTxMessage::SendValidationMessage(
vec![peer],
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments_packet),
Expand All @@ -1559,7 +1559,7 @@ async fn adjust_required_routing_and_propagate<Context, BlockFilter, RoutingModi
}

for (peer, approvals_packet) in peer_approvals {
ctx.send_message(NetworkBridgeMessage::SendValidationMessage(
ctx.send_message(NetworkBridgeTxMessage::SendValidationMessage(
vec![peer],
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(approvals_packet),
Expand All @@ -1582,7 +1582,7 @@ async fn modify_reputation(
"Reputation change for peer",
);

sender.send_message(NetworkBridgeMessage::ReportPeer(peer_id, rep)).await;
sender.send_message(NetworkBridgeTxMessage::ReportPeer(peer_id, rep)).await;
}

#[overseer::contextbounds(ApprovalDistribution, prefix = self::overseer)]
Expand Down
62 changes: 31 additions & 31 deletions node/network/approval-distribution/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -229,8 +229,8 @@ async fn expect_reputation_change(
) {
assert_matches!(
overseer_recv(virtual_overseer).await,
AllMessages::NetworkBridge(
NetworkBridgeMessage::ReportPeer(
AllMessages::NetworkBridgeTx(
NetworkBridgeTxMessage::ReportPeer(
rep_peer,
rep,
)
Expand Down Expand Up @@ -299,7 +299,7 @@ fn try_import_the_same_assignment() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments)
Expand Down Expand Up @@ -461,7 +461,7 @@ fn peer_sending_us_the_same_we_just_sent_them_is_ok() {
// we should send them the assignment
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments)
Expand Down Expand Up @@ -528,7 +528,7 @@ fn import_approval_happy_path() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments)
Expand Down Expand Up @@ -564,7 +564,7 @@ fn import_approval_happy_path() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(approvals)
Expand Down Expand Up @@ -787,7 +787,7 @@ fn update_peer_view() {
// we should send relevant assignments to the peer
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments)
Expand Down Expand Up @@ -838,7 +838,7 @@ fn update_peer_view() {
// we should send relevant assignments to the peer
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(assignments)
Expand Down Expand Up @@ -1025,7 +1025,7 @@ fn sends_assignments_even_when_state_is_approved() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1038,7 +1038,7 @@ fn sends_assignments_even_when_state_is_approved() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand Down Expand Up @@ -1200,7 +1200,7 @@ fn propagates_locally_generated_assignment_to_both_dimensions() {

let assignment_sent_peers = assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1221,7 +1221,7 @@ fn propagates_locally_generated_assignment_to_both_dimensions() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand Down Expand Up @@ -1304,7 +1304,7 @@ fn propagates_assignments_along_unshared_dimension() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand Down Expand Up @@ -1353,7 +1353,7 @@ fn propagates_assignments_along_unshared_dimension() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand Down Expand Up @@ -1448,7 +1448,7 @@ fn propagates_to_required_after_connect() {

let assignment_sent_peers = assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1469,7 +1469,7 @@ fn propagates_to_required_after_connect() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand All @@ -1486,7 +1486,7 @@ fn propagates_to_required_after_connect() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1500,7 +1500,7 @@ fn propagates_to_required_after_connect() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand Down Expand Up @@ -1574,7 +1574,7 @@ fn sends_to_more_peers_after_getting_topology() {
let mut expected_indices = vec![0, 10, 20, 30, 50, 51, 52, 53];
let assignment_sent_peers = assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1596,7 +1596,7 @@ fn sends_to_more_peers_after_getting_topology() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand All @@ -1621,7 +1621,7 @@ fn sends_to_more_peers_after_getting_topology() {
for _ in 0..expected_indices_assignments.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1642,7 +1642,7 @@ fn sends_to_more_peers_after_getting_topology() {
for _ in 0..expected_indices_approvals.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand Down Expand Up @@ -1732,7 +1732,7 @@ fn originator_aggression_l1() {

let prev_sent_indices = assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(_)
Expand All @@ -1746,7 +1746,7 @@ fn originator_aggression_l1() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
_,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(_)
Expand Down Expand Up @@ -1782,7 +1782,7 @@ fn originator_aggression_l1() {
for _ in 0..unsent_indices.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand All @@ -1802,7 +1802,7 @@ fn originator_aggression_l1() {
for _ in 0..unsent_indices.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Approvals(sent_approvals)
Expand Down Expand Up @@ -1891,7 +1891,7 @@ fn non_originator_aggression_l1() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
_,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(_)
Expand Down Expand Up @@ -1996,7 +1996,7 @@ fn non_originator_aggression_l2() {

let prev_sent_indices = assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(_)
Expand Down Expand Up @@ -2066,7 +2066,7 @@ fn non_originator_aggression_l2() {
for _ in 0..unsent_indices.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand Down Expand Up @@ -2157,7 +2157,7 @@ fn resends_messages_periodically() {

assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand Down Expand Up @@ -2206,7 +2206,7 @@ fn resends_messages_periodically() {
for _ in 0..expected_y.len() {
assert_matches!(
overseer_recv(overseer).await,
AllMessages::NetworkBridge(NetworkBridgeMessage::SendValidationMessage(
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendValidationMessage(
sent_peers,
Versioned::V1(protocol_v1::ValidationProtocol::ApprovalDistribution(
protocol_v1::ApprovalDistributionMessage::Assignments(sent_assignments)
Expand Down
9 changes: 6 additions & 3 deletions node/network/availability-distribution/src/pov_requester/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ use polkadot_node_network_protocol::request_response::{
use polkadot_node_primitives::PoV;
use polkadot_node_subsystem::{
jaeger,
messages::{IfDisconnected, NetworkBridgeMessage},
messages::{IfDisconnected, NetworkBridgeTxMessage},
overseer,
};
use polkadot_node_subsystem_util::runtime::RuntimeInfo;
Expand Down Expand Up @@ -62,7 +62,7 @@ pub async fn fetch_pov<Context>(
);
let full_req = Requests::PoVFetchingV1(req);

ctx.send_message(NetworkBridgeMessage::SendRequests(
ctx.send_message(NetworkBridgeTxMessage::SendRequests(
vec![full_req],
IfDisconnected::ImmediateError,
))
Expand Down Expand Up @@ -195,7 +195,10 @@ mod tests {
)) => {
tx.send(Ok(Some(make_session_info()))).unwrap();
},
AllMessages::NetworkBridge(NetworkBridgeMessage::SendRequests(mut reqs, _)) => {
AllMessages::NetworkBridgeTx(NetworkBridgeTxMessage::SendRequests(
mut reqs,
_,
)) => {
let req = assert_matches!(
reqs.pop(),
Some(Requests::PoVFetchingV1(outgoing)) => {outgoing}
Expand Down
Loading