diff --git a/apps/mobile/src/connection/environment-cache-store.test.ts b/apps/mobile/src/connection/environment-cache-store.test.ts index d8031de9e7a39..7891983b4c23f 100644 --- a/apps/mobile/src/connection/environment-cache-store.test.ts +++ b/apps/mobile/src/connection/environment-cache-store.test.ts @@ -498,10 +498,11 @@ describe("mobile SQLite environment cache store", () => { Effect.provideService(MobileDatabase.MobileDatabase, memory.database), ); - yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT); + yield* store.saveShell(ENVIRONMENT_ID, SHELL_SNAPSHOT, "1.0.0"); yield* store.saveThread(ENVIRONMENT_ID, THREAD_SNAPSHOT); const shell = Option.getOrThrow(yield* store.loadShell(ENVIRONMENT_ID)); + expect(shell.verifiedServerVersion).toBe("1.0.0"); const thread = Option.getOrThrow(yield* store.loadThread(ENVIRONMENT_ID, THREAD_ID)); expect(DateTime.formatIso(shell.threads[0]!.updatedAt)).toBe("2026-07-29T12:00:00.000Z"); diff --git a/apps/mobile/src/connection/environment-cache-store.ts b/apps/mobile/src/connection/environment-cache-store.ts index b55af4718d15f..b264d0a84ee1a 100644 --- a/apps/mobile/src/connection/environment-cache-store.ts +++ b/apps/mobile/src/connection/environment-cache-store.ts @@ -5,6 +5,7 @@ import { encodeStoredThreadSnapshot, Persistence, makeStoredShellSnapshotEncoder, + withVerifiedServerVersion, } from "@t3tools/client-runtime/platform"; import { type EnvironmentId, ServerConfig, VcsListRefsResult } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; @@ -96,7 +97,7 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { const database = yield* MobileDatabase.MobileDatabase; attachProjectFaviconDatabase(database); const encodeStoredShellSnapshot = makeStoredShellSnapshotEncoder(); - return Persistence.EnvironmentCacheStore.of({ + const store = Persistence.EnvironmentCacheStore.of({ loadShell: Effect.fn("MobileEnvironmentCache.loadShell")((environmentId) => loadDecodedCache({ database, @@ -106,7 +107,9 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { operation: "load-shell", decode: decodeStoredShellSnapshot, select: (stored) => - stored.environmentId === environmentId ? Option.some(stored.snapshot) : Option.none(), + stored.environmentId === environmentId + ? Option.some(withVerifiedServerVersion(stored)) + : Option.none(), }).pipe( Effect.zipWith( Effect.promise(() => projectFaviconDatabaseCache.hydrate()), @@ -115,16 +118,25 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { ), ), ), - saveShell: Effect.fn("MobileEnvironmentCache.saveShell")(function* (environmentId, snapshot) { - const payload = yield* encodeStoredShellSnapshot({ - schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, - environmentId, - snapshot, - }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); - yield* database - .saveCache(environmentId, "shell", "snapshot", ORCHESTRATION_CACHE_SCHEMA_VERSION, payload) - .pipe(Effect.mapError(mapDatabaseError("save-shell"))); - }), + saveShell: Effect.fn("MobileEnvironmentCache.saveShell")( + function* (environmentId, snapshot, verifiedServerVersion) { + const payload = yield* encodeStoredShellSnapshot({ + schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, + environmentId, + snapshot, + ...(verifiedServerVersion === undefined ? {} : { verifiedServerVersion }), + }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); + yield* database + .saveCache( + environmentId, + "shell", + "snapshot", + ORCHESTRATION_CACHE_SCHEMA_VERSION, + payload, + ) + .pipe(Effect.mapError(mapDatabaseError("save-shell"))); + }, + ), loadThread: Effect.fn("MobileEnvironmentCache.loadThread")((environmentId, threadId) => loadDecodedCache({ database, @@ -229,6 +241,7 @@ export const make = Effect.fn("MobileEnvironmentCacheStore.make")(function* () { ), ), }); + return Persistence.shareServerConfigReads(store); }); export const layer = Layer.effect(Persistence.EnvironmentCacheStore, make()); diff --git a/apps/mobile/src/features/observability/devSpanLog.test.ts b/apps/mobile/src/features/observability/devSpanLog.test.ts new file mode 100644 index 0000000000000..30e28383c7a82 --- /dev/null +++ b/apps/mobile/src/features/observability/devSpanLog.test.ts @@ -0,0 +1,48 @@ +import { expect, vi } from "vite-plus/test"; +import { it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Tracer from "effect/Tracer"; +import { RelayClientTracer, withRelayClientTracing } from "@t3tools/shared/relayTracing"; + +it.effect("logs relay spans while preserving their exporter and attributes", () => + Effect.gen(function* () { + vi.stubGlobal("__DEV__", true); + vi.stubEnv("EXPO_PUBLIC_T3CODE_LOG_SPANS", "1"); + using log = vi.spyOn(console, "log").mockImplementation(() => {}); + const exported: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (time, exit) => { + end(time, exit); + exported.push(span.name); + }; + return span; + }, + }); + try { + const DevSpanLog = yield* Effect.promise(() => import("./devSpanLog")); + yield* Effect.gen(function* () { + const span = yield* Effect.makeSpan("relay.test", { attributes: { route: "relay" } }); + span.end(span.status.startTime + 2_000_000n, Exit.succeed(undefined)); + }).pipe( + withRelayClientTracing, + Effect.provide( + DevSpanLog.layer.pipe( + Layer.provideMerge(Layer.succeed(RelayClientTracer, Option.some(tracer))), + ), + ), + ); + expect(exported).toEqual(["relay.test"]); + expect(log).toHaveBeenCalledWith(expect.stringContaining('"name":"relay.test"')); + expect(log).toHaveBeenCalledWith(expect.stringContaining('"attributes":{"route":"relay"}')); + } finally { + vi.unstubAllGlobals(); + vi.unstubAllEnvs(); + } + }), +); diff --git a/apps/mobile/src/features/observability/devSpanLog.ts b/apps/mobile/src/features/observability/devSpanLog.ts new file mode 100644 index 0000000000000..00d00bd7af568 --- /dev/null +++ b/apps/mobile/src/features/observability/devSpanLog.ts @@ -0,0 +1,48 @@ +import { RelayClientTracer } from "@t3tools/shared/relayTracing"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Tracer from "effect/Tracer"; + +// Simulators share the host clock, so these lines line up with the server's trace file. +const MIN_LOGGED_SPAN_NANOS = 1_000_000n; + +function loggedTracer(delegate: Tracer.Tracer): Tracer.Tracer { + return Tracer.make({ + ...delegate, + span(options) { + const span = delegate.span(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + const startTime = span.status.startTime; + if (endTime - startTime < MIN_LOGGED_SPAN_NANOS) return; + console.log( + `[t3-span] ${JSON.stringify({ + name: span.name, + startMs: Number(startTime / 1_000_000n), + endMs: Number(endTime / 1_000_000n), + ok: exit._tag === "Success", + attributes: Object.fromEntries(span.attributes), + })}`, + ); + }; + return span; + }, + }); +} + +/** Dev builds started with `EXPO_PUBLIC_T3CODE_LOG_SPANS=1` print every span of 1 ms or more. */ +export const layer = + __DEV__ && process.env.EXPO_PUBLIC_T3CODE_LOG_SPANS === "1" + ? Layer.effectContext( + Effect.gen(function* () { + const local = yield* Tracer.Tracer; + const relay = yield* RelayClientTracer; + return Context.make(Tracer.Tracer, loggedTracer(local)).pipe( + Context.add(RelayClientTracer, Option.map(relay, loggedTracer)), + ); + }), + ) + : Layer.empty; diff --git a/apps/mobile/src/lib/runtime.ts b/apps/mobile/src/lib/runtime.ts index 375cbc6b29fb8..e816cf7cbe0fc 100644 --- a/apps/mobile/src/lib/runtime.ts +++ b/apps/mobile/src/lib/runtime.ts @@ -8,6 +8,7 @@ import { layerRemoteHttpClient } from "@t3tools/client-runtime/rpc"; import * as Dpop from "../features/cloud/dpop"; import * as ManagedRelayLayer from "../features/cloud/managedRelayLayer"; import { resolveCloudPublicConfig } from "../features/cloud/publicConfig"; +import * as DevSpanLog from "../features/observability/devSpanLog"; import * as Tracing from "../features/observability/tracing"; import * as Persistence from "../persistence/layer"; import { disposeOnFoundationReplace, type FoundationHotModule } from "./foundation-fast-refresh"; @@ -33,9 +34,11 @@ const layerRuntime = Layer.merge( ManagedRelayLayer.layer(configuredRelayUrl()), Socket.layerWebSocketConstructorGlobal, ).pipe( + Layer.provideMerge( + DevSpanLog.layer.pipe(Layer.provideMerge(Tracing.layer.pipe(Layer.provide(layerHttpClient)))), + ), Layer.provideMerge(Dpop.layer), Layer.provideMerge(layerHttpClient), - Layer.provideMerge(Tracing.layer.pipe(Layer.provide(layerHttpClient))), Layer.provideMerge(Persistence.layer), // These layers read the keychain and the database while they build, and a // failed build is kept for the life of the process. diff --git a/apps/mobile/src/persistence/mobile-secure-storage.ts b/apps/mobile/src/persistence/mobile-secure-storage.ts index 7269e7969e5d3..fc499c005ba20 100644 --- a/apps/mobile/src/persistence/mobile-secure-storage.ts +++ b/apps/mobile/src/persistence/mobile-secure-storage.ts @@ -33,7 +33,7 @@ const make = MobileSecureStorage.of({ Effect.tryPromise({ try: () => SecureStore.getItemAsync(key), catch: (cause) => new MobileSecureStorageError({ operation: "read", key, cause }), - }), + }).pipe(Effect.tap(() => Effect.annotateCurrentSpan("secure_storage.key", key))), ), setItem: Effect.fn("MobileSecureStorage.setItem")((key, value) => Effect.tryPromise({ diff --git a/apps/mobile/src/persistence/mobile-storage.test.ts b/apps/mobile/src/persistence/mobile-storage.test.ts index 363875ade020d..bc9a49a75fa88 100644 --- a/apps/mobile/src/persistence/mobile-storage.test.ts +++ b/apps/mobile/src/persistence/mobile-storage.test.ts @@ -62,6 +62,10 @@ describe("mobile device identity", () => { const storage = yield* makeStorage; expect(yield* storage.loadOrCreateAgentAwarenessDeviceId).toBe("existing-device"); + // Every connect asks again; only the first may pay for a keychain read. + expect(yield* storage.loadOrCreateAgentAwarenessDeviceId).toBe("existing-device"); + expect(yield* storage.loadAgentAwarenessDeviceId).toBe("existing-device"); + expect(getItemAsync).toHaveBeenCalledTimes(1); expect(uuidv4).not.toHaveBeenCalled(); expect(setItemAsync).not.toHaveBeenCalled(); }), diff --git a/apps/mobile/src/persistence/mobile-storage.ts b/apps/mobile/src/persistence/mobile-storage.ts index f15f1c820d02e..0cc766eafb29a 100644 --- a/apps/mobile/src/persistence/mobile-storage.ts +++ b/apps/mobile/src/persistence/mobile-storage.ts @@ -125,20 +125,28 @@ export const make = Effect.fn("MobileStorage.make")(function* () { yield* secureStorage.setItem(key, encoded); }); + // The id never changes once written, and every connect reads it; a keychain read can take + // hundreds of milliseconds while the app starts. + let knownDeviceId: string | null = null; const loadOrCreateAgentAwarenessDeviceId = Effect.gen(function* () { + if (knownDeviceId !== null) return knownDeviceId; const existing = yield* secureStorage.getItem(AGENT_AWARENESS_DEVICE_ID_KEY); - if (existing?.trim()) return existing; + if (existing?.trim()) return (knownDeviceId = existing); const deviceId = yield* Effect.tryPromise({ try: () => import("../lib/uuid").then(({ uuidv4 }) => uuidv4()), catch: (cause) => new MobileDeviceIdGenerationError({ cause }), }); yield* secureStorage.setItem(AGENT_AWARENESS_DEVICE_ID_KEY, deviceId); - return deviceId; + return (knownDeviceId = deviceId); }).pipe(deviceIdLock.withPermits(1)); - const loadAgentAwarenessDeviceId = secureStorage - .getItem(AGENT_AWARENESS_DEVICE_ID_KEY) - .pipe(Effect.map((existing) => (existing?.trim() ? existing : null))); + const loadAgentAwarenessDeviceId = Effect.suspend(() => + knownDeviceId !== null + ? Effect.succeed(knownDeviceId) + : secureStorage + .getItem(AGENT_AWARENESS_DEVICE_ID_KEY) + .pipe(Effect.map((existing) => (existing?.trim() ? (knownDeviceId = existing) : null))), + ); const loadAgentAwarenessRegistrationRecord = readJson( AGENT_AWARENESS_REGISTRATION_KEY, diff --git a/apps/server/src/environment/DirectEndpoints.ts b/apps/server/src/environment/DirectEndpoints.ts index 99e33b98f9709..990ef4ee9949c 100644 --- a/apps/server/src/environment/DirectEndpoints.ts +++ b/apps/server/src/environment/DirectEndpoints.ts @@ -24,11 +24,13 @@ import { isWildcardHost, VIRTUAL_INTERFACE, } from "../startupAccess.ts"; +import { makeBackgroundRefreshed } from "./backgroundRefresh.ts"; type NetworkInterfacesMap = ReturnType; // Stay below the config discovery timeout so slow Tailscale drops only the tailnet name. const TAILSCALE_ENDPOINT_TIMEOUT = Duration.seconds(3); +const SERVED_URL_MAX_AGE = Duration.seconds(30); export class DirectEndpoints extends Context.Service< DirectEndpoints, @@ -119,6 +121,30 @@ export const make = Effect.gen(function* () { : Effect.succeed(false); yield* Effect.forkDetach(onCloudVm); + // Serve setup can fail without stopping startup; advertise only after it answers as this server. + const servedUrl = yield* makeBackgroundRefreshed( + readTailscaleStatus.pipe( + Effect.map((status) => status.magicDnsName), + Effect.flatMap((magicDnsName) => { + if (magicDnsName === null) return Effect.succeed(null); + const httpBaseUrl = buildTailscaleHttpsBaseUrl({ + magicDnsName, + servePort: config.tailscaleServePort, + }); + return probeTailscaleHttpsEndpoint({ baseUrl: httpBaseUrl }).pipe( + Effect.map((reachable) => (reachable ? httpBaseUrl : null)), + ); + }), + Effect.timeoutOption(TAILSCALE_ENDPOINT_TIMEOUT), + Effect.map(Option.flatMap(Option.fromNullishOr)), + Effect.orElseSucceed(() => Option.none()), + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.provideService(HttpClient.HttpClient, httpClient), + ), + SERVED_URL_MAX_AGE, + { warm: config.tailscaleServeEnabled }, + ); + const resolve = Effect.gen(function* () { const endpoints = [ ...resolveBoundEndpoints({ @@ -128,28 +154,10 @@ export const make = Effect.gen(function* () { onCloudVm: yield* onCloudVm, }), ]; - // Serve setup can fail without stopping startup; advertise only after it answers as this server. if (config.tailscaleServeEnabled) { - const servedUrl = yield* readTailscaleStatus.pipe( - Effect.map((status) => status.magicDnsName), - Effect.flatMap((magicDnsName) => { - if (magicDnsName === null) return Effect.succeed(null); - const httpBaseUrl = buildTailscaleHttpsBaseUrl({ - magicDnsName, - servePort: config.tailscaleServePort, - }); - return probeTailscaleHttpsEndpoint({ baseUrl: httpBaseUrl }).pipe( - Effect.map((reachable) => (reachable ? httpBaseUrl : null)), - ); - }), - Effect.timeoutOption(TAILSCALE_ENDPOINT_TIMEOUT), - Effect.map(Option.flatMap(Option.fromNullishOr)), - Effect.orElseSucceed(() => Option.none()), - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), - Effect.provideService(HttpClient.HttpClient, httpClient), - ); - if (Option.isSome(servedUrl)) { - endpoints.push({ kind: "tailnet", httpBaseUrl: servedUrl.value }); + const served = yield* servedUrl; + if (Option.isSome(served)) { + endpoints.push({ kind: "tailnet", httpBaseUrl: served.value }); } } return endpoints; diff --git a/apps/server/src/environment/RemoteOpenTargets.test.ts b/apps/server/src/environment/RemoteOpenTargets.test.ts index 3e66d5f707a0d..3a482409b93f4 100644 --- a/apps/server/src/environment/RemoteOpenTargets.test.ts +++ b/apps/server/src/environment/RemoteOpenTargets.test.ts @@ -68,15 +68,16 @@ const resolveTargets = (input: { readonly hostname: string; readonly username?: string | null; }) => { + // The service warms its cache while the layer builds, so host facts must reach construction. const resolve = Effect.flatMap(RemoteOpenTargets.RemoteOpenTargets, (service) => service.resolveTargets(), ).pipe( - Effect.provideService(HostProcess.Hostname, input.hostname), Effect.provide( RemoteOpenTargets.layer.pipe( Layer.provide(Layer.mergeAll(layerNet(input.sshd), layerSpawner(input.tailscale))), ), ), + Effect.provideService(HostProcess.Hostname, input.hostname), ); return input.username === undefined diff --git a/apps/server/src/environment/RemoteOpenTargets.ts b/apps/server/src/environment/RemoteOpenTargets.ts index 8b4997f695d15..e9ad02d6f87c0 100644 --- a/apps/server/src/environment/RemoteOpenTargets.ts +++ b/apps/server/src/environment/RemoteOpenTargets.ts @@ -3,11 +3,15 @@ import * as HostProcess from "@t3tools/shared/HostProcess"; import * as NetService from "@t3tools/shared/Net"; import { readTailscaleSshEnabled, readTailscaleStatus } from "@t3tools/tailscale"; import * as Context from "effect/Context"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import { makeBackgroundRefreshed } from "./backgroundRefresh.ts"; + const SSH_PORT = 22; +const TARGETS_MAX_AGE = Duration.seconds(30); export class RemoteOpenTargets extends Context.Service< RemoteOpenTargets, @@ -77,10 +81,13 @@ export const make = Effect.gen(function* () { return targets; }); + const cachedTargets = yield* makeBackgroundRefreshed(resolveTargets, TARGETS_MAX_AGE, { + warm: true, + }); return RemoteOpenTargets.of({ resolveTargets: Effect.fn("RemoteOpenTargets.resolveTargets")(function () { - return resolveTargets; + return cachedTargets; }), }); }); diff --git a/apps/server/src/environment/backgroundRefresh.test.ts b/apps/server/src/environment/backgroundRefresh.test.ts new file mode 100644 index 0000000000000..dbadf9b0c8b9b --- /dev/null +++ b/apps/server/src/environment/backgroundRefresh.test.ts @@ -0,0 +1,88 @@ +import { it } from "@effect/vitest"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Ref from "effect/Ref"; +import { TestClock } from "effect/testing"; +import { describe, expect } from "vite-plus/test"; + +import { makeBackgroundRefreshed } from "./backgroundRefresh.ts"; + +describe("makeBackgroundRefreshed", () => { + it.effect("shares the first compute and reuses it while fresh", () => + Effect.gen(function* () { + const computes = yield* Ref.make(0); + const read = yield* makeBackgroundRefreshed( + Ref.updateAndGet(computes, (n) => n + 1), + "30 seconds", + ); + + const results = yield* Effect.all([read, read, read], { concurrency: "unbounded" }); + yield* TestClock.adjust("29 seconds"); + const later = yield* read; + + expect(results).toEqual([1, 1, 1]); + expect(later).toBe(1); + expect(yield* Ref.get(computes)).toBe(1); + }).pipe(Effect.scoped), + ); + + it.effect("answers a stale read at once and refreshes behind it", () => + Effect.gen(function* () { + const computes = yield* Ref.make(0); + const gate = yield* Deferred.make(); + const read = yield* makeBackgroundRefreshed( + Effect.gen(function* () { + const n = yield* Ref.updateAndGet(computes, (count) => count + 1); + if (n > 1) yield* Deferred.await(gate); + return n; + }), + "30 seconds", + ); + + expect(yield* read).toBe(1); + yield* TestClock.adjust("30 seconds"); + // The refresh is blocked on the gate, so these reads only pass if they skip waiting for it. + expect(yield* read).toBe(1); + expect(yield* read).toBe(1); + + yield* Deferred.succeed(gate, undefined); + yield* TestClock.adjust("1 millis"); + expect(yield* read).toBe(2); + expect(yield* Ref.get(computes)).toBe(2); + }).pipe(Effect.scoped), + ); + + it.effect("releases first readers after a defect and lets later reads retry", () => + Effect.gen(function* () { + const computes = yield* Ref.make(0); + const read = yield* makeBackgroundRefreshed( + Effect.gen(function* () { + const attempt = yield* Ref.updateAndGet(computes, (n) => n + 1); + if (attempt === 1) return yield* Effect.die("probe failed"); + return attempt; + }), + "30 seconds", + ); + const reader = yield* Effect.forkChild(Effect.exit(read)); + yield* TestClock.adjust("1 millis"); + const result = reader.pollUnsafe(); + expect(result).toMatchObject({ _tag: "Success", value: { _tag: "Failure" } }); + expect(yield* read).toBe(2); + }).pipe(Effect.scoped), + ); + + it.effect("warm computes before the first read", () => + Effect.gen(function* () { + const started = yield* Deferred.make(); + const read = yield* makeBackgroundRefreshed( + Deferred.succeed(started, undefined).pipe(Effect.as("targets")), + "30 seconds", + { warm: true }, + ); + + yield* Deferred.await(started); + yield* TestClock.adjust("1 millis"); + expect(yield* read).toBe("targets"); + }).pipe(Effect.scoped), + ); +}); diff --git a/apps/server/src/environment/backgroundRefresh.ts b/apps/server/src/environment/backgroundRefresh.ts new file mode 100644 index 0000000000000..1569409c70386 --- /dev/null +++ b/apps/server/src/environment/backgroundRefresh.ts @@ -0,0 +1,78 @@ +import * as Clock from "effect/Clock"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Option from "effect/Option"; +import * as Ref from "effect/Ref"; + +interface RefreshState { + readonly value: Option.Option; + readonly computedAtNanos: bigint; + readonly pending: Deferred.Deferred | undefined; +} + +/** + * Serves the last computed value at once and recomputes it in the background once it is older + * than `maxAge`. Only reads before the first compute finishes wait; `warm` starts that compute + * now. Server config is on every client's connect path, so slow host probes must not run inside it. + */ +export const makeBackgroundRefreshed = Effect.fnUntraced(function* ( + compute: Effect.Effect, + maxAge: Duration.Input, + options?: { readonly warm?: boolean }, +) { + const scope = yield* Effect.scope; + const maxAgeNanos = Duration.toNanosUnsafe(maxAge); + const state = yield* Ref.make>({ + value: Option.none(), + computedAtNanos: 0n, + pending: undefined, + }); + + const startRefresh = Effect.gen(function* () { + const pending = yield* Deferred.make(); + const selected = yield* Ref.modify(state, (current) => + current.pending === undefined + ? ([pending, { ...current, pending }] as const) + : ([current.pending, current] as const), + ); + if (selected === pending) { + yield* compute.pipe( + Effect.onExit((exit) => + Effect.gen(function* () { + if (Exit.isSuccess(exit)) { + const computedAtNanos = yield* Clock.monotonicTimeNanos; + yield* Ref.set(state, { + value: Option.some(exit.value), + computedAtNanos, + pending: undefined, + }); + } else { + yield* Ref.update(state, (current) => ({ ...current, pending: undefined })); + } + yield* Deferred.done(pending, exit); + }), + ), + Effect.exit, + Effect.interruptible, + Effect.forkIn(scope), + ); + } + return selected; + }).pipe(Effect.uninterruptible); + + if (options?.warm === true) yield* startRefresh; + + return Effect.gen(function* () { + const current = yield* Ref.get(state); + if (Option.isNone(current.value)) { + return yield* Deferred.await(yield* startRefresh); + } + const now = yield* Clock.monotonicTimeNanos; + if (now - current.computedAtNanos >= maxAgeNanos) { + yield* startRefresh; + } + return current.value.value; + }); +}); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ac60becc2d78a..fcf1e5fd82cb7 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -820,12 +820,14 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( _tag: "Request", id: configRequest, tag: WS_METHODS.subscribeServerConfig, - payload: { workspaceSnapshotsOnDemand: true }, + payload: { workspaceSnapshotsOnDemand: true, providerCatalogHashes: {} }, headers: Headers.empty, }); const snapshot = (yield* Queue.take(responses)) as ServerConfigStreamEvent; if (snapshot.type !== "snapshot") return yield* Effect.die("Missing initial snapshot"); assert.isUndefined(snapshot.config.providers[0]?.workspaceSnapshots); + assert.isDefined(snapshot.config.providers[0]?.catalogHash); + assert.deepEqual(snapshot.config.providers[0]?.slashCommands, provider.slashCommands); yield* server.write(0, { _tag: "Ack", requestId: configRequest }); // A rescan that only touches a snapshot sends nothing; the next real change does. @@ -841,6 +843,36 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( if (update.type !== "providerStatuses") return yield* Effect.die("Expected provider update"); assert.equal(update.payload.providers[0]?.status, "warning"); assert.isUndefined(update.payload.providers[0]?.workspaceSnapshots); + assert.isTrue(update.payload.providers[0]?.catalogOmitted); + assert.deepEqual(update.payload.providers[0]?.slashCommands, []); + + yield* server.write(0, { _tag: "Ack", requestId: configRequest }); + yield* Queue.offer(providerChanges, [ + { ...provider, slashCommands: [{ name: "new-command" }] }, + ]); + yield* Queue.take(pulled); + yield* TestClock.adjust("1 second"); + const catalogUpdate = (yield* Queue.take(responses)) as ServerConfigStreamEvent; + if (catalogUpdate.type !== "providerStatuses") + return yield* Effect.die("Expected provider update"); + assert.deepEqual(catalogUpdate.payload.providers[0]?.slashCommands, [{ name: "new-command" }]); + assert.isUndefined(catalogUpdate.payload.providers[0]?.catalogOmitted); + + const reconnectRequest = RpcMessage.RequestId("reconnect"); + yield* server.write(0, { + _tag: "Request", + id: reconnectRequest, + tag: WS_METHODS.subscribeServerConfig, + payload: { + providerCatalogHashes: { [instanceId]: snapshot.config.providers[0]?.catalogHash }, + }, + headers: Headers.empty, + }); + const reconnect = (yield* Queue.take(responses)) as ServerConfigStreamEvent; + if (reconnect.type !== "snapshot") return yield* Effect.die("Missing reconnect snapshot"); + assert.isTrue(reconnect.config.providers[0]?.catalogOmitted); + assert.deepEqual(reconnect.config.providers[0]?.slashCommands, []); + yield* server.write(0, { _tag: "Interrupt", requestId: reconnectRequest, interruptors: [] }); yield* server.write(0, { _tag: "Request", diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index e1d6303b23506..7904473a45b01 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1,4 +1,6 @@ import { OrchestrationDispatchCommandError } from "@t3tools/contracts"; +// @effect-diagnostics-next-line nodeBuiltinImport:off -- Catalog hashes are computed synchronously while mapping stream events. +import * as NodeCrypto from "node:crypto"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as Orchestrator from "./orchestration-v2/Orchestrator.ts"; @@ -367,6 +369,27 @@ function withoutWorkspaceSnapshots(providers: ReadonlyArray): Se return providers.map(({ workspaceSnapshots: _workspaceSnapshots, ...provider }) => provider); } +function compactProviderCatalogs( + event: ServerConfigStreamEvent, + known: Map, +): ServerConfigStreamEvent { + if (event.type !== "snapshot" && event.type !== "providerStatuses") return event; + const providers = event.type === "snapshot" ? event.config.providers : event.payload.providers; + const compact = providers.map((provider) => { + const catalogHash = NodeCrypto.createHash("sha256") + .update(JSON.stringify([provider.slashCommands, provider.skills])) + .digest("hex"); + const omitted = known.get(provider.instanceId) === catalogHash; + known.set(provider.instanceId, catalogHash); + return omitted + ? { ...provider, catalogHash, catalogOmitted: true, slashCommands: [], skills: [] } + : { ...provider, catalogHash }; + }); + return event.type === "snapshot" + ? { ...event, config: { ...event.config, providers: compact } } + : { ...event, payload: { ...event.payload, providers: compact } }; +} + function unexpectedCompatibilityError(error: never): never { throw new Error(`Unhandled compatibility error: ${String(error)}`); } @@ -3164,10 +3187,15 @@ export const layerWsRpc = ( ), ); - return Stream.concat( + const configEvents = Stream.concat( rpcInitialItems([{ version: 1 as const, type: "snapshot" as const, config }]), withLateEditorConfig(config, liveUpdates, externalLauncher), ); + if (input.providerCatalogHashes === undefined) return configEvents; + const known = new Map(Object.entries(input.providerCatalogHashes)); + return configEvents.pipe( + Stream.map((event) => compactProviderCatalogs(event, known)), + ); }), ), [WS_METHODS.subscribeServerLifecycle]: (_input) => diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 7d9db4afdc494..ec7ca7b895aec 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -16,6 +16,7 @@ import { replaceCatalogValue, Persistence, makeStoredShellSnapshotEncoder, + withVerifiedServerVersion, } from "@t3tools/client-runtime/platform"; import { TokenStore } from "@t3tools/client-runtime/authorization"; import { @@ -747,7 +748,7 @@ export const layer = Layer.effectContext( Effect.mapError((cause) => persistenceError("load-shell", cause)), Effect.map((stored) => stored.environmentId === environmentId - ? Option.some(stored.snapshot) + ? Option.some(withVerifiedServerVersion(stored)) : Option.none(), ), ), @@ -758,11 +759,12 @@ export const layer = Layer.effectContext( ); }), saveShell: Effect.fn("storage.saveShell")( - function* (environmentId, snapshot) { + function* (environmentId, snapshot, verifiedServerVersion) { const encoded = yield* encodeStoredShellSnapshot({ schemaVersion: ORCHESTRATION_CACHE_SCHEMA_VERSION, environmentId, snapshot, + ...(verifiedServerVersion === undefined ? {} : { verifiedServerVersion }), }).pipe(Effect.mapError((cause) => persistenceError("save-shell", cause))); yield* writeDatabaseValue(database, SHELL_STORE_NAME, environmentId, encoded); }, @@ -967,7 +969,10 @@ export const layer = Layer.effectContext( Context.add(ProfileStore.ConnectionProfileStore, profileStore), Context.add(CredentialStore.ConnectionCredentialStore, credentialStore), Context.add(TokenStore.RemoteDpopAccessTokenStore, remoteTokenStore), - Context.add(Persistence.EnvironmentCacheStore, cacheStore), + Context.add( + Persistence.EnvironmentCacheStore, + Persistence.shareServerConfigReads(cacheStore), + ), ); }), ); diff --git a/packages/client-runtime/src/authorization/layer.test.ts b/packages/client-runtime/src/authorization/layer.test.ts index b98d019e1af07..e6f3f962f3504 100644 --- a/packages/client-runtime/src/authorization/layer.test.ts +++ b/packages/client-runtime/src/authorization/layer.test.ts @@ -24,6 +24,10 @@ import * as ConnectionCredentialStore from "../connection/credentialStore.ts"; import { BearerConnectionTarget } from "../connection/model.ts"; import * as ConnectionProfileStore from "../connection/profileStore.ts"; import * as ConnectionResolver from "../connection/resolver.ts"; +import * as ConnectionDriver from "../connection/driver.ts"; +import * as RpcSession from "../rpc/session.ts"; +import { ConnectionTransientError } from "../connection/model.ts"; +import { CheckedEnvironmentDescriptors } from "../environment/descriptor.ts"; const ENVIRONMENT_ID = EnvironmentId.make("environment-1"); const ENDPOINT = { @@ -385,6 +389,153 @@ describe("RemoteEnvironmentAuthorization", () => { }), ); + it.effect("trusts the descriptor its route check just fetched, but not another machine's", () => + Effect.gen(function* () { + const harness = yield* makeHarness({ responses: [websocketTicket("checked-ticket")] }); + const authorize = (environmentId: EnvironmentId) => + Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + return yield* remote.authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl: ENDPOINT.httpBaseUrl, + wsBaseUrl: ENDPOINT.wsBaseUrl, + bearerToken: "bearer-token", + connectionMethod: "direct", + }); + }).pipe( + Effect.provideService( + CheckedEnvironmentDescriptors, + new Map([ + [ + ENDPOINT.httpBaseUrl, + { + ...DESCRIPTOR, + platform: { os: "linux" as const, arch: "x64" as const }, + environmentId, + }, + ], + ]), + ), + Effect.provide(harness.layer), + Effect.result, + ); + + expect(yield* authorize(ENVIRONMENT_ID)).toMatchObject({ + _tag: "Success", + success: { socketUrl: expect.stringContaining("wsTicket=checked-ticket") }, + }); + expect(yield* authorize(EnvironmentId.make("someone-else"))).toMatchObject({ + _tag: "Failure", + failure: { _tag: "ConnectionBlockedError", reason: "configuration" }, + }); + expect(harness.fetch.calls.map(([url]) => new URL(String(url)).pathname)).toEqual([ + "/api/auth/websocket-ticket/rpc", + ]); + }), + ); + + it.effect( + "rechecks a LAN descriptor after another route attempt before sending credentials", + () => + Effect.gen(function* () { + const lanUrl = "http://192.168.1.2:3773"; + const route = (connectionId: string, httpBaseUrl: string) => ({ + target: new BearerConnectionTarget({ + environmentId: ENVIRONMENT_ID, + label: DESCRIPTOR.label, + connectionId, + }), + profile: Option.some( + new BearerConnectionProfile({ + connectionId, + environmentId: ENVIRONMENT_ID, + label: DESCRIPTOR.label, + httpBaseUrl, + wsBaseUrl: httpBaseUrl.replace("http", "ws"), + }), + ), + }); + const harness = yield* makeHarness({ + responses: [ + Response.json(DESCRIPTOR), + Response.json(DESCRIPTOR), + websocketTicket("first-ticket"), + Response.json({ ...DESCRIPTOR, environmentId: "another-machine" }), + ], + }); + const result = yield* Effect.gen(function* () { + const remote = yield* RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization; + const driver = yield* ConnectionDriver.ConnectionDriver.pipe( + Effect.provide( + ConnectionDriver.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.succeed(ConnectionResolver.ConnectionResolver, { + prepare: (entry) => + Effect.gen(function* () { + const profile = Option.getOrThrow(entry.profile); + if (profile._tag !== "BearerConnectionProfile") + return yield* Effect.die("Unexpected profile"); + return { + ...(yield* remote.authorizeBearer({ + expectedEnvironmentId: ENVIRONMENT_ID, + httpBaseUrl: profile.httpBaseUrl, + wsBaseUrl: profile.wsBaseUrl, + bearerToken: "bearer-token", + connectionMethod: "direct", + })), + target: entry.target, + }; + }), + prepareForUpdate: () => Effect.die("Unexpected update"), + }), + Layer.succeed(RpcSession.RpcSessionFactory, { + connect: () => + Effect.fail( + new ConnectionTransientError({ + reason: "endpoint-unavailable", + detail: "First route disconnected", + }), + ), + }), + ), + ), + ), + ), + ); + return yield* driver + .connect( + { + ...route("first", "http://127.0.0.1:3773"), + alternateRoutes: [route("lan", lanUrl)], + enabled: true, + }, + () => Effect.void, + ) + .pipe(Effect.result); + }).pipe(Effect.provide(harness.layer)); + + expect(result._tag).toBe("Failure"); + for (const [url, init] of harness.fetch.calls.filter(([url]) => + String(url).startsWith(lanUrl), + )) { + expect(new URL(String(url)).pathname).toBe("/.well-known/t3/environment"); + expect(new Headers(init.headers).has("authorization")).toBe(false); + expect(init.cache).toBe("no-store"); + } + expect( + harness.fetch.calls + .filter(([url]) => String(url).startsWith(lanUrl)) + .map(([url]) => new URL(String(url)).pathname), + ).toEqual(["/.well-known/t3/environment", "/.well-known/t3/environment"]); + expect( + harness.fetch.calls.filter(([url]) => + String(url).endsWith("/api/auth/websocket-ticket/rpc"), + ), + ).toHaveLength(1); + }), + ); + it.effect("reuses a same-account persisted environment token without contacting the relay", () => Effect.gen(function* () { const cached = new TokenStore.RemoteDpopAccessToken({ diff --git a/packages/client-runtime/src/authorization/service.ts b/packages/client-runtime/src/authorization/service.ts index 2962393d7b4a2..e8752707ac8b4 100644 --- a/packages/client-runtime/src/authorization/service.ts +++ b/packages/client-runtime/src/authorization/service.ts @@ -22,7 +22,10 @@ import { ConnectionTransientError, type ConnectionAttemptError, } from "../connection/model.ts"; -import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts"; +import { + CheckedEnvironmentDescriptors, + fetchRemoteEnvironmentDescriptor, +} from "../environment/descriptor.ts"; import { environmentEndpointUrl } from "../environment/endpoint.ts"; import * as ClientCapabilities from "../platform/capabilities.ts"; import * as ManagedRelay from "../relay/managedRelay.ts"; @@ -37,6 +40,7 @@ import * as Result from "effect/Result"; import * as Scope from "effect/Scope"; import * as Semaphore from "effect/Semaphore"; import * as HttpClient from "effect/http/HttpClient"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; import { DPOP_ACCESS_TOKEN_REFRESH_SKEW_MS, @@ -136,9 +140,13 @@ export const make = Effect.gen(function* () { readonly httpBaseUrl: string; readonly connectionMethod: ClientConnectionMethod; }) { - const descriptor = yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( - Effect.provideService(HttpClient.HttpClient, httpClient), - ); + const checked = (yield* CheckedEnvironmentDescriptors).get(input.httpBaseUrl); + const descriptor = + checked ?? + (yield* fetchDescriptor(input.httpBaseUrl, input.connectionMethod).pipe( + Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }), + Effect.provideService(HttpClient.HttpClient, httpClient), + )); if (descriptor.environmentId !== input.expectedEnvironmentId) { return yield* environmentMismatchError({ expected: input.expectedEnvironmentId, diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts index 0fd08c63364e7..dd9221fb78a7f 100644 --- a/packages/client-runtime/src/connection/driver.ts +++ b/packages/client-runtime/src/connection/driver.ts @@ -1,4 +1,6 @@ +import type { ExecutionEnvironmentDescriptor } from "@t3tools/contracts"; import * as Context from "effect/Context"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; @@ -23,7 +25,10 @@ import { usableRoutes, } from "./routes.ts"; import * as RpcSession from "../rpc/session.ts"; -import { fetchRemoteEnvironmentDescriptor } from "../environment/descriptor.ts"; +import { + CheckedEnvironmentDescriptors, + fetchRemoteEnvironmentDescriptor, +} from "../environment/descriptor.ts"; export type ConnectionDriverProgress = | { @@ -46,6 +51,10 @@ const ROUTE_CHECK_TIMEOUT_MS = 2_500; // wait the full probe timeout for a LAN the device has already left. A route // that misses it connects over relay, then upgrades with a second reconnect. const ROUTE_CHECK_HEAD_START_MS = 500; +// Once a route proves itself reachable, a preferred route still checking gets +// only this much longer. A live LAN answers within a few milliseconds, and a +// slower one is picked up by the upgrade check that runs after connecting. +const ROUTE_CHECK_GRACE_AFTER_ANSWER_MS = 150; export class ConnectionDriver extends Context.Service< ConnectionDriver, @@ -71,11 +80,29 @@ export const connectOverRoutes = Effect.fn("ConnectionDriver.connectOverRoutes") ) => Effect.Effect, ) { const routes = usableRoutes(entry); + const firstAnswer = yield* Deferred.make(); const checks = routes.length === 1 ? [] - : yield* Effect.forEach(routes, (route) => Effect.forkChild(checkRoute(route))); - const headStart = yield* Effect.forkChild(Effect.sleep(ROUTE_CHECK_HEAD_START_MS)); + : yield* Effect.forEach(routes, (route) => + Effect.forkChild( + checkRoute(route).pipe( + Effect.tap((check) => + check === "answered" ? Deferred.succeed(firstAnswer, undefined) : Effect.void, + ), + ), + ), + ); + // Unchecked routes (T3 Connect) prove nothing, so only a real answer shortens the head start. + const headStart = yield* Effect.forkChild( + Deferred.await(firstAnswer).pipe( + Effect.andThen(Effect.sleep(ROUTE_CHECK_GRACE_AFTER_ANSWER_MS)), + Effect.timeoutOrElse({ + duration: ROUTE_CHECK_HEAD_START_MS, + orElse: () => Effect.void, + }), + ), + ); const attemptScope = yield* Scope.Scope; let transient: ConnectionAttemptError | undefined; let blocked: ConnectionAttemptError | undefined; @@ -143,24 +170,36 @@ export const make = Effect.gen(function* () { // The descriptor is public, so this sends no credential to whatever // answers at a saved LAN address on a different network. - const checkDescriptor = (httpBaseUrl: string, environmentId: string) => + const checkDescriptor = ( + httpBaseUrl: string, + environmentId: string, + onAnswer: (descriptor: ExecutionEnvironmentDescriptor) => void = () => {}, + ) => fetchRemoteEnvironmentDescriptor({ httpBaseUrl, timeoutMs: ROUTE_CHECK_TIMEOUT_MS, }).pipe( - Effect.map((descriptor): RouteCheck => - descriptor.environmentId === environmentId ? "answered" : "silent", - ), + Effect.map((descriptor): RouteCheck => { + if (descriptor.environmentId !== environmentId) return "silent"; + onAnswer(descriptor); + return "answered"; + }), Effect.orElseSucceed((): RouteCheck => "silent"), // A cached descriptor would answer for a host that cannot be reached. Effect.provideService(FetchHttpClient.RequestInit, { cache: "no-store" }), Effect.provideService(HttpClient.HttpClient, httpClient), ); - const checkRoute = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => { + const checkRoute = ( + entry: ConnectionCatalogEntry, + route: ConnectionRoute, + checked: Map, + ) => { const httpBaseUrl = routeHttpBaseUrl(route); if (httpBaseUrl === null) return Effect.succeed("unchecked"); - return checkDescriptor(httpBaseUrl, entry.target.environmentId).pipe( + return checkDescriptor(httpBaseUrl, entry.target.environmentId, (descriptor) => + checked.set(httpBaseUrl, descriptor), + ).pipe( Effect.withSpan("ConnectionDriver.checkRoute", { attributes: { "connection.target.kind": route.target._tag }, }), @@ -185,11 +224,18 @@ export const make = Effect.gen(function* () { "connection.route.count": connectionRoutes(entry).length, }); yield* reportProgress({ stage: "preparing" }); + const checked = new Map(); + let firstRouteAttempt = true; return yield* connectOverRoutes( entry, - (route) => checkRoute(entry, route), + (route) => checkRoute(entry, route, checked), Effect.fnUntraced(function* (route) { - const prepared = yield* resolver.prepare(routeEntry(entry, route)); + // Another route attempt may outlive the network where these probes answered. + const descriptors = firstRouteAttempt ? checked : new Map(); + firstRouteAttempt = false; + const prepared = yield* resolver + .prepare(routeEntry(entry, route)) + .pipe(Effect.provideService(CheckedEnvironmentDescriptors, descriptors)); yield* reportProgress({ stage: "opening", prepared }); const session = yield* sessions.connect(prepared); yield* reportProgress({ stage: "synchronizing", prepared }); @@ -199,14 +245,17 @@ export const make = Effect.gen(function* () { ); }); - const preflight = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => - (sendsTokenInClear(connectionRoutes(entry), route) - ? Effect.succeed("silent" as const) - : checkRoute(entry, route) + const preflight = (entry: ConnectionCatalogEntry, route: ConnectionRoute) => { + const checked = new Map(); + return ( + sendsTokenInClear(connectionRoutes(entry), route) + ? Effect.succeed("silent" as const) + : checkRoute(entry, route, checked) ).pipe( Effect.flatMap((check) => check === "answered" ? resolver.prepare(routeEntry(entry, route)).pipe( + Effect.provideService(CheckedEnvironmentDescriptors, checked), Effect.as(true), Effect.orElseSucceed(() => false), ) @@ -214,6 +263,7 @@ export const make = Effect.gen(function* () { ), Effect.withSpan("ConnectionDriver.preflight"), ); + }; return ConnectionDriver.of({ connect, preflight, checkHost }); }); diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts index 347bde9cc44c8..5f320e99fa2ce 100644 --- a/packages/client-runtime/src/connection/supervisor.test.ts +++ b/packages/client-runtime/src/connection/supervisor.test.ts @@ -97,6 +97,29 @@ const LAN_THEN_RELAY_ENTRY: ConnectionCatalogEntry = { alternateRoutes: [{ target: RELAY_TARGET, profile: Option.none() }], enabled: true, }; +const TAILNET_TARGET = new BearerConnectionTarget({ + environmentId: TARGET.environmentId, + label: TARGET.label, + connectionId: "bearer:tailnet", +}); +const LAN_THEN_TAILNET_ENTRY: ConnectionCatalogEntry = { + ...LAN_THEN_RELAY_ENTRY, + alternateRoutes: [ + { + target: TAILNET_TARGET, + profile: Option.some( + new BearerConnectionProfile({ + connectionId: TAILNET_TARGET.connectionId, + environmentId: TARGET.environmentId, + label: TARGET.label, + httpBaseUrl: "http://100.100.1.2:3773/", + wsBaseUrl: "ws://100.100.1.2:3773/", + network: "tailscale", + }), + ), + }, + ], +}; const LAN_ENTRY: ConnectionCatalogEntry = { target: LAN_ROUTE.target, @@ -516,6 +539,29 @@ describe("EnvironmentSupervisor", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("retries a direct host soon after it drops a stable session", () => + Effect.gen(function* () { + const reconnectsWithin = Effect.fn("test.reconnectsWithin")(function* ( + entry: ConnectionCatalogEntry, + ) { + const harness = yield* makeHarness(); + const supervisor = yield* EnvironmentSupervisor.make(entry, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + yield* TestClock.adjust("31 seconds"); + yield* harness.closeLatestSession(); + yield* awaitState(supervisor.state, (state) => state.phase === "backoff"); + yield* TestClock.adjust("600 millis"); + return (yield* SubscriptionRef.get(supervisor.state)).phase !== "backoff"; + }); + + // A restarting host is back within a second or two; a relay keeps the normal ladder. + expect(yield* Effect.scoped(reconnectsWithin(TARGET_ENTRY))).toBe(true); + expect(yield* Effect.scoped(reconnectsWithin(RELAY_ENTRY))).toBe(false); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("does not report a session lost while suspended as a failed connection", () => Effect.gen(function* () { const allowReconnect = yield* Deferred.make(); @@ -1474,6 +1520,62 @@ describe("EnvironmentSupervisor routes", () => { }).pipe(Effect.provide(TestClock.layer())), ); + it.effect("stops waiting for a dead LAN route soon after the tailnet answers", () => + Effect.gen(function* () { + const checkingLan = yield* Deferred.make(); + const harness = yield* makeHarness({ + checkRoute: (route) => + connectionRouteId(route.target) === LAN_TARGET.connectionId + ? Deferred.succeed(checkingLan, undefined).pipe( + Effect.andThen(Effect.sleep("2500 millis")), + Effect.as("silent" as const), + ) + : Effect.sleep("20 millis").pipe(Effect.as("answered" as const)), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_TAILNET_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + yield* Deferred.await(checkingLan); + // Well inside the 500 ms head start: the tailnet answer cut it short. + yield* TestClock.adjust("170 millis"); + yield* awaitState(supervisor.state, (state) => state.phase === "connected"); + expect( + connectionRouteId( + Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target, + ), + ).toBe(TAILNET_TARGET.connectionId); + }).pipe(Effect.provide(TestClock.layer())), + ); + + it.effect("moves to a LAN route that answers just after connecting elsewhere", () => + Effect.gen(function* () { + const lanChecks = yield* Ref.make(0); + const harness = yield* makeHarness({ + checkRoute: (route) => + route.target._tag === "BearerConnectionTarget" + ? Ref.updateAndGet(lanChecks, (count) => count + 1).pipe( + Effect.map((count) => (count === 1 ? "silent" : "answered")), + ) + : Effect.succeed("unchecked"), + prepare: (_attempt, target) => Effect.succeed(preparedFor(target)), + }); + const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, { + initiallyDesired: true, + }).pipe(Effect.provide(harness.dependencies)); + + // No clock advance: the upgrade must not wait for the periodic check. + yield* awaitState( + supervisor.state, + (state) => state.phase === "connected" && state.generation === 2, + ); + expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe( + "BearerConnectionTarget", + ); + }).pipe(Effect.provide(TestClock.layer())), + ); + it.effect("still tries a slow LAN route when relay fails", () => Effect.gen(function* () { const checkingLan = yield* Deferred.make(); diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts index ce7632df0f52b..f38a7db5c560b 100644 --- a/packages/client-runtime/src/connection/supervisor.ts +++ b/packages/client-runtime/src/connection/supervisor.ts @@ -40,6 +40,9 @@ import { } from "./routes.ts"; const RETRY_BASE_DELAY_MS = 1_000; +// A direct host that drops a long-lived session is usually restarting, and a refused +// connection costs it almost nothing. Relays and SSH tunnels keep the normal ladder. +const RESTART_RETRY_BASE_DELAY_MS = 250; const RETRY_MAX_DELAY_MS = 300_000; // A cold server with a large database can take longer than 15 seconds to // answer, and that is not a reason to drop a working connection. @@ -134,8 +137,8 @@ export interface EnvironmentSupervisorOptions { }) => Effect.Effect>; } -function retryDelayMs(failureCount: number, random: number): number { - const ceiling = Math.min(RETRY_MAX_DELAY_MS, RETRY_BASE_DELAY_MS * 2 ** (failureCount + 1)); +function retryDelayMs(failureCount: number, random: number, baseDelayMs: number): number { + const ceiling = Math.min(RETRY_MAX_DELAY_MS, baseDelayMs * 2 ** (failureCount + 1)); return Math.round(ceiling / 2 + (ceiling / 2) * random); } @@ -667,9 +670,10 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( ), ); // Ticks always run: the check skips the preferred route itself, and the - // route list can grow while connected (learned routes). + // route list can grow while connected (learned routes). The first tick is + // immediate, so a preferred route that answered just after the connect + // attempt moved past it is used now rather than a minute later. yield* Stream.tick(BETTER_ROUTE_CHECK_INTERVAL).pipe( - Stream.drop(1), Stream.runForEach(() => requestBetterRouteCheck(lease)), Effect.forkScoped, ); @@ -995,17 +999,18 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( let generation = 0; let latestFailure: ConnectionAttemptError | null = null; let pendingRetry = Option.none(); + let restartLadder = false; const resetRetryLadder = () => { failureCount = 0; pendingRetry = Option.none(); + restartLadder = false; }; let replacing = false; for (;;) { if (yield* Ref.getAndSet(resetRetryState, false)) { - failureCount = 0; + resetRetryLadder(); latestFailure = null; - pendingRetry = Option.none(); } const currentIntent = yield* Ref.get(intent); if (!currentIntent.desired) { @@ -1042,6 +1047,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( latestFailure = null; if (outcome.stable) { resetRetryLadder(); + restartLadder = (yield* currentRoutes).every( + (route) => + route.target._tag === "PrimaryConnectionTarget" || + route.target._tag === "BearerConnectionTarget", + ); } } if (outcome._tag === "Interrupted") { @@ -1083,7 +1093,11 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* ( } failureCount += 1; - const delayMs = retryDelayMs(failureCount - 1, yield* Random.next); + const delayMs = retryDelayMs( + failureCount - 1, + yield* Random.next, + restartLadder ? RESTART_RETRY_BASE_DELAY_MS : RETRY_BASE_DELAY_MS, + ); pendingRetry = Option.map(attemptSpan, (previousAttempt) => ({ previousAttempt, failureCount, diff --git a/packages/client-runtime/src/environment/descriptor.ts b/packages/client-runtime/src/environment/descriptor.ts index 1f92b296c5e33..15c775a77e97a 100644 --- a/packages/client-runtime/src/environment/descriptor.ts +++ b/packages/client-runtime/src/environment/descriptor.ts @@ -1,3 +1,5 @@ +import type { ExecutionEnvironmentDescriptor } from "@t3tools/contracts"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import { environmentEndpointUrl } from "./endpoint.ts"; @@ -5,6 +7,16 @@ import { executeEnvironmentHttpRequest, makeEnvironmentHttpApiGroupClient } from const DEFAULT_REMOTE_REQUEST_TIMEOUT_MS = 10_000; +/** + * Descriptors fetched by route checks, reusable only before the first route attempt. + * Verifying the same origin again before sending credentials would cost a round trip. + */ +export class CheckedEnvironmentDescriptors extends Context.Reference< + ReadonlyMap +>("@t3tools/client-runtime/environment/CheckedEnvironmentDescriptors", { + defaultValue: () => new Map(), +}) {} + export const fetchRemoteEnvironmentDescriptor = Effect.fn( "clientRuntime.environment.fetchRemoteEnvironmentDescriptor", )(function* (input: { readonly httpBaseUrl: string; readonly timeoutMs?: number }) { diff --git a/packages/client-runtime/src/platform/orchestrationCache.ts b/packages/client-runtime/src/platform/orchestrationCache.ts index e7800e20bd33a..031f949b71cf4 100644 --- a/packages/client-runtime/src/platform/orchestrationCache.ts +++ b/packages/client-runtime/src/platform/orchestrationCache.ts @@ -1,5 +1,6 @@ import { EnvironmentId, + type OrchestrationV2ShellSnapshot, OrchestrationV2ShellSnapshotJson, OrchestrationV2ThreadDetailSnapshot, OrchestrationV2ThreadProjectionJson, @@ -17,8 +18,19 @@ export const StoredOrchestrationShellSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshotJson, + // The server build whose stream last verified `snapshot`; a cold start on that build resumes. + verifiedServerVersion: Schema.optionalKey(Schema.String), }); +export function withVerifiedServerVersion(stored: { + readonly snapshot: S; + readonly verifiedServerVersion?: string; +}): S & { readonly verifiedServerVersion?: string } { + return stored.verifiedServerVersion === undefined + ? stored.snapshot + : { ...stored.snapshot, verifiedServerVersion: stored.verifiedServerVersion }; +} + export const StoredOrchestrationThreadSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(ORCHESTRATION_THREAD_CACHE_SCHEMA_VERSION), environmentId: EnvironmentId, diff --git a/packages/client-runtime/src/platform/persistence.ts b/packages/client-runtime/src/platform/persistence.ts index e043279d3e576..45af1d3685f35 100644 --- a/packages/client-runtime/src/platform/persistence.ts +++ b/packages/client-runtime/src/platform/persistence.ts @@ -86,6 +86,7 @@ export class EnvironmentCacheStore extends Context.Service< readonly saveShell: ( environmentId: EnvironmentId, snapshot: OrchestrationV2ShellSnapshot, + verifiedServerVersion?: string, ) => Effect.Effect; readonly loadThread: ( environmentId: EnvironmentId, @@ -139,6 +140,39 @@ export class EnvironmentCacheStore extends Context.Service< } >()("@t3tools/client-runtime/platform/persistence/EnvironmentCacheStore") {} +export function shareServerConfigReads(store: EnvironmentCacheStore["Service"]) { + const reads = new Map< + EnvironmentId, + Effect.Effect, ConnectionPersistenceError> + >(); + return EnvironmentCacheStore.of({ + ...store, + loadServerConfig: Effect.fn("EnvironmentCacheStore.sharedServerConfig")( + function* (environmentId) { + let read = reads.get(environmentId); + if (read === undefined) { + read = yield* Effect.cached(store.loadServerConfig(environmentId)); + reads.set(environmentId, read); + } + return yield* read.pipe( + Effect.tapError(() => Effect.sync(() => reads.delete(environmentId))), + ); + }, + ), + saveServerConfig: Effect.fn("EnvironmentCacheStore.sharedSaveServerConfig")( + (environmentId, config) => + store + .saveServerConfig(environmentId, config) + .pipe(Effect.tap(() => Effect.sync(() => reads.delete(environmentId)))), + ), + clear: Effect.fn("EnvironmentCacheStore.sharedClear")((environmentId) => + store + .clear(environmentId) + .pipe(Effect.tap(() => Effect.sync(() => reads.delete(environmentId)))), + ), + }); +} + export class EnvironmentOwnedDataCleanup extends Context.Reference<{ readonly clear: (environmentId: EnvironmentId) => Effect.Effect; }>("@t3tools/client-runtime/platform/persistence/EnvironmentOwnedDataCleanup", { diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index adca6b40b184a..a31f42a915cdf 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -1,6 +1,8 @@ import { DEFAULT_SERVER_SETTINGS, EnvironmentId, + ProviderInstanceId, + ProviderDriverKind, ServerConfig, type ServerConfig as ServerConfigType, ServerConfigStreamEvent, @@ -200,6 +202,7 @@ const SOURCE_EVENT: ServerConfigStreamEventType = { const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( options: RpcSession.RpcSessionOptions = {}, + cachedConfig?: ServerConfigType, ) { const sentChanges = yield* Queue.unbounded(); const sockets = Object.assign([] as TestWebSocket[], { changes: yield* Queue.unbounded() }); @@ -209,7 +212,16 @@ const makeFactory = Effect.fn("TestRpcSessionFactory.make")(function* ( Queue.offerUnsafe(sockets.changes, undefined); return socket as unknown as globalThis.WebSocket; }); - const layer = RpcSession.layer(options).pipe(Layer.provide(layerConstructor)); + const layer = RpcSession.layer(options).pipe( + Layer.provide( + Layer.mergeAll( + layerConstructor, + Layer.mock(Persistence.EnvironmentCacheStore)({ + loadServerConfig: () => Effect.succeed(Option.fromNullishOr(cachedConfig)), + }), + ), + ), + ); const factory = yield* RpcSession.RpcSessionFactory.pipe(Effect.provide(layer)); return { factory, sockets }; }); @@ -275,6 +287,89 @@ const publishConfigEvents = Effect.fn("TestRpcSessionFactory.publishConfigEvents }); describe("RpcSessionFactory", () => { + it.effect("shares cached config reads and reloads after a save", () => + Effect.gen(function* () { + let reads = 0; + let stored = SERVER_CONFIG; + const raw = yield* Persistence.EnvironmentCacheStore.pipe( + Effect.provide( + Layer.mock(Persistence.EnvironmentCacheStore)({ + loadServerConfig: () => + Effect.sync(() => { + reads++; + return Option.some(stored); + }), + saveServerConfig: (_environmentId, config) => + Effect.sync(() => { + stored = config; + }), + }), + ), + ); + const cache = Persistence.shareServerConfigReads(raw); + yield* Effect.all( + [ + cache.loadServerConfig(TARGET.environmentId), + cache.loadServerConfig(TARGET.environmentId), + ], + { concurrency: "unbounded" }, + ); + expect(reads).toBe(1); + const updated = { ...SERVER_CONFIG, cwd: "/updated" }; + yield* cache.saveServerConfig(TARGET.environmentId, updated); + expect(Option.getOrNull(yield* cache.loadServerConfig(TARGET.environmentId))).toEqual( + updated, + ); + expect(reads).toBe(2); + }), + ); + it.effect("restores a cached provider catalog omitted by the server", () => + Effect.gen(function* () { + const provider = { + instanceId: ProviderInstanceId.make("claudeAgent"), + driver: ProviderDriverKind.make("claudeAgent"), + enabled: true, + installed: true, + version: "1.0.0", + status: "ready" as const, + auth: { status: "authenticated" as const }, + checkedAt: "2026-10-25T00:00:00.000Z", + models: [], + slashCommands: [{ name: "command", description: "Description" }], + skills: [], + catalogHash: "hash", + }; + const cached = { ...SERVER_CONFIG, providers: [provider] }; + const { factory, sockets } = yield* makeFactory({}, cached); + const session = yield* factory.connect(PREPARED); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig( + socket, + encodeServerConfig({ + ...SERVER_CONFIG, + providers: [{ ...provider, slashCommands: [], catalogOmitted: true }], + }), + { providerCatalogHashes: { claudeAgent: "hash" } }, + ); + const config = yield* session.initialConfig; + expect(config.providers).toEqual([provider]); + + const replacement = yield* factory.connect(PREPARED); + const replacementSocket = yield* awaitSocket(sockets, 1); + replacementSocket.open(); + yield* completeInitialConfig( + replacementSocket, + encodeServerConfig({ + ...SERVER_CONFIG, + providers: [ + { ...provider, catalogHash: "different", slashCommands: [], catalogOmitted: true }, + ], + }), + ); + expect(Exit.isFailure(yield* Effect.exit(replacement.initialConfig))).toBe(true); + }).pipe(Effect.scoped), + ); it.effect("owns one scoped websocket attempt and exposes readiness and closure", () => Effect.gen(function* () { const { factory, sockets } = yield* makeFactory(); diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 250c7889c6fce..81403ed54cdf1 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -1,6 +1,7 @@ import { type ServerConfig, type ServerConfigStreamEvent, + type ServerProvider, WsSubscribeServerConfigRpc, WS_METHODS, } from "@t3tools/contracts"; @@ -42,6 +43,7 @@ import { withoutEnvironmentThemes, } from "../state/serverConfigProjection.ts"; import { environmentMismatchError } from "../connection/errors.ts"; +import * as Persistence from "../platform/persistence.ts"; const SOCKET_OPEN_TIMEOUT = "15 seconds"; @@ -167,6 +169,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( options: RpcSessionOptions = {}, ) { const webSocketConstructor = yield* Socket.WebSocketConstructor; + const cache = yield* Effect.serviceOption(Persistence.EnvironmentCacheStore); const serverConfigInput: ServerConfigSubscriptionInput = { ...(options.environmentThemes === true ? { environmentThemes: true } : {}), ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), @@ -262,13 +265,55 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( const configSubscriptionClosed = yield* Deferred.make(); const serverConfigState = yield* Ref.make(Option.none()); const serverConfigUpdates = yield* PubSub.sliding(64); + const cachedConfig = Option.isSome(cache) + ? yield* cache.value + .loadServerConfig(connection.environmentId) + .pipe(Effect.orElseSucceed(() => Option.none())) + : Option.none(); + const catalogs = new Map( + Option.getOrUndefined(cachedConfig)?.providers.map((provider) => [ + provider.instanceId, + provider, + ]) ?? [], + ); + const hashes = Object.fromEntries( + [...catalogs.values()].flatMap((provider) => + provider.catalogHash === undefined ? [] : [[provider.instanceId, provider.catalogHash]], + ), + ); + const restoreCatalogs = (event: ServerConfigStreamEvent): ServerConfigStreamEvent => { + if (event.type !== "snapshot" && event.type !== "providerStatuses") return event; + const providers = + event.type === "snapshot" ? event.config.providers : event.payload.providers; + const restored = providers.map((provider) => { + const previous = catalogs.get(provider.instanceId); + if ( + provider.catalogOmitted && + (previous === undefined || previous.catalogHash !== provider.catalogHash) + ) { + throw new Error(`Missing provider catalog for ${provider.instanceId}`); + } + const { catalogOmitted: _omitted, ...rest } = provider; + const next = + provider.catalogOmitted && previous !== undefined + ? { ...rest, slashCommands: previous.slashCommands, skills: previous.skills } + : rest; + catalogs.set(provider.instanceId, next); + return next; + }); + return event.type === "snapshot" + ? { ...event, config: { ...event.config, providers: restored } } + : { ...event, payload: { ...event.payload, providers: restored } }; + }; const configSubscriptionEndedError = new ConnectionTransientErrorClass({ reason: "remote-unavailable", detail: `${connection.label} config subscription ended.`, }); - const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig]( - onDemandSnapshots(serverConfigInput), - ).pipe( + const serverConfigSource = protocolClient[WS_METHODS.subscribeServerConfig]({ + ...onDemandSnapshots(serverConfigInput), + providerCatalogHashes: hashes, + }).pipe( + Stream.map(restoreCatalogs), Stream.runForEach((event) => Effect.gen(function* () { const buffered = yield* Ref.modify(serverConfigState, (current) => { diff --git a/packages/client-runtime/src/state/shell-sync.test.ts b/packages/client-runtime/src/state/shell-sync.test.ts index 9d1df33d1f70b..cac0f7b47133c 100644 --- a/packages/client-runtime/src/state/shell-sync.test.ts +++ b/packages/client-runtime/src/state/shell-sync.test.ts @@ -7,6 +7,7 @@ import { type ThreadPullRequestLink, } from "@t3tools/contracts"; import { describe, expect, it } from "@effect/vitest"; +import { vi } from "vite-plus/test"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -18,6 +19,7 @@ import * as Scope from "effect/Scope"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as TestClock from "effect/testing/TestClock"; +import * as Tracer from "effect/Tracer"; import { AVAILABLE_CONNECTION_STATE, @@ -29,6 +31,7 @@ import * as Persistence from "../platform/persistence.ts"; import * as RpcSession from "../rpc/session.ts"; import type { WsRpcProtocolClient } from "../rpc/protocol.ts"; import { type EnvironmentShellState, makeEnvironmentShellState } from "./shell.ts"; +import type { DeferredShellSnapshot } from "./shellPullRequests.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; import { v2Project, v2ShellSnapshot, v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; @@ -122,6 +125,96 @@ const makePersistenceHarness = Effect.fn("test.makeShellPersistenceHarness")(fun return { state, send }; }); +const FAMILY_SNAPSHOT: OrchestrationV2ShellSnapshot = { + ...LIVE_SHELL_SNAPSHOT, + olderSettled: { before: "2026-01-01T00:00:00.000Z", asOfSequence: 1, counts: [] }, +}; + +/** A server that omits idle subagents and older settled threads, so cursors can resume. */ +const makeFamilyHarness = Effect.fn("test.makeFamilyHarness")(function* (input: { + readonly cached: Option.Option; + readonly initialVersion: string; + readonly snapshot?: DeferredShellSnapshot; +}) { + const subscribeInputs = yield* Queue.unbounded(); + const saved = yield* Queue.unbounded<{ readonly verifiedServerVersion?: string }>(); + const httpLoads = yield* Ref.make(0); + const serverSnapshot = yield* Ref.make(input.snapshot ?? FAMILY_SNAPSHOT); + const familySession = ( + serverVersion: string, + events = Stream.make>({ + kind: "synchronized", + }).pipe(Stream.concat(Stream.never)), + ): RpcSession.RpcSession => { + const client = { + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (subscribeInput: unknown) => + Stream.unwrap(Queue.offer(subscribeInputs, subscribeInput).pipe(Effect.as(events))), + } as unknown as WsRpcProtocolClient; + return { + ...session(client), + initialConfig: Effect.succeed({ + shellResumeCompletionMarker: true, + shellThreadFamilies: true, + shellOlderSettled: true, + environment: { serverVersion }, + } as never), + }; + }; + const sessions = yield* SubscriptionRef.make(Option.some(familySession(input.initialVersion))); + const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({ + target: TARGET, + state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE), + session: sessions, + prepared: yield* SubscriptionRef.make(Option.some(PREPARED)), + connect: Effect.void, + disconnect: Effect.void, + retryNow: Effect.void, + } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]); + const cache = Persistence.EnvironmentCacheStore.of({ + loadShell: () => Effect.succeed(input.cached), + saveShell: (_environmentId, _snapshot, verifiedServerVersion) => + Queue.offer(saved, verifiedServerVersion === undefined ? {} : { verifiedServerVersion }), + loadThread: () => Effect.succeedNone, + saveThread: () => Effect.void, + removeThread: () => Effect.void, + loadServerConfig: () => Effect.succeedNone, + saveServerConfig: () => Effect.void, + loadVcsRefs: () => Effect.succeedNone, + saveVcsRefs: () => Effect.void, + removeVcsRefs: () => Effect.void, + clearVcsRefs: () => Effect.void, + clear: () => Effect.void, + }); + const snapshotLoader = ShellSnapshotLoader.ShellSnapshotLoader.of({ + load: () => + Ref.update(httpLoads, (count) => count + 1).pipe( + Effect.andThen(Ref.get(serverSnapshot)), + Effect.asSome, + ), + }); + const shellState = yield* makeEnvironmentShellState().pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.provideService(Persistence.EnvironmentCacheStore, cache), + Effect.provideService(ShellSnapshotLoader.ShellSnapshotLoader, snapshotLoader), + ); + const stateWhere = (predicate: (state: EnvironmentShellState) => boolean) => + SubscriptionRef.changes(shellState).pipe( + Stream.filter(predicate), + Stream.runHead, + Effect.map(Option.getOrThrow), + ); + return { + sessions, + familySession, + subscribeInputs, + saved, + httpLoads, + stateWhere, + shellState, + serverSnapshot, + }; +}); + describe("environment shell synchronization", () => { it.effect("skips unchanged shell writes, persists resets and flushes changed rows", () => Effect.gen(function* () { @@ -726,6 +819,142 @@ describe("environment shell synchronization", () => { }), ); + it.effect("reloads restored history on a replacement session even when the build matches", () => + Effect.gen(function* () { + const h = yield* makeFamilyHarness({ cached: Option.none(), initialVersion: "1.0.0" }); + + yield* Queue.take(h.subscribeInputs); + const first = yield* h.stateWhere((state) => state.status === "live"); + + yield* SubscriptionRef.set(h.sessions, yield* SubscriptionRef.get(h.sessions)); + yield* Queue.take(h.subscribeInputs); + expect(yield* Ref.get(h.httpLoads)).toBe(1); + yield* Ref.set(h.serverSnapshot, { + ...FAMILY_SNAPSHOT, + projects: [{ ...v2Project, title: "Restored history" }], + }); + yield* SubscriptionRef.set(h.sessions, Option.some(h.familySession("1.0.0"))); + expect(yield* Queue.take(h.subscribeInputs)).toMatchObject({ afterSequence: 1 }); + const reloaded = yield* h.stateWhere( + (state) => state.status === "live" && state.generation > first.generation, + ); + expect(Option.getOrThrow(reloaded.snapshot).projects[0]?.title).toBe("Restored history"); + expect(yield* Ref.get(h.httpLoads)).toBe(2); + + // A different build may shape rows differently, so it reloads the snapshot. + yield* SubscriptionRef.set(h.sessions, Option.some(h.familySession("1.1.0"))); + yield* Queue.take(h.subscribeInputs); + expect(yield* Ref.get(h.httpLoads)).toBe(3); + }), + ); + + it.effect( + "does not verify a replacement session with a buffered snapshot from its predecessor", + () => + Effect.gen(function* () { + const consumerStarted = yield* Deferred.make(); + let observeConsumer = false; + // Effect.fn starts its span before acquiring the snapshot lock. + const tracer = Tracer.make({ + span: (options) => { + if (observeConsumer && options.name === "EnvironmentShellState.applyItems") { + Deferred.doneUnsafe(consumerStarted, Exit.succeed(undefined)); + } + return new Tracer.NativeSpan(options); + }, + }); + const links = + yield* Deferred.make>>(); + const h = yield* makeFamilyHarness({ + cached: Option.none(), + initialVersion: "1.0.0", + snapshot: { ...FAMILY_SNAPSHOT, loadPullRequests: Deferred.await(links) }, + }).pipe(Effect.withTracer(tracer)); + yield* Queue.take(h.subscribeInputs); + yield* h.stateWhere((state) => state.status === "live"); + const configStarted = yield* Deferred.make(); + const configGate = yield* Deferred.make(); + const replacement = h.familySession("2.0.0"); + const next = { + ...replacement, + initialConfig: Deferred.succeed(configStarted, undefined).pipe( + Effect.andThen(Deferred.await(configGate)), + Effect.andThen(replacement.initialConfig), + ), + }; + const events = yield* Queue.unbounded(); + const old = h.familySession("1.0.0", Stream.fromQueue(events)); + yield* SubscriptionRef.set(h.sessions, Option.some(old)); + yield* Queue.take(h.subscribeInputs); + const held = yield* Deferred.make(); + const release = yield* Deferred.make(); + const writer = yield* Effect.forkChild( + h.shellState.semaphore.withPermit( + Deferred.succeed(held, undefined).pipe(Effect.andThen(Deferred.await(release))), + ), + ); + yield* Deferred.await(held); + // The fill reaches the state lock only after taking snapshotWriteLock. + const fillLocked = yield* Deferred.make(); + const withPermit = h.shellState.semaphore.withPermit.bind(h.shellState.semaphore); + using observeFill = vi + .spyOn(h.shellState.semaphore, "withPermit") + .mockImplementation((effect: Effect.Effect) => + Deferred.succeed(fillLocked, undefined).pipe(Effect.andThen(withPermit(effect))), + ); + yield* Deferred.succeed(links, new Map()); + yield* Deferred.await(fillLocked); + observeFill.mockRestore(); + observeConsumer = true; + yield* Queue.offer(events, { + kind: "snapshot", + snapshot: { ...FAMILY_SNAPSHOT, snapshotSequence: 99 }, + }); + yield* Deferred.await(consumerStarted); + yield* SubscriptionRef.set(h.sessions, Option.some(next)); + yield* Deferred.await(configStarted); + yield* Deferred.succeed(release, undefined); + yield* Fiber.join(writer); + yield* h.stateWhere((state) => + Option.exists(state.snapshot, (snapshot) => snapshot.snapshotSequence === 99), + ); + yield* Deferred.succeed(configGate, undefined); + expect(yield* Queue.take(h.subscribeInputs)).toMatchObject({ afterSequence: 1 }); + expect(yield* Ref.get(h.httpLoads)).toBe(3); + }), + ); + + it.effect("reloads cached history on a cold start even when the build matches", () => + Effect.gen(function* () { + const cached = { + ...FAMILY_SNAPSHOT, + verifiedServerVersion: "1.0.0", + projects: [{ ...v2Project, title: "Stale cached history" }], + }; + const sameBuild = yield* makeFamilyHarness({ + cached: Option.some(cached), + initialVersion: "1.0.0", + }); + expect(yield* Queue.take(sameBuild.subscribeInputs)).toMatchObject({ afterSequence: 1 }); + expect(yield* Ref.get(sameBuild.httpLoads)).toBe(1); + const reloaded = yield* sameBuild.stateWhere((state) => state.status === "live"); + expect(Option.getOrThrow(reloaded.snapshot).projects[0]?.title).toBe(v2Project.title); + + const updatedBuild = yield* makeFamilyHarness({ + cached: Option.some(cached), + initialVersion: "1.1.0", + }); + yield* Queue.take(updatedBuild.subscribeInputs); + expect(yield* Ref.get(updatedBuild.httpLoads)).toBe(1); + // The reloaded snapshot is saved under the build that served it. + yield* updatedBuild.stateWhere((state) => state.status === "live"); + yield* TestClock.adjust("10 seconds"); + expect(yield* Queue.take(updatedBuild.saved)).toMatchObject({ + verifiedServerVersion: "1.1.0", + }); + }), + ); + it.effect("fills deferred cached pull request links without overwriting newer rows", () => Effect.gen(function* () { const link = (number: number): ThreadPullRequestLink => ({ diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index bce13d315b425..d48ba83e6cc56 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -8,6 +8,7 @@ import { type ThreadId, type ThreadPullRequestLink, } from "@t3tools/contracts"; +import * as Array from "effect/Array"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Fiber from "effect/Fiber"; @@ -26,7 +27,7 @@ import * as ConnectionWakeups from "../connection/wakeups.ts"; import { safeErrorLogAttributes } from "../errors/safeLog.ts"; import * as Persistence from "../platform/persistence.ts"; import { runCachePersistence } from "./cachePersistence.ts"; -import { subscribeDynamic } from "../rpc/client.ts"; +import { subscribeDynamicWithSession } from "../rpc/client.ts"; import type { RpcSession } from "../rpc/session.ts"; import * as ShellSnapshotLoader from "./shellSnapshotHttp.ts"; import type { DeferredShellSnapshot } from "./shellPullRequests.ts"; @@ -74,8 +75,10 @@ function shellStatusForSnapshot( } function withoutDeferredPullRequests(value: DeferredShellSnapshot): OrchestrationV2ShellSnapshot { - if (value.loadPullRequests === undefined) return value; - const { loadPullRequests: _deferred, ...snapshot } = value; + if (value.loadPullRequests === undefined && value.verifiedServerVersion === undefined) { + return value; + } + const { loadPullRequests: _deferred, verifiedServerVersion: _verified, ...snapshot } = value; return snapshot; } @@ -107,13 +110,20 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const awaitingCompletion = yield* Ref.make(false); const lastAuthoritativeSession = yield* Ref.make(null); const activeSubscriptionSession = yield* Ref.make(null); + // Keep the server build that produced each authoritative snapshot as cache provenance. + const verifiedServerVersion = yield* Ref.make( + Option.getOrUndefined(cached)?.verifiedServerVersion ?? null, + ); + const activeServerVersion = yield* Ref.make(null); const latestLiveSnapshot = yield* Ref.make>( Option.none(), ); + let latestLiveServerVersion: string | null = null; const persistence = yield* Queue.sliding(1); const persistenceLock = yield* Semaphore.make(1); let lastPersisted: OrchestrationV2ShellSnapshot | undefined; + let lastPersistedServerVersion: string | null = null; // Rows still waiting for their deferred pull request links, which must not be saved: // the cache would keep them without links until the next server refresh. let pendingLinkRows: ReadonlySet = new Set(); @@ -122,11 +132,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ) { while (true) { const latest = yield* Ref.get(latestLiveSnapshot); - if (Option.isNone(latest) || latest.value === lastPersisted) return; + const serverVersion = latestLiveServerVersion; + if ( + Option.isNone(latest) || + (latest.value === lastPersisted && serverVersion === lastPersistedServerVersion) + ) { + return; + } if (latest.value.threads.some((thread) => pendingLinkRows.has(thread))) return; const snapshot = latest.value; if ( lastPersisted !== undefined && + serverVersion === lastPersistedServerVersion && snapshot.snapshotSequence >= lastPersisted.snapshotSequence && snapshot.schemaVersion === lastPersisted.schemaVersion && snapshot.projects === lastPersisted.projects && @@ -138,17 +155,20 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") lastPersisted = snapshot; return; } - const saved = yield* cache.saveShell(environmentId, snapshot).pipe( - Effect.as(true), - Effect.catch((error) => - Effect.logWarning("Could not persist environment shell cache.").pipe( - Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error) }), - Effect.as(false), + const saved = yield* cache + .saveShell(environmentId, snapshot, serverVersion ?? undefined) + .pipe( + Effect.as(true), + Effect.catch((error) => + Effect.logWarning("Could not persist environment shell cache.").pipe( + Effect.annotateLogs({ environmentId, ...safeErrorLogAttributes(error) }), + Effect.as(false), + ), ), - ), - ); + ); if (!saved) return; lastPersisted = snapshot; + lastPersistedServerVersion = serverVersion; if (!flush) return; } }, persistenceLock.withPermit); @@ -206,9 +226,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const snapshotWriteLock = yield* Semaphore.make(1); const applyItems = Effect.fn("EnvironmentShellState.applyItems")(function* ( + session: RpcSession, items: ReadonlyArray, pendingLinkIds?: ReadonlySet, ) { + if ((yield* Ref.get(activeSubscriptionSession)) !== session) return; const initial = yield* SubscriptionRef.get(state); let waiting = yield* Ref.get(awaitingCompletion); /* oxlint-disable oxc/no-accumulating-spread -- This copies a fixed-shape state record, not a growing collection. */ @@ -218,7 +240,11 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (item.kind === "synchronized") { waiting = false; if (Option.isSome(next.snapshot)) { - next = { ...next, status: "live", error: Option.none() }; + next = { + ...next, + status: "live", + error: Option.none(), + }; } continue; } @@ -269,16 +295,15 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") next.snapshot.value.threads.filter((thread) => pendingLinkIds.has(thread.id)), ); } + if (receivedSnapshot) { + yield* Ref.set(lastAuthoritativeSession, session); + yield* Ref.set(verifiedServerVersion, yield* Ref.get(activeServerVersion)); + } if (Option.isSome(next.snapshot)) { yield* Ref.set(latestLiveSnapshot, next.snapshot); + latestLiveServerVersion = yield* Ref.get(verifiedServerVersion); } yield* SubscriptionRef.set(state, next); - if (receivedSnapshot) { - const session = yield* Ref.get(activeSubscriptionSession); - if (session !== null) { - yield* Ref.set(lastAuthoritativeSession, session); - } - } if (next.snapshot !== initial.snapshot && Option.isSome(next.snapshot)) { yield* Queue.offer(persistence, next.snapshot.value); } @@ -343,6 +368,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") if (Option.isNone(live)) return; const latest = Option.isSome(filled) ? filled : live; yield* Ref.set(latestLiveSnapshot, latest); + latestLiveServerVersion = yield* Ref.get(verifiedServerVersion); yield* Queue.offer(persistence, latest.value); }, (effect) => snapshotWriteLock.withPermit(effect), @@ -359,28 +385,34 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* setSynchronizing; yield* Effect.forkScoped( - subscribeDynamic( + subscribeDynamicWithSession( ORCHESTRATION_V2_WS_METHODS.subscribeShell, Effect.fn("EnvironmentShellState.makeSubscribeInput")(function* (session) { - yield* Ref.set(activeSubscriptionSession, session); - const { omitIdleSubagents, omitOlderSettled } = yield* session.initialConfig.pipe( - Effect.map((config) => ({ - omitIdleSubagents: config.shellThreadFamilies === true, - omitOlderSettled: - config.shellOlderSettled === true && config.shellThreadFamilies === true, - })), - Effect.orElseSucceed(() => ({ omitIdleSubagents: false, omitOlderSettled: false })), + const config = yield* session.initialConfig.pipe(Effect.option); + const omitIdleSubagents = Option.exists( + config, + (value) => value.shellThreadFamilies === true, + ); + const omitOlderSettled = + omitIdleSubagents && Option.exists(config, (value) => value.shellOlderSettled === true); + const serverVersion = Option.match(config, { + onNone: () => null, + onSome: (value) => value.environment?.serverVersion ?? null, + }); + yield* snapshotWriteLock.withPermit( + Effect.gen(function* () { + yield* Ref.set(activeSubscriptionSession, session); + yield* Ref.set(activeServerVersion, serverVersion); + yield* Ref.set(awaitingCompletion, true); + yield* setSynchronizing; + }), ); - yield* Ref.set(awaitingCompletion, true); - yield* setSynchronizing; - - // Foreground resubscriptions on the same live session can resume from - // the in-memory cursor. A new session reloads the authoritative HTTP - // snapshot so a valid cursor cannot preserve incomplete cached data. - const hasAuthoritativeSnapshot = (yield* Ref.get(lastAuthoritativeSession)) === session; - let canResume = hasAuthoritativeSnapshot; + + // A matching build can still serve restored history after reconnecting. + // Only a snapshot verified on this live session is safe to resume. + let canResume = (yield* Ref.get(lastAuthoritativeSession)) === session; let current = yield* SubscriptionRef.get(state); - if (!hasAuthoritativeSnapshot || Option.isNone(current.snapshot)) { + if (!canResume || Option.isNone(current.snapshot)) { const prepared = yield* SubscriptionRef.get(supervisor.prepared).pipe( Effect.flatMap( Option.match({ @@ -412,7 +444,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") ? [] : httpSnapshot.value.threads.map((thread) => thread.id), ); - yield* applyItems([{ kind: "snapshot", snapshot }], pendingLinkIds); + yield* applyItems(session, [{ kind: "snapshot", snapshot }], pendingLinkIds); canResume = true; current = yield* SubscriptionRef.get(state); if (Option.isSome(current.snapshot)) { @@ -439,7 +471,18 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") retryExpectedFailureAfter: "250 millis", resubscribe: foregroundResubscriptions, }, - ).pipe(Stream.runForEachArray((items) => applyItems(items))), + ).pipe( + Stream.runForEachArray((items) => + Effect.forEach( + Array.groupWith(items, ([left], [right]) => left === right), + (group) => + applyItems( + group[0][0], + group.map(([, item]) => item), + ), + ), + ), + ), ); yield* SubscriptionRef.changes(supervisor.state).pipe( Stream.runForEach((connectionState) => { diff --git a/packages/client-runtime/src/state/shellPullRequests.ts b/packages/client-runtime/src/state/shellPullRequests.ts index 340b1403f2937..e31979a450e46 100644 --- a/packages/client-runtime/src/state/shellPullRequests.ts +++ b/packages/client-runtime/src/state/shellPullRequests.ts @@ -11,6 +11,7 @@ export interface DeferredShellSnapshot extends OrchestrationV2ShellSnapshot { readonly loadPullRequests?: Effect.Effect< ReadonlyMap> >; + readonly verifiedServerVersion?: string; } // A zero sleep is a microtask yield, which on React Native runs before the next frame. diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index e4823f30272bd..a28a5ac9a8405 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1812,6 +1812,7 @@ export const WsSubscribeServerConfigRpc = Rpc.make(WS_METHODS.subscribeServerCon usageLimitsCommand: Schema.optional(Schema.Boolean), /** Omit per-cwd provider snapshots; the client fetches its cwd through `server.refreshProviders`. */ workspaceSnapshotsOnDemand: Schema.optional(Schema.Boolean), + providerCatalogHashes: Schema.optional(Schema.Record(Schema.String, Schema.String)), }), success: ServerConfigStreamEvent, error: Schema.Union([KeybindingsConfigError, ServerSettingsError, EnvironmentAuthorizationError]), diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index 134d8160642fc..ad866e4601ebc 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -258,6 +258,8 @@ export const ServerProvider = Schema.Struct({ Schema.withDecodingDefault(Effect.succeed([])), ), skills: Schema.Array(ServerProviderSkill).pipe(Schema.withDecodingDefault(Effect.succeed([]))), + catalogHash: Schema.optionalKey(Schema.String), + catalogOmitted: Schema.optionalKey(Schema.Boolean), workspaceSnapshots: Schema.optionalKey(Schema.Array(ServerProviderWorkspaceSnapshot)), usageLimits: Schema.optional(ServerProviderUsageLimits), versionAdvisory: Schema.optionalKey(ServerProviderVersionAdvisory),