diff --git a/apps/server/src/environment/DirectEndpoints.test.ts b/apps/server/src/environment/DirectEndpoints.test.ts index 08b16959e8646..7a0fa846a90e5 100644 --- a/apps/server/src/environment/DirectEndpoints.test.ts +++ b/apps/server/src/environment/DirectEndpoints.test.ts @@ -1,18 +1,27 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { it as effectIt } from "@effect/vitest"; +import * as HostProcess from "@t3tools/shared/HostProcess"; +import * as NetService from "@t3tools/shared/Net"; import type * as NodeOS from "node:os"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as HttpClient from "effect/http/HttpClient"; import * as HttpClientResponse from "effect/http/HttpClientResponse"; +import * as ChildProcess from "effect/process/ChildProcess"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import * as TestClock from "effect/testing/TestClock"; import { describe, expect, it } from "vite-plus/test"; import * as ServerConfig from "../config.ts"; import * as DirectEndpoints from "./DirectEndpoints.ts"; import { resolveBoundEndpoints } from "./DirectEndpoints.ts"; +import * as RemoteOpenTargets from "./RemoteOpenTargets.ts"; +import * as TailscaleStatusReader from "./TailscaleStatusReader.ts"; const INTERFACES: ReturnType = { lo0: [ @@ -171,26 +180,31 @@ const TAILSCALE_STATUS_JSON = JSON.stringify({ Self: { DNSName: "bb-1.tail1234.ts.net.", TailscaleIPs: ["100.64.1.2"] }, }); -const layerTailscaleUp = Layer.succeed( - ChildProcessSpawner.ChildProcessSpawner, - ChildProcessSpawner.make(() => - Effect.succeed( - ChildProcessSpawner.makeHandle({ - pid: ChildProcessSpawner.ProcessId(1), - exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(0)), - isRunning: Effect.succeed(false), - kill: () => Effect.void, - unref: Effect.succeed(Effect.void), - stdin: Sink.drain, - stdout: Stream.make(new TextEncoder().encode(TAILSCALE_STATUS_JSON)), - stderr: Stream.empty, - all: Stream.empty, - getInputFd: () => Sink.drain, - getOutputFd: () => Stream.empty, +// `tailscale status` exits non-zero while tailscaled is unavailable, and prints the node once it is up. +const layerTailscale = (isUp: () => boolean) => + Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.sync(() => { + const up = isUp(); + return ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1), + exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(up ? 0 : 1)), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: up ? Stream.make(new TextEncoder().encode(TAILSCALE_STATUS_JSON)) : Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); }), ), - ), -); + ); + +const layerTailscaleUp = layerTailscale(() => true); const layerServeProbe = (status: number) => Layer.succeed( @@ -217,6 +231,7 @@ const resolveWithServe = (probeStatus: number) => Effect.flatMap(DirectEndpoints.DirectEndpoints, (service) => service.resolve()).pipe( Effect.provide( DirectEndpoints.layer.pipe( + Layer.provide(TailscaleStatusReader.layer), Layer.provide( Layer.mergeAll(layerServeConfig, layerTailscaleUp, layerServeProbe(probeStatus)), ), @@ -239,3 +254,103 @@ describe("DirectEndpoints Tailscale Serve", () => { }), ); }); + +// A CLI that cannot reach tailscaled: `tailscale status` never answers, so each read waits out its timeout. +const layerTailscaleHung = (onStatusSpawn: () => void) => + Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make((command) => { + if (ChildProcess.isStandardCommand(command) && command.args[0] === "status") onStatusSpawn(); + return Effect.succeed( + ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1), + exitCode: Effect.never, + isRunning: Effect.succeed(true), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }), + ); + }), + ); + +const layerSshdListening = Layer.succeed(NetService.NetService, { + canListenOnHost: () => Effect.succeed(true), + isPortAvailableOnLoopback: () => Effect.succeed(true), + hasListenerOnHost: () => Effect.succeed(true), + reserveLoopbackPort: () => Effect.succeed(40_000), + findAvailablePort: (preferred) => Effect.succeed(preferred), +}); + +const MDNS_ONLY = [{ kind: "mdns", host: "bb-1.local", username: "t3-user" }]; + +// What ws.loadServerConfig resolves for every client that connects. The layers live in the +// test's scope, as the server's do, so the shared status read outlives each connect. +const connectWithTailscale = (layerSpawner: Layer.Layer) => + Effect.gen(function* () { + const context = yield* Layer.build( + Layer.mergeAll(DirectEndpoints.layer, RemoteOpenTargets.layer).pipe( + Layer.provide(TailscaleStatusReader.layer), + Layer.provide( + Layer.mergeAll(layerServeConfig, layerSpawner, layerServeProbe(200), layerSshdListening), + ), + ), + ); + const remoteOpenTargets = Context.get(context, RemoteOpenTargets.RemoteOpenTargets); + const directEndpoints = Context.get(context, DirectEndpoints.DirectEndpoints); + return Effect.all([remoteOpenTargets.resolveTargets(), directEndpoints.resolve()], { + concurrency: "unbounded", + }).pipe( + Effect.provideService(HostProcess.Hostname, "bb-1"), + Effect.provideService(HostProcess.Username, "t3-user"), + ); + }); + +describe("Tailscale status shared by DirectEndpoints and RemoteOpenTargets", () => { + effectIt.effect("reads `tailscale status` once per connect, not once per service", () => + Effect.gen(function* () { + let statusReads = 0; + const connect = yield* connectWithTailscale(layerTailscaleHung(() => statusReads++)); + const first = yield* Effect.forkChild(connect); + yield* TestClock.adjust("4 seconds"); + expect(yield* Fiber.join(first)).toEqual([MDNS_ONLY, []]); + expect(statusReads, "tailscale status spawns for one connect").toBe(1); + }), + ); + + effectIt.effect( + "answers a later connect from the settled failed read without waiting again", + () => + Effect.gen(function* () { + const connect = yield* connectWithTailscale(layerTailscaleHung(() => {})); + const first = yield* Effect.forkChild(connect); + yield* TestClock.adjust("4 seconds"); + expect(yield* Fiber.join(first)).toEqual([MDNS_ONLY, []]); + + const later = yield* connect.pipe(Effect.timeoutOption("1 second"), Effect.forkChild); + yield* TestClock.adjust("1 second"); + expect(yield* Fiber.join(later), "later connect still waiting on tailscale status").toEqual( + Option.some([MDNS_ONLY, []]), + ); + }), + ); + + effectIt.effect("answers a connect from a recovered read that settles within the grace", () => + Effect.gen(function* () { + let tailscaleUp = false; + const connect = yield* connectWithTailscale(layerTailscale(() => tailscaleUp)); + expect(yield* connect, "startup read fails").toEqual([MDNS_ONLY, []]); + + tailscaleUp = true; + expect(yield* connect).toEqual([ + [{ kind: "tailscale", host: "bb-1.tail1234.ts.net", username: "t3-user" }, ...MDNS_ONLY], + [{ kind: "tailnet", httpBaseUrl: "https://bb-1.tail1234.ts.net/" }], + ]); + }), + ); +}); diff --git a/apps/server/src/environment/DirectEndpoints.ts b/apps/server/src/environment/DirectEndpoints.ts index 99e33b98f9709..b0e0deba5d4a5 100644 --- a/apps/server/src/environment/DirectEndpoints.ts +++ b/apps/server/src/environment/DirectEndpoints.ts @@ -1,16 +1,11 @@ import type { ServerDirectEndpoint } from "@t3tools/contracts"; -import { - buildTailscaleHttpsBaseUrl, - probeTailscaleHttpsEndpoint, - readTailscaleStatus, -} from "@t3tools/tailscale"; +import { buildTailscaleHttpsBaseUrl, probeTailscaleHttpsEndpoint } 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 Option from "effect/Option"; import { HttpClient } from "effect/http"; -import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import * as NodeNet from "node:net"; import * as NodeOS from "node:os"; @@ -24,6 +19,7 @@ import { isWildcardHost, VIRTUAL_INTERFACE, } from "../startupAccess.ts"; +import * as TailscaleStatusReader from "./TailscaleStatusReader.ts"; type NetworkInterfacesMap = ReturnType; @@ -109,7 +105,7 @@ const probeCloudMetadata = Effect.callback((resume) => { /** @public */ export const make = Effect.gen(function* () { const config = yield* ServerConfig.ServerConfig; - const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + const tailscaleStatus = yield* TailscaleStatusReader.TailscaleStatusReader; const httpClient = yield* HttpClient.HttpClient; // Only the service default listens on every interface without being asked, // so only it checks once whether it runs in a cloud. The check starts now so @@ -130,7 +126,7 @@ export const make = Effect.gen(function* () { ]; // Serve setup can fail without stopping startup; advertise only after it answers as this server. if (config.tailscaleServeEnabled) { - const servedUrl = yield* readTailscaleStatus.pipe( + const servedUrl = yield* tailscaleStatus.read().pipe( Effect.map((status) => status.magicDnsName), Effect.flatMap((magicDnsName) => { if (magicDnsName === null) return Effect.succeed(null); @@ -145,7 +141,6 @@ export const make = Effect.gen(function* () { 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)) { diff --git a/apps/server/src/environment/RemoteOpenTargets.test.ts b/apps/server/src/environment/RemoteOpenTargets.test.ts index 3e66d5f707a0d..c2e103b8f83ef 100644 --- a/apps/server/src/environment/RemoteOpenTargets.test.ts +++ b/apps/server/src/environment/RemoteOpenTargets.test.ts @@ -12,6 +12,7 @@ import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { afterEach, describe, expect, vi } from "vite-plus/test"; import * as RemoteOpenTargets from "./RemoteOpenTargets.ts"; +import * as TailscaleStatusReader from "./TailscaleStatusReader.ts"; afterEach(() => { vi.restoreAllMocks(); @@ -74,6 +75,7 @@ const resolveTargets = (input: { Effect.provideService(HostProcess.Hostname, input.hostname), Effect.provide( RemoteOpenTargets.layer.pipe( + Layer.provide(TailscaleStatusReader.layer), Layer.provide(Layer.mergeAll(layerNet(input.sshd), layerSpawner(input.tailscale))), ), ), diff --git a/apps/server/src/environment/RemoteOpenTargets.ts b/apps/server/src/environment/RemoteOpenTargets.ts index 8b4997f695d15..212cb4f4d2528 100644 --- a/apps/server/src/environment/RemoteOpenTargets.ts +++ b/apps/server/src/environment/RemoteOpenTargets.ts @@ -1,12 +1,14 @@ import { type RemoteOpenTarget } from "@t3tools/contracts"; import * as HostProcess from "@t3tools/shared/HostProcess"; import * as NetService from "@t3tools/shared/Net"; -import { readTailscaleSshEnabled, readTailscaleStatus } from "@t3tools/tailscale"; +import { readTailscaleSshEnabled } from "@t3tools/tailscale"; import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; +import * as TailscaleStatusReader from "./TailscaleStatusReader.ts"; + const SSH_PORT = 22; export class RemoteOpenTargets extends Context.Service< @@ -20,6 +22,7 @@ export class RemoteOpenTargets extends Context.Service< export const make = Effect.gen(function* () { const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; const net = yield* NetService.NetService; + const tailscaleStatus = yield* TailscaleStatusReader.TailscaleStatusReader; const resolveTargets = Effect.gen(function* () { // Check both loopback families: sshd can be bound IPv6-only. @@ -30,10 +33,9 @@ export const make = Effect.gen(function* () { ); const username = yield* HostProcess.Username; - const magicDnsName = yield* readTailscaleStatus.pipe( + const magicDnsName = yield* tailscaleStatus.read().pipe( Effect.map((status) => status.magicDnsName), Effect.orElseSucceed(() => null), - Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), ); const targets: Array = []; diff --git a/apps/server/src/environment/TailscaleStatusReader.ts b/apps/server/src/environment/TailscaleStatusReader.ts new file mode 100644 index 0000000000000..6801d979dd4dc --- /dev/null +++ b/apps/server/src/environment/TailscaleStatusReader.ts @@ -0,0 +1,95 @@ +import { + readTailscaleStatus, + type TailscaleCommandError, + type TailscaleStatus, + type TailscaleStatusParseError, +} from "@t3tools/tailscale"; +import * as Context from "effect/Context"; +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 Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import type * as Scope from "effect/Scope"; +import * as Stream from "effect/Stream"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; + +type TailscaleStatusError = TailscaleCommandError | TailscaleStatusParseError; + +export class TailscaleStatusReader extends Context.Service< + TailscaleStatusReader, + { + readonly read: () => Effect.Effect; + /** Emits when a read settles to a different outcome than the previous settled read. */ + readonly subscribeChanges: Effect.Effect, never, Scope.Scope>; + } +>()("t3/environment/TailscaleStatusReader") {} + +/** Answer from the last settled read without refreshing, for re-resolves that a settled change drives. */ +export const SettledReadOnly = Context.Reference( + "t3/environment/TailscaleStatusReader/SettledReadOnly", + { defaultValue: () => false }, +); + +// A healthy CLI answers in tens of milliseconds, so a connect still gets a fresh read from it. +const FRESH_READ_GRACE = Duration.millis(250); + +// Callers treat every failure alike, so failures compare equal regardless of cause. +const outcomeKey = (exit: Exit.Exit) => + Exit.isSuccess(exit) ? JSON.stringify(exit.value) : null; + +const make = Effect.gen(function* () { + const spawner = yield* ChildProcessSpawner.ChildProcessSpawner; + // The layer's scope, so a caller that times out or disconnects never interrupts the shared read. + const scope = yield* Effect.scope; + const changes = yield* PubSub.unbounded(); + let settled: Exit.Exit | undefined; + let inFlight: Deferred.Deferred | undefined; + + const refresh = Effect.suspend(() => { + if (inFlight !== undefined) return Effect.succeed(inFlight); + const deferred = Deferred.makeUnsafe(); + inFlight = deferred; + return readTailscaleStatus.pipe( + Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), + Effect.onExit((exit) => + Effect.suspend(() => { + const changed = settled === undefined || outcomeKey(settled) !== outcomeKey(exit); + settled = exit; + inFlight = undefined; + return Deferred.done(deferred, exit).pipe( + Effect.andThen(changed ? PubSub.publish(changes, undefined) : Effect.void), + ); + }), + ), + Effect.interruptible, + Effect.forkIn(scope), + Effect.as(deferred), + ); + }).pipe( + // Claiming `inFlight` and forking are one step: an interrupt between them would strand the claim. + Effect.uninterruptible, + ); + yield* refresh; + + return TailscaleStatusReader.of({ + // A caller gets the refresh it starts or joins if that settles within the grace, else the + // previous read's result; a later outcome reaches open subscribers through `subscribeChanges`. + read: Effect.fn("TailscaleStatusReader.read")(function* () { + const previous = settled; + if (previous !== undefined && (yield* SettledReadOnly)) return yield* previous; + const pending = yield* refresh; + if (previous === undefined) return yield* Deferred.await(pending); + const fresh = yield* Deferred.await(pending).pipe( + Effect.exit, + Effect.timeoutOption(FRESH_READ_GRACE), + ); + return yield* Option.getOrElse(fresh, () => previous); + }), + subscribeChanges: PubSub.subscribe(changes).pipe(Effect.map(Stream.fromSubscription)), + }); +}); + +export const layer = Layer.effect(TailscaleStatusReader, make); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 741fc73103b7d..11e2f43fb9309 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -129,6 +129,7 @@ import * as EventLoopMonitor from "./observability/EventLoopMonitor.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as DirectEndpoints from "./environment/DirectEndpoints.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; +import * as TailscaleStatusReader from "./environment/TailscaleStatusReader.ts"; import * as AuthHttp from "./auth/http.ts"; import * as ReplayMarkers from "./auth/replayMarkers.ts"; import * as ServerSecretStore from "./auth/ServerSecretStore.ts"; @@ -701,8 +702,11 @@ const layerRuntimeDependencies = layerRuntimeCoreDependencies.pipe( Layer.provideMerge(TriageService.layer.pipe(Layer.provide(GitVcsDriver.layer))), Layer.provideMerge(AnalyticsService.layer), Layer.provideMerge(ExternalLauncher.layer), - Layer.provideMerge(RemoteOpenTargets.layer), - Layer.provideMerge(DirectEndpoints.layer), + Layer.provideMerge( + Layer.mergeAll(RemoteOpenTargets.layer, DirectEndpoints.layer).pipe( + Layer.provideMerge(TailscaleStatusReader.layer), + ), + ), Layer.provideMerge(ServerLifecycleEvents.layer), Layer.provide(NetService.layer), ); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ac60becc2d78a..2c6728ccf676c 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -17,15 +17,20 @@ import { type ServerConfigStreamEvent, type ServerProvider, } from "@t3tools/contracts"; +import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; import * as Queue from "effect/Queue"; +import * as Sink from "effect/Sink"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; +import * as NetService from "@t3tools/shared/Net"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; import { RpcGroup, RpcMessage, RpcServer } from "effect/rpc"; import * as SqlClient from "effect/sql/SqlClient"; @@ -73,6 +78,7 @@ import * as PullRequestService from "./pullRequest/PullRequestService.ts"; import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts"; import * as RelayClient from "@t3tools/shared/relayClient"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; +import * as TailscaleStatusReader from "./environment/TailscaleStatusReader.ts"; import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as ReviewService from "./review/ReviewService.ts"; @@ -123,7 +129,7 @@ import { subscribeOrchestrationV2Shell, subscribeOrchestrationV2Thread, WS_RPC_SERVER_OPTIONS, - withLateEditorConfig, + withLateConfig, } from "./ws.ts"; const snapshotConfig = (fields: Partial) => @@ -138,7 +144,7 @@ const settingsUpdated = (settings: object): ServerConfigStreamEvent => ({ it.effect("resends late editors without rolling back updates already sent", () => Effect.gen(function* () { const settingsSent = yield* Deferred.make(); - const events = yield* withLateEditorConfig( + const events = yield* withLateConfig( snapshotConfig({ settings: { enableProviderUpdateChecks: true } as never }), Stream.make(settingsUpdated({ enableProviderUpdateChecks: false })), { @@ -146,6 +152,7 @@ it.effect("resends late editors without rolling back updates already sent", () = resolveFileManagerRevealKind: () => Deferred.await(settingsSent).pipe(Effect.as("file-explorer" as const)), }, + Stream.empty, ).pipe( Stream.tap((event) => event.type === "settingsUpdated" ? Deferred.succeed(settingsSent, undefined) : Effect.void, @@ -167,13 +174,14 @@ it.effect("resends late editors without rolling back updates already sent", () = it.effect("resends a file manager reveal kind that missed the snapshot", () => Effect.gen(function* () { - const events = yield* withLateEditorConfig( + const events = yield* withLateConfig( snapshotConfig({ availableEditors: ["file-manager"] }), Stream.empty, { resolveAvailableEditors: () => Effect.succeed(["file-manager"]), resolveFileManagerRevealKind: () => Effect.succeed("file-explorer"), }, + Stream.empty, ).pipe(Stream.runCollect); const [late] = Array.from(events); @@ -589,6 +597,9 @@ it.effect( resolveTargets: () => Effect.succeed([]), }), Layer.mock(DirectEndpoints.DirectEndpoints)({ resolve: () => Effect.succeed([]) }), + Layer.mock(TailscaleStatusReader.TailscaleStatusReader)({ + subscribeChanges: Effect.succeed(Stream.empty), + }), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ scratchRoot: Effect.succeedNone, namedProjectsRoot: process.cwd(), @@ -765,6 +776,9 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( resolveTargets: () => Effect.succeed([]), }), Layer.mock(DirectEndpoints.DirectEndpoints)({ resolve: () => Effect.succeed([]) }), + Layer.mock(TailscaleStatusReader.TailscaleStatusReader)({ + subscribeChanges: Effect.succeed(Stream.empty), + }), Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ scratchRoot: Effect.succeedNone, namedProjectsRoot: process.cwd(), @@ -869,6 +883,213 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( ), ); +// `tailscale status` exits non-zero while tailscaled is unavailable, and prints the node once it is up. +// Each spawn asks `nextRead` for its outcome and for the effect the process exits after. +const layerTailscale = ( + nextRead: () => { readonly up: boolean; readonly exits: Effect.Effect }, +) => + Layer.succeed( + ChildProcessSpawner.ChildProcessSpawner, + ChildProcessSpawner.make(() => + Effect.sync(() => { + const { up, exits } = nextRead(); + return ChildProcessSpawner.makeHandle({ + pid: ChildProcessSpawner.ProcessId(1), + exitCode: exits.pipe(Effect.as(ChildProcessSpawner.ExitCode(up ? 0 : 1))), + isRunning: Effect.succeed(false), + kill: () => Effect.void, + unref: Effect.succeed(Effect.void), + stdin: Sink.drain, + stdout: up + ? Stream.make( + new TextEncoder().encode( + JSON.stringify({ + Self: { DNSName: "bb-1.tail1234.ts.net.", TailscaleIPs: ["100.64.1.2"] }, + }), + ), + ) + : Stream.empty, + stderr: Stream.empty, + all: Stream.empty, + getInputFd: () => Sink.drain, + getOutputFd: () => Stream.empty, + }); + }), + ), + ); + +const targetKinds = (event: ServerConfigStreamEvent) => + event.type === "snapshot" ? event.config.remoteOpenTargets?.map((target) => target.kind) : []; + +// Builds the shared Tailscale services in the caller's scope, as the server does, then +// waits for the startup read so a test controls which reads its connect sees. +const subscribeServerConfigWithTailscale = Effect.fn(function* ( + layerSpawner: Layer.Layer, + beforeConnect: Effect.Effect, +) { + const tailscaleServices = yield* Layer.build( + RemoteOpenTargets.layer.pipe( + Layer.provideMerge(TailscaleStatusReader.layer), + Layer.provide( + Layer.mergeAll( + layerSpawner, + Layer.mock(NetService.NetService)({ hasListenerOnHost: () => Effect.succeed(true) }), + ), + ), + ), + ); + const reader = Context.get(tailscaleServices, TailscaleStatusReader.TailscaleStatusReader); + assert.isTrue(Exit.isFailure(yield* Effect.exit(reader.read())), "startup read fails"); + yield* beforeConnect; + + const responses = yield* Queue.unbounded(); + const configRpcServices = Layer.mergeAll( + layerUnusedConfigRpcServices, + Layer.succeedContext(tailscaleServices), + Layer.mock(ExternalLauncher.ExternalLauncher)({ + resolveAvailableEditors: () => Effect.succeed([]), + // @effect-diagnostics-next-line effectSucceedWithVoid:off - This optional result requires undefined, not the wider void type. + resolveFileManagerRevealKind: () => Effect.succeed(undefined), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([]), + streamChanges: Stream.never, + }), + Layer.mock(UsageLimitSources.UsageLimitSources)({ streamChanges: Stream.make([]) }), + Layer.mock(DirectEndpoints.DirectEndpoints)({ resolve: () => Effect.succeed([]) }), + Layer.mock(ManagedProjectFolders.ManagedProjectFolders)({ + scratchRoot: Effect.succeedNone, + namedProjectsRoot: process.cwd(), + }), + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getDescriptor: Effect.succeed({ + environmentId: EnvironmentId.make("config-test"), + label: "Config test", + platform: { os: "linux", arch: "x64" }, + serverVersion: "0.0.0", + capabilities: { repositoryIdentity: false }, + }), + }), + Layer.mock(EnvironmentAuth.EnvironmentAuth)({ + getDescriptor: () => + Effect.succeed({ + policy: "loopback-browser", + bootstrapMethods: [], + sessionMethods: ["browser-session-cookie"], + sessionCookieName: "config-test-session", + }), + }), + ); + const server = yield* RpcServer.makeNoSerialization(RpcGroup.make(WsSubscribeServerConfigRpc), { + ...WS_RPC_SERVER_OPTIONS, + onFromServer: (response) => + response._tag === "Chunk" + ? Effect.forEach(response.values, (event) => Queue.offer(responses, event), { + discard: true, + }) + : Effect.void, + }).pipe( + Effect.provide( + layerWsRpc( + { + sessionId: AuthSessionId.make("config-test"), + subject: "config-test", + method: "browser-session-cookie", + scopes: [], + }, + {}, + yield* ServerBrowser.ServerBrowser, + ).pipe(Layer.provide(configRpcServices)), + ), + ); + const requestId = RpcMessage.RequestId("config"); + yield* server.write(0, { + _tag: "Request", + id: requestId, + tag: WS_METHODS.subscribeServerConfig, + payload: {}, + headers: Headers.empty, + }); + return { + responses, + ack: server.write(0, { _tag: "Ack", requestId }), + interrupt: server.write(0, { _tag: "Interrupt", requestId, interruptors: [] }), + }; +}); + +const provideConfigSources = (effect: Effect.Effect) => + effect.pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ); + +it.effect("sends the recovered Tailscale target to a subscription opened on a stale failure", () => + Effect.gen(function* () { + let tailscaleUp = false; + const recoveredReadSpawned = yield* Deferred.make(); + const recoveredReadAnswers = yield* Deferred.make(); + const subscription = yield* subscribeServerConfigWithTailscale( + layerTailscale(() => { + if (!tailscaleUp) return { up: false, exits: Effect.void }; + Deferred.doneUnsafe(recoveredReadSpawned, Exit.void); + return { up: true, exits: Deferred.await(recoveredReadAnswers) }; + }), + Effect.sync(() => { + tailscaleUp = true; + }), + ); + + // The recovered read outlasts the connect's grace, so it must arrive as a follow-up. + yield* Deferred.await(recoveredReadSpawned); + yield* TestClock.adjust("1 second"); + const first = yield* Queue.take(subscription.responses); + assert.equal(first.type, "snapshot"); + assert.deepEqual(targetKinds(first), ["mdns"], "first snapshot answers from the stale failure"); + yield* subscription.ack; + yield* Deferred.succeed(recoveredReadAnswers, undefined); + const followUp = yield* Queue.take(subscription.responses); + assert.equal(followUp.type, "snapshot"); + assert.deepEqual(targetKinds(followUp), ["tailscale", "mdns"]); + yield* subscription.interrupt; + }).pipe(provideConfigSources), +); + +it.effect("re-resolves a subscription after a Tailscale change without reading again", () => + Effect.gen(function* () { + let statusReads = 0; + const secondReadSpawned = yield* Deferred.make(); + const secondReadAnswers = yield* Deferred.make(); + // A flapping tailscaled: every read answers the opposite of the last one. + const subscription = yield* subscribeServerConfigWithTailscale( + layerTailscale(() => { + statusReads++; + if (statusReads % 2 === 1) return { up: false, exits: Effect.void }; + if (statusReads === 2) { + Deferred.doneUnsafe(secondReadSpawned, Exit.void); + return { up: true, exits: Deferred.await(secondReadAnswers) }; + } + return { up: true, exits: Effect.void }; + }), + Effect.void, + ); + + yield* Deferred.await(secondReadSpawned); + yield* TestClock.adjust("1 second"); + assert.deepEqual(targetKinds(yield* Queue.take(subscription.responses)), ["mdns"]); + yield* subscription.ack; + yield* Deferred.succeed(secondReadAnswers, undefined); + assert.deepEqual(targetKinds(yield* Queue.take(subscription.responses)), ["tailscale", "mdns"]); + assert.equal(statusReads, 2, "only startup and the connect read tailscale status"); + yield* subscription.interrupt; + }).pipe(provideConfigSources), +); + // Compaction keeps the oldest and newest event per message, so a cursor older // than the deleted rows still replays each message where the client would // first have seen it (reducers append on first sight) and reaches the same diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 34eb11815b13f..bdfc23215a553 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -221,6 +221,7 @@ import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts"; import * as ServerEnvironment from "./environment/ServerEnvironment.ts"; import * as DirectEndpoints from "./environment/DirectEndpoints.ts"; import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts"; +import * as TailscaleStatusReader from "./environment/TailscaleStatusReader.ts"; import * as DefectReporter from "./observability/DefectReporter.ts"; import * as EventLoopMonitor from "./observability/EventLoopMonitor.ts"; import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; @@ -310,10 +311,17 @@ const resolveEditorConfig = Effect.fn("ws.resolveEditorConfig")(function* }; }); -export const withLateEditorConfig = ( +type NetworkTargets = Pick; + +const sameNetworkTargets = (left: NetworkTargets, right: NetworkTargets) => + JSON.stringify([left.remoteOpenTargets, left.directEndpoints]) === + JSON.stringify([right.remoteOpenTargets, right.directEndpoints]); + +export const withLateConfig = ( config: ClientServerConfig, liveUpdates: Stream.Stream, launcher: EditorDiscovery, + lateNetworkTargets: Stream.Stream, ) => { const lateEditorConfig = Stream.fromEffect(launcher.resolveAvailableEditors()).pipe( Stream.filter( @@ -333,7 +341,15 @@ export const withLateEditorConfig = ( Stream.map((editorConfig) => ({ type: "editorsResolved" as const, editorConfig })), ); - return Stream.merge(liveUpdates, lateEditorConfig).pipe( + return Stream.merge( + liveUpdates, + Stream.merge( + lateEditorConfig, + lateNetworkTargets.pipe( + Stream.map((targets) => ({ type: "networkTargetsResolved" as const, targets })), + ), + ), + ).pipe( Stream.mapAccum( (): ClientServerConfig => config, (current, event): readonly [ClientServerConfig, ReadonlyArray] => { @@ -348,6 +364,11 @@ export const withLateEditorConfig = ( const next = { ...rest, ...event.editorConfig }; return [next, [{ version: 1, type: "snapshot", config: next }]]; } + case "networkTargetsResolved": { + if (sameNetworkTargets(current, event.targets)) return [current, []]; + const next = { ...current, ...event.targets }; + return [next, [{ version: 1, type: "snapshot", config: next }]]; + } case "keybindingsUpdated": return [{ ...current, ...event.payload }, [event]]; case "providerStatuses": @@ -1309,6 +1330,7 @@ export const layerWsRpc = ( const externalLauncher = yield* ExternalLauncher.ExternalLauncher; const remoteOpenTargets = yield* RemoteOpenTargets.RemoteOpenTargets; const directEndpoints = yield* DirectEndpoints.DirectEndpoints; + const tailscaleStatus = yield* TailscaleStatusReader.TailscaleStatusReader; const gitWorkflow = yield* GitWorkflowService.GitWorkflowService; const review = yield* ReviewService.ReviewService; const vcsProvisioning = yield* VcsProvisioningService.VcsProvisioningService; @@ -1716,6 +1738,17 @@ export const layerWsRpc = ( ), ); + // Concurrent, so both services share one Tailscale read and its grace window. + const resolveNetworkTargets = Effect.fn("ws.resolveNetworkTargets")(function* () { + return yield* Effect.all( + { + remoteOpenTargets: resolveAvailableEditorsForConfig(remoteOpenTargets.resolveTargets()), + directEndpoints: resolveAvailableEditorsForConfig(directEndpoints.resolve()), + }, + { concurrency: "unbounded" }, + ); + }); + const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { readonly usageLimitsCommand: boolean; readonly workspaceSnapshotsOnDemand: boolean; @@ -1749,10 +1782,7 @@ export const layerWsRpc = ( issues: keybindingsConfig.issues, providers, ...editorConfig, - remoteOpenTargets: yield* resolveAvailableEditorsForConfig( - remoteOpenTargets.resolveTargets(), - ), - directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()), + ...(yield* resolveNetworkTargets()), observability: { logsDirectoryPath: config.logsDir, localTracingEnabled: true, @@ -3076,6 +3106,8 @@ export const layerWsRpc = ( // Buffer changes through discovery and the initial snapshot's RPC ACK wait. const keybindingsChanges = yield* keybindings.subscribeChanges; const settingsChanges = yield* serverSettings.subscribeChanges; + // A Tailscale refresh that outlasts the snapshot's grace arrives as a follow-up. + const tailscaleChanges = yield* tailscaleStatus.subscribeChanges; const config = yield* loadServerConfig({ usageLimitsCommand, workspaceSnapshotsOnDemand, @@ -3167,7 +3199,20 @@ export const layerWsRpc = ( return Stream.concat( rpcInitialItems([{ version: 1 as const, type: "snapshot" as const, config }]), - withLateEditorConfig(config, liveUpdates, externalLauncher), + withLateConfig( + config, + liveUpdates, + externalLauncher, + // Re-resolving from the settled change must not start a refresh, or a flapping + // CLI would chain refreshes through every open subscription. + tailscaleChanges.pipe( + Stream.mapEffect(() => + resolveNetworkTargets().pipe( + Effect.provideService(TailscaleStatusReader.SettledReadOnly, true), + ), + ), + ), + ), ); }), ),