| ... | ... | @@ -6,12 +6,28 @@ import * as Exit from "effect/Exit"; |
| 6 | 6 | import * as Fiber from "effect/Fiber"; |
| 7 | 7 | import * as Latch from "effect/Latch"; |
| 8 | 8 | import * as Layer from "effect/Layer"; |
| 9 | import * as Option from "effect/Option"; |
| 10 | import * as Schema from "effect/Schema"; |
| 9 | 11 | import * as Stream from "effect/Stream"; |
| 12 | import * as SubscriptionRef from "effect/SubscriptionRef"; |
| 10 | 13 | import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity"; |
| 11 | 14 | |
| 15 | import { |
| 16 | AVAILABLE_CONNECTION_STATE, |
| 17 | ConnectionBlockedError, |
| 18 | ConnectionTransientError, |
| 19 | PrimaryConnectionTarget, |
| 20 | type PreparedConnection, |
| 21 | type SupervisorConnectionState, |
| 22 | } from "../connection/model.ts"; |
| 23 | import * as EnvironmentRegistry from "../connection/registry.ts"; |
| 24 | import * as EnvironmentSupervisor from "../connection/supervisor.ts"; |
| 25 | import { EnvironmentRpcUnavailableError } from "../rpc/client.ts"; |
| 26 | import type * as RpcSession from "../rpc/session.ts"; |
| 12 | 27 | import { |
| 13 | 28 | environmentRpcKey, |
| 14 | 29 | createAtomCommandScheduler, |
| 30 | createEnvironmentQueryAtomFamily, |
| 15 | 31 | createRuntimeCommand, |
| 16 | 32 | scheduleAtomCommandEffect, |
| 17 | 33 | executeAtomCommand, |
| ... | ... | @@ -24,6 +40,98 @@ import { |
| 24 | 40 | squashAtomCommandFailure, |
| 25 | 41 | } from "./runtime.ts"; |
| 26 | 42 | |
| 43 | const QUERY_ENVIRONMENT = new PrimaryConnectionTarget({ |
| 44 | environmentId: EnvironmentId.make("query-environment"), |
| 45 | label: "Query environment", |
| 46 | httpBaseUrl: "https://query.example.test", |
| 47 | wsBaseUrl: "wss://query.example.test", |
| 48 | }); |
| 49 | |
| 50 | const QUERY_RPC_SESSION = {} as RpcSession.RpcSession; |
| 51 | |
| 52 | class TestQueryError extends Schema.TaggedErrorClass<TestQueryError>()("TestQueryError", { |
| 53 | message: Schema.String, |
| 54 | }) {} |
| 55 | |
| 56 | const OFFLINE_QUERY_FAILURE = new ConnectionTransientError({ |
| 57 | reason: "transport", |
| 58 | detail: "Relay is unavailable.", |
| 59 | }); |
| 60 | |
| 61 | const BLOCKED_QUERY_FAILURE = new ConnectionBlockedError({ |
| 62 | reason: "permission", |
| 63 | detail: "Access denied.", |
| 64 | }); |
| 65 | |
| 66 | function queryConnectionState( |
| 67 | overrides: Partial<SupervisorConnectionState> = {}, |
| 68 | ): SupervisorConnectionState { |
| 69 | return { |
| 70 | ...AVAILABLE_CONNECTION_STATE, |
| 71 | desired: true, |
| 72 | network: "online", |
| 73 | phase: "connected", |
| 74 | attempt: 1, |
| 75 | generation: 1, |
| 76 | ...overrides, |
| 77 | }; |
| 78 | } |
| 79 | |
| 80 | const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* <A, E>( |
| 81 | execute: Effect.Effect<A, E>, |
| 82 | ) { |
| 83 | const supervisorState = yield* SubscriptionRef.make(queryConnectionState()); |
| 84 | const supervisorSession = yield* SubscriptionRef.make(Option.some(QUERY_RPC_SESSION)); |
| 85 | const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ |
| 86 | target: QUERY_ENVIRONMENT, |
| 87 | state: supervisorState, |
| 88 | session: supervisorSession, |
| 89 | prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()), |
| 90 | connect: Effect.void, |
| 91 | disconnect: Effect.void, |
| 92 | retryNow: Effect.void, |
| 93 | } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); |
| 94 | const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (_environmentId, effect) => |
| 95 | Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor); |
| 96 | const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = ( |
| 97 | _environmentId, |
| 98 | stream, |
| 99 | ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor); |
| 100 | const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({ |
| 101 | run, |
| 102 | followStream, |
| 103 | stateChanges: () => SubscriptionRef.changes(supervisorState), |
| 104 | } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]); |
| 105 | const runtime = Atom.runtime( |
| 106 | Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry), |
| 107 | ); |
| 108 | const family = createEnvironmentQueryAtomFamily(runtime, { |
| 109 | label: "test.environment-query", |
| 110 | staleTimeMs: 60_000, |
| 111 | execute: () => execute, |
| 112 | }); |
| 113 | |
| 114 | return { |
| 115 | atom: family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined }), |
| 116 | supervisorSession, |
| 117 | supervisorState, |
| 118 | }; |
| 119 | }); |
| 120 | |
| 121 | const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* <A, E>( |
| 122 | atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>, |
| 123 | ) { |
| 124 | const registry = AtomRegistry.make(); |
| 125 | const unmount = registry.mount(atom); |
| 126 | yield* Effect.addFinalizer(() => |
| 127 | Effect.sync(() => { |
| 128 | unmount(); |
| 129 | registry.dispose(); |
| 130 | }), |
| 131 | ); |
| 132 | return registry; |
| 133 | }); |
| 134 | |
| 27 | 135 | describe("settleAsyncResult", () => { |
| 28 | 136 | it("preserves successful values and typed failures", async () => { |
| 29 | 137 | const success = await settleAsyncResult(() => Promise.resolve(Exit.succeed("done"))); |
| ... | ... | @@ -163,6 +271,283 @@ describe("environmentRpcKey", () => { |
| 163 | 271 | }); |
| 164 | 272 | }); |
| 165 | 273 | |
| 274 | describe("environment query lifecycle", () => { |
| 275 | it.effect( |
| 276 | "retries an interrupted query without exposing a failure during session replacement", |
| 277 | () => |
| 278 | Effect.scoped( |
| 279 | Effect.gen(function* () { |
| 280 | const firstStarted = Latch.makeUnsafe(); |
| 281 | const failFirst = Latch.makeUnsafe(); |
| 282 | const firstSettled = Latch.makeUnsafe(); |
| 283 | const unavailable = new EnvironmentRpcUnavailableError({ |
| 284 | environmentId: QUERY_ENVIRONMENT.environmentId, |
| 285 | message: "Query environment is not connected.", |
| 286 | }); |
| 287 | let executions = 0; |
| 288 | const execute = Effect.suspend(() => { |
| 289 | executions += 1; |
| 290 | if (executions > 1) { |
| 291 | return Effect.succeed("recovered"); |
| 292 | } |
| 293 | firstStarted.openUnsafe(); |
| 294 | return failFirst.await.pipe( |
| 295 | Effect.andThen(Effect.fail(unavailable)), |
| 296 | Effect.ensuring( |
| 297 | Effect.sync(() => { |
| 298 | firstSettled.openUnsafe(); |
| 299 | }), |
| 300 | ), |
| 301 | ); |
| 302 | }); |
| 303 | const harness = yield* makeEnvironmentQueryHarness(execute); |
| 304 | const registry = AtomRegistry.make(); |
| 305 | const observed: Array<AsyncResult.AsyncResult<string, unknown>> = []; |
| 306 | const unsubscribe = registry.subscribe( |
| 307 | harness.atom, |
| 308 | (result) => { |
| 309 | observed.push(result); |
| 310 | }, |
| 311 | { immediate: true }, |
| 312 | ); |
| 313 | yield* Effect.addFinalizer(() => |
| 314 | Effect.sync(() => { |
| 315 | unsubscribe(); |
| 316 | registry.dispose(); |
| 317 | }), |
| 318 | ); |
| 319 | |
| 320 | yield* firstStarted.await; |
| 321 | yield* SubscriptionRef.set(harness.supervisorSession, Option.none()); |
| 322 | yield* Effect.yieldNow; |
| 323 | failFirst.openUnsafe(); |
| 324 | yield* firstSettled.await; |
| 325 | yield* Effect.yieldNow; |
| 326 | |
| 327 | expect(observed.some(AsyncResult.isFailure)).toBe(false); |
| 328 | |
| 329 | yield* SubscriptionRef.set( |
| 330 | harness.supervisorState, |
| 331 | queryConnectionState({ phase: "connecting", stage: "preparing" }), |
| 332 | ); |
| 333 | yield* Effect.yieldNow; |
| 334 | yield* SubscriptionRef.set( |
| 335 | harness.supervisorState, |
| 336 | queryConnectionState({ |
| 337 | phase: "backoff", |
| 338 | stage: null, |
| 339 | lastFailure: new ConnectionTransientError({ |
| 340 | reason: "transport", |
| 341 | detail: "Relay session is reconnecting.", |
| 342 | }), |
| 343 | retryAt: 1, |
| 344 | }), |
| 345 | ); |
| 346 | yield* Effect.yieldNow; |
| 347 | |
| 348 | yield* SubscriptionRef.set(harness.supervisorSession, Option.some(QUERY_RPC_SESSION)); |
| 349 | yield* SubscriptionRef.set( |
| 350 | harness.supervisorState, |
| 351 | queryConnectionState({ generation: 2 }), |
| 352 | ); |
| 353 | expect( |
| 354 | yield* AtomRegistry.getResult(registry, harness.atom, { |
| 355 | suspendOnWaiting: true, |
| 356 | }), |
| 357 | ).toBe("recovered"); |
| 358 | }), |
| 359 | ), |
| 360 | ); |
| 361 | |
| 362 | it.effect.each([ |
| 363 | { |
| 364 | condition: "after a manual disconnect", |
| 365 | state: queryConnectionState({ |
| 366 | desired: false, |
| 367 | phase: "available", |
| 368 | stage: null, |
| 369 | attempt: 0, |
| 370 | }), |
| 371 | expectedFailure: null, |
| 372 | }, |
| 373 | { |
| 374 | condition: "while the environment is offline", |
| 375 | state: queryConnectionState({ |
| 376 | network: "offline", |
| 377 | phase: "offline", |
| 378 | stage: null, |
| 379 | lastFailure: OFFLINE_QUERY_FAILURE, |
| 380 | }), |
| 381 | expectedFailure: OFFLINE_QUERY_FAILURE, |
| 382 | }, |
| 383 | { |
| 384 | condition: "when connection recovery is blocked", |
| 385 | state: queryConnectionState({ |
| 386 | phase: "blocked", |
| 387 | stage: null, |
| 388 | lastFailure: BLOCKED_QUERY_FAILURE, |
| 389 | }), |
| 390 | expectedFailure: BLOCKED_QUERY_FAILURE, |
| 391 | }, |
| 392 | ] as const)("settles as unavailable $condition", ({ state, expectedFailure }) => |
| 393 | Effect.scoped( |
| 394 | Effect.gen(function* () { |
| 395 | const harness = yield* makeEnvironmentQueryHarness(Effect.succeed("connected")); |
| 396 | const registry = yield* mountEnvironmentQuery(harness.atom); |
| 397 | |
| 398 | expect( |
| 399 | yield* AtomRegistry.getResult(registry, harness.atom, { |
| 400 | suspendOnWaiting: true, |
| 401 | }), |
| 402 | ).toBe("connected"); |
| 403 | |
| 404 | yield* SubscriptionRef.set(harness.supervisorState, state); |
| 405 | yield* Effect.yieldNow; |
| 406 | |
| 407 | const result = registry.get(harness.atom); |
| 408 | expect(AsyncResult.isFailure(result)).toBe(true); |
| 409 | expect(result.waiting).toBe(false); |
| 410 | if (AsyncResult.isFailure(result) && expectedFailure !== null) { |
| 411 | expect(Cause.squash(result.cause)).toBe(expectedFailure); |
| 412 | } |
| 413 | }), |
| 414 | ), |
| 415 | ); |
| 416 | |
| 417 | it.effect("keeps a genuine query failure settled while reconnecting", () => |
| 418 | Effect.scoped( |
| 419 | Effect.gen(function* () { |
| 420 | const expectedFailure = new TestQueryError({ message: "Query failed." }); |
| 421 | const firstStarted = Latch.makeUnsafe(); |
| 422 | const failFirst = Latch.makeUnsafe(); |
| 423 | const refreshStarted = Latch.makeUnsafe(); |
| 424 | const finishRefresh = Latch.makeUnsafe(); |
| 425 | let executions = 0; |
| 426 | const execute = Effect.suspend(() => { |
| 427 | executions += 1; |
| 428 | if (executions === 1) { |
| 429 | firstStarted.openUnsafe(); |
| 430 | return failFirst.await.pipe(Effect.andThen(Effect.fail(expectedFailure))); |
| 431 | } |
| 432 | refreshStarted.openUnsafe(); |
| 433 | return finishRefresh.await.pipe(Effect.as("recovered")); |
| 434 | }); |
| 435 | const harness = yield* makeEnvironmentQueryHarness(execute); |
| 436 | const registry = yield* mountEnvironmentQuery(harness.atom); |
| 437 | |
| 438 | yield* firstStarted.await; |
| 439 | failFirst.openUnsafe(); |
| 440 | const initial = yield* AtomRegistry.getResult(registry, harness.atom, { |
| 441 | suspendOnWaiting: true, |
| 442 | }).pipe(Effect.exit); |
| 443 | expect(Exit.isFailure(initial)).toBe(true); |
| 444 | if (Exit.isFailure(initial)) { |
| 445 | expect(Cause.squash(initial.cause)).toBe(expectedFailure); |
| 446 | } |
| 447 | |
| 448 | yield* SubscriptionRef.set( |
| 449 | harness.supervisorState, |
| 450 | queryConnectionState({ phase: "connecting", stage: "opening" }), |
| 451 | ); |
| 452 | yield* Effect.yieldNow; |
| 453 | |
| 454 | const refreshing = registry.get(harness.atom); |
| 455 | expect(AsyncResult.isFailure(refreshing)).toBe(true); |
| 456 | expect(refreshing.waiting).toBe(true); |
| 457 | if (AsyncResult.isFailure(refreshing)) { |
| 458 | expect(Cause.squash(refreshing.cause)).toBe(expectedFailure); |
| 459 | } |
| 460 | |
| 461 | yield* SubscriptionRef.set( |
| 462 | harness.supervisorState, |
| 463 | queryConnectionState({ generation: 2 }), |
| 464 | ); |
| 465 | yield* refreshStarted.await; |
| 466 | finishRefresh.openUnsafe(); |
| 467 | expect( |
| 468 | yield* AtomRegistry.getResult(registry, harness.atom, { |
| 469 | suspendOnWaiting: true, |
| 470 | }), |
| 471 | ).toBe("recovered"); |
| 472 | }), |
| 473 | ), |
| 474 | ); |
| 475 | |
| 476 | it.effect("retains the last successful value while reconnecting", () => |
| 477 | Effect.scoped( |
| 478 | Effect.gen(function* () { |
| 479 | const refreshStarted = Latch.makeUnsafe(); |
| 480 | const finishRefresh = Latch.makeUnsafe(); |
| 481 | let executions = 0; |
| 482 | const execute = Effect.suspend(() => { |
| 483 | executions += 1; |
| 484 | if (executions === 1) { |
| 485 | return Effect.succeed("cached"); |
| 486 | } |
| 487 | refreshStarted.openUnsafe(); |
| 488 | return finishRefresh.await.pipe(Effect.as("updated")); |
| 489 | }); |
| 490 | const harness = yield* makeEnvironmentQueryHarness(execute); |
| 491 | const registry = yield* mountEnvironmentQuery(harness.atom); |
| 492 | |
| 493 | expect( |
| 494 | yield* AtomRegistry.getResult(registry, harness.atom, { |
| 495 | suspendOnWaiting: true, |
| 496 | }), |
| 497 | ).toBe("cached"); |
| 498 | |
| 499 | yield* SubscriptionRef.set( |
| 500 | harness.supervisorState, |
| 501 | queryConnectionState({ phase: "connecting", stage: "opening" }), |
| 502 | ); |
| 503 | yield* Effect.yieldNow; |
| 504 | expect(registry.get(harness.atom)).toMatchObject({ |
| 505 | _tag: "Success", |
| 506 | value: "cached", |
| 507 | waiting: true, |
| 508 | }); |
| 509 | |
| 510 | yield* SubscriptionRef.set( |
| 511 | harness.supervisorState, |
| 512 | queryConnectionState({ |
| 513 | phase: "backoff", |
| 514 | stage: null, |
| 515 | lastFailure: new ConnectionTransientError({ |
| 516 | reason: "transport", |
| 517 | detail: "Retrying.", |
| 518 | }), |
| 519 | retryAt: 1, |
| 520 | }), |
| 521 | ); |
| 522 | yield* Effect.yieldNow; |
| 523 | expect(registry.get(harness.atom)).toMatchObject({ |
| 524 | _tag: "Success", |
| 525 | value: "cached", |
| 526 | waiting: true, |
| 527 | }); |
| 528 | |
| 529 | yield* SubscriptionRef.set( |
| 530 | harness.supervisorState, |
| 531 | queryConnectionState({ generation: 2 }), |
| 532 | ); |
| 533 | yield* refreshStarted.await; |
| 534 | expect(registry.get(harness.atom)).toMatchObject({ |
| 535 | _tag: "Success", |
| 536 | value: "cached", |
| 537 | waiting: true, |
| 538 | }); |
| 539 | |
| 540 | finishRefresh.openUnsafe(); |
| 541 | expect( |
| 542 | yield* AtomRegistry.getResult(registry, harness.atom, { |
| 543 | suspendOnWaiting: true, |
| 544 | }), |
| 545 | ).toBe("updated"); |
| 546 | }), |
| 547 | ), |
| 548 | ); |
| 549 | }); |
| 550 | |
| 166 | 551 | describe("Atom.fn mutation semantics", () => { |
| 167 | 552 | it.effect("interrupts the previous invocation when the same mutation atom is written again", () => |
| 168 | 553 | Effect.gen(function* () { |
Tristan Knight <admin@snappeh.com> 2026-08-25 09:10:03+01:00
GitHub <noreply@github.com> 2026-08-25 01:10:03-07:00