Reactivity hardening (sequenced bench bisect) (#207)

Author: jlukic

packages/reactivity/src/dependency.js

@@ -42,13 +42,7 @@ export class Dependency {
     }
   }
 
-  // called after flush
-  cleanUp(reaction) {
-    this.subscribers.delete(reaction);
-  }
-
-  // identical for now but called from stop()
-  unsubscribe(reaction) {
+  remove(reaction) {
     this.subscribers.delete(reaction);
   }
 }

packages/reactivity/src/reaction.js

@@ -85,27 +85,25 @@ export class Reaction {
     Scheduler.current = this;
     try {
       for (const dep of this.dependencies) {
-        dep.cleanUp(this);
+        dep.remove(this);
       }
       this.dependencies.clear();
       this.callback(this);
-      this.firstRun = false;
     }
     finally {
+      // firstRun advances even on throw so a re-invalidation re-tracks from a known baseline
+      this.firstRun = false;
       Scheduler.current = previousReaction;
     }
   }
 
   invalidate(context) {
-    // Set this reaction as active and about to be run
-    this.active = true;
-
-    // Pass through trace for debugging
+    if (!this.active) {
+      return;
+    }
     if (context) {
       this.addContext(context);
     }
-
-    // Schedule this reaction to occur in the next flush
     Scheduler.scheduleReaction(this);
   }
 
@@ -114,7 +112,9 @@ export class Reaction {
       return;
     }
     this.active = false;
-    this.dependencies.forEach(dep => dep.unsubscribe(this));
+    Scheduler.pendingReactions.delete(this);
+    this.dependencies.forEach(dep => dep.remove(this));
+    this.dependencies.clear();
     this.fireCleanups();
   }
 

packages/reactivity/src/scheduler.js

@@ -7,6 +7,7 @@ export class Scheduler {
   static pendingReactions = new Set();
   static afterFlushCallbacks = [];
   static isFlushScheduled = false;
+  static isFlushing = false;
 
   static scheduleReaction(reaction) {
     Scheduler.pendingReactions.add(reaction);
@@ -24,43 +25,66 @@ export class Scheduler {
 
   static flush() {
     Scheduler.isFlushScheduled = false;
+    Scheduler.isFlushing = true;
     // capture first error but finish draining so one faulty reaction or afterFlush callback can't jam the queue
     let firstError;
     let iterations = 0;
-    while (Scheduler.pendingReactions.size > 0) {
-      if (++iterations > Scheduler.maxFlushIterations) {
-        console.error('Reactive cycle detected: flush exceeded maximum iterations');
-        Scheduler.pendingReactions.clear();
-        break;
-      }
-      const reactions = [...Scheduler.pendingReactions];
-      Scheduler.pendingReactions.clear();
-      for (let i = 0; i < reactions.length; i++) {
-        try {
-          reactions[i].run();
+    try {
+      // alternate: drain all pending reactions, then run one snapshot of afterFlush callbacks.
+      // afterFlushes registered during the batch land in the next alternation, which drains
+      // any reactions they queued before the next batch runs.
+      while (Scheduler.pendingReactions.size > 0 || Scheduler.afterFlushCallbacks.length > 0) {
+        while (Scheduler.pendingReactions.size > 0) {
+          if (++iterations > Scheduler.maxFlushIterations) {
+            console.error('Reactive cycle detected: flush exceeded maximum iterations');
+            Scheduler.pendingReactions.clear();
+            Scheduler.afterFlushCallbacks.length = 0;
+            break;
+          }
+          // set-swap: avoid the per-pass array spread. new invalidations land in the next pass.
+          const toRun = Scheduler.pendingReactions;
+          Scheduler.pendingReactions = new Set();
+          for (const r of toRun) {
+            if (r.stopped) { continue; }
+            try {
+              r.run();
+            }
+            catch (e) {
+              if (!firstError) { firstError = e; }
+            }
+          }
         }
-        catch (e) {
-          if (!firstError) { firstError = e; }
+        if (Scheduler.afterFlushCallbacks.length > 0) {
+          if (++iterations > Scheduler.maxFlushIterations) {
+            console.error('Reactive cycle detected: flush exceeded maximum iterations');
+            Scheduler.afterFlushCallbacks.length = 0;
+            break;
+          }
+          const callbacks = Scheduler.afterFlushCallbacks;
+          Scheduler.afterFlushCallbacks = [];
+          for (let i = 0; i < callbacks.length; i++) {
+            try {
+              callbacks[i]();
+            }
+            catch (e) {
+              if (!firstError) { firstError = e; }
+            }
+          }
         }
       }
     }
-
-    const callbacks = Scheduler.afterFlushCallbacks;
-    Scheduler.afterFlushCallbacks = [];
-    for (let i = 0; i < callbacks.length; i++) {
-      try {
-        callbacks[i]();
-      }
-      catch (e) {
-        if (!firstError) { firstError = e; }
-      }
+    finally {
+      Scheduler.isFlushing = false;
     }
 
     if (firstError) { throw firstError; }
   }
 
   static afterFlush(callback) {
     Scheduler.afterFlushCallbacks.push(callback);
+    if (!Scheduler.isFlushing) {
+      Scheduler.scheduleFlush();
+    }
   }
 
   static getSource() {

packages/reactivity/src/signal.js

@@ -145,52 +145,44 @@ export class Signal {
   // derive a new signal from this signal's value
   derive(computeFn, options = {}) {
     const derivedSignal = new Signal(undefined, options);
+    // weak so the reaction's closure doesn't pin derived through source.dep.subscribers
+    const derivedRef = new WeakRef(derivedSignal);
+    const source = this;
 
-    // check if signal has been garbage collected
-    // if it has we need to clean up reaction
-    const sourceRef = new WeakRef(this);
-
-    // Create reaction that updates the derived signal
     const reaction = Reaction.create(() => {
-      const source = sourceRef.deref();
-      if (!source) {
-        reaction.stop(); // Auto-cleanup if source is gone
+      const d = derivedRef.deref();
+      if (!d) {
+        reaction.stop();
         return;
       }
-      const result = computeFn(source.get());
-      derivedSignal.set(result);
+      d.set(computeFn(source.get()));
     });
 
-    // scope to parent reaction when called from inside one
     if (Reaction.current) {
       Reaction.current.onCleanup(() => reaction.stop());
     }
 
-    // Store reaction reference for potential cleanup
-    derivedSignal._derivedReaction = reaction;
-
     return derivedSignal;
   }
 
   // static method for computing from multiple signals
   static computed(computeFn, options = {}) {
     const computedSignal = new Signal(undefined, options);
+    const computedRef = new WeakRef(computedSignal);
 
-    // Create reaction that updates the computed signal
-    // No WeakRef needed - computed signal and reaction have same lifecycle
     const reaction = Reaction.create(() => {
-      const result = computeFn();
-      computedSignal.set(result);
+      const c = computedRef.deref();
+      if (!c) {
+        reaction.stop();
+        return;
+      }
+      c.set(computeFn());
     });
 
-    // scope to parent reaction when called from inside one
     if (Reaction.current) {
       Reaction.current.onCleanup(() => reaction.stop());
     }
 
-    // Store reaction reference for potential cleanup
-    computedSignal._computedReaction = reaction;
-
     return computedSignal;
   }
 

packages/reactivity/test/unit/internals.test.js

@@ -119,6 +119,27 @@ describe('Scheduler — flush', () => {
     errorSpy.mockRestore();
   });
 
+  // the cycle cap spans both queues with a unified iteration counter — a reaction
+  // that schedules an afterFlush that re-invalidates the reaction must also hit it
+  it('breaks a reaction↔afterFlush cycle and logs an error', () => {
+    const errorSpy = vi.spyOn(console, 'error').mockImplementation(() => {});
+    const trigger = new Signal(0);
+
+    Reaction.create(() => {
+      trigger.get();
+      Reaction.afterFlush(() => trigger.set(trigger.peek() + 1));
+    });
+
+    trigger.set(1);
+    Reaction.flush();
+
+    expect(errorSpy).toHaveBeenCalled();
+    expect(errorSpy.mock.calls.some(c => /cycle detected/i.test(c[0]))).toBe(true);
+    expect(Scheduler.pendingReactions.size).toBe(0);
+    expect(Scheduler.afterFlushCallbacks.length).toBe(0);
+    errorSpy.mockRestore();
+  });
+
   // exception in one reaction must not silently swallow others in the same batch
   // either it propagates or framework isolates each, this test pins which
   it('continues processing remaining reactions when one throws', () => {
@@ -210,8 +231,7 @@ describe('Scheduler — flush', () => {
     expect(settled).toHaveBeenCalledTimes(1);
   });
 
-  // late-registered afterFlush queues for the next flush, otherwise self-registering callbacks would infinite-loop
-  it('does not run afterFlush callbacks registered DURING afterFlush in the same pass', () => {
+  it('drains afterFlush callbacks registered during afterFlush in the same flush', () => {
     let runCount = 0;
     const recursive = () => {
       runCount++;
@@ -222,10 +242,45 @@ describe('Scheduler — flush', () => {
     Reaction.afterFlush(recursive);
 
     Reaction.flush();
-    expect(runCount).toBe(1);
+    expect(runCount).toBe(5);
+  });
+
+  it('schedules a flush when afterFlush registers with no pending work', async () => {
+    const cb = vi.fn();
+    Reaction.afterFlush(cb);
+    await Promise.resolve();
+    expect(cb).toHaveBeenCalledTimes(1);
+  });
+
+  it('keeps draining when an afterFlush callback throws', () => {
+    const survivor = vi.fn();
+    Reaction.afterFlush(() => {
+      throw new Error('boom');
+    });
+    Reaction.afterFlush(survivor);
+
+    expect(() => Reaction.flush()).toThrow('boom');
+    expect(survivor).toHaveBeenCalledTimes(1);
+  });
+
+  it('drains reactions queued by an afterFlush callback before the next callback runs', () => {
+    const source = new Signal('initial');
+    const derived = new Signal('initial');
+
+    Reaction.create(() => {
+      derived.set(`reaction-saw-${source.get()}`);
+    });
+
+    let observedInCb2;
+    Reaction.afterFlush(() => {
+      source.set('updated-by-cb1');
+      Reaction.afterFlush(() => {
+        observedInCb2 = derived.peek();
+      });
+    });
 
     Reaction.flush();
-    expect(runCount).toBe(2);
+    expect(observedInCb2).toBe('reaction-saw-updated-by-cb1');
   });
 });
 
@@ -262,6 +317,32 @@ describe('Scheduler — current reaction context', () => {
     expect(s.hasDependents()).toBe(false);
   });
 
+  it('advances firstRun even when the callback throws, so re-invalidation tracks fresh deps', () => {
+    const trigger = new Signal(0);
+    let throwOnce = true;
+    const callback = vi.fn();
+    let reaction;
+
+    // Reaction.create throws because the first run does, so capture the instance via the callback arg
+    expect(() => {
+      Reaction.create((r) => {
+        reaction = r;
+        trigger.get();
+        callback(r.firstRun);
+        if (throwOnce) {
+          throwOnce = false;
+          throw new Error('first run throws');
+        }
+      });
+    }).toThrow('first run throws');
+
+    expect(reaction.firstRun).toBe(false);
+
+    trigger.set(1);
+    Reaction.flush();
+    expect(callback).toHaveBeenLastCalledWith(false);
+  });
+
   it('nonreactive nests correctly — restores outer reaction when inner returns', () => {
     const outer = new Signal('outer');
     const inner = new Signal('inner');
@@ -467,6 +548,33 @@ describe('Reaction.guard', () => {
     outer.stop();
     expect(counter.dependency.subscribers.size).toBe(0);
   });
+
+  it('propagates value changes after the first f() throws', () => {
+    const source = new Signal('first');
+    let throwOnce = true;
+    const downstream = vi.fn();
+
+    expect(() => {
+      Reaction.create(() => {
+        const v = Reaction.guard(() => {
+          const x = source.get();
+          if (throwOnce) {
+            throwOnce = false;
+            throw new Error('first run throws');
+          }
+          return x;
+        });
+        downstream(v);
+      });
+    }).toThrow('first run throws');
+
+    expect(downstream).not.toHaveBeenCalled();
+
+    // a signal change re-fires the inner guard, which now succeeds and propagates upward
+    source.set('second');
+    Reaction.flush();
+    expect(downstream).toHaveBeenCalledWith('second');
+  });
 });
 
 /*******************************