BlueGit

golden-age paperdoll-renderer

Files

paperdoll-renderer / src / resume.rs
Code

SSH

HTTPS

Username and a token as the password.

resume.rs

WIP: 60 views, default 4 fps, atlas writer + .ppda reader, output schema 4
timfa Grok Bot · pushed by timfa · 7 days ago
bcc9bf42 History
resume.rs 18.2 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";
27const MAX_RENDER_JSON: u64 = 1024 * 1024;
28const MAX_JOURNAL: u64 = 256 * 1024 * 1024;
29const MAX_CANDIDATES: usize = 20;
30const MAX_SCANNED_RUNS: usize = 2000;
31
32/// Outcome of the pre-render verification of a resumed run.
33#[derive(Debug, Clone, Default, PartialEq, Eq)]
34pub struct ResumeReport {
35 pub attempt: u32,
36 pub planned: usize,
37 pub verified: usize,
38 pub missing: usize,
39 pub quarantined: Vec<String>,
40 pub unjournaled: usize,
41 pub damaged_journal_lines: usize,
42 pub previous_marker: Option<String>,
43}
44
45/// Open write session on one run directory (fresh or resumed).
46pub struct RunSession {
47 pub root: PathBuf,
48 pub attempt: u32,
49 /// Planned PNGs already on disk with a verified journal checksum.
50 pub verified: BTreeMap<String, String>,
51 /// Existing regular PNGs without a journal entry; compared on regeneration.
52 pub unjournaled: BTreeSet<String>,
53 pub report: ResumeReport,
54 journal: File,
55 _lock: File,
56}
57
58fn io(e: std::io::Error) -> String {
59 e.to_string()
60}
61
62fn lock_run(root: &Path) -> Result<File, String> {
63 let path = root.join(LOCK);
64 if let Ok(meta) = fs::symlink_metadata(&path) {
65 if !meta.is_file() {
66 return Err("run.lock is geen regulier bestand".into());
67 }
68 }
69 let file = OpenOptions::new()
70 .create(true)
71 .truncate(false)
72 .write(true)
73 .open(&path)
74 .map_err(io)?;
75 match file.try_lock() {
76 Ok(()) => Ok(file),
77 Err(fs::TryLockError::WouldBlock) => {
78 Err("deze run wordt op dit moment al gerenderd (run.lock bezet)".into())
79 }
80 Err(fs::TryLockError::Error(e)) => Err(format!("run.lock: {e}")),
81 }
82}
83
84fn is_locked(root: &Path) -> bool {
85 let Ok(file) = OpenOptions::new().write(true).open(root.join(LOCK)) else {
86 return false;
87 };
88 matches!(file.try_lock(), Err(fs::TryLockError::WouldBlock))
89}
90
91fn open_journal(root: &Path) -> Result<File, String> {
92 let path = root.join(JOURNAL);
93 if let Ok(meta) = fs::symlink_metadata(&path) {
94 if !meta.is_file() {
95 return Err("progress.jsonl is geen regulier bestand".into());
96 }
97 }
98 OpenOptions::new()
99 .create(true)
100 .append(true)
101 .open(path)
102 .map_err(io)
103}
104
105fn read_regular(path: &Path, max: u64) -> Result<Option<Vec<u8>>, String> {
106 let meta = match fs::symlink_metadata(path) {
107 Ok(meta) => meta,
108 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
109 Err(e) => return Err(e.to_string()),
110 };
111 if !meta.is_file() {
112 return Err(format!("{} is geen regulier bestand", path.display()));
113 }
114 if meta.len() > max {
115 return Err(format!("{} groter dan {max} bytes", path.display()));
116 }
117 let mut bytes = Vec::new();
118 File::open(path)
119 .map_err(io)?
120 .take(max + 1)
121 .read_to_end(&mut bytes)
122 .map_err(io)?;
123 if bytes.len() as u64 > max {
124 return Err(format!("{} groter dan {max} bytes", path.display()));
125 }
126 Ok(Some(bytes))
127}
128
129/// Journal entries: last valid entry per path wins; returns (entries, damaged lines, ends with newline).
130pub fn read_journal(root: &Path) -> Result<(BTreeMap<String, String>, usize, bool), String> {
131 let Some(bytes) = read_regular(&root.join(JOURNAL), MAX_JOURNAL)? else {
132 return Ok((BTreeMap::new(), 0, true));
133 };
134 let mut entries = BTreeMap::new();
135 let mut damaged = 0;
136 for line in bytes.split(|b| *b == b'\n') {
137 if line.is_empty() {
138 continue;
139 }
140 let parsed = serde_json::from_slice::<Value>(line).ok().and_then(|v| {
141 let path = v.get("path")?.as_str()?.to_string();
142 let sha = v.get("sha256")?.as_str()?.to_string();
143 digest::is_sha256_hex(&sha).then_some((path, sha))
144 });
145 match parsed {
146 Some((path, sha)) => {
147 entries.insert(path, sha);
148 }
149 None => damaged += 1,
150 }
151 }
152 Ok((entries, damaged, bytes.last().is_none_or(|b| *b == b'\n')))
153}
154
155/// Move a bad file/link aside inside the run without deleting or clobbering anything.
156pub(crate) fn quarantine(root: &Path, rel: &str, attempt: u32) -> Result<String, String> {
157 let path = root.join(rel);
158 let name = path
159 .file_name()
160 .and_then(|n| n.to_str())
161 .ok_or("ongeldige bestandsnaam")?
162 .to_string();
163 for n in 0..1000 {
164 let target = path.with_file_name(format!("{name}.corrupt-a{attempt:03}-{n}"));
165 match fs::hard_link(&path, &target) {
166 Ok(()) => {
167 fs::remove_file(&path).map_err(io)?;
168 let rel_target = Path::new(rel)
169 .with_file_name(target.file_name().unwrap_or_default())
170 .to_string_lossy()
171 .replace('\\', "/");
172 return Ok(rel_target);
173 }
174 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
175 Err(e) => return Err(format!("quarantaine van {rel} mislukt: {e}")),
176 }
177 }
178 Err(format!("geen vrije quarantainenaam voor {rel}"))
179}
180
181/// Move a file to a fresh unique name (no clobber), returning the new file name.
182fn archive(root: &Path, name: &str, prefix: &str) -> Result<String, String> {
183 for n in 0..1000 {
184 let target = if n == 0 {
185 format!("{prefix}-{name}")
186 } else {
187 format!("{prefix}-{n}-{name}")
188 };
189 match fs::hard_link(root.join(name), root.join(&target)) {
190 Ok(()) => {
191 fs::remove_file(root.join(name)).map_err(io)?;
192 return Ok(target);
193 }
194 Err(e) if e.kind() == std::io::ErrorKind::AlreadyExists => continue,
195 Err(e) => return Err(format!("archiveren van {name} mislukt: {e}")),
196 }
197 }
198 Err(format!("geen vrije archiefnaam voor {name}"))
199}
200
201/// Top-level keys whose values differ between two JSON objects.
202pub fn contract_differences(existing: &Value, expected: &Value) -> Vec<String> {
203 match (existing.as_object(), expected.as_object()) {
204 (Some(a), Some(b)) => a
205 .keys()
206 .chain(b.keys())
207 .collect::<BTreeSet<_>>()
208 .into_iter()
209 .filter(|k| a.get(*k) != b.get(*k))
210 .cloned()
211 .collect(),
212 _ => vec!["render.json".into()],
213 }
214}
215
216fn normalized(doc: &Value) -> Result<Value, String> {
217 serde_json::from_str(&serde_json::to_string(doc).map_err(|e| e.to_string())?)
218 .map_err(|e| e.to_string())
219}
220
221fn read_render(root: &Path) -> Result<Value, String> {
222 let bytes =
223 read_regular(&root.join("render.json"), MAX_RENDER_JSON)?.ok_or("render.json ontbreekt")?;
224 serde_json::from_slice(&bytes).map_err(|e| format!("render.json: {e}"))
225}
226
227fn check_contract(root: &Path, render_doc: &Value) -> Result<(), String> {
228 let existing = read_render(root)?;
229 let expected = normalized(render_doc)?;
230 if existing != expected {
231 return Err(format!(
232 "huidige instellingen/bronnen passen niet bij deze run (verschil in: {})",
233 contract_differences(&existing, &expected).join(", ")
234 ));
235 }
236 Ok(())
237}
238
239/// Direct `run-*` child directory of the output directory, not a link.
240fn run_root(output: &Path, dir: &Path) -> Result<PathBuf, String> {
241 let meta = fs::symlink_metadata(dir).map_err(io)?;
242 if !meta.is_dir() {
243 return Err("runmap is geen gewone map".into());
244 }
245 let out = output.canonicalize().map_err(io)?;
246 let root = dir.canonicalize().map_err(io)?;
247 let name_ok = root
248 .file_name()
249 .and_then(|n| n.to_str())
250 .is_some_and(|n| n.starts_with("run-"));
251 if root.parent() != Some(out.as_path()) || !name_ok {
252 return Err("runmap ligt niet direct in de uitvoermap".into());
253 }
254 Ok(root)
255}
256
257impl RunSession {
258 /// A new, empty run directory created by the caller.
259 pub fn fresh(root: &Path) -> Result<Self, String> {
260 let lock = lock_run(root)?;
261 let journal = open_journal(root)?;
262 Ok(Self {
263 root: root.to_path_buf(),
264 attempt: 0,
265 verified: BTreeMap::new(),
266 unjournaled: BTreeSet::new(),
267 report: ResumeReport::default(),
268 journal,
269 _lock: lock,
270 })
271 }
272
273 /// Lock, contract-check and verify an incomplete run before re-rendering.
274 /// `planned` lists every relative target PNG of the job in render order.
275 pub fn resume(
276 output: &Path,
277 dir: &Path,
278 render_doc: &Value,
279 planned: &[String],
280 progress: &mut dyn FnMut(usize, usize),
281 ) -> Result<Self, String> {
282 let root = run_root(output, dir)?;
283 if fs::symlink_metadata(root.join("complete.json")).is_ok() {
284 return Err("deze run is al voltooid (complete.json bestaat)".into());
285 }
286 let lock = lock_run(&root)?;
287 check_contract(&root, render_doc)?;
288 let planned_set: HashSet<&str> = planned.iter().map(String::as_str).collect();
289 let (journal_entries, damaged, newline) = read_journal(&root)?;
290 if let Some(unknown) = journal_entries
291 .keys()
292 .find(|k| !planned_set.contains(k.as_str()))
293 {
294 return Err(format!("journal noemt een onbekend pad: {unknown}"));
295 }
296 let previous_attempts = read_regular(&root.join(RESUME_LOG), MAX_RENDER_JSON)?
297 .map(|b| b.split(|c| *c == b'\n').filter(|l| !l.is_empty()).count())
298 .unwrap_or(0);
299 let attempt = previous_attempts as u32 + 1;
300 let mut report = ResumeReport {
301 attempt,
302 planned: planned.len(),
303 damaged_journal_lines: damaged,
304 ..Default::default()
305 };
306 let mut verified = BTreeMap::new();
307 let mut unjournaled = BTreeSet::new();
308 for (i, rel) in planned.iter().enumerate() {
309 if i % 256 == 0 {
310 progress(i, planned.len());
311 }
312 let path = root.join(rel);
313 let meta = match fs::symlink_metadata(&path) {
314 Ok(meta) => meta,
315 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
316 report.missing += 1;
317 continue;
318 }
319 Err(e) => return Err(e.to_string()),
320 };
321 if meta.is_dir() {
322 return Err(format!("map staat op de plaats van PNG {rel}"));
323 }
324 if !meta.is_file() || meta.len() > crate::export_set::MAX_PNG_BYTES {
325 report.quarantined.push(quarantine(&root, rel, attempt)?);
326 continue;
327 }
328 match journal_entries.get(rel) {
329 Some(expected) => {
330 let bytes = fs::read(&path).map_err(io)?;
331 if digest::sha256_hex(&bytes) == *expected {
332 verified.insert(rel.clone(), expected.clone());
333 } else {
334 report.quarantined.push(quarantine(&root, rel, attempt)?);
335 }
336 }
337 None => {
338 unjournaled.insert(rel.clone());
339 }
340 }
341 }
342 progress(planned.len(), planned.len());
343 report.verified = verified.len();
344 report.unjournaled = unjournaled.len();
345 for marker in ["cancelled.json", "failed.json"] {
346 if fs::symlink_metadata(root.join(marker)).is_ok() {
347 let archived = archive(&root, marker, &format!("attempt-{:03}", attempt - 1))?;
348 report.previous_marker = Some(archived);
349 }
350 }
351 let mut journal = open_journal(&root)?;
352 if !newline {
353 journal.write_all(b"\n").map_err(io)?;
354 journal.sync_data().map_err(io)?;
355 }
356 let entry = serde_json::json!({
357 "attempt": attempt,
358 "planned_png_files": report.planned,
359 "verified": report.verified,
360 "missing": report.missing,
361 "quarantined": report.quarantined,
362 "unjournaled_existing": report.unjournaled,
363 "damaged_journal_lines": report.damaged_journal_lines,
364 "previous_marker": report.previous_marker,
365 });
366 let mut log = OpenOptions::new()
367 .create(true)
368 .append(true)
369 .open(root.join(RESUME_LOG))
370 .map_err(io)?;
371 log.write_all(format!("{entry}\n").as_bytes()).map_err(io)?;
372 log.sync_data().map_err(io)?;
373 Ok(Self {
374 root,
375 attempt,
376 verified,
377 unjournaled,
378 report,
379 journal,
380 _lock: lock,
381 })
382 }
383
384 fn record(&mut self, rel: &str, sha: &str) -> Result<(), String> {
385 let line = serde_json::json!({"path": rel, "sha256": sha}).to_string() + "\n";
386 self.journal.write_all(line.as_bytes()).map_err(io)?;
387 self.journal.sync_data().map_err(io)
388 }
389
390 /// Publish encoded PNG bytes for a planned path, never overwriting a file in place.
391 pub fn publish(&mut self, rel: &str, bytes: &[u8]) -> Result<String, String> {
392 if let Some(sha) = self.verified.get(rel) {
393 return Ok(sha.clone());
394 }
395 let sha = digest::sha256_hex(bytes);
396 if self.unjournaled.remove(rel) {
397 let path = self.root.join(rel);
398 match fs::symlink_metadata(&path) {
399 Ok(meta) if meta.is_file() => {
400 let existing = fs::read(&path).map_err(io)?;
401 if existing == bytes {
402 self.record(rel, &sha)?;
403 return Ok(sha);
404 }
405 let q = quarantine(&self.root, rel, self.attempt)?;
406 self.report.quarantined.push(q);
407 }
408 Ok(_) => {
409 let q = quarantine(&self.root, rel, self.attempt)?;
410 self.report.quarantined.push(q);
411 }
412 Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
413 Err(e) => return Err(e.to_string()),
414 }
415 }
416 crate::production::publish_bytes(&self.root, Path::new(rel), bytes)?;
417 self.record(rel, &sha)?;
418 Ok(sha)
419 }
420}
421
422/// An incomplete run of the current sex/set found in the output directory.
423#[derive(Debug, Clone, PartialEq, Eq)]
424pub struct ResumeCandidate {
425 pub dir: PathBuf,
426 pub name: String,
427 /// Distinct planned PNGs recorded in the journal (not yet re-verified).
428 pub journaled: usize,
429 pub planned: u64,
430 /// "geannuleerd", "mislukt" or "afgebroken" (no terminal marker: crash/kill).
431 pub state: String,
432 /// Why this run cannot be resumed now; None when resumable.
433 pub blocked: Option<String>,
434}
435
436fn run_stamp(name: &str) -> u128 {
437 name.strip_prefix("run-")
438 .and_then(|r| r.split('-').next())
439 .and_then(|n| n.parse().ok())
440 .unwrap_or(0)
441}
442
443/// Incomplete runs of the same sex/set as `render_doc`, newest first.
444pub fn candidates(output: &Path, render_doc: &Value) -> Result<Vec<ResumeCandidate>, String> {
445 let expected = normalized(render_doc)?;
446 let read = match fs::read_dir(output) {
447 Ok(read) => read,
448 Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
449 Err(e) => return Err(e.to_string()),
450 };
451 let mut found = Vec::new();
452 for entry in read.take(MAX_SCANNED_RUNS) {
453 let entry = entry.map_err(io)?;
454 let name = entry.file_name().to_string_lossy().to_string();
455 let Ok(kind) = entry.file_type() else {
456 continue;
457 };
458 if !name.starts_with("run-") || !kind.is_dir() {
459 continue;
460 }
461 let dir = entry.path();
462 if fs::symlink_metadata(dir.join("complete.json")).is_ok() {
463 continue;
464 }
465 let Ok(existing) = read_render(&dir) else {
466 continue;
467 };
468 if existing.get("sex") != expected.get("sex") || existing.get("set") != expected.get("set")
469 {
470 continue;
471 }
472 let state = if dir.join("cancelled.json").exists() {
473 "geannuleerd"
474 } else if dir.join("failed.json").exists() {
475 "mislukt"
476 } else {
477 "afgebroken"
478 };
479 let journaled = read_journal(&dir).map(|(e, _, _)| e.len()).unwrap_or(0);
480 let blocked = if is_locked(&dir) {
481 Some("wordt op dit moment gerenderd".to_string())
482 } else if existing != expected {
483 Some(format!(
484 "past niet meer bij de huidige instellingen/bronnen (verschil in: {})",
485 contract_differences(&existing, &expected).join(", ")
486 ))
487 } else {
488 None
489 };
490 found.push(ResumeCandidate {
491 dir,
492 planned: existing
493 .get("png_files")
494 .and_then(Value::as_u64)
495 .unwrap_or(0),
496 name,
497 journaled,
498 state: state.into(),
499 blocked,
500 });
501 }
502 found.sort_by_key(|c| std::cmp::Reverse(run_stamp(&c.name)));
503 found.truncate(MAX_CANDIDATES);
504 Ok(found)
505}