GetStream · andremion · Jul 30, 2026 · Jul 30, 2026 · Jul 31, 2026 · Jul 31, 2026
@@ -94,7 +94,12 @@ jobs:
         if: failure()
         with:
           name: logs_${{ env.ANDROID_API_LEVEL }}
-          path: fastlane/stream-chat-test-mock-server/logs/*
+          # The allure results carry the retry rule's failure artifacts (logcat,
+          # screenshot, hierarchy per failed attempt), which TestOps deduplicates
+          # away when the same test passes on another API level.
+          path: |
+            fastlane/stream-chat-test-mock-server/logs/*
+            fastlane/allure-results/*
   slack:
     permissions:
       actions: write

@@ -11,6 +11,15 @@ on:
         type: string
         required: true
         default: main
+      api_level:
+        description: 'Android API level to run on'
+        type: string
+        required: false
+        default: '36'
+      test_class:
+        description: 'Fully qualified test class to run on its own. Every batch job then runs it, giving independent samples of a flaky test.'
+        type: string
+        required: false
 
 concurrency:
   group: ${{ github.workflow }}-${{ github.ref }}
@@ -70,7 +79,7 @@ jobs:
           - batch: 2
       fail-fast: false
     env:
-      ANDROID_API_LEVEL: 36
+      ANDROID_API_LEVEL: ${{ inputs.api_level || '36' }}
       LAUNCH_ID: ${{ needs.allure_testops_launch.outputs.launch_id }}
     steps:
       - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1
@@ -90,7 +99,7 @@ jobs:
           profile: pixel
           arch : x86_64
           emulator-options: -no-snapshot-save -no-window -no-audio -no-boot-anim -gpu swiftshader_indirect -camera-back none -camera-front none
-          script: bundle exec fastlane run_e2e_test batch:${{ matrix.batch }} batch_count:${{ strategy.job-total }} mock_server_branch:${{ inputs.mock_server_branch }}
+          script: bundle exec fastlane run_e2e_test batch:${{ matrix.batch }} batch_count:${{ strategy.job-total }} mock_server_branch:${{ inputs.mock_server_branch }} ${{ inputs.test_class && format('test_class:{0}', inputs.test_class) || '' }}
       - name: Allure TestOps Upload
         if: ${{ env.LAUNCH_ID != '' && (success() || failure()) }}
         run: bundle exec fastlane allure_upload
@@ -109,4 +118,9 @@ jobs:
         timeout-minutes: 10
         with:
           name: logs_${{ matrix.batch }}
-          path: fastlane/stream-chat-test-mock-server/logs/*
+          # The allure results carry the retry rule's failure artifacts (logcat,
+          # screenshot, hierarchy per failed attempt), which TestOps deduplicates
+          # away when the same test passes in another batch.
+          path: |
+            fastlane/stream-chat-test-mock-server/logs/*
+            fastlane/allure-results/*
@@ -107,8 +107,18 @@ lane :run_e2e_test do |options|
 
   backend_test_class = 'io.getstream.chat.android.compose.tests.BackendTests'
 
+  targeted_class = options[:test_class].to_s
+  # The value is interpolated into a shell command, so only a class name is accepted,
+  # optionally with a single method. A typo also fails here instead of in the shell.
+  UI.user_error!("Invalid test_class: #{targeted_class}") unless
+    targeted_class.empty? || targeted_class.match?(/\A[\w.$]+(#\w+)?\z/)
+
   if options[:use_backend]
     test_filter = "-e class #{backend_test_class}"
+  elsif !targeted_class.empty?
+    # Targeted run: a single class and no batching, so every batch job runs it and gives an
+    # independent sample. Used to reproduce a failure that only shows on one API level.
+    test_filter = "-e class #{targeted_class}"
   else
     run_tests_in_batches = batch_tests(
       batch: options[:batch],

@@ -340,7 +340,7 @@ internal class ChannelEventHandlerImpl(
                 }
             }
 
-            is NotificationMarkUnreadEvent -> state.updateRead(event.toChannelUserRead())
+            is NotificationMarkUnreadEvent -> state.replaceRead(event.toChannelUserRead())
             is MessageDeliveredEvent -> state.updateDelivered(event.toChannelUserRead())
             // Invitation events
             is NotificationInviteAcceptedEvent -> {

@@ -25,6 +25,7 @@ import io.getstream.chat.android.client.events.HasChannel
 import io.getstream.chat.android.client.events.TypingStartEvent
 import io.getstream.chat.android.client.events.UserStartWatchingEvent
 import io.getstream.chat.android.client.events.UserStopWatchingEvent
+import io.getstream.chat.android.client.extensions.getCreatedAtOrDefault
 import io.getstream.chat.android.client.extensions.internal.NEVER
 import io.getstream.chat.android.client.internal.state.message.attachments.internal.AttachmentUrlValidator
 import io.getstream.chat.android.client.internal.state.plugin.logic.channel.internal.SearchLogic
@@ -925,17 +926,22 @@ internal class ChannelStateLogic(
             processedMessageIds.put(message.id, true)
             return
         }
-        // Skip update if the event is outdated
+        // Skip update for messages the user has already read. The read dates only advance
+        // with server-confirmed values, so a replayed event for a message older than the
+        // last read must not count again.
         val currentRead = mutableState.read.value
-        if (currentRead != null && currentRead.lastReceivedEventDate.after(eventReceivedDate)) {
+        if (currentRead != null && !message.getCreatedAtOrDefault(Date()).after(currentRead.lastRead)) {
             processedMessageIds.put(message.id, true)
             return
         }
-        // Update the unread count
+        // Update the unread count. Replays are excluded by processedMessageIds and the
+        // last read date; the event clock must not gate the count, because a queued event
+        // can legitimately carry an older date than an already processed one. The clock
+        // only moves forward, so an out of order event cannot regress it.
         currentRead?.let {
             updateRead(
                 it.copy(
-                    lastReceivedEventDate = eventReceivedDate,
+                    lastReceivedEventDate = maxOf(it.lastReceivedEventDate, eventReceivedDate),
                     unreadMessages = it.unreadMessages.inc(),
                 ),
             )

@@ -63,6 +63,7 @@ import kotlinx.coroutines.flow.asStateFlow
 import kotlinx.coroutines.flow.update
 import java.util.Date
 import java.util.concurrent.ConcurrentHashMap
+import java.util.concurrent.atomic.AtomicInteger
 import kotlin.math.max
 
 /**
@@ -167,7 +168,8 @@ internal class ChannelStateImpl(
             pending.filter { pagination.isInWindow(it) }
         }
 
-    private val logger by taggedLogger("ChannelStateImpl")
+    private val seq = seqGenerator.incrementAndGet()
+    private val logger by taggedLogger("ChannelStateImpl-$seq")
 
     override val repliedMessage: StateFlow<Message?> = _repliedMessage.asStateFlow()
 
@@ -1054,9 +1056,28 @@ internal class ChannelStateImpl(
         } else {
             reads
         }
+        val before = read.value?.unreadMessages
         _reads.update { current ->
             current + readsToUpsert.associateBy(ChannelUserRead::getUserId)
         }
+        val after = read.value?.unreadMessages
+        if (before != after) {
+            val incoming = reads.firstOrNull { it.getUserId() == currentUser.value?.id }?.unreadMessages
+            logger.d { "[updateReads] cid: $cid; unreadMessages: $before -> $after; incoming: $incoming" }
+        }
+    }
+
+    /**
+     * Replaces the read state of [read]'s user with no arbitration against the local one.
+     *
+     * For read states the server reports explicitly, where the reported position and count are
+     * the truth even when the position moves backwards, as a mark unread does. [updateReads]
+     * cannot be used for those, because it protects the local count against stale payloads.
+     */
+    fun replaceRead(read: ChannelUserRead) {
+        val previous = _reads.value[read.getUserId()]?.unreadMessages
+        logger.d { "[replaceRead] cid: $cid; unreadMessages: $previous -> ${read.unreadMessages}" }
+        _reads.update { current -> current + (read.getUserId() to read) }
     }
 
     /**
@@ -1067,63 +1088,62 @@ internal class ChannelStateImpl(
      * @param message The new message that triggered the event.
      */
     fun updateCurrentUserRead(eventReceivedDate: Date, message: Message) {
-        // Skip update if the message was already processed
-        val isProcessed = processedMessageIds[message.id] == true
-        if (isProcessed) {
-            return
-        }
-        // Skip update if the channel is muted
-        val isMuted = muted.value
-        if (isMuted) {
-            processedMessageIds.put(message.id, true)
-            return
-        }
-        // Skip update for thread replies not shown in channel
-        val isThreadReplyNotInChannel = message.parentId != null && !message.showInChannel
-        if (isThreadReplyNotInChannel) {
-            processedMessageIds.put(message.id, true)
-            return
-        }
-        // Skip update for messages from current user
-        val isFromCurrentUser = message.user.id == currentUser.value?.id
-        if (isFromCurrentUser) {
-            processedMessageIds.put(message.id, true)
-            return
-        }
-        // Skip update for messages from muted users
-        val isFromMutedUser = mutedUsers.value.any { it.target?.id == message.user.id }
-        if (isFromMutedUser) {
-            processedMessageIds.put(message.id, true)
+        if (processedMessageIds[message.id] == true) {
+            logger.d { "[updateCurrentUserRead] skipped msgId(${message.id}): already processed" }
             return
         }
-        // Skip update for messages from shadow banned users
-        if (message.shadowed) {
+        val currentRead = read.value
+        val skipReason = unreadCountSkipReason(currentRead, message)
+        if (skipReason != null) {
+            logger.d { "[updateCurrentUserRead] skipped msgId(${message.id}): $skipReason" }
             processedMessageIds.put(message.id, true)
             return
         }
-        // Skip update for silent messages
-        if (message.silent) {
+        if (currentRead == null) {
+            logger.d { "[updateCurrentUserRead] skipped msgId(${message.id}): no read state for the current user" }
             processedMessageIds.put(message.id, true)
             return
         }
-        // Skip update if the event is outdated
-        val currentRead = read.value
-        if (currentRead != null && currentRead.lastReceivedEventDate.after(eventReceivedDate)) {
-            processedMessageIds.put(message.id, true)
-            return
+        // Written straight into the map instead of through updateRead: this is the local count,
+        // not an incoming read, so it must not pass through the merge that arbitrates external
+        // reads, and the read-modify-write has to be atomic to survive a burst of events. The
+        // clock only moves forward, so an out of order event cannot regress it.
+        val userId = currentRead.getUserId()
+        _reads.update { current ->
+            val existing = current[userId] ?: currentRead
+            current + (
+                userId to existing.copy(
+                    lastReceivedEventDate = maxOf(existing.lastReceivedEventDate, eventReceivedDate),
+                    unreadMessages = existing.unreadMessages.inc(),
+                )
+                )
         }
-        // Update the unread count
-        currentRead?.let {
-            updateRead(
-                it.copy(
-                    lastReceivedEventDate = eventReceivedDate,
-                    unreadMessages = it.unreadMessages.inc(),
-                ),
-            )
+        logger.d {
+            "[updateCurrentUserRead] counted msgId(${message.id}); unreadMessages: ${read.value?.unreadMessages}"
         }
         processedMessageIds.put(message.id, true)
     }
 
+    /**
+     * Why [message] does not increment the current user's unread count, or `null` when it does.
+     *
+     * The returned reason is logged, so a wrong count can be traced to the exact condition that
+     * suppressed it.
+     */
+    private fun unreadCountSkipReason(currentRead: ChannelUserRead?, message: Message): String? = when {
+        muted.value -> "channel is muted"
+        message.parentId != null && !message.showInChannel -> "thread reply not shown in channel"
+        message.user.id == currentUser.value?.id -> "own message"
+        mutedUsers.value.any { it.target?.id == message.user.id } -> "author is muted"
+        message.shadowed -> "message is shadowed"
+        message.silent -> "message is silent"
+        // The read dates only advance with server-confirmed values, so a replayed event for a
+        // message older than the last read must not count again.
+        currentRead != null && !message.getCreatedAtOrDefault(Date()).after(currentRead.lastRead) ->
+            "already read: createdAt(${message.createdAt}) <= lastRead(${currentRead.lastRead})"
+        else -> null
+    }
+
     /**
      * Marks the channel as read for the current user if the following conditions are met:
      * 1. Read events are enabled in the channel configuration.
@@ -1149,12 +1169,11 @@ internal class ChannelStateImpl(
             return true
         }
         return if (lastMessage.id != currentUserRead.lastReadMessageId) {
-            // The last message is different from the last read message, we can mark the channel as read
-            val updatedRead = currentUserRead.copy(
-                lastReceivedEventDate = lastMessage.getCreatedAtOrDefault(Date()),
-                lastRead = lastMessage.getCreatedAtOrDefault(Date()),
-                unreadMessages = 0,
-            )
+            // Zero the count optimistically, but leave the read dates untouched: they stay
+            // anchored to what the server confirmed, and the server's own message.read echo
+            // advances them. Stamping them with the newest loaded message's date here made
+            // messages that were still queued behind this call count as already seen.
+            val updatedRead = currentUserRead.copy(unreadMessages = 0)
             _reads.update { current ->
                 current + (updatedRead.getUserId() to updatedRead)
             }
@@ -1614,12 +1633,22 @@ internal class ChannelStateImpl(
         localRead: ChannelUserRead,
         serverRead: ChannelUserRead,
     ): ChannelUserRead {
-        return if (localRead.lastReceivedEventDate.after(serverRead.lastReceivedEventDate)) {
-            // Local state is more recent, preserve it but merge other fields from server
+        // The read position is the only thing an incoming read can tell us that the local state
+        // does not already know: the count between two read positions is maintained locally, one
+        // increment per new message event. So the incoming count is taken only when the position
+        // moved forward, which is what happens when the channel is read elsewhere. At the same or
+        // an earlier position the incoming count may be stale, as with the channel list's own copy
+        // of the channel, so the local count is kept. A mark unread moves the position backwards on
+        // purpose and its count is authoritative, which is why it is applied by [replaceRead]
+        // instead of going through this arbitration. Note
+        // that lastReceivedEventDate cannot be used to decide this: the server does not send it,
+        // it is synthesised while mapping, so comparing it against a locally advanced clock is
+        // meaningless and lets a stale snapshot (for example the channel list's own copy of the
+        // channel) overwrite a count that is correct.
+        return if (!serverRead.lastRead.after(localRead.lastRead)) {
             logger.d {
-                "[updateReads] Local read state is more recent, preserving: " +
-                    "local.lastReceivedEventDate=${localRead.lastReceivedEventDate}, " +
-                    "server.lastReceivedEventDate=${serverRead.lastReceivedEventDate}, " +
+                "[updateReads] keeping the local count: " +
+                    "local.lastRead=${localRead.lastRead}, server.lastRead=${serverRead.lastRead}, " +
                     "local.unreadMessages=${localRead.unreadMessages}, " +
                     "server.unreadMessages=${serverRead.unreadMessages}"
             }
@@ -1632,7 +1661,7 @@ internal class ChannelStateImpl(
                 // lastReceivedEventDate and unreadMessages are preserved from local (not set in copy)
             )
         } else {
-            // Server data is more recent, use it
+            // The read position moved forward, so the incoming count describes it best.
             serverRead
         }
     }
@@ -1675,6 +1704,9 @@ internal class ChannelStateImpl(
     }
 
     private companion object {
+        /** Distinguishes instances of the same channel in the logs. */
+        private val seqGenerator = AtomicInteger()
+
         /**
          * Hard limit for cached latest messages to prevent unbounded memory growth while in search mode.
          * When the user is viewing messages around a specific message (e.g., from a deep link or search),

@@ -17,7 +17,6 @@
 package io.getstream.chat.android.client.internal.state.plugin.state.channel.internal
 
 import io.getstream.chat.android.client.channel.state.ChannelState
-import io.getstream.chat.android.client.extensions.getCreatedAtOrDefault
 import io.getstream.chat.android.client.extensions.getCreatedAtOrNull
 import io.getstream.chat.android.client.extensions.internal.updateUsers
 import io.getstream.chat.android.client.extensions.internal.wasCreatedAfter
@@ -611,15 +610,12 @@ internal class ChannelStateLegacyImpl(
                     currentUserRead
                         .takeIf { it.lastReadMessageId != lastMessage.id || it.unreadMessages > 0 }
                         ?.let {
-                            upsertReads(
-                                listOf(
-                                    it.copy(
-                                        lastReceivedEventDate = lastMessage.getCreatedAtOrDefault(Date()),
-                                        lastRead = lastMessage.getCreatedAtOrDefault(Date()),
-                                        unreadMessages = 0,
-                                    ),
-                                ),
-                            )
+                            // Zero the count optimistically, but leave the read dates untouched:
+                            // they stay anchored to what the server confirmed, and the server's
+                            // own message.read echo advances them. Stamping them with the newest
+                            // loaded message's date here made messages that were still queued
+                            // behind this call count as already seen.
+                            upsertReads(listOf(it.copy(unreadMessages = 0)))
                             true
                         }
             }

@@ -1025,12 +1025,15 @@ internal class ChannelEventHandlerImplTest {
     }
 
     @Test
-    fun `When NotificationMarkUnreadEvent is handled, Then read is updated`() {
+    fun `When NotificationMarkUnreadEvent is handled, Then the read is replaced`() {
+        // Replaced rather than merged: a mark unread moves the read position backwards on purpose,
+        // and updateRead would keep the local count instead of the one the server reported.
         val event = randomNotificationMarkUnreadEvent(cid = cid)
 
         handler.handle(event)
 
-        verify(state).updateRead(event.toChannelUserRead())
+        verify(state).replaceRead(event.toChannelUserRead())
+        verify(state, never()).updateRead(any())
     }
 
     @Test