From 9ad5588e92424ac943d53d6d7d377131ec9ada1c Mon Sep 17 00:00:00 2001 From: Anthony Ettinger Date: Mon, 9 Nov 2026 06:26:35 +0000 Subject: [PATCH] feat(music): live node progress for songs; c0mpute.com songs follow the brief's length; v0.2.85 - Gateway: POST /v1/music takes an x-request-id and records each stage of that request (finding a worker, queued behind busy workers, rendering on node X with the node's own ETA, fetching the result, retrying and why). GET /v1/progress/{id} returns it. Reported through a task-local, so the shared auction path needs no new arguments; the winning bid is now read as soon as it is chosen, not only on failure. - c0mpute.com/chat shows that line while a song renders (relayed by Infernet as `node`, sanitized). - c0mpute.com's /api/v1/music no longer defaults duration to 30: that was sent to Infernet as an explicit choice and overrode the length in the brief, so every c0mpute.com song came out 30 s. Co-Authored-By: Claude Opus 5.5 --- Cargo.lock | 32 ++-- Cargo.toml | 2 +- apps/tui/package.json | 2 +- apps/web/package.json | 2 +- apps/web/src/app/api/v1/music/[id]/route.ts | 4 +- apps/web/src/app/chat/chat-app.tsx | 16 +- .../web/src/app/releases/latest.json/route.ts | 2 +- apps/web/src/lib/chat/music.test.ts | 15 +- apps/web/src/lib/chat/music.ts | 67 +++++++- node/crates/c0mpute-core/src/job_progress.rs | 155 ++++++++++++++++++ node/crates/c0mpute-core/src/lib.rs | 1 + .../crates/c0mpute-core/src/speech_gateway.rs | 92 ++++++++++- package.json | 2 +- packages/shared/package.json | 2 +- plugins/ad-generator-agent/module.toml | 2 +- plugins/ads-manager/module.toml | 2 +- plugins/coinpay/module.toml | 2 +- plugins/crawl/module.toml | 2 +- plugins/ctv-ads/module.toml | 2 +- plugins/diffusion/module.toml | 2 +- plugins/embed/module.toml | 2 +- plugins/gameserver/module.toml | 2 +- plugins/hosting/module.toml | 2 +- plugins/infernet/module.toml | 2 +- plugins/live-stream/module.toml | 2 +- plugins/music/module.toml | 2 +- plugins/ocr/module.toml | 2 +- plugins/podcast-ads/module.toml | 2 +- plugins/podcasting/module.toml | 2 +- plugins/podcasts/module.toml | 2 +- plugins/render/module.toml | 2 +- plugins/secure-chat/module.toml | 2 +- plugins/storage/module.toml | 2 +- plugins/train/module.toml | 2 +- plugins/transcode/module.toml | 2 +- plugins/tts/module.toml | 2 +- plugins/video/module.toml | 2 +- plugins/webhook/module.toml | 2 +- plugins/whisper/module.toml | 2 +- plugins/workflow-step/module.toml | 2 +- 40 files changed, 377 insertions(+), 69 deletions(-) create mode 100644 node/crates/c0mpute-core/src/job_progress.rs diff --git a/Cargo.lock b/Cargo.lock index 363a7a54..01f363ac 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -478,7 +478,7 @@ checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" [[package]] name = "c0mpute-api" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "c0mpute-proto", @@ -493,7 +493,7 @@ dependencies = [ [[package]] name = "c0mpute-bench" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "blake3", @@ -504,7 +504,7 @@ dependencies = [ [[package]] name = "c0mpute-cli" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "axum", @@ -534,7 +534,7 @@ dependencies = [ [[package]] name = "c0mpute-core" -version = "0.2.84" +version = "0.2.85" dependencies = [ "aes-gcm", "anyhow", @@ -580,7 +580,7 @@ dependencies = [ [[package]] name = "c0mpute-doctor" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "serde", @@ -592,7 +592,7 @@ dependencies = [ [[package]] name = "c0mpute-envelope" -version = "0.2.84" +version = "0.2.85" dependencies = [ "blake3", "bs58", @@ -605,7 +605,7 @@ dependencies = [ [[package]] name = "c0mpute-gateway" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "axum", @@ -627,7 +627,7 @@ dependencies = [ [[package]] name = "c0mpute-market" -version = "0.2.84" +version = "0.2.85" dependencies = [ "c0mpute-envelope", "serde", @@ -637,7 +637,7 @@ dependencies = [ [[package]] name = "c0mpute-net" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "async-trait", @@ -654,7 +654,7 @@ dependencies = [ [[package]] name = "c0mpute-placement" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "async-trait", @@ -675,7 +675,7 @@ dependencies = [ [[package]] name = "c0mpute-proto" -version = "0.2.84" +version = "0.2.85" dependencies = [ "blake3", "hex", @@ -687,7 +687,7 @@ dependencies = [ [[package]] name = "c0mpute-secure-chat" -version = "0.2.84" +version = "0.2.85" dependencies = [ "aes-gcm", "anyhow", @@ -710,7 +710,7 @@ dependencies = [ [[package]] name = "c0mpute-store" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "blake3", @@ -728,7 +728,7 @@ dependencies = [ [[package]] name = "c0mpute-transcode" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "c0mpute-proto", @@ -741,7 +741,7 @@ dependencies = [ [[package]] name = "c0mpute-update" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "hex", @@ -755,7 +755,7 @@ dependencies = [ [[package]] name = "c0mpute-verify" -version = "0.2.84" +version = "0.2.85" dependencies = [ "anyhow", "blake3", diff --git a/Cargo.toml b/Cargo.toml index 4e5b2421..b2799924 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -23,7 +23,7 @@ members = [ ] [workspace.package] -version = "0.2.84" +version = "0.2.85" edition = "2024" license = "MIT" repository = "https://github.com/profullstack/c0mpute" diff --git a/apps/tui/package.json b/apps/tui/package.json index ec6664a9..e4e434c0 100644 --- a/apps/tui/package.json +++ b/apps/tui/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/tui", - "version": "0.2.84", + "version": "0.2.85", "private": true, "type": "module", "bin": { diff --git a/apps/web/package.json b/apps/web/package.json index 26394bdf..a3ebf163 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute", - "version": "0.2.84", + "version": "0.2.85", "private": true, "type": "module", "scripts": { diff --git a/apps/web/src/app/api/v1/music/[id]/route.ts b/apps/web/src/app/api/v1/music/[id]/route.ts index 4ed5d5f1..f8d1857c 100644 --- a/apps/web/src/app/api/v1/music/[id]/route.ts +++ b/apps/web/src/app/api/v1/music/[id]/route.ts @@ -1,5 +1,5 @@ import { isJobId } from "@/lib/chat/images"; -import { AUDIO_TYPES } from "@/lib/chat/music"; +import { AUDIO_TYPES, sanitizeNodeProgress } from "@/lib/chat/music"; import { currentUser, infernet, json, unauthorized } from "@/lib/chat/server"; export const dynamic = "force-dynamic"; @@ -33,7 +33,7 @@ export async function GET(req: Request, { params }: { params: Promise<{ id: stri // Infernet writes lyrics from a brief first; pass what it wrote along. const pick = (k: string) => (typeof body[k] === "string" || typeof body[k] === "number" ? body[k] : undefined); return json( - { id, status: "pending", seconds: body.seconds ?? null, stage: pick("stage"), caption: pick("caption"), lyrics: pick("lyrics"), duration: pick("duration") }, + { id, status: "pending", seconds: body.seconds ?? null, stage: pick("stage"), caption: pick("caption"), lyrics: pick("lyrics"), duration: pick("duration"), node: sanitizeNodeProgress(body.node) }, 202, { "cache-control": "no-store" }, ); diff --git a/apps/web/src/app/chat/chat-app.tsx b/apps/web/src/app/chat/chat-app.tsx index 88009b8b..5615ff39 100644 --- a/apps/web/src/app/chat/chat-app.tsx +++ b/apps/web/src/app/chat/chat-app.tsx @@ -3,7 +3,7 @@ import { useCallback, useEffect, useRef, useState } from "react"; import { CodeBlock } from "@/components/CodeBlock"; import { parseImageCommand, textHistory } from "@/lib/chat/images"; -import { audioExtension, parseMusicCommand, type MusicCommand } from "@/lib/chat/music"; +import { audioExtension, describeNodeProgress, parseMusicCommand, sanitizeNodeProgress, type MusicCommand } from "@/lib/chat/music"; // An image turn keeps its picture only for this session (a blob: URL); what is // saved to localStorage is the text placeholder, and the model never sees it. @@ -280,12 +280,14 @@ export function ChatApp({ error }: { error: string | null }) { const p = await poll.json().catch(() => ({})); const secs = Math.round((Date.now() - started) / 1000); if (typeof p.caption === "string") written = { caption: p.caption, lyrics: typeof p.lyrics === "string" ? p.lyrics : "" }; - setLast({ - content: - p.stage === "writing" - ? `// writing lyrics and a style from your brief… ${secs}s` - : `// composing with ACE-Step 1.5 on a c0mpute worker… ${secs}s${written?.lyrics ? `\n\n${written.lyrics}` : ""}`, - }); + const node = sanitizeNodeProgress(p.node); + const head = + p.stage === "writing" + ? `// writing lyrics and a style from your brief… ${secs}s` + : `// ${node ? describeNodeProgress(node) : `composing with ACE-Step 1.5 on a c0mpute worker…`} · ${secs}s total${ + typeof p.duration === "number" ? ` · ${p.duration}s song` : "" + }`; + setLast({ content: `${head}${written?.lyrics ? `\n\n${written.lyrics}` : ""}` }); continue; } if (!poll.ok) { diff --git a/apps/web/src/app/releases/latest.json/route.ts b/apps/web/src/app/releases/latest.json/route.ts index 7fd61665..296d44ba 100644 --- a/apps/web/src/app/releases/latest.json/route.ts +++ b/apps/web/src/app/releases/latest.json/route.ts @@ -28,7 +28,7 @@ export const dynamic = "force-dynamic"; * this in sync with the workspace version in `Cargo.toml` on each release so a * freshly built CLI reports "already latest" rather than a phantom upgrade. */ -const FALLBACK_VERSION = "0.2.84"; +const FALLBACK_VERSION = "0.2.85"; const FALLBACK_MIN_REQUIRED = "0.0.1"; type Channel = "stable" | "beta" | "nightly"; diff --git a/apps/web/src/lib/chat/music.test.ts b/apps/web/src/lib/chat/music.test.ts index e2669191..a33f1a9f 100644 --- a/apps/web/src/lib/chat/music.test.ts +++ b/apps/web/src/lib/chat/music.test.ts @@ -1,5 +1,5 @@ import { describe, expect, test } from "bun:test"; -import { audioExtension, parseMusicCommand, parseMusicRequest } from "./music"; +import { audioExtension, describeNodeProgress, parseMusicCommand, parseMusicRequest, sanitizeNodeProgress } from "./music"; describe("chat music", () => { test("parses /music and /song, with optional lyrics", () => { @@ -12,12 +12,12 @@ describe("chat music", () => { expect(parseMusicCommand("play music")).toBeNull(); }); test("validates the request", () => { - expect(parseMusicRequest({ prompt: " jazz " })).toEqual({ prompt: "jazz", lyrics: "", duration: 30 }); + expect(parseMusicRequest({ prompt: " jazz " })).toEqual({ prompt: "jazz", lyrics: "" }); expect(parseMusicRequest({ prompt: "x", duration: 60, lyrics: "hi" })).toEqual({ prompt: "x", lyrics: "hi", duration: 60 }); expect(typeof parseMusicRequest({ prompt: "x", duration: 5 })).toBe("string"); expect(typeof parseMusicRequest({ prompt: "x", duration: 200 })).toBe("string"); expect(typeof parseMusicRequest({ prompt: "" })).toBe("string"); - expect(parseMusicRequest({ prompt: "x".repeat(100_000), lyrics: "y".repeat(100_000) })).toMatchObject({ duration: 30 }); + expect(parseMusicRequest({ prompt: "x".repeat(100_000), lyrics: "y".repeat(100_000) })).toMatchObject({ lyrics: "y".repeat(100_000) }); expect(typeof parseMusicRequest({ prompt: "x".repeat(100_001) })).toBe("string"); expect(typeof parseMusicRequest(null)).toBe("string"); }); @@ -26,4 +26,13 @@ describe("chat music", () => { expect(audioExtension("audio/wav")).toBe("wav"); expect(audioExtension("")).toBe("mp3"); }); + test("node progress is sanitized and described", () => { + expect(sanitizeNodeProgress({ stage: "evil" })).toBeUndefined(); + expect(sanitizeNodeProgress(null)).toBeUndefined(); + const p = sanitizeNodeProgress({ stage: "rendering", worker: "12D3KooWG9TmqCHN63oFoxb2bwTGRYqwW8e1C2Ua9e14XtPja8R9", eta_s: 600, stage_elapsed_s: 125, attempt: 1, junk: 1 }); + expect(p).toMatchObject({ stage: "rendering", eta_s: 600 }); + expect(describeNodeProgress(p!)).toBe("rendering on node …XtPja8R9 · 2:05 in · ~7:55 left (node's estimate)"); + expect(describeNodeProgress({ stage: "queued", stage_elapsed_s: 61 })).toBe("every music worker is busy; queued 1:01"); + expect(describeNodeProgress({ stage: "retrying", worker: "abcdefgh12345678", detail: "worker failed" })).toContain("trying another worker"); + }); }); diff --git a/apps/web/src/lib/chat/music.ts b/apps/web/src/lib/chat/music.ts index dcfe64d8..057da887 100644 --- a/apps/web/src/lib/chat/music.ts +++ b/apps/web/src/lib/chat/music.ts @@ -27,20 +27,83 @@ export const MAX_SECONDS = 180; /** Validate a POST body; returns the body to forward or an error string. */ export function parseMusicRequest( body: unknown, -): { prompt: string; lyrics: string; duration: number } | string { +): { prompt: string; lyrics: string; duration?: number } | string { const b = (body ?? {}) as Record; const prompt = typeof b.prompt === "string" ? b.prompt.trim() : ""; // Long briefs and whole songs are fine; Infernet clips what the model reads. if (!prompt || prompt.length > MAX_TEXT_CHARS) return `prompt must contain 1 to ${MAX_TEXT_CHARS} characters`; const lyrics = typeof b.lyrics === "string" ? b.lyrics.trim() : ""; if (lyrics.length > MAX_TEXT_CHARS) return `lyrics must be at most ${MAX_TEXT_CHARS} characters`; - const duration = b.duration === undefined || b.duration === null ? 30 : Number(b.duration); + // No duration means "whatever the brief asks for": Infernet reads a length + // from the prompt ("150-second song") or lets its writer pick. Defaulting + // to 30 here overrode that and every c0mpute.com song came out 30 s. + if (b.duration === undefined || b.duration === null) return { prompt, lyrics }; + const duration = Number(b.duration); if (!Number.isInteger(duration) || duration < MIN_SECONDS || duration > MAX_SECONDS) { return `duration must be a whole number of seconds from ${MIN_SECONDS} to ${MAX_SECONDS}`; } return { prompt, lyrics, duration }; } +/** What the c0mpute node doing a song reports, relayed by Infernet. */ +export type NodeProgress = { + stage: string; + worker?: string; + eta_s?: number; + elapsed_s?: number; + stage_elapsed_s?: number; + attempt?: number; + detail?: string; +}; + +const STAGES = ["starting", "finding_worker", "queued", "rendering", "fetching_result", "retrying", "done", "failed"]; + +/** Keep only known, well-typed fields of an upstream progress object. */ +export function sanitizeNodeProgress(v: unknown): NodeProgress | undefined { + if (!v || typeof v !== "object") return undefined; + const o = v as Record; + if (typeof o.stage !== "string" || !STAGES.includes(o.stage)) return undefined; + const num = (k: string) => (typeof o[k] === "number" && Number.isFinite(o[k]) && (o[k] as number) >= 0 ? (o[k] as number) : undefined); + const str = (k: string, max: number) => (typeof o[k] === "string" ? (o[k] as string).slice(0, max) : undefined); + return { + stage: o.stage, + worker: str("worker", 80), + eta_s: num("eta_s"), + elapsed_s: num("elapsed_s"), + stage_elapsed_s: num("stage_elapsed_s"), + attempt: num("attempt"), + detail: str("detail", 200), + }; +} + +const mmss = (s: number) => `${Math.floor(s / 60)}:${String(Math.floor(s % 60)).padStart(2, "0")}`; + +/** One line for the chat: what the network is doing with this song now. */ +export function describeNodeProgress(p: NodeProgress): string { + const node = p.worker ? `node …${p.worker.slice(-8)}` : "a node"; + const retry = (p.attempt ?? 1) > 1 ? ` (attempt ${p.attempt})` : ""; + switch (p.stage) { + case "starting": + case "finding_worker": + return `finding a c0mpute music worker${retry}`; + case "queued": + return `every music worker is busy; queued ${mmss(p.stage_elapsed_s ?? 0)}`; + case "rendering": { + const left = p.eta_s !== undefined ? Math.max(0, p.eta_s - (p.stage_elapsed_s ?? 0)) : undefined; + const eta = left === undefined ? "" : left > 0 ? ` · ~${mmss(left)} left (node's estimate)` : " · past the node's estimate, still working"; + return `rendering on ${node}${retry} · ${mmss(p.stage_elapsed_s ?? 0)} in${eta}`; + } + case "fetching_result": + return `${node} finished; fetching the song`; + case "retrying": + return `${node} failed${p.detail ? ` (${p.detail})` : ""}; trying another worker`; + case "done": + return "song ready"; + default: + return "song failed"; + } +} + export const AUDIO_TYPES = ["audio/mpeg", "audio/wav", "audio/flac", "audio/ogg"]; export function audioExtension(type: string): string { diff --git a/node/crates/c0mpute-core/src/job_progress.rs b/node/crates/c0mpute-core/src/job_progress.rs new file mode 100644 index 00000000..0b346604 --- /dev/null +++ b/node/crates/c0mpute-core/src/job_progress.rs @@ -0,0 +1,155 @@ +//! Live progress for long gateway jobs (songs), so a caller waiting minutes +//! on `POST /v1/music` can show what is happening instead of a timer. +//! +//! The caller names its request with an `x-request-id` header; the gateway +//! records each stage under that id (finding a worker, queued behind busy +//! workers, rendering on node X with its ETA, fetching the result, retrying) +//! and `GET /v1/progress/{id}` returns the latest. The job code reports +//! through a task-local, so the shared auction path needs no new arguments. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{SystemTime, UNIX_EPOCH}; + +use serde::Serialize; + +/// How long a finished request's progress stays readable. +const KEEP_MS: u64 = 30 * 60 * 1000; +const MAX_ENTRIES: usize = 512; + +#[derive(Clone, Debug, Serialize)] +pub struct JobProgress { + /// starting | finding_worker | queued | rendering | fetching_result | + /// retrying | done | failed + pub stage: &'static str, + /// Auction attempt (a worker that fails or vanishes means another). + pub attempt: u32, + /// The winning worker's peer id while it renders. + pub worker: Option, + /// The c0mpute job id of the current attempt. + pub job_id: Option, + /// The winning worker's own estimate, from its measured job times. + pub eta_s: Option, + /// Why the last attempt was retried, when it was. + pub detail: Option, + pub started_at_ms: u64, + pub stage_since_ms: u64, +} + +fn now_ms() -> u64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[derive(Clone, Default)] +pub struct ProgressBoard(Arc>>); + +impl ProgressBoard { + pub fn start(&self, key: &str) { + let now = now_ms(); + let Ok(mut map) = self.0.lock() else { return }; + map.retain(|_, p| now.saturating_sub(p.started_at_ms) < KEEP_MS); + if map.len() >= MAX_ENTRIES + && let Some(oldest) = map.iter().min_by_key(|(_, p)| p.started_at_ms).map(|(k, _)| k.clone()) + { + map.remove(&oldest); + } + map.insert( + key.to_string(), + JobProgress { + stage: "starting", + attempt: 0, + worker: None, + job_id: None, + eta_s: None, + detail: None, + started_at_ms: now, + stage_since_ms: now, + }, + ); + } + + pub fn update(&self, key: &str, f: impl FnOnce(&mut JobProgress)) { + if let Ok(mut map) = self.0.lock() + && let Some(p) = map.get_mut(key) + { + f(p); + } + } + + /// The entry plus how long the request and its current stage have run. + pub fn snapshot(&self, key: &str) -> Option { + let map = self.0.lock().ok()?; + let p = map.get(key)?; + let now = now_ms(); + let mut v = serde_json::to_value(p).ok()?; + v["elapsed_s"] = (now.saturating_sub(p.started_at_ms) / 1000).into(); + v["stage_elapsed_s"] = (now.saturating_sub(p.stage_since_ms) / 1000).into(); + Some(v) + } +} + +/// A caller-chosen request id: short and URL-safe. +pub fn valid_key(key: &str) -> bool { + !key.is_empty() + && key.len() <= 64 + && key.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-' || b == b'_') +} + +tokio::task_local! { + /// The board and request id of the job running on this task, if any. + pub static CURRENT: (ProgressBoard, String); +} + +/// Change the current request's progress; a no-op outside a tracked job. +pub fn report(f: impl FnOnce(&mut JobProgress)) { + let _ = CURRENT.try_with(|(board, key)| board.update(key, f)); +} + +/// Move the current request to `stage` (the stage clock restarts only on a change). +pub fn set_stage(stage: &'static str) { + report(|p| { + if p.stage != stage { + p.stage = stage; + p.stage_since_ms = now_ms(); + } + }); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn stages_are_recorded_under_the_request_id_only_inside_its_scope() { + let board = ProgressBoard::default(); + board.start("req-1"); + set_stage("rendering"); // outside any scope: ignored + assert_eq!(board.snapshot("req-1").unwrap()["stage"], "starting"); + CURRENT + .scope((board.clone(), "req-1".to_string()), async { + set_stage("queued"); + report(|p| { + p.worker = Some("12D3KooW".into()); + p.eta_s = Some(140); + }); + }) + .await; + let s = board.snapshot("req-1").unwrap(); + assert_eq!(s["stage"], "queued"); + assert_eq!(s["worker"], "12D3KooW"); + assert_eq!(s["eta_s"], 140); + assert!(s["elapsed_s"].is_u64()); + assert!(board.snapshot("nope").is_none()); + } + + #[test] + fn request_ids_must_be_short_and_url_safe() { + assert!(valid_key("17a1b21d-a95e-4345-a3fe-d14dfd6868e3")); + assert!(!valid_key("")); + assert!(!valid_key("../etc")); + assert!(!valid_key(&"a".repeat(65))); + } +} diff --git a/node/crates/c0mpute-core/src/lib.rs b/node/crates/c0mpute-core/src/lib.rs index d97efaa7..49e81ed8 100644 --- a/node/crates/c0mpute-core/src/lib.rs +++ b/node/crates/c0mpute-core/src/lib.rs @@ -13,6 +13,7 @@ pub mod crawl; pub mod diffusion; pub mod dm; pub mod music; +pub mod job_progress; pub mod hosting; pub mod live; pub mod dispatch; diff --git a/node/crates/c0mpute-core/src/speech_gateway.rs b/node/crates/c0mpute-core/src/speech_gateway.rs index a26ef1ab..f68413f8 100644 --- a/node/crates/c0mpute-core/src/speech_gateway.rs +++ b/node/crates/c0mpute-core/src/speech_gateway.rs @@ -60,6 +60,8 @@ struct GatewayState { sessions: crate::speech_session::SessionStore, /// Workers that won and did not deliver, benched for a while. penalties: crate::penalty_box::PenaltyBox, + /// Live stages of long requests (songs), by the caller's x-request-id. + progress: crate::job_progress::ProgressBoard, } #[derive(Debug)] @@ -176,6 +178,7 @@ pub fn router( ocr_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_OCR_JOBS)), embed_slots: Arc::new(Semaphore::new(MAX_CONCURRENT_EMBED_JOBS)), sessions: crate::speech_session::SessionStore::new(), + progress: crate::job_progress::ProgressBoard::default(), penalties, }; Router::new() @@ -185,6 +188,7 @@ pub fn router( .route("/v1/stt", post(transcribe)) .route("/v1/images", post(generate_image)) .route("/v1/music", post(generate_music)) + .route("/v1/progress/{request}", get(job_progress)) .route("/v1/videos", post(generate_video)) // OCR takes a 32 MiB document, above the router-wide audio limit. .route( @@ -521,10 +525,29 @@ async fn run_speech_auction( } else { Duration::from_secs(15 * 60) }); - run_auction(state.net.clone(), auction) - .await + // Read the winner as soon as it is chosen (not only on failure), so a + // caller polling /v1/progress sees which node is working and its ETA. + let auction_run = run_auction(state.net.clone(), auction); + tokio::pin!(auction_run); + let mut winner_seen: Option = None; + let result = loop { + tokio::select! { + result = &mut auction_run => break result, + Some((job_id, bid)) = winner_rx.recv() => { + crate::job_progress::set_stage("rendering"); + crate::job_progress::report(|p| { + p.worker = Some(bid.bidder_peer_id.clone()); + p.job_id = Some(job_id); + p.eta_s = Some(bid.eta_seconds); + }); + winner_seen = Some(bid.bidder_peer_id); + } + } + }; + result .map_err(|error| { - let winner = winner_rx.try_recv().ok().map(|(_, bid)| bid.bidder_peer_id); + let winner = winner_seen + .or_else(|| winner_rx.try_recv().ok().map(|(_, bid)| bid.bidder_peer_id)); tracing::warn!(workload, err = %error, winner = ?winner, "speech auction failed"); ( ApiError( @@ -572,6 +595,16 @@ async fn run_checked_job( while attempt < MAX_SPEECH_WORKER_ATTEMPTS { attempt += 1; + crate::job_progress::report(|p| { + p.attempt = attempt as u32; + p.worker = None; + p.eta_s = None; + }); + crate::job_progress::report(|p| { + if p.stage != "queued" { + p.stage = "finding_worker"; + } + }); let outcome = match run_speech_auction( state, workload, @@ -599,7 +632,8 @@ async fn run_checked_job( if (workload == crate::diffusion::WORKLOAD_TXT2IMG && started.elapsed() < IMAGE_QUEUE_WAIT) || (workload == crate::music::WORKLOAD_TXT2MUSIC && started.elapsed() < MUSIC_QUEUE_WAIT) => { - tracing::info!(workload, waited_secs = started.elapsed().as_secs(), "no free image worker yet; waiting"); + tracing::info!(workload, waited_secs = started.elapsed().as_secs(), "no free worker yet; waiting"); + crate::job_progress::set_stage("queued"); attempt -= 1; last_error.get_or_insert(error); tokio::time::sleep(Duration::from_secs(15)).await; @@ -608,6 +642,7 @@ async fn run_checked_job( Err((error, None)) => return Err(last_error.unwrap_or(error)), }; let worker = outcome.accepted_bid.bidder_peer_id.clone(); + crate::job_progress::set_stage("fetching_result"); let result = if outcome.receipt.status != c0mpute_net::topics::JobStatus::Completed { Err(ApiError( StatusCode::BAD_GATEWAY, @@ -636,6 +671,9 @@ async fn run_checked_job( } Err(error) => { state.penalties.strike(&worker); + let why = error.1.clone(); + crate::job_progress::set_stage("retrying"); + crate::job_progress::report(|p| p.detail = Some(why)); tracing::warn!( workload, attempt, @@ -1004,7 +1042,12 @@ async fn generate_music( .map_err(|_| ApiError(StatusCode::SERVICE_UNAVAILABLE, "music queue is unavailable".into()))?; let spec = serde_json::to_value(input.clip_for_model()) .map_err(|error| gateway_failure(error.into()))?; - let (outcome, bytes) = run_speech_job( + let request_id = headers + .get("x-request-id") + .and_then(|v| v.to_str().ok()) + .filter(|k| crate::job_progress::valid_key(k)) + .map(str::to_string); + let job = run_speech_job( &state, crate::music::WORKLOAD_TXT2MUSIC, spec, @@ -1013,8 +1056,21 @@ async fn generate_music( crate::music::MODEL_TAG.to_string(), ], crate::music::MAX_RESULT_BYTES, - ) - .await?; + ); + let result = match &request_id { + Some(key) => { + state.progress.start(key); + crate::job_progress::CURRENT + .scope((state.progress.clone(), key.clone()), async { + let result = job.await; + crate::job_progress::set_stage(if result.is_ok() { "done" } else { "failed" }); + result + }) + .await + } + None => job.await, + }; + let (outcome, bytes) = result?; let content_type = crate::music::audio_content_type(&bytes).unwrap_or("audio/mpeg"); let mut response = (StatusCode::OK, bytes).into_response(); response @@ -1031,6 +1087,28 @@ async fn generate_music( Ok(response) } +/// `GET /v1/progress/{request}`: the live stage of a request sent with that +/// `x-request-id` (see `job_progress`). 404 until it starts and 30 min after. +async fn job_progress( + State(state): State, + headers: HeaderMap, + Path(request): Path, +) -> ApiResult { + authenticate(&state, &headers)?; + if !crate::job_progress::valid_key(&request) { + return Err(ApiError(StatusCode::BAD_REQUEST, "bad request id".into())); + } + let snapshot = state + .progress + .snapshot(&request) + .ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such request".into()))?; + let mut response = Json(snapshot).into_response(); + response + .headers_mut() + .insert(header::CACHE_CONTROL, HeaderValue::from_static("no-store")); + Ok(response) +} + async fn generate_image( State(state): State, headers: HeaderMap, diff --git a/package.json b/package.json index 27f5ab00..65a2277c 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "c0mpute-monorepo", - "version": "0.2.84", + "version": "0.2.85", "private": true, "workspaces": [ "apps/*", diff --git a/packages/shared/package.json b/packages/shared/package.json index ff998df7..3e5fd486 100644 --- a/packages/shared/package.json +++ b/packages/shared/package.json @@ -1,6 +1,6 @@ { "name": "@c0mpute/shared", - "version": "0.2.84", + "version": "0.2.85", "private": true, "type": "module", "main": "./dist/index.js", diff --git a/plugins/ad-generator-agent/module.toml b/plugins/ad-generator-agent/module.toml index 54b71660..123b1368 100644 --- a/plugins/ad-generator-agent/module.toml +++ b/plugins/ad-generator-agent/module.toml @@ -21,7 +21,7 @@ [module] id = "ad-generator-agent" name = "Ad Generator Agent — every ad type from one URL" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "available" diff --git a/plugins/ads-manager/module.toml b/plugins/ads-manager/module.toml index ad8c2a1c..77b61171 100644 --- a/plugins/ads-manager/module.toml +++ b/plugins/ads-manager/module.toml @@ -38,7 +38,7 @@ [module] id = "ads-manager" name = "Ads Manager — shared ad operations layer" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "partial" diff --git a/plugins/coinpay/module.toml b/plugins/coinpay/module.toml index 8ff70d14..a53866cd 100644 --- a/plugins/coinpay/module.toml +++ b/plugins/coinpay/module.toml @@ -7,7 +7,7 @@ [module] id = "coinpay" name = "CoinPay — DID, payments, escrow, reputation" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "available" diff --git a/plugins/crawl/module.toml b/plugins/crawl/module.toml index 75114027..52f5df82 100644 --- a/plugins/crawl/module.toml +++ b/plugins/crawl/module.toml @@ -8,7 +8,7 @@ [module] id = "crawl" name = "Crawl — distributed web fetching" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "data" status = "available" diff --git a/plugins/ctv-ads/module.toml b/plugins/ctv-ads/module.toml index 213584be..a7a62cf4 100644 --- a/plugins/ctv-ads/module.toml +++ b/plugins/ctv-ads/module.toml @@ -28,7 +28,7 @@ [module] id = "ctv-ads" name = "CTV Ads — VAST and VMAP from every gateway" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "partial" diff --git a/plugins/diffusion/module.toml b/plugins/diffusion/module.toml index d62bbdbf..818aff70 100644 --- a/plugins/diffusion/module.toml +++ b/plugins/diffusion/module.toml @@ -11,7 +11,7 @@ [module] id = "diffusion" name = "Diffusion — image generation" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/embed/module.toml b/plugins/embed/module.toml index bd5fbeb4..bfae4ebd 100644 --- a/plugins/embed/module.toml +++ b/plugins/embed/module.toml @@ -19,7 +19,7 @@ [module] id = "embed" name = "Embeddings — vector generation" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/gameserver/module.toml b/plugins/gameserver/module.toml index d222e1da..04f8b7d4 100644 --- a/plugins/gameserver/module.toml +++ b/plugins/gameserver/module.toml @@ -55,7 +55,7 @@ [module] id = "gameserver" name = "Game server — Minecraft / Factorio / Valheim" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "infrastructure" status = "available" diff --git a/plugins/hosting/module.toml b/plugins/hosting/module.toml index d477d89c..96461f7d 100644 --- a/plugins/hosting/module.toml +++ b/plugins/hosting/module.toml @@ -30,7 +30,7 @@ [module] id = "hosting" name = "Hosting — p2p static sites" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "data" status = "available" diff --git a/plugins/infernet/module.toml b/plugins/infernet/module.toml index ed19829f..ad62ed20 100644 --- a/plugins/infernet/module.toml +++ b/plugins/infernet/module.toml @@ -22,7 +22,7 @@ [module] id = "infernet" name = "Infernet Protocol — AI inference" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/live-stream/module.toml b/plugins/live-stream/module.toml index 52b019e8..fca70bfb 100644 --- a/plugins/live-stream/module.toml +++ b/plugins/live-stream/module.toml @@ -34,7 +34,7 @@ [module] id = "live-stream" name = "Live Stream — p2p HLS live streaming" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "media" status = "available" diff --git a/plugins/music/module.toml b/plugins/music/module.toml index 0de0e3ee..d593b89f 100644 --- a/plugins/music/module.toml +++ b/plugins/music/module.toml @@ -15,7 +15,7 @@ [module] id = "music" name = "Music — song generation (ACE-Step 1.5)" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/ocr/module.toml b/plugins/ocr/module.toml index f51d9244..8a0f9778 100644 --- a/plugins/ocr/module.toml +++ b/plugins/ocr/module.toml @@ -19,7 +19,7 @@ [module] id = "ocr" name = "OCR — document text extraction" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/podcast-ads/module.toml b/plugins/podcast-ads/module.toml index 5a093c7b..df1cffcb 100644 --- a/plugins/podcast-ads/module.toml +++ b/plugins/podcast-ads/module.toml @@ -31,7 +31,7 @@ [module] id = "podcast-ads" name = "Podcast Ads — dynamic ad insertion" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "partial" diff --git a/plugins/podcasting/module.toml b/plugins/podcasting/module.toml index 530f5243..526069a4 100644 --- a/plugins/podcasting/module.toml +++ b/plugins/podcasting/module.toml @@ -24,7 +24,7 @@ [module] id = "podcasting" name = "Podcasting — p2p podcast publishing" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "media" status = "available" diff --git a/plugins/podcasts/module.toml b/plugins/podcasts/module.toml index ce4ed5ce..27e715dd 100644 --- a/plugins/podcasts/module.toml +++ b/plugins/podcasts/module.toml @@ -24,7 +24,7 @@ [module] id = "podcasts" name = "Podcasts — p2p podcast listener" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "media" status = "available" diff --git a/plugins/render/module.toml b/plugins/render/module.toml index e953f87f..ad87730c 100644 --- a/plugins/render/module.toml +++ b/plugins/render/module.toml @@ -33,7 +33,7 @@ [module] id = "render" name = "Render — Blender 3D rendering" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "media" status = "available" diff --git a/plugins/secure-chat/module.toml b/plugins/secure-chat/module.toml index efbae4f9..5373be86 100644 --- a/plugins/secure-chat/module.toml +++ b/plugins/secure-chat/module.toml @@ -21,7 +21,7 @@ [module] id = "secure-chat" name = "Secure Chat — E2E encrypted p2p messaging" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "applications" status = "available" diff --git a/plugins/storage/module.toml b/plugins/storage/module.toml index 3b0591cd..68ff8bcb 100644 --- a/plugins/storage/module.toml +++ b/plugins/storage/module.toml @@ -20,7 +20,7 @@ [module] id = "storage" name = "Storage — Reed-Solomon shard service" -version = "0.2.84" +version = "0.2.85" kind = "service" category = "data" status = "available" diff --git a/plugins/train/module.toml b/plugins/train/module.toml index e2a14754..120c313a 100644 --- a/plugins/train/module.toml +++ b/plugins/train/module.toml @@ -41,7 +41,7 @@ [module] id = "train" name = "Fine-tune — LoRA training" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/transcode/module.toml b/plugins/transcode/module.toml index 9ea03792..5d8e7829 100644 --- a/plugins/transcode/module.toml +++ b/plugins/transcode/module.toml @@ -7,7 +7,7 @@ [module] id = "transcode" name = "FFmpeg Transcoding" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "media" status = "available" diff --git a/plugins/tts/module.toml b/plugins/tts/module.toml index 730d34a2..f5090ecf 100644 --- a/plugins/tts/module.toml +++ b/plugins/tts/module.toml @@ -15,7 +15,7 @@ [module] id = "tts" name = "TTS — text-to-speech" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/video/module.toml b/plugins/video/module.toml index 8b194296..8dad8108 100644 --- a/plugins/video/module.toml +++ b/plugins/video/module.toml @@ -50,7 +50,7 @@ [module] id = "video" name = "Video — generation (Wan 2.2, LTX-2.3 with audio)" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/webhook/module.toml b/plugins/webhook/module.toml index b62b5fee..f9d679e9 100644 --- a/plugins/webhook/module.toml +++ b/plugins/webhook/module.toml @@ -7,7 +7,7 @@ [module] id = "webhook" name = "Webhooks — paid, end-to-end encrypted delivery" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "infrastructure" status = "available" diff --git a/plugins/whisper/module.toml b/plugins/whisper/module.toml index 192c7a8f..d407af1c 100644 --- a/plugins/whisper/module.toml +++ b/plugins/whisper/module.toml @@ -7,7 +7,7 @@ [module] id = "whisper" name = "Whisper — speech-to-text" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "ai" status = "available" diff --git a/plugins/workflow-step/module.toml b/plugins/workflow-step/module.toml index 07f541a9..6eda5fcd 100644 --- a/plugins/workflow-step/module.toml +++ b/plugins/workflow-step/module.toml @@ -14,7 +14,7 @@ [module] id = "workflow-step" name = "Workflow step — webhook workflow steps (meshhook)" -version = "0.2.84" +version = "0.2.85" kind = "workload" category = "infrastructure" status = "available"