Define revision purge and copy/move crash recovery
Some checks failed
Use-case regression gate / usecase-regressions (push) Has been cancelled

Page saves keep every revision. An explicit purge drops only inactive
revisions older than a timezone-aware cutoff, keeps live bytes and deleted
page history, and stops if a stored hash does not match. Interrupted copies
and moves resume from a journal without deleting a changed source or granting
destination access. Private corpus review remains open.

Assistant: grok
Assistant-Session: 01a12cc2-e3b2-77a3-8aef-9cc60e577fee
This commit is contained in:
tegwick 2026-10-11 23:24:23 +02:00
parent 5121ae0aad
commit f94ef6a9d1
6 changed files with 565 additions and 28 deletions

View file

@ -88,6 +88,13 @@ def page_path_for(space: Space, page_slug: str) -> str:
def render_markdown(text: str) -> str:
"""Render the supported Markdown subset.
Extensions are fenced code, tables, newline-to-break and sane lists.
Raw HTML, script, event handlers and non-http(s)/mailto URLs are removed.
Relative asset URLs are kept so copied page images still resolve. This is
not acceptance of a private corpus or of arbitrary Bubble markup.
"""
raw_html = md_lib.markdown(
text,
extensions=["fenced_code", "tables", "nl2br", "sane_lists"],

View file

@ -21,8 +21,10 @@ import re
import shutil
import tempfile
import threading
import uuid
from contextlib import contextmanager
from dataclasses import dataclass, field
from datetime import UTC, datetime
from functools import wraps
from pathlib import Path
from urllib.parse import quote, unquote, urlsplit
@ -42,7 +44,7 @@ def content_write_lock():
"""Serialize cooperating app/import workers on one mounted content instance."""
with _WRITER_LOCK:
if getattr(_WRITER_STATE, "active", False):
yield
yield False
return
root = content_root()
root.mkdir(parents=True, exist_ok=True)
@ -50,16 +52,20 @@ def content_write_lock():
try:
fcntl.flock(fd, fcntl.LOCK_EX)
_WRITER_STATE.active = True
yield
try:
yield True
finally:
_WRITER_STATE.active = False
finally:
os.close(fd)
def content_writer(function):
@wraps(function)
def locked(*args, **kwargs):
with content_write_lock():
with content_write_lock() as outermost:
if outermost and function.__name__ != "recover_interrupted_content":
_recover_interrupted_content()
return function(*args, **kwargs)
return locked
@ -251,14 +257,8 @@ def page_aliases(space_slug):
raise PageOpsError("Page-address history needs operator review") from exc
@content_writer
def remember_page_address(source_space, source_page, dest_space, dest_page):
source_page = normalize_slug(source_page)
dest_page = normalize_slug(dest_page)
space_dir(dest_space)
aliases = page_aliases(source_space)
aliases[source_page] = [dest_space, dest_page]
directory = space_dir(source_space)
def _store_aliases(space_slug, aliases):
directory = space_dir(space_slug)
temporary = None
try:
with tempfile.NamedTemporaryFile(dir=directory, delete=False) as stream:
@ -273,6 +273,295 @@ def remember_page_address(source_space, source_page, dest_space, dest_page):
temporary.unlink(missing_ok=True)
@content_writer
def remember_page_address(source_space, source_page, dest_space, dest_page):
source_page = normalize_slug(source_page)
dest_page = normalize_slug(dest_page)
space_dir(dest_space)
aliases = page_aliases(source_space)
aliases[source_page] = [dest_space, dest_page]
_store_aliases(source_space, aliases)
def _recovery_dir():
path = content_root() / ".recovery"
if path.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
path.mkdir(parents=True, exist_ok=True)
return path
def _write_record(record):
directory = _recovery_dir()
destination = directory / f"{record['id']}.json"
if destination.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
temporary = None
try:
with tempfile.NamedTemporaryFile(dir=directory, delete=False) as stream:
temporary = Path(stream.name)
payload = (json.dumps(record, sort_keys=True) + "\n").encode("utf-8")
stream.write(payload)
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, destination)
_fsync_directory(directory)
finally:
if temporary is not None:
temporary.unlink(missing_ok=True)
return record
def _read_record(record_id):
path = _recovery_dir() / f"{record_id}.json"
if path.is_symlink() or not path.is_file():
return None
try:
record = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError):
return "held"
if not isinstance(record, dict) or record.get("id") != record_id:
return "held"
return record
def _clear_record(record):
path = _recovery_dir() / f"{record['id']}.json"
if path.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
path.unlink(missing_ok=True)
_fsync_directory(path.parent)
def _start_record(op, source_space, source_page, dest_space, dest_page, source_revision):
return _write_record({
"id": uuid.uuid4().hex,
"version": 1,
"op": op,
"source_space": source_space,
"source_page": source_page,
"dest_space": dest_space,
"dest_page": dest_page,
"source_revision": source_revision,
"dest_revision": "",
"assets": [],
"phase": "started",
})
def _note_asset(record, name):
if not isinstance(name, str) or not name or "/" in name or name in (".", ".."):
raise PageOpsError("Recovery asset name needs operator review")
if name not in record["assets"]:
record["assets"].append(name)
_write_record(record)
def _file_revision(path):
return hashlib.sha256(path.read_bytes()).hexdigest()
def _discard_recorded_assets(space_slug, names):
if not names:
return
root = assets_dir(space_slug)
for name in names:
if not isinstance(name, str) or not name or "/" in name or name in (".", ".."):
raise PageOpsError("Recovery asset name needs operator review")
path = root / name
if path.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
if path.is_file():
path.unlink()
_fsync_directory(root)
def _discard_unconfirmed_page(space_slug, page_slug):
path = page_file(space_slug, page_slug)
if path.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
if not path.is_file():
return
_remember(space_slug, page_slug, path.read_bytes())
path.unlink()
_fsync_directory(path.parent)
def _drop_matching_alias(record):
aliases = page_aliases(record["source_space"])
expected = [record["dest_space"], record["dest_page"]]
if aliases.get(record["source_page"]) != expected:
return
aliases.pop(record["source_page"])
_store_aliases(record["source_space"], aliases)
def _record_is_actionable(record):
if not isinstance(record, dict) or record.get("version") != 1:
return False
if record.get("op") not in {"copy", "transfer"}:
return False
if record.get("phase") not in {"started", "published", "aliased"}:
return False
try:
for key in ("source_space", "dest_space"):
if not _SLUG_RE.fullmatch(record[key]):
return False
for key in ("source_page", "dest_page"):
normalize_slug(record[key])
except (KeyError, TypeError, PageOpsError):
return False
if record["dest_page"] == "index":
return False
if record["op"] == "transfer" and record["source_page"] == "index":
return False
if not isinstance(record.get("assets"), list):
return False
return isinstance(record.get("source_revision"), str)
def _finish_transfer_record(record, source_path, dest_confirmed):
"""Complete a published move, or keep a source that changed after it started."""
source_same = source_path.is_file() and _file_revision(source_path) == record["source_revision"]
if source_same and dest_confirmed:
if record["phase"] != "aliased":
remember_page_address(
record["source_space"], record["source_page"],
record["dest_space"], record["dest_page"],
)
if _file_revision(source_path) != record["source_revision"]:
return "held"
delete_page(record["source_space"], record["source_page"])
_clear_record(record)
return "cleared"
if source_path.is_file() and not source_same:
_drop_matching_alias(record)
_clear_record(record)
return "cleared"
if dest_confirmed or page_file(record["dest_space"], record["dest_page"]).is_file():
_clear_record(record)
return "cleared"
_discard_recorded_assets(record["dest_space"], record["assets"])
_clear_record(record)
return "cleared"
def _recover_record(record):
if record == "held" or not _record_is_actionable(record):
logger.error("content recovery left a journal for operator review")
return "held"
source_path = page_file(record["source_space"], record["source_page"])
dest_path = page_file(record["dest_space"], record["dest_page"])
dest_confirmed = bool(
record.get("dest_revision")
and dest_path.is_file()
and _file_revision(dest_path) == record["dest_revision"]
)
source_exists = source_path.is_file()
if record["op"] == "copy":
if dest_confirmed or (not source_exists and dest_path.is_file()):
_clear_record(record)
return "cleared"
if source_exists:
_discard_unconfirmed_page(record["dest_space"], record["dest_page"])
_discard_recorded_assets(record["dest_space"], record["assets"])
_clear_record(record)
return "cleared"
if dest_confirmed:
return _finish_transfer_record(record, source_path, dest_confirmed)
if record["phase"] == "aliased" and source_exists:
_drop_matching_alias(record)
if source_exists:
_discard_unconfirmed_page(record["dest_space"], record["dest_page"])
_discard_recorded_assets(record["dest_space"], record["assets"])
_clear_record(record)
return "cleared"
if dest_path.is_file():
_clear_record(record)
return "cleared"
_discard_recorded_assets(record["dest_space"], record["assets"])
_clear_record(record)
return "cleared"
def _recover_interrupted_content():
directory = content_root() / ".recovery"
if not directory.exists():
return {"cleared": 0, "held": 0}
if directory.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
cleared = held = 0
for path in sorted(directory.glob("*.json")):
if path.is_symlink():
held += 1
logger.error("content recovery left a journal for operator review")
continue
try:
record = json.loads(path.read_text(encoding="utf-8"))
except (OSError, ValueError, TypeError):
held += 1
logger.error("content recovery left a journal for operator review")
continue
if _recover_record(record) == "held":
held += 1
else:
cleared += 1
return {"cleared": cleared, "held": held}
@content_writer
def recover_interrupted_content():
"""Finish or roll back a copy/move whose process stopped between files.
A confirmed destination is kept. The source is removed only when its bytes
still match the journal. Address aliases are written or removed only for
that journal's source page, and they do not grant destination access.
"""
return _recover_interrupted_content()
@content_writer
def purge_revisions(*, older_than):
"""Delete expired inactive revisions. Never runs as part of a page save.
The live page bytes are kept. Every stored revision of a page that no
longer has a file is kept, because that directory is its recovery copy.
A revision whose bytes do not match its content-address name stops the
purge before anything is deleted.
"""
if older_than.tzinfo is None:
raise PageOpsError("Revision purge cutoff must include a timezone")
root = content_root() / ".revisions"
if not root.exists():
return 0
if root.is_symlink():
raise PageOpsError("Content paths must not be symlinks")
victims = []
for path in sorted(root.rglob("*.md")):
if path.is_symlink() or any(parent.is_symlink() for parent in path.parents):
raise PageOpsError("Content paths must not be symlinks")
relative = path.relative_to(root)
if len(relative.parts) != 3:
raise PageOpsError("Stored revision integrity check failed")
data = path.read_bytes()
digest = hashlib.sha256(data).hexdigest()
if path.stem != digest:
raise PageOpsError("Stored revision integrity check failed")
space_slug, page_slug, _name = relative.parts
live = page_file(space_slug, page_slug)
if live.is_file() and _file_revision(live) == digest:
continue
if not live.is_file():
continue
modified = datetime.fromtimestamp(path.stat().st_mtime, tz=UTC)
if modified < older_than:
victims.append(path)
for path in victims:
path.unlink()
_fsync_directory(path.parent)
return len(victims)
@content_writer
def write_page(
space_slug: str,
@ -371,7 +660,7 @@ _BODY_ASSET = re.compile(
)
def _copy_assets(source_space, dest_space, dest_page, visual, body):
def _copy_assets(source_space, dest_space, dest_page, visual, body, on_created=None):
"""Copy local Visual and Markdown link/image destinations as one staged set."""
from .assets import asset_parts, open_local_asset
@ -408,6 +697,8 @@ def _copy_assets(source_space, dest_space, dest_page, visual, body):
stream.flush()
os.fsync(stream.fileno())
copied[path] = f"assets/{quote(dest_name)}"
if on_created is not None:
on_created(dest_name)
suffix = (f"?{parsed.query}" if parsed.query else "")
suffix += f"#{parsed.fragment}" if parsed.fragment else ""
return copied[path] + suffix
@ -437,6 +728,8 @@ def copy_page(
dest_page: str,
*,
title: str | None = None,
journal: bool = True,
on_asset=None,
) -> PageDoc:
src = read_page(source_space, source_page)
if not src.exists:
@ -446,20 +739,42 @@ def copy_page(
or dest_page in page_aliases(dest_space)):
raise PageOpsError("Destination page already exists")
ensure_space_tree(dest_space)
source_revision = _file_revision(Path(src.path))
record = _start_record(
"copy", source_space, src.slug, dest_space, dest_page, source_revision,
) if journal else None
created = []
def note_asset(name):
if record is not None:
_note_asset(record, name)
if on_asset is not None:
on_asset(name)
try:
visual, body, created = _copy_assets(
source_space, dest_space, dest_page, src.visual, src.body,
source_space, dest_space, dest_page, src.visual, src.body, note_asset,
)
new_title = title if title is not None else (src.title or dest_page)
if title is None and new_title == src.title and source_space == dest_space:
new_title = f"{src.title} (copy)" if src.title else f"{dest_page}"
try:
return write_page(
doc = write_page(
dest_space, dest_page, title=new_title, abstractor=src.abstractor,
visual=visual, body=body, expected_revision="",
extra_meta={k: v for k, v in src.meta.items()
if k not in {"title", "abstractor", "visual", "space"}},
)
if record is not None:
record["dest_revision"] = doc.revision
record["phase"] = "published"
_write_record(record)
_clear_record(record)
return doc
except BaseException:
if record is not None:
stored = _read_record(record["id"])
if stored is not None:
_recover_record(stored)
if not page_file(dest_space, dest_page).exists():
for path in created:
path.unlink(missing_ok=True)
@ -478,10 +793,39 @@ def transfer_page(
if source_page == "index":
raise PageOpsError("Cannot transfer the space index page")
src = read_page(source_space, source_page)
copy_page(source_space, source_page, dest_space, dest_page, title=src.title)
if not src.exists:
raise PageOpsError("Source page not found")
# The journal is authority to remove an unconfirmed destination. Refuse an
# existing page before writing that journal.
if (page_file(dest_space, dest_page).is_file()
or dest_page in page_aliases(dest_space)):
raise PageOpsError("Destination page already exists")
record = _start_record(
"transfer", source_space, source_page, dest_space, dest_page,
_file_revision(Path(src.path)),
)
try:
doc = copy_page(
source_space, source_page, dest_space, dest_page, title=src.title,
journal=False, on_asset=lambda name: _note_asset(record, name),
)
record["dest_revision"] = doc.revision
record["phase"] = "published"
_write_record(record)
remember_page_address(source_space, source_page, dest_space, dest_page)
record["phase"] = "aliased"
_write_record(record)
source_path = page_file(source_space, source_page)
if _file_revision(source_path) != record["source_revision"]:
raise PageOpsError("Source changed before the move finished")
delete_page(source_space, source_page)
_clear_record(record)
return read_page(dest_space, dest_page)
except BaseException:
stored = _read_record(record["id"])
if stored is not None:
_recover_record(stored)
raise
@content_writer

View file

@ -6,9 +6,10 @@ superseded. Continue work in the registered residual owners: WP-0012 recovery/
service readiness, WP-0013 sharing/content/portability, WP-0014 further cohorts/
Bubble off-ramp, and WP-0015 communication operations/flow acceptance.
2026-10-11: WP-0013 is active. T01 has the global slug constraint and migration
refusal in the checkout; the live database is not migrated. T02 and T03 remain
todo. WP-0012-T01 and WP-0015-T01/T03 still wait on their named owners.
2026-10-11: WP-0013 is active. T01 has the global slug constraint in the checkout;
the live database is not migrated. T02 has revision purge and copy/move crash
recovery in the checkout; private corpus review and any live purge remain open.
T03 remains todo. WP-0012-T01 and WP-0015-T01/T03 still wait on their named owners.
UC-008 weekly calls and UC-009 personal recovery are accepted. App revision16
and receiver revision7 are deployed; the receiver uses its existing single-caller

View file

@ -16,6 +16,21 @@ def test_render_markdown_strips_script():
assert "Hi" in html
def test_supported_markdown_keeps_local_assets_and_drops_raw_html():
html = render_markdown(
"```\nkept\n```\n\n"
"![Diagram](assets/diagram.png)\n\n"
"<script>alert(1)</script>\n"
"<div onclick=\"alert(1)\">raw</div>\n\n"
"[bad](javascript:alert(1))\n"
)
assert "<script>" not in html
assert "onclick" not in html
assert "javascript:" not in html
assert "assets/diagram.png" in html
assert "<pre>" in html and "kept" in html
@pytest.mark.django_db
def test_load_space_page_unbound(tmp_path, settings):
settings.CONTENT_ROOT = str(tmp_path / "empty-content")

View file

@ -0,0 +1,158 @@
"""Revision retention and crash recovery for copy/move. Synthetic content only."""
import os
from datetime import UTC, datetime, timedelta
from pathlib import Path
import pytest
from django.urls import reverse
from coulomb_social.apps.spaces import pageops
from coulomb_social.apps.spaces.services import create_space, set_space_access
from tests.test_space_permissions import make_member
@pytest.fixture
def trees(tmp_path, settings):
settings.CONTENT_ROOT = str(tmp_path / "content")
for slug in ("source", "destination"):
pageops.init_space_content(slug, title=slug)
return tmp_path / "content"
def _notes(body="Retained content"):
asset = pageops.assets_dir("source") / "diagram.png"
asset.write_bytes(b"synthetic image")
return pageops.write_page(
"source", "notes", title="Notes", body=f"![Diagram](assets/diagram.png)\n\n{body}\n",
)
def test_restart_drops_unconfirmed_destination_and_keeps_source(trees):
source = _notes()
before = Path(source.path).read_bytes()
partial = pageops.assets_dir("destination") / "notes-diagram.png"
partial.write_bytes(b"synthetic image")
record = pageops._start_record(
"transfer", "source", "notes", "destination", "notes", source.revision,
)
pageops._note_asset(record, "notes-diagram.png")
pageops.write_page("source", "other", title="Other", body="Unrelated")
assert not partial.exists()
assert Path(source.path).read_bytes() == before
assert pageops.page_aliases("source") == {}
assert not pageops.read_page("destination", "notes").exists
assert not list((trees / ".recovery").glob("*.json"))
def test_restart_finishes_a_published_move_without_changing_bytes(trees):
source = _notes()
copied = pageops.copy_page(
"source", "notes", "destination", "notes", title="Notes", journal=True,
)
record = pageops._start_record(
"transfer", "source", "notes", "destination", "notes", source.revision,
)
record["dest_revision"] = copied.revision
record["phase"] = "published"
pageops._write_record(record)
result = pageops.recover_interrupted_content()
assert result == {"cleared": 1, "held": 0}
assert not Path(source.path).exists()
assert pageops.page_aliases("source")["notes"] == ["destination", "notes"]
assert pageops.read_page("destination", "notes").revision == copied.revision
assert not list((trees / ".recovery").glob("*.json"))
def test_restart_keeps_a_source_that_changed_and_drops_its_alias(trees):
source = _notes()
copied = pageops.copy_page(
"source", "notes", "destination", "notes", title="Notes", journal=True,
)
pageops.remember_page_address("source", "notes", "destination", "notes")
record = pageops._start_record(
"transfer", "source", "notes", "destination", "notes", source.revision,
)
record["dest_revision"] = copied.revision
record["phase"] = "aliased"
pageops._write_record(record)
Path(source.path).write_bytes(b"changed after the crash window\n")
pageops.recover_interrupted_content()
assert Path(source.path).read_bytes() == b"changed after the crash window\n"
assert pageops.page_aliases("source") == {}
assert pageops.read_page("destination", "notes").revision == copied.revision
@pytest.mark.django_db
def test_recovered_move_still_requires_the_destination_grant(trees, client):
owner, reader = make_member("owner"), make_member("reader")
source = create_space(owner, title="Shared", slug="source")
target = create_space(owner, title="Private", slug="private")
set_space_access(owner, source, reader.pk, "viewer")
doc = pageops.write_page(source.slug, "notes", title="Notes", body="Retained content")
copied = pageops.copy_page(
source.slug, "notes", target.slug, "moved", title="Notes", journal=True,
)
record = pageops._start_record(
"transfer", source.slug, "notes", target.slug, "moved", doc.revision,
)
record["dest_revision"] = copied.revision
record["phase"] = "published"
pageops._write_record(record)
pageops.recover_interrupted_content()
client.force_login(reader.user)
old_url = reverse("spaces:detail", args=[source.slug]) + "?page=notes"
assert client.get(old_url).status_code == 404
set_space_access(owner, target, reader.pk, "viewer")
response = client.get(old_url, follow=True)
assert response.status_code == 200
assert b"Retained content" in response.content
def test_existing_destination_is_refused_without_a_journal(trees):
source = _notes()
pageops.write_page("destination", "notes", title="Keep", body="Already here")
before = pageops.page_file("destination", "notes").read_bytes()
with pytest.raises(pageops.PageOpsError, match="already exists"):
pageops.transfer_page("source", "notes", "destination", "notes")
assert pageops.page_file("destination", "notes").read_bytes() == before
assert Path(source.path).read_bytes()
assert not list((trees / ".recovery").glob("*.json"))
def test_purge_keeps_live_bytes_and_deleted_page_history(trees):
first = pageops.write_page("source", "notes", title="Notes", body="First")
pageops.write_page(
"source", "notes", title="Notes", body="Second", expected_revision=first.revision,
)
removed = pageops.write_page("source", "gone", title="Gone", body="Recover me")
pageops.delete_page("source", "gone")
revision_dir = trees / ".revisions" / "source" / "notes"
old = next(path for path in revision_dir.glob("*.md") if path.stem == first.revision)
live = next(path for path in revision_dir.glob("*.md") if path.stem != first.revision)
deleted = next((trees / ".revisions" / "source" / "gone").glob("*.md"))
past = (datetime.now(UTC) - timedelta(days=30)).timestamp()
for path in (old, live, deleted):
os.utime(path, (past, past))
removed_count = pageops.purge_revisions(
older_than=datetime.now(UTC) - timedelta(days=7),
)
assert removed_count == 1
assert not old.exists()
assert live.exists()
assert deleted.exists()
assert pageops.read_page("source", "notes").body == "Second\n"
assert removed.body == "Recover me\n"
def test_purge_stops_before_deleting_when_a_revision_hash_does_not_match(trees):
doc = pageops.write_page("source", "notes", title="Notes", body="First")
pageops.write_page(
"source", "notes", title="Notes", body="Second", expected_revision=doc.revision,
)
stored = list((trees / ".revisions" / "source" / "notes").glob("*.md"))
stored[0].write_bytes(b"tampered")
past = datetime.now(UTC) - timedelta(days=1)
with pytest.raises(pageops.PageOpsError, match="integrity"):
pageops.purge_revisions(older_than=past)
assert len(list((trees / ".revisions" / "source" / "notes").glob("*.md"))) == 2

View file

@ -40,13 +40,25 @@ The live database has not been migrated.
```task
id: CSOC-WP-0013-T02
status: todo
status: progress
priority: high
state_hub_task_id: "8d06cdee-e0ce-5c3e-bf98-705deed0ee62"
```
Review custom Markdown/raw HTML and body-linked assets with privately nominated corpus examples. Define revision retention/purge and failed multi-file copy/move crash recovery without losing source or incoming-address authorization. Current immutable revisions, stale-edit refusal, supported Markdown attachments and aliases are deployed; preserve them.
2026-10-11: Supported Markdown is fenced code, tables, newline breaks and sane
lists. Raw HTML, script, event handlers and non-http(s)/mailto URLs are removed.
Relative asset URLs stay. No private corpus has been nominated, so this is not
corpus acceptance. Revisions are not purged on save. `purge_revisions` deletes
only inactive revisions older than an explicit timezone-aware cutoff, keeps the
live page bytes, and keeps every revision of a deleted page. A hash mismatch
aborts the purge before any deletion. Copy and move write a journal under
CONTENT_ROOT/.recovery. The next content write finishes a confirmed move or
discards an unconfirmed destination. The source is deleted only when its bytes
still match the journal. Alias changes do not grant destination access. The
live content root has not been purged or recovered.
## T03 — Review productive sharing and personal portability
```task