fix: community browse discovers members via DHT

- Add CommunityMembers wire type stored under community_info_key(share_id)
- Validate CommunityMembers entries in validate_dht_value_for_known_keyspaces
- Merge member lists on dht_store receipt (relay accumulates all members)
- Announce self as community member in DHT when joining/creating a community
- Re-announce community memberships after relay tunnel registration
- browse_community now does DHT find_value to discover member peer addresses
- browse_community includes local node's own published community shares
- Add 4 tests: store/find, wrong-key rejection, merge-on-store, upsert roundtrip

Previously browse_community only queried bootstrap peers (relay) which
typically has NOT joined the community, returning 0 participants / 0 shares.
Now community members announce themselves via the DHT so other members
can discover and query them.

Author: techartdev

crates/scp2p-cli/src/main.rs

@@ -55,6 +55,8 @@ async fn main() -> anyhow::Result<()> {
         EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new(&args.log_level));
     fmt().with_env_filter(filter).with_target(false).init();
 
-    let quic_port = args.quic_port.unwrap_or_else(|| args.port.saturating_sub(1));
+    let quic_port = args
+        .quic_port
+        .unwrap_or_else(|| args.port.saturating_sub(1));
     shell::run(args.db, args.bootstrap, args.port, quic_port).await
 }

crates/scp2p-cli/src/shell.rs

@@ -14,10 +14,10 @@ use inquire::{InquireError, Select, Text};
 use rand::rngs::OsRng;
 use scp2p_core::{
     BoxedStream, Capabilities, FetchPolicy, Node, NodeConfig, NodeId, OwnedRelayAwareTransport,
-    PeerAddr, PeerConnector, PeerRecord, PersistedCommunity, RelayAwareTransport,
-    RequestTransport, SearchQuery, ShareId, ShareVisibility, SqliteStore, Store,
-    TransportProtocol, build_tls_server_handle, quic_connect_bi_session_insecure,
-    start_quic_server, tls_connect_session_insecure,
+    PeerAddr, PeerConnector, PeerRecord, PersistedCommunity, RelayAwareTransport, RequestTransport,
+    SearchQuery, ShareId, ShareVisibility, SqliteStore, Store, TransportProtocol,
+    build_tls_server_handle, quic_connect_bi_session_insecure, start_quic_server,
+    tls_connect_session_insecure,
 };
 use tracing::{info, warn};
 
@@ -107,7 +107,12 @@ impl Ctx {
 
 // ── Entry point ───────────────────────────────────────────────────────────────
 
-pub async fn run(db: String, bootstrap_raw: Vec<String>, port: u16, quic_port: u16) -> anyhow::Result<()> {
+pub async fn run(
+    db: String,
+    bootstrap_raw: Vec<String>,
+    port: u16,
+    quic_port: u16,
+) -> anyhow::Result<()> {
     let pb = spinner("Opening database…");
     let store = SqliteStore::open(&db)?;
     let config = NodeConfig {
@@ -205,16 +210,11 @@ pub async fn run(db: String, bootstrap_raw: Vec<String>, port: u16, quic_port: u
                                         relay_via: None,
                                     })
                                     .await;
-                                let _ = tunnel_handle
-                                    .reannounce_content_providers(self_addr)
-                                    .await;
+                                let _ = tunnel_handle.reannounce_content_providers(self_addr).await;
                             }
                             // Push to relay immediately.
                             let _ = tunnel_handle
-                                .dht_republish_once(
-                                    tunnel_transport.as_ref(),
-                                    &tunnel_bootstrap,
-                                )
+                                .dht_republish_once(tunnel_transport.as_ref(), &tunnel_bootstrap)
                                 .await;
                             break;
                         }
@@ -317,7 +317,11 @@ async fn cmd_status(ctx: &Ctx) -> anyhow::Result<()> {
     println!("  TCP port : {}", ctx.port);
     println!(
         "  QUIC port: {}",
-        if ctx.quic_port > 0 { ctx.quic_port.to_string() } else { "disabled".to_owned() }
+        if ctx.quic_port > 0 {
+            ctx.quic_port.to_string()
+        } else {
+            "disabled".to_owned()
+        }
     );
     println!("  Subscriptions  : {}", s.subscriptions.len());
     println!("  Manifests      : {}", s.manifests.len());
@@ -1125,7 +1129,11 @@ fn print_banner(ctx: &Ctx) {
     println!("  TCP port : {}", ctx.port);
     println!(
         "  QUIC port: {}",
-        if ctx.quic_port > 0 { ctx.quic_port.to_string() } else { "disabled".to_owned() }
+        if ctx.quic_port > 0 {
+            ctx.quic_port.to_string()
+        } else {
+            "disabled".to_owned()
+        }
     );
     println!(
         "  Network  : {}",

crates/scp2p-core/examples/quic_probe.rs

@@ -2,10 +2,10 @@
 //! Usage: cargo run --example quic_probe -- 178.104.13.182:7000
 use std::net::SocketAddr;
 
-use scp2p_core::transport_net::{quic_connect_bi_session_insecure, tls_connect_session_insecure};
-use scp2p_core::capabilities::Capabilities;
 use ed25519_dalek::SigningKey;
 use rand::rngs::OsRng;
+use scp2p_core::capabilities::Capabilities;
+use scp2p_core::transport_net::{quic_connect_bi_session_insecure, tls_connect_session_insecure};
 
 #[tokio::main]
 async fn main() {
@@ -21,7 +21,10 @@ async fn main() {
     match quic_connect_bi_session_insecure(addr, &key, Capabilities::default(), None).await {
         Ok(session) => {
             println!("QUIC SUCCESS — connected");
-            println!("  remote pubkey: {}", hex::encode(session.session.remote_node_pubkey));
+            println!(
+                "  remote pubkey: {}",
+                hex::encode(session.session.remote_node_pubkey)
+            );
         }
         Err(e) => {
             println!("QUIC FAILED — {e:#}");
@@ -35,7 +38,10 @@ async fn main() {
     match tls_connect_session_insecure(tcp_addr, &key2, Capabilities::default(), None).await {
         Ok((_stream, session)) => {
             println!("TCP SUCCESS — connected");
-            println!("  remote pubkey: {}", hex::encode(session.remote_node_pubkey));
+            println!(
+                "  remote pubkey: {}",
+                hex::encode(session.remote_node_pubkey)
+            );
         }
         Err(e) => {
             println!("TCP FAILED — {e:#}");

crates/scp2p-core/src/api/helpers.rs

@@ -14,7 +14,7 @@ use std::{
 };
 
 use crate::{
-    dht_keys::{content_provider_key, manifest_loc_key, share_head_key},
+    dht_keys::{community_info_key, content_provider_key, manifest_loc_key, share_head_key},
     ids::{NodeId, ShareId},
     manifest::{ManifestV1, PublicShareSummary, ShareHead},
     net_fetch::RequestTransport,
@@ -25,10 +25,11 @@ use crate::{
     },
     search::IndexedItem,
     wire::{
-        CommunityPublicShareList, CommunityStatus, Envelope, FLAG_RESPONSE, FindNode,
-        FindNodeResult, FindValue, FindValueResult, GetCommunityStatus, ListCommunityPublicShares,
-        ListPublicShares, MsgType, Providers, PublicShareList, RelayListRequest, RelayListResponse,
-        RelayPayloadKind as WireRelayPayloadKind, Store as WireStore, WirePayload,
+        CommunityMembers, CommunityPublicShareList, CommunityStatus, Envelope, FLAG_RESPONSE,
+        FindNode, FindNodeResult, FindValue, FindValueResult, GetCommunityStatus,
+        ListCommunityPublicShares, ListPublicShares, MsgType, Providers, PublicShareList,
+        RelayListRequest, RelayListResponse, RelayPayloadKind as WireRelayPayloadKind,
+        Store as WireStore, WirePayload,
     },
 };
 
@@ -181,6 +182,14 @@ pub(super) fn validate_dht_value_for_known_keyspaces(
         }
         anyhow::bail!("relay announcement key does not match any valid rendezvous slot");
     }
+    // Community member lists are stored at community_info_key(share_id).
+    if let Ok(cm) = crate::cbor::from_slice::<CommunityMembers>(value) {
+        let expected = community_info_key(&ShareId(cm.community_share_id));
+        if expected != key {
+            anyhow::bail!("community members value does not match community info key");
+        }
+        return Ok(());
+    }
     // Reject values that do not match any recognized keyspace to prevent
     // arbitrary data storage and potential abuse (§4.5).
     anyhow::bail!("DHT value does not match any recognized keyspace")

crates/scp2p-core/src/api/mod.rs

@@ -1455,11 +1455,7 @@ impl NodeHandle {
     /// Update the locally stored name for a community.
     ///
     /// Called when a remote peer reports the community name during browse.
-    pub async fn update_community_name(
-        &self,
-        share_id: ShareId,
-        name: &str,
-    ) -> anyhow::Result<()> {
+    pub async fn update_community_name(&self, share_id: ShareId, name: &str) -> anyhow::Result<()> {
         let mut state = self.state.write().await;
         if let Some(membership) = state.communities.get_mut(&share_id.0)
             && membership.name.is_none()

crates/scp2p-core/src/api/node_dht.rs

@@ -19,12 +19,12 @@ use crate::transport_net::{
 use crate::{
     capabilities::Capabilities,
     dht::{ALPHA, DEFAULT_TTL_SECS, DhtInsertResult, DhtNodeRecord, DhtValue, K, MAX_VALUE_SIZE},
-    dht_keys::share_head_key,
+    dht_keys::{community_info_key, share_head_key},
     ids::{NodeId, ShareId},
     manifest::ShareHead,
     net_fetch::RequestTransport,
     peer::PeerAddr,
-    wire::{FindNode, Store as WireStore},
+    wire::{CommunityMembers, FindNode, Store as WireStore},
 };
 
 use super::{
@@ -101,13 +101,12 @@ impl NodeHandle {
 
     pub async fn dht_store(&self, req: WireStore) -> anyhow::Result<()> {
         validate_dht_value_for_known_keyspaces(req.key, &req.value)?;
+        let now = now_unix_secs()?;
+        let value = merge_community_members_if_applicable(req.key, req.value, self, now).await;
         let mut state = self.state.write().await;
-        state.dht.store(
-            req.key,
-            req.value,
-            req.ttl_secs.max(DEFAULT_TTL_SECS),
-            now_unix_secs()?,
-        )
+        state
+            .dht
+            .store(req.key, value, req.ttl_secs.max(DEFAULT_TTL_SECS), now)
     }
 
     pub async fn dht_find_value(&self, key: [u8; 32]) -> anyhow::Result<Option<DhtValue>> {
@@ -610,4 +609,102 @@ impl NodeHandle {
         merge_peer_list(&mut peers, known);
         peers
     }
+
+    /// Insert or update the local DHT entry for a community the node has
+    /// joined so that other peers can discover this node as a community
+    /// member via `community_info_key(share_id)`.
+    pub async fn upsert_community_member(
+        &self,
+        community_share_id: ShareId,
+        self_addr: PeerAddr,
+    ) -> anyhow::Result<()> {
+        let key = community_info_key(&community_share_id);
+        let now = now_unix_secs()?;
+        let mut state = self.state.write().await;
+        let mut cm: CommunityMembers = state
+            .dht
+            .find_value(key, now)
+            .and_then(|v| crate::cbor::from_slice(&v.value).ok())
+            .unwrap_or(CommunityMembers {
+                community_share_id: community_share_id.0,
+                members: vec![],
+                updated_at: now,
+            });
+        if !cm.members.contains(&self_addr) {
+            cm.members.push(self_addr);
+        }
+        cm.updated_at = now;
+        state
+            .dht
+            .store(key, crate::cbor::to_vec(&cm)?, DEFAULT_TTL_SECS, now)?;
+        Ok(())
+    }
+
+    /// Re-announce DHT community member entries for all joined communities.
+    ///
+    /// Called during `dht_republish_once` to keep community member
+    /// announcements fresh and ensure they survive app restarts.
+    pub async fn reannounce_community_memberships(
+        &self,
+        self_addr: PeerAddr,
+    ) -> anyhow::Result<usize> {
+        let community_ids: Vec<[u8; 32]> = {
+            let state = self.state.read().await;
+            state.communities.keys().copied().collect()
+        };
+        let mut count = 0usize;
+        for cid in community_ids {
+            if let Err(e) = self
+                .upsert_community_member(ShareId(cid), self_addr.clone())
+                .await
+            {
+                debug!(
+                    community = %hex::encode(&cid[..8]),
+                    error = %e,
+                    "reannounce_community_memberships: failed"
+                );
+            } else {
+                count += 1;
+            }
+        }
+        if count > 0 {
+            debug!(count, "reannounce_community_memberships: updated");
+        }
+        Ok(count)
+    }
+}
+
+/// If `value` deserializes as [`CommunityMembers`], merge its member list
+/// with any existing entry already stored in the local DHT at `key`.
+/// Otherwise return `value` unchanged.
+async fn merge_community_members_if_applicable(
+    key: [u8; 32],
+    value: Vec<u8>,
+    node: &NodeHandle,
+    now: u64,
+) -> Vec<u8> {
+    let Ok(incoming) = crate::cbor::from_slice::<CommunityMembers>(&value) else {
+        return value;
+    };
+    let expected = community_info_key(&ShareId(incoming.community_share_id));
+    if expected != key {
+        return value;
+    }
+    let existing = {
+        let mut state = node.state.write().await;
+        state
+            .dht
+            .find_value(key, now)
+            .and_then(|v| crate::cbor::from_slice::<CommunityMembers>(&v.value).ok())
+    };
+    let Some(mut merged) = existing else {
+        return value;
+    };
+    for member in incoming.members {
+        if !merged.members.contains(&member) {
+            merged.members.push(member);
+        }
+    }
+    merged.updated_at = merged.updated_at.max(incoming.updated_at);
+    crate::cbor::to_vec(&merged).unwrap_or(value)
 }

crates/scp2p-core/src/api/node_net.rs

@@ -1023,28 +1023,29 @@ impl NodeHandle {
                     payload,
                 }),
             WirePayload::GetCommunityStatus(msg) => {
-                let (joined, proof, name) = match VerifyingKey::from_bytes(&msg.community_share_pubkey) {
-                    Ok(pubkey) if ShareId::from_pubkey(&pubkey).0 == msg.community_share_id => {
-                        let state = self.state.read().await;
-                        match state.communities.get(&msg.community_share_id) {
-                            Some(membership) => {
-                                let proof = membership
-                                    .token
-                                    .as_ref()
-                                    .and_then(|t| crate::cbor::to_vec(t).ok());
-                                (true, proof, membership.name.clone())
+                let (joined, proof, name) =
+                    match VerifyingKey::from_bytes(&msg.community_share_pubkey) {
+                        Ok(pubkey) if ShareId::from_pubkey(&pubkey).0 == msg.community_share_id => {
+                            let state = self.state.read().await;
+                            match state.communities.get(&msg.community_share_id) {
+                                Some(membership) => {
+                                    let proof = membership
+                                        .token
+                                        .as_ref()
+                                        .and_then(|t| crate::cbor::to_vec(t).ok());
+                                    (true, proof, membership.name.clone())
+                                }
+                                None => (false, None, None),
                             }
-                            None => (false, None, None),
                         }
-                    }
-                    _ => {
-                        return Some(error_envelope(
-                            req_type,
-                            req_id,
-                            "community share_id does not match share_pubkey",
-                        ));
-                    }
-                };
+                        _ => {
+                            return Some(error_envelope(
+                                req_type,
+                                req_id,
+                                "community share_id does not match share_pubkey",
+                            ));
+                        }
+                    };
                 crate::cbor::to_vec(&CommunityStatus {
                     community_share_id: msg.community_share_id,
                     joined,
@@ -1523,7 +1524,10 @@ impl NodeHandle {
         }
 
         if refreshed > 0 {
-            debug!(refreshed, "reannounce_published_share_data: DHT entries restored");
+            debug!(
+                refreshed,
+                "reannounce_published_share_data: DHT entries restored"
+            );
         }
 
         Ok(refreshed)

crates/scp2p-core/src/api/node_publish.rs

@@ -240,12 +240,8 @@ impl NodeHandle {
                 chunk_list_hash: desc.chunk_list_hash,
             });
             // Register using the already-computed descriptor — no re-read.
-            self.register_content_precomputed(
-                provider.clone(),
-                desc,
-                file_path.to_path_buf(),
-            )
-            .await?;
+            self.register_content_precomputed(provider.clone(), desc, file_path.to_path_buf())
+                .await?;
 
             if total > 50 && (idx + 1) % 500 == 0 {
                 info!(
@@ -349,10 +345,7 @@ impl NodeHandle {
     ///
     /// Called after relay tunnel registration so that provider entries
     /// contain the relayed address, enabling NAT-traversed downloads.
-    pub async fn reannounce_content_providers(
-        &self,
-        self_addr: PeerAddr,
-    ) -> anyhow::Result<usize> {
+    pub async fn reannounce_content_providers(&self, self_addr: PeerAddr) -> anyhow::Result<usize> {
         let now = now_unix_secs()?;
         let mut state = self.state.write().await;