diff --git a/docs/hosting.md b/docs/hosting.md index 347cb05d..27c385ab 100644 --- a/docs/hosting.md +++ b/docs/hosting.md @@ -68,7 +68,141 @@ publishes it as `gateway` on `/api/nodes`. Nothing is advertised without that se and never in private mode. Clients that are not libp2p peers (the Moshpit edge, for one) use this list to find gateways to fetch sites from. +## Paid hosting: the d1sks marketplace + +Free copies are kept by volunteers. To pay hosting nodes to keep your site, pick them +from the [d1sks](https://d1sks.pwa.pit.world/disks?hosting=1) OpenDisk marketplace when +you publish: + +```sh +c0mpute hosting disks --country DE,FI # what --disks auto would pick +c0mpute hosting publish ./public --name my-site --disks auto +c0mpute hosting publish ./public --name my-site --disks auto \ + --country DE,FI --max-price 0.01 --copies 3 +c0mpute hosting publish ./public --name my-site \ + --disks 'country=DE&min_standing=50&max_price=0.01' +``` + +`--disks` is `auto` or a d1sks query (`country`, `max_price`, `min_gib`, `min_standing`; +unknown keys are refused). `--copies` is how many preferred hosts to pick (1-5, default 3). + +**d1sks is discovery only.** `publish` asks it once. It keeps only disks that are c0mpute +hosting nodes (the descriptor has a `c0mpute.peer_id` and `accepts.hosting: true` or the +`hosting` role), drops any that ask more than `--max-price`, sit outside `--country` or +lack room for the site, and ranks the rest: cheapest first, then highest d1sks standing +(unrated last), then most free space, then peer id. The first `--copies` become the +site's **preferred hosts**. The agreed price is the highest ask among them, so each gets +at least what it asks. If d1sks is unreachable or nothing matches, the site publishes as a +plain free site. Placement, proofs, serving and settlement never touch d1sks. + +### Record v2 + +The preferred hosts go into the signed site record: + +| Field | | +|---|---| +| `v` | `2` when the record carries marketplace terms; plain sites stay `1`, byte for byte | +| `preferred_hosts` | 1-5 base58 peer ids | +| `price_per_gib_month` | agreed USD price, canonical decimal (`"0.004"`) | +| `payer` | hex ed25519 key that pays; absent = the owner | + +A v2 record signs under its own domain (`c0mpute-hosting-v2`), so terms cannot be +stripped or added without breaking the signature. New nodes read v1 and v2. Nodes older +than this release ignore v2 records (they cannot verify them), so a v2 site is held only +by upgraded nodes. + +### Placement + +Still leaderless, still computed the same way on every node: + +1. A preferred host that is live, asks no more than the agreed price (hosting nodes + gossip their price) and has not lost preference replicates whether or not the site + already has three copies. +2. The rendezvous fill runs exactly as before and does not wait for preferred hosts, so + the site reaches three copies even if every preferred host is offline. +3. Once preferred hosts hold copies, free copies above `3 - preferred holders` are + released one per round, lowest rank first. The total never drops below three. + +Two preferred hosts among five nodes settle at three copies: both preferred plus one free. + +### Proof of storage + +Any node with `hosting.challenge` on asks each holder for one random chunk of the site, +straight from that peer, and checks its blake3 hash. The challenger signs a receipt +(`site, manifest, holder, chunk, ok, rtt, time`) and gossips it on `c0mpute/hosting/v1`. + +- `challenge = "own"` (default): the sites this machine's publisher key owns or pays for. +- `challenge = "all"`: also a random sample of everyone else's (a gateway might); at most + 32 challenges per round. +- `challenge = "off"`. + +A preferred host loses preference when at least 3 of its latest 5 receipts signed by the +site's owner or payer are failures. Receipts from anyone else are kept and shown but +cannot demote anybody. The site's copies are kept up by the normal fill meanwhile. + +This proves the peer returned the right bytes for a chunk it could not predict, within the +timeout. It does not prove the peer keeps its own copy rather than relaying from another +holder. + +### Pricing and settlement + +The publishing machine's node is the payer. Keep a c0mpute node running there: it signs +the receipts and settles once per `hosting.settle_every_secs` (default a day) over the +window since the last round: + +```text +owed = price_per_gib_month x site GiB x (window / 30 days) x passed / challenged +``` + +counting only receipts signed by the owner or payer. Under half passed, or never +challenged, earns nothing for that window. Amounts are integer micro-USD, rounded down. +Each transfer has an idempotency key (site, holder, window) recorded in +`/hosting/ledger.json`, so nothing is paid twice; windows never overlap. + +Hosting nodes say what they ask and where to be paid: + +```toml +[hosting] +price_usd_per_gib_month = 0.004 # in the OpenDisk descriptor and hosting gossip +payout_wallet = "0x..." # where payers send money +payout_coin = "USDC_POL" # CoinPay currency code +challenge = "own" # own | all | off +challenge_every_secs = 600 +settle_every_secs = 86400 +``` + +### Settlement switch (off by default) + +With `[payments]` off, which is the default, settlement is a dry run: the node logs what it +would pay (`hosting settlement (dry run ...)`) and moves nothing. To pay for real, from the +payer's own CoinPay business wallet: + +1. In CoinPay, create a business API key with the `payouts:create` scope on a business + whose payout wallet holds the coin your hosts want (e.g. USDC on Polygon). +2. Put the key in the node's environment, never in config.toml: + `export C0MPUTE_COINPAY_PAYOUT_KEY=...` (the variable name is + `payments.coinpay.api_key_env`). +3. Turn payments on in `~/.config/c0mpute/config.toml`: + + ```toml + [payments] + enabled = true + rail = "coinpay" + max_usd_per_round = "10" # hard cap per settlement round + [payments.coinpay] + url = "https://coinpayportal.com" + recipient_email = "you@example.org" # CoinPay files every payout under an email + ``` + +4. Restart the node. Each round then calls CoinPay `POST /api/payouts/create` once per + preferred holder that is owed something and has a payout wallet, up to the cap. + +`enabled = true` without the key is an error in the log and settlement stays a dry run. +CoinPay has no balance call for an API key, so the round cap is the spending limit. +Payments move only between the payer and the hosts it chose; c0mpute holds nothing. + ## Not yet -Paid reservations, epoch settlement and proof-of-serve challenges (DIP-0015) are not -implemented. Hosting is free while it is in beta. +Escrow (money held until a window is proven) and proof-of-serve challenges from DIP-0015 +are not implemented: payers pay from their own wallet after the fact. Free hosting stays +free. diff --git a/docs/storage-opendisk.md b/docs/storage-opendisk.md index df4985bd..2d468631 100644 --- a/docs/storage-opendisk.md +++ b/docs/storage-opendisk.md @@ -8,6 +8,10 @@ Storage is a default role. Every node that runs it (that is every node unless | `/storage/v1/objects/{hash}`, `/shards/{hash}`, `/manifests/{hash}`, `/status` | The CIP-002 storage API. Reads are open: a blake3 hash is the capability, and private-tier objects are ciphertext. Writes need an `x-coinpay-auth` envelope signed by the node's own ed25519 key or a key in `storage.trusted_writers`, so a stranger cannot fill the disk. | | `/.well-known/opendisk.json` | The node's [OpenDisk 0.1](https://d1sks.pwa.pit.world/docs) descriptor: `key` (`ed25519:` + the node's identity key), live `capacity.free_gib` / `total_gib`, `price.per_gib_month`, and a `c0mpute` block with the peer id and storage API URL. | +A node with the `hosting` role also says so: `accepts.hosting: true`, `c0mpute.roles`, and +`c0mpute.hosting { price_per_gib_month, max_gib }`. d1sks lists those at `?hosting=1`, and +`c0mpute hosting publish --disks auto` picks paid hosts from them (see [hosting](hosting.md)). + ## Listing on d1sks A node lists itself; nothing central does it for the node. When the gateway announces a diff --git a/node/crates/c0mpute-cli/src/main.rs b/node/crates/c0mpute-cli/src/main.rs index ca4126db..b02c0a37 100644 --- a/node/crates/c0mpute-cli/src/main.rs +++ b/node/crates/c0mpute-cli/src/main.rs @@ -619,6 +619,37 @@ enum HostingCmd { /// Give up waiting for network copies after this many seconds. #[arg(long, default_value_t = 600)] wait_secs: u64, + /// Pay hosting nodes from the d1sks OpenDisk marketplace to hold the + /// site: `auto`, or a d1sks query such as + /// `country=DE,FI&max_price=0.01&min_standing=50`. Without it the + /// site is free and the network keeps 3 copies as usual. + #[arg(long)] + disks: Option, + /// Highest price per GiB-month (USD) a preferred host may ask. + #[arg(long)] + max_price: Option, + /// Only hosts in these countries (ISO codes, comma separated). + #[arg(long)] + country: Option, + /// How many preferred (paid) hosts to pick, 1-5. + #[arg(long, default_value_t = 3)] + copies: usize, + /// The OpenDisk marketplace to pick hosts from (discovery only). + #[arg(long, default_value = c0mpute_core::opendisk::DEFAULT_D1SKS_URL)] + d1sks: String, + }, + /// Show the c0mpute hosting nodes d1sks lists, ranked the way + /// `publish --disks` would pick them. + Disks { + /// `auto` or a d1sks query (see `publish --disks`). + #[arg(long, default_value = "auto")] + query: String, + #[arg(long)] + max_price: Option, + #[arg(long)] + country: Option, + #[arg(long, default_value = c0mpute_core::opendisk::DEFAULT_D1SKS_URL)] + d1sks: String, }, /// List the sites a gateway knows, with live copy counts. Ls { @@ -3205,6 +3236,22 @@ async fn run_ads(cmd: AdsCmd) -> Result<()> { } } +/// `--disks` plus the `--max-price` / `--country` shorthands as one query. +fn disk_query( + spec: &str, + max_price: Option<&str>, + country: Option<&str>, +) -> Result { + let mut q = c0mpute_core::hosting::market::DiskQuery::parse(spec)?; + if let Some(p) = max_price { + q.max_price = Some(c0mpute_core::payments::parse_usd(p).with_context(|| format!("--max-price {p:?}"))?); + } + if let Some(c) = country { + q.add_countries(c)?; + } + Ok(q) +} + async fn run_hosting(cmd: HostingCmd) -> Result<()> { match cmd { HostingCmd::Publish { @@ -3212,18 +3259,49 @@ async fn run_hosting(cmd: HostingCmd) -> Result<()> { name, domains, wait_secs, + disks, + max_price, + country, + copies, + d1sks, } => { let domains: Vec = domains.into_iter().map(|d| d.to_ascii_lowercase()).collect(); - let (record, copies) = c0mpute_core::hosting::publish_dir( + let market = match disks { + Some(spec) => Some(c0mpute_core::hosting::MarketOptions { + query: disk_query(&spec, max_price.as_deref(), country.as_deref())?, + copies: copies.clamp(1, c0mpute_core::hosting::site::MAX_PREFERRED_HOSTS), + d1sks_url: d1sks, + }), + None => { + if max_price.is_some() || country.is_some() { + anyhow::bail!("--max-price and --country pick marketplace hosts; add --disks auto"); + } + None + } + }; + let (record, copies) = c0mpute_core::hosting::publish_dir_with( &dir, &name, domains.clone(), std::time::Duration::from_secs(wait_secs), + market, ) .await?; println!("site: {}", record.site_id); println!("version: {}", record.seq); println!("copies: {copies}/{}", c0mpute_core::hosting::TARGET_REPLICAS); + if !record.preferred_hosts.is_empty() { + println!( + "paid: {} preferred host(s) at ${}/GiB-month, paid by {}", + record.preferred_hosts.len(), + record.price_per_gib_month.as_deref().unwrap_or("?"), + record.payer_key() + ); + println!( + " keep a c0mpute node running on this machine: it challenges the \ + holders and settles (dry run until [payments] is enabled)" + ); + } println!("path: /s/{}/ (on any c0mpute gateway, e.g. http://:7777/s/{}/)", record.site_id, record.site_id); for d in &domains { println!( @@ -3242,6 +3320,31 @@ async fn run_hosting(cmd: HostingCmd) -> Result<()> { } Ok(()) } + HostingCmd::Disks { + query, + max_price, + country, + d1sks, + } => { + let q = disk_query(&query, max_price.as_deref(), country.as_deref())?; + let disks = c0mpute_core::hosting::market::query(&d1sks, &q).await?; + let ranked = c0mpute_core::hosting::market::rank(&disks, &q, 0); + if ranked.is_empty() { + println!("no c0mpute hosting disks on {d1sks} match ({} listed)", disks.len()); + } + for d in ranked { + println!( + "{:<54} ${:<9} {:>8.1} GiB standing {:<5} {:<8} {}", + d.peer_id, + c0mpute_core::payments::format_usd(d.price), + d.free_bytes as f64 / (1024.0 * 1024.0 * 1024.0), + d.standing.map(|s| format!("{s:.0}")).unwrap_or_else(|| "-".into()), + d.countries.join(","), + d.name + ); + } + Ok(()) + } HostingCmd::Ls { gateway } => { let url = format!("{}/hosting/v1/sites", gateway.trim_end_matches('/')); let v: serde_json::Value = reqwest::get(&url) diff --git a/node/crates/c0mpute-core/src/config.rs b/node/crates/c0mpute-core/src/config.rs index ee197a90..8cc7acf7 100644 --- a/node/crates/c0mpute-core/src/config.rs +++ b/node/crates/c0mpute-core/src/config.rs @@ -60,6 +60,10 @@ pub struct Config { /// Game server rentals (`gameserver` role, opt-in). #[serde(default)] pub gameserver: GameserverConfig, + /// Money movement (hosting settlement; ads billing later). Off unless + /// `enabled = true` and the rail's key is in the environment. + #[serde(default)] + pub payments: crate::payments::PaymentsConfig, } /// Ads on a gateway (`ads-manager`, `ctv-ads`, `podcast-ads`). Every gateway @@ -127,16 +131,83 @@ pub struct HostingConfig { /// Most bytes of other people's sites this node will hold. #[serde(default = "default_hosting_max_bytes")] pub max_bytes: u64, + /// What this node asks to be a preferred (paid) host, USD per GiB-month. + /// Published in the OpenDisk descriptor and in hosting gossip; a site + /// offering less does not count this node as preferred. + #[serde(default = "default_hosting_price")] + pub price_usd_per_gib_month: f64, + /// Where marketplace payments for holding sites go (a wallet address on + /// `payout_coin`'s chain). Without it this node can still be preferred, + /// but nobody can pay it. + #[serde(default)] + pub payout_wallet: Option, + #[serde(default = "crate::payments::default_coin")] + pub payout_coin: String, + /// Which sites this node challenges for proof of storage: `own` (sites + /// its publisher key owns or pays for), `all` (also a sample of everyone + /// else's, as a gateway might), or `off`. + #[serde(default = "default_challenge")] + pub challenge: String, + /// Seconds between challenge rounds. + #[serde(default = "default_challenge_every")] + pub challenge_every_secs: u64, + /// Seconds between settlement rounds (paying preferred holders of the + /// sites this node pays for). Payments only move when `[payments]` is on. + #[serde(default = "default_settle_every")] + pub settle_every_secs: u64, } impl Default for HostingConfig { fn default() -> Self { Self { max_bytes: default_hosting_max_bytes(), + price_usd_per_gib_month: default_hosting_price(), + payout_wallet: None, + payout_coin: crate::payments::default_coin(), + challenge: default_challenge(), + challenge_every_secs: default_challenge_every(), + settle_every_secs: default_settle_every(), } } } +impl HostingConfig { + /// The hosting price in micro-USD per GiB-month (the default when unusable). + pub fn price_micros(&self) -> crate::payments::Micros { + let p = if self.price_usd_per_gib_month.is_finite() && self.price_usd_per_gib_month >= 0.0 { + self.price_usd_per_gib_month + } else { + default_hosting_price() + }; + (p * crate::payments::MICROS_PER_USD as f64).round() as crate::payments::Micros + } + + /// The payout address, when set and well formed. + pub fn payee(&self) -> Option { + let p = crate::payments::Payee { + wallet: self.payout_wallet.clone()?.trim().to_string(), + coin: self.payout_coin.trim().to_string(), + }; + p.valid().then_some(p) + } +} + +fn default_hosting_price() -> f64 { + crate::opendisk::DEFAULT_PRICE_USD_PER_GIB_MONTH +} + +fn default_challenge() -> String { + "own".into() +} + +fn default_challenge_every() -> u64 { + 600 +} + +fn default_settle_every() -> u64 { + 24 * 60 * 60 +} + fn default_hosting_max_bytes() -> u64 { crate::hosting::DEFAULT_MAX_BYTES } @@ -384,6 +455,7 @@ impl Default for Config { ads: AdsConfig::default(), embed: EmbedConfig::default(), gameserver: GameserverConfig::default(), + payments: Default::default(), } } } diff --git a/node/crates/c0mpute-core/src/hosting/market.rs b/node/crates/c0mpute-core/src/hosting/market.rs new file mode 100644 index 00000000..fa2f2800 --- /dev/null +++ b/node/crates/c0mpute-core/src/hosting/market.rs @@ -0,0 +1,373 @@ +//! Picking paid hosts from the d1sks OpenDisk marketplace. +//! +//! d1sks is discovery only. `c0mpute hosting publish --disks …` asks it once +//! for disks that are c0mpute hosting nodes, ranks them here, and writes the +//! chosen peer ids and the agreed price into the signed site record. From +//! then on placement, proofs, serving and settlement are peer to peer; if +//! d1sks is down the site publishes without preferred hosts and the network +//! keeps its usual three free copies. + +use std::time::Duration; + +use anyhow::{Context, Result, bail}; +use serde_json::Value; + +use super::site::{HostingTerms, MAX_PREFERRED_HOSTS}; +use crate::payments::{Micros, parse_usd}; + +const GIB: f64 = 1024.0 * 1024.0 * 1024.0; + +/// What to ask d1sks for. `--disks auto` is the empty query; otherwise a +/// d1sks-style query string such as `country=DE,FI&max_price=0.01&min_gib=5`. +#[derive(Clone, Debug, Default, PartialEq, Eq)] +pub struct DiskQuery { + /// ISO 3166-1 alpha-2 codes, upper case; empty = anywhere. + pub countries: Vec, + /// Highest hosting price accepted, micro-USD per GiB-month. + pub max_price: Option, + /// Least free space a disk must report, GiB. + pub min_gib: Option, + /// Least d1sks standing (0-100). + pub min_standing: Option, +} + +impl DiskQuery { + /// Parse `auto` or a query string. Unknown keys are an error, so a typo + /// does not silently widen the search. + pub fn parse(spec: &str) -> Result { + let spec = spec.trim().trim_start_matches('?'); + let mut q = DiskQuery::default(); + if spec.is_empty() || spec == "auto" { + return Ok(q); + } + for pair in spec.split('&').filter(|p| !p.is_empty()) { + let (k, v) = pair.split_once('=').with_context(|| format!("{pair:?} is not key=value"))?; + let v = v.trim(); + match k.trim() { + "country" => q.add_countries(v)?, + "max_price" => q.max_price = Some(parse_usd(v).with_context(|| format!("max_price {v:?}"))?), + "min_gib" => q.min_gib = Some(v.parse().with_context(|| format!("min_gib {v:?}"))?), + "min_standing" => { + let s: u32 = v.parse().with_context(|| format!("min_standing {v:?}"))?; + if s > 100 { + bail!("min_standing is 0-100"); + } + q.min_standing = Some(s); + } + other => bail!("unknown d1sks query key {other:?} (country, max_price, min_gib, min_standing)"), + } + } + Ok(q) + } + + /// Add comma-separated country codes. + pub fn add_countries(&mut self, list: &str) -> Result<()> { + for c in list.split(',').map(|c| c.trim()).filter(|c| !c.is_empty()) { + let c = c.to_ascii_uppercase(); + if c.len() != 2 || !c.chars().all(|ch| ch.is_ascii_uppercase()) { + bail!("{c:?} is not an ISO country code"); + } + if !self.countries.contains(&c) { + self.countries.push(c); + } + } + Ok(()) + } + + /// The d1sks listing URL for this query. `hosting=1` asks d1sks to list + /// only c0mpute hosting nodes; the answer is filtered here again anyway, + /// so an older d1sks that ignores the parameter still works. + pub fn url(&self, base: &str) -> String { + let mut u = format!("{}/api/v1/disks?hosting=1&sort=price&limit=200", base.trim_end_matches('/')); + if !self.countries.is_empty() { + u.push_str(&format!("&country={}", self.countries.join(","))); + } + if let Some(p) = self.max_price { + u.push_str(&format!("&max_price={}", crate::payments::format_usd(p))); + } + if let Some(g) = self.min_gib { + u.push_str(&format!("&min_gib={g}")); + } + if let Some(s) = self.min_standing { + u.push_str(&format!("&min_standing={s}")); + } + u + } +} + +/// A c0mpute hosting node as d1sks lists it. +#[derive(Clone, Debug, PartialEq)] +pub struct HostDisk { + pub peer_id: String, + pub name: String, + /// Hosting price, micro-USD per GiB-month. + pub price: Micros, + /// Free bytes it offers for hosting. + pub free_bytes: u64, + /// d1sks standing 0-100, when it has one. + pub standing: Option, + pub countries: Vec, +} + +fn usd_micros(v: &Value) -> Option { + let f = v.as_f64()?; + if !f.is_finite() || f < 0.0 { + return None; + } + Some((f * crate::payments::MICROS_PER_USD as f64).round() as Micros) +} + +/// The c0mpute hosting nodes in a d1sks `/api/v1/disks` answer. A disk +/// counts when its descriptor (`data`) carries a valid `c0mpute.peer_id` and +/// says it hosts (`accepts.hosting: true` or `"hosting"` in `c0mpute.roles`). +pub fn parse_disks(body: &Value) -> Vec { + let Some(rows) = body["disks"].as_array() else { + return Vec::new(); + }; + let mut out = Vec::new(); + for row in rows { + if row["status"].as_str().is_some_and(|s| s != "active") { + continue; + } + let data = &row["data"]; + let c = &data["c0mpute"]; + let Some(peer_id) = c["peer_id"].as_str() else { + continue; + }; + if peer_id.len() > 64 || peer_id.parse::().is_err() { + continue; + } + let roles_host = c["roles"] + .as_array() + .is_some_and(|r| r.iter().any(|x| x.as_str() == Some("hosting"))); + if !(data["accepts"]["hosting"].as_bool() == Some(true) || roles_host) { + continue; + } + let price = usd_micros(&c["hosting"]["price_per_gib_month"]) + .or_else(|| usd_micros(&row["per_gib_month"])) + .or_else(|| usd_micros(&data["price"]["per_gib_month"])); + let Some(price) = price else { + continue; + }; + let disk_free = row["free_gib"] + .as_f64() + .or_else(|| data["capacity"]["free_gib"].as_f64()) + .unwrap_or(0.0); + let free_gib = match c["hosting"]["max_gib"].as_f64() { + Some(m) => disk_free.min(m), + None => disk_free, + }; + let countries = row["countries"] + .as_array() + .or_else(|| data["location"]["countries"].as_array()) + .map(|a| { + a.iter() + .filter_map(|x| x.as_str().map(|s| s.to_ascii_uppercase())) + .collect() + }) + .unwrap_or_default(); + out.push(HostDisk { + peer_id: peer_id.to_string(), + name: row["name"].as_str().or(data["name"].as_str()).unwrap_or("").to_string(), + price, + free_bytes: (free_gib.max(0.0) * GIB) as u64, + standing: row["standing"].as_f64().or_else(|| data["record"]["standing"].as_f64()), + countries, + }); + } + out +} + +/// Disks that fit the query and the site, best first: cheapest, then +/// highest standing (unrated last), then most free space, then peer id, so +/// the same listing always gives the same order. +pub fn rank(disks: &[HostDisk], q: &DiskQuery, site_bytes: u64) -> Vec { + let mut v: Vec = disks + .iter() + .filter(|d| q.max_price.is_none_or(|m| d.price <= m)) + .filter(|d| d.free_bytes >= site_bytes) + .filter(|d| q.min_gib.is_none_or(|g| d.free_bytes as f64 >= g as f64 * GIB)) + .filter(|d| q.min_standing.is_none_or(|s| d.standing.is_some_and(|x| x >= s as f64))) + .filter(|d| q.countries.is_empty() || d.countries.iter().any(|c| q.countries.contains(c))) + .cloned() + .collect(); + v.sort_by(|a, b| { + a.price + .cmp(&b.price) + .then_with(|| { + let (x, y) = (a.standing.unwrap_or(-1.0), b.standing.unwrap_or(-1.0)); + y.total_cmp(&x) + }) + .then(b.free_bytes.cmp(&a.free_bytes)) + .then(a.peer_id.cmp(&b.peer_id)) + }); + let mut seen = std::collections::HashSet::new(); + v.retain(|d| seen.insert(d.peer_id.clone())); + v +} + +/// The first `copies` of a ranking (at most [`MAX_PREFERRED_HOSTS`]), and +/// the terms to sign: the agreed price is the highest ask among them, so +/// every chosen host gets at least what it asks. +pub fn choose(ranked: &[HostDisk], copies: usize, payer: Option) -> Option<(Vec, HostingTerms)> { + let n = copies.clamp(1, MAX_PREFERRED_HOSTS); + let chosen: Vec = ranked.iter().take(n).cloned().collect(); + if chosen.is_empty() { + return None; + } + let price = chosen.iter().map(|d| d.price).max().unwrap_or(0); + let terms = HostingTerms { + preferred_hosts: chosen.iter().map(|d| d.peer_id.clone()).collect(), + price_per_gib_month: price, + payer, + }; + Some((chosen, terms)) +} + +/// Ask d1sks for hosting disks matching `q`. +pub async fn query(d1sks: &str, q: &DiskQuery) -> Result> { + let url = q.url(d1sks); + let client = reqwest::Client::builder() + .timeout(Duration::from_secs(15)) + .user_agent(concat!("c0mpute/", env!("CARGO_PKG_VERSION"))) + .build()?; + let res = client.get(&url).send().await.with_context(|| format!("GET {url}"))?; + if !res.status().is_success() { + bail!("{url} answered {}", res.status()); + } + let body: Value = res.json().await.with_context(|| format!("{url} is not JSON"))?; + Ok(parse_disks(&body)) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn pid(seed: u8) -> String { + let kp = c0mpute_net::libp2p::identity::Keypair::ed25519_from_bytes([seed; 32]).unwrap(); + kp.public().to_peer_id().to_base58() + } + + /// A d1sks `/api/v1/disks` answer: rows as `src/disks.js` returns them, + /// with the raw descriptor in `data`. + fn fixture() -> Value { + serde_json::json!({ + "disks": [ + { // cheapest hosting node, DE + "name": "a", "status": "active", "free_gib": 100.0, "per_gib_month": 0.004, + "countries": ["DE"], "standing": 90, + "data": { "accepts": { "hosting": true }, + "c0mpute": { "peer_id": pid(1), "roles": ["storage", "hosting"], + "hosting": { "price_per_gib_month": 0.003, "max_gib": 5.0 } } } + }, + { // same price, lower standing, FI + "name": "b", "status": "active", "free_gib": 50.0, "per_gib_month": 0.003, + "countries": ["FI"], "standing": 40, + "data": { "c0mpute": { "peer_id": pid(2), "roles": ["hosting"] } } + }, + { // same price, unrated, US + "name": "c", "status": "active", "free_gib": 500.0, "per_gib_month": 0.003, + "countries": ["US"], "standing": null, + "data": { "accepts": { "hosting": true }, "c0mpute": { "peer_id": pid(3) } } + }, + { // pricey, DE + "name": "d", "status": "active", "free_gib": 900.0, "per_gib_month": 0.02, + "countries": ["DE"], "standing": 99, + "data": { "accepts": { "hosting": true }, "c0mpute": { "peer_id": pid(4) } } + }, + { // storage only: not a hosting node + "name": "e", "status": "active", "free_gib": 900.0, "per_gib_month": 0.001, + "countries": ["DE"], + "data": { "c0mpute": { "peer_id": pid(5), "roles": ["storage"] } } + }, + { // not a c0mpute node at all + "name": "f", "status": "active", "free_gib": 900.0, "per_gib_month": 0.001, + "data": { "accepts": { "hosting": true } } + }, + { // bad peer id + "name": "g", "status": "active", "free_gib": 900.0, "per_gib_month": 0.001, + "data": { "accepts": { "hosting": true }, "c0mpute": { "peer_id": "nope" } } + }, + { // gone + "name": "h", "status": "gone", "free_gib": 900.0, "per_gib_month": 0.001, + "data": { "accepts": { "hosting": true }, "c0mpute": { "peer_id": pid(8) } } + }, + { // tiny disk, FI + "name": "i", "status": "active", "free_gib": 0.0001, "per_gib_month": 0.001, + "countries": ["FI"], + "data": { "accepts": { "hosting": true }, "c0mpute": { "peer_id": pid(9) } } + } + ], + "total": 9 + }) + } + + #[test] + fn query_strings_parse_strictly() { + assert_eq!(DiskQuery::parse("auto").unwrap(), DiskQuery::default()); + assert_eq!(DiskQuery::parse("").unwrap(), DiskQuery::default()); + let q = DiskQuery::parse("country=de, fi&max_price=0.01&min_gib=5&min_standing=50").unwrap(); + assert_eq!(q.countries, vec!["DE", "FI"]); + assert_eq!(q.max_price, Some(10_000)); + assert_eq!(q.min_gib, Some(5)); + assert_eq!(q.min_standing, Some(50)); + assert_eq!( + q.url("https://d1sks.example/"), + "https://d1sks.example/api/v1/disks?hosting=1&sort=price&limit=200&country=DE,FI&max_price=0.01&min_gib=5&min_standing=50" + ); + assert!(DiskQuery::parse("country=Germany").is_err()); + assert!(DiskQuery::parse("max_price=cheap").is_err()); + assert!(DiskQuery::parse("colour=blue").is_err(), "unknown keys are refused"); + assert!(DiskQuery::parse("min_standing=101").is_err()); + assert!(DiskQuery::parse("country").is_err()); + } + + #[test] + fn only_c0mpute_hosting_nodes_are_parsed() { + let disks = parse_disks(&fixture()); + let names: Vec<&str> = disks.iter().map(|d| d.name.as_str()).collect(); + assert_eq!(names, vec!["a", "b", "c", "d", "i"]); + let a = &disks[0]; + assert_eq!(a.peer_id, pid(1)); + assert_eq!(a.price, 3_000, "the hosting price wins over the storage price"); + assert_eq!(a.free_bytes, 5 * 1024 * 1024 * 1024, "capped at hosting max_gib"); + assert_eq!(a.standing, Some(90.0)); + assert!(parse_disks(&serde_json::json!({"oops": 1})).is_empty()); + } + + #[test] + fn ranking_is_price_then_standing_then_space() { + let disks = parse_disks(&fixture()); + let site = 1024 * 1024; + let order: Vec = rank(&disks, &DiskQuery::default(), site) + .into_iter() + .map(|d| d.name) + .collect(); + // i is too small for the site; a/b/c tie on price, ranked by standing + // (unrated last); d is the most expensive. + assert_eq!(order, vec!["a", "b", "c", "d"]); + + let q = DiskQuery::parse("country=DE,FI&max_price=0.01").unwrap(); + let order: Vec = rank(&disks, &q, site).into_iter().map(|d| d.name).collect(); + assert_eq!(order, vec!["a", "b"]); + + let q = DiskQuery::parse("min_standing=50").unwrap(); + let order: Vec = rank(&disks, &q, site).into_iter().map(|d| d.name).collect(); + assert_eq!(order, vec!["a", "d"]); + } + + #[test] + fn choosing_agrees_on_the_highest_ask() { + let disks = parse_disks(&fixture()); + let ranked = rank(&disks, &DiskQuery::default(), 1024 * 1024); + let (chosen, terms) = choose(&ranked, 4, None).unwrap(); + assert_eq!(chosen.len(), 4); + assert_eq!(terms.preferred_hosts[0], pid(1)); + assert_eq!(terms.price_per_gib_month, 20_000, "d asks $0.02, so all four are offered $0.02"); + let (two, terms) = choose(&ranked, 2, None).unwrap(); + assert_eq!(two.len(), 2); + assert_eq!(terms.price_per_gib_month, 3_000); + assert_eq!(choose(&ranked, 99, None).unwrap().0.len(), 4, "capped by what exists"); + assert!(choose(&[], 3, None).is_none()); + } +} diff --git a/node/crates/c0mpute-core/src/hosting/mod.rs b/node/crates/c0mpute-core/src/hosting/mod.rs index 2c1003b0..f2f60433 100644 --- a/node/crates/c0mpute-core/src/hosting/mod.rs +++ b/node/crates/c0mpute-core/src/hosting/mod.rs @@ -13,9 +13,17 @@ //! chunks when this node holds them and otherwise fetched over libp2p from //! the fastest live holder (round-robin among equally fast ones), failing //! over to the next. +//! * Marketplace sites (v2 records) name preferred hosts picked from d1sks +//! ([`market`]). Holders are challenged with random chunk fetches and the +//! challenger signs availability receipts ([`proof`]); preferred holders +//! that pass are paid per GiB-month by the site's payer ([`settle`], +//! through [`crate::payments`], off unless the operator turns it on). pub mod dns; +pub mod market; +pub mod proof; pub mod replicas; +pub mod settle; pub mod site; use std::collections::{HashMap, HashSet}; @@ -39,6 +47,8 @@ use serde::{Deserialize, Serialize}; use tokio::sync::{RwLock, Semaphore}; use tracing::{debug, info, warn}; +use crate::payments::{Micros, Payee}; +use proof::Receipt; use replicas::{Action, ReplicaMap}; use site::{SiteManifest, SiteRecord}; @@ -100,7 +110,17 @@ pub enum HostingMsg { /// can fetch from a holder it has no connection to yet. #[serde(default)] addrs: Vec, + /// The hosting price this node asks to be a preferred host, USD per + /// GiB-month (decimal). Absent from older nodes. + #[serde(default, skip_serializing_if = "Option::is_none")] + price: Option, + /// Where it is paid for marketplace copies. + #[serde(default, skip_serializing_if = "Option::is_none")] + payout: Option, }, + /// A signed availability receipt (proof-of-storage challenge result). + /// Older nodes do not know this variant and skip it. + Receipt { receipt: Receipt }, } fn one() -> u32 { @@ -145,8 +165,44 @@ struct Inner { /// Optional handler for requests with a query string (podcast ad /// insertion installs one); it may decline and let the plain serve run. hook: std::sync::OnceLock>, + /// Signs challenge receipts (and is the payer identity for settlement). + challenger: Option, + challenger_hex: Option, + challenge: ChallengeScope, + challenge_every: Duration, + price: Option, + payout: Option, + /// Receipts this node signed, for settlement (kept on disk too). + own_receipts: RwLock>, + receipts_path: Option, } +/// Which sites a node challenges. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ChallengeScope { + Off, + /// Sites whose owner or payer is the challenger key. + Own, + /// Own sites, plus a sample of every other site's holders. + All, +} + +impl ChallengeScope { + pub fn parse(s: &str) -> Self { + match s.trim() { + "off" | "none" | "false" => ChallengeScope::Off, + "all" => ChallengeScope::All, + _ => ChallengeScope::Own, + } + } +} + +/// Most (site, holder) challenges in one round, so a gateway that knows many +/// sites stays at a few MiB of challenge traffic per round. +const CHALLENGES_PER_ROUND: usize = 32; +/// Receipts kept in memory/on disk for settlement. +const OWN_RECEIPTS_MAX: usize = 200_000; + /// A handler a gateway can put in front of plain site serving for requests /// that carry a query string (e.g. `?ads=1` on a podcast episode). #[async_trait::async_trait] @@ -166,6 +222,16 @@ pub struct Options { pub catalog_path: Option, pub timing: Timing, pub dns: dns::Verifier, + /// Key that signs proof-of-storage receipts; no challenges without it. + pub challenger: Option, + pub challenge: ChallengeScope, + pub challenge_every: Duration, + /// Hosting price this node asks (micro-USD per GiB-month), gossiped. + pub price: Option, + /// Payout address, gossiped so payers can pay this node. + pub payout: Option, + /// JSONL file of this node's own receipts (survives restarts). + pub receipts_path: Option, } impl Default for Options { @@ -176,6 +242,12 @@ impl Default for Options { catalog_path: None, timing: Timing::default(), dns: dns::Verifier::default(), + challenger: None, + challenge: ChallengeScope::Own, + challenge_every: Duration::from_secs(600), + price: None, + payout: None, + receipts_path: None, } } } @@ -190,6 +262,23 @@ impl Hosting { .and_then(|b| serde_json::from_slice::(&b).ok()) .unwrap_or_default(); let me = net.peer_id().to_base58(); + let own_receipts: Vec = opts + .receipts_path + .as_ref() + .and_then(|p| std::fs::read_to_string(p).ok()) + .map(|text| { + let mut v: Vec = text + .lines() + .filter_map(|l| serde_json::from_str::(l).ok()) + .filter(|r| r.verify()) + .collect(); + if v.len() > OWN_RECEIPTS_MAX { + v.drain(..v.len() - OWN_RECEIPTS_MAX); + } + v + }) + .unwrap_or_default(); + let challenger_hex = opts.challenger.as_ref().and_then(site::owner_hex); let inner = Arc::new(Inner { net: net.clone(), store, @@ -207,6 +296,14 @@ impl Hosting { fetch_slots: Semaphore::new(8), dns: opts.dns, hook: std::sync::OnceLock::new(), + challenger: opts.challenger, + challenger_hex, + challenge: opts.challenge, + challenge_every: opts.challenge_every, + price: opts.price, + payout: opts.payout.filter(|p| p.valid()), + own_receipts: RwLock::new(own_receipts), + receipts_path: opts.receipts_path, }); let h = Hosting(inner); // Sites held before a restart: keep them only if their chunks are @@ -221,6 +318,7 @@ impl Hosting { h.spawn_gossip_reader(); h.spawn_announcer(); h.spawn_repair(); + h.spawn_challenger(); Ok(h) } @@ -274,6 +372,8 @@ impl Hosting { parts, hosts, addrs, + price, + payout, } => { if let Some(pid) = msg.source { for a in addrs.iter().take(8) { @@ -293,12 +393,17 @@ impl Hosting { entry.0 += 1; if part + 1 >= parts { let (_, all) = partial.remove(&source).unwrap_or_default(); + let price = price.as_deref().and_then(|p| crate::payments::parse_usd(p).ok()); + let payout = payout.filter(|p| p.valid()); h.0.map .write() .await - .apply_holding_as(&source, all, Instant::now(), hosts); + .apply_holding_meta(&source, all, Instant::now(), hosts, price, payout); } } + HostingMsg::Receipt { receipt } => { + h.0.map.write().await.apply_receipt(receipt); + } } } }); @@ -331,12 +436,20 @@ impl Hosting { .iter() .map(|h| (h.site_id.clone(), (h.manifest.clone(), h.seq))) .collect(); + let (price, payout) = if self.0.hosts { + (self.0.price, self.0.payout.clone()) + } else { + (None, None) + }; if self.0.hosts || !mine.is_empty() { - self.0 - .map - .write() - .await - .apply_holding_as(&self.0.me, mine, Instant::now(), self.0.hosts); + self.0.map.write().await.apply_holding_meta( + &self.0.me, + mine, + Instant::now(), + self.0.hosts, + price, + payout.clone(), + ); } else { return; } @@ -363,6 +476,8 @@ impl Hosting { parts, hosts: self.0.hosts, addrs: addrs.clone(), + price: price.map(crate::payments::format_usd), + payout: payout.clone(), }) .await; } @@ -649,6 +764,198 @@ impl Hosting { v } + // ── proof of storage ───────────────────────────────────────────────── + + fn spawn_challenger(&self) { + if self.0.challenger.is_none() || self.0.challenge == ChallengeScope::Off { + return; + } + let h = self.clone(); + tokio::spawn(async move { + loop { + tokio::time::sleep(jitter(h.0.challenge_every)).await; + if h.is_stopped() { + return; + } + let n = h.challenge_round().await.len(); + if n > 0 { + debug!(receipts = n, "hosting: challenge round"); + } + } + }); + } + + /// The hex key receipts are signed with (the payer identity), if any. + pub fn challenger_key(&self) -> Option<&str> { + self.0.challenger_hex.as_deref() + } + + /// One challenge round: for each site in scope, ask every live holder + /// for one random chunk, straight from that peer, and sign a receipt. + /// Receipts are applied locally and gossiped. Returns them. + pub async fn challenge_round(&self) -> Vec { + let (Some(kp), Some(me_key)) = (self.0.challenger.clone(), self.0.challenger_hex.clone()) else { + return Vec::new(); + }; + if self.0.challenge == ChallengeScope::Off { + return Vec::new(); + } + let (mut own, mut others): (Vec<(SiteRecord, String)>, Vec<(SiteRecord, String)>) = (Vec::new(), Vec::new()); + { + let map = self.0.map.read().await; + for rec in map.records.values() { + let mine = rec.owner == me_key || rec.payer_key() == me_key; + if !mine && self.0.challenge != ChallengeScope::All { + continue; + } + for holder in map.holders(&rec.site_id) { + if holder == self.0.me { + continue; + } + if mine { + own.push((rec.clone(), holder)); + } else { + others.push((rec.clone(), holder)); + } + } + } + } + own.sort_by(|a, b| a.0.site_id.cmp(&b.0.site_id).then(a.1.cmp(&b.1))); + // Others: a random sample each round. + { + use rand::seq::SliceRandom; + others.shuffle(&mut rand::thread_rng()); + } + let picks: Vec<(SiteRecord, String)> = own.into_iter().chain(others).take(CHALLENGES_PER_ROUND).collect(); + let mut out = Vec::new(); + for (rec, holder) in picks { + match self.challenge_one(&kp, &rec, &holder).await { + Ok(r) => out.push(r), + Err(e) => debug!(site = %rec.site_id, %holder, err = %e, "hosting: no challenge"), + } + } + out + } + + /// Challenge one holder of one site and record the signed receipt. + pub async fn challenge_one( + &self, + kp: &c0mpute_net::libp2p::identity::Keypair, + rec: &SiteRecord, + holder: &str, + ) -> Result { + let manifest = self.manifest(rec).await?; + let nonce: [u8; 32] = rand::random(); + let chunk = proof::pick_chunk(&manifest, &nonce).context("site has no chunks")?; + let hash = Hash::from_hex(&chunk).context("bad chunk hash")?; + let pid: c0mpute_net::PeerId = holder.parse().context("holder is not a peer id")?; + let req = ChunkRequest { + chunk_hash: hash, + shard_index: None, + }; + let started = Instant::now(); + let got = tokio::time::timeout(self.0.timing.fetch_timeout, self.0.net.fetch_from_peer(pid, &req)).await; + let ok = matches!(&got, Ok(Ok(bytes)) if Hash::of(bytes) == hash); + let rtt_ms = if ok { started.elapsed().as_millis().min(u32::MAX as u128) as u32 } else { 0 }; + let receipt = Receipt::sign( + kp, + proof::Challenge { + site_id: &rec.site_id, + manifest: &rec.manifest, + holder, + chunk: &chunk, + ok, + rtt_ms, + at: crate::payments::now_secs(), + }, + )?; + self.record_receipt(receipt.clone()).await; + Ok(receipt) + } + + /// Keep a receipt this node signed: in the map, in the settlement log, + /// and on gossip. + async fn record_receipt(&self, r: Receipt) { + self.0.map.write().await.apply_receipt(r.clone()); + { + let mut own = self.0.own_receipts.write().await; + own.push(r.clone()); + if own.len() > OWN_RECEIPTS_MAX { + let extra = own.len() - OWN_RECEIPTS_MAX; + own.drain(..extra); + } + } + if let Some(p) = &self.0.receipts_path + && let Ok(line) = serde_json::to_string(&r) + { + use std::io::Write; + if let Some(d) = p.parent() { + let _ = std::fs::create_dir_all(d); + } + if let Ok(mut f) = std::fs::OpenOptions::new().create(true).append(true).open(p) { + let _ = writeln!(f, "{line}"); + } + } + self.publish(&HostingMsg::Receipt { receipt: r }).await; + } + + /// Receipts this node signed. + pub async fn own_receipts(&self) -> Vec { + self.0.own_receipts.read().await.clone() + } + + /// Receipts the network gossiped for one holder of one site. + pub async fn receipts_for(&self, site_id: &str, holder: &str) -> Vec { + self.0.map.read().await.receipts_for(site_id, holder).to_vec() + } + + /// The preferred hosts of a site that count right now. + pub async fn preferred(&self, site_id: &str) -> Vec { + self.0.map.read().await.effective_preferred(site_id) + } + + // ── settlement ─────────────────────────────────────────────────────── + + /// What this node (as payer) owes preferred holders for `[from, to)`. + pub async fn dues(&self, from: u64, to: u64) -> Vec { + let Some(payer) = self.0.challenger_hex.clone() else { + return Vec::new(); + }; + let (records, payouts, mut receipts) = { + let map = self.0.map.read().await; + let records: Vec = map.records.values().cloned().collect(); + let payouts: HashMap = map + .peers + .iter() + .filter_map(|(p, h)| h.payout.clone().map(|po| (p.clone(), po))) + .collect(); + let receipts: Vec = map.receipts.values().flatten().cloned().collect(); + (records, payouts, receipts) + }; + receipts.extend(self.own_receipts().await); + settle::dues(&records, &receipts, &payouts, &payer, from, to) + } + + /// One settlement round for `[from, to)`: work out the dues and pay them + /// through `rail` (skipping anything the ledger already paid, capped at + /// `cap` micro-USD). + pub async fn settle_round( + &self, + rail: &dyn crate::payments::Rail, + ledger: &crate::payments::Ledger, + from: u64, + to: u64, + cap: Micros, + ) -> (Vec, Vec<(crate::payments::Transfer, crate::payments::Outcome)>) { + let dues = self.dues(from, to).await; + let Some(payer) = self.0.challenger_hex.clone() else { + return (dues, Vec::new()); + }; + let transfers = settle::transfers(&dues, &payer); + let outcomes = crate::payments::pay_all(rail, ledger, &transfers, cap).await; + (dues, outcomes) + } + // ── fetching ───────────────────────────────────────────────────────── async fn fetch_chunk(&self, hash: &Hash, holders: &[String]) -> Result> { @@ -821,6 +1128,8 @@ async fn list_sites(State(h): State) -> Response { "domains": r.domains, "copies": copies, "path": format!("/s/{}/", r.site_id), + "preferred_hosts": r.preferred_hosts, + "price_per_gib_month": r.price_per_gib_month, }) }) .collect(); @@ -908,11 +1217,19 @@ mod tests { parts: 1, hosts: true, addrs: vec![], + price: Some("0.004".into()), + payout: None, }; let bytes = serde_json::to_vec(&m).unwrap(); assert!(bytes.len() < 1024); let back: HostingMsg = serde_json::from_slice(&bytes).unwrap(); assert!(matches!(back, HostingMsg::Holding { .. })); + // A holding message from an older node (no price/payout) still parses. + let old = br#"{"t":"holding","sites":[],"part":0,"parts":1,"hosts":true,"addrs":[]}"#; + assert!(matches!( + serde_json::from_slice::(old).unwrap(), + HostingMsg::Holding { price: None, payout: None, .. } + )); } #[test] @@ -930,6 +1247,11 @@ mod tests { parts: 1, hosts: true, addrs: vec!["/ip4/203.0.113.7/tcp/40001".into(); 8], + price: Some("0.004".into()), + payout: Some(Payee { + wallet: "0x".to_string() + &"a".repeat(40), + coin: "USDC_POL".into(), + }), }) .unwrap(); assert!(bytes.len() < 64 * 1024, "{} bytes", bytes.len()); @@ -953,6 +1275,68 @@ pub async fn publish_dir( name: &str, domains: Vec, wait: Duration, +) -> Result<(SiteRecord, usize)> { + publish_dir_with(dir, name, domains, wait, None).await +} + +/// Marketplace options for [`publish_dir_with`]. +#[derive(Clone, Debug)] +pub struct MarketOptions { + pub query: market::DiskQuery, + /// Preferred (paid) hosts to pick, 1-5. + pub copies: usize, + /// The OpenDisk marketplace to ask (discovery only). + pub d1sks_url: String, +} + +/// Pick preferred hosts from d1sks for a site of `size` bytes. `None` (with +/// a note on stderr) when d1sks is unreachable or nothing matches: the site +/// then publishes as a plain free site, which the network still keeps. +pub async fn pick_terms(m: &MarketOptions, size: u64) -> Option { + let disks = match market::query(&m.d1sks_url, &m.query).await { + Ok(d) => d, + Err(e) => { + eprintln!("d1sks: {e}; publishing without preferred hosts (the network still keeps 3 free copies)"); + return None; + } + }; + let ranked = market::rank(&disks, &m.query, size); + match market::choose(&ranked, m.copies, None) { + Some((chosen, terms)) => { + eprintln!( + "marketplace: {} preferred host(s) at ${}/GiB-month:", + chosen.len(), + crate::payments::format_usd(terms.price_per_gib_month) + ); + for d in &chosen { + eprintln!( + " {} {} ${}/GiB-month {}", + d.peer_id, + d.name, + crate::payments::format_usd(d.price), + d.countries.join(",") + ); + } + Some(terms) + } + None => { + eprintln!( + "d1sks: no c0mpute hosting disk matches ({} listed); publishing without preferred hosts", + disks.len() + ); + None + } + } +} + +/// [`publish_dir`], optionally picking paid hosts from the d1sks marketplace +/// first and signing them into a v2 record. +pub async fn publish_dir_with( + dir: &std::path::Path, + name: &str, + domains: Vec, + wait: Duration, + market: Option, ) -> Result<(SiteRecord, usize)> { for d in &domains { if !site::valid_domain(d) { @@ -967,13 +1351,18 @@ pub async fn publish_dir( .duration_since(std::time::UNIX_EPOCH) .map(|d| d.as_millis() as u64) .unwrap_or(0); - let record = SiteRecord::sign( + let terms = match &market { + Some(m) => pick_terms(m, built.manifest.total_bytes()).await, + None => None, + }; + let record = SiteRecord::sign_with_terms( &keypair, name, &manifest_hash, seq, built.manifest.total_bytes(), domains, + terms, )?; let hosting = Hosting::start(supervisor.libp2p.clone(), supervisor.store.clone(), Options::default()).await?; // Let the swarm reach peers before the first gossip. @@ -991,12 +1380,25 @@ pub async fn publish_dir( while started.elapsed() < wait { tokio::time::sleep(Duration::from_secs(5)).await; hosting.regossip(&record).await; - let others = hosting.holders(&record.site_id).await.len(); + let holders = hosting.holders(&record.site_id).await; + let others = holders.len(); if others != copies { copies = others; - eprintln!(" {copies}/{TARGET_REPLICAS} network copies"); + let paid = holders.iter().filter(|h| record.prefers(h)).count(); + if record.preferred_hosts.is_empty() { + eprintln!(" {copies}/{TARGET_REPLICAS} network copies"); + } else { + eprintln!( + " {copies}/{TARGET_REPLICAS} network copies ({paid}/{} preferred)", + record.preferred_hosts.len() + ); + } } - if copies >= TARGET_REPLICAS { + let paid_done = record.preferred_hosts.is_empty() + || holders.iter().filter(|h| record.prefers(h)).count() >= record.preferred_hosts.len(); + // A marketplace site waits a little longer for its preferred hosts, + // but never past `wait`. + if copies >= TARGET_REPLICAS && (paid_done || started.elapsed() > wait / 2) { break; } } diff --git a/node/crates/c0mpute-core/src/hosting/proof.rs b/node/crates/c0mpute-core/src/hosting/proof.rs new file mode 100644 index 00000000..c16d762c --- /dev/null +++ b/node/crates/c0mpute-core/src/hosting/proof.rs @@ -0,0 +1,234 @@ +//! Proof of storage for hosted sites: anyone challenges a holder by fetching +//! one random chunk of the site straight from that peer, checks it against +//! its hash, and signs an availability [`Receipt`] saying whether it passed. +//! +//! Receipts are gossiped on the hosting topic, so every node sees the same +//! history. Two things read them: +//! +//! * placement: a preferred host whose recent receipts from the site's owner +//! or payer are mostly failures loses its preference ([`demoted`]), and the +//! normal rendezvous fill takes over; +//! * settlement: the payer pays preferred holders in proportion to the +//! challenges they passed (`settle`). +//! +//! What this proves: the peer returned the right bytes for a chunk nobody +//! could predict, within the timeout. It does not prove the peer keeps its +//! own copy rather than relaying from another holder; relaying costs the +//! relayer bandwidth and latency, and the rest of the network still has the +//! data, so the site stays up either way. + +use anyhow::Result; +use c0mpute_proto::Hash; +use serde::{Deserialize, Serialize}; + +use super::site::SiteManifest; + +const SIGNING_DOMAIN: &str = "c0mpute-hosting-receipt-v1"; + +/// Receipts per (site, holder) a node keeps for placement decisions. +pub const KEEP_PER_HOLDER: usize = 16; +/// A preferred host is judged on its most recent this-many trusted receipts… +pub const RECENT: usize = 5; +/// …and loses preference when at least this many of them failed. +pub const FAIL_LIMIT: usize = 3; + +/// "I asked `holder` for chunk `chunk` of `site_id` at `at`; it passed or not." +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Receipt { + pub v: u8, + pub site_id: String, + /// Manifest (version) the chunk belongs to. + pub manifest: String, + /// The challenged peer (base58 peer id). + pub holder: String, + /// The chunk asked for (hex blake3). + pub chunk: String, + pub ok: bool, + /// Round-trip of the fetch, ms (0 on a failure). + pub rtt_ms: u32, + /// Unix seconds. + pub at: u64, + /// Challenger's ed25519 public key, hex. + pub challenger: String, + pub sig: String, +} + +/// The unsigned fields of a receipt. +pub struct Challenge<'a> { + pub site_id: &'a str, + pub manifest: &'a str, + pub holder: &'a str, + pub chunk: &'a str, + pub ok: bool, + pub rtt_ms: u32, + pub at: u64, +} + +impl Receipt { + pub fn signing_bytes(&self) -> Vec { + format!( + "{SIGNING_DOMAIN}\n{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}", + self.site_id, self.manifest, self.holder, self.chunk, self.ok, self.rtt_ms, self.at, self.challenger + ) + .into_bytes() + } + + pub fn sign(keypair: &c0mpute_net::libp2p::identity::Keypair, c: Challenge<'_>) -> Result { + let challenger = super::site::owner_hex(keypair).ok_or_else(|| anyhow::anyhow!("challenger needs an ed25519 key"))?; + let mut r = Receipt { + v: 1, + site_id: c.site_id.to_string(), + manifest: c.manifest.to_string(), + holder: c.holder.to_string(), + chunk: c.chunk.to_string(), + ok: c.ok, + rtt_ms: c.rtt_ms, + at: c.at, + challenger, + sig: String::new(), + }; + let sig = keypair + .sign(&r.signing_bytes()) + .map_err(|e| anyhow::anyhow!("sign receipt: {e}"))?; + r.sig = hex::encode(sig); + Ok(r) + } + + /// True when the fields are sane and the challenger signed them. + pub fn verify(&self) -> bool { + if self.v != 1 + || self.site_id.len() != 20 + || self.holder.is_empty() + || self.holder.len() > 64 + || Hash::from_hex(&self.manifest).is_err() + || Hash::from_hex(&self.chunk).is_err() + { + return false; + } + let (Ok(key), Ok(sig)) = (hex::decode(&self.challenger), hex::decode(&self.sig)) else { + return false; + }; + let Ok(public) = c0mpute_net::libp2p::identity::ed25519::PublicKey::try_from_bytes(&key) else { + return false; + }; + public.verify(&self.signing_bytes(), &sig) + } +} + +/// The chunk to ask for, picked by `nonce` (the challenger passes fresh +/// randomness, so a holder cannot keep only the chunks it expects). +pub fn pick_chunk(manifest: &SiteManifest, nonce: &[u8]) -> Option { + let all = manifest.chunk_hashes(); + if all.is_empty() { + return None; + } + let h = Hash::of(nonce).0; + let n = u64::from_le_bytes(h[..8].try_into().expect("8 bytes")); + Some(all[(n % all.len() as u64) as usize].clone()) +} + +/// Whether a holder has lost preference: of its most recent [`RECENT`] +/// receipts signed by a `trusted` key (the site's owner or payer), at least +/// [`FAIL_LIMIT`] failed. `receipts` may be in any order and may include +/// untrusted ones; those are ignored, so a stranger cannot demote anybody. +pub fn demoted(receipts: &[Receipt], trusted: &[&str]) -> bool { + let mut mine: Vec<&Receipt> = receipts + .iter() + .filter(|r| trusted.contains(&r.challenger.as_str())) + .collect(); + mine.sort_by(|a, b| b.at.cmp(&a.at).then(a.sig.cmp(&b.sig))); + mine.iter().take(RECENT).filter(|r| !r.ok).count() >= FAIL_LIMIT +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hosting::site::FileEntry; + use std::collections::BTreeMap; + + fn manifest(n: usize) -> SiteManifest { + let mut files = BTreeMap::new(); + for i in 0..n { + files.insert( + format!("f{i}"), + FileEntry { + size: 1, + content_type: "text/plain".into(), + chunks: vec![Hash::of(&[i as u8]).to_hex()], + }, + ); + } + SiteManifest { + v: 1, + name: "s".into(), + files, + } + } + + fn receipt(kp: &c0mpute_net::libp2p::identity::Keypair, ok: bool, at: u64) -> Receipt { + let m = Hash::of(b"m").to_hex(); + let c = Hash::of(b"c").to_hex(); + Receipt::sign( + kp, + Challenge { + site_id: "0123456789abcdef0123", + manifest: &m, + holder: "peer", + chunk: &c, + ok, + rtt_ms: 5, + at, + }, + ) + .unwrap() + } + + #[test] + fn receipts_verify_and_reject_tampering() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let r = receipt(&kp, true, 100); + assert!(r.verify()); + let back: Receipt = serde_json::from_slice(&serde_json::to_vec(&r).unwrap()).unwrap(); + assert!(back.verify()); + let mut t = r.clone(); + t.ok = false; + assert!(!t.verify(), "a pass cannot be turned into a fail"); + let mut t = r.clone(); + t.holder = "other".into(); + assert!(!t.verify()); + let mut t = r.clone(); + t.at = 101; + assert!(!t.verify()); + } + + #[test] + fn chunk_choice_follows_the_nonce() { + let m = manifest(50); + let picks: std::collections::HashSet = + (0u32..64).filter_map(|i| pick_chunk(&m, &i.to_le_bytes())).collect(); + assert!(picks.len() > 20, "random nonces spread over the site: {}", picks.len()); + assert_eq!(pick_chunk(&m, b"x"), pick_chunk(&m, b"x")); + assert!(pick_chunk(&manifest(0), b"x").is_none()); + } + + #[test] + fn only_trusted_recent_failures_demote() { + let owner = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let stranger = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let owner_hex = crate::hosting::site::owner_hex(&owner).unwrap(); + let trusted = [owner_hex.as_str()]; + // Strangers' failures do nothing. + let noise: Vec = (0..10).map(|i| receipt(&stranger, false, i)).collect(); + assert!(!demoted(&noise, &trusted)); + // Two recent owner failures: still preferred. Three: demoted. + let mut rs = vec![receipt(&owner, true, 1), receipt(&owner, false, 2), receipt(&owner, false, 3)]; + assert!(!demoted(&rs, &trusted)); + rs.push(receipt(&owner, false, 4)); + assert!(demoted(&rs, &trusted)); + // Passing again pushes the failures out of the window. + for at in 5..10 { + rs.push(receipt(&owner, true, at)); + } + assert!(!demoted(&rs, &trusted)); + } +} diff --git a/node/crates/c0mpute-core/src/hosting/replicas.rs b/node/crates/c0mpute-core/src/hosting/replicas.rs index 797b1424..5af1c97a 100644 --- a/node/crates/c0mpute-core/src/hosting/replicas.rs +++ b/node/crates/c0mpute-core/src/hosting/replicas.rs @@ -6,13 +6,22 @@ //! too few live copies, every node ranks the candidate peers the same way //! (rendezvous hashing on site id + peer id) and the top-ranked ones take a //! copy. Nobody is elected; everybody computes the same answer. +//! +//! Marketplace sites (v2 records) name preferred hosts. A preferred host that +//! is live, asks no more than the agreed price and has not lost preference +//! through failed challenges replicates whether or not the site already has +//! three copies; the rendezvous fill still brings every site to three copies +//! on its own, and once preferred hosts hold copies the surplus free copies +//! are released one at a time. A site never drops below three copies. use std::collections::HashMap; use std::time::{Duration, Instant}; use c0mpute_proto::Hash; +use super::proof::{self, Receipt}; use super::site::SiteRecord; +use crate::payments::{Micros, Payee}; /// What one peer says it holds: site id -> (manifest hash, seq). #[derive(Clone, Debug, Default)] @@ -22,6 +31,10 @@ pub struct Holding { /// A hosting node (its copies count toward the target). A publisher that /// is only seeding its own site until the network has copies does not. pub hosts: bool, + /// The hosting price it asks (micro-USD per GiB-month), when it says. + pub price: Option, + /// Where it wants marketplace payments, when it says. + pub payout: Option, } #[derive(Debug, Default)] @@ -32,6 +45,8 @@ pub struct ReplicaMap { pub peers: HashMap, /// Smoothed fetch round-trip per peer, in ms. pub rtt_ms: HashMap, + /// Recent availability receipts per (site id, holder), newest last. + pub receipts: HashMap<(String, String), Vec>, } impl ReplicaMap { @@ -61,6 +76,19 @@ impl ReplicaMap { sites: HashMap, now: Instant, hosts: bool, + ) { + self.apply_holding_meta(peer, sites, now, hosts, None, None); + } + + /// A holding list plus what the peer asks for and where it is paid. + pub fn apply_holding_meta( + &mut self, + peer: &str, + sites: HashMap, + now: Instant, + hosts: bool, + price: Option, + payout: Option, ) { self.peers.insert( peer.to_string(), @@ -68,10 +96,72 @@ impl ReplicaMap { sites, seen: Some(now), hosts, + price, + payout, }, ); } + /// Keep a verified receipt (deduplicated, newest [`proof::KEEP_PER_HOLDER`] + /// per holder). Returns true when it was new. + pub fn apply_receipt(&mut self, r: Receipt) -> bool { + if !r.verify() { + return false; + } + let list = self.receipts.entry((r.site_id.clone(), r.holder.clone())).or_default(); + if list.iter().any(|x| x.sig == r.sig) { + return false; + } + list.push(r); + list.sort_by(|a, b| a.at.cmp(&b.at).then(a.sig.cmp(&b.sig))); + if list.len() > proof::KEEP_PER_HOLDER { + let extra = list.len() - proof::KEEP_PER_HOLDER; + list.drain(..extra); + } + true + } + + /// Receipts kept for one holder of one site, oldest first. + pub fn receipts_for(&self, site_id: &str, holder: &str) -> &[Receipt] { + self.receipts + .get(&(site_id.to_string(), holder.to_string())) + .map(|v| v.as_slice()) + .unwrap_or(&[]) + } + + /// True when `holder` failed too many recent challenges by the site's + /// owner or payer (see [`proof::demoted`]). + pub fn is_demoted(&self, site_id: &str, holder: &str) -> bool { + let Some(rec) = self.records.get(site_id) else { + return false; + }; + let trusted = [rec.owner.as_str(), rec.payer_key()]; + proof::demoted(self.receipts_for(site_id, holder), &trusted) + } + + /// The preferred hosts that count right now, in the record's order: live + /// hosting peers that ask no more than the agreed price and have not lost + /// preference. + pub fn effective_preferred(&self, site_id: &str) -> Vec { + let Some(rec) = self.records.get(site_id) else { + return Vec::new(); + }; + let Some(terms) = rec.terms() else { + return Vec::new(); + }; + terms + .preferred_hosts + .into_iter() + .filter(|p| { + self.peers.get(p).is_some_and(|h| { + h.hosts && h.price.is_none_or(|ask| ask <= terms.price_per_gib_month) + }) + }) + .filter(|p| !self.is_demoted(site_id, p)) + .take(super::site::MAX_PREFERRED_HOSTS) + .collect() + } + /// Forget peers not heard from within `ttl`. pub fn prune(&mut self, now: Instant, ttl: Duration) { self.peers @@ -210,19 +300,41 @@ pub enum Action { pub fn decide(map: &ReplicaMap, site_id: &str, me: &str, i_host: bool, target: usize, max: usize) -> Action { let holders = map.holders(site_id); let i_hold = holders.iter().any(|p| p == me); + let preferred = map.effective_preferred(site_id); if i_hold { - if surplus(site_id, &holders, max).as_deref() == Some(me) { + if preferred.is_empty() { + if surplus(site_id, &holders, max).as_deref() == Some(me) { + return Action::Drop; + } + return Action::Nothing; + } + // Marketplace site: free copies make room for paid ones, never below + // `target` in total. One release per round, lowest rank first. + let free: Vec = holders.iter().filter(|p| !preferred.contains(p)).cloned().collect(); + let paid = holders.len() - free.len(); + let allowed = target.saturating_sub(paid); + if free.len() > allowed && surplus(site_id, &free, 0).as_deref() == Some(me) { return Action::Drop; } return Action::Nothing; } - if !i_host || holders.len() >= target { + if !i_host { + return Action::Nothing; + } + // A preferred host takes its copy whatever the count: it is paid to. + if preferred.iter().any(|p| p == me) { + return Action::Replicate; + } + if holders.len() >= target { return Action::Nothing; } + // The rendezvous fill. Preferred hosts that do not hold yet are left out: + // they replicate on their own, and counting on them would leave the site + // short if one never manages to. let mut candidates: Vec = map .live_peers() .into_iter() - .filter(|p| !holders.contains(p)) + .filter(|p| !holders.contains(p) && !preferred.contains(p)) .collect(); if !candidates.iter().any(|p| p == me) { candidates.push(me.to_string()); @@ -345,6 +457,186 @@ mod tests { assert!(m.live_peers().is_empty()); } + fn pid() -> String { + c0mpute_net::libp2p::identity::Keypair::generate_ed25519() + .public() + .to_peer_id() + .to_base58() + } + + fn paid_rec(kp: &c0mpute_net::libp2p::identity::Keypair, preferred: Vec, price: u64) -> SiteRecord { + SiteRecord::sign_with_terms( + kp, + "paid", + &Hash::of(b"paid"), + 1, + 10, + vec![], + Some(super::super::site::HostingTerms { + preferred_hosts: preferred, + price_per_gib_month: price, + payer: None, + }), + ) + .unwrap() + } + + /// Run repair rounds over a shared map until nothing changes; every peer + /// decides from the same map, like the real network once gossip settles. + /// Returns the holder count after each round. + fn settle(m: &mut ReplicaMap, site: &str, manifest: &str, peers: &[String]) -> Vec { + let mut counts = Vec::new(); + for _ in 0..20 { + let actions: Vec<(String, Action)> = peers + .iter() + .map(|p| (p.clone(), decide(m, site, p, true, 3, 5))) + .collect(); + if actions.iter().all(|(_, a)| *a == Action::Nothing) { + break; + } + let now = Instant::now(); + for (p, a) in actions { + let mut h = m.peers.get(&p).cloned().unwrap_or_default(); + match a { + Action::Replicate => { + h.sites.insert(site.to_string(), (manifest.to_string(), 1)); + } + Action::Drop => { + h.sites.remove(site); + } + Action::Nothing => {} + } + m.apply_holding_meta(&p, h.sites, now, true, h.price, h.payout); + } + counts.push(m.holders(site).len()); + } + counts + } + + fn hosting_peers(m: &mut ReplicaMap, peers: &[String]) { + for p in peers { + m.apply_holding(p, HashMap::new(), Instant::now()); + } + } + + #[test] + fn preferred_hosts_replicate_first_and_the_site_settles_at_three() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let peers: Vec = (0..5).map(|_| pid()).collect(); + let preferred = vec![peers[1].clone(), peers[3].clone()]; + let r = paid_rec(&kp, preferred.clone(), 4_000); + let mut m = ReplicaMap::default(); + m.apply_record(r.clone()); + hosting_peers(&mut m, &peers); + assert_eq!(m.effective_preferred(&r.site_id), preferred); + // Round one: both preferred hosts replicate, whatever their rank. + for p in &preferred { + assert_eq!(decide(&m, &r.site_id, p, true, 3, 5), Action::Replicate); + } + let counts = settle(&mut m, &r.site_id, &r.manifest, &peers); + assert!(counts.iter().all(|c| *c >= 3), "never below three copies: {counts:?}"); + let holders = m.holders(&r.site_id); + assert_eq!(holders.len(), 3, "settles at the target: {counts:?}"); + for p in &preferred { + assert!(holders.contains(p), "preferred {p} holds a copy"); + } + } + + #[test] + fn preferred_placement_is_the_same_on_every_node() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let peers: Vec = (0..7).map(|_| pid()).collect(); + let r = paid_rec(&kp, vec![peers[6].clone()], 4_000); + let run = |order: Vec| { + let mut m = ReplicaMap::default(); + m.apply_record(r.clone()); + hosting_peers(&mut m, &order); + settle(&mut m, &r.site_id, &r.manifest, &order); + m.holders(&r.site_id) + }; + let a = run(peers.clone()); + let mut rev = peers.clone(); + rev.reverse(); + let b = run(rev); + assert_eq!(a, b, "same inputs, same copies, whatever order peers are seen in"); + assert_eq!(a.len(), 3); + assert!(a.contains(&peers[6])); + } + + #[test] + fn without_live_preferred_hosts_the_fill_still_keeps_three() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let peers: Vec = (0..5).map(|_| pid()).collect(); + // Both preferred hosts are offline (not in the map). + let r = paid_rec(&kp, vec![pid(), pid()], 4_000); + let mut m = ReplicaMap::default(); + m.apply_record(r.clone()); + hosting_peers(&mut m, &peers); + assert!(m.effective_preferred(&r.site_id).is_empty()); + settle(&mut m, &r.site_id, &r.manifest, &peers); + assert_eq!(m.holders(&r.site_id).len(), 3); + } + + #[test] + fn a_preferred_host_asking_more_than_the_agreed_price_is_not_preferred() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let peers: Vec = (0..4).map(|_| pid()).collect(); + let r = paid_rec(&kp, vec![peers[0].clone(), peers[1].clone()], 4_000); + let mut m = ReplicaMap::default(); + m.apply_record(r.clone()); + hosting_peers(&mut m, &peers); + m.apply_holding_meta(&peers[0], HashMap::new(), Instant::now(), true, Some(9_000), None); + m.apply_holding_meta(&peers[1], HashMap::new(), Instant::now(), true, Some(4_000), None); + assert_eq!(m.effective_preferred(&r.site_id), vec![peers[1].clone()]); + } + + #[test] + fn a_preferred_host_that_fails_challenges_loses_preference() { + use super::super::proof::{Challenge, Receipt}; + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let peers: Vec = (0..5).map(|_| pid()).collect(); + let bad = peers[2].clone(); + let r = paid_rec(&kp, vec![bad.clone()], 4_000); + let mut m = ReplicaMap::default(); + m.apply_record(r.clone()); + hosting_peers(&mut m, &peers); + settle(&mut m, &r.site_id, &r.manifest, &peers); + assert!(m.holders(&r.site_id).contains(&bad)); + let chunk = Hash::of(b"c").to_hex(); + let fail = |signer: &c0mpute_net::libp2p::identity::Keypair, at: u64| { + Receipt::sign( + signer, + Challenge { + site_id: &r.site_id, + manifest: &r.manifest, + holder: &bad, + chunk: &chunk, + ok: false, + rtt_ms: 0, + at, + }, + ) + .unwrap() + }; + // A stranger's failures change nothing. + let stranger = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + for at in 0..5 { + assert!(m.apply_receipt(fail(&stranger, at))); + } + assert_eq!(m.effective_preferred(&r.site_id), vec![bad.clone()]); + // Three failures signed by the owner: demoted. + for at in 10..13 { + assert!(m.apply_receipt(fail(&kp, at))); + } + assert!(!m.apply_receipt(fail(&kp, 12)), "duplicates are ignored"); + assert!(m.is_demoted(&r.site_id, &bad)); + assert!(m.effective_preferred(&r.site_id).is_empty()); + // The site still has three copies; the demoted host is now an + // ordinary holder. + settle(&mut m, &r.site_id, &r.manifest, &peers); + assert!(m.holders(&r.site_id).len() >= 3); + } + #[test] fn proxy_order_prefers_fast_and_rotates_ties() { let mut m = ReplicaMap::default(); diff --git a/node/crates/c0mpute-core/src/hosting/settle.rs b/node/crates/c0mpute-core/src/hosting/settle.rs new file mode 100644 index 00000000..c670d2f3 --- /dev/null +++ b/node/crates/c0mpute-core/src/hosting/settle.rs @@ -0,0 +1,234 @@ +//! Settlement for marketplace sites: what a payer owes each preferred holder +//! for a window of time, from the availability receipts it trusts. +//! +//! Pure functions, no I/O. The node runs them over its own receipts and +//! hands the result to a [`crate::payments::Rail`]; with payments off +//! (the default) it only logs what it would pay. +//! +//! The rule, per preferred holder of a site whose payer is this node: +//! +//! ```text +//! owed = price_per_gib_month × site_GiB × (window / 30 days) × passed / challenged +//! ``` +//! +//! counting only receipts in the window signed by the site's owner or payer. +//! A holder that passed fewer than half its challenges, or was never +//! challenged, is owed nothing for the window. Amounts round down to the +//! micro-USD. + +use std::collections::HashMap; + +use super::proof::Receipt; +use super::site::SiteRecord; +use crate::payments::{Micros, Payee, Transfer}; + +/// The month in GiB-month: 30 days. +pub const MONTH_SECS: u64 = 30 * 24 * 60 * 60; +const GIB: u128 = 1024 * 1024 * 1024; + +/// Owed for holding `size` bytes at `price` micro-USD per GiB-month for +/// `secs` seconds, having passed `passed` of `challenged` challenges. +pub fn owed(price: Micros, size: u64, secs: u64, passed: u64, challenged: u64) -> Micros { + if challenged == 0 || passed * 2 < challenged { + return 0; + } + let full = (price as u128 * size as u128).saturating_mul(secs as u128) / (GIB * MONTH_SECS as u128); + let earned = full.saturating_mul(passed as u128) / challenged as u128; + earned.min(u64::MAX as u128) as Micros +} + +/// One holder's due for one site and window. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct Due { + pub site_id: String, + pub holder: String, + pub payee: Option, + pub amount: Micros, + pub passed: u64, + pub challenged: u64, + /// Idempotency key: same site, holder and window, same payment. + pub key: String, +} + +/// Everything `payer` owes for `[from, to)` (unix seconds). +pub fn dues( + records: &[SiteRecord], + receipts: &[Receipt], + payouts: &HashMap, + payer: &str, + from: u64, + to: u64, +) -> Vec { + if to <= from { + return Vec::new(); + } + let mut out = Vec::new(); + for rec in records { + let Some(terms) = rec.terms() else { + continue; + }; + if rec.payer_key() != payer || !rec.verify() { + continue; + } + let trusted = [rec.owner.as_str(), rec.payer_key()]; + for holder in &terms.preferred_hosts { + let mut seen = std::collections::HashSet::new(); + let (mut passed, mut challenged) = (0u64, 0u64); + for r in receipts { + if r.site_id == rec.site_id + && r.holder == *holder + && (from..to).contains(&r.at) + && trusted.contains(&r.challenger.as_str()) + && seen.insert(r.sig.clone()) + && r.verify() + { + challenged += 1; + if r.ok { + passed += 1; + } + } + } + let amount = owed(terms.price_per_gib_month, rec.size, to - from, passed, challenged); + out.push(Due { + site_id: rec.site_id.clone(), + holder: holder.clone(), + payee: payouts.get(holder).cloned(), + amount, + passed, + challenged, + key: format!("c0mpute-hosting:{}:{}:{from}:{to}", rec.site_id, holder), + }); + } + } + out.sort_by(|a, b| a.site_id.cmp(&b.site_id).then(a.holder.cmp(&b.holder))); + out +} + +/// The transfers to make: dues with something owed and a payout address. +pub fn transfers(dues: &[Due], payer: &str) -> Vec { + dues.iter() + .filter(|d| d.amount > 0) + .filter_map(|d| { + let payee = d.payee.clone().filter(|p| p.valid())?; + Some(Transfer { + key: d.key.clone(), + payer: payer.to_string(), + payee, + amount: d.amount, + memo: format!( + "c0mpute hosting: site {} held by {} ({}/{} challenges passed)", + d.site_id, d.holder, d.passed, d.challenged + ), + }) + }) + .collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::hosting::proof::Challenge; + use crate::hosting::site::HostingTerms; + use c0mpute_proto::Hash; + + const MIB: u64 = 1024 * 1024; + + #[test] + fn owed_follows_price_size_time_and_passes() { + // $1/GiB-month, 1 GiB, a whole month, every challenge passed = $1. + assert_eq!(owed(1_000_000, 1024 * MIB, MONTH_SECS, 10, 10), 1_000_000); + // Half the size, half the time = a quarter. + assert_eq!(owed(1_000_000, 512 * MIB, MONTH_SECS / 2, 4, 4), 250_000); + // 3 of 4 passed = three quarters. + assert_eq!(owed(1_000_000, 1024 * MIB, MONTH_SECS, 3, 4), 750_000); + // Under half passed, or never challenged: nothing. + assert_eq!(owed(1_000_000, 1024 * MIB, MONTH_SECS, 1, 3), 0); + assert_eq!(owed(1_000_000, 1024 * MIB, MONTH_SECS, 0, 0), 0); + // Exactly half passes. + assert_eq!(owed(1_000_000, 1024 * MIB, MONTH_SECS, 2, 4), 500_000); + // Small sites round down, never up. + assert_eq!(owed(4_000, MIB, 3600, 1, 1), 0); + // No overflow at the extremes. + assert!(owed(u64::MAX, u64::MAX, u64::MAX, 1, 1) > 0); + } + + fn pid() -> String { + c0mpute_net::libp2p::identity::Keypair::generate_ed25519() + .public() + .to_peer_id() + .to_base58() + } + + #[test] + fn dues_count_only_trusted_receipts_in_the_window() { + let owner = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let owner_hex = crate::hosting::site::owner_hex(&owner).unwrap(); + let stranger = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let (a, b) = (pid(), pid()); + let rec = SiteRecord::sign_with_terms( + &owner, + "paid", + &Hash::of(b"m"), + 1, + 256 * MIB, + vec![], + Some(HostingTerms { + preferred_hosts: vec![a.clone(), b.clone()], + price_per_gib_month: 8_000_000, + payer: None, + }), + ) + .unwrap(); + let chunk = Hash::of(b"c").to_hex(); + let r = |kp: &c0mpute_net::libp2p::identity::Keypair, holder: &str, ok: bool, at: u64| { + Receipt::sign( + kp, + Challenge { + site_id: &rec.site_id, + manifest: &rec.manifest, + holder, + chunk: &chunk, + ok, + rtt_ms: 1, + at, + }, + ) + .unwrap() + }; + let mut receipts = vec![ + r(&owner, &a, true, 100), + r(&owner, &a, true, 200), + r(&owner, &b, true, 100), + r(&owner, &b, false, 200), + r(&owner, &b, false, 300), + // Outside the window, or not trusted: ignored. + r(&owner, &a, false, 5_000_000), + r(&stranger, &a, false, 150), + r(&stranger, &b, true, 150), + ]; + receipts.push(receipts[0].clone()); // a duplicate counts once + let mut payouts = HashMap::new(); + payouts.insert( + a.clone(), + Payee { + wallet: "0xaaaaaaaaaaaaaaaa".into(), + coin: "USDC_POL".into(), + }, + ); + let ds = dues(&[rec.clone()], &receipts, &payouts, &owner_hex, 0, MONTH_SECS); + assert_eq!(ds.len(), 2); + let da = ds.iter().find(|d| d.holder == a).unwrap(); + let db = ds.iter().find(|d| d.holder == b).unwrap(); + assert_eq!((da.passed, da.challenged), (2, 2)); + assert_eq!(da.amount, 2_000_000, "a quarter GiB, a month, $8/GiB-month, all passed"); + assert_eq!((db.passed, db.challenged), (1, 3)); + assert_eq!(db.amount, 0, "under half passed"); + // Only `a` is owed and has a payout address. + let ts = transfers(&ds, &owner_hex); + assert_eq!(ts.len(), 1); + assert_eq!(ts[0].amount, 2_000_000); + assert_eq!(ts[0].key, da.key); + // Somebody else's sites are not ours to pay. + assert!(dues(&[rec], &receipts, &payouts, &"c".repeat(64), 0, MONTH_SECS).is_empty()); + } +} diff --git a/node/crates/c0mpute-core/src/hosting/site.rs b/node/crates/c0mpute-core/src/hosting/site.rs index c2eeee30..12a21c3e 100644 --- a/node/crates/c0mpute-core/src/hosting/site.rs +++ b/node/crates/c0mpute-core/src/hosting/site.rs @@ -28,6 +28,12 @@ pub const MAX_NAME_CHARS: usize = 63; pub const MAX_DOMAINS: usize = 16; const SIGNING_DOMAIN: &str = "c0mpute-hosting-v1"; +/// Records that carry marketplace terms (v2) sign under their own domain, so +/// a v2 signature can never be replayed as a v1 record or the other way round. +const SIGNING_DOMAIN_V2: &str = "c0mpute-hosting-v2"; +/// Most preferred (paid) hosts one record can name. Equal to the replica +/// ceiling, so preferred copies alone never push a site over it. +pub const MAX_PREFERRED_HOSTS: usize = 5; /// One file: its size, MIME type and the chunks that make it up, in order. #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] @@ -109,10 +115,39 @@ pub struct SiteRecord { /// after DNS confirms it (see `dns`). #[serde(default)] pub domains: Vec, + /// v2: hosting nodes (base58 peer ids) the publisher picked, usually from + /// the d1sks marketplace, and pays per GiB-month. They replicate first; + /// the normal rendezvous fill still keeps at least three copies. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub preferred_hosts: Vec, + /// v2: the agreed price per GiB-month, USD decimal ("0.004"). + #[serde(default, skip_serializing_if = "Option::is_none")] + pub price_per_gib_month: Option, + /// v2: who pays (hex ed25519 key); the owner when absent. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub payer: Option, /// ed25519 signature (hex) over [`SiteRecord::signing_bytes`]. pub sig: String, } +/// Marketplace terms a v2 record carries. +#[derive(Clone, Debug, PartialEq, Eq, Default)] +pub struct HostingTerms { + pub preferred_hosts: Vec, + /// Micro-USD per GiB-month. + pub price_per_gib_month: crate::payments::Micros, + /// Hex ed25519 key of the payer; `None` = the owner pays. + pub payer: Option, +} + +fn valid_peer_id(p: &str) -> bool { + p.len() <= 64 && p.parse::().is_ok() +} + +fn valid_key_hex(k: &str) -> bool { + k.len() == 64 && k.chars().all(|c| c.is_ascii_digit() || ('a'..='f').contains(&c)) +} + /// The id a site with this owner and name always has: 20 hex characters of /// blake3, so it fits a DNS label and a URL path. pub fn site_id_for(owner_hex: &str, name: &str) -> String { @@ -156,8 +191,9 @@ pub fn valid_path(p: &str) -> bool { impl SiteRecord { pub fn signing_bytes(&self) -> Vec { - format!( - "{SIGNING_DOMAIN}\n{}\n{}\n{}\n{}\n{}\n{}\n{}", + let base = format!( + "{}\n{}\n{}\n{}\n{}\n{}\n{}\n{}", + if self.v >= 2 { SIGNING_DOMAIN_V2 } else { SIGNING_DOMAIN }, self.site_id, self.name, self.owner, @@ -165,18 +201,55 @@ impl SiteRecord { self.seq, self.size, self.domains.join(",") + ); + if self.v < 2 { + return base.into_bytes(); + } + format!( + "{base}\n{}\n{}\n{}", + self.preferred_hosts.join(","), + self.price_per_gib_month.as_deref().unwrap_or(""), + self.payer.as_deref().unwrap_or("") ) .into_bytes() } - /// Build and sign a record with the owner's libp2p (ed25519) keypair. - pub fn sign( + /// The marketplace terms, when this is a v2 record. + pub fn terms(&self) -> Option { + if self.v < 2 { + return None; + } + Some(HostingTerms { + preferred_hosts: self.preferred_hosts.clone(), + price_per_gib_month: self + .price_per_gib_month + .as_deref() + .and_then(|p| crate::payments::parse_usd(p).ok()) + .unwrap_or(0), + payer: self.payer.clone(), + }) + } + + /// Who pays for this site's preferred copies: the payer key, else the owner. + pub fn payer_key(&self) -> &str { + self.payer.as_deref().unwrap_or(&self.owner) + } + + /// True when `peer` is one of the record's preferred hosts. + pub fn prefers(&self, peer: &str) -> bool { + self.preferred_hosts.iter().any(|p| p == peer) + } + + /// Build and sign a record with marketplace terms (v2). `terms = None` + /// signs a plain v1 record, exactly as [`SiteRecord::sign`] does. + pub fn sign_with_terms( keypair: &c0mpute_net::libp2p::identity::Keypair, name: &str, manifest: &Hash, seq: u64, size: u64, domains: Vec, + terms: Option, ) -> Result { let owner = owner_hex(keypair).context("hosting needs an ed25519 identity")?; let mut rec = SiteRecord { @@ -188,8 +261,17 @@ impl SiteRecord { seq, size, domains, + preferred_hosts: Vec::new(), + price_per_gib_month: None, + payer: None, sig: String::new(), }; + if let Some(t) = terms { + rec.v = 2; + rec.preferred_hosts = t.preferred_hosts; + rec.price_per_gib_month = Some(crate::payments::format_usd(t.price_per_gib_month)); + rec.payer = t.payer; + } rec.check_fields()?; let sig = keypair .sign(&rec.signing_bytes()) @@ -198,9 +280,48 @@ impl SiteRecord { Ok(rec) } + /// Build and sign a record with the owner's libp2p (ed25519) keypair. + pub fn sign( + keypair: &c0mpute_net::libp2p::identity::Keypair, + name: &str, + manifest: &Hash, + seq: u64, + size: u64, + domains: Vec, + ) -> Result { + Self::sign_with_terms(keypair, name, manifest, seq, size, domains, None) + } + fn check_fields(&self) -> Result<()> { - if self.v != 1 { - bail!("unsupported site record version {}", self.v); + match self.v { + 1 => { + if !self.preferred_hosts.is_empty() || self.price_per_gib_month.is_some() || self.payer.is_some() { + bail!("a v1 site record carries no marketplace terms"); + } + } + 2 => { + if self.preferred_hosts.is_empty() || self.preferred_hosts.len() > MAX_PREFERRED_HOSTS { + bail!("a v2 site record names 1-{MAX_PREFERRED_HOSTS} preferred hosts"); + } + if !self.preferred_hosts.iter().all(|p| valid_peer_id(p)) { + bail!("preferred host is not a peer id"); + } + let mut seen = std::collections::HashSet::new(); + if !self.preferred_hosts.iter().all(|p| seen.insert(p)) { + bail!("preferred hosts repeat"); + } + match self.price_per_gib_month.as_deref().map(crate::payments::parse_usd) { + // Canonical form only, so every node signs/verifies the same bytes. + Some(Ok(m)) if self.price_per_gib_month.as_deref() == Some(crate::payments::format_usd(m).as_str()) => {} + _ => bail!("price_per_gib_month must be a canonical USD decimal"), + } + if let Some(p) = &self.payer + && !valid_key_hex(p) + { + bail!("payer must be a hex ed25519 key"); + } + } + v => bail!("unsupported site record version {v}"), } if !valid_name(&self.name) { bail!("site name must be 1-63 of a-z, 0-9 and '-'"); @@ -446,6 +567,110 @@ mod tests { assert!(!stolen.verify()); } + fn peer() -> String { + c0mpute_net::libp2p::identity::Keypair::generate_ed25519() + .public() + .to_peer_id() + .to_base58() + } + + #[test] + fn v1_records_are_unchanged_and_still_verify() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let rec = SiteRecord::sign(&kp, "old", &Hash::of(b"m"), 1, 10, vec![]).unwrap(); + assert_eq!(rec.v, 1); + let json = serde_json::to_string(&rec).unwrap(); + // A v1 record serializes with no new keys, so old nodes read it as before. + assert!(!json.contains("preferred_hosts") && !json.contains("price_per_gib_month") && !json.contains("payer")); + // And a v1 record as an old node wrote it (no new keys) parses and verifies. + let back: SiteRecord = serde_json::from_str(&json).unwrap(); + assert!(back.verify()); + assert!(back.terms().is_none()); + assert_eq!(back.payer_key(), back.owner); + } + + #[test] + fn v2_records_sign_terms_and_reject_tampering() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let (a, b) = (peer(), peer()); + let terms = HostingTerms { + preferred_hosts: vec![a.clone(), b.clone()], + price_per_gib_month: 4_000, + payer: None, + }; + let rec = SiteRecord::sign_with_terms(&kp, "paid", &Hash::of(b"m"), 1, 10, vec![], Some(terms.clone())).unwrap(); + assert_eq!(rec.v, 2); + assert!(rec.verify()); + assert_eq!(rec.price_per_gib_month.as_deref(), Some("0.004")); + assert_eq!(rec.terms().unwrap(), terms); + assert!(rec.prefers(&a) && !rec.prefers("nobody")); + let back: SiteRecord = serde_json::from_slice(&serde_json::to_vec(&rec).unwrap()).unwrap(); + assert!(back.verify()); + + let mut t = rec.clone(); + t.preferred_hosts.reverse(); + assert!(!t.verify(), "preferred hosts are signed"); + let mut t = rec.clone(); + t.price_per_gib_month = Some("0.005".into()); + assert!(!t.verify(), "price is signed"); + let mut t = rec.clone(); + t.payer = Some("b".repeat(64)); + assert!(!t.verify(), "payer is signed"); + // Downgrading to v1 (dropping the terms) breaks the signature too. + let mut t = rec.clone(); + t.v = 1; + t.preferred_hosts.clear(); + t.price_per_gib_month = None; + assert!(!t.verify(), "a v2 signature is not a v1 signature"); + } + + #[test] + fn v2_terms_are_validated() { + let kp = c0mpute_net::libp2p::identity::Keypair::generate_ed25519(); + let sign = |t: HostingTerms| SiteRecord::sign_with_terms(&kp, "x", &Hash::of(b"m"), 1, 10, vec![], Some(t)); + assert!(sign(HostingTerms::default()).is_err(), "no preferred hosts"); + assert!( + sign(HostingTerms { + preferred_hosts: vec!["not-a-peer".into()], + price_per_gib_month: 1, + payer: None + }) + .is_err() + ); + let p = peer(); + assert!( + sign(HostingTerms { + preferred_hosts: vec![p.clone(), p.clone()], + price_per_gib_month: 1, + payer: None + }) + .is_err(), + "duplicates" + ); + assert!( + sign(HostingTerms { + preferred_hosts: (0..6).map(|_| peer()).collect(), + price_per_gib_month: 1, + payer: None + }) + .is_err(), + "too many" + ); + assert!( + sign(HostingTerms { + preferred_hosts: vec![p.clone()], + price_per_gib_month: 1, + payer: Some("zz".into()) + }) + .is_err(), + "bad payer" + ); + // A v1 record smuggling terms is rejected. + let mut r = SiteRecord::sign(&kp, "x", &Hash::of(b"m"), 1, 10, vec![]).unwrap(); + r.preferred_hosts = vec![p]; + assert!(!r.verify()); + } + #[test] fn names_domains_paths() { assert!(valid_name("blog-2")); diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index 5ed89964..44bdcbbb 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/node/crates/c0mpute-core/src/lib.rs @@ -21,6 +21,7 @@ pub mod jobs_gateway; pub mod net_guard; pub mod ocr; pub mod opendisk; +pub mod payments; pub mod penalty_box; pub mod pieces; pub mod podcast; diff --git a/node/crates/c0mpute-core/src/opendisk.rs b/node/crates/c0mpute-core/src/opendisk.rs index 8bf5e79a..487b24db 100644 --- a/node/crates/c0mpute-core/src/opendisk.rs +++ b/node/crates/c0mpute-core/src/opendisk.rs @@ -118,6 +118,12 @@ pub fn descriptor( Some(n) => format!("{n} · c0mpute node {short}"), None => format!("c0mpute node {short}"), }; + let hosts = config.roles.contains(&c0mpute_proto::Role::Hosting); + let roles: Vec = config + .roles + .iter() + .filter_map(|r| serde_json::to_value(r).ok()) + .collect(); let mut doc = json!({ "opendisk": "0.1", "name": name, @@ -133,6 +139,10 @@ pub fn descriptor( "accepts": { "visibility": ["public", "private"], "encrypted_only": false, + // A c0mpute hosting node holds static sites for the network and + // can be picked as a paid (preferred) host by + // `c0mpute hosting publish --disks auto`. + "hosting": hosts, }, "updated": chrono_now(), // Unknown to OpenDisk, kept by readers: how to reach this disk on @@ -142,8 +152,16 @@ pub fn descriptor( "version": env!("CARGO_PKG_VERSION"), "storage_api": origin.map(|o| format!("{}/storage/v1", o.trim_end_matches('/'))), "redundancy": "reed-solomon 10/14", + "roles": roles, }, }); + if hosts { + doc["c0mpute"]["hosting"] = json!({ + "price_per_gib_month": config.hosting.price_micros() as f64 / crate::payments::MICROS_PER_USD as f64, + "max_gib": gib(config.hosting.max_bytes), + "publish": "c0mpute hosting publish --name --disks auto", + }); + } if let Some(k) = key_hex { doc["key"] = Value::String(format!("ed25519:{k}")); } @@ -287,6 +305,45 @@ mod tests { assert_eq!(d["key"], format!("ed25519:{key}")); assert_eq!(d["c0mpute"]["storage_api"], "https://disk.example/storage/v1"); assert!(d["updated"].as_str().unwrap().ends_with('Z')); + // Default roles include hosting: the disk says so, for d1sks to filter on. + assert_eq!(d["accepts"]["hosting"], true); + assert!(d["c0mpute"]["roles"].as_array().unwrap().iter().any(|r| r == "hosting")); + assert_eq!(d["c0mpute"]["hosting"]["price_per_gib_month"], 0.004); + assert_eq!(d["c0mpute"]["hosting"]["max_gib"], 5.0); + } + + #[test] + fn a_node_without_the_hosting_role_does_not_offer_hosting() { + let mut cfg = Config::default(); + cfg.roles.retain(|r| *r != c0mpute_proto::Role::Hosting); + let d = descriptor(&cfg, "12D3KooWSFxwveBjrGy4n1Djg3vpQD5aHPCXYCQHZWruffig7P4J", None, None, Capacity { + free_bytes: G, + total_bytes: G, + }); + assert_eq!(d["accepts"]["hosting"], false); + assert!(d["c0mpute"].get("hosting").is_none()); + } + + #[test] + fn the_descriptor_parses_as_a_marketplace_hosting_disk() { + // What d1sks stores is the descriptor under `data`; the publisher's + // ranking reads it back as a hosting node. + let cfg = Config::default(); + let peer = c0mpute_net::libp2p::identity::Keypair::generate_ed25519() + .public() + .to_peer_id() + .to_base58(); + let d = descriptor(&cfg, &peer, None, Some("https://disk.example"), Capacity { + free_bytes: 100 * G, + total_bytes: 200 * G, + }); + let listing = json!({ "disks": [{ "name": "n", "status": "active", "free_gib": 100.0, + "per_gib_month": 0.004, "countries": [], "data": d }] }); + let disks = crate::hosting::market::parse_disks(&listing); + assert_eq!(disks.len(), 1); + assert_eq!(disks[0].peer_id, peer); + assert_eq!(disks[0].price, 4_000); + assert_eq!(disks[0].free_bytes, 5 * G, "hosting offers its hosting budget, not the whole disk"); } #[test] diff --git a/node/crates/c0mpute-core/src/payments.rs b/node/crates/c0mpute-core/src/payments.rs new file mode 100644 index 00000000..834d398c --- /dev/null +++ b/node/crates/c0mpute-core/src/payments.rs @@ -0,0 +1,604 @@ +//! Payments: the one small interface every c0mpute feature that moves money +//! goes through (hosting settlement today; ads billing is meant to use the +//! same [`Rail`]). +//! +//! * Amounts are integer micro-USD ([`Micros`]), never floats. +//! * A [`Transfer`] carries an idempotency key; the [`Ledger`] records every +//! key that was paid, so a retried settlement round never pays twice. +//! * Rails: [`FakeRail`] (in memory, for tests) and [`CoinPayRail`] (CoinPay +//! `POST /api/payouts/create` with a business API key holding the +//! `payouts:create` scope). +//! +//! Nothing here moves money unless the operator turns it on: [`PaymentsConfig`] +//! defaults to `enabled = false`, and [`rail_from_config`] returns `None` +//! until `enabled = true` **and** the CoinPay key is in the environment. + +use std::collections::{HashMap, HashSet}; +use std::path::PathBuf; +use std::sync::Mutex; + +use anyhow::{Context, Result, bail}; +use serde::{Deserialize, Serialize}; + +/// Micro-USD (1 USD = 1_000_000). +pub type Micros = u64; + +pub const MICROS_PER_USD: u64 = 1_000_000; + +/// Parse a decimal USD amount ("0.004", "12", "1.5") into micro-USD. At most +/// six decimals; no sign, no exponent. +pub fn parse_usd(s: &str) -> Result { + let s = s.trim(); + if s.is_empty() || s.len() > 24 { + bail!("not a USD amount: {s:?}"); + } + let (int, frac) = match s.split_once('.') { + Some((i, f)) => (i, f), + None => (s, ""), + }; + if int.is_empty() && frac.is_empty() { + bail!("not a USD amount: {s:?}"); + } + if !int.chars().all(|c| c.is_ascii_digit()) || !frac.chars().all(|c| c.is_ascii_digit()) { + bail!("not a USD amount: {s:?}"); + } + if frac.len() > 6 { + bail!("USD amounts have at most 6 decimals: {s:?}"); + } + let whole: u64 = if int.is_empty() { 0 } else { int.parse().context("USD amount too large")? }; + let mut f = frac.to_string(); + while f.len() < 6 { + f.push('0'); + } + let frac_micros: u64 = f.parse().unwrap_or(0); + whole + .checked_mul(MICROS_PER_USD) + .and_then(|w| w.checked_add(frac_micros)) + .context("USD amount too large") +} + +/// Micro-USD as a canonical decimal string ("0.004", "12", "1.5"). +pub fn format_usd(m: Micros) -> String { + let whole = m / MICROS_PER_USD; + let frac = m % MICROS_PER_USD; + if frac == 0 { + return whole.to_string(); + } + let f = format!("{frac:06}"); + format!("{whole}.{}", f.trim_end_matches('0')) +} + +/// Where a payee wants to be paid. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Payee { + /// Wallet address on `coin`'s chain. + pub wallet: String, + /// CoinPay currency code, e.g. `USDC_POL`. + #[serde(default = "default_coin")] + pub coin: String, +} + +pub fn default_coin() -> String { + "USDC_POL".to_string() +} + +impl Payee { + /// Basic shape check; the rail validates the address for real. + pub fn valid(&self) -> bool { + let w = self.wallet.trim(); + (8..=128).contains(&w.len()) + && w.chars().all(|c| c.is_ascii_alphanumeric() || c == ':' || c == '_' || c == '-') + && (2..=16).contains(&self.coin.len()) + && self.coin.chars().all(|c| c.is_ascii_uppercase() || c.is_ascii_digit() || c == '_') + } +} + +/// One payment to make. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Transfer { + /// Same key, same payment: rails and the ledger refuse to pay a key twice. + pub key: String, + /// Who pays (hex ed25519 key of the payer identity; informational for + /// the rail, which pays from the account its credentials belong to). + pub payer: String, + pub payee: Payee, + pub amount: Micros, + /// What it is for (goes into the rail's metadata). + pub memo: String, +} + +/// What a rail answered for a payment. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +pub struct Paid { + pub key: String, + /// The rail's own id for the payment. + pub rail_ref: String, +} + +/// A way to move money. Keep it small: ads billing and hosting settlement +/// both need only these two calls. +#[async_trait::async_trait] +pub trait Rail: Send + Sync { + fn name(&self) -> &str; + /// The payer's spendable balance, when the rail can tell. `None` means + /// unknown (callers then rely on their own spending cap). + async fn balance(&self, payer: &str) -> Result>; + /// Pay one transfer. Must refuse a key it already paid. + async fn pay(&self, t: &Transfer) -> Result; +} + +// ── fake rail ─────────────────────────────────────────────────────────── + +/// In-memory rail for tests: prepaid balances per payer, a list of payments, +/// and the same idempotency rule a real rail has. +#[derive(Default)] +pub struct FakeRail { + inner: Mutex, +} + +#[derive(Default)] +struct FakeState { + balances: HashMap, + paid: Vec, + keys: HashSet, +} + +impl FakeRail { + pub fn new() -> Self { + Self::default() + } + + /// Prepay `amount` for `payer`. + pub fn fund(&self, payer: &str, amount: Micros) { + let mut s = self.inner.lock().unwrap(); + *s.balances.entry(payer.to_string()).or_default() += amount; + } + + pub fn payments(&self) -> Vec { + self.inner.lock().unwrap().paid.clone() + } + + /// Total received by a wallet. + pub fn received(&self, wallet: &str) -> Micros { + self.inner + .lock() + .unwrap() + .paid + .iter() + .filter(|t| t.payee.wallet == wallet) + .map(|t| t.amount) + .sum() + } +} + +#[async_trait::async_trait] +impl Rail for FakeRail { + fn name(&self) -> &str { + "fake" + } + + async fn balance(&self, payer: &str) -> Result> { + Ok(Some(*self.inner.lock().unwrap().balances.get(payer).unwrap_or(&0))) + } + + async fn pay(&self, t: &Transfer) -> Result { + let mut s = self.inner.lock().unwrap(); + if s.keys.contains(&t.key) { + bail!("transfer {} already paid", t.key); + } + let bal = s.balances.get(&t.payer).copied().unwrap_or(0); + if bal < t.amount { + bail!("insufficient balance: {} < {}", format_usd(bal), format_usd(t.amount)); + } + s.balances.insert(t.payer.clone(), bal - t.amount); + s.keys.insert(t.key.clone()); + s.paid.push(t.clone()); + Ok(Paid { + key: t.key.clone(), + rail_ref: format!("fake-{}", s.paid.len()), + }) + } +} + +// ── CoinPay rail ──────────────────────────────────────────────────────── + +/// CoinPay payouts from the payer's own CoinPay business wallet. +pub struct CoinPayRail { + base: String, + api_key: String, + recipient_email: String, + client: reqwest::Client, +} + +impl CoinPayRail { + pub fn new(base: &str, api_key: &str, recipient_email: &str) -> Result { + if api_key.trim().is_empty() { + bail!("CoinPay API key is empty"); + } + if !recipient_email.contains('@') { + bail!("payments.coinpay.recipient_email must be an email address"); + } + Ok(Self { + base: base.trim_end_matches('/').to_string(), + api_key: api_key.trim().to_string(), + recipient_email: recipient_email.to_string(), + client: reqwest::Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .user_agent(concat!("c0mpute/", env!("CARGO_PKG_VERSION"))) + .build()?, + }) + } + + /// The JSON body for `POST /api/payouts/create`. + pub fn payout_body(&self, t: &Transfer) -> serde_json::Value { + serde_json::json!({ + "recipient_email": self.recipient_email, + "recipient_wallet": t.payee.wallet, + "cryptocurrency": t.payee.coin, + "amount_usd": format_usd(t.amount), + "metadata": { + "source": "c0mpute", + "idempotency_key": t.key, + "payer": t.payer, + "memo": t.memo, + }, + }) + } +} + +#[async_trait::async_trait] +impl Rail for CoinPayRail { + fn name(&self) -> &str { + "coinpay" + } + + async fn balance(&self, _payer: &str) -> Result> { + // CoinPay has no balance call for a business API key; callers cap + // spending with `payments.max_usd_per_round`. + Ok(None) + } + + async fn pay(&self, t: &Transfer) -> Result { + let res = self + .client + .post(format!("{}/api/payouts/create", self.base)) + .bearer_auth(&self.api_key) + .json(&self.payout_body(t)) + .send() + .await + .context("CoinPay payout request")?; + let status = res.status(); + let body: serde_json::Value = res.json().await.unwrap_or(serde_json::Value::Null); + if !status.is_success() || body["success"] != serde_json::Value::Bool(true) { + bail!("CoinPay payout answered {status}: {}", body["error"]); + } + Ok(Paid { + key: t.key.clone(), + rail_ref: body["payout"]["id"].as_str().unwrap_or("").to_string(), + }) + } +} + +// ── config ────────────────────────────────────────────────────────────── + +/// `[payments]` in config.toml. Off by default: no money moves until an +/// operator sets `enabled = true` and puts the CoinPay key in the +/// environment variable `coinpay.api_key_env` names. +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct PaymentsConfig { + #[serde(default)] + pub enabled: bool, + /// Only `coinpay` is a real rail. + #[serde(default = "default_rail")] + pub rail: String, + /// Most this node pays out in one settlement round, USD (decimal string). + #[serde(default = "default_round_cap")] + pub max_usd_per_round: String, + #[serde(default)] + pub coinpay: CoinPayConfig, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct CoinPayConfig { + #[serde(default = "default_coinpay_url")] + pub url: String, + /// The environment variable holding a CoinPay business API key with the + /// `payouts:create` scope. The key itself never goes in config.toml. + #[serde(default = "default_key_env")] + pub api_key_env: String, + /// CoinPay requires a recipient email on every payout; this is the + /// address its payout records are filed under (yours). + #[serde(default)] + pub recipient_email: String, +} + +fn default_rail() -> String { + "coinpay".into() +} +fn default_round_cap() -> String { + "10".into() +} +fn default_coinpay_url() -> String { + "https://coinpayportal.com".into() +} +fn default_key_env() -> String { + "C0MPUTE_COINPAY_PAYOUT_KEY".into() +} + +impl Default for CoinPayConfig { + fn default() -> Self { + Self { + url: default_coinpay_url(), + api_key_env: default_key_env(), + recipient_email: String::new(), + } + } +} + +impl Default for PaymentsConfig { + fn default() -> Self { + Self { + enabled: false, + rail: default_rail(), + max_usd_per_round: default_round_cap(), + coinpay: CoinPayConfig::default(), + } + } +} + +impl PaymentsConfig { + pub fn round_cap(&self) -> Micros { + parse_usd(&self.max_usd_per_round).unwrap_or(10 * MICROS_PER_USD) + } +} + +/// The rail this node pays with, or `None` when payments are off. Errors only +/// when payments are on but misconfigured, so an operator sees why. +pub fn rail_from_config(cfg: &PaymentsConfig) -> Result>> { + if !cfg.enabled { + return Ok(None); + } + match cfg.rail.as_str() { + "coinpay" => { + let key = std::env::var(&cfg.coinpay.api_key_env).unwrap_or_default(); + if key.trim().is_empty() { + bail!( + "payments.enabled = true but ${} is not set (a CoinPay business API key with payouts:create)", + cfg.coinpay.api_key_env + ); + } + let rail = CoinPayRail::new(&cfg.coinpay.url, &key, &cfg.coinpay.recipient_email)?; + Ok(Some(std::sync::Arc::new(rail))) + } + other => bail!("unknown payments.rail {other:?} (only \"coinpay\")"), + } +} + +// ── ledger ────────────────────────────────────────────────────────────── + +/// Every transfer key this node has paid, with the rail's reference. Written +/// to disk after each payment so a crash between rounds never pays twice. +#[derive(Default)] +pub struct Ledger { + path: Option, + paid: Mutex>, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct LedgerEntry { + pub amount: Micros, + pub payee: String, + pub rail_ref: String, + pub at: u64, +} + +impl Ledger { + pub fn in_memory() -> Self { + Self::default() + } + + pub fn open(path: PathBuf) -> Self { + let paid = std::fs::read(&path) + .ok() + .and_then(|b| serde_json::from_slice(&b).ok()) + .unwrap_or_default(); + Self { + path: Some(path), + paid: Mutex::new(paid), + } + } + + pub fn contains(&self, key: &str) -> bool { + self.paid.lock().unwrap().contains_key(key) + } + + pub fn entries(&self) -> HashMap { + self.paid.lock().unwrap().clone() + } + + fn record(&self, t: &Transfer, paid: &Paid) { + let snapshot = { + let mut m = self.paid.lock().unwrap(); + m.insert( + t.key.clone(), + LedgerEntry { + amount: t.amount, + payee: t.payee.wallet.clone(), + rail_ref: paid.rail_ref.clone(), + at: now_secs(), + }, + ); + serde_json::to_vec_pretty(&*m).unwrap_or_default() + }; + if let Some(p) = &self.path { + if let Some(d) = p.parent() { + let _ = std::fs::create_dir_all(d); + } + let tmp = p.with_extension("tmp"); + if std::fs::write(&tmp, &snapshot).is_ok() { + let _ = std::fs::rename(&tmp, p); + } + } + } +} + +/// What happened to one transfer in [`pay_all`]. +#[derive(Clone, Debug, PartialEq, Eq)] +pub enum Outcome { + Paid(Paid), + AlreadyPaid, + /// Skipped: over the round cap or the payer's balance. + OverBudget, + Failed(String), +} + +/// Pay `transfers` in order through `rail`, skipping keys the ledger already +/// has, and stopping at `cap` (and at the payer's balance when the rail +/// knows it). +pub async fn pay_all(rail: &dyn Rail, ledger: &Ledger, transfers: &[Transfer], cap: Micros) -> Vec<(Transfer, Outcome)> { + let mut out = Vec::new(); + let mut spent: Micros = 0; + let mut balances: HashMap> = HashMap::new(); + for t in transfers { + if ledger.contains(&t.key) { + out.push((t.clone(), Outcome::AlreadyPaid)); + continue; + } + if t.amount == 0 { + continue; + } + if spent.saturating_add(t.amount) > cap { + out.push((t.clone(), Outcome::OverBudget)); + continue; + } + if !balances.contains_key(&t.payer) { + let b = rail.balance(&t.payer).await.unwrap_or(None); + balances.insert(t.payer.clone(), b); + } + if let Some(Some(b)) = balances.get(&t.payer) + && *b < t.amount + { + out.push((t.clone(), Outcome::OverBudget)); + continue; + } + match rail.pay(t).await { + Ok(p) => { + ledger.record(t, &p); + spent += t.amount; + if let Some(Some(b)) = balances.get_mut(&t.payer) { + *b -= t.amount; + } + out.push((t.clone(), Outcome::Paid(p))); + } + Err(e) => out.push((t.clone(), Outcome::Failed(e.to_string()))), + } + } + out +} + +pub fn now_secs() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn payee(w: &str) -> Payee { + Payee { + wallet: w.into(), + coin: default_coin(), + } + } + + fn t(key: &str, payer: &str, w: &str, amount: Micros) -> Transfer { + Transfer { + key: key.into(), + payer: payer.into(), + payee: payee(w), + amount, + memo: "test".into(), + } + } + + #[test] + fn usd_round_trips() { + assert_eq!(parse_usd("0.004").unwrap(), 4_000); + assert_eq!(parse_usd("12").unwrap(), 12_000_000); + assert_eq!(parse_usd("1.5").unwrap(), 1_500_000); + assert_eq!(parse_usd(".25").unwrap(), 250_000); + assert!(parse_usd("-1").is_err()); + assert!(parse_usd("1e3").is_err()); + assert!(parse_usd("0.0000001").is_err()); + assert!(parse_usd("").is_err()); + assert!(parse_usd(".").is_err()); + assert_eq!(format_usd(4_000), "0.004"); + assert_eq!(format_usd(12_000_000), "12"); + assert_eq!(format_usd(1_500_000), "1.5"); + for m in [0, 1, 999_999, 1_000_001, 123_456_789] { + assert_eq!(parse_usd(&format_usd(m)).unwrap(), m); + } + } + + #[test] + fn payments_are_off_by_default() { + let cfg = PaymentsConfig::default(); + assert!(!cfg.enabled); + assert!(rail_from_config(&cfg).unwrap().is_none()); + let on = PaymentsConfig { + enabled: true, + coinpay: CoinPayConfig { + api_key_env: "C0MPUTE_TEST_KEY_THAT_IS_NEVER_SET".into(), + ..Default::default() + }, + ..Default::default() + }; + assert!(rail_from_config(&on).is_err(), "on without a key is an error, not a silent rail"); + } + + #[test] + fn coinpay_body_carries_the_idempotency_key() { + let rail = CoinPayRail::new("https://coinpay.invalid/", "k", "ops@example.org").unwrap(); + let b = rail.payout_body(&t("k1", "aa", "0xabcdef0123456789", 1_234_567)); + assert_eq!(b["amount_usd"], "1.234567"); + assert_eq!(b["recipient_wallet"], "0xabcdef0123456789"); + assert_eq!(b["cryptocurrency"], "USDC_POL"); + assert_eq!(b["metadata"]["idempotency_key"], "k1"); + } + + #[tokio::test] + async fn fake_rail_pays_once_within_balance_and_cap() { + let rail = FakeRail::new(); + rail.fund("payer", 1_000); + let ledger = Ledger::in_memory(); + let ts = vec![ + t("a", "payer", "wallet-aaaaaaaa", 400), + t("b", "payer", "wallet-bbbbbbbb", 400), + t("c", "payer", "wallet-cccccccc", 400), + ]; + let out = pay_all(&rail, &ledger, &ts, 10_000).await; + assert!(matches!(out[0].1, Outcome::Paid(_))); + assert!(matches!(out[1].1, Outcome::Paid(_))); + assert_eq!(out[2].1, Outcome::OverBudget, "balance runs out"); + // Same round again: nothing pays twice. + rail.fund("payer", 1_000); + let again = pay_all(&rail, &ledger, &ts, 10_000).await; + assert_eq!(again[0].1, Outcome::AlreadyPaid); + assert_eq!(again[1].1, Outcome::AlreadyPaid); + assert!(matches!(again[2].1, Outcome::Paid(_))); + assert_eq!(rail.received("wallet-aaaaaaaa"), 400); + // The cap stops a round. + let capped = pay_all(&rail, &ledger, &[t("d", "payer", "wallet-dddddddd", 500)], 499).await; + assert_eq!(capped[0].1, Outcome::OverBudget); + } + + #[test] + fn payee_shape() { + assert!(payee("0xabcdef0123456789").valid()); + assert!(!payee("short").valid()); + assert!(!payee("0xabc def0123456789").valid()); + } +} diff --git a/node/crates/c0mpute-core/src/supervisor.rs b/node/crates/c0mpute-core/src/supervisor.rs index 81e4bacf..6d385b7d 100644 --- a/node/crates/c0mpute-core/src/supervisor.rs +++ b/node/crates/c0mpute-core/src/supervisor.rs @@ -555,18 +555,95 @@ impl Supervisor { /// Start the hosting service: gossip, replica map, self-healing copies. pub async fn start_hosting(&self) -> Result { let hosts = self.config.roles.contains(&c0mpute_proto::Role::Hosting); - let catalog_path = Some(self.config.storage.root.join("hosting").join("catalog.json")); - crate::hosting::Hosting::start( + let dir = self.config.storage.root.join("hosting"); + let catalog_path = Some(dir.join("catalog.json")); + let cfg = &self.config.hosting; + // Receipts are signed with this machine's publisher key: the key that + // owns (and pays for) the sites published from here. + let challenge = crate::hosting::ChallengeScope::parse(&cfg.challenge); + let challenger = if challenge == crate::hosting::ChallengeScope::Off || self.config.private.enabled { + None + } else { + crate::hosting::publisher_keypair().ok() + }; + let hosting = crate::hosting::Hosting::start( self.libp2p.clone(), self.store.clone(), crate::hosting::Options { hosts, - max_bytes: self.config.hosting.max_bytes, + max_bytes: cfg.max_bytes, catalog_path, + challenger, + challenge, + challenge_every: std::time::Duration::from_secs(cfg.challenge_every_secs.max(30)), + price: Some(cfg.price_micros()), + payout: cfg.payee(), + receipts_path: Some(dir.join("receipts.jsonl")), ..Default::default() }, ) - .await + .await?; + self.spawn_settlement(hosting.clone(), dir); + Ok(hosting) + } + + /// Settle with the preferred holders of the sites this machine pays for, + /// once per `hosting.settle_every_secs`. With `[payments]` off (the + /// default) it only logs what it would pay; nothing moves. + fn spawn_settlement(&self, hosting: crate::hosting::Hosting, dir: std::path::PathBuf) { + if hosting.challenger_key().is_none() { + return; + } + let every = self.config.hosting.settle_every_secs.max(60); + let payments = self.config.payments.clone(); + let rail = match crate::payments::rail_from_config(&payments) { + Ok(r) => r, + Err(e) => { + tracing::warn!(err = %e, "payments: misconfigured, hosting settlement stays dry-run"); + None + } + }; + let ledger = std::sync::Arc::new(crate::payments::Ledger::open(dir.join("ledger.json"))); + let cursor_path = dir.join("settled-until"); + tokio::spawn(async move { + loop { + tokio::time::sleep(std::time::Duration::from_secs(every)).await; + let now = crate::payments::now_secs(); + let from = std::fs::read_to_string(&cursor_path) + .ok() + .and_then(|s| s.trim().parse::().ok()) + .filter(|f| *f < now) + .unwrap_or(now.saturating_sub(every)); + match &rail { + Some(rail) => { + let (dues, outcomes) = hosting + .settle_round(rail.as_ref(), &ledger, from, now, payments.round_cap()) + .await; + for (t, o) in &outcomes { + tracing::info!(key = %t.key, amount = %crate::payments::format_usd(t.amount), wallet = %t.payee.wallet, outcome = ?o, "hosting settlement"); + } + tracing::debug!(dues = dues.len(), "hosting settlement round"); + // Always move on: windows never overlap, so a key is + // never paid twice. A failed transfer is logged above + // and not retried (an underpayment, never a double one). + let _ = std::fs::write(&cursor_path, now.to_string()); + } + None => { + for d in hosting.dues(from, now).await.into_iter().filter(|d| d.amount > 0) { + tracing::info!( + site = %d.site_id, + holder = %d.holder, + amount = %crate::payments::format_usd(d.amount), + passed = d.passed, + challenged = d.challenged, + "hosting settlement (dry run: [payments] is off, nothing paid)" + ); + } + let _ = std::fs::write(&cursor_path, now.to_string()); + } + } + } + }); } /// Start the ads service on a gateway: campaign index, VAST/VMAP, and diff --git a/node/crates/c0mpute-core/tests/hosting_market_e2e.rs b/node/crates/c0mpute-core/tests/hosting_market_e2e.rs new file mode 100644 index 00000000..b0959012 --- /dev/null +++ b/node/crates/c0mpute-core/tests/hosting_market_e2e.rs @@ -0,0 +1,252 @@ +//! End-to-end marketplace hosting on loopback libp2p: a publisher names two +//! of five hosting nodes as preferred (paid) hosts in a v2 record; both take +//! copies and the site settles at three copies with no coordinator; the +//! publisher challenges every holder (random chunk, hash-verified, signed +//! receipts on gossip); a settlement round pays exactly the two preferred +//! holders on a fake rail, and a second round pays nothing twice. + +use std::sync::Arc; +use std::time::Duration; + +use c0mpute_core::hosting::site::{HostingTerms, SiteRecord, build_from_dir, owner_hex}; +use c0mpute_core::hosting::{ChallengeScope, Hosting, Options, Timing}; +use c0mpute_core::payments::{FakeRail, Ledger, Outcome, Payee}; +use c0mpute_net::{ChunkSource, Libp2pNetwork, NetworkConfig}; +use c0mpute_proto::Hash; +use c0mpute_store::ChunkStore; +use libp2p::multiaddr::Protocol; +use tempfile::tempdir; + +struct StoreSource(ChunkStore); + +#[async_trait::async_trait] +impl ChunkSource for StoreSource { + async fn read_chunk(&self, hash: &Hash) -> anyhow::Result> { + self.0.get(hash).await + } +} + +async fn node(dir: &std::path::Path, bootstrap: Vec) -> (Arc, ChunkStore) { + let store = ChunkStore::open(dir.join("store")).await.unwrap(); + let net = Libp2pNetwork::spawn( + NetworkConfig::for_dir(dir.join("id")) + .with_listen(vec!["/ip4/127.0.0.1/tcp/0".parse().unwrap()]) + .with_local_source(Arc::new(StoreSource(store.clone()))) + .with_bootstrap(bootstrap), + ) + .await + .unwrap(); + (Arc::new(net), store) +} + +async fn listen_addr(net: &Libp2pNetwork) -> libp2p::Multiaddr { + for _ in 0..50 { + if let Ok(addrs) = net.listen_addrs().await + && let Some(a) = addrs.into_iter().next() + { + return a.with(Protocol::P2p(net.peer_id())); + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + panic!("no listen addr"); +} + +fn fast() -> Timing { + Timing { + announce_every: Duration::from_millis(700), + records_every: Duration::from_millis(1500), + repair_every: Duration::from_millis(500), + holder_ttl: Duration::from_secs(4), + fetch_timeout: Duration::from_secs(20), + } +} + +async fn wait_for(what: &str, secs: u64, mut f: F) +where + F: FnMut() -> Fut, + Fut: std::future::Future, +{ + let deadline = tokio::time::Instant::now() + Duration::from_secs(secs); + while tokio::time::Instant::now() < deadline { + if f().await { + return; + } + tokio::time::sleep(Duration::from_millis(250)).await; + } + panic!("timed out waiting for {what}"); +} + +fn wallet(i: usize) -> String { + format!("0x{:040x}", 0xa11ce + i) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +async fn preferred_hosts_hold_pass_challenges_and_get_paid() { + let _ = tracing_subscriber::fmt() + .with_env_filter("c0mpute_core=info,c0mpute_net=warn") + .with_test_writer() + .try_init(); + let tmp = tempdir().unwrap(); + + // Five hosting nodes, each asking $0.004/GiB-month with a payout wallet. + let host_opts = |dir: &std::path::Path, i: usize| Options { + hosts: true, + catalog_path: Some(dir.join("catalog.json")), + timing: fast(), + price: Some(4_000), + payout: Some(Payee { + wallet: wallet(i), + coin: "USDC_POL".into(), + }), + ..Default::default() + }; + let (hub_net, hub_store) = node(&tmp.path().join("h0"), vec![]).await; + let hub_addr = listen_addr(&hub_net).await; + let mut hosts: Vec = vec![Hosting::start(hub_net, hub_store, host_opts(&tmp.path().join("h0"), 0)).await.unwrap()]; + for i in 1..5 { + let d = tmp.path().join(format!("h{i}")); + let (n, s) = node(&d, vec![hub_addr.clone()]).await; + hosts.push(Hosting::start(n, s, host_opts(&d, i)).await.unwrap()); + } + + // The publisher: owns and pays for the site, challenges its holders. + let owner = libp2p::identity::Keypair::generate_ed25519(); + let owner_key = owner_hex(&owner).unwrap(); + let pd = tmp.path().join("pub"); + let (pn, ps) = node(&pd, vec![hub_addr.clone()]).await; + let publisher = Hosting::start( + pn, + ps, + Options { + hosts: false, + timing: fast(), + challenger: Some(owner.clone()), + challenge: ChallengeScope::Own, + // Rounds are driven by the test. + challenge_every: Duration::from_secs(3600), + receipts_path: Some(pd.join("receipts.jsonl")), + ..Default::default() + }, + ) + .await + .unwrap(); + + // A gateway that challenges everything with its own key: its receipts + // travel on gossip but do not count for the owner's settlement. + let gd = tmp.path().join("gw"); + let (gn, gs) = node(&gd, vec![hub_addr.clone()]).await; + let gateway = Hosting::start( + gn, + gs, + Options { + hosts: false, + timing: fast(), + challenger: Some(libp2p::identity::Keypair::generate_ed25519()), + challenge: ChallengeScope::All, + challenge_every: Duration::from_secs(3600), + ..Default::default() + }, + ) + .await + .unwrap(); + + tokio::time::sleep(Duration::from_secs(3)).await; + + // Two of the five are preferred, at $1/GiB-month. + let preferred: Vec = vec![hosts[1].peer_id().to_string(), hosts[3].peer_id().to_string()]; + let site_dir = tmp.path().join("site"); + std::fs::create_dir_all(&site_dir).unwrap(); + std::fs::write(site_dir.join("index.html"), b"

paid hosting

").unwrap(); + let big: Vec = (0..(5 * 1024 * 1024)).map(|i| (i % 253) as u8).collect(); + std::fs::write(site_dir.join("big.bin"), &big).unwrap(); + let built = build_from_dir(&site_dir, "market").unwrap(); + let rec = SiteRecord::sign_with_terms( + &owner, + "market", + &Hash::of(&built.manifest.to_bytes()), + 1, + built.manifest.total_bytes(), + vec![], + Some(HostingTerms { + preferred_hosts: preferred.clone(), + price_per_gib_month: 1_000_000, + payer: None, + }), + ) + .unwrap(); + assert_eq!(rec.v, 2); + let site = rec.site_id.clone(); + publisher.publish_site(&built, rec.clone()).await.unwrap(); + + // Both preferred hosts hold copies and the site settles at exactly three. + let (p, s, r, pref) = (publisher.clone(), site.clone(), rec.clone(), preferred.clone()); + wait_for("preferred copies and three in total", 90, || { + let (p, s, r, pref) = (p.clone(), s.clone(), r.clone(), pref.clone()); + async move { + p.regossip(&r).await; + let hs = p.holders(&s).await; + hs.len() == 3 && pref.iter().all(|x| hs.contains(x)) + } + }) + .await; + tokio::time::sleep(Duration::from_secs(3)).await; + let holders = hosts[0].holders(&site).await; + assert_eq!(holders.len(), 3, "stays at three: {holders:?}"); + for x in &preferred { + assert!(holders.contains(x), "preferred {x} holds a copy"); + } + assert_eq!(publisher.preferred(&site).await, preferred); + + // A challenge round: every holder passes. + let receipts = publisher.challenge_round().await; + assert_eq!(receipts.len(), 3, "one challenge per holder: {receipts:?}"); + assert!(receipts.iter().all(|r| r.ok && r.verify() && r.challenger == owner_key)); + let pd2 = pd.clone(); + assert_eq!( + std::fs::read_to_string(pd2.join("receipts.jsonl")).unwrap().lines().count(), + 3, + "receipts are kept on disk for settlement" + ); + // The receipts reach other nodes over gossip. + let h0 = hosts[0].clone(); + let (s, x) = (site.clone(), preferred[0].clone()); + wait_for("receipts on gossip", 20, || { + let (h0, s, x) = (h0.clone(), s.clone(), x.clone()); + async move { !h0.receipts_for(&s, &x).await.is_empty() } + }) + .await; + + // The gateway's receipts arrive too, but are not the payer's. + let g_receipts = gateway.challenge_round().await; + assert!(!g_receipts.is_empty()); + + // Settlement on a fake rail, prepaid by the owner. + let rail = FakeRail::new(); + rail.fund(&owner_key, 10_000_000); + let ledger = Ledger::in_memory(); + let now = c0mpute_core::payments::now_secs(); + let month = c0mpute_core::hosting::settle::MONTH_SECS; + let (from, to) = (now + 1 - month, now + 1); + let (dues, outcomes) = publisher.settle_round(&rail, &ledger, from, to, 5_000_000).await; + assert_eq!(dues.len(), 2, "dues for the two preferred hosts only: {dues:?}"); + let expected = c0mpute_core::hosting::settle::owed(1_000_000, rec.size, month, 1, 1); + assert!(expected > 4_000, "5 MiB at $1/GiB-month for a month is ~$0.0049: {expected}"); + for d in &dues { + assert_eq!((d.passed, d.challenged), (1, 1), "gateway receipts do not count: {d:?}"); + assert_eq!(d.amount, expected); + } + assert_eq!(outcomes.len(), 2); + assert!(outcomes.iter().all(|(_, o)| matches!(o, Outcome::Paid(_))), "{outcomes:?}"); + let idx = |peer: &str| hosts.iter().position(|h| h.peer_id() == peer).unwrap(); + for x in &preferred { + assert_eq!(rail.received(&wallet(idx(x))), expected, "preferred {x} paid"); + } + let free = holders.iter().find(|h| !preferred.contains(h)).unwrap(); + assert_eq!(rail.received(&wallet(idx(free))), 0, "the free copy is not paid"); + assert_eq!(rail.payments().len(), 2); + + // Same window again: nothing pays twice. + let (_, again) = publisher.settle_round(&rail, &ledger, from, to, 5_000_000).await; + assert!(again.iter().all(|(_, o)| *o == Outcome::AlreadyPaid), "{again:?}"); + assert_eq!(rail.payments().len(), 2); +} diff --git a/plugins/hosting/module.toml b/plugins/hosting/module.toml index d1b2849b..4e5063f9 100644 --- a/plugins/hosting/module.toml +++ b/plugins/hosting/module.toml @@ -18,8 +18,14 @@ # fetched over libp2p from the fastest live holder, round-robin among # equally fast ones, failing over to the next. # -# Not yet: paid reservations, epoch settlement and proof-of-serve challenges -# from DIP-0015. Hosting is free while it is in beta. +# - Paid hosting: `publish --disks auto` picks preferred hosts from the d1sks +# OpenDisk marketplace (discovery only) into a signed v2 record; they +# replicate first, holders are challenged with random chunk fetches +# (signed receipts on gossip), and the payer settles per GiB-month through +# CoinPay payouts, OFF by default ([payments] enabled = false). +# +# Not yet: escrow, and proof-of-serve challenges from DIP-0015. Free hosting +# stays free. [module] id = "hosting" @@ -29,14 +35,14 @@ kind = "service" category = "data" status = "available" builtin = true -description = "Static sites kept alive by the network: three self-healing copies on hosting nodes, served by every gateway at /s// and on DNS-verified custom domains. Signed, content-addressed, no coordinator. Free in beta." +description = "Static sites kept alive by the network: three self-healing copies on hosting nodes, served by every gateway at /s// and on DNS-verified custom domains. Signed, content-addressed, no coordinator. Free by default; optionally pay hosts picked from the d1sks marketplace per GiB-month, with proof-of-storage challenges (settlement off until enabled)." author = "Profullstack" license = "MIT" source = "https://github.com/profullstack/c0mpute/tree/master/node/crates/c0mpute-core/src/hosting" -keywords = ["hosting", "static", "p2p", "censorship-resistant", "self-healing", "dweb"] +keywords = ["hosting", "static", "p2p", "censorship-resistant", "self-healing", "dweb", "marketplace", "opendisk"] [module.requirements] -c0mpute = ">=0.2.72" +c0mpute = ">=NEXT" os = ["linux", "darwin"] arch = ["x86_64", "aarch64"] capabilities = ["disk", "network:ingress?"]