Fix: leaving a group we hold no encryption keys for retried forever

.gitignore

@@ -6,6 +6,7 @@ project.properties
 bin/
 gen/
 .idea/
+.claude/
 *.iml
 out
 build

app/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt

@@ -24,6 +24,7 @@ import org.session.libsession.messaging.utilities.Data
 import org.session.libsession.utilities.ConfigFactoryProtocol
 import org.session.libsession.utilities.ConfigUpdateNotification
 import org.session.libsession.utilities.withGroupConfigs
+import org.session.libsignal.exceptions.NonRetryableException
 import org.session.libsignal.utilities.AccountId
 import org.session.libsignal.utilities.Log
 import org.thoughtcrime.securesms.api.error.UnhandledStatusCodeException
@@ -94,13 +95,18 @@ class MessageSendJob @AssistedInject constructor(
         val isSync = destination is Destination.Contact && destination.publicKey == storage.getUserPublicKey()
 
         try {
-            // Shouldn't send message to group when the group has no keys available
+            // A group we hold no encryption keys for can't be sent to, and the keys can only
+            // arrive by an admin granting them to us, which may never happen. Typed so a caller
+            // waiting on this send can tell it apart from a transient failure and give up
+            // deliberately rather than waiting for keys that aren't coming.
             if (destination is Destination.ClosedGroup) {
-                requireNotNull(withTimeoutOrNull(20_000L) {
+                val keysAvailable = withTimeoutOrNull(20_000L) {
                     configFactory
                         .waitForGroupEncryptionKeys(AccountId(destination.publicKey))
-                }) {
-                    "Timeout waiting for group keys to become available"
+                } != null
+
+                if (!keysAvailable) {
+                    throw NonRetryableException("Timeout waiting for group keys to become available")
                 }
             }
 

app/src/main/java/org/thoughtcrime/securesms/groups/GroupLeavingWorker.kt

@@ -61,8 +61,10 @@ class GroupLeavingWorker @AssistedInject constructor(
         return groupScope.launchAndWait(groupId, "GroupLeavingWorker") {
             val group = configFactory.getGroup(groupId)
 
-            // Make sure we only have one group leaving control message
+            // Make sure we only have one group leaving control message, and that the error
+            // message from an earlier attempt doesn't sit alongside it
             storage.deleteGroupInfoMessages(groupId, UpdateMessageData.Kind.GroupLeaving::class.java)
+            storage.deleteGroupInfoMessages(groupId, UpdateMessageData.Kind.GroupErrorQuit::class.java)
             storage.insertGroupInfoLeaving(groupId)
 
             // Best effort to unsubscribe ourselves from the registration server.
@@ -98,16 +100,27 @@ class GroupLeavingWorker @AssistedInject constructor(
             try {
                 if (group?.destroyed != true) {
                     // Only send the left/left notification group message when we are not kicked and we are not the only admin (only admin has a special treatment)
-                    val weAreTheOnlyAdmin = configFactory.withGroupConfigs(groupId) { config ->
+                    val (weAreTheOnlyAdmin, weHoldTheGroupKeys) = configFactory.withGroupConfigs(groupId) { config ->
                         val allMembers = config.groupMembers.all()
-                        allMembers.count { it.admin } == 1 &&
+                        val onlyAdmin = allMembers.count { it.admin } == 1 &&
                                 allMembers.first { it.admin }
                                     .accountId() == storage.getUserPublicKey()
+
+                        onlyAdmin to config.groupKeys.keys().isNotEmpty()
                     }
 
-                    if (group != null && !group.kicked && !weAreTheOnlyAdmin) {
+                    // Holding no keys means there is nothing to encrypt the departure with, and it
+                    // is not a state that resolves itself: the flag above is only set when a
+                    // revocation message is actually received, and receiving one clears the keys,
+                    // so a revocation we were never told about looks exactly like this. Attempting
+                    // the send would cost the user 20 seconds per message to learn that.
+                    if (group != null && !group.kicked && !weAreTheOnlyAdmin && weHoldTheGroupKeys) {
                         val address = Address.fromSerialized(groupId.hexString)
-                        val statusChannel = Channel<kotlin.Result<Unit>>()
+                        // The jobs report with trySend, which delivers nothing unless a receiver
+                        // is already parked, and they run on their own dispatcher: either result
+                        // can land before we reach the wait below. Unbuffered, that result is lost
+                        // and the leave waits for it forever, holding this group's scope with it.
+                        val statusChannel = Channel<kotlin.Result<Unit>>(capacity = Channel.UNLIMITED)
 
                         // Always send a "XXX left" message to the group if we can
                         messageSender.send(
@@ -133,8 +146,19 @@ class GroupLeavingWorker @AssistedInject constructor(
                         )
 
                         // Wait for both messages to be sent
-                        repeat(2) {
-                            statusChannel.receive().getOrThrow()
+                        try {
+                            repeat(2) {
+                                statusChannel.receive().getOrThrow()
+                            }
+                        } catch (e: CancellationException) {
+                            throw e
+                        } catch (e: NonRetryableException) {
+                            // Our access to the group can be revoked before we get around to
+                            // leaving it, which leaves us without the keys to encrypt the
+                            // departure to the group. Nothing will grant them back, so honour the
+                            // leave locally instead: the alternative is a group the user can never
+                            // leave, however many times they ask.
+                            Log.e(TAG, "Unable to announce leaving group $groupId. Proceeding...", e)
                         }
                     }
 
@@ -170,7 +194,10 @@ class GroupLeavingWorker @AssistedInject constructor(
             } catch (e: Exception) {
                 storage.insertGroupInfoErrorQuit(groupId)
                 Log.e(TAG, "Failed to leave group $groupId", e)
-                if (e is NonRetryableException) {
+                // WorkManager's retry has backoff but no attempt cap, so an error that never
+                // resolves itself would have us re-attempting the leave — and reporting the
+                // failure to the conversation — for as long as the group exists
+                if (e is NonRetryableException || runAttemptCount >= MAX_RETRIES) {
                     Result.failure()
                 } else {
                     Result.retry()
@@ -184,6 +211,8 @@ class GroupLeavingWorker @AssistedInject constructor(
     companion object {
         private const val TAG = "GroupLeavingWorker"
 
+        private const val MAX_RETRIES = 2
+
         private const val KEY_GROUP_ID = "group_id"
         private const val KEY_DELETE_GROUP = "delete_group"
 

app/src/test/java/org/session/libsession/messaging/jobs/MessageSendJobTest.kt

@@ -0,0 +1,114 @@
+package org.session.libsession.messaging.jobs
+
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.MutableSharedFlow
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.flow.StateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.runTest
+import network.loki.messenger.libsession_util.ReadableGroupKeysConfig
+import org.junit.Assert.assertTrue
+import org.junit.Before
+import org.junit.Test
+import org.session.libsession.messaging.messages.Destination
+import org.session.libsession.messaging.messages.control.GroupUpdated
+import org.session.libsession.utilities.ConfigFactoryProtocol
+import org.session.libsession.utilities.ConfigUpdateNotification
+import org.session.libsession.utilities.GroupConfigs
+import org.session.libsignal.exceptions.NonRetryableException
+import org.session.libsignal.utilities.Log
+import org.session.protos.SessionProtos
+import org.session.libsignal.utilities.AccountId
+import org.thoughtcrime.securesms.NoOpLogger
+import kotlin.time.Duration.Companion.seconds
+
+class MessageSendJobTest {
+    private val groupId = "03${"11".repeat(32)}"
+
+    @Before
+    fun setUp() {
+        Log.initialize(NoOpLogger)
+    }
+
+    @Test
+    fun `sending to a group we hold no keys for fails non-retryably`() = runTest {
+        val statusChannel = Channel<Result<Unit>>(capacity = 1)
+
+        job(statusChannel, groupKeys = emptyList()).execute("test")
+
+        val error = statusChannel.receive().exceptionOrNull()
+        assertTrue("expected NonRetryableException, got $error", error is NonRetryableException)
+    }
+
+    @Test
+    fun `sending waits for keys that arrive while it is waiting`() = runTest {
+        val statusChannel = Channel<Result<Unit>>(capacity = 1)
+        val keys = MutableStateFlow(emptyList<ByteArray>())
+        val configUpdates = MutableSharedFlow<ConfigUpdateNotification>()
+
+        launch {
+            // Well inside the 20s the send is prepared to wait, and virtual time, so it is free
+            delay(5.seconds)
+            keys.value = listOf(ByteArray(32))
+            configUpdates.emit(ConfigUpdateNotification.GroupConfigsUpdated(AccountId(groupId), fromMerge = true))
+        }
+
+        job(statusChannel, groupKeys = keys, configUpdates = configUpdates).execute("test")
+
+        assertTrue(statusChannel.receive().isSuccess)
+    }
+
+    @Test
+    fun `sending to a group we hold keys for is sent`() = runTest {
+        val statusChannel = Channel<Result<Unit>>(capacity = 1)
+
+        job(statusChannel, groupKeys = listOf(ByteArray(32))).execute("test")
+
+        assertTrue(statusChannel.receive().isSuccess)
+    }
+
+    private fun job(
+        statusChannel: Channel<Result<Unit>>,
+        groupKeys: List<ByteArray>,
+    ): MessageSendJob = job(statusChannel, MutableStateFlow(groupKeys))
+
+    private fun job(
+        statusChannel: Channel<Result<Unit>>,
+        groupKeys: StateFlow<List<ByteArray>>,
+        // The real notification flow never completes; an ending flow would fail the wait
+        // outright instead of exercising the timeout
+        configUpdates: Flow<ConfigUpdateNotification> = MutableSharedFlow(),
+    ): MessageSendJob {
+        val keysConfig = mockk<ReadableGroupKeysConfig> {
+            every { keys() } answers { groupKeys.value }
+        }
+
+        val configFactory = mockk<ConfigFactoryProtocol> {
+            every { configUpdateNotifications } returns configUpdates
+            every { dangerouslyAccessGroupConfigs(any()) } returns Pair(
+                mockk<GroupConfigs> { every { this@mockk.groupKeys } returns keysConfig },
+                {},
+            )
+        }
+
+        return MessageSendJob(
+            message = GroupUpdated(
+                SessionProtos.GroupUpdateMessage.newBuilder()
+                    .setMemberLeftMessage(SessionProtos.GroupUpdateMemberLeftMessage.getDefaultInstance())
+                    .build()
+            ),
+            destination = Destination.ClosedGroup(groupId),
+            statusCallback = statusChannel,
+            attachmentUploadJobFactory = mockk(relaxed = true),
+            messageDataProvider = mockk(relaxed = true),
+            storage = mockk(relaxed = true),
+            configFactory = configFactory,
+            messageSender = mockk(relaxed = true),
+            jobQueue = mockk(relaxed = true),
+        )
+    }
+}