Download a model onto the machine that will serve it
The Models tab was about this backend's own disk, which is the wrong disk for every session that runs anywhere else: llama.cpp reads the file where it runs. So the models of a machine live under that machine's llama.cpp provider now, beside the settings deciding how each is loaded, and the download that produces one happens there. A download is a detached `curl` on that machine, started by a script this server writes and never spoken to again. Its state is a file beside the partial, so nothing about it is held here: it survives the app closing, this backend restarting and a second device watching, and the progress is `wc -c` of the partial against the size HuggingFace published rather than anything remembered. A run whose process is gone is reported failed, since `kill -0` is asked at each listing, and there is no "finished" state -- a download that finished is a model, in the list beside the ones still going. Resuming is guarded by the published sha256, which is also checked before the file takes its real name. Two other things the same screens wanted: A provider is drawn as a card rather than as a line of text, bordered against the machine card it sits in -- the tint it had was one step along the surface ladder and rendered as one flat block -- with room to tap and no chevron. Nothing in a raw block wraps any more; the block scrolls sideways instead, one offset for all its lines, so a diff or a column-aligned test run still reads as one. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
iris-aiandClaude Opus 5 committed
1 parent
8c323fc7a9
commit
81c30dcda1
14 files changed
+1224
-1073
No files matched your search
@@ -175,7 +175,6 @@ async fn main() -> Result<()> {
|
||||
let models_dir = args
|
||||
.models_dir
|
||||
.unwrap_or_else(|| data_home("ai-app").join("models"));
|
||||
let models = Arc::new(models::ModelStore::new(models_dir.clone()));
|
||||
let manager = Arc::new(
|
||||
SessionManager::new(config_path.clone(), data_dir, models_dir.clone())
|
||||
.with_context(|| format!("failed to load {}", config_path.display()))?
|
||||
@@ -290,7 +289,6 @@ async fn main() -> Result<()> {
|
||||
Arc::clone(&provider_logins),
|
||||
Arc::clone(&manager),
|
||||
))
|
||||
.merge(routes::models_router(Arc::clone(&models)))
|
||||
.layer(axum::middleware::from_fn_with_state(
|
||||
Arc::clone(&manager),
|
||||
auth::require_token,
|
||||
|
||||
+383
-493
@@ -1,46 +1,45 @@
|
||||
//! GGUF models on this machine, and the downloads that produce them.
|
||||
//! GGUF models on the machines this backend can run them on, and the
|
||||
//! downloads that produce them.
|
||||
//!
|
||||
//! The registry pattern again: one owner, one lock, so what is on disk and what
|
||||
//! this server believes cannot come apart. Three things shape the design, all
|
||||
//! consequences of a model file being gigabytes rather than kilobytes:
|
||||
//! **A model file belongs to the machine that will serve it.** Everything
|
||||
//! here is addressed by a transport and a directory rather than by this
|
||||
//! process's own disk, because `llama-server` reads the file where it runs
|
||||
//! and a list or a download naming anywhere else is a claim about the wrong
|
||||
//! filesystem.
|
||||
//!
|
||||
//! **A download belongs to the model, not to whoever asked for it.** It is
|
||||
//! keyed by the model it produces and lives here, so any device can watch it --
|
||||
//! including one that did not start it. State in a per-connection channel would
|
||||
//! not survive the phone locking its screen, which for an hour-long download is
|
||||
//! the normal case.
|
||||
//! **So a download runs there too** (2026-09-19, replacing the fetch this
|
||||
//! server used to perform itself onto its own disk). It is a detached `curl`
|
||||
//! on that machine, started by a script written here and never spoken to
|
||||
//! again. Three things follow, and all three are why it is shaped this way:
|
||||
//!
|
||||
//! **Every run has an id, and its outcome outlives it.** Without those, "not
|
||||
//! downloading" is three answers at once -- it finished, it never started, or a
|
||||
//! different run finished while you were away.
|
||||
//!
|
||||
//! **Progress is measured, never estimated.** `total` is whatever
|
||||
//! `Content-Length` said and nothing else; when the server does not send one it
|
||||
//! stays `None` and the phone shows that it does not know.
|
||||
//! - **Its state is a file beside the model**, `x.gguf.download`, written by
|
||||
//! that script at each step. Nothing about it is held in this process, so
|
||||
//! one download survives the phone locking, this backend restarting, and
|
||||
//! being watched from a second device -- and what is on that disk is the
|
||||
//! answer rather than something this server remembers about it.
|
||||
//! - **Progress is measured, never estimated.** The bytes are `wc -c` of the
|
||||
//! partial file and the total is the size HuggingFace published for it. A
|
||||
//! download whose process is gone is reported failed rather than left
|
||||
//! saying "running" for ever, which is what `kill -0` at each listing is
|
||||
//! for.
|
||||
//! - **Resume is guarded by identity.** The partial records the sha256 the
|
||||
//! file is meant to have; one written against a different sha is discarded
|
||||
//! rather than resumed onto, and the finished file is checked against it
|
||||
//! before it takes the real name.
|
||||
|
||||
use std::collections::HashMap;
|
||||
use std::io::{Read, Seek, SeekFrom, Write};
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||
use std::sync::{Arc, Mutex};
|
||||
use std::path::Path;
|
||||
|
||||
use anyhow::{Context, Result, bail};
|
||||
use anyhow::{Context, Result};
|
||||
use serde::Serialize;
|
||||
|
||||
use wg_app_link::private;
|
||||
|
||||
use crate::session::transport::{Launch, Transport};
|
||||
|
||||
/// Identifies this client to HuggingFace. They ask for one, and a request
|
||||
/// without it is more likely to be rate-limited.
|
||||
const USER_AGENT: &str = concat!("ai-server/", env!("CARGO_PKG_VERSION"));
|
||||
|
||||
/// Read size per loop iteration. Big enough that the syscall overhead is
|
||||
/// nothing against a multi-gigabyte file, small enough that a cancel is noticed
|
||||
/// promptly -- the flag is only checked between chunks.
|
||||
const CHUNK: usize = 256 * 1024;
|
||||
|
||||
/// A model file sitting on this machine, ready to run.
|
||||
/// A model file sitting on a machine, ready to run there.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct LocalModel {
|
||||
@@ -107,486 +106,367 @@ pub fn labels(models: &[LocalModel]) -> Vec<String> {
|
||||
.collect()
|
||||
}
|
||||
|
||||
/// The model's own name, read out of the file itself.
|
||||
/// What a download is doing, or did.
|
||||
///
|
||||
/// Absent for every way of not finding out -- see [`crate::gguf`]. The file is
|
||||
/// opened and read only as far as the name, which is the first few hundred
|
||||
/// bytes, so this is affordable once per model per listing.
|
||||
fn name_of(path: &Path) -> Option<String> {
|
||||
let mut file = std::fs::File::open(path).ok()?;
|
||||
crate::gguf::name(&mut file)
|
||||
}
|
||||
|
||||
/// What a run is doing, or did. Flat rather than a tagged enum carrying its
|
||||
/// message, because the phone switches on this and a string it can compare is
|
||||
/// easier to render than a variant it has to destructure.
|
||||
/// Flat rather than a tagged enum carrying its message, because the phone
|
||||
/// switches on this and a word it can compare is easier to render than a
|
||||
/// variant it has to destructure. There is no `finished`: a download that
|
||||
/// finished is a model, and it is in the list beside this one -- an outcome
|
||||
/// kept here as well would be the same fact stated twice, in two places that
|
||||
/// could disagree.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum DownloadState {
|
||||
Running,
|
||||
/// Reading the finished file back to check it against the hash HuggingFace
|
||||
/// publishes. Its own state because it takes real time on a multi-gigabyte
|
||||
/// file and "still working" is the honest thing to show, rather than a bar
|
||||
/// sitting at 100% for half a minute.
|
||||
/// Reading the finished file back to check it against the hash
|
||||
/// HuggingFace publishes. Its own state because it takes real time on a
|
||||
/// multi-gigabyte file, and "still working" is the honest thing to show
|
||||
/// where a bar sitting at 100% for half a minute is not.
|
||||
Verifying,
|
||||
Finished,
|
||||
Failed,
|
||||
Cancelled,
|
||||
}
|
||||
|
||||
/// One download run, as the phone sees it.
|
||||
impl DownloadState {
|
||||
fn from_word(word: &str) -> Option<Self> {
|
||||
match word {
|
||||
"running" => Some(Self::Running),
|
||||
"verifying" => Some(Self::Verifying),
|
||||
"failed" => Some(Self::Failed),
|
||||
"cancelled" => Some(Self::Cancelled),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// One download on one machine, as the phone sees it.
|
||||
#[derive(Debug, Clone, Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct DownloadStatus {
|
||||
pub struct Download {
|
||||
/// `owner/repo/file.gguf`, the same key the finished model will have.
|
||||
/// One download per model at a time, so this identifies it.
|
||||
pub key: String,
|
||||
/// Distinguishes this run from any earlier one for the same model, so a
|
||||
/// device that was watching run 3 can tell it is now looking at run 4.
|
||||
pub run: u64,
|
||||
pub repo: String,
|
||||
pub file: String,
|
||||
pub state: DownloadState,
|
||||
pub done: u64,
|
||||
/// What `Content-Length` said, or absent when the server did not say.
|
||||
/// Absent means "unknown", never "zero".
|
||||
/// What HuggingFace published as the file's size, or absent when it
|
||||
/// published none. Absent means "not known", never "zero".
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub total: Option<u64>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub error: Option<String>,
|
||||
pub started: f64,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
pub finished: Option<f64>,
|
||||
}
|
||||
|
||||
/// The mutable half of a run, behind one lock.
|
||||
#[derive(Debug)]
|
||||
struct Progress {
|
||||
state: DownloadState,
|
||||
done: u64,
|
||||
total: Option<u64>,
|
||||
error: Option<String>,
|
||||
started: f64,
|
||||
finished: Option<f64>,
|
||||
/// What HuggingFace says about one file before anything is fetched: how big
|
||||
/// it is, and what it should hash to.
|
||||
///
|
||||
/// Both read here rather than on the machine doing the download, because this
|
||||
/// is the side with internet trust to spend (see [`search`]) and because the
|
||||
/// sha is what makes a resume safe -- the machine is handed the answer rather
|
||||
/// than asked to find it.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct Published {
|
||||
pub bytes: Option<u64>,
|
||||
/// The LFS object id, which for these repositories is the sha256 of the
|
||||
/// content -- a free integrity check rather than a second source of
|
||||
/// truth. Absent for a repository that publishes none, and then the file
|
||||
/// is not verified and a partial is never resumed onto.
|
||||
pub sha256: Option<String>,
|
||||
}
|
||||
|
||||
/// A run, shared between the thread doing the work and everyone watching.
|
||||
struct Run {
|
||||
id: u64,
|
||||
key: String,
|
||||
repo: String,
|
||||
file: String,
|
||||
progress: Mutex<Progress>,
|
||||
/// Set by [`ModelStore::cancel`]; the download loop checks it between
|
||||
/// chunks and stops, leaving the partial file for a later resume.
|
||||
cancel: AtomicBool,
|
||||
}
|
||||
|
||||
impl Run {
|
||||
fn status(&self) -> DownloadStatus {
|
||||
let p = self.progress.lock().unwrap();
|
||||
DownloadStatus {
|
||||
key: self.key.clone(),
|
||||
run: self.id,
|
||||
repo: self.repo.clone(),
|
||||
file: self.file.clone(),
|
||||
state: p.state,
|
||||
done: p.done,
|
||||
total: p.total,
|
||||
error: p.error.clone(),
|
||||
started: p.started,
|
||||
finished: p.finished,
|
||||
/// Refuses a key that is anything but `owner/repo/file.gguf`.
|
||||
///
|
||||
/// The key comes from a phone, so it is treated as hostile: it reaches a
|
||||
/// shell as a positional argument, which is what stops it being read as a
|
||||
/// command, but nothing there stops `../..` from naming a file outside the
|
||||
/// models directory. Rejected rather than rewritten, since a silently
|
||||
/// corrected path downloads the right bytes to the wrong place.
|
||||
fn checked(key: &str) -> Result<()> {
|
||||
let bad = |why: &str| anyhow::anyhow!("\"{key}\" is not a model this can store: {why}");
|
||||
for part in key.split('/') {
|
||||
if part.is_empty() || part == "." || part == ".." {
|
||||
return Err(bad(
|
||||
"a path component that would leave the models directory",
|
||||
));
|
||||
}
|
||||
if part.contains('\\') || part.chars().any(char::is_control) {
|
||||
return Err(bad("a path component with a character no file here has"));
|
||||
}
|
||||
}
|
||||
|
||||
fn finish(&self, state: DownloadState, error: Option<String>) {
|
||||
let mut p = self.progress.lock().unwrap();
|
||||
p.state = state;
|
||||
p.error = error;
|
||||
p.finished = Some(crate::session::now());
|
||||
}
|
||||
}
|
||||
|
||||
/// Every model this machine has, and every download in flight or finished.
|
||||
pub struct ModelStore {
|
||||
dir: PathBuf,
|
||||
/// Keyed by model key: one run per model at a time, and the last run for a
|
||||
/// model stays here after it ends so its outcome can still be read.
|
||||
runs: Mutex<HashMap<String, Arc<Run>>>,
|
||||
next_run: AtomicU64,
|
||||
}
|
||||
|
||||
impl ModelStore {
|
||||
pub fn new(dir: PathBuf) -> Self {
|
||||
Self {
|
||||
dir,
|
||||
runs: Mutex::new(HashMap::new()),
|
||||
next_run: AtomicU64::new(1),
|
||||
}
|
||||
}
|
||||
|
||||
/// Where a model's file lives, refusing anything that would escape the
|
||||
/// models directory.
|
||||
///
|
||||
/// The repo and file come from a phone, so they are treated as hostile:
|
||||
/// every component must be an ordinary name. Rejecting rather than
|
||||
/// sanitising, since a silently rewritten path would download the right
|
||||
/// bytes to the wrong place.
|
||||
fn path_for(&self, repo: &str, file: &str) -> Result<PathBuf> {
|
||||
let mut path = self.dir.clone();
|
||||
for part in repo.split('/').chain(file.split('/')) {
|
||||
if part.is_empty() || part == "." || part == ".." || part.contains('\\') {
|
||||
bail!("\"{repo}/{file}\" is not a name this can store: \"{part}\"");
|
||||
}
|
||||
path.push(part);
|
||||
}
|
||||
Ok(path)
|
||||
}
|
||||
|
||||
pub fn key_for(repo: &str, file: &str) -> String {
|
||||
format!("{repo}/{file}")
|
||||
}
|
||||
|
||||
/// Every `.gguf` found under the models directory, newest first. Read from
|
||||
/// disk on each call rather than cached: a file deleted by hand should stop
|
||||
/// being offered.
|
||||
pub fn list(&self) -> Vec<LocalModel> {
|
||||
let mut found = Vec::new();
|
||||
collect(&self.dir, &self.dir, &mut found);
|
||||
found.sort_by(|a, b| a.key.cmp(&b.key));
|
||||
found
|
||||
}
|
||||
|
||||
/// The status of every run this server remembers.
|
||||
pub fn downloads(&self) -> Vec<DownloadStatus> {
|
||||
let runs = self.runs.lock().unwrap();
|
||||
let mut all: Vec<_> = runs.values().map(|run| run.status()).collect();
|
||||
all.sort_by_key(|status| std::cmp::Reverse(status.run));
|
||||
all
|
||||
}
|
||||
|
||||
/// Starts fetching `file` from `repo`, or returns the run already doing so.
|
||||
/// Idempotent on purpose: a phone that lost its connection will press the
|
||||
/// button again, and that must join the existing run rather than start a
|
||||
/// second one writing the same file.
|
||||
pub fn start(self: &Arc<Self>, repo: &str, file: &str) -> Result<DownloadStatus> {
|
||||
let key = Self::key_for(repo, file);
|
||||
let target = self.path_for(repo, file)?;
|
||||
if target.is_file() {
|
||||
bail!("{key} is already downloaded");
|
||||
}
|
||||
|
||||
let mut runs = self.runs.lock().unwrap();
|
||||
if let Some(existing) = runs.get(&key)
|
||||
&& existing.progress.lock().unwrap().state == DownloadState::Running
|
||||
{
|
||||
return Ok(existing.status());
|
||||
}
|
||||
|
||||
let run = Arc::new(Run {
|
||||
id: self.next_run.fetch_add(1, Ordering::Relaxed),
|
||||
key: key.clone(),
|
||||
repo: repo.to_string(),
|
||||
file: file.to_string(),
|
||||
progress: Mutex::new(Progress {
|
||||
state: DownloadState::Running,
|
||||
done: 0,
|
||||
total: None,
|
||||
error: None,
|
||||
started: crate::session::now(),
|
||||
finished: None,
|
||||
}),
|
||||
cancel: AtomicBool::new(false),
|
||||
});
|
||||
runs.insert(key, Arc::clone(&run));
|
||||
let status = run.status();
|
||||
drop(runs);
|
||||
|
||||
// A dedicated thread rather than the blocking pool: this holds its
|
||||
// thread for as long as the download takes, which is minutes to hours,
|
||||
// and the pool exists for short work.
|
||||
let store = Arc::clone(self);
|
||||
std::thread::spawn(move || {
|
||||
let outcome = store.fetch(&run, &target);
|
||||
match outcome {
|
||||
Ok(()) if run.cancel.load(Ordering::Relaxed) => {
|
||||
run.finish(DownloadState::Cancelled, None);
|
||||
tracing::info!("download {} cancelled", run.key);
|
||||
}
|
||||
Ok(()) => {
|
||||
run.finish(DownloadState::Finished, None);
|
||||
tracing::info!("download {} finished", run.key);
|
||||
}
|
||||
Err(err) => {
|
||||
let message = format!("{err:#}");
|
||||
tracing::warn!("download {} failed: {message}", run.key);
|
||||
run.finish(DownloadState::Failed, Some(message));
|
||||
}
|
||||
}
|
||||
});
|
||||
Ok(status)
|
||||
}
|
||||
|
||||
/// Asks a running download to stop. The partial file stays, so starting
|
||||
/// again resumes rather than refetching.
|
||||
pub fn cancel(&self, key: &str) -> Result<DownloadStatus> {
|
||||
let runs = self.runs.lock().unwrap();
|
||||
let Some(run) = runs.get(key) else {
|
||||
bail!("no download for {key}");
|
||||
};
|
||||
run.cancel.store(true, Ordering::Relaxed);
|
||||
Ok(run.status())
|
||||
}
|
||||
|
||||
/// Removes a downloaded model, and any partial file for it.
|
||||
pub fn delete(&self, key: &str) -> Result<()> {
|
||||
let (repo, file) = key.rsplit_once('/').context("a key is repo/file")?;
|
||||
let target = self.path_for(repo, file)?;
|
||||
let partial = partial_of(&target);
|
||||
if !target.is_file() && !partial.is_file() {
|
||||
bail!("{key} is not downloaded");
|
||||
}
|
||||
for path in [&target, &partial] {
|
||||
if path.is_file() {
|
||||
std::fs::remove_file(path).with_context(|| format!("remove {}", path.display()))?;
|
||||
}
|
||||
}
|
||||
self.runs.lock().unwrap().remove(key);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn fetch(&self, run: &Run, target: &Path) -> Result<()> {
|
||||
let partial = partial_of(target);
|
||||
let identity = identity_of(target);
|
||||
if let Some(parent) = target.parent() {
|
||||
private::create_dir(parent)?;
|
||||
}
|
||||
|
||||
// What we have, and what it was part of. A partial with no recorded
|
||||
// identity is not resumable -- it could be a fragment of any revision.
|
||||
let known = std::fs::read_to_string(&identity)
|
||||
.ok()
|
||||
.map(|s| s.trim().to_string());
|
||||
let have = match known {
|
||||
Some(_) => partial.metadata().map(|m| m.len()).unwrap_or(0),
|
||||
None => 0,
|
||||
};
|
||||
|
||||
let url = format!(
|
||||
"https://huggingface.co/{}/resolve/main/{}",
|
||||
run.repo,
|
||||
run.file.replace(' ', "%20")
|
||||
);
|
||||
let (mut response, mut resumed) = request(&url, have)?;
|
||||
let mut etag = etag_of(&response);
|
||||
|
||||
// HuggingFace's CDN ignores `If-Range` -- probed 2026-08-28: a
|
||||
// deliberately stale validator still answers 206 with the ranged bytes.
|
||||
// So the header cannot be relied on to restart us, and the check is done
|
||||
// here instead: if what arrived is not the revision our partial belongs
|
||||
// to, resuming would splice two files into something of exactly the
|
||||
// right length and the wrong contents.
|
||||
if resumed && etag.is_some() && etag != known {
|
||||
tracing::info!(
|
||||
"{} changed upstream since the partial was written -- starting again",
|
||||
run.key,
|
||||
);
|
||||
let (fresh, fresh_resumed) = request(&url, 0)?;
|
||||
response = fresh;
|
||||
resumed = fresh_resumed;
|
||||
etag = etag_of(&response);
|
||||
}
|
||||
|
||||
// On a 206, Content-Length is the length of the *range*, not of the file
|
||||
// -- taken at face value it would fill the bar at 72 MB of a 234 MB
|
||||
// model. The whole size is the last field of Content-Range, which has
|
||||
// the further merit of not depending on where the range began.
|
||||
let total: Option<u64> = if resumed {
|
||||
response
|
||||
.headers()
|
||||
.get("content-range")
|
||||
.and_then(|v| v.to_str().ok())
|
||||
.and_then(|v| {
|
||||
v.rsplit_once('/')
|
||||
.map(|(_, whole)| whole.trim().to_string())
|
||||
})
|
||||
.and_then(|whole| whole.parse().ok())
|
||||
} else {
|
||||
response
|
||||
.headers()
|
||||
.get("content-length")
|
||||
.and_then(|v| v.to_str().ok()?.parse().ok())
|
||||
};
|
||||
let mut done = if resumed { have } else { 0 };
|
||||
{
|
||||
let mut p = run.progress.lock().unwrap();
|
||||
p.done = done;
|
||||
p.total = total;
|
||||
}
|
||||
|
||||
// `truncate(false)` is the whole resume story: the file is opened to be
|
||||
// seeked into and appended to, and truncating would throw away exactly
|
||||
// the bytes the Range request just asked the server not to send again.
|
||||
// Stated rather than left to the default.
|
||||
let mut file = std::fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.write(true)
|
||||
.truncate(false)
|
||||
.open(&partial)
|
||||
.with_context(|| format!("open {}", partial.display()))?;
|
||||
if resumed {
|
||||
file.seek(SeekFrom::Start(have))
|
||||
.context("seek to resume point")?;
|
||||
} else {
|
||||
file.set_len(0)
|
||||
.context("truncate a partial we cannot resume onto")?;
|
||||
}
|
||||
// Written before the body, so an interrupted download leaves a partial
|
||||
// that can still say which revision it belongs to.
|
||||
if let Some(etag) = &etag {
|
||||
std::fs::write(&identity, etag).ok();
|
||||
}
|
||||
|
||||
let mut reader = response.body_mut().as_reader();
|
||||
let mut buffer = vec![0u8; CHUNK];
|
||||
loop {
|
||||
if run.cancel.load(Ordering::Relaxed) {
|
||||
file.flush().ok();
|
||||
return Ok(());
|
||||
}
|
||||
let read = reader
|
||||
.read(&mut buffer)
|
||||
.context("reading from HuggingFace")?;
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
file.write_all(&buffer[..read])
|
||||
.context("writing the model file")?;
|
||||
done += read as u64;
|
||||
run.progress.lock().unwrap().done = done;
|
||||
}
|
||||
file.flush().context("flushing the model file")?;
|
||||
drop(file);
|
||||
|
||||
// Checked before the rename, so a file that fails never gets the real
|
||||
// name and `list` never offers it. With the identity check above this
|
||||
// should not fire; it is here because a wrong model is the kind of
|
||||
// failure that surfaces as bad output rather than as an error.
|
||||
if let Some(expected) = published_sha256(&run.repo, &run.file) {
|
||||
run.progress.lock().unwrap().state = DownloadState::Verifying;
|
||||
let actual = sha256_of(&partial)?;
|
||||
if actual != expected {
|
||||
std::fs::remove_file(&partial).ok();
|
||||
std::fs::remove_file(&identity).ok();
|
||||
bail!(
|
||||
"{} arrived corrupted -- HuggingFace publishes sha256 {expected}, what \
|
||||
arrived hashes to {actual}. It has been deleted; downloading again \
|
||||
starts clean.",
|
||||
run.key,
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
// Renamed only once complete, so a file at its real name is always a
|
||||
// whole model -- `list` needs no other way to tell.
|
||||
std::fs::rename(&partial, target)
|
||||
.with_context(|| format!("finish {}", target.display()))?;
|
||||
std::fs::remove_file(&identity).ok();
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// The sha256 of a file, read in chunks -- these are gigabytes, and reading one
|
||||
/// into memory to hash it would be the largest allocation this server makes.
|
||||
fn sha256_of(path: &Path) -> Result<String> {
|
||||
use sha2::{Digest, Sha256};
|
||||
let mut file =
|
||||
std::fs::File::open(path).with_context(|| format!("reopen {}", path.display()))?;
|
||||
let mut hasher = Sha256::new();
|
||||
let mut buffer = vec![0u8; CHUNK];
|
||||
loop {
|
||||
let read = file.read(&mut buffer).context("reading back to verify")?;
|
||||
if read == 0 {
|
||||
break;
|
||||
}
|
||||
hasher.update(&buffer[..read]);
|
||||
}
|
||||
// Hex by hand, as `wg_app_link::enroll::token_hash_hex` also has to, since
|
||||
// this sha2 version's output type does not implement LowerHex.
|
||||
Ok(hasher
|
||||
.finalize()
|
||||
.iter()
|
||||
.map(|byte| format!("{byte:02x}"))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// One GET, ranged when there is something to resume onto.
|
||||
fn request(url: &str, from: u64) -> Result<(ureq::http::Response<ureq::Body>, bool)> {
|
||||
let mut get = ureq::get(url).header("User-Agent", USER_AGENT);
|
||||
if from > 0 {
|
||||
get = get.header("Range", &format!("bytes={from}-"));
|
||||
}
|
||||
let response = get.call().with_context(|| format!("GET {url}"))?;
|
||||
// Trust the status, not the request: a server that ignores Range answers
|
||||
// 200 with the whole file, and appending to that would corrupt it.
|
||||
let resumed = response.status() == 206;
|
||||
Ok((response, resumed))
|
||||
}
|
||||
|
||||
fn etag_of(response: &ureq::http::Response<ureq::Body>) -> Option<String> {
|
||||
Some(
|
||||
response
|
||||
.headers()
|
||||
.get("etag")?
|
||||
.to_str()
|
||||
.ok()?
|
||||
.trim()
|
||||
.to_string(),
|
||||
)
|
||||
}
|
||||
|
||||
/// `x.gguf` -> `x.gguf.part.etag`, holding which revision the partial beside it
|
||||
/// is a piece of.
|
||||
fn identity_of(target: &Path) -> PathBuf {
|
||||
let mut name = target.as_os_str().to_os_string();
|
||||
name.push(".part.etag");
|
||||
PathBuf::from(name)
|
||||
}
|
||||
|
||||
/// `x.gguf` -> `x.gguf.part`, the in-progress name.
|
||||
fn partial_of(target: &Path) -> PathBuf {
|
||||
let mut name = target.as_os_str().to_os_string();
|
||||
name.push(".part");
|
||||
PathBuf::from(name)
|
||||
}
|
||||
|
||||
/// Walks `dir` collecting `.gguf` files, keyed by their path under `root`.
|
||||
fn collect(root: &Path, dir: &Path, found: &mut Vec<LocalModel>) {
|
||||
let Ok(entries) = std::fs::read_dir(dir) else {
|
||||
return;
|
||||
let Some((_, file)) = key.rsplit_once('/') else {
|
||||
return Err(bad("a key is owner/repo/file.gguf"));
|
||||
};
|
||||
for entry in entries.flatten() {
|
||||
let path = entry.path();
|
||||
if path.is_dir() {
|
||||
collect(root, &path, found);
|
||||
continue;
|
||||
}
|
||||
if path.extension().is_none_or(|e| e != "gguf") {
|
||||
continue;
|
||||
}
|
||||
let Ok(relative) = path.strip_prefix(root) else {
|
||||
continue;
|
||||
};
|
||||
let key = relative.to_string_lossy().replace('\\', "/");
|
||||
let Some((repo, file)) = key.rsplit_once('/') else {
|
||||
continue;
|
||||
};
|
||||
found.push(LocalModel {
|
||||
key: key.clone(),
|
||||
repo: repo.to_string(),
|
||||
file: file.to_string(),
|
||||
bytes: entry.metadata().map(|m| m.len()).unwrap_or(0),
|
||||
name: name_of(&path),
|
||||
});
|
||||
if !file.ends_with(".gguf") {
|
||||
return Err(bad("only .gguf files are models"));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn key_for(repo: &str, file: &str) -> String {
|
||||
format!("{repo}/{file}")
|
||||
}
|
||||
|
||||
/// Expands a leading `~` the way that machine's own shell would, which is the
|
||||
/// only place that knows what it stands for. Prefixed to every script here;
|
||||
/// each takes the directory as `$1`.
|
||||
const EXPAND: &str = "d=$1; case $d in \"~\") d=$HOME;; \"~/\"*) d=$HOME/${d#\"~/\"};; esac;";
|
||||
|
||||
/// The download itself, run detached on the machine that will hold the file.
|
||||
///
|
||||
/// Passed to [`START`] as an argument rather than written inside it, so that
|
||||
/// neither script has to quote the other: `sh -c "$worker"` takes this whole
|
||||
/// text as one string the outer shell never parses.
|
||||
///
|
||||
/// `-sS` rather than plain: curl's progress meter goes to stderr, and the
|
||||
/// same file is what a failure is read out of -- a message buried in three
|
||||
/// screens of redrawn bar is one the phone cannot show. Progress is read off
|
||||
/// the partial file instead, which is the measurement rather than a report
|
||||
/// of it.
|
||||
///
|
||||
/// The state file is written whole and moved into place, because a listing
|
||||
/// can arrive in the middle of any of these writes and a truncated one is a
|
||||
/// download that blinks out of the phone's list for a poll.
|
||||
///
|
||||
/// `curl` is waited on rather than run in the foreground, because a cancel
|
||||
/// arrives as a signal and a shell blocked in `wait` is the one that can run
|
||||
/// a trap for it. The trap writes nothing when the partial has already gone:
|
||||
/// that is how [`REMOVE`] deletes a running download without the dying worker
|
||||
/// recreating the state file behind it.
|
||||
const WORKER: &str = r#"
|
||||
t=$1; url=$2; sha=$3; total=$4; p=$t.part; s=$t.download; e=$t.part.err
|
||||
st() { printf 'pid=%s\nsha=%s\ntotal=%s\nstate=%s\nerror=%s\n' "$$" "$sha" "$total" "$1" "$2" > "$s.tmp" && mv "$s.tmp" "$s"; }
|
||||
trap '[ -n "$c" ] && kill "$c" 2>/dev/null; [ -f "$p" ] && st cancelled ""; exit 0' TERM INT
|
||||
c=
|
||||
st running ""
|
||||
have=$(wc -c 2>/dev/null < "$p" | tr -d " ")
|
||||
if [ -z "$total" ] || [ "$have" != "$total" ]; then
|
||||
curl -fsSL --retry 3 --retry-delay 2 -C - -o "$p" "$url" 2> "$e" &
|
||||
c=$!
|
||||
if ! wait "$c"; then
|
||||
st failed "$(tr -s '\n\t\r' ' ' < "$e" | tail -c 300)"
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
rm -f "$e"
|
||||
if [ -n "$sha" ] && command -v sha256sum > /dev/null 2>&1; then
|
||||
st verifying ""
|
||||
got=$(sha256sum "$p" | cut -d' ' -f1)
|
||||
if [ "$got" != "$sha" ]; then
|
||||
rm -f "$p"
|
||||
st failed "it arrived corrupted -- HuggingFace publishes sha256 $sha, what arrived hashes to $got. It has been deleted; downloading again starts clean."
|
||||
exit 1
|
||||
fi
|
||||
fi
|
||||
mv "$p" "$t" && rm -f "$s"
|
||||
"#;
|
||||
|
||||
/// Starts [`WORKER`], or leaves alone whatever is already happening.
|
||||
///
|
||||
/// Idempotent on purpose, in both directions: a phone that lost its
|
||||
/// connection presses the button again, and that must join the run already
|
||||
/// going rather than start a second one writing the same file. A model
|
||||
/// already downloaded is likewise the answer rather than an error, which is
|
||||
/// the ordinary case once two devices can both see the list.
|
||||
const START: &str = r#"
|
||||
k=$2; url=$3; sha=$4; total=$5; worker=$6
|
||||
t=$d/$k; s=$t.download
|
||||
[ -f "$t" ] && exit 0
|
||||
pid=$(sed -n 's/^pid=//p' "$s" 2>/dev/null)
|
||||
if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then exit 0; fi
|
||||
command -v curl > /dev/null 2>&1 || { echo "there is no curl on this machine to download with" >&2; exit 1; }
|
||||
mkdir -p "${t%/*}" || { echo "cannot make ${t%/*} to download into" >&2; exit 1; }
|
||||
had=$(sed -n 's/^sha=//p' "$s" 2>/dev/null)
|
||||
if [ -z "$sha" ] || [ "$had" != "$sha" ]; then rm -f "$t.part"; fi
|
||||
nohup sh -c "$worker" sh "$t" "$url" "$sha" "$total" > /dev/null 2>&1 &
|
||||
"#;
|
||||
|
||||
/// Every download the machine is holding state for, with the bytes read off
|
||||
/// the partial file rather than from anything remembered here.
|
||||
///
|
||||
/// `kill -0` is why a stopped machine does not leave a download claiming to
|
||||
/// be running for ever: a state file whose process is gone is the failure it
|
||||
/// actually is.
|
||||
///
|
||||
/// A state file beside a finished model is skipped, because the file is the
|
||||
/// truth and the state file is only ever a report about producing it. That is
|
||||
/// also what keeps the one race here self-correcting: two devices asking for
|
||||
/// the same model in the same second get two workers writing one partial,
|
||||
/// whose sha will not match -- and whichever of them loses the rename would
|
||||
/// otherwise leave a failure recorded against a model that is sitting there.
|
||||
const LIST: &str = r#"
|
||||
[ -d "$d" ] || exit 0
|
||||
cd "$d" || exit 0
|
||||
find . -type f -name '*.download' | while read -r f; do
|
||||
k=${f#./}; k=${k%.download}
|
||||
[ -f "$k" ] && continue
|
||||
size=$(wc -c 2>/dev/null < "$k.part" | tr -d " ")
|
||||
[ -n "$size" ] || size=0
|
||||
pid=$(sed -n 's/^pid=//p' "$f")
|
||||
alive=no
|
||||
if [ -n "$pid" ] && kill -0 "$pid" 2>/dev/null; then alive=yes; fi
|
||||
printf '%s\037%s\037%s\037%s\000' "$size" "$alive" "$(tr '\n' '\036' < "$f")" "$k"
|
||||
done
|
||||
"#;
|
||||
|
||||
/// Asks a running download to stop; the partial stays, so starting again
|
||||
/// resumes rather than refetching.
|
||||
///
|
||||
/// A record whose process has already gone is written off here rather than
|
||||
/// left to be reported as a failure, because somebody asking for it to stop
|
||||
/// and it being stopped is not a fault.
|
||||
const CANCEL: &str = r#"
|
||||
k=$2; s=$d/$k.download
|
||||
pid=$(sed -n 's/^pid=//p' "$s" 2>/dev/null)
|
||||
[ -n "$pid" ] || { echo "nothing is downloading $k" >&2; exit 1; }
|
||||
if ! kill -TERM "$pid" 2>/dev/null; then
|
||||
sed 's/^state=.*/state=cancelled/' "$s" > "$s.tmp" && mv "$s.tmp" "$s"
|
||||
fi
|
||||
"#;
|
||||
|
||||
/// Takes a model off a machine: the file, a partial, and the state beside it.
|
||||
///
|
||||
/// The partial goes *before* the signal, so that a worker dying from it finds
|
||||
/// nothing to write about and leaves no state file behind -- see [`WORKER`]'s
|
||||
/// trap.
|
||||
const REMOVE: &str = r#"
|
||||
k=$2; t=$d/$k
|
||||
if [ ! -f "$t" ] && [ ! -f "$t.part" ] && [ ! -f "$t.download" ]; then
|
||||
echo "$k is not on this machine" >&2
|
||||
exit 1
|
||||
fi
|
||||
rm -f "$t.part"
|
||||
pid=$(sed -n 's/^pid=//p' "$t.download" 2>/dev/null)
|
||||
[ -n "$pid" ] && kill -TERM "$pid" 2>/dev/null
|
||||
rm -f "$t" "$t.part.err" "$t.download" "$t.download.tmp"
|
||||
# The repository's own directory, and only that one: it was made by the
|
||||
# download and is litter once empty. Never its parent, which for a key with
|
||||
# one component would be the models directory itself.
|
||||
rmdir "${t%/*}" 2>/dev/null
|
||||
exit 0
|
||||
"#;
|
||||
|
||||
/// Runs one of the scripts above on `transport`, against `dir`.
|
||||
async fn run(transport: &Transport, script: &str, dir: &str, rest: &[&str]) -> Result<String> {
|
||||
let mut args = vec![
|
||||
"-c".to_string(),
|
||||
format!("{EXPAND}{script}"),
|
||||
"sh".to_string(),
|
||||
dir.to_string(),
|
||||
];
|
||||
args.extend(rest.iter().map(|arg| (*arg).to_string()));
|
||||
transport.capture(&Launch::new("sh", args, None)).await
|
||||
}
|
||||
|
||||
/// Starts fetching `file` from `repo` onto the machine `transport` reaches.
|
||||
pub async fn start(
|
||||
transport: &Transport,
|
||||
dir: &str,
|
||||
repo: &str,
|
||||
file: &str,
|
||||
published: &Published,
|
||||
) -> Result<()> {
|
||||
let key = key_for(repo, file);
|
||||
checked(&key)?;
|
||||
let url = format!(
|
||||
"https://huggingface.co/{repo}/resolve/main/{}",
|
||||
file.replace(' ', "%20")
|
||||
);
|
||||
run(
|
||||
transport,
|
||||
START,
|
||||
dir,
|
||||
&[
|
||||
&key,
|
||||
&url,
|
||||
published.sha256.as_deref().unwrap_or(""),
|
||||
&published.bytes.map(|b| b.to_string()).unwrap_or_default(),
|
||||
WORKER,
|
||||
],
|
||||
)
|
||||
.await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Every download in flight or stopped on that machine, in key order --
|
||||
/// which is the order the models they become are listed in.
|
||||
pub async fn downloads(transport: &Transport, dir: &str) -> Result<Vec<Download>> {
|
||||
let out = run(transport, LIST, dir, &[]).await?;
|
||||
let mut found: Vec<Download> = out
|
||||
.split('\0')
|
||||
.filter(|record| !record.is_empty())
|
||||
// The key last, so a name with a separator in it cannot eat a field,
|
||||
// and the state file's own lines separated by RS inside field three.
|
||||
.filter_map(|record| {
|
||||
let mut fields = record.splitn(4, '\u{1f}');
|
||||
let done: u64 = fields.next()?.trim().parse().unwrap_or(0);
|
||||
let alive = fields.next()? == "yes";
|
||||
let status: HashMap<&str, &str> = fields
|
||||
.next()?
|
||||
.split('\u{1e}')
|
||||
.filter_map(|line| line.split_once('='))
|
||||
.collect();
|
||||
let key = fields.next()?.trim().to_string();
|
||||
let (repo, file) = key.rsplit_once('/')?;
|
||||
let recorded = status
|
||||
.get("state")
|
||||
.and_then(|word| DownloadState::from_word(word.trim()));
|
||||
let error = status
|
||||
.get("error")
|
||||
.map(|text| text.trim())
|
||||
.filter(|text| !text.is_empty())
|
||||
.map(str::to_string);
|
||||
// A word this server does not know means a state file written by
|
||||
// a different version of the script. Whether there is a process
|
||||
// behind it is measured either way, so that is what it is
|
||||
// reported as -- dropping the record instead would make a
|
||||
// download that is using disk invisible.
|
||||
let recorded = recorded.unwrap_or(if alive {
|
||||
DownloadState::Running
|
||||
} else {
|
||||
DownloadState::Failed
|
||||
});
|
||||
// A record that says it is working and has no process behind it
|
||||
// is the one case this has to correct: the machine was restarted,
|
||||
// or the worker was killed by something other than a cancel.
|
||||
let stopped = matches!(recorded, DownloadState::Running | DownloadState::Verifying);
|
||||
Some(Download {
|
||||
state: if stopped && !alive {
|
||||
DownloadState::Failed
|
||||
} else {
|
||||
recorded
|
||||
},
|
||||
error: error.or_else(|| {
|
||||
(stopped && !alive).then(|| {
|
||||
"it stopped before it finished -- the machine may have been restarted. \
|
||||
Downloading again carries on from here."
|
||||
.to_string()
|
||||
})
|
||||
}),
|
||||
done,
|
||||
total: status.get("total").and_then(|t| t.trim().parse().ok()),
|
||||
repo: repo.to_string(),
|
||||
file: file.to_string(),
|
||||
key,
|
||||
})
|
||||
})
|
||||
.collect();
|
||||
found.sort_by(|a, b| a.key.cmp(&b.key));
|
||||
Ok(found)
|
||||
}
|
||||
|
||||
/// Asks the machine to stop one download. The partial stays behind.
|
||||
pub async fn cancel(transport: &Transport, dir: &str, key: &str) -> Result<()> {
|
||||
checked(key)?;
|
||||
run(transport, CANCEL, dir, &[key]).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Removes a model from a machine, downloaded or half-downloaded.
|
||||
pub async fn remove(transport: &Transport, dir: &str, key: &str) -> Result<()> {
|
||||
checked(key)?;
|
||||
run(transport, REMOVE, dir, &[key]).await?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Where a machine reached over ssh keeps its models, when its machine does
|
||||
@@ -707,8 +587,8 @@ pub struct RemoteRepo {
|
||||
pub struct RemoteFile {
|
||||
pub path: String,
|
||||
pub bytes: u64,
|
||||
/// Already on this machine, so the phone can say so rather than offering to
|
||||
/// fetch it again.
|
||||
/// Already on the machine being looked at, so the phone can say so rather
|
||||
/// than offering to fetch it again.
|
||||
pub have: bool,
|
||||
}
|
||||
|
||||
@@ -745,27 +625,37 @@ pub fn search(query: &str) -> Result<Vec<RemoteRepo>> {
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// The sha256 HuggingFace publishes for one file, if it publishes one. It is
|
||||
/// the LFS object id, which for these repositories is the sha256 of the content
|
||||
/// -- so it is a free integrity check rather than a second source of truth.
|
||||
fn published_sha256(repo: &str, file: &str) -> Option<String> {
|
||||
/// Asks HuggingFace for both halves of [`Published`] in the one call they
|
||||
/// come from.
|
||||
pub fn published(repo: &str, file: &str) -> Result<Published> {
|
||||
let url = format!("https://huggingface.co/api/models/{repo}/tree/main?expand=true");
|
||||
let body = get_json(&url).ok()?;
|
||||
body.as_array()?.iter().find_map(|f| {
|
||||
(f.get("path")?.as_str()? == file)
|
||||
.then(|| f.get("lfs")?.get("oid")?.as_str().map(str::to_string))?
|
||||
let body = get_json(&url)?;
|
||||
let entry = body
|
||||
.as_array()
|
||||
.and_then(|files| {
|
||||
files
|
||||
.iter()
|
||||
.find(|f| f.get("path").and_then(serde_json::Value::as_str) == Some(file))
|
||||
})
|
||||
.with_context(|| format!("{repo} does not have a file called {file}"))?;
|
||||
Ok(Published {
|
||||
bytes: entry.get("size").and_then(serde_json::Value::as_u64),
|
||||
sha256: entry
|
||||
.get("lfs")
|
||||
.and_then(|lfs| lfs.get("oid"))
|
||||
.and_then(serde_json::Value::as_str)
|
||||
.map(str::to_string),
|
||||
})
|
||||
}
|
||||
|
||||
/// The GGUF files in one repository, largest last, with the ones already
|
||||
/// downloaded marked.
|
||||
pub fn files(repo: &str, store: &ModelStore) -> Result<Vec<RemoteFile>> {
|
||||
pub fn files(repo: &str, have: &std::collections::HashSet<String>) -> Result<Vec<RemoteFile>> {
|
||||
let url = format!("https://huggingface.co/api/models/{repo}/tree/main");
|
||||
let body = get_json(&url)?;
|
||||
let list = body
|
||||
.as_array()
|
||||
.context("HuggingFace returned something that is not a list")?;
|
||||
let have: std::collections::HashSet<String> = store.list().into_iter().map(|m| m.key).collect();
|
||||
let mut files: Vec<RemoteFile> = list
|
||||
.iter()
|
||||
.filter_map(|f| {
|
||||
@@ -778,7 +668,7 @@ pub fn files(repo: &str, store: &ModelStore) -> Result<Vec<RemoteFile>> {
|
||||
.get("size")
|
||||
.and_then(serde_json::Value::as_u64)
|
||||
.unwrap_or(0),
|
||||
have: have.contains(&ModelStore::key_for(repo, &path)),
|
||||
have: have.contains(&key_for(repo, &path)),
|
||||
path,
|
||||
})
|
||||
})
|
||||
|
||||
+113
-59
@@ -18,6 +18,12 @@
|
||||
//! GET /machines/{id}/providers/{provider}/auth/{attempt} sign-in state
|
||||
//! POST /machines/{id}/providers/{provider}/auth/{attempt}/code submit browser code
|
||||
//! DELETE /machines/{id}/providers/{provider}/auth/{attempt} cancel sign-in
|
||||
//! GET /machines/{id}/models its GGUFs, and what is being fetched onto it
|
||||
//! GET /machines/{id}/models/files?repo=R the GGUFs in one repository, marked
|
||||
//! with the ones that machine already has
|
||||
//! POST /machines/{id}/models/download {repo, file}; joins the run already going
|
||||
//! POST /machines/{id}/models/cancel {key}; the partial stays, so starting again resumes
|
||||
//! POST /machines/{id}/models/delete {key}; the model, a partial, and the state beside it
|
||||
//! GET /machines/{id}/dir?path=P entries of directory P, and P resolved
|
||||
//! GET /machines/{id}/file?path=P content of file P, or why not
|
||||
//! PUT /machines/{id}/file {path, content, ifSha256} -> new size/mtime/sha256
|
||||
@@ -79,12 +85,7 @@
|
||||
//! GET /defaults {effort} -- what a new session starts at
|
||||
//! POST /defaults {effort} -- null for the CLI's own default
|
||||
//! GET /usage cached usage windows per provider
|
||||
//! GET /models downloaded GGUFs, and what is being fetched
|
||||
//! GET /models/search?q=Q HuggingFace repositories matching Q
|
||||
//! GET /models/files?repo=R the GGUFs in one repository
|
||||
//! POST /models/download {repo, file}; rejoins the run already going
|
||||
//! POST /models/cancel {key}; the partial stays, so starting again resumes
|
||||
//! POST /models/delete {key}
|
||||
//! ```
|
||||
//!
|
||||
//! Everything here works purely in the common event model; nothing may
|
||||
@@ -137,6 +138,19 @@ pub fn router(manager: Arc<SessionManager>) -> Router {
|
||||
"/machines/{id}",
|
||||
get(read_machine).put(update_machine).delete(delete_machine),
|
||||
)
|
||||
// The GGUFs on a machine, and the downloads putting them there. Under
|
||||
// the machine because that is whose disk they are on: the file has to
|
||||
// be where `llama-server` will read it, and one backend serves
|
||||
// several machines. Keys are `owner/repo/file.gguf` and so contain
|
||||
// slashes, which is why none of these puts one in the path -- a key
|
||||
// travels in the body or a query string, and the routes stay
|
||||
// addressable without escaping rules nobody would get right.
|
||||
.route("/machines/{id}/models", get(machine_models))
|
||||
.route("/machines/{id}/models/files", get(repo_files))
|
||||
.route("/machines/{id}/models/download", post(start_download))
|
||||
.route("/machines/{id}/models/cancel", post(cancel_download))
|
||||
.route("/machines/{id}/models/delete", post(delete_model))
|
||||
.route("/models/search", get(search_models))
|
||||
// The models on a configured machine, for a llama session there.
|
||||
.route(
|
||||
"/machines/{id}/providers/{provider}/models",
|
||||
@@ -325,6 +339,12 @@ struct MachineInfo {
|
||||
struct ProviderInfo {
|
||||
name: String,
|
||||
kind: crate::config::DriverKind,
|
||||
/// The program discovery found, which is the honest answer to "what is
|
||||
/// this" and is not something the phone may change -- see `machines`.
|
||||
/// Shown on the provider's card, so that two machines offering the same
|
||||
/// provider from different builds say so.
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
command: Option<String>,
|
||||
models: Vec<String>,
|
||||
permission_modes: Vec<&'static str>,
|
||||
#[serde(skip_serializing_if = "Option::is_none")]
|
||||
@@ -351,6 +371,7 @@ fn info_for(machine: crate::config::MachineConfig) -> MachineInfo {
|
||||
.map(|provider| ProviderInfo {
|
||||
name: provider.name,
|
||||
kind: provider.kind,
|
||||
command: provider.command,
|
||||
models: provider.models,
|
||||
permission_modes: provider.kind.permission_modes().to_vec(),
|
||||
default_permission_mode: provider.kind.default_permission_mode(),
|
||||
@@ -458,6 +479,7 @@ async fn probe_machine(
|
||||
.map(|provider| ProviderInfo {
|
||||
name: provider.name,
|
||||
kind: provider.kind,
|
||||
command: provider.command,
|
||||
models: provider.models,
|
||||
permission_modes: provider.kind.permission_modes().to_vec(),
|
||||
default_permission_mode: provider.kind.default_permission_mode(),
|
||||
@@ -733,7 +755,7 @@ async fn provider_view(
|
||||
machine: machine.name.clone(),
|
||||
name: provider.name.clone(),
|
||||
kind: provider.kind,
|
||||
command: provider.command.clone(),
|
||||
command: provider.command,
|
||||
model_params: provider.kind.model_params(),
|
||||
max_loaded: provider.max_loaded,
|
||||
models,
|
||||
@@ -2492,46 +2514,11 @@ async fn send_event(
|
||||
.await
|
||||
}
|
||||
|
||||
/// Separate router because its state is the model store, like `usage`'s.
|
||||
///
|
||||
/// Keys are `owner/repo/file.gguf` and so contain slashes, which is why
|
||||
/// nothing here puts one in the path: a key travels in the body or a query
|
||||
/// string, and the routes stay addressable without escaping rules nobody would
|
||||
/// get right from a phone.
|
||||
pub fn models_router(store: Arc<crate::models::ModelStore>) -> Router {
|
||||
Router::new()
|
||||
.route("/models", get(list_models))
|
||||
.route("/models/search", get(search_models))
|
||||
.route("/models/files", get(repo_files))
|
||||
.route("/models/download", post(start_download))
|
||||
.route("/models/cancel", post(cancel_download))
|
||||
.route("/models/delete", post(delete_model))
|
||||
.with_state(store)
|
||||
}
|
||||
|
||||
/// What this machine has and what it is fetching, in one answer. Both
|
||||
/// together deliberately: a phone showing the model list needs both to draw
|
||||
/// one screen, and two routes would let it render a model as absent while its
|
||||
/// download sits at 99%.
|
||||
#[derive(serde::Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct ModelsResponse {
|
||||
local: Vec<crate::models::LocalModel>,
|
||||
downloads: Vec<crate::models::DownloadStatus>,
|
||||
}
|
||||
|
||||
async fn list_models(
|
||||
State(store): State<Arc<crate::models::ModelStore>>,
|
||||
) -> Result<axum::Json<ModelsResponse>, ApiError> {
|
||||
let listing = tokio::task::spawn_blocking(move || ModelsResponse {
|
||||
local: store.list(),
|
||||
downloads: store.downloads(),
|
||||
})
|
||||
.await
|
||||
.context("listing models panicked")?;
|
||||
Ok(axum::Json(listing))
|
||||
}
|
||||
|
||||
/// Searching HuggingFace is the one thing here that is about no machine in
|
||||
/// particular, so it is the one route left at the top level. Proxied through
|
||||
/// this server rather than called from the phone, which trusts exactly one
|
||||
/// certificate -- this server's -- and has no general internet trust to spend
|
||||
/// on huggingface.co.
|
||||
#[derive(Deserialize)]
|
||||
struct SearchQuery {
|
||||
q: String,
|
||||
@@ -2548,16 +2535,63 @@ async fn search_models(
|
||||
Ok(axum::Json(found))
|
||||
}
|
||||
|
||||
/// One machine's models and its downloads, in one answer.
|
||||
///
|
||||
/// Both together deliberately: this is one screen, and two routes would let
|
||||
/// the phone draw a model as absent while its download sits at 99%.
|
||||
#[derive(serde::Serialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
struct MachineModels {
|
||||
local: Vec<crate::models::LocalModel>,
|
||||
downloads: Vec<crate::models::Download>,
|
||||
}
|
||||
|
||||
/// Where a machine keeps its models, and how to reach it. Every route below
|
||||
/// needs both, and a machine that is not there is the same 404 each time.
|
||||
fn models_on(
|
||||
manager: &Arc<SessionManager>,
|
||||
id: &str,
|
||||
) -> Result<(crate::session::transport::Transport, String), ApiError> {
|
||||
let machine = machine_by_id(manager, id)?;
|
||||
let transport = crate::session::transport::Transport::for_machine(&machine);
|
||||
let dir = crate::models::dir_on(&transport, manager.models_dir());
|
||||
Ok((transport, dir))
|
||||
}
|
||||
|
||||
async fn machine_models(
|
||||
State(manager): State<Arc<SessionManager>>,
|
||||
UrlPath(id): UrlPath<String>,
|
||||
) -> Result<axum::Json<MachineModels>, ApiError> {
|
||||
let (transport, dir) = models_on(&manager, &id)?;
|
||||
let local = crate::models::on_machine(&transport, &dir)
|
||||
.await
|
||||
.map_err(from_machine)?;
|
||||
let downloads = crate::models::downloads(&transport, &dir)
|
||||
.await
|
||||
.map_err(from_machine)?;
|
||||
Ok(axum::Json(MachineModels { local, downloads }))
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
struct RepoQuery {
|
||||
repo: String,
|
||||
}
|
||||
|
||||
/// The GGUFs in one repository, marked with what this machine already has --
|
||||
/// which is why this is under a machine and the search is not.
|
||||
async fn repo_files(
|
||||
State(store): State<Arc<crate::models::ModelStore>>,
|
||||
State(manager): State<Arc<SessionManager>>,
|
||||
UrlPath(id): UrlPath<String>,
|
||||
Query(query): Query<RepoQuery>,
|
||||
) -> Result<axum::Json<Vec<crate::models::RemoteFile>>, ApiError> {
|
||||
let files = tokio::task::spawn_blocking(move || crate::models::files(&query.repo, &store))
|
||||
let (transport, dir) = models_on(&manager, &id)?;
|
||||
let have: std::collections::HashSet<String> = crate::models::on_machine(&transport, &dir)
|
||||
.await
|
||||
.map_err(from_machine)?
|
||||
.into_iter()
|
||||
.map(|model| model.key)
|
||||
.collect();
|
||||
let files = tokio::task::spawn_blocking(move || crate::models::files(&query.repo, &have))
|
||||
.await
|
||||
.context("listing repository files panicked")?
|
||||
.map_err(bad_request)?;
|
||||
@@ -2570,13 +2604,25 @@ struct DownloadRequest {
|
||||
file: String,
|
||||
}
|
||||
|
||||
/// Starts a download, or rejoins the one already running for that model.
|
||||
/// Starts a download on that machine, or leaves the one already running
|
||||
/// alone. What HuggingFace publishes about the file is read here and handed
|
||||
/// over, because this is the side with internet trust and because the machine
|
||||
/// needs the sha to resume safely.
|
||||
async fn start_download(
|
||||
State(store): State<Arc<crate::models::ModelStore>>,
|
||||
State(manager): State<Arc<SessionManager>>,
|
||||
UrlPath(id): UrlPath<String>,
|
||||
axum::Json(body): axum::Json<DownloadRequest>,
|
||||
) -> Result<axum::Json<crate::models::DownloadStatus>, ApiError> {
|
||||
let status = store.start(&body.repo, &body.file).map_err(bad_request)?;
|
||||
Ok(axum::Json(status))
|
||||
) -> Result<StatusCode, ApiError> {
|
||||
let (transport, dir) = models_on(&manager, &id)?;
|
||||
let (repo, file) = (body.repo.clone(), body.file.clone());
|
||||
let published = tokio::task::spawn_blocking(move || crate::models::published(&repo, &file))
|
||||
.await
|
||||
.context("asking HuggingFace about a file panicked")?
|
||||
.map_err(bad_request)?;
|
||||
crate::models::start(&transport, &dir, &body.repo, &body.file, &published)
|
||||
.await
|
||||
.map_err(from_machine)?;
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
||||
#[derive(Deserialize)]
|
||||
@@ -2585,17 +2631,25 @@ struct KeyRequest {
|
||||
}
|
||||
|
||||
async fn cancel_download(
|
||||
State(store): State<Arc<crate::models::ModelStore>>,
|
||||
State(manager): State<Arc<SessionManager>>,
|
||||
UrlPath(id): UrlPath<String>,
|
||||
axum::Json(body): axum::Json<KeyRequest>,
|
||||
) -> Result<axum::Json<crate::models::DownloadStatus>, ApiError> {
|
||||
let status = store.cancel(&body.key).map_err(bad_request)?;
|
||||
Ok(axum::Json(status))
|
||||
) -> Result<StatusCode, ApiError> {
|
||||
let (transport, dir) = models_on(&manager, &id)?;
|
||||
crate::models::cancel(&transport, &dir, &body.key)
|
||||
.await
|
||||
.map_err(from_machine)?;
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
|
||||
async fn delete_model(
|
||||
State(store): State<Arc<crate::models::ModelStore>>,
|
||||
State(manager): State<Arc<SessionManager>>,
|
||||
UrlPath(id): UrlPath<String>,
|
||||
axum::Json(body): axum::Json<KeyRequest>,
|
||||
) -> Result<StatusCode, ApiError> {
|
||||
store.delete(&body.key).map_err(bad_request)?;
|
||||
let (transport, dir) = models_on(&manager, &id)?;
|
||||
crate::models::remove(&transport, &dir, &body.key)
|
||||
.await
|
||||
.map_err(from_machine)?;
|
||||
Ok(StatusCode::NO_CONTENT)
|
||||
}
|
||||
Reference in new issue
Block a user