feat(peer-store): introduce libp2p-peer-store

Cargo.toml

@@ -64,7 +64,7 @@ members = [
     "transports/websocket-websys",
     "transports/websocket",
     "transports/webtransport-websys",
-    "wasm-tests/webtransport-tests",
+    "wasm-tests/webtransport-tests", "misc/peer-store",
 ]
 resolver = "2"
 

misc/peer-store/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "libp2p-peer-store"
+version = "0.1.0"
+edition = "2021"
+rust-version.workspace = true
+
+[dependencies]
+libp2p-core = { workspace = true }
+libp2p-swarm = { workspace = true }
+lru = "*"
+
+[lints]
+workspace = true

misc/peer-store/src/behaviour.rs

@@ -0,0 +1,156 @@
+use std::{
+    collections::{HashSet, VecDeque},
+    task::Poll,
+};
+
+use libp2p_core::{Multiaddr, PeerId};
+use libp2p_swarm::{dummy, FromSwarm, NetworkBehaviour};
+
+use crate::store::Store;
+
+/// Events of this behaviour that will be emmitted to the swarm.
+pub enum Event {
+    RecordUpdated { peer: PeerId },
+}
+
+pub struct Behaviour<S> {
+    store: S,
+    /// Peers that are currently connected.
+    connected_peers: HashSet<PeerId>,
+    /// Events that will be emitted.
+    pending_events: VecDeque<Event>,
+}
+
+impl<S> Behaviour<S>
+where
+    S: Store + 'static,
+{
+    pub fn new(store: S) -> Self {
+        Self {
+            store,
+            connected_peers: HashSet::new(),
+            pending_events: VecDeque::new(),
+        }
+    }
+    /// List peers that are currently connected to this peer.
+    pub fn list_connected(&self) -> impl Iterator<Item = &PeerId> {
+        self.connected_peers.iter()
+    }
+    /// Try to get all observed address of the given peer.  
+    /// Returns `None` when the peer is not in the store.
+    pub fn address_of_peer<'a, 'b>(
+        &'a self,
+        peer: &'b PeerId,
+    ) -> Option<impl Iterator<Item = super::AddressRecord<'a>> + use<'a, 'b, S>> {
+        self.store.addresses_of_peer(peer)
+    }
+    /// Manually update a record.  
+    /// This will always cause an `Event::RecordUpdated` to be emitted.
+    pub fn update_record(&mut self, peer: &PeerId, address: &Multiaddr) {
+        self.store.on_address_update(peer, address);
+        self.pending_events
+            .push_back(Event::RecordUpdated { peer: *peer });
+    }
+    fn on_peer_connect(&mut self, peer: &PeerId) {
+        self.connected_peers.insert(*peer);
+    }
+    fn on_peer_disconnect(&mut self, peer: &PeerId) {
+        self.connected_peers.remove(peer);
+    }
+    fn on_address_update(&mut self, peer: &PeerId, address: &Multiaddr) {
+        if self.store.on_address_update(peer, address) {
+            self.pending_events
+                .push_back(Event::RecordUpdated { peer: *peer });
+        }
+    }
+}
+
+impl<S> NetworkBehaviour for Behaviour<S>
+where
+    S: Store + 'static,
+{
+    type ConnectionHandler = dummy::ConnectionHandler;
+
+    type ToSwarm = Event;
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _connection_id: libp2p_swarm::ConnectionId,
+        peer: libp2p_core::PeerId,
+        _local_addr: &libp2p_core::Multiaddr,
+        remote_addr: &libp2p_core::Multiaddr,
+    ) -> Result<libp2p_swarm::THandler<Self>, libp2p_swarm::ConnectionDenied> {
+        self.on_address_update(&peer, remote_addr);
+        Ok(dummy::ConnectionHandler)
+    }
+
+    fn handle_pending_outbound_connection(
+        &mut self,
+        _connection_id: libp2p_swarm::ConnectionId,
+        maybe_peer: Option<PeerId>,
+        _addresses: &[Multiaddr],
+        _effective_role: libp2p_core::Endpoint,
+    ) -> Result<Vec<Multiaddr>, libp2p_swarm::ConnectionDenied> {
+        if maybe_peer.is_none() {
+            return Ok(Vec::with_capacity(0));
+        }
+        if let Some(i) = self
+            .store
+            .addresses_of_peer(&maybe_peer.expect("already handled"))
+        {
+            return Ok(i.map(|r| r.address).cloned().collect());
+        }
+        Ok(Vec::with_capacity(0))
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _connection_id: libp2p_swarm::ConnectionId,
+        peer: libp2p_core::PeerId,
+        addr: &libp2p_core::Multiaddr,
+        _role_override: libp2p_core::Endpoint,
+        _port_use: libp2p_core::transport::PortUse,
+    ) -> Result<libp2p_swarm::THandler<Self>, libp2p_swarm::ConnectionDenied> {
+        self.on_address_update(&peer, addr);
+        Ok(dummy::ConnectionHandler)
+    }
+
+    fn on_swarm_event(&mut self, event: libp2p_swarm::FromSwarm) {
+        match event {
+            FromSwarm::ConnectionClosed(info) => {
+                if info.remaining_established < 1 {
+                    self.on_peer_disconnect(&info.peer_id);
+                }
+            }
+            FromSwarm::ConnectionEstablished(info) => {
+                if info.other_established == 0 {
+                    self.on_peer_connect(&info.peer_id);
+                }
+            }
+            FromSwarm::NewExternalAddrOfPeer(info) => {
+                self.on_address_update(&info.peer_id, info.addr);
+            }
+            _ => {}
+        }
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        _peer_id: libp2p_core::PeerId,
+        _connection_id: libp2p_swarm::ConnectionId,
+        _event: libp2p_swarm::THandlerOutEvent<Self>,
+    ) {
+        unreachable!("No event will be produced by a dummy handler.")
+    }
+
+    fn poll(
+        &mut self,
+        _cx: &mut std::task::Context<'_>,
+    ) -> std::task::Poll<libp2p_swarm::ToSwarm<Self::ToSwarm, libp2p_swarm::THandlerInEvent<Self>>>
+    {
+        if let Some(ev) = self.pending_events.pop_front() {
+            return Poll::Ready(libp2p_swarm::ToSwarm::GenerateEvent(ev));
+        }
+        Poll::Pending
+    }
+}

misc/peer-store/src/lib.rs

@@ -0,0 +1,24 @@
+mod behaviour;
+mod store;
+
+use std::time::SystemTime;
+
+pub use behaviour::{Behaviour, Event};
+use libp2p_core::Multiaddr;
+pub use store::{MemoryStore, Store};
+
+pub struct AddressRecord<'a> {
+    address: &'a Multiaddr,
+    last_seen: &'a SystemTime,
+}
+impl<'a> AddressRecord<'a> {
+    /// The address of this record.
+    pub fn address(&self) -> &Multiaddr {
+        self.address
+    }
+    /// How much time has passed since the address is last reported wrt. current time.  
+    /// This may fail because of system time change.
+    pub fn last_seen(&self) -> Result<std::time::Duration, std::time::SystemTimeError> {
+        self.last_seen.elapsed()
+    }
+}

misc/peer-store/src/store.rs

@@ -0,0 +1,86 @@
+use std::{collections::HashMap, num::NonZeroUsize, time::SystemTime};
+
+use libp2p_core::{Multiaddr, PeerId};
+use lru::LruCache;
+
+/// A store that
+/// - keep track of currently connected peers;
+/// - contains all observed addresses of peers;
+pub trait Store {
+    /// Update an address record.  
+    /// Return `true` when the address is new.  
+    fn on_address_update(&mut self, peer: &PeerId, address: &Multiaddr) -> bool;
+    fn addresses_of_peer(
+        &self,
+        peer: &PeerId,
+    ) -> Option<impl Iterator<Item = super::AddressRecord>>;
+}
+
+pub(crate) struct PeerAddressRecord {
+    addresses: LruCache<Multiaddr, AddressRecord>,
+}
+impl PeerAddressRecord {
+    pub(crate) fn records(&self) -> impl Iterator<Item = super::AddressRecord> {
+        self.addresses
+            .iter()
+            .map(|(address, record)| super::AddressRecord {
+                address,
+                last_seen: &record.last_seen,
+            })
+    }
+    pub(crate) fn new(address: &Multiaddr) -> Self {
+        let mut address_book = LruCache::new(NonZeroUsize::new(8).expect("8 is greater than 0"));
+        address_book.get_or_insert(address.clone(), AddressRecord::new);
+        Self {
+            addresses: address_book,
+        }
+    }
+    pub(crate) fn on_address_update(&mut self, address: &Multiaddr) -> bool {
+        if let Some(record) = self.addresses.get_mut(address) {
+            record.update_last_seen();
+            false
+        } else {
+            self.addresses.get_or_insert(address.clone(), AddressRecord::new);
+            true
+        }
+    }
+}
+
+pub(crate) struct AddressRecord {
+    /// The time when the address is last seen.
+    last_seen: SystemTime,
+}
+impl AddressRecord {
+    pub(crate) fn new() -> Self {
+        Self {
+            last_seen: SystemTime::now(),
+        }
+    }
+    pub(crate) fn update_last_seen(&mut self) {
+        self.last_seen = SystemTime::now();
+    }
+}
+
+/// A in-memory store.
+pub struct MemoryStore {
+    /// An address book of peers regardless of their status(connected or not).
+    address_book: HashMap<PeerId, PeerAddressRecord>,
+}
+
+impl Store for MemoryStore {
+    fn on_address_update(&mut self, peer: &PeerId, address: &Multiaddr) -> bool {
+        if let Some(record) = self.address_book.get_mut(peer) {
+            record.on_address_update(address)
+        } else {
+            self.address_book
+                .insert(*peer, PeerAddressRecord::new(address));
+            true
+        }
+    }
+    fn addresses_of_peer(
+        &self,
+        peer: &PeerId,
+    ) -> Option<impl Iterator<Item = super::AddressRecord>> {
+        self.address_book.get(peer).map(|record| record.records())
+    }
+}