refactor(core): decouple effects from change detection

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.

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,