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 ee15eafd..b0febf42 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<Pair<Long, Long>, Job>()
     private val missingMessageCooldownUntil = ConcurrentHashMap<Pair<Long, Long>, Long>()
     private val sendQueue = Channel<suspend () -> Unit>(Channel.BUFFERED)
-    override val newMessageFlow = MutableSharedFlow<MessageModel>()
-    override val messageEditedFlow = MutableSharedFlow<MessageModel>()
+    // 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<MessageModel>(scope)
+    override val newMessageFlow = newMessages.events
+    private val messageEdits = OrderedEventFlow<MessageModel>(scope)
+    override val messageEditedFlow = messageEdits.events
     private val messageReads = OrderedEventFlow<ReadUpdate>(scope)
     override val messageReadFlow = messageReads.events
-    override val messageUploadProgressFlow = MutableSharedFlow<MessageUploadProgressEvent>()
+    override val messageUploadProgressFlow = MutableSharedFlow<MessageUploadProgressEvent>(
+        extraBufferCapacity = 64,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
     private val fileDownloads = OrderedEventFlow<FileDownloadEvent>(scope)
     override val fileDownloadFlow = fileDownloads.events
     private val messageDownloads = OrderedEventFlow<MessageDownloadEvent>(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 e5c0070f..69650ce9 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<String, Long>()
-    private val _updates = MutableSharedFlow<TdApi.Update>(
-        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<TdApi.Update> = _updates
+    /**
+     * Observation stream. Conflates under load, so it must not be used to drive durable
+     * state; use [lane] for that.
+     */
+    val updates: SharedFlow<TdApi.Update> = 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 11b95685..7215bd7b 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 00000000..8032f197
--- /dev/null
+++ b/data/src/main/java/org/monogram/data/di/TdUpdatePipeline.kt
@@ -0,0 +1,232 @@
+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<TdApi.Update>(Channel.UNLIMITED)
+
+    private val _updates = MutableSharedFlow<TdApi.Update>(
+        replay = OBSERVER_REPLAY,
+        extraBufferCapacity = OBSERVER_BUFFER,
+        onBufferOverflow = BufferOverflow.DROP_OLDEST
+    )
+
+    /** Observation only — may conflate. Durable consumers must use [lane]. */
+    val updates: SharedFlow<TdApi.Update> = _updates.asSharedFlow()
+
+    private val lanes = CopyOnWriteArrayList<Lane>()
+
+    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<TdApi.Update>(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
+            // 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()
+
+        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 1364c726..d3870c93 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<UpdateDispatcher>().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 511bf1c1..62939307 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 <T : TdApi.Object> execute(function: TdApi.Function<T>): T
+
+    /**
+     * Observation stream. Conflates when a collector falls behind; use [lane] for
+     * anything that drives durable state.
+     */
     val updates: SharedFlow<TdApi.Update>
+
     val isAuthenticated: StateFlow<Boolean>
+
+    /**
+     * 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 6d2fe735..5af3b075 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<Boolean>
         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 65124604..e9d88c6d 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<TdApi.Update>
 
+    /**
+     * 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<TdApi.UpdateAuthorizationState>
 
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 8e2a938e..b8234345 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<TdApi.Update> = 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 <reified T : TdApi.Update> flow(): Flow<T> =
         updates.filterIsInstance<T>()
 
@@ -56,36 +68,44 @@ class UpdateDispatcherImpl(
     override val installedStickerSets = flow<TdApi.UpdateInstalledStickerSets>()
     override val newChat = flow<TdApi.UpdateNewChat>()
     override val attachmentMenuBots = flow<TdApi.UpdateAttachmentMenuBots>()
-    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 887b7a1d..466ddbb4 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<TdApi.UpdateFile>,
+    private val updates: UpdateDispatcher,
     private val scope: CoroutineScope
 ) {
     val customEmojiPaths = SynchronizedLruMap<Long, String>(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 2ec968da..e95b38a9 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 77744333..1009ce34 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 c3ff6166..5fd9416d 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 e85c8541..d92c6c8b 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 6e92d727..dabc1514 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<TdApi.Update>()
         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 6cc97471..02868f99 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<TdApi.Update>()
         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 1f97c437..35276a2c 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<TdApi.Update>()
         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 00000000..eb7de34f
--- /dev/null
+++ b/data/src/test/java/org/monogram/data/di/TdUpdatePipelineTest.kt
@@ -0,0 +1,298 @@
+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 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, 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 {
+
+    private val scopes = mutableListOf<CoroutineScope>()
+    private val pipelines = mutableListOf<TdUpdatePipeline>()
+
+    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()
+    }
+
+    /**
+     * 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(timeout = 60_000)
+    fun `lane receives every update exactly once and in order`() = runBlocking {
+        val pipeline = pipeline()
+        val received = Collections.synchronizedList(mutableListOf<Int>())
+        pipeline.lane("state", scope()) { received.add(seqOf(it)) }
+
+        val total = 5_000
+        submitFromCallbackThread(pipeline, total)
+        awaitCount(total, "updates on the lane") { received.size }
+
+        assertEquals(total, received.size)
+        assertEquals((0 until total).toList(), received.toList())
+    }
+
+    @Test(timeout = 60_000)
+    fun `lane filter is applied and does not create gaps`() = runBlocking {
+        val pipeline = pipeline()
+        val received = Collections.synchronizedList(mutableListOf<Int>())
+        pipeline.lane("even", scope(), filter = { seqOf(it) % 2 == 0 }) { received.add(seqOf(it)) }
+
+        submitFromCallbackThread(pipeline, 1_000)
+        awaitCount(500, "even updates") { received.size }
+
+        assertEquals(500, received.size)
+        assertEquals((0 until 1_000 step 2).toList(), received.toList())
+    }
+
+    @Test(timeout = 60_000)
+    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 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"))
+    }
+
+    /**
+     * 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 gate = CompletableDeferred<Unit>()
+        val blocked = Collections.synchronizedList(mutableListOf<Int>())
+        val healthy = Collections.synchronizedList(mutableListOf<Int>())
+
+        pipeline.lane("blocked", scope()) { gate.await(); blocked.add(seqOf(it)) }
+        pipeline.lane("healthy", scope()) { healthy.add(seqOf(it)) }
+
+        val total = 2_000
+        submitFromCallbackThread(pipeline, total)
+
+        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)
+
+        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(timeout = 60_000)
+    fun `ingestion does not wait for consumers`() = runBlocking {
+        val pipeline = pipeline()
+        val gate = CompletableDeferred<Unit>()
+        val processed = AtomicInteger()
+        pipeline.lane("blocked", scope()) {
+            gate.await()
+            processed.incrementAndGet()
+        }
+
+        // Every submit returns even though the lane handler has never returned.
+        val total = 10_000
+        submitFromCallbackThread(pipeline, total)
+        assertEquals("ingestion must not block on a consumer", 0, processed.get())
+
+        gate.complete(Unit)
+        awaitCount(total, "updates drained after unblocking") { processed.get() }
+        assertEquals(total, processed.get())
+    }
+
+    @Test(timeout = 60_000)
+    fun `lane deregisters when its scope is cancelled`() = runBlocking {
+        val pipeline = pipeline()
+        val ownScope = CoroutineScope(SupervisorJob() + Dispatchers.Default)
+        val seen = AtomicInteger()
+        pipeline.lane("transient", ownScope) { seen.incrementAndGet() }
+
+        submitFromCallbackThread(pipeline, 10)
+        awaitCount(10, "updates before cancellation") { seen.get() }
+
+        ownScope.cancel()
+        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 neither reach nor accumulate for the dead lane.
+        submitFromCallbackThread(pipeline, 100, from = 10)
+        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. 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(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 observerFrozen = CompletableDeferred<Unit>()
+        val release = CompletableDeferred<Unit>()
+        val sentinelSeen = CompletableDeferred<Unit>()
+
+        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)
+            }
+        }
+        awaitObserverCount(pipeline, 1)
+
+        // 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
+
+        // 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) }
+
+        // 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 must conflate: it saw ${observerSeen.get()} of $submitted",
+            observerSeen.get() < submitted
+        )
+        observer.cancel()
+    }
+
+    @Test(timeout = 60_000)
+    fun `metrics expose the backlog of a stalled lane`() = runBlocking {
+        val pipeline = pipeline()
+        val gate = CompletableDeferred<Unit>()
+        pipeline.lane("stalled", scope()) { gate.await() }
+
+        submitFromCallbackThread(pipeline, 250)
+        val backlogPattern = Regex("stalled: backlog=(\\d+)")
+        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"))
+    }
+
+    private companion object {
+        /** Distinct from every generated sequence number. */
+        private const val SENTINEL = -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 f43acde5..7a3fe0c7 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 <T : TdApi.Object> execute(function: TdApi.Function<T>): 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 d1b68e5d..8be7a5ed 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<TdApi.UpdateConnectionState>
     ) : UpdateDispatcher {
         override val all: Flow<TdApi.Update> = 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<TdApi.UpdateNewMessage> = MutableSharedFlow()
         override val activeNotifications: Flow<TdApi.UpdateActiveNotifications> =
             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 6a287135..12855192 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<TdApi.Update>()
         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 e3031ce9..8e253644 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<TdApi.UpdateFile>()
+        val updates = MutableSharedFlow<TdApi.Update>()
         val queue = RecordingQueue()
         val registry = FileMessageRegistry()
         val handler = FileUpdateHandler(
             registry = registry,
             queue = queue,
-            fileUpdatesSource = updates,
+            updates = UpdateDispatcherImpl(FakeTelegramGateway(updates)),
             scope = scope
         )
         val fileCompleted = mutableListOf<Pair<Long, String>>()
@@ -68,6 +73,24 @@ class FileUpdateHandlerTest {
         remote = TdApi.RemoteFile()
     }
 
+    private class FakeTelegramGateway(
+        override val updates: MutableSharedFlow<TdApi.Update>
+    ) : TelegramGateway {
+        override val isAuthenticated = MutableStateFlow(false)
+
+        override suspend fun <T : TdApi.Object> execute(function: TdApi.Function<T>): 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<Int>()
 
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 edfc53e2..54918a7f 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<TdApi.Update>()
         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 b767b72e..d90494b7 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<Boolean> = 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<TdApi.Update> = 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<TdApi.UpdateAuthorizationState> = MutableSharedFlow()
     override val newMessage: Flow<TdApi.UpdateNewMessage> = MutableSharedFlow()
     override val activeNotifications: Flow<TdApi.UpdateActiveNotifications> = 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 00000000..5a5dc63d
--- /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<TdApi.Update>,
+    scope: CoroutineScope,
+    context: CoroutineContext,
+    filter: (TdApi.Update) -> Boolean,
+    handler: suspend (TdApi.Update) -> Unit,
+) {
+    scope.launch(context) {
+        source.filter(filter).collect { handler(it) }
+    }
+}