fix(android): ignore stale sync coordination objects
This commit is contained in:
parent
d94ff495e3
commit
bad79ed926
2 changed files with 43 additions and 7 deletions
|
|
@ -36,6 +36,27 @@ internal class SyncGenerationForegroundCoordinator(private val repository: Train
|
||||||
return path
|
return path
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private fun awaitCorrelated(
|
||||||
|
path: File,
|
||||||
|
deadline: Long,
|
||||||
|
predicate: (JSONObject) -> Boolean,
|
||||||
|
): ByteArray {
|
||||||
|
while (true) {
|
||||||
|
if (path.isFile && path.length() <= 64 * 1024) {
|
||||||
|
try {
|
||||||
|
val raw = path.readBytes()
|
||||||
|
if (predicate(JSONObject(raw.toString(Charsets.UTF_8)))) return raw
|
||||||
|
} catch (_: Exception) {
|
||||||
|
// Publication replaces the file atomically, but an old or
|
||||||
|
// concurrently observed object is not evidence of failure.
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (System.nanoTime() >= deadline)
|
||||||
|
throw SyncGenerationException("timeout waiting for correlated ${path.name}")
|
||||||
|
Thread.sleep(50)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
fun run(
|
fun run(
|
||||||
directory: File,
|
directory: File,
|
||||||
timeout: Duration = Duration.ofMinutes(5),
|
timeout: Duration = Duration.ofMinutes(5),
|
||||||
|
|
@ -88,13 +109,20 @@ internal class SyncGenerationForegroundCoordinator(private val repository: Train
|
||||||
.toString(),
|
.toString(),
|
||||||
)
|
)
|
||||||
afterPublication?.invoke()
|
afterPublication?.invoke()
|
||||||
service.acceptAcknowledgement(
|
// CONTRACT: durable objects from earlier conversations remain in
|
||||||
await(File(directory, "desktop-consumption-ack-v1.json"), deadline).readBytes()
|
// the exchange directory. Only this run's exact generation ACK
|
||||||
)
|
// can advance its lineage; stale bytes are retained and ignored.
|
||||||
val reference =
|
val desktopAck =
|
||||||
JSONObject(
|
awaitCorrelated(File(directory, "desktop-consumption-ack-v1.json"), deadline) {
|
||||||
await(File(directory, "desktop-generation-v1.json"), deadline).readText()
|
it.optString("run_id") == captured.runId &&
|
||||||
)
|
it.optString("generation_id") == captured.generationId
|
||||||
|
}
|
||||||
|
service.acceptAcknowledgement(desktopAck)
|
||||||
|
val referenceRaw =
|
||||||
|
awaitCorrelated(File(directory, "desktop-generation-v1.json"), deadline) {
|
||||||
|
it.optString("run_id") == captured.runId
|
||||||
|
}
|
||||||
|
val reference = JSONObject(referenceRaw.toString(Charsets.UTF_8))
|
||||||
val acknowledgement =
|
val acknowledgement =
|
||||||
service.consume(File(directory, reference.getString("relative_path")))
|
service.consume(File(directory, reference.getString("relative_path")))
|
||||||
publish(File(directory, "android-consumption-ack-v1.json"), acknowledgement)
|
publish(File(directory, "android-consumption-ack-v1.json"), acknowledgement)
|
||||||
|
|
|
||||||
|
|
@ -57,6 +57,14 @@ class SyncDeploymentConversationTest {
|
||||||
var localWorker: Process? = null
|
var localWorker: Process? = null
|
||||||
try {
|
try {
|
||||||
transport.mkdirs()
|
transport.mkdirs()
|
||||||
|
// A previous interrupted exchange leaves durable coordination
|
||||||
|
// objects behind. The next run must wait for correlated replacements.
|
||||||
|
File(transport, "desktop-consumption-ack-v1.json").writeText(
|
||||||
|
"""{"run_id":"sy_11111111-1111-4111-8111-111111111111","generation_id":"gen_11111111-1111-4111-8111-111111111111"}"""
|
||||||
|
)
|
||||||
|
File(transport, "desktop-generation-v1.json").writeText(
|
||||||
|
"""{"run_id":"sy_11111111-1111-4111-8111-111111111111","generation_id":"gen_22222222-2222-4222-8222-222222222222"}"""
|
||||||
|
)
|
||||||
val exercise =
|
val exercise =
|
||||||
(repository.createExercise(
|
(repository.createExercise(
|
||||||
NewExerciseProfile(
|
NewExerciseProfile(
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue