diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/MtpPublicationVisibility.kt b/android/app/src/main/java/com/labfytools/trainlog/data/MtpPublicationVisibility.kt new file mode 100644 index 0000000..3ce0463 --- /dev/null +++ b/android/app/src/main/java/com/labfytools/trainlog/data/MtpPublicationVisibility.kt @@ -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) +} + +internal object ImmediateMtpPublicationVisibility : MtpPublicationVisibility { + override fun confirm(files: List) = Unit +} + +internal class AndroidMtpPublicationVisibility(context: Context) : MtpPublicationVisibility { + private val applicationContext = context.applicationContext + + override fun confirm(files: List) { + 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") + } + } +} 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 7371b77..ba8095a 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 @@ -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 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 f029336..8a9b2f7 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 @@ -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() diff --git a/android/app/src/main/java/com/labfytools/trainlog/data/SyncRequestOutbox.kt b/android/app/src/main/java/com/labfytools/trainlog/data/SyncRequestOutbox.kt index b3d468e..0df0f0c 100644 --- a/android/app/src/main/java/com/labfytools/trainlog/data/SyncRequestOutbox.kt +++ b/android/app/src/main/java/com/labfytools/trainlog/data/SyncRequestOutbox.kt @@ -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) } 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 f830f1d..2e7e17a 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 @@ -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 diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt index 4978bab..8ece9de 100644 --- a/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncDeploymentConversationTest.kt @@ -101,7 +101,15 @@ class SyncDeploymentConversationTest { } val conversations = if (suppliedTransport == null) 24 else 1 repeat(conversations) { - val coordinator = SyncGenerationForegroundCoordinator(repository) + val visibilityEvents = mutableListOf>() + val tracePhases = mutableListOf() + 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() 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 new file mode 100644 index 0000000..ba71b83 --- /dev/null +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncGenerationConversationLockTest.kt @@ -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 { + 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() + } + } +} diff --git a/android/app/src/test/java/com/labfytools/trainlog/data/SyncRequestOutboxTest.kt b/android/app/src/test/java/com/labfytools/trainlog/data/SyncRequestOutboxTest.kt new file mode 100644 index 0000000..2fc5ece --- /dev/null +++ b/android/app/src/test/java/com/labfytools/trainlog/data/SyncRequestOutboxTest.kt @@ -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() + } + } +} diff --git a/tests/test_syncd_routing.py b/tests/test_syncd_routing.py index 70c72de..f2b80d0 100644 --- a/tests/test_syncd_routing.py +++ b/tests/test_syncd_routing.py @@ -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)