Claims say when each build cache was last kept 1e6b9fb1e73f
When both runners hold a job's build cache, the hub should send the job to the fresher copy (#54): a later disk built later sources, so less is rebuilt. Claims now carry cache_times beside caches: each slot's newest disk's time, set when a passing job keeps it (Unix ms). An older hub ignores the field. refs #54
- Author
- Andy McCall @andymccall <[email protected]>
- When
- · 2026-10-14 01:37 UTC
- Signature
- ✔ signed by Andy McCall
- Hash
1e6b9fb1e73fdbe8e22c781af20f6da643afffddf74404f399eb9f925d85b04f- Depends on
- 3007d8f011fa 82cf6e91d398 a083cc802892 b14aeaae683a d60aac6c3f8a e0918b887850
- Depended on by
- 2594278f49b2 a4b3f8cb9aae
- 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
- 10 atoms, 0 tree operations, 4 files
Browse the tree at this change
modified crates/flow-core/src/api.rs +5 −0
| … | 56 unchanged lines |
||
| 57 | 57 | /// where its cache is. Absent: an older runner. |
|
| 58 | 58 | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 59 | 59 | pub caches: Option<Vec<String>>, |
|
| 60 | + | /// When each of [`ClaimRequest::caches`] last kept a disk, Unix ms, so |
|
| 61 | + | /// the hub can send a job to the runner whose copy is freshest when |
|
| 62 | + | /// several hold one (oxygenflow#54). Absent: an older runner. |
|
| 63 | + | #[serde(default, skip_serializing_if = "Option::is_none")] |
|
| 64 | + | pub cache_times: Option<std::collections::BTreeMap<String, u64>>, |
|
| 60 | 65 | } |
|
| 61 | 66 | ||
| 62 | 67 | /// A name safe as one path segment: kept readable, with a short hash so |
|
| … | 296 unchanged lines |
modified crates/flow/src/cache.rs +24 −9
| … | 163 unchanged lines |
||
| 164 | 164 | ||
| 165 | 165 | /// The repository and job slots holding at least one disk, as |
|
| 166 | 166 | /// [`flow_core::api::cache_slot`] names them, for the claim |
|
| 167 | - | /// (oxygenflow#51). |
|
| 168 | - | pub fn slots(&self) -> Vec<String> { |
|
| 169 | - | let mut out = Vec::new(); |
|
| 167 | + | /// (oxygenflow#51), each with when its newest disk was kept (Unix ms): |
|
| 168 | + | /// a disk's time is set when it is kept, so the newest is the freshest |
|
| 169 | + | /// build (oxygenflow#54). |
|
| 170 | + | pub fn slot_times(&self) -> std::collections::BTreeMap<String, u64> { |
|
| 171 | + | let mut out = std::collections::BTreeMap::new(); |
|
| 170 | 172 | for repo in std::fs::read_dir(&self.root).into_iter().flatten().flatten() { |
|
| 171 | 173 | for job in std::fs::read_dir(repo.path()).into_iter().flatten().flatten() { |
|
| 172 | - | let mut any = false; |
|
| 173 | - | walk(&job.path(), &mut |p| any |= p.extension().is_some_and(|e| e == "ext4")); |
|
| 174 | - | if any { |
|
| 175 | - | out.push(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy())); |
|
| 174 | + | let mut newest: Option<SystemTime> = None; |
|
| 175 | + | walk(&job.path(), &mut |p| { |
|
| 176 | + | if p.extension().is_some_and(|e| e == "ext4") { |
|
| 177 | + | let t = modified(p); |
|
| 178 | + | newest = Some(newest.map_or(t, |n| n.max(t))); |
|
| 179 | + | } |
|
| 180 | + | }); |
|
| 181 | + | if let Some(t) = newest { |
|
| 182 | + | let ms = t.duration_since(SystemTime::UNIX_EPOCH).map(|d| d.as_millis() as u64).unwrap_or(0); |
|
| 183 | + | out.insert(format!("{}/{}", repo.file_name().to_string_lossy(), job.file_name().to_string_lossy()), ms); |
|
| 176 | 184 | } |
|
| 177 | 185 | } |
|
| 178 | 186 | } |
|
| 179 | - | out.sort(); |
|
| 180 | 187 | out |
|
| 181 | 188 | } |
|
| 182 | 189 | ||
| … | 112 unchanged lines |
||
| 295 | 302 | assert!(taken.how.contains("older key"), "{}", taken.how); |
|
| 296 | 303 | assert_eq!(taken.sources.as_ref(), Some(&built)); |
|
| 297 | 304 | pool.keep(&main, &m, &key2, &work.path().join("c.ext4"), &built).unwrap(); |
|
| 298 | - | assert_eq!(pool.slots(), vec![flow_core::api::cache_slot("andy/hub", "test")], "what the claim reports (#51)"); |
|
| 305 | + | assert_eq!( |
|
| 306 | + | pool.slot_times().into_keys().collect::<Vec<_>>(), |
|
| 307 | + | vec![flow_core::api::cache_slot("andy/hub", "test")], |
|
| 308 | + | "what the claim reports (#51)" |
|
| 309 | + | ); |
|
| 310 | + | // With when it was last kept (#54): just now. |
|
| 311 | + | let now = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis() as u64; |
|
| 312 | + | let kept = pool.slot_times()[&flow_core::api::cache_slot("andy/hub", "test")]; |
|
| 313 | + | assert!(now.abs_diff(kept) < 60_000, "kept {kept}, now {now}"); |
|
| 299 | 314 | // Another bookmark copies it, leaving main's in place. |
|
| 300 | 315 | let feature = Ctx { bookmark: "feature", ..main }; |
|
| 301 | 316 | let taken = pool.take(&feature, &m, &key2, &work.path().join("d.ext4")).unwrap(); |
|
| … | 45 unchanged lines |
modified crates/flow/src/runner.rs +3 −1
| … | 544 unchanged lines |
||
| 545 | 545 | std::thread::sleep(Duration::from_millis(500)); |
|
| 546 | 546 | continue; |
|
| 547 | 547 | }; |
|
| 548 | + | let cache_times = crate::cache::Pool::new(&home.root, cfg.cache_gb).slot_times(); |
|
| 548 | 549 | let req = ClaimRequest { |
|
| 549 | 550 | runner: name.clone(), |
|
| 550 | 551 | runtimes: runtimes.clone(), |
|
| … | 10 unchanged lines |
||
| 561 | 562 | fetches_images: true, |
|
| 562 | 563 | facts: Some(crate::facts::now(home, Some(speed))), |
|
| 563 | 564 | swarm: swarm.as_ref().map(|s| s.addr.clone()), |
|
| 564 | - | caches: Some(crate::cache::Pool::new(&home.root, cfg.cache_gb).slots()), |
|
| 565 | + | caches: Some(cache_times.keys().cloned().collect()), |
|
| 566 | + | cache_times: Some(cache_times), |
|
| 565 | 567 | }; |
|
| 566 | 568 | match hub.claim(&req) { |
|
| 567 | 569 | Ok(Some(claim)) => { |
|
| … | 853 unchanged lines |
modified docs/spec.md +5 −1
| … | 315 unchanged lines |
||
| 316 | 316 | `flow_core::api::cache_slot(repo, job)`), and the hub ranks a runner |
|
| 317 | 317 | holding a job's cache above one holding its image (§11.4): a job another |
|
| 318 | 318 | free runner holds the cache for is held for it for up to 15 seconds, and |
|
| 319 | - | `placed` says "holds its build cache". |
|
| 319 | + | `placed` says "holds its build cache". When several hold it, the |
|
| 320 | + | freshest copy wins (oxygenflow#54): claims also say when each slot last |
|
| 321 | + | kept a disk (`cache_times`, Unix ms), and a copy kept at least a minute |
|
| 322 | + | later ranks above an older one, before the image and speed are |
|
| 323 | + | weighed; `placed` says how old the copy is. |
|
| 320 | 324 | - The syscall tier keeps its disks empty for now. |
|
| 321 | 325 | - A cache is a speed-up, never a source of truth: jobs must build |
|
| 322 | 326 | correctly from an empty disk, which a key change or an eviction gives |
|
| … | 318 unchanged lines |