From bd85ff76d39ef595b8e23668961af3f95ef88f4d Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:15:54 +0530 Subject: [PATCH 01/31] fix(web): wait for a pending Push connect and reconnect when the credential changes --- templates/web/src/services/push.ts.twig | 130 ++++++++++++++++++++---- 1 file changed, 110 insertions(+), 20 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 9fcbdc133c..40883bf1a9 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -127,6 +127,8 @@ export class Push extends Service { * this instance so reconnects resume the same broker session. */ private readonly fallbackClientId = randomId(); + private connectedKey = ''; + private connectionEpoch = 0; private readonly subscriptions = new Map< string, @@ -308,7 +310,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 +393,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; } @@ -478,20 +503,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 !== '') { @@ -511,20 +582,9 @@ export class Push extends Service { } 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.', - ); - } + const { authMethod, credential } = this.credential(); + this.connectedKey = this.credentialKey(); + const epoch = this.connectionEpoch; const project = this.client.config.project ?? ''; // Client id, when the app did not set one with Client.setPushClientId(). Unlike the @@ -545,6 +605,9 @@ export class Push extends Service { 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 +657,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 +682,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 +701,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 +797,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)); From 9873f3f6d36ef99058d76ea66818530938a26850 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:15:54 +0530 Subject: [PATCH 02/31] fix(web): give each tab its own default Push client id --- templates/web/src/services/push.ts.twig | 37 ++++++++++++++++++------- 1 file changed, 27 insertions(+), 10 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 40883bf1a9..4f8c976831 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,8 +125,8 @@ 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, kept in sessionStorage when it can be (see + * tabId): random, but kept for this instance so reconnects resume the same broker session. */ private readonly fallbackClientId = randomId(); private connectedKey = ''; @@ -548,6 +550,19 @@ export class Push extends Service { ); } + private tabId(): string { + try { + const stored = window.sessionStorage.getItem(TAB_ID_KEY); + if (stored) { + return stored; + } + window.sessionStorage.setItem(TAB_ID_KEY, this.fallbackClientId); + } catch { + return this.fallbackClientId; + } + return this.fallbackClientId; + } + private credentialKey(): string { try { const { authMethod, credential } = this.credential(); @@ -589,16 +604,18 @@ export class Push extends Service { 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); const clientId = this.client.config.pushClientId || - (authMethod === 'appwrite-jwt' - ? userIdFromJwt(credential) - : userIdFromSession(credential)) || - this.fallbackClientId; + (userId ? `${userId}-${this.tabId()}` : this.tabId()); const generation = this.generation; const { connect, Buffer } = await loadTransport(); From edfb511a9835a2a8eaf23e35cbe461bd2dd95e53 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:15:54 +0530 Subject: [PATCH 03/31] test(web): a credential change reconnects Push, also while it is connecting --- tests/e2e/Base.php | 5 +++++ tests/e2e/WebChromiumTest.php | 3 ++- tests/e2e/languages/web/index.html | 27 +++++++++++++++++++++++++++ 3 files changed, 34 insertions(+), 1 deletion(-) diff --git a/tests/e2e/Base.php b/tests/e2e/Base.php index cfa31c456a..70b4b08676 100644 --- a/tests/e2e/Base.php +++ b/tests/e2e/Base.php @@ -197,6 +197,11 @@ 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 PUSH_SESSION_COOKIE_RESPONSES = [ 'Push user cookie topic:passed', 'Push user cookie switch:passed', diff --git a/tests/e2e/WebChromiumTest.php b/tests/e2e/WebChromiumTest.php index 94b7bb639d..c3719bd470 100644 --- a/tests/e2e/WebChromiumTest.php +++ b/tests/e2e/WebChromiumTest.php @@ -66,6 +66,7 @@ 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 ]; } diff --git a/tests/e2e/languages/web/index.html b/tests/e2e/languages/web/index.html index fe50ec3658..84b9362605 100644 --- a/tests/e2e/languages/web/index.html +++ b/tests/e2e/languages/web/index.html @@ -710,6 +710,33 @@ await kickedPush.subscribe(['e2e-disconnect/kicked-by-test'], () => {}); } catch (e) {} console.log((await kickedError) === 'kicked-by-test' ? 'Push disconnect error:passed' : 'Push disconnect error:failed'); + + 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').setSession(e2eSession); + const pendingPush = new Push(pendingClient); + let pendingOpened = false; + pendingPush.onOpen(() => (pendingOpened = true)); + const firstPending = outcome(pendingPush.subscribe(['e2e-switch'], () => {})); + await new Promise((resolve) => setTimeout(resolve, 0)); + const switchedWhileConnecting = !pendingOpened; + 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' || (!switchedWhileConnecting && firstPendingOutcome === '')) + ? 'Push credential pending switch:passed' + : `Push credential pending switch:failed (first: ${firstPendingOutcome}, second: ${secondPending}, connecting: ${switchedWhileConnecting})`, + ); kickedPush.close(); }); From 1ee6657d11c87bad2546bb33ded62a17a8247af5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:18:19 +0530 Subject: [PATCH 04/31] fix(android): serialize Push connects and reconnect when the credential changes --- .../java/io/package/services/Push.kt.twig | 81 +++++++++++++++---- 1 file changed, 66 insertions(+), 15 deletions(-) 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 { From 79fcd0938c43e9ff561a6d6934a38e12f49ee5ac Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:18:19 +0530 Subject: [PATCH 05/31] test(android): a credential change reconnects Push, also while a subscribe is connecting --- tests/e2e/Android16Java17Test.php | 1 + tests/e2e/Android5Java17Test.php | 1 + tests/e2e/languages/android/Tests.kt | 34 ++++++++++++++++++++++++++++ 3 files changed, 36 insertions(+) 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/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 6db07f61a0..5dba0198df 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -667,6 +667,40 @@ class ServiceTest { }, ) + val outcome = { block: () -> Unit -> + try { + block() + "" + } catch (e: Exception) { + e.message ?: "" + } + } + val switchClient = pushClient().setSession(e2eSession) + 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().setSession(e2eSession) + val pendingPush = Push(pendingClient, ApplicationProvider.getApplicationContext()) + val firstPending = java.util.concurrent.CompletableFuture.supplyAsync { + outcome { pendingPush.subscribe("e2e-switch") { } } + } + Thread.sleep(50) + 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() + 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. From 61bc7893944a1a620bcc62d9d1d564f8fa6cf503 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:22:42 +0530 Subject: [PATCH 06/31] fix(flutter): reconnect Push when the credential changes and keep other subscriptions on a failed connect --- templates/flutter/lib/src/mqtt_io.dart.twig | 94 ++++++++++++++------- 1 file changed, 64 insertions(+), 30 deletions(-) diff --git a/templates/flutter/lib/src/mqtt_io.dart.twig b/templates/flutter/lib/src/mqtt_io.dart.twig index 993c3c7930..4c8bb18654 100644 --- a/templates/flutter/lib/src/mqtt_io.dart.twig +++ b/templates/flutter/lib/src/mqtt_io.dart.twig @@ -84,8 +84,11 @@ 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; + 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 +259,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 +344,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 +460,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 +487,23 @@ class PushIO implements Push { _nativeActive = true; } + void _dropConnection() { + _connectionEpoch++; + _connecting = 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 +876,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 +892,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 +921,7 @@ class PushIO implements Push { Future _open() async { final (authMethod, credential) = _credential(); + _connectedKey = _credentialKey; final epoch = _connectionEpoch; final project = client.config['project'] ?? ''; @@ -917,7 +939,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); @@ -975,12 +997,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 +1010,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) { @@ -1287,7 +1320,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(); } From b97e9097258de8239c8074bc7facb6916f03f843 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:22:42 +0530 Subject: [PATCH 07/31] fix(flutter): reconnect web Push when the credential changes and keep other subscriptions on a failed connect --- .../flutter/lib/src/mqtt_browser.dart.twig | 118 +++++++++++++++--- 1 file changed, 102 insertions(+), 16 deletions(-) diff --git a/templates/flutter/lib/src/mqtt_browser.dart.twig b/templates/flutter/lib/src/mqtt_browser.dart.twig index 42eddce3fe..9ce8423276 100644 --- a/templates/flutter/lib/src/mqtt_browser.dart.twig +++ b/templates/flutter/lib/src/mqtt_browser.dart.twig @@ -62,6 +62,9 @@ class PushWeb implements Push { MqttBrowserClient? _mqtt; Future? _connecting; + int _connectionEpoch = 0; + String _connectedKey = ''; + bool _resubscribeOnConnect = false; void Function()? _onOpen; void Function()? _onClose; @@ -253,9 +256,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 +349,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 +431,22 @@ class PushWeb implements Push { } Future _connect() { + if ((_mqtt != null || _connecting != null) && + _connectedKey != _credentialKey) { + _connectionEpoch++; + _connecting = 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 +458,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 @@ -498,14 +561,20 @@ class PushWeb implements Push { _mqtt = mqtt; final failure = await _connectOrFail(mqtt, (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 +582,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) { @@ -810,3 +892,7 @@ String? _stringField(String encoded, String key) { return null; } } + +class _Superseded implements Exception { + const _Superseded(); +} From 74553969e1dd95347ac1f960af8c98bb101adfb5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:22:42 +0530 Subject: [PATCH 08/31] test(flutter): a credential change reconnects Push, also while it is connecting --- tests/e2e/FlutterBetaTest.php | 1 + tests/e2e/FlutterStableTest.php | 1 + tests/e2e/FlutterWebTest.php | 3 +- tests/e2e/languages/flutter/tests.dart | 64 ++++++++++++++++++++++ tests/e2e/languages/flutter/web_tests.dart | 62 +++++++++++++++++++++ 5 files changed, 130 insertions(+), 1 deletion(-) 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/languages/flutter/tests.dart b/tests/e2e/languages/flutter/tests.dart index 999eb1ad25..4e88d2ba60 100644 --- a/tests/e2e/languages/flutter/tests.dart +++ b/tests/e2e/languages/flutter/tests.dart @@ -670,6 +670,70 @@ 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') + .setSession(e2eSession); + final pendingPush = Push(pendingClient); + var pendingOpened = false; + pendingPush.onOpen(() => pendingOpened = true); + final firstPending = outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + await Future.delayed(Duration.zero); + final switchedWhileConnecting = !pendingOpened; + 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' || + (!switchedWhileConnecting && firstPendingOutcome == '')) + ? 'Push credential pending switch:passed' + : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending, connecting: $switchedWhileConnecting)', + ); + // 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..c0ae4f7917 100644 --- a/tests/e2e/languages/flutter/web_tests.dart +++ b/tests/e2e/languages/flutter/web_tests.dart @@ -163,4 +163,66 @@ 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') + .setSession(e2eSession); + final pendingPush = Push(pendingClient); + var pendingOpened = false; + pendingPush.onOpen(() => pendingOpened = true); + final firstPending = outcome( + () => pendingPush.subscribe(['e2e-switch'], (_) {}), + ); + await Future.delayed(Duration.zero); + final switchedWhileConnecting = !pendingOpened; + 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' || + (!switchedWhileConnecting && firstPendingOutcome == '')) + ? 'Push credential pending switch:passed' + : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending, connecting: $switchedWhileConnecting)', + ); } From fcc632dd9f571ea46d09612f793f66322d47a5d5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:24:44 +0530 Subject: [PATCH 09/31] fix(unity): share one Push connect, reconnect on credential change and keep other subscriptions on a failed connect --- templates/unity/Assets/Runtime/Push.cs.twig | 144 ++++++++++++++++++-- 1 file changed, 129 insertions(+), 15 deletions(-) diff --git a/templates/unity/Assets/Runtime/Push.cs.twig b/templates/unity/Assets/Runtime/Push.cs.twig index 878954d01d..0782614f51 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; @@ -439,11 +443,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 +565,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 +632,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 +646,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); + } + + var session = _client.GetSession(); + if (!string.IsNullOrEmpty(session)) + { + return ("appwrite-session", session); } - else if (!string.IsNullOrEmpty(session)) + + throw new {{ spec.info.title | caseUcfirst }}Exception("No credential set on the client; call Client.SetJWT() or Client.SetSession() first."); + } + + private string CredentialKey() + { + try { - authMethod = "appwrite-session"; - credential = session; + var (authMethod, credential) = Credential(); + return string.Join("|", PushEndpoint, _client.Config.GetValueOrDefault("project") ?? "", _client.Config.GetValueOrDefault("pushClientId") ?? "", authMethod, credential); } - else + catch (Exception) { - throw new {{ spec.info.title | caseUcfirst }}Exception("No credential set on the client; call Client.SetJWT() or Client.SetSession() first."); + 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). @@ -732,19 +798,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 +1082,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 From e9b863429a7315969e43ce9d531b83c353ed465e Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:24:44 +0530 Subject: [PATCH 10/31] test(unity): a credential change reconnects Push, also while it is connecting --- tests/e2e/Unity2021Test.php | 3 +- tests/e2e/languages/unity/Tests.cs | 45 ++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 1 deletion(-) 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/languages/unity/Tests.cs b/tests/e2e/languages/unity/Tests.cs index 41d2cd9c66..27956d247a 100644 --- a/tests/e2e/languages/unity/Tests.cs +++ b/tests/e2e/languages/unity/Tests.cs @@ -653,6 +653,51 @@ 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().SetSession(e2eSession); + var pendingPushObject = new GameObject("PushPendingSwitchTest"); + var pendingPush = pendingPushObject.AddComponent(); + pendingPush.Initialize(pendingClient); + var pendingOpened = false; + pendingPush.OnOpen(() => pendingOpened = true); + var firstPending = Outcome(() => pendingPush.Subscribe("e2e-switch", (message) => { })); + var switchedWhileConnecting = !pendingOpened && !firstPending.IsCompleted; + 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" || (!switchedWhileConnecting && firstPendingOutcome == "")) + ? "Push credential pending switch:passed" + : $"Push credential pending switch:failed (first: {firstPendingOutcome}, second: {secondPending}, connecting: {switchedWhileConnecting})"); + // Cleanup Realtime GameObject if (realtimeObject) { From c1ba3727128e083bed7d8f7be1a1ff3a1d55ff6a Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:28:44 +0530 Subject: [PATCH 11/31] fix(apple): reconnect Push when the credential changes, also while it is connecting --- .../apple/Sources/Services/Push.swift.twig | 92 ++++++++++++++++++- 1 file changed, 88 insertions(+), 4 deletions(-) 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 {} From b5d98e462aa48251feea337aae424776e1ff8f6f Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:28:44 +0530 Subject: [PATCH 12/31] test(apple): a credential change reconnects Push, also while it is connecting --- tests/e2e/AppleSwift61Test.php | 3 +- tests/e2e/languages/apple/Tests.swift | 48 +++++++++++++++++++++++++++ 2 files changed, 50 insertions(+), 1 deletion(-) 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/languages/apple/Tests.swift b/tests/e2e/languages/apple/Tests.swift index 0659bd24e0..419028d0ed 100644 --- a/tests/e2e/languages/apple/Tests.swift +++ b/tests/e2e/languages/apple/Tests.swift @@ -629,6 +629,54 @@ 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 switchClient = Client() + .setProject("console") + .setSelfSigned() + .setPushEndpoint("mqtt://mqtt:1883") + .setSession(e2eSession) + let switchPush = Push(switchClient) + _ = 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(!switched.isEmpty ? "Push credential switch:passed" : "Push credential switch:failed") + + let pendingClient = Client() + .setProject("console") + .setSelfSigned() + .setPushEndpoint("mqtt://mqtt:1883") + .setSession(e2eSession) + let pendingPush = Push(pendingClient) + let firstPending = Task { await outcome { _ = try await pendingPush.subscribe("e2e-switch") { _ in } } } + _ = pendingClient.setJWT("deny:switched-pending") + let secondPending = await outcome { _ = try await pendingPush.subscribe("e2e-switch") { _ in } } + let firstPendingOutcome = await withTaskGroup(of: String?.self) { group -> String in + group.addTask { await firstPending.value } + group.addTask { + try? await Task.sleep(nanoseconds: 10_000_000_000) + return nil + } + let first = await group.next() ?? nil + group.cancelAll() + return first ?? "timeout" + } + pendingPush.close() + print( + !secondPending.isEmpty && firstPendingOutcome != "timeout" + ? "Push credential pending switch:passed" + : "Push credential pending switch:failed (first: \(firstPendingOutcome), second: \(secondPending))" + ) } func parse(from json: String) -> String? { From 6e8b31d86e65f550b4b2b3de25d044d04117c109 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:39:02 +0530 Subject: [PATCH 13/31] fix(flutter): let a superseded Push connect stop waiting for its CONNACK --- templates/flutter/lib/src/mqtt_browser.dart.twig | 10 ++++++++-- templates/flutter/lib/src/mqtt_io.dart.twig | 10 ++++++++-- 2 files changed, 16 insertions(+), 4 deletions(-) diff --git a/templates/flutter/lib/src/mqtt_browser.dart.twig b/templates/flutter/lib/src/mqtt_browser.dart.twig index 9ce8423276..2cd292d6f2 100644 --- a/templates/flutter/lib/src/mqtt_browser.dart.twig +++ b/templates/flutter/lib/src/mqtt_browser.dart.twig @@ -63,6 +63,7 @@ class PushWeb implements Push { MqttBrowserClient? _mqtt; Future? _connecting; int _connectionEpoch = 0; + Completer? _openCancel; String _connectedKey = ''; bool _resubscribeOnConnect = false; @@ -435,6 +436,8 @@ class PushWeb implements Push { _connectedKey != _credentialKey) { _connectionEpoch++; _connecting = null; + _openCancel?.complete(); + _openCancel = null; for (final waiting in _subAcks.values) { for (final completer in waiting) { if (!completer.isCompleted) { @@ -560,7 +563,9 @@ 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; } @@ -669,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(); @@ -680,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 { diff --git a/templates/flutter/lib/src/mqtt_io.dart.twig b/templates/flutter/lib/src/mqtt_io.dart.twig index 4c8bb18654..a44c156be0 100644 --- a/templates/flutter/lib/src/mqtt_io.dart.twig +++ b/templates/flutter/lib/src/mqtt_io.dart.twig @@ -87,6 +87,7 @@ class PushIO implements Push { // 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; @@ -490,6 +491,8 @@ class PushIO implements Push { void _dropConnection() { _connectionEpoch++; _connecting = null; + _openCancel?.complete(); + _openCancel = null; for (final waiting in _subAcks.values) { for (final completer in waiting) { if (!completer.isCompleted) { @@ -987,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; @@ -1090,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(); @@ -1101,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 { From b5468d95f20f99a89fe0c53bbf9bcf5f41c7b167 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 14/31] test(mock): delay CONNACK for a slow credential and echo the client id on e2e-whoami --- mock-server/app/mqtt.php | 13 +++++++++++++ 1 file changed, 13 insertions(+) diff --git a/mock-server/app/mqtt.php b/mock-server/app/mqtt.php index 29d8191fde..e5c578891f 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: Packet::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) { From 02b48113a2aeed6dfc19231e438c7a1b43aa98a2 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 15/31] fix(web): hold the tab's Push client id with a Web Lock so a duplicated tab picks its own --- templates/web/src/services/push.ts.twig | 60 ++++++++++++++++++++----- 1 file changed, 48 insertions(+), 12 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 4f8c976831..b4cc34d113 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -130,6 +130,7 @@ export class Push extends Service { */ private readonly fallbackClientId = randomId(); private connectedKey = ''; + private tabIdClaim: Promise | null = null; private connectionEpoch = 0; private readonly subscriptions = new Map< @@ -474,6 +475,9 @@ export class Push extends Service { this.generation++; this.everConnected = false; this.subscriptions.clear(); + const claim = this.tabIdClaim; + this.tabIdClaim = null; + claim?.then(({ release }) => release()); } /** @@ -550,17 +554,9 @@ export class Push extends Service { ); } - private tabId(): string { - try { - const stored = window.sessionStorage.getItem(TAB_ID_KEY); - if (stored) { - return stored; - } - window.sessionStorage.setItem(TAB_ID_KEY, this.fallbackClientId); - } catch { - return this.fallbackClientId; - } - return this.fallbackClientId; + private async tabId(): Promise { + this.tabIdClaim ??= claimTabId(this.fallbackClientId); + return (await this.tabIdClaim).id; } private credentialKey(): string { @@ -613,9 +609,10 @@ export class Push extends Service { authMethod === 'appwrite-jwt' ? userIdFromJwt(credential) : userIdFromSession(credential); + const tabId = this.client.config.pushClientId ? '' : await this.tabId(); const clientId = this.client.config.pushClientId || - (userId ? `${userId}-${this.tabId()}` : this.tabId()); + (userId ? `${userId}-${tabId}` : tabId); const generation = this.generation; const { connect, Buffer } = await loadTransport(); @@ -864,6 +861,45 @@ function userIdFromSession(session?: string): string { } } +type TabIdClaim = { id: string; release: () => void }; + +async function claimTabId(fallback: string): Promise { + let stored: string | null; + try { + stored = window.sessionStorage.getItem(TAB_ID_KEY); + } catch { + return { id: fallback, release: () => {} }; + } + const held = stored ? await holdTabId(stored) : null; + const id = held && stored ? stored : fallback; + const release = held ?? (await holdTabId(fallback)) ?? (() => {}); + try { + window.sessionStorage.setItem(TAB_ID_KEY, id); + } catch { + return { id, release }; + } + return { id, release }; +} + +function holdTabId(id: string): Promise<(() => void) | null> { + const locks = + typeof navigator === 'undefined' ? undefined : navigator.locks; + if (!locks) { + return Promise.resolve(() => {}); + } + return new Promise((resolve) => { + locks + .request(`${TAB_ID_KEY}:${id}`, { ifAvailable: true }, (lock) => { + if (!lock) { + resolve(null); + return undefined; + } + return new Promise((release) => resolve(() => release())); + }) + .catch(() => resolve(() => {})); + }); +} + /** Base64 to UTF-8 text without `buffer`, which only loads once Push connects. */ function decodeBase64(value: string): string { return new TextDecoder().decode( From d9a98fe3cdecfd8ac1132e8be86a06ae7563a02d Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 16/31] docs(web): describe the per-tab default in setPushClientId --- templates/web/src/client.ts.twig | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/templates/web/src/client.ts.twig b/templates/web/src/client.ts.twig index 5d13ac602c..0298f7a3cc 100644 --- a/templates/web/src/client.ts.twig +++ b/templates/web/src/client.ts.twig @@ -501,7 +501,9 @@ 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 to one per user and tab when unset: stable across reloads of + * the tab, and different in every other tab, so tabs do not take over each other's session. + * An explicit id shared by several tabs makes them do that. * * @param {string} pushClientId * From 97f7b324d4c72b738aa709399875717e7fdcf601 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 17/31] fix(unity): read the Push client id on every connect and ignore a replaced client's disconnect --- templates/unity/Assets/Runtime/Push.cs.twig | 16 ++++++++++------ 1 file changed, 10 insertions(+), 6 deletions(-) diff --git a/templates/unity/Assets/Runtime/Push.cs.twig b/templates/unity/Assets/Runtime/Push.cs.twig index 0782614f51..d5757b0413 100644 --- a/templates/unity/Assets/Runtime/Push.cs.twig +++ b/templates/unity/Assets/Runtime/Push.cs.twig @@ -65,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; @@ -728,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() @@ -767,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) From c52af3e780308978ebce8bdd045dd2764ee91f65 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 18/31] test(web): check concurrent refusals and per-tab client ids, and switch while connecting --- tests/e2e/Base.php | 5 ++++ tests/e2e/WebChromiumTest.php | 3 ++- tests/e2e/languages/web/index.html | 42 ++++++++++++++++++++++++------ 3 files changed, 41 insertions(+), 9 deletions(-) diff --git a/tests/e2e/Base.php b/tests/e2e/Base.php index 70b4b08676..9b745039e1 100644 --- a/tests/e2e/Base.php +++ b/tests/e2e/Base.php @@ -202,6 +202,11 @@ abstract class Base extends TestCase '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/WebChromiumTest.php b/tests/e2e/WebChromiumTest.php index c3719bd470..789f8c7e31 100644 --- a/tests/e2e/WebChromiumTest.php +++ b/tests/e2e/WebChromiumTest.php @@ -67,6 +67,7 @@ final class WebChromiumTest extends Base ...Base::PUSH_RESPONSES, ...Base::PUSH_SESSION_REPLAY_RESPONSES, ...Base::PUSH_ERROR_RESPONSES, - ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES + ...Base::PUSH_CREDENTIAL_SWITCH_RESPONSES, + ...Base::WEB_PUSH_CONNECTION_RESPONSES ]; } diff --git a/tests/e2e/languages/web/index.html b/tests/e2e/languages/web/index.html index 84b9362605..e186a742d3 100644 --- a/tests/e2e/languages/web/index.html +++ b/tests/e2e/languages/web/index.html @@ -720,22 +720,48 @@ 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').setSession(e2eSession); + const pendingClient = new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setJWT('slow:pending'); const pendingPush = new Push(pendingClient); - let pendingOpened = false; - pendingPush.onOpen(() => (pendingOpened = true)); const firstPending = outcome(pendingPush.subscribe(['e2e-switch'], () => {})); - await new Promise((resolve) => setTimeout(resolve, 0)); - const switchedWhileConnecting = !pendingOpened; + 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' || (!switchedWhileConnecting && firstPendingOutcome === '')) + secondPending === 'switched-pending' && firstPendingOutcome === 'switched-pending' ? 'Push credential pending switch:passed' - : `Push credential pending switch:failed (first: ${firstPendingOutcome}, second: ${secondPending}, connecting: ${switchedWhileConnecting})`, + : `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(', ')})`, + ); + + const whoami = async () => { + const tabPush = new Push(new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setSession(e2eSession)); + const id = await Promise.race([ + new Promise((resolve) => { tabPush.subscribe(['e2e-whoami'], (m) => resolve(m.data)); }), + new Promise((resolve) => setTimeout(() => resolve(''), 5000)), + ]); + return { tabPush, id }; + }; + const firstTab = await whoami(); + const secondTab = await whoami(); + firstTab.tabPush.close(); + secondTab.tabPush.close(); + console.log( + firstTab.id.startsWith('e2e-session-user-') && secondTab.id.startsWith('e2e-session-user-') && firstTab.id !== secondTab.id + ? 'Push tab client id:passed' + : `Push tab client id:failed (${firstTab.id}, ${secondTab.id})`, ); kickedPush.close(); From bd41ba23a945ae9239515b25e26add59b0b62681 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 14:54:34 +0530 Subject: [PATCH 19/31] test: switch the credential while a slow connect is pending, and tighten the Apple checks --- tests/e2e/languages/android/Tests.kt | 4 +-- tests/e2e/languages/apple/Tests.swift | 42 ++++++++++++++-------- tests/e2e/languages/flutter/tests.dart | 12 +++---- tests/e2e/languages/flutter/web_tests.dart | 12 +++---- tests/e2e/languages/unity/Tests.cs | 11 +++--- 5 files changed, 41 insertions(+), 40 deletions(-) diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 5dba0198df..50b9e7e9ba 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -683,12 +683,12 @@ class ServiceTest { switchPush.close() writeToFile(if (switched == "switched-user") "Push credential switch:passed" else "Push credential switch:failed ($switched)") - val pendingClient = pushClient().setSession(e2eSession) + 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(50) + 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") diff --git a/tests/e2e/languages/apple/Tests.swift b/tests/e2e/languages/apple/Tests.swift index 419028d0ed..1a61e472b4 100644 --- a/tests/e2e/languages/apple/Tests.swift +++ b/tests/e2e/languages/apple/Tests.swift @@ -640,40 +640,52 @@ class Tests: XCTestCase { } } + 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) - _ = try? await switchPush.subscribe("e2e-switch") { _ in } + 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(!switched.isEmpty ? "Push credential switch:passed" : "Push credential switch:failed") + 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") - .setSession(e2eSession) + .setJWT("slow:pending") let pendingPush = Push(pendingClient) - let firstPending = Task { await outcome { _ = try await pendingPush.subscribe("e2e-switch") { _ in } } } - _ = pendingClient.setJWT("deny:switched-pending") + 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 } } - let firstPendingOutcome = await withTaskGroup(of: String?.self) { group -> String in - group.addTask { await firstPending.value } - group.addTask { - try? await Task.sleep(nanoseconds: 10_000_000_000) - return nil - } - let first = await group.next() ?? nil - group.cancelAll() - return first ?? "timeout" + 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.isEmpty && firstPendingOutcome != "timeout" + secondPending == deniedSwitch && firstPendingOutcome == deniedSwitch ? "Push credential pending switch:passed" : "Push credential pending switch:failed (first: \(firstPendingOutcome), second: \(secondPending))" ) diff --git a/tests/e2e/languages/flutter/tests.dart b/tests/e2e/languages/flutter/tests.dart index 4e88d2ba60..bec7c90aa5 100644 --- a/tests/e2e/languages/flutter/tests.dart +++ b/tests/e2e/languages/flutter/tests.dart @@ -708,15 +708,12 @@ void main() async { .setSelfSigned() .setProject('console') .setPushEndpoint('mqtt://mqtt:1883') - .setSession(e2eSession); + .setJWT('slow:pending'); final pendingPush = Push(pendingClient); - var pendingOpened = false; - pendingPush.onOpen(() => pendingOpened = true); final firstPending = outcome( () => pendingPush.subscribe(['e2e-switch'], (_) {}), ); - await Future.delayed(Duration.zero); - final switchedWhileConnecting = !pendingOpened; + await Future.delayed(const Duration(milliseconds: 100)); pendingClient.setJWT('deny:switched-pending'); final secondPending = await outcome( () => pendingPush.subscribe(['e2e-switch'], (_) {}), @@ -728,10 +725,9 @@ void main() async { pendingPush.close(); print( secondPending == 'switched-pending' && - (firstPendingOutcome == 'switched-pending' || - (!switchedWhileConnecting && firstPendingOutcome == '')) + firstPendingOutcome == 'switched-pending' ? 'Push credential pending switch:passed' - : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending, connecting: $switchedWhileConnecting)', + : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending)', ); // Background delivery on Android through the public API. The SDK's native Android plugin needs diff --git a/tests/e2e/languages/flutter/web_tests.dart b/tests/e2e/languages/flutter/web_tests.dart index c0ae4f7917..d698e17c91 100644 --- a/tests/e2e/languages/flutter/web_tests.dart +++ b/tests/e2e/languages/flutter/web_tests.dart @@ -200,15 +200,12 @@ void main() async { final pendingClient = Client() .setProject('console') .setPushEndpoint('ws://mqtt:8083') - .setSession(e2eSession); + .setJWT('slow:pending'); final pendingPush = Push(pendingClient); - var pendingOpened = false; - pendingPush.onOpen(() => pendingOpened = true); final firstPending = outcome( () => pendingPush.subscribe(['e2e-switch'], (_) {}), ); - await Future.delayed(Duration.zero); - final switchedWhileConnecting = !pendingOpened; + await Future.delayed(const Duration(milliseconds: 100)); pendingClient.setJWT('deny:switched-pending'); final secondPending = await outcome( () => pendingPush.subscribe(['e2e-switch'], (_) {}), @@ -220,9 +217,8 @@ void main() async { pendingPush.close(); print( secondPending == 'switched-pending' && - (firstPendingOutcome == 'switched-pending' || - (!switchedWhileConnecting && firstPendingOutcome == '')) + firstPendingOutcome == 'switched-pending' ? 'Push credential pending switch:passed' - : 'Push credential pending switch:failed (first: $firstPendingOutcome, second: $secondPending, connecting: $switchedWhileConnecting)', + : '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 27956d247a..7599a5c56e 100644 --- a/tests/e2e/languages/unity/Tests.cs +++ b/tests/e2e/languages/unity/Tests.cs @@ -679,24 +679,21 @@ async Task Outcome(System.Func subscribing) ? "Push credential switch:passed" : $"Push credential switch:failed ({switched})"); - var pendingClient = PushClient().SetSession(e2eSession); + var pendingClient = PushClient().SetJWT("slow:pending"); var pendingPushObject = new GameObject("PushPendingSwitchTest"); var pendingPush = pendingPushObject.AddComponent(); pendingPush.Initialize(pendingClient); - var pendingOpened = false; - pendingPush.OnOpen(() => pendingOpened = true); var firstPending = Outcome(() => pendingPush.Subscribe("e2e-switch", (message) => { })); - var switchedWhileConnecting = !pendingOpened && !firstPending.IsCompleted; + 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" || (!switchedWhileConnecting && firstPendingOutcome == "")) + LogResult(secondPending == "switched-pending" && firstPendingOutcome == "switched-pending" ? "Push credential pending switch:passed" - : $"Push credential pending switch:failed (first: {firstPendingOutcome}, second: {secondPending}, connecting: {switchedWhileConnecting})"); + : $"Push credential pending switch:failed (first: {firstPendingOutcome}, second: {secondPending})"); // Cleanup Realtime GameObject if (realtimeObject) From 7588b12eca3971a698a4ce3514610540777d9b0c Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:06:32 +0530 Subject: [PATCH 20/31] fix(web): claim the Push tab id once per page and hold it for the page's lifetime --- templates/web/src/services/push.ts.twig | 54 ++++++++++++------------- 1 file changed, 25 insertions(+), 29 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index b4cc34d113..da23f1d564 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -125,12 +125,11 @@ 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). /** - * This tab's part of the default client id, kept in sessionStorage when it can be (see - * tabId): 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(); private connectedKey = ''; - private tabIdClaim: Promise | null = null; private connectionEpoch = 0; private readonly subscriptions = new Map< @@ -475,9 +474,6 @@ export class Push extends Service { this.generation++; this.everConnected = false; this.subscriptions.clear(); - const claim = this.tabIdClaim; - this.tabIdClaim = null; - claim?.then(({ release }) => release()); } /** @@ -554,11 +550,6 @@ export class Push extends Service { ); } - private async tabId(): Promise { - this.tabIdClaim ??= claimTabId(this.fallbackClientId); - return (await this.tabIdClaim).id; - } - private credentialKey(): string { try { const { authMethod, credential } = this.credential(); @@ -609,7 +600,9 @@ export class Push extends Service { authMethod === 'appwrite-jwt' ? userIdFromJwt(credential) : userIdFromSession(credential); - const tabId = this.client.config.pushClientId ? '' : await this.tabId(); + const tabId = this.client.config.pushClientId + ? '' + : await claimTabId(this.fallbackClientId); const clientId = this.client.config.pushClientId || (userId ? `${userId}-${tabId}` : tabId); @@ -861,42 +854,45 @@ function userIdFromSession(session?: string): string { } } -type TabIdClaim = { id: string; release: () => void }; +let tabIdClaim: Promise | null = null; + +function claimTabId(fallback: string): Promise { + tabIdClaim ??= claimStoredTabId(fallback); + return tabIdClaim; +} -async function claimTabId(fallback: string): Promise { +async function claimStoredTabId(fallback: string): Promise { let stored: string | null; try { stored = window.sessionStorage.getItem(TAB_ID_KEY); } catch { - return { id: fallback, release: () => {} }; + return fallback; + } + if (stored && (await holdTabId(stored))) { + return stored; } - const held = stored ? await holdTabId(stored) : null; - const id = held && stored ? stored : fallback; - const release = held ?? (await holdTabId(fallback)) ?? (() => {}); + await holdTabId(fallback); try { - window.sessionStorage.setItem(TAB_ID_KEY, id); + window.sessionStorage.setItem(TAB_ID_KEY, fallback); } catch { - return { id, release }; + return fallback; } - return { id, release }; + return fallback; } -function holdTabId(id: string): Promise<(() => void) | null> { +function holdTabId(id: string): Promise { const locks = typeof navigator === 'undefined' ? undefined : navigator.locks; if (!locks) { - return Promise.resolve(() => {}); + return Promise.resolve(true); } return new Promise((resolve) => { locks .request(`${TAB_ID_KEY}:${id}`, { ifAvailable: true }, (lock) => { - if (!lock) { - resolve(null); - return undefined; - } - return new Promise((release) => resolve(() => release())); + resolve(lock !== null); + return lock ? new Promise(() => {}) : undefined; }) - .catch(() => resolve(() => {})); + .catch(() => resolve(true)); }); } From 42a6226cda6c08b67ac286f7b7aa358709dfd9f5 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:06:32 +0530 Subject: [PATCH 21/31] test(web): compare client ids within a page and with a popup that shares its sessionStorage --- tests/e2e/WebChromiumTest.php | 1 + tests/e2e/languages/web/index.html | 24 +++++++++++++++++------- tests/e2e/languages/web/tab.html | 22 ++++++++++++++++++++++ tests/e2e/languages/web/tests.js | 2 +- 4 files changed, 41 insertions(+), 8 deletions(-) create mode 100644 tests/e2e/languages/web/tab.html diff --git a/tests/e2e/WebChromiumTest.php b/tests/e2e/WebChromiumTest.php index 789f8c7e31..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] diff --git a/tests/e2e/languages/web/index.html b/tests/e2e/languages/web/index.html index e186a742d3..0461f95a3c 100644 --- a/tests/e2e/languages/web/index.html +++ b/tests/e2e/languages/web/index.html @@ -752,16 +752,26 @@ new Promise((resolve) => { tabPush.subscribe(['e2e-whoami'], (m) => resolve(m.data)); }), new Promise((resolve) => setTimeout(() => resolve(''), 5000)), ]); - return { tabPush, id }; + tabPush.close(); + return id; }; - const firstTab = await whoami(); - const secondTab = await whoami(); - firstTab.tabPush.close(); - secondTab.tabPush.close(); + const pageId = await whoami(); + const samePageId = await whoami(); + 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( - firstTab.id.startsWith('e2e-session-user-') && secondTab.id.startsWith('e2e-session-user-') && firstTab.id !== secondTab.id + pageId !== '' && pageId === samePageId && popupId !== '' && popupId !== pageId ? 'Push tab client id:passed' - : `Push tab client id:failed (${firstTab.id}, ${secondTab.id})`, + : `Push tab client id:failed (${pageId}, ${samePageId}, ${popupId})`, ); kickedPush.close(); 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); }); From 2311a422b52c6fab2012bc5959a6296c3c82cca4 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:06:32 +0530 Subject: [PATCH 22/31] test(android): switch with its own user and show what a failed background restore saw --- tests/e2e/languages/android/Tests.kt | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 50b9e7e9ba..f21d4bd18a 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -675,7 +675,9 @@ class ServiceTest { e.message ?: "" } } - val switchClient = pushClient().setSession(e2eSession) + 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") @@ -761,7 +763,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)" }, ) From 02ed5de737752b872585f2d9ffd02d7692dd5e5a Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:18:16 +0530 Subject: [PATCH 23/31] test(mock): publish the e2e-whoami reply at QoS 0 with a literal the library accepts --- mock-server/app/mqtt.php | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mock-server/app/mqtt.php b/mock-server/app/mqtt.php index e5c578891f..e4ffa532d2 100644 --- a/mock-server/app/mqtt.php +++ b/mock-server/app/mqtt.php @@ -155,7 +155,7 @@ 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: Packet::QOS_0)); + \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. From e53ea71790274490afa305f8f3e4c042c049a27b Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:24:45 +0530 Subject: [PATCH 24/31] fix(web): give each concurrent Push connection a distinct client id The tab id was claimed once per page and shared, so two Push instances alive at the same time connected with the same MQTT client id and took over each other's broker session. Now the first live connection takes the tab's stored id (stable across reloads, for replay) and any connection alive at the same time takes a distinct locked id, released on close so a later connection can reuse it. --- templates/web/src/client.ts.twig | 7 +- templates/web/src/services/push.ts.twig | 90 ++++++++++++++++++------- 2 files changed, 71 insertions(+), 26 deletions(-) diff --git a/templates/web/src/client.ts.twig b/templates/web/src/client.ts.twig index 0298f7a3cc..e173886357 100644 --- a/templates/web/src/client.ts.twig +++ b/templates/web/src/client.ts.twig @@ -501,9 +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 one per user and tab when unset: stable across reloads of - * the tab, and different in every other tab, so tabs do not take over each other's session. - * An explicit id shared by several tabs makes them do that. + * 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 da23f1d564..65c281eb50 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -129,6 +129,8 @@ export class Push extends Service { * 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 = ''; private connectedKey = ''; private connectionEpoch = 0; @@ -470,6 +472,10 @@ export class Push extends Service { this.mqtt.end(true); this.mqtt = null; } + if (this.tabId !== '') { + releaseTabId(this.tabId); + this.tabId = ''; + } this.connecting = null; this.generation++; this.everConnected = false; @@ -600,9 +606,12 @@ export class Push extends Service { authMethod === 'appwrite-jwt' ? userIdFromJwt(credential) : userIdFromSession(credential); - const tabId = this.client.config.pushClientId - ? '' - : await claimTabId(this.fallbackClientId); + // Claim once per instance and reuse it across reconnects and credential switches, so this + // connection keeps one id (its replay cursor). Concurrent instances get distinct ids. + if (!this.client.config.pushClientId && this.tabId === '') { + this.tabId = await claimTabId(this.fallbackClientId); + } + const tabId = this.tabId; const clientId = this.client.config.pushClientId || (userId ? `${userId}-${tabId}` : tabId); @@ -854,30 +863,65 @@ function userIdFromSession(session?: string): string { } } -let tabIdClaim: Promise | null = null; - -function claimTabId(fallback: string): Promise { - tabIdClaim ??= claimStoredTabId(fallback); - return tabIdClaim; -} +/** 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; -async function claimStoredTabId(fallback: string): Promise { - let stored: string | null; - try { - stored = window.sessionStorage.getItem(TAB_ID_KEY); - } catch { - return fallback; - } - if (stored && (await holdTabId(stored))) { +/** + * 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; } - await holdTabId(fallback); - try { - window.sessionStorage.setItem(TAB_ID_KEY, fallback); - } catch { - return fallback; + // 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(); } - return fallback; + 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 { From 0b4b0a2ac8610d7af807006a53725228bf2dc893 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:25:27 +0530 Subject: [PATCH 25/31] test(web): assert concurrent Push instances get distinct client ids Open two Push instances at the same time in one page and require the broker to report different client ids for them, alongside the existing reopen-resumes-same-id and popup-differs checks. The prior test closed one instance before opening the next, so it never exercised two live at once. Also close the disconnect-error Push before this block frees its id. --- tests/e2e/languages/web/index.html | 44 ++++++++++++++++++++---------- 1 file changed, 30 insertions(+), 14 deletions(-) diff --git a/tests/e2e/languages/web/index.html b/tests/e2e/languages/web/index.html index 0461f95a3c..d5f4eff8fb 100644 --- a/tests/e2e/languages/web/index.html +++ b/tests/e2e/languages/web/index.html @@ -710,6 +710,7 @@ await kickedPush.subscribe(['e2e-disconnect/kicked-by-test'], () => {}); } catch (e) {} 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); @@ -746,17 +747,31 @@ : `Push concurrent refused:failed (${refused.join(', ')})`, ); - const whoami = async () => { - const tabPush = new Push(new Client().setProject('console').setPushEndpoint('ws://mqtt:8083').setSession(e2eSession)); - const id = await Promise.race([ - new Promise((resolve) => { tabPush.subscribe(['e2e-whoami'], (m) => resolve(m.data)); }), - new Promise((resolve) => setTimeout(() => resolve(''), 5000)), - ]); - tabPush.close(); - return id; - }; - const pageId = await whoami(); - const samePageId = await whoami(); + // 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) => { @@ -769,11 +784,12 @@ new Promise((resolve) => setTimeout(() => resolve(''), 5000)), ]); console.log( - pageId !== '' && pageId === samePageId && popupId !== '' && popupId !== pageId + pageId !== '' && pageId === samePageId && + liveAId !== '' && liveBId !== '' && liveAId !== liveBId && + popupId !== '' && popupId !== pageId ? 'Push tab client id:passed' - : `Push tab client id:failed (${pageId}, ${samePageId}, ${popupId})`, + : `Push tab client id:failed (page ${pageId}, same ${samePageId}, liveA ${liveAId}, liveB ${liveBId}, popup ${popupId})`, ); - kickedPush.close(); }); From 9eea9f1deed4419919b542b430982cf4fb13ee35 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:45:13 +0530 Subject: [PATCH 26/31] fix(web): claim the Push client id once per instance to avoid stranding it A credential switch could start a second open() while the first was still awaiting its client-id claim; both passed the empty-tabId check and claimed, and the loser overwrote this.tabId with its fallback, so the stored id it had claimed was never released and later connections could not resume its replay cursor. Memoize the claim per instance so concurrent opens share one, and release a claim that resolves after close(). --- templates/web/src/services/push.ts.twig | 33 ++++++++++++++++++++++--- 1 file changed, 29 insertions(+), 4 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 65c281eb50..9768623f62 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -131,6 +131,9 @@ export class Push extends Service { private readonly fallbackClientId = randomId(); /** The default client id claimed for this instance's connection, released on close(). */ private tabId = ''; + /** Memoized claim of {@link tabId}, so concurrent open() calls (a credential switch racing the + * first connect) claim it only once and never strand the tab's stored id. */ + private tabIdClaim: Promise | null = null; private connectedKey = ''; private connectionEpoch = 0; @@ -476,6 +479,8 @@ export class Push extends Service { releaseTabId(this.tabId); this.tabId = ''; } + // A claim still in flight releases itself once it resolves (its generation will differ). + this.tabIdClaim = null; this.connecting = null; this.generation++; this.everConnected = false; @@ -589,6 +594,25 @@ export class Push extends Service { : `${wsEndpoint.slice(0, at + separator.length)}push.${wsEndpoint.slice(at + separator.length)}`; } + /** + * The default client id for this connection, claimed once per instance and reused across + * reconnects and credential switches. Memoized so concurrent open() calls (a credential switch + * racing the first connect) claim only once, rather than each claiming and stranding the tab's + * stored id. A claim that resolves after close() releases itself. + */ + private claimClientId(): Promise { + const generation = this.generation; + this.tabIdClaim ??= claimTabId(this.fallbackClientId).then((id) => { + if (this.generation !== generation) { + releaseTabId(id); // closed while claiming: no connection will use this id + return id; + } + this.tabId = id; + return id; + }); + return this.tabIdClaim; + } + private async open(): Promise { const { authMethod, credential } = this.credential(); this.connectedKey = this.credentialKey(); @@ -606,10 +630,11 @@ export class Push extends Service { authMethod === 'appwrite-jwt' ? userIdFromJwt(credential) : userIdFromSession(credential); - // Claim once per instance and reuse it across reconnects and credential switches, so this - // connection keeps one id (its replay cursor). Concurrent instances get distinct ids. - if (!this.client.config.pushClientId && this.tabId === '') { - this.tabId = await claimTabId(this.fallbackClientId); + // 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 (!this.client.config.pushClientId) { + await this.claimClientId(); } const tabId = this.tabId; const clientId = From 959aed35f54ca5555b628fc927f3e18621c7cb20 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 15:54:26 +0530 Subject: [PATCH 27/31] fix(web): make the Push client-id claim safe across close and reopen Capture the generation before awaiting the claim and re-check it after, so a close() during a pending claim stops the open before it connects. Key the memoized claim by generation and chain a reopen's claim after the previous one settles and releases its id, so a close+reopen while a claim is pending takes the tab's stored id back instead of a fallback, and a claim resolving after its generation ended releases its id rather than stranding it. --- templates/web/src/services/push.ts.twig | 52 ++++++++++++++++--------- 1 file changed, 34 insertions(+), 18 deletions(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 9768623f62..394edb9729 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -131,9 +131,10 @@ export class Push extends Service { private readonly fallbackClientId = randomId(); /** The default client id claimed for this instance's connection, released on close(). */ private tabId = ''; - /** Memoized claim of {@link tabId}, so concurrent open() calls (a credential switch racing the - * first connect) claim it only once and never strand the tab's stored id. */ + /** 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; @@ -479,9 +480,9 @@ export class Push extends Service { releaseTabId(this.tabId); this.tabId = ''; } - // A claim still in flight releases itself once it resolves (its generation will differ). - this.tabIdClaim = null; 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(); @@ -595,21 +596,32 @@ export class Push extends Service { } /** - * The default client id for this connection, claimed once per instance and reused across - * reconnects and credential switches. Memoized so concurrent open() calls (a credential switch - * racing the first connect) claim only once, rather than each claiming and stranding the tab's - * stored id. A claim that resolves after close() releases itself. + * 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; - this.tabIdClaim ??= claimTabId(this.fallbackClientId).then((id) => { - if (this.generation !== generation) { - releaseTabId(id); // closed while claiming: no connection will use this id - return id; - } - this.tabId = id; - return id; - }); + 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; } @@ -617,6 +629,7 @@ export class Push extends Service { 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 @@ -632,16 +645,19 @@ export class Push extends Service { : 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. + // 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 || (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.'); From 459d84c4cf24f0d087fddb6804d67faf8c0fca61 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 16:09:10 +0530 Subject: [PATCH 28/31] style(web): wrap the long client-id claim condition for prettier The generated src/services/push.ts failed the web format:check because the memoized-claim guard exceeded prettier's print width. Wrap it as prettier formats it so the Build SDK step passes. --- templates/web/src/services/push.ts.twig | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/templates/web/src/services/push.ts.twig b/templates/web/src/services/push.ts.twig index 394edb9729..35253fb53b 100644 --- a/templates/web/src/services/push.ts.twig +++ b/templates/web/src/services/push.ts.twig @@ -605,7 +605,10 @@ export class Push extends Service { */ private claimClientId(): Promise { const generation = this.generation; - if (this.tabIdClaim === null || this.tabIdClaimGeneration !== generation) { + if ( + this.tabIdClaim === null || + this.tabIdClaimGeneration !== generation + ) { const previous = this.tabIdClaim ?? Promise.resolve(''); this.tabIdClaimGeneration = generation; const claim = previous From 09ed0676d44f732671f0fdc6272229aeee54123f Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 16:17:45 +0530 Subject: [PATCH 29/31] test: log recent SUBSCRIBEs when an Android background restore sees a duplicate (temporary) --- mock-server/app/mqtt.php | 9 +++++++++ tests/e2e/languages/android/Tests.kt | 15 ++++++++++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/mock-server/app/mqtt.php b/mock-server/app/mqtt.php index e4ffa532d2..358d1459bd 100644 --- a/mock-server/app/mqtt.php +++ b/mock-server/app/mqtt.php @@ -65,6 +65,9 @@ class MockHandler implements Handler /** @var array project|client id => its connection, while subscribed to "e2e-replay" */ private array $replayOnline = []; + /** @var list */ + private array $subscribeLog = []; + public function onConnect(Connect $connect, Connection $connection): Connack|Auth { $connection->prefix = $connect->userProperties()['projectId'] ?? ''; @@ -133,6 +136,12 @@ public function onSubscribe(Subscribe $subscribe, Connection $connection): Subac // delivers it when that client id subscribes again, like the real broker's replay position. // A SUBSCRIBE to "e2e-replay-publish" stands in for the server publishing a message. foreach ($subscribe->filters() as $filter) { + $this->subscribeLog[] = \sprintf('%.3f %s %s', \microtime(true), $connection->getClientId(), $filter->topic); + $this->subscribeLog = \array_slice($this->subscribeLog, -40); + if ($filter->topic === 'e2e-subscribe-log') { + $log = \implode("\n", $this->subscribeLog); + \Swoole\Timer::after(100, fn () => $connection->publish('e2e-subscribe-log', $log, qos: 0)); + } if (\str_starts_with($filter->topic, 'e2e-disconnect/')) { $reason = \substr($filter->topic, \strlen('e2e-disconnect/')); \Swoole\Timer::after(100, fn () => $connection->disconnect(Disconnect::NOT_AUTHORIZED, $reason)); diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index f21d4bd18a..1fb6fcd209 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -756,6 +756,19 @@ class ServiceTest { org.robolectric.Shadows.shadowOf(android.os.Looper.getMainLooper()).idle() Thread.sleep(100) } + val subscribeLog = java.util.concurrent.atomic.AtomicReference("") + if (E2EPushReceiver.messages.size != 1) { + val logLatch = java.util.concurrent.CountDownLatch(1) + val logPush = Push(pushClient().setJWT("eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJ1c2VySWQiOiJlMmUtdXNlciJ9.e2e"), context) + runCatching { + logPush.subscribe("e2e-subscribe-log") { message -> + subscribeLog.set(message.data.replace("\n", " | ")) + logLatch.countDown() + } + } + logLatch.await(5, java.util.concurrent.TimeUnit.SECONDS) + logPush.close() + } val posted = org.robolectric.Shadows.shadowOf(notifications).allNotifications.firstOrNull() val postedTitle = posted?.extras?.getCharSequence(android.app.Notification.EXTRA_TITLE)?.toString() val postedText = posted?.extras?.getCharSequence(android.app.Notification.EXTRA_TEXT)?.toString() @@ -763,7 +776,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 (messages: ${E2EPushReceiver.messages.toList()}, title: $postedTitle, text: $postedText)" + "Push background restore:failed (messages: ${E2EPushReceiver.messages.toList()}, title: $postedTitle, text: $postedText, subscribes: ${subscribeLog.get()})" }, ) From 4347feebf0a28bb9023e68c819aa09ac26aea09b Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 16:24:40 +0530 Subject: [PATCH 30/31] Revert "test: log recent SUBSCRIBEs when an Android background restore sees a duplicate (temporary)" This reverts commit 09ed0676d44f732671f0fdc6272229aeee54123f. --- mock-server/app/mqtt.php | 9 --------- tests/e2e/languages/android/Tests.kt | 15 +-------------- 2 files changed, 1 insertion(+), 23 deletions(-) diff --git a/mock-server/app/mqtt.php b/mock-server/app/mqtt.php index 358d1459bd..e4ffa532d2 100644 --- a/mock-server/app/mqtt.php +++ b/mock-server/app/mqtt.php @@ -65,9 +65,6 @@ class MockHandler implements Handler /** @var array project|client id => its connection, while subscribed to "e2e-replay" */ private array $replayOnline = []; - /** @var list */ - private array $subscribeLog = []; - public function onConnect(Connect $connect, Connection $connection): Connack|Auth { $connection->prefix = $connect->userProperties()['projectId'] ?? ''; @@ -136,12 +133,6 @@ public function onSubscribe(Subscribe $subscribe, Connection $connection): Subac // delivers it when that client id subscribes again, like the real broker's replay position. // A SUBSCRIBE to "e2e-replay-publish" stands in for the server publishing a message. foreach ($subscribe->filters() as $filter) { - $this->subscribeLog[] = \sprintf('%.3f %s %s', \microtime(true), $connection->getClientId(), $filter->topic); - $this->subscribeLog = \array_slice($this->subscribeLog, -40); - if ($filter->topic === 'e2e-subscribe-log') { - $log = \implode("\n", $this->subscribeLog); - \Swoole\Timer::after(100, fn () => $connection->publish('e2e-subscribe-log', $log, qos: 0)); - } if (\str_starts_with($filter->topic, 'e2e-disconnect/')) { $reason = \substr($filter->topic, \strlen('e2e-disconnect/')); \Swoole\Timer::after(100, fn () => $connection->disconnect(Disconnect::NOT_AUTHORIZED, $reason)); diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index 1fb6fcd209..f21d4bd18a 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -756,19 +756,6 @@ class ServiceTest { org.robolectric.Shadows.shadowOf(android.os.Looper.getMainLooper()).idle() Thread.sleep(100) } - val subscribeLog = java.util.concurrent.atomic.AtomicReference("") - if (E2EPushReceiver.messages.size != 1) { - val logLatch = java.util.concurrent.CountDownLatch(1) - val logPush = Push(pushClient().setJWT("eyJhbGciOiJub25lIiwidHlwIjoiSldUIn0.eyJ1c2VySWQiOiJlMmUtdXNlciJ9.e2e"), context) - runCatching { - logPush.subscribe("e2e-subscribe-log") { message -> - subscribeLog.set(message.data.replace("\n", " | ")) - logLatch.countDown() - } - } - logLatch.await(5, java.util.concurrent.TimeUnit.SECONDS) - logPush.close() - } val posted = org.robolectric.Shadows.shadowOf(notifications).allNotifications.firstOrNull() val postedTitle = posted?.extras?.getCharSequence(android.app.Notification.EXTRA_TITLE)?.toString() val postedText = posted?.extras?.getCharSequence(android.app.Notification.EXTRA_TEXT)?.toString() @@ -776,7 +763,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 (messages: ${E2EPushReceiver.messages.toList()}, title: $postedTitle, text: $postedText, subscribes: ${subscribeLog.get()})" + "Push background restore:failed (messages: ${E2EPushReceiver.messages.toList()}, title: $postedTitle, text: $postedText)" }, ) From 2731c99aef0d2f293a9642976d251675e5039528 Mon Sep 17 00:00:00 2001 From: ArnabChatterjee20k Date: Tue, 29 Sep 2026 16:24:40 +0530 Subject: [PATCH 31/31] test(android): let the mock's delayed publishes drain before the background test --- tests/e2e/languages/android/Tests.kt | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/e2e/languages/android/Tests.kt b/tests/e2e/languages/android/Tests.kt index f21d4bd18a..a5d061de98 100644 --- a/tests/e2e/languages/android/Tests.kt +++ b/tests/e2e/languages/android/Tests.kt @@ -695,6 +695,7 @@ class ServiceTest { 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"