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

Cargo.toml

@@ -26,6 +26,7 @@ members = [
     "misc/memory-connection-limits",
     "misc/metrics",
     "misc/multistream-select",
+    "misc/peer-store",
     "misc/quick-protobuf-codec",
     "misc/quickcheck-ext",
     "misc/rw-stream-sink",
@@ -64,7 +65,7 @@ members = [
     "transports/websocket-websys",
     "transports/websocket",
     "transports/webtransport-websys",
-    "wasm-tests/webtransport-tests",
+    "wasm-tests/webtransport-tests", "misc/peer-store",
 ]
 resolver = "2"
 
@@ -89,6 +90,7 @@ libp2p-memory-connection-limits = { version = "0.3.1", path = "misc/memory-conne
 libp2p-metrics = { version = "0.15.0", path = "misc/metrics" }
 libp2p-mplex = { version = "0.42.0", path = "muxers/mplex" }
 libp2p-noise = { version = "0.45.1", path = "transports/noise" }
+libp2p-peer-store = { version = "0.1.0", path = "misc/peer-store" }
 libp2p-perf = { version = "0.4.0", path = "protocols/perf" }
 libp2p-ping = { version = "0.45.1", path = "protocols/ping" }
 libp2p-plaintext = { version = "0.42.0", path = "transports/plaintext" }

libp2p/CHANGELOG.md

@@ -1,3 +1,8 @@
+## 0.55.0(unreleased)
+
+- Introduce `libp2p-peer-store`.
+  See [PR 5724](https://github.com/libp2p/rust-libp2p/pull/5724).
+
 ## 0.54.2
 
 - Add `with_connection_timeout` on `SwarmBuilder` to allow configuration of the connection_timeout parameter.

libp2p/Cargo.toml

@@ -68,6 +68,7 @@ mdns = ["dep:libp2p-mdns"]
 memory-connection-limits = ["dep:libp2p-memory-connection-limits"]
 metrics = ["dep:libp2p-metrics"]
 noise = ["dep:libp2p-noise"]
+# peer-store = ["dep:libp2p-peer-store"]
 ping = ["dep:libp2p-ping", "libp2p-metrics?/ping"]
 plaintext = ["dep:libp2p-plaintext"]
 pnet = ["dep:libp2p-pnet"]
@@ -110,6 +111,7 @@ libp2p-identity = { workspace = true, features = ["rand"] }
 libp2p-kad = { workspace = true, optional = true }
 libp2p-metrics = { workspace = true, optional = true }
 libp2p-noise = { workspace = true, optional = true }
+libp2p-peer-store = { workspace = true }
 libp2p-ping = { workspace = true, optional = true }
 libp2p-plaintext = { workspace = true, optional = true }
 libp2p-pnet = { workspace = true, optional = true }

libp2p/src/lib.rs

@@ -81,6 +81,8 @@ pub use libp2p_metrics as metrics;
 #[cfg(feature = "noise")]
 #[doc(inline)]
 pub use libp2p_noise as noise;
+#[doc(inline)]
+pub use libp2p_peer_store as peer_store;
 #[cfg(feature = "ping")]
 #[doc(inline)]
 pub use libp2p_ping as ping;

misc/peer-store/CHANGELOG.md

@@ -0,0 +1,4 @@
+## 0.1.0
+
+- Introduce `libp2p-peer-store`.
+  See [PR 5724](https://github.com/libp2p/rust-libp2p/pull/5724).

misc/peer-store/Cargo.toml

@@ -0,0 +1,21 @@
+[package]
+name = "libp2p-peer-store"
+edition = "2021"
+version = "0.1.0"
+authors = ["drHuangMHT <[email protected]>"]
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+rust-version.workspace = true
+
+[dependencies]
+libp2p-core = { workspace = true }
+libp2p-swarm = { workspace = true }
+lru = "*"
+futures-timer = "*"
+futures-util = "*"
+
+[dev-dependencies]
+libp2p-identity = { workspace = true, features = ["rand"] }
+
+[lints]
+workspace = true

misc/peer-store/src/behaviour.rs

@@ -0,0 +1,233 @@
+use std::{
+    collections::{HashSet, VecDeque},
+    task::Poll,
+    time::Duration,
+};
+
+use futures_timer::Delay;
+use futures_util::FutureExt;
+use libp2p_core::{Multiaddr, PeerId};
+use libp2p_swarm::{dummy, FromSwarm, NetworkBehaviour};
+
+use crate::store::{AddressSource, Store};
+
+/// Events generated by [`Behaviour`] and emitted back to the [`libp2p_swarm::Swarm`].
+pub enum Event {
+    /// The peer's record has been updated.  
+    /// Manually updating a record will always emit this event
+    /// even if it provides no new information.
+    RecordUpdated { peer: PeerId },
+}
+
+pub struct Config {
+    /// The interval for garbage collecting records.
+    check_record_ttl_interval: Duration,
+}
+
+pub struct Behaviour<S> {
+    store: S,
+    /// Peers that are currently connected.
+    connected_peers: HashSet<PeerId>,
+    /// Pending Events to be emitted back to the  [`libp2p_swarm::Swarm`].
+    pending_events: VecDeque<Event>,
+    record_ttl_timer: Option<Delay>,
+    config: Config,
+}
+
+impl<'a, S> Behaviour<S>
+where
+    S: Store<'a> + 'static,
+{
+    pub fn new(store: S, config: Config) -> Self {
+        Self {
+            store,
+            connected_peers: HashSet::new(),
+            pending_events: VecDeque::new(),
+            record_ttl_timer: Some(Delay::new(config.check_record_ttl_interval)),
+            config,
+        }
+    }
+
+    /// 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<'b>(
+        &'a self,
+        peer: &'b PeerId,
+    ) -> Option<impl Iterator<Item = &'a Multiaddr> + use<'a, 'b, S>> {
+        self.store.addresses_of_peer(peer)
+    }
+
+    /// Manually update a record.  
+    /// This will always emit an `Event::RecordUpdated`.
+    pub fn update_address(&mut self, peer: &PeerId, address: &Multiaddr) {
+        self.store
+            .update_address(peer, address, AddressSource::Manual, false);
+        self.pending_events
+            .push_back(Event::RecordUpdated { peer: *peer });
+    }
+
+    /// Should be called when other protocol emits a [`PeerRecord`](libp2p_core::PeerRecord).  
+    /// This will always emit an `Event::RecordUpdated`.
+    pub fn on_signed_peer_record(
+        &mut self,
+        peer: &PeerId,
+        record: libp2p_core::PeerRecord,
+        source: AddressSource,
+    ) {
+        self.store
+            .update_certified_address(peer, record, source, false);
+        self.pending_events
+            .push_back(Event::RecordUpdated { peer: *peer });
+    }
+
+    /// Get a immutable reference to the internal store.
+    pub fn store(&self) -> &S {
+        &self.store
+    }
+
+    /// Get a mutable reference to the internal store.
+    pub fn store_mut(&mut self) -> &mut S {
+        &mut self.store
+    }
+
+    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,
+        source: AddressSource,
+        should_expire: bool,
+    ) {
+        if self
+            .store
+            .update_address(peer, address, source, should_expire)
+        {
+            self.pending_events
+                .push_back(Event::RecordUpdated { peer: *peer });
+        }
+    }
+    fn handle_store_event(&mut self, event: super::store::Event) {
+        use super::store::Event::*;
+        match event {
+            RecordUpdated(peer) => self.pending_events.push_back(Event::RecordUpdated { peer }),
+        }
+    }
+}
+
+impl<'a, S> NetworkBehaviour for Behaviour<S>
+where
+    S: Store<'a> + '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, AddressSource::DirectConnection, false);
+        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::new());
+        }
+        let peer = maybe_peer.expect("already handled");
+        if let Some(unsigned) = self.store.addresses_of_peer(&peer) {
+            if let Some(signed) = self.store.certified_addresses_of_peer(&peer) {
+                return Ok(signed.chain(unsigned).cloned().collect());
+            }
+            return Ok(unsigned.cloned().collect());
+        }
+        Ok(Vec::new())
+    }
+
+    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, AddressSource::DirectConnection, false);
+        Ok(dummy::ConnectionHandler)
+    }
+
+    fn on_swarm_event(&mut self, event: libp2p_swarm::FromSwarm) {
+        if let Some(ev) = self.store.on_swarm_event(&event) {
+            self.handle_store_event(ev);
+        };
+        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);
+                }
+            }
+            _ => {}
+        }
+    }
+
+    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));
+        }
+        self.poll_record_ttl(cx);
+        Poll::Pending
+    }
+}
+
+impl<'a, S> Behaviour<S>
+where
+    S: Store<'a>,
+{
+    /// Garbage collect records.
+    fn poll_record_ttl(&mut self, cx: &mut std::task::Context<'_>) {
+        if let Some(mut timer) = self.record_ttl_timer.take() {
+            if let Poll::Ready(()) = timer.poll_unpin(cx) {
+                self.store.check_ttl();
+                self.record_ttl_timer = Some(Delay::new(self.config.check_record_ttl_interval));
+            }
+            self.record_ttl_timer = Some(timer)
+        }
+    }
+}

misc/peer-store/src/lib.rs

@@ -0,0 +1,6 @@
+mod behaviour;
+pub mod memory_store;
+mod store;
+
+pub use behaviour::{Behaviour, Config, Event};
+pub use store::Store;