Keep a re-keyed shared key when its grant lands but the call fails

app/src/main/assets/pombo_bridge.html

@@ -2298,6 +2298,21 @@
         return { ok: true };
     },
 
+    // Which side of a re-key holds PUBLISH, stream by stream (web streamr.js
+    // rekeyGrantsState). Read-only; throws when the chain cannot be read.
+    async rekeyGrantsState(a) {
+        if (!client) throw new Error('client not connected');
+        const holds = (streamId, address) => client.hasPermission(
+            { streamId: streamId, userId: address.toLowerCase(), permission: 'publish', allowPublic: false });
+        const next = [];
+        const old = [];
+        for (const streamId of a.streamIds) {
+            next.push(await holds(streamId, a.newAddress));
+            old.push(a.oldAddress ? await holds(streamId, a.oldAddress) : false);
+        }
+        return { next: next, old: old };
+    },
+
     // Retention is a SECOND transaction and fails on its own. It gets its own
     // retries here, because the native caller retries this whole call and a
     // swallowed failure never reaches it. When it never lands the caller is

app/src/main/java/com/pombo/android/ChannelManager.kt

@@ -336,6 +336,30 @@ class ChannelManager(
         isPreviewChannel = { messageStreamId ->
             _channels.value.none { it.messageStreamId == messageStreamId }
         },
+        // The streams Moderation's re-key grants write, read back off the chain.
+        readRekeyGrants = { messageStreamId, interactions, newAddress, oldAddress ->
+            val channel = channelByStream(messageStreamId)
+                ?: throw IllegalStateException("Unknown channel for $messageStreamId")
+            val streamIds = (if (interactions) listOf(
+                channel.interactionsStreamId.ifEmpty { StreamConstants.deriveInteractionsId(messageStreamId) },
+                channel.ephemeralStreamId
+            ) else listOf(messageStreamId, channel.ephemeralStreamId)).filter { it.isNotEmpty() }
+            val res = bridge.call("rekeyGrantsState", JSONObject()
+                .put("streamIds", JSONArray(streamIds))
+                .put("newAddress", newAddress)
+                .put("oldAddress", oldAddress ?: JSONObject.NULL))
+            fun held(key: String) = res.getJSONArray(key).let { arr -> List(arr.length()) { arr.getBoolean(it) } }
+            com.pombo.android.core.EpochKeyManager.RekeyGrants(streamIds, held("next"), held("old"))
+        },
+        onRekeyUnsettled = { _, warning -> onGateWarning?.invoke(warning) },
+        onRekeyPending = { onLocalStateChanged() },
+        completePublishGrants = { _, newAddress, revoke, streamIds ->
+            val assignments = JSONArray().apply {
+                put(JSONObject().put("userId", newAddress).put("permissions", JSONArray(listOf("publish"))))
+                revoke.forEach { put(JSONObject().put("userId", it).put("permissions", JSONArray())) }
+            }
+            streamIds.forEach { setPermissionsRetry(it, assignments) }
+        },
         myPrivateKey = myPrivateKey,
         myUsername = myUsername,
         publishRoster = { keysStreamId, data ->

app/src/main/java/com/pombo/android/core/SyncMerge.kt

@@ -27,6 +27,8 @@ object SyncMerge {
     // a snapshot that holds nothing.
     private const val UNSTAMPED_VALUE_TS = 1L
 
+    private val PENDING_SLOTS = listOf("pubKeyPending", "intKeyPending")
+
     private fun isEmptySlice(value: Any?): Boolean = when (value) {
         null, JSONObject.NULL -> true
         is String -> value.isEmpty()
@@ -296,14 +298,30 @@ object SyncMerge {
                 if (x == null) y else if (y == null) x
                 else if (y.optInt("rev") > x.optInt("rev")) y else x
             }
-            for (slot in listOf("pubKey", "pubAnnounce", "intKey", "intAnnounce")) {
+            for (slot in listOf("pubKey", "pubAnnounce", "intKey", "intAnnounce") + PENDING_SLOTS) {
                 higherRev(b.optJSONObject(slot), i.optJSONObject(slot))?.let { entry.put(slot, it) }
             }
+            dropSettledPending(entry)
             result.put(streamId, entry)
         }
         return result
     }
 
+    /**
+     * A pending re-key travels with the keys; a key or announce at its rev or
+     * above means some device already settled it.
+     */
+    private fun dropSettledPending(entry: JSONObject) {
+        for ((pending, key, announce) in listOf(
+            Triple("pubKeyPending", "pubKey", "pubAnnounce"),
+            Triple("intKeyPending", "intKey", "intAnnounce")
+        )) {
+            val rev = entry.optJSONObject(pending)?.optInt("rev") ?: continue
+            val settled = maxOf(entry.optJSONObject(key)?.optInt("rev") ?: 0, entry.optJSONObject(announce)?.optInt("rev") ?: 0)
+            if (rev <= settled) entry.remove(pending)
+        }
+    }
+
     private fun editedAt(message: JSONObject): Long =
         (message.opt("_editedAt") as? Number)?.toLong() ?: 0L
 
@@ -418,11 +436,12 @@ object SyncMerge {
         if (incoming.optInt("currentEpoch") > local.optInt("currentEpoch")) {
             local.put("currentEpoch", incoming.optInt("currentEpoch"))
         }
-        for (field in listOf("pubKey", "pubAnnounce", "intKey", "intAnnounce")) {
+        for (field in listOf("pubKey", "pubAnnounce", "intKey", "intAnnounce") + PENDING_SLOTS) {
             val inc = incoming.optJSONObject(field) ?: continue
             val cur = local.optJSONObject(field)
             if (cur == null || inc.optInt("rev") > cur.optInt("rev")) local.put(field, inc)
         }
+        dropSettledPending(local)
         val helloEpochs = sortedSetOf<Int>()
         for (src in listOfNotNull(local.optJSONArray("helloEpochs"), incoming.optJSONArray("helloEpochs"))) {
             for (k in 0 until src.length()) src.optInt(k).takeIf { it > 0 }?.let { helloEpochs.add(it) }

app/src/main/java/com/pombo/android/core/channels/Moderation.kt

@@ -691,13 +691,12 @@ internal class Moderation(private val manager: ChannelManager) {
             "the publish key only exists on Sealed channels"
         }
         val keysId = channel.keysStreamId.ifEmpty { StreamConstants.deriveKeysId(channel.messageStreamId) }
-        return epochKeys.rekeyPublishKey(channel.messageStreamId, keysId) { newAddress, oldAddress ->
+        return epochKeys.rekeyPublishKey(channel.messageStreamId, keysId) { newAddress, revoke ->
             val assignments = JSONArray().apply {
                 // PUBLISH alone: the shared key writes, the clone reads.
                 put(JSONObject().put("userId", newAddress)
                     .put("permissions", JSONArray(listOf("publish"))))
-                if (oldAddress != null) put(JSONObject().put("userId", oldAddress)
-                    .put("permissions", JSONArray()))
+                revoke.forEach { put(JSONObject().put("userId", it).put("permissions", JSONArray())) }
             }
             setPermissionsRetry(channel.messageStreamId, assignments)
             setPermissionsRetry(channel.ephemeralStreamId, assignments)
@@ -713,12 +712,11 @@ internal class Moderation(private val manager: ChannelManager) {
         val keysId = channel.keysStreamId.ifEmpty { StreamConstants.deriveKeysId(channel.messageStreamId) }
         val interactionsId = channel.interactionsStreamId
             .ifEmpty { StreamConstants.deriveInteractionsId(channel.messageStreamId) }
-        return epochKeys.rekeyInteractionsKey(channel.messageStreamId, keysId) { newAddress, oldAddress ->
+        return epochKeys.rekeyInteractionsKey(channel.messageStreamId, keysId) { newAddress, revoke ->
             val assignments = JSONArray().apply {
                 put(JSONObject().put("userId", newAddress)
                     .put("permissions", JSONArray(listOf("publish"))))
-                if (oldAddress != null) put(JSONObject().put("userId", oldAddress)
-                    .put("permissions", JSONArray()))
+                revoke.forEach { put(JSONObject().put("userId", it).put("permissions", JSONArray())) }
             }
             setPermissionsRetry(listOf(interactionsId, channel.ephemeralStreamId), assignments)
         }

app/src/test/java/com/pombo/android/core/RekeyInteractionsTest.kt

@@ -55,21 +55,21 @@ class RekeyInteractionsTest {
 
     @Test
     fun `moves the grants first, then announces the new key at the next rev`() = runBlocking {
-        var granted: Pair<String, String?>? = null
+        var granted: Pair<String, List<String>>? = null
         val keys = manager()
 
-        val rev = keys.rekeyInteractionsKey(stream, keysStream) { next, old ->
-            events += "grants"; granted = next to old
+        val rev = keys.rekeyInteractionsKey(stream, keysStream) { next, revoke ->
+            events += "grants"; granted = next to revoke
         }
 
         assertEquals(2, rev)
         // "adopted" is what hands the new key to the account's other devices through sync.
-        assertEquals(listOf("grants", "announce", "adopted"), events)
+        assertEquals(listOf("grants", "adopted", "announce"), events)
         val ann = published.single()
         assertEquals("i", ann.getString("k"))
         assertEquals(2, ann.getInt("rev"))
         assertEquals(granted!!.first, ann.getString("addr"))
-        assertEquals(oldAddress, granted!!.second)
+        assertEquals(listOf(oldAddress), granted!!.second)
         assertNotEquals(oldAddress, ann.getString("addr"))
         assertEquals(ann.getString("keyId"), keys.interactionsKeyFor(stream)?.keyId)
     }

app/src/test/java/com/pombo/android/core/RekeyPublishKeyTest.kt

@@ -51,13 +51,13 @@ class RekeyPublishKeyTest {
     }
 
     @Test
-    fun `hands the new publish key to sync once it is announced`() = runBlocking {
+    fun `hands the new publish key to sync before announcing it`() = runBlocking {
         val keys = manager()
 
         val rev = keys.rekeyPublishKey(stream, keysStream) { _, _ -> events += "grants" }
 
         assertEquals(2, rev)
-        assertEquals(listOf("grants", "announce", "adopted"), events)
+        assertEquals(listOf("grants", "adopted", "announce"), events)
         assertEquals(keys.publishKeyFor(stream)?.keyId, adopted.single())
     }
 }