1use async_trait::async_trait;
2
3use crate::dht::types::Chord;
4use crate::dht::types::CorrectChord;
5use crate::dht::PeerRingAction;
6use crate::dht::TopoInfo;
7use crate::error::Error;
8use crate::error::Result;
9use crate::message::effects::MessageSendFunctor;
10use crate::message::effects::PayloadRelayFunctor;
11use crate::message::types::ConnectNodeReport;
12use crate::message::types::ConnectNodeSend;
13use crate::message::types::FindSuccessorReport;
14use crate::message::types::FindSuccessorSend;
15use crate::message::types::Message;
16use crate::message::types::QueryForTopoInfoReport;
17use crate::message::types::QueryForTopoInfoSend;
18use crate::message::types::Then;
19use crate::message::FindSuccessorReportHandler;
20use crate::message::FindSuccessorThen;
21use crate::message::HandleMsg;
22use crate::message::MessageHandler;
23use crate::message::MessagePayload;
24
25#[cfg_attr(feature = "wasm", async_trait(?Send))]
27#[cfg_attr(not(feature = "wasm"), async_trait)]
28impl HandleMsg<QueryForTopoInfoSend> for MessageHandler {
29 async fn handle(&self, ctx: &MessagePayload, msg: &QueryForTopoInfoSend) -> Result<()> {
30 let info: TopoInfo = TopoInfo::try_from(self.dht.as_ref())?;
31 if msg.targets(self.dht.did) {
32 self.run_effects([PayloadRelayFunctor::send_report_message(
33 ctx,
34 Message::QueryForTopoInfoReport(msg.resp(info)),
35 )
36 .into()])
37 .await?
38 }
39 Ok(())
40 }
41}
42
43#[cfg_attr(feature = "wasm", async_trait(?Send))]
45#[cfg_attr(not(feature = "wasm"), async_trait)]
46impl HandleMsg<QueryForTopoInfoReport> for MessageHandler {
47 async fn handle(&self, _ctx: &MessagePayload, msg: &QueryForTopoInfoReport) -> Result<()> {
48 match msg.then {
49 <QueryForTopoInfoReport as Then>::Then::SyncSuccessor => {
50 for peer in msg.info.successors.iter() {
51 if self.transport.get_connection(*peer).is_some() {
52 self.join_dht(*peer).await?;
53 }
54 }
55 }
56 <QueryForTopoInfoReport as Then>::Then::Stabilization => {
57 if let Some(peer) = msg.info.predecessor {
60 self.connect_dht_peer(peer).await?;
61 }
62 for peer in msg.info.successors.iter() {
63 self.connect_dht_peer(*peer).await?;
64 }
65 let ev = self.dht.stabilize(msg.info.clone())?;
66 self.handle_dht_events(&ev).await?;
67 }
68 }
69 Ok(())
70 }
71}
72
73#[cfg_attr(feature = "wasm", async_trait(?Send))]
74#[cfg_attr(not(feature = "wasm"), async_trait)]
75impl HandleMsg<ConnectNodeSend> for MessageHandler {
76 async fn handle(&self, ctx: &MessagePayload, msg: &ConnectNodeSend) -> Result<()> {
77 if msg.network_id != self.transport.network_id {
78 return Ok(());
79 }
80
81 if ctx.should_forward_from(self.dht.did) {
82 self.run_effects([PayloadRelayFunctor::forward_payload(ctx, None).into()])
83 .await
84 } else {
85 let answer = self
86 .transport
87 .answer_remote_connection(
88 ctx.relay.try_origin_sender()?,
89 self.inner_callback(),
90 msg,
91 )
92 .await?;
93 self.run_effects([PayloadRelayFunctor::send_report_message(
94 ctx,
95 Message::ConnectNodeReport(answer),
96 )
97 .into()])
98 .await
99 }
100 }
101}
102
103#[cfg_attr(feature = "wasm", async_trait(?Send))]
104#[cfg_attr(not(feature = "wasm"), async_trait)]
105impl HandleMsg<ConnectNodeReport> for MessageHandler {
106 async fn handle(&self, ctx: &MessagePayload, msg: &ConnectNodeReport) -> Result<()> {
107 if ctx.should_forward_from(self.dht.did) {
108 self.run_effects([PayloadRelayFunctor::forward_payload(ctx, None).into()])
109 .await
110 } else {
111 self.transport
112 .accept_remote_connection(ctx.relay.try_origin_sender()?, msg)
113 .await
114 }
115 }
116}
117
118#[cfg_attr(feature = "wasm", async_trait(?Send))]
119#[cfg_attr(not(feature = "wasm"), async_trait)]
120impl HandleMsg<FindSuccessorSend> for MessageHandler {
121 async fn handle(&self, ctx: &MessagePayload, msg: &FindSuccessorSend) -> Result<()> {
122 match self.dht.find_successor(msg.did)? {
123 PeerRingAction::Some(did) => {
124 if msg.accepts_local_successor(self.dht.did) {
125 match &msg.then {
126 FindSuccessorThen::Report(handler) => {
127 self.run_effects([PayloadRelayFunctor::send_report_message(
128 ctx,
129 Message::FindSuccessorReport(FindSuccessorReport {
130 did,
131 handler: handler.clone(),
132 }),
133 )
134 .into()])
135 .await
136 }
137 }
138 } else {
139 self.run_effects([PayloadRelayFunctor::forward_payload(ctx, Some(did)).into()])
140 .await
141 }
142 }
143 PeerRingAction::RemoteAction(next, _) => {
144 self.run_effects([PayloadRelayFunctor::reset_destination(ctx, next).into()])
145 .await
146 }
147 act => Err(Error::unexpected_peer_ring_action(act)),
148 }
149 }
150}
151
152#[cfg_attr(feature = "wasm", async_trait(?Send))]
153#[cfg_attr(not(feature = "wasm"), async_trait)]
154impl HandleMsg<FindSuccessorReport> for MessageHandler {
155 async fn handle(&self, ctx: &MessagePayload, msg: &FindSuccessorReport) -> Result<()> {
156 if ctx.should_forward_from(self.dht.did) {
157 return self
158 .run_effects([PayloadRelayFunctor::forward_payload(ctx, None).into()])
159 .await;
160 }
161
162 match &msg.handler {
163 FindSuccessorReportHandler::FixFingerTable { index } => {
164 self.dht.apply_fixed_finger(*index, msg.did)?;
165 if msg.reports_remote_successor(self.dht.did) {
166 let offer_msg = self
167 .transport
168 .prepare_connection_offer(msg.did, self.inner_callback())
169 .await?;
170 self.run_effects([MessageSendFunctor::send_message(
171 Message::ConnectNodeSend(offer_msg),
172 msg.did,
173 )
174 .into()])
175 .await?;
176 }
177 }
178 FindSuccessorReportHandler::Connect => {
179 if msg.reports_remote_successor(self.dht.did) {
180 let offer_msg = self
181 .transport
182 .prepare_connection_offer(msg.did, self.inner_callback())
183 .await?;
184 self.run_effects([MessageSendFunctor::send_message(
185 Message::ConnectNodeSend(offer_msg),
186 msg.did,
187 )
188 .into()])
189 .await?;
190 }
191 }
192 _ => {}
193 }
194
195 Ok(())
196 }
197}
198
199#[cfg(not(feature = "wasm"))]
200#[cfg(test)]
201pub mod tests {
202 use tokio::time::sleep;
204 use tokio::time::Duration;
205
206 use super::*;
207 use crate::dht::successor::SuccessorReader;
208 use crate::ecc::tests::gen_ordered_keys;
209 use crate::ecc::SecretKey;
210 use crate::tests::default::assert_no_more_msg;
211 use crate::tests::default::prepare_node;
212 use crate::tests::default::wait_for_msgs;
213 use crate::tests::default::Node;
214 use crate::tests::manually_establish_connection;
215
216 #[tokio::test]
258 async fn test_triple_nodes_connection_1_2_3() -> Result<()> {
259 let keys = gen_ordered_keys(3);
260 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
261 test_triple_ordered_nodes_connection(key1, key2, key3).await?;
262 Ok(())
263 }
264
265 #[tokio::test]
267 async fn test_triple_nodes_connection_2_3_1() -> Result<()> {
268 let keys = gen_ordered_keys(3);
269 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
270 test_triple_ordered_nodes_connection(key2, key3, key1).await?;
271 Ok(())
272 }
273
274 #[tokio::test]
276 async fn test_triple_nodes_connection_3_1_2() -> Result<()> {
277 let keys = gen_ordered_keys(3);
278 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
279 test_triple_ordered_nodes_connection(key3, key1, key2).await?;
280 Ok(())
281 }
282
283 #[tokio::test]
303 async fn test_triple_nodes_connection_3_2_1() -> Result<()> {
304 let keys = gen_ordered_keys(3);
305 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
306 test_triple_desc_ordered_nodes_connection(key3, key2, key1).await?;
307 Ok(())
308 }
309
310 #[tokio::test]
312 async fn test_triple_nodes_connection_2_1_3() -> Result<()> {
313 let keys = gen_ordered_keys(3);
314 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
315 test_triple_desc_ordered_nodes_connection(key2, key1, key3).await?;
316 Ok(())
317 }
318
319 #[tokio::test]
321 async fn test_triple_nodes_connection_1_3_2() -> Result<()> {
322 let keys = gen_ordered_keys(3);
323 let (key1, key2, key3) = (keys[0], keys[1], keys[2]);
324 test_triple_desc_ordered_nodes_connection(key1, key3, key2).await?;
325 Ok(())
326 }
327
328 async fn test_triple_ordered_nodes_connection(
329 key1: SecretKey,
330 key2: SecretKey,
331 key3: SecretKey,
332 ) -> Result<(Node, Node, Node)> {
333 let node1 = prepare_node(key1).await;
334 let node2 = prepare_node(key2).await;
335 let node3 = prepare_node(key3).await;
336
337 println!("========================================");
338 println!("|| now we connect node1 and node2 ||");
339 println!("========================================");
340
341 manually_establish_connection(&node1.swarm, &node2.swarm).await;
342 wait_for_msgs([&node1, &node2, &node3]).await;
343 assert_no_more_msg([&node1, &node2, &node3]).await;
344
345 node1.assert_transports(vec![node2.did()]);
346 node2.assert_transports(vec![node1.did()]);
347 node3.assert_transports(vec![]);
348 assert_eq!(node1.dht().successors().list()?, vec![node2.did()]);
349 assert_eq!(node2.dht().successors().list()?, vec![node1.did()]);
350 assert_eq!(node3.dht().successors().list()?, vec![]);
351
352 println!("========================================");
353 println!("|| now we start join node3 to node2 ||");
354 println!("========================================");
355
356 manually_establish_connection(&node3.swarm, &node2.swarm).await;
357 wait_for_msgs([&node1, &node2, &node3]).await;
358 assert_no_more_msg([&node1, &node2, &node3]).await;
359
360 println!("=== Check state before connect via DHT ===");
361 node1.assert_transports(vec![node2.did()]);
362 node2.assert_transports(vec![node1.did(), node3.did()]);
363 node3.assert_transports(vec![node2.did()]);
364 assert_eq!(node1.dht().successors().list()?, vec![node2.did(),]);
365 assert_eq!(node2.dht().successors().list()?, vec![
366 node3.did(),
367 node1.did()
368 ]);
369 assert_eq!(node3.dht().successors().list()?, vec![node2.did()]);
370
371 println!("=============================================");
372 println!("|| now we connect node1 to node3 via DHT ||");
373 println!("=============================================");
374
375 assert!(node1.swarm.transport.get_connection(node3.did()).is_none());
377 assert_eq!(node1.dht().successors().max()?, node2.did());
379
380 node1.swarm.connect(node3.did()).await?;
381 wait_for_msgs([&node1, &node2, &node3]).await;
382 assert_no_more_msg([&node1, &node2, &node3]).await;
383
384 println!("=== Check state after connect via DHT ===");
385 node1.assert_transports(vec![node2.did(), node3.did()]);
386 node2.assert_transports(vec![node1.did(), node3.did()]);
387 node3.assert_transports(vec![node1.did(), node2.did()]);
388 assert_eq!(node1.dht().successors().list()?, vec![
389 node2.did(),
390 node3.did()
391 ]);
392 assert_eq!(node2.dht().successors().list()?, vec![
393 node3.did(),
394 node1.did()
395 ]);
396 assert_eq!(node3.dht().successors().list()?, vec![
397 node1.did(),
398 node2.did()
399 ]);
400
401 Ok((node1, node2, node3))
402 }
403
404 async fn test_triple_desc_ordered_nodes_connection(
405 key1: SecretKey,
406 key2: SecretKey,
407 key3: SecretKey,
408 ) -> Result<(Node, Node, Node)> {
409 let node1 = prepare_node(key1).await;
410 let node2 = prepare_node(key2).await;
411 let node3 = prepare_node(key3).await;
412
413 println!("========================================");
414 println!("|| now we connect node1 and node2 ||");
415 println!("========================================");
416
417 manually_establish_connection(&node1.swarm, &node2.swarm).await;
418 wait_for_msgs([&node1, &node2, &node3]).await;
419 assert_no_more_msg([&node1, &node2, &node3]).await;
420
421 assert_eq!(node1.dht().successors().list()?, vec![node2.did()]);
422 assert_eq!(node2.dht().successors().list()?, vec![node1.did()]);
423 assert_eq!(node3.dht().successors().list()?, vec![]);
424
425 println!("========================================");
426 println!("|| now we start join node3 to node2 ||");
427 println!("========================================");
428
429 manually_establish_connection(&node3.swarm, &node2.swarm).await;
430 wait_for_msgs([&node1, &node2, &node3]).await;
431 assert_no_more_msg([&node1, &node2, &node3]).await;
432
433 println!("=== Check state before connect via DHT ===");
434 node1.assert_transports(vec![node2.did()]);
435 node2.assert_transports(vec![node1.did(), node3.did()]);
436 node3.assert_transports(vec![node2.did()]);
437 assert_eq!(node1.dht().successors().list()?, vec![node2.did()]);
438 assert_eq!(node2.dht().successors().list()?, vec![
439 node1.did(),
440 node3.did()
441 ]);
442 assert_eq!(node3.dht().successors().list()?, vec![node2.did()]);
443
444 println!("=============================================");
445 println!("|| now we connect node1 to node3 via DHT ||");
446 println!("=============================================");
447
448 assert!(node1.swarm.transport.get_connection(node3.did()).is_none());
450 assert_eq!(node1.dht().successors().max()?, node2.did());
452
453 node1.swarm.connect(node3.did()).await?;
454 wait_for_msgs([&node1, &node2, &node3]).await;
455 assert_no_more_msg([&node1, &node2, &node3]).await;
456
457 println!("=== Check state after connect via DHT ===");
458 node1.assert_transports(vec![node2.did(), node3.did()]);
459 node2.assert_transports(vec![node1.did(), node3.did()]);
460 node3.assert_transports(vec![node1.did(), node2.did()]);
461 assert_eq!(node1.dht().successors().list()?, vec![
462 node3.did(),
463 node2.did()
464 ]);
465 assert_eq!(node2.dht().successors().list()?, vec![
466 node1.did(),
467 node3.did()
468 ]);
469 assert_eq!(node3.dht().successors().list()?, vec![
470 node2.did(),
471 node1.did()
472 ]);
473
474 Ok((node1, node2, node3))
475 }
476
477 #[tokio::test]
478 async fn test_fourth_node_connection() -> Result<()> {
479 let keys = gen_ordered_keys(4);
480 let (key1, key2, key3, key4) = (keys[0], keys[1], keys[2], keys[3]);
481 let (node1, node2, node3) = test_triple_ordered_nodes_connection(key1, key2, key3).await?;
482 let node4 = prepare_node(key4).await;
487
488 manually_establish_connection(&node4.swarm, &node2.swarm).await;
497 wait_until("node4 joined: DHT successors converged", || {
501 Ok(
502 node1.dht().successors().list()? == vec![node2.did(), node3.did(), node4.did()]
503 && node2.dht().successors().list()?
504 == vec![node3.did(), node4.did(), node1.did()]
505 && node3.dht().successors().list()? == vec![node1.did(), node2.did()]
506 && node4.dht().successors().list()? == vec![node1.did(), node2.did()],
507 )
508 })
509 .await?;
510
511 println!("=== Check state before connect via DHT ===");
512 node1.assert_transports(vec![node2.did(), node3.did(), node4.did()]);
513 node2.assert_transports(vec![node3.did(), node4.did(), node1.did()]);
514 node3.assert_transports(vec![node1.did(), node2.did()]);
515 node4.assert_transports(vec![node1.did(), node2.did()]);
517 assert_eq!(node1.dht().successors().list()?, vec![
518 node2.did(),
519 node3.did(),
520 node4.did(),
521 ]);
522 assert_eq!(node2.dht().successors().list()?, vec![
523 node3.did(),
524 node4.did(),
525 node1.did(),
526 ]);
527 assert_eq!(node3.dht().successors().list()?, vec![
528 node1.did(),
529 node2.did(),
530 ]);
531 assert_eq!(node4.dht().successors().list()?, vec![
532 node1.did(),
533 node2.did(),
534 ]);
535
536 println!("========================================");
537 println!("| test node4 connect node3 via dht |");
538 println!("========================================");
539 println!(
540 "node1.did(): {:?}, node2.did(): {:?}, node3.did(): {:?}, node4.did(): {:?}",
541 node1.did(),
542 node2.did(),
543 node3.did(),
544 node4.did(),
545 );
546 println!("==================================================");
547
548 node4.swarm.connect(node3.did()).await?;
549 wait_until("node4 connected node3: DHT successors converged", || {
552 Ok(
553 node1.dht().successors().list()? == vec![node2.did(), node3.did(), node4.did()]
554 && node2.dht().successors().list()?
555 == vec![node3.did(), node4.did(), node1.did()]
556 && node3.dht().successors().list()?
557 == vec![node4.did(), node1.did(), node2.did()]
558 && node4.dht().successors().list()?
559 == vec![node1.did(), node2.did(), node3.did()],
560 )
561 })
562 .await?;
563
564 println!("=== Check state after connect via DHT ===");
565 node1.assert_transports(vec![node2.did(), node3.did(), node4.did()]);
566 node2.assert_transports(vec![node3.did(), node4.did(), node1.did()]);
567 node3.assert_transports(vec![node4.did(), node1.did(), node2.did()]);
568 node4.assert_transports(vec![node1.did(), node2.did(), node3.did()]);
569 assert_eq!(node1.dht().successors().list()?, vec![
570 node2.did(),
571 node3.did(),
572 node4.did()
573 ]);
574 assert_eq!(node2.dht().successors().list()?, vec![
575 node3.did(),
576 node4.did(),
577 node1.did(),
578 ]);
579 assert_eq!(node3.dht().successors().list()?, vec![
580 node4.did(),
581 node1.did(),
582 node2.did(),
583 ]);
584 assert_eq!(node4.dht().successors().list()?, vec![
585 node1.did(),
586 node2.did(),
587 node3.did(),
588 ]);
589
590 Ok(())
591 }
592
593 async fn wait_until(msg: &str, mut cond: impl FnMut() -> Result<bool>) -> Result<()> {
601 for _ in 0..300 {
602 if cond()? {
603 return Ok(());
604 }
605 sleep(Duration::from_millis(200)).await;
606 }
607 Err(Error::InvalidMessage(format!("timeout waiting for: {msg}")))
608 }
609
610 #[tokio::test]
611 async fn test_finger_when_disconnect() -> Result<()> {
612 let key1 = SecretKey::random();
613 let key2 = SecretKey::random();
614
615 let node1 = prepare_node(key1).await;
616 let node2 = prepare_node(key2).await;
617
618 {
619 assert!(node1.dht().lock_finger()?.is_empty());
620 assert!(node1.dht().lock_finger()?.is_empty());
621 }
622
623 manually_establish_connection(&node1.swarm, &node2.swarm).await;
624
625 wait_until("node1 and node2 to join each other's DHT", || {
629 let finger1 = node1.dht().lock_finger()?.clone().clone_finger();
630 let finger2 = node2.dht().lock_finger()?.clone().clone_finger();
631 Ok(finger1.into_iter().any(|x| x == Some(node2.did()))
632 && finger2.into_iter().any(|x| x == Some(node1.did())))
633 })
634 .await?;
635
636 node1.assert_transports(vec![node2.did()]);
637 node2.assert_transports(vec![node1.did()]);
638
639 println!("===================================");
640 println!("| test disconnect node1 and node2 |");
641 println!("===================================");
642 node1.swarm.disconnect(node2.did()).await?;
643
644 wait_until("both sides to drop the connection", || {
648 Ok(node1.swarm.transport.get_connection(node2.did()).is_none()
649 && node2.swarm.transport.get_connection(node1.did()).is_none())
650 })
651 .await?;
652
653 node1.assert_transports(vec![]);
654 node2.assert_transports(vec![]);
655 {
656 let finger1 = node1.dht().lock_finger()?.clone().clone_finger();
657 let finger2 = node2.dht().lock_finger()?.clone().clone_finger();
658 assert!(finger1.into_iter().all(|x| x.is_none()));
659 assert!(finger2.into_iter().all(|x| x.is_none()));
660 }
661
662 Ok(())
663 }
664}