Line data Source code
1 : /*
2 : * Copyright (C) 2004-2026 Savoir-faire Linux Inc.
3 : *
4 : * This program is free software: you can redistribute it and/or modify
5 : * it under the terms of the GNU General Public License as published by
6 : * the Free Software Foundation, either version 3 of the License, or
7 : * (at your option) any later version.
8 : *
9 : * This program is distributed in the hope that it will be useful,
10 : * but WITHOUT ANY WARRANTY; without even the implied warranty of
11 : * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 : * GNU General Public License for more details.
13 : *
14 : * You should have received a copy of the GNU General Public License
15 : * along with this program. If not, see <https://www.gnu.org/licenses/>.
16 : */
17 :
18 : #include "swarm_manager.h"
19 : #include "jamidht/timestamp.h"
20 : #include <dhtnet/multiplexed_socket.h>
21 : #include <dhtnet/channel_utils.h>
22 : #include <opendht/thread_pool.h>
23 :
24 : namespace jami {
25 :
26 : using namespace swarm_protocol;
27 :
28 : static bool
29 11 : isNewerLease(const MobileLease& candidate, const MobileLease& reference)
30 : {
31 11 : return candidate.expires_at > reference.expires_at
32 11 : || (candidate.expires_at == reference.expires_at && candidate.issued_at > reference.issued_at);
33 : }
34 :
35 1184 : SwarmManager::SwarmManager(const NodeId& id,
36 : bool isMobile,
37 : const std::mt19937_64& rand,
38 : ToConnectCb&& toConnectCb,
39 : std::string conversationId,
40 : MobileLeaseProvider mobileLeaseProvider,
41 : MobileLeaseIssuerValidator mobileLeaseIssuerValidator,
42 : CertificateProvider certificateProvider,
43 1184 : CertificateFetcher certificateFetcher)
44 1184 : : id_(id)
45 1184 : , isMobile_(isMobile)
46 1184 : , conversationId_(std::move(conversationId))
47 1184 : , rd(rand)
48 1184 : , mobileLeaseProvider_(std::move(mobileLeaseProvider))
49 1184 : , mobileLeaseIssuerValidator_(std::move(mobileLeaseIssuerValidator))
50 1184 : , certificateProvider_(std::move(certificateProvider))
51 1184 : , certificateFetcher_(std::move(certificateFetcher))
52 3552 : , toConnectCb_(toConnectCb)
53 : {
54 1184 : routing_table.setId(id);
55 1184 : }
56 :
57 2368 : SwarmManager::~SwarmManager()
58 : {
59 1184 : if (!isShutdown_)
60 280 : shutdown();
61 1184 : }
62 :
63 : bool
64 4040 : SwarmManager::setKnownNodes(const std::vector<NodeId>& known_nodes)
65 : {
66 4040 : isShutdown_ = false;
67 4040 : std::vector<NodeId> newNodes;
68 : {
69 4040 : std::lock_guard lock(mutex);
70 8381 : for (const auto& nodeId : known_nodes) {
71 4341 : if (addKnownNode(nodeId)) {
72 1147 : newNodes.emplace_back(nodeId);
73 : }
74 : }
75 4040 : }
76 :
77 4040 : if (newNodes.empty()) {
78 : // Members coming back online are re-announced with the devices we already
79 : // know, so addKnownNode() reports nothing new and the block below is
80 : // skipped. When the routing table holds no connected node, that announce
81 : // is the only chance we get to retry: nothing else calls maintainBuckets()
82 : // and the swarm stays silent until the account is reloaded. Retry the
83 : // candidates we already have instead of dropping the notification.
84 3056 : if (isConnected())
85 1616 : return false;
86 1440 : dht::ThreadPool::io().run([w = weak()] {
87 1440 : if (auto shared = w.lock())
88 1440 : shared->maintainBuckets();
89 1440 : });
90 1440 : return false;
91 : }
92 :
93 984 : dht::ThreadPool::io().run([w = weak(), newNodes = std::move(newNodes)] {
94 984 : auto shared = w.lock();
95 984 : if (!shared)
96 0 : return;
97 : // If we detect a new node which already got a TCP link
98 : // we can use it to speed-up the bootstrap (because opening
99 : // a new channel will be easy)
100 984 : std::set<NodeId> toConnect;
101 2131 : for (const auto& nodeId : newNodes) {
102 1147 : if (shared->toConnectCb_ && shared->toConnectCb_(nodeId))
103 410 : toConnect.emplace(nodeId);
104 : }
105 984 : shared->maintainBuckets(toConnect);
106 984 : });
107 984 : return true;
108 4040 : }
109 :
110 : void
111 643 : SwarmManager::setMobileNodes(const std::vector<NodeId>& mobile_nodes)
112 : {
113 643 : bool changed = false;
114 : {
115 643 : std::lock_guard lock(mutex);
116 643 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
117 643 : if (!conversationId_.empty() && now >= LEGACY_MOBILE_NODE_SUNSET)
118 1 : return;
119 843 : for (const auto& nodeId : mobile_nodes) {
120 201 : changed |= addMobileNodes(nodeId);
121 201 : if (!conversationId_.empty() && !mobileNodeLeases_.contains(nodeId))
122 0 : changed |= legacyMobileNodeExpiries_.try_emplace(nodeId, LEGACY_MOBILE_NODE_SUNSET).second;
123 : }
124 642 : scheduleMobileLeaseExpiryInternal();
125 643 : }
126 642 : if (changed)
127 21 : emitMobileNodesChanged();
128 : }
129 :
130 : void
131 16 : SwarmManager::setMobileNodes(const std::vector<MobileNodeInfo>& mobile_nodes, bool requireLease)
132 : {
133 16 : bool changed = false;
134 16 : size_t records = 0;
135 34 : for (const auto& mobile : mobile_nodes) {
136 18 : if (records++ == MAX_MOBILE_NODE_INFOS)
137 0 : break;
138 18 : changed |= setMobileNodeInfo(mobile, requireLease);
139 : }
140 16 : if (changed)
141 6 : emitMobileNodesChanged();
142 16 : }
143 :
144 : bool
145 26 : SwarmManager::setMobileNodeInfo(const MobileNodeInfo& mobile,
146 : bool requireLease,
147 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& source)
148 : {
149 26 : if (mobile.id == id_)
150 0 : return false;
151 :
152 26 : if (!mobile.lease) {
153 6 : if (requireLease)
154 1 : return false;
155 : const auto now = static_cast<uint64_t>(
156 5 : std::chrono::duration_cast<std::chrono::seconds>(std::chrono::system_clock::now().time_since_epoch())
157 5 : .count());
158 5 : if (!conversationId_.empty() && now >= LEGACY_MOBILE_NODE_SUNSET)
159 1 : return false;
160 4 : std::lock_guard lock(mutex);
161 4 : bool changed = addMobileNodes(mobile.id);
162 4 : if (!conversationId_.empty() && !mobileNodeLeases_.contains(mobile.id)) {
163 0 : changed |= legacyMobileNodeExpiries_.try_emplace(mobile.id, LEGACY_MOBILE_NODE_SUNSET).second;
164 0 : scheduleMobileLeaseExpiryInternal();
165 : }
166 4 : return changed;
167 4 : }
168 :
169 20 : const auto& lease = *mobile.lease;
170 20 : if (lease.device_id != mobile.id || !precheckLease(lease))
171 2 : return false;
172 :
173 : {
174 : // Gossip re-announces the same lease every round: skip the whole
175 : // resolution when what we already verified is at least as good.
176 18 : std::lock_guard lock(mutex);
177 18 : auto known = mobileNodeLeases_.find(mobile.id);
178 18 : if (known != mobileNodeLeases_.end() && !isNewerLease(lease, known->second))
179 1 : return false;
180 18 : }
181 :
182 : // The certificate is never gossiped: resolve it from the account certificate
183 : // store, which already holds every device we ever connected to (the swarm
184 : // channel pins its TLS peer certificate) and everything resolved before.
185 17 : if (certificateProvider_) {
186 12 : if (auto certificate = certificateProvider_(mobile.id)) {
187 6 : if (!verifyLease(*certificate, lease))
188 1 : return false;
189 5 : std::lock_guard lock(mutex);
190 5 : return commitLeaseInternal(lease);
191 17 : }
192 : }
193 :
194 11 : std::lock_guard lock(mutex);
195 11 : enqueuePendingLeaseInternal(lease, source);
196 11 : return false;
197 11 : }
198 :
199 : void
200 3459 : SwarmManager::addChannel(const std::shared_ptr<dhtnet::ChannelSocketInterface>& channel)
201 : {
202 : // JAMI_WARNING("[SwarmManager {}] addChannel! with {}", fmt::ptr(this), channel->deviceId().to_view());
203 3459 : if (channel) {
204 3459 : auto emit = false;
205 3459 : auto added = false;
206 : {
207 3459 : std::lock_guard lock(mutex);
208 3459 : emit = routing_table.isEmpty();
209 3459 : auto bucket = routing_table.findBucket(channel->deviceId());
210 3459 : added = routing_table.addNode(channel, bucket);
211 3459 : }
212 3459 : if (added) {
213 2093 : std::error_code ec;
214 2093 : resetNodeExpiry(ec, channel, id_);
215 : }
216 3459 : receiveMessage(channel);
217 3459 : if (emit && onConnectionChanged_) {
218 : // If it's the first channel we add, we're now connected!
219 763 : JAMI_DEBUG("[SwarmManager {}] Bootstrap: Connected!", fmt::ptr(this));
220 763 : onConnectionChanged_(true);
221 : }
222 : }
223 3459 : }
224 :
225 : void
226 1100 : SwarmManager::removeNode(const NodeId& nodeId)
227 : {
228 1100 : std::unique_lock lk(mutex);
229 1100 : if (isConnectedWith(nodeId)) {
230 968 : removeNodeInternal(nodeId);
231 968 : lk.unlock();
232 968 : maintainBuckets();
233 : }
234 1100 : }
235 :
236 : void
237 255 : SwarmManager::changeMobility(const NodeId& nodeId, bool isMobile)
238 : {
239 : {
240 255 : std::lock_guard lock(mutex);
241 255 : auto bucket = routing_table.findBucket(nodeId);
242 255 : bucket->changeMobility(nodeId, isMobile);
243 255 : }
244 255 : emitMobileNodesChanged();
245 255 : }
246 :
247 : bool
248 1778 : SwarmManager::isConnectedWith(const NodeId& deviceId)
249 : {
250 1778 : return routing_table.hasNode(deviceId);
251 : }
252 :
253 : void
254 1282 : SwarmManager::shutdown()
255 : {
256 1282 : if (isShutdown_) {
257 36 : return;
258 : }
259 1246 : isShutdown_ = true;
260 1246 : std::lock_guard lock(mutex);
261 1246 : mobileLeaseExpiryTimer_.cancel();
262 1246 : for (auto& [peer, state] : outstandingCertRequests_)
263 0 : if (state.timer)
264 0 : state.timer->cancel();
265 1246 : outstandingCertRequests_.clear();
266 1246 : routing_table.shutdownAllNodes();
267 1246 : }
268 :
269 : void
270 34 : SwarmManager::restart()
271 : {
272 34 : isShutdown_ = false;
273 34 : std::lock_guard lock(mutex);
274 34 : scheduleMobileLeaseExpiryInternal();
275 34 : }
276 :
277 : bool
278 5014 : SwarmManager::addKnownNode(const NodeId& nodeId)
279 : {
280 5014 : return routing_table.addKnownNode(nodeId);
281 : }
282 :
283 : bool
284 216 : SwarmManager::addMobileNodes(const NodeId& nodeId)
285 : {
286 216 : if (id_ != nodeId) {
287 214 : return routing_table.addMobileNode(nodeId);
288 : }
289 2 : return false;
290 : }
291 :
292 : bool
293 640 : SwarmManager::isMobileNodeCurrentInternal(const NodeId& nodeId, int64_t now) const
294 : {
295 640 : if (auto lease = mobileNodeLeases_.find(nodeId); lease != mobileNodeLeases_.end())
296 28 : return lease->second.expires_at > now;
297 612 : if (auto legacy = legacyMobileNodeExpiries_.find(nodeId); legacy != legacyMobileNodeExpiries_.end())
298 0 : return legacy->second > now;
299 612 : return conversationId_.empty();
300 : }
301 :
302 : bool
303 34 : SwarmManager::precheckLease(const MobileLease& lease) const
304 : {
305 34 : if (lease.format_version != 1 || lease.conversation_id != conversationId_ || lease.conversation_id.empty()
306 33 : || lease.conversation_id.size() > MAX_MOBILE_LEASE_IDENTIFIER_SIZE || lease.signature.empty()
307 33 : || lease.signature.size() > MAX_MOBILE_LEASE_SIGNATURE_SIZE || !lease.issuer_id
308 68 : || !mobileLeaseIssuerValidator_ || !mobileLeaseIssuerValidator_(lease.issuer_id))
309 3 : return false;
310 :
311 31 : constexpr auto MAX_CLOCK_SKEW = std::chrono::seconds(5 * 60);
312 31 : const auto now = std::chrono::system_clock::now();
313 31 : const auto issued = std::chrono::system_clock::time_point(std::chrono::seconds(lease.issued_at));
314 31 : const auto expires = std::chrono::system_clock::time_point(std::chrono::seconds(lease.expires_at));
315 62 : if (issued > now + MAX_CLOCK_SKEW || expires <= now || expires <= issued
316 62 : || expires - issued > MAX_MOBILE_LEASE_DURATION)
317 0 : return false;
318 31 : return true;
319 : }
320 :
321 : bool
322 13 : SwarmManager::verifyLease(const dht::crypto::Certificate& certificate, const MobileLease& lease) const
323 : {
324 : try {
325 26 : if (certificate.getLongId() != lease.device_id || !certificate.issuer
326 26 : || certificate.issuer->getId() != lease.issuer_id)
327 0 : return false;
328 13 : dht::crypto::TrustList trust;
329 13 : trust.add(*certificate.issuer);
330 13 : if (!trust.verify(certificate))
331 0 : return false;
332 13 : auto certificateExpiry = toSecondsSinceEpoch(certificate.getExpiration());
333 13 : if (lease.expires_at > certificateExpiry)
334 0 : return false;
335 13 : const auto payload = mobileLeasePayload(lease);
336 13 : return certificate.getPublicKey().checkSignature(payload, lease.signature);
337 13 : } catch (const std::exception& e) {
338 0 : JAMI_WARNING("Ignoring invalid mobile lease for {}: {}", lease.device_id, e.what());
339 0 : return false;
340 0 : }
341 : }
342 :
343 : bool
344 11 : SwarmManager::commitLeaseInternal(const MobileLease& lease)
345 : {
346 11 : auto pending = pendingMobileLeases_.find(lease.device_id);
347 11 : if (pending != pendingMobileLeases_.end()) {
348 6 : if (!isNewerLease(pending->second.lease, lease))
349 6 : pendingMobileLeases_.erase(pending);
350 : }
351 :
352 11 : bool changed = addMobileNodes(lease.device_id);
353 11 : auto current = mobileNodeLeases_.find(lease.device_id);
354 11 : if (current == mobileNodeLeases_.end() || isNewerLease(lease, current->second)) {
355 11 : mobileNodeLeases_.insert_or_assign(lease.device_id, lease);
356 11 : legacyMobileNodeExpiries_.erase(lease.device_id);
357 11 : changed = true;
358 : }
359 11 : scheduleMobileLeaseExpiryInternal();
360 11 : return changed;
361 : }
362 :
363 : void
364 11 : SwarmManager::enqueuePendingLeaseInternal(const MobileLease& lease,
365 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& source)
366 : {
367 11 : const auto& nodeId = lease.device_id;
368 11 : auto pending = pendingMobileLeases_.find(nodeId);
369 11 : if (pending != pendingMobileLeases_.end()) {
370 1 : if (isNewerLease(lease, pending->second.lease))
371 0 : pending->second.lease = lease;
372 1 : if (source)
373 1 : pending->second.source = source;
374 : } else {
375 10 : if (pendingMobileLeases_.size() >= MAX_PENDING_MOBILE_LEASES) {
376 : // Evict the entry that would expire first: it is the least useful to keep resolving.
377 0 : auto oldest = std::min_element(pendingMobileLeases_.begin(),
378 : pendingMobileLeases_.end(),
379 0 : [](const auto& a, const auto& b) {
380 0 : return a.second.lease.expires_at < b.second.lease.expires_at;
381 : });
382 0 : if (oldest != pendingMobileLeases_.end() && oldest->second.lease.expires_at >= lease.expires_at)
383 0 : return;
384 0 : pendingMobileLeases_.erase(oldest);
385 : }
386 10 : pendingMobileLeases_.emplace(nodeId, PendingLease {lease, source});
387 : }
388 :
389 11 : if (certFetchInFlight_.count(nodeId))
390 0 : return;
391 :
392 11 : if (source) {
393 6 : certFetchInFlight_.emplace(nodeId);
394 6 : dht::ThreadPool::io().run([w = weak(), source, nodeId] {
395 6 : if (auto shared = w.lock())
396 18 : shared->requestCertificates(source, {nodeId});
397 6 : });
398 : } else {
399 5 : certFetchInFlight_.emplace(nodeId);
400 5 : dht::ThreadPool::io().run([w = weak(), nodeId] {
401 5 : if (auto shared = w.lock())
402 5 : shared->fetchCertificateFromDht(nodeId);
403 5 : });
404 : }
405 : }
406 :
407 : void
408 4 : SwarmManager::abandonLeaseInternal(const NodeId& nodeId)
409 : {
410 4 : certFetchInFlight_.erase(nodeId);
411 4 : pendingMobileLeases_.erase(nodeId);
412 4 : }
413 :
414 : void
415 6 : SwarmManager::requestCertificates(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
416 : const std::vector<NodeId>& ids)
417 : {
418 6 : if (!socket || ids.empty() || isShutdown_) {
419 0 : std::lock_guard lock(mutex);
420 0 : for (const auto& id : ids)
421 0 : certFetchInFlight_.erase(id);
422 0 : return;
423 0 : }
424 6 : const auto peer = NodeId(socket->deviceId());
425 6 : CertRequest request;
426 : {
427 6 : std::lock_guard lock(mutex);
428 6 : auto& state = outstandingCertRequests_[peer];
429 6 : if (state.timer) {
430 : // One request in flight per peer: drop the resolution so that the
431 : // next gossip round asks again.
432 2 : for (const auto& id : ids)
433 1 : certFetchInFlight_.erase(id);
434 1 : return;
435 : }
436 10 : for (const auto& id : ids) {
437 5 : if (request.ids.size() >= MAX_CERT_REQUEST_IDS) {
438 0 : certFetchInFlight_.erase(id);
439 0 : continue;
440 : }
441 5 : request.ids.emplace_back(id);
442 : }
443 5 : if (request.ids.empty()) {
444 0 : outstandingCertRequests_.erase(peer);
445 0 : return;
446 : }
447 5 : state.ids.insert(request.ids.begin(), request.ids.end());
448 5 : state.timer = std::make_shared<asio::steady_timer>(*Manager::instance().ioContext());
449 5 : state.timer->expires_after(CERT_REQUEST_TIMEOUT);
450 5 : state.timer->async_wait([w = weak(), peer](const asio::error_code& ec) {
451 5 : if (ec == asio::error::operation_aborted)
452 4 : return;
453 1 : auto shared = w.lock();
454 1 : if (!shared)
455 0 : return;
456 1 : std::vector<NodeId> unanswered;
457 : {
458 1 : std::lock_guard lock(shared->mutex);
459 1 : auto it = shared->outstandingCertRequests_.find(peer);
460 1 : if (it == shared->outstandingCertRequests_.end())
461 0 : return;
462 1 : unanswered.assign(it->second.ids.begin(), it->second.ids.end());
463 1 : shared->outstandingCertRequests_.erase(it);
464 1 : }
465 : // The peer did not answer: fall back to the DHT.
466 2 : for (const auto& id : unanswered)
467 1 : shared->fetchCertificateFromDht(id);
468 1 : });
469 6 : }
470 :
471 5 : Message msg;
472 5 : msg.is_mobile = isMobile_;
473 5 : msg.cert_request = std::move(request);
474 :
475 5 : msgpack::sbuffer buffer;
476 5 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
477 5 : pk.pack(msg);
478 :
479 5 : std::error_code ec;
480 5 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
481 5 : if (ec)
482 0 : JAMI_ERROR("{}", ec.message());
483 6 : }
484 :
485 : void
486 2 : SwarmManager::onCertRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const CertRequest& request)
487 : {
488 2 : if (!socket || request.ids.empty() || !certificateProvider_)
489 0 : return;
490 :
491 2 : std::vector<NodeId> toAnswer;
492 : {
493 2 : std::lock_guard lock(mutex);
494 5 : for (const auto& id : request.ids) {
495 3 : if (toAnswer.size() >= MAX_CERT_REQUEST_IDS)
496 0 : break;
497 : // A peer must not be able to use the swarm as a generic certificate
498 : // oracle: only serve certificates for devices we ourselves announced
499 : // as mobile in this conversation, plus our own when we are mobile.
500 3 : if (id != id_ && !mobileNodeLeases_.count(id))
501 2 : continue;
502 1 : toAnswer.emplace_back(id);
503 : }
504 2 : }
505 2 : if (toAnswer.empty())
506 1 : return;
507 :
508 1 : CertResponse response;
509 1 : size_t totalSize = 0;
510 2 : for (const auto& id : toAnswer) {
511 1 : auto certificate = certificateProvider_(id);
512 1 : if (!certificate)
513 0 : continue;
514 1 : auto packed = certificate->getPacked();
515 2 : if (packed.empty() || packed.size() > MAX_MOBILE_CERTIFICATE_SIZE
516 2 : || totalSize + packed.size() > MAX_MOBILE_CERTIFICATES_SIZE)
517 0 : continue;
518 1 : totalSize += packed.size();
519 1 : response.certificates.emplace_back(std::move(packed));
520 1 : }
521 1 : if (response.certificates.empty())
522 0 : return;
523 :
524 1 : Message msg;
525 1 : msg.is_mobile = isMobile_;
526 1 : msg.cert_response = std::move(response);
527 :
528 1 : msgpack::sbuffer buffer;
529 1 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
530 1 : pk.pack(msg);
531 :
532 1 : std::error_code ec;
533 1 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
534 1 : if (ec)
535 0 : JAMI_ERROR("{}", ec.message());
536 2 : }
537 :
538 : void
539 5 : SwarmManager::onCertResponse(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const CertResponse& response)
540 : {
541 5 : if (!socket)
542 0 : return;
543 5 : const auto peer = NodeId(socket->deviceId());
544 :
545 5 : std::set<NodeId> requested;
546 : {
547 5 : std::lock_guard lock(mutex);
548 5 : auto it = outstandingCertRequests_.find(peer);
549 5 : if (it == outstandingCertRequests_.end())
550 1 : return; // Unsolicited.
551 4 : requested = std::move(it->second.ids);
552 4 : if (it->second.timer)
553 4 : it->second.timer->cancel();
554 4 : outstandingCertRequests_.erase(it);
555 5 : }
556 :
557 4 : size_t totalSize = 0;
558 8 : for (const auto& packed : response.certificates) {
559 8 : if (packed.empty() || packed.size() > MAX_MOBILE_CERTIFICATE_SIZE
560 8 : || totalSize + packed.size() > MAX_MOBILE_CERTIFICATES_SIZE)
561 0 : break;
562 4 : totalSize += packed.size();
563 : try {
564 4 : auto certificate = std::make_shared<dht::crypto::Certificate>(packed);
565 4 : const auto nodeId = certificate->getLongId();
566 4 : if (!requested.erase(nodeId))
567 1 : continue; // Not something we asked for.
568 3 : onCertificateResolved(nodeId, certificate);
569 4 : } catch (const std::exception& e) {
570 0 : JAMI_WARNING("Ignoring invalid certificate from {}: {}", peer, e.what());
571 0 : }
572 : }
573 :
574 : // Whatever the peer could not provide is worth one DHT lookup.
575 5 : for (const auto& nodeId : requested)
576 1 : fetchCertificateFromDht(nodeId);
577 5 : }
578 :
579 : void
580 7 : SwarmManager::fetchCertificateFromDht(const NodeId& nodeId)
581 : {
582 7 : if (isShutdown_)
583 0 : return;
584 7 : if (!certificateFetcher_) {
585 2 : std::lock_guard lock(mutex);
586 2 : abandonLeaseInternal(nodeId);
587 2 : return;
588 2 : }
589 5 : certificateFetcher_(nodeId, [w = weak(), nodeId](const std::shared_ptr<dht::crypto::Certificate>& certificate) {
590 5 : auto shared = w.lock();
591 5 : if (!shared)
592 0 : return;
593 5 : if (certificate && certificate->getLongId() == nodeId)
594 4 : shared->onCertificateResolved(nodeId, certificate);
595 : else {
596 1 : std::lock_guard lock(shared->mutex);
597 1 : shared->abandonLeaseInternal(nodeId);
598 1 : }
599 5 : });
600 : }
601 :
602 : void
603 7 : SwarmManager::onCertificateResolved(const NodeId& nodeId, const std::shared_ptr<dht::crypto::Certificate>& certificate)
604 : {
605 7 : if (!certificate)
606 0 : return;
607 :
608 7 : std::optional<MobileLease> lease;
609 : {
610 7 : std::lock_guard lock(mutex);
611 7 : certFetchInFlight_.erase(nodeId);
612 7 : auto pending = pendingMobileLeases_.find(nodeId);
613 7 : if (pending == pendingMobileLeases_.end())
614 0 : return;
615 7 : lease = pending->second.lease;
616 7 : }
617 :
618 : // Re-run the cheap checks: the lease may have expired while we were resolving.
619 7 : if (!precheckLease(*lease) || !verifyLease(*certificate, *lease)) {
620 1 : std::lock_guard lock(mutex);
621 1 : auto pending = pendingMobileLeases_.find(nodeId);
622 1 : if (pending != pendingMobileLeases_.end() && !isNewerLease(pending->second.lease, *lease))
623 1 : abandonLeaseInternal(nodeId);
624 1 : return;
625 1 : }
626 :
627 6 : bool changed = false;
628 : {
629 6 : std::lock_guard lock(mutex);
630 6 : changed = commitLeaseInternal(*lease);
631 6 : }
632 6 : if (changed)
633 6 : emitMobileNodesChanged();
634 7 : }
635 :
636 : std::optional<MobileNodeInfo>
637 4125 : SwarmManager::localMobileNodeInfo()
638 : {
639 4125 : if (!isMobile_ || !mobileLeaseProvider_)
640 4122 : return std::nullopt;
641 :
642 3 : std::lock_guard renewalLock(mobileLeaseRenewalMtx_);
643 :
644 3 : auto renewalThresholdTime = toSecondsSinceEpoch(std::chrono::system_clock::now() + MOBILE_LEASE_RENEWAL_THRESHOLD);
645 : {
646 3 : std::lock_guard lock(mutex);
647 5 : if (localMobileNodeInfo_ && localMobileNodeInfo_->lease
648 5 : && localMobileNodeInfo_->lease->expires_at > renewalThresholdTime)
649 1 : return localMobileNodeInfo_;
650 3 : }
651 :
652 2 : auto renewed = mobileLeaseProvider_();
653 4 : if (!renewed || renewed->id != id_ || !renewed->lease || renewed->lease->device_id != id_
654 4 : || !precheckLease(*renewed->lease))
655 0 : return std::nullopt;
656 2 : std::lock_guard lock(mutex);
657 2 : localMobileNodeInfo_ = std::move(renewed);
658 2 : return localMobileNodeInfo_;
659 3 : }
660 :
661 : void
662 704 : SwarmManager::scheduleMobileLeaseExpiryInternal()
663 : {
664 704 : mobileLeaseExpiryTimer_.cancel();
665 704 : if ((mobileNodeLeases_.empty() && legacyMobileNodeExpiries_.empty()) || isShutdown_)
666 693 : return;
667 :
668 11 : auto nearestExpiry = std::numeric_limits<int64_t>::max();
669 23 : for (const auto& [node, lease] : mobileNodeLeases_)
670 12 : nearestExpiry = std::min(nearestExpiry, lease.expires_at);
671 11 : for (const auto& [node, expiry] : legacyMobileNodeExpiries_)
672 0 : nearestExpiry = std::min(nearestExpiry, expiry);
673 11 : auto expiryTime = timePointFromSeconds(nearestExpiry);
674 11 : const auto now = std::chrono::system_clock::now();
675 11 : constexpr auto MAX_TIMER_DELAY_SECONDS = std::chrono::minutes(1);
676 11 : const auto delay = std::min<std::chrono::system_clock::duration>(expiryTime > now ? expiryTime - now : std::chrono::seconds(0), MAX_TIMER_DELAY_SECONDS);
677 11 : mobileLeaseExpiryTimer_.expires_after(delay);
678 11 : mobileLeaseExpiryTimer_.async_wait([w = weak()](const asio::error_code& ec) {
679 11 : if (auto shared = w.lock())
680 11 : shared->expireMobileLeases(ec);
681 11 : });
682 : }
683 :
684 : void
685 4 : SwarmManager::expireMobileLeases(const asio::error_code& ec)
686 : {
687 4 : if (ec == asio::error::operation_aborted)
688 3 : return;
689 :
690 1 : bool changed = false;
691 : {
692 1 : std::lock_guard lock(mutex);
693 1 : auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
694 2 : for (auto it = mobileNodeLeases_.begin(); it != mobileNodeLeases_.end();) {
695 1 : if (it->second.expires_at > now) {
696 0 : ++it;
697 0 : continue;
698 : }
699 1 : const auto nodeId = it->first;
700 1 : it = mobileNodeLeases_.erase(it);
701 1 : auto legacy = legacyMobileNodeExpiries_.find(nodeId);
702 1 : if (legacy == legacyMobileNodeExpiries_.end() || legacy->second <= now) {
703 1 : routing_table.removeMobileNode(nodeId);
704 1 : routing_table.findBucket(nodeId)->changeMobility(nodeId, false);
705 : }
706 1 : changed = true;
707 : }
708 1 : for (auto it = legacyMobileNodeExpiries_.begin(); it != legacyMobileNodeExpiries_.end();) {
709 0 : if (it->second > now) {
710 0 : ++it;
711 0 : continue;
712 : }
713 0 : const auto nodeId = it->first;
714 0 : it = legacyMobileNodeExpiries_.erase(it);
715 0 : auto lease = mobileNodeLeases_.find(nodeId);
716 0 : if (lease == mobileNodeLeases_.end() || lease->second.expires_at <= now) {
717 0 : routing_table.removeMobileNode(nodeId);
718 0 : routing_table.findBucket(nodeId)->changeMobility(nodeId, false);
719 : }
720 0 : changed = true;
721 : }
722 1 : for (auto it = pendingMobileLeases_.begin(); it != pendingMobileLeases_.end();) {
723 0 : if (it->second.lease.expires_at > now)
724 0 : ++it;
725 : else
726 0 : it = pendingMobileLeases_.erase(it);
727 : }
728 1 : scheduleMobileLeaseExpiryInternal();
729 1 : }
730 1 : if (changed)
731 1 : emitMobileNodesChanged();
732 : }
733 :
734 : void
735 292 : SwarmManager::emitMobileNodesChanged()
736 : {
737 292 : std::lock_guard emissionLock(mobileNodesEmissionMtx_);
738 292 : auto mobileNodes = getKnownMobileNodes();
739 292 : auto mobileNodeInfos = getKnownMobileNodeInfos();
740 292 : OnMobileNodesChanged callback;
741 292 : OnMobileNodeInfosChanged infosCallback;
742 : {
743 292 : std::lock_guard callbackLock(onMobileNodesChangedMtx_);
744 292 : callback = onMobileNodesChanged_;
745 292 : infosCallback = onMobileNodeInfosChanged_;
746 292 : }
747 292 : if (callback)
748 19 : callback(mobileNodes);
749 292 : if (infosCallback)
750 1 : infosCallback(mobileNodeInfos);
751 292 : }
752 :
753 : void
754 4822 : SwarmManager::maintainBuckets(const std::set<NodeId>& toConnect)
755 : {
756 4822 : std::set<NodeId> nodes = toConnect;
757 4822 : std::unique_lock lock(mutex);
758 4822 : auto& buckets = routing_table.getBuckets();
759 11817 : for (auto it = buckets.begin(); it != buckets.end(); ++it) {
760 6995 : auto& bucket = *it;
761 6995 : bool myBucket = routing_table.contains(it, id_);
762 9168 : auto connecting_nodes = myBucket ? bucket.getConnectingNodesSize()
763 2173 : : bucket.getConnectingNodesSize() + bucket.getNodesSize();
764 6995 : if (connecting_nodes < Bucket::BUCKET_MAX_SIZE) {
765 5245 : auto nodesToTry = bucket.getKnownNodesRandom(Bucket::BUCKET_MAX_SIZE - connecting_nodes, rd);
766 6918 : for (auto& node : nodesToTry)
767 1673 : routing_table.addConnectingNode(node);
768 :
769 5245 : nodes.insert(nodesToTry.begin(), nodesToTry.end());
770 5245 : }
771 : }
772 4822 : lock.unlock();
773 6521 : for (const auto& node : nodes)
774 1699 : tryConnect(node);
775 4822 : }
776 :
777 : void
778 2093 : SwarmManager::sendRequest(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
779 : const NodeId& nodeId,
780 : Query q,
781 : int numberNodes)
782 : {
783 2093 : auto selfMobileInfo = localMobileNodeInfo();
784 4186 : dht::ThreadPool::io().run(
785 4186 : [socket, isMobile = isMobile_, selfMobileInfo = std::move(selfMobileInfo), nodeId, q, numberNodes] {
786 2093 : msgpack::sbuffer buffer;
787 2093 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
788 2093 : Message msg;
789 2093 : msg.is_mobile = isMobile;
790 2093 : msg.self_mobile_info = selfMobileInfo;
791 2093 : msg.request = Request {q, numberNodes, nodeId};
792 2093 : pk.pack(msg);
793 :
794 2093 : std::error_code ec;
795 2093 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
796 2093 : if (ec) {
797 4 : JAMI_ERROR("{}", ec.message());
798 : }
799 2093 : });
800 2093 : }
801 :
802 : void
803 2029 : SwarmManager::sendAnswer(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket, const Message& msg_)
804 : {
805 2029 : if (msg_.request->q != Query::FIND)
806 0 : return;
807 :
808 2029 : auto selfMobileInfo = localMobileNodeInfo();
809 2029 : Message msg;
810 : {
811 2029 : std::lock_guard lock(mutex);
812 2029 : auto nodes = routing_table.closestNodes(msg_.request->nodeId, msg_.request->num);
813 2029 : auto bucket = routing_table.findBucket(msg_.request->nodeId);
814 2029 : const auto& m_nodes = bucket->getMobileNodes();
815 2029 : std::vector<NodeId> responseMobileNodes;
816 2029 : responseMobileNodes.reserve(m_nodes.size());
817 2029 : std::vector<MobileNodeInfo> mobileNodeInfos;
818 2029 : mobileNodeInfos.reserve(m_nodes.size());
819 2029 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
820 2115 : for (const auto& node : m_nodes) {
821 86 : if (!isMobileNodeCurrentInternal(node, now))
822 0 : continue;
823 86 : responseMobileNodes.emplace_back(node);
824 86 : if (mobileNodeInfos.size() >= MAX_MOBILE_NODE_INFOS)
825 0 : continue;
826 86 : auto lease = mobileNodeLeases_.find(node);
827 86 : if (lease == mobileNodeLeases_.end()) {
828 86 : if (msg_.v >= 3)
829 86 : continue;
830 0 : mobileNodeInfos.emplace_back(MobileNodeInfo {node, std::nullopt});
831 : } else {
832 0 : mobileNodeInfos.emplace_back(MobileNodeInfo {node, lease->second});
833 : }
834 : }
835 2029 : Response toResponse {Query::FOUND, nodes, std::move(responseMobileNodes), std::move(mobileNodeInfos)};
836 :
837 2029 : msg.is_mobile = isMobile_;
838 2029 : msg.self_mobile_info = std::move(selfMobileInfo);
839 2029 : msg.response = std::move(toResponse);
840 2029 : }
841 :
842 2029 : msgpack::sbuffer buffer;
843 2029 : msgpack::packer<msgpack::sbuffer> pk(&buffer);
844 2029 : pk.pack(msg);
845 :
846 2029 : std::error_code ec;
847 2029 : socket->write(reinterpret_cast<const unsigned char*>(buffer.data()), buffer.size(), ec);
848 2029 : if (ec) {
849 8 : JAMI_ERROR("{}", ec.message());
850 8 : return;
851 : }
852 2045 : }
853 :
854 : void
855 3459 : SwarmManager::receiveMessage(const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket)
856 : {
857 6918 : socket->setOnRecv(dhtnet::buildMsgpackReader<Message>(
858 6918 : [w = weak(), wsocket = std::weak_ptr<dhtnet::ChannelSocketInterface>(socket)](Message&& msg) {
859 4057 : auto shared = w.lock();
860 4057 : auto socket = wsocket.lock();
861 4057 : if (!shared || !socket)
862 0 : return std::make_error_code(std::errc::operation_canceled);
863 :
864 4057 : auto validMobileAnnouncement = msg.v < 3 || shared->conversationId_.empty();
865 4057 : if (msg.self_mobile_info && msg.self_mobile_info->id == socket->deviceId()) {
866 : // The peer's own certificate is authenticated by the channel's TLS
867 : // handshake and pinned when the swarm channel was added, so this
868 : // resolves locally without any lookup.
869 5 : validMobileAnnouncement = msg.self_mobile_info->lease
870 10 : && msg.self_mobile_info->lease->device_id == msg.self_mobile_info->id
871 10 : && shared->precheckLease(*msg.self_mobile_info->lease);
872 5 : if (validMobileAnnouncement && shared->setMobileNodeInfo(*msg.self_mobile_info, true, socket))
873 1 : shared->emitMobileNodesChanged();
874 : }
875 4057 : if (msg.is_mobile && validMobileAnnouncement) {
876 253 : if (msg.v < 3 && !shared->conversationId_.empty())
877 3 : shared->setMobileNodes(std::vector<NodeId> {socket->deviceId()});
878 253 : shared->changeMobility(socket->deviceId(), true);
879 : }
880 :
881 4057 : if (msg.cert_request) {
882 2 : shared->onCertRequest(socket, *msg.cert_request);
883 4055 : } else if (msg.cert_response) {
884 5 : shared->onCertResponse(socket, *msg.cert_response);
885 4050 : } else if (msg.request) {
886 2029 : shared->sendAnswer(socket, msg);
887 :
888 2021 : } else if (msg.response) {
889 2015 : shared->setKnownNodes(msg.response->nodes);
890 2015 : const auto requireLease = msg.v >= 3 && !shared->conversationId_.empty();
891 2015 : bool changed = false;
892 2015 : size_t records = 0;
893 2019 : for (const auto& mobile : msg.response->mobile_node_infos) {
894 4 : if (records++ == MAX_MOBILE_NODE_INFOS)
895 0 : break;
896 4 : changed |= shared->setMobileNodeInfo(mobile, requireLease, socket);
897 : }
898 2015 : if (changed)
899 0 : shared->emitMobileNodesChanged();
900 2015 : const auto acceptLegacy = msg.v < 3 || shared->conversationId_.empty();
901 2015 : if (acceptLegacy)
902 611 : shared->setMobileNodes(msg.response->mobile_nodes);
903 : }
904 4057 : return std::error_code();
905 4057 : }));
906 :
907 3459 : socket->onShutdown([w = weak(), deviceId = socket->deviceId()](const std::error_code&) {
908 1958 : dht::ThreadPool::io().run([w, deviceId] {
909 1958 : auto shared = w.lock();
910 1958 : if (shared && !shared->isShutdown_) {
911 1100 : shared->removeNode(deviceId);
912 : }
913 1958 : });
914 1958 : });
915 3459 : }
916 :
917 : void
918 2093 : SwarmManager::resetNodeExpiry(const asio::error_code& ec,
919 : const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket,
920 : NodeId node)
921 : {
922 2093 : NodeId idToFind;
923 2093 : std::list<Bucket>::iterator bucket;
924 :
925 2093 : if (ec == asio::error::operation_aborted)
926 0 : return;
927 :
928 2093 : if (!node) {
929 0 : bucket = routing_table.findBucket(socket->deviceId());
930 0 : idToFind = bucket->randomId(rd);
931 : } else {
932 2093 : bucket = routing_table.findBucket(node);
933 2093 : idToFind = node;
934 : }
935 :
936 2093 : sendRequest(socket, idToFind, Query::FIND, Bucket::BUCKET_MAX_SIZE);
937 :
938 2093 : if (!node) {
939 0 : auto& nodeTimer = bucket->getNodeTimer(socket);
940 0 : nodeTimer.expires_after(FIND_PERIOD);
941 0 : nodeTimer.async_wait(std::bind(&jami::SwarmManager::resetNodeExpiry,
942 0 : shared_from_this(),
943 : std::placeholders::_1,
944 : socket,
945 0 : NodeId {}));
946 : }
947 : }
948 :
949 : void
950 2372 : SwarmManager::tryConnect(const NodeId& nodeId, bool noNewSocket)
951 : {
952 2372 : if (needSocketCb_)
953 2368 : needSocketCb_(
954 4736 : nodeId.toString(),
955 4736 : [w = weak(), nodeId](const std::shared_ptr<dhtnet::ChannelSocketInterface>& socket) {
956 2208 : auto shared = w.lock();
957 2208 : if (!shared || shared->isShutdown_)
958 562 : return true;
959 1646 : if (socket) {
960 1493 : shared->addChannel(socket);
961 1493 : return true;
962 : }
963 153 : std::unique_lock lk(shared->mutex);
964 153 : auto bucket = shared->routing_table.findBucket(nodeId);
965 153 : bucket->removeConnectingNode(nodeId);
966 153 : if (!bucket->hasMobileNode(nodeId))
967 152 : bucket->addKnownNode(nodeId);
968 153 : if (shared->routing_table.getActiveNodesCount() == 0 && shared->onConnectionChanged_) {
969 98 : lk.unlock();
970 98 : JAMI_LOG("[SwarmManager {:p}] Bootstrap: all connections failed", fmt::ptr(shared.get()));
971 98 : shared->onConnectionChanged_(false);
972 : }
973 153 : return true;
974 2208 : },
975 : noNewSocket);
976 2372 : }
977 :
978 : void
979 968 : SwarmManager::removeNodeInternal(const NodeId& nodeId)
980 : {
981 968 : routing_table.removeNode(nodeId);
982 968 : }
983 :
984 : void
985 690 : SwarmManager::connectNode(const NodeId& nodeId)
986 : {
987 : {
988 690 : std::lock_guard lock(mutex);
989 690 : if (isShutdown_)
990 12 : return;
991 678 : if (isConnectedWith(nodeId))
992 5 : return;
993 673 : addKnownNode(nodeId);
994 673 : if (!routing_table.addConnectingNode(nodeId))
995 0 : return;
996 690 : }
997 673 : tryConnect(nodeId, true);
998 : }
999 :
1000 : std::vector<NodeId>
1001 17 : SwarmManager::getAllNodes() const
1002 : {
1003 17 : std::lock_guard lock(mutex);
1004 34 : return routing_table.getAllNodes();
1005 17 : }
1006 :
1007 : std::vector<NodeId>
1008 2529 : SwarmManager::getConnectedNodes() const
1009 : {
1010 2529 : std::lock_guard lock(mutex);
1011 5058 : return routing_table.getConnectedNodes();
1012 2529 : }
1013 :
1014 : std::vector<NodeId>
1015 64 : SwarmManager::getMobileNodesToNotify()
1016 : {
1017 64 : std::lock_guard lock(mutex);
1018 128 : return routing_table.getMobileNodesToNotify();
1019 64 : }
1020 :
1021 : std::vector<NodeId>
1022 329 : SwarmManager::getKnownMobileNodes() const
1023 : {
1024 329 : std::lock_guard lock(mutex);
1025 658 : return routing_table.getKnownMobileNodes();
1026 329 : }
1027 :
1028 : std::vector<MobileNodeInfo>
1029 346 : SwarmManager::getKnownMobileNodeInfos() const
1030 : {
1031 346 : std::lock_guard lock(mutex);
1032 346 : std::vector<MobileNodeInfo> infos;
1033 346 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
1034 899 : for (const auto& node : routing_table.getKnownMobileNodes()) {
1035 553 : if (!isMobileNodeCurrentInternal(node, now))
1036 32 : continue;
1037 521 : auto lease = mobileNodeLeases_.find(node);
1038 521 : infos.emplace_back(MobileNodeInfo {node,
1039 1042 : lease == mobileNodeLeases_.end()
1040 521 : ? std::nullopt
1041 28 : : std::optional<MobileLease>(lease->second)});
1042 346 : }
1043 692 : return infos;
1044 346 : }
1045 :
1046 : std::vector<MobileNodeInfo>
1047 2461 : SwarmManager::getMobileNodeInfosToNotify()
1048 : {
1049 2461 : std::lock_guard lock(mutex);
1050 2461 : std::vector<MobileNodeInfo> infos;
1051 2461 : const auto now = toSecondsSinceEpoch(std::chrono::system_clock::now());
1052 2462 : for (const auto& node : routing_table.getMobileNodesToNotify()) {
1053 1 : if (!isMobileNodeCurrentInternal(node, now))
1054 1 : continue;
1055 0 : auto lease = mobileNodeLeases_.find(node);
1056 0 : infos.emplace_back(MobileNodeInfo {node,
1057 0 : lease == mobileNodeLeases_.end()
1058 0 : ? std::nullopt
1059 0 : : std::optional<MobileLease>(lease->second)});
1060 2461 : }
1061 4922 : return infos;
1062 2461 : }
1063 :
1064 : std::vector<std::map<std::string, std::string>>
1065 2 : SwarmManager::getRoutingTableInfo() const
1066 : {
1067 2 : std::lock_guard lock(mutex);
1068 2 : auto stats = routing_table.getRoutingTableStats();
1069 2 : const auto toNotify = routing_table.getMobileNodesToNotify();
1070 2 : std::set<std::string> responsible;
1071 4 : for (const auto& node : toNotify)
1072 2 : responsible.emplace(node.toString());
1073 2 : std::vector<std::map<std::string, std::string>> result;
1074 2 : result.reserve(stats.size());
1075 11 : for (const auto& stat : stats) {
1076 72 : result.push_back({{"id", stat.id},
1077 9 : {"device", stat.id},
1078 9 : {"status", stat.status},
1079 9 : {"remoteAddress", stat.remoteAddress},
1080 9 : {"mobile", stat.isMobile ? "true" : "false"},
1081 9 : {"responsible", responsible.count(stat.id) ? "true" : "false"}});
1082 9 : if (stat.connectionTime != std::chrono::system_clock::time_point::min()) {
1083 0 : auto tt = std::chrono::system_clock::to_time_t(stat.connectionTime);
1084 0 : result.back().emplace("connectionTime", std::to_string(tt));
1085 : }
1086 : }
1087 4 : return result;
1088 29 : }
1089 :
1090 : bool
1091 9255 : SwarmManager::isConnected() const
1092 : {
1093 9255 : std::lock_guard lock(mutex);
1094 18510 : return !routing_table.isEmpty();
1095 9255 : }
1096 :
1097 : void
1098 16 : SwarmManager::deleteNode(const std::vector<NodeId>& nodes)
1099 : {
1100 16 : bool mobileNodesChanged = false;
1101 : {
1102 16 : std::lock_guard lock(mutex);
1103 16 : auto mobileNodes = routing_table.getKnownMobileNodes();
1104 32 : for (const auto& node : nodes) {
1105 16 : routing_table.deleteNode(node);
1106 16 : mobileNodesChanged |= mobileNodeLeases_.erase(node) != 0;
1107 16 : mobileNodesChanged |= legacyMobileNodeExpiries_.erase(node) != 0;
1108 16 : pendingMobileLeases_.erase(node);
1109 : }
1110 16 : scheduleMobileLeaseExpiryInternal();
1111 16 : mobileNodesChanged |= mobileNodes != routing_table.getKnownMobileNodes();
1112 16 : }
1113 16 : if (mobileNodesChanged)
1114 2 : emitMobileNodesChanged();
1115 16 : maintainBuckets();
1116 16 : }
1117 :
1118 : } // namespace jami
|
