diff --git a/docs/reviews/history_delete_http500_fix_v1_evidence.md b/docs/reviews/history_delete_http500_fix_v1_evidence.md index 36a1f3b..2d5c8ed 100644 --- a/docs/reviews/history_delete_http500_fix_v1_evidence.md +++ b/docs/reviews/history_delete_http500_fix_v1_evidence.md @@ -53,3 +53,7 @@ production full-generation consumer now filters only live facts already dominated by causal state applied from that complete envelope; standalone snapshot imports retain their strict rejection. A regression exercises this ordering and proves the tombstone and absent session remain unchanged. +The recovery envelope also republishes both consumed and rejected correlated +ACKs. This lets Android close an interrupted generation with the real durable +desktop rejection evidence and reclaim normal bounded capacity; no generation, +ACK, tombstone or archive is manually purged or synthesized. diff --git a/tests/test_sync_error_classification.py b/tests/test_sync_error_classification.py index 21bc875..1645edd 100644 --- a/tests/test_sync_error_classification.py +++ b/tests/test_sync_error_classification.py @@ -1,4 +1,5 @@ import json +import sqlite3 import subprocess import sys import tempfile @@ -15,6 +16,32 @@ import sync_peer_worker # noqa: E402 class SyncErrorClassificationTest(unittest.TestCase): + def test_recovery_envelope_includes_consumed_and_rejected_acknowledgements(self): + with sqlite3.connect(":memory:") as database: + database.execute( + "CREATE TABLE sync_consumed_generations(" + "generation_id TEXT,producer_peer_id TEXT,consumer_peer_id TEXT," + "consumed_at TEXT,result TEXT,ack_json TEXT)" + ) + for generation, result in (("gen_consumed", "consumed"), + ("gen_rejected", "rejected"), + ("gen_other", "consumed")): + database.execute( + "INSERT INTO sync_consumed_generations VALUES(?,?,?,?,?,?)", + (generation, "android" if generation != "gen_other" else "other", + "desktop", generation, result, + json.dumps({"generation_id": generation, "result": result})), + ) + + acknowledgements = sync_peer_worker.recovery_acknowledgements( + database, "android", "desktop" + ) + + self.assertEqual( + [(value["generation_id"], value["result"]) for value in acknowledgements], + [("gen_consumed", "consumed"), ("gen_rejected", "rejected")], + ) + def test_mtp_refreshes_a_retained_peer_advertisement_before_validation(self): expected_peer = "peer_00000000-0000-4000-8000-000000000001" with tempfile.TemporaryDirectory() as directory: diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py index 1c2ec0a..4f63522 100644 --- a/tools/sync_peer_worker.py +++ b/tools/sync_peer_worker.py @@ -67,6 +67,17 @@ def publish_json(path: Path, value: dict) -> None: os.replace(temporary, path) +def recovery_acknowledgements(db, producer_peer_id: str, consumer_peer_id: str) -> list[dict]: + """Load bounded terminal ACK evidence for interrupted peer conversations.""" + rows = db.execute( + "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? " + "AND consumer_peer_id=? AND result IN('consumed','rejected') " + "ORDER BY consumed_at,generation_id LIMIT 32", + (producer_peer_id, consumer_peer_id), + ).fetchall() + return [json.loads(row[0]) for row in rows] + + def run_adapter( adapter: Path, operation: str, @@ -309,18 +320,14 @@ def main() -> int: request_path = args.transport_root / "request-v1.json" publish_json(request_path, request) with closing(generation.connect_database(args.database)) as db: - acknowledgement_rows = db.execute( - "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? " - "AND consumer_peer_id=? AND result='consumed' ORDER BY consumed_at,generation_id LIMIT 32", - (peer["peer_id"], desktop_peer), - ).fetchall() + acknowledgements = recovery_acknowledgements(db, peer["peer_id"], desktop_peer) archive_acknowledgements = { "format": "trainlog-sync-archive-acknowledgements", "version": 1, "run_id": args.run_id, "android_peer_id": peer["peer_id"], "desktop_peer_id": desktop_peer, - "acknowledgements": [json.loads(row[0]) for row in acknowledgement_rows], + "acknowledgements": acknowledgements, } archive_ack_path = args.transport_root / "desktop-archive-acknowledgements-v1.json" publish_json(archive_ack_path, archive_acknowledgements)