A runner tells the hub which jobs it holds at each claim (refs oxygenhub#71) 1daca0b6bb8c
ClaimRequest.holding lists the jobs running or with a result not yet reported; the hub ends any other job it has as running on that runner.
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-10-02 03:02 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
1daca0b6bb8c1eb3e887aac0a920eb826064cb643ec01d140663692be7a01ee0- Depends on
- 3007d8f011fa 8c0a4e3a8a30 bb643c252515 c3db750ed96e e0918b887850 f98a8f7881e0
- Depended on by
- b14aeaae683a 62829d15ef09 4cf0c99a1486 1abadd4a2fa6 280ef951d193 02b127d2c721
- In bookmarks
- main newer-pipeline push-uses-login
- In tags
- v0.2.0 v0.3.0 v0.3.1 v0.4.0 v0.5.0 v0.5.1 v0.5.2 v0.5.3
- Edits
- 9 atoms, 0 tree operations, 3 files
Browse the tree at this change
modified CHANGELOG.md +1 −0
| … | 3 unchanged lines |
||
| 4 | 4 | ||
| 5 | 5 | ## Unreleased |
|
| 6 | 6 | ||
| 7 | + | - **A runner says which jobs it holds (oxygenhub#71).** Each claim lists the jobs the runner is running or still has a result for, so the hub ends at once any job it lost by restarting, instead of an hour later at the time limit. |
|
| 7 | 8 | - **A runner says which images it holds (oxygenhub#73).** Each claim lists the rootfs images in the runner's `cas/` (read each time, so an image imported while it runs counts), and the hub offers only jobs whose image is among them. plex-1 had failed metalgear's jobs for want of one. |
|
| 8 | 9 | ||
| 9 | 10 | - **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. |
|
| … | 3 unchanged lines |
modified crates/flow-core/src/api.rs +6 −0
| … | 27 unchanged lines |
||
| 28 | 28 | /// offered. |
|
| 29 | 29 | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 30 | 30 | pub images: Option<Vec<String>>, |
|
| 31 | + | /// The jobs this runner holds now: running, or finished with a result |
|
| 32 | + | /// not yet reported. The hub ends any other job it has as running on |
|
| 33 | + | /// this runner, which the runner lost by restarting or never received |
|
| 34 | + | /// (oxygenhub#71). Absent: the runner does not say. |
|
| 35 | + | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 36 | + | pub holding: Option<Vec<i64>>, |
|
| 31 | 37 | } |
|
| 32 | 38 | ||
| 33 | 39 | /// A runner's description: shown on the hub's Runners page, and what |
|
| … | 158 unchanged lines |
modified crates/flow/src/runner.rs +11 −2
| … | 263 unchanged lines |
||
| 264 | 264 | let used = std::sync::Mutex::new(Used::default()); |
|
| 265 | 265 | let mut idle = Duration::from_secs(2); |
|
| 266 | 266 | let pending = report_pending(home); |
|
| 267 | + | // The jobs held now, told to the hub at each claim so it can end the |
|
| 268 | + | // ones this runner lost: a job is held from its claim until its result |
|
| 269 | + | // is reported. |
|
| 270 | + | let held = std::sync::Mutex::new(std::collections::BTreeSet::<i64>::new()); |
|
| 271 | + | held.lock().unwrap().extend(pending.iter().map(|(job, _)| *job)); |
|
| 267 | 272 | std::thread::scope(|scope| { |
|
| 268 | 273 | for (job, f) in pending { |
|
| 269 | - | let hub = &hub; |
|
| 274 | + | let (hub, held) = (&hub, &held); |
|
| 270 | 275 | scope.spawn(move || { |
|
| 271 | 276 | eprintln!("job {job}: reporting the result kept from before the restart"); |
|
| 272 | 277 | report_finish(home, hub, job, &f); |
|
| 278 | + | held.lock().unwrap().remove(&job); |
|
| 273 | 279 | }); |
|
| 274 | 280 | } |
|
| 275 | 281 | loop { |
|
| … | 13 unchanged lines |
||
| 289 | 295 | }), |
|
| 290 | 296 | // Read each time: an image imported while the runner runs counts. |
|
| 291 | 297 | images: Some(home.images()), |
|
| 298 | + | holding: Some(held.lock().unwrap().iter().copied().collect()), |
|
| 292 | 299 | }; |
|
| 293 | 300 | match hub.claim(&req) { |
|
| 294 | 301 | Ok(Some(claim)) => { |
|
| … | 5 unchanged lines |
||
| 300 | 307 | u.memory_mib += mem; |
|
| 301 | 308 | u.cpus += cpus; |
|
| 302 | 309 | } |
|
| 303 | - | let (home, hub, cfg, used, key) = (home, &hub, &cfg, &used, &key); |
|
| 310 | + | held.lock().unwrap().insert(claim.job_id); |
|
| 311 | + | let (home, hub, cfg, used, key, held) = (home, &hub, &cfg, &used, &key, &held); |
|
| 304 | 312 | scope.spawn(move || { |
|
| 305 | 313 | let id = claim.job_id; |
|
| 306 | 314 | eprintln!( |
|
| … | 15 unchanged lines |
||
| 322 | 330 | finish.error.as_deref().map(|e| format!(" ({e})")).unwrap_or_default() |
|
| 323 | 331 | ); |
|
| 324 | 332 | report_finish(home, hub, id, &finish); |
|
| 333 | + | held.lock().unwrap().remove(&id); |
|
| 325 | 334 | let mut u = used.lock().unwrap(); |
|
| 326 | 335 | u.jobs -= 1; |
|
| 327 | 336 | u.memory_mib -= mem; |
|
| … | 377 unchanged lines |