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.
@@ -107,7 +109,11 @@ class GroupLeavingWorker @AssistedInject constructor(
 
                     if (group != null && !group.kicked && !weAreTheOnlyAdmin) {
                         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 +139,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 +187,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 +204,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,83 @@
+package org.session.libsession.messaging.jobs
+
+import io.mockk.every
+import io.mockk.mockk
+import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.flow.MutableSharedFlow
+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.thoughtcrime.securesms.NoOpLogger
+
+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 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 {
+        val keysConfig = mockk<ReadableGroupKeysConfig> {
+            every { keys() } returns groupKeys
+        }
+
+        val configFactory = mockk<ConfigFactoryProtocol> {
+            // The real notification flow never completes; an ending flow would fail the
+            // wait outright instead of exercising the timeout
+            every { configUpdateNotifications } returns MutableSharedFlow<ConfigUpdateNotification>()
+            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),
+        )
+    }
+}

app/src/test/java/org/thoughtcrime/securesms/groups/GroupLeavingWorkerTest.kt

@@ -0,0 +1,141 @@
+package org.thoughtcrime.securesms.groups
+
+import androidx.work.Data
+import androidx.work.ListenableWorker
+import androidx.work.WorkerParameters
+import io.mockk.every
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.channels.SendChannel
+import kotlinx.coroutines.flow.MutableStateFlow
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.test.TestScope
+import kotlinx.coroutines.test.runTest
+import network.loki.messenger.libsession_util.ReadableGroupMembersConfig
+import network.loki.messenger.libsession_util.ReadableUserGroupsConfig
+import network.loki.messenger.libsession_util.util.GroupInfo
+import org.junit.Assert.assertEquals
+import org.junit.Before
+import org.junit.Test
+import org.session.libsession.messaging.groups.GroupScope
+import org.session.libsession.messaging.sending_receiving.MessageSender
+import org.session.libsession.messaging.utilities.UpdateMessageData
+import org.session.libsession.utilities.GroupConfigs
+import org.session.libsession.utilities.UserConfigs
+import org.session.libsignal.exceptions.NonRetryableException
+import org.session.libsignal.utilities.AccountId
+import org.session.libsignal.utilities.Log
+import org.thoughtcrime.securesms.NoOpLogger
+import org.thoughtcrime.securesms.database.Storage
+import org.thoughtcrime.securesms.dependencies.ConfigFactory
+
+class GroupLeavingWorkerTest {
+    private val groupId = AccountId("03${"11".repeat(32)}")
+
+    private val storage = mockk<Storage>(relaxed = true)
+    private val configFactory = mockk<ConfigFactory>(relaxed = true)
+    private val messageSender = mockk<MessageSender>()
+
+    @Before
+    fun setUp() {
+        Log.initialize(NoOpLogger)
+
+        val group = mockk<GroupInfo.ClosedGroupInfo> {
+            every { kicked } returns false
+            every { destroyed } returns false
+        }
+
+        every { configFactory.dangerouslyAccessUserConfigs() } returns Pair(
+            mockk<UserConfigs> {
+                every { userGroups } returns mockk<ReadableUserGroupsConfig> {
+                    every { getClosedGroup(groupId.hexString) } returns group
+                }
+            },
+            {},
+        )
+
+        // No admins, so we are not the only admin and the leave takes the announce-and-go path
+        every { configFactory.dangerouslyAccessGroupConfigs(groupId) } returns Pair(
+            mockk<GroupConfigs> {
+                every { groupMembers } returns mockk<ReadableGroupMembersConfig> {
+                    every { all() } returns emptyList()
+                }
+            },
+            {},
+        )
+    }
+
+    @Test
+    fun `group is left locally when the departure cannot be announced`() = runTest {
+        answerSendWith(Result.failure(NonRetryableException("no keys for this group")))
+
+        val result = worker(runAttemptCount = 0).doWork()
+
+        assertEquals(ListenableWorker.Result.success(), result)
+        verify { configFactory.removeGroup(groupId) }
+        verify(exactly = 0) { storage.insertGroupInfoErrorQuit(any()) }
+    }
+
+    @Test
+    fun `retries stop once the attempts are used up`() = runTest {
+        answerSendWith(Result.failure(RuntimeException("network went away")))
+
+        assertEquals(ListenableWorker.Result.retry(), worker(runAttemptCount = 0).doWork())
+        assertEquals(ListenableWorker.Result.failure(), worker(runAttemptCount = 2).doWork())
+    }
+
+    @Test
+    fun `the error message replaces the one from the previous attempt`() = runTest {
+        answerSendWith(Result.failure(RuntimeException("network went away")))
+
+        worker(runAttemptCount = 0).doWork()
+
+        verify(exactly = 1) { storage.insertGroupInfoErrorQuit(groupId) }
+        verify {
+            storage.deleteGroupInfoMessages(groupId, UpdateMessageData.Kind.GroupErrorQuit::class.java)
+        }
+    }
+
+    @Test
+    fun `leave completes when both results arrive before the worker waits for them`() = runTest {
+        answerSendImmediatelyWith(Result.success(Unit))
+
+        val result = worker(runAttemptCount = 0).doWork()
+
+        assertEquals(ListenableWorker.Result.success(), result)
+        verify { configFactory.removeGroup(groupId) }
+    }
+
+    /** Reports the result once the worker is already waiting for it. */
+    private fun TestScope.answerSendWith(result: Result<Unit>) {
+        every { messageSender.send(any(), any(), any()) } answers {
+            val statusChannel = thirdArg<SendChannel<Result<Unit>>>()
+            launch { statusChannel.send(result) }
+        }
+    }
+
+    /** Reports the result as the send is made, before the worker gets as far as waiting for it. */
+    private fun answerSendImmediatelyWith(result: Result<Unit>) {
+        every { messageSender.send(any(), any(), any()) } answers {
+            thirdArg<SendChannel<Result<Unit>>>().trySend(result)
+        }
+    }
+
+    private fun CoroutineScope.worker(runAttemptCount: Int) = GroupLeavingWorker(
+        context = mockk(relaxed = true),
+        params = mockk<WorkerParameters>(relaxed = true) {
+            every { inputData } returns Data.Builder()
+                .putString("group_id", groupId.hexString)
+                .build()
+            every { this@mockk.runAttemptCount } returns runAttemptCount
+        },
+        storage = storage,
+        configFactory = configFactory,
+        groupScope = GroupScope(this),
+        tokenFetcher = mockk { every { token } returns MutableStateFlow(null) },
+        serverApiExecutor = mockk(relaxed = true),
+        pushUnregisterApiFactory = mockk(relaxed = true),
+        messageSender = messageSender,
+    )
+}