From c5ed5070ebd8d0396b338117f1cfaa711889f679 Mon Sep 17 00:00:00 2001 From: pasichDev Date: Sat, 5 Sep 2026 13:26:06 +0300 Subject: [PATCH] fix(sync): give the merge a notion of an ancestor, and settle conflicts as a set MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Running the app on a phone found two defects no review had: every ordinary local edit raised a conflict on the next sync, and a task conflict never converged at all. The merge had no ancestor. It reported a conflict whenever the two hashes differed, so a note edited on this device alone was offered against the version it had just replaced — after every edit, with the wrong click discarding the user's work. sync_metadata now carries the canonical hash of the version this device last applied or published (schema 21 to 22), and the merge uses it: when the remote still equals that base only this side moved, so the local version is published without a dialog, and the same in reverse. Both sides moved, or no base recorded, keeps the old rule, so a fresh install and rows migrated from 2.6.51 behave exactly as before and fill the column in on their first sync. The task loop had a second cause. Without a base, a deletion merged by timestamp and published the version it replaced as an alternative, so the record ended up with two open conflicts; isSuperseded judged the second by timestamp alone and dropped it unsettled, and it returned on the next sync against the freshly re-timed version. One resolution per sync, forever. isSuperseded now compares content: a row is dropped only when the local record matches neither version it offers, which a resolution never causes. Rows whose winner is no longer the live version are retired when a record is applied, so the other device is not left holding a phantom. Also from the fourth review: accepting 2.6.50's repeated attachment id decoded to a shape this store could never rebuild, so those notes would have self-conflicted forever — the store and the decoder now agree on keeping the repeat; the backup pick could restore twice or be lost on rotation; the supersession mark carried the listing's time rather than its own; and a repeated attachment now travels the way 2.6.50 wrote it, so a mixed fleet needs no alias field. Verified on a Pixel 7a as two peers against one Drive: A deletes a task and publishes with no dialog, B edits it and gets exactly one honest conflict, B resolves once and two further syncs are clean, and A gets the task back with B's edit. Migration 21 to 22 ran on a live database with every row intact. 341 unit tests, 89 instrumentation tests, 0 failures; lint 0 errors. --- CHANGELOG.md | 2 +- app/build.gradle | 2 +- .../22.json | 584 ++++++++++++++++++ .../data/sync/InMemoryBundleBackend.java | 90 +++ .../com/pasich/mynotes/db/MigrationTest.java | 28 +- .../pasich/mynotes/db/RoomSyncStoreTest.java | 150 ++++- .../db/TwoPeerStoreConvergenceTest.java | 372 +++++++++++ .../mynotes/data/database/AppDatabase.java | 17 + .../data/database/dao/SyncMetadataDao.java | 6 + .../database/entities/SyncMetadataEntity.java | 9 + .../mynotes/data/sync/BundleHistory.java | 193 ++++++ .../data/sync/GoogleDriveSyncBackend.java | 161 +---- .../mynotes/data/sync/LegacyNotePayload.java | 43 +- .../mynotes/data/sync/RoomSyncStore.java | 196 +++++- .../mynotes/data/sync/SyncBundleCodec.java | 27 +- .../pasich/mynotes/data/sync/SyncMerger.java | 19 + .../pasich/mynotes/data/sync/SyncRecord.java | 32 +- .../pasich/mynotes/di/ApplicationModule.java | 3 +- .../ui/view/activity/BackupActivity.java | 108 +++- .../utils/constants/DatabaseConstants.java | 2 +- .../data/sync/GoogleDriveSyncBackendTest.java | 45 +- .../data/sync/LegacyNotePayloadTest.java | 37 +- .../data/sync/SyncBundleCodecTest.java | 19 +- .../mynotes/data/sync/SyncMergerTest.java | 54 ++ .../sync/SyncMutationCoordinatorTest.java | 6 + .../mynotes/data/sync/SyncServiceTest.java | 43 +- .../data/sync/TwoPeerConvergenceTest.java | 470 ++++++++++++++ 27 files changed, 2436 insertions(+), 282 deletions(-) create mode 100644 app/schemas/com.pasich.mynotes.data.database.AppDatabase/22.json create mode 100644 app/src/androidTest/java/com/pasich/mynotes/data/sync/InMemoryBundleBackend.java create mode 100644 app/src/androidTest/java/com/pasich/mynotes/db/TwoPeerStoreConvergenceTest.java create mode 100644 app/src/main/java/com/pasich/mynotes/data/sync/BundleHistory.java create mode 100644 app/src/test/java/com/pasich/mynotes/data/sync/TwoPeerConvergenceTest.java diff --git a/CHANGELOG.md b/CHANGELOG.md index 3afefd17..fe4b6ce9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,6 +1,6 @@ # CHANGELOG -## [2.6.51] - 05.09.2026 +## [2.6.52] - 05.09.2026 **New** diff --git a/app/build.gradle b/app/build.gradle index 4ccc7e06..feea5444 100644 --- a/app/build.gradle +++ b/app/build.gradle @@ -21,7 +21,7 @@ apply from: "$projectDir/gradle/libs-task.gradle" apply from: "$projectDir/gradle/changelog-task.gradle" -def appVersionCode = 51 +def appVersionCode = 52 def appVersionName = "2.6.${appVersionCode}" def gitCommitHashProvider = providers.exec { diff --git a/app/schemas/com.pasich.mynotes.data.database.AppDatabase/22.json b/app/schemas/com.pasich.mynotes.data.database.AppDatabase/22.json new file mode 100644 index 00000000..36cee7f4 --- /dev/null +++ b/app/schemas/com.pasich.mynotes.data.database.AppDatabase/22.json @@ -0,0 +1,584 @@ +{ + "formatVersion": 1, + "database": { + "version": 22, + "identityHash": "b937db5e7ea37a3712fc161529ab8277", + "entities": [ + { + "tableName": "tags", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `name` TEXT NOT NULL, `visibility` INTEGER NOT NULL, `systemAction` INTEGER NOT NULL, `position` INTEGER NOT NULL)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "nameTag", + "columnName": "name", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "visibility", + "columnName": "visibility", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "systemAction", + "columnName": "systemAction", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "position", + "columnName": "position", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "notes", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `title` TEXT, `value` TEXT, `date` INTEGER NOT NULL, `tag` TEXT, `valueJson` TEXT, `hasRichContent` INTEGER NOT NULL, `attachments` TEXT, `isTrash` INTEGER NOT NULL, `reminderTime` INTEGER, `isPinned` INTEGER NOT NULL, `reminderRepeat` TEXT NOT NULL, `reminderIntervalMinutes` INTEGER NOT NULL)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "title", + "columnName": "title", + "affinity": "TEXT" + }, + { + "fieldPath": "value", + "columnName": "value", + "affinity": "TEXT" + }, + { + "fieldPath": "date", + "columnName": "date", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "tag", + "columnName": "tag", + "affinity": "TEXT" + }, + { + "fieldPath": "valueJson", + "columnName": "valueJson", + "affinity": "TEXT" + }, + { + "fieldPath": "hasRichContent", + "columnName": "hasRichContent", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "attachments", + "columnName": "attachments", + "affinity": "TEXT" + }, + { + "fieldPath": "isTrash", + "columnName": "isTrash", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "reminderTime", + "columnName": "reminderTime", + "affinity": "INTEGER" + }, + { + "fieldPath": "isPinned", + "columnName": "isPinned", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "reminderRepeat", + "columnName": "reminderRepeat", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "reminderIntervalMinutes", + "columnName": "reminderIntervalMinutes", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "tasks", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `title` TEXT NOT NULL, `description` TEXT, `isDone` INTEGER NOT NULL DEFAULT 0, `categoryId` INTEGER NOT NULL DEFAULT 0, `createdAt` INTEGER NOT NULL DEFAULT 0, `position` INTEGER NOT NULL DEFAULT 0, `reminderTime` INTEGER, `reminderIntervalMinutes` INTEGER NOT NULL DEFAULT 0)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "title", + "columnName": "title", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "description", + "columnName": "description", + "affinity": "TEXT" + }, + { + "fieldPath": "isDone", + "columnName": "isDone", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "categoryId", + "columnName": "categoryId", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "createdAt", + "columnName": "createdAt", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "position", + "columnName": "position", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "reminderTime", + "columnName": "reminderTime", + "affinity": "INTEGER" + }, + { + "fieldPath": "reminderIntervalMinutes", + "columnName": "reminderIntervalMinutes", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "task_categories", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `name` TEXT NOT NULL, `colorHex` TEXT NOT NULL DEFAULT '#6750A4', `position` INTEGER NOT NULL DEFAULT 0)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "name", + "columnName": "name", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "colorHex", + "columnName": "colorHex", + "affinity": "TEXT", + "notNull": true, + "defaultValue": "'#6750A4'" + }, + { + "fieldPath": "position", + "columnName": "position", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "sync_metadata", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`recordType` TEXT NOT NULL, `localId` INTEGER NOT NULL, `stableId` TEXT NOT NULL, `updatedAt` INTEGER NOT NULL, `deletedAt` INTEGER, `syncedVersionId` TEXT, PRIMARY KEY(`recordType`, `localId`))", + "fields": [ + { + "fieldPath": "recordType", + "columnName": "recordType", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "localId", + "columnName": "localId", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "stableId", + "columnName": "stableId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "updatedAt", + "columnName": "updatedAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "deletedAt", + "columnName": "deletedAt", + "affinity": "INTEGER" + }, + { + "fieldPath": "syncedVersionId", + "columnName": "syncedVersionId", + "affinity": "TEXT" + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "recordType", + "localId" + ] + }, + "indices": [ + { + "name": "index_sync_metadata_recordType_stableId", + "unique": true, + "columnNames": [ + "recordType", + "stableId" + ], + "orders": [], + "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_sync_metadata_recordType_stableId` ON `${TABLE_NAME}` (`recordType`, `stableId`)" + }, + { + "name": "index_sync_metadata_updatedAt", + "unique": false, + "columnNames": [ + "updatedAt" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_sync_metadata_updatedAt` ON `${TABLE_NAME}` (`updatedAt`)" + }, + { + "name": "index_sync_metadata_deletedAt", + "unique": false, + "columnNames": [ + "deletedAt" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_sync_metadata_deletedAt` ON `${TABLE_NAME}` (`deletedAt`)" + } + ] + }, + { + "tableName": "sync_pending_preferences", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER NOT NULL, `payloadJson` TEXT NOT NULL, `targetHash` TEXT NOT NULL, `baselineHash` TEXT NOT NULL, `recordUpdatedAt` INTEGER NOT NULL, `quarantined` INTEGER NOT NULL, `conflictId` INTEGER NOT NULL, `conflictResolution` TEXT NOT NULL, PRIMARY KEY(`id`))", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "payloadJson", + "columnName": "payloadJson", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "targetHash", + "columnName": "targetHash", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "baselineHash", + "columnName": "baselineHash", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "recordUpdatedAt", + "columnName": "recordUpdatedAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "quarantined", + "columnName": "quarantined", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "conflictId", + "columnName": "conflictId", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "conflictResolution", + "columnName": "conflictResolution", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "sync_conflicts", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `recordType` TEXT NOT NULL, `stableId` TEXT NOT NULL, `versionPairHash` TEXT NOT NULL, `winnerSource` TEXT NOT NULL, `loserSource` TEXT NOT NULL, `winnerVersionId` TEXT NOT NULL, `loserVersionId` TEXT NOT NULL, `winnerJson` TEXT NOT NULL, `loserJson` TEXT NOT NULL, `winnerUpdatedAt` INTEGER NOT NULL, `loserUpdatedAt` INTEGER NOT NULL, `winnerTombstone` INTEGER NOT NULL, `loserTombstone` INTEGER NOT NULL, `resolution` TEXT NOT NULL, `resolved` INTEGER NOT NULL, `createdAt` INTEGER NOT NULL, `resolvedAt` INTEGER NOT NULL)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "recordType", + "columnName": "recordType", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "stableId", + "columnName": "stableId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "versionPairHash", + "columnName": "versionPairHash", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "winnerSource", + "columnName": "winnerSource", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "loserSource", + "columnName": "loserSource", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "winnerVersionId", + "columnName": "winnerVersionId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "loserVersionId", + "columnName": "loserVersionId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "winnerJson", + "columnName": "winnerJson", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "loserJson", + "columnName": "loserJson", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "winnerUpdatedAt", + "columnName": "winnerUpdatedAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "loserUpdatedAt", + "columnName": "loserUpdatedAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "winnerTombstone", + "columnName": "winnerTombstone", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "loserTombstone", + "columnName": "loserTombstone", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "resolution", + "columnName": "resolution", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "resolved", + "columnName": "resolved", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "createdAt", + "columnName": "createdAt", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "resolvedAt", + "columnName": "resolvedAt", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + }, + "indices": [ + { + "name": "index_sync_conflicts_recordType_stableId_versionPairHash", + "unique": true, + "columnNames": [ + "recordType", + "stableId", + "versionPairHash" + ], + "orders": [], + "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_sync_conflicts_recordType_stableId_versionPairHash` ON `${TABLE_NAME}` (`recordType`, `stableId`, `versionPairHash`)" + }, + { + "name": "index_sync_conflicts_resolved", + "unique": false, + "columnNames": [ + "resolved" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_sync_conflicts_resolved` ON `${TABLE_NAME}` (`resolved`)" + }, + { + "name": "index_sync_conflicts_createdAt", + "unique": false, + "columnNames": [ + "createdAt" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_sync_conflicts_createdAt` ON `${TABLE_NAME}` (`createdAt`)" + } + ] + }, + { + "tableName": "sync_state", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER NOT NULL, `status` TEXT NOT NULL, `backendIdentifier` TEXT, `lastSuccessfulSyncAt` INTEGER, `attemptStartedAt` INTEGER, `errorMessage` TEXT, `conflictCount` INTEGER NOT NULL, PRIMARY KEY(`id`))", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "status", + "columnName": "status", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "backendIdentifier", + "columnName": "backendIdentifier", + "affinity": "TEXT" + }, + { + "fieldPath": "lastSuccessfulSyncAt", + "columnName": "lastSuccessfulSyncAt", + "affinity": "INTEGER" + }, + { + "fieldPath": "attemptStartedAt", + "columnName": "attemptStartedAt", + "affinity": "INTEGER" + }, + { + "fieldPath": "errorMessage", + "columnName": "errorMessage", + "affinity": "TEXT" + }, + { + "fieldPath": "conflictCount", + "columnName": "conflictCount", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "id" + ] + } + } + ], + "setupQueries": [ + "CREATE TABLE IF NOT EXISTS room_master_table (id INTEGER PRIMARY KEY,identity_hash TEXT)", + "INSERT OR REPLACE INTO room_master_table (id,identity_hash) VALUES(42, 'b937db5e7ea37a3712fc161529ab8277')" + ] + } +} \ No newline at end of file diff --git a/app/src/androidTest/java/com/pasich/mynotes/data/sync/InMemoryBundleBackend.java b/app/src/androidTest/java/com/pasich/mynotes/data/sync/InMemoryBundleBackend.java new file mode 100644 index 00000000..55fffc8a --- /dev/null +++ b/app/src/androidTest/java/com/pasich/mynotes/data/sync/InMemoryBundleBackend.java @@ -0,0 +1,90 @@ +package com.pasich.mynotes.data.sync; + +import androidx.annotation.NonNull; +import androidx.annotation.Nullable; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.time.Clock; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.UUID; + +/** + * Drive without the network: the same codec and the same history fold as the Drive backend, over + * bundles kept in memory, so two real stores can be synced against each other on the device. + */ +public final class InMemoryBundleBackend implements SyncBackend { + private final Clock clock; + private final SyncBundleCodec codec = new SyncBundleCodec(); + private final Map bundles = new LinkedHashMap<>(); + private final Map attachments = new LinkedHashMap<>(); + private String lastReadToken = ""; + + public InMemoryBundleBackend(@NonNull Clock clock) { + this.clock = clock; + } + + @NonNull + @Override + public String getIdentifier() { + return "memory"; + } + + @NonNull + @Override + public RemoteSnapshot readSnapshotResult() throws IOException { + Map decoded = new LinkedHashMap<>(); + for (byte[] bytes : bundles.values()) { + SyncBundleCodec.DecodedBundle bundle = codec.decode(new ByteArrayInputStream(bytes)); + decoded.put(bundle.getBundleId(), bundle); + } + BundleHistory.Fold fold = BundleHistory.fold(decoded, new SyncMerger()); + lastReadToken = UUID.randomUUID().toString(); + return new RemoteSnapshot( + fold.merged, + fold.conflicts, + fold.frontier, + fold.alternatives, + fold.resolvedAlternativeIds, + lastReadToken); + } + + @Override + public void publish(@NonNull SyncPublication publication) throws IOException { + if (!publication.getReadContext().getReadToken().equals(lastReadToken)) { + throw new IOException("stale read context"); + } + byte[] bytes = + codec.encode( + publication.getSnapshot(), + clock.instant(), + publication.getReadContext().getFrontierBundleIds(), + publication.getUnresolvedAlternatives(), + publication.getResolvedAlternativeIds()); + bundles.put(codec.decode(new ByteArrayInputStream(bytes)).getBundleId(), bytes); + } + + @Override + public boolean hasAttachment(@NonNull String sha256) { + return attachments.containsKey(sha256); + } + + @Nullable + @Override + public InputStream readAttachment(@NonNull String sha256) { + byte[] bytes = attachments.get(sha256); + return bytes == null ? null : new ByteArrayInputStream(bytes); + } + + @Override + public void writeAttachment( + @NonNull String sha256, long sizeBytes, @NonNull InputStream content) + throws IOException { + java.io.ByteArrayOutputStream out = new java.io.ByteArrayOutputStream(); + byte[] buffer = new byte[8192]; + int read; + while ((read = content.read(buffer)) != -1) out.write(buffer, 0, read); + attachments.put(sha256, out.toByteArray()); + } +} diff --git a/app/src/androidTest/java/com/pasich/mynotes/db/MigrationTest.java b/app/src/androidTest/java/com/pasich/mynotes/db/MigrationTest.java index 4cecfcb2..342b7f34 100644 --- a/app/src/androidTest/java/com/pasich/mynotes/db/MigrationTest.java +++ b/app/src/androidTest/java/com/pasich/mynotes/db/MigrationTest.java @@ -199,9 +199,30 @@ public void migrate20to21_addsTheConflictBookkeepingToTheJournal() throws IOExce } } + @Test + public void migrate21to22_addsTheSyncedVersionAndLeavesItUnknownForExistingRows() + throws IOException { + SupportSQLiteDatabase db = helper.createDatabase(TEST_DB, 21); + db.execSQL( + "INSERT INTO sync_metadata (recordType, localId, stableId, updatedAt, deletedAt) " + + "VALUES ('note', 1, 'stable', 1000, NULL)"); + db.close(); + + SupportSQLiteDatabase migrated = + helper.runMigrationsAndValidate(TEST_DB, 22, true, AppDatabase.MIGRATION_21_22); + try (android.database.Cursor cursor = + migrated.query("SELECT syncedVersionId FROM sync_metadata WHERE localId = 1")) { + assertThat(cursor.moveToFirst()).isTrue(); + // Unknown until the next sync fills it in; the merge then behaves as before for it. + assertThat(cursor.isNull(0)).isTrue(); + } finally { + migrated.close(); + } + } + @Test public void migrateFromTheLastReleasedVersion_reachesTheCurrentSchema() throws IOException { - // 17 is what 2.6.48 shipped; 18, 19 and 20 all land in the same release after it. + // 17 is what 2.6.48 shipped; everything after it lands in later releases of this line. SupportSQLiteDatabase db = helper.createDatabase(TEST_DB, 17); db.execSQL( "INSERT INTO notes " @@ -213,12 +234,13 @@ public void migrateFromTheLastReleasedVersion_reachesTheCurrentSchema() throws I SupportSQLiteDatabase migrated = helper.runMigrationsAndValidate( TEST_DB, - 21, + 22, true, AppDatabase.MIGRATION_17_18, AppDatabase.MIGRATION_18_19, AppDatabase.MIGRATION_19_20, - AppDatabase.MIGRATION_20_21); + AppDatabase.MIGRATION_20_21, + AppDatabase.MIGRATION_21_22); try (android.database.Cursor cursor = migrated.query("SELECT COUNT(*) FROM notes")) { assertThat(cursor.moveToFirst()).isTrue(); assertThat(cursor.getInt(0)).isEqualTo(1); diff --git a/app/src/androidTest/java/com/pasich/mynotes/db/RoomSyncStoreTest.java b/app/src/androidTest/java/com/pasich/mynotes/db/RoomSyncStoreTest.java index 31f2ffc2..89cac525 100644 --- a/app/src/androidTest/java/com/pasich/mynotes/db/RoomSyncStoreTest.java +++ b/app/src/androidTest/java/com/pasich/mynotes/db/RoomSyncStoreTest.java @@ -418,6 +418,122 @@ public void aNoteReceivedFromAnotherDeviceHashesIdenticallyWhenRebuiltThere() th } } + // ------------------------------------------------- an ordinary edit after a clean sync + + @Test + public void aLocalEditAfterACleanSyncIsPublishedWithoutAConflict() throws Exception { + // Reproduced on a Pixel: fully synced, edit a note, sync — the conflict dialog offered + // the edit against the text it had replaced, after every single edit. + int noteId = seedNote("Original", "body", null); + SyncRecord published = onlyNote(store.readSnapshot()); + // What a clean sync does with the merged snapshot: apply it, which records the version + // the remote now holds. + store.applySnapshot( + new SyncSnapshot(Collections.singletonList(published)), Collections.emptyList()); + Note note = db.noteDao().getNoteSync(noteId); + note.setTitle("Edited here"); + db.noteDao().addNote(note); + db.syncMetadataDao().touch(SyncMetadata.RECORD_TYPE_NOTE, noteId, 5_000L); + + SyncRecord edited = onlyNote(store.readSnapshot()); + com.pasich.mynotes.data.sync.SyncMergeResult merge = + new com.pasich.mynotes.data.sync.SyncMerger() + .merge( + new SyncSnapshot(Collections.singletonList(edited)), + new SyncSnapshot(Collections.singletonList(published))); + + assertThat(edited.getBaseVersionId()).isEqualTo(published.getCanonicalPayloadHash()); + assertThat(merge.getConflicts()).isEmpty(); + assertThat( + merge.getMergedSnapshot() + .getRecords() + .get(0) + .getPayload() + .get("b") + .getAsString()) + .isEqualTo("Edited here"); + } + + @Test + public void anEditOnBothSidesAfterACleanSyncIsStillAConflict() throws Exception { + int noteId = seedNote("Original", "body", null); + SyncRecord published = onlyNote(store.readSnapshot()); + store.applySnapshot( + new SyncSnapshot(Collections.singletonList(published)), Collections.emptyList()); + Note note = db.noteDao().getNoteSync(noteId); + note.setTitle("Edited here"); + db.noteDao().addNote(note); + db.syncMetadataDao().touch(SyncMetadata.RECORD_TYPE_NOTE, noteId, 5_000L); + JsonObject elsewhere = published.getPayload(); + elsewhere.addProperty("b", "Edited elsewhere"); + SyncRecord remote = + SyncRecord.live( + SyncRecord.Type.NOTE, + published.getId(), + java.time.Instant.ofEpochMilli(6_000L), + elsewhere); + + com.pasich.mynotes.data.sync.SyncMergeResult merge = + new com.pasich.mynotes.data.sync.SyncMerger() + .merge( + new SyncSnapshot( + Collections.singletonList(onlyNote(store.readSnapshot()))), + new SyncSnapshot(Collections.singletonList(remote))); + + assertThat(merge.getConflicts()).hasSize(1); + } + + @Test + public void aNoteWithADuplicatedBlockReceivedUnder2650RebuildsToWhatTheDecoderProduces() + throws Exception { + // 2.6.50 published such a note with one attachment id referenced twice, and its receivers + // restored one column entry per reference under that id. The decoder keeps the id; the + // store used to re-key the repeat, so the two versions hashed differently at the same + // timestamp and the note conflicted with itself on every sync. + byte[] bytes = "photo bytes".getBytes(StandardCharsets.UTF_8); + String hash = sha256(bytes); + String restoredId = "7d444840-9dc0-11d1-b245-5ffdce74fad2"; + int noteId = seedNote("Shopping", "Milk", null); + File folder = new File(context.getFilesDir(), "attachments/note_" + noteId); + assertThat(folder.mkdirs() || folder.isDirectory()).isTrue(); + try (FileOutputStream out = + new FileOutputStream(new File(folder, restoredId + "-" + hash))) { + out.write(bytes); + } + String receiverUrl = + com.pasich.mynotes.extendedEditor.attach.AttachmentStorage.urlFor( + noteId, restoredId + "-" + hash); + String entry = + "{\"url\":\"" + + receiverUrl + + "\",\"name\":\"photo.png\",\"id\":\"" + + restoredId + + "\"}"; + String block = + "{\"type\":\"attaches\",\"data\":{\"file\":{\"url\":\"" + + receiverUrl + + "\",\"name\":\"photo.png\"}}}"; + Note received = db.noteDao().getNoteSync(noteId); + received.setAttachments("[" + entry + "," + entry + "]"); + received.setValueJson("[" + block + "," + block + "]"); + db.noteDao().addNote(received); + // The same note as 2.6.50 put it on the wire: the sender's own URLs, [X, X]. + String senderUrl = "editorjs://attachments/note_77/photo.png"; + String senderBlock = + "{\"type\":\"attaches\",\"data\":{\"file\":{\"url\":\"" + + senderUrl + + "\",\"name\":\"photo.png\"}}}"; + Note sender = new Note().create("Shopping", "Milk", 1_000L, ""); + sender.setValueJson("[" + senderBlock + "," + senderBlock + "]"); + SyncRecord onDrive = + published2650WithIds( + sender, "photo.png", hash, bytes.length, restoredId, restoredId); + + SyncRecord rebuilt = onlyNote(store.readSnapshot()); + + assertThat(rebuilt.getCanonicalPayloadHash()).isEqualTo(onDrive.getCanonicalPayloadHash()); + } + // ------------------------------------------------- notes published by 2.6.50 @Test @@ -535,24 +651,34 @@ private static SyncRecord published2650( (stableId + "\n0\n" + blockUrl + "\n" + displayName) .getBytes(StandardCharsets.UTF_8)) .toString(); + return published2650WithIds(note, displayName, hash, size, legacyId); + } + + /** As above, with the manifest ids given: one entry per id, repeats included. */ + private static SyncRecord published2650WithIds( + Note note, String displayName, String hash, long size, String... ids) + throws IOException { + String stableId = "11111111-1111-4111-8111-111111111111"; JsonObject payload = new com.google.gson.Gson().toJsonTree(note).getAsJsonObject(); payload.remove("a"); payload.remove("h"); - JsonObject entry = new JsonObject(); - entry.addProperty("id", legacyId); - entry.addProperty("sha256", hash); - entry.addProperty("mimeType", "image/png"); - entry.addProperty("size", size); - entry.addProperty("path", "attachments/" + hash); - entry.addProperty("displayName", displayName); com.google.gson.JsonArray manifest = new com.google.gson.JsonArray(); - manifest.add(entry); - payload.add("attachmentsManifest", manifest); com.google.gson.JsonArray hashes = new com.google.gson.JsonArray(); - hashes.add(hash); - payload.add("attachmentHashes", hashes); JsonObject names = new JsonObject(); - names.addProperty(legacyId, displayName); + for (String id : ids) { + JsonObject entry = new JsonObject(); + entry.addProperty("id", id); + entry.addProperty("sha256", hash); + entry.addProperty("mimeType", "image/png"); + entry.addProperty("size", size); + entry.addProperty("path", "attachments/" + hash); + entry.addProperty("displayName", displayName); + manifest.add(entry); + hashes.add(hash); + names.addProperty(id, displayName); + } + payload.add("attachmentsManifest", manifest); + payload.add("attachmentHashes", hashes); payload.add("attachmentNames", names); SyncRecord asPublished = SyncRecord.live( diff --git a/app/src/androidTest/java/com/pasich/mynotes/db/TwoPeerStoreConvergenceTest.java b/app/src/androidTest/java/com/pasich/mynotes/db/TwoPeerStoreConvergenceTest.java new file mode 100644 index 00000000..80009103 --- /dev/null +++ b/app/src/androidTest/java/com/pasich/mynotes/db/TwoPeerStoreConvergenceTest.java @@ -0,0 +1,372 @@ +package com.pasich.mynotes.db; + +import static com.google.common.truth.Truth.assertThat; +import static org.mockito.Mockito.mock; + +import android.content.Context; +import androidx.room.Room; +import androidx.test.ext.junit.runners.AndroidJUnit4; +import androidx.test.platform.app.InstrumentationRegistry; +import com.pasich.mynotes.data.database.AppDatabase; +import com.pasich.mynotes.data.database.entities.SyncConflictEntity; +import com.pasich.mynotes.data.database.entities.SyncMetadataEntity; +import com.pasich.mynotes.data.model.Note; +import com.pasich.mynotes.data.model.Tag; +import com.pasich.mynotes.data.model.Task; +import com.pasich.mynotes.data.model.TaskCategory; +import com.pasich.mynotes.data.preferences.PreferenceHelper; +import com.pasich.mynotes.data.sync.InMemoryBundleBackend; +import com.pasich.mynotes.data.sync.RoomSyncStore; +import com.pasich.mynotes.data.sync.SyncMetadata; +import com.pasich.mynotes.data.sync.SyncResolution; +import com.pasich.mynotes.data.sync.SyncService; +import com.pasich.mynotes.data.sync.SyncState; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneId; +import java.time.ZoneOffset; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.junit.runner.RunWith; + +/** + * Two real stores, one shared history, a conflict settled through the store's own resolution. + * + *

Reproduced on a Pixel with tasks: peer A deletes, peer B edits, B keeps the live version, and + * the same conflict came back on every sync. Run for every record type: the protocol is shared, but + * each type has its own DAO path through the store. + */ +@RunWith(AndroidJUnit4.class) +public class TwoPeerStoreConvergenceTest { + + private static final String STABLE_ID = "11111111-1111-4111-8111-111111111111"; + + private Context context; + private final MutableClock clock = new MutableClock(Instant.parse("2026-09-05T12:00:00Z")); + private InMemoryBundleBackend drive; + private Peer a; + private Peer b; + + @Before + public void setUp() { + context = InstrumentationRegistry.getInstrumentation().getTargetContext(); + drive = new InMemoryBundleBackend(clock); + a = new Peer(); + b = new Peer(); + } + + @After + public void tearDown() { + a.db.close(); + b.db.close(); + } + + @Test + public void aTaskDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() + throws Exception { + settlesAfterOneResolution(new TaskRecord(), false); + } + + @Test + public void aTaskSyncedBeforeBasesWereRecordedStillSettlesAfterOneResolution() + throws Exception { + settlesAfterOneResolution(new TaskRecord(), true); + } + + @Test + public void aNoteDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() + throws Exception { + settlesAfterOneResolution(new NoteRecord(), false); + } + + @Test + public void aTagDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() + throws Exception { + settlesAfterOneResolution(new TagRecord(), false); + } + + @Test + public void aCategoryDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() + throws Exception { + settlesAfterOneResolution(new CategoryRecord(), false); + } + + private void settlesAfterOneResolution(RecordKind kind, boolean migratedWithoutBases) + throws Exception { + kind.create(a, "TaskX"); + clock.advance(); + assertClean(a.sync()); + clock.advance(); + assertClean(b.sync()); + clock.advance(); + assertClean(a.sync()); + assertThat(kind.title(b)).isEqualTo("TaskX"); + if (migratedWithoutBases) { + // Rows a build that recorded no synced version left behind. + a.forgetBases(); + b.forgetBases(); + } + + // A deletes and syncs; B edits and syncs. + kind.delete(a); + clock.advance(); + a.sync(); + clock.advance(); + kind.edit(b, "TaskX-B"); + clock.advance(); + assertThat(b.sync().getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(b.store.getUnresolvedConflicts()).isNotEmpty(); + + // B keeps the live version — every open conflict, as the dialog offers them one by one. + clock.advance(); + for (SyncConflictEntity conflict : b.store.getUnresolvedConflicts()) { + b.store.resolveConflict( + conflict.id, + conflict.winnerTombstone + ? SyncResolution.KEEP_ALTERNATIVE + : SyncResolution.KEEP_WINNER); + } + assertThat(b.store.getUnresolvedConflicts()).isEmpty(); + clock.advance(); + SyncState publishing = b.sync(); + assertThat(publishing.getErrorMessage()).isNull(); + assertClean(publishing); + assertThat(b.store.getUnresolvedConflicts()).isEmpty(); + clock.advance(); + assertClean(b.sync()); + assertThat(b.store.getUnresolvedConflicts()).isEmpty(); + + // A gets the record back — the tombstone revival — with the edit, and stays clean. + clock.advance(); + SyncState onA = a.sync(); + assertThat(onA.getErrorMessage()).isNull(); + assertClean(onA); + assertThat(kind.title(a)).isEqualTo("TaskX-B"); + assertThat(kind.title(b)).isEqualTo("TaskX-B"); + // Nothing left to ask either user, including the deletion A reported against the version + // it replaced when its synced version was still unknown. + assertThat(a.store.getUnresolvedConflicts()).isEmpty(); + clock.advance(); + assertClean(a.sync()); + clock.advance(); + assertClean(b.sync()); + assertThat(a.store.getUnresolvedConflicts()).isEmpty(); + assertThat(b.store.getUnresolvedConflicts()).isEmpty(); + } + + private static void assertClean(SyncState state) { + assertThat(state.getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(state.getConflictCount()).isEqualTo(0); + } + + // ------------------------------------------------------------------ peers and records + + private final class Peer { + final AppDatabase db; + final RoomSyncStore store; + long localId; + + Peer() { + db = + Room.inMemoryDatabaseBuilder(context, AppDatabase.class) + .allowMainThreadQueries() + .build(); + store = new RoomSyncStore(context, db, mock(PreferenceHelper.class)); + } + + SyncState sync() { + return new SyncService(store, new com.pasich.mynotes.data.sync.SyncMerger(), clock) + .sync(drive); + } + + void forgetBases() { + db.getOpenHelper() + .getWritableDatabase() + .execSQL("UPDATE sync_metadata SET syncedVersionId = NULL"); + } + + void metadata(String recordType, long id) { + localId = id; + db.syncMetadataDao() + .insertIfAbsent( + new SyncMetadataEntity( + recordType, id, STABLE_ID, clock.millis(), null)); + } + + void touch(String recordType) { + db.syncMetadataDao().touch(recordType, localIdOf(recordType), clock.millis()); + } + + void markDeleted(String recordType) { + db.syncMetadataDao().markDeleted(recordType, localIdOf(recordType), clock.millis()); + } + + long localIdOf(String recordType) { + SyncMetadataEntity metadata = db.syncMetadataDao().getByStableId(recordType, STABLE_ID); + return metadata == null ? -1L : metadata.localId; + } + } + + /** How one record type is created, edited, deleted and read through the peer's own DAOs. */ + private interface RecordKind { + void create(Peer peer, String title); + + void edit(Peer peer, String title); + + void delete(Peer peer); + + String title(Peer peer); + } + + private static final class TaskRecord implements RecordKind { + @Override + public void create(Peer peer, String title) { + Task task = new Task(title, 0); + task.setCreatedAt(1_000L); + peer.metadata(SyncMetadata.RECORD_TYPE_TASK, peer.db.taskDao().insertTask(task)); + } + + @Override + public void edit(Peer peer, String title) { + Task task = peer.db.taskDao().getTaskSync((int) peer.localIdOf("task")); + task.setTitle(title); + peer.db.taskDao().updateTask(task); + peer.touch(SyncMetadata.RECORD_TYPE_TASK); + } + + @Override + public void delete(Peer peer) { + peer.db.taskDao().deleteById((int) peer.localIdOf("task")); + peer.markDeleted(SyncMetadata.RECORD_TYPE_TASK); + } + + @Override + public String title(Peer peer) { + long id = peer.localIdOf("task"); + Task task = id < 0 ? null : peer.db.taskDao().getTaskSync((int) id); + return task == null ? null : task.getTitle(); + } + } + + private static final class NoteRecord implements RecordKind { + @Override + public void create(Peer peer, String title) { + Note note = new Note().create(title, "body", 1_000L, ""); + peer.metadata(SyncMetadata.RECORD_TYPE_NOTE, peer.db.noteDao().addNote(note)); + } + + @Override + public void edit(Peer peer, String title) { + Note note = peer.db.noteDao().getNoteSync((int) peer.localIdOf("note")); + note.setTitle(title); + peer.db.noteDao().addNote(note); + peer.touch(SyncMetadata.RECORD_TYPE_NOTE); + } + + @Override + public void delete(Peer peer) { + peer.db.noteDao().deleteById((int) peer.localIdOf("note")); + peer.markDeleted(SyncMetadata.RECORD_TYPE_NOTE); + } + + @Override + public String title(Peer peer) { + long id = peer.localIdOf("note"); + Note note = id < 0 ? null : peer.db.noteDao().getNoteSync((int) id); + return note == null ? null : note.getTitle(); + } + } + + private static final class TagRecord implements RecordKind { + @Override + public void create(Peer peer, String title) { + peer.metadata( + SyncMetadata.RECORD_TYPE_TAG, + peer.db.tagsDao().addTag(new Tag().create(title))); + } + + @Override + public void edit(Peer peer, String title) { + Tag tag = peer.db.tagsDao().getTagSync(peer.localIdOf("tag")); + tag.setNameTag(title); + peer.db.tagsDao().updateTag(tag); + peer.touch(SyncMetadata.RECORD_TYPE_TAG); + } + + @Override + public void delete(Peer peer) { + peer.db.tagsDao().deleteById(peer.localIdOf("tag")); + peer.markDeleted(SyncMetadata.RECORD_TYPE_TAG); + } + + @Override + public String title(Peer peer) { + long id = peer.localIdOf("tag"); + Tag tag = id < 0 ? null : peer.db.tagsDao().getTagSync(id); + return tag == null ? null : tag.getNameTag(); + } + } + + private static final class CategoryRecord implements RecordKind { + @Override + public void create(Peer peer, String title) { + TaskCategory category = new TaskCategory(); + category.setName(title); + peer.metadata( + SyncMetadata.RECORD_TYPE_CATEGORY, + peer.db.taskCategoryDao().insertCategory(category)); + } + + @Override + public void edit(Peer peer, String title) { + TaskCategory category = + peer.db.taskCategoryDao().getCategorySync((int) peer.localIdOf("category")); + category.setName(title); + peer.db.taskCategoryDao().updateCategory(category); + peer.touch(SyncMetadata.RECORD_TYPE_CATEGORY); + } + + @Override + public void delete(Peer peer) { + peer.db.taskCategoryDao().deleteById((int) peer.localIdOf("category")); + peer.markDeleted(SyncMetadata.RECORD_TYPE_CATEGORY); + } + + @Override + public String title(Peer peer) { + long id = peer.localIdOf("category"); + TaskCategory category = + id < 0 ? null : peer.db.taskCategoryDao().getCategorySync((int) id); + return category == null ? null : category.getName(); + } + } + + private static final class MutableClock extends Clock { + private Instant now; + + MutableClock(Instant start) { + now = start; + } + + void advance() { + now = now.plus(Duration.ofMinutes(1)); + } + + @Override + public ZoneId getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + } +} diff --git a/app/src/main/java/com/pasich/mynotes/data/database/AppDatabase.java b/app/src/main/java/com/pasich/mynotes/data/database/AppDatabase.java index 2453f749..3da62cf1 100644 --- a/app/src/main/java/com/pasich/mynotes/data/database/AppDatabase.java +++ b/app/src/main/java/com/pasich/mynotes/data/database/AppDatabase.java @@ -237,6 +237,23 @@ public void migrate(@NonNull SupportSQLiteDatabase database) { } }; + /** + * Remembers, per record, the version the remote is known to hold. + * + *

The merge had no notion of a common ancestor, so a record edited on this device alone was + * offered to the user as a conflict against the version they had just replaced — after every + * edit. Null for existing rows: the next sync fills it in, and until then the merge behaves as + * before. + */ + public static final Migration MIGRATION_21_22 = + new Migration(21, 22) { + @Override + public void migrate(@NonNull SupportSQLiteDatabase database) { + database.execSQL( + "ALTER TABLE `sync_metadata` ADD COLUMN `syncedVersionId` TEXT"); + } + }; + private static void insertMetadataForExistingRecords( SupportSQLiteDatabase database, String recordType, diff --git a/app/src/main/java/com/pasich/mynotes/data/database/dao/SyncMetadataDao.java b/app/src/main/java/com/pasich/mynotes/data/database/dao/SyncMetadataDao.java index 80ab8197..81fd0d4f 100644 --- a/app/src/main/java/com/pasich/mynotes/data/database/dao/SyncMetadataDao.java +++ b/app/src/main/java/com/pasich/mynotes/data/database/dao/SyncMetadataDao.java @@ -18,6 +18,12 @@ public interface SyncMetadataDao { "UPDATE sync_metadata SET updatedAt = :updatedAt, deletedAt = :deletedAt WHERE recordType = :recordType AND localId = :localId") void setVersion(String recordType, long localId, long updatedAt, Long deletedAt); + /** Records the version the remote is known to hold; see {@code syncedVersionId}. */ + @Query( + "UPDATE sync_metadata SET syncedVersionId = :versionId " + + "WHERE recordType = :recordType AND localId = :localId") + void setSyncedVersion(String recordType, long localId, String versionId); + @Query( "SELECT * FROM sync_metadata " + "WHERE recordType = :recordType AND localId = :localId LIMIT 1") diff --git a/app/src/main/java/com/pasich/mynotes/data/database/entities/SyncMetadataEntity.java b/app/src/main/java/com/pasich/mynotes/data/database/entities/SyncMetadataEntity.java index a0e0a798..23d95f22 100644 --- a/app/src/main/java/com/pasich/mynotes/data/database/entities/SyncMetadataEntity.java +++ b/app/src/main/java/com/pasich/mynotes/data/database/entities/SyncMetadataEntity.java @@ -33,6 +33,15 @@ public class SyncMetadataEntity { public Long deletedAt; + /** + * Canonical hash of the version this device last published or applied for the record — the + * version the remote is known to hold. A merge that finds the remote still equal to it knows + * that only this device has moved and publishes without asking; without it every ordinary local + * edit was offered to the user as a conflict against the text they had just replaced. Null + * until the record has been through a sync on this build. + */ + public String syncedVersionId; + public SyncMetadataEntity( @NonNull String recordType, long localId, diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/BundleHistory.java b/app/src/main/java/com/pasich/mynotes/data/sync/BundleHistory.java new file mode 100644 index 00000000..0f2808a5 --- /dev/null +++ b/app/src/main/java/com/pasich/mynotes/data/sync/BundleHistory.java @@ -0,0 +1,193 @@ +package com.pasich.mynotes.data.sync; + +import androidx.annotation.NonNull; +import java.io.IOException; +import java.util.ArrayDeque; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.Comparator; +import java.util.Deque; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Set; + +/** + * Turns every bundle an account holds into the one remote state a sync merges against. + * + *

Bundles are immutable and name their parents, so the remote state is the fold of the frontier + * heads plus the unresolved alternatives and settled versions they carry. Kept apart from the Drive + * transport so the same fold serves an in-memory backend in tests: the convergence of two peers is + * a property of this fold and the store together, and has to be provable without a network. + */ +final class BundleHistory { + + /** Everything a read of the history yields. */ + static final class Fold { + final List frontier; + final SyncSnapshot merged; + final List conflicts; + final List alternatives; + final Set resolvedAlternativeIds; + + private Fold( + List frontier, + SyncSnapshot merged, + List conflicts, + List alternatives, + Set resolvedAlternativeIds) { + this.frontier = Collections.unmodifiableList(frontier); + this.merged = merged; + this.conflicts = Collections.unmodifiableList(conflicts); + this.alternatives = Collections.unmodifiableList(alternatives); + this.resolvedAlternativeIds = Collections.unmodifiableSet(resolvedAlternativeIds); + } + } + + private BundleHistory() {} + + @NonNull + static Fold fold( + @NonNull Map bundlesByLogicalId, + @NonNull SyncMerger merger) + throws IOException { + validateBundleDag(bundlesByLogicalId); + List frontier = computeFrontier(bundlesByLogicalId); + SyncSnapshot merged = SyncSnapshot.empty(); + List conflicts = new ArrayList<>(); + for (String bundleId : frontier) { + // Both sides are remote bundle heads. Naming them explicitly stops the accumulator + // being reported to the user as "this device". + SyncMergeResult result = + merger.merge( + merged, + bundlesByLogicalId.get(bundleId).getSnapshot(), + SyncMergeResult.Source.REMOTE, + SyncMergeResult.Source.REMOTE); + merged = result.getMergedSnapshot(); + conflicts.addAll(result.getConflicts()); + } + // Alternatives and the resolutions that retire them travel with the bundles, so a device + // that has never seen a conflict still discovers it and a device that resolved one still + // retires it everywhere. + Set resolvedAlternativeIds = new HashSet<>(); + for (String bundleId : frontier) { + resolvedAlternativeIds.addAll( + bundlesByLogicalId.get(bundleId).getResolvedAlternativeIds()); + } + Map alternativesByVersion = new java.util.LinkedHashMap<>(); + for (String bundleId : frontier) { + for (SyncRecord alternative : bundlesByLogicalId.get(bundleId).getAlternatives()) { + String versionId = alternative.getCanonicalPayloadHash(); + if (resolvedAlternativeIds.contains(versionId)) continue; + SyncRecord winner = merged.find(alternative.getType(), alternative.getId()); + if (winner == null || winner.getCanonicalPayloadHash().equals(versionId)) { + // Nothing to choose between: the alternative is the current value, or its + // record no longer exists at all. + continue; + } + alternativesByVersion.putIfAbsent(versionId, alternative); + } + } + List alternatives = new ArrayList<>(alternativesByVersion.values()); + for (SyncRecord alternative : alternatives) { + SyncRecord winner = merged.find(alternative.getType(), alternative.getId()); + conflicts.add( + new SyncMergeResult.Conflict( + winner, + alternative, + SyncMergeResult.Source.REMOTE, + SyncMergeResult.Source.REMOTE)); + } + return new Fold(frontier, merged, conflicts, alternatives, resolvedAlternativeIds); + } + + /** + * Checks the ancestry graph without requiring every historical bundle to still exist. + * + *

A missing ancestor used to be fatal, which inverted the rule that cleanup must never be + * needed for correctness: one bundle trashed by hand, or aged out of Drive's own trash, and + * sync failed forever with no way back. It is safe to tolerate because a bundle is a complete + * snapshot rather than a delta — every descendant already contains everything its ancestors + * held, including their unresolved alternatives — so an absent ancestor removes nothing from + * the state a head describes. It also cannot be a frontier head itself, since a head is a + * bundle no present bundle claims as a parent. + */ + private static void validateBundleDag( + @NonNull Map bundles) throws IOException { + Map> parentsById = new HashMap<>(); + for (Map.Entry entry : bundles.entrySet()) { + parentsById.put(entry.getKey(), entry.getValue().getParentBundleIds()); + } + validateAncestry(parentsById); + } + + /** + * Rejects a cycle in the parent graph. + * + *

Iterative on purpose: the recursive walk went one frame deeper per ancestor, so a long + * linear history — exactly what an account that syncs after every edit accumulates — could + * overflow the worker's stack, and a {@code StackOverflowError} is not an {@code IOException} + * the sync knows how to report. + */ + static void validateAncestry(@NonNull Map> parentsById) + throws IOException { + Set visited = new HashSet<>(); + Set visiting = new HashSet<>(); + Deque stack = new ArrayDeque<>(); + for (String root : parentsById.keySet()) { + if (visited.contains(root)) { + continue; + } + visiting.add(root); + stack.push(new Frame(root, parentsById.get(root).iterator())); + while (!stack.isEmpty()) { + Frame frame = stack.peek(); + if (!frame.parents.hasNext()) { + stack.pop(); + visiting.remove(frame.bundleId); + visited.add(frame.bundleId); + continue; + } + String parent = frame.parents.next(); + Collection grandparents = parentsById.get(parent); + if (grandparents == null || visited.contains(parent)) { + // An ancestor that is no longer stored, or one already walked. + continue; + } + if (!visiting.add(parent)) { + throw new IOException("Drive bundle ancestry contains a cycle"); + } + stack.push(new Frame(parent, grandparents.iterator())); + } + } + } + + private static final class Frame { + private final String bundleId; + private final Iterator parents; + + private Frame(String bundleId, Iterator parents) { + this.bundleId = bundleId; + this.parents = parents; + } + } + + @NonNull + private static List computeFrontier( + @NonNull Map bundles) { + Set ancestors = new HashSet<>(); + for (SyncBundleCodec.DecodedBundle bundle : bundles.values()) { + ancestors.addAll(bundle.getParentBundleIds()); + } + List frontier = new ArrayList<>(); + for (String bundleId : bundles.keySet()) { + if (!ancestors.contains(bundleId)) frontier.add(bundleId); + } + frontier.sort(Comparator.naturalOrder()); + return frontier; + } +} diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackend.java b/app/src/main/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackend.java index 777a8bfa..ab9c2005 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackend.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackend.java @@ -17,15 +17,11 @@ import java.net.URLEncoder; import java.nio.charset.StandardCharsets; import java.time.Clock; -import java.util.ArrayDeque; import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.Comparator; -import java.util.Deque; import java.util.HashMap; -import java.util.HashSet; -import java.util.Iterator; import java.util.List; import java.util.Map; import java.util.Set; @@ -212,54 +208,12 @@ public synchronized RemoteSnapshot readSnapshotResult() throws IOException { bundlesByLogicalId.put(decoded.getBundleId(), decoded); } } - validateBundleDag(bundlesByLogicalId); - List frontier = computeFrontier(bundlesByLogicalId); - SyncSnapshot merged = SyncSnapshot.empty(); - List conflicts = new ArrayList<>(); - for (String bundleId : frontier) { - // Both sides are Drive bundle heads. Naming them explicitly stops the accumulator - // being reported to the user as "this device". - SyncMergeResult result = - merger.merge( - merged, - bundlesByLogicalId.get(bundleId).getSnapshot(), - SyncMergeResult.Source.REMOTE, - SyncMergeResult.Source.REMOTE); - merged = result.getMergedSnapshot(); - conflicts.addAll(result.getConflicts()); - } - // Alternatives and the resolutions that retire them travel with the bundles, so a device - // that has never seen a conflict still discovers it and a device that resolved one still - // retires it everywhere. - Set resolvedAlternativeIds = new HashSet<>(); - for (String bundleId : frontier) { - resolvedAlternativeIds.addAll( - bundlesByLogicalId.get(bundleId).getResolvedAlternativeIds()); - } - Map alternativesByVersion = new java.util.LinkedHashMap<>(); - for (String bundleId : frontier) { - for (SyncRecord alternative : bundlesByLogicalId.get(bundleId).getAlternatives()) { - String versionId = alternative.getCanonicalPayloadHash(); - if (resolvedAlternativeIds.contains(versionId)) continue; - SyncRecord winner = merged.find(alternative.getType(), alternative.getId()); - if (winner == null || winner.getCanonicalPayloadHash().equals(versionId)) { - // Nothing to choose between: the alternative is the current value, or its - // record no longer exists at all. - continue; - } - alternativesByVersion.putIfAbsent(versionId, alternative); - } - } - List alternatives = new ArrayList<>(alternativesByVersion.values()); - for (SyncRecord alternative : alternatives) { - SyncRecord winner = merged.find(alternative.getType(), alternative.getId()); - conflicts.add( - new SyncMergeResult.Conflict( - winner, - alternative, - SyncMergeResult.Source.REMOTE, - SyncMergeResult.Source.REMOTE)); - } + BundleHistory.Fold fold = BundleHistory.fold(bundlesByLogicalId, merger); + List frontier = fold.frontier; + SyncSnapshot merged = fold.merged; + List conflicts = fold.conflicts; + List alternatives = fold.alternatives; + Set resolvedAlternativeIds = fold.resolvedAlternativeIds; markSupersededBundles(bundleFiles, frontier); lastReadBundles = Collections.unmodifiableList(bundleFiles); @@ -300,6 +254,7 @@ private void markSupersededBundles( output.write(jsonBytes(patch)); } ensureSuccess(connection); + noteDriveTime(connection); } finally { connection.disconnect(); } @@ -614,92 +569,6 @@ private Map attachmentSizes(@NonNull Collection notes) return sizes; } - /** - * Checks the ancestry graph without requiring every historical bundle to still exist. - * - *

A missing ancestor used to be fatal, which inverted the rule that cleanup must never be - * needed for correctness: one bundle trashed by hand, or aged out of Drive's own trash, and - * sync failed forever with no way back. It is safe to tolerate because a bundle is a complete - * snapshot rather than a delta — every descendant already contains everything its ancestors - * held, including their unresolved alternatives — so an absent ancestor removes nothing from - * the state a head describes. It also cannot be a frontier head itself, since a head is a - * bundle no present bundle claims as a parent. - */ - private static void validateBundleDag( - @NonNull Map bundles) throws IOException { - Map> parentsById = new HashMap<>(); - for (Map.Entry entry : bundles.entrySet()) { - parentsById.put(entry.getKey(), entry.getValue().getParentBundleIds()); - } - validateAncestry(parentsById); - } - - /** - * Rejects a cycle in the parent graph. - * - *

Iterative on purpose: the recursive walk went one frame deeper per ancestor, so a long - * linear history — exactly what an account that syncs after every edit accumulates — could - * overflow the worker's stack, and a {@code StackOverflowError} is not an {@code IOException} - * the sync knows how to report. - */ - static void validateAncestry(@NonNull Map> parentsById) - throws IOException { - Set visited = new HashSet<>(); - Set visiting = new HashSet<>(); - Deque stack = new ArrayDeque<>(); - for (String root : parentsById.keySet()) { - if (visited.contains(root)) { - continue; - } - visiting.add(root); - stack.push(new Frame(root, parentsById.get(root).iterator())); - while (!stack.isEmpty()) { - Frame frame = stack.peek(); - if (!frame.parents.hasNext()) { - stack.pop(); - visiting.remove(frame.bundleId); - visited.add(frame.bundleId); - continue; - } - String parent = frame.parents.next(); - Collection grandparents = parentsById.get(parent); - if (grandparents == null || visited.contains(parent)) { - // An ancestor that is no longer stored, or one already walked. - continue; - } - if (!visiting.add(parent)) { - throw new IOException("Drive bundle ancestry contains a cycle"); - } - stack.push(new Frame(parent, grandparents.iterator())); - } - } - } - - private static final class Frame { - private final String bundleId; - private final Iterator parents; - - private Frame(String bundleId, Iterator parents) { - this.bundleId = bundleId; - this.parents = parents; - } - } - - @NonNull - private static List computeFrontier( - @NonNull Map bundles) { - Set ancestors = new HashSet<>(); - for (SyncBundleCodec.DecodedBundle bundle : bundles.values()) { - ancestors.addAll(bundle.getParentBundleIds()); - } - List frontier = new ArrayList<>(); - for (String bundleId : bundles.keySet()) { - if (!ancestors.contains(bundleId)) frontier.add(bundleId); - } - frontier.sort(Comparator.naturalOrder()); - return frontier; - } - /** One physical bundle file in a root; the logical id is known once it has been decoded. */ private static final class BundleFile { private final String fileId; @@ -1452,13 +1321,22 @@ private JsonObject requestJsonIdempotent( return requestExecutor.executeIdempotent(() -> requestJson(method, url, contentType, body)); } - @NonNull - private JsonObject readJsonResponse(@NonNull HttpURLConnection connection) throws IOException { - ensureSuccess(connection); + /** + * Remembers Drive's clock from a response. Every response is asked, downloads included: the + * supersession mark written after the last download used to carry the time of the listing + * before it, so the grace ran short by the whole download phase. + */ + private void noteDriveTime(@NonNull HttpURLConnection connection) { long serverTime = connection.getHeaderFieldDate("Date", 0L); if (serverTime > 0L) { lastDriveTimeMillis = serverTime; } + } + + @NonNull + private JsonObject readJsonResponse(@NonNull HttpURLConnection connection) throws IOException { + ensureSuccess(connection); + noteDriveTime(connection); try (InputStream input = connection.getInputStream()) { return GSON.fromJson( new String( @@ -1483,6 +1361,7 @@ private byte[] requestBytesOnce(@NonNull String method, @NonNull String url, int HttpURLConnection connection = open(method, url); try { ensureSuccess(connection); + noteDriveTime(connection); try (InputStream input = connection.getInputStream()) { return readBounded(input, maxBytes, "Drive response exceeds the sync size limit"); } diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/LegacyNotePayload.java b/app/src/main/java/com/pasich/mynotes/data/sync/LegacyNotePayload.java index 6facbaeb..347d14c1 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/LegacyNotePayload.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/LegacyNotePayload.java @@ -34,6 +34,13 @@ private LegacyNotePayload() {} /** * Rewrites {@code payload} in place when it is a 2.6.50-shaped note. * + *

Two kinds of id occur in such a note. One 2.6.50 derived itself, recognisable as exactly + * the old derivation of the block URL, name and position; it becomes the current derivation, + * which is what the upgraded store computes for the same column. One it restored from another + * device is a canonical id the store keeps as it is — repeated or not. Either way the blocks + * are put into wire form by position, which is how the store maps them when the column and the + * blocks line up. + * * @return true when the payload was upgraded. */ static boolean upgrade(@NonNull String noteStableId, @NonNull JsonObject payload) { @@ -50,7 +57,7 @@ static boolean upgrade(@NonNull String noteStableId, @NonNull JsonObject payload if (urls.size() != manifest.size()) { return false; } - List newIds = new ArrayList<>(manifest.size()); + List ids = new ArrayList<>(manifest.size()); for (int index = 0; index < manifest.size(); index++) { JsonElement element = manifest.get(index); if (!element.isJsonObject()) { @@ -58,34 +65,34 @@ static boolean upgrade(@NonNull String noteStableId, @NonNull JsonObject payload } JsonObject entry = element.getAsJsonObject(); String url = urls.get(index); + // A block already in wire form, or naming something that is not a local file, is + // not 2.6.50's shape. if (AttachmentUrl.parse(url) == null || !isString(entry.get("id")) || !isString(entry.get("sha256")) || !isString(entry.get("displayName"))) { return false; } + String id = entry.get("id").getAsString(); String displayName = entry.get("displayName").getAsString(); - if (!entry.get("id") - .getAsString() - .equals( - AttachmentLogicalIds.deriveLegacy( - noteStableId, index, url, displayName))) { - return false; - } - newIds.add( - AttachmentLogicalIds.derive( - noteStableId, - index, - url, - displayName, - entry.get("sha256").getAsString())); + ids.add( + id.equals( + AttachmentLogicalIds.deriveLegacy( + noteStableId, index, url, displayName)) + ? AttachmentLogicalIds.derive( + noteStableId, + index, + url, + displayName, + entry.get("sha256").getAsString()) + : id); } JsonObject names = new JsonObject(); for (int index = 0; index < manifest.size(); index++) { JsonObject entry = manifest.get(index).getAsJsonObject(); - entry.addProperty("id", newIds.get(index)); - names.addProperty(newIds.get(index), entry.get("displayName").getAsString()); + entry.addProperty("id", ids.get(index)); + names.addProperty(ids.get(index), entry.get("displayName").getAsString()); } payload.add("attachmentNames", names); int[] position = {0}; @@ -93,7 +100,7 @@ static boolean upgrade(@NonNull String noteStableId, @NonNull JsonObject payload "f", EditorAttachmentBlocks.rewriteUrls( blocks.getAsString(), - url -> AttachmentWireUrl.forLogicalId(newIds.get(position[0]++)))); + url -> AttachmentWireUrl.forLogicalId(ids.get(position[0]++)))); return true; } diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/RoomSyncStore.java b/app/src/main/java/com/pasich/mynotes/data/sync/RoomSyncStore.java index c5a95150..a460420f 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/RoomSyncStore.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/RoomSyncStore.java @@ -74,6 +74,15 @@ public final class RoomSyncStore implements SyncStore { /** Whether every note folder has been indexed into {@link #localAttachments}. */ private volatile boolean noteFoldersIndexed; + /** + * Version hash of every record the last build produced, by record key. + * + *

When the apply skips a record because it was edited during the sync, the version that was + * nonetheless published may be this device's own; remembering it as the synced version keeps + * the next merge from calling the follow-up edit a conflict. + */ + private final Map builtVersionIds = new ConcurrentHashMap<>(); + /** * Attachments the last build described from the column alone because their file is gone, keyed * by hash, so the service can name the note when no endpoint holds the blob either. @@ -187,15 +196,19 @@ public SnapshotBuildResult buildSnapshot() throws IOException { List records = new ArrayList<>(); List problems = new ArrayList<>(); rememberedOnlyAttachments.clear(); + builtVersionIds.clear(); try { for (SyncMetadataEntity metadata : database.syncMetadataDao().getAll()) { + // Every record carries the version the remote is known to hold, so the merge can + // tell an edit made here alone from one made on both sides. if (metadata.deletedAt != null) { records.add( SyncRecord.tombstone( - SyncRecord.Type.fromWireValue(metadata.recordType), - metadata.stableId, - Instant.ofEpochMilli(metadata.updatedAt), - Instant.ofEpochMilli(metadata.deletedAt))); + SyncRecord.Type.fromWireValue(metadata.recordType), + metadata.stableId, + Instant.ofEpochMilli(metadata.updatedAt), + Instant.ofEpochMilli(metadata.deletedAt)) + .withBaseVersion(metadata.syncedVersionId)); continue; } JsonObject payload = payload(metadata, problems); @@ -204,15 +217,19 @@ public SnapshotBuildResult buildSnapshot() throws IOException { database.syncMetadataDao().get(metadata.recordType, metadata.localId); records.add( SyncRecord.live( - SyncRecord.Type.fromWireValue(metadata.recordType), - metadata.stableId, - Instant.ofEpochMilli( - current == null - ? metadata.updatedAt - : current.updatedAt), - payload)); + SyncRecord.Type.fromWireValue(metadata.recordType), + metadata.stableId, + Instant.ofEpochMilli( + current == null + ? metadata.updatedAt + : current.updatedAt), + payload) + .withBaseVersion(metadata.syncedVersionId)); } } + for (SyncRecord record : records) { + builtVersionIds.put(recordKey(record), record.getCanonicalPayloadHash()); + } } finally { hashCache.flush(); } @@ -285,6 +302,11 @@ private void applySnapshotInternal( record.getUpdatedAt() .toEpochMilli(), null)); + database.syncMetadataDao() + .setSyncedVersion( + record.getType().getWireValue(), + localId, + record.getCanonicalPayloadHash()); } transactionFailureInjector.afterRecordApplied(record); continue; @@ -316,6 +338,16 @@ private void applySnapshotInternal( + metadata.recordType + "; it was edited during this sync"); skippedKeys.add(key); + if (record.getCanonicalPayloadHash() + .equals(builtVersionIds.get(key))) { + // The version just published is this device's own; the + // edit made meanwhile is a step past it, not a conflict. + database.syncMetadataDao() + .setSyncedVersion( + metadata.recordType, + metadata.localId, + record.getCanonicalPayloadHash()); + } continue; } if (record.isTombstone()) { @@ -326,6 +358,12 @@ private void applySnapshotInternal( metadata.localId, record.getUpdatedAt().toEpochMilli(), record.getDeletedAt().toEpochMilli()); + database.syncMetadataDao() + .setSyncedVersion( + metadata.recordType, + metadata.localId, + record.getCanonicalPayloadHash()); + retireConflictsSupersededBy(record); transactionFailureInjector.afterRecordApplied(record); continue; } @@ -339,6 +377,12 @@ private void applySnapshotInternal( metadata.localId, record.getUpdatedAt().toEpochMilli(), null); + database.syncMetadataDao() + .setSyncedVersion( + metadata.recordType, + metadata.localId, + record.getCanonicalPayloadHash()); + retireConflictsSupersededBy(record); transactionFailureInjector.afterRecordApplied(record); } if (preferencesMetadata[0] != null) { @@ -370,6 +414,14 @@ private void applySnapshotInternal( preferencesMetadata[0].localId, stagedPreferencesUpdatedAt, null); + if (preferencesRecord != null) { + database.syncMetadataDao() + .setSyncedVersion( + preferencesMetadata[0].recordType, + preferencesMetadata[0].localId, + preferencesRecord + .getCanonicalPayloadHash()); + } database.syncPendingPreferencesDao() .upsert( new SyncPendingPreferencesEntity( @@ -991,9 +1043,9 @@ public void resolveConflict(long conflictId, @NonNull SyncResolution resolution) if (pending == null || pending.resolved) return; if (isSuperseded(pending)) { - // The record moved on after this conflict was recorded — an edit here, or a newer - // version applied from another device — so both stored versions are older than what - // the user now has. Applying either would overwrite the newer edit with a version + // The record's content changed after this conflict was recorded — an edit here, or a + // newer version applied from another device — so both stored versions are older than + // what the user now has. Applying either would overwrite the newer edit with a version // that was never offered against it. The alternative still travels with the bundle // and comes back as a fresh conflict against the current version at the next sync. Log.w(TAG, "Dropping a conflict that the record's newer version has superseded"); @@ -1034,12 +1086,65 @@ public void resolveConflict(long conflictId, @NonNull SyncResolution resolution) } } - /** True when the local record is newer than both versions the conflict offers. */ - private boolean isSuperseded(@NonNull SyncConflictEntity conflict) { + /** + * True when the local record has moved on from both versions the conflict offers. + * + *

Newer by timestamp is not enough. Resolving one conflict re-times the record without + * changing it, and a record with two open conflicts — a live edit against a deletion and + * against an older version, say — was then judged to have moved past its second conflict, which + * was dropped unsettled instead of offered. Its alternative came back at the next sync as a + * fresh conflict, was dropped again after the next resolution, and the account never settled. + * Only content the user actually changed since the conflict was recorded counts. + */ + private boolean isSuperseded(@NonNull SyncConflictEntity conflict) throws IOException { SyncMetadataEntity metadata = database.syncMetadataDao().getByStableId(conflict.recordType, conflict.stableId); - return metadata != null - && metadata.updatedAt > Math.max(conflict.winnerUpdatedAt, conflict.loserUpdatedAt); + if (metadata == null + || metadata.updatedAt + <= Math.max(conflict.winnerUpdatedAt, conflict.loserUpdatedAt)) { + return false; + } + String current = contentDigest(metadata); + return !current.equals(contentDigest(conflict.recordType, conflict.winnerJson)) + && !current.equals(contentDigest(conflict.recordType, conflict.loserJson)); + } + + /** A digest of what the local record says, independent of when it last changed. */ + @NonNull + private String contentDigest(@NonNull SyncMetadataEntity metadata) { + if (metadata.deletedAt != null) { + return "tombstone"; + } + JsonObject payload; + if (SyncMetadata.RECORD_TYPE_PREFERENCES.equals(metadata.recordType)) { + // Not through payload(): that path records a local edit as a side effect. + payload = gson.toJsonTree(preferenceHelper.getListPreferences()).getAsJsonObject(); + } else { + payload = payload(metadata, new ArrayList<>()); + } + return payload == null ? "absent" : contentDigest(metadata.recordType, payload); + } + + @NonNull + private static String contentDigest(@NonNull String recordType, @NonNull String recordJson) { + JsonObject root = JsonParser.parseString(recordJson).getAsJsonObject(); + JsonElement deletedAt = root.get("deletedAt"); + if (deletedAt != null && !deletedAt.isJsonNull()) { + return "tombstone"; + } + JsonObject payload = root.getAsJsonObject("payload"); + return contentDigest(recordType, payload == null ? new JsonObject() : payload); + } + + /** The version hash at a fixed timestamp, so only the content takes part. */ + @NonNull + private static String contentDigest(@NonNull String recordType, @NonNull JsonObject payload) { + return SyncRecord.live( + SyncRecord.Type.fromWireValue(recordType), + PREFERENCES_STABLE_ID, + Instant.EPOCH, + payload) + .getCanonicalPayloadHash(); } /** @@ -1145,6 +1250,22 @@ private void pinResolvedConflictAttachments(@NonNull SyncRecord selected) throws } } + /** + * Drops open conflicts for a record whose winner is not the version just applied. + * + *

Such a row offers, pre-selected, a version that is no longer the live one; kept, it showed + * as a phantom conflict until the user tapped it, at which point it was dropped anyway. Its + * alternative is not lost: if it is still unsettled it travels with the bundle and comes back + * against the current version at the next sync. + */ + private void retireConflictsSupersededBy(@NonNull SyncRecord applied) { + database.syncConflictDao() + .deleteSupersededUnresolved( + applied.getType().getWireValue(), + applied.getId(), + applied.getCanonicalPayloadHash()); + } + /** * Stores the conflicts this apply produced and retires the ones it makes meaningless. * @@ -1511,7 +1632,7 @@ private boolean addAttachmentMetadata( JsonArray hashes = new JsonArray(); JsonObject names = new JsonObject(); Map logicalIdByUrl = new HashMap<>(); - Set logicalIdsInNote = new HashSet<>(); + List logicalIdsInOrder = new ArrayList<>(); boolean complete = true; for (int attachmentIndex = 0; attachmentIndex < attachments.size(); attachmentIndex++) { JsonElement element = attachments.get(attachmentIndex); @@ -1601,9 +1722,10 @@ private boolean addAttachmentMetadata( } String displayName = displayNameFor(attachment, file, hash); String logicalId = attachment.id; - // A column entry may repeat an id — a duplicated block whose file was later replaced. - // The manifest is keyed by id, so the repeat is given its own, as if it had none. - if (!isCanonicalUuid(logicalId) || logicalIdsInNote.contains(logicalId)) { + // A repeated canonical id — a duplicated block — is kept as it is: 2.6.50 published + // such notes that way, and re-keying the repeat here made the decoded and the + // rebuilt version disagree forever. + if (!isCanonicalUuid(logicalId)) { // Existing editor data predates logical attachment IDs. Deriving from the stable // note, source URL, position and content keeps the migration deterministic while // allowing equal-content references to remain distinct logical attachments. The @@ -1623,7 +1745,7 @@ private boolean addAttachmentMetadata( displayName, hash); } - logicalIdsInNote.add(logicalId); + logicalIdsInOrder.add(logicalId); hashes.add(hash); names.addProperty(logicalId, displayName); logicalIdByUrl.put(comparableUrl(attachment.url), logicalId); @@ -1653,15 +1775,25 @@ private boolean addAttachmentMetadata( JsonElement valueJson = payload.get("f"); if (valueJson != null && valueJson.isJsonPrimitive()) { String local = valueJson.getAsString(); - String wire = - EditorAttachmentBlocks.rewriteUrls( - local, - url -> { - String logicalId = logicalIdByUrl.get(comparableUrl(url)); - return logicalId == null - ? null - : AttachmentWireUrl.forLogicalId(logicalId); - }); + // The column is the blocks' file list in document order, so when the two line up + // the mapping is positional — the only mapping a decoder of another device's bundle + // can reproduce, and the only one that survives two blocks naming one file. The + // URL match is the fallback for a column that no longer lines up with its blocks. + EditorAttachmentBlocks.UrlMapper mapper; + if (EditorAttachmentBlocks.fileUrls(local).size() == logicalIdsInOrder.size()) { + int[] position = {0}; + mapper = + url -> AttachmentWireUrl.forLogicalId(logicalIdsInOrder.get(position[0]++)); + } else { + mapper = + url -> { + String logicalId = logicalIdByUrl.get(comparableUrl(url)); + return logicalId == null + ? null + : AttachmentWireUrl.forLogicalId(logicalId); + }; + } + String wire = EditorAttachmentBlocks.rewriteUrls(local, mapper); if (wire != null && !wire.equals(local)) { payload.addProperty("f", wire); } diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/SyncBundleCodec.java b/app/src/main/java/com/pasich/mynotes/data/sync/SyncBundleCodec.java index 3920eef4..75b091b0 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/SyncBundleCodec.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/SyncBundleCodec.java @@ -360,29 +360,23 @@ private static AttachmentPlan planAttachments( for (JsonElement element : manifestEntries) { AttachmentManifestEntry attachment = AttachmentManifestEntry.fromJson(element.getAsJsonObject()); - List recordWireIds = plan.wireIdsFor(record); String wireId = attachment.id; AttachmentManifestEntry sameId = plan.byWireId.get(wireId); if (sameId != null && !sameId.sameRemoteFile(attachment)) { - wireId = aliasFor(attachment, 0); + wireId = aliasFor(attachment); sameId = plan.byWireId.get(wireId); if (sameId != null && !sameId.sameRemoteFile(attachment)) { throw new IOException( "Two notes reference conflicting attachment metadata"); } } - // One record may repeat an entry — 2.6.50 published a duplicated block that way, - // and its receivers hold such manifests as conflict alternatives. Each repeat gets - // a wire id of its own, so a note never references one entry twice on the wire - // and every repeat maps back to the record's own id on the way in. - for (int occurrence = 1; recordWireIds.contains(wireId); occurrence++) { - wireId = aliasFor(attachment, occurrence); - sameId = plan.byWireId.get(wireId); - } if (sameId == null) { plan.byWireId.put(wireId, attachment.withId(wireId)); } - recordWireIds.add(wireId); + // A record repeating one entry — a duplicated block — references it twice under + // the one id, exactly as 2.6.50 published it: the shape every released client + // reads, with no alias for the older ones to stumble over. + plan.wireIdsFor(record).add(wireId); AttachmentManifestEntry previous = seenByHash.putIfAbsent(attachment.sha256, attachment); if (previous != null && !previous.sameRemoteFile(attachment)) { @@ -398,12 +392,11 @@ private static AttachmentPlan planAttachments( /** Deterministic, so two devices publishing the same collision write the same bundle. */ @NonNull - private static String aliasFor(@NonNull AttachmentManifestEntry attachment, int occurrence) { - String source = "attachment-alias\n" + attachment.id + "\n" + attachment.sha256; - if (occurrence > 0) { - source += "\n" + occurrence; - } - return UUID.nameUUIDFromBytes(source.getBytes(StandardCharsets.UTF_8)).toString(); + private static String aliasFor(@NonNull AttachmentManifestEntry attachment) { + return UUID.nameUUIDFromBytes( + ("attachment-alias\n" + attachment.id + "\n" + attachment.sha256) + .getBytes(StandardCharsets.UTF_8)) + .toString(); } /** The manifest entries by wire id, and each record's wire ids in manifest order. */ diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/SyncMerger.java b/app/src/main/java/com/pasich/mynotes/data/sync/SyncMerger.java index d639693c..9bd139cc 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/SyncMerger.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/SyncMerger.java @@ -12,6 +12,12 @@ *

A missing version never means deletion. For a record present at both endpoints, newest {@code * updatedAt} wins. Equal but different versions use the lexicographically smaller canonical SHA-256 * as a stable tiebreaker, so every device reaches the same result regardless of merge order. + * + *

A conflict is reported only when both sides moved. The local record may carry the version its + * device last synchronized; while the remote still equals it, only the local side has changed and + * the local version is published without asking — and the other way round. Without that, every + * ordinary local edit was offered to the user as a conflict against the text they had just + * replaced, after every single edit, with the wrong tap discarding their work. */ public final class SyncMerger { @@ -78,6 +84,19 @@ private void mergeVersions( SyncSnapshot.RecordKey key, SyncMergeResult.Source localSource, SyncMergeResult.Source remoteSource) { + String base = local.getBaseVersionId(); + if (base != null) { + if (base.equals(remote.getCanonicalPayloadHash())) { + // The remote is exactly what this device last synchronized: nobody else moved. + merged.put(key, local); + return; + } + if (base.equals(local.getCanonicalPayloadHash())) { + // This device has not moved since it last synchronized; the remote has. + merged.put(key, remote); + return; + } + } int timestampComparison = local.getUpdatedAt().compareTo(remote.getUpdatedAt()); if (timestampComparison > 0) { merged.put(key, local); diff --git a/app/src/main/java/com/pasich/mynotes/data/sync/SyncRecord.java b/app/src/main/java/com/pasich/mynotes/data/sync/SyncRecord.java index c0ed704e..cf9ba499 100644 --- a/app/src/main/java/com/pasich/mynotes/data/sync/SyncRecord.java +++ b/app/src/main/java/com/pasich/mynotes/data/sync/SyncRecord.java @@ -63,12 +63,22 @@ public static Type fromWireValue(@NonNull String wireValue) { @Nullable private final Instant deletedAt; private final JsonObject payload; + /** + * Version id of the remote version this record's device last synchronized, or null. + * + *

Not part of the payload or the hash: it is what the local store knows about the remote, + * carried alongside the record so the merge can tell "only this side moved" from "both moved". + */ + @Nullable private final String baseVersionId; + private SyncRecord( @NonNull Type type, @NonNull String id, @NonNull Instant updatedAt, @Nullable Instant deletedAt, - @NonNull JsonObject payload) { + @NonNull JsonObject payload, + @Nullable String baseVersionId) { + this.baseVersionId = baseVersionId; this.type = Objects.requireNonNull(type, "type"); validateCanonicalUuid(id); this.id = id; @@ -120,7 +130,7 @@ public static SyncRecord live( @NonNull String id, @NonNull Instant updatedAt, @NonNull JsonObject payload) { - return new SyncRecord(type, id, updatedAt, null, payload); + return new SyncRecord(type, id, updatedAt, null, payload, null); } @NonNull @@ -129,7 +139,23 @@ public static SyncRecord tombstone( @NonNull String id, @NonNull Instant updatedAt, @NonNull Instant deletedAt) { - return new SyncRecord(type, id, updatedAt, deletedAt, new JsonObject()); + return new SyncRecord(type, id, updatedAt, deletedAt, new JsonObject(), null); + } + + /** + * The same version, annotated with the remote version its device last synchronized. + * + * @param baseVersionId canonical hash of that version, or null when nothing was ever synced. + */ + @NonNull + public SyncRecord withBaseVersion(@Nullable String baseVersionId) { + return new SyncRecord(type, id, updatedAt, deletedAt, payload, baseVersionId); + } + + /** The remote version this record's device last synchronized, or null when unknown. */ + @Nullable + public String getBaseVersionId() { + return baseVersionId; } @NonNull diff --git a/app/src/main/java/com/pasich/mynotes/di/ApplicationModule.java b/app/src/main/java/com/pasich/mynotes/di/ApplicationModule.java index 16a4902a..e72cd797 100644 --- a/app/src/main/java/com/pasich/mynotes/di/ApplicationModule.java +++ b/app/src/main/java/com/pasich/mynotes/di/ApplicationModule.java @@ -77,7 +77,8 @@ AppDatabase providesAppDatabase(@ApplicationContext Context context) { AppDatabase.MIGRATION_17_18, AppDatabase.MIGRATION_18_19, AppDatabase.MIGRATION_19_20, - AppDatabase.MIGRATION_20_21) + AppDatabase.MIGRATION_20_21, + AppDatabase.MIGRATION_21_22) .build(); } diff --git a/app/src/main/java/com/pasich/mynotes/ui/view/activity/BackupActivity.java b/app/src/main/java/com/pasich/mynotes/ui/view/activity/BackupActivity.java index 225879d6..e0fd6171 100644 --- a/app/src/main/java/com/pasich/mynotes/ui/view/activity/BackupActivity.java +++ b/app/src/main/java/com/pasich/mynotes/ui/view/activity/BackupActivity.java @@ -156,6 +156,16 @@ public class BackupActivity extends BaseActivity } }); private boolean restoreSuccess = false; + + /** + * The backup document being judged, kept across a rotation. + * + *

The verdict is delivered to the instance that asked; when that instance has been destroyed + * meanwhile the pick used to be dropped without a word. The recreated instance asks again. + */ + @Nullable private Uri pendingRestorePick; + + private static final String STATE_PENDING_RESTORE_PICK = "pendingRestorePick"; private OtherAppImportDialog importDialog; private Dialog progressDialog; private RoomSyncStore roomSyncStore; @@ -208,6 +218,13 @@ public void onCreate(Bundle savedInstanceState) { setupTabs(); + if (savedInstanceState != null) { + Uri pick = savedInstanceState.getParcelable(STATE_PENDING_RESTORE_PICK); + if (pick != null) { + validatePickedBackup(pick); + } + } + getOnBackPressedDispatcher() .addCallback( new OnBackPressedCallback(true) { @@ -226,36 +243,61 @@ public void handleOnBackPressed() { * froze the screen. The verdict is delivered back to the main thread. */ private void validatePickedBackup(@NonNull Uri uri) { - runInBackground( - () -> - BackupFileValidator.isValidBackupFile( - this, - uri, - new BackupFileValidator.BackupValidatorCallback() { - @Override - public void onValid(String fileName) { - runOnUiThread( - () -> { - if (!isFinishing() && !isDestroyed()) { - presenter.readFileBackupLocal(uri); - } - }); - } + if (pendingRestorePick != null) { + // One pick at a time: a second one queued behind the first restored everything twice. + return; + } + pendingRestorePick = uri; + // Up before the verdict, not after: it is what stops the user from starting a second + // restore while this one is being judged, and it is the only feedback during the wait. + showProcessRestoreDialog(); + // A thread of its own rather than the sync executor: queued behind a running Drive sync + // the verdict waited minutes behind a modal dialog with nothing happening. + new Thread( + () -> + BackupFileValidator.isValidBackupFile( + this, + uri, + new BackupFileValidator.BackupValidatorCallback() { + @Override + public void onValid(String fileName) { + runOnUiThread( + () -> { + if (isFinishing() || isDestroyed()) { + return; + } + pendingRestorePick = null; + dismissProgressDialog(); + presenter.readFileBackupLocal(uri); + }); + } - @Override - public void onInvalid(String errorMessage) { - runOnUiThread( - () -> { - if (!isFinishing() && !isDestroyed()) { - onInfoSnack( - errorMessage, - null, - SnackBarInfo.Error, - Snackbar.LENGTH_LONG); - } - }); - } - })); + @Override + public void onInvalid(String errorMessage) { + runOnUiThread( + () -> { + if (isFinishing() || isDestroyed()) { + return; + } + pendingRestorePick = null; + dismissProgressDialog(); + onInfoSnack( + errorMessage, + null, + SnackBarInfo.Error, + Snackbar.LENGTH_LONG); + }); + } + }), + "backup-pick-validation") + .start(); + } + + private void dismissProgressDialog() { + if (progressDialog != null) { + progressDialog.dismiss(); + progressDialog = null; + } } /** The account tab draws itself from the state the next updateSyncUi() pushes. */ @@ -939,6 +981,14 @@ private boolean finishActivity() { return true; } + @Override + protected void onSaveInstanceState(@NonNull Bundle outState) { + super.onSaveInstanceState(outState); + if (pendingRestorePick != null) { + outState.putParcelable(STATE_PENDING_RESTORE_PICK, pendingRestorePick); + } + } + @Override protected void onDestroy() { super.onDestroy(); diff --git a/app/src/main/java/com/pasich/mynotes/utils/constants/DatabaseConstants.java b/app/src/main/java/com/pasich/mynotes/utils/constants/DatabaseConstants.java index 455b0d00..80068834 100644 --- a/app/src/main/java/com/pasich/mynotes/utils/constants/DatabaseConstants.java +++ b/app/src/main/java/com/pasich/mynotes/utils/constants/DatabaseConstants.java @@ -3,5 +3,5 @@ public class DatabaseConstants { public static final String DB_NAME = "MyNotes.db"; - public static final int DB_VERSION = 21; + public static final int DB_VERSION = 22; } diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackendTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackendTest.java index b41155d2..896485c1 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackendTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/GoogleDriveSyncBackendTest.java @@ -650,7 +650,7 @@ public void validateAncestry_walksALongLinearHistoryWithoutOverflowingTheStack() null, () -> { try { - GoogleDriveSyncBackend.validateAncestry(parents); + BundleHistory.validateAncestry(parents); } catch (Throwable error) { failure[0] = error; } @@ -672,7 +672,7 @@ public void validateAncestry_stillRejectsACycle() { parents.put("c", List.of("a", "missing")); try { - GoogleDriveSyncBackend.validateAncestry(parents); + BundleHistory.validateAncestry(parents); throw new AssertionError("Expected the cycle to be refused"); } catch (IOException expected) { assertThat(expected).hasMessageThat().contains("cycle"); @@ -762,6 +762,29 @@ public void publish_retiresABundleOnlyOnceItHasBeenSupersededForTheWholeGrace() .isEqualTo("Fourth"); } + @Test + public void theSupersessionMarkCarriesDrivesTimeAtTheMarkNotAtTheListing() throws Exception { + // The mark is written after every bundle has been downloaded; stamping it with the time + // of the listing that preceded the downloads shortened the grace by the whole read. + SyncBundleCodec codec = new SyncBundleCodec(); + byte[] base = codec.encode(snapshotWithTitle("Base"), CLOCK.instant()); + String baseId = codec.decode(new ByteArrayInputStream(base)).getBundleId(); + byte[] head = + codec.encode( + snapshotWithTitle("Head"), CLOCK.instant(), Collections.singleton(baseId)); + server.seedOwnedBundleBytes(base); + server.seedOwnedBundleBytes(head); + long tenMinutes = 10L * 60L * 1000L; + server.advanceClockOnEachDownload(tenMinutes); + + backend().readSnapshotResult(); + + // Two downloads happened before the mark, so Drive's clock had moved twenty minutes. + assertThat(server.supersessionMarkTimes()).hasSize(1); + assertThat(server.supersessionMarkTimes().get(0)) + .isAtLeast(CLOCK.millis() + 2L * tenMinutes); + } + @Test public void publish_measuresTheGraceOnDrivesClockNotThePhones() throws Exception { // A phone running two hours fast used to see every fresh mark as two hours old and @@ -1264,6 +1287,8 @@ private static final class FakeDriveServer implements AutoCloseable { /** Drive's clock, as the fake stamps files with it; starts at the tests' fixed CLOCK. */ private volatile long serverNowMillis = CLOCK.millis(); + private volatile long clockStepPerDownload; + private SyncSnapshot updateBeforeNextPatch; private final AtomicInteger nextId = new AtomicInteger(1); @@ -1487,6 +1512,21 @@ void advanceClock(long millis) { serverNowMillis += millis; } + /** Models slow downloads: Drive's clock moves by {@code millis} on each media read. */ + void advanceClockOnEachDownload(long millis) { + clockStepPerDownload = millis; + } + + /** The times recorded in the clients' supersession marks, in file order. */ + List supersessionMarkTimes() { + List times = new ArrayList<>(); + for (DriveFile file : files.values()) { + String mark = file.appProperties.get("mynotesBundleSuperseded"); + if (mark != null) times.add(Long.parseLong(mark)); + } + return times; + } + /** * Turns every supersession mark into a bare marker, as a client without server time writes. */ @@ -1724,6 +1764,7 @@ private Response handleFileRead(Request request, URI uri, String id) { } if ("media".equals(parseQuery(uri).get("alt"))) { mediaReads.merge(id, 1, Integer::sum); + serverNowMillis += clockStepPerDownload; return Response.binary(200, file.content, file.eTag()); } return Response.json(200, fileMetadata(file).toString(), file.eTag()); diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/LegacyNotePayloadTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/LegacyNotePayloadTest.java index ccc8cc42..e3a226ae 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/LegacyNotePayloadTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/LegacyNotePayloadTest.java @@ -67,18 +67,35 @@ public void theUpgradedPayloadHashesLikeTheCurrentBuildOfTheSameNote() { } @Test - public void leavesAPayloadWhoseIdsWereNotDerivedTheOldWayAlone() { - // An id a receiving device restored, or one already derived with the hash: not provably - // the old shape, so it is not touched. + public void keepsACanonicalIdButStillPutsTheBlocksIntoWireForm() { + // An id a receiving 2.6.50 device restored, repeated because the block was duplicated. The + // upgraded store keeps such ids as they are and maps the blocks by position; the decoded + // note has to come out the same way, or the two hash differently at the same timestamp + // and the note conflicts with itself on every sync. + String restoredId = "7d444840-9dc0-11d1-b245-5ffdce74fad2"; JsonObject payload = legacyPayload(LOCAL_URL, "photo.png"); - payload.getAsJsonArray("attachmentsManifest") - .get(0) - .getAsJsonObject() - .addProperty("id", "7d444840-9dc0-11d1-b245-5ffdce74fad2"); - String before = payload.toString(); + JsonObject entry = payload.getAsJsonArray("attachmentsManifest").get(0).getAsJsonObject(); + entry.addProperty("id", restoredId); + payload.getAsJsonArray("attachmentsManifest").add(entry.deepCopy()); + payload.getAsJsonArray("attachmentHashes").add(HASH); + payload.addProperty( + "f", + "[{\"type\":\"attaches\",\"data\":{\"file\":{\"url\":\"" + + LOCAL_URL + + "\"}}},{\"type\":\"attaches\",\"data\":{\"file\":{\"url\":\"" + + LOCAL_URL + + "\"}}}]"); - assertThat(LegacyNotePayload.upgrade(NOTE_ID, payload)).isFalse(); - assertThat(payload.toString()).isEqualTo(before); + assertThat(LegacyNotePayload.upgrade(NOTE_ID, payload)).isTrue(); + + JsonArray manifest = payload.getAsJsonArray("attachmentsManifest"); + assertThat(manifest.get(0).getAsJsonObject().get("id").getAsString()).isEqualTo(restoredId); + assertThat(manifest.get(1).getAsJsonObject().get("id").getAsString()).isEqualTo(restoredId); + assertThat(EditorAttachmentBlocks.fileUrls(payload.get("f").getAsString())) + .containsExactly( + AttachmentWireUrl.forLogicalId(restoredId), + AttachmentWireUrl.forLogicalId(restoredId)); + assertThat(payload.getAsJsonObject("attachmentNames").entrySet()).hasSize(1); } @Test diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/SyncBundleCodecTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/SyncBundleCodecTest.java index adc0980a..09e6b1f7 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/SyncBundleCodecTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/SyncBundleCodecTest.java @@ -427,8 +427,7 @@ public void encode_keepsBothBlobsWhenOneNoteCarriesOneIdWithTwoContents() throws } @Test - public void encode_republishesANoteThatRepeatsOneAttachmentWithoutCollapsingIt() - throws Exception { + public void encode_republishesANoteThatRepeatsOneAttachmentTheWay2650Did() throws Exception { // 2.6.50 receivers hold manifests with one entry repeated — a duplicated block whose file // was not replaced — and republish them as conflict alternatives after the upgrade. // Refusing the shape at encode failed every publish until the conflict was resolved; @@ -474,17 +473,19 @@ public void encode_republishesANoteThatRepeatsOneAttachmentWithoutCollapsingIt() .isEqualTo(ATTACHMENT_ID); assertThat(alternative.getCanonicalPayloadHash()) .isEqualTo(repeated.getCanonicalPayloadHash()); - // Never the shape the validator has to refuse: one wire id, two references. + // On the wire exactly as 2.6.50 published it — two references to the one entry and no + // alias field — so a device still on 2.6.50 reads it without stumbling over a field it + // does not know. JsonObject records = com.google.gson.JsonParser.parseString( unzipToStrings(bundle).get(SyncBundleCodec.ENTRY_RECORDS)) .getAsJsonObject(); - JsonArray wireIds = - records.getAsJsonArray("alternatives") - .get(0) - .getAsJsonObject() - .getAsJsonArray("attachmentIds"); - assertThat(wireIds.get(0).getAsString()).isNotEqualTo(wireIds.get(1).getAsString()); + JsonObject wireAlternative = + records.getAsJsonArray("alternatives").get(0).getAsJsonObject(); + JsonArray wireIds = wireAlternative.getAsJsonArray("attachmentIds"); + assertThat(wireIds.get(0).getAsString()).isEqualTo(ATTACHMENT_ID); + assertThat(wireIds.get(1).getAsString()).isEqualTo(ATTACHMENT_ID); + assertThat(wireAlternative.has(SyncBundleCodec.FIELD_ATTACHMENT_ID_ALIASES)).isFalse(); } @Test diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/SyncMergerTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/SyncMergerTest.java index 05061c4f..313a4616 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/SyncMergerTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/SyncMergerTest.java @@ -27,6 +27,60 @@ public void merge_identicalVersionIsNoOp() { assertThat(result.getDiscardedRecords()).isEmpty(); } + @Test + public void merge_publishesALocalEditWithoutAConflictWhileTheRemoteIsWhatWasLastSynced() { + // What every user does every day: edit a note after a clean sync. The remote still holds + // the version this device last published, so nobody else has moved — yet the merge asked + // the user to choose between their edit and the text they had just replaced. + SyncRecord synced = note(NOTE_ID, TEN, "Milk"); + SyncRecord edited = + note(NOTE_ID, TWENTY, "Milk and bread") + .withBaseVersion(synced.getCanonicalPayloadHash()); + + SyncMergeResult result = merger.merge(snapshot(edited), snapshot(synced)); + + assertThat(result.getMergedSnapshot().getRecords()).containsExactly(edited); + assertThat(result.getConflicts()).isEmpty(); + } + + @Test + public void merge_takesTheRemoteWithoutAConflictWhenOnlyTheOtherSideMoved() { + SyncRecord synced = note(NOTE_ID, TEN, "Milk"); + SyncRecord unchanged = synced.withBaseVersion(synced.getCanonicalPayloadHash()); + SyncRecord editedElsewhere = note(NOTE_ID, TWENTY, "Milk and eggs"); + + SyncMergeResult result = merger.merge(snapshot(unchanged), snapshot(editedElsewhere)); + + assertThat(result.getMergedSnapshot().getRecords()).containsExactly(editedElsewhere); + assertThat(result.getConflicts()).isEmpty(); + } + + @Test + public void merge_stillReportsAConflictWhenBothSidesMovedFromTheSameBase() { + SyncRecord synced = note(NOTE_ID, TEN, "Milk"); + SyncRecord editedHere = + note(NOTE_ID, TWENTY, "Milk and bread") + .withBaseVersion(synced.getCanonicalPayloadHash()); + SyncRecord editedElsewhere = note(NOTE_ID, TWENTY.plusSeconds(1), "Milk and eggs"); + + SyncMergeResult result = merger.merge(snapshot(editedHere), snapshot(editedElsewhere)); + + assertThat(result.getConflicts()).hasSize(1); + assertThat(result.getMergedSnapshot().getRecords()).containsExactly(editedElsewhere); + } + + @Test + public void merge_withoutAKnownBaseBehavesAsBefore() { + // A fresh install, or a record last synced by a build that did not record the base. + SyncRecord local = note(NOTE_ID, TWENTY, "Local"); + SyncRecord remote = note(NOTE_ID, TEN, "Remote"); + + SyncMergeResult result = merger.merge(snapshot(local), snapshot(remote)); + + assertThat(result.getMergedSnapshot().getRecords()).containsExactly(local); + assertThat(result.getConflicts()).hasSize(1); + } + @Test public void merge_keepsLocalOnlyAndRemoteOnlyRecords() { SyncRecord localNote = note(NOTE_ID, TEN, "Local"); diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/SyncMutationCoordinatorTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/SyncMutationCoordinatorTest.java index 22de4a23..f50a9f22 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/SyncMutationCoordinatorTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/SyncMutationCoordinatorTest.java @@ -623,6 +623,12 @@ public long insertIfAbsent(SyncMetadataEntity metadata) { return 1L; } + @Override + public void setSyncedVersion(String recordType, long localId, String versionId) { + SyncMetadataEntity current = rows.get(key(recordType, localId)); + if (current != null) current.syncedVersionId = versionId; + } + @Override public void setVersion(String recordType, long localId, long updatedAt, Long deletedAt) { SyncMetadataEntity current = get(recordType, localId); diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/SyncServiceTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/SyncServiceTest.java index a406d0b3..8ed63e97 100644 --- a/app/src/test/java/com/pasich/mynotes/data/sync/SyncServiceTest.java +++ b/app/src/test/java/com/pasich/mynotes/data/sync/SyncServiceTest.java @@ -305,6 +305,32 @@ public void sync_replacesCorruptLocalAttachmentWithVerifiedRemoteBlob() throws E assertThat(backend.writeSnapshotCalls).isEqualTo(1); } + @Test + public void sync_publishesAnOrdinaryLocalEditWithoutAskingTheUser() { + // Reproduced on a Pixel: fully synced, edit one note, sync — the conflict dialog offered + // the edit against the text it replaced, after every edit, with the wrong tap discarding + // the work. The remote still holds what this device last published, so only this side + // moved. + FakeStore store = new FakeStore(snapshot(note(TEN, "Milk"))); + FakeBackend backend = new FakeBackend(SyncSnapshot.empty()); + assertThat(new SyncService(store, new SyncMerger(), CLOCK).sync(backend).getStatus()) + .isEqualTo(SyncState.Status.SUCCESS); + store.snapshot = snapshot(note(TWENTY, "Milk and bread")); + + SyncState state = new SyncService(store, new SyncMerger(), CLOCK).sync(backend); + + assertThat(state.getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(state.getConflictCount()).isEqualTo(0); + assertThat(store.appliedConflicts).isEmpty(); + assertThat( + backend.snapshot + .find(SyncRecord.Type.NOTE, NOTE_ID) + .getPayload() + .get("value") + .getAsString()) + .isEqualTo("Milk and bread"); + } + @Test public void runWhileNoSyncRuns_waitsForTheSyncInFlightToFinish() throws Exception { // Disconnect wipes state, conflicts and the blob cache. Done concurrently with the @@ -518,15 +544,24 @@ private static final class FakeStore implements SyncStore { this.snapshot = snapshot; } + /** The version the remote is known to hold, per record, as RoomSyncStore remembers it. */ + private final Map syncedVersions = new HashMap<>(); + @Override public SyncSnapshot readSnapshot() { - return snapshot; + List withBases = new ArrayList<>(); + for (SyncRecord record : snapshot.getRecords()) { + String base = syncedVersions.get(record.getType() + ":" + record.getId()); + // Same instance when nothing is known, so identity-based assertions still hold. + withBases.add(base == null ? record : record.withBaseVersion(base)); + } + return new SyncSnapshot(withBases); } @Override public SnapshotBuildResult buildSnapshot() { return snapshotBuildResult == null - ? SnapshotBuildResult.publishable(snapshot) + ? SnapshotBuildResult.publishable(readSnapshot()) : snapshotBuildResult; } @@ -537,6 +572,10 @@ public void applySnapshot(SyncSnapshot snapshot, List appliedSnapshot = snapshot; appliedConflicts = new ArrayList<>(conflicts); applyCalls++; + for (SyncRecord record : snapshot.getRecords()) { + syncedVersions.put( + record.getType() + ":" + record.getId(), record.getCanonicalPayloadHash()); + } } @Override diff --git a/app/src/test/java/com/pasich/mynotes/data/sync/TwoPeerConvergenceTest.java b/app/src/test/java/com/pasich/mynotes/data/sync/TwoPeerConvergenceTest.java new file mode 100644 index 00000000..e30b79f8 --- /dev/null +++ b/app/src/test/java/com/pasich/mynotes/data/sync/TwoPeerConvergenceTest.java @@ -0,0 +1,470 @@ +package com.pasich.mynotes.data.sync; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.gson.JsonObject; +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.io.InputStream; +import java.time.Clock; +import java.time.Duration; +import java.time.Instant; +import java.time.ZoneOffset; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.LinkedHashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +import java.util.UUID; +import org.junit.Test; + +/** + * Two peers, one shared bundle history, a conflict settled by hand. + * + *

Reproduced on a Pixel: peer A deletes a task and syncs, peer B renames it and syncs, B gets + * the right live-versus-deleted conflict, keeps the live version — and the same conflict returned + * on every sync forever, because the resolution never reached Drive. The stores here model exactly + * what RoomSyncStore does with versions, bases and resolutions, against the real codec, merger, + * service and bundle fold; only the transport is in memory. + */ +public class TwoPeerConvergenceTest { + + private static final String RECORD_ID = "550e8400-e29b-41d4-a716-446655440000"; + private static final Instant START = Instant.parse("2026-09-05T12:00:00Z"); + + private final MutableClock clock = new MutableClock(START); + private final InMemoryBundleBackend drive = new InMemoryBundleBackend(clock); + + @Test + public void aTaskDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() { + deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type.TASK); + } + + @Test + public void aNoteDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() { + deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type.NOTE); + } + + @Test + public void aTagDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() { + deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type.TAG); + } + + @Test + public void aCategoryDeletedOnOnePeerAndEditedOnTheOtherSettlesAfterOneResolution() { + deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type.CATEGORY); + } + + @Test + public void aTaskSyncedBeforeBasesWereRecordedStillSettlesAfterOneResolution() { + deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type.TASK, true); + } + + private void deletedHereEditedThereSettlesAfterOneResolution(SyncRecord.Type type) { + deletedHereEditedThereSettlesAfterOneResolution(type, false); + } + + private void deletedHereEditedThereSettlesAfterOneResolution( + SyncRecord.Type type, boolean migratedWithoutBases) { + Peer a = new Peer("A", clock, drive); + Peer b = new Peer("B", clock, drive); + a.create(type, RECORD_ID, "TaskX"); + clock.advance(); + assertThat(a.sync().getConflictCount()).isEqualTo(0); + clock.advance(); + assertThat(b.sync().getConflictCount()).isEqualTo(0); + clock.advance(); + assertThat(a.sync().getConflictCount()).isEqualTo(0); + assertThat(b.store.pendingConflicts()).isEmpty(); + assertThat(a.store.pendingConflicts()).isEmpty(); + if (migratedWithoutBases) { + // Rows migrated from a build that recorded no synced version: null until the next + // sync fills them in. + a.store.bases.clear(); + b.store.bases.clear(); + } + + // A deletes and syncs; B edits and syncs. A genuine conflict, seen once on B. + a.delete(type, RECORD_ID); + clock.advance(); + // A record whose synced version is not yet known merges as before: the deletion is + // reported against the version it replaced, and that version travels as an alternative. + assertThat(a.sync().getConflictCount()).isEqualTo(migratedWithoutBases ? 1 : 0); + clock.advance(); + b.edit(type, RECORD_ID, "TaskX-B"); + clock.advance(); + SyncState firstOnB = b.sync(); + assertThat(firstOnB.getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(b.store.pendingConflicts()).hasSize(migratedWithoutBases ? 2 : 1); + + // B keeps the live version, in one sitting — the dialog offers every open conflict — and + // syncs: the choice reaches the history, and nothing is offered again — on B, on B once + // more, or on A, which gets the record back. + clock.advance(); + b.store.resolveAllKeepingLive(clock.instant()); + clock.advance(); + SyncState publishing = b.sync(); + assertThat(publishing.getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(publishing.getErrorMessage()).isNull(); + assertThat(b.store.pendingConflicts()).isEmpty(); + clock.advance(); + SyncState secondOnB = b.sync(); + assertThat(secondOnB.getConflictCount()).isEqualTo(0); + assertThat(b.store.pendingConflicts()).isEmpty(); + clock.advance(); + SyncState onA = a.sync(); + assertThat(onA.getStatus()).isEqualTo(SyncState.Status.SUCCESS); + assertThat(onA.getConflictCount()).isEqualTo(0); + assertThat(a.store.pendingConflicts()).isEmpty(); + assertThat(a.store.titleOf(type, RECORD_ID)).isEqualTo("TaskX-B"); + assertThat(b.store.titleOf(type, RECORD_ID)).isEqualTo("TaskX-B"); + clock.advance(); + assertThat(a.sync().getConflictCount()).isEqualTo(0); + clock.advance(); + assertThat(b.sync().getConflictCount()).isEqualTo(0); + } + + // ------------------------------------------------------------------ the two peers + + private static final class Peer { + final String name; + final PeerStore store; + private final Clock clock; + private final SyncBackend backend; + + Peer(String name, Clock clock, SyncBackend backend) { + this.name = name; + this.clock = clock; + this.backend = backend; + this.store = new PeerStore(clock); + } + + SyncState sync() { + return new SyncService(store, new SyncMerger(), clock).sync(backend); + } + + void create(SyncRecord.Type type, String id, String title) { + store.put(SyncRecord.live(type, id, clock.instant(), payload(title)), null); + } + + void edit(SyncRecord.Type type, String id, String title) { + store.touchLive(type, id, payload(title)); + } + + void delete(SyncRecord.Type type, String id) { + store.tombstone(type, id); + } + + private static JsonObject payload(String title) { + JsonObject payload = new JsonObject(); + payload.addProperty("title", title); + return payload; + } + } + + /** RoomSyncStore's versioning, bases, conflict rows and resolutions, without Room. */ + static final class PeerStore implements SyncStore { + private final Clock clock; + private final Map records = new LinkedHashMap<>(); + private final Map bases = new LinkedHashMap<>(); + private final List conflicts = new ArrayList<>(); + private SyncState state = SyncState.idle(); + + PeerStore(Clock clock) { + this.clock = clock; + } + + static final class ConflictRow { + final SyncMergeResult.Conflict conflict; + boolean resolved; + + ConflictRow(SyncMergeResult.Conflict conflict) { + this.conflict = conflict; + } + } + + void put(SyncRecord record, String base) { + records.put(key(record), record); + if (base != null) bases.put(key(record), base); + } + + void touchLive(SyncRecord.Type type, String id, JsonObject payload) { + SyncRecord current = records.get(type + ":" + id); + records.put(type + ":" + id, SyncRecord.live(type, id, next(current), payload)); + } + + void tombstone(SyncRecord.Type type, String id) { + SyncRecord current = records.get(type + ":" + id); + Instant at = next(current); + records.put(type + ":" + id, SyncRecord.tombstone(type, id, at, at)); + } + + private Instant next(SyncRecord current) { + Instant now = clock.instant(); + return current == null || now.isAfter(current.getUpdatedAt()) + ? now + : current.getUpdatedAt().plusMillis(1); + } + + String titleOf(SyncRecord.Type type, String id) { + SyncRecord record = records.get(type + ":" + id); + return record == null || record.isTombstone() + ? null + : record.getPayload().get("title").getAsString(); + } + + List pendingConflicts() { + List pending = new ArrayList<>(); + for (ConflictRow row : conflicts) { + if (!row.resolved) pending.add(row); + } + return pending; + } + + /** The dialog loop: every open conflict, one after another, keeping the live version. */ + void resolveAllKeepingLive(Instant resolvedAt) { + for (ConflictRow row : new ArrayList<>(pendingConflicts())) { + resolveKeepingLive(row, resolvedAt); + } + } + + /** + * What resolveConflict does: drop a row whose record's content has moved on, otherwise + * re-time the chosen version and mark the row settled. + */ + void resolveKeepingLive(ConflictRow row, Instant resolvedAt) { + SyncRecord chosen = + row.conflict.getWinner().isTombstone() + ? row.conflict.getLoser() + : row.conflict.getWinner(); + String key = key(chosen); + SyncRecord current = records.get(key); + if (current != null + && current.getUpdatedAt() + .isAfter( + row.conflict + .getWinner() + .getUpdatedAt() + .isAfter(row.conflict.getLoser().getUpdatedAt()) + ? row.conflict.getWinner().getUpdatedAt() + : row.conflict.getLoser().getUpdatedAt()) + && !contentDigest(current).equals(contentDigest(row.conflict.getWinner())) + && !contentDigest(current).equals(contentDigest(row.conflict.getLoser()))) { + conflicts.remove(row); + return; + } + Instant updatedAt = + current != null && !resolvedAt.isAfter(current.getUpdatedAt()) + ? current.getUpdatedAt().plusMillis(1) + : resolvedAt; + records.put( + key, + SyncRecord.live( + chosen.getType(), chosen.getId(), updatedAt, chosen.getPayload())); + row.resolved = true; + } + + @Override + public SyncSnapshot readSnapshot() { + List withBases = new ArrayList<>(); + for (Map.Entry entry : records.entrySet()) { + withBases.add(entry.getValue().withBaseVersion(bases.get(entry.getKey()))); + } + return new SyncSnapshot(withBases); + } + + @Override + public void applySnapshot(SyncSnapshot snapshot, List incoming) { + Set skipped = new LinkedHashSet<>(); + for (SyncRecord record : snapshot.getRecords()) { + String key = key(record); + SyncRecord current = records.get(key); + if (current != null && current.getUpdatedAt().isAfter(record.getUpdatedAt())) { + skipped.add(key); + continue; + } + records.put(key, record.withBaseVersion(null)); + bases.put(key, record.getCanonicalPayloadHash()); + conflicts.removeIf( + row -> + !row.resolved + && key.equals( + row.conflict.getType() + ":" + row.conflict.getId()) + && !row.conflict + .getWinnerVersionId() + .equals(record.getCanonicalPayloadHash())); + } + for (SyncMergeResult.Conflict conflict : incoming) { + String key = conflict.getType() + ":" + conflict.getId(); + if (skipped.contains(key)) continue; + conflicts.removeIf( + row -> + !row.resolved + && key.equals( + row.conflict.getType() + ":" + row.conflict.getId()) + && !row.conflict + .getWinnerVersionId() + .equals(conflict.getWinnerVersionId())); + boolean duplicate = false; + for (ConflictRow row : conflicts) { + if (row.conflict.getWinnerVersionId().equals(conflict.getWinnerVersionId()) + && row.conflict + .getLoserVersionId() + .equals(conflict.getLoserVersionId())) { + duplicate = true; + } + } + if (!duplicate) conflicts.add(new ConflictRow(conflict)); + } + } + + @Override + public Set getResolvedAlternativeIds() { + Set settled = new LinkedHashSet<>(); + for (ConflictRow row : conflicts) { + if (row.resolved) { + settled.add(row.conflict.getWinnerVersionId()); + settled.add(row.conflict.getLoserVersionId()); + } + } + return settled; + } + + @Override + public Collection getAttachmentHashes(SyncSnapshot snapshot) { + return Collections.emptyList(); + } + + @Override + public boolean hasAttachment(String sha256) { + return false; + } + + @Override + public InputStream readAttachment(String sha256) { + return new ByteArrayInputStream(new byte[0]); + } + + @Override + public void writeAttachment(String sha256, long sizeBytes, InputStream content) {} + + @Override + public SyncState readState() { + return state; + } + + @Override + public void writeState(SyncState state) { + this.state = state; + } + + private static String key(SyncRecord record) { + return record.getType() + ":" + record.getId(); + } + + private static String contentDigest(SyncRecord record) { + return record.isTombstone() + ? "tombstone" + : SyncRecord.live( + record.getType(), + record.getId(), + Instant.EPOCH, + record.getPayload()) + .getCanonicalPayloadHash(); + } + } + + /** Drive without the network: the same codec and the same fold over the same history. */ + static final class InMemoryBundleBackend implements SyncBackend { + private final Clock clock; + private final SyncBundleCodec codec = new SyncBundleCodec(); + private final Map bundles = new LinkedHashMap<>(); + private String lastReadToken = ""; + + InMemoryBundleBackend(Clock clock) { + this.clock = clock; + } + + @Override + public String getIdentifier() { + return "memory"; + } + + @Override + public RemoteSnapshot readSnapshotResult() throws IOException { + Map decoded = new LinkedHashMap<>(); + for (byte[] bytes : bundles.values()) { + SyncBundleCodec.DecodedBundle bundle = + codec.decode(new ByteArrayInputStream(bytes)); + decoded.put(bundle.getBundleId(), bundle); + } + BundleHistory.Fold fold = BundleHistory.fold(decoded, new SyncMerger()); + lastReadToken = UUID.randomUUID().toString(); + return new RemoteSnapshot( + fold.merged, + fold.conflicts, + fold.frontier, + fold.alternatives, + fold.resolvedAlternativeIds, + lastReadToken); + } + + @Override + public void publish(SyncPublication publication) throws IOException { + if (!publication.getReadContext().getReadToken().equals(lastReadToken)) { + throw new IOException("stale read context"); + } + byte[] bytes = + codec.encode( + publication.getSnapshot(), + clock.instant(), + publication.getReadContext().getFrontierBundleIds(), + publication.getUnresolvedAlternatives(), + publication.getResolvedAlternativeIds()); + bundles.put(codec.decode(new ByteArrayInputStream(bytes)).getBundleId(), bytes); + } + + @Override + public boolean hasAttachment(String sha256) { + return false; + } + + @Override + public InputStream readAttachment(String sha256) { + return null; + } + + @Override + public void writeAttachment(String sha256, long sizeBytes, InputStream content) {} + } + + private static final class MutableClock extends Clock { + private Instant now; + + MutableClock(Instant start) { + now = start; + } + + void advance() { + now = now.plus(Duration.ofMinutes(1)); + } + + @Override + public ZoneOffset getZone() { + return ZoneOffset.UTC; + } + + @Override + public Clock withZone(java.time.ZoneId zone) { + return this; + } + + @Override + public Instant instant() { + return now; + } + } +}