🐛 Fix signals.is() subscription race (#217)

.policies/no-sleep-test-sync.md

@@ -60,7 +60,7 @@ it("processes items concurrently", function* () {
 From `signals/helpers.test.ts`:
 
 ```typescript
-import { sleep, spawn, withResolvers } from "effection";
+import { spawn, withResolvers } from "effection";
 import { createBooleanSignal, is } from "@effectionx/signals";
 
 it("waits until the value of the stream matches the predicate", function* () {
@@ -76,7 +76,7 @@ it("waits until the value of the stream matches the predicate", function* () {
   });
 
   yield* spawn(function* () {
-    yield* sleep(1);
+    // is() observes the change deterministically; no sleep needed first.
     open.set(true);
   });
 
@@ -115,10 +115,12 @@ it("resolves when the assertion passes within the timeout", function* () {
 
 ### Compliant: is() with signals for state changes
 
-From `stream-helpers/test-helpers/faucet.test.ts`:
+Wait for a signal to reach a target state with `is()`. The producer publishes
+immediately — `is()` observes the matching change deterministically, so no
+`sleep()` is required to "give the consumer time to subscribe":
 
 ```typescript
-import { each, sleep, spawn } from "effection";
+import { each, spawn } from "effection";
 import { createArraySignal, is } from "@effectionx/signals";
 import { useFaucet } from "@effectionx/stream-helpers";
 
@@ -134,7 +136,6 @@ it("creates a faucet that can pour items", function* () {
   });
 
   yield* spawn(function* () {
-    yield* sleep(1);
     yield* faucet.pour([1, 2, 3]);
   });
 

signals/README.md

@@ -102,6 +102,12 @@ matches the predicate. It's useful when you want to wait for a signal to enter a
 specific state. Some of the common use cases are waiting for an array to reach a
 given length or for a boolean signal to become true or false.
 
+`is` observes the signal's current state as well as any matching change, so it
+completes deterministically without the producer having to `yield` or `sleep`
+before publishing. It establishes its subscription before checking the current
+value, so a matching update that lands while it is starting to wait is never
+lost.
+
 ```ts
 import { run, spawn } from "effection";
 import { createBooleanSignal, is } from "@effectionx/signals";
@@ -114,6 +120,7 @@ await run(function* () {
     console.log("floodgates are open!");
   });
 
+  // No sleep or yield needed before publishing.
   open.set(true);
 });
 ```

signals/helpers.test.ts

@@ -1,8 +1,10 @@
 import { describe, it } from "@effectionx/vitest";
 import { expect } from "expect";
-import { sleep, spawn, withResolvers } from "effection";
+import { spawn, withResolvers } from "effection";
+import { timebox } from "@effectionx/timebox";
 
 import { createBooleanSignal } from "./boolean.ts";
+import { createArraySignal } from "./array.ts";
 import { is } from "./helpers.ts";
 
 describe("is", () => {
@@ -19,12 +21,66 @@ describe("is", () => {
     });
 
     yield* spawn(function* () {
-      yield* sleep(0);
       open.set(true);
     });
 
     yield* operation;
 
     expect(update).toEqual(["floodgates are open!"]);
   });
+
+  it("completes immediately when the current value already matches", function* () {
+    const open = yield* createBooleanSignal(true);
+
+    const result = yield* timebox(1000, () =>
+      is(open, (open) => open === true),
+    );
+
+    expect(result.timeout).toEqual(false);
+    expect(open.valueOf()).toEqual(true);
+  });
+
+  it("completes on the first matching update after non-matching updates", function* () {
+    const count = yield* createArraySignal<number>([]);
+
+    const { resolve, operation } = withResolvers<void>();
+    const seen: number[] = [];
+
+    yield* spawn(function* () {
+      yield* is(count, (xs) => xs.length >= 3);
+      seen.push(count.length);
+      resolve();
+    });
+
+    yield* spawn(function* () {
+      count.push(1);
+      count.push(2);
+      count.push(3);
+    });
+
+    yield* operation;
+
+    expect(seen).toEqual([3]);
+    expect(count.valueOf()).toEqual([1, 2, 3]);
+  });
+
+  // Regression for #217: a matching change that lands between observing the
+  // initial state and establishing the subscription must not be lost. The
+  // producer publishes immediately with no sleep or preliminary yield. The
+  // timebox deadline is only a diagnostic bound — it does not coordinate the
+  // producer and consumer — so a lost update surfaces as a timeout failure
+  // rather than a hang.
+  it("completes on a match that lands during subscription setup", function* () {
+    const open = yield* createBooleanSignal(false);
+
+    const result = yield* timebox(1000, function* () {
+      yield* spawn(function* () {
+        open.set(true);
+      });
+      yield* is(open, (open) => open === true);
+    });
+
+    expect(result.timeout).toEqual(false);
+    expect(open.valueOf()).toEqual(true);
+  });
 });

signals/helpers.ts

@@ -1,25 +1,35 @@
-import { each, type Operation } from "effection";
+import type { Operation } from "effection";
 import type { ValueSignal } from "./types.ts";
 
 /**
  * Returns an operation that will wait until the value of the stream matches the predicate.
+ *
+ * The subscription is established before the current value is checked, so a
+ * matching change that lands between observing the initial state and consuming
+ * the stream is not lost. The producer does not need to yield or sleep before
+ * publishing.
+ *
  * @param stream - The stream to check.
  * @param predicate - The predicate to check the value against.
  * @returns An operation that will wait until the value of the stream matches the predicate.
  */
-export function* is<T>(
+export function is<T>(
   stream: ValueSignal<T>,
   predicate: (item: T) => boolean,
 ): Operation<void> {
-  const result = predicate(stream.valueOf());
-  if (result) {
-    return;
-  }
-  for (const value of yield* each(stream)) {
-    const result = predicate(value);
-    if (result) {
-      return;
-    }
-    yield* each.next();
-  }
+  return {
+    *[Symbol.iterator]() {
+      const subscription = yield* stream;
+      if (predicate(stream.valueOf())) {
+        return;
+      }
+      let next = yield* subscription.next();
+      while (!next.done) {
+        if (predicate(next.value)) {
+          return;
+        }
+        next = yield* subscription.next();
+      }
+    },
+  };
 }

signals/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/signals",
   "description": "Reactive signals and computed values for Effection operations",
-  "version": "0.5.3",
+  "version": "0.5.4",
   "keywords": ["reactivity"],
   "type": "module",
   "main": "./dist/mod.js",
@@ -31,6 +31,7 @@
   },
   "sideEffects": false,
   "devDependencies": {
+    "@effectionx/timebox": "workspace:*",
     "@effectionx/vitest": "workspace:*",
     "effection": "^4"
   }

signals/tsconfig.json

@@ -7,6 +7,9 @@
   "include": ["**/*.ts"],
   "exclude": ["**/*.test.ts", "dist"],
   "references": [
+    {
+      "path": "../timebox"
+    },
     {
       "path": "../vitest"
     }

stream-helpers/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/stream-helpers",
   "description": "Type-safe stream operators like filter, map, reduce, and forEach",
-  "version": "0.8.2",
+  "version": "0.8.3",
   "keywords": ["streams"],
   "type": "module",
   "main": "./dist/mod.js",

worker/package.json

@@ -1,7 +1,7 @@
 {
   "name": "@effectionx/worker",
   "description": "Web Worker integration with two-way messaging and graceful shutdown",
-  "version": "0.5.2",
+  "version": "0.5.3",
   "keywords": ["platform"],
   "type": "module",
   "main": "./dist/mod.js",