Skip to content

Commit e11b644

Browse files
tris203tris203
committed
fix(client-runtime): retry queries after connection interruption
- Preserve query results while reconnecting - Mark unavailable environments as settled failures
1 parent f035a0f commit e11b644

2 files changed

Lines changed: 374 additions & 22 deletions

File tree

‎packages/client-runtime/src/state/runtime.test.ts‎

Lines changed: 349 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,12 +6,27 @@ import * as Exit from "effect/Exit";
66
import * as Fiber from "effect/Fiber";
77
import * as Latch from "effect/Latch";
88
import * as Layer from "effect/Layer";
9+
import * as Option from "effect/Option";
10+
import * as Schema from "effect/Schema";
911
import * as Stream from "effect/Stream";
12+
import * as SubscriptionRef from "effect/SubscriptionRef";
1013
import { AsyncResult, Atom, AtomRegistry } from "effect/unstable/reactivity";
1114

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 type * as RpcSession from "../rpc/session.ts";
1226
import {
1327
environmentRpcKey,
1428
createAtomCommandScheduler,
29+
createEnvironmentQueryAtomFamily,
1530
createRuntimeCommand,
1631
scheduleAtomCommandEffect,
1732
executeAtomCommand,
@@ -24,6 +39,79 @@ import {
2439
squashAtomCommandFailure,
2540
} from "./runtime.ts";
2641

42+
const QUERY_ENVIRONMENT = new PrimaryConnectionTarget({
43+
environmentId: EnvironmentId.make("query-environment"),
44+
label: "Query environment",
45+
httpBaseUrl: "https://query.example.test",
46+
wsBaseUrl: "wss://query.example.test",
47+
});
48+
49+
class TestQueryError extends Schema.TaggedErrorClass<TestQueryError>()("TestQueryError", {
50+
message: Schema.String,
51+
}) {}
52+
53+
function queryConnectionState(
54+
overrides: Partial<SupervisorConnectionState> = {},
55+
): SupervisorConnectionState {
56+
return {
57+
...AVAILABLE_CONNECTION_STATE,
58+
desired: true,
59+
network: "online",
60+
phase: "connected",
61+
attempt: 1,
62+
generation: 1,
63+
...overrides,
64+
};
65+
}
66+
67+
const makeEnvironmentQueryHarness = Effect.fn("TestEnvironmentQuery.makeHarness")(function* <A, E>(
68+
execute: Effect.Effect<A, E>,
69+
) {
70+
const supervisorState = yield* SubscriptionRef.make(queryConnectionState());
71+
const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
72+
target: QUERY_ENVIRONMENT,
73+
state: supervisorState,
74+
session: yield* SubscriptionRef.make<Option.Option<RpcSession.RpcSession>>(Option.none()),
75+
prepared: yield* SubscriptionRef.make<Option.Option<PreparedConnection>>(Option.none()),
76+
connect: Effect.void,
77+
disconnect: Effect.void,
78+
retryNow: Effect.void,
79+
} satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
80+
const run: EnvironmentRegistry.EnvironmentRegistry["Service"]["run"] = (_environmentId, effect) =>
81+
Effect.provideService(effect, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
82+
const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
83+
run,
84+
stateChanges: () => SubscriptionRef.changes(supervisorState),
85+
} as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
86+
const runtime = Atom.runtime(
87+
Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
88+
);
89+
const family = createEnvironmentQueryAtomFamily(runtime, {
90+
label: "test.environment-query",
91+
staleTimeMs: 60_000,
92+
execute: () => execute,
93+
});
94+
95+
return {
96+
atom: family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined }),
97+
supervisorState,
98+
};
99+
});
100+
101+
const mountEnvironmentQuery = Effect.fn("TestEnvironmentQuery.mount")(function* <A, E>(
102+
atom: Atom.Atom<AsyncResult.AsyncResult<A, E>>,
103+
) {
104+
const registry = AtomRegistry.make();
105+
const unmount = registry.mount(atom);
106+
yield* Effect.addFinalizer(() =>
107+
Effect.sync(() => {
108+
unmount();
109+
registry.dispose();
110+
}),
111+
);
112+
return registry;
113+
});
114+
27115
describe("settleAsyncResult", () => {
28116
it("preserves successful values and typed failures", async () => {
29117
const success = await settleAsyncResult(() => Promise.resolve(Exit.succeed("done")));
@@ -163,6 +251,267 @@ describe("environmentRpcKey", () => {
163251
});
164252
});
165253

254+
describe("environment query lifecycle", () => {
255+
it.effect(
256+
"retries an interrupted query without exposing a failure during session replacement",
257+
() =>
258+
Effect.scoped(
259+
Effect.gen(function* () {
260+
const firstStarted = Latch.makeUnsafe();
261+
const interruptFirst = Latch.makeUnsafe();
262+
const firstSettled = Latch.makeUnsafe();
263+
let executions = 0;
264+
const execute = Effect.suspend(() => {
265+
executions += 1;
266+
if (executions > 1) {
267+
return Effect.succeed("recovered");
268+
}
269+
firstStarted.openUnsafe();
270+
return interruptFirst.await.pipe(
271+
Effect.andThen(Effect.interrupt),
272+
Effect.ensuring(
273+
Effect.sync(() => {
274+
firstSettled.openUnsafe();
275+
}),
276+
),
277+
);
278+
});
279+
const harness = yield* makeEnvironmentQueryHarness(execute);
280+
const registry = AtomRegistry.make();
281+
const observed: Array<AsyncResult.AsyncResult<string, unknown>> = [];
282+
const unsubscribe = registry.subscribe(
283+
harness.atom,
284+
(result) => {
285+
observed.push(result);
286+
},
287+
{ immediate: true },
288+
);
289+
yield* Effect.addFinalizer(() =>
290+
Effect.sync(() => {
291+
unsubscribe();
292+
registry.dispose();
293+
}),
294+
);
295+
296+
yield* firstStarted.await;
297+
yield* SubscriptionRef.set(
298+
harness.supervisorState,
299+
queryConnectionState({ phase: "connecting", stage: "preparing" }),
300+
);
301+
yield* Effect.yieldNow;
302+
yield* SubscriptionRef.set(
303+
harness.supervisorState,
304+
queryConnectionState({
305+
phase: "backoff",
306+
stage: null,
307+
lastFailure: new ConnectionTransientError({
308+
reason: "transport",
309+
detail: "Relay session is reconnecting.",
310+
}),
311+
retryAt: 1,
312+
}),
313+
);
314+
yield* Effect.yieldNow;
315+
interruptFirst.openUnsafe();
316+
yield* firstSettled.await;
317+
yield* Effect.yieldNow;
318+
319+
expect(observed.some(AsyncResult.isFailure)).toBe(false);
320+
321+
yield* SubscriptionRef.set(
322+
harness.supervisorState,
323+
queryConnectionState({ generation: 2 }),
324+
);
325+
expect(
326+
yield* AtomRegistry.getResult(registry, harness.atom, {
327+
suspendOnWaiting: true,
328+
}),
329+
).toBe("recovered");
330+
}),
331+
),
332+
);
333+
334+
it.effect.each([
335+
{
336+
condition: "after a manual disconnect",
337+
state: queryConnectionState({
338+
desired: false,
339+
phase: "available",
340+
stage: null,
341+
attempt: 0,
342+
}),
343+
},
344+
{
345+
condition: "while the environment is offline",
346+
state: queryConnectionState({ network: "offline", phase: "offline", stage: null }),
347+
},
348+
{
349+
condition: "when connection recovery is blocked",
350+
state: queryConnectionState({
351+
phase: "blocked",
352+
stage: null,
353+
lastFailure: new ConnectionBlockedError({
354+
reason: "permission",
355+
detail: "Access denied.",
356+
}),
357+
}),
358+
},
359+
] as const)("settles as unavailable $condition", ({ state }) =>
360+
Effect.scoped(
361+
Effect.gen(function* () {
362+
const harness = yield* makeEnvironmentQueryHarness(Effect.succeed("connected"));
363+
const registry = yield* mountEnvironmentQuery(harness.atom);
364+
365+
expect(
366+
yield* AtomRegistry.getResult(registry, harness.atom, {
367+
suspendOnWaiting: true,
368+
}),
369+
).toBe("connected");
370+
371+
yield* SubscriptionRef.set(harness.supervisorState, state);
372+
yield* Effect.yieldNow;
373+
374+
const result = registry.get(harness.atom);
375+
expect(AsyncResult.isFailure(result)).toBe(true);
376+
expect(result.waiting).toBe(false);
377+
}),
378+
),
379+
);
380+
381+
it.effect("keeps a genuine query failure settled while reconnecting", () =>
382+
Effect.scoped(
383+
Effect.gen(function* () {
384+
const expectedFailure = new TestQueryError({ message: "Query failed." });
385+
const firstStarted = Latch.makeUnsafe();
386+
const failFirst = Latch.makeUnsafe();
387+
const refreshStarted = Latch.makeUnsafe();
388+
const finishRefresh = Latch.makeUnsafe();
389+
let executions = 0;
390+
const execute = Effect.suspend(() => {
391+
executions += 1;
392+
if (executions === 1) {
393+
firstStarted.openUnsafe();
394+
return failFirst.await.pipe(Effect.andThen(Effect.fail(expectedFailure)));
395+
}
396+
refreshStarted.openUnsafe();
397+
return finishRefresh.await.pipe(Effect.as("recovered"));
398+
});
399+
const harness = yield* makeEnvironmentQueryHarness(execute);
400+
const registry = yield* mountEnvironmentQuery(harness.atom);
401+
402+
yield* firstStarted.await;
403+
failFirst.openUnsafe();
404+
const initial = yield* AtomRegistry.getResult(registry, harness.atom, {
405+
suspendOnWaiting: true,
406+
}).pipe(Effect.exit);
407+
expect(Exit.isFailure(initial)).toBe(true);
408+
if (Exit.isFailure(initial)) {
409+
expect(Cause.squash(initial.cause)).toBe(expectedFailure);
410+
}
411+
412+
yield* SubscriptionRef.set(
413+
harness.supervisorState,
414+
queryConnectionState({ phase: "connecting", stage: "opening" }),
415+
);
416+
yield* Effect.yieldNow;
417+
418+
const refreshing = registry.get(harness.atom);
419+
expect(AsyncResult.isFailure(refreshing)).toBe(true);
420+
expect(refreshing.waiting).toBe(true);
421+
if (AsyncResult.isFailure(refreshing)) {
422+
expect(Cause.squash(refreshing.cause)).toBe(expectedFailure);
423+
}
424+
425+
yield* SubscriptionRef.set(
426+
harness.supervisorState,
427+
queryConnectionState({ generation: 2 }),
428+
);
429+
yield* refreshStarted.await;
430+
finishRefresh.openUnsafe();
431+
expect(
432+
yield* AtomRegistry.getResult(registry, harness.atom, {
433+
suspendOnWaiting: true,
434+
}),
435+
).toBe("recovered");
436+
}),
437+
),
438+
);
439+
440+
it.effect("retains the last successful value while reconnecting", () =>
441+
Effect.scoped(
442+
Effect.gen(function* () {
443+
const refreshStarted = Latch.makeUnsafe();
444+
const finishRefresh = Latch.makeUnsafe();
445+
let executions = 0;
446+
const execute = Effect.suspend(() => {
447+
executions += 1;
448+
if (executions === 1) {
449+
return Effect.succeed("cached");
450+
}
451+
refreshStarted.openUnsafe();
452+
return finishRefresh.await.pipe(Effect.as("updated"));
453+
});
454+
const harness = yield* makeEnvironmentQueryHarness(execute);
455+
const registry = yield* mountEnvironmentQuery(harness.atom);
456+
457+
expect(
458+
yield* AtomRegistry.getResult(registry, harness.atom, {
459+
suspendOnWaiting: true,
460+
}),
461+
).toBe("cached");
462+
463+
yield* SubscriptionRef.set(
464+
harness.supervisorState,
465+
queryConnectionState({ phase: "connecting", stage: "opening" }),
466+
);
467+
yield* Effect.yieldNow;
468+
expect(registry.get(harness.atom)).toMatchObject({
469+
_tag: "Success",
470+
value: "cached",
471+
waiting: true,
472+
});
473+
474+
yield* SubscriptionRef.set(
475+
harness.supervisorState,
476+
queryConnectionState({
477+
phase: "backoff",
478+
stage: null,
479+
lastFailure: new ConnectionTransientError({
480+
reason: "transport",
481+
detail: "Retrying.",
482+
}),
483+
retryAt: 1,
484+
}),
485+
);
486+
yield* Effect.yieldNow;
487+
expect(registry.get(harness.atom)).toMatchObject({
488+
_tag: "Success",
489+
value: "cached",
490+
waiting: true,
491+
});
492+
493+
yield* SubscriptionRef.set(
494+
harness.supervisorState,
495+
queryConnectionState({ generation: 2 }),
496+
);
497+
yield* refreshStarted.await;
498+
expect(registry.get(harness.atom)).toMatchObject({
499+
_tag: "Success",
500+
value: "cached",
501+
waiting: true,
502+
});
503+
504+
finishRefresh.openUnsafe();
505+
expect(
506+
yield* AtomRegistry.getResult(registry, harness.atom, {
507+
suspendOnWaiting: true,
508+
}),
509+
).toBe("updated");
510+
}),
511+
),
512+
);
513+
});
514+
166515
describe("Atom.fn mutation semantics", () => {
167516
it.effect("interrupts the previous invocation when the same mutation atom is written again", () =>
168517
Effect.gen(function* () {

0 commit comments

Comments
 (0)