diff --git a/mock-server/app/mqtt.php b/mock-server/app/mqtt.php index 29d8191fde..e4ffa532d2 100644 --- a/mock-server/app/mqtt.php +++ b/mock-server/app/mqtt.php @@ -9,6 +9,8 @@ * - CONNECT is accepted unless the credential is "deny" or "deny:", which is refused * (with 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:" 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/" makes the broker DISCONNECT the client with * 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) { diff --git a/templates/android/library/src/main/java/io/package/services/Push.kt.twig b/templates/android/library/src/main/java/io/package/services/Push.kt.twig index 9dc5e79c92..fe0a715fa5 100644 --- a/templates/android/library/src/main/java/io/package/services/Push.kt.twig +++ b/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() @@ -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) { - 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 { diff --git a/templates/apple/Sources/Services/Push.swift.twig b/templates/apple/Sources/Services/Push.swift.twig index a110bb1d50..d177d5ad9e 100644 --- a/templates/apple/Sources/Services/Push.swift.twig +++ b/templates/apple/Sources/Services/Push.swift.twig @@ -100,6 +100,10 @@ open class Push: Service { private var mqtt: MQTTClient? private var connecting: Task? + private var connectedKey = "" + private var connectionEpoch = 0 + private var resubscribeOnConnect = false + private var retired = Set() 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 {} diff --git a/templates/flutter/lib/src/mqtt_browser.dart.twig b/templates/flutter/lib/src/mqtt_browser.dart.twig index 42eddce3fe..2cd292d6f2 100644 --- a/templates/flutter/lib/src/mqtt_browser.dart.twig +++ b/templates/flutter/lib/src/mqtt_browser.dart.twig @@ -62,6 +62,10 @@ class PushWeb implements Push { MqttBrowserClient? _mqtt; Future? _connecting; + int _connectionEpoch = 0; + Completer? _openCancel; + String _connectedKey = ''; + bool _resubscribeOnConnect = false; void Function()? _onOpen; void Function()? _onClose; @@ -253,9 +257,19 @@ class PushWeb implements Push { topicList = _topicList(client, topics); mqtt = await _connect(); } catch (e) { + if (e is _Superseded) { + return subscribe( + topics, + callback, + background: background, + title: title, + retry: retry, + ); + } _notify(e); rethrow; } + final epoch = _connectionEpoch; // The own-user topic (topics == null) defaults to background delivery, like FCM/APNs. final wantsBackground = background ?? topics == null; @@ -336,6 +350,18 @@ class PushWeb implements Push { } await Future.wait(acks); } catch (e) { + if (epoch != _connectionEpoch) { + for (final id in ids) { + _subscriptions.remove(id); + } + return subscribe( + topics, + callback, + background: background, + title: title, + retry: retry, + ); + } unsubscribe(); _notify(e); rethrow; @@ -406,6 +432,24 @@ class PushWeb implements Push { } Future _connect() { + if ((_mqtt != null || _connecting != null) && + _connectedKey != _credentialKey) { + _connectionEpoch++; + _connecting = null; + _openCancel?.complete(); + _openCancel = null; + for (final waiting in _subAcks.values) { + for (final completer in waiting) { + if (!completer.isCompleted) { + completer.completeError(const _Superseded()); + } + } + } + _subAcks.clear(); + _mqtt?.disconnect(); + _mqtt = null; + _resubscribeOnConnect = _subscriptions.isNotEmpty; + } final mqtt = _mqtt; if (mqtt != null && mqtt.connectionStatus?.state == MqttConnectionState.connected) { @@ -417,32 +461,54 @@ class PushWeb implements Push { _connecting = connecting; // A failed open must not poison future attempts — clear the cached in-flight // future so a later subscribe/publish can retry. + final pending = connecting; unawaited( - connecting.catchError((Object _) { - _connecting = null; + pending.catchError((Object _) { + if (identical(_connecting, pending)) { + _connecting = null; + } }), ); } return connecting.then((_) => _mqtt!); } - Future _open() async { - // Read the credential set on the client (via Client.setJWT / setSession). + // The credential set on the client (via Client.setJWT / setSession), as (method, credential). + (String, String) _credential() { final jwt = client.config['jwt'] ?? client.config['jWT']; final session = client.config['session']; - final String authMethod; - final String credential; if (jwt != null && jwt.isNotEmpty) { - authMethod = 'appwrite-jwt'; - credential = jwt; - } else if (session != null && session.isNotEmpty) { - authMethod = 'appwrite-session'; - credential = session; - } else { - throw {{spec.info.title | caseUcfirst}}Exception( - 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', - ); + return ('appwrite-jwt', jwt); } + if (session != null && session.isNotEmpty) { + return ('appwrite-session', session); + } + throw {{spec.info.title | caseUcfirst}}Exception( + 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', + ); + } + + String get _credentialKey { + try { + final (authMethod, credential) = _credential(); + return [ + _host, + _port, + _path, + client.config['project'] ?? '', + _resolvedClientId, + authMethod, + credential, + ].join('|'); + } catch (_) { + return ''; + } + } + + Future _open() async { + final (authMethod, credential) = _credential(); + _connectedKey = _credentialKey; + final epoch = _connectionEpoch; final project = client.config['project'] ?? ''; // An empty client id makes the broker derive a stable id server-side (keyed on the @@ -497,15 +563,23 @@ class PushWeb implements Push { _mqtt = mqtt; - final failure = await _connectOrFail(mqtt, (error) { + final cancel = Completer(); + _openCancel = cancel; + final failure = await _connectOrFail(mqtt, cancel.future, (error) { + if (epoch != _connectionEpoch) { + return; + } // Let go of the client first: it is still being wound down, and a close() from // onError must not touch it. if (identical(_mqtt, mqtt)) { _mqtt = null; } - close(); _notify(error); }); + if (epoch != _connectionEpoch) { + mqtt.disconnect(); + throw const _Superseded(); + } if (failure != null) { throw failure; } @@ -513,6 +587,19 @@ class PushWeb implements Push { // Attach the delivery listener only after CONNACK — the `updates` stream is not // available until the connection is established. mqtt.updates.listen(_onData); + + if (_resubscribeOnConnect) { + _resubscribeOnConnect = false; + final filters = _subscriptions.values.map((s) => s.filter).toSet(); + for (final filter in filters) { + mqtt.subscribeWithSubscriptionList([ + MqttSubscription.withMaximumQos( + MqttSubscriptionTopic(filter), + _effectiveQos(filter), + ), + ]); + } + } } void _onData(List> events) { @@ -587,6 +674,7 @@ class PushWeb implements Push { /// watched for the refusal as well, and the pending connect() is then let go. Future _connectOrFail( MqttClient mqtt, + Future cancelled, void Function(Object error) failed, ) async { final connecting = mqtt.connect(); @@ -598,7 +686,7 @@ Future _connectOrFail( }); Object? failure; try { - await Future.any([connecting, refused.future]); + await Future.any([connecting, refused.future, cancelled]); } catch (e) { failure = e; } finally { @@ -810,3 +898,7 @@ String? _stringField(String encoded, String key) { return null; } } + +class _Superseded implements Exception { + const _Superseded(); +} diff --git a/templates/flutter/lib/src/mqtt_io.dart.twig b/templates/flutter/lib/src/mqtt_io.dart.twig index 993c3c7930..a44c156be0 100644 --- a/templates/flutter/lib/src/mqtt_io.dart.twig +++ b/templates/flutter/lib/src/mqtt_io.dart.twig @@ -84,8 +84,12 @@ class PushIO implements Push { final PushNative? _native = PushNative.instance; // Whether the subscriptions currently live on the native plugin's connection. bool _nativeActive = false; - // Bumped when the subscriptions move to the native host, superseding an _open in flight. + // Bumped when the subscriptions move to the native host, or the credential changes, + // superseding an _open in flight. int _connectionEpoch = 0; + Completer? _openCancel; + String _connectedKey = ''; + bool _resubscribeOnConnect = false; StreamSubscription>? _nativeListener; // Every Push on Android shares the one native host (one connection per client id), which hosts @@ -256,7 +260,7 @@ class PushIO implements Push { // Every error the service hits goes through here and reaches onError. It is never thrown // from here, so an error outside any call cannot crash the app. void _report(Object error) { - if (error is _MovedToNative) { + if (error is _Superseded) { // Internal: a connection superseded by the native host, not a failure. return; } @@ -341,12 +345,18 @@ class PushIO implements Push { // off the in-process one if it was up) and covers the whole topic union. await _hostInIsolate(); } else { - try { - await _subscribeInProcess(ids); - } on _MovedToNative { - // Another Push moved this one's subscriptions to the native host meanwhile: this - // subscribe goes there too. - await _hostNative(); + while (true) { + try { + await _subscribeInProcess(ids); + break; + } on _Superseded { + // Another Push moved this one's subscriptions to the native host meanwhile: this + // subscribe goes there too. + if (_nativeActive) { + await _hostNative(); + break; + } + } } } } catch (e) { @@ -451,20 +461,7 @@ class PushIO implements Push { void _joinNative(PushNative native) { // An _open still in flight sees this and closes what it opened, and a subscribe still waiting // for its SUBACK continues on the native host. - _connectionEpoch++; - _connecting = null; - for (final waiting in _subAcks.values) { - for (final completer in waiting) { - if (!completer.isCompleted) { - completer.completeError(const _MovedToNative()); - } - } - } - _subAcks.clear(); - if (_mqtt != null) { - _mqtt?.disconnect(); - _mqtt = null; - } + _dropConnection(); _nativeListener ??= native.listen( onMessage: (message) { final sub = _subscriptions[message.id]; @@ -491,6 +488,25 @@ class PushIO implements Push { _nativeActive = true; } + void _dropConnection() { + _connectionEpoch++; + _connecting = null; + _openCancel?.complete(); + _openCancel = null; + for (final waiting in _subAcks.values) { + for (final completer in waiting) { + if (!completer.isCompleted) { + completer.completeError(const _Superseded()); + } + } + } + _subAcks.clear(); + if (_mqtt != null) { + _mqtt?.disconnect(); + _mqtt = null; + } + } + // Move this Push's subscriptions back onto its own connection. void _leaveToInProcess() { _nativeHosts.remove(this); @@ -863,6 +879,11 @@ class PushIO implements Push { } Future _connect() { + if ((_mqtt != null || _connecting != null) && + _connectedKey != _credentialKey) { + _dropConnection(); + _resubscribeOnConnect = _subscriptions.isNotEmpty; + } final mqtt = _mqtt; if (mqtt != null && mqtt.connectionStatus?.state == MqttConnectionState.connected) { @@ -874,9 +895,12 @@ class PushIO implements Push { _connecting = connecting; // A failed open must not poison future attempts — clear the cached in-flight // future so a later subscribe/publish can retry. + final pending = connecting; unawaited( - connecting.catchError((Object _) { - _connecting = null; + pending.catchError((Object _) { + if (identical(_connecting, pending)) { + _connecting = null; + } }), ); } @@ -900,6 +924,7 @@ class PushIO implements Push { Future _open() async { final (authMethod, credential) = _credential(); + _connectedKey = _credentialKey; final epoch = _connectionEpoch; final project = client.config['project'] ?? ''; @@ -917,7 +942,7 @@ class PushIO implements Push { // The subscriptions moved to the native host while this was waiting: its connection would // share the host's client id and take over the broker session. if (epoch != _connectionEpoch) { - throw const _MovedToNative(); + throw const _Superseded(); } final mqtt = MqttServerClient.withPort(_host, clientId, _port); @@ -965,7 +990,9 @@ class PushIO implements Push { _mqtt = mqtt; - final failure = await _connectOrFail(mqtt, (error) { + final cancel = Completer(); + _openCancel = cancel; + final failure = await _connectOrFail(mqtt, cancel.future, (error) { // Superseded by the native host, which closed this connection on purpose. if (epoch != _connectionEpoch) { return; @@ -975,12 +1002,11 @@ class PushIO implements Push { if (identical(_mqtt, mqtt)) { _mqtt = null; } - _teardown(); _notify(error); }); if (epoch != _connectionEpoch) { mqtt.disconnect(); - throw const _MovedToNative(); + throw const _Superseded(); } if (failure != null) { throw failure; @@ -989,6 +1015,18 @@ class PushIO implements Push { // Attach the delivery listener only after CONNACK — the `updates` stream is not // available until the connection is established. mqtt.updates.listen(_onData); + + if (_resubscribeOnConnect) { + _resubscribeOnConnect = false; + for (final filter in _backgroundTopics) { + mqtt.subscribeWithSubscriptionList([ + MqttSubscription.withMaximumQos( + MqttSubscriptionTopic(filter), + _effectiveQos(filter), + ), + ]); + } + } } void _onData(List> events) { @@ -1057,6 +1095,7 @@ class PushIO implements Push { /// watched for the refusal as well, and the pending connect() is then let go. Future _connectOrFail( MqttClient mqtt, + Future cancelled, void Function(Object error) failed, ) async { final connecting = mqtt.connect(); @@ -1068,7 +1107,7 @@ Future _connectOrFail( }); Object? failure; try { - await Future.any([connecting, refused.future]); + await Future.any([connecting, refused.future, cancelled]); } catch (e) { failure = e; } finally { @@ -1287,7 +1326,8 @@ const String _backgroundDisplaced = 'Background delivery stopped: a Push with another credential started background delivery, ' 'and the device hosts one. These subscriptions still deliver while the app runs.'; -// An in-process connect superseded by moving the subscriptions to the native host. -class _MovedToNative implements Exception { - const _MovedToNative(); +// An in-process connect superseded by moving the subscriptions to the native host, or by a +// connect with another credential. +class _Superseded implements Exception { + const _Superseded(); } diff --git a/templates/unity/Assets/Runtime/Push.cs.twig b/templates/unity/Assets/Runtime/Push.cs.twig index 878954d01d..d5757b0413 100644 --- a/templates/unity/Assets/Runtime/Push.cs.twig +++ b/templates/unity/Assets/Runtime/Push.cs.twig @@ -53,6 +53,10 @@ namespace {{ spec.info.title | caseUcfirst }} private bool _reconnecting; // The last connect options, reused when reconnecting after an unexpected drop. private MqttClientOptions _options; + private Task _connecting; + private string _connectedKey = ""; + private int _connectionEpoch; + private bool _resubscribeOnConnect; private Action _onOpen; private Action _onClose; @@ -61,10 +65,9 @@ namespace {{ spec.info.title | caseUcfirst }} // Fixed keep-alive (PINGREQ interval) in seconds — not exposed as an option. private const int KeepAliveSeconds = 60; - // Resolved once, then reused: a reconnect rebuilds the MQTT client, and if each rebuild - // drew a new random id the broker would treat it as a new session and drop this client's - // replay cursor. A stable id from Client.SetPushClientId() resumes replay across - // restarts; otherwise one is generated. Reusing one id keeps it stable. + // The client id of the last connect: Client.SetPushClientId()'s id, else empty so the + // broker derives a stable one. Read again on every connect, so a changed id applies to + // the next connection. private string _resolvedClientId; // Local id -> (topic filter, callback, retry). The id is never sent to the broker; @@ -439,11 +442,16 @@ namespace {{ spec.info.title | caseUcfirst }} { client = await Connect(); } + catch (SupersededException) + { + return await Subscribe(topicList, callback, retry); + } catch (Exception e) { NotifyError(e); throw; } + var epoch = _connectionEpoch; var ids = new List(); foreach (var topic in topicList) @@ -556,6 +564,19 @@ namespace {{ spec.info.title | caseUcfirst }} } catch (Exception e) { + if (epoch != _connectionEpoch) + { + lock (_subscriptions) + { + foreach (var id in ids) + { + _subscriptions.Remove(id); + } + } + + return await Subscribe(topicList, callback, retry); + } + unsubscribe(); NotifyError(e); throw; @@ -610,6 +631,7 @@ namespace {{ spec.info.title | caseUcfirst }} { // Mark this a deliberate close so the disconnect handler does not reconnect. _closing = true; + _connecting = null; if (_mqtt != null) { _ = _mqtt.DisconnectAsync(); @@ -623,32 +645,75 @@ namespace {{ spec.info.title | caseUcfirst }} } } - private async Task Connect() + private Task Connect() { + if ((_mqtt != null || _connecting != null) && _connectedKey != CredentialKey()) + { + _connectionEpoch++; + _connecting = null; + var previous = _mqtt; + _mqtt = null; + if (previous != null) + { + _ = previous.DisconnectAsync(); + previous.Dispose(); + } + + lock (_subscriptions) + { + _resubscribeOnConnect = _subscriptions.Count > 0; + } + } + + if (_connecting != null && !_connecting.IsCompleted) + { + return _connecting; + } + if (_mqtt != null && _mqtt.IsConnected) { - return _mqtt; + return Task.FromResult(_mqtt); } - // Read the credential set on the client (via Client.SetJWT / SetSession). + _connecting = Open(); + return _connecting; + } + + private (string AuthMethod, string Credential) Credential() + { var jwt = _client.GetJWT(); - var session = _client.GetSession(); - string authMethod; - string credential; if (!string.IsNullOrEmpty(jwt)) { - authMethod = "appwrite-jwt"; - credential = jwt; + return ("appwrite-jwt", jwt); } - else if (!string.IsNullOrEmpty(session)) + + var session = _client.GetSession(); + if (!string.IsNullOrEmpty(session)) { - authMethod = "appwrite-session"; - credential = session; + return ("appwrite-session", session); } - else + + throw new {{ spec.info.title | caseUcfirst }}Exception("No credential set on the client; call Client.SetJWT() or Client.SetSession() first."); + } + + private string CredentialKey() + { + try { - throw new {{ spec.info.title | caseUcfirst }}Exception("No credential set on the client; call Client.SetJWT() or Client.SetSession() first."); + var (authMethod, credential) = Credential(); + return string.Join("|", PushEndpoint, _client.Config.GetValueOrDefault("project") ?? "", _client.Config.GetValueOrDefault("pushClientId") ?? "", authMethod, credential); } + catch (Exception) + { + return ""; + } + } + + private async Task Open() + { + var (authMethod, credential) = Credential(); + _connectedKey = CredentialKey(); + var epoch = _connectionEpoch; // When no explicit push endpoint is set, default to a `push.` host on the regular // endpoint (mirroring how realtime derives from the endpoint, on the push subdomain). @@ -662,9 +727,9 @@ namespace {{ spec.info.title | caseUcfirst }} // Empty when the app did not set one: the broker derives a stable id server-side // (keyed on the credential/project) and keys its offline-replay cursor on it, so // replay resumes across restarts; Client.SetPushClientId() overrides with an - // explicit id. Resolved once so a reconnect rebuild reuses the same id. + // explicit id. var pushClientId = _client.Config.GetValueOrDefault("pushClientId"); - _resolvedClientId ??= string.IsNullOrEmpty(pushClientId) ? "" : pushClientId; + _resolvedClientId = string.IsNullOrEmpty(pushClientId) ? "" : pushClientId; var clientId = _resolvedClientId; var optionsBuilder = new MqttClientOptionsBuilder() @@ -701,6 +766,11 @@ namespace {{ spec.info.title | caseUcfirst }} }; client.DisconnectedAsync += e => { + if (_mqtt != client) + { + return Task.CompletedTask; + } + var error = DisconnectError(e); _mainThread.Enqueue(() => _onClose?.Invoke()); if (error != null) @@ -732,19 +802,63 @@ namespace {{ spec.info.title | caseUcfirst }} } catch (Exception e) { - Close(); + if (epoch != _connectionEpoch) + { + throw new SupersededException(); + } + + DropClient(client); throw ConnectError(e); } + if (epoch != _connectionEpoch) + { + DropClient(client); + throw new SupersededException(); + } + if (result.ResultCode != MqttClientConnectResultCode.Success) { - Close(); + DropClient(client); throw ConnectRefused(result); } + if (_resubscribeOnConnect) + { + _resubscribeOnConnect = false; + var rebuild = new MqttClientSubscribeOptionsBuilder(); + var any = false; + lock (_subscriptions) + { + foreach (var filter in new HashSet(_subscriptions.Values.Select(entry => entry.Filter))) + { + var topic = filter; + rebuild = rebuild.WithTopicFilter(f => f + .WithTopic(topic) + .WithQualityOfServiceLevel(EffectiveQos(topic))); + any = true; + } + } + + if (any) + { + _ = SubscribeInBackground(client, rebuild.Build()); + } + } + return client; } + private void DropClient(IMqttClient client) + { + if (_mqtt == client) + { + _mqtt = null; + } + + client.Dispose(); + } + // The error for a failed connect attempt: a refused CONNECT carries the broker's reason. private static Exception ConnectError(Exception error) { @@ -972,6 +1086,10 @@ namespace {{ spec.info.title | caseUcfirst }} public override string ToString() => _value; } + internal class SupersededException : Exception + { + } + /// /// Handle for a live subscription created with . /// Unlike the realtime service's bare unsubscribe callable, this is returned as a handle for diff --git a/templates/web/src/client.ts.twig b/templates/web/src/client.ts.twig index 5d13ac602c..e173886357 100644 --- a/templates/web/src/client.ts.twig +++ b/templates/web/src/client.ts.twig @@ -501,7 +501,10 @@ class Client { * * A stable client id for the {{ spec.info.title | caseUcfirst }}Push service. The broker keys its * offline-replay cursor on this id, so pass a stable value to resume replay across - * reloads/restarts. Defaults to a per-connection id when unset. + * reloads/restarts. Defaults, when unset, to one per user and tab: stable across reloads of + * the tab, different in every other tab, and distinct for each connection open at the same + * time, so no two connections take over each other's session. An explicit id shared by + * several tabs or connections makes them do that. * * @param {string} pushClientId * diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 9fcbdc133c..35253fb53b 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -51,6 +51,8 @@ export interface PushSubscription { type AuthMethod = 'appwrite-jwt' | 'appwrite-session' | ''; +const TAB_ID_KEY = '{{ spec.info.title | caseLower }}-push-client-id'; + interface Transport { connect: typeof import('mqtt').connect; Buffer: typeof import('buffer').Buffer; @@ -123,10 +125,18 @@ export class Push extends Service { // Local id -> subscription state. The id is never sent to the broker; it only lets // the same topic carry more than one callback (each with its own options). /** - * The client id when the credential names no user (and none was set): random, but kept for - * this instance so reconnects resume the same broker session. + * This tab's part of the default client id when the tab has none yet (see claimTabId): + * random, but kept for this instance so reconnects resume the same broker session. */ private readonly fallbackClientId = randomId(); + /** The default client id claimed for this instance's connection, released on close(). */ + private tabId = ''; + /** In-flight claim of {@link tabId}, memoized per generation: a credential switch (same + * generation) shares the one claim; a close/reopen (new generation) chains a fresh claim. */ + private tabIdClaim: Promise | null = null; + private tabIdClaimGeneration = -1; + private connectedKey = ''; + private connectionEpoch = 0; private readonly subscriptions = new Map< string, @@ -308,7 +318,20 @@ export class Push extends Service { this.requestNotificationPermission(); } - const client = await this.connect(); + let client: MqttClient; + try { + client = await this.connect(); + } catch (err) { + if (err instanceof SupersededError) { + return this.subscribeWith( + topicsOrCallback, + callbackOrOptions, + topicOptions, + ); + } + throw err; + } + const epoch = this.connectionEpoch; const ids: string[] = []; const acks: Promise[] = []; @@ -378,6 +401,16 @@ export class Push extends Service { try { await Promise.all(acks); } catch (err) { + if (epoch !== this.connectionEpoch) { + for (const id of ids) { + this.subscriptions.delete(id); + } + return this.subscribeWith( + topicsOrCallback, + callbackOrOptions, + topicOptions, + ); + } unsubscribe(); throw err; } @@ -443,7 +476,13 @@ export class Push extends Service { this.mqtt.end(true); this.mqtt = null; } + if (this.tabId !== '') { + releaseTabId(this.tabId); + this.tabId = ''; + } this.connecting = null; + // Bumping the generation ends the current claim: one still in flight releases its id when + // it resolves, and the next open() chains a fresh claim after it (see claimClientId). this.generation++; this.everConnected = false; this.subscriptions.clear(); @@ -478,20 +517,66 @@ export class Push extends Service { // --- internals --- private connect(): Promise { - if (this.mqtt) { + if ( + (this.mqtt || this.connecting) && + this.connectedKey !== this.credentialKey() + ) { + this.connectionEpoch++; + this.mqtt?.end(true); + this.mqtt = null; + this.connecting = null; + this.everConnected = this.subscriptions.size > 0; + } + if (this.mqtt && !this.connecting) { return Promise.resolve(this.mqtt); } if (!this.connecting) { - this.connecting = this.open(); + const connecting = this.open(); + this.connecting = connecting; // A failed open must not poison future attempts — clear the cached // in-flight promise so a later subscribe/publish can retry. - this.connecting.catch(() => { - this.connecting = null; + connecting.catch(() => { + if (this.connecting === connecting) { + this.connecting = null; + } }); } return this.connecting.then(() => this.mqtt!); } + private credential(): { authMethod: AuthMethod; credential: string } { + if (this.client.config.jwt) { + return { + authMethod: 'appwrite-jwt', + credential: this.client.config.jwt, + }; + } + if (this.client.config.session) { + return { + authMethod: 'appwrite-session', + credential: this.client.config.session, + }; + } + throw new Error( + 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', + ); + } + + private credentialKey(): string { + try { + const { authMethod, credential } = this.credential(); + return [ + this.endpointUrl(), + this.client.config.project ?? '', + this.client.config.pushClientId ?? '', + authMethod, + credential, + ].join('|'); + } catch { + return ''; + } + } + private endpointUrl(): string { // The push endpoint set on the client (Client.setPushEndpoint) wins. if (this.client.config.endpointPush !== '') { @@ -510,41 +595,79 @@ export class Push extends Service { : `${wsEndpoint.slice(0, at + separator.length)}push.${wsEndpoint.slice(at + separator.length)}`; } - private async open(): Promise { - // Read the credential set on the client (via Client.setJWT / setSession). - let authMethod: AuthMethod; - let credential: string; - if (this.client.config.jwt) { - authMethod = 'appwrite-jwt'; - credential = this.client.config.jwt; - } else if (this.client.config.session) { - authMethod = 'appwrite-session'; - credential = this.client.config.session; - } else { - throw new Error( - 'No credential set on the client; call Client.setJWT() or Client.setSession() first.', - ); + /** + * Claim this connection's default client id, memoized per generation. A credential switch (same + * generation) shares the one claim, so concurrent open() calls never each claim. A close/reopen + * (a new generation) chains a fresh claim after the previous one has settled and released its + * id, so the reopened connection can take the tab's stored id back rather than a fallback. A + * claim that resolves after its generation has ended releases its id instead of keeping it, so + * a stale open connects with nothing reserved and the stored id is never stranded. + */ + private claimClientId(): Promise { + const generation = this.generation; + if ( + this.tabIdClaim === null || + this.tabIdClaimGeneration !== generation + ) { + const previous = this.tabIdClaim ?? Promise.resolve(''); + this.tabIdClaimGeneration = generation; + const claim = previous + .catch(() => '') + .then(() => claimTabId(this.fallbackClientId)) + .then((id) => { + if (this.generation !== generation) { + releaseTabId(id); + return ''; + } + this.tabId = id; + return id; + }); + this.tabIdClaim = claim; + return claim; } + return this.tabIdClaim; + } + + private async open(): Promise { + const { authMethod, credential } = this.credential(); + this.connectedKey = this.credentialKey(); + const epoch = this.connectionEpoch; + const generation = this.generation; const project = this.client.config.project ?? ''; // Client id, when the app did not set one with Client.setPushClientId(). Unlike the // native SDKs (which send an empty id and let the broker derive a stable one), mqtt.js - // rejects an empty id with clean start off, so derive a stable per-user id here from the - // credential this connection authenticates with: the JWT's userId, or the session's user - // id. A stable id keeps the broker's replay cursor resuming across reloads. The - // credential itself is never used as the id. + // rejects an empty id with clean start off, so derive one per user and tab from the + // credential this connection authenticates with (the JWT's userId, or the session's user + // id) and an id kept in the tab's sessionStorage. It stays the same across reloads, so + // the broker's replay cursor resumes, and differs between tabs and browsers, so two of + // them do not take over each other's session. The credential itself is never used. + const userId = + authMethod === 'appwrite-jwt' + ? userIdFromJwt(credential) + : userIdFromSession(credential); + // Claim once per instance (memoized) and reuse it across reconnects and credential + // switches, so concurrent open() calls never each claim and strand the tab's stored id. + // Concurrent instances still get distinct ids. If close() ran while the claim was pending + // this open is stale, so stop before connecting (the claim releases its own id). + if (!this.client.config.pushClientId) { + await this.claimClientId(); + if (generation !== this.generation) { + throw new Error('Push was closed while connecting.'); + } + } + const tabId = this.tabId; const clientId = this.client.config.pushClientId || - (authMethod === 'appwrite-jwt' - ? userIdFromJwt(credential) - : userIdFromSession(credential)) || - this.fallbackClientId; + (userId ? `${userId}-${tabId}` : tabId); - const generation = this.generation; const { connect, Buffer } = await loadTransport(); if (generation !== this.generation) { throw new Error('Push was closed while connecting.'); } + if (epoch !== this.connectionEpoch) { + throw new SupersededError(); + } this.Buffer = Buffer; const options: IClientOptions = { @@ -594,6 +717,9 @@ export class Push extends Service { } }); client.on('error', (err) => { + if (epoch !== this.connectionEpoch) { + return; + } const error = this.connectError(err, refusedReason); // While the first connect is pending its caller receives the error (and reports it // through fail()); afterwards, e.g. on reconnect, report it here. @@ -616,6 +742,10 @@ export class Push extends Service { await new Promise((resolve, reject) => { const onConnect = (packet: IConnackPacket) => { cleanup(); + if (epoch !== this.connectionEpoch) { + reject(new SupersededError()); + return; + } const rc = packet.reasonCode ?? packet.returnCode ?? 0; if (rc !== 0) { this.close(); @@ -631,16 +761,30 @@ export class Push extends Service { }; const onError = (err: Error) => { cleanup(); + if (epoch !== this.connectionEpoch) { + reject(new SupersededError()); + return; + } this.close(); reject(this.connectError(err, refusedReason)); }; + const onClose = () => { + if (epoch !== this.connectionEpoch) { + cleanup(); + reject(new SupersededError()); + } + }; const cleanup = () => { - this.opening = false; + if (epoch === this.connectionEpoch) { + this.opening = false; + } client.removeListener('connect', onConnect); client.removeListener('error', onError); + client.removeListener('close', onClose); }; client.on('connect', onConnect); client.on('error', onError); + client.on('close', onClose); client.connect(); }); } @@ -713,6 +857,12 @@ export class Push extends Service { } } +class SupersededError extends Error { + constructor() { + super('The push connection was superseded'); + } +} + /** Normalize anything thrown or passed to an error handler into an `Error`. */ function toError(error: unknown): Error { return error instanceof Error ? error : new Error(String(error)); @@ -757,6 +907,83 @@ function userIdFromSession(session?: string): string { } } +/** Ids in use by live connections in this page, so two Push instances never share a client id. */ +const claimedTabIds = new Set(); +/** The tab's stored id, resolved and locked once per page (see claimStoredTabId). */ +let storedTabIdClaim: Promise | null = null; + +/** + * A default client id for one Push connection in this page. The first connection alive takes the + * tab's stored id (kept in sessionStorage and locked for the page's lifetime, so it stays the same + * across reloads and the broker's replay cursor resumes). Any connection alive at the same time + * takes a distinct locked id instead, so simultaneous Push instances never share a client id and + * so cannot take over each other's broker session. `fallback` is this instance's own random id. + */ +async function claimTabId(fallback: string): Promise { + const stored = await claimStoredTabId(fallback); + if (stored !== '' && !claimedTabIds.has(stored)) { + claimedTabIds.add(stored); + return stored; + } + // The tab's id is taken by another live connection here: take a distinct, locked id without + // touching the tab's stored cursor. + let id = fallback; + while (claimedTabIds.has(id)) { + id = randomId(); + } + await holdTabId(id); + claimedTabIds.add(id); + return id; +} + +/** Free an id when its connection closes, so a later connection can reuse it (e.g. the stored id). */ +function releaseTabId(id: string): void { + claimedTabIds.delete(id); +} + +/** + * The tab's stored id, resolved and locked once per page: the value in sessionStorage when a lock + * on it is free (this tab owns it), else a fresh id that is locked and stored. '' when + * sessionStorage cannot be read, so the caller falls back to its own id. + */ +function claimStoredTabId(fallback: string): Promise { + storedTabIdClaim ??= (async () => { + let stored: string | null; + try { + stored = window.sessionStorage.getItem(TAB_ID_KEY); + } catch { + return ''; + } + if (stored && (await holdTabId(stored))) { + return stored; + } + await holdTabId(fallback); + try { + window.sessionStorage.setItem(TAB_ID_KEY, fallback); + } catch { + // Not persisted, but still usable as this page's stored id for this session. + } + return fallback; + })(); + return storedTabIdClaim; +} + +function holdTabId(id: string): Promise { + const locks = + typeof navigator === 'undefined' ? undefined : navigator.locks; + if (!locks) { + return Promise.resolve(true); + } + return new Promise((resolve) => { + locks + .request(`${TAB_ID_KEY}:${id}`, { ifAvailable: true }, (lock) => { + resolve(lock !== null); + return lock ? new Promise(() => {}) : undefined; + }) + .catch(() => resolve(true)); + }); +} + /** Base64 to UTF-8 text without `buffer`, which only loads once Push connects. */ function decodeBase64(value: string): string { return new TextDecoder().decode( diff --git a/tests/e2e/Android16Java17Test.php b/tests/e2e/Android16Java17Test.php index fa21b982be..f6e5483452 100644 --- a/tests/e2e/Android16Java17Test.php +++ b/tests/e2e/Android16Java17Test.php @@ -59,6 +59,7 @@ final class Android16Java17Test extends Base ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, ...Base::PUSH_BACKGROUND_RESPONSES, ...Base::ANDROID_PUSH_OPT_OUT_RESPONSES ]; diff --git a/tests/e2e/Android5Java17Test.php b/tests/e2e/Android5Java17Test.php index 21c178639d..8575883508 100644 --- a/tests/e2e/Android5Java17Test.php +++ b/tests/e2e/Android5Java17Test.php @@ -59,6 +59,7 @@ final class Android5Java17Test extends Base ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, ...Base::PUSH_BACKGROUND_RESPONSES, ...Base::ANDROID_PUSH_OPT_OUT_RESPONSES ]; diff --git a/tests/e2e/AppleSwift61Test.php b/tests/e2e/AppleSwift61Test.php index bcade9f18d..0e807ceb9e 100644 --- a/tests/e2e/AppleSwift61Test.php +++ b/tests/e2e/AppleSwift61Test.php @@ -58,6 +58,7 @@ final class AppleSwift61Test extends Base ...Base::CHANNEL_HELPER_RESPONSES, ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, - ...Base::PUSH_ERROR_RESPONSES + ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES ]; } diff --git a/tests/e2e/Base.php b/tests/e2e/Base.php index cfa31c456a..9b745039e1 100644 --- a/tests/e2e/Base.php +++ b/tests/e2e/Base.php @@ -197,6 +197,16 @@ abstract class Base extends TestCase 'Push session replay:passed', ]; + protected const PUSH_CREDENTIAL_SWITCH_RESPONSES = [ + 'Push credential switch:passed', + 'Push credential pending switch:passed', + ]; + + protected const WEB_PUSH_CONNECTION_RESPONSES = [ + 'Push concurrent refused:passed', + 'Push tab client id:passed', + ]; + protected const PUSH_SESSION_COOKIE_RESPONSES = [ 'Push user cookie topic:passed', 'Push user cookie switch:passed', diff --git a/tests/e2e/FlutterBetaTest.php b/tests/e2e/FlutterBetaTest.php index 550f9985da..5485736b30 100644 --- a/tests/e2e/FlutterBetaTest.php +++ b/tests/e2e/FlutterBetaTest.php @@ -59,6 +59,7 @@ final class FlutterBetaTest extends Base ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, ...Base::FLUTTER_PUSH_NATIVE_RESPONSES, ...Base::PUSH_NATIVE_DISPLACED_RESPONSES ]; diff --git a/tests/e2e/FlutterStableTest.php b/tests/e2e/FlutterStableTest.php index ce2dc1a0b4..ba0e38c255 100644 --- a/tests/e2e/FlutterStableTest.php +++ b/tests/e2e/FlutterStableTest.php @@ -79,6 +79,7 @@ final class FlutterStableTest extends Base ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, ...Base::FLUTTER_PUSH_NATIVE_RESPONSES, ...Base::PUSH_NATIVE_DISPLACED_RESPONSES, ...Base::PUSH_BACKGROUND_RESPONSES diff --git a/tests/e2e/FlutterWebTest.php b/tests/e2e/FlutterWebTest.php index 05dcd925f4..a5974fdf8c 100644 --- a/tests/e2e/FlutterWebTest.php +++ b/tests/e2e/FlutterWebTest.php @@ -43,6 +43,7 @@ final class FlutterWebTest extends Base #[Override] protected array $expectedOutput = [ ...Base::PUSH_RESPONSES, - ...Base::PUSH_ERROR_RESPONSES + ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES ]; } diff --git a/tests/e2e/Unity2021Test.php b/tests/e2e/Unity2021Test.php index 63c6c50ff5..3c0d37536b 100644 --- a/tests/e2e/Unity2021Test.php +++ b/tests/e2e/Unity2021Test.php @@ -112,6 +112,7 @@ public function tearDown(): void ...Base::CHANNEL_HELPER_RESPONSES, ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, - ...Base::PUSH_ERROR_RESPONSES + ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES ]; } diff --git a/tests/e2e/WebChromiumTest.php b/tests/e2e/WebChromiumTest.php index 94b7bb639d..b2f34bf38f 100644 --- a/tests/e2e/WebChromiumTest.php +++ b/tests/e2e/WebChromiumTest.php @@ -27,6 +27,7 @@ final class WebChromiumTest extends Base 'cp tests/e2e/languages/web/tests.js tests/e2e/sdks/web/tests.js', 'cp tests/e2e/languages/web/node.js tests/e2e/sdks/web/node.js', 'cp tests/e2e/languages/web/index.html tests/e2e/sdks/web/index.html', + 'cp tests/e2e/languages/web/tab.html tests/e2e/sdks/web/tab.html', 'docker run --rm -v $(pwd):/app -w /app/tests/e2e/sdks/web mcr.microsoft.com/playwright:v1.56.1-jammy sh -c "npm install && npm run build"', ]; #[Override] @@ -66,6 +67,8 @@ final class WebChromiumTest extends Base ...Base::OPERATOR_HELPER_RESPONSES, ...Base::PUSH_RESPONSES, ...Base::PUSH_SESSION_REPLAY_RESPONSES, - ...Base::PUSH_ERROR_RESPONSES + ...Base::PUSH_ERROR_RESPONSES, + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, + ...Base::WEB_PUSH_CONNECTION_RESPONSES ]; } diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 6db07f61a0..a5d061de98 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -667,6 +667,43 @@ class ServiceTest { }, ) + val outcome = { block: () -> Unit -> + try { + block() + "" + } catch (e: Exception) { + e.message ?: "" + } + } + val switchClient = pushClient().setSession( + android.util.Base64.encodeToString("{\"id\":\"e2e-switch-user\",\"secret\":\"e2e-secret\"}".toByteArray(), android.util.Base64.NO_WRAP), + ) + val switchPush = Push(switchClient, ApplicationProvider.getApplicationContext()) + switchPush.subscribe("e2e-switch") { } + switchClient.setJWT("deny:switched-user") + val switched = outcome { switchPush.subscribe("e2e-switch") { } } + switchPush.close() + writeToFile(if (switched == "switched-user") "Push credential switch:passed" else "Push credential switch:failed ($switched)") + + val pendingClient = pushClient().setJWT("slow:pending") + val pendingPush = Push(pendingClient, ApplicationProvider.getApplicationContext()) + val firstPending = java.util.concurrent.CompletableFuture.supplyAsync { + outcome { pendingPush.subscribe("e2e-switch") { } } + } + Thread.sleep(100) + pendingClient.setJWT("deny:switched-pending") + val secondPending = outcome { pendingPush.subscribe("e2e-switch") { } } + val firstPendingOutcome = runCatching { firstPending.get(10, java.util.concurrent.TimeUnit.SECONDS) }.getOrDefault("timeout") + pendingPush.close() + Thread.sleep(500) + writeToFile( + if (secondPending == "switched-pending" && firstPendingOutcome in listOf("", "switched-pending")) { + "Push credential pending switch:passed" + } else { + "Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending)" + }, + ) + // Background delivery, used the way an app does: subscribe with background on and a // PushReceiver declared, then the process dies, and the scheduled wake-up brings the // next message to the receiver and a notification. Sign-out stops it. @@ -727,7 +764,7 @@ class ServiceTest { if (E2EPushReceiver.messages.toList() == listOf("push-payload") && postedTitle == "E2E title" && postedText == "push-payload") { "Push background restore:passed" } else { - "Push background restore:failed" + "Push background restore:failed (messages: ${E2EPushReceiver.messages.toList()}, title: $postedTitle, text: $postedText)" }, ) diff --git a/tests/e2e/languages/apple/Tests.swift b/tests/e2e/languages/apple/Tests.swift index 0659bd24e0..1a61e472b4 100644 --- a/tests/e2e/languages/apple/Tests.swift +++ b/tests/e2e/languages/apple/Tests.swift @@ -629,6 +629,66 @@ class Tests: XCTestCase { let kickedError = await firstError(Push(client), subscribingTo: "e2e-disconnect/kicked-by-test") print(!kickedError.isEmpty ? "Push disconnect error:passed" : "Push disconnect error:failed") + + func outcome(_ subscribing: () async throws -> Void) async -> String { + do { + try await subscribing() + return "" + } catch { + let message = (error as? AppwriteError)?.message ?? error.localizedDescription + return message.isEmpty ? "failed" : message + } + } + + let deniedSwitch = await outcome { + _ = try await Push( + Client() + .setProject("console") + .setSelfSigned() + .setPushEndpoint("mqtt://mqtt:1883") + .setJWT("deny:switched-user") + ).subscribe("e2e-switch") { _ in } + } + + let switchClient = Client() + .setProject("console") + .setSelfSigned() + .setPushEndpoint("mqtt://mqtt:1883") + .setSession(e2eSession) + let switchPush = Push(switchClient) + let initial = await outcome { _ = try await switchPush.subscribe("e2e-switch") { _ in } } + _ = switchClient.setJWT("deny:switched-user") + let switched = await outcome { _ = try await switchPush.subscribe("e2e-switch") { _ in } } + switchPush.close() + print( + initial.isEmpty && !deniedSwitch.isEmpty && switched == deniedSwitch + ? "Push credential switch:passed" + : "Push credential switch:failed (initial: \(initial), switched: \(switched), denied: \(deniedSwitch))" + ) + + let pendingClient = Client() + .setProject("console") + .setSelfSigned() + .setPushEndpoint("mqtt://mqtt:1883") + .setJWT("slow:pending") + let pendingPush = Push(pendingClient) + let firstPending = ErrorCollector() + Task { + firstPending.record(await outcome { _ = try await pendingPush.subscribe("e2e-switch") { _ in } }) + } + try? await Task.sleep(nanoseconds: 100_000_000) + _ = pendingClient.setJWT("deny:switched-user") + let secondPending = await outcome { _ = try await pendingPush.subscribe("e2e-switch") { _ in } } + for _ in 0..<100 where firstPending.message == nil { + try? await Task.sleep(nanoseconds: 100_000_000) + } + let firstPendingOutcome = firstPending.message ?? "timeout" + pendingPush.close() + print( + secondPending == deniedSwitch && firstPendingOutcome == deniedSwitch + ? "Push credential pending switch:passed" + : "Push credential pending switch:failed (first: \(firstPendingOutcome), second: \(secondPending))" + ) } func parse(from json: String) -> String? { diff --git a/tests/e2e/languages/flutter/tests.dart b/tests/e2e/languages/flutter/tests.dart index 999eb1ad25..bec7c90aa5 100644 --- a/tests/e2e/languages/flutter/tests.dart +++ b/tests/e2e/languages/flutter/tests.dart @@ -670,6 +670,66 @@ void main() async { ); kickedPush.close(); + Future outcome(Future Function() subscribing) async { + try { + await subscribing(); + return ''; + } catch (e) { + final message = e.toString(); + if (message.contains('switched-user')) { + return 'switched-user'; + } + if (message.contains('switched-pending')) { + return 'switched-pending'; + } + return message; + } + } + + final switchClient = Client() + .setSelfSigned() + .setProject('console') + .setPushEndpoint('mqtt://mqtt:1883') + .setSession(e2eSession); + final switchPush = Push(switchClient); + await switchPush.subscribe(['e2e-switch'], (_) {}); + switchClient.setJWT('deny:switched-user'); + final switched = await outcome( + () => switchPush.subscribe(['e2e-switch'], (_) {}), + ); + switchPush.close(); + print( + switched == 'switched-user' + ? 'Push credential switch:passed' + : 'Push credential switch:failed ($switched)', + ); + + final pendingClient = Client() + .setSelfSigned() + .setProject('console') + .setPushEndpoint('mqtt://mqtt:1883') + .setJWT('slow:pending'); + final pendingPush = Push(pendingClient); + final firstPending = outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + await Future.delayed(const Duration(milliseconds: 100)); + pendingClient.setJWT('deny:switched-pending'); + final secondPending = await outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + final firstPendingOutcome = await firstPending.timeout( + const Duration(seconds: 5), + onTimeout: () => 'timeout', + ); + pendingPush.close(); + print( + secondPending == 'switched-pending' && + firstPendingOutcome == 'switched-pending' + ? 'Push credential pending switch:passed' + : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending)', + ); + // Background delivery on Android through the public API. The SDK's native Android plugin needs // a device, so a stand-in answers on its channels, as the mock server stands in for the broker; // the plugin itself is exercised over the same channels by the Robolectric run. diff --git a/tests/e2e/languages/flutter/web_tests.dart b/tests/e2e/languages/flutter/web_tests.dart index 76d2aff295..d698e17c91 100644 --- a/tests/e2e/languages/flutter/web_tests.dart +++ b/tests/e2e/languages/flutter/web_tests.dart @@ -163,4 +163,62 @@ void main() async { : 'Push disconnect error:failed', ); kickedPush.close(); + + Future outcome(Future Function() subscribing) async { + try { + await subscribing(); + return ''; + } catch (e) { + final message = e.toString(); + if (message.contains('switched-user')) { + return 'switched-user'; + } + if (message.contains('switched-pending')) { + return 'switched-pending'; + } + return message; + } + } + + final switchClient = Client() + .setProject('console') + .setPushEndpoint('ws://mqtt:8083') + .setSession(e2eSession); + final switchPush = Push(switchClient); + await switchPush.subscribe(['e2e-switch'], (_) {}); + switchClient.setJWT('deny:switched-user'); + final switched = await outcome( + () => switchPush.subscribe(['e2e-switch'], (_) {}), + ); + switchPush.close(); + print( + switched == 'switched-user' + ? 'Push credential switch:passed' + : 'Push credential switch:failed ($switched)', + ); + + final pendingClient = Client() + .setProject('console') + .setPushEndpoint('ws://mqtt:8083') + .setJWT('slow:pending'); + final pendingPush = Push(pendingClient); + final firstPending = outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + await Future.delayed(const Duration(milliseconds: 100)); + pendingClient.setJWT('deny:switched-pending'); + final secondPending = await outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + final firstPendingOutcome = await firstPending.timeout( + const Duration(seconds: 5), + onTimeout: () => 'timeout', + ); + pendingPush.close(); + print( + secondPending == 'switched-pending' && + firstPendingOutcome == 'switched-pending' + ? 'Push credential pending switch:passed' + : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending)', + ); } diff --git a/tests/e2e/languages/unity/Tests.cs b/tests/e2e/languages/unity/Tests.cs index 41d2cd9c66..7599a5c56e 100644 --- a/tests/e2e/languages/unity/Tests.cs +++ b/tests/e2e/languages/unity/Tests.cs @@ -653,6 +653,48 @@ async Task FirstError(Client errorClient, string topic) ? "Push disconnect error:passed" : "Push disconnect error:failed"); + async Task Outcome(System.Func subscribing) + { + try + { + await subscribing(); + return ""; + } + catch (System.Exception e) + { + return e.Message; + } + } + + var switchClient = PushClient().SetSession(e2eSession); + var switchPushObject = new GameObject("PushSwitchTest"); + var switchPush = switchPushObject.AddComponent(); + switchPush.Initialize(switchClient); + await switchPush.Subscribe("e2e-switch", (message) => { }); + switchClient.SetJWT("deny:switched-user"); + var switched = await Outcome(() => switchPush.Subscribe("e2e-switch", (message) => { })); + switchPush.Close(); + Object.DestroyImmediate(switchPushObject); + LogResult(switched == "switched-user" + ? "Push credential switch:passed" + : $"Push credential switch:failed ({switched})"); + + var pendingClient = PushClient().SetJWT("slow:pending"); + var pendingPushObject = new GameObject("PushPendingSwitchTest"); + var pendingPush = pendingPushObject.AddComponent(); + pendingPush.Initialize(pendingClient); + var firstPending = Outcome(() => pendingPush.Subscribe("e2e-switch", (message) => { })); + await Task.Delay(100); + pendingClient.SetJWT("deny:switched-pending"); + var secondPending = await Outcome(() => pendingPush.Subscribe("e2e-switch", (message) => { })); + var firstWinner = await Task.WhenAny(firstPending, Task.Delay(5000)); + var firstPendingOutcome = firstWinner == firstPending ? firstPending.Result : "timeout"; + pendingPush.Close(); + Object.DestroyImmediate(pendingPushObject); + LogResult(secondPending == "switched-pending" && firstPendingOutcome == "switched-pending" + ? "Push credential pending switch:passed" + : $"Push credential pending switch:failed (first: {firstPendingOutcome}, second: {secondPending})"); + // Cleanup Realtime GameObject if (realtimeObject) { diff --git a/tests/e2e/languages/web/index.html b/tests/e2e/languages/web/index.html index fe50ec3658..d5f4eff8fb 100644 --- a/tests/e2e/languages/web/index.html +++ b/tests/e2e/languages/web/index.html @@ -712,6 +712,85 @@ console.log((await kickedError) === 'kicked-by-test' ? 'Push disconnect error:passed' : 'Push disconnect error:failed'); kickedPush.close(); + const outcome = (subscribing) => subscribing.then(() => '', (e) => (e instanceof Error ? e.message : String(e))); + const switchClient = new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setSession(e2eSession); + const switchPush = new Push(switchClient); + await switchPush.subscribe(['e2e-switch'], () => {}); + switchClient.setJWT('deny:switched-user'); + const switched = await outcome(switchPush.subscribe(['e2e-switch'], () => {})); + switchPush.close(); + console.log(switched === 'switched-user' ? 'Push credential switch:passed' : `Push credential switch:failed (${switched})`); + + const pendingClient = new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setJWT('slow:pending'); + const pendingPush = new Push(pendingClient); + const firstPending = outcome(pendingPush.subscribe(['e2e-switch'], () => {})); + await new Promise((resolve) => setTimeout(resolve, 100)); + pendingClient.setJWT('deny:switched-pending'); + const secondPending = await outcome(pendingPush.subscribe(['e2e-switch'], () => {})); + const firstPendingOutcome = await Promise.race([firstPending, new Promise((resolve) => setTimeout(() => resolve('timeout'), 5000))]); + pendingPush.close(); + console.log( + secondPending === 'switched-pending' && firstPendingOutcome === 'switched-pending' + ? 'Push credential pending switch:passed' + : `Push credential pending switch:failed (first: ${firstPendingOutcome}, second: ${secondPending})`, + ); + + const refusedPush = new Push(new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setJWT('deny:concurrent-refused')); + const refused = await Promise.all([ + outcome(refusedPush.subscribe(['e2e-push'], () => {})), + outcome(refusedPush.subscribe(['e2e-push'], () => {})), + ]); + refusedPush.close(); + console.log( + refused.every((message) => message === 'concurrent-refused') + ? 'Push concurrent refused:passed' + : `Push concurrent refused:failed (${refused.join(', ')})`, + ); + + // The client id the broker sees for a Push. The mock echoes the connection's client id + // on e2e-whoami, so this is an observable identity, not the SDK's formatting. + const newTabPush = () => new Push(new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setSession(e2eSession)); + const whoamiOf = (tabPush) => Promise.race([ + new Promise((resolve) => { tabPush.subscribe(['e2e-whoami'], (m) => resolve(m.data)); }), + new Promise((resolve) => setTimeout(() => resolve(''), 5000)), + ]); + + // Closing and reopening in the same tab resumes the same client id, so the broker replays. + const firstTab = newTabPush(); + const pageId = await whoamiOf(firstTab); + firstTab.close(); + const secondTab = newTabPush(); + const samePageId = await whoamiOf(secondTab); + secondTab.close(); + + // Two Push instances alive at the same time in one page must get distinct client ids, so + // neither takes over the other's broker session. + const liveA = newTabPush(); + const liveB = newTabPush(); + const [liveAId, liveBId] = await Promise.all([whoamiOf(liveA), whoamiOf(liveB)]); + liveA.close(); + liveB.close(); + + // A popup copies its opener's sessionStorage, like a duplicated tab, and still gets its own id. + const popupId = await Promise.race([ + new Promise((resolve) => { + window.addEventListener('message', (event) => { + if (event.data && typeof event.data.whoami === 'string') { + resolve(event.data.whoami); + } + }); + window.open('tab.html', '_blank'); + }), + new Promise((resolve) => setTimeout(() => resolve(''), 5000)), + ]); + console.log( + pageId !== '' && pageId === samePageId && + liveAId !== '' && liveBId !== '' && liveAId !== liveBId && + popupId !== '' && popupId !== pageId + ? 'Push tab client id:passed' + : `Push tab client id:failed (page ${pageId}, same ${samePageId}, liveA ${liveAId}, liveB ${liveBId}, popup ${popupId})`, + ); + }); diff --git a/tests/e2e/languages/web/tab.html b/tests/e2e/languages/web/tab.html new file mode 100644 index 0000000000..ed90258009 --- /dev/null +++ b/tests/e2e/languages/web/tab.html @@ -0,0 +1,22 @@ + + + + + + + Appwrite tab + + + + + + + diff --git a/tests/e2e/languages/web/tests.js b/tests/e2e/languages/web/tests.js index 309a040d14..604d942be2 100644 --- a/tests/e2e/languages/web/tests.js +++ b/tests/e2e/languages/web/tests.js @@ -31,5 +31,5 @@ server.listen(3000, async () => { await browser.close(); server.close(); exit(0); - }, 15000); + }, 30000); });