From f974c791b86d110e62de43af9fc6f29db63ea8a6 Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Tue, 27 Oct 2026 05:13:59 +0000 Subject: [PATCH] Operator listings: dynamic pricing with an operator-set minimum; v0.2.35 Nodes now price their own bids instead of bidding the buyer's cap: - [pricing] in config.toml: mode dynamic (default) or fixed, min_usd floor. Dynamic bids 60% of the cap when idle and rises to the cap as the runner queue fills; never under the floor; skips jobs capped below it. - `c0mpute pricing [show|set --min X --fixed Y --dynamic]`. - The listing rides in the capability ad's hardware blob; the status aggregator re-validates it into /nodes `pricing`. - c0mpute.com/market shows price/job per GPU node and documents setting it; /providers and llms.txt updated. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 32 ++-- Cargo.toml | 2 +- apps/tui/package.json | 2 +- apps/web/package.json | 2 +- apps/web/public/llms.txt | 1 + apps/web/src/app/market/page.tsx | 23 ++- apps/web/src/app/providers/page.tsx | 6 +- apps/web/src/lib/market.ts | 7 + apps/web/src/lib/nodes.ts | 19 ++ node/crates/c0mpute-cli/src/main.rs | 72 +++++++- node/crates/c0mpute-core/src/capabilities.rs | 11 +- node/crates/c0mpute-core/src/config.rs | 4 + node/crates/c0mpute-core/src/dispatch.rs | 54 +++++- node/crates/c0mpute-core/src/lib.rs | 6 +- node/crates/c0mpute-core/src/pricing.rs | 172 ++++++++++++++++++ .../c0mpute-core/src/status_aggregator.rs | 87 ++++++++- node/crates/c0mpute-core/src/supervisor.rs | 21 +-- package.json | 2 +- packages/shared/package.json | 2 +- 19 files changed, 465 insertions(+), 60 deletions(-) create mode 100644 node/crates/c0mpute-core/src/pricing.rs diff --git a/Cargo.lock b/Cargo.lock index d173d48..647549f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -478,7 +478,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "c0mpute-api" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "c0mpute-proto", @@ -493,7 +493,7 @@ dependencies = [ [[package]] name = "c0mpute-bench" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "blake3", @@ -504,7 +504,7 @@ dependencies = [ [[package]] name = "c0mpute-cli" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "axum", @@ -534,7 +534,7 @@ dependencies = [ [[package]] name = "c0mpute-core" -version = "0.2.34" +version = "0.2.35" dependencies = [ "aes-gcm", "anyhow", @@ -573,7 +573,7 @@ dependencies = [ [[package]] name = "c0mpute-doctor" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "serde", @@ -585,7 +585,7 @@ dependencies = [ [[package]] name = "c0mpute-envelope" -version = "0.2.34" +version = "0.2.35" dependencies = [ "blake3", "bs58", @@ -598,7 +598,7 @@ dependencies = [ [[package]] name = "c0mpute-gateway" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "axum", @@ -620,7 +620,7 @@ dependencies = [ [[package]] name = "c0mpute-market" -version = "0.2.34" +version = "0.2.35" dependencies = [ "c0mpute-envelope", "serde", @@ -630,7 +630,7 @@ dependencies = [ [[package]] name = "c0mpute-net" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "async-trait", @@ -647,7 +647,7 @@ dependencies = [ [[package]] name = "c0mpute-placement" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "async-trait", @@ -668,7 +668,7 @@ dependencies = [ [[package]] name = "c0mpute-proto" -version = "0.2.34" +version = "0.2.35" dependencies = [ "blake3", "hex", @@ -680,7 +680,7 @@ dependencies = [ [[package]] name = "c0mpute-secure-chat" -version = "0.2.34" +version = "0.2.35" dependencies = [ "aes-gcm", "anyhow", @@ -703,7 +703,7 @@ dependencies = [ [[package]] name = "c0mpute-store" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "blake3", @@ -721,7 +721,7 @@ dependencies = [ [[package]] name = "c0mpute-transcode" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "c0mpute-proto", @@ -733,7 +733,7 @@ dependencies = [ [[package]] name = "c0mpute-update" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "hex", @@ -747,7 +747,7 @@ dependencies = [ [[package]] name = "c0mpute-verify" -version = "0.2.34" +version = "0.2.35" dependencies = [ "anyhow", "blake3", diff --git a/Cargo.toml b/Cargo.toml index ce898fa..f06c127 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ members = [ ] [workspace.package] -version = "0.2.34" +version = "0.2.35" edition = "2024" license = "MIT" repository = "https://github.com/profullstack/c0mpute" diff --git a/apps/tui/package.json b/apps/tui/package.json index 700ce5f..0112244 100644 --- a/apps/tui/package.json +++ b/apps/tui/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/tui", - "version": "0.2.34", + "version": "0.2.35", "private": true, "type": "module", "bin": { diff --git a/apps/web/package.json b/apps/web/package.json index 784833c..e3ab893 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute", - "version": "0.2.34", + "version": "0.2.35", "private": true, "type": "module", "scripts": { diff --git a/apps/web/public/llms.txt b/apps/web/public/llms.txt index 6c0702d..88fd5d0 100644 --- a/apps/web/public/llms.txt +++ b/apps/web/public/llms.txt @@ -9,6 +9,7 @@ ## Run a node - [Run a node](https://c0mpute.com/providers): `curl -fsSL https://c0mpute.com/install.sh | sh` installs c0mpute and starts the worker (Linux, macOS, Windows via WSL2; CPU or GPU). The goal is 100,000 nodes, enough pooled capacity to train and serve open-weight models in the open. +- [Market](https://c0mpute.com/market): GPU nodes online with each operator's listing. `c0mpute pricing set --min ` = dynamic pricing (idle bids 60% of the buyer's cap, rising to the cap as the queue fills) never under the floor; `--fixed ` = one price. The listing (`pricing: {mode, min_usd, fixed_usd}`) is in https://c0mpute.com/api/nodes. - [Node board](https://c0mpute.com/nodes): every node online now, ranked by `c0mpute bench` score. JSON at https://c0mpute.com/api/nodes. - Referral links: https://c0mpute.com/r/ gives an install line with `C0MPUTE_REF=`; nodes installed with it show the handle on the board. Public scoreboard, no payout. diff --git a/apps/web/src/app/market/page.tsx b/apps/web/src/app/market/page.tsx index 1ba629e..caa509a 100644 --- a/apps/web/src/app/market/page.tsx +++ b/apps/web/src/app/market/page.tsx @@ -1,7 +1,7 @@ import Link from "next/link"; import { CodeBlock } from "@/components/CodeBlock"; -import { getNodesPayload } from "@/lib/nodes"; -import { MARKET_FEE_PCT, OPERATOR_SHARE_PCT, splitPrice } from "@/lib/market"; +import { getNodesPayload, priceLabel } from "@/lib/nodes"; +import { DYNAMIC_IDLE_SHARE, MARKET_FEE_PCT, OPERATOR_SHARE_PCT, splitPrice } from "@/lib/market"; export const metadata = { title: "market — c0mpute", @@ -91,6 +91,7 @@ export default async function MarketPage() { node class bench + price / job version seen @@ -101,6 +102,7 @@ export default async function MarketPage() { {n.id} {n.class} {n.bench_score?.toLocaleString() ?? "-"} + {priceLabel(n.pricing)} {n.version ?? "-"} {ago(n.last_seen_s)} @@ -141,6 +143,23 @@ $ c0mpute transcode submit input.mov --preset hls --max-price 1.25`}

{`$ curl -fsSL https://c0mpute.com/install.sh | sh $ infernet payout set USDC 0xYourAddress --network polygon`} +

+ Set your price.{" "} + Every node is listed with dynamic pricing by default: when it is idle + it bids {Math.round(DYNAMIC_IDLE_SHARE * 100)}% of the buyer's + cap to win work, and the bid rises toward the full cap as its job + queue fills. You set a minimum and it never bids under it; jobs capped + below your minimum are skipped. Or ask one fixed price. +

+ {`$ c0mpute pricing set --min 0.05 # dynamic, never under $0.05 a job +$ c0mpute pricing set --fixed 0.40 # always $0.40 a job +$ c0mpute pricing # show your listing +$ c0mpute worker stop && c0mpute worker start -d # apply`} +

+ Your listing (mode and minimum) rides in the node's capability + ad and shows in the table above and in /api/nodes. + The {MARKET_FEE_PCT}% market fee comes out of the price you win the job at. +

→ running a node

diff --git a/apps/web/src/app/providers/page.tsx b/apps/web/src/app/providers/page.tsx index eda1d63..c691369 100644 --- a/apps/web/src/app/providers/page.tsx +++ b/apps/web/src/app/providers/page.tsx @@ -112,8 +112,10 @@ $ c0mpute worker status # the installer already started it`}

[ what you earn ]

- You set your rates. A buyer sets a cap. When you quote under it and - the work validates, settlement releases and both sides sign a + You set your rates: dynamic by default (undercut when idle, up to the + buyer's cap when busy) or one fixed price, never under the + minimum you set with c0mpute pricing set --min 0.05. A + buyer sets a cap. When you quote under it and the work validates, settlement releases and both sides sign a receipt. A direct protocol deal has{" "} no protocol fee. Jobs bought through the market, including paid diff --git a/apps/web/src/lib/market.ts b/apps/web/src/lib/market.ts index 6037c15..2836db2 100644 --- a/apps/web/src/lib/market.ts +++ b/apps/web/src/lib/market.ts @@ -13,3 +13,10 @@ export function splitPrice(usd: number): { operator: number; fee: number } { const fee = Math.round((cents * MARKET_FEE_PCT) / 100); return { operator: (cents - fee) / 100, fee: fee / 100 }; } + +/** + * Share of the buyer's cap an idle node bids in dynamic mode; the bid rises to + * the full cap as its queue fills, never under the operator's minimum. Mirrors + * `DYNAMIC_IDLE_SHARE` in node/crates/c0mpute-core/src/pricing.rs. + */ +export const DYNAMIC_IDLE_SHARE = 0.6; diff --git a/apps/web/src/lib/nodes.ts b/apps/web/src/lib/nodes.ts index 885f5da..81ed080 100644 --- a/apps/web/src/lib/nodes.ts +++ b/apps/web/src/lib/nodes.ts @@ -9,9 +9,28 @@ export interface NodeRow { bench_score: number | null; version: string | null; ref: string | null; + /** The operator's listing; absent on nodes older than 0.2.35. */ + pricing?: Pricing | null; last_seen_s: number; } +/** How a node prices its bids, set by its operator with `c0mpute pricing set`. */ +export interface Pricing { + mode: "dynamic" | "fixed"; + min_usd: number; + fixed_usd?: number; +} + +/** "dynamic · min $0.02" / "fixed $0.30" / "-" for the market table. */ +export function priceLabel(p: Pricing | null | undefined): string { + if (!p) return "-"; + const usd = (n: number) => `$${n < 0.01 && n > 0 ? n : n.toFixed(2)}`; + if (p.mode === "fixed" && p.fixed_usd != null) { + return `fixed ${usd(Math.max(p.fixed_usd, p.min_usd))}`; + } + return p.min_usd > 0 ? `dynamic · min ${usd(p.min_usd)}` : "dynamic"; +} + export interface NodesPayload { generated_at: string; workers_online: number; diff --git a/node/crates/c0mpute-cli/src/main.rs b/node/crates/c0mpute-cli/src/main.rs index 5c91f62..b9c78c8 100644 --- a/node/crates/c0mpute-cli/src/main.rs +++ b/node/crates/c0mpute-cli/src/main.rs @@ -76,6 +76,15 @@ enum Cmd { #[arg(long, value_delimiter = ',', global = true)] threads: Option>, }, + /// Your listing on c0mpute.com/market: how this node prices its bids. + /// + /// Dynamic (default) undercuts the buyer's cap when idle and rises as the + /// queue fills; fixed always asks one price. Neither goes under `--min`. + /// A running worker picks a change up on restart. + Pricing { + #[command(subcommand)] + cmd: Option, + }, /// Worker lifecycle. Worker { #[command(subcommand)] @@ -484,6 +493,7 @@ async fn run_app(cli: Cli) -> Result<()> { threads, } => run_bench(cmd, quick, json, threads), Cmd::StatusAggregator { bind } => c0mpute_core::status_aggregator::run(bind).await, + Cmd::Pricing { cmd } => run_pricing(cmd, &config_path), Cmd::Worker { cmd } => run_worker(cmd, &config_path).await, Cmd::Job { cmd } => run_job(cmd).await, Cmd::Storage { cmd } => storage::run(cmd, &config_path).await, @@ -496,7 +506,9 @@ async fn run_app(cli: Cli) -> Result<()> { Cmd::Uninstall { all, purge, yes } => run_uninstall(all, purge, yes), Cmd::Coinpay { args } => delegate("coinpay", &args), Cmd::Login => run_login(), - Cmd::Infernet { args } if args.first().map(String::as_str) == Some("temporal") => temporal::run(&args[1..]).await, + Cmd::Infernet { args } if args.first().map(String::as_str) == Some("temporal") => { + temporal::run(&args[1..]).await + } Cmd::Infernet { mut args } => { // Default the network to c0mpute when caller didn't specify. if matches!(args.first().map(String::as_str), Some("run")) @@ -723,6 +735,64 @@ async fn run_worker(cmd: WorkerCmd, config_path: &std::path::Path) -> Result<()> // bench // ──────────────────────────────────────────────────────────────────────── +#[derive(Subcommand, Debug)] +enum PricingCmd { + /// Print the current listing (the default). + Show, + /// Set the floor and the mode. + Set { + /// Lowest price per job you accept, in USD. + #[arg(long)] + min: Option, + /// Always ask this price per job (fixed mode). + #[arg(long, conflicts_with = "dynamic")] + fixed: Option, + /// Price by demand: undercut when idle, up to the buyer's cap when busy. + #[arg(long)] + dynamic: bool, + }, +} + +fn run_pricing(cmd: Option, config_path: &std::path::Path) -> Result<()> { + use c0mpute_core::pricing::PricingMode; + let mut cfg = Config::load_or_default(config_path)?; + if let Some(PricingCmd::Set { + min, + fixed, + dynamic, + }) = cmd + { + if let Some(m) = min { + cfg.pricing.min_usd = m; + } + if let Some(f) = fixed { + cfg.pricing.mode = PricingMode::Fixed; + cfg.pricing.fixed_usd = Some(f); + } else if dynamic { + cfg.pricing.mode = PricingMode::Dynamic; + cfg.pricing.fixed_usd = None; + } + cfg.pricing.validate().map_err(anyhow::Error::msg)?; + cfg.save(config_path)?; + println!( + "saved to {} (apply: c0mpute worker stop && c0mpute worker start -d)", + config_path.display() + ); + } + let p = &cfg.pricing; + println!("mode : {}", p.mode.as_str()); + println!("min : ${} per job", p.min_usd); + if let (PricingMode::Fixed, Some(f)) = (p.mode, p.fixed_usd) { + println!("price : ${f} per job"); + } + if let Some(q) = p.quote(1.0, 0.0) { + println!("example : on a $1.00 job you bid ${q:.2} when idle"); + } else { + println!("example : you skip jobs capped at $1.00 (under your floor)"); + } + Ok(()) +} + #[derive(Subcommand, Debug)] enum BenchCmd { /// Run the benchmark and save the report (the default). diff --git a/node/crates/c0mpute-core/src/capabilities.rs b/node/crates/c0mpute-core/src/capabilities.rs index 6043d97..6edfed9 100644 --- a/node/crates/c0mpute-core/src/capabilities.rs +++ b/node/crates/c0mpute-core/src/capabilities.rs @@ -127,8 +127,7 @@ pub fn hardware_blob(config: &Config) -> serde_json::Value { "version": env!("CARGO_PKG_VERSION"), "free_disk_root": config.storage.root.display().to_string(), }); - if let Some(report) = bench_report_path().and_then(|p| c0mpute_bench::load(&p).ok().flatten()) - { + if let Some(report) = bench_report_path().and_then(|p| c0mpute_bench::load(&p).ok().flatten()) { blob["cpu"] = serde_json::Value::String(report.metadata.cpu.clone()); blob["cores"] = serde_json::Value::from(report.metadata.cores); blob["bench_at"] = serde_json::Value::String(report.metadata.start_time.clone()); @@ -136,6 +135,7 @@ pub fn hardware_blob(config: &Config) -> serde_json::Value { if let Some(code) = referral_code() { blob["ref"] = serde_json::Value::String(code); } + blob["pricing"] = config.pricing.listing(); blob } @@ -411,9 +411,7 @@ mod tests { let mut found: Option = None; for _ in 0..50 { let entries = registry.list().await; - if let Some((_, ad)) = - entries.iter().find(|(p, _)| *p == net_a.peer_id()) - { + if let Some((_, ad)) = entries.iter().find(|(p, _)| *p == net_a.peer_id()) { found = Some(ad.clone()); break; } @@ -425,8 +423,7 @@ mod tests { assert_eq!(ad.bench_score, Some(8438), "bench score travels in the ad"); // find_with_tag should return A. - let storage_workers = - registry.find_with_tag("c0mpute:role:storage").await; + let storage_workers = registry.find_with_tag("c0mpute:role:storage").await; assert_eq!(storage_workers.len(), 1); assert_eq!(storage_workers[0].0, net_a.peer_id()); } diff --git a/node/crates/c0mpute-core/src/config.rs b/node/crates/c0mpute-core/src/config.rs index afd2566..84cc5d4 100644 --- a/node/crates/c0mpute-core/src/config.rs +++ b/node/crates/c0mpute-core/src/config.rs @@ -31,6 +31,9 @@ pub struct Config { /// Override the release-feed URL. #[serde(default)] pub update_feed_url: Option, + /// The operator's listing: dynamic or fixed price per job, with a floor. + #[serde(default)] + pub pricing: crate::pricing::PricingConfig, } impl Default for Config { @@ -44,6 +47,7 @@ impl Default for Config { update_auto: default_auto_update(), update_interval_secs: default_update_interval_secs(), update_feed_url: None, + pricing: Default::default(), } } } diff --git a/node/crates/c0mpute-core/src/dispatch.rs b/node/crates/c0mpute-core/src/dispatch.rs index 79a7d26..31d65ca 100644 --- a/node/crates/c0mpute-core/src/dispatch.rs +++ b/node/crates/c0mpute-core/src/dispatch.rs @@ -25,6 +25,7 @@ use c0mpute_net::{GossipMessage, Libp2pNetwork}; use tokio::sync::{broadcast, mpsc}; use tracing::{debug, info, warn}; +use crate::pricing::PricingConfig; use crate::runner::AcceptedJob; /// Reject job offers older than this — replay protection + freshness @@ -56,6 +57,20 @@ pub fn run_worker_subscriber_filtered( advertised_tags: Vec, runner_tx: Option>, filter: Option, +) { + run_worker_subscriber_priced(net, workload_type, advertised_tags, runner_tx, filter, None) +} + +/// [`run_worker_subscriber_filtered`] that bids with the operator's +/// [`PricingConfig`] (dynamic with a floor, or fixed) instead of the buyer's +/// cap. `None` keeps the cap-price bid. +pub fn run_worker_subscriber_priced( + net: Arc, + workload_type: String, + advertised_tags: Vec, + runner_tx: Option>, + filter: Option, + pricing: Option, ) { tokio::spawn(async move { let topic = job_topic(&workload_type); @@ -92,7 +107,16 @@ pub fn run_worker_subscriber_filtered( continue; } } - handle_offer(&net, &topic, &offer, &advertised_tags).await; + let load = runner_tx.as_ref().map(queue_load).unwrap_or(0.0); + handle_offer( + &net, + &topic, + &offer, + &advertised_tags, + pricing.as_ref(), + load, + ) + .await; continue; } if let Ok(accept) = serde_json::from_slice::(&msg.data) { @@ -127,11 +151,22 @@ pub fn run_worker_subscriber_filtered( }); } +/// How full the runner's queue is: 0.0 empty, 1.0 full. +fn queue_load(tx: &mpsc::Sender) -> f64 { + let max = tx.max_capacity(); + if max == 0 { + return 0.0; + } + (max - tx.capacity()) as f64 / max as f64 +} + async fn handle_offer( net: &Libp2pNetwork, topic: &str, offer: &JobOffer, advertised_tags: &[String], + pricing: Option<&PricingConfig>, + load: f64, ) { let now_ms = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) @@ -160,14 +195,23 @@ async fn handle_offer( return; } - // Phase 1: publish a bid at the buyer's max price (cooperative - // pricing). Phase 2 will compute a real per-worker price from - // hardware + queue depth + reputation tier. + // The operator's listing prices the bid (dynamic by queue load, never + // under their floor); without one, bid the buyer's cap. + let price_usd = match pricing { + Some(p) => match p.quote(offer.max_price_usd, load) { + Some(price) => price, + None => { + debug!(job_id = %offer.job_id, cap = offer.max_price_usd, min = p.min_usd, "dispatch: cap under our floor, not bidding"); + return; + } + }, + None => offer.max_price_usd, + }; let bid = JobBid { job_id: offer.job_id.clone(), bidder_peer_id: net.peer_id().to_base58(), bidder_did: None, - price_usd: offer.max_price_usd, + price_usd, eta_seconds: 60, // placeholder; real ETA from the worker pool published_at_ms: now_ms, }; diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index b72153f..307de21 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/node/crates/c0mpute-core/src/lib.rs @@ -8,6 +8,7 @@ pub mod buyer; pub mod capabilities; pub mod config; pub mod dispatch; +pub mod pricing; pub mod register; pub mod runner; pub mod status_aggregator; @@ -17,7 +18,10 @@ pub mod temporal; pub use buyer::{AuctionOutcome, JobAuction, run_auction}; pub use capabilities::{Registry, advertise_loop, tags_from_config}; pub use config::Config; -pub use dispatch::{OfferFilter, run_worker_subscriber, run_worker_subscriber_filtered, workload_types_from_roles}; +pub use dispatch::{ + OfferFilter, run_worker_subscriber, run_worker_subscriber_filtered, + run_worker_subscriber_priced, workload_types_from_roles, +}; pub use register::{Registration, run_register}; pub use runner::TranscodeJobInline; pub use supervisor::Supervisor; diff --git a/node/crates/c0mpute-core/src/pricing.rs b/node/crates/c0mpute-core/src/pricing.rs new file mode 100644 index 0000000..b3f1860 --- /dev/null +++ b/node/crates/c0mpute-core/src/pricing.rs @@ -0,0 +1,172 @@ +//! How a worker prices its bids: the operator's listing. +//! +//! The operator sets a floor (`min_usd`, per job) and a mode: +//! +//! - `dynamic` (the default): an idle node quotes [`DYNAMIC_IDLE_SHARE`] of +//! the buyer's cap to win work, and the quote rises toward the full cap +//! as the node's job queue fills. Never below the floor. +//! - `fixed`: always quote `fixed_usd` (raised to the floor if it is lower). +//! +//! Either way the node does not bid when its price would exceed the buyer's +//! cap. The listing rides in the capability ad's hardware blob, so the node +//! board and c0mpute.com/market show it. + +use serde::{Deserialize, Serialize}; + +/// Share of the buyer's cap an idle node quotes in dynamic mode. +pub const DYNAMIC_IDLE_SHARE: f64 = 0.6; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum PricingMode { + #[default] + Dynamic, + Fixed, +} + +impl PricingMode { + pub fn as_str(self) -> &'static str { + match self { + PricingMode::Dynamic => "dynamic", + PricingMode::Fixed => "fixed", + } + } +} + +#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)] +pub struct PricingConfig { + #[serde(default)] + pub mode: PricingMode, + /// The lowest price per job this node accepts, in USD. 0 = no floor. + #[serde(default)] + pub min_usd: f64, + /// The price per job in `fixed` mode, in USD. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub fixed_usd: Option, +} + +impl PricingConfig { + /// Reject negative or non-finite numbers and a fixed mode with no price. + pub fn validate(&self) -> Result<(), String> { + let ok = |v: f64| v.is_finite() && v >= 0.0; + if !ok(self.min_usd) { + return Err(format!( + "min price must be a number >= 0, got {}", + self.min_usd + )); + } + match (self.mode, self.fixed_usd) { + (PricingMode::Fixed, None) => Err("fixed mode needs a price (--fixed )".into()), + (_, Some(f)) if !ok(f) => Err(format!("fixed price must be a number >= 0, got {f}")), + _ => Ok(()), + } + } + + /// The price this node bids on a job capped at `max_usd`, with its queue + /// `load` between 0.0 (idle) and 1.0 (full). `None` = do not bid. + pub fn quote(&self, max_usd: f64, load: f64) -> Option { + if !max_usd.is_finite() || max_usd < 0.0 { + return None; + } + let floor = self.min_usd.max(0.0); + let price = match self.mode { + PricingMode::Fixed => self.fixed_usd.unwrap_or(max_usd).max(floor), + PricingMode::Dynamic => { + let load = load.clamp(0.0, 1.0); + let share = DYNAMIC_IDLE_SHARE + (1.0 - DYNAMIC_IDLE_SHARE) * load; + (max_usd * share).max(floor) + } + }; + let price = (price * 1e6).round() / 1e6; + (price <= max_usd + 1e-9).then_some(price.min(max_usd)) + } + + /// The public listing carried in the capability ad. + pub fn listing(&self) -> serde_json::Value { + let mut v = serde_json::json!({ "mode": self.mode.as_str(), "min_usd": self.min_usd }); + if let (PricingMode::Fixed, Some(f)) = (self.mode, self.fixed_usd) { + v["fixed_usd"] = serde_json::Value::from(f); + } + v + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn dynamic(min: f64) -> PricingConfig { + PricingConfig { + mode: PricingMode::Dynamic, + min_usd: min, + fixed_usd: None, + } + } + + #[test] + fn idle_dynamic_node_undercuts_the_cap() { + assert_eq!(dynamic(0.0).quote(1.0, 0.0), Some(0.6)); + } + + #[test] + fn dynamic_price_rises_with_load() { + let p = dynamic(0.0); + assert_eq!(p.quote(1.0, 0.5), Some(0.8)); + assert_eq!(p.quote(1.0, 1.0), Some(1.0)); + assert_eq!(p.quote(1.0, 7.0), Some(1.0), "load is clamped"); + } + + #[test] + fn never_quotes_under_the_floor() { + assert_eq!(dynamic(0.75).quote(1.0, 0.0), Some(0.75)); + } + + #[test] + fn no_bid_when_the_floor_is_over_the_cap() { + assert_eq!(dynamic(1.5).quote(1.0, 0.0), None); + } + + #[test] + fn fixed_quotes_its_price_and_respects_the_cap() { + let p = PricingConfig { + mode: PricingMode::Fixed, + min_usd: 0.0, + fixed_usd: Some(0.3), + }; + assert_eq!(p.quote(1.0, 1.0), Some(0.3)); + assert_eq!(p.quote(0.2, 0.0), None); + let floored = PricingConfig { min_usd: 0.5, ..p }; + assert_eq!(floored.quote(1.0, 0.0), Some(0.5)); + } + + #[test] + fn validate_catches_bad_input() { + assert!(dynamic(-1.0).validate().is_err()); + assert!(dynamic(f64::NAN).validate().is_err()); + let p = PricingConfig { + mode: PricingMode::Fixed, + min_usd: 0.0, + fixed_usd: None, + }; + assert!(p.validate().is_err()); + assert!(dynamic(0.02).validate().is_ok()); + } + + #[test] + fn default_config_parses_from_an_empty_section() { + let p: PricingConfig = toml::from_str("").unwrap(); + assert_eq!(p, PricingConfig::default()); + assert_eq!(p.mode, PricingMode::Dynamic); + let p: PricingConfig = + toml::from_str("mode = \"fixed\"\nmin_usd = 0.1\nfixed_usd = 0.2").unwrap(); + assert_eq!(p.quote(1.0, 0.0), Some(0.2)); + } + + #[test] + fn listing_is_public_and_small() { + assert_eq!( + dynamic(0.02).listing(), + serde_json::json!({"mode": "dynamic", "min_usd": 0.02}) + ); + } +} diff --git a/node/crates/c0mpute-core/src/status_aggregator.rs b/node/crates/c0mpute-core/src/status_aggregator.rs index c6ebc15..940e999 100644 --- a/node/crates/c0mpute-core/src/status_aggregator.rs +++ b/node/crates/c0mpute-core/src/status_aggregator.rs @@ -61,7 +61,12 @@ const DHT_CRAWL_INTERVAL: Duration = Duration::from_secs(30); // The temporal workloads (DIP 0026) are listed so seeds running the // aggregator also relay them for buyers that only know the seeds; their // session calls are sealed, so the aggregator sees only ciphertext. -const KNOWN_WORKLOADS: &[&str] = &["ffmpeg.transcode", "infernet.inference", "infernet.temporal.train", "infernet.temporal.infer"]; +const KNOWN_WORKLOADS: &[&str] = &[ + "ffmpeg.transcode", + "infernet.inference", + "infernet.temporal.train", + "infernet.temporal.infer", +]; /// Roles we always surface (even at zero) so the "workers by role" table /// stays populated. Extra roles observed on the wire are added dynamically. @@ -255,7 +260,8 @@ async fn build_payload(state: &AppState) -> StatusPayload { } } } - let bench = BenchSummary::from_scores(ads.iter().filter_map(|(_, ad)| ad.bench_score).collect()); + let bench = + BenchSummary::from_scores(ads.iter().filter_map(|(_, ad)| ad.bench_score).collect()); // Jobs — from the correlated job state. let mut workload_types: BTreeMap = KNOWN_WORKLOADS @@ -362,9 +368,20 @@ struct NodeRow { version: Option, #[serde(rename = "ref")] referral: Option, + /// The operator's listing (`mode`, `min_usd`, `fixed_usd`), if the node + /// advertises one. Re-validated here: only known fields, sane numbers. + pricing: Option, last_seen_s: u64, } +/// The ad's pricing listing, rebuilt from validated fields only. +fn listing(hw: &serde_json::Value) -> Option { + let v = hw.get("pricing")?; + let p: crate::pricing::PricingConfig = serde_json::from_value(v.clone()).ok()?; + let cap = |x: f64| x <= 1_000_000.0; + (p.validate().is_ok() && cap(p.min_usd) && p.fixed_usd.is_none_or(cap)).then(|| p.listing()) +} + #[derive(Serialize, Debug, PartialEq)] struct ReferrerRow { #[serde(rename = "ref")] @@ -404,12 +421,17 @@ fn build_nodes(ads: &[(String, c0mpute_net::topics::CapabilityAd)], now: u64) -> class: node_class(&ad.tags), cores: ad.hardware.get("cores").and_then(|v| v.as_u64()), bench_score: ad.bench_score, - version: ad.hardware.get("version").and_then(|v| v.as_str()).map(|s| s.chars().take(24).collect()), + version: ad + .hardware + .get("version") + .and_then(|v| v.as_str()) + .map(|s| s.chars().take(24).collect()), referral: ad .hardware .get("ref") .and_then(|v| v.as_str()) .and_then(crate::capabilities::sanitize_ref), + pricing: listing(&ad.hardware), last_seen_s: now.saturating_sub(ad.published_at_ms) / 1000, }) .collect(); @@ -530,7 +552,8 @@ pub async fn run(bind: SocketAddr) -> Result<()> { // Best-effort bootstrap from c0mpute.com/bootstrap.json; mDNS covers LAN. let bootstrap_addrs = - bootstrap::fetch_or_fallback(bootstrap::DEFAULT_BOOTSTRAP_URL, bootstrap::default_seeds()).await; + bootstrap::fetch_or_fallback(bootstrap::DEFAULT_BOOTSTRAP_URL, bootstrap::default_seeds()) + .await; info!( count = bootstrap_addrs.len(), "aggregator: bootstrap peers loaded" @@ -629,9 +652,26 @@ mod tests { ad }; let ads = vec![ - ("12D3KooWaaaaaaaaAAAAAAAA".to_string(), mk(&["c0mpute:cpu"], None, serde_json::json!({"ref": "Chovy"}))), - ("12D3KooWbbbbbbbbBBBBBBBB".to_string(), mk(&["c0mpute:gpu:nvidia"], Some(9000), serde_json::json!({"cores": 16, "version": "0.2.32", "ref": "chovy", "cpu": "secret model"}))), - ("12D3KooWccccccccCCCCCCCC".to_string(), mk(&["c0mpute:cpu"], Some(100), serde_json::json!({"ref": ""}))), + ( + "12D3KooWaaaaaaaaAAAAAAAA".to_string(), + mk(&["c0mpute:cpu"], None, serde_json::json!({"ref": "Chovy"})), + ), + ( + "12D3KooWbbbbbbbbBBBBBBBB".to_string(), + mk( + &["c0mpute:gpu:nvidia"], + Some(9000), + serde_json::json!({"cores": 16, "version": "0.2.32", "ref": "chovy", "cpu": "secret model"}), + ), + ), + ( + "12D3KooWccccccccCCCCCCCC".to_string(), + mk( + &["c0mpute:cpu"], + Some(100), + serde_json::json!({"ref": ""}), + ), + ), ]; let p = build_nodes(&ads, 40_000); assert_eq!(p.workers_online, 3); @@ -641,11 +681,42 @@ mod tests { assert_eq!(p.nodes[0].last_seen_s, 30); assert_eq!(p.nodes[2].bench_score, None); assert_eq!(p.nodes[1].referral, None, "invalid ref dropped"); - assert_eq!(p.referrers, vec![ReferrerRow { referral: "chovy".into(), nodes: 2 }]); + assert_eq!( + p.referrers, + vec![ReferrerRow { + referral: "chovy".into(), + nodes: 2 + }] + ); let json = serde_json::to_string(&p.nodes).unwrap(); assert!(!json.contains("secret model"), "cpu model never published"); } + #[test] + fn node_board_carries_only_a_valid_pricing_listing() { + let ok = serde_json::json!({"pricing": {"mode": "dynamic", "min_usd": 0.02, "junk": "x"}}); + assert_eq!( + listing(&ok), + Some(serde_json::json!({"mode": "dynamic", "min_usd": 0.02})) + ); + let fixed = + serde_json::json!({"pricing": {"mode": "fixed", "min_usd": 0.0, "fixed_usd": 0.5}}); + assert_eq!(listing(&fixed).unwrap()["fixed_usd"], 0.5); + assert_eq!( + listing(&serde_json::json!({"pricing": {"mode": "fixed"}})), + None + ); + assert_eq!( + listing(&serde_json::json!({"pricing": {"min_usd": -1}})), + None + ); + assert_eq!( + listing(&serde_json::json!({"pricing": {"mode": "auction"}})), + None + ); + assert_eq!(listing(&serde_json::json!({})), None); + } + #[test] fn iso8601_known_epochs() { assert_eq!(iso8601_utc(0), "1970-01-01T00:00:00Z"); diff --git a/node/crates/c0mpute-core/src/supervisor.rs b/node/crates/c0mpute-core/src/supervisor.rs index d41dc1c..4966c74 100644 --- a/node/crates/c0mpute-core/src/supervisor.rs +++ b/node/crates/c0mpute-core/src/supervisor.rs @@ -31,11 +31,9 @@ impl Supervisor { // /identity.key. Bootstrap list is empty for now — // DIP-0010 wires up c0mpute.com/bootstrap.json once we run // public seed nodes. - let identity_dir = config::config_dir() - .unwrap_or_else(|| std::path::PathBuf::from(".")); + let identity_dir = config::config_dir().unwrap_or_else(|| std::path::PathBuf::from(".")); - let local_source: Arc = - Arc::new(StoreSource(store.clone())); + let local_source: Arc = Arc::new(StoreSource(store.clone())); // Fetch bootstrap list from c0mpute.com (best-effort) and merge // with the hardcoded fallback list. mDNS handles LAN discovery @@ -46,17 +44,13 @@ impl Supervisor { bootstrap::default_seeds(), ) .await; - info!( - count = bootstrap_addrs.len(), - "bootstrap peers loaded" - ); + info!(count = bootstrap_addrs.len(), "bootstrap peers loaded"); let net_cfg = NetworkConfig::for_dir(identity_dir) .with_local_source(local_source) .with_bootstrap(bootstrap_addrs); - let libp2p_net: Arc = - Arc::new(Libp2pNetwork::spawn(net_cfg).await?); + let libp2p_net: Arc = Arc::new(Libp2pNetwork::spawn(net_cfg).await?); info!( peer_id = %libp2p_net.peer_id(), "libp2p network up" @@ -126,11 +120,13 @@ impl Supervisor { } _ => None, }; - dispatch::run_worker_subscriber( + dispatch::run_worker_subscriber_priced( net.clone(), workload_type, tags.clone(), runner_tx, + None, + Some(self.config.pricing.clone()), ); } @@ -153,8 +149,7 @@ impl Supervisor { .update_feed_url .clone() .unwrap_or_else(|| c0mpute_update::DEFAULT_RELEASE_FEED.to_string()); - let interval = - std::time::Duration::from_secs(self.config.update_interval_secs.max(60)); + let interval = std::time::Duration::from_secs(self.config.update_interval_secs.max(60)); let current = env!("CARGO_PKG_VERSION").to_string(); info!( interval_secs = interval.as_secs(), diff --git a/package.json b/package.json index 3da95a6..67472e1 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute-monorepo", - "version": "0.2.34", + "version": "0.2.35", "private": true, "workspaces": [ "apps/*", diff --git a/packages/shared/package.json b/packages/shared/package.json index b683b26..bdea75b 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/shared", - "version": "0.2.34", + "version": "0.2.35", "private": true, "type": "module", "main": "./dist/index.js",