Detect configs expired from the swarm and re-store them

app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt

@@ -2,7 +2,6 @@ package org.session.libsession.messaging.sending_receiving.pollers
 
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.async
-import kotlinx.coroutines.launch
 import kotlinx.coroutines.supervisorScope
 import kotlinx.coroutines.withTimeoutOrNull
 import network.loki.messenger.libsession_util.Namespace
@@ -13,6 +12,7 @@ import org.session.libsession.messaging.sending_receiving.MessageParser
 import org.session.libsession.messaging.sending_receiving.ReceivedMessageProcessor
 import org.session.libsession.network.SnodeClock
 import org.session.libsession.network.snode.SwarmDirectory
+import org.session.libsession.snode.SnodeMessage
 import org.session.libsession.snode.model.RetrieveMessageResponse
 import org.session.libsession.utilities.Address
 import org.session.libsession.utilities.Address.Companion.toAddress
@@ -31,13 +31,13 @@ import org.thoughtcrime.securesms.api.swarm.SwarmApiExecutor
 import org.thoughtcrime.securesms.api.swarm.SwarmApiRequest
 import org.thoughtcrime.securesms.api.swarm.SwarmSnodeSelector
 import org.thoughtcrime.securesms.api.swarm.execute
+import org.thoughtcrime.securesms.configs.ExpiredConfigRecovery
 import org.thoughtcrime.securesms.database.ReceivedMessageHashDatabase
 import org.thoughtcrime.securesms.preferences.PreferenceKey
 import org.thoughtcrime.securesms.preferences.PreferenceStorage
 import org.thoughtcrime.securesms.util.AppVisibilityManager
 import org.thoughtcrime.securesms.util.NetworkConnectivity
 import javax.inject.Inject
-import kotlin.time.Duration.Companion.days
 import kotlin.time.Duration.Companion.seconds
 
 class Poller @Inject constructor(
@@ -56,6 +56,7 @@ class Poller @Inject constructor(
     private val swarmSnodeSelector: SwarmSnodeSelector,
     private val swarmDirectory: SwarmDirectory,
     private val snodeApiExecutor: SnodeApiExecutor,
+    private val expiredConfigRecovery: ExpiredConfigRecovery,
     appVisibilityManager: AppVisibilityManager,
 ) : BasePoller<Unit>(
     debugLabel = "MainPoller",
@@ -140,12 +141,25 @@ class Poller @Inject constructor(
         }
     }
 
-    private fun processConfig(messages: List<RetrieveMessageResponse.Message>, forConfig: UserConfigType) {
+    /**
+     * @return whether everything fetched was taken in. A merge failure is swallowed deliberately — one
+     *  bad config message shouldn't fail the whole poll — but it does mean local state is *not* level
+     *  with the swarm, which the caller needs to know before it permits an expired-config re-store.
+     */
+    private fun processConfig(
+        messages: List<RetrieveMessageResponse.Message>,
+        forConfig: UserConfigType,
+    ): Boolean {
         if (messages.isEmpty()) {
             log("No messages to process for $forConfig")
-            return
+            return true
         }
 
+        // Note this marks each hash as SEEN, before the merge below has had a chance to fail — and the
+        // last-hash cursor advances on a successful fetch too. So "seen" is not "incorporated", and a
+        // message that fails to merge is never offered to us again by either mechanism. Anything that
+        // needs to know local state took everything in must track that separately; it cannot infer it
+        // from a later poll coming back clean, because it always will.
         val newMessages = messages
             .asSequence()
             .filterNot { msg ->
@@ -158,18 +172,30 @@ class Poller @Inject constructor(
             .map { it.toConfigMessage() }
             .toList()
 
+        var tookEverythingIn = true
+
         if (newMessages.isNotEmpty()) {
             try {
-                configFactory.mergeUserConfigs(
+                val merged = configFactory.mergeUserConfigs(
                     userConfigType = forConfig,
                     messages = newMessages
                 )
+
+                // Merging is tolerant of a message that won't parse or verify: it skips it, takes the
+                // rest, and returns normally. So a clean return is not evidence everything landed —
+                // compare the count. 2-of-3 merging is indistinguishable from 3-of-3 otherwise.
+                if (merged < newMessages.size) {
+                    logE("Only merged $merged of ${newMessages.size} messages for config $forConfig")
+                    tookEverythingIn = false
+                }
             } catch (e: Exception) {
                 logE("Error while merging user configs for $forConfig", e)
+                tookEverythingIn = false
             }
         }
 
         log("Processed ${newMessages.size} new messages for config $forConfig")
+        return tookEverythingIn
     }
 
     private fun RetrieveMessageResponse.Message.toConfigMessage(): ConfigMessage {
@@ -241,39 +267,56 @@ class Poller @Inject constructor(
                 }
         }
 
-        if (hashesToExtend.isNotEmpty()) {
-            launch {
-                try {
+        // The extension response doubles as our only signal that a config has been swept from the
+        // swarm, so keep hold of it. It's awaited after the merge below, because putting a config
+        // back before merging what we just fetched is how a long-offline device overwrites newer
+        // state with older.
+        val extendTask = hashesToExtend.takeIf { it.isNotEmpty() }?.let { hashes ->
+            async {
+                runCatching {
                     swarmApiExecutor.execute(
                         SwarmApiRequest(
                             swarmPubKeyHex = userAuth.accountId.hexString,
                             api = alterTtlApiFactory.create(
-                                messageHashes = hashesToExtend,
+                                messageHashes = hashes,
                                 auth = userAuth,
                                 alterType = AlterTtlApi.AlterType.Extend,
-                                newExpiry = snodeClock.currentTimeMillis() + 14.days.inWholeMilliseconds
+                                newExpiry = snodeClock.currentTimeMillis() + SnodeMessage.CONFIG_TTL
                             ),
                             swarmNodeOverride = snode,
                         )
                     )
-                } catch (e: Exception) {
+                }.onFailure { e ->
                     if (e is CancellationException) throw e
 
                     logE("Error while extending TTL for hashes", e)
                 }
             }
         }
 
+        // Everything above is launched before anything is awaited, and that ordering is load-bearing:
+        // requests sharing a snode coalesce into one batch inside a 100ms window (BatchApiExecutor), so a
+        // group of retrieves plus the extend go out as roughly one round-trip. Nothing enforces it —
+        // inserting an await longer than the window between those launches silently splits the batch and
+        // no test or error would show it, and the window itself is only documented two layers down.
+
         // From here, we will await on the results of pending tasks
 
+        var mergedAnyConfig = false
+        var tookEverythingIn = true
+
         // Always process the configs before the messages
         for (task in configFetchTasks) {
             val (configType, result) = task.await()
 
             val messages = result.getOrThrow().messages
-            processConfig(messages = messages, forConfig = configType)
+            if (!processConfig(messages = messages, forConfig = configType)) {
+                tookEverythingIn = false
+            }
 
             if (messages.isNotEmpty()) {
+                mergedAnyConfig = true
+
                 lokiApiDatabase.setLastMessageHashValue(
                     snode = snode,
                     publicKey = userPublicKey,
@@ -296,6 +339,37 @@ class Poller @Inject constructor(
                 namespace = Namespace.DEFAULT()
             )
         }
+
+        // Left until last: the configs above have been taken in, which is what makes it safe to put
+        // back anything the swarm has lost, and nothing else should wait on the expire response.
+        //
+        // Reached whether or not there was anything to merge, and it must stay that way — a device whose
+        // configs have expired gets nothing back, so gating this on `mergedAnyConfig` would make recovery
+        // unreachable for exactly the devices that need it.
+        //
+        // It is *not* reached when any config namespace failed to fetch, because the per-namespace
+        // `getOrThrow()` above throws out of the poll first. That's what keeps "the swarm has nothing"
+        // apart from both "nothing answered" and "some namespaces answered and some didn't" — a partial
+        // answer tells us nothing about the namespaces that stayed silent, so it must not count as
+        // level. If you ever restructure this loop to collect failures instead of throwing, that
+        // property has to be preserved deliberately: "at least one namespace answered" is not enough.
+        //
+        // A failed *merge* is the third case, and it's the one that hides: processConfig swallows those
+        // so one bad message can't fail the whole poll, which means a successful fetch is not by itself
+        // proof we took anything in. If we didn't, the swarm still holds config we haven't incorporated
+        // and we are not level with it — so say so rather than authorising a re-store.
+        if (tookEverythingIn) {
+            expiredConfigRecovery.markLocalStateLevelWithSwarm(
+                swarmPubKeyHex = userAuth.accountId.hexString,
+                mergedConfigMessagesForDiagnosticsOnly = mergedAnyConfig,
+            )
+        } else {
+            expiredConfigRecovery.markMergeIncompleteForSwarm(userAuth.accountId.hexString)
+        }
+
+        extendTask?.await()?.getOrNull()?.let { result ->
+            expiredConfigRecovery.onUserConfigsChecked(auth = userAuth, report = result.expiry)
+        }
     }
 
     private suspend fun pollInitialUserProfile() = supervisorScope {

app/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt

@@ -83,7 +83,16 @@ interface ConfigFactoryProtocol {
      */
     fun dangerouslyAccessMutableGroupConfigs(groupId: AccountId): Pair<MutableGroupConfigs, () -> Unit>
 
-    fun mergeUserConfigs(userConfigType: UserConfigType, messages: List<ConfigMessage>)
+    /**
+     * Merges [messages] into the given config.
+     *
+     * @return how many of [messages] were actually taken in. This is deliberately a count rather than a
+     *  success flag: merging is tolerant of partial failure — a message that fails to parse or verify is
+     *  skipped and the rest are merged — so "no exception" says nothing about how much was incorporated.
+     *  Anything whose correctness depends on local state being level with the swarm must compare this
+     *  against `messages.size` rather than treating a normal return as proof.
+     */
+    fun mergeUserConfigs(userConfigType: UserConfigType, messages: List<ConfigMessage>): Int
 
     /**
      * Create a new group config instance. Note this does not save the group configs to the database.
@@ -111,12 +120,18 @@ interface ConfigFactoryProtocol {
                        domain: String,
                        closedGroupSessionId: AccountId): ByteArray?
 
+    /**
+     * Merges the given group config messages.
+     *
+     * @return how many messages across all three namespaces were actually taken in — see
+     *  [mergeUserConfigs] for why this is a count and not a flag.
+     */
     fun mergeGroupConfigMessages(
         groupId: AccountId,
         keys: List<ConfigMessage>,
         info: List<ConfigMessage>,
         members: List<ConfigMessage>
-    )
+    ): Int
 
     fun confirmUserConfigsPushed(
         contacts: Pair<ConfigPush, ConfigPushResult>? = null,

app/src/main/java/org/thoughtcrime/securesms/api/snode/AlterTtlApi.kt

@@ -3,11 +3,15 @@ package org.thoughtcrime.securesms.api.snode
 import dagger.assisted.Assisted
 import dagger.assisted.AssistedFactory
 import dagger.assisted.AssistedInject
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonArray
 import kotlinx.serialization.json.JsonElement
 import kotlinx.serialization.json.JsonPrimitive
+import kotlinx.serialization.json.decodeFromJsonElement
 import org.session.libsession.network.SnodeClock
 import org.session.libsession.snode.SwarmAuth
+import org.session.libsignal.utilities.Log
 import org.thoughtcrime.securesms.api.ApiExecutorContext
 
 class AlterTtlApi @AssistedInject constructor(
@@ -17,8 +21,27 @@ class AlterTtlApi @AssistedInject constructor(
     @Assisted private val newExpiry: Long,
     errorManager: SnodeApiErrorManager,
     private val snodeClock: SnodeClock,
-) : AbstractSnodeApi<Unit>(errorManager) {
-    override fun deserializeSuccessResponse(ctx: ApiExecutorContext, body: JsonElement) {}
+    private val json: Json,
+) : AbstractSnodeApi<AlterTtlApi.Result>(errorManager) {
+    override fun deserializeSuccessResponse(ctx: ApiExecutorContext, body: JsonElement): Result {
+        // By the time we're reading this the expiries have already been altered, which is this
+        // request's actual job. Reading the response is only how we notice configs going missing, so a
+        // surprise in its shape must not turn a successful alteration into a failed request.
+        val report = runCatching {
+            val response: Response = json.decodeFromJsonElement(body)
+
+            detectMissingConfigHashes(
+                requestedHashes = messageHashes,
+                extendRequested = alterType == AlterType.Extend,
+                swarm = response.swarm,
+            )
+        }.getOrElse { e ->
+            Log.w("AlterTtlApi", "Unable to read the expire response for missing configs", e)
+            ConfigExpiryReport.Inconclusive.ResponseUnreadable
+        }
+
+        return Result(expiry = report)
+    }
 
     override val methodName: String
         get() = "expire"
@@ -48,12 +71,25 @@ class AlterTtlApi @AssistedInject constructor(
 
     }
 
+    /**
+     * @property expiry Which of the requested hashes the swarm turned out to have lost. Only an
+     *  extend request can tell us this — see [detectMissingConfigHashes].
+     */
+    data class Result(
+        val expiry: ConfigExpiryReport,
+    )
+
     enum class AlterType(val value: String) {
         Extend("extend"),
         Shorten("shorten"),
         Unspecified("")
     }
 
+    @Serializable
+    private class Response(
+        val swarm: Map<String, SnodeExpiryState> = emptyMap()
+    )
+
     @AssistedFactory
     interface Factory {
         fun create(
@@ -63,4 +99,4 @@ class AlterTtlApi @AssistedInject constructor(
             newExpiry: Long
         ): AlterTtlApi
     }
-}
+}