From 271fb8cf81e8189cde25e7f42e1fe5ab6322c617 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 02:14:28 +0000 Subject: [PATCH 1/5] test(server): reproduce #26775, connects re-read a failing tailscale status per service Fixes nothing yet: these tests fail on main. Both RemoteOpenTargets and DirectEndpoints spawn their own `tailscale status` per connect, and a later connect waits on a fresh read after a failed one has already settled. Co-Authored-By: Claude Opus 5.5 --- .../src/environment/DirectEndpoints.test.ts | 97 +++++++++++++++++++ 1 file changed, 97 insertions(+) diff --git a/apps/server/src/environment/DirectEndpoints.test.ts b/apps/server/src/environment/DirectEndpoints.test.ts index 08b16959e8646..1fd3dc2b6368e 100644 --- a/apps/server/src/environment/DirectEndpoints.test.ts +++ b/apps/server/src/environment/DirectEndpoints.test.ts @@ -1,18 +1,25 @@ 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 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"; const INTERFACES: ReturnType = { lo0: [ @@ -239,3 +246,93 @@ 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. +const connectWithTailscaleHung = (onStatusSpawn: () => void) => + Effect.gen(function* () { + const remoteOpenTargets = yield* RemoteOpenTargets.RemoteOpenTargets; + const directEndpoints = yield* DirectEndpoints.DirectEndpoints; + return Effect.all([remoteOpenTargets.resolveTargets(), directEndpoints.resolve()], { + concurrency: "unbounded", + }).pipe( + Effect.provideService(HostProcess.Hostname, "bb-1"), + Effect.provideService(HostProcess.Username, "t3-user"), + ); + }).pipe( + Effect.provide( + Layer.mergeAll(DirectEndpoints.layer, RemoteOpenTargets.layer).pipe( + Layer.provide( + Layer.mergeAll( + layerServeConfig, + layerTailscaleHung(onStatusSpawn), + layerServeProbe(200), + layerSshdListening, + ), + ), + ), + ), + ); + +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* connectWithTailscaleHung(() => 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* connectWithTailscaleHung(() => {}); + 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, []]), + ); + }), + ); +}); From bceee1ab94ac31e9ed9b0e436d0942b2b2971bbc Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 02:25:23 +0000 Subject: [PATCH 2/5] fix(server): connects share one tailscale status read instead of waiting on two Every connect ran `tailscale status --json` twice, once for RemoteOpenTargets and once for DirectEndpoints, and kept nothing between connects. When the CLI times out, each connect and reconnect waited ~6.5 s for its first config. A server-wide TailscaleStatusReader starts the read when the server builds, answers later callers from the last settled result (success or failure), and starts at most one background refresh at a time. The read runs in the layer's scope, so a caller timing out never interrupts it. Fixes pingdotgg/t3code#26775 Co-Authored-By: Claude Opus 5.5 --- .../src/environment/DirectEndpoints.test.ts | 30 +++++----- .../server/src/environment/DirectEndpoints.ts | 13 ++--- .../src/environment/RemoteOpenTargets.test.ts | 2 + .../src/environment/RemoteOpenTargets.ts | 8 ++- .../src/environment/TailscaleStatusReader.ts | 58 +++++++++++++++++++ apps/server/src/server.ts | 8 ++- 6 files changed, 92 insertions(+), 27 deletions(-) create mode 100644 apps/server/src/environment/TailscaleStatusReader.ts diff --git a/apps/server/src/environment/DirectEndpoints.test.ts b/apps/server/src/environment/DirectEndpoints.test.ts index 1fd3dc2b6368e..6a83e9350cb2c 100644 --- a/apps/server/src/environment/DirectEndpoints.test.ts +++ b/apps/server/src/environment/DirectEndpoints.test.ts @@ -3,6 +3,7 @@ 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"; @@ -20,6 +21,7 @@ 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: [ @@ -224,6 +226,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)), ), @@ -281,20 +284,13 @@ const layerSshdListening = Layer.succeed(NetService.NetService, { const MDNS_ONLY = [{ kind: "mdns", host: "bb-1.local", username: "t3-user" }]; -// What ws.loadServerConfig resolves for every client that connects. +// 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 connectWithTailscaleHung = (onStatusSpawn: () => void) => Effect.gen(function* () { - const remoteOpenTargets = yield* RemoteOpenTargets.RemoteOpenTargets; - const directEndpoints = yield* DirectEndpoints.DirectEndpoints; - return Effect.all([remoteOpenTargets.resolveTargets(), directEndpoints.resolve()], { - concurrency: "unbounded", - }).pipe( - Effect.provideService(HostProcess.Hostname, "bb-1"), - Effect.provideService(HostProcess.Username, "t3-user"), - ); - }).pipe( - Effect.provide( + const context = yield* Layer.build( Layer.mergeAll(DirectEndpoints.layer, RemoteOpenTargets.layer).pipe( + Layer.provide(TailscaleStatusReader.layer), Layer.provide( Layer.mergeAll( layerServeConfig, @@ -304,8 +300,16 @@ const connectWithTailscaleHung = (onStatusSpawn: () => void) => ), ), ), - ), - ); + ); + 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", () => 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..cb091e19d681e --- /dev/null +++ b/apps/server/src/environment/TailscaleStatusReader.ts @@ -0,0 +1,58 @@ +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 Effect from "effect/Effect"; +import type * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as ChildProcessSpawner from "effect/process/ChildProcessSpawner"; + +type TailscaleStatusError = TailscaleCommandError | TailscaleStatusParseError; + +export class TailscaleStatusReader extends Context.Service< + TailscaleStatusReader, + { + readonly read: () => Effect.Effect; + } +>()("t3/environment/TailscaleStatusReader") {} + +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; + 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.sync(() => { + settled = exit; + inFlight = undefined; + }).pipe(Effect.andThen(Deferred.done(deferred, exit))), + ), + Effect.forkIn(scope), + Effect.as(deferred), + ); + }); + yield* refresh; + + return TailscaleStatusReader.of({ + // A caller gets the previous read's result; the refresh it starts serves the next caller. + read: Effect.fn("TailscaleStatusReader.read")(function* () { + const previous = settled; + const pending = yield* refresh; + return yield* previous ?? Deferred.await(pending); + }), + }); +}); + +export const layer = Layer.effect(TailscaleStatusReader, make); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 741fc73103b7d..0bb15f5e65893 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.provide(TailscaleStatusReader.layer), + ), + ), Layer.provideMerge(ServerLifecycleEvents.layer), Layer.provide(NetService.layer), ); From 447455dc0e32a5f775939042b56e1bc0766afd21 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 02:44:20 +0000 Subject: [PATCH 3/5] fix(server): open config subscriptions get Tailscale targets that recover after connect A client that connected while the shared `tailscale status` read held a stale failure kept an mDNS-only target and no tailnet route for its whole session, because the refresh its connect started never reached its subscription. TailscaleStatusReader now publishes when a read settles to a different outcome. subscribeServerConfig subscribes to it before discovery, re-resolves remote open targets and direct endpoints on each change, and folds them into the existing late-config accumulator, which sends a snapshot only when they differ. Co-Authored-By: Claude Opus 5.5 --- .../src/environment/TailscaleStatusReader.ts | 24 ++- apps/server/src/server.ts | 2 +- apps/server/src/ws.test.ts | 161 +++++++++++++++++- apps/server/src/ws.ts | 49 +++++- 4 files changed, 221 insertions(+), 15 deletions(-) diff --git a/apps/server/src/environment/TailscaleStatusReader.ts b/apps/server/src/environment/TailscaleStatusReader.ts index cb091e19d681e..5ee98f21b4da8 100644 --- a/apps/server/src/environment/TailscaleStatusReader.ts +++ b/apps/server/src/environment/TailscaleStatusReader.ts @@ -7,8 +7,11 @@ import { import * as Context from "effect/Context"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; -import type * as Exit from "effect/Exit"; +import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; +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; @@ -17,13 +20,20 @@ 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") {} +// 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; @@ -34,10 +44,14 @@ const make = Effect.gen(function* () { return readTailscaleStatus.pipe( Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner), Effect.onExit((exit) => - Effect.sync(() => { + Effect.suspend(() => { + const changed = settled === undefined || outcomeKey(settled) !== outcomeKey(exit); settled = exit; inFlight = undefined; - }).pipe(Effect.andThen(Deferred.done(deferred, exit))), + return Deferred.done(deferred, exit).pipe( + Effect.andThen(changed ? PubSub.publish(changes, undefined) : Effect.void), + ); + }), ), Effect.forkIn(scope), Effect.as(deferred), @@ -46,12 +60,14 @@ const make = Effect.gen(function* () { yield* refresh; return TailscaleStatusReader.of({ - // A caller gets the previous read's result; the refresh it starts serves the next caller. + // A caller gets the previous read's result; the refresh it starts serves the next caller + // and, when its outcome differs, reaches open subscribers through `subscribeChanges`. read: Effect.fn("TailscaleStatusReader.read")(function* () { const previous = settled; const pending = yield* refresh; return yield* previous ?? Deferred.await(pending); }), + subscribeChanges: PubSub.subscribe(changes).pipe(Effect.map(Stream.fromSubscription)), }); }); diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 0bb15f5e65893..11e2f43fb9309 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -704,7 +704,7 @@ const layerRuntimeDependencies = layerRuntimeCoreDependencies.pipe( Layer.provideMerge(ExternalLauncher.layer), Layer.provideMerge( Layer.mergeAll(RemoteOpenTargets.layer, DirectEndpoints.layer).pipe( - Layer.provide(TailscaleStatusReader.layer), + Layer.provideMerge(TailscaleStatusReader.layer), ), ), Layer.provideMerge(ServerLifecycleEvents.layer), diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index ac60becc2d78a..6037ffe7e7359 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,147 @@ 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. +const layerTailscaleRecovering = (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( + 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, + }); + }), + ), + ); + +it.effect("sends the recovered Tailscale target to a subscription opened on a stale failure", () => + Effect.gen(function* () { + let tailscaleUp = false; + const tailscaleServices = yield* Layer.build( + RemoteOpenTargets.layer.pipe( + Layer.provideMerge(TailscaleStatusReader.layer), + Layer.provide( + Layer.mergeAll( + layerTailscaleRecovering(() => tailscaleUp), + 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"); + tailscaleUp = true; + + 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, + }); + const targetKinds = (event: ServerConfigStreamEvent) => + event.type === "snapshot" ? event.config.remoteOpenTargets?.map((target) => target.kind) : []; + + const first = yield* Queue.take(responses); + assert.equal(first.type, "snapshot"); + assert.deepEqual(targetKinds(first), ["mdns"], "first snapshot answers from the stale failure"); + yield* server.write(0, { _tag: "Ack", requestId }); + const followUp = yield* Queue.take(responses); + assert.equal(followUp.type, "snapshot"); + assert.deepEqual(targetKinds(followUp), ["tailscale", "mdns"]); + yield* server.write(0, { _tag: "Interrupt", requestId, interruptors: [] }); + }).pipe( + Effect.provide( + Layer.mergeAll( + Layer.fresh(layerConfigSources), + Layer.mock(PreviewAutomationBroker.PreviewAutomationBroker)({}), + Layer.mock(ServerBrowser.ServerBrowser)({}), + ), + ), + Effect.scoped, + ), +); + // 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..07fa634bfa09c 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,15 @@ export const layerWsRpc = ( ), ); + const resolveNetworkTargets = Effect.fn("ws.resolveNetworkTargets")(function* () { + return { + remoteOpenTargets: yield* resolveAvailableEditorsForConfig( + remoteOpenTargets.resolveTargets(), + ), + directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()), + }; + }); + const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { readonly usageLimitsCommand: boolean; readonly workspaceSnapshotsOnDemand: boolean; @@ -1749,10 +1780,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 +3104,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; + // Discovery can start the Tailscale refresh this snapshot misses. + const tailscaleChanges = yield* tailscaleStatus.subscribeChanges; const config = yield* loadServerConfig({ usageLimitsCommand, workspaceSnapshotsOnDemand, @@ -3167,7 +3197,12 @@ export const layerWsRpc = ( return Stream.concat( rpcInitialItems([{ version: 1 as const, type: "snapshot" as const, config }]), - withLateEditorConfig(config, liveUpdates, externalLauncher), + withLateConfig( + config, + liveUpdates, + externalLauncher, + tailscaleChanges.pipe(Stream.mapEffect(resolveNetworkTargets)), + ), ); }), ), From fdf8e89b6b0d1762a683454e2c479f8dc4de74dd Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 03:06:03 +0000 Subject: [PATCH 4/5] fix(server): connects wait briefly for a fresh Tailscale read, change re-resolves never refresh A connect after Tailscale recovered got the stale failure in its first snapshot, and clients only learn direct routes from that snapshot. read() now waits up to 250 ms for the refresh it starts or joins before falling back to the previous result, and ws resolves RemoteOpenTargets and DirectEndpoints concurrently so they share that one grace window. Change-driven re-resolves provide SettledReadOnly, so they answer from the read that just settled instead of starting another refresh; a flapping CLI no longer chains refreshes through every open subscription. Co-Authored-By: Claude Opus 5.5 --- .../src/environment/DirectEndpoints.test.ts | 68 +++-- .../src/environment/TailscaleStatusReader.ts | 23 +- apps/server/src/ws.test.ts | 258 +++++++++++------- apps/server/src/ws.ts | 26 +- 4 files changed, 241 insertions(+), 134 deletions(-) diff --git a/apps/server/src/environment/DirectEndpoints.test.ts b/apps/server/src/environment/DirectEndpoints.test.ts index 6a83e9350cb2c..7a0fa846a90e5 100644 --- a/apps/server/src/environment/DirectEndpoints.test.ts +++ b/apps/server/src/environment/DirectEndpoints.test.ts @@ -180,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( @@ -286,18 +291,13 @@ 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 connectWithTailscaleHung = (onStatusSpawn: () => void) => +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, - layerTailscaleHung(onStatusSpawn), - layerServeProbe(200), - layerSshdListening, - ), + Layer.mergeAll(layerServeConfig, layerSpawner, layerServeProbe(200), layerSshdListening), ), ), ); @@ -315,7 +315,7 @@ 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* connectWithTailscaleHung(() => statusReads++); + 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, []]); @@ -327,7 +327,7 @@ describe("Tailscale status shared by DirectEndpoints and RemoteOpenTargets", () "answers a later connect from the settled failed read without waiting again", () => Effect.gen(function* () { - const connect = yield* connectWithTailscaleHung(() => {}); + const connect = yield* connectWithTailscale(layerTailscaleHung(() => {})); const first = yield* Effect.forkChild(connect); yield* TestClock.adjust("4 seconds"); expect(yield* Fiber.join(first)).toEqual([MDNS_ONLY, []]); @@ -339,4 +339,18 @@ describe("Tailscale status shared by DirectEndpoints and RemoteOpenTargets", () ); }), ); + + 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/TailscaleStatusReader.ts b/apps/server/src/environment/TailscaleStatusReader.ts index 5ee98f21b4da8..b17982429d1d7 100644 --- a/apps/server/src/environment/TailscaleStatusReader.ts +++ b/apps/server/src/environment/TailscaleStatusReader.ts @@ -6,9 +6,11 @@ import { } 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"; @@ -25,6 +27,15 @@ export class TailscaleStatusReader extends Context.Service< } >()("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; @@ -60,12 +71,18 @@ const make = Effect.gen(function* () { yield* refresh; return TailscaleStatusReader.of({ - // A caller gets the previous read's result; the refresh it starts serves the next caller - // and, when its outcome differs, reaches open subscribers through `subscribeChanges`. + // 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; - return yield* previous ?? Deferred.await(pending); + 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)), }); diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 6037ffe7e7359..2c6728ccf676c 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -884,15 +884,18 @@ 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. -const layerTailscaleRecovering = (isUp: () => boolean) => +// 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 = isUp(); + const { up, exits } = nextRead(); return ChildProcessSpawner.makeHandle({ pid: ChildProcessSpawner.ProcessId(1), - exitCode: Effect.succeed(ChildProcessSpawner.ExitCode(up ? 0 : 1)), + exitCode: exits.pipe(Effect.as(ChildProcessSpawner.ExitCode(up ? 0 : 1))), isRunning: Effect.succeed(false), kill: () => Effect.void, unref: Effect.succeed(Effect.void), @@ -915,104 +918,107 @@ const layerTailscaleRecovering = (isUp: () => boolean) => ), ); -it.effect("sends the recovered Tailscale target to a subscription opened on a stale failure", () => - Effect.gen(function* () { - let tailscaleUp = false; - const tailscaleServices = yield* Layer.build( - RemoteOpenTargets.layer.pipe( - Layer.provideMerge(TailscaleStatusReader.layer), - Layer.provide( - Layer.mergeAll( - layerTailscaleRecovering(() => tailscaleUp), - Layer.mock(NetService.NetService)({ hasListenerOnHost: () => Effect.succeed(true) }), - ), +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"); - tailscaleUp = 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(), + 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(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", }), - }), - 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, - }); - const targetKinds = (event: ServerConfigStreamEvent) => - event.type === "snapshot" ? event.config.remoteOpenTargets?.map((target) => target.kind) : []; - - const first = yield* Queue.take(responses); - assert.equal(first.type, "snapshot"); - assert.deepEqual(targetKinds(first), ["mdns"], "first snapshot answers from the stale failure"); - yield* server.write(0, { _tag: "Ack", requestId }); - const followUp = yield* Queue.take(responses); - assert.equal(followUp.type, "snapshot"); - assert.deepEqual(targetKinds(followUp), ["tailscale", "mdns"]); - yield* server.write(0, { _tag: "Interrupt", requestId, interruptors: [] }); + }), + ); + 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), @@ -1021,7 +1027,67 @@ it.effect("sends the recovered Tailscale target to a subscription opened on a st ), ), 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 diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 07fa634bfa09c..bdfc23215a553 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1738,13 +1738,15 @@ export const layerWsRpc = ( ), ); + // Concurrent, so both services share one Tailscale read and its grace window. const resolveNetworkTargets = Effect.fn("ws.resolveNetworkTargets")(function* () { - return { - remoteOpenTargets: yield* resolveAvailableEditorsForConfig( - remoteOpenTargets.resolveTargets(), - ), - directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()), - }; + return yield* Effect.all( + { + remoteOpenTargets: resolveAvailableEditorsForConfig(remoteOpenTargets.resolveTargets()), + directEndpoints: resolveAvailableEditorsForConfig(directEndpoints.resolve()), + }, + { concurrency: "unbounded" }, + ); }); const loadServerConfig = Effect.fn("ws.loadServerConfig")(function* (options: { @@ -3104,7 +3106,7 @@ 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; - // Discovery can start the Tailscale refresh this snapshot misses. + // A Tailscale refresh that outlasts the snapshot's grace arrives as a follow-up. const tailscaleChanges = yield* tailscaleStatus.subscribeChanges; const config = yield* loadServerConfig({ usageLimitsCommand, @@ -3201,7 +3203,15 @@ export const layerWsRpc = ( config, liveUpdates, externalLauncher, - tailscaleChanges.pipe(Stream.mapEffect(resolveNetworkTargets)), + // 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), + ), + ), + ), ), ); }), From c526075b8dc32284b64b796acf63a349403c509e Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 29 Oct 2026 03:14:39 +0000 Subject: [PATCH 5/5] fix(server): a refresh claim and its fork cannot be split by an interrupt Co-Authored-By: Claude Opus 5.5 --- apps/server/src/environment/TailscaleStatusReader.ts | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/apps/server/src/environment/TailscaleStatusReader.ts b/apps/server/src/environment/TailscaleStatusReader.ts index b17982429d1d7..6801d979dd4dc 100644 --- a/apps/server/src/environment/TailscaleStatusReader.ts +++ b/apps/server/src/environment/TailscaleStatusReader.ts @@ -64,10 +64,14 @@ const make = Effect.gen(function* () { ); }), ), + 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({