Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 42 additions & 3 deletions .github/workflows/video-runner.yml
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,19 @@
# torch image is several GB and not worth building to learn that a validation
# rule changed.
#
# The image the node pulls is the tag in plugins/video/module.toml under
# [module.dispatch]; keep the two in step.
# Tags on every master build: :stable (what operators pull by hand),
# :candidate (kept for older nodes) and :<sha>. Nodes pull by DIGEST, the one
# pinned in plugins/video/module.toml [module.dispatch] and RUNNER_DIGEST in
# node/crates/c0mpute-core/src/video.rs (a unit test keeps those two equal).
# The job summary prints the new digest; bump both pins to it after a runner
# change.
#
# Visibility: the package must be PUBLIC for outside operators to pull it.
# GHCR creates org packages private and has no API to change that, so it is
# one manual click by an org admin: github.com/orgs/profullstack/packages ->
# c0mpute-video-runner -> Package settings -> Change visibility -> Public.
# The last step below checks an anonymous pull and warns until it works.
# Until then `c0mpute video setup` builds the same image from source.
name: video runner

on:
Expand Down Expand Up @@ -81,15 +92,21 @@ jobs:
EXTRA: ${{ inputs.tag }}
run: |
set -eu
tags="${IMAGE}:candidate,${IMAGE}:${GITHUB_SHA}"
tags="${IMAGE}:stable,${IMAGE}:candidate,${IMAGE}:${GITHUB_SHA}"
if [ -n "${EXTRA:-}" ]; then
tags="${tags},${IMAGE}:${EXTRA}"
fi
echo "tags=${tags}" >> "$GITHUB_OUTPUT"

- name: Build and push
id: push
uses: docker/build-push-action@v6
with:
# Links the package to this public repo (and its README/license).
labels: |
org.opencontainers.image.source=https://github.com/${{ github.repository }}
org.opencontainers.image.description=c0mpute video runner (Wan 2.2, LTX-2.3)
org.opencontainers.image.licenses=MIT
# The context is plugins/video so module.toml — the runner's model
# registry — is part of the image.
context: plugins/video
Expand All @@ -103,3 +120,25 @@ jobs:
# Proves the thing we just pushed can parse the manifest, encode and
# probe. Needs no GPU, so a hosted runner can do it.
run: docker run --rm "${IMAGE}:${GITHUB_SHA}" selftest

- name: Digest to pin, and can a stranger pull it?
env:
DIGEST: ${{ steps.push.outputs.digest }}
run: |
set -u
{
echo "### video runner published"
echo ""
echo "Pin this in plugins/video/module.toml and RUNNER_DIGEST in video.rs:"
echo ""
echo " ${IMAGE}@${DIGEST}"
} >> "$GITHUB_STEP_SUMMARY"
# Anonymous pull: a fresh docker config with no credentials.
export DOCKER_CONFIG="$(mktemp -d)"
if docker manifest inspect "${IMAGE}@${DIGEST}" >/dev/null 2>&1; then
echo "Anonymous pull: OK (package is public)." >> "$GITHUB_STEP_SUMMARY"
else
msg="ghcr.io package c0mpute-video-runner is not publicly pullable. An org admin must set it Public once: github.com/orgs/profullstack/packages -> c0mpute-video-runner -> Package settings -> Change visibility. Until then nodes build the runner from source."
echo "::warning::${msg}"
echo "Anonymous pull: FAILED. ${msg}" >> "$GITHUB_STEP_SUMMARY"
fi
6 changes: 6 additions & 0 deletions apps/web/src/app/plugins/page.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -186,6 +186,12 @@ function PluginCard({ p }: { p: PluginManifest }) {

<ModelList p={p} />

{typeof p.requirements?.hardware === "string" && (
<p className="text-xs text-[var(--color-dim)]">
<span className="accent">requires:</span> {p.requirements.hardware}
</p>
)}

{p.status === "partial" && p.statusNote && (
<p className="text-xs text-[var(--color-dim)]">
<span className="accent">partial:</span> {p.statusNote}
Expand Down
2 changes: 2 additions & 0 deletions apps/web/src/lib/plugins.ts
Original file line number Diff line number Diff line change
Expand Up @@ -113,6 +113,8 @@ export interface PluginManifest {
os?: string[];
arch?: string[];
capabilities?: string[];
/** What a node needs to serve it, in words (GPU, docker, public IP). */
hardware?: string;
};
workloads?: Record<string, { command?: string; validation?: string }>;
models?: Record<string, ModelOption>;
Expand Down
80 changes: 74 additions & 6 deletions node/crates/c0mpute-cli/src/gameserver.rs
Original file line number Diff line number Diff line change
Expand Up @@ -34,12 +34,16 @@ enum Sub {
/// Restore a world: the hash `gameserver stop` printed, or a .tar of the game's data dir.
#[arg(long)]
world: Option<String>,
/// Valheim server password (5-32 letters/digits; peers on the job topic can see it).
/// Valheim server password (5-32 letters/digits). Sent sealed to the
/// winning worker after the auction, never in the public offer.
#[arg(long)]
password: Option<String>,
/// Server name / MOTD.
#[arg(long)]
name: Option<String>,
/// Minutes between live world backups, sealed to your key (0 = only at stop/expiry).
#[arg(long, default_value_t = gameserver::DEFAULT_BACKUP_MINS)]
backup_mins: u32,
/// Most you pay for the whole reservation, USD.
#[arg(long, default_value_t = 1.0)]
max_price: f64,
Expand All @@ -56,6 +60,32 @@ enum Sub {
#[arg(long)]
json: bool,
},
/// Watch a server: probe its port, mirror its backups, and if it stays
/// down for --fail-after-secs, start the same world on another worker.
Watch {
id: String,
#[arg(long, default_value_t = 30)]
probe_every_secs: u64,
#[arg(long, default_value_t = 300)]
fail_after_secs: u64,
/// Only probe and mirror backups; never start a replacement.
#[arg(long)]
no_failover: bool,
/// Most you pay for a replacement, USD.
#[arg(long, default_value_t = 1.0)]
max_price: f64,
/// Only accept replacement bids from these worker peer ids. Repeatable.
#[arg(long)]
worker: Vec<String>,
},
/// Show a server's probe log (what `watch` recorded).
Health {
id: String,
#[arg(long, default_value_t = 20)]
last: usize,
#[arg(long)]
json: bool,
},
/// Stop a server early (or collect an expired one's backup) and download its world.
Stop {
id: String,
Expand Down Expand Up @@ -85,7 +115,7 @@ fn left(expires_at: u64) -> String {
pub async fn run(args: &[String]) -> Result<()> {
let cli = GameserverCli::try_parse_from(std::iter::once("c0mpute gameserver".to_string()).chain(args.iter().cloned())).unwrap_or_else(|e| e.exit());
match cli.cmd {
Sub::Start { game, hours, ram, cpus, accept_eula, world, password, name, max_price, worker, timeout_secs, json } => {
Sub::Start { game, hours, ram, cpus, accept_eula, world, password, name, backup_mins, max_price, worker, timeout_secs, json } => {
let game = Game::parse(&game).with_context(|| format!("unknown game {game:?}: minecraft, factorio or valheim"))?;
let req = StartRequest {
game,
Expand All @@ -96,8 +126,10 @@ pub async fn run(args: &[String]) -> Result<()> {
world,
password,
server_name: name,
backup_mins,
max_price_usd: max_price,
workers: worker,
exclude: Vec::new(),
timeout: Duration::from_secs(timeout_secs.clamp(60, 3600)),
};
eprintln!("looking for a worker that hosts {} ...", game.name());
Expand All @@ -107,6 +139,7 @@ pub async fn run(args: &[String]) -> Result<()> {
} else {
println!("{} {} is up at {}", r.game.name(), r.id, r.address);
println!(" worker {} ${} stops in {}", r.worker, r.price_usd, left(r.expires_at));
println!(" keep it healthy (probe, mirror backups, fail over): c0mpute gameserver watch {}", r.id);
println!(" stop early (and keep the world): c0mpute gameserver stop {}", r.id);
}
Ok(())
Expand All @@ -121,15 +154,50 @@ pub async fn run(args: &[String]) -> Result<()> {
println!("no servers rented from this machine");
}
for r in all {
let status = match (r.stopped_at, &r.backup) {
(Some(_), Some(b)) => format!("stopped world {b}"),
(Some(_), None) => "stopped".into(),
(None, _) => left(r.expires_at),
let status = match (r.stopped_at, &r.backup, &r.superseded_by) {
(_, _, Some(next)) => format!("failed over to {next}"),
(Some(_), Some(b), _) => format!("stopped world {b}"),
(Some(_), None, _) => "stopped".into(),
(None, _, _) => match &r.latest_backup {
Some(b) => format!("{} backup {}", left(r.expires_at), &b[..12.min(b.len())]),
None => left(r.expires_at),
},
};
println!("{:20} {:10} {:28} {}", r.id, r.game.name(), r.address, status);
}
Ok(())
}
Sub::Watch { id, probe_every_secs, fail_after_secs, no_failover, max_price, worker } => {
let opts = gameserver::WatchOptions {
probe_every: Duration::from_secs(probe_every_secs.clamp(5, 3600)),
fail_after: Duration::from_secs(fail_after_secs.clamp(30, 86_400)),
failover: !no_failover,
max_price_usd: max_price,
workers: worker,
..Default::default()
};
eprintln!("watching {id} (probe every {}s, fail over after {}s down)", opts.probe_every.as_secs(), opts.fail_after.as_secs());
match gameserver::watch(&id, opts).await? {
gameserver::WatchEnd::Ended(r) => println!("{} ended (reservation over or stopped)", r.id),
gameserver::WatchEnd::FailedOver { from, to } => println!("{} failed over to {} at {}", from.id, to.id, to.address),
}
Ok(())
}
Sub::Health { id, last, json } => {
let log = gameserver::health_log(&gameserver::buyer_dir(), &id);
let tail = &log[log.len().saturating_sub(last)..];
if json {
println!("{}", serde_json::to_string(tail)?);
return Ok(());
}
if tail.is_empty() {
println!("no probes recorded for {id} (run `c0mpute gameserver watch {id}`)");
}
for p in tail {
println!("{} {} {:16} {:>5}ms {}", p.ts, if p.ok { "up " } else { "DOWN" }, p.method, p.rtt_ms, p.detail);
}
Ok(())
}
Sub::Stop { id, timeout_secs, json } => {
let r = gameserver::stop(&id, Duration::from_secs(timeout_secs.clamp(30, 3600))).await?;
if json {
Expand Down
85 changes: 4 additions & 81 deletions node/crates/c0mpute-cli/src/main.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2367,62 +2367,6 @@ async fn run_stt(cmd: SttCmd) -> Result<()> {
}
}

/// Build the runner image from a shallow clone of the public repo.
///
/// The fallback for `c0mpute video setup` when the registry will not serve
/// the image. Slower than a pull (it downloads the Python stack rather than
/// finished layers) but it needs no registry account, which is the point.
async fn build_video_runner() -> Result<()> {
use c0mpute_core::video;

for bin in ["docker", "git"] {
if tokio::process::Command::new(bin)
.arg("--version")
.stdout(std::process::Stdio::null())
.stderr(std::process::Stdio::null())
.status()
.await
.is_err()
{
anyhow::bail!("building the runner needs both docker and git; {bin} is missing");
}
}

// A fresh directory per run: a half-finished clone from a previous
// attempt would make `git clone` fail rather than resume.
let checkout = std::env::temp_dir().join(format!("c0mpute-video-src-{}", std::process::id()));
let _ = tokio::fs::remove_dir_all(&checkout).await;

println!("cloning {} (shallow)", video::RUNNER_SOURCE_REPO);
let status = tokio::process::Command::new("git")
.args(["clone", "--depth", "1", video::RUNNER_SOURCE_REPO])
.arg(&checkout)
.status()
.await
.context("could not run git")?;
if !status.success() {
anyhow::bail!("git clone failed with {status}");
}

println!("building {} (several minutes)", video::RUNNER_IMAGE);
let args = video::build_argv(&checkout, video::RUNNER_IMAGE);
let status = tokio::process::Command::new("docker")
.args(&args)
.status()
.await
.context("could not run docker build")?;

// Clean up whether or not the build worked: the checkout is scratch and
// a stale one in /tmp is worse than no cache at all.
let _ = tokio::fs::remove_dir_all(&checkout).await;

if !status.success() {
anyhow::bail!("docker build failed with {status}");
}
println!("built {}", video::RUNNER_IMAGE);
Ok(())
}

/// `c0mpute video` — the surface `plugins/video/module.toml` declares.
///
/// Everything here talks to the runner image rather than to the network: a
Expand All @@ -2433,29 +2377,8 @@ async fn run_video(cmd: VideoCmd) -> Result<()> {

match cmd {
VideoCmd::Setup { build } => {
let mut have_image = false;
if !build {
println!("pulling {} (this is a large image)", video::RUNNER_IMAGE);
let status = tokio::process::Command::new("docker")
.args(["pull", video::RUNNER_IMAGE])
.status()
.await
.context("could not run docker; install a container runtime first")?;
have_image = status.success();
if !have_image {
// A registry is the kind of dependency that is private by
// default, rate-limited or down, and none of those are a
// reason a node with a card cannot serve video. The repo
// is public, so build the same image from source.
println!();
println!("could not pull the runner image (private, rate-limited or offline).");
println!("building it from source instead — same Dockerfile, same result.");
println!();
}
}
if !have_image {
build_video_runner().await?;
}
let how = video::setup(build, &std::env::temp_dir(), &|line: &str| println!("{line}")).await?;
println!("runner image ready as {} ({how:?})", video::RUNNER_IMAGE);
match video::readiness(video::RUNNER_IMAGE) {
None => {
println!("this node can serve video jobs");
Expand All @@ -2482,7 +2405,7 @@ async fn run_video(cmd: VideoCmd) -> Result<()> {
}
if selftest {
println!("running the runner selftest in the container");
let status = tokio::process::Command::new("docker")
let status = c0mpute_core::container::command()
.args(["run", "--rm", video::RUNNER_IMAGE, "selftest"])
.status()
.await
Expand All @@ -2497,7 +2420,7 @@ async fn run_video(cmd: VideoCmd) -> Result<()> {
// The image carries the manifest it was built from, so this is
// the model set that node would actually serve — not what this
// CLI happened to be compiled with.
let out = tokio::process::Command::new("docker")
let out = c0mpute_core::container::command()
.args([
"run",
"--rm",
Expand Down
57 changes: 57 additions & 0 deletions node/crates/c0mpute-core/src/container.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,57 @@
//! The container CLI the container workloads (`video`, `gameserver`) drive.
//!
//! `docker` by default. `C0MPUTE_DOCKER` names another binary with the same
//! command line: `podman`, a wrapper that adds `--context`, or a test double
//! that stands in for the daemon in the loopback end-to-end tests.

use std::ffi::OsString;
use std::path::PathBuf;

/// Environment variable that overrides the container CLI.
pub const ENV: &str = "C0MPUTE_DOCKER";

/// The container CLI to run: `$C0MPUTE_DOCKER`, else `docker`.
pub fn cli() -> OsString {
match std::env::var_os(ENV) {
Some(v) if !v.is_empty() => v,
_ => OsString::from("docker"),
}
}

/// Where [`cli`] resolves: the path itself when it names a file, else the
/// first match on `PATH`. `None` means there is no container runtime.
pub fn which() -> Option<PathBuf> {
let bin = cli();
let as_path = PathBuf::from(&bin);
if as_path.components().count() > 1 {
return as_path.is_file().then_some(as_path);
}
std::env::var_os("PATH").and_then(|paths| {
std::env::split_paths(&paths)
.map(|dir| dir.join(&bin))
.find(|candidate| candidate.is_file())
})
}

/// A `tokio` command for the container CLI.
pub fn command() -> tokio::process::Command {
tokio::process::Command::new(cli())
}

/// A blocking command for the container CLI (readiness probes only).
pub fn command_sync() -> std::process::Command {
std::process::Command::new(cli())
}

#[cfg(test)]
mod tests {
use super::*;

#[test]
fn defaults_to_docker() {
// The variable is not set in the unit-test process.
if std::env::var_os(ENV).is_none() {
assert_eq!(cli(), OsString::from("docker"));
}
}
}
Loading
Loading