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

feat: v0.8 Iteration C — narrow mesh, N1-N4 uniques index, anchor convection

Browse source 
Topology rework. The mesh's job is width; depth now comes from knowledge.

Slots:
- Local/Wide collapse into ONE mesh pool: 20 desktop / 15 mobile
- Temp referral slots +4..+10 ABOVE the cap (10 desktop / 4 mobile), carry no
  knowledge exchange, graduate into a freed mesh slot or expire (5 min TTL),
  never evict an established mesh peer
- Fixes the Wide-never-fills bug: outbound paths hardcoded Local while the
  growth gate counted Local only, so growth stopped at 71 and 20 slots were
  inbound-only. Also closes register_connection paths that enforced no cap.

Knowledge — two-pool uniques announce (replaces N1/N2/N3 share lists):
- Pool 1 (forwardable): our N0-N2 uniques, deduped. Receiver stores shifted one
  bounce deeper as N1-N3, reporter-tagged.
- Pool 2 (terminal): our N3, deduped against pool 1. Receiver stores as N4,
  USED for search/resolution but NEVER re-announced. No N5 anywhere.
- Uniques merge mesh peers + social directs + CDN file authors + holder peers,
  so receivers cannot tell how we know an ID
- Anchor entries carry addresses; every other entry is a bare ID. The pools
  double as the anchor directory.
- Dedup at both store and announce; per-bounce caps; 2 MB payload ceiling
- Device-tiered depth: mobile holds 3 bounces and drops the terminal pool

Anchor convection (register loop retired — 0xC0 AnchorRegister deleted):
- Rolling in-memory caller window (no registration DB): a caller gets the 2
  most recent viable prior callers, its address goes to the next 2, then
  rotates out; still-connected callers preferred; RelayIntroduce coordinated
  when both ends are live
- Request classes: entry (<2 connections) always served; top-up refused
  cheaply under load, and the refusal feeds the adaptive weights
- Stochastic per-disconnect action {nothing | anchor intro | mesh intro} with
  weights from local anchor density + refusal feedback. No thresholds, no
  jitter timer. Recovery (mesh < 2) stays immediate and non-stochastic.
- Anchors mined from the uniques pools incl. retained pools of disconnected
  peers; known_anchors demoted to a bootstrap cache

Security fixes from review: remote address poisoning (hearsay anchors could be
dialled — known_anchors/is_anchor now written only from completed handshakes),
N4 leaking back into announcements via the anchor mirror (infinite propagation),
missing payload size caps, unbounded candidate scoring, and a bulk SQLite write
held under the ConnectionManager mutex.

deploy.sh: version regex now captures pre-release suffixes (0.8.0-alpha would
have truncated to 0.8.0 and broken every artifact path), announce channel
derives from the suffix, and the anchor runs --publish-registry at genesis.

228 core tests; a3 integration 9/9; new c_topology_test.sh 33/33 (both
independently re-run). EDM corpse, session-relay gating, Iteration A/B intact.

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 13:03:12 -04:00
parent 36e3871c4b
commit e756fabb11
14 changed files with 5503 additions and 2367 deletions
Download patch file Download diff file Expand all files Collapse all files

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

File diff suppressed because it is too large Load diff

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

@ -14,7 +14,7 @@ use crate::content::compute_post_id;
use crate::crypto;
use crate::storage::Storage;
use crate::types::{
GroupKeyDistributionContent, GroupKeyRecord, GroupMemberKey, NodeId, Post, PostId,
GroupKeyDistributionContent, GroupKeyRecord, NodeId, Post, PostId,
PostVisibility, PostingIdentity, VisibilityIntent,
};

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

@ -10,16 +10,15 @@ use tracing::{debug, error, info, warn};
use crate::activity::{ActivityCategory, ActivityLevel, ActivityLog};
use crate::blob::BlobStore;
use crate::connection::{initial_exchange_accept, initial_exchange_connect, ConnHandle, ConnectionActor, ConnectionManager, ExchangeResult};
use crate::content::verify_post_id;
use crate::protocol::{
read_message_type, read_payload, write_typed_message, BlobRequestPayload, BlobResponsePayload,
MessageType, ProfileUpdatePayload,
PullSyncRequestPayload, PullSyncResponsePayload, RefuseRedirectPayload,
RefuseRedirectPayload,
ALPN,
};
use crate::storage::StoragePool;
use crate::types::{
DeviceProfile, DeviceRole, NodeId, PeerSlotKind, Post, PostId,
DeviceProfile, DeviceRole, NodeId, MeshSlot, Post, PostId,
PostVisibility, PublicProfile, SessionReachMethod, WormResult,
};
@ -734,13 +733,16 @@ impl Network {
recv: iroh::endpoint::RecvStream,
is_mesh: &AtomicBool,
) -> bool {
// Try to allocate a slot
let accepted = conn_handle.accept_connection(
// Try to allocate a slot. `None` = both the mesh pool and the temp
// referral band are full.
let slot = conn_handle.accept_connection(
conn.clone(), remote_node_id, Some(remote_sock),
).await;
info!(peer = hex::encode(remote_node_id), accepted, "try_mesh_upgrade: accept_connection result");
if !accepted {
info!(peer = hex::encode(remote_node_id), accepted = slot.is_some(), "try_mesh_upgrade: accept_connection result");
let slot = match slot {
Some(s) => s,
None => {
let redirect = conn_handle.pick_random_redirect_peer(&remote_node_id).await;
let payload = RefuseRedirectPayload {
reason: "slots full".to_string(),
@ -752,13 +754,19 @@ impl Network {
debug!(peer = hex::encode(remote_node_id), "Refused mesh connection (slots full)");
conn_handle.add_session(remote_node_id, conn.clone(), crate::types::SessionReachMethod::Direct, Some(remote_sock)).await;
conn_handle.log_activity(ActivityLevel::Info, ActivityCategory::Connection, format!("Refused {} mesh (slots full), added session", &hex::encode(remote_node_id)[..8]), Some(remote_node_id));
return false;
}
return false;
}
};
{
let s = storage.get().await;
// bugs-fixed #2: address always persisted, temp slots included.
let _ = s.upsert_peer(&remote_node_id, &[remote_sock], None);
let _ = s.add_mesh_peer(&remote_node_id, PeerSlotKind::Local, 0);
// But mesh_peers ONLY for a real mesh slot — that omission is what
// keeps temp referral peers out of our uniques announce.
if slot.is_mesh() {
let _ = s.add_mesh_peer(&remote_node_id);
}
if s.has_social_route(&remote_node_id).unwrap_or(false) {
let _ = s.touch_social_route_connect(
&remote_node_id,
@ -774,9 +782,10 @@ impl Network {
let our_nat_type = conn_handle.nat_type().await;
let our_http_capable = conn_handle.is_http_capable();
let our_http_addr = conn_handle.http_addr();
match initial_exchange_accept(storage, &our_node_id, send, recv, remote_node_id, anchor_addr, Some(remote_sock), our_nat_type, our_http_capable, our_http_addr, conn_handle.device_role(), None, false).await {
let our_depth = conn_handle.max_bounce();
match initial_exchange_accept(storage, &our_node_id, send, recv, remote_node_id, anchor_addr, Some(remote_sock), our_nat_type, our_http_capable, our_http_addr, conn_handle.device_role(), None, false, our_depth, slot.is_mesh()).await {
Ok(()) => {
info!(peer = hex::encode(remote_node_id), "Initial exchange complete (upgraded to mesh)");
info!(peer = hex::encode(remote_node_id), slot = %slot, "Initial exchange complete (upgraded to mesh)");
conn_handle.log_activity(ActivityLevel::Info, ActivityCategory::Connection, format!("Upgraded {} to mesh", &hex::encode(remote_node_id)[..8]), Some(remote_node_id));
}
Err(e) => {
@ -819,12 +828,15 @@ impl Network {
let conn = ConnectionManager::connect_to_unlocked(&self.endpoint, addr).await?;
// Register the established connection
self.conn_handle.register_connection(peer_id, conn.clone(), addrs, PeerSlotKind::Local).await;
let slot = match self.conn_handle.register_connection(peer_id, conn.clone(), addrs).await {
Some(s) => s,
None => anyhow::bail!("no slot available (mesh + temp full)"),
};
let anchor_addr = self.conn_handle.build_anchor_advertised_addr().await;
let our_nat_type = self.conn_handle.nat_type().await;
// Initial exchange WITHOUT holding conn_mgr lock
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, peer_id, anchor_addr, our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None).await? {
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, peer_id, anchor_addr, our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None, self.conn_handle.max_bounce(), slot.is_mesh()).await? {
ExchangeResult::Accepted { duplicate_active } => {
if duplicate_active {
self.duplicate_detected.store(true, std::sync::atomic::Ordering::Relaxed);
@ -870,27 +882,38 @@ impl Network {
}
}
/// Pull from all connected peers.
pub async fn pull_from_all(&self) -> anyhow::Result<PullStats> {
/// TRANSITIONAL content sync from every connected peer, plus the
/// engagement fetch that Iteration A's registry and greetings ride on.
///
/// Routes through the actor path (`content_sync_unlocked`), NOT a second
/// local implementation: the old `Network::pull_from_peer` stored posts
/// with `store_post_with_visibility` instead of `control::receive_post`, so
/// it silently dropped `intent` (control posts — deletes, revocations, key
/// distributions — were never processed) and never called
/// `touch_file_holder`, which is why posts received on the startup pull
/// registered no CDN holder.
pub async fn content_sync_all(&self) -> anyhow::Result<PullStats> {
let peers = self.conn_handle.connected_peers().await;
let mut total_posts = 0;
let mut success = 0;
for peer_id in peers {
// Uses Network::pull_from_peer which doesn't hold conn_mgr lock during I/O
let result = self.pull_from_peer(&peer_id).await;
let result = self.conn_handle.content_sync_from_peer(&peer_id).await;
match result {
Ok(stats) => {
total_posts += stats.posts_received;
success += 1;
// Also fetch engagement data
// Engagement fetch is what carries Iteration A's registry
// entries and greeting comments (BlobHeaderRequest/Response
// + BlobHeaderDiff). It MUST stay driven on a timer — this
// is the pull side of that safety net.
let _ = self.conn_handle.fetch_engagement_from_peer(&peer_id).await;
if stats.posts_received > 0 {
info!(
peer = hex::encode(peer_id),
posts = stats.posts_received,
"Pulled posts"
"Content sync: received posts"
);
}
}
@ -898,7 +921,7 @@ impl Network {
debug!(
peer = hex::encode(peer_id),
error = %e,
"Pull failed"
"Content sync failed"
);
}
}
@ -910,82 +933,85 @@ impl Network {
})
}
/// Broadcast routing diff to all connected peers.
/// Uses ConnHandle to get diff data, then sends outside the lock.
pub async fn broadcast_diff(&self) -> anyhow::Result<usize> {
use crate::protocol::{NodeListUpdatePayload, write_typed_message, MessageType};
let snapshot = self.conn_handle.get_diff_data().await;
if snapshot.n1_added.is_empty() && snapshot.n1_removed.is_empty()
&& snapshot.n2_added.is_empty() && snapshot.n2_removed.is_empty()
{
return Ok(0);
/// Run the uniques-index exchange (0x40/0x41) against every MESH peer.
///
/// This is what a "pull" means in v0.8: an exchange of the IDs each side
/// has a live path to. Symmetric, so one round trip refreshes both indexes.
/// Temp referral slots are skipped — they carry no knowledge.
pub async fn uniques_pull_all(&self) -> usize {
let conns = self.conn_handle.get_connection_map().await;
let mut exchanged = 0;
for (peer_id, conn, slot, _) in conns {
if !slot.is_mesh() {
continue;
}
let fut = crate::connection::ConnectionManager::uniques_pull_unlocked(
&self.conn_mgr,
conn,
&peer_id,
);
match tokio::time::timeout(std::time::Duration::from_secs(20), fut).await {
Ok(Ok(n)) => {
exchanged += 1;
if n > 0 {
debug!(peer = hex::encode(peer_id), rows = n, "Uniques pull: index updated");
}
}
Ok(Err(e)) => debug!(peer = hex::encode(peer_id), error = %e, "Uniques pull failed"),
Err(_) => debug!(peer = hex::encode(peer_id), "Uniques pull timed out"),
}
}
exchanged
}
let payload = NodeListUpdatePayload {
seq: snapshot.diff_seq,
n1_added: snapshot.n1_added,
n1_removed: snapshot.n1_removed,
n2_added: snapshot.n2_added,
n2_removed: snapshot.n2_removed,
/// Broadcast our uniques announce to every MESH peer.
///
/// Temporary referral slots are excluded — they carry no knowledge
/// exchange. Peers that advertised `depth < 4` get the terminal pool
/// stripped before send (the biggest saving on a cellular link).
///
/// The announce is always a full snapshot; `snapshot.unchanged` is what
/// stands in for incremental diffing, so an idle mesh sends nothing.
pub async fn broadcast_uniques(&self) -> anyhow::Result<usize> {
use crate::protocol::{write_typed_message, MessageType, UniquesSlice};
let snapshot = self.conn_handle.get_announce_data().await;
let payload = match snapshot.payload {
Some(p) if !snapshot.unchanged => p,
_ => return Ok(0),
};
let mut sent = 0;
for (peer_id, conn) in &snapshot.connections {
for (peer_id, conn, peer_depth) in &snapshot.connections {
let per_peer = crate::protocol::UniquesAnnouncePayload {
seq: payload.seq,
full: payload.full,
fwd: payload.fwd.clone(),
term: if *peer_depth >= 4 { payload.term.clone() } else { UniquesSlice::default() },
depth: payload.depth,
};
let result = async {
let mut send = conn.open_uni().await?;
write_typed_message(&mut send, MessageType::NodeListUpdate, &payload).await?;
write_typed_message(&mut send, MessageType::UniquesAnnounce, &per_peer).await?;
send.finish()?;
anyhow::Ok(())
}.await;
if result.is_ok() {
sent += 1;
} else {
debug!(peer = hex::encode(peer_id), "Failed to send routing diff");
debug!(peer = hex::encode(peer_id), "Failed to send uniques announce");
}
}
Ok(sent)
}
/// Broadcast full N1/N2 state to all mesh peers (periodic catch-up for missed diffs).
/// Periodic catch-up: force the next announce even if the pools are
/// unchanged, so a peer that missed one resynchronises.
/// (The old `broadcast_full_state` is subsumed — every announce is already
/// a full snapshot.)
pub async fn broadcast_full_state(&self) -> anyhow::Result<usize> {
use crate::protocol::{NodeListUpdatePayload, write_typed_message, MessageType};
let snapshot = self.conn_handle.get_diff_data().await;
// Build full state: all current N1 and N2 as "added", nothing removed
let all_n1 = self.conn_handle.connected_peers().await;
let all_n2 = {
let storage = self.storage.get().await;
storage.build_n2_share().unwrap_or_default()
};
if all_n1.is_empty() && all_n2.is_empty() {
return Ok(0);
}
let payload = NodeListUpdatePayload {
seq: snapshot.diff_seq,
n1_added: all_n1,
n1_removed: vec![],
n2_added: all_n2,
n2_removed: vec![],
};
let mut sent = 0;
for (_peer_id, conn) in &snapshot.connections {
let result = async {
let mut send = conn.open_uni().await?;
write_typed_message(&mut send, MessageType::NodeListUpdate, &payload).await?;
send.finish()?;
anyhow::Ok(())
}.await;
if result.is_ok() {
sent += 1;
}
}
Ok(sent)
self.conn_handle.force_announce().await;
self.broadcast_uniques().await
}
/// Push a profile update to all audience members (ephemeral-capable).
@ -1191,7 +1217,7 @@ impl Network {
}
/// Get connection info for display.
pub async fn connection_info(&self) -> Vec<(NodeId, PeerSlotKind, u64)> {
pub async fn connection_info(&self) -> Vec<(NodeId, MeshSlot, u64)> {
self.conn_handle.connection_info().await
}
@ -1226,11 +1252,15 @@ impl Network {
}
match self.connect_to_peer(peer_id, addr.clone()).await {
Ok(()) => Ok(()),
Ok(()) => {
self.promote_proven_anchor(peer_id, &addr).await;
Ok(())
}
Err(e) if e.to_string().contains("mesh refused") => {
// Anchor refused mesh — reconnect as session for registration
let conn = ConnectionManager::connect_to_unlocked(&self.endpoint, addr).await?;
let conn = ConnectionManager::connect_to_unlocked(&self.endpoint, addr.clone()).await?;
self.conn_handle.add_session(peer_id, conn, crate::types::SessionReachMethod::Direct, None).await;
self.promote_proven_anchor(peer_id, &addr).await;
self.conn_handle.log_activity(
ActivityLevel::Info,
ActivityCategory::Anchor,
@ -1243,6 +1273,28 @@ impl Network {
}
}
/// An anchor we have COMPLETED a QUIC handshake to. iroh has proven the
/// node ID owns this path, so — and only so — the entry may enter the
/// authoritative address state: `peers.is_anchor` and the `known_anchors`
/// bootstrap cache.
///
/// This is the counterpart to removing the announce-path mirror. Hearsay
/// anchor claims live in `reachable` only (mineable via `list_pool_anchors`,
/// which is what "the pools ARE the anchor directory" means); proven ones
/// live here, which is also what makes `known_anchors.last_seen_ms`
/// ordering meaningful again — and keeps the DNS-refreshed bootstrap anchor
/// from being evicted by a flood of gossip-sourced entries.
async fn promote_proven_anchor(&self, peer_id: NodeId, addr: &iroh::EndpointAddr) {
let socks: Vec<std::net::SocketAddr> = addr.ip_addrs().copied().collect();
if socks.is_empty() {
return;
}
let storage = self.storage.get().await;
let _ = storage.upsert_peer(&peer_id, &socks, None);
let _ = storage.set_peer_anchor(&peer_id, true);
let _ = storage.upsert_known_anchor(&peer_id, &socks);
}
/// Register an already-established QUIC connection as a mesh peer.
/// Sends InitialExchange first — if the remote accepts (responds with
/// InitialExchange), registers as mesh and spawns the stream loop.
@ -1259,13 +1311,15 @@ impl Network {
let anchor_addr = self.conn_handle.build_anchor_advertised_addr().await;
let our_nat_type = self.conn_handle.nat_type().await;
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, peer_id, anchor_addr, our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None).await {
// Decide the slot BEFORE the exchange: the slot class is what says
// whether this handshake carries a uniques announce.
// bugs-fixed #3: a connection is registered before anything can drop it.
let slot = match self.conn_handle.register_connection(peer_id, conn.clone(), vec![]).await {
Some(s) => s,
None => anyhow::bail!("no slot available (mesh + temp full)"),
};
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, peer_id, anchor_addr, our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None, self.conn_handle.max_bounce(), slot.is_mesh()).await {
Ok(ExchangeResult::Accepted { .. }) => {
self.conn_handle.register_connection(peer_id, conn.clone(), vec![], PeerSlotKind::Local).await;
{
let s = self.storage.get().await;
let _ = s.add_mesh_peer(&peer_id, PeerSlotKind::Local, 0);
}
// Spawn the per-connection stream loop
let conn_data = self.conn_handle.get_connection_map().await;
@ -1280,10 +1334,16 @@ impl Network {
Ok(ExchangeResult::Refused { redirect }) => {
let redir_info = redirect.as_ref().map(|r| r.n.clone());
info!(peer = hex::encode(peer_id), redirect = ?redir_info, "Mesh refused after hole punch, keeping as session");
self.conn_handle.disconnect_peer(&peer_id).await;
self.conn_handle.add_session(peer_id, conn, crate::types::SessionReachMethod::HolePunch, None).await;
anyhow::bail!("mesh refused: slots full");
}
Err(e) => Err(e),
Err(e) => {
// Roll back the slot we reserved — otherwise a failed handshake
// holds a mesh slot until the zombie sweep.
self.conn_handle.disconnect_peer(&peer_id).await;
Err(e)
}
}
}
@ -1344,7 +1404,9 @@ impl Network {
None,
);
if remaining == 0 {
// Recovery threshold is mesh < 2 (design.html §lifecycle), not zero:
// a single remaining peer is a partition waiting to happen.
if remaining < 2 {
self.conn_handle.notify_recovery();
} else {
self.conn_handle.notify_growth();
@ -1364,7 +1426,7 @@ impl Network {
for peer_id in &newly_connected {
let conn = self.conn_handle.get_connection(peer_id).await;
if let Some(conn) = conn {
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, *peer_id, anchor_addr.clone(), our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None).await {
match initial_exchange_connect(&self.storage, &self.our_node_id, &conn, *peer_id, anchor_addr.clone(), our_nat_type, self.is_http_capable(), self.http_addr(), Some(self.device_role), None, self.conn_handle.max_bounce(), true).await {
Ok(ExchangeResult::Accepted { .. }) => {}
Ok(ExchangeResult::Refused { redirect }) => {
debug!(peer = hex::encode(peer_id), "Auto-connect refused, disconnecting");
@ -1444,10 +1506,10 @@ impl Network {
loop {
// Check slots + pick candidate via ConnHandle (no lock contention)
let available = self.conn_handle.available_local_slots().await;
let available = self.conn_handle.available_mesh_slots().await;
if available == 0 {
debug!("Growth loop: local slots full");
self.log_activity(ActivityLevel::Info, ActivityCategory::Growth, "Local slots full".into(), None);
debug!("Growth loop: mesh slots full");
self.log_activity(ActivityLevel::Info, ActivityCategory::Growth, "Mesh slots full".into(), None);
break;
}
let candidate = {
@ -1476,25 +1538,39 @@ impl Network {
self.log_activity(ActivityLevel::Info, ActivityCategory::Growth, format!("Trying candidate {} (score {:.1})", &hex::encode(candidate_id)[..8], score), Some(candidate_id));
// Resolve address via ConnHandle (no lock during I/O)
let asked_a_reporter;
let addr_str = {
let local_addr = self.conn_handle.resolve_peer_addr_local(&candidate_id).await;
if let Some(endpoint_addr) = local_addr {
asked_a_reporter = true;
endpoint_addr.ip_addrs().next().map(|a| a.to_string())
} else {
// Network resolution: get reporter connections, resolve outside lock
// Network resolution: get reporter connections, resolve
// outside lock. The horizon here must match the horizon
// the CANDIDATE SCORER uses (2..=4, same as
// `ConnectionManager::resolve_address`) — when it did
// not, the growth loop scored N4 candidates it then had
// no reporter to ask about, and burned a
// `consecutive_failures` slot plus an UNREACHABLE_EXPIRY
// suppression on every one of them.
let reporters_and_conns = {
let storage = self.storage.get().await;
let n2 = storage.find_in_n2(&candidate_id).unwrap_or_default();
let n3 = storage.find_in_n3(&candidate_id).unwrap_or_default();
let mut reporter_set: std::collections::HashSet<NodeId> =
std::collections::HashSet::new();
for bounce in 2..=4u8 {
for r in storage.find_reporters_at(&candidate_id, bounce).unwrap_or_default() {
reporter_set.insert(r);
}
}
drop(storage);
let conn_map = self.conn_handle.get_connection_map().await;
let reporter_set: std::collections::HashSet<NodeId> = n2.into_iter().chain(n3).collect();
conn_map.into_iter()
.filter(|(nid, _, _, _)| reporter_set.contains(nid))
.map(|(_, conn, _, _)| conn)
.collect::<Vec<_>>()
};
asked_a_reporter = !reporters_and_conns.is_empty();
let mut resolved = None;
for conn in reporters_and_conns {
let result: anyhow::Result<Option<String>> = async {
@ -1521,10 +1597,19 @@ impl Network {
None => {
debug!(
peer = hex::encode(candidate_id),
"Growth loop: no address, marking unreachable"
asked_a_reporter,
"Growth loop: no address"
);
self.log_activity(ActivityLevel::Warn, ActivityCategory::Growth, format!("No address for {}", &hex::encode(candidate_id)[..8]), Some(candidate_id));
self.conn_handle.mark_unreachable(&candidate_id);
// Only a candidate we could actually ask about, and
// that came back with nothing, has earned an
// UNREACHABLE_EXPIRY suppression. Failing for want of a
// connected reporter says nothing about the candidate —
// suppressing it there would hide it even after a
// shallower reporter shows up.
if asked_a_reporter {
self.conn_handle.mark_unreachable(&candidate_id);
}
consecutive_failures += 1;
if consecutive_failures >= 3 {
debug!("Growth loop: 3 consecutive failures, backing off");
@ -1560,7 +1645,7 @@ impl Network {
consecutive_failures = 0;
// Broadcast diff so peers learn about our new connection
let _ = self.broadcast_diff().await;
let _ = self.broadcast_uniques().await;
// Brief pause to let InitialExchange update N2/N3 before next pick
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
@ -1618,7 +1703,7 @@ impl Network {
if introduced {
consecutive_failures = 0;
let _ = self.broadcast_diff().await;
let _ = self.broadcast_uniques().await;
tokio::time::sleep(std::time::Duration::from_millis(500)).await;
} else {
self.conn_handle.mark_unreachable(&candidate_id);
@ -1635,64 +1720,6 @@ impl Network {
}
}
/// Pull posts from a peer (persistent if available, ephemeral otherwise).
pub async fn pull_from_peer(&self, peer_id: &NodeId) -> anyhow::Result<PullStats> {
let conn = self.get_connection(peer_id).await?;
let (our_follows, follows_sync, our_personas) = {
let storage = self.storage.get().await;
(
storage.list_follows()?,
storage.get_follows_with_last_sync().unwrap_or_default(),
storage.list_posting_identities().unwrap_or_default(),
)
};
// Merged pull: include every posting identity we hold so DMs addressed
// to any of our personas match on recipient. Our network NodeId is
// never an author nor a wrapped_key recipient — including it would
// never match and would leak the network↔posting boundary.
let mut query_list = our_follows;
for pi in &our_personas {
if !query_list.contains(&pi.node_id) {
query_list.push(pi.node_id);
}
}
let (mut send, mut recv) = conn.open_bi().await?;
write_typed_message(
&mut send,
MessageType::PullSyncRequest,
&PullSyncRequestPayload {
follows: query_list,
since_ms: follows_sync,
},
)
.await?;
send.finish()?;
let msg_type = read_message_type(&mut recv).await?;
if msg_type != MessageType::PullSyncResponse {
anyhow::bail!("expected PullSyncResponse, got {:?}", msg_type);
}
let response: PullSyncResponsePayload = read_payload(&mut recv, 64 * 1024 * 1024).await?;
let now_ms = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_millis() as u64;
let storage = self.storage.get().await;
let mut posts_received = 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)? {
posts_received += 1;
}
// Protocol v4: update last_sync_ms for the author
let _ = storage.update_follow_last_sync(&sp.post.author, now_ms);
}
}
Ok(PullStats {
peers_pulled: 1,
posts_received,
})
}
/// Send a uni-stream message. Uses persistent connection if available, ephemeral otherwise.
pub async fn send_to_peer_uni<T: Serialize>(
&self,
@ -1940,41 +1967,146 @@ impl Network {
Some(addr)
}
// ---- Anchor referral delegation ----
// ---- Anchor convection delegation (design.html §anchors) ----
/// Request peer referrals from an anchor peer (mesh or session).
/// Does NOT hold conn_mgr lock during network I/O.
pub async fn request_anchor_referrals(
/// Ask an anchor for peers. The request itself enrols us in that anchor's
/// rolling window — there is no registration message; 0xC0 is retired.
///
/// Does NOT hold the conn_mgr lock during network I/O. A refusal comes back
/// as a single small message, so the `refused` case costs one round trip,
/// not a timeout.
pub async fn request_convection(
&self,
anchor: &NodeId,
) -> anyhow::Result<Vec<crate::protocol::AnchorReferral>> {
class: crate::protocol::ConvectionClass,
) -> anyhow::Result<crate::protocol::ConvectionResponsePayload> {
let conn = self.conn_handle.get_any_connection(anchor).await
.ok_or_else(|| anyhow::anyhow!("anchor peer not connected (mesh or session)"))?;
// Self-reported addresses, best first: the stable advertised address
// (if we are an anchor ourselves), then the UPnP external mapping, then
// whatever the endpoint knows. The anchor prefers what it OBSERVES over
// all of these; these only help when it cannot observe us.
let endpoint = self.conn_handle.endpoint().await;
let our_addrs: Vec<String> = endpoint.addr().ip_addrs()
let mut our_addrs: Vec<String> = endpoint.addr().ip_addrs()
.filter(|s| is_publicly_routable(s))
.map(|s| s.to_string())
.collect();
if let Some(ext) = self.conn_handle.upnp_external_addr().await {
let ext_str = ext.to_string();
if !our_addrs.contains(&ext_str) {
our_addrs.insert(0, ext_str);
}
}
if let Some(anchor_addr) = self.conn_handle.build_anchor_advertised_addr().await {
if !our_addrs.contains(&anchor_addr) {
our_addrs.insert(0, anchor_addr);
}
}
let request = crate::protocol::AnchorReferralRequestPayload {
let request = crate::protocol::ConvectionRequestPayload {
requester: self.our_node_id,
requester_addresses: our_addrs,
class,
};
let (mut send, mut recv) = conn.open_bi().await?;
crate::protocol::write_typed_message(&mut send, crate::protocol::MessageType::AnchorReferralRequest, &request).await?;
crate::protocol::write_typed_message(&mut send, crate::protocol::MessageType::ConvectionRequest, &request).await?;
send.finish()?;
let msg_type = crate::protocol::read_message_type(&mut recv).await?;
if msg_type != crate::protocol::MessageType::AnchorReferralResponse {
anyhow::bail!("expected AnchorReferralResponse, got {:?}", msg_type);
if msg_type != crate::protocol::MessageType::ConvectionResponse {
anyhow::bail!("expected ConvectionResponse, got {:?}", msg_type);
}
let response: crate::protocol::AnchorReferralResponsePayload = crate::protocol::read_payload(&mut recv, 4096).await?;
let response: crate::protocol::ConvectionResponsePayload =
crate::protocol::read_payload(&mut recv, 4096).await?;
// Touch session to prevent idle reaping
self.conn_handle.touch_session_if_exists(anchor);
Ok(response.referrals)
// Adaptive feedback (round 5): a refusal shifts weight toward mesh
// introductions and other anchors; a served request drifts it back.
if response.refused {
debug!(anchor = hex::encode(anchor), "Convection: refused (top-up, no capacity)");
self.conn_handle.record_convection_refusal(anchor);
} else {
self.conn_handle.record_convection_success(anchor);
}
// The NAT filter probe used to ride the register message. It has to
// ride something, and this is now the only message we send an anchor.
if self.conn_handle.nat_filtering().await == crate::types::NatFiltering::Unknown {
if let Err(e) = self.request_nat_filter_probe(anchor).await {
debug!(error = %e, "NAT filter probe request failed");
}
}
Ok(response)
}
/// Act on one convection response: connect to each referral, landing it in
/// whatever slot class is free (mesh first, then the temp referral band
/// above the cap). Returns how many connections were established.
///
/// `introduced` referrals skip the extra introduction round trip — the
/// anchor already pushed a RelayIntroduce toward them naming us, so they
/// are punching outward while we dial.
pub async fn act_on_convection(
&self,
anchor: &NodeId,
response: &crate::protocol::ConvectionResponsePayload,
) -> usize {
let mut connected = 0;
for referral in &response.referrals {
if referral.node_id == self.our_node_id {
continue;
}
if self.conn_handle.is_connected(&referral.node_id).await {
continue;
}
let mut ok = false;
// bugs-fixed #8 (receive side): a referral address is a third
// party's claim. Dialling `127.0.0.1:22` or `192.168.1.1:443`
// because an anchor said so is attacker-directed internal probing
// from our own host, and every such connect() enters iroh's
// per-endpoint path store permanently (v0.7.3 EDM lesson).
let usable: Option<&String> = referral.addresses.iter().find(|a| {
a.parse::<std::net::SocketAddr>()
.map(|sa| crate::connection::convection_addr_ok(&sa))
.unwrap_or(false)
});
if let Some(addr_str) = usable {
let connect_str = format!("{}@{}", hex::encode(referral.node_id), addr_str);
if let Ok((rid, raddr)) = crate::parse_connect_string(&connect_str) {
let fut = self.connect_to_peer(rid, raddr);
match tokio::time::timeout(std::time::Duration::from_secs(15), fut).await {
Ok(Ok(())) => {
info!(peer = hex::encode(rid), "Convection: connected to referred peer");
ok = true;
}
Ok(Err(e)) => debug!(error = %e, peer = hex::encode(rid), "Convection: direct connect failed"),
Err(_) => debug!(peer = hex::encode(rid), "Convection: direct connect timed out"),
}
}
}
if !ok && !referral.introduced {
// No coordinated introduction was possible — ask for one.
match self.connect_via_introduction_as_mesh(referral.node_id, *anchor).await {
Ok(()) => {
info!(peer = hex::encode(referral.node_id), "Convection: connected via introduction");
ok = true;
}
Err(e) => debug!(error = %e, peer = hex::encode(referral.node_id), "Convection: introduction failed"),
}
}
if ok {
connected += 1;
}
}
if connected > 0 {
self.notify_growth().await;
}
connected
}
/// Request NAT filter probe from an anchor (determines address-restricted vs port-restricted).
@ -2028,55 +2160,6 @@ impl Network {
Ok(())
}
/// Register our address with an anchor peer (mesh or session).
/// Also runs NAT filter probe if filtering is still Unknown.
/// Does NOT hold conn_mgr lock during network I/O.
pub async fn send_anchor_register(&self, anchor: &NodeId) -> anyhow::Result<()> {
let conn = self.conn_handle.get_any_connection(anchor).await
.ok_or_else(|| anyhow::anyhow!("anchor peer not connected (mesh or session)"))?;
let endpoint = self.conn_handle.endpoint().await;
let mut our_addrs: Vec<String> = endpoint.addr().ip_addrs()
.map(|s| s.to_string())
.collect();
// Prepend UPnP external address (most useful for remote peers)
if let Some(ext) = self.conn_handle.upnp_external_addr().await {
let ext_str = ext.to_string();
if !our_addrs.contains(&ext_str) {
our_addrs.insert(0, ext_str);
}
}
// Prepend stable anchor advertised address
if let Some(anchor_addr) = self.conn_handle.build_anchor_advertised_addr().await {
if !our_addrs.contains(&anchor_addr) {
our_addrs.insert(0, anchor_addr);
}
}
let payload = crate::protocol::AnchorRegisterPayload {
node_id: self.our_node_id,
addresses: our_addrs,
};
let mut send = conn.open_uni().await?;
crate::protocol::write_typed_message(&mut send, crate::protocol::MessageType::AnchorRegister, &payload).await?;
send.finish()?;
// Touch session to prevent idle reaping
self.conn_handle.touch_session_if_exists(anchor);
debug!(anchor = hex::encode(anchor), "Registered with anchor");
// Also run NAT filter probe if filtering is still Unknown
let needs_probe = self.conn_handle.nat_filtering().await == crate::types::NatFiltering::Unknown;
if needs_probe {
if let Err(e) = self.request_nat_filter_probe(anchor).await {
debug!(error = %e, "NAT filter probe request failed");
}
}
Ok(())
}
/// Check if a peer is a known anchor.
pub async fn is_anchor_peer(&self, node_id: &NodeId) -> bool {
let storage = self.storage.get().await;

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

File diff suppressed because it is too large Load diff

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

@ -27,13 +27,25 @@ pub struct SyncPost {
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum MessageType {
NodeListUpdate = 0x01,
UniquesAnnounce = 0x01,
InitialExchange = 0x02,
AddressRequest = 0x03,
AddressResponse = 0x04,
RefuseRedirect = 0x05,
PullSyncRequest = 0x40,
PullSyncResponse = 0x41,
/// The uniques-index exchange (design.html §sync). A pull is no longer a
/// post transfer: it is "if you want these IDs, talk to me and I'll help
/// you find them". Symmetric — request carries our pools, response carries
/// theirs, one round trip updates both sides.
UniquesPullRequest = 0x40,
UniquesPullResponse = 0x41,
/// TRANSITIONAL (Iteration D folds this into the update-cadence scheduler +
/// CDN replication). Content still has to move somewhere while
/// PostFetchRequest (0xD4) serves Public posts only: DMs, group posts,
/// Friends and FoFClosed posts have no other carrier. Splitting it off
/// 0x40/0x41 keeps §sync honest — a *pull* is not a post transfer, because
/// post transfer is not a pull.
ContentSyncRequest = 0x46,
ContentSyncResponse = 0x47,
ProfileUpdate = 0x50,
WormQuery = 0x60,
WormResponse = 0x61,
@ -49,9 +61,11 @@ pub enum MessageType {
RelayIntroduceResult = 0xB1,
SessionRelay = 0xB2,
CircleProfileUpdate = 0xB4,
AnchorRegister = 0xC0,
AnchorReferralRequest = 0xC1,
AnchorReferralResponse = 0xC2,
// 0xC0 retired in v0.8 — anchors keep a rolling window of recent callers
// fed by the request itself; there is no registration message and no
// registration database.
ConvectionRequest = 0xC1,
ConvectionResponse = 0xC2,
AnchorProbeRequest = 0xC3,
AnchorProbeResult = 0xC4,
PortScanHeartbeat = 0xC5,
@ -73,13 +87,15 @@ pub enum MessageType {
impl MessageType {
pub fn from_byte(b: u8) -> Option<Self> {
match b {
0x01 => Some(Self::NodeListUpdate),
0x01 => Some(Self::UniquesAnnounce),
0x02 => Some(Self::InitialExchange),
0x03 => Some(Self::AddressRequest),
0x04 => Some(Self::AddressResponse),
0x05 => Some(Self::RefuseRedirect),
0x40 => Some(Self::PullSyncRequest),
0x41 => Some(Self::PullSyncResponse),
0x40 => Some(Self::UniquesPullRequest),
0x41 => Some(Self::UniquesPullResponse),
0x46 => Some(Self::ContentSyncRequest),
0x47 => Some(Self::ContentSyncResponse),
0x50 => Some(Self::ProfileUpdate),
0x60 => Some(Self::WormQuery),
0x61 => Some(Self::WormResponse),
@ -94,9 +110,8 @@ impl MessageType {
0xB1 => Some(Self::RelayIntroduceResult),
0xB2 => Some(Self::SessionRelay),
0xB4 => Some(Self::CircleProfileUpdate),
0xC0 => Some(Self::AnchorRegister),
0xC1 => Some(Self::AnchorReferralRequest),
0xC2 => Some(Self::AnchorReferralResponse),
0xC1 => Some(Self::ConvectionRequest),
0xC2 => Some(Self::ConvectionResponse),
0xC3 => Some(Self::AnchorProbeRequest),
0xC4 => Some(Self::AnchorProbeResult),
0xC5 => Some(Self::PortScanHeartbeat),
@ -124,13 +139,152 @@ impl MessageType {
// --- Payload structs ---
/// Initial exchange: N1/N2 node lists + profile + deletes + post_ids + peer addresses
/// A packed set of 32-byte IDs: base64 of the raw concatenated bytes.
///
/// `NodeId = [u8; 32]` serialised through serde_json becomes a JSON array of 32
/// decimal numbers (~100-130 bytes/ID — a 3-4x blowup). The uniques pools can
/// carry thousands of IDs, so they ride packed instead.
pub type PackedIds = String;
/// Encode a set of IDs into a packed base64 blob.
pub fn pack_ids(ids: &[NodeId]) -> PackedIds {
use base64::Engine;
let mut raw = Vec::with_capacity(ids.len() * 32);
for id in ids {
raw.extend_from_slice(id);
}
base64::engine::general_purpose::STANDARD.encode(raw)
}
/// Decode a packed base64 blob back into IDs. Trailing partial IDs are dropped.
pub fn unpack_ids(packed: &str) -> Vec<NodeId> {
use base64::Engine;
let raw = match base64::engine::general_purpose::STANDARD.decode(packed) {
Ok(r) => r,
Err(_) => return Vec::new(),
};
raw.chunks_exact(32)
.map(|c| {
let mut id = [0u8; 32];
id.copy_from_slice(c);
id
})
.collect()
}
/// How many 32-byte IDs a packed base64 blob holds, WITHOUT decoding it.
///
/// Standard base64 emits 4 chars per 3 input bytes (padded), so decoded length
/// is `chars / 4 * 3` minus padding; integer division by 32 absorbs the padding
/// slack for any non-empty blob. Exact for every blob `pack_ids` produces.
pub fn packed_id_count(packed: &str) -> usize {
(packed.len() / 4) * 3 / 32
}
/// The ONLY entry type in a uniques pool that carries an address.
///
/// design.html §layers: anchor entries carry addresses so the pools double as
/// the anchor directory; every other entry is a bare ID (there is no point
/// sharing an address that needs an introduction anyway). Addresses MUST be
/// filtered through `is_publicly_routable()` before they go on the wire.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnchorEntry {
/// hex node id (anchors are always node-class)
pub i: String,
/// e.g. ["1.2.3.4:4433"]
pub a: Vec<String>,
}
/// One slice of a uniques pool. Node-class and author-class IDs live in
/// separate arrays so a receiver structurally cannot mistake a persona ID for
/// a connectable node ID.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct UniquesSlice {
/// Connectable network IDs, bare.
#[serde(default)]
pub nodes: PackedIds,
/// Posting/persona/throwaway IDs, bare. CDN-resolvable only — never a
/// connect target.
#[serde(default)]
pub authors: PackedIds,
/// Address-bearing anchor entries (node-class by definition).
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub anchors: Vec<AnchorEntry>,
}
impl UniquesSlice {
pub fn is_empty(&self) -> bool {
self.nodes.is_empty() && self.authors.is_empty() && self.anchors.is_empty()
}
/// Number of IDs carried (for logging / budget checks).
///
/// `pack_ids` concatenates ALL IDs and base64-encodes the blob ONCE, so the
/// encoded length is `4 * ceil(32n/3)` characters — about 42.67 per ID, not
/// 44 (44 is the encoding of exactly ONE 32-byte ID). Counting `len()/44`
/// undercounts by ~3% at scale and by 33% for a 3-ID blob. Since this backs
/// budget checks, it is computed from the DECODED byte length instead.
pub fn len(&self) -> usize {
packed_id_count(&self.nodes) + packed_id_count(&self.authors) + self.anchors.len()
}
}
/// 0x01 — the two-pool uniques announce (design.html §layers).
///
/// POOL 1 (`fwd`) is FORWARDABLE and indexed by OUR bounce:
/// fwd[0] = N0 — ourselves (and our own anchor address, if we are an anchor)
/// fwd[1] = N1 — our own uniques: mesh peers, social directs, CDN file
/// authors, CDN file-holder peers, merged so the receiver
/// cannot tell which source an entry came from
/// fwd[2] = N2 — what our N1 reported to us
/// The receiver stores `fwd[i]` at bounce `i + 1` (their N1..N3), tagged to us
/// as reporter, and MAY re-announce it shifted one deeper.
///
/// POOL 2 (`term`) is TERMINAL: our N3, deduplicated against every fwd slice.
/// The receiver stores it as their N4 and USES it for search / address
/// resolution, but NEVER re-announces it. The structural separation — a
/// distinct field, not a flag in a merged list — is what makes "never
/// re-announce" un-forgettable at the announce builder: the builder simply
/// never reads bounce-4 rows. There is no N5 anywhere.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct UniquesAnnouncePayload {
pub seq: u64,
/// Always true in v0.8 — the announce is a full snapshot that replaces this
/// reporter's rows. (Incremental diffing over four layers with anchor
/// metadata was not worth the divergence risk; the sender skips the send
/// entirely when nothing changed, which recovers most of the saving.)
pub full: bool,
/// Pool 1, forwardable. Always length 3 (bounces 0, 1, 2).
pub fwd: Vec<UniquesSlice>,
/// Pool 2, terminal. Empty when the recipient advertised `depth < 4`.
#[serde(default)]
pub term: UniquesSlice,
/// Deepest bounce the SENDER retains (4 desktop, 3 mobile). Lets the peer
/// skip building/sending the terminal pool for us — the single biggest
/// bandwidth saving on a cellular link.
pub depth: u8,
}
impl UniquesAnnouncePayload {
pub fn empty(seq: u64, depth: u8) -> Self {
Self {
seq,
full: true,
fwd: vec![UniquesSlice::default(), UniquesSlice::default(), UniquesSlice::default()],
term: UniquesSlice::default(),
depth,
}
}
}
/// Initial exchange: uniques announce + profile + deletes + post_ids + peer addresses
#[derive(Debug, Serialize, Deserialize)]
pub struct InitialExchangePayload {
/// Our connections + social contacts NodeIds (no addresses)
pub n1_node_ids: Vec<NodeId>,
/// Our deduplicated N2 NodeIds (no addresses)
pub n2_node_ids: Vec<NodeId>,
/// Our uniques pools (full snapshot). `None` means "this connection carries
/// no knowledge" — i.e. the sender placed us in a temporary referral slot.
/// Explicit on the wire rather than a pair of empty vectors so the receiver
/// can tell "nothing to share" from "not sharing".
#[serde(default)]
pub uniques: Option<UniquesAnnouncePayload>,
/// Our profile
pub profile: Option<PublicProfile>,
/// Our delete records
@ -162,28 +316,32 @@ pub struct InitialExchangePayload {
pub duplicate_active: Option<bool>,
}
/// Incremental N1/N2 changes
/// 0x40/0x41 — the uniques-index exchange, both directions.
///
/// The requester sends its pools; the responder answers with its own. One
/// round trip refreshes both indexes, which is what makes an index *exchange*
/// rather than a fetch. Content does not ride this path.
#[derive(Debug, Serialize, Deserialize)]
pub struct NodeListUpdatePayload {
pub seq: u64,
pub n1_added: Vec<NodeId>,
pub n1_removed: Vec<NodeId>,
pub n2_added: Vec<NodeId>,
pub n2_removed: Vec<NodeId>,
pub struct UniquesPullPayload {
/// The sender's two pools. `None` = "I carry no knowledge on this link"
/// (temp referral slot), mirroring `InitialExchangePayload.uniques`.
#[serde(default)]
pub uniques: Option<UniquesAnnouncePayload>,
}
/// Pull-based post sync request
/// 0x46 — TRANSITIONAL content sync request. Identical in shape to the v0.7
/// pull request; only the opcode moved. See `MessageType::ContentSyncRequest`.
#[derive(Debug, Serialize, Deserialize)]
pub struct PullSyncRequestPayload {
pub struct ContentSyncRequestPayload {
/// Our follows (for the responder to filter)
pub follows: Vec<NodeId>,
/// Per-author last-sync timestamps (Vec of tuples for serde compat)
pub since_ms: Vec<(NodeId, u64)>,
}
/// Pull-based post sync response
/// 0x47 — TRANSITIONAL content sync response.
#[derive(Debug, Serialize, Deserialize)]
pub struct PullSyncResponsePayload {
pub struct ContentSyncResponsePayload {
pub posts: Vec<SyncPost>,
}
@ -387,33 +545,63 @@ pub struct CircleProfileUpdatePayload {
pub updated_at: u64,
}
// --- Anchor referral payloads ---
// --- Anchor convection payloads (design.html §anchors) ---
/// Node registers its address with an anchor (uni-stream)
#[derive(Debug, Serialize, Deserialize)]
pub struct AnchorRegisterPayload {
pub node_id: NodeId,
pub addresses: Vec<String>,
/// The one request-class bit (round-5 ruling).
///
/// ENTRY = bootstrap or recovery, i.e. the caller has fewer than 2 mesh
/// connections. Always served — a node that cannot get in cannot come back.
/// TOP_UP = growth with a working mesh. Served capacity-permitting and refused
/// CHEAPLY (a single response message, no timeout burned). Top-ups are the
/// convection medium — they are what keeps referral windows fresh — so they
/// are not throttled aggressively; the refusal doubles as the adaptive load
/// signal the caller feeds back into its action weights.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum ConvectionClass {
Entry,
TopUp,
}
/// Node requests peer referrals from an anchor (bi-stream request)
impl ConvectionClass {
/// Entry class is decided purely by "can this node function right now".
pub fn for_mesh_count(mesh_count: usize) -> Self {
if mesh_count < 2 { Self::Entry } else { Self::TopUp }
}
pub fn is_entry(self) -> bool {
matches!(self, Self::Entry)
}
}
/// 0xC1 — "I'm joining / I need peers". The request itself is what enrols the
/// caller in the anchor's rolling window; there is no separate registration.
#[derive(Debug, Serialize, Deserialize)]
pub struct AnchorReferralRequestPayload {
pub struct ConvectionRequestPayload {
pub requester: NodeId,
pub requester_addresses: Vec<String>,
pub class: ConvectionClass,
}
/// Anchor responds with peer referrals (bi-stream response)
/// 0xC2 — up to 2 recent prior callers, or a cheap refusal.
#[derive(Debug, Serialize, Deserialize)]
pub struct AnchorReferralResponsePayload {
pub referrals: Vec<AnchorReferral>,
pub struct ConvectionResponsePayload {
pub referrals: Vec<ConvectionReferral>,
/// Cheap refusal of a TOP_UP request: one message, immediately, so the
/// caller never burns a timeout. Never set for ENTRY-class requests.
#[serde(default)]
pub refused: bool,
}
/// A single peer referral from an anchor
/// A single referral out of the anchor's rolling caller window.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnchorReferral {
pub struct ConvectionReferral {
pub node_id: NodeId,
pub addresses: Vec<String>,
/// The anchor had BOTH ends live and has pushed a RelayIntroduce toward
/// this peer naming the caller as requester — so the peer is already
/// punching outward. Dial straight through instead of paying for a second
/// introduction round trip.
#[serde(default)]
pub introduced: bool,
}
// --- Anchor probe payloads ---
@ -634,13 +822,15 @@ mod tests {
#[test]
fn message_type_roundtrip() {
let types = [
MessageType::NodeListUpdate,
MessageType::UniquesAnnounce,
MessageType::InitialExchange,
MessageType::AddressRequest,
MessageType::AddressResponse,
MessageType::RefuseRedirect,
MessageType::PullSyncRequest,
MessageType::PullSyncResponse,
MessageType::UniquesPullRequest,
MessageType::UniquesPullResponse,
MessageType::ContentSyncRequest,
MessageType::ContentSyncResponse,
MessageType::ProfileUpdate,
MessageType::WormQuery,
MessageType::WormResponse,
@ -655,9 +845,8 @@ mod tests {
MessageType::RelayIntroduceResult,
MessageType::SessionRelay,
MessageType::CircleProfileUpdate,
MessageType::AnchorRegister,
MessageType::AnchorReferralRequest,
MessageType::AnchorReferralResponse,
MessageType::ConvectionRequest,
MessageType::ConvectionResponse,
MessageType::AnchorProbeRequest,
MessageType::AnchorProbeResult,
MessageType::PortScanHeartbeat,
@ -779,56 +968,105 @@ mod tests {
}
#[test]
fn anchor_register_payload_roundtrip() {
let payload = AnchorRegisterPayload {
node_id: [1u8; 32],
addresses: vec!["192.168.1.5:4433".to_string(), "10.0.0.1:4433".to_string()],
};
let json = serde_json::to_string(&payload).unwrap();
let decoded: AnchorRegisterPayload = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.node_id, [1u8; 32]);
assert_eq!(decoded.addresses.len(), 2);
assert_eq!(decoded.addresses[0], "192.168.1.5:4433");
fn convection_request_carries_the_class_bit() {
for class in [ConvectionClass::Entry, ConvectionClass::TopUp] {
let payload = ConvectionRequestPayload {
requester: [2u8; 32],
requester_addresses: vec!["10.0.0.2:4433".to_string()],
class,
};
let json = serde_json::to_string(&payload).unwrap();
let decoded: ConvectionRequestPayload = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.requester, [2u8; 32]);
assert_eq!(decoded.requester_addresses, vec!["10.0.0.2:4433"]);
assert_eq!(decoded.class, class);
}
// The class is derived purely from "can this node function right now".
assert!(ConvectionClass::for_mesh_count(0).is_entry());
assert!(ConvectionClass::for_mesh_count(1).is_entry());
assert!(!ConvectionClass::for_mesh_count(2).is_entry());
assert!(!ConvectionClass::for_mesh_count(20).is_entry());
}
#[test]
fn anchor_referral_request_payload_roundtrip() {
let payload = AnchorReferralRequestPayload {
requester: [2u8; 32],
requester_addresses: vec!["10.0.0.2:4433".to_string()],
};
let json = serde_json::to_string(&payload).unwrap();
let decoded: AnchorReferralRequestPayload = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.requester, [2u8; 32]);
assert_eq!(decoded.requester_addresses, vec!["10.0.0.2:4433"]);
}
#[test]
fn anchor_referral_response_payload_roundtrip() {
let payload = AnchorReferralResponsePayload {
fn convection_response_roundtrip_including_cheap_refusal() {
let payload = ConvectionResponsePayload {
referrals: vec![
AnchorReferral {
ConvectionReferral {
node_id: [3u8; 32],
addresses: vec!["10.0.0.3:4433".to_string()],
introduced: false,
},
AnchorReferral {
ConvectionReferral {
node_id: [4u8; 32],
addresses: vec!["10.0.0.4:4433".to_string(), "192.168.1.4:4433".to_string()],
addresses: vec!["10.0.0.4:4433".to_string()],
introduced: true,
},
],
refused: false,
};
let json = serde_json::to_string(&payload).unwrap();
let decoded: AnchorReferralResponsePayload = serde_json::from_str(&json).unwrap();
let decoded: ConvectionResponsePayload = serde_json::from_str(&json).unwrap();
assert_eq!(decoded.referrals.len(), 2);
assert_eq!(decoded.referrals[0].node_id, [3u8; 32]);
assert_eq!(decoded.referrals[0].addresses, vec!["10.0.0.3:4433"]);
assert_eq!(decoded.referrals[1].node_id, [4u8; 32]);
assert_eq!(decoded.referrals[1].addresses.len(), 2);
assert!(!decoded.referrals[0].introduced);
assert!(decoded.referrals[1].introduced);
assert!(!decoded.refused);
// Empty referrals
let empty = AnchorReferralResponsePayload { referrals: vec![] };
let json2 = serde_json::to_string(&empty).unwrap();
let decoded2: AnchorReferralResponsePayload = serde_json::from_str(&json2).unwrap();
// A refusal is one small message with no referrals — the caller reads
// it immediately instead of burning a timeout.
let refusal = ConvectionResponsePayload { referrals: vec![], refused: true };
let json2 = serde_json::to_string(&refusal).unwrap();
assert!(json2.len() < 64, "refusal must stay tiny: {}", json2);
let decoded2: ConvectionResponsePayload = serde_json::from_str(&json2).unwrap();
assert!(decoded2.refused);
assert!(decoded2.referrals.is_empty());
}
#[test]
fn uniques_pull_is_an_index_exchange_not_a_post_transfer() {
let payload = UniquesPullPayload {
uniques: Some(UniquesAnnouncePayload::empty(7, 4)),
};
let json = serde_json::to_string(&payload).unwrap();
let decoded: UniquesPullPayload = serde_json::from_str(&json).unwrap();
let u = decoded.uniques.unwrap();
assert_eq!(u.seq, 7);
assert_eq!(u.depth, 4);
assert_eq!(u.fwd.len(), 3);
// No post field exists on this payload at all — content rides 0x46/0x47.
assert!(!json.contains("posts"));
// A temp referral slot carries no knowledge in either direction.
let none = UniquesPullPayload { uniques: None };
let json2 = serde_json::to_string(&none).unwrap();
let decoded2: UniquesPullPayload = serde_json::from_str(&json2).unwrap();
assert!(decoded2.uniques.is_none());
}
/// `len()` backs budget checks, so it must be EXACT — not the old
/// `bytes / 44` approximation, which assumed one base64 encoding per ID
/// when `pack_ids` encodes the whole concatenated blob once.
#[test]
fn uniques_slice_len_counts_packed_ids_exactly() {
for n in [0usize, 1, 2, 3, 7, 100, 1000] {
let ids: Vec<crate::types::NodeId> = (0..n)
.map(|i| {
let mut id = [0u8; 32];
id[0] = (i % 251) as u8;
id[1] = (i / 251) as u8;
id
})
.collect();
let packed = pack_ids(&ids);
assert_eq!(packed_id_count(&packed), n, "packed_id_count for n={}", n);
assert_eq!(unpack_ids(&packed).len(), n);
let slice = UniquesSlice {
nodes: packed.clone(),
authors: packed.clone(),
anchors: vec![AnchorEntry { i: "aa".into(), a: vec!["1.2.3.4:1".into()] }],
};
assert_eq!(slice.len(), n * 2 + 1, "UniquesSlice::len for n={}", n);
}
}
}

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

File diff suppressed because it is too large Load diff

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

@ -649,26 +649,67 @@ impl NatProfile {
}
/// Device profile — determines connection slot budget
///
/// v0.8 (Iteration C): the Local/Wide split is gone. There is ONE mesh pool
/// (`mesh_slots`) plus a strictly-additional band of temporary referral slots
/// (`temp_referral_slots`) that live ABOVE the mesh cap and carry no knowledge
/// exchange. See design.html §connections.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DeviceProfile {
/// Desktop: 71 local + 20 wide = 91 mesh connections
/// Desktop: 20 mesh + up to 10 temp referral
Desktop,
/// Mobile: 7 local + 5 wide = 12 mesh connections
/// Mobile: 15 mesh + up to 4 temp referral
Mobile,
}
impl DeviceProfile {
pub fn local_slots(&self) -> usize {
/// The single mesh pool size (v0.8 narrow mesh, deep knowledge).
///
/// `ITSGOIN_TEST_MESH_SLOTS` overrides it so integration tests can reach
/// the temp-referral boundary with a handful of nodes instead of 21. Test
/// gate only — same pattern as `ITSGOIN_TEST_ALLOW_REGISTRY_GENESIS`.
///
/// Read ONCE into a `OnceLock`: this is called from `ConnectionManager`
/// construction and from every path that wants a fresh cap, and an env
/// lookup per call is a syscall-shaped cost on a hot path (and lets the
/// mesh cap change under a running node, which nothing expects).
pub fn mesh_slots(&self) -> usize {
static OVERRIDE: std::sync::OnceLock<Option<usize>> = std::sync::OnceLock::new();
let over = *OVERRIDE.get_or_init(|| {
std::env::var("ITSGOIN_TEST_MESH_SLOTS")
.ok()
.and_then(|v| v.parse::<usize>().ok())
.filter(|n| *n > 0)
});
if let Some(n) = over {
return n;
}
match self {
DeviceProfile::Desktop => 71,
DeviceProfile::Mobile => 7,
DeviceProfile::Desktop => 20,
DeviceProfile::Mobile => 15,
}
}
pub fn wide_slots(&self) -> usize {
/// Temporary referral slots, allocated ABOVE `mesh_slots`. These exist only
/// to facilitate introductions: they carry no uniques exchange, are never
/// persisted to `mesh_peers`, and either graduate into a freed mesh slot or
/// expire. Ruling: +4..+10 above the cap.
pub fn temp_referral_slots(&self) -> usize {
match self {
DeviceProfile::Desktop => 20,
DeviceProfile::Mobile => 5,
DeviceProfile::Desktop => 10,
DeviceProfile::Mobile => 4,
}
}
/// Deepest bounce this device stores in the `reachable` index.
/// Desktop keeps the full 4-bounce horizon; mobile caps at 3 (it drops the
/// terminal pool on receipt and advertises `depth = 3` so senders skip
/// building it). Alternative considered and NOT implemented: Bloom-compress
/// N4 on mobile (see `apply_uniques_announce` doc comment).
pub fn max_bounce(&self) -> u8 {
match self {
DeviceProfile::Desktop => 4,
DeviceProfile::Mobile => 3,
}
}
@ -708,24 +749,103 @@ impl std::fmt::Display for SessionReachMethod {
}
}
/// Slot kind for mesh connections
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PeerSlotKind {
/// Diverse local connections (Desktop: 71, Mobile: 7)
Local,
/// Bloom-sourced random distant connections (Desktop: 20, Mobile: 5)
Wide,
/// Which class of slot a live connection occupies.
///
/// v0.8: `Mesh` is the real pool (capped at `DeviceProfile::mesh_slots`) and is
/// the ONLY class that participates in the uniques announce. `TempReferral`
/// slots sit above the cap, exist purely to broker introductions, and are never
/// written to `mesh_peers` — which is what structurally keeps them out of our
/// own announcements. They graduate into a freed mesh slot or expire.
///
/// NOT on the wire — purely local accounting.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum MeshSlot {
Mesh,
TempReferral { expires_at_ms: u64 },
}
impl std::fmt::Display for PeerSlotKind {
impl MeshSlot {
pub fn is_mesh(&self) -> bool {
matches!(self, MeshSlot::Mesh)
}
pub fn is_temp(&self) -> bool {
!self.is_mesh()
}
}
impl std::fmt::Display for MeshSlot {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
PeerSlotKind::Local => write!(f, "local"),
PeerSlotKind::Wide => write!(f, "wide"),
MeshSlot::Mesh => write!(f, "mesh"),
MeshSlot::TempReferral { .. } => write!(f, "temp"),
}
}
}
/// Class of an ID carried in a uniques pool.
///
/// A `Node` ID is a network identity we may connect to / hole-punch / relay to.
/// An `Author` ID is a posting identity (persona, throwaway, greeting ID). It
/// has NO address and NO device linkage (Iteration A stripped
/// `AuthorManifest.author_addresses`), so it must NEVER enter the address
/// resolution cascade or the growth-candidate scorer — it resolves only through
/// CDN holder search. The pools keep the two classes in separate arrays so a
/// receiver structurally cannot confuse them.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum IdClass {
Node = 0,
Author = 1,
}
impl IdClass {
pub fn as_i64(self) -> i64 {
self as i64
}
pub fn from_i64(v: i64) -> Self {
if v == 1 { IdClass::Author } else { IdClass::Node }
}
}
/// One entry in the uniques index / announce pools.
///
/// `addresses` is non-empty ONLY when `is_anchor` — that is the §layers privacy
/// invariant, enforced at the store (`add_reach` blanks the column otherwise)
/// and at the wire builder (only `AnchorEntry` has an address field at all).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReachEntry {
pub id: NodeId,
pub id_class: IdClass,
pub is_anchor: bool,
pub addresses: Vec<String>,
}
impl ReachEntry {
/// A bare node-class entry (no address).
pub fn node(id: NodeId) -> Self {
Self { id, id_class: IdClass::Node, is_anchor: false, addresses: Vec::new() }
}
/// A bare author-class entry (no address, never a connect target).
pub fn author(id: NodeId) -> Self {
Self { id, id_class: IdClass::Author, is_anchor: false, addresses: Vec::new() }
}
/// An anchor entry — the only kind that carries an address.
pub fn anchor(id: NodeId, addresses: Vec<String>) -> Self {
Self { id, id_class: IdClass::Node, is_anchor: true, addresses }
}
}
/// The two announce pools, pre-wire.
///
/// `fwd` is always length 3 (our N0, N1, N2) and is FORWARDABLE — the receiver
/// stores `fwd[i]` at bounce `i + 1` and may re-announce it shifted.
/// `term` is our N3 and is TERMINAL — the receiver stores it at bounce 4, uses
/// it, and never re-announces it.
#[derive(Debug, Clone, Default)]
pub struct UniquesPools {
pub fwd: Vec<Vec<ReachEntry>>,
pub term: Vec<ReachEntry>,
}
// --- Social Routing Cache ---
/// How we last reached a social contact

2
crates/core/src/upnp.rs
View file

@ -69,7 +69,7 @@
use std::net::SocketAddr;
use std::num::NonZeroU16;
use std::time::Duration;
use tracing::{debug, info, warn};
use tracing::{debug, info};
/// An active port mapping. While this value is held, the underlying
/// `portmapper::Client` keeps the lease alive in a background task.