Two more tests

Author: simolus3

packages/common/src/client/ConnectionManager.ts

@@ -355,6 +355,8 @@ class ActiveSubscription {
 }
 
 class SyncStreamSubscriptionHandle implements SyncStreamSubscription {
+  private active: boolean = false;
+
   constructor(readonly subscription: ActiveSubscription) {
     subscription.refcount++;
     _finalizer?.register(this, subscription);
@@ -373,8 +375,11 @@ class SyncStreamSubscriptionHandle implements SyncStreamSubscription {
   }
 
   unsubscribe(): void {
-    _finalizer?.unregister(this);
-    this.subscription.decrementRefCount();
+    if (this.active) {
+      this.active = false;
+      _finalizer?.unregister(this);
+      this.subscription.decrementRefCount();
+    }
   }
 }
 

packages/node/tests/sync-stream.test.ts

@@ -146,4 +146,40 @@ describe('Sync streams', () => {
     let status = await statusPromise;
     expect(status.forStream(subscription)).not.toBeNull();
   });
+
+  mockSyncServiceTest('unsubscribing multiple times has no effect', async ({ syncService }) => {
+    const database = await syncService.createDatabase();
+    const a = await database.syncStream('a').subscribe();
+    const aAgain = await database.syncStream('a').subscribe();
+    a.unsubscribe();
+    a.unsubscribe();
+
+    // Pretend the streams are expired - they should still be requested because the core extension extends the lifetime
+    // of streams currently referenced before connecting.
+    await database.execute('UPDATE ps_stream_subscriptions SET expires_at = unixepoch() - 1000');
+    await database.connect(new TestConnector(), defaultOptions);
+
+    expect(syncService.connectedListeners[0]).toMatchObject({
+      streams: {
+        include_defaults: true,
+        subscriptions: [{}]
+      }
+    });
+    aAgain.unsubscribe();
+  });
+
+  mockSyncServiceTest('unsubscribeAll', async ({ syncService }) => {
+    const database = await syncService.createDatabase();
+    const a = await database.syncStream('a').subscribe();
+    database.syncStream('a').unsubscribeAll();
+
+    await database.connect(new TestConnector(), defaultOptions);
+    expect(syncService.connectedListeners[0]).toMatchObject({
+      streams: {
+        include_defaults: true,
+        subscriptions: []
+      }
+    });
+    a.unsubscribe();
+  });
 });