fix(): SSR memory leaks (#2294)

Co-authored-by: Valentin Funk <funk.valentin@gmail.com>

karma.conf.js

@@ -19,6 +19,7 @@ module.exports = function(config) {
       'node_modules/zone.js/dist/jasmine-patch.js',
       'node_modules/zone.js/dist/async-test.js',
       'node_modules/zone.js/dist/fake-async-test.js',
+      'node_modules/zone.js/dist/task-tracking.js',
 
       'node_modules/rxjs/bundles/rxjs.umd.{js,map}',
 

package.json

@@ -9,7 +9,7 @@
     "test:watch": "concurrently \"npm run build:watch\" \"npm run delayed_karma\"",
     "test:debug": "npm run build && karma start",
     "karma": "karma start",
-    "test:universal": "npm run build && cp -R dist/packages-dist test/universal-test/node_modules/angularfire2 && cd test/universal-test && npm run prerender",
+    "test:universal": "npm run build && cd test/universal-test && yarn && npm run prerender",
     "delayed_karma": "sleep 10 && karma start",
     "build": "rimraf dist && node tools/build.js && npm pack ./dist/packages-dist",
     "changelog": "conventional-changelog -p angular -i CHANGELOG.md -s -r 1",

src/analytics/analytics.service.ts

@@ -1,8 +1,8 @@
 import { Injectable, Optional, NgZone, OnDestroy, ComponentFactoryResolver, Inject, PLATFORM_ID, Injector, NgModuleFactory } from '@angular/core';
 import { Subscription, from, Observable, of } from 'rxjs';
-import { filter, withLatestFrom, switchMap, map, tap, pairwise, startWith, groupBy, mergeMap } from 'rxjs/operators';
+import { filter, withLatestFrom, switchMap, map, tap, pairwise, startWith, groupBy, mergeMap, observeOn } from 'rxjs/operators';
 import { Router, NavigationEnd, ActivationEnd, ROUTES } from '@angular/router';
-import { runOutsideAngular } from '@angular/fire';
+import { ɵAngularFireSchedulers } from '@angular/fire';
 import { AngularFireAnalytics, DEBUG_MODE } from './analytics';
 import { User } from 'firebase/app';
 import { Title } from '@angular/platform-browser';
@@ -160,15 +160,17 @@ export class UserTrackingService implements OnDestroy {
         zone: NgZone,
         @Inject(PLATFORM_ID) platformId:Object
     ) {
+        const schedulers = new ɵAngularFireSchedulers(zone);
+
         if (!isPlatformServer(platformId)) {
             zone.runOutsideAngular(() => {
                 // @ts-ignore zap the import in the UMD
                 this.disposable = from(import('firebase/auth')).pipe(
+                    observeOn(schedulers.outsideAngular),
                     switchMap(() => analytics.app),
                     map(app => app.auth()),
                     switchMap(auth => new Observable<User|null>(auth.onAuthStateChanged.bind(auth))),
-                    switchMap(user => analytics.setUserId(user ? user.uid : null!)),
-                    runOutsideAngular(zone)
+                    switchMap(user => analytics.setUserId(user ? user.uid : null!))
                 ).subscribe();
             });
         }

src/analytics/analytics.ts

@@ -1,8 +1,8 @@
 import { Injectable, Inject, Optional, NgZone, InjectionToken, PLATFORM_ID } from '@angular/core';
 import { of } from 'rxjs';
 import { isPlatformBrowser } from '@angular/common';
-import { map, tap, shareReplay, switchMap } from 'rxjs/operators';
-import { FirebaseAppConfig, FirebaseOptions, runOutsideAngular, ɵlazySDKProxy, FirebaseAnalytics, FIREBASE_OPTIONS, FIREBASE_APP_NAME, _firebaseAppFactory } from '@angular/fire';
+import { map, tap, shareReplay, switchMap, observeOn } from 'rxjs/operators';
+import { FirebaseAppConfig, FirebaseOptions, ɵlazySDKProxy, FirebaseAnalytics, FIREBASE_OPTIONS, FIREBASE_APP_NAME, _firebaseAppFactory, ɵAngularFireSchedulers } from '@angular/fire';
 import { analytics, app } from 'firebase';
 
 export interface Config {[key:string]: any};
@@ -58,6 +58,8 @@ export class AngularFireAnalytics {
     zone: NgZone
   ) {
 
+    const schedulers = new ɵAngularFireSchedulers(zone);
+
     if (isPlatformBrowser(platformId)) {
 
       window[DATA_LAYER_NAME] = window[DATA_LAYER_NAME] || [];
@@ -84,15 +86,15 @@ export class AngularFireAnalytics {
     if (debugModeEnabled)   { this.updateConfig({ [DEBUG_MODE_KEY]:  1 }) }
 
     const analytics = of(undefined).pipe(
+      observeOn(schedulers.outsideAngular),
       // @ts-ignore zapping in the UMD in the build script
-      switchMap(() => zone.runOutsideAngular(() => import('firebase/analytics'))),
+      switchMap(() => import('firebase/analytics')),
       map(() => _firebaseAppFactory(options, zone, nameOrConfig)),
       // SEMVER no need to cast once we drop older Firebase
       map(app => <analytics.Analytics>app.analytics()),
       tap(analytics => {
         if (analyticsCollectionEnabled === false) { analytics.setAnalyticsCollectionEnabled(false) }
       }),
-      runOutsideAngular(zone),
       shareReplay({ bufferSize: 1, refCount: false }),
     );
 

src/auth/auth.ts

@@ -1,7 +1,7 @@
 import { Injectable, Inject, Optional, NgZone, PLATFORM_ID } from '@angular/core';
 import { Observable, of, from } from 'rxjs';
 import { switchMap } from 'rxjs/operators';
-import { FIREBASE_OPTIONS, FIREBASE_APP_NAME, FirebaseOptions, FirebaseAppConfig, FirebaseAuth, _firebaseAppFactory, FirebaseZoneScheduler } from '@angular/fire';
+import { FIREBASE_OPTIONS, FIREBASE_APP_NAME, FirebaseOptions, FirebaseAppConfig, FirebaseAuth, _firebaseAppFactory, ɵAngularFireSchedulers, ɵkeepUnstableUntilFirstFactory } from '@angular/fire';
 import { User, auth } from 'firebase/app';
 
 @Injectable()
@@ -38,31 +38,22 @@ export class AngularFireAuth {
     @Inject(FIREBASE_OPTIONS) options:FirebaseOptions,
     @Optional() @Inject(FIREBASE_APP_NAME) nameOrConfig:string|FirebaseAppConfig|null|undefined,
     @Inject(PLATFORM_ID) platformId: Object,
-    private zone: NgZone
+    zone: NgZone
   ) {
-    const scheduler = new FirebaseZoneScheduler(zone, platformId);
+    const keepUnstableUntilFirst = ɵkeepUnstableUntilFirstFactory(new ɵAngularFireSchedulers(zone), platformId);
+
     this.auth = zone.runOutsideAngular(() => {
       const app = _firebaseAppFactory(options, zone, nameOrConfig);
       return app.auth();
     });
 
-    this.authState = scheduler.keepUnstableUntilFirst(
-      scheduler.runOutsideAngular(
-        new Observable(subscriber => {
-          const unsubscribe = this.auth.onAuthStateChanged(subscriber);
-          return { unsubscribe };
-        })
-      )
-    );
+    this.authState = new Observable<User | null>(subscriber => {
+      return zone.runOutsideAngular(() => this.auth.onIdTokenChanged(subscriber));
+    }).pipe(keepUnstableUntilFirst);;
 
-    this.user = scheduler.keepUnstableUntilFirst(
-      scheduler.runOutsideAngular(
-        new Observable(subscriber => {
-          const unsubscribe = this.auth.onIdTokenChanged(subscriber);
-          return { unsubscribe };
-        })
-      )
-    );
+    this.user = new Observable<User | null>(subscriber => {
+      return zone.runOutsideAngular(() => this.auth.onIdTokenChanged(subscriber));
+    }).pipe(keepUnstableUntilFirst);
 
     this.idToken = this.user.pipe(switchMap(user => {
       return user ? from(user.getIdToken()) : of(null)

src/core/angularfire2.spec.ts

@@ -1,10 +1,14 @@
 import { TestBed, inject } from '@angular/core/testing';
-import { PlatformRef, NgModule, CompilerFactory } from '@angular/core';
+import { PlatformRef, NgModule, CompilerFactory, NgZone } from '@angular/core';
 import { FirebaseApp, AngularFireModule } from '@angular/fire';
-import { Subscription } from 'rxjs';
+import { Subscription, Observable, Subject, of } from 'rxjs';
 import { COMMON_CONFIG } from './test-config';
 import { BrowserModule } from '@angular/platform-browser';
 import { database } from 'firebase/app';
+import { ɵZoneScheduler, ɵkeepUnstableUntilFirstFactory, ɵAngularFireSchedulers } from './angularfire2';
+import { ɵPLATFORM_BROWSER_ID, ɵPLATFORM_SERVER_ID } from '@angular/common';
+import { tap  } from 'rxjs/operators';
+import { TestScheduler } from 'rxjs/testing';
 
 describe('angularfire', () => {
   let subscription:Subscription;
@@ -40,6 +44,196 @@ describe('angularfire', () => {
     app.delete().then(done, done.fail);
   });
 
+  describe('ZoneScheduler', () => {
+    it('should execute the scheduled work inside the specified zone', done => {
+      let ngZone = Zone.current.fork({
+        name: 'ngZone'
+      });
+      const rootZone = Zone.current;
+
+      // Mimic real behavior: Executing in Angular
+      ngZone.run(() => {
+        const outsideAngularScheduler = new ɵZoneScheduler(rootZone);
+        outsideAngularScheduler.schedule(() => {
+          expect(Zone.current.name).not.toEqual('ngZone');
+          done();
+        });
+      });
+    });
+
+    it('should execute nested scheduled work inside the specified zone', done => {
+      const testScheduler = new TestScheduler(null!);
+      testScheduler.run(helpers => {
+        const outsideAngularScheduler = new ɵZoneScheduler(Zone.current, testScheduler);
+
+        let ngZone = Zone.current.fork({
+          name: 'ngZone'
+        });
+
+        let callbacksRan = 0;
+
+        // Mimic real behavior: Executing in Angular
+        ngZone.run(() => {
+          outsideAngularScheduler.schedule(() => {
+            callbacksRan++;
+            expect(Zone.current.name).not.toEqual('ngZone');
+
+            ngZone.run(() => {
+              // Sync queueing
+              outsideAngularScheduler.schedule(() => {
+                callbacksRan++;
+                expect(Zone.current.name).not.toEqual('ngZone');
+              });
+
+              // Async (10ms delay) nested scheduling
+              outsideAngularScheduler.schedule(() => {
+                callbacksRan++;
+                expect(Zone.current.name).not.toEqual('ngZone');
+              }, 10);
+
+              // Simulate flush from inside angular-
+              helpers.flush();
+              done();
+              expect(callbacksRan).toEqual(3);
+            })
+          });
+          helpers.flush();
+        });
+      });
+    })
+  })
+
+  describe('keepUnstableUntilFirstFactory', () => {
+    let schedulers: ɵAngularFireSchedulers;
+    let outsideZone: Zone;
+    let insideZone: Zone;
+    beforeAll(() => {
+      outsideZone = Zone.current;
+      insideZone = Zone.current.fork({
+        name: 'ngZone'
+      });
+      const ngZone = {
+        run: insideZone.run.bind(insideZone),
+        runGuarded: insideZone.runGuarded.bind(insideZone),
+        runOutsideAngular: outsideZone.runGuarded.bind(outsideZone),
+        runTask: insideZone.run.bind(insideZone)
+      } as NgZone;
+      schedulers = new ɵAngularFireSchedulers(ngZone);
+    })
+
+    it('should re-schedule emissions asynchronously', done => {
+      const keepUnstableOp = ɵkeepUnstableUntilFirstFactory(schedulers, ɵPLATFORM_SERVER_ID);
+
+      let ran = false;
+      of(null).pipe(
+        keepUnstableOp,
+        tap(() => ran = true)
+      ).subscribe(() => {
+        expect(ran).toEqual(true);
+        done();
+      }, () => fail("Should not error"));
+
+      expect(ran).toEqual(false);
+    });
+
+    [ɵPLATFORM_SERVER_ID, ɵPLATFORM_BROWSER_ID].map(platformId =>
+      it(`should subscribe outside angular and observe inside angular (${platformId})`, done => {
+        const keepUnstableOp = ɵkeepUnstableUntilFirstFactory(schedulers, platformId);
+
+        insideZone.run(() => {
+          new Observable(s => {
+            expect(Zone.current).toEqual(outsideZone);
+            s.next("test");
+          }).pipe(
+            keepUnstableOp,
+            tap(() => {
+              expect(Zone.current).toEqual(insideZone);
+            })
+          ).subscribe(() => {
+            expect(Zone.current).toEqual(insideZone);
+            done();
+          }, err => {
+            fail(err);
+          });
+        });
+      })
+    );
+
+    it('should block until first emission on server platform', done => {
+      const testScheduler = new TestScheduler(null!);
+      testScheduler.run(helpers => {
+        const outsideZone = Zone.current;
+        const taskTrack = new Zone['TaskTrackingZoneSpec']();
+        const insideZone = Zone.current.fork(taskTrack);
+        const trackingSchedulers: ɵAngularFireSchedulers = {
+          ngZone: {
+            run: insideZone.run.bind(insideZone),
+            runGuarded: insideZone.runGuarded.bind(insideZone),
+            runOutsideAngular: outsideZone.runGuarded.bind(outsideZone),
+            runTask: insideZone.run.bind(insideZone)
+          } as NgZone,
+          outsideAngular: new ɵZoneScheduler(outsideZone, testScheduler),
+          insideAngular: new ɵZoneScheduler(insideZone, testScheduler),
+        };
+        const keepUnstableOp = ɵkeepUnstableUntilFirstFactory(trackingSchedulers, ɵPLATFORM_SERVER_ID);
+
+        const s = new Subject();
+        s.pipe(
+          keepUnstableOp,
+        ).subscribe(() => { }, err => { fail(err); }, () => { });
+
+        // Flush to ensure all async scheduled functions are run
+        helpers.flush();
+        // Should now be blocked until first item arrives
+        expect(taskTrack.macroTasks.length).toBe(1);
+        expect(taskTrack.macroTasks[0].source).toBe('firebaseZoneBlock');
+
+        // Emit next item
+        s.next(123);
+        helpers.flush();
+
+        // Should not be blocked after first item
+        expect(taskTrack.macroTasks.length).toBe(0);
+
+        done();
+      });
+    })
+
+    it('should not block on client platform', done => {
+      const testScheduler = new TestScheduler(null!);
+      testScheduler.run(helpers => {
+        const outsideZone = Zone.current;
+        const taskTrack = new Zone['TaskTrackingZoneSpec']();
+        const insideZone = Zone.current.fork(taskTrack);
+        const trackingSchedulers: ɵAngularFireSchedulers = {
+          ngZone: {
+            run: insideZone.run.bind(insideZone),
+            runGuarded: insideZone.runGuarded.bind(insideZone),
+            runOutsideAngular: outsideZone.runGuarded.bind(outsideZone),
+            runTask: insideZone.run.bind(insideZone)
+          } as NgZone,
+          outsideAngular: new ɵZoneScheduler(outsideZone, testScheduler),
+          insideAngular: new ɵZoneScheduler(insideZone, testScheduler),
+        };
+        const keepUnstableOp = ɵkeepUnstableUntilFirstFactory(trackingSchedulers, ɵPLATFORM_BROWSER_ID);
+
+        const s = new Subject();
+        s.pipe(
+          keepUnstableOp,
+        ).subscribe(() => { }, err => { fail(err); }, () => { });
+
+        // Flush to ensure all async scheduled functions are run
+        helpers.flush();
+
+        // Zone should not be blocked
+        expect(taskTrack.macroTasks.length).toBe(0);
+        expect(taskTrack.microTasks.length).toBe(0);
+
+        done();
+      });
+    })
+  })
+
   describe('FirebaseApp', () => {
     it('should provide a FirebaseApp for the FirebaseApp binding', () => {
       expect(typeof app.delete).toBe('function');