From 6f5e44d2071a7ac40b5d06c1069dcf30bea028bc Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Wed, 19 Aug 2026 14:00:42 +0200 Subject: [PATCH 1/2] fix(shared-internals): apply stream subscription changes made during connect() --- .changeset/brave-otters-listen.md | 8 + .../src/client/ConnectionManager.ts | 15 ++ .../tests/client/ConnectionManager.test.ts | 202 ++++++++++++++++++ 3 files changed, 225 insertions(+) create mode 100644 .changeset/brave-otters-listen.md create mode 100644 packages/shared-internals/tests/client/ConnectionManager.test.ts diff --git a/.changeset/brave-otters-listen.md b/.changeset/brave-otters-listen.md new file mode 100644 index 000000000..b7cd5aecf --- /dev/null +++ b/.changeset/brave-otters-listen.md @@ -0,0 +1,8 @@ +--- +'@powersync/shared-internals': patch +--- + +Fix sync stream subscription changes being lost when they happen while `connect()` is still creating the sync +implementation. A `subscribe()` or `unsubscribe()` in that window was silently dropped and never retried, leaving a +stream syncing after nothing was subscribed to it, or a stream that the client reports as subscribed but which is never +requested (so no data arrives and `waitForFirstSync()` never resolves). diff --git a/packages/shared-internals/src/client/ConnectionManager.ts b/packages/shared-internals/src/client/ConnectionManager.ts index a45b2e3fa..4f9f37880 100644 --- a/packages/shared-internals/src/client/ConnectionManager.ts +++ b/packages/shared-internals/src/client/ConnectionManager.ts @@ -234,6 +234,10 @@ export class ConnectionManager extends BaseObserver { this.pendingConnectionOptions = null; + // An update from subscriptionsMayHaveChanged() is lost between this read and the + // implementation being ready: there is nothing to send it to yet, or it is sent to an + // implementation that discards it. Compare against this and re-send the current set below. + const subscriptionsAtStart = this.subscriptionIdentity; const { sync, onDispose } = await this.options.createSyncImplementation(connector, { subscriptions: this.activeStreams, serializedSchema: schema @@ -242,6 +246,10 @@ export class ConnectionManager extends BaseObserver { this.syncStreamImplementation = sync; this.syncDisposer = onDispose; await this.syncStreamImplementation.waitForReady(); + + if (this.subscriptionIdentity !== subscriptionsAtStart) { + this.syncStreamImplementation.updateSubscriptions(this.activeStreams); + } resolve(); } catch (error) { reject(error); @@ -364,6 +372,13 @@ export class ConnectionManager extends BaseObserver { return [...this.locallyActiveSubscriptions.values()].map((a) => ({ name: a.name, params: a.parameters })); } + /** + * Identifies the current set of subscriptions, for detecting a change across an await. + */ + private get subscriptionIdentity() { + return [...this.locallyActiveSubscriptions.keys()].join('\n'); + } + private subscriptionsMayHaveChanged() { this.syncStreamImplementation?.updateSubscriptions(this.activeStreams); } diff --git a/packages/shared-internals/tests/client/ConnectionManager.test.ts b/packages/shared-internals/tests/client/ConnectionManager.test.ts new file mode 100644 index 000000000..55a93e447 --- /dev/null +++ b/packages/shared-internals/tests/client/ConnectionManager.test.ts @@ -0,0 +1,202 @@ +import { + BaseObserver, + LogLevels, + PowerSyncBackendConnector, + SyncStatus, + SyncStreamConnectionMethod, + createConsoleLogger +} from '@powersync/common'; +import { describe, expect, test } from 'vitest'; +import { ConnectionManager, InternalSubscriptionAdapter } from '../../src/client/ConnectionManager.js'; +import { ResolvedSyncOptions } from '../../src/client/sync/options.js'; +import { + StreamingSyncImplementation, + StreamingSyncImplementationListener, + SubscribedStream +} from '../../src/client/sync/stream/AbstractStreamingSyncImplementation.js'; + +const names = (subscriptions: SubscribedStream[]) => subscriptions.map((s) => s.name); + +class MockStreamingSyncImplementation + extends BaseObserver + implements StreamingSyncImplementation +{ + isConnected = false; + isReady = false; + readonly receivedUpdates: SubscribedStream[][] = []; + + constructor( + readonly snapshot: SubscribedStream[], + private duringReady: (() => void | Promise) | null = null + ) { + super(); + } + + // What this implementation would request from the service: the set connect() created it with, + // unless a later update replaced it. + get effectiveSubscriptions(): SubscribedStream[] { + return this.receivedUpdates.at(-1) ?? this.snapshot; + } + + async connect(_options: ResolvedSyncOptions) { + this.isConnected = true; + } + + async disconnect() { + this.isConnected = false; + } + + async getWriteCheckpoint() { + return '1'; + } + + triggerCrudUpload() {} + + async waitForReady() { + await this.duringReady?.(); + this.duringReady = null; + this.isReady = true; + } + + async waitUntilStatusMatches(_predicate: (status: SyncStatus) => boolean) {} + + updateSubscriptions(subscriptions: SubscribedStream[]) { + if (!this.isReady) { + // Modelled on SharedWebStreamingSyncImplementation, which discards updates while its Comlink + // port is unresolved. + return; + } + this.receivedUpdates.push(subscriptions); + } + + markConnectionMayHaveChanged() {} + + async dispose() { + super.dispose(); + } +} + +const connector: PowerSyncBackendConnector = { + fetchCredentials: async () => null, + uploadData: async () => {} +}; + +const adapter: InternalSubscriptionAdapter = { + firstStatusMatching: async () => {}, + resolveOfflineSyncStatus: async () => {}, + rustSubscriptionsCommand: async () => {} +}; + +// The connect window: after connect() has read the current subscriptions, and before there is a +// ready implementation to send updates to. duringCreate and duringReady run inside it. +function managerWithCreateHook() { + const syncs: MockStreamingSyncImplementation[] = []; + let duringCreate: (() => void | Promise) | null = null; + let duringReady: (() => void | Promise) | null = null; + + const manager = new ConnectionManager({ + logger: createConsoleLogger({ minLevel: LogLevels.error }), + defaultConnectionMethod: SyncStreamConnectionMethod.HTTP, + createSyncImplementation: async (_connector, options) => { + await duringCreate?.(); + const sync = new MockStreamingSyncImplementation(options.subscriptions, duringReady); + duringReady = null; + syncs.push(sync); + return { sync, onDispose: () => {} }; + } + }); + + return { + manager, + syncs, + duringCreate(action: () => void | Promise) { + duringCreate = async () => { + duringCreate = null; + await action(); + }; + }, + duringReady(action: () => void | Promise) { + duringReady = action; + }, + get sync() { + return syncs.at(-1)!; + } + }; +} + +describe('ConnectionManager', () => { + describe('subscription changes inside the connect window', () => { + test('applies an unsubscribe made inside the connect window', async () => { + const harness = managerWithCreateHook(); + await harness.manager.stream(adapter, 'stream_a', null).subscribe(); + const b = await harness.manager.stream(adapter, 'stream_b', null).subscribe(); + + harness.duringCreate(() => b.unsubscribe()); + await harness.manager.connect(connector, {}, {}); + + expect(names(harness.sync.snapshot)).toEqual(['stream_a', 'stream_b']); + expect(names(harness.sync.effectiveSubscriptions)).toEqual(['stream_a']); + + await harness.manager.disconnect(); + }); + + test('applies a subscribe made inside the connect window', async () => { + const harness = managerWithCreateHook(); + await harness.manager.stream(adapter, 'stream_a', null).subscribe(); + + harness.duringCreate(() => harness.manager.stream(adapter, 'stream_b', null).subscribe()); + await harness.manager.connect(connector, {}, {}); + + expect(names(harness.sync.snapshot)).toEqual(['stream_a']); + expect(names(harness.sync.effectiveSubscriptions)).toEqual(['stream_a', 'stream_b']); + + await harness.manager.disconnect(); + }); + + test('sends no update when nothing changed inside the connect window', async () => { + const harness = managerWithCreateHook(); + await harness.manager.stream(adapter, 'stream_a', null).subscribe(); + + await harness.manager.connect(connector, {}, {}); + + // Not an optimisation: SharedSyncImplementation sets subscriptions on its implementation from + // the merged cross-tab set and leaves its own connection manager's activeStreams empty, so an + // unconditional update from here would overwrite that merged set with an empty one. + expect(harness.sync.receivedUpdates).toEqual([]); + expect(names(harness.sync.snapshot)).toEqual(['stream_a']); + + await harness.manager.disconnect(); + }); + + test('waits for the implementation to be ready before sending the update', async () => { + const harness = managerWithCreateHook(); + await harness.manager.stream(adapter, 'stream_a', null).subscribe(); + const b = await harness.manager.stream(adapter, 'stream_b', null).subscribe(); + + // Later in the window: the implementation exists, but is still coming up, so the update it + // receives is discarded and has to be re-sent once it is ready. + harness.duringReady(() => b.unsubscribe()); + await harness.manager.connect(connector, {}, {}); + + expect(names(harness.sync.snapshot)).toEqual(['stream_a', 'stream_b']); + expect(names(harness.sync.effectiveSubscriptions)).toEqual(['stream_a']); + + await harness.manager.disconnect(); + }); + + test('applies a change made inside the connect window of a reconnect', async () => { + const harness = managerWithCreateHook(); + const a = await harness.manager.stream(adapter, 'stream_a', null).subscribe(); + await harness.manager.connect(connector, {}, {}); + + harness.duringCreate(() => a.unsubscribe()); + await harness.manager.connect(connector, {}, {}); + + expect(harness.syncs).toHaveLength(2); + expect(names(harness.sync.snapshot)).toEqual(['stream_a']); + expect(names(harness.sync.effectiveSubscriptions)).toEqual([]); + + await harness.manager.disconnect(); + }); + }); +}); From 897cc30cfd49f1d83ad937d4fdcdef0cb5a36eee Mon Sep 17 00:00:00 2001 From: joshuabrink Date: Thu, 20 Aug 2026 12:05:31 +0200 Subject: [PATCH 2/2] chore: clean up comments --- .changeset/brave-otters-listen.md | 7 +++---- packages/shared-internals/src/client/ConnectionManager.ts | 4 +--- 2 files changed, 4 insertions(+), 7 deletions(-) diff --git a/.changeset/brave-otters-listen.md b/.changeset/brave-otters-listen.md index b7cd5aecf..9fcb73977 100644 --- a/.changeset/brave-otters-listen.md +++ b/.changeset/brave-otters-listen.md @@ -2,7 +2,6 @@ '@powersync/shared-internals': patch --- -Fix sync stream subscription changes being lost when they happen while `connect()` is still creating the sync -implementation. A `subscribe()` or `unsubscribe()` in that window was silently dropped and never retried, leaving a -stream syncing after nothing was subscribed to it, or a stream that the client reports as subscribed but which is never -requested (so no data arrives and `waitForFirstSync()` never resolves). +Fix stream subscription changes made while `connect()` is still bringing up the sync implementation being +lost until the next reconnect. A stream could keep syncing after being unsubscribed, or be reported as +subscribed while never being requested, leaving `waitForFirstSync()` unresolved. diff --git a/packages/shared-internals/src/client/ConnectionManager.ts b/packages/shared-internals/src/client/ConnectionManager.ts index 4f9f37880..bb3d9e634 100644 --- a/packages/shared-internals/src/client/ConnectionManager.ts +++ b/packages/shared-internals/src/client/ConnectionManager.ts @@ -234,9 +234,6 @@ export class ConnectionManager extends BaseObserver { this.pendingConnectionOptions = null; - // An update from subscriptionsMayHaveChanged() is lost between this read and the - // implementation being ready: there is nothing to send it to yet, or it is sent to an - // implementation that discards it. Compare against this and re-send the current set below. const subscriptionsAtStart = this.subscriptionIdentity; const { sync, onDispose } = await this.options.createSyncImplementation(connector, { subscriptions: this.activeStreams, @@ -247,6 +244,7 @@ export class ConnectionManager extends BaseObserver { this.syncDisposer = onDispose; await this.syncStreamImplementation.waitForReady(); + // Subscriptions changed while creating the sync stream implementation, update it now. if (this.subscriptionIdentity !== subscriptionsAtStart) { this.syncStreamImplementation.updateSubscriptions(this.activeStreams); }