From 0e651134dff4be02e636eace1a198289d06967c9 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 16:03:05 +0200 Subject: [PATCH 1/4] feat(sync): keep only the devices the user chooses on a restore --- ...11-07-feat-sync2-restore-choose-devices.md | 2 + .../10.json | 649 ++++++++++++++++++ .../aedev/flow/sync2/data/GroupState.kt | 8 + .../aedev/flow/sync2/data/LinkedDatabase.kt | 3 +- .../aedev/flow/sync2/engine/EpochKeys.kt | 59 ++ .../flow/sync2/engine/LinkedException.kt | 2 + .../github/aedev/flow/sync2/engine/Session.kt | 52 +- .../aedev/flow/sync2/engine/Snapshots.kt | 2 +- .../aedev/flow/sync2/linking/DeviceNames.kt | 20 +- .../aedev/flow/sync2/linking/Membership.kt | 18 +- .../aedev/flow/sync2/linking/Recovery.kt | 136 +++- .../aedev/flow/sync2/live/LiveChannel.kt | 4 +- .../aedev/flow/sync2/live/LiveEnvelopes.kt | 2 +- .../aedev/flow/sync2/service/LinkedActions.kt | 46 +- .../aedev/flow/sync2/service/LinkedError.kt | 5 + .../aedev/flow/sync2/service/LinkedSync.kt | 16 + .../io/github/aedev/flow/sync2/store/Store.kt | 14 +- .../sync/linked/LinkedDevicesSection.kt | 3 + .../sync/linked/LinkedDevicesViewModel.kt | 40 +- .../ui/screens/sync/linked/LinkedErrors.kt | 1 + .../ui/screens/sync/linked/LinkedSetup.kt | 44 ++ app/src/main/res/values/strings.xml | 4 + .../aedev/flow/sync2/LinkedErrorTest.kt | 1 + .../aedev/flow/sync2/RestoreChoiceTest.kt | 421 ++++++++++++ .../aedev/flow/sync2/WebDavGroupTest.kt | 4 +- .../ui/screens/sync/linked/LinkedSetupTest.kt | 29 +- app/src/test/resources/sync2/README.md | 5 +- app/src/test/resources/sync2/forks.json | 173 +++++ app/src/test/resources/sync2/roster.json | 177 +++++ 29 files changed, 1856 insertions(+), 84 deletions(-) create mode 100644 .github/nightly-notes/2026-11-07-feat-sync2-restore-choose-devices.md create mode 100644 app/schemas/io.github.aedev.flow.sync2.data.LinkedDatabase/10.json create mode 100644 app/src/main/java/io/github/aedev/flow/sync2/engine/EpochKeys.kt create mode 100644 app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt diff --git a/.github/nightly-notes/2026-11-07-feat-sync2-restore-choose-devices.md b/.github/nightly-notes/2026-11-07-feat-sync2-restore-choose-devices.md new file mode 100644 index 000000000..de6a3ac5b --- /dev/null +++ b/.github/nightly-notes/2026-11-07-feat-sync2-restore-choose-devices.md @@ -0,0 +1,2 @@ +## Improved +- Restoring linked devices with your recovery words now asks which devices you still have, and signs out the rest diff --git a/app/schemas/io.github.aedev.flow.sync2.data.LinkedDatabase/10.json b/app/schemas/io.github.aedev.flow.sync2.data.LinkedDatabase/10.json new file mode 100644 index 000000000..e8660e84b --- /dev/null +++ b/app/schemas/io.github.aedev.flow.sync2.data.LinkedDatabase/10.json @@ -0,0 +1,649 @@ +{ + "formatVersion": 1, + "database": { + "version": 10, + "identityHash": "453d83ce298638949789b3516be2f4a6", + "entities": [ + { + "tableName": "sent", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`link` TEXT NOT NULL DEFAULT '', `collection` TEXT NOT NULL, `recordKey` TEXT NOT NULL, `hash` TEXT NOT NULL, `sentAt` INTEGER NOT NULL, PRIMARY KEY(`link`, `collection`, `recordKey`))", + "fields": [ + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true, + "defaultValue": "''" + }, + { + "fieldPath": "collection", + "columnName": "collection", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "recordKey", + "columnName": "recordKey", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "hash", + "columnName": "hash", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "sentAt", + "columnName": "sentAt", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "link", + "collection", + "recordKey" + ] + } + }, + { + "tableName": "group_record", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`groupId` TEXT NOT NULL, `storeUrl` TEXT NOT NULL, `instance` TEXT, `link` TEXT NOT NULL, `ownSeq` INTEGER NOT NULL, `ownPrev` TEXT NOT NULL, `cursor` INTEGER NOT NULL, `recoverySaved` INTEGER NOT NULL DEFAULT 0, `serverCursor` INTEGER NOT NULL DEFAULT 0, `uploaded` INTEGER NOT NULL DEFAULT 0, `loginDue` INTEGER, `loginSent` INTEGER, `recoveryNoticeDevice` TEXT, `recoveryNoticeAt` INTEGER, `recoveryDue` INTEGER NOT NULL DEFAULT 0, `restoreKeep` TEXT, `id` INTEGER NOT NULL, PRIMARY KEY(`id`))", + "fields": [ + { + "fieldPath": "groupId", + "columnName": "groupId", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "storeUrl", + "columnName": "storeUrl", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "instance", + "columnName": "instance", + "affinity": "TEXT" + }, + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "ownSeq", + "columnName": "ownSeq", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "ownPrev", + "columnName": "ownPrev", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "cursor", + "columnName": "cursor", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "recoverySaved", + "columnName": "recoverySaved", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "serverCursor", + "columnName": "serverCursor", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "uploaded", + "columnName": "uploaded", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "loginDue", + "columnName": "loginDue", + "affinity": "INTEGER" + }, + { + "fieldPath": "loginSent", + "columnName": "loginSent", + "affinity": "INTEGER" + }, + { + "fieldPath": "recoveryNoticeDevice", + "columnName": "recoveryNoticeDevice", + "affinity": "TEXT" + }, + { + "fieldPath": "recoveryNoticeAt", + "columnName": "recoveryNoticeAt", + "affinity": "INTEGER" + }, + { + "fieldPath": "recoveryDue", + "columnName": "recoveryDue", + "affinity": "INTEGER", + "notNull": true, + "defaultValue": "0" + }, + { + "fieldPath": "restoreKeep", + "columnName": "restoreKeep", + "affinity": "TEXT" + }, + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "id" + ] + } + }, + { + "tableName": "roster", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`seq` INTEGER NOT NULL, `entry` TEXT NOT NULL, PRIMARY KEY(`seq`))", + "fields": [ + { + "fieldPath": "seq", + "columnName": "seq", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "entry", + "columnName": "entry", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "seq" + ] + } + }, + { + "tableName": "roster_pin", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`hash` TEXT NOT NULL, PRIMARY KEY(`hash`))", + "fields": [ + { + "fieldPath": "hash", + "columnName": "hash", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "hash" + ] + } + }, + { + "tableName": "streams", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `seq` INTEGER NOT NULL, `hash` TEXT NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "seq", + "columnName": "seq", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "hash", + "columnName": "hash", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "snapshots", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`link` TEXT NOT NULL, `writtenMs` INTEGER NOT NULL, `mergedBytes` INTEGER NOT NULL, PRIMARY KEY(`link`))", + "fields": [ + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "writtenMs", + "columnName": "writtenMs", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "mergedBytes", + "columnName": "mergedBytes", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "link" + ] + } + }, + { + "tableName": "acks", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `vv` TEXT NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "vv", + "columnName": "vv", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "manifests", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `seq` INTEGER NOT NULL, `manifest` TEXT NOT NULL, PRIMARY KEY(`did`, `seq`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "seq", + "columnName": "seq", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "manifest", + "columnName": "manifest", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did", + "seq" + ] + } + }, + { + "tableName": "links", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`link` TEXT NOT NULL, PRIMARY KEY(`link`))", + "fields": [ + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "link" + ] + } + }, + { + "tableName": "collected", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `through` INTEGER NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "through", + "columnName": "through", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "held", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`cursor` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `path` TEXT NOT NULL, `size` INTEGER NOT NULL)", + "fields": [ + { + "fieldPath": "cursor", + "columnName": "cursor", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "path", + "columnName": "path", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "size", + "columnName": "size", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "cursor" + ] + }, + "indices": [ + { + "name": "index_held_path", + "unique": true, + "columnNames": [ + "path" + ], + "orders": [], + "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_held_path` ON `${TABLE_NAME}` (`path`)" + } + ] + }, + { + "tableName": "live_counter", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `seq` INTEGER NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "seq", + "columnName": "seq", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "consent", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `allowed` INTEGER NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "allowed", + "columnName": "allowed", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "lan_address", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`did` TEXT NOT NULL, `host` TEXT NOT NULL, `port` INTEGER NOT NULL, PRIMARY KEY(`did`))", + "fields": [ + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "host", + "columnName": "host", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "port", + "columnName": "port", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "did" + ] + } + }, + { + "tableName": "profile_links", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`profile` TEXT NOT NULL, `link` TEXT NOT NULL, `caughtUp` INTEGER NOT NULL, PRIMARY KEY(`profile`))", + "fields": [ + { + "fieldPath": "profile", + "columnName": "profile", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "caughtUp", + "columnName": "caughtUp", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "profile" + ] + }, + "indices": [ + { + "name": "index_profile_links_link", + "unique": true, + "columnNames": [ + "link" + ], + "orders": [], + "createSql": "CREATE UNIQUE INDEX IF NOT EXISTS `index_profile_links_link` ON `${TABLE_NAME}` (`link`)" + } + ] + }, + { + "tableName": "kept_batches", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`id` INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL, `link` TEXT NOT NULL, `sender` TEXT NOT NULL, `collection` TEXT NOT NULL)", + "fields": [ + { + "fieldPath": "id", + "columnName": "id", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "sender", + "columnName": "sender", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "collection", + "columnName": "collection", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": true, + "columnNames": [ + "id" + ] + }, + "indices": [ + { + "name": "index_kept_batches_link", + "unique": false, + "columnNames": [ + "link" + ], + "orders": [], + "createSql": "CREATE INDEX IF NOT EXISTS `index_kept_batches_link` ON `${TABLE_NAME}` (`link`)" + } + ] + }, + { + "tableName": "profile_names", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`link` TEXT NOT NULL, `name` TEXT NOT NULL, `hlc` TEXT NOT NULL, `deleted` INTEGER NOT NULL, PRIMARY KEY(`link`))", + "fields": [ + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "name", + "columnName": "name", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "hlc", + "columnName": "hlc", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "deleted", + "columnName": "deleted", + "affinity": "INTEGER", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "link" + ] + } + }, + { + "tableName": "link_reads", + "createSql": "CREATE TABLE IF NOT EXISTS `${TABLE_NAME}` (`link` TEXT NOT NULL, `did` TEXT NOT NULL, `seq` INTEGER NOT NULL, `hash` TEXT NOT NULL, PRIMARY KEY(`link`, `did`))", + "fields": [ + { + "fieldPath": "link", + "columnName": "link", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "did", + "columnName": "did", + "affinity": "TEXT", + "notNull": true + }, + { + "fieldPath": "seq", + "columnName": "seq", + "affinity": "INTEGER", + "notNull": true + }, + { + "fieldPath": "hash", + "columnName": "hash", + "affinity": "TEXT", + "notNull": true + } + ], + "primaryKey": { + "autoGenerate": false, + "columnNames": [ + "link", + "did" + ] + } + } + ], + "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, '453d83ce298638949789b3516be2f4a6')" + ] + } +} \ No newline at end of file diff --git a/app/src/main/java/io/github/aedev/flow/sync2/data/GroupState.kt b/app/src/main/java/io/github/aedev/flow/sync2/data/GroupState.kt index fc47e273c..3740dc8ea 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/data/GroupState.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/data/GroupState.kt @@ -27,6 +27,9 @@ val GroupRecord.webDavUrl: String? get() = storeUrl.takeIf { it.startsWith(WEBDA /** The Flow server the group lives on, or null for a group on WebDAV or on its devices alone. */ val GroupRecord.serverUrl: String? get() = storeUrl.takeIf { it != LAN_STORE && webDavUrl == null } +/** The devices a restore keeps, or null when no restore is unfinished. */ +val GroupRecord.keptDevices: Set? get() = restoreKeep?.split(',')?.filterTo(HashSet()) { it.isNotEmpty() } + /** A roster entry the user kept between sides of a contested roster. It never leaves this device. */ @Entity(tableName = "roster_pin") data class RosterPinRow( @@ -58,6 +61,8 @@ data class GroupRecord( val recoveryNoticeAt: Long? = null, /** The device that set the recovery key left: nothing is written until it is replaced. */ @ColumnInfo(defaultValue = "0") val recoveryDue: Boolean = false, + /** The devices a restore keeps, comma-separated, until it has signed out the rest (spec/protocol.md, Recovery). */ + val restoreKeep: String? = null, @PrimaryKey val id: Int = 1, ) @@ -118,6 +123,9 @@ interface GroupDao { @Query("UPDATE group_record SET recoveryDue = :due WHERE id = 1") suspend fun setRecoveryDue(due: Boolean) + @Query("UPDATE group_record SET restoreKeep = NULL WHERE id = 1") + suspend fun forgetRestoreKeep() + @Query("UPDATE group_record SET recoverySaved = :saved WHERE id = 1") suspend fun setRecoverySaved(saved: Boolean) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/data/LinkedDatabase.kt b/app/src/main/java/io/github/aedev/flow/sync2/data/LinkedDatabase.kt index 8c2d45655..ab6cf1805 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/data/LinkedDatabase.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/data/LinkedDatabase.kt @@ -62,7 +62,7 @@ interface SentDao { ProfileNameRow::class, LinkReadRow::class, ], - version = 9, + version = 10, exportSchema = true, autoMigrations = [ AutoMigration(from = 1, to = 2), @@ -73,6 +73,7 @@ interface SentDao { AutoMigration(from = 6, to = 7), AutoMigration(from = 7, to = 8), AutoMigration(from = 8, to = 9), + AutoMigration(from = 9, to = 10), ], ) abstract class LinkedDatabase : RoomDatabase() { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/EpochKeys.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/EpochKeys.kt new file mode 100644 index 000000000..e8623beed --- /dev/null +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/EpochKeys.kt @@ -0,0 +1,59 @@ +package io.github.aedev.flow.sync2.engine + +import io.github.aedev.flow.sync2.crypto.ProtocolError +import io.github.aedev.flow.sync2.crypto.ProtocolException +import io.github.aedev.flow.sync2.crypto.checkEpochKey +import io.github.aedev.flow.sync2.crypto.openChain +import io.github.aedev.flow.sync2.crypto.reject +import io.github.aedev.flow.sync2.store.Store +import io.github.aedev.flow.sync2.store.kcName +import java.util.TreeMap + +/** The epoch keys held so far, and older ones walked down the history chain from the oldest of them. */ +internal class EpochKeys( + private val store: Store, + private val group: ByteArray, + /** The `kc` of each epoch on the chain. */ + private val commitments: Map, +) { + private val keys = TreeMap() + + operator fun set( + epoch: Int, + key: ByteArray, + ) { + keys[epoch] = key + } + + fun held(epoch: Int): ByteArray = keys.getValue(epoch) + + suspend fun key(epoch: Int): ByteArray { + while (epoch !in keys) { + val (known, key) = keys.firstEntry() ?: reject(ProtocolError.KEY_COMMITMENT) + if (known <= epoch) reject(ProtocolError.KEY_COMMITMENT) + val knownCommitment = commitments[known] ?: reject(ProtocolError.KEY_COMMITMENT) + val kc = kcName(knownCommitment) + val chain = store.key(known, kc, "chain") ?: reject(ProtocolError.KEY_COMMITMENT) + val below = commitments[known - 1] ?: reject(ProtocolError.KEY_COMMITMENT) + keys[known - 1] = + forgettingIfBad(known, kc, "chain") { + openChain(key, group, known, chain).also { checkEpochKey(it, group, known - 1, below) } + } + } + return keys.getValue(epoch) + } + + /** A wrap or chain box that fails its checks isn't kept: it may be fine next time the store is asked. */ + suspend fun forgettingIfBad( + epoch: Int, + kc: String, + name: String, + open: () -> T, + ): T = + try { + open() + } catch (e: ProtocolException) { + store.forgetKey(epoch, kc, name) + throw e + } +} diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/LinkedException.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/LinkedException.kt index 00e205b52..a71d4e9fe 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/LinkedException.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/LinkedException.kt @@ -10,6 +10,8 @@ sealed class LinkedException( class RecoveryDue : LinkedException("The recovery key has to be replaced before anything is written") + class RestoreUnfinished : LinkedException("A restore has to sign out the devices the user didn't choose before anything is written") + class RemovalUnconfirmed : LinkedException("The roster that leaves this device out replaces entries it had applied") class Contested : LinkedException("Two sides of the roster removed each other") diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt index 3cacc3f33..3eabdef33 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/Session.kt @@ -14,7 +14,6 @@ import io.github.aedev.flow.sync2.crypto.Purpose import io.github.aedev.flow.sync2.crypto.checkEpochKey import io.github.aedev.flow.sync2.crypto.decodePayload import io.github.aedev.flow.sync2.crypto.encodePayload -import io.github.aedev.flow.sync2.crypto.openChain import io.github.aedev.flow.sync2.crypto.purposeKey import io.github.aedev.flow.sync2.crypto.reject import io.github.aedev.flow.sync2.crypto.sha256 @@ -57,7 +56,6 @@ import io.github.aedev.flow.sync2.store.sealObject import kotlinx.serialization.json.JsonObject import kotlinx.serialization.json.JsonPrimitive import java.security.MessageDigest -import java.util.TreeMap /** What a log object carries, once checked and opened. */ internal sealed interface Content { @@ -121,7 +119,7 @@ class Session private constructor( ) { internal val me = SyncBytes.toHex(device.did) private val commitments = HashMap() - private val keys = TreeMap() + private val keys = EpochKeys(store, group, commitments) /** The `kc` of each entry that lost a fork and stands, by the epoch it starts. */ private val standing = standing.mapNotNull { it.epoch }.groupBy({ it.n }, { it.kc }) @@ -135,7 +133,16 @@ class Session private constructor( var record = record private set - internal fun currentKey(): ByteArray = keys.getValue(state.epoch) + internal fun currentKey(): ByteArray = keys.held(state.epoch) + + /** A restore hasn't signed out the devices the user didn't choose yet, so they hold this key (spec/protocol.md, Recovery). */ + val removalPending: Boolean get() = record.restoreKeep != null + + /** The current key for sealing anything other devices will read. */ + internal fun sealingKey(): ByteArray { + if (removalPending) throw LinkedException.RestoreUnfinished() + return currentKey() + } init { for (entry in entries) { @@ -233,41 +240,13 @@ class Session private constructor( ): ByteArray { val kc = kcName(commitment) val wrap = store.key(epoch, kc, me) ?: reject(ProtocolError.KEY_COMMITMENT) - return forgettingIfBad(epoch, kc, me) { + return keys.forgettingIfBad(epoch, kc, me) { unwrapEpochKey(device.kemSecret, group, epoch, device.did, wrap).also { checkEpochKey(it, group, epoch, commitment) } } } - /** A wrap or chain box that fails its checks isn't kept: it may be fine next time the store is asked. */ - private suspend fun forgettingIfBad( - epoch: Int, - kc: String, - name: String, - open: () -> T, - ): T = - try { - open() - } catch (e: ProtocolException) { - store.forgetKey(epoch, kc, name) - throw e - } - /** The key of [epoch], walking the history chain back from the oldest key this device holds. */ - internal suspend fun epochKey(epoch: Int): ByteArray { - while (epoch !in keys) { - val (known, key) = keys.firstEntry() ?: reject(ProtocolError.KEY_COMMITMENT) - if (known <= epoch) reject(ProtocolError.KEY_COMMITMENT) - val knownCommitment = commitments[known] ?: reject(ProtocolError.KEY_COMMITMENT) - val kc = kcName(knownCommitment) - val chain = store.key(known, kc, "chain") ?: reject(ProtocolError.KEY_COMMITMENT) - val below = commitments[known - 1] ?: reject(ProtocolError.KEY_COMMITMENT) - keys[known - 1] = - forgettingIfBad(known, kc, "chain") { - openChain(key, group, known, chain).also { checkEpochKey(it, group, known - 1, below) } - } - } - return keys.getValue(epoch) - } + internal suspend fun epochKey(epoch: Int): ByteArray = keys.key(epoch) /** * Opens something sealed under [epoch] with the chain's key, or else with the key of an entry that @@ -301,7 +280,7 @@ class Session private constructor( val seq = record.ownSeq + 1 val epoch = state.epoch val header = Header(type, group, epoch, device.did, seq, SyncBytes.fromHex(record.ownPrev)) - val key = purposeKey(keys.getValue(epoch), group, epoch, type.purpose) + val key = purposeKey(sealingKey(), group, epoch, type.purpose) val sealed = sealObject(header, key, encodePayload(content), device.sigSeed) store.putObject(StorePaths.log(me, seq), sealed) record = record.copy(ownSeq = seq, ownPrev = SyncBytes.toHex(sha256(sealed))) @@ -368,6 +347,9 @@ class Session private constructor( return records } + /** How far every other device's log can be read now, with nothing saved. */ + internal suspend fun readablePositions(): Map = read(links = null).positions + /** Batches of [links] only, or every batch when [links] is null. */ private suspend fun read(links: Set?): Pulled { fun wanted(link: String?) = links == null || link in links diff --git a/app/src/main/java/io/github/aedev/flow/sync2/engine/Snapshots.kt b/app/src/main/java/io/github/aedev/flow/sync2/engine/Snapshots.kt index 7fe1ff8e1..5f8e9dcbe 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/engine/Snapshots.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/engine/Snapshots.kt @@ -89,7 +89,7 @@ suspend fun Session.writeSnapshot( ) { if (state.recoveryDue) return val epoch = state.epoch - val key = purposeKey(currentKey(), group, epoch, ObjectType.SNAPSHOT.purpose) + val key = purposeKey(sealingKey(), group, epoch, ObjectType.SNAPSHOT.purpose) val parts = exported.flatMap { (collection, wire) -> chunks(wire.lines).map { lines -> diff --git a/app/src/main/java/io/github/aedev/flow/sync2/linking/DeviceNames.kt b/app/src/main/java/io/github/aedev/flow/sync2/linking/DeviceNames.kt index 50ee1ab7a..adcf3bf83 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/linking/DeviceNames.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/linking/DeviceNames.kt @@ -4,6 +4,7 @@ import io.github.aedev.flow.sync.crypto.SyncBytes import io.github.aedev.flow.sync2.crypto.ProtocolException import io.github.aedev.flow.sync2.engine.LinkedException import io.github.aedev.flow.sync2.engine.Session +import io.github.aedev.flow.sync2.roster.Entry import io.github.aedev.flow.sync2.roster.RosterState import io.github.aedev.flow.sync2.roster.applyEntry import io.github.aedev.flow.sync2.roster.openDeviceMeta @@ -12,13 +13,19 @@ import io.github.aedev.flow.sync2.roster.openDeviceMeta * Device names and platforms, as the latest roster entry naming each device says. Names are only * shown, so one that can't be read is left out instead of stopping sync. */ -suspend fun deviceNames(session: Session): Map { +suspend fun deviceNames(session: Session): Map = deviceNames(session.entries, session.group, session::epochKey) + +internal suspend fun deviceNames( + entries: List, + group: ByteArray, + epochKey: suspend (Int) -> ByteArray, +): Map { val names = HashMap() var state: RosterState? = null - for (entry in session.entries) { + for (entry in entries) { val after = applyEntry(state, entry) for ((did, sealed) in entry.meta.orEmpty()) { - openName(session, after.epoch, entry.seq, did, sealed)?.let { names[did] = it } + openName(group, epochKey, after.epoch, entry.seq, did, sealed)?.let { names[did] = it } } state = after } @@ -26,7 +33,8 @@ suspend fun deviceNames(session: Session): Map { } private suspend fun openName( - session: Session, + group: ByteArray, + epochKey: suspend (Int) -> ByteArray, epoch: Int, seq: Long, did: String, @@ -35,8 +43,8 @@ private suspend fun openName( try { val content = openDeviceMeta( - session.epochKey(epoch), - session.group, + epochKey(epoch), + group, epoch, seq, SyncBytes.fromHex(did), diff --git a/app/src/main/java/io/github/aedev/flow/sync2/linking/Membership.kt b/app/src/main/java/io/github/aedev/flow/sync2/linking/Membership.kt index 207ec87ed..507cbd059 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/linking/Membership.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/linking/Membership.kt @@ -24,6 +24,7 @@ import io.github.aedev.flow.sync2.roster.memberSubject import io.github.aedev.flow.sync2.roster.recoveryMember import io.github.aedev.flow.sync2.roster.sealDeviceMeta import io.github.aedev.flow.sync2.store.Appended +import io.github.aedev.flow.sync2.store.StreamPosition /** A fresh epoch key, wrapped to every member after a change, with the previous key chained. */ class NewEpoch( @@ -61,19 +62,24 @@ suspend fun appendChecked( if (session.store.appendRoster(session.state.head, entry, wraps, chain) != Appended.Done) reject(ProtocolError.ROSTER_CHAIN) } -/** Removes [removed] and starts a new epoch it never receives. The store also refuses it at once. */ +/** + * Removes [removed] and starts a new epoch it never receives. The store also refuses it at once. + * Each log's `cut` is what this device saved reading it, unless a caller that saves nothing yet gives [read]. + */ suspend fun removeDevices( session: Session, device: MemberSecrets, removed: List, + read: Map? = null, ): String? { val state = session.state val positions = - session.db - .groups() - .streams() - .associateBy { it.did } - val cut = removed.mapNotNull { did -> positions[did]?.let { did to Cut(it.seq, it.hash) } }.toMap() + read?.mapValues { (_, at) -> Cut(at.seq, SyncBytes.toHex(at.hash)) } + ?: session.db + .groups() + .streams() + .associate { it.did to Cut(it.seq, it.hash) } + val cut = removed.mapNotNull { did -> positions[did]?.takeIf { it.seq > 0 }?.let { did to it } }.toMap() val replacing = installerGone(state.members, removed) val secret = SyncCrypto.randomBytes(32) val next = recoveryKeys(secret) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt b/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt index 6eee4cad6..8f08a260f 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt @@ -14,6 +14,8 @@ import io.github.aedev.flow.sync2.data.GroupRecord import io.github.aedev.flow.sync2.data.LinkedDatabase import io.github.aedev.flow.sync2.data.WEBDAV_STORE import io.github.aedev.flow.sync2.data.addEntries +import io.github.aedev.flow.sync2.data.keptDevices +import io.github.aedev.flow.sync2.engine.EpochKeys import io.github.aedev.flow.sync2.engine.LinkedException import io.github.aedev.flow.sync2.engine.Session import io.github.aedev.flow.sync2.roster.Draft @@ -51,6 +53,94 @@ private suspend fun append( return after } +/** A current device a restore can keep or sign out; [name] is empty when it can't be read. */ +data class RestoreDevice( + val did: String, + val name: String, + val platform: String, +) + +/** The roster a store shows the recovery member, and the current key it opens. */ +private class RecoveryView( + val entries: List, + val state: RosterState, + val group: ByteArray, + val key: ByteArray, +) + +private suspend fun readAsRecovery( + asRecovery: Store, + recovery: MemberSecrets, +): RecoveryView { + val entries = asRecovery.rosterAfter(-1) + val state = replayRoster(entries) + val recoveryId = SyncBytes.toHex(recovery.did) + if (state.members[recoveryId]?.used != false) throw LinkedException.RecoveryUsed() + val group = SyncBytes.fromHex(state.group) + val commitment = SyncBytes.b64urlDecode(state.kc) + val wrap = asRecovery.key(state.epoch, kcName(commitment), recoveryId) ?: reject(ProtocolError.KEY_COMMITMENT) + val current = unwrapEpochKey(recovery.kemSecret, group, state.epoch, recovery.did, wrap) + checkEpochKey(current, group, state.epoch, commitment) + return RecoveryView(entries, state, group, current) +} + +/** The devices a restore asks about (spec/protocol.md, Recovery, step 3). It writes nothing. */ +suspend fun previewRestore( + asRecovery: Store, + recovery: MemberSecrets, +): List { + val view = readAsRecovery(asRecovery, recovery) + val commitments = view.entries.mapNotNull { it.epoch }.associate { it.n to SyncBytes.b64urlDecode(it.kc) } + val keys = EpochKeys(asRecovery, view.group, commitments).also { it[view.state.epoch] = view.key } + val names = deviceNames(view.entries, view.group, keys::key) + return view.state.members + .filterValues { it.role == Role.DEVICE } + .keys + .sorted() + .map { did -> RestoreDevice(did, names[did]?.name.orEmpty(), names[did]?.platform.orEmpty()) } +} + +/** The devices a restore has yet to sign out: all current ones but this one and those the user kept. */ +internal fun Session.unchosenDevices(): List { + val keep = record.keptDevices ?: return emptyList() + return state.members + .filter { (did, member) -> member.role == Role.DEVICE && did != me && did !in keep } + .keys + .sorted() +} + +/** Where a restore's last step left things: the session to go on with, and new words to show when the removal had to replace them. */ +class FinishedRestore( + val session: Session, + val words: String?, +) + +/** + * Signs out the devices the user didn't choose (spec/protocol.md, Recovery, step 6), having read every log without + * writing, so that nothing is sealed under a key they hold. [reread] opens the roster again from the store. A removal + * that lost a fork stays due for the next sync, and so does one that dies on the way: it starts over from the roster. + */ +suspend fun finishRestore( + session: Session, + device: MemberSecrets, + reread: suspend () -> Session, +): FinishedRestore { + var current = session + var words: String? = null + val unchosen = current.unchosenDevices() + if (unchosen.isNotEmpty()) { + words = removeDevices(current, device, unchosen, current.readablePositions()) + if (words != null) current.db.groups().setRecoverySaved(false) + current.db.snapshots().snapshotSoon() + current = reread() + } + if (current.unchosenDevices().isEmpty()) { + current.db.groups().forgetRestoreKeep() + current = reread() + } + return FinishedRestore(current, words) +} + /** * Restores a group with the recovery key (spec/protocol.md, Recovery): the recovery member adds * [device] and starts an epoch, then [device] replaces the used key straight away. [asRecovery] is @@ -63,14 +153,11 @@ suspend fun restore( device: MemberSecrets, meta: DeviceMeta, ): Restored { - var state = replayRoster(asRecovery.rosterAfter(-1)) + val view = readAsRecovery(asRecovery, recovery) + var state = view.state val recoveryId = SyncBytes.toHex(recovery.did) - if (state.members[recoveryId]?.used != false) throw LinkedException.RecoveryUsed() - val group = SyncBytes.fromHex(state.group) - val commitment = SyncBytes.b64urlDecode(state.kc) - val wrap = asRecovery.key(state.epoch, kcName(commitment), recoveryId) ?: reject(ProtocolError.KEY_COMMITMENT) - val current = unwrapEpochKey(recovery.kemSecret, group, state.epoch, recovery.did, wrap) - checkEpochKey(current, group, state.epoch, commitment) + val group = view.group + val current = view.key val kems = state.members.mapValues { (_, member) -> member.kem } val restored = newEpoch(group, state.epoch + 1, kems + (SyncBytes.toHex(device.did) to device.kemPublic), current) @@ -138,9 +225,31 @@ suspend fun replaceRecovery( return replacement.words } +/** The devices of the group on the Flow server at [storeUrl], for the user to choose from. */ +suspend fun previewServerRestore( + storeUrl: String, + words: String, +): List = previewOn(words) { ServerStore.connect(storeUrl, it, null, null) } + +/** The devices of the group in a WebDAV folder, for the user to choose from. */ +suspend fun previewWebDavRestore( + files: WebDavStore, + words: String, +): List = previewOn(words) { FileLayout(files, SyncBytes.toHex(it.did)) } + +private suspend fun previewOn( + words: String, + store: (recovery: MemberSecrets) -> Store, +): List { + val secret = recoverySecret(words) ?: throw LinkedException.WrongWords() + val recovery = recoveryKeys(secret) + return previewRestore(store(recovery), recovery).also { recovery.wipe() } +} + /** - * Restores onto this device from the recovery words. Its profile link stays open until the first - * sync reads every log ([io.github.aedev.flow.sync2.service.LinkedProfiles.links]). + * Restores onto this device from the recovery words, keeping the devices in [keep]. Its profile link stays open until + * the first sync reads every log ([io.github.aedev.flow.sync2.service.LinkedProfiles.links]), which also signs out + * every other device ([finishRestore]). */ suspend fun restoreGroup( db: LinkedDatabase, @@ -148,10 +257,11 @@ suspend fun restoreGroup( words: String, device: MemberSecrets, meta: DeviceMeta, + keep: Set, ): String { val asDevice = ServerStore.connect(storeUrl, device, null, null) var asRecovery: ServerStore? = null - return restoreOn(db, words, device, meta, { recovery -> + return restoreOn(db, words, device, meta, keep, { recovery -> ServerStore.connect(storeUrl, recovery, null, null).also { asRecovery = it } to asDevice }) { @@ -168,9 +278,10 @@ suspend fun restoreWebDavGroup( words: String, device: MemberSecrets, meta: DeviceMeta, + keep: Set, ): String { val layout = FileLayout(files, SyncBytes.toHex(device.did)) - return restoreOn(db, words, device, meta, { layout to layout }) { state -> + return restoreOn(db, words, device, meta, keep, { layout to layout }) { state -> GroupRecord(state.group, WEBDAV_STORE + files.location, null, link = "") } } @@ -181,6 +292,7 @@ private suspend fun restoreOn( words: String, device: MemberSecrets, meta: DeviceMeta, + keep: Set, stores: (recovery: MemberSecrets) -> Pair, record: suspend (RosterState) -> GroupRecord, ): String { @@ -191,6 +303,6 @@ private suspend fun restoreOn( recovery.wipe() val saved = record(replayRoster(restored.entries)) db.groups().addEntries(restored.entries) - db.groups().saveGroup(saved) + db.groups().saveGroup(saved.copy(restoreKeep = keep.sorted().joinToString(","))) return restored.words } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveChannel.kt b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveChannel.kt index 537ff8c00..b98dab3e9 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveChannel.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveChannel.kt @@ -66,10 +66,10 @@ class LiveChannel return { transports.update { it - transport } } } - /** Sends [message] to every reachable device; returns its counter, or null when this device isn't linked. */ + /** Sends [message] to every reachable device; returns its counter, or null when this device isn't linked or a restore is unfinished. */ suspend fun send(message: LiveMessage): Long? = lock.withLock { - val session = session() ?: return null + val session = session()?.takeUnless { it.removalPending } ?: return null val seq = (db.live().counter(session.me) ?: 0) + 1 // Saved before it leaves, so a crash can't make a counter repeat. db.live().setCounter(LiveCounterRow(session.me, seq)) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt index 503a00171..e7bf71642 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/live/LiveEnvelopes.kt @@ -27,7 +27,7 @@ internal fun Session.sealLive( ): ByteArray { val epoch = state.epoch val header = Header(ObjectType.LIVE, group, epoch, device.did, seq, ByteArray(32)) - val key = purposeKey(currentKey(), group, epoch, Purpose.LIVE) + val key = purposeKey(sealingKey(), group, epoch, Purpose.LIVE) return sealObject(header, key, encodePayload(LiveMessage.encode(message)), device.sigSeed) } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedActions.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedActions.kt index 76da361f7..43ad5bd66 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedActions.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedActions.kt @@ -24,6 +24,7 @@ import io.github.aedev.flow.sync2.lan.LanSync import io.github.aedev.flow.sync2.lan.lanPort import io.github.aedev.flow.sync2.linking.Code import io.github.aedev.flow.sync2.linking.DeviceMeta +import io.github.aedev.flow.sync2.linking.RestoreDevice import io.github.aedev.flow.sync2.linking.StoreLocation import io.github.aedev.flow.sync2.linking.addDevice import io.github.aedev.flow.sync2.linking.addScanned @@ -36,6 +37,8 @@ import io.github.aedev.flow.sync2.linking.deviceNames import io.github.aedev.flow.sync2.linking.join import io.github.aedev.flow.sync2.linking.listen import io.github.aedev.flow.sync2.linking.newOffer +import io.github.aedev.flow.sync2.linking.previewServerRestore +import io.github.aedev.flow.sync2.linking.previewWebDavRestore import io.github.aedev.flow.sync2.linking.readCode import io.github.aedev.flow.sync2.linking.removeDevices import io.github.aedev.flow.sync2.linking.renameDevice @@ -134,8 +137,12 @@ class LinkedActions private suspend fun linked(): GroupRecord = db.groups().group() ?: throw LinkedException.NotLinked() /** Fresh device keys for a group this device is about to create, join or restore (spec/crypto.md, Device keys). */ - private suspend fun newMembership(): MemberSecrets { + private suspend fun checkNotLinked() { if (db.groups().group() != null) throw LinkedException.AlreadyLinked() + } + + private suspend fun newMembership(): MemberSecrets { + checkNotLinked() return keys.fresh() } @@ -211,20 +218,42 @@ class LinkedActions linkedNow() } - /** Restores the group with the recovery words and returns the new words to show once. */ + /** The devices of the group the recovery words open, for the user to choose from. Writes nothing. */ + suspend fun previewRestore( + url: String, + words: String, + ): List { + checkNotLinked() + return previewServerRestore(url.trim(), words) + } + + /** The devices of the group in a WebDAV folder that the recovery words open. Writes nothing. */ + suspend fun previewRestoreFromWebDav( + url: String, + user: String, + password: String, + words: String, + ): List { + checkNotLinked() + return previewWebDavRestore(WebDavStore.connect(url.trim(), user, password), words) + } + + /** Restores the group with the recovery words, keeping the devices in [keep], and returns the new words to show once. */ suspend fun restore( url: String, words: String, - ): String = restoreGroup(db, url.trim(), words, newMembership(), meta()).also { linkedNow() } + keep: Set, + ): String = restoreGroup(db, url.trim(), words, newMembership(), meta(), keep).also { linkedNow() } - /** Restores the group from a WebDAV folder with the recovery words and returns the new words to show once. */ + /** Restores the group from a WebDAV folder with the recovery words, keeping the devices in [keep], and returns the new words to show once. */ suspend fun restoreFromWebDav( url: String, user: String, password: String, words: String, + keep: Set, ): String { - val restored = restoreWebDavGroup(db, WebDavStore.connect(url.trim(), user, password), words, newMembership(), meta()) + val restored = restoreWebDavGroup(db, WebDavStore.connect(url.trim(), user, password), words, newMembership(), meta(), keep) credentials.save(WebDavLogin(user, password)) return restored.also { linkedNow() } } @@ -265,7 +294,10 @@ class LinkedActions /** The devices yet to read the login this device sent, or null with no change under way. */ suspend fun loginWaiting(): List? = loginWaiting(db, SyncBytes.toHex(keys.load().did)) - suspend fun confirmRecoverySaved() = db.groups().setRecoverySaved(true) + suspend fun confirmRecoverySaved() { + db.groups().setRecoverySaved(true) + sync.replacedWords.value = null + } suspend fun rename( id: String, @@ -329,6 +361,8 @@ class LinkedActions val syncing: StateFlow = sync.syncing + val replacedWords: StateFlow = sync.replacedWords + fun syncNow() = LinkedSyncWorker.syncSoon(context, announce = true) /** Shows a member's code, answered through the server's mailbox or over the LAN. */ diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt index 80c3019a7..3b7bbe6e1 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedError.kt @@ -28,6 +28,7 @@ enum class LinkedError( ROLLED_BACK(retryable = true), RECOVERY_DUE(retryable = false), ROSTER_CHANGED(retryable = true), + RESTORE_UNFINISHED(retryable = true), GROUP_OUTDATED, DECLINED, TIMED_OUT, @@ -108,6 +109,10 @@ fun Throwable.linkedError(): LinkedError = LinkedError.RECOVERY_DUE } + is LinkedException.RestoreUnfinished -> { + LinkedError.RESTORE_UNFINISHED + } + is LinkedException.RosterRolledBack -> { LinkedError.ROLLED_BACK } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt index ce6e3476e..b11fa9a29 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt @@ -32,6 +32,7 @@ import io.github.aedev.flow.sync2.engine.isDue import io.github.aedev.flow.sync2.engine.writeSnapshot import io.github.aedev.flow.sync2.linking.DeviceMeta import io.github.aedev.flow.sync2.linking.deviceNames +import io.github.aedev.flow.sync2.linking.finishRestore import io.github.aedev.flow.sync2.linking.rotateEpoch import io.github.aedev.flow.sync2.store.FileLayout import io.github.aedev.flow.sync2.store.GroupStore @@ -96,6 +97,9 @@ class LinkedSync val lastSync: StateFlow = last.asStateFlow() + /** New recovery words a restore's removal had to make, to show once. */ + val replacedWords = MutableStateFlow(null) + val syncing: StateFlow = active.asStateFlow() /** Each device a sync the user asked for applied changes from. */ @@ -223,6 +227,18 @@ class LinkedSync } val reading = if (offline == null) store else GroupStore(local, null) var session = Session.open(db, reading, device) + if (session.removalPending) { + // It needs the store to take the removal; what it read stays for this sync once that's done. + if (offline != null) throw offline + val finished = + finishRestore(session, device) { + store.download(me, record.serverCursor, objects = false) + Session.open(db, reading, device) + } + finished.words?.let { replacedWords.value = it } + session = finished.session + if (session.removalPending) return + } val links = profiles.links(session) ?: return // A device left: start the epoch it never receives before writing anything, then read the roster back (spec/protocol.md, Leaving). if (session.state.rotate && !session.state.recoveryDue && offline == null) { diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt index 0919e1adc..e23ce5517 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/Store.kt @@ -47,6 +47,13 @@ interface Store { /** Deletes an object; one that is already gone counts as deleted. */ suspend fun deleteObject(path: String) + + /** Lets go of a held wrap or chain box that failed its checks, so the next sync asks for it again. */ + suspend fun forgetKey( + epoch: Int, + kc: String, + name: String, + ) = Unit } /** A group's shared store: a Flow server, or files through [FileLayout]. Both carry the pairing mailbox. */ @@ -58,13 +65,6 @@ interface Remote : interface ChangeList : Store { /** Objects that arrived after [cursor], and the cursor to pass next time. */ suspend fun changes(cursor: Long): Pair, Long> - - /** Lets go of a held wrap or chain box that failed its checks, so the next sync asks for it again. */ - suspend fun forgetKey( - epoch: Int, - kc: String, - name: String, - ) = Unit } /** Where a wrap or chain box lives under `keys/`: its epoch, its ``, and a member's id or `chain`. */ diff --git a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesSection.kt b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesSection.kt index 8be7602be..f243bfd6a 100644 --- a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesSection.kt +++ b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesSection.kt @@ -119,6 +119,9 @@ internal fun LinkedDevicesSection( onShowCode = viewModel::showCode, onRestore = viewModel::restore, onRestoreFromWebDav = viewModel::restoreFromWebDav, + restoreDevices = state.restoreDevices, + onConfirmRestore = viewModel::confirmRestore, + onCancelRestore = viewModel::cancelRestore, ) } } diff --git a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesViewModel.kt b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesViewModel.kt index 24355496c..a37924a7a 100644 --- a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesViewModel.kt +++ b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedDevicesViewModel.kt @@ -7,6 +7,7 @@ import dagger.hilt.android.lifecycle.HiltViewModel import dagger.hilt.android.qualifiers.ApplicationContext import io.github.aedev.flow.sync2.device.WebDavLogin import io.github.aedev.flow.sync2.engine.ContestSide +import io.github.aedev.flow.sync2.linking.RestoreDevice import io.github.aedev.flow.sync2.live.LiveConsent import io.github.aedev.flow.sync2.live.LiveDevices import io.github.aedev.flow.sync2.live.deviceKind @@ -26,6 +27,7 @@ import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.combine +import kotlinx.coroutines.flow.filterNotNull import kotlinx.coroutines.flow.stateIn import kotlinx.coroutines.flow.update import kotlinx.coroutines.launch @@ -45,6 +47,8 @@ internal data class LinkedUiState( val kinds: Map = emptyMap(), /** Recovery words to show once, until the user proves they wrote them down. */ val words: String? = null, + /** The group's devices while the user chooses which a restore keeps. */ + val restoreDevices: List? = null, val busy: Boolean = false, val syncing: Boolean = false, val error: LinkedError? = null, @@ -63,6 +67,12 @@ internal data class LinkedUiState( val linked: Boolean get() = storeUrl != null } +/** A restore waiting for the user's choice: the group's [devices], and how to go on with the ones to keep. */ +private class PendingRestore( + val devices: List, + val run: suspend (keep: Set) -> String, +) + /** What the screen itself holds: the words being shown and the outcome of the last action. */ private data class Local( val devices: List? = null, @@ -70,6 +80,7 @@ private data class Local( /** Devices met on this network while the app runs. */ val nearby: Set = emptySet(), val words: String? = null, + val restore: PendingRestore? = null, val busy: Boolean = false, val error: LinkedError? = null, val contest: List = emptyList(), @@ -101,6 +112,12 @@ internal class LinkedDevicesViewModel private val local = MutableStateFlow(Local()) + init { + viewModelScope.launch { + actions.replacedWords.filterNotNull().collect { words -> local.update { it.copy(words = words) } } + } + } + val state: StateFlow = combine( actions.group, @@ -120,6 +137,7 @@ internal class LinkedDevicesViewModel nearby, if (own != null) kinds + (own to ownKind) else kinds, l.words, + l.restore?.devices, l.busy, syncing, l.error ?: lastSync?.error, @@ -197,7 +215,12 @@ internal class LinkedDevicesViewModel user: String, password: String, words: String, - ) = run { local.update { it.copy(words = actions.restoreFromWebDav(url, user, password, words)) } } + ) = run { + val devices = actions.previewRestoreFromWebDav(url, user, password, words) + local.update { state -> + state.copy(restore = PendingRestore(devices) { actions.restoreFromWebDav(url, user, password, words, it) }) + } + } fun updateLogin( user: String, @@ -231,7 +254,20 @@ internal class LinkedDevicesViewModel fun restore( url: String, words: String, - ) = run { local.update { it.copy(words = actions.restore(url, words)) } } + ) = run { + val devices = actions.previewRestore(url, words) + local.update { state -> state.copy(restore = PendingRestore(devices) { actions.restore(url, words, it) }) } + } + + /** Goes on with the restore the user was asked about, keeping [keep]. */ + fun confirmRestore(keep: Set) = + run { + val pending = local.value.restore ?: return@run + val words = pending.run(keep) + local.update { it.copy(words = words, restore = null) } + } + + fun cancelRestore() = local.update { it.copy(restore = null, error = null) } fun confirmWordsSaved() = run { diff --git a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedErrors.kt b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedErrors.kt index ba4fb2539..c5404cb2e 100644 --- a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedErrors.kt +++ b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedErrors.kt @@ -27,6 +27,7 @@ internal fun LinkedError.message(): String = LinkedError.RECOVERY_DUE -> R.string.linked_error_recovery_due LinkedError.ROLLED_BACK -> R.string.linked_error_rolled_back LinkedError.ROSTER_CHANGED -> R.string.linked_error_roster_changed + LinkedError.RESTORE_UNFINISHED -> R.string.linked_error_restore_unfinished LinkedError.GROUP_OUTDATED -> R.string.linked_error_group_outdated LinkedError.DECLINED -> R.string.linked_error_declined LinkedError.TIMED_OUT -> R.string.linked_error_timed_out diff --git a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetup.kt b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetup.kt index 10220682c..d8fbd9a1b 100644 --- a/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetup.kt +++ b/app/src/main/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetup.kt @@ -1,5 +1,6 @@ package io.github.aedev.flow.ui.screens.sync.linked +import androidx.compose.foundation.layout.Row import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.padding import androidx.compose.foundation.text.KeyboardOptions @@ -14,6 +15,7 @@ import androidx.compose.material.icons.outlined.QrCode2 import androidx.compose.material.icons.outlined.QrCodeScanner import androidx.compose.material.icons.outlined.Wifi import androidx.compose.material3.AlertDialog +import androidx.compose.material3.Checkbox import androidx.compose.material3.MaterialTheme import androidx.compose.material3.OutlinedTextField import androidx.compose.material3.Text @@ -24,6 +26,7 @@ import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember import androidx.compose.runtime.saveable.rememberSaveable import androidx.compose.runtime.setValue +import androidx.compose.ui.Alignment import androidx.compose.ui.Modifier import androidx.compose.ui.res.stringResource import androidx.compose.ui.text.input.KeyboardType @@ -31,6 +34,7 @@ import androidx.compose.ui.text.input.PasswordVisualTransformation import androidx.compose.ui.unit.dp import io.github.aedev.flow.R import io.github.aedev.flow.sync2.device.WebDavLogin +import io.github.aedev.flow.sync2.linking.RestoreDevice import io.github.aedev.flow.sync2.service.LinkedError import io.github.aedev.flow.ui.screens.sync.SyncActionRow import io.github.aedev.flow.ui.screens.sync.SyncCard @@ -74,6 +78,10 @@ internal fun LinkedSetup( onShowCode: () -> Unit, onRestore: (url: String, words: String) -> Unit, onRestoreFromWebDav: (url: String, user: String, password: String, words: String) -> Unit, + /** The group's devices once the words open it, while the user chooses which to keep. */ + restoreDevices: List?, + onConfirmRestore: (keep: Set) -> Unit, + onCancelRestore: () -> Unit, ) { var setup by rememberSaveable { mutableStateOf(Setup.CHOOSE) } // A WebDAV login entered before scanning; kept in memory only, never in saved state. @@ -85,6 +93,10 @@ internal fun LinkedSetup( Text(error.message(), Modifier.padding(16.dp), color = MaterialTheme.colorScheme.onErrorContainer) } } + if (restoreDevices != null) { + RestoreChoice(restoreDevices, busy, onConfirmRestore, onCancelRestore) + return + } when (setup) { Setup.CHOOSE -> { SyncStepHeader( @@ -265,6 +277,38 @@ internal fun LinkedSetup( } } +/** Which of the group's [devices] the user still has; a restore signs out the rest, so none is ticked to begin with. */ +@Composable +private fun RestoreChoice( + devices: List, + busy: Boolean, + onConfirm: (keep: Set) -> Unit, + onBack: () -> Unit, +) { + var keep by rememberSaveable { mutableStateOf(emptyList()) } + SyncStepHeader( + icon = Icons.Outlined.Key, + title = stringResource(R.string.linked_restore_choose_title), + body = stringResource(R.string.linked_restore_choose_body), + ) + for (device in devices) { + Row(verticalAlignment = Alignment.CenterVertically) { + Checkbox( + checked = device.did in keep, + onCheckedChange = { keep = if (it) keep + device.did else keep - device.did }, + ) + Text(device.name.ifEmpty { stringResource(R.string.linked_restore_unknown_device, device.did.take(8)) }) + } + } + SyncActionRow( + confirmLabel = stringResource(R.string.linked_restore), + onConfirm = { onConfirm(keep.toSet()) }, + confirmEnabled = !busy, + dismissLabel = stringResource(R.string.linked_back), + onDismiss = onBack, + ) +} + /** * A server that already holds a group, which nobody here can join: join it from a device still in it, * or reset the server, after a warning, and create the new group there. diff --git a/app/src/main/res/values/strings.xml b/app/src/main/res/values/strings.xml index 6f5e3f253..239a0e56a 100644 --- a/app/src/main/res/values/strings.xml +++ b/app/src/main/res/values/strings.xml @@ -2798,6 +2798,9 @@ Recovery words Create Restore + Which devices do you still have? + Restoring signs out every device you don\'t tick. Tick one only if you have it with you, not just because its name looks right. If you still have a working device, go back and link this one from it instead. + Unknown device (%1$s) Back Scan the code on a linked device Server @@ -2826,6 +2829,7 @@ It wasn\'t me Make new recovery words The device that set your recovery key has left the group. Until you make new words on one of your devices, this one receives changes but doesn\'t send any. + The restore isn\'t finished yet. Flow will finish it on the next sync. Make new recovery words first. The device that set them has left the group. This device set your current recovery words, so your other devices will stop sending changes until you make new words on one of them. %1$s added these devices. Tick any you don\'t recognise to remove them too. diff --git a/app/src/test/java/io/github/aedev/flow/sync2/LinkedErrorTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/LinkedErrorTest.kt index a9535817c..30660c305 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/LinkedErrorTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/LinkedErrorTest.kt @@ -25,6 +25,7 @@ class LinkedErrorTest { assertEquals(LinkedError.REFUSED, StoreException.Refused(401, "unauthorized").linkedError()) assertEquals(LinkedError.GROUP_MISSING, StoreException.Refused(503, "unclaimed").linkedError()) assertEquals(LinkedError.STOPPED, LinkedException.Stopped().linkedError()) + assertEquals(LinkedError.RESTORE_UNFINISHED, LinkedException.RestoreUnfinished().linkedError()) } @Test diff --git a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt new file mode 100644 index 000000000..8918885b0 --- /dev/null +++ b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt @@ -0,0 +1,421 @@ +package io.github.aedev.flow.sync2 + +import android.app.Application +import android.content.Context +import androidx.test.core.app.ApplicationProvider +import androidx.test.ext.junit.runners.AndroidJUnit4 +import io.github.aedev.flow.sync.crypto.SyncBytes +import io.github.aedev.flow.sync.protocol.CollectionWire +import io.github.aedev.flow.sync.protocol.SyncCollection +import io.github.aedev.flow.sync2.crypto.MemberSecrets +import io.github.aedev.flow.sync2.crypto.ProtocolException +import io.github.aedev.flow.sync2.crypto.purposeKey +import io.github.aedev.flow.sync2.crypto.sha256 +import io.github.aedev.flow.sync2.crypto.unwrapEpochKey +import io.github.aedev.flow.sync2.data.GroupRecord +import io.github.aedev.flow.sync2.data.LinkedDatabase +import io.github.aedev.flow.sync2.data.addEntries +import io.github.aedev.flow.sync2.data.keptDevices +import io.github.aedev.flow.sync2.device.DeviceKeys +import io.github.aedev.flow.sync2.device.StoreCredentials +import io.github.aedev.flow.sync2.device.WebDavLogin +import io.github.aedev.flow.sync2.engine.LinkedException +import io.github.aedev.flow.sync2.engine.Session +import io.github.aedev.flow.sync2.engine.writeSnapshot +import io.github.aedev.flow.sync2.linking.DeviceMeta +import io.github.aedev.flow.sync2.linking.createWebDavGroup +import io.github.aedev.flow.sync2.linking.finishRestore +import io.github.aedev.flow.sync2.linking.previewRestore +import io.github.aedev.flow.sync2.linking.previewWebDavRestore +import io.github.aedev.flow.sync2.linking.removeDevices +import io.github.aedev.flow.sync2.linking.renameDevice +import io.github.aedev.flow.sync2.linking.replaceRecovery +import io.github.aedev.flow.sync2.linking.restore +import io.github.aedev.flow.sync2.linking.restoreWebDavGroup +import io.github.aedev.flow.sync2.linking.rotateEpoch +import io.github.aedev.flow.sync2.live.LiveChannel +import io.github.aedev.flow.sync2.live.LiveMessage +import io.github.aedev.flow.sync2.live.sealLive +import io.github.aedev.flow.sync2.roster.Op +import io.github.aedev.flow.sync2.roster.Role +import io.github.aedev.flow.sync2.roster.buildEntry +import io.github.aedev.flow.sync2.roster.memberSubject +import io.github.aedev.flow.sync2.service.LinkedProfiles +import io.github.aedev.flow.sync2.service.LinkedReset +import io.github.aedev.flow.sync2.service.LinkedSync +import io.github.aedev.flow.sync2.service.SyncIndicator +import io.github.aedev.flow.sync2.store.FileLayout +import io.github.aedev.flow.sync2.store.GroupStore +import io.github.aedev.flow.sync2.store.KeyName +import io.github.aedev.flow.sync2.store.LocalStore +import io.github.aedev.flow.sync2.store.ObjectType +import io.github.aedev.flow.sync2.store.StoreLogin +import io.github.aedev.flow.sync2.store.StorePaths +import io.github.aedev.flow.sync2.store.kcName +import io.github.aedev.flow.sync2.store.parseObject +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.TestScope +import kotlinx.coroutines.test.runTest +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNotNull +import org.junit.Assert.assertNull +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.annotation.Config +import java.io.IOException + +/** A restore keeps only the devices the user chooses, and seals nothing before the others are signed out (spec/protocol.md, Recovery). */ +@RunWith(AndroidJUnit4::class) +@Config(sdk = [34], application = Application::class) +class RestoreChoiceTest { + private val context: Context = ApplicationProvider.getApplicationContext() + private val dav = WebDavStandIn() + private val export = mapOf(SyncCollection.WATCH_HISTORY to CollectionWire(listOf("""{"title":"t","videoId":"aaaaaaaaaaa"}"""), 1, "")) + private val now = LiveMessage.Now(video = "aaaaaaaaaaa", title = "t", state = LiveMessage.PLAYING) + + @After + fun stop() = dav.close() + + private fun id(member: MemberSecrets) = SyncBytes.toHex(member.did) + + /** A device restored on [store], as `restoreGroup` leaves it: the choice saved, the others not signed out yet. */ + private suspend fun TestScope.restoredKeeping( + g: LinkedGroup, + store: MemoryStore = g.store, + keep: Set = emptySet(), + ): Pair { + val device = newMember() + val restored = restore(store, store, g.recovery, device, RESTORED) + val db = emptyDb() + db.groups().addEntries(restored.entries) + db.groups().saveGroup( + GroupRecord(SyncBytes.toHex(g.group), "memory", null, link = "", restoreKeep = keep.sorted().joinToString(",")), + ) + return device to db + } + + private suspend fun finish( + store: MemoryStore, + db: LinkedDatabase, + device: MemberSecrets, + ) = finishRestore(Session.open(db, store, device), device) { Session.open(db, store, device) } + + private fun MemoryStore.objectsOf(device: MemberSecrets) = objects.keys.filter { it.startsWith("dev/${id(device)}/") } + + @Test + fun the_devices_are_listed_by_name_and_nothing_is_written() = + runTest { + val g = LinkedGroup(this).apply { create() } + val dbA = g.deviceDb() + renameDevice(Session.open(dbA, g.store, g.a), g.a, id(g.a), DeviceMeta("app-a", "Phone", "android")) + // The name is sealed under an older epoch, so reading it walks the history chain. + rotateEpoch(Session.open(dbA, g.store, g.a), g.a) + val entries = g.store.roster.size + val keys = g.store.keys.size + + val devices = previewRestore(g.store, g.recovery) + + assertEquals(mapOf(id(g.a) to "Phone", id(g.b) to ""), devices.associate { it.did to it.name }) + assertEquals(entries, g.store.roster.size) + assertEquals(keys, g.store.keys.size) + } + + @Test + fun used_words_are_refused_before_the_question() = + runTest { + val g = LinkedGroup(this).apply { create() } + restore(g.store, g.store, g.recovery, newMember(), RESTORED) + val error = runCatching { previewRestore(g.store, g.recovery) }.exceptionOrNull() + assertTrue("$error", error is LinkedException.RecoveryUsed) + } + + @Test + fun a_restore_signs_out_every_device_not_chosen_and_keeps_what_they_wrote() = + runTest { + val g = LinkedGroup(this).apply { create() } + Session.open(g.deviceDb(), g.store, g.a).push(export, LINK) + val lastOfA = + g.store.objects + .getValue(StorePaths.log(id(g.a), 1)) + .second + val (device, db) = restoredKeeping(g) + + val finished = finish(g.store, db, device) + + val last = g.store.roster.last() + assertEquals( + listOf(Op.RESTORE, Op.RECOVERY, Op.REMOVE), + g.store.roster + .takeLast(3) + .map { it.op }, + ) + assertEquals(listOf(id(g.a), id(g.b)).sorted(), last.remove) + val cut = last.cut.orEmpty() + assertEquals(setOf(id(g.a)), cut.keys) + assertEquals(1L, cut.getValue(id(g.a)).seq) + assertEquals(SyncBytes.toHex(sha256(lastOfA)), cut.getValue(id(g.a)).hash) + val devices = + g.store + .head() + .members + .filterValues { it.role == Role.DEVICE } + .keys + assertEquals(setOf(id(device)), devices) + assertNull(finished.words) + assertFalse(finished.session.removalPending) + assertNull(db.groups().group()?.keptDevices) + assertTrue(id(g.a) in finished.session.pull(setOf(LINK)).batches) + } + + @Test + fun a_device_that_was_chosen_stays_and_is_told_about_the_restore() = + runTest { + val g = LinkedGroup(this).apply { create() } + val dbA = g.deviceDb() + val (device, db) = restoredKeeping(g, keep = setOf(id(g.a))) + + finish(g.store, db, device) + + assertEquals( + listOf(id(g.b)), + g.store.roster + .last() + .remove, + ) + assertEquals(3, Session.open(dbA, g.store, g.a).state.epoch) + assertNotNull(dbA.groups().group()?.recoveryNoticeDevice) + } + + @Test + fun with_every_device_chosen_nothing_is_removed() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g, keep = setOf(id(g.a), id(g.b))) + + val finished = finish(g.store, db, device) + + assertEquals( + Op.RECOVERY, + g.store.roster + .last() + .op, + ) + assertFalse(finished.session.removalPending) + } + + @Test + fun a_device_that_appears_after_the_question_is_signed_out_too() = + runTest { + val g = LinkedGroup(this).apply { create() } + val keep = previewRestore(g.store, g.recovery).first { it.did == id(g.a) }.did + val late = newMember() + val state = g.store.head() + val add = + buildEntry( + g.group, + state, + draft(Op.ADD, g.a).copy(add = listOf(memberSubject(late.sigPublic, late.kemPublic, Role.DEVICE))), + listOf(g.a, late), + ) + g.store.appendRoster(state.head, add, wraps(g.group, 0, g.epochKey, listOf(late)), null) + val (device, db) = restoredKeeping(g, keep = setOf(keep)) + + finish(g.store, db, device) + + assertEquals( + listOf(id(g.b), id(late)).sorted(), + g.store.roster + .last() + .remove, + ) + } + + @Test + fun nothing_is_sealed_while_the_removal_is_pending() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + val session = Session.open(db, g.store, device) + assertTrue(session.removalPending) + + suspend fun refused( + path: String, + write: suspend () -> Unit, + ) { + val error = runCatching { write() }.exceptionOrNull() + assertTrue("$path: $error", error is LinkedException.RestoreUnfinished) + assertEquals(path, emptyList(), g.store.objectsOf(device)) + } + refused("a batch") { session.push(export, LINK) } + refused("a snapshot") { session.writeSnapshot(export, LINK) } + refused("a store login") { session.writeLogin(StoreLogin("user", "password")) } + refused("a live message") { session.sealLive(now, 1) } + Session.open(g.deviceDb(), g.store, g.a).push(export, LINK) + refused("an ack") { session.finish(session.pull(setOf(LINK))) } + } + + @Test + fun the_live_channel_sends_nothing_while_the_removal_is_pending() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + val local = heldStore(db) + local.addEntries(g.store.roster) + g.store.keys.forEach { (key, wrap) -> local.putKey(key.epoch, key.kc, key.name, wrap) } + val channel = LiveChannel(context, db, mockk { coEvery { load() } returns device }, local) + + assertNull(channel.send(now)) + + db.groups().forgetRestoreKeep() + assertNotNull(channel.send(now)) + } + + @Test + fun a_restore_that_stopped_before_the_removal_resumes_there() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + assertEquals( + Op.RECOVERY, + g.store.roster + .last() + .op, + ) + + // The next run starts from what was saved, with a new session. + finish(g.store, db, device) + + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + } + + @Test + fun a_restore_that_stopped_after_the_removal_does_not_remove_again() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + val stopped = + runCatching { + finishRestore(Session.open(db, g.store, device), device) { throw IOException("stopped") } + }.exceptionOrNull() + assertTrue("$stopped", stopped is IOException) + val entries = g.store.roster.size + assertNotNull(db.groups().group()?.keptDevices) + + val finished = finish(g.store, db, device) + + assertEquals(entries, g.store.roster.size) + assertFalse(finished.session.removalPending) + } + + @Test + fun a_removal_that_has_to_replace_the_words_hands_them_on() = + runTest { + val g = LinkedGroup(this).apply { create() } + val dbA = g.deviceDb() + val (device, db) = restoredKeeping(g) + replaceRecovery(Session.open(dbA, g.store, g.a), g.a) + + val finished = finish(g.store, db, device) + + assertNotNull(finished.words) + assertEquals(false, db.groups().group()?.recoverySaved) + assertFalse(g.store.head().recoveryDue) + } + + @Test + fun a_device_an_older_view_still_shows_cannot_open_what_is_written_next() = + runTest { + val g = LinkedGroup(this).apply { create() } + val older = MemoryStore() + older.roster += g.store.roster + older.keys += g.store.keys + removeDevices(Session.open(g.deviceDb(), g.store, g.a), g.a, listOf(id(g.b))) + assertFalse(id(g.b) in g.store.head().members) + assertTrue(id(g.b) in older.head().members) + + val (device, db) = restoredKeeping(g, store = older) + val session = finish(older, db, device).session + session.push(export, LINK) + + val epoch = session.state.epoch + assertFalse(id(g.b) in older.head().members) + assertFalse(older.holdsKey(epoch, id(g.b))) + val parsed = parseObject(older.objects.getValue(StorePaths.log(id(device), 1)).second, g.group) + assertEquals(epoch, parsed.header.epoch) + // The key before it was wrapped to B, as rule 5 asks, and it seals nothing. + val commitment = SyncBytes.b64urlDecode(checkNotNull(older.roster.first { it.epoch?.n == epoch - 1 }.epoch).kc) + val wrap = older.keys.getValue(KeyName(epoch - 1, kcName(commitment), id(g.b))) + val lastHeld = unwrapEpochKey(g.b.kemSecret, g.group, epoch - 1, g.b.did, wrap) + val error = runCatching { parsed.open(purposeKey(lastHeld, g.group, epoch, ObjectType.CHANGES.purpose)) }.exceptionOrNull() + assertTrue("$error", error is ProtocolException) + } + + private fun sync( + db: LinkedDatabase, + device: MemberSecrets, + local: LocalStore = heldStore(db), + ) = LinkedSync( + context, + db, + mockk { coEvery { load() } returns device }, + mockk(), + mockk(), + local, + SyncIndicator(context), + mockk { every { leaveFailed } returns MutableStateFlow(null) }, + mockk { coEvery { webDav() } returns WebDavLogin(WebDavStandIn.USER, WebDavStandIn.PASSWORD) }, + mockk(), + mockk { coEvery { links(any()) } returns null }, + mockk(), + ) + + @Test + fun the_first_sync_of_a_restored_device_signs_out_the_others() = + runTest { + val first = newMember() + val (words, _) = createWebDavGroup(emptyDb(), dav.store(), first, RESTORED, LINK) + val devices = previewWebDavRestore(dav.store(), words) + assertEquals(listOf(RESTORED.name), devices.map { it.name }) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + assertEquals(emptySet(), db.groups().group()?.keptDevices) + + val result = sync(db, device).syncNow() + + assertNull(result.error) + val roster = FileLayout(dav.store(), id(device)).rosterView() + assertEquals(Op.REMOVE, roster.maxBy { it.seq }.op) + assertEquals(listOf(id(first)), roster.maxBy { it.seq }.remove) + assertNull(db.groups().group()?.keptDevices) + } + + @Test + fun a_restored_device_that_cannot_reach_the_store_stays_pending() = + runTest { + val (words, _) = createWebDavGroup(emptyDb(), dav.store(), newMember(), RESTORED, LINK) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + val local = heldStore(db) + GroupStore(local, FileLayout(dav.store(), id(device))).download(id(device), 0, objects = false) + dav.close() + + val result = sync(db, device, local).syncNow() + + assertNotNull(result.error) + assertEquals(emptySet(), db.groups().group()?.keptDevices) + assertTrue(local.rosterAfter(-1).none { it.op == Op.REMOVE }) + } +} diff --git a/app/src/test/java/io/github/aedev/flow/sync2/WebDavGroupTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/WebDavGroupTest.kt index c23a30dbf..07fa52169 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/WebDavGroupTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/WebDavGroupTest.kt @@ -105,10 +105,10 @@ class WebDavGroupTest { runTest { val (words, _) = createWebDavGroup(emptyDb(), dav.store(), newMember(), meta, LINK) val db = emptyDb() - val newWords = restoreWebDavGroup(db, dav.store(), words, newMember(), meta) + val newWords = restoreWebDavGroup(db, dav.store(), words, newMember(), meta, emptySet()) assertNotNull(db.groups().group()?.webDavUrl) assertFalse(newWords == words) - val error = runCatching { restoreWebDavGroup(emptyDb(), dav.store(), words, newMember(), meta) }.exceptionOrNull() + val error = runCatching { restoreWebDavGroup(emptyDb(), dav.store(), words, newMember(), meta, emptySet()) }.exceptionOrNull() assertTrue("used words don't restore twice: $error", error is LinkedException.RecoveryUsed) } } diff --git a/app/src/test/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetupTest.kt b/app/src/test/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetupTest.kt index 55c1ec0d6..03bdf4b8f 100644 --- a/app/src/test/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetupTest.kt +++ b/app/src/test/java/io/github/aedev/flow/ui/screens/sync/linked/LinkedSetupTest.kt @@ -6,15 +6,19 @@ import androidx.compose.runtime.getValue import androidx.compose.runtime.mutableStateOf import androidx.compose.runtime.remember import androidx.compose.runtime.setValue +import androidx.compose.ui.semantics.SemanticsActions import androidx.compose.ui.test.assertIsEnabled import androidx.compose.ui.test.assertIsNotEnabled +import androidx.compose.ui.test.isToggleable import androidx.compose.ui.test.junit4.v2.createComposeRule import androidx.compose.ui.test.onNodeWithText import androidx.compose.ui.test.performClick +import androidx.compose.ui.test.performSemanticsAction import androidx.compose.ui.test.performTextInput import androidx.test.core.app.ApplicationProvider import androidx.test.ext.junit.runners.AndroidJUnit4 import io.github.aedev.flow.R +import io.github.aedev.flow.sync2.linking.RestoreDevice import io.github.aedev.flow.sync2.service.LinkedError import org.junit.Assert.assertEquals import org.junit.Rule @@ -36,7 +40,9 @@ class LinkedSetupTest { private val created = ArrayList>() private val reset = ArrayList>() - private fun show() = + private val kept = ArrayList>() + + private fun show(restoreDevices: List? = null) = rule.setContent { var error by remember { mutableStateOf(null) } LinkedSetup( @@ -52,6 +58,9 @@ class LinkedSetupTest { onShowCode = {}, onRestore = { _, _ -> }, onRestoreFromWebDav = { _, _, _, _ -> }, + restoreDevices = restoreDevices, + onConfirmRestore = { kept += it }, + onCancelRestore = {}, ) } @@ -87,6 +96,24 @@ class LinkedSetupTest { rule.onNodeWithText(text(R.string.linked_create_lan_title)).assertDoesNotExist() } + @Test + fun a_restore_keeps_no_device_unless_it_is_ticked() { + val devices = listOf(RestoreDevice("aa".repeat(16), "Phone", "android"), RestoreDevice("bb".repeat(16), "", "android")) + show(devices) + rule.onNodeWithText(text(R.string.linked_restore_choose_body)).assertExists() + rule.onNodeWithText(context.getString(R.string.linked_restore_unknown_device, "bbbbbbbb")).assertExists() + rule.onNodeWithText(text(R.string.linked_restore)).performClick() + assertEquals(listOf(emptySet()), kept) + } + + @Test + fun a_ticked_device_is_kept() { + show(listOf(RestoreDevice("aa".repeat(16), "Phone", "android"))) + rule.onAllNodes(isToggleable())[0].performSemanticsAction(SemanticsActions.OnClick) + rule.onNodeWithText(text(R.string.linked_restore)).performClick() + assertEquals(listOf(setOf("aa".repeat(16))), kept) + } + @Test fun a_server_that_already_has_a_group_offers_joining_or_a_confirmed_reset() { show() diff --git a/app/src/test/resources/sync2/README.md b/app/src/test/resources/sync2/README.md index e4df2893a..88a128d1b 100644 --- a/app/src/test/resources/sync2/README.md +++ b/app/src/test/resources/sync2/README.md @@ -14,9 +14,10 @@ All byte strings are lowercase hex, except inside roster entries, which use the | `keys.json` | Member ids, epoch commitment, purpose keys, HPKE wrap with a fixed ephemeral key, wrong-epoch and wrong-member unwraps, history chain, wrap and chain paths | | `recovery.json` | Recovery secret, 24 words and derived keys; words with a bad checksum, 23 words, and a valid 12-word phrase | | `envelope.json` | Log objects of each type, including a store login, opened in stream order, a live message, a snapshot blob; truncation, magic, suite, group, unknown sender, signature, tampering, replay, gap, a log read from the middle, wrong `prev`, swapped epoch, `seq` above 253 − 1, blob hash | -| `roster.json` | A chain through genesis, add, rename, remove, restore, recovery change, add, leave and rotate, with the state after each entry (including whether a rotate is due) and the device names it carries; bad and missing signatures, unknown fields, wrong `seq`, truncated chain, removed signer, weak key, re-adding a removed device, wrong epoch sets and numbers, self-removal, renaming the recovery key, a non-member signer, a used recovery key; a leave by a non-member, by the recovery member and by the last device, a leave removing another device or carrying an epoch, a rotate without an epoch, with a removal, or wrapped to a device that left; six forks and their winners, including leave against remove and rotate, and concurrent rotates; a genesis whose group id isn't the one its keys commit to | +| `roster.json` | A chain through genesis, add, rename, remove, restore, recovery change, add, leave and rotate, with the state after each entry (including whether a rotate is due, and each member's epochs and `cut`) and the device names it carries; bad and missing signatures, unknown fields, wrong `seq`, truncated chain, removed signer, weak key, re-adding a removed device, wrong epoch sets and numbers, self-removal, renaming the recovery key, a non-member signer, a used recovery key; a leave by a non-member, by the recovery member and by the last device, a leave removing another device or carrying an epoch, a rotate without an epoch, with a removal, or wrapped to a device that left; six forks and their winners, including leave against remove and rotate, and concurrent rotates; a genesis whose group id isn't the one its keys commit to; a `cut` on the wrong op, for a device not removed, empty, or at `seq` 0; `replacements`, entries on top of the chain where the device that set the recovery key is removed or has left | | `forks.json` | Named entries, and cases that settle a store's view of them: `applied` is what the device had applied, `view` is the store's answer, `pins` the chosen entries, `linear` a Flow server. Each case gives the outcome and, in any order of `view`, the same chain, standing entries that lost and start an epoch, contested set and sides | -| `pairing.json` | Pairing keys, SAS, mailbox slot, and a sealed `join` message | +| `senders.json` | Step 2 of [reading an envelope](../store.md#reading-an-envelope): a roster, then `senders` with who may write under which epoch and how far along its log, then `walks` of a log down from a `cut`, with the hashes found from the lowest `seq` up or the check that refuses it. An empty string in `log` is an object the store doesn't have | +| `pairing.json` | Commitment, proof, the scanner's first message, pairing keys, SAS, mailbox slot, and a sealed `join` message; `refused` holds a shower key that doesn't match the commitment and a proof made with another token | | `lan.json` | Discovery tag, hello AAD, transcripts, session keys, frame AAD | ## Using them diff --git a/app/src/test/resources/sync2/forks.json b/app/src/test/resources/sync2/forks.json index 860f51336..99cabb894 100644 --- a/app/src/test/resources/sync2/forks.json +++ b/app/src/test/resources/sync2/forks.json @@ -562,6 +562,60 @@ "6e1946f855d4d7de61c0d3e3e4491cae": "4aPE0cvXlKz8NZR0T2W6e2_mF1q57W3xCPWOsLDnm0_9tTnsjTltq4cbrR6M8mKwqoObdlHLQV2iuQUk2tFXDQ" } }, + "restoredKeepsNone": { + "v": 2, + "group": "669a44456c08e4a12e4ceb3538860317", + "seq": 5, + "prev": "2d558a93916b6e84b40709e5d8c6c2ab8767619970ce6658feef10698f0ef97e", + "op": "remove", + "signer": "b2d4dcb5ace60ab7f499b462f715ffcd", + "remove": [ + "894e311b2b2058e3bcc2ef3bde8ec864", + "addf195306ed8f3c4d02642d8b6ebd6a", + "d8576f2de42861ceb8030cc722b4c669" + ], + "cut": { + "d8576f2de42861ceb8030cc722b4c669": { + "seq": 1, + "hash": "c26f8df94d53d79f589e6bd279730004666d0e2d177ee7730a41d9047b7d4229" + } + }, + "epoch": { + "n": 3, + "kc": "OEuCk5x1sPm47KRWu-suMstQN8LQYHwkQLh-3rwUzTc", + "to": [ + "6e1946f855d4d7de61c0d3e3e4491cae", + "b2d4dcb5ace60ab7f499b462f715ffcd" + ] + }, + "sigs": { + "b2d4dcb5ace60ab7f499b462f715ffcd": "B1jqw4xYJPAhu0AKQDycIriDFQmDoRfkjJecMl6NGfZLj_sfeAouRcCYWtWZtahEZg-NXtICW7th_-46HrnjCQ" + } + }, + "restoredKeepsA": { + "v": 2, + "group": "669a44456c08e4a12e4ceb3538860317", + "seq": 5, + "prev": "2d558a93916b6e84b40709e5d8c6c2ab8767619970ce6658feef10698f0ef97e", + "op": "remove", + "signer": "b2d4dcb5ace60ab7f499b462f715ffcd", + "remove": [ + "894e311b2b2058e3bcc2ef3bde8ec864", + "addf195306ed8f3c4d02642d8b6ebd6a" + ], + "epoch": { + "n": 3, + "kc": "U5UZ6k_DWcTW-28X3oFfc6ACWFdLf3I7fodW9kCNzvQ", + "to": [ + "6e1946f855d4d7de61c0d3e3e4491cae", + "b2d4dcb5ace60ab7f499b462f715ffcd", + "d8576f2de42861ceb8030cc722b4c669" + ] + }, + "sigs": { + "b2d4dcb5ace60ab7f499b462f715ffcd": "E0ikYKN_DPFTwztMiuuVCadgrppNgTUXY2XFVZqy4zevXjcMS1mM-oU0sdSXOB8JYT8_2dvCsSzmB6H3Cj3dBw" + } + }, "aRotate": { "v": 2, "group": "669a44456c08e4a12e4ceb3538860317", @@ -1064,6 +1118,125 @@ "extendsApplied": true } }, + { + "name": "restore-signs-out-a-device-whose-removal-it-never-read", + "applied": [ + "genesis", + "addB", + "addC", + "aRemovesC" + ], + "view": [ + "genesis", + "addB", + "addC", + "aRemovesC", + "restore", + "restored", + "restoredKeepsNone" + ], + "pins": [], + "linear": false, + "expect": { + "outcome": "apply", + "chain": [ + "genesis", + "addB", + "addC", + "restore", + "restored", + "restoredKeepsNone" + ], + "standing": [], + "extendsApplied": false + } + }, + { + "name": "restore-signing-out-a-device-on-a-linear-store", + "applied": [ + "genesis", + "addB", + "addC", + "aRemovesC" + ], + "view": [ + "genesis", + "addB", + "addC", + "aRemovesC", + "restore", + "restored", + "restoredKeepsNone" + ], + "pins": [], + "linear": true, + "expect": { + "outcome": "rewritten" + } + }, + { + "name": "restore-keeping-the-device-that-removed-another", + "applied": [ + "genesis", + "addB", + "addC", + "aRemovesC" + ], + "view": [ + "genesis", + "addB", + "addC", + "aRemovesC", + "restore", + "restored", + "restoredKeepsA" + ], + "pins": [], + "linear": false, + "expect": { + "outcome": "apply", + "chain": [ + "genesis", + "addB", + "addC", + "aRemovesC" + ], + "standing": [ + "restore", + "restored", + "restoredKeepsA" + ], + "extendsApplied": true + } + }, + { + "name": "restore-signing-out-the-device-that-replaced-its-words", + "applied": [], + "view": [ + "genesis", + "addB", + "addC", + "replaceWords", + "restore", + "restored", + "restoredKeepsNone" + ], + "pins": [], + "linear": false, + "expect": { + "outcome": "contested", + "contested": [ + "replaceWords", + "restore", + "restored", + "restoredKeepsNone" + ], + "sides": [ + "replaceWords", + "restoredKeepsNone" + ] + } + }, { "name": "restore-against-the-removal-that-replaces-its-key", "applied": [], diff --git a/app/src/test/resources/sync2/roster.json b/app/src/test/resources/sync2/roster.json index 819e13785..9c3a67ef2 100644 --- a/app/src/test/resources/sync2/roster.json +++ b/app/src/test/resources/sync2/roster.json @@ -2461,6 +2461,183 @@ } ], "error": "roster-op" + }, + { + "name": "restore-then-remove-the-other-devices", + "steps": [ + { + "v": 2, + "group": "669a44456c08e4a12e4ceb3538860317", + "seq": 9, + "prev": "18eb362b0fb928a0f35b684a8cbfae6f0073cfcd3ba0c5c6dbd7af74343f3224", + "op": "restore", + "signer": "6e1946f855d4d7de61c0d3e3e4491cae", + "add": [ + { + "did": "b2d4dcb5ace60ab7f499b462f715ffcd", + "sig": "RsTU5IjEf1rjeyLp-09xbUp-OfjjWUqwFdO04YajNZ0", + "kem": "GXeu_UDlHVNWxtd9eoaDmQgIMjNwiRnINQ5z03xH_18", + "role": "device" + } + ], + "epoch": { + "n": 5, + "kc": "t3FcIp-eA9A7wQzAGmAeZWCCb-9ut5XU7Je9TSgOfIw", + "to": [ + "6e1946f855d4d7de61c0d3e3e4491cae", + "addf195306ed8f3c4d02642d8b6ebd6a", + "b2d4dcb5ace60ab7f499b462f715ffcd", + "d8576f2de42861ceb8030cc722b4c669" + ] + }, + "sigs": { + "6e1946f855d4d7de61c0d3e3e4491cae": "Lm4m-c5647BTraOvYQVk2d0qp9ZxtzKW9Bw641HKINIwoHTBYuQaozDUCnRQOnJ1UpOhz_bwekpYHzHQR1RwBA", + "b2d4dcb5ace60ab7f499b462f715ffcd": "Zwr3xThm5PExuX3A0fZnlZDIqNE5fgPh0DdkE_JqwkrBNvtC5oe6nAB6vL-ygIfvscB5TFW-0ODysKoQ_4ozCw" + } + }, + { + "v": 2, + "group": "669a44456c08e4a12e4ceb3538860317", + "seq": 10, + "prev": "bf22756ef83c16e04ae4b185b694786f4811f7fffad8a7434ca53a9b8b9f4b93", + "op": "recovery", + "signer": "b2d4dcb5ace60ab7f499b462f715ffcd", + "remove": [ + "6e1946f855d4d7de61c0d3e3e4491cae" + ], + "add": [ + { + "did": "2b4bc2fbe53e4adaad7b3c9270a335f5", + "sig": "7Ia_W6iMAVlxUYngARu37l8gwiFhc9zFF4mkxPi5jEQ", + "kem": "K4o73nvvMWnnFXK8heU7G_VEa7kvVeQ0wyD3NDeYSRg", + "role": "recovery" + } + ], + "epoch": { + "n": 6, + "kc": "7DcLcYbUbnlQRas76vQS4ZiStJRHmCWrzzg5IMMIC5Q", + "to": [ + "2b4bc2fbe53e4adaad7b3c9270a335f5", + "addf195306ed8f3c4d02642d8b6ebd6a", + "b2d4dcb5ace60ab7f499b462f715ffcd", + "d8576f2de42861ceb8030cc722b4c669" + ] + }, + "sigs": { + "b2d4dcb5ace60ab7f499b462f715ffcd": "qokhQg4drEVSAzq1KcDMGGiHRLLrNWIgXYOcf7XLG64oONWy-DqsVzA-rHgpnJbNIsuNCjvCZSbhT48QZhrlBw", + "2b4bc2fbe53e4adaad7b3c9270a335f5": "4HQMdWtGKO9nhB8gKV2wGL7pYshoS6xPHvRDlaixOUruzkq3HrkL15PGPn9NG1ooy4aFdui-1TIt9wCJsUzbCg" + } + }, + { + "v": 2, + "group": "669a44456c08e4a12e4ceb3538860317", + "seq": 11, + "prev": "3222ecd63ebba6fdf0cd66addd7e342fc84f22cde1b1b82f3307191ebb45aeb1", + "op": "remove", + "signer": "b2d4dcb5ace60ab7f499b462f715ffcd", + "remove": [ + "addf195306ed8f3c4d02642d8b6ebd6a", + "d8576f2de42861ceb8030cc722b4c669" + ], + "cut": { + "d8576f2de42861ceb8030cc722b4c669": { + "seq": 1, + "hash": "c26f8df94d53d79f589e6bd279730004666d0e2d177ee7730a41d9047b7d4229" + } + }, + "epoch": { + "n": 7, + "kc": "XYPM-00Tr4D-k5DLBkOxpZwRiwzqY1Dd8U9UdcG-4X8", + "to": [ + "2b4bc2fbe53e4adaad7b3c9270a335f5", + "b2d4dcb5ace60ab7f499b462f715ffcd" + ] + }, + "sigs": { + "b2d4dcb5ace60ab7f499b462f715ffcd": "DGiqV7M-bAVYPifm3ki0VNuR1JY5D6KHXsGTjiu153Qkn6hzKWcHavoSQhSVkvBOPnkNZE_VlXHoposU25iqBg" + } + } + ], + "state": { + "seq": 11, + "head": "7be6485aba52772a2d633177680fb8bb2546df68fe8307250e110e8d55e9b506", + "epoch": 7, + "rotate": false, + "recoveryDue": false, + "members": [ + { + "did": "2b4bc2fbe53e4adaad7b3c9270a335f5", + "role": "recovery", + "used": false, + "installer": "b2d4dcb5ace60ab7f499b462f715ffcd" + }, + { + "did": "b2d4dcb5ace60ab7f499b462f715ffcd", + "role": "device", + "used": false + } + ], + "senders": [ + { + "did": "1edce5f29fbda5bba7ada194b0a578c6", + "role": "device", + "since": 3, + "until": 3, + "cut": { + "seq": 1, + "hash": "301bf18e9cce13b93870928117aecd06ff793354402b03810bff6476e123865b" + } + }, + { + "did": "2b4bc2fbe53e4adaad7b3c9270a335f5", + "role": "recovery", + "since": 6 + }, + { + "did": "6e1946f855d4d7de61c0d3e3e4491cae", + "role": "recovery", + "since": 3, + "until": 5 + }, + { + "did": "894e311b2b2058e3bcc2ef3bde8ec864", + "role": "device", + "since": 0, + "until": 0, + "cut": { + "seq": 2, + "hash": "4d4573758bd47b198e4305475ff362af7b9806c157caff61d657837e94134ffe" + } + }, + { + "did": "ab6095990ebaaf36d4c49e86fc3a65c3", + "role": "recovery", + "since": 0, + "until": 2 + }, + { + "did": "addf195306ed8f3c4d02642d8b6ebd6a", + "role": "device", + "since": 2, + "until": 6 + }, + { + "did": "b2d4dcb5ace60ab7f499b462f715ffcd", + "role": "device", + "since": 5 + }, + { + "did": "d8576f2de42861ceb8030cc722b4c669", + "role": "device", + "since": 0, + "until": 6, + "cut": { + "seq": 1, + "hash": "c26f8df94d53d79f589e6bd279730004666d0e2d177ee7730a41d9047b7d4229" + } + } + ] + } } ], "geneses": [ From 802336f331754905dd75d207bff6a099b304543d Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 17:05:59 +0200 Subject: [PATCH 2/4] fix(sync): hold a restore's new words before reading the roster again --- .../aedev/flow/sync2/linking/Recovery.kt | 20 +++-- .../aedev/flow/sync2/service/LinkedSync.kt | 6 +- .../aedev/flow/sync2/RestoreChoiceTest.kt | 74 ++++++++++++++++--- 3 files changed, 75 insertions(+), 25 deletions(-) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt b/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt index 8f08a260f..5c240d3e2 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/linking/Recovery.kt @@ -109,28 +109,26 @@ internal fun Session.unchosenDevices(): List { .sorted() } -/** Where a restore's last step left things: the session to go on with, and new words to show when the removal had to replace them. */ -class FinishedRestore( - val session: Session, - val words: String?, -) - /** * Signs out the devices the user didn't choose (spec/protocol.md, Recovery, step 6), having read every log without * writing, so that nothing is sealed under a key they hold. [reread] opens the roster again from the store. A removal * that lost a fork stays due for the next sync, and so does one that dies on the way: it starts over from the roster. + * New words the removal had to make go to [holdWords] before the roster is read again, which can fail. Returns the + * session to go on with. */ suspend fun finishRestore( session: Session, device: MemberSecrets, + holdWords: (String) -> Unit, reread: suspend () -> Session, -): FinishedRestore { +): Session { var current = session - var words: String? = null val unchosen = current.unchosenDevices() if (unchosen.isNotEmpty()) { - words = removeDevices(current, device, unchosen, current.readablePositions()) - if (words != null) current.db.groups().setRecoverySaved(false) + removeDevices(current, device, unchosen, current.readablePositions())?.let { words -> + holdWords(words) + current.db.groups().setRecoverySaved(false) + } current.db.snapshots().snapshotSoon() current = reread() } @@ -138,7 +136,7 @@ suspend fun finishRestore( current.db.groups().forgetRestoreKeep() current = reread() } - return FinishedRestore(current, words) + return current } /** diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt index b11fa9a29..4183c7bf8 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt @@ -230,13 +230,11 @@ class LinkedSync if (session.removalPending) { // It needs the store to take the removal; what it read stays for this sync once that's done. if (offline != null) throw offline - val finished = - finishRestore(session, device) { + session = + finishRestore(session, device, { replacedWords.value = it }) { store.download(me, record.serverCursor, objects = false) Session.open(db, reading, device) } - finished.words?.let { replacedWords.value = it } - session = finished.session if (session.removalPending) return } val links = profiles.links(session) ?: return diff --git a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt index 8918885b0..ff62b93ef 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt @@ -10,6 +10,8 @@ import io.github.aedev.flow.sync.protocol.SyncCollection import io.github.aedev.flow.sync2.crypto.MemberSecrets import io.github.aedev.flow.sync2.crypto.ProtocolException import io.github.aedev.flow.sync2.crypto.purposeKey +import io.github.aedev.flow.sync2.crypto.recoveryKeys +import io.github.aedev.flow.sync2.crypto.recoverySecret import io.github.aedev.flow.sync2.crypto.sha256 import io.github.aedev.flow.sync2.crypto.unwrapEpochKey import io.github.aedev.flow.sync2.data.GroupRecord @@ -104,7 +106,8 @@ class RestoreChoiceTest { store: MemoryStore, db: LinkedDatabase, device: MemberSecrets, - ) = finishRestore(Session.open(db, store, device), device) { Session.open(db, store, device) } + holdWords: (String) -> Unit = {}, + ) = finishRestore(Session.open(db, store, device), device, holdWords) { Session.open(db, store, device) } private fun MemoryStore.objectsOf(device: MemberSecrets) = objects.keys.filter { it.startsWith("dev/${id(device)}/") } @@ -167,10 +170,9 @@ class RestoreChoiceTest { .filterValues { it.role == Role.DEVICE } .keys assertEquals(setOf(id(device)), devices) - assertNull(finished.words) - assertFalse(finished.session.removalPending) + assertFalse(finished.removalPending) assertNull(db.groups().group()?.keptDevices) - assertTrue(id(g.a) in finished.session.pull(setOf(LINK)).batches) + assertTrue(id(g.a) in finished.pull(setOf(LINK)).batches) } @Test @@ -206,7 +208,7 @@ class RestoreChoiceTest { .last() .op, ) - assertFalse(finished.session.removalPending) + assertFalse(finished.removalPending) } @Test @@ -306,7 +308,7 @@ class RestoreChoiceTest { val (device, db) = restoredKeeping(g) val stopped = runCatching { - finishRestore(Session.open(db, g.store, device), device) { throw IOException("stopped") } + finishRestore(Session.open(db, g.store, device), device, {}) { throw IOException("stopped") } }.exceptionOrNull() assertTrue("$stopped", stopped is IOException) val entries = g.store.roster.size @@ -315,7 +317,7 @@ class RestoreChoiceTest { val finished = finish(g.store, db, device) assertEquals(entries, g.store.roster.size) - assertFalse(finished.session.removalPending) + assertFalse(finished.removalPending) } @Test @@ -326,13 +328,46 @@ class RestoreChoiceTest { val (device, db) = restoredKeeping(g) replaceRecovery(Session.open(dbA, g.store, g.a), g.a) - val finished = finish(g.store, db, device) + var words: String? = null + finish(g.store, db, device) { words = it } - assertNotNull(finished.words) + assertNotNull(words) assertEquals(false, db.groups().group()?.recoverySaved) assertFalse(g.store.head().recoveryDue) } + @Test + fun new_words_are_held_and_marked_unsaved_when_the_next_read_of_the_roster_fails() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + db.groups().setRecoverySaved(true) + replaceRecovery(Session.open(g.deviceDb(), g.store, g.a), g.a) + + var words: String? = null + val stopped = + runCatching { + finishRestore(Session.open(db, g.store, device), device, { words = it }) { throw IOException("stopped") } + }.exceptionOrNull() + + assertTrue("$stopped", stopped is IOException) + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + val next = recoveryKeys(checkNotNull(recoverySecret(checkNotNull(words)))) + assertEquals( + Role.RECOVERY, + g.store + .head() + .members[id(next)] + ?.role, + ) + assertEquals(false, db.groups().group()?.recoverySaved) + } + @Test fun a_device_an_older_view_still_shows_cannot_open_what_is_written_next() = runTest { @@ -345,7 +380,7 @@ class RestoreChoiceTest { assertTrue(id(g.b) in older.head().members) val (device, db) = restoredKeeping(g, store = older) - val session = finish(older, db, device).session + val session = finish(older, db, device) session.push(export, LINK) val epoch = session.state.epoch @@ -401,6 +436,25 @@ class RestoreChoiceTest { assertNull(db.groups().group()?.keptDevices) } + @Test + fun the_sync_hands_on_words_its_removal_had_to_make() = + runTest { + val first = newMember() + val dbFirst = emptyDb() + val (words, _) = createWebDavGroup(dbFirst, dav.store(), first, RESTORED, LINK) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + val storeFirst = GroupStore(heldStore(dbFirst), FileLayout(dav.store(), id(first))) + storeFirst.download(id(first), 0, objects = false) + replaceRecovery(Session.open(dbFirst, storeFirst, first), first) + + val sync = sync(db, device) + assertNull(sync.syncNow().error) + + assertNotNull(sync.replacedWords.value) + } + @Test fun a_restored_device_that_cannot_reach_the_store_stays_pending() = runTest { From b2ecf1f7e4c238b01b69805b5af2daa671916d6a Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sat, 7 Nov 2026 19:34:01 +0200 Subject: [PATCH 3/4] fix(sync): end every read of a store at a limit and hold a restore's removal until it's done --- .../aedev/flow/sync2/service/LinkedSync.kt | 2 + .../aedev/flow/sync2/store/FileLayout.kt | 11 +- .../aedev/flow/sync2/store/GroupStore.kt | 16 +- .../aedev/flow/sync2/store/ReadBudget.kt | 21 ++ .../aedev/flow/sync2/store/ServerStore.kt | 4 +- .../aedev/flow/sync2/RestoreChoiceTest.kt | 342 ++++++++++++++++++ .../aedev/flow/sync2/ServerStoreTest.kt | 16 + 7 files changed, 405 insertions(+), 7 deletions(-) create mode 100644 app/src/main/java/io/github/aedev/flow/sync2/store/ReadBudget.kt diff --git a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt index 4183c7bf8..f44ef84df 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/service/LinkedSync.kt @@ -230,6 +230,8 @@ class LinkedSync if (session.removalPending) { // It needs the store to take the removal; what it read stays for this sync once that's done. if (offline != null) throw offline + // A read that isn't finished doesn't remove: the next sync goes on from what it holds (spec/protocol.md, Recovery, step 6). + if (store.cutShort) throw LinkedException.RestoreUnfinished() session = finishRestore(session, device, { replacedWords.value = it }) { store.download(me, record.serverCursor, objects = false) diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt index 53e23debe..8d50a1765 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/FileLayout.kt @@ -83,11 +83,14 @@ class FileLayout( /** * Copies every other sender's log objects after the last one [local] holds, reading the next `seq` * until there is none, or a file that isn't that object (spec/store.md, WebDAV). A sender never read starts at its oldest object. - * When the store tags folders, nothing is read while the tag stays the same. + * When the store tags folders, nothing is read while the tag stays the same. False when [budget] ended the read. */ - suspend fun pullLogs(local: LocalStore) { + suspend fun pullLogs( + local: LocalStore, + budget: ReadBudget = ReadBudget(), + ): Boolean { val tag = files.tag("dev") - if (tag != null && tags[files.location] == tag) return + if (tag != null && tags[files.location] == tag) return true val check = ObjectCheck(me, local.resolved().chain) val tips = local.logTips() var whole = true @@ -102,10 +105,12 @@ class FileLayout( break } local.putObject(path, bytes) + if (!budget.take(1)) return false seq++ } } if (whole) tag?.let { tags[files.location] = it } + return true } /** The oldest log object of [did] still stored, from its first shard only. */ diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt index 98e72883e..04578b43d 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/GroupStore.kt @@ -11,7 +11,12 @@ import io.github.aedev.flow.sync2.roster.entryHash class GroupStore( val local: LocalStore, val server: Remote?, + private val newBudget: () -> ReadBudget = ::ReadBudget, ) : ChangeList { + /** A read of objects stopped on its [ReadBudget], short of the end. The next one goes on from where it got to. */ + var cutShort = false + private set + override suspend fun rosterAfter(after: Long): List = local.rosterAfter(after) /** What the shared store answers, which decides, or what this device holds when it has none. */ @@ -82,8 +87,8 @@ class GroupStore( /** * Copies what the server holds that this device doesn't: roster entries, this device's wraps and - * the chain boxes, and, with [objects], the objects after [cursor]. Returns the next cursor, which - * stays before the first page holding an object this device couldn't keep, so that page is read again. + * the chain boxes, and, with [objects], the objects after [cursor], up to its [ReadBudget]. Returns the next + * cursor, which stays before the first page holding an object this device couldn't keep, so that page is read again. */ suspend fun download( me: String, @@ -104,11 +109,12 @@ class GroupStore( } if (!objects) return cursor if (server is FileLayout) { - server.pullLogs(local) + cutShort = !server.pullLogs(local, newBudget()) return cursor } if (server !is ChangeList) return cursor val check = ObjectCheck(me, resolved.chain) + val budget = newBudget() var next = cursor var unkept: Long? = null while (true) { @@ -121,6 +127,10 @@ class GroupStore( // A cursor that doesn't move would hand out the same page forever. if (changed.isEmpty() || after <= next) return unkept ?: next next = after + if (!budget.take(changed.size)) { + cutShort = true + return unkept ?: next + } } } diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/ReadBudget.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/ReadBudget.kt new file mode 100644 index 000000000..1d12e33c5 --- /dev/null +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/ReadBudget.kt @@ -0,0 +1,21 @@ +package io.github.aedev.flow.sync2.store + +private const val REQUESTS = 4096 +private const val OBJECTS = 200_000 + +/** + * How much one read may take from a store (spec/store.md, Make progress or stop). A cursor that keeps advancing + * passes that rule, so a total has to end the read too. Both limits sit far above what a group holds, since garbage + * collection keeps its logs short. + */ +class ReadBudget( + private var requests: Int = REQUESTS, + private var objects: Int = OBJECTS, +) { + /** Counts one answer naming [named] objects; false once the read should end. */ + fun take(named: Int): Boolean { + requests-- + objects -= named + return requests > 0 && objects > 0 + } +} diff --git a/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt b/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt index f0015e45e..d4bfeea7c 100644 --- a/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt +++ b/app/src/main/java/io/github/aedev/flow/sync2/store/ServerStore.kt @@ -8,6 +8,7 @@ import io.github.aedev.flow.sync2.crypto.sign import io.github.aedev.flow.sync2.crypto.tag import io.github.aedev.flow.sync2.engine.LinkedException import io.github.aedev.flow.sync2.roster.Entry +import io.github.aedev.flow.sync2.roster.MAX_ROSTER_ENTRIES import io.github.aedev.flow.sync2.roster.entryJson import io.github.aedev.flow.sync2.roster.parseEntry import kotlinx.coroutines.Dispatchers @@ -336,7 +337,7 @@ class ServerStore private constructor( ?.map(::parseEntry) ?: throw StoreException.Malformed() - /** The whole roster, page by page. A page that doesn't advance ends the read. */ + /** The whole roster, page by page. A page that doesn't advance, or one past a roster's limit, ends the read. */ override suspend fun rosterView(): List { val entries = ArrayList() var after = -1L @@ -346,6 +347,7 @@ class ServerStore private constructor( if (last <= after) throw StoreException.Malformed() after = last entries += page + if (entries.size > MAX_ROSTER_ENTRIES) throw StoreException.Malformed() } } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt index ff62b93ef..34887aed6 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt @@ -7,6 +7,7 @@ import androidx.test.ext.junit.runners.AndroidJUnit4 import io.github.aedev.flow.sync.crypto.SyncBytes import io.github.aedev.flow.sync.protocol.CollectionWire import io.github.aedev.flow.sync.protocol.SyncCollection +import io.github.aedev.flow.sync2.changes.batchContent import io.github.aedev.flow.sync2.crypto.MemberSecrets import io.github.aedev.flow.sync2.crypto.ProtocolException import io.github.aedev.flow.sync2.crypto.purposeKey @@ -42,15 +43,23 @@ import io.github.aedev.flow.sync2.roster.Op import io.github.aedev.flow.sync2.roster.Role import io.github.aedev.flow.sync2.roster.buildEntry import io.github.aedev.flow.sync2.roster.memberSubject +import io.github.aedev.flow.sync2.service.LinkedError import io.github.aedev.flow.sync2.service.LinkedProfiles import io.github.aedev.flow.sync2.service.LinkedReset import io.github.aedev.flow.sync2.service.LinkedSync import io.github.aedev.flow.sync2.service.SyncIndicator +import io.github.aedev.flow.sync2.store.ChangeList +import io.github.aedev.flow.sync2.store.ChangedObject import io.github.aedev.flow.sync2.store.FileLayout import io.github.aedev.flow.sync2.store.GroupStore import io.github.aedev.flow.sync2.store.KeyName import io.github.aedev.flow.sync2.store.LocalStore +import io.github.aedev.flow.sync2.store.MAX_OBJECT_SIZE +import io.github.aedev.flow.sync2.store.Mailbox import io.github.aedev.flow.sync2.store.ObjectType +import io.github.aedev.flow.sync2.store.ReadBudget +import io.github.aedev.flow.sync2.store.Remote +import io.github.aedev.flow.sync2.store.StoreException import io.github.aedev.flow.sync2.store.StoreLogin import io.github.aedev.flow.sync2.store.StorePaths import io.github.aedev.flow.sync2.store.kcName @@ -58,6 +67,7 @@ import io.github.aedev.flow.sync2.store.parseObject import io.mockk.coEvery import io.mockk.every import io.mockk.mockk +import io.mockk.spyk import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.test.TestScope import kotlinx.coroutines.test.runTest @@ -472,4 +482,336 @@ class RestoreChoiceTest { assertEquals(emptySet(), db.groups().group()?.keptDevices) assertTrue(local.rosterAfter(-1).none { it.op == Op.REMOVE }) } + + /** The store one object to a page, so a read can stop between objects. Asking for a path in [failing] throws what it names. */ + private class OnePerPage( + private val store: MemoryStore, + ) : Remote, + ChangeList by store, + Mailbox by store { + val asked = ArrayList() + val failing = HashMap() + + override suspend fun changes(cursor: Long): Pair, Long> { + val page = store.changes(cursor).first.take(1) + return page to (page.firstOrNull()?.let { store.objects.getValue(it.path).first } ?: cursor) + } + + override suspend fun getObject(path: String): ByteArray? { + failing[path]?.let { throw it } + asked += path + return store.getObject(path) + } + } + + /** A store whose cursor always moves on: each page names one more object, which it doesn't have. */ + private class EndlessPages( + store: MemoryStore, + ) : Remote, + ChangeList by store, + Mailbox by store { + private var pages = 0L + + override suspend fun changes(cursor: Long): Pair, Long> { + check(++pages < 10_000) { "the pages never ended" } + return listOf(ChangedObject(StorePaths.log("00".repeat(16), pages), 1)) to pages + } + } + + private fun syncOn( + db: LinkedDatabase, + device: MemberSecrets, + server: Remote, + budget: () -> ReadBudget = ::ReadBudget, + ): LinkedSync { + val local = heldStore(db) + return spyk(sync(db, device, local)) { coEvery { groupStore(any(), any()) } coAnswers { GroupStore(local, server, budget) } } + } + + private suspend fun Session.wrote(count: Int): List { + repeat( + count, + ) { write(ObjectType.CHANGES, batchContent(SyncCollection.WATCH_HISTORY, "00".repeat(32), LINK, export.values.single().lines)) } + return (1..count).map { StorePaths.log(me, it.toLong()) } + } + + private suspend fun TestScope.wrote( + g: LinkedGroup, + member: MemberSecrets, + count: Int, + ) = Session.open(g.deviceDb(), g.store, member).wrote(count) + + private fun copyOf(store: MemoryStore) = + MemoryStore().also { + it.roster += store.roster + it.keys += store.keys + it.objects += store.objects + } + + private suspend fun assertStillPending( + store: MemoryStore, + db: LinkedDatabase, + device: MemberSecrets, + entries: Int, + ) { + assertEquals("nothing was added to the roster", entries, store.roster.size) + assertEquals("nothing was sealed", emptyList(), store.objectsOf(device)) + assertNotNull("the choice is still saved", db.groups().group()?.keptDevices) + assertEquals("no place was saved in the logs", 0, db.groups().streams().size) + } + + @Test + fun a_read_the_connection_stopped_goes_on_at_the_next_sync_and_cuts_as_one_read_would() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + wrote(g, g.b, 2) + val whole = copyOf(g.store) + val (device, db) = restoredKeeping(g) + val entries = g.store.roster.size + val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.Network(IOException("dropped")) } + val sync = syncOn(db, device, server) + + assertEquals(LinkedError.UNREACHABLE, sync.syncNow().error) + assertStillPending(g.store, db, device, entries) + val asked = server.asked.toList() + + server.failing.clear() + assertNull(sync.syncNow().error) + + assertEquals("what it held isn't asked for again", emptyList(), server.asked.drop(asked.size).filter { it in asked }) + val (other, otherDb) = restoredKeeping(g, store = whole) + assertNull(syncOn(otherDb, other, whole).syncNow().error) + assertEquals( + whole.roster.last().cut, + g.store.roster + .last() + .cut, + ) + assertEquals( + setOf(id(g.a), id(g.b)), + g.store.roster + .last() + .cut + ?.keys, + ) + assertEquals( + 3L, + g.store.roster + .last() + .cut + ?.getValue(id(g.a)) + ?.seq, + ) + } + + @Test + fun a_log_that_stops_at_an_object_the_store_lacks_ends_there_and_the_removal_goes_ahead() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + val first = + g.store.objects + .getValue(paths[0]) + .second + g.store.take(paths[1]) + val (device, db) = restoredKeeping(g) + + assertNull(syncOn(db, device, g.store).syncNow().error) + + val last = g.store.roster.last() + assertEquals(Op.REMOVE, last.op) + assertEquals(1L, last.cut?.getValue(id(g.a))?.seq) + assertEquals(SyncBytes.toHex(sha256(first)), last.cut?.getValue(id(g.a))?.hash) + assertNull(db.groups().group()?.keptDevices) + } + + @Test + fun a_log_that_stops_at_a_bad_copy_ends_there_and_the_removal_goes_ahead() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + g.store.take(paths[1]) + g.store.put(paths[1], ByteArray(512)) + val (device, db) = restoredKeeping(g) + + assertNull(syncOn(db, device, g.store).syncNow().error) + + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + assertEquals( + 1L, + g.store.roster + .last() + .cut + ?.getValue(id(g.a)) + ?.seq, + ) + } + + @Test + fun an_object_a_server_answers_over_the_size_limit_holds_the_removal_until_it_has_it_right() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + val (device, db) = restoredKeeping(g) + val entries = g.store.roster.size + val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.Malformed() } + val sync = syncOn(db, device, server) + + assertNotNull(sync.syncNow().error) + assertStillPending(g.store, db, device, entries) + + server.failing.clear() + assertNull(sync.syncNow().error) + + assertEquals( + 3L, + g.store.roster + .last() + .cut + ?.getValue(id(g.a)) + ?.seq, + ) + } + + @Test + fun an_object_a_file_store_holds_over_the_size_limit_holds_the_removal_until_it_has_it_right() = + runTest { + val first = newMember() + val dbFirst = emptyDb() + val (words, _) = createWebDavGroup(dbFirst, dav.store(), first, RESTORED, LINK) + val storeFirst = GroupStore(heldStore(dbFirst), FileLayout(dav.store(), id(first))) + storeFirst.download(id(first), 0, objects = false) + val paths = Session.open(dbFirst, storeFirst, first).wrote(3) + storeFirst.upload(id(first), 0) + val genuine = checkNotNull(dav.files["flow/${paths[1]}"]) + dav.files["flow/${paths[1]}"] = ByteArray(MAX_OBJECT_SIZE + 1) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + val sync = sync(db, device) + val roster = FileLayout(dav.store(), id(device)) + + assertNotNull(sync.syncNow().error) + assertTrue(roster.rosterView().none { it.op == Op.REMOVE }) + assertEquals(emptySet(), db.groups().group()?.keptDevices) + assertTrue(dav.files.keys.none { it.startsWith("flow/dev/${id(device)}/") }) + + dav.files["flow/${paths[1]}"] = genuine + assertNull(sync.syncNow().error) + + val removal = roster.rosterView().maxBy { it.seq } + assertEquals(Op.REMOVE, removal.op) + assertEquals(3L, removal.cut?.getValue(id(first))?.seq) + assertEquals(SyncBytes.toHex(sha256(checkNotNull(dav.files["flow/${paths[2]}"]))), removal.cut?.getValue(id(first))?.hash) + } + + @Test + fun a_store_whose_pages_never_end_holds_the_restore_but_each_sync_ends() = + runTest { + val g = LinkedGroup(this).apply { create() } + val (device, db) = restoredKeeping(g) + val entries = g.store.roster.size + val sync = syncOn(db, device, EndlessPages(g.store)) + + repeat(2) { + assertEquals(LinkedError.RESTORE_UNFINISHED, sync.syncNow().error) + assertStillPending(g.store, db, device, entries) + } + } + + @Test + fun a_read_stopped_on_its_limit_removes_nothing_goes_on_and_cuts_as_one_read_would() = + runTest { + val g = LinkedGroup(this).apply { create() } + wrote(g, g.a, 3) + wrote(g, g.b, 2) + val whole = copyOf(g.store) + val (device, db) = restoredKeeping(g) + val entries = g.store.roster.size + val server = OnePerPage(g.store) + val sync = syncOn(db, device, server) { ReadBudget(requests = 2) } + + repeat(2) { + assertEquals(LinkedError.RESTORE_UNFINISHED, sync.syncNow().error) + assertStillPending(g.store, db, device, entries) + } + assertNull(sync.syncNow().error) + + assertEquals("each object was asked for once", server.asked.distinct(), server.asked) + val (other, otherDb) = restoredKeeping(g, store = whole) + assertNull(syncOn(otherDb, other, whole).syncNow().error) + val cut = + g.store.roster + .last() + .cut + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + assertEquals(whole.roster.last().cut, cut) + assertEquals(mapOf(id(g.a) to 3L, id(g.b) to 2L), cut?.mapValues { it.value.seq }) + assertNull(db.groups().group()?.keptDevices) + } + + @Test + fun a_read_of_a_file_store_stopped_on_its_limit_goes_on_at_the_next_sync() = + runTest { + val first = newMember() + val dbFirst = emptyDb() + val (words, _) = createWebDavGroup(dbFirst, dav.store(), first, RESTORED, LINK) + val storeFirst = GroupStore(heldStore(dbFirst), FileLayout(dav.store(), id(first))) + storeFirst.download(id(first), 0, objects = false) + val paths = Session.open(dbFirst, storeFirst, first).wrote(3) + storeFirst.upload(id(first), 0) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + val roster = FileLayout(dav.store(), id(device)) + val sync = syncOn(db, device, roster) { ReadBudget(requests = 2) } + + assertEquals(LinkedError.RESTORE_UNFINISHED, sync.syncNow().error) + assertTrue(roster.rosterView().none { it.op == Op.REMOVE }) + assertEquals(emptySet(), db.groups().group()?.keptDevices) + assertTrue(dav.files.keys.none { it.startsWith("flow/dev/${id(device)}/") }) + + assertNull(sync.syncNow().error) + + val removal = roster.rosterView().maxBy { it.seq } + assertEquals(Op.REMOVE, removal.op) + assertEquals(3L, removal.cut?.getValue(id(first))?.seq) + assertEquals(SyncBytes.toHex(sha256(checkNotNull(dav.files["flow/${paths[2]}"]))), removal.cut?.getValue(id(first))?.hash) + } + + @Test + fun a_sync_with_no_restore_under_way_just_goes_on_from_where_its_read_stopped() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + val db = g.deviceDb() + val sync = syncOn(db, g.b, OnePerPage(g.store)) { ReadBudget(requests = 2) } + + assertNull(sync.syncNow().error) + assertEquals( + g.store.objects + .getValue(paths[1]) + .first, + db.groups().group()?.serverCursor, + ) + + assertNull(sync.syncNow().error) + assertEquals( + g.store.objects + .getValue(paths[2]) + .first, + db.groups().group()?.serverCursor, + ) + } } diff --git a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt index b05c889f0..64164b7e3 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/ServerStoreTest.kt @@ -7,6 +7,7 @@ import io.github.aedev.flow.sync2.crypto.verify import io.github.aedev.flow.sync2.engine.LinkedException import io.github.aedev.flow.sync2.roster.Entry import io.github.aedev.flow.sync2.roster.Op +import io.github.aedev.flow.sync2.roster.canonicalEntry import io.github.aedev.flow.sync2.store.Appended import io.github.aedev.flow.sync2.store.ServerStore import io.github.aedev.flow.sync2.store.StoreException @@ -173,6 +174,21 @@ class ServerStoreTest { assertTrue(runCatching { store().changes(0) }.exceptionOrNull() is StoreException.Malformed) } + @Test + fun a_roster_past_its_limit_is_refused() = + runTest { + val g = LinkedGroup(this).apply { create() } + val entry = canonicalEntry(g.store.roster.first()) + var pages = 0 + serve { + val page = ++pages + val entries = (1..1000).map { entry.replaceFirst(Regex("\"seq\":\\d+"), "\"seq\":${(page - 1) * 1000 + it}") } + respondJson(200, if (page > 10) """{"entries":[]}""" else """{"entries":$entries}""") + } + val error = runCatching { store().rosterView() }.exceptionOrNull() + assertTrue("$error", error is StoreException.Malformed) + } + @Test fun the_status_says_when_each_device_was_last_seen() = runTest { From 2916e64ad21d3d73641fdbd52dc8b0c6adbd9ec6 Mon Sep 17 00:00:00 2001 From: AE <98699388+A-EDev@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:21:22 +0200 Subject: [PATCH 4/4] test(sync): expect a restore to cut at an item that stops only itself and wait out an outage --- .../aedev/flow/sync2/RestoreChoiceTest.kt | 172 ++++++++++++++++-- 1 file changed, 156 insertions(+), 16 deletions(-) diff --git a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt index 34887aed6..33b45ad4e 100644 --- a/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt +++ b/app/src/test/java/io/github/aedev/flow/sync2/RestoreChoiceTest.kt @@ -504,6 +504,24 @@ class RestoreChoiceTest { } } + /** The store with its chain boxes failing as [stop] says, until it's cleared. */ + private class ChainBoxFails( + private val store: MemoryStore, + ) : Remote, + ChangeList by store, + Mailbox by store { + var stop: StoreException? = null + + override suspend fun key( + epoch: Int, + kc: String, + name: String, + ): ByteArray? { + if (name == "chain") stop?.let { throw it } + return store.key(epoch, kc, name) + } + } + /** A store whose cursor always moves on: each page names one more object, which it doesn't have. */ private class EndlessPages( store: MemoryStore, @@ -654,13 +672,38 @@ class RestoreChoiceTest { } @Test - fun an_object_a_server_answers_over_the_size_limit_holds_the_removal_until_it_has_it_right() = + fun a_log_that_stops_at_an_object_a_server_answers_over_the_size_limit_ends_there() = runTest { val g = LinkedGroup(this).apply { create() } val paths = wrote(g, g.a, 3) + val first = + g.store.objects + .getValue(paths[0]) + .second + val (device, db) = restoredKeeping(g) + val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.TooLarge() } + + assertNull(syncOn(db, device, server).syncNow().error) + + val last = g.store.roster.last() + assertEquals(Op.REMOVE, last.op) + assertEquals(1L, last.cut?.getValue(id(g.a))?.seq) + assertEquals(SyncBytes.toHex(sha256(first)), last.cut?.getValue(id(g.a))?.hash) + assertNull(db.groups().group()?.keptDevices) + } + + @Test + fun a_server_error_on_an_object_holds_the_restore_until_the_store_recovers() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + val last = + g.store.objects + .getValue(paths[2]) + .second val (device, db) = restoredKeeping(g) val entries = g.store.roster.size - val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.Malformed() } + val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.Refused(503, "busy") } val sync = syncOn(db, device, server) assertNotNull(sync.syncNow().error) @@ -669,6 +712,34 @@ class RestoreChoiceTest { server.failing.clear() assertNull(sync.syncNow().error) + val removal = g.store.roster.last() + assertEquals(Op.REMOVE, removal.op) + assertEquals(3L, removal.cut?.getValue(id(g.a))?.seq) + assertEquals(SyncBytes.toHex(sha256(last)), removal.cut?.getValue(id(g.a))?.hash) + } + + @Test + fun a_server_error_on_a_chain_box_holds_the_restore_until_the_store_recovers() = + runTest { + val g = LinkedGroup(this).apply { create() } + wrote(g, g.a, 3) + val (device, db) = restoredKeeping(g) + val entries = g.store.roster.size + val server = ChainBoxFails(g.store).apply { stop = StoreException.Refused(503, "busy") } + val sync = syncOn(db, device, server) + + assertNotNull(sync.syncNow().error) + assertStillPending(g.store, db, device, entries) + + server.stop = null + assertNull(sync.syncNow().error) + + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) assertEquals( 3L, g.store.roster @@ -680,7 +751,83 @@ class RestoreChoiceTest { } @Test - fun an_object_a_file_store_holds_over_the_size_limit_holds_the_removal_until_it_has_it_right() = + fun a_refused_object_ends_its_log_there() = + runTest { + val g = LinkedGroup(this).apply { create() } + val paths = wrote(g, g.a, 3) + val (device, db) = restoredKeeping(g) + val server = OnePerPage(g.store).apply { failing[paths[1]] = StoreException.Refused(403, "refused") } + + assertNull(syncOn(db, device, server).syncNow().error) + + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + assertEquals( + 1L, + g.store.roster + .last() + .cut + ?.getValue(id(g.a)) + ?.seq, + ) + } + + @Test + fun a_refused_chain_box_or_one_over_the_limit_drops_the_older_log_with_no_cut() = + runTest { + for (stop in listOf(StoreException.Refused(403, "refused"), StoreException.TooLarge())) { + val g = LinkedGroup(this).apply { create() } + wrote(g, g.a, 3) + val (device, db) = restoredKeeping(g) + + assertNull(syncOn(db, device, ChainBoxFails(g.store).apply { this.stop = stop }).syncNow().error) + + assertEquals( + Op.REMOVE, + g.store.roster + .last() + .op, + ) + assertNull( + g.store.roster + .last() + .cut, + ) + } + } + + @Test + fun a_snapshot_blob_over_the_limit_with_its_log_collected_drops_the_log_with_no_cut() = + runTest { + val first = newMember() + val dbFirst = emptyDb() + val (words, _) = createWebDavGroup(dbFirst, dav.store(), first, RESTORED, LINK) + val storeFirst = GroupStore(heldStore(dbFirst), FileLayout(dav.store(), id(first))) + storeFirst.download(id(first), 0, objects = false) + val session = Session.open(dbFirst, storeFirst, first) + session.wrote(2) + session.writeSnapshot(export, LINK) + storeFirst.upload(id(first), 0) + dav.files.remove("flow/${StorePaths.log(id(first), 1)}") + val blob = dav.files.keys.first { it.endsWith(".b") } + dav.files[blob] = ByteArray(MAX_OBJECT_SIZE + 1) + val device = newMember() + val db = emptyDb() + restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) + + assertNull(sync(db, device).syncNow().error) + + val removal = FileLayout(dav.store(), id(device)).rosterView().maxBy { it.seq } + assertEquals(Op.REMOVE, removal.op) + assertNull(removal.cut) + } + + @Test + fun a_log_that_stops_at_an_object_a_file_store_holds_over_the_size_limit_ends_there() = runTest { val first = newMember() val dbFirst = emptyDb() @@ -689,26 +836,19 @@ class RestoreChoiceTest { storeFirst.download(id(first), 0, objects = false) val paths = Session.open(dbFirst, storeFirst, first).wrote(3) storeFirst.upload(id(first), 0) - val genuine = checkNotNull(dav.files["flow/${paths[1]}"]) + val firstObject = checkNotNull(dav.files["flow/${paths[0]}"]) dav.files["flow/${paths[1]}"] = ByteArray(MAX_OBJECT_SIZE + 1) val device = newMember() val db = emptyDb() restoreWebDavGroup(db, dav.store(), words, device, RESTORED, keep = emptySet()) - val sync = sync(db, device) - val roster = FileLayout(dav.store(), id(device)) - assertNotNull(sync.syncNow().error) - assertTrue(roster.rosterView().none { it.op == Op.REMOVE }) - assertEquals(emptySet(), db.groups().group()?.keptDevices) - assertTrue(dav.files.keys.none { it.startsWith("flow/dev/${id(device)}/") }) + assertNull(sync(db, device).syncNow().error) - dav.files["flow/${paths[1]}"] = genuine - assertNull(sync.syncNow().error) - - val removal = roster.rosterView().maxBy { it.seq } + val removal = FileLayout(dav.store(), id(device)).rosterView().maxBy { it.seq } assertEquals(Op.REMOVE, removal.op) - assertEquals(3L, removal.cut?.getValue(id(first))?.seq) - assertEquals(SyncBytes.toHex(sha256(checkNotNull(dav.files["flow/${paths[2]}"]))), removal.cut?.getValue(id(first))?.hash) + assertEquals(1L, removal.cut?.getValue(id(first))?.seq) + assertEquals(SyncBytes.toHex(sha256(firstObject)), removal.cut?.getValue(id(first))?.hash) + assertNull(db.groups().group()?.keptDevices) } @Test