Runner: a heartbeat while it has no room to claim a4b3f8cb9aae
A runner whose slots are full (or that is draining) claims nothing, so the hub heard nothing from it through a quiet job and showed it offline while it worked. It now posts /api/v1/flow/runners/heartbeat every 30 s while it can't claim (once, no retries; a hub without the endpoint answers 404, which is fine). Refs oxygenhub#243
- Author
- Andy McCall @andymccall <[email protected]>
- via
- agent
Claude Code <[email protected]> · delegate for changes, until 2026-11-13 19:00Z · key ea5f78160088
- When
- · 2026-10-16 22:21 UTC
- Signature
- ✔ signed by Claude Code
- Hash
a4b3f8cb9aae8aa50c1f6cc4b8cfc136f7da55dedd90ddf74e9a64b0b5b1c45a- Depends on
- 1e6b9fb1e73f 3007d8f011fa 89171fca079c b11e30f5031d bb643c252515 e0918b887850
- Depended on by
- cedd12342afa 85275e1ef3b7 06a4c472b385 35f7a849bcc5
- In bookmarks
- main
- 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
- 6 atoms, 0 tree operations, 1 file
Browse the tree at this change
modified crates/flow/src/runner.rs +29 −0
| … | 101 unchanged lines |
||
| 102 | 102 | Ok(()) |
|
| 103 | 103 | } |
|
| 104 | 104 | ||
| 105 | + | /// Say this runner is there while it claims nothing: its slots are full, |
|
| 106 | + | /// or it is draining (oxygenhub#243). Once, no retries: the next comes |
|
| 107 | + | /// soon. A hub without the endpoint (404) is not an error. |
|
| 108 | + | pub fn heartbeat(&self) -> Result<()> { |
|
| 109 | + | match self.post_with("/flow/runners/heartbeat", &serde_json::json!({}), &[]) { |
|
| 110 | + | Ok(_) => Ok(()), |
|
| 111 | + | Err(e) if format!("{e:#}").contains("404") => Ok(()), |
|
| 112 | + | Err(e) => Err(e), |
|
| 113 | + | } |
|
| 114 | + | } |
|
| 115 | + | ||
| 105 | 116 | pub fn log(&self, job: i64, records: Vec<Record>) -> Result<()> { |
|
| 106 | 117 | self.post(&format!("/flow/jobs/{job}/log"), &LogBatch { records })?; |
|
| 107 | 118 | Ok(()) |
|
| … | 412 unchanged lines |
||
| 520 | 531 | let mut speed = measure(); |
|
| 521 | 532 | let mut benched = Instant::now(); |
|
| 522 | 533 | let mut said_draining = false; |
|
| 534 | + | // When the hub last heard from us (oxygenhub#243): a claim, or a |
|
| 535 | + | // heartbeat while there is no room to claim. |
|
| 536 | + | let mut heard = Instant::now(); |
|
| 537 | + | let beat = |heard: &mut Instant| { |
|
| 538 | + | if heard.elapsed() >= HUB_HEARTBEAT { |
|
| 539 | + | if let Err(e) = hub.heartbeat() { |
|
| 540 | + | eprintln!("flow runner {name}: heartbeat: {e:#}"); |
|
| 541 | + | } |
|
| 542 | + | *heard = Instant::now(); |
|
| 543 | + | } |
|
| 544 | + | }; |
|
| 523 | 545 | loop { |
|
| 524 | 546 | // Draining (oxygenflow#50): told to stop, claim nothing more and |
|
| 525 | 547 | // stop once every job held is done and reported. |
|
| … | 7 unchanged lines |
||
| 533 | 555 | eprintln!("flow runner {name}: draining: claiming nothing more, stopping when the {jobs} job(s) held are done (stop again to stop now)"); |
|
| 534 | 556 | said_draining = true; |
|
| 535 | 557 | } |
|
| 558 | + | beat(&mut heard); |
|
| 536 | 559 | std::thread::sleep(Duration::from_millis(500)); |
|
| 537 | 560 | continue; |
|
| 538 | 561 | } |
|
| … | 3 unchanged lines |
||
| 542 | 565 | } |
|
| 543 | 566 | let room = budget.room(*used.lock().unwrap()); |
|
| 544 | 567 | let Some((free_memory_mib, free_cpus)) = room else { |
|
| 568 | + | beat(&mut heard); |
|
| 545 | 569 | std::thread::sleep(Duration::from_millis(500)); |
|
| 546 | 570 | continue; |
|
| 547 | 571 | }; |
|
| 572 | + | heard = Instant::now(); |
|
| 548 | 573 | let cache_times = crate::cache::Pool::new(&home.root, cfg.cache_gb).slot_times(); |
|
| 549 | 574 | let req = ClaimRequest { |
|
| 550 | 575 | runner: name.clone(), |
|
| … | 208 unchanged lines |
||
| 759 | 784 | self.last = Instant::now(); |
|
| 760 | 785 | } |
|
| 761 | 786 | } |
|
| 787 | + | ||
| 788 | + | /// How often a runner with no room to claim tells the hub it is there: |
|
| 789 | + | /// well inside the hub's 90 s before it shows a runner offline. |
|
| 790 | + | const HUB_HEARTBEAT: Duration = Duration::from_secs(30); |
|
| 762 | 791 | ||
| 763 | 792 | /// Held while a job's workspace is prepared (see `run_claim`). |
|
| 764 | 793 | static PREPARE: std::sync::Mutex<()> = std::sync::Mutex::new(()); |
|
| … | 658 unchanged lines |