File size: 8,744 Bytes
bc50ea0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 | """Recover interrupted PNG publication without touching earlier scene frames.
The journal is written only after a whole scene has been staged and synced.
It grants ownership of exactly the next frame range, never arbitrary PNGs.
All calls are made under the export folder's process lock.
"""
from contextlib import contextmanager
from copy import deepcopy
import json
from pathlib import Path
import re
import shutil
import tempfile
import uuid
from . import processing_persistence as persistence
PENDING = ".png_pending.json"
FORMAT = "h3_png_publication_v1"
def _same_record(left, right):
# A fallback copy changes stat hints, not scene identity or pixel bytes.
def identity(record):
record = deepcopy(record)
if isinstance(record, dict):
for clip in record.get("clips", []):
for item in clip.get("files", []):
item.pop("mtime_ns", None)
return record
return identity(left) == identity(right)
def _cleanup_stage(chain, stage):
if any(stage.glob("conflict_*")):
chain._LOG.warning("H3 PNG recovery preserved conflicting interrupted files in %s", stage)
return
shutil.rmtree(stage, ignore_errors=True)
@contextmanager
def staging(chain, directory, index):
stage = Path(tempfile.mkdtemp(prefix=".png_scene_%04d_" % index, dir=directory))
try:
yield stage
finally:
# Keep staged pixels if publication or its acknowledgement is uncertain.
# On restart the journal can finish the scene without decoding again.
try:
pending = (directory / PENDING).exists()
except OSError:
pending = True
if not pending:
_cleanup_stage(chain, stage)
def _matches(chain, path, item):
return path.is_file() and path.stat().st_size == item["size"] and chain._file_sha256(str(path)) == item["sha256"]
def _read(path):
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else None
def _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=False):
stage = safe_path(root, directory / journal["stage"])
record = deepcopy(journal["record"])
for item in record["clips"][-1]["files"]:
chain._png_export_check_interrupted()
target = safe_path(root, directory / item["file"])
source = safe_path(root, stage / item["file"])
if recovering and _matches(chain, target, item):
pass
else:
if not _matches(chain, source, item):
raise ValueError("Interrupted PNG staging is missing or changed: %s. Saved scenes were kept." % source)
if target.exists():
if not recovering or not target.is_file():
raise FileExistsError("PNG publication will not overwrite %s" % target)
# A killed network copy can leave a short final file. Preserve
# any conflicting bytes (including manual edits), then recover
# from the verified complete staging file. Never delete them.
preserved = safe_path(root, stage / ("conflict_%s_%s" % (uuid.uuid4().hex, target.name)))
target.rename(preserved)
persistence.sync_directory(stage)
persistence.sync_directory(directory)
chain._LOG.warning("H3 PNG recovery preserved an interrupted/conflicting file at %s", preserved)
publish_frame(source, target)
item["mtime_ns"] = target.stat().st_mtime_ns
persistence.sync_directory(directory)
persistence.atomic_json(directory / "export.json", record)
(directory / PENDING).unlink()
persistence.sync_directory(directory)
_cleanup_stage(chain, stage)
return record
def publish(chain, root, directory, stage, previous, record, safe_path, publish_frame):
pending = safe_path(root, directory / PENDING)
if pending.exists():
raise ValueError("Recover the pending PNG publication before appending another scene.")
for item in record["clips"][-1]["files"]:
if safe_path(root, directory / item["file"]).exists():
raise FileExistsError("PNG publication will not overwrite %s" % item["file"])
persistence.sync_directory(stage)
persistence.sync_directory(directory)
journal = {"format": FORMAT, "stage": stage.name,
"previous": previous, "record": record}
persistence.atomic_json(pending, journal)
try:
return _finish(chain, root, directory, journal, safe_path, publish_frame)
except BaseException:
# Normal cancellation rolls back only verified files from this attempt.
# An uncertain index commit / unreachable share retains the journal and
# media. In particular, never undo a successfully replaced export.json.
try:
current = _read(safe_path(root, directory / "export.json"))
if not _same_record(current, record) and _same_record(current, previous):
for item in record["clips"][-1]["files"]:
target = safe_path(root, directory / item["file"])
if target.exists():
if not _matches(chain, target, item):
raise ValueError("Uncertain interrupted PNG publication")
target.unlink()
persistence.sync_directory(directory)
pending.unlink(missing_ok=True)
persistence.sync_directory(directory)
except (OSError, ValueError):
chain._LOG.warning("H3 PNG publication will be recovered on retry: %s", pending)
raise
def recover(chain, root, directory, previous, config, contracts, safe_path, publish_frame):
pending = safe_path(root, directory / PENDING)
journal = _read(pending)
if journal is None:
return previous
if (not isinstance(journal, dict) or journal.get("format") != FORMAT
or not re.fullmatch(r"\.png_scene_[0-9]{4,}_[a-zA-Z0-9_-]+", str(journal.get("stage", "")))):
raise ValueError("Invalid pending PNG journal; existing files were kept.")
stage = safe_path(root, directory / journal["stage"])
record, base = journal.get("record"), journal.get("previous")
if (not isinstance(record, dict) or record.get("format") != "h3_video_png_sequence_v1"
or record.get("settings") != config or not isinstance(record.get("clips"), list)
or not record["clips"] or (base is not None and not isinstance(base, dict))):
raise ValueError("Pending PNG scene belongs to different settings; existing files were kept.")
old_clips = (base or {}).get("clips", [])
if not isinstance(old_clips, list):
raise ValueError("Invalid pending PNG prefix; existing files were kept.")
clip = record["clips"][-1]
first = config["first_frame_number"] + sum(c["delivered_frames"] for c in old_clips)
if (record["clips"][:-1] != old_clips or not isinstance(clip, dict)
or clip.get("source_contract") != contracts.get(clip.get("index"))
or (old_clips and clip["index"] != old_clips[-1]["index"] + 1)
or clip.get("first_frame_number") != first
or not isinstance(clip.get("files"), list)
or len(clip["files"]) != clip.get("delivered_frames")
or not clip["files"]
or not journal["stage"].startswith(".png_scene_%04d_" % clip["index"])
or record.get("frame_count") != first + len(clip["files"]) - config["first_frame_number"]):
raise ValueError("Pending PNG scene has a different branch/order; existing files were kept.")
for offset, item in enumerate(clip["files"]):
if (not isinstance(item, dict) or item.get("file") != "frame_%08d.png" % (first + offset)
or not re.fullmatch(r"[0-9a-f]{64}", str(item.get("sha256", "")))
or not isinstance(item.get("size"), int) or item["size"] <= 0):
raise ValueError("Invalid pending PNG frame address; existing files were kept.")
if _same_record(previous, record):
# The index committed before the process stopped; normal validation has
# already checked every published PNG. Only private staging is obsolete.
pending.unlink()
persistence.sync_directory(directory)
_cleanup_stage(chain, stage)
return previous
if not _same_record(previous, base):
raise ValueError("PNG export changed since the interrupted publication; existing files were kept.")
result = _finish(chain, root, directory, journal, safe_path, publish_frame, recovering=True)
chain._LOG.info("H3 PNG recovered committed scene %d from its publication journal", clip["index"])
return result
|