Stop failing a send permanently when the request timed out

stream-chat-android-core/src/main/java/io/getstream/chat/android/client/errors/ChatError.kt

@@ -19,15 +19,16 @@ package io.getstream.chat.android.client.errors
 import io.getstream.chat.android.core.internal.InternalStreamChatApi
 import io.getstream.result.Error
 import io.getstream.result.Error.NetworkError.Companion.UNKNOWN_STATUS_CODE
-import java.net.ConnectException
-import java.net.UnknownHostException
+import java.io.IOException
 
 /**
  * Represents the error in the SDK.
  */
 private const val HTTP_TOO_MANY_REQUESTS = 429
 private const val HTTP_TIMEOUT = 408
 private const val HTTP_API_ERROR = 500
+private const val MESSAGE_DUPLICATE_PREFIX = "a message with ID"
+private const val MESSAGE_DUPLICATE_SUFFIX = "already exists"
 
 /**
  * Creates [Error.NetworkError] from [ChatErrorCode] with custom status code and optional cause.
@@ -71,14 +72,28 @@ public fun Error.isPermanent(): Boolean {
 
         when {
             statusCode in temporaryErrors -> false
-            cause is UnknownHostException || cause is ConnectException -> false
+            // Transport failures leave the outcome unknown; a send that did land is caught by
+            // isMessageAlreadyExists when it is retried
+            cause is IOException -> false
             else -> true
         }
     } else {
         false
     }
 }
 
+/**
+ * @return If the error reports that the message being sent is already stored on the server, which happens
+ * when a send that did reach the backend is retried. The backend answers with the generic validation code
+ * shared by every input error, so the message shape has to be matched as well.
+ */
+@InternalStreamChatApi
+public fun Error.isMessageAlreadyExists(): Boolean =
+    this is Error.NetworkError &&
+        serverErrorCode == ChatErrorCode.VALIDATION_ERROR.code &&
+        message.contains(MESSAGE_DUPLICATE_PREFIX) &&
+        message.contains(MESSAGE_DUPLICATE_SUFFIX)
+
 /**
  * Copies the original [Error] objects with custom message.
  *

stream-chat-android-core/src/test/java/io/getstream/chat/android/client/errors/ChatErrorTest.kt

@@ -22,6 +22,8 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.params.ParameterizedTest
 import org.junit.jupiter.params.provider.Arguments
 import org.junit.jupiter.params.provider.MethodSource
+import java.net.SocketException
+import java.net.SocketTimeoutException
 import java.net.UnknownHostException
 
 internal class ChatErrorTest {
@@ -63,6 +65,15 @@ internal class ChatErrorTest {
         error.isPermanent() `should be equal to` isPermanent
     }
 
+    @ParameterizedTest
+    @MethodSource("isMessageAlreadyExistsArguments")
+    fun `Verify isMessageAlreadyExists() extension function returns proper value`(
+        error: Error,
+        isAlreadyExists: Boolean,
+    ) {
+        error.isMessageAlreadyExists() `should be equal to` isAlreadyExists
+    }
+
     @ParameterizedTest
     @MethodSource("copyWithMessageArguments")
     fun testCopyWithMessage(
@@ -95,6 +106,14 @@ internal class ChatErrorTest {
                 networkError(ChatErrorCode.NETWORK_FAILED, statusCode = 400, cause = UnknownHostException()),
                 false,
             ),
+            Arguments.of(
+                networkError(ChatErrorCode.NETWORK_FAILED, statusCode = 400, cause = SocketTimeoutException()),
+                false,
+            ),
+            Arguments.of(
+                networkError(ChatErrorCode.NETWORK_FAILED, statusCode = 400, cause = SocketException()),
+                false,
+            ),
             Arguments.of(networkError(ChatErrorCode.NETWORK_FAILED, 400), true),
             Arguments.of(networkError(ChatErrorCode.PARSER_ERROR, 400), true),
             Arguments.of(networkError(ChatErrorCode.SOCKET_CLOSED, 400), true),
@@ -144,6 +163,31 @@ internal class ChatErrorTest {
             )
         }
 
+        @JvmStatic
+        fun isMessageAlreadyExistsArguments() = listOf(
+            Arguments.of(alreadyExistsError(), true),
+            Arguments.of(alreadyExistsError(message = "a message with ID abc already exists"), true),
+            Arguments.of(alreadyExistsError(message = "channel members are limited to 100"), false),
+            Arguments.of(alreadyExistsError(message = "poll with ID `p1` already exists"), false),
+            Arguments.of(alreadyExistsError(message = "vote already exists for user `u1` on poll `p1`"), false),
+            Arguments.of(
+                alreadyExistsError(
+                    message = "SendMessage failed with error: \"a message with ID abc already exists\"",
+                ),
+                true,
+            ),
+            Arguments.of(
+                alreadyExistsError(code = ChatErrorCode.AUTHENTICATION_ERROR.code),
+                false,
+            ),
+            Arguments.of(Error.GenericError("a message with ID abc already exists"), false),
+        )
+
+        private fun alreadyExistsError(
+            code: Int = ChatErrorCode.VALIDATION_ERROR.code,
+            message: String = "a message with ID abc already exists",
+        ): Error.NetworkError = Error.NetworkError(message, code, 400)
+
         private fun networkError(
             code: ChatErrorCode,
             statusCode: Int = 400,

stream-chat-android-offline/src/main/java/io/getstream/chat/android/offline/plugin/listener/internal/SendMessageListenerDatabase.kt

@@ -16,6 +16,7 @@
 
 package io.getstream.chat.android.offline.plugin.listener.internal
 
+import io.getstream.chat.android.client.errors.isMessageAlreadyExists
 import io.getstream.chat.android.client.errors.isPermanent
 import io.getstream.chat.android.client.extensions.enrichWithCid
 import io.getstream.chat.android.client.extensions.internal.users
@@ -79,6 +80,18 @@ internal class SendMessageListenerDatabase(
         message: Message,
         error: Error,
     ) {
+        if (error.isMessageAlreadyExists()) {
+            StreamLog.w(TAG) { "[handleSendMessageFailure] message already stored server side" }
+            messageRepository.insertMessage(
+                message.copy(
+                    syncStatus = SyncStatus.COMPLETED,
+                    // The server stored the message but its reply never arrived, so the local date
+                    // stands in until the next channel query replaces the row with the server copy.
+                    createdAt = message.createdAt ?: message.createdLocallyAt,
+                ),
+            )
+            return
+        }
         val isPermanentError = error.isPermanent()
         StreamLog.w(TAG) { "[handleSendMessageFailure] isPermanentError: $isPermanentError" }
 

stream-chat-android-offline/src/test/java/io/getstream/chat/android/offline/plugin/listener/internal/SendMessageListenerDatabaseTest.kt

@@ -16,6 +16,7 @@
 
 package io.getstream.chat.android.offline.plugin.listener.internal
 
+import io.getstream.chat.android.client.errors.ChatErrorCode
 import io.getstream.chat.android.client.extensions.internal.users
 import io.getstream.chat.android.client.persistance.repository.MessageRepository
 import io.getstream.chat.android.client.persistance.repository.UserRepository
@@ -35,6 +36,7 @@ import org.mockito.kotlin.mock
 import org.mockito.kotlin.never
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
+import java.util.Date
 
 @OptIn(ExperimentalCoroutinesApi::class)
 internal class SendMessageListenerDatabaseTest {
@@ -110,6 +112,38 @@ internal class SendMessageListenerDatabaseTest {
         )
     }
 
+    @Test
+    fun `when the send is rejected as a duplicate, the message should be stored as completed`() = runTest {
+        whenever(messageRepository.selectMessage(any())) doReturn null
+
+        val createdLocallyAt = Date()
+        val testMessage = randomMessage(
+            syncStatus = SyncStatus.SYNC_NEEDED,
+            createdAt = null,
+            createdLocallyAt = createdLocallyAt,
+        )
+        val alreadyExists = Error.NetworkError(
+            message = "a message with ID ${testMessage.id} already exists",
+            serverErrorCode = ChatErrorCode.VALIDATION_ERROR.code,
+            statusCode = 400,
+        )
+
+        sendMessageListenerDatabase.onMessageSendResult(
+            result = Result.Failure(alreadyExists),
+            channelType = randomString(),
+            channelId = randomString(),
+            message = testMessage,
+        )
+
+        verify(messageRepository).insertMessage(
+            argThat { message ->
+                message.id == testMessage.id &&
+                    message.syncStatus == SyncStatus.COMPLETED &&
+                    message.createdAt == createdLocallyAt
+            },
+        )
+    }
+
     @Test
     fun `when message is already in database and completed, it should not be inserted again`() = runTest {
         whenever(messageRepository.selectMessage(any())) doReturn randomMessage(syncStatus = SyncStatus.COMPLETED)

stream-chat-android-state/src/main/java/io/getstream/chat/android/state/plugin/listener/internal/SendMessageListenerState.kt

@@ -16,6 +16,7 @@
 
 package io.getstream.chat.android.state.plugin.listener.internal
 
+import io.getstream.chat.android.client.errors.isMessageAlreadyExists
 import io.getstream.chat.android.client.errors.isPermanent
 import io.getstream.chat.android.client.extensions.enrichWithCid
 import io.getstream.chat.android.client.plugin.listeners.SendMessageListener
@@ -85,6 +86,16 @@ internal class SendMessageListenerState(private val logic: LogicRegistry) : Send
         message: Message,
         error: Error,
     ) {
+        if (error.isMessageAlreadyExists()) {
+            StreamLog.w(TAG) { "[handleSendMessageFailure] message already stored server side" }
+            message.copy(
+                syncStatus = SyncStatus.COMPLETED,
+                // The server stored the message but its reply never arrived, so the local date stands
+                // in until the next channel query replaces the row with the server copy.
+                createdAt = message.createdAt ?: message.createdLocallyAt,
+            ).also(::updateState)
+            return
+        }
         val isPermanentError = error.isPermanent()
         StreamLog.w(TAG) { "[handleSendMessageFailure] isPermanentError: $isPermanentError" }
         message.copy(

stream-chat-android-state/src/test/java/io/getstream/chat/android/state/internal/SyncManagerTest.kt

@@ -82,6 +82,7 @@ import org.junit.jupiter.api.Test
 import org.junit.jupiter.api.TestInstance
 import org.mockito.kotlin.any
 import org.mockito.kotlin.anyOrNull
+import org.mockito.kotlin.argThat
 import org.mockito.kotlin.argumentCaptor
 import org.mockito.kotlin.doReturn
 import org.mockito.kotlin.eq
@@ -514,6 +515,28 @@ internal class SyncManagerTest {
         verify(repositoryFacade, never()).insertMessage(any())
     }
 
+    @Test
+    fun `retryMessages should not fail a message the server already stored`() = runTest(testDispatcher) {
+        val message = localRandomMessage()
+        val channelClient: ChannelClient = mock()
+        val alreadyExists = Error.NetworkError(
+            message = "a message with ID ${message.id} already exists",
+            serverErrorCode = ChatErrorCode.VALIDATION_ERROR.code,
+            statusCode = 400,
+        )
+        whenever(repositoryFacade.selectMessageIdsBySyncState(SyncStatus.SYNC_NEEDED)) doReturn listOf(message.id)
+        whenever(repositoryFacade.selectMessage(message.id)) doReturn message
+        whenever(chatClient.channel(message.cid)) doReturn channelClient
+        whenever(channelClient.sendMessage(message)) doReturn TestCall(Result.Failure(alreadyExists))
+
+        val sut = buildSyncManager()
+        sut.retryMessages()
+
+        verify(repositoryFacade, never()).insertMessage(
+            argThat { message -> message.syncStatus == SyncStatus.FAILED_PERMANENTLY },
+        )
+    }
+
     @Test
     fun `when isAutomaticSyncOnReconnectEnabled is true, getSyncHistory should be called on connected event`() =
         runTest(testDispatcher) {

stream-chat-android-state/src/test/java/io/getstream/chat/android/state/plugin/listener/internal/SendMessageListenerStateTest.kt

@@ -16,6 +16,7 @@
 
 package io.getstream.chat.android.state.plugin.listener.internal
 
+import io.getstream.chat.android.client.errors.ChatErrorCode
 import io.getstream.chat.android.client.extensions.cidToTypeAndId
 import io.getstream.chat.android.models.SyncStatus
 import io.getstream.chat.android.randomMessage
@@ -153,6 +154,36 @@ internal class SendMessageListenerStateTest {
         )
     }
 
+    @Test
+    fun `when the send is rejected as a duplicate, the message should be marked completed`() = runTest {
+        val createdLocallyAt = Date()
+        val testMessage = randomMessage(
+            syncStatus = SyncStatus.SYNC_NEEDED,
+            createdAt = null,
+            createdLocallyAt = createdLocallyAt,
+        )
+        val alreadyExists = Error.NetworkError(
+            message = "a message with ID ${testMessage.id} already exists",
+            serverErrorCode = ChatErrorCode.VALIDATION_ERROR.code,
+            statusCode = 400,
+        )
+
+        sendMessageListener.onMessageSendResult(
+            result = Result.Failure(alreadyExists),
+            channelType = randomString(),
+            channelId = randomString(),
+            message = testMessage,
+        )
+
+        verify(channelLogic).upsertMessage(
+            argThat { message ->
+                message.id == testMessage.id &&
+                    message.syncStatus == SyncStatus.COMPLETED &&
+                    message.createdAt == createdLocallyAt
+            },
+        )
+    }
+
     @Test
     fun `when no old message exists, createdLocallyAt should be null on success`() = runTest {
         val testMessage = randomMessage(