fix(sync): reconcile concurrent Sleep edits by user edit time
This commit is contained in:
parent
9568bb60d3
commit
39a9df7b07
8 changed files with 487 additions and 36 deletions
|
|
@ -26,6 +26,12 @@ new features, modules, protocols, or speculative refactors.
|
|||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -676,8 +676,9 @@ class TrainlogRepository(
|
|||
}
|
||||
} catch (_: RuntimeException) { false }
|
||||
|
||||
/** CONTRACT: one transaction appends an immutable revision and advances
|
||||
* the guarded tip. Wall time never resolves a concurrent edit. */
|
||||
/** 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. */
|
||||
fun saveSleepDiary(draft: SleepDiaryDraft): SaveSleepDiaryResult =
|
||||
saveSleepDiary(draft, quickPublicationAt = null)
|
||||
|
||||
|
|
@ -1555,8 +1556,21 @@ class TrainlogRepository(
|
|||
}
|
||||
val remoteDescendsFromLocal =
|
||||
local != null && ancestry?.drop(1)?.contains(local.first) == true
|
||||
if (local != null && !remoteDescendsFromLocal && parent != local.first)
|
||||
return SleepDiaryImportResult.Rejected("concurrent sleep diary revision")
|
||||
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
|
||||
}
|
||||
}
|
||||
/* 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. */
|
||||
|
|
@ -1606,21 +1620,77 @@ 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))
|
||||
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),
|
||||
) }
|
||||
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),
|
||||
) }
|
||||
}
|
||||
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)
|
||||
|
|
@ -1629,6 +1699,40 @@ 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,6 +13,7 @@ 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
|
||||
|
|
@ -25,6 +26,147 @@ 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)
|
||||
|
|
|
|||
|
|
@ -163,7 +163,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. |
|
||||
| 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. |
|
||||
| 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,14 +32,25 @@ 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; sibling revisions and resurrection after deletion are rejected
|
||||
in the generation consumption transaction.
|
||||
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 a known concurrent revision is
|
||||
reported as `data_conflict` rather than an opaque `internal_error`. This
|
||||
classification does not retry, merge, or acknowledge the rejected generation.
|
||||
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.
|
||||
|
||||
This does not change `TRAINLOG_FORMAT_V1`, manifests, ACKs or causal
|
||||
ownership.
|
||||
|
|
|
|||
|
|
@ -1,5 +1,15 @@
|
|||
# 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
|
||||
|
|
|
|||
|
|
@ -288,6 +288,89 @@ class SleepExchangeTest(unittest.TestCase):
|
|||
"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)
|
||||
|
|
|
|||
|
|
@ -5,6 +5,7 @@ import argparse
|
|||
import json
|
||||
import re
|
||||
import sqlite3
|
||||
import uuid
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
|
||||
|
|
@ -378,6 +379,42 @@ 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"]
|
||||
|
|
@ -484,8 +521,14 @@ 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]:
|
||||
fail("concurrent sleep diary revision")
|
||||
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
|
||||
# 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
|
||||
|
|
@ -505,17 +548,30 @@ 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]))
|
||||
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"]])
|
||||
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"]],
|
||||
)
|
||||
# An imported revision has already crossed the synchronization
|
||||
# boundary; recording both markers prevents it appearing as a draft.
|
||||
db.execute(
|
||||
|
|
@ -527,6 +583,45 @@ 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
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue