From ab76936df36d0f17072c866c62e78739546caa3b Mon Sep 17 00:00:00 2001 From: Kuro Date: Wed, 7 Oct 2026 16:50:18 +0000 Subject: [PATCH] feat(s1): compute Market API + auth token + workloads + tests (28 verts, preuve live 201/401) --- ROADMAP.md | 13 + docs/JOBS_QUICKSTART.md | 66 ++ helium-mesh/src/bench.rs | 190 ++++++ helium-mesh/src/daemon.rs | 999 ++++++++++++++++++++++++++++++- helium-mesh/src/identity.rs | 23 +- helium-mesh/src/lib.rs | 2 + helium-mesh/src/main.rs | 202 ++++++- helium-mesh/src/market.rs | 258 ++++++-- helium-mesh/src/storage.rs | 4 +- helium-mesh/src/templates.rs | 113 ++++ helium-mesh/src/workload.rs | 92 ++- helium/scripts/demo-live.sh | 13 + helium/scripts/demo-s3-cancel.sh | 29 + helium/scripts/demo-s3-quota.sh | 34 ++ helium/scripts/fc-vm.sh | 42 +- helium/scripts/helium-health.sh | 92 +++ install.sh | 32 + 17 files changed, 2086 insertions(+), 118 deletions(-) create mode 100644 docs/JOBS_QUICKSTART.md create mode 100644 helium-mesh/src/bench.rs create mode 100644 helium-mesh/src/templates.rs create mode 100644 helium/scripts/demo-live.sh create mode 100644 helium/scripts/demo-s3-cancel.sh create mode 100644 helium/scripts/demo-s3-quota.sh create mode 100644 helium/scripts/helium-health.sh diff --git a/ROADMAP.md b/ROADMAP.md index b185ecd..0f0a76b 100644 --- a/ROADMAP.md +++ b/ROADMAP.md @@ -52,6 +52,19 @@ CLI (Rust) -> Discovery (libp2p DHT privee) -> Matching (SQLite) -> Tunnel (Wire - [ ] Invitations QR / X / Discord - [ ] Monitoring & scheduling +### v0.3+ — Kuro x Helium Compute (planifie) +Kuro peut demander du compute au mesh quand il en a besoin. +Loopback par defaut, acces distant uniquement via reseau prive +WireGuard ou tunnel SSH, avec token d API. + +- [x] API jobs : `POST /requests`, `POST /matches/:id/accept`, `GET /workloads` (S1, token requis) +- [x] Scheduler avec quotas par projet, file, annulation et timeout (S3 : 429 quota, cancel, TTL, demos CLI) +- [x] Connecteur Kuro : `GET /api/compute/offers`, `POST /api/compute/request` (S2, file locale) +- [x] Templates `jupyter`/`train`/`batch-scan` + `docs/JOBS_QUICKSTART.md` (Q5) +- [ ] Providers legers (Android branches, classe CPU batch-only ; bench soutenu : fait) +- [ ] 2 demos de bout en bout (<3 etapes chacune) +- [ ] Gate de sortie : 1 job via tunnel prive + tests verts + --- ## MVP Scope diff --git a/docs/JOBS_QUICKSTART.md b/docs/JOBS_QUICKSTART.md new file mode 100644 index 0000000..62884e4 --- /dev/null +++ b/docs/JOBS_QUICKSTART.md @@ -0,0 +1,66 @@ +# Jobs Quickstart — Kuro x Helium Compute (Q5) + +Run a job in 1 command. Everything below targets the local node +(`127.0.0.1:8787` + `helium` CLI). Remote access goes through the private +WireGuard mesh or an SSH tunnel — never expose the ports. + +## 1. Manual job (no market) + +```bash +helium workload run --template jupyter --name my-first-job +helium workload list +``` + +## 2. Market flow (credits + scheduler) + +```bash +helium faucet --party me --amount 100 +helium market offer --amount 32 --price 0.5 +helium market request --template train --max-price 1.0 --hours 2 +helium market match --request-id +helium market accept --match-id # settles + records the workload +helium workload list +``` + +## 3. Via Kuro (any project) + +```bash +curl -s -X POST 127.0.0.1:8767/api/compute/request \ + -H 'Content-Type: application/json' \ + -d '{"project":"openquant","template":"batch-scan","hours":1}' +``` + +## Templates + +| name | image | gpus | for | +|---|---|---|---| +| `jupyter` | jupyter/pytorch-notebook:latest | 1 | JupyterLab :8888 | +| `train` | pytorch/pytorch:latest | 1 | training scripts | +| `batch-scan` | ubuntu:22.04 | 0 | CPU scans, ETL | + +Explicit `--image` / `--gpus` (or API fields) override the template. +Unknown names fail with the known list — locally and over HTTP (400). + +## Bench (score soutenu, jamais le pic) + +```bash +helium bench --seconds 60 # humain +helium bench --seconds 300 --json # score officiel (5 min, throttling visible) +helium market offer --amount 32 --price 0.5 --bench # offre + score inline +helium market request --template train --min-bench 10 # filtre les offres < 10 GFLOPS +``` + +Scores relatifs (meme binaire, meme classe CPU), binaire `--release` +exige pour un score officiel. `derate_pct` = throttling thermique. + +## Knobs (env) + +| var | default | meaning | +|---|---|---| +| `HELIUM_MAX_ACTIVE_JOBS_PER_REQUESTER` | 2 | concurrent jobs per requester (429 beyond) | +| `HELIUM_REQUEST_TTL_HOURS` | 24 | open requests expire after this | +| `HELIUM_HOME` | ~/.helium | state dir (tests, portable nodes) | +| `HELIUM_API_URL` / `HELIUM_API_TOKEN` | 127.0.0.1:8787 | Kuro-side client config | + +Shell note (Windows): run demos under Git-Bash/PowerShell — WSL interop +does not forward custom env to Windows .exe. diff --git a/helium-mesh/src/bench.rs b/helium-mesh/src/bench.rs new file mode 100644 index 0000000..cc09de3 --- /dev/null +++ b/helium-mesh/src/bench.rs @@ -0,0 +1,190 @@ +//! helium bench — sustained provider score, never the marketing peak. +//! +//! Measures what a job will actually get: multi-thread FP32 FMA throughput +//! sampled in windows (peak = best early window, sustained = mean of the +//! late windows), one-shot memory copy bandwidth, and available RAM. +//! Thermal throttling shows up as derate_pct — that is the point: a phone +//! advertising 80 TOPS but sustaining 40% of it scores 40%. +//! +//! Scores are RELATIVE: compare runs of the same binary on the same CPU +//! class. Always bench with a --release build (dev profile is ~10x slower +//! and meaningless). Cross-machine absolute calibration (reference table +//! per CPU class) is future work, not needed for match filtering. + +use serde::Serialize; +use std::time::{Duration, Instant}; + +/// FP32 multiply-adds per loop iteration (2 FLOP each). +const FMAS_PER_ITER: u64 = 2; +/// Elements per worker buffer (fits L2, streams through it). +const BUF_LEN: usize = 4096; +/// Target wall time of one sampling window. +const WINDOW: Duration = Duration::from_millis(200); + +#[derive(Debug, Clone, Serialize)] +pub struct BenchReport { + /// Requested wall time (seconds). + pub seconds: u64, + /// Best early-window throughput (GFLOPS, all threads). + pub gflops_peak: f64, + /// Mean late-window throughput (GFLOPS, all threads). + pub gflops_sustained: f64, + /// 100 * (1 - sustained/peak), clamped >= 0. Throttling indicator. + pub derate_pct: f64, + /// Single memory copy bandwidth (GB/s). + pub mem_bw_gbs: f64, + pub ram_total_gb: f64, + pub ram_avail_gb: f64, + /// Provisional class from sustained numbers ("cpu" or "cpu-le"). + pub suggested_class: String, +} + +/// Provisional floor for a full "cpu" class: 20 sustained GFLOPS + 8 GB free. +/// Below either -> "cpu-le" (phones, old laptops, throttled boxes). +pub fn suggest_class(gflops_sustained: f64, ram_avail_gb: f64) -> &'static str { + if gflops_sustained >= 20.0 && ram_avail_gb >= 8.0 { + "cpu" + } else { + "cpu-le" + } +} + +fn worker_fma(windows: &mut Vec, deadline: Instant) { + let a: Vec = (0..BUF_LEN).map(|i| (i as f32) * 0.0001 + 1.0).collect(); + let b: Vec = (0..BUF_LEN).map(|i| (i as f32) * -0.0002 + 2.0).collect(); + // Eight independent accumulators: a single chain would serialize on FMA + // latency (~5 cycles) and report one core at a fraction of its power. + let mut acc = [0.0f32; 8]; + let mut window_ops: u64 = 0; + let mut window_start = Instant::now(); + while Instant::now() < deadline { + for i in (0..BUF_LEN).step_by(8) { + acc[0] = a[i].mul_add(b[i], acc[0]); + acc[1] = a[i + 1].mul_add(b[i + 1], acc[1]); + acc[2] = a[i + 2].mul_add(b[i + 2], acc[2]); + acc[3] = a[i + 3].mul_add(b[i + 3], acc[3]); + acc[4] = a[i + 4].mul_add(b[i + 4], acc[4]); + acc[5] = a[i + 5].mul_add(b[i + 5], acc[5]); + acc[6] = a[i + 6].mul_add(b[i + 6], acc[6]); + acc[7] = a[i + 7].mul_add(b[i + 7], acc[7]); + } + // Keep the accumulators live so the loop is not optimized away. + if acc.iter().any(|x| x.is_infinite()) { + acc = [0.0; 8]; + } + window_ops += BUF_LEN as u64 * FMAS_PER_ITER; + let elapsed = window_start.elapsed(); + if elapsed >= WINDOW { + windows.push(window_ops as f64 / elapsed.as_secs_f64() / 1e9); + window_ops = 0; + window_start = Instant::now(); + } + } + if window_start.elapsed() > Duration::from_millis(10) && window_ops > 0 { + windows.push(window_ops as f64 / window_start.elapsed().as_secs_f64() / 1e9); + } + std::hint::black_box(acc); +} + +fn copy_bandwidth_gbs() -> f64 { + const BYTES: usize = 256 * 1024 * 1024; + let src = vec![0xA5u8; BYTES]; + let mut dst = vec![0u8; BYTES]; + let start = Instant::now(); + dst.copy_from_slice(&src); + let secs = start.elapsed().as_secs_f64().max(1e-9); + std::hint::black_box(dst); + // Count read + write traffic. + (2 * BYTES) as f64 / secs / 1e9 +} + +fn ram_gb() -> (f64, f64) { + let mut sys = sysinfo::System::new(); + sys.refresh_memory(); + ( + sys.total_memory() as f64 / 1e9, + sys.available_memory() as f64 / 1e9, + ) +} + +/// Run the benchmark for (about) `seconds`. Short runs (< 5 s) still work +/// but only long runs (5 min recommended) expose thermal throttling. +pub fn run_bench(seconds: u64) -> BenchReport { + let seconds = seconds.max(1); + let deadline = Instant::now() + Duration::from_secs(seconds); + let threads = std::thread::available_parallelism() + .map(|n| n.get()) + .unwrap_or(1) + .min(64); + let mut handles = Vec::with_capacity(threads); + for _ in 0..threads { + handles.push(std::thread::spawn(move || { + let mut windows = Vec::new(); + worker_fma(&mut windows, deadline); + windows + })); + } + let mut agg: Vec = Vec::new(); + for h in handles { + let windows = h.join().unwrap_or_default(); + for (i, rate) in windows.into_iter().enumerate() { + if agg.len() <= i { + agg.push(0.0); + } + agg[i] += rate; + } + } + // Peak = best of the first quarter of windows, sustained = mean of the + // last quarter. Single-window runs report the same value twice. + let q = (agg.len() / 4).max(1); + let peak = agg.iter().take(q).cloned().fold(0.0f64, f64::max); + let tail = &agg[agg.len().saturating_sub(q)..]; + let sustained = if tail.is_empty() { + 0.0 + } else { + tail.iter().sum::() / tail.len() as f64 + }; + let derate_pct = if peak > 0.0 { + (100.0 * (1.0 - sustained / peak)).max(0.0) + } else { + 0.0 + }; + let (ram_total_gb, ram_avail_gb) = ram_gb(); + BenchReport { + seconds, + gflops_peak: peak, + gflops_sustained: sustained, + derate_pct, + mem_bw_gbs: copy_bandwidth_gbs(), + ram_total_gb, + ram_avail_gb, + suggested_class: suggest_class(sustained, ram_avail_gb).to_string(), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn short_bench_reports_sane_numbers() { + let r = run_bench(1); + assert_eq!(r.seconds, 1); + assert!(r.gflops_peak > 0.0, "peak must be positive"); + assert!(r.gflops_sustained > 0.0, "sustained must be positive"); + assert!(r.gflops_sustained <= r.gflops_peak * 1.5, "windows are noisy, not magic"); + assert!((0.0..=100.0).contains(&r.derate_pct)); + assert!(r.mem_bw_gbs > 0.0); + assert!(r.ram_total_gb > 0.0); + assert!(r.ram_avail_gb >= 0.0); + assert!(r.suggested_class == "cpu" || r.suggested_class == "cpu-le"); + } + + #[test] + fn class_thresholds() { + assert_eq!(suggest_class(20.0, 8.0), "cpu"); + assert_eq!(suggest_class(100.0, 32.0), "cpu"); + assert_eq!(suggest_class(19.9, 64.0), "cpu-le"); + assert_eq!(suggest_class(200.0, 7.9), "cpu-le"); + } +} diff --git a/helium-mesh/src/daemon.rs b/helium-mesh/src/daemon.rs index 8220870..aec680e 100644 --- a/helium-mesh/src/daemon.rs +++ b/helium-mesh/src/daemon.rs @@ -6,6 +6,7 @@ use crate::identity::IdentityManager; use crate::market::{Market, MatchRow}; use crate::mesh::MeshManager; use crate::discovery::DhtDiscovery; +use crate::workload::WorkloadManager; pub struct HeliumDaemon { bind_address: SocketAddr, @@ -87,19 +88,408 @@ impl HeliumDaemon { } /// HTTP API (powers the future web UI and remote monitoring). + /// S1 (Programme F): job submission for Kuro. Everything except + /// /health requires the node API token (`helium token` prints it). async fn api_server(bind: std::net::SocketAddr) -> Result<()> { - let app = axum::Router::new() - .route("/health", axum::routing::get(|| async { "ok" })) - .route("/status", axum::routing::get(api_status)); let listener = tokio::net::TcpListener::bind(bind) .await .with_context(|| format!("cannot bind API to {bind}"))?; - axum::serve(listener, app) + axum::serve(listener, build_router()) .await .context("API server failed")?; Ok(()) } +} + +/// S1 job API routes. Auth: Bearer token on everything except /health. +fn build_router() -> axum::Router { + use axum::routing::{get, post}; + axum::Router::new() + .route("/health", get(|| async { "ok" })) + .route("/status", get(api_status)) + .route("/requests", post(api_create_request).get(api_list_requests)) + .route("/requests/:id", get(api_get_request)) + .route("/requests/:id/matches", get(api_request_matches)) + .route("/requests/:id/cancel", post(api_cancel_request)) + .route("/offers", get(api_list_offers)) + .route("/matches/:id/accept", post(api_accept_match)) + .route("/workloads", get(api_list_workloads)) + .route("/workloads/:id", get(api_get_workload)) + .route("/workloads/:id/logs", get(api_workload_logs)) +} + +type ApiOut = (axum::http::StatusCode, axum::Json); + +fn authed(headers: &axum::http::HeaderMap) -> bool { + crate::identity::authorized(headers, &crate::identity::api_token()) +} + +fn unauthorized() -> ApiOut { + ( + axum::http::StatusCode::UNAUTHORIZED, + axum::Json(serde_json::json!({ + "error": "missing or invalid bearer token (run `helium token` on the node)" + })), + ) +} + +fn db_error(context: &str) -> ApiOut { + ( + axum::http::StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ "error": format!("{context}: cannot open local state") })), + ) +} + +/// Missing row -> 404, anything else (already matched, no credits) -> 409. +fn market_status(err: &anyhow::Error) -> axum::http::StatusCode { + let msg = err.to_string().to_lowercase(); + if msg.contains("not found") || msg.contains("no row") { + axum::http::StatusCode::NOT_FOUND + } else { + axum::http::StatusCode::CONFLICT + } +} + +fn market_error(err: anyhow::Error) -> ApiOut { + ( + market_status(&err), + axum::Json(serde_json::json!({ "error": err.to_string() })), + ) +} + +#[derive(Debug, serde::Deserialize)] +struct CreateRequestBody { + /// Who asks (e.g. "kuro:openquant"). Empty = this node. + #[serde(default)] + requester: String, + #[serde(default = "default_rtype")] + rtype: String, + amount: u32, + max_price: f64, + hours: u32, + #[serde(default)] + wg_pubkey: String, + #[serde(default)] + endpoint: String, + /// S2: container image wanted ("" = node default, see Q5 templates). + #[serde(default)] + image: String, + /// S2: GPUs wanted (0 = unset, node decides). + #[serde(default)] + gpus: u32, + /// Q5: named template (jupyter, train, batch-scan); explicit fields win. + #[serde(default)] + template: String, + /// Bench floor: offers below this sustained GFLOPS never match (0 = any). + #[serde(default)] + min_bench: f64, +} + +fn default_rtype() -> String { + "gpu".to_string() +} + +/// Pure validation so it stays unit-testable without touching disk. +fn validate_request(rtype: &str, amount: u32, max_price: f64, hours: u32, image: &str, gpus: u32, template: &str, min_bench: f64) -> std::result::Result<(), String> { + let rtype = rtype.trim(); + if rtype.is_empty() { + return Err("rtype is required (e.g. gpu)".to_string()); + } + if rtype.len() > 32 { + return Err("rtype must be 32 chars or less".to_string()); + } + if amount == 0 || amount > 1024 { + return Err("amount must be between 1 and 1024".to_string()); + } + if max_price.is_nan() || max_price <= 0.0 || max_price > 1000.0 { + return Err("max_price must be greater than 0 and at most 1000".to_string()); + } + if hours == 0 || hours > 720 { + return Err("hours must be between 1 and 720".to_string()); + } + if image.len() > 256 { + return Err("image must be 256 chars or less".to_string()); + } + if gpus > 16 { + return Err("gpus must be 16 or less".to_string()); + } + if image.len() > 256 { + return Err("image must be 256 chars or less".to_string()); + } + if template.trim().len() > 64 { + return Err("template must be 64 chars or less".to_string()); + } + if min_bench.is_nan() || !(0.0..=1e6).contains(&min_bench) { + return Err("min_bench must be between 0 and 1000000".to_string()); + } + Ok(()) +} + +async fn api_create_request( + headers: axum::http::HeaderMap, + axum::Json(body): axum::Json, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + if let Err(detail) = validate_request(&body.rtype, body.amount, body.max_price, body.hours, &body.image, body.gpus, &body.template, body.min_bench) { + return (StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({ "error": detail }))); + } + let requester = if body.requester.trim().is_empty() { + IdentityManager::load() + .await + .ok() + .and_then(|idm| idm.node_id().ok()) + .unwrap_or_else(|| "api-client".to_string()) + } else { + body.requester.trim().to_string() + }; + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + let (image, gpus) = match crate::templates::apply(&body.template, &body.image, body.gpus) { + Ok(v) => v, + Err(detail) => { + return (StatusCode::BAD_REQUEST, axum::Json(serde_json::json!({ "error": detail }))) + } + }; + let id = match market.add_request(&crate::market::NewRequest { + requester, + rtype: body.rtype.trim().to_string(), + amount: body.amount, + max_price: body.max_price, + hours: body.hours, + wg_pubkey: body.wg_pubkey.clone(), + endpoint: body.endpoint.clone(), + image, + gpus, + min_bench: body.min_bench, + }) { + Ok(id) => id, + Err(e) => return market_error(e), + }; + match market.get_request(&id) { + Ok(req) => ( + StatusCode::CREATED, + axum::Json(serde_json::to_value(&req).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => market_error(e), + } +} + +async fn api_list_requests(headers: axum::http::HeaderMap) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + match market.list_requests() { + Ok(rows) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&rows).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => market_error(e), + } +} + +async fn api_get_request( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + match market.get_request(&id) { + Ok(req) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&req).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => market_error(e), + } +} + +async fn api_list_offers(headers: axum::http::HeaderMap) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + match market.list_offers() { + Ok(rows) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&rows).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => market_error(e), + } +} + +async fn api_request_matches( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + match market.find_matches(&id) { + Ok(rows) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&rows).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => market_error(e), + } +} + +async fn api_accept_match( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let me = IdentityManager::load() + .await + .ok() + .and_then(|idm| idm.node_id().ok()) + .unwrap_or_default(); + match settle_match(&id, &me).await { + Ok(report) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&report).unwrap_or(serde_json::Value::Null)), + ), + Err(e) => { + let msg = e.to_string(); + if msg.contains("quota exceeded") { + ( + StatusCode::TOO_MANY_REQUESTS, + axum::Json(serde_json::json!({ "error": msg })), + ) + } else if msg.contains("recording workload") + || msg.contains("opening market") + || msg.contains("opening workloads") + { + ( + StatusCode::INTERNAL_SERVER_ERROR, + axum::Json(serde_json::json!({ "error": msg })), + ) + } else { + market_error(e) + } + } + } +} + +async fn api_cancel_request( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let market = match Market::open() { + Ok(m) => m, + Err(_) => return db_error("market"), + }; + match market.cancel_request(&id) { + Ok(status) => ( + StatusCode::OK, + axum::Json(serde_json::json!({ "id": id, "status": status })), + ), + Err(e) => market_error(e), + } +} + +async fn api_list_workloads(headers: axum::http::HeaderMap) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let manager = match WorkloadManager::new().await { + Ok(m) => m, + Err(_) => return db_error("workloads"), + }; + ( + StatusCode::OK, + axum::Json( + serde_json::to_value(&manager.list_workloads().await) + .unwrap_or(serde_json::Value::Null), + ), + ) +} + +async fn api_get_workload( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let manager = match WorkloadManager::new().await { + Ok(m) => m, + Err(_) => return db_error("workloads"), + }; + match manager.get_workload(&id).await { + Some(w) => ( + StatusCode::OK, + axum::Json(serde_json::to_value(&w).unwrap_or(serde_json::Value::Null)), + ), + None => ( + StatusCode::NOT_FOUND, + axum::Json(serde_json::json!({ "error": format!("workload {id} not found") })), + ), + } +} + +async fn api_workload_logs( + axum::extract::Path(id): axum::extract::Path, + headers: axum::http::HeaderMap, +) -> ApiOut { + use axum::http::StatusCode; + if !authed(&headers) { + return unauthorized(); + } + let manager = match WorkloadManager::new().await { + Ok(m) => m, + Err(_) => return db_error("workloads"), + }; + match manager.get_workload(&id).await { + Some(w) => ( + StatusCode::OK, + axum::Json(serde_json::json!({ + "id": w.id, + "status": w.status.to_string(), + "logs": w.logs, + })), + ), + None => ( + StatusCode::NOT_FOUND, + axum::Json(serde_json::json!({ "error": format!("workload {id} not found") })), + ), + } +} +/// Background loops (second half of the daemon impl). +impl HeliumDaemon { /// Auto-match loop: settle affordable open requests without manual CLI. /// Policy = the request itself (max price cap = consent). Runs every 30s. /// After settling, provisions the WireGuard tunnel both ways (best effort). @@ -116,6 +506,12 @@ impl HeliumDaemon { continue; } }; + // S3: evict open requests older than the TTL so the queue cannot rot. + match market.expire_stale(request_ttl_hours()) { + Ok(0) => {} + Ok(n) => info!("auto-match expired {n} stale request(s)"), + Err(e) => debug!("auto-match: cannot expire stale requests: {e}"), + }; let me = crate::identity::IdentityManager::load() .await .ok() @@ -134,14 +530,14 @@ impl HeliumDaemon { match market.find_matches(&req.id) { Ok(matches) if !matches.is_empty() => { let best = &matches[0]; - match market.accept(&best.id) { - Ok((from, to, cost)) => { + match settle_match(&best.id, &me).await { + Ok(report) => { info!( - "auto-matched {}: {} -> {} : {:.2} credits (score {:.2})", - best.id, from, to, cost, best.score + "auto-matched {}: {} -> {} : {:.2} credits (workload {})", + report.match_id, report.from, report.to, + report.cost, report.workload_id ); settled += 1; - Self::provision_tunnel(&market, &me, best); } Err(e) => debug!("auto-accept skipped for {}: {}", req.id, e), } @@ -160,31 +556,32 @@ impl HeliumDaemon { /// Provider side: allocate a tunnel IP and add the borrower as peer. /// Borrower side: add the provider (endpoint + key from the offer). /// Best effort: warns and keeps going when not root or incomplete data. + /// Returns a one-line outcome for workload logs (S2). fn provision_tunnel( market: &Market, me: &str, m: &MatchRow, - ) { + ) -> String { use crate::networking::{PeerConfig, WireGuardInterface}; let offer = match market.get_offer(&m.offer_id) { Ok(o) => o, Err(e) => { warn!("provision: {e}"); - return; + return format!("tunnel skipped: {e}"); } }; let request = match market.get_request(&m.request_id) { Ok(r) => r, Err(e) => { warn!("provision: {e}"); - return; + return format!("tunnel skipped: {e}"); } }; let wg = WireGuardInterface::new("helium-poc"); if !wg.exists() { debug!("provision: no local helium-poc interface, skipping"); - return; + return "tunnel skipped: no local helium-poc interface".to_string(); } if offer.provider == me && !request.wg_pubkey.is_empty() { @@ -207,11 +604,18 @@ impl HeliumDaemon { request.requester, m.id ); Self::maybe_boot_vm(offer.amount); + format!("tunnel ready: borrower {} -> {ip} (match {})", request.requester, m.id) + } + Err(e) => { + warn!("provision: cannot add borrower peer: {e}"); + format!("tunnel skipped: cannot add borrower peer: {e}") } - Err(e) => warn!("provision: cannot add borrower peer: {e}"), } } - Err(e) => warn!("provision: IPAM failed: {e}"), + Err(e) => { + warn!("provision: IPAM failed: {e}"); + format!("tunnel skipped: IPAM failed: {e}") + } } } else if request.requester == me && !offer.wg_pubkey.is_empty() @@ -224,14 +628,21 @@ impl HeliumDaemon { keepalive: Some(25), }; match wg.add_peer(&peer) { - Ok(()) => info!( - "tunnel ready: provider {} via {} (match {})", - offer.provider, offer.endpoint, m.id - ), - Err(e) => warn!("provision: cannot add provider peer: {e}"), + Ok(()) => { + info!( + "tunnel ready: provider {} via {} (match {})", + offer.provider, offer.endpoint, m.id + ); + format!("tunnel ready: provider {} via {} (match {})", offer.provider, offer.endpoint, m.id) + } + Err(e) => { + warn!("provision: cannot add provider peer: {e}"); + format!("tunnel skipped: cannot add provider peer: {e}") + } } } else { debug!("provision: not a party or missing keys, skipping"); + "tunnel skipped: not a party or missing keys".to_string() } } @@ -287,6 +698,92 @@ impl HeliumDaemon { } } +/// Image recorded when the request names none. Real templates arrive in Q5. +const DEFAULT_JOB_IMAGE: &str = "helium/job:latest"; + +/// S3 scheduler knobs (env-overridable, documented in the S3 demo scripts). +const DEFAULT_QUOTA_JOBS_PER_REQUESTER: u32 = 2; +const DEFAULT_REQUEST_TTL_HOURS: u32 = 24; + +fn env_u32(key: &str, default: u32) -> u32 { + std::env::var(key).ok().and_then(|v| v.parse().ok()).unwrap_or(default) +} + +fn quota_jobs_per_requester() -> u32 { + env_u32("HELIUM_MAX_ACTIVE_JOBS_PER_REQUESTER", DEFAULT_QUOTA_JOBS_PER_REQUESTER) +} + +fn request_ttl_hours() -> u32 { + env_u32("HELIUM_REQUEST_TTL_HOURS", DEFAULT_REQUEST_TTL_HOURS) +} + +/// S2: outcome of settling one match, shared by the CLI, the API and the loop. +#[derive(Debug, Clone, serde::Serialize)] +pub struct SettleReport { + pub match_id: String, + pub from: String, + pub to: String, + pub cost: f64, + pub workload_id: String, + pub workload_status: String, + pub tunnel: String, +} + +/// Accept a proposed match, provision the tunnel and record the workload. +/// Single entry point so CLI, API and auto-match behave the same (S2). +/// Opens its own market connection: Market is Send but not Sync, so no +/// shared reference may cross an await (tokio::spawn + axum need Send). +/// Workload-stage failures carry a "recording workload" context so the API +/// can answer 500 instead of a misleading 409. +pub async fn settle_match(match_id: &str, me: &str) -> Result { + let market = Market::open().context("opening market")?; + let m = market.get_match(match_id)?; + let request = market.get_request(&m.request_id)?; + // S3 quota first: reject before any credit moves, request stays open. + let manager = WorkloadManager::new() + .await + .context("opening workloads")?; + let quota = quota_jobs_per_requester(); + let active = manager.count_active_for(&request.requester).await; + if active >= quota as usize { + anyhow::bail!( + "quota exceeded for {}: {active}/{quota} active jobs (HELIUM_MAX_ACTIVE_JOBS_PER_REQUESTER)", + request.requester + ); + } + let (from, to, cost) = market.accept(match_id)?; + let tunnel = HeliumDaemon::provision_tunnel(&market, me, &m); + let image = if request.image.trim().is_empty() { + DEFAULT_JOB_IMAGE.to_string() + } else { + request.image.clone() + }; + let workload = manager + .create_workload( + format!("job-{}", request.id), + image, + request.gpus, + &request.id, + &request.requester, + ) + .await + .context("recording workload: cannot create")?; + let line = format!("matched {match_id} ({} -> {} : {:.2} credits); {}", from, to, cost, tunnel); + manager + .mark_running(&workload.id, line) + .await + .context("recording workload: cannot mark running")?; + Ok(SettleReport { + match_id: match_id.to_string(), + from, + to, + cost, + workload_id: workload.id, + workload_status: "running".to_string(), + tunnel, + }) +} + async fn api_status( headers: axum::http::HeaderMap, ) -> Result, axum::http::StatusCode> { @@ -316,12 +813,31 @@ async fn api_status( ), Err(_) => (0, 0), }; + let (workloads_pending, workloads_running) = match WorkloadManager::new().await { + Ok(mgr) => { + let mut pending = 0; + let mut running = 0; + for w in mgr.list_workloads().await { + match w.status { + crate::workload::WorkloadStatus::Pending => pending += 1, + crate::workload::WorkloadStatus::Running => running += 1, + _ => {} + } + } + (pending, running) + } + Err(_) => (0, 0), + }; Ok(axum::Json(serde_json::json!({ "node": node_id, "mesh_joined": joined, "mesh_peers": peers, "open_offers": offers, "open_requests": requests, + "workloads_pending": workloads_pending, + "workloads_running": workloads_running, + "quota_jobs_per_requester": quota_jobs_per_requester(), + "request_ttl_hours": request_ttl_hours(), "discovered_lan": DhtDiscovery::known_peers().len(), "discovered_dht": DhtDiscovery::kad_peers().len(), "version": env!("CARGO_PKG_VERSION"), @@ -348,6 +864,54 @@ async fn shutdown_signal() { mod tests { use super::*; + /// Process-global serial lock for env-mutating HTTP tests: HELIUM_HOME + /// (like HOME) is shared by all test threads in the process. + static ISOLATED_HOME_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(()); + + /// Redirect node state to a temp dir (dirs ignores process env on + /// Windows, so HELIUM_HOME exists for exactly this). The lock guard MUST + /// stay alive until after restore (declare it first so it drops last). + type HomeGuard = ( + tempfile::TempDir, + std::sync::MutexGuard<'static, ()>, + Vec<(String, Option)>, + ); + + fn lock_isolated_home() -> HomeGuard { + let guard = ISOLATED_HOME_LOCK.lock().unwrap_or_else(|e| e.into_inner()); + let tmp = tempfile::tempdir().expect("tempdir for isolated HOME"); + let keys = ["HELIUM_HOME", "HOME", "USERPROFILE"]; + let old: Vec<(String, Option)> = keys + .iter() + .map(|k| (k.to_string(), std::env::var_os(k))) + .collect(); + std::env::set_var("HELIUM_HOME", tmp.path()); + std::env::set_var("HOME", tmp.path()); + std::env::set_var("USERPROFILE", tmp.path()); + (tmp, guard, old) + } + + fn restore_home(old: Vec<(String, Option)>) { + for (k, v) in old { + match v { + Some(x) => std::env::set_var(&k, x), + None => std::env::remove_var(&k), + } + } + } + + fn new_offer(provider: &str) -> crate::market::NewOffer { + crate::market::NewOffer { + provider: provider.to_string(), + rtype: "gpu".to_string(), + amount: 64, + price: 0.5, + endpoint: String::new(), + wg_pubkey: String::new(), + bench_gflops: 0.0, + } + } + #[test] fn vm_mem_sizing() { assert_eq!(HeliumDaemon::vm_mem_for_offer(0), 512); // floor @@ -355,4 +919,399 @@ mod tests { assert_eq!(HeliumDaemon::vm_mem_for_offer(2), 2048); assert_eq!(HeliumDaemon::vm_mem_for_offer(64), 2048); // cap } + + #[test] + fn s1_request_validation() { + assert!(validate_request("gpu", 16, 1.0, 2, "", 0, "", 0.0).is_ok()); + assert!(validate_request("ram", 1, 0.01, 1, "ubuntu:22.04", 1, "", 0.0).is_ok()); + assert!(validate_request("gpu", 16, 1.0, 2, "", 0, "jupyter", 0.0).is_ok()); + assert!(validate_request("", 16, 1.0, 2, "", 0, "", 0.0).is_err()); // missing rtype + assert!(validate_request(" ", 16, 1.0, 2, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 0, 1.0, 2, "", 0, "", 0.0).is_err()); // amount 0 + assert!(validate_request("gpu", 1025, 1.0, 2, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 16, 0.0, 2, "", 0, "", 0.0).is_err()); // free + assert!(validate_request("gpu", 16, -1.0, 2, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 16, f64::NAN, 2, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 16, 1000.01, 2, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 16, 1.0, 0, "", 0, "", 0.0).is_err()); // hours 0 + assert!(validate_request("gpu", 16, 1.0, 721, "", 0, "", 0.0).is_err()); + assert!(validate_request("gpu", 16, 1.0, 2, &"x".repeat(257), 0, "", 0.0).is_err()); // image long + assert!(validate_request("gpu", 16, 1.0, 2, "", 17, "", 0.0).is_err()); // too many gpus + assert!(validate_request("gpu", 16, 1.0, 2, "", 0, &"t".repeat(65), 0.0).is_err()); // template long + assert!(validate_request("gpu", 16, 1.0, 2, "", 0, "", -1.0).is_err()); // min_bench negative + assert!(validate_request("gpu", 16, 1.0, 2, "", 0, "", f64::NAN).is_err()); + } + + #[test] + fn s1_market_error_mapping() { + use axum::http::StatusCode; + let not_found = anyhow::anyhow!("request req-abcdef12 not found"); + assert_eq!(market_status(¬_found), StatusCode::NOT_FOUND); + let no_rows = anyhow::anyhow!("Query returned no rows"); + assert_eq!(market_status(&no_rows), StatusCode::NOT_FOUND); + let taken = anyhow::anyhow!("match match-abc is not proposed (status: accepted)"); + assert_eq!(market_status(&taken), StatusCode::CONFLICT); + let broke = anyhow::anyhow!("insufficient credits for alice: 0.00 < 1.00"); + assert_eq!(market_status(&broke), StatusCode::CONFLICT); + } + + /// S1 proof over real HTTP: 401 without token, 201 on valid request, + /// 400 on invalid values, 404 on unknown ids, full match/accept loop. + /// Runs against an isolated HOME so no real node state is touched. + /// Test-only serial lock: no deadlock (holders never wait on each other + /// while holding), but it parks a worker thread, hence the allow. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn s1_job_api_http() { + let (_tmp, _home_guard, old_env) = lock_isolated_home(); + + let token = crate::identity::api_token(); + assert!(!token.is_empty()); + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + let server = tokio::spawn(async move { + axum::serve(listener, build_router()).await + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + + // No token -> 401, token -> 200 (health stays open by design). + let r_health = client.get(format!("{base}/health")).send().await.expect("health"); + let r_status_anon = client.get(format!("{base}/status")).send().await.expect("status anon"); + let r_status_auth = client + .get(format!("{base}/status")) + .bearer_auth(&token) + .send() + .await + .expect("status auth"); + let r_create_anon = client + .post(format!("{base}/requests")) + .json(&serde_json::json!({"rtype": "gpu", "amount": 16, "max_price": 1.0, "hours": 2})) + .send() + .await + .expect("create anon"); + // Invalid values with token -> 400, never 201. + let r_create_bad = client + .post(format!("{base}/requests")) + .bearer_auth(&token) + .json(&serde_json::json!({"rtype": "gpu", "amount": 0, "max_price": 1.0, "hours": 2})) + .send() + .await + .expect("create bad"); + // Valid request -> 201 with an id. + let r_create = client + .post(format!("{base}/requests")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "requester": "kuro:test", "rtype": "gpu", + "amount": 16, "max_price": 1.0, "hours": 2, + })) + .send() + .await + .expect("create"); + let created: serde_json::Value = r_create.json().await.expect("create body"); + let req_id = created.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(); + + let r_get = client + .get(format!("{base}/requests/{req_id}")) + .bearer_auth(&token) + .send() + .await + .expect("get request"); + let r_get_missing = client + .get(format!("{base}/requests/req-nope1234")) + .bearer_auth(&token) + .send() + .await + .expect("get missing"); + let r_offers = client + .get(format!("{base}/offers")) + .bearer_auth(&token) + .send() + .await + .expect("offers"); + let r_workloads = client + .get(format!("{base}/workloads")) + .bearer_auth(&token) + .send() + .await + .expect("workloads"); + let r_workload_missing = client + .get(format!("{base}/workloads/wl-nope1234")) + .bearer_auth(&token) + .send() + .await + .expect("workload missing"); + let r_accept_missing = client + .post(format!("{base}/matches/match-nope12/accept")) + .bearer_auth(&token) + .send() + .await + .expect("accept missing"); + + // Full loop: provider offer + credits, then match + accept over HTTP. + let market = Market::open().expect("open market in isolated HOME"); + market.faucet("kuro:test", 100.0).expect("faucet"); + market.add_offer(&new_offer("provider:test")).expect("offer"); + let r_matches = client + .get(format!("{base}/requests/{req_id}/matches")) + .bearer_auth(&token) + .send() + .await + .expect("matches"); + let matches: serde_json::Value = r_matches.json().await.expect("matches body"); + let match_id = matches.get(0).and_then(|m| m.get("id")).and_then(|v| v.as_str()).unwrap_or("").to_string(); + let r_accept = client + .post(format!("{base}/matches/{match_id}/accept")) + .bearer_auth(&token) + .send() + .await + .expect("accept"); + + server.abort(); + restore_home(old_env); + + assert_eq!(r_health.status().as_u16(), 200); + assert_eq!(r_status_anon.status().as_u16(), 401); + assert_eq!(r_status_auth.status().as_u16(), 200); + assert_eq!(r_create_anon.status().as_u16(), 401); + assert_eq!(r_create_bad.status().as_u16(), 400); + assert!(!req_id.is_empty(), "201 body must carry the request id: {created}"); + assert_eq!(r_get.status().as_u16(), 200); + assert_eq!(r_get_missing.status().as_u16(), 404); + assert_eq!(r_offers.status().as_u16(), 200); + assert_eq!(r_workloads.status().as_u16(), 200); + assert_eq!(r_workload_missing.status().as_u16(), 404); + assert_eq!(r_accept_missing.status().as_u16(), 404); + assert!(!match_id.is_empty(), "matches must list the test offer: {matches}"); + assert_eq!(r_accept.status().as_u16(), 200); + } + + /// S2 proof: accepting a match records a Running workload carrying the + /// requested image/gpus, with the tunnel outcome in its first log line + /// (ready, or skipped with the reason — no silent skip). + /// Test-only serial lock: no deadlock (holders never wait on each other + /// while holding), but it parks a worker thread, hence the allow. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn s2_accept_creates_workload() { + let (_tmp, _home_guard, old_env) = lock_isolated_home(); + + let token = crate::identity::api_token(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + let server = tokio::spawn(async move { + axum::serve(listener, build_router()).await + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + + let r_create = client + .post(format!("{base}/requests")) + .bearer_auth(&token) + .json(&serde_json::json!({ + "requester": "kuro:s2", "rtype": "gpu", + "amount": 16, "max_price": 1.0, "hours": 2, + "image": "ubuntu:22.04", "gpus": 1, + })) + .send() + .await + .expect("create"); + let created: serde_json::Value = r_create.json().await.expect("create body"); + let req_id = created.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string(); + + let market = Market::open().expect("open market in isolated HOME"); + market.faucet("kuro:s2", 100.0).expect("faucet"); + market.add_offer(&new_offer("provider:s2")).expect("offer"); + + let r_matches = client + .get(format!("{base}/requests/{req_id}/matches")) + .bearer_auth(&token) + .send() + .await + .expect("matches"); + let matches: serde_json::Value = r_matches.json().await.expect("matches body"); + let match_id = matches.get(0).and_then(|m| m.get("id")).and_then(|v| v.as_str()).unwrap_or("").to_string(); + + let r_accept = client + .post(format!("{base}/matches/{match_id}/accept")) + .bearer_auth(&token) + .send() + .await + .expect("accept"); + let settled: serde_json::Value = r_accept.json().await.expect("accept body"); + let workload_id = settled.get("workload_id").and_then(|v| v.as_str()).unwrap_or("").to_string(); + + let r_workloads = client + .get(format!("{base}/workloads")) + .bearer_auth(&token) + .send() + .await + .expect("workloads"); + let workloads: serde_json::Value = r_workloads.json().await.expect("workloads body"); + let r_logs = client + .get(format!("{base}/workloads/{workload_id}/logs")) + .bearer_auth(&token) + .send() + .await + .expect("logs"); + let logs: serde_json::Value = r_logs.json().await.expect("logs body"); + + server.abort(); + restore_home(old_env); + + assert!(!req_id.is_empty(), "201 must carry the request id"); + assert!(!match_id.is_empty(), "matches must list the test offer"); + assert!(!workload_id.is_empty(), "accept must carry the workload id: {settled}"); + assert_eq!(settled.get("workload_status").and_then(|v| v.as_str()), Some("running")); + let mine = workloads.as_array().expect("workloads array").iter() + .find(|w| w.get("id").and_then(|v| v.as_str()) == Some(workload_id.as_str())) + .expect("workload listed"); + assert_eq!(mine.get("status").and_then(|v| v.as_str()), Some("Running")); + assert_eq!(mine.get("docker_image").and_then(|v| v.as_str()), Some("ubuntu:22.04")); + assert_eq!(mine.get("gpu_count").and_then(|v| v.as_u64()), Some(1)); + assert!(mine.get("name").and_then(|v| v.as_str()).unwrap_or("").contains(&req_id)); + let lines = logs.get("logs").and_then(|v| v.as_array()).expect("logs array"); + assert!(!lines.is_empty(), "workload must log the tunnel outcome"); + assert!(lines[0].as_str().unwrap_or("").contains("tunnel"), "first log explains the tunnel: {lines:?}"); + } + + /// S3 proof over real HTTP with the default quota (2 active jobs): + /// two accepts run, the third is rejected 429 with a clear message, + /// cancel withdraws the request, and /status reports the scheduler. + /// Test-only serial lock: no deadlock (holders never wait on each other + /// while holding), but it parks a worker thread, hence the allow. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn s3_quota_cancel_and_status() { + let (_tmp, _home_guard, old_env) = lock_isolated_home(); + + let token = crate::identity::api_token(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + let server = tokio::spawn(async move { + axum::serve(listener, build_router()).await + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + let authed = |b: reqwest::RequestBuilder| b.bearer_auth(&token); + + let mut req_ids = Vec::new(); + for _ in 0..3 { + let created: serde_json::Value = authed(client + .post(format!("{base}/requests")) + .json(&serde_json::json!({ + "requester": "kuro:s3", "rtype": "gpu", + "amount": 8, "max_price": 2.0, "hours": 1, + }))) + .send().await.expect("create").json().await.expect("create body"); + req_ids.push(created.get("id").and_then(|v| v.as_str()).unwrap_or("").to_string()); + } + + let market = Market::open().expect("open market in isolated HOME"); + market.faucet("kuro:s3", 1000.0).expect("faucet"); + market.add_offer(&new_offer("provider:s3")).expect("offer"); + market.add_offer(&new_offer("provider:s3b")).expect("offer"); + market.add_offer(&new_offer("provider:s3c")).expect("offer"); + + let mut statuses = Vec::new(); + for rid in &req_ids { + let matches: serde_json::Value = authed(client + .get(format!("{base}/requests/{rid}/matches"))) + .send().await.expect("matches").json().await.expect("matches body"); + let mid = matches.get(0).and_then(|m| m.get("id")).and_then(|v| v.as_str()).unwrap_or(""); + let resp = authed(client + .post(format!("{base}/matches/{mid}/accept"))) + .send().await.expect("accept"); + statuses.push(resp.status().as_u16()); + } + + // Third request: cancel it, then every settle path must refuse. + let r_cancel = authed(client + .post(format!("{base}/requests/{}/cancel", req_ids[2]))) + .send().await.expect("cancel"); + let cancelled: serde_json::Value = r_cancel.json().await.expect("cancel body"); + let r_get = authed(client + .get(format!("{base}/requests/{}", req_ids[2]))) + .send().await.expect("get cancelled"); + let got: serde_json::Value = r_get.json().await.expect("get body"); + let r_matches_after = authed(client + .get(format!("{base}/requests/{}/matches", req_ids[2]))) + .send().await.expect("matches after cancel"); + let r_status = authed(client + .get(format!("{base}/status"))) + .send().await.expect("status"); + let status: serde_json::Value = r_status.json().await.expect("status body"); + + server.abort(); + restore_home(old_env); + + assert!(req_ids.iter().all(|id| !id.is_empty()), "201 must carry ids"); + assert_eq!(statuses, vec![200, 200, 429], "third concurrent job hits the quota: {statuses:?}"); + assert_eq!(cancelled.get("status").and_then(|v| v.as_str()), Some("cancelled")); + assert_eq!(got.get("status").and_then(|v| v.as_str()), Some("cancelled")); + assert_eq!(r_matches_after.status().as_u16(), 409, "cancelled requests match nothing"); + assert_eq!(status.get("workloads_running").and_then(|v| v.as_u64()), Some(2)); + assert_eq!(status.get("quota_jobs_per_requester").and_then(|v| v.as_u64()), Some(2)); + } + + /// Q5 proof: template=jupyter resolves server-side (image+gpus recorded), + /// explicit fields win, unknown names fail 400 with the known list. + #[allow(clippy::await_holding_lock)] + #[tokio::test] + async fn q5_template_resolves_over_http() { + let (_tmp, _home_guard, old_env) = lock_isolated_home(); + let token = crate::identity::api_token(); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind ephemeral port"); + let addr = listener.local_addr().expect("local addr"); + let server = tokio::spawn(async move { + axum::serve(listener, build_router()).await + }); + let base = format!("http://{addr}"); + let client = reqwest::Client::new(); + let authed = |b: reqwest::RequestBuilder| b.bearer_auth(&token); + + let r_tpl: serde_json::Value = authed(client + .post(format!("{base}/requests")) + .json(&serde_json::json!({ + "requester": "kuro:q5", "rtype": "gpu", + "amount": 8, "max_price": 2.0, "hours": 1, + "template": "jupyter", + }))) + .send().await.expect("create template").json().await.expect("body"); + let r_over: serde_json::Value = authed(client + .post(format!("{base}/requests")) + .json(&serde_json::json!({ + "requester": "kuro:q5", "rtype": "gpu", + "amount": 8, "max_price": 2.0, "hours": 1, + "template": "jupyter", "image": "custom:9", "gpus": 4, + }))) + .send().await.expect("create override").json().await.expect("body"); + let r_bad = authed(client + .post(format!("{base}/requests")) + .json(&serde_json::json!({ + "requester": "kuro:q5", "rtype": "gpu", + "amount": 8, "max_price": 2.0, "hours": 1, + "template": "nope", + }))) + .send().await.expect("create bad"); + + server.abort(); + restore_home(old_env); + + assert_eq!(r_tpl.get("image").and_then(|v| v.as_str()), Some("jupyter/pytorch-notebook:latest")); + assert_eq!(r_tpl.get("gpus").and_then(|v| v.as_u64()), Some(1)); + assert_eq!(r_over.get("image").and_then(|v| v.as_str()), Some("custom:9")); + assert_eq!(r_over.get("gpus").and_then(|v| v.as_u64()), Some(4)); + assert_eq!(r_bad.status().as_u16(), 400); + } } diff --git a/helium-mesh/src/identity.rs b/helium-mesh/src/identity.rs index 928dd14..b4decd1 100644 --- a/helium-mesh/src/identity.rs +++ b/helium-mesh/src/identity.rs @@ -11,12 +11,25 @@ const IDENTITY_FILE: &str = "identity.json"; const CONFIG_FILE: &str = "config.toml"; const TOKEN_FILE: &str = "api_token"; +/// Node state directory. HELIUM_HOME (absolute path) overrides it wholesale: +/// tests, portable nodes, multi-node on one machine. Unset = ~/.helium. +pub fn helium_dir() -> Result { + if let Ok(dir) = std::env::var("HELIUM_HOME") { + if !dir.trim().is_empty() { + return Ok(PathBuf::from(dir)); + } + } + let home = dirs::home_dir() + .context("Failed to determine home directory")?; + Ok(home.join(HELIUM_DIR)) +} + /// Read (or generate) the local API token. Missing file on old nodes = /// fail-open with empty token (see `authorized`); new nodes get one at init. pub fn api_token() -> String { - let path = match dirs::home_dir() { - Some(h) => h.join(HELIUM_DIR).join(TOKEN_FILE), - None => return String::new(), + let path = match helium_dir() { + Ok(d) => d.join(TOKEN_FILE), + Err(_) => return String::new(), }; if let Ok(t) = std::fs::read_to_string(&path) { let t = t.trim().to_string(); @@ -233,9 +246,7 @@ impl IdentityManager { /// Get helium directory path fn helium_dir() -> Result { - let home = dirs::home_dir() - .context("Failed to determine home directory")?; - Ok(home.join(HELIUM_DIR)) + helium_dir() } /// Sign a message diff --git a/helium-mesh/src/lib.rs b/helium-mesh/src/lib.rs index a820118..1e2ab89 100644 --- a/helium-mesh/src/lib.rs +++ b/helium-mesh/src/lib.rs @@ -1,3 +1,4 @@ +pub mod bench; pub mod daemon; pub mod discovery; pub mod identity; @@ -5,6 +6,7 @@ pub mod market; pub mod mesh; pub mod networking; pub mod storage; +pub mod templates; pub mod tui; pub mod workload; diff --git a/helium-mesh/src/main.rs b/helium-mesh/src/main.rs index 0461d1a..a1be6da 100644 --- a/helium-mesh/src/main.rs +++ b/helium-mesh/src/main.rs @@ -1,7 +1,8 @@ use anyhow::Result; use clap::{Parser, Subcommand}; -use tracing::{info, warn}; +use tracing::info; +mod bench; mod daemon; mod discovery; mod identity; @@ -9,6 +10,7 @@ mod market; mod mesh; mod networking; mod storage; +mod templates; mod tui; mod workload; @@ -67,6 +69,16 @@ enum Commands { command: MarketCommands, }, + /// Benchmark this node (sustained score, never the peak) + Bench { + /// Wall time in seconds (300 recommended for official scores) + #[arg(short, long, default_value_t = 60)] + seconds: u64, + /// Machine-readable JSON output + #[arg(long)] + json: bool, + }, + /// Show credits balance Credits { #[arg(short, long)] @@ -97,6 +109,9 @@ enum Commands { /// Show this node's API token (share it with peers you trust) Token, + /// Self-contained demo: faucet, offers, request, match, settle (in-memory, nothing saved) + Demo, + /// Remote disk: register and inspect shared datasets Storage { #[command(subcommand)] @@ -129,12 +144,21 @@ enum MeshCommands { enum WorkloadCommands { /// Run a workload on available GPU Run { - #[arg(short, long)] + #[arg(short, long, default_value = "")] image: String, - #[arg(short, long)] + #[arg(short, long, default_value_t = 0)] gpu_count: u32, + /// Named template (jupyter, train, batch-scan); explicit flags win + #[arg(short, long, default_value = "")] + template: String, + /// Job name (default: job-) + #[arg(short, long)] + name: Option, }, + /// List available job templates + Templates, + /// List running workloads List, @@ -161,6 +185,12 @@ enum MarketCommands { endpoint: String, #[arg(short, long, default_value = "")] wg_pubkey: String, + /// Run `helium bench` inline and attach the sustained score + #[arg(long)] + bench: bool, + /// Bench wall time in seconds (with --bench) + #[arg(long, default_value_t = 60)] + bench_seconds: u64, }, /// List open offers @@ -179,12 +209,25 @@ enum MarketCommands { amount: u32, #[arg(short, long, default_value_t = 1.0)] max_price: f64, - #[arg(short, long, default_value_t = 2)] + /// -h est reserve a --help (clap) : pas de short ici. + #[arg(long, default_value_t = 2)] hours: u32, #[arg(short, long, default_value = "")] wg_pubkey: String, #[arg(short, long, default_value = "")] endpoint: String, + /// Container image wanted ("" = node default) + #[arg(short, long, default_value = "")] + image: String, + /// GPUs wanted (0 = unset) + #[arg(short, long, default_value_t = 0)] + gpus: u32, + /// Named template (jupyter, train, batch-scan); explicit flags win + #[arg(short, long, default_value = "")] + template: String, + /// Bench floor: offers below this sustained GFLOPS never match + #[arg(long, default_value_t = 0.0)] + min_bench: f64, }, /// Find matches for a request @@ -198,6 +241,12 @@ enum MarketCommands { #[arg(short, long)] match_id: String, }, + + /// Withdraw an open request (no credits moved) + Cancel { + #[arg(short, long)] + request_id: String, + }, } #[derive(Subcommand)] @@ -352,6 +401,18 @@ async fn main() -> Result<()> { } }, + Commands::Bench { seconds, json } => { + let report = bench::run_bench(seconds); + if json { + println!("{}", serde_json::to_string_pretty(&report)?); + } else { + println!("sustained: {:.1} GFLOPS (peak {:.1}, derate {:.0}%, {}s)", + report.gflops_sustained, report.gflops_peak, report.derate_pct, report.seconds); + println!("memory: {:.1} GB/s | RAM: {:.1}/{:.1} GB free | class: {}", + report.mem_bw_gbs, report.ram_avail_gb, report.ram_total_gb, report.suggested_class); + } + } + Commands::Daemon { bind } => { info!("Starting Helium daemon on {}...", bind); let daemon = HeliumDaemon::new(bind).await?; @@ -377,20 +438,60 @@ async fn main() -> Result<()> { println!(" DHT peers (Kademlia): {}", kad.len()); } - Commands::Workload { .. } => { - warn!("Workload commands not yet implemented"); - println!("Workload management coming in Phase 2"); - } + Commands::Workload { command } => match command { + WorkloadCommands::Run { image, gpu_count, template, name } => { + let (image, gpu_count) = templates::apply(&template, &image, gpu_count) + .map_err(|e| anyhow::anyhow!("{e}"))?; + let name = name.unwrap_or_else(|| format!("job-{}", &uuid::Uuid::new_v4().to_string()[..8])); + let w = workload::WorkloadManager::new().await? + .create_workload(name, image, gpu_count, "", "").await?; + println!("Workload created: {} ({}, {} gpu) [{}]", w.id, w.name, w.gpu_count, w.status); + } + WorkloadCommands::Templates => { + for t in templates::list() { + println!("{}: {} [image={} gpus={}]", t.name, t.description, t.docker_image, t.gpus); + } + } + WorkloadCommands::List => { + let rows = workload::WorkloadManager::new().await?.list_workloads().await; + if rows.is_empty() { + println!("No workloads"); + } + for w in rows { + println!("{}: {} [{}] image={} gpus={}", w.id, w.name, w.status, w.docker_image, w.gpu_count); + } + } + WorkloadCommands::Stop { id } => { + workload::WorkloadManager::new().await?.stop_workload(&id).await?; + println!("Stopped workload {id}"); + } + }, Commands::Market { command } => { let market = Market::open()?; match command { - MarketCommands::Offer { by, rtype, amount, price, endpoint, wg_pubkey } => { + MarketCommands::Offer { by, rtype, amount, price, endpoint, wg_pubkey, bench, bench_seconds } => { let by = match by { Some(p) => p, None => IdentityManager::load().await?.node_id()?, }; - let id = market.add_offer(&by, &rtype, amount, price, &endpoint, &wg_pubkey)?; + let bench_gflops = if bench { + let report = bench::run_bench(bench_seconds); + println!("bench: sustained {:.1} GFLOPS (peak {:.1}, derate {:.0}%, {}s)", + report.gflops_sustained, report.gflops_peak, report.derate_pct, report.seconds); + report.gflops_sustained + } else { + 0.0 + }; + let id = market.add_offer(&market::NewOffer { + provider: by.clone(), + rtype: rtype.clone(), + amount, + price, + endpoint, + wg_pubkey, + bench_gflops, + })?; println!("Offer created: {id} ({rtype} {amount} @ {price}/h by {by})"); } MarketCommands::List => { @@ -398,9 +499,9 @@ async fn main() -> Result<()> { if offers.is_empty() { println!("No open offers"); } - for o in offers { - println!("{}: {} {} @ {:.2}/h by {} [{}]", o.id, o.rtype, o.amount, o.price_per_hour, o.provider, o.status); - } + for o in offers { + println!("{}: {} {} @ {:.2}/h by {} bench={:.1} [{}]", o.id, o.rtype, o.amount, o.price_per_hour, o.provider, o.bench_gflops, o.status); + } } MarketCommands::Requests => { let requests = market.list_requests()?; @@ -411,12 +512,28 @@ async fn main() -> Result<()> { println!("{}: {} {} max {:.2}/h x {}h by {} [{}]", r.id, r.rtype, r.amount, r.max_price, r.hours, r.requester, r.status); } } - MarketCommands::Request { by, rtype, amount, max_price, hours, wg_pubkey, endpoint } => { + MarketCommands::Request { by, rtype, amount, max_price, hours, wg_pubkey, endpoint, image, gpus, template, min_bench } => { let by = match by { Some(p) => p, None => IdentityManager::load().await?.node_id()?, }; - let id = market.add_request(&by, &rtype, amount, max_price, hours, &wg_pubkey, &endpoint)?; + if !(min_bench >= 0.0) || min_bench > 1e6 { + anyhow::bail!("--min-bench must be between 0 and 1000000"); + } + let (image, gpus) = templates::apply(&template, &image, gpus) + .map_err(|e| anyhow::anyhow!("{e}"))?; + let id = market.add_request(&market::NewRequest { + requester: by.clone(), + rtype: rtype.clone(), + amount, + max_price, + hours, + wg_pubkey, + endpoint, + image, + gpus, + min_bench, + })?; println!("Request created: {id} ({rtype} {amount}, max {max_price}/h x {hours}h by {by})"); } MarketCommands::Match { request_id } => { @@ -429,8 +546,15 @@ async fn main() -> Result<()> { } } MarketCommands::Accept { match_id } => { - let (from, to, cost) = market.accept(&match_id)?; - println!("Settled: {from} -> {to} : {cost:.2} credits"); + let me = IdentityManager::load().await.ok() + .and_then(|idm| idm.node_id().ok()).unwrap_or_default(); + let report = crate::daemon::settle_match(&match_id, &me).await?; + println!("Settled: {} -> {} : {:.2} credits (workload {})", + report.from, report.to, report.cost, report.workload_id); + } + MarketCommands::Cancel { request_id } => { + let status = market.cancel_request(&request_id)?; + println!("Request {request_id}: {status}"); } } } @@ -533,6 +657,50 @@ async fn main() -> Result<()> { println!("{}", crate::identity::api_token()); } + Commands::Demo => { + let market = Market::open_memory()?; + println!("Helium demo — full market loop in memory (nothing saved):"); + market.faucet("demo-alice", 100.0)?; + println!(" faucet: demo-alice +100.00"); + let o1 = market.add_offer(&market::NewOffer { + provider: "demo-bob".to_string(), rtype: "gpu".to_string(), + amount: 64, price: 0.9, endpoint: String::new(), + wg_pubkey: String::new(), bench_gflops: 0.0, + })?; + let o2 = market.add_offer(&market::NewOffer { + provider: "demo-carol".to_string(), rtype: "gpu".to_string(), + amount: 16, price: 0.2, endpoint: String::new(), + wg_pubkey: String::new(), bench_gflops: 0.0, + })?; + println!(" offer: {o1} (gpu 64 @ 0.90) + {o2} (gpu 16 @ 0.20)"); + let r = market.add_request(&market::NewRequest { + requester: "demo-alice".to_string(), + rtype: "gpu".to_string(), + amount: 16, + max_price: 1.0, + hours: 1, + wg_pubkey: String::new(), + endpoint: String::new(), + image: String::new(), + gpus: 0, + min_bench: 0.0, + })?; + println!(" request: {r} (gpu 16, max 1.00/h)"); + let matches = market.find_matches(&r)?; + for m in &matches { + println!(" match: {} score {:.2} cost {:.2}", m.id, m.score, m.est_cost); + } + let (from, to, cost) = market.accept(&matches[0].id)?; + println!(" settled: {from} -> {to} : {cost:.2} credits"); + println!( + " balances: alice {:.2}, bob {:.2}, carol {:.2}", + market.balance("demo-alice")?, + market.balance("demo-bob")?, + market.balance("demo-carol")? + ); + println!("Done. Try it for real: helium faucet && helium market offer ..."); + } + Commands::Storage { command } => { let store = StorageManager::new().await?; match command { diff --git a/helium-mesh/src/market.rs b/helium-mesh/src/market.rs index 602251d..6e95dbc 100644 --- a/helium-mesh/src/market.rs +++ b/helium-mesh/src/market.rs @@ -12,7 +12,7 @@ use rusqlite::{params, Connection}; const DB_FILE: &str = "market.db"; -#[derive(Debug, Clone)] +#[derive(Debug, Clone, serde::Serialize)] pub struct Offer { pub id: String, pub provider: String, @@ -21,10 +21,25 @@ pub struct Offer { pub price_per_hour: f64, pub endpoint: String, pub wg_pubkey: String, + /// Sustained bench score (GFLOPS, `helium bench`). 0 = unmeasured. + #[serde(default)] + pub bench_gflops: f64, pub status: String, } +/// Groups the 7 offer fields so add_offer stays a 2-arg call. #[derive(Debug, Clone)] +pub struct NewOffer { + pub provider: String, + pub rtype: String, + pub amount: u32, + pub price: f64, + pub endpoint: String, + pub wg_pubkey: String, + pub bench_gflops: f64, +} + +#[derive(Debug, Clone, serde::Serialize)] pub struct Request { pub id: String, pub requester: String, @@ -34,10 +49,32 @@ pub struct Request { pub hours: u32, pub wg_pubkey: String, pub endpoint: String, + /// S2: container image wanted for the job ("" = node default, see Q5 templates). + pub image: String, + /// S2: GPUs wanted (0 = unset, node decides). + pub gpus: u32, + /// Bench floor: offers below this sustained GFLOPS never match (0 = any). + #[serde(default)] + pub min_bench: f64, pub status: String, } +/// S4: groups the 9 request fields so add_request stays a 2-arg call. #[derive(Debug, Clone)] +pub struct NewRequest { + pub requester: String, + pub rtype: String, + pub amount: u32, + pub max_price: f64, + pub hours: u32, + pub wg_pubkey: String, + pub endpoint: String, + pub image: String, + pub gpus: u32, + pub min_bench: f64, +} + +#[derive(Debug, Clone, serde::Serialize)] pub struct MatchRow { pub id: String, pub offer_id: String, @@ -52,9 +89,8 @@ pub struct Market { } fn helium_db_path() -> Result { - let home = dirs::home_dir().context("no home directory")?; - let dir = home.join(".helium"); - std::fs::create_dir_all(&dir).context("cannot create .helium dir")?; + let dir = crate::identity::helium_dir()?; + std::fs::create_dir_all(&dir).context("cannot create helium dir")?; Ok(dir.join(DB_FILE)) } @@ -63,12 +99,15 @@ CREATE TABLE IF NOT EXISTS offers ( id TEXT PRIMARY KEY, provider TEXT NOT NULL, rtype TEXT NOT NULL, amount INTEGER NOT NULL, price_per_hour REAL NOT NULL, endpoint TEXT NOT NULL DEFAULT '', wg_pubkey TEXT NOT NULL DEFAULT '', + bench_gflops REAL NOT NULL DEFAULT 0.0, status TEXT NOT NULL DEFAULT 'open', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS requests ( id TEXT PRIMARY KEY, requester TEXT NOT NULL, rtype TEXT NOT NULL, amount INTEGER NOT NULL, max_price REAL NOT NULL, hours INTEGER NOT NULL, wg_pubkey TEXT NOT NULL DEFAULT '', endpoint TEXT NOT NULL DEFAULT '', + image TEXT NOT NULL DEFAULT '', gpus INTEGER NOT NULL DEFAULT 0, + min_bench REAL NOT NULL DEFAULT 0.0, status TEXT NOT NULL DEFAULT 'open', created_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS matches ( @@ -93,6 +132,10 @@ fn migrate(conn: &Connection) -> Result<()> { "ALTER TABLE offers ADD COLUMN wg_pubkey TEXT NOT NULL DEFAULT ''", "ALTER TABLE requests ADD COLUMN wg_pubkey TEXT NOT NULL DEFAULT ''", "ALTER TABLE requests ADD COLUMN endpoint TEXT NOT NULL DEFAULT ''", + "ALTER TABLE requests ADD COLUMN image TEXT NOT NULL DEFAULT ''", + "ALTER TABLE requests ADD COLUMN gpus INTEGER NOT NULL DEFAULT 0", + "ALTER TABLE requests ADD COLUMN min_bench REAL NOT NULL DEFAULT 0.0", + "ALTER TABLE offers ADD COLUMN bench_gflops REAL NOT NULL DEFAULT 0.0", ] { match conn.execute_batch(stmt) { Ok(()) => {} @@ -115,13 +158,18 @@ fn match_score( req_type: &str, req_amount: u32, req_max_price: f64, + req_min_bench: f64, off_type: &str, off_amount: u32, off_price: f64, + off_bench: f64, ) -> Option { if req_type != off_type { return None; } + if off_bench < req_min_bench { + return None; + } if off_price > req_max_price { return None; } @@ -133,7 +181,7 @@ fn match_score( } let price_score = 1.0 - (off_price / req_max_price); let capacity_score = (off_amount as f64 / req_amount as f64).min(2.0) / 2.0; - Some(((price_score * 0.5) + (capacity_score * 0.5)).min(1.0).max(0.0)) + Some(((price_score * 0.5) + (capacity_score * 0.5)).clamp(0.0, 1.0)) } impl Market { @@ -145,35 +193,27 @@ impl Market { Ok(Self { conn }) } - #[cfg(test)] - fn open_memory() -> Result { + /// In-memory database: demos, tests, zero disk writes. + pub fn open_memory() -> Result { let conn = Connection::open_in_memory().context("cannot open memory db")?; conn.execute_batch(SCHEMA).context("cannot init schema")?; migrate(&conn)?; Ok(Self { conn }) } - pub fn add_offer( - &self, - provider: &str, - rtype: &str, - amount: u32, - price: f64, - endpoint: &str, - wg_pubkey: &str, - ) -> Result { + pub fn add_offer(&self, off: &NewOffer) -> Result { let id = format!("offer-{}", &uuid::Uuid::new_v4().to_string()[..8]); self.conn.execute( - "INSERT INTO offers (id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, status, created_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'open', ?8)", - params![id, provider, rtype, amount as i64, price, endpoint, wg_pubkey, now()], + "INSERT INTO offers (id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, bench_gflops, status, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'open', ?9)", + params![id, off.provider, off.rtype, off.amount as i64, off.price, off.endpoint, off.wg_pubkey, off.bench_gflops, now()], )?; Ok(id) } pub fn list_offers(&self) -> Result> { let mut stmt = self.conn.prepare( - "SELECT id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, status FROM offers + "SELECT id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, bench_gflops, status FROM offers WHERE status = 'open' ORDER BY price_per_hour ASC", )?; let rows = stmt.query_map([], |r| { @@ -185,7 +225,8 @@ impl Market { price_per_hour: r.get(4)?, endpoint: r.get(5)?, wg_pubkey: r.get(6)?, - status: r.get(7)?, + bench_gflops: r.get(7)?, + status: r.get(8)?, }) })?; Ok(rows.collect::, _>>()?) @@ -193,7 +234,7 @@ impl Market { pub fn get_offer(&self, id: &str) -> Result { self.conn.query_row( - "SELECT id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, status FROM offers WHERE id = ?1", + "SELECT id, provider, rtype, amount, price_per_hour, endpoint, wg_pubkey, bench_gflops, status FROM offers WHERE id = ?1", params![id], |r| { Ok(Offer { @@ -204,34 +245,26 @@ impl Market { price_per_hour: r.get(4)?, endpoint: r.get(5)?, wg_pubkey: r.get(6)?, - status: r.get(7)?, + bench_gflops: r.get(7)?, + status: r.get(8)?, }) }, ).with_context(|| format!("offer {id} not found")) } - pub fn add_request( - &self, - requester: &str, - rtype: &str, - amount: u32, - max_price: f64, - hours: u32, - wg_pubkey: &str, - endpoint: &str, - ) -> Result { + pub fn add_request(&self, req: &NewRequest) -> Result { let id = format!("req-{}", &uuid::Uuid::new_v4().to_string()[..8]); self.conn.execute( - "INSERT INTO requests (id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, status, created_at) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, 'open', ?9)", - params![id, requester, rtype, amount as i64, max_price, hours as i64, wg_pubkey, endpoint, now()], + "INSERT INTO requests (id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, image, gpus, min_bench, status, created_at) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, 'open', ?12)", + params![id, req.requester, req.rtype, req.amount as i64, req.max_price, req.hours as i64, req.wg_pubkey, req.endpoint, req.image, req.gpus as i64, req.min_bench, now()], )?; Ok(id) } pub fn list_requests(&self) -> Result> { let mut stmt = self.conn.prepare( - "SELECT id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, status FROM requests + "SELECT id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, image, gpus, min_bench, status FROM requests WHERE status = 'open' ORDER BY created_at ASC", )?; let rows = stmt.query_map([], |r| { @@ -244,7 +277,10 @@ impl Market { hours: r.get::<_, i64>(5)? as u32, wg_pubkey: r.get(6)?, endpoint: r.get(7)?, - status: r.get(8)?, + image: r.get(8)?, + gpus: r.get::<_, i64>(9)? as u32, + min_bench: r.get(10)?, + status: r.get(11)?, }) })?; Ok(rows.collect::, _>>()?) @@ -252,7 +288,7 @@ impl Market { pub fn get_request(&self, id: &str) -> Result { self.conn.query_row( - "SELECT id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, status FROM requests WHERE id = ?1", + "SELECT id, requester, rtype, amount, max_price, hours, wg_pubkey, endpoint, image, gpus, min_bench, status FROM requests WHERE id = ?1", params![id], |r| { Ok(Request { @@ -264,7 +300,10 @@ impl Market { hours: r.get::<_, i64>(5)? as u32, wg_pubkey: r.get(6)?, endpoint: r.get(7)?, - status: r.get(8)?, + image: r.get(8)?, + gpus: r.get::<_, i64>(9)? as u32, + min_bench: r.get(10)?, + status: r.get(11)?, }) }, ).with_context(|| format!("request {id} not found")) @@ -282,9 +321,11 @@ impl Market { &req.rtype, req.amount, req.max_price, + req.min_bench, &o.rtype, o.amount, o.price_per_hour, + o.bench_gflops, ) { let est_cost = score * o.price_per_hour * req.hours as f64; let id = format!("match-{}", &uuid::Uuid::new_v4().to_string()[..8]); @@ -307,6 +348,23 @@ impl Market { Ok(out) } + pub fn get_match(&self, id: &str) -> Result { + self.conn.query_row( + "SELECT id, offer_id, request_id, score, est_cost, status FROM matches WHERE id = ?1", + params![id], + |r| { + Ok(MatchRow { + id: r.get(0)?, + offer_id: r.get(1)?, + request_id: r.get(2)?, + score: r.get(3)?, + est_cost: r.get(4)?, + status: r.get(5)?, + }) + }, + ).with_context(|| format!("match {id} not found")) + } + /// Accept a proposed match: closes offer+request, settles credits. pub fn accept(&self, match_id: &str) -> Result<(String, String, f64)> { let (offer_id, request_id, est_cost, status): (String, String, f64, String) = @@ -342,6 +400,44 @@ impl Market { Ok((request.requester, offer.provider, est_cost)) } + /// S3: withdraw an open request. No credits moved (only accept settles). + pub fn cancel_request(&self, id: &str) -> Result { + let req = self.get_request(id)?; + if req.status != "open" { + anyhow::bail!("request {id} is not open (status: {})", req.status); + } + self.conn.execute( + "UPDATE requests SET status = 'cancelled' WHERE id = ?1", + params![id], + )?; + Ok("cancelled".to_string()) + } + + /// S3: expire open requests older than `ttl_hours`. Returns evicted count. + pub fn expire_stale(&self, ttl_hours: u32) -> Result { + let cutoff = chrono::Utc::now() - chrono::Duration::hours(ttl_hours as i64); + let mut stmt = self.conn.prepare( + "SELECT id, created_at FROM requests WHERE status = 'open'", + )?; + let rows: Vec<(String, String)> = stmt + .query_map([], |r| Ok((r.get(0)?, r.get(1)?)))? + .collect::, _>>()?; + let mut evicted = 0; + for (id, created_at) in rows { + let stale = chrono::DateTime::parse_from_rfc3339(&created_at) + .map(|dt| dt.with_timezone(&chrono::Utc) < cutoff) + .unwrap_or(false); + if stale { + self.conn.execute( + "UPDATE requests SET status = 'expired' WHERE id = ?1", + params![id], + )?; + evicted += 1; + } + } + Ok(evicted) + } + /// Allocate the next free 10.0.0.x tunnel IP for a peer (provider side). pub fn alloc_ip(&self, peer_pubkey: &str, match_id: &str) -> Result { let mut stmt = self.conn.prepare("SELECT ip FROM allocs")?; @@ -350,7 +446,7 @@ impl Market { .collect::, _>>()?; for last in 3..=254u8 { let ip = format!("10.0.0.{last}"); - if !used.iter().any(|u| *u == ip) { + if !used.contains(&ip) { self.conn.execute( "INSERT INTO allocs (ip, peer_pubkey, match_id, created_at) VALUES (?1, ?2, ?3, ?4)", params![ip, peer_pubkey, match_id, now()], @@ -399,19 +495,39 @@ mod tests { #[test] fn score_rejects_mismatch() { - assert!(match_score("gpu", 16, 1.0, "ram", 32, 0.5).is_none()); - assert!(match_score("gpu", 16, 1.0, "gpu", 32, 2.0).is_none()); - assert!(match_score("gpu", 64, 1.0, "gpu", 32, 0.5).is_none()); - let s = match_score("gpu", 16, 1.0, "gpu", 32, 0.5).unwrap(); + assert!(match_score("gpu", 16, 1.0, 0.0, "ram", 32, 0.5, 0.0).is_none()); + assert!(match_score("gpu", 16, 1.0, 0.0, "gpu", 32, 2.0, 0.0).is_none()); + assert!(match_score("gpu", 64, 1.0, 0.0, "gpu", 32, 0.5, 0.0).is_none()); + let s = match_score("gpu", 16, 1.0, 0.0, "gpu", 32, 0.5, 0.0).unwrap(); assert!((0.0..=1.0).contains(&s)); } + #[test] + fn score_enforces_bench_floor() { + // Unbenched offer (0.0) never matches a request demanding a floor. + assert!(match_score("gpu", 16, 1.0, 10.0, "gpu", 32, 0.5, 0.0).is_none()); + assert!(match_score("gpu", 16, 1.0, 10.0, "gpu", 32, 0.5, 9.9).is_none()); + assert!(match_score("gpu", 16, 1.0, 10.0, "gpu", 32, 0.5, 10.0).is_some()); + } + + fn new_offer(provider: &str, endpoint: &str, wg_pubkey: &str) -> NewOffer { + NewOffer { + provider: provider.to_string(), + rtype: "gpu".to_string(), + amount: 32, + price: 0.5, + endpoint: endpoint.to_string(), + wg_pubkey: wg_pubkey.to_string(), + bench_gflops: 0.0, + } + } + #[test] fn migrate_is_idempotent() -> Result<()> { let m = Market::open_memory()?; migrate(&m.conn)?; migrate(&m.conn)?; - m.add_offer("bob", "gpu", 32, 0.5, "1.2.3.4:51820", "PUB")?; + m.add_offer(&new_offer("bob", "1.2.3.4:51820", "PUB"))?; Ok(()) } @@ -419,8 +535,22 @@ mod tests { fn full_flow_in_memory() -> Result<()> { let m = Market::open_memory()?; m.faucet("alice", 100.0)?; - let oid = m.add_offer("bob", "gpu", 32, 0.5, "1.2.3.4:51820", "PUBBOB")?; - let rid = m.add_request("alice", "gpu", 16, 1.0, 2, "PUBALICE", "")?; + let oid = m.add_offer(&new_offer("bob", "1.2.3.4:51820", "PUBBOB"))?; + let rid = m.add_request(&NewRequest { + requester: "alice".to_string(), + rtype: "gpu".to_string(), + amount: 16, + max_price: 1.0, + hours: 2, + wg_pubkey: "PUBALICE".to_string(), + endpoint: String::new(), + image: "ubuntu:22.04".to_string(), + gpus: 1, + min_bench: 0.0, + })?; + let req = m.get_request(&rid)?; + assert_eq!(req.image, "ubuntu:22.04"); + assert_eq!(req.gpus, 1); let offer = m.get_offer(&oid)?; assert_eq!(offer.endpoint, "1.2.3.4:51820"); let matches = m.find_matches(&rid)?; @@ -437,4 +567,36 @@ mod tests { assert!(m.find_matches(&rid).is_err()); // request filled Ok(()) } + + #[test] + fn cancel_and_expiry() -> Result<()> { + let m = Market::open_memory()?; + let new_req = |requester: &str| NewRequest { + requester: requester.to_string(), + rtype: "gpu".to_string(), + amount: 16, + max_price: 1.0, + hours: 2, + wg_pubkey: String::new(), + endpoint: String::new(), + image: String::new(), + gpus: 0, + min_bench: 0.0, + }; + let rid = m.add_request(&new_req("alice"))?; + assert_eq!(m.cancel_request(&rid)?, "cancelled"); + assert_eq!(m.get_request(&rid)?.status, "cancelled"); + assert!(m.cancel_request(&rid).is_err()); // no longer open + assert!(m.find_matches(&rid).is_err()); // cannot match it anymore + assert!(m.cancel_request("req-missing").is_err()); // unknown id + let rid2 = m.add_request(&new_req("bob"))?; + m.conn.execute( + "UPDATE requests SET created_at = '2000-01-01T00:00:00+00:00' WHERE id = ?1", + params![rid2], + )?; + assert_eq!(m.expire_stale(24)?, 1); + assert_eq!(m.get_request(&rid2)?.status, "expired"); + assert_eq!(m.expire_stale(24)?, 0); // idempotent + Ok(()) + } } diff --git a/helium-mesh/src/storage.rs b/helium-mesh/src/storage.rs index 700e5bc..5d7ae87 100644 --- a/helium-mesh/src/storage.rs +++ b/helium-mesh/src/storage.rs @@ -318,9 +318,7 @@ impl StorageManager { } fn helium_dir() -> Result { - let home = dirs::home_dir() - .context("Failed to determine home directory")?; - Ok(home.join(".helium")) + crate::identity::helium_dir() } } diff --git a/helium-mesh/src/templates.rs b/helium-mesh/src/templates.rs new file mode 100644 index 0000000..49c20a6 --- /dev/null +++ b/helium-mesh/src/templates.rs @@ -0,0 +1,113 @@ +//! Q5: named job templates — one word to a ready-to-run job definition. +//! +//! A template only declares WHAT to run (image, GPUs, ports, command). +//! Pulling the image / booting it is the provider's job (Firecracker rootfs +//! today, `docker run` where Docker exists). Unknown names are rejected with +//! the known list so typos fail fast, locally and over HTTP. + +#[derive(Debug, Clone, PartialEq)] +pub struct JobTemplate { + pub name: &'static str, + pub description: &'static str, + pub docker_image: &'static str, + pub gpus: u32, + pub ports: &'static [u16], + pub command: &'static str, +} + +const JUPYTER_PORTS: &[u16] = &[8888]; + +/// The whole registry. Keep it short: every entry is a promise to users. +pub fn list() -> Vec { + vec![ + JobTemplate { + name: "jupyter", + description: "JupyterLab with PyTorch (port 8888, no token)", + docker_image: "jupyter/pytorch-notebook:latest", + gpus: 1, + ports: JUPYTER_PORTS, + command: "start-notebook.sh --NotebookApp.token= --NotebookApp.ip=0.0.0.0", + }, + JobTemplate { + name: "train", + description: "PyTorch training container (1 GPU, run your script)", + docker_image: "pytorch/pytorch:latest", + gpus: 1, + ports: &[], + command: "python train.py", + }, + JobTemplate { + name: "batch-scan", + description: "CPU batch job on Ubuntu (Kuro scans, ETL, benchmarks)", + docker_image: "ubuntu:22.04", + gpus: 0, + ports: &[], + command: "bash run.sh", + }, + ] +} + +/// Case-insensitive lookup. Empty input = no template (caller decides). +pub fn resolve(name: &str) -> Option { + let want = name.trim().to_lowercase(); + if want.is_empty() { + return None; + } + list().into_iter().find(|t| t.name == want) +} + +/// Known names for error messages and CLI help. +pub fn names() -> Vec<&'static str> { + list().iter().map(|t| t.name).collect() +} + +/// Merge explicit values over a template: explicit image/gpus win, +/// template fills the blanks. Unknown template name = Err with the list. +pub fn apply(template: &str, image: &str, gpus: u32) -> Result<(String, u32), String> { + let template = template.trim(); + if template.is_empty() { + return Ok((image.trim().to_string(), gpus)); + } + match resolve(template) { + None => Err(format!("unknown template '{}' (known: {})", template, names().join(", "))), + Some(t) => { + let img = if image.trim().is_empty() { + t.docker_image.to_string() + } else { + image.trim().to_string() + }; + Ok((img, if gpus == 0 { t.gpus } else { gpus })) + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn registry_lists_three() { + let all = list(); + assert_eq!(all.len(), 3); + assert!(all.iter().any(|t| t.name == "jupyter")); + } + + #[test] + fn resolve_is_forgiving() { + assert_eq!(resolve("Jupyter").unwrap().docker_image, "jupyter/pytorch-notebook:latest"); + assert_eq!(resolve(" train ").unwrap().gpus, 1); + assert!(resolve("").is_none()); + assert!(resolve("nope").is_none()); + } + + #[test] + fn apply_prefers_explicit() { + let (img, gpus) = apply("jupyter", "", 0).unwrap(); + assert_eq!((img.as_str(), gpus), ("jupyter/pytorch-notebook:latest", 1)); + let (img, gpus) = apply("jupyter", "custom:1", 4).unwrap(); + assert_eq!((img.as_str(), gpus), ("custom:1", 4)); + assert!(apply("nope", "", 0).is_err()); + let (img, gpus) = apply("", "ubuntu:22.04", 0).unwrap(); + assert_eq!((img.as_str(), gpus), ("ubuntu:22.04", 0)); + } +} diff --git a/helium-mesh/src/workload.rs b/helium-mesh/src/workload.rs index e4e5638..1b7b46a 100644 --- a/helium-mesh/src/workload.rs +++ b/helium-mesh/src/workload.rs @@ -18,6 +18,12 @@ pub struct Workload { pub docker_image: String, pub gpu_count: u32, pub target_node: String, + /// S3: originating market request ("" for manual `workload run` jobs). + #[serde(default)] + pub request_id: String, + /// S3: requester the job counts against for quotas ("" = unlinked). + #[serde(default)] + pub requester: String, pub created_at: chrono::DateTime, pub started_at: Option>, pub completed_at: Option>, @@ -53,7 +59,11 @@ pub struct WorkloadManager { impl WorkloadManager { pub async fn new() -> Result { let base_path = Self::helium_dir()?; - + // Q5 fix: first `workload run` on a fresh node failed writing — + // the dir was never created (Market does create_dir_all, we didn't). + tokio::fs::create_dir_all(&base_path) + .await + .context("Failed to create helium directory")?; let workloads = if let Ok(data) = tokio::fs::read_to_string(base_path.join(WORKLOADS_FILE)).await { serde_json::from_str(&data).unwrap_or_default() } else { @@ -66,7 +76,14 @@ impl WorkloadManager { }) } - pub async fn create_workload(&self, name: String, docker_image: String, gpu_count: u32) -> Result { + pub async fn create_workload( + &self, + name: String, + docker_image: String, + gpu_count: u32, + request_id: &str, + requester: &str, + ) -> Result { let id = Uuid::new_v4().to_string(); let workload = Workload { @@ -76,16 +93,24 @@ impl WorkloadManager { docker_image, gpu_count, target_node: String::new(), // Will be assigned by scheduler + request_id: request_id.to_string(), + requester: requester.to_string(), created_at: chrono::Utc::now(), started_at: None, completed_at: None, logs: vec![], }; - let mut workloads = self.workloads.write().await; - workloads.insert(id.clone(), workload.clone()); - - self.save().await?; + // Serialize under the write guard, write the file after releasing it: + // save-while-locked deadlocks (RwLock is not reentrant). + let data = { + let mut workloads = self.workloads.write().await; + workloads.insert(id.clone(), workload.clone()); + serde_json::to_string_pretty(&*workloads) + .context("Failed to serialize workloads")? + }; + tokio::fs::write(self.base_path.join(WORKLOADS_FILE), data).await + .context("Failed to write workloads file")?; info!("Created workload {} ({})", id, workload.name); @@ -102,32 +127,57 @@ impl WorkloadManager { workloads.get(id).cloned() } - pub async fn stop_workload(&self, id: &str) -> Result<()> { - let mut workloads = self.workloads.write().await; + /// S3: non-terminal (Pending/Running) jobs counting against a requester. + pub async fn count_active_for(&self, requester: &str) -> usize { + use WorkloadStatus::{Pending, Running}; + let workloads = self.workloads.read().await; + workloads + .values() + .filter(|w| w.requester == requester && matches!(w.status, Pending | Running)) + .count() + } - if let Some(workload) = workloads.get_mut(id) { - workload.status = WorkloadStatus::Stopped; - self.save().await?; - info!("Stopped workload {}", id); - } + pub async fn stop_workload(&self, id: &str) -> Result<()> { + let data = { + let mut workloads = self.workloads.write().await; + + if let Some(workload) = workloads.get_mut(id) { + workload.status = WorkloadStatus::Stopped; + } + serde_json::to_string_pretty(&*workloads) + .context("Failed to serialize workloads")? + }; + tokio::fs::write(self.base_path.join(WORKLOADS_FILE), data).await + .context("Failed to write workloads file")?; + info!("Stopped workload {}", id); Ok(()) } - async fn save(&self) -> Result<()> { - let workloads = self.workloads.read().await; - let data = serde_json::to_string_pretty(&*workloads) - .context("Failed to serialize workloads")?; - + /// S2: move a freshly matched workload to Running with a first log line. + pub async fn mark_running(&self, id: &str, log_line: String) -> Result<()> { + let data = { + let mut workloads = self.workloads.write().await; + + match workloads.get_mut(id) { + Some(workload) => { + workload.status = WorkloadStatus::Running; + workload.started_at = Some(chrono::Utc::now()); + workload.logs.push(log_line); + } + None => anyhow::bail!("workload {id} not found"), + } + serde_json::to_string_pretty(&*workloads) + .context("Failed to serialize workloads")? + }; tokio::fs::write(self.base_path.join(WORKLOADS_FILE), data).await .context("Failed to write workloads file")?; + info!("Workload {} running", id); Ok(()) } fn helium_dir() -> Result { - let home = dirs::home_dir() - .context("Failed to determine home directory")?; - Ok(home.join(".helium")) + crate::identity::helium_dir() } } diff --git a/helium/scripts/demo-live.sh b/helium/scripts/demo-live.sh new file mode 100644 index 0000000..03ef96e --- /dev/null +++ b/helium/scripts/demo-live.sh @@ -0,0 +1,13 @@ +#!/usr/bin/env bash +# demo-live.sh — vitrine Q5 + bench en 3 etapes (hermetique). +set -euo pipefail +export HELIUM_HOME="$(mktemp -d)/helium" +BIN="./helium-mesh/target/debug/helium.exe" +echo "=== 1. templates disponibles" +"$BIN" workload templates 2>/dev/null +echo "=== 2. bench rapide (10s)" +"$BIN" bench --seconds 10 2>/dev/null +echo "=== 3. job via template" +"$BIN" workload run --template batch-scan --name demo-live 2>/dev/null +"$BIN" workload list 2>/dev/null +echo "DEMO-OK" diff --git a/helium/scripts/demo-s3-cancel.sh b/helium/scripts/demo-s3-cancel.sh new file mode 100644 index 0000000..19418d1 --- /dev/null +++ b/helium/scripts/demo-s3-cancel.sh @@ -0,0 +1,29 @@ +#!/usr/bin/env bash +# demo-s3-cancel.sh — S3 : cancel une demande ouverte (aucun credit ne bouge). +# Usage (1 commande) : ./demo-s3-cancel.sh +# Hermetique : HELIUM_HOME temporaire, rien n est ecrit dans ~/.helium. +# Shell : Git-Bash ou PowerShell (l interop WSL ne transmet pas les env +# custom aux .exe Windows — sous WSL, HELIUM_HOME serait ignore). +# Requiert : binaire compile (cd ../helium-mesh && cargo build). +# Note : le timeout auto (TTL) tourne dans le daemon (match_loop, defaut +# HELIUM_REQUEST_TTL_HOURS=24) ; il est prouve par le test cancel_and_expiry. +set -euo pipefail + +DIR="$(cd "$(dirname "$0")/../../helium-mesh/target/debug" && pwd)" +BIN="$DIR/helium" +[ -x "$BIN" ] || BIN="$DIR/helium.exe" +[ -x "$BIN" ] || { echo "compile d'abord : (cd helium-mesh && cargo build)"; exit 1; } +export HELIUM_HOME="$(mktemp -d)/helium" +ME="demo-cancel" + +echo "[1/3] demande ouverte" +"$BIN" faucet --party "$ME" --amount 100 >/dev/null +REQ=$("$BIN" market request --by "$ME" --amount 8 --max-price 2 --hours 1 | awk '{print $3}') +echo " $REQ" + +echo "[2/3] cancel" +"$BIN" market cancel --request-id "$REQ" + +echo "[3/3] la demande annulee ne matche plus et ne s accepte plus" +"$BIN" market match --request-id "$REQ" || echo " (refus attendu : not open)" +echo "OK : cancel sans mouvement de credits." diff --git a/helium/scripts/demo-s3-quota.sh b/helium/scripts/demo-s3-quota.sh new file mode 100644 index 0000000..a6327ec --- /dev/null +++ b/helium/scripts/demo-s3-quota.sh @@ -0,0 +1,34 @@ +#!/usr/bin/env bash +# demo-s3-quota.sh — S3 : quotas scheduler (2 jobs OK, 3e refuse). +# Usage (1 commande) : ./demo-s3-quota.sh +# Hermetique : HELIUM_HOME temporaire, rien n est ecrit dans ~/.helium. +# Shell : Git-Bash ou PowerShell (l interop WSL ne transmet pas les env +# custom aux .exe Windows — sous WSL, HELIUM_HOME serait ignore). +# Requiert : binaire compile (cd ../helium-mesh && cargo build). +set -euo pipefail + +DIR="$(cd "$(dirname "$0")/../../helium-mesh/target/debug" && pwd)" +BIN="$DIR/helium" +[ -x "$BIN" ] || BIN="$DIR/helium.exe" +[ -x "$BIN" ] || { echo "compile d'abord : (cd helium-mesh && cargo build)"; exit 1; } +export HELIUM_HOME="$(mktemp -d)/helium" +ME="demo-quota" + +echo "[1/3] credits + offre provider" +"$BIN" faucet --party "$ME" --amount 1000 >/dev/null +"$BIN" market offer --by provider-demo --amount 64 --price 0.1 +"$BIN" market offer --by provider-demo2 --amount 64 --price 0.1 +"$BIN" market offer --by provider-demo3 --amount 64 --price 0.1 + +echo "[2/3] 3 demandes du meme projet, 3 accepts (quota defaut = 2)" +for i in 1 2 3; do + REQ=$("$BIN" market request --by "$ME" --amount 8 --max-price 2 --hours 1 | awk '{print $3}') + MATCH=$("$BIN" market match --request-id "$REQ" | head -1 | awk '{print $1}' | sed 's/:$//') + echo "--- accept $REQ via $MATCH" + "$BIN" market accept --match-id "$MATCH" || echo " (refus attendu au 3e : quota depasse, 429 via l API)" +done + +echo "[3/3] workloads crees" +"$BIN" workload list +echo "OK : 2 Running, 3e refuse avec message 'quota exceeded'." +echo "Knob : HELIUM_MAX_ACTIVE_JOBS_PER_REQUESTER=N (defaut 2)." diff --git a/helium/scripts/fc-vm.sh b/helium/scripts/fc-vm.sh index aec28f2..5c2e7a2 100755 --- a/helium/scripts/fc-vm.sh +++ b/helium/scripts/fc-vm.sh @@ -28,7 +28,12 @@ done TAP_IP="${GUEST_IP%.*}.1" SOCK=/tmp/fc-helium.socket +# Cle unique par VM si elle existe, sinon cle CI (1er boot seulement). +# La cle CI est PUBLIQUE (artefact Firecracker) : rotate_guest_key() la +# remplace des le 1er boot. Le fichier CI est garde pour re-bootstrap. +VM_KEY="$DIR/vm_key" KEY="$DIR/ubuntu-22.04.id_rsa" +[ -f "$VM_KEY" ] && KEY="$VM_KEY" SSH="ssh -i $KEY -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=10 root@$GUEST_IP" info() { echo "[fc-vm] $*"; } @@ -98,6 +103,7 @@ cmd_up() { wait_ssh # Route par defaut volatile (perdue a chaque boot) : le guest sort via tap. $SSH 'ip route replace default via 172.16.0.1 dev eth0 2>/dev/null || ip route replace default via 172.16.0.1' || warn "route guest non posee" + rotate_guest_key || warn "rotation cle guest echouee (cle CI gardee)" info "guest OK: $($SSH 'hostname; cat /proc/meminfo | head -1')" } @@ -109,8 +115,37 @@ cmd_down() { info "vm arretee" } -cmd_resize() { - # Grow the rootfs image (VM must be down). Ex: fc-vm.sh resize +1536M +rotate_guest_key() { + # Remplace la cle CI publique par une cle unique par VM. + # Regenere si absente OU cassee (une cle cassee n'a jamais pu servir). + local vmkey="$DIR/vm_key" newpub oldpub + local SSH_NEW="ssh -i $vmkey -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=10 root@$GUEST_IP" + if [ -f "$vmkey" ] && $SSH_NEW true 2>/dev/null; then + info "cle guest unique deja active" + return 0 + fi + # Canal casse (ex: vm_key generee mais jamais injectee) : repli cle CI. + if ! $SSH true 2>/dev/null; then + SSH="ssh -i $DIR/ubuntu-22.04.id_rsa -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null -o ConnectTimeout=10 root@$GUEST_IP" + $SSH true 2>/dev/null || { echo "[fc-vm][ERR] aucun acces SSH guest" >&2; return 1; } + info "repli cle CI pour bootstrap" + fi + rm -f "$vmkey" "$vmkey.pub" # ssh-keygen demanderait confirmation sinon + ssh-keygen -t ed25519 -N "" -f "$vmkey" -q + chmod 600 "$vmkey" + newpub="$(cat "$vmkey.pub")" + $SSH "grep -q -F '$newpub' ~/.ssh/authorized_keys 2>/dev/null || echo '$newpub' >> ~/.ssh/authorized_keys" + $SSH_NEW true 2>/dev/null || { echo "[fc-vm][ERR] nouvelle cle refusee, CI gardee" >&2; return 1; } + oldpub="$(ssh-keygen -y -f "$DIR/ubuntu-22.04.id_rsa")" + $SSH "sed -i '\|$oldpub|d' ~/.ssh/authorized_keys" + if $SSH true 2>/dev/null; then + warn "cle CI toujours acceptee (non bloquant)" + else + info "cle CI revoquee, cle unique active" + fi +} + +cmd_resize() { # Grow the rootfs image (VM must be down). Ex: fc-vm.sh resize +1536M # Inside the guest afterwards: resize2fs /dev/vda size="${1:?usage: fc-vm.sh resize +SIZE (ex: +1536M)}" if [ -S /tmp/fc-helium.socket ]; then @@ -142,5 +177,6 @@ case "$CMD" in ssh) $SSH "$@" ;; expose) cmd_expose "$@" ;; resize) cmd_resize "$@" ;; - *) echo "usage: $0 {up|down|ssh|expose|resize} [options]"; exit 1 ;; + rotate-key) rotate_guest_key ;; + *) echo "usage: $0 {up|down|ssh|expose|resize|rotate-key} [options]"; exit 1 ;; esac diff --git a/helium/scripts/helium-health.sh b/helium/scripts/helium-health.sh new file mode 100644 index 0000000..220ae6b --- /dev/null +++ b/helium/scripts/helium-health.sh @@ -0,0 +1,92 @@ +#!/usr/bin/env bash +# helium-health.sh — controle sante provider (tunnel, daemon, VM, guest, disque). +# Usage: sudo ./helium-health.sh [--json] +# Sortie: lignes OK/WARN/FAIL + code retour (0 ok, 1 echec, 2 avertissements). +set -uo pipefail + +JSON=0 +[ "${1:-}" = "--json" ] && JSON=1 +fails=0; warns=0 +results="" + +check() { # check + local name="$1"; shift + if out="$("$@" 2>&1)"; then + results="${results}OK $name\n" + else + results="${results}FAIL $name :: $(echo "$out" | head -1)\n" + fails=$((fails + 1)) + fi +} + +check_warn() { # check_warn + local name="$1"; shift + if out="$("$@" 2>&1)"; then + results="${results}OK $name\n" + else + results="${results}WARN $name :: $(echo "$out" | head -1)\n" + warns=$((warns + 1)) + fi +} + +SUDO="" +[ "$(id -u)" != "0" ] && SUDO="sudo -n " + +# 1. KVM + binaire +check "kvm-present" test -e /dev/kvm +check "helium-bin" command -v helium + +# 2. Tunnel WireGuard +check "wg-iface" ip link show helium-poc +if $SUDO wg show helium-poc latest-handshakes 2>/dev/null | grep -q .; then + results="${results}OK wg-handshake\n" +else + # handshake absent = peut-etre normal (pas de trafic recent) + results="${results}WARN wg-handshake :: aucun handshake (tunnel idle ?)\n" + warns=$((warns + 1)) +fi + +# 3. Services systemd +check "svc-wg" systemctl is-active --quiet wg-quick@helium-poc +check "svc-daemon" systemctl is-active --quiet helium-daemon +check_warn "svc-vm" systemctl is-active --quiet helium-vm + +# 4. MicroVM + guest +if [ -S /tmp/fc-helium.socket ]; then + results="${results}OK vm-socket\n" + GKEY=/srv/helium-vm/ubuntu-22.04.id_rsa + [ -f /srv/helium-vm/vm_key ] && GKEY=/srv/helium-vm/vm_key + if ssh -i "$GKEY" -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null \ + -o ConnectTimeout=8 root@172.16.0.2 true 2>/dev/null; then + results="${results}OK guest-ssh\n" + else + results="${results}FAIL guest-ssh :: injoignable\n" + fails=$((fails + 1)) + fi +else + results="${results}WARN vm-socket :: VM eteinte (normal si pas de match)\n" + warns=$((warns + 1)) +fi + +# 5. Disque + DB +root_free=$(df -m / | awk 'NR==2{print $4}') +if [ "${root_free:-0}" -lt 1024 ]; then + results="${results}FAIL disk :: ${root_free}Mo libres\n" + fails=$((fails + 1)) +else + results="${results}OK disk :: ${root_free}Mo libres\n" +fi +user_home=$(getent passwd "${SUDO_USER:-$(id -un)}" | cut -d: -f6) +[ -f "$user_home/.helium/market.db" ] && results="${results}OK market-db\n" || { + results="${results}WARN market-db :: absente\n" + warns=$((warns + 1)) +} + +if [ "$JSON" = "1" ]; then + printf '{"fails":%d,"warns":%d}\n' "$fails" "$warns" +else + printf "$results" +fi +[ "$fails" -eq 0 ] || exit 1 +[ "$warns" -eq 0 ] || exit 2 +exit 0 diff --git a/install.sh b/install.sh index f476fba..6099358 100755 --- a/install.sh +++ b/install.sh @@ -7,6 +7,7 @@ # sh install.sh --source # build depuis sources (cargo) # sh install.sh --dir ~/.local/bin # dossier d'install custom # sudo sh install.sh --provider --daemon # provider complet + service +# sudo sh install.sh --provider --daemon --vm # + microVM persistante # sh install.sh --borrower --endpoint IP:51820 --pubkey CLE set -euo pipefail @@ -25,6 +26,7 @@ INSTALL_DIR="" CHECK_ONLY=0 FROM_SOURCE=0 WITH_DAEMON=0 +WITH_VM=0 ROLE="" ENDPOINT="" PEER_PUBKEY="" @@ -37,6 +39,7 @@ for arg in "$@"; do --check-only) CHECK_ONLY=1 ;; --source) FROM_SOURCE=1 ;; --daemon) WITH_DAEMON=1 ;; + --vm) WITH_VM=1 ;; --provider) ROLE="provider" ;; --borrower) ROLE="borrower" ;; --endpoint|--pubkey|--endpoint=*|--pubkey=*) ;; @@ -380,6 +383,9 @@ main() { if [ "$WITH_DAEMON" = "1" ]; then install_daemon_service || warn "service daemon non installe" fi + if [ "$WITH_VM" = "1" ]; then + install_vm_service || warn "service VM non installe" + fi if command -v "$BINARY" >/dev/null 2>&1; then "$BINARY" --version 2>/dev/null || true elif [ -x "$INSTALL_DIR/$BINARY" ]; then @@ -408,6 +414,32 @@ install_helium_binary() { else install_from_release "$VERSION" || { warn "fallback build source"; install_from_source; }; fi } +# Service systemd pour la microVM (remonte seule au boot). +install_vm_service() { + [ "$PLAT_OS" = "linux" ] || { err "service VM = Linux requis"; return 1; } + [ -x "/srv/helium-vm/fc-vm.sh" ] || { err "fc-vm.sh absent (/srv/helium-vm)"; return 1; } + $SUDO tee /etc/systemd/system/helium-vm.service >/dev/null <<'EOF' +[Unit] +Description=Helium Firecracker microVM (2vCPU/2Go, guest 172.16.0.2) +After=network-online.target wg-quick@helium-poc.service +Wants=network-online.target + +[Service] +Type=oneshot +RemainAfterExit=yes +ExecStart=/srv/helium-vm/fc-vm.sh up +ExecStop=/srv/helium-vm/fc-vm.sh down + +[Install] +WantedBy=multi-user.target +EOF + $SUDO systemctl daemon-reload + $SUDO systemctl enable --now helium-vm + sleep 2 + $SUDO systemctl is-active --quiet helium-vm || { err "service helium-vm inactif"; return 1; } + ok "service helium-vm actif (remonte au boot)" +} + # Service systemd pour `helium daemon` (redemarrage auto). install_daemon_service() { [ "$PLAT_OS" = "linux" ] || { err "service daemon = Linux (systemd) requis"; return 1; }