diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 5a5b1e3..dea1b54 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -21,9 +21,20 @@ jobs: - uses: actions/checkout@v4 - uses: dtolnay/rust-toolchain@stable - uses: Swatinem/rust-cache@v2 + # Skip when this version is already on crates.io — a version can never be + # re-uploaded, so publishing one that exists (released by hand, or a re-run + # of this job) is a guaranteed failure rather than a real problem. - name: Publish to crates.io if: ${{ env.CARGO_REGISTRY_TOKEN != '' }} - run: cargo publish + run: | + name=$(cargo metadata --no-deps --format-version 1 | jq -r '.packages[0].name') + version=$(cargo metadata --no-deps --format-version 1 | jq -r '.packages[0].version') + if curl -sf -H 'User-Agent: cortiq-gateway-release' \ + "https://crates.io/api/v1/crates/$name/$version" >/dev/null; then + echo "$name $version is already published — skipping" + else + cargo publish + fi # --- prebuilt binaries + GitHub Release ----------------------------------- release: diff --git a/src/admin/mod.rs b/src/admin/mod.rs index a8c0a76..e8bae64 100644 --- a/src/admin/mod.rs +++ b/src/admin/mod.rs @@ -1399,7 +1399,16 @@ async fn setup_complete( let _ = std::fs::create_dir_all(dir); } let lang = b.lang.unwrap_or_else(|| "en".into()); - let _ = std::fs::write(&marker, format!("done lang={lang} ts={}", std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default().as_secs())); + let _ = std::fs::write( + &marker, + format!( + "done lang={lang} ts={}", + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs() + ), + ); let mut registered: Vec = Vec::new(); if b.register.unwrap_or(true) { @@ -1473,7 +1482,9 @@ async fn setup_featured(State(state): State) -> ApiResult { } } } - if ids.is_empty() { return ok(json!({ "models": fallback })); } + if ids.is_empty() { + return ok(json!({ "models": fallback })); + } // fetch per-variant sizes in parallel (tree API, 10s each) let token_owned = token.map(|s| s.to_string()); let variant_lists: Vec<(String, Vec<(String, u64)>)> = @@ -1512,7 +1523,9 @@ async fn setup_featured(State(state): State) -> ApiResult { if let Some(arr) = models.as_array() { for m in arr { let id = m["id"].as_str().unwrap_or("").to_string(); - if !id.to_ascii_lowercase().contains("cmf") { continue; } + if !id.to_ascii_lowercase().contains("cmf") { + continue; + } let likes = m["likes"].as_u64().unwrap_or(0); let vars = variant_map.get(&id); // extract size token from id @@ -1538,7 +1551,11 @@ async fn setup_featured(State(state): State) -> ApiResult { .and_then(|vs| vs.first().map(|(n, _)| n.as_str())) .unwrap_or(&id) .to_ascii_lowercase(); - let quant = if quant_src.contains("q1t") || quant_src.contains("2bit") || lower.contains("q1t") || lower.contains("2bit") { + let quant = if quant_src.contains("q1t") + || quant_src.contains("2bit") + || lower.contains("q1t") + || lower.contains("2bit") + { "Q1T" } else if quant_src.contains("q2tp") || lower.contains("q2tp") { "Q2TP" @@ -1554,7 +1571,11 @@ async fn setup_featured(State(state): State) -> ApiResult { "" }; let mut badge = if quant.is_empty() || size_token == "CMF" { - if size_token == "CMF" { "CMF · ready".to_string() } else { size_token.to_string() } + if size_token == "CMF" { + "CMF · ready".to_string() + } else { + size_token.to_string() + } } else { format!("{size_token} · {quant}") }; @@ -1568,7 +1589,11 @@ async fn setup_featured(State(state): State) -> ApiResult { if vs.len() == 1 { let (name, sz) = &vs[0]; let s = crate::import::format_bytes(*sz); - (Some(s.clone()), Some(*sz), json!([{"name": name, "size": s, "size_bytes": sz}])) + ( + Some(s.clone()), + Some(*sz), + json!([{"name": name, "size": s, "size_bytes": sz}]), + ) } else { // dedupe same-size variants (e.g. q2tp v2/v3 both 112GB → keep latest v3) let mut seen_sz = std::collections::HashSet::new(); @@ -1576,22 +1601,31 @@ async fn setup_featured(State(state): State) -> ApiResult { // iterate reversed to keep latest (v3) over v2 when same size for (n, sz) in vs.iter().rev() { let s = crate::import::format_bytes(*sz); - if seen_sz.insert(s) { uniq_vs.push((n.clone(), *sz)); } + if seen_sz.insert(s) { + uniq_vs.push((n.clone(), *sz)); + } } uniq_vs.reverse(); - let parts: Vec = uniq_vs.iter().map(|(_, sz)| crate::import::format_bytes(*sz)).collect(); + let parts: Vec = uniq_vs + .iter() + .map(|(_, sz)| crate::import::format_bytes(*sz)) + .collect(); let max_sz = uniq_vs.iter().map(|(_, sz)| *sz).max(); let vars_json: Vec = uniq_vs.iter().map(|(n, sz)| json!({"name": n, "size": crate::import::format_bytes(*sz), "size_bytes": sz})).collect(); (Some(parts.join(" / ")), max_sz, json!(vars_json)) } } else if let Some((s, sz)) = fallback_map.get(&id) { - (Some(s.clone()), Some(*sz), json!([{"name": id.split('/').last().unwrap_or(&id), "size": s, "size_bytes": sz}])) + ( + Some(s.clone()), + Some(*sz), + json!([{"name": id.split('/').next_back().unwrap_or(&id), "size": s, "size_bytes": sz}]), + ) } else { (None::, None::, Value::Null) }; // check if already downloaded locally (case-insensitive, handles sharded dirs) let models_dir = live.cfg.cmf.models_dir.clone(); - let base = crate::import::sanitize_for_id(id.split('/').last().unwrap_or(&id)); + let base = crate::import::sanitize_for_id(id.split('/').next_back().unwrap_or(&id)); let mut downloaded = false; let mut downloaded_variants: Vec = Vec::new(); if let Some(vs) = variants_json.as_array() { @@ -1615,7 +1649,12 @@ async fn setup_featured(State(state): State) -> ApiResult { if local_sz == 0 && entry.path().is_dir() { local_sz = std::fs::read_dir(entry.path()) .ok() - .map(|r| r.filter_map(|e| e.ok()).filter_map(|e| e.metadata().ok()).map(|m| m.len()).sum()) + .map(|r| { + r.filter_map(|e| e.ok()) + .filter_map(|e| e.metadata().ok()) + .map(|m| m.len()) + .sum() + }) .unwrap_or(0); } break; @@ -1627,18 +1666,23 @@ async fn setup_featured(State(state): State) -> ApiResult { let p = std::path::Path::new(&models_dir).join(format!("{vbase}.cmf")); if p.exists() || p.is_dir() { exists = true; - local_sz = std::fs::metadata(&p) - .map(|m| m.len()) - .unwrap_or_else(|_| { + local_sz = + std::fs::metadata(&p).map(|m| m.len()).unwrap_or_else(|_| { std::fs::read_dir(&p) .ok() - .map(|r| r.filter_map(|e| e.ok()).filter_map(|e| e.metadata().ok()).map(|m| m.len()).sum()) + .map(|r| { + r.filter_map(|e| e.ok()) + .filter_map(|e| e.metadata().ok()) + .map(|m| m.len()) + .sum() + }) .unwrap_or(0) }); } } let remote_sz = v["size_bytes"].as_u64().unwrap_or(0); - let is_complete = exists && local_sz != 0 && remote_sz != 0 && local_sz == remote_sz; + let is_complete = + exists && local_sz != 0 && remote_sz != 0 && local_sz == remote_sz; // downloaded = only when complete (not partial) downloaded = downloaded || is_complete; let mut vv = v.clone(); @@ -1648,8 +1692,7 @@ async fn setup_featured(State(state): State) -> ApiResult { // the job that produces it — the UI passes both back verbatim so // download / progress / delete can never disagree on the file. vv["local_name"] = json!(vbase); - vv["job"] = - json!(crate::import::gen_id(&format!("{id}::{vname}"))); + vv["job"] = json!(crate::import::gen_id(&format!("{id}::{vname}"))); if exists && !is_complete && local_sz != 0 { // partial or outdated — show as update available if not currently downloading // we don't have job state here, so just mark update if sizes differ @@ -1660,11 +1703,17 @@ async fn setup_featured(State(state): State) -> ApiResult { downloaded_variants.push(vv); } } - let final_variants = if downloaded_variants.is_empty() { variants_json } else { json!(downloaded_variants) }; - let any_update = downloaded_variants.iter().any(|v| v["update_available"].as_bool().unwrap_or(false)); + let final_variants = if downloaded_variants.is_empty() { + variants_json + } else { + json!(downloaded_variants) + }; + let any_update = downloaded_variants + .iter() + .any(|v| v["update_available"].as_bool().unwrap_or(false)); out.push(json!({ "id": id, - "label": id.split('/').last().unwrap_or(&id), + "label": id.split('/').next_back().unwrap_or(&id), "local_name": base, "badge": badge, "desc": m["pipeline_tag"].as_str().unwrap_or("text-generation"), @@ -1678,7 +1727,11 @@ async fn setup_featured(State(state): State) -> ApiResult { })); } } - if out.is_empty() { fallback } else { json!(out) } + if out.is_empty() { + fallback + } else { + json!(out) + } }; ok(json!({ "models": arr })) } @@ -1747,8 +1800,12 @@ async fn delete_setup_file( } let live = state.live(); let models_dir = live.cfg.cmf.models_dir.clone(); - let base = crate::import::sanitize_for_id(repo.split('/').last().unwrap_or(repo)); - let variant = q.variant.as_deref().map(|s| s.trim()).filter(|s| !s.is_empty()); + let base = crate::import::sanitize_for_id(repo.split('/').next_back().unwrap_or(repo)); + let variant = q + .variant + .as_deref() + .map(|s| s.trim()) + .filter(|s| !s.is_empty()); // Exact stem wins; sanitizing it also keeps the delete inside models_dir. let vbase = match q.name.as_deref().map(str::trim).filter(|s| !s.is_empty()) { Some(n) => crate::import::sanitize_for_id(n), diff --git a/src/config.rs b/src/config.rs index 754ec22..256b5d9 100644 --- a/src/config.rs +++ b/src/config.rs @@ -774,7 +774,10 @@ pub fn default_config_path() -> String { if std::path::Path::new("Cargo.toml").exists() && std::path::Path::new("config").exists() { return repo_path.to_string(); } - data_dir().join("gateway.toml").to_string_lossy().into_owned() + data_dir() + .join("gateway.toml") + .to_string_lossy() + .into_owned() } impl Config { diff --git a/src/import.rs b/src/import.rs index bd06942..5fa8882 100644 --- a/src/import.rs +++ b/src/import.rs @@ -206,10 +206,9 @@ impl JobStore { if let Some(j) = self.jobs.lock().unwrap().get_mut(id) { let f = frac.clamp(0.0, 1.0); let cur = j.progress.unwrap_or(0.0); - // monotonic only — never go backwards (curl reprints, re-download) - if f > cur + 0.001 { - j.progress = Some(f); - } else if j.progress.is_none() { + // monotonic only — never go backwards (curl reprints, re-download), + // but an unset bar takes whatever the first reading is + if j.progress.is_none() || f > cur + 0.001 { j.progress = Some(f); } if !phase.is_empty() { @@ -306,7 +305,9 @@ impl JobStore { let key = format!("{}::{}", j.repo, j.variant.clone().unwrap_or_default()); match map.get(&key) { Some(prev) if prev.started >= j.started => {} - _ => { map.insert(key, j); } + _ => { + map.insert(key, j); + } } } let mut v: Vec = map.into_values().collect(); @@ -434,7 +435,11 @@ pub async fn hf_repo_variants(id: &str, token: Option<&str>) -> Vec<(String, u64 let sz = item .get("size") .and_then(|v| v.as_u64()) - .or_else(|| item.get("lfs").and_then(|l| l.get("size")).and_then(|v| v.as_u64())) + .or_else(|| { + item.get("lfs") + .and_then(|l| l.get("size")) + .and_then(|v| v.as_u64()) + }) .unwrap_or(0); if sz == 0 { continue; @@ -559,7 +564,8 @@ async fn download_cmf_repo( } let mut cmf_files: Vec<(String, u64)> = Vec::new(); - let mut sharded_dirs: std::collections::HashMap> = std::collections::HashMap::new(); + let mut sharded_dirs: std::collections::HashMap> = + std::collections::HashMap::new(); let mut dir_sizes: std::collections::HashMap = std::collections::HashMap::new(); if let Ok(resp) = req.send().await { if resp.status().is_success() { @@ -573,13 +579,20 @@ async fn download_cmf_repo( let sz = item .get("size") .and_then(|v| v.as_u64()) - .or_else(|| item.get("lfs").and_then(|l| l.get("size")).and_then(|v| v.as_u64())) + .or_else(|| { + item.get("lfs") + .and_then(|l| l.get("size")) + .and_then(|v| v.as_u64()) + }) .unwrap_or(0); if path.to_ascii_lowercase().ends_with(".cmf") { cmf_files.push((path.to_string(), sz)); } else if path.contains('/') && path.starts_with("parts-") { let top = path.split('/').next().unwrap_or(path).to_string(); - sharded_dirs.entry(top.clone()).or_default().push(path.to_string()); + sharded_dirs + .entry(top.clone()) + .or_default() + .push(path.to_string()); *dir_sizes.entry(top).or_default() += sz; } } @@ -591,12 +604,19 @@ async fn download_cmf_repo( // Decide what to download: single .cmf file vs sharded variant directory let is_sharded = !sharded_dirs.is_empty(); - let chosen_variant = variant.as_deref().map(|v| v.trim()).filter(|v| !v.is_empty()); + let chosen_variant = variant + .as_deref() + .map(|v| v.trim()) + .filter(|v| !v.is_empty()); if is_sharded { // pick variant: explicit > first alphabetically let target_dir = if let Some(v) = chosen_variant { - if sharded_dirs.contains_key(v) { v.to_string() } else { sharded_dirs.keys().next().cloned().unwrap_or_default() } + if sharded_dirs.contains_key(v) { + v.to_string() + } else { + sharded_dirs.keys().next().cloned().unwrap_or_default() + } } else { // prefer smallest? use first sorted let mut keys: Vec = sharded_dirs.keys().cloned().collect(); @@ -611,9 +631,23 @@ async fn download_cmf_repo( return; } let total_sz = dir_sizes.get(&target_dir).copied().unwrap_or(0); - let q = if target_dir.contains("q2tp") { "Q2TP" } else if target_dir.contains("q4tp") { "Q4TP" } else { "READY_CMF" }; + let q = if target_dir.contains("q2tp") { + "Q2TP" + } else if target_dir.contains("q4tp") { + "Q4TP" + } else { + "READY_CMF" + }; store.update_quant(&id, q.to_string()); - store.push_line(&id, format!("→ downloading {} ({} files, {}) from HuggingFace...", target_dir, files.len(), format_bytes(total_sz))); + store.push_line( + &id, + format!( + "→ downloading {} ({} files, {}) from HuggingFace...", + target_dir, + files.len(), + format_bytes(total_sz) + ), + ); store.set_total(&id, total_sz); store.set_progress(&id, 0.001, "downloading".into()); @@ -635,7 +669,13 @@ async fn download_cmf_repo( let url = format!("https://huggingface.co/{repo}/resolve/main/{rel}"); store.push_line(&id, format!(" [{}/{}] {filename}", idx + 1, files.len())); let mut cmd = tokio::process::Command::new("curl"); - cmd.arg("-f").arg("-L").arg("-s").arg("-S").arg("-o").arg(&dest).arg(&url); + cmd.arg("-f") + .arg("-L") + .arg("-s") + .arg("-S") + .arg("-o") + .arg(&dest) + .arg(&url); if let Some(t) = &hf_token { cmd.arg("-H").arg(format!("Authorization: Bearer {t}")); } @@ -758,7 +798,11 @@ async fn download_cmf_repo( .unwrap_or(0) }; store.push_line(&id, { - let sz = if total > 0 { format!(" ({})", format_bytes(total)) } else { String::new() }; + let sz = if total > 0 { + format!(" ({})", format_bytes(total)) + } else { + String::new() + }; format!("→ downloading {cmf_path}{sz} from HuggingFace...") }); store.set_total(&id, total); @@ -773,7 +817,12 @@ async fn download_cmf_repo( let _poller = spawn_size_poller(store.clone(), id.clone(), output_abs.clone()); let mut cmd = tokio::process::Command::new("curl"); - cmd.arg("-f").arg("-L").arg("-s").arg("-S").arg("-o").arg(&output_abs); + cmd.arg("-f") + .arg("-L") + .arg("-s") + .arg("-S") + .arg("-o") + .arg(&output_abs); if let Some(t) = &hf_token { cmd.arg("-H").arg(format!("Authorization: Bearer {t}")); @@ -905,7 +954,12 @@ pub fn start_import(store: Arc, cfg: &CmfCfg, p: ImportParams) -> Resu sanitize(&p.name) }; // for sharded variants make output distinct (e.g. DeepSeek parts-q2tp vs q4tp); single-file CMF ignores variant - if let Some(v) = p.variant.as_deref().map(|s| s.trim()).filter(|s| !s.is_empty()) { + if let Some(v) = p + .variant + .as_deref() + .map(|s| s.trim()) + .filter(|s| !s.is_empty()) + { if v.starts_with("parts-") { let vs = sanitize(v); if !base.ends_with(&vs) { @@ -921,7 +975,11 @@ pub fn start_import(store: Arc, cfg: &CmfCfg, p: ImportParams) -> Resu .map(|d| d.join(format!("{base}.cmf")).to_string_lossy().to_string()) .unwrap_or_else(|_| output.clone()); - let id = gen_id(&format!("{}::{}", p.repo, p.variant.as_deref().unwrap_or(""))); + let id = gen_id(&format!( + "{}::{}", + p.repo, + p.variant.as_deref().unwrap_or("") + )); // if same repo/variant is already running, don't reset progress — just return it if let Some(existing) = store.get(&id) { if existing.state == "running" { diff --git a/src/main.rs b/src/main.rs index b1f941f..caee898 100644 --- a/src/main.rs +++ b/src/main.rs @@ -90,7 +90,10 @@ fn should_open_browser(args: &Args) -> bool { if args.no_browser { return false; } - if std::env::var("CORTIQ_NO_BROWSER").map(|v| !v.is_empty() && v != "0").unwrap_or(false) { + if std::env::var("CORTIQ_NO_BROWSER") + .map(|v| !v.is_empty() && v != "0") + .unwrap_or(false) + { return false; } if std::env::var("CI").is_ok() || std::env::var("NO_BROWSER").is_ok() { @@ -121,7 +124,9 @@ fn is_port_free(host: &str, port: u16) -> bool { let probe_host = if host == "0.0.0.0" { "127.0.0.1" } else { host }; if let Ok(mut addrs) = (probe_host, port).to_socket_addrs() { if let Some(a) = addrs.next() { - if std::net::TcpStream::connect_timeout(&a, std::time::Duration::from_millis(200)).is_ok() { + if std::net::TcpStream::connect_timeout(&a, std::time::Duration::from_millis(200)) + .is_ok() + { return false; } } @@ -134,7 +139,11 @@ fn resolve_listen(cfg: &mut Config, no_search: bool) -> String { return cfg.listen.clone(); } let (host, port) = parse_listen(&cfg.listen); - let clean_host = if host.is_empty() { "0.0.0.0".to_string() } else { host }; + let clean_host = if host.is_empty() { + "0.0.0.0".to_string() + } else { + host + }; if is_port_free(&clean_host, port) { return cfg.listen.clone(); } @@ -147,7 +156,9 @@ fn resolve_listen(cfg: &mut Config, no_search: bool) -> String { } } if !is_port_free(&clean_host, port) { - eprintln!("warning: port {port} busy and no free port found in next 50, will try to bind anyway"); + eprintln!( + "warning: port {port} busy and no free port found in next 50, will try to bind anyway" + ); } cfg.listen.clone() } @@ -244,7 +255,11 @@ async fn main() -> anyhow::Result<()> { listen.clone() }; // On first run open the onboarding wizard, otherwise the normal console. - let admin_path = if first_run { "/admin#/onboarding" } else { "/admin" }; + let admin_path = if first_run { + "/admin#/onboarding" + } else { + "/admin" + }; let url = if admin_token.is_empty() { format!("http://{host_for_url}{admin_path}") } else {