Skip to main content

rings_core/message/handlers/
connection.rs

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/// QueryForTopoInfoSend is direct message
26#[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/// Try join received node into DHT after received from TopoInfo.
44#[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                // Establish stabilization-learned candidates first so the
58                // resulting Notify/Query actions can usually send immediately.
59                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    //! tests
203    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    // node1.key < node2.key < node3.key
217    //
218    // Firstly, we connect node1 to node2, node2 to node3.
219    // Then, we connect node1 to node3 via DHT.
220    //
221    // After full connected, the topological structure should be:
222    //
223    // Node1 ------------ Node2 ------------ Node3
224    //   |-------------------------------------|
225    //
226    // --------- Connect node1 and node2
227    // 0. Node1 and node2 will set each other as their successor in DHTJoin handler.
228    //
229    // 1. Node1 send FindSuccessorSend(node1) to node2.
230    //    Meanwhile, node2 send FindSuccessorSend(node2) to node1.
231    //
232    // 2. Node1 respond by sending FindSuccessorReport(node2) to node2.
233    //    Meanwhile, node2 respond by sending FindSuccessorReport(node1) to node1.
234    //    But no node should update local successor by those reports.
235    //
236    // --------- Join node3 to node2
237    // 0. Node2 and node3 will set each other as their successor in DHTJoin handler.
238    //
239    // 1. Node3 send FindSuccessorSend(node3) to node2.
240    //    Meanwhile, node2 send FindSuccessorSend(node2) to node3.
241    //
242    // 2. Node3 respond by sending FindSuccessorReport(node2) to node2.
243    //    Meanwhile, node2 respond by sending FindSuccessorReport(node3) to node3.
244    //    But no node should update local successor by those reports.
245    //
246    // --------- Connect node1 to node3 via DHT
247    // 0. After checking finger table locally, node1 pick node2 to send ConnectNodeSend(node3).
248    //
249    // 1. Node2 relay ConnectNodeSend(node3) to node3.
250    //
251    // 2. Node3 respond by sending ConnectNodeReport(node1) to node2.
252    //
253    // 3. Node2 relay ConnectNodeReport(node1) to node1.
254    //
255    // --------- Communications after successful connection
256    //
257    #[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    // The 2_3_1 should have same behavior as 1_2_3 since they are all clockwise.
266    #[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    // The 3_1_2 should have same behavior as 1_2_3 since they are all clockwise.
275    #[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    // node1.key > node2.key > node3.key
284    //
285    // All the processes are the same as test_triple_nodes_1_2_3. Except the following:
286    //
287    // --------- Join node3 to node2
288    // 0. Node3 will set node2 as successor in DHTJoin handler.
289    //
290    //    Node2 will not set node3 as successor in DHTJoin handler.
291    //    Because node2.processor.max() is node1, and node1.bias(node1) < node1.bias(node3).
292    //    That means node1 is closer to node2 than node3 on the clock circle.
293    //
294    // 1. Node3 send FindSuccessorSend(node3) to node2. Node2 relay it to Node1.
295    //    Meanwhile, node2 send FindSuccessorSend(node2) to node3.
296    //
297    // 2. Node3 respond by sending FindSuccessorReport(node2) to node2.
298    //    Meanwhile, node1 respond by sending FindSuccessorReport(node2) to node3 through node2.
299    //
300    // --------- Communications after successful connection
301    //
302    #[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    // The 2_1_3 should have same behavior as 3_2_1 since they are all anti-clockwise.
311    #[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    // The 1_3_2 should have same behavior as 3_2_1 since they are all anti-clockwise.
320    #[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        // check node1 and node3 is not connected to each other
376        assert!(node1.swarm.transport.get_connection(node3.did()).is_none());
377        // node1's successor should be node2 now
378        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        // check node1 and node3 is not connected to each other
449        assert!(node1.swarm.transport.get_connection(node3.did()).is_none());
450        // node1's successor should be node2 now
451        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        // we now have three connected nodes
483        // node1 -> node2 -> node3
484        //  |-<-----<---------<--|
485
486        let node4 = prepare_node(key4).await;
487
488        // Unless we use a fixed did value, we cannot fully predict the communication order between node4 and the nodes,
489        // because we do not know the distance between node4 and each node.
490        //
491        // Therefore, here we only guarantee that messages can be processed correctly without checking the specific message order.
492        //
493        // In addition, we check the final state to ensure the entire process meets expectations.
494
495        // connect node4 to node2
496        manually_establish_connection(&node4.swarm, &node2.swarm).await;
497        // Poll for convergence rather than sleeping a fixed amount: under the
498        // release-LTO CI run with native WebRTC, 6s is not always enough and the
499        // assertions below would flake. The expected final state is unchanged.
500        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 will connect node1 after connecting node2, because node2 notified node4 that node1 is its predecessor.
516        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        // Same as above: poll for the post-connect converged state instead of a
550        // fixed 6s sleep so the test is robust under CI contention.
551        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    /// Poll `cond` every 200ms until it returns true, failing after ~60s.
594    /// Used instead of fixed sleeps so the test is deterministic regardless of
595    /// how long the WebRTC handshake/teardown takes on a given machine.
596    ///
597    /// The window is generous on purpose: ICE paces connectivity checks at
598    /// ~200ms each, so on a host with many network interfaces (lots of
599    /// candidate pairs) establishing the connection can legitimately take ~20s.
600    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        // The data channels open and `on_data_channel_open -> join_dht` runs
626        // asynchronously, so poll until both sides have joined each other rather
627        // than asserting after a fixed wait.
628        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        // node1 closes locally; node2 learns via the data channel closing and
645        // tears its side down promptly (without waiting for the ICE `Failed`
646        // timeout). Poll until both sides have removed the connection.
647        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}