Compare commits
No commits in common. "39a9df7b07775ddd8ef773bf5421a0cbf857bba7" and "8b3a91cf50410f25240041dc6894c0a1b3777262" have entirely different histories.
39a9df7b07
...
8b3a91cf50
14 changed files with 35 additions and 696 deletions
18
CHANGELOG.md
18
CHANGELOG.md
|
|
@ -14,24 +14,6 @@ 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.
|
||||
- Preserved the desktop consumer's known Sleep revision rejection diagnostic
|
||||
through the full-generation worker and classified it as `data_conflict` in
|
||||
Web run state. A synthetic divergent-final-wake regression keeps the true
|
||||
conflict rejected; this reporting fix does not reconcile real Sleep data or
|
||||
turn a rejected generation into a successful exchange.
|
||||
- Resolved safely comparable concurrent Sleep V2 edits by the last actual
|
||||
user-edit instant, regardless of device or the event times being recorded.
|
||||
Complete winning snapshots and both original immutable revisions are kept;
|
||||
a local winner gains a fresh causal successor for peer convergence. Equal
|
||||
timestamps, changed entry envelopes, and deletion transitions still reject
|
||||
instead of guessing. Existing rejected generation ACKs are not rewritten.
|
||||
|
||||
## 0.1.6 — 2026-09-25
|
||||
|
||||
|
|
|
|||
|
|
@ -606,18 +606,6 @@ 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
|
||||
|
|
|
|||
|
|
@ -676,9 +676,8 @@ class TrainlogRepository(
|
|||
}
|
||||
} catch (_: RuntimeException) { false }
|
||||
|
||||
/** CONTRACT: one local edit transaction appends an immutable revision and
|
||||
* advances the guarded tip. Cross-device conflict selection belongs only
|
||||
* to the versioned import path after both edit instants are validated. */
|
||||
/** CONTRACT: one transaction appends an immutable revision and advances
|
||||
* the guarded tip. Wall time never resolves a concurrent edit. */
|
||||
fun saveSleepDiary(draft: SleepDiaryDraft): SaveSleepDiaryResult =
|
||||
saveSleepDiary(draft, quickPublicationAt = null)
|
||||
|
||||
|
|
@ -1556,21 +1555,8 @@ class TrainlogRepository(
|
|||
}
|
||||
val remoteDescendsFromLocal =
|
||||
local != null && ancestry?.drop(1)?.contains(local.first) == true
|
||||
var winningLocalRevision: String? = null
|
||||
if (local != null && !remoteDescendsFromLocal && parent != local.first) {
|
||||
val localWins = sleepConcurrentLocalWins(
|
||||
db, item, entryId, local.first, wireVersion,
|
||||
)
|
||||
if (localWins) {
|
||||
if (knownOwner != null) {
|
||||
// A retained losing branch must not generate another
|
||||
// reconciliation revision when its artifact replays.
|
||||
unchanged++
|
||||
continue
|
||||
}
|
||||
winningLocalRevision = local.first
|
||||
}
|
||||
}
|
||||
if (local != null && !remoteDescendsFromLocal && parent != local.first)
|
||||
return SleepDiaryImportResult.Rejected("concurrent sleep diary revision")
|
||||
/* The snapshot may start at a non-root current revision on a
|
||||
* fresh peer. Preserve its causal parent identity even though
|
||||
* the historical parent payload is intentionally absent. */
|
||||
|
|
@ -1620,77 +1606,21 @@ class TrainlogRepository(
|
|||
arrayOf(entryId,draft.nightStartDate,draft.nightEndDate,draft.createdAt,draft.updatedAt,revisionId,if (deleted) 1 else 0))
|
||||
else db.execSQL("UPDATE sleep_diary_entries SET night_start_date=?,night_end_date=?,updated_at=?,current_revision_id=?,deleted=? WHERE entry_id=? AND current_revision_id=?",
|
||||
arrayOf(draft.nightStartDate,draft.nightEndDate,draft.updatedAt,revisionId,if (deleted) 1 else 0,entryId,local.first))
|
||||
if (knownOwner == null) {
|
||||
db.execSQL(
|
||||
"INSERT INTO sleep_diary_revisions(revision_id,entry_id," +
|
||||
"parent_revision_id,created_at,sleep_quality,wake_quality,day_form," +
|
||||
"treatment_and_notes) VALUES(?,?,?,?,?,?,?,?)",
|
||||
arrayOf(
|
||||
revisionId, entryId, parent, draft.updatedAt,
|
||||
draft.sleepQuality?.wireValue, draft.wakeQuality?.wireValue,
|
||||
draft.dayForm?.wireValue, draft.treatmentAndNotes,
|
||||
),
|
||||
)
|
||||
draft.events.forEach { event ->
|
||||
db.execSQL(
|
||||
"INSERT INTO sleep_diary_events(revision_id,event_id,event_type," +
|
||||
"start_at,end_at) VALUES(?,?,?,?,?)",
|
||||
arrayOf(revisionId, event.eventId, event.type.wireValue,
|
||||
event.startAt, event.endAt),
|
||||
)
|
||||
}
|
||||
draft.intakes.forEach { intake -> db.execSQL(
|
||||
"INSERT INTO sleep_medication_intakes(" +
|
||||
"revision_id,intake_id,medication_id,medication_name,taken_at," +
|
||||
"dose_value,dose_unit,quantity,note,created_at" +
|
||||
") VALUES(?,?,?,?,?,?,?,?,?,?)",
|
||||
arrayOf<Any?>(revisionId, intake.intakeId, intake.medicationId,
|
||||
intake.medicationName, intake.takenAt, intake.doseValue, intake.doseUnit,
|
||||
intake.quantity, intake.note, intake.createdAt),
|
||||
) }
|
||||
}
|
||||
db.execSQL("INSERT INTO sleep_diary_revisions VALUES(?,?,?,?,?,?,?,?)",
|
||||
arrayOf(revisionId,entryId,parent,draft.updatedAt,draft.sleepQuality?.wireValue,draft.wakeQuality?.wireValue,draft.dayForm?.wireValue,draft.treatmentAndNotes))
|
||||
draft.events.forEach { event -> db.execSQL("INSERT INTO sleep_diary_events VALUES(?,?,?,?,?)",
|
||||
arrayOf(revisionId,event.eventId,event.type.wireValue,event.startAt,event.endAt)) }
|
||||
draft.intakes.forEach { intake -> db.execSQL(
|
||||
"INSERT INTO sleep_medication_intakes(" +
|
||||
"revision_id,intake_id,medication_id,medication_name,taken_at," +
|
||||
"dose_value,dose_unit,quantity,note,created_at" +
|
||||
") VALUES(?,?,?,?,?,?,?,?,?,?)",
|
||||
arrayOf<Any?>(revisionId, intake.intakeId, intake.medicationId,
|
||||
intake.medicationName, intake.takenAt, intake.doseValue, intake.doseUnit,
|
||||
intake.quantity,
|
||||
intake.note, intake.createdAt),
|
||||
) }
|
||||
markSleepDiaryImported(db, entryId, revisionId, root.getString("generated_at"))
|
||||
if (winningLocalRevision != null) {
|
||||
val successor = sleepId("slr")
|
||||
// INVARIANT: this technical successor inherits the latest
|
||||
// user's edit timestamp and exact snapshot, not the import
|
||||
// time. Its remote parent makes the peer converge, while
|
||||
// both original sibling revisions stay immutable.
|
||||
db.execSQL(
|
||||
"INSERT INTO sleep_diary_revisions(revision_id,entry_id," +
|
||||
"parent_revision_id,created_at,sleep_quality,wake_quality,day_form," +
|
||||
"treatment_and_notes) " +
|
||||
"SELECT ?,entry_id,?,created_at,sleep_quality,wake_quality,day_form," +
|
||||
"treatment_and_notes FROM sleep_diary_revisions WHERE revision_id=?",
|
||||
arrayOf(successor, revisionId, winningLocalRevision),
|
||||
)
|
||||
db.execSQL(
|
||||
"INSERT INTO sleep_diary_events(revision_id,event_id,event_type," +
|
||||
"start_at,end_at) " +
|
||||
"SELECT ?,event_id,event_type,start_at,end_at FROM sleep_diary_events " +
|
||||
"WHERE revision_id=?",
|
||||
arrayOf(successor, winningLocalRevision),
|
||||
)
|
||||
db.execSQL(
|
||||
"INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id," +
|
||||
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) " +
|
||||
"SELECT ?,intake_id,medication_id,medication_name,taken_at,dose_value," +
|
||||
"dose_unit,quantity,note,created_at FROM sleep_medication_intakes " +
|
||||
"WHERE revision_id=?",
|
||||
arrayOf(successor, winningLocalRevision),
|
||||
)
|
||||
db.execSQL(
|
||||
"UPDATE sleep_diary_entries SET updated_at=(SELECT created_at FROM " +
|
||||
"sleep_diary_revisions WHERE revision_id=?),current_revision_id=? " +
|
||||
"WHERE entry_id=? AND current_revision_id=?",
|
||||
arrayOf(winningLocalRevision, successor, entryId, revisionId),
|
||||
)
|
||||
db.execSQL(
|
||||
"UPDATE sleep_diary_publication_state SET validated_revision_id=?," +
|
||||
"validated_at=? WHERE entry_id=?",
|
||||
arrayOf(successor, root.getString("generated_at"), entryId),
|
||||
)
|
||||
}
|
||||
advanced++
|
||||
}
|
||||
SleepDiaryImportResult.Applied(advanced, unchanged)
|
||||
|
|
@ -1699,40 +1629,6 @@ class TrainlogRepository(
|
|||
}
|
||||
}
|
||||
|
||||
private fun sleepConcurrentLocalWins(
|
||||
db: SQLiteDatabase,
|
||||
item: JSONObject,
|
||||
entryId: String,
|
||||
localRevision: String,
|
||||
wireVersion: Int,
|
||||
): Boolean {
|
||||
// WHY: generation/ACK timestamps are transport events. Only revision
|
||||
// creation times reflect editing actions on the two clients.
|
||||
// CONTRACT: V2 sibling tips with an unchanged entry envelope and
|
||||
// deletion state may use last-user-edit-wins. An equal edit instant
|
||||
// or inconsistent provenance remains an explicit conflict.
|
||||
if (wireVersion != 2) throw IllegalArgumentException("concurrent sleep diary revision")
|
||||
val local = db.rawQuery(
|
||||
"SELECT e.night_start_date,e.night_end_date,e.created_at,e.updated_at,e.deleted," +
|
||||
"r.created_at FROM sleep_diary_entries e JOIN sleep_diary_revisions r " +
|
||||
"ON r.revision_id=e.current_revision_id WHERE e.entry_id=? AND r.revision_id=?",
|
||||
arrayOf(entryId, localRevision),
|
||||
).use { cursor ->
|
||||
if (!cursor.moveToFirst()) null else List(6) { index -> cursor.getString(index) }
|
||||
} ?: throw IllegalArgumentException("concurrent sleep diary revision")
|
||||
if (local[0] != item.getString("night_start_date") ||
|
||||
local[1] != item.getString("night_end_date") ||
|
||||
local[2] != item.getString("created_at") ||
|
||||
(local[4] == "1") != item.getBoolean("deleted") || local[3] != local[5]
|
||||
) throw IllegalArgumentException("concurrent sleep diary revision")
|
||||
val localAt = OffsetDateTime.parse(local[5]).toInstant()
|
||||
val remoteAt = OffsetDateTime.parse(item.getString("updated_at")).toInstant()
|
||||
if (localAt == remoteAt) throw IllegalArgumentException(
|
||||
"concurrent sleep diary revision has indeterminate edit order",
|
||||
)
|
||||
return localAt > remoteAt
|
||||
}
|
||||
|
||||
private fun sleepDiaryRevisionMatches(
|
||||
db: SQLiteDatabase,
|
||||
item: JSONObject,
|
||||
|
|
|
|||
|
|
@ -13,7 +13,6 @@ import com.labfytools.trainlog.model.MedicationIntake
|
|||
import com.labfytools.trainlog.model.SleepMedication
|
||||
import java.util.UUID
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotEquals
|
||||
import org.junit.Assert.assertThrows
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Test
|
||||
|
|
@ -26,147 +25,6 @@ import org.robolectric.annotation.Config
|
|||
class SleepDiaryRepositoryTest {
|
||||
private val context: Context = ApplicationProvider.getApplicationContext()
|
||||
|
||||
@Test
|
||||
fun concurrentV2SleepEditsChooseLatestUserEditAndConverge() {
|
||||
for (localNewer in listOf(true, false)) {
|
||||
val sourceName = "sleep-lww-source-${UUID.randomUUID()}.db"
|
||||
val destinationName = "sleep-lww-destination-${UUID.randomUUID()}.db"
|
||||
val source = TrainlogRepository(context, sourceName)
|
||||
val destination = TrainlogRepository(context, destinationName)
|
||||
try {
|
||||
val initial = source.saveSleepDiary(
|
||||
SleepDiaryDraft(
|
||||
nightStartDate = "2026-10-24",
|
||||
nightEndDate = "2026-10-25",
|
||||
createdAt = "2026-10-24T22:00:00+02:00",
|
||||
updatedAt = "2026-10-24T22:00:00+02:00",
|
||||
sleepQuality = null,
|
||||
wakeQuality = null,
|
||||
dayForm = null,
|
||||
treatmentAndNotes = "",
|
||||
events = listOf(
|
||||
SleepDiaryEvent(
|
||||
"", SleepEventType.SLEEP,
|
||||
"2026-10-24T23:00:00+02:00",
|
||||
"2026-10-25T07:00:00+01:00",
|
||||
),
|
||||
),
|
||||
),
|
||||
) as TrainlogRepository.SaveSleepDiaryResult.Saved
|
||||
assertTrue(source.validateSleepDiary(
|
||||
initial.entryId,
|
||||
initial.revisionId,
|
||||
"2026-10-25T08:00:00+01:00",
|
||||
) is TrainlogRepository.SaveSleepDiaryResult.Saved)
|
||||
assertEquals(
|
||||
TrainlogRepository.SleepDiaryImportResult.Applied(1, 0),
|
||||
destination.applySleepDiaryV1Json(source.buildSleepDiaryV1Json()),
|
||||
)
|
||||
|
||||
fun edit(repository: TrainlogRepository, at: String, wakeAt: String): String {
|
||||
val current = repository.listSleepDiary().single()
|
||||
val saved = repository.saveSleepDiary(
|
||||
SleepDiaryDraft(
|
||||
entryId = current.entryId,
|
||||
expectedRevision = current.revisionId,
|
||||
nightStartDate = current.nightStartDate,
|
||||
nightEndDate = current.nightEndDate,
|
||||
createdAt = current.createdAt,
|
||||
updatedAt = at,
|
||||
sleepQuality = null,
|
||||
wakeQuality = null,
|
||||
dayForm = null,
|
||||
treatmentAndNotes = "",
|
||||
events = current.events.map { it.copy(endAt = wakeAt) },
|
||||
),
|
||||
) as TrainlogRepository.SaveSleepDiaryResult.Saved
|
||||
assertTrue(repository.validateSleepDiary(
|
||||
current.entryId, saved.revisionId, at,
|
||||
) is TrainlogRepository.SaveSleepDiaryResult.Saved)
|
||||
return saved.revisionId
|
||||
}
|
||||
|
||||
val localAt = if (localNewer) "2026-10-25T10:00:00+01:00"
|
||||
else "2026-10-25T09:00:00+01:00"
|
||||
val remoteAt = if (localNewer) "2026-10-25T09:00:00+01:00"
|
||||
else "2026-10-25T10:00:00+01:00"
|
||||
val localRevision = edit(destination, localAt, "2026-10-25T06:00:00+01:00")
|
||||
val remoteRevision = edit(source, remoteAt, "2026-10-25T07:30:00+01:00")
|
||||
val remoteSnapshot = source.buildSleepDiaryV1Json()
|
||||
val tiedSnapshot = org.json.JSONObject(remoteSnapshot)
|
||||
val tiedEntry = tiedSnapshot.getJSONArray("entries").getJSONObject(0)
|
||||
val tiedRevision = "slr_${UUID.randomUUID()}"
|
||||
tiedEntry.put("revision_id", tiedRevision)
|
||||
tiedEntry.put("updated_at", localAt)
|
||||
tiedEntry.put(
|
||||
"ancestry",
|
||||
org.json.JSONArray().put(tiedRevision).put(initial.revisionId),
|
||||
)
|
||||
assertEquals(
|
||||
TrainlogRepository.SleepDiaryImportResult.Rejected(
|
||||
"concurrent sleep diary revision has indeterminate edit order",
|
||||
),
|
||||
destination.applySleepDiaryV1Json(tiedSnapshot.toString()),
|
||||
)
|
||||
assertEquals(localRevision, destination.listSleepDiary().single().revisionId)
|
||||
assertEquals(
|
||||
TrainlogRepository.SleepDiaryImportResult.Applied(1, 0),
|
||||
destination.applySleepDiaryV1Json(remoteSnapshot),
|
||||
)
|
||||
val winner = destination.listSleepDiary().single()
|
||||
assertEquals(
|
||||
if (localNewer) "2026-10-25T06:00:00+01:00"
|
||||
else "2026-10-25T07:30:00+01:00",
|
||||
winner.events.single().endAt,
|
||||
)
|
||||
assertEquals(
|
||||
TrainlogRepository.SleepDiaryImportResult.Applied(0, 1),
|
||||
destination.applySleepDiaryV1Json(remoteSnapshot),
|
||||
)
|
||||
val reconciled = destination.buildSleepDiaryV1Json()
|
||||
if (localNewer) {
|
||||
assertNotEquals(localRevision, winner.revisionId)
|
||||
assertEquals(
|
||||
remoteRevision,
|
||||
org.json.JSONObject(reconciled).getJSONArray("entries")
|
||||
.getJSONObject(0).getString("parent_revision_id"),
|
||||
)
|
||||
} else {
|
||||
assertEquals(remoteRevision, winner.revisionId)
|
||||
}
|
||||
assertEquals(
|
||||
TrainlogRepository.SleepDiaryImportResult.Applied(
|
||||
if (localNewer) 1 else 0, if (localNewer) 0 else 1,
|
||||
),
|
||||
source.applySleepDiaryV1Json(reconciled),
|
||||
)
|
||||
assertEquals(winner.revisionId, source.listSleepDiary().single().revisionId)
|
||||
val retained = SQLiteDatabase.openDatabase(
|
||||
context.getDatabasePath(destinationName).path,
|
||||
null,
|
||||
SQLiteDatabase.OPEN_READONLY,
|
||||
)
|
||||
try {
|
||||
assertEquals(2, retained.rawQuery(
|
||||
"SELECT COUNT(*) FROM sleep_diary_revisions " +
|
||||
"WHERE revision_id IN (?,?)",
|
||||
arrayOf(localRevision, remoteRevision),
|
||||
).use { cursor ->
|
||||
assertTrue(cursor.moveToFirst())
|
||||
cursor.getInt(0)
|
||||
})
|
||||
} finally {
|
||||
retained.close()
|
||||
}
|
||||
} finally {
|
||||
source.close()
|
||||
destination.close()
|
||||
context.deleteDatabase(sourceName)
|
||||
context.deleteDatabase(destinationName)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Test fun createEditConflictDeleteAndReopen() {
|
||||
val name = "sleep-${UUID.randomUUID()}.db"
|
||||
var repository = TrainlogRepository(context, name)
|
||||
|
|
|
|||
|
|
@ -4,7 +4,6 @@ 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
|
||||
|
|
@ -55,109 +54,6 @@ 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,11 +34,6 @@ 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
|
||||
|
|
@ -163,7 +158,7 @@ parallel implementation of its rules.
|
|||
| Web Dashboard tiles | `WEB_DASHBOARD_TILES_V1=PASS/FROZEN` |
|
||||
| Web Dashboard visualizations | `WEB_DASHBOARD_VISUALIZATIONS_V1=PASS/FROZEN` |
|
||||
| Web Dashboard V1 | `WEB_DASHBOARD_V1=PASS/FROZEN` |
|
||||
| Sleep Diary V1/V2 | `SLEEP_DIARY_V1=PASS` remains frozen and readable. The active Sleep exchange requires `sleep-diary-v2` / `sleep-diary-v2.json`: structured per-intake quantity and quick-publishable current revisions supplement the existing causal diary model without changing V1. Distinct concurrent V2 edits with comparable user-edit instants retain the complete latest snapshot and both immutable histories; ambiguous ordering and unsafe deletion/envelope changes still reject. |
|
||||
| Sleep Diary V1/V2 | `SLEEP_DIARY_V1=PASS` remains frozen and readable. The active Sleep exchange requires `sleep-diary-v2` / `sleep-diary-v2.json`: structured per-intake quantity and quick-publishable current revisions supplement the existing causal diary model without changing V1. |
|
||||
| Web Sessions V1 | `TRAINLOG_WEB_SESSIONS_V1=PASS/FROZEN` (controlled desktop/Android deployment validated) |
|
||||
| Web Sessions deletion and Programs V1 | `TRAINLOG_WEB_SESSIONS_DELETE_AND_PROGRAMS_V1=PASS` (private grouped rollout validated) |
|
||||
| Programs presentation, Android projection, and deletion | `TRAINLOG_PROGRAMS_PRESENTATION_ANDROID_DELETE_V1=PASS` (private coordinated deployment and restart/replay validated) |
|
||||
|
|
|
|||
|
|
@ -32,25 +32,8 @@ offset-bearing timestamps, appreciations, notes, logical deletion, and an
|
|||
integer `quantity` in `1..99` on every medication intake. `dose_value` and
|
||||
`dose_unit` remain the per-unit snapshot; quantity is never encoded in notes.
|
||||
V1 input omits `quantity` and is normalized to one unit. Exact revision replay
|
||||
is unchanged; resurrection after deletion is rejected in the generation
|
||||
consumption transaction. For distinct V2 sibling tips of the same entry with
|
||||
an unchanged night envelope and deletion state, the last actual user edit
|
||||
wins, independent of device. Revision edit timestamps are compared as instants;
|
||||
event times, generation publication, import, and ACK times never establish
|
||||
priority. An equal edit instant or inconsistent provenance remains an explicit
|
||||
conflict. This selection uses the complete winning snapshot, not a field or
|
||||
event union, so an intentional removal is not resurrected. The losing revision
|
||||
remains immutable in local history. When the local snapshot wins, a fresh
|
||||
technical successor descends from the received revision and retains the
|
||||
winning user-edit timestamp and exact payload, allowing the other peer to
|
||||
advance causally without rewriting either original revision or generation.
|
||||
|
||||
A durable Sleep rejection keeps its exact ACK and does not advance either
|
||||
causal tip. The full-generation worker carries the consumer's bounded Sleep
|
||||
diagnostic through to the Web run state, where an unresolved concurrent
|
||||
revision is reported as `data_conflict` rather than an opaque `internal_error`.
|
||||
An old rejected generation remains rejected after a later, newly captured
|
||||
generation resolves its distinct conflict.
|
||||
is unchanged; sibling revisions and resurrection after deletion are rejected
|
||||
in the generation consumption transaction.
|
||||
|
||||
This does not change `TRAINLOG_FORMAT_V1`, manifests, ACKs or causal
|
||||
ownership.
|
||||
|
|
|
|||
|
|
@ -1,15 +1,5 @@
|
|||
# Tests and validation
|
||||
|
||||
## Sleep V2 concurrent-edit selection in 0.1.7
|
||||
|
||||
Synthetic Desktop and Android regressions cover both edit-time orderings,
|
||||
including a later user edit that records an earlier sleep event. They check
|
||||
the complete winning snapshot, retained sibling revisions, no event/intake
|
||||
union or duplication, causal successor import on the other peer, exact replay,
|
||||
and rejection when edit instants are equal. The real paired-device validation
|
||||
completed separate automatic, Web-with-automatic-service, and post-reconnection
|
||||
Bluetooth generation/ACK cycles without changing an old rejected ACK.
|
||||
|
||||
## Sleep revision collision recovery
|
||||
|
||||
Collision regressions distinguish exact replay from same-ID parent or payload
|
||||
|
|
|
|||
|
|
@ -246,131 +246,6 @@ class SleepExchangeTest(unittest.TestCase):
|
|||
with self.assertRaisesRegex(ValueError, "positive"):
|
||||
exchange.validate(self.document(invalid))
|
||||
|
||||
def test_contradictory_final_get_up_siblings_remain_unresolved(self):
|
||||
base = "slr_20000000-0000-4000-8000-000000000002"
|
||||
desktop_tip = "slr_40000000-0000-4000-8000-000000000004"
|
||||
android_parent = "slr_50000000-0000-4000-8000-000000000005"
|
||||
android_tip = "slr_60000000-0000-4000-8000-000000000006"
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, self.document(entry(base))))
|
||||
|
||||
desktop = entry(desktop_tip, base)
|
||||
desktop["events"].append({
|
||||
"event_id": "sle_70000000-0000-4000-8000-000000000007",
|
||||
"type": "final_get_up", "start_at": "2026-10-25T06:30:00+01:00",
|
||||
"end_at": None,
|
||||
})
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, self.document(desktop)))
|
||||
before = self.db.execute(
|
||||
"SELECT revision_id,parent_revision_id FROM sleep_diary_revisions "
|
||||
"ORDER BY revision_id"
|
||||
).fetchall()
|
||||
|
||||
android = entry(android_tip, android_parent)
|
||||
android["ancestry"] = [android_tip, android_parent, base]
|
||||
android["events"][0]["end_at"] = "2026-10-25T06:50:00+01:00"
|
||||
android["events"].append({
|
||||
"event_id": "sle_80000000-0000-4000-8000-000000000008",
|
||||
"type": "final_get_up", "start_at": "2026-10-25T06:50:00+01:00",
|
||||
"end_at": None,
|
||||
})
|
||||
with self.assertRaisesRegex(ValueError, "concurrent sleep diary revision"):
|
||||
exchange.apply(self.db, self.document(android))
|
||||
|
||||
self.assertEqual(before, self.db.execute(
|
||||
"SELECT revision_id,parent_revision_id FROM sleep_diary_revisions "
|
||||
"ORDER BY revision_id"
|
||||
).fetchall())
|
||||
self.assertEqual(desktop_tip, self.db.execute(
|
||||
"SELECT current_revision_id FROM sleep_diary_entries"
|
||||
).fetchone()[0])
|
||||
self.assertEqual(1, self.db.execute(
|
||||
"SELECT COUNT(*) FROM sleep_diary_events WHERE revision_id=? "
|
||||
"AND event_type='final_get_up'", (desktop_tip,)
|
||||
).fetchone()[0])
|
||||
|
||||
def test_v2_newer_local_edit_converges_without_using_event_time(self):
|
||||
base = "slr_20000000-0000-4000-8000-000000000002"
|
||||
local_id = "slr_40000000-0000-4000-8000-000000000004"
|
||||
remote_id = "slr_50000000-0000-4000-8000-000000000005"
|
||||
|
||||
def v2(item):
|
||||
document = self.document(item)
|
||||
document["version"] = 2
|
||||
document["entries"][0]["intakes"][0]["quantity"] = 1
|
||||
return document
|
||||
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, v2(entry(base))))
|
||||
self.db.commit()
|
||||
peer = sqlite3.connect(":memory:")
|
||||
self.db.backup(peer)
|
||||
try:
|
||||
local = entry(local_id, base)
|
||||
local["updated_at"] = "2026-10-25T20:00:00+01:00"
|
||||
local["events"][0]["end_at"] = "2026-10-25T06:00:00+01:00"
|
||||
remote = entry(remote_id, base)
|
||||
remote["updated_at"] = "2026-10-25T19:00:00+01:00"
|
||||
remote["events"][0]["end_at"] = "2026-10-25T07:00:00+01:00"
|
||||
remote["events"].append({
|
||||
"event_id": "sle_60000000-0000-4000-8000-000000000006",
|
||||
"type": "final_get_up",
|
||||
"start_at": "2026-10-25T07:00:00+01:00",
|
||||
"end_at": None,
|
||||
})
|
||||
extra_intake = dict(remote["intakes"][0])
|
||||
extra_intake["intake_id"] = "mdi_60000000-0000-4000-8000-000000000006"
|
||||
extra_intake["quantity"] = 2
|
||||
remote["intakes"].append(extra_intake)
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, v2(local)))
|
||||
self.assertEqual((1, 0), exchange.apply(peer, v2(remote)))
|
||||
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, exchange.build(peer)))
|
||||
successor = exchange.build(self.db)["entries"][0]
|
||||
self.assertEqual(remote_id, successor["parent_revision_id"])
|
||||
self.assertEqual(local["events"], successor["events"])
|
||||
self.assertEqual(1, len(successor["events"]))
|
||||
self.assertEqual(local["updated_at"], successor["updated_at"])
|
||||
self.assertEqual(1, len(successor["intakes"]))
|
||||
self.assertEqual(1, successor["intakes"][0]["quantity"])
|
||||
self.assertEqual((0, 1), exchange.apply(self.db, exchange.build(peer)))
|
||||
self.assertEqual((1, 0), exchange.apply(peer, exchange.build(self.db)))
|
||||
self.assertEqual((0, 1), exchange.apply(peer, exchange.build(self.db)))
|
||||
self.assertEqual(
|
||||
successor["revision_id"],
|
||||
exchange.build(peer)["entries"][0]["revision_id"],
|
||||
)
|
||||
self.assertEqual(4, self.db.execute(
|
||||
"SELECT COUNT(*) FROM sleep_diary_revisions"
|
||||
).fetchone()[0])
|
||||
self.assertEqual("ok", self.db.execute("PRAGMA integrity_check").fetchone()[0])
|
||||
finally:
|
||||
peer.close()
|
||||
|
||||
def test_v2_newer_remote_edit_wins_and_equal_time_rejects(self):
|
||||
base = "slr_20000000-0000-4000-8000-000000000002"
|
||||
local = entry("slr_40000000-0000-4000-8000-000000000004", base)
|
||||
remote = entry("slr_50000000-0000-4000-8000-000000000005", base)
|
||||
|
||||
def v2(item):
|
||||
document = self.document(item)
|
||||
document["version"] = 2
|
||||
document["entries"][0]["intakes"][0]["quantity"] = 1
|
||||
return document
|
||||
|
||||
exchange.apply(self.db, v2(entry(base)))
|
||||
local["updated_at"] = "2026-10-25T19:00:00+01:00"
|
||||
remote["updated_at"] = "2026-10-25T20:00:00+01:00"
|
||||
local["events"][0]["end_at"] = "2026-10-25T07:00:00+01:00"
|
||||
remote["events"][0]["end_at"] = "2026-10-25T06:00:00+01:00"
|
||||
exchange.apply(self.db, v2(local))
|
||||
tied = entry("slr_60000000-0000-4000-8000-000000000006", base)
|
||||
tied["updated_at"] = local["updated_at"]
|
||||
with self.assertRaisesRegex(ValueError, "indeterminate edit order"):
|
||||
exchange.apply(self.db, v2(tied))
|
||||
self.assertEqual((1, 0), exchange.apply(self.db, v2(remote)))
|
||||
self.assertEqual(remote["revision_id"], exchange.build(self.db)["entries"][0]["revision_id"])
|
||||
self.assertEqual(remote["events"], exchange.build(self.db)["entries"][0]["events"])
|
||||
self.assertEqual((0, 1), exchange.apply(self.db, v2(local)))
|
||||
|
||||
def test_draft_is_durable_but_not_exported_until_validated(self):
|
||||
revision = "slr_20000000-0000-4000-8000-000000000002"
|
||||
item = entry(revision)
|
||||
|
|
|
|||
|
|
@ -159,12 +159,6 @@ class SyncErrorClassificationTest(unittest.TestCase):
|
|||
sync_orchestrator.failure_code(RuntimeError("revision conflict")),
|
||||
"data_conflict",
|
||||
)
|
||||
self.assertEqual(
|
||||
sync_orchestrator.failure_code(RuntimeError(
|
||||
"desktop rejected Android generation: concurrent sleep diary revision"
|
||||
)),
|
||||
"data_conflict",
|
||||
)
|
||||
|
||||
def test_mtp_outbox_excludes_retained_staging_and_unrelated_files(self):
|
||||
with tempfile.TemporaryDirectory() as directory:
|
||||
|
|
|
|||
|
|
@ -26,19 +26,6 @@ class SyncPeerWorkerTest(unittest.TestCase):
|
|||
def tearDown(self):
|
||||
self.temporary.cleanup()
|
||||
|
||||
def test_rejected_android_generation_retains_consumer_diagnostic(self):
|
||||
error = worker.rejected_android_generation_error(
|
||||
{"diagnostic": "concurrent sleep diary revision"}
|
||||
)
|
||||
self.assertEqual(
|
||||
str(error),
|
||||
"desktop rejected Android generation: concurrent sleep diary revision",
|
||||
)
|
||||
self.assertEqual(
|
||||
str(worker.rejected_android_generation_error({"diagnostic": ""})),
|
||||
"desktop rejected Android generation",
|
||||
)
|
||||
|
||||
def test_drive_adapter_uses_remaining_conversation_deadline(self):
|
||||
completed = subprocess.CompletedProcess([], 0, b"", b"")
|
||||
with mock.patch.object(worker.subprocess, "run", return_value=completed) as run:
|
||||
|
|
|
|||
|
|
@ -5,7 +5,6 @@ import argparse
|
|||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import uuid
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -379,42 +378,6 @@ def revision_is_ancestor(db: sqlite3.Connection, parent_sql: str, owner_id: str,
|
|||
return False
|
||||
|
||||
|
||||
def concurrent_entry_winner(db: sqlite3.Connection, item: dict, local_revision: str,
|
||||
wire_version: int) -> dict | None:
|
||||
"""Choose a complete Sleep snapshot only when user-edit instants are ordered.
|
||||
|
||||
WHY: an import or ACK is not a new user edit. Revision creation times are
|
||||
recorded by the editing clients, whereas generation times are transport
|
||||
times. CONTRACT: only V2 sibling tips with the same entry envelope and no
|
||||
deletion transition admit last-user-edit-wins. Equal or inconsistent
|
||||
timestamps remain conflicts, never a peer-order tie break.
|
||||
"""
|
||||
if wire_version != 2:
|
||||
fail("concurrent sleep diary revision")
|
||||
row = db.execute(
|
||||
"SELECT e.night_start_date,e.night_end_date,e.created_at,e.updated_at,e.deleted,"
|
||||
"r.created_at FROM sleep_diary_entries e JOIN sleep_diary_revisions r "
|
||||
"ON r.revision_id=e.current_revision_id WHERE e.entry_id=? AND r.revision_id=?",
|
||||
(item["entry_id"], local_revision),
|
||||
).fetchone()
|
||||
if (row is None or tuple(row[:3]) !=
|
||||
(item["night_start_date"], item["night_end_date"], item["created_at"]) or
|
||||
bool(row[4]) != item["deleted"] or row[3] != row[5]):
|
||||
fail("concurrent sleep diary revision")
|
||||
local_at = datetime.fromisoformat(timestamp(row[5]).replace("Z", "+00:00"))
|
||||
remote_at = datetime.fromisoformat(timestamp(item["updated_at"]).replace("Z", "+00:00"))
|
||||
if local_at == remote_at:
|
||||
fail("concurrent sleep diary revision has indeterminate edit order")
|
||||
if local_at < remote_at:
|
||||
return None
|
||||
current = db.execute(
|
||||
"SELECT entry_id,night_start_date,night_end_date,created_at,updated_at,"
|
||||
"current_revision_id,deleted FROM sleep_diary_entries WHERE entry_id=?",
|
||||
(item["entry_id"],),
|
||||
).fetchone()
|
||||
return _persisted_entry(db, current)
|
||||
|
||||
|
||||
def apply(db: sqlite3.Connection, root: dict) -> tuple[int, int]:
|
||||
validate(root)
|
||||
wire_version = root["version"]
|
||||
|
|
@ -521,14 +484,8 @@ def apply(db: sqlite3.Connection, root: dict) -> tuple[int, int]:
|
|||
unchanged += 1
|
||||
continue
|
||||
remote_descends_from_local = local is not None and local[0] in item.get("ancestry", [])[1:]
|
||||
winning_local = None
|
||||
if local is not None and not remote_descends_from_local and item["parent_revision_id"] != local[0]:
|
||||
winning_local = concurrent_entry_winner(db, item, local[0], wire_version)
|
||||
if known_owner is not None and winning_local is not None:
|
||||
# A previously retained losing branch cannot cause a fresh
|
||||
# reconciliation revision on every immutable replay.
|
||||
unchanged += 1
|
||||
continue
|
||||
fail("concurrent sleep diary revision")
|
||||
# WHY: one-tap Sleep creates several local revisions before its first
|
||||
# synchronization. The current snapshot carries their bounded lineage
|
||||
# by identity, so a new desktop must be able to seed the tip without
|
||||
|
|
@ -548,30 +505,17 @@ def apply(db: sqlite3.Connection, root: dict) -> tuple[int, int]:
|
|||
"current_revision_id=?,deleted=? WHERE entry_id=? AND current_revision_id=?",
|
||||
(item["night_start_date"], item["night_end_date"], item["updated_at"],
|
||||
item["revision_id"], int(item["deleted"]), item["entry_id"], local[0]))
|
||||
if known_owner is None:
|
||||
db.execute(
|
||||
"INSERT INTO sleep_diary_revisions(revision_id,entry_id,parent_revision_id,"
|
||||
"created_at,sleep_quality,wake_quality,day_form,treatment_and_notes) "
|
||||
"VALUES(?,?,?,?,?,?,?,?)",
|
||||
(item["revision_id"], item["entry_id"], item["parent_revision_id"],
|
||||
item["updated_at"], item["sleep_quality"], item["wake_quality"],
|
||||
item["day_form"], item["treatment_and_notes"]),
|
||||
)
|
||||
db.executemany(
|
||||
"INSERT INTO sleep_diary_events(revision_id,event_id,event_type,start_at,end_at) "
|
||||
"VALUES(?,?,?,?,?)",
|
||||
[(item["revision_id"], event["event_id"], event["type"],
|
||||
event["start_at"], event["end_at"]) for event in item["events"]],
|
||||
)
|
||||
db.executemany(
|
||||
"INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id,"
|
||||
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) "
|
||||
"VALUES(?,?,?,?,?,?,?,?,?,?)",
|
||||
[(item["revision_id"], intake["intake_id"], intake["medication_id"],
|
||||
intake["medication_name"], intake["taken_at"], intake["dose_value"],
|
||||
intake["dose_unit"], intake.get("quantity", 1), intake["note"],
|
||||
intake["created_at"]) for intake in item["intakes"]],
|
||||
)
|
||||
db.execute("INSERT INTO sleep_diary_revisions VALUES(?,?,?,?,?,?,?,?)",
|
||||
(item["revision_id"], item["entry_id"], item["parent_revision_id"], item["updated_at"],
|
||||
item["sleep_quality"], item["wake_quality"], item["day_form"], item["treatment_and_notes"]))
|
||||
db.executemany("INSERT INTO sleep_diary_events VALUES(?,?,?,?,?)",
|
||||
[(item["revision_id"], event["event_id"], event["type"], event["start_at"], event["end_at"])
|
||||
for event in item["events"]])
|
||||
db.executemany("INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id,medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) VALUES(?,?,?,?,?,?,?,?,?,?)",
|
||||
[(item["revision_id"], intake["intake_id"], intake["medication_id"],
|
||||
intake["medication_name"], intake["taken_at"], intake["dose_value"],
|
||||
intake["dose_unit"], intake.get("quantity", 1), intake["note"], intake["created_at"])
|
||||
for intake in item["intakes"]])
|
||||
# An imported revision has already crossed the synchronization
|
||||
# boundary; recording both markers prevents it appearing as a draft.
|
||||
db.execute(
|
||||
|
|
@ -583,45 +527,6 @@ def apply(db: sqlite3.Connection, root: dict) -> tuple[int, int]:
|
|||
(item["entry_id"], item["revision_id"], root["generated_at"],
|
||||
item["revision_id"], root["generated_at"]),
|
||||
)
|
||||
if winning_local is not None:
|
||||
successor = "slr_" + str(uuid.uuid4())
|
||||
# INVARIANT: the successor's edit timestamp remains the winning
|
||||
# user edit, not the later technical reconciliation time. Its
|
||||
# complete payload is copied exactly; the losing branch stays
|
||||
# immutable and addressable by its original revision identity.
|
||||
db.execute(
|
||||
"INSERT INTO sleep_diary_revisions(revision_id,entry_id,parent_revision_id,"
|
||||
"created_at,sleep_quality,wake_quality,day_form,treatment_and_notes) "
|
||||
"VALUES(?,?,?,?,?,?,?,?)",
|
||||
(successor, item["entry_id"], item["revision_id"],
|
||||
winning_local["updated_at"], winning_local["sleep_quality"],
|
||||
winning_local["wake_quality"], winning_local["day_form"],
|
||||
winning_local["treatment_and_notes"]),
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO sleep_diary_events(revision_id,event_id,event_type,start_at,end_at) "
|
||||
"SELECT ?,event_id,event_type,start_at,end_at "
|
||||
"FROM sleep_diary_events WHERE revision_id=?",
|
||||
(successor, winning_local["revision_id"]),
|
||||
)
|
||||
db.execute(
|
||||
"INSERT INTO sleep_medication_intakes(revision_id,intake_id,medication_id,"
|
||||
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at) "
|
||||
"SELECT ?,intake_id,medication_id,"
|
||||
"medication_name,taken_at,dose_value,dose_unit,quantity,note,created_at "
|
||||
"FROM sleep_medication_intakes WHERE revision_id=?",
|
||||
(successor, winning_local["revision_id"]),
|
||||
)
|
||||
db.execute(
|
||||
"UPDATE sleep_diary_entries SET updated_at=?,current_revision_id=? "
|
||||
"WHERE entry_id=? AND current_revision_id=?",
|
||||
(winning_local["updated_at"], successor, item["entry_id"], item["revision_id"]),
|
||||
)
|
||||
db.execute(
|
||||
"UPDATE sleep_diary_publication_state SET validated_revision_id=?,"
|
||||
"validated_at=? WHERE entry_id=?",
|
||||
(successor, root["generated_at"], item["entry_id"]),
|
||||
)
|
||||
applied += 1
|
||||
return applied, unchanged
|
||||
|
||||
|
|
|
|||
|
|
@ -131,8 +131,7 @@ def failure_code(error: BaseException) -> str:
|
|||
return "device_unavailable"
|
||||
if "capacity exhausted" in diagnostic or "retained generation capacity" in diagnostic:
|
||||
return "capacity_exhausted"
|
||||
if ("conflict" in diagnostic or "wrong consumer" in diagnostic or
|
||||
"concurrent sleep diary revision" in diagnostic):
|
||||
if "conflict" in diagnostic or "wrong consumer" in diagnostic:
|
||||
return "data_conflict"
|
||||
return "internal_error"
|
||||
|
||||
|
|
|
|||
|
|
@ -72,15 +72,6 @@ def emit(run_id: str, phase: str, **values: object) -> None:
|
|||
print(json.dumps(value, sort_keys=True, separators=(",", ":")), flush=True)
|
||||
|
||||
|
||||
def rejected_android_generation_error(ack: dict) -> RuntimeError:
|
||||
"""Keep the durable peer rejection visible without changing its ACK."""
|
||||
# WHY: the browser cannot classify a known causal conflict when the worker
|
||||
# drops the consumer's bounded diagnostic at this process boundary.
|
||||
diagnostic = ack.get("diagnostic", "")
|
||||
detail = f": {diagnostic}" if diagnostic else ""
|
||||
return RuntimeError(f"desktop rejected Android generation{detail}")
|
||||
|
||||
|
||||
def bounded(path: Path, limit: int = 65536) -> bytes:
|
||||
with path.open("rb") as stream:
|
||||
raw = stream.read(limit + 1)
|
||||
|
|
@ -814,7 +805,7 @@ def main() -> int:
|
|||
deadline,
|
||||
)
|
||||
if ack["result"] != "consumed":
|
||||
raise rejected_android_generation_error(ack)
|
||||
raise RuntimeError("desktop rejected Android generation")
|
||||
emit(
|
||||
args.run_id,
|
||||
"local_import_committed",
|
||||
|
|
|
|||
Loading…
Reference in a new issue