distributed_topic_tracker/gossip/merge/
bubble.rs

1use std::{collections::HashSet, time::Duration};
2use actor_helper::{Action, Actor, Handle};
3
4use crate::{gossip::GossipRecordContent, GossipReceiver, GossipSender, RecordPublisher};
5use anyhow::Result;
6
7#[derive(Debug, Clone)]
8pub struct BubbleMerge {
9    _api: Handle<BubbleMergeActor>,
10}
11
12#[derive(Debug)]
13struct BubbleMergeActor {
14    rx: tokio::sync::mpsc::Receiver<Action<BubbleMergeActor>>,
15
16    record_publisher: RecordPublisher,
17    gossip_receiver: GossipReceiver,
18    gossip_sender: GossipSender,
19    ticker: tokio::time::Interval,
20}
21
22impl BubbleMerge {
23    pub fn new(
24        record_publisher: RecordPublisher,
25        gossip_sender: GossipSender,
26        gossip_receiver: GossipReceiver,
27    ) -> Result<Self> {
28        let (api, rx) = Handle::channel(32);
29
30        let mut ticker = tokio::time::interval(Duration::from_secs(10));
31        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
32
33        tokio::spawn(async move {
34            let mut actor = BubbleMergeActor {
35                rx,
36                record_publisher,
37                gossip_receiver,
38                gossip_sender,
39                ticker,
40            };
41            let _ = actor.run().await;
42        });
43
44        Ok(Self { _api: api })
45    }
46}
47
48impl Actor for BubbleMergeActor {
49    async fn run(&mut self) -> Result<()> {
50        loop {
51            tokio::select! {
52                Some(action) = self.rx.recv() => {
53                    action(self).await;
54                }
55                _ = self.ticker.tick() => {
56                    let _ = self.merge().await;
57                    self.ticker.reset_after(Duration::from_secs(rand::random::<u64>() % 50));
58                }
59                _ = tokio::signal::ctrl_c() => break,
60            }
61        }
62        Ok(())
63    }
64}
65
66impl BubbleMergeActor {
67    // Cluster size as bubble indicator
68    async fn merge(&mut self) -> Result<()> {
69        let unix_minute = crate::unix_minute(0);
70        let records = self.record_publisher.get_records(unix_minute).await;
71        let neighbors = self.gossip_receiver.neighbors().await;
72        if neighbors.len() < 4 && !records.is_empty() {
73            let node_ids = records
74                .iter()
75                .filter_map(|record| {
76                    if let Ok(content) = record.content::<GossipRecordContent>() {
77                        Some(content.active_peers
78                        .iter()
79                        .filter_map(|&active_peer| {
80                            if active_peer == [0; 32]
81                                || neighbors.contains(&active_peer)
82                                || active_peer.eq(record.node_id().to_vec().as_slice())
83                                || active_peer.eq(self.record_publisher.pub_key().as_bytes())
84                            {
85                                None
86                            } else {
87                                iroh::NodeId::from_bytes(&active_peer).ok()
88                            }
89                        })
90                        .collect::<Vec<_>>())
91                    } else {
92                        None
93                    }
94                })
95                .flatten()
96                .collect::<HashSet<_>>();
97            self.gossip_sender
98                .join_peers(
99                    node_ids.iter().cloned().collect::<Vec<_>>(),
100                    Some(super::MAX_JOIN_PEERS_COUNT),
101                )
102                .await?;
103        }
104        Ok(())
105    }
106}
distributed_topic_tracker/gossip/merge/bubble.rs

distributed_topic_tracker/gossip/merge/
bubble.rs
