Fix stale get reads notify cbs

integration-tests/create-state/state-modifications.test.ts

@@ -171,6 +171,37 @@ describe("state modifications", () => {
   });
 
   describe("glitch-free push-pull", () => {
+    test("allows source subscribers to synchronously read the latest derived value", () => {
+      const source = new StateCore(1);
+      const doubled = source.map((value) => value * 2);
+
+      expect(doubled.get()).toBe(2);
+
+      const reads: number[] = [];
+      source.on(() => reads.push(doubled.get() as number));
+
+      source.set(2);
+
+      expect(reads).toEqual([4]);
+      expect(doubled.get()).toBe(4);
+    });
+
+    test("allows derived subscribers to synchronously read the latest downstream value", () => {
+      const source = new StateCore(1);
+      const doubled = source.map((value) => value * 2);
+      const quadrupled = doubled.map((value) => value * 2);
+
+      expect(quadrupled.get()).toBe(4);
+
+      const reads: number[] = [];
+      doubled.on(() => reads.push(quadrupled.get() as number));
+
+      source.set(2);
+
+      expect(reads).toEqual([8]);
+      expect(quadrupled.get()).toBe(8);
+    });
+
     test("does not emit an intermediate value in a deep, uneven diamond graph", () => {
       const source = new StateCore(1);
       const short = source.map((value) => value + 100);

src/create-state/state-core.ts

@@ -22,6 +22,8 @@ type CoreOptions<T> = {
 
 const defaultEquality: EqualityFn<any> = (value1, value2) => value1 === value2;
 
+type PendingNotification = { node: StateCore<any>; value: any; prevValue: any };
+
 class StateCore<T> {
   private _value: CoreValue<T>;
   private _prevValue: CoreValue<T> = emptyValue;
@@ -67,13 +69,22 @@ class StateCore<T> {
     const prevValue = this._value;
     this._prevValue = prevValue;
     this._value = newValue;
-    this.notifySubscribers(newValue, prevValue);
 
     this._children.forEach((child) => {
       child._dirty = true;
     });
 
-    this.flush();
+    const pendingNotifications = this.flush();
+
+    // technically, we can notify subscribers earlier, but then synchronous reading
+    // would return stale state values.
+    this.notifySubscribers(newValue, prevValue);
+
+    // we intentionally process derived notifications after direct subscribers to
+    // the signal
+    pendingNotifications.forEach(({ node, value, prevValue }) => {
+      node.notifySubscribers(value, prevValue);
+    });
   }
 
   update(fn: (currentValue: CoreValue<T>) => T) {
@@ -205,7 +216,9 @@ class StateCore<T> {
     return { changed };
   }
 
-  private flush() {
+  private flush(): PendingNotification[] {
+    const pendingNotifications: PendingNotification[] = [];
+
     /**
      * We utilize a min-heap to ensure that we don't trigger notifications
      * before all the previous necessary work is done. Otherwise it can
@@ -256,12 +269,18 @@ class StateCore<T> {
 
       if (!changed) continue;
 
-      child.notifySubscribers(value, child._prevValue);
+      pendingNotifications.push({
+        node: child,
+        prevValue: child._prevValue,
+        value,
+      });
       child._children.forEach((grandchild) => {
         grandchild._dirty = true;
         enqueue(grandchild);
       });
     }
+
+    return pendingNotifications;
   }
 
   private notifySubscribers(value: CoreValue<T>, prevValue: CoreValue<T>) {