diff --git a/Cargo.lock b/Cargo.lock index 14e46037..ca98578e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -486,7 +486,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "c0mpute-api" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "c0mpute-proto", @@ -501,7 +501,7 @@ dependencies = [ [[package]] name = "c0mpute-bench" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "blake3", @@ -512,7 +512,7 @@ dependencies = [ [[package]] name = "c0mpute-cli" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "async-trait", @@ -549,7 +549,7 @@ dependencies = [ [[package]] name = "c0mpute-core" -version = "0.3.59" +version = "0.3.60" dependencies = [ "aes-gcm", "anyhow", @@ -603,7 +603,7 @@ dependencies = [ [[package]] name = "c0mpute-doctor" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "serde", @@ -615,7 +615,7 @@ dependencies = [ [[package]] name = "c0mpute-envelope" -version = "0.3.59" +version = "0.3.60" dependencies = [ "blake3", "bs58", @@ -628,7 +628,7 @@ dependencies = [ [[package]] name = "c0mpute-gateway" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "axum", @@ -650,7 +650,7 @@ dependencies = [ [[package]] name = "c0mpute-market" -version = "0.3.59" +version = "0.3.60" dependencies = [ "c0mpute-envelope", "serde", @@ -660,7 +660,7 @@ dependencies = [ [[package]] name = "c0mpute-net" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "async-trait", @@ -681,7 +681,7 @@ dependencies = [ [[package]] name = "c0mpute-placement" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "async-trait", @@ -702,7 +702,7 @@ dependencies = [ [[package]] name = "c0mpute-proto" -version = "0.3.59" +version = "0.3.60" dependencies = [ "blake3", "hex", @@ -714,7 +714,7 @@ dependencies = [ [[package]] name = "c0mpute-sealed" -version = "0.3.59" +version = "0.3.60" dependencies = [ "aes", "aes-gcm", @@ -736,7 +736,7 @@ dependencies = [ [[package]] name = "c0mpute-secure-chat" -version = "0.3.59" +version = "0.3.60" dependencies = [ "aes-gcm", "anyhow", @@ -759,7 +759,7 @@ dependencies = [ [[package]] name = "c0mpute-store" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "blake3", @@ -778,7 +778,7 @@ dependencies = [ [[package]] name = "c0mpute-transcode" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "c0mpute-proto", @@ -791,7 +791,7 @@ dependencies = [ [[package]] name = "c0mpute-update" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "hex", @@ -806,7 +806,7 @@ dependencies = [ [[package]] name = "c0mpute-verify" -version = "0.3.59" +version = "0.3.60" dependencies = [ "anyhow", "blake3", diff --git a/Cargo.toml b/Cargo.toml index b313a438..c59c60e2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -24,7 +24,7 @@ members = [ ] [workspace.package] -version = "0.3.59" +version = "0.3.60" edition = "2024" license = "MIT" repository = "https://github.com/profullstack/c0mpute" diff --git a/dips/0034-agent-swarms.md b/dips/0034-agent-swarms.md index 6f195efa..e4f30684 100644 --- a/dips/0034-agent-swarms.md +++ b/dips/0034-agent-swarms.md @@ -82,7 +82,9 @@ more challenges than endorsements; an owner reject always wins. A task is task is done; a job is **complete** when its final task is done. A bounty is **payable** when an owner accepted the result, or at least two -peers endorsed it and endorsements outnumber challenges. +reviewers with **standing** endorsed it and they outnumber challenges. A +reviewer has standing once someone other than itself has published +`task.paid` for one of its results in that swarm. ### Next action @@ -148,9 +150,11 @@ then publishes `task.paid`. A result without a `payee` cannot be paid. * Membership is the key. A leaked link cannot be revoked; make a new swarm. * Gossip and relay peers see topic hashes, sizes and timing, not content. -* Sybil identities can endorse each other. That is why peer endorsement pays - only at two endorsers and more endorsements than challenges, and why owner - votes decide. Reputation from receipts is the next layer. +* Sybil identities can endorse each other. Their endorsements still finish + tasks, but only reviewers with standing (paid by someone else) make a + bounty payable, and owner votes decide. Standing can be bought by paying + your own sock puppets, which costs real money per identity; cross-swarm + reputation from receipts is the next layer. * Posters can refuse to pay. `task.paid` makes payment public, and workers skip lobby posters with unpaid accepted work. * Lease griefing (claim and never submit) costs one lease; leases are capped diff --git a/docs/protocol/relay.md b/docs/protocol/relay.md index 42d6f10a..0f269ed5 100644 --- a/docs/protocol/relay.md +++ b/docs/protocol/relay.md @@ -21,9 +21,11 @@ libp2p protocols, so any node can be dialed by peer id. **Every node is a relay server.** A reservation it grants carries its confirmed public addresses (from AutoNAT, or from `C0MPUTE_P2P_EXTERNAL`, which a gateway with a fixed open port sets and which is confirmed at -start). A node behind NAT has none, and clients reject a reservation -without addresses, so in practice the public gateways and cloud workers -carry the relays. +start). A node with no confirmed external address refuses reservations +outright (a reservation rate limiter gated on it, since v0.3.60): before +that, nodes behind NAT or Docker granted empty reservations that clients +then rejected with `NoAddressesInReservation`. In practice the public +gateways and cloud workers carry the relays. Server limits (raised from libp2p's defaults, which fit hole-punch coordination only, so a job result can cross a relay when punching @@ -43,8 +45,14 @@ confirmed us public, the node listens on `/p2p//p2p-circuit`, which asks the relay for a reservation. It keeps at most two reservations, so losing one relay never makes it unreachable. A relay that refuses (or later drops) a reservation -is left alone for 10 minutes. When AutoNAT later confirms the node public, -it gives its reservations back. +is left alone for 10 minutes, or 6 hours when it answered with no +addresses. When AutoNAT later confirms the node public, it gives its +reservations back. + +Public is sticky for 30 minutes: after AutoNAT said Public, a Private +verdict starts relay reservations only once that grace has passed (one +probe server that cannot dial back, being behind NAT itself, used to flip +public nodes). A node with `C0MPUTE_P2P_EXTERNAL` set is always public. The accepted reservation shows up as a listen address, diff --git a/node/crates/c0mpute-core/src/agent_swarm/reducer.rs b/node/crates/c0mpute-core/src/agent_swarm/reducer.rs index 0ea90b98..32a25ecb 100644 --- a/node/crates/c0mpute-core/src/agent_swarm/reducer.rs +++ b/node/crates/c0mpute-core/src/agent_swarm/reducer.rs @@ -17,7 +17,8 @@ //! * A task is done while one of its results stands; a job is complete when //! its final task is done. //! * A bounty is payable when an owner accepted the result, or at least two -//! peers endorsed it and endorsements outnumber challenges. +//! reviewers with standing (someone else once paid them) endorsed it and +//! they outnumber challenges. use std::collections::{BTreeMap, BTreeSet}; @@ -383,14 +384,30 @@ impl State { } } - /// Whether the job's bounty is owed for this result. + /// Whether `peer` has standing as a reviewer: someone other than itself + /// marked one of its results paid. Paying costs real money, so a swarm + /// of fresh identities endorsing each other has none. + pub fn has_standing(&self, peer: &str) -> bool { + self.submissions + .values() + .any(|s| s.author == peer && s.paid.as_ref().is_some_and(|p| p.by != peer)) + } + + /// Whether the job's bounty is owed for this result: an owner accepted + /// it, or at least two reviewers with standing endorsed it and they + /// outnumber every challenge. pub fn payable(&self, sub: &Submission) -> bool { match self.owner_votes(sub) { (_, true) => false, (true, false) => true, (false, false) => { - let (e, c) = Self::tally(sub); - e >= 2 && e > c + let trusted = sub + .reviews + .iter() + .filter(|(who, r)| r.verdict == Verdict::Accept && self.has_standing(who)) + .count(); + let (_, c) = Self::tally(sub); + trusted >= 2 && trusted > c } } } @@ -750,6 +767,49 @@ mod tests { assert!(State::build(&[job.clone(), close(&poster, 2)]).jobs[&job.id].closed); } + #[test] + fn only_reviewers_with_standing_make_a_result_payable() { + let paid = |sub: &str, by: &libp2p_kp::Keypair, l| { + ev(by, SW, l, l, Body::TaskPaid(TaskPaid { submission: sub.into(), amount_usd: "1".into(), rail_ref: String::new() })) + }; + let (poster, _) = signer(); + let (w, _) = signer(); + let (s1, _) = signer(); + let (s2, _) = signer(); + // Job 1: s1 and s2 each earn standing (paid by the poster). + let j1 = ev(&poster, SW, 1, 1, post(vec![("a", vec![]), ("b", vec![])], "1")); + let (a, b) = (make_task_id(&j1.id, "a"), make_task_id(&j1.id, "b")); + let sa = ev(&s1, SW, 3, 3, submit(&a, "x")); + let sb = ev(&s2, SW, 5, 5, submit(&b, "y")); + // Job 2: w's result endorsed by s1 and s2, no owner vote. + let j2 = ev(&poster, SW, 10, 10, post(vec![("c", vec![])], "1")); + let c = make_task_id(&j2.id, "c"); + let sc = ev(&w, SW, 12, 12, submit(&c, "z")); + let mut log = vec![ + j1.clone(), ev(&s1, SW, 2, 2, claim(&a)), sa.clone(), ev(&s2, SW, 4, 4, claim(&b)), sb.clone(), + j2.clone(), ev(&w, SW, 11, 11, claim(&c)), sc.clone(), + ev(&s1, SW, 13, 13, review(&sc.id, true)), ev(&s2, SW, 14, 14, review(&sc.id, true)), + ]; + let s = State::build(&log); + assert!(s.stands(&s.submissions[&sc.id]), "two endorsements finish the task"); + assert!(!s.payable(&s.submissions[&sc.id]), "fresh identities cannot make it payable"); + // Self-marked payment gives no standing. + log.push(paid(&sa.id, &s1, 15)); + assert!(!State::build(&log).has_standing(&libp2p_kp::peer(&s1))); + // Paid by the poster: both reviewers gain standing; now payable. + log.push(paid(&sa.id, &poster, 16)); + log.push(paid(&sb.id, &poster, 17)); + let s = State::build(&log); + assert!(s.payable(&s.submissions[&sc.id])); + } + + mod libp2p_kp { + pub use c0mpute_net::libp2p::identity::Keypair; + pub fn peer(k: &Keypair) -> String { + k.public().to_peer_id().to_string() + } + } + #[test] fn unpaid_posters_are_visible() { let (poster, pp) = signer(); diff --git a/node/crates/c0mpute-net/src/swarm.rs b/node/crates/c0mpute-net/src/swarm.rs index a43d17e3..9d7cd363 100644 --- a/node/crates/c0mpute-net/src/swarm.rs +++ b/node/crates/c0mpute-net/src/swarm.rs @@ -97,8 +97,12 @@ pub const MAX_RELAY_RESERVATIONS: usize = 2; /// Relay server limits. libp2p's defaults (2 min, 128 KiB per circuit) fit /// hole-punch coordination only; a job result has to be able to cross a /// relayed connection when hole punching fails. -fn relay_server_config() -> relay::Config { - relay::Config { +/// Relay server limits. `ready` is true while this node has a confirmed +/// external address: a reservation hands the client our external addresses, +/// so with none it is useless (the client fails with +/// `NoAddressesInReservation`) and is refused up front instead. +fn relay_server_config(ready: Arc) -> relay::Config { + let mut cfg = relay::Config { max_reservations: 256, max_reservations_per_peer: 4, reservation_duration: Duration::from_secs(60 * 60), @@ -107,6 +111,37 @@ fn relay_server_config() -> relay::Config { max_circuit_duration: Duration::from_secs(10 * 60), max_circuit_bytes: 32 * 1024 * 1024, ..relay::Config::default() + }; + cfg.reservation_rate_limiters.push(Box::new( + // web_time::Instant is std's Instant off wasm. + move |_peer: PeerId, _addr: &Multiaddr, _now: std::time::Instant| { + ready.load(std::sync::atomic::Ordering::Relaxed) + }, + )); + cfg +} + +/// After AutoNAT last said Public, a Private verdict is trusted only once +/// this long has passed: one probe server that cannot dial back (it sits +/// behind NAT itself) flipped public nodes to Private and set them +/// reserving relays they did not need. +pub const PUBLIC_GRACE: Duration = Duration::from_secs(30 * 60); + +/// Whether this node counts as publicly reachable: an operator-declared +/// external address, AutoNAT's current verdict, or a Public verdict within +/// [`PUBLIC_GRACE`]. +pub fn effectively_public(declared: bool, nat_public: bool, since_last_public: Option) -> bool { + declared || nat_public || since_last_public.is_some_and(|d| d < PUBLIC_GRACE) +} + +/// How long to leave a relay alone after its reservation closed with +/// `reason` (its Debug text). A relay with no external address will not get +/// one soon: hours, not minutes. +pub fn relay_backoff(reason: &str) -> Duration { + if reason.contains("NoAddressesInReservation") { + Duration::from_secs(6 * 60 * 60) + } else { + RELAY_RETRY_AFTER } } @@ -442,6 +477,8 @@ impl Libp2pNetwork { sw.add_external_address(a.clone()); } task.nat_public = true; + task.declared_public = true; + task.relay_ready.store(true, std::sync::atomic::Ordering::Relaxed); } for addr in &config.listen { task.swarm @@ -947,9 +984,17 @@ struct SwarmTask { relay_listeners: HashMap, /// Relays whose reservation failed, and when: not asked again for /// [`RELAY_RETRY_AFTER`] (identify re-runs every few minutes). - relay_failed: HashMap, + relay_failed: HashMap, /// AutoNAT's latest verdict: true once it has seen us dialed back. nat_public: bool, + /// The operator declared a public address (`C0MPUTE_P2P_EXTERNAL` or + /// [`Libp2pNetwork::add_external_address`]): public whatever AutoNAT says. + declared_public: bool, + /// When AutoNAT last said Public ([`PUBLIC_GRACE`]). + last_public: Option, + /// Shared with the relay server's reservation gate: true while we hold + /// a confirmed external address. + relay_ready: Arc, } /// How long a relay that refused or dropped our reservation is left alone. @@ -985,6 +1030,8 @@ impl SwarmTask { rpc_handlers: RpcHandlers, cmd_weak: mpsc::WeakSender, ) -> Result { + let relay_ready = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let relay_ready_cfg = relay_ready.clone(); let swarm = libp2p::SwarmBuilder::with_existing_identity(keypair.clone()) .with_tokio() .with_tcp( @@ -1047,7 +1094,7 @@ impl SwarmTask { Some(mdns::tokio::Behaviour::new(mdns::Config::default(), local_peer_id)?) }; let mdns = libp2p::swarm::behaviour::toggle::Toggle::from(mdns); - let relay = relay::Behaviour::new(local_peer_id, relay_server_config()); + let relay = relay::Behaviour::new(local_peer_id, relay_server_config(relay_ready_cfg.clone())); let dcutr = dcutr::Behaviour::new(local_peer_id); let autonat = autonat::Behaviour::new( local_peer_id, @@ -1098,6 +1145,9 @@ impl SwarmTask { relay_listeners: HashMap::new(), relay_failed: HashMap::new(), nat_public: false, + declared_public: false, + last_public: None, + relay_ready, }) } @@ -1152,6 +1202,8 @@ impl SwarmTask { Cmd::AddExternal { addr } => { self.swarm.lock().await.add_external_address(addr); self.nat_public = true; + self.declared_public = true; + self.relay_ready.store(true, std::sync::atomic::Ordering::Relaxed); } Cmd::Advertised { reply } => { let swarm = self.swarm.lock().await; @@ -1348,10 +1400,20 @@ impl SwarmTask { Ok(_) => info!(peer = %ev.remote_peer_id, "dcutr: hole punch succeeded; direct connection"), Err(e) => debug!(peer = %ev.remote_peer_id, err = %e, "dcutr: hole punch failed; staying relayed"), }, + SwarmEvent::ExternalAddrConfirmed { .. } | SwarmEvent::ExternalAddrExpired { .. } => { + let has = self.swarm.lock().await.external_addresses().next().is_some(); + let was = self.relay_ready.swap(has, std::sync::atomic::Ordering::Relaxed); + if has != was { + info!(relaying = has, "relay server: {}", if has { "external address confirmed; accepting reservations" } else { "no external address; refusing reservations" }); + } + } SwarmEvent::Behaviour(C0mputeBehaviourEvent::Autonat(autonat::Event::StatusChanged { old, new })) => { info!(?old, ?new, "autonat: reachability changed"); self.nat_public = new.is_public(); if self.nat_public { + self.last_public = Some(std::time::Instant::now()); + } + if effectively_public(self.declared_public, self.nat_public, self.last_public.map(|t| t.elapsed())) { // Dialable directly: give the relay slots back. let held: Vec<(PeerId, ListenerId)> = self.relay_listeners.drain().collect(); let mut sw = self.swarm.lock().await; @@ -1371,7 +1433,8 @@ impl SwarmTask { if let Some(relay) = relay { self.relay_listeners.remove(&relay); if reason.is_err() { - self.relay_failed.insert(relay, std::time::Instant::now()); + let backoff = relay_backoff(&format!("{reason:?}")); + self.relay_failed.insert(relay, (std::time::Instant::now(), backoff)); } info!(%relay, ?reason, "relay: reservation closed"); } @@ -1586,11 +1649,12 @@ impl SwarmTask { /// TCP address. The circuit listener's address joins our listen /// addresses, so identify and Kad hand it to peers like any other. async fn maybe_reserve_relay(&mut self, relay: PeerId, listen_addrs: &[Multiaddr]) { - if self.relay_failed.get(&relay).is_some_and(|t| t.elapsed() < RELAY_RETRY_AFTER) { + if self.relay_failed.get(&relay).is_some_and(|(t, wait)| t.elapsed() < *wait) { return; } + let public = effectively_public(self.declared_public, self.nat_public, self.last_public.map(|t| t.elapsed())); if self.relay_listeners.contains_key(&relay) - || !wants_relay_reservation(relay_client_disabled(), self.nat_public, self.relay_listeners.len()) + || !wants_relay_reservation(relay_client_disabled(), public, self.relay_listeners.len()) { return; } @@ -1959,6 +2023,31 @@ mod tests { } } + #[test] + fn public_grace_and_relay_backoff() { + assert!(effectively_public(true, false, None), "declared address wins"); + assert!(effectively_public(false, true, None)); + assert!(effectively_public(false, false, Some(Duration::from_secs(60))), "one Private right after Public is noise"); + assert!(!effectively_public(false, false, Some(PUBLIC_GRACE)), "grace runs out"); + assert!(!effectively_public(false, false, None)); + assert_eq!( + relay_backoff("Err(Custom { kind: Other, error: Other(Left(Left(Left(Reservation(Protocol(NoAddressesInReservation)))))) })"), + Duration::from_secs(6 * 60 * 60) + ); + assert_eq!(relay_backoff("Err(Io(ConnectionReset))"), RELAY_RETRY_AFTER); + } + + #[test] + fn relay_server_refuses_reservations_without_an_external_address() { + let ready = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let mut cfg = relay_server_config(ready.clone()); + let addr: Multiaddr = "/ip4/8.8.8.8/tcp/1".parse().unwrap(); + let ours = cfg.reservation_rate_limiters.last_mut().unwrap(); + assert!(!ours.try_next(PeerId::random(), &addr, std::time::Instant::now())); + ready.store(true, std::sync::atomic::Ordering::Relaxed); + assert!(ours.try_next(PeerId::random(), &addr, std::time::Instant::now())); + } + #[test] fn relay_reservation_decision_and_addrs() { assert!(wants_relay_reservation(false, false, 0));