From b019aba3fcb7a78d1d75253df268aab155c043c8 Mon Sep 17 00:00:00 2001 From: Cody Henthorne Date: Mon, 27 Jul 2026 12:25:48 -0400 Subject: [PATCH] Ensure group updates render before message content when not from p2p. --- .../securesms/groups/GroupManager.java | 5 +- .../securesms/groups/GroupManagerV2.java | 5 +- .../v2/processing/GroupsV2StateProcessor.kt | 48 +++++++----- .../messages/DataMessageProcessor.kt | 3 +- .../messages/MessageContentProcessor.kt | 10 ++- .../securesms/mms/IncomingMessage.kt | 5 +- ...le.kt => MessageTableTest_groupUpdates.kt} | 76 ++++++++++++++++++- 7 files changed, 119 insertions(+), 33 deletions(-) rename app/src/test/java/org/thoughtcrime/securesms/database/{SmsDatabaseTest_collapseJoinRequestEventsIfPossible.kt => MessageTableTest_groupUpdates.kt} (80%) diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManager.java b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManager.java index 216c045b32..969974bd2d 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManager.java +++ b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManager.java @@ -175,11 +175,12 @@ public final class GroupManager { int revision, long timestamp, @Nullable byte[] signedGroupChange, - @Nullable String serverGuid) + @Nullable String serverGuid, + @Nullable Long receivedTime) throws GroupChangeBusyException, IOException, GroupNotAMemberException { try (GroupManagerV2.GroupUpdater updater = new GroupManagerV2(context).updater(groupMasterKey)) { - return updater.updateLocalToServerRevision(revision, timestamp, groupRecord, groupSecretParams, signedGroupChange, serverGuid); + return updater.updateLocalToServerRevision(revision, timestamp, groupRecord, groupSecretParams, signedGroupChange, serverGuid, receivedTime); } } diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManagerV2.java b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManagerV2.java index 9a2d257a45..44ab1da1a0 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManagerV2.java +++ b/app/src/main/java/org/thoughtcrime/securesms/groups/GroupManagerV2.java @@ -847,11 +847,12 @@ final class GroupManagerV2 { @NonNull Optional localRecord, @Nullable GroupSecretParams groupSecretParams, @Nullable byte[] signedGroupChange, - @Nullable String serverGuid) + @Nullable String serverGuid, + @Nullable Long receivedTime) throws IOException, GroupNotAMemberException { return GroupsV2StateProcessor.forGroup(serviceIds, groupMasterKey, groupSecretParams) - .updateLocalGroupToRevision(revision, timestamp, getDecryptedGroupChange(signedGroupChange), localRecord, serverGuid); + .updateLocalGroupToRevision(revision, timestamp, getDecryptedGroupChange(signedGroupChange), localRecord, serverGuid, receivedTime); } @WorkerThread diff --git a/app/src/main/java/org/thoughtcrime/securesms/groups/v2/processing/GroupsV2StateProcessor.kt b/app/src/main/java/org/thoughtcrime/securesms/groups/v2/processing/GroupsV2StateProcessor.kt index c9c414ad18..14df359fc9 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/groups/v2/processing/GroupsV2StateProcessor.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/groups/v2/processing/GroupsV2StateProcessor.kt @@ -178,7 +178,8 @@ class GroupsV2StateProcessor private constructor( timestamp: Long, signedGroupChange: DecryptedGroupChange? = null, groupRecord: Optional = SignalDatabase.groups.getGroup(groupId), - serverGuid: String? = null + serverGuid: String? = null, + receivedTime: Long? = null ): GroupUpdateResult { if (localIsAtLeast(groupRecord, targetRevision)) { return GroupUpdateResult(GroupUpdateResult.UpdateStatus.GROUP_CONSISTENT_OR_AHEAD, null) @@ -187,14 +188,14 @@ class GroupsV2StateProcessor private constructor( val currentLocalState: DecryptedGroup? = groupRecord.filter { it.hasV2GroupProperties }.map { it.requireV2GroupProperties().decryptedGroup }.orNull()?.let { if (it.isEmptyPlaceholder()) null else it } if (signedGroupChange != null && canApplyP2pChange(targetRevision, signedGroupChange, currentLocalState, groupRecord)) { - when (val p2pUpdateResult = updateViaPeerGroupChange(timestamp, serverGuid, signedGroupChange, currentLocalState!!, forceApply = false)) { + when (val p2pUpdateResult = updateViaPeerGroupChange(timestamp, serverGuid, signedGroupChange, currentLocalState!!, forceApply = false, receivedTime = receivedTime)) { InternalUpdateResult.NoUpdateNeeded -> return GroupUpdateResult.CONSISTENT_OR_AHEAD is InternalUpdateResult.Updated -> return GroupUpdateResult.updated(p2pUpdateResult.updatedLocalState) else -> Log.w(TAG, "$logPrefix P2P update was not successfully processed, falling back to server update") } } - val serverUpdateResult = updateViaServer(targetRevision, timestamp, serverGuid, groupRecord) + val serverUpdateResult = updateViaServer(targetRevision, timestamp, serverGuid, groupRecord, receivedTime) when (serverUpdateResult) { InternalUpdateResult.NoUpdateNeeded -> return GroupUpdateResult.CONSISTENT_OR_AHEAD @@ -210,7 +211,7 @@ class GroupsV2StateProcessor private constructor( Log.w(TAG, "$logPrefix Server says we're not a member. Ignoring P2P group change because this change doesn't add or remove us.") } else { Log.i(TAG, "$logPrefix Server says we're not a member. Force applying P2P group change because it adds or removes us.") - when (val forcedP2pUpdateResult = updateViaPeerGroupChange(timestamp, serverGuid, signedGroupChange, currentLocalState, forceApply = true)) { + when (val forcedP2pUpdateResult = updateViaPeerGroupChange(timestamp, serverGuid, signedGroupChange, currentLocalState, forceApply = true, receivedTime = receivedTime)) { is InternalUpdateResult.Updated -> return GroupUpdateResult.updated(forcedP2pUpdateResult.updatedLocalState) InternalUpdateResult.NoUpdateNeeded -> return GroupUpdateResult.CONSISTENT_OR_AHEAD is InternalUpdateResult.NotAMember, is InternalUpdateResult.UpdateFailed -> Log.w(TAG, "$logPrefix Unable to apply P2P group change when not a member: $forcedP2pUpdateResult") @@ -276,7 +277,8 @@ class GroupsV2StateProcessor private constructor( serverGuid: String?, signedGroupChange: DecryptedGroupChange, currentLocalState: DecryptedGroup, - forceApply: Boolean + forceApply: Boolean, + receivedTime: Long? = null ): InternalUpdateResult { val updatedGroupState = try { if (forceApply) { @@ -297,7 +299,8 @@ class GroupsV2StateProcessor private constructor( groupStateDiff = groupStateDiff, groupSendEndorsements = null, forceSave = forceApply, - persistProfileKeys = !forceApply + persistProfileKeys = !forceApply, + receivedTime = receivedTime ) } @@ -305,13 +308,14 @@ class GroupsV2StateProcessor private constructor( targetRevision: Int, timestamp: Long, serverGuid: String?, - groupRecord: Optional = SignalDatabase.groups.getGroup(groupId) + groupRecord: Optional = SignalDatabase.groups.getGroup(groupId), + receivedTime: Long? = null ): InternalUpdateResult { var currentLocalState: DecryptedGroup? = groupRecord.filter { it.hasV2GroupProperties }.map { it.requireV2GroupProperties().decryptedGroup }.orNull()?.let { if (it.isEmptyPlaceholder()) null else it } if (targetRevision == LATEST && (currentLocalState == null || currentLocalState.revision == RESTORE_PLACEHOLDER_REVISION)) { Log.i(TAG, "$logPrefix Latest revision only, update to latest directly") - return updateToLatestViaServer(timestamp, currentLocalState, reconstructChange = false, forceUpdate = false) + return updateToLatestViaServer(timestamp, currentLocalState, reconstructChange = false, forceUpdate = false, receivedTime = receivedTime) } Log.i(TAG, "$logPrefix Paging from server targetRevision: ${if (targetRevision == LATEST) "latest" else targetRevision}") @@ -322,7 +326,7 @@ class GroupsV2StateProcessor private constructor( val joinedAtFailure = InternalUpdateResult.from(joinedAtResult.getCause()!!) if (joinedAtFailure is InternalUpdateResult.NotAMember) { Log.i(TAG, "$logPrefix Not a member, try to update to latest directly") - return updateToLatestViaServer(timestamp, currentLocalState, reconstructChange = currentLocalState != null, forceUpdate = true) + return updateToLatestViaServer(timestamp, currentLocalState, reconstructChange = currentLocalState != null, forceUpdate = true, receivedTime = receivedTime) } else { return joinedAtFailure } @@ -375,7 +379,7 @@ class GroupsV2StateProcessor private constructor( if (addMessagesForAllUpdates) { Log.d(TAG, "$logPrefix Inserting group changes into chat history") - runningTimestamp = profileAndMessageHelper.insertUpdateMessages(runningTimestamp, currentLocalState, applyGroupStateDiffResult.processedLogEntries, serverGuid) + runningTimestamp = profileAndMessageHelper.insertUpdateMessages(runningTimestamp, currentLocalState, applyGroupStateDiffResult.processedLogEntries, serverGuid, receivedTime) } remoteGroupStateDiff @@ -406,7 +410,7 @@ class GroupsV2StateProcessor private constructor( if (!addMessagesForAllUpdates) { Log.i(TAG, "$logPrefix Inserting single update message for restore placeholder") - profileAndMessageHelper.insertUpdateMessages(runningTimestamp, null, setOf(AppliedGroupChangeLog(currentLocalState!!, null)), serverGuid) + profileAndMessageHelper.insertUpdateMessages(runningTimestamp, null, setOf(AppliedGroupChangeLog(currentLocalState!!, null)), serverGuid, receivedTime) } profileAndMessageHelper.persistLearnedProfileKeys(profileKeys) @@ -423,7 +427,7 @@ class GroupsV2StateProcessor private constructor( return InternalUpdateResult.Updated(currentLocalState!!) } - private fun updateToLatestViaServer(timestamp: Long, currentLocalState: DecryptedGroup?, reconstructChange: Boolean, forceUpdate: Boolean): InternalUpdateResult { + private fun updateToLatestViaServer(timestamp: Long, currentLocalState: DecryptedGroup?, reconstructChange: Boolean, forceUpdate: Boolean, receivedTime: Long? = null): InternalUpdateResult { val result = groupsApi.getGroupAsResult(groupSecretParams, groupsV2Authorization.getAuthorizationForToday(serviceIds, groupSecretParams)) val groupResponse = if (result is NetworkResult.Success) { @@ -440,7 +444,8 @@ class GroupsV2StateProcessor private constructor( serverGuid = null, groupStateDiff = remoteGroupStateDiff, groupSendEndorsements = groupOperations.receiveGroupSendEndorsements(serviceIds.aci, groupResponse.group, groupResponse.groupSendEndorsementsResponse), - forceSave = forceUpdate && groupResponse.group.members.asSequence().mapNotNull { ACI.parseOrNull(it.aciBytes) }.any { serviceIds.matches(it) } + forceSave = forceUpdate && groupResponse.group.members.asSequence().mapNotNull { ACI.parseOrNull(it.aciBytes) }.any { serviceIds.matches(it) }, + receivedTime = receivedTime ) } @@ -572,7 +577,8 @@ class GroupsV2StateProcessor private constructor( groupStateDiff: GroupStateDiff, groupSendEndorsements: ReceivedGroupSendEndorsements?, forceSave: Boolean, - persistProfileKeys: Boolean = true + persistProfileKeys: Boolean = true, + receivedTime: Long? = null ): InternalUpdateResult { val currentLocalState: DecryptedGroup? = groupStateDiff.previousGroupState val applyGroupStateDiffResult = GroupStatePatcher.applyGroupStateDiff(groupStateDiff, GroupStatePatcher.LATEST) @@ -598,10 +604,10 @@ class GroupsV2StateProcessor private constructor( if (currentLocalState == null || currentLocalState.revision == RESTORE_PLACEHOLDER_REVISION) { if (!updatedGroupState.terminated) { Log.i(TAG, "$logPrefix Inserting single update message for no local state or restore placeholder") - profileAndMessageHelper.insertUpdateMessages(timestamp, null, setOf(AppliedGroupChangeLog(updatedGroupState, null)), null) + profileAndMessageHelper.insertUpdateMessages(timestamp, null, setOf(AppliedGroupChangeLog(updatedGroupState, null)), null, receivedTime) } } else { - profileAndMessageHelper.insertUpdateMessages(timestamp, currentLocalState, applyGroupStateDiffResult.processedLogEntries, serverGuid) + profileAndMessageHelper.insertUpdateMessages(timestamp, currentLocalState, applyGroupStateDiffResult.processedLogEntries, serverGuid, receivedTime) } if (persistProfileKeys) { @@ -770,7 +776,8 @@ class GroupsV2StateProcessor private constructor( timestamp: Long, previousGroupState: DecryptedGroup?, processedLogEntries: Collection, - serverGuid: String? + serverGuid: String?, + receivedTime: Long? = null ): Long { var runningTimestamp = timestamp var runningGroupState = previousGroupState @@ -792,7 +799,8 @@ class GroupsV2StateProcessor private constructor( storeMessage( decryptedGroupV2Context = GroupProtoUtil.createDecryptedGroupV2Context(masterKey, GroupMutation(runningGroupState, entry.change, entry.group), null), timestamp = runningTimestamp, - serverGuid = serverGuid + serverGuid = serverGuid, + receivedTime = receivedTime ) runningTimestamp++ } @@ -974,7 +982,7 @@ class GroupsV2StateProcessor private constructor( } @VisibleForTesting - fun storeMessage(decryptedGroupV2Context: DecryptedGroupV2Context, timestamp: Long, serverGuid: String?) { + fun storeMessage(decryptedGroupV2Context: DecryptedGroupV2Context, timestamp: Long, serverGuid: String?, receivedTime: Long? = null) { val editor: Optional = getEditor(decryptedGroupV2Context) val serviceIds = SignalStore.account.getServiceIds() @@ -1018,7 +1026,7 @@ class GroupsV2StateProcessor private constructor( val isNotifiable = isGroupAdd || isGroupTerminate - val groupMessage = IncomingMessage.groupUpdate(RecipientId.from(editor.get()), timestamp, groupId, updateDescription, isNotifiable, serverGuid) + val groupMessage = IncomingMessage.groupUpdate(RecipientId.from(editor.get()), timestamp, groupId, updateDescription, isNotifiable, serverGuid, receivedTime) val insertResult = SignalDatabase.messages.insertMessageInbox(groupMessage) if (insertResult.isPresent) { diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/DataMessageProcessor.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/DataMessageProcessor.kt index 1a72e17f95..9ad6a15936 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/DataMessageProcessor.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/DataMessageProcessor.kt @@ -155,7 +155,8 @@ object DataMessageProcessor { senderRecipient = senderRecipient, groupSecretParams = groupSecretParams, serverGuid = UuidUtil.getStringUUID(envelope.serverGuid, envelope.serverGuidBinary), - batchCache = batchCache + batchCache = batchCache, + receivedTime = receivedTime ) SignalTrace.endSection() diff --git a/app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.kt b/app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.kt index 1224f7f286..432db1a11b 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/messages/MessageContentProcessor.kt @@ -248,10 +248,11 @@ open class MessageContentProcessor(private val context: Context) { senderRecipient: Recipient, groupSecretParams: GroupSecretParams? = null, serverGuid: String? = null, - batchCache: BatchCache? = null + batchCache: BatchCache? = null, + receivedTime: Long? = null ): Gv2PreProcessResult { val preUpdateGroupRecord = batchCache?.groupRecordCache[groupId] ?: SignalDatabase.groups.getGroup(groupId) - val groupUpdateResult = updateGv2GroupFromServerOrP2PChange(context, timestamp, groupV2, preUpdateGroupRecord, groupSecretParams, serverGuid) + val groupUpdateResult = updateGv2GroupFromServerOrP2PChange(context, timestamp, groupV2, preUpdateGroupRecord, groupSecretParams, serverGuid, receivedTime) if (groupUpdateResult == null) { log(timestamp, "Ignoring GV2 message for group we are not currently in $groupId") return Gv2PreProcessResult.IGNORE @@ -304,13 +305,14 @@ open class MessageContentProcessor(private val context: Context) { groupV2: GroupContextV2, localRecord: Optional, groupSecretParams: GroupSecretParams? = null, - serverGuid: String? = null + serverGuid: String? = null, + receivedTime: Long? = null ): GroupUpdateResult? { return try { val signedGroupChange: ByteArray? = if (groupV2.hasSignedGroupChange) groupV2.signedGroupChange else null val updatedTimestamp = if (signedGroupChange != null) timestamp else timestamp + 1 if (groupV2.revision != null) { - GroupManager.updateGroupFromServer(context, groupV2.groupMasterKey, localRecord, groupSecretParams, groupV2.revision!!, updatedTimestamp, signedGroupChange, serverGuid) + GroupManager.updateGroupFromServer(context, groupV2.groupMasterKey, localRecord, groupSecretParams, groupV2.revision!!, updatedTimestamp, signedGroupChange, serverGuid, receivedTime) } else { warn(timestamp, "Ignore group update message without a revision") null diff --git a/app/src/main/java/org/thoughtcrime/securesms/mms/IncomingMessage.kt b/app/src/main/java/org/thoughtcrime/securesms/mms/IncomingMessage.kt index 671490403e..f9c0a55028 100644 --- a/app/src/main/java/org/thoughtcrime/securesms/mms/IncomingMessage.kt +++ b/app/src/main/java/org/thoughtcrime/securesms/mms/IncomingMessage.kt @@ -98,15 +98,14 @@ class IncomingMessage( ) } - @JvmStatic - fun groupUpdate(from: RecipientId, timestamp: Long, groupId: GroupId, update: GV2UpdateDescription, isNotifiable: Boolean, serverGuid: String?): IncomingMessage { + fun groupUpdate(from: RecipientId, timestamp: Long, groupId: GroupId, update: GV2UpdateDescription, isNotifiable: Boolean, serverGuid: String?, receivedTime: Long? = null): IncomingMessage { val messageExtras = MessageExtras(gv2UpdateDescription = update) val groupContext = MessageGroupContext(update.gv2ChangeDescription!!) return IncomingMessage( from = from, sentTimeMillis = timestamp, - receivedTimeMillis = System.currentTimeMillis(), + receivedTimeMillis = receivedTime?.minus(1) ?: System.currentTimeMillis(), serverTimeMillis = timestamp, serverGuid = serverGuid, groupId = groupId, diff --git a/app/src/test/java/org/thoughtcrime/securesms/database/SmsDatabaseTest_collapseJoinRequestEventsIfPossible.kt b/app/src/test/java/org/thoughtcrime/securesms/database/MessageTableTest_groupUpdates.kt similarity index 80% rename from app/src/test/java/org/thoughtcrime/securesms/database/SmsDatabaseTest_collapseJoinRequestEventsIfPossible.kt rename to app/src/test/java/org/thoughtcrime/securesms/database/MessageTableTest_groupUpdates.kt index 43734cde59..97b079090a 100644 --- a/app/src/test/java/org/thoughtcrime/securesms/database/SmsDatabaseTest_collapseJoinRequestEventsIfPossible.kt +++ b/app/src/test/java/org/thoughtcrime/securesms/database/MessageTableTest_groupUpdates.kt @@ -3,6 +3,8 @@ package org.thoughtcrime.securesms.database import android.app.Application import assertk.assertThat import assertk.assertions.isEqualTo +import assertk.assertions.isGreaterThan +import assertk.assertions.isLessThan import assertk.assertions.isNull import assertk.assertions.isPresent import io.mockk.every @@ -37,7 +39,7 @@ import java.util.UUID @Suppress("ClassName", "TestFunctionName") @RunWith(RobolectricTestRunner::class) @Config(manifest = Config.NONE, application = Application::class) -class SmsDatabaseTest_collapseJoinRequestEventsIfPossible { +class MessageTableTest_groupUpdates { @get:Rule val recipientRule = RecipientTestRule() @@ -295,6 +297,78 @@ class SmsDatabaseTest_collapseJoinRequestEventsIfPossible { } } + @Test + fun groupUpdateSortsAboveTriggeringMessage() { + val receivedTime = 5_000L + + val message = sms.insertMessageInbox( + IncomingMessage( + type = MessageType.NORMAL, + from = alice, + sentTimeMillis = receivedTime, + serverTimeMillis = receivedTime, + receivedTimeMillis = receivedTime, + body = "What up", + groupId = groupId, + isUnidentified = true + ) + ).get() + + val update = sms.insertMessageInbox(joinUpdateMessage(sentTime = 4_000L, receivedTime = receivedTime)).get() + + val messageRecord = sms.getMessageRecordOrNull(message.messageId)!! + val updateRecord = sms.getMessageRecordOrNull(update.messageId)!! + + assertThat(updateRecord.dateReceived, "update anchored just before triggering message").isEqualTo(receivedTime - 1) + assertThat(updateRecord.dateReceived, "update sorts above triggering message").isLessThan(messageRecord.dateReceived) + } + + @Test + fun groupUpdateSortsBelowOlderMessages() { + val olderMessage = sms.insertMessageInbox( + IncomingMessage( + type = MessageType.NORMAL, + from = alice, + sentTimeMillis = 1_000L, + serverTimeMillis = 1_000L, + receivedTimeMillis = 1_000L, + body = "What up", + groupId = groupId, + isUnidentified = true + ) + ).get() + + val update = sms.insertMessageInbox(joinUpdateMessage(sentTime = 500L, receivedTime = 5_000L)).get() + + val olderRecord = sms.getMessageRecordOrNull(olderMessage.messageId)!! + val updateRecord = sms.getMessageRecordOrNull(update.messageId)!! + + assertThat(updateRecord.dateReceived, "update sorts below older messages despite older sent time").isGreaterThan(olderRecord.dateReceived) + } + + private fun joinUpdateMessage(sentTime: Long, receivedTime: Long): IncomingMessage { + val updateContext = groupContext(masterKey = masterKey) { + change = groupChange(editor = aliceServiceId) { + addMember(bobServiceId) + } + } + + val updateDescription = GV2UpdateDescription( + gv2ChangeDescription = updateContext, + groupChangeUpdate = GroupsV2UpdateMessageConverter.translateDecryptedChangeUpdate(SignalStore.account.getServiceIds(), updateContext) + ) + + return IncomingMessage.groupUpdate( + from = alice, + timestamp = sentTime, + groupId = groupId, + update = updateDescription, + isNotifiable = true, + serverGuid = null, + receivedTime = receivedTime + ) + } + private fun smsMessage(sender: RecipientId, body: String? = ""): IncomingMessage { wallClock++ return IncomingMessage(