From 38d304d014caf9d571889e1cc4dc5ea583f3588e Mon Sep 17 00:00:00 2001 From: fy59 Date: Sun, 20 Sep 2026 17:54:43 +0200 Subject: [PATCH] fix(android): prioritize explicit sync conversations --- .../trainlog/data/DriveSyncTransport.kt | 32 ++- .../data/SyncBackgroundActionability.kt | 44 +++ .../trainlog/data/SyncBackgroundService.kt | 33 ++- .../trainlog/data/SyncConversationArbiter.kt | 116 ++++++++ .../SyncGenerationForegroundCoordinator.kt | 146 +++++++--- .../com/labfytools/trainlog/ui/SyncScreen.kt | 124 ++++++--- .../app/src/main/res/values-en/strings.xml | 1 + android/app/src/main/res/values/strings.xml | 1 + .../data/SyncConversationOwnershipTest.kt | 251 ++++++++++++++++++ .../SyncGenerationConversationLockTest.kt | 8 +- .../data/SyncGenerationServiceTest.kt | 20 +- tests/test_syncd_routing.py | 52 ++++ 12 files changed, 741 insertions(+), 87 deletions(-) create mode 100644 android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt create mode 100644 android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt create mode 100644 android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt b/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt index 99eec1a..7e5af45 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/DriveSyncTransport.kt @@ -291,10 +291,15 @@ class DriveSyncWorker(context: Context, parameters: WorkerParameters) : transport.pull(local) stage = "coordinate" val coordinator = SyncGenerationCoordinator(repository) + val terminalLedger = BackgroundSyncRunLedger(applicationContext) coordinator.publishPeer(local) stage = "peer-publication" transport.pushPeer(local) - if (!coordinator.hasActionableRequest(local)) { + val actionability = coordinator.classifyBackgroundRequest(local, terminalLedger.terminal()) + if ( + actionability == BackgroundRequestActionability.NOT_ACTIONABLE || + actionability == BackgroundRequestActionability.RESUMABLE_BUT_STALE + ) { if (coordinator.hasReplayableAcknowledgement(local)) { stage = "acknowledgement-replay" transport.pushAcknowledgement(local) @@ -322,8 +327,31 @@ class DriveSyncWorker(context: Context, parameters: WorkerParameters) : stage = "acknowledgement" transport.pushAcknowledgement(local) }, + intent = SyncConversationIntent.RESUME_BACKGROUND, ) - if (result is ForegroundGenerationResult.Completed) Result.success() else Result.retry() + when (result) { + is ForegroundGenerationResult.Completed -> { + terminalLedger.recordTerminal( + result.runId, + coordinator.remoteEvidence(local, result.runId), + ) + Result.success() + } + ForegroundGenerationResult.Busy, + is ForegroundGenerationResult.Cancelled -> Result.success() + is ForegroundGenerationResult.Superseded -> { + result.runId?.let { + terminalLedger.recordTerminal(it, coordinator.remoteEvidence(local, it)) + } + Result.success() + } + is ForegroundGenerationResult.Failed -> { + result.runId?.let { + terminalLedger.recordTerminal(it, coordinator.remoteEvidence(local, it)) + } + Result.retry() + } + } } catch (error: Exception) { /* SECURITY: stage and exception class are fixed application * metadata. Provider messages may contain private URIs and are diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt new file mode 100644 index 0000000..b11c76c --- /dev/null +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundActionability.kt @@ -0,0 +1,44 @@ +package com.labfytools.trainlog.data + +import android.content.Context + +internal enum class BackgroundRequestActionability { + NEW_REQUEST, + NEW_REMOTE_EVIDENCE, + ACTIVE_HANDOFF, + RESUMABLE_BUT_STALE, + NOT_ACTIONABLE, +} + +internal data class BackgroundTerminalRun( + val runId: String, + val remoteEvidence: String, +) + +/** One-entry durable suppression ledger for the current background request. */ +internal class BackgroundSyncRunLedger(context: Context) { + private val preferences = + context.applicationContext.getSharedPreferences(PREFERENCES, Context.MODE_PRIVATE) + + fun terminal(): BackgroundTerminalRun? { + val runId = preferences.getString(RUN_ID, null) ?: return null + return BackgroundTerminalRun(runId, preferences.getString(REMOTE_EVIDENCE, "").orEmpty()) + } + + fun recordTerminal(runId: String, remoteEvidence: String) { + /* CONTRACT: only the request currently present on the transport needs + * suppression. A replacement request has a distinct run_id and is + * immediately actionable, so this ledger remains permanently bounded + * to one record and is not a second synchronization database. */ + preferences.edit() + .putString(RUN_ID, runId) + .putString(REMOTE_EVIDENCE, remoteEvidence) + .apply() + } + + companion object { + private const val PREFERENCES = "trainlog-background-sync-terminal" + private const val RUN_ID = "run_id" + private const val REMOTE_EVIDENCE = "remote_evidence" + } +} diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt index ba8095a..6d0a73b 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt @@ -76,6 +76,7 @@ class SyncBackgroundService : Service() { requireNotNull(repository), AndroidMtpPublicationVisibility(applicationContext), ) + val terminalLedger = BackgroundSyncRunLedger(applicationContext) while (!stopped.get() && BackgroundSyncSettings(this).enabled) { val directory = canonicalExchangeDirectory() /* WHY: waiting inside run() owns the process-wide conversation @@ -84,8 +85,36 @@ class SyncBackgroundService : Service() { * the coordinator still validates all protocol evidence. * INVARIANT: no request means no business work and no lock. */ coordinator.publishPeer(directory) - if (coordinator.hasActionableRequest(directory)) { - coordinator.run(directory, Duration.ofMinutes(5)) + val actionability = + coordinator.classifyBackgroundRequest(directory, terminalLedger.terminal()) + if ( + actionability == BackgroundRequestActionability.NEW_REQUEST || + actionability == BackgroundRequestActionability.NEW_REMOTE_EVIDENCE || + actionability == BackgroundRequestActionability.ACTIVE_HANDOFF + ) { + when ( + val result = coordinator.run( + directory, + Duration.ofMinutes(5), + intent = SyncConversationIntent.RESUME_BACKGROUND, + ) + ) { + is ForegroundGenerationResult.Completed -> + terminalLedger.recordTerminal( + result.runId, + coordinator.remoteEvidence(directory, result.runId), + ) + is ForegroundGenerationResult.Failed -> + result.runId?.let { + terminalLedger.recordTerminal(it, coordinator.remoteEvidence(directory, it)) + } + is ForegroundGenerationResult.Superseded -> + result.runId?.let { + terminalLedger.recordTerminal(it, coordinator.remoteEvidence(directory, it)) + } + ForegroundGenerationResult.Busy, + is ForegroundGenerationResult.Cancelled -> Unit + } } if (!stopped.get()) runCatching { Thread.sleep(5_000) } } diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt new file mode 100644 index 0000000..4c63c2d --- /dev/null +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncConversationArbiter.kt @@ -0,0 +1,116 @@ +package com.labfytools.trainlog.data + +import java.time.Duration +import java.util.concurrent.locks.ReentrantLock +import kotlin.concurrent.withLock + +internal enum class SyncConversationOwner { + FOREGROUND_EXPLICIT, + BACKGROUND_AUTOMATIC, +} + +internal class SyncConversationYieldedException : Exception("conversation superseded by explicit sync") + +/** Process-wide ownership authority for Android generation conversations. */ +internal class SyncConversationArbiter { + private val lock = ReentrantLock() + private val ownershipChanged = lock.newCondition() + private var active: Lease? = null + private var pendingExplicit = 0 + + /** + * Announces user intent before the foreground service is refreshed. + * + * WHY: starting the service first allowed it to acquire the old binary + * lock and silently absorb the click. CONTRACT: an announcement prevents + * new background ownership and cooperatively supersedes an existing + * background lease. INVARIANT: ownership is still exclusive until that + * lease observes yield and releases in its finally block. + */ + fun announceExplicit(): ExplicitClaim = lock.withLock { + pendingExplicit += 1 + active?.takeIf { it.owner == SyncConversationOwner.BACKGROUND_AUTOMATIC } + ?.requestYield() + ownershipChanged.signalAll() + ExplicitClaim(this) + } + + fun tryAcquireBackground(): Lease? = lock.withLock { + if (active != null || pendingExplicit > 0) return null + Lease(this, SyncConversationOwner.BACKGROUND_AUTOMATIC).also { active = it } + } + + private fun acquireExplicit(claim: ExplicitClaim, timeout: Duration): Lease? { + return lock.withLock { + check(claim.announced) + var remaining = timeout.toNanos() + while (active != null) { + if (remaining <= 0L) { + releaseAnnouncement(claim) + return@withLock null + } + remaining = ownershipChanged.awaitNanos(remaining) + } + releaseAnnouncement(claim) + Lease(this, SyncConversationOwner.FOREGROUND_EXPLICIT).also { active = it } + } + } + + private fun releaseAnnouncement(claim: ExplicitClaim) { + if (!claim.announced) return + claim.announced = false + pendingExplicit -= 1 + check(pendingExplicit >= 0) + ownershipChanged.signalAll() + } + + private fun release(lease: Lease) = lock.withLock { + if (active === lease) { + active = null + ownershipChanged.signalAll() + } + } + + internal class ExplicitClaim internal constructor(private val arbiter: SyncConversationArbiter) : + AutoCloseable { + internal var announced = true + private var acquired = false + + fun acquire(timeout: Duration): Lease? { + check(!acquired) + val lease = arbiter.acquireExplicit(this, timeout) + acquired = lease != null + return lease + } + + override fun close() = arbiter.lock.withLock { + if (!acquired) arbiter.releaseAnnouncement(this) + } + } + + internal class Lease internal constructor( + private val arbiter: SyncConversationArbiter, + val owner: SyncConversationOwner, + ) : AutoCloseable { + @Volatile private var yieldRequested = false + @Volatile private var closed = false + + internal fun requestYield() { + yieldRequested = true + } + + fun throwIfYieldRequested() { + if (yieldRequested) throw SyncConversationYieldedException() + } + + override fun close() { + if (closed) return + closed = true + arbiter.release(this) + } + } + + companion object { + val process = SyncConversationArbiter() + } +} 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 8a9b2f7..41d78e3 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 @@ -13,8 +13,15 @@ import org.json.JSONObject internal sealed interface ForegroundGenerationResult { data class Completed(val runId: String) : ForegroundGenerationResult + data object Busy : ForegroundGenerationResult + data class Superseded(val runId: String?) : ForegroundGenerationResult + data class Cancelled(val runId: String?) : ForegroundGenerationResult + data class Failed(val message: String, val runId: String?) : ForegroundGenerationResult +} - data class Error(val message: String) : ForegroundGenerationResult +internal enum class SyncConversationIntent { + NEW_EXPLICIT, + RESUME_BACKGROUND, } internal fun interface SyncGenerationTrace { @@ -43,6 +50,7 @@ internal class SyncGenerationCoordinator( private val repository: TrainlogRepository, private val visibility: MtpPublicationVisibility = ImmediateMtpPublicationVisibility, private val trace: SyncGenerationTrace = AndroidSyncGenerationTrace, + private val arbiter: SyncConversationArbiter = SyncConversationArbiter.process, ) { private val service = SyncGenerationService(repository) @@ -54,14 +62,6 @@ internal class SyncGenerationCoordinator( result: String? = null, ) = trace.record(runId, phase, artifact, generationId, result) - companion object { - /* WHY: UI and background triggers share one repository protocol and - * must never create concurrent runs. CONTRACT: this process-wide lock - * guards only one bounded conversation; SQLite remains the durable - * replay authority after process restart. */ - private val conversationLock = java.util.concurrent.locks.ReentrantLock() - } - internal fun resumableGeneration(runId: String, consumerPeerId: String? = null): CapturedSyncGeneration? = repository.inSyncGenerationTransaction { db -> db.rawQuery( @@ -90,10 +90,10 @@ internal class SyncGenerationCoordinator( } } - internal fun hasActionableRequest(directory: File): Boolean { + private fun request(directory: File): JSONObject? { val path = File(directory, "request-v1.json") - if (!path.isFile || path.length() !in 1..64 * 1024) return false - val request = runCatching { JSONObject(path.readText()) }.getOrNull() ?: return false + if (!path.isFile || path.length() !in 1..64 * 1024) return null + val request = runCatching { JSONObject(path.readText()) }.getOrNull() ?: return null val runId = request.optString("run_id") val desktopPeerId = request.optString("desktop_peer_id") if ( @@ -102,16 +102,64 @@ internal class SyncGenerationCoordinator( !runId.matches(Regex("^sy_[0-9a-f-]{36}$")) || !desktopPeerId.matches(Regex("^peer_[0-9a-f-]{36}$")) || request.optString("android_peer_id") != service.peerId() - ) return false - if (resumableGeneration(runId, desktopPeerId) != null) return true - return !repository.inSyncGenerationTransaction { db -> + ) return null + return request + } + + private fun hasGeneration(runId: String): Boolean = + repository.inSyncGenerationTransaction { db -> db.rawQuery( "SELECT 1 FROM sync_generations WHERE run_id=? LIMIT 1", arrayOf(runId), ).use { it.moveToFirst() } } + + internal fun remoteEvidence(directory: File, runId: String): String { + val evidence = + listOf( + Triple("desktop-consumption-ack-v1.json", "trainlog-sync-ack", 1), + Triple("desktop-generation-v1.json", "trainlog-sync-generation-reference", 1), + ) + .mapNotNull { (name, format, version) -> + val path = File(directory, name) + if (!path.isFile || path.length() !in 1..64 * 1024) return@mapNotNull null + val raw = runCatching { path.readBytes() }.getOrNull() ?: return@mapNotNull null + val value = runCatching { JSONObject(raw.toString(Charsets.UTF_8)) }.getOrNull() + ?: return@mapNotNull null + if ( + value.optString("format") != format || + value.optInt("version", -1) != version || + value.optString("run_id") != runId + ) return@mapNotNull null + "$name:${java.security.MessageDigest.getInstance("SHA-256").digest(raw) + .joinToString("") { "%02x".format(it) }}" + } + return evidence.sorted().joinToString("|") } + internal fun classifyBackgroundRequest( + directory: File, + terminal: BackgroundTerminalRun?, + ): BackgroundRequestActionability { + val request = request(directory) ?: return BackgroundRequestActionability.NOT_ACTIONABLE + val runId = request.getString("run_id") + val desktopPeerId = request.getString("desktop_peer_id") + val resumable = resumableGeneration(runId, desktopPeerId) != null + if (terminal?.runId == runId) { + val currentEvidence = remoteEvidence(directory, runId) + return if (currentEvidence.isNotEmpty() && currentEvidence != terminal.remoteEvidence) + BackgroundRequestActionability.NEW_REMOTE_EVIDENCE + else if (resumable) BackgroundRequestActionability.RESUMABLE_BUT_STALE + else BackgroundRequestActionability.NOT_ACTIONABLE + } + if (!hasGeneration(runId)) return BackgroundRequestActionability.NEW_REQUEST + if (resumable) return BackgroundRequestActionability.ACTIVE_HANDOFF + return BackgroundRequestActionability.NOT_ACTIONABLE + } + + internal fun currentRequestRunId(directory: File): String? = + request(directory)?.optString("run_id") + internal fun hasReplayableAcknowledgement(directory: File): Boolean { val request = runCatching { JSONObject(File(directory, "request-v1.json").readText()) }.getOrNull() @@ -287,8 +335,10 @@ internal class SyncGenerationCoordinator( deadline: Long, predicate: (JSONObject) -> Boolean, pump: (() -> Unit)? = null, + lease: SyncConversationArbiter.Lease, ): ByteArray { while (true) { + lease.throwIfYieldRequested() if (path.isFile && path.length() <= 64 * 1024) { try { val raw = path.readBytes() @@ -312,11 +362,27 @@ internal class SyncGenerationCoordinator( afterPublication: (() -> Unit)? = null, pollTransport: (() -> Unit)? = null, afterAcknowledgement: (() -> Unit)? = null, + intent: SyncConversationIntent = SyncConversationIntent.RESUME_BACKGROUND, + explicitClaim: SyncConversationArbiter.ExplicitClaim? = null, ): ForegroundGenerationResult { - if (!conversationLock.tryLock()) { - return ForegroundGenerationResult.Error("Synchronization already in progress.") + val claim = + if (intent == SyncConversationIntent.NEW_EXPLICIT) + explicitClaim ?: arbiter.announceExplicit() + else null + val lease = try { + claim?.acquire(timeout) ?: if (claim == null) arbiter.tryAcquireBackground() else null + } catch (_: InterruptedException) { + Thread.currentThread().interrupt() + claim?.close() + return ForegroundGenerationResult.Cancelled(null) } + if (lease == null) { + claim?.close() + return ForegroundGenerationResult.Busy + } + var currentRunId: String? = null return try { + val baselineRunId = request(directory)?.optString("run_id") val peer = publishPeer(directory) phase(null, "peer_published", "android-peer-v1.json") /* WHY: trainlog-syncd is intentionally driven by the established @@ -325,23 +391,23 @@ internal class SyncGenerationCoordinator( * identity before its caller emits one fresh request signal. * INVARIANT: a failed attempt's request, run, generations, and * ACK evidence remain immutable; retry creates new correlation. */ - afterPeerPublication?.invoke() - phase(null, "legacy_trigger_published", "trainlog-sync-request-v1.json") + afterPeerPublication?.let { + it.invoke() + phase(null, "legacy_trigger_published", "trainlog-sync-request-v1.json") + } val deadline = System.nanoTime() + timeout.toNanos() val requestRaw = awaitCorrelated(File(directory, "request-v1.json"), deadline, { candidate -> val runId = candidate.optString("run_id") - runId.startsWith("sy_") && - (resumableGeneration(runId) != null || - !repository.inSyncGenerationTransaction { db -> - db.rawQuery( - "SELECT 1 FROM sync_generations WHERE run_id=? LIMIT 1", - arrayOf(runId), - ).use { it.moveToFirst() } - }) - }, pollTransport) + runId.startsWith("sy_") && when (intent) { + SyncConversationIntent.NEW_EXPLICIT -> runId != baselineRunId + SyncConversationIntent.RESUME_BACKGROUND -> + resumableGeneration(runId) != null || !hasGeneration(runId) + } + }, pollTransport, lease) val request = JSONObject(requestRaw.toString(Charsets.UTF_8)) val runId = request.getString("run_id") + currentRunId = runId phase(runId, "generation_request_observed", "request-v1.json") if (request.getString("android_peer_id") != peer) throw SyncGenerationException("request targets another Android peer") @@ -354,6 +420,7 @@ internal class SyncGenerationCoordinator( it.optString("android_peer_id") == peer }, pollTransport, + lease, ) reconcileArchiveAcknowledgements( archiveAckRaw, @@ -370,6 +437,7 @@ internal class SyncGenerationCoordinator( * an acknowledged outbound generation remains resumable only to * finish its missing inbound half. */ phase(runId, "generation_capture_started") + lease.throwIfYieldRequested() val captured = resumableGeneration( request.getString("run_id"), @@ -379,11 +447,13 @@ internal class SyncGenerationCoordinator( request.getString("desktop_peer_id"), request.getString("run_id"), ) + lease.throwIfYieldRequested() phase(runId, "generation_captured", generationId = captured.generationId) val published = service.publish(captured, File(directory, "android-objects")) visibility.confirm( published.walkTopDown().filter(File::isFile).sortedBy { it.name == "manifest.json" }.toList() ) + lease.throwIfYieldRequested() phase( runId, "generation_objects_published", @@ -410,6 +480,7 @@ internal class SyncGenerationCoordinator( captured.generationId, ) afterPublication?.invoke() + lease.throwIfYieldRequested() // CONTRACT: durable objects from earlier conversations remain in // the exchange directory. Only this run's exact generation ACK // can advance its lineage; stale bytes are retained and ignored. @@ -417,13 +488,13 @@ internal class SyncGenerationCoordinator( awaitCorrelated(File(directory, "desktop-consumption-ack-v1.json"), deadline, { it.optString("run_id") == captured.runId && it.optString("generation_id") == captured.generationId - }, pollTransport) + }, pollTransport, lease) service.acceptAcknowledgement(desktopAck) phase(runId, "desktop_ack_observed", "desktop-consumption-ack-v1.json", captured.generationId) val referenceRaw = awaitCorrelated(File(directory, "desktop-generation-v1.json"), deadline, { it.optString("run_id") == captured.runId - }, pollTransport) + }, pollTransport, lease) val reference = JSONObject(referenceRaw.toString(Charsets.UTF_8)) phase( runId, @@ -458,6 +529,13 @@ internal class SyncGenerationCoordinator( afterAcknowledgement?.invoke() phase(runId, "completed", result = "completed") ForegroundGenerationResult.Completed(captured.runId) + } catch (_: SyncConversationYieldedException) { + phase(currentRunId, "superseded", result = "foreground_priority") + ForegroundGenerationResult.Superseded(currentRunId) + } catch (_: InterruptedException) { + Thread.currentThread().interrupt() + phase(currentRunId, "cancelled", result = "interrupted") + ForegroundGenerationResult.Cancelled(currentRunId) } catch (error: Exception) { if (error.message == "generation retention capacity exhausted") { try { @@ -486,9 +564,13 @@ internal class SyncGenerationCoordinator( "error", result = error.javaClass.simpleName, ) - ForegroundGenerationResult.Error(error.message ?: "generation synchronization failed") + ForegroundGenerationResult.Failed( + error.message ?: "generation synchronization failed", + currentRunId, + ) } finally { - conversationLock.unlock() + lease.close() + claim?.close() } } } diff --git a/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt b/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt index 2e7e17a..11c7c4b 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt @@ -29,6 +29,9 @@ import com.labfytools.trainlog.data.SyncRequestOutbox import com.labfytools.trainlog.data.SyncRequestResult import com.labfytools.trainlog.data.TrainlogRepository import com.labfytools.trainlog.data.ForegroundGenerationResult +import com.labfytools.trainlog.data.BackgroundSyncRunLedger +import com.labfytools.trainlog.data.SyncConversationArbiter +import com.labfytools.trainlog.data.SyncConversationIntent import com.labfytools.trainlog.data.SyncGenerationForegroundCoordinator import com.labfytools.trainlog.data.BackgroundSyncSettings import com.labfytools.trainlog.data.SyncBackgroundService @@ -52,39 +55,72 @@ internal suspend fun publishBundleAndRequest( exporter: SyncExporter, requestOutbox: SyncRequestOutbox, visibility: MtpPublicationVisibility = com.labfytools.trainlog.data.ImmediateMtpPublicationVisibility, + explicitClaim: SyncConversationArbiter.ExplicitClaim = + SyncConversationArbiter.process.announceExplicit(), + terminalLedger: BackgroundSyncRunLedger? = null, ): SyncRequestResult = withContext(Dispatchers.IO) { - val opened = exporter.openStorageSnapshot() - if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) { - return@withContext SyncRequestResult.Error("prepare:${opened.message}") - } - val snapshot = (opened as com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Ready).snapshot - /* CONTRACT: the request signal remains compatible with an unconfigured - * desktop, so publish the strict standalone V3 snapshot before admission. - * A configured daemon still routes the signal exclusively to the complete - * generation below; this compatibility publication never selects V3. */ - when (val publication = exporter.exportMobileBundle(snapshot)) { - SyncExportResult.Unsupported -> return@withContext SyncRequestResult.Unsupported - is SyncExportResult.Error -> - return@withContext SyncRequestResult.Error("prepare:${publication.message}") - is SyncExportResult.Exported -> Unit - } - return@withContext when ( - val result = SyncGenerationForegroundCoordinator(repository, visibility).run( - canonicalExchangeDirectory(), - afterPeerPublication = { - when (val request = requestOutbox.requestSync(snapshot)) { - is SyncRequestResult.Requested -> Unit - SyncRequestResult.Unsupported -> - throw IllegalStateException("Generation synchronization is unsupported.") - is SyncRequestResult.Error -> throw IllegalStateException(request.message) - is SyncRequestResult.Completed -> - throw IllegalStateException("Unexpected completed request signal.") + try { + val opened = exporter.openStorageSnapshot() + if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) { + return@withContext SyncRequestResult.Error("prepare:${opened.message}") + } + val snapshot = + (opened as com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Ready).snapshot + /* CONTRACT: the request signal remains compatible with an unconfigured + * desktop, so publish the strict standalone V3 snapshot before admission. + * A configured daemon still routes the signal exclusively to the complete + * generation below; this compatibility publication never selects V3. */ + when (val publication = exporter.exportMobileBundle(snapshot)) { + SyncExportResult.Unsupported -> return@withContext SyncRequestResult.Unsupported + is SyncExportResult.Error -> + return@withContext SyncRequestResult.Error("prepare:${publication.message}") + is SyncExportResult.Exported -> Unit + } + val directory = canonicalExchangeDirectory() + val coordinator = SyncGenerationForegroundCoordinator(repository, visibility) + return@withContext when ( + val result = coordinator.run( + directory, + afterPeerPublication = { + when (val request = requestOutbox.requestSync(snapshot)) { + is SyncRequestResult.Requested -> Unit + SyncRequestResult.Unsupported -> + throw IllegalStateException("Generation synchronization is unsupported.") + is SyncRequestResult.Error -> throw IllegalStateException(request.message) + is SyncRequestResult.Completed -> + throw IllegalStateException("Unexpected completed request signal.") + } + }, + intent = SyncConversationIntent.NEW_EXPLICIT, + explicitClaim = explicitClaim, + ) + ) { + is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId) + ForegroundGenerationResult.Busy -> + SyncRequestResult.Error("sync_interrupted") + is ForegroundGenerationResult.Superseded -> + SyncRequestResult.Error("sync_interrupted").also { + result.runId?.let { runId -> + terminalLedger?.recordTerminal( + runId, + coordinator.remoteEvidence(directory, runId), + ) + } } - }, - ) - ) { - is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId) - is ForegroundGenerationResult.Error -> SyncRequestResult.Error(result.message) + is ForegroundGenerationResult.Cancelled -> + SyncRequestResult.Error("sync_interrupted") + is ForegroundGenerationResult.Failed -> + SyncRequestResult.Error(result.message).also { + result.runId?.let { runId -> + terminalLedger?.recordTerminal( + runId, + coordinator.remoteEvidence(directory, runId), + ) + } + } + } + } finally { + explicitClaim.close() } } @@ -147,8 +183,11 @@ fun SyncScreen( add(strings.resources.getQuantityString(R.plurals.catalog_reconciled, reconciled, reconciled)) skipped?.let { add(strings.resources.getQuantityString(R.plurals.catalog_present, it, it)) } }.joinToString(", ") - fun visibleError(raw: String): String = raw.removePrefix("prepare:").let { - if (raw.startsWith("prepare:")) strings.getString(R.string.sync_prepare_error, it) else raw + fun visibleError(raw: String): String = when { + raw == "sync_interrupted" -> strings.getString(R.string.sync_interrupted) + raw.startsWith("prepare:") -> + strings.getString(R.string.sync_prepare_error, raw.removePrefix("prepare:")) + else -> raw } var status by @@ -393,15 +432,15 @@ fun SyncScreen( * published from one prepared repository snapshot before * that signal becomes visible; otherwise the PC can mix a * fresh V3 file with stale or absent causal companions. */ - if (backgroundSettings.enabled) { - /* WHY: Samsung may freeze the activity after USB mode - * changes or a permission surface opens. CONTRACT: a - * user-enabled foreground listener is refreshed before - * the UI-owned conversation begins; the process lock - * prevents it from creating a second generation. */ - SyncBackgroundService.start(context) - } + /* WHY: Samsung may freeze the activity after USB mode + * changes or a permission surface opens. CONTRACT: + * announce explicit priority before waking the service. + * The service may keep Android alive, but cannot become + * conversation owner while this claim is pending. */ + val explicitClaim = SyncConversationArbiter.process.announceExplicit() + if (backgroundSettings.enabled) SyncBackgroundService.start(context) syncRunning = true + status = strings.getString(R.string.sync_running) coroutineScope.launch { try { when ( @@ -410,6 +449,8 @@ fun SyncScreen( exporter, requestOutbox, AndroidMtpPublicationVisibility(context), + explicitClaim, + BackgroundSyncRunLedger(context), ) ) { is SyncRequestResult.Completed -> { @@ -435,6 +476,7 @@ fun SyncScreen( } } } finally { + explicitClaim.close() syncRunning = false } } diff --git a/android/app/src/main/res/values-en/strings.xml b/android/app/src/main/res/values-en/strings.xml index 3f9891b..683de90 100644 --- a/android/app/src/main/res/values-en/strings.xml +++ b/android/app/src/main/res/values-en/strings.xml @@ -94,6 +94,7 @@ Synchronize The Android snapshot is maintained automatically. Synchronization in progress… + Synchronization interrupted. You can start a new attempt. Synchronize now Background synchronization Allows Trainlog to receive PC synchronizations without opening the app. diff --git a/android/app/src/main/res/values/strings.xml b/android/app/src/main/res/values/strings.xml index 3a21413..bfc9fce 100644 --- a/android/app/src/main/res/values/strings.xml +++ b/android/app/src/main/res/values/strings.xml @@ -94,6 +94,7 @@ Synchroniser Le snapshot Android est maintenu automatiquement. Synchronisation en cours… + Synchronisation interrompue. Une nouvelle tentative peut être lancée. Synchroniser maintenant Synchronisation en arrière-plan Permet à Trainlog de recevoir les synchronisations du PC sans ouvrir l’application. diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt new file mode 100644 index 0000000..e1babfd --- /dev/null +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncConversationOwnershipTest.kt @@ -0,0 +1,251 @@ +package com.labfytools.trainlog.data + +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import java.io.File +import java.nio.file.Files +import java.time.Duration +import java.util.UUID +import java.util.concurrent.CountDownLatch +import java.util.concurrent.TimeUnit +import java.util.concurrent.atomic.AtomicReference +import org.json.JSONArray +import org.json.JSONObject +import org.junit.Assert.assertEquals +import org.junit.Assert.assertNotEquals +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner +import org.robolectric.annotation.Config + +@RunWith(RobolectricTestRunner::class) +@Config(sdk = [35]) +class SyncConversationOwnershipTest { + private val context: Context = ApplicationProvider.getApplicationContext() + + private fun request(root: File, runId: String, androidPeer: String, desktopPeer: String) { + File(root, "request-v1.json").writeText( + JSONObject() + .put("format", "trainlog-sync-generation-request") + .put("version", 1) + .put("run_id", runId) + .put("android_peer_id", androidPeer) + .put("desktop_peer_id", desktopPeer) + .toString() + ) + } + + private fun archiveAcknowledgements( + root: File, + runId: String, + androidPeer: String, + desktopPeer: String, + ) { + File(root, "desktop-archive-acknowledgements-v1.json").writeText( + JSONObject() + .put("format", "trainlog-sync-archive-acknowledgements") + .put("version", 1) + .put("run_id", runId) + .put("android_peer_id", androidPeer) + .put("desktop_peer_id", desktopPeer) + .put("acknowledgements", JSONArray()) + .toString() + ) + } + + @Test + fun explicitClickYieldsOldBackgroundAndPublishesFreshTriggerForNewRun() { + val databaseName = "generation-ownership-${UUID.randomUUID()}.db" + val root = Files.createTempDirectory("trainlog-generation-ownership-").toFile() + val repository = TrainlogRepository(context, databaseName) + val arbiter = SyncConversationArbiter() + val oldObserved = CountDownLatch(1) + val capturedRuns = mutableListOf() + val desktopPeer = "peer_${UUID.randomUUID()}" + val oldRun = "sy_${UUID.randomUUID()}" + val newRun = "sy_${UUID.randomUUID()}" + val service = SyncGenerationService(repository) + val background = + SyncGenerationCoordinator( + repository, + arbiter = arbiter, + trace = SyncGenerationTrace { runId, phase, _, _, _ -> + if (phase == "generation_request_observed" && runId == oldRun) oldObserved.countDown() + }, + ) + val foreground = + SyncGenerationCoordinator( + repository, + arbiter = arbiter, + trace = SyncGenerationTrace { runId, phase, _, _, _ -> + if (phase == "generation_captured") synchronized(capturedRuns) { + capturedRuns += checkNotNull(runId) + } + }, + ) + val executor = java.util.concurrent.Executors.newFixedThreadPool(2) + try { + request(root, oldRun, service.peerId(), desktopPeer) + File(root, "trainlog-sync-request-v1.json").writeText( + """{"format":"trainlog-sync-request","version":1,"request_id":"sr_old"}""" + ) + val old = executor.submit { + background.run(root, Duration.ofSeconds(3)) + } + assertTrue(oldObserved.await(2, TimeUnit.SECONDS)) + + val claim = arbiter.announceExplicit() + val snapshot = + (DirectExchangePublisher { directExchangeDirectoryForPath(root) }.snapshot() + as DirectExchangeSnapshotResult.Ready).snapshot + val outbox = SyncRequestOutbox { directExchangeDirectoryForPath(root) } + var freshRequestId = "" + val explicit = executor.submit { + foreground.run( + root, + Duration.ofMillis(700), + afterPeerPublication = { + freshRequestId = + (outbox.requestSync(snapshot) as SyncRequestResult.Requested).requestId + request(root, newRun, service.peerId(), desktopPeer) + archiveAcknowledgements(root, newRun, service.peerId(), desktopPeer) + }, + intent = SyncConversationIntent.NEW_EXPLICIT, + explicitClaim = claim, + ) + } + + assertTrue(old.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Superseded) + assertTrue(explicit.get(3, TimeUnit.SECONDS) is ForegroundGenerationResult.Failed) + assertNotEquals("sr_old", freshRequestId) + assertEquals( + freshRequestId, + JSONObject(File(root, "trainlog-sync-request-v1.json").readText()) + .getString("request_id"), + ) + assertEquals(listOf(newRun), capturedRuns) + repository.inSyncGenerationTransaction { database -> + database.rawQuery("SELECT run_id FROM sync_generations", null).use { cursor -> + assertTrue(cursor.moveToFirst()) + assertEquals(newRun, cursor.getString(0)) + assertEquals(false, cursor.moveToNext()) + } + } + } finally { + executor.shutdownNow() + repository.close() + context.deleteDatabase(databaseName) + root.deleteRecursively() + } + } + + @Test + fun terminalBackgroundRunStaysSuppressedUntilNewRequestOrRemoteEvidence() { + val databaseName = "generation-suppression-${UUID.randomUUID()}.db" + val root = Files.createTempDirectory("trainlog-generation-suppression-").toFile() + val repository = TrainlogRepository(context, databaseName) + try { + val service = SyncGenerationService(repository) + val coordinator = SyncGenerationCoordinator(repository, arbiter = SyncConversationArbiter()) + val desktopPeer = "peer_${UUID.randomUUID()}" + val runId = "sy_${UUID.randomUUID()}" + request(root, runId, service.peerId(), desktopPeer) + service.capture(root, desktopPeer, runId) + val terminal = BackgroundTerminalRun(runId, coordinator.remoteEvidence(root, runId)) + + repeat(5) { + assertEquals( + BackgroundRequestActionability.RESUMABLE_BUT_STALE, + coordinator.classifyBackgroundRequest(root, terminal), + ) + } + + File(root, "desktop-consumption-ack-v1.json").writeText( + JSONObject() + .put("format", "trainlog-sync-ack") + .put("version", 1) + .put("run_id", runId) + .put("proof", "new") + .toString() + ) + assertEquals( + BackgroundRequestActionability.NEW_REMOTE_EVIDENCE, + coordinator.classifyBackgroundRequest(root, terminal), + ) + + request(root, "sy_${UUID.randomUUID()}", service.peerId(), desktopPeer) + assertEquals( + BackgroundRequestActionability.NEW_REQUEST, + coordinator.classifyBackgroundRequest(root, terminal), + ) + } finally { + repository.close() + context.deleteDatabase(databaseName) + root.deleteRecursively() + } + } + + @Test + fun cancelledExplicitConversationHandsSameGenerationToBackgroundWithoutDuplication() { + val databaseName = "generation-handoff-${UUID.randomUUID()}.db" + val root = Files.createTempDirectory("trainlog-generation-handoff-").toFile() + val repository = TrainlogRepository(context, databaseName) + val arbiter = SyncConversationArbiter() + val published = CountDownLatch(1) + val result = AtomicReference() + val desktopPeer = "peer_${UUID.randomUUID()}" + val oldRunId = "sy_${UUID.randomUUID()}" + val runId = "sy_${UUID.randomUUID()}" + try { + val service = SyncGenerationService(repository) + request(root, oldRunId, service.peerId(), desktopPeer) + val coordinator = + SyncGenerationCoordinator( + repository, + arbiter = arbiter, + trace = SyncGenerationTrace { _, phase, _, _, _ -> + if (phase == "generation_reference_published") published.countDown() + }, + ) + val claim = arbiter.announceExplicit() + val thread = Thread { + result.set( + coordinator.run( + root, + Duration.ofSeconds(5), + afterPeerPublication = { + request(root, runId, service.peerId(), desktopPeer) + archiveAcknowledgements(root, runId, service.peerId(), desktopPeer) + }, + intent = SyncConversationIntent.NEW_EXPLICIT, + explicitClaim = claim, + ) + ) + } + thread.start() + assertTrue(published.await(3, TimeUnit.SECONDS)) + thread.interrupt() + thread.join(3_000) + assertTrue(result.get() is ForegroundGenerationResult.Cancelled) + assertEquals( + BackgroundRequestActionability.ACTIVE_HANDOFF, + coordinator.classifyBackgroundRequest(root, null), + ) + + val resumed = coordinator.run(root, Duration.ofMillis(250)) + assertTrue(resumed is ForegroundGenerationResult.Failed) + repository.inSyncGenerationTransaction { database -> + database.rawQuery("SELECT COUNT(*) FROM sync_generations WHERE run_id=?", arrayOf(runId)) + .use { cursor -> + assertTrue(cursor.moveToFirst()) + assertEquals(1, cursor.getInt(0)) + } + } + } finally { + repository.close() + context.deleteDatabase(databaseName) + root.deleteRecursively() + } + } +} diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt index ba71b83..de31766 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt @@ -38,12 +38,8 @@ class SyncGenerationConversationLockTest { } assertTrue(entered.await(2, TimeUnit.SECONDS)) val duplicate = background.run(root, Duration.ofMillis(10)) - assertTrue(duplicate is ForegroundGenerationResult.Error) - assertEquals( - "Synchronization already in progress.", - (duplicate as ForegroundGenerationResult.Error).message, - ) - assertTrue(active.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Error) + assertEquals(ForegroundGenerationResult.Busy, duplicate) + assertTrue(active.get(2, TimeUnit.SECONDS) is ForegroundGenerationResult.Failed) repository.inSyncGenerationTransaction { database -> database.rawQuery("SELECT COUNT(*) FROM sync_generations", null).use { cursor -> assertTrue(cursor.moveToFirst()) diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt index 785d5ee..7cf221c 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationServiceTest.kt @@ -357,9 +357,15 @@ class SyncGenerationServiceTest { val repository = TrainlogRepository(context, databaseName) try { val coordinator = SyncGenerationCoordinator(repository) - assertEquals(false, coordinator.hasActionableRequest(root)) + assertEquals( + BackgroundRequestActionability.NOT_ACTIONABLE, + coordinator.classifyBackgroundRequest(root, null), + ) File(root, "request-v1.json").writeText("not-json") - assertEquals(false, coordinator.hasActionableRequest(root)) + assertEquals( + BackgroundRequestActionability.NOT_ACTIONABLE, + coordinator.classifyBackgroundRequest(root, null), + ) val peer = SyncGenerationService(repository).peerId() val request = @@ -370,11 +376,17 @@ class SyncGenerationServiceTest { .put("android_peer_id", "peer_${UUID.randomUUID()}") .put("desktop_peer_id", "peer_${UUID.randomUUID()}") File(root, "request-v1.json").writeText(request.toString()) - assertEquals(false, coordinator.hasActionableRequest(root)) + assertEquals( + BackgroundRequestActionability.NOT_ACTIONABLE, + coordinator.classifyBackgroundRequest(root, null), + ) request.put("android_peer_id", peer) File(root, "request-v1.json").writeText(request.toString()) - assertEquals(true, coordinator.hasActionableRequest(root)) + assertEquals( + BackgroundRequestActionability.NEW_REQUEST, + coordinator.classifyBackgroundRequest(root, null), + ) } finally { repository.close() context.deleteDatabase(databaseName) diff --git a/tests/test_syncd_routing.py b/tests/test_syncd_routing.py index f2b80d0..c05ec7f 100644 --- a/tests/test_syncd_routing.py +++ b/tests/test_syncd_routing.py @@ -74,6 +74,58 @@ class SyncdRoutingTest(unittest.TestCase): self.assertEqual(trigger, state["trigger"]) self.assertEqual("full_generation_v1", state["effective_mode"]) + def test_consumed_request_id_does_not_create_a_second_desktop_run(self): + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + transport = root / "transport" + transport.mkdir() + request = "sr_22222222-2222-4222-8222-222222222222" + config = root / "config.json" + config.write_text( + json.dumps( + { + "format": "trainlog-sync-orchestrator-config", + "version": 1, + "enabled": True, + "mode": "mtp", + "expected_peer_id": "peer_11111111-1111-4111-8111-111111111111", + "transport_root": str(transport), + "owned_root": str(root / "owned"), + "timeout_seconds": 30, + } + ) + ) + state = root / "trainlog.db.sync-run.json" + state.write_text(json.dumps({"request_id": request, "run_id": "sy_old"})) + args = argparse.Namespace( + mtp_adapter=root / "adapter", + state=state, + trigger="android", + orchestrator=root / "sync_orchestrator.py", + database=root / "trainlog.db", + ) + commands = [] + + def execute(command, **_kwargs): + commands.append(command) + (transport / syncd.REQUEST_NAME).write_text( + json.dumps( + { + "format": "trainlog-sync-request", + "version": 1, + "request_id": request, + "requested_at": "2026-09-20T12:00:00Z", + } + ) + ) + return subprocess.CompletedProcess(command, 0, "PASS\n", "") + + with mock.patch.object(syncd.subprocess, "run", side_effect=execute): + self.assertEqual(3, syncd.run_full_generation(args, config)) + self.assertEqual(str(args.mtp_adapter), commands[0][0]) + self.assertEqual(1, len(commands)) + self.assertEqual("sy_old", json.loads(state.read_text())["run_id"]) + def test_android_production_trigger_no_longer_uses_opt_in_or_legacy_export(self): source = (ROOT / "android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt").read_text() production = source.split("internal suspend fun publishBundleAndRequest", 1)[1].split(