coulomb/coulomb-social
5
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Prepare exact mirror expiry with durable interruption recovery
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 6m27s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a18e2d-f513-7742-b8e7-12af0b744ff3
This commit is contained in:
tegwick 2026-10-31 04:07:32 +01:00
parent 51a527128a
commit 235f13c0d2
6 changed files with 1019 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

51
docs/release/mirror-expiry-executor-preflight-2026-10-31.json Normal file
View file

@ -0,0 +1,51 @@
{
"candidate_points": 1,
"candidate_versions": 2,
"decryption_identity_read": false,
"dedicated_local_expiry_pair_present": false,
"expiry_caller_used": false,
"independent_reader_used": true,
"kept_generations": 3,
"native_expiry_boundary_verified": false,
"native_expiry_executed": false,
"native_return": {
"deletion_admitted": false,
"mirror_bytes_verified": 8669264,
"proposal_digest": "6af0564eeb560c2fb80b436e271646ef794912c1484c6ea21409ed16b588cd72",
"provider_mutations": false,
"status": "read_only_expiry_preflight_verified"
},
"private_proposal_sha256": "6af0564eeb560c2fb80b436e271646ef794912c1484c6ea21409ed16b588cd72",
"protected_kit_and_proof_version_heads_verified": true,
"protected_kit_body_refetched": false,
"provider_mutations": false,
"publisher_or_custodian_used": false,
"recorded_at": "2026-10-31T03:05:28.028375+00:00",
"recurring_retention_activated": false,
"schema": "coulomb.mirror-expiry-preflight-proof.v1",
"sources_manifest_sha256": "4ec2d2ec6884c7a260861e6d098d2470e44583970a3a3b9cae48dede9d0b2c47",
"targets": [
{
"bucket": "railiance-coulomb-social-archives",
"bytes": 2181848,
"object_key": "archives/platform-pg/application-archives/coulomb-social/2026/10/31/20261031T002756Z-c31b84b48b3874bdaa87cbf5.tar.age",
"role": "ciphertext",
"sha256": "f7be502c7e122283e70b10d2fd8a2cbf9550d2261192ba11f94d945a9f39af51",
"version_id": "1793408525635522"
},
{
"bucket": "railiance-coulomb-social-archives",
"bytes": 1664,
"object_key": "archives/platform-pg/application-archives/coulomb-social/2026/10/31/20261031T002756Z-c31b84b48b3874bdaa87cbf5.tar.age.cbbf4c0b9cab6829293e00435f225e9cc3c0167930307c289428c1f9ded2a6f7.receipt.json",
"role": "receipt",
"sha256": "cbbf4c0b9cab6829293e00435f225e9cc3c0167930307c289428c1f9ded2a6f7",
"version_id": "1793408541632331"
}
],
"validation": {
"browser_journeys_included": true,
"make_check_passed": 441,
"skipped": 0,
"synthetic_expiry_cases_passed": 26
}
}

10
docs/release/mirror-expiry-executor-preflight-2026-10-31.sources.txt Normal file
View file

@ -0,0 +1,10 @@
/home/worsch/coulomb-social/deploy/backup/mirror-platform-runtime.json: 9253c5f6e956b67f499d3e12d2851be7fabd3bc56b00496b8810bbd0be8209f1
/home/worsch/coulomb-social/scripts/backup_capacity.py: cb699a79e67b129239b5aa5d6f3b0d4ffaa9773a8b073227ad17fdecb5b9d893
/home/worsch/coulomb-social/scripts/check-paired-backup-health.py: c003871fe91c5c31788fd9bfc50aef3b2a275c06c5a42c2c3165b92d5ea80db9
/home/worsch/coulomb-social/scripts/expire-backup-mirror.py: 387cd0536add3615c0def9eaa8ce5e2eae17e9ada4c4ed98bda89fcd7730eadb
/home/worsch/coulomb-social/scripts/mirror-paired-backup.py: 9ceb26d9083af016c4b787907e4dce7bdf437c782fb2d15eb28d9e334e6e431c
/home/worsch/coulomb-social/scripts/observe-backup-retention.py: 9dc7e862ef2888998744d47cb458f0401b870e96cf283df5dd375b627b3f176f
/home/worsch/coulomb-social/scripts/review-backup-retention.py: 929c0416add6bfddee5bce9e6aaeb436d2d9de8084db48d061f66186f7127794
/home/worsch/coulomb-social/tests/test_backup_mirror_expiry.py: 65f3f025c30c261e0054b8b0a2fef7e753e8b984088b5c473b8c7272743bb7c5
/home/worsch/railiance-platform/scripts/application_archive_retention_plan.py: 08bc57308f53bbfd0d58fdbb9b5671dd4285e31c5b5137e201524df606aa3f9f
/home/worsch/railiance-platform/scripts/primary_application_archive.py: a3f0a61e869a6012edb4b5288be14e3b0c4a9a1273b090320ab146d12e5aa94d

112
docs/release/mirror-expiry-executor-review-2026-10-31.md Normal file
View file

@ -0,0 +1,112 @@
# Mirror expiry executor — prepared, not admitted
`scripts/expire-backup-mirror.py` now implements exact-version expiry with durable
interruption recovery. The current proposal identifies only the October 31 00:27
mirror ciphertext and its public receipt. Both remain retained. The independently
verified October 29, October 30 and October 31 02:07 generations remain in the
keep set, together with all protected-kit versions and unknown objects/fixtures.
Proposal digest:
`6af0564eeb560c2fb80b436e271646ef794912c1484c6ea21409ed16b588cd72`.
Private proposal:
`~/.config/railiance-platform/coulomb-social-recovery/expiry-review-20261031/proposal.json`.
This is a source/read-only-preflight result, not deletion admission or a recurring
retention service. No expiry credentials were created or used.
## Selection and execution boundaries
Preparation recomputes the existing pinned 14-daily/four-weekly selector and the
consumer retention review from the exact input inventory, protected recovery
proof and fresh observation. The proposal binds those inputs, the resulting
keep/target sets, six local source files, the owner selector/helper pins and the
platform runtime manifest. Changed sources, foreign targets, changed protected
bindings and modified selection refuse before provider access.
The independent reader performs every verification. It rereads complete mirror
and kit version inventories and their lifecycle/versioning state, compares exact
known HEAD identities/hashes/sizes, and streams/hashes every known mirrored
ciphertext and receipt. The protected kit and recovery proof receive fresh HEAD
checks against their previously verified immutable versions. The full kit is not
restored on each check; the prior provider-only recovery proof remains separate.
The expiry caller has one operation: `DeleteObject` with an explicit immutable
`VersionId`, always in the Coulomb mirror bucket. It never receives primary or kit
targets and is never used as a fallback reader. Each candidate's ciphertext is
removed and independently confirmed absent before its associated receipt is
eligible. Kept generations and protected custody are rechecked immediately before
each delete and in postflight. Unknown objects are preserved.
The owner-private attempt directory contains a flock and a proposal-bound journal.
Intent is flushed/fsynced and atomically published before each request. A lost
reply leaves intent unresolved; resume checks both current version listing and
exact-version HEAD. An already absent version is reconciled without another
delete. A still-present version can be retried only under the same proposal and
fresh keep-set checks. A 403 or transient read failure is not proof of absence.
Journal write failure prevents the request. Terminal replay performs verification
without repeating deletes.
The local lock serializes this attempt directory only. It does not fence another
host or provider actor. Any observed version addition, unrelated disappearance,
delete marker, changed lifecycle or changed custody stops the run with its journal
retained. If such drift occurs after partial expiry, owner reconciliation and a
new reviewed continuation are required; do not discard the journal to force a
fresh run. Exact-version deletion is irreversible. The prepared first run needs
an owner-coordinated window; distributed recurring execution remains unimplemented.
## Verification already performed
The native read-only command succeeded with the dedicated reader, verifying
8,669,264 bytes across four ciphertexts and two public receipts. Complete mirror
and kit inventories still match the proposal. The protected versions' metadata
matches their recovery proof. No expiry pair, decryption identity or provider
mutation was used. Evidence: [preflight proof](mirror-expiry-executor-preflight-2026-10-31.json)
and its [source manifest](mirror-expiry-executor-preflight-2026-10-31.sources.txt).
```sh
/path/to/locked-sdk/python scripts/expire-backup-mirror.py --verify \
--proposal ~/.config/railiance-platform/coulomb-social-recovery/expiry-review-20261031/proposal.json \
--expect-digest 6af0564eeb560c2fb80b436e271646ef794912c1484c6ea21409ed16b588cd72 \
--platform-scripts /home/worsch/railiance-platform/scripts \
--runtime-manifest deploy/backup/mirror-platform-runtime.json \
--credentials-directory ~/.config/railiance-platform/coulomb-social-recovery
```
Twenty-six synthetic tests cover exact ordering and preserved versions, durable
intent, lost replies before/after deletion, disk failure, current inventory/hash/
body/custody drift, unsuccessful deletion, keep loss mid-run, 403 versus absence,
proposal/source/runtime tampering, journal ordering, private atomic checkpoints
and offline default preview. Full `make check`: 441 passes, no skipped tests,
including mandatory browser journeys. Synthetic execution is not a native expiry
boundary or deletion receipt.
## Remaining owner action
No admitted local `expiry-pair.json` is present. The current recovery contract
names publisher, reader and custodian roles; their existing deletion denials must
remain intact. Warden's `scaleway-bootstrap` route is still draft/non-resolvable,
not a routine execution lane. Source readiness does not supply provider
administration authority or justify retaining a personal Console session.
The proposed additional caller needs project-scoped object-deletion IAM capability
intersected with a bucket-policy allow for `s3:DeleteObjectVersion` only under
`railiance-coulomb-social-archives/archives/platform-pg/application-archives/coulomb-social/*`.
It needs no read/list rights because the existing independent reader performs
those checks. This split refines the earlier combined read/expiry caller proposal.
Scaleway documents version deletion as a separate action and its current bucket
policy as an allowlist intersected with IAM permissions:
[official bucket-policy reference](https://www.scaleway.com/en/docs/object-storage/api-cli/bucket-policy/).
The live policy must be read, diffed and reviewed before adding a new principal;
this packet is not a replacement policy or a provider apply command.
Required before the first attended expiry: owner admission of the new caller and
credential custody/renewal, native exact-version deletion of a newly created
synthetic fixture, and denials for unversioned deletion, writes, primary/kit/foreign
prefix access, lifecycle/bucket administration, escrow and IAM. Verify publisher
and reader deletion denials remain unchanged. Then refresh/rebase the proposal
if needed and record founder execution plus independent postflight. The CLI
requires the separate pair and exact terminal confirmation; it is not an approval
service. No current application backup has been approved for deletion here.
Recurring retention integration, node operating delivery, capacity admission,
received alerts, historical-copy disposition and service-contract conformance
remain original WP-0012 obligations.

501
scripts/expire-backup-mirror.py Normal file
View file

@ -0,0 +1,501 @@
#!/usr/bin/env python3
"""Prepare, verify or perform an attended exact-version mirror expiry.
Preparation/preview is offline; verification uses only the independent reader.
Execution requires a separately admitted expiry pair and founder confirmation.
This tool does not grant authority, provision credentials or activate a schedule.
"""
from __future__ import annotations
import argparse
import fcntl
import hashlib
import importlib.util
import json
import os
import sys
import tempfile
from datetime import UTC, datetime
from pathlib import Path
def module(name, filename):
spec = importlib.util.spec_from_file_location(name, Path(__file__).with_name(filename))
loaded = importlib.util.module_from_spec(spec)
spec.loader.exec_module(loaded)
return loaded
observer = module("expiry_observer", "observe-backup-retention.py")
reviewer = module("expiry_reviewer", "review-backup-retention.py")
health = observer.health
Refused = health.Refused
MIRROR = health.MIRROR_BUCKET
KIT = observer.KIT
SOURCE_FILES = (
"expire-backup-mirror.py",
"observe-backup-retention.py",
"review-backup-retention.py",
"mirror-paired-backup.py",
"check-paired-backup-health.py",
"backup_capacity.py",
)
def canonical(value):
return observer.worker.canonical(value)
def sha(data):
return hashlib.sha256(data).hexdigest()
def source_hashes():
return {name: sha(Path(__file__).with_name(name).read_bytes()) for name in SOURCE_FILES}
def key(item):
return item["bucket"], item["object_key"], item["version_id"]
def version_map(observations):
result = {}
for bucket in observations:
if bucket["bucket"] not in (MIRROR, KIT):
continue
if bucket.get("delete_markers"):
raise Refused("delete_markers_require_separate_review")
for item in bucket["versions"]:
identity = bucket["bucket"], item["Key"], item["VersionId"]
observer.version_id(item["VersionId"])
if identity in result:
raise Refused("duplicate_version")
result[identity] = item["Size"]
return result
def derive(inventory, recovery, observation, platform_scripts, created_at):
owner = reviewer.load_owner(platform_scripts)
selection = owner.plan("coulomb-social", inventory, recovery, created_at)
review = reviewer.review(inventory, recovery, observation, selection, created_at)
if not review["generation_selection_valid"] or review["blockers"]:
raise Refused("valid_generation_selection_required")
targets = []
for candidate in review["expiration_candidates"]:
targets.append(
{
"bucket": MIRROR,
"object_key": candidate["object_key"],
"version_id": candidate["version_id"],
"sha256": candidate["ciphertext_sha256"],
"bytes": candidate["ciphertext_bytes"],
"role": "ciphertext",
}
)
receipt = candidate["associated_receipt"]
targets.append(
{
"bucket": MIRROR,
"object_key": receipt["receipt_object_key"],
"version_id": receipt["receipt_version_id"],
"sha256": receipt["receipt_sha256"],
"bytes": receipt["receipt_bytes"],
"role": "receipt",
}
)
ids = [key(t) for t in targets]
if len(ids) != len(set(ids)):
raise Refused("duplicate_expiry_target")
keep = {key(t) for t in review["keep"]}
if keep.intersection(ids) or any(t[0] != MIRROR for t in ids):
raise Refused("expiry_overlaps_keep_or_other_bucket")
baseline = version_map(observation["observations"])
if any(baseline.get(key(t)) != t["bytes"] for t in targets):
raise Refused("expiry_target_not_in_complete_inventory")
return {"keep": review["keep"], "targets": targets}
def prepare(inventory, recovery, observation, platform_scripts, now, runtime_manifest):
derived = derive(inventory, recovery, observation, platform_scripts, now)
return {
"schema": "coulomb.mirror-expiry-proposal.v1",
"application": "coulomb-social",
"project_id": health.PROJECT,
"created_at": now.isoformat(),
"sources": source_hashes(),
"owner_sources": reviewer.OWNER_FILES,
"runtime_manifest_sha256": sha(runtime_manifest.read_bytes()),
"inventory": inventory,
"recovery": recovery,
"observation": observation,
**derived,
"requires_separate_expiry_admission": True,
"requires_founder_attended_execution": True,
}
def validate(proposal, expected, platform_scripts, now, runtime_manifest):
if sha(canonical(proposal)) != expected:
raise Refused("proposal_digest_differs")
if (
proposal.get("schema") != "coulomb.mirror-expiry-proposal.v1"
or proposal.get("application") != "coulomb-social"
or proposal.get("project_id") != health.PROJECT
or proposal.get("sources") != source_hashes()
or proposal.get("owner_sources") != reviewer.OWNER_FILES
or proposal.get("runtime_manifest_sha256") != sha(runtime_manifest.read_bytes())
or proposal.get("requires_separate_expiry_admission") is not True
or proposal.get("requires_founder_attended_execution") is not True
):
raise Refused("reviewed_expiry_sources_and_scope_required")
created = health.timestamp(proposal["created_at"])
if created > now:
raise Refused("future_proposal")
# Validate original selection at proposal time. Every run, including resume,
# independently rereads current custody and the entire frozen version set.
derived = derive(
proposal["inventory"],
proposal["recovery"],
proposal["observation"],
platform_scripts,
created,
)
if any(proposal[k] != v for k, v in derived.items()):
raise Refused("proposal_selection_differs_from_owner")
def new_journal(digest, count):
return {
"schema": "coulomb.mirror-expiry-attempt.v1",
"proposal_digest": digest,
"steps": ["pending"] * count,
"status": "prepared",
}
def validate_journal(journal, digest, count):
if (
journal.get("schema") != "coulomb.mirror-expiry-attempt.v1"
or journal.get("proposal_digest") != digest
or len(journal.get("steps", [])) != count
or journal.get("status") not in {"prepared", "running", "completed"}
):
raise Refused("journal_binding_differs")
tail = False
for state in journal["steps"]:
if state not in {"pending", "intent", "absent_verified"}:
raise Refused("invalid_journal_step")
if state == "absent_verified":
if tail:
raise Refused("out_of_order_expiry_journal")
elif state == "intent":
if tail:
raise Refused("out_of_order_expiry_journal")
tail = True
else:
tail = True
if journal["status"] == "completed" and any(s != "absent_verified" for s in journal["steps"]):
raise Refused("incomplete_terminal_journal")
def exact_head(reader, target):
value = reader.head_object(
Bucket=target["bucket"], Key=target["object_key"], VersionId=target["version_id"]
)
if (
value.get("VersionId") != target["version_id"]
or value.get("ContentLength") != target["bytes"]
or value.get("Metadata", {}).get("sha256") != target["sha256"]
or value.get("DeleteMarker")
):
raise Refused("retained_version_head_differs")
def confirm_absent(reader, target):
try:
reader.head_object(
Bucket=target["bucket"], Key=target["object_key"], VersionId=target["version_id"]
)
except Exception as error:
if observer.status(error) == 404:
return
raise
raise Refused("expired_version_still_present")
def verify(reader, proposal, journal):
"""Fresh native reads; permit disappearance only after durable own intent."""
digest = sha(canonical(proposal))
validate_journal(journal, digest, len(proposal["targets"]))
current = [observer.observe_bucket(reader, bucket) for bucket in (MIRROR, KIT)]
for bucket in current:
rules = bucket["lifecycle"]["Rules"]
if (bucket["bucket"] == KIT and rules) or any(r.get("Status") == "Enabled" for r in rules):
raise Refused("generation_or_protected_custody_lifecycle_changed")
baseline = version_map(proposal["observation"]["observations"])
present = version_map(current)
allowed_absent = {
key(t)
for t, state in zip(proposal["targets"], journal["steps"])
if state in {"intent", "absent_verified"}
}
missing = baseline.keys() - present.keys()
if (
present.keys() - baseline.keys()
or not missing <= allowed_absent
or any(baseline[k] != size for k, size in present.items())
):
raise Refused("provider_inventory_changed_outside_own_expiry")
for target, state in zip(proposal["targets"], journal["steps"]):
if key(target) in missing:
confirm_absent(reader, target)
elif state == "absent_verified":
raise Refused("expired_version_reappeared")
heads = [h for h in proposal["observation"]["version_heads"] if h["bucket"] in (MIRROR, KIT)]
for target in heads:
if key(target) in present:
exact_head(reader, target)
# Independently stream/hash every retained known mirror object, including
# the recovered anchor. Kit identity is HEAD-checked against its prior full
# provider-only recovery proof; the large kit is not restored on each expiry.
kept = {key(t) for t in proposal["keep"]}
verified_bytes = 0
for target in heads:
if key(target) not in present or (target["bucket"] != MIRROR):
continue
# Also validate candidates/sidecars before touching them. No plaintext.
value = reader.get_object(
Bucket=MIRROR, Key=target["object_key"], VersionId=target["version_id"]
)
total = 0
hasher = hashlib.sha256()
with value["Body"] as stream:
while chunk := stream.read(min(1024 * 1024, target["bytes"] + 1 - total)):
total += len(chunk)
if total > target["bytes"]:
raise Refused("retained_version_body_oversized")
hasher.update(chunk)
if (
value.get("VersionId") != target["version_id"]
or value.get("ContentLength") != target["bytes"]
or total != target["bytes"]
or hasher.hexdigest() != target["sha256"]
):
raise Refused("retained_version_body_differs")
verified_bytes += total
if not kept <= present.keys():
raise Refused("kept_generation_missing")
return {
"absent": missing,
"mirror_bytes_verified": verified_bytes,
"protected_versions_head_verified": True,
}
def execute(reader, expiry, proposal, journal, save):
"""Exact single-version deletes; caller is never used for reads or fallback."""
digest = sha(canonical(proposal))
validate_journal(journal, digest, len(proposal["targets"]))
state = verify(reader, proposal, journal)
for index, target in enumerate(proposal["targets"]):
if key(target) in state["absent"]:
journal["steps"][index] = "absent_verified"
save()
continue
# Recheck custody and the complete keep set immediately before each action.
state = verify(reader, proposal, journal)
if key(target) in state["absent"]:
journal["steps"][index] = "absent_verified"
save()
continue
journal["steps"][index] = "intent"
journal["status"] = "running"
save() # Must be durable before any destructive request, including retries.
value = expiry.delete_object(
Bucket=MIRROR, Key=target["object_key"], VersionId=target["version_id"]
)
if (
value.get("ResponseMetadata", {}).get("HTTPStatusCode") != 204
or value.get("DeleteMarker") is True
):
raise Refused("uncertain_exact_version_delete")
state = verify(reader, proposal, journal)
if key(target) not in state["absent"]:
raise Refused("deletion_not_independently_verified")
journal["steps"][index] = "absent_verified"
save()
verify(reader, proposal, journal)
journal.update(status="completed", completed_at=datetime.now(UTC).isoformat())
save()
return {
"status": "exact_mirror_versions_expired_verified",
"proposal_digest": digest,
"versions_absent_verified": len(proposal["targets"]),
"kept_generations": len(proposal["keep"]),
"protected_custody_verified": True,
"primary_or_kit_deleted": False,
"unknown_objects_preserved": True,
}
def checkpoint(path, value):
if path.exists() or path.is_symlink():
health.regular(path)
fd, temporary = tempfile.mkstemp(prefix="expiry-journal-", dir=path.parent)
try:
with os.fdopen(fd, "wb") as stream:
stream.write(canonical(value))
stream.flush()
os.fsync(stream.fileno())
os.replace(temporary, path)
directory = os.open(path.parent, os.O_RDONLY | os.O_DIRECTORY)
try:
os.fsync(directory)
finally:
os.close(directory)
finally:
if os.path.exists(temporary):
os.unlink(temporary)
def main():
parser = argparse.ArgumentParser(description=__doc__)
modes = parser.add_mutually_exclusive_group()
modes.add_argument("--prepare", action="store_true")
modes.add_argument("--verify", action="store_true")
modes.add_argument("--execute", action="store_true")
for name in (
"inventory",
"recovery",
"observation",
"platform-scripts",
"runtime-manifest",
"proposal",
"output",
"credentials-directory",
"state-dir",
):
parser.add_argument("--" + name, type=Path)
parser.add_argument("--expect-digest")
args = parser.parse_args()
if not any((args.prepare, args.verify, args.execute)):
print(
json.dumps(
{"status": "preview", "provider_contacted": False, "deletion_admitted": False}
)
)
return
if args.prepare:
proposal = prepare(
*[
json.loads(getattr(args, name).read_bytes())
for name in ("inventory", "recovery", "observation")
],
args.platform_scripts,
datetime.now(UTC),
args.runtime_manifest,
)
observer.worker.private_directory(args.output.parent)
if any((p / ".git").exists() for p in [args.output.parent, *args.output.parents]):
raise Refused("private_proposal_outside_git_required")
fd = os.open(args.output, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, "wb") as stream:
stream.write(canonical(proposal))
print(
json.dumps(
{
"status": "proposal_prepared",
"proposal_digest": sha(canonical(proposal)),
"candidate_points": len(proposal["targets"]) // 2,
"provider_contacted": False,
"deletion_admitted": False,
}
)
)
return
proposal = health.read_json(args.proposal)
validate(
proposal,
args.expect_digest,
args.platform_scripts,
datetime.now(UTC),
args.runtime_manifest,
)
observer.worker.load_runtime(args.platform_scripts, args.runtime_manifest)
from application_recovery_identities import private_directory, private_json
from application_recovery_resources import s3
private_directory(args.credentials_directory)
reader_pair = private_json(args.credentials_directory / "reader-pair.json")
digest = sha(canonical(proposal))
if args.verify:
reader = s3(reader_pair, health.PROJECT)
try:
result = verify(reader, proposal, new_journal(digest, len(proposal["targets"])))
print(
json.dumps(
{
"status": "read_only_expiry_preflight_verified",
"proposal_digest": digest,
"provider_mutations": False,
"mirror_bytes_verified": result["mirror_bytes_verified"],
"deletion_admitted": False,
}
)
)
finally:
reader.close()
return
if not sys.stdin.isatty():
raise Refused("founder_attended_terminal_required")
private_directory(args.state_dir)
# This new pair must be provisioned/admitted separately. There is no fallback
# to the publisher, reader or custodian when the dedicated file is absent.
expiry_pair = private_json(args.credentials_directory / "expiry-pair.json")
if expiry_pair["AWS_ACCESS_KEY_ID"] == reader_pair["AWS_ACCESS_KEY_ID"]:
raise Refused("separate_expiry_caller_required")
expected = "EXPIRE-MIRROR " + digest
if input("Type " + expected + " only after separate expiry admission: ").strip() != expected:
raise Refused("not_confirmed")
fd = os.open(args.state_dir / "expiry.lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
with os.fdopen(fd, "w") as lock:
health.regular(args.state_dir / "expiry.lock")
fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB)
path = args.state_dir / (digest + ".json")
journal = (
health.read_json(path)
if path.exists() or path.is_symlink()
else new_journal(digest, len(proposal["targets"]))
)
validate_journal(journal, digest, len(proposal["targets"]))
checkpoint(path, journal)
reader = s3(reader_pair, health.PROJECT)
expiry = None
try:
expiry = s3(expiry_pair, health.PROJECT)
print(
json.dumps(
execute(reader, expiry, proposal, journal, lambda: checkpoint(path, journal))
)
)
finally:
reader.close()
if expiry is not None:
expiry.close()
if __name__ == "__main__":
try:
main()
except Exception as error:
print(
json.dumps(
{
"status": "expiry_requires_review",
"failure_type": type(error).__name__,
"journal_preserved": True,
}
)
)
raise SystemExit(1)

329
tests/test_backup_mirror_expiry.py Normal file
View file

@ -0,0 +1,329 @@
"""Synthetic expiry must preserve custody and recover ambiguous single-version deletes."""
import copy
import hashlib
import importlib.util
import json
import subprocess
import sys
from datetime import timedelta
from pathlib import Path
from types import SimpleNamespace
import pytest
from tests.test_backup_health import NOW
from tests.test_backup_health import health_state as health_state
from tests.test_backup_health import monitor as monitor
from tests.test_backup_retention_observation import ProviderError, ReadOnlyStore
from tests.test_backup_retention_review import evidence as evidence
from tests.test_backup_retention_review import reviewer as reviewer
RUNTIME_MANIFEST = (
Path(__file__).resolve().parents[1] / "deploy/backup/mirror-platform-runtime.json"
)
@pytest.fixture
def executor(monkeypatch):
scripts = Path(__file__).resolve().parents[1] / "scripts"
monkeypatch.syspath_prepend(str(scripts))
spec = importlib.util.spec_from_file_location(
"mirror_expiry", scripts / "expire-backup-mirror.py"
)
value = importlib.util.module_from_spec(spec)
spec.loader.exec_module(value)
return value
class Store(ReadOnlyStore):
def head_object(self, Bucket, Key, VersionId):
if (Bucket, Key, VersionId) not in self.heads:
raise ProviderError(404, "NoSuchVersion")
return super().head_object(Bucket, Key, VersionId)
def remove(self, identity):
bucket, key, version = identity
self.buckets[bucket]["versions"] = [
v
for v in self.buckets[bucket]["versions"]
if (v["Key"], v["VersionId"]) != (key, version)
]
self.heads.pop(identity, None)
self.payloads.pop(identity, None)
class ExpiryCaller:
"""No get/list/put methods: reads must use the independent reader."""
def __init__(self, store):
self.store = store
self.calls = []
self.failure = None
self.after_delete = None
def delete_object(self, *, Bucket, Key, VersionId):
self.calls.append((Bucket, Key, VersionId))
if self.failure == "before":
self.failure = None
raise TimeoutError("synthetic uncertain request")
if self.failure != "no_effect":
self.store.remove((Bucket, Key, VersionId))
if self.after_delete:
self.after_delete()
if self.failure == "after":
self.failure = None
raise TimeoutError("synthetic lost reply")
return {"ResponseMetadata": {"HTTPStatusCode": 204}, "VersionId": VersionId}
@pytest.fixture
def scenario(executor, evidence, monkeypatch):
inventory, recovery, observation, selection = evidence
for name in list(inventory[0]["recovery_mirror"]):
if name.startswith("receipt_"):
del inventory[0]["recovery_mirror"][name]
store = Store(evidence)
payload = b"synthetic encrypted paired point"
ref = inventory[0]["recovery_mirror"]
store.payloads[ref["bucket"], ref["object_key"], ref["version_id"]] = payload
def add_version(bucket, key, version, data):
row = {"Key": key, "VersionId": version, "Size": len(data)}
head = {
"bucket": bucket,
"object_key": key,
"version_id": version,
"bytes": len(data),
"sha256": hashlib.sha256(data).hexdigest(),
}
next(b for b in observation["observations"] if b["bucket"] == bucket)["versions"].append(
row
)
observation["version_heads"].append(head)
store.buckets[bucket]["versions"].append(dict(row))
store.heads[bucket, key, version] = {
"VersionId": version,
"ContentLength": len(data),
"Metadata": {"sha256": head["sha256"]},
}
store.payloads[bucket, key, version] = data
for minutes, label in [(30, "candidate"), (0, "newest")]:
point = copy.deepcopy(inventory[0])
stamp = NOW - timedelta(minutes=minutes)
name = stamp.strftime("%Y%m%dT%H%M%SZ-") + "f" * 24 + ".tar.age"
key = executor.health.PREFIX + stamp.strftime("%Y/%m/%d/") + name
point.update(
captured_at=stamp.isoformat(),
local_archive=name,
object_key=key,
destination="s3://" + executor.observer.PRIMARY + "/" + key,
version_id=label + "-primary",
)
mirror = point["recovery_mirror"]
mirror.update(object_key="archives/" + key, version_id=label + "-mirror")
add_version(executor.observer.PRIMARY, key, point["version_id"], payload)
add_version(executor.MIRROR, mirror["object_key"], mirror["version_id"], payload)
if label == "candidate":
sidecar = executor.canonical(executor.observer.worker.public_receipt(point, point))
digest = hashlib.sha256(sidecar).hexdigest()
mirror.update(
receipt_object_key=mirror["object_key"] + "." + digest + ".receipt.json",
receipt_version_id="candidate-receipt",
receipt_bytes=len(sidecar),
receipt_sha256=digest,
receipt_readback_verified=True,
)
add_version(executor.MIRROR, mirror["receipt_object_key"], "candidate-receipt", sidecar)
inventory.append(point)
identity = {
k: point[k]
for k in ("object_key", "version_id", "ciphertext_sha256", "ciphertext_bytes")
}
selection["expiration_candidates" if label == "candidate" else "keep"].append(identity)
# Preserve a synthetic boundary fixture despite its not being a generation.
add_version(executor.MIRROR, "archives/synthetic-fixture", "fixture", b"fixture")
monkeypatch.setattr(
executor.reviewer,
"load_owner",
lambda _: SimpleNamespace(plan=lambda *args: copy.deepcopy(selection)),
)
proposal = executor.prepare(
inventory, recovery, observation, Path("unused"), NOW, RUNTIME_MANIFEST
)
digest = executor.sha(executor.canonical(proposal))
executor.validate(proposal, digest, Path("unused"), NOW, RUNTIME_MANIFEST)
journal = executor.new_journal(digest, len(proposal["targets"]))
return proposal, journal, store, ExpiryCaller(store)
def test_exact_two_version_expiry_preserves_kept_unknown_primary_and_kit(executor, scenario):
proposal, journal, store, caller = scenario
before = set(store.heads)
snapshots = []
result = executor.execute(
store, caller, proposal, journal, lambda: snapshots.append(copy.deepcopy(journal))
)
expected = [executor.key(t) for t in proposal["targets"]]
assert caller.calls == expected
assert set(store.heads) == before - set(expected)
assert result["versions_absent_verified"] == 2 and journal["status"] == "completed"
assert any(s["steps"] == ["intent", "pending"] for s in snapshots)
assert any(s["steps"] == ["absent_verified", "intent"] for s in snapshots)
assert all(bucket == executor.MIRROR and version for bucket, _, version in caller.calls)
executor.execute(store, caller, proposal, journal, lambda: None)
assert caller.calls == expected # Terminal replay verifies; it does not delete again.
@pytest.mark.parametrize("failure,expected_calls", [("before", 3), ("after", 2)])
def test_ambiguous_delete_reconciles_before_retry(executor, scenario, failure, expected_calls):
proposal, journal, store, caller = scenario
durable = []
def save():
durable.append(copy.deepcopy(journal))
caller.failure = failure
with pytest.raises(TimeoutError):
executor.execute(store, caller, proposal, journal, save)
assert durable[-1]["steps"] == ["intent", "pending"]
resumed = copy.deepcopy(durable[-1])
executor.execute(store, caller, proposal, resumed, lambda: None)
assert len(caller.calls) == expected_calls and resumed["status"] == "completed"
def test_durable_checkpoint_failure_prevents_delete(executor, scenario):
proposal, journal, store, caller = scenario
def full_disk():
raise OSError("synthetic disk full")
with pytest.raises(OSError):
executor.execute(store, caller, proposal, journal, full_disk)
assert caller.calls == []
@pytest.mark.parametrize(
"change",
["missing_keep", "missing_unknown", "new_version", "hash", "body", "lifecycle", "versioning"],
)
def test_changed_custody_refuses_all_expiry(executor, scenario, change):
proposal, journal, store, caller = scenario
keep = executor.key(proposal["keep"][0])
if change == "missing_keep":
store.remove(keep)
elif change == "missing_unknown":
store.remove((executor.MIRROR, "archives/synthetic-fixture", "fixture"))
elif change == "new_version":
store.buckets[executor.MIRROR]["versions"].append(
{"Key": "archives/new", "VersionId": "new", "Size": 1}
)
elif change == "hash":
store.heads[keep]["Metadata"]["sha256"] = "0" * 64
elif change == "body":
store.payloads[keep] = b"changed"
elif change == "lifecycle":
store.buckets[executor.KIT]["lifecycle"]["Rules"] = [{"Status": "Enabled"}]
else:
store.buckets[executor.MIRROR]["versioning"] = {"Status": "Suspended"}
with pytest.raises(ValueError):
executor.execute(store, caller, proposal, journal, lambda: None)
assert caller.calls == []
def test_missing_candidate_without_own_intent_is_not_adopted(executor, scenario):
proposal, journal, store, caller = scenario
store.remove(executor.key(proposal["targets"][0]))
with pytest.raises(ValueError, match="inventory_changed"):
executor.execute(store, caller, proposal, journal, lambda: None)
assert caller.calls == []
def test_provider_success_without_actual_absence_stops_before_receipt(executor, scenario):
proposal, journal, store, caller = scenario
caller.failure = "no_effect"
with pytest.raises(ValueError, match="deletion_not_independently_verified"):
executor.execute(store, caller, proposal, journal, lambda: None)
assert len(caller.calls) == 1 and journal["steps"] == ["intent", "pending"]
def test_keep_loss_after_first_delete_stops_remaining_actions(executor, scenario):
proposal, journal, store, caller = scenario
caller.after_delete = lambda: store.remove(executor.key(proposal["keep"][0]))
with pytest.raises(ValueError, match="inventory_changed"):
executor.execute(store, caller, proposal, journal, lambda: None)
assert len(caller.calls) == 1 and journal["steps"] == ["intent", "pending"]
def test_forbidden_is_not_proof_of_deleted_version(executor, scenario):
proposal, journal, store, caller = scenario
target = proposal["targets"][0]
store.remove(executor.key(target))
journal["steps"][0] = "intent"
original = store.head_object
def denied(**kwargs):
if kwargs["Key"] == target["object_key"]:
raise ProviderError(403, "AccessDenied")
return original(**kwargs)
store.head_object = denied
with pytest.raises(ProviderError):
executor.execute(store, caller, proposal, journal, lambda: None)
assert caller.calls == []
@pytest.mark.parametrize("change", ["digest", "source", "runtime", "target", "protected", "future"])
def test_capsule_mutation_is_rejected(executor, scenario, change):
proposal, journal, _, _ = scenario
expected = journal["proposal_digest"]
if change == "digest":
expected = "0" * 64
elif change == "source":
proposal["sources"]["expire-backup-mirror.py"] = "0" * 64
elif change == "runtime":
proposal["runtime_manifest_sha256"] = "0" * 64
elif change == "target":
proposal["targets"][0]["bucket"] = executor.KIT
elif change == "protected":
proposal["recovery"]["database_sha256"] = "0" * 64
else:
proposal["created_at"] = (NOW + timedelta(hours=1)).isoformat()
if change != "digest":
expected = executor.sha(executor.canonical(proposal))
with pytest.raises(ValueError):
executor.validate(proposal, expected, Path("unused"), NOW, RUNTIME_MANIFEST)
@pytest.mark.parametrize(
"steps", [["pending", "absent_verified"], ["intent", "intent"], ["unknown", "pending"]]
)
def test_out_of_order_journal_refuses_without_provider_calls(executor, scenario, steps):
proposal, journal, store, caller = scenario
journal["steps"] = steps
with pytest.raises(ValueError):
executor.execute(store, caller, proposal, journal, lambda: None)
assert store.calls == [] and caller.calls == []
def test_checkpoint_is_private_and_refuses_linked_journal(executor, tmp_path):
target = tmp_path / "journal.json"
executor.checkpoint(target, {"status": "synthetic"})
assert target.stat().st_mode & 0o777 == 0o600
assert json.loads(target.read_text()) == {"status": "synthetic"}
link = tmp_path / "link.json"
link.symlink_to(target)
with pytest.raises(ValueError):
executor.checkpoint(link, {"status": "wrong"})
assert json.loads(target.read_text()) == {"status": "synthetic"}
def test_default_preview_needs_no_provider_or_credentials():
script = Path(__file__).resolve().parents[1] / "scripts/expire-backup-mirror.py"
result = subprocess.run([sys.executable, str(script)], capture_output=True, check=True)
assert json.loads(result.stdout) == {
"status": "preview",
"provider_contacted": False,
"deletion_admitted": False,
}

16
workplans/CSOC-WP-0012-durable-recovery-service-readiness.md
View file

@ -145,6 +145,22 @@ proof: docs/release/generation-retention-review-2026-10-31.{md,json}. Separate
expiry executor/caller admission, native boundaries and historical-copy
disposition remain open; the fresh review is not recurring retention enforcement.
2026-10-31 expiry-executor preparation: exact proposal6af0564e recomputes the
pinned generation selector and binds source/runtime, inventory, protected recovery
and targets. Prepared executor journals intent durably before each version-specific
mirror delete, verifies kept/protected objects before every action, and reconciles
lost replies by native listing/HEAD before retrying. Candidate ciphertext precedes
its receipt; unknown versions, primary and kit objects remain outside deletion.
Twenty-six synthetic cases and full make check441passes validate the source.
Native read-only preflight streamed/verified8,669,264mirror bytes and exact protected
version metadata with no mutation. Current candidate remains retained. Review:
docs/release/mirror-expiry-executor-review-2026-10-31.md; exact preflight/source proof:
mirror-expiry-executor-preflight-2026-10-31.{json,sources.txt}. No admitted local
expiry pair is present. Separate caller/custody admission, native boundaries,
attended execution and recurring integration remain open; no source or fixture
test is substituted for those returns. Inventory drift during a partial run
requires owner reconciliation with the journal retained.
## T02 — Activate and prove the first encrypted paired recovery point
```task