Skip to content

data: stop silently dropping TDLib updates - #348

Merged
gdlbo merged 2 commits into
monogram-android:tdlib-updates-fixfrom
tdrkDev:fix/tdlib-update-pipeline-loss
Aug 5, 2026
Merged

data: stop silently dropping TDLib updates#348
gdlbo merged 2 commits into
monogram-android:tdlib-updates-fixfrom
tdrkDev:fix/tdlib-update-pipeline-loss

Conversation

@tdrkDev

@tdrkDev tdrkDev commented Aug 5, 2026

Copy link
Copy Markdown
Contributor

(c) Claude Opus 5 & Claude Fable 5

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.

tdrkDev and others added 2 commits August 5, 2026 16:48
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<T>() 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 <noreply@anthropic.com>
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 <noreply@anthropic.com>
@gdlbo
gdlbo changed the base branch from develop to tdlib-updates-fix August 5, 2026 14:10
@gdlbo
gdlbo merged commit 9d76094 into monogram-android:tdlib-updates-fix Aug 5, 2026
1 check passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants