cliquenet/net/server.rs
1use std::{collections::HashMap, mem, net::IpAddr, sync::Arc, time::Duration};
2
3use bytes::{Bytes, BytesMut};
4use tokio::{
5 net::{TcpListener, TcpStream},
6 select, spawn,
7 sync::{
8 mpsc::{UnboundedReceiver, UnboundedSender},
9 watch,
10 },
11 task::{JoinHandle, JoinSet},
12};
13use tokio_util::{sync::CancellationToken, task::JoinMap};
14use tracing::{debug, error, info, trace, warn};
15
16use crate::{
17 Config, Metrics, NetAddr, PublicKey, Role,
18 connection::Connection,
19 delay::DelayQueue,
20 error::NetworkError,
21 msg::{MsgId, Slot, Trailer, hello::Hello},
22 net::{Command, PeerCommand, PeerMessage, RetryPolicy, SendAction, peer::Peer},
23 queue::Queue,
24 util::until,
25};
26
27pub struct Server {
28 key: PublicKey,
29 conf: Arc<Config>,
30 role: Role,
31 msgid: MsgId,
32 lower_bound: Slot,
33 parties: HashMap<PublicKey, Party>,
34 ibound: UnboundedSender<PeerMessage>,
35 obound: UnboundedReceiver<Command>,
36 next_slot: watch::Receiver<Slot>,
37 accept_tasks: JoinSet<Result<Connection, NetworkError>>,
38 #[allow(clippy::type_complexity)]
39 hello_tasks: JoinMap<PublicKey, Result<(Hello, Connection, Option<Hello>), NetworkError>>,
40 connect_tasks: JoinMap<PublicKey, Connection>,
41 peer_tasks: JoinMap<PublicKey, Peer>,
42 metrics: Arc<dyn Metrics>,
43}
44
45struct Party {
46 role: Role,
47 addr: NetAddr,
48 outbox: Queue<(RetryPolicy, Bytes)>,
49 retry: DelayQueue,
50 peer: PeerState,
51}
52
53/// The states of a peer.
54///
55/// When a party is created there exists no peer yet. After a connection has
56/// been accepted or a connect attempt succeeded, a peer task is created and
57/// the state transitions to `Connected`. The cancellation token is used if
58/// the peer task should stop in order to replace the peer's connection. When
59/// cancelled the next connection is stored in `Replace` and once the peer
60/// task finishes and returns the peer, its connection is set and a new peer
61/// task (with the same peer object) is spawned and the state transitions to
62/// `Connected` again.
63///
64/// The `Reconnect` state is entered if the peer itself errors and its task
65/// finishes. We keep the peer here until we have a new connection to resume.
66/// Once a connect or accept task finishes we set the connection and spawn
67/// a new peer task again.
68enum PeerState {
69 /// Initial state.
70 None,
71 /// A peer task is running.
72 ///
73 /// The cancellation token can be used to interrupt the peer. The task
74 /// will end and the peer object is returned.
75 Connected(CancellationToken),
76 /// A peer has errored and wants a fresh connection.
77 ///
78 /// We store the peer here until a new connection is available.
79 Reconnect(Peer),
80 /// The server wants to replace a peer's connection.
81 ///
82 /// Only entered from `Connected` after the server has cancelled the
83 /// peer task. While waiting for the task to return the peer object
84 /// we park the connection here.
85 Replace(Connection),
86}
87
88impl Server {
89 pub(super) fn spawn(
90 conf: Arc<Config>,
91 listener: TcpListener,
92 role: Role,
93 tx: UnboundedSender<PeerMessage>,
94 rx: UnboundedReceiver<Command>,
95 sx: watch::Receiver<Slot>,
96 metrics: Arc<dyn Metrics>,
97 ) -> JoinHandle<()> {
98 let our_key = conf.keypair.public_key();
99 let parties = conf
100 .parties
101 .iter()
102 .filter(|&(k, _)| *k != our_key)
103 .map(|(k, a)| {
104 let p = Party::new(conf.clone(), Role::Active, a.clone());
105 (*k, p)
106 })
107 .collect();
108
109 let this = Self {
110 key: our_key,
111 conf,
112 role,
113 ibound: tx,
114 obound: rx,
115 parties,
116 accept_tasks: JoinSet::new(),
117 connect_tasks: JoinMap::new(),
118 hello_tasks: JoinMap::new(),
119 peer_tasks: JoinMap::new(),
120 msgid: MsgId::new(0),
121 next_slot: sx,
122 lower_bound: Slot::MIN,
123 metrics,
124 };
125
126 spawn(this.run(listener))
127 }
128
129 async fn run(mut self, listener: TcpListener) {
130 // Connect to all peers.
131 for (k, a) in self
132 .parties
133 .iter()
134 .map(|(k, p)| (*k, p.addr.clone()))
135 .collect::<Vec<_>>()
136 {
137 self.spawn_connect(k, a)
138 }
139
140 loop {
141 select! {
142 x = listener.accept(), if self.accept_tasks.len() < self.conf.max_accept_tasks => {
143 match x {
144 Ok((stream, addr)) => {
145 debug!(
146 name = %self.conf.name,
147 node = %self.key,
148 %addr,
149 "accepted new tcp connection"
150 );
151 self.spawn_accept(stream)
152 }
153 Err(err) => {
154 warn!(
155 name = %self.conf.name,
156 node = %self.key,
157 %err,
158 "error accepting tcp connection"
159 )
160 }
161 }
162 }
163
164 Some(h) = self.accept_tasks.join_next() => match h {
165 Ok(Ok(conn)) => {
166 self.metrics.set(&self.key, ACCEPT_TASKS, self.accept_tasks.len());
167 if conn.key == self.key {
168 warn!(
169 name = %self.conf.name,
170 node = %self.key,
171 peer = %conn.key,
172 addr = %conn.addr,
173 "rejecting connection with the same key"
174 );
175 self.spawn_hello(conn, Hello::BackOff(Duration::MAX));
176 continue
177 }
178 let Some(party) = self.parties.get_mut(&conn.key) else {
179 info!(
180 name = %self.conf.name,
181 node = %self.key,
182 peer = %conn.key,
183 addr = %conn.addr,
184 "unknown party"
185 );
186 self.spawn_hello(conn, Hello::BackOff(self.conf.backoff_duration));
187 continue
188 };
189 if party.ip_addr_mismatch(conn.addr.ip()) {
190 warn!(
191 name = %self.conf.name,
192 node = %self.key,
193 peer = %conn.key,
194 addr = %conn.addr,
195 "party has invalid ip addr"
196 );
197 self.spawn_hello(conn, Hello::BackOff(self.conf.backoff_duration));
198 continue
199 }
200 self.spawn_hello(conn, Hello::Ok);
201 }
202 Ok(Err(err)) => {
203 self.metrics.set(&self.key, ACCEPT_TASKS, self.accept_tasks.len());
204 warn!(name = %self.conf.name, node = %self.key, %err, "handshake failed")
205 }
206 Err(err) => {
207 self.metrics.set(&self.key, ACCEPT_TASKS, self.accept_tasks.len());
208 if err.is_panic() {
209 error!(
210 name = %self.conf.name,
211 node = %self.key,
212 %err,
213 "handshake task panic"
214 )
215 }
216 }
217 },
218
219 Some(r) = self.hello_tasks.join_next() => match r {
220 (key, Ok(Ok((our_hello, conn, their_hello)))) => {
221 self.metrics.set(&self.key, HELLO_TASKS, self.hello_tasks.len());
222 if key == self.key {
223 // This case has been addressed already by rejecting the peer,
224 // i.e. we told the peer to backoff forever.
225 continue
226 }
227 let Some(party) = self.parties.get_mut(&key) else {
228 info!(
229 name = %self.conf.name,
230 node = %self.key,
231 peer = %key,
232 addr = %conn.addr,
233 "unknown party"
234 );
235 continue
236 };
237 if !(our_hello.is_ok() && their_hello.is_some_and(|h| h.is_ok())) {
238 warn!(
239 name = %self.conf.name,
240 node = %self.key,
241 peer = %key,
242 addr = %conn.addr,
243 ours = ?our_hello,
244 theirs = ?their_hello,
245 "hello failed"
246 );
247 continue
248 }
249 match party.peer.take() {
250 PeerState::None => {
251 self.connect_tasks.abort(&key);
252 let peer = Peer::builder()
253 .config(self.conf.clone())
254 .budget(self.conf.peer_budget)
255 .inbound(self.ibound.clone())
256 .messages(party.outbox.clone())
257 .retry(party.retry.clone())
258 .metrics(self.metrics.clone())
259 .build();
260 let cancel = CancellationToken::new();
261 party.peer = PeerState::Connected(cancel.clone());
262 self.spawn_peer(key, peer, conn, cancel);
263 }
264 PeerState::Reconnect(peer) => {
265 self.connect_tasks.abort(&key);
266 let cancel = CancellationToken::new();
267 party.peer = PeerState::Connected(cancel.clone());
268 self.spawn_peer(key, peer, conn, cancel);
269 }
270 PeerState::Connected(cancel) => {
271 if key > self.key {
272 info!(
273 name = %self.conf.name,
274 node = %self.key,
275 peer = %key,
276 addr = %conn.addr,
277 "replacing connection with accepted one"
278 );
279 cancel.cancel();
280 party.peer = PeerState::Replace(conn);
281 } else {
282 party.peer = PeerState::Connected(cancel);
283 }
284 }
285 PeerState::Replace(_) => {
286 party.peer = PeerState::Replace(conn);
287 }
288 }
289 }
290 (key, Ok(Err(err))) => {
291 self.metrics.set(&self.key, HELLO_TASKS, self.hello_tasks.len());
292 warn!(
293 name = %self.conf.name,
294 node = %self.key,
295 peer = %key,
296 %err,
297 "hello task error"
298 )
299 }
300 (key, Err(err)) => {
301 self.metrics.set(&self.key, HELLO_TASKS, self.hello_tasks.len());
302 if err.is_panic() {
303 error!(
304 name = %self.conf.name,
305 node = %self.key,
306 peer = %key,
307 %err,
308 "hello task panic"
309 )
310 }
311 }
312 },
313
314 Some(x) = self.connect_tasks.join_next() => match x {
315 (key, Ok(conn)) => {
316 self.metrics.set(&self.key, CONNECT_TASKS, self.connect_tasks.len());
317 let Some(party) = self.parties.get_mut(&key) else {
318 debug!(
319 name = %self.conf.name,
320 node = %self.key,
321 peer = %key,
322 addr = %conn.addr,
323 "party has been removed"
324 );
325 continue
326 };
327 match party.peer.take() {
328 PeerState::None => {
329 let peer = Peer::builder()
330 .config(self.conf.clone())
331 .budget(self.conf.peer_budget)
332 .inbound(self.ibound.clone())
333 .messages(party.outbox.clone())
334 .retry(party.retry.clone())
335 .metrics(self.metrics.clone())
336 .build();
337 let cancel = CancellationToken::new();
338 party.peer = PeerState::Connected(cancel.clone());
339 self.spawn_peer(key, peer, conn, cancel);
340 }
341 PeerState::Reconnect(peer) => {
342 let cancel = CancellationToken::new();
343 party.peer = PeerState::Connected(cancel.clone());
344 self.spawn_peer(key, peer, conn, cancel);
345 }
346 PeerState::Connected(cancel) => {
347 if key < self.key {
348 info!(
349 name = %self.conf.name,
350 node = %self.key,
351 peer = %key,
352 addr = %conn.addr,
353 "replacing connection with outgoing one"
354 );
355 cancel.cancel();
356 party.peer = PeerState::Replace(conn);
357 } else {
358 party.peer = PeerState::Connected(cancel);
359 }
360 }
361 PeerState::Replace(_) => {
362 party.peer = PeerState::Replace(conn);
363 }
364 }
365 }
366 (key, Err(err)) => {
367 self.metrics.set(&self.key, CONNECT_TASKS, self.connect_tasks.len());
368 if err.is_panic() {
369 error!(
370 name = %self.conf.name,
371 node = %self.key,
372 peer = %key,
373 %err,
374 "connect task panic"
375 );
376 if let Some(party) = self.parties.get_mut(&key)
377 && matches!(party.peer, PeerState::None | PeerState::Reconnect(_))
378 {
379 let addr = party.addr.clone();
380 self.spawn_connect(key, addr);
381 }
382 }
383 }
384 },
385
386 Some(p) = self.peer_tasks.join_next() => match p {
387 (key, Ok(peer)) => {
388 self.metrics.set(&self.key, PEER_TASKS, self.peer_tasks.len());
389 if self.ibound.is_closed() {
390 return
391 }
392 let Some(party) = self.parties.get_mut(&key) else {
393 debug!(
394 name = %self.conf.name,
395 node = %self.key,
396 peer = %key,
397 "party has been removed"
398 );
399 continue
400 };
401 if let PeerState::Replace(conn) = party.peer.take() {
402 let cancel = CancellationToken::new();
403 party.peer = PeerState::Connected(cancel.clone());
404 self.spawn_peer(key, peer, conn, cancel);
405 } else {
406 let addr = party.addr.clone();
407 party.peer = PeerState::Reconnect(peer);
408 self.spawn_connect(key, addr);
409 }
410 }
411 (key, Err(err)) => {
412 self.metrics.set(&self.key, PEER_TASKS, self.peer_tasks.len());
413 if err.is_panic() {
414 error!(
415 name = %self.conf.name,
416 node = %self.key,
417 peer = %key,
418 %err,
419 "peer task panic"
420 );
421 if self.ibound.is_closed() {
422 return
423 }
424 if let Some(party) = self.parties.get_mut(&key) {
425 let addr = party.addr.clone();
426 party.peer = PeerState::None;
427 self.spawn_connect(key, addr);
428 }
429 }
430 }
431 },
432
433 r = self.next_slot.changed() => {
434 if r.is_err() {
435 return
436 }
437 let s = *self.next_slot.borrow_and_update();
438 debug_assert!(s > self.lower_bound); // ensured by `NetworkSender::gc`
439 self.lower_bound = s;
440 self.metrics.set(&self.key, LOWER_BOUND, u64::from(s) as usize);
441 for party in self.parties.values() {
442 party.outbox.gc(s);
443 party.retry.gc(s);
444 }
445 }
446
447 cmd = self.obound.recv() => {
448 self.metrics.set(&self.key, CHANNEL_SIZE, self.obound.len());
449 match cmd {
450 Some(Command::Peer(PeerCommand::Add(role, parties))) => {
451 for (k, a) in parties {
452 if k == self.key {
453 self.role = role;
454 continue
455 }
456 if let Some(p) = self.parties.get_mut(&k) {
457 if p.addr == a {
458 p.role = role;
459 } else {
460 info!(
461 name = %self.conf.name,
462 node = %self.key,
463 peer = %k,
464 addr = %a,
465 "updating party address"
466 );
467 p.addr = a.clone();
468 p.role = role;
469 self.connect_tasks.abort(&k);
470 if let PeerState::Connected(cancel) = &p.peer {
471 cancel.cancel()
472 } else {
473 self.spawn_connect(k, a)
474 }
475 }
476 continue
477 }
478 info!(
479 name = %self.conf.name,
480 node = %self.key,
481 peer = %k,
482 addr = %a,
483 "adding new peer"
484 );
485 self.parties.insert(k, Party::new(self.conf.clone(), role, a.clone()));
486 self.spawn_connect(k, a)
487 }
488 }
489 Some(Command::Peer(PeerCommand::Remove(peers))) => {
490 for k in &peers {
491 if *k == self.key {
492 info!(
493 name = %self.conf.name,
494 node = %self.key,
495 "removing self sets role to passive"
496 );
497 self.role = Role::Passive;
498 continue
499 }
500 info!(
501 name = %self.conf.name,
502 node = %self.key,
503 peer = %k,
504 "removing peer"
505 );
506 self.parties.remove(k);
507 self.connect_tasks.abort(k);
508 self.peer_tasks.abort(k);
509 }
510 }
511 Some(Command::Peer(PeerCommand::Assign(role, peers))) => {
512 for k in &peers {
513 if *k == self.key {
514 self.role = role;
515 continue
516 }
517 if let Some(p) = self.parties.get_mut(k) {
518 info!(
519 name = %self.conf.name,
520 node = %self.key,
521 peer = %k,
522 %role,
523 "assigning role to peer"
524 );
525 p.role = role
526 } else {
527 warn!(
528 name = %self.conf.name,
529 node = %self.key,
530 peer = %k,
531 role = %role,
532 "peer to assign role to not found"
533 );
534 }
535 }
536 }
537 Some(Command::Send(cmd)) => match cmd.action {
538 SendAction::Unicast(to, m) => {
539 if cmd.slot < self.lower_bound {
540 continue
541 }
542
543 if to == self.key {
544 trace!(name = %self.conf.name, node = %self.key, "sending message");
545 if let Err(err) = self.ibound.send((self.key, m.into(), None)) {
546 warn!(
547 name = %self.conf.name,
548 node = %self.key,
549 err = %err,
550 "channel closed"
551 );
552 return
553 }
554 trace!(name = %self.conf.name, node = %self.key, "message delivered");
555 continue
556 }
557
558 let msgid = self.next_msgid();
559 let bytes = append_trailer(cmd.retry, cmd.slot, msgid, m);
560
561 if let Some(party) = self.parties.get(&to) {
562 party.outbox.enqueue(cmd.slot, msgid, (cmd.retry, bytes));
563 } else {
564 warn!(
565 name = %self.conf.name,
566 node = %self.key,
567 peer = %to,
568 "unicast target not found"
569 );
570 }
571 }
572 SendAction::Multicast(parties, m) => {
573 if cmd.slot < self.lower_bound {
574 continue
575 }
576
577 let msgid = self.next_msgid();
578 let bytes = append_trailer(cmd.retry, cmd.slot, msgid, m);
579
580 if parties.contains(&self.key) {
581 let bytes = remove_trailer(bytes.clone());
582 trace!(name = %self.conf.name, node = %self.key, "sending message");
583 if let Err(err) = self.ibound.send((self.key, bytes, None)) {
584 warn!(
585 name = %self.conf.name,
586 node = %self.key,
587 err = %err,
588 "channel closed"
589 );
590 return
591 }
592 trace!(name = %self.conf.name, node = %self.key, "message delivered");
593 }
594
595 for (to, party) in &self.parties {
596 if !parties.contains(to) {
597 continue
598 }
599 trace!(name = %self.conf.name, node = %self.key, %to, "sending message");
600 party.outbox.enqueue(cmd.slot, msgid, (cmd.retry, bytes.clone()));
601 }
602 }
603 SendAction::Broadcast(m) => {
604 if cmd.slot < self.lower_bound {
605 continue
606 }
607
608 let msgid = self.next_msgid();
609 let bytes = append_trailer(cmd.retry, cmd.slot, msgid, m);
610
611 if self.role.is_active() {
612 let bytes = remove_trailer(bytes.clone());
613 trace!(name = %self.conf.name, node = %self.key, "sending message");
614 if let Err(err) = self.ibound.send((self.key, bytes, None)) {
615 warn!(
616 name = %self.conf.name,
617 node = %self.key,
618 err = %err,
619 "channel closed"
620 );
621 return
622 }
623 trace!(name = %self.conf.name, node = %self.key, "message delivered");
624 }
625 for (key, party) in &self.parties {
626 if party.role.is_active() {
627 trace!(
628 name = %self.conf.name,
629 node = %self.key,
630 to = %key,
631 "sending message"
632 );
633 party.outbox.enqueue(cmd.slot, msgid, (cmd.retry, bytes.clone()));
634 }
635 }
636 }
637 }
638 Some(Command::Shutdown(tx)) => {
639 debug!(name = %self.conf.name, node = %self.key, "shutting down");
640 let _ = tx.send(());
641 return
642 }
643 None => return
644 }
645 }
646 }
647 }
648 }
649
650 fn spawn_connect(&mut self, key: PublicKey, addr: NetAddr) {
651 if self.key == key {
652 return;
653 }
654 debug!(
655 name = %self.conf.name,
656 node = %self.key,
657 peer = %key,
658 addr = %addr,
659 "spawning connect task"
660 );
661 let conn = Connection::connect(self.conf.clone(), key, addr);
662 self.connect_tasks.spawn(key, conn);
663 self.metrics.add(&key, CONNECT_ATTEMPTS, 1);
664 self.metrics
665 .set(&self.key, CONNECT_TASKS, self.connect_tasks.len());
666 }
667
668 fn spawn_accept(&mut self, stream: TcpStream) {
669 debug!(name = %self.conf.name, node = %self.key, "spawning accept task");
670 let conn = Connection::accept(self.conf.clone(), stream);
671 self.accept_tasks.spawn(conn);
672 self.metrics
673 .set(&self.key, ACCEPT_TASKS, self.accept_tasks.len());
674 }
675
676 fn spawn_hello(&mut self, mut conn: Connection, ours: Hello) {
677 debug!(
678 name = %self.conf.name,
679 node = %self.key,
680 peer = %conn.key,
681 addr = %conn.addr,
682 "spawning hello task"
683 );
684
685 self.metrics.add(&conn.key, HELLOS, 1);
686
687 // Under normal circumstances we should never witness more hellos than parties.
688 // An adversary may nevertheless pass the accept stage and since we do not
689 // know the party we inform it to back off. If the remote does not send us
690 // their hello in due time this will occupy the hello task (and a file
691 // descriptor). To put a cap on that we make the reading of hellos contingent
692 // on how many ongoing hellos we already process. Under attack we do not
693 // bother with reading the remote's (unless it is a peer we want), but only
694 // send our hello and drop the connection. NB that this may cause the remote
695 // to see a connection reset.
696 let under_pressure = self.hello_tasks.len() > 2 * self.parties.len();
697
698 self.hello_tasks.abort(&conn.key);
699 self.hello_tasks.spawn(
700 conn.key,
701 until(self.conf.handshake_timeout, async move {
702 let theirs = if ours.is_ok() || !under_pressure {
703 Some(conn.recv_hello().await?)
704 } else {
705 None
706 };
707 conn.send_hello(ours).await?;
708 Ok::<_, NetworkError>((ours, conn, theirs))
709 }),
710 );
711
712 self.metrics
713 .set(&self.key, HELLO_TASKS, self.hello_tasks.len());
714 }
715
716 fn spawn_peer(
717 &mut self,
718 key: PublicKey,
719 mut peer: Peer,
720 conn: Connection,
721 cancel: CancellationToken,
722 ) {
723 debug!(
724 name = %self.conf.name,
725 node = %self.key,
726 peer = %key,
727 addr = %conn.addr,
728 "spawning peer task"
729 );
730 let node = self.key;
731 let name = self.conf.name.clone();
732 let metrics = self.metrics.clone();
733 let addr = conn.addr;
734 self.peer_tasks.spawn(key, async move {
735 let Err(err) = peer.start(conn, cancel).await;
736 if !matches!(err, NetworkError::PeerInterrupt) {
737 warn!(
738 %name,
739 %node,
740 peer = %key,
741 %addr,
742 %err,
743 "peer failure"
744 );
745 metrics.add(&key, ERRORS, 1)
746 }
747 peer
748 });
749 self.metrics
750 .set(&self.key, PEER_TASKS, self.peer_tasks.len());
751 }
752
753 fn next_msgid(&mut self) -> MsgId {
754 let current = self.msgid;
755 self.msgid = MsgId::new(self.msgid.0.wrapping_add(1));
756 current
757 }
758}
759
760impl Party {
761 fn new(c: Arc<Config>, r: Role, a: NetAddr) -> Self {
762 Self {
763 addr: a,
764 role: r,
765 outbox: Queue::new(),
766 retry: DelayQueue::new(c),
767 peer: PeerState::None,
768 }
769 }
770
771 fn ip_addr_mismatch(&self, addr: IpAddr) -> bool {
772 let NetAddr::Inet(ip, _) = &self.addr else {
773 return false;
774 };
775 ip.to_canonical() != addr.to_canonical()
776 }
777}
778
779impl PeerState {
780 fn take(&mut self) -> Self {
781 mem::replace(self, Self::None)
782 }
783}
784
785fn append_trailer(pol: RetryPolicy, slot: Slot, id: MsgId, bytes: Vec<u8>) -> Bytes {
786 let t = match pol {
787 RetryPolicy::Default => Trailer::Std { slot, id },
788 RetryPolicy::NoRetry => Trailer::NoAck { slot },
789 };
790 let mut msg = BytesMut::from(Bytes::from(bytes));
791 msg.extend_from_slice(t.to_bytes().as_ref());
792 msg.freeze()
793}
794
795fn remove_trailer(mut bytes: Bytes) -> Bytes {
796 let _t = Trailer::from_bytes(&mut bytes);
797 debug_assert!(_t.is_some());
798 bytes
799}
800
801// Metrics labels /////////////////////////////////////////////////////////////
802
803/// Current number of accept tasks.
804const ACCEPT_TASKS: &str = "accept_tasks";
805
806/// Current number of channel items.
807const CHANNEL_SIZE: &str = "channel_size";
808
809/// Total number of connect attempts.
810const CONNECT_ATTEMPTS: &str = "connect_attempts";
811
812/// Current number of connect tasks.
813const CONNECT_TASKS: &str = "connect_tasks";
814
815/// Total number of peer errors.
816const ERRORS: &str = "errors";
817
818/// Total number of hello exchanges.
819const HELLOS: &str = "hellos";
820
821/// Current number of hello tasks.
822const HELLO_TASKS: &str = "hello_tasks";
823
824/// Current GC lower bound.
825const LOWER_BOUND: &str = "lower_bound";
826
827/// Current number of peer tasks.
828const PEER_TASKS: &str = "peer_tasks";