fix: keep Push connections consistent across concurrent subscribes and credential changes

mock-server/app/mqtt.php

@@ -9,6 +9,8 @@
  *  - CONNECT is accepted unless the credential is "deny" or "deny:<reason>", which is refused
  *    (with <reason> as the Reason String) so a test can assert a rejected connection and that
  *    its reason reaches the SDK; the projectId user property becomes the connection prefix.
+ *    A "slow:<anything>" credential is accepted after 500 ms, so a test can act while it connects.
+ *  - SUBSCRIBE to "e2e-whoami" publishes the connection's client id on "e2e-whoami".
  *  - SUBSCRIBE to "e2e-disconnect/<reason>" makes the broker DISCONNECT the client with <reason>
  *    as the Reason String.
  *  - "e2e-replay" keeps a replay position per project and client id, like the real broker does
@@ -79,6 +81,14 @@ public function onConnect(Connect $connect, Connection $connection): Connack|Aut
             return Connack::refuse(Connack::NOT_AUTHORIZED, $properties);
         }
 
+        if (\str_starts_with($credential, 'slow:')) {
+            if (\Swoole\Coroutine::getCid() > 0) {
+                \Swoole\Coroutine::sleep(0.5);
+            } else {
+                \usleep(500_000);
+            }
+        }
+
         // Mirror the real broker: derive a stable per-connection id when the client sends
         // none, so an empty client id (the SDKs' default in reliable mode) is accepted.
         $clientId = $connect->clientId;
@@ -144,6 +154,9 @@ public function onSubscribe(Subscribe $subscribe, Connection $connection): Subac
                     }
                 });
             }
+            if ($filter->topic === 'e2e-whoami') {
+                \Swoole\Timer::after(100, fn () => $connection->publish('e2e-whoami', $connection->getClientId(), qos: 0));
+            }
             if ($filter->topic === 'e2e-replay-publish') {
                 // Only the publisher's project's clients, like a real publish.
                 foreach (\array_keys($this->replayQueues) as $key) {

templates/android/library/src/main/java/io/package/services/Push.kt.twig

@@ -132,6 +132,12 @@ class Push(
     @Volatile
     private var mqtt: Mqtt5AsyncClient? = null
 
+    private var connectedConfig: PushConfig? = null
+    private val connectLock = Any()
+
+    @Volatile
+    private var connectionEpoch = 0
+
     // Local id -> subscription (filter + callback + per-sub options). The id is never sent to
     // the broker; it only lets the same topic carry more than one callback.
     private val subscriptions = ConcurrentHashMap<String, Sub>()
@@ -159,7 +165,10 @@ class Push(
         var background: Boolean,
         var title: String?,
         var retry: Boolean,
-    )
+    ) {
+        @Volatile
+        var client: Mqtt5AsyncClient? = null
+    }
 
     // The broker subscription for a filter uses the highest QoS any subscription on it wants,
     // so a QoS-0 subscription never downgrades a QoS-1 one sharing the filter.
@@ -427,15 +436,28 @@ class Push(
 
     // In-process host: connect and register this call's new filters, blocking on each SUBACK.
     private fun subscribeInProcess(ids: List<String>) {
-        val client = connect()
-        for (id in ids) {
-            val sub = subscriptions[id] ?: continue
-            client.subscribeWith()
-                .topicFilter(sub.filter)
-                .qos(effectiveQos(sub.filter))
-                .callback { publish -> sub.callback(publish.toPushMessage()) }
-                .send()
-                .get()
+        while (true) {
+            val (client, epoch) = synchronized(connectLock) { connect() to connectionEpoch }
+            try {
+                for (id in ids) {
+                    val sub = subscriptions[id] ?: continue
+                    if (sub.client === client) {
+                        continue
+                    }
+                    client.subscribeWith()
+                        .topicFilter(sub.filter)
+                        .qos(effectiveQos(sub.filter))
+                        .callback { publish -> sub.callback(publish.toPushMessage()) }
+                        .send()
+                        .get()
+                    sub.client = client
+                }
+                return
+            } catch (e: Exception) {
+                if (epoch == connectionEpoch) {
+                    throw e
+                }
+            }
         }
     }
 
@@ -446,6 +468,7 @@ class Push(
             mqtt?.disconnect()
             mqtt = null
         }
+        subscriptions.values.forEach { it.client = null }
         PushBackground.errorHandler = { deliverBackgroundError(it) }
         PushBackground.host(appContext, buildConfig(), currentListeners())
         backgroundActive = true
@@ -456,12 +479,16 @@ class Push(
     private fun ensureInProcessSubscribed() {
         val client = connect()
         for (sub in subscriptions.values) {
+            if (sub.client === client) {
+                continue
+            }
             client.subscribeWith()
                 .topicFilter(sub.filter)
                 .qos(effectiveQos(sub.filter))
                 .callback { publish -> sub.callback(publish.toPushMessage()) }
                 .send()
                 .reportFailure(client)
+            sub.client = client
         }
     }
 
@@ -478,6 +505,7 @@ class Push(
                 .callback { publish -> sub.callback(publish.toPushMessage()) }
                 .send()
                 .reportFailure(client)
+            sub.client = client
         }
     }
 
@@ -593,15 +621,20 @@ class Push(
         }
     }
 
-    private fun connect(): Mqtt5AsyncClient {
+    private fun connect(): Mqtt5AsyncClient = synchronized(connectLock) {
+        val config = buildConfig()
         // Reuse the existing client even while it is mid automatic-reconnect. Building a second
         // client here would give two clients the same stable id, so they would fight over the
         // broker session and callbacks could stay attached to the abandoned one.
         mqtt?.let { existing ->
-            return existing
+            if (config == connectedConfig) {
+                return existing
+            }
+            connectionEpoch++
+            existing.disconnect()
+            mqtt = null
         }
 
-        val config = buildConfig()
         // The listeners surface the lifecycle to onOpen/onClose and every connection error to
         // onError, except a refused initial CONNECT: pushConnect throws that one, and the
         // failing call reports it.
@@ -612,15 +645,33 @@ class Push(
             onError = { reportAsync(it) },
         )
         mqtt = client
+        connectedConfig = config
 
         try {
             client.pushConnect(config)
         } catch (e: Exception) {
-            close()
+            client.disconnect()
+            if (mqtt === client) {
+                mqtt = null
+                connectedConfig = null
+            }
             throw e
         }
 
-        return client
+        for (sub in subscriptions.values) {
+            if (sub.client == null || sub.client === client) {
+                continue
+            }
+            client.subscribeWith()
+                .topicFilter(sub.filter)
+                .qos(effectiveQos(sub.filter))
+                .callback { publish -> sub.callback(publish.toPushMessage()) }
+                .send()
+                .reportFailure(client)
+            sub.client = client
+        }
+
+        client
     }
 
     private fun buildConfig(): PushConfig {

templates/apple/Sources/Services/Push.swift.twig

@@ -100,6 +100,10 @@ open class Push: Service {
 
     private var mqtt: MQTTClient?
     private var connecting: Task<MQTTClient, Swift.Error>?
+    private var connectedKey = ""
+    private var connectionEpoch = 0
+    private var resubscribeOnConnect = false
+    private var retired = Set<ObjectIdentifier>()
 
     private var onOpenCb: (() -> Void)?
     private var onCloseCb: (() -> Void)?
@@ -248,7 +252,13 @@ open class Push: Service {
             requestNotificationAuthorization()
         }
 
-        let mqtt = try await connect()
+        let mqtt: MQTTClient
+        do {
+            mqtt = try await connect()
+        } catch is SupersededError {
+            return try await subscribe(topics.map { $0 as TopicValue }, background: background, title: title, retry: retry, callback: callback)
+        }
+        let epoch = sync.sync { connectionEpoch }
 
         var ids: [String] = []
         for topic in topics {
@@ -263,6 +273,16 @@ open class Push: Service {
             }
             try await brokerSubscribe(mqtt, infos)
         } catch {
+            let superseded = sync.sync { () -> Bool in
+                guard connectionEpoch != epoch else { return false }
+                for id in ids {
+                    subscriptions.removeValue(forKey: id)
+                }
+                return true
+            }
+            if superseded {
+                return try await subscribe(topics.map { $0 as TopicValue }, background: background, title: title, retry: retry, callback: callback)
+            }
             // The SUBSCRIBE failed; roll this call's registrations back (and drop any broker
             // filter no other local subscriber owns) so a failed subscribe leaks no callback.
             let filters = sync.sync { () -> [String] in
@@ -405,6 +425,19 @@ open class Push: Service {
     }
 
     private func connect() async throws -> MQTTClient {
+        let key = credentialKey()
+        let superseded = sync.sync { () -> MQTTClient? in
+            guard self.mqtt != nil || self.connecting != nil, self.connectedKey != key else { return nil }
+            self.connectionEpoch += 1
+            self.connecting = nil
+            let existing = self.mqtt
+            self.mqtt = nil
+            self.resubscribeOnConnect = !self.subscriptions.isEmpty
+            return existing
+        }
+        if let superseded {
+            retire(superseded)
+        }
         if let existing = sync.sync(execute: { self.mqtt }) {
             return existing
         }
@@ -414,16 +447,32 @@ open class Push: Service {
             if let inflight = self.connecting {
                 return inflight
             }
+            self.connectedKey = key
+            let epoch = self.connectionEpoch
             let created = Task { () throws -> MQTTClient in
                 do {
                     let client = try await self.open()
-                    self.sync.sync {
+                    let current = self.sync.sync { () -> Bool in
+                        guard self.connectionEpoch == epoch else { return false }
                         self.mqtt = client
                         self.connecting = nil
+                        return true
                     }
+                    guard current else {
+                        self.retire(client)
+                        throw SupersededError()
+                    }
+                    self.resubscribeIfSuperseded(client)
                     return client
                 } catch {
-                    self.sync.sync { self.connecting = nil }
+                    let current = self.sync.sync { () -> Bool in
+                        guard self.connectionEpoch == epoch else { return false }
+                        self.connecting = nil
+                        return true
+                    }
+                    if !current {
+                        throw SupersededError()
+                    }
                     throw error
                 }
             }
@@ -433,6 +482,36 @@ open class Push: Service {
         return try await task.value
     }
 
+    private func credentialKey() -> String {
+        let jwt = client.config["jwt"] ?? ""
+        let session = client.config["session"] ?? ""
+        let credential = jwt.isEmpty ? "appwrite-session|\(session)" : "appwrite-jwt|\(jwt)"
+        return [host, String(port), client.config["project"] ?? "", resolvedClientId, credential].joined(separator: "|")
+    }
+
+    private func retire(_ client: MQTTClient) {
+        sync.sync { _ = retired.insert(ObjectIdentifier(client)) }
+        DispatchQueue.global().async {
+            try? client.disconnect().wait()
+            try? client.syncShutdownGracefully()
+        }
+    }
+
+    private func resubscribeIfSuperseded(_ client: MQTTClient) {
+        let infos = sync.sync { () -> [MQTTSubscribeInfoV5] in
+            guard resubscribeOnConnect else { return [] }
+            resubscribeOnConnect = false
+            return Set(subscriptions.values.map { $0.filter }).map {
+                MQTTSubscribeInfoV5(topicFilter: $0, qos: effectiveQos($0))
+            }
+        }
+        if !infos.isEmpty {
+            inBackground { [weak self] in
+                try await self?.brokerSubscribe(client, infos)
+            }
+        }
+    }
+
     // Read the credential off the client and build the CONNECT properties (enhanced auth plus
     // the project id). Rebuilt on every connect so a reconnect picks up a refreshed credential.
     private func connectProperties() throws -> MQTTProperties {
@@ -522,7 +601,10 @@ open class Push: Service {
 
         // Surface the connection lifecycle to the caller's onClose hook, and recover from an
         // unexpected drop (a deliberate close() is skipped via the `closing` flag).
-        mqtt.addCloseListener(named: "{{ spec.info.title | caseLower }}-push") { [weak self] result in
+        mqtt.addCloseListener(named: "{{ spec.info.title | caseLower }}-push") { [weak self, weak mqtt] result in
+            if let self, let mqtt, self.sync.sync(execute: { self.retired.remove(ObjectIdentifier(mqtt)) != nil }) {
+                return
+            }
             if case .failure(let error) = result {
                 self?.surface(Push.describe(error))
             } else if let self, !self.sync.sync(execute: { self.closing }) {
@@ -783,3 +865,5 @@ private func matches(filter: String, topic: String) -> Bool {
     }
     return filterParts.count == topicParts.count
 }
+
+private struct SupersededError: Swift.Error {}