444 lines
21 KiB
Python
444 lines
21 KiB
Python
#!/usr/bin/env python3
|
|
"""Recover one unpublished local Sleep branch from an immutable peer collision."""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import re
|
|
import sqlite3
|
|
import uuid
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Callable
|
|
|
|
import sleep_diary_exchange as exchange
|
|
|
|
|
|
REVISION_ID = re.compile(r"slr_[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}")
|
|
|
|
|
|
class RecoveryError(ValueError):
|
|
"""A collision cannot be recovered without changing established facts."""
|
|
|
|
|
|
def _new_revision_id() -> str:
|
|
return f"slr_{uuid.uuid4()}"
|
|
|
|
|
|
def _canonical_digest(value: object) -> str:
|
|
payload = json.dumps(value, ensure_ascii=False, sort_keys=True,
|
|
separators=(",", ":")).encode("utf-8")
|
|
return hashlib.sha256(payload).hexdigest()
|
|
|
|
|
|
def _incoming_entry(document: dict, collision_id: str) -> dict:
|
|
matches = [item for item in document["entries"] if item["revision_id"] == collision_id]
|
|
if len(matches) != 1:
|
|
raise RecoveryError("collision revision is not unique in incoming artifact")
|
|
return matches[0]
|
|
|
|
|
|
def _merge_scalar(local: object, remote: object, label: str) -> object:
|
|
if local == remote:
|
|
return local
|
|
if local is None:
|
|
return remote
|
|
if remote is None:
|
|
return local
|
|
raise RecoveryError(f"human decision required for {label}")
|
|
|
|
|
|
def _merge_identity_list(local: list[dict], remote: list[dict], identity: str,
|
|
label: str) -> list[dict]:
|
|
merged = {value[identity]: value for value in local}
|
|
for value in remote:
|
|
existing = merged.get(value[identity])
|
|
if existing is not None and existing != value:
|
|
raise RecoveryError(f"{label} identity reused with different content")
|
|
merged[value[identity]] = value
|
|
return [merged[key] for key in sorted(merged)]
|
|
|
|
|
|
def _copy_local_revision(db: sqlite3.Connection, entry_id: str, source_id: str,
|
|
target_id: str) -> None:
|
|
db.execute(
|
|
"INSERT INTO sleep_diary_revisions "
|
|
"SELECT ?,entry_id,parent_revision_id,created_at,sleep_quality,wake_quality,day_form,"
|
|
"treatment_and_notes FROM sleep_diary_revisions WHERE revision_id=? AND entry_id=?",
|
|
(target_id, source_id, entry_id),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO sleep_diary_events "
|
|
"SELECT ?,event_id,event_type,start_at,end_at FROM sleep_diary_events WHERE revision_id=?",
|
|
(target_id, source_id),
|
|
)
|
|
db.execute(
|
|
"INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id,"
|
|
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) "
|
|
"SELECT ?,intake_id,medication_id,medication_name,taken_at,dose_value,dose_unit,"
|
|
"quantity,note,created_at FROM sleep_medication_intakes WHERE revision_id=?",
|
|
(target_id, source_id),
|
|
)
|
|
|
|
|
|
def _insert_revision(db: sqlite3.Connection, item: dict, revision_id: str,
|
|
parent_revision_id: str | None, created_at: str) -> None:
|
|
db.execute(
|
|
"INSERT INTO sleep_diary_revisions VALUES(?,?,?,?,?,?,?,?)",
|
|
(revision_id, item["entry_id"], parent_revision_id, created_at,
|
|
item["sleep_quality"], item["wake_quality"], item["day_form"],
|
|
item["treatment_and_notes"]),
|
|
)
|
|
db.executemany(
|
|
"INSERT INTO sleep_diary_events VALUES(?,?,?,?,?)",
|
|
[(revision_id, value["event_id"], value["type"], value["start_at"], value["end_at"])
|
|
for value in item["events"]],
|
|
)
|
|
db.executemany(
|
|
"INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id,"
|
|
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) "
|
|
"VALUES(?,?,?,?,?,?,?,?,?,?)",
|
|
[(revision_id, value["intake_id"], value["medication_id"],
|
|
value["medication_name"], value["taken_at"], value["dose_value"],
|
|
value["dose_unit"], value["quantity"], value["note"], value["created_at"])
|
|
for value in item["intakes"]],
|
|
)
|
|
|
|
|
|
def recover(db: sqlite3.Connection, document: dict, collision_id: str, recovered_at: str,
|
|
id_factory: Callable[[], str] = _new_revision_id,
|
|
fail_after_reidentification: bool = False,
|
|
rejected_generation: tuple[dict, str] | None = None) -> dict:
|
|
"""Preserve both branches and advance through one non-conflicting successor.
|
|
|
|
CONTRACT: only a local current revision that has never crossed an immutable
|
|
publication boundary may be re-identified. The incoming revision retains its
|
|
published identity and exact payload. A fresh successor descends from that
|
|
published revision, while the re-identified local side branch remains durable.
|
|
INVARIANT: every mutation commits in one IMMEDIATE transaction or none does.
|
|
"""
|
|
exchange.validate(document)
|
|
if document["version"] != 2 or not REVISION_ID.fullmatch(collision_id):
|
|
raise RecoveryError("Sleep Diary V2 collision identity required")
|
|
incoming = _incoming_entry(document, collision_id)
|
|
incoming_payload = exchange.incoming_revision_payload(incoming, 2)
|
|
owner = db.execute(
|
|
"SELECT entry_id FROM sleep_diary_revisions WHERE revision_id=?", (collision_id,)
|
|
).fetchone()
|
|
if owner is None or owner[0] != incoming["entry_id"]:
|
|
raise RecoveryError("local collision revision is unavailable")
|
|
entry_id = owner[0]
|
|
current = db.execute(
|
|
"SELECT night_start_date,night_end_date,created_at,updated_at,current_revision_id,deleted "
|
|
"FROM sleep_diary_entries WHERE entry_id=?", (entry_id,)
|
|
).fetchone()
|
|
if current is None or current[4] != collision_id:
|
|
raise RecoveryError("local collision revision is not the current tip")
|
|
local_payload = exchange.persisted_revision_payload(db, entry_id, collision_id)
|
|
if local_payload == incoming_payload:
|
|
raise RecoveryError("revision is already an identical replay")
|
|
if tuple(current[:3]) != (incoming["night_start_date"], incoming["night_end_date"],
|
|
incoming["created_at"]) or bool(current[5]) != incoming["deleted"]:
|
|
raise RecoveryError("entry envelope requires a human decision")
|
|
if db.execute(
|
|
"SELECT 1 FROM sleep_diary_revisions WHERE parent_revision_id=? LIMIT 1", (collision_id,)
|
|
).fetchone() is not None:
|
|
raise RecoveryError("local collision revision already has descendants")
|
|
publication = db.execute(
|
|
"SELECT validated_revision_id,acknowledged_revision_id FROM sleep_diary_publication_state "
|
|
"WHERE entry_id=?", (entry_id,)
|
|
).fetchone()
|
|
if publication is not None and collision_id in publication:
|
|
raise RecoveryError("local collision revision already crossed publication state")
|
|
try:
|
|
immutable_reference = db.execute(
|
|
"SELECT 1 FROM sleep_diary_generation_entries WHERE entry_id=? AND revision_id=? LIMIT 1",
|
|
(entry_id, collision_id),
|
|
).fetchone()
|
|
except sqlite3.OperationalError:
|
|
immutable_reference = None
|
|
if immutable_reference is not None:
|
|
raise RecoveryError("local collision revision is referenced by an immutable generation")
|
|
if incoming["parent_revision_id"] is not None and db.execute(
|
|
"SELECT 1 FROM sleep_diary_revisions WHERE revision_id=? AND entry_id=?",
|
|
(incoming["parent_revision_id"], entry_id),
|
|
).fetchone() is None:
|
|
raise RecoveryError("incoming causal parent is unavailable locally")
|
|
|
|
merged = dict(incoming)
|
|
merged["sleep_quality"] = _merge_scalar(
|
|
local_payload["sleep_quality"], incoming["sleep_quality"], "sleep_quality"
|
|
)
|
|
merged["wake_quality"] = _merge_scalar(
|
|
local_payload["wake_quality"], incoming["wake_quality"], "wake_quality"
|
|
)
|
|
merged["day_form"] = _merge_scalar(
|
|
local_payload["day_form"], incoming["day_form"], "day_form"
|
|
)
|
|
if local_payload["treatment_and_notes"] != incoming["treatment_and_notes"]:
|
|
raise RecoveryError("human decision required for treatment_and_notes")
|
|
merged["events"] = _merge_identity_list(
|
|
local_payload["events"], incoming["events"], "event_id", "sleep event"
|
|
)
|
|
merged["intakes"] = _merge_identity_list(
|
|
local_payload["intakes"], incoming["intakes"], "intake_id", "medication intake"
|
|
)
|
|
|
|
local_revision_id = id_factory()
|
|
reconciliation_revision_id = id_factory()
|
|
if (not REVISION_ID.fullmatch(local_revision_id) or
|
|
not REVISION_ID.fullmatch(reconciliation_revision_id) or
|
|
len({collision_id, local_revision_id, reconciliation_revision_id}) != 3):
|
|
raise RecoveryError("revision identity generator returned an invalid or duplicate id")
|
|
|
|
db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
_copy_local_revision(db, entry_id, collision_id, local_revision_id)
|
|
changed = db.execute(
|
|
"UPDATE sleep_diary_entries SET current_revision_id=? "
|
|
"WHERE entry_id=? AND current_revision_id=?",
|
|
(local_revision_id, entry_id, collision_id),
|
|
).rowcount
|
|
if changed != 1:
|
|
raise RecoveryError("guarded local branch re-identification failed")
|
|
if fail_after_reidentification:
|
|
raise RecoveryError("injected recovery failure")
|
|
db.execute("DELETE FROM sleep_diary_events WHERE revision_id=?", (collision_id,))
|
|
db.execute("DELETE FROM sleep_medication_intakes WHERE revision_id=?", (collision_id,))
|
|
db.execute("DELETE FROM sleep_diary_revisions WHERE revision_id=?", (collision_id,))
|
|
_insert_revision(db, incoming, collision_id, incoming["parent_revision_id"],
|
|
incoming["updated_at"])
|
|
merged["updated_at"] = recovered_at
|
|
_insert_revision(db, merged, reconciliation_revision_id, collision_id, recovered_at)
|
|
changed = db.execute(
|
|
"UPDATE sleep_diary_entries SET updated_at=?,current_revision_id=? "
|
|
"WHERE entry_id=? AND current_revision_id=?",
|
|
(recovered_at, reconciliation_revision_id, entry_id, local_revision_id),
|
|
).rowcount
|
|
if changed != 1:
|
|
raise RecoveryError("guarded reconciliation advance failed")
|
|
rejection_rearmed = False
|
|
if rejected_generation is not None:
|
|
rejection_rearmed = _rearm_desktop_rejection(
|
|
db, rejected_generation[0], rejected_generation[1]
|
|
)
|
|
if db.execute("PRAGMA foreign_key_check").fetchone() is not None:
|
|
raise RecoveryError("foreign key check failed during recovery")
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
merged_payload = dict(merged)
|
|
merged_payload["revision_id"] = reconciliation_revision_id
|
|
merged_payload["parent_revision_id"] = collision_id
|
|
return {
|
|
"entry_id": entry_id,
|
|
"collision_revision_id": collision_id,
|
|
"local_reidentified_revision_id": local_revision_id,
|
|
"reconciliation_revision_id": reconciliation_revision_id,
|
|
"reconciliation_parent_revision_id": collision_id,
|
|
"local_payload_sha256": _canonical_digest(local_payload),
|
|
"incoming_payload_sha256": _canonical_digest(incoming_payload),
|
|
"merged_payload_sha256": _canonical_digest(
|
|
exchange.incoming_revision_payload(merged_payload, 2)
|
|
),
|
|
"event_count": len(merged["events"]),
|
|
"intake_count": len(merged["intakes"]),
|
|
"desktop_rejection_rearmed": rejection_rearmed,
|
|
}
|
|
|
|
|
|
def _rearm_desktop_rejection(db: sqlite3.Connection, manifest: dict,
|
|
manifest_digest: str) -> bool:
|
|
generation_id = manifest["generation_id"]
|
|
row = db.execute(
|
|
"SELECT run_id,producer_peer_id,consumer_peer_id,parent_generation_id,manifest_sha256,"
|
|
"result,durability,diagnostic FROM sync_consumed_generations WHERE generation_id=?",
|
|
(generation_id,),
|
|
).fetchone()
|
|
if row is None:
|
|
return False
|
|
expected = (
|
|
manifest["run_id"], manifest["producer"]["peer_id"], manifest["consumer_peer_id"],
|
|
manifest["parent_generation_id"], manifest_digest, "rejected", "sqlite-commit-rejection",
|
|
)
|
|
if tuple(row[:7]) != expected:
|
|
raise RecoveryError("generation rejection ledger does not match immutable manifest")
|
|
# The deployed consumer could lose ValueError text while recording this
|
|
# exact collision (StopIteration has an empty message). The collision has
|
|
# already been proven directly against the manifest-owned Sleep artifact.
|
|
if row[7] not in (
|
|
"",
|
|
"sleep diary revision identity reused with different content",
|
|
"schema desktop v8 à v35 requis",
|
|
):
|
|
raise RecoveryError("generation rejection has an unrelated diagnostic")
|
|
if db.execute(
|
|
"SELECT 1 FROM sync_consumed_generations WHERE parent_generation_id=? LIMIT 1",
|
|
(generation_id,),
|
|
).fetchone() is not None:
|
|
raise RecoveryError("rejected generation already has a durable descendant")
|
|
ack = db.execute(
|
|
"SELECT ack_id,result,durability,manifest_sha256 FROM sync_acknowledgements "
|
|
"WHERE generation_id=?", (generation_id,),
|
|
).fetchall()
|
|
expected_ack_id = "ack_" + generation_id.removeprefix("gen_")
|
|
if ack != [(expected_ack_id, "rejected", "sqlite-commit-rejection", manifest_digest)]:
|
|
raise RecoveryError("generation rejection ACK does not match immutable manifest")
|
|
db.execute("DELETE FROM sync_acknowledgements WHERE generation_id=?", (generation_id,))
|
|
db.execute("DELETE FROM sync_consumed_generations WHERE generation_id=?", (generation_id,))
|
|
return True
|
|
|
|
|
|
def rearm_desktop_consumer(db: sqlite3.Connection, document: dict, manifest: dict,
|
|
manifest_digest: str, collision_id: str) -> dict:
|
|
"""Rearm an exact retry after recovery or a recovery-adjacent schema rejection.
|
|
|
|
CONTRACT: the published collision identity must already contain the exact
|
|
manifest-owned payload. Only the matching rejected consumption and ACK are
|
|
removed; immutable generation bytes and recovered Sleep history are unchanged.
|
|
"""
|
|
exchange.validate(document)
|
|
incoming = _incoming_entry(document, collision_id)
|
|
persisted = exchange.persisted_revision_payload(db, incoming["entry_id"], collision_id)
|
|
if persisted != exchange.incoming_revision_payload(incoming, 2):
|
|
raise RecoveryError("recovered collision payload no longer matches immutable artifact")
|
|
db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
rearmed = _rearm_desktop_rejection(db, manifest, manifest_digest)
|
|
if not rearmed:
|
|
raise RecoveryError("desktop has no matching rejected generation to rearm")
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return {
|
|
"generation_id": manifest["generation_id"],
|
|
"manifest_sha256": manifest_digest,
|
|
"status": "ready_for_retry",
|
|
"immutable_artifacts_rewritten": False,
|
|
}
|
|
|
|
|
|
def rearm_android_producer(db: sqlite3.Connection, manifest: dict, manifest_digest: str,
|
|
collision_id: str) -> dict:
|
|
"""Rearm one rejected immutable Android generation after causal recovery.
|
|
|
|
CONTRACT: manifest bytes and generation-to-Sleep mapping remain unchanged.
|
|
Only the exact rejection ACK is removed and its producer state returns to
|
|
waiting_acknowledgement so the normal retry can record a consumed ACK.
|
|
"""
|
|
generation_id = manifest["generation_id"]
|
|
db.execute("BEGIN IMMEDIATE")
|
|
try:
|
|
row = db.execute(
|
|
"SELECT run_id,producer_peer_id,consumer_peer_id,generated_at,parent_generation_id,"
|
|
"manifest_sha256,status FROM sync_generations WHERE generation_id=?",
|
|
(generation_id,),
|
|
).fetchone()
|
|
expected = (
|
|
manifest["run_id"], manifest["producer"]["peer_id"], manifest["consumer_peer_id"],
|
|
manifest["generated_at"], manifest["parent_generation_id"], manifest_digest, "rejected",
|
|
)
|
|
if row is None or tuple(row) != expected:
|
|
raise RecoveryError("Android rejected generation does not match immutable manifest")
|
|
mapping = db.execute(
|
|
"SELECT revision_id FROM sleep_diary_generation_entries WHERE generation_id=? "
|
|
"AND entry_id=(SELECT entry_id FROM sleep_diary_revisions WHERE revision_id=?)",
|
|
(generation_id, collision_id),
|
|
).fetchone()
|
|
if mapping != (collision_id,):
|
|
raise RecoveryError("Android generation does not own the collided Sleep revision")
|
|
ack = db.execute(
|
|
"SELECT ack_id,result,durability,manifest_sha256 FROM sync_acknowledgements "
|
|
"WHERE generation_id=?", (generation_id,),
|
|
).fetchall()
|
|
expected_ack_id = "ack_" + generation_id.removeprefix("gen_")
|
|
if ack != [(expected_ack_id, "rejected", "sqlite-commit-rejection", manifest_digest)]:
|
|
raise RecoveryError("Android rejection ACK does not match immutable manifest")
|
|
db.execute("DELETE FROM sync_acknowledgements WHERE generation_id=?", (generation_id,))
|
|
changed = db.execute(
|
|
"UPDATE sync_generations SET status='waiting_acknowledgement',acknowledged_at=NULL "
|
|
"WHERE generation_id=? AND status='rejected'", (generation_id,),
|
|
).rowcount
|
|
if changed != 1:
|
|
raise RecoveryError("Android generation rearm guard failed")
|
|
if db.execute("PRAGMA foreign_key_check").fetchone() is not None:
|
|
raise RecoveryError("foreign key check failed during Android rearm")
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return {
|
|
"generation_id": generation_id,
|
|
"manifest_sha256": manifest_digest,
|
|
"status": "waiting_acknowledgement",
|
|
"immutable_artifacts_rewritten": False,
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser()
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
desktop = subparsers.add_parser("recover-desktop")
|
|
desktop.add_argument("--database", required=True, type=Path)
|
|
desktop.add_argument("--generation-directory", required=True, type=Path)
|
|
desktop.add_argument("--collision-id", required=True)
|
|
desktop.add_argument("--recovered-at", default=None)
|
|
desktop.add_argument("--report", type=Path)
|
|
desktop_retry = subparsers.add_parser("rearm-desktop")
|
|
desktop_retry.add_argument("--database", required=True, type=Path)
|
|
desktop_retry.add_argument("--generation-directory", required=True, type=Path)
|
|
desktop_retry.add_argument("--collision-id", required=True)
|
|
desktop_retry.add_argument("--report", type=Path)
|
|
android = subparsers.add_parser("rearm-android")
|
|
android.add_argument("--database", required=True, type=Path)
|
|
android.add_argument("--generation-directory", required=True, type=Path)
|
|
android.add_argument("--collision-id", required=True)
|
|
android.add_argument("--report", type=Path)
|
|
args = parser.parse_args()
|
|
import sync_generation_exchange as generation
|
|
|
|
manifest_path = args.generation_directory / "manifest.json"
|
|
preliminary = json.loads(manifest_path.read_text(encoding="utf-8"))
|
|
manifest, manifest_digest = generation.validate_published(
|
|
args.generation_directory, preliminary["consumer_peer_id"]
|
|
)
|
|
sleep_descriptor = next(
|
|
(value for value in manifest["artifacts"] if value["logical_name"] == "sleep-diary"),
|
|
None,
|
|
)
|
|
if sleep_descriptor is None:
|
|
raise RecoveryError("generation has no Sleep Diary artifact")
|
|
artifact = args.generation_directory / Path(sleep_descriptor["filename"]).name
|
|
document = exchange.load(artifact)
|
|
with exchange.connect_database(args.database) as db:
|
|
version = db.execute("PRAGMA user_version").fetchone()[0]
|
|
if args.command == "recover-desktop":
|
|
if version != 35:
|
|
raise RecoveryError("desktop schema v35 pre-immutability recovery required")
|
|
recovered_at = args.recovered_at or datetime.now().astimezone().isoformat()
|
|
report = recover(
|
|
db, document, args.collision_id, recovered_at,
|
|
rejected_generation=(manifest, manifest_digest),
|
|
)
|
|
elif args.command == "rearm-desktop":
|
|
if version != 36:
|
|
raise RecoveryError("desktop schema v36 post-recovery retry required")
|
|
report = rearm_desktop_consumer(
|
|
db, document, manifest, manifest_digest, args.collision_id
|
|
)
|
|
else:
|
|
if version not in (32, 33):
|
|
raise RecoveryError("Android schema v32 or v33 required")
|
|
report = rearm_android_producer(
|
|
db, manifest, manifest_digest, args.collision_id
|
|
)
|
|
encoded = json.dumps(report, ensure_ascii=False, sort_keys=True, indent=2) + "\n"
|
|
if args.report is not None:
|
|
args.report.write_text(encoded, encoding="utf-8")
|
|
print(encoded, end="")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|