From c905a1e2a139ec41fd5aad9be123ef810a4f739f Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 16:59:36 +0000 Subject: [PATCH 1/2] feat(net): winning workers dial the buyer back so it can fetch results; bump v0.2.66 Live: dev2's gateway got 'timed out fetching the c0mpute result' from every remote image worker. Workers listen on random, firewalled ports and c0mpute has no relay, so a buyer could only fetch from workers on its own host. - JobAccept.buyer_addrs (optional, old nodes ignore it): the buyer's public multiaddrs from C0MPUTE_P2P_EXTERNAL. - C0MPUTE_P2P_LISTEN: fixed listen addresses (a buyer gateway opens one port). - Before publishing a receipt the worker dials those addresses (public TCP only, must end in the accept's own buyer peer id, max 4), so the buyer's fetch reuses that connection. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 32 +- Cargo.toml | 2 +- apps/tui/package.json | 2 +- apps/web/package.json | 2 +- .../web/src/app/releases/latest.json/route.ts | 2 +- node/crates/c0mpute-core/src/buyer.rs | 4 + node/crates/c0mpute-core/src/dispatch.rs | 1 + node/crates/c0mpute-core/src/lib.rs | 1 + node/crates/c0mpute-core/src/reachability.rs | 122 ++++ node/crates/c0mpute-core/src/runner.rs | 14 + .../c0mpute-core/src/status_aggregator.rs | 2 + node/crates/c0mpute-core/src/supervisor.rs | 10 +- node/crates/c0mpute-core/src/webhook.rs | 577 +++++++++++++++--- node/crates/c0mpute-net/src/topics.rs | 12 +- package.json | 2 +- packages/shared/package.json | 2 +- plugins/ad-generator-agent/module.toml | 2 +- plugins/ads-manager/module.toml | 2 +- plugins/coinpay/module.toml | 2 +- plugins/crawl/module.toml | 2 +- plugins/ctv-ads/module.toml | 2 +- plugins/diffusion/module.toml | 2 +- plugins/embed/module.toml | 2 +- plugins/gameserver/module.toml | 2 +- plugins/hosting/module.toml | 2 +- plugins/infernet/module.toml | 2 +- plugins/live-stream/module.toml | 2 +- plugins/ocr/module.toml | 2 +- plugins/podcast-ads/module.toml | 2 +- plugins/podcasting/module.toml | 2 +- plugins/podcasts/module.toml | 2 +- plugins/render/module.toml | 2 +- plugins/secure-chat/module.toml | 2 +- plugins/storage/module.toml | 2 +- plugins/train/module.toml | 2 +- plugins/transcode/module.toml | 2 +- plugins/tts/module.toml | 2 +- plugins/video/module.toml | 2 +- plugins/webhook/module.toml | 2 +- plugins/whisper/module.toml | 2 +- plugins/workflow-step/module.toml | 2 +- 41 files changed, 692 insertions(+), 145 deletions(-) create mode 100644 node/crates/c0mpute-core/src/reachability.rs diff --git a/Cargo.lock b/Cargo.lock index f979f14b..91265a8b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -478,7 +478,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "c0mpute-api" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "c0mpute-proto", @@ -493,7 +493,7 @@ dependencies = [ [[package]] name = "c0mpute-bench" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "blake3", @@ -504,7 +504,7 @@ dependencies = [ [[package]] name = "c0mpute-cli" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "axum", @@ -534,7 +534,7 @@ dependencies = [ [[package]] name = "c0mpute-core" -version = "0.2.65" +version = "0.2.66" dependencies = [ "aes-gcm", "anyhow", @@ -578,7 +578,7 @@ dependencies = [ [[package]] name = "c0mpute-doctor" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "serde", @@ -590,7 +590,7 @@ dependencies = [ [[package]] name = "c0mpute-envelope" -version = "0.2.65" +version = "0.2.66" dependencies = [ "blake3", "bs58", @@ -603,7 +603,7 @@ dependencies = [ [[package]] name = "c0mpute-gateway" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "axum", @@ -625,7 +625,7 @@ dependencies = [ [[package]] name = "c0mpute-market" -version = "0.2.65" +version = "0.2.66" dependencies = [ "c0mpute-envelope", "serde", @@ -635,7 +635,7 @@ dependencies = [ [[package]] name = "c0mpute-net" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "async-trait", @@ -652,7 +652,7 @@ dependencies = [ [[package]] name = "c0mpute-placement" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "async-trait", @@ -673,7 +673,7 @@ dependencies = [ [[package]] name = "c0mpute-proto" -version = "0.2.65" +version = "0.2.66" dependencies = [ "blake3", "hex", @@ -685,7 +685,7 @@ dependencies = [ [[package]] name = "c0mpute-secure-chat" -version = "0.2.65" +version = "0.2.66" dependencies = [ "aes-gcm", "anyhow", @@ -708,7 +708,7 @@ dependencies = [ [[package]] name = "c0mpute-store" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "blake3", @@ -726,7 +726,7 @@ dependencies = [ [[package]] name = "c0mpute-transcode" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "c0mpute-proto", @@ -739,7 +739,7 @@ dependencies = [ [[package]] name = "c0mpute-update" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "hex", @@ -753,7 +753,7 @@ dependencies = [ [[package]] name = "c0mpute-verify" -version = "0.2.65" +version = "0.2.66" dependencies = [ "anyhow", "blake3", diff --git a/Cargo.toml b/Cargo.toml index b294e01f..f39beda4 100644 --- a/Cargo.toml +++ b/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" diff --git a/apps/tui/package.json b/apps/tui/package.json index 9cfae0cb..0ed320ba 100644 --- a/apps/tui/package.json +++ b/apps/tui/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/tui", - "version": "0.2.65", + "version": "0.2.66", "private": true, "type": "module", "bin": { diff --git a/apps/web/package.json b/apps/web/package.json index 7451627a..4f0e2948 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute", - "version": "0.2.65", + "version": "0.2.66", "private": true, "type": "module", "scripts": { diff --git a/apps/web/src/app/releases/latest.json/route.ts b/apps/web/src/app/releases/latest.json/route.ts index 317b449a..96a6fc78 100644 --- a/apps/web/src/app/releases/latest.json/route.ts +++ b/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"; diff --git a/node/crates/c0mpute-core/src/buyer.rs b/node/crates/c0mpute-core/src/buyer.rs index 92a1c493..9d2ee7bb 100644 --- a/node/crates/c0mpute-core/src/buyer.rs +++ b/node/crates/c0mpute-core/src/buyer.rs @@ -309,6 +309,10 @@ pub async fn run_auction(net: Arc, 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 { diff --git a/node/crates/c0mpute-core/src/dispatch.rs b/node/crates/c0mpute-core/src/dispatch.rs index 428f1f9a..742c6a1a 100644 --- a/node/crates/c0mpute-core/src/dispatch.rs +++ b/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(), } } diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index b5df0f8a..73e4be7d 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/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; diff --git a/node/crates/c0mpute-core/src/reachability.rs b/node/crates/c0mpute-core/src/reachability.rs new file mode 100644 index 00000000..b6802d94 --- /dev/null +++ b/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 { + raw.unwrap_or("") + .split(',') + .map(str::trim) + .filter(|s| !s.is_empty()) + .filter_map(|s| s.parse::().ok()) + .collect() +} + +/// `C0MPUTE_P2P_LISTEN`: comma-separated multiaddrs to listen on. +pub fn listen_addrs(raw: Option<&str>) -> Vec { + parse_list(raw) +} + +/// `C0MPUTE_P2P_EXTERNAL` as the strings a buyer puts in its accepts, each +/// ending in `/p2p/`. +pub fn buyer_external_addrs(raw: Option<&str>, buyer_peer_id: &str) -> Vec { + 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 { + addrs + .iter() + .filter_map(|s| s.parse::().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 = (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()); + } +} diff --git a/node/crates/c0mpute-core/src/runner.rs b/node/crates/c0mpute-core/src/runner.rs index e87178cf..47bc56a6 100644 --- a/node/crates/c0mpute-core/src/runner.rs +++ b/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 { diff --git a/node/crates/c0mpute-core/src/status_aggregator.rs b/node/crates/c0mpute-core/src/status_aggregator.rs index f164427c..4be2709b 100644 --- a/node/crates/c0mpute-core/src/status_aggregator.rs +++ b/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 { diff --git a/node/crates/c0mpute-core/src/supervisor.rs b/node/crates/c0mpute-core/src/supervisor.rs index 2748dc86..0c21f711 100644 --- a/node/crates/c0mpute-core/src/supervisor.rs +++ b/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 = Arc::new(Libp2pNetwork::spawn(net_cfg).await?); info!( diff --git a/node/crates/c0mpute-core/src/webhook.rs b/node/crates/c0mpute-core/src/webhook.rs index a091d848..6f3dbd52 100644 --- a/node/crates/c0mpute-core/src/webhook.rs +++ b/node/crates/c0mpute-core/src/webhook.rs @@ -153,7 +153,9 @@ impl WebhookSpec { /// Worst-case seconds the worker spends on this delivery. pub fn budget_secs(&self) -> u64 { let tries = u64::from(self.max_attempts); - let waits: u64 = (1..tries).map(|n| backoff(self.backoff_ms, n as u32).as_millis() as u64).sum(); + let waits: u64 = (1..tries) + .map(|n| backoff(self.backoff_ms, n as u32).as_millis() as u64) + .sum(); (tries * u64::from(self.timeout_ms) + waits).div_ceil(1000) } } @@ -215,8 +217,10 @@ impl WebhookCall { let mut bytes = 0; for (k, v) in &self.headers { bytes += k.len() + v.len(); - let name = reqwest::header::HeaderName::from_bytes(k.as_bytes()).with_context(|| format!("bad header name {k:?}"))?; - reqwest::header::HeaderValue::from_str(v).with_context(|| format!("bad value for header {k}"))?; + let name = reqwest::header::HeaderName::from_bytes(k.as_bytes()) + .with_context(|| format!("bad header name {k:?}"))?; + reqwest::header::HeaderValue::from_str(v) + .with_context(|| format!("bad value for header {k}"))?; if RESERVED_HEADERS.contains(&name.as_str()) { bail!("header {k} is set by the worker"); } @@ -224,7 +228,11 @@ impl WebhookCall { if bytes > MAX_HEADER_BYTES { bail!("headers larger than {MAX_HEADER_BYTES} bytes"); } - if self.secret.as_deref().is_some_and(|s| s.is_empty() || s.len() > MAX_SECRET) { + if self + .secret + .as_deref() + .is_some_and(|s| s.is_empty() || s.len() > MAX_SECRET) + { bail!("secret must be 1..={MAX_SECRET} bytes"); } if let Some(id) = &self.id { @@ -281,7 +289,10 @@ impl DeliveryResult { } fn refused(e: impl std::fmt::Display) -> Self { - Self { error: Some(format!("refused: {e}")), ..Default::default() } + Self { + error: Some(format!("refused: {e}")), + ..Default::default() + } } } @@ -319,17 +330,28 @@ pub struct WebhookSealedResult { /// One key per direction, bound to the job: a sealed result can never be /// replayed as a request, nor a message moved to another job. pub fn request_key(secret: &StaticSecret, peer_pub_hex: &str, job_id: &str) -> Result<[u8; 32]> { - crypto::session_key(secret, peer_pub_hex, &format!("c0mpute.webhook/v1/{job_id}/request")) + crypto::session_key( + secret, + peer_pub_hex, + &format!("c0mpute.webhook/v1/{job_id}/request"), + ) } pub fn result_key(secret: &StaticSecret, peer_pub_hex: &str, job_id: &str) -> Result<[u8; 32]> { - crypto::session_key(secret, peer_pub_hex, &format!("c0mpute.webhook/v1/{job_id}/result")) + crypto::session_key( + secret, + peer_pub_hex, + &format!("c0mpute.webhook/v1/{job_id}/result"), + ) } pub fn seal_call(key: &[u8; 32], call: &WebhookCall) -> Result { let sealed = crypto::seal(key, &serde_json::to_vec(call)?)?; if sealed.len() > MAX_SEALED { - bail!("sealed request is {} bytes; one delivery carries at most {MAX_SEALED} (shrink the body or headers)", sealed.len()); + bail!( + "sealed request is {} bytes; one delivery carries at most {MAX_SEALED} (shrink the body or headers)", + sealed.len() + ); } Ok(sealed) } @@ -382,7 +404,10 @@ fn floor_char_boundary(s: &str, mut i: usize) -> usize { } fn unix_secs() -> u64 { - std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0) + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) } // ── Standard Webhooks signature ───────────────────────────────────────── @@ -390,12 +415,16 @@ fn unix_secs() -> u64 { /// The signing key: `whsec_` decodes to bytes (Standard Webhooks); /// anything else is used as raw bytes. fn signing_key(secret: &str) -> Vec { - secret.strip_prefix("whsec_").and_then(|b| B64.decode(b).ok()).unwrap_or_else(|| secret.as_bytes().to_vec()) + secret + .strip_prefix("whsec_") + .and_then(|b| B64.decode(b).ok()) + .unwrap_or_else(|| secret.as_bytes().to_vec()) } /// `webhook-signature` value: `v1,`. pub fn sign(secret: &str, id: &str, timestamp: u64, body: &[u8]) -> String { - let mut mac = as Mac>::new_from_slice(&signing_key(secret)).expect("hmac accepts any key length"); + let mut mac = as Mac>::new_from_slice(&signing_key(secret)) + .expect("hmac accepts any key length"); mac.update(id.as_bytes()); mac.update(b"."); mac.update(timestamp.to_string().as_bytes()); @@ -420,12 +449,18 @@ async fn resolve_public(u: &url::Url, guard: Guard) -> Result<(String, Vec = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((lookup.as_str(), port))) - .await - .map_err(|_| anyhow!("resolving {host} timed out"))? - .with_context(|| format!("resolve {host}"))? - .collect(); + let lookup = host + .trim_start_matches('[') + .trim_end_matches(']') + .to_string(); + let addrs: Vec = tokio::time::timeout( + Duration::from_secs(10), + tokio::net::lookup_host((lookup.as_str(), port)), + ) + .await + .map_err(|_| anyhow!("resolving {host} timed out"))? + .with_context(|| format!("resolve {host}"))? + .collect(); if addrs.is_empty() { bail!("{host} did not resolve"); } @@ -447,7 +482,12 @@ fn retryable(status: Option) -> bool { /// Send `call` with up to `spec.max_attempts` tries. Never fails: a refused /// or unreachable target is reported in the result. -pub async fn deliver(call: &WebhookCall, spec: &WebhookSpec, default_id: &str, guard: Guard) -> DeliveryResult { +pub async fn deliver( + call: &WebhookCall, + spec: &WebhookSpec, + default_id: &str, + guard: Guard, +) -> DeliveryResult { if let Err(e) = call.validate().and_then(|_| spec.validate()) { return DeliveryResult::refused(e); } @@ -455,9 +495,15 @@ pub async fn deliver(call: &WebhookCall, spec: &WebhookSpec, default_id: &str, g Ok(u) => u, Err(e) => return DeliveryResult::refused(e), }; - let id = call.id.clone().unwrap_or_else(|| format!("msg_{default_id}")); + let id = call + .id + .clone() + .unwrap_or_else(|| format!("msg_{default_id}")); let started = Instant::now(); - let mut out = DeliveryResult { webhook_id: call.secret.as_ref().map(|_| id.clone()), ..Default::default() }; + let mut out = DeliveryResult { + webhook_id: call.secret.as_ref().map(|_| id.clone()), + ..Default::default() + }; for n in 1..=spec.max_attempts { if n > 1 { tokio::time::sleep(backoff(spec.backoff_ms, n - 1)).await; @@ -470,7 +516,11 @@ pub async fn deliver(call: &WebhookCall, spec: &WebhookSpec, default_id: &str, g // A refusal is final; retrying cannot make the address public. out.attempts = n; out.error = Some(format!("refused: {e:#}")); - out.log.push(Attempt { status: None, error: out.error.clone(), latency_ms: 0 }); + out.log.push(Attempt { + status: None, + error: out.error.clone(), + latency_ms: 0, + }); break; } Ok((host, addrs)) => send_once(call, spec, &u, &host, &addrs, &id).await, @@ -486,12 +536,20 @@ pub async fn deliver(call: &WebhookCall, spec: &WebhookSpec, default_id: &str, g out.response_body = r.body; out.truncated = r.truncated; out.error = None; - out.log.push(Attempt { status: Some(r.status), error: None, latency_ms }); + out.log.push(Attempt { + status: Some(r.status), + error: None, + latency_ms, + }); } Err(e) => { let msg = format!("{e:#}"); out.error = Some(msg.clone()); - out.log.push(Attempt { status: None, error: Some(msg), latency_ms }); + out.log.push(Attempt { + status: None, + error: Some(msg), + latency_ms, + }); } } if !retryable(out.log.last().and_then(|a| a.status)) { @@ -509,7 +567,14 @@ struct Answer { truncated: bool, } -async fn send_once(call: &WebhookCall, spec: &WebhookSpec, u: &url::Url, host: &str, addrs: &[std::net::SocketAddr], id: &str) -> Result { +async fn send_once( + call: &WebhookCall, + spec: &WebhookSpec, + u: &url::Url, + host: &str, + addrs: &[std::net::SocketAddr], + id: &str, +) -> Result { let timeout = Duration::from_millis(u64::from(spec.timeout_ms)); let client = reqwest::Client::builder() .redirect(reqwest::redirect::Policy::none()) @@ -534,12 +599,17 @@ async fn send_once(call: &WebhookCall, spec: &WebhookSpec, u: &url::Url, host: & req = req .header("webhook-id", id) .header("webhook-timestamp", ts.to_string()) - .header("webhook-signature", sign(secret, id, ts, call.body.as_bytes())); + .header( + "webhook-signature", + sign(secret, id, ts, call.body.as_bytes()), + ); } if !call.body.is_empty() { req = req.body(call.body.clone()); } - let mut resp = tokio::time::timeout(timeout, req.send()).await.map_err(|_| anyhow!("timed out after {}ms", spec.timeout_ms))??; + let mut resp = tokio::time::timeout(timeout, req.send()) + .await + .map_err(|_| anyhow!("timed out after {}ms", spec.timeout_ms))??; let status = resp.status().as_u16(); let headers = resp .headers() @@ -570,7 +640,12 @@ async fn send_once(call: &WebhookCall, spec: &WebhookSpec, u: &url::Url, host: & if tokio::time::timeout(timeout, read).await.is_err() { truncated = true; } - Ok(Answer { status, headers, body: String::from_utf8_lossy(&body).into_owned(), truncated }) + Ok(Answer { + status, + headers, + body: String::from_utf8_lossy(&body).into_owned(), + truncated, + }) } // ── worker ────────────────────────────────────────────────────────────── @@ -585,7 +660,11 @@ pub fn validate_price(price_usd: f64) -> Result<()> { /// The listing every webhook bid is priced with: a fixed price per delivery. pub fn pricing(price_usd: f64) -> PricingConfig { - PricingConfig { mode: PricingMode::Fixed, min_usd: price_usd, fixed_usd: Some(price_usd) } + PricingConfig { + mode: PricingMode::Fixed, + min_usd: price_usd, + fixed_usd: Some(price_usd), + } } /// Bid only on valid webhook offers whose cap covers our price. @@ -598,7 +677,9 @@ pub fn offer_filter(price_usd: f64) -> OfferFilter { .as_ref() .and_then(|v| serde_json::from_value::(v.clone()).ok()) .is_some_and(|s| { - let deadline_ok = offer.deadline_unix_ms.saturating_sub(offer.published_at_ms) / 1000 >= s.budget_secs(); + let deadline_ok = offer.deadline_unix_ms.saturating_sub(offer.published_at_ms) + / 1000 + >= s.budget_secs(); s.validate().is_ok() && deadline_ok }) }) @@ -613,7 +694,12 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re let topic = job_topic(WORKLOAD); let me = net.peer_id().to_base58(); let mut rx = net.messages(); - let spec: WebhookSpec = serde_json::from_value(accept.spec_inline.clone().context("accept lacks spec_inline")?)?; + let spec: WebhookSpec = serde_json::from_value( + accept + .spec_inline + .clone() + .context("accept lacks spec_inline")?, + )?; spec.validate()?; let kp = crypto::keypair(); let req_key = request_key(&kp.secret, &spec.client_pubkey, &accept.job_id)?; @@ -646,14 +732,22 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re if msg.topic != topic { continue; } - let Ok(r) = serde_json::from_slice::(&msg.data) else { continue }; + let Ok(r) = serde_json::from_slice::(&msg.data) else { + continue; + }; // Only the buyer that won us may hand us the request. - if r.t != "webhook_request" || r.job_id != accept.job_id || r.worker_peer_id != me || !from(&msg, &accept.buyer_peer_id) { + if r.t != "webhook_request" + || r.job_id != accept.job_id + || r.worker_peer_id != me + || !from(&msg, &accept.buyer_peer_id) + { continue; } match open_call(&req_key, &r.sealed) { Ok(c) => break c, - Err(e) => debug!(job_id = %accept.job_id, err = %e, "webhook: unreadable request ignored"), + Err(e) => { + debug!(job_id = %accept.job_id, err = %e, "webhook: unreadable request ignored") + } } }; @@ -661,10 +755,19 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re drop(call); let sealed = seal_result(&res_key, &result)?; let hash = sha256_hex(sealed.as_bytes()); - let msg = serde_json::to_vec(&WebhookSealedResult { t: "webhook_result".into(), job_id: accept.job_id.clone(), worker_peer_id: me.clone(), sealed })?; + let msg = serde_json::to_vec(&WebhookSealedResult { + t: "webhook_result".into(), + job_id: accept.job_id.clone(), + worker_peer_id: me.clone(), + sealed, + })?; let _ = net.publish(&topic, msg.clone()).await; tokio::time::sleep(Duration::from_millis(50)).await; - let status = if result.answered() { JobStatus::Completed } else { JobStatus::Failed }; + let status = if result.answered() { + JobStatus::Completed + } else { + JobStatus::Failed + }; info!(job_id = %accept.job_id, status = ?result.status, attempts = result.attempts, delivered = result.delivered, "webhook: delivery done"); publish_receipt(&net, &accept, Some(hash), status).await; @@ -674,7 +777,12 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re tokio::spawn(async move { let linger = Instant::now() + RESULT_LINGER; loop { - let m = match tokio::time::timeout(linger.saturating_duration_since(Instant::now()), rx.recv()).await { + let m = match tokio::time::timeout( + linger.saturating_duration_since(Instant::now()), + rx.recv(), + ) + .await + { Ok(Ok(m)) => m, Ok(Err(broadcast::error::RecvError::Lagged(_))) => continue, _ => return, @@ -694,11 +802,17 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re /// Runner for `webhook.deliver`: each accepted job runs in its own task on /// one of the node's delivery slots. -pub fn spawn_runner(net: Arc, load: WorkerLoad, guard: Guard) -> mpsc::Sender { +pub fn spawn_runner( + net: Arc, + load: WorkerLoad, + guard: Guard, +) -> mpsc::Sender { let (tx, mut rx) = mpsc::channel::(64); tokio::spawn(async move { while let Some(job) = rx.recv().await { - let Ok(slot) = load.compute_slots().acquire_owned().await else { return }; + let Ok(slot) = load.compute_slots().acquire_owned().await else { + return; + }; let net = net.clone(); let load = load.clone(); tokio::spawn(async move { @@ -724,20 +838,42 @@ pub struct Serving { /// `c0mpute webhook serve`: advertise `webhook:deliver` and deliver paid, /// sealed webhooks, `concurrency` at a time. -pub async fn serve(net_opts: &crate::temporal::node::NetOptions, price_usd: f64, concurrency: usize) -> Result { +pub async fn serve( + net_opts: &crate::temporal::node::NetOptions, + price_usd: f64, + concurrency: usize, +) -> Result { validate_price(price_usd)?; let concurrency = concurrency.clamp(1, 256); let net = crate::temporal::node::spawn_node(net_opts).await?; let tags = vec![TAG.to_string()]; let listing = pricing(price_usd); let hardware = serde_json::json!({"version": env!("CARGO_PKG_VERSION"), "webhook": {"concurrency": concurrency}, "pricing": listing.listing()}); - tokio::spawn(crate::capabilities::advertise_loop(net.clone(), tags.clone(), hardware, crate::capabilities::local_bench_score(), crate::capabilities::DEFAULT_ADVERTISE_INTERVAL)); + tokio::spawn(crate::capabilities::advertise_loop( + net.clone(), + tags.clone(), + hardware, + crate::capabilities::local_bench_score(), + crate::capabilities::DEFAULT_ADVERTISE_INTERVAL, + )); let load = WorkerLoad::new(Arc::new(Semaphore::new(concurrency)), concurrency); let tx = spawn_runner(net.clone(), load.clone(), Guard::default()); - crate::dispatch::run_worker_subscriber_with_load(net.clone(), WORKLOAD.into(), tags.clone(), Some(tx), Some(offer_filter(price_usd)), Some(listing), Some(load)); + crate::dispatch::run_worker_subscriber_with_load( + net.clone(), + WORKLOAD.into(), + tags.clone(), + Some(tx), + Some(offer_filter(price_usd)), + Some(listing), + Some(load), + ); info!(peer = %net.peer_id().to_base58(), price_usd, concurrency, "webhook: serving"); tokio::time::sleep(Duration::from_millis(300)).await; - Ok(Serving { net, tags, price_usd }) + Ok(Serving { + net, + tags, + price_usd, + }) } // ── buyer ─────────────────────────────────────────────────────────────── @@ -773,7 +909,13 @@ pub async fn send(net: Arc, req: &SendRequest) -> Result, req: &SendRequest) -> Result) -> Result<()> { +async fn key_up( + net: &Libp2pNetwork, + topic: &str, + secret: &StaticSecret, + call: &WebhookCall, + ready: WebhookReady, + keys: &mut HashMap, +) -> Result<()> { let key = request_key(secret, &ready.worker_pubkey, &ready.job_id)?; let sealed = seal_call(&key, call)?; - let payload = serde_json::to_vec(&WebhookSealed { t: "webhook_request".into(), job_id: ready.job_id.clone(), worker_peer_id: ready.worker_peer_id.clone(), sealed })?; + let payload = serde_json::to_vec(&WebhookSealed { + t: "webhook_request".into(), + job_id: ready.job_id.clone(), + worker_peer_id: ready.worker_peer_id.clone(), + sealed, + })?; let _ = net.publish(topic, payload.clone()).await; debug!(job_id = %ready.job_id, worker = %ready.worker_peer_id, "webhook: sealed request sent"); - keys.insert(ready.job_id, Exchange { worker_pubkey: ready.worker_pubkey, payload }); + keys.insert( + ready.job_id, + Exchange { + worker_pubkey: ready.worker_pubkey, + payload, + }, + ); Ok(()) } @@ -905,14 +1082,29 @@ mod tests { use axum::http::{HeaderMap, StatusCode}; use axum::routing::any; - const LOCAL: Guard = Guard { allow_private: true }; + const LOCAL: Guard = Guard { + allow_private: true, + }; fn spec(attempts: u32) -> WebhookSpec { - WebhookSpec { v: 1, client_pubkey: "11".repeat(32), max_attempts: attempts, timeout_ms: 5_000, backoff_ms: 100 } + WebhookSpec { + v: 1, + client_pubkey: "11".repeat(32), + max_attempts: attempts, + timeout_ms: 5_000, + backoff_ms: 100, + } } fn call(url: &str) -> WebhookCall { - WebhookCall { url: url.into(), method: "POST".into(), headers: vec![("x-tenant".into(), "acme".into())], body: r#"{"event":"paid"}"#.into(), secret: None, id: None } + WebhookCall { + url: url.into(), + method: "POST".into(), + headers: vec![("x-tenant".into(), "acme".into())], + body: r#"{"event":"paid"}"#.into(), + secret: None, + id: None, + } } #[derive(Clone, Default)] @@ -926,7 +1118,12 @@ mod tests { /// to `/other`, `/big` 100 KiB. async fn target() -> (String, Seen) { let seen = Seen::default(); - async fn handle(State(s): State, uri: axum::http::Uri, headers: HeaderMap, body: axum::body::Bytes) -> (StatusCode, HeaderMap, Vec) { + async fn handle( + State(s): State, + uri: axum::http::Uri, + headers: HeaderMap, + body: axum::body::Bytes, + ) -> (StatusCode, HeaderMap, Vec) { let mut h = HeaderMap::new(); match uri.path() { "/other" => { @@ -966,18 +1163,59 @@ mod tests { let bk = request_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(); let wk = request_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(); let sealed = seal_call(&bk, &c).unwrap(); - assert!(!sealed.contains("example.com") && !sealed.contains("whsec_"), "ciphertext must not leak the call"); + assert!( + !sealed.contains("example.com") && !sealed.contains("whsec_"), + "ciphertext must not leak the call" + ); assert_eq!(open_call(&wk, &sealed).unwrap(), c); // Another job, another direction, another worker: none can open it. - assert!(open_call(&request_key(&worker.secret, &buyer.public_hex, "job-2").unwrap(), &sealed).is_err()); - assert!(open_call(&result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), &sealed).is_err()); + assert!( + open_call( + &request_key(&worker.secret, &buyer.public_hex, "job-2").unwrap(), + &sealed + ) + .is_err() + ); + assert!( + open_call( + &result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), + &sealed + ) + .is_err() + ); let eve = crypto::keypair(); - assert!(open_call(&request_key(&eve.secret, &buyer.public_hex, "job-1").unwrap(), &sealed).is_err()); - - let r = DeliveryResult { delivered: true, status: Some(200), attempts: 1, response_body: "ok".into(), ..Default::default() }; - let rs = seal_result(&result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), &r).unwrap(); - assert_eq!(open_result(&result_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(), &rs).unwrap(), r); - assert!(open_result(&bk, &rs).is_err(), "a result cannot be opened as a request"); + assert!( + open_call( + &request_key(&eve.secret, &buyer.public_hex, "job-1").unwrap(), + &sealed + ) + .is_err() + ); + + let r = DeliveryResult { + delivered: true, + status: Some(200), + attempts: 1, + response_body: "ok".into(), + ..Default::default() + }; + let rs = seal_result( + &result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), + &r, + ) + .unwrap(); + assert_eq!( + open_result( + &result_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(), + &rs + ) + .unwrap(), + r + ); + assert!( + open_result(&bk, &rs).is_err(), + "a result cannot be opened as a request" + ); } #[test] @@ -989,7 +1227,11 @@ mod tests { // JSON escaping can blow a body under the cap past one message. c.body = "\u{1}".repeat(MAX_BODY / 2); assert!(seal_call(&key, &c).is_err()); - let r = DeliveryResult { status: Some(200), response_body: "\u{1}".repeat(MAX_RESPONSE_BODY), ..Default::default() }; + let r = DeliveryResult { + status: Some(200), + response_body: "\u{1}".repeat(MAX_RESPONSE_BODY), + ..Default::default() + }; let s = seal_result(&key, &r).unwrap(); assert!(s.len() <= MAX_SEALED); assert!(open_result(&key, &s).unwrap().truncated); @@ -998,13 +1240,23 @@ mod tests { #[test] fn call_validation() { assert!(call("https://example.com/h").validate().is_ok()); - for bad in ["ftp://example.com/", "file:///etc/passwd", "https://user:pw@example.com/", "not a url"] { + for bad in [ + "ftp://example.com/", + "file:///etc/passwd", + "https://user:pw@example.com/", + "not a url", + ] { assert!(call(bad).validate().is_err(), "{bad}"); } let mut c = call("https://example.com/h"); c.method = "CONNECT".into(); assert!(c.validate().is_err()); - for h in ["Host", "content-length", "webhook-signature", "Transfer-Encoding"] { + for h in [ + "Host", + "content-length", + "webhook-signature", + "Transfer-Encoding", + ] { let mut c = call("https://example.com/h"); c.headers = vec![(h.into(), "x".into())]; assert!(c.validate().is_err(), "{h}"); @@ -1031,10 +1283,18 @@ mod tests { ] { let r = deliver(&call(&url), &spec(3), "j", Guard::default()).await; assert!(!r.answered(), "{url} must not be reached"); - assert!(r.error.as_deref().unwrap_or("").starts_with("refused"), "{url}: {:?}", r.error); + assert!( + r.error.as_deref().unwrap_or("").starts_with("refused"), + "{url}: {:?}", + r.error + ); assert_eq!(r.attempts, 1, "{url}: a refusal is not retried"); } - assert_eq!(seen.hits.load(Ordering::SeqCst), 0, "nothing reached the local target"); + assert_eq!( + seen.hits.load(Ordering::SeqCst), + 0, + "nothing reached the local target" + ); } #[tokio::test] @@ -1044,8 +1304,16 @@ mod tests { assert_eq!(r.status, Some(302)); assert!(!r.delivered); assert_eq!(r.attempts, 1, "a 3xx is final"); - assert_eq!(seen.other_hits.load(Ordering::SeqCst), 0, "the redirect target was never fetched"); - assert!(r.response_headers.iter().any(|(k, v)| k == "location" && v == "/other")); + assert_eq!( + seen.other_hits.load(Ordering::SeqCst), + 0, + "the redirect target was never fetched" + ); + assert!( + r.response_headers + .iter() + .any(|(k, v)| k == "location" && v == "/other") + ); } #[tokio::test] @@ -1055,7 +1323,10 @@ mod tests { c.secret = Some("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw".into()); c.id = Some("msg_1".into()); let r = deliver(&c, &spec(3), "j", LOCAL).await; - assert!(r.delivered && r.status == Some(200) && r.attempts == 1, "{r:?}"); + assert!( + r.delivered && r.status == Some(200) && r.attempts == 1, + "{r:?}" + ); assert_eq!(r.response_body, "thanks"); assert_eq!(r.webhook_id.as_deref(), Some("msg_1")); let (h, body) = seen.last.lock().unwrap().clone().unwrap(); @@ -1065,15 +1336,26 @@ mod tests { assert_eq!(h["webhook-id"], "msg_1"); let ts: u64 = h["webhook-timestamp"].to_str().unwrap().parse().unwrap(); // Verify the way a receiver does, with an independent HMAC. - let mut mac = as Mac>::new_from_slice(&B64.decode("MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw").unwrap()).unwrap(); + let mut mac = as Mac>::new_from_slice( + &B64.decode("MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw").unwrap(), + ) + .unwrap(); mac.update(format!("msg_1.{ts}.{}", c.body).as_bytes()); - assert_eq!(h["webhook-signature"].to_str().unwrap(), format!("v1,{}", B64.encode(mac.finalize().into_bytes()))); + assert_eq!( + h["webhook-signature"].to_str().unwrap(), + format!("v1,{}", B64.encode(mac.finalize().into_bytes())) + ); } #[test] fn signature_matches_the_standard_webhooks_reference_vector() { // From the Standard Webhooks spec / svix reference libraries. - let sig = sign("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", "msg_p5jXN8AQM9LWM0D4loKWxJek", 1614265330, br#"{"test": 2432232314}"#); + let sig = sign( + "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", + "msg_p5jXN8AQM9LWM0D4loKWxJek", + 1614265330, + br#"{"test": 2432232314}"#, + ); assert_eq!(sig, "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE="); } @@ -1086,11 +1368,24 @@ mod tests { assert_eq!(seen.hits.load(Ordering::SeqCst), 3); let r = deliver(&call(&format!("{base}/bad")), &spec(3), "j", LOCAL).await; - assert_eq!((r.attempts, r.status), (1, Some(400)), "a 4xx is not retried"); + assert_eq!( + (r.attempts, r.status), + (1, Some(400)), + "a 4xx is not retried" + ); assert_eq!(seen.hits.load(Ordering::SeqCst), 4); - let r = deliver(&call(&format!("{base}/fail")), &spec(MAX_ATTEMPTS + 1), "j", LOCAL).await; - assert!(r.error.unwrap().contains("max_attempts"), "over the cap is refused outright"); + let r = deliver( + &call(&format!("{base}/fail")), + &spec(MAX_ATTEMPTS + 1), + "j", + LOCAL, + ) + .await; + assert!( + r.error.unwrap().contains("max_attempts"), + "over the cap is refused outright" + ); assert_eq!(seen.hits.load(Ordering::SeqCst), 4); } @@ -1114,19 +1409,32 @@ mod tests { let mut s = spec(1); s.client_pubkey = "zz".into(); assert!(s.validate().is_err()); - let s = WebhookSpec { v: 1, client_pubkey: "11".repeat(32), max_attempts: 3, timeout_ms: 10_000, backoff_ms: 1_000 }; + let s = WebhookSpec { + v: 1, + client_pubkey: "11".repeat(32), + max_attempts: 3, + timeout_ms: 10_000, + backoff_ms: 1_000, + }; assert_eq!(s.budget_secs(), 30 + 1 + 2); assert_eq!(backoff(10_000, 5), Duration::from_millis(BACKOFF_CEIL_MS)); } async fn node(dir: &std::path::Path, boot: Vec) -> Arc { - let cfg = c0mpute_net::NetworkConfig::for_dir(dir.to_path_buf()).with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]).with_bootstrap(boot); + let cfg = c0mpute_net::NetworkConfig::for_dir(dir.to_path_buf()) + .with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]) + .with_bootstrap(boot); Arc::new(Libp2pNetwork::spawn(cfg).await.unwrap()) } async fn dial_addr(net: &Libp2pNetwork) -> libp2p::Multiaddr { for _ in 0..50 { - if let Some(a) = net.listen_addrs().await.ok().and_then(|v| v.into_iter().next()) { + if let Some(a) = net + .listen_addrs() + .await + .ok() + .and_then(|v| v.into_iter().next()) + { return a.with(libp2p::multiaddr::Protocol::P2p(net.peer_id())); } tokio::time::sleep(Duration::from_millis(50)).await; @@ -1149,7 +1457,15 @@ mod tests { let load = WorkerLoad::new(Arc::new(Semaphore::new(4)), 4); let tx = spawn_runner(worker.clone(), load.clone(), LOCAL); - crate::dispatch::run_worker_subscriber_with_load(worker.clone(), WORKLOAD.into(), vec![TAG.into()], Some(tx), Some(offer_filter(DEFAULT_PRICE_USD)), Some(pricing(DEFAULT_PRICE_USD)), Some(load)); + crate::dispatch::run_worker_subscriber_with_load( + worker.clone(), + WORKLOAD.into(), + vec![TAG.into()], + Some(tx), + Some(offer_filter(DEFAULT_PRICE_USD)), + Some(pricing(DEFAULT_PRICE_USD)), + Some(load), + ); let topic = job_topic(WORKLOAD); bystander.subscribe(&topic).await.unwrap(); let mut spy = bystander.messages(); @@ -1165,14 +1481,34 @@ mod tests { let secret = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"; // gitleaks:allow (public Standard Webhooks spec test key) let mut c = call(&format!("{base}/ok")); c.secret = Some(secret.into()); - c.headers.push(("authorization".into(), "Bearer sk_live_topsecret".into())); - let req = SendRequest { call: c.clone(), max_attempts: 2, timeout_ms: 5_000, backoff_ms: 100, workers: vec![], max_price_usd: DEFAULT_MAX_PRICE_USD, bid_window: Duration::from_secs(3), external_ref: Some("coinpay:inv_e2e".into()) }; - let out = tokio::time::timeout(Duration::from_secs(90), send(buyer.clone(), &req)).await.expect("send finishes").expect("send succeeds"); + c.headers + .push(("authorization".into(), "Bearer sk_live_topsecret".into())); + let req = SendRequest { + call: c.clone(), + max_attempts: 2, + timeout_ms: 5_000, + backoff_ms: 100, + workers: vec![], + max_price_usd: DEFAULT_MAX_PRICE_USD, + bid_window: Duration::from_secs(3), + external_ref: Some("coinpay:inv_e2e".into()), + }; + let out = tokio::time::timeout(Duration::from_secs(90), send(buyer.clone(), &req)) + .await + .expect("send finishes") + .expect("send succeeds"); assert_eq!(out.worker_peer_id, worker.peer_id().to_base58()); assert_eq!(out.receipt_status, JobStatus::Completed); - assert!(out.result.delivered && out.result.status == Some(200), "{:?}", out.result); + assert!( + out.result.delivered && out.result.status == Some(200), + "{:?}", + out.result + ); assert_eq!(out.result.response_body, "thanks"); - assert_eq!(out.price_usd, DEFAULT_PRICE_USD, "paid the worker's price, not zero and not the cap"); + assert_eq!( + out.price_usd, DEFAULT_PRICE_USD, + "paid the worker's price, not zero and not the cap" + ); assert_eq!(seen.hits.load(Ordering::SeqCst), 1); let (h, _) = seen.last.lock().unwrap().clone().unwrap(); assert_eq!(h["authorization"], "Bearer sk_live_topsecret"); @@ -1180,9 +1516,22 @@ mod tests { tokio::time::sleep(Duration::from_millis(500)).await; let wire = String::from_utf8_lossy(&sniffed.lock().unwrap()).into_owned(); - assert!(wire.contains("webhook_request") && wire.contains("webhook_result"), "the bystander saw the sealed exchange"); - for leak in [base.as_str(), "sk_live_topsecret", secret, "x-tenant", "\"event\"", "thanks"] { - assert!(!wire.contains(leak), "plaintext {leak:?} leaked onto gossip"); + assert!( + wire.contains("webhook_request") && wire.contains("webhook_result"), + "the bystander saw the sealed exchange" + ); + for leak in [ + base.as_str(), + "sk_live_topsecret", + secret, + "x-tenant", + "\"event\"", + "thanks", + ] { + assert!( + !wire.contains(leak), + "plaintext {leak:?} leaked onto gossip" + ); } } @@ -1211,24 +1560,57 @@ mod tests { // Our bid is always our price, never the buyer's cap, never zero. let p = pricing(DEFAULT_PRICE_USD); assert_eq!(p.quote(0.01, 0.0), Some(DEFAULT_PRICE_USD)); - assert_eq!(p.quote(DEFAULT_PRICE_USD / 2.0, 0.0), None, "a cap under our price gets no bid"); + assert_eq!( + p.quote(DEFAULT_PRICE_USD / 2.0, 0.0), + None, + "a cap under our price gets no bid" + ); let f = offer_filter(DEFAULT_PRICE_USD); let good = serde_json::to_value(spec(3)).unwrap(); assert!(f(&offer(DEFAULT_MAX_PRICE_USD, good.clone()))); assert!(!f(&offer(0.0, good.clone())), "a free offer gets no bid"); assert!(!f(&offer(f64::NAN, good.clone()))); - assert!(!f(&offer(1.0, serde_json::to_value(spec(9)).unwrap())), "an invalid spec gets no bid"); - assert!(!f(&offer(1.0, serde_json::json!({"url": "http://169.254.169.254/"}))), "plaintext calls are not a webhook spec"); + assert!( + !f(&offer(1.0, serde_json::to_value(spec(9)).unwrap())), + "an invalid spec gets no bid" + ); + assert!( + !f(&offer( + 1.0, + serde_json::json!({"url": "http://169.254.169.254/"}) + )), + "plaintext calls are not a webhook spec" + ); let mut short = offer(1.0, good); short.deadline_unix_ms = short.published_at_ms + 1_000; - assert!(!f(&short), "a deadline shorter than the retry budget gets no bid"); + assert!( + !f(&short), + "a deadline shorter than the retry budget gets no bid" + ); } #[test] 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 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()), + 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(); @@ -1241,6 +1623,9 @@ mod tests { assert!(bid.matches(&a, 0).is_err(), "over the cap"); let mut a = accept; a.buyer_peer_id = "mallory".into(); - assert!(bid.matches(&a, 0).is_err(), "only the offer's buyer accepts"); + assert!( + bid.matches(&a, 0).is_err(), + "only the offer's buyer accepts" + ); } } diff --git a/node/crates/c0mpute-net/src/topics.rs b/node/crates/c0mpute-net/src/topics.rs index ad9ea8d9..bf58cb3c 100644 --- a/node/crates/c0mpute-net/src/topics.rs +++ b/node/crates/c0mpute-net/src/topics.rs @@ -194,6 +194,12 @@ pub struct JobAccept { /// [`JobOffer::external_ref`]. #[serde(default, skip_serializing_if = "Option::is_none")] pub external_ref: Option, + /// The buyer's publicly dialable multiaddrs (`.../p2p/`). + /// A winning worker behind NAT dials one of these before publishing its + /// receipt, so the buyer can fetch the result over that connection; the + /// buyer usually cannot dial a worker's random, firewalled port. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub buyer_addrs: Vec, } /// Worker's completion receipt. Includes the output hash so the buyer @@ -239,7 +245,11 @@ mod tests { assert_eq!(offer.external_ref, None); // And a new offer omits the field entirely when unset, so the bytes // an old node sees are unchanged. - assert!(!serde_json::to_string(&offer).unwrap().contains("external_ref")); + assert!( + !serde_json::to_string(&offer) + .unwrap() + .contains("external_ref") + ); } #[test] diff --git a/package.json b/package.json index 1e1c82ec..7199b696 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute-monorepo", - "version": "0.2.65", + "version": "0.2.66", "private": true, "workspaces": [ "apps/*", diff --git a/packages/shared/package.json b/packages/shared/package.json index e563b77a..c5358510 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/shared", - "version": "0.2.65", + "version": "0.2.66", "private": true, "type": "module", "main": "./dist/index.js", diff --git a/plugins/ad-generator-agent/module.toml b/plugins/ad-generator-agent/module.toml index 6407daa5..15ccd5b1 100644 --- a/plugins/ad-generator-agent/module.toml +++ b/plugins/ad-generator-agent/module.toml @@ -21,7 +21,7 @@ [module] id = "ad-generator-agent" name = "Ad Generator Agent — every ad type from one URL" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "Turns a landing-page URL into a complete ad set (display banners, text link, terminal, feed item, image, GIF, 5-second video pre-roll and audio) by buying c0mpute work: copy from Infernet inference, voiceover from CPU text-to-speech, artwork from diffusion when workers offer it. No paid AI API by default." diff --git a/plugins/ads-manager/module.toml b/plugins/ads-manager/module.toml index e7cd3a97..51081b93 100644 --- a/plugins/ads-manager/module.toml +++ b/plugins/ads-manager/module.toml @@ -18,7 +18,7 @@ [module] id = "ads-manager" name = "Ads Manager — shared ad operations layer" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "Unified ad operations for the c0mpute network: campaign management, creative registry, publisher inventory, OpenRTB auction primitives, CoinPay escrow, and reporting. Required dependency for ctv-ads and all future ad format plugins." diff --git a/plugins/coinpay/module.toml b/plugins/coinpay/module.toml index 384df7a5..d61cbc62 100644 --- a/plugins/coinpay/module.toml +++ b/plugins/coinpay/module.toml @@ -7,7 +7,7 @@ [module] id = "coinpay" name = "CoinPay — DID, payments, escrow, reputation" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "Decentralized identifiers, wallet linking, escrow, signed receipts, and worker reputation. The trust spine of the c0mpute network." diff --git a/plugins/crawl/module.toml b/plugins/crawl/module.toml index 8a1925e6..1b0e1e8c 100644 --- a/plugins/crawl/module.toml +++ b/plugins/crawl/module.toml @@ -14,7 +14,7 @@ [module] id = "crawl" name = "Crawl — distributed web fetching" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "data" description = "Decentralized web crawling and page-rendering pool. Static fetches and JS-rendered pages, robots-aware, WARC-capable." diff --git a/plugins/ctv-ads/module.toml b/plugins/ctv-ads/module.toml index cf9af9a5..5e4679c6 100644 --- a/plugins/ctv-ads/module.toml +++ b/plugins/ctv-ads/module.toml @@ -20,7 +20,7 @@ [module] id = "ctv-ads" name = "CTV Ads — Connected TV ad marketplace" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "Decentralised Connected TV advertising on the c0mpute network. VAST 4.x endpoint, OpenRTB real-time auction, cryptographically verified impressions, CoinPay settlement. Supports Roku, Fire TV, Apple TV, Smart TVs, and FAST channels. Live-stream integration for ad-monetised broadcasts." diff --git a/plugins/diffusion/module.toml b/plugins/diffusion/module.toml index 105e7ed3..003704c9 100644 --- a/plugins/diffusion/module.toml +++ b/plugins/diffusion/module.toml @@ -11,7 +11,7 @@ [module] id = "diffusion" name = "Diffusion — image generation" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Decentralized text-to-image on plain CPUs: Z-Image-Turbo through stable-diffusion.cpp, provisioned by every node with the RAM for it." diff --git a/plugins/embed/module.toml b/plugins/embed/module.toml index 141085ee..7bcb5ab9 100644 --- a/plugins/embed/module.toml +++ b/plugins/embed/module.toml @@ -13,7 +13,7 @@ [module] id = "embed" name = "Embeddings — vector generation" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Bulk text-embedding generation (sentence-transformers, BGE, Nomic) with CPU fallback and optional GPU acceleration." diff --git a/plugins/gameserver/module.toml b/plugins/gameserver/module.toml index 1371728d..b8596fda 100644 --- a/plugins/gameserver/module.toml +++ b/plugins/gameserver/module.toml @@ -14,7 +14,7 @@ [module] id = "gameserver" name = "Game server — Minecraft / Factorio / Valheim" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "infrastructure" description = "Continuous-reservation hosting for game servers. Multi-replica placement, persistent world state on c0mpute storage, proof-of-serve settlement." diff --git a/plugins/hosting/module.toml b/plugins/hosting/module.toml index db3c90f7..47e044c0 100644 --- a/plugins/hosting/module.toml +++ b/plugins/hosting/module.toml @@ -12,7 +12,7 @@ [module] id = "hosting" name = "Hosting — censorship-resistant static sites" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "data" description = "Static-site hosting with multi-replica placement, mutable naming, epoch settlement, and slashable proof-of-serve. See DIP-0015." diff --git a/plugins/infernet/module.toml b/plugins/infernet/module.toml index 4a62af1a..8a21aa16 100644 --- a/plugins/infernet/module.toml +++ b/plugins/infernet/module.toml @@ -10,7 +10,7 @@ [module] id = "infernet" name = "Infernet Protocol — AI inference" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Decentralized AI inference workloads on c0mpute. Runs models on consumer/prosumer GPUs with model-hash and runtime-image verification. Also trains and serves temporal policies (CfC/LTC liquid networks, GRU baseline) and fine-tunes language models (LoRA) with live progress, on CPU or GPU." diff --git a/plugins/live-stream/module.toml b/plugins/live-stream/module.toml index b3836369..de07dfbc 100644 --- a/plugins/live-stream/module.toml +++ b/plugins/live-stream/module.toml @@ -23,7 +23,7 @@ [module] id = "live-stream" name = "Live Stream — p2p HLS live streaming" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "media" description = "Decentralized live video streaming via BitTorrent-style HLS segment swarming. Broadcaster pushes RTMP/SRT; c0mpute nodes swarm segments to viewers. No central CDN. Censorship-resistant. Per-GB micropayments via CoinPay." diff --git a/plugins/ocr/module.toml b/plugins/ocr/module.toml index 62c19543..48df133f 100644 --- a/plugins/ocr/module.toml +++ b/plugins/ocr/module.toml @@ -12,7 +12,7 @@ [module] id = "ocr" name = "OCR — document text extraction" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Decentralized OCR for PDFs and scanned documents (Tesseract, PaddleOCR, docTR) with CPU or GPU runners." diff --git a/plugins/podcast-ads/module.toml b/plugins/podcast-ads/module.toml index 3fe9b393..c88da7e3 100644 --- a/plugins/podcast-ads/module.toml +++ b/plugins/podcast-ads/module.toml @@ -22,7 +22,7 @@ [module] id = "podcast-ads" name = "Podcast Ads — audio ad marketplace" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "Decentralised audio advertising for c0mpute-hosted podcasts. DAAST 1.0 endpoint, dynamic ad insertion, Podcasting 2.0 revenue splits, and CoinPay settlement. Publishers sell ad slots; advertisers run campaigns — zero platform cut beyond protocol split." diff --git a/plugins/podcasting/module.toml b/plugins/podcasting/module.toml index a5f47c0a..bb77998a 100644 --- a/plugins/podcasting/module.toml +++ b/plugins/podcasting/module.toml @@ -20,7 +20,7 @@ [module] id = "podcasting" name = "Podcasting — p2p podcast publishing" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "media" description = "Censorship-resistant podcast hosting on the c0mpute network. Upload episodes, stream live shows, auto-transcribe, and monetise via CoinPay subscriptions, pay-per-episode, value-for-value, and audio ads — with zero platform cut. RSS 2.0 + Podcasting 2.0 compatible with all major podcast apps." diff --git a/plugins/podcasts/module.toml b/plugins/podcasts/module.toml index bd332708..9df2636e 100644 --- a/plugins/podcasts/module.toml +++ b/plugins/podcasts/module.toml @@ -19,7 +19,7 @@ [module] id = "podcasts" name = "Podcasts — p2p podcast listener" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "media" description = "Discover, subscribe to, and play podcasts hosted on the c0mpute network or any standard RSS feed. Stream live episodes, tip creators directly via CoinPay, and participate in value-for-value streaming payments. Works alongside any existing podcast app via OPML export and local RSS proxy." diff --git a/plugins/render/module.toml b/plugins/render/module.toml index d44cabb4..bbf6162d 100644 --- a/plugins/render/module.toml +++ b/plugins/render/module.toml @@ -13,7 +13,7 @@ [module] id = "render" name = "Render — Blender 3D rendering" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "media" description = "Decentralized 3D rendering using Blender headless (Cycles/Eevee) across consumer GPUs and CPUs." diff --git a/plugins/secure-chat/module.toml b/plugins/secure-chat/module.toml index c7c92af6..39bb771b 100644 --- a/plugins/secure-chat/module.toml +++ b/plugins/secure-chat/module.toml @@ -18,7 +18,7 @@ [module] id = "secure-chat" name = "Secure Chat — E2E encrypted p2p messaging" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "applications" description = "End-to-end encrypted peer-to-peer messaging across the c0mpute network. No central server. Keys stay with the user. Identity and reputation via CoinPay DID. Group chats, sealed-sender mode, and offline store-and-forward included." diff --git a/plugins/storage/module.toml b/plugins/storage/module.toml index 2cacb309..d26f4d58 100644 --- a/plugins/storage/module.toml +++ b/plugins/storage/module.toml @@ -12,7 +12,7 @@ [module] id = "storage" name = "Storage — Reed-Solomon shard service" -version = "0.2.65" +version = "0.2.66" kind = "service" category = "data" description = "Reed-Solomon 10/14 shard storage with auto-repair. Compute-locality, E2E encryption, content-addressed, pay-as-you-go. See DIP-0012." diff --git a/plugins/train/module.toml b/plugins/train/module.toml index 23a7158c..b4ca6f35 100644 --- a/plugins/train/module.toml +++ b/plugins/train/module.toml @@ -13,7 +13,7 @@ [module] id = "train" name = "Fine-tune — LoRA/QLoRA training" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Decentralized model fine-tuning (LoRA, QLoRA) on consumer/prosumer GPUs using axolotl or unsloth." diff --git a/plugins/transcode/module.toml b/plugins/transcode/module.toml index 2df63d14..aba9ffb8 100644 --- a/plugins/transcode/module.toml +++ b/plugins/transcode/module.toml @@ -7,7 +7,7 @@ [module] id = "transcode" name = "FFmpeg Transcoding" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "media" description = "Decentralized audio/video transcoding via FFmpeg with hardware acceleration (NVENC, QSV, AMF, VideoToolbox, AV1)." diff --git a/plugins/tts/module.toml b/plugins/tts/module.toml index 35edc264..aae10d76 100644 --- a/plugins/tts/module.toml +++ b/plugins/tts/module.toml @@ -15,7 +15,7 @@ [module] id = "tts" name = "TTS — text-to-speech" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "CPU text-to-speech inference over c0mpute using Piper, Kokoro, or Pocket TTS." diff --git a/plugins/video/module.toml b/plugins/video/module.toml index d66f5fd8..9695e925 100644 --- a/plugins/video/module.toml +++ b/plugins/video/module.toml @@ -44,7 +44,7 @@ [module] id = "video" name = "Video — generation (Wan 2.2, LTX-2.3 with audio)" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "Decentralized video generation on consumer/prosumer GPUs: Wan 2.2 text/image-to-video, and LTX-2.3 video with a synchronized soundtrack." diff --git a/plugins/webhook/module.toml b/plugins/webhook/module.toml index 0b14a5d0..422400fa 100644 --- a/plugins/webhook/module.toml +++ b/plugins/webhook/module.toml @@ -7,7 +7,7 @@ [module] id = "webhook" name = "Webhooks — paid, end-to-end encrypted delivery" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "infrastructure" description = "Send webhooks from c0mpute workers, paid per delivery. The URL, headers, body and signing secret are sealed to the one worker that wins the job; relays and the market see only ciphertext. Standard Webhooks signatures, bounded retries with backoff, SSRF-guarded delivery (public addresses only, no redirects)." diff --git a/plugins/whisper/module.toml b/plugins/whisper/module.toml index ea3fd748..64fcc3e5 100644 --- a/plugins/whisper/module.toml +++ b/plugins/whisper/module.toml @@ -7,7 +7,7 @@ [module] id = "whisper" name = "Whisper — speech-to-text" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "ai" description = "CPU speech-to-text inference over c0mpute using faster-whisper." diff --git a/plugins/workflow-step/module.toml b/plugins/workflow-step/module.toml index e1e0f930..bb92c224 100644 --- a/plugins/workflow-step/module.toml +++ b/plugins/workflow-step/module.toml @@ -14,7 +14,7 @@ [module] id = "workflow-step" name = "Workflow step — webhook workflow steps (meshhook)" -version = "0.2.65" +version = "0.2.66" kind = "workload" category = "infrastructure" description = "Runs one step of a webhook workflow on a c0mpute worker: a JMESPath transform or a guarded HTTP call. Applications reach it through `c0mpute jobs serve`." From 970ae7c5099357eb91f4cf4b312f1c369f4fdc53 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Sun, 8 Nov 2026 17:03:10 +0000 Subject: [PATCH 2/2] chore: keep the diff to the reachability change (undo rustfmt churn) Co-Authored-By: Claude Opus 5.5 --- node/crates/c0mpute-core/src/webhook.rs | 577 ++++-------------------- node/crates/c0mpute-net/src/topics.rs | 6 +- 2 files changed, 97 insertions(+), 486 deletions(-) diff --git a/node/crates/c0mpute-core/src/webhook.rs b/node/crates/c0mpute-core/src/webhook.rs index 6f3dbd52..5c45702e 100644 --- a/node/crates/c0mpute-core/src/webhook.rs +++ b/node/crates/c0mpute-core/src/webhook.rs @@ -153,9 +153,7 @@ impl WebhookSpec { /// Worst-case seconds the worker spends on this delivery. pub fn budget_secs(&self) -> u64 { let tries = u64::from(self.max_attempts); - let waits: u64 = (1..tries) - .map(|n| backoff(self.backoff_ms, n as u32).as_millis() as u64) - .sum(); + let waits: u64 = (1..tries).map(|n| backoff(self.backoff_ms, n as u32).as_millis() as u64).sum(); (tries * u64::from(self.timeout_ms) + waits).div_ceil(1000) } } @@ -217,10 +215,8 @@ impl WebhookCall { let mut bytes = 0; for (k, v) in &self.headers { bytes += k.len() + v.len(); - let name = reqwest::header::HeaderName::from_bytes(k.as_bytes()) - .with_context(|| format!("bad header name {k:?}"))?; - reqwest::header::HeaderValue::from_str(v) - .with_context(|| format!("bad value for header {k}"))?; + let name = reqwest::header::HeaderName::from_bytes(k.as_bytes()).with_context(|| format!("bad header name {k:?}"))?; + reqwest::header::HeaderValue::from_str(v).with_context(|| format!("bad value for header {k}"))?; if RESERVED_HEADERS.contains(&name.as_str()) { bail!("header {k} is set by the worker"); } @@ -228,11 +224,7 @@ impl WebhookCall { if bytes > MAX_HEADER_BYTES { bail!("headers larger than {MAX_HEADER_BYTES} bytes"); } - if self - .secret - .as_deref() - .is_some_and(|s| s.is_empty() || s.len() > MAX_SECRET) - { + if self.secret.as_deref().is_some_and(|s| s.is_empty() || s.len() > MAX_SECRET) { bail!("secret must be 1..={MAX_SECRET} bytes"); } if let Some(id) = &self.id { @@ -289,10 +281,7 @@ impl DeliveryResult { } fn refused(e: impl std::fmt::Display) -> Self { - Self { - error: Some(format!("refused: {e}")), - ..Default::default() - } + Self { error: Some(format!("refused: {e}")), ..Default::default() } } } @@ -330,28 +319,17 @@ pub struct WebhookSealedResult { /// One key per direction, bound to the job: a sealed result can never be /// replayed as a request, nor a message moved to another job. pub fn request_key(secret: &StaticSecret, peer_pub_hex: &str, job_id: &str) -> Result<[u8; 32]> { - crypto::session_key( - secret, - peer_pub_hex, - &format!("c0mpute.webhook/v1/{job_id}/request"), - ) + crypto::session_key(secret, peer_pub_hex, &format!("c0mpute.webhook/v1/{job_id}/request")) } pub fn result_key(secret: &StaticSecret, peer_pub_hex: &str, job_id: &str) -> Result<[u8; 32]> { - crypto::session_key( - secret, - peer_pub_hex, - &format!("c0mpute.webhook/v1/{job_id}/result"), - ) + crypto::session_key(secret, peer_pub_hex, &format!("c0mpute.webhook/v1/{job_id}/result")) } pub fn seal_call(key: &[u8; 32], call: &WebhookCall) -> Result { let sealed = crypto::seal(key, &serde_json::to_vec(call)?)?; if sealed.len() > MAX_SEALED { - bail!( - "sealed request is {} bytes; one delivery carries at most {MAX_SEALED} (shrink the body or headers)", - sealed.len() - ); + bail!("sealed request is {} bytes; one delivery carries at most {MAX_SEALED} (shrink the body or headers)", sealed.len()); } Ok(sealed) } @@ -404,10 +382,7 @@ fn floor_char_boundary(s: &str, mut i: usize) -> usize { } fn unix_secs() -> u64 { - std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map(|d| d.as_secs()) - .unwrap_or(0) + std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs()).unwrap_or(0) } // ── Standard Webhooks signature ───────────────────────────────────────── @@ -415,16 +390,12 @@ fn unix_secs() -> u64 { /// The signing key: `whsec_` decodes to bytes (Standard Webhooks); /// anything else is used as raw bytes. fn signing_key(secret: &str) -> Vec { - secret - .strip_prefix("whsec_") - .and_then(|b| B64.decode(b).ok()) - .unwrap_or_else(|| secret.as_bytes().to_vec()) + secret.strip_prefix("whsec_").and_then(|b| B64.decode(b).ok()).unwrap_or_else(|| secret.as_bytes().to_vec()) } /// `webhook-signature` value: `v1,`. pub fn sign(secret: &str, id: &str, timestamp: u64, body: &[u8]) -> String { - let mut mac = as Mac>::new_from_slice(&signing_key(secret)) - .expect("hmac accepts any key length"); + let mut mac = as Mac>::new_from_slice(&signing_key(secret)).expect("hmac accepts any key length"); mac.update(id.as_bytes()); mac.update(b"."); mac.update(timestamp.to_string().as_bytes()); @@ -449,18 +420,12 @@ async fn resolve_public(u: &url::Url, guard: Guard) -> Result<(String, Vec = tokio::time::timeout( - Duration::from_secs(10), - tokio::net::lookup_host((lookup.as_str(), port)), - ) - .await - .map_err(|_| anyhow!("resolving {host} timed out"))? - .with_context(|| format!("resolve {host}"))? - .collect(); + let lookup = host.trim_start_matches('[').trim_end_matches(']').to_string(); + let addrs: Vec = tokio::time::timeout(Duration::from_secs(10), tokio::net::lookup_host((lookup.as_str(), port))) + .await + .map_err(|_| anyhow!("resolving {host} timed out"))? + .with_context(|| format!("resolve {host}"))? + .collect(); if addrs.is_empty() { bail!("{host} did not resolve"); } @@ -482,12 +447,7 @@ fn retryable(status: Option) -> bool { /// Send `call` with up to `spec.max_attempts` tries. Never fails: a refused /// or unreachable target is reported in the result. -pub async fn deliver( - call: &WebhookCall, - spec: &WebhookSpec, - default_id: &str, - guard: Guard, -) -> DeliveryResult { +pub async fn deliver(call: &WebhookCall, spec: &WebhookSpec, default_id: &str, guard: Guard) -> DeliveryResult { if let Err(e) = call.validate().and_then(|_| spec.validate()) { return DeliveryResult::refused(e); } @@ -495,15 +455,9 @@ pub async fn deliver( Ok(u) => u, Err(e) => return DeliveryResult::refused(e), }; - let id = call - .id - .clone() - .unwrap_or_else(|| format!("msg_{default_id}")); + let id = call.id.clone().unwrap_or_else(|| format!("msg_{default_id}")); let started = Instant::now(); - let mut out = DeliveryResult { - webhook_id: call.secret.as_ref().map(|_| id.clone()), - ..Default::default() - }; + let mut out = DeliveryResult { webhook_id: call.secret.as_ref().map(|_| id.clone()), ..Default::default() }; for n in 1..=spec.max_attempts { if n > 1 { tokio::time::sleep(backoff(spec.backoff_ms, n - 1)).await; @@ -516,11 +470,7 @@ pub async fn deliver( // A refusal is final; retrying cannot make the address public. out.attempts = n; out.error = Some(format!("refused: {e:#}")); - out.log.push(Attempt { - status: None, - error: out.error.clone(), - latency_ms: 0, - }); + out.log.push(Attempt { status: None, error: out.error.clone(), latency_ms: 0 }); break; } Ok((host, addrs)) => send_once(call, spec, &u, &host, &addrs, &id).await, @@ -536,20 +486,12 @@ pub async fn deliver( out.response_body = r.body; out.truncated = r.truncated; out.error = None; - out.log.push(Attempt { - status: Some(r.status), - error: None, - latency_ms, - }); + out.log.push(Attempt { status: Some(r.status), error: None, latency_ms }); } Err(e) => { let msg = format!("{e:#}"); out.error = Some(msg.clone()); - out.log.push(Attempt { - status: None, - error: Some(msg), - latency_ms, - }); + out.log.push(Attempt { status: None, error: Some(msg), latency_ms }); } } if !retryable(out.log.last().and_then(|a| a.status)) { @@ -567,14 +509,7 @@ struct Answer { truncated: bool, } -async fn send_once( - call: &WebhookCall, - spec: &WebhookSpec, - u: &url::Url, - host: &str, - addrs: &[std::net::SocketAddr], - id: &str, -) -> Result { +async fn send_once(call: &WebhookCall, spec: &WebhookSpec, u: &url::Url, host: &str, addrs: &[std::net::SocketAddr], id: &str) -> Result { let timeout = Duration::from_millis(u64::from(spec.timeout_ms)); let client = reqwest::Client::builder() .redirect(reqwest::redirect::Policy::none()) @@ -599,17 +534,12 @@ async fn send_once( req = req .header("webhook-id", id) .header("webhook-timestamp", ts.to_string()) - .header( - "webhook-signature", - sign(secret, id, ts, call.body.as_bytes()), - ); + .header("webhook-signature", sign(secret, id, ts, call.body.as_bytes())); } if !call.body.is_empty() { req = req.body(call.body.clone()); } - let mut resp = tokio::time::timeout(timeout, req.send()) - .await - .map_err(|_| anyhow!("timed out after {}ms", spec.timeout_ms))??; + let mut resp = tokio::time::timeout(timeout, req.send()).await.map_err(|_| anyhow!("timed out after {}ms", spec.timeout_ms))??; let status = resp.status().as_u16(); let headers = resp .headers() @@ -640,12 +570,7 @@ async fn send_once( if tokio::time::timeout(timeout, read).await.is_err() { truncated = true; } - Ok(Answer { - status, - headers, - body: String::from_utf8_lossy(&body).into_owned(), - truncated, - }) + Ok(Answer { status, headers, body: String::from_utf8_lossy(&body).into_owned(), truncated }) } // ── worker ────────────────────────────────────────────────────────────── @@ -660,11 +585,7 @@ pub fn validate_price(price_usd: f64) -> Result<()> { /// The listing every webhook bid is priced with: a fixed price per delivery. pub fn pricing(price_usd: f64) -> PricingConfig { - PricingConfig { - mode: PricingMode::Fixed, - min_usd: price_usd, - fixed_usd: Some(price_usd), - } + PricingConfig { mode: PricingMode::Fixed, min_usd: price_usd, fixed_usd: Some(price_usd) } } /// Bid only on valid webhook offers whose cap covers our price. @@ -677,9 +598,7 @@ pub fn offer_filter(price_usd: f64) -> OfferFilter { .as_ref() .and_then(|v| serde_json::from_value::(v.clone()).ok()) .is_some_and(|s| { - let deadline_ok = offer.deadline_unix_ms.saturating_sub(offer.published_at_ms) - / 1000 - >= s.budget_secs(); + let deadline_ok = offer.deadline_unix_ms.saturating_sub(offer.published_at_ms) / 1000 >= s.budget_secs(); s.validate().is_ok() && deadline_ok }) }) @@ -694,12 +613,7 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re let topic = job_topic(WORKLOAD); let me = net.peer_id().to_base58(); let mut rx = net.messages(); - let spec: WebhookSpec = serde_json::from_value( - accept - .spec_inline - .clone() - .context("accept lacks spec_inline")?, - )?; + let spec: WebhookSpec = serde_json::from_value(accept.spec_inline.clone().context("accept lacks spec_inline")?)?; spec.validate()?; let kp = crypto::keypair(); let req_key = request_key(&kp.secret, &spec.client_pubkey, &accept.job_id)?; @@ -732,22 +646,14 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re if msg.topic != topic { continue; } - let Ok(r) = serde_json::from_slice::(&msg.data) else { - continue; - }; + let Ok(r) = serde_json::from_slice::(&msg.data) else { continue }; // Only the buyer that won us may hand us the request. - if r.t != "webhook_request" - || r.job_id != accept.job_id - || r.worker_peer_id != me - || !from(&msg, &accept.buyer_peer_id) - { + if r.t != "webhook_request" || r.job_id != accept.job_id || r.worker_peer_id != me || !from(&msg, &accept.buyer_peer_id) { continue; } match open_call(&req_key, &r.sealed) { Ok(c) => break c, - Err(e) => { - debug!(job_id = %accept.job_id, err = %e, "webhook: unreadable request ignored") - } + Err(e) => debug!(job_id = %accept.job_id, err = %e, "webhook: unreadable request ignored"), } }; @@ -755,19 +661,10 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re drop(call); let sealed = seal_result(&res_key, &result)?; let hash = sha256_hex(sealed.as_bytes()); - let msg = serde_json::to_vec(&WebhookSealedResult { - t: "webhook_result".into(), - job_id: accept.job_id.clone(), - worker_peer_id: me.clone(), - sealed, - })?; + let msg = serde_json::to_vec(&WebhookSealedResult { t: "webhook_result".into(), job_id: accept.job_id.clone(), worker_peer_id: me.clone(), sealed })?; let _ = net.publish(&topic, msg.clone()).await; tokio::time::sleep(Duration::from_millis(50)).await; - let status = if result.answered() { - JobStatus::Completed - } else { - JobStatus::Failed - }; + let status = if result.answered() { JobStatus::Completed } else { JobStatus::Failed }; info!(job_id = %accept.job_id, status = ?result.status, attempts = result.attempts, delivered = result.delivered, "webhook: delivery done"); publish_receipt(&net, &accept, Some(hash), status).await; @@ -777,12 +674,7 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re tokio::spawn(async move { let linger = Instant::now() + RESULT_LINGER; loop { - let m = match tokio::time::timeout( - linger.saturating_duration_since(Instant::now()), - rx.recv(), - ) - .await - { + let m = match tokio::time::timeout(linger.saturating_duration_since(Instant::now()), rx.recv()).await { Ok(Ok(m)) => m, Ok(Err(broadcast::error::RecvError::Lagged(_))) => continue, _ => return, @@ -802,17 +694,11 @@ async fn run_job(net: Arc, accept: JobAccept, guard: Guard) -> Re /// Runner for `webhook.deliver`: each accepted job runs in its own task on /// one of the node's delivery slots. -pub fn spawn_runner( - net: Arc, - load: WorkerLoad, - guard: Guard, -) -> mpsc::Sender { +pub fn spawn_runner(net: Arc, load: WorkerLoad, guard: Guard) -> mpsc::Sender { let (tx, mut rx) = mpsc::channel::(64); tokio::spawn(async move { while let Some(job) = rx.recv().await { - let Ok(slot) = load.compute_slots().acquire_owned().await else { - return; - }; + let Ok(slot) = load.compute_slots().acquire_owned().await else { return }; let net = net.clone(); let load = load.clone(); tokio::spawn(async move { @@ -838,42 +724,20 @@ pub struct Serving { /// `c0mpute webhook serve`: advertise `webhook:deliver` and deliver paid, /// sealed webhooks, `concurrency` at a time. -pub async fn serve( - net_opts: &crate::temporal::node::NetOptions, - price_usd: f64, - concurrency: usize, -) -> Result { +pub async fn serve(net_opts: &crate::temporal::node::NetOptions, price_usd: f64, concurrency: usize) -> Result { validate_price(price_usd)?; let concurrency = concurrency.clamp(1, 256); let net = crate::temporal::node::spawn_node(net_opts).await?; let tags = vec![TAG.to_string()]; let listing = pricing(price_usd); let hardware = serde_json::json!({"version": env!("CARGO_PKG_VERSION"), "webhook": {"concurrency": concurrency}, "pricing": listing.listing()}); - tokio::spawn(crate::capabilities::advertise_loop( - net.clone(), - tags.clone(), - hardware, - crate::capabilities::local_bench_score(), - crate::capabilities::DEFAULT_ADVERTISE_INTERVAL, - )); + tokio::spawn(crate::capabilities::advertise_loop(net.clone(), tags.clone(), hardware, crate::capabilities::local_bench_score(), crate::capabilities::DEFAULT_ADVERTISE_INTERVAL)); let load = WorkerLoad::new(Arc::new(Semaphore::new(concurrency)), concurrency); let tx = spawn_runner(net.clone(), load.clone(), Guard::default()); - crate::dispatch::run_worker_subscriber_with_load( - net.clone(), - WORKLOAD.into(), - tags.clone(), - Some(tx), - Some(offer_filter(price_usd)), - Some(listing), - Some(load), - ); + crate::dispatch::run_worker_subscriber_with_load(net.clone(), WORKLOAD.into(), tags.clone(), Some(tx), Some(offer_filter(price_usd)), Some(listing), Some(load)); info!(peer = %net.peer_id().to_base58(), price_usd, concurrency, "webhook: serving"); tokio::time::sleep(Duration::from_millis(300)).await; - Ok(Serving { - net, - tags, - price_usd, - }) + Ok(Serving { net, tags, price_usd }) } // ── buyer ─────────────────────────────────────────────────────────────── @@ -909,13 +773,7 @@ pub async fn send(net: Arc, req: &SendRequest) -> Result, req: &SendRequest) -> Result, -) -> Result<()> { +async fn key_up(net: &Libp2pNetwork, topic: &str, secret: &StaticSecret, call: &WebhookCall, ready: WebhookReady, keys: &mut HashMap) -> Result<()> { let key = request_key(secret, &ready.worker_pubkey, &ready.job_id)?; let sealed = seal_call(&key, call)?; - let payload = serde_json::to_vec(&WebhookSealed { - t: "webhook_request".into(), - job_id: ready.job_id.clone(), - worker_peer_id: ready.worker_peer_id.clone(), - sealed, - })?; + let payload = serde_json::to_vec(&WebhookSealed { t: "webhook_request".into(), job_id: ready.job_id.clone(), worker_peer_id: ready.worker_peer_id.clone(), sealed })?; let _ = net.publish(topic, payload.clone()).await; debug!(job_id = %ready.job_id, worker = %ready.worker_peer_id, "webhook: sealed request sent"); - keys.insert( - ready.job_id, - Exchange { - worker_pubkey: ready.worker_pubkey, - payload, - }, - ); + keys.insert(ready.job_id, Exchange { worker_pubkey: ready.worker_pubkey, payload }); Ok(()) } @@ -1082,29 +905,14 @@ mod tests { use axum::http::{HeaderMap, StatusCode}; use axum::routing::any; - const LOCAL: Guard = Guard { - allow_private: true, - }; + const LOCAL: Guard = Guard { allow_private: true }; fn spec(attempts: u32) -> WebhookSpec { - WebhookSpec { - v: 1, - client_pubkey: "11".repeat(32), - max_attempts: attempts, - timeout_ms: 5_000, - backoff_ms: 100, - } + WebhookSpec { v: 1, client_pubkey: "11".repeat(32), max_attempts: attempts, timeout_ms: 5_000, backoff_ms: 100 } } fn call(url: &str) -> WebhookCall { - WebhookCall { - url: url.into(), - method: "POST".into(), - headers: vec![("x-tenant".into(), "acme".into())], - body: r#"{"event":"paid"}"#.into(), - secret: None, - id: None, - } + WebhookCall { url: url.into(), method: "POST".into(), headers: vec![("x-tenant".into(), "acme".into())], body: r#"{"event":"paid"}"#.into(), secret: None, id: None } } #[derive(Clone, Default)] @@ -1118,12 +926,7 @@ mod tests { /// to `/other`, `/big` 100 KiB. async fn target() -> (String, Seen) { let seen = Seen::default(); - async fn handle( - State(s): State, - uri: axum::http::Uri, - headers: HeaderMap, - body: axum::body::Bytes, - ) -> (StatusCode, HeaderMap, Vec) { + async fn handle(State(s): State, uri: axum::http::Uri, headers: HeaderMap, body: axum::body::Bytes) -> (StatusCode, HeaderMap, Vec) { let mut h = HeaderMap::new(); match uri.path() { "/other" => { @@ -1163,59 +966,18 @@ mod tests { let bk = request_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(); let wk = request_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(); let sealed = seal_call(&bk, &c).unwrap(); - assert!( - !sealed.contains("example.com") && !sealed.contains("whsec_"), - "ciphertext must not leak the call" - ); + assert!(!sealed.contains("example.com") && !sealed.contains("whsec_"), "ciphertext must not leak the call"); assert_eq!(open_call(&wk, &sealed).unwrap(), c); // Another job, another direction, another worker: none can open it. - assert!( - open_call( - &request_key(&worker.secret, &buyer.public_hex, "job-2").unwrap(), - &sealed - ) - .is_err() - ); - assert!( - open_call( - &result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), - &sealed - ) - .is_err() - ); + assert!(open_call(&request_key(&worker.secret, &buyer.public_hex, "job-2").unwrap(), &sealed).is_err()); + assert!(open_call(&result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), &sealed).is_err()); let eve = crypto::keypair(); - assert!( - open_call( - &request_key(&eve.secret, &buyer.public_hex, "job-1").unwrap(), - &sealed - ) - .is_err() - ); - - let r = DeliveryResult { - delivered: true, - status: Some(200), - attempts: 1, - response_body: "ok".into(), - ..Default::default() - }; - let rs = seal_result( - &result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), - &r, - ) - .unwrap(); - assert_eq!( - open_result( - &result_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(), - &rs - ) - .unwrap(), - r - ); - assert!( - open_result(&bk, &rs).is_err(), - "a result cannot be opened as a request" - ); + assert!(open_call(&request_key(&eve.secret, &buyer.public_hex, "job-1").unwrap(), &sealed).is_err()); + + let r = DeliveryResult { delivered: true, status: Some(200), attempts: 1, response_body: "ok".into(), ..Default::default() }; + let rs = seal_result(&result_key(&worker.secret, &buyer.public_hex, "job-1").unwrap(), &r).unwrap(); + assert_eq!(open_result(&result_key(&buyer.secret, &worker.public_hex, "job-1").unwrap(), &rs).unwrap(), r); + assert!(open_result(&bk, &rs).is_err(), "a result cannot be opened as a request"); } #[test] @@ -1227,11 +989,7 @@ mod tests { // JSON escaping can blow a body under the cap past one message. c.body = "\u{1}".repeat(MAX_BODY / 2); assert!(seal_call(&key, &c).is_err()); - let r = DeliveryResult { - status: Some(200), - response_body: "\u{1}".repeat(MAX_RESPONSE_BODY), - ..Default::default() - }; + let r = DeliveryResult { status: Some(200), response_body: "\u{1}".repeat(MAX_RESPONSE_BODY), ..Default::default() }; let s = seal_result(&key, &r).unwrap(); assert!(s.len() <= MAX_SEALED); assert!(open_result(&key, &s).unwrap().truncated); @@ -1240,23 +998,13 @@ mod tests { #[test] fn call_validation() { assert!(call("https://example.com/h").validate().is_ok()); - for bad in [ - "ftp://example.com/", - "file:///etc/passwd", - "https://user:pw@example.com/", - "not a url", - ] { + for bad in ["ftp://example.com/", "file:///etc/passwd", "https://user:pw@example.com/", "not a url"] { assert!(call(bad).validate().is_err(), "{bad}"); } let mut c = call("https://example.com/h"); c.method = "CONNECT".into(); assert!(c.validate().is_err()); - for h in [ - "Host", - "content-length", - "webhook-signature", - "Transfer-Encoding", - ] { + for h in ["Host", "content-length", "webhook-signature", "Transfer-Encoding"] { let mut c = call("https://example.com/h"); c.headers = vec![(h.into(), "x".into())]; assert!(c.validate().is_err(), "{h}"); @@ -1283,18 +1031,10 @@ mod tests { ] { let r = deliver(&call(&url), &spec(3), "j", Guard::default()).await; assert!(!r.answered(), "{url} must not be reached"); - assert!( - r.error.as_deref().unwrap_or("").starts_with("refused"), - "{url}: {:?}", - r.error - ); + assert!(r.error.as_deref().unwrap_or("").starts_with("refused"), "{url}: {:?}", r.error); assert_eq!(r.attempts, 1, "{url}: a refusal is not retried"); } - assert_eq!( - seen.hits.load(Ordering::SeqCst), - 0, - "nothing reached the local target" - ); + assert_eq!(seen.hits.load(Ordering::SeqCst), 0, "nothing reached the local target"); } #[tokio::test] @@ -1304,16 +1044,8 @@ mod tests { assert_eq!(r.status, Some(302)); assert!(!r.delivered); assert_eq!(r.attempts, 1, "a 3xx is final"); - assert_eq!( - seen.other_hits.load(Ordering::SeqCst), - 0, - "the redirect target was never fetched" - ); - assert!( - r.response_headers - .iter() - .any(|(k, v)| k == "location" && v == "/other") - ); + assert_eq!(seen.other_hits.load(Ordering::SeqCst), 0, "the redirect target was never fetched"); + assert!(r.response_headers.iter().any(|(k, v)| k == "location" && v == "/other")); } #[tokio::test] @@ -1323,10 +1055,7 @@ mod tests { c.secret = Some("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw".into()); c.id = Some("msg_1".into()); let r = deliver(&c, &spec(3), "j", LOCAL).await; - assert!( - r.delivered && r.status == Some(200) && r.attempts == 1, - "{r:?}" - ); + assert!(r.delivered && r.status == Some(200) && r.attempts == 1, "{r:?}"); assert_eq!(r.response_body, "thanks"); assert_eq!(r.webhook_id.as_deref(), Some("msg_1")); let (h, body) = seen.last.lock().unwrap().clone().unwrap(); @@ -1336,26 +1065,15 @@ mod tests { assert_eq!(h["webhook-id"], "msg_1"); let ts: u64 = h["webhook-timestamp"].to_str().unwrap().parse().unwrap(); // Verify the way a receiver does, with an independent HMAC. - let mut mac = as Mac>::new_from_slice( - &B64.decode("MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw").unwrap(), - ) - .unwrap(); + let mut mac = as Mac>::new_from_slice(&B64.decode("MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw").unwrap()).unwrap(); mac.update(format!("msg_1.{ts}.{}", c.body).as_bytes()); - assert_eq!( - h["webhook-signature"].to_str().unwrap(), - format!("v1,{}", B64.encode(mac.finalize().into_bytes())) - ); + assert_eq!(h["webhook-signature"].to_str().unwrap(), format!("v1,{}", B64.encode(mac.finalize().into_bytes()))); } #[test] fn signature_matches_the_standard_webhooks_reference_vector() { // From the Standard Webhooks spec / svix reference libraries. - let sig = sign( - "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", - "msg_p5jXN8AQM9LWM0D4loKWxJek", - 1614265330, - br#"{"test": 2432232314}"#, - ); + let sig = sign("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw", "msg_p5jXN8AQM9LWM0D4loKWxJek", 1614265330, br#"{"test": 2432232314}"#); assert_eq!(sig, "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE="); } @@ -1368,24 +1086,11 @@ mod tests { assert_eq!(seen.hits.load(Ordering::SeqCst), 3); let r = deliver(&call(&format!("{base}/bad")), &spec(3), "j", LOCAL).await; - assert_eq!( - (r.attempts, r.status), - (1, Some(400)), - "a 4xx is not retried" - ); + assert_eq!((r.attempts, r.status), (1, Some(400)), "a 4xx is not retried"); assert_eq!(seen.hits.load(Ordering::SeqCst), 4); - let r = deliver( - &call(&format!("{base}/fail")), - &spec(MAX_ATTEMPTS + 1), - "j", - LOCAL, - ) - .await; - assert!( - r.error.unwrap().contains("max_attempts"), - "over the cap is refused outright" - ); + let r = deliver(&call(&format!("{base}/fail")), &spec(MAX_ATTEMPTS + 1), "j", LOCAL).await; + assert!(r.error.unwrap().contains("max_attempts"), "over the cap is refused outright"); assert_eq!(seen.hits.load(Ordering::SeqCst), 4); } @@ -1409,32 +1114,19 @@ mod tests { let mut s = spec(1); s.client_pubkey = "zz".into(); assert!(s.validate().is_err()); - let s = WebhookSpec { - v: 1, - client_pubkey: "11".repeat(32), - max_attempts: 3, - timeout_ms: 10_000, - backoff_ms: 1_000, - }; + let s = WebhookSpec { v: 1, client_pubkey: "11".repeat(32), max_attempts: 3, timeout_ms: 10_000, backoff_ms: 1_000 }; assert_eq!(s.budget_secs(), 30 + 1 + 2); assert_eq!(backoff(10_000, 5), Duration::from_millis(BACKOFF_CEIL_MS)); } async fn node(dir: &std::path::Path, boot: Vec) -> Arc { - let cfg = c0mpute_net::NetworkConfig::for_dir(dir.to_path_buf()) - .with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]) - .with_bootstrap(boot); + let cfg = c0mpute_net::NetworkConfig::for_dir(dir.to_path_buf()).with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]).with_bootstrap(boot); Arc::new(Libp2pNetwork::spawn(cfg).await.unwrap()) } async fn dial_addr(net: &Libp2pNetwork) -> libp2p::Multiaddr { for _ in 0..50 { - if let Some(a) = net - .listen_addrs() - .await - .ok() - .and_then(|v| v.into_iter().next()) - { + if let Some(a) = net.listen_addrs().await.ok().and_then(|v| v.into_iter().next()) { return a.with(libp2p::multiaddr::Protocol::P2p(net.peer_id())); } tokio::time::sleep(Duration::from_millis(50)).await; @@ -1457,15 +1149,7 @@ mod tests { let load = WorkerLoad::new(Arc::new(Semaphore::new(4)), 4); let tx = spawn_runner(worker.clone(), load.clone(), LOCAL); - crate::dispatch::run_worker_subscriber_with_load( - worker.clone(), - WORKLOAD.into(), - vec![TAG.into()], - Some(tx), - Some(offer_filter(DEFAULT_PRICE_USD)), - Some(pricing(DEFAULT_PRICE_USD)), - Some(load), - ); + crate::dispatch::run_worker_subscriber_with_load(worker.clone(), WORKLOAD.into(), vec![TAG.into()], Some(tx), Some(offer_filter(DEFAULT_PRICE_USD)), Some(pricing(DEFAULT_PRICE_USD)), Some(load)); let topic = job_topic(WORKLOAD); bystander.subscribe(&topic).await.unwrap(); let mut spy = bystander.messages(); @@ -1481,34 +1165,14 @@ mod tests { let secret = "whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw"; // gitleaks:allow (public Standard Webhooks spec test key) let mut c = call(&format!("{base}/ok")); c.secret = Some(secret.into()); - c.headers - .push(("authorization".into(), "Bearer sk_live_topsecret".into())); - let req = SendRequest { - call: c.clone(), - max_attempts: 2, - timeout_ms: 5_000, - backoff_ms: 100, - workers: vec![], - max_price_usd: DEFAULT_MAX_PRICE_USD, - bid_window: Duration::from_secs(3), - external_ref: Some("coinpay:inv_e2e".into()), - }; - let out = tokio::time::timeout(Duration::from_secs(90), send(buyer.clone(), &req)) - .await - .expect("send finishes") - .expect("send succeeds"); + c.headers.push(("authorization".into(), "Bearer sk_live_topsecret".into())); + let req = SendRequest { call: c.clone(), max_attempts: 2, timeout_ms: 5_000, backoff_ms: 100, workers: vec![], max_price_usd: DEFAULT_MAX_PRICE_USD, bid_window: Duration::from_secs(3), external_ref: Some("coinpay:inv_e2e".into()) }; + let out = tokio::time::timeout(Duration::from_secs(90), send(buyer.clone(), &req)).await.expect("send finishes").expect("send succeeds"); assert_eq!(out.worker_peer_id, worker.peer_id().to_base58()); assert_eq!(out.receipt_status, JobStatus::Completed); - assert!( - out.result.delivered && out.result.status == Some(200), - "{:?}", - out.result - ); + assert!(out.result.delivered && out.result.status == Some(200), "{:?}", out.result); assert_eq!(out.result.response_body, "thanks"); - assert_eq!( - out.price_usd, DEFAULT_PRICE_USD, - "paid the worker's price, not zero and not the cap" - ); + assert_eq!(out.price_usd, DEFAULT_PRICE_USD, "paid the worker's price, not zero and not the cap"); assert_eq!(seen.hits.load(Ordering::SeqCst), 1); let (h, _) = seen.last.lock().unwrap().clone().unwrap(); assert_eq!(h["authorization"], "Bearer sk_live_topsecret"); @@ -1516,22 +1180,9 @@ mod tests { tokio::time::sleep(Duration::from_millis(500)).await; let wire = String::from_utf8_lossy(&sniffed.lock().unwrap()).into_owned(); - assert!( - wire.contains("webhook_request") && wire.contains("webhook_result"), - "the bystander saw the sealed exchange" - ); - for leak in [ - base.as_str(), - "sk_live_topsecret", - secret, - "x-tenant", - "\"event\"", - "thanks", - ] { - assert!( - !wire.contains(leak), - "plaintext {leak:?} leaked onto gossip" - ); + assert!(wire.contains("webhook_request") && wire.contains("webhook_result"), "the bystander saw the sealed exchange"); + for leak in [base.as_str(), "sk_live_topsecret", secret, "x-tenant", "\"event\"", "thanks"] { + assert!(!wire.contains(leak), "plaintext {leak:?} leaked onto gossip"); } } @@ -1560,57 +1211,24 @@ mod tests { // Our bid is always our price, never the buyer's cap, never zero. let p = pricing(DEFAULT_PRICE_USD); assert_eq!(p.quote(0.01, 0.0), Some(DEFAULT_PRICE_USD)); - assert_eq!( - p.quote(DEFAULT_PRICE_USD / 2.0, 0.0), - None, - "a cap under our price gets no bid" - ); + assert_eq!(p.quote(DEFAULT_PRICE_USD / 2.0, 0.0), None, "a cap under our price gets no bid"); let f = offer_filter(DEFAULT_PRICE_USD); let good = serde_json::to_value(spec(3)).unwrap(); assert!(f(&offer(DEFAULT_MAX_PRICE_USD, good.clone()))); assert!(!f(&offer(0.0, good.clone())), "a free offer gets no bid"); assert!(!f(&offer(f64::NAN, good.clone()))); - assert!( - !f(&offer(1.0, serde_json::to_value(spec(9)).unwrap())), - "an invalid spec gets no bid" - ); - assert!( - !f(&offer( - 1.0, - serde_json::json!({"url": "http://169.254.169.254/"}) - )), - "plaintext calls are not a webhook spec" - ); + assert!(!f(&offer(1.0, serde_json::to_value(spec(9)).unwrap())), "an invalid spec gets no bid"); + assert!(!f(&offer(1.0, serde_json::json!({"url": "http://169.254.169.254/"}))), "plaintext calls are not a webhook spec"); let mut short = offer(1.0, good); short.deadline_unix_ms = short.published_at_ms + 1_000; - assert!( - !f(&short), - "a deadline shorter than the retry budget gets no bid" - ); + assert!(!f(&short), "a deadline shorter than the retry budget gets no bid"); } #[test] 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()), - buyer_addrs: Vec::new(), - }; + 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()), 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(); @@ -1623,9 +1241,6 @@ mod tests { assert!(bid.matches(&a, 0).is_err(), "over the cap"); let mut a = accept; a.buyer_peer_id = "mallory".into(); - assert!( - bid.matches(&a, 0).is_err(), - "only the offer's buyer accepts" - ); + assert!(bid.matches(&a, 0).is_err(), "only the offer's buyer accepts"); } } diff --git a/node/crates/c0mpute-net/src/topics.rs b/node/crates/c0mpute-net/src/topics.rs index bf58cb3c..302a74c5 100644 --- a/node/crates/c0mpute-net/src/topics.rs +++ b/node/crates/c0mpute-net/src/topics.rs @@ -245,11 +245,7 @@ mod tests { assert_eq!(offer.external_ref, None); // And a new offer omits the field entirely when unset, so the bytes // an old node sees are unchanged. - assert!( - !serde_json::to_string(&offer) - .unwrap() - .contains("external_ref") - ); + assert!(!serde_json::to_string(&offer).unwrap().contains("external_ref")); } #[test]