refactor(core): decouple effects from change detection (angular#51049)

Previously effects were queued as they became dirty, and this queue was
flushed at various checkpoints during the change detection cycle. The result
was that change detection _was_ the effect runner, and without executing CD,
effects would not execute. This leads a particular tradeoff:

* effects are subject to unidirectional data flow (bad for dx)
* effects don't cause a new round of CD (good/bad depending on use case)
* effects can be used to implement control flow efficiently (desirable)

This commit changes the scheduling mechanism. Effects are now scheduled via
the microtask queue. This changes the tradeoffs:

* effects are no longer limited by unidirectional data flow (easy dx)
* effects registered in the Angular zone will trigger CD after they run
  (same as `Promise.resolve` really)
* the public `effect()` type of effect probably isn't a good building block
  for our built-in control flow, and we'll need a new internal abstraction.

As `effect()` is in developer preview, changing the execution timing is not
considered breaking even though it may impact current users.

PR Close angular#51049

goldens/public-api/core/testing/index.md

@@ -20,6 +20,7 @@ import { PlatformRef } from '@angular/core';
 import { ProviderToken } from '@angular/core';
 import { SchemaMetadata } from '@angular/core';
 import { Type } from '@angular/core';
+import { ɵFlushableEffectRunner } from '@angular/core';
 
 // @public
 export const __core_private_testing_placeholder__ = "";
@@ -29,7 +30,7 @@ export function async(fn: Function): (done: any) => any;
 
 // @public
 export class ComponentFixture<T> {
-    constructor(componentRef: ComponentRef<T>, ngZone: NgZone | null, _autoDetect: boolean);
+    constructor(componentRef: ComponentRef<T>, ngZone: NgZone | null, effectRunner: ɵFlushableEffectRunner | null, _autoDetect: boolean);
     autoDetectChanges(autoDetect?: boolean): void;
     changeDetectorRef: ChangeDetectorRef;
     checkNoChanges(): void;
@@ -110,6 +111,7 @@ export interface TestBed {
     createComponent<T>(component: Type<T>): ComponentFixture<T>;
     // (undocumented)
     execute(tokens: any[], fn: Function, context?: any): any;
+    flushEffects(): void;
     // @deprecated (undocumented)
     get<T>(token: ProviderToken<T>, notFoundValue?: T, flags?: InjectFlags): any;
     // @deprecated (undocumented)

packages/core/src/core_reactivity_export_internal.ts

@@ -23,5 +23,8 @@ export {
   effect,
   EffectRef,
   EffectCleanupFn,
+  EffectScheduler as ɵEffectScheduler,
+  ZoneAwareQueueingScheduler as ɵZoneAwareQueueingScheduler,
+  FlushableEffectRunner as ɵFlushableEffectRunner,
 } from './render3/reactivity/effect';
-// clang-format on    
+// clang-format on

packages/core/src/render3/component_ref.ts

@@ -44,7 +44,7 @@ import {CONTEXT, HEADER_OFFSET, INJECTOR, LView, LViewEnvironment, LViewFlags, T
 import {MATH_ML_NAMESPACE, SVG_NAMESPACE} from './namespaces';
 import {createElementNode, setupStaticAttributes, writeDirectClass} from './node_manipulation';
 import {extractAttrsAndClassesFromSelector, stringifyCSSSelectorList} from './node_selector_matcher';
-import {EffectManager} from './reactivity/effect';
+import {EffectScheduler} from './reactivity/effect';
 import {enterView, getCurrentTNode, getLView, leaveView} from './state';
 import {computeStaticStyling} from './styling/static_styling';
 import {mergeHostAttrs, setUpAttributes} from './util/attrs_utils';
@@ -188,14 +188,13 @@ export class ComponentFactory<T> extends AbstractComponentFactory<T> {
     }
     const sanitizer = rootViewInjector.get(Sanitizer, null);
 
-    const effectManager = rootViewInjector.get(EffectManager, null);
-
     const afterRenderEventManager = rootViewInjector.get(AfterRenderEventManager, null);
 
     const environment: LViewEnvironment = {
       rendererFactory,
       sanitizer,
-      effectManager,
+      // We don't use inline effects (yet).
+      inlineEffectRunner: null,
       afterRenderEventManager,
     };
 

packages/core/src/render3/instructions/change_detection.ts

@@ -48,7 +48,7 @@ export function detectChangesInternal<T>(
 
       // One final flush of the effects queue to catch any effects created in `ngAfterViewInit` or
       // other post-order hooks.
-      environment.effectManager?.flush();
+      environment.inlineEffectRunner?.flush();
 
       // Invoke all callbacks registered via `after*Render`, if needed.
       afterRenderEventManager?.end();
@@ -117,7 +117,7 @@ export function refreshView<T>(
   // since they were assigned. We do not want to execute lifecycle hooks in that mode.
   const isInCheckNoChangesPass = ngDevMode && isInCheckNoChangesMode();
 
-  !isInCheckNoChangesPass && lView[ENVIRONMENT].effectManager?.flush();
+  !isInCheckNoChangesPass && lView[ENVIRONMENT].inlineEffectRunner?.flush();
 
   enterView(lView);
   try {

packages/core/src/render3/interfaces/view.ts

@@ -12,7 +12,7 @@ import {DehydratedView} from '../../hydration/interfaces';
 import {SchemaMetadata} from '../../metadata/schema';
 import {Sanitizer} from '../../sanitization/sanitizer';
 import type {ReactiveLViewConsumer} from '../reactive_lview_consumer';
-import type {EffectManager} from '../reactivity/effect';
+import type {FlushableEffectRunner} from '../reactivity/effect';
 import type {AfterRenderEventManager} from '../after_render_hooks';
 
 import {LContainer} from './container';
@@ -372,7 +372,7 @@ export interface LViewEnvironment {
   sanitizer: Sanitizer|null;
 
   /** Container for reactivity system `effect`s. */
-  effectManager: EffectManager|null;
+  inlineEffectRunner: FlushableEffectRunner|null;
 
   /** Container for after render hooks */
   afterRenderEventManager: AfterRenderEventManager|null;

packages/core/src/render3/reactivity/effect.ts

@@ -7,11 +7,14 @@
  */
 
 import {assertInInjectionContext} from '../../di/contextual';
+import {InjectionToken} from '../../di/injection_token';
 import {Injector} from '../../di/injector';
 import {inject} from '../../di/injector_compatibility';
 import {ɵɵdefineInjectable} from '../../di/interface/defs';
+import {ErrorHandler} from '../../error_handler';
 import {DestroyRef} from '../../linker/destroy_ref';
-import {Watch, watch} from '../../signals';
+import {isInNotificationPhase, watch, Watch, WatchCleanupFn, WatchCleanupRegisterFn} from '../../signals';
+
 
 /**
  * An effect can, optionally, register a cleanup function. If registered, the cleanup is executed
@@ -27,73 +30,201 @@ export type EffectCleanupFn = () => void;
  */
 export type EffectCleanupRegisterFn = (cleanupFn: EffectCleanupFn) => void;
 
+export interface SchedulableEffect {
+  run(): void;
+  creationZone: unknown;
+}
+
 /**
- * Tracks all effects registered within a given application and runs them via `flush`.
+ * Not public API, which guarantees `EffectScheduler` only ever comes from the application root
+ * injector.
  */
-export class EffectManager {
-  private all = new Set<Watch>();
-  private queue = new Map<Watch, Zone|null>();
-
-  create(
-      effectFn: (onCleanup: (cleanupFn: EffectCleanupFn) => void) => void,
-      destroyRef: DestroyRef|null, allowSignalWrites: boolean): EffectRef {
-    const zone = (typeof Zone === 'undefined') ? null : Zone.current;
-    const w = watch(effectFn, (watch) => {
-      if (!this.all.has(watch)) {
-        return;
-      }
-
-      this.queue.set(watch, zone);
-    }, allowSignalWrites);
+export const APP_EFFECT_SCHEDULER = new InjectionToken('', {
+  providedIn: 'root',
+  factory: () => inject(EffectScheduler),
+});
 
-    this.all.add(w);
-
-    // Effects start dirty.
-    w.notify();
+/**
+ * A scheduler which manages the execution of effects.
+ */
+export abstract class EffectScheduler {
+  /**
+   * Schedule the given effect to be executed at a later time.
+   *
+   * It is an error to attempt to execute any effects synchronously during a scheduling operation.
+   */
+  abstract scheduleEffect(e: SchedulableEffect): void;
 
-    let unregisterOnDestroy: (() => void)|undefined;
+  /** @nocollapse */
+  static ɵprov = /** @pureOrBreakMyCode */ ɵɵdefineInjectable({
+    token: EffectScheduler,
+    providedIn: 'root',
+    factory: () => new ZoneAwareMicrotaskScheduler(),
+  });
+}
 
-    const destroy = () => {
-      w.cleanup();
-      unregisterOnDestroy?.();
-      this.all.delete(w);
-      this.queue.delete(w);
-    };
+/**
+ * Interface to an `EffectScheduler` capable of running scheduled effects synchronously.
+ */
+export interface FlushableEffectRunner {
+  /**
+   * Run any scheduled effects.
+   */
+  flush(): void;
+}
 
-    unregisterOnDestroy = destroyRef?.onDestroy(destroy);
+/**
+ * An `EffectScheduler` which is capable of queueing scheduled effects per-zone, and flushing them
+ * as an explicit operation.
+ */
+export class ZoneAwareQueueingScheduler implements EffectScheduler, FlushableEffectRunner {
+  private queuedEffectCount = 0;
+  private queues = new Map<Zone|null, Set<SchedulableEffect>>();
 
-    return {
-      destroy,
-    };
-  }
+  scheduleEffect(handle: SchedulableEffect): void {
+    const zone = handle.creationZone as Zone | null;
+    if (!this.queues.has(zone)) {
+      this.queues.set(zone, new Set());
+    }
 
-  flush(): void {
-    if (this.queue.size === 0) {
+    const queue = this.queues.get(zone)!;
+    if (queue.has(handle)) {
       return;
     }
+    this.queuedEffectCount++;
+    queue.add(handle);
+  }
 
-    for (const [watch, zone] of this.queue) {
-      this.queue.delete(watch);
-      if (zone) {
-        zone.run(() => watch.run());
-      } else {
-        watch.run();
+  /**
+   * Run all scheduled effects.
+   *
+   * Execution order of effects within the same zone is guaranteed to be FIFO, but there is no
+   * ordering guarantee between effects scheduled in different zones.
+   */
+  flush(): void {
+    while (this.queuedEffectCount > 0) {
+      for (const [zone, queue] of this.queues) {
+        // `zone` here must be defined.
+        if (zone === null) {
+          this.flushQueue(queue);
+        } else {
+          zone.run(() => this.flushQueue(queue));
+        }
       }
     }
   }
 
-  get isQueueEmpty(): boolean {
-    return this.queue.size === 0;
+  private flushQueue(queue: Set<SchedulableEffect>): void {
+    for (const handle of queue) {
+      queue.delete(handle);
+      this.queuedEffectCount--;
+
+      // TODO: what happens if this throws an error?
+      handle.run();
+    }
   }
 
   /** @nocollapse */
   static ɵprov = /** @pureOrBreakMyCode */ ɵɵdefineInjectable({
-    token: EffectManager,
+    token: ZoneAwareQueueingScheduler,
     providedIn: 'root',
-    factory: () => new EffectManager(),
+    factory: () => new ZoneAwareQueueingScheduler(),
   });
 }
 
+/**
+ * A wrapper around `ZoneAwareQueueingScheduler` that schedules flushing via the microtask queue
+ * when.
+ */
+export class ZoneAwareMicrotaskScheduler implements EffectScheduler {
+  private hasQueuedFlush = false;
+  private delegate = new ZoneAwareQueueingScheduler();
+  private flushTask = () => {
+    // Leave `hasQueuedFlush` as `true` so we don't queue another microtask if more effects are
+    // scheduled during flushing. The flush of the `ZoneAwareQueueingScheduler` delegate is
+    // guaranteed to empty the queue.
+    this.delegate.flush();
+    this.hasQueuedFlush = false;
+
+    // This is a variable initialization, not a method.
+    // tslint:disable-next-line:semicolon
+  };
+
+  scheduleEffect(handle: SchedulableEffect): void {
+    this.delegate.scheduleEffect(handle);
+
+    if (!this.hasQueuedFlush) {
+      queueMicrotask(this.flushTask);
+      this.hasQueuedFlush = true;
+    }
+  }
+}
+
+/**
+ * Core reactive node for an Angular effect.
+ *
+ * `EffectHandle` combines the reactive graph's `Watch` base node for effects with the framework's
+ * scheduling abstraction (`EffectScheduler`) as well as automatic cleanup via `DestroyRef` if
+ * available/requested.
+ */
+class EffectHandle implements EffectRef, SchedulableEffect {
+  private alive = true;
+  unregisterOnDestroy: (() => void)|undefined;
+  protected watcher: Watch;
+
+  constructor(
+      private scheduler: EffectScheduler,
+      private effectFn: (onCleanup: EffectCleanupRegisterFn) => void,
+      public creationZone: Zone|null, destroyRef: DestroyRef|null,
+      private errorHandler: ErrorHandler|null, allowSignalWrites: boolean) {
+    this.watcher =
+        watch((onCleanup) => this.runEffect(onCleanup), () => this.schedule(), allowSignalWrites);
+    this.unregisterOnDestroy = destroyRef?.onDestroy(() => this.destroy());
+  }
+
+  private runEffect(onCleanup: WatchCleanupRegisterFn): void {
+    if (!this.alive) {
+      // Running a destroyed effect is a no-op.
+      return;
+    }
+    if (ngDevMode && isInNotificationPhase()) {
+      throw new Error(`Schedulers cannot synchronously execute effects while scheduling.`);
+    }
+
+    try {
+      this.effectFn(onCleanup);
+    } catch (err) {
+      this.errorHandler?.handleError(err);
+    }
+  }
+
+  run(): void {
+    this.watcher.run();
+  }
+
+  private schedule(): void {
+    if (!this.alive) {
+      return;
+    }
+
+    this.scheduler.scheduleEffect(this);
+  }
+
+  notify(): void {
+    this.watcher.notify();
+  }
+
+  destroy(): void {
+    this.alive = false;
+
+    this.watcher.cleanup();
+    this.unregisterOnDestroy?.();
+
+    // Note: if the effect is currently scheduled, it's not un-scheduled, and so the scheduler will
+    // retain a reference to it. Attempting to execute it will be a no-op.
+  }
+}
+
 /**
  * A global reactive effect, which can be manually destroyed.
  *
@@ -147,7 +278,16 @@ export function effect(
     options?: CreateEffectOptions): EffectRef {
   !options?.injector && assertInInjectionContext(effect);
   const injector = options?.injector ?? inject(Injector);
-  const effectManager = injector.get(EffectManager);
+  const errorHandler = injector.get(ErrorHandler, null, {optional: true});
   const destroyRef = options?.manualCleanup !== true ? injector.get(DestroyRef) : null;
-  return effectManager.create(effectFn, destroyRef, !!options?.allowSignalWrites);
+
+  const handle = new EffectHandle(
+      injector.get(APP_EFFECT_SCHEDULER), effectFn,
+      (typeof Zone === 'undefined') ? null : Zone.current, destroyRef, errorHandler,
+      options?.allowSignalWrites ?? false);
+
+  // Effects start dirty.
+  handle.notify();
+
+  return handle;
 }

packages/core/src/signals/index.ts

@@ -9,8 +9,8 @@
 export {defaultEquals, isSignal, Signal, SIGNAL, ValueEqualityFn} from './src/api';
 export {computed, CreateComputedOptions} from './src/computed';
 export {setThrowInvalidWriteToSignalError} from './src/errors';
-export {consumerAfterComputation, consumerBeforeComputation, consumerDestroy, producerAccessed, producerNotifyConsumers, producerUpdatesAllowed, producerUpdateValueVersion, REACTIVE_NODE, ReactiveNode, setActiveConsumer} from './src/graph';
+export {consumerAfterComputation, consumerBeforeComputation, consumerDestroy, isInNotificationPhase, producerAccessed, producerNotifyConsumers, producerUpdatesAllowed, producerUpdateValueVersion, REACTIVE_NODE, ReactiveNode, setActiveConsumer} from './src/graph';
 export {CreateSignalOptions, setPostSignalSetFn, signal, WritableSignal} from './src/signal';
 export {untracked} from './src/untracked';
-export {Watch, watch, WatchCleanupFn} from './src/watch';
+export {Watch, watch, WatchCleanupFn, WatchCleanupRegisterFn} from './src/watch';
 export {setAlternateWeakRefImpl} from './src/weak_ref';

packages/core/src/signals/src/graph.ts

@@ -26,6 +26,10 @@ export function setActiveConsumer(consumer: ReactiveNode|null): ReactiveNode|nul
   return prev;
 }
 
+export function isInNotificationPhase(): boolean {
+  return inNotificationPhase;
+}
+
 export const REACTIVE_NODE = {
   version: 0 as Version,
   dirty: false,