Fix Android generation sync retry trigger
This commit is contained in:
parent
189307b548
commit
1be4ab0397
3 changed files with 36 additions and 4 deletions
|
|
@ -112,6 +112,7 @@ internal class SyncGenerationForegroundCoordinator(private val repository: Train
|
||||||
fun run(
|
fun run(
|
||||||
directory: File,
|
directory: File,
|
||||||
timeout: Duration = Duration.ofMinutes(5),
|
timeout: Duration = Duration.ofMinutes(5),
|
||||||
|
afterPeerPublication: (() -> Unit)? = null,
|
||||||
afterPublication: (() -> Unit)? = null,
|
afterPublication: (() -> Unit)? = null,
|
||||||
): ForegroundGenerationResult =
|
): ForegroundGenerationResult =
|
||||||
try {
|
try {
|
||||||
|
|
@ -139,6 +140,13 @@ internal class SyncGenerationForegroundCoordinator(private val repository: Train
|
||||||
)
|
)
|
||||||
.toString(),
|
.toString(),
|
||||||
)
|
)
|
||||||
|
/* 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
|
||||||
|
* 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()
|
||||||
val deadline = System.nanoTime() + timeout.toNanos()
|
val deadline = System.nanoTime() + timeout.toNanos()
|
||||||
val requestRaw =
|
val requestRaw =
|
||||||
awaitCorrelated(File(directory, "request-v1.json"), deadline) { candidate ->
|
awaitCorrelated(File(directory, "request-v1.json"), deadline) { candidate ->
|
||||||
|
|
|
||||||
|
|
@ -54,7 +54,26 @@ internal suspend fun publishBundleAndRequest(
|
||||||
value.opt("version") is Int && value.getInt("version") == 1 && value.opt("enabled") == true
|
value.opt("version") is Int && value.getInt("version") == 1 && value.opt("enabled") == true
|
||||||
} catch (_: Exception) { false }
|
} catch (_: Exception) { false }
|
||||||
if (!enabled) return@withContext SyncRequestResult.Error("Invalid generation synchronization opt-in.")
|
if (!enabled) return@withContext SyncRequestResult.Error("Invalid generation synchronization opt-in.")
|
||||||
return@withContext when (val result = SyncGenerationForegroundCoordinator(repository).run(canonicalExchangeDirectory())) {
|
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
|
||||||
|
return@withContext when (
|
||||||
|
val result = SyncGenerationForegroundCoordinator(repository).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.")
|
||||||
|
}
|
||||||
|
},
|
||||||
|
)
|
||||||
|
) {
|
||||||
is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId)
|
is ForegroundGenerationResult.Completed -> SyncRequestResult.Completed(result.runId)
|
||||||
is ForegroundGenerationResult.Error -> SyncRequestResult.Error(result.message)
|
is ForegroundGenerationResult.Error -> SyncRequestResult.Error(result.message)
|
||||||
}
|
}
|
||||||
|
|
|
||||||
|
|
@ -103,11 +103,15 @@ class SyncDeploymentConversationTest {
|
||||||
repeat(conversations) {
|
repeat(conversations) {
|
||||||
val coordinator = SyncGenerationForegroundCoordinator(repository)
|
val coordinator = SyncGenerationForegroundCoordinator(repository)
|
||||||
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
|
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
|
||||||
|
var requestSignals = 0
|
||||||
val result =
|
val result =
|
||||||
executor.submit<ForegroundGenerationResult> {
|
executor.submit<ForegroundGenerationResult> {
|
||||||
coordinator.run(transport, Duration.ofSeconds(30)) {
|
coordinator.run(
|
||||||
release?.let { awaitFile(File(it)) }
|
transport,
|
||||||
}
|
Duration.ofSeconds(30),
|
||||||
|
afterPeerPublication = { requestSignals += 1 },
|
||||||
|
afterPublication = { release?.let { awaitFile(File(it)) } },
|
||||||
|
)
|
||||||
}
|
}
|
||||||
val peer = JSONObject(awaitFile(File(transport, "android-peer-v1.json")).readText())
|
val peer = JSONObject(awaitFile(File(transport, "android-peer-v1.json")).readText())
|
||||||
if (suppliedTransport == null) {
|
if (suppliedTransport == null) {
|
||||||
|
|
@ -138,6 +142,7 @@ class SyncDeploymentConversationTest {
|
||||||
"conversation ${it + 1}: $completed",
|
"conversation ${it + 1}: $completed",
|
||||||
completed is ForegroundGenerationResult.Completed,
|
completed is ForegroundGenerationResult.Completed,
|
||||||
)
|
)
|
||||||
|
assertEquals("one fresh request signal per conversation", 1, requestSignals)
|
||||||
var workerOutput = ""
|
var workerOutput = ""
|
||||||
localWorker?.let { worker ->
|
localWorker?.let { worker ->
|
||||||
workerOutput = worker.inputStream.bufferedReader().readText()
|
workerOutput = worker.inputStream.bufferedReader().readText()
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue