BlueGit

golden-age paperdoll-renderer

Files

paperdoll-renderer / src / resume.rs
Code

SSH

HTTPS

Username and a token as the password.

resume.rs

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