OXYGENHUB
# 6907a989e82a oxygenhub / crates / hub / src / web / runners.rs
20.3 KiB · Rust · last changed Admin › Runners › Images: what images are where 23 days ago source blame history raw
1 
//! Admin → Runners (hub spec §13.2): the OxygenFlow runners of this hub,
2 
//! what they are and are running, their history, registration tokens, and
3 
//! draining, disabling and removing them.
4
5 
use super::settings::csrf;
6 
use super::{filters, Error, Page, Shell};
7 
use crate::auth::Viewer;
8 
use crate::db::{self, accounts, runners};
9 
use crate::Shared;
10 
use askama::Template;
11 
use axum::extract::{Form, Path, Query, State};
12 
use axum::response::{IntoResponse, Redirect, Response};
13 
use serde::Deserialize;
14 
use std::collections::HashMap;
15
16 
fn require_admin(viewer: &Viewer, shell: &Shell) -> Result<accounts::Account, Error> {
17 
    match viewer.account() {
18 
        Some(a) if a.is_admin => Ok(a.clone()),
19 
        _ => Err(Error::not_found(shell.clone())),
20 
    }
21 
}
22
23 
pub struct RunnerRow {
24 
    pub r: runners::Runner,
25 
    /// Jobs it is running now.
26 
    pub running: i64,
27 
    /// The repositories it serves, in words.
28 
    pub scope: String,
29 
    /// Where its peers reach it in the image swarm, and its site
30 
    /// (oxygenhub#144); empty when it is not in one.
31 
    pub swarm: String,
32 
}
33
34 
/// A scope in words: the whole site, an owner's repositories, or one
35 
/// repository.
36 
async fn scope_words(app: &Shared, s: &runners::Scope) -> String {
37 
    match s.repo_id {
38 
        Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() {
39 
            Some(r) => format!("{}/{}", r.owner, r.name),
40 
            None => "a removed repository".into(),
41 
        },
42 
        None if s.owner.is_empty() => "the whole site".into(),
43 
        None => format!("{}'s repositories", s.owner),
44 
    }
45 
}
46
47 
/// A scope as typed: empty for the whole site, an owner, or
48 
/// `owner/repository`.
49 
async fn parse_scope(app: &Shared, text: &str) -> Result<runners::Scope, String> {
50 
    let text = text.trim().trim_matches('/');
51 
    if text.is_empty() {
52 
        return Ok(runners::Scope::default());
53 
    }
54 
    if let Some((owner, name)) = text.split_once('/') {
55 
        return match db::repos::get(&app.db, owner, name).await {
56 
            Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }),
57 
            _ => Err(format!("There is no repository {text}.")),
58 
        };
59 
    }
60 
    let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some();
61 
    let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some();
62 
    if user || org {
63 
        Ok(runners::Scope { owner: text.to_string(), repo_id: None })
64 
    } else {
65 
        Err(format!("There is no user or organisation {text}."))
66 
    }
67 
}
68
69 
#[derive(Template)]
70 
#[template(path = "admin_runners.html")]
71 
pub struct RunnersPage {
72 
    shell: Shell,
73 
    runners: Vec<RunnerRow>,
74 
    /// Registration tokens waiting to be used: (created by, created, expires).
75 
    pending: Vec<(String, i64, i64)>,
76 
    /// A token just made, shown once, with the command that uses it.
77 
    new_token: Option<String>,
78 
    hub: String,
79 
    notice: Option<String>,
80 
    error: Option<String>,
81 
    /// Running jobs, then queued ones in the order runners claim them.
82 
    queue: Vec<db::flow::QueueEntry>,
83 
}
84
85 
async fn page(
86 
    app: &Shared,
87 
    shell: Shell,
88 
    new_token: Option<String>,
89 
    notice: Option<String>,
90 
    error: Option<String>,
91 
) -> Result<Page<RunnersPage>, Error> {
92 
    let e = |e| Error::from_db(shell.clone(), e);
93 
    let list = runners::list(&app.db).await.map_err(e)?;
94 
    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();
95 
    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;
96 
    let mut rows = Vec::with_capacity(list.len());
97 
    for r in list {
98 
        let scope = scope_words(app, &r.scope()).await;
99 
        let swarm = app
100 
            .swarm
101 
            .addr(r.id)
102 
            .map(|s| if s.site.is_empty() { s.addr } else { format!("{} · site {}", s.addr, s.site) })
103 
            .unwrap_or_default();
104 
        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, r });
105 
    }
106 
    let queue = db::flow::queue(&app.db).await.map_err(e)?;
107 
    Ok(Page(RunnersPage { hub: app.config.base_url().to_string(), shell, runners: rows, pending, new_token, notice, error, queue }))
108 
}
109
110 
/// `GET /admin/runners`.
111 
pub async fn list(State(app): State<Shared>, viewer: Viewer) -> Result<Page<RunnersPage>, Error> {
112 
    let shell = Shell::new(&app, "/admin/runners", &viewer);
113 
    require_admin(&viewer, &shell)?;
114 
    page(&app, shell, None, None, None).await
115 
}
116
117 
#[derive(Deserialize)]
118 
pub struct CsrfOnly {
119 
    #[serde(rename = "_csrf")]
120 
    csrf_token: String,
121 
}
122
123 
#[derive(Deserialize)]
124 
pub struct ScopeForm {
125 
    #[serde(rename = "_csrf")]
126 
    csrf_token: String,
127 
    /// Empty for the whole site, an owner, or `owner/repository`.
128 
    #[serde(default)]
129 
    scope: String,
130 
}
131
132 
/// `POST /admin/runners/token`: a one-time registration token, shown once,
133 
/// for a runner of the scope given.
134 
pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {
135 
    let shell = Shell::new(&app, "/admin/runners", &viewer);
136 
    let me = require_admin(&viewer, &shell)?;
137 
    csrf(&viewer, &f.csrf_token, &shell)?;
138 
    let scope = match parse_scope(&app, &f.scope).await {
139 
        Ok(s) => s,
140 
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
141 
    };
142 
    let token = format!("flowreg_{}", crate::auth::random_hex(24));
143 
    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)
144 
        .await
145 
        .map_err(|e| Error::from_db(shell.clone(), e))?;
146 
    let words = scope_words(&app, &scope).await;
147 
    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();
148 
    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await
149 
}
150
151 
/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.
152 
pub async fn set_scope(
153 
    State(app): State<Shared>,
154 
    viewer: Viewer,
155 
    Path(id): Path<i64>,
156 
    Form(f): Form<ScopeForm>,
157 
) -> Result<Page<RunnersPage>, Error> {
158 
    let shell = Shell::new(&app, "/admin/runners", &viewer);
159 
    let me = require_admin(&viewer, &shell)?;
160 
    csrf(&viewer, &f.csrf_token, &shell)?;
161 
    let r =
162 
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
163 
    let scope = match parse_scope(&app, &f.scope).await {
164 
        Ok(s) => s,
165 
        Err(e) => return page(&app, shell, None, None, Some(e)).await,
166 
    };
167 
    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;
168 
    let words = scope_words(&app, &scope).await;
169 
    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();
170 
    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await
171 
}
172
173 
#[derive(Deserialize)]
174 
pub struct LimitsForm {
175 
    #[serde(rename = "_csrf")]
176 
    csrf_token: String,
177 
    /// Comma-separated.
178 
    #[serde(default)]
179 
    labels: String,
180 
    /// Minutes; empty for the site's limit.
181 
    #[serde(default)]
182 
    timeout_minutes: String,
183 
}
184
185 
/// `POST /admin/runners/{id}/limits`: the labels an administrator gives a
186 
/// runner and its own job time limit (#98).
187 
pub async fn set_limits(
188 
    State(app): State<Shared>,
189 
    viewer: Viewer,
190 
    Path(id): Path<i64>,
191 
    Form(f): Form<LimitsForm>,
192 
) -> Result<Response, Error> {
193 
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
194 
    let me = require_admin(&viewer, &shell)?;
195 
    csrf(&viewer, &f.csrf_token, &shell)?;
196 
    let r =
197 
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
198 
    let mut labels: Vec<String> = Vec::new();
199 
    for l in f.labels.split([',', ' ']).map(str::trim).filter(|l| !l.is_empty()) {
200 
        if !l.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) || l.len() > 40 {
201 
            return Err(Error::bad(shell, format!("label {l:?}: letters, digits, - _ and . only, up to 40 characters")));
202 
        }
203 
        if !labels.iter().any(|x| x == l) {
204 
            labels.push(l.to_string());
205 
        }
206 
    }
207 
    let timeout = match f.timeout_minutes.trim() {
208 
        "" => None,
209 
        t => match t.parse::<i64>() {
210 
            Ok(m) if (1..=24 * 60).contains(&m) => Some(m),
211 
            _ => return Err(Error::bad(shell, "a time limit is a whole number of minutes from 1 to 1440")),
212 
        },
213 
    };
214 
    runners::set_limits(&app.db, id, &labels.join(","), timeout).await.map_err(|e| Error::from_db(shell.clone(), e))?;
215 
    let words = format!(
216 
        "labels {}; time limit {}",
217 
        if labels.is_empty() { "none".to_string() } else { labels.join(", ") },
218 
        timeout.map(|m| format!("{m} min")).unwrap_or_else(|| "the site's".into())
219 
    );
220 
    accounts::audit(&app.db, Some(&me), "flow.runner.limits", &r.name, &words).await.ok();
221 
    Ok(Redirect::to(&format!("/admin/runners/{id}")).into_response())
222 
}
223
224 
#[derive(Deserialize)]
225 
pub struct StateForm {
226 
    #[serde(rename = "_csrf")]
227 
    csrf_token: String,
228 
    /// `active`, `draining` or `disabled`.
229 
    state: String,
230 
}
231
232 
/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.
233 
pub async fn set_state(
234 
    State(app): State<Shared>,
235 
    viewer: Viewer,
236 
    Path(id): Path<i64>,
237 
    Form(f): Form<StateForm>,
238 
) -> Result<Response, Error> {
239 
    let shell = Shell::new(&app, "/admin/runners", &viewer);
240 
    let me = require_admin(&viewer, &shell)?;
241 
    csrf(&viewer, &f.csrf_token, &shell)?;
242 
    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {
243 
        return Err(Error::bad(shell, "a runner is active, draining or disabled"));
244 
    }
245 
    let r =
246 
        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;
247 
    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;
248 
    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();
249 
    Ok(Redirect::to("/admin/runners").into_response())
250 
}
251
252 
/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.
253 
pub async fn remove(
254 
    State(app): State<Shared>,
255 
    viewer: Viewer,
256 
    Path(id): Path<i64>,
257 
    Form(f): Form<CsrfOnly>,
258 
) -> Result<Page<RunnersPage>, Error> {
259 
    let shell = Shell::new(&app, "/admin/runners", &viewer);
260 
    let me = require_admin(&viewer, &shell)?;
261 
    csrf(&viewer, &f.csrf_token, &shell)?;
262 
    let e = |e| Error::from_db(shell.clone(), e);
263 
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
264 
    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);
265 
    if busy {
266 
        return page(
267 
            &app,
268 
            shell,
269 
            None,
270 
            None,
271 
            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),
272 
        )
273 
        .await;
274 
    }
275 
    runners::remove(&app.db, id).await.map_err(e)?;
276 
    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();
277 
    let also = if r.source == "config" {
278 
        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."
279 
    } else {
280 
        ""
281 
    };
282 
    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await
283 
}
284
285 
#[derive(Template)]
286 
#[template(path = "admin_runner.html")]
287 
pub struct RunnerPage {
288 
    shell: Shell,
289 
    r: runners::Runner,
290 
    running: i64,
291 
    history: Vec<runners::RunnerJob>,
292 
    declines: Vec<crate::db::flow::RunnerDecline>,
293 
    older: Option<i64>,
294 
    jobs: i64,
295 
    passed: i64,
296 
    busy_ms: i64,
297 
}
298
299 
impl RunnerPage {
300 
    /// Time spent running jobs, as pages show durations.
301 
    pub fn busy(&self) -> String {
302 
        crate::flow::human_ms(self.busy_ms)
303 
    }
304 
}
305
306 
#[derive(Deserialize, Default)]
307 
pub struct HistoryQuery {
308 
    before: Option<i64>,
309 
}
310
311 
const PER_PAGE: i64 = 50;
312
313 
/// `GET /admin/runners/{id}`: a runner and the jobs it ran.
314 
pub async fn show(
315 
    State(app): State<Shared>,
316 
    viewer: Viewer,
317 
    Path(id): Path<i64>,
318 
    Query(q): Query<HistoryQuery>,
319 
) -> Result<Page<RunnerPage>, Error> {
320 
    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);
321 
    require_admin(&viewer, &shell)?;
322 
    let e = |e| Error::from_db(shell.clone(), e);
323 
    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;
324 
    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;
325 
    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();
326 
    history.truncate(PER_PAGE as usize);
327 
    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;
328 
    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);
329 
    let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?;
330 
    Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms }))
331 
}
332
333 
pub fn duration(j: &runners::RunnerJob) -> String {
334 
    match (j.started_ms, j.finished_ms) {
335 
        (Some(a), Some(b)) => crate::flow::human_ms(b - a),
336 
        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),
337 
        _ => "–".into(),
338 
    }
339 
}
340
341 
// ---------------------------------------------------------------- settings
342
343 
#[derive(Template)]
344 
#[template(path = "admin_runner_settings.html")]
345 
pub struct SettingsPage {
346 
    shell: Shell,
347 
    default_m: u64,
348 
    max_m: u64,
349 
    grace_m: i64,
350 
    /// What hub.toml says, shown beside a setting the page overrides.
351 
    file_default_m: u64,
352 
    flow_repos: Vec<String>,
353 
    notice: Option<String>,
354 
    error: Option<String>,
355 
}
356
357 
async fn settings_page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Page<SettingsPage> {
358 
    let l = crate::flow::limits(app).await;
359 
    Page(SettingsPage {
360 
        shell,
361 
        default_m: l.default_s / 60,
362 
        max_m: l.max_s / 60,
363 
        grace_m: l.grace_ms / 60_000,
364 
        file_default_m: app.config.flow.job_timeout_minutes,
365 
        flow_repos: app.config.flow.repos.clone(),
366 
        notice,
367 
        error,
368 
    })
369 
}
370
371 
/// `GET /admin/runners/settings` (oxygenhub#89).
372 
pub async fn settings(State(app): State<Shared>, viewer: Viewer) -> Result<Page<SettingsPage>, Error> {
373 
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
374 
    require_admin(&viewer, &shell)?;
375 
    Ok(settings_page(&app, shell, None, None).await)
376 
}
377
378 
#[derive(Deserialize)]
379 
pub struct SettingsForm {
380 
    #[serde(rename = "_csrf")]
381 
    csrf_token: String,
382 
    default_minutes: String,
383 
    max_minutes: String,
384 
    grace_minutes: String,
385 
}
386
387 
/// `POST /admin/runners/settings`: the site's job time limits.
388 
pub async fn save_settings(State(app): State<Shared>, viewer: Viewer, Form(f): Form<SettingsForm>) -> Result<Page<SettingsPage>, Error> {
389 
    let shell = Shell::new(&app, "/admin/runners/settings", &viewer);
390 
    let me = require_admin(&viewer, &shell)?;
391 
    csrf(&viewer, &f.csrf_token, &shell)?;
392 
    let num = |v: &str| v.trim().parse::<u64>().ok().filter(|n| *n > 0);
393 
    let (Some(d), Some(m), Some(g)) = (num(&f.default_minutes), num(&f.max_minutes), num(&f.grace_minutes)) else {
394 
        return Ok(settings_page(&app, shell, None, Some("Each limit is a whole number of minutes, at least 1.".into())).await);
395 
    };
396 
    if d > m {
397 
        return Ok(settings_page(&app, shell, None, Some("The default cannot be longer than the maximum.".into())).await);
398 
    }
399 
    if m > 24 * 60 {
400 
        return Ok(settings_page(&app, shell, None, Some("A job may run for a day at most (1440 minutes).".into())).await);
401 
    }
402 
    let e = |e| Error::from_db(shell.clone(), e);
403 
    accounts::set_setting(&app.db, crate::flow::SETTING_DEFAULT, &d.to_string()).await.map_err(e)?;
404 
    accounts::set_setting(&app.db, crate::flow::SETTING_MAX, &m.to_string()).await.map_err(e)?;
405 
    accounts::set_setting(&app.db, crate::flow::SETTING_GRACE, &g.to_string()).await.map_err(e)?;
406 
    accounts::audit(
407 
        &app.db,
408 
        Some(&me),
409 
        "flow.settings",
410 
        "",
411 
        &format!("job time limit {d} min by default, {m} min at most; lost after {g} min more"),
412 
    )
413 
    .await
414 
    .ok();
415 
    Ok(settings_page(&app, shell, Some("Saved. Jobs claimed from now on run under these limits.".into()), None).await)
416 
}
417
418 
/// Where one image is: the hub, a runner holding it, or a runner fetching
419 
/// it from the swarm now.
420 
pub struct Place {
421 
    pub name: String,
422 
    /// `hub`, `holds` or `fetching`.
423 
    pub kind: &'static str,
424 
    pub title: String,
425 
}
426
427 
pub struct ImageRow {
428 
    pub hash: String,
429 
    pub arch: String,
430 
    /// Its size, when the hub keeps it (runners do not say).
431 
    pub size: Option<i64>,
432 
    pub places: Vec<Place>,
433 
    /// Who sent it to the hub, and when.
434 
    pub sent: Option<(String, i64)>,
435 
    /// Repositories whose jobs ran on it in the last 90 days.
436 
    pub uses: Vec<(String, i64, i64)>,
437 
    pub last_used: i64,
438 
}
439
440 
impl ImageRow {
441 
    /// Nowhere a job could get it: not on the hub, on no runner.
442 
    pub fn stranded(&self) -> bool {
443 
        self.places.is_empty()
444 
    }
445 
}
446
447 
#[derive(Template)]
448 
#[template(path = "admin_images.html")]
449 
pub struct ImagesPage {
450 
    shell: Shell,
451 
    images: Vec<ImageRow>,
452 
    on_hub: usize,
453 
    /// Runners in the image swarm, as "name (site)".
454 
    swarm: Vec<String>,
455 
}
456
457 
/// `GET /admin/runners/images` (C0, docs/cylinder.md): every runner image,
458 
/// where it is and what uses it.
459 
pub async fn images(State(app): State<Shared>, viewer: Viewer) -> Result<Page<ImagesPage>, Error> {
460 
    let shell = Shell::new(&app, "/admin/runners/images", &viewer);
461 
    require_admin(&viewer, &shell)?;
462 
    let e = |e| Error::from_db(shell.clone(), e);
463 
    let kept = crate::images::kept(&app.db).await.map_err(e)?;
464 
    let uses = crate::images::uses(&app.db, db::now_ms() - 90 * 24 * 3600 * 1000).await.map_err(e)?;
465 
    let archs = crate::images::reported_archs(&app.db).await.map_err(e)?;
466 
    let list = runners::list(&app.db).await.map_err(e)?;
467 
    let fetching = app.swarm.fetching_now();
468 
    let mut rows: HashMap<String, ImageRow> = HashMap::new();
469 
    let row = |rows: &mut HashMap<String, ImageRow>, hash: &str| -> () {
470 
        rows.entry(hash.to_ascii_lowercase()).or_insert_with(|| ImageRow {
471 
            hash: hash.to_ascii_lowercase(),
472 
            arch: archs.get(&hash.to_ascii_lowercase()).cloned().unwrap_or_default(),
473 
            size: None,
474 
            places: Vec::new(),
475 
            sent: None,
476 
            uses: Vec::new(),
477 
            last_used: 0,
478 
        });
479 
    };
480 
    for k in &kept {
481 
        row(&mut rows, &k.hash);
482 
        let r = rows.get_mut(&k.hash.to_ascii_lowercase()).expect("added");
483 
        r.size = Some(k.size);
484 
        if r.arch.is_empty() {
485 
            r.arch = k.arch.clone();
486 
        }
487 
        r.sent = Some((k.uploader.clone(), k.created_ms));
488 
        r.places.push(Place { name: "hub".into(), kind: "hub", title: format!("{} chunks kept by the hub", k.chunks) });
489 
    }
490 
    let mut swarm = Vec::new();
491 
    for runner in &list {
492 
        if let Some(a) = app.swarm.addr(runner.id) {
493 
            swarm.push(if a.site.is_empty() { runner.name.clone() } else { format!("{} ({})", runner.name, a.site) });
494 
        }
495 
        for h in runner.image_list().unwrap_or_default() {
496 
            row(&mut rows, &h);
497 
            rows.get_mut(&h.to_ascii_lowercase()).expect("added").places.push(Place {
498 
                name: runner.name.clone(),
499 
                kind: "holds",
500 
                title: format!("{} holds it ({})", runner.name, runner.status()),
501 
            });
502 
        }
503 
    }
504 
    for (image, who) in &fetching {
505 
        row(&mut rows, image);
506 
        for id in who {
507 
            if let Some(runner) = list.iter().find(|r| r.id == *id) {
508 
                rows.get_mut(image).expect("added").places.push(Place {
509 
                    name: runner.name.clone(),
510 
                    kind: "fetching",
511 
                    title: format!("{} is fetching it from the swarm now", runner.name),
512 
                });
513 
            }
514 
        }
515 
    }
516 
    for u in uses {
517 
        if !jet::images::is_hash(&u.image) {
518 
            continue;
519 
        }
520 
        row(&mut rows, &u.image);
521 
        let r = rows.get_mut(&u.image).expect("added");
522 
        r.last_used = r.last_used.max(u.last_ms);
523 
        r.uses.push((format!("{}/{}", u.owner, u.name), u.jobs, u.last_ms));
524 
    }
525 
    let mut images: Vec<ImageRow> = rows.into_values().collect();
526 
    images.sort_by(|a, b| b.last_used.cmp(&a.last_used).then_with(|| a.hash.cmp(&b.hash)));
527 
    let on_hub = kept.len();
528 
    Ok(Page(ImagesPage { shell, images, on_hub, swarm }))
529 
}