Skip to content

Commit 3f6a24d

Browse files
AndyScherzingerAndyScherzinger
committed
fix(chat): merge connected chat blocks atomically
The upsert -> getConnectedChatBlocks -> replaceConnectedChatBlocks sequence in ChatMessageSyncer.updateBlocks was not atomic: with the open-path delta, long poll/insurance, signaling and background catch-up all able to update the same conversation concurrently, two callers could each upsert a block and query connectivity before seeing the other's write, leaving overlapping blocks behind. The whole sequence now runs as a single Room @transaction in ChatBlocksDao.upsertAndMergeConnectedChatBlocks. getConnectedChatBlocks and deleteChatBlocks became suspend so no blocking DAO calls run inside the suspending transaction. Assisted-by: Claude Code:claude-fable-5 Signed-off-by: Andy Scherzinger <info@andy-scherzinger.de>
1 parent ea40bb0 commit 3f6a24d

6 files changed

Lines changed: 176 additions & 82 deletions

File tree

‎app/src/androidTest/java/com/nextcloud/talk/data/database/dao/ChatBlocksDaoTest.kt‎

Lines changed: 116 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -244,7 +244,7 @@ class ChatBlocksDaoTest {
244244
newestMessageId = searchedChatBlock.newestMessageId
245245
)
246246

247-
assertEquals(5, results.first().size)
247+
assertEquals(5, results.size)
248248
}
249249

250250
@Test
@@ -314,7 +314,7 @@ class ChatBlocksDaoTest {
314314
newestMessageId = searchedChatBlock.newestMessageId
315315
)
316316

317-
assertEquals(1, resultsForThreadIdNull.first().size)
317+
assertEquals(1, resultsForThreadIdNull.size)
318318

319319
val resultsForThreadId123 = chatBlocksDao.getConnectedChatBlocks(
320320
internalConversationId = conversation1.internalId,
@@ -323,7 +323,7 @@ class ChatBlocksDaoTest {
323323
newestMessageId = searchedChatBlock.newestMessageId
324324
)
325325

326-
assertEquals(2, resultsForThreadId123.first().size)
326+
assertEquals(2, resultsForThreadId123.size)
327327
}
328328

329329
@Test
@@ -384,7 +384,7 @@ class ChatBlocksDaoTest {
384384
threadId = null,
385385
oldestMessageId = 10,
386386
newestMessageId = 35
387-
).first()
387+
)
388388
assertEquals(3, connectedBlocks.size)
389389

390390
val mergedBlock = ChatBlockEntity(
@@ -412,10 +412,121 @@ class ChatBlocksDaoTest {
412412
threadId = null,
413413
oldestMessageId = 10,
414414
newestMessageId = 40
415-
).first().size
415+
).size
416416
)
417417
}
418418

419+
@Test
420+
fun testUpsertAndMergeConnectedChatBlocksMergesOverlappingBlocks() =
421+
runTest {
422+
val user = createUserEntity("account1", "Account 1")
423+
usersDao.saveUser(user)
424+
val account1 = usersDao.getUserWithUserId("account1").blockingGet()
425+
426+
conversationsDao.upsertConversations(
427+
account1.id,
428+
listOf(
429+
createConversationEntity(
430+
accountId = account1.id,
431+
token = "abc",
432+
roomName = "Conversation One"
433+
)
434+
)
435+
)
436+
437+
val conversation = conversationsDao.getConversationsForUser(account1.id).first()[0]
438+
439+
chatBlocksDao.upsertChatBlock(
440+
ChatBlockEntity(
441+
internalConversationId = conversation.internalId,
442+
accountId = conversation.accountId,
443+
token = conversation.token,
444+
threadId = null,
445+
oldestMessageId = 10,
446+
newestMessageId = 20,
447+
hasHistory = false
448+
)
449+
)
450+
chatBlocksDao.upsertChatBlock(
451+
ChatBlockEntity(
452+
internalConversationId = conversation.internalId,
453+
accountId = conversation.accountId,
454+
token = conversation.token,
455+
threadId = null,
456+
oldestMessageId = 25,
457+
newestMessageId = 35,
458+
hasHistory = true
459+
)
460+
)
461+
462+
// the new block overlaps both existing blocks, so all three must merge into one
463+
chatBlocksDao.upsertAndMergeConnectedChatBlocks(
464+
ChatBlockEntity(
465+
internalConversationId = conversation.internalId,
466+
accountId = conversation.accountId,
467+
token = conversation.token,
468+
threadId = null,
469+
oldestMessageId = 18,
470+
newestMessageId = 27,
471+
hasHistory = true
472+
)
473+
)
474+
475+
val blocks = chatBlocksDao.getChatBlocksForConversation(conversation.internalId)
476+
assertEquals(1, blocks.size)
477+
assertEquals(10L, blocks[0].oldestMessageId)
478+
assertEquals(35L, blocks[0].newestMessageId)
479+
assertEquals(false, blocks[0].hasHistory)
480+
}
481+
482+
@Test
483+
fun testUpsertAndMergeConnectedChatBlocksKeepsDisjointBlocksSeparate() =
484+
runTest {
485+
val user = createUserEntity("account1", "Account 1")
486+
usersDao.saveUser(user)
487+
val account1 = usersDao.getUserWithUserId("account1").blockingGet()
488+
489+
conversationsDao.upsertConversations(
490+
account1.id,
491+
listOf(
492+
createConversationEntity(
493+
accountId = account1.id,
494+
token = "abc",
495+
roomName = "Conversation One"
496+
)
497+
)
498+
)
499+
500+
val conversation = conversationsDao.getConversationsForUser(account1.id).first()[0]
501+
502+
chatBlocksDao.upsertChatBlock(
503+
ChatBlockEntity(
504+
internalConversationId = conversation.internalId,
505+
accountId = conversation.accountId,
506+
token = conversation.token,
507+
threadId = null,
508+
oldestMessageId = 10,
509+
newestMessageId = 20,
510+
hasHistory = true
511+
)
512+
)
513+
514+
chatBlocksDao.upsertAndMergeConnectedChatBlocks(
515+
ChatBlockEntity(
516+
internalConversationId = conversation.internalId,
517+
accountId = conversation.accountId,
518+
token = conversation.token,
519+
threadId = null,
520+
oldestMessageId = 30,
521+
newestMessageId = 40,
522+
hasHistory = true
523+
)
524+
)
525+
526+
val blocks = chatBlocksDao.getChatBlocksForConversation(conversation.internalId)
527+
assertEquals(2, blocks.size)
528+
}
529+
419530
private fun createUserEntity(userId: String, userName: String) =
420531
UserEntity(
421532
userId = userId,

‎app/src/main/java/com/nextcloud/talk/chat/data/network/ChatMessageSyncer.kt‎

Lines changed: 1 addition & 46 deletions
Original file line numberDiff line numberDiff line change
@@ -632,7 +632,7 @@ class ChatMessageSyncer @Inject constructor(
632632
newestMessageId = newestMessageIdForNewChatBlock,
633633
hasHistory = hasHistory
634634
)
635-
updateBlocks(target, newChatBlock)
635+
chatBlocksDao.upsertAndMergeConnectedChatBlocks(newChatBlock)
636636

637637
return chatMessageEntities
638638
}
@@ -826,51 +826,6 @@ class ChatMessageSyncer @Inject constructor(
826826
return blockContainingQueriedMessage
827827
}
828828

829-
suspend fun updateBlocks(target: SyncTarget, chatBlock: ChatBlockEntity) {
830-
chatBlocksDao.upsertChatBlock(chatBlock)
831-
832-
val connectedChatBlocks =
833-
chatBlocksDao.getConnectedChatBlocks(
834-
internalConversationId = target.internalConversationId,
835-
threadId = target.threadId,
836-
oldestMessageId = chatBlock.oldestMessageId,
837-
newestMessageId = chatBlock.newestMessageId
838-
).first()
839-
840-
if (connectedChatBlocks.size == 1) {
841-
Log.d(TAG, "This chatBlock is not connected to others")
842-
val chatBlockFromDb = connectedChatBlocks[0]
843-
Log.d(TAG, "chatBlockFromDb.oldestMessageId: " + chatBlockFromDb.oldestMessageId)
844-
Log.d(TAG, "chatBlockFromDb.newestMessageId: " + chatBlockFromDb.newestMessageId)
845-
} else if (connectedChatBlocks.size > 1) {
846-
Log.d(TAG, "Found " + connectedChatBlocks.size + " chat blocks that are connected")
847-
val oldestIdFromDbChatBlocks =
848-
connectedChatBlocks.minByOrNull { it.oldestMessageId }!!.oldestMessageId
849-
val newestIdFromDbChatBlocks =
850-
connectedChatBlocks.maxByOrNull { it.newestMessageId }!!.newestMessageId
851-
852-
val hasNoHistory = connectedChatBlocks.any { !it.hasHistory }
853-
val hasHistory = !hasNoHistory
854-
Log.d(TAG, "hasHistory = $hasHistory")
855-
856-
val newChatBlock = ChatBlockEntity(
857-
internalConversationId = target.internalConversationId,
858-
accountId = target.accountId,
859-
token = target.roomToken,
860-
threadId = target.threadId,
861-
oldestMessageId = oldestIdFromDbChatBlocks,
862-
newestMessageId = newestIdFromDbChatBlocks,
863-
hasHistory = hasHistory
864-
)
865-
chatBlocksDao.replaceConnectedChatBlocks(connectedChatBlocks, newChatBlock)
866-
Log.d(TAG, "A new chat block was created that covers all the range of the found chatblocks")
867-
Log.d(TAG, "new chatBlock - oldest MessageId: $oldestIdFromDbChatBlocks")
868-
Log.d(TAG, "new chatBlock - newest MessageId: $newestIdFromDbChatBlocks")
869-
} else {
870-
Log.d(TAG, "No chat block found ....")
871-
}
872-
}
873-
874829
companion object {
875830
val TAG: String = ChatMessageSyncer::class.java.simpleName
876831

‎app/src/main/java/com/nextcloud/talk/chat/data/network/OfflineFirstChatRepository.kt‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -458,7 +458,7 @@ class OfflineFirstChatRepository @Inject constructor(
458458
newestMessageId = newestId,
459459
hasHistory = true
460460
)
461-
syncer.updateBlocks(syncTarget, block)
461+
chatBlocksDao.upsertAndMergeConnectedChatBlocks(block)
462462

463463
ChatMessageRepository.MessagesRange(
464464
oldestMessageId = oldestId,

‎app/src/main/java/com/nextcloud/talk/data/database/dao/ChatBlocksDao.kt‎

Lines changed: 43 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,6 +7,7 @@
77

88
package com.nextcloud.talk.data.database.dao
99

10+
import android.util.Log
1011
import androidx.room.Dao
1112
import androidx.room.Delete
1213
import androidx.room.Insert
@@ -19,7 +20,7 @@ import kotlinx.coroutines.flow.Flow
1920
@Dao
2021
interface ChatBlocksDao {
2122
@Delete
22-
fun deleteChatBlocks(blocks: List<ChatBlockEntity>)
23+
suspend fun deleteChatBlocks(blocks: List<ChatBlockEntity>)
2324

2425
@Query(
2526
"""
@@ -54,12 +55,12 @@ interface ChatBlocksDao {
5455
ORDER BY newestMessageId ASC
5556
"""
5657
)
57-
fun getConnectedChatBlocks(
58+
suspend fun getConnectedChatBlocks(
5859
internalConversationId: String,
5960
threadId: Long?,
6061
oldestMessageId: Long,
6162
newestMessageId: Long
62-
): Flow<List<ChatBlockEntity>>
63+
): List<ChatBlockEntity>
6364

6465
@Query(
6566
"""
@@ -74,6 +75,41 @@ interface ChatBlocksDao {
7475
@Insert(onConflict = OnConflictStrategy.REPLACE)
7576
suspend fun upsertChatBlock(chatBlock: ChatBlockEntity)
7677

78+
/**
79+
* Upserts [chatBlock] and merges all chat blocks it overlaps into one covering their combined
80+
* range, as a single atomic operation.
81+
*
82+
* The open-path delta fetch, long polling, the insurance request, signaling and the background
83+
* catch-up can all update the blocks of the same conversation concurrently. Without one
84+
* transaction around upsert, connectivity query and merge, two callers could each upsert
85+
* their block and query connectivity before seeing the other's write, leaving overlapping
86+
* blocks behind.
87+
*/
88+
@Transaction
89+
suspend fun upsertAndMergeConnectedChatBlocks(chatBlock: ChatBlockEntity) {
90+
upsertChatBlock(chatBlock)
91+
92+
val connectedChatBlocks = getConnectedChatBlocks(
93+
internalConversationId = chatBlock.internalConversationId,
94+
threadId = chatBlock.threadId,
95+
oldestMessageId = chatBlock.oldestMessageId,
96+
newestMessageId = chatBlock.newestMessageId
97+
)
98+
if (connectedChatBlocks.size > 1) {
99+
val mergedBlock = chatBlock.copy(
100+
oldestMessageId = connectedChatBlocks.minOf { it.oldestMessageId },
101+
newestMessageId = connectedChatBlocks.maxOf { it.newestMessageId },
102+
hasHistory = connectedChatBlocks.all { it.hasHistory }
103+
)
104+
replaceConnectedChatBlocks(connectedChatBlocks, mergedBlock)
105+
Log.d(
106+
TAG,
107+
"Merged ${connectedChatBlocks.size} connected chat blocks into " +
108+
"${mergedBlock.oldestMessageId}..${mergedBlock.newestMessageId}"
109+
)
110+
}
111+
}
112+
77113
@Transaction
78114
suspend fun replaceConnectedChatBlocks(connectedBlocks: List<ChatBlockEntity>, mergedBlock: ChatBlockEntity) {
79115
val newestConnectedBlock = connectedBlocks.maxByOrNull { it.newestMessageId }
@@ -131,4 +167,8 @@ interface ChatBlocksDao {
131167
"""
132168
)
133169
suspend fun getChatBlocksForConversation(internalConversationId: String): List<ChatBlockEntity>
170+
171+
companion object {
172+
private const val TAG = "ChatBlocksDao"
173+
}
134174
}

‎app/src/main/java/com/nextcloud/talk/utils/preview/ComposePreviewUtilsDaos.kt‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -274,7 +274,7 @@ class DummyConversationDaoImpl : ConversationsDao {
274274
}
275275

276276
class DummyChatBlocksDaoImpl : ChatBlocksDao {
277-
override fun deleteChatBlocks(blocks: List<ChatBlockEntity>) {
277+
override suspend fun deleteChatBlocks(blocks: List<ChatBlockEntity>) {
278278
/* */
279279
}
280280

@@ -284,12 +284,12 @@ class DummyChatBlocksDaoImpl : ChatBlocksDao {
284284
messageId: Long
285285
): Flow<List<ChatBlockEntity>> = flowOf()
286286

287-
override fun getConnectedChatBlocks(
287+
override suspend fun getConnectedChatBlocks(
288288
internalConversationId: String,
289289
threadId: Long?,
290290
oldestMessageId: Long,
291291
newestMessageId: Long
292-
): Flow<List<ChatBlockEntity>> = flowOf()
292+
): List<ChatBlockEntity> = emptyList()
293293

294294
override fun getNewestMessageIdFromChatBlocks(internalConversationId: String, threadId: Long?): Long = 0L
295295

0 commit comments

Comments
 (0)