fix(sync): release Android lease after desktop rejection
This commit is contained in:
parent
8b3a91cf50
commit
3d5dfa224a
4 changed files with 128 additions and 0 deletions
|
|
@ -14,6 +14,13 @@ new features, modules, protocols, or speculative refactors.
|
|||
`sync_in_progress`, or interruption) remain eligible after reconnect;
|
||||
completed requests and business failures remain consumed. Generation, ACK,
|
||||
causal, schema, MTP, Drive, and manual-sync contracts are unchanged.
|
||||
- Fixed Android generation coordination after a durable desktop rejection.
|
||||
The Android producer now retains and reports the correlated `rejected` ACK
|
||||
but immediately releases its conversation lease instead of waiting for a
|
||||
reciprocal desktop generation that the rejecting worker cannot publish.
|
||||
This prevents a rejected stale revision from starving a subsequent Web or
|
||||
automatic Bluetooth request; rejection, generation, ACK, causal, schema,
|
||||
and transport contracts are unchanged.
|
||||
|
||||
## 0.1.6 — 2026-09-25
|
||||
|
||||
|
|
|
|||
|
|
@ -606,6 +606,18 @@ internal class SyncGenerationCoordinator(
|
|||
}, pollTransport, lease)
|
||||
service.acceptAcknowledgement(desktopAck)
|
||||
phase(runId, "desktop_ack_observed", "desktop-consumption-ack-v1.json", captured.generationId)
|
||||
/* WHY: a durable desktop rejection closes this Android-produced
|
||||
* generation, so the desktop worker deliberately cannot publish a
|
||||
* reciprocal generation for this conversation. CONTRACT: retain
|
||||
* and validate the rejection ACK, then end this run immediately.
|
||||
* INVARIANT: a rejected generation never holds the process-wide
|
||||
* conversation lease while waiting for an impossible reference. */
|
||||
val desktopAcknowledgement = JSONObject(desktopAck.toString(Charsets.UTF_8))
|
||||
if (desktopAcknowledgement.getString("result") == "rejected") {
|
||||
val diagnostic = desktopAcknowledgement.getString("diagnostic")
|
||||
phase(runId, "desktop_rejected", "desktop-consumption-ack-v1.json", captured.generationId)
|
||||
return ForegroundGenerationResult.Failed(diagnostic, captured.runId)
|
||||
}
|
||||
val referenceRaw =
|
||||
awaitCorrelated(File(directory, "desktop-generation-v1.json"), deadline, {
|
||||
it.optString("run_id") == captured.runId
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import android.content.Context
|
|||
import androidx.test.core.app.ApplicationProvider
|
||||
import java.io.File
|
||||
import java.nio.file.Files
|
||||
import java.security.MessageDigest
|
||||
import java.time.Duration
|
||||
import java.util.UUID
|
||||
import java.util.concurrent.CountDownLatch
|
||||
|
|
@ -54,6 +55,109 @@ class SyncConversationOwnershipTest {
|
|||
)
|
||||
}
|
||||
|
||||
private fun canonical(value: Any?): String =
|
||||
when (value) {
|
||||
null,
|
||||
JSONObject.NULL -> "null"
|
||||
is JSONObject ->
|
||||
value.keys().asSequence().toList().sorted().joinToString(",", "{", "}") {
|
||||
JSONObject.quote(it) + ":" + canonical(value.get(it))
|
||||
}
|
||||
is JSONArray ->
|
||||
(0 until value.length()).joinToString(",", "[", "]") { canonical(value.get(it)) }
|
||||
is String -> JSONObject.quote(value)
|
||||
is Boolean,
|
||||
is Int,
|
||||
is Long -> value.toString()
|
||||
else -> error("unsupported test ACK value")
|
||||
}
|
||||
|
||||
private fun rejectedDesktopAcknowledgement(
|
||||
runId: String,
|
||||
generationId: String,
|
||||
producerPeerId: String,
|
||||
consumerPeerId: String,
|
||||
manifestSha256: String,
|
||||
): String {
|
||||
/* This fixture signs the exact wire ACK because the production
|
||||
* validator must persist it before the coordinator may release its
|
||||
* lease; an unsigned rejection would only test malformed-input logic. */
|
||||
val acknowledgement =
|
||||
JSONObject()
|
||||
.put("format", "trainlog-sync-ack")
|
||||
.put("version", 1)
|
||||
.put("ack_id", "ack_${generationId.removePrefix("gen_")}")
|
||||
.put("run_id", runId)
|
||||
.put("generation_id", generationId)
|
||||
.put("producer_peer_id", producerPeerId)
|
||||
.put("consumer_peer_id", consumerPeerId)
|
||||
.put("manifest_sha256", manifestSha256)
|
||||
.put("result", "rejected")
|
||||
.put("durability", "sqlite-commit-rejection")
|
||||
.put("consumed_at", "2026-09-28T12:00:00+02:00")
|
||||
.put("diagnostic", "concurrent sleep diary revision")
|
||||
val digest = MessageDigest.getInstance("SHA-256")
|
||||
.digest(canonical(acknowledgement).toByteArray())
|
||||
.joinToString("") { "%02x".format(it) }
|
||||
return acknowledgement.put("payload_sha256", digest).toString()
|
||||
}
|
||||
|
||||
@Test
|
||||
fun rejectedDesktopAcknowledgementEndsConversationAndReleasesBackgroundLease() {
|
||||
val databaseName = "generation-rejection-${UUID.randomUUID()}.db"
|
||||
val root = Files.createTempDirectory("trainlog-generation-rejection-").toFile()
|
||||
val repository = TrainlogRepository(context, databaseName)
|
||||
val arbiter = SyncConversationArbiter()
|
||||
val referencePublished = CountDownLatch(1)
|
||||
val desktopPeer = "peer_${UUID.randomUUID()}"
|
||||
val runId = "sy_${UUID.randomUUID()}"
|
||||
try {
|
||||
val service = SyncGenerationService(repository)
|
||||
request(root, runId, service.peerId(), desktopPeer)
|
||||
archiveAcknowledgements(root, runId, service.peerId(), desktopPeer)
|
||||
val coordinator =
|
||||
SyncGenerationCoordinator(
|
||||
repository,
|
||||
arbiter = arbiter,
|
||||
trace = SyncGenerationTrace { _, phase, _, _, _ ->
|
||||
if (phase == "generation_reference_published") referencePublished.countDown()
|
||||
},
|
||||
)
|
||||
val executor = java.util.concurrent.Executors.newSingleThreadExecutor()
|
||||
try {
|
||||
val active = executor.submit<ForegroundGenerationResult> {
|
||||
coordinator.run(root, Duration.ofSeconds(5))
|
||||
}
|
||||
assertTrue(referencePublished.await(3, TimeUnit.SECONDS))
|
||||
val reference = JSONObject(File(root, "android-generation-v1.json").readText())
|
||||
File(root, "desktop-consumption-ack-v1.json").writeText(
|
||||
rejectedDesktopAcknowledgement(
|
||||
runId,
|
||||
reference.getString("generation_id"),
|
||||
service.peerId(),
|
||||
desktopPeer,
|
||||
reference.getString("manifest_sha256"),
|
||||
)
|
||||
)
|
||||
val result = active.get(2, TimeUnit.SECONDS)
|
||||
assertEquals(
|
||||
ForegroundGenerationResult.Failed("concurrent sleep diary revision", runId),
|
||||
result,
|
||||
)
|
||||
assertTrue(
|
||||
"The rejected run must release the lease for its next request.",
|
||||
coordinator.run(root, Duration.ofMillis(100)) !is ForegroundGenerationResult.Busy,
|
||||
)
|
||||
} finally {
|
||||
executor.shutdownNow()
|
||||
}
|
||||
} finally {
|
||||
repository.close()
|
||||
context.deleteDatabase(databaseName)
|
||||
root.deleteRecursively()
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
fun explicitClickYieldsOldBackgroundAndPublishesFreshTriggerForNewRun() {
|
||||
val databaseName = "generation-ownership-${UUID.randomUUID()}.db"
|
||||
|
|
|
|||
|
|
@ -34,6 +34,11 @@ arbiter: explicit foreground intent has priority over automatic background
|
|||
resumption, while a foreground service may remain alive without owning the
|
||||
conversation. An automatic run yields cooperatively at safe boundaries; its
|
||||
durable generation is retained for a correlated handoff rather than deleted.
|
||||
A correlated desktop `rejected` ACK is also terminal for the Android producer:
|
||||
it is retained as durable rejection evidence, then releases the conversation
|
||||
lease immediately because no reciprocal desktop generation can follow that
|
||||
rejection. This keeps an explicit Web request or a later automatic Bluetooth
|
||||
request from being starved by an impossible post-rejection wait.
|
||||
|
||||
Direct-MTP publication now has an explicit Android visibility boundary.
|
||||
Durable immutable generation files are submitted to MediaProvider and all scan
|
||||
|
|
|
|||
Loading…
Reference in a new issue