diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index f7e46fb777faa..f3c282d8bd312 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -14,6 +14,7 @@ import { WS_METHODS, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; @@ -36,6 +37,7 @@ import { AVAILABLE_CONNECTION_STATE, PrimaryConnectionTarget, type PreparedConnection, + type SupervisorConnectionState, } from "../connection/model.ts"; import * as EnvironmentRegistry from "../connection/registry.ts"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; @@ -1394,3 +1396,84 @@ describe("provider setup commands", () => { }).pipe(Effect.scoped), ); }); + +describe("resource telemetry history", () => { + it.effect("refreshes an initially empty history while mounted and stops after unmount", () => + Effect.gen(function* () { + vi.useFakeTimers(); + try { + yield* Effect.scoped( + Effect.gen(function* () { + let reads = 0; + const client = { + [WS_METHODS.serverGetResourceTelemetryHistory]: () => + Effect.sync(() => ({ + retainedSampleCount: reads++, + })), + } as unknown as WsRpcProtocolClient; + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make({ + ...AVAILABLE_CONNECTION_STATE, + phase: "connected", + }), + session: yield* SubscriptionRef.make(Option.some(session(client))), + prepared: yield* SubscriptionRef.make(Option.none()), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + }); + const environments = EnvironmentRegistry.EnvironmentRegistry.of({ + run: (_id, effect) => + Effect.provideService( + effect, + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ), + followStream: (_id, stream) => + Stream.provideService( + stream, + EnvironmentSupervisor.EnvironmentSupervisor, + supervisor, + ), + } as EnvironmentRegistry.EnvironmentRegistry["Service"]); + const runtime = Atom.runtime( + Layer.merge( + Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environments), + Layer.succeed(Persistence.EnvironmentCacheStore, { + loadServerConfig: () => Effect.succeedNone, + } as unknown as Persistence.EnvironmentCacheStore["Service"]), + ), + ); + const atoms = createServerEnvironmentAtoms(runtime, { + initialConfigValueAtom: () => Atom.make(CONFIG), + }); + const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (value) => + Effect.sync(() => value.dispose()), + ); + const history = atoms.resourceTelemetryHistory({ + environmentId: TARGET.environmentId, + input: { windowMs: 900_000, bucketMs: 30_000 }, + }); + const unmount = registry.mount(history); + const initial = yield* AtomRegistry.getResult(registry, history, { + suspendOnWaiting: true, + }); + expect(initial.retainedSampleCount).toBe(0); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(5_000)); + expect(reads).toBe(2); + const updated = yield* AtomRegistry.getResult(registry, history, { + suspendOnWaiting: true, + }); + expect(updated.retainedSampleCount).toBe(1); + unmount(); + yield* Effect.promise(() => vi.advanceTimersByTimeAsync(10_000)); + expect(reads).toBe(2); + }), + ); + } finally { + vi.useRealTimers(); + } + }), + ); +}); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 04a2d66321517..bc291114567ec 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -1362,6 +1362,8 @@ export function createServerEnvironmentAtoms( label: "environment-data:server:resource-telemetry-history", tag: WS_METHODS.serverGetResourceTelemetryHistory, staleTimeMs: 5_000, + refreshIntervalMs: 5_000, + idleTtlMs: 0, }), searchAcpRegistry: createEnvironmentRpcQueryAtomFamily(runtime, { label: "environment-data:server:acp-registry:search",