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",
@@ -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" }
@@ -100,7 +102,7 @@ libp2p-request-response = { version = "0.28.0", path = "protocols/request-respon
 libp2p-server = { version = "0.12.8", path = "misc/server" }
 libp2p-stream = { version = "0.2.0-alpha.1", path = "protocols/stream" }
 libp2p-swarm = { version = "0.45.2", path = "swarm" }
-libp2p-swarm-derive = { version = "=0.35.0", path = "swarm-derive" } # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.
+libp2p-swarm-derive = { version = "=0.35.0", path = "swarm-derive" }                            # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.
 libp2p-swarm-test = { version = "0.5.0", path = "swarm-test" }
 libp2p-tcp = { version = "0.42.0", path = "transports/tcp" }
 libp2p-tls = { version = "0.5.0", path = "transports/tls" }

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,179 @@
+use std::{collections::VecDeque, task::Poll};
+
+use libp2p_core::{Multiaddr, PeerId};
+use libp2p_swarm::{dummy, NetworkBehaviour};
+
+use crate::store::{AddressSource, Store};
+
+/// Events generated by [`Behaviour`] and emitted back to the [`libp2p_swarm::Swarm`].
+pub enum Event<S: Store> {
+    /// 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,
+    },
+    Store(S::ToSwarm),
+}
+
+pub struct Behaviour<S: Store> {
+    store: S,
+    /// Pending Events to be emitted back to the  [`libp2p_swarm::Swarm`].
+    pending_events: VecDeque<Event<S>>,
+}
+
+impl<'a, S> Behaviour<S>
+where
+    S: Store + 'static,
+{
+    pub fn new(store: S) -> Self {
+        Self {
+            store,
+            pending_events: VecDeque::new(),
+        }
+    }
+
+    /// 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,
+        signed_record: &libp2p_core::PeerRecord,
+        source: AddressSource,
+    ) {
+        self.store
+            .update_certified_address(signed_record, source, false);
+        self.pending_events.push_back(Event::RecordUpdated {
+            peer: signed_record.peer_id(),
+        });
+    }
+
+    /// 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_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<S> NetworkBehaviour for Behaviour<S>
+where
+    S: Store + 'static,
+    <S as Store>::ToSwarm: Send + Sync,
+{
+    type ConnectionHandler = dummy::ConnectionHandler;
+
+    type ToSwarm = Event<S>;
+
+    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");
+        Ok(self
+            .store
+            .addresses_of_peer(&peer)
+            .map(|i| i.cloned().collect())
+            .unwrap_or_default())
+    }
+
+    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);
+        };
+    }
+
+    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));
+        }
+        if let Some(ev) = self.store.poll(cx) {
+            self.pending_events.push_back(Event::Store(ev));
+        };
+        Poll::Pending
+    }
+}

misc/peer-store/src/lib.rs

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