fix(android): prioritize explicit sync conversations
This commit is contained in:
fy59
parent
ca2ea9379b
commit
38d304d014
12 changed files with 741 additions and 87 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
}
|
||||
}
|
||||
|
|
@ -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) }
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
|
|
@ -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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -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
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -94,6 +94,7 @@
|
|||
<string name="sync_action">Synchronize</string>
|
||||
<string name="sync_snapshot_automatic">The Android snapshot is maintained automatically.</string>
|
||||
<string name="sync_running">Synchronization in progress…</string>
|
||||
<string name="sync_interrupted">Synchronization interrupted. You can start a new attempt.</string>
|
||||
<string name="sync_now">Synchronize now</string>
|
||||
<string name="background_sync_title">Background synchronization</string>
|
||||
<string name="background_sync_description">Allows Trainlog to receive PC synchronizations without opening the app.</string>
|
||||
|
|
|
|||
|
|
@ -94,6 +94,7 @@
|
|||
<string name="sync_action">Synchroniser</string>
|
||||
<string name="sync_snapshot_automatic">Le snapshot Android est maintenu automatiquement.</string>
|
||||
<string name="sync_running">Synchronisation en cours…</string>
|
||||
<string name="sync_interrupted">Synchronisation interrompue. Une nouvelle tentative peut être lancée.</string>
|
||||
<string name="sync_now">Synchroniser maintenant</string>
|
||||
<string name="background_sync_title">Synchronisation en arrière-plan</string>
|
||||
<string name="background_sync_description">Permet à Trainlog de recevoir les synchronisations du PC sans ouvrir l’application.</string>
|
||||
|
|
|
|||
|
|
@ -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<String>()
|
||||
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<ForegroundGenerationResult> {
|
||||
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<ForegroundGenerationResult> {
|
||||
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<ForegroundGenerationResult>()
|
||||
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()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -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())
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
Loading…
Reference in a new issue