fix: wrapper commands run straight through, and a smaller memory boost

CHANGELOG.md

@@ -12,6 +12,16 @@ and the versions follow [semver](https://semver.org).
 
 ### Changed
 
+- A command that wraps other taskguard jobs, such as `taskguard bun run
+  ci:local` around a turbo run, runs straight through from its second run on,
+  and the jobs inside it queue on their own. Queued as one job, it booked its
+  biggest run, up to 20 GB, also when turbo's cache left almost nothing to do;
+  on one machine such wrappers waited 995 minutes in 17 days. A long-lived
+  job, a job in a pool, and a job with a minimum stay queued.
+  `taskguard prune 'KEY' --apply` queues the command again. (#24)
+- A run starved of memory raises the next run's need by 25% only when it
+  reached 90% of its need. One starved far below its need was squeezed by
+  other programs, and the extra room went unused. (#24)
 - macOS: the kernel's memory warning level no longer stops every new job.
   At warning, jobs start while memory fits without any leeway, and not while
   a first run still grows (at most two minutes). A dev stack kept the level

README.md

@@ -312,6 +312,18 @@ straight through, unqueued and unmeasured. A call nested inside a
 running job starts at once and takes no second slot, even when a task runner in
 between drops environment variables.
 
+A command that wraps other taskguard jobs, such as `taskguard bun run ci:local`
+around a turbo run whose package scripts start with `taskguard`, runs straight
+through from its second run on. Queued as one job, it booked the memory of its
+biggest run, a full build, also when turbo's cache left almost nothing to do,
+and the jobs inside it ran unqueued under its claim. Now each nested call
+leaves a note for the job it runs in, and a command whose recent runs started
+taskguard jobs inside them runs straight through, with one line saying so. The
+jobs inside it then queue on their own, each by its own need. A long-lived job,
+a job in a pool and a job with a minimum stay queued: their claim is on
+purpose. `taskguard prune 'KEY' --apply` forgets the history, so the command
+queues again.
+
 Long-running commands are never queued: watch modes, dev servers, `--version`
 and `--help` run straight through (the list is in `src/builtin.rs`). Add your
 own with `passthrough = ["storybook", "my-dev-server"]` in a config file; it
@@ -359,7 +371,9 @@ instructions, so it knows the whole tool at once. `taskguard help COMMAND`, or
 taskguard also measures whether a job was held back while it ran: its threads
 waited for a core more than half as long as they ran, it paged memory in while the machine was under memory pressure, or it
 took 1.5 times its usual time while the machine was full. A run starved of
-memory teaches a higher need for the next run; a run starved of CPU does not
+memory that reached 90% of its need teaches a higher need for the next run
+(+25%); one starved far below its need was squeezed by others, and more room
+would not have helped it. A run starved of CPU does not
 raise the CPU need (see "Needs are learned"). With hints on it prints advice
 that can be pasted:
 

src/db.rs

@@ -59,7 +59,8 @@ CREATE TABLE IF NOT EXISTS runs (
   paused_s REAL,
   package TEXT,
   steady_cores_wanted REAL,
-  steady_mem_kb INTEGER
+  steady_mem_kb INTEGER,
+  nested_jobs INTEGER
 );
 CREATE INDEX IF NOT EXISTS runs_key ON runs(key, ended_at);
 CREATE INDEX IF NOT EXISTS runs_ended ON runs(ended_at);
@@ -175,6 +176,7 @@ CREATE TABLE IF NOT EXISTS pruned_runs (
   package TEXT,
   steady_cores_wanted REAL,
   steady_mem_kb INTEGER,
+  nested_jobs INTEGER,
   pruned_at REAL NOT NULL,
   prune_reason TEXT NOT NULL
 );
@@ -186,7 +188,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";
+  steady_mem_kb, nested_jobs";
 
 pub fn now() -> f64 {
     std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map(|d| d.as_secs_f64()).unwrap_or(0.0)
@@ -235,8 +237,15 @@ pub struct Learned {
     /// peaks and the median of the cores wanted on average. None for others.
     pub steady_mem_kb: Option<u64>,
     pub steady_cpu: Option<f64>,
+    /// The most taskguard jobs one of the recent runs started inside itself,
+    /// when any did: the command wraps other taskguard jobs. None otherwise.
+    pub nested_jobs: Option<u64>,
 }
 
+/// A run starved of memory raises the need when its peak reached this share
+/// of the need it started with.
+pub const BOOST_REACHED: f64 = 0.9;
+
 /// How history turns into needs.
 #[derive(Debug, Clone, PartialEq)]
 pub struct Learn {
@@ -477,6 +486,8 @@ pub struct RunResult {
     /// A long-lived job after its start-up: cores wanted on average, and
     /// the memory peak. None for other jobs.
     pub steady: Option<(f64, u64)>,
+    /// How many taskguard calls ran nested inside this run.
+    pub nested_jobs: u64,
 }
 
 fn median(v: &mut [f64]) -> Option<f64> {
@@ -529,24 +540,33 @@ impl Db {
                 conn.execute_batch(&format!("ALTER TABLE runs ADD COLUMN {col} {kind}"))?;
             }
         }
+        // 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"))?;
+            }
+        }
         Ok(Db { conn })
     }
 
     pub fn learned(&self, key: &str, rules: &Learn) -> Result<Learned> {
         let (keep, boost_runs) = (rules.keep, rules.boost_runs);
         let mut stmt = self.conn.prepare_cached(
             "SELECT peak_mem_kb, coalesce(cores_used, cores_wanted), ended_at - started_at - coalesce(paused_s, 0), starved, steady_mem_kb,
-                    steady_cores_wanted, id, ended_at, exit
+                    steady_cores_wanted, id, ended_at, exit, need_mem_kb, nested_jobs
              FROM runs WHERE key = ?1 AND ended_at IS NOT NULL AND peak_mem_kb > 0
              ORDER BY ended_at DESC, id DESC LIMIT ?2",
         )?;
-        // peak memory, cores used, duration, starved, and after the start-up: memory peak, cores wanted
-        type Row = (i64, Option<f64>, Option<f64>, Option<String>, Option<i64>, Option<f64>);
+        // peak memory, cores used, duration, starved, after the start-up: memory peak and cores wanted,
+        // then the memory need it started with and the jobs nested inside it
+        type Row = (i64, Option<f64>, Option<f64>, Option<String>, Option<i64>, Option<f64>, Option<i64>, Option<i64>);
         let rows: Vec<(Row, Outlier)> = stmt
             .query_map(params![key, keep.max(boost_runs) as i64], |r| {
                 let peak: i64 = r.get(0)?;
                 let run = Outlier { run_id: r.get(6)?, peak_kb: peak as u64, ended_at: r.get(7)?, exit: r.get(8)? };
-                Ok(((peak, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?), run))
+                Ok(((peak, r.get(1)?, r.get(2)?, r.get(3)?, r.get(4)?, r.get(5)?, r.get(9)?, r.get(10)?), run))
             })?
             .collect::<std::result::Result<_, _>>()?;
         let (rows, runs): (Vec<Row>, Vec<Outlier>) = rows.into_iter().unzip();
@@ -566,10 +586,28 @@ impl Db {
             .collect();
         let cpu = cpu_need(&cpu_runs, rules);
         let dur = median_of(recent.iter().filter_map(|r| r.2).collect());
-        let boosted = rows.iter().take(boost_runs).any(|r| r.3.as_deref() == Some("memory"));
+        // A run starved of memory raises the need only when it reached what it
+        // was given: then its need was too small. One that starved far below
+        // its need was squeezed by others, and more room for it would have
+        // gone unused. A run from before the need was recorded still counts.
+        let boosted = rows
+            .iter()
+            .take(boost_runs)
+            .any(|r| r.3.as_deref() == Some("memory") && r.6.is_none_or(|need| r.0 as f64 >= need as f64 * BOOST_REACHED));
         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());
-        Ok(Learned { runs: recent.len(), mem_kb: mem, outliers, cpu, dur_s: dur, mem_boosted: boosted, steady_mem_kb, steady_cpu })
+        let nested_jobs = recent.iter().filter_map(|r| r.7).filter(|n| *n > 0).max().map(|n| n as u64);
+        Ok(Learned {
+            runs: recent.len(),
+            mem_kb: mem,
+            outliers,
+            cpu,
+            dur_s: dur,
+            mem_boosted: boosted,
+            steady_mem_kb,
+            steady_cpu,
+            nested_jobs,
+        })
     }
 
     /// Every key with finished runs, or those that match `pattern`: a key, or
@@ -737,7 +775,7 @@ impl Db {
         self.conn.execute(
             "UPDATE runs SET ended_at = ?2, exit = ?3, peak_mem_kb = ?4, cores_used = ?5, cores_wanted = ?6,
                 cpu_seconds = ?7, runnable_seconds = ?8, pageins = ?9, starved = ?10, starved_detail = ?11,
-                machine_full_frac = ?12, paused_s = ?13, steady_cores_wanted = ?14, steady_mem_kb = ?15
+                machine_full_frac = ?12, paused_s = ?13, steady_cores_wanted = ?14, steady_mem_kb = ?15, nested_jobs = ?16
              WHERE id = ?1",
             params![
                 id,
@@ -754,7 +792,8 @@ impl Db {
                 r.machine_full_frac,
                 (r.paused_s > 0.0).then_some(r.paused_s),
                 r.steady.filter(|_| r.measured).map(|s| s.0),
-                r.steady.filter(|_| r.measured).map(|s| s.1 as i64)
+                r.steady.filter(|_| r.measured).map(|s| s.1 as i64),
+                r.nested_jobs as i64
             ],
         )?;
         Ok(())
@@ -1184,6 +1223,42 @@ mod tests {
         assert_eq!(db.starved_streak("k").unwrap(), 1);
     }
 
+    #[test]
+    fn a_starved_run_raises_the_need_only_when_it_reached_it() {
+        let tmp = tempfile::tempdir().unwrap();
+        let db = Db::open_dir(tmp.path()).unwrap();
+        let starved = |peak: u64, need: u64| {
+            let id = db.insert_run(&NewRun { ns: "n", key: "k", need_mem_kb: need, ..Default::default() }).unwrap();
+            db.mark_started(id, now() - 10.0, 0.0, None).unwrap();
+            let r = RunResult { ended_at: now(), peak_mem_kb: peak, measured: true, starved: Some("memory".into()), ..Default::default() };
+            db.finish_run(id, &r).unwrap();
+        };
+        // Squeezed by others at a third of its need: more room would have gone unused.
+        starved(1000, 3000);
+        assert!(!db.learned("k", &Learn::default()).unwrap().mem_boosted);
+        // It reached its need: the need was too small.
+        starved(2900, 3000);
+        assert!(db.learned("k", &Learn::default()).unwrap().mem_boosted);
+    }
+
+    #[test]
+    fn a_run_with_jobs_inside_it_makes_its_command_a_wrapper() {
+        let tmp = tempfile::tempdir().unwrap();
+        let db = Db::open_dir(tmp.path()).unwrap();
+        let finish = |nested: u64| {
+            let id = db.insert_run(&NewRun { ns: "n", key: "ci", ..Default::default() }).unwrap();
+            db.mark_started(id, now() - 10.0, 0.0, None).unwrap();
+            db.finish_run(id, &RunResult { ended_at: now(), peak_mem_kb: 1000, measured: true, nested_jobs: nested, ..Default::default() })
+                .unwrap();
+        };
+        finish(0);
+        assert_eq!(db.learned("ci", &Learn::default()).unwrap().nested_jobs, None);
+        finish(37);
+        // A later run with turbo's cache full starts none; the command still wraps.
+        finish(0);
+        assert_eq!(db.learned("ci", &Learn::default()).unwrap().nested_jobs, Some(37));
+    }
+
     /// A run that used `used` cores while it wanted `wanted`.
     fn run_cpu(db: &Db, key: &str, used: f64, wanted: f64, starved: Option<&str>) {
         let id = db.insert_run(&NewRun { ns: "n", key, ..Default::default() }).unwrap();

src/queue.rs

@@ -258,7 +258,7 @@ pub struct Guard {
 
 impl Queue {
     pub fn open(dir: &Path) -> Result<Queue> {
-        for d in ["wait", "run", "viewers", "nudge", "pipeline", "stall"] {
+        for d in ["wait", "run", "viewers", "nudge", "pipeline", "stall", "nested"] {
             fs::create_dir_all(dir.join(d)).with_context(|| format!("creating {}", dir.join(d).display()))?;
         }
         Ok(Queue { dir: dir.to_path_buf() })
@@ -285,6 +285,24 @@ impl Queue {
         self.dir.join("run").join(pid.to_string())
     }
 
+    /// A taskguard call nested inside the running job of `outer` (a pid)
+    /// leaves `nested/<outer>.<pid>`, so that job learns that it wraps other
+    /// taskguard jobs.
+    pub fn note_nested(&self, outer: i32, pid: i32) {
+        let _ = fs::write(self.dir.join("nested").join(format!("{outer}.{pid}")), "");
+    }
+
+    /// How many nested calls left a note for the job of `owner`. The notes
+    /// are removed, so a later job with the same pid starts from none.
+    pub fn take_nested(&self, owner: i32) -> u64 {
+        let prefix = format!("{owner}.");
+        let Ok(rd) = fs::read_dir(self.dir.join("nested")) else { return 0 };
+        rd.filter_map(|e| e.ok())
+            .filter(|e| e.file_name().to_str().is_some_and(|n| n.starts_with(&prefix)))
+            .filter(|e| fs::remove_file(e.path()).is_ok())
+            .count() as u64
+    }
+
     /// When the task-runner run `id` ("turbo:4242", owned by `pid`) queued
     /// its first job. The first job of the run writes it; the file outlives
     /// that job, so the run keeps its age between waves of jobs.
@@ -2266,6 +2284,18 @@ mod tests {
         assert!(blockers(decide(&m, &LIM, &[], std::slice::from_ref(&me), &me, 1000.0, &[])).is_empty(), "it eases: the job starts");
     }
 
+    #[test]
+    fn nested_calls_leave_notes_for_their_outer_job() {
+        let tmp = tempfile::tempdir().unwrap();
+        let q = Queue::open(tmp.path()).unwrap();
+        q.note_nested(100, 201);
+        q.note_nested(100, 202);
+        q.note_nested(1000, 203);
+        assert_eq!(q.take_nested(100), 2, "only its own notes, not those of pid 1000");
+        assert_eq!(q.take_nested(100), 0, "taken once");
+        assert_eq!(q.take_nested(1000), 1);
+    }
+
     #[test]
     fn at_warning_jobs_start_while_memory_fits_and_no_first_run_grows() {
         // A dev stack past its start-up holds macOS at warning with half of

src/runner.rs

@@ -97,6 +97,16 @@ fn held_by_ancestor(q: &Queue) -> bool {
     sys::ancestors(me, &procs).iter().skip(1).any(|p| q.run_path(*p).exists())
 }
 
+/// Tell the running job this nested call runs inside that it wraps other
+/// taskguard jobs: see `queue::Queue::note_nested`.
+fn note_nested(q: &Queue) {
+    let procs = sys::list_procs();
+    let me = std::process::id() as i32;
+    if let Some(outer) = sys::ancestors(me, &procs).into_iter().skip(1).find(|p| q.run_path(*p).exists()) {
+        q.note_nested(outer, me);
+    }
+}
+
 /// Task runners whose run groups the jobs it starts. A job belongs to the
 /// nearest one above it.
 const LAUNCHERS: &[&str] = &["turbo", "nx", "make", "gmake", "moon", "lage", "just"];
@@ -182,7 +192,11 @@ pub fn run(mut o: Opts) -> Result<i32> {
     }
     let dir = config::state_dir();
     let q = Queue::open(&dir)?;
-    if std::env::var("TASKGUARD_DISABLE").is_ok_and(|v| v == "1") || held_by_ancestor(&q) {
+    if std::env::var("TASKGUARD_DISABLE").is_ok_and(|v| v == "1") {
+        exec(&o.cmd);
+    }
+    if held_by_ancestor(&q) {
+        note_nested(&q);
         exec(&o.cmd);
     }
 
@@ -238,6 +252,27 @@ pub fn run(mut o: Opts) -> Result<i32> {
         }
         _ => learned,
     };
+    // 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
+    // job, it books its biggest run, a full build, also when turbo's cache
+    // leaves almost nothing to do, and the jobs inside it run straight through
+    // under its claim. So it runs straight through instead, and the jobs
+    // inside it queue on their own, each by its own need. Not a long-lived job,
+    // whose start-up builds belong to its claim, nor a job in a pool or with a
+    // minimum, which someone gave it on purpose.
+    if let Some(n) = learned.nested_jobs
+        && startup.is_none()
+        && pool.is_none()
+        && min_mem.is_none()
+        && min_cpu.is_none()
+    {
+        if !o.quiet {
+            say(&format!(
+                "straight through {jkey} - its recent runs started up to {n} taskguard jobs inside it, and those queue on their own (taskguard prune '{jkey}' --apply queues it again)"
+            ));
+        }
+        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.
@@ -295,6 +330,8 @@ pub fn run(mut o: Opts) -> Result<i32> {
         real.strip_prefix(&real_checkout).map(|r| r.display().to_string()).unwrap_or(p)
     });
     let me_pid = std::process::id() as i32;
+    // Notes left for an earlier job with the same pid.
+    q.take_nested(me_pid);
     let run_id = database
         .as_ref()
         .and_then(|d| {
@@ -810,6 +847,7 @@ pub fn run(mut o: Opts) -> Result<i32> {
                 measured,
                 paused_s: me.paused_s,
                 steady: tracker.steady(ended, me.paused_s - paused_before_steady),
+                nested_jobs: q.take_nested(me_pid),
             },
         );
     }

tests/cli.rs

@@ -287,6 +287,28 @@ fn a_nested_call_runs_inside_the_outer_slot() {
     assert_eq!(runs, 1, "the nested call takes no slot and records no run");
 }
 
+#[test]
+fn a_command_that_wraps_taskguard_jobs_runs_straight_through_next_time() {
+    // The first run queues as one job; its nested call leaves a note. The
+    // next run of the same command runs straight through, and the call
+    // inside it queues as a job of its own.
+    let e = Env::new("");
+    let inner = format!("{} -- sh -c 'echo inner'", env!("CARGO_BIN_EXE_taskguard"));
+    let first = e.run(&["--", "sh", "-c", &inner]);
+    assert!(first.status.success(), "{}", stderr(&first));
+    assert!(!stderr(&first).contains("straight through"), "{}", stderr(&first));
+    let nested: i64 = e.db().query_row("SELECT nested_jobs FROM runs", [], |r| r.get(0)).unwrap();
+    assert_eq!(nested, 1);
+
+    let second = e.run(&["--", "sh", "-c", &inner]);
+    assert!(second.status.success(), "{}", stderr(&second));
+    assert!(stderr(&second).contains("straight through"), "{}", stderr(&second));
+    let keys: Vec<String> =
+        e.db().prepare("SELECT key FROM runs ORDER BY id").unwrap().query_map([], |r| r.get(0)).unwrap().map(Result::unwrap).collect();
+    assert_eq!(keys.len(), 2, "the wrapper once, then the job inside it: {keys:?}");
+    assert_ne!(keys[0], keys[1]);
+}
+
 /// Wait until a waiting entry exists for `pid`, so a nudge reaches a job that waits.
 fn wait_until_queued(e: &Env, pid: u32) {
     let t = Instant::now();