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

fix(sync): make Android generation publication visible to MTP

Browse source
This commit is contained in:
fy59 2026-09-20 15:30:00 +02:00
parent ba5021b436
commit a2eaf224e8
9 changed files with 302 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

40
android/app/src/main/java/com/labfytools/trainlog/data/MtpPublicationVisibility.kt Normal file
View file

@ -0,0 +1,40 @@
package com.labfytools.trainlog.data
import android.content.Context
import android.media.MediaScannerConnection
import java.io.File
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
/** Confirms that durable external-storage files have reached Android's MTP index. */
internal fun interface MtpPublicationVisibility {
fun confirm(files: List<File>)
}
internal object ImmediateMtpPublicationVisibility : MtpPublicationVisibility {
override fun confirm(files: List<File>) = Unit
}
internal class AndroidMtpPublicationVisibility(context: Context) : MtpPublicationVisibility {
private val applicationContext = context.applicationContext
override fun confirm(files: List<File>) {
val paths = files.asSequence().filter(File::isFile).map { it.absolutePath }.distinct().toList()
if (paths.isEmpty()) return
val completed = CountDownLatch(paths.size)
/* WHY: direct File writes are durable before MediaProvider necessarily
* advertises them through Samsung's MTP database. CONTRACT: every
* immutable artifact is scanned before its mutable reference, and a
* publication is not reported complete until every callback arrives.
* INVARIANT: scanning changes transport visibility only; canonical
* names, bytes, generation identities and commit markers are intact. */
MediaScannerConnection.scanFile(
applicationContext,
paths.toTypedArray(),
Array(paths.size) { "application/json" },
) { _, _ -> completed.countDown() }
if (!completed.await(15, TimeUnit.SECONDS)) {
throw SyncGenerationException("timeout confirming MTP publication visibility")
}
}
}

5
android/app/src/main/java/com/labfytools/trainlog/data/SyncBackgroundService.kt
View file

@ -72,7 +72,10 @@ class SyncBackgroundService : Service() {
}
if (!launched.compareAndSet(false, true)) return START_NOT_STICKY
executor.execute {
val coordinator = SyncGenerationCoordinator(requireNotNull(repository))
val coordinator = SyncGenerationCoordinator(
requireNotNull(repository),
AndroidMtpPublicationVisibility(applicationContext),
)
while (!stopped.get() && BackgroundSyncSettings(this).enabled) {
val directory = canonicalExchangeDirectory()
/* WHY: waiting inside run() owns the process-wide conversation

98
android/app/src/main/java/com/labfytools/trainlog/data/SyncGenerationForegroundCoordinator.kt
View file

@ -1,5 +1,6 @@
package com.labfytools.trainlog.data
import android.util.Log
import java.io.File
import java.io.FileOutputStream
import java.nio.file.AtomicMoveNotSupportedException
@ -16,10 +17,43 @@ internal sealed interface ForegroundGenerationResult {
data class Error(val message: String) : ForegroundGenerationResult
}
internal fun interface SyncGenerationTrace {
fun record(runId: String?, phase: String, artifact: String?, generationId: String?, result: String?)
}
internal object AndroidSyncGenerationTrace : SyncGenerationTrace {
override fun record(
runId: String?, phase: String, artifact: String?, generationId: String?, result: String?
) {
Log.i(
"TrainlogSyncGeneration",
listOfNotNull(
"run_id=${runId ?: "pending"}",
"phase=$phase",
artifact?.let { "artifact=$it" },
generationId?.let { "generation_id=$it" },
result?.let { "result=$it" },
).joinToString(" "),
)
}
}
/** Owns the bounded, user-initiated Android side of a generation conversation. */
internal class SyncGenerationCoordinator(private val repository: TrainlogRepository) {
internal class SyncGenerationCoordinator(
private val repository: TrainlogRepository,
private val visibility: MtpPublicationVisibility = ImmediateMtpPublicationVisibility,
private val trace: SyncGenerationTrace = AndroidSyncGenerationTrace,
) {
private val service = SyncGenerationService(repository)
private fun phase(
runId: String?,
phase: String,
artifact: String? = null,
generationId: String? = null,
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
@ -229,9 +263,12 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
* an identical canonical descriptor is already a complete durable
* publication. INVARIANT: changed or unreadable bytes still use the
* crash-safe replacement path above. */
if (runCatching { path.isFile && path.readText() == descriptor }.getOrDefault(false))
if (runCatching { path.isFile && path.readText() == descriptor }.getOrDefault(false)) {
visibility.confirm(listOf(path))
return peer
}
publish(path, descriptor)
visibility.confirm(listOf(path))
return peer
}
@ -281,6 +318,7 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
}
return try {
val peer = publishPeer(directory)
phase(null, "peer_published", "android-peer-v1.json")
/* WHY: trainlog-syncd is intentionally driven by the established
* Android request artifact, not by polling generation internals.
* CONTRACT: each explicit foreground attempt publishes the peer
@ -288,6 +326,7 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
* 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")
val deadline = System.nanoTime() + timeout.toNanos()
val requestRaw =
awaitCorrelated(File(directory, "request-v1.json"), deadline, { candidate ->
@ -302,6 +341,8 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
})
}, pollTransport)
val request = JSONObject(requestRaw.toString(Charsets.UTF_8))
val runId = request.getString("run_id")
phase(runId, "generation_request_observed", "request-v1.json")
if (request.getString("android_peer_id") != peer)
throw SyncGenerationException("request targets another Android peer")
val archiveAckRaw =
@ -320,6 +361,7 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
peer,
request.getString("desktop_peer_id"),
)
phase(runId, "archive_ack_observed", "desktop-archive-acknowledgements-v1.json")
/* WHY: transport can fail after immutable generation publication.
* CONTRACT: an explicit retry for the same run republishes that
* exact generation and resumes ACK handling; it never recaptures
@ -327,6 +369,7 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
* INVARIANT: a completed inbound ledger makes the run ineligible;
* an acknowledged outbound generation remains resumable only to
* finish its missing inbound half. */
phase(runId, "generation_capture_started")
val captured =
resumableGeneration(
request.getString("run_id"),
@ -336,9 +379,20 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
request.getString("desktop_peer_id"),
request.getString("run_id"),
)
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()
)
phase(
runId,
"generation_objects_published",
published.relativeTo(directory).path,
captured.generationId,
)
val generationReference = File(directory, "android-generation-v1.json")
publish(
File(directory, "android-generation-v1.json"),
generationReference,
JSONObject()
.put("format", "trainlog-sync-generation-reference")
.put("version", 1)
@ -348,6 +402,13 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
.put("relative_path", published.relativeTo(directory).path)
.toString(),
)
visibility.confirm(listOf(generationReference))
phase(
runId,
"generation_reference_published",
generationReference.name,
captured.generationId,
)
afterPublication?.invoke()
// CONTRACT: durable objects from earlier conversations remain in
// the exchange directory. Only this run's exact generation ACK
@ -358,11 +419,18 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
it.optString("generation_id") == captured.generationId
}, pollTransport)
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)
val reference = JSONObject(referenceRaw.toString(Charsets.UTF_8))
phase(
runId,
"desktop_generation_observed",
"desktop-generation-v1.json",
reference.optString("generation_id"),
)
/* WHY: non-atomic transports can expose a newly downloaded
* reference locally before the corresponding immutable objects
* have reached this process. CONTRACT: correlation validates the
@ -373,8 +441,22 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
pollTransport?.invoke()
val acknowledgement =
service.consume(File(directory, reference.getString("relative_path")))
publish(File(directory, "android-consumption-ack-v1.json"), acknowledgement)
phase(
runId,
"desktop_generation_consumed",
generationId = reference.optString("generation_id"),
)
val acknowledgementPath = File(directory, "android-consumption-ack-v1.json")
publish(acknowledgementPath, acknowledgement)
visibility.confirm(listOf(acknowledgementPath))
phase(
runId,
"android_ack_published",
acknowledgementPath.name,
reference.optString("generation_id"),
)
afterAcknowledgement?.invoke()
phase(runId, "completed", result = "completed")
ForegroundGenerationResult.Completed(captured.runId)
} catch (error: Exception) {
if (error.message == "generation retention capacity exhausted") {
@ -391,11 +473,19 @@ internal class SyncGenerationCoordinator(private val repository: TrainlogReposit
.put("action", "Archive acknowledged generations, then retry explicitly.")
.toString(),
)
visibility.confirm(listOf(File(directory, "android-generation-error-v1.json")))
} catch (_: Exception) {
// Preserve the original local failure when error reporting
// cannot itself be durably published.
}
}
phase(
runCatching {
JSONObject(File(directory, "request-v1.json").readText()).optString("run_id")
}.getOrNull(),
"error",
result = error.javaClass.simpleName,
)
ForegroundGenerationResult.Error(error.message ?: "generation synchronization failed")
} finally {
conversationLock.unlock()

10
android/app/src/main/java/com/labfytools/trainlog/data/SyncRequestOutbox.kt
View file

@ -33,13 +33,16 @@ sealed interface SyncRequestResult {
class SyncRequestOutbox private constructor(
private val publisher: DirectExchangePublisher,
private val visibility: MtpPublicationVisibility,
) {
constructor(context: Context) : this(
DirectExchangePublisher { directExchangeDirectory(context) },
AndroidMtpPublicationVisibility(context),
)
internal constructor(directoryProvider: () -> DirectExchangeDirectoryAccess?) : this(
DirectExchangePublisher(directoryProvider),
ImmediateMtpPublicationVisibility,
)
suspend fun requestSync(): SyncRequestResult = withContext(Dispatchers.IO) {
@ -87,7 +90,12 @@ class SyncRequestOutbox private constructor(
"trainlog-sync-request-v1.json"
val error = snapshot.writeJson(displayName, payload)
return if (error == null) {
SyncRequestResult.Requested(requestId)
try {
visibility.confirm(listOf(java.io.File(canonicalExchangeDirectory(), displayName)))
SyncRequestResult.Requested(requestId)
} catch (error: Exception) {
SyncRequestResult.Error(error.message ?: "Publication MTP impossible.")
}
} else {
SyncRequestResult.Error(error)
}

22
android/app/src/main/java/com/labfytools/trainlog/ui/SyncScreen.kt
View file

@ -34,6 +34,8 @@ import com.labfytools.trainlog.data.BackgroundSyncSettings
import com.labfytools.trainlog.data.SyncBackgroundService
import com.labfytools.trainlog.data.DriveSyncSettings
import com.labfytools.trainlog.data.canonicalExchangeDirectory
import com.labfytools.trainlog.data.AndroidMtpPublicationVisibility
import com.labfytools.trainlog.data.MtpPublicationVisibility
import com.labfytools.trainlog.data.logDirectExchange
import com.labfytools.trainlog.data.directStoragePermissionIntent
import com.labfytools.trainlog.ui.theme.LocalTrainlogColors
@ -49,6 +51,7 @@ internal suspend fun publishBundleAndRequest(
repository: TrainlogRepository,
exporter: SyncExporter,
requestOutbox: SyncRequestOutbox,
visibility: MtpPublicationVisibility = com.labfytools.trainlog.data.ImmediateMtpPublicationVisibility,
): SyncRequestResult = withContext(Dispatchers.IO) {
val opened = exporter.openStorageSnapshot()
if (opened is com.labfytools.trainlog.data.DirectExchangeSnapshotResult.Error) {
@ -66,7 +69,7 @@ internal suspend fun publishBundleAndRequest(
is SyncExportResult.Exported -> Unit
}
return@withContext when (
val result = SyncGenerationForegroundCoordinator(repository).run(
val result = SyncGenerationForegroundCoordinator(repository, visibility).run(
canonicalExchangeDirectory(),
afterPeerPublication = {
when (val request = requestOutbox.requestSync(snapshot)) {
@ -390,10 +393,25 @@ 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)
}
syncRunning = true
coroutineScope.launch {
try {
when (val result = publishBundleAndRequest(repository, exporter, requestOutbox)) {
when (
val result = publishBundleAndRequest(
repository,
exporter,
requestOutbox,
AndroidMtpPublicationVisibility(context),
)
) {
is SyncRequestResult.Completed -> {
// The generation coordinator has already consumed the
// desktop return and published its durable ACK. Waiting

38
android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt
View file

@ -101,7 +101,15 @@ class SyncDeploymentConversationTest {
}
val conversations = if (suppliedTransport == null) 24 else 1
repeat(conversations) {
val coordinator = SyncGenerationForegroundCoordinator(repository)
val visibilityEvents = mutableListOf<List<String>>()
val tracePhases = mutableListOf<String>()
val coordinator = SyncGenerationForegroundCoordinator(
repository,
MtpPublicationVisibility { files ->
visibilityEvents += files.map(File::getName)
},
SyncGenerationTrace { _, phase, _, _, _ -> tracePhases += phase },
)
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
var requestSignals = 0
var refreshedAfterDesktopReference = false
@ -156,6 +164,34 @@ class SyncDeploymentConversationTest {
"transport is refreshed after the correlated desktop reference",
refreshedAfterDesktopReference,
)
val referenceVisibility = visibilityEvents.indexOfFirst {
it == listOf("android-generation-v1.json")
}
val objectVisibility = visibilityEvents.indexOfFirst { "manifest.json" in it }
assertTrue("generation objects must become visible", objectVisibility >= 0)
assertTrue(
"reference must follow complete object visibility",
referenceVisibility > objectVisibility,
)
assertTrue(
tracePhases.containsAll(
listOf(
"peer_published",
"legacy_trigger_published",
"generation_request_observed",
"archive_ack_observed",
"generation_capture_started",
"generation_captured",
"generation_objects_published",
"generation_reference_published",
"desktop_ack_observed",
"desktop_generation_observed",
"desktop_generation_consumed",
"android_ack_published",
"completed",
)
)
)
var workerOutput = ""
localWorker?.let { worker ->
workerOutput = worker.inputStream.bufferedReader().readText()

60
android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt Normal file
View file

@ -0,0 +1,60 @@
package com.labfytools.trainlog.data
import android.content.Context
import androidx.test.core.app.ApplicationProvider
import java.nio.file.Files
import java.time.Duration
import java.util.UUID
import java.util.concurrent.CountDownLatch
import java.util.concurrent.TimeUnit
import org.junit.Assert.assertEquals
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 SyncGenerationConversationLockTest {
private val context: Context = ApplicationProvider.getApplicationContext()
@Test
fun foregroundConversationCannotBeStolenByBackgroundCoordinator() {
val databaseName = "generation-lock-${UUID.randomUUID()}.db"
val root = Files.createTempDirectory("trainlog-generation-lock-").toFile()
val repository = TrainlogRepository(context, databaseName)
val entered = CountDownLatch(1)
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
try {
val foreground = SyncGenerationCoordinator(repository)
val background = SyncGenerationCoordinator(repository)
val active = executor.submit<ForegroundGenerationResult> {
foreground.run(
root,
Duration.ofMillis(300),
afterPeerPublication = { entered.countDown() },
)
}
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)
repository.inSyncGenerationTransaction { database ->
database.rawQuery("SELECT COUNT(*) FROM sync_generations", null).use { cursor ->
assertTrue(cursor.moveToFirst())
assertEquals(0, cursor.getInt(0))
}
}
} finally {
executor.shutdownNow()
repository.close()
context.deleteDatabase(databaseName)
root.deleteRecursively()
}
}
}

35
android/app/src/test/java/com/labfytools/trainlog/data/SyncRequestOutboxTest.kt Normal file
View file

@ -0,0 +1,35 @@
package com.labfytools.trainlog.data
import java.nio.file.Files
import org.json.JSONObject
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 SyncRequestOutboxTest {
@Test
fun distinctExplicitRequestsAlwaysPublishFreshIdentities() {
val directory = Files.createTempDirectory("trainlog-request-outbox-").toFile()
try {
val outbox = SyncRequestOutbox { directExchangeDirectoryForPath(directory) }
val snapshot =
(DirectExchangePublisher { directExchangeDirectoryForPath(directory) }.snapshot()
as DirectExchangeSnapshotResult.Ready).snapshot
val first = outbox.requestSync(snapshot) as SyncRequestResult.Requested
val firstBytes = java.io.File(directory, "trainlog-sync-request-v1.json").readText()
val second = outbox.requestSync(snapshot) as SyncRequestResult.Requested
val secondBytes = java.io.File(directory, "trainlog-sync-request-v1.json").readText()
assertNotEquals(first.requestId, second.requestId)
assertNotEquals(firstBytes, secondBytes)
assertTrue(JSONObject(secondBytes).getString("request_id") == second.requestId)
} finally {
directory.deleteRecursively()
}
}
}

4
tests/test_syncd_routing.py
View file

@ -79,7 +79,9 @@ class SyncdRoutingTest(unittest.TestCase):
production = source.split("internal suspend fun publishBundleAndRequest", 1)[1].split(
"internal suspend fun publishLegacyBundleAndRequest", 1
)[0]
self.assertIn("SyncGenerationForegroundCoordinator(repository).run", production)
self.assertIn(
"SyncGenerationForegroundCoordinator(repository, visibility).run", production
)
self.assertNotIn("generation-opt-in", production)
self.assertIn("exportMobileBundle", production)
self.assertIn("generation below", production)