1import { describe, expect, it, vi } from "@effect/vitest";
2import { EnvironmentId } from "@t3tools/contracts";
3import * as Cause from "effect/Cause";
4import * as Deferred from "effect/Deferred";
5import * as Effect from "effect/Effect";
6import * as Exit from "effect/Exit";
7import * as Fiber from "effect/Fiber";
8import * as Latch from "effect/Latch";
9import * as Layer from "effect/Layer";
10import * as Option from "effect/Option";
11import * as Queue from "effect/Queue";
12import * as Schema from "effect/Schema";
13import * as Stream from "effect/Stream";
14import * as SubscriptionRef from "effect/SubscriptionRef";
15import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
16
17import {
18 AVAILABLE_CONNECTION_STATE,
19 ConnectionBlockedError,
20 ConnectionTransientError,
21 PrimaryConnectionTarget,
22 type PreparedConnection,
23 type SupervisorConnectionState,
24} from "../connection/model.ts";
25import * as EnvironmentRegistry from "../connection/registry.ts";
26import * as EnvironmentSupervisor from "../connection/supervisor.ts";
27import { EnvironmentRpcUnavailableError } from "../rpc/client.ts";
28import type * as RpcSession from "../rpc/session.ts";
29import type { QueryPersistence } from "./queryPersistence.ts";
30import {
31 environmentRpcKey,
32 createAtomCommandScheduler,
33 createEnvironmentQueryAtomFamily,
34 createEnvironmentSubscriptionAtomFamily,
35 createRuntimeCommand,
36 scheduleAtomCommandEffect,
37 executeAtomCommand,
38 executeAtomQuery,
39 isAtomCommandInterrupted,
40 mapAtomCommandResult,
41 runAtomCommand,
42 settleAsyncResult,
43 settlePromise,
44 squashAtomCommandFailure,
45} from "./runtime.ts";
46
47const QUERY_ENVIRONMENT = new PrimaryConnectionTarget({
48 environmentId: EnvironmentId.make("query-environment"),
49 label: "Query environment",
50 httpBaseUrl: "https://query.example.test",
51 wsBaseUrl: "wss://query.example.test",
52});
53
54const QUERY_RPC_SESSION = {} as RpcSession.RpcSession;
55
56class TestQueryError extends Schema.TaggedError<TestQueryError>()("TestQueryError", {
57 message: Schema.String,
58}) {}
59
60const OFFLINE_QUERY_FAILURE = new ConnectionTransientError({
61 reason: "transport",
62 detail: "Relay is unavailable.",
63});
64
65const BLOCKED_QUERY_FAILURE = new ConnectionBlockedError({
66 reason: "permission",
67 detail: "Access denied.",
68});
69
70function queryConnectionState(
71 overrides: Partial<SupervisorConnectionState> = {},
72): SupervisorConnectionState {
73 return {
74 ...AVAILABLE_CONNECTION_STATE,
75 desired: true,
76 network: "online",
77 phase: "connected",
78 attempt: 1,
79 generation: 1,
80 ...overrides,
81 };
82}
83
84const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* <A, E>(
85 execute: Effect.Effect<A, E>,
86 {
87 refreshTrigger,
88 idleTtlMs,
89 persistence,
90 initialState,
91 isFinal,
92 }: {
93 readonly refreshTrigger?: Atom.Atom<unknown>;
94 readonly isFinal?: (value: A) => boolean;
95 readonly idleTtlMs?: number;
96 readonly persistence?: QueryPersistence<A>;
97 readonly initialState?: SupervisorConnectionState;
98 } = {},
99) {
100 const supervisorState = yield* SubscriptionRef.make(initialState ?? queryConnectionState());
101 const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION));
102 const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
103 target: QUERY_ENVIRONMENT,
104 state: supervisorState,
105 session: supervisorSession,
106 prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()),
107 connect: Effect.void,
108 disconnect: Effect.void,
109 retryNow: Effect.void,
110 } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
111 const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (_environmentId, effect) =>
112 Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
113 const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
114 _environmentId,
115 stream,
116 ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
117 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
118 run,
119 followStream,
120 stateChanges: () => SubscriptionRef.changes(supervisorState),
121 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
122 const runtime = Atom.runtime(
123 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
124 );
125 const family = createEnvironmentQueryAtomFamily(runtime, {
126 label: "test.environment-query",
127 staleTimeMs: 60_000,
128 ...(persistence === undefined ? {} : { persistence }),
129 execute: () => execute,
130 ...(refreshTrigger === undefined ? {} : { refreshTrigger: () => refreshTrigger }),
131 ...(idleTtlMs === undefined ? {} : { idleTtlMs }),
132 ...(isFinal === undefined ? {} : { isFinal }),
133 });
134
135 return {
136 atom: family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined }),
137 supervisorSession,
138 supervisorState,
139 };
140});
141
142/**
143 * Reads a query's settled value without leaving it mounted. `getResult` resumes from inside the
144 * query's listener, so it only unsubscribes once the reader yields.
145 */
146const readEnvironmentQuery = <A, E>(
147 registry: AtomRegistry.AtomRegistry,
148 atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
149) =>
150 AtomRegistry.getResult(registry, atom, { suspendOnWaiting: true }).pipe(
151 Effect.tap(() => Effect.yieldNow),
152 );
153
154const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* <A, E>(
155 atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
156) {
157 const registry = AtomRegistry.make();
158 const unmount = registry.mount(atom);
159 yield* Effect.addFinalizer(() =>
160 Effect.sync(() => {
161 unmount();
162 registry.dispose();
163 }),
164 );
165 return registry;
166});
167
168describe("settleAsyncResult", () => {
169 it("preserves successful values and typed failures", async () => {
170 const success = await settleAsyncResult(() => Promise.resolve(Exit.succeed("done")));
171 expect(AsyncResult.isSuccess(success)).toBe(true);
172 if (AsyncResult.isSuccess(success)) {
173 expect(success.value).toBe("done");
174 }
175
176 const expectedFailure = new Error("request failed");
177 const failure = await settleAsyncResult(() => Promise.resolve(Exit.fail(expectedFailure)));
178 expect(AsyncResult.isFailure(failure)).toBe(true);
179 if (AsyncResult.isFailure(failure)) {
180 expect(Cause.hasDies(failure.cause)).toBe(false);
181 expect(Cause.squash(failure.cause)).toBe(expectedFailure);
182 }
183 });
184
185 it("encodes thrown and rejected promises as defects", async () => {
186 const thrownDefect = new Error("thrown defect");
187 const thrown = await settleAsyncResult<void, never>(() => {
188 throw thrownDefect;
189 });
190 expect(AsyncResult.isFailure(thrown)).toBe(true);
191 if (AsyncResult.isFailure(thrown)) {
192 expect(Cause.hasDies(thrown.cause)).toBe(true);
193 expect(Cause.squash(thrown.cause)).toBe(thrownDefect);
194 }
195
196 const rejectedDefect = new Error("rejected defect");
197 const rejected = await settleAsyncResult<void, never>(() => Promise.reject(rejectedDefect));
198 expect(AsyncResult.isFailure(rejected)).toBe(true);
199 if (AsyncResult.isFailure(rejected)) {
200 expect(Cause.hasDies(rejected.cause)).toBe(true);
201 expect(Cause.squash(rejected.cause)).toBe(rejectedDefect);
202 }
203 });
204});
205
206describe("atom command result helpers", () => {
207 it("maps successful command values", () => {
208 const result = mapAtomCommandResult(AsyncResult.success(2), (value) => value * 3);
209
210 expect(result._tag).toBe("Success");
211 if (result._tag === "Success") {
212 expect(result.value).toBe(6);
213 }
214 });
215
216 it("preserves failures while mapping", () => {
217 const result = mapAtomCommandResult(
218 AsyncResult.failure<number, string>(Cause.fail("nope")),
219 (value) => value * 3,
220 );
221
222 expect(result._tag).toBe("Failure");
223 if (result._tag === "Failure") {
224 expect(Cause.squash(result.cause)).toBe("nope");
225 }
226 });
227
228 it("distinguishes interruption from other failures", () => {
229 const interrupted = AsyncResult.failure(Cause.interrupt(1));
230 const failed = AsyncResult.failure(Cause.fail("nope"));
231
232 expect(isAtomCommandInterrupted(interrupted)).toBe(true);
233 expect(isAtomCommandInterrupted(failed)).toBe(false);
234 expect(squashAtomCommandFailure(failed)).toBe("nope");
235 });
236
237 it("settles raw promise boundaries as successes or defects", async () => {
238 const success = await settlePromise(() => Promise.resolve("done"));
239 expect(success._tag).toBe("Success");
240
241 const defect = new Error("raw promise rejected");
242 const failure = await settlePromise(() => Promise.reject(defect));
243 expect(failure._tag).toBe("Failure");
244 if (failure._tag === "Failure") {
245 expect(Cause.hasDies(failure.cause)).toBe(true);
246 expect(Cause.squash(failure.cause)).toBe(defect);
247 }
248 });
249
250 it("reports expected failures and defects through separate policies", async () => {
251 const warnings: string[] = [];
252 const errors: string[] = [];
253 const reporter = {
254 warn: (message: string) => {
255 warnings.push(message);
256 },
257 error: (message: string) => {
258 errors.push(message);
259 },
260 };
261
262 await executeAtomCommand(() => Promise.resolve(Exit.fail("nope")), { label: "save" }, reporter);
263 await executeAtomCommand(
264 () => Promise.resolve(Exit.fail("ignored")),
265 { label: "quiet save", reportFailure: false },
266 reporter,
267 );
268 await executeAtomCommand(
269 () => Promise.reject(new Error("defect")),
270 { label: "quiet save", reportFailure: false },
271 reporter,
272 );
273 await executeAtomCommand(
274 () => Promise.resolve(Exit.interrupt(1)),
275 { label: "interrupted" },
276 reporter,
277 );
278
279 expect(warnings).toEqual(["[atom-command] save failed"]);
280 expect(errors).toEqual(["[atom-command] quiet save defected"]);
281 });
282});
283
284describe("environmentRpcKey", () => {
285 it("isolates subscription state by environment and cwd", () => {
286 const environmentId = EnvironmentId.make("environment-1");
287 const originalTarget = {
288 environmentId,
289 input: { cwd: "/repo/original" },
290 };
291 const nextTarget = {
292 environmentId,
293 input: { cwd: "/repo/next" },
294 };
295
296 expect(environmentRpcKey(originalTarget)).not.toBe(environmentRpcKey(nextTarget));
297 expect(environmentRpcKey(originalTarget)).toBe(environmentRpcKey({ ...originalTarget }));
298 expect(
299 environmentRpcKey({
300 environmentId: EnvironmentId.make("environment-2"),
301 input: originalTarget.input,
302 }),
303 ).not.toBe(environmentRpcKey(originalTarget));
304 });
305});
306
307describe("finite environment subscription lifecycle", () => {
308 it.effect(
309 "exposes early values and settles on the terminal value without waiting for a reconnect",
310 () =>
311 Effect.scoped(
312 Effect.gen(function* () {
313 const releaseFinal = yield* Deferred.make<void>();
314 const sawEarly = yield* Deferred.make<void>();
315 const sawFinal = yield* Deferred.make<void>();
316 const closed = yield* Deferred.make<void>();
317 const stream = Stream.concat(
318 Stream.succeed({ complete: false, count: 1 }),
319 Stream.fromEffect(
320 Deferred.await(releaseFinal).pipe(Effect.as({ complete: true, count: 123 })),
321 ),
322 ).pipe(Stream.ensuring(Deferred.succeed(closed, undefined)));
323 // Following a registered environment normally remains open for replacements.
324 const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
325 followStream: (_environmentId: EnvironmentId, source: typeof stream) =>
326 Stream.concat(source, Stream.never),
327 } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
328 const runtime = Atom.runtime(
329 Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
330 );
331 const family = createEnvironmentSubscriptionAtomFamily(runtime, {
332 label: "test.finite-subscription",
333 subscribe: () => stream,
334 completeWhen: (value) => value.complete,
335 idleTtlMs: 0,
336 });
337 const atom = family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined });
338 const registry = AtomRegistry.make();
339 const observed: Array<
340 AsyncResult.AsyncResult<{ complete: boolean; count: number }, unknown>
341 > = [];
342 const unsubscribe = registry.subscribe(
343 atom,
344 (result) => {
345 observed.push(result);
346 if (AsyncResult.isSuccess(result)) {
347 if (!result.value.complete) Deferred.doneUnsafe(sawEarly, Exit.void);
348 if (result.value.complete && !result.waiting)
349 Deferred.doneUnsafe(sawFinal, Exit.void);
350 }
351 },
352 { immediate: true },
353 );
354 yield* Effect.addFinalizer(() =>
355 Effect.sync(() => {
356 unsubscribe();
357 registry.dispose();
358 }),
359 );
360 yield* Deferred.await(sawEarly);
361 const early = registry.get(atom);
362 expect(AsyncResult.isSuccess(early) && early.value.count).toBe(1);
363 expect(early.waiting).toBe(true);
364 yield* Deferred.succeed(releaseFinal, undefined);
365 yield* Deferred.await(sawFinal);
366 yield* Deferred.await(closed);
367 const final = registry.get(atom);
368 expect(AsyncResult.isSuccess(final) && final.value.count).toBe(123);
369 expect(final.waiting).toBe(false);
370 expect(observed.some((result) => AsyncResult.isFailure(result))).toBe(false);
371 }),
372 ),
373 );
374});
375
376describe("environment query lifecycle", () => {
377 it.effect(
378 "retries an interrupted query without exposing a failure during session replacement",
379 () =>
380 Effect.scoped(
381 Effect.gen(function* () {
382 const firstStarted = Latch.makeUnsafe();
383 const failFirst = Latch.makeUnsafe();
384 const firstSettled = Latch.makeUnsafe();
385 const unavailable = new EnvironmentRpcUnavailableError({
386 environmentId: QUERY_ENVIRONMENT.environmentId,
387 message: "Query environment is not connected.",
388 });
389 let executions = 0;
390 const execute = Effect.suspend(() => {
391 executions += 1;
392 if (executions > 1) {
393 return Effect.succeed("recovered");
394 }
395 firstStarted.openUnsafe();
396 return failFirst.await.pipe(
397 Effect.andThen(Effect.fail(unavailable)),
398 Effect.ensuring(
399 Effect.sync(() => {
400 firstSettled.openUnsafe();
401 }),
402 ),
403 );
404 });
405 const harness = yield* makeEnvironmentQueryHarness(execute);
406 const registry = AtomRegistry.make();
407 const observed: Array<AsyncResult.AsyncResult<string, unknown>> = [];
408 const unsubscribe = registry.subscribe(
409 harness.atom,
410 (result) => {
411 observed.push(result);
412 },
413 { immediate: true },
414 );
415 yield* Effect.addFinalizer(() =>
416 Effect.sync(() => {
417 unsubscribe();
418 registry.dispose();
419 }),
420 );
421
422 yield* firstStarted.await;
423 yield* SubscriptionRef.set(harness.supervisorSession, Option.none());
424 yield* Effect.yieldNow;
425 failFirst.openUnsafe();
426 yield* firstSettled.await;
427 yield* Effect.yieldNow;
428
429 expect(observed.some(AsyncResult.isFailure)).toBe(false);
430
431 yield* SubscriptionRef.set(
432 harness.supervisorState,
433 queryConnectionState({ phase: "connecting", stage: "preparing" }),
434 );
435 yield* Effect.yieldNow;
436 yield* SubscriptionRef.set(
437 harness.supervisorState,
438 queryConnectionState({
439 phase: "backoff",
440 stage: null,
441 lastFailure: new ConnectionTransientError({
442 reason: "transport",
443 detail: "Relay session is reconnecting.",
444 }),
445 retryAt: 1,
446 }),
447 );
448 yield* Effect.yieldNow;
449
450 yield* SubscriptionRef.set(harness.supervisorSession, Option.some(QUERY_RPC_SESSION));
451 yield* SubscriptionRef.set(
452 harness.supervisorState,
453 queryConnectionState({ generation: 2 }),
454 );
455 expect(
456 yield* AtomRegistry.getResult(registry, harness.atom, {
457 suspendOnWaiting: true,
458 }),
459 ).toBe("recovered");
460 }),
461 ),
462 );
463
464 it.effect.each([
465 {
466 condition: "after a manual disconnect",
467 state: queryConnectionState({
468 desired: false,
469 phase: "available",
470 stage: null,
471 attempt: 0,
472 }),
473 expectedFailure: null,
474 },
475 {
476 condition: "while the environment is offline",
477 state: queryConnectionState({
478 network: "offline",
479 phase: "offline",
480 stage: null,
481 lastFailure: OFFLINE_QUERY_FAILURE,
482 }),
483 expectedFailure: OFFLINE_QUERY_FAILURE,
484 },
485 {
486 condition: "when connection recovery is blocked",
487 state: queryConnectionState({
488 phase: "blocked",
489 stage: null,
490 lastFailure: BLOCKED_QUERY_FAILURE,
491 }),
492 expectedFailure: BLOCKED_QUERY_FAILURE,
493 },
494 ] as const)("settles as unavailable $condition", ({ state, expectedFailure }) =>
495 Effect.scoped(
496 Effect.gen(function* () {
497 const harness = yield* makeEnvironmentQueryHarness(Effect.succeed("connected"));
498 const registry = yield* mountEnvironmentQuery(harness.atom);
499
500 expect(
501 yield* AtomRegistry.getResult(registry, harness.atom, {
502 suspendOnWaiting: true,
503 }),
504 ).toBe("connected");
505
506 yield* SubscriptionRef.set(harness.supervisorState, state);
507 yield* Effect.yieldNow;
508
509 const result = registry.get(harness.atom);
510 expect(AsyncResult.isFailure(result)).toBe(true);
511 expect(result.waiting).toBe(false);
512 if (AsyncResult.isFailure(result) && expectedFailure !== null) {
513 expect(Cause.squash(result.cause)).toBe(expectedFailure);
514 }
515 }),
516 ),
517 );
518
519 it.effect("keeps a genuine query failure settled while reconnecting", () =>
520 Effect.scoped(
521 Effect.gen(function* () {
522 const expectedFailure = new TestQueryError({ message: "Query failed." });
523 const firstStarted = Latch.makeUnsafe();
524 const failFirst = Latch.makeUnsafe();
525 const refreshStarted = Latch.makeUnsafe();
526 const finishRefresh = Latch.makeUnsafe();
527 let executions = 0;
528 const execute = Effect.suspend(() => {
529 executions += 1;
530 if (executions === 1) {
531 firstStarted.openUnsafe();
532 return failFirst.await.pipe(Effect.andThen(Effect.fail(expectedFailure)));
533 }
534 refreshStarted.openUnsafe();
535 return finishRefresh.await.pipe(Effect.as("recovered"));
536 });
537 const harness = yield* makeEnvironmentQueryHarness(execute);
538 const registry = yield* mountEnvironmentQuery(harness.atom);
539
540 yield* firstStarted.await;
541 failFirst.openUnsafe();
542 const initial = yield* AtomRegistry.getResult(registry, harness.atom, {
543 suspendOnWaiting: true,
544 }).pipe(Effect.exit);
545 expect(Exit.isFailure(initial)).toBe(true);
546 if (Exit.isFailure(initial)) {
547 expect(Cause.squash(initial.cause)).toBe(expectedFailure);
548 }
549
550 yield* SubscriptionRef.set(
551 harness.supervisorState,
552 queryConnectionState({ phase: "connecting", stage: "opening" }),
553 );
554 yield* Effect.yieldNow;
555
556 const refreshing = registry.get(harness.atom);
557 expect(AsyncResult.isFailure(refreshing)).toBe(true);
558 expect(refreshing.waiting).toBe(true);
559 if (AsyncResult.isFailure(refreshing)) {
560 expect(Cause.squash(refreshing.cause)).toBe(expectedFailure);
561 }
562
563 yield* SubscriptionRef.set(
564 harness.supervisorState,
565 queryConnectionState({ generation: 2 }),
566 );
567 yield* refreshStarted.await;
568 finishRefresh.openUnsafe();
569 expect(
570 yield* AtomRegistry.getResult(registry, harness.atom, {
571 suspendOnWaiting: true,
572 }),
573 ).toBe("recovered");
574 }),
575 ),
576 );
577
578 it.effect("retains the last successful value while reconnecting", () =>
579 Effect.scoped(
580 Effect.gen(function* () {
581 const refreshStarted = Latch.makeUnsafe();
582 const finishRefresh = Latch.makeUnsafe();
583 let executions = 0;
584 const execute = Effect.suspend(() => {
585 executions += 1;
586 if (executions === 1) {
587 return Effect.succeed("cached");
588 }
589 refreshStarted.openUnsafe();
590 return finishRefresh.await.pipe(Effect.as("updated"));
591 });
592 const harness = yield* makeEnvironmentQueryHarness(execute);
593 const registry = yield* mountEnvironmentQuery(harness.atom);
594
595 expect(
596 yield* AtomRegistry.getResult(registry, harness.atom, {
597 suspendOnWaiting: true,
598 }),
599 ).toBe("cached");
600
601 yield* SubscriptionRef.set(
602 harness.supervisorState,
603 queryConnectionState({ phase: "connecting", stage: "opening" }),
604 );
605 yield* Effect.yieldNow;
606 expect(registry.get(harness.atom)).toMatchObject({
607 _tag: "Success",
608 value: "cached",
609 waiting: true,
610 });
611
612 yield* SubscriptionRef.set(
613 harness.supervisorState,
614 queryConnectionState({
615 phase: "backoff",
616 stage: null,
617 lastFailure: new ConnectionTransientError({
618 reason: "transport",
619 detail: "Retrying.",
620 }),
621 retryAt: 1,
622 }),
623 );
624 yield* Effect.yieldNow;
625 expect(registry.get(harness.atom)).toMatchObject({
626 _tag: "Success",
627 value: "cached",
628 waiting: true,
629 });
630
631 yield* SubscriptionRef.set(
632 harness.supervisorState,
633 queryConnectionState({ generation: 2 }),
634 );
635 yield* refreshStarted.await;
636 expect(registry.get(harness.atom)).toMatchObject({
637 _tag: "Success",
638 value: "cached",
639 waiting: true,
640 });
641
642 finishRefresh.openUnsafe();
643 expect(
644 yield* AtomRegistry.getResult(registry, harness.atom, {
645 suspendOnWaiting: true,
646 }),
647 ).toBe("updated");
648 }),
649 ),
650 );
651
652 it.effect("refreshes on its trigger while read, and once on its next read when idle", () =>
653 Effect.scoped(
654 Effect.gen(function* () {
655 let executions = 0;
656 const trigger = Atom.make(0).pipe(Atom.keepAlive);
657 const harness = yield* makeEnvironmentQueryHarness(
658 Effect.sync(() => (executions += 1)),
659 { refreshTrigger: trigger },
660 );
661 const registry = AtomRegistry.make();
662 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
663 const read = readEnvironmentQuery(registry, harness.atom);
664
665 const unmount = registry.mount(harness.atom);
666 expect(yield* read).toBe(1);
667 registry.set(trigger, 1);
668 expect(yield* read).toBe(2);
669
670 // Unmounted, the query stays alive on its idle TTL, but nothing reads it.
671 unmount();
672 registry.set(trigger, 2);
673 registry.set(trigger, 3);
674 expect(executions).toBe(2);
675
676 expect(yield* read).toBe(3);
677 }),
678 ),
679 );
680
681 it.effect("stops refreshing on its trigger once the value is final", () =>
682 Effect.scoped(
683 Effect.gen(function* () {
684 let executions = 0;
685 const trigger = Atom.make(0).pipe(Atom.keepAlive);
686 const harness = yield* makeEnvironmentQueryHarness(
687 Effect.sync(() => (executions += 1)),
688 { refreshTrigger: trigger, isFinal: (value) => value >= 2 },
689 );
690 const registry = AtomRegistry.make();
691 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
692 const read = readEnvironmentQuery(registry, harness.atom);
693
694 registry.mount(harness.atom);
695 expect(yield* read).toBe(1);
696 registry.set(trigger, 1);
697 expect(yield* read).toBe(2);
698 registry.set(trigger, 2);
699 expect(yield* read).toBe(2);
700 expect(executions).toBe(2);
701 }),
702 ),
703 );
704
705 it.effect("keeps an idle query when its trigger recomputes to the same value", () =>
706 Effect.scoped(
707 Effect.gen(function* () {
708 let executions = 0;
709 const settings = Atom.make({ theme: "dark" }).pipe(Atom.keepAlive);
710 const trigger = Atom.make((get) => JSON.stringify([get(settings) !== null]));
711 const harness = yield* makeEnvironmentQueryHarness(
712 Effect.sync(() => (executions += 1)),
713 { refreshTrigger: trigger },
714 );
715 const registry = AtomRegistry.make();
716 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
717 const read = readEnvironmentQuery(registry, harness.atom);
718
719 const unmount = registry.mount(harness.atom);
720 expect(yield* read).toBe(1);
721 unmount();
722 registry.set(settings, { theme: "light" });
723 expect(yield* read).toBe(1);
724 expect(executions).toBe(1);
725 }),
726 ),
727 );
728
729 it.effect("refreshes an idle query the registry swept and rebuilt after its trigger moved", () =>
730 Effect.scoped(
731 Effect.gen(function* () {
732 // Only the registry's idle-TTL sweeps run on these; effects keep their own scheduler.
733 vi.useFakeTimers({ toFake: ["setTimeout", "clearTimeout", "Date"] });
734 yield* Effect.addFinalizer(() => Effect.sync(() => vi.useRealTimers()));
735 let executions = 0;
736 const trigger = Atom.make(0).pipe(Atom.keepAlive);
737 const harness = yield* makeEnvironmentQueryHarness(
738 Effect.sync(() => (executions += 1)),
739 { refreshTrigger: trigger, idleTtlMs: 1_000 },
740 );
741 const registry = AtomRegistry.make({ timeoutResolution: 100 });
742 yield* Effect.addFinalizer(() => Effect.sync(() => registry.dispose()));
743 const read = readEnvironmentQuery(registry, harness.atom);
744
745 const unmount = registry.mount(harness.atom);
746 expect(yield* read).toBe(1);
747 unmount();
748 // Lets the registry start the query node's idle TTL.
749 yield* Effect.yieldNow;
750 vi.advanceTimersByTime(500);
751 registry.set(trigger, 1);
752 yield* Effect.yieldNow;
753 // Sweeps the query node but not the data it read, whose idle TTL started with the trigger.
754 vi.advanceTimersByTime(700);
755 expect(executions).toBe(1);
756
757 expect(yield* read).toBe(2);
758 }),
759 ),
760 );
761});
762
763describe("Atom.fn mutation semantics", () => {
764 it.effect("interrupts the previous invocation when the same mutation atom is written again", () =>
765 Effect.gen(function* () {
766 const firstLatch = Latch.makeUnsafe();
767 const secondLatch = Latch.makeUnsafe();
768 const interrupted: string[] = [];
769 const mutation = Atom.fn((id: "first" | "second") =>
770 (id === "first" ? firstLatch : secondLatch).await.pipe(
771 Effect.as(id),
772 Effect.onInterrupt(() =>
773 Effect.sync(() => {
774 interrupted.push(id);
775 }),
776 ),
777 ),
778 );
779 const registry = AtomRegistry.make();
780 const unmount = registry.mount(mutation);
781
782 registry.set(mutation, "first");
783 registry.set(mutation, "second");
784 yield* Effect.yieldNow;
785
786 expect(interrupted).toEqual(["first"]);
787
788 secondLatch.openUnsafe();
789 expect(
790 yield* AtomRegistry.getResult(registry, mutation, {
791 suspendOnWaiting: true,
792 }),
793 ).toBe("second");
794
795 unmount();
796 registry.dispose();
797 }),
798 );
799
800 it.effect("keeps stream mutations waiting until the final emitted value", () =>
801 Effect.gen(function* () {
802 const completionLatch = Latch.makeUnsafe();
803 const mutation = Atom.fn(() =>
804 Stream.make("progress").pipe(
805 Stream.concat(Stream.fromEffect(completionLatch.await.pipe(Effect.as("done")))),
806 ),
807 );
808 const registry = AtomRegistry.make();
809 const unmount = registry.mount(mutation);
810
811 registry.set(mutation, undefined);
812
813 const progress = registry.get(mutation);
814 expect(AsyncResult.isSuccess(progress)).toBe(true);
815 if (AsyncResult.isSuccess(progress)) {
816 expect(progress.value).toBe("progress");
817 expect(progress.waiting).toBe(true);
818 }
819
820 completionLatch.openUnsafe();
821 expect(
822 yield* AtomRegistry.getResult(registry, mutation, {
823 suspendOnWaiting: true,
824 }),
825 ).toBe("done");
826
827 unmount();
828 registry.dispose();
829 }),
830 );
831
832 it.effect(
833 "allows concurrent effects to finish but does not correlate results to individual writes",
834 () =>
835 Effect.gen(function* () {
836 const firstLatch = Latch.makeUnsafe();
837 const secondLatch = Latch.makeUnsafe();
838 const mutation = Atom.fn<never, "first" | "second", "first" | "second">(
839 (id: "first" | "second") =>
840 (id === "first" ? firstLatch : secondLatch).await.pipe(Effect.as(id)),
841 { concurrent: true },
842 );
843 const registry = AtomRegistry.make();
844 const unmount = registry.mount(mutation);
845
846 registry.set(mutation, "first");
847 const firstResult = yield* AtomRegistry.getResult(registry, mutation, {
848 suspendOnWaiting: true,
849 }).pipe(Effect.forkChild({ startImmediately: true }));
850 registry.set(mutation, "second");
851 const secondResult = yield* AtomRegistry.getResult(registry, mutation, {
852 suspendOnWaiting: true,
853 }).pipe(Effect.forkChild({ startImmediately: true }));
854
855 secondLatch.openUnsafe();
856 yield* Effect.yieldNow;
857
858 const stillWaiting = registry.get(mutation);
859 expect(stillWaiting.waiting).toBe(true);
860
861 firstLatch.openUnsafe();
862
863 expect(yield* Fiber.join(firstResult)).toBe("first");
864 expect(yield* Fiber.join(secondResult)).toBe("first");
865
866 unmount();
867 registry.dispose();
868 }),
869 );
870});
871
872describe("executeAtomQuery", () => {
873 it("keeps concurrent query results correlated to their atoms", async () => {
874 const firstLatch = Latch.makeUnsafe();
875 const secondLatch = Latch.makeUnsafe();
876 const firstAtom = Atom.make(firstLatch.await.pipe(Effect.as("first")));
877 const secondAtom = Atom.make(secondLatch.await.pipe(Effect.as("second")));
878 const registry = AtomRegistry.make();
879
880 const firstResult = executeAtomQuery(registry, firstAtom);
881 const secondResult = executeAtomQuery(registry, secondAtom);
882
883 secondLatch.openUnsafe();
884 firstLatch.openUnsafe();
885
886 const [first, second] = await Promise.all([firstResult, secondResult]);
887 expect(first._tag).toBe("Success");
888 expect(second._tag).toBe("Success");
889 if (first._tag === "Success" && second._tag === "Success") {
890 expect(first.value).toBe("first");
891 expect(second.value).toBe("second");
892 }
893
894 registry.dispose();
895 });
896
897 it("settles when its caller aborts a waiting query", async () => {
898 const registry = AtomRegistry.make();
899 const controller = new AbortController();
900 const resultPromise = executeAtomQuery(registry, Atom.make(Effect.never), {
901 reportDefect: false,
902 signal: controller.signal,
903 });
904
905 controller.abort();
906
907 const result = await resultPromise;
908 expect(result._tag).toBe("Failure");
909 if (result._tag === "Failure") {
910 expect(Cause.hasInterruptsOnly(result.cause)).toBe(true);
911 }
912 registry.dispose();
913 });
914});
915
916describe("runtime command runner", () => {
917 it("encodes custom command rejections as defects", async () => {
918 const defect = new Error("custom command rejected");
919 const registry = AtomRegistry.make();
920 const result = await runAtomCommand(
921 registry,
922 {
923 label: "test.rejected-command",
924 run: () => Promise.reject(defect),
925 },
926 undefined,
927 { reportDefect: false },
928 );
929
930 expect(result._tag).toBe("Failure");
931 if (result._tag === "Failure") {
932 expect(Cause.hasDies(result.cause)).toBe(true);
933 expect(Cause.squash(result.cause)).toBe(defect);
934 }
935 registry.dispose();
936 });
937
938 it("settles generated command scheduler defects from direct callers", async () => {
939 const defect = new Error("invalid command key");
940 const runtime = Atom.runtime(Layer.empty);
941 const command = createRuntimeCommand(runtime, {
942 label: "test.invalid-key",
943 concurrency: {
944 mode: "serial",
945 key: () => {
946 throw defect;
947 },
948 },
949 execute: () => Effect.void,
950 });
951 const registry = AtomRegistry.make();
952
953 const result = await command.run(registry, undefined);
954 expect(result._tag).toBe("Failure");
955 if (result._tag === "Failure") {
956 expect(Cause.hasDies(result.cause)).toBe(true);
957 expect(Cause.squash(result.cause)).toBe(defect);
958 }
959 registry.dispose();
960 });
961
962 it("correlates parallel invocation results", async () => {
963 const firstLatch = Latch.makeUnsafe();
964 const secondLatch = Latch.makeUnsafe();
965 const runtime = Atom.runtime(Layer.empty);
966 const command = createRuntimeCommand(runtime, {
967 label: "test.parallel",
968 execute: (id: "first" | "second") =>
969 (id === "first" ? firstLatch : secondLatch).await.pipe(Effect.as(id)),
970 });
971 const registry = AtomRegistry.make();
972
973 const first = command.run(registry, "first");
974 const second = command.run(registry, "second");
975 secondLatch.openUnsafe();
976 firstLatch.openUnsafe();
977
978 expect(await first).toMatchObject({ _tag: "Success", value: "first", waiting: false });
979 expect(await second).toMatchObject({ _tag: "Success", value: "second", waiting: false });
980 registry.dispose();
981 });
982
983 it("serializes commands that share a scheduler and lane", async () => {
984 const firstLatch = Latch.makeUnsafe();
985 const events: string[] = [];
986 const runtime = Atom.runtime(Layer.empty);
987 const scheduler = createAtomCommandScheduler();
988 const concurrency = { mode: "serial" as const, key: () => "shared" };
989 const firstCommand = createRuntimeCommand(runtime, {
990 label: "test.first",
991 scheduler,
992 concurrency,
993 execute: () =>
994 Effect.sync(() => events.push("first:start")).pipe(
995 Effect.andThen(firstLatch.await),
996 Effect.tap(() => Effect.sync(() => events.push("first:end"))),
997 ),
998 });
999 const secondCommand = createRuntimeCommand(runtime, {
1000 label: "test.second",
1001 scheduler,
1002 concurrency,
1003 execute: () => Effect.sync(() => events.push("second:start")),
1004 });
1005 const registry = AtomRegistry.make();
1006
1007 const first = firstCommand.run(registry, undefined);
1008 const second = secondCommand.run(registry, undefined);
1009 await Promise.resolve();
1010 expect(events).toEqual(["first:start"]);
1011
1012 firstLatch.openUnsafe();
1013 await Promise.all([first, second]);
1014 expect(events).toEqual(["first:start", "first:end", "second:start"]);
1015 registry.dispose();
1016 });
1017
1018 it.effect("releases a shared scheduler lane before the outer command finishes", () =>
1019 Effect.gen(function* () {
1020 const handoffStarted = Latch.makeUnsafe();
1021 const handoffComplete = Latch.makeUnsafe();
1022 const resumeComplete = Latch.makeUnsafe();
1023 const runtime = Atom.runtime(Layer.empty);
1024 const scheduler = createAtomCommandScheduler();
1025 const concurrency = { mode: "serial" as const, key: () => "shared" };
1026 const updateCommand = createRuntimeCommand(runtime, {
1027 label: "test.update",
1028 execute: (_input: void, registry) =>
1029 scheduleAtomCommandEffect(
1030 registry,
1031 scheduler,
1032 concurrency,
1033 undefined,
1034 Effect.sync(() => handoffStarted.openUnsafe()).pipe(
1035 Effect.andThen(handoffComplete.await),
1036 ),
1037 ).pipe(Effect.andThen(resumeComplete.await)),
1038 });
1039 const configCommand = createRuntimeCommand(runtime, {
1040 label: "test.config",
1041 scheduler,
1042 concurrency,
1043 execute: () => Effect.succeed("configured"),
1044 });
1045 const registry = AtomRegistry.make();
1046
1047 const update = updateCommand.run(registry, undefined);
1048 yield* handoffStarted.await;
1049 const config = configCommand.run(registry, undefined);
1050 handoffComplete.openUnsafe();
1051
1052 expect(yield* Effect.promise(() => config)).toMatchObject({
1053 _tag: "Success",
1054 value: "configured",
1055 waiting: false,
1056 });
1057 resumeComplete.openUnsafe();
1058 expect(yield* Effect.promise(() => update)).toMatchObject({
1059 _tag: "Success",
1060 waiting: false,
1061 });
1062 registry.dispose();
1063 }),
1064 );
1065
1066 it.effect("keeps an outer reconnect observer alive after a scheduled handoff returns", () =>
1067 Effect.gen(function* () {
1068 const commitStarting = yield* Deferred.make<void>();
1069 const observerArmed = yield* Deferred.make<void>();
1070 const reconnected = yield* Deferred.make<void>();
1071 const states = yield* Queue.unbounded<{ readonly phase: string }>();
1072 yield* Queue.offer(states, { phase: "connected" });
1073 const runtime = Atom.runtime(Layer.empty);
1074 const scheduler = createAtomCommandScheduler();
1075 const concurrency = { mode: "serial" as const, key: () => "shared" };
1076 const command = createRuntimeCommand(runtime, {
1077 label: "test.desktop-update-handoff",
1078 execute: (_input: void, registry) =>
1079 Effect.gen(function* () {
1080 yield* Deferred.await(commitStarting).pipe(
1081 Effect.andThen(
1082 Stream.fromQueue(states).pipe(
1083 Stream.tap(() => Deferred.succeed(observerArmed, undefined)),
1084 Stream.dropWhile((state) => state.phase === "connected"),
1085 Stream.filter((state) => state.phase === "connected"),
1086 Stream.runHead,
1087 ),
1088 ),
1089 Effect.andThen(Deferred.succeed(reconnected, undefined)),
1090 Effect.forkChild,
1091 );
1092 yield* scheduleAtomCommandEffect(
1093 registry,
1094 scheduler,
1095 concurrency,
1096 undefined,
1097 Deferred.succeed(commitStarting, undefined).pipe(
1098 Effect.andThen(Deferred.await(observerArmed)),
1099 ),
1100 );
1101 yield* Queue.offer(states, { phase: "backoff" });
1102 yield* Queue.offer(states, { phase: "connected" });
1103 yield* Deferred.await(reconnected);
1104 }),
1105 });
1106 const registry = AtomRegistry.make();
1107
1108 expect(yield* Effect.promise(() => command.run(registry, undefined))).toMatchObject({
1109 _tag: "Success",
1110 });
1111 registry.dispose();
1112 }),
1113 );
1114
1115 it("deduplicates single-flight commands by key", async () => {
1116 const latch = Latch.makeUnsafe();
1117 let executions = 0;
1118 const runtime = Atom.runtime(Layer.empty);
1119 const command = createRuntimeCommand(runtime, {
1120 label: "test.single-flight",
1121 concurrency: { mode: "singleFlight", key: (key: string) => key },
1122 execute: () =>
1123 Effect.sync(() => executions++).pipe(Effect.andThen(latch.await), Effect.as("done")),
1124 });
1125 const registry = AtomRegistry.make();
1126
1127 const first = command.run(registry, "same");
1128 const second = command.run(registry, "same");
1129 latch.openUnsafe();
1130
1131 expect(await first).toMatchObject({ _tag: "Success", value: "done", waiting: false });
1132 expect(await second).toMatchObject({ _tag: "Success", value: "done", waiting: false });
1133 expect(executions).toBe(1);
1134 registry.dispose();
1135 });
1136
1137 it("coalesces pending latest-value commands without interrupting the active call", async () => {
1138 const firstLatch = Latch.makeUnsafe();
1139 const executed: number[] = [];
1140 const runtime = Atom.runtime(Layer.empty);
1141 const command = createRuntimeCommand(runtime, {
1142 label: "test.latest",
1143 concurrency: { mode: "latest", key: () => "shared" },
1144 execute: (value: number) =>
1145 Effect.sync(() => executed.push(value)).pipe(
1146 Effect.andThen(value === 1 ? firstLatch.await : Effect.void),
1147 Effect.as(value),
1148 ),
1149 });
1150 const registry = AtomRegistry.make();
1151
1152 const first = command.run(registry, 1);
1153 await Promise.resolve();
1154 const second = command.run(registry, 2);
1155 const third = command.run(registry, 3);
1156 firstLatch.openUnsafe();
1157
1158 expect(await first).toMatchObject({ _tag: "Success", value: 1, waiting: false });
1159 expect(await second).toMatchObject({ _tag: "Success", value: 3, waiting: false });
1160 expect(await third).toMatchObject({ _tag: "Success", value: 3, waiting: false });
1161 expect(executed).toEqual([1, 3]);
1162 registry.dispose();
1163 });
1164});
1165
1166describe("persisted environment queries", () => {
1167 it.effect(
1168 "restores cached data in a new registry while offline, then refreshes on reconnect",
1169 () =>
1170 Effect.scoped(
1171 Effect.gen(function* () {
1172 let saved = Option.none<string>();
1173 const persistence: QueryPersistence<string> = {
1174 load: () => Effect.sync(() => saved),
1175 save: (_environmentId, _key, value) =>
1176 Effect.sync(() => {
1177 saved = Option.some(value);
1178 }),
1179 };
1180 const first = yield* makeEnvironmentQueryHarness(Effect.succeed("first"), {
1181 persistence,
1182 });
1183 const firstRegistry = yield* mountEnvironmentQuery(first.atom);
1184 expect(yield* readEnvironmentQuery(firstRegistry, first.atom)).toBe("first");
1185 expect(saved).toEqual(Option.some("first"));
1186
1187 const restarted = yield* makeEnvironmentQueryHarness(Effect.succeed("updated"), {
1188 persistence,
1189 initialState: queryConnectionState({ phase: "offline", network: "offline" }),
1190 });
1191 const registry = yield* mountEnvironmentQuery(restarted.atom);
1192 expect(yield* readEnvironmentQuery(registry, restarted.atom)).toBe("first");
1193 yield* SubscriptionRef.set(restarted.supervisorState, queryConnectionState());
1194 yield* Effect.yieldNow;
1195 expect(yield* readEnvironmentQuery(registry, restarted.atom)).toBe("updated");
1196 expect(saved).toEqual(Option.some("updated"));
1197 }),
1198 ),
1199 );
1200
1201 it.effect("keeps cached content available when revalidation fails", () =>
1202 Effect.scoped(
1203 Effect.gen(function* () {
1204 const harness = yield* makeEnvironmentQueryHarness(
1205 Effect.fail(new TestQueryError({ message: "failed" })),
1206 {
1207 persistence: { load: () => Effect.succeedSome("cached"), save: () => Effect.void },
1208 },
1209 );
1210 const registry = yield* mountEnvironmentQuery(harness.atom);
1211 yield* Effect.result(readEnvironmentQuery(registry, harness.atom));
1212 expect(Option.getOrNull(AsyncResult.value(registry.get(harness.atom)))).toBe("cached");
1213 }),
1214 ),
1215 );
1216
1217 it.effect("shows persisted data before a live request finishes", () =>
1218 Effect.scoped(
1219 Effect.gen(function* () {
1220 const response = yield* Deferred.make<string>();
1221 const seeded = yield* Deferred.make<void>();
1222 const refreshed = yield* Deferred.make<void>();
1223 const harness = yield* makeEnvironmentQueryHarness(Deferred.await(response), {
1224 persistence: {
1225 load: () => Effect.succeedSome("cached"),
1226 save: () => Effect.void,
1227 },
1228 });
1229 const registry = yield* mountEnvironmentQuery(harness.atom);
1230 const unsubscribe = registry.subscribe(
1231 harness.atom,
1232 (value) => {
1233 if (Option.getOrNull(AsyncResult.value(value)) === "cached")
1234 Deferred.doneUnsafe(seeded, Exit.void);
1235 if (Option.getOrNull(AsyncResult.value(value)) === "fresh")
1236 Deferred.doneUnsafe(refreshed, Exit.void);
1237 },
1238 { immediate: true },
1239 );
1240 yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));
1241 yield* Deferred.await(seeded);
1242 expect(Option.getOrNull(AsyncResult.value(registry.get(harness.atom)))).toBe("cached");
1243 yield* Deferred.succeed(response, "fresh");
1244 yield* Deferred.await(refreshed);
1245 expect(yield* readEnvironmentQuery(registry, harness.atom)).toBe("fresh");
1246 }),
1247 ),
1248 );
1249});