diff --git a/.gitignore b/.gitignore index 90be087c09..d9a4cfa843 100644 --- a/.gitignore +++ b/.gitignore @@ -6,6 +6,7 @@ project.properties bin/ gen/ .idea/ +.claude/ *.iml out build diff --git a/app/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt b/app/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt index 7d3a5e6d8a..b62bbd7ede 100644 --- a/app/src/main/java/org/session/libsession/messaging/jobs/MessageSendJob.kt +++ b/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") } } diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupLeavingWorker.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupLeavingWorker.kt index e0cbc987f7..829ceab778 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupLeavingWorker.kt +++ b/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>() + // 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>(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" diff --git a/app/src/test/java/org/session/libsession/messaging/jobs/MessageSendJobTest.kt b/app/src/test/java/org/session/libsession/messaging/jobs/MessageSendJobTest.kt new file mode 100644 index 0000000000..68e8c6f3d4 --- /dev/null +++ b/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>(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>(capacity = 1) + + job(statusChannel, groupKeys = listOf(ByteArray(32))).execute("test") + + assertTrue(statusChannel.receive().isSuccess) + } + + private fun job( + statusChannel: Channel>, + groupKeys: List, + ): MessageSendJob { + val keysConfig = mockk { + every { keys() } returns groupKeys + } + + val configFactory = mockk { + // The real notification flow never completes; an ending flow would fail the + // wait outright instead of exercising the timeout + every { configUpdateNotifications } returns MutableSharedFlow() + every { dangerouslyAccessGroupConfigs(any()) } returns Pair( + mockk { 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), + ) + } +} diff --git a/app/src/test/java/org/thoughtcrime/securesms/groups/GroupLeavingWorkerTest.kt b/app/src/test/java/org/thoughtcrime/securesms/groups/GroupLeavingWorkerTest.kt new file mode 100644 index 0000000000..8b374cfbac --- /dev/null +++ b/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(relaxed = true) + private val configFactory = mockk(relaxed = true) + private val messageSender = mockk() + + @Before + fun setUp() { + Log.initialize(NoOpLogger) + + val group = mockk { + every { kicked } returns false + every { destroyed } returns false + } + + every { configFactory.dangerouslyAccessUserConfigs() } returns Pair( + mockk { + every { userGroups } returns mockk { + 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 { + every { groupMembers } returns mockk { + 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) { + every { messageSender.send(any(), any(), any()) } answers { + val statusChannel = thirdArg>>() + 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) { + every { messageSender.send(any(), any(), any()) } answers { + thirdArg>>().trySend(result) + } + } + + private fun CoroutineScope.worker(runAttemptCount: Int) = GroupLeavingWorker( + context = mockk(relaxed = true), + params = mockk(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, + ) +}