feat(net): workers dial the buyer back to deliver results (C0MPUTE_P2P_LISTEN/EXTERNAL); v0.2.66

Cargo.toml

@@ -23,7 +23,7 @@ members = [
 ]
 
 [workspace.package]
-version = "0.2.65"
+version = "0.2.66"
 edition = "2024"
 license = "MIT"
 repository = "https://github.com/profullstack/c0mpute"

apps/tui/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@c0mpute/tui",
-  "version": "0.2.65",
+  "version": "0.2.66",
   "private": true,
   "type": "module",
   "bin": {

apps/web/package.json

@@ -1,6 +1,6 @@
 {
   "name": "c0mpute",
-  "version": "0.2.65",
+  "version": "0.2.66",
   "private": true,
   "type": "module",
   "scripts": {

apps/web/src/app/releases/latest.json/route.ts

@@ -28,7 +28,7 @@ export const dynamic = "force-dynamic";
  * this in sync with the workspace version in `Cargo.toml` on each release so a
  * freshly built CLI reports "already latest" rather than a phantom upgrade.
  */
-const FALLBACK_VERSION = "0.2.65";
+const FALLBACK_VERSION = "0.2.66";
 const FALLBACK_MIN_REQUIRED = "0.0.1";
 
 type Channel = "stable" | "beta" | "nightly";

node/crates/c0mpute-core/src/buyer.rs

@@ -309,6 +309,10 @@ pub async fn run_auction(net: Arc<Libp2pNetwork>, auction: JobAuction) -> Result
         spec_inline: Some(auction.spec_inline.clone()),
         published_at_ms: unix_ms(),
         external_ref: auction.external_ref.clone(),
+        buyer_addrs: crate::reachability::buyer_external_addrs(
+            std::env::var("C0MPUTE_P2P_EXTERNAL").ok().as_deref(),
+            &offer.buyer_peer_id,
+        ),
     };
     publish_json(&net, &topic, &accept).await?;
     if let Some(tx) = &auction.accept_notify {

node/crates/c0mpute-core/src/dispatch.rs

@@ -572,6 +572,7 @@ mod tests {
             spec_inline: Some(serde_json::json!({"input_url": "https://example.com/a.mp4"})),
             published_at_ms: 0,
             external_ref: Some("infernet:job_42".into()),
+            buyer_addrs: Vec::new(),
         }
     }
 

node/crates/c0mpute-core/src/lib.rs

@@ -13,6 +13,7 @@ pub(crate) mod gateway_util;
 pub mod jobs_gateway;
 pub mod net_guard;
 pub mod pricing;
+pub mod reachability;
 pub mod private;
 pub mod register;
 pub mod resources;

node/crates/c0mpute-core/src/reachability.rs

@@ -0,0 +1,122 @@
+//! Getting results back from workers that cannot be dialed.
+//!
+//! Workers listen on a random port, usually behind NAT or a firewall, and
+//! c0mpute has no relay, so a buyer often cannot open a connection to the
+//! worker that holds its result. A buyer gateway, on the other hand, is one
+//! machine we control: it listens on a fixed open port (`C0MPUTE_P2P_LISTEN`)
+//! and advertises its public address (`C0MPUTE_P2P_EXTERNAL`) in every
+//! `JobAccept`. The winning worker dials that address before publishing its
+//! receipt, and the buyer's result fetch reuses the connection.
+
+use c0mpute_net::Multiaddr;
+use c0mpute_net::Protocol;
+
+/// At most this many buyer addresses are dialed, whatever the accept says.
+pub const MAX_BUYER_ADDRS: usize = 4;
+
+fn parse_list(raw: Option<&str>) -> Vec<Multiaddr> {
+    raw.unwrap_or("")
+        .split(',')
+        .map(str::trim)
+        .filter(|s| !s.is_empty())
+        .filter_map(|s| s.parse::<Multiaddr>().ok())
+        .collect()
+}
+
+/// `C0MPUTE_P2P_LISTEN`: comma-separated multiaddrs to listen on.
+pub fn listen_addrs(raw: Option<&str>) -> Vec<Multiaddr> {
+    parse_list(raw)
+}
+
+/// `C0MPUTE_P2P_EXTERNAL` as the strings a buyer puts in its accepts, each
+/// ending in `/p2p/<buyer_peer_id>`.
+pub fn buyer_external_addrs(raw: Option<&str>, buyer_peer_id: &str) -> Vec<String> {
+    parse_list(raw)
+        .into_iter()
+        .filter_map(|addr| {
+            let has_peer = addr.iter().any(|p| matches!(p, Protocol::P2p(_)));
+            let s = addr.to_string();
+            if has_peer {
+                s.ends_with(&format!("/p2p/{buyer_peer_id}")).then_some(s)
+            } else {
+                Some(format!("{s}/p2p/{buyer_peer_id}"))
+            }
+        })
+        .take(MAX_BUYER_ADDRS)
+        .collect()
+}
+
+/// The buyer addresses a worker may dial: public IPv4/IPv6 over TCP, ending
+/// in the accept's own buyer peer id, at most [`MAX_BUYER_ADDRS`]. Anything
+/// else is dropped, so an accept cannot make a worker dial a private network
+/// or a third party.
+pub fn dialable_buyer_addrs(addrs: &[String], buyer_peer_id: &str) -> Vec<Multiaddr> {
+    addrs
+        .iter()
+        .filter_map(|s| s.parse::<Multiaddr>().ok())
+        .filter(|addr| {
+            let mut ip_ok = false;
+            let mut tcp = false;
+            let mut peer_ok = false;
+            for p in addr.iter() {
+                match p {
+                    Protocol::Ip4(ip) => ip_ok = crate::net_guard::is_public_ip(ip.into()),
+                    Protocol::Ip6(ip) => ip_ok = crate::net_guard::is_public_ip(ip.into()),
+                    Protocol::Tcp(_) => tcp = true,
+                    Protocol::P2p(peer) => peer_ok = peer.to_base58() == buyer_peer_id,
+                    _ => {}
+                }
+            }
+            ip_ok && tcp && peer_ok
+        })
+        .take(MAX_BUYER_ADDRS)
+        .collect()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    const PEER: &str = "12D3KooWADm5FfNxzPRanv1vacpPD7eh1bWv6pqmDCGE69Zof2n6";
+    const OTHER: &str = "12D3KooWN99F7k4Fj6SdUqZ2VG4ZBLuw1e1ftwdPDBx3GsiDXzqv";
+
+    #[test]
+    fn external_addrs_get_the_buyer_peer_id() {
+        let a = buyer_external_addrs(Some("/ip4/23.95.228.174/tcp/46341"), PEER);
+        assert_eq!(a, vec![format!("/ip4/23.95.228.174/tcp/46341/p2p/{PEER}")]);
+        let wrong = buyer_external_addrs(Some(&format!("/ip4/1.2.3.4/tcp/1/p2p/{OTHER}")), PEER);
+        assert!(wrong.is_empty(), "an address for another peer is dropped");
+        assert!(buyer_external_addrs(None, PEER).is_empty());
+        assert!(buyer_external_addrs(Some("not a multiaddr"), PEER).is_empty());
+    }
+
+    #[test]
+    fn workers_only_dial_public_tcp_addrs_of_the_buyer() {
+        let good = format!("/ip4/23.95.228.174/tcp/46341/p2p/{PEER}");
+        let private = format!("/ip4/10.0.0.5/tcp/46341/p2p/{PEER}");
+        let loopback = format!("/ip4/127.0.0.1/tcp/46341/p2p/{PEER}");
+        let other_peer = format!("/ip4/23.95.228.174/tcp/46341/p2p/{OTHER}");
+        let no_peer = "/ip4/23.95.228.174/tcp/46341".to_string();
+        let addrs = vec![good.clone(), private, loopback, other_peer, no_peer];
+        let ok = dialable_buyer_addrs(&addrs, PEER);
+        assert_eq!(ok.len(), 1);
+        assert_eq!(ok[0].to_string(), good);
+    }
+
+    #[test]
+    fn at_most_four_addrs() {
+        let addrs: Vec<String> = (1..10)
+            .map(|i| format!("/ip4/23.95.228.{i}/tcp/46341/p2p/{PEER}"))
+            .collect();
+        assert_eq!(dialable_buyer_addrs(&addrs, PEER).len(), MAX_BUYER_ADDRS);
+    }
+
+    #[test]
+    fn listen_list_parses() {
+        assert_eq!(
+            listen_addrs(Some("/ip4/0.0.0.0/tcp/46341, /ip6/::/tcp/46341")).len(),
+            2
+        );
+        assert!(listen_addrs(Some("")).is_empty());
+    }
+}

node/crates/c0mpute-core/src/runner.rs

@@ -593,6 +593,20 @@ pub(crate) async fn publish_receipt(
         signature: None,
         external_ref: accept.external_ref.clone(),
     };
+    // Reach back to the buyer first: it can fetch the result only over a
+    // connection, and it usually cannot dial our random, firewalled port.
+    if receipt.output_hash.is_some() {
+        let addrs =
+            crate::reachability::dialable_buyer_addrs(&accept.buyer_addrs, &accept.buyer_peer_id);
+        if !addrs.is_empty() {
+            for addr in addrs {
+                if let Err(e) = net.dial(addr).await {
+                    tracing::debug!(err = %e, "dial back to buyer failed");
+                }
+            }
+            tokio::time::sleep(std::time::Duration::from_millis(1500)).await;
+        }
+    }
     let topic = job_topic(&accept.workload_type);
     if let Ok(payload) = serde_json::to_vec(&receipt) {
         if let Err(e) = net.publish(&topic, payload).await {

node/crates/c0mpute-core/src/status_aggregator.rs

@@ -883,6 +883,7 @@ mod tests {
             spec_inline: None,
             published_at_ms: now - 5_000,
             external_ref: None,
+            buyer_addrs: Vec::new(),
         })
         .await;
         assert_eq!(jobs.in_flight.read().await.len(), 1);
@@ -920,6 +921,7 @@ mod tests {
             spec_inline: None,
             published_at_ms: now,
             external_ref: None,
+            buyer_addrs: Vec::new(),
         })
         .await;
         jobs.on_receipt(JobReceipt {

node/crates/c0mpute-core/src/supervisor.rs

@@ -48,9 +48,17 @@ impl Supervisor {
         .await;
         info!(count = bootstrap_addrs.len(), "bootstrap peers loaded");
 
-        let net_cfg = NetworkConfig::for_dir(identity_dir)
+        let mut net_cfg = NetworkConfig::for_dir(identity_dir)
             .with_local_source(local_source)
             .with_bootstrap(bootstrap_addrs);
+        // A fixed, firewall-opened port for nodes that must be dialable
+        // (buyer gateways): C0MPUTE_P2P_LISTEN=/ip4/0.0.0.0/tcp/46341
+        let listen =
+            crate::reachability::listen_addrs(std::env::var("C0MPUTE_P2P_LISTEN").ok().as_deref());
+        if !listen.is_empty() {
+            info!(?listen, "libp2p listening on configured addresses");
+            net_cfg = net_cfg.with_listen(listen);
+        }
 
         let libp2p_net: Arc<Libp2pNetwork> = Arc::new(Libp2pNetwork::spawn(net_cfg).await?);
         info!(

node/crates/c0mpute-core/src/webhook.rs

@@ -1228,7 +1228,7 @@ mod tests {
     fn accepts_must_match_the_webhook_offer_we_bid_on() {
         let s = serde_json::to_value(spec(3)).unwrap();
         let bid = crate::dispatch::OpenBid { cap: 0.002, price: 0.001, buyer_peer_id: "buyer".into(), spec_inline: Some(s.clone()), external_ref: Some("coinpay:inv_1".into()), expires_at_ms: u64::MAX / 2 };
-        let accept = JobAccept { job_id: "j".into(), buyer_peer_id: "buyer".into(), winning_bidder_peer_id: "us".into(), agreed_price_usd: 0.001, workload_type: WORKLOAD.into(), spec_inline: Some(s), published_at_ms: 0, external_ref: Some("coinpay:inv_1".into()) };
+        let accept = JobAccept { job_id: "j".into(), buyer_peer_id: "buyer".into(), winning_bidder_peer_id: "us".into(), agreed_price_usd: 0.001, workload_type: WORKLOAD.into(), spec_inline: Some(s), published_at_ms: 0, external_ref: Some("coinpay:inv_1".into()), buyer_addrs: Vec::new() };
         assert!(bid.matches(&accept, 0).is_ok());
         // A forged accept swapping in another buyer key (to read the call) is refused.
         let mut a = accept.clone();