Sologretto/itsgoin
1
0
Fork 0
Code Issues 1 Pull requests Projects Releases Packages Wiki Activity Actions

refactor: v0.8 Iteration B — cruft purge + ALPN bump to itsgoin/4

Browse source 
Zero-users ruling removes all wire-compat obligations. Net -1,443 LOC of Rust.

Protocol break:
- ALPN_V2 b"itsgoin/3" -> ALPN b"itsgoin/4" (renamed: the versioned name was
  what rotted). Old nodes are now refused at the QUIC handshake instead of
  half-interoperating with the v0.8 manifest/digest formats. Deploy gate from
  Iteration A is discharged.

Dead wire surface removed (MessageType 46 -> 40):
- DeleteRecord 0x51 (deletes ride ControlOp::DeletePost + InitialExchange)
- VisibilityUpdate 0x52 + its two dead senders (ControlOp::UpdateVisibility)
- SocialDisconnectNotice 0x71 (zero senders; checkin timeouts carry the signal)
- MeshPrefer 0xB3 + request_prefer + handler (preferred peers are gone)
- Legacy dual pull-matching half (have_post_ids) — behavior-identical
- PullSyncResponse.visibility_updates + counters (control posts cover it)
- serde(default) stripped from wire-only payloads (kept where
  skip_serializing_if makes it load-bearing; persisted-row defaults untouched)

Preferred-peer subsystem eliminated (design ruling: CDN file_holders replaced
the N+10 direct push/pull it served): slot tier, preferred_peers table (dropped),
preferred_tree semantics, rebalance Priority 0, find_relays_for preferred tiers,
7-day prune + 30-day watcher, dead FromStr impl. Slots are now Local/Wide only
(91 desktop / 12 mobile) pending Iteration C's single ~20-slot pool.

Other dead code: hostlist encoder + base64url helper, always-empty
downstream_addrs parameter chain, start_upnp_renewal_cycle no-op + 4 callers,
RELAY_TARGET_RATE_LIMIT, GetSecretSeed, stale comments swept.

Preserved deliberately: EDM scanner corpse (awaiting raw-UDP refactor),
PortScanHeartbeat, Iteration A startup migrations, session-relay opt-in gating.

Docs: design.html + tech.html synced (message counts, ALPN, purge status,
Rework asides flipped past-tense).

190 core tests pass; CLI + desktop build; A3 integration 9/9 (independently
re-run post-purge).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LGiPD2cF75mnvneSCjdDC5
This commit is contained in:
Scott Reimers 2026-07-30 09:26:16 -04:00
parent 17dbf076cb
commit 36e3871c4b
15 changed files with 152 additions and 1476 deletions
Download patch file Download diff file Expand all files Collapse all files

496
crates/core/src/connection.rs
View file

@ -15,13 +15,13 @@ use crate::protocol::{
AnchorReferralRequestPayload, AnchorReferralResponsePayload, AnchorRegisterPayload,
BlobHeaderDiffPayload,
BlobHeaderRequestPayload, BlobHeaderResponsePayload, BlobRequestPayload, BlobResponsePayload,
CircleProfileUpdatePayload, InitialExchangePayload, MeshPreferPayload,
CircleProfileUpdatePayload, InitialExchangePayload,
MessageType, NodeListUpdatePayload, PostDownstreamRegisterPayload,
ProfileUpdatePayload, PullSyncRequestPayload, PullSyncResponsePayload,
RefuseRedirectPayload, RelayIntroducePayload, RelayIntroduceResultPayload, SessionRelayPayload,
SocialAddressUpdatePayload, SocialCheckinPayload, SocialDisconnectNoticePayload,
SyncPost, VisibilityUpdatePayload, WormQueryPayload, WormResponsePayload,
ReplicationRequestPayload, ReplicationResponsePayload, ALPN_V2,
SocialAddressUpdatePayload, SocialCheckinPayload,
SyncPost, WormQueryPayload, WormResponsePayload,
ReplicationRequestPayload, ReplicationResponsePayload, ALPN,
};
use crate::storage::StoragePool;
use crate::types::{
@ -37,24 +37,15 @@ const WORM_TOTAL_TIMEOUT_MS: u64 = 3000;
const WORM_COOLDOWN_MS: i64 = 300_000; // 5 min
const WORM_DEDUP_EXPIRY_MS: u64 = 10_000; // 10 sec
const SESSION_IDLE_TIMEOUT_MS: u64 = 300_000; // 5 min
#[allow(dead_code)]
const RELAY_COOLDOWN_MS: i64 = 300_000; // 5 min
const RELAY_INTRO_DEDUP_EXPIRY_MS: u64 = 30_000; // 30 sec
#[allow(dead_code)]
const RELAY_INTRO_TIMEOUT_MS: u64 = 15_000; // 15 sec
const HOLE_PUNCH_TIMEOUT_MS: u64 = 30_000; // 30 sec overall window
const HOLE_PUNCH_ATTEMPT_MS: u64 = 2_000; // 2 sec per attempt before retry
/// Max bytes relayed per pipe before closing
const RELAY_MAX_BYTES: u64 = 50 * 1024 * 1024; // 50 MB
/// Relay pipe idle timeout
const RELAY_PIPE_IDLE_MS: u64 = 120_000; // 2 min
/// How long a preferred peer can be unreachable before being pruned (7 days)
const PREFERRED_UNREACHABLE_PRUNE_MS: u64 = 7 * 24 * 60 * 60 * 1000;
/// How long reconnect watchers live before expiry (30 days)
const WATCHER_EXPIRY_MS: i64 = 30 * 24 * 60 * 60 * 1000;
/// Max pending introductions per target in 5 minutes
#[allow(dead_code)]
const RELAY_TARGET_RATE_LIMIT: usize = 5;
/// Grace period before removing disconnected peers from referral list
const REFERRAL_DISCONNECT_GRACE_MS: u64 = 120_000; // 2 min
@ -84,7 +75,7 @@ pub(crate) async fn hole_punch_parallel(
target: &NodeId,
addresses: &[String],
) -> Option<iroh::endpoint::Connection> {
use crate::protocol::ALPN_V2;
use crate::protocol::ALPN;
// Filter to address families this endpoint can actually reach
let reachable = filter_reachable_families(endpoint, addresses);
@ -116,7 +107,7 @@ pub(crate) async fn hole_punch_parallel(
handles.push(tokio::spawn(async move {
tokio::time::timeout(
std::time::Duration::from_millis(HOLE_PUNCH_ATTEMPT_MS),
ep.connect(a, ALPN_V2),
ep.connect(a, ALPN),
).await
}));
}
@ -348,7 +339,7 @@ async fn edm_port_scan_disabled_v0_7_3(
join_set.spawn(async move {
if let Ok(Ok(conn)) = tokio::time::timeout(
std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
ep.connect(addr, ALPN_V2),
ep.connect(addr, ALPN),
).await {
let _ = tx.send(conn).await;
}
@ -378,7 +369,7 @@ async fn edm_port_scan_disabled_v0_7_3(
join_set.spawn(async move {
if let Ok(Ok(conn)) = tokio::time::timeout(
std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
ep.connect(addr, ALPN_V2),
ep.connect(addr, ALPN),
).await {
let _ = tx.send(conn).await;
}
@ -419,7 +410,7 @@ async fn edm_port_scan_disabled_v0_7_3(
join_set.spawn(async move {
if let Ok(Ok(conn)) = tokio::time::timeout(
std::time::Duration::from_millis(SCAN_CONNECT_TIMEOUT_MS),
ep.connect(addr, ALPN_V2),
ep.connect(addr, ALPN),
).await {
let _ = tx.send(conn).await;
}
@ -525,7 +516,7 @@ async fn hole_punch_single(
match tokio::time::timeout(
std::time::Duration::from_millis(HOLE_PUNCH_ATTEMPT_MS),
endpoint.connect(endpoint_addr, ALPN_V2),
endpoint.connect(endpoint_addr, ALPN),
).await {
Ok(Ok(conn)) => {
tracing::info!(peer = hex::encode(target), "Quick single punch succeeded");
@ -597,7 +588,6 @@ pub struct SessionConnection {
pub struct PullSyncStats {
pub posts_received: usize,
pub visibility_updates: usize,
}
/// Entry in the anchor's referral list — connection-backed, self-pruning.
@ -666,8 +656,6 @@ pub struct ConnectionManager {
#[allow(dead_code)]
is_anchor: Arc<AtomicBool>,
diff_seq: AtomicU64,
#[allow(dead_code)]
secret_seed: [u8; 32],
blob_store: Arc<BlobStore>,
/// Dedup map for worm queries: worm_id → timestamp_ms
seen_worms: HashMap<WormId, u64>,
@ -675,8 +663,6 @@ pub struct ConnectionManager {
last_n1_set: HashSet<NodeId>,
/// Last broadcast N2 set (for computing diffs)
last_n2_set: HashSet<NodeId>,
/// Max preferred (bilateral) mesh slots
preferred_slots: usize,
/// Max local (diverse) mesh slots
local_slots: usize,
/// Max wide (bloom-sourced) mesh slots
@ -769,7 +755,6 @@ impl ConnectionManager {
storage: Arc<StoragePool>,
our_node_id: NodeId,
is_anchor: Arc<AtomicBool>,
secret_seed: [u8; 32],
blob_store: Arc<BlobStore>,
profile: DeviceProfile,
activity_log: Arc<std::sync::Mutex<ActivityLog>>,
@ -791,12 +776,10 @@ impl ConnectionManager {
our_node_id,
is_anchor,
diff_seq: AtomicU64::new(0),
secret_seed,
blob_store,
seen_worms: HashMap::new(),
last_n1_set: HashSet::new(),
last_n2_set: HashSet::new(),
preferred_slots: profile.preferred_slots(),
local_slots: profile.local_slots(),
wide_slots: profile.wide_slots(),
sessions: HashMap::new(),
@ -1188,7 +1171,7 @@ impl ConnectionManager {
let addr = iroh::EndpointAddr::from(eid).with_ip_addr(sock_addr);
match tokio::time::timeout(
std::time::Duration::from_secs(15),
self.endpoint.connect(addr, ALPN_V2),
self.endpoint.connect(addr, ALPN),
).await {
Ok(Ok(_conn)) => true,
_ => false,
@ -1369,7 +1352,7 @@ impl ConnectionManager {
debug!(peer = hex::encode(remote_node_id), "Replacing existing connection");
}
let total_slots = self.preferred_slots + self.local_slots + self.wide_slots;
let total_slots = self.local_slots + self.wide_slots;
let total = self.connections.len();
if total >= total_slots && !self.connections.contains_key(&remote_node_id) {
debug!(peer = hex::encode(remote_node_id), "Slots full, rejecting");
@ -1486,7 +1469,7 @@ impl ConnectionManager {
) -> anyhow::Result<iroh::endpoint::Connection> {
let conn = tokio::time::timeout(
std::time::Duration::from_secs(15),
endpoint.connect(addr, ALPN_V2),
endpoint.connect(addr, ALPN),
).await
.map_err(|_| anyhow::anyhow!("connect timed out (15s)"))?
.map_err(|e| anyhow::anyhow!("connect failed: {e}"))?;
@ -1522,7 +1505,6 @@ impl ConnectionManager {
let request = PullSyncRequestPayload {
follows: query_list,
have_post_ids: vec![],
since_ms: follows_sync,
};
@ -1537,7 +1519,6 @@ impl ConnectionManager {
let response: PullSyncResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
let mut posts_received = 0;
let mut vis_updates = 0;
let mut new_post_ids: Vec<PostId> = Vec::new();
let now_ms = crate::connection::now_ms();
let mut synced_authors: HashSet<NodeId> = HashSet::new();
@ -1565,7 +1546,7 @@ impl ConnectionManager {
}
}
// Brief storage lock: upstream + last_sync + visibility updates
// Brief storage lock: upstream + last_sync
{
let s = storage.get().await;
for pid in &new_post_ids {
@ -1579,15 +1560,6 @@ impl ConnectionManager {
for author in &synced_authors {
let _ = s.update_follow_last_sync(author, now_ms);
}
for vu in response.visibility_updates {
if let Some(post) = s.get_post(&vu.post_id)? {
if post.author == vu.author {
if s.update_post_visibility(&vu.post_id, &vu.visibility)? {
vis_updates += 1;
}
}
}
}
}
// Register as downstream (spawned, no lock needed)
@ -1604,7 +1576,7 @@ impl ConnectionManager {
});
}
Ok(PullSyncStats { posts_received, visibility_updates: vis_updates })
Ok(PullSyncStats { posts_received })
}
/// Fetch engagement headers from a peer — standalone version that doesn't require conn_mgr lock.
@ -2021,36 +1993,6 @@ impl ConnectionManager {
Ok(count)
}
/// Broadcast a visibility update to all connected peers.
pub async fn broadcast_visibility_update(
&self,
update: &crate::types::VisibilityUpdate,
) -> usize {
let payload = VisibilityUpdatePayload {
updates: vec![update.clone()],
};
let mut sent = 0;
for (peer_id, pc) in &self.connections {
let result = async {
let mut send = pc.connection.open_uni().await?;
write_typed_message(&mut send, MessageType::VisibilityUpdate, &payload).await?;
send.finish()?;
anyhow::Ok(())
}
.await;
match result {
Ok(()) => sent += 1,
Err(e) => {
debug!(peer = hex::encode(peer_id), error = %e, "Failed to push visibility update");
}
}
}
sent
}
/// Pull posts from a connected peer.
pub async fn pull_from_peer(&self, peer_id: &NodeId) -> anyhow::Result<PullSyncStats> {
let pc = self
@ -2077,7 +2019,6 @@ impl ConnectionManager {
let request = PullSyncRequestPayload {
follows: query_list,
have_post_ids: vec![], // v4: empty, using since_ms instead
since_ms: follows_sync,
};
@ -2092,7 +2033,6 @@ impl ConnectionManager {
let response: PullSyncResponsePayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
let mut posts_received = 0;
let mut vis_updates = 0;
let mut new_post_ids: Vec<PostId> = Vec::new();
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@ -2126,7 +2066,7 @@ impl ConnectionManager {
}
// Lock RELEASED
// Brief lock 2: upstream + last_sync + visibility updates
// Brief lock 2: upstream + last_sync
{
let storage = self.storage.get().await;
for pid in &new_post_ids {
@ -2140,19 +2080,6 @@ impl ConnectionManager {
for author in &synced_authors {
let _ = storage.update_follow_last_sync(author, now_ms);
}
for vu in response.visibility_updates {
if vu.author != *peer_id {
// Only accept visibility updates authored by the responding peer
// (or their forwarded data — but for now, only from the peer)
}
if let Some(post) = storage.get_post(&vu.post_id)? {
if post.author == vu.author {
if storage.update_post_visibility(&vu.post_id, &vu.visibility)? {
vis_updates += 1;
}
}
}
}
}
// Register as downstream with the sender for new posts (cap at 50 to avoid flooding)
@ -2171,7 +2098,6 @@ impl ConnectionManager {
Ok(PullSyncStats {
posts_received,
visibility_updates: vis_updates,
})
}
@ -2296,26 +2222,20 @@ impl ConnectionManager {
let request: PullSyncRequestPayload = read_payload(&mut recv, MAX_PAYLOAD).await?;
let their_follows: HashSet<NodeId> = request.follows.into_iter().collect();
let their_post_ids: HashSet<[u8; 32]> = request.have_post_ids.into_iter().collect();
// Protocol v4: build per-author since_ms lookup
// Per-author since_ms lookup
let since_ms_map: HashMap<NodeId, u64> = request.since_ms.into_iter().collect();
let use_since_ms = !since_ms_map.is_empty();
// Phase 1: Brief lock — load data
let (all_posts, group_members, own_posting_ids) = {
let (all_posts, group_members) = {
let s = storage.get().await;
let posts = s.list_posts_with_visibility()?;
let members = s.get_all_group_members().unwrap_or_default();
let own_ids: Vec<NodeId> = s.list_posting_identities()
.unwrap_or_default()
.into_iter().map(|p| p.node_id).collect();
(posts, members, own_ids)
(posts, members)
};
// Phase 2: Filter without lock (pure CPU)
let mut candidates_to_send = Vec::new();
let mut vis_updates_to_send = Vec::new();
for (id, post, visibility) in all_posts {
let should_send =
@ -2325,35 +2245,19 @@ impl ConnectionManager {
continue;
}
let peer_has_post = if use_since_ms {
if let Some(&since) = since_ms_map.get(&post.author) {
post.timestamp_ms <= since + 60_000
} else {
false
}
let peer_has_post = if let Some(&since) = since_ms_map.get(&post.author) {
post.timestamp_ms <= since + 60_000
} else {
their_post_ids.contains(&id)
false
};
if !peer_has_post {
candidates_to_send.push((id, post, visibility));
} else {
// "Own post" = authored by ANY of our posting identities;
// our_node_id is the NETWORK id and never authors posts.
// (Redundant with control-post propagation — deletion
// candidate in the v0.8 cruft purge.)
if own_posting_ids.contains(&post.author) {
vis_updates_to_send.push(crate::types::VisibilityUpdate {
post_id: id,
author: post.author,
visibility,
});
}
}
}
// Phase 3: Brief re-lock for is_deleted checks + intent fetch on filtered posts
let (posts, vis_updates) = {
let posts = {
let s = storage.get().await;
let posts_to_send: Vec<SyncPost> = candidates_to_send.into_iter()
.filter(|(id, _, _)| !s.is_deleted(id).unwrap_or(false))
@ -2362,12 +2266,11 @@ impl ConnectionManager {
SyncPost { id, post, visibility, intent }
})
.collect();
(posts_to_send, vis_updates_to_send)
posts_to_send
};
let response = PullSyncResponsePayload {
posts,
visibility_updates: vis_updates,
};
write_typed_message(&mut send, MessageType::PullSyncResponse, &response).await?;
@ -2815,7 +2718,7 @@ impl ConnectionManager {
addr = addr.with_ip_addr(sock);
}
let conn = endpoint.connect(addr, ALPN_V2).await.ok()?;
let conn = endpoint.connect(addr, ALPN).await.ok()?;
let resp = Self::send_worm_query_raw(&conn, &payload).await.ok()?;
let is_hit = resp.found || resp.post_holder.is_some() || resp.blob_holder.is_some();
@ -3160,7 +3063,7 @@ impl ConnectionManager {
}
// Signal growth loop to fill the empty slot (don't wait 10min for rebalance)
let total_slots = self.preferred_slots + self.local_slots + self.wide_slots;
let total_slots = self.local_slots + self.wide_slots;
if remaining < total_slots {
self.notify_growth();
}
@ -3280,72 +3183,6 @@ impl ConnectionManager {
let newly_connected: Vec<NodeId> = Vec::new();
let mut pending_connects: Vec<(NodeId, iroh::EndpointAddr, String, PeerSlotKind)> = Vec::new();
// Priority 0 (NEW): Reconnect preferred peers
{
let preferred_peers = {
let storage = self.storage.get().await;
storage.list_preferred_peers().unwrap_or_default()
};
let now = now_ms();
for peer_id in &preferred_peers {
if self.connections.contains_key(peer_id) || *peer_id == self.our_node_id {
continue;
}
// Prune preferred peers unreachable for 7+ days
if let Some(&failed_at) = self.unreachable_peers.get(peer_id) {
if now.saturating_sub(failed_at) > PREFERRED_UNREACHABLE_PRUNE_MS {
info!(peer = hex::encode(peer_id), "Removing preferred peer unreachable for 7 days+");
let storage = self.storage.get().await;
let _ = storage.remove_preferred_peer(peer_id);
continue;
}
}
// Evict lowest-diversity non-preferred peer if at capacity
let total_slots = self.preferred_slots + self.local_slots + self.wide_slots;
if self.connections.len() >= total_slots {
let evict_candidate = self.find_non_preferred_eviction_candidate().await;
if let Some(evict_id) = evict_candidate {
info!(
evicting = hex::encode(evict_id),
for_preferred = hex::encode(peer_id),
"Evicting non-preferred peer for preferred reconnection"
);
self.log_activity(ActivityLevel::Info, ActivityCategory::Rebalance, format!("Evicting {} for preferred {}", &hex::encode(evict_id)[..8], &hex::encode(peer_id)[..8]), Some(evict_id));
self.disconnect_peer(&evict_id).await;
} else {
debug!(peer = hex::encode(peer_id), "No non-preferred peer to evict");
continue;
}
}
// Collect for connection outside the lock
let addr_str = if !self.is_likely_unreachable(peer_id) {
let storage = self.storage.get().await;
let addr = storage.get_peer_record(peer_id).ok().flatten()
.and_then(|r| r.addresses.first().map(|a| a.to_string()))
.or_else(|| {
storage.get_social_route(peer_id).ok().flatten()
.and_then(|r| r.addresses.first().map(|a| a.to_string()))
});
drop(storage);
addr
} else {
None
};
if let Some(addr_s) = addr_str {
if let Ok(eid) = iroh::EndpointId::from_bytes(peer_id) {
let mut addr = iroh::EndpointAddr::from(eid);
if let Ok(sock) = addr_s.parse::<std::net::SocketAddr>() {
addr = addr.with_ip_addr(sock);
}
pending_connects.push((*peer_id, addr, addr_s, PeerSlotKind::Preferred));
}
}
}
}
// Priority 1+2: Fill empty local slots with diverse candidates
let local_count = self.count_kind(PeerSlotKind::Local);
if local_count < self.local_slots {
@ -3353,15 +3190,11 @@ impl ConnectionManager {
let storage = self.storage.get().await;
let mut cands = Vec::new();
// Priority 1: reconnect recently-dead non-preferred peers
// Priority 1: reconnect recently-dead peers
for peer_id in &dead {
if *peer_id == self.our_node_id {
continue;
}
// Skip preferred peers — handled above
if storage.is_preferred_peer(peer_id).unwrap_or(false) {
continue;
}
if let Ok(Some(rec)) = storage.get_peer_record(peer_id) {
let addr = rec.addresses.first().map(|a| a.to_string());
cands.push((*peer_id, addr));
@ -3431,24 +3264,6 @@ impl ConnectionManager {
Ok((newly_connected, pending_connects))
}
/// Find the lowest-diversity non-preferred peer to evict.
async fn find_non_preferred_eviction_candidate(&self) -> Option<NodeId> {
let storage = self.storage.get().await;
let mut worst: Option<(NodeId, usize)> = None;
for (peer_id, mc) in &self.connections {
if mc.slot_kind == PeerSlotKind::Preferred {
continue; // Never evict preferred
}
let unique = storage.count_unique_n2_for_reporter(peer_id, &[]).unwrap_or(0);
match &worst {
None => worst = Some((*peer_id, unique)),
Some((_, worst_unique)) if unique < *worst_unique => worst = Some((*peer_id, unique)),
_ => {}
}
}
worst.map(|(nid, _)| nid)
}
pub fn is_connected(&self, peer_id: &NodeId) -> bool {
self.connections.contains_key(peer_id)
}
@ -3478,120 +3293,6 @@ impl ConnectionManager {
self.connections.values().filter(|pc| pc.slot_kind == kind).count()
}
// ---- Preferred peer negotiation ----
/// Request bilateral preferred peer status with a connected mesh peer.
/// On success, both sides persist the agreement and upgrade the slot.
pub async fn request_prefer(&mut self, peer_id: &NodeId) -> anyhow::Result<bool> {
let pc = self.connections.get(peer_id)
.ok_or_else(|| anyhow::anyhow!("peer not connected"))?;
// Check if we have room
let preferred_count = self.count_kind(PeerSlotKind::Preferred);
if preferred_count >= self.preferred_slots {
anyhow::bail!("preferred slots full ({}/{})", preferred_count, self.preferred_slots);
}
let request = MeshPreferPayload {
requesting: true,
accepted: false,
reject_reason: None,
};
let (mut send, mut recv) = pc.connection.open_bi().await?;
write_typed_message(&mut send, MessageType::MeshPrefer, &request).await?;
send.finish()?;
let msg_type = read_message_type(&mut recv).await?;
if msg_type != MessageType::MeshPrefer {
anyhow::bail!("expected MeshPrefer response, got {:?}", msg_type);
}
let response: MeshPreferPayload = read_payload(&mut recv, 4096).await?;
if response.accepted {
// Persist agreement
let storage = self.storage.get().await;
storage.add_preferred_peer(peer_id)?;
storage.add_mesh_peer(peer_id, PeerSlotKind::Preferred, 100)?;
drop(storage);
// Upgrade slot in-memory
if let Some(mc) = self.connections.get_mut(peer_id) {
mc.slot_kind = PeerSlotKind::Preferred;
}
info!(peer = hex::encode(peer_id), "Preferred peer agreement established");
Ok(true)
} else {
debug!(
peer = hex::encode(peer_id),
reason = ?response.reject_reason,
"Preferred peer request rejected"
);
Ok(false)
}
}
/// Handle an incoming MeshPrefer request from a connected peer.
pub async fn handle_mesh_prefer(
&mut self,
from_peer: NodeId,
mut send: iroh::endpoint::SendStream,
mut recv: iroh::endpoint::RecvStream,
) -> anyhow::Result<()> {
let request: MeshPreferPayload = read_payload(&mut recv, 4096).await?;
if !request.requesting {
// Not a request — ignore
return Ok(());
}
// Check if we have room for a preferred peer
let preferred_count = self.count_kind(PeerSlotKind::Preferred);
let can_accept = preferred_count < self.preferred_slots
&& self.connections.contains_key(&from_peer);
let response = if can_accept {
// Persist agreement
let storage = self.storage.get().await;
storage.add_preferred_peer(&from_peer)?;
storage.add_mesh_peer(&from_peer, PeerSlotKind::Preferred, 100)?;
drop(storage);
// Upgrade slot in-memory
if let Some(mc) = self.connections.get_mut(&from_peer) {
mc.slot_kind = PeerSlotKind::Preferred;
}
info!(peer = hex::encode(from_peer), "Accepted preferred peer request");
MeshPreferPayload {
requesting: false,
accepted: true,
reject_reason: None,
}
} else {
let reason = if !self.connections.contains_key(&from_peer) {
"not connected".to_string()
} else {
format!("preferred slots full ({}/{})", preferred_count, self.preferred_slots)
};
debug!(peer = hex::encode(from_peer), reason = %reason, "Rejecting preferred peer request");
MeshPreferPayload {
requesting: false,
accepted: false,
reject_reason: Some(reason),
}
};
write_typed_message(&mut send, MessageType::MeshPrefer, &response).await?;
send.finish()?;
Ok(())
}
// reconnect_preferred removed — direct connect now happens outside the lock
// via pending_connects in the actor dispatch. Relay fallback for preferred peers
// is handled by the growth loop's normal relay introduction path.
// ---- Session connection management ----
/// Add a session connection. Evicts oldest idle session if at capacity.
@ -3820,38 +3521,7 @@ impl ConnectionManager {
let mut candidates = Vec::new();
let storage = self.storage.get().await;
// Step 1 (NEW): Check target's preferred_tree from social_routes (~100 NodeIds)
// Intersect with our connections → TTL=0 candidates (they know target or are stably nearby)
if let Ok(Some(route)) = storage.get_social_route(target) {
if !route.preferred_tree.is_empty() {
// Prefer our own preferred peers first within the tree
for tree_node in &route.preferred_tree {
if candidates.len() >= 3 {
break;
}
if let Some(mc) = self.connections.get(tree_node) {
if mc.slot_kind == PeerSlotKind::Preferred
&& !candidates.iter().any(|(nid, _)| nid == tree_node)
{
candidates.push((*tree_node, 0));
}
}
}
// Then any connected tree node
for tree_node in &route.preferred_tree {
if candidates.len() >= 3 {
break;
}
if self.connections.contains_key(tree_node)
&& !candidates.iter().any(|(nid, _)| nid == tree_node)
{
candidates.push((*tree_node, 0));
}
}
}
}
// Step 2: Our mesh peers that have target in their N1 (our N2 entry tagged to them)
// Step 1: Our mesh peers that have target in their N1 (our N2 entry tagged to them)
if candidates.len() < 3 {
if let Ok(reporters) = storage.find_in_n2(target) {
for reporter in reporters {
@ -3865,28 +3535,7 @@ impl ConnectionManager {
}
}
// Step 3: Intersect preferred_tree with N3 → TTL=1 candidates
if candidates.len() < 3 {
if let Ok(Some(route)) = storage.get_social_route(target) {
for tree_node in &route.preferred_tree {
if candidates.len() >= 3 {
break;
}
if let Ok(reporters) = storage.find_in_n3(tree_node) {
for reporter in reporters {
if self.connections.contains_key(&reporter)
&& !candidates.iter().any(|(nid, _)| *nid == reporter)
&& candidates.len() < 3
{
candidates.push((reporter, 1));
}
}
}
}
}
}
// Step 4: Fallback — full N3 scan for target
// Step 2: Fallback — full N3 scan for target
if candidates.len() < 3 {
if let Ok(reporters) = storage.find_in_n3(target) {
for reporter in reporters {
@ -4600,7 +4249,7 @@ impl ConnectionManager {
}
};
// Attempt reconnect for unexpected disconnects (not intentional SocialDisconnectNotice)
// Attempt reconnect for unexpected disconnects
if is_current {
if let Some(addr) = peer_addr {
let cm_arc = Arc::clone(&conn_mgr);
@ -4953,7 +4602,7 @@ impl ConnectionManager {
// Build a temporary endpoint with a random port
let secret_key = iroh::SecretKey::generate(&mut rand::rng());
let temp_ep = iroh::Endpoint::builder()
.alpns(vec![ALPN_V2.to_vec()])
.alpns(vec![ALPN.to_vec()])
.secret_key(secret_key)
.bind()
.await?;
@ -4964,7 +4613,7 @@ impl ConnectionManager {
// Try connecting with a short timeout (2s)
let result = tokio::time::timeout(
std::time::Duration::from_secs(2),
temp_ep.connect(addr, ALPN_V2),
temp_ep.connect(addr, ALPN),
).await;
// Clean up the temporary endpoint
@ -5005,54 +4654,6 @@ impl ConnectionManager {
let _ = storage.store_profile(&profile);
}
}
MessageType::DeleteRecord => {
let payload: crate::protocol::DeleteRecordPayload =
read_payload(recv, MAX_PAYLOAD).await?;
let cm = conn_mgr.lock().await;
// Collect blob CIDs + CDN peers before async work
let mut blob_cleanup: Vec<([u8; 32], Vec<(NodeId, Vec<String>)>)> = Vec::new();
{
let storage = cm.storage.get().await;
for dr in &payload.records {
if crypto::verify_delete_signature(&dr.author, &dr.post_id, &dr.signature) {
// Collect blobs for CDN cleanup before deleting
let blob_cids = storage.get_blobs_for_post(&dr.post_id).unwrap_or_default();
for cid in blob_cids {
let holders = storage.get_file_holders(&cid).unwrap_or_default();
blob_cleanup.push((cid, holders));
}
let _ = storage.store_delete(dr);
let _ = storage.apply_delete(dr);
// Delete blob metadata + CDN metadata
let deleted_cids = storage.delete_blobs_for_post(&dr.post_id).unwrap_or_default();
for cid in &deleted_cids {
let _ = storage.cleanup_cdn_for_blob(cid);
let _ = cm.blob_store.delete(cid);
}
}
}
}
drop(cm);
// BlobDeleteNotice removed in v0.6.2: orphaned blobs on remote
// holders are evicted naturally via LRU rather than by a
// persona-signed push.
let _ = blob_cleanup;
}
MessageType::VisibilityUpdate => {
let payload: crate::protocol::VisibilityUpdatePayload =
read_payload(recv, MAX_PAYLOAD).await?;
let cm = conn_mgr.lock().await;
let storage = cm.storage.get().await;
for vu in payload.updates {
if let Some(post) = storage.get_post(&vu.post_id)? {
if post.author == vu.author {
let _ = storage.update_post_visibility(&vu.post_id, &vu.visibility);
}
}
}
}
MessageType::SocialAddressUpdate => {
let payload: SocialAddressUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
let cm = conn_mgr.lock().await;
@ -5308,19 +4909,6 @@ impl ConnectionManager {
debug!(peer = hex::encode(remote_node_id), stored, relayed = relay_conns.len(), "Received manifest push");
}
MessageType::SocialDisconnectNotice => {
let payload: SocialDisconnectNoticePayload = read_payload(recv, MAX_PAYLOAD).await?;
let cm = conn_mgr.lock().await;
let storage = cm.storage.get().await;
if storage.has_social_route(&payload.node_id).unwrap_or(false) {
let _ = storage.set_social_route_status(&payload.node_id, SocialStatus::Disconnected);
}
debug!(
peer = hex::encode(remote_node_id),
target = hex::encode(payload.node_id),
"Received social disconnect notice"
);
}
MessageType::CircleProfileUpdate => {
let payload: CircleProfileUpdatePayload = read_payload(recv, MAX_PAYLOAD).await?;
let cm = conn_mgr.lock().await;
@ -5942,10 +5530,6 @@ impl ConnectionManager {
let cm = Arc::clone(conn_mgr);
Self::handle_session_relay(cm, recv, send, remote_node_id).await?;
}
MessageType::MeshPrefer => {
let mut cm = conn_mgr.lock().await;
cm.handle_mesh_prefer(remote_node_id, send, recv).await?;
}
MessageType::AnchorReferralRequest => {
let payload: AnchorReferralRequestPayload = read_payload(&mut recv, 4096).await?;
let mut cm = conn_mgr.lock().await;
@ -6502,7 +6086,6 @@ pub enum ConnResponse {
OptString(Option<String>),
OptEndpointAddr(Option<iroh::EndpointAddr>),
Endpoint(iroh::Endpoint),
SecretSeed([u8; 32]),
Storage(Arc<StoragePool>),
BlobStore(Arc<BlobStore>),
ActiveRelayPipes(Arc<AtomicU64>),
@ -6595,9 +6178,6 @@ pub enum ConnCommand {
GetEndpoint {
reply: oneshot::Sender<iroh::Endpoint>,
},
GetSecretSeed {
reply: oneshot::Sender<[u8; 32]>,
},
GetStorage {
reply: oneshot::Sender<Arc<StoragePool>>,
},
@ -6984,12 +6564,6 @@ impl ConnHandle {
rx.await.expect("actor dropped")
}
pub async fn secret_seed(&self) -> [u8; 32] {
let (tx, rx) = oneshot::channel();
let _ = self.tx.send(ConnCommand::GetSecretSeed { reply: tx }).await;
rx.await.unwrap_or([0u8; 32])
}
pub async fn storage(&self) -> Arc<StoragePool> {
let (tx, rx) = oneshot::channel();
let _ = self.tx.send(ConnCommand::GetStorage { reply: tx }).await;
@ -7700,7 +7274,7 @@ impl ConnectionActor {
let endpoint_id = iroh::EndpointId::from_bytes(&ref_id).ok()?;
let mut addr = iroh::EndpointAddr::from(endpoint_id);
if let Ok(sock) = ref_addr.parse::<std::net::SocketAddr>() { addr = addr.with_ip_addr(sock); }
let conn = endpoint.connect(addr, ALPN_V2).await.ok()?;
let conn = endpoint.connect(addr, ALPN).await.ok()?;
let resp = ConnectionManager::send_worm_query_raw(&conn, &payload).await.ok()?;
let is_hit = resp.found || resp.post_holder.is_some() || resp.blob_holder.is_some();
if is_hit {
@ -7930,10 +7504,6 @@ impl ConnectionActor {
let cm = self.cm.lock().await;
let _ = reply.send(cm.endpoint().clone());
}
ConnCommand::GetSecretSeed { reply } => {
let cm = self.cm.lock().await;
let _ = reply.send(cm.secret_seed);
}
ConnCommand::GetStorage { reply } => {
let cm = self.cm.lock().await;
let _ = reply.send(cm.storage_ref());

7
crates/core/src/fof.rs
View file

@ -572,9 +572,10 @@ pub fn decrypt_fof_comment_payload(
// re-propagated via neighbor-manifest diffs.
/// Maximum allowed wrap_slots / pub_post_set entries on an incoming
/// FoF post. The bucket rule caps at `real + rand(0..=128)` above 256;
/// at a 4096-vouchee max realistic graph that's ~4224. Round up for
/// headroom; anything larger is presumed attacker-shaped.
/// FoF post. The bucket rule (`next_vouch_batch_bucket` in profile.rs) is
/// deterministic: ≤8 → 8; ≤256 → next power of two; >256 → next multiple
/// of 128. A 4096-vouchee realistic max buckets to 4096; 8192 gives 2x
/// headroom, anything larger is presumed attacker-shaped.
const MAX_FOF_WRAP_SLOTS: usize = 8192;
/// Maximum allowed revocation_list entries in a t=0 published gating

14
crates/core/src/group_key_distribution.rs
View file

@ -1,15 +1,9 @@
//! Group-key distribution as an encrypted post.
//!
//! v0.6.2 replaces the v0.6.1 `GroupKeyDistribute` wire push (admin →
//! member, uni-stream) with a standard public post that carries the group
//! seed inside `PostVisibility::Encrypted`. Each member is a recipient; the
//! Group-key distribution as an encrypted post: the group seed travels
//! inside `PostVisibility::Encrypted`. Each member is a recipient; the
//! post's CEK is wrapped per member using the admin's posting key. Members
//! receive the post via normal CDN / pull paths, decrypt with their posting
//! secret, and recover the seed + metadata.
//!
//! Removing the direct push eliminates the wire-level signal that a given
//! network endpoint is coordinating group membership with another specific
//! endpoint.
//! secret, and recover the seed + metadata. (No direct wire push — that
//! would signal which endpoints coordinate group membership.)
//!
//! Note: Members are identified by their **posting** NodeIds (the
//! author/recipient namespace since the v0.6.1 identity split), not network

137
crates/core/src/http.rs
View file

@ -8,7 +8,6 @@ use std::sync::Arc;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::TcpStream;
use tokio::sync::Mutex;
use tracing::{debug, info};
use crate::blob::BlobStore;
@ -106,7 +105,6 @@ pub async fn run_http_server(
port: u16,
storage: Arc<StoragePool>,
blob_store: Arc<BlobStore>,
downstream_addrs: Arc<Mutex<HashMap<[u8; 32], Vec<SocketAddr>>>>,
) -> anyhow::Result<()> {
let addr: SocketAddr = ([0, 0, 0, 0], port).into();
// Use SO_REUSEADDR + SO_REUSEPORT so TCP punch sockets can share the port
@ -156,10 +154,9 @@ pub async fn run_http_server(
let storage = Arc::clone(&storage);
let blob_store = Arc::clone(&blob_store);
let budget = Arc::clone(&budget);
let downstream_addrs = Arc::clone(&downstream_addrs);
tokio::spawn(async move {
handle_connection(stream, ip, slot, &storage, &blob_store, &downstream_addrs).await;
handle_connection(stream, ip, slot, &storage, &blob_store).await;
let mut b = budget.lock().unwrap();
match slot {
SlotKind::Content => b.release_content(ip),
@ -182,7 +179,6 @@ async fn handle_connection(
slot: SlotKind,
storage: &Arc<StoragePool>,
blob_store: &Arc<BlobStore>,
downstream_addrs: &Arc<Mutex<HashMap<[u8; 32], Vec<SocketAddr>>>>,
) {
// Keep-alive loop: handle sequential requests on the same connection
loop {
@ -222,7 +218,7 @@ async fn handle_connection(
}
}
SlotKind::Redirect => {
if !try_redirect(&mut stream, &post_id, storage, downstream_addrs).await {
if !try_redirect(&mut stream, &post_id, storage).await {
return;
}
}
@ -368,7 +364,6 @@ async fn try_redirect(
stream: &mut TcpStream,
post_id: &[u8; 32],
storage: &Arc<StoragePool>,
_downstream_addrs: &Arc<Mutex<HashMap<[u8; 32], Vec<SocketAddr>>>>,
) -> bool {
// Get downstream peers for this post
let downstream_peers = {
@ -583,114 +578,6 @@ pub fn html_escape(s: &str) -> String {
out
}
// --- Share link generation ---
/// Encode a list of socket addresses as compact binary, then base64url.
/// Per IPv4: [0x04][4 bytes IP][2 bytes port] = 7 bytes
/// Per IPv6: [0x06][16 bytes IP][2 bytes port] = 19 bytes
pub fn encode_hostlist(hosts: &[SocketAddr]) -> String {
let mut buf = Vec::with_capacity(hosts.len() * 19);
for host in hosts.iter().take(5) {
match host {
SocketAddr::V4(v4) => {
buf.push(0x04);
buf.extend_from_slice(&v4.ip().octets());
buf.extend_from_slice(&v4.port().to_be_bytes());
}
SocketAddr::V6(v6) => {
buf.push(0x06);
buf.extend_from_slice(&v6.ip().octets());
buf.extend_from_slice(&v6.port().to_be_bytes());
}
}
}
base64url_encode(&buf)
}
/// Decode a base64url-encoded hostlist back to socket addresses.
pub fn decode_hostlist(encoded: &str) -> Vec<SocketAddr> {
let buf = match base64url_decode(encoded) {
Some(b) => b,
None => return Vec::new(),
};
let mut addrs = Vec::new();
let mut i = 0;
while i < buf.len() {
match buf[i] {
0x04 if i + 7 <= buf.len() => {
let ip = std::net::Ipv4Addr::new(buf[i + 1], buf[i + 2], buf[i + 3], buf[i + 4]);
let port = u16::from_be_bytes([buf[i + 5], buf[i + 6]]);
addrs.push(SocketAddr::new(ip.into(), port));
i += 7;
}
0x06 if i + 19 <= buf.len() => {
let mut octets = [0u8; 16];
octets.copy_from_slice(&buf[i + 1..i + 17]);
let ip = std::net::Ipv6Addr::from(octets);
let port = u16::from_be_bytes([buf[i + 17], buf[i + 18]]);
addrs.push(SocketAddr::new(ip.into(), port));
i += 19;
}
_ => break, // malformed
}
}
addrs
}
// --- Minimal base64url implementation (no external dependency) ---
const B64_CHARS: &[u8] = b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789-_";
fn base64url_encode(data: &[u8]) -> String {
let mut out = String::with_capacity((data.len() * 4 + 2) / 3);
let mut i = 0;
while i + 2 < data.len() {
let n = ((data[i] as u32) << 16) | ((data[i + 1] as u32) << 8) | data[i + 2] as u32;
out.push(B64_CHARS[((n >> 18) & 0x3F) as usize] as char);
out.push(B64_CHARS[((n >> 12) & 0x3F) as usize] as char);
out.push(B64_CHARS[((n >> 6) & 0x3F) as usize] as char);
out.push(B64_CHARS[(n & 0x3F) as usize] as char);
i += 3;
}
let remaining = data.len() - i;
if remaining == 2 {
let n = ((data[i] as u32) << 16) | ((data[i + 1] as u32) << 8);
out.push(B64_CHARS[((n >> 18) & 0x3F) as usize] as char);
out.push(B64_CHARS[((n >> 12) & 0x3F) as usize] as char);
out.push(B64_CHARS[((n >> 6) & 0x3F) as usize] as char);
} else if remaining == 1 {
let n = (data[i] as u32) << 16;
out.push(B64_CHARS[((n >> 18) & 0x3F) as usize] as char);
out.push(B64_CHARS[((n >> 12) & 0x3F) as usize] as char);
}
out // no padding
}
fn base64url_decode(s: &str) -> Option<Vec<u8>> {
let mut buf = Vec::with_capacity(s.len() * 3 / 4);
let mut accum: u32 = 0;
let mut bits: u32 = 0;
for c in s.bytes() {
let val = match c {
b'A'..=b'Z' => c - b'A',
b'a'..=b'z' => c - b'a' + 26,
b'0'..=b'9' => c - b'0' + 52,
b'-' => 62,
b'_' => 63,
b'=' => continue, // skip padding
_ => return None,
};
accum = (accum << 6) | val as u32;
bits += 6;
if bits >= 8 {
bits -= 8;
buf.push((accum >> bits) as u8);
accum &= (1 << bits) - 1;
}
}
Some(buf)
}
#[cfg(test)]
mod tests {
use super::*;
@ -713,26 +600,6 @@ mod tests {
assert_eq!(html_escape("a&b"), "a&amp;b");
}
#[test]
fn test_base64url_roundtrip() {
let data = b"hello world";
let encoded = base64url_encode(data);
let decoded = base64url_decode(&encoded).unwrap();
assert_eq!(decoded, data);
}
#[test]
fn test_hostlist_roundtrip() {
use std::net::{Ipv4Addr, Ipv6Addr};
let hosts = vec![
SocketAddr::new(Ipv4Addr::new(192, 168, 1, 1).into(), 4433),
SocketAddr::new(Ipv6Addr::LOCALHOST.into(), 8080),
];
let encoded = encode_hostlist(&hosts);
let decoded = decode_hostlist(&encoded);
assert_eq!(decoded, hosts);
}
#[test]
fn test_parse_request_line() {
let req = b"GET /p/abc123 HTTP/1.1\r\nHost: example.com\r\n\r\n";

61
crates/core/src/network.rs
View file

@ -15,7 +15,7 @@ use crate::protocol::{
read_message_type, read_payload, write_typed_message, BlobRequestPayload, BlobResponsePayload,
MessageType, ProfileUpdatePayload,
PullSyncRequestPayload, PullSyncResponsePayload, RefuseRedirectPayload,
ALPN_V2,
ALPN,
};
use crate::storage::StoragePool;
use crate::types::{
@ -95,7 +95,6 @@ impl Network {
secret_key: iroh::SecretKey,
storage: Arc<StoragePool>,
bind_addr: Option<SocketAddr>,
secret_seed: [u8; 32],
blob_store: Arc<BlobStore>,
profile: DeviceProfile,
activity_log: Arc<std::sync::Mutex<ActivityLog>>,
@ -103,7 +102,7 @@ impl Network {
let mut builder = iroh::Endpoint::builder()
.secret_key(secret_key)
.relay_mode(iroh::RelayMode::Disabled)
.alpns(vec![ALPN_V2.to_vec()])
.alpns(vec![ALPN.to_vec()])
.clear_address_lookup(); // Remove default pkarr + DNS (no dns.iroh.link publishing)
// mDNS LAN discovery only: enables automatic peer discovery on local network
@ -236,7 +235,6 @@ impl Network {
Arc::clone(&storage),
our_node_id,
Arc::clone(&is_anchor),
secret_seed,
blob_store,
profile,
Arc::clone(&activity_log),
@ -877,7 +875,6 @@ impl Network {
let peers = self.conn_handle.connected_peers().await;
let mut total_posts = 0;
let mut total_vis = 0;
let mut success = 0;
for peer_id in peers {
@ -886,7 +883,6 @@ impl Network {
match result {
Ok(stats) => {
total_posts += stats.posts_received;
total_vis += stats.visibility_updates;
success += 1;
// Also fetch engagement data
let _ = self.conn_handle.fetch_engagement_from_peer(&peer_id).await;
@ -911,7 +907,6 @@ impl Network {
Ok(PullStats {
peers_pulled: success,
posts_received: total_posts,
visibility_updates: total_vis,
})
}
@ -995,12 +990,11 @@ impl Network {
/// Push a profile update to all audience members (ephemeral-capable).
pub async fn push_profile(&self, profile: &PublicProfile) -> usize {
// v0.6.1: profiles broadcast on the wire are keyed by the network
// NodeId. They carry ONLY routing metadata (anchors, recent_peers,
// preferred_peers) — no display name / bio / avatar. Attaching a
// human-readable name to the network id would correlate the QUIC
// endpoint to a specific person. Persona-level display data will
// travel via signed posts from v0.6.2 onward.
// Profiles broadcast on the wire are keyed by the network NodeId.
// They carry ONLY routing metadata (anchors, recent_peers) — no
// display name / bio / avatar. Attaching a human-readable name to
// the network id would correlate the QUIC endpoint to a specific
// person. Persona-level display data travels via signed posts.
let push_profile = profile.sanitized_for_network_broadcast();
let payload = ProfileUpdatePayload {
profiles: vec![push_profile],
@ -1050,32 +1044,6 @@ impl Network {
sent
}
/// Push a visibility update to all connected peers.
/// Gets connections snapshot, sends I/O outside the lock.
pub async fn push_visibility(&self, update: &crate::types::VisibilityUpdate) -> usize {
use crate::protocol::{VisibilityUpdatePayload, write_typed_message, MessageType};
let conns = self.conn_handle.get_connection_map().await;
let payload = VisibilityUpdatePayload {
updates: vec![update.clone()],
};
let mut sent = 0;
for (peer_id, conn, _, _) in &conns {
let result = async {
let mut send = conn.open_uni().await?;
write_typed_message(&mut send, MessageType::VisibilityUpdate, &payload).await?;
send.finish()?;
anyhow::Ok(())
}.await;
if result.is_ok() {
sent += 1;
} else {
debug!(peer = hex::encode(peer_id), "Failed to push visibility update");
}
}
sent
}
/// Push an updated manifest to all known holders of the file (flat set,
/// up to 5 most-recent). Replaces the legacy downstream-tree push.
pub async fn push_manifest_to_downstream(
@ -1694,7 +1662,6 @@ impl Network {
MessageType::PullSyncRequest,
&PullSyncRequestPayload {
follows: query_list,
have_post_ids: vec![], // v4: empty, using since_ms instead
since_ms: follows_sync,
},
)
@ -1711,7 +1678,6 @@ impl Network {
.as_millis() as u64;
let storage = self.storage.get().await;
let mut posts_received = 0;
let mut vis_updates = 0;
for sp in &response.posts {
if !storage.is_deleted(&sp.id)? && verify_post_id(&sp.id, &sp.post) {
if storage.store_post_with_visibility(&sp.id, &sp.post, &sp.visibility)? {
@ -1721,19 +1687,9 @@ impl Network {
let _ = storage.update_follow_last_sync(&sp.post.author, now_ms);
}
}
for vu in response.visibility_updates {
if let Some(post) = storage.get_post(&vu.post_id)? {
if post.author == vu.author {
if storage.update_post_visibility(&vu.post_id, &vu.visibility)? {
vis_updates += 1;
}
}
}
}
Ok(PullStats {
peers_pulled: 1,
posts_received,
visibility_updates: vis_updates,
})
}
@ -1864,7 +1820,7 @@ impl Network {
if let Some(addr) = addr {
match tokio::time::timeout(
std::time::Duration::from_secs(5),
self.endpoint.connect(addr, ALPN_V2),
self.endpoint.connect(addr, ALPN),
).await {
Ok(Ok(conn)) => {
self.conn_handle.mark_reachable(peer_id);
@ -2314,7 +2270,6 @@ impl Network {
pub struct PullStats {
pub peers_pulled: usize,
pub posts_received: usize,
pub visibility_updates: usize,
}
/// Decide whether a post should be sent to a requesting peer.

57
crates/core/src/node.rs
View file

@ -31,6 +31,11 @@ use crate::types::{
/// with "UnknownIssuer" because they pin the wrong cert identity.
const DEFAULT_ANCHOR: &str = "ab2b7258ef0b75b2c6ee8bf6595232055f6199d584d3c0fc10b15a1ed549aa13@itsgoin.net:4433";
/// Cooldown between relay-introduction attempts toward the same target (5 min)
const RELAY_COOLDOWN_MS: i64 = 300_000;
/// Timeout for a single relay-introduction round trip
const RELAY_INTRO_TIMEOUT_SECS: u64 = 15;
/// A distsoc node: ties together identity, storage, and networking
pub struct Node {
pub data_dir: PathBuf,
@ -342,7 +347,7 @@ impl Node {
// Load or generate identity key (network secret — QUIC endpoint only,
// never used as content author under the v0.6.1+ clean model).
let key_path = data_dir.join("identity.key");
let (mut secret_key, mut secret_seed) = if key_path.exists() {
let (mut secret_key, secret_seed) = if key_path.exists() {
let key_bytes = std::fs::read(&key_path)?;
let bytes: [u8; 32] = key_bytes
.try_into()
@ -415,7 +420,6 @@ impl Node {
std::fs::write(&key_path, new_seed)?;
info!("v0.6.1 migration: rotated network key to decouple from default posting key");
secret_key = new_key;
secret_seed = new_seed;
}
}
}
@ -429,9 +433,9 @@ impl Node {
let last_rebalance_ms = Arc::new(AtomicU64::new(0));
let last_anchor_register_ms = Arc::new(AtomicU64::new(0));
// Start network (v2: single ALPN, connection manager)
// Start network (single ALPN, connection manager)
let network = Arc::new(
Network::new(secret_key, Arc::clone(&storage), bind_addr, secret_seed, Arc::clone(&blob_store), profile, Arc::clone(&activity_log)).await?,
Network::new(secret_key, Arc::clone(&storage), bind_addr, Arc::clone(&blob_store), profile, Arc::clone(&activity_log)).await?,
);
let node_id = network.node_id_bytes();
@ -492,7 +496,7 @@ impl Node {
// NETWORK id may still carry persona fields from the unified-id
// era. Strip them in place so no code path can ever re-broadcast
// persona data bound to the device network id. Topology fields
// (anchors/recent_peers/preferred_peers) are preserved.
// (anchors/recent_peers) are preserved.
//
// (b) The manifest signature digest dropped author_addresses, so
// rows signed by pre-v0.8 builds no longer verify. Re-sign our
@ -1302,7 +1306,7 @@ impl Node {
continue;
}
match kind {
PeerSlotKind::Preferred | PeerSlotKind::Local => social.push(nid),
PeerSlotKind::Local => social.push(nid),
PeerSlotKind::Wide => wide.push(nid),
}
}
@ -1982,7 +1986,6 @@ impl Node {
let peer_addresses = storage.build_peer_addresses_for(node_id)?;
let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default().as_millis() as u64;
let preferred_tree = storage.build_preferred_tree_for(node_id).unwrap_or_default();
storage.upsert_social_route(&SocialRouteEntry {
node_id: *node_id,
addresses,
@ -1992,7 +1995,6 @@ impl Node {
last_connected_ms: 0,
last_seen_ms: now,
reach_method: ReachMethod::Direct,
preferred_tree,
})?;
Ok(())
@ -2161,7 +2163,6 @@ impl Node {
updated_at: timestamp_ms,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: true,
avatar_cid,
})
@ -2183,10 +2184,9 @@ impl Node {
let recent_peers = self.current_recent_peers().await;
let profile = {
let storage = self.storage.get().await;
let preferred_peers = storage.list_preferred_peers().unwrap_or_default();
// v0.8: the network-id-keyed profile row is TOPOLOGY ONLY
// (anchors, recent_peers, preferred_peers). Persona fields
// (anchors, recent_peers). Persona fields
// (display_name, bio, avatar_cid, public_visible) live
// exclusively on posting-id-keyed rows written by profile
// posts — never copy them onto the network row, or a legacy
@ -2198,7 +2198,6 @@ impl Node {
updated_at: now,
anchors,
recent_peers,
preferred_peers,
public_visible: true,
avatar_cid: None,
};
@ -3438,7 +3437,6 @@ impl Node {
updated_at: now,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: visible,
avatar_cid: None,
},
@ -3630,8 +3628,7 @@ impl Node {
let now = control_post.timestamp_ms;
// Clean up blob storage local-side. Blobs in remote holders become
// orphans and get evicted naturally via LRU — BlobDeleteNotice is
// gone in v0.6.2.
// orphans and get evicted naturally via LRU.
let blob_cids = {
let storage = self.storage.get().await;
let cids = storage.delete_blobs_for_post(post_id)?;
@ -3805,7 +3802,7 @@ impl Node {
storage.store_post_with_visibility(&new_post_id, &new_post, &new_vis)?;
}
// delete_post already pushes the DeleteRecord.
// delete_post propagates the deletion as a signed control post.
// Replacement post propagates via the CDN to remaining recipients.
self.delete_post(post_id).await?;
@ -4021,7 +4018,7 @@ impl Node {
{
let on_cooldown = {
let storage = self.storage.get().await;
storage.is_relay_cooldown(&peer_id, 300_000).unwrap_or(false)
storage.is_relay_cooldown(&peer_id, RELAY_COOLDOWN_MS).unwrap_or(false)
};
if !on_cooldown {
@ -4041,7 +4038,7 @@ impl Node {
);
let intro_result = tokio::time::timeout(
std::time::Duration::from_secs(15),
std::time::Duration::from_secs(RELAY_INTRO_TIMEOUT_SECS),
self.network.send_relay_introduce_standalone(relay_peer, &peer_id, *ttl),
).await;
@ -4278,7 +4275,7 @@ impl Node {
/// Start pull cycle: Protocol v4 tiered pull — 60s ticks, full pull on first tick,
/// then only pull for stale authors (last_sync_ms > 4 hours old).
pub fn start_pull_cycle(self: &Arc<Self>, _interval_secs: u64) -> tokio::task::JoinHandle<()> {
pub fn start_pull_cycle(self: &Arc<Self>) -> tokio::task::JoinHandle<()> {
let node = Arc::clone(self);
tokio::spawn(async move {
let mut interval =
@ -4404,7 +4401,7 @@ impl Node {
})
}
/// Start recovery loop: triggered when mesh drops below 2 connections.
/// Start recovery loop: triggered when the mesh empties completely.
/// Immediately reconnects to anchors and requests referrals.
pub fn start_recovery_loop(&self) -> tokio::task::JoinHandle<()> {
let network = Arc::clone(&self.network);
@ -5029,18 +5026,6 @@ impl Node {
})
}
/// No-op since v0.7.2: the `portmapper::Client` held inside the active
/// `PortMapping` auto-renews internally in its own background service.
/// Retained for API compatibility with callers in CLI and Tauri until
/// they're cleaned up. Returns `None`.
///
/// TODO(v0.7.x): wire a watcher on `mapping.watch_external()` to clear
/// anchor mode if the external address stays `None` for more than ~5min
/// (parity with the old "3 renewal failures" behavior).
pub fn start_upnp_renewal_cycle(&self) -> Option<tokio::task::JoinHandle<()>> {
None
}
// --- HTTP Post Delivery ---
/// Start the HTTP server for serving public posts to browsers.
@ -5056,9 +5041,6 @@ impl Node {
}
let storage = Arc::clone(&self.storage);
let blob_store = Arc::clone(&self.blob_store);
let downstream_addrs = Arc::new(tokio::sync::Mutex::new(
std::collections::HashMap::<[u8; 32], Vec<std::net::SocketAddr>>::new(),
));
// Advertise HTTP capability to peers
let http_addr = self.network.http_addr();
@ -5076,7 +5058,7 @@ impl Node {
info!("Starting HTTP server on TCP port {}", port);
Some(tokio::spawn(async move {
if let Err(e) = crate::http::run_http_server(port, storage, blob_store, downstream_addrs).await {
if let Err(e) = crate::http::run_http_server(port, storage, blob_store).await {
warn!("HTTP server stopped: {}", e);
}
}))
@ -6963,7 +6945,8 @@ pub fn compute_blob_priority_standalone(
impl Node {
/// Start the active replication cycle: periodically ask peers to hold our
/// under-replicated recent content. Only Available/Persistent devices initiate.
/// under-replicated recent content. All devices initiate — phones need
/// their content replicated before they go to sleep.
pub fn start_replication_cycle(self: &Arc<Self>, interval_secs: u64) -> tokio::task::JoinHandle<()> {
let node = Arc::clone(self);
tokio::spawn(async move {

1
crates/core/src/profile.rs
View file

@ -68,7 +68,6 @@ pub fn apply_profile_post_if_applicable(
updated_at: content.timestamp_ms,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: true,
avatar_cid: content.avatar_cid,
};

145
crates/core/src/protocol.rs
View file

@ -2,21 +2,14 @@ use serde::{Deserialize, Serialize};
use crate::types::{
BlobHeaderDiffOp, CdnManifest, DeleteRecord, GroupEpoch, GroupId, NodeId,
PeerWithAddress, Post, PostId, PostVisibility, PublicProfile, VisibilityUpdate, WormId,
PeerWithAddress, Post, PostId, PostVisibility, PublicProfile, WormId,
};
/// Single ALPN for Discovery Protocol v3 (N1/N2/N3 architecture)
///
/// DEPLOY GATE (v0.8): the manifest wire format + signature digest changed
/// incompatibly in this tree (AuthorManifest.author_addresses removed from
/// struct AND digest; CdnManifest reduced to 2 fields; GroupKeyRequest
/// 0xA1/0xA2 removed). Mixed-version nodes half-interop: old-signed
/// manifests fail the new digest check, new CdnManifest JSON fails old
/// deserialization. Scott ruled ALPN -> b"itsgoin/4" for v0.8 (2026-07-29,
/// scheduled Iteration B). Bump it BEFORE any build/deploy of this change
/// set so mixed generations refuse cleanly at handshake — ask Scott first
/// (feedback_ask_before_release).
pub const ALPN_V2: &[u8] = b"itsgoin/3";
/// Wire-protocol ALPN. One protocol version per build, no negotiation —
/// mismatched versions are refused at the QUIC handshake. v0.8 = itsgoin/4:
/// clean break from <=v0.7.x (zero-users ruling, design.html Appendix D
/// item 3).
pub const ALPN: &[u8] = b"itsgoin/4";
/// A post bundled with its visibility metadata for sync
#[derive(Debug, Serialize, Deserialize)]
@ -25,9 +18,8 @@ pub struct SyncPost {
pub post: Post,
pub visibility: PostVisibility,
/// Optional originator's intent, so receivers can filter control posts
/// out of the feed and process their ControlOp payload. Absent on
/// pre-v0.6.2 senders; receivers treat as "unknown"/regular post.
#[serde(default)]
/// out of the feed and process their ControlOp payload. None = regular
/// post.
pub intent: Option<crate::types::VisibilityIntent>,
}
@ -42,16 +34,10 @@ pub enum MessageType {
RefuseRedirect = 0x05,
PullSyncRequest = 0x40,
PullSyncResponse = 0x41,
// 0x42 (PostNotification), 0x43 (PostPush), 0x44 (AudienceRequest),
// 0x45 (AudienceResponse) retired in v0.6.2: persona-signed direct pushes
// are gone. Public posts propagate via the CDN; encrypted posts via pull.
ProfileUpdate = 0x50,
DeleteRecord = 0x51,
VisibilityUpdate = 0x52,
WormQuery = 0x60,
WormResponse = 0x61,
SocialAddressUpdate = 0x70,
SocialDisconnectNotice = 0x71,
SocialCheckin = 0x72,
// 0x80-0x81 reserved
BlobRequest = 0x90,
@ -59,13 +45,9 @@ pub enum MessageType {
ManifestRefreshRequest = 0x92,
ManifestRefreshResponse = 0x93,
ManifestPush = 0x94,
// 0x95 (BlobDeleteNotice) retired in v0.6.2 — remote holders evict via LRU.
// 0xA0 (GroupKeyDistribute) retired in v0.6.2 — group seeds now travel
// as encrypted posts via the CDN. See `group_key_distribution` module.
RelayIntroduce = 0xB0,
RelayIntroduceResult = 0xB1,
SessionRelay = 0xB2,
MeshPrefer = 0xB3,
CircleProfileUpdate = 0xB4,
AnchorRegister = 0xC0,
AnchorReferralRequest = 0xC1,
@ -99,12 +81,9 @@ impl MessageType {
0x40 => Some(Self::PullSyncRequest),
0x41 => Some(Self::PullSyncResponse),
0x50 => Some(Self::ProfileUpdate),
0x51 => Some(Self::DeleteRecord),
0x52 => Some(Self::VisibilityUpdate),
0x60 => Some(Self::WormQuery),
0x61 => Some(Self::WormResponse),
0x70 => Some(Self::SocialAddressUpdate),
0x71 => Some(Self::SocialDisconnectNotice),
0x72 => Some(Self::SocialCheckin),
0x90 => Some(Self::BlobRequest),
0x91 => Some(Self::BlobResponse),
@ -114,7 +93,6 @@ impl MessageType {
0xB0 => Some(Self::RelayIntroduce),
0xB1 => Some(Self::RelayIntroduceResult),
0xB2 => Some(Self::SessionRelay),
0xB3 => Some(Self::MeshPrefer),
0xB4 => Some(Self::CircleProfileUpdate),
0xC0 => Some(Self::AnchorRegister),
0xC1 => Some(Self::AnchorReferralRequest),
@ -160,34 +138,24 @@ pub struct InitialExchangePayload {
/// Our post IDs (for replica tracking)
pub post_ids: Vec<PostId>,
/// Our N+10:Addresses (connected peers with addresses) for social routing
#[serde(default)]
pub peer_addresses: Vec<PeerWithAddress>,
/// If sender is an anchor, their stable advertised address (e.g. "174.127.120.52:4433")
#[serde(default)]
pub anchor_addr: Option<String>,
/// What the sender sees as the receiver's address (STUN-like observed addr)
#[serde(default)]
pub your_observed_addr: Option<String>,
/// Sender's NAT type ("public", "easy", "hard", "unknown")
#[serde(default)]
pub nat_type: Option<String>,
/// Sender's NAT mapping behavior ("eim", "edm", "unknown")
#[serde(default)]
pub nat_mapping: Option<String>,
/// Sender's NAT filtering behavior ("open", "port_restricted", "unknown")
#[serde(default)]
pub nat_filtering: Option<String>,
/// Whether the sender is running an HTTP server for direct browser access
#[serde(default)]
pub http_capable: bool,
/// External HTTP address if known (e.g. "1.2.3.4:4433")
#[serde(default)]
pub http_addr: Option<String>,
/// CDN replication device role: "intermittent", "available", "persistent"
#[serde(default)]
pub device_role: Option<String>,
/// CDN cache pressure: 0-255 availability score (255 = lots of capacity)
#[serde(default)]
pub cache_pressure: Option<u8>,
/// Set by anchor when it detects this NodeId is already connected from a different address
#[serde(default, skip_serializing_if = "Option::is_none")]
@ -209,11 +177,7 @@ pub struct NodeListUpdatePayload {
pub struct PullSyncRequestPayload {
/// Our follows (for the responder to filter)
pub follows: Vec<NodeId>,
/// Post IDs we already have (backward compat — empty for v4 senders)
#[serde(default)]
pub have_post_ids: Vec<PostId>,
/// Protocol v4: per-author timestamps (Vec of tuples for serde compat)
#[serde(default)]
/// Per-author last-sync timestamps (Vec of tuples for serde compat)
pub since_ms: Vec<(NodeId, u64)>,
}
@ -221,7 +185,6 @@ pub struct PullSyncRequestPayload {
#[derive(Debug, Serialize, Deserialize)]
pub struct PullSyncResponsePayload {
pub posts: Vec<SyncPost>,
pub visibility_updates: Vec<VisibilityUpdate>,
}
/// Profile update (pushed via uni-stream)
@ -230,18 +193,6 @@ pub struct ProfileUpdatePayload {
pub profiles: Vec<PublicProfile>,
}
/// Delete record (pushed via uni-stream)
#[derive(Debug, Serialize, Deserialize)]
pub struct DeleteRecordPayload {
pub records: Vec<DeleteRecord>,
}
/// Visibility update (pushed via uni-stream)
#[derive(Debug, Serialize, Deserialize)]
pub struct VisibilityUpdatePayload {
pub updates: Vec<VisibilityUpdate>,
}
/// Address resolution request (bi-stream: ask reporter for a hop-2 peer's address)
#[derive(Debug, Serialize, Deserialize)]
pub struct AddressRequestPayload {
@ -254,10 +205,8 @@ pub struct AddressResponsePayload {
pub target: NodeId,
pub address: Option<String>,
/// Set when the target is known-disconnected (requester registered as watcher)
#[serde(default)]
pub disconnected_at: Option<u64>,
/// Target's N+10:Addresses if known
#[serde(default)]
pub peer_addresses: Vec<PeerWithAddress>,
}
@ -274,15 +223,12 @@ pub struct WormQueryPayload {
pub worm_id: WormId,
pub target: NodeId,
/// Additional IDs to search for (up to 10 recent_peers of target)
#[serde(default)]
pub needle_peers: Vec<NodeId>,
pub ttl: u8,
pub visited: Vec<NodeId>,
/// Optional: also search for a specific post by ID
#[serde(default)]
pub post_id: Option<PostId>,
/// Optional: also search for a specific blob by CID
#[serde(default)]
pub blob_id: Option<[u8; 32]>,
}
@ -292,19 +238,15 @@ pub struct WormResponsePayload {
pub worm_id: WormId,
pub found: bool,
/// Which needle was actually found (target or one of its recent_peers)
#[serde(default)]
pub found_id: Option<NodeId>,
pub addresses: Vec<String>,
pub reporter: Option<NodeId>,
pub hop: Option<u8>,
/// One random wide-peer referral: (node_id, address) for bloom round
#[serde(default)]
pub wide_referral: Option<(NodeId, String)>,
/// Node that holds the requested post (may differ from found_id)
#[serde(default)]
pub post_holder: Option<NodeId>,
/// Node that holds the requested blob
#[serde(default)]
pub blob_holder: Option<NodeId>,
}
@ -318,12 +260,6 @@ pub struct SocialAddressUpdatePayload {
pub peer_addresses: Vec<PeerWithAddress>,
}
/// Disconnect notice: "peer X disconnected"
#[derive(Debug, Serialize, Deserialize)]
pub struct SocialDisconnectNoticePayload {
pub node_id: NodeId,
}
/// Lightweight keepalive checkin (bidirectional)
#[derive(Debug, Serialize, Deserialize)]
pub struct SocialCheckinPayload {
@ -339,7 +275,6 @@ pub struct SocialCheckinPayload {
pub struct BlobRequestPayload {
pub cid: [u8; 32],
/// Requester's addresses so the host can record downstream
#[serde(default)]
pub requester_addresses: Vec<String>,
}
@ -349,16 +284,12 @@ pub struct BlobResponsePayload {
pub cid: [u8; 32],
pub found: bool,
/// Base64-encoded blob bytes (empty if not found)
#[serde(default)]
pub data_b64: String,
/// Author manifest + host info (if available)
#[serde(default)]
pub manifest: Option<CdnManifest>,
/// Whether host accepted requester as downstream
#[serde(default)]
pub cdn_registered: bool,
/// If not registered (host full), try these peers
#[serde(default)]
pub cdn_redirect_peers: Vec<PeerWithAddress>,
}
@ -442,18 +373,6 @@ pub struct SessionRelayPayload {
pub target: NodeId,
}
/// Mesh preference negotiation (bi-stream: request + response)
#[derive(Debug, Serialize, Deserialize)]
pub struct MeshPreferPayload {
/// true = "I want us to be preferred peers" (request)
pub requesting: bool,
/// true = "I agree to be preferred peers" (response only)
pub accepted: bool,
/// Reason for rejection (response only, when accepted=false)
#[serde(default)]
pub reject_reason: Option<String>,
}
/// Circle profile update: encrypted profile variant for a circle (uni-stream push)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CircleProfileUpdatePayload {
@ -522,7 +441,9 @@ pub struct AnchorProbeResultPayload {
pub observed_addr: Option<String>,
}
/// Port scan heartbeat during scanning hole punch (informational)
/// Port scan heartbeat during scanning hole punch (informational).
/// Reserved wire surface for the EDM scanner (`edm_port_scan_disabled_v0_7_3`
/// in connection.rs) — no sender/handler until the raw-UDP scan refactor.
#[derive(Debug, Serialize, Deserialize)]
pub struct PortScanHeartbeatPayload {
pub peer: NodeId,
@ -573,7 +494,6 @@ pub struct BlobHeaderResponsePayload {
/// True if the sender has a newer header than requested
pub updated: bool,
/// JSON-serialized BlobHeader (if updated)
#[serde(default)]
pub header_json: Option<String>,
}
@ -722,12 +642,9 @@ mod tests {
MessageType::PullSyncRequest,
MessageType::PullSyncResponse,
MessageType::ProfileUpdate,
MessageType::DeleteRecord,
MessageType::VisibilityUpdate,
MessageType::WormQuery,
MessageType::WormResponse,
MessageType::SocialAddressUpdate,
MessageType::SocialDisconnectNotice,
MessageType::SocialCheckin,
MessageType::BlobRequest,
MessageType::BlobResponse,
@ -737,7 +654,6 @@ mod tests {
MessageType::RelayIntroduce,
MessageType::RelayIntroduceResult,
MessageType::SessionRelay,
MessageType::MeshPrefer,
MessageType::CircleProfileUpdate,
MessageType::AnchorRegister,
MessageType::AnchorReferralRequest,
@ -828,43 +744,6 @@ mod tests {
assert_eq!(decoded2.reject_reason.unwrap(), "target not reachable");
}
#[test]
fn mesh_prefer_payload_roundtrip() {
// Request
let request = MeshPreferPayload {
requesting: true,
accepted: false,
reject_reason: None,
};
let json = serde_json::to_string(&request).unwrap();
let decoded: MeshPreferPayload = serde_json::from_str(&json).unwrap();
assert!(decoded.requesting);
assert!(!decoded.accepted);
assert!(decoded.reject_reason.is_none());
// Accepted response
let accept = MeshPreferPayload {
requesting: false,
accepted: true,
reject_reason: None,
};
let json2 = serde_json::to_string(&accept).unwrap();
let decoded2: MeshPreferPayload = serde_json::from_str(&json2).unwrap();
assert!(!decoded2.requesting);
assert!(decoded2.accepted);
// Rejected response
let reject = MeshPreferPayload {
requesting: false,
accepted: false,
reject_reason: Some("slots full".to_string()),
};
let json3 = serde_json::to_string(&reject).unwrap();
let decoded3: MeshPreferPayload = serde_json::from_str(&json3).unwrap();
assert!(!decoded3.accepted);
assert_eq!(decoded3.reject_reason.unwrap(), "slots full");
}
#[test]
fn session_relay_payload_roundtrip() {
let payload = SessionRelayPayload {

573
crates/core/src/storage.rs
View file

@ -308,10 +308,6 @@ impl Storage {
target_id BLOB PRIMARY KEY,
failed_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS preferred_peers (
node_id BLOB PRIMARY KEY,
agreed_at INTEGER NOT NULL
);
CREATE TABLE IF NOT EXISTS circle_profiles (
author BLOB NOT NULL,
circle_name TEXT NOT NULL,
@ -707,25 +703,10 @@ impl Storage {
)?;
}
// Add preferred_peers column to profiles if missing (Preferred Mesh Peers migration)
let has_preferred_peers = self.conn.prepare(
"SELECT COUNT(*) FROM pragma_table_info('profiles') WHERE name='preferred_peers'"
)?.query_row([], |row| row.get::<_, i64>(0))?;
if has_preferred_peers == 0 {
self.conn.execute_batch(
"ALTER TABLE profiles ADD COLUMN preferred_peers TEXT NOT NULL DEFAULT '[]';"
)?;
}
// Add preferred_tree column to social_routes if missing (Preferred Tree migration)
let has_pref_tree = self.conn.prepare(
"SELECT COUNT(*) FROM pragma_table_info('social_routes') WHERE name='preferred_tree'"
)?.query_row([], |row| row.get::<_, i64>(0))?;
if has_pref_tree == 0 {
self.conn.execute_batch(
"ALTER TABLE social_routes ADD COLUMN preferred_tree TEXT NOT NULL DEFAULT '[]';"
)?;
}
// v0.8: preferred-peer tier removed — drop the agreement table.
// (The unused profiles.preferred_peers / social_routes.preferred_tree
// columns are left in place on old dirs; no code reads them.)
self.conn.execute_batch("DROP TABLE IF EXISTS preferred_peers;")?;
// Add public_visible column to profiles if missing (Phase D-4 migration)
let has_public_visible = self.conn.prepare(
@ -851,65 +832,6 @@ impl Storage {
"CREATE INDEX IF NOT EXISTS idx_group_keys_root ON group_keys(canonical_root_post_id);"
)?;
// v0.6.2 import bugfix: the import pipeline up through f b0e293
// filed EVERY encrypted post as VisibilityIntent::Friends, including
// DMs. The Messages tab in the UI only shows posts whose intent is
// `Direct` (or `unknown` with the right visibility shape), so DMs
// imported from an "everything" bundle silently disappeared.
//
// This one-time migration reclassifies short-recipient encrypted
// posts filed as Friends back to Direct, using the same small-list
// heuristic as the import code (recipients <= 3 → Direct). A guard
// flag in the settings kv ensures this sweep runs once per DB.
let imported_dm_fixup_done = self.get_setting("mig_import_dm_fixup_v1")?.is_some();
if !imported_dm_fixup_done {
let mut fixed = 0i64;
let mut stmt = self.conn.prepare(
"SELECT id, visibility FROM posts
WHERE visibility_intent = '\"Friends\"'
AND visibility LIKE '%Encrypted%'"
)?;
let rows: Vec<(Vec<u8>, String)> = stmt
.query_map([], |row| Ok((row.get::<_, Vec<u8>>(0)?, row.get::<_, String>(1)?)))?
.filter_map(|r| r.ok())
.collect();
drop(stmt);
for (id_bytes, vis_json) in rows {
// Parse out the recipients array from the Encrypted
// visibility. Short recipient list (≤ 3) = likely DM.
let vis: Result<PostVisibility, _> = serde_json::from_str(&vis_json);
let should_fix = match vis {
Ok(PostVisibility::Encrypted { recipients }) => recipients.len() <= 3,
_ => false,
};
if !should_fix { continue; }
// Build a Direct-intent value with the recipient list recovered
// from the visibility — semantically equivalent to what the
// user's original send-side intent would have been.
let vis_parsed: PostVisibility = match serde_json::from_str(&vis_json) {
Ok(v) => v,
Err(_) => continue,
};
let recipients: Vec<NodeId> = match vis_parsed {
PostVisibility::Encrypted { recipients } => {
recipients.iter().map(|wk| wk.recipient).collect()
}
_ => continue,
};
let intent = crate::types::VisibilityIntent::Direct(recipients);
let intent_json = serde_json::to_string(&intent).unwrap_or_default();
let n = self.conn.execute(
"UPDATE posts SET visibility_intent = ?1 WHERE id = ?2",
params![intent_json, id_bytes],
)?;
fixed += n as i64;
}
self.set_setting("mig_import_dm_fixup_v1", &fixed.to_string())?;
if fixed > 0 {
tracing::info!(count = fixed, "Migrated imported DMs from Friends-intent to Direct-intent");
}
}
// Add device_role column to peers if missing (Active CDN replication)
let has_device_role = self.conn.prepare(
"SELECT COUNT(*) FROM pragma_table_info('peers') WHERE name='device_role'"
@ -942,22 +864,6 @@ impl Storage {
)?;
}
// 0.6.1-beta: seed file_holders from legacy upstream/downstream tables
// before they're dropped. Idempotent — only fires on an empty
// file_holders table.
self.seed_file_holders_from_legacy()?;
// 0.6.1-beta: drop legacy directional tables — replaced by file_holders.
self.conn.execute_batch(
"DROP TABLE IF EXISTS blob_upstream;
DROP TABLE IF EXISTS blob_downstream;
DROP TABLE IF EXISTS post_upstream;
DROP TABLE IF EXISTS post_downstream;",
)?;
// 0.6.2-beta: seed post_recipients index from existing encrypted posts.
self.seed_post_recipients_from_posts()?;
// FoF Layer 2: add comment columns for pub_x_index / group_sig /
// encrypted_payload. Old DBs have NULL → deserializes to None.
let has_comment_pub_x = self.conn.prepare(
@ -1584,7 +1490,7 @@ impl Storage {
pub fn list_discoverable_profiles(&self, self_id: &NodeId) -> anyhow::Result<Vec<crate::types::PublicProfile>> {
let mut stmt = self.conn.prepare(
"SELECT p.node_id, p.display_name, p.bio, p.updated_at,
p.anchors, p.recent_peers, p.preferred_peers,
p.anchors, p.recent_peers,
p.public_visible, p.avatar_cid
FROM profiles p
WHERE p.display_name != ''
@ -1603,9 +1509,8 @@ impl Storage {
let updated_at = row.get::<_, i64>(3)? as u64;
let anchors = parse_anchors_json(&row.get::<_, String>(4).unwrap_or_default());
let recent_peers = parse_anchors_json(&row.get::<_, String>(5).unwrap_or_default());
let preferred_peers = parse_anchors_json(&row.get::<_, String>(6).unwrap_or_default());
let public_visible: i64 = row.get(7).unwrap_or(1);
let avatar_cid: Option<Vec<u8>> = row.get(8).ok();
let public_visible: i64 = row.get(6).unwrap_or(1);
let avatar_cid: Option<Vec<u8>> = row.get(7).ok();
let avatar_cid = avatar_cid.and_then(|v| if v.len() == 32 {
let mut arr = [0u8; 32]; arr.copy_from_slice(&v); Some(arr)
} else { None });
@ -1616,7 +1521,6 @@ impl Storage {
updated_at,
anchors,
recent_peers,
preferred_peers,
public_visible: public_visible != 0,
avatar_cid,
});
@ -2009,12 +1913,9 @@ impl Storage {
let recent_peers_json = serde_json::to_string(
&profile.recent_peers.iter().map(hex::encode).collect::<Vec<_>>()
)?;
let preferred_peers_json = serde_json::to_string(
&profile.preferred_peers.iter().map(hex::encode).collect::<Vec<_>>()
)?;
let avatar_cid_slice = profile.avatar_cid.as_ref().map(|c| c.as_slice());
self.conn.execute(
"INSERT OR REPLACE INTO profiles (node_id, display_name, bio, updated_at, anchors, recent_peers, preferred_peers, public_visible, avatar_cid) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)",
"INSERT OR REPLACE INTO profiles (node_id, display_name, bio, updated_at, anchors, recent_peers, public_visible, avatar_cid) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
params![
profile.node_id.as_slice(),
profile.display_name,
@ -2022,7 +1923,6 @@ impl Storage {
profile.updated_at as i64,
anchors_json,
recent_peers_json,
preferred_peers_json,
profile.public_visible as i64,
avatar_cid_slice,
],
@ -2161,15 +2061,14 @@ impl Storage {
/// Get a profile by node ID
pub fn get_profile(&self, node_id: &NodeId) -> anyhow::Result<Option<PublicProfile>> {
let mut stmt = self.conn.prepare(
"SELECT node_id, display_name, bio, updated_at, anchors, recent_peers, preferred_peers, public_visible, avatar_cid FROM profiles WHERE node_id = ?1",
"SELECT node_id, display_name, bio, updated_at, anchors, recent_peers, public_visible, avatar_cid FROM profiles WHERE node_id = ?1",
)?;
let mut rows = stmt.query(params![node_id.as_slice()])?;
if let Some(row) = rows.next()? {
let anchors = parse_anchors_json(&row.get::<_, String>(4)?);
let recent_peers = parse_anchors_json(&row.get::<_, String>(5).unwrap_or_else(|_| "[]".to_string()));
let preferred_peers = parse_anchors_json(&row.get::<_, String>(6).unwrap_or_else(|_| "[]".to_string()));
let public_visible = row.get::<_, i64>(7).unwrap_or(1) != 0;
let avatar_cid = row.get::<_, Option<Vec<u8>>>(8).unwrap_or(None)
let public_visible = row.get::<_, i64>(6).unwrap_or(1) != 0;
let avatar_cid = row.get::<_, Option<Vec<u8>>>(7).unwrap_or(None)
.and_then(|b| <[u8; 32]>::try_from(b.as_slice()).ok());
Ok(Some(PublicProfile {
node_id: blob_to_nodeid(row.get(0)?)?,
@ -2178,7 +2077,6 @@ impl Storage {
updated_at: row.get::<_, i64>(3)? as u64,
anchors,
recent_peers,
preferred_peers,
public_visible,
avatar_cid,
}))
@ -2191,7 +2089,7 @@ impl Storage {
pub fn list_profiles(&self) -> anyhow::Result<Vec<PublicProfile>> {
let mut stmt = self
.conn
.prepare("SELECT node_id, display_name, bio, updated_at, anchors, recent_peers, preferred_peers, public_visible, avatar_cid FROM profiles")?;
.prepare("SELECT node_id, display_name, bio, updated_at, anchors, recent_peers, public_visible, avatar_cid FROM profiles")?;
let rows = stmt.query_map([], |row| {
let node_id_bytes: Vec<u8> = row.get(0)?;
let display_name: String = row.get(1)?;
@ -2199,14 +2097,13 @@ impl Storage {
let updated_at: i64 = row.get(3)?;
let anchors_json: String = row.get(4)?;
let recent_peers_json: String = row.get::<_, String>(5).unwrap_or_else(|_| "[]".to_string());
let preferred_peers_json: String = row.get::<_, String>(6).unwrap_or_else(|_| "[]".to_string());
let public_visible: i64 = row.get::<_, i64>(7).unwrap_or(1);
let avatar_cid_bytes: Option<Vec<u8>> = row.get::<_, Option<Vec<u8>>>(8).unwrap_or(None);
Ok((node_id_bytes, display_name, bio, updated_at, anchors_json, recent_peers_json, preferred_peers_json, public_visible, avatar_cid_bytes))
let public_visible: i64 = row.get::<_, i64>(6).unwrap_or(1);
let avatar_cid_bytes: Option<Vec<u8>> = row.get::<_, Option<Vec<u8>>>(7).unwrap_or(None);
Ok((node_id_bytes, display_name, bio, updated_at, anchors_json, recent_peers_json, public_visible, avatar_cid_bytes))
})?;
let mut profiles = Vec::new();
for row in rows {
let (node_id_bytes, display_name, bio, updated_at, anchors_json, recent_peers_json, preferred_peers_json, public_visible, avatar_cid_bytes) = row?;
let (node_id_bytes, display_name, bio, updated_at, anchors_json, recent_peers_json, public_visible, avatar_cid_bytes) = row?;
let avatar_cid = avatar_cid_bytes.and_then(|b| <[u8; 32]>::try_from(b.as_slice()).ok());
profiles.push(PublicProfile {
node_id: blob_to_nodeid(node_id_bytes)?,
@ -2215,7 +2112,6 @@ impl Storage {
updated_at: updated_at as u64,
anchors: parse_anchors_json(&anchors_json),
recent_peers: parse_anchors_json(&recent_peers_json),
preferred_peers: parse_anchors_json(&preferred_peers_json),
public_visible: public_visible != 0,
avatar_cid,
});
@ -3862,105 +3758,6 @@ impl Storage {
Ok(())
}
// ---- Preferred Peers ----
/// Add a bilateral preferred peer agreement.
pub fn add_preferred_peer(&self, node_id: &NodeId) -> anyhow::Result<()> {
let now = now_ms();
self.conn.execute(
"INSERT OR REPLACE INTO preferred_peers (node_id, agreed_at) VALUES (?1, ?2)",
params![node_id.as_slice(), now],
)?;
Ok(())
}
/// Remove a preferred peer agreement.
pub fn remove_preferred_peer(&self, node_id: &NodeId) -> anyhow::Result<()> {
self.conn.execute(
"DELETE FROM preferred_peers WHERE node_id = ?1",
params![node_id.as_slice()],
)?;
Ok(())
}
/// List all preferred peers.
pub fn list_preferred_peers(&self) -> anyhow::Result<Vec<NodeId>> {
let mut stmt = self.conn.prepare(
"SELECT node_id FROM preferred_peers ORDER BY agreed_at DESC",
)?;
let mut result = Vec::new();
let mut rows = stmt.query([])?;
while let Some(row) = rows.next()? {
result.push(blob_to_nodeid(row.get(0)?)?);
}
Ok(result)
}
/// Check if a peer is a preferred peer.
pub fn is_preferred_peer(&self, node_id: &NodeId) -> anyhow::Result<bool> {
let count: i64 = self.conn.query_row(
"SELECT COUNT(*) FROM preferred_peers WHERE node_id = ?1",
params![node_id.as_slice()],
|row| row.get(0),
)?;
Ok(count > 0)
}
/// Count preferred peers.
pub fn count_preferred_peers(&self) -> anyhow::Result<usize> {
let count: i64 = self.conn.query_row(
"SELECT COUNT(*) FROM preferred_peers",
[],
|row| row.get(0),
)?;
Ok(count as usize)
}
// ---- Preferred Tree ----
/// Build 2-layer preferred peer tree from stored profiles.
/// Layer 0: target. Layer 1: target's preferred_peers. Layer 2: each L1 peer's preferred_peers.
/// Returns ~100 unique NodeIds.
pub fn build_preferred_tree_for(&self, target: &NodeId) -> anyhow::Result<Vec<NodeId>> {
let mut tree = std::collections::HashSet::new();
// Layer 0: target itself
tree.insert(*target);
// Layer 1: target's preferred peers from their profile
let l1_peers = match self.get_profile(target)? {
Some(profile) => profile.preferred_peers,
None => return Ok(tree.into_iter().collect()),
};
for pp in &l1_peers {
tree.insert(*pp);
}
// Layer 2: each L1 peer's preferred peers
for pp in &l1_peers {
if let Some(profile) = self.get_profile(pp)? {
for pp2 in &profile.preferred_peers {
tree.insert(*pp2);
}
}
}
Ok(tree.into_iter().collect())
}
/// Update the preferred_tree JSON for a social route.
pub fn update_social_route_preferred_tree(&self, node_id: &NodeId, tree: &[NodeId]) -> anyhow::Result<()> {
let json = serde_json::to_string(
&tree.iter().map(hex::encode).collect::<Vec<_>>()
)?;
self.conn.execute(
"UPDATE social_routes SET preferred_tree = ?1 WHERE node_id = ?2",
params![json, node_id.as_slice()],
)?;
Ok(())
}
// ---- Social Routes ----
/// Insert or update a social route entry.
@ -3969,17 +3766,14 @@ impl Storage {
&entry.addresses.iter().map(|a| a.to_string()).collect::<Vec<_>>()
)?;
let peer_addrs_json = serde_json::to_string(&entry.peer_addresses)?;
let pref_tree_json = serde_json::to_string(
&entry.preferred_tree.iter().map(hex::encode).collect::<Vec<_>>()
)?;
self.conn.execute(
"INSERT INTO social_routes (node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method, preferred_tree)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9)
"INSERT INTO social_routes (node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method)
VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)
ON CONFLICT(node_id) DO UPDATE SET
addresses = ?2, peer_addresses = ?3, relation = ?4, status = ?5,
last_connected_ms = MAX(social_routes.last_connected_ms, ?6),
last_seen_ms = MAX(social_routes.last_seen_ms, ?7),
reach_method = ?8, preferred_tree = ?9",
reach_method = ?8",
params![
entry.node_id.as_slice(),
addrs_json,
@ -3989,7 +3783,6 @@ impl Storage {
entry.last_connected_ms as i64,
entry.last_seen_ms as i64,
entry.reach_method.to_string(),
pref_tree_json,
],
)?;
Ok(())
@ -3998,7 +3791,7 @@ impl Storage {
/// Get a single social route entry.
pub fn get_social_route(&self, node_id: &NodeId) -> anyhow::Result<Option<SocialRouteEntry>> {
let mut stmt = self.conn.prepare(
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method, preferred_tree
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method
FROM social_routes WHERE node_id = ?1",
)?;
let mut rows = stmt.query(params![node_id.as_slice()])?;
@ -4079,7 +3872,7 @@ impl Storage {
/// List all social routes, sorted by last_seen DESC.
pub fn list_social_routes(&self) -> anyhow::Result<Vec<SocialRouteEntry>> {
let mut stmt = self.conn.prepare(
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method, preferred_tree
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method
FROM social_routes ORDER BY last_seen_ms DESC",
)?;
let mut entries = Vec::new();
@ -4094,7 +3887,7 @@ impl Storage {
pub fn list_stale_social_routes(&self, max_age_ms: u64) -> anyhow::Result<Vec<SocialRouteEntry>> {
let cutoff = now_ms() - max_age_ms as i64;
let mut stmt = self.conn.prepare(
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method, preferred_tree
"SELECT node_id, addresses, peer_addresses, relation, status, last_connected_ms, last_seen_ms, reach_method
FROM social_routes WHERE last_seen_ms < ?1 ORDER BY last_seen_ms ASC",
)?;
let mut entries = Vec::new();
@ -4137,9 +3930,6 @@ impl Storage {
// Build peer_addresses from the contact's profile recent_peers
let peer_addresses = self.build_peer_addresses_for(&nid)?;
// Build preferred peer tree
let preferred_tree = self.build_preferred_tree_for(&nid).unwrap_or_default();
// Only insert if not already present (don't overwrite runtime state)
if !self.has_social_route(&nid)? {
self.upsert_social_route(&SocialRouteEntry {
@ -4151,12 +3941,8 @@ impl Storage {
last_connected_ms: 0,
last_seen_ms: now,
reach_method: ReachMethod::Direct,
preferred_tree,
})?;
count += 1;
} else {
// Update the preferred tree for existing routes
self.update_social_route_preferred_tree(&nid, &preferred_tree)?;
}
}
@ -4780,36 +4566,6 @@ impl Storage {
Ok(out)
}
/// Seed the post_recipients index from existing encrypted posts.
/// One-time idempotent migration for users upgrading from pre-0.6.2.
pub fn seed_post_recipients_from_posts(&self) -> anyhow::Result<()> {
let existing: i64 = self.conn.prepare("SELECT COUNT(*) FROM post_recipients")?
.query_row([], |row| row.get(0))?;
if existing > 0 {
return Ok(());
}
// Scan all posts, parse visibility, index recipients.
let mut stmt = self.conn.prepare("SELECT id, visibility FROM posts")?;
let rows = stmt.query_map([], |row| {
let id_bytes: Vec<u8> = row.get(0)?;
let vis_json: String = row.get(1)?;
Ok((id_bytes, vis_json))
})?;
let entries: Vec<([u8; 32], PostVisibility)> = rows
.filter_map(|r| r.ok())
.filter_map(|(id_bytes, vis_json)| {
let pid = <[u8; 32]>::try_from(id_bytes.as_slice()).ok()?;
let vis: PostVisibility = serde_json::from_str(&vis_json).ok()?;
Some((pid, vis))
})
.collect();
drop(stmt);
for (pid, vis) in entries {
self.index_post_recipients(&pid, &vis)?;
}
Ok(())
}
// --- Posting identities (multi-persona plumbing) ---
pub fn upsert_posting_identity(&self, id: &PostingIdentity) -> anyhow::Result<()> {
@ -5840,55 +5596,6 @@ impl Storage {
Ok(())
}
/// One-time migration: seed file_holders from the legacy upstream/downstream
/// tables so a user upgrading from pre-0.6.1 doesn't start with empty holder
/// sets. Idempotent — inserts use ON CONFLICT DO NOTHING semantics via the
/// PRIMARY KEY. Skips tables that don't exist on fresh installs.
pub fn seed_file_holders_from_legacy(&self) -> anyhow::Result<()> {
// Skip if file_holders already populated (idempotent re-run protection).
let existing: i64 = self.conn.prepare("SELECT COUNT(*) FROM file_holders")?
.query_row([], |row| row.get(0))?;
if existing > 0 {
return Ok(());
}
let now = now_ms() as i64;
let table_exists = |name: &str| -> anyhow::Result<bool> {
let count: i64 = self.conn.prepare(
"SELECT COUNT(*) FROM sqlite_master WHERE type='table' AND name=?1",
)?.query_row(params![name], |row| row.get(0))?;
Ok(count > 0)
};
if table_exists("post_upstream")? {
self.conn.execute(
"INSERT OR IGNORE INTO file_holders (file_id, peer_id, peer_addresses, last_interaction_ms, direction)
SELECT post_id, peer_node_id, '[]', ?1, 'received' FROM post_upstream",
params![now],
)?;
}
if table_exists("post_downstream")? {
self.conn.execute(
"INSERT OR IGNORE INTO file_holders (file_id, peer_id, peer_addresses, last_interaction_ms, direction)
SELECT post_id, peer_node_id, '[]', ?1, 'sent' FROM post_downstream",
params![now],
)?;
}
if table_exists("blob_upstream")? {
self.conn.execute(
"INSERT OR IGNORE INTO file_holders (file_id, peer_id, peer_addresses, last_interaction_ms, direction)
SELECT cid, source_node_id, source_addresses, ?1, 'received' FROM blob_upstream",
params![now],
)?;
}
if table_exists("blob_downstream")? {
self.conn.execute(
"INSERT OR IGNORE INTO file_holders (file_id, peer_id, peer_addresses, last_interaction_ms, direction)
SELECT cid, peer_node_id, peer_addresses, ?1, 'sent' FROM blob_downstream",
params![now],
)?;
}
Ok(())
}
// --- Engagement: reactions ---
/// Store a reaction (upsert by reactor+post_id+emoji).
@ -6659,8 +6366,6 @@ fn row_to_social_route(row: &rusqlite::Row) -> anyhow::Result<SocialRouteEntry>
let last_seen_ms = row.get::<_, i64>(6)? as u64;
let method_str: String = row.get(7)?;
let reach_method: ReachMethod = method_str.parse().unwrap_or(ReachMethod::Direct);
let pref_tree_json: String = row.get::<_, String>(8).unwrap_or_else(|_| "[]".to_string());
let preferred_tree = parse_anchors_json(&pref_tree_json);
Ok(SocialRouteEntry {
node_id,
@ -6671,7 +6376,6 @@ fn row_to_social_route(row: &rusqlite::Row) -> anyhow::Result<SocialRouteEntry>
last_connected_ms,
last_seen_ms,
reach_method,
preferred_tree,
})
}
@ -6899,7 +6603,6 @@ mod tests {
last_connected_ms: 0,
last_seen_ms: 1000,
reach_method: ReachMethod::Direct,
preferred_tree: vec![],
}).unwrap();
// Disconnected routes should NOT be in N1 share
@ -6994,7 +6697,6 @@ mod tests {
last_connected_ms: 1000,
last_seen_ms: 2000,
reach_method: ReachMethod::Direct,
preferred_tree: vec![],
};
s.upsert_social_route(&entry).unwrap();
@ -7054,7 +6756,6 @@ mod tests {
last_connected_ms: 0,
last_seen_ms: 0,
reach_method: ReachMethod::Direct,
preferred_tree: vec![],
};
s.upsert_social_route(&entry).unwrap();
@ -7396,237 +7097,20 @@ mod tests {
assert_eq!(got_pubkey, expected_pubkey);
}
// ---- Preferred peers tests ----
#[test]
fn preferred_peers_crud() {
let s = temp_storage();
let peer_a = make_node_id(1);
let peer_b = make_node_id(2);
assert_eq!(s.count_preferred_peers().unwrap(), 0);
assert!(!s.is_preferred_peer(&peer_a).unwrap());
s.add_preferred_peer(&peer_a).unwrap();
s.add_preferred_peer(&peer_b).unwrap();
assert_eq!(s.count_preferred_peers().unwrap(), 2);
assert!(s.is_preferred_peer(&peer_a).unwrap());
assert!(s.is_preferred_peer(&peer_b).unwrap());
let list = s.list_preferred_peers().unwrap();
assert_eq!(list.len(), 2);
assert!(list.contains(&peer_a));
assert!(list.contains(&peer_b));
s.remove_preferred_peer(&peer_a).unwrap();
assert!(!s.is_preferred_peer(&peer_a).unwrap());
assert_eq!(s.count_preferred_peers().unwrap(), 1);
}
#[test]
fn preferred_peers_idempotent() {
let s = temp_storage();
let peer = make_node_id(1);
s.add_preferred_peer(&peer).unwrap();
s.add_preferred_peer(&peer).unwrap(); // no error on duplicate
assert_eq!(s.count_preferred_peers().unwrap(), 1);
}
#[test]
fn profile_stores_preferred_peers() {
let s = temp_storage();
let nid = make_node_id(1);
let pref_a = make_node_id(10);
let pref_b = make_node_id(11);
let profile = PublicProfile {
node_id: nid,
display_name: "test".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![pref_a, pref_b],
public_visible: true,
avatar_cid: None,
};
s.store_profile(&profile).unwrap();
let got = s.get_profile(&nid).unwrap().unwrap();
assert_eq!(got.preferred_peers.len(), 2);
assert!(got.preferred_peers.contains(&pref_a));
assert!(got.preferred_peers.contains(&pref_b));
assert!(got.public_visible);
assert!(got.avatar_cid.is_none());
}
#[test]
fn preferred_slot_counts() {
assert_eq!(DeviceProfile::Desktop.preferred_slots(), 10);
fn slot_counts() {
assert_eq!(DeviceProfile::Desktop.local_slots(), 71);
assert_eq!(DeviceProfile::Mobile.preferred_slots(), 3);
assert_eq!(DeviceProfile::Mobile.local_slots(), 7);
// Total unchanged
assert_eq!(
DeviceProfile::Desktop.preferred_slots() + DeviceProfile::Desktop.local_slots() + DeviceProfile::Desktop.wide_slots(),
101
DeviceProfile::Desktop.local_slots() + DeviceProfile::Desktop.wide_slots(),
91
);
assert_eq!(
DeviceProfile::Mobile.preferred_slots() + DeviceProfile::Mobile.local_slots() + DeviceProfile::Mobile.wide_slots(),
15
DeviceProfile::Mobile.local_slots() + DeviceProfile::Mobile.wide_slots(),
12
);
}
#[test]
fn peer_slot_kind_preferred_roundtrip() {
let kind: PeerSlotKind = "preferred".parse().unwrap();
assert_eq!(kind, PeerSlotKind::Preferred);
assert_eq!(kind.to_string(), "preferred");
}
// ---- Preferred tree tests ----
#[test]
fn build_preferred_tree_empty() {
let s = temp_storage();
let target = make_node_id(1);
// No profile stored — tree should just contain the target
let tree = s.build_preferred_tree_for(&target).unwrap();
assert_eq!(tree.len(), 1);
assert!(tree.contains(&target));
}
#[test]
fn build_preferred_tree_two_layers() {
let s = temp_storage();
let target = make_node_id(1);
let l1_a = make_node_id(10);
let l1_b = make_node_id(11);
let l2_a1 = make_node_id(20);
let l2_a2 = make_node_id(21);
let l2_b1 = make_node_id(30);
// Target's profile with 2 preferred peers
s.store_profile(&PublicProfile {
node_id: target,
display_name: "target".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![l1_a, l1_b],
public_visible: true,
avatar_cid: None,
}).unwrap();
// L1 peer A's profile with 2 preferred peers
s.store_profile(&PublicProfile {
node_id: l1_a,
display_name: "l1a".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![l2_a1, l2_a2],
public_visible: true,
avatar_cid: None,
}).unwrap();
// L1 peer B's profile with 1 preferred peer
s.store_profile(&PublicProfile {
node_id: l1_b,
display_name: "l1b".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![l2_b1],
public_visible: true,
avatar_cid: None,
}).unwrap();
let tree = s.build_preferred_tree_for(&target).unwrap();
// Should contain: target, l1_a, l1_b, l2_a1, l2_a2, l2_b1 = 6 unique nodes
assert_eq!(tree.len(), 6);
assert!(tree.contains(&target));
assert!(tree.contains(&l1_a));
assert!(tree.contains(&l1_b));
assert!(tree.contains(&l2_a1));
assert!(tree.contains(&l2_a2));
assert!(tree.contains(&l2_b1));
}
#[test]
fn build_preferred_tree_deduplicates() {
let s = temp_storage();
let target = make_node_id(1);
let shared = make_node_id(10);
let l1_a = make_node_id(11);
// Target's preferred peers include shared
s.store_profile(&PublicProfile {
node_id: target,
display_name: "target".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![l1_a, shared],
public_visible: true,
avatar_cid: None,
}).unwrap();
// L1 peer's preferred peers also include shared
s.store_profile(&PublicProfile {
node_id: l1_a,
display_name: "l1a".to_string(),
bio: "".to_string(),
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![shared, target],
public_visible: true,
avatar_cid: None,
}).unwrap();
let tree = s.build_preferred_tree_for(&target).unwrap();
// Should contain: target, l1_a, shared = 3 unique nodes (no duplicates)
assert_eq!(tree.len(), 3);
}
#[test]
fn social_route_preferred_tree_roundtrip() {
use crate::types::{ReachMethod, SocialRelation, SocialRouteEntry, SocialStatus};
let s = temp_storage();
let nid = make_node_id(1);
let tree_node = make_node_id(10);
let entry = SocialRouteEntry {
node_id: nid,
addresses: vec![],
peer_addresses: vec![],
relation: SocialRelation::Follow,
status: SocialStatus::Online,
last_connected_ms: 0,
last_seen_ms: 1000,
reach_method: ReachMethod::Direct,
preferred_tree: vec![tree_node],
};
s.upsert_social_route(&entry).unwrap();
let got = s.get_social_route(&nid).unwrap().unwrap();
assert_eq!(got.preferred_tree.len(), 1);
assert!(got.preferred_tree.contains(&tree_node));
// Update preferred tree
let new_tree = vec![make_node_id(20), make_node_id(21)];
s.update_social_route_preferred_tree(&nid, &new_tree).unwrap();
let got2 = s.get_social_route(&nid).unwrap().unwrap();
assert_eq!(got2.preferred_tree.len(), 2);
}
// ---- Circle Profile tests ----
#[test]
@ -7688,7 +7172,6 @@ mod tests {
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: true,
avatar_cid: None,
}).unwrap();
@ -7735,7 +7218,6 @@ mod tests {
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: false,
avatar_cid: None,
}).unwrap();
@ -7760,7 +7242,6 @@ mod tests {
updated_at: 1000,
anchors: vec![],
recent_peers: vec![],
preferred_peers: vec![],
public_visible: true,
avatar_cid: None,
}).unwrap();

47
crates/core/src/types.rs
View file

@ -71,8 +71,8 @@ pub struct Attachment {
/// Public profile — plaintext, synced to all peers.
///
/// v0.8 keying (dual-purpose struct — the `node_id` decides which class):
/// - NETWORK-id rows carry TOPOLOGY ONLY (anchors, recent_peers,
/// preferred_peers); persona fields stay empty. They travel via
/// - NETWORK-id rows carry TOPOLOGY ONLY (anchors, recent_peers);
/// persona fields stay empty. They travel via
/// ProfileUpdate / InitialExchange, always through
/// `sanitized_for_network_broadcast()`.
/// - POSTING-id rows carry PERSONA fields (display_name, bio, avatar_cid,
@ -99,9 +99,6 @@ pub struct PublicProfile {
/// Up to 10 currently-connected peer NodeIds (for 11-needle worm search)
#[serde(default)]
pub recent_peers: Vec<NodeId>,
/// Bilateral preferred peer NodeIds (stable relay hubs)
#[serde(default)]
pub preferred_peers: Vec<NodeId>,
/// Whether display_name/bio are visible to non-circle peers
#[serde(default = "default_true")]
pub public_visible: bool,
@ -113,7 +110,7 @@ pub struct PublicProfile {
impl PublicProfile {
/// Return a copy with persona-level display data (display_name, bio,
/// avatar_cid) stripped, leaving only the routing metadata (anchors,
/// recent_peers, preferred_peers). v0.6.1 broadcasts the profile under
/// recent_peers). v0.6.1 broadcasts the profile under
/// the network NodeId; attaching a human-readable name to that key would
/// correlate the network endpoint to a specific person. Persona display
/// data will travel via signed posts from v0.6.2 onward.
@ -465,14 +462,6 @@ pub struct DeleteRecord {
pub signature: Vec<u8>,
}
/// An update to a post's visibility (new wrapped keys after revocation)
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VisibilityUpdate {
pub post_id: PostId,
pub author: NodeId,
pub visibility: PostVisibility,
}
/// How to handle revoking a recipient's access to past encrypted posts
#[derive(Debug, Clone, Copy)]
pub enum RevocationMode {
@ -662,20 +651,13 @@ impl NatProfile {
/// Device profile — determines connection slot budget
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeviceProfile {
/// Desktop: 81 local + 20 wide = 101 mesh connections
/// Desktop: 71 local + 20 wide = 91 mesh connections
Desktop,
/// Mobile: 10 local + 5 wide = 15 mesh connections
/// Mobile: 7 local + 5 wide = 12 mesh connections
Mobile,
}
impl DeviceProfile {
pub fn preferred_slots(&self) -> usize {
match self {
DeviceProfile::Desktop => 10,
DeviceProfile::Mobile => 3,
}
}
pub fn local_slots(&self) -> usize {
match self {
DeviceProfile::Desktop => 71,
@ -729,36 +711,21 @@ impl std::fmt::Display for SessionReachMethod {
/// Slot kind for mesh connections
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PeerSlotKind {
/// Bilateral preferred connections (Desktop: 10, Mobile: 3)
Preferred,
/// Diverse local connections (Desktop: 71, Mobile: 7)
Local,
/// Bloom-sourced random distant connections (20 slots)
/// Bloom-sourced random distant connections (Desktop: 20, Mobile: 5)
Wide,
}
impl std::fmt::Display for PeerSlotKind {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PeerSlotKind::Preferred => write!(f, "preferred"),
PeerSlotKind::Local => write!(f, "local"),
PeerSlotKind::Wide => write!(f, "wide"),
}
}
}
impl std::str::FromStr for PeerSlotKind {
type Err = anyhow::Error;
fn from_str(s: &str) -> Result<Self, Self::Err> {
match s {
"preferred" => Ok(PeerSlotKind::Preferred),
"local" | "social" => Ok(PeerSlotKind::Local),
"wide" => Ok(PeerSlotKind::Wide),
_ => Err(anyhow::anyhow!("unknown slot kind: {}", s)),
}
}
}
// --- Social Routing Cache ---
/// How we last reached a social contact
@ -906,8 +873,6 @@ pub struct SocialRouteEntry {
pub last_connected_ms: u64,
pub last_seen_ms: u64,
pub reach_method: ReachMethod,
/// 2-layer preferred peer tree (~100 nodes) for fast relay candidate search
pub preferred_tree: Vec<NodeId>,
}
// --- Engagement System ---

27
crates/core/src/web.rs
View file

@ -3,8 +3,9 @@
//! renders HTML, serves blobs. No permanent storage of fetched content.
//!
//! Routes (behind Apache reverse proxy):
//! GET /p/<postid_hex>/<author_hex> → render post HTML (fetched on-demand)
//! GET /b/<blobid_hex> → serve blob (images/videos)
//! GET /p/<postid_hex> → render post HTML (fetched on-demand; anchor
//! resolves holders itself)
//! GET /b/<blobid_hex> → serve blob (images/videos)
use std::net::SocketAddr;
use std::sync::Arc;
@ -89,7 +90,7 @@ fn extract_header<'a>(buf: &'a [u8], name: &str) -> Option<&'a str> {
None
}
/// Handle GET /p/<postid_hex>/<author_hex>
/// Handle GET /p/<postid_hex>
///
/// Three-tier serving:
/// 1. Redirect to a CDN holder with a public/punchable HTTP server
@ -112,26 +113,11 @@ async fn serve_post(stream: &mut TcpStream, path: &str, node: &Arc<Node>, browse
_ => return,
};
// Parse optional author_id (after the slash)
let author_id: Option<NodeId> = if rest.len() > 65 {
let author_hex = &rest[65..];
if author_hex.len() == 64 && author_hex.chars().all(|c| c.is_ascii_hexdigit()) {
hex::decode(author_hex).ok().and_then(|b| b.try_into().ok())
} else {
None
}
} else {
None
};
// Single lock: gather holders, local post, AND author name if local
let (holders, local_post, local_author_name) = {
let store = node.storage.get().await;
let mut holders = Vec::new();
if let Some(author) = author_id {
holders.push(author);
}
if let Ok(file_holders) = store.get_file_holders(&post_id) {
for (peer, _addrs) in file_holders {
if !holders.contains(&peer) {
@ -177,8 +163,9 @@ async fn serve_post(stream: &mut TcpStream, path: &str, node: &Arc<Node>, browse
}
}
// Fetch via content search + PostFetch
let author = author_id.unwrap_or([0u8; 32]);
// Fetch via content search + PostFetch (no author hint — holders resolved
// from file_holders / worm search)
let author: NodeId = [0u8; 32];
info!("Web: proxying post {} via QUIC (no redirect candidate found)", post_hex);
let search_result = tokio::time::timeout(