Manual P2P Discovery

.npmrc

@@ -8,3 +8,4 @@ node-linker=hoisted
 auto-install-peers=true
 max-old-space-size=4096
 enable-pre-post-scripts=true
+package-manager-strict=false

apps/server/docker/Dockerfile

@@ -111,9 +111,15 @@ ADD --chmod=755 --checksum=sha256:1d127c69218f2cd14964036f2b057c4b2652cda3996c69
 
 COPY --chmod=755 entrypoint.sh /usr/bin/
 
+# P2P config
+ENV SD_DOCKER=true
+
 # Expose webserver
 EXPOSE 8080
 
+# Expose P2P
+EXPOSE 7373
+
 # Create the data directory to store the database
 VOLUME [ "/data" ]
 

core/src/api/mod.rs

@@ -1,7 +1,7 @@
 use crate::{
  invalidate_query,
  node::{
- config::{NodeConfig, NodeConfigP2P, NodePreferences},
+ config::{is_in_docker, NodeConfig, NodeConfigP2P, NodePreferences},
  get_hardware_model_name, HardwareModel,
  },
  old_job::JobProgressEvent,
@@ -117,6 +117,7 @@ struct NodeState {
  config: SanitisedNodeConfig,
  data_path: String,
  device_model: Option<String>,
+ is_in_docker: bool,
 }
 
 pub(crate) fn mount() -> Arc<Router> {
@@ -152,6 +153,7 @@ pub(crate) fn mount() -> Arc<Router> {
  .expect("Found non-UTF-8 path")
  .to_string(),
  device_model: Some(device_model),
+ is_in_docker: is_in_docker(),
  })
  })
  })

core/src/api/nodes.rs

@@ -1,3 +1,5 @@
+use std::collections::HashSet;
+
 use crate::{
  invalidate_query,
  node::config::{P2PDiscoveryState, Port},
@@ -24,6 +26,7 @@ pub(crate) fn mount() -> AlphaRouter<Ctx> {
  pub p2p_ipv6_enabled: Option<bool>,
  pub p2p_discovery: Option<P2PDiscoveryState>,
  pub p2p_remote_access: Option<bool>,
+ pub p2p_manual_peers: Option<HashSet<String>>,
  pub image_labeler_version: Option<String>,
  }
  R.mutation(|node, args: ChangeNodeNameArgs| async move {
@@ -60,6 +63,9 @@ pub(crate) fn mount() -> AlphaRouter<Ctx> {
  if let Some(remote_access) = args.p2p_remote_access {
  config.p2p.remote_access = remote_access;
  };
+ if let Some(manual_peers) = args.p2p_manual_peers {
+ config.p2p.manual_peers = manual_peers;
+ };
 
  #[cfg(feature = "ai")]
  if let Some(version) = args.image_labeler_version {

core/src/api/p2p.rs

@@ -41,6 +41,7 @@ pub(crate) fn mount() -> AlphaRouter<Ctx> {
  false => DiscoveryMethod::Local,
  },
  metadata,
+ addrs: peer.addrs(),
  });
  }
 

core/src/node/config.rs

@@ -8,6 +8,7 @@ use sd_p2p::Identity;
 use sd_utils::error::FileIOError;
 
 use std::{
+ collections::HashSet,
  path::{Path, PathBuf},
  sync::Arc,
 };
@@ -46,17 +47,25 @@ pub enum Port {
 
 impl Port {
  pub fn get(&self) -> u16 {
+ if is_in_docker() {
+ return 7373;
+ }
+
  match self {
  Port::Random => 0,
  Port::Discrete(port) => *port,
  }
  }
 
- pub fn is_random(&self) -> bool {
+ pub fn is_default(&self) -> bool {
  matches!(self, Port::Random)
  }
 }
 
+pub fn is_in_docker() -> bool {
+ std::env::var("SD_DOCKER").as_deref() == Ok("true")
+}
+
 fn default_as_true() -> bool {
  true
 }
@@ -73,14 +82,24 @@ fn skip_if_false(value: &bool) -> bool {
 pub struct NodeConfigP2P {
  #[serde(default)]
  pub discovery: P2PDiscoveryState,
- #[serde(default, skip_serializing_if = "Port::is_random")]
+ #[serde(default, skip_serializing_if = "Port::is_default")]
  pub port: Port,
  #[serde(default = "default_as_true", skip_serializing_if = "skip_if_true")]
  pub ipv4: bool,
  #[serde(default = "default_as_true", skip_serializing_if = "skip_if_true")]
  pub ipv6: bool,
  #[serde(default, skip_serializing_if = "skip_if_false")]
  pub remote_access: bool,
+ /// A list of peer addresses to try and manually connect to, instead of relying on discovery.
+ ///
+ /// All of these are valid values:
+ /// - `localhost`
+ /// - `otbeaumont.me` or `otbeaumont.me:3000`
+ /// - `127.0.0.1` or `127.0.0.1:300`
+ /// - `[::1]` or `[::1]:3000`
+ /// which is why we use `String` not `SocketAddr`
+ #[serde(default)]
+ pub manual_peers: HashSet<String>,
 }
 
 impl Default for NodeConfigP2P {
@@ -91,6 +110,7 @@ impl Default for NodeConfigP2P {
  ipv4: true,
  ipv6: true,
  remote_access: false,
+ manual_peers: Default::default(),
  }
  }
 }

core/src/p2p/events.rs

@@ -1,4 +1,4 @@
-use std::sync::Arc;
+use std::{collections::HashSet, net::SocketAddr, sync::Arc};
 
 use sd_p2p::{flume::bounded, HookEvent, HookId, PeerConnectionCandidate, RemoteIdentity, P2P};
 use serde::Serialize;
@@ -40,6 +40,7 @@ pub enum P2PEvent {
  connection: ConnectionMethod,
  discovery: DiscoveryMethod,
  metadata: PeerMetadata,
+ addrs: HashSet<SocketAddr>,
  },
  // Delete a peer
  PeerDelete {
@@ -113,6 +114,7 @@ impl P2PEvents {
  false => DiscoveryMethod::Local,
  },
  metadata,
+ addrs: peer.addrs(),
  }
  }
  HookEvent::PeerUnavailable(identity) => P2PEvent::PeerDelete { identity },

core/src/p2p/manager.rs

@@ -20,13 +20,13 @@ use sd_p2p::{
 use sd_p2p_tunnel::Tunnel;
 use serde_json::json;
 use std::{
- collections::HashMap,
+ collections::{HashMap, HashSet},
  convert::Infallible,
  sync::{atomic::AtomicBool, Arc, Mutex, PoisonError},
  time::Duration,
 };
 use tower_service::Service;
-use tracing::error;
+use tracing::{error, warn};
 
 use tokio::sync::oneshot;
 use tracing::info;
@@ -184,6 +184,24 @@ impl P2PManager {
  .ipv6 = Some(err.to_string());
  }
 
+ let mut addrs = HashSet::new();
+ for addr in config.p2p.manual_peers {
+ // TODO: We should probs track these errors for the UI
+ let Ok(addr) = tokio::net::lookup_host(&addr)
+ .await
+ .map_err(|err| {
+ warn!("Failed to parse manual peer address '{addr}': {err}");
+ })
+ .and_then(|mut i| i.next().ok_or(()))
+ else {
+ continue;
+ };
+
+ addrs.insert(addr);
+ }
+
+ self.quic.set_manual_peer_addrs(addrs);
+
  let should_revert = match config.p2p.discovery {
  P2PDiscoveryState::Everyone
  // TODO: Make `ContactsOnly` work

crates/p2p/src/peer.rs

@@ -179,6 +179,20 @@ impl Peer {
  .collect()
  }
 
+ pub fn addrs(&self) -> HashSet<SocketAddr> {
+ self.state
+ .read()
+ .unwrap_or_else(PoisonError::into_inner)
+ .discovered
+ .values()
+ .flatten()
+ .filter_map(|addr| match addr {
+ PeerConnectionCandidate::SocketAddr(addr) => Some(*addr),
+ _ => None,
+ })
+ .collect()
+ }
+
  /// Construct a new Quic stream to the peer.
  pub async fn new_stream(&self) -> Result<UnicastStream, NewStreamError> {
  let (addrs, connect_tx) = {

crates/p2p/src/quic/transport.rs

@@ -1,5 +1,5 @@
 use std::{
- collections::HashMap,
+ collections::{HashMap, HashSet},
  net::{Ipv4Addr, Ipv6Addr, SocketAddr},
  str::FromStr,
  sync::{Arc, Mutex, PoisonError, RwLock},
@@ -58,6 +58,9 @@ enum InternalEvent {
  relays: Vec<RelayServerEntry>,
  result: oneshot::Sender<Result<(), String>>,
  },
+ RegisterPeerAddr {
+ addrs: HashSet<SocketAddr>,
+ },
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -162,6 +165,12 @@ impl QuicTransport {
  .clone()
  }
 
+ pub fn set_manual_peer_addrs(&self, addrs: HashSet<SocketAddr>) {
+ self.internal_tx
+ .send(InternalEvent::RegisterPeerAddr { addrs })
+ .ok();
+ }
+
  // `None` on the port means disabled. Use `0` for random port.
  pub async fn set_ipv4_enabled(&self, port: Option<u16>) -> Result<(), String> {
  self.setup_listener(
@@ -238,6 +247,9 @@ async fn start(
  let mut incoming = control.accept(PROTOCOL).unwrap();
  let map = Arc::new(RwLock::new(HashMap::new()));
  let mut relay_config = Vec::new();
+ let mut manual_addrs = HashSet::new();
+ let mut interval = tokio::time::interval(Duration::from_secs(15));
+ // let manual_hook_id = p2p.register_hook("manual", flume::unbounded().0);
 
  loop {
  tokio::select! {
@@ -427,6 +439,9 @@ async fn start(
  // TODO: Proper error handling
  result.send(Ok(())).ok();
  },
+ InternalEvent::RegisterPeerAddr { addrs } => {
+ manual_addrs = addrs;
+ }
  },
  Some(req) = connect_rx.recv() => {
  let mut control = control.clone();
@@ -460,6 +475,17 @@ async fn start(
  }
  });
  }
+ _ = interval.tick() => {
+ let addrs = manual_addrs.clone();
+
+ for addr in addrs {
+ let err = swarm.dial(socketaddr_to_quic_multiaddr(&addr));
+ debug!("Attempting connection to {:?} {}", addr, match err {
+ Ok(_) => "successfully".into(),
+ Err(err) => format!("with error: {err}"),
+ });
+ }
+ }
  }
  }
 }

interface/ErrorFallback.tsx

@@ -24,6 +24,8 @@ export function RouterErrorBoundary() {
  localStorage.setItem(RENDERING_ERROR_LOCAL_STORAGE_KEY, 'true');
  };
 
+ useEffect(() => console.error(error), [error]);
+
  return (
  <ErrorPage
  message={(error as any).toString()}