diff --git a/coulomb_social/apps/spaces/date_backfill_journal.py b/coulomb_social/apps/spaces/date_backfill_journal.py index 5155c63..c5266bf 100644 --- a/coulomb_social/apps/spaces/date_backfill_journal.py +++ b/coulomb_social/apps/spaces/date_backfill_journal.py @@ -139,7 +139,7 @@ def apply_date_patch(root, plan, journal, *, backup_sha256, checkpoint=lambda ph _replace(journal, json.dumps(record, sort_keys=True).encode()) -def reconcile_date_patch(root, journal): +def reconcile_date_patch(root, journal, *, checkpoint=lambda phase: None): """With writers STILL fenced, use committed SQL to reconcile pending bytes. Unknown SQL/file state refuses rather than overwriting a later edit. A lost @@ -161,6 +161,7 @@ def reconcile_date_patch(root, journal): raise ValueError("Journal patch binding changed") with transaction.atomic(durable=True): target = _target(plan) + checkpoint("locked") prior = source_timestamp(plan["observed_sql_created_at"]) desired = source_timestamp(plan["source_created_at"]) _check_pages(index, plan, {_digest(original), _digest(patched)}) @@ -171,6 +172,8 @@ def reconcile_date_patch(root, journal): else: raise ValueError("Unknown SQL state; manual reconciliation required") _replace(index, data) + checkpoint("file") + checkpoint("commit") record["state"] = result _replace(journal, json.dumps(record, sort_keys=True).encode()) return result diff --git a/coulomb_social/apps/spaces/date_backfill_operation.py b/coulomb_social/apps/spaces/date_backfill_operation.py new file mode 100644 index 0000000..e92e4b2 --- /dev/null +++ b/coulomb_social/apps/spaces/date_backfill_operation.py @@ -0,0 +1,243 @@ +"""Application-side checks for the supervised creation-date executor. + +The external supervisor still owns exact admission, pod/image/PVC verification, +fresh native backup/readback, and fencing all writers before and after this call. +Receipt metadata is validated here; this module never contacts the provider. +""" + +from __future__ import annotations + +import base64 +import copy +import fcntl +import hashlib +import json +import os +import re +import stat +from contextlib import contextmanager +from datetime import UTC, datetime +from pathlib import Path + +from django.db import connection + +from .bubble_transfer import source_timestamp +from .date_backfill import prepare_index_date_patch +from .date_backfill_journal import apply_date_patch, reconcile_date_patch + +PREFIX = "platform-pg/application-archives/coulomb-social/" + + +def canonical(data): + return json.dumps(data, sort_keys=True, separators=(",", ":"), default=str).encode() + + +def sha(data): + return hashlib.sha256(data).hexdigest() + + +def validate_backup(receipt, image, recipient, *, now=None, require_fresh=True): + """Check declarations and binding only, not the claimed provider readback.""" + now = now or datetime.now(UTC) + captured = source_timestamp(receipt["captured_at"]) + completed = source_timestamp(receipt["completed_at"]) + if (now.tzinfo is None or captured > completed or completed > now + or (require_fresh and (now - captured).total_seconds() > 600) + or receipt.get("schema") != "platform.application-primary-archive.v1" + or receipt.get("status") != "offsite_fetched_verified" + or receipt.get("application") != "coulomb-social" + or receipt.get("image") != image or receipt.get("age_recipient") != recipient + or not re.fullmatch(r"forgejo\.coulomb\.social/coulomb/" + r"coulomb-social@sha256:[a-f0-9]{64}", image) + or not isinstance(recipient, str) or not recipient.startswith("age1") + or receipt.get("immutable_create") is not True + or receipt.get("remote_retrieval_verified") is not True + or not isinstance(receipt.get("version_id"), str) + or receipt["version_id"] in {"", "null"} + or not isinstance(receipt.get("object_key"), str) + or not receipt["object_key"].startswith(PREFIX) + or receipt.get("destination") != "s3://railiance-platform-pg-backup/" + + receipt["object_key"]): + raise ValueError("Fresh exact application backup metadata required") + for field in ("encrypted_sha256", "ciphertext_sha256", "database_sha256", "content_sha256"): + if not re.fullmatch(r"[a-f0-9]{64}", receipt.get(field, "")): + raise ValueError("Complete paired backup hashes required") + if (receipt["encrypted_sha256"] != receipt["ciphertext_sha256"] + or type(receipt.get("encrypted_bytes")) is not int or receipt["encrypted_bytes"] <= 0 + or type(receipt.get("ciphertext_bytes")) is not int + or receipt["ciphertext_bytes"] != receipt["encrypted_bytes"]): + raise ValueError("Consistent paired ciphertext binding required") + return sha(canonical(receipt)) + + +def _snapshot(root): + """Keep row values in memory only; return fingerprints to the supervisor.""" + root = Path(root).absolute() + if any(p.is_symlink() for p in [root, *root.parents]) or not root.is_dir(): + raise ValueError("Real content directory required") + if connection.vendor == "postgresql": + with connection.cursor() as cursor: + cursor.execute("SELECT current_schema()") + if cursor.fetchone()[0] != "public": + raise ValueError("Expected public application schema required") + cursor.execute("SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY 1") + names = [r[0] for r in cursor.fetchall()] + else: + names = sorted(connection.introspection.table_names()) + rows = {} + with connection.cursor() as cursor: + for table in names: + qualified = ("public." if connection.vendor == "postgresql" else "") + qualified += connection.ops.quote_name(table) + cursor.execute("SELECT * FROM " + qualified) + columns = [c[0] for c in cursor.description] + rows[table] = [dict(zip(columns, row)) for row in cursor.fetchall()] + files = {} + for path in sorted(root.rglob("*")): + mode = path.lstat().st_mode + if stat.S_ISDIR(mode): + continue + if not stat.S_ISREG(mode) or path.stat().st_nlink != 1: + raise ValueError("Content link or special file refused") + if path.relative_to(root).parts[0] == ".recovery": + raise ValueError("Content recovery must be resolved before date correction") + with path.open("rb") as stream: + files[path.relative_to(root).as_posix()] = hashlib.file_digest( + stream, "sha256").hexdigest() + return {"rows": rows, "files": files} + + +def fingerprint(snapshot): + tables = {name: sorted(canonical(row).decode() for row in rows) + for name, rows in snapshot["rows"].items()} + return {"tables_sha256": sha(canonical(tables)), + "files_sha256": sha(canonical(snapshot["files"])), + "tables": len(tables), "rows": sum(len(rows) for rows in tables.values()), + "files": len(snapshot["files"])} + + +def source_fingerprint(root): + return fingerprint(_snapshot(root)) + + +def _lock_tables(): + if connection.vendor != "postgresql": + raise ValueError("PostgreSQL execution required") + if not connection.in_atomic_block: + raise ValueError("SQL locks require a transaction") + with connection.cursor() as cursor: + cursor.execute("SELECT tablename FROM pg_tables WHERE schemaname='public' ORDER BY 1") + tables = [r[0] for r in cursor.fetchall()] + for table in tables: + cursor.execute("LOCK TABLE public." + connection.ops.quote_name(table) + + " IN ACCESS EXCLUSIVE MODE NOWAIT") + + +@contextmanager +def _content_lock(root): + path = Path(root) / ".writer.lock" + if any(p.is_symlink() for p in [path, *path.absolute().parents]): + raise ValueError("Real existing writer lock required") + fd = os.open(path, os.O_RDONLY | os.O_NOFOLLOW) + try: + if not stat.S_ISREG(os.fstat(fd).st_mode) or os.fstat(fd).st_nlink != 1: + raise ValueError("Regular writer lock required") + fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB) + yield + finally: + os.close(fd) + + +def _set_date(snapshot, plan, date): + selected = plan["selection"] + matches = [row for row in snapshot["rows"]["spaces_space"] + if str(row["id"]) == selected["space_id"]] + if (len(matches) != 1 or matches[0]["tenant_id"] != selected["tenant_id"] + or matches[0]["slug"] != selected["space_slug"] + or sum(row["slug"] == selected["space_slug"] + for row in snapshot["rows"]["spaces_space"]) != 1): + raise ValueError("Exact unambiguous SQL identity required") + matches[0]["created_at"] = source_timestamp(date) + + +def execute_date_operation(root, plan, journal, baseline, backup, *, image, recipient, + now=None, reconcile=False, checkpoint=lambda phase: None): + """Verify whole-store pre/postconditions around the durable correction. + + On any error the external supervisor MUST keep writers fenced. Reconciliation + uses the ORIGINAL bound backup even if old; taking a new backup of uncertain + mixed SQL/file state is not a substitute. No automatic fresh retry is allowed. + """ + if connection.vendor != "postgresql": + raise ValueError("PostgreSQL execution required") + binding = validate_backup(backup, image, recipient, now=now, require_fresh=not reconcile) + root, journal = Path(root), Path(journal) + name = f"spaces/{plan['selection']['space_slug']}/pages/index.md" + with _content_lock(root): + if reconcile: + record = json.loads(journal.read_bytes()) + if (record["state"] != "pending" or record["plan"] != plan + or record["backup_receipt_sha256"] != binding): + raise ValueError("Original pending operation binding required") + original = base64.b64decode(record["original"], validate=True) + else: + original = (root / name).read_bytes() + patched = prepare_index_date_patch(plan, original) + before = _snapshot(root) + normalized = copy.deepcopy(before) + _set_date(normalized, plan, plan["observed_sql_created_at"]) + normalized["files"][name] = sha(original) + if fingerprint(normalized) != baseline: + raise ValueError("Whole-store reviewed baseline changed") + expected = copy.deepcopy(normalized) + _set_date(expected, plan, plan["source_created_at"]) + expected["files"][name] = sha(patched) + expected_fingerprint = fingerprint(expected) + + def verify(phase): + if phase in {"intent", "locked"}: + _lock_tables() + current = _snapshot(root) + if reconcile: + # Only the selected date/index may be in either known state. + allowed_dates = {source_timestamp(plan["observed_sql_created_at"]), + source_timestamp(plan["source_created_at"])} + target = next(row for row in current["rows"]["spaces_space"] + if str(row["id"]) == plan["selection"]["space_id"]) + if (target["created_at"] not in allowed_dates + or current["files"][name] not in {sha(original), sha(patched)}): + raise ValueError("Unknown date operation state") + corrected = target["created_at"] == source_timestamp(plan["source_created_at"]) + verified = copy.deepcopy(current) + _set_date(verified, plan, plan["observed_sql_created_at"]) + verified["files"][name] = sha(original) + if fingerprint(verified) != baseline: + raise ValueError("Unrelated rows or files changed") + outcome["fingerprint"] = expected_fingerprint if corrected else baseline + elif fingerprint(current) != baseline: + raise ValueError("Whole-store baseline changed after SQL locks") + elif phase == "sql": + sql_only = copy.deepcopy(expected) + sql_only["files"][name] = sha(original) + if fingerprint(_snapshot(root)) != fingerprint(sql_only): + raise ValueError("Unexpected SQL correction result") + elif phase in {"file", "commit"}: + if fingerprint(_snapshot(root)) != outcome["fingerprint"]: + raise ValueError("Whole-store correction postflight differs") + checkpoint(phase) + + outcome = {"fingerprint": expected_fingerprint} + if reconcile: + result = reconcile_date_patch(root, journal, checkpoint=verify) + else: + if fingerprint(before) != baseline: + raise ValueError("Observed date operation state changed") + apply_date_patch(root, plan, journal, backup_sha256=binding, checkpoint=verify) + result = "completed" + if fingerprint(_snapshot(root)) != outcome["fingerprint"]: + raise ValueError("Final complete-store comparison failed") + return {"status": result, "whole_store_postflight": True, + "backup_metadata_validated": True, "backup_provider_readback_performed_here": False, + "external_writer_fence_verified_here": False, + "all_source_tables_compared": baseline["tables"], + "all_source_files_compared": baseline["files"]} diff --git a/docs/release/source-date-journal-review-2026-10-30.md b/docs/release/source-date-journal-review-2026-10-30.md index d55f83f..f8f48aa 100644 --- a/docs/release/source-date-journal-review-2026-10-30.md +++ b/docs/release/source-date-journal-review-2026-10-30.md @@ -65,9 +65,55 @@ uv run python scripts/rehearse-source-date-journal.py \ --output /tmp/date-journal-real-pair.json ``` +## Complete store operation checks + +`date_backfill_operation.py` now wraps the journal primitive with complete-store +preconditions and postflight. It accepts only PostgreSQL execution in the public +application schema, takes the existing content-writer lock without waiting, and +acquires ACCESS EXCLUSIVE NOWAIT locks on every public table inside the correction +transaction. A busy writer or SQL lock conflict refuses. The reviewed baseline +binds all table rows and all content-file hashes, including unrelated records, +assets and aliases. The expected result is derived by changing only the selected +row's creation date and the byte-preserving index patch. Every SQL/file/commit +boundary is compared against that result; an error requires the supervisor to +keep writers fenced. + +Backup metadata must declare the exact app/image/public recipient, immutable +version, complete paired-store/ciphertext hashes, matching own-prefix destination +and successful readback, with capture/completion ordered and at most ten minutes +old for a new operation. The module validates those declarations and retains +their canonical digest. It does not contact the provider or establish that the +declarations are true; native capture/readback and source-to-backup bookends are +the external supervisor's responsibility. + +Reconciliation requires the original pending plan and backup digest. It permits +that original backup to have aged, reconstructs the reviewed baseline by undoing +only the known date/index change in memory, and refuses any unrelated drift. +It never substitutes a new capture of potentially inconsistent state. The +committed SQL date still determines rollback versus completion. No generic +restore, historical data overwrite or automatic retry is added. + +The extended isolated runner accepts `--backup-receipt` to exercise this exact +operation on a disposable clone of the recovered pair. Its four real process +deaths reconcile correctly, including recovery with the original binding a day +later. SQL drift, an unrelated new file and a busy content writer refuse before +an intent is created. A second real PostgreSQL connection's UPDATE times out +against the operation's table lock. Successful commit and replay refusal preserve +all unrelated rows/files; all 13 tables and 82 files are compared. The retained +recovery source remains unchanged and the clone is removed. Proof: +`source-date-operation-real-pair-2026-10-30.json`. + +This rehearsal uses the historical verified backup receipt and an explicitly +historical metadata-check clock. It proves the operation logic on real recovered +data, not freshness of a newly captured point or a production writer fence. +The candidate image uses explicit source overlays; no image is released here. +The focused planner/journal/operation suite passes 50 cases; `make check` passes +364 cases including protected browsers, with zero skips. + Remaining wrapper requirements: bind exact reviewed source/plan/target and fresh backup/readback hashes; verify the external writer fence and full live SQL/content -baseline; supervise the privileged action; refuse outstanding journal state; +baseline; invoke this operation in the pinned helper; supervise the privileged +action; refuse outstanding journal state; retain the journal on durable private storage; perform full unrelated-row/file postflight and resolve every pending state before resuming writers. Rehearse that wrapper against the independently recovered real pair before productive use. diff --git a/docs/release/source-date-operation-real-pair-2026-10-30.json b/docs/release/source-date-operation-real-pair-2026-10-30.json new file mode 100644 index 0000000..49707df --- /dev/null +++ b/docs/release/source-date-operation-real-pair-2026-10-30.json @@ -0,0 +1,70 @@ +{ + "schema": "coulomb.recovered-date-journal-rehearsal.v1", + "status": "guarded_recovered_date_operation_verified", + "phases": [ + { + "phase": "intent", + "child_exit": 73, + "reconciled": "rolled_back", + "all_tables_and_files_match_expected": true + }, + { + "phase": "sql", + "child_exit": 73, + "reconciled": "rolled_back", + "all_tables_and_files_match_expected": true + }, + { + "phase": "file", + "child_exit": 73, + "reconciled": "rolled_back", + "all_tables_and_files_match_expected": true + }, + { + "phase": "commit", + "child_exit": 73, + "reconciled": "completed", + "all_tables_and_files_match_expected": true + } + ], + "successful_commit_verified": true, + "replay_refused": true, + "only_selected_sql_date_and_index_changed": true, + "all_tables_compared": 13, + "all_files_compared": 82, + "grants_and_local_edits_preserved": true, + "whole_store_drift_refused": true, + "busy_content_writer_refused": true, + "concurrent_sql_writer_blocked": true, + "recovery_retains_original_old_backup_binding": true, + "historical_backup_metadata_clock_used": true, + "fresh_native_backup_or_external_fence_proved": false, + "candidate_image_id": "sha256:512819463a1902b43a615beee14be7bf97e6a05e016385082b9380692ad1a8cd", + "source_overlay_used": true, + "modules_sha256": { + "date_backfill.py": "84ce5d3369389e1af9ab999aad3804456fd2f66e654e290496be6930cd4885e4", + "date_backfill_journal.py": "47af997c02b3910058e5dc8c593dd8135a68be94e909c672a946f54601eb621e", + "date_backfill_operation.py": "50f962f6c42fd47dbbbdea3f06d38ccfef063e784bdb6361b8a2325cbe825e18" + }, + "runner_sha256": "3cab58a736658422b99b582ce69f935b34ae648edb59f559e72c739e73e261b9", + "shared_runner_sha256": "812fc4b6325ea1b4c9c3698ad1175fe5a95d77ce5c1dc1631174f46e93bbb956", + "private_plan_sha256": "44369d5abbf92d3e6d52e909cdc1e1c3a90b8817be7027c3a43846fbfc381ef8", + "restore_receipt_sha256": "5f745cdae0a34c4a98f0e4429ceea49475b4e5cd9d2f8f09dea6be000e200258", + "historical_backup_receipt_sha256": "308421ffb233adf6f67c4554fc3ed8b67e8bad2f4345d3da303daa705a05c663", + "network_internal": true, + "published_ports": 0, + "original_recovered_sql_and_files_unchanged": true, + "disposable_sql_clone_removed": true, + "production_access": false, + "production_writes": false, + "supervised_live_executor_prepared": false, + "full_gate": { + "command": "make check", + "tests": 364, + "failures": 0, + "errors": 0, + "skipped": 0 + }, + "focused_cases_in_full_gate": 50, + "operation_tests_sha256": "1b48e485ad3d31499b20811dff7147e34633174505b8f79f2d6a3d71d7bfdc34" +} diff --git a/scripts/rehearse-source-date-journal.py b/scripts/rehearse-source-date-journal.py index 77e0f4c..2caaf8b 100644 --- a/scripts/rehearse-source-date-journal.py +++ b/scripts/rehearse-source-date-journal.py @@ -93,6 +93,136 @@ print(json.dumps({"status": "recovered_pair_journal_verified", "phases": phases, "grants_and_local_edits_preserved": True})) ''' +GUARDED_CHECKS = COMMON + r''' +import os, subprocess, sys, psycopg +from datetime import timedelta +from coulomb_social.apps.spaces.date_backfill_operation import ( + execute_date_operation, source_fingerprint, _content_lock) + +backup = json.loads(Path("/review/backup.json").read_bytes()) +backup = backup.get("receipt", backup) +# A historical metadata-check clock, never evidence of a new live backup. +review_now = source_timestamp(backup["completed_at"]) + timedelta(seconds=1) +baseline = source_fingerprint(root) +Path("/journal/baseline.json").write_text(json.dumps(baseline)) +before_files, before = files(), database() +prior = source_timestamp(plan["observed_sql_created_at"]) +desired = source_timestamp(plan["source_created_at"]) +assert prior != desired +expected = copy.deepcopy(before) +target = next(r for r in expected["spaces_space"] if str(r["id"]) == selection["space_id"]) +assert target["created_at"] == prior +target["created_at"] = desired +expected_files = dict(before_files) +expected_files[str(index.relative_to(root))] = hashlib.sha256(patched).hexdigest() + +def execute(journal, **options): + return execute_date_operation(root, plan, journal, baseline, backup, + image=backup["image"], recipient=backup["age_recipient"], now=review_now, **options) + +child = """ +import django, json, os +from datetime import timedelta +from pathlib import Path +django.setup() +from coulomb_social.apps.spaces.bubble_transfer import source_timestamp +from coulomb_social.apps.spaces.date_backfill_operation import execute_date_operation +backup = json.loads(Path('/review/backup.json').read_bytes()) +backup = backup.get('receipt', backup) +execute_date_operation(Path('/rehearsal-content'), + json.loads(Path('/review/plan.json').read_bytes()), Path(os.environ['JOURNAL']), + json.loads(Path('/journal/baseline.json').read_bytes()), backup, + image=backup['image'], recipient=backup['age_recipient'], + now=source_timestamp(backup['completed_at']) + timedelta(seconds=1), + checkpoint=lambda phase: os._exit(73) if phase == os.environ['CRASH_PHASE'] else None) +""" +phases = [] +for phase in ["intent", "sql", "file", "commit"]: + journal = Path("/journal") / (phase + ".json") + result = subprocess.run([sys.executable, "-c", child], capture_output=True, timeout=45, + env={**os.environ, "JOURNAL": str(journal), "CRASH_PHASE": phase}) + assert result.returncode == 73 + assert json.loads(journal.read_bytes())["state"] == "pending" + # Recovery binds the original point, even a day later. It never captures a + # replacement point from an uncertain combination of SQL and file state. + proof = execute_date_operation(root, plan, journal, baseline, backup, + image=backup["image"], recipient=backup["age_recipient"], + now=review_now + timedelta(days=1), reconcile=True) + committed = phase == "commit" + assert proof["status"] == ("completed" if committed else "rolled_back") + assert proof["whole_store_postflight"] + assert digest(database()) == digest(expected if committed else before) + assert files() == (expected_files if committed else before_files) + phases.append({"phase": phase, "child_exit": 73, "reconciled": proof["status"], + "all_tables_and_files_match_expected": True}) + if committed: + assert Space.objects.filter(pk=selection["space_id"], created_at=desired).update( + created_at=prior) == 1 + index.write_bytes(original) + assert digest(database()) == digest(before) and files() == before_files + +for kind in ["sql", "file", "writer"]: + journal = Path("/journal") / ("refused-" + kind + ".json") + old_title = Space.objects.get(pk=selection["space_id"]).title + stray = root / "unrelated-guard-fixture.bin" + try: + if kind == "sql": + Space.objects.filter(pk=selection["space_id"]).update(title="synthetic drift") + elif kind == "file": + stray.write_bytes(b"synthetic unrelated drift") + try: + if kind == "writer": + with _content_lock(root): + execute(journal) + else: + execute(journal) + except (ValueError, BlockingIOError): + pass + else: + raise AssertionError("whole-store drift or busy writer unexpectedly accepted") + assert not journal.exists() + finally: + if kind == "sql": + Space.objects.filter(pk=selection["space_id"]).update(title=old_title) + stray.unlink(missing_ok=True) + assert digest(database()) == digest(before) and files() == before_files + +lock_proved = [] +def probe_lock(phase): + if phase == "sql": + with psycopg.connect(os.environ['DATABASE_URL']) as other: + with other.cursor() as cursor: + cursor.execute("SET lock_timeout='100ms'") + try: + cursor.execute("UPDATE spaces_space SET title=title WHERE id=%s", + (selection["space_id"],)) + except psycopg.errors.LockNotAvailable: + lock_proved.append(True) + else: + raise AssertionError("unrelated SQL writer was not blocked") + +final = Path("/journal/success.json") +proof = execute(final, checkpoint=probe_lock) +assert lock_proved == [True] +assert digest(database()) == digest(expected) and files() == expected_files +try: + execute(final) +except (ValueError, FileExistsError): + pass +else: + raise AssertionError("guarded operation replay unexpectedly accepted") +assert digest(database()) == digest(expected) and files() == expected_files +print(json.dumps({"status": "guarded_recovered_date_operation_verified", "phases": phases, + "successful_commit_verified": True, "replay_refused": True, + "only_selected_sql_date_and_index_changed": True, + "all_tables_compared": len(before), "all_files_compared": len(before_files), + "grants_and_local_edits_preserved": True, "whole_store_drift_refused": True, + "busy_content_writer_refused": True, "concurrent_sql_writer_blocked": True, + "recovery_retains_original_old_backup_binding": True, + "historical_backup_metadata_clock_used": True, + "fresh_native_backup_or_external_fence_proved": False})) +''' + def sha(data): return hashlib.sha256(data).hexdigest() @@ -103,6 +233,8 @@ def main(): parser.add_argument("--restore-receipt", type=Path, required=True) parser.add_argument("--plan", type=Path, required=True) parser.add_argument("--candidate-image", required=True) + parser.add_argument("--backup-receipt", type=Path, + help="Enable whole-store operation checks with historical metadata clock") parser.add_argument("--output", type=Path, required=True) args = parser.parse_args() if args.output.exists() or args.output.is_symlink(): @@ -142,12 +274,18 @@ def main(): repo = Path(__file__).resolve().parents[1] modules = {name: (repo / "coulomb_social/apps/spaces" / name).read_bytes() for name in ["date_backfill.py", "date_backfill_journal.py"]} + if args.backup_receipt: + modules["date_backfill_operation.py"] = ( + repo / "coulomb_social/apps/spaces/date_backfill_operation.py").read_bytes() + backup_bytes = args.backup_receipt.read_bytes() if args.backup_receipt else None plan_bytes = args.plan.read_bytes() clone = "csoc_date_" + uuid.uuid4().hex with tempfile.TemporaryDirectory(prefix="csoc-date-journal-rehearsal-") as temporary: work = Path(temporary) shutil.copytree(root, work / "content") (work / "plan.json").write_bytes(plan_bytes) + if backup_bytes: + (work / "backup.json").write_bytes(backup_bytes) (work / "journal").mkdir(mode=0o700) for name, data in modules.items(): (work / name).write_bytes(data) @@ -167,6 +305,8 @@ def main(): "-v", str(work / "content") + ":/rehearsal-content", "-v", str(work / "plan.json") + ":/review/plan.json:ro", "-v", str(work / "journal") + ":/journal"] + if backup_bytes: + command += ["-v", str(work / "backup.json") + ":/review/backup.json:ro"] for member in modules: command += ["-v", str(work / member) + ":/app/coulomb_social/apps/spaces/" + member + ":ro"] @@ -194,7 +334,7 @@ def main(): run(["docker", "exec", state["database_container"], "createdb", "-U", "postgres", "-T", "coulomb_social_restore", clone]) try: - verdict = application(clone, CHECKS) + verdict = application(clone, GUARDED_CHECKS if backup_bytes else CHECKS) after = application("coulomb_social_restore", SNAPSHOT) # SNAPSHOT reads the disposable content copy; the source DB itself # must be identical. Original files are checked separately below. @@ -213,6 +353,7 @@ def main(): "shared_runner_sha256": sha(Path(__file__).with_name( "rehearse-source-date-backfill.py").read_bytes()), "private_plan_sha256": sha(plan_bytes), "restore_receipt_sha256": sha(raw_receipt), + "historical_backup_receipt_sha256": sha(backup_bytes) if backup_bytes else None, "network_internal": True, "published_ports": 0, "original_recovered_sql_and_files_unchanged": True, "disposable_sql_clone_removed": True, diff --git a/tests/test_date_backfill_operation.py b/tests/test_date_backfill_operation.py new file mode 100644 index 0000000..3d45337 --- /dev/null +++ b/tests/test_date_backfill_operation.py @@ -0,0 +1,105 @@ +"""Receipt guards and full-store fingerprints use synthetic records only.""" + +import copy +from datetime import UTC, datetime, timedelta + +import pytest + +from coulomb_social.apps.spaces.date_backfill_operation import ( + _content_lock, + _set_date, + execute_date_operation, + source_fingerprint, + validate_backup, +) +from coulomb_social.apps.spaces.models import Space + +IMAGE = "forgejo.coulomb.social/coulomb/coulomb-social@sha256:" + "a" * 64 +RECIPIENT = "age1synthetic-public-recipient" +NOW = datetime(2026, 10, 30, 12, 5, tzinfo=UTC) + + +def backup(): + key = "platform-pg/application-archives/coulomb-social/synthetic.tar.age" + return { + "schema": "platform.application-primary-archive.v1", "application": "coulomb-social", + "status": "offsite_fetched_verified", "captured_at": "2026-10-30T12:00:00Z", + "completed_at": "2026-10-30T12:01:00Z", "image": IMAGE, "age_recipient": RECIPIENT, + "immutable_create": True, "remote_retrieval_verified": True, "version_id": "synthetic-v1", + "object_key": key, "destination": "s3://railiance-platform-pg-backup/" + key, + "encrypted_sha256": "b" * 64, "ciphertext_sha256": "b" * 64, + "database_sha256": "c" * 64, "content_sha256": "d" * 64, + "encrypted_bytes": 1234, "ciphertext_bytes": 1234, + } + + +@pytest.mark.parametrize("change", [ + {"status": "uploaded"}, {"remote_retrieval_verified": False}, {"immutable_create": False}, + {"version_id": "null"}, {"image": IMAGE + "0"}, {"age_recipient": "age1other"}, + {"object_key": "platform-pg/application-archives/another/synthetic.tar.age"}, + {"destination": "s3://another/synthetic"}, {"ciphertext_sha256": "e" * 64}, + {"encrypted_bytes": True}, {"ciphertext_bytes": True}, {"database_sha256": "missing"}, + {"captured_at": "2026-10-30T11:00:00Z"}, {"completed_at": "2026-10-31T12:00:00Z"}, + {"captured_at": "2026-10-30T12:02:00Z"}, +]) +def test_stale_unverified_or_changed_backup_metadata_refuses(change): + receipt = {**backup(), **change} + with pytest.raises(ValueError): + validate_backup(receipt, IMAGE, RECIPIENT, now=NOW) + + +def test_backup_binding_is_stable_and_recovery_retains_original_even_when_old(): + receipt = backup() + original = copy.deepcopy(receipt) + digest = validate_backup(receipt, IMAGE, RECIPIENT, now=NOW) + reordered = dict(reversed(list(receipt.items()))) + assert validate_backup(reordered, IMAGE, RECIPIENT, now=NOW) == digest + assert receipt == original + with pytest.raises(ValueError): + validate_backup(receipt, IMAGE, RECIPIENT, now=NOW + timedelta(days=1)) + assert validate_backup(receipt, IMAGE, RECIPIENT, now=NOW + timedelta(days=1), + require_fresh=False) == digest + + +def test_legacy_slug_collision_refuses_even_when_selected_sql_identity_is_exact(): + selected = {"space_id": "selected", "tenant_id": "tenant:a", "space_slug": "same"} + snapshot = {"rows": {"spaces_space": [ + {"id": "selected", "tenant_id": "tenant:a", "slug": "same", "created_at": NOW}, + {"id": "another", "tenant_id": "tenant:b", "slug": "same", "created_at": NOW}, + ]}} + before = copy.deepcopy(snapshot) + with pytest.raises(ValueError, match="unambiguous"): + _set_date(snapshot, {"selection": selected}, "2020-01-01T00:00:00Z") + assert snapshot == before + + +@pytest.mark.django_db +def test_fingerprints_cover_other_tables_and_all_content_not_only_selected_pages(tmp_path): + Space.objects.create(tenant_id="tenant:synthetic", slug="synthetic") + baseline = source_fingerprint(tmp_path) + assert baseline["tables"] >= 4 + (tmp_path / "arbitrary-unrelated-asset.bin").write_bytes(b"unrelated") + after_file = source_fingerprint(tmp_path) + assert after_file["tables_sha256"] == baseline["tables_sha256"] + assert after_file["files_sha256"] != baseline["files_sha256"] + Space.objects.update(title="another update") + after_sql = source_fingerprint(tmp_path) + assert after_sql["tables_sha256"] != baseline["tables_sha256"] + assert after_sql["files_sha256"] == after_file["files_sha256"] + + +@pytest.mark.django_db +def test_sqlite_is_observation_only_never_an_execution_backend(tmp_path): + with pytest.raises(ValueError, match="PostgreSQL execution required"): + execute_date_operation(tmp_path, {}, tmp_path / "journal.json", {}, {}, + image=IMAGE, recipient=RECIPIENT) + assert not (tmp_path / "journal.json").exists() + + +def test_writer_lock_refuses_another_holder_and_never_creates_missing_lock(tmp_path): + with pytest.raises(FileNotFoundError), _content_lock(tmp_path): + pass + assert not list(tmp_path.iterdir()) + (tmp_path / ".writer.lock").touch() + with _content_lock(tmp_path), pytest.raises(BlockingIOError), _content_lock(tmp_path): + pass diff --git a/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md b/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md index fa6dd99..6135d85 100644 --- a/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md +++ b/workplans/CSOC-WP-0014-cohort-transfer-bubble-offramp.md @@ -74,6 +74,13 @@ uncaptured article references. Targeted continuation completed 57 visits, captur All four prior per-space graph exceptions persist. Non-secret counts/hashes: docs/release/bubble-offline-preparation-2026-10-28.json. +2026-10-30 source recheck: the retained targeted reference capture is terminal +(`completed_reference_pass`), with 244 articles and five remaining references. +The saved Bubble browser state fails its local usability check; no new browser +capture or server-session acceptance is claimed. Requested renewed attended login +through the existing capture-auth-state.sh workflow; original private checkpoints +remain untouched. No guessed body, date, grant or source exception was substituted. + ## T02 — Review remaining journeys and record transfer decisions ```task @@ -188,6 +195,21 @@ explicit; no new image, productive date correction or admission is inferred. The journal's real-pair rehearsal is complete; the external supervised wrapper, fresh verified backup/fence and productive postflight are still required. +2026-10-30 whole-store executor follow-through: added date_backfill_operation.py +around the journal core. It validates exact fresh backup metadata, binds all SQL +rows/content-file hashes, takes the existing content lock and all-public-table +PostgreSQL exclusive locks, and checks the only permitted date/index delta at +SQL/file/commit boundaries. Reconciliation retains the original backup binding +even when old and refuses unrelated drift. Extended native recovered-pair tests +prove four real process deaths, original-binding recovery a day later, SQL/file/ +busy-writer refusal, blocking of a second PostgreSQL writer, successful commit and +replay refusal across all13tables/82files. Source remains unchanged; clone removed. +Proof: docs/release/source-date-operation-real-pair-2026-10-30.json. The historical +metadata clock is explicit; no fresh native backup, external fence, release or +productive correction is implied.50 focused and364 full gate cases pass, zero skips. +The exact supervised node wrapper, live backup/source bookends, pinned helper/PVC +admission, durable journal placement and operator postflight/resume remain open. + ## Completion Finish when every task above has its stated evidence and acceptance, or hand off any newly discovered residual to a live work record first. Cross-owner dependencies remain their existing canonical records.