fy59/trainlog
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions

Recover terminal generation acknowledgements

Browse source
This commit is contained in:
fy59 2026-09-18 23:22:35 +02:00
parent 2c0fa714c3
commit 59673a0739
3 changed files with 44 additions and 6 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
docs/reviews/history_delete_http500_fix_v1_evidence.md
View file

@ -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.

27
tests/test_sync_error_classification.py
View file

@ -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:

19
tools/sync_peer_worker.py
View file

@ -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)