From 0750a47ceff570477265d30b573368fc8c802f81 Mon Sep 17 00:00:00 2001 From: fy59 Date: Wed, 23 Sep 2026 16:19:17 +0200 Subject: [PATCH] fix(sync): recover lost generation acknowledgements --- .../SyncGenerationForegroundCoordinator.kt | 82 +++++++++++++++++ .../data/SyncDeploymentConversationTest.kt | 67 ++++++++++++++ tests/test_sync_deployment_tools.py | 1 + tools/package_sync_candidate.py | 1 + tools/sync_generation_exchange.py | 7 +- tools/sync_peer_worker.py | 91 ++++++++++++++++++- 6 files changed, 245 insertions(+), 4 deletions(-) diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt index feb60d2..009dd5e 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt @@ -54,6 +54,11 @@ internal class SyncGenerationCoordinator( ) { private val service = SyncGenerationService(repository) + private companion object { + const val ACK_RECOVERY_CAPABILITY = "generation-ack-recovery-v1" + const val MAX_COORDINATION_BYTES = 65_536 + } + private fun phase( runId: String?, phase: String, @@ -196,6 +201,45 @@ internal class SyncGenerationCoordinator( if (actual != expected) throw SyncGenerationException("invalid $label fields") } + private fun requestSupports(request: JSONObject, capability: String): Boolean { + val capabilities = request.optJSONArray("capabilities") ?: return false + for (index in 0 until capabilities.length()) { + if (capabilities.optString(index) == capability) return true + } + return false + } + + private fun recoveryAcknowledgements( + desktopPeerId: String, + androidPeerId: String, + ): JSONArray = + repository.inSyncGenerationTransaction { db -> + val acknowledgements = JSONArray() + db.rawQuery( + "SELECT ack_json FROM sync_consumed_generations WHERE producer_peer_id=? " + + "AND consumer_peer_id=? AND result IN('consumed','rejected') " + + "ORDER BY consumed_at DESC,generation_id DESC LIMIT 32", + arrayOf(desktopPeerId, androidPeerId), + ) + .use { cursor -> + while (cursor.moveToNext()) { + val acknowledgement = JSONObject(cursor.getString(0)) + /* ACKs written before cross-runtime rejection diagnostics + * were normalized can hash '/' differently in Android + * and Python JSON canonicalizers. They remain immutable + * evidence, but must not block recovery of newer, + * compatible ACKs. Consumed ACKs are unaffected. */ + if ( + acknowledgement.optString("result") != "rejected" || + !acknowledgement.optString("diagnostic").contains("/") + ) { + acknowledgements.put(acknowledgement) + } + } + } + acknowledgements + } + private fun reconcileArchiveAcknowledgements( raw: ByteArray, runId: String, @@ -299,6 +343,7 @@ internal class SyncGenerationCoordinator( "mobile-history-v4", "execution-draft-v1", "generation-archive-v1", + ACK_RECOVERY_CAPABILITY, "session-preparations-v2", "programs-v1", ) @@ -433,6 +478,43 @@ internal class SyncGenerationCoordinator( request.getString("desktop_peer_id"), ) phase(runId, "archive_ack_observed", "desktop-archive-acknowledgements-v1.json") + if (requestSupports(request, ACK_RECOVERY_CAPABILITY)) { + /* WHY: Android may commit consumption after the desktop + * producer has already published its immutable generation, then + * transport can fail before the ACK reaches that producer. + * CONTRACT: return only exact durable consumer ACK evidence, + * bounded to the newest 32 rows, before either side captures a + * new generation. INVARIANT: recovery never fabricates lineage + * and never rewrites Android consumption history. */ + val recoveryPath = + File(directory, "android-archive-acknowledgements-v1.json") + val recoveryEnvelope = + JSONObject() + .put("format", "trainlog-sync-archive-acknowledgements") + .put("version", 1) + .put("run_id", runId) + .put("android_peer_id", peer) + .put("desktop_peer_id", request.getString("desktop_peer_id")) + .put( + "acknowledgements", + recoveryAcknowledgements( + request.getString("desktop_peer_id"), + peer, + ), + ) + .toString() + if (recoveryEnvelope.toByteArray(Charsets.UTF_8).size > MAX_COORDINATION_BYTES) + throw SyncGenerationException( + "Android archive acknowledgement envelope exceeds bound" + ) + publish(recoveryPath, recoveryEnvelope) + visibility.confirm(listOf(recoveryPath)) + phase( + runId, + "android_archive_ack_published", + recoveryPath.name, + ) + } /* WHY: transport can fail after immutable generation publication. * CONTRACT: an explicit retry for the same run republishes that * exact generation and resumes ACK handling; it never recaptures diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt index 063476f..c5bc7c0 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt @@ -56,6 +56,7 @@ class SyncDeploymentConversationTest { val databaseName = "deployment-peer-${UUID.randomUUID()}.db" var repository = TrainlogRepository(context, databaseName) var localWorker: Process? = null + var lostDesktopAckGeneration: String? = null try { transport.mkdirs() // A previous interrupted exchange leaves durable coordination @@ -180,6 +181,7 @@ class SyncDeploymentConversationTest { "full_generation_trigger_published", "generation_request_observed", "archive_ack_observed", + "android_archive_ack_published", "generation_capture_started", "generation_captured", "generation_objects_published", @@ -215,6 +217,71 @@ class SyncDeploymentConversationTest { } } } + if (suppliedTransport == null && it == 11) { + /* Reproduce the inverse deployed failure: Android has + * durably consumed the newest desktop generation, but its + * ACK was lost before the desktop producer committed it. + * The next conversation must recover that exact Android + * evidence before capturing another desktop generation. */ + SQLiteDatabase.openDatabase( + desktop.absolutePath, + null, + SQLiteDatabase.OPEN_READWRITE, + ).use { database -> + val generation = + database.rawQuery( + "SELECT generation_id FROM sync_generations " + + "WHERE status='acknowledged' ORDER BY generated_at DESC LIMIT 1", + null, + ).use { cursor -> + assertTrue(cursor.moveToFirst()) + cursor.getString(0) + } + database.delete( + "sync_acknowledgements", + "generation_id=?", + arrayOf(generation), + ) + database.execSQL( + "UPDATE sync_generations SET status='waiting_acknowledgement'," + + "acknowledged_at=NULL WHERE generation_id=?", + arrayOf(generation), + ) + database.rawQuery( + "SELECT status FROM sync_generations WHERE generation_id=?", + arrayOf(generation), + ).use { cursor -> + assertTrue(cursor.moveToFirst()) + assertEquals("waiting_acknowledgement", cursor.getString(0)) + } + lostDesktopAckGeneration = generation + } + } + if (suppliedTransport == null && it == 12) { + val recovered = checkNotNull(lostDesktopAckGeneration) + SQLiteDatabase.openDatabase( + desktop.absolutePath, + null, + SQLiteDatabase.OPEN_READONLY, + ).use { database -> + database.rawQuery( + "SELECT status FROM sync_generations WHERE generation_id=?", + arrayOf(recovered), + ).use { cursor -> + assertTrue(cursor.moveToFirst()) + assertEquals("acknowledged", cursor.getString(0)) + } + database.rawQuery( + "SELECT parent_generation_id,status FROM sync_generations " + + "ORDER BY generated_at DESC LIMIT 1", + null, + ).use { cursor -> + assertTrue(cursor.moveToFirst()) + assertEquals(recovered, cursor.getString(0)) + assertEquals("acknowledged", cursor.getString(1)) + } + } + } localWorker = null } repository.close() diff --git a/tests/test_sync_deployment_tools.py b/tests/test_sync_deployment_tools.py index 5b8d60f..829b6cf 100644 --- a/tests/test_sync_deployment_tools.py +++ b/tests/test_sync_deployment_tools.py @@ -78,6 +78,7 @@ class DeploymentToolsTest(unittest.TestCase): self.assertIn("heart-rate-v1", inventory["protocols"]) self.assertIn("session-timeline-v1", inventory["protocols"]) self.assertIn("bluetooth-files-v1", inventory["protocols"]) + self.assertIn("generation-ack-recovery-v1", inventory["protocols"]) self.assertTrue((output / "bin/trainlog").is_file()) self.assertTrue((output / "bin/trainlog-sync-once").is_file()) self.assertTrue((output / "bin/trainlog-syncd").is_file()) diff --git a/tools/package_sync_candidate.py b/tools/package_sync_candidate.py index d3840de..a3ee020 100644 --- a/tools/package_sync_candidate.py +++ b/tools/package_sync_candidate.py @@ -129,6 +129,7 @@ def main() -> int: "generation-manifest-v1", "generation-ack-v1", "generation-archive-v1", + "generation-ack-recovery-v1", "bluetooth-files-v1", "session-preparations-v2", "ai-session-drafts-v2", diff --git a/tools/sync_generation_exchange.py b/tools/sync_generation_exchange.py index 74f2995..a8eeae2 100644 --- a/tools/sync_generation_exchange.py +++ b/tools/sync_generation_exchange.py @@ -791,8 +791,7 @@ def validate_ack(raw: bytes, pending: sqlite3.Row | tuple) -> dict: return value -def accept_ack(database: Path, path: Path) -> str: - raw = path.read_bytes() +def accept_ack_bytes(database: Path, raw: bytes) -> str: with closing(connect_database(database)) as db: require_schema(db); db.execute("BEGIN IMMEDIATE") value = strict_json(raw, MAX_MANIFEST) @@ -862,6 +861,10 @@ def accept_ack(database: Path, path: Path) -> str: db.commit(); return status +def accept_ack(database: Path, path: Path) -> str: + return accept_ack_bytes(database, path.read_bytes()) + + def main() -> None: parser = argparse.ArgumentParser() sub = parser.add_subparsers(dest="mode", required=True) diff --git a/tools/sync_peer_worker.py b/tools/sync_peer_worker.py index be87971..3e031a3 100644 --- a/tools/sync_peer_worker.py +++ b/tools/sync_peer_worker.py @@ -35,6 +35,8 @@ CAPS = { "session-preparations-v2", } +ACK_RECOVERY_CAP = "generation-ack-recovery-v1" + MTP_OPERATION_TIMEOUT_SECONDS = 30.0 MTP_POLL_INTERVAL_SECONDS = 3.0 @@ -115,6 +117,67 @@ def recovery_acknowledgements(db, producer_peer_id: str, consumer_peer_id: str) ] +def recovery_acknowledgement_envelope( + raw: bytes, + run_id: str, + android_peer_id: str, + desktop_peer_id: str, +) -> list[dict]: + """Validate the Android consumer's bounded ACK recovery envelope.""" + value = generation.strict_json(raw, 65536) + expected = { + "format", + "version", + "run_id", + "android_peer_id", + "desktop_peer_id", + "acknowledgements", + } + if ( + not isinstance(value, dict) + or set(value) != expected + or value.get("format") != "trainlog-sync-archive-acknowledgements" + or type(value.get("version")) is not int + or value.get("version") != 1 + or value.get("run_id") != run_id + or value.get("android_peer_id") != android_peer_id + or value.get("desktop_peer_id") != desktop_peer_id + ): + raise RuntimeError("Android archive acknowledgement envelope is not correlated") + acknowledgements = value.get("acknowledgements") + if not isinstance(acknowledgements, list) or len(acknowledgements) > 32: + raise RuntimeError("invalid Android archive acknowledgement count") + if not all(isinstance(ack, dict) for ack in acknowledgements): + raise RuntimeError("invalid Android archive acknowledgement") + return acknowledgements + + +def reconcile_recovery_acknowledgements( + database: Path, + raw: bytes, + run_id: str, + android_peer_id: str, + desktop_peer_id: str, +) -> int: + """Repair producer state only from exact ACK evidence retained by Android.""" + acknowledgements = recovery_acknowledgement_envelope( + raw, + run_id, + android_peer_id, + desktop_peer_id, + ) + reconciled = 0 + for acknowledgement in acknowledgements: + result = generation.accept_ack_bytes( + database, + generation.canonical(acknowledgement), + ) + if result not in ("acknowledged", "rejected", "unchanged"): + raise RuntimeError("unexpected recovered ACK result") + reconciled += 1 + return reconciled + + def run_adapter( adapter: Path, operation: str, @@ -630,7 +693,7 @@ def main() -> int: "run_id": args.run_id, "desktop_peer_id": desktop_peer, "android_peer_id": peer["peer_id"], - "capabilities": sorted(CAPS), + "capabilities": sorted(CAPS | {ACK_RECOVERY_CAP}), } request_path = args.transport_root / "request-v1.json" publish_json(request_path, request) @@ -678,12 +741,36 @@ def main() -> int: producer_peer_id=peer["peer_id"], consumer_peer_id=desktop_peer, ) - inbound_ref = args.transport_root / "android-generation-v1.json" + def pump_inbound(): if pump is not None: pump() correlated_peer_error(args.transport_root, args.run_id) + if ACK_RECOVERY_CAP in set(peer.get("capabilities", [])): + android_archive_ack = args.transport_root / "android-archive-acknowledgements-v1.json" + wait_json( + android_archive_ack, + deadline, + lambda value: value.get("run_id") == args.run_id + and value.get("android_peer_id") == peer["peer_id"] + and value.get("desktop_peer_id") == desktop_peer, + pump_inbound, + ) + recovered = reconcile_recovery_acknowledgements( + args.database, + bounded(android_archive_ack), + args.run_id, + peer["peer_id"], + desktop_peer, + ) + emit( + args.run_id, + "android_archive_ack_observed", + recovered_acknowledgements=recovered, + ) + + inbound_ref = args.transport_root / "android-generation-v1.json" inbound = wait_json( inbound_ref, deadline, lambda value: value.get("run_id") == args.run_id, pump_inbound )