| 1 | //! Recovery of incomplete/cancelled/crashed production runs (design §7/§12.6). |
| 2 | //! |
| 3 | //! Every production run directory holds an exclusive `run.lock` while a |
| 4 | //! worker writes it and an append-only `progress.jsonl` journal with one |
| 5 | //! `{"path","sha256"}` line per published target PNG (fsynced after the PNG |
| 6 | //! itself was published). A run may be resumed only when its `render.json` |
| 7 | //! is byte-for-byte the same contract document that the current job would |
| 8 | //! write (project, sex/set, body/equipment SHA-256, target, FPS, framing, |
| 9 | //! pivot, camera, clip inventory, maps, counts). Before any rendering every |
| 10 | //! planned PNG is checked: journaled files whose bytes match their digest are |
| 11 | //! kept untouched; missing ones are regenerated; corrupt files and foreign |
| 12 | //! links are moved aside (`<name>.corrupt-aNNN-K`, never deleted); existing |
| 13 | //! files without a journal entry (crash between publish and journal) are |
| 14 | //! compared with the regenerated bytes and kept when identical. |
| 15 | use crate::digest; |
| 16 | use serde_json::Value; |
| 17 | use std::{ |
| 18 | collections::{BTreeMap, BTreeSet, HashSet}, |
| 19 | fs::{self, File, OpenOptions}, |
| 20 | io::{Read, Write}, |
| 21 | path::{Path, PathBuf}, |
| 22 | }; |
| 23 | |
| 24 | pub const JOURNAL: &str = "progress.jsonl"; |
| 25 | pub const LOCK: &str = "run.lock"; |
| 26 | pub const RESUME_LOG: &str = "resume.jsonl"; |
| 27 | /// Prefix of the temporary build folder of a render (also the resume target). |
| 28 | pub const IN_PROGRESS_PREFIX: &str = "inprogress-"; |
| 29 | /// Prefix of a previous result folder moved aside during the final swap. |
| 30 | pub const REPLACED_PREFIX: &str = ".replaced-"; |
| 31 | const MAX_RENDER_JSON: u64 = 1024 * 1024; |
| 32 | /// Subfolder of a result folder with every loose per-frame map folder (Tim t993u). |
| 33 | pub const RAW_DIR: &str = "raw files"; |
| 34 | /// Completion marker of a build; lives only in the in-progress folder. |
| 35 | pub const COMPLETE_MARKER: &str = "complete.json"; |
| 36 | |
| 37 | fn has_ppda(dir: &Path) -> bool { |
| 38 | fs::read_dir(dir).is_ok_and(|read| { |
| 39 | read.flatten() |
| 40 | .any(|e| e.file_name().to_string_lossy().ends_with("-atlas.ppda")) |
| 41 | }) |
| 42 | } |
| 43 | |
| 44 | /// True when `dir` holds a completed build: its `complete.json` marker, or a |
| 45 | /// build already stripped of bookkeeping (no `render.json`, a `.ppda` present). |
| 46 | pub fn is_complete(dir: &Path) -> bool { |
| 47 | fs::symlink_metadata(dir.join(COMPLETE_MARKER)).is_ok() |
| 48 | || (fs::symlink_metadata(dir.join("render.json")).is_err() && has_ppda(dir)) |
| 49 | } |
| 50 | |
| 51 | /// Remove render/resume/lock bookkeeping from a completed, unlocked build so |
| 52 | /// the result folder holds only consumer output (`.ppda`, atlases, |
| 53 | /// `raw files/`). The `.ppda` carries everything the reader still needs. |
| 54 | /// Idempotent and crash-safe: `complete.json` is removed last. |
| 55 | pub fn strip_bookkeeping(root: &Path) -> Result<(), String> { |
| 56 | if !is_complete(root) { |
| 57 | return Err("opschonen geweigerd: render is niet voltooid".into()); |
| 58 | } |
| 59 | if is_locked(root) { |
| 60 | return Err("opschonen geweigerd: run.lock is bezet".into()); |
| 61 | } |
| 62 | for entry in fs::read_dir(root).map_err(io)? { |
| 63 | let entry = entry.map_err(io)?; |
| 64 | let name = entry.file_name().to_string_lossy().to_string(); |
| 65 | let bookkeeping = name == "render.json" |
| 66 | || name == JOURNAL |
| 67 | || name == RESUME_LOG |
| 68 | || name == LOCK |
| 69 | || (name.starts_with("attempt-") && name.ends_with(".json")); |
| 70 | if bookkeeping && entry.file_type().map_err(io)?.is_file() { |
| 71 | fs::remove_file(entry.path()).map_err(|e| format!("opschonen {name}: {e}"))?; |
| 72 | } |
| 73 | } |
| 74 | sync_dir(root); |
| 75 | match fs::remove_file(root.join(COMPLETE_MARKER)) { |
| 76 | Ok(()) => {} |
| 77 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} |
| 78 | Err(e) => return Err(format!("opschonen complete.json: {e}")), |
| 79 | } |
| 80 | sync_dir(root); |
| 81 | Ok(()) |
| 82 | } |
| 83 | const MAX_JOURNAL: u64 = 256 * 1024 * 1024; |
| 84 | const MAX_CANDIDATES: usize = 20; |
| 85 | const MAX_SCANNED_RUNS: usize = 2000; |
| 86 | |
| 87 | /// Outcome of the pre-render verification of a resumed run. |
| 88 | #[derive(Debug, Clone, Default, PartialEq, Eq)] |
| 89 | pub struct ResumeReport { |
| 90 | pub attempt: u32, |
| 91 | pub planned: usize, |
| 92 | pub verified: usize, |
| 93 | pub missing: usize, |
| 94 | pub quarantined: Vec<String>, |
| 95 | pub unjournaled: usize, |
| 96 | pub damaged_journal_lines: usize, |
| 97 | pub previous_marker: Option<String>, |
| 98 | } |
| 99 | |
| 100 | /// Open write session on one run directory (fresh or resumed). |
| 101 | pub struct RunSession { |
| 102 | pub root: PathBuf, |
| 103 | pub attempt: u32, |
| 104 | /// Planned PNGs already on disk with a verified journal checksum. |
| 105 | pub verified: BTreeMap<String, String>, |
| 106 | /// Existing regular PNGs without a journal entry; compared on regeneration. |
| 107 | pub unjournaled: BTreeSet<String>, |
| 108 | pub report: ResumeReport, |
| 109 | journal: File, |
| 110 | _lock: File, |
| 111 | } |
| 112 | |
| 113 | fn io(e: std::io::Error) -> String { |
| 114 | e.to_string() |
| 115 | } |
| 116 | |
| 117 | fn lock_run(root: &Path) -> Result<File, String> { |
| 118 | let path = root.join(LOCK); |
| 119 | if let Ok(meta) = fs::symlink_metadata(&path) { |
| 120 | if !meta.is_file() { |
| 121 | return Err("run.lock is geen regulier bestand".into()); |
| 122 | } |
| 123 | } |
| 124 | let file = OpenOptions::new() |
| 125 | .create(true) |
| 126 | .truncate(false) |
| 127 | .write(true) |
| 128 | .open(&path) |
| 129 | .map_err(io)?; |
| 130 | match file.try_lock() { |
| 131 | Ok(()) => Ok(file), |
| 132 | Err(fs::TryLockError::WouldBlock) => { |
| 133 | Err("deze run wordt op dit moment al gerenderd (run.lock bezet)".into()) |
| 134 | } |
| 135 | Err(fs::TryLockError::Error(e)) => Err(format!("run.lock: {e}")), |
| 136 | } |
| 137 | } |
| 138 | |
| 139 | fn is_locked(root: &Path) -> bool { |
| 140 | let Ok(file) = OpenOptions::new().write(true).open(root.join(LOCK)) else { |
| 141 | return false; |
| 142 | }; |
| 143 | matches!(file.try_lock(), Err(fs::TryLockError::WouldBlock)) |
| 144 | } |
| 145 | |
| 146 | fn open_journal(root: &Path) -> Result<File, String> { |
| 147 | let path = root.join(JOURNAL); |
| 148 | if let Ok(meta) = fs::symlink_metadata(&path) { |
| 149 | if !meta.is_file() { |
| 150 | return Err("progress.jsonl is geen regulier bestand".into()); |
| 151 | } |
| 152 | } |
| 153 | OpenOptions::new() |
| 154 | .create(true) |
| 155 | .append(true) |
| 156 | .open(path) |
| 157 | .map_err(io) |
| 158 | } |
| 159 | |
| 160 | fn read_regular(path: &Path, max: u64) -> Result<Option<Vec<u8>>, String> { |
| 161 | let meta = match fs::symlink_metadata(path) { |
| 162 | Ok(meta) => meta, |
| 163 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None), |
| 164 | Err(e) => return Err(e.to_string()), |
| 165 | }; |
| 166 | if !meta.is_file() { |
| 167 | return Err(format!("{} is geen regulier bestand", path.display())); |
| 168 | } |
| 169 | if meta.len() > max { |
| 170 | return Err(format!("{} groter dan {max} bytes", path.display())); |
| 171 | } |
| 172 | let mut bytes = Vec::new(); |
| 173 | File::open(path) |
| 174 | .map_err(io)? |
| 175 | .take(max + 1) |
| 176 | .read_to_end(&mut bytes) |
| 177 | .map_err(io)?; |
| 178 | if bytes.len() as u64 > max { |
| 179 | return Err(format!("{} groter dan {max} bytes", path.display())); |
| 180 | } |
| 181 | Ok(Some(bytes)) |
| 182 | } |
| 183 | |
| 184 | /// Journal entries: last valid entry per path wins; returns (entries, damaged lines, ends with newline). |
| 185 | pub fn read_journal(root: &Path) -> Result<(BTreeMap<String, String>, usize, bool), String> { |
| 186 | let Some(bytes) = read_regular(&root.join(JOURNAL), MAX_JOURNAL)? else { |
| 187 | return Ok((BTreeMap::new(), 0, true)); |
| 188 | }; |
| 189 | let mut entries = BTreeMap::new(); |
| 190 | let mut damaged = 0; |
| 191 | for line in bytes.split(|b| *b == b'\n') { |
| 192 | if line.is_empty() { |
| 193 | continue; |
| 194 | } |
| 195 | let parsed = serde_json::from_slice::<Value>(line).ok().and_then(|v| { |
| 196 | let path = v.get("path")?.as_str()?.to_string(); |
| 197 | let sha = v.get("sha256")?.as_str()?.to_string(); |
| 198 | digest::is_sha256_hex(&sha).then_some((path, sha)) |
| 199 | }); |
| 200 | match parsed { |
| 201 | Some((path, sha)) => { |
| 202 | entries.insert(path, sha); |
| 203 | } |
| 204 | None => damaged += 1, |
| 205 | } |
| 206 | } |
| 207 | Ok((entries, damaged, bytes.last().is_none_or(|b| *b == b'\n'))) |
| 208 | } |
| 209 | |
| 210 | /// Move a bad file/link aside inside the run without deleting or clobbering anything. |
| 211 | pub(crate) fn quarantine(root: &Path, rel: &str, attempt: u32) -> Result<String, String> { |
| 212 | let path = root.join(rel); |
| 213 | let name = path |
| 214 | .file_name() |
| 215 | .and_then(|n| n.to_str()) |
| 216 | .ok_or("ongeldige bestandsnaam")? |
| 217 | .to_string(); |
| 218 | for n in 0..1000 { |
| 219 | let target = path.with_file_name(format!("{name}.corrupt-a{attempt:03}-{n}")); |
| 220 | match fs::hard_link(&path, &target) { |
| 221 | Ok(()) => { |
| 222 | fs::remove_file(&path).map_err(io)?; |
| 223 | let rel_target = Path::new(rel) |
| 224 | .with_file_name(target.file_name().unwrap_or_default()) |
| 225 | .to_string_lossy() |
| 226 | .replace('\\', "/"); |
| 227 | return Ok(rel_target); |
| 228 | } |
| 229 | Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue, |
| 230 | Err(e) => return Err(format!("quarantaine van {rel} mislukt: {e}")), |
| 231 | } |
| 232 | } |
| 233 | Err(format!("geen vrije quarantainenaam voor {rel}")) |
| 234 | } |
| 235 | |
| 236 | /// Move a file to a fresh unique name (no clobber), returning the new file name. |
| 237 | fn archive(root: &Path, name: &str, prefix: &str) -> Result<String, String> { |
| 238 | for n in 0..1000 { |
| 239 | let target = if n == 0 { |
| 240 | format!("{prefix}-{name}") |
| 241 | } else { |
| 242 | format!("{prefix}-{n}-{name}") |
| 243 | }; |
| 244 | match fs::hard_link(root.join(name), root.join(&target)) { |
| 245 | Ok(()) => { |
| 246 | fs::remove_file(root.join(name)).map_err(io)?; |
| 247 | return Ok(target); |
| 248 | } |
| 249 | Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue, |
| 250 | Err(e) => return Err(format!("archiveren van {name} mislukt: {e}")), |
| 251 | } |
| 252 | } |
| 253 | Err(format!("geen vrije archiefnaam voor {name}")) |
| 254 | } |
| 255 | |
| 256 | /// Top-level keys whose values differ between two JSON objects. |
| 257 | pub fn contract_differences(existing: &Value, expected: &Value) -> Vec<String> { |
| 258 | match (existing.as_object(), expected.as_object()) { |
| 259 | (Some(a), Some(b)) => a |
| 260 | .keys() |
| 261 | .chain(b.keys()) |
| 262 | .collect::<BTreeSet<_>>() |
| 263 | .into_iter() |
| 264 | .filter(|k| a.get(*k) != b.get(*k)) |
| 265 | .cloned() |
| 266 | .collect(), |
| 267 | _ => vec!["render.json".into()], |
| 268 | } |
| 269 | } |
| 270 | |
| 271 | fn normalized(doc: &Value) -> Result<Value, String> { |
| 272 | serde_json::from_str(&serde_json::to_string(doc).map_err(|e| e.to_string())?) |
| 273 | .map_err(|e| e.to_string()) |
| 274 | } |
| 275 | |
| 276 | fn read_render(root: &Path) -> Result<Value, String> { |
| 277 | let bytes = |
| 278 | read_regular(&root.join("render.json"), MAX_RENDER_JSON)?.ok_or("render.json ontbreekt")?; |
| 279 | serde_json::from_slice(&bytes).map_err(|e| format!("render.json: {e}")) |
| 280 | } |
| 281 | |
| 282 | fn check_contract(root: &Path, render_doc: &Value) -> Result<(), String> { |
| 283 | let existing = read_render(root)?; |
| 284 | let expected = normalized(render_doc)?; |
| 285 | if existing != expected { |
| 286 | return Err(format!( |
| 287 | "huidige instellingen/bronnen passen niet bij deze run (verschil in: {})", |
| 288 | contract_differences(&existing, &expected).join(", ") |
| 289 | )); |
| 290 | } |
| 291 | Ok(()) |
| 292 | } |
| 293 | |
| 294 | /// Direct `inprogress-*` child directory of the output directory, not a link. |
| 295 | fn run_root(output: &Path, dir: &Path) -> Result<PathBuf, String> { |
| 296 | let meta = fs::symlink_metadata(dir).map_err(io)?; |
| 297 | if !meta.is_dir() { |
| 298 | return Err("runmap is geen gewone map".into()); |
| 299 | } |
| 300 | let out = output.canonicalize().map_err(io)?; |
| 301 | let root = dir.canonicalize().map_err(io)?; |
| 302 | let name_ok = root |
| 303 | .file_name() |
| 304 | .and_then(|n| n.to_str()) |
| 305 | .is_some_and(|n| n.starts_with(IN_PROGRESS_PREFIX)); |
| 306 | if root.parent() != Some(out.as_path()) || !name_ok { |
| 307 | return Err("runmap ligt niet direct in de uitvoermap".into()); |
| 308 | } |
| 309 | Ok(root) |
| 310 | } |
| 311 | |
| 312 | impl RunSession { |
| 313 | /// A new, empty run directory created by the caller. |
| 314 | pub fn fresh(root: &Path) -> Result<Self, String> { |
| 315 | let lock = lock_run(root)?; |
| 316 | let journal = open_journal(root)?; |
| 317 | Ok(Self { |
| 318 | root: root.to_path_buf(), |
| 319 | attempt: 0, |
| 320 | verified: BTreeMap::new(), |
| 321 | unjournaled: BTreeSet::new(), |
| 322 | report: ResumeReport::default(), |
| 323 | journal, |
| 324 | _lock: lock, |
| 325 | }) |
| 326 | } |
| 327 | |
| 328 | /// Lock, contract-check and verify an incomplete run before re-rendering. |
| 329 | /// `planned` lists every relative target PNG of the job in render order. |
| 330 | pub fn resume( |
| 331 | output: &Path, |
| 332 | dir: &Path, |
| 333 | render_doc: &Value, |
| 334 | planned: &[String], |
| 335 | progress: &mut dyn FnMut(usize, usize), |
| 336 | ) -> Result<Self, String> { |
| 337 | let root = run_root(output, dir)?; |
| 338 | if is_complete(&root) { |
| 339 | return Err("deze run is al voltooid (complete.json bestaat)".into()); |
| 340 | } |
| 341 | let lock = lock_run(&root)?; |
| 342 | check_contract(&root, render_doc)?; |
| 343 | let planned_set: HashSet<&str> = planned.iter().map(String::as_str).collect(); |
| 344 | let (journal_entries, damaged, newline) = read_journal(&root)?; |
| 345 | if let Some(unknown) = journal_entries |
| 346 | .keys() |
| 347 | .find(|k| !planned_set.contains(k.as_str())) |
| 348 | { |
| 349 | return Err(format!("journal noemt een onbekend pad: {unknown}")); |
| 350 | } |
| 351 | let previous_attempts = read_regular(&root.join(RESUME_LOG), MAX_RENDER_JSON)? |
| 352 | .map(|b| b.split(|c| *c == b'\n').filter(|l| !l.is_empty()).count()) |
| 353 | .unwrap_or(0); |
| 354 | let attempt = previous_attempts as u32 + 1; |
| 355 | let mut report = ResumeReport { |
| 356 | attempt, |
| 357 | planned: planned.len(), |
| 358 | damaged_journal_lines: damaged, |
| 359 | ..Default::default() |
| 360 | }; |
| 361 | let mut verified = BTreeMap::new(); |
| 362 | let mut unjournaled = BTreeSet::new(); |
| 363 | for (i, rel) in planned.iter().enumerate() { |
| 364 | if i % 256 == 0 { |
| 365 | progress(i, planned.len()); |
| 366 | } |
| 367 | let path = root.join(rel); |
| 368 | let meta = match fs::symlink_metadata(&path) { |
| 369 | Ok(meta) => meta, |
| 370 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => { |
| 371 | report.missing += 1; |
| 372 | continue; |
| 373 | } |
| 374 | Err(e) => return Err(e.to_string()), |
| 375 | }; |
| 376 | if meta.is_dir() { |
| 377 | return Err(format!("map staat op de plaats van PNG {rel}")); |
| 378 | } |
| 379 | if !meta.is_file() || meta.len() > crate::export_set::MAX_PNG_BYTES { |
| 380 | report.quarantined.push(quarantine(&root, rel, attempt)?); |
| 381 | continue; |
| 382 | } |
| 383 | match journal_entries.get(rel) { |
| 384 | Some(expected) => { |
| 385 | let bytes = fs::read(&path).map_err(io)?; |
| 386 | if digest::sha256_hex(&bytes) == *expected { |
| 387 | verified.insert(rel.clone(), expected.clone()); |
| 388 | } else { |
| 389 | report.quarantined.push(quarantine(&root, rel, attempt)?); |
| 390 | } |
| 391 | } |
| 392 | None => { |
| 393 | unjournaled.insert(rel.clone()); |
| 394 | } |
| 395 | } |
| 396 | } |
| 397 | progress(planned.len(), planned.len()); |
| 398 | report.verified = verified.len(); |
| 399 | report.unjournaled = unjournaled.len(); |
| 400 | for marker in ["cancelled.json", "failed.json"] { |
| 401 | if fs::symlink_metadata(root.join(marker)).is_ok() { |
| 402 | let archived = archive(&root, marker, &format!("attempt-{:03}", attempt - 1))?; |
| 403 | report.previous_marker = Some(archived); |
| 404 | } |
| 405 | } |
| 406 | let mut journal = open_journal(&root)?; |
| 407 | if !newline { |
| 408 | journal.write_all(b"\n").map_err(io)?; |
| 409 | journal.sync_data().map_err(io)?; |
| 410 | } |
| 411 | let entry = serde_json::json!({ |
| 412 | "attempt": attempt, |
| 413 | "planned_png_files": report.planned, |
| 414 | "verified": report.verified, |
| 415 | "missing": report.missing, |
| 416 | "quarantined": report.quarantined, |
| 417 | "unjournaled_existing": report.unjournaled, |
| 418 | "damaged_journal_lines": report.damaged_journal_lines, |
| 419 | "previous_marker": report.previous_marker, |
| 420 | }); |
| 421 | let mut log = OpenOptions::new() |
| 422 | .create(true) |
| 423 | .append(true) |
| 424 | .open(root.join(RESUME_LOG)) |
| 425 | .map_err(io)?; |
| 426 | log.write_all(format!("{entry}\n").as_bytes()).map_err(io)?; |
| 427 | log.sync_data().map_err(io)?; |
| 428 | Ok(Self { |
| 429 | root, |
| 430 | attempt, |
| 431 | verified, |
| 432 | unjournaled, |
| 433 | report, |
| 434 | journal, |
| 435 | _lock: lock, |
| 436 | }) |
| 437 | } |
| 438 | |
| 439 | fn record(&mut self, rel: &str, sha: &str) -> Result<(), String> { |
| 440 | let line = serde_json::json!({"path": rel, "sha256": sha}).to_string() + "\n"; |
| 441 | self.journal.write_all(line.as_bytes()).map_err(io)?; |
| 442 | self.journal.sync_data().map_err(io) |
| 443 | } |
| 444 | |
| 445 | /// Publish encoded PNG bytes for a planned path, never overwriting a file in place. |
| 446 | pub fn publish(&mut self, rel: &str, bytes: &[u8]) -> Result<String, String> { |
| 447 | if let Some(sha) = self.verified.get(rel) { |
| 448 | return Ok(sha.clone()); |
| 449 | } |
| 450 | let sha = digest::sha256_hex(bytes); |
| 451 | if self.unjournaled.remove(rel) { |
| 452 | let path = self.root.join(rel); |
| 453 | match fs::symlink_metadata(&path) { |
| 454 | Ok(meta) if meta.is_file() => { |
| 455 | let existing = fs::read(&path).map_err(io)?; |
| 456 | if existing == bytes { |
| 457 | self.record(rel, &sha)?; |
| 458 | return Ok(sha); |
| 459 | } |
| 460 | let q = quarantine(&self.root, rel, self.attempt)?; |
| 461 | self.report.quarantined.push(q); |
| 462 | } |
| 463 | Ok(_) => { |
| 464 | let q = quarantine(&self.root, rel, self.attempt)?; |
| 465 | self.report.quarantined.push(q); |
| 466 | } |
| 467 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => {} |
| 468 | Err(e) => return Err(e.to_string()), |
| 469 | } |
| 470 | } |
| 471 | crate::production::publish_bytes(&self.root, Path::new(rel), bytes)?; |
| 472 | self.record(rel, &sha)?; |
| 473 | Ok(sha) |
| 474 | } |
| 475 | } |
| 476 | |
| 477 | /// An incomplete run of the current sex/set found in the output directory. |
| 478 | #[derive(Debug, Clone, PartialEq, Eq)] |
| 479 | pub struct ResumeCandidate { |
| 480 | pub dir: PathBuf, |
| 481 | pub name: String, |
| 482 | /// Distinct planned PNGs recorded in the journal (not yet re-verified). |
| 483 | pub journaled: usize, |
| 484 | pub planned: u64, |
| 485 | /// "geannuleerd", "mislukt" or "afgebroken" (no terminal marker: crash/kill). |
| 486 | pub state: String, |
| 487 | /// Why this run cannot be resumed now; None when resumable. |
| 488 | pub blocked: Option<String>, |
| 489 | } |
| 490 | |
| 491 | fn run_stamp(name: &str) -> u128 { |
| 492 | name.strip_prefix(IN_PROGRESS_PREFIX) |
| 493 | .and_then(|r| r.split('-').next()) |
| 494 | .and_then(|n| n.parse().ok()) |
| 495 | .unwrap_or(0) |
| 496 | } |
| 497 | |
| 498 | /// Incomplete runs of the same sex/set as `render_doc`, newest first. |
| 499 | pub fn candidates(output: &Path, render_doc: &Value) -> Result<Vec<ResumeCandidate>, String> { |
| 500 | let expected = normalized(render_doc)?; |
| 501 | let read = match fs::read_dir(output) { |
| 502 | Ok(read) => read, |
| 503 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()), |
| 504 | Err(e) => return Err(e.to_string()), |
| 505 | }; |
| 506 | let mut found = Vec::new(); |
| 507 | for entry in read.take(MAX_SCANNED_RUNS) { |
| 508 | let entry = entry.map_err(io)?; |
| 509 | let name = entry.file_name().to_string_lossy().to_string(); |
| 510 | let Ok(kind) = entry.file_type() else { |
| 511 | continue; |
| 512 | }; |
| 513 | if !name.starts_with(IN_PROGRESS_PREFIX) || !kind.is_dir() { |
| 514 | continue; |
| 515 | } |
| 516 | let dir = entry.path(); |
| 517 | if is_complete(&dir) { |
| 518 | continue; |
| 519 | } |
| 520 | let Ok(existing) = read_render(&dir) else { |
| 521 | continue; |
| 522 | }; |
| 523 | if existing.get("sex") != expected.get("sex") || existing.get("set") != expected.get("set") |
| 524 | { |
| 525 | continue; |
| 526 | } |
| 527 | let state = if dir.join("cancelled.json").exists() { |
| 528 | "geannuleerd" |
| 529 | } else if dir.join("failed.json").exists() { |
| 530 | "mislukt" |
| 531 | } else { |
| 532 | "afgebroken" |
| 533 | }; |
| 534 | let journaled = read_journal(&dir).map(|(e, _, _)| e.len()).unwrap_or(0); |
| 535 | let blocked = if is_locked(&dir) { |
| 536 | Some("wordt op dit moment gerenderd".to_string()) |
| 537 | } else if existing != expected { |
| 538 | Some(format!( |
| 539 | "past niet meer bij de huidige instellingen/bronnen (verschil in: {})", |
| 540 | contract_differences(&existing, &expected).join(", ") |
| 541 | )) |
| 542 | } else { |
| 543 | None |
| 544 | }; |
| 545 | found.push(ResumeCandidate { |
| 546 | dir, |
| 547 | planned: existing |
| 548 | .get("png_files") |
| 549 | .and_then(Value::as_u64) |
| 550 | .unwrap_or(0), |
| 551 | name, |
| 552 | journaled, |
| 553 | state: state.into(), |
| 554 | blocked, |
| 555 | }); |
| 556 | } |
| 557 | found.sort_by_key(|c| std::cmp::Reverse(run_stamp(&c.name))); |
| 558 | found.truncate(MAX_CANDIDATES); |
| 559 | Ok(found) |
| 560 | } |
| 561 | |
| 562 | /// Fixed result folder name: `<sex>-<settype>-atlas`. |
| 563 | pub fn result_dir_name(sex: &str, set: &str) -> String { |
| 564 | format!("{sex}-{set}-atlas") |
| 565 | } |
| 566 | |
| 567 | /// Name for a new in-progress build folder (stamp first, so resume sorts by age). |
| 568 | pub fn in_progress_name(stamp: u128, pid: u32, sex: &str, set: &str) -> String { |
| 569 | format!("{IN_PROGRESS_PREFIX}{stamp}-{pid}-{sex}-{set}") |
| 570 | } |
| 571 | |
| 572 | #[cfg(unix)] |
| 573 | fn sync_dir(dir: &Path) { |
| 574 | if let Ok(f) = File::open(dir) { |
| 575 | let _ = f.sync_all(); |
| 576 | } |
| 577 | } |
| 578 | #[cfg(not(unix))] |
| 579 | fn sync_dir(_dir: &Path) {} |
| 580 | |
| 581 | fn unique_stamp() -> u128 { |
| 582 | std::time::SystemTime::now() |
| 583 | .duration_since(std::time::UNIX_EPOCH) |
| 584 | .map(|d| d.as_nanos()) |
| 585 | .unwrap_or(0) |
| 586 | } |
| 587 | |
| 588 | /// Replace `output/<final_name>` by the completed build folder `built`. |
| 589 | /// |
| 590 | /// The previous result is first renamed aside (`.replaced-<name>-<stamp>`), |
| 591 | /// then the new folder is renamed into place, and only then the old one is |
| 592 | /// removed. If the second rename fails the old result is restored. A crash in |
| 593 | /// between is repaired by [`recover`]. The run lock must already be released. |
| 594 | pub fn promote(output: &Path, built: &Path, final_name: &str) -> Result<PathBuf, String> { |
| 595 | let target = output.join(final_name); |
| 596 | let retired = match fs::symlink_metadata(&target) { |
| 597 | Ok(_) => { |
| 598 | let aside = output.join(format!("{REPLACED_PREFIX}{final_name}-{}", unique_stamp())); |
| 599 | fs::rename(&target, &aside) |
| 600 | .map_err(|e| format!("oude resultaatmap opzijzetten: {e}"))?; |
| 601 | Some(aside) |
| 602 | } |
| 603 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => None, |
| 604 | Err(e) => return Err(e.to_string()), |
| 605 | }; |
| 606 | if let Err(e) = fs::rename(built, &target) { |
| 607 | if let Some(aside) = &retired { |
| 608 | let _ = fs::rename(aside, &target); |
| 609 | } |
| 610 | sync_dir(output); |
| 611 | return Err(format!("nieuwe resultaatmap plaatsen: {e}")); |
| 612 | } |
| 613 | sync_dir(output); |
| 614 | if let Some(aside) = retired { |
| 615 | // Best effort: a leftover is removed by the next `recover`. |
| 616 | let _ = fs::remove_dir_all(&aside); |
| 617 | } |
| 618 | Ok(target) |
| 619 | } |
| 620 | |
| 621 | /// Repair interrupted swaps in `output` (called before each render): |
| 622 | /// a moved-aside previous result is restored when its result folder is missing |
| 623 | /// (or removed when the new one is in place), and a fully completed in-progress |
| 624 | /// folder that never got swapped is promoted. Returns human-readable notes. |
| 625 | pub fn recover(output: &Path) -> Result<Vec<String>, String> { |
| 626 | let mut notes = Vec::new(); |
| 627 | let read = match fs::read_dir(output) { |
| 628 | Ok(read) => read, |
| 629 | Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(notes), |
| 630 | Err(e) => return Err(e.to_string()), |
| 631 | }; |
| 632 | let mut replaced = Vec::new(); |
| 633 | let mut built = Vec::new(); |
| 634 | for entry in read.take(MAX_SCANNED_RUNS).flatten() { |
| 635 | let name = entry.file_name().to_string_lossy().to_string(); |
| 636 | let Ok(kind) = entry.file_type() else { |
| 637 | continue; |
| 638 | }; |
| 639 | if !kind.is_dir() { |
| 640 | continue; |
| 641 | } |
| 642 | if name.starts_with(REPLACED_PREFIX) { |
| 643 | replaced.push(name); |
| 644 | } else if name.starts_with(IN_PROGRESS_PREFIX) { |
| 645 | built.push(name); |
| 646 | } |
| 647 | } |
| 648 | replaced.sort(); |
| 649 | for name in replaced { |
| 650 | let rest = &name[REPLACED_PREFIX.len()..]; |
| 651 | let Some((final_name, _)) = rest.rsplit_once('-') else { |
| 652 | continue; |
| 653 | }; |
| 654 | let target = output.join(final_name); |
| 655 | if fs::symlink_metadata(&target).is_ok() { |
| 656 | let _ = fs::remove_dir_all(output.join(&name)); |
| 657 | } else { |
| 658 | fs::rename(output.join(&name), &target).map_err(io)?; |
| 659 | notes.push(format!("vorige resultaatmap {final_name} hersteld")); |
| 660 | } |
| 661 | } |
| 662 | built.sort(); |
| 663 | for name in built { |
| 664 | let dir = output.join(&name); |
| 665 | if !is_complete(&dir) || is_locked(&dir) { |
| 666 | continue; |
| 667 | } |
| 668 | let Ok(atlas) = crate::atlas::AtlasSet::open_run(&dir) else { |
| 669 | continue; |
| 670 | }; |
| 671 | if crate::export_set::ExportRun::open(&dir).is_err() { |
| 672 | continue; |
| 673 | } |
| 674 | strip_bookkeeping(&dir)?; |
| 675 | let final_name = result_dir_name(&atlas.ppda.sex, &atlas.ppda.settype); |
| 676 | promote(output, &dir, &final_name)?; |
| 677 | notes.push(format!( |
| 678 | "voltooide render {name} alsnog geplaatst als {final_name}" |
| 679 | )); |
| 680 | } |
| 681 | Ok(notes) |
| 682 | } |