[feat] Basic IPNS support

Author: ppodolsky

iroh-p2p/src/dht_records.rs

@@ -0,0 +1,36 @@
+use ahash::AHashMap;
+use libp2p::kad::{GetRecordOk, GetRecordResult, PeerRecord, QueryId};
+use tokio::sync::mpsc::Sender;
+
+pub struct DhtGetQuery {
+    response_channel: Sender<anyhow::Result<PeerRecord>>,
+}
+
+impl DhtGetQuery {
+    pub fn new(response_channel: Sender<anyhow::Result<PeerRecord>>) -> DhtGetQuery {
+        DhtGetQuery { response_channel }
+    }
+}
+
+#[derive(Default)]
+pub struct DhtRecords {
+    current_queries: AHashMap<QueryId, DhtGetQuery>,
+}
+
+impl DhtRecords {
+    pub fn insert(&mut self, query_id: QueryId, query: DhtGetQuery) {
+        self.current_queries.insert(query_id, query);
+    }
+
+    pub fn handle_get_record_result(&mut self, id: QueryId, get_record_result: GetRecordResult) {
+        if let Some(query) = self.current_queries.remove(&id) {
+            match get_record_result {
+                Ok(GetRecordOk::FoundRecord(record)) => {
+                    tokio::spawn(async move { query.response_channel.send(Ok(record)).await.ok() });
+                }
+                Ok(GetRecordOk::FinishedWithNoAdditionalRecord { cache_candidates }) => {}
+                Err(_) => todo!(),
+            }
+        };
+    }
+}

iroh-p2p/src/lib.rs

@@ -1,6 +1,7 @@
 mod behaviour;
 pub mod cli;
 pub mod config;
+mod dht_records;
 mod keys;
 pub mod metrics;
 mod node;

iroh-p2p/src/node.rs

@@ -34,6 +34,7 @@ use tracing::{debug, error, info, trace, warn};
 use iroh_bitswap::{BitswapEvent, Block};
 use iroh_rpc_client::Lookup;
 
+use crate::dht_records::{DhtGetQuery, DhtRecords};
 use crate::keys::{Keychain, Storage};
 use crate::providers::Providers;
 use crate::rpc::{P2p, ProviderRequestKey};
@@ -87,6 +88,7 @@ pub struct Node<KeyStorage: Storage> {
     use_dht: bool,
     bitswap_sessions: BitswapSessions,
     providers: Providers,
+    dht_records: DhtRecords,
     listen_addrs: Vec<Multiaddr>,
 }
 
@@ -175,6 +177,7 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             use_dht: libp2p_config.kademlia,
             bitswap_sessions: Default::default(),
             providers: Providers::new(4),
+            dht_records: DhtRecords::default(),
             listen_addrs,
         })
     }
@@ -522,7 +525,6 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             }
             Event::Kademlia(e) => {
                 libp2p_metrics().record(&e);
-
                 if let KademliaEvent::OutboundQueryProgressed {
                     id, result, step, ..
                 } = e
@@ -627,6 +629,11 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
                                 });
                             }
                         }
+                        QueryResult::GetRecord(get_record_result) => self
+                            .dht_records
+                            .handle_get_record_result(id, get_record_result),
+                        QueryResult::PutRecord(_) => todo!(),
+                        QueryResult::RepublishRecord(_) => todo!(),
                         other => {
                             debug!("Libp2p => Unhandled Kademlia query result: {:?}", other)
                         }
@@ -1061,6 +1068,16 @@ impl<KeyStorage: Storage> Node<KeyStorage> {
             RpcMessage::Shutdown => {
                 return Ok(true);
             }
+            RpcMessage::DhtGet {
+                key,
+                response_channel,
+            } => {
+                if let Some(kad) = self.swarm.behaviour_mut().kad.as_mut() {
+                    let query_id = kad.get_record(key);
+                    self.dht_records
+                        .insert(query_id, DhtGetQuery::new(response_channel));
+                }
+            }
         }
 
         Ok(false)

iroh-p2p/src/rpc.rs

@@ -6,7 +6,6 @@ use futures::{
     stream::{BoxStream, Stream},
     TryFutureExt,
 };
-use iroh_bitswap::Block;
 use iroh_rpc_client::{
     create_server, Lookup, P2pServer, ServerError, ServerSocket, HEALTH_POLL_WAIT,
 };
@@ -27,6 +26,17 @@ use tokio::sync::mpsc::{channel, Sender};
 use tokio::sync::oneshot;
 use tracing::{debug, info, trace};
 
+use async_trait::async_trait;
+use iroh_bitswap::Block;
+use iroh_rpc_types::p2p::{
+    BitswapRequest, BitswapResponse, ConnectByPeerIdRequest, ConnectRequest, DhtGetRequest,
+    DisconnectRequest, GetListeningAddrsResponse, GetPeersResponse, GossipsubAllPeersResponse,
+    GossipsubPublishResponse, GossipsubSubscribeResponse,
+    GossipsubTopicsResponse, Key as ProviderKey, LookupRequest,
+    NotifyNewBlocksBitswapRequest, StopSessionBitswapRequest, VersionResponse,
+};
+use libp2p::kad::PeerRecord;
+
 use super::node::DEFAULT_PROVIDER_LIMIT;
 use crate::VERSION;
 
@@ -231,6 +241,24 @@ impl P2p {
         Ok(stream)
     }
 
+    #[tracing::instrument(skip(self, req))]
+    async fn dht_get(
+        &self,
+        req: DhtGetRequest,
+    ) -> anyhow::Result<BoxStream<'static, anyhow::Result<iroh_rpc_types::p2p::PeerRecord>>> {
+        // ToDo: parametrize size of channel in a proper way
+        let (sender, receiver) = channel(64);
+        let msg = RpcMessage::DhtGet {
+            key: Key::from(req.key.0.to_vec()),
+            response_channel: sender,
+        };
+        self.sender.send(msg).await?;
+        let receiver = tokio_stream::wrappers::ReceiverStream::new(receiver)
+            .map(|peer_record| peer_record.map(|peer_record| peer_record.into()));
+
+        Ok(Box::pin(receiver))
+    }
+
     #[tracing::instrument(skip(self, req))]
     async fn start_providing(self, req: StartProvidingRequest) -> Result<()> {
         trace!("received StartProviding request: {:?}", req.key);
@@ -620,6 +648,10 @@ pub enum RpcMessage {
     ExternalAddrs(oneshot::Sender<Vec<Multiaddr>>),
     Listeners(oneshot::Sender<Vec<Multiaddr>>),
     LocalPeerId(oneshot::Sender<PeerId>),
+    DhtGet {
+        key: Key,
+        response_channel: Sender<Result<PeerRecord>>,
+    },
     BitswapRequest {
         ctx: u64,
         cids: Vec<Cid>,

iroh-resolver/src/resolver.rs

@@ -35,6 +35,11 @@ use crate::dns_resolver::{Config, DnsResolver};
 
 pub const IROH_STORE: &str = "iroh-store";
 
+pub(crate) mod ipns_pb {
+    #![allow(clippy::all)]
+    include!(concat!(env!("OUT_DIR"), "/ipns_pb.rs"));
+}
+
 // ToDo: Remove this function
 // Related issue: https://github.com/n0-computer/iroh/issues/593
 fn from_peer_id(id: &str) -> Option<libipld::Multihash> {
@@ -1066,8 +1071,12 @@ impl<T: ContentLoader> Resolver<T> {
                 },
                 PathType::Ipns => match current.root() {
                     CidOrDomain::Cid(ref c) => {
-                        let c = self.load_ipns_record(c).await?;
-                        current = Path::from_cid(c);
+                        let resolved_record = Bytes::from(self.load_ipns_record(c).await?.value);
+                        let proto = ipns_pb::IpnsEntry::decode(resolved_record).unwrap();
+                        current = Path::from_str(
+                            &String::from_utf8(proto.value.clone().unwrap()).unwrap(),
+                        )
+                        .unwrap();
                     }
                     CidOrDomain::Domain(ref domain) => {
                         let mut records = self.dns_resolver.resolve_dnslink(domain).await?;
@@ -1102,7 +1111,7 @@ impl<T: ContentLoader> Resolver<T> {
 
     #[tracing::instrument(skip(self))]
     async fn load_ipns_record(&self, cid: &Cid) -> Result<Cid> {
-        todo!()
+        self.loader.load_record_from_dht(cid).await
     }
 }
 

iroh-rpc-client/src/network.rs

@@ -1,3 +1,4 @@
+use crate::{StatusType, HEALTH_POLL_WAIT};
 use anyhow::Result;
 use async_stream::stream;
 use bytes::Bytes;
@@ -9,8 +10,6 @@ use libp2p::{Multiaddr, PeerId};
 use std::collections::{HashMap, HashSet};
 use tracing::{debug, warn};
 
-use crate::{StatusType, HEALTH_POLL_WAIT};
-
 #[derive(Debug, Clone)]
 pub struct P2pClient {
     client: quic_rpc::RpcClient<P2pService, crate::ChannelTypes>,
@@ -101,6 +100,17 @@ impl P2pClient {
         Ok(providers_stream)
     }
 
+    #[tracing::instrument(skip(self))]
+    pub async fn dht_get(
+        &self,
+        key: &libp2p::kad::record::Key,
+    ) -> Result<impl Stream<Item = Result<iroh_rpc_types::p2p::PeerRecord>>> {
+        let key = Key(key.to_vec().into());
+        let res = self.client.server_streaming(DhtGetRequest { key }).await?;
+        let peer_records_stream = res.map(|p| Ok(p??));
+        Ok(peer_records_stream)
+    }
+
     #[tracing::instrument(skip(self))]
     pub async fn start_providing(&self, key: &Cid) -> Result<()> {
         let key = Key(key.hash().to_bytes().into());

iroh-rpc-types/src/p2p.rs

@@ -8,6 +8,7 @@ use quic_rpc::{
 };
 use serde::{Deserialize, Serialize};
 use std::collections::BTreeMap;
+use std::time::{Duration, Instant};
 
 use crate::{RpcResult, VersionRequest, VersionResponse, WatchRequest, WatchResponse};
 
@@ -40,6 +41,20 @@ pub struct ListenersResponse {
     pub addrs: Vec<Multiaddr>,
 }
 
+#[derive(Serialize, Deserialize, Debug)]
+pub struct DhtGetRequest {
+    pub key: Key,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+pub struct PeerRecord {
+    pub peer_id: Option<Bytes>,
+    pub key: Bytes,
+    pub value: Bytes,
+    pub ttl: u32,
+    pub publisher: Option<Bytes>,
+}
+
 #[derive(Serialize, Deserialize, Debug)]
 pub struct BitswapRequest {
     pub cid: Cid,
@@ -219,6 +234,7 @@ pub enum P2pRequest {
     Watch(WatchRequest),
     Version(VersionRequest),
     Shutdown(ShutdownRequest),
+    DhtGet(DhtGetRequest),
     FetchBitswap(BitswapRequest),
     FetchProviderDht(FetchProvidersDhtRequest),
     StopSessionBitswap(StopSessionBitswapRequest),
@@ -250,6 +266,7 @@ pub enum P2pRequest {
 pub enum P2pResponse {
     Watch(WatchResponse),
     Version(VersionResponse),
+    DhtGet(RpcResult<PeerRecord>),
     FetchBitswap(RpcResult<BitswapResponse>),
     FetchProviderDht(RpcResult<FetchProvidersDhtResponse>),
     GetListeningAddrs(RpcResult<GetListeningAddrsResponse>),
@@ -291,6 +308,14 @@ impl RpcMsg<P2pService> for ShutdownRequest {
     type Response = RpcResult<()>;
 }
 
+impl Msg<P2pService> for DhtGetRequest {
+    type Response = RpcResult<PeerRecord>;
+
+    type Update = Self;
+
+    type Pattern = ServerStreaming;
+}
+
 impl RpcMsg<P2pService> for BitswapRequest {
     type Response = RpcResult<BitswapResponse>;
 }
@@ -394,3 +419,53 @@ impl RpcMsg<P2pService> for ExternalAddrsRequest {
 impl RpcMsg<P2pService> for ListenersRequest {
     type Response = RpcResult<ListenersResponse>;
 }
+
+impl From<libp2p::kad::PeerRecord> for PeerRecord {
+    fn from(peer_record: libp2p::kad::PeerRecord) -> Self {
+        PeerRecord {
+            peer_id: peer_record.peer.map(|peer| Bytes::from(peer.to_bytes())),
+            key: peer_record.record.key.to_vec().into(),
+            ttl: peer_record
+                .record
+                .expires
+                .map(|t| {
+                    let now = Instant::now();
+                    if t > now {
+                        (t - now).as_secs() as u32
+                    } else {
+                        1 // because 0 means "does not expire"
+                    }
+                })
+                .unwrap_or(0),
+            value: peer_record.record.value.into(),
+            publisher: peer_record
+                .record
+                .publisher
+                .map(|publisher| Bytes::from(publisher.to_bytes())),
+        }
+    }
+}
+
+impl From<PeerRecord> for libp2p::kad::PeerRecord {
+    fn from(peer_record: PeerRecord) -> Self {
+        let peer = peer_record
+            .peer_id
+            .as_ref()
+            .map(|peer_id| libp2p::PeerId::from_bytes(peer_id).unwrap());
+        let key = libp2p::kad::record::Key::from(peer_record.key.to_vec());
+        let value = peer_record.value;
+        let publisher = peer_record
+            .publisher
+            .as_ref()
+            .map(|publisher| libp2p::PeerId::from_bytes(publisher).unwrap());
+        let expires = (peer_record.ttl > 0)
+            .then(|| Instant::now() + Duration::from_secs(peer_record.ttl as u64));
+        let record = libp2p::kad::Record {
+            key,
+            value: value.to_vec(),
+            publisher,
+            expires,
+        };
+        libp2p::kad::PeerRecord { peer, record }
+    }
+}

iroh-share/src/p2p_node.rs

@@ -127,6 +127,10 @@ impl ContentLoader for Loader {
     async fn has_cid(&self, cid: &Cid) -> Result<bool> {
         Ok(self.client.try_store()?.has(*cid).await?)
     }
+
+    async fn load_record_from_dht(&self, _: &Cid) -> Result<iroh_rpc_types::p2p::PeerRecord> {
+        todo!()
+    }
 }
 
 impl P2pNode {

iroh-unixfs/Cargo.toml

@@ -22,6 +22,7 @@ fastmurmur3.workspace = true
 futures.workspace = true
 iroh-metrics = { workspace = true, features = ["resolver", "gateway"] }
 iroh-rpc-client.workspace = true
+iroh-rpc-types.workspace = true
 iroh-util.workspace = true
 libipld.workspace = true
 libp2p = { workspace = true, features = ["serde"] }

iroh-unixfs/build.rs

@@ -1,6 +1,13 @@
 fn main() {
     prost_build::Config::new()
-        .bytes([".unixfs_pb.Data", ".merkledag_pb.PBNode.Data"])
-        .compile_protos(&["src/unixfs.proto", "src/merkledag.proto"], &["src"])
+        .bytes([
+            ".unixfs_pb.Data.Data",
+            ".merkledag_pb.PBNode.Data",
+            ".ipns_pb.IpnsEntry.Data",
+        ])
+        .compile_protos(
+            &["src/unixfs.proto", "src/merkledag.proto", "src/ipns.proto"],
+            &["src"],
+        )
         .unwrap();
 }