e5e2965ac456
chore: Trace sync protocol receive stages (v2.5.18)
- time
- 2026-07-27 18:09 UTC · 2 months ago
- machine
- koh2
- agent
- codex-gpt-5
- session
- peach-sync-e2e
- parent
- ec914e23f6d4
↓ peach-android-e5e2965a.tar.gzatlas at this save
files changed3 files
app/build.gradle+2 −2app/src/main/java/com/peachpasswords/sync/SyncCoordinator.kt+37 −17app/src/test/java/com/peachpasswords/sync/SyncCoordinatorTest.kt+15 −3app/build.gradle+2 −2
@@ -16,8 +16,8 @@1616 applicationId "com.peachpasswords"1717 minSdk 261818 targetSdk 3519- versionCode 24011720- versionName "2.5.17"19+ versionCode 24011820+ versionName "2.5.18"2121 2222 testInstrumentationRunner "androidx.test.runner.AndroidJUnitRunner"2323 vectorDrawables {app/src/main/java/com/peachpasswords/sync/SyncCoordinator.kt+37 −17
@@ -105,7 +105,7 @@105105 session.sendMsg(106106 SyncMessage.Hello(ourVaultId, vault.getOurDeviceId(), localVersion, nowMs),107107 )108- val hello = session.recvAs<SyncMessage.Hello>("SYNC_HELLO")108+ val hello = session.recvAs<SyncMessage.Hello>("initiator", "SYNC_HELLO")109109 if (hello.vaultId != ourVaultId) {110110 session.sendMsg(SyncMessage.Error("vault_id_mismatch"))111111 return@withTimeout SyncOutcome(SyncOutcomeStatus.VAULT_MISMATCH)@@ -114,25 +114,25 @@114114 115115 if (hello.version == localVersion) {116116 session.sendMsg(SyncMessage.Done(0, 0))117- session.recvAs<SyncMessage.Done>("SYNC_DONE")117+ session.recvAs<SyncMessage.Done>("initiator", "SYNC_DONE")118118 return@withTimeout SyncOutcome(SyncOutcomeStatus.COMPLETED)119119 }120120 121121 val ourEntries = vault.buildEntryManifest()122122 val ourTombstones = vault.buildTombstoneManifest()123123 session.sendMsg(SyncMessage.Manifest(ourEntries, ourTombstones))124- val peerManifest = session.recvAs<SyncMessage.Manifest>("SYNC_MANIFEST")124+ val peerManifest = session.recvAs<SyncMessage.Manifest>("initiator", "SYNC_MANIFEST")125125 126126 val wantFromPeer = diffNewer(peerManifest.entries, ourEntries)127127 session.sendMsg(SyncMessage.Request(wantFromPeer))128- val peerRequest = session.recvAs<SyncMessage.Request>("SYNC_REQUEST")128+ val peerRequest = session.recvAs<SyncMessage.Request>("initiator", "SYNC_REQUEST")129129 130130 val ourPayload = vault.getEncryptedEntries(peerRequest.entryIds)131131 session.sendMsg(SyncMessage.Entries(ourPayload))132132 entriesSent = ourPayload.size133133 134134 if (wantFromPeer.isNotEmpty()) {135- val incoming = session.recvAs<SyncMessage.Entries>("SYNC_ENTRIES")135+ val incoming = session.recvAs<SyncMessage.Entries>("initiator", "SYNC_ENTRIES")136136 vault.applyEntries(incoming.entries)137137 entriesReceived = incoming.entries.size138138 }@@ -141,13 +141,19 @@141141 session.sendMsg(SyncMessage.Tombstones(newerForPeer))142142 tombstonesSent = newerForPeer.size143143 144- val peerTombstones = session.recvAs<SyncMessage.Tombstones>("SYNC_TOMBSTONES")144+ val peerTombstones = session.recvAs<SyncMessage.Tombstones>(145+ "initiator",146+ "SYNC_TOMBSTONES",147+ )145148 vault.applyTombstones(peerTombstones.tombstones)146149 tombstonesReceived = peerTombstones.tombstones.size147150 148151 if (vault.supportsDeviceManifest) {149152 session.sendMsg(SyncMessage.DeviceManifest(vault.buildOutgoingDeviceManifest()))150- val theirs = session.recvAs<SyncMessage.DeviceManifest>("SYNC_DEVICE_MANIFEST")153+ val theirs = session.recvAs<SyncMessage.DeviceManifest>(154+ "initiator",155+ "SYNC_DEVICE_MANIFEST",156+ )151157 if (theirs.manifest != null) {152158 vault.applyIncomingDeviceManifest(153159 theirs.manifest,@@ -157,7 +163,7 @@157163 }158164 159165 session.sendMsg(SyncMessage.Done(entriesReceived, entriesSent))160- session.recvAs<SyncMessage.Done>("SYNC_DONE")166+ session.recvAs<SyncMessage.Done>("initiator", "SYNC_DONE")161167 SyncOutcome(162168 SyncOutcomeStatus.COMPLETED,163169 entriesReceived, entriesSent, tombstonesReceived, tombstonesSent,@@ -192,7 +198,7 @@192198 val ourVaultId = vault.computeVaultId()193199 val localVersion = vault.getLocalVersion()194200 195- val hello = session.recvAs<SyncMessage.Hello>("SYNC_HELLO")201+ val hello = session.recvAs<SyncMessage.Hello>("responder", "SYNC_HELLO")196202 // Always answer HELLO so the initiator's recv completes, even on a197203 // mismatch where we bail immediately after.198204 session.sendMsg(@@ -204,22 +210,23 @@204210 warnOnSkew(nowMs, hello.deviceTime)205211 206212 if (hello.version == localVersion) {207- session.recvAs<SyncMessage.Done>("SYNC_DONE")213+ session.recvAs<SyncMessage.Done>("responder", "SYNC_DONE")208214 session.sendMsg(SyncMessage.Done(0, 0))209215 return@withTimeout SyncOutcome(SyncOutcomeStatus.COMPLETED)210216 }211217 212218 val ourEntries = vault.buildEntryManifest()213219 val ourTombstones = vault.buildTombstoneManifest()214- val peerManifest = session.recvAs<SyncMessage.Manifest>("SYNC_MANIFEST")220+ val peerManifest = session.recvAs<SyncMessage.Manifest>("responder", "SYNC_MANIFEST")215221 session.sendMsg(SyncMessage.Manifest(ourEntries, ourTombstones))216222 217- val initiatorWants = session.recvAs<SyncMessage.Request>("SYNC_REQUEST").entryIds223+ val initiatorWants =224+ session.recvAs<SyncMessage.Request>("responder", "SYNC_REQUEST").entryIds218225 val wantFromPeer = diffNewer(peerManifest.entries, ourEntries)219226 session.sendMsg(SyncMessage.Request(wantFromPeer))220227 221228 // Drain the initiator's unconditional SYNC_ENTRIES first.222- val incoming = session.recvAs<SyncMessage.Entries>("SYNC_ENTRIES")229+ val incoming = session.recvAs<SyncMessage.Entries>("responder", "SYNC_ENTRIES")223230 if (wantFromPeer.isNotEmpty()) {224231 vault.applyEntries(incoming.entries)225232 entriesReceived = incoming.entries.size@@ -231,7 +238,10 @@231238 entriesSent = ourPayload.size232239 }233240 234- val peerTombstones = session.recvAs<SyncMessage.Tombstones>("SYNC_TOMBSTONES")241+ val peerTombstones = session.recvAs<SyncMessage.Tombstones>(242+ "responder",243+ "SYNC_TOMBSTONES",244+ )235245 vault.applyTombstones(peerTombstones.tombstones)236246 tombstonesReceived = peerTombstones.tombstones.size237247 @@ -240,7 +250,10 @@240250 tombstonesSent = newerForPeer.size241251 242252 if (vault.supportsDeviceManifest) {243- val theirs = session.recvAs<SyncMessage.DeviceManifest>("SYNC_DEVICE_MANIFEST")253+ val theirs = session.recvAs<SyncMessage.DeviceManifest>(254+ "responder",255+ "SYNC_DEVICE_MANIFEST",256+ )244257 session.sendMsg(SyncMessage.DeviceManifest(vault.buildOutgoingDeviceManifest()))245258 if (theirs.manifest != null) {246259 vault.applyIncomingDeviceManifest(@@ -250,7 +263,7 @@250263 }251264 }252265 253- session.recvAs<SyncMessage.Done>("SYNC_DONE")266+ session.recvAs<SyncMessage.Done>("responder", "SYNC_DONE")254267 session.sendMsg(SyncMessage.Done(entriesReceived, entriesSent))255268 SyncOutcome(256269 SyncOutcomeStatus.COMPLETED,@@ -284,7 +297,14 @@284297 send(bytes)285298 }286299 287-private suspend inline fun <reified T : SyncMessage> SyncSession.recvAs(expected: String): T {300+private suspend inline fun <reified T : SyncMessage> SyncSession.recvAs(301+ role: String,302+ expected: String,303+): T {304+ // Stage-only diagnostics: no device names, entry ids, counts, payloads,305+ // room capabilities, or key material. The final marker before a timeout306+ // identifies the protocol edge where the peers stopped agreeing.307+ PeachSyncLog.i("sync-stage: role=$role wait=$expected")288308 val bytes = receive()289309 requireSyncFrameSize(bytes.size)290310 val msg = decodeSyncMessage(bytes)app/src/test/java/com/peachpasswords/sync/SyncCoordinatorTest.kt+15 −3
@@ -69,9 +69,14 @@6969 val a = FakeAdapter("A", version = 1).apply { put("x", 100) }7070 val b = FakeAdapter("B", version = 2).apply { put("y", 200) }7171 72- val ra = async { runSyncInitiator(sa, a, nowMs = 1_000) }73- val rb = async { runSyncResponder(sb, b, nowMs = 1_000) }74- val (outA, outB) = ra.await() to rb.await()72+ lateinit var outA: SyncOutcome73+ lateinit var outB: SyncOutcome74+ val logs = capturePeachSyncLogs {75+ val ra = async { runSyncInitiator(sa, a, nowMs = 1_000) }76+ val rb = async { runSyncResponder(sb, b, nowMs = 1_000) }77+ outA = ra.await()78+ outB = rb.await()79+ }7580 7681 assertEquals(SyncOutcomeStatus.COMPLETED, outA.status)7782 assertEquals(SyncOutcomeStatus.COMPLETED, outB.status)@@ -80,6 +85,13 @@8085 assertTrue(b.blobs.containsKey("x"))8186 assertEquals(1, outA.entriesReceived)8287 assertEquals(1, outA.entriesSent)88+ val messages = logs.map { it.second }89+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_HELLO"))90+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_MANIFEST"))91+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_REQUEST"))92+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_ENTRIES"))93+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_TOMBSTONES"))94+ assertTrue(messages.contains("sync-stage: role=responder wait=SYNC_DONE"))8395 }8496 8597 @Test fun equal_versions_short_circuit_without_transfer() = runBlocking {