diff --git a/sphinx/application/data/features/feature-connect-manager/src/main/java/chat/sphinx/feature_connect_manager/ConnectManagerImpl.kt b/sphinx/application/data/features/feature-connect-manager/src/main/java/chat/sphinx/feature_connect_manager/ConnectManagerImpl.kt index f8b0ef92..8b4d7470 100644 --- a/sphinx/application/data/features/feature-connect-manager/src/main/java/chat/sphinx/feature_connect_manager/ConnectManagerImpl.kt +++ b/sphinx/application/data/features/feature-connect-manager/src/main/java/chat/sphinx/feature_connect_manager/ConnectManagerImpl.kt @@ -20,10 +20,13 @@ import com.ensarsarajcic.kotlinx.serialization.msgpack.MsgPackDynamicSerializer import io.matthewnelson.crypto_common.annotations.RawPasswordAccess import kotlinx.coroutines.CoroutineScope import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.delay import kotlinx.coroutines.flow.MutableStateFlow import kotlinx.coroutines.flow.StateFlow import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.isActive import kotlinx.coroutines.launch import okio.base64.encodeBase64 import org.cryptonode.jncryptor.AES256JNCryptorOutputStream @@ -111,6 +114,14 @@ class ConnectManagerImpl: ConnectManager() private var restoreMnemonicWords: List? = emptyList() private var inviterContact: NewContact? = null private var hasAttemptedReconnect = false + + @Volatile private var connectionInProgress = false + @Volatile private var reconnectAttemptCount = 0 + @Volatile private var lastInboundTimeMs: Long = 0L + private var connectionTimeoutJob: Job? = null + private var watchdogJob: Job? = null + private var reconnectJob: Job? = null + private var tribeServer: String? = null private var serverDefaultTribe: String? = null private var router: String? = null @@ -139,6 +150,11 @@ class ConnectManagerImpl: ConnectManager() const val TYPE_GROUP_JOIN = 14 const val TYPE_MEMBER_APPROVE = 20 const val TYPE_CONTACT_KEY_RECORD = 33 + + // MQTT reliability + const val MQTT_KEEP_ALIVE_SECS = 15 + const val CONNECTION_TIMEOUT_MS = 15_000L + const val WATCHDOG_INTERVAL_MS = 30_000L // keepAlive × 2 } private val _ownerInfoStateFlow: MutableStateFlow by lazy { @@ -184,12 +200,84 @@ class ConnectManagerImpl: ConnectManager() return _mixerIp } + private fun forceCloseMqttClient() { + val dead = mqttClient + mqttClient = null + dead?.setCallback(null) + try { dead?.disconnect() } catch (_: Exception) {} + } + + private fun startConnectionTimeoutGuard() { + connectionTimeoutJob?.cancel() + connectionTimeoutJob = userStateScope.launch { + delay(CONNECTION_TIMEOUT_MS) + if (connectionInProgress) { + Log.w("MQTT_MESSAGES", "[MQTT] Connection timed out after 15s — force-closing and retrying") + forceCloseMqttClient() + connectionInProgress = false + startReconnectWithDelay() + } + } + } + + private fun startWatchdog() { + stopWatchdog() + lastInboundTimeMs = System.currentTimeMillis() + watchdogJob = userStateScope.launch { + while (isActive) { + delay(WATCHDOG_INTERVAL_MS) + if (!isMqttConnected) break + val elapsed = System.currentTimeMillis() - lastInboundTimeMs + if (elapsed >= WATCHDOG_INTERVAL_MS) { + Log.w("MQTT_MESSAGES", "[MQTT] Watchdog: silent for 30s — forcing reconnect") + stopWatchdog() + forceCloseMqttClient() + isMqttConnected = false + connectionInProgress = false + startReconnectWithDelay() + break + } + } + } + } + + private fun stopWatchdog() { + watchdogJob?.cancel() + watchdogJob = null + } + + private fun startReconnectWithDelay() { + reconnectJob?.cancel() + val factor = Math.pow(2.0, reconnectAttemptCount.toDouble()) + val jitter = 0.75 + Math.random() * 0.5 + val delayMs = minOf((1_000.0 * factor * jitter).toLong(), 60_000L) + reconnectAttemptCount++ + Log.d("MQTT_MESSAGES", "[MQTT] Scheduling reconnect attempt $reconnectAttemptCount in ${delayMs}ms") + reconnectJob = userStateScope.launch { + delay(delayMs) + if (mixerIp != null && walletMnemonic != null) { + initializeMqttAndSubscribe( + mixerIp!!, + walletMnemonic!!, + ownerInfoStateFlow.value + ) + } + } + } + private fun connectToMQTT( serverURI: String, clientId: String, key: String, password: String, ) { + if (connectionInProgress) { + Log.d("MQTT_MESSAGES", "[MQTT] connectToMQTT skipped — connection already in progress") + return + } + connectionInProgress = true + forceCloseMqttClient() + try { mqttClient = MqttAsyncClient(serverURI, clientId, null) @@ -216,49 +304,47 @@ class ConnectManagerImpl: ConnectManager() this.userName = key this.password = password.toCharArray() this.socketFactory = sslContext?.socketFactory + this.keepAliveInterval = MQTT_KEEP_ALIVE_SECS + this.connectionTimeout = MQTT_KEEP_ALIVE_SECS } mqttClient?.connect(options, null, object : IMqttActionListener { override fun onSuccess(asyncActionToken: IMqttToken?) { + connectionTimeoutJob?.cancel() + connectionTimeoutJob = null isMqttConnected = true + connectionInProgress = false + reconnectAttemptCount = 0 hasAttemptedReconnect = false - + startWatchdog() subscribeOwnerMQTT() - notifyListeners { onNetworkStatusChange(false, isLoading = true) } - Log.d("MQTT_MESSAGES", "MQTT CONNECTED!") + Log.d("MQTT_MESSAGES", "[MQTT] CONNECTED!") } override fun onFailure(asyncActionToken: IMqttToken?, exception: Throwable?) { isMqttConnected = false - - // If it's the first time trying to connect, try to reconnect otherwise - // prevent infinite loop - if (!hasAttemptedReconnect) { - hasAttemptedReconnect = true - reconnectWithBackOff() - } -// notifyListeners { -// onConnectManagerError(ConnectManagerError.MqttConnectError(exception?.message)) -// } - Log.d("MQTT_MESSAGES", "Failed to connect to MQTT: ${exception?.message}") + connectionInProgress = false + connectionTimeoutJob?.cancel() + startReconnectWithDelay() + Log.d("MQTT_MESSAGES", "[MQTT] Failed to connect: ${exception?.message}") } }) + startConnectionTimeoutGuard() mqttClient?.setCallback(object : MqttCallback { override fun connectionLost(cause: Throwable?) { isMqttConnected = false - reconnectWithBackOff() - -// notifyListeners { -// onConnectManagerError(ConnectManagerError.MqttConnectError(cause?.message)) -// } - Log.d("MQTT_MESSAGES", "MQTT DISCONNECTED! $cause ${cause?.message}") + connectionInProgress = false + stopWatchdog() + startReconnectWithDelay() + Log.d("MQTT_MESSAGES", "[MQTT] DISCONNECTED! $cause ${cause?.message}") } override fun messageArrived(topic: String?, message: MqttMessage?) { + lastInboundTimeMs = System.currentTimeMillis() // Handle incoming messages here if (topic?.contains("/ping") == true) { @@ -280,12 +366,9 @@ class ConnectManagerImpl: ConnectManager() }) } catch (e: MqttException) { isMqttConnected = false -// notifyListeners { -// onConnectManagerError(ConnectManagerError.MqttConnectError(e.message)) -// } - - reconnectWithBackOff() - Log.d("MQTT_MESSAGES", "MQTT DISCONNECTED! exception ${e.printStackTrace()}") + connectionInProgress = false + startReconnectWithDelay() + Log.d("MQTT_MESSAGES", "[MQTT] MqttException: ${e.message}") } } @@ -1256,16 +1339,8 @@ class ConnectManagerImpl: ConnectManager() if (!isConnected()) { resetMQTT() - - if (mixerIp != null && walletMnemonic != null) { - - initializeMqttAndSubscribe( - mixerIp!!, - walletMnemonic!!, - ownerInfoStateFlow.value, - ) - } - Log.d("MQTT_MESSAGES", "onReconnectMqtt") + startReconnectWithDelay() + Log.d("MQTT_MESSAGES", "[MQTT] onReconnectMqtt") } else { getReadMessages() getMutedChats() @@ -1280,9 +1355,14 @@ class ConnectManagerImpl: ConnectManager() if (isAppFirstInit) { isAppFirstInit = false return - } else { - reconnectWithBackOff() } + // Tear down any stale in-flight state from background + reconnectAttemptCount = 0 + connectionInProgress = false + stopWatchdog() + connectionTimeoutJob?.cancel() + connectionTimeoutJob = null + reconnectWithBackOff() } override fun retrieveLspIp(): String? { @@ -2281,9 +2361,14 @@ class ConnectManagerImpl: ConnectManager() } override fun resetMQTT() { - if (mqttClient?.isConnected == true) { - mqttClient?.disconnect() - } + forceCloseMqttClient() + stopWatchdog() + connectionTimeoutJob?.cancel() + connectionTimeoutJob = null + reconnectJob?.cancel() + reconnectJob = null + isMqttConnected = false + connectionInProgress = false } private fun storeUserState(state: ByteArray) {