A job's result is kept until the hub has it (refs andymccall/oxygenhub#71) 3007d8f011fa

- **A job's result is kept until the hub has it (oxygenhub#71).** The runner offered a finished job's result three times over seven seconds and then dropped it, so a hub that was briefly down (Earth's reset) left the job running forever. The result is now kept on disk (`pending/` in the Flow home) and offered again, every 5 s doubling to a minute, for up to two hours, through any server error or lost connection; a runner that restarts sends what it kept. A refusal (the hub ended the job already) is final.

    /// Report a job's end until the hub has it (oxygenhub#71): while the hub

    /// cannot be reached or fails (5xx), try again after `first_wait`,

    /// doubling to a minute, for up to `patience`. A refusal (4xx: the job was ended

    /// already, say) is final. A result lost here leaves the job running

    /// on the hub until its time limit.

    pub fn finish_until_accepted(&self, job: i64, f: &Finish, patience: Duration, first_wait: Duration) -> Result<()> {

        let url = format!("{}/api/v1/flow/jobs/{job}/finish", self.base);

        let deadline = std::time::Instant::now() + patience;

        let mut wait = first_wait;

        loop {

            let why = match self.agent.post(&url).set("Authorization", &format!("Bearer {}", self.token)).send_json(f) {

                Ok(_) => return Ok(()),

                Err(ureq::Error::Status(code, r)) if code < 500 => {

                    bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default())

                }

                Err(ureq::Error::Status(code, _)) => format!("HTTP {code}"),

                Err(e) => e.to_string(),

            };

            if std::time::Instant::now() + wait > deadline {

                bail!("{url}: gave up after {} minutes: {why}", patience.as_secs() / 60);

            }

            eprintln!("job {job}: the hub did not take the result ({why}); trying again in {} s", wait.as_secs());

            std::thread::sleep(wait);

            wait = (wait * 2).min(Duration::from_secs(60));

        }

    }

}

/// How long a runner keeps trying to report a job's end.

const FINISH_PATIENCE: Duration = Duration::from_secs(2 * 3600);

/// Where a job's result waits until the hub has it, so a runner that

/// restarts still reports it.

fn pending_path(home: &Home, job: i64) -> PathBuf {

    home.root.join("pending").join(format!("{job}.json"))

}

/// Report `f`, keeping it on disk until the hub has it.

fn report_finish(home: &Home, hub: &Hub, job: i64, f: &Finish) {

    let path = pending_path(home, job);

    let kept = std::fs::create_dir_all(path.parent().unwrap())

        .and_then(|_| std::fs::write(&path, serde_json::to_vec(f).unwrap_or_default()))

        .is_ok();

    match hub.finish_until_accepted(job, f, FINISH_PATIENCE, Duration::from_secs(5)) {

        Ok(()) => {}

        Err(e) => eprintln!("job {job}: reporting failed: {e:#}"),

    }

    if kept {

        let _ = std::fs::remove_file(&path);

}

/// Results a previous run of the runner could not report, to send again.

fn report_pending(home: &Home) -> Vec<(i64, Finish)> {

    let Ok(dir) = std::fs::read_dir(home.root.join("pending")) else {

        return Vec::new();

    };

    dir.filter_map(|e| {

        let p = e.ok()?.path();

        let job: i64 = p.file_stem()?.to_str()?.parse().ok()?;

        let f: Finish = serde_json::from_slice(&std::fs::read(&p).ok()?).ok()?;

        Some((job, f))

    })

    .collect()

    let pending = report_pending(home);

    std::thread::scope(|scope| {

        for (job, f) in pending {

            let hub = &hub;

            scope.spawn(move || {

                eprintln!("job {job}: reporting the result kept from before the restart");

                report_finish(home, hub, job, &f);

            });

        }

        loop {

            let room = budget.room(*used.lock().unwrap());

            let Some((free_memory_mib, free_cpus)) = room else {

                std::thread::sleep(Duration::from_millis(500));

                continue;

            };

            let req = ClaimRequest {

                runner: name.clone(),

                runtimes: vec!["microvm".into()],

                free_memory_mib,

                free_cpus,

                info: Some(RunnerInfo {

                    version: env!("OXY_VERSION_LONG").into(),

                    ..RunnerInfo::here(cfg.labels.clone(), budget.max_jobs as u32, cfg.memory_mib, cfg.cpus)

                }),

            };

            match hub.claim(&req) {

                Ok(Some(claim)) => {

                    idle = Duration::from_secs(2);

                    let (mem, cpus) = (claim.sandbox.memory_mib, claim.sandbox.cpus);

                    {

                        let mut u = used.lock().unwrap();

                        u.jobs += 1;

                        u.memory_mib += mem;

                        u.cpus += cpus;

                    let (home, hub, cfg, used, key) = (home, &hub, &cfg, &used, &key);

                    scope.spawn(move || {

                        let id = claim.job_id;

                        eprintln!(

                            "job {id}: {} {} on {}",

                            claim.repo,

                            claim.job.name,

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

                        );

                        let finish = run_claim(home, hub, cfg, key, &claim).unwrap_or_else(|e| Finish {

                            exit_code: None,

                            duration_ms: 0,

                            receipt: None,

                            signed_receipt: None,

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

                        });

                        eprintln!(

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

                            finish.exit_code,

                            finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default()

                        );

                        report_finish(home, hub, id, &finish);

                        let mut u = used.lock().unwrap();

                        u.jobs -= 1;

                        u.memory_mib -= mem;

                        u.cpus -= cpus;

                    });

                }

                Ok(None) => std::thread::sleep(idle),

                Err(e) => {

                    eprintln!("claim: {e:#}");

                    std::thread::sleep(idle);

                    idle = (idle * 2).min(Duration::from_secs(60));

                }

    }

    /// oxygenhub#71: a job's result is offered until the hub takes it,

    /// through server errors, and a refusal ends the trying at once.

    #[test]

    fn a_result_is_reported_until_the_hub_takes_it() {

        let f = Finish { exit_code: Some(0), duration_ms: 1, receipt: None, signed_receipt: None, error: None };

        let quick = Duration::from_millis(10);

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

        hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).unwrap();

        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 5, "kept trying through four server errors");

        let (base, seen) = fake_hub(vec![409]);

        assert!(hub_at(base).finish_until_accepted(7, &f, Duration::from_secs(30), quick).is_err());

        assert_eq!(seen.load(std::sync::atomic::Ordering::SeqCst), 1, "a refusal is final");