From c5c37617728ed3b0df572d6a82fe6f876871c3d1 Mon Sep 17 00:00:00 2001 From: Vincent Date: Fri, 16 Oct 2026 08:36:17 +0200 Subject: [PATCH] feat(dump): JSON dump of all taskguard knows about a job `taskguard dump PATTERN | --cmd "COMMAND" | --run ID` prints one JSON document per job: how it matches, the rules in force and who set them, what the next run reserves and why (step by step), each recent run with its wait spans, adjustments and role in the need, pruned runs, and live queue entries. The need math moves from runner.rs into need.rs, so the runner and dump use the same code. Each run now stores its need steps in a new nullable column, runs.need_why (also in pruned_runs and RUN_COLUMNS), so old runs can be explained after the fact. Co-Authored-By: Claude Opus 5.5 (1M context) --- CHANGELOG.md | 12 ++ README.md | 1 + src/db.rs | 127 +++++++++++++++++---- src/dump.rs | 247 +++++++++++++++++++++++++++++++++++++++++ src/help.rs | 42 ++++++- src/main.rs | 6 + src/need.rs | 297 ++++++++++++++++++++++++++++++++++++++++++++++++++ src/runner.rs | 52 +++------ tests/cli.rs | 32 +++++- 9 files changed, 755 insertions(+), 61 deletions(-) create mode 100644 src/dump.rs create mode 100644 src/need.rs diff --git a/CHANGELOG.md b/CHANGELOG.md index b265fef..c2971f4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,18 @@ and the versions follow [semver](https://semver.org). ## [Unreleased] +### Added + +- `taskguard dump PATTERN | --cmd "COMMAND" | --run ID` prints everything + taskguard knows about a job as one JSON document: how the command matches, + the rules in force and which file set them, what the next run reserves and + why, step by step (history, boost, first guess, caps, minimums), each recent + run with what it waited for, what it changed, and its role in the need + (outlier, failed and ignored, boost), pruned runs, and the live queue entry. + Use it to find out why a job reserved what it did. +- Each run now stores how its need was made (`runs.need_why`), so `dump` + can explain an old run too, not only the next one. + ## [0.10.0] - 2026-10-16 ### Fixed diff --git a/README.md b/README.md index c72b0c6..9cf82d6 100644 --- a/README.md +++ b/README.md @@ -509,6 +509,7 @@ jobs. Set `TASKGUARD=/path/to/taskguard` to try a local build. | `taskguard prune --undo [PATTERN]` | Put pruned runs back | | `taskguard doctor` | Configuration, live readings, and leftover tsc-queue shims | | `taskguard doctor --explain "COMMAND"` | How one command is matched, pooled and learned | +| `taskguard dump PATTERN \| --cmd "COMMAND" \| --run ID` | Everything taskguard knows about a job, as JSON: how it matches, what its next run reserves and why, step by step, and each recent run with what it waited for and what it did to the need. Attach it to a bug report | | `taskguard import-history` | Load tsc-queue's memory history | | `taskguard help COMMAND`, `taskguard COMMAND --help` | Help for one command: what it does, its options, examples | | `taskguard help --all` | Every command in full, in one text, for LLM agents | diff --git a/src/db.rs b/src/db.rs index 1861573..0b08557 100644 --- a/src/db.rs +++ b/src/db.rs @@ -6,6 +6,7 @@ use anyhow::{Context, Result}; use rusqlite::{Connection, OptionalExtension, params}; +use serde::Serialize; use std::path::Path; use std::time::Duration; @@ -60,7 +61,8 @@ CREATE TABLE IF NOT EXISTS runs ( package TEXT, steady_cores_wanted REAL, steady_mem_kb INTEGER, - nested_jobs INTEGER + nested_jobs INTEGER, + need_why TEXT ); CREATE INDEX IF NOT EXISTS runs_key ON runs(key, ended_at); CREATE INDEX IF NOT EXISTS runs_ended ON runs(ended_at); @@ -177,6 +179,7 @@ CREATE TABLE IF NOT EXISTS pruned_runs ( steady_cores_wanted REAL, steady_mem_kb INTEGER, nested_jobs INTEGER, + need_why TEXT, pruned_at REAL NOT NULL, prune_reason TEXT NOT NULL ); @@ -188,7 +191,7 @@ CREATE INDEX IF NOT EXISTS pruned_runs_key ON pruned_runs(key); const RUN_COLUMNS: &str = "id, ns, pool, key, label, cwd, cmd, pid, script, package_json, queued_at, started_at, ended_at, exit, waited_s, main_blocker, now, min_cpu, min_mem_kb, need_cpu, need_mem_kb, peak_mem_kb, cores_used, cores_wanted, cpu_seconds, runnable_seconds, pageins, starved, starved_detail, machine_full_frac, imported, paused_s, package, steady_cores_wanted, - steady_mem_kb, nested_jobs"; + steady_mem_kb, nested_jobs, need_why"; pub fn now() -> f64 { std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs_f64()).unwrap_or(0.0) @@ -217,7 +220,7 @@ pub struct Estimate { } /// What history says about one key. -#[derive(Debug, Clone, Default, PartialEq)] +#[derive(Debug, Clone, Default, PartialEq, Serialize)] pub struct Learned { pub runs: usize, /// Max of the recent peaks, with outliers held back (see `Outliers`). @@ -290,7 +293,7 @@ impl Default for Learn { } /// A recent run whose memory peak stands far above the others. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize)] pub struct Outlier { pub run_id: i64, pub peak_kb: u64, @@ -311,6 +314,20 @@ impl Outlier { } } +/// Whether a run adds the 25% memory boost: it was starved of memory and +/// reached what it was given (`starved`, the need it started with, its peak, +/// and whether it failed). See `Db::learned`. +pub fn boosts(starved: Option<&str>, need_kb: Option, peak_kb: i64, failed: bool) -> bool { + starved == Some("memory") + && match need_kb { + None => !failed, + Some(need) => { + let peak = peak_kb as f64; + peak >= need as f64 * BOOST_REACHED && (!failed || peak <= need as f64 * BOOST_FAILED_MAX) + } + } +} + /// The memory need when every recent run failed: the second-highest peak, /// at least `new_job_mem_kb` but never more than the highest. One failed run /// far above the others alone does not set the need, as the job is then as @@ -327,7 +344,7 @@ pub fn failed_need(peaks: &[u64], new_job_mem_kb: u64) -> Option { /// took 20 GB where the others took 5 GB. A single such run may be a one-off, /// so it does not set the memory need at once: it counts by `weight`, which /// grows each time it happens again, until it counts in full. -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, Serialize)] pub struct Outliers { pub runs: Vec, /// The highest peak below them. @@ -501,6 +518,8 @@ pub struct NewRun<'a> { pub min_mem_kb: Option, pub need_cpu: f64, pub need_mem_kb: u64, + /// How the need was made, as JSON, for `taskguard dump`. + pub need_why: Option<&'a str>, } #[derive(Debug, Clone, Default)] @@ -579,10 +598,13 @@ impl Db { } // Databases made before wrapper jobs were learned: how many taskguard // jobs a run started inside itself. Both tables, so a prune keeps it. - for table in ["runs", "pruned_runs"] { - let has: bool = conn.prepare(&format!("SELECT 1 FROM pragma_table_info('{table}') WHERE name = 'nested_jobs'"))?.exists([])?; - if !has { - conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN nested_jobs INTEGER"))?; + // And how the need a run started with was made, as JSON (`need::Need`). + for (col, kind) in [("nested_jobs", "INTEGER"), ("need_why", "TEXT")] { + for table in ["runs", "pruned_runs"] { + let has: bool = conn.prepare(&format!("SELECT 1 FROM pragma_table_info('{table}') WHERE name = '{col}'"))?.exists([])?; + if !has { + conn.execute_batch(&format!("ALTER TABLE {table} ADD COLUMN {col} {kind}"))?; + } } } Ok(Db { conn }) @@ -642,16 +664,7 @@ impl Db { // One that failed counts only when its peak also stayed within what // the boost gives: a run that went to 50x its need and crashed tells // nothing about its need. - let boosted = rows.iter().zip(&runs).take(boost_runs).any(|(r, o)| { - r.3.as_deref() == Some("memory") - && match (r.6, o.failed()) { - (None, failed) => !failed, - (Some(need), failed) => { - let peak = r.0 as f64; - peak >= need as f64 * BOOST_REACHED && (!failed || peak <= need as f64 * BOOST_FAILED_MAX) - } - } - }); + let boosted = rows.iter().zip(&runs).take(boost_runs).any(|(r, o)| boosts(r.3.as_deref(), r.6, r.0, o.failed())); let steady_mem_kb = recent.iter().filter_map(|r| r.4).max().map(|m| m as u64); let steady_cpu = median_of(recent.iter().filter_map(|r| r.5).collect()); let nested_jobs = recent.iter().filter_map(|r| r.7).filter(|n| *n > 0).max().map(|n| n as u64); @@ -669,6 +682,75 @@ impl Db { }) } + /// Rows as JSON objects, one field per column, for `taskguard dump`: every + /// column, the ones a newer version added too. + fn rows_json(&self, sql: &str, args: impl rusqlite::Params) -> Result> { + use rusqlite::types::ValueRef; + let mut stmt = self.conn.prepare(sql)?; + let names: Vec = stmt.column_names().iter().map(|n| n.to_string()).collect(); + let rows = stmt + .query_map(args, |r| { + let mut o = serde_json::Map::new(); + for (i, n) in names.iter().enumerate() { + let v = match r.get_ref(i)? { + ValueRef::Null => serde_json::Value::Null, + ValueRef::Integer(v) => v.into(), + ValueRef::Real(v) => v.into(), + ValueRef::Text(t) => { + let t = String::from_utf8_lossy(t); + // need_why holds JSON: keep it as JSON. + match (n.as_str(), serde_json::from_str(&t)) { + ("need_why", Ok(v)) => v, + _ => t.into_owned().into(), + } + } + ValueRef::Blob(b) => format!("{} bytes", b.len()).into(), + }; + o.insert(n.clone(), v); + } + Ok(serde_json::Value::Object(o)) + })? + .collect::>()?; + Ok(rows) + } + + /// The newest `limit` runs of a key: those still waiting or running + /// first, then the finished ones in the order `learned` reads them. + pub fn runs_json(&self, key: &str, limit: usize) -> Result> { + self.rows_json( + "SELECT * FROM runs WHERE key = ?1 ORDER BY ended_at IS NULL DESC, ended_at DESC, id DESC LIMIT ?2", + params![key, limit as i64], + ) + } + + /// What a run waited for, span by span. + pub fn wait_spans_json(&self, run_id: i64) -> Result> { + self.rows_json("SELECT * FROM wait_spans WHERE run_id = ?1 ORDER BY from_ts", [run_id]) + } + + /// The changes a run made to its key's needs (boosts, outliers, advice). + pub fn adjustments_json(&self, run_id: i64) -> Result> { + self.rows_json("SELECT * FROM adjustments WHERE run_id = ?1 ORDER BY ts", [run_id]) + } + + /// The pruned runs of a key. + pub fn pruned_json(&self, key: &str) -> Result> { + self.rows_json( + "SELECT id, ended_at, exit, peak_mem_kb, pruned_at, prune_reason FROM pruned_runs WHERE key = ?1 ORDER BY ended_at DESC", + [key], + ) + } + + /// The key of a run, in the history or among the pruned runs. + pub fn key_of_run(&self, run_id: i64) -> Result> { + Ok(self + .conn + .query_row("SELECT key FROM runs WHERE id = ?1 UNION ALL SELECT key FROM pruned_runs WHERE id = ?1 LIMIT 1", [run_id], |r| { + r.get(0) + }) + .optional()?) + } + /// Every key with finished runs, or those that match `pattern`: a key, or /// a glob with `*` and `?`. pub fn keys(&self, pattern: Option<&str>) -> Result> { @@ -794,8 +876,8 @@ impl Db { pub fn insert_run(&self, r: &NewRun) -> Result { self.conn.execute( "INSERT INTO runs (ns, pool, key, label, cwd, cmd, pid, script, package_json, queued_at, now, - min_cpu, min_mem_kb, need_cpu, need_mem_kb, package) - VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16)", + min_cpu, min_mem_kb, need_cpu, need_mem_kb, package, need_why) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, ?16, ?17)", params![ r.ns, r.pool, @@ -812,7 +894,8 @@ impl Db { r.min_mem_kb.map(|v| v as i64), r.need_cpu, r.need_mem_kb as i64, - r.package + r.package, + r.need_why ], )?; Ok(self.conn.last_insert_rowid()) diff --git a/src/dump.rs b/src/dump.rs new file mode 100644 index 0000000..ab5cce6 --- /dev/null +++ b/src/dump.rs @@ -0,0 +1,247 @@ +//! `taskguard dump`: everything taskguard knows about a job, as one JSON +//! document, to find out later why it reserved what it did and to tune the +//! rules from real histories. + +use crate::config::{self, Config}; +use crate::db::{self, Db}; +use crate::key; +use crate::machine; +use crate::matcher; +use crate::need; +use crate::queue::Queue; +use anyhow::{Result, bail}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; + +/// The options of `dump`. +#[derive(Debug, Default, PartialEq)] +struct Args { + pattern: Option, + cmd: Option, + run: Option, + runs: usize, +} + +fn parse(args: &[String]) -> Result { + let mut a = Args { runs: 20, ..Default::default() }; + let mut i = 0; + while i < args.len() { + let arg = args[i].as_str(); + let (name, inline) = match arg.split_once('=') { + Some((n, v)) if arg.starts_with("--") => (n, Some(v.to_string())), + _ => (arg, None), + }; + let mut value = || -> Result { + if let Some(v) = inline.clone() { + return Ok(v); + } + i += 1; + args.get(i).cloned().ok_or_else(|| anyhow::anyhow!("{name} needs a value")) + }; + match name { + "--cmd" => a.cmd = Some(value()?), + "--run" => a.run = Some(value()?.parse().map_err(|_| anyhow::anyhow!("--run needs a run id, a number"))?), + "--runs" => a.runs = value()?.parse().map_err(|_| anyhow::anyhow!("--runs needs a number"))?, + _ if arg.starts_with('-') => bail!("unknown option {arg}"), + _ if a.pattern.is_none() => a.pattern = Some(arg.to_string()), + _ => bail!("one key pattern only; quote it so the shell leaves the * alone"), + } + i += 1; + } + if [a.pattern.is_some(), a.cmd.is_some(), a.run.is_some()].iter().filter(|x| **x).count() != 1 { + bail!("name the job one way: a key PATTERN, --cmd \"COMMAND\", or --run ID"); + } + Ok(a) +} + +/// `taskguard dump PATTERN | --cmd "COMMAND" | --run ID [--runs N]` +pub fn dump(args: &[String]) -> Result { + let a = parse(args)?; + let dir = config::state_dir(); + let db = Db::open_dir(&dir)?; + let here = std::env::current_dir()?; + let mut jobs = Vec::new(); + if let Some(cmdline) = &a.cmd { + let argv = matcher::split_shell(cmdline); + let k = key::run_key(&here, &argv); + jobs.push(job(&db, &dir, &k, Some((&argv, &here)), &a)?); + } else { + let keys = match (a.run, &a.pattern) { + (Some(id), _) => match db.key_of_run(id)? { + Some(k) => vec![k], + None => bail!("no run {id} in the history"), + }, + (None, Some(p)) => { + let all = db.keys(Some(p))?; + if all.is_empty() { + bail!("no key matches {p:?}; taskguard history lists them"); + } + all + } + _ => unreachable!(), + }; + for k in keys { + jobs.push(job(&db, &dir, &k, None, &a)?); + } + } + let m = machine::measure(None, 0); + let out = json!({ + "taskguard": env!("CARGO_PKG_VERSION"), + "generated_at": db::now(), + "state_dir": dir, + "machine": m, + "focus_run": a.run, + "jobs": jobs, + }); + println!("{}", serde_json::to_string_pretty(&out)?); + Ok(0) +} + +/// One job: how it is matched, what it learned, what its next run reserves +/// and why, its recent runs with what each did to the need, and its live +/// queue entries. `given` is the command and directory when the user named +/// one; otherwise both come from the newest run. +fn job(db: &Db, dir: &Path, k: &str, given: Option<(&Vec, &PathBuf)>, a: &Args) -> Result { + let rows = db.runs_json(k, a.runs)?; + let newest = rows.first(); + let text = |r: Option<&Value>, f: &str| r.and_then(|r| r[f].as_str()).map(str::to_string); + // The settings of the directory the job runs in, when it still exists. + let (argv, cwd) = match given { + Some((argv, cwd)) => (argv.clone(), cwd.clone()), + None => ( + text(newest, "cmd").map(|c| matcher::split_shell(&c)).unwrap_or_default(), + text(newest, "cwd").map(PathBuf::from).filter(|p| p.is_dir()).unwrap_or(std::env::current_dir()?), + ), + }; + let cfg = Config::load(&cwd, &key::checkout_root(&cwd))?; + let cls = cfg.classify(&argv, k)?; + let rules = cfg.learn(); + let (learned, history_key) = + if given.is_some() { need::history(db, &cfg, &cwd, &argv, k) } else { (db.learned(k, &rules)?, k.to_string()) }; + let pool = cls.pool.as_ref().map(|p| p.0.clone()); + let guess = (learned.runs == 0).then(|| need::guess(Some(db), &cfg, None, &argv, cls.label.as_deref(), pool.as_deref())); + // A --min-cpu or --min-mem flag is not in the config: the newest run + // recorded what it was given. + let num = |f: &str| newest.and_then(|r| r[f].as_f64()); + let min_cpu = cls.min_cpu.or(num("min_cpu")); + let min_mem = cls.min_mem_kb.or(num("min_mem_kb").map(|v| v as u64)); + let next = need::derive(&need::inputs(&cfg, &learned, Some(&history_key), guess.as_ref(), min_cpu, min_mem)); + + // What each run did to the need, in the order `Db::learned` reads them. + let outliers: Vec = learned.outliers.iter().flat_map(|o| o.runs.iter().map(|r| r.run_id)).collect(); + let ignored: Vec = learned.failed.iter().map(|r| r.run_id).collect(); + let mut seen = 0; + let mut runs = Vec::new(); + for mut r in rows { + let id = r["id"].as_i64().unwrap_or(0); + let peak = r["peak_mem_kb"].as_i64().unwrap_or(0); + let exit = r["exit"].as_i64(); + let failed = exit.is_some_and(|e| e != 0); + let finished = !r["ended_at"].is_null(); + let counted = finished && peak > 0; + let place = counted.then(|| { + seen += 1; + seen - 1 + }); + let in_window = place.is_some_and(|p| p < rules.keep); + let boost = + place.is_some_and(|p| p < rules.boost_runs) && db::boosts(r["starved"].as_str(), r["need_mem_kb"].as_i64(), peak, failed); + let mut notes = Vec::new(); + if !finished { + notes.push("still waiting or running".to_string()); + } else if !counted { + notes.push("no memory reading: not learned from".into()); + } else if !in_window { + notes.push(format!("older than the last {} runs: no longer counts", rules.keep)); + } + if outliers.contains(&id) { + notes.push("an outlier: counts by outlier_weights".into()); + } + if ignored.contains(&id) { + notes.push("failed, and its peak is ignored".into()); + } else if failed && in_window { + notes.push("failed: its peak counts only up to what passing runs took".into()); + } + if boost { + notes.push("starved of memory at its need: adds the 25% boost".into()); + } + r["role"] = json!({ + "in_window": in_window, + "failed": failed, + "outlier": outliers.contains(&id), + "failed_ignored": ignored.contains(&id), + "boost": boost, + "notes": notes, + }); + r["wait_spans"] = db.wait_spans_json(id)?.into(); + r["adjustments"] = db.adjustments_json(id)?.into(); + runs.push(r); + } + + let q = Queue::open(dir)?; + let live: Vec = + q.running().into_iter().chain(q.waiting()).filter(|e| e.key == k).map(|e| serde_json::to_value(e).unwrap_or_default()).collect(); + let origin: serde_json::Map = + cfg.origin.iter().filter(|(_, v)| *v != "built-in").map(|(k, v)| (k.clone(), Value::from(v.as_str()))).collect(); + + Ok(json!({ + "key": k, + "history_key": history_key, + "match": { + "cmd": argv.join(" "), + "cwd": cwd, + "namespace": key::namespace(&cwd), + "passthrough": cls.passthrough, + "now": cls.now, + "pool": cls.pool.as_ref().map(|(n, s, pc)| json!({"name": n, "max_slots": s, "per_checkout": pc})), + "label": cls.label, + "min_cpu": min_cpu, + "min_mem_kb": min_mem, + "priority": cls.priority, + "long_lived_startup_s": cfg.long_lived(&cls, pool.as_deref()), + "why": cls.why, + }, + "config": { + "layers": cfg.layers, + "set_by": origin, + "rules": { + "keep": rules.keep, + "boost_runs": rules.boost_runs, + "outlier_ratio": rules.outlier_ratio, + "outlier_min_kb": rules.outlier_min_kb, + "outlier_weights": rules.outlier_weights, + "new_job_mem_kb": rules.new_job_mem_kb, + "cpu_max": cfg.cpu_max, + "mem_max": cfg.mem_max, + "noise_mem": cfg.noise_mem, + "noise_cpu": cfg.noise_cpu, + "partial_fit": cfg.partial_fit, + "max_backfill": cfg.max_backfill, + }, + }, + "learned": learned, + "next_need": next, + "runs": runs, + "pruned": db.pruned_json(k)?, + "live": live, + })) +} + +#[cfg(test)] +mod tests { + use super::*; + + fn args(s: &str) -> Vec { + s.split_whitespace().map(String::from).collect() + } + + #[test] + fn the_job_is_named_one_way() { + assert_eq!(parse(&args("'k*'")).unwrap().pattern.as_deref(), Some("'k*'")); + let a = parse(&args("--run 42 --runs=5")).unwrap(); + assert_eq!((a.run, a.runs), (Some(42), 5)); + assert!(parse(&args("")).is_err()); + assert!(parse(&args("k --run 1")).is_err()); + assert!(parse(&args("--run x")).is_err()); + } +} diff --git a/src/help.rs b/src/help.rs index f4236ed..d8dbb23 100644 --- a/src/help.rs +++ b/src/help.rs @@ -185,6 +185,43 @@ examples: taskguard prune --undo 'packages/api:*' --apply "; +const DUMP: &str = "\ +taskguard dump PATTERN | --cmd \"COMMAND\" | --run ID [--runs N] + +Everything taskguard knows about a job, as one JSON document: to find out +why a job reserved what it did, and to tune the rules from real histories. +Attach it to a bug report. + +Name the job one way: + PATTERN a history key, or a glob with * and ? (taskguard history + lists the keys); every key that matches gets an entry + --cmd \"COMMAND\" the job this command is in this directory, matched as + doctor --explain does + --run ID the job of one run (runs[].id, or wait_spans[].run_id) + +options: + --runs N how many recent runs to include per job (default 20) + +per job: + match how the command matches: pool, label, minimums, and which + rule set what + config the settings files read, the ones that changed a setting, + and the rules that turn history into needs + learned what the history says: memory, CPU, outliers, failed runs + whose peak is ignored, the 25% boost + next_need what the next run reserves, step by step, each step with + why (history, boost, first_guess, cpu_cap, min_cpu, + min_mem, mem_cap) + runs each run as stored, newest first: its need and how that + was made (need_why; older runs have none), its peak, exit, what it + waited for (wait_spans), what it changed (adjustments), + and its role in the next need (in the window, outlier, + failed and ignored, boost) + pruned runs taken out with taskguard prune + live the job's queue entries while it waits or runs, with what + holds it back now +"; + const DOCTOR: &str = "\ taskguard doctor [--explain \"COMMAND\"] @@ -231,6 +268,7 @@ pub fn text(cmd: &str) -> Option { "history" => HISTORY.into(), "outliers" => OUTLIERS.into(), "prune" => PRUNE.into(), + "dump" => DUMP.into(), "doctor" => DOCTOR.into(), "import-history" => IMPORT.into(), "version" => VERSION.into(), @@ -240,8 +278,8 @@ pub fn text(cmd: &str) -> Option { } /// Every command in the order `help --all` prints them. -pub const ALL: [&str; 12] = - ["run", "wait", "top", "status", "pause", "start", "history", "outliers", "prune", "doctor", "import-history", "version"]; +pub const ALL: [&str; 13] = + ["run", "wait", "top", "status", "pause", "start", "history", "outliers", "prune", "dump", "doctor", "import-history", "version"]; /// The full help, every command in one text: for LLM agents and for reading /// it all at once. `usage` is the overview that comes first. diff --git a/src/main.rs b/src/main.rs index 7c75e49..f26ddc9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -7,12 +7,14 @@ mod commands; mod config; mod dash; mod db; +mod dump; mod groups; mod help; mod insight; mod key; mod machine; mod matcher; +mod need; mod queue; mod recorder; mod report; @@ -39,6 +41,8 @@ usage: taskguard outliers [PATTERN] memory peaks far above a job's other runs, and how much they count taskguard prune PATTERN | --outliers | --undo drop runs from the history, or put them back (a dry run without --apply) + taskguard dump PATTERN | --cmd \"COMMAND\" | --run ID + all taskguard knows about a job, as JSON: what it reserves and why taskguard doctor [--explain \"COMMAND\"] configuration, readings, and how a command matches taskguard import-history load tsc-queue's history taskguard run [options] -- COMMAND the same as the first form, for a command named like a subcommand @@ -65,6 +69,7 @@ const SUBCOMMANDS: &[&str] = &[ "history", "outliers", "prune", + "dump", "doctor", "import-history", "version", @@ -187,6 +192,7 @@ fn dispatch(args: &[String]) -> Result { "history" => commands::history(), "outliers" => commands::outliers(rest), "prune" => commands::prune(rest), + "dump" => dump::dump(rest), "doctor" => commands::doctor(rest), "import-history" => commands::import_history(), "top" => top::run(rest), diff --git a/src/need.rs b/src/need.rs new file mode 100644 index 0000000..353d413 --- /dev/null +++ b/src/need.rs @@ -0,0 +1,297 @@ +//! How a job's history becomes the CPU and memory it reserves, step by step. +//! +//! The runner books what `derive` returns, and stores its steps with the run +//! (`runs.need_why`), so `taskguard dump` can show later why a run reserved +//! what it did. `dump` calls the same function for the next run, so the two +//! never tell different stories. + +use crate::config::Config; +use crate::db::{Db, Learned}; +use crate::report::gb; +use serde::Serialize; +use std::path::Path; + +/// One step that set or changed the need. `cpu` and `mem_kb` are the need +/// after the step; a step that leaves one alone repeats it. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct Step { + pub step: &'static str, + pub cpu: f64, + pub mem_kb: u64, + pub why: String, +} + +/// The need a run starts with, and how it was made. +#[derive(Debug, Clone, PartialEq, Serialize)] +pub struct Need { + pub cpu: f64, + pub mem_kb: u64, + /// A minimum raised the need above what was learned. + pub raised_by_min: bool, + /// The history said more memory than `mem_max` allows; this much. + #[serde(skip_serializing_if = "Option::is_none")] + pub mem_capped_from: Option, + /// The key whose history counted: the job's own, or tsc-queue's. + #[serde(skip_serializing_if = "Option::is_none")] + pub history_key: Option, + pub steps: Vec, +} + +/// A first guess for a job with no history. +#[derive(Debug, Clone, PartialEq, Default)] +pub struct Guess { + pub mem_kb: Option, + pub cpu: Option, + pub dur_s: Option, + /// Where it comes from, in words. + pub from: String, +} + +/// Everything `derive` reads. +pub struct Inputs<'a> { + pub learned: &'a Learned, + pub history_key: Option<&'a str>, + /// Set when the job has no runs of its own. + pub guess: Option<&'a Guess>, + pub min_cpu: Option, + pub min_mem_kb: Option, + pub ncpu: f64, + pub mem_total_kb: u64, + pub cpu_max: f64, + pub mem_max: f64, + pub new_job_mem_kb: u64, + pub keep: usize, + pub boost_runs: usize, +} + +/// The need a run starts with, and each step that made it. +pub fn derive(i: &Inputs) -> Need { + let l = i.learned; + let mut steps = Vec::new(); + let (mut cpu, mut mem) = (0.0, 0u64); + if l.runs > 0 { + cpu = l.cpu.unwrap_or(0.0); + mem = l.mem_kb.unwrap_or(0); + steps.push(Step { step: "history", cpu, mem_kb: mem, why: history_why(l, i.keep, i.history_key) }); + if l.mem_boosted { + mem = (mem as f64 * 1.25) as u64; + steps.push(Step { + step: "boost", + cpu, + mem_kb: mem, + why: format!("+25%: one of the last {} runs was starved of memory and reached its need", i.boost_runs), + }); + } + } else if let Some(g) = i.guess { + cpu = g.cpu.unwrap_or(1.0); + mem = g.mem_kb.unwrap_or(i.new_job_mem_kb); + steps.push(Step { step: "first_guess", cpu, mem_kb: mem, why: format!("no runs yet: {}", g.from) }); + } + // What the history alone says: a first guess is no learned need. + let learned_mem = if l.runs > 0 { mem } else { 0 }; + // A job that once wanted more cores than the machine has is capped at the + // limit; otherwise it could only ever start on an empty machine. + let cpu_cap = (i.ncpu * i.cpu_max / 100.0).min(i.ncpu); + if cpu > cpu_cap { + cpu = cpu_cap; + steps.push(Step { step: "cpu_cap", cpu, mem_kb: mem, why: format!("capped at cpu_max {:.0}% of {} cores", i.cpu_max, i.ncpu) }); + } + if let Some(m) = i.min_cpu.filter(|m| *m > cpu) { + cpu = m; + steps.push(Step { step: "min_cpu", cpu, mem_kb: mem, why: format!("raised to min_cpu {m:.1}") }); + } + if let Some(m) = i.min_mem_kb.filter(|m| *m > mem) { + mem = m; + steps.push(Step { step: "min_mem", cpu, mem_kb: mem, why: format!("raised to min_mem {}", gb(m)) }); + } + // A job that once took more memory than the limit allows is capped at the + // limit, as CPU is: otherwise it could only ever start on an empty machine. + let mem_cap = (i.mem_total_kb as f64 * i.mem_max / 100.0) as u64; + let mut mem_capped_from = None; + if mem_cap > 0 && mem > mem_cap && i.min_mem_kb.is_none_or(|m| m <= mem_cap) { + mem_capped_from = Some(mem); + mem = mem_cap; + steps.push(Step { + step: "mem_cap", + cpu, + mem_kb: mem, + why: format!("capped at mem_max {:.0}% of {}", i.mem_max, gb(i.mem_total_kb)), + }); + } + let raised_by_min = i.min_cpu.is_some_and(|m| m > l.cpu.unwrap_or(0.0)) || i.min_mem_kb.is_some_and(|m| m > learned_mem); + Need { cpu, mem_kb: mem, raised_by_min, mem_capped_from, history_key: i.history_key.map(str::to_string), steps } +} + +/// Why the history says what it says about memory and CPU, in words. +fn history_why(l: &Learned, keep: usize, key: Option<&str>) -> String { + let n = l.runs; + let mut why = Vec::new(); + if let Some(k) = key.filter(|k| k.starts_with("tsc-queue:")) { + why.push(format!("no runs of its own; the history of {k}")); + } + let mem = l.mem_kb.map(gb).unwrap_or("?".into()); + if let Some(o) = &l.outliers { + let top = o.runs.iter().map(|r| r.peak_kb).max().unwrap_or(0); + why.push(format!( + "memory {mem}: {} outlier run(s) up to {} above the usual {}, {}", + o.runs.len(), + gb(top), + gb(o.normal_kb), + o.why(&o.runs[0]) + )); + } else { + why.push(format!("memory {mem}: the highest peak of the last {n} run(s) (keep {keep})")); + } + if !l.failed.is_empty() { + let top = l.failed.iter().map(|r| r.peak_kb).max().unwrap_or(0); + why.push(format!("{} failed run(s) up to {} ignored", l.failed.len(), gb(top))); + } + if let Some(c) = l.cpu { + why.push(format!("cpu {c:.1}: the cores the recent runs used, newest counting most")); + } + why.join("; ") +} + +/// The history that counts for a job: its own, or for a compile with none, +/// what tsc-queue learned. Returns it with the key it came from. +pub fn history(db: &Db, cfg: &Config, cwd: &Path, argv: &[String], key: &str) -> (Learned, String) { + let learned = db.learned(key, &cfg.learn()).unwrap_or_default(); + let tool = tool(argv); + if learned.runs == 0 && matches!(tool.as_str(), "tsc" | "tsgo") { + let legacy = format!("tsc-queue:{}", crate::commands::legacy_key(cwd)); + if let Ok(l) = db.learned(&legacy, &cfg.learn()) + && l.runs > 0 + { + return (l, legacy); + } + } + (learned, key.to_string()) +} + +/// A first guess for a job with no history: what similar jobs needed, or +/// new_job_mem without any. +pub fn guess(db: Option<&Db>, cfg: &Config, package: Option<&str>, argv: &[String], label: Option<&str>, pool: Option<&str>) -> Guess { + let cmdline = argv.join(" "); + let same = package.map(|p| (p, cmdline.as_str())); + let est = db.and_then(|d| d.estimate(same, label, &tool(argv), pool).ok().flatten()); + match est { + Some(e) if e.mem_kb.is_some() => Guess { mem_kb: e.mem_kb, cpu: e.cpu, dur_s: e.dur_s, from: e.from }, + e => Guess { + mem_kb: Some(cfg.new_job_mem_kb), + cpu: e.as_ref().and_then(|e| e.cpu), + dur_s: e.and_then(|e| e.dur_s), + from: "the new_job_mem default".into(), + }, + } +} + +/// The program a command runs: `tsc` for `bunx tsc -p .`. +pub fn tool(argv: &[String]) -> String { + crate::matcher::effective(argv).first().map(|c| c.rsplit('/').next().unwrap_or(c).to_string()).unwrap_or_default() +} + +/// The `Inputs` of a job from the config and this machine. +pub fn inputs<'a>( + cfg: &Config, + learned: &'a Learned, + history_key: Option<&'a str>, + guess: Option<&'a Guess>, + min_cpu: Option, + min_mem_kb: Option, +) -> Inputs<'a> { + Inputs { + learned, + history_key, + guess: guess.filter(|_| learned.runs == 0), + min_cpu, + min_mem_kb, + ncpu: crate::sys::ncpu() as f64, + mem_total_kb: crate::sys::mem_total_kb(), + cpu_max: cfg.cpu_max, + mem_max: cfg.mem_max, + new_job_mem_kb: cfg.new_job_mem_kb, + keep: cfg.hist_keep, + boost_runs: cfg.boost_runs, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const GB: u64 = 1024 * 1024; + + fn base(l: &Learned) -> Inputs<'_> { + Inputs { + learned: l, + history_key: Some("k"), + guess: None, + min_cpu: None, + min_mem_kb: None, + ncpu: 10.0, + mem_total_kb: 32 * GB, + cpu_max: 100.0, + mem_max: 85.0, + new_job_mem_kb: 1536 * 1024, + keep: 10, + boost_runs: 5, + } + } + + fn names(n: &Need) -> Vec<&str> { + n.steps.iter().map(|s| s.step).collect() + } + + #[test] + fn the_history_sets_the_need_and_each_step_says_why() { + let l = Learned { runs: 4, mem_kb: Some(3 * GB), cpu: Some(2.0), ..Default::default() }; + let n = derive(&base(&l)); + assert_eq!((n.cpu, n.mem_kb, n.raised_by_min), (2.0, 3 * GB, false)); + assert_eq!(names(&n), ["history"]); + assert!(n.steps[0].why.contains("highest peak of the last 4 run(s)"), "{}", n.steps[0].why); + } + + #[test] + fn the_boost_min_and_cap_each_leave_a_step() { + let l = Learned { runs: 4, mem_kb: Some(24 * GB), cpu: Some(12.0), mem_boosted: true, ..Default::default() }; + let mut i = base(&l); + i.min_cpu = Some(11.0); + let n = derive(&i); + // 24 GB * 1.25 = 30 GB, above mem_max 85% of 32 GB (27.2 GB). + assert_eq!(names(&n), ["history", "boost", "cpu_cap", "min_cpu", "mem_cap"]); + assert_eq!(n.mem_capped_from, Some(30 * GB)); + assert_eq!(n.mem_kb, ((32 * GB) as f64 * 85.0 / 100.0) as u64); + assert_eq!(n.cpu, 11.0); + // The minimum only lifts the capped need, not what was learned (12). + assert!(!n.raised_by_min); + } + + #[test] + fn a_min_above_the_cap_is_kept() { + let l = Learned::default(); + let mut i = base(&l); + i.min_mem_kb = Some(30 * GB); + let n = derive(&i); + assert_eq!((n.mem_kb, n.mem_capped_from), (30 * GB, None)); + assert!(n.raised_by_min); + } + + #[test] + fn a_first_run_takes_the_guess() { + let l = Learned::default(); + let g = Guess { mem_kb: Some(2 * GB), cpu: None, dur_s: None, from: "typical of 3 tsc jobs".into() }; + let mut i = base(&l); + i.guess = Some(&g); + let n = derive(&i); + assert_eq!((n.cpu, n.mem_kb), (1.0, 2 * GB)); + assert_eq!(n.steps[0].why, "no runs yet: typical of 3 tsc jobs"); + } + + #[test] + fn ignored_failed_runs_are_named() { + let failed = crate::db::Outlier { run_id: 7, peak_kb: 14 * GB, ended_at: 0.0, exit: Some(1) }; + let l = Learned { runs: 3, mem_kb: Some(GB / 2), failed: vec![failed], ..Default::default() }; + let n = derive(&base(&l)); + assert!(n.steps[0].why.contains("1 failed run(s) up to 14.0 GB ignored"), "{}", n.steps[0].why); + } +} diff --git a/src/runner.rs b/src/runner.rs index 58df647..3f0d0c3 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -7,6 +7,7 @@ use crate::insight::{self, AdviceInput, RunFacts, Tracker}; use crate::key; use crate::machine; use crate::matcher; +use crate::need; use crate::queue::{self, Decision, Entry, Queue}; use crate::report::{self, Lines, say}; use crate::sys; @@ -272,14 +273,10 @@ pub fn run(mut o: Opts) -> Result { let startup = cfg.long_lived(&cls, pool.as_deref()); let database = Db::open_dir(&dir).ok(); - let learned = database.as_ref().and_then(|d| d.learned(&jkey, &cfg.learn()).ok()).unwrap_or_default(); // A compile with no history of its own falls back on what tsc-queue learned. - let tool = matcher::effective(&o.cmd).first().map(|c| c.rsplit('/').next().unwrap_or(c).to_string()).unwrap_or_default(); - let learned = match (&database, learned.runs, tool.as_str()) { - (Some(d), 0, "tsc" | "tsgo") => { - d.learned(&format!("tsc-queue:{}", crate::commands::legacy_key(&cwd)), &cfg.learn()).unwrap_or(learned) - } - _ => learned, + let (learned, history_key) = match &database { + Some(d) => need::history(d, &cfg, &cwd, &o.cmd, &jkey), + None => (Default::default(), jkey.clone()), }; // A command whose recent runs started other taskguard jobs inside it, such // as `bun run ci:local` around a turbo run, is a wrapper. Queued as one @@ -302,45 +299,27 @@ pub fn run(mut o: Opts) -> Result { } exec(&o.cmd); } - let learned_mem = learned.mem_kb.map(|m| if learned.mem_boosted { (m as f64 * 1.25) as u64 } else { m }); - // A job that once wanted more cores than the machine has is capped at the - // limit; otherwise it could only ever start on an empty machine. - let cpu_cap = (sys::ncpu() as f64 * cfg.cpu_max / 100.0).min(sys::ncpu() as f64); // A job with no history still reserves room: what similar jobs needed, or // new_job_mem without any. Counting a first run as zero let a burst of new // compiles start into a machine that could not hold them, and froze it: // each grew to several GB within seconds, faster than the readings showed. let package = package_name(&cwd); let cmdline = o.cmd.join(" "); - let (est_mem, est_cpu, est_dur, estimate_from) = if learned.runs == 0 { - let same = package.as_deref().map(|p| (p, cmdline.as_str())); - let est = database.as_ref().and_then(|d| d.estimate(same, cls.label.as_deref(), &tool, pool.as_deref()).ok().flatten()); - let from = match &est { - Some(e) if e.mem_kb.is_some() => e.from.clone(), - _ => "the new_job_mem default".to_string(), - }; - let (m, c, d) = est.map(|e| (e.mem_kb, e.cpu, e.dur_s)).unwrap_or((None, None, None)); - (Some(m.unwrap_or(cfg.new_job_mem_kb)), Some(c.unwrap_or(1.0)), d, Some(from)) - } else { - (None, None, None, None) - }; - let need_cpu = learned.cpu.or(est_cpu).unwrap_or(0.0).min(cpu_cap).max(min_cpu.unwrap_or(0.0)); - let need_mem = learned_mem.or(est_mem).unwrap_or(0).max(min_mem.unwrap_or(0)); - // A job that once took more memory than the limit allows is capped at the - // limit, as CPU is: otherwise it could only ever start on an empty machine. - let mem_cap = (sys::mem_total_kb() as f64 * cfg.mem_max / 100.0) as u64; - let need_mem = if mem_cap > 0 && need_mem > mem_cap && min_mem.is_none_or(|m| m <= mem_cap) { + let guess = (learned.runs == 0) + .then(|| need::guess(database.as_ref(), &cfg, package.as_deref(), &o.cmd, cls.label.as_deref(), pool.as_deref())); + let need = need::derive(&need::inputs(&cfg, &learned, Some(&history_key), guess.as_ref(), min_cpu, min_mem)); + if let Some(from) = need.mem_capped_from { say(&format!( "warning {jkey} - its history says {} of memory, more than mem_max {:.0}% of RAM allows; it reserves {} (see taskguard outliers)", - report::gb(need_mem), + report::gb(from), cfg.mem_max, - report::gb(mem_cap) + report::gb(need.mem_kb) )); - mem_cap - } else { - need_mem - }; - let raised = min_cpu.is_some_and(|m| m > learned.cpu.unwrap_or(0.0)) || min_mem.is_some_and(|m| m > learned_mem.unwrap_or(0)); + } + let (need_cpu, need_mem, raised) = (need.cpu, need.mem_kb, need.raised_by_min); + let need_why = serde_json::to_string(&need).ok(); + let estimate_from = guess.as_ref().map(|g| g.from.clone()); + let est_dur = guess.as_ref().and_then(|g| g.dur_s); // `bunx` and `npx` set the variable to their own name; that is no script. let script = std::env::var("npm_lifecycle_event").ok().filter(|s| !matches!(s.as_str(), "bunx" | "npx" | "pnpx" | "dlx" | "exec")); @@ -380,6 +359,7 @@ pub fn run(mut o: Opts) -> Result { min_mem_kb: min_mem, need_cpu, need_mem_kb: need_mem, + need_why: need_why.as_deref(), }) .ok() }) diff --git a/tests/cli.rs b/tests/cli.rs index 533c4d8..8a7b065 100644 --- a/tests/cli.rs +++ b/tests/cli.rs @@ -223,6 +223,36 @@ fn a_run_is_learned_and_explained() { assert!(v["machine"]["ncpu"].as_u64().unwrap() >= 1); } +#[test] +fn dump_shows_what_a_job_reserves_and_why() { + let e = Env::new(""); + assert!(e.run(&["--key", "k", "--", "sh", "-c", "exit 0"]).status.success()); + let dump = |args: &[&str]| -> serde_json::Value { + let o = e.run(args); + assert!(o.status.success(), "{args:?}: {}", stderr(&o)); + serde_json::from_slice(&o.stdout).unwrap() + }; + let v = dump(&["dump", "k"]); + let job = &v["jobs"][0]; + assert_eq!(job["key"], "k"); + // The run stored how its need was made: a first run guesses. + let run = &job["runs"][0]; + assert_eq!(run["need_why"]["steps"][0]["step"], "first_guess", "{run}"); + assert_eq!(run["role"]["in_window"], true); + // The next run reserves from the history, and says so. + assert_eq!(job["next_need"]["steps"][0]["step"], "history", "{job}"); + assert!(job["config"]["rules"]["keep"].as_u64().is_some()); + // The same job by one of its runs. + let id = run["id"].as_i64().unwrap().to_string(); + let by_run = dump(&["dump", "--run", &id]); + assert_eq!(by_run["jobs"][0]["key"], "k"); + assert_eq!(by_run["focus_run"].as_i64().unwrap().to_string(), id); + // A command, matched as doctor --explain does. + let by_cmd = dump(&["dump", "--cmd", "bunx --bun playwright test"]); + assert_eq!(by_cmd["jobs"][0]["match"]["pool"]["name"], "e2e"); + assert!(!e.run(&["dump", "nope*"]).status.success()); +} + #[test] fn a_cpu_starved_run_is_flagged_and_advised() { let e = Env::new(""); @@ -559,7 +589,7 @@ fn every_command_has_help_and_help_all_has_them_all() { let h = out(&["outliers", "--ratio", "1.2", "--help"]); assert!(h.starts_with("taskguard outliers"), "{h}"); assert_eq!(out(&["help", "outliers"]), h); - for cmd in ["top", "status", "pause", "resume", "history", "prune", "doctor", "import-history", "version", "help"] { + for cmd in ["top", "status", "pause", "resume", "history", "prune", "dump", "doctor", "import-history", "version", "help"] { let h = out(&[cmd, "--help"]); assert!(h.contains(&format!("taskguard {cmd}")), "{cmd}: {h}"); }