diff --git a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt b/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt index 117ed214e8..ba892c25ee 100644 --- a/app/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt +++ b/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( debugLabel = "MainPoller", @@ -140,12 +141,25 @@ class Poller @Inject constructor( } } - private fun processConfig(messages: List, 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, + 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,22 +267,26 @@ 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) @@ -264,16 +294,29 @@ class Poller @Inject constructor( } } + // 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 { diff --git a/app/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt b/app/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt index 3a58174cee..dd516a0424 100644 --- a/app/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt +++ b/app/src/main/java/org/session/libsession/utilities/ConfigFactoryProtocol.kt @@ -83,7 +83,16 @@ interface ConfigFactoryProtocol { */ fun dangerouslyAccessMutableGroupConfigs(groupId: AccountId): Pair Unit> - fun mergeUserConfigs(userConfigType: UserConfigType, messages: List) + /** + * 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): 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, info: List, members: List - ) + ): Int fun confirmUserConfigsPushed( contacts: Pair? = null, diff --git a/app/src/main/java/org/thoughtcrime/securesms/api/snode/AlterTtlApi.kt b/app/src/main/java/org/thoughtcrime/securesms/api/snode/AlterTtlApi.kt index a1efe6be25..e1a2405ee0 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/api/snode/AlterTtlApi.kt +++ b/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(errorManager) { - override fun deserializeSuccessResponse(ctx: ApiExecutorContext, body: JsonElement) {} + private val json: Json, +) : AbstractSnodeApi(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 = emptyMap() + ) + @AssistedFactory interface Factory { fun create( @@ -63,4 +99,4 @@ class AlterTtlApi @AssistedInject constructor( newExpiry: Long ): AlterTtlApi } -} \ No newline at end of file +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetection.kt b/app/src/main/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetection.kt new file mode 100644 index 0000000000..3bd4c13039 --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetection.kt @@ -0,0 +1,175 @@ +package org.thoughtcrime.securesms.api.snode + +import kotlinx.serialization.Serializable + +/** + * Interpretation of an `expire` response: which of the requested message hashes the swarm no longer + * holds. + * + * The `expire` RPC is recursive: one snode fans the request out to the whole swarm and returns a + * `swarm` dict keyed by snode pubkey, where each entry is that snode's own answer. An entry reports + * `updated` (hashes whose expiry row it actually modified) and `unchanged` (hashes it still holds + * but didn't modify), so a hash in *neither* array means that snode's database has no such message. + * + * @see detectMissingConfigHashes for the rules, which are subtle enough to be worth reading. + */ +sealed interface ConfigExpiryReport { + /** + * The response can't tell us anything, and **no hash may be treated as missing**. Every consumer + * treats these identically — the distinction is not a behavioural one, it exists so that a test can + * name *which* condition it is exercising. + * + * That is not cosmetic. When several conditions share one value, a fixture set up for one of them + * usually satisfies another as well, and the test then passes without touching the guard it names: + * two tests here did exactly that, and both went green with their guard deleted. Distinct values make + * every fixture self-isolating, because asserting the cause *is* asserting the guard ran. + */ + sealed interface Inconclusive : ConfigExpiryReport { + /** + * No extension was requested, so the server omits `unchanged` by definition and the response says + * nothing about absence whatever else it contains. + */ + data object ExtendNotRequested : Inconclusive + + /** We asked about no hashes, so there is nothing the answer could be about. */ + data object NothingAsked : Inconclusive + + /** + * Every sub-response was unusable — each one either `failed` or omitted `unchanged`. Distinct from + * [ExtendNotRequested] because here we *did* ask and the swarm still told us nothing. + */ + data object NoUsableSubResponse : Inconclusive + + /** + * The response body couldn't be decoded at all, so detection never ran. Deliberately *not* folded + * into [NoUsableSubResponse]: that one means the swarm answered and told us nothing, this one means + * we failed to read what it said. Collapsing the two would reintroduce, in the type meant to + * prevent it, exactly the ambiguity this split exists to remove. + */ + data object ResponseUnreadable : Inconclusive + } + + /** At least one snode gave a usable answer. [missingHashes] may be empty, meaning all healthy. */ + data class Checked(val missingHashes: Set) : ConfigExpiryReport +} + +/** + * One snode's answer within a recursive `expire` response. + * + * [unchanged] is nullable *and that matters*: the server only includes the key at all when the + * request set `extend` or `shorten`, so an absent key means "this response can't be used for + * detection", whereas a present-but-empty one means "I modified everything I hold". + */ +@Serializable +class SnodeExpiryState( + val failed: Boolean = false, + val updated: List = emptyList(), + val unchanged: Map? = null, +) + +/** + * Whether an expiry check is authoritative about a group having expired, and if so what it says. + * + * The group keys config decides this on its own — info or members going missing drives a re-store, never + * the banner. + * + * **The rule is not "every keys hash is missing".** It is *every keys hash is missing **and** this device + * cannot put them back*, and those are one rule rather than a rule plus an override. libsession retains the + * raw bytes of the keys messages it has loaded, and re-storing those bytes lands on the same hash without + * being re-signed — so a device holding them, admin or member, is looking at a group it can repair rather + * than an expired one. Flagging it would raise a banner that is false at the moment it appears. + * + * Which is why [canRepairKeys] is an input here and not a check the caller applies to the answer. Returning + * "expired" for a group this device can repair, and leaving a second site to know better, would make the + * value mean something other than its name — and no single test could pin the rule, because half of it would + * live somewhere else. + * + * The result is deliberately three-valued, because this check does not supersede the existing "we + * merged config messages and ended up with no keys at all" one — they answer different questions: + * + * - `true` / `false` — the group is beyond this device's reach, or it is not. Detection wins. + * - `null` — detection has nothing to say, so the existing check decides. Either no eligible snode + * answered, or the device held no keys hashes to ask about in the first place, in which case no + * request was even sent. Silence here is *not* "nothing is missing". + * + * @param keysHashes the group keys hashes that were requested — kept separate from the info and + * members hashes on purpose, since a flat union of the three can't be attributed back. + * @param canRepairKeys whether this device holds the bytes of the keys messages it asked about, so it could + * re-store them. A fact rather than a collaborator: the caller does the asking, this rule does the deciding, + * and detection gains no dependency on recovery machinery. + * + * A lambda so it is only consulted once the guards above have passed — answering it means taking the config + * lock, and on the overwhelmingly common inconclusive or all-present poll the answer cannot change the + * outcome. Keeping the ordering inside the rule is what stops the caller re-deriving "is this conclusive" + * in order to avoid the cost, which would put half the rule back at the call site. + */ +fun groupExpiredFromExpiryCheck( + report: ConfigExpiryReport?, + keysHashes: Set, + canRepairKeys: () -> Boolean, +): Boolean? { + if (report !is ConfigExpiryReport.Checked || keysHashes.isEmpty()) { + return null + } + + if (!keysHashes.all { it in report.missingHashes }) { + return false + } + + // Every keys hash is gone from the swarm. Whether that makes the group expired is a separate + // question, and it is this one: expired means nobody here can put them back. + return !canRepairKeys() +} + +/** + * Works out which of [requestedHashes] the swarm has lost, given the per-snode answers in + * [swarm]. + * + * The rules, in the order they bite: + * + * 1. Detection only works on a request that asked to extend. Without `extend` (or `shorten`) the + * server omits `unchanged` entirely, so every hash it didn't touch would look absent — which for + * a healthy config is *all of them*. A group member is the dangerous case: their subaccount + * lacks delete access, so the server forces extend-only semantics on the update while still + * omitting `unchanged`. + * 2. A sub-response carrying `failed` contributes nothing at all — not evidence of presence, not of + * absence. Reading a timeout as "that snode doesn't have it" would turn every network blip into + * a re-push storm. If nothing is left to read, the answer is + * [ConfigExpiryReport.Inconclusive.NoUsableSubResponse]. + * 3. One usable snode reporting a hash as absent is enough to call it missing; agreement is not + * required. Re-storing is idempotent so a false positive costs a request, whereas waiting for + * consensus would lean on the swarm replication that is itself the unreliable part. + * + * Multipart configs need no special handling here — each part is a separate message with its own + * hash, so they are simply requested and judged individually. Requiring *all* parts to be present + * before calling a config healthy is the caller's job. + */ +fun detectMissingConfigHashes( + requestedHashes: Collection, + extendRequested: Boolean, + swarm: Map, +): ConfigExpiryReport { + if (!extendRequested) { + return ConfigExpiryReport.Inconclusive.ExtendNotRequested + } + + // Asking about nothing tells us nothing. The tempting short-circuit here is + // `Checked(emptySet())` — "no hashes requested, so none are missing" — which is locally reasonable + // and wrong: a *conclusive* report outranks the caller's own fallback checks, so this would make + // detection the authority for precisely the case it is supposed to defer on, and the fallback + // unreachable. All three Session clients wrote that short-circuit independently. + if (requestedHashes.isEmpty()) { + return ConfigExpiryReport.Inconclusive.NothingAsked + } + + val usable = swarm.values.filter { !it.failed && it.unchanged != null } + if (usable.isEmpty()) { + return ConfigExpiryReport.Inconclusive.NoUsableSubResponse + } + + return ConfigExpiryReport.Checked( + requestedHashes.filterTo(mutableSetOf()) { hash -> + usable.any { state -> hash !in state.updated && hash !in state.unchanged!! } + } + ) +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/configs/ConfigRestoreSource.kt b/app/src/main/java/org/thoughtcrime/securesms/configs/ConfigRestoreSource.kt new file mode 100644 index 0000000000..ad6190cacd --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/configs/ConfigRestoreSource.kt @@ -0,0 +1,273 @@ +package org.thoughtcrime.securesms.configs + +import network.loki.messenger.libsession_util.MutableConfig +import network.loki.messenger.libsession_util.Namespace +import network.loki.messenger.libsession_util.ReadableGroupKeysConfig +import network.loki.messenger.libsession_util.util.Bytes +import network.loki.messenger.libsession_util.util.ConfigPush +import org.session.libsession.utilities.ConfigFactoryProtocol +import org.session.libsession.utilities.UserConfigType +import org.session.libsession.utilities.getGroup +import org.session.libsession.utilities.withGroupConfigs +import org.session.libsession.utilities.withMutableGroupConfigs +import org.session.libsession.utilities.withMutableUserConfigs +import org.session.libsignal.utilities.AccountId +import org.session.libsignal.utilities.Log +import javax.inject.Inject +import javax.inject.Singleton + +private const val TAG = "ConfigRestoreSource" + +/** One config's worth of messages to put back on the swarm. */ +class PendingRestore( + val label: String, + val push: ConfigPush, + + /** + * Every hash this restore accounts for, which for a multipart config is *all* of its parts and + * not just the missing one — we re-store the whole config, so a later poll reporting a different + * part missing is already covered. + */ + val claimedHashes: Set, + + /** + * Whether this restore is a group's *keys* config. Recovery needs to know, because a successful keys + * re-store is what clears the expired-group banner — and it has to be a typed fact rather than a match + * on [label], which is a human-readable string nobody has promised to keep stable. + */ + val isGroupKeys: Boolean = false, + + namespace: () -> Int, +) { + /** + * Resolved lazily because libsession's [Namespace] is a native class whose initialiser loads the + * shared library. Keeping it out of construction is what lets the guards that decide whether a + * config is restorable at all be covered by plain JVM unit tests. + */ + val namespace: Int by lazy(namespace) +} + +/** + * Whether a config the swarm has apparently lost should be put back. + * + * Two conditions, both of which exist to stop recovery becoming a way to *change* state: + * + * - The device must still consider one of the missing hashes current. A hash that has dropped out of + * [activeHashes] has been superseded locally, so re-storing it would resurrect state we've already + * moved on from. + * - The config must be clean. Recovery re-uploads existing state and never creates new state; a config + * with changes of its own is already on its way up via [ConfigUploader]. + * + * The clean check carries more weight than that on its own suggests, for a reason that isn't visible + * from here: a config dumped while *dirty* is reloaded as a mutable message with no trusted + * signature, so it loses the signature it was received with. For a group member — who has no signing + * key to make a new one — the bytes would then no longer reproduce the original hash. Recovery only + * ever touches clean configs, so the property it depends on holds exactly where it runs. + * + * **How reachable the clean check is, since it looks redundant and mostly is.** Dirtying a config moves + * its current hashes into the *old* set and clears them (libsession `base.cpp`, `set_state`), so a dirty + * config's [activeHashes] usually no longer contains anything the swarm reported missing — and the first + * condition rejects it before this one is consulted. + * + * It is **not** wholly redundant, though, and the exception is the reason to keep it: [activeHashes] is + * current hashes *plus the parts of any pending multipart set* that is neither done nor expired, and that + * second component survives dirtying. So a config that went dirty while a multipart set was still + * arriving, one of whose part hashes the swarm has lost, reaches this check with a non-empty + * intersection. Rare, and exactly the case where re-uploading would fight the uploader. + * + * That reachability rests on libsession's behaviour rather than ours, and **cannot be asserted here**: it + * would need `activeHashes()` to run against the real native library, which no JVM unit test in this + * project can load. The tests below reach this branch through a mocked config, which can present + * dirty-with-intersecting-hashes freely. So do not delete this check on the grounds that no test drives + * it from a realistic state, and do not delete the tests on the grounds the check looks unreachable. + */ +internal fun shouldRestore( + label: String, + activeHashes: Set, + needsPush: Boolean, + missingHashes: Set, +): Boolean { + if (missingHashes.none { it in activeHashes }) { + return false + } + + if (needsPush) { + Log.d(TAG, "Skipping recovery of $label: it has changes pending") + return false + } + + return true +} + +/** + * Turns "the swarm has lost these hashes" into the specific config messages worth re-uploading, + * applying the guards in [shouldRestore] plus the group-specific ones. + */ +@Singleton +class ConfigRestoreSource @Inject constructor( + private val configFactory: ConfigFactoryProtocol, +) { + fun userConfigsToRestore(missingHashes: Set): List { + return configFactory.withMutableUserConfigs { configs -> + UserConfigType.entries.mapNotNull { type -> + configs.getConfig(type).toRestore( + label = "user config $type", + missingHashes = missingHashes, + namespace = { type.namespace }, + ) + } + } + } + + fun groupConfigsToRestore( + groupId: AccountId, + missingHashes: Set, + ): List { + val group = configFactory.getGroup(groupId) + + // Re-storing is impossible for these anyway — the credentials were cleared and the subaccount + // token revoked — so trying only generates auth failures. + if (group == null || group.kicked || group.destroyed) { + Log.d(TAG, "Not recovering configs for a group we're no longer in") + return emptyList() + } + + return configFactory.withMutableGroupConfigs(groupId) { configs -> + // Any member can re-store these, admin or not — and that is the point, because a group + // whose admins have gone quiet is exactly the group whose configs expire. A member holds + // info and members read-only, but a read-only config re-emits the signature it received + // verbatim, and that signature survives the dump round trip, so the bytes it produces are + // identical to the admin's. Do not gate this on adminKey. + // + // What *is* admin-only is the prune below: libsession never hands a read-only + // config its obsolete hashes, and a member subaccount has no Delete access anyway. Those + // two facts cancel rather than compound — a member re-stores and simply never prunes. + // + // Keys are recoverable too, and NOT via an admin path: libsession retains the raw bytes of + // every keys message this device has *loaded*, and pushing those bytes back lands on the same + // hash without being re-signed. A member holding them can repair the group; an admin + // immediately after its own rekey holds nothing for the message it just created and is the + // device *least* able to. An admin rekey is the remedy only when no device anywhere still + // holds the bytes — which is what the banner is for. + listOfNotNull( + configs.groupInfo.toRestore( + label = "group info for $groupId", + missingHashes = missingHashes, + namespace = { Namespace.GROUP_INFO() }, + ), + configs.groupMembers.toRestore( + label = "group members for $groupId", + missingHashes = missingHashes, + namespace = { Namespace.GROUP_MEMBERS() }, + ), + configs.groupKeys.keysToRestore( + label = "group keys for $groupId", + missingHashes = missingHashes, + ), + ) + } + } + + /** + * Whether this device could put back a group's keys, and if so the bytes to send. + * + * The one predicate behind both the expired-group flag and the re-store itself, shared rather than + * duplicated on purpose: if the two ever disagreed in the direction "flag says repairable, recovery + * declines", the banner would never appear AND nothing would be fixed — silently, and for good. + * + * Returns null when no hash the swarm has lost is one we hold bytes for. Holding bytes for messages + * that are all still present is not a reason to write anything. + */ + private fun ReadableGroupKeysConfig.retainedKeysCovering( + missingHashes: Set, + ): Map? { + val retained = activeKeyMessages() + return retained.takeIf { missingHashes.any { hash -> hash in it.keys } } + } + + /** + * Whether the expired-group banner should be withheld for [groupId] because this device can repair it. + * + * Deliberately **not** gated on the things that gate the *action* — foreground, backoff, being level + * with the swarm. Those decide whether to write now; this decides whether the group is beyond reach at + * all, and a group whose repair is merely deferred until the app is foregrounded is not expired. Gating + * this on them would raise a banner that a later poll takes away, which is the flicker v119(a) exists + * to avoid. + */ + fun canRepairGroupKeys(groupId: AccountId, missingHashes: Set): Boolean { + val group = configFactory.getGroup(groupId) + if (group == null || group.kicked || group.destroyed) { + // No credentials and a revoked subaccount token: the bytes are irrelevant, we cannot store. + return false + } + + return configFactory.withGroupConfigs(groupId) { configs -> + configs.groupKeys.retainedKeysCovering(missingHashes) != null + } + } + + /** + * The keys equivalent of [toRestore], and deliberately not the same function. + * + * **Every retained message goes back, not just the missing ones.** A generation is a rekey plus every + * supplemental issued against it, and a member who receives only part of a generation cannot derive the + * key — so a partial re-store is worse than none. The retained set is not keyed by generation and + * carries no generation field, so grouping is not expressible here; re-storing all of it is a strict + * superset of "the affected generation" and therefore satisfies the requirement a fortiori. It is + * bounded (retention follows the key expiry) and idempotent, since each message is byte-identical to + * what the swarm already had and lands on the same hash. + * + * None of the [shouldRestore] guards apply. There is no `needsPush` for keys — a pending rekey is a + * *new* message, not a re-store of an old one, and it goes out through the uploader. There is no + * obsolete-hash list either, so nothing is pruned and no delete is issued. + * + * Returns null when this device holds no bytes for any missing hash, which is the case the + * expired-group banner exists for. + */ + private fun ReadableGroupKeysConfig.keysToRestore( + label: String, + missingHashes: Set, + ): PendingRestore? { + val retained = retainedKeysCovering(missingHashes) ?: return null + + Log.d(TAG, "Restoring all ${retained.size} retained keys message(s) for $label") + + return PendingRestore( + label = label, + push = ConfigPush( + messages = retained.values.map { Bytes(it) }, + seqNo = 0L, + obsoleteHashes = emptyList(), + ), + claimedHashes = retained.keys, + isGroupKeys = true, + namespace = { Namespace.GROUP_KEYS() }, + ) + } + + /** + * [namespace] stays a lambda all the way into [PendingRestore] — see the note there. Note that a + * bound reference such as `Namespace::GROUP_INFO` would defeat the point: it resolves its receiver + * eagerly, loading the class at the point the reference is created. + */ + private fun MutableConfig.toRestore( + label: String, + missingHashes: Set, + namespace: () -> Int, + ): PendingRestore? { + // Read the hashes before pushing: push() marks pending multipart sets as done, which can drop + // them back out of activeHashes. + val active = activeHashes().toSet() + + if (!shouldRestore(label, active, needsPush(), missingHashes)) { + return null + } + + return PendingRestore( + label = label, + push = push(), + claimedHashes = active, + namespace = namespace, + ) + } +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecovery.kt b/app/src/main/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecovery.kt new file mode 100644 index 0000000000..0b54527c69 --- /dev/null +++ b/app/src/main/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecovery.kt @@ -0,0 +1,570 @@ +package org.thoughtcrime.securesms.configs + +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.async +import kotlinx.coroutines.awaitAll +import kotlinx.coroutines.coroutineScope +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import org.session.libsession.network.SnodeClock +import org.session.libsession.snode.SnodeMessage +import org.session.libsession.snode.SwarmAuth +import org.session.libsignal.utilities.AccountId +import org.session.libsignal.utilities.Base64 +import org.session.libsignal.utilities.Log +import org.session.libsignal.utilities.retryWithUniformInterval +import org.thoughtcrime.securesms.api.snode.ConfigExpiryReport +import org.thoughtcrime.securesms.api.snode.DeleteMessageApi +import org.thoughtcrime.securesms.api.snode.StoreMessageApi +import org.thoughtcrime.securesms.api.swarm.SwarmApiExecutor +import org.thoughtcrime.securesms.api.swarm.SwarmApiRequest +import org.thoughtcrime.securesms.api.swarm.execute +import org.thoughtcrime.securesms.util.AppVisibilityManager +import java.util.Collections +import java.util.concurrent.ConcurrentHashMap +import javax.inject.Inject +import javax.inject.Singleton +import kotlin.time.Duration.Companion.hours +import kotlin.time.Duration.Companion.minutes +import kotlin.time.Duration.Companion.seconds + +private const val TAG = "ExpiredConfigRecovery" + +/** + * Most sub-requests recovery will put in one batch — and therefore also the most it has in flight at once. + * + * **Derived from the storage server's limit, deliberately not from a concurrency preference.** Requests + * sharing a snode coalesce into one batch inside a 100ms window (`BatchApiExecutor`), which has no size cap + * or chunking of its own — it flushes whatever accumulated when the deadline fires. The server then rejects + * an oversized batch **whole** rather than truncating it (`BATCH_REQUEST_MAX = 20`, `request_handler.h`; + * `parse_error` in `client_rpc_endpoints.cpp`). So an unchunked round loses *everything*, and it surfaces as + * a request failure rather than a size error — so it reads as a network problem and gets retried into the + * same wall. + * + * That is easy to reach: `MAX_MULTIPART_SIZE / MAX_MESSAGE_SIZE` means one config can split into ~66 parts, + * each its own store, and a round batches every config for the swarm together. + * + * **Both decode paths cap at 20 inclusive** and reject 21. They *look* like they disagree — `> MAX` on the + * JSON path, `>= MAX` on the bt path — but the bt check sits *before* its `push_back` inside the loop, so on + * the nth request it sees size n-1 and 20 requests never trip it. Reading the operators is not the same as + * reading the loop. + * + * So 20 is the real ceiling and this is set to it. An earlier version used 19 as "headroom", which does not + * survive scrutiny: the chunker controls the count exactly, so there is nothing for headroom to absorb. + * + * A concurrency limiter must never be relied on to bound this — that was the previous arrangement here and + * it held by coincidence. The boundary has to be explicit and derived from the limit, which is why this one + * number does both jobs rather than having a separate semaphore that could drift below it. + */ +private const val MAX_BATCH_SUB_REQUESTS = 20 + +/** + * How long to leave a swarm alone after its first failed recovery round, doubling per consecutive failure + * up to [RECOVERY_RETRY_BACKOFF_CEILING_MS] and resetting once anything stores. + * + * This is a **deferral, never an exclusion**, and the distinction is the whole design. A cap on attempts + * would shut out a device whose stores keep failing for the rest of the session — which is the same + * population being repaired, so a run of transient failures would cost exactly the wrong devices their + * recovery. The ceiling therefore bounds the *interval* and never the *number of attempts*: retry stays + * available indefinitely, it just gets cheaper over time. + */ +private val RECOVERY_RETRY_BACKOFF_MS = 60.seconds.inWholeMilliseconds + +/** Upper bound on the *interval* — deliberately not a bound on attempts. See above. */ +private val RECOVERY_RETRY_BACKOFF_CEILING_MS = 30.minutes.inWholeMilliseconds + +/** + * How long a successfully re-stored hash stays barred from being re-stored again. + * + * **Bounded in time rather than scoped to the session, and the reason is the TTL.** The bar exists only to + * stop a swarm that reports the same hash missing on every poll costing a store every poll — a burst + * measured in seconds. "Never again this session" is unbounded, and a session can outlive the 30-day config + * TTL: a mobile client backgrounded for a month, or a desktop client, which by design runs for weeks. In + * that window a hash we successfully put back can expire from the swarm a *second* time, and a + * session-scoped bar would block the very recovery that should put it back — excluding long-lived sessions, + * which is exactly where configs expire. + * + * **1 hour, and the figure is not load-bearing** — the property is "hours". It is still ~100x the margin the + * bar needs (polls are seconds apart, replication lag seconds to minutes) and 1/720th of the TTL, so it + * cannot interact with a genuine second expiry. + * + * Erring short rather than long because the two failure modes are asymmetric: too long re-creates the exact + * defect this bound exists to fix, while too short costs a redundant store — and a redundant re-store is + * byte-identical and idempotent, so it is one request that changes nothing. When one side of a trade costs + * correctness and the other costs a no-op request, err toward the cheap side. Standardised across the three + * clients; don't tune it as though something depended on it. + */ +private val RESTORED_HASH_BAR_MS = 1.hours.inWholeMilliseconds + +/** + * Puts config messages back on the swarm after they've been swept for exceeding their TTL. + * + * A config has a 30 day TTL, refreshed every time we poll. Go quiet for longer than that and your + * account state is deleted, so restoring from seed gives you an empty account — even though any + * device still logged in is holding a perfectly good copy. This class uploads that copy again. + * + * It is safe to do so for one specific reason, which is worth understanding because it's what makes + * the whole thing non-destructive: **config encryption is deterministic**, and the storage server + * derives a message's hash from its ciphertext alone (no timestamp, no TTL). Re-uploading an + * unchanged config therefore produces *the same message hash it had before* — it isn't a new message + * competing with existing state, it's the same message going back where it was. `store` is purely + * additive, so there is no new seqno, no merge, no fork, and nothing can be overwritten. Two devices + * recovering at once compute the same hash and the second is a no-op. + * + * The variant to never introduce here is "mark the config dirty to force an upload". That bumps the + * seqno and triggers a merge, which is exactly what this design exists to avoid, and it is what + * makes a long-offline device destructive rather than helpful. + * + * @see org.thoughtcrime.securesms.api.snode.detectMissingConfigHashes for how "missing" is decided. + * @see ConfigRestoreSource for which configs are eligible. + */ +@Singleton +class ExpiredConfigRecovery @Inject constructor( + private val restoreSource: ConfigRestoreSource, + private val clock: SnodeClock, + private val appVisibilityManager: AppVisibilityManager, + private val swarmApiExecutor: SwarmApiExecutor, + private val storeMessageApiFactory: StoreMessageApi.Factory, + private val deleteMessageApiFactory: DeleteMessageApi.Factory, +) { + /** + * Swarms our local state is known to be **level** with, this session — i.e. there is nothing on the + * swarm we haven't already taken in. + * + * That property, not "a poll happened" or "a merge happened", is what makes a re-store safe: a + * device that has taken in whatever the swarm had re-stores the incorporated result, which is + * correct by construction. Re-storing while the swarm still holds config we haven't seen is the + * dangerous ordering, and it's what this guard exists to prevent. + */ + private val swarmsLevelWithLocalState = + Collections.newSetFromMap(ConcurrentHashMap()) + + /** + * Swarms where a poll this session failed to take in everything it fetched. + * + * **Sticky for the session, and that is the point.** A config message we couldn't merge is not + * offered to us again: the dedup table marks a hash as seen before the merge is attempted, and the + * poller's `lastHash` advances on a successful *fetch*, so the swarm won't return it on the next + * poll either. So the very next poll looks completely clean while local state is still missing what + * that message carried — and without this, it would mark the swarm level and authorise a re-store. + * + * Recording it once and refusing for the rest of the session is the cheap, correct answer: recovery + * is a best-effort repair, so deferring it to the next app start costs almost nothing, whereas + * acting on a view we know to be incomplete is the thing the guard exists to prevent. + */ + private val swarmsWithIncompleteMerge = + Collections.newSetFromMap(ConcurrentHashMap()) + + /** + * Hashes claimed by a recovery round, so that a swarm reporting the same hash missing on every poll + * costs one store rather than one per poll. + * + * Note this is **not** "at most one attempt per hash per session", and anything reasoning from that + * stronger claim will be wrong. The bar is on a store that **succeeded**: a round that *fails* + * releases its claims so a later poll can retry (see [runRestore]), because otherwise the storm guard + * would be the thing making a partial upload permanent — and worse, a single transient network failure + * would cost a device its repair for the whole session. What bounds the retrying is + * [RECOVERY_RETRY_BACKOFF_MS], not this set. + * + * ⚠️ This set answers exactly one question — *"is there any point acting on this hash again?"* — and it + * is written by two different causes that happen to share that answer: a store that succeeded, and a + * hash a guard ruled out. **It is therefore not a record of what was restored**, and a future consumer + * asking that (a metric, a UI, a "did recovery help?" check) must not read it. If you need to + * distinguish them, add a second set rather than reinterpreting this one: they are answers to different + * questions that currently coincide, and one value cannot be wrong about one without being wrong about + * the other while reading correctly at both call sites. + */ + private val attemptedHashes = ConcurrentHashMap() + + /** Hashes still barred — i.e. claimed within [RESTORED_HASH_BAR_MS]. */ + private fun currentlyBarred(): Set { + val now = clock.currentTimeMillis() + // Prune as we go: an expired entry is indistinguishable from an absent one, and letting the map + // grow for the life of the process would be a slow leak on a long-lived session — the very case + // this bar was made time-bounded for. + attemptedHashes.entries.removeAll { now - it.value >= RESTORED_HASH_BAR_MS } + return attemptedHashes.keys + } + + /** Consecutive failed rounds per swarm, and when the last one was — see [RECOVERY_RETRY_BACKOFF_MS]. */ + private val backoffState = ConcurrentHashMap() + + private class BackoffState(val failedAt: Long, val consecutiveFailures: Int) { + /** Doubles per consecutive failure, capped. Bounds the wait, not the attempt count. */ + val waitMs: Long + get() = minOf( + RECOVERY_RETRY_BACKOFF_CEILING_MS, + RECOVERY_RETRY_BACKOFF_MS shl (consecutiveFailures - 1).coerceAtMost(20), + ) + } + + /** + * Records that local state is level with [swarmPubKeyHex] — call this only after a poll that + * **succeeded**, and only once whatever config it returned has been taken in. + * + * Two ways to get this wrong, both of which have bitten a Session client: + * + * Requiring a merge before re-storing reads as the careful choice and is the one thing that would + * quietly make this whole feature a no-op for the devices it exists for: a device whose configs have + * expired gets *nothing* back when it polls, so there is nothing to merge, and it would never + * recover — while every device whose configs were fine would. An empty poll establishes the property + * we need directly: nothing is on the swarm that we haven't already taken in. Hence + * [mergedConfigMessagesForDiagnosticsOnly] is logged and **must not** affect the outcome. + * + * A **failed** poll is the case that must not count — it says nothing about swarm state, so treating + * it as level reintroduces the same hazard from the other side. On this client that's structural + * rather than checked here: a failed retrieve throws (see `AutoRetryApiExecutor`, which rethrows once + * retries are exhausted), so callers never reach this line. Empty and failed are different *types* + * here, not the same value — which is the trap on clients where both arrive as an empty array. + * + * So: do not add a condition here, and do not call this from a path that can be reached on failure. + * + * @param mergedConfigMessagesForDiagnosticsOnly Logged, never acted on. The clumsy name is + * deliberate, because the obvious reading of a shorter one is that it should influence the decision + * — which is the bug. **Deleting this parameter removes the only mechanism by which anyone can + * demonstrate that this guard works**: without it, "polled but merged nothing" cannot be expressed + * in a test, so `V22 - a successful poll that merged nothing still permits recovery` collapses into + * a duplicate of the happy path and can no longer fail. A guard whose test cannot fail is precisely + * the defect this parameter exists to make impossible. Remove it as a decision, not as a tidy-up. + */ + fun markLocalStateLevelWithSwarm( + swarmPubKeyHex: String, + mergedConfigMessagesForDiagnosticsOnly: Boolean, + ) { + // An earlier poll this session already lost something we can never be offered again, so a clean + // poll now doesn't mean what it looks like it means. See [swarmsWithIncompleteMerge]. + if (swarmPubKeyHex in swarmsWithIncompleteMerge) { + return + } + + Log.d( + TAG, + "Local state is level with the swarm " + + "(merged config: $mergedConfigMessagesForDiagnosticsOnly)" + ) + swarmsLevelWithLocalState.add(swarmPubKeyHex) + } + + /** + * Records that a poll of [swarmPubKeyHex] did **not** take in everything it fetched — a merge that + * threw, or one that skipped a message it couldn't parse and returned normally. + * + * This is not merely the absence of [markLocalStateLevelWithSwarm]: it withdraws the swarm for the rest + * of the session, because the message we missed will not come back. See [swarmsWithIncompleteMerge]. + */ + fun markMergeIncompleteForSwarm(swarmPubKeyHex: String) { + Log.w(TAG, "A poll didn't take in everything it fetched; no recovery this session") + swarmsWithIncompleteMerge.add(swarmPubKeyHex) + swarmsLevelWithLocalState.remove(swarmPubKeyHex) + } + + /** + * Whether local state is known to be level with [swarmPubKeyHex] this session — the + * precondition for recovery, and the only thing that may authorise a re-store. + */ + private fun localStateIsLevelWithSwarm(swarmPubKeyHex: String): Boolean = + swarmPubKeyHex in swarmsLevelWithLocalState + + /** + * Acts on an expiry check for the current user's own configs. + * + * [auth] is the auth the poll itself used, so recovery signs as whoever noticed the problem. + */ + suspend fun onUserConfigsChecked(auth: SwarmAuth, report: ConfigExpiryReport) { + recover(auth, report, restoreSource::userConfigsToRestore) + } + + /** Acts on an expiry check for a group's configs. */ + /** + * Groups whose keys this device has just successfully put back on the swarm. + * + * The expired-group banner is otherwise cleared reactively, when a keys message is *handled* — and the + * device that did the re-storing already holds that hash, so it may never handle it again. Relying on + * the reactive path alone would leave the banner up forever over keys that are back on the swarm, which + * is the one case it must not do. + * + * `replay = 1` deliberately. The collector is started eagerly at process start, so in practice nothing + * can be emitted before it subscribes — recovery only runs from a poll. But a dropped emission here + * fails silently and permanently, while a replayed one merely clears a flag that is already clear, so + * the asymmetry decides it rather than an argument about reachability. + */ + val keysRestored: SharedFlow get() = mutableKeysRestored.asSharedFlow() + + private val mutableKeysRestored = MutableSharedFlow(replay = 1) + + /** + * Whether the group's keys are within this device's reach, for the expired-group flag. Delegated so the + * poller has one recovery-side collaborator, and so the flag and the re-store share a single predicate. + */ + fun canRepairGroupKeys(groupId: AccountId, missingHashes: Set): Boolean = + restoreSource.canRepairGroupKeys(groupId, missingHashes) + + suspend fun onGroupConfigsChecked( + groupId: AccountId, + auth: SwarmAuth, + report: ConfigExpiryReport, + ) { + recover( + auth = auth, + report = report, + gather = { missing -> restoreSource.groupConfigsToRestore(groupId, missing) }, + // Per SUCCEEDED restore, not per round: a round that stored info and members but failed on keys + // must leave the banner up. Emitting from the round would make that case pass for the wrong + // reason. + onRestored = { restore -> + if (restore.isGroupKeys) { + Log.i(TAG, "Group keys restored for $groupId; clearing any expired flag") + mutableKeysRestored.tryEmit(groupId) + } + }, + ) + } + + private suspend fun recover( + auth: SwarmAuth, + report: ConfigExpiryReport, + gather: (missingHashes: Set) -> List, + onRestored: (PendingRestore) -> Unit = {}, + ) { + val missing = (report as? ConfigExpiryReport.Checked)?.missingHashes.orEmpty() + if (missing.isEmpty()) { + return + } + + // Detection runs wherever polling runs, including in the background. Acting on it doesn't: + // recovery is N extra store requests, and the largest N is the long-offline case, which is + // also the one most likely to be running in a constrained background window. None of this is + // latency sensitive, so it can wait for the app to be open. + if (!appVisibilityManager.isAppVisible.value) { + Log.d(TAG, "Not recovering ${missing.size} config message(s) while in the background") + return + } + + if (!localStateIsLevelWithSwarm(auth.accountId.hexString)) { + Log.d(TAG, "Not recovering config messages until local state is level with this swarm") + return + } + + // Back off after a failed round rather than counting attempts. Claiming hashes on attempt is what + // stops a swarm reporting the same hash missing on every poll costing a store every poll — but on + // its own it also banks a failure as if it had succeeded, so a half-uploaded config would stay + // half-uploaded for the session on the very device the feature exists to repair. Failed rounds + // therefore release their claims (see [runRestore]), and this is what keeps that from becoming the + // storm the claiming prevented. + // + // Deliberately a rate limit and not a cap: a limit on attempts would exclude a device whose store + // keeps failing, which is the same population being repaired. + backoffState[auth.accountId.hexString]?.let { state -> + val sinceFailure = clock.currentTimeMillis() - state.failedAt + if (sinceFailure < state.waitMs) { + Log.d(TAG, "Not recovering: backing off for ${state.waitMs - sinceFailure}ms") + return + } + } + + val fresh = missing - currentlyBarred() + if (fresh.isEmpty()) { + return + } + + // An inspection that THREW is not a guard verdict, so nothing is barred and no backoff is consumed + // — the hashes stay retryable. Catching it here also keeps recovery from breaking the poll it rides + // on: `gather` runs libsession code (`push()` throws when a config has no keys), and the handoff in + // Poller.poll() sits at the end of the poll with nothing above it to catch this. + val restores = try { + gather(fresh) + } catch (e: Exception) { + if (e is CancellationException) throw e + + Log.w(TAG, "Could not inspect configs for recovery; leaving these hashes retryable", e) + return + } + + // "Not stored" is three cases, not two. A store that FAILED is retryable — that's the whole point + // of releasing claims below. But a hash a *guard* ruled out is barred like a success, because + // nothing about it changes on the timescale the bar covers: it isn't current, or its config is + // dirty and about to be pushed under a new hash anyway, or the group is gone, or it's a keys hash + // this device holds no retained bytes for — which is a fact about what we loaded, so it will not + // become true by asking again soon. + // + // Barred through the same expiring map as a successful store, deliberately: over hours a kicked + // group can be rejoined, a destroyed one replaced, a dirty config settle. "Nothing will change" + // was only ever true for a burst of polls, and re-examining a guard costs no network call — so + // there is nothing to buy by making this permanent. + // + // Folding these into "failure" costs no requests, which is exactly why it doesn't look like a + // problem — but it re-runs the gather on every poll, which takes the config write lock and re-logs + // the same rejection every few seconds for the rest of the session. + val guardRejected = fresh - restores.flatMapTo(mutableSetOf()) { it.claimedHashes } + if (guardRejected.isNotEmpty()) { + Log.d(TAG, "${guardRejected.size} hash(es) ruled out by a guard; not revisiting for now") + guardRejected.forEach { attemptedHashes[it] = clock.currentTimeMillis() } + } + + if (restores.isEmpty()) { + return + } + + // Claim the hashes before doing any work, so that concurrent polls of the same swarm — and + // every subsequent poll this session — leave them alone. + val claimedAt = clock.currentTimeMillis() + restores.forEach { r -> r.claimedHashes.forEach { attemptedHashes[it] = claimedAt } } + + Log.i(TAG, "Recovering expired configs: ${restores.joinToString { it.label }}") + + val outcomes = runRound(auth, restores) + + // Per restore that actually landed, index-aligned with [restores]. Anything keyed off the round as + // a whole would fire for a round in which this particular config failed. + outcomes.forEachIndexed { index, landed -> if (landed) onRestored(restores[index]) } + + // Decided once for the whole round rather than per config, so a mixed round can't race itself into + // an arbitrary state depending on which config finished last. + val swarm = auth.accountId.hexString + if (outcomes.any { it }) { + // Something landed, so the swarm is reachable and whatever failed deserves a prompt retry. + backoffState.remove(swarm) + } else { + val consecutive = (backoffState[swarm]?.consecutiveFailures ?: 0) + 1 + backoffState[swarm] = BackoffState(clock.currentTimeMillis(), consecutive) + Log.d(TAG, "Recovery round failed ($consecutive consecutive); backing off") + } + } + + /** + * Runs one round's stores and deletes, chunked so no batch can exceed the server's sub-request limit. + * + * @return per-restore success, index-aligned with [restores]. A restore succeeded only if *every* one of + * its messages stored: a multipart config with one bad part is not stored at all. + */ + private suspend fun runRound(auth: SwarmAuth, restores: List): List { + val succeeded = MutableList(restores.size) { true } + + val stores = restores.flatMapIndexed { index, restore -> + restore.push.messages.map { message -> Op.Store(index, restore.namespace, message.data) } + } + + // Sequential chunks, concurrent within a chunk. Awaiting a chunk fully means its batch has already + // round-tripped, so the next chunk cannot join it — that is what makes the boundary real rather + // than a hope about timing. + // + // One config's parts SPAN chunks freely, and must: at ~66 parts a large config needs four of them. + // What counts as *stored* is every part of the config — tracked per restore below — not which + // transport the parts travel in. Skipping an over-sized config instead would make anything past + // ~1.5MB permanently unrecoverable, which is the largest-accounts population the feature is for. + runChunked(auth, stores) { op, failure -> + // One chunk failing must not bar another restore's hashes, so failures are recorded against + // the owning restore rather than the round. + succeeded[op.restoreIndex] = false + Log.w(TAG, "Failed to store ${restores[op.restoreIndex].label}", failure) + } + + // Deletes are built only AFTER the stores are in, and only for configs that fully landed. + // + // `push()` has already cleared the obsolete hashes, so this is the only chance to delete them — and + // deleting them when the replacement store *failed* removes the swarm's older copy without adding + // the new one, leaving a seed restore in that window with nothing rather than stale state. + // + // Skipping them instead leaks nothing in the case that matters, and the argument is a timeline + // rather than a probability. Obsolete hashes are never TTL-extended — the extend set comes from + // `activeHashes()`, which is `_curr_hashes` plus pending multiparts and excludes `_old_hashes`. So + // with `T_old < T_cur` (a hash becomes obsolete when its successor is created) and `T_cur <= now-30d` + // (we are here because the current hash genuinely expired), `T_old + 30d < now`: **the obsolete hash + // expired first, necessarily.** The delete is a no-op precisely where recovery is legitimate. + // + // Where it is *not* a no-op is a false positive — one out-of-sync snode is enough to trigger a round + // (D1), so the obsolete hash may be recent and still live. That is the case where deleting it does + // real harm. Leaking something already gone and deleting something still there are not symmetric. + val deletes = restores.mapIndexedNotNull { index, restore -> + restore.push.obsoleteHashes + .takeIf { it.isNotEmpty() && succeeded[index] } + ?.let { Op.Delete(index, it) } + } + + runChunked(auth, deletes) { op, failure -> + // Not load bearing: the superseded messages expire on their own TTL regardless. + Log.w(TAG, "Failed to delete obsolete hashes for ${restores[op.restoreIndex].label}", failure) + } + + succeeded.forEachIndexed { index, ok -> + if (ok) { + Log.i(TAG, "Recovered ${restores[index].label}") + } else { + // Release the claims so a later round can retry. The bar applies to a hash whose store SUCCEEDED, + // not one that was merely attempted — and "succeeded" means every message's own + // sub-response was 2xx, since each is a separate execute() that throws on its own non-2xx. + // A multipart config with one bad part therefore stays retryable in full: a + // half-uploaded config can't be reconstructed, so banking it would leave it broken. + attemptedHashes.keys.removeAll(restores[index].claimedHashes) + } + } + + return succeeded + } + + /** Runs [ops] in chunks no larger than the server's sub-request limit, reporting each failure. */ + private suspend fun runChunked( + auth: SwarmAuth, + ops: List, + onFailure: (T, Throwable) -> Unit, + ) { + for (chunk in ops.chunked(MAX_BATCH_SUB_REQUESTS)) { + val outcomes = coroutineScope { + chunk.map { op -> async { runCatching { execute(auth, op) } } }.awaitAll() + } + + outcomes.forEachIndexed { position, outcome -> + val failure = outcome.exceptionOrNull() ?: return@forEachIndexed + if (failure is CancellationException) throw failure + + onFailure(chunk[position], failure) + } + } + } + + private suspend fun execute(auth: SwarmAuth, op: Op) { + retryWithUniformInterval { + when (op) { + is Op.Store -> swarmApiExecutor.execute( + SwarmApiRequest( + swarmPubKeyHex = auth.accountId.hexString, + api = storeMessageApiFactory.create( + namespace = op.namespace, + message = SnodeMessage( + auth.accountId.hexString, + Base64.encodeBytes(op.data), + SnodeMessage.CONFIG_TTL, + clock.currentTimeMillis(), + ), + auth = auth, + ) + ) + ) + + // push() hands back — and clears — libsession's obsolete hash set even for a clean config, + // so dropping these would lose messages a later real push would have deleted, permanently: + // the next push() returns an empty list. Issuing the delete keeps this behaviourally + // identical to ConfigUploader.pushConfig, minus the confirmPushed (nothing to confirm — the + // seqno never moved). + is Op.Delete -> swarmApiExecutor.execute( + SwarmApiRequest( + swarmPubKeyHex = auth.accountId.hexString, + api = deleteMessageApiFactory.create( + swarmAuth = auth, + messageHashes = op.hashes, + ) + ) + ) + } + } + } + + /** One sub-request, tagged with the restore it belongs to so a failure can be attributed. */ + private sealed interface Op { + val restoreIndex: Int + + class Store(override val restoreIndex: Int, val namespace: Int, val data: ByteArray) : Op + class Delete(override val restoreIndex: Int, val hashes: List) : Op + } +} diff --git a/app/src/main/java/org/thoughtcrime/securesms/dependencies/ConfigFactory.kt b/app/src/main/java/org/thoughtcrime/securesms/dependencies/ConfigFactory.kt index 23a45531fa..beac852e7c 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/dependencies/ConfigFactory.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/dependencies/ConfigFactory.kt @@ -271,11 +271,13 @@ class ConfigFactory @Inject constructor( override fun mergeUserConfigs( userConfigType: UserConfigType, messages: List - ) { + ): Int { if (messages.isEmpty()) { - return + return 0 } + var mergedCount = 0 + val result = doWithMutableUserConfigs(fromMerge = true) { configs -> val config = when (userConfigType) { UserConfigType.CONTACTS -> configs.contacts @@ -286,7 +288,10 @@ class ConfigFactory @Inject constructor( // Merge the list of config messages, we'll be told which messages have been merged // and we will then find out which message has the max timestamp - val maxTimestamp = config.merge(messages.map { it.hash to it.data }.toTypedArray()) + val mergedHashes = config.merge(messages.map { it.hash to it.data }.toTypedArray()) + mergedCount = mergedHashes.size + + val maxTimestamp = mergedHashes .asSequence() .mapNotNull { hash -> messages.firstOrNull { it.hash == hash } } .maxOfOrNull { it.timestamp } @@ -307,6 +312,8 @@ class ConfigFactory @Inject constructor( timestamp = timestamp ) } + + return mergedCount } override fun createGroupConfigs(groupId: AccountId, adminKey: ByteArray): MutableGroupConfigs { @@ -404,28 +411,42 @@ class ConfigFactory @Inject constructor( keys: List, info: List, members: List - ) { + ): Int { + var mergedCount = 0 + val changed = doWithMutableGroupConfigs(groupId, fromMerge = true) { configs -> - // Keys must be loaded first as they are used to decrypt the other config messages - val keysLoaded = keys.fold(false) { acc, msg -> - configs.groupKeys.loadKey(msg.data, msg.hash, msg.timestamp, configs.groupInfo.pointer, configs.groupMembers.pointer) || acc + // Keys must be loaded first as they are used to decrypt the other config messages. + // Counted rather than folded to a flag: every key is still attempted, but callers whose + // correctness depends on having taken everything in need to know how many actually landed. + val keysLoaded = keys.count { msg -> + configs.groupKeys.loadKey(msg.data, msg.hash, msg.timestamp, configs.groupInfo.pointer, configs.groupMembers.pointer) } - val infoMerged = info.isNotEmpty() && - configs.groupInfo.merge(info.map { it.hash to it.data }.toTypedArray()).isNotEmpty() + val infoMerged = if (info.isEmpty()) { + emptyList() + } else { + configs.groupInfo.merge(info.map { it.hash to it.data }.toTypedArray()) + } - val membersMerged = members.isNotEmpty() && - configs.groupMembers.merge(members.map { it.hash to it.data }.toTypedArray()).isNotEmpty() + val membersMerged = if (members.isEmpty()) { + emptyList() + } else { + configs.groupMembers.merge(members.map { it.hash to it.data }.toTypedArray()) + } configs.dumpIfNeeded(clock) - val changed = (keysLoaded || infoMerged || membersMerged) + mergedCount = keysLoaded + infoMerged.size + membersMerged.size + + val changed = keysLoaded > 0 || infoMerged.isNotEmpty() || membersMerged.isNotEmpty() changed to changed } if (changed) { configToDatabaseSync.get().syncGroupConfigs(groupId) } + + return mergedCount } override fun confirmUserConfigsPushed( diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/ExpiredGroupManager.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/ExpiredGroupManager.kt index 23a8280736..6807562b98 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/ExpiredGroupManager.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/groups/ExpiredGroupManager.kt @@ -3,11 +3,14 @@ package org.thoughtcrime.securesms.groups import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.SharingStarted import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.mapNotNull +import kotlinx.coroutines.flow.merge import kotlinx.coroutines.flow.scan import kotlinx.coroutines.flow.stateIn import org.session.libsignal.utilities.AccountId import org.session.libsignal.utilities.Log +import org.thoughtcrime.securesms.configs.ExpiredConfigRecovery import org.thoughtcrime.securesms.dependencies.ManagerScope import org.thoughtcrime.securesms.dependencies.OnAppStartupComponent import javax.inject.Inject @@ -24,11 +27,21 @@ import javax.inject.Singleton @Singleton class ExpiredGroupManager @Inject constructor( pollerManager: GroupPollerManager, + expiredConfigRecovery: ExpiredConfigRecovery, @ManagerScope scope: CoroutineScope ) : OnAppStartupComponent { @Suppress("OPT_IN_USAGE") - val expiredGroups: StateFlow> = pollerManager.watchAllGroupPollingState() - .mapNotNull { (groupId, state) -> + val expiredGroups: StateFlow> = merge( + // A successful keys re-store enters by the same door as a poll reporting "not expired", rather than + // mutating the set from outside. Two writers would race the scan and make "last known state" depend + // on call ordering instead of flow ordering; this way the add/remove and skip-null rules below stay + // the only place the set is decided. + // + // It cannot be left to the reactive path: that clears the flag when a keys message is *handled*, and + // the device that re-stored it already holds that hash, so it may never handle it again. + expiredConfigRecovery.keysRestored.map { groupId -> groupId to false }, + pollerManager.watchAllGroupPollingState() + .mapNotNull { (groupId, state) -> val expired = state.lastPolledResult?.getOrNull()?.groupExpired if (expired == null) { @@ -38,8 +51,9 @@ class ExpiredGroupManager @Inject constructor( return@mapNotNull null } - groupId to expired - } + groupId to expired + } + ) // This scan keep track of all expired groups. Whenever there is a new state for a group // poller, we compare the state with the previous state and update the set of expired groups. diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt index 4170031648..8cdff00f7c 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupPoller.kt @@ -14,6 +14,7 @@ import org.session.libsession.messaging.sending_receiving.MessageParser import org.session.libsession.messaging.sending_receiving.ReceivedMessageProcessor import org.session.libsession.messaging.sending_receiving.pollers.BasePoller import org.session.libsession.network.SnodeClock +import org.session.libsession.snode.SnodeMessage import org.session.libsession.snode.model.RetrieveMessageResponse import org.session.libsession.utilities.Address import org.session.libsession.utilities.ConfigFactoryProtocol @@ -27,15 +28,16 @@ import org.session.libsignal.utilities.AccountId import org.session.libsignal.utilities.Snode import org.thoughtcrime.securesms.api.snode.AlterTtlApi import org.thoughtcrime.securesms.api.snode.RetrieveMessageApi +import org.thoughtcrime.securesms.api.snode.groupExpiredFromExpiryCheck 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.util.AppVisibilityManager import org.thoughtcrime.securesms.util.NetworkConnectivity import kotlin.coroutines.cancellation.CancellationException -import kotlin.time.Duration.Companion.days class GroupPoller @AssistedInject constructor( @Assisted private val groupId: AccountId, @@ -51,6 +53,7 @@ class GroupPoller @AssistedInject constructor( private val alterTtlApiApiFactory: AlterTtlApi.Factory, private val swarmApiExecutor: SwarmApiExecutor, private val swarmSnodeSelector: SwarmSnodeSelector, + private val expiredConfigRecovery: ExpiredConfigRecovery, networkConnectivity: NetworkConnectivity, appVisibilityManager: AppVisibilityManager, ): BasePoller( @@ -62,6 +65,20 @@ class GroupPoller @AssistedInject constructor( val groupExpired: Boolean? ) + /** + * The active hashes of a group's three configs, kept attributed. + * + * Only [all] is sent on the wire; the individual sets exist so the response can be read back + * per-config, which is what lets the keys config alone decide whether the group is expired. + */ + private data class GroupConfigHashes( + val keys: Set, + val info: Set, + val members: Set, + ) { + val all: Set get() = keys + info + members + } + override suspend fun doPollOnce(isFirstPollSinceAppStarted: Boolean): GroupPollResult = pollSemaphore.withPermit { var groupExpired: Boolean? = null @@ -71,12 +88,16 @@ class GroupPoller @AssistedInject constructor( val groupAuth = configFactoryProtocol.getGroupAuth(groupId) ?: return@supervisorScope - val configHashesToExtends = configFactoryProtocol.withGroupConfigs(groupId) { - buildSet { - addAll(it.groupKeys.activeHashes()) - addAll(it.groupInfo.activeHashes()) - addAll(it.groupMembers.activeHashes()) - } + // Keep the three sets apart rather than merging them here: the expire response is read + // back to find out which configs the swarm has lost, and only the *keys* hashes decide + // whether the group is expired. A flat union can't be attributed, so merge only where + // the request payload is built. + val configHashes = configFactoryProtocol.withGroupConfigs(groupId) { + GroupConfigHashes( + keys = it.groupKeys.activeHashes().toSet(), + info = it.groupInfo.activeHashes().toSet(), + members = it.groupMembers.activeHashes().toSet(), + ) } val group = configFactoryProtocol.getGroup(groupId) @@ -90,8 +111,6 @@ class GroupPoller @AssistedInject constructor( log("Start polling group($groupId) message snode = ${snode.ip}") - val adminKey = group.adminKey - val pollingTasks = mutableListOf>>() val receiveRevokeMessage = async { @@ -113,22 +132,34 @@ class GroupPoller @AssistedInject constructor( ).messages } - if (configHashesToExtends.isNotEmpty() && adminKey != null) { - pollingTasks += "extending group config TTL" to async { - swarmApiExecutor.execute( - SwarmApiRequest( - swarmNodeOverride = snode, - swarmPubKeyHex = groupId.hexString, - api = alterTtlApiApiFactory.create( - messageHashes = configHashesToExtends, - auth = groupAuth, - alterType = AlterTtlApi.AlterType.Extend, - newExpiry = clock.currentTimeMillis() + 14.days.inWholeMilliseconds, + // Any member can extend, not just an admin: the request authenticates with + // `groupAuth`, and the storage server explicitly supports a member doing this. Gating + // it on an admin key meant a group whose admins went quiet lost its configs at 30 + // days while active members polled it daily. + // + // The response also doubles as our only signal that a config has been swept from the + // swarm, so keep hold of it. It's read 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: Deferred? = + if (configHashes.all.isNotEmpty()) { + async { + swarmApiExecutor.execute( + SwarmApiRequest( + swarmNodeOverride = snode, + swarmPubKeyHex = groupId.hexString, + api = alterTtlApiApiFactory.create( + messageHashes = configHashes.all, + auth = groupAuth, + alterType = AlterTtlApi.AlterType.Extend, + newExpiry = clock.currentTimeMillis() + SnodeMessage.CONFIG_TTL, + ) ) ) - ) + }.also { pollingTasks += "extending group config TTL" to it } + } else { + null } - } val groupMessageRetrieval = async { val lastHash = lokiApiDatabase.getLastMessageHashValue( @@ -183,7 +214,8 @@ class GroupPoller @AssistedInject constructor( pollingTasks += "polling and handling group config keys and messages" to async { val result = runCatching { val (keysMessage, infoMessage, membersMessage) = groupConfigRetrieval.awaitAll() - handleGroupConfigMessages(keysMessage, infoMessage, membersMessage) + val tookEverythingIn = + handleGroupConfigMessages(keysMessage, infoMessage, membersMessage) saveLastMessageHash(snode, keysMessage, Namespace.GROUP_KEYS()) saveLastMessageHash(snode, infoMessage, Namespace.GROUP_INFO()) saveLastMessageHash(snode, membersMessage, Namespace.GROUP_MEMBERS()) @@ -203,6 +235,59 @@ class GroupPoller @AssistedInject constructor( namespace = Namespace.GROUP_MESSAGES() ) } + + // 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. Its failure is already reported via pollingTasks. + val expiryReport = extendTask?.let { task -> + runCatching { task.await() }.getOrNull()?.expiry + } + + // Reached whether or not there was anything to merge, and it must stay that + // way — a group whose configs have expired returns nothing, so gating this on + // having merged something would make recovery unreachable for exactly the + // groups that need it. + // + // It is *not* reached when any of the three namespaces failed: awaitAll() + // rethrows the first failure, so a partial answer never counts as level. Nor + // when the merge threw, since handleGroupConfigMessages lets that propagate to + // the outer runCatching. + // + // `tookEverythingIn` covers the case neither of those catches: a merge that + // skips a message it can't parse and returns normally. No error, nothing to + // catch, and the swarm still holds config we haven't incorporated. + if (tookEverythingIn) { + expiredConfigRecovery.markLocalStateLevelWithSwarm( + swarmPubKeyHex = groupId.hexString, + mergedConfigMessagesForDiagnosticsOnly = keysMessage.isNotEmpty() || + infoMessage.isNotEmpty() || + membersMessage.isNotEmpty(), + ) + } else { + expiredConfigRecovery.markMergeIncompleteForSwarm(groupId.hexString) + } + + // The keys hashes alone decide this, and only when the check actually had an + // answer — otherwise the empty-keys check above stands. "Expired" now means the + // keys are gone AND this device cannot put them back, so the repairable question + // is part of the rule rather than something applied to its answer. + groupExpiredFromExpiryCheck( + report = expiryReport, + keysHashes = configHashes.keys, + canRepairKeys = { + expiredConfigRecovery.canRepairGroupKeys(groupId, configHashes.keys) + }, + )?.let { + groupExpired = it + } + + if (expiryReport != null) { + expiredConfigRecovery.onGroupConfigsChecked( + groupId = groupId, + auth = groupAuth, + report = expiryReport, + ) + } } // Revoke message must be handled regardless, and at the end @@ -265,13 +350,19 @@ class GroupPoller @AssistedInject constructor( groupRevokedMessageHandler.handleRevokeMessage(groupId, messages.map { it.data }) } + /** + * @return whether every message handed to the merge was actually taken in. Merging skips a message + * it can't parse or verify and returns normally, so a clean return is not evidence of that — the + * count has to be compared. Callers whose correctness depends on being level with the swarm need + * this; callers that just want the configs applied can ignore it. + */ private fun handleGroupConfigMessages( keysResponse: List, infoResponse: List, membersResponse: List - ) { + ): Boolean { if (keysResponse.isEmpty() && infoResponse.isEmpty() && membersResponse.isEmpty()) { - return + return true } log("Handling group config messages(" + @@ -280,12 +371,19 @@ class GroupPoller @AssistedInject constructor( "members = ${membersResponse.size})" ) - configFactoryProtocol.mergeGroupConfigMessages( + val given = keysResponse.size + infoResponse.size + membersResponse.size + val merged = configFactoryProtocol.mergeGroupConfigMessages( groupId = groupId, keys = keysResponse.map { it.toConfigMessage() }, info = infoResponse.map { it.toConfigMessage() }, members = membersResponse.map { it.toConfigMessage() }, ) + + if (merged < given) { + logE("Only merged $merged of $given group config messages") + } + + return merged == given } private fun handleMessages(messages: List) { diff --git a/app/src/test/java/org/thoughtcrime/securesms/api/snode/AlterTtlApiTest.kt b/app/src/test/java/org/thoughtcrime/securesms/api/snode/AlterTtlApiTest.kt new file mode 100644 index 0000000000..c7c2cb334d --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/api/snode/AlterTtlApiTest.kt @@ -0,0 +1,171 @@ +package org.thoughtcrime.securesms.api.snode + +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.test.runTest +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.JsonArray +import kotlinx.serialization.json.JsonObject +import kotlinx.serialization.json.JsonPrimitive +import kotlinx.serialization.json.boolean +import okhttp3.HttpUrl.Companion.toHttpUrl +import org.junit.Rule +import org.junit.Test +import org.session.libsession.network.SnodeClock +import org.session.libsession.snode.SwarmAuth +import org.session.libsignal.utilities.AccountId +import org.session.libsignal.utilities.IdPrefix +import org.session.libsignal.utilities.Snode +import org.thoughtcrime.securesms.api.ApiExecutorContext +import org.thoughtcrime.securesms.util.MockLoggingRule +import kotlin.test.assertEquals +import kotlin.test.assertNull +import kotlin.test.assertTrue + +/** + * Tests the `expire` request as it actually goes **on the wire**, and the response as it actually + * comes back — rather than what the caller passed in. + * + * That distinction is the whole point here. Three separate instances of the same bug have now been + * found across the Session clients: a flag accepted at one layer and silently dropped before the wire, + * with no error anywhere (Desktop's `extends` typo, Desktop's dead `shortenOrExtend` parameter, iOS's + * `updateExpiry` dropping `shortenOnly`/`extendOnly`). A test asserting on the caller's argument would + * have caught none of them. And `extend` reaching the server is load bearing for detection: without it + * the server omits `unchanged` entirely, and every healthy hash then reads as missing. + */ +class AlterTtlApiTest { + @get:Rule + val loggingRule = MockLoggingRule() + + private val h1 = "hash-one" + private val h2 = "hash-two" + + private val json = Json { ignoreUnknownKeys = true } + private val userId = AccountId(IdPrefix.STANDARD, ByteArray(32) { 1 }) + private val snode = Snode( + url = "https://snode.example".toHttpUrl(), + publicKeySet = Snode.KeySet("k1", "k2"), + ) + + @Test + fun `an extend request puts extend true on the wire`() { + val params = buildParams(AlterTtlApi.AlterType.Extend) + + assertEquals(true, params.bool("extend")) + assertNull(params["shorten"]) + } + + @Test + fun `a shorten request puts shorten true on the wire, and no extend`() { + val params = buildParams(AlterTtlApi.AlterType.Shorten) + + assertEquals(true, params.bool("shorten")) + assertNull(params["extend"]) + } + + @Test + fun `an unspecified request sets neither flag`() { + val params = buildParams(AlterTtlApi.AlterType.Unspecified) + + assertNull(params["extend"]) + assertNull(params["shorten"]) + } + + @Test + fun `the requested hashes and the new expiry go on the wire`() { + val params = buildParams(AlterTtlApi.AlterType.Extend) + + assertEquals( + listOf(h1, h2), + (params.getValue("messages") as JsonArray).map { (it as JsonPrimitive).content }, + ) + assertEquals("12345", (params.getValue("expiry") as JsonPrimitive).content) + } + + @Test + fun `a real expire response is read for missing hashes`() = runTest { + val report = handle( + AlterTtlApi.AlterType.Extend, + """ + { + "swarm": { + "aa": { "updated": ["$h1"], "unchanged": {}, "expiry": 12345, "signature": "sig" }, + "bb": { "updated": ["$h1", "$h2"], "unchanged": {}, "expiry": 12345, "signature": "sig" }, + "cc": { "failed": true, "timeout": true } + }, + "t": 999 + } + """.trimIndent() + ) + + // "aa" has h2 in neither array, and one eligible snode reporting absence is enough. "cc" + // contributes nothing at all. + assertEquals(ConfigExpiryReport.Checked(setOf(h2)), report) + } + + @Test + fun `a shorten response is never read for missing hashes`() = runTest { + // The body is deliberately **fully readable** — `unchanged` present and empty — so the shorten + // short-circuit is the ONLY thing that can produce Inconclusive here. With an absent `unchanged` + // key (the first version of this fixture) the response is unreadable anyway, and the test passed + // whether or not the alter type was checked at all: it would have gone green against an + // implementation that read shorten responses for absence. Read this way it discriminates — + // drop the alter-type guard and both hashes come back as Checked missing. + // + // ⚠️ Deliberately counterfactual: a real shorten response omits `unchanged`, so production trips + // both causes together and no realistic fixture can isolate either one. Since the assertion now + // names the cause, "correcting" this body to omit `unchanged` fails the test LOUDLY (it would come + // back NoUsableSubResponse) rather than passing vacuously — which is the whole point of the causes + // being distinguishable, and why the fixture and the assertion have to be read together. + val report = handle( + AlterTtlApi.AlterType.Shorten, + """{ "swarm": { "aa": { "updated": [], "unchanged": {}, "expiry": 1 } } }""" + ) + + assertEquals(ConfigExpiryReport.Inconclusive.ExtendNotRequested, report) + } + + /** + * The expiries have already been altered by the time we read the body — that is the request's + * actual job. Detection is a bonus read of the same response, so a surprise in its shape must not + * turn a successful alteration into a failed request. + */ + @Test + fun `an unreadable response degrades to inconclusive instead of failing the request`() = runTest { + val report = handle(AlterTtlApi.AlterType.Extend, """{ "swarm": "not-a-dict" }""") + + // ResponseUnreadable, not NoUsableSubResponse: detection never ran at all here. Asserting the + // specific cause is what proves the degradation happened in the decode and not somewhere inside + // the rules. + assertEquals(ConfigExpiryReport.Inconclusive.ResponseUnreadable, report) + } + + private fun api(alterType: AlterTtlApi.AlterType) = AlterTtlApi( + messageHashes = listOf(h1, h2), + auth = mockk().also { + every { it.accountId } returns userId + every { it.ed25519PublicKeyHex } returns null + every { it.sign(any()) } returns mapOf("signature" to "a-signature") + }, + alterType = alterType, + newExpiry = 12345L, + errorManager = mockk(relaxed = true), + snodeClock = mockk().also { every { it.currentTimeMillis() } returns 555L }, + json = json, + ) + + private fun buildParams(alterType: AlterTtlApi.AlterType): JsonObject { + val params = api(alterType).buildParams(ApiExecutorContext()) as JsonObject + + // Whatever else changes, the request must stay authenticated. + assertTrue("signature" in params) + return params + } + + private suspend fun handle(alterType: AlterTtlApi.AlterType, body: String): ConfigExpiryReport = + api(alterType) + .handleResponse(ApiExecutorContext(), snode, 200, Json.parseToJsonElement(body)) + .expiry + + private fun JsonObject.bool(key: String) = (getValue(key) as JsonPrimitive).boolean +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetectionTest.kt b/app/src/test/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetectionTest.kt new file mode 100644 index 0000000000..a6423fb4f6 --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/api/snode/ConfigExpiryDetectionTest.kt @@ -0,0 +1,398 @@ +package org.thoughtcrime.securesms.api.snode + +import kotlinx.serialization.Serializable +import kotlinx.serialization.json.Json +import kotlinx.serialization.json.decodeFromJsonElement +import kotlinx.serialization.json.jsonObject +import org.junit.Test +import kotlin.test.assertEquals +import kotlin.test.fail + +/** + * How an `expire` response is read: which of the hashes we asked about the swarm has lost. + * + * The V-numbers in the test names are a vocabulary shared with the other Session clients, which + * implement this detection separately from the same set of cases. Keeping the numbering aligned is what + * lets a disagreement between two clients be pinned to one specific rule rather than a whole feature; + * they are not references to anything outside this repo. + * + * V10-V13 cover the guards around the recovery action rather than the reading of the response, so + * they live in [org.thoughtcrime.securesms.configs.ExpiredConfigRecoveryTest]. + */ +class ConfigExpiryDetectionTest { + private val h1 = "hash-one" + private val h2 = "hash-two" + private val requested = listOf(h1, h2) + + @Test + fun `V1 - everything updated means nothing missing`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1, h2), unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(emptySet()), report) + } + + @Test + fun `V2 - unchanged counts as present`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1), unchanged = mapOf(h2 to 1L)), + ) + + assertEquals(ConfigExpiryReport.Checked(emptySet()), report) + } + + @Test + fun `V3 - absent from both arrays is missing`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1), unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(setOf(h2)), report) + } + + @Test + fun `V4 - one snode reporting absence is enough, even when another has it`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1), unchanged = emptyMap()), + "snodeB" to SnodeExpiryState(updated = listOf(h1, h2), unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(setOf(h2)), report) + } + + @Test + fun `V5 - a failed sub-response is excluded, not read as absence`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1, h2), unchanged = emptyMap()), + "snodeB" to SnodeExpiryState(failed = true), + ) + + assertEquals(ConfigExpiryReport.Checked(emptySet()), report) + } + + @Test + fun `V6 - all sub-responses failed is inconclusive`() { + val report = detect( + "snodeA" to SnodeExpiryState(failed = true), + "snodeB" to SnodeExpiryState(failed = true), + ) + + assertEquals(ConfigExpiryReport.Inconclusive.NoUsableSubResponse, report) + } + + /** + * V5's companion, and the only fixture that isolates the `failed` term. + * + * A sub-response is usable only if it is `!failed` **and** carries `unchanged`. Every other failed-node + * fixture — V5's and V6's — leaves `unchanged` absent, so the second term does the excluding and the + * first is never consulted: deleting `!it.failed` passes the whole suite. Verified, not assumed, by + * dropping that term alongside a control mutation known to kill two tests; the control fired and this + * hole survived it. + * + * So the fixture below is a node reporting failure that *nevertheless* carries `unchanged`. With the + * term present it is excluded and the healthy node's answer stands. Without it, the failed node's empty + * arrays are read as authority and **every requested hash is reported missing** — a false positive that + * authorises re-storing configs the swarm still holds, on the word of a snode that said it failed. + */ + @Test + fun `a failed sub-response is excluded even when it carries an unchanged map`() { + val report = detect( + "healthy" to SnodeExpiryState(updated = listOf(h1, h2), unchanged = emptyMap()), + "broken" to SnodeExpiryState(failed = true, unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(emptySet()), report) + } + + /** + * V7 and V15 are the same wire response, and the point of testing it is the distinction it draws + * with V8: `unchanged: {}` is a *valid answer* meaning "I hold none of the rest", so total loss + * must come out as "everything missing" rather than being mistaken for unavailability — which + * would disable recovery in precisely the case it exists for. + */ + @Test + fun `V7 and V15 - a snode holding nothing reports every hash missing`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = emptyList(), unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(setOf(h1, h2)), report) + } + + /** + * V8 — no extend was asked for, so the response says nothing about absence whatever it contains. + * + * The sub-response is deliberately **readable** (`unchanged` present and empty) so that the extend + * flag is the only thing that can produce Inconclusive. The first version passed `unchanged = null`, + * which is unreadable on its own (V8b) — so the test went green whether or not the extend flag was + * consulted at all, and a mutation deleting that guard left it passing. It now fails on that mutation. + * + * ⚠️ **This fixture is deliberately counterfactual and must stay that way.** A real server omits + * `unchanged` when it decides the request wasn't an extend, so production triggers *both* causes at + * once — which is exactly why the realistic fixture cannot isolate either. Restoring `unchanged = null` + * to make it "accurate" now fails this test *loudly* — it would come back `NoUsableSubResponse` where + * the assertion names `ExtendNotRequested` — which is the point of the causes being distinguishable, + * and the reason the fixture and the assertion have to be read together. The realistic shape is + * covered by V8b, whose subject *is* the unreadable response. + */ + @Test + fun `V8 - detection is unavailable without an extend request`() { + val report = detectMissingConfigHashes( + requestedHashes = requested, + extendRequested = false, + swarm = mapOf("snodeA" to SnodeExpiryState(updated = listOf(h1), unchanged = emptyMap())), + ) + + assertEquals(ConfigExpiryReport.Inconclusive.ExtendNotRequested, report) + } + + /** + * The same trap as V8, but reached the other way round: the server omits `unchanged` whenever it + * decides the request wasn't an extend, and a group member gets extend-only forced on server-side + * *while* the array is suppressed. A response we can't read must not be read as "all gone". + */ + @Test + fun `V8b - a missing unchanged key is unavailability, not absence`() { + val report = detect( + "snodeA" to SnodeExpiryState(updated = listOf(h1), unchanged = null), + ) + + assertEquals(ConfigExpiryReport.Inconclusive.NoUsableSubResponse, report) + } + + @Test + fun `V8c - an unreadable sub-response is skipped while a readable one still counts`() { + // Both ways a sub-response becomes unusable, in one fixture, because this vector's subject is + // unusable sub-responses and there are two routes to it. The first version had no `failed` node at + // all, so it exercised the readability route twice over and was insensitive to the other. + // + // Each node is shaped so its exclusion is load-bearing: + // + // - `failed` claims to hold NOTHING. Include it and h1 is reported missing too, so the verdict + // changes. A failed node that *held* the hashes would prove nothing, since one snode reporting + // absence is already enough (V4) and a node holding them cannot cancel that. + // - `noUnchanged` omits h2 from `updated`, so including it reaches the `unchanged!!` deref that + // the readability filter is what protects. + val report = detect( + "failed" to SnodeExpiryState(failed = true, updated = emptyList(), unchanged = emptyMap()), + "noUnchanged" to SnodeExpiryState(updated = listOf(h1), unchanged = null), + "readable" to SnodeExpiryState(updated = listOf(h1), unchanged = emptyMap()), + ) + + assertEquals(ConfigExpiryReport.Checked(setOf(h2)), report) + } + + @Test + fun `V9 - each part of a multipart config is judged on its own hash`() { + val report = detectMissingConfigHashes( + requestedHashes = listOf("P1", "P2", "P3"), + extendRequested = true, + swarm = mapOf( + "snodeA" to SnodeExpiryState(updated = listOf("P1", "P3"), unchanged = emptyMap()), + ), + ) + + // Only P2 is re-stored, and the config is emphatically *not* healthy just because two of its + // three parts are — a partially present multipart config decodes to nothing. + assertEquals(ConfigExpiryReport.Checked(setOf("P2")), report) + } + + /** + * The rules above are only worth anything if the wire format actually preserves the distinction + * they hinge on, so parse it rather than hand-building the states: `"unchanged": {}` means "I + * modified everything I hold", while no `unchanged` key at all means "you can't tell from this". + */ + @Test + fun `an absent unchanged key parses differently to an empty one`() { + val json = Json { ignoreUnknownKeys = true } + + val parsed: SwarmResponse = json.decodeFromJsonElement( + Json.parseToJsonElement( + """ + { + "swarm": { + "empty": { "updated": ["hash-one"], "unchanged": {}, "expiry": 123, "signature": "sig" }, + "absent": { "updated": ["hash-one"], "expiry": 123, "signature": "sig" }, + "holding": { "updated": ["hash-one"], "unchanged": { "hash-two": 456 } }, + "failed": { "failed": true, "timeout": true } + }, + "t": 1234 + } + """.trimIndent() + ).jsonObject + ) + + assertEquals(emptyMap(), parsed.swarm.getValue("empty").unchanged) + assertEquals(null, parsed.swarm.getValue("absent").unchanged) + assertEquals(mapOf(h2 to 456L), parsed.swarm.getValue("holding").unchanged) + assertEquals(true, parsed.swarm.getValue("failed").failed) + + // ...and the two together behave as V8c says they should. + assertEquals( + ConfigExpiryReport.Checked(setOf(h2)), + detectMissingConfigHashes( + requestedHashes = requested, + extendRequested = true, + swarm = parsed.swarm - "holding", + ) + ) + } + + // --- Which mechanism decides that a group has expired --- + + @Test + fun `V16 - every requested keys hash gone, and unrepairable, means the group is expired`() { + assertEquals( + true, + groupExpiredFromExpiryCheck( + ConfigExpiryReport.Checked(setOf("keys-1")), + keysHashes = setOf("keys-1"), + canRepairKeys = { false }, + ) + ) + } + + /** + * V16a — the threshold is *every* keys hash, not any. With only one keys hash V16 can't tell the + * two apart, and a device legitimately holds several: a generation is one rekey message plus N + * per-member key supplements. + * + * A surviving keys message means existing members are unaffected, and the banner's remedy isn't + * free — an admin rekey dirties info and members and bumps both seqnos — so a spurious flag causes + * spurious writes. The accepted residual: a supplement is written when a member is *added*, so it + * can outlive its generation's rekey message, and a new device that isn't one of the session ids + * that supplement encrypts to can't derive the generation while the group still reads as healthy. + */ + @Test + fun `V16a - a surviving keys hash clears the expired state even if another is gone`() { + assertEquals( + false, + groupExpiredFromExpiryCheck( + ConfigExpiryReport.Checked(setOf("keys-1")), + keysHashes = setOf("keys-1", "keys-2"), + canRepairKeys = mustNotAsk, + ) + ) + } + + @Test + fun `V19 - a missing info hash while the keys survive does not expire the group`() { + // GroupInfo's hash is gone, so it gets re-stored — but the banner is the keys config's call + // alone, and the keys hash is present. + assertEquals( + false, + groupExpiredFromExpiryCheck( + ConfigExpiryReport.Checked(setOf("info-1")), + keysHashes = setOf("keys-1"), + canRepairKeys = mustNotAsk, + ) + ) + } + + /** + * V14 — a response to a request that asked about nothing is **inconclusive**, not a conclusive + * "nothing is missing". + * + * `Checked(emptySet())` is the natural short-circuit and it is wrong in a way that hides itself: a + * conclusive report outranks the caller's fallback checks, so detection would become the authority + * for exactly the case it is meant to defer on, and the fallback would be unreachable. All three + * Session clients wrote this short-circuit independently, two of them with a test asserting the + * wrong value — so this test earns its place despite looking trivial. + */ + @Test + fun `V14 - an empty ask is inconclusive, not a conclusive nothing-missing`() { + val report = detectMissingConfigHashes( + requestedHashes = emptyList(), + extendRequested = true, + swarm = mapOf("snodeA" to SnodeExpiryState(updated = emptyList(), unchanged = emptyMap())), + ) + + assertEquals(ConfigExpiryReport.Inconclusive.NothingAsked, report) + } + + @Test + fun `V16b - holding no keys hashes leaves the expired state to the empty-fetch check`() { + // No hashes means no expire request was sent at all, so there is nothing to conclude. Null, + // not false. + assertEquals( + null, + groupExpiredFromExpiryCheck( + ConfigExpiryReport.Checked(emptySet()), + keysHashes = emptySet(), + canRepairKeys = mustNotAsk, + ) + ) + } + + /** + * Every cause, not a sampled one: the causes are distinguishable precisely so a test can name them, and + * a new cause added later must not quietly acquire the power to flag a group expired. + */ + @Test + fun `an inconclusive check leaves the expired state alone, whatever made it inconclusive`() { + val causes = listOf( + ConfigExpiryReport.Inconclusive.ExtendNotRequested, + ConfigExpiryReport.Inconclusive.NothingAsked, + ConfigExpiryReport.Inconclusive.NoUsableSubResponse, + ConfigExpiryReport.Inconclusive.ResponseUnreadable, + ) + + for (cause in causes) { + assertEquals( + null, + groupExpiredFromExpiryCheck(cause, setOf("keys-1"), canRepairKeys = mustNotAsk), + "$cause", + ) + } + } + + @Test + fun `no check at all leaves the expired state alone`() { + assertEquals(null, groupExpiredFromExpiryCheck(null, setOf("keys-1"), canRepairKeys = mustNotAsk)) + } + + /** + * V23a — every keys hash is gone from the swarm, and this device holds the bytes, so the group is **not** + * expired: it is repairable from here. + * + * This is the case the rule change exists for, and it is the only one that distinguishes the new rule + * from the old. V16 is the same wire response with the answer inverted by this one fact, which is why + * both are needed — either alone would pass against an implementation that ignored the repairable + * question entirely. + * + * The banner must be **withheld** rather than raised and later cleared: it is a visible conversation + * banner, so correcting it after the fact is a flicker on a group that was never out of reach. + */ + @Test + fun `V23a - keys all gone but held locally is repairable, not expired`() { + assertEquals( + false, + groupExpiredFromExpiryCheck( + ConfigExpiryReport.Checked(setOf("keys-1")), + keysHashes = setOf("keys-1"), + canRepairKeys = { true }, + ) + ) + } + + /** + * Answering "can this device repair the keys?" means taking the config lock, so the rule must not ask + * until its own guards have passed — otherwise every inconclusive poll pays for an answer that cannot + * change the outcome. Passing this where the question is unreachable is what pins that ordering; a + * plain Boolean parameter could not express it, because Kotlin would evaluate it before the call. + */ + private val mustNotAsk: () -> Boolean = + { fail("canRepairKeys was consulted for a report that cannot depend on it") } + + private fun detect(vararg swarm: Pair) = + detectMissingConfigHashes( + requestedHashes = requested, + extendRequested = true, + swarm = swarm.toMap(), + ) + + /** Mirrors the private response type in [AlterTtlApi]. */ + @Serializable + private class SwarmResponse(val swarm: Map = emptyMap()) +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/configs/ConfigRestoreSourceTest.kt b/app/src/test/java/org/thoughtcrime/securesms/configs/ConfigRestoreSourceTest.kt new file mode 100644 index 0000000000..7c76d62a1f --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/configs/ConfigRestoreSourceTest.kt @@ -0,0 +1,324 @@ +package org.thoughtcrime.securesms.configs + +import io.mockk.every +import io.mockk.mockk +import network.loki.messenger.libsession_util.MutableConfig +import network.loki.messenger.libsession_util.MutableGroupInfoConfig +import network.loki.messenger.libsession_util.MutableGroupKeysConfig +import network.loki.messenger.libsession_util.MutableGroupMembersConfig +import network.loki.messenger.libsession_util.ReadableUserGroupsConfig +import network.loki.messenger.libsession_util.util.Bytes +import network.loki.messenger.libsession_util.util.ConfigPush +import network.loki.messenger.libsession_util.util.GroupInfo +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.session.libsession.utilities.ConfigFactoryProtocol +import org.session.libsession.utilities.MutableGroupConfigs +import org.session.libsession.utilities.UserConfigs +import org.session.libsignal.utilities.AccountId +import org.session.libsignal.utilities.IdPrefix +import org.thoughtcrime.securesms.util.MockLoggingRule +import kotlin.test.assertEquals +import kotlin.test.assertFalse +import kotlin.test.assertTrue + +/** + * Whether a given config is eligible to be put back at all (V11 and V12), plus the [shouldRestore] + * rules those cases rest on. + * + * Everything here goes through the *group* path, and [shouldRestore] is also exercised directly. The + * user path can't be driven from a JVM unit test: reaching it touches `UserConfigType`, whose class + * initialiser resolves `Namespace` and loads the libsession native library, which no unit test in + * this project has. The rules being checked are the same ones the user path calls. + */ +class ConfigRestoreSourceTest { + @get:Rule + val loggingRule = MockLoggingRule() + + private val h1 = "hash-one" + private val h2 = "hash-two" + private val groupId = AccountId(IdPrefix.GROUP, ByteArray(32) { 2 }) + private val adminKey = Bytes(ByteArray(64) { 3 }) + + private lateinit var configFactory: ConfigFactoryProtocol + private lateinit var source: ConfigRestoreSource + + @Before + fun setUp() { + configFactory = mockk() + source = ConfigRestoreSource(configFactory) + + givenGroup(adminKey = adminKey) + givenGroupConfigs() + } + + @Test + fun `V11 - a hash the device no longer considers current must not be put back`() { + // h2 has dropped out of activeHashes: it's been superseded locally, so re-storing it would + // resurrect state we've already moved on from. + givenGroupConfigs(activeHashes = listOf(h1)) + + assertEquals(emptyList(), source.groupConfigsToRestore(groupId, setOf(h2))) + + // Reachability control. Without it this is an absence assertion satisfied by total inaction: + // break the shared stub so every config reports holding nothing, and the line above passes + // while exercising none of the rule it names. h1 IS still current, so it must produce restores + // through the same fixture. + assertEquals(2, source.groupConfigsToRestore(groupId, setOf(h1)).size) + } + + @Test + fun `V12 - a kicked group must not be put back`() { + givenGroup(adminKey = adminKey, kicked = true) + + assertEquals(emptyList(), source.groupConfigsToRestore(groupId, setOf(h2))) + } + + @Test + fun `V12b - a destroyed group must not be put back`() { + givenGroup(adminKey = adminKey, destroyed = true) + + assertEquals(emptyList(), source.groupConfigsToRestore(groupId, setOf(h2))) + } + + @Test + fun `a group that has left config entirely must not be put back`() { + givenGroup(adminKey = adminKey, present = false) + + assertEquals(emptyList(), source.groupConfigsToRestore(groupId, setOf(h2))) + } + + /** + * V20 — a non-admin member re-storing is a supported path, not a workaround, and gating it on + * admin status would remove recovery from exactly the groups that need it: a group whose admins + * have gone quiet is the group whose configs expire. A read-only config re-emits the signature it + * received verbatim and that signature survives the dump round trip, so a member's bytes are + * identical to an admin's. + */ + @Test + fun `V20 - a non-admin member re-stores group info`() { + givenGroup(adminKey = null) + + val restores = source.groupConfigsToRestore(groupId, setOf(h2)) + + assertEquals( + listOf("group info for $groupId", "group members for $groupId"), + restores.map { it.label }, + ) + } + + /** + * V21 — and a member is never handed obsolete hashes to prune, because libsession skips the + * hand-back for read-only configs (while still clearing the set). An empty list here is the + * expected result, not a failure. + */ + @Test + fun `V21 - a member re-store carries no obsolete hashes`() { + givenGroup(adminKey = null) + givenGroupConfigs(obsoleteHashes = emptyList()) + + val restores = source.groupConfigsToRestore(groupId, setOf(h2)) + + assertEquals(2, restores.size) + assertEquals(emptyList(), restores.flatMap { it.push.obsoleteHashes }) + } + + /** + * V23 — keys ARE restorable when this device holds their bytes, and a **member** can do it: the retained + * message carries the admin's signature already, so pushing it back lands on the same hash without being + * re-signed. This fixture is a non-admin deliberately. + */ + @Test + fun `V23 - a missing keys hash whose bytes are held is restorable by a member`() { + givenGroup(adminKey = null) + givenGroupConfigs(retainedKeys = mapOf("keys-1" to "keys-one".toByteArray())) + + val restores = source.groupConfigsToRestore(groupId, setOf("keys-1")) + + val keys = restores.single { it.isGroupKeys } + assertEquals(setOf("keys-1"), keys.claimedHashes) + assertEquals(listOf("keys-one"), keys.push.messages.map { String(it.data) }) + // No obsolete-hash list for keys, so nothing is ever pruned on this path. + assertEquals(emptyList(), keys.push.obsoleteHashes) + } + + /** + * V23b — a supplemental is retained and re-stored **at all**. + * + * A generation is a rekey plus every supplemental issued against it, and a member who receives only part + * of one cannot derive the key. Retention is keyed by message hash, not by generation, so all of it goes + * back whenever any of it is missing — which is a strict superset of "the affected generation" and the + * only form expressible on this API, since the retained map carries no generation field. + * + * What this pins is that supplementals are not silently dropped: it is the test that fails if someone + * "tidies" the cache to be keyed by generation, or re-stores only the hash the swarm reported. + */ + @Test + fun `V23b - every retained keys message is re-stored, supplementals included`() { + givenGroup(adminKey = null) + givenGroupConfigs( + retainedKeys = mapOf( + "rekey-1" to "the-rekey".toByteArray(), + "supplement-1" to "for-alice".toByteArray(), + "supplement-2" to "for-bob".toByteArray(), + ) + ) + + // Only ONE hash is reported missing; all three must still go back. + val keys = source.groupConfigsToRestore(groupId, setOf("supplement-1")).single { it.isGroupKeys } + + assertEquals(setOf("rekey-1", "supplement-1", "supplement-2"), keys.claimedHashes) + assertEquals(3, keys.push.messages.size) + } + + @Test + fun `canRepairGroupKeys is true only when a MISSING hash is one we hold`() { + givenGroup(adminKey = null) + givenGroupConfigs(retainedKeys = mapOf("keys-1" to "keys-one".toByteArray())) + + assertTrue(source.canRepairGroupKeys(groupId, setOf("keys-1"))) + + // Holding bytes for messages the swarm still has is not a reason to write, and must not withhold + // the banner for a group whose *other* keys are genuinely gone. + assertFalse(source.canRepairGroupKeys(groupId, setOf("keys-99"))) + } + + /** + * A kicked group cannot be repaired however many bytes we hold — the credentials are gone and the + * subaccount token is revoked, so the store would only generate auth failures. The flag must stand. + */ + @Test + fun `canRepairGroupKeys is false for a kicked group even holding the bytes`() { + givenGroup(adminKey = null, kicked = true) + givenGroupConfigs(retainedKeys = mapOf("keys-1" to "keys-one".toByteArray())) + + assertFalse(source.canRepairGroupKeys(groupId, setOf("keys-1"))) + } + + /** + * The keys config is not in the restorable set at all, and a missing keys hash goes straight to the + * expired-group flag instead. + * + * ⚠️ This is a **platform** limitation with a known expiry date, not a property of the format: libsession + * retains the bytes of active keys messages and exposes them, and once the Android wrapper binds that + * accessor a member will be able to repair a group's keys by pushing the retained bytes back. At that + * point this test inverts rather than being deleted — keys become restorable when their bytes are held, + * and the flag is only for a device that holds none. + * + * Deliberately unlabelled: V16 is the *detection* rule (all keys hashes gone ⇒ group expired) and lives + * in ConfigExpiryDetectionTest. This is the same input on the *restore* path, which is a separate + * question the vector table doesn't have a row for. It carried "V16" until a sweep found that label + * already on the detection test. + */ + @Test + fun `a missing group keys hash produces no re-store while the wrapper exposes no key bytes`() { + // The keys hash is not among any restorable config's active hashes, so nothing matches. + assertEquals(emptyList(), source.groupConfigsToRestore(groupId, setOf("keys-hash"))) + + // Reachability control, same reasoning as V11: a hash that IS restorable must produce restores + // through this fixture, or the assertion above proves only that nothing ran. + assertEquals(2, source.groupConfigsToRestore(groupId, setOf(h2)).size) + } + + /** + * Note what this does *not* establish: the mock presents dirty-with-intersecting-hashes directly, and + * a real config reaches that state only through the pending-multipart component of `activeHashes()`, + * since dirtying clears the current hashes. Driving it realistically needs the native library. See + * [shouldRestore]'s doc for why the check stays regardless. + */ + @Test + fun `a config with changes of its own is left to the uploader`() { + assertFalse( + shouldRestore("test", setOf(h1, h2), needsPush = true, missingHashes = setOf(h2)) + ) + } + + @Test + fun `a clean config still holding the missing hash is restorable`() { + assertTrue( + shouldRestore("test", setOf(h1, h2), needsPush = false, missingHashes = setOf(h2)) + ) + } + + @Test + fun `a config no longer holding the missing hash is not restorable`() { + assertFalse( + shouldRestore("test", setOf(h1), needsPush = false, missingHashes = setOf(h2)) + ) + } + + @Test + fun `one missing part is enough to restore a multipart config`() { + assertTrue( + shouldRestore( + "test", + setOf("P1", "P2", "P3"), + needsPush = false, + missingHashes = setOf("P2"), + ) + ) + } + + private fun T.withState( + activeHashes: List, + needsPush: Boolean = false, + obsoleteHashes: List = emptyList(), + ): T = apply { + every { activeHashes() } returns activeHashes + every { needsPush() } returns needsPush + every { push() } returns ConfigPush( + messages = listOf(Bytes("config-data".toByteArray())), + seqNo = 7L, + obsoleteHashes = obsoleteHashes, + ) + } + + private fun givenGroupConfigs( + activeHashes: List = listOf(h1, h2), + obsoleteHashes: List = emptyList(), + retainedKeys: Map = emptyMap(), + ) { + val configs = mockk() + every { configs.groupInfo } returns + mockk().withState(activeHashes, obsoleteHashes = obsoleteHashes) + every { configs.groupMembers } returns + mockk().withState(activeHashes, obsoleteHashes = obsoleteHashes) + every { configs.groupKeys } returns mockk(relaxed = true).also { + every { it.activeKeyMessages() } returns retainedKeys + } + + every { configFactory.dangerouslyAccessMutableGroupConfigs(groupId) } returns (configs to {}) + // canRepairGroupKeys reads through the *read-only* accessor, so it needs its own stub — the + // mutable one is not a superset here. + every { configFactory.dangerouslyAccessGroupConfigs(groupId) } returns (configs to {}) + } + + private fun givenGroup( + adminKey: Bytes?, + kicked: Boolean = false, + destroyed: Boolean = false, + present: Boolean = true, + ) { + val userGroups = mockk() + every { userGroups.getClosedGroup(groupId.hexString) } returns if (!present) { + null + } else { + GroupInfo.ClosedGroupInfo( + groupAccountId = groupId.hexString, + adminKey = adminKey, + authData = if (adminKey == null) Bytes(ByteArray(100) { 4 }) else null, + priority = 0L, + invited = false, + name = "A group", + kicked = kicked, + destroyed = destroyed, + joinedAtSecs = 0L, + ) + } + + val configs = mockk() + every { configs.userGroups } returns userGroups + every { configFactory.dangerouslyAccessUserConfigs() } returns (configs to {}) + } +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecoveryTest.kt b/app/src/test/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecoveryTest.kt new file mode 100644 index 0000000000..0075701e95 --- /dev/null +++ b/app/src/test/java/org/thoughtcrime/securesms/configs/ExpiredConfigRecoveryTest.kt @@ -0,0 +1,963 @@ +package org.thoughtcrime.securesms.configs + +import io.mockk.coEvery +import io.mockk.coVerify +import io.mockk.every +import io.mockk.mockk +import io.mockk.verify +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.test.runTest +import network.loki.messenger.libsession_util.util.Bytes +import network.loki.messenger.libsession_util.util.ConfigPush +import org.junit.Before +import org.junit.Rule +import org.junit.Test +import org.session.libsession.network.SnodeClock +import org.session.libsession.snode.SwarmAuth +import org.session.libsession.snode.model.StoreMessageResponse +import org.session.libsignal.utilities.AccountId +import org.session.libsignal.utilities.IdPrefix +import org.thoughtcrime.securesms.api.snode.ConfigExpiryReport +import org.thoughtcrime.securesms.api.snode.DeleteMessageApi +import org.thoughtcrime.securesms.api.snode.StoreMessageApi +import org.thoughtcrime.securesms.api.swarm.SwarmApiExecutor +import org.thoughtcrime.securesms.api.swarm.SwarmApiRequest +import org.thoughtcrime.securesms.util.AppVisibilityManager +import org.thoughtcrime.securesms.util.MockLoggingRule +import java.time.Instant +import java.util.concurrent.atomic.AtomicInteger +import kotlin.time.Duration.Companion.hours +import kotlin.time.Duration.Companion.minutes +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** + * The session-scoped guards around the recovery action (V10 and V13 and their variants). + * + * The other cases live where the logic they exercise does: V1-V9 in + * [org.thoughtcrime.securesms.api.snode.ConfigExpiryDetectionTest], and V11/V12 in + * [ConfigRestoreSourceTest], which owns the per-config eligibility rules. + */ +class ExpiredConfigRecoveryTest { + @get:Rule + val loggingRule = MockLoggingRule() + + private val h2 = "hash-two" + private val userId = AccountId(IdPrefix.STANDARD, ByteArray(32) { 1 }) + private val groupId = AccountId(IdPrefix.GROUP, ByteArray(32) { 2 }) + + private lateinit var restoreSource: ConfigRestoreSource + private lateinit var appVisibilityManager: AppVisibilityManager + private lateinit var swarmApiExecutor: SwarmApiExecutor + private lateinit var deleteMessageApiFactory: DeleteMessageApi.Factory + private lateinit var recovery: ExpiredConfigRecovery + + /** Store requests issued, so [assertRecoveryStillReachable] can assert an *increase*. */ + private val storeCalls = AtomicInteger(0) + + /** Mutable so a test can background the app and then restore it for the positive control. */ + private val appVisible = MutableStateFlow(true) + + /** Controllable so the retry backoff can be advanced past. */ + private var now = 0L + private lateinit var clock: SnodeClock + + @Before + fun setUp() { + restoreSource = mockk() + appVisibilityManager = mockk() + swarmApiExecutor = mockk() + deleteMessageApiFactory = mockk(relaxed = true) + + storeCalls.set(0) + appVisible.value = true + now = 0L + clock = mockk() + every { clock.currentTimeMillis() } answers { now } + every { appVisibilityManager.isAppVisible } returns appVisible + coEvery { swarmApiExecutor.send(any(), any()) } answers { + when (secondArg>().api) { + is DeleteMessageApi -> DeleteMessageApi.SuccessResponse(1, 1) + else -> { + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + } + } + + givenRestorable(claimedHashes = setOf(h2), messageCount = 1) + + recovery = ExpiredConfigRecovery( + restoreSource = restoreSource, + clock = clock, + appVisibilityManager = appVisibilityManager, + swarmApiExecutor = swarmApiExecutor, + storeMessageApiFactory = mockk(relaxed = true), + deleteMessageApiFactory = deleteMessageApiFactory, + ) + } + + @Test + fun `a missing hash on a merged swarm is put back`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(1) + } + + @Test + fun `V10 - recovery must not run before any successful poll of the swarm`() = runTest { + // No successful poll at all — which is a different thing from "polled but merged nothing", the + // case V22 covers. Detection still happened; we just must not act on it, because re-storing + // while the swarm holds config we haven't taken in can put back state that has since been + // deliberately changed. + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + @Test + fun `V10b - a successful poll of a different swarm does not unlock this one`() = runTest { + recovery.markLocalStateLevelWithSwarm( + AccountId(IdPrefix.GROUP, ByteArray(32) { 9 }).hexString, + mergedConfigMessagesForDiagnosticsOnly = true, + ) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + /** + * V22 — a successful poll that returned **no config messages** must still permit recovery, and this + * is the most consequential assertion in the file. + * + * The intuitive reading of the guard is "wait until a merge has happened". That reading makes the + * whole feature a no-op for precisely the devices it was built for: a device whose configs have + * expired gets nothing back when it polls, so there is nothing to merge, so it would never recover — + * while every device whose configs were fine would. And it fails *silently*: detection runs, the + * guard declines, nothing happens, no error is raised, and a suite that hands the recovery a merge + * in its setup stays green throughout. + * + * Asserting that recovery is skipped here would be encoding that bug, not testing for it. + */ + @Test + fun `V22 - a successful poll that merged nothing still permits recovery`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = false) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(1) + } + + /** + * V22a — **recorded as N/A-by-construction on this platform, not as passing.** Read the reason before + * treating this as coverage. + * + * The vector guards clients where a failed poll and an empty poll arrive as the *same value* (an empty + * array), so that keying off "result is empty" silently treats "nothing answered" as "we're level". + * Elsewhere V22a is the only vector that discriminates — V22 passes under the wrong implementation. + * + * That trap cannot occur here, and the reason is structural rather than tested: a failed retrieve + * **throws** (`AutoRetryApiExecutor` rethrows once retries are exhausted; `AbstractSnodeApi` throws on + * any non-2xx), so failure is an *exception* and empty is an *empty list* — different types, and the + * poller never reaches [ExpiredConfigRecovery.markLocalStateLevelWithSwarm] on the failing path. This + * client also polls a **single** snode per request, so there is no aggregate-of-many-snodes step in + * which failures could be flattened into emptiness at all. + * + * So the assertion below exercises no code that V10 doesn't already cover — it documents the hazard + * rather than testing against it, and counting it as a passing row would inflate apparent coverage. + * The thing that actually protects this property is the type distinction above; keep it that way. + */ + @Test + fun `V22a - N-A by construction - a poll that answered nothing must not permit recovery`() = runTest { + // Deliberately no markLocalStateLevelWithSwarm call: that is what a failed poll looks like here, + // because the poller throws before reaching it. + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + /** + * V22c — a poll that took in fewer messages than it fetched must not permit recovery. + * + * The tolerance itself is correct: a config message that won't parse or verify is skipped and the + * rest are merged. What's wrong is reading that silence as "everything landed" — 2-of-3 merging is + * indistinguishable from 3-of-3 unless the count is compared. + */ + @Test + fun `V22c - a poll that merged only some of what it fetched must not permit recovery`() = runTest { + recovery.markMergeIncompleteForSwarm(userId.hexString) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + /** + * V22c, the part that matters most and the one a per-poll check misses: the verdict has to be + * **sticky for the session**. + * + * A message we couldn't merge is never offered again — the dedup table marks a hash as seen before + * the merge is attempted, and the poller's `lastHash` advances on a successful *fetch* — so the very + * next poll comes back completely clean while local state is still missing what that message + * carried. A guard that only looks at the current poll therefore self-heals into the wrong answer + * one poll later, which is worse than failing outright because nothing ever looks wrong again. + */ + @Test + fun `V22d - a later clean poll must not undo an earlier incomplete merge`() = runTest { + recovery.markMergeIncompleteForSwarm(userId.hexString) + + // The next poll fetches nothing at all and looks perfectly healthy. + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = false) + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + /** + * And a swarm withdrawn this way doesn't take unrelated swarms down with it. + * + * Deliberately unlabelled: the withdrawal being per-swarm is a consequence of V22c, not a vector of its + * own. It carried "V22c" until a sweep found that label already on the test above. + */ + @Test + fun `withdrawing one swarm leaves others recoverable`() = runTest { + recovery.markMergeIncompleteForSwarm( + AccountId(IdPrefix.GROUP, ByteArray(32) { 9 }).hexString + ) + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(1) + } + + + /** + * ...but the retrying is rate-limited, or releasing the claims would reintroduce the storm. + * + * Deliberately a backoff and **not** a cap on attempts. A cap would exclude a device whose stores keep + * failing for the rest of the session — which is the same population the feature exists to repair, so + * a transient network failure would cost a device its recovery entirely. See V13a. + */ + @Test + fun `retries after failure are rate-limited, not counted`() = runTest { + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("store rejected") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + repeat(10) { + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + } + + // One round's worth, however many polls report it missing inside the backoff window. + assertStoreCount(ATTEMPTS_PER_STORE) + } + + /** + * V13a — a store that failed transiently MUST be retried; only a store that **succeeded** bars the + * hash for the session. + * + * Read as a pair with V13. V13 alone passes on a barred-on-attempt implementation, which is why the + * spec carried the weaker wording for 39 revisions: barring on attempt buys none of the anti-storm + * property and silently excludes any device whose one attempt hit a blip — on a feature that exists + * for devices something has already gone wrong for. + */ + @Test + fun `V13a - a transiently failed store is retried once the backoff elapses`() = runTest { + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("transient") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(ATTEMPTS_PER_STORE) + + // The blip passes, and so does the backoff window. + coEvery { swarmApiExecutor.send(any(), any()) } answers { + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + now += 61_000L + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(ATTEMPTS_PER_STORE + 1) + } + + /** + * V13h — a round needing more than the server's sub-request limit must be **chunked**, and all of it + * must land. + * + * The server rejects an oversized batch *whole* rather than truncating it, and it surfaces as a request + * failure rather than a size error — so it reads as a network problem and gets retried into the same + * wall. One config can split into ~66 parts (`MAX_MULTIPART_SIZE / MAX_MESSAGE_SIZE`), each its own + * store, so a single large config is already three times over. + * + * And it is the worst instance of this feature's recurring trap — a limit that excludes the very + * population the repair exists for: **the accounts with the largest configs have the most to lose and + * are exactly the ones whose recovery would be rejected wholesale.** + * + * Asserts the **boundary**, not eventual success — the fixture is 25 parts plus a delete, so it + * genuinely crosses 20, and the assertion is on the observed chunk sizes. A test that only checked + * "everything eventually stored" would pass on an implementation that got lucky with a small fixture. + */ + @Test + fun `V13h - a round exceeding the batch limit is chunked and all of it lands`() = runTest { + val parts = (1..25).map { "P$it" } + every { restoreSource.userConfigsToRestore(any()) } returns listOf( + PendingRestore( + label = "user config CONTACTS", + push = ConfigPush( + messages = parts.map { Bytes(it.toByteArray()) }, + seqNo = 7L, + obsoleteHashes = listOf("old-1"), + ), + claimedHashes = parts.toSet(), + namespace = { CONTACTS_NAMESPACE }, + ) + ) + + // Measure peak concurrency, which is what the batch window actually sees. The mock SUSPENDS, so a + // chunk's requests are genuinely in flight together and the peak is the chunk size; a synchronous + // mock would complete each call before the next began and report a peak of 1 whatever the + // implementation did. + var inFlight = 0 + var peakInFlight = 0 + coEvery { swarmApiExecutor.send(any(), any()) } coAnswers { + inFlight++ + peakInFlight = maxOf(peakInFlight, inFlight) + delay(1) + inFlight-- + when (secondArg>().api) { + // Must keep setUp's delete branch: returning a store response for a delete throws a cast + // error, which the retry wrapper then repeats — inflating the count by 4 and looking like + // a chunking bug rather than a fixture one. + is DeleteMessageApi -> DeleteMessageApi.SuccessResponse(1, 1) + else -> { + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + } + } + + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P1"))) + + // All 25 stores plus the delete went out... + coVerify(exactly = 26) { swarmApiExecutor.send(any(), any()) } + // ...in chunks that never exceeded the limit... + assertTrue(peakInFlight <= 20, "a chunk exceeded the limit: peak was $peakInFlight") + // ...and the peak proves a chunk was genuinely observed. Without this the assertion above passes + // on a fully sequential implementation, which would report a peak of 1 and prove nothing about + // the boundary — the degenerate case this vector exists to rule out. + assertTrue(peakInFlight > 1, "no chunk was observed at all: peak was $peakInFlight") + } + + /** + * V13g — the bar on a successfully re-stored hash must be **time-bounded**, not session-scoped. + * + * The bar exists to stop a swarm reporting the same hash missing on every poll costing a store every + * poll — a burst measured in seconds. "Never again this session" is unbounded in time, and a session can + * outlive the 30-day config TTL (a backgrounded mobile client, or desktop, which runs for weeks by + * design). In that window a hash we successfully put back can expire a *second* time, and a + * session-scoped bar blocks the recovery that should put it back — excluding long-lived sessions, which + * is exactly where configs expire. + * + * ⚠️ Driven by **advancing the clock**, never by rebuilding the recovery instance. A fresh instance + * clears in-memory state, so a restart-driven version of this test passes on the session-scoped + * implementation too — it would be measuring construction rather than expiry. A session-scoped + * implementation passes V13/V13a/V13b and fails only this. + */ + @Test + fun `V13g - a successfully re-stored hash is barred for a bounded time, not the session`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(1) + + // Still barred a few minutes later — the anti-storm property has to survive a burst of polls. + now += 5.minutes.inWholeMilliseconds + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(1) + + // Past the bar, the same hash reported missing again is a genuine second expiry: re-store it. + now += 1.hours.inWholeMilliseconds + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(2) + } + + /** + * V13b — "stored" must mean every *sub-response's own* code was 2xx, not merely that the outer call + * returned. A batch returns 200 while its sub-requests carry their own codes, so a client that discards + * the response bars every hash for the session having written nothing — and the bar's bookkeeping looks + * perfectly correct while the semantics are wrong. + * + * Here each message is its own `execute()`, which throws on its own non-2xx, so a multipart config with + * one bad part stays retryable **in full**. This test drives that: part 2 of 3 fails. + */ + @Test + fun `V13b - one failed part of a multipart store leaves the whole config retryable`() = runTest { + givenRestorable(claimedHashes = setOf("P1", "P2", "P3"), messageCount = 3) + // One part stores, the rest fail permanently — including through the retry wrapper, which is what + // makes this a genuine partial store rather than a transient blip that retries into success. Keyed + // on a count of successes rather than of calls, so it doesn't depend on which part runs first. + var stored = 0 + coEvery { swarmApiExecutor.send(any(), any()) } answers { + if (stored >= 1) throw RuntimeException("sub-request 500") + stored++ + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P2"))) + + // The round failed, so nothing is barred — including the parts that did store. + coEvery { swarmApiExecutor.send(any(), any()) } answers { + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + now += 61_000L + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P2"))) + + // All three parts re-stored on the retry, not just the one that failed. + verify(exactly = 2) { restoreSource.userConfigsToRestore(any()) } + } + + /** + * V13c — the wait **doubles** per consecutive failed round. A flat-rate implementation passes V13a and + * fails only this. + */ + @Test + fun `V13c - the second wait is 120s, not 60`() = runTest { + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("still failing") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(ATTEMPTS_PER_STORE) + + // 61s clears the first 60s window: a second round runs and also fails. + now += 61_000L + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(2 * ATTEMPTS_PER_STORE) + + // Another 61s is NOT enough now — the window has doubled to 120s. + now += 61_000L + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(2 * ATTEMPTS_PER_STORE) + + // ...but 121s from the second failure is. + now += 61_000L + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + assertStoreCount(3 * ATTEMPTS_PER_STORE) + } + + /** + * V13d — retry never stops. The ceiling bounds the *interval*, never the number of attempts: a cap on + * attempts is the population-exclusion shape this feature has already produced twice, and it would + * exclude exactly the swarms most in need of repair. + */ + @Test + fun `V13d - retrying never stops, however long the session fails`() = runTest { + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("permanently failing") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + // Twenty hours of a persistently failing swarm, stepping past each (growing) window. + // Counted via the gather rather than via storeCalls, which only counts stores that SUCCEED — a + // detail that made the first version of this test read 0 rounds against a working implementation. + repeat(40) { + now += 31.minutes.inWholeMilliseconds + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + } + + // Every single step past the ceiling produced a round — nothing ever gave up. + verify(exactly = 40) { restoreSource.userConfigsToRestore(any()) } + } + + /** + * V13 — a swarm reporting the same hash missing on every poll must cost one store, not one per poll. + * + * ⚠️ This test does **not** advance the clock, so it cannot tell a session-scoped bar from a + * time-bounded one — both pass it. Hence "while the bar holds" in the name rather than a bare "once": + * the unqualified claim would be a property no assertion here checks. V13g is the only test that + * separates them. + */ + @Test + fun `V13 - a hash reported missing on every poll is put back once while the bar holds`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + repeat(3) { + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + } + + assertStoreCount(1) + } + + /** + * V18 — one missing part re-stores the whole config. There's no way to do otherwise: the active + * hashes come back as an unordered set, so a part hash can't be mapped to its index in the message + * vector `push()` returns. Re-storing the present parts is harmless anyway — they're byte-identical, + * so it's a no-op TTL refresh. + */ + @Test + fun `V18 - one missing part of a multipart config re-stores all of them`() = runTest { + givenRestorable(claimedHashes = setOf("P1", "P2", "P3"), messageCount = 3) + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P2"))) + + assertStoreCount(3) + } + + /** + * And every part is then claimed, so a later poll naming a *different* part changes nothing. + * + * Deliberately unlabelled: this is a consequence of V18 on this client, not a vector of its own. It + * carried "V13b" until a sweep found that label already on a different test above — the same silent + * collision that is only supposed to happen *between* clients. + */ + @Test + fun `every part of a re-stored multipart config is claimed, not just the missing one`() = runTest { + givenRestorable(claimedHashes = setOf("P1", "P2", "P3"), messageCount = 3) + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P2"))) + assertStoreCount(3) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf("P1"))) + assertStoreCount(3) + } + + /** + * V17 — `push()` drains libsession's obsolete-hash set and clears it unconditionally, even for a + * clean config. So a recovery path that takes the list and throws it away loses those hashes + * permanently and leaves the superseded messages on the swarm forever. The delete has to be issued + * exactly as a normal push would. + */ + @Test + fun `V17 - obsolete hashes returned by a re-store are deleted`() = runTest { + givenRestorable(claimedHashes = setOf(h2), messageCount = 1, obsoleteHashes = listOf("old-1")) + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + coVerify(exactly = 1) { deleteMessageApiFactory.create(any(), listOf("old-1")) } + } + + /** + * A failed store must not take its config's obsolete hashes with it. + * + * `push()` has already cleared them, so deleting them without storing the replacement removes the + * swarm's older copy without adding the new one — and a device restoring from seed in that window gets + * *nothing* rather than stale state. Skipping the delete leaks them instead, which is close to free: + * obsolete hashes are never TTL-extended, so by the time our refreshed current hash has expired an + * un-refreshed older one is long gone. + */ + @Test + fun `a failed store does not delete its config's obsolete hashes`() = runTest { + givenRestorable(claimedHashes = setOf(h2), messageCount = 1, obsoleteHashes = listOf("old-1")) + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("store rejected") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + coVerify(exactly = 0) { deleteMessageApiFactory.create(any(), any()) } + + // Reachability control: the same fixture DOES delete once the store succeeds, so the absence above + // is the store failing rather than the delete path never being wired. + coEvery { swarmApiExecutor.send(any(), any()) } answers { + when (secondArg>().api) { + is DeleteMessageApi -> DeleteMessageApi.SuccessResponse(1, 1) + else -> { + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + } + } + now += 2.hours.inWholeMilliseconds + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + coVerify(exactly = 1) { deleteMessageApiFactory.create(any(), listOf("old-1")) } + } + + /** + * V17b — V17's negative counterpart: with nothing obsolete there is no delete at all. The natural bug + * is issuing an empty delete. + * + * An absence assertion, so it carries a reachability control — the `assertStoreCount(1)` below, which a + * dead harness could not satisfy. This is the + * normal case for a member re-store, where libsession never hands back the hashes — an empty list + * is the expected result, not a sign anything failed. + */ + @Test + fun `V17b - no obsolete hashes means no delete request`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(1) + coVerify(exactly = 0) { deleteMessageApiFactory.create(any(), any()) } + } + + /** Every cause, so that a cause added later cannot quietly become one that authorises a store. */ + @Test + fun `an inconclusive report triggers nothing, whatever made it inconclusive`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + listOf( + ConfigExpiryReport.Inconclusive.ExtendNotRequested, + ConfigExpiryReport.Inconclusive.NothingAsked, + ConfigExpiryReport.Inconclusive.NoUsableSubResponse, + ConfigExpiryReport.Inconclusive.ResponseUnreadable, + ).forEach { recovery.onUserConfigsChecked(userAuth(), it) } + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + @Test + fun `a report with nothing missing triggers nothing`() = runTest { + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(emptySet())) + + assertStoreCount(0) + assertRecoveryStillReachable() + } + + @Test + fun `recovery waits for the foreground`() = runTest { + appVisible.value = false + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(0) + + // Restore the foreground before the control: with the app backgrounded for the whole test, the + // control could not produce a store either, and a control that can't succeed proves nothing. + appVisible.value = true + assertRecoveryStillReachable() + } + + /** + * V13f — a throwing inspection must not escape, and this is the load-bearing half. + * + * The handoff sits at the very end of `Poller.poll()` with nothing above it to catch, and `gather` runs + * libsession code — `push()` throws when a config has no encryption keys. So an escaping exception + * doesn't merely mis-handle a hash: it fails the **entire poll**, every poll, because the condition + * doesn't clear. No messages processed, no configs merged, failure counter climbing. + * + * A best-effort repair feature must never be able to break the thing it rides on — recovery's whole + * premise is that polling continues, so this removes its own precondition and takes everything else + * polling does with it. And the failure signature is that every recovery test still passes, which is + * why this needs its own vector rather than being inferred from the others. + * + * The inspection is the only part that runs library code that can throw, and it is also the part that + * *looks* like a pure read — which is why it doesn't get wrapped. + */ + @Test + fun `V13f - a throwing inspection does not escape, bar anything, or consume the backoff`() = runTest { + every { restoreSource.userConfigsToRestore(any()) } throws + IllegalStateException("Cannot push data without an encryption key!") + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + // Must not throw — if this escapes, the caller is the poll itself. + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + // Nothing barred and no backoff consumed: an inspection that threw reached no verdict, so the + // hash stays retryable and the next poll tries again immediately. + every { restoreSource.userConfigsToRestore(any()) } returns listOf( + PendingRestore( + label = "user config CONTACTS", + push = ConfigPush( + messages = listOf(Bytes("config-data".toByteArray())), + seqNo = 7L, + obsoleteHashes = emptyList(), + ), + claimedHashes = setOf(h2), + namespace = { CONTACTS_NAMESPACE }, + ) + ) + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + assertStoreCount(1) + } + + /** + * V13e — a hash a **guard** ruled out is barred like a success, not treated as a retryable failure. + * + * (Two of the Session clients independently used "V13b" for two different tests, which is silent by + * construction and only surfaces when suites are compared — which is this feature's entire + * verification model.) + * + * Folding guard-rejections into "failure" costs no requests, which is why it doesn't look like a + * problem — but the gather re-runs on every poll, taking the config write lock and re-logging the same + * rejection every few seconds for the rest of the session. Only a store that *failed* is retryable. + */ + @Test + fun `V13e - a hash ruled out by a guard is not re-inspected on a later poll`() = runTest { + every { restoreSource.userConfigsToRestore(any()) } returns emptyList() + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + // Gathered once and then left alone, however many polls keep reporting it missing. + verify(exactly = 1) { restoreSource.userConfigsToRestore(any()) } + } + + /** + * The one must-not vector that **cannot** use [assertRecoveryStillReachable], because its premise *is* + * the death mode: "nothing was eligible" and "the harness returned nothing" are the same observation. + * A reachability control here would be asserting that an empty gather produces a store. + * + * So it proves the path was reached a different way — by asserting the guards were all passed and the + * gather actually ran. That is the assertion a dead harness cannot satisfy. + */ + @Test + fun `nothing eligible means no requests`() = runTest { + every { restoreSource.userConfigsToRestore(any()) } returns emptyList() + recovery.markLocalStateLevelWithSwarm(userId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onUserConfigsChecked(userAuth(), ConfigExpiryReport.Checked(setOf(h2))) + + verify(exactly = 1) { restoreSource.userConfigsToRestore(setOf(h2)) } + assertStoreCount(0) + } + + /** + * V23d — a group already flagged expired must have the flag cleared **by the re-store itself**. + * + * The reactive path cannot do this. It clears the flag when a keys message is *handled*, and the device + * that re-stored the bytes already holds that hash, so it may never handle it again — leaving a banner up + * permanently over keys that are back on the swarm. So a successful keys re-store emits directly. + */ + @Test + fun `V23d - a successful keys re-store announces itself so the flag can be cleared`() = runTest { + givenGroupKeysRestorable() + recovery.markLocalStateLevelWithSwarm(groupId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onGroupConfigsChecked(groupId, authFor(groupId.hexString), keysMissing) + + assertEquals(listOf(groupId), recovery.keysRestored.replayCache) + } + + /** + * V23c — a keys re-store that FAILED must not announce anything, so the flag stays up. + * + * What this pins is only that a wholly failed round is silent. It does NOT pin per-restore emission over + * per-round — every store fails here, so a round-level signal stays silent too and both implementations + * pass. The mixed round below is the test that separates them; this comment used to claim that job and + * was wrong, which a mutation to round-level emission demonstrated by surviving. + */ + @Test + fun `V23c - a failed keys re-store announces nothing, so the flag stands`() = runTest { + givenGroupKeysRestorable() + coEvery { swarmApiExecutor.send(any(), any()) } throws RuntimeException("store failed") + recovery.markLocalStateLevelWithSwarm(groupId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onGroupConfigsChecked(groupId, authFor(groupId.hexString), keysMissing) + + assertEquals(emptyList(), recovery.keysRestored.replayCache) + } + + /** + * A restore that is not the keys config must not clear the banner however well it goes — the flag is the + * keys config's alone, and info or members landing says nothing about whether the keys are back. + */ + @Test + fun `a successful non-keys restore announces nothing`() = runTest { + every { restoreSource.groupConfigsToRestore(groupId, any()) } returns listOf( + PendingRestore( + label = "group info for $groupId", + push = ConfigPush(listOf(Bytes("info".toByteArray())), 7L, emptyList()), + claimedHashes = setOf("info-1"), + namespace = { GROUP_INFO_NAMESPACE }, + ) + ) + recovery.markLocalStateLevelWithSwarm(groupId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onGroupConfigsChecked( + groupId, + authFor(groupId.hexString), + ConfigExpiryReport.Checked(setOf("info-1")), + ) + + assertStoreCount(1) + assertEquals(emptyList(), recovery.keysRestored.replayCache) + } + + /** + * The case that actually distinguishes per-restore emission from per-round: info lands, keys does not. + * + * V23c alone does NOT pin this, and I claimed it did. It fails *every* store, so `outcomes.any { it }` + * is false and a round-level implementation stays silent too — both pass. Verified by mutating the + * emission to round-level: V23c, V23d and the non-keys control all survived it. This test is what dies. + * + * Getting it wrong would clear the banner on a group whose keys never made it back, on the strength of + * its *info* config having stored. + */ + @Test + fun `a round where info lands and keys fails announces nothing`() = runTest { + // A factory that hands back a distinguishable api per namespace, so the keys store can be failed + // precisely rather than by call ordering — which is concurrent within a chunk and would make this + // test depend on which store happened to run first. + val keysApis = mutableSetOf() + val factory = mockk() + every { factory.create(any(), any(), any()) } answers { + mockk(relaxed = true).also { + if (thirdArg() == GROUP_KEYS_NAMESPACE) keysApis += it + } + } + recovery = ExpiredConfigRecovery( + restoreSource = restoreSource, + clock = clock, + appVisibilityManager = appVisibilityManager, + swarmApiExecutor = swarmApiExecutor, + storeMessageApiFactory = factory, + deleteMessageApiFactory = deleteMessageApiFactory, + ) + coEvery { swarmApiExecutor.send(any(), any()) } answers { + val request = secondArg>() + if (request.api in keysApis) throw RuntimeException("keys store 500") + storeCalls.incrementAndGet() + StoreMessageResponse(hash = h2, timestamp = Instant.EPOCH) + } + + every { restoreSource.groupConfigsToRestore(groupId, any()) } returns listOf( + PendingRestore( + label = "group info for $groupId", + push = ConfigPush(listOf(Bytes("info".toByteArray())), 7L, emptyList()), + claimedHashes = setOf("info-1"), + namespace = { GROUP_INFO_NAMESPACE }, + ), + PendingRestore( + label = "group keys for $groupId", + push = ConfigPush(listOf(Bytes("keys-bytes".toByteArray())), 0L, emptyList()), + claimedHashes = setOf("keys-1"), + isGroupKeys = true, + namespace = { GROUP_KEYS_NAMESPACE }, + ), + ) + recovery.markLocalStateLevelWithSwarm(groupId.hexString, mergedConfigMessagesForDiagnosticsOnly = true) + + recovery.onGroupConfigsChecked( + groupId, + authFor(groupId.hexString), + ConfigExpiryReport.Checked(setOf("info-1", "keys-1")), + ) + + // Info stored, so the round is not a failure — and the banner must still stand. + assertEquals(emptyList(), recovery.keysRestored.replayCache) + } + + private val keysMissing = ConfigExpiryReport.Checked(setOf("keys-1")) + + private fun givenGroupKeysRestorable() { + every { restoreSource.groupConfigsToRestore(groupId, any()) } returns listOf( + PendingRestore( + label = "group keys for $groupId", + push = ConfigPush(listOf(Bytes("keys-bytes".toByteArray())), 0L, emptyList()), + claimedHashes = setOf("keys-1"), + isGroupKeys = true, + namespace = { GROUP_KEYS_NAMESPACE }, + ) + ) + } + + private fun givenRestorable( + claimedHashes: Set, + messageCount: Int, + obsoleteHashes: List = emptyList(), + ) { + every { restoreSource.userConfigsToRestore(any()) } returns listOf( + PendingRestore( + label = "user config CONTACTS", + push = ConfigPush( + messages = List(messageCount) { Bytes("config-data-$it".toByteArray()) }, + seqNo = 7L, + obsoleteHashes = obsoleteHashes, + ), + claimedHashes = claimedHashes, + namespace = { CONTACTS_NAMESPACE }, + ) + ) + } + + /** + * Counts `store` requests. Deletes go through the same executor, so they're excluded by the API + * each request was built from rather than by counting calls. + */ + private suspend fun assertStoreCount(expected: Int) { + coVerify(exactly = expected) { + swarmApiExecutor.send(any(), match { it.api is StoreMessageApi }) + } + } + + /** + * Positive control for a must-not vector, and the reason every one of them calls it. + * + * A negative assertion cannot establish anything about its own harness: "no store happened" is + * produced equally by the guard correctly declining and by the path never running at all. Killing a + * stub in a way that dies *quietly* — returning an empty list, which is a perfectly legitimate + * "nothing was eligible" — made all nine must-not vectors here pass green while executing none of the + * code they name. A death that *throws* doesn't show this, which is why it has to be a quiet one. + * + * So each must-not vector finishes by proving a store is still reachable through the same harness and + * the same instance. A different swarm and a different hash are used so this can't perturb whatever + * the vector just asserted. + */ + private suspend fun assertRecoveryStillReachable() { + val controlSwarm = AccountId(IdPrefix.STANDARD, ByteArray(32) { 7 }).hexString + val before = storeCallCount() + + recovery.markLocalStateLevelWithSwarm( + controlSwarm, + mergedConfigMessagesForDiagnosticsOnly = true, + ) + recovery.onUserConfigsChecked( + authFor(controlSwarm), + ConfigExpiryReport.Checked(setOf("control-hash")), + ) + + assertTrue( + storeCallCount() > before, + "Harness is dead: a fully satisfied guard produced no store, so the assertion above " + + "proves nothing about the guard under test.", + ) + } + + private fun storeCallCount(): Int = storeCalls.get() + + private fun authFor(swarmPubKeyHex: String): SwarmAuth = mockk().also { + every { it.accountId } returns AccountId(swarmPubKeyHex) + } + + private fun userAuth(): SwarmAuth = mockk().also { + every { it.accountId } returns userId + } + + private companion object { + /** Hardcoded rather than read from libsession's native `Namespace`, which unit tests can't load. */ + const val CONTACTS_NAMESPACE = 3 + const val GROUP_KEYS_NAMESPACE = 12 + const val GROUP_INFO_NAMESPACE = 13 + + /** + * Requests one *failing* store produces: `retryWithUniformInterval` wraps each store with three + * retries, so a failed recovery round costs four requests per message rather than one. Worth + * naming — it's the multiplier on the round cap, so the real storm bound is + * `MAX_RECOVERY_ROUNDS_PER_SWARM × this × messages-per-config`. + */ + const val ATTEMPTS_PER_STORE = 4 + } +}