From de6afbb657b36e08b2bc68f319010a4777cd7d8e Mon Sep 17 00:00:00 2001 From: Brandon McAnsh Date: Fri, 6 Nov 2026 16:31:14 -0500 Subject: [PATCH] fix(chat): apply every GetDelta batch, not only the first GetDelta streams a catch-up as MessageBatch responses of at most 100 messages, and every response carries latest_sequence, the head as of stream open. performDeltaSync took .first(), wrote that batch, then set the cursor to the head. Anything past the first 100 messages was never fetched, and no later sync asked for it, because the cursor already said it was there. The stream is now consumed to the end. Messages are written 500 at a time (DELTA_FLUSH_MESSAGES), and each write moves the cursor to the checkpoint_sequence of the last batch it holds. The cursor reaches the head only when the stream completes. A stream that fails, ends early or runs past 60 s keeps what it delivered, and the next sync resumes from that checkpoint. The timeout exists because the coordinator runs the sync inline in its event handling, and an open stream would hold it. On RESET_REQUIRED after some batches, the delivered batches' checkpoint is kept before LoadMessages runs. loadMessages only moves the cursor forward, so this marks rows that are stored and nothing more. DeltaSyncBatchesTest drives the delegate with a multi-batch fake. Its first three cases failed before this change: a 12-batch catch-up wrote 100 of 1,200 messages. --- .../internal/delegates/EventStreamDelegate.kt | 100 +++++++++--- .../shared/chat/DeltaSyncBatchesTest.kt | 144 ++++++++++++++++++ 2 files changed, 219 insertions(+), 25 deletions(-) create mode 100644 apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/DeltaSyncBatchesTest.kt diff --git a/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt b/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt index 5f155d4d09..442aa4f839 100644 --- a/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt +++ b/apps/flipcash/shared/chat/src/commonMain/kotlin/com/flipcash/shared/chat/internal/delegates/EventStreamDelegate.kt @@ -40,6 +40,7 @@ import com.getcode.opencode.model.core.ID import com.getcode.opencode.model.financial.Fiat import com.getcode.utils.TraceType import com.getcode.utils.trace +import kotlinx.coroutines.CancellationException import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Job import kotlinx.coroutines.awaitCancellation @@ -47,10 +48,10 @@ import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.distinctUntilChanged -import kotlinx.coroutines.flow.first import kotlinx.coroutines.flow.firstOrNull import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.receiveAsFlow +import kotlinx.coroutines.flow.transformWhile import kotlinx.coroutines.launch import kotlinx.coroutines.withTimeoutOrNull import javax.inject.Inject @@ -111,6 +112,12 @@ class EventStreamDelegate @Inject constructor( private val GAP_FILL_DELAY = 2.seconds private val HEARTBEAT_INTERVAL = 30.seconds private val REOPEN_BACKOFF_BASE = 1.seconds + + /** How many caught-up messages are written per transaction; `GetDelta` sends at most 100 per batch. */ + internal const val DELTA_FLUSH_MESSAGES = 500 + + /** Bounds a catch-up stream, which runs inline in the coordinator's event handling. */ + private val DELTA_SYNC_TIMEOUT = 60.seconds } sealed interface Event { @@ -301,39 +308,82 @@ class EventStreamDelegate @Inject constructor( * The cursor is read from `chat_metadata`, which holds what the client has actually applied: * nothing writes the server's head there — a feed sync refreshes only the server-owned columns * (`ChatMetadataDao.upsert`), and the mapper drops the head on the way in. + * + * `GetDelta` streams the catch-up as batches of at most 100 messages, and only the completed + * stream reaches the head (`latest_sequence`). Batches are buffered and written + * [DELTA_FLUSH_MESSAGES] at a time, each write followed by the checkpoint it reached, so a long + * catch-up costs observers a few invalidations rather than one per batch. A stream that fails, + * drops or times out keeps what it delivered, and the next sync resumes from that checkpoint. */ internal suspend fun performDeltaSync(chatId: ChatId) { val afterSequence = metadataDataSource.getLatestEventSequence(chatId) trace(tag = TAG, message = "Delta sync for $chatId from sequence $afterSequence", type = TraceType.Process) - try { - val result = messagingController.getDelta(chatId, afterSequence).first() - result - .onSuccess { delta -> - if (delta.messages.isNotEmpty()) { - messageDataSource.upsert(chatId, delta.messages) + val buffered = mutableListOf() + var checkpoint = afterSequence + var head: Long? = null + var applied = 0 + var failure: Throwable? = null + + // One write that only moves forward: the cursor, and the newest message only when it is + // newer than the stored one. See ChatMetadataDao.applyCatchUp. + suspend fun flush(cursor: Long) { + if (buffered.isNotEmpty()) messageDataSource.upsert(chatId, buffered.toList()) + metadataDataSource.applyCatchUp( + chatId, + if (cursor > afterSequence) cursor else 0L, + buffered.maxByOrNull { it.messageId }, + ) + applied += buffered.size + buffered.clear() + } + + val completed = try { + withTimeoutOrNull(DELTA_SYNC_TIMEOUT) { + messagingController.getDelta(chatId, afterSequence) + .transformWhile { result -> + result.onSuccess { emit(it) }.onFailure { failure = it } + result.isSuccess } - // One write that only moves forward: the cursor, and the newest message only - // when it is newer than the stored one. See ChatMetadataDao.applyCatchUp. - metadataDataSource.applyCatchUp( - chatId, - if (delta.latestSequence > afterSequence) delta.latestSequence else 0L, - delta.messages.maxByOrNull { it.messageId }, - ) - sequenceTracker.resetTo(chatId, delta.latestSequence) - trace(tag = TAG, message = "Delta sync complete: ${delta.messages.size} messages, sequence ${delta.latestSequence}", type = TraceType.Process) - } - .onFailure { error -> - if (error is GetDeltaError.ResetRequired) { - trace(tag = TAG, message = "Delta sync reset required for $chatId, falling back to full load", type = TraceType.Process) - sequenceTracker.clear(chatId) - _events.send(Event.LoadMessages(chatId)) - } else { - trace(tag = TAG, message = "Delta sync failed for $chatId: ${error.message}", type = TraceType.Error) + .collect { delta -> + buffered += delta.messages + checkpoint = maxOf(checkpoint, delta.checkpointSequence) + head = delta.latestSequence + if (buffered.size >= DELTA_FLUSH_MESSAGES) flush(checkpoint) } - } + true + } ?: false + } catch (e: CancellationException) { + throw e } catch (e: Exception) { trace(tag = TAG, message = "Delta sync exception for $chatId: ${e.message}", type = TraceType.Error) + false + } + + val latest = head + if (completed && failure == null && latest != null) { + flush(maxOf(latest, checkpoint)) + sequenceTracker.resetTo(chatId, latest) + trace(tag = TAG, message = "Delta sync complete: $applied messages, sequence $latest", type = TraceType.Process) + return + } + + // Every batch received before the stream stopped was whole; keep them, and resume from + // their checkpoint rather than the head they never reached. + flush(checkpoint) + val error = failure + when { + error is GetDeltaError.ResetRequired -> { + trace(tag = TAG, message = "Delta sync reset required for $chatId, falling back to full load", type = TraceType.Process) + sequenceTracker.clear(chatId) + _events.send(Event.LoadMessages(chatId)) + } + error != null -> + trace(tag = TAG, message = "Delta sync failed for $chatId: ${error.message}", type = TraceType.Error) + completed -> + trace(tag = TAG, message = "Delta sync for $chatId ended without a response", type = TraceType.Error) + else -> + trace(tag = TAG, message = "Delta sync for $chatId stopped at sequence $checkpoint after $applied messages", type = TraceType.Error) } } diff --git a/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/DeltaSyncBatchesTest.kt b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/DeltaSyncBatchesTest.kt new file mode 100644 index 0000000000..1073d3f082 --- /dev/null +++ b/apps/flipcash/shared/chat/src/commonTest/kotlin/com/flipcash/shared/chat/DeltaSyncBatchesTest.kt @@ -0,0 +1,144 @@ +package com.flipcash.shared.chat + +import com.flipcash.app.persistence.sources.ChatMessageDataSource +import com.flipcash.app.persistence.sources.ChatMetadataDataSource +import com.flipcash.services.controllers.ChatMessagingController +import com.flipcash.services.controllers.EventStreamingController +import com.flipcash.services.models.GetDeltaError +import com.flipcash.services.models.chat.ChatId +import com.flipcash.services.models.chat.ChatMessage +import com.flipcash.services.models.chat.MessageContent +import com.flipcash.services.repository.DeltaUpdate +import com.flipcash.services.user.UserManager +import com.flipcash.shared.chat.internal.ChatStateHolder +import com.flipcash.shared.chat.internal.delegates.EventStreamDelegate +import io.mockk.coEvery +import io.mockk.every +import io.mockk.mockk +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.emptyFlow +import kotlinx.coroutines.flow.flow +import kotlinx.coroutines.test.runTest +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue +import kotlin.time.Instant + +/** + * `GetDelta` is a server stream: the catch-up arrives as a run of `MessageBatch`es of up to 100 + * messages each, and the cursor may only move to `latest_sequence` once the stream completes. + */ +class DeltaSyncBatchesTest { + + private val chat = ChatId("aabbccdd") + + private val written = mutableListOf>() + private val cursors = mutableListOf() + private var stored = 0L + + private val messageDataSource = mockk(relaxed = true).also { + coEvery { it.upsert(chat, any()) } answers { written += secondArg>() } + } + private val metadataDataSource = mockk(relaxed = true).also { + coEvery { it.getLatestEventSequence(chat) } answers { stored } + coEvery { it.applyCatchUp(chat, any(), any()) } answers { + val sequence = secondArg() + if (sequence > 0) { + cursors += sequence + stored = maxOf(stored, sequence) + } + } + } + private val messagingController = mockk(relaxed = true) + + private val delegate = EventStreamDelegate( + eventStreamingController = mockk(relaxed = true).also { + every { it.streamFailures } returns emptyFlow() + }, + messagingController = messagingController, + metadataDataSource = metadataDataSource, + messageDataSource = messageDataSource, + memberDataSource = mockk(relaxed = true), + tokenCoordinator = mockk(relaxed = true), + userManager = mockk(relaxed = true), + stateHolder = ChatStateHolder(), + analytics = mockk(relaxed = true), + exchange = mockk(relaxed = true), + ) + + private fun message(sequence: Long) = ChatMessage( + messageId = sequence, + senderId = listOf(2), + content = listOf(MessageContent.Text("m$sequence")), + timestamp = Instant.fromEpochMilliseconds(sequence * 1000), + unreadSeq = sequence, + eventSequence = sequence, + ) + + /** [count] batches of [size] messages, sequences 1 through count * size. */ + private fun batches(count: Int, size: Int = 100): List { + val head = (count * size).toLong() + return (0 until count).map { b -> + val first = b * size + 1L + val last = first + size - 1 + DeltaUpdate( + messages = (first..last).map(::message), + latestSequence = head, + checkpointSequence = last, + ) + } + } + + private fun stream(responses: List, thenFail: Throwable? = null): Flow> = flow { + for (response in responses) emit(Result.success(response)) + if (thenFail != null) throw thenFail + } + + @Test + fun `every batch in the stream is written and the cursor ends at the head`() = runTest { + every { messagingController.getDelta(chat, 0L, any()) } returns stream(batches(3)) + + delegate.performDeltaSync(chat) + + assertEquals((1L..300L).toList(), written.flatten().map { it.messageId }.sorted()) + assertEquals(300L, stored) + } + + @Test + fun `a stream that drops keeps what it delivered and resumes from the last checkpoint`() = runTest { + every { messagingController.getDelta(chat, 0L, any()) } returns + stream(batches(3).take(2), thenFail = RuntimeException("stream reset")) + + delegate.performDeltaSync(chat) + + assertEquals((1L..200L).toList(), written.flatten().map { it.messageId }.sorted()) + assertEquals(200L, stored, "the cursor must not move past what was applied") + } + + @Test + fun `a reset after the first batch keeps that batch and does not jump to the head`() = runTest { + val responses = batches(3) + every { messagingController.getDelta(chat, 0L, any()) } returns flow { + emit(Result.success(responses[0])) + emit(Result.failure(GetDeltaError.ResetRequired())) + emit(Result.success(responses[1])) + } + + delegate.performDeltaSync(chat) + + assertEquals((1L..100L).toList(), written.flatten().map { it.messageId }.sorted()) + assertEquals(100L, stored) + } + + @Test + fun `a long catch-up is written in fewer transactions than it has batches`() = runTest { + every { messagingController.getDelta(chat, 0L, any()) } returns stream(batches(12)) + + delegate.performDeltaSync(chat) + + assertEquals(1200, written.sumOf { it.size }) + assertTrue(written.size < 12, "wrote ${written.size} times for 12 batches") + assertEquals(cursors.sorted(), cursors, "the cursor only moves forward") + assertEquals(1200L, stored) + } +}