fix(voip): Android DDP thread safety and VoipPayload bundle parity (#7168)

Author: diegolmello

android/app/build.gradle

@@ -155,6 +155,7 @@ dependencies {
     implementation 'androidx.security:security-crypto:1.1.0'
 
     testImplementation 'junit:junit:4.13.2'
+    testImplementation 'org.robolectric:robolectric:4.14.1'
 
     // For ProcessLifecycleOwner (app foreground detection)
     implementation 'androidx.lifecycle:lifecycle-process:2.8.7'

android/app/src/main/java/chat/rocket/reactnative/voip/DDPClient.kt

@@ -11,6 +11,8 @@ import okhttp3.WebSocketListener
 import org.json.JSONArray
 import org.json.JSONObject
 import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.AtomicInteger
 
 class DDPClient {
     private data class QueuedMethodCall(
@@ -28,17 +30,29 @@ class DDPClient {
 
     private var webSocket: WebSocket? = null
     private val client: OkHttpClient = sharedClient
-    private var sendCounter = 0
+    private val sendCounter = AtomicInteger(0)
+
+    @Volatile
     private var isConnected = false
+
     private val mainHandler = Handler(Looper.getMainLooper())
 
     private val pendingCallbacks = mutableMapOf<String, (JSONObject) -> Unit>()
     private val queuedMethodCalls = mutableListOf<QueuedMethodCall>()
+
+    @Volatile
     private var connectedCallback: ((Boolean) -> Unit)? = null
 
+    @Volatile
+    private var connectTimeoutRunnable: Runnable? = null
+
+    private val connectResultDelivered = AtomicBoolean(false)
+
     var onCollectionMessage: ((JSONObject) -> Unit)? = null
 
     fun connect(host: String, callback: (Boolean) -> Unit) {
+        resetConnectHandshakeState()
+
         val wsUrl = buildWebSocketURL(host)
 
         Log.d(TAG, "Connecting to $wsUrl")
@@ -60,16 +74,26 @@ class DDPClient {
             }
 
             override fun onMessage(webSocket: WebSocket, text: String) {
+                if (webSocket !== this@DDPClient.webSocket) return
                 handleMessage(text)
             }
 
             override fun onFailure(webSocket: WebSocket, t: Throwable, response: Response?) {
+                if (webSocket !== this@DDPClient.webSocket) return
                 Log.e(TAG, "WebSocket failure: ${t.message}")
                 isConnected = false
-                mainHandler.post { callback(false) }
+                mainHandler.post {
+                    // Re-check identity on main thread: disconnect()+connect() can interleave
+                    // between the outer guard (OkHttp thread) and this runnable's execution.
+                    // Without this re-check, a stale failure can hijack a newly installed
+                    // connectedCallback via the CAS in tryDeliverConnectOutcome.
+                    if (webSocket !== this@DDPClient.webSocket) return@post
+                    tryDeliverConnectOutcome(false)
+                }
             }
 
             override fun onClosed(webSocket: WebSocket, code: Int, reason: String) {
+                if (webSocket !== this@DDPClient.webSocket) return
                 Log.d(TAG, "WebSocket closed: $code $reason")
                 isConnected = false
             }
@@ -134,6 +158,7 @@ class DDPClient {
     fun disconnect() {
         Log.d(TAG, "Disconnecting")
         isConnected = false
+        cancelConnectTimeout()
         synchronized(pendingCallbacks) { pendingCallbacks.clear() }
         clearQueuedMethodCalls()
         connectedCallback = null
@@ -143,10 +168,10 @@ class DDPClient {
     }
 
     private fun nextMessage(msg: String): JSONObject {
-        sendCounter++
+        val nextId = sendCounter.incrementAndGet()
         return JSONObject().apply {
             put("msg", msg)
-            put("id", "ddp-$sendCounter")
+            put("id", "ddp-$nextId")
         }
     }
 
@@ -211,14 +236,44 @@ class DDPClient {
         }
     }
 
+    private fun resetConnectHandshakeState() {
+        connectResultDelivered.set(false)
+    }
+
+    /**
+     * Delivers at most one outcome for the WebSocket connect handshake ([connect] callback).
+     * Uses [AtomicBoolean] so [onFailure], the connect-timeout runnable, and `"connected"` cannot
+     * all report conflicting results.
+     */
+    private fun tryDeliverConnectOutcome(success: Boolean, connectTimeout: Boolean = false) {
+        if (!connectResultDelivered.compareAndSet(false, true)) {
+            return
+        }
+        cancelConnectTimeout()
+        val cb = connectedCallback
+        connectedCallback = null
+        if (connectTimeout) {
+            Log.e(TAG, "Connect timeout")
+        }
+        mainHandler.post {
+            cb?.invoke(success)
+        }
+    }
+
     private fun waitForConnected(timeoutMs: Long, callback: (Boolean) -> Unit) {
         connectedCallback = callback
-        mainHandler.postDelayed({
-            val cb = connectedCallback ?: return@postDelayed
-            connectedCallback = null
-            Log.e(TAG, "Connect timeout")
-            cb(false)
-        }, timeoutMs)
+        cancelConnectTimeout()
+        val runnable = Runnable {
+            connectTimeoutRunnable = null
+            tryDeliverConnectOutcome(false, connectTimeout = true)
+        }
+        connectTimeoutRunnable = runnable
+        mainHandler.postDelayed(runnable, timeoutMs)
+    }
+
+    private fun cancelConnectTimeout() {
+        connectTimeoutRunnable?.let { mainHandler.removeCallbacks(it) }
+        connectTimeoutRunnable = null
     }
 
     private fun handleMessage(text: String) {
@@ -231,10 +286,7 @@ class DDPClient {
         when (json.optString("msg")) {
             "connected" -> {
                 isConnected = true
-                mainHandler.removeCallbacksAndMessages(null)
-                val cb = connectedCallback
-                connectedCallback = null
-                cb?.let { mainHandler.post { it(true) } }
+                tryDeliverConnectOutcome(true)
             }
 
             "ping" -> {
@@ -301,4 +353,24 @@ class DDPClient {
         val scheme = if (useSsl) "wss" else "ws"
         return "$scheme://$normalizedHost/websocket"
     }
+
+    internal fun testStartConnectTimeout(timeoutMs: Long, callback: (Boolean) -> Unit) {
+        resetConnectHandshakeState()
+        waitForConnected(timeoutMs, callback)
+    }
+
+    internal fun testDeliverRawMessage(text: String) {
+        handleMessage(text)
+    }
+
+    internal fun testDeliverConnectFailure(fromWebSocket: WebSocket? = null) {
+        mainHandler.post {
+            if (fromWebSocket != null && fromWebSocket !== this@DDPClient.webSocket) return@post
+            tryDeliverConnectOutcome(false)
+        }
+    }
+
+    internal fun testSetActiveWebSocket(ws: WebSocket?) {
+        this.webSocket = ws
+    }
 }

android/app/src/main/java/chat/rocket/reactnative/voip/VoipModule.kt

@@ -2,11 +2,11 @@ package chat.rocket.reactnative.voip
 
 import android.util.Log
 import com.facebook.react.bridge.ReactApplicationContext
-import com.facebook.react.bridge.Arguments
 import com.facebook.react.bridge.WritableMap
 import com.facebook.react.modules.core.DeviceEventManagerModule
-import java.lang.ref.WeakReference
 import chat.rocket.reactnative.networking.NativeVoipSpec
+import java.lang.ref.WeakReference
+import java.util.concurrent.atomic.AtomicReference
 
 /**
  * Native module to expose VoIP call data to JavaScript.
@@ -20,7 +20,8 @@ class VoipModule(reactContext: ReactApplicationContext) : NativeVoipSpec(reactCo
         private const val EVENT_VOIP_ACCEPT_FAILED = "VoipAcceptFailed"
 
         private var reactContextRef: WeakReference<ReactApplicationContext>? = null
-        private var initialEventsData: VoipPayload? = null
+
+        private val initialEventsData = AtomicReference<VoipPayload?>(null)
 
         /**
          * Sets the React context reference for event emission.
@@ -54,7 +55,7 @@ class VoipModule(reactContext: ReactApplicationContext) : NativeVoipSpec(reactCo
          */
         @JvmStatic
         fun storeInitialEvents(voipPayload: VoipPayload) {
-            initialEventsData = voipPayload
+            initialEventsData.set(voipPayload)
             emitInitialEventsEvent(voipPayload)
         }
 
@@ -64,7 +65,7 @@ class VoipModule(reactContext: ReactApplicationContext) : NativeVoipSpec(reactCo
         @JvmStatic
         fun storeAcceptFailureForJs(payload: VoipPayload) {
             val failed = payload.copy(voipAcceptFailed = true)
-            initialEventsData = failed
+            initialEventsData.set(failed)
             emitVoipAcceptFailedEvent(failed)
         }
 
@@ -85,7 +86,7 @@ class VoipModule(reactContext: ReactApplicationContext) : NativeVoipSpec(reactCo
         @JvmStatic
         fun clearInitialEventsInternal() {
             try {
-                initialEventsData = null
+                initialEventsData.set(null)
                 Log.d(TAG, "Cleared initial events")
             } catch (e: Exception) {
                 Log.e(TAG, "Error clearing initial events", e)
@@ -103,18 +104,22 @@ class VoipModule(reactContext: ReactApplicationContext) : NativeVoipSpec(reactCo
      * Returns null if no initial events.
      */
     override fun getInitialEvents(): WritableMap? {
-        val data = initialEventsData ?: return null
+        while (true) {
+            val data = initialEventsData.get() ?: return null
 
-        if (data.isExpired()) {
-            Log.d(TAG, "Discarding expired VoIP initial event: ${data.callId}")
-            clearInitialEventsInternal()
-            return null
-        }
+            if (data.isExpired()) {
+                Log.d(TAG, "Discarding expired VoIP initial event: ${data.callId}")
+                if (initialEventsData.compareAndSet(data, null)) {
+                    return null
+                }
+                continue
+            }
 
-        val result = data.toWritableMap()
-        clearInitialEventsInternal()
-        
-        return result
+            val result = data.toWritableMap()
+            if (initialEventsData.compareAndSet(data, null)) {
+                return result
+            }
+        }
     }
 
     /**

android/app/src/main/java/chat/rocket/reactnative/voip/VoipPayload.kt

@@ -72,6 +72,7 @@ data class VoipPayload(
             putString("avatarUrl", avatarUrl)
             putString("createdAt", createdAt)
             putInt("notificationId", notificationId)
+            putBoolean("voipAcceptFailed", voipAcceptFailed)
             // Useful flag for MainActivity to know it's handling a VoIP action
             putBoolean("voipAction", true)
         }