tegwick
e75d3b0f6a
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 7m14s
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
235 lines
10 KiB
Python
235 lines
10 KiB
Python
"""Opt-in corpus checkpoints; no automatic repair of uncertain DB/file commits.
|
|
|
|
The caller still owns source freezing, external writer fencing, paired backups
|
|
and admission. A started receipt survives a crash and stops automatic retries.
|
|
Completed receipts skip their space, preserving any subsequent local edits.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import fcntl
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import stat
|
|
import tempfile
|
|
import uuid
|
|
from contextlib import contextmanager
|
|
from pathlib import Path
|
|
|
|
from django.db import connection
|
|
|
|
from . import pageops
|
|
from .import_tree import import_space_tree
|
|
from .models import Space, SpaceMembership
|
|
|
|
SCHEMA = "coulomb.corpus-import-checkpoint.v1"
|
|
|
|
|
|
def digest(value):
|
|
return hashlib.sha256(json.dumps(value, sort_keys=True, default=str).encode()).hexdigest()
|
|
|
|
|
|
def tree_digest(tree):
|
|
files = []
|
|
for path in sorted(tree.rglob("*")):
|
|
mode = path.lstat().st_mode
|
|
if stat.S_ISDIR(mode):
|
|
continue
|
|
if not stat.S_ISREG(mode):
|
|
raise ValueError("Import tree must contain only real files and directories")
|
|
with path.open("rb") as stream:
|
|
checksum = hashlib.file_digest(stream, "sha256").hexdigest()
|
|
files.append([path.relative_to(tree).as_posix(), checksum])
|
|
return digest(files)
|
|
|
|
|
|
def private_path(path, *, directory=False):
|
|
if any(p.is_symlink() for p in (path, *path.parents)):
|
|
raise ValueError("Receipt path must not contain symlinks")
|
|
if path.exists():
|
|
info = path.stat()
|
|
expected = stat.S_ISDIR if directory else stat.S_ISREG
|
|
if (not expected(info.st_mode) or info.st_uid != os.getuid()
|
|
or info.st_mode & 0o077 or (not directory and info.st_nlink != 1)):
|
|
raise ValueError("Receipt path must be private and owned by this operator")
|
|
|
|
|
|
def sync_directory(path):
|
|
fd = os.open(path, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
|
|
|
|
def sync_tree(tree):
|
|
# Content must be durable before its completed receipt. The underlying
|
|
# importer commits SQL first; a crash before this finishes remains started.
|
|
directories = [tree]
|
|
for path in tree.rglob("*"):
|
|
mode = path.lstat().st_mode
|
|
if stat.S_ISDIR(mode):
|
|
directories.append(path)
|
|
elif stat.S_ISREG(mode):
|
|
fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW)
|
|
try:
|
|
os.fsync(fd)
|
|
finally:
|
|
os.close(fd)
|
|
else:
|
|
raise ValueError("Imported tree contains an unsupported file")
|
|
for directory in sorted(directories, key=lambda p: len(p.parts), reverse=True):
|
|
sync_directory(directory)
|
|
sync_directory(tree.parent)
|
|
sync_directory(tree.parent.parent)
|
|
|
|
|
|
class CorpusReceipts:
|
|
def __init__(self, directory, manifest, plan):
|
|
self.directory = Path(directory).expanduser().absolute()
|
|
private_path(self.directory, directory=True)
|
|
self.directory = self.directory.resolve()
|
|
root = pageops.content_root().resolve()
|
|
for protected in (root, *(tree for tree, _, _, _ in plan)):
|
|
if (self.directory.is_relative_to(protected)
|
|
or protected.is_relative_to(self.directory)):
|
|
raise ValueError("Receipt directory must be separate from source and content")
|
|
self.path = self.directory / "checkpoint.json"
|
|
self.plan = plan
|
|
self.source_hashes = [tree_digest(tree) for tree, _, _, _ in plan]
|
|
self.binding = digest({
|
|
"manifest": hashlib.sha256(Path(manifest).read_bytes()).hexdigest(),
|
|
"content_root": str(root),
|
|
"database": {key: connection.settings_dict.get(key)
|
|
for key in ("ENGINE", "HOST", "PORT", "NAME", "USER")},
|
|
"units": [
|
|
[str(tree), tenant, slug, source_hash,
|
|
[[str(member.pk), role] for member, role in grants]]
|
|
for (tree, tenant, slug, grants), source_hash
|
|
in zip(plan, self.source_hashes, strict=True)
|
|
],
|
|
})
|
|
|
|
def read(self):
|
|
private_path(self.directory, directory=True)
|
|
private_path(self.path)
|
|
if not self.path.exists():
|
|
return {"schema": SCHEMA, "binding": self.binding,
|
|
"spaces": [{"state": "pending"} for _ in self.plan]}
|
|
data = json.loads(self.path.read_bytes())
|
|
if (not isinstance(data, dict) or data.get("schema") != SCHEMA
|
|
or data.get("binding") != self.binding
|
|
or not isinstance(data.get("spaces"), list)
|
|
or len(data["spaces"]) != len(self.plan)):
|
|
raise ValueError("Receipt does not match this corpus, mapping or destination")
|
|
return data
|
|
|
|
def validate(self, data):
|
|
for unit, (_, tenant, slug, _) in zip(data["spaces"], self.plan, strict=True):
|
|
if not isinstance(unit, dict):
|
|
raise ValueError("Invalid space checkpoint")
|
|
state = unit.get("state")
|
|
if state == "completed":
|
|
try:
|
|
uuid.UUID(unit["space_id"])
|
|
if any(not re.fullmatch(r"[a-f0-9]{64}", unit[key])
|
|
for key in ("files_sha256", "space_sha256", "grants_sha256")):
|
|
raise ValueError
|
|
if any(type(unit[key]) is not int or unit[key] < 0
|
|
for key in ("pages", "assets")):
|
|
raise ValueError
|
|
except (KeyError, TypeError, ValueError, AttributeError) as exc:
|
|
raise ValueError("Invalid completed space checkpoint") from exc
|
|
# Never recreate an independently deleted/replaced space or infer
|
|
# that a later edit or grant change should be reversed.
|
|
if not Space.objects.filter(pk=unit.get("space_id"), tenant_id=tenant,
|
|
slug=slug).exists():
|
|
raise ValueError("Completed space identity changed; reconciliation required")
|
|
elif state == "pending":
|
|
if (Space.objects.filter(slug=slug).exists()
|
|
or pageops.space_dir(slug).exists()):
|
|
raise ValueError("Unreceipted destination exists; reconciliation required")
|
|
else:
|
|
raise ValueError(
|
|
"Uncertain space import; reconcile the started receipt before retry")
|
|
|
|
def save(self, data):
|
|
private_path(self.path)
|
|
fd, temporary = tempfile.mkstemp(prefix=".checkpoint-", dir=self.directory)
|
|
try:
|
|
with os.fdopen(fd, "w") as stream:
|
|
json.dump(data, stream, sort_keys=True)
|
|
stream.flush()
|
|
os.fsync(stream.fileno())
|
|
os.replace(temporary, self.path)
|
|
sync_directory(self.directory)
|
|
finally:
|
|
if os.path.exists(temporary):
|
|
os.unlink(temporary)
|
|
|
|
@contextmanager
|
|
def lock(self):
|
|
# Parent must already exist; do not chmod an operator's existing path.
|
|
self.directory.mkdir(mode=0o700, exist_ok=True)
|
|
private_path(self.directory, directory=True)
|
|
sync_directory(self.directory.parent)
|
|
lock = self.directory / ".lock"
|
|
private_path(lock)
|
|
fd = os.open(lock, os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
|
|
try:
|
|
try:
|
|
fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
except BlockingIOError as exc:
|
|
raise ValueError("Another importer owns this receipt directory") from exc
|
|
yield
|
|
finally:
|
|
os.close(fd)
|
|
|
|
def run(self):
|
|
if connection.in_atomic_block or not connection.get_autocommit():
|
|
raise ValueError("Receipted import requires independent database commits")
|
|
self.validate(self.read()) # Invalid inputs remain write-free.
|
|
with self.lock(), pageops.content_write_lock():
|
|
data = self.read()
|
|
self.validate(data)
|
|
pageops.recover_interrupted_content()
|
|
self.save(data)
|
|
imported = skipped = 0
|
|
for index, (tree, tenant, slug, grants) in enumerate(self.plan):
|
|
if data["spaces"][index]["state"] == "completed":
|
|
skipped += 1
|
|
continue
|
|
if tree_digest(tree) != self.source_hashes[index]:
|
|
raise ValueError("Source changed after preflight")
|
|
# Persist uncertainty BEFORE the first import write. Any failure,
|
|
# including a lost commit reply or receipt fsync, stays uncertain.
|
|
space_id = uuid.uuid4()
|
|
data["spaces"][index] = {"state": "started", "space_id": str(space_id)}
|
|
self.save(data)
|
|
report = import_space_tree(tree, tenant_id=tenant, slug=slug,
|
|
memberships=grants, replace=False,
|
|
new_space_id=space_id)
|
|
if tree_digest(tree) != self.source_hashes[index]:
|
|
raise ValueError("Source changed during import; reconciliation required")
|
|
space = Space.objects.get(tenant_id=tenant, slug=slug)
|
|
if space.pk != space_id:
|
|
raise ValueError("Imported space identity changed; reconciliation required")
|
|
observed_grants = sorted((str(m), role) for m, role in
|
|
SpaceMembership.objects.filter(space=space).values_list("member_id", "role"))
|
|
if observed_grants != sorted((str(m.pk), role) for m, role in grants):
|
|
raise ValueError("Imported grants differ from the reviewed mapping")
|
|
sync_tree(pageops.space_dir(slug))
|
|
# A fingerprint records the committed return, not acceptance of
|
|
# source fidelity or authorization to import more cohorts.
|
|
data["spaces"][index] = {
|
|
"state": "completed", "space_id": str(space.pk),
|
|
"files_sha256": tree_digest(pageops.space_dir(slug)),
|
|
"space_sha256": digest(Space.objects.filter(pk=space.pk).values().get()),
|
|
"grants_sha256": digest(observed_grants),
|
|
"pages": report.pages_copied, "assets": report.assets_copied,
|
|
}
|
|
self.save(data)
|
|
imported += 1
|
|
return imported, skipped
|