"""Recoverable date correction primitive for a future supervised executor. Caller MUST fence every content/SQL writer and verify a fresh immutable backup before calling. This module neither grants admission nor proves that fencing. Pending journals require reconciliation before writers may resume. """ from __future__ import annotations import base64 import hashlib import json import os import re import tempfile from pathlib import Path from django.db import transaction from .bubble_transfer import source_timestamp from .date_backfill import prepare_index_date_patch from .models import Space def _digest(data): return hashlib.sha256(data).hexdigest() def _sync_directory(path): fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY) try: os.fsync(fd) finally: os.close(fd) def _replace(path, data): previous = path.stat() fd, temporary = tempfile.mkstemp(prefix=".date-patch-", dir=path.parent) try: with os.fdopen(fd, "wb") as stream: os.fchown(stream.fileno(), previous.st_uid, previous.st_gid) os.fchmod(stream.fileno(), previous.st_mode & 0o777) stream.write(data) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, path) _sync_directory(path.parent) finally: if os.path.exists(temporary): os.unlink(temporary) def _index(root, plan): # Refuse symlinks at every component, not just at index.md. root = Path(root).absolute() if any(p.is_symlink() for p in [root, *root.parents]): raise ValueError("Real content root required") slug = plan["selection"]["space_slug"] if not re.fullmatch(r"[a-z0-9]+(?:-[a-z0-9]+)*", slug): raise ValueError("Invalid destination slug") path = root / "spaces" / slug / "pages" / "index.md" if any(p.is_symlink() for p in [path, *path.parents]): raise ValueError("Symlink destination refused") return path.resolve() def _check_pages(index, plan, permitted_index_hashes): expected = {p["slug"]: p["observed_sha256"] for p in plan["pages"]} if len(expected) != len(plan["pages"]) or "index" in expected: raise ValueError("Ambiguous observed pages") paths = list(index.parent.glob("*.md")) if any(p.is_symlink() or not p.is_file() for p in paths): raise ValueError("Real page files required") observed = {p.stem: _digest(p.read_bytes()) for p in paths} index_hash = observed.pop("index", None) if observed != expected or index_hash not in permitted_index_hashes: raise ValueError("Observed page bytes changed") def _target(plan): selected = plan["selection"] return Space.objects.select_for_update().get( pk=selected["space_id"], tenant_id=selected["tenant_id"], slug=selected["space_slug"]) def _journal_path(path): path = Path(path).absolute() if (any(p.is_symlink() for p in [path, *path.parents]) or path.parent.stat().st_mode & 0o077): raise ValueError("Private real journal directory required") return path.resolve() def apply_date_patch(root, plan, journal, *, backup_sha256, checkpoint=lambda phase: None): """Single attempt. Never automatically retry a pending or completed journal. backup_sha256 binds the caller's verified fresh-backup receipt, not its data. The journal contains private original bytes and must stay outside public content. checkpoint is a rehearsal fault injector; production callers leave it unset. """ if not re.fullmatch(r"[a-f0-9]{64}", backup_sha256): raise ValueError("Verified backup receipt digest required") index = _index(root, plan) journal = _journal_path(journal) if journal.is_relative_to(Path(root).resolve()): raise ValueError("Journal must be outside served content") original = index.read_bytes() patched = prepare_index_date_patch(plan, original) record = { "schema": "coulomb.source-date-journal.v1", "state": "pending", "plan": plan, "backup_receipt_sha256": backup_sha256, "original": base64.b64encode(original).decode(), "patched_sha256": _digest(patched), } # Durable intent is created before either store changes. Exclusive creation # prevents accidental replay after a crash or a successful earlier attempt. with transaction.atomic(durable=True): target = _target(plan) prior = source_timestamp(plan["observed_sql_created_at"]) desired = source_timestamp(plan["source_created_at"]) if target.created_at != prior: raise ValueError("Observed SQL date changed") _check_pages(index, plan, {_digest(original)}) fd = os.open(journal, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "wb") as stream: stream.write(json.dumps(record, sort_keys=True).encode()) stream.flush() os.fsync(stream.fileno()) _sync_directory(journal.parent) checkpoint("intent") if Space.objects.filter(pk=target.pk, created_at=prior).update(created_at=desired) != 1: raise ValueError("SQL precondition changed") checkpoint("sql") _replace(index, patched) checkpoint("file") checkpoint("commit") record["state"] = "completed" _replace(journal, json.dumps(record, sort_keys=True).encode()) def reconcile_date_patch(root, journal): """With writers STILL fenced, use committed SQL to reconcile pending bytes. Unknown SQL/file state refuses rather than overwriting a later edit. A lost commit reply is resolved by the persisted SQL date. Keep the journal as proof. """ journal = _journal_path(journal) if journal.stat().st_mode & 0o077: raise ValueError("Private journal required") record = json.loads(journal.read_bytes()) if record["schema"] != "coulomb.source-date-journal.v1" or record["state"] != "pending": raise ValueError("Pending journal required") plan = record["plan"] index = _index(root, plan) if journal.is_relative_to(Path(root).resolve()): raise ValueError("Journal must be outside served content") original = base64.b64decode(record["original"], validate=True) patched = prepare_index_date_patch(plan, original) if _digest(patched) != record["patched_sha256"]: raise ValueError("Journal patch binding changed") with transaction.atomic(durable=True): target = _target(plan) prior = source_timestamp(plan["observed_sql_created_at"]) desired = source_timestamp(plan["source_created_at"]) _check_pages(index, plan, {_digest(original), _digest(patched)}) if target.created_at == desired: result, data = "completed", patched elif target.created_at == prior: result, data = "rolled_back", original else: raise ValueError("Unknown SQL state; manual reconciliation required") _replace(index, data) record["state"] = result _replace(journal, json.dumps(record, sort_keys=True).encode()) return result