1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
use super::*;
impl NostrDiscovery {
pub(super) fn spawn_notify_loop(
self: Arc<Self>,
mut notifications: broadcast::Receiver<RelayPoolNotification>,
) -> JoinHandle<()> {
tokio::spawn(async move {
let started_at = Instant::now();
let mut first_event_seen = false;
info!("nostr notify loop entered");
loop {
let notification = match notifications.recv().await {
Ok(notification) => notification,
Err(broadcast::error::RecvError::Lagged(skipped)) => {
warn!(
skipped,
"nostr notification channel lagged; advert/signal events dropped"
);
continue;
}
Err(broadcast::error::RecvError::Closed) => {
warn!("nostr notification channel closed; notify loop exiting");
break;
}
};
if !first_event_seen {
first_event_seen = true;
info!(
elapsed_ms = started_at.elapsed().as_millis() as u64,
"nostr notify loop received first event"
);
}
if let RelayPoolNotification::Event { event, .. } = notification {
if event.kind == Kind::Custom(ADVERT_KIND) {
let Ok(verified_event) = VerifiedEvent::try_from(event.as_ref()) else {
continue;
};
let author_key = NostrPeerKey::from_public_key_ref(verified_event.pubkey());
let author_npub = verified_event.pubkey().to_bech32().expect("infallible");
if let Some(valid_until_ms) = self.event_valid_until_ms(&event)
&& let Ok(advert) =
Self::parse_overlay_advert_event(verified_event, &self.config.app)
{
let mut cache = self.advert_cache.write().await;
let should_replace = cache
.get(&author_key)
.map(|existing| existing.created_at <= event.created_at.as_secs())
.unwrap_or(true);
if should_replace && author_key != self.self_peer_key() {
debug!(
peer = %short_npub(&author_npub),
endpoints = %endpoint_summary(&advert.endpoints),
event = %short_id(&event.id.to_string()),
"advert: peer cached"
);
}
if should_replace {
cache.insert(
author_key,
CachedOverlayAdvert {
author_npub,
advert,
created_at: event.created_at.as_secs(),
valid_until_ms,
},
);
}
}
self.prune_advert_cache().await;
continue;
}
if event.kind != Kind::Custom(SIGNAL_KIND) {
continue;
}
let unwrapped = match unwrap_signal_event(&self.keys, &event).await {
Ok(unwrapped) => unwrapped,
Err(err) => {
trace!(error = %err, "failed to unwrap traversal signal");
continue;
}
};
let sender_npub = match unwrapped.sender.to_bech32() {
Ok(npub) => npub,
Err(err) => {
debug!(error = %err, "failed to encode traversal sender npub");
continue;
}
};
if let Ok(answer) =
serde_json::from_str::<TraversalAnswer>(&unwrapped.rumor.content)
&& answer.message_type == "answer"
&& answer.recipient_npub == self.npub
{
if let Some(tx) = self
.pending_answers
.lock()
.await
.remove(&answer.in_reply_to)
{
let _ = tx.send(SignalEnvelope {
payload: answer,
event_id: Some(event.id),
sender_npub: sender_npub.clone(),
});
}
continue;
}
if let Ok(offer) =
serde_json::from_str::<TraversalOffer>(&unwrapped.rumor.content)
&& offer.message_type == "offer"
&& offer.recipient_npub == self.npub
{
let Ok(permit) = self.offer_slots.clone().try_acquire_owned() else {
warn!(
sender_npub = %sender_npub,
limit = self.config.max_concurrent_incoming_offers,
"rate-limited inbound traversal offer (max_concurrent_incoming_offers reached); offer dropped"
);
continue;
};
let runtime = Arc::clone(&self);
tokio::spawn(async move {
let _permit = permit;
if let Err(err) = runtime
.handle_incoming_offer(offer, unwrapped.sender, sender_npub)
.await
{
debug!(error = %err, "failed to handle traversal offer");
}
});
}
}
}
})
}
}