diff --git a/.policies/no-sleep-test-sync.md b/.policies/no-sleep-test-sync.md index 05b039a0..0d36f299 100644 --- a/.policies/no-sleep-test-sync.md +++ b/.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]); }); diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 521465ee..fb995c29 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -262,6 +262,9 @@ importers: specifier: ^5 version: 5.1.4 devDependencies: + '@effectionx/timebox': + specifier: workspace:* + version: link:../timebox '@effectionx/vitest': specifier: workspace:* version: link:../vitest diff --git a/signals/README.md b/signals/README.md index 9d6c153a..5ed3d9bf 100644 --- a/signals/README.md +++ b/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); }); ``` diff --git a/signals/helpers.test.ts b/signals/helpers.test.ts index af8a1f2e..f99c3a34 100644 --- a/signals/helpers.test.ts +++ b/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,7 +21,6 @@ describe("is", () => { }); yield* spawn(function* () { - yield* sleep(0); open.set(true); }); @@ -27,4 +28,59 @@ describe("is", () => { 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([]); + + const { resolve, operation } = withResolvers(); + 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); + }); }); diff --git a/signals/helpers.ts b/signals/helpers.ts index c822cac6..805bf687 100644 --- a/signals/helpers.ts +++ b/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( +export function is( stream: ValueSignal, predicate: (item: T) => boolean, ): Operation { - 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(); + } + }, + }; } diff --git a/signals/package.json b/signals/package.json index a1320e90..5caa3096 100644 --- a/signals/package.json +++ b/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" } diff --git a/signals/tsconfig.json b/signals/tsconfig.json index 49b10377..08cda0a8 100644 --- a/signals/tsconfig.json +++ b/signals/tsconfig.json @@ -7,6 +7,9 @@ "include": ["**/*.ts"], "exclude": ["**/*.test.ts", "dist"], "references": [ + { + "path": "../timebox" + }, { "path": "../vitest" } diff --git a/stream-helpers/package.json b/stream-helpers/package.json index 810ddeb0..d789af09 100644 --- a/stream-helpers/package.json +++ b/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", diff --git a/worker/package.json b/worker/package.json index bd6e7fa8..c68c4e60 100644 --- a/worker/package.json +++ b/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",