Admin runner settings: job time limits; each job runs under its own 5dead1f4f26d

-- Each job's own time limit, fixed when it is claimed (oxygenflow#23,

-- oxygenhub#89): the pipeline's timeout, else the site default, capped at

-- the site maximum. 0 for jobs claimed before: the site default.

ALTER TABLE flow_jobs ADD COLUMN timeout_s INTEGER NOT NULL DEFAULT 0;

{% extends "base.html" %}

{% block title %}Runner settings · Administration{% endblock %}

{% block content %}

<div class="head"><h1>Runners <small><a href="/admin">Administration</a></small></h1></div>

<div class="bar"><span class="modes"><a href="/admin/runners">runners</a><a href="/admin/runners/settings" class="on">settings</a></span></div>

{% if let Some(n) = notice %}<p class="note">{{ n }}</p>{% endif %}

{% if let Some(e) = error %}<p class="error">{{ e }}</p>{% endif %}

<form class="card wide" method="post" action="/admin/runners/settings">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Job time limits</h2>

  <p class="small mute">A pipeline sets a job's limit with <code>timeout "90m"</code> on the job or its sandbox. These apply to every repository on this hub, to jobs claimed from now on.</p>

  <label for="default_minutes">Default, for a job whose pipeline sets none (minutes)</label>

  <input type="number" id="default_minutes" name="default_minutes" min="1" max="1440" value="{{ default_m }}" required>

  <p class="small mute">hub.toml says {{ file_default_m }}; this page's value wins once saved.</p>

  <label for="max_minutes">Maximum a pipeline may ask for (minutes)</label>

  <input type="number" id="max_minutes" name="max_minutes" min="1" max="1440" value="{{ max_m }}" required>

  <p class="small mute">A longer <code>timeout</code> is cut to this.</p>

  <label for="grace_minutes">Lost after (minutes past a job's limit)</label>

  <input type="number" id="grace_minutes" name="grace_minutes" min="1" max="120" value="{{ grace_m }}" required>

  <p class="small mute">A runner stops a job at its limit and says so at once; a job still running this long after is one its runner no longer reports, and the hub ends it.</p>

  <button type="submit">Save</button>

</form>

<section class="panel">

  <h2>Repositories that run Flow</h2>

  {% if flow_repos.is_empty() %}<p class="mute">None listed in hub.toml; each repository's administrator switches Flow on in its Settings.</p>{% else %}<ul class="plain">{% for r in flow_repos %}<li><code>{{ r }}</code></li>{% endfor %}</ul><p class="small mute">From hub.toml's <code>[flow] repos</code>; others are switched on in each repository's Settings.</p>{% endif %}

</section>

{% endblock %}

- **Runner settings for the site, and a job's own time limit (#89, oxygenflow#23).** Admin → Runners → Settings sets the default job time limit, the most a pipeline may ask for, and how long past its limit a silent job is given up; saved in the database, over hub.toml's `job_timeout_minutes`. A job runs under its pipeline's `timeout` (on the job or its sandbox), capped at the maximum, else the default; the limit is recorded on the job when it is claimed, and the lost-job reaper uses each job's own.

/// Jobs still running at `now` past their own time limit (`default_s` for

/// one claimed without) plus `grace_ms`.

pub async fn running_past_limit(pool: &Pool, now: i64, default_s: u64, grace_ms: i64) -> sqlx::Result<Vec<Job>> {

    sqlx::query_as(

        "SELECT * FROM flow_jobs WHERE status = 'running' \

         AND COALESCE(started_ms, 0) + (CASE WHEN timeout_s > 0 THEN timeout_s ELSE ? END) * 1000 + ? < ? ORDER BY started_ms",

    )

    .bind(default_s as i64)

    .bind(grace_ms)

    .bind(now)

    .fetch_all(pool)

    .await

}

/// Record the limit a claimed job runs under.

pub async fn set_job_timeout(pool: &Pool, job_id: i64, timeout_s: u64) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET timeout_s = ? WHERE id = ?").bind(timeout_s as i64).bind(job_id).execute(pool).await?;

    Ok(())

}

    // The job's limit: its own timeout, else its sandbox's, else the site

    // default, never past the site maximum; recorded for the reaper.

    let limits = limits(&app).await;

    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;

    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;

    let timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);

    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;

        job: spec,

        sandbox,

        timeout_s,

/// The site's job time limits (oxygenhub#89): set on Admin → Runners →

/// Settings, else from `hub.toml`.

pub struct Limits {

    /// For a job whose pipeline names none.

    pub default_s: u64,

    /// The most a pipeline may ask for.

    pub max_s: u64,

    /// Grace past a job's limit before the hub gives it up: a runner stops

    /// a job at its limit and reports it within moments, so one still

    /// running this long after is one the runner no longer reports (it was

    /// restarted, or lost the hub while the job ended, oxygenhub#71).

    pub grace_ms: i64,

}

pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";

pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";

pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";

pub async fn limits(app: &Shared) -> Limits {

    let minutes = |key: &'static str| async move {

        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)

    };

    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));

    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);

    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);

    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }

}

/// End every job still running past its time limit plus the grace:

    let limits = limits(app).await;

    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;

        .route("/admin/runners/settings", get(runners::settings).post(runners::save_settings))

// ---------------------------------------------------------------- settings

#[derive(Template)]

#[template(path = "admin_runner_settings.html")]

pub struct SettingsPage {

    shell: Shell,

    default_m: u64,

    max_m: u64,

    grace_m: i64,

    /// What hub.toml says, shown beside a setting the page overrides.

    file_default_m: u64,

    flow_repos: Vec<String>,

    notice: Option<String>,

    error: Option<String>,

}

async fn settings_page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Page<SettingsPage> {

    let l = crate::flow::limits(app).await;

    Page(SettingsPage {

        shell,

        default_m: l.default_s / 60,

        max_m: l.max_s / 60,

        grace_m: l.grace_ms / 60_000,

        file_default_m: app.config.flow.job_timeout_minutes,

        flow_repos: app.config.flow.repos.clone(),

        notice,

        error,

    })

}

/// `GET /admin/runners/settings` (oxygenhub#89).

pub async fn settings(State(app): State<Shared>, viewer: Viewer) -> Result<Page<SettingsPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);

    require_admin(&viewer, &shell)?;

    Ok(settings_page(&app, shell, None, None).await)

}

#[derive(Deserialize)]

pub struct SettingsForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    default_minutes: String,

    max_minutes: String,

    grace_minutes: String,

}

/// `POST /admin/runners/settings`: the site's job time limits.

pub async fn save_settings(State(app): State<Shared>, viewer: Viewer, Form(f): Form<SettingsForm>) -> Result<Page<SettingsPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let num = |v: &str| v.trim().parse::<u64>().ok().filter(|n| *n > 0);

    let (Some(d), Some(m), Some(g)) = (num(&f.default_minutes), num(&f.max_minutes), num(&f.grace_minutes)) else {

        return Ok(settings_page(&app, shell, None, Some("Each limit is a whole number of minutes, at least 1.".into())).await);

    };

    if d > m {

        return Ok(settings_page(&app, shell, None, Some("The default cannot be longer than the maximum.".into())).await);

    }

    if m > 24 * 60 {

        return Ok(settings_page(&app, shell, None, Some("A job may run for a day at most (1440 minutes).".into())).await);

    }

    let e = |e| Error::from_db(shell.clone(), e);

    accounts::set_setting(&app.db, crate::flow::SETTING_DEFAULT, &d.to_string()).await.map_err(e)?;

    accounts::set_setting(&app.db, crate::flow::SETTING_MAX, &m.to_string()).await.map_err(e)?;

    accounts::set_setting(&app.db, crate::flow::SETTING_GRACE, &g.to_string()).await.map_err(e)?;

    accounts::audit(

        &app.db,

        Some(&me),

        "flow.settings",

        "",

        &format!("job time limit {d} min by default, {m} min at most; lost after {g} min more"),

    )

    .await

    .ok();

    Ok(settings_page(&app, shell, Some("Saved. Jobs claimed from now on run under these limits.".into()), None).await)

}

/// oxygenhub#89: an admin sets the site's job time limits on Admin →

/// Runners → Settings; nobody else can.

#[tokio::test(flavor = "multi_thread")]

async fn admins_set_the_job_time_limits() {

    let h = hub().await;

    let app = &h.app;

    let andy = register(app, "andy", "").await;

    let page = get(app, "/admin/runners/settings", Some(&andy)).await;

    assert_eq!(page.status, StatusCode::OK);

    assert!(page.body.contains("Job time limits") && page.body.contains("value=\"60\""), "the hub.toml default shows");

    let csrf = csrf_of(&page.body);

    let r = post(app, "/admin/runners/settings", Some(&andy), &format!("_csrf={csrf}&default_minutes=200&max_minutes=120&grace_minutes=5"))

        .await;

    assert!(r.body.contains("cannot be longer than the maximum"), "{}", r.body);

    let r = post(app, "/admin/runners/settings", Some(&andy), &format!("_csrf={csrf}&default_minutes=90&max_minutes=240&grace_minutes=10"))

        .await;

    assert!(r.body.contains("Saved."), "{}", r.body);

    let l = oxygenhub::flow::limits(app).await;

    assert_eq!((l.default_s, l.max_s, l.grace_ms), (5400, 14400, 600_000));

    assert!(get(app, "/admin/runners", Some(&andy)).await.body.contains("href=\"/admin/runners/settings\""));

    let code = {

        let csrf = csrf_of(&get(app, "/admin", Some(&andy)).await.body);

        let r = post(app, "/admin/invite", Some(&andy), &format!("_csrf={csrf}")).await;

        r.body.split("register?invite=").nth(1).unwrap().split('<').next().unwrap().trim().to_string()

    };

    let bob = register(app, "bob", &code).await;

    assert_ne!(get(app, "/admin/runners/settings", Some(&bob)).await.status, StatusCode::OK, "not for others");

}

/// oxygenflow#23 and oxygenhub#89: a job runs under its pipeline's

/// timeout, capped at the site's maximum, else the site's default; and the

/// hub gives a job up only past its own limit.

#[tokio::test(flavor = "multi_thread")]

async fn a_job_runs_under_its_own_time_limit_within_the_sites() {

    let h = hub().await;

    db::accounts::create_account(&h.app.db, "andy", "[email protected]", "x", "salt", true).await.unwrap();

    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_DEFAULT, "30").await.unwrap();

    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_MAX, "120").await.unwrap();

    create(&h.app, "andy", "limits").await;

    let url = format!("{}/andy/limits", h.base);

    let work = tempfile::tempdir().unwrap();

    let c = client(&url, work.path());

    let extra = "job \"slow\" {\n    sandbox \"alpine\"\n    timeout \"90m\"\n    exec \"true\"\n}\njob \"huge\" {\n    sandbox \"alpine\"\n    timeout \"5h\"\n    exec \"true\"\n}\n";

    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));

    write(work.path(), "test.sh", "echo ok\n");

    c.commit("Add a pipeline", &[]).unwrap();

    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    assert_eq!(runs_of(&h.app, "andy", "limits").await.len(), 1);

    let mut limits = std::collections::HashMap::new();

    let mut ids = std::collections::HashMap::new();

    while let Some(j) = claim(&h.app).await {

        let name = j["job"]["name"].as_str().unwrap().to_string();

        limits.insert(name.clone(), j["timeout_s"].as_u64().unwrap());

        ids.insert(name, j["job_id"].as_i64().unwrap());

    }

    assert_eq!(limits["slow"], 5400, "its own");

    assert_eq!(limits["huge"], 7200, "cut to the site's maximum");

    assert_eq!(limits["build"], 1800, "the site's default");

    // 40 minutes in: past the default and its grace, within slow's limit.

    let started = db::now_ms() - 40 * 60 * 1000;

    sqlx::query("UPDATE flow_jobs SET started_ms = ?").bind(started).execute(&h.app.db).await.unwrap();

    oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap();

    let status = |id: i64| {

        let app = h.app.clone();

        async move { db::flow::job(&app.db, id).await.unwrap().unwrap().status }

    };

    assert_eq!(status(ids["build"]).await, "error", "past its 30 minutes");

    assert_eq!(status(ids["slow"]).await, "running", "within its 90");

    assert_eq!(status(ids["huge"]).await, "running", "within its 120");

}

<div class="bar"><span class="modes"><a href="/admin/runners" class="on">runners</a><a href="/admin/runners/settings">settings</a></span></div>