Fix various read sync inconsistencies

ts/messageModifiers/ReadSyncs.ts

@@ -103,7 +103,7 @@ export class ReadSyncs extends Collection {
       window.Whisper.Notifications.removeBy({ messageId: found.id });
 
       const message = window.MessageController.register(found.id, found);
-      const readAt = sync.get('readAt');
+      const readAt = Math.min(sync.get('readAt'), Date.now());
 
       // If message is unread, we mark it read. Otherwise, we update the expiration
       //   timer to the time specified by the read sync if it's earlier than
@@ -116,10 +116,7 @@ export class ReadSyncs extends Collection {
           // onReadMessage may result in messages older than this one being
           //   marked read. We want those messages to have the same expire timer
           //   start time as this one, so we pass the readAt value through.
-          const conversation = message.getConversation();
+          message.getConversation()?.onReadMessage(message, readAt);
-          if (conversation) {
-            conversation.onReadMessage(message, readAt);
-          }
         };
 
         if (window.startupProcessingQueue) {
@@ -127,6 +124,7 @@ export class ReadSyncs extends Collection {
           if (conversation) {
             window.startupProcessingQueue.add(
               conversation.get('id'),
+              message.get('sent_at'),
               updateConversation
             );
           }

ts/models/messages.ts

@@ -185,6 +185,8 @@ export class MessageModel extends window.Backbone.Model<MessageAttributesType> {
 
   isSelected?: boolean;
 
+  private pendingMarkRead?: number;
+
   syncPromise?: Promise<CallbackResultType | void>;
 
   initialize(attributes: unknown): void {
@@ -3284,54 +3286,63 @@ export class MessageModel extends window.Backbone.Model<MessageAttributesType> {
 
       const viewSyncs = ViewSyncs.getSingleton().forMessage(message);
 
-      if (
+      if (readSyncs.length !== 0 || viewSyncs.length !== 0) {
-        (readSyncs.length !== 0 || viewSyncs.length !== 0) &&
+        const markReadAt = Math.min(
-        message.get('expireTimer')
-      ) {
-        const existingExpirationStartTimestamp = message.get(
-          'expirationStartTimestamp'
-        );
-        const candidateTimestamps: Array<number> = [
           Date.now(),
-          ...(existingExpirationStartTimestamp
-            ? [existingExpirationStartTimestamp]
-            : []),
           ...readSyncs.map(sync => sync.get('readAt')),
-          ...viewSyncs.map(sync => sync.get('viewedAt')),
+          ...viewSyncs.map(sync => sync.get('viewedAt'))
-        ];
-        message.set(
-          'expirationStartTimestamp',
-          Math.min(...candidateTimestamps)
         );
-        changed = true;
-      }
 
-      let newReadStatus: undefined | ReadStatus.Read | ReadStatus.Viewed;
+        if (message.get('expireTimer')) {
-      if (viewSyncs.length) {
+          const existingExpirationStartTimestamp = message.get(
-        newReadStatus = ReadStatus.Viewed;
+            'expirationStartTimestamp'
-      } else if (readSyncs.length) {
+          );
-        newReadStatus = ReadStatus.Read;
+          message.set(
-      }
+            'expirationStartTimestamp',
-
+            Math.min(existingExpirationStartTimestamp ?? Date.now(), markReadAt)
-      if (newReadStatus !== undefined) {
+          );
-        message.set('readStatus', newReadStatus);
+          changed = true;
-        // This is primarily to allow the conversation to mark all older
-        // messages as read, as is done when we receive a read sync for
-        // a message we already know about.
-        const c = message.getConversation();
-        if (c) {
-          c.onReadMessage(message);
         }
-        changed = true;
-      }
 
-      if (isFirstRun && !viewSyncs.length && !readSyncs.length) {
+        let newReadStatus: ReadStatus.Read | ReadStatus.Viewed;
+        if (viewSyncs.length) {
+          newReadStatus = ReadStatus.Viewed;
+        } else {
+          strictAssert(
+            readSyncs.length !== 0,
+            'Should have either view or read syncs'
+          );
+          newReadStatus = ReadStatus.Read;
+        }
+
+        message.set('readStatus', newReadStatus);
+        changed = true;
+
+        this.pendingMarkRead = Math.min(
+          this.pendingMarkRead ?? Date.now(),
+          markReadAt
+        );
+      } else if (isFirstRun) {
         conversation.set({
           unreadCount: (conversation.get('unreadCount') || 0) + 1,
           isArchived: false,
         });
       }
 
+      if (!isFirstRun && this.pendingMarkRead) {
+        const markReadAt = this.pendingMarkRead;
+        this.pendingMarkRead = undefined;
+
+        // This is primarily to allow the conversation to mark all older
+        // messages as read, as is done when we receive a read sync for
+        // a message we already know about.
+        //
+        // We run this when `isFirstRun` is false so that it triggers when the
+        // message and the other ones accompanying it in the batch are fully in
+        // the database.
+        message.getConversation()?.onReadMessage(message, markReadAt);
+      }
+
       // Check for out-of-order view once open syncs
       if (isTapToView(message.attributes)) {
         const viewOnceOpenSync = ViewOnceOpenSyncs.getSingleton().forMessage(

ts/util/StartupQueue.ts

@@ -1,30 +1,41 @@
 // Copyright 2021 Signal Messenger, LLC
 // SPDX-License-Identifier: AGPL-3.0-only
+/* eslint-disable no-restricted-syntax */
+
+import * as Errors from '../types/errors';
+
+type EntryType = Readonly<{
+  value: number;
+  callback(): void;
+}>;
 
 export class StartupQueue {
-  set: Set<string>;
+  private readonly map = new Map<string, EntryType>();
 
-  items: Array<() => void>;
+  public add(id: string, value: number, f: () => void): void {
-
+    const existing = this.map.get(id);
-  constructor() {
+    if (existing && existing.value >= value) {
-    this.set = new Set();
-    this.items = [];
-  }
-
-  add(id: string, f: () => void): void {
-    if (this.set.has(id)) {
       return;
     }
 
-    this.items.push(f);
+    this.map.set(id, { value, callback: f });
-    this.set.add(id);
   }
 
-  flush(): void {
+  public flush(): void {
-    const { items } = this;
+    window.log.info('StartupQueue: Processing', this.map.size, 'actions');
-    window.log.info('StartupQueue: Processing', items.length, 'actions');
+
-    items.forEach(f => f());
+    const values = Array.from(this.map.values());
-    this.items = [];
+    this.map.clear();
-    this.set.clear();
+
+    for (const { callback } of values) {
+      try {
+        callback();
+      } catch (error) {
+        window.log.error(
+          'StartupQueue: Failed to process item due to error',
+          Errors.toLogFormat(error)
+        );
+      }
+    }
   }
 }