From 9eeaeaf49020c90fe058dec1fed9daeb99b494c5 Mon Sep 17 00:00:00 2001 From: tdrkDev Date: Wed, 5 Aug 2026 16:23:35 +0700 Subject: [PATCH 1/2] data: stop silently dropping TDLib updates Updates were fanned out through a single MutableSharedFlow(replay = 3, extraBufferCapacity = 64, DROP_OLDEST) with 18 independent subscriptions on it. Several of those collectors perform suspending Room writes or TDLib requests inline, so they are orders of magnitude slower than the arrival rate. Once any subscriber falls more than 67 values behind, DROP_OLDEST fast-forwards it past everything it had not consumed; tryEmit still returns true and nothing is logged, so the loss is invisible to both producer and consumer. The flow was never blocking TDLib - tryEmit with a non-SUSPEND overflow policy cannot fail or suspend - it was hiding the fact that downstream could not keep up. Losing these updates is not recoverable: updateNewChat is the only introduction of a chat object, and after it is missed every later cache.updateChat for that chat is a silent no-op, so the chat is blackholed permanently. Under a 3000 updates/s burst the persisted mirror ended up with 139 of 300 chats and 121 of 300 correct titles. Introduce TdUpdatePipeline, which offers two explicit contracts: - lane(): lossless and strictly ordered. Private unbounded queue, private worker, per-update exception isolation. For consumers that write to Room, mutate ChatCache, or issue TDLib requests. - updates: lossy by design, DROP_OLDEST. For consumers that render state they can re-read. Ingestion from the TDLib callback thread is one non-suspending enqueue; a dedicated pump thread does the fan-out. That thread also delivers every query result, so it must stay O(1): fanning out on it directly measured 38us/update with eight lanes attached, against a ~8us total budget, because each trySend has to resume a parked receiver. Through the pump it is 6-9ns and flat in the number of lanes. There is deliberately no bounded lane. Channel(capacity, DROP_OLDEST) does not report what it discarded, which is the defect being fixed. Migrated to lanes: messages, chat-list, users, notifications, files and notification-settings. Story, TelegramLink, Sponsor, Sticker, AttachMenuBot, Privacy, ConnectionManager and Auth stay on the observation flow - cheap handlers over state that is re-fetched or independently defended. The users lane carries updateUser and updateUserStatus together; status is handed to the existing LatestByKeyBatcher, which conflates per user id, so the lane stays cheap without losing a user's presence entirely. Also replaces MessageRepositoryImpl's `.catch { Log.e("CRITICAL: Update loop died") }`, which ended the subscription permanently on the first exception, and moves TdMessageRemoteDataSource's remaining rendezvous MutableSharedFlow() declarations onto OrderedEventFlow - with the default arguments every emit parked until all subscribers had taken the value, piling up ~5200 coroutines under load. Upload progress keeps a bounded DROP_OLDEST buffer instead, since progress ticks are conflatable. Verified: 300/300 chats and titles persisted under the same burst, ordering preserved, callback thread cost per update 4.84us -> 2.25us, request results delivered in the window 165 -> 269. :data, :presentation and :app compile; 129 data unit tests pass, including 8 new TdUpdatePipeline contract tests. Not addressed here: the loss-critical repositories are still lazily constructed, so their lanes register late and miss anything TDLib sent beforehand. Lanes fix loss under load, not late registration. Co-Authored-By: Claude Fable 5 --- .../remote/TdMessageRemoteDataSource.kt | 25 +- .../java/org/monogram/data/di/TdLibClient.kt | 45 +++- .../monogram/data/di/TdNotificationManager.kt | 15 +- .../org/monogram/data/di/TdUpdatePipeline.kt | 231 ++++++++++++++++++ .../java/org/monogram/data/di/dataModule.kt | 2 +- .../monogram/data/gateway/TelegramGateway.kt | 21 ++ .../data/gateway/TelegramGatewayImpl.kt | 15 ++ .../monogram/data/gateway/UpdateDispatcher.kt | 27 ++ .../data/gateway/UpdateDispatcherImpl.kt | 86 ++++--- .../monogram/data/infra/FileUpdateHandler.kt | 16 +- .../repository/ChatsListRepositoryImpl.kt | 15 +- .../data/repository/MessageRepositoryImpl.kt | 28 +-- .../NotificationSettingsRepositoryImpl.kt | 76 +++--- .../repository/user/UserUpdateSynchronizer.kt | 29 ++- .../remote/TdChatRemoteSourceTest.kt | 8 + .../remote/TdGifRemoteSourceTest.kt | 8 + .../remote/TdStickerRemoteSourceTest.kt | 8 + .../monogram/data/di/TdUpdatePipelineTest.kt | 229 +++++++++++++++++ .../data/gateway/UpdateDispatcherImplTest.kt | 10 + .../data/infra/ConnectionManagerTest.kt | 8 + .../data/infra/FileDownloadQueueTest.kt | 8 + .../data/infra/FileUpdateHandlerTest.kt | 27 +- .../data/infra/SponsorSyncManagerTest.kt | 8 + .../TelegramLinkRepositoryImplTest.kt | 16 ++ .../monogram/data/testing/FakeUpdateLane.kt | 28 +++ 25 files changed, 863 insertions(+), 126 deletions(-) create mode 100644 data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt create mode 100644 data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt create mode 100644 data/src/test/java/org/monogram/data/testing/FakeUpdateLane.kt diff --git a/data/src/main/java/org/monogram/data/datasource/remote/TdMessageRemoteDataSource.kt b/data/src/main/java/org/monogram/data/datasource/remote/TdMessageRemoteDataSource.kt index ee15eafdb..b0febf422 100644 --- a/data/src/main/java/org/monogram/data/datasource/remote/TdMessageRemoteDataSource.kt +++ b/data/src/main/java/org/monogram/data/datasource/remote/TdMessageRemoteDataSource.kt @@ -92,11 +92,22 @@ class TdMessageRemoteDataSource( private val refreshJobs = ConcurrentHashMap, Job>() private val missingMessageCooldownUntil = ConcurrentHashMap, Long>() private val sendQueue = Channel Unit>(Channel.BUFFERED) - override val newMessageFlow = MutableSharedFlow() - override val messageEditedFlow = MutableSharedFlow() + // These are fed from `scope.launch { ... }` inside update handling. With the default + // arguments (replay 0, no buffer, SUSPEND) a MutableSharedFlow is a rendezvous channel: + // every emit parks until *all* subscribers have taken the value, which piled emitters up + // without bound during bursts. OrderedEventFlow.enqueue is non-suspending and lossless, + // so the event streams go through it; progress ticks are conflatable and get an explicit + // bounded buffer instead. + private val newMessages = OrderedEventFlow(scope) + override val newMessageFlow = newMessages.events + private val messageEdits = OrderedEventFlow(scope) + override val messageEditedFlow = messageEdits.events private val messageReads = OrderedEventFlow(scope) override val messageReadFlow = messageReads.events - override val messageUploadProgressFlow = MutableSharedFlow() + override val messageUploadProgressFlow = MutableSharedFlow( + extraBufferCapacity = 64, + onBufferOverflow = BufferOverflow.DROP_OLDEST + ) private val fileDownloads = OrderedEventFlow(scope) override val fileDownloadFlow = fileDownloads.events private val messageDownloads = OrderedEventFlow(scope) @@ -1992,7 +2003,7 @@ class TdMessageRemoteDataSource( scope.launch(dispatcherProvider.io) { try { val model = mapMessageToModel(message) - newMessageFlow.emit(model) + newMessages.enqueue(model) } catch (e: Exception) { Log.e("TdMessageRemote", "Error mapping NewMessage", e) } } } @@ -2039,7 +2050,7 @@ class TdMessageRemoteDataSource( errorCode = update.error?.code ?: 0 ) ) - messageEditedFlow.emit(model) + messageEdits.enqueue(model) } } is TdApi.UpdateMessageContent -> { @@ -2216,7 +2227,7 @@ class TdMessageRemoteDataSource( if (messageId == 0L) return val msg = cache.getMessage(chatId, messageId) ?: return val model = mapMessageToModel(msg) - messageEditedFlow.emit(model) + messageEdits.enqueue(model) } private suspend fun mapMessageToModel(message: TdApi.Message): MessageModel { @@ -2445,7 +2456,7 @@ class TdMessageRemoteDataSource( delay(150) val msg = cache.getMessage(chatId, messageId) ?: return@launch try { - messageEditedFlow.emit(mapMessageToModel(msg)) + messageEdits.enqueue(mapMessageToModel(msg)) } catch (e: CancellationException) { throw e } catch (e: Exception) { diff --git a/data/src/main/java/org/monogram/data/di/TdLibClient.kt b/data/src/main/java/org/monogram/data/di/TdLibClient.kt index e5c0070fc..69650ce93 100644 --- a/data/src/main/java/org/monogram/data/di/TdLibClient.kt +++ b/data/src/main/java/org/monogram/data/di/TdLibClient.kt @@ -1,9 +1,8 @@ package org.monogram.data.di import android.util.Log -import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.delay -import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.asStateFlow @@ -16,16 +15,21 @@ import org.monogram.data.BuildConfig import org.monogram.data.gateway.TdLibException import org.monogram.data.gateway.isExpectedProxyFailure import java.util.concurrent.ConcurrentHashMap +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext import kotlin.coroutines.resume internal class TdLibClient { private val TAG = "TdLibClient" private val retryAfterUntilMsByScope = ConcurrentHashMap() - private val _updates = MutableSharedFlow( - replay = 3, - extraBufferCapacity = 64, - onBufferOverflow = BufferOverflow.DROP_OLDEST - ) + + /** + * Authoritative ingestion for TDLib updates. + * + * Consumers that own durable state subscribe with [lane] (lossless, ordered); + * consumers that only render subscribe to [updates] (conflating). + */ + private val pipeline = TdUpdatePipeline() private val _isAuthenticated = MutableStateFlow(false) val isAuthenticated = _isAuthenticated.asStateFlow() @@ -44,17 +48,40 @@ internal class TdLibClient { } } - val updates: SharedFlow = _updates + /** + * Observation stream. Conflates under load, so it must not be used to drive durable + * state; use [lane] for that. + */ + val updates: SharedFlow = pipeline.updates + + /** + * Lossless, strictly ordered subscription for consumers that own durable state: + * Room writes, [org.monogram.data.chats.ChatCache] mutation, or TDLib requests. + * + * Register during startup — updates delivered before the lane exists are not replayed. + */ + fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext = EmptyCoroutineContext, + filter: (TdApi.Update) -> Boolean = { true }, + handler: suspend (TdApi.Update) -> Unit, + ): TdUpdatePipeline.Lane = pipeline.lane(name, scope, context, filter, handler) + + /** Diagnostics: ingest/lane backlogs, processed counts and handler failures. */ + fun updateMetrics(): String = pipeline.metrics() private val client = Client.create( { result -> if (result is TdApi.Update) { + // Kept on the callback thread on purpose: these gate sendSuspend, so they + // must not depend on the update pipeline making progress. if (result is TdApi.UpdateAuthorizationState) { val state = result.authorizationState _isInitialized.value = state !is TdApi.AuthorizationStateWaitTdlibParameters _isAuthenticated.value = state is TdApi.AuthorizationStateReady } - _updates.tryEmit(result) + pipeline.submit(result) } }, { error -> diff --git a/data/src/main/java/org/monogram/data/di/TdNotificationManager.kt b/data/src/main/java/org/monogram/data/di/TdNotificationManager.kt index 11b956853..7215bd7b6 100644 --- a/data/src/main/java/org/monogram/data/di/TdNotificationManager.kt +++ b/data/src/main/java/org/monogram/data/di/TdNotificationManager.kt @@ -167,14 +167,13 @@ class TdNotificationManager( } } - scope.launch { - updates.all.collect { update -> - runCatching { - handleCoreUpdate(update) - }.onFailure { - Log.e(TAG, "Failed to handle update ${update.javaClass.simpleName}", it) - } - } + // handleCoreUpdate issues TDLib requests inline (getChat, membership checks), so + // this consumer is orders of magnitude slower than the update rate and would be + // the first to be conflated away on the observation flow. updateNotificationGroup + // carries added/removed deltas and updateActiveNotifications arrives exactly once + // before them, so none of it may be dropped or reordered. + updates.lane(name = "notifications", scope = scope) { update -> + handleCoreUpdate(update) } scope.launch { diff --git a/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt b/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt new file mode 100644 index 000000000..2ff826221 --- /dev/null +++ b/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt @@ -0,0 +1,231 @@ +package org.monogram.data.di + +import android.util.Log +import kotlinx.coroutines.CancellationException +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.asCoroutineDispatcher +import kotlinx.coroutines.cancel +import kotlinx.coroutines.channels.BufferOverflow +import kotlinx.coroutines.channels.Channel +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.launch +import org.drinkless.tdlib.TdApi +import java.util.concurrent.CopyOnWriteArrayList +import java.util.concurrent.Executors +import java.util.concurrent.atomic.AtomicLong +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext + +/** + * Fan-out for TDLib updates, with two deliberately different delivery contracts. + * + * [submit] is called on TDLib's single "TDLib thread" (`Client.ResponseReceiver`, see + * `Client.java`). That same thread also delivers every query result, so it must never + * block and must stay O(1): blocking it stalls all in-flight requests, and TDLib's + * native output queue then grows without bound. [submit] therefore performs exactly one + * non-suspending enqueue and nothing else; a dedicated pump thread does the fan-out. + * (Fanning out on the callback thread instead was measured at ~38us/update with eight + * lanes attached, against a ~8us total budget.) + * + * Two ways to consume: + * + * - [lane] — **lossless and strictly ordered.** Private unbounded queue, private worker, + * per-update exception isolation. Use it whenever the handler writes to Room, mutates + * [org.monogram.data.chats.ChatCache], or issues a TDLib request. The order a lane sees + * is exactly TDLib's delivery order. + * - [updates] — **lossy by design.** A shared [SharedFlow] with `DROP_OLDEST`. Use it only + * for consumers that render state they can re-read; a drop there costs a redraw, never a + * state transition. + * + * There is deliberately no bounded lane. A bounded lane would silently discard updates, + * which is the defect this class exists to remove. + */ +internal class TdUpdatePipeline { + + private val ingest = Channel(Channel.UNLIMITED) + + private val _updates = MutableSharedFlow( + replay = OBSERVER_REPLAY, + extraBufferCapacity = OBSERVER_BUFFER, + onBufferOverflow = BufferOverflow.DROP_OLDEST + ) + + /** Observation only — may conflate. Durable consumers must use [lane]. */ + val updates: SharedFlow = _updates.asSharedFlow() + + private val lanes = CopyOnWriteArrayList() + + private val submitted = AtomicLong() + private val dispatched = AtomicLong() + private val rejected = AtomicLong() + + private val pumpExecutor = Executors.newSingleThreadExecutor { runnable -> + Thread(runnable, PUMP_THREAD_NAME).apply { isDaemon = true } + } + private val pumpScope = CoroutineScope(SupervisorJob() + pumpExecutor.asCoroutineDispatcher()) + + init { + pumpScope.launch { + for (update in ingest) { + // CopyOnWriteArrayList: indexed access, no iterator allocation on the hot path. + for (index in lanes.indices) { + lanes.getOrNull(index)?.offer(update) + } + _updates.tryEmit(update) + val count = dispatched.incrementAndGet() + if (count % BACKLOG_SAMPLE_EVERY == 0L) reportBacklogIfHigh() + } + } + } + + /** + * Called on the TDLib callback thread for every update. One unbounded enqueue, + * measured at single-digit nanoseconds and independent of the number of lanes. + */ + fun submit(update: TdApi.Update) { + submitted.incrementAndGet() + // UNLIMITED: only fails once the pipeline has been shut down. + if (ingest.trySend(update).isFailure) { + rejected.incrementAndGet() + } + } + + /** + * Registers a lossless, strictly ordered consumer. The lane stops and deregisters + * when [scope] is cancelled. + * + * Register lanes during application startup. Updates delivered before a lane exists + * are not retained for it, so a durable consumer that is constructed lazily will miss + * everything TDLib sent beforehand. + * + * @param filter evaluated on the pump thread; keep it to cheap type checks. + * @param context extra context for the worker, e.g. `Dispatchers.IO` for a lane that + * writes to Room. Defaults to [scope]'s dispatcher. + */ + fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext = EmptyCoroutineContext, + filter: (TdApi.Update) -> Boolean = { true }, + handler: suspend (TdApi.Update) -> Unit, + ): Lane { + val lane = Lane(name, filter) + // Register before starting the worker: if `scope` is already cancelled the worker + // completes immediately, and its completion handler must be able to find the lane. + // Otherwise the lane would linger with nothing draining its queue. + lanes.add(lane) + val job = scope.launch(context) { + for (update in lane.queue) { + // Per-update isolation. `.catch { }` on a Flow ends the subscription for + // good; this keeps the lane alive and counts the failure instead. + try { + handler(update) + } catch (e: CancellationException) { + throw e + } catch (e: Throwable) { + lane.failures.incrementAndGet() + Log.e(TAG, "lane '$name' failed on ${update.javaClass.simpleName}", e) + } + lane.processed.incrementAndGet() + } + } + lane.worker = job + job.invokeOnCompletion { cause -> + lanes.remove(lane) + lane.queue.close() + if (cause != null && cause !is CancellationException) { + Log.e(TAG, "lane '$name' terminated unexpectedly", cause) + } + } + return lane + } + + /** Snapshot for diagnostics; safe to call from any thread. */ + fun metrics(): String = buildString { + append("submitted=").append(submitted.get()) + append(" dispatched=").append(dispatched.get()) + append(" ingestBacklog=").append(submitted.get() - dispatched.get()) + if (rejected.get() > 0) append(" rejected=").append(rejected.get()) + append(" observers=").append(_updates.subscriptionCount.value) + for (lane in lanes) { + append(" | ").append(lane.name) + .append(": backlog=").append(lane.backlog()) + .append(" processed=").append(lane.processed.get()) + .append(" failures=").append(lane.failures.get()) + } + } + + fun shutdown() { + ingest.close() + lanes.forEach { it.cancel() } + pumpScope.cancel() + pumpExecutor.shutdown() + } + + private fun reportBacklogIfHigh() { + val ingestBacklog = submitted.get() - dispatched.get() + var worstLane: Lane? = null + var worstBacklog = 0L + for (lane in lanes) { + val backlog = lane.backlog() + if (backlog > worstBacklog) { + worstBacklog = backlog + worstLane = lane + } + } + if (worstBacklog < BACKLOG_WARN_AT && ingestBacklog < BACKLOG_WARN_AT) return + Log.w(TAG, "update backlog is high (worst lane '${worstLane?.name}'): ${metrics()}") + } + + internal class Lane( + val name: String, + private val filter: (TdApi.Update) -> Boolean, + ) { + // Unbounded on purpose: a lane exists precisely because its consumer must not lose + // updates. Backlog is reported through [metrics] rather than being discarded. + val queue = Channel(Channel.UNLIMITED) + val queued = AtomicLong() + val processed = AtomicLong() + val failures = AtomicLong() + + @Volatile + var worker: Job? = null + + fun offer(update: TdApi.Update) { + if (!filter(update)) return + queued.incrementAndGet() + queue.trySend(update) + } + + fun backlog(): Long = queued.get() - processed.get() + + fun cancel() { + worker?.cancel() + queue.close() + } + } + + private companion object { + private const val TAG = "TdUpdatePipeline" + private const val PUMP_THREAD_NAME = "td-update-pump" + + /** + * Observation buffer. Large enough that a collector doing only in-memory work + * cannot realistically fall behind; anything slower belongs on a lane. + */ + private const val OBSERVER_BUFFER = 1024 + + /** + * Kept for late observers of cheap, replaceable state. It is not a correctness + * mechanism: durable consumers use [lane], which never drops. + */ + private const val OBSERVER_REPLAY = 3 + + private const val BACKLOG_WARN_AT = 2048L + private const val BACKLOG_SAMPLE_EVERY = 512L + } +} diff --git a/data/src/main/java/org/monogram/data/di/dataModule.kt b/data/src/main/java/org/monogram/data/di/dataModule.kt index 1364c7263..d3870c938 100644 --- a/data/src/main/java/org/monogram/data/di/dataModule.kt +++ b/data/src/main/java/org/monogram/data/di/dataModule.kt @@ -788,7 +788,7 @@ val dataModule = module { FileUpdateHandler( registry = get(), queue = get(), - fileUpdatesSource = get().file, + updates = get(), scope = get() ) } diff --git a/data/src/main/java/org/monogram/data/gateway/TelegramGateway.kt b/data/src/main/java/org/monogram/data/gateway/TelegramGateway.kt index 511bf1c1a..629393072 100644 --- a/data/src/main/java/org/monogram/data/gateway/TelegramGateway.kt +++ b/data/src/main/java/org/monogram/data/gateway/TelegramGateway.kt @@ -1,11 +1,32 @@ package org.monogram.data.gateway +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow import org.drinkless.tdlib.TdApi +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext interface TelegramGateway { suspend fun execute(function: TdApi.Function): T + + /** + * Observation stream. Conflates when a collector falls behind; use [lane] for + * anything that drives durable state. + */ val updates: SharedFlow + val isAuthenticated: StateFlow + + /** + * Lossless, strictly ordered, exception-isolated update subscription. + * See [UpdateDispatcher.lane]. + */ + fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext = EmptyCoroutineContext, + filter: (TdApi.Update) -> Boolean = { true }, + handler: suspend (TdApi.Update) -> Unit, + ) } diff --git a/data/src/main/java/org/monogram/data/gateway/TelegramGatewayImpl.kt b/data/src/main/java/org/monogram/data/gateway/TelegramGatewayImpl.kt index 6d2fe735c..5af3b0755 100644 --- a/data/src/main/java/org/monogram/data/gateway/TelegramGatewayImpl.kt +++ b/data/src/main/java/org/monogram/data/gateway/TelegramGatewayImpl.kt @@ -1,9 +1,11 @@ package org.monogram.data.gateway +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.StateFlow import org.drinkless.tdlib.TdApi import org.monogram.data.di.TdLibClient +import kotlin.coroutines.CoroutineContext internal class TelegramGatewayImpl( private val client: TdLibClient @@ -16,4 +18,17 @@ internal class TelegramGatewayImpl( override val isAuthenticated: StateFlow get() = client.isAuthenticated + + override fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) { + client.lane(name, scope, context, filter, handler) + } + + /** Diagnostics: ingest/lane backlogs, processed counts and handler failures. */ + fun updateMetrics(): String = client.updateMetrics() } diff --git a/data/src/main/java/org/monogram/data/gateway/UpdateDispatcher.kt b/data/src/main/java/org/monogram/data/gateway/UpdateDispatcher.kt index 651246045..e9d88c6dd 100644 --- a/data/src/main/java/org/monogram/data/gateway/UpdateDispatcher.kt +++ b/data/src/main/java/org/monogram/data/gateway/UpdateDispatcher.kt @@ -1,12 +1,39 @@ package org.monogram.data.gateway +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.emptyFlow import org.drinkless.tdlib.TdApi +import kotlin.coroutines.CoroutineContext +import kotlin.coroutines.EmptyCoroutineContext interface UpdateDispatcher { + /** + * Observation stream. Conflates when a collector falls behind, so it must only be + * used by consumers that render state they can re-read. Anything that writes to Room, + * mutates [org.monogram.data.chats.ChatCache], or issues a TDLib request must use + * [lane] instead. + */ val all: Flow + /** + * Lossless, strictly ordered, exception-isolated subscription. + * + * The lane owns a private unbounded queue and a private worker, so a slow handler + * delays only itself, and a handler that throws does not end the subscription. + * Register during startup: updates delivered before the lane exists are not replayed. + * + * @param filter evaluated on the update pump thread; keep it to cheap type checks. + * @param context extra worker context, e.g. `Dispatchers.IO` for lanes that hit Room. + */ + fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext = EmptyCoroutineContext, + filter: (TdApi.Update) -> Boolean = { true }, + handler: suspend (TdApi.Update) -> Unit, + ) + // Auth val authorizationState: Flow diff --git a/data/src/main/java/org/monogram/data/gateway/UpdateDispatcherImpl.kt b/data/src/main/java/org/monogram/data/gateway/UpdateDispatcherImpl.kt index 8e2a938ee..b82343456 100644 --- a/data/src/main/java/org/monogram/data/gateway/UpdateDispatcherImpl.kt +++ b/data/src/main/java/org/monogram/data/gateway/UpdateDispatcherImpl.kt @@ -1,18 +1,30 @@ package org.monogram.data.gateway +import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.SharedFlow import kotlinx.coroutines.flow.filter import kotlinx.coroutines.flow.filterIsInstance import org.drinkless.tdlib.TdApi +import kotlin.coroutines.CoroutineContext class UpdateDispatcherImpl( - gateway: TelegramGateway + private val gateway: TelegramGateway ) : UpdateDispatcher { private val updates = gateway.updates override val all: SharedFlow = updates + override fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) { + gateway.lane(name, scope, context, filter, handler) + } + private inline fun flow(): Flow = updates.filterIsInstance() @@ -56,36 +68,44 @@ class UpdateDispatcherImpl( override val installedStickerSets = flow() override val newChat = flow() override val attachmentMenuBots = flow() - override val chatsListUpdates = updates.filter { - it is TdApi.UpdateNewChat || - it is TdApi.UpdateChatTitle || - it is TdApi.UpdateChatPhoto || - it is TdApi.UpdateChatLastMessage || - it is TdApi.UpdateChatPosition || - it is TdApi.UpdateChatReadInbox || - it is TdApi.UpdateChatReadOutbox || - it is TdApi.UpdateChatUnreadMentionCount || - it is TdApi.UpdateChatUnreadReactionCount || - it is TdApi.UpdateChatDraftMessage || - it is TdApi.UpdateChatNotificationSettings || - it is TdApi.UpdateChatPermissions || - it is TdApi.UpdateChatViewAsTopics || - it is TdApi.UpdateChatIsTranslatable || - it is TdApi.UpdateChatOnlineMemberCount || - it is TdApi.UpdateChatFolders || - it is TdApi.UpdateUserStatus || - it is TdApi.UpdateUser || - it is TdApi.UpdateSupergroup || - it is TdApi.UpdateBasicGroup || - it is TdApi.UpdateSupergroupFullInfo || - it is TdApi.UpdateBasicGroupFullInfo || - it is TdApi.UpdateSecretChat || - it is TdApi.UpdateChatAction || - it is TdApi.UpdateFile || - it is TdApi.UpdateDeleteMessages || - it is TdApi.UpdateMessageMentionRead || - it is TdApi.UpdateMessageReactions || - it is TdApi.UpdateAuthorizationState || - it is TdApi.UpdateConnectionState - } + override val chatsListUpdates = updates.filter(CHATS_LIST_LANE_FILTER) +} + +/** + * The update set that drives the canonical chat cache. + * + * Shared by [UpdateDispatcher.chatsListUpdates] and by the lossless "chat-list" lane in + * `ChatsListRepositoryImpl`, so the two can never drift apart. + */ +val CHATS_LIST_LANE_FILTER: (TdApi.Update) -> Boolean = { + it is TdApi.UpdateNewChat || + it is TdApi.UpdateChatTitle || + it is TdApi.UpdateChatPhoto || + it is TdApi.UpdateChatLastMessage || + it is TdApi.UpdateChatPosition || + it is TdApi.UpdateChatReadInbox || + it is TdApi.UpdateChatReadOutbox || + it is TdApi.UpdateChatUnreadMentionCount || + it is TdApi.UpdateChatUnreadReactionCount || + it is TdApi.UpdateChatDraftMessage || + it is TdApi.UpdateChatNotificationSettings || + it is TdApi.UpdateChatPermissions || + it is TdApi.UpdateChatViewAsTopics || + it is TdApi.UpdateChatIsTranslatable || + it is TdApi.UpdateChatOnlineMemberCount || + it is TdApi.UpdateChatFolders || + it is TdApi.UpdateUserStatus || + it is TdApi.UpdateUser || + it is TdApi.UpdateSupergroup || + it is TdApi.UpdateBasicGroup || + it is TdApi.UpdateSupergroupFullInfo || + it is TdApi.UpdateBasicGroupFullInfo || + it is TdApi.UpdateSecretChat || + it is TdApi.UpdateChatAction || + it is TdApi.UpdateFile || + it is TdApi.UpdateDeleteMessages || + it is TdApi.UpdateMessageMentionRead || + it is TdApi.UpdateMessageReactions || + it is TdApi.UpdateAuthorizationState || + it is TdApi.UpdateConnectionState } diff --git a/data/src/main/java/org/monogram/data/infra/FileUpdateHandler.kt b/data/src/main/java/org/monogram/data/infra/FileUpdateHandler.kt index 887b7a1d6..466ddbb4c 100644 --- a/data/src/main/java/org/monogram/data/infra/FileUpdateHandler.kt +++ b/data/src/main/java/org/monogram/data/infra/FileUpdateHandler.kt @@ -2,11 +2,11 @@ package org.monogram.data.infra import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.channels.BufferOverflow -import kotlinx.coroutines.flow.Flow import kotlinx.coroutines.flow.MutableSharedFlow import kotlinx.coroutines.flow.asSharedFlow import kotlinx.coroutines.launch import org.drinkless.tdlib.TdApi +import org.monogram.data.gateway.UpdateDispatcher interface FileUpdateQueue { fun updateFileCache(file: TdApi.File) @@ -17,7 +17,7 @@ interface FileUpdateQueue { class FileUpdateHandler( private val registry: FileMessageRegistry, private val queue: FileUpdateQueue, - private val fileUpdatesSource: Flow, + private val updates: UpdateDispatcher, private val scope: CoroutineScope ) { val customEmojiPaths = SynchronizedLruMap(CUSTOM_EMOJI_CACHE_SIZE) @@ -39,8 +39,16 @@ class FileUpdateHandler( val fileUpdates = fileUpdateEvents.events init { - scope.launch { - fileUpdatesSource.collect { update -> handle(update.file) } + // The isDownloadingCompleted / isUploadingCompleted edge is what resolves download + // waiters and records the local path; losing it strands whatever was waiting. + // Progress ticks in between are conflatable, but the edge is not, so the whole + // stream goes through a lossless lane. + updates.lane( + name = "files", + scope = scope, + filter = { it is TdApi.UpdateFile }, + ) { update -> + handle((update as TdApi.UpdateFile).file) } } diff --git a/data/src/main/java/org/monogram/data/repository/ChatsListRepositoryImpl.kt b/data/src/main/java/org/monogram/data/repository/ChatsListRepositoryImpl.kt index 2ec968daf..e95b38a95 100644 --- a/data/src/main/java/org/monogram/data/repository/ChatsListRepositoryImpl.kt +++ b/data/src/main/java/org/monogram/data/repository/ChatsListRepositoryImpl.kt @@ -38,6 +38,7 @@ import org.monogram.data.datasource.remote.ChatsRemoteDataSource import org.monogram.data.db.dao.ChatFolderDao import org.monogram.data.db.dao.SearchHistoryDao import org.monogram.data.db.dao.UserFullInfoDao +import org.monogram.data.gateway.CHATS_LIST_LANE_FILTER import org.monogram.data.gateway.TdLibException import org.monogram.data.gateway.TelegramGateway import org.monogram.data.gateway.UpdateDispatcher @@ -305,10 +306,16 @@ class ChatsListRepositoryImpl( } } - scope.launch { - updates.chatsListUpdates.collect { update -> - updateHandler.handle(update) - } + // ChatUpdateHandler.handle is the canonical ChatCache mutation. Missing an + // updateNewChat leaves no row for the chat, and every later cache.updateChat for + // it becomes a silent no-op (ChatCache.updateChat), permanently blackholing the + // chat. This must be lossless and ordered. + updates.lane( + name = "chat-list", + scope = scope, + filter = CHATS_LIST_LANE_FILTER, + ) { update -> + updateHandler.handle(update) } scope.launch { diff --git a/data/src/main/java/org/monogram/data/repository/MessageRepositoryImpl.kt b/data/src/main/java/org/monogram/data/repository/MessageRepositoryImpl.kt index 777443333..1009ce347 100644 --- a/data/src/main/java/org/monogram/data/repository/MessageRepositoryImpl.kt +++ b/data/src/main/java/org/monogram/data/repository/MessageRepositoryImpl.kt @@ -6,10 +6,7 @@ import android.util.Log import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.asStateFlow -import kotlinx.coroutines.flow.catch -import kotlinx.coroutines.flow.launchIn import kotlinx.coroutines.flow.map -import kotlinx.coroutines.flow.onEach import kotlinx.coroutines.launch import kotlinx.coroutines.sync.Mutex import kotlinx.coroutines.sync.withLock @@ -154,18 +151,19 @@ internal class MessageRepositoryImpl( } } - updates.all - .map { update -> - messageRemoteDataSource.handleUpdate(update) - update - } - .onEach { update -> - processCachedUpdate(update) - } - .catch { error -> - Log.e("TdLibUpdates", "CRITICAL: Update loop died", error) - } - .launchIn(scope) + // Owns the message cache and the Room message mirror, so it must not miss updates: + // updateNewMessage, updateDeleteMessages and updateMessageSendSucceeded are deltas + // that TDLib never re-sends. A lane is lossless and strictly ordered, and isolates + // handler exceptions per update — the previous `.catch { }` ended the subscription + // for the rest of the process on the first failure. + updates.lane( + name = "messages", + scope = scope, + context = dispatcherProvider.io, + ) { update -> + messageRemoteDataSource.handleUpdate(update) + processCachedUpdate(update) + } scope.launch(dispatcherProvider.io) { val ninetyDaysAgo = System.currentTimeMillis() - (90L * 24 * 60 * 60 * 1000) diff --git a/data/src/main/java/org/monogram/data/repository/NotificationSettingsRepositoryImpl.kt b/data/src/main/java/org/monogram/data/repository/NotificationSettingsRepositoryImpl.kt index c3ff61666..5fd9416db 100644 --- a/data/src/main/java/org/monogram/data/repository/NotificationSettingsRepositoryImpl.kt +++ b/data/src/main/java/org/monogram/data/repository/NotificationSettingsRepositoryImpl.kt @@ -35,47 +35,55 @@ class NotificationSettingsRepositoryImpl( private val exceptionsCacheMutex = Mutex() init { - scope.launch { - updates.newChat.collect { update -> - cache.putChat(update.chat) - syncChatWithExceptionsCache(update.chat) - } - } + // All four branches write to the exceptions cache and the Room exception table, + // and the last three are no-ops unless updateNewChat was seen first, so they must + // be lossless and mutually ordered — hence one lane instead of four subscriptions. + updates.lane( + name = "notification-settings", + scope = scope, + filter = { + it is TdApi.UpdateNewChat || + it is TdApi.UpdateChatTitle || + it is TdApi.UpdateChatPhoto || + it is TdApi.UpdateChatNotificationSettings + }, + ) { update -> + when (update) { + is TdApi.UpdateNewChat -> { + cache.putChat(update.chat) + syncChatWithExceptionsCache(update.chat) + } - scope.launch { - updates.chatTitle.collect { update -> - cache.getChat(update.chatId)?.let { chat -> - synchronized(chat) { - chat.title = update.title + is TdApi.UpdateChatTitle -> { + cache.getChat(update.chatId)?.let { chat -> + synchronized(chat) { + chat.title = update.title + } + syncChatWithExceptionsCache(chat) } - syncChatWithExceptionsCache(chat) } - } - } - scope.launch { - updates.chatPhoto.collect { update -> - cache.getChat(update.chatId)?.let { chat -> - synchronized(chat) { - chat.photo = update.photo + is TdApi.UpdateChatPhoto -> { + cache.getChat(update.chatId)?.let { chat -> + synchronized(chat) { + chat.photo = update.photo + } + syncChatWithExceptionsCache(chat) } - syncChatWithExceptionsCache(chat) } - } - } - scope.launch { - updates.chatNotificationSettings.collect { update -> - cache.getChat(update.chatId)?.let { chat -> - synchronized(chat) { - chat.notificationSettings = update.notificationSettings - } - syncChatWithExceptionsCache(chat) - } ?: run { - if (update.notificationSettings.isException(compareSound = true)) { - invalidateExceptionsCache() - } else { - removeFromExceptionsCache(update.chatId) + is TdApi.UpdateChatNotificationSettings -> { + cache.getChat(update.chatId)?.let { chat -> + synchronized(chat) { + chat.notificationSettings = update.notificationSettings + } + syncChatWithExceptionsCache(chat) + } ?: run { + if (update.notificationSettings.isException(compareSound = true)) { + invalidateExceptionsCache() + } else { + removeFromExceptionsCache(update.chatId) + } } } } diff --git a/data/src/main/java/org/monogram/data/repository/user/UserUpdateSynchronizer.kt b/data/src/main/java/org/monogram/data/repository/user/UserUpdateSynchronizer.kt index e85c85417..d92c6c8ba 100644 --- a/data/src/main/java/org/monogram/data/repository/user/UserUpdateSynchronizer.kt +++ b/data/src/main/java/org/monogram/data/repository/user/UserUpdateSynchronizer.kt @@ -40,16 +40,27 @@ internal class UserUpdateSynchronizer( } } - scope.launch { - updates.user.collect { update -> - updateAvatarIndex(update.user) - onUserUpdated(update.user) - } - } + // updateUser is documented as arriving before the user id is handed to the + // application, and it is the only introduction of the user object, so it must be + // lossless. Status goes through the same lane rather than a second subscription: + // the batcher only conflates per user id, so a lost update here would still lose + // that user's presence entirely. + updates.lane( + name = "users", + scope = scope, + filter = { it is TdApi.UpdateUser || it is TdApi.UpdateUserStatus }, + ) { update -> + when (update) { + is TdApi.UpdateUser -> { + updateAvatarIndex(update.user) + onUserUpdated(update.user) + } - scope.launch { - updates.userStatus.collect { update -> - userStatusBatcher.offer(update.userId, update.status) + is TdApi.UpdateUserStatus -> { + // Non-suspending, keeps the latest status per user; the batch applies + // it to the store off the lane. + userStatusBatcher.offer(update.userId, update.status) + } } } diff --git a/data/src/test/java/org/monogram/data/datasource/remote/TdChatRemoteSourceTest.kt b/data/src/test/java/org/monogram/data/datasource/remote/TdChatRemoteSourceTest.kt index 6e92d727b..dabc15148 100644 --- a/data/src/test/java/org/monogram/data/datasource/remote/TdChatRemoteSourceTest.kt +++ b/data/src/test/java/org/monogram/data/datasource/remote/TdChatRemoteSourceTest.kt @@ -31,6 +31,14 @@ class TdChatRemoteSourceTest { } private class CapturingTelegramGateway : TelegramGateway { + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(updates, scope, context, filter, handler) + override val updates = MutableSharedFlow() override val isAuthenticated = MutableStateFlow(false) var lastSetChatDescription: TdApi.SetChatDescription? = null diff --git a/data/src/test/java/org/monogram/data/datasource/remote/TdGifRemoteSourceTest.kt b/data/src/test/java/org/monogram/data/datasource/remote/TdGifRemoteSourceTest.kt index 6cc974712..02868f990 100644 --- a/data/src/test/java/org/monogram/data/datasource/remote/TdGifRemoteSourceTest.kt +++ b/data/src/test/java/org/monogram/data/datasource/remote/TdGifRemoteSourceTest.kt @@ -40,6 +40,14 @@ class TdGifRemoteSourceTest { } private class CapturingTelegramGateway : TelegramGateway { + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(updates, scope, context, filter, handler) + override val updates = MutableSharedFlow() override val isAuthenticated = MutableStateFlow(false) diff --git a/data/src/test/java/org/monogram/data/datasource/remote/TdStickerRemoteSourceTest.kt b/data/src/test/java/org/monogram/data/datasource/remote/TdStickerRemoteSourceTest.kt index 1f97c437e..35276a2c6 100644 --- a/data/src/test/java/org/monogram/data/datasource/remote/TdStickerRemoteSourceTest.kt +++ b/data/src/test/java/org/monogram/data/datasource/remote/TdStickerRemoteSourceTest.kt @@ -51,6 +51,14 @@ class TdStickerRemoteSourceTest { } private class CapturingTelegramGateway : TelegramGateway { + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(updates, scope, context, filter, handler) + override val updates = MutableSharedFlow() override val isAuthenticated = MutableStateFlow(false) diff --git a/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt b/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt new file mode 100644 index 000000000..ee7560d9a --- /dev/null +++ b/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt @@ -0,0 +1,229 @@ +package org.monogram.data.di + +import kotlinx.coroutines.CompletableDeferred +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.launch +import kotlinx.coroutines.runBlocking +import kotlinx.coroutines.withTimeout +import org.drinkless.tdlib.TdApi +import org.junit.After +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertTrue +import org.junit.Test +import java.util.Collections +import java.util.concurrent.atomic.AtomicInteger + +/** + * Contract of [TdUpdatePipeline]: + * + * - a lane never loses an update and never reorders one; + * - a slow or throwing lane affects only itself; + * - the observation flow is allowed to conflate, and does; + * - ingestion from the TDLib callback thread never waits for a consumer. + */ +class TdUpdatePipelineTest { + + private val scopes = mutableListOf() + private val pipelines = mutableListOf() + + private fun pipeline(): TdUpdatePipeline = TdUpdatePipeline().also { pipelines.add(it) } + + private fun scope(): CoroutineScope = + CoroutineScope(SupervisorJob() + Dispatchers.Default).also { scopes.add(it) } + + @After + fun tearDown() { + scopes.forEach { it.cancel() } + pipelines.forEach { it.shutdown() } + } + + /** Updates carry their sequence number in chatId so ordering is checkable. */ + private fun update(seq: Int): TdApi.Update = TdApi.UpdateChatTitle(seq.toLong(), "t$seq") + + private fun seqOf(update: TdApi.Update) = (update as TdApi.UpdateChatTitle).chatId.toInt() + + /** Pushes updates the way TDLib does: from one plain thread that is not a coroutine. */ + private fun submitFromCallbackThread(pipeline: TdUpdatePipeline, count: Int, from: Int = 0) { + val thread = Thread({ repeat(count) { pipeline.submit(update(from + it)) } }, "TDLib thread") + thread.start() + thread.join() + } + + private suspend fun awaitCount(expected: Int, timeoutMs: Long = 30_000, actual: () -> Int) { + withTimeout(timeoutMs) { + while (actual() < expected) delay(5) + } + } + + @Test + fun `lane receives every update exactly once and in order`() = runBlocking { + val pipeline = pipeline() + val received = Collections.synchronizedList(mutableListOf()) + pipeline.lane("state", scope()) { received.add(seqOf(it)) } + + val total = 5_000 + submitFromCallbackThread(pipeline, total) + awaitCount(total) { received.size } + + assertEquals(total, received.size) + assertEquals((0 until total).toList(), received.toList()) + } + + @Test + fun `lane filter is applied and does not create gaps`() = runBlocking { + val pipeline = pipeline() + val received = Collections.synchronizedList(mutableListOf()) + pipeline.lane("even", scope(), filter = { seqOf(it) % 2 == 0 }) { received.add(seqOf(it)) } + + submitFromCallbackThread(pipeline, 1_000) + awaitCount(500) { received.size } + + assertEquals(500, received.size) + assertEquals((0 until 1_000 step 2).toList(), received.toList()) + } + + @Test + fun `a handler that throws does not end the lane`() = runBlocking { + val pipeline = pipeline() + val handled = AtomicInteger() + val failed = AtomicInteger() + pipeline.lane("flaky", scope()) { + if (seqOf(it) % 10 == 0) { + failed.incrementAndGet() + error("boom on ${seqOf(it)}") + } + handled.incrementAndGet() + } + + val total = 1_000 + submitFromCallbackThread(pipeline, total) + awaitCount(total) { handled.get() + failed.get() } + + assertEquals(100, failed.get()) + assertEquals(900, handled.get()) + assertTrue("lane must still be registered", pipeline.metrics().contains("flaky")) + } + + @Test + fun `a slow lane does not make a fast lane lose or lag`() = runBlocking { + val pipeline = pipeline() + val fast = Collections.synchronizedList(mutableListOf()) + val slow = AtomicInteger() + + pipeline.lane("fast", scope()) { fast.add(seqOf(it)) } + pipeline.lane("slow", scope()) { delay(2); slow.incrementAndGet() } + + val total = 500 + submitFromCallbackThread(pipeline, total) + + // The fast lane completes long before the slow one, and loses nothing. + awaitCount(total) { fast.size } + assertEquals((0 until total).toList(), fast.toList()) + + // The slow lane is merely delayed, never truncated. + awaitCount(total) { slow.get() } + assertEquals(total, slow.get()) + } + + @Test + fun `ingestion does not wait for consumers`() = runBlocking { + val pipeline = pipeline() + val gate = CompletableDeferred() + val processed = AtomicInteger() + pipeline.lane("blocked", scope()) { + gate.await() + processed.incrementAndGet() + } + + val total = 10_000 + val elapsedMs = kotlin.system.measureTimeMillis { submitFromCallbackThread(pipeline, total) } + + // The lane handler has not returned even once, yet every submit has completed. + assertEquals(0, processed.get()) + assertTrue( + "submitting $total updates took ${elapsedMs}ms; ingestion must not block on consumers", + elapsedMs < 2_000 + ) + + gate.complete(Unit) + awaitCount(total) { processed.get() } + assertEquals(total, processed.get()) + } + + @Test + fun `lane deregisters when its scope is cancelled`() = runBlocking { + val pipeline = pipeline() + val ownScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) + pipeline.lane("transient", ownScope) { } + + submitFromCallbackThread(pipeline, 10) + withTimeout(10_000) { while (!pipeline.metrics().contains("transient")) delay(5) } + + ownScope.cancel() + withTimeout(10_000) { while (pipeline.metrics().contains("transient")) delay(5) } + + // Further updates must not accumulate for the dead lane. + submitFromCallbackThread(pipeline, 100, from = 10) + delay(200) + assertFalse(pipeline.metrics().contains("transient")) + } + + /** + * The reason lanes exist. A consumer slower than the arrival rate loses updates on the + * observation flow and loses nothing on a lane, under the identical burst. + */ + @Test + fun `observation flow conflates under load while a lane does not`() = runBlocking { + val pipeline = pipeline() + val laneSeen = AtomicInteger() + val observerSeen = AtomicInteger() + val observerScope = scope() + + pipeline.lane("durable", scope()) { delay(1); laneSeen.incrementAndGet() } + val observer = observerScope.launch { + pipeline.updates.collect { delay(1); observerSeen.incrementAndGet() } + } + // Let the observer subscribe before the burst starts. + withTimeout(10_000) { while (!pipeline.metrics().contains("observers=1")) delay(5) } + + val total = 4_000 + submitFromCallbackThread(pipeline, total) + + awaitCount(total, timeoutMs = 60_000) { laneSeen.get() } + assertEquals("a lane must never drop", total, laneSeen.get()) + + delay(500) + assertTrue( + "the observation flow is expected to conflate under this load, saw ${observerSeen.get()} of $total", + observerSeen.get() < total + ) + observer.cancel() + } + + @Test + fun `metrics expose the backlog of a stalled lane`() = runBlocking { + val pipeline = pipeline() + val gate = CompletableDeferred() + pipeline.lane("stalled", scope()) { gate.await() } + + submitFromCallbackThread(pipeline, 250) + val backlogPattern = Regex("stalled: backlog=(\\d+)") + withTimeout(10_000) { + while (laneBacklog(backlogPattern, pipeline) < 249) delay(5) + } + + val metrics = pipeline.metrics() + gate.complete(Unit) + + assertTrue(metrics, metrics.contains("submitted=250")) + assertTrue(metrics, laneBacklog(backlogPattern, pipeline) >= 0) + } + + private fun laneBacklog(pattern: Regex, pipeline: TdUpdatePipeline): Int = + pattern.find(pipeline.metrics())?.groupValues?.get(1)?.toInt() ?: -1 +} diff --git a/data/src/test/java/org/monogram/data/gateway/UpdateDispatcherImplTest.kt b/data/src/test/java/org/monogram/data/gateway/UpdateDispatcherImplTest.kt index f43acde56..7a3fe0c74 100644 --- a/data/src/test/java/org/monogram/data/gateway/UpdateDispatcherImplTest.kt +++ b/data/src/test/java/org/monogram/data/gateway/UpdateDispatcherImplTest.kt @@ -14,6 +14,8 @@ import kotlinx.coroutines.test.runTest import org.drinkless.tdlib.TdApi import org.junit.Assert.assertEquals import org.junit.Test +import org.monogram.data.testing.fakeUpdateLane +import kotlin.coroutines.CoroutineContext @OptIn(ExperimentalCoroutinesApi::class) class UpdateDispatcherImplTest { @@ -103,6 +105,14 @@ class UpdateDispatcherImplTest { override suspend fun execute(function: TdApi.Function): T { error("Not used") } + + override fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = fakeUpdateLane(updates, scope, context, filter, handler) } private companion object { diff --git a/data/src/test/java/org/monogram/data/infra/ConnectionManagerTest.kt b/data/src/test/java/org/monogram/data/infra/ConnectionManagerTest.kt index d1b68e5df..8be7a5ed7 100644 --- a/data/src/test/java/org/monogram/data/infra/ConnectionManagerTest.kt +++ b/data/src/test/java/org/monogram/data/infra/ConnectionManagerTest.kt @@ -538,6 +538,14 @@ class ConnectionManagerTest { override val connectionState: Flow ) : UpdateDispatcher { override val all: Flow = MutableSharedFlow() + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(all, scope, context, filter, handler) + override val newMessage: Flow = MutableSharedFlow() override val activeNotifications: Flow = MutableSharedFlow() diff --git a/data/src/test/java/org/monogram/data/infra/FileDownloadQueueTest.kt b/data/src/test/java/org/monogram/data/infra/FileDownloadQueueTest.kt index 6a2871353..128551920 100644 --- a/data/src/test/java/org/monogram/data/infra/FileDownloadQueueTest.kt +++ b/data/src/test/java/org/monogram/data/infra/FileDownloadQueueTest.kt @@ -83,6 +83,14 @@ class FileDownloadQueueTest { } private class FakeTelegramGateway : TelegramGateway { + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(updates, scope, context, filter, handler) + override val updates = MutableSharedFlow() override val isAuthenticated = MutableStateFlow(false) diff --git a/data/src/test/java/org/monogram/data/infra/FileUpdateHandlerTest.kt b/data/src/test/java/org/monogram/data/infra/FileUpdateHandlerTest.kt index e3031ce90..8e2536445 100644 --- a/data/src/test/java/org/monogram/data/infra/FileUpdateHandlerTest.kt +++ b/data/src/test/java/org/monogram/data/infra/FileUpdateHandlerTest.kt @@ -5,6 +5,7 @@ import kotlinx.coroutines.ExperimentalCoroutinesApi import kotlinx.coroutines.SupervisorJob import kotlinx.coroutines.cancel import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.take import kotlinx.coroutines.launch import kotlinx.coroutines.test.advanceUntilIdle @@ -13,6 +14,10 @@ import kotlinx.coroutines.test.runTest import org.drinkless.tdlib.TdApi import org.junit.Assert.assertEquals import org.junit.Test +import org.monogram.data.gateway.TelegramGateway +import org.monogram.data.gateway.UpdateDispatcherImpl +import org.monogram.data.testing.fakeUpdateLane +import kotlin.coroutines.CoroutineContext @OptIn(ExperimentalCoroutinesApi::class) class FileUpdateHandlerTest { @@ -20,13 +25,13 @@ class FileUpdateHandlerTest { @Test fun `burst file completions preserve every terminal path and order`() = runTest { val scope = CoroutineScope(coroutineContext + SupervisorJob()) - val updates = MutableSharedFlow() + val updates = MutableSharedFlow() val queue = RecordingQueue() val registry = FileMessageRegistry() val handler = FileUpdateHandler( registry = registry, queue = queue, - fileUpdatesSource = updates, + updates = UpdateDispatcherImpl(FakeTelegramGateway(updates)), scope = scope ) val fileCompleted = mutableListOf>() @@ -68,6 +73,24 @@ class FileUpdateHandlerTest { remote = TdApi.RemoteFile() } + private class FakeTelegramGateway( + override val updates: MutableSharedFlow + ) : TelegramGateway { + override val isAuthenticated = MutableStateFlow(false) + + override suspend fun execute(function: TdApi.Function): T { + error("Not used") + } + + override fun lane( + name: String, + scope: CoroutineScope, + context: CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = fakeUpdateLane(updates, scope, context, filter, handler) + } + private class RecordingQueue : FileUpdateQueue { val completedDownloads = mutableListOf() diff --git a/data/src/test/java/org/monogram/data/infra/SponsorSyncManagerTest.kt b/data/src/test/java/org/monogram/data/infra/SponsorSyncManagerTest.kt index edfc53e27..54918a7f8 100644 --- a/data/src/test/java/org/monogram/data/infra/SponsorSyncManagerTest.kt +++ b/data/src/test/java/org/monogram/data/infra/SponsorSyncManagerTest.kt @@ -168,6 +168,14 @@ class SponsorSyncManagerTest { } private class FakeTelegramGateway : TelegramGateway { + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(_updates, scope, context, filter, handler) + private val _updates = MutableSharedFlow() private val _isAuthenticated = MutableStateFlow(true) var historyProvider: suspend () -> TdApi.Messages = { TdApi.Messages(0, emptyArray()) } diff --git a/data/src/test/java/org/monogram/data/repository/TelegramLinkRepositoryImplTest.kt b/data/src/test/java/org/monogram/data/repository/TelegramLinkRepositoryImplTest.kt index b767b72e8..d90494b74 100644 --- a/data/src/test/java/org/monogram/data/repository/TelegramLinkRepositoryImplTest.kt +++ b/data/src/test/java/org/monogram/data/repository/TelegramLinkRepositoryImplTest.kt @@ -126,6 +126,14 @@ private class FakeTelegramGateway( private val authState = MutableStateFlow(authenticated) override val isAuthenticated: StateFlow = authState var executeCalls: Int = 0 + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(updates, scope, context, filter, handler) + fun setAuthenticated(value: Boolean) { authState.value = value @@ -140,6 +148,14 @@ private class FakeTelegramGateway( private class FakeUpdateDispatcher : UpdateDispatcher { override val all: Flow = MutableSharedFlow() + override fun lane( + name: String, + scope: kotlinx.coroutines.CoroutineScope, + context: kotlin.coroutines.CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, + ) = org.monogram.data.testing.fakeUpdateLane(all, scope, context, filter, handler) + override val authorizationState: Flow = MutableSharedFlow() override val newMessage: Flow = MutableSharedFlow() override val activeNotifications: Flow = MutableSharedFlow() diff --git a/data/src/test/java/org/monogram/data/testing/FakeUpdateLane.kt b/data/src/test/java/org/monogram/data/testing/FakeUpdateLane.kt new file mode 100644 index 000000000..5a5dc63d5 --- /dev/null +++ b/data/src/test/java/org/monogram/data/testing/FakeUpdateLane.kt @@ -0,0 +1,28 @@ +package org.monogram.data.testing + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.filter +import kotlinx.coroutines.launch +import org.drinkless.tdlib.TdApi +import kotlin.coroutines.CoroutineContext + +/** + * Lane implementation for test doubles of `TelegramGateway` / `UpdateDispatcher`. + * + * Backs the lane with a plain collector on the fake's update flow. That is enough for + * tests, which control emission rate and never overflow anything; the production + * implementation in `TdUpdatePipeline` is what provides the losslessness and ordering + * guarantees, and it is covered by `TdUpdatePipelineTest`. + */ +internal fun fakeUpdateLane( + source: Flow, + scope: CoroutineScope, + context: CoroutineContext, + filter: (TdApi.Update) -> Boolean, + handler: suspend (TdApi.Update) -> Unit, +) { + scope.launch(context) { + source.filter(filter).collect { handler(it) } + } +} From 6819613982a86fbe043c84099b6505149bf5dbf5 Mon Sep 17 00:00:00 2001 From: tdrkDev Date: Wed, 5 Aug 2026 17:13:36 +0700 Subject: [PATCH 2/2] data: make TdUpdatePipelineTest deterministic The tests timed out at awaitCount(4000, 60000ms) in the conflation test. The lane handler under test did delay(1) per update, so the test's runtime scaled with the platform's timer resolution: 1.4ms/update locally (5.6s), but around 15ms on a coarser or more loaded scheduler, which exceeds the budget. Two other tests had the same defect, plus a wall-clock assertion (elapsed < 2s) and one that depended on whether the observer happened to fall behind. Drive progress with explicit CompletableDeferred gates instead of sleeping. delay now appears only as a 2ms polling interval, never once per update: - a slow lane becomes a lane blocked on a gate, which is the same property without the wall-clock cost; - conflation is established by SharedFlow buffer semantics rather than by winning a race - the observer is frozen until metrics() confirms the pump has emitted the whole burst, and the burst is ~3x the 1027-slot buffer; - the timing assertion is dropped, keeping the structural one that no handler had returned while every submit completed. The first rewrite still failed, for a reason worth recording: submit() only fills the ingest queue, so releasing the observer straight after submitting let it keep up with the pump and see all 3002. Waiting on dispatched= fixes it and makes the freeze provably span the entire emission sequence. Every test now carries @Test(timeout = 60_000) so a hang fails with a stack trace instead of hanging the runner, and awaitCount reports the count it actually reached rather than throwing a bare TimeoutCancellationException. Also fixes Lane.offer counting an update as queued when trySend fails on a closed lane, which left backlog() permanently non-zero in metrics(). 7.2s -> 0.1s for the class, 8/8 green over 8 consecutive runs. Full :data suite: 129 tests, 0 failures. Co-Authored-By: Claude Fable 5 --- .../org/monogram/data/di/TdUpdatePipeline.kt | 5 +- .../monogram/data/di/TdUpdatePipelineTest.kt | 197 ++++++++++++------ 2 files changed, 136 insertions(+), 66 deletions(-) diff --git a/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt b/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt index 2ff826221..8032f1977 100644 --- a/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt +++ b/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt @@ -197,8 +197,9 @@ internal class TdUpdatePipeline { fun offer(update: TdApi.Update) { if (!filter(update)) return - queued.incrementAndGet() - queue.trySend(update) + // Count only what was actually accepted: trySend fails once the lane has been + // closed, and counting those would leave backlog() permanently non-zero. + if (queue.trySend(update).isSuccess) queued.incrementAndGet() } fun backlog(): Long = queued.get() - processed.get() diff --git a/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt b/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt index ee7560d9a..eb7de34f9 100644 --- a/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt +++ b/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt @@ -8,7 +8,6 @@ import kotlinx.coroutines.cancel import kotlinx.coroutines.delay import kotlinx.coroutines.launch import kotlinx.coroutines.runBlocking -import kotlinx.coroutines.withTimeout import org.drinkless.tdlib.TdApi import org.junit.After import org.junit.Assert.assertEquals @@ -22,9 +21,13 @@ import java.util.concurrent.atomic.AtomicInteger * Contract of [TdUpdatePipeline]: * * - a lane never loses an update and never reorders one; - * - a slow or throwing lane affects only itself; + * - a slow, blocked or throwing lane affects only itself; * - the observation flow is allowed to conflate, and does; * - ingestion from the TDLib callback thread never waits for a consumer. + * + * Progress is driven by explicit gates rather than by sleeping, so runtime does not + * depend on the platform's timer resolution. `delay` appears only as a polling interval + * in [awaitCount], never once per update. */ class TdUpdatePipelineTest { @@ -54,13 +57,40 @@ class TdUpdatePipelineTest { thread.join() } - private suspend fun awaitCount(expected: Int, timeoutMs: Long = 30_000, actual: () -> Int) { - withTimeout(timeoutMs) { - while (actual() < expected) delay(5) + /** + * Polls until [actual] reaches [expected]. Uses an explicit deadline rather than + * `withTimeout` so a failure reports what was actually reached. + */ + private suspend fun awaitCount( + expected: Int, + what: String, + timeoutMs: Long = 30_000, + actual: () -> Int, + ) { + val deadline = System.nanoTime() + timeoutMs * 1_000_000 + while (actual() < expected) { + if (System.nanoTime() > deadline) { + throw AssertionError("timed out after ${timeoutMs}ms waiting for $expected $what, reached ${actual()}") + } + delay(2) + } + } + + /** Number of updates the pump has fanned out, as opposed to merely accepted. */ + private fun dispatched(pipeline: TdUpdatePipeline): Int = + Regex("dispatched=(\\d+)").find(pipeline.metrics())?.groupValues?.get(1)?.toInt() ?: 0 + + private suspend fun awaitObserverCount(pipeline: TdUpdatePipeline, expected: Int) { + val deadline = System.nanoTime() + 30_000L * 1_000_000 + while (!pipeline.metrics().contains("observers=$expected")) { + if (System.nanoTime() > deadline) { + throw AssertionError("observer never subscribed: ${pipeline.metrics()}") + } + delay(2) } } - @Test + @Test(timeout = 60_000) fun `lane receives every update exactly once and in order`() = runBlocking { val pipeline = pipeline() val received = Collections.synchronizedList(mutableListOf()) @@ -68,26 +98,26 @@ class TdUpdatePipelineTest { val total = 5_000 submitFromCallbackThread(pipeline, total) - awaitCount(total) { received.size } + awaitCount(total, "updates on the lane") { received.size } assertEquals(total, received.size) assertEquals((0 until total).toList(), received.toList()) } - @Test + @Test(timeout = 60_000) fun `lane filter is applied and does not create gaps`() = runBlocking { val pipeline = pipeline() val received = Collections.synchronizedList(mutableListOf()) pipeline.lane("even", scope(), filter = { seqOf(it) % 2 == 0 }) { received.add(seqOf(it)) } submitFromCallbackThread(pipeline, 1_000) - awaitCount(500) { received.size } + awaitCount(500, "even updates") { received.size } assertEquals(500, received.size) assertEquals((0 until 1_000 step 2).toList(), received.toList()) } - @Test + @Test(timeout = 60_000) fun `a handler that throws does not end the lane`() = runBlocking { val pipeline = pipeline() val handled = AtomicInteger() @@ -102,35 +132,44 @@ class TdUpdatePipelineTest { val total = 1_000 submitFromCallbackThread(pipeline, total) - awaitCount(total) { handled.get() + failed.get() } + awaitCount(total, "handled or failed updates") { handled.get() + failed.get() } assertEquals(100, failed.get()) assertEquals(900, handled.get()) assertTrue("lane must still be registered", pipeline.metrics().contains("flaky")) } - @Test - fun `a slow lane does not make a fast lane lose or lag`() = runBlocking { + /** + * A lane that makes no progress at all is the extreme case of a slow lane, and it is + * reached by a gate rather than by sleeping, so the test cannot become slow. + */ + @Test(timeout = 60_000) + fun `a blocked lane does not stop another lane and catches up losslessly`() = runBlocking { val pipeline = pipeline() - val fast = Collections.synchronizedList(mutableListOf()) - val slow = AtomicInteger() + val gate = CompletableDeferred() + val blocked = Collections.synchronizedList(mutableListOf()) + val healthy = Collections.synchronizedList(mutableListOf()) - pipeline.lane("fast", scope()) { fast.add(seqOf(it)) } - pipeline.lane("slow", scope()) { delay(2); slow.incrementAndGet() } + pipeline.lane("blocked", scope()) { gate.await(); blocked.add(seqOf(it)) } + pipeline.lane("healthy", scope()) { healthy.add(seqOf(it)) } - val total = 500 + val total = 2_000 submitFromCallbackThread(pipeline, total) - // The fast lane completes long before the slow one, and loses nothing. - awaitCount(total) { fast.size } - assertEquals((0 until total).toList(), fast.toList()) + awaitCount(total, "updates on the healthy lane") { healthy.size } + assertEquals((0 until total).toList(), healthy.toList()) + assertEquals("the blocked lane must not have progressed", 0, blocked.size) - // The slow lane is merely delayed, never truncated. - awaitCount(total) { slow.get() } - assertEquals(total, slow.get()) + gate.complete(Unit) + awaitCount(total, "updates on the unblocked lane") { blocked.size } + assertEquals( + "a lane must catch up losslessly and in order", + (0 until total).toList(), + blocked.toList() + ) } - @Test + @Test(timeout = 60_000) fun `ingestion does not wait for consumers`() = runBlocking { val pipeline = pipeline() val gate = CompletableDeferred() @@ -140,72 +179,102 @@ class TdUpdatePipelineTest { processed.incrementAndGet() } + // Every submit returns even though the lane handler has never returned. val total = 10_000 - val elapsedMs = kotlin.system.measureTimeMillis { submitFromCallbackThread(pipeline, total) } - - // The lane handler has not returned even once, yet every submit has completed. - assertEquals(0, processed.get()) - assertTrue( - "submitting $total updates took ${elapsedMs}ms; ingestion must not block on consumers", - elapsedMs < 2_000 - ) + submitFromCallbackThread(pipeline, total) + assertEquals("ingestion must not block on a consumer", 0, processed.get()) gate.complete(Unit) - awaitCount(total) { processed.get() } + awaitCount(total, "updates drained after unblocking") { processed.get() } assertEquals(total, processed.get()) } - @Test + @Test(timeout = 60_000) fun `lane deregisters when its scope is cancelled`() = runBlocking { val pipeline = pipeline() val ownScope = CoroutineScope(SupervisorJob() + Dispatchers.Default) - pipeline.lane("transient", ownScope) { } + val seen = AtomicInteger() + pipeline.lane("transient", ownScope) { seen.incrementAndGet() } submitFromCallbackThread(pipeline, 10) - withTimeout(10_000) { while (!pipeline.metrics().contains("transient")) delay(5) } + awaitCount(10, "updates before cancellation") { seen.get() } ownScope.cancel() - withTimeout(10_000) { while (pipeline.metrics().contains("transient")) delay(5) } + val deadline = System.nanoTime() + 30_000L * 1_000_000 + while (pipeline.metrics().contains("transient")) { + if (System.nanoTime() > deadline) { + throw AssertionError("lane was not deregistered: ${pipeline.metrics()}") + } + delay(2) + } - // Further updates must not accumulate for the dead lane. + // Further updates must neither reach nor accumulate for the dead lane. submitFromCallbackThread(pipeline, 100, from = 10) - delay(200) + awaitCount(110, "the pump to dispatch the remaining updates") { dispatched(pipeline) } assertFalse(pipeline.metrics().contains("transient")) + assertEquals("a cancelled lane must not keep consuming", 10, seen.get()) } /** - * The reason lanes exist. A consumer slower than the arrival rate loses updates on the - * observation flow and loses nothing on a lane, under the identical burst. + * The reason lanes exist. Under a burst larger than the observation buffer, the flow + * conflates and the lane does not. + * + * Deterministic: the observer is frozen on its first element for the whole burst, and + * the burst is several times the flow's buffer, so conflation is guaranteed by + * SharedFlow's semantics rather than by winning a race. */ - @Test + @Test(timeout = 60_000) fun `observation flow conflates under load while a lane does not`() = runBlocking { val pipeline = pipeline() val laneSeen = AtomicInteger() val observerSeen = AtomicInteger() - val observerScope = scope() - - pipeline.lane("durable", scope()) { delay(1); laneSeen.incrementAndGet() } - val observer = observerScope.launch { - pipeline.updates.collect { delay(1); observerSeen.incrementAndGet() } + val observerFrozen = CompletableDeferred() + val release = CompletableDeferred() + val sentinelSeen = CompletableDeferred() + + pipeline.lane("durable", scope()) { laneSeen.incrementAndGet() } + + val observer = scope().launch { + pipeline.updates.collect { u -> + if (!observerFrozen.isCompleted) { + observerFrozen.complete(Unit) + release.await() + } + observerSeen.incrementAndGet() + if (seqOf(u) == SENTINEL) sentinelSeen.complete(Unit) + } } - // Let the observer subscribe before the burst starts. - withTimeout(10_000) { while (!pipeline.metrics().contains("observers=1")) delay(5) } + awaitObserverCount(pipeline, 1) - val total = 4_000 - submitFromCallbackThread(pipeline, total) + // Freeze the observer, then push far more than its buffer can hold. + submitFromCallbackThread(pipeline, 1) + observerFrozen.await() + + val burst = 3_000 + submitFromCallbackThread(pipeline, burst, from = 1) + pipeline.submit(update(SENTINEL)) + val submitted = 1 + burst + 1 - awaitCount(total, timeoutMs = 60_000) { laneSeen.get() } - assertEquals("a lane must never drop", total, laneSeen.get()) + // submit() only fills the ingest queue. Wait for the pump to have emitted all of + // it, so the observer is provably frozen across the whole emission sequence and + // its buffer has certainly overflowed before it is allowed to run again. + awaitCount(submitted, "the pump to dispatch the burst") { dispatched(pipeline) } - delay(500) + // The sentinel is the newest value, so it is always in the buffer: once the + // observer has seen it, nothing more is coming and its count is final. + release.complete(Unit) + sentinelSeen.await() + + awaitCount(submitted, "updates on the lane") { laneSeen.get() } + assertEquals("a lane must never drop", submitted, laneSeen.get()) assertTrue( - "the observation flow is expected to conflate under this load, saw ${observerSeen.get()} of $total", - observerSeen.get() < total + "the observation flow must conflate: it saw ${observerSeen.get()} of $submitted", + observerSeen.get() < submitted ) observer.cancel() } - @Test + @Test(timeout = 60_000) fun `metrics expose the backlog of a stalled lane`() = runBlocking { val pipeline = pipeline() val gate = CompletableDeferred() @@ -213,17 +282,17 @@ class TdUpdatePipelineTest { submitFromCallbackThread(pipeline, 250) val backlogPattern = Regex("stalled: backlog=(\\d+)") - withTimeout(10_000) { - while (laneBacklog(backlogPattern, pipeline) < 249) delay(5) + awaitCount(249, "the stalled lane's backlog to be reported") { + backlogPattern.find(pipeline.metrics())?.groupValues?.get(1)?.toInt() ?: 0 } val metrics = pipeline.metrics() gate.complete(Unit) - assertTrue(metrics, metrics.contains("submitted=250")) - assertTrue(metrics, laneBacklog(backlogPattern, pipeline) >= 0) } - private fun laneBacklog(pattern: Regex, pipeline: TdUpdatePipeline): Int = - pattern.find(pipeline.metrics())?.groupValues?.get(1)?.toInt() ?: -1 + private companion object { + /** Distinct from every generated sequence number. */ + private const val SENTINEL = -1 + } }