From e58a2f0ef48b001fa2185a19db0133b7ae9ee124 Mon Sep 17 00:00:00 2001 From: Lukas Remis Date: Fri, 18 Sep 2026 18:21:30 +0200 Subject: [PATCH] Sync: a flag set on the server left the thread counters stale MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reported from the running app: a message showed as unread in the list — bold, blue dot — while the reading pane showed it as read, and only a manual "mark as read" reconciled them. The two views read different things. The pane reads the message's own flags; the list reads `threads.unread_count`, denormalized from them. `apply_delta` updated message flags with a bare UPDATE and never recomputed the thread, so a flag set anywhere but in this app — read on a phone, filed by a server-side rule — moved one and not the other. Auto-mark-on-read could not fix it either: it sees a message that is already \Seen and correctly does nothing, leaving the counter wrong forever. Expunges had the same hole, which also left `message_count` overstated. `apply_delta` now collects the affected threads and re-denormalizes them after the deletes, so a thread emptied by an expunge is removed rather than recounted. Migration 0018 repairs rows that already drifted — recomputed from the messages, so it is a no-op on a healthy database. One thread was stale here; installs that have been syncing longer will have more. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_018SqJYt7jD7oUEkoY68sBZo --- backend/app/sync/engine.py | 38 ++++++++-- .../versions/0018_repair_thread_counters.py | 55 ++++++++++++++ backend/tests/sync/test_secondary_poll.py | 76 +++++++++++++++++++ 3 files changed, 163 insertions(+), 6 deletions(-) create mode 100644 backend/migrations/versions/0018_repair_thread_counters.py diff --git a/backend/app/sync/engine.py b/backend/app/sync/engine.py index 3401e1f..cb839f4 100644 --- a/backend/app/sync/engine.py +++ b/backend/app/sync/engine.py @@ -23,6 +23,7 @@ from app.providers.imap_smtp import ImapSmtpProvider from app.search.factory import get_search_backend from app.sync.folder_names import infer_special_use from app.sync.provider_factory import build_provider +from app.threading_.threader import denormalize_thread logger = logging.getLogger(__name__) @@ -371,22 +372,42 @@ async def apply_delta( if r is not None: results.append(r) max_uid = max(max_uid, remote.uid) + # Threads whose denormalized counters this batch invalidates. A flag set + # on the server (read on a phone, filed by a server-side rule) and an + # expunge both change what `threads.unread_count` should say, and neither + # goes through apply_flags_to_messages — so without this the row keeps its + # old count and the list shows a message as unread that the reading pane + # shows as read. + touched_threads: set[uuid.UUID] = set() + for change in delta.flag_changes: await session.execute( update(Message) .where(Message.mailbox_id == mailbox.id, Message.uid == change.uid) .values(flags=sorted(change.flags)) ) - if delta.expunged_uids: - # Server-side expunges (QRESYNC/VANISHED, SPEC §6.3). Capture the row ids - # first so we can drop them from the search index after the delete. - gone_ids = list( + if delta.flag_changes: + touched_threads.update( await session.scalars( - select(Message.id).where( - Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids) + select(Message.thread_id).where( + Message.mailbox_id == mailbox.id, + Message.uid.in_([c.uid for c in delta.flag_changes]), + Message.thread_id.is_not(None), ) ) ) + if delta.expunged_uids: + # Server-side expunges (QRESYNC/VANISHED, SPEC §6.3). Capture the row ids + # first so we can drop them from the search index after the delete. + gone = ( + await session.execute( + select(Message.id, Message.thread_id).where( + Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids) + ) + ) + ).all() + gone_ids = [row[0] for row in gone] + touched_threads.update(row[1] for row in gone if row[1] is not None) await session.execute( delete(Message).where( Message.mailbox_id == mailbox.id, Message.uid.in_(delta.expunged_uids) @@ -398,6 +419,11 @@ async def apply_delta( .where(Draft.account_id == account.id, Draft.imap_uid.in_(delta.expunged_uids)) .values(imap_uid=None) ) + # After the deletes, so a thread left with no messages is removed rather + # than recounted (denormalize_thread handles that case). + for tid in touched_threads: + await denormalize_thread(session, tid) + updates = {"last_seen_uid": max_uid} if delta.new_highestmodseq is not None: updates["highestmodseq"] = delta.new_highestmodseq diff --git a/backend/migrations/versions/0018_repair_thread_counters.py b/backend/migrations/versions/0018_repair_thread_counters.py new file mode 100644 index 0000000..f82361a --- /dev/null +++ b/backend/migrations/versions/0018_repair_thread_counters.py @@ -0,0 +1,55 @@ +"""repair thread counters left stale by server-side flag changes and expunges + +`threads.unread_count`, `pinned` and `message_count` are denormalized from the +thread's messages. Two sync paths updated messages without recomputing them: a +flag change reported by the server (mail read on a phone, or filed by a +server-side rule) and a QRESYNC expunge. The row then disagreed with its own +messages — the message list showing a message as unread that the reading pane +showed as read, and only a manual "mark as read" reconciling it. + +The code paths are fixed; this repairs rows that already drifted. Recomputes from +the messages themselves, so it is safe to run on a healthy database: threads that +already agree are left untouched. + +Revision ID: 0018 +Revises: 0017 +Create Date: 2026-09-18 +""" + +from __future__ import annotations + +from collections.abc import Sequence + +from alembic import op + +revision: str = "0018" +down_revision: str | None = "0017" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.execute( + r""" + UPDATE threads t + SET unread_count = s.unread, + pinned = s.pinned, + message_count = s.total + FROM ( + SELECT thread_id, + count(*) AS total, + count(*) FILTER (WHERE NOT ('\Seen' = ANY(flags))) AS unread, + bool_or('\Flagged' = ANY(flags)) AS pinned + FROM messages + WHERE thread_id IS NOT NULL + GROUP BY thread_id + ) AS s + WHERE t.id = s.thread_id + AND (t.unread_count, t.pinned, t.message_count) IS DISTINCT FROM + (s.unread, s.pinned, s.total) + """ + ) + + +def downgrade() -> None: + """Nothing to undo: this only makes rows agree with the messages they describe.""" diff --git a/backend/tests/sync/test_secondary_poll.py b/backend/tests/sync/test_secondary_poll.py index 4da6cee..0adc8c0 100644 --- a/backend/tests/sync/test_secondary_poll.py +++ b/backend/tests/sync/test_secondary_poll.py @@ -241,3 +241,79 @@ async def test_apply_delta_expunges_vanished_uids() -> None: # type: ignore[no- await cleanup.execute(delete(User).where(User.id == user.id)) await session.close() await engine.dispose() + + +async def test_server_side_flag_change_updates_the_thread_counters() -> None: + """A flag set elsewhere (phone, webmail, server rule) must re-denormalize. + + `threads.unread_count` is what the message list renders, while the reading + pane renders the message's own flags. When a delta updated flags without + recomputing the thread, the two disagreed forever: the list kept showing a + message as unread that the pane showed as read, and only a manual "mark as + read" — which goes through a different path — reconciled them. + """ + from app.models import Thread + from app.threading_.threader import denormalize_thread + + engine = create_async_engine(get_settings().database_url, poolclass=NullPool) + sm = async_sessionmaker(engine, expire_on_commit=False) + session = sm() + user = User(handle=f"fc{uuid.uuid4().hex[:8]}@test.local", password_hash=hash_password("pw")) + session.add(user) + await session.flush() + account = Account( + user_id=user.id, + email="me@example.com", + imap_config={}, + smtp_config={}, + secret_ct=b"", + discovery_src="manual", + ) + session.add(account) + await session.flush() + inbox = Mailbox( + account_id=account.id, user_id=user.id, path="INBOX", special_use="inbox", last_seen_uid=5 + ) + thread = Thread(user_id=user.id, subject_norm="t", last_message_at=datetime.now(UTC)) + session.add_all([inbox, thread]) + await session.flush() + message = Message( + user_id=user.id, + account_id=account.id, + mailbox_id=inbox.id, + thread_id=thread.id, + uid=5, + internal_date=datetime.now(UTC), + blob_sha256="0" * 64, + flags=[], + ) + session.add(message) + await session.flush() + await denormalize_thread(session, thread.id) + await session.commit() + assert (await session.get(Thread, thread.id)).unread_count == 1 + + class FakeProvider: + qresync_enabled = True + + async def close(self) -> None: + return None + + class _FlagChange: + uid = 5 + flags = ["\\Seen"] + + delta = MailboxDelta([], [_FlagChange()], [], None, None) + await sync_engine.apply_delta(FakeProvider(), account, inbox, delta) # type: ignore[arg-type] + + try: + session.expunge_all() + refreshed = await session.get(Thread, thread.id) + assert refreshed.unread_count == 0, ( + "the server said the message was read; the thread counter did not follow" + ) + finally: + async with session_scope() as cleanup: + await cleanup.execute(delete(User).where(User.id == user.id)) + await session.close() + await engine.dispose()