fix(core): async EventEmitter error should not prevent stability (#61028)

This commit wraps the `fn` invocation with `try-finally`, ensuring that the pending task (added in [this commit](d5c6ee4)) is always removed.

Prior to this commit, if a subscriber threw an error, it would prevent the application from becoming stable — though this shouldn't happen under normal scenarios because the error should be handled by the RxJS error handler or Angular's error handler.

Errors should not silently prevent the application from being rendered on the server.

PR Close #61028

Author: arturovt

packages/core/src/event_emitter.ts

@@ -176,9 +176,12 @@ class EventEmitter_ extends Subject<any> implements OutputRef<any> {
     return (value: unknown) => {
       const taskId = this.pendingTasks?.add();
       setTimeout(() => {
-        fn(value);
-        if (taskId !== undefined) {
-          this.pendingTasks?.remove(taskId);
+        try {
+          fn(value);
+        } finally {
+          if (taskId !== undefined) {
+            this.pendingTasks?.remove(taskId);
+          }
         }
       });
     };

packages/core/test/event_emitter_spec.ts

@@ -7,10 +7,10 @@
  */
 
 import {TestBed} from '../testing';
-import {filter, tap} from 'rxjs/operators';
+import {filter} from 'rxjs/operators';
 
 import {EventEmitter} from '../src/event_emitter';
-import {ApplicationRef} from '../public_api';
+import {ApplicationRef, NgZone} from '../public_api';
 
 describe('EventEmitter', () => {
   let emitter: EventEmitter<number>;
@@ -206,6 +206,21 @@ describe('EventEmitter', () => {
     expect(emitValue!).toEqual(1);
   });
 
+  it('should not prevent app from becoming stable if subscriber throws an error', async () => {
+    const logs: string[] = [];
+    const ngZone = TestBed.inject(NgZone);
+    const appRef = TestBed.inject(ApplicationRef);
+    appRef.isStable.subscribe((isStable) => logs.push(`isStable=${isStable}`));
+    const emitter = TestBed.runInInjectionContext(() => new EventEmitter<number>(true));
+    emitter.subscribe(() => {
+      throw new Error('Given this is some TypeError...');
+    });
+    // Emit inside the Angular zone so that the error is not captured by Jasmine in `afterAll`.
+    ngZone.run(() => emitter.emit(1));
+    await appRef.whenStable();
+    expect(logs).toEqual(['isStable=true', 'isStable=false', 'isStable=true']);
+  });
+
   // TODO: vsavkin: add tests cases
   // should call dispose on the subscription if generator returns {done:true}
   // should call dispose on the subscription on throw