fix(common): untrack subscription and unsubscription in async pipe (a…

…ngular#50522)

This commit wraps the actual subscription/unsubscription in the `async`
pipe with `untracked`, to ensure that any signal reads/writes which might
take place in `Observable` side effects are not attributed to the template.

Fixes angular#50382

PR Close angular#50522

packages/common/src/pipes/async_pipe.ts

@@ -6,7 +6,7 @@
  * found in the LICENSE file at https://angular.io/license
  */
 
-import {ChangeDetectorRef, EventEmitter, OnDestroy, Pipe, PipeTransform, ɵisPromise, ɵisSubscribable} from '@angular/core';
+import {ChangeDetectorRef, EventEmitter, OnDestroy, Pipe, PipeTransform, untracked, ɵisPromise, ɵisSubscribable} from '@angular/core';
 import {Observable, Subscribable, Unsubscribable} from 'rxjs';
 
 import {invalidPipeArgumentError} from './invalid_pipe_argument_error';
@@ -19,16 +19,24 @@ interface SubscriptionStrategy {
 
 class SubscribableStrategy implements SubscriptionStrategy {
   createSubscription(async: Subscribable<any>, updateLatestValue: any): Unsubscribable {
-    return async.subscribe({
+    // Subscription can be side-effectful, and we don't want any signal reads which happen in the
+    // side effect of the subscription to be tracked by a component's template when that
+    // subscription is triggered via the async pipe. So we wrap the subscription in `untracked` to
+    // decouple from the current reactive context.
+    //
+    // `untracked` also prevents signal _writes_ which happen in the subscription side effect from
+    // being treated as signal writes during the template evaluation (which throws errors).
+    return untracked(() => async.subscribe({
       next: updateLatestValue,
       error: (e: any) => {
         throw e;
       }
-    });
+    }));
   }
 
   dispose(subscription: Unsubscribable): void {
-    subscription.unsubscribe();
+    // See the comment in `createSubscription` above on the use of `untracked`.
+    untracked(() => subscription.unsubscribe());
   }
 }
 

packages/common/test/pipes/async_pipe_spec.ts

@@ -7,9 +7,9 @@
  */
 
 import {AsyncPipe} from '@angular/common';
-import {ChangeDetectorRef, Component, EventEmitter} from '@angular/core';
+import {ChangeDetectorRef, Component, computed, EventEmitter, signal} from '@angular/core';
 import {TestBed} from '@angular/core/testing';
-import {of, Subscribable, Unsubscribable} from 'rxjs';
+import {Observable, of, Subscribable, Unsubscribable} from 'rxjs';
 
 {
   describe('AsyncPipe', () => {
@@ -114,6 +114,29 @@ import {of, Subscribable, Unsubscribable} from 'rxjs';
           expect(firstResult).toBeNaN();
           expect(secondResult).toBeNaN();
         });
+
+        it('should not track signal reads in subscriptions', () => {
+          const trigger = signal(false);
+
+          const obs = new Observable(() => {
+            // Whenever `obs` is subscribed, synchronously read `trigger`.
+            trigger();
+          });
+
+          let trackCount = 0;
+          const tracker = computed(() => {
+            // Subscribe to `obs` within this `computed`. If the subscription side effect runs
+            // within the computed, then changes to `trigger` will invalidate this computed.
+            pipe.transform(obs);
+
+            // The computed returns how many times it's run.
+            return ++trackCount;
+          });
+
+          expect(tracker()).toBe(1);
+          trigger.set(true);
+          expect(tracker()).toBe(1);
+        });
       });
 
       describe('ngOnDestroy', () => {