A job's finish report says where its time went: image, prepare, run, finish (refs #35) 1215f05ff8d0

## Unreleased

- **The swarm reads images in order (#97).** A fetcher took equally rare chunks at random, so a peer read its whole image in random order, a seek per chunk on a spinning disk: plex-1 got 2.3 GiB from earth-1 at about 2 MB/s. Neighbouring chunks now come in runs, shuffled so runners starting together still spread, and the picker no longer rescans what it has: picking a 142,244-chunk image went from 19 s to 0.35 s.

- **The swarm moves an image nobody pushed (#94).** A runner that serves its peers tells the hub, at start and every ten minutes, which images it holds, and sends the manifest of each the hub keeps none of: the manifest alone, not the image. With a hub that keeps such manifests (oxygenhub#342), runners in the same site fetch the image from it. A fetch told the hub holds no chunks asks peers alone, and fails plainly when none can serve it. Design: docs/jetstream.md §8.1.1.

- **A WASM job stopped while its module compiles stops (#93).** The time limit or a cancel moved the engine's epoch once; a store made after that set its deadline past it, so the job ran on forever (seen as a test spinning 35 minutes under load). Once stopped, the epoch now moves every tick until the job ends.

    pub peak_disk_mib: Option<u64>,

    #[serde(default, skip_serializing_if = "Option::is_none")]

    pub peak_memory_mib: Option<u64>,

}

/// Longest text of one record the hub keeps; longer lines are cut.

        "schema-version 1\nsandbox \"w\" {\n  runtime \"wasm\"\n  network \"loopback\"\n}\njob \"t\" { sandbox \"w\"; exec \"a.wasm\"; }\n";

    assert!(problems(networked).iter().any(|(_, m)| m.contains("has no network")));

}

        error: Some(CANCELLED.into()),

        peak_disk_mib: None,

        peak_memory_mib: None,

    }

}

/// Fetch `claim`'s image from the hub when this home lacks it

/// (oxygenflow#39), saying so in the job's log. A hub that does not hold it

/// leaves it missing, and the preflight declines the job as before.

    let hash = &claim.sandbox.rootfs;

    if hash.is_empty() || home.rootfs(hash).is_file() || !jet::images::is_hash(hash) {

    }

    let note = |text: String| {

        eprintln!("job {}: {text}", claim.job_id);

    static FETCHING: std::sync::Mutex<()> = std::sync::Mutex::new(());

    let _one = FETCHING.lock().unwrap_or_else(|p| p.into_inner());

    if home.rootfs(hash).is_file() {

    }

    let client = match crate::images::Client::new(&cfg.hub, &cfg.token, cfg.ca.as_deref()) {

        Ok(c) => c,

    };

    let fetched = match swarm {

        Some(sw) => {

            note(format!("image {} is not on this runner: fetching it from its peers and the hub", &hash[..12]));

            client.fetch(home, hash)

        }

    };

    match fetched {

        Ok(Some(m)) => note(format!(

            "image {}: fetched {} chunks, {}{} in {:.1} s",

        Ok(None) => note(format!("image {}: the hub does not hold it either", &hash[..12])),

        Err(e) => note(format!("image {}: fetch failed: {e:#}", &hash[..12])),

    }

}

/// Why this runner cannot run `claim` now (spec §11.5), checked before

                    let (home, hub, cfg, used, key, held, swarm) = (home, &hub, &cfg, &used, &key, &held, swarm.as_ref());

                    scope.spawn(move || {

                        let id = claim.job_id;

                        if let Some(why) = preflight(home, &claim) {

                            eprintln!("job {id}: declined: {why}");

                            crate::metrics::DECLINED.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                            claim.heads.first().map(|h| &h[..12]).unwrap_or("-")

                        );

                        crate::metrics::RUNNING.fetch_add(1, std::sync::atomic::Ordering::Relaxed);

                            exit_code: None,

                            duration_ms: 0,

                            receipt: None,

                            error: Some(format!("{e:#}")),

                            peak_disk_mib: None,

                            peak_memory_mib: None,

                        });

                        eprintln!(

                            "job {id}: exit {:?}{}",

    format!("{:.1} s", d.as_secs_f64())

}

    let started = Instant::now();

    let mut sink = LogSink {

        hub,

            );

        }

        let timeout = Duration::from_secs(claim.timeout_s);

        let mut outcome = if wasm_tier {

            crate::wasm::run(&claim.sandbox, &job, &scratch.join("tree"), &scratch, timeout, &mut sink)?

        } else if syscall_tier {

        } else {

            microvm::run(home, &claim.sandbox, &job, &claim.secrets, &scratch.join("workspace.ext4"), &scratch, timeout, &mut sink)?

        };

        if outcome.cancelled {

            // Handed back as the runner leaves: its checkpoints first, for

            // the next attempt (oxygenflow#77).

                Ok(None) => {}

                Err(e) => note(&mut sink, format!("cache: not {}: {e:#}", if kept { "kept" } else { "discarded" })),

            }

        }

        Ok(Finish {

            exit_code: outcome.exit_code,

            receipt,

            signed_receipt,

            error: output_error.or_else(|| outcome.timed_out.then(|| "stopped at the time limit".to_string())),

            error: None,

            peak_disk_mib: None,

            peak_memory_mib: None,

        };

        let quick = Duration::from_millis(10);

        let (base, seen) = fake_hub(vec![502, 503, 500, 500, 204]);