syncweb_core/node/
discovery.rs

1use distributed_topic_tracker::{
2    AutoDiscoveryGossip, Config, DefaultSecretRotation, RecordPublisher, RotationHandle, Topic, TopicId,
3};
4use ed25519_dalek::SigningKey;
5use iroh::{Endpoint, EndpointAddr, PublicKey, TransportAddr};
6use iroh_base::CustomAddr;
7use iroh_docs::NamespaceId;
8use iroh_gossip::net::Gossip;
9use serde::{Deserialize, Serialize};
10use std::{collections::HashMap, sync::Arc, time::Duration};
11use tokio::sync::Mutex;
12
13use crate::error::{Result, SyncwebError};
14
15/// How often a node republishes its address presence record and refreshes its
16/// view of peers for the DHT topic.
17///
18/// The tracker's bootstrap records only carry node ids, so a peer that learned
19/// another node out of band (a node-id-only ticket, the DHT itself, or a prior
20/// connection) has no address to dial. Each node therefore also publishes a
21/// presence record carrying its full [`EndpointAddr`] (direct addresses, iroh
22/// relay, and the Syncthing relay custom address) and, in the same background
23/// loop, reads other nodes' presence records so `find_peers` can return
24/// resolved addresses instead of bare node ids.
25const PRESENCE_INTERVAL: Duration = Duration::from_secs(10);
26
27/// Marker distinguishing a presence record from the tracker's own bootstrap
28/// records, which carry a fixed `GossipRecordContent` layout. A peer ignores
29/// any record whose tag does not match.
30const PRESENCE_TAG: [u8; 8] = *b"syncweb!";
31
32/// A peer address announcement stored in the namespace's DHT topic.
33#[derive(Clone, Debug, Serialize, Deserialize)]
34struct PresenceRecord {
35    tag: [u8; 8],
36    node_id: [u8; 32],
37    addrs: Vec<TransportAddr>,
38}
39
40/// A joined topic and the record publishers used to read/write the DHT.
41#[derive(Clone)]
42struct TopicEntry {
43    topic: Topic,
44    bootstrap: RecordPublisher,
45    presence: RecordPublisher,
46}
47
48/// Peer addresses discovered per topic, keyed by topic hash.
49type PeerCache = Arc<Mutex<HashMap<[u8; 32], HashMap<PublicKey, EndpointAddr>>>>;
50
51#[derive(Clone)]
52pub struct TopicTracker {
53    gossip: Gossip,
54    topics: Arc<Mutex<HashMap<[u8; 32], TopicEntry>>>,
55    peers: PeerCache,
56    endpoint: Endpoint,
57    relay_addr: Option<CustomAddr>,
58}
59
60impl TopicTracker {
61    #[must_use]
62    pub fn new(gossip: &Gossip, endpoint: &Endpoint, relay_addr: Option<CustomAddr>) -> Self {
63        Self {
64            gossip: gossip.clone(),
65            topics: Arc::new(Mutex::new(HashMap::new())),
66            peers: Arc::new(Mutex::new(HashMap::new())),
67            endpoint: endpoint.clone(),
68            relay_addr,
69        }
70    }
71
72    /// # Errors
73    ///
74    /// Returns an error if joining the topic fails.
75    pub async fn announce(&self, namespace_id: NamespaceId) -> Result<()> {
76        self.topic(namespace_id).await?;
77        Ok(())
78    }
79
80    /// Return the peers discovered for a namespace so far, with any addresses
81    /// that have been resolved via the DHT presence records. This is
82    /// deliberately non-blocking: the DHT reads happen in the background loop,
83    /// so this call never stalls a daemon task on network latency.
84    ///
85    /// # Errors
86    ///
87    /// Returns an error if the topic cannot be joined or gossip neighbors cannot be retrieved.
88    pub async fn find_peers(&self, namespace_id: NamespaceId) -> Result<Vec<EndpointAddr>> {
89        let entry = self.topic(namespace_id).await?;
90        let key = *namespace_id.as_bytes();
91        let own_id = self.endpoint.id();
92        let mut peers: HashMap<PublicKey, EndpointAddr> =
93            self.peers.lock().await.get(&key).cloned().unwrap_or_default();
94
95        // Include currently-connected gossip neighbors (as bare node ids) so a
96        // blob fetch or sync that runs before the background collector's first
97        // pass can still reach already-connected peers.
98        let receiver = entry
99            .topic
100            .gossip_receiver()
101            .await
102            .map_err(|error| SyncwebError::operation("failed to get discovery receiver", error))?;
103        let neighbors = receiver
104            .neighbors()
105            .await
106            .map_err(|error| SyncwebError::operation("failed to find discovery peers", error))?;
107        for neighbor in neighbors {
108            if neighbor != own_id {
109                peers.entry(neighbor).or_insert_with(|| EndpointAddr::new(neighbor));
110            }
111        }
112
113        Ok(peers.into_values().collect())
114    }
115
116    async fn topic(&self, namespace_id: NamespaceId) -> Result<TopicEntry> {
117        let key = *namespace_id.as_bytes();
118        let existing = self.topics.lock().await.get(&key).cloned();
119        if let Some(entry) = existing {
120            return Ok(entry);
121        }
122
123        let topic_id = TopicId::from_hash(&key);
124        let secret = namespace_id.as_bytes().to_vec();
125        let node_secret = self.endpoint.secret_key().to_bytes();
126        let signing_key = SigningKey::from_bytes(&node_secret);
127        let bootstrap = RecordPublisher::builder(topic_id.clone(), signing_key, secret.clone())
128            .config(Config::default())
129            .secret_rotation(RotationHandle::new(DefaultSecretRotation))
130            .build();
131        let topic = self
132            .gossip
133            .subscribe_and_join_with_auto_discovery_no_wait(bootstrap.clone())
134            .await
135            .map_err(|error| SyncwebError::operation("failed to join discovery topic", error))?;
136        let presence = RecordPublisher::builder(topic_id, presence_signing_key(&node_secret), secret)
137            .config(Config::default())
138            .build();
139
140        let entry = TopicEntry {
141            topic,
142            bootstrap,
143            presence,
144        };
145        self.topics.lock().await.insert(key, entry.clone());
146        Self::spawn_discovery_loop(
147            entry.clone(),
148            self.endpoint.clone(),
149            self.relay_addr.clone(),
150            self.peers.clone(),
151            key,
152        );
153        Ok(entry)
154    }
155
156    fn spawn_discovery_loop(
157        entry: TopicEntry,
158        endpoint: Endpoint,
159        relay_addr: Option<CustomAddr>,
160        peers: PeerCache,
161        topic_key: [u8; 32],
162    ) {
163        tokio::spawn(async move {
164            let cancel = tokio_util::sync::CancellationToken::new();
165            let mut ticker = tokio::time::interval(PRESENCE_INTERVAL);
166            ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
167            let own_id = endpoint.id();
168            loop {
169                ticker.tick().await;
170                publish_presence(&entry, &endpoint, relay_addr.as_ref(), cancel.clone()).await;
171                let discovered = collect_peers(&entry, own_id, cancel.clone()).await;
172                peers.lock().await.insert(topic_key, discovered);
173            }
174        });
175    }
176}
177
178/// Derive a stable signing key for presence records from the node secret so
179/// presence records do not collide with the tracker's bootstrap records in the
180/// DHT's per-publisher dedup.
181fn presence_signing_key(node_secret: &[u8; 32]) -> SigningKey {
182    use sha2::Digest;
183    let mut hasher = sha2::Sha256::new();
184    hasher.update(b"syncweb-presence");
185    hasher.update(node_secret);
186    let seed: [u8; 32] = hasher.finalize().into();
187    SigningKey::from_bytes(&seed)
188}
189
190/// The addresses to advertise: the endpoint's current direct and relay
191/// addresses plus the Syncthing relay custom address when relaying is enabled.
192fn presence_addrs(endpoint: &Endpoint, relay_addr: Option<&CustomAddr>) -> Vec<TransportAddr> {
193    let mut addrs: Vec<TransportAddr> = endpoint.addr().addrs.into_iter().collect();
194    if let Some(relay) = relay_addr {
195        addrs.push(TransportAddr::Custom(relay.clone()));
196    }
197    addrs
198}
199
200/// Publish this node's address presence record for the current minute.
201async fn publish_presence(
202    entry: &TopicEntry,
203    endpoint: &Endpoint,
204    relay_addr: Option<&CustomAddr>,
205    cancel: tokio_util::sync::CancellationToken,
206) {
207    let content = PresenceRecord {
208        tag: PRESENCE_TAG,
209        node_id: *endpoint.id().as_bytes(),
210        addrs: presence_addrs(endpoint, relay_addr),
211    };
212    let Ok(record) = entry
213        .presence
214        .new_record(distributed_topic_tracker::unix_minute(0), content)
215    else {
216        return;
217    };
218    if let Err(error) = entry.presence.publish_record(record, cancel).await {
219        tracing::debug!(%error, "failed to publish discovery presence record");
220    }
221}
222
223/// Read peer addresses (presence records and node-id-only bootstrap records)
224/// and gossip neighbors for the current and previous minute.
225async fn collect_peers(
226    entry: &TopicEntry,
227    own_id: PublicKey,
228    cancel: tokio_util::sync::CancellationToken,
229) -> HashMap<PublicKey, EndpointAddr> {
230    let mut peers: HashMap<PublicKey, EndpointAddr> = HashMap::new();
231    let now = distributed_topic_tracker::unix_minute(0);
232    for minute in [now.saturating_sub(1), now] {
233        let records = entry
234            .bootstrap
235            .get_records(minute, cancel.clone())
236            .await
237            .unwrap_or_default();
238        for record in records {
239            if let Ok(presence) = record.content::<PresenceRecord>()
240                && presence.tag == PRESENCE_TAG
241            {
242                if let Some(node_id) = PublicKey::from_bytes(&presence.node_id).ok()
243                    && node_id != own_id
244                {
245                    peers
246                        .entry(node_id)
247                        .or_insert_with(|| EndpointAddr::from_parts(node_id, presence.addrs.clone()));
248                }
249                continue;
250            }
251            if let Some(node_id) = PublicKey::from_bytes(&record.pub_key()).ok()
252                && node_id != own_id
253            {
254                peers.entry(node_id).or_insert_with(|| EndpointAddr::new(node_id));
255            }
256        }
257    }
258
259    if let Ok(receiver) = entry.topic.gossip_receiver().await
260        && let Ok(neighbors) = receiver
261            .neighbors()
262            .await
263            .map_err(|error| SyncwebError::operation("failed to find discovery peers", error))
264    {
265        for neighbor in neighbors {
266            if neighbor != own_id {
267                peers.entry(neighbor).or_insert_with(|| EndpointAddr::new(neighbor));
268            }
269        }
270    }
271
272    peers
273}
syncweb_core/node/discovery.rs

syncweb_core/node/
discovery.rs
